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.

437 lines
10 KiB

  1. package mempool
  2. import (
  3. "context"
  4. "os"
  5. "strings"
  6. "sync"
  7. "testing"
  8. "time"
  9. "github.com/stretchr/testify/require"
  10. "github.com/tendermint/tendermint/abci/example/kvstore"
  11. abci "github.com/tendermint/tendermint/abci/types"
  12. "github.com/tendermint/tendermint/config"
  13. tmsync "github.com/tendermint/tendermint/internal/libs/sync"
  14. "github.com/tendermint/tendermint/internal/p2p"
  15. "github.com/tendermint/tendermint/internal/p2p/p2ptest"
  16. "github.com/tendermint/tendermint/libs/log"
  17. tmrand "github.com/tendermint/tendermint/libs/rand"
  18. protomem "github.com/tendermint/tendermint/proto/tendermint/mempool"
  19. "github.com/tendermint/tendermint/types"
  20. )
  21. type reactorTestSuite struct {
  22. network *p2ptest.Network
  23. logger log.Logger
  24. reactors map[types.NodeID]*Reactor
  25. mempoolChannels map[types.NodeID]*p2p.Channel
  26. mempools map[types.NodeID]*TxMempool
  27. kvstores map[types.NodeID]*kvstore.Application
  28. peerChans map[types.NodeID]chan p2p.PeerUpdate
  29. peerUpdates map[types.NodeID]*p2p.PeerUpdates
  30. nodes []types.NodeID
  31. }
  32. func setupReactors(ctx context.Context, t *testing.T, numNodes int, chBuf uint) *reactorTestSuite {
  33. t.Helper()
  34. cfg, err := config.ResetTestRoot(strings.ReplaceAll(t.Name(), "/", "|"))
  35. require.NoError(t, err)
  36. t.Cleanup(func() { os.RemoveAll(cfg.RootDir) })
  37. rts := &reactorTestSuite{
  38. logger: log.TestingLogger().With("testCase", t.Name()),
  39. network: p2ptest.MakeNetwork(ctx, t, p2ptest.NetworkOptions{NumNodes: numNodes}),
  40. reactors: make(map[types.NodeID]*Reactor, numNodes),
  41. mempoolChannels: make(map[types.NodeID]*p2p.Channel, numNodes),
  42. mempools: make(map[types.NodeID]*TxMempool, numNodes),
  43. kvstores: make(map[types.NodeID]*kvstore.Application, numNodes),
  44. peerChans: make(map[types.NodeID]chan p2p.PeerUpdate, numNodes),
  45. peerUpdates: make(map[types.NodeID]*p2p.PeerUpdates, numNodes),
  46. }
  47. chDesc := GetChannelDescriptor(cfg.Mempool)
  48. rts.mempoolChannels = rts.network.MakeChannelsNoCleanup(t, chDesc)
  49. for nodeID := range rts.network.Nodes {
  50. rts.kvstores[nodeID] = kvstore.NewApplication()
  51. mempool := setup(ctx, t, 0)
  52. rts.mempools[nodeID] = mempool
  53. rts.peerChans[nodeID] = make(chan p2p.PeerUpdate)
  54. rts.peerUpdates[nodeID] = p2p.NewPeerUpdates(rts.peerChans[nodeID], 1)
  55. rts.network.Nodes[nodeID].PeerManager.Register(rts.peerUpdates[nodeID])
  56. rts.reactors[nodeID] = NewReactor(
  57. rts.logger.With("nodeID", nodeID),
  58. cfg.Mempool,
  59. rts.network.Nodes[nodeID].PeerManager,
  60. mempool,
  61. rts.mempoolChannels[nodeID],
  62. rts.peerUpdates[nodeID],
  63. )
  64. rts.nodes = append(rts.nodes, nodeID)
  65. require.NoError(t, rts.reactors[nodeID].Start(ctx))
  66. require.True(t, rts.reactors[nodeID].IsRunning())
  67. }
  68. require.Len(t, rts.reactors, numNodes)
  69. t.Cleanup(func() {
  70. for nodeID := range rts.reactors {
  71. if rts.reactors[nodeID].IsRunning() {
  72. require.NoError(t, rts.reactors[nodeID].Stop())
  73. require.False(t, rts.reactors[nodeID].IsRunning())
  74. }
  75. }
  76. })
  77. return rts
  78. }
  79. func (rts *reactorTestSuite) start(t *testing.T) {
  80. t.Helper()
  81. rts.network.Start(t)
  82. require.Len(t,
  83. rts.network.RandomNode().PeerManager.Peers(),
  84. len(rts.nodes)-1,
  85. "network does not have expected number of nodes")
  86. }
  87. func (rts *reactorTestSuite) assertMempoolChannelsDrained(t *testing.T) {
  88. t.Helper()
  89. rts.stop(t)
  90. for _, mch := range rts.mempoolChannels {
  91. require.Empty(t, mch.Out, "checking channel %q (len=%d)", mch.ID, len(mch.Out))
  92. }
  93. }
  94. func (rts *reactorTestSuite) stop(t *testing.T) {
  95. for id, r := range rts.reactors {
  96. require.NoError(t, r.Stop(), "stopping reactor %s", id)
  97. r.Wait()
  98. require.False(t, r.IsRunning(), "reactor %s did not stop", id)
  99. }
  100. }
  101. func (rts *reactorTestSuite) waitForTxns(t *testing.T, txs []types.Tx, ids ...types.NodeID) {
  102. t.Helper()
  103. // ensure that the transactions get fully broadcast to the
  104. // rest of the network
  105. wg := &sync.WaitGroup{}
  106. for name, pool := range rts.mempools {
  107. if !p2ptest.NodeInSlice(name, ids) {
  108. continue
  109. }
  110. wg.Add(1)
  111. go func(pool *TxMempool) {
  112. defer wg.Done()
  113. require.Eventually(t, func() bool { return len(txs) == pool.Size() },
  114. time.Minute,
  115. 100*time.Millisecond,
  116. )
  117. }(pool)
  118. }
  119. wg.Wait()
  120. }
  121. func TestReactorBroadcastDoesNotPanic(t *testing.T) {
  122. ctx, cancel := context.WithCancel(context.Background())
  123. defer cancel()
  124. const numNodes = 2
  125. rts := setupReactors(ctx, t, numNodes, 0)
  126. observePanic := func(r interface{}) {
  127. t.Fatal("panic detected in reactor")
  128. }
  129. primary := rts.nodes[0]
  130. secondary := rts.nodes[1]
  131. primaryReactor := rts.reactors[primary]
  132. primaryMempool := primaryReactor.mempool
  133. secondaryReactor := rts.reactors[secondary]
  134. primaryReactor.observePanic = observePanic
  135. secondaryReactor.observePanic = observePanic
  136. firstTx := &WrappedTx{}
  137. primaryMempool.insertTx(firstTx)
  138. // run the router
  139. rts.start(t)
  140. closer := tmsync.NewCloser()
  141. primaryReactor.peerWG.Add(1)
  142. go primaryReactor.broadcastTxRoutine(secondary, closer)
  143. wg := &sync.WaitGroup{}
  144. for i := 0; i < 50; i++ {
  145. next := &WrappedTx{}
  146. wg.Add(1)
  147. go func() {
  148. defer wg.Done()
  149. primaryMempool.insertTx(next)
  150. }()
  151. }
  152. err := primaryReactor.Stop()
  153. require.NoError(t, err)
  154. primaryReactor.peerWG.Wait()
  155. wg.Wait()
  156. }
  157. func TestReactorBroadcastTxs(t *testing.T) {
  158. numTxs := 1000
  159. numNodes := 10
  160. ctx, cancel := context.WithCancel(context.Background())
  161. defer cancel()
  162. rts := setupReactors(ctx, t, numNodes, 0)
  163. primary := rts.nodes[0]
  164. secondaries := rts.nodes[1:]
  165. txs := checkTxs(t, rts.reactors[primary].mempool, numTxs, UnknownPeerID)
  166. // run the router
  167. rts.start(t)
  168. // Wait till all secondary suites (reactor) received all mempool txs from the
  169. // primary suite (node).
  170. rts.waitForTxns(t, convertTex(txs), secondaries...)
  171. rts.stop(t)
  172. }
  173. // regression test for https://github.com/tendermint/tendermint/issues/5408
  174. func TestReactorConcurrency(t *testing.T) {
  175. numTxs := 5
  176. numNodes := 2
  177. ctx, cancel := context.WithCancel(context.Background())
  178. defer cancel()
  179. rts := setupReactors(ctx, t, numNodes, 0)
  180. primary := rts.nodes[0]
  181. secondary := rts.nodes[1]
  182. rts.start(t)
  183. var wg sync.WaitGroup
  184. for i := 0; i < 1000; i++ {
  185. wg.Add(2)
  186. // 1. submit a bunch of txs
  187. // 2. update the whole mempool
  188. txs := checkTxs(t, rts.reactors[primary].mempool, numTxs, UnknownPeerID)
  189. go func() {
  190. defer wg.Done()
  191. mempool := rts.mempools[primary]
  192. mempool.Lock()
  193. defer mempool.Unlock()
  194. deliverTxResponses := make([]*abci.ResponseDeliverTx, len(txs))
  195. for i := range txs {
  196. deliverTxResponses[i] = &abci.ResponseDeliverTx{Code: 0}
  197. }
  198. require.NoError(t, mempool.Update(1, convertTex(txs), deliverTxResponses, nil, nil))
  199. }()
  200. // 1. submit a bunch of txs
  201. // 2. update none
  202. _ = checkTxs(t, rts.reactors[secondary].mempool, numTxs, UnknownPeerID)
  203. go func() {
  204. defer wg.Done()
  205. mempool := rts.mempools[secondary]
  206. mempool.Lock()
  207. defer mempool.Unlock()
  208. err := mempool.Update(1, []types.Tx{}, make([]*abci.ResponseDeliverTx, 0), nil, nil)
  209. require.NoError(t, err)
  210. }()
  211. // flush the mempool
  212. rts.mempools[secondary].Flush()
  213. }
  214. wg.Wait()
  215. }
  216. func TestReactorNoBroadcastToSender(t *testing.T) {
  217. numTxs := 1000
  218. numNodes := 2
  219. ctx, cancel := context.WithCancel(context.Background())
  220. defer cancel()
  221. rts := setupReactors(ctx, t, numNodes, uint(numTxs))
  222. primary := rts.nodes[0]
  223. secondary := rts.nodes[1]
  224. peerID := uint16(1)
  225. _ = checkTxs(t, rts.mempools[primary], numTxs, peerID)
  226. rts.start(t)
  227. time.Sleep(100 * time.Millisecond)
  228. require.Eventually(t, func() bool {
  229. return rts.mempools[secondary].Size() == 0
  230. }, time.Minute, 100*time.Millisecond)
  231. rts.assertMempoolChannelsDrained(t)
  232. }
  233. func TestReactor_MaxTxBytes(t *testing.T) {
  234. numNodes := 2
  235. cfg := config.TestConfig()
  236. ctx, cancel := context.WithCancel(context.Background())
  237. defer cancel()
  238. rts := setupReactors(ctx, t, numNodes, 0)
  239. primary := rts.nodes[0]
  240. secondary := rts.nodes[1]
  241. // Broadcast a tx, which has the max size and ensure it's received by the
  242. // second reactor.
  243. tx1 := tmrand.Bytes(cfg.Mempool.MaxTxBytes)
  244. err := rts.reactors[primary].mempool.CheckTx(
  245. ctx,
  246. tx1,
  247. nil,
  248. TxInfo{
  249. SenderID: UnknownPeerID,
  250. },
  251. )
  252. require.NoError(t, err)
  253. rts.start(t)
  254. rts.reactors[primary].mempool.Flush()
  255. rts.reactors[secondary].mempool.Flush()
  256. // broadcast a tx, which is beyond the max size and ensure it's not sent
  257. tx2 := tmrand.Bytes(cfg.Mempool.MaxTxBytes + 1)
  258. err = rts.mempools[primary].CheckTx(ctx, tx2, nil, TxInfo{SenderID: UnknownPeerID})
  259. require.Error(t, err)
  260. rts.assertMempoolChannelsDrained(t)
  261. }
  262. func TestDontExhaustMaxActiveIDs(t *testing.T) {
  263. // we're creating a single node network, but not starting the
  264. // network.
  265. ctx, cancel := context.WithCancel(context.Background())
  266. defer cancel()
  267. rts := setupReactors(ctx, t, 1, MaxActiveIDs+1)
  268. nodeID := rts.nodes[0]
  269. peerID, err := types.NewNodeID("0011223344556677889900112233445566778899")
  270. require.NoError(t, err)
  271. // ensure the reactor does not panic (i.e. exhaust active IDs)
  272. for i := 0; i < MaxActiveIDs+1; i++ {
  273. rts.peerChans[nodeID] <- p2p.PeerUpdate{
  274. Status: p2p.PeerStatusUp,
  275. NodeID: peerID,
  276. }
  277. rts.mempoolChannels[nodeID].Out <- p2p.Envelope{
  278. To: peerID,
  279. Message: &protomem.Txs{
  280. Txs: [][]byte{},
  281. },
  282. }
  283. }
  284. require.Eventually(
  285. t,
  286. func() bool {
  287. for _, mch := range rts.mempoolChannels {
  288. if len(mch.Out) > 0 {
  289. return false
  290. }
  291. }
  292. return true
  293. },
  294. time.Minute,
  295. 10*time.Millisecond,
  296. )
  297. rts.assertMempoolChannelsDrained(t)
  298. }
  299. func TestMempoolIDsPanicsIfNodeRequestsOvermaxActiveIDs(t *testing.T) {
  300. if testing.Short() {
  301. t.Skip("skipping test in short mode")
  302. }
  303. // 0 is already reserved for UnknownPeerID
  304. ids := NewMempoolIDs()
  305. peerID, err := types.NewNodeID("0011223344556677889900112233445566778899")
  306. require.NoError(t, err)
  307. for i := 0; i < MaxActiveIDs-1; i++ {
  308. ids.ReserveForPeer(peerID)
  309. }
  310. require.Panics(t, func() {
  311. ids.ReserveForPeer(peerID)
  312. })
  313. }
  314. func TestBroadcastTxForPeerStopsWhenPeerStops(t *testing.T) {
  315. if testing.Short() {
  316. t.Skip("skipping test in short mode")
  317. }
  318. ctx, cancel := context.WithCancel(context.Background())
  319. defer cancel()
  320. rts := setupReactors(ctx, t, 2, 0)
  321. primary := rts.nodes[0]
  322. secondary := rts.nodes[1]
  323. rts.start(t)
  324. // disconnect peer
  325. rts.peerChans[primary] <- p2p.PeerUpdate{
  326. Status: p2p.PeerStatusDown,
  327. NodeID: secondary,
  328. }
  329. time.Sleep(500 * time.Millisecond)
  330. txs := checkTxs(t, rts.reactors[primary].mempool, 4, UnknownPeerID)
  331. require.Equal(t, 4, len(txs))
  332. require.Equal(t, 4, rts.mempools[primary].Size())
  333. require.Equal(t, 0, rts.mempools[secondary].Size())
  334. }