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.

736 lines
20 KiB

cleanup: Reduce and normalize import path aliasing. (#6975) The code in the Tendermint repository makes heavy use of import aliasing. This is made necessary by our extensive reuse of common base package names, and by repetition of similar names across different subdirectories. Unfortunately we have not been very consistent about which packages we alias in various circumstances, and the aliases we use vary. In the spirit of the advice in the style guide and https://github.com/golang/go/wiki/CodeReviewComments#imports, his change makes an effort to clean up and normalize import aliasing. This change makes no API or behavioral changes. It is a pure cleanup intended o help make the code more readable to developers (including myself) trying to understand what is being imported where. Only unexported names have been modified, and the changes were generated and applied mechanically with gofmt -r and comby, respecting the lexical and syntactic rules of Go. Even so, I did not fix every inconsistency. Where the changes would be too disruptive, I left it alone. The principles I followed in this cleanup are: - Remove aliases that restate the package name. - Remove aliases where the base package name is unambiguous. - Move overly-terse abbreviations from the import to the usage site. - Fix lexical issues (remove underscores, remove capitalization). - Fix import groupings to more closely match the style guide. - Group blank (side-effecting) imports and ensure they are commented. - Add aliases to multiple imports with the same base package name.
3 years ago
cleanup: Reduce and normalize import path aliasing. (#6975) The code in the Tendermint repository makes heavy use of import aliasing. This is made necessary by our extensive reuse of common base package names, and by repetition of similar names across different subdirectories. Unfortunately we have not been very consistent about which packages we alias in various circumstances, and the aliases we use vary. In the spirit of the advice in the style guide and https://github.com/golang/go/wiki/CodeReviewComments#imports, his change makes an effort to clean up and normalize import aliasing. This change makes no API or behavioral changes. It is a pure cleanup intended o help make the code more readable to developers (including myself) trying to understand what is being imported where. Only unexported names have been modified, and the changes were generated and applied mechanically with gofmt -r and comby, respecting the lexical and syntactic rules of Go. Even so, I did not fix every inconsistency. Where the changes would be too disruptive, I left it alone. The principles I followed in this cleanup are: - Remove aliases that restate the package name. - Remove aliases where the base package name is unambiguous. - Move overly-terse abbreviations from the import to the usage site. - Fix lexical issues (remove underscores, remove capitalization). - Fix import groupings to more closely match the style guide. - Group blank (side-effecting) imports and ensure they are commented. - Add aliases to multiple imports with the same base package name.
3 years ago
cleanup: Reduce and normalize import path aliasing. (#6975) The code in the Tendermint repository makes heavy use of import aliasing. This is made necessary by our extensive reuse of common base package names, and by repetition of similar names across different subdirectories. Unfortunately we have not been very consistent about which packages we alias in various circumstances, and the aliases we use vary. In the spirit of the advice in the style guide and https://github.com/golang/go/wiki/CodeReviewComments#imports, his change makes an effort to clean up and normalize import aliasing. This change makes no API or behavioral changes. It is a pure cleanup intended o help make the code more readable to developers (including myself) trying to understand what is being imported where. Only unexported names have been modified, and the changes were generated and applied mechanically with gofmt -r and comby, respecting the lexical and syntactic rules of Go. Even so, I did not fix every inconsistency. Where the changes would be too disruptive, I left it alone. The principles I followed in this cleanup are: - Remove aliases that restate the package name. - Remove aliases where the base package name is unambiguous. - Move overly-terse abbreviations from the import to the usage site. - Fix lexical issues (remove underscores, remove capitalization). - Fix import groupings to more closely match the style guide. - Group blank (side-effecting) imports and ensure they are commented. - Add aliases to multiple imports with the same base package name.
3 years ago
cleanup: Reduce and normalize import path aliasing. (#6975) The code in the Tendermint repository makes heavy use of import aliasing. This is made necessary by our extensive reuse of common base package names, and by repetition of similar names across different subdirectories. Unfortunately we have not been very consistent about which packages we alias in various circumstances, and the aliases we use vary. In the spirit of the advice in the style guide and https://github.com/golang/go/wiki/CodeReviewComments#imports, his change makes an effort to clean up and normalize import aliasing. This change makes no API or behavioral changes. It is a pure cleanup intended o help make the code more readable to developers (including myself) trying to understand what is being imported where. Only unexported names have been modified, and the changes were generated and applied mechanically with gofmt -r and comby, respecting the lexical and syntactic rules of Go. Even so, I did not fix every inconsistency. Where the changes would be too disruptive, I left it alone. The principles I followed in this cleanup are: - Remove aliases that restate the package name. - Remove aliases where the base package name is unambiguous. - Move overly-terse abbreviations from the import to the usage site. - Fix lexical issues (remove underscores, remove capitalization). - Fix import groupings to more closely match the style guide. - Group blank (side-effecting) imports and ensure they are commented. - Add aliases to multiple imports with the same base package name.
3 years ago
cleanup: Reduce and normalize import path aliasing. (#6975) The code in the Tendermint repository makes heavy use of import aliasing. This is made necessary by our extensive reuse of common base package names, and by repetition of similar names across different subdirectories. Unfortunately we have not been very consistent about which packages we alias in various circumstances, and the aliases we use vary. In the spirit of the advice in the style guide and https://github.com/golang/go/wiki/CodeReviewComments#imports, his change makes an effort to clean up and normalize import aliasing. This change makes no API or behavioral changes. It is a pure cleanup intended o help make the code more readable to developers (including myself) trying to understand what is being imported where. Only unexported names have been modified, and the changes were generated and applied mechanically with gofmt -r and comby, respecting the lexical and syntactic rules of Go. Even so, I did not fix every inconsistency. Where the changes would be too disruptive, I left it alone. The principles I followed in this cleanup are: - Remove aliases that restate the package name. - Remove aliases where the base package name is unambiguous. - Move overly-terse abbreviations from the import to the usage site. - Fix lexical issues (remove underscores, remove capitalization). - Fix import groupings to more closely match the style guide. - Group blank (side-effecting) imports and ensure they are commented. - Add aliases to multiple imports with the same base package name.
3 years ago
cleanup: Reduce and normalize import path aliasing. (#6975) The code in the Tendermint repository makes heavy use of import aliasing. This is made necessary by our extensive reuse of common base package names, and by repetition of similar names across different subdirectories. Unfortunately we have not been very consistent about which packages we alias in various circumstances, and the aliases we use vary. In the spirit of the advice in the style guide and https://github.com/golang/go/wiki/CodeReviewComments#imports, his change makes an effort to clean up and normalize import aliasing. This change makes no API or behavioral changes. It is a pure cleanup intended o help make the code more readable to developers (including myself) trying to understand what is being imported where. Only unexported names have been modified, and the changes were generated and applied mechanically with gofmt -r and comby, respecting the lexical and syntactic rules of Go. Even so, I did not fix every inconsistency. Where the changes would be too disruptive, I left it alone. The principles I followed in this cleanup are: - Remove aliases that restate the package name. - Remove aliases where the base package name is unambiguous. - Move overly-terse abbreviations from the import to the usage site. - Fix lexical issues (remove underscores, remove capitalization). - Fix import groupings to more closely match the style guide. - Group blank (side-effecting) imports and ensure they are commented. - Add aliases to multiple imports with the same base package name.
3 years ago
cleanup: Reduce and normalize import path aliasing. (#6975) The code in the Tendermint repository makes heavy use of import aliasing. This is made necessary by our extensive reuse of common base package names, and by repetition of similar names across different subdirectories. Unfortunately we have not been very consistent about which packages we alias in various circumstances, and the aliases we use vary. In the spirit of the advice in the style guide and https://github.com/golang/go/wiki/CodeReviewComments#imports, his change makes an effort to clean up and normalize import aliasing. This change makes no API or behavioral changes. It is a pure cleanup intended o help make the code more readable to developers (including myself) trying to understand what is being imported where. Only unexported names have been modified, and the changes were generated and applied mechanically with gofmt -r and comby, respecting the lexical and syntactic rules of Go. Even so, I did not fix every inconsistency. Where the changes would be too disruptive, I left it alone. The principles I followed in this cleanup are: - Remove aliases that restate the package name. - Remove aliases where the base package name is unambiguous. - Move overly-terse abbreviations from the import to the usage site. - Fix lexical issues (remove underscores, remove capitalization). - Fix import groupings to more closely match the style guide. - Group blank (side-effecting) imports and ensure they are commented. - Add aliases to multiple imports with the same base package name.
3 years ago
  1. //nolint:unused
  2. package pex_test
  3. import (
  4. "context"
  5. "errors"
  6. "strings"
  7. "testing"
  8. "time"
  9. "github.com/stretchr/testify/require"
  10. dbm "github.com/tendermint/tm-db"
  11. "github.com/tendermint/tendermint/crypto/ed25519"
  12. "github.com/tendermint/tendermint/internal/p2p"
  13. "github.com/tendermint/tendermint/internal/p2p/p2ptest"
  14. "github.com/tendermint/tendermint/internal/p2p/pex"
  15. "github.com/tendermint/tendermint/libs/log"
  16. p2pproto "github.com/tendermint/tendermint/proto/tendermint/p2p"
  17. "github.com/tendermint/tendermint/types"
  18. )
  19. const (
  20. checkFrequency = 500 * time.Millisecond
  21. defaultBufferSize = 2
  22. shortWait = 10 * time.Second
  23. longWait = 60 * time.Second
  24. firstNode = 0
  25. secondNode = 1
  26. thirdNode = 2
  27. )
  28. func TestReactorBasic(t *testing.T) {
  29. ctx, cancel := context.WithCancel(context.Background())
  30. defer cancel()
  31. // start a network with one mock reactor and one "real" reactor
  32. testNet := setupNetwork(ctx, t, testOptions{
  33. MockNodes: 1,
  34. TotalNodes: 2,
  35. })
  36. testNet.connectAll(ctx, t)
  37. testNet.start(ctx, t)
  38. // assert that the mock node receives a request from the real node
  39. testNet.listenForRequest(ctx, t, secondNode, firstNode, shortWait)
  40. // assert that when a mock node sends a request it receives a response (and
  41. // the correct one)
  42. testNet.sendRequest(ctx, t, firstNode, secondNode)
  43. testNet.listenForResponse(ctx, t, secondNode, firstNode, shortWait, []p2pproto.PexAddress(nil))
  44. }
  45. func TestReactorConnectFullNetwork(t *testing.T) {
  46. ctx, cancel := context.WithCancel(context.Background())
  47. defer cancel()
  48. testNet := setupNetwork(ctx, t, testOptions{
  49. TotalNodes: 4,
  50. })
  51. // make every node be only connected with one other node (it actually ends up
  52. // being two because of two way connections but oh well)
  53. testNet.connectN(ctx, t, 1)
  54. testNet.start(ctx, t)
  55. // assert that all nodes add each other in the network
  56. for idx := 0; idx < len(testNet.nodes); idx++ {
  57. testNet.requireNumberOfPeers(t, idx, len(testNet.nodes)-1, longWait)
  58. }
  59. }
  60. func TestReactorSendsRequestsTooOften(t *testing.T) {
  61. ctx, cancel := context.WithCancel(context.Background())
  62. defer cancel()
  63. r := setupSingle(ctx, t)
  64. badNode := newNodeID(t, "b")
  65. r.pexInCh <- p2p.Envelope{
  66. From: badNode,
  67. Message: &p2pproto.PexRequest{},
  68. }
  69. resp := <-r.pexOutCh
  70. msg, ok := resp.Message.(*p2pproto.PexResponse)
  71. require.True(t, ok)
  72. require.Empty(t, msg.Addresses)
  73. r.pexInCh <- p2p.Envelope{
  74. From: badNode,
  75. Message: &p2pproto.PexRequest{},
  76. }
  77. peerErr := <-r.pexErrCh
  78. require.Error(t, peerErr.Err)
  79. require.Empty(t, r.pexOutCh)
  80. require.Contains(t, peerErr.Err.Error(), "sent PEX request too soon")
  81. require.Equal(t, badNode, peerErr.NodeID)
  82. }
  83. func TestReactorSendsResponseWithoutRequest(t *testing.T) {
  84. t.Skip("This test needs updated https://github.com/tendermint/tendermint/issue/7634")
  85. ctx, cancel := context.WithCancel(context.Background())
  86. defer cancel()
  87. testNet := setupNetwork(ctx, t, testOptions{
  88. MockNodes: 1,
  89. TotalNodes: 3,
  90. })
  91. testNet.connectAll(ctx, t)
  92. testNet.start(ctx, t)
  93. // firstNode sends the secondNode an unrequested response
  94. // NOTE: secondNode will send a request by default during startup so we send
  95. // two responses to counter that.
  96. testNet.sendResponse(ctx, t, firstNode, secondNode, []int{thirdNode})
  97. testNet.sendResponse(ctx, t, firstNode, secondNode, []int{thirdNode})
  98. // secondNode should evict the firstNode
  99. testNet.listenForPeerUpdate(ctx, t, secondNode, firstNode, p2p.PeerStatusDown, shortWait)
  100. }
  101. func TestReactorNeverSendsTooManyPeers(t *testing.T) {
  102. t.Skip("This test needs updated https://github.com/tendermint/tendermint/issue/7634")
  103. ctx, cancel := context.WithCancel(context.Background())
  104. defer cancel()
  105. testNet := setupNetwork(ctx, t, testOptions{
  106. MockNodes: 1,
  107. TotalNodes: 2,
  108. })
  109. testNet.connectAll(ctx, t)
  110. testNet.start(ctx, t)
  111. testNet.addNodes(ctx, t, 110)
  112. nodes := make([]int, 110)
  113. for i := 0; i < len(nodes); i++ {
  114. nodes[i] = i + 2
  115. }
  116. testNet.addAddresses(t, secondNode, nodes)
  117. // first we check that even although we have 110 peers, honest pex reactors
  118. // only send 100 (test if secondNode sends firstNode 100 addresses)
  119. testNet.pingAndlistenForNAddresses(ctx, t, secondNode, firstNode, shortWait, 100)
  120. }
  121. func TestReactorErrorsOnReceivingTooManyPeers(t *testing.T) {
  122. ctx, cancel := context.WithCancel(context.Background())
  123. defer cancel()
  124. r := setupSingle(ctx, t)
  125. peer := p2p.NodeAddress{Protocol: p2p.MemoryProtocol, NodeID: randomNodeID(t)}
  126. added, err := r.manager.Add(peer)
  127. require.NoError(t, err)
  128. require.True(t, added)
  129. addresses := make([]p2pproto.PexAddress, 101)
  130. for i := 0; i < len(addresses); i++ {
  131. nodeAddress := p2p.NodeAddress{Protocol: p2p.MemoryProtocol, NodeID: randomNodeID(t)}
  132. addresses[i] = p2pproto.PexAddress{
  133. URL: nodeAddress.String(),
  134. }
  135. }
  136. r.peerCh <- p2p.PeerUpdate{
  137. NodeID: peer.NodeID,
  138. Status: p2p.PeerStatusUp,
  139. }
  140. select {
  141. // wait for a request and then send a response with too many addresses
  142. case req := <-r.pexOutCh:
  143. if _, ok := req.Message.(*p2pproto.PexRequest); !ok {
  144. t.Fatal("expected v2 pex request")
  145. }
  146. r.pexInCh <- p2p.Envelope{
  147. From: peer.NodeID,
  148. Message: &p2pproto.PexResponse{
  149. Addresses: addresses,
  150. },
  151. }
  152. case <-time.After(10 * time.Second):
  153. t.Fatal("pex failed to send a request within 10 seconds")
  154. }
  155. peerErr := <-r.pexErrCh
  156. require.Error(t, peerErr.Err)
  157. require.Empty(t, r.pexOutCh)
  158. require.Contains(t, peerErr.Err.Error(), "peer sent too many addresses")
  159. require.Equal(t, peer.NodeID, peerErr.NodeID)
  160. }
  161. func TestReactorSmallPeerStoreInALargeNetwork(t *testing.T) {
  162. ctx, cancel := context.WithCancel(context.Background())
  163. defer cancel()
  164. testNet := setupNetwork(ctx, t, testOptions{
  165. TotalNodes: 8,
  166. MaxPeers: 4,
  167. MaxConnected: 3,
  168. BufferSize: 8,
  169. })
  170. testNet.connectN(ctx, t, 1)
  171. testNet.start(ctx, t)
  172. // test that all nodes reach full capacity
  173. for _, nodeID := range testNet.nodes {
  174. require.Eventually(t, func() bool {
  175. // nolint:scopelint
  176. return testNet.network.Nodes[nodeID].PeerManager.PeerRatio() >= 0.9
  177. }, longWait, checkFrequency)
  178. }
  179. }
  180. func TestReactorLargePeerStoreInASmallNetwork(t *testing.T) {
  181. ctx, cancel := context.WithCancel(context.Background())
  182. defer cancel()
  183. testNet := setupNetwork(ctx, t, testOptions{
  184. TotalNodes: 3,
  185. MaxPeers: 25,
  186. MaxConnected: 25,
  187. BufferSize: 5,
  188. })
  189. testNet.connectN(ctx, t, 1)
  190. testNet.start(ctx, t)
  191. // assert that all nodes add each other in the network
  192. for idx := 0; idx < len(testNet.nodes); idx++ {
  193. testNet.requireNumberOfPeers(t, idx, len(testNet.nodes)-1, longWait)
  194. }
  195. }
  196. func TestReactorWithNetworkGrowth(t *testing.T) {
  197. t.Skip("This test needs updated https://github.com/tendermint/tendermint/issue/7634")
  198. ctx, cancel := context.WithCancel(context.Background())
  199. defer cancel()
  200. testNet := setupNetwork(ctx, t, testOptions{
  201. TotalNodes: 5,
  202. BufferSize: 5,
  203. })
  204. testNet.connectAll(ctx, t)
  205. testNet.start(ctx, t)
  206. // assert that all nodes add each other in the network
  207. for idx := 0; idx < len(testNet.nodes); idx++ {
  208. testNet.requireNumberOfPeers(t, idx, len(testNet.nodes)-1, shortWait)
  209. }
  210. // now we inject 10 more nodes
  211. testNet.addNodes(ctx, t, 10)
  212. for i := 5; i < testNet.total; i++ {
  213. node := testNet.nodes[i]
  214. require.NoError(t, testNet.reactors[node].Start(ctx))
  215. require.True(t, testNet.reactors[node].IsRunning())
  216. // we connect all new nodes to a single entry point and check that the
  217. // node can distribute the addresses to all the others
  218. testNet.connectPeers(ctx, t, 0, i)
  219. }
  220. require.Len(t, testNet.reactors, 15)
  221. // assert that all nodes add each other in the network
  222. for idx := 0; idx < len(testNet.nodes); idx++ {
  223. testNet.requireNumberOfPeers(t, idx, len(testNet.nodes)-1, longWait)
  224. }
  225. }
  226. type singleTestReactor struct {
  227. reactor *pex.Reactor
  228. pexInCh chan p2p.Envelope
  229. pexOutCh chan p2p.Envelope
  230. pexErrCh chan p2p.PeerError
  231. pexCh *p2p.Channel
  232. peerCh chan p2p.PeerUpdate
  233. manager *p2p.PeerManager
  234. }
  235. func setupSingle(ctx context.Context, t *testing.T) *singleTestReactor {
  236. t.Helper()
  237. nodeID := newNodeID(t, "a")
  238. chBuf := 2
  239. pexInCh := make(chan p2p.Envelope, chBuf)
  240. pexOutCh := make(chan p2p.Envelope, chBuf)
  241. pexErrCh := make(chan p2p.PeerError, chBuf)
  242. pexCh := p2p.NewChannel(
  243. p2p.ChannelID(pex.PexChannel),
  244. new(p2pproto.PexMessage),
  245. pexInCh,
  246. pexOutCh,
  247. pexErrCh,
  248. )
  249. peerCh := make(chan p2p.PeerUpdate, chBuf)
  250. peerUpdates := p2p.NewPeerUpdates(peerCh, chBuf)
  251. peerManager, err := p2p.NewPeerManager(nodeID, dbm.NewMemDB(), p2p.PeerManagerOptions{})
  252. require.NoError(t, err)
  253. chCreator := func(context.Context, *p2p.ChannelDescriptor) (*p2p.Channel, error) {
  254. return pexCh, nil
  255. }
  256. reactor, err := pex.NewReactor(ctx, log.NewNopLogger(), peerManager, chCreator, peerUpdates)
  257. require.NoError(t, err)
  258. require.NoError(t, reactor.Start(ctx))
  259. t.Cleanup(reactor.Wait)
  260. return &singleTestReactor{
  261. reactor: reactor,
  262. pexInCh: pexInCh,
  263. pexOutCh: pexOutCh,
  264. pexErrCh: pexErrCh,
  265. pexCh: pexCh,
  266. peerCh: peerCh,
  267. manager: peerManager,
  268. }
  269. }
  270. type reactorTestSuite struct {
  271. network *p2ptest.Network
  272. logger log.Logger
  273. reactors map[types.NodeID]*pex.Reactor
  274. pexChannels map[types.NodeID]*p2p.Channel
  275. peerChans map[types.NodeID]chan p2p.PeerUpdate
  276. peerUpdates map[types.NodeID]*p2p.PeerUpdates
  277. nodes []types.NodeID
  278. mocks []types.NodeID
  279. total int
  280. opts testOptions
  281. }
  282. type testOptions struct {
  283. MockNodes int
  284. TotalNodes int
  285. BufferSize int
  286. MaxPeers uint16
  287. MaxConnected uint16
  288. }
  289. // setup setups a test suite with a network of nodes. Mocknodes represent the
  290. // hollow nodes that the test can listen and send on
  291. func setupNetwork(ctx context.Context, t *testing.T, opts testOptions) *reactorTestSuite {
  292. t.Helper()
  293. require.Greater(t, opts.TotalNodes, opts.MockNodes)
  294. if opts.BufferSize == 0 {
  295. opts.BufferSize = defaultBufferSize
  296. }
  297. networkOpts := p2ptest.NetworkOptions{
  298. NumNodes: opts.TotalNodes,
  299. BufferSize: opts.BufferSize,
  300. NodeOpts: p2ptest.NodeOptions{
  301. MaxPeers: opts.MaxPeers,
  302. MaxConnected: opts.MaxConnected,
  303. },
  304. }
  305. chBuf := opts.BufferSize
  306. realNodes := opts.TotalNodes - opts.MockNodes
  307. rts := &reactorTestSuite{
  308. logger: log.NewNopLogger().With("testCase", t.Name()),
  309. network: p2ptest.MakeNetwork(ctx, t, networkOpts),
  310. reactors: make(map[types.NodeID]*pex.Reactor, realNodes),
  311. pexChannels: make(map[types.NodeID]*p2p.Channel, opts.TotalNodes),
  312. peerChans: make(map[types.NodeID]chan p2p.PeerUpdate, opts.TotalNodes),
  313. peerUpdates: make(map[types.NodeID]*p2p.PeerUpdates, opts.TotalNodes),
  314. total: opts.TotalNodes,
  315. opts: opts,
  316. }
  317. // NOTE: we don't assert that the channels get drained after stopping the
  318. // reactor
  319. rts.pexChannels = rts.network.MakeChannelsNoCleanup(ctx, t, pex.ChannelDescriptor())
  320. idx := 0
  321. for nodeID := range rts.network.Nodes {
  322. rts.peerChans[nodeID] = make(chan p2p.PeerUpdate, chBuf)
  323. rts.peerUpdates[nodeID] = p2p.NewPeerUpdates(rts.peerChans[nodeID], chBuf)
  324. rts.network.Nodes[nodeID].PeerManager.Register(ctx, rts.peerUpdates[nodeID])
  325. chCreator := func(context.Context, *p2p.ChannelDescriptor) (*p2p.Channel, error) {
  326. return rts.pexChannels[nodeID], nil
  327. }
  328. // the first nodes in the array are always mock nodes
  329. if idx < opts.MockNodes {
  330. rts.mocks = append(rts.mocks, nodeID)
  331. } else {
  332. var err error
  333. rts.reactors[nodeID], err = pex.NewReactor(
  334. ctx,
  335. rts.logger.With("nodeID", nodeID),
  336. rts.network.Nodes[nodeID].PeerManager,
  337. chCreator,
  338. rts.peerUpdates[nodeID],
  339. )
  340. require.NoError(t, err)
  341. }
  342. rts.nodes = append(rts.nodes, nodeID)
  343. idx++
  344. }
  345. require.Len(t, rts.reactors, realNodes)
  346. t.Cleanup(func() {
  347. for _, reactor := range rts.reactors {
  348. if reactor.IsRunning() {
  349. reactor.Wait()
  350. require.False(t, reactor.IsRunning())
  351. }
  352. }
  353. })
  354. return rts
  355. }
  356. // starts up the pex reactors for each node
  357. func (r *reactorTestSuite) start(ctx context.Context, t *testing.T) {
  358. t.Helper()
  359. for _, reactor := range r.reactors {
  360. require.NoError(t, reactor.Start(ctx))
  361. require.True(t, reactor.IsRunning())
  362. }
  363. }
  364. func (r *reactorTestSuite) addNodes(ctx context.Context, t *testing.T, nodes int) {
  365. t.Helper()
  366. for i := 0; i < nodes; i++ {
  367. node := r.network.MakeNode(ctx, t, p2ptest.NodeOptions{
  368. MaxPeers: r.opts.MaxPeers,
  369. MaxConnected: r.opts.MaxConnected,
  370. })
  371. r.network.Nodes[node.NodeID] = node
  372. nodeID := node.NodeID
  373. r.pexChannels[nodeID] = node.MakeChannelNoCleanup(ctx, t, pex.ChannelDescriptor())
  374. r.peerChans[nodeID] = make(chan p2p.PeerUpdate, r.opts.BufferSize)
  375. r.peerUpdates[nodeID] = p2p.NewPeerUpdates(r.peerChans[nodeID], r.opts.BufferSize)
  376. r.network.Nodes[nodeID].PeerManager.Register(ctx, r.peerUpdates[nodeID])
  377. chCreator := func(context.Context, *p2p.ChannelDescriptor) (*p2p.Channel, error) {
  378. return r.pexChannels[nodeID], nil
  379. }
  380. var err error
  381. r.reactors[nodeID], err = pex.NewReactor(
  382. ctx,
  383. r.logger.With("nodeID", nodeID),
  384. r.network.Nodes[nodeID].PeerManager,
  385. chCreator,
  386. r.peerUpdates[nodeID],
  387. )
  388. require.NoError(t, err)
  389. r.nodes = append(r.nodes, nodeID)
  390. r.total++
  391. }
  392. }
  393. func (r *reactorTestSuite) listenFor(
  394. ctx context.Context,
  395. t *testing.T,
  396. node types.NodeID,
  397. conditional func(msg *p2p.Envelope) bool,
  398. assertion func(t *testing.T, msg *p2p.Envelope) bool,
  399. waitPeriod time.Duration,
  400. ) {
  401. ctx, cancel := context.WithTimeout(ctx, waitPeriod)
  402. defer cancel()
  403. iter := r.pexChannels[node].Receive(ctx)
  404. for iter.Next(ctx) {
  405. envelope := iter.Envelope()
  406. if conditional(envelope) && assertion(t, envelope) {
  407. return
  408. }
  409. }
  410. if errors.Is(ctx.Err(), context.DeadlineExceeded) {
  411. require.Fail(t, "timed out waiting for message",
  412. "node=%v, waitPeriod=%s", node, waitPeriod)
  413. }
  414. }
  415. func (r *reactorTestSuite) listenForRequest(ctx context.Context, t *testing.T, fromNode, toNode int, waitPeriod time.Duration) {
  416. to, from := r.checkNodePair(t, toNode, fromNode)
  417. conditional := func(msg *p2p.Envelope) bool {
  418. _, ok := msg.Message.(*p2pproto.PexRequest)
  419. return ok && msg.From == from
  420. }
  421. assertion := func(t *testing.T, msg *p2p.Envelope) bool {
  422. require.Equal(t, &p2pproto.PexRequest{}, msg.Message)
  423. return true
  424. }
  425. r.listenFor(ctx, t, to, conditional, assertion, waitPeriod)
  426. }
  427. func (r *reactorTestSuite) pingAndlistenForNAddresses(
  428. ctx context.Context,
  429. t *testing.T,
  430. fromNode, toNode int,
  431. waitPeriod time.Duration,
  432. addresses int,
  433. ) {
  434. t.Helper()
  435. to, from := r.checkNodePair(t, toNode, fromNode)
  436. conditional := func(msg *p2p.Envelope) bool {
  437. _, ok := msg.Message.(*p2pproto.PexResponse)
  438. return ok && msg.From == from
  439. }
  440. assertion := func(t *testing.T, msg *p2p.Envelope) bool {
  441. m, ok := msg.Message.(*p2pproto.PexResponse)
  442. if !ok {
  443. require.Fail(t, "expected pex response v2")
  444. return true
  445. }
  446. // assert the same amount of addresses
  447. if len(m.Addresses) == addresses {
  448. return true
  449. }
  450. // if we didn't get the right length, we wait and send the
  451. // request again
  452. time.Sleep(300 * time.Millisecond)
  453. r.sendRequest(ctx, t, toNode, fromNode)
  454. return false
  455. }
  456. r.sendRequest(ctx, t, toNode, fromNode)
  457. r.listenFor(ctx, t, to, conditional, assertion, waitPeriod)
  458. }
  459. func (r *reactorTestSuite) listenForResponse(
  460. ctx context.Context,
  461. t *testing.T,
  462. fromNode, toNode int,
  463. waitPeriod time.Duration,
  464. addresses []p2pproto.PexAddress,
  465. ) {
  466. to, from := r.checkNodePair(t, toNode, fromNode)
  467. conditional := func(msg *p2p.Envelope) bool {
  468. _, ok := msg.Message.(*p2pproto.PexResponse)
  469. return ok && msg.From == from
  470. }
  471. assertion := func(t *testing.T, msg *p2p.Envelope) bool {
  472. require.Equal(t, &p2pproto.PexResponse{Addresses: addresses}, msg.Message)
  473. return true
  474. }
  475. r.listenFor(ctx, t, to, conditional, assertion, waitPeriod)
  476. }
  477. func (r *reactorTestSuite) listenForPeerUpdate(
  478. ctx context.Context,
  479. t *testing.T,
  480. onNode, withNode int,
  481. status p2p.PeerStatus,
  482. waitPeriod time.Duration,
  483. ) {
  484. on, with := r.checkNodePair(t, onNode, withNode)
  485. sub := r.network.Nodes[on].PeerManager.Subscribe(ctx)
  486. timesUp := time.After(waitPeriod)
  487. for {
  488. select {
  489. case <-ctx.Done():
  490. require.Fail(t, "operation canceled")
  491. return
  492. case peerUpdate := <-sub.Updates():
  493. if peerUpdate.NodeID == with {
  494. require.Equal(t, status, peerUpdate.Status)
  495. return
  496. }
  497. case <-timesUp:
  498. require.Fail(t, "timed out waiting for peer status", "%v with status %v",
  499. with, status)
  500. return
  501. }
  502. }
  503. }
  504. func (r *reactorTestSuite) getAddressesFor(nodes []int) []p2pproto.PexAddress {
  505. addresses := make([]p2pproto.PexAddress, len(nodes))
  506. for idx, node := range nodes {
  507. nodeID := r.nodes[node]
  508. addresses[idx] = p2pproto.PexAddress{
  509. URL: r.network.Nodes[nodeID].NodeAddress.String(),
  510. }
  511. }
  512. return addresses
  513. }
  514. func (r *reactorTestSuite) sendRequest(ctx context.Context, t *testing.T, fromNode, toNode int) {
  515. t.Helper()
  516. to, from := r.checkNodePair(t, toNode, fromNode)
  517. require.NoError(t, r.pexChannels[from].Send(ctx, p2p.Envelope{
  518. To: to,
  519. Message: &p2pproto.PexRequest{},
  520. }))
  521. }
  522. func (r *reactorTestSuite) sendResponse(
  523. ctx context.Context,
  524. t *testing.T,
  525. fromNode, toNode int,
  526. withNodes []int,
  527. ) {
  528. t.Helper()
  529. from, to := r.checkNodePair(t, fromNode, toNode)
  530. addrs := r.getAddressesFor(withNodes)
  531. require.NoError(t, r.pexChannels[from].Send(ctx, p2p.Envelope{
  532. To: to,
  533. Message: &p2pproto.PexResponse{
  534. Addresses: addrs,
  535. },
  536. }))
  537. }
  538. func (r *reactorTestSuite) requireNumberOfPeers(
  539. t *testing.T,
  540. nodeIndex, numPeers int,
  541. waitPeriod time.Duration,
  542. ) {
  543. t.Helper()
  544. require.Eventuallyf(t, func() bool {
  545. actualNumPeers := len(r.network.Nodes[r.nodes[nodeIndex]].PeerManager.Peers())
  546. return actualNumPeers >= numPeers
  547. }, waitPeriod, checkFrequency, "peer failed to connect with the asserted amount of peers "+
  548. "index=%d, node=%q, waitPeriod=%s expected=%d actual=%d",
  549. nodeIndex, r.nodes[nodeIndex], waitPeriod, numPeers,
  550. len(r.network.Nodes[r.nodes[nodeIndex]].PeerManager.Peers()),
  551. )
  552. }
  553. func (r *reactorTestSuite) connectAll(ctx context.Context, t *testing.T) {
  554. r.connectN(ctx, t, r.total-1)
  555. }
  556. // connects all nodes with n other nodes
  557. func (r *reactorTestSuite) connectN(ctx context.Context, t *testing.T, n int) {
  558. if n >= r.total {
  559. require.Fail(t, "connectN: n must be less than the size of the network - 1")
  560. }
  561. for i := 0; i < r.total; i++ {
  562. for j := 0; j < n; j++ {
  563. r.connectPeers(ctx, t, i, (i+j+1)%r.total)
  564. }
  565. }
  566. }
  567. // connects node1 to node2
  568. func (r *reactorTestSuite) connectPeers(ctx context.Context, t *testing.T, sourceNode, targetNode int) {
  569. t.Helper()
  570. node1, node2 := r.checkNodePair(t, sourceNode, targetNode)
  571. n1 := r.network.Nodes[node1]
  572. if n1 == nil {
  573. require.Fail(t, "connectPeers: source node %v is not part of the testnet", node1)
  574. return
  575. }
  576. n2 := r.network.Nodes[node2]
  577. if n2 == nil {
  578. require.Fail(t, "connectPeers: target node %v is not part of the testnet", node2)
  579. return
  580. }
  581. sourceSub := n1.PeerManager.Subscribe(ctx)
  582. targetSub := n2.PeerManager.Subscribe(ctx)
  583. sourceAddress := n1.NodeAddress
  584. targetAddress := n2.NodeAddress
  585. added, err := n1.PeerManager.Add(targetAddress)
  586. require.NoError(t, err)
  587. if !added {
  588. return
  589. }
  590. select {
  591. case peerUpdate := <-targetSub.Updates():
  592. require.Equal(t, peerUpdate.NodeID, node1)
  593. require.Equal(t, peerUpdate.Status, p2p.PeerStatusUp)
  594. case <-time.After(2 * time.Second):
  595. require.Fail(t, "timed out waiting for peer", "%v accepting %v",
  596. targetNode, sourceNode)
  597. }
  598. select {
  599. case peerUpdate := <-sourceSub.Updates():
  600. require.Equal(t, peerUpdate.NodeID, node2)
  601. require.Equal(t, peerUpdate.Status, p2p.PeerStatusUp)
  602. case <-time.After(2 * time.Second):
  603. require.Fail(t, "timed out waiting for peer", "%v dialing %v",
  604. sourceNode, targetNode)
  605. }
  606. added, err = n2.PeerManager.Add(sourceAddress)
  607. require.NoError(t, err)
  608. require.True(t, added)
  609. }
  610. func (r *reactorTestSuite) checkNodePair(t *testing.T, first, second int) (types.NodeID, types.NodeID) {
  611. require.NotEqual(t, first, second)
  612. require.Less(t, first, r.total)
  613. require.Less(t, second, r.total)
  614. return r.nodes[first], r.nodes[second]
  615. }
  616. func (r *reactorTestSuite) addAddresses(t *testing.T, node int, addrs []int) {
  617. peerManager := r.network.Nodes[r.nodes[node]].PeerManager
  618. for _, addr := range addrs {
  619. require.Less(t, addr, r.total)
  620. address := r.network.Nodes[r.nodes[addr]].NodeAddress
  621. added, err := peerManager.Add(address)
  622. require.NoError(t, err)
  623. require.True(t, added)
  624. }
  625. }
  626. func newNodeID(t *testing.T, id string) types.NodeID {
  627. nodeID, err := types.NewNodeID(strings.Repeat(id, 2*types.NodeIDByteLength))
  628. require.NoError(t, err)
  629. return nodeID
  630. }
  631. func randomNodeID(t *testing.T) types.NodeID {
  632. return types.NodeIDFromPubKey(ed25519.GenPrivKey().PubKey())
  633. }