You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
393 lines
10 KiB
393 lines
10 KiB
package mempool
|
|
|
|
import (
|
|
"encoding/hex"
|
|
"errors"
|
|
"net"
|
|
"sync"
|
|
"testing"
|
|
"time"
|
|
|
|
"github.com/fortytw2/leaktest"
|
|
"github.com/go-kit/kit/log/term"
|
|
"github.com/stretchr/testify/assert"
|
|
"github.com/stretchr/testify/require"
|
|
|
|
"github.com/tendermint/tendermint/abci/example/kvstore"
|
|
abci "github.com/tendermint/tendermint/abci/types"
|
|
cfg "github.com/tendermint/tendermint/config"
|
|
"github.com/tendermint/tendermint/libs/log"
|
|
tmrand "github.com/tendermint/tendermint/libs/rand"
|
|
"github.com/tendermint/tendermint/p2p"
|
|
"github.com/tendermint/tendermint/p2p/mock"
|
|
memproto "github.com/tendermint/tendermint/proto/tendermint/mempool"
|
|
"github.com/tendermint/tendermint/proxy"
|
|
"github.com/tendermint/tendermint/types"
|
|
)
|
|
|
|
const (
|
|
numTxs = 1000
|
|
timeout = 120 * time.Second // ridiculously high because CircleCI is slow
|
|
)
|
|
|
|
type peerState struct {
|
|
height int64
|
|
}
|
|
|
|
func (ps peerState) GetHeight() int64 {
|
|
return ps.height
|
|
}
|
|
|
|
// Send a bunch of txs to the first reactor's mempool and wait for them all to
|
|
// be received in the others.
|
|
func TestReactorBroadcastTxsMessage(t *testing.T) {
|
|
config := cfg.TestConfig()
|
|
// if there were more than two reactors, the order of transactions could not be
|
|
// asserted in waitForTxsOnReactors (due to transactions gossiping). If we
|
|
// replace Connect2Switches (full mesh) with a func, which connects first
|
|
// reactor to others and nothing else, this test should also pass with >2 reactors.
|
|
const N = 2
|
|
reactors := makeAndConnectReactors(config, N)
|
|
defer func() {
|
|
for _, r := range reactors {
|
|
if err := r.Stop(); err != nil {
|
|
assert.NoError(t, err)
|
|
}
|
|
}
|
|
}()
|
|
for _, r := range reactors {
|
|
for _, peer := range r.Switch.Peers().List() {
|
|
peer.Set(types.PeerStateKey, peerState{1})
|
|
}
|
|
}
|
|
|
|
txs := checkTxs(t, reactors[0].mempool, numTxs, UnknownPeerID)
|
|
waitForTxsOnReactors(t, txs, reactors)
|
|
}
|
|
|
|
// regression test for https://github.com/tendermint/tendermint/issues/5408
|
|
func TestReactorConcurrency(t *testing.T) {
|
|
config := cfg.TestConfig()
|
|
const N = 2
|
|
reactors := makeAndConnectReactors(config, N)
|
|
defer func() {
|
|
for _, r := range reactors {
|
|
if err := r.Stop(); err != nil {
|
|
assert.NoError(t, err)
|
|
}
|
|
}
|
|
}()
|
|
for _, r := range reactors {
|
|
for _, peer := range r.Switch.Peers().List() {
|
|
peer.Set(types.PeerStateKey, peerState{1})
|
|
}
|
|
}
|
|
var wg sync.WaitGroup
|
|
|
|
const numTxs = 5
|
|
|
|
for i := 0; i < 1000; i++ {
|
|
wg.Add(2)
|
|
|
|
// 1. submit a bunch of txs
|
|
// 2. update the whole mempool
|
|
txs := checkTxs(t, reactors[0].mempool, numTxs, UnknownPeerID)
|
|
go func() {
|
|
defer wg.Done()
|
|
|
|
reactors[0].mempool.Lock()
|
|
defer reactors[0].mempool.Unlock()
|
|
|
|
deliverTxResponses := make([]*abci.ResponseDeliverTx, len(txs))
|
|
for i := range txs {
|
|
deliverTxResponses[i] = &abci.ResponseDeliverTx{Code: 0}
|
|
}
|
|
err := reactors[0].mempool.Update(1, txs, deliverTxResponses, nil, nil)
|
|
assert.NoError(t, err)
|
|
}()
|
|
|
|
// 1. submit a bunch of txs
|
|
// 2. update none
|
|
_ = checkTxs(t, reactors[1].mempool, numTxs, UnknownPeerID)
|
|
go func() {
|
|
defer wg.Done()
|
|
|
|
reactors[1].mempool.Lock()
|
|
defer reactors[1].mempool.Unlock()
|
|
err := reactors[1].mempool.Update(1, []types.Tx{}, make([]*abci.ResponseDeliverTx, 0), nil, nil)
|
|
assert.NoError(t, err)
|
|
}()
|
|
|
|
// 1. flush the mempool
|
|
reactors[1].mempool.Flush()
|
|
}
|
|
|
|
wg.Wait()
|
|
}
|
|
|
|
// Send a bunch of txs to the first reactor's mempool, claiming it came from peer
|
|
// ensure peer gets no txs.
|
|
func TestReactorNoBroadcastToSender(t *testing.T) {
|
|
config := cfg.TestConfig()
|
|
const N = 2
|
|
reactors := makeAndConnectReactors(config, N)
|
|
defer func() {
|
|
for _, r := range reactors {
|
|
if err := r.Stop(); err != nil {
|
|
assert.NoError(t, err)
|
|
}
|
|
}
|
|
}()
|
|
for _, r := range reactors {
|
|
for _, peer := range r.Switch.Peers().List() {
|
|
peer.Set(types.PeerStateKey, peerState{1})
|
|
}
|
|
}
|
|
|
|
const peerID = 1
|
|
checkTxs(t, reactors[0].mempool, numTxs, peerID)
|
|
ensureNoTxs(t, reactors[peerID], 100*time.Millisecond)
|
|
}
|
|
|
|
func TestReactor_MaxBatchBytes(t *testing.T) {
|
|
config := cfg.TestConfig()
|
|
config.Mempool.MaxBatchBytes = 1024
|
|
|
|
const N = 2
|
|
reactors := makeAndConnectReactors(config, N)
|
|
defer func() {
|
|
for _, r := range reactors {
|
|
if err := r.Stop(); err != nil {
|
|
assert.NoError(t, err)
|
|
}
|
|
}
|
|
}()
|
|
for _, r := range reactors {
|
|
for _, peer := range r.Switch.Peers().List() {
|
|
peer.Set(types.PeerStateKey, peerState{1})
|
|
}
|
|
}
|
|
|
|
// Broadcast a tx, which has the max size (minus proto overhead)
|
|
// => ensure it's received by the second reactor.
|
|
tx1 := tmrand.Bytes(1018)
|
|
err := reactors[0].mempool.CheckTx(tx1, nil, TxInfo{SenderID: UnknownPeerID})
|
|
require.NoError(t, err)
|
|
waitForTxsOnReactors(t, []types.Tx{tx1}, reactors)
|
|
|
|
reactors[0].mempool.Flush()
|
|
reactors[1].mempool.Flush()
|
|
|
|
// Broadcast a tx, which is beyond the max size
|
|
// => ensure it's not sent
|
|
tx2 := tmrand.Bytes(1020)
|
|
err = reactors[0].mempool.CheckTx(tx2, nil, TxInfo{SenderID: UnknownPeerID})
|
|
require.NoError(t, err)
|
|
ensureNoTxs(t, reactors[1], 100*time.Millisecond)
|
|
// => ensure the second reactor did not disconnect from us
|
|
out, in, _ := reactors[1].Switch.NumPeers()
|
|
assert.Equal(t, 1, out+in)
|
|
}
|
|
|
|
func TestBroadcastTxForPeerStopsWhenPeerStops(t *testing.T) {
|
|
if testing.Short() {
|
|
t.Skip("skipping test in short mode.")
|
|
}
|
|
|
|
config := cfg.TestConfig()
|
|
const N = 2
|
|
reactors := makeAndConnectReactors(config, N)
|
|
defer func() {
|
|
for _, r := range reactors {
|
|
if err := r.Stop(); err != nil {
|
|
assert.NoError(t, err)
|
|
}
|
|
}
|
|
}()
|
|
|
|
// stop peer
|
|
sw := reactors[1].Switch
|
|
sw.StopPeerForError(sw.Peers().List()[0], errors.New("some reason"))
|
|
|
|
// check that we are not leaking any go-routines
|
|
// i.e. broadcastTxRoutine finishes when peer is stopped
|
|
leaktest.CheckTimeout(t, 10*time.Second)()
|
|
}
|
|
|
|
func TestBroadcastTxForPeerStopsWhenReactorStops(t *testing.T) {
|
|
if testing.Short() {
|
|
t.Skip("skipping test in short mode.")
|
|
}
|
|
|
|
config := cfg.TestConfig()
|
|
const N = 2
|
|
reactors := makeAndConnectReactors(config, N)
|
|
|
|
// stop reactors
|
|
for _, r := range reactors {
|
|
if err := r.Stop(); err != nil {
|
|
assert.NoError(t, err)
|
|
}
|
|
}
|
|
|
|
// check that we are not leaking any go-routines
|
|
// i.e. broadcastTxRoutine finishes when reactor is stopped
|
|
leaktest.CheckTimeout(t, 10*time.Second)()
|
|
}
|
|
|
|
func TestMempoolIDsBasic(t *testing.T) {
|
|
ids := newMempoolIDs()
|
|
|
|
peer := mock.NewPeer(net.IP{127, 0, 0, 1})
|
|
|
|
ids.ReserveForPeer(peer)
|
|
assert.EqualValues(t, 1, ids.GetForPeer(peer))
|
|
ids.Reclaim(peer)
|
|
|
|
ids.ReserveForPeer(peer)
|
|
assert.EqualValues(t, 2, ids.GetForPeer(peer))
|
|
ids.Reclaim(peer)
|
|
}
|
|
|
|
func TestMempoolIDsPanicsIfNodeRequestsOvermaxActiveIDs(t *testing.T) {
|
|
if testing.Short() {
|
|
return
|
|
}
|
|
|
|
// 0 is already reserved for UnknownPeerID
|
|
ids := newMempoolIDs()
|
|
|
|
for i := 0; i < maxActiveIDs-1; i++ {
|
|
peer := mock.NewPeer(net.IP{127, 0, 0, 1})
|
|
ids.ReserveForPeer(peer)
|
|
}
|
|
|
|
assert.Panics(t, func() {
|
|
peer := mock.NewPeer(net.IP{127, 0, 0, 1})
|
|
ids.ReserveForPeer(peer)
|
|
})
|
|
}
|
|
|
|
func TestDontExhaustMaxActiveIDs(t *testing.T) {
|
|
config := cfg.TestConfig()
|
|
const N = 1
|
|
reactors := makeAndConnectReactors(config, N)
|
|
defer func() {
|
|
for _, r := range reactors {
|
|
if err := r.Stop(); err != nil {
|
|
assert.NoError(t, err)
|
|
}
|
|
}
|
|
}()
|
|
reactor := reactors[0]
|
|
|
|
for i := 0; i < maxActiveIDs+1; i++ {
|
|
peer := mock.NewPeer(nil)
|
|
reactor.Receive(MempoolChannel, peer, []byte{0x1, 0x2, 0x3})
|
|
reactor.AddPeer(peer)
|
|
}
|
|
}
|
|
|
|
// mempoolLogger is a TestingLogger which uses a different
|
|
// color for each validator ("validator" key must exist).
|
|
func mempoolLogger() log.Logger {
|
|
return log.TestingLoggerWithColorFn(func(keyvals ...interface{}) term.FgBgColor {
|
|
for i := 0; i < len(keyvals)-1; i += 2 {
|
|
if keyvals[i] == "validator" {
|
|
return term.FgBgColor{Fg: term.Color(uint8(keyvals[i+1].(int) + 1))}
|
|
}
|
|
}
|
|
return term.FgBgColor{}
|
|
})
|
|
}
|
|
|
|
// connect N mempool reactors through N switches
|
|
func makeAndConnectReactors(config *cfg.Config, n int) []*Reactor {
|
|
reactors := make([]*Reactor, n)
|
|
logger := mempoolLogger()
|
|
for i := 0; i < n; i++ {
|
|
app := kvstore.NewApplication()
|
|
cc := proxy.NewLocalClientCreator(app)
|
|
mempool, cleanup := newMempoolWithApp(cc)
|
|
defer cleanup()
|
|
|
|
reactors[i] = NewReactor(config.Mempool, mempool) // so we dont start the consensus states
|
|
reactors[i].SetLogger(logger.With("validator", i))
|
|
}
|
|
|
|
p2p.MakeConnectedSwitches(config.P2P, n, func(i int, s *p2p.Switch) *p2p.Switch {
|
|
s.AddReactor("MEMPOOL", reactors[i])
|
|
return s
|
|
|
|
}, p2p.Connect2Switches)
|
|
return reactors
|
|
}
|
|
|
|
func waitForTxsOnReactors(t *testing.T, txs types.Txs, reactors []*Reactor) {
|
|
// wait for the txs in all mempools
|
|
wg := new(sync.WaitGroup)
|
|
for i, reactor := range reactors {
|
|
wg.Add(1)
|
|
go func(r *Reactor, reactorIndex int) {
|
|
defer wg.Done()
|
|
waitForTxsOnReactor(t, txs, r, reactorIndex)
|
|
}(reactor, i)
|
|
}
|
|
|
|
done := make(chan struct{})
|
|
go func() {
|
|
wg.Wait()
|
|
close(done)
|
|
}()
|
|
|
|
timer := time.After(timeout)
|
|
select {
|
|
case <-timer:
|
|
t.Fatal("Timed out waiting for txs")
|
|
case <-done:
|
|
}
|
|
}
|
|
|
|
func waitForTxsOnReactor(t *testing.T, txs types.Txs, reactor *Reactor, reactorIndex int) {
|
|
mempool := reactor.mempool
|
|
for mempool.Size() < len(txs) {
|
|
time.Sleep(time.Millisecond * 100)
|
|
}
|
|
|
|
reapedTxs := mempool.ReapMaxTxs(len(txs))
|
|
for i, tx := range txs {
|
|
assert.Equalf(t, tx, reapedTxs[i],
|
|
"txs at index %d on reactor %d don't match: %v vs %v", i, reactorIndex, tx, reapedTxs[i])
|
|
}
|
|
}
|
|
|
|
// ensure no txs on reactor after some timeout
|
|
func ensureNoTxs(t *testing.T, reactor *Reactor, timeout time.Duration) {
|
|
time.Sleep(timeout) // wait for the txs in all mempools
|
|
assert.Zero(t, reactor.mempool.Size())
|
|
}
|
|
|
|
func TestMempoolVectors(t *testing.T) {
|
|
testCases := []struct {
|
|
testName string
|
|
tx []byte
|
|
expBytes string
|
|
}{
|
|
{"tx 1", []byte{123}, "0a030a017b"},
|
|
{"tx 2", []byte("proto encoding in mempool"), "0a1b0a1970726f746f20656e636f64696e6720696e206d656d706f6f6c"},
|
|
}
|
|
|
|
for _, tc := range testCases {
|
|
tc := tc
|
|
|
|
msg := memproto.Message{
|
|
Sum: &memproto.Message_Txs{
|
|
Txs: &memproto.Txs{Txs: [][]byte{tc.tx}},
|
|
},
|
|
}
|
|
bz, err := msg.Marshal()
|
|
require.NoError(t, err, tc.testName)
|
|
|
|
require.Equal(t, tc.expBytes, hex.EncodeToString(bz), tc.testName)
|
|
}
|
|
}
|