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.

969 lines
29 KiB

p2p: make PeerManager.DialNext() and EvictNext() block (#5947) See #5936 and #5938 for background. The plan was initially to have `DialNext()` and `EvictNext()` return a channel. However, implementing this became unnecessarily complicated and error-prone. As an example, the channel would be both consumed and populated (via method calls) by the same driving method (e.g. `Router.dialPeers()`) which could easily cause deadlocks where a method call blocked while sending on the channel that the caller itself was responsible for consuming (but couldn't since it was busy making the method call). It would also require a set of goroutines in the peer manager that would interact with the goroutines in the router in non-obvious ways, and fully populating the channel on startup could cause deadlocks with other startup tasks. Several issues like these made the solution hard to reason about. I therefore simply made `DialNext()` and `EvictNext()` block until the next peer was available, using internal triggers to wake these methods up in a non-blocking fashion when any relevant state changes occurred. This proved much simpler to reason about, since there are no goroutines in the peer manager (except for trivial retry timers), nor any blocking channel sends, and it instead relies entirely on the existing goroutine structure of the router for concurrency. This also happens to be the same pattern used by the `Transport.Accept()` API, following Go stdlib conventions, so all router goroutines end up using a consistent pattern as well.
4 years ago
p2p: make PeerManager.DialNext() and EvictNext() block (#5947) See #5936 and #5938 for background. The plan was initially to have `DialNext()` and `EvictNext()` return a channel. However, implementing this became unnecessarily complicated and error-prone. As an example, the channel would be both consumed and populated (via method calls) by the same driving method (e.g. `Router.dialPeers()`) which could easily cause deadlocks where a method call blocked while sending on the channel that the caller itself was responsible for consuming (but couldn't since it was busy making the method call). It would also require a set of goroutines in the peer manager that would interact with the goroutines in the router in non-obvious ways, and fully populating the channel on startup could cause deadlocks with other startup tasks. Several issues like these made the solution hard to reason about. I therefore simply made `DialNext()` and `EvictNext()` block until the next peer was available, using internal triggers to wake these methods up in a non-blocking fashion when any relevant state changes occurred. This proved much simpler to reason about, since there are no goroutines in the peer manager (except for trivial retry timers), nor any blocking channel sends, and it instead relies entirely on the existing goroutine structure of the router for concurrency. This also happens to be the same pattern used by the `Transport.Accept()` API, following Go stdlib conventions, so all router goroutines end up using a consistent pattern as well.
4 years ago
  1. package p2p_test
  2. import (
  3. "context"
  4. "errors"
  5. "fmt"
  6. "io"
  7. "runtime"
  8. "strings"
  9. "sync"
  10. "testing"
  11. "time"
  12. "github.com/fortytw2/leaktest"
  13. "github.com/gogo/protobuf/proto"
  14. gogotypes "github.com/gogo/protobuf/types"
  15. "github.com/stretchr/testify/mock"
  16. "github.com/stretchr/testify/require"
  17. dbm "github.com/tendermint/tm-db"
  18. "github.com/tendermint/tendermint/crypto"
  19. tmsync "github.com/tendermint/tendermint/internal/libs/sync"
  20. "github.com/tendermint/tendermint/internal/p2p"
  21. "github.com/tendermint/tendermint/internal/p2p/mocks"
  22. "github.com/tendermint/tendermint/internal/p2p/p2ptest"
  23. "github.com/tendermint/tendermint/libs/log"
  24. "github.com/tendermint/tendermint/types"
  25. )
  26. func echoReactor(ctx context.Context, channel *p2p.Channel) {
  27. iter := channel.Receive(ctx)
  28. for iter.Next(ctx) {
  29. envelope := iter.Envelope()
  30. value := envelope.Message.(*p2ptest.Message).Value
  31. if err := channel.Send(ctx, p2p.Envelope{
  32. To: envelope.From,
  33. Message: &p2ptest.Message{Value: value},
  34. }); err != nil {
  35. return
  36. }
  37. }
  38. }
  39. func TestRouter_Network(t *testing.T) {
  40. ctx, cancel := context.WithCancel(context.Background())
  41. defer cancel()
  42. t.Cleanup(leaktest.Check(t))
  43. // Create a test network and open a channel where all peers run echoReactor.
  44. network := p2ptest.MakeNetwork(ctx, t, p2ptest.NetworkOptions{NumNodes: 8})
  45. local := network.RandomNode()
  46. peers := network.Peers(local.NodeID)
  47. channels := network.MakeChannels(ctx, t, chDesc)
  48. network.Start(ctx, t)
  49. channel := channels[local.NodeID]
  50. for _, peer := range peers {
  51. go echoReactor(ctx, channels[peer.NodeID])
  52. }
  53. // Sending a message to each peer should work.
  54. for _, peer := range peers {
  55. p2ptest.RequireSendReceive(ctx, t, channel, peer.NodeID,
  56. &p2ptest.Message{Value: "foo"},
  57. &p2ptest.Message{Value: "foo"},
  58. )
  59. }
  60. // Sending a broadcast should return back a message from all peers.
  61. p2ptest.RequireSend(ctx, t, channel, p2p.Envelope{
  62. Broadcast: true,
  63. Message: &p2ptest.Message{Value: "bar"},
  64. })
  65. expect := []*p2p.Envelope{}
  66. for _, peer := range peers {
  67. expect = append(expect, &p2p.Envelope{
  68. From: peer.NodeID,
  69. ChannelID: 1,
  70. Message: &p2ptest.Message{Value: "bar"},
  71. })
  72. }
  73. p2ptest.RequireReceiveUnordered(ctx, t, channel, expect)
  74. // We then submit an error for a peer, and watch it get disconnected and
  75. // then reconnected as the router retries it.
  76. peerUpdates := local.MakePeerUpdatesNoRequireEmpty(ctx, t)
  77. require.NoError(t, channel.SendError(ctx, p2p.PeerError{
  78. NodeID: peers[0].NodeID,
  79. Err: errors.New("boom"),
  80. }))
  81. p2ptest.RequireUpdates(t, peerUpdates, []p2p.PeerUpdate{
  82. {NodeID: peers[0].NodeID, Status: p2p.PeerStatusDown},
  83. {NodeID: peers[0].NodeID, Status: p2p.PeerStatusUp},
  84. })
  85. }
  86. func TestRouter_Channel_Basic(t *testing.T) {
  87. t.Cleanup(leaktest.Check(t))
  88. ctx, cancel := context.WithCancel(context.Background())
  89. defer cancel()
  90. // Set up a router with no transports (so no peers).
  91. peerManager, err := p2p.NewPeerManager(selfID, dbm.NewMemDB(), p2p.PeerManagerOptions{})
  92. require.NoError(t, err)
  93. router, err := p2p.NewRouter(
  94. ctx,
  95. log.TestingLogger(),
  96. p2p.NopMetrics(),
  97. selfInfo,
  98. selfKey,
  99. peerManager,
  100. nil,
  101. nil,
  102. p2p.RouterOptions{},
  103. )
  104. require.NoError(t, err)
  105. require.NoError(t, router.Start(ctx))
  106. t.Cleanup(router.Wait)
  107. // Opening a channel should work.
  108. chctx, chcancel := context.WithCancel(ctx)
  109. defer chcancel()
  110. channel, err := router.OpenChannel(chctx, chDesc)
  111. require.NoError(t, err)
  112. require.Contains(t, router.NodeInfo().Channels, byte(chDesc.ID))
  113. require.NotNil(t, channel)
  114. // Opening the same channel again should fail.
  115. _, err = router.OpenChannel(ctx, chDesc)
  116. require.Error(t, err)
  117. // Opening a different channel should work.
  118. chDesc2 := &p2p.ChannelDescriptor{ID: 2, MessageType: &p2ptest.Message{}}
  119. _, err = router.OpenChannel(ctx, chDesc2)
  120. require.NoError(t, err)
  121. require.Contains(t, router.NodeInfo().Channels, byte(chDesc2.ID))
  122. // Closing the channel, then opening it again should be fine.
  123. chcancel()
  124. time.Sleep(200 * time.Millisecond) // yes yes, but Close() is async...
  125. channel, err = router.OpenChannel(ctx, chDesc)
  126. require.NoError(t, err)
  127. // We should be able to send on the channel, even though there are no peers.
  128. p2ptest.RequireSend(ctx, t, channel, p2p.Envelope{
  129. To: types.NodeID(strings.Repeat("a", 40)),
  130. Message: &p2ptest.Message{Value: "foo"},
  131. })
  132. // A message to ourselves should be dropped.
  133. p2ptest.RequireSend(ctx, t, channel, p2p.Envelope{
  134. To: selfID,
  135. Message: &p2ptest.Message{Value: "self"},
  136. })
  137. p2ptest.RequireEmpty(ctx, t, channel)
  138. }
  139. // Channel tests are hairy to mock, so we use an in-memory network instead.
  140. func TestRouter_Channel_SendReceive(t *testing.T) {
  141. ctx, cancel := context.WithCancel(context.Background())
  142. defer cancel()
  143. t.Cleanup(leaktest.Check(t))
  144. // Create a test network and open a channel on all nodes.
  145. network := p2ptest.MakeNetwork(ctx, t, p2ptest.NetworkOptions{NumNodes: 3})
  146. ids := network.NodeIDs()
  147. aID, bID, cID := ids[0], ids[1], ids[2]
  148. channels := network.MakeChannels(ctx, t, chDesc)
  149. a, b, c := channels[aID], channels[bID], channels[cID]
  150. otherChannels := network.MakeChannels(ctx, t, p2ptest.MakeChannelDesc(9))
  151. network.Start(ctx, t)
  152. // Sending a message a->b should work, and not send anything
  153. // further to a, b, or c.
  154. p2ptest.RequireSend(ctx, t, a, p2p.Envelope{To: bID, Message: &p2ptest.Message{Value: "foo"}})
  155. p2ptest.RequireReceive(ctx, t, b, p2p.Envelope{From: aID, Message: &p2ptest.Message{Value: "foo"}})
  156. p2ptest.RequireEmpty(ctx, t, a, b, c)
  157. // Sending a nil message a->b should be dropped.
  158. p2ptest.RequireSend(ctx, t, a, p2p.Envelope{To: bID, Message: nil})
  159. p2ptest.RequireEmpty(ctx, t, a, b, c)
  160. // Sending a different message type should be dropped.
  161. p2ptest.RequireSend(ctx, t, a, p2p.Envelope{To: bID, Message: &gogotypes.BoolValue{Value: true}})
  162. p2ptest.RequireEmpty(ctx, t, a, b, c)
  163. // Sending to an unknown peer should be dropped.
  164. p2ptest.RequireSend(ctx, t, a, p2p.Envelope{
  165. To: types.NodeID(strings.Repeat("a", 40)),
  166. Message: &p2ptest.Message{Value: "a"},
  167. })
  168. p2ptest.RequireEmpty(ctx, t, a, b, c)
  169. // Sending without a recipient should be dropped.
  170. p2ptest.RequireSend(ctx, t, a, p2p.Envelope{Message: &p2ptest.Message{Value: "noto"}})
  171. p2ptest.RequireEmpty(ctx, t, a, b, c)
  172. // Sending to self should be dropped.
  173. p2ptest.RequireSend(ctx, t, a, p2p.Envelope{To: aID, Message: &p2ptest.Message{Value: "self"}})
  174. p2ptest.RequireEmpty(ctx, t, a, b, c)
  175. // Removing b and sending to it should be dropped.
  176. network.Remove(ctx, t, bID)
  177. p2ptest.RequireSend(ctx, t, a, p2p.Envelope{To: bID, Message: &p2ptest.Message{Value: "nob"}})
  178. p2ptest.RequireEmpty(ctx, t, a, b, c)
  179. // After all this, sending a message c->a should work.
  180. p2ptest.RequireSend(ctx, t, c, p2p.Envelope{To: aID, Message: &p2ptest.Message{Value: "bar"}})
  181. p2ptest.RequireReceive(ctx, t, a, p2p.Envelope{From: cID, Message: &p2ptest.Message{Value: "bar"}})
  182. p2ptest.RequireEmpty(ctx, t, a, b, c)
  183. // None of these messages should have made it onto the other channels.
  184. for _, other := range otherChannels {
  185. p2ptest.RequireEmpty(ctx, t, other)
  186. }
  187. }
  188. func TestRouter_Channel_Broadcast(t *testing.T) {
  189. t.Cleanup(leaktest.Check(t))
  190. ctx, cancel := context.WithCancel(context.Background())
  191. defer cancel()
  192. // Create a test network and open a channel on all nodes.
  193. network := p2ptest.MakeNetwork(ctx, t, p2ptest.NetworkOptions{NumNodes: 4})
  194. ids := network.NodeIDs()
  195. aID, bID, cID, dID := ids[0], ids[1], ids[2], ids[3]
  196. channels := network.MakeChannels(ctx, t, chDesc)
  197. a, b, c, d := channels[aID], channels[bID], channels[cID], channels[dID]
  198. network.Start(ctx, t)
  199. // Sending a broadcast from b should work.
  200. p2ptest.RequireSend(ctx, t, b, p2p.Envelope{Broadcast: true, Message: &p2ptest.Message{Value: "foo"}})
  201. p2ptest.RequireReceive(ctx, t, a, p2p.Envelope{From: bID, Message: &p2ptest.Message{Value: "foo"}})
  202. p2ptest.RequireReceive(ctx, t, c, p2p.Envelope{From: bID, Message: &p2ptest.Message{Value: "foo"}})
  203. p2ptest.RequireReceive(ctx, t, d, p2p.Envelope{From: bID, Message: &p2ptest.Message{Value: "foo"}})
  204. p2ptest.RequireEmpty(ctx, t, a, b, c, d)
  205. // Removing one node from the network shouldn't prevent broadcasts from working.
  206. network.Remove(ctx, t, dID)
  207. p2ptest.RequireSend(ctx, t, a, p2p.Envelope{Broadcast: true, Message: &p2ptest.Message{Value: "bar"}})
  208. p2ptest.RequireReceive(ctx, t, b, p2p.Envelope{From: aID, Message: &p2ptest.Message{Value: "bar"}})
  209. p2ptest.RequireReceive(ctx, t, c, p2p.Envelope{From: aID, Message: &p2ptest.Message{Value: "bar"}})
  210. p2ptest.RequireEmpty(ctx, t, a, b, c, d)
  211. }
  212. func TestRouter_Channel_Wrapper(t *testing.T) {
  213. t.Cleanup(leaktest.Check(t))
  214. ctx, cancel := context.WithCancel(context.Background())
  215. defer cancel()
  216. // Create a test network and open a channel on all nodes.
  217. network := p2ptest.MakeNetwork(ctx, t, p2ptest.NetworkOptions{NumNodes: 2})
  218. ids := network.NodeIDs()
  219. aID, bID := ids[0], ids[1]
  220. chDesc := &p2p.ChannelDescriptor{
  221. ID: chID,
  222. MessageType: &wrapperMessage{},
  223. Priority: 5,
  224. SendQueueCapacity: 10,
  225. RecvMessageCapacity: 10,
  226. }
  227. channels := network.MakeChannels(ctx, t, chDesc)
  228. a, b := channels[aID], channels[bID]
  229. network.Start(ctx, t)
  230. // Since wrapperMessage implements p2p.Wrapper and handles Message, it
  231. // should automatically wrap and unwrap sent messages -- we prepend the
  232. // wrapper actions to the message value to signal this.
  233. p2ptest.RequireSend(ctx, t, a, p2p.Envelope{To: bID, Message: &p2ptest.Message{Value: "foo"}})
  234. p2ptest.RequireReceive(ctx, t, b, p2p.Envelope{From: aID, Message: &p2ptest.Message{Value: "unwrap:wrap:foo"}})
  235. // If we send a different message that can't be wrapped, it should be dropped.
  236. p2ptest.RequireSend(ctx, t, a, p2p.Envelope{To: bID, Message: &gogotypes.BoolValue{Value: true}})
  237. p2ptest.RequireEmpty(ctx, t, b)
  238. // If we send the wrapper message itself, it should also be passed through
  239. // since WrapperMessage supports it, and should only be unwrapped at the receiver.
  240. p2ptest.RequireSend(ctx, t, a, p2p.Envelope{
  241. To: bID,
  242. Message: &wrapperMessage{Message: p2ptest.Message{Value: "foo"}},
  243. })
  244. p2ptest.RequireReceive(ctx, t, b, p2p.Envelope{
  245. From: aID,
  246. Message: &p2ptest.Message{Value: "unwrap:foo"},
  247. })
  248. }
  249. // WrapperMessage prepends the value with "wrap:" and "unwrap:" to test it.
  250. type wrapperMessage struct {
  251. p2ptest.Message
  252. }
  253. var _ p2p.Wrapper = (*wrapperMessage)(nil)
  254. func (w *wrapperMessage) Wrap(inner proto.Message) error {
  255. switch inner := inner.(type) {
  256. case *p2ptest.Message:
  257. w.Message.Value = fmt.Sprintf("wrap:%v", inner.Value)
  258. case *wrapperMessage:
  259. *w = *inner
  260. default:
  261. return fmt.Errorf("invalid message type %T", inner)
  262. }
  263. return nil
  264. }
  265. func (w *wrapperMessage) Unwrap() (proto.Message, error) {
  266. return &p2ptest.Message{Value: fmt.Sprintf("unwrap:%v", w.Message.Value)}, nil
  267. }
  268. func TestRouter_Channel_Error(t *testing.T) {
  269. t.Cleanup(leaktest.Check(t))
  270. ctx, cancel := context.WithCancel(context.Background())
  271. defer cancel()
  272. // Create a test network and open a channel on all nodes.
  273. network := p2ptest.MakeNetwork(ctx, t, p2ptest.NetworkOptions{NumNodes: 3})
  274. network.Start(ctx, t)
  275. ids := network.NodeIDs()
  276. aID, bID := ids[0], ids[1]
  277. channels := network.MakeChannels(ctx, t, chDesc)
  278. a := channels[aID]
  279. // Erroring b should cause it to be disconnected. It will reconnect shortly after.
  280. sub := network.Nodes[aID].MakePeerUpdates(ctx, t)
  281. p2ptest.RequireError(ctx, t, a, p2p.PeerError{NodeID: bID, Err: errors.New("boom")})
  282. p2ptest.RequireUpdates(t, sub, []p2p.PeerUpdate{
  283. {NodeID: bID, Status: p2p.PeerStatusDown},
  284. {NodeID: bID, Status: p2p.PeerStatusUp},
  285. })
  286. }
  287. func TestRouter_AcceptPeers(t *testing.T) {
  288. testcases := map[string]struct {
  289. peerInfo types.NodeInfo
  290. peerKey crypto.PubKey
  291. ok bool
  292. }{
  293. "valid handshake": {peerInfo, peerKey.PubKey(), true},
  294. "empty handshake": {types.NodeInfo{}, nil, false},
  295. "invalid key": {peerInfo, selfKey.PubKey(), false},
  296. "self handshake": {selfInfo, selfKey.PubKey(), false},
  297. "incompatible peer": {
  298. types.NodeInfo{
  299. NodeID: peerID,
  300. ListenAddr: "0.0.0.0:0",
  301. Network: "other-network",
  302. Moniker: string(peerID),
  303. },
  304. peerKey.PubKey(),
  305. false,
  306. },
  307. }
  308. bctx, bcancel := context.WithCancel(context.Background())
  309. defer bcancel()
  310. for name, tc := range testcases {
  311. tc := tc
  312. t.Run(name, func(t *testing.T) {
  313. ctx, cancel := context.WithCancel(bctx)
  314. defer cancel()
  315. t.Cleanup(leaktest.Check(t))
  316. // Set up a mock transport that handshakes.
  317. closer := tmsync.NewCloser()
  318. mockConnection := &mocks.Connection{}
  319. mockConnection.On("String").Maybe().Return("mock")
  320. mockConnection.On("Handshake", mock.Anything, selfInfo, selfKey).
  321. Return(tc.peerInfo, tc.peerKey, nil)
  322. mockConnection.On("Close").Run(func(_ mock.Arguments) { closer.Close() }).Return(nil).Maybe()
  323. mockConnection.On("RemoteEndpoint").Return(p2p.Endpoint{})
  324. if tc.ok {
  325. mockConnection.On("ReceiveMessage", mock.Anything).Return(chID, nil, io.EOF).Maybe()
  326. }
  327. mockTransport := &mocks.Transport{}
  328. mockTransport.On("String").Maybe().Return("mock")
  329. mockTransport.On("Protocols").Return([]p2p.Protocol{"mock"})
  330. mockTransport.On("Close").Return(nil).Maybe()
  331. mockTransport.On("Accept", mock.Anything).Once().Return(mockConnection, nil)
  332. mockTransport.On("Accept", mock.Anything).Maybe().Return(nil, io.EOF)
  333. // Set up and start the router.
  334. peerManager, err := p2p.NewPeerManager(selfID, dbm.NewMemDB(), p2p.PeerManagerOptions{})
  335. require.NoError(t, err)
  336. sub := peerManager.Subscribe(ctx)
  337. router, err := p2p.NewRouter(
  338. ctx,
  339. log.TestingLogger(),
  340. p2p.NopMetrics(),
  341. selfInfo,
  342. selfKey,
  343. peerManager,
  344. []p2p.Transport{mockTransport},
  345. nil,
  346. p2p.RouterOptions{},
  347. )
  348. require.NoError(t, err)
  349. require.NoError(t, router.Start(ctx))
  350. if tc.ok {
  351. p2ptest.RequireUpdate(t, sub, p2p.PeerUpdate{
  352. NodeID: tc.peerInfo.NodeID,
  353. Status: p2p.PeerStatusUp,
  354. })
  355. // force a context switch so that the
  356. // connection is handled.
  357. time.Sleep(time.Millisecond)
  358. } else {
  359. select {
  360. case <-closer.Done():
  361. case <-time.After(100 * time.Millisecond):
  362. require.Fail(t, "connection not closed")
  363. }
  364. }
  365. require.NoError(t, router.Stop())
  366. mockTransport.AssertExpectations(t)
  367. mockConnection.AssertExpectations(t)
  368. })
  369. }
  370. }
  371. func TestRouter_AcceptPeers_Error(t *testing.T) {
  372. t.Cleanup(leaktest.Check(t))
  373. ctx, cancel := context.WithCancel(context.Background())
  374. defer cancel()
  375. // Set up a mock transport that returns an error, which should prevent
  376. // the router from calling Accept again.
  377. mockTransport := &mocks.Transport{}
  378. mockTransport.On("String").Maybe().Return("mock")
  379. mockTransport.On("Protocols").Return([]p2p.Protocol{"mock"})
  380. mockTransport.On("Accept", mock.Anything).Once().Return(nil, errors.New("boom"))
  381. mockTransport.On("Close").Return(nil)
  382. // Set up and start the router.
  383. peerManager, err := p2p.NewPeerManager(selfID, dbm.NewMemDB(), p2p.PeerManagerOptions{})
  384. require.NoError(t, err)
  385. router, err := p2p.NewRouter(
  386. ctx,
  387. log.TestingLogger(),
  388. p2p.NopMetrics(),
  389. selfInfo,
  390. selfKey,
  391. peerManager,
  392. []p2p.Transport{mockTransport},
  393. nil,
  394. p2p.RouterOptions{},
  395. )
  396. require.NoError(t, err)
  397. require.NoError(t, router.Start(ctx))
  398. time.Sleep(time.Second)
  399. require.NoError(t, router.Stop())
  400. mockTransport.AssertExpectations(t)
  401. }
  402. func TestRouter_AcceptPeers_ErrorEOF(t *testing.T) {
  403. t.Cleanup(leaktest.Check(t))
  404. ctx, cancel := context.WithCancel(context.Background())
  405. defer cancel()
  406. // Set up a mock transport that returns io.EOF once, which should prevent
  407. // the router from calling Accept again.
  408. mockTransport := &mocks.Transport{}
  409. mockTransport.On("String").Maybe().Return("mock")
  410. mockTransport.On("Protocols").Return([]p2p.Protocol{"mock"})
  411. mockTransport.On("Accept", mock.Anything).Once().Return(nil, io.EOF)
  412. mockTransport.On("Close").Return(nil)
  413. // Set up and start the router.
  414. peerManager, err := p2p.NewPeerManager(selfID, dbm.NewMemDB(), p2p.PeerManagerOptions{})
  415. require.NoError(t, err)
  416. router, err := p2p.NewRouter(
  417. ctx,
  418. log.TestingLogger(),
  419. p2p.NopMetrics(),
  420. selfInfo,
  421. selfKey,
  422. peerManager,
  423. []p2p.Transport{mockTransport},
  424. nil,
  425. p2p.RouterOptions{},
  426. )
  427. require.NoError(t, err)
  428. require.NoError(t, router.Start(ctx))
  429. time.Sleep(time.Second)
  430. require.NoError(t, router.Stop())
  431. mockTransport.AssertExpectations(t)
  432. }
  433. func TestRouter_AcceptPeers_HeadOfLineBlocking(t *testing.T) {
  434. t.Cleanup(leaktest.Check(t))
  435. ctx, cancel := context.WithCancel(context.Background())
  436. defer cancel()
  437. // Set up a mock transport that returns a connection that blocks during the
  438. // handshake. It should be able to accept several of these in parallel, i.e.
  439. // a single connection can't halt other connections being accepted.
  440. acceptCh := make(chan bool, 3)
  441. closeCh := make(chan time.Time)
  442. mockConnection := &mocks.Connection{}
  443. mockConnection.On("String").Maybe().Return("mock")
  444. mockConnection.On("Handshake", mock.Anything, selfInfo, selfKey).
  445. WaitUntil(closeCh).Return(types.NodeInfo{}, nil, io.EOF)
  446. mockConnection.On("Close").Return(nil)
  447. mockConnection.On("RemoteEndpoint").Return(p2p.Endpoint{})
  448. mockTransport := &mocks.Transport{}
  449. mockTransport.On("String").Maybe().Return("mock")
  450. mockTransport.On("Protocols").Return([]p2p.Protocol{"mock"})
  451. mockTransport.On("Close").Return(nil)
  452. mockTransport.On("Accept", mock.Anything).Times(3).Run(func(_ mock.Arguments) {
  453. acceptCh <- true
  454. }).Return(mockConnection, nil)
  455. mockTransport.On("Accept", mock.Anything).Once().Return(nil, io.EOF)
  456. // Set up and start the router.
  457. peerManager, err := p2p.NewPeerManager(selfID, dbm.NewMemDB(), p2p.PeerManagerOptions{})
  458. require.NoError(t, err)
  459. router, err := p2p.NewRouter(
  460. ctx,
  461. log.TestingLogger(),
  462. p2p.NopMetrics(),
  463. selfInfo,
  464. selfKey,
  465. peerManager,
  466. []p2p.Transport{mockTransport},
  467. nil,
  468. p2p.RouterOptions{},
  469. )
  470. require.NoError(t, err)
  471. require.NoError(t, router.Start(ctx))
  472. require.Eventually(t, func() bool {
  473. return len(acceptCh) == 3
  474. }, time.Second, 10*time.Millisecond, "num", len(acceptCh))
  475. close(closeCh)
  476. time.Sleep(100 * time.Millisecond)
  477. require.NoError(t, router.Stop())
  478. mockTransport.AssertExpectations(t)
  479. mockConnection.AssertExpectations(t)
  480. }
  481. func TestRouter_DialPeers(t *testing.T) {
  482. testcases := map[string]struct {
  483. dialID types.NodeID
  484. peerInfo types.NodeInfo
  485. peerKey crypto.PubKey
  486. dialErr error
  487. ok bool
  488. }{
  489. "valid dial": {peerInfo.NodeID, peerInfo, peerKey.PubKey(), nil, true},
  490. "empty handshake": {peerInfo.NodeID, types.NodeInfo{}, nil, nil, false},
  491. "invalid key": {peerInfo.NodeID, peerInfo, selfKey.PubKey(), nil, false},
  492. "unexpected node ID": {peerInfo.NodeID, selfInfo, selfKey.PubKey(), nil, false},
  493. "dial error": {peerInfo.NodeID, peerInfo, peerKey.PubKey(), errors.New("boom"), false},
  494. "incompatible peer": {
  495. peerInfo.NodeID,
  496. types.NodeInfo{
  497. NodeID: peerID,
  498. ListenAddr: "0.0.0.0:0",
  499. Network: "other-network",
  500. Moniker: string(peerID),
  501. },
  502. peerKey.PubKey(),
  503. nil,
  504. false,
  505. },
  506. }
  507. bctx, bcancel := context.WithCancel(context.Background())
  508. defer bcancel()
  509. for name, tc := range testcases {
  510. tc := tc
  511. t.Run(name, func(t *testing.T) {
  512. t.Cleanup(leaktest.Check(t))
  513. ctx, cancel := context.WithCancel(bctx)
  514. defer cancel()
  515. address := p2p.NodeAddress{Protocol: "mock", NodeID: tc.dialID}
  516. endpoint := p2p.Endpoint{Protocol: "mock", Path: string(tc.dialID)}
  517. // Set up a mock transport that handshakes.
  518. closer := tmsync.NewCloser()
  519. mockConnection := &mocks.Connection{}
  520. mockConnection.On("String").Maybe().Return("mock")
  521. if tc.dialErr == nil {
  522. mockConnection.On("Handshake", mock.Anything, selfInfo, selfKey).
  523. Return(tc.peerInfo, tc.peerKey, nil)
  524. mockConnection.On("Close").Run(func(_ mock.Arguments) { closer.Close() }).Return(nil).Maybe()
  525. }
  526. if tc.ok {
  527. mockConnection.On("ReceiveMessage", mock.Anything).Return(chID, nil, io.EOF).Maybe()
  528. }
  529. mockTransport := &mocks.Transport{}
  530. mockTransport.On("String").Maybe().Return("mock")
  531. mockTransport.On("Protocols").Return([]p2p.Protocol{"mock"})
  532. mockTransport.On("Close").Return(nil).Maybe()
  533. mockTransport.On("Accept", mock.Anything).Maybe().Return(nil, io.EOF)
  534. if tc.dialErr == nil {
  535. mockTransport.On("Dial", mock.Anything, endpoint).Once().Return(mockConnection, nil)
  536. // This handles the retry when a dialed connection gets closed after ReceiveMessage
  537. // returns io.EOF above.
  538. mockTransport.On("Dial", mock.Anything, endpoint).Maybe().Return(nil, io.EOF)
  539. } else {
  540. mockTransport.On("Dial", mock.Anything, endpoint).Once().
  541. Run(func(_ mock.Arguments) { closer.Close() }).
  542. Return(nil, tc.dialErr)
  543. }
  544. // Set up and start the router.
  545. peerManager, err := p2p.NewPeerManager(selfID, dbm.NewMemDB(), p2p.PeerManagerOptions{})
  546. require.NoError(t, err)
  547. added, err := peerManager.Add(address)
  548. require.NoError(t, err)
  549. require.True(t, added)
  550. sub := peerManager.Subscribe(ctx)
  551. router, err := p2p.NewRouter(
  552. ctx,
  553. log.TestingLogger(),
  554. p2p.NopMetrics(),
  555. selfInfo,
  556. selfKey,
  557. peerManager,
  558. []p2p.Transport{mockTransport},
  559. nil,
  560. p2p.RouterOptions{},
  561. )
  562. require.NoError(t, err)
  563. require.NoError(t, router.Start(ctx))
  564. if tc.ok {
  565. p2ptest.RequireUpdate(t, sub, p2p.PeerUpdate{
  566. NodeID: tc.peerInfo.NodeID,
  567. Status: p2p.PeerStatusUp,
  568. })
  569. // force a context switch so that the
  570. // connection is handled.
  571. time.Sleep(time.Millisecond)
  572. } else {
  573. select {
  574. case <-closer.Done():
  575. case <-time.After(100 * time.Millisecond):
  576. require.Fail(t, "connection not closed")
  577. }
  578. }
  579. require.NoError(t, router.Stop())
  580. mockTransport.AssertExpectations(t)
  581. mockConnection.AssertExpectations(t)
  582. })
  583. }
  584. }
  585. func TestRouter_DialPeers_Parallel(t *testing.T) {
  586. t.Cleanup(leaktest.Check(t))
  587. ctx, cancel := context.WithCancel(context.Background())
  588. defer cancel()
  589. a := p2p.NodeAddress{Protocol: "mock", NodeID: types.NodeID(strings.Repeat("a", 40))}
  590. b := p2p.NodeAddress{Protocol: "mock", NodeID: types.NodeID(strings.Repeat("b", 40))}
  591. c := p2p.NodeAddress{Protocol: "mock", NodeID: types.NodeID(strings.Repeat("c", 40))}
  592. // Set up a mock transport that returns a connection that blocks during the
  593. // handshake. It should dial all peers in parallel.
  594. dialCh := make(chan bool, 3)
  595. closeCh := make(chan time.Time)
  596. mockConnection := &mocks.Connection{}
  597. mockConnection.On("String").Maybe().Return("mock")
  598. mockConnection.On("Handshake", mock.Anything, selfInfo, selfKey).
  599. WaitUntil(closeCh).Return(types.NodeInfo{}, nil, io.EOF)
  600. mockConnection.On("Close").Return(nil)
  601. mockTransport := &mocks.Transport{}
  602. mockTransport.On("String").Maybe().Return("mock")
  603. mockTransport.On("Protocols").Return([]p2p.Protocol{"mock"})
  604. mockTransport.On("Close").Return(nil)
  605. mockTransport.On("Accept", mock.Anything).Once().Return(nil, io.EOF)
  606. for _, address := range []p2p.NodeAddress{a, b, c} {
  607. endpoint := p2p.Endpoint{Protocol: address.Protocol, Path: string(address.NodeID)}
  608. mockTransport.On("Dial", mock.Anything, endpoint).Run(func(_ mock.Arguments) {
  609. dialCh <- true
  610. }).Return(mockConnection, nil)
  611. }
  612. // Set up and start the router.
  613. peerManager, err := p2p.NewPeerManager(selfID, dbm.NewMemDB(), p2p.PeerManagerOptions{})
  614. require.NoError(t, err)
  615. added, err := peerManager.Add(a)
  616. require.NoError(t, err)
  617. require.True(t, added)
  618. added, err = peerManager.Add(b)
  619. require.NoError(t, err)
  620. require.True(t, added)
  621. added, err = peerManager.Add(c)
  622. require.NoError(t, err)
  623. require.True(t, added)
  624. router, err := p2p.NewRouter(
  625. ctx,
  626. log.TestingLogger(),
  627. p2p.NopMetrics(),
  628. selfInfo,
  629. selfKey,
  630. peerManager,
  631. []p2p.Transport{mockTransport},
  632. nil,
  633. p2p.RouterOptions{
  634. DialSleep: func(_ context.Context) {},
  635. NumConcurrentDials: func() int {
  636. ncpu := runtime.NumCPU()
  637. if ncpu <= 3 {
  638. return 3
  639. }
  640. return ncpu
  641. },
  642. },
  643. )
  644. require.NoError(t, err)
  645. require.NoError(t, router.Start(ctx))
  646. require.Eventually(t,
  647. func() bool {
  648. return len(dialCh) == 3
  649. },
  650. 5*time.Second,
  651. 100*time.Millisecond,
  652. "reached %d rather than 3", len(dialCh))
  653. close(closeCh)
  654. time.Sleep(500 * time.Millisecond)
  655. require.NoError(t, router.Stop())
  656. mockTransport.AssertExpectations(t)
  657. mockConnection.AssertExpectations(t)
  658. }
  659. func TestRouter_EvictPeers(t *testing.T) {
  660. t.Cleanup(leaktest.Check(t))
  661. ctx, cancel := context.WithCancel(context.Background())
  662. defer cancel()
  663. // Set up a mock transport that we can evict.
  664. closeCh := make(chan time.Time)
  665. closeOnce := sync.Once{}
  666. mockConnection := &mocks.Connection{}
  667. mockConnection.On("String").Maybe().Return("mock")
  668. mockConnection.On("Handshake", mock.Anything, selfInfo, selfKey).
  669. Return(peerInfo, peerKey.PubKey(), nil)
  670. mockConnection.On("ReceiveMessage", mock.Anything).WaitUntil(closeCh).Return(chID, nil, io.EOF)
  671. mockConnection.On("RemoteEndpoint").Return(p2p.Endpoint{})
  672. mockConnection.On("Close").Run(func(_ mock.Arguments) {
  673. closeOnce.Do(func() {
  674. close(closeCh)
  675. })
  676. }).Return(nil)
  677. mockTransport := &mocks.Transport{}
  678. mockTransport.On("String").Maybe().Return("mock")
  679. mockTransport.On("Protocols").Return([]p2p.Protocol{"mock"})
  680. mockTransport.On("Close").Return(nil)
  681. mockTransport.On("Accept", mock.Anything).Once().Return(mockConnection, nil)
  682. mockTransport.On("Accept", mock.Anything).Maybe().Return(nil, io.EOF)
  683. // Set up and start the router.
  684. peerManager, err := p2p.NewPeerManager(selfID, dbm.NewMemDB(), p2p.PeerManagerOptions{})
  685. require.NoError(t, err)
  686. sub := peerManager.Subscribe(ctx)
  687. router, err := p2p.NewRouter(
  688. ctx,
  689. log.TestingLogger(),
  690. p2p.NopMetrics(),
  691. selfInfo,
  692. selfKey,
  693. peerManager,
  694. []p2p.Transport{mockTransport},
  695. nil,
  696. p2p.RouterOptions{},
  697. )
  698. require.NoError(t, err)
  699. require.NoError(t, router.Start(ctx))
  700. // Wait for the mock peer to connect, then evict it by reporting an error.
  701. p2ptest.RequireUpdate(t, sub, p2p.PeerUpdate{
  702. NodeID: peerInfo.NodeID,
  703. Status: p2p.PeerStatusUp,
  704. })
  705. peerManager.Errored(peerInfo.NodeID, errors.New("boom"))
  706. p2ptest.RequireUpdate(t, sub, p2p.PeerUpdate{
  707. NodeID: peerInfo.NodeID,
  708. Status: p2p.PeerStatusDown,
  709. })
  710. require.NoError(t, router.Stop())
  711. mockTransport.AssertExpectations(t)
  712. mockConnection.AssertExpectations(t)
  713. }
  714. func TestRouter_ChannelCompatability(t *testing.T) {
  715. t.Cleanup(leaktest.Check(t))
  716. ctx, cancel := context.WithCancel(context.Background())
  717. defer cancel()
  718. incompatiblePeer := types.NodeInfo{
  719. NodeID: peerID,
  720. ListenAddr: "0.0.0.0:0",
  721. Network: "test",
  722. Moniker: string(peerID),
  723. Channels: []byte{0x03},
  724. }
  725. mockConnection := &mocks.Connection{}
  726. mockConnection.On("String").Maybe().Return("mock")
  727. mockConnection.On("Handshake", mock.Anything, selfInfo, selfKey).
  728. Return(incompatiblePeer, peerKey.PubKey(), nil)
  729. mockConnection.On("RemoteEndpoint").Return(p2p.Endpoint{})
  730. mockConnection.On("Close").Return(nil)
  731. mockTransport := &mocks.Transport{}
  732. mockTransport.On("String").Maybe().Return("mock")
  733. mockTransport.On("Protocols").Return([]p2p.Protocol{"mock"})
  734. mockTransport.On("Close").Return(nil)
  735. mockTransport.On("Accept", mock.Anything).Once().Return(mockConnection, nil)
  736. mockTransport.On("Accept", mock.Anything).Once().Return(nil, io.EOF)
  737. // Set up and start the router.
  738. peerManager, err := p2p.NewPeerManager(selfID, dbm.NewMemDB(), p2p.PeerManagerOptions{})
  739. require.NoError(t, err)
  740. router, err := p2p.NewRouter(
  741. ctx,
  742. log.TestingLogger(),
  743. p2p.NopMetrics(),
  744. selfInfo,
  745. selfKey,
  746. peerManager,
  747. []p2p.Transport{mockTransport},
  748. nil,
  749. p2p.RouterOptions{},
  750. )
  751. require.NoError(t, err)
  752. require.NoError(t, router.Start(ctx))
  753. time.Sleep(1 * time.Second)
  754. require.NoError(t, router.Stop())
  755. require.Empty(t, peerManager.Peers())
  756. mockConnection.AssertExpectations(t)
  757. mockTransport.AssertExpectations(t)
  758. }
  759. func TestRouter_DontSendOnInvalidChannel(t *testing.T) {
  760. t.Cleanup(leaktest.Check(t))
  761. ctx, cancel := context.WithCancel(context.Background())
  762. defer cancel()
  763. peer := types.NodeInfo{
  764. NodeID: peerID,
  765. ListenAddr: "0.0.0.0:0",
  766. Network: "test",
  767. Moniker: string(peerID),
  768. Channels: []byte{0x02},
  769. }
  770. mockConnection := &mocks.Connection{}
  771. mockConnection.On("String").Maybe().Return("mock")
  772. mockConnection.On("Handshake", mock.Anything, selfInfo, selfKey).
  773. Return(peer, peerKey.PubKey(), nil)
  774. mockConnection.On("RemoteEndpoint").Return(p2p.Endpoint{})
  775. mockConnection.On("Close").Return(nil)
  776. mockConnection.On("ReceiveMessage", mock.Anything).Return(chID, nil, io.EOF)
  777. mockTransport := &mocks.Transport{}
  778. mockTransport.On("AddChannelDescriptors", mock.Anything).Return()
  779. mockTransport.On("String").Maybe().Return("mock")
  780. mockTransport.On("Protocols").Return([]p2p.Protocol{"mock"})
  781. mockTransport.On("Close").Return(nil)
  782. mockTransport.On("Accept", mock.Anything).Once().Return(mockConnection, nil)
  783. mockTransport.On("Accept", mock.Anything).Maybe().Return(nil, io.EOF)
  784. // Set up and start the router.
  785. peerManager, err := p2p.NewPeerManager(selfID, dbm.NewMemDB(), p2p.PeerManagerOptions{})
  786. require.NoError(t, err)
  787. sub := peerManager.Subscribe(ctx)
  788. router, err := p2p.NewRouter(
  789. ctx,
  790. log.TestingLogger(),
  791. p2p.NopMetrics(),
  792. selfInfo,
  793. selfKey,
  794. peerManager,
  795. []p2p.Transport{mockTransport},
  796. nil,
  797. p2p.RouterOptions{},
  798. )
  799. require.NoError(t, err)
  800. require.NoError(t, router.Start(ctx))
  801. p2ptest.RequireUpdate(t, sub, p2p.PeerUpdate{
  802. NodeID: peerInfo.NodeID,
  803. Status: p2p.PeerStatusUp,
  804. })
  805. channel, err := router.OpenChannel(ctx, chDesc)
  806. require.NoError(t, err)
  807. require.NoError(t, channel.Send(ctx, p2p.Envelope{
  808. To: peer.NodeID,
  809. Message: &p2ptest.Message{Value: "Hi"},
  810. }))
  811. require.NoError(t, router.Stop())
  812. mockTransport.AssertExpectations(t)
  813. }