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.

1796 lines
59 KiB

  1. package p2p_test
  2. import (
  3. "context"
  4. "errors"
  5. "strings"
  6. "testing"
  7. "time"
  8. "github.com/fortytw2/leaktest"
  9. "github.com/stretchr/testify/require"
  10. dbm "github.com/tendermint/tm-db"
  11. "github.com/tendermint/tendermint/p2p"
  12. )
  13. // FIXME: We should probably have some randomized property-based tests for the
  14. // PeerManager too, which runs a bunch of random operations with random peers
  15. // and ensures certain invariants always hold. The logic can be complex, with
  16. // many interactions, and it's hard to cover all scenarios with handwritten
  17. // tests.
  18. func TestPeerManagerOptions_Validate(t *testing.T) {
  19. nodeID := p2p.NodeID("00112233445566778899aabbccddeeff00112233")
  20. testcases := map[string]struct {
  21. options p2p.PeerManagerOptions
  22. ok bool
  23. }{
  24. "zero options is valid": {p2p.PeerManagerOptions{}, true},
  25. // PersistentPeers
  26. "valid PersistentPeers NodeID": {p2p.PeerManagerOptions{
  27. PersistentPeers: []p2p.NodeID{"00112233445566778899aabbccddeeff00112233"},
  28. }, true},
  29. "invalid PersistentPeers NodeID": {p2p.PeerManagerOptions{
  30. PersistentPeers: []p2p.NodeID{"foo"},
  31. }, false},
  32. "uppercase PersistentPeers NodeID": {p2p.PeerManagerOptions{
  33. PersistentPeers: []p2p.NodeID{"00112233445566778899AABBCCDDEEFF00112233"},
  34. }, false},
  35. "PersistentPeers at MaxConnected": {p2p.PeerManagerOptions{
  36. PersistentPeers: []p2p.NodeID{nodeID, nodeID, nodeID},
  37. MaxConnected: 3,
  38. }, true},
  39. "PersistentPeers above MaxConnected": {p2p.PeerManagerOptions{
  40. PersistentPeers: []p2p.NodeID{nodeID, nodeID, nodeID},
  41. MaxConnected: 2,
  42. }, false},
  43. "PersistentPeers above MaxConnected below MaxConnectedUpgrade": {p2p.PeerManagerOptions{
  44. PersistentPeers: []p2p.NodeID{nodeID, nodeID, nodeID},
  45. MaxConnected: 2,
  46. MaxConnectedUpgrade: 2,
  47. }, false},
  48. // MaxPeers
  49. "MaxPeers without MaxConnected": {p2p.PeerManagerOptions{
  50. MaxPeers: 3,
  51. }, false},
  52. "MaxPeers below MaxConnected+MaxConnectedUpgrade": {p2p.PeerManagerOptions{
  53. MaxPeers: 2,
  54. MaxConnected: 2,
  55. MaxConnectedUpgrade: 1,
  56. }, false},
  57. "MaxPeers at MaxConnected+MaxConnectedUpgrade": {p2p.PeerManagerOptions{
  58. MaxPeers: 3,
  59. MaxConnected: 2,
  60. MaxConnectedUpgrade: 1,
  61. }, true},
  62. // MaxRetryTime
  63. "MaxRetryTime below MinRetryTime": {p2p.PeerManagerOptions{
  64. MinRetryTime: 7 * time.Second,
  65. MaxRetryTime: 5 * time.Second,
  66. }, false},
  67. "MaxRetryTime at MinRetryTime": {p2p.PeerManagerOptions{
  68. MinRetryTime: 5 * time.Second,
  69. MaxRetryTime: 5 * time.Second,
  70. }, true},
  71. "MaxRetryTime without MinRetryTime": {p2p.PeerManagerOptions{
  72. MaxRetryTime: 5 * time.Second,
  73. }, false},
  74. // MaxRetryTimePersistent
  75. "MaxRetryTimePersistent below MinRetryTime": {p2p.PeerManagerOptions{
  76. MinRetryTime: 7 * time.Second,
  77. MaxRetryTimePersistent: 5 * time.Second,
  78. }, false},
  79. "MaxRetryTimePersistent at MinRetryTime": {p2p.PeerManagerOptions{
  80. MinRetryTime: 5 * time.Second,
  81. MaxRetryTimePersistent: 5 * time.Second,
  82. }, true},
  83. "MaxRetryTimePersistent without MinRetryTime": {p2p.PeerManagerOptions{
  84. MaxRetryTimePersistent: 5 * time.Second,
  85. }, false},
  86. }
  87. for name, tc := range testcases {
  88. tc := tc
  89. t.Run(name, func(t *testing.T) {
  90. err := tc.options.Validate()
  91. if tc.ok {
  92. require.NoError(t, err)
  93. } else {
  94. require.Error(t, err)
  95. }
  96. })
  97. }
  98. }
  99. func TestNewPeerManager(t *testing.T) {
  100. // Zero options should be valid.
  101. _, err := p2p.NewPeerManager(selfID, dbm.NewMemDB(), p2p.PeerManagerOptions{})
  102. require.NoError(t, err)
  103. // Invalid options should error.
  104. _, err = p2p.NewPeerManager(selfID, dbm.NewMemDB(), p2p.PeerManagerOptions{
  105. PersistentPeers: []p2p.NodeID{"foo"},
  106. })
  107. require.Error(t, err)
  108. // Invalid database should error.
  109. _, err = p2p.NewPeerManager(selfID, nil, p2p.PeerManagerOptions{})
  110. require.Error(t, err)
  111. // Empty self ID should error.
  112. _, err = p2p.NewPeerManager("", nil, p2p.PeerManagerOptions{})
  113. require.Error(t, err)
  114. }
  115. func TestNewPeerManager_Persistence(t *testing.T) {
  116. aID := p2p.NodeID(strings.Repeat("a", 40))
  117. aAddresses := []p2p.NodeAddress{
  118. {Protocol: "tcp", NodeID: aID, Hostname: "127.0.0.1", Port: 26657, Path: "/path"},
  119. {Protocol: "memory", NodeID: aID},
  120. }
  121. bID := p2p.NodeID(strings.Repeat("b", 40))
  122. bAddresses := []p2p.NodeAddress{
  123. {Protocol: "tcp", NodeID: bID, Hostname: "b10c::1", Port: 26657, Path: "/path"},
  124. {Protocol: "memory", NodeID: bID},
  125. }
  126. cID := p2p.NodeID(strings.Repeat("c", 40))
  127. cAddresses := []p2p.NodeAddress{
  128. {Protocol: "tcp", NodeID: cID, Hostname: "host.domain", Port: 80},
  129. {Protocol: "memory", NodeID: cID},
  130. }
  131. // Create an initial peer manager and add the peers.
  132. db := dbm.NewMemDB()
  133. peerManager, err := p2p.NewPeerManager(selfID, db, p2p.PeerManagerOptions{
  134. PersistentPeers: []p2p.NodeID{aID},
  135. PeerScores: map[p2p.NodeID]p2p.PeerScore{bID: 1},
  136. })
  137. require.NoError(t, err)
  138. defer peerManager.Close()
  139. for _, addr := range append(append(aAddresses, bAddresses...), cAddresses...) {
  140. added, err := peerManager.Add(addr)
  141. require.NoError(t, err)
  142. require.True(t, added)
  143. }
  144. require.ElementsMatch(t, aAddresses, peerManager.Addresses(aID))
  145. require.ElementsMatch(t, bAddresses, peerManager.Addresses(bID))
  146. require.ElementsMatch(t, cAddresses, peerManager.Addresses(cID))
  147. require.Equal(t, map[p2p.NodeID]p2p.PeerScore{
  148. aID: p2p.PeerScorePersistent,
  149. bID: 1,
  150. cID: 0,
  151. }, peerManager.Scores())
  152. peerManager.Close()
  153. // Creating a new peer manager with the same database should retain the
  154. // peers, but they should have updated scores from the new PersistentPeers
  155. // configuration.
  156. peerManager, err = p2p.NewPeerManager(selfID, db, p2p.PeerManagerOptions{
  157. PersistentPeers: []p2p.NodeID{bID},
  158. PeerScores: map[p2p.NodeID]p2p.PeerScore{cID: 1},
  159. })
  160. require.NoError(t, err)
  161. defer peerManager.Close()
  162. require.ElementsMatch(t, aAddresses, peerManager.Addresses(aID))
  163. require.ElementsMatch(t, bAddresses, peerManager.Addresses(bID))
  164. require.ElementsMatch(t, cAddresses, peerManager.Addresses(cID))
  165. require.Equal(t, map[p2p.NodeID]p2p.PeerScore{
  166. aID: 0,
  167. bID: p2p.PeerScorePersistent,
  168. cID: 1,
  169. }, peerManager.Scores())
  170. }
  171. func TestNewPeerManager_SelfIDChange(t *testing.T) {
  172. a := p2p.NodeAddress{Protocol: "memory", NodeID: p2p.NodeID(strings.Repeat("a", 40))}
  173. b := p2p.NodeAddress{Protocol: "memory", NodeID: p2p.NodeID(strings.Repeat("b", 40))}
  174. db := dbm.NewMemDB()
  175. peerManager, err := p2p.NewPeerManager(selfID, db, p2p.PeerManagerOptions{})
  176. require.NoError(t, err)
  177. added, err := peerManager.Add(a)
  178. require.NoError(t, err)
  179. require.True(t, added)
  180. added, err = peerManager.Add(b)
  181. require.NoError(t, err)
  182. require.True(t, added)
  183. require.ElementsMatch(t, []p2p.NodeID{a.NodeID, b.NodeID}, peerManager.Peers())
  184. peerManager.Close()
  185. // If we change our selfID to one of the peers in the peer store, it
  186. // should be removed from the store.
  187. peerManager, err = p2p.NewPeerManager(a.NodeID, db, p2p.PeerManagerOptions{})
  188. require.NoError(t, err)
  189. require.Equal(t, []p2p.NodeID{b.NodeID}, peerManager.Peers())
  190. }
  191. func TestPeerManager_Add(t *testing.T) {
  192. aID := p2p.NodeID(strings.Repeat("a", 40))
  193. bID := p2p.NodeID(strings.Repeat("b", 40))
  194. cID := p2p.NodeID(strings.Repeat("c", 40))
  195. peerManager, err := p2p.NewPeerManager(selfID, dbm.NewMemDB(), p2p.PeerManagerOptions{
  196. PersistentPeers: []p2p.NodeID{aID, cID},
  197. MaxPeers: 2,
  198. MaxConnected: 2,
  199. })
  200. require.NoError(t, err)
  201. // Adding a couple of addresses should work.
  202. aAddresses := []p2p.NodeAddress{
  203. {Protocol: "tcp", NodeID: aID, Hostname: "localhost"},
  204. {Protocol: "memory", NodeID: aID},
  205. }
  206. for _, addr := range aAddresses {
  207. added, err := peerManager.Add(addr)
  208. require.NoError(t, err)
  209. require.True(t, added)
  210. }
  211. require.ElementsMatch(t, aAddresses, peerManager.Addresses(aID))
  212. // Adding a different peer should be fine.
  213. bAddress := p2p.NodeAddress{Protocol: "tcp", NodeID: bID, Hostname: "localhost"}
  214. added, err := peerManager.Add(bAddress)
  215. require.NoError(t, err)
  216. require.True(t, added)
  217. require.Equal(t, []p2p.NodeAddress{bAddress}, peerManager.Addresses(bID))
  218. require.ElementsMatch(t, aAddresses, peerManager.Addresses(aID))
  219. // Adding an existing address again should be a noop.
  220. added, err = peerManager.Add(aAddresses[0])
  221. require.NoError(t, err)
  222. require.False(t, added)
  223. require.ElementsMatch(t, aAddresses, peerManager.Addresses(aID))
  224. // Adding a third peer with MaxPeers=2 should cause bID, which is
  225. // the lowest-scored peer (not in PersistentPeers), to be removed.
  226. added, err = peerManager.Add(p2p.NodeAddress{
  227. Protocol: "tcp", NodeID: cID, Hostname: "localhost"})
  228. require.NoError(t, err)
  229. require.True(t, added)
  230. require.ElementsMatch(t, []p2p.NodeID{aID, cID}, peerManager.Peers())
  231. // Adding an invalid address should error.
  232. _, err = peerManager.Add(p2p.NodeAddress{Path: "foo"})
  233. require.Error(t, err)
  234. // Adding self should error
  235. _, err = peerManager.Add(p2p.NodeAddress{Protocol: "memory", NodeID: selfID})
  236. require.Error(t, err)
  237. }
  238. func TestPeerManager_DialNext(t *testing.T) {
  239. a := p2p.NodeAddress{Protocol: "memory", NodeID: p2p.NodeID(strings.Repeat("a", 40))}
  240. peerManager, err := p2p.NewPeerManager(selfID, dbm.NewMemDB(), p2p.PeerManagerOptions{})
  241. require.NoError(t, err)
  242. // Add an address. DialNext should return it.
  243. added, err := peerManager.Add(a)
  244. require.NoError(t, err)
  245. require.True(t, added)
  246. address, err := peerManager.DialNext(ctx)
  247. require.NoError(t, err)
  248. require.Equal(t, a, address)
  249. // Since there are no more undialed peers, the next call should block
  250. // until it times out.
  251. timeoutCtx, cancel := context.WithTimeout(ctx, 100*time.Millisecond)
  252. defer cancel()
  253. _, err = peerManager.DialNext(timeoutCtx)
  254. require.Error(t, err)
  255. require.Equal(t, context.DeadlineExceeded, err)
  256. }
  257. func TestPeerManager_DialNext_Retry(t *testing.T) {
  258. a := p2p.NodeAddress{Protocol: "memory", NodeID: p2p.NodeID(strings.Repeat("a", 40))}
  259. options := p2p.PeerManagerOptions{
  260. MinRetryTime: 100 * time.Millisecond,
  261. MaxRetryTime: 500 * time.Millisecond,
  262. }
  263. peerManager, err := p2p.NewPeerManager(selfID, dbm.NewMemDB(), options)
  264. require.NoError(t, err)
  265. added, err := peerManager.Add(a)
  266. require.NoError(t, err)
  267. require.True(t, added)
  268. // Do five dial retries (six dials total). The retry time should double for
  269. // each failure. At the forth retry, MaxRetryTime should kick in.
  270. ctx, cancel := context.WithTimeout(ctx, 5*time.Second)
  271. defer cancel()
  272. for i := 0; i <= 5; i++ {
  273. start := time.Now()
  274. dial, err := peerManager.DialNext(ctx)
  275. require.NoError(t, err)
  276. require.Equal(t, a, dial)
  277. elapsed := time.Since(start).Round(time.Millisecond)
  278. switch i {
  279. case 0:
  280. require.LessOrEqual(t, elapsed, options.MinRetryTime)
  281. case 1:
  282. require.GreaterOrEqual(t, elapsed, options.MinRetryTime)
  283. case 2:
  284. require.GreaterOrEqual(t, elapsed, 2*options.MinRetryTime)
  285. case 3:
  286. require.GreaterOrEqual(t, elapsed, 4*options.MinRetryTime)
  287. case 4, 5:
  288. require.GreaterOrEqual(t, elapsed, options.MaxRetryTime)
  289. require.LessOrEqual(t, elapsed, 8*options.MinRetryTime)
  290. default:
  291. require.Fail(t, "unexpected retry")
  292. }
  293. require.NoError(t, peerManager.DialFailed(a))
  294. }
  295. }
  296. func TestPeerManager_DialNext_WakeOnAdd(t *testing.T) {
  297. a := p2p.NodeAddress{Protocol: "memory", NodeID: p2p.NodeID(strings.Repeat("a", 40))}
  298. peerManager, err := p2p.NewPeerManager(selfID, dbm.NewMemDB(), p2p.PeerManagerOptions{})
  299. require.NoError(t, err)
  300. // Spawn a goroutine to add a peer after a delay.
  301. go func() {
  302. time.Sleep(200 * time.Millisecond)
  303. added, err := peerManager.Add(a)
  304. require.NoError(t, err)
  305. require.True(t, added)
  306. }()
  307. // This will block until peer is added above.
  308. ctx, cancel := context.WithTimeout(ctx, 3*time.Second)
  309. defer cancel()
  310. dial, err := peerManager.DialNext(ctx)
  311. require.NoError(t, err)
  312. require.Equal(t, a, dial)
  313. }
  314. func TestPeerManager_DialNext_WakeOnDialFailed(t *testing.T) {
  315. peerManager, err := p2p.NewPeerManager(selfID, dbm.NewMemDB(), p2p.PeerManagerOptions{
  316. MaxConnected: 1,
  317. })
  318. require.NoError(t, err)
  319. a := p2p.NodeAddress{Protocol: "memory", NodeID: p2p.NodeID(strings.Repeat("a", 40))}
  320. b := p2p.NodeAddress{Protocol: "memory", NodeID: p2p.NodeID(strings.Repeat("b", 40))}
  321. // Add and dial a.
  322. added, err := peerManager.Add(a)
  323. require.NoError(t, err)
  324. require.True(t, added)
  325. dial, err := peerManager.TryDialNext()
  326. require.NoError(t, err)
  327. require.Equal(t, a, dial)
  328. // Add b. We shouldn't be able to dial it, due to MaxConnected.
  329. added, err = peerManager.Add(b)
  330. require.NoError(t, err)
  331. require.True(t, added)
  332. dial, err = peerManager.TryDialNext()
  333. require.NoError(t, err)
  334. require.Zero(t, dial)
  335. // Spawn a goroutine to fail a's dial attempt.
  336. go func() {
  337. time.Sleep(200 * time.Millisecond)
  338. require.NoError(t, peerManager.DialFailed(a))
  339. }()
  340. // This should make b available for dialing (not a, retries are disabled).
  341. ctx, cancel := context.WithTimeout(ctx, 3*time.Second)
  342. defer cancel()
  343. dial, err = peerManager.DialNext(ctx)
  344. require.NoError(t, err)
  345. require.Equal(t, b, dial)
  346. }
  347. func TestPeerManager_DialNext_WakeOnDialFailedRetry(t *testing.T) {
  348. options := p2p.PeerManagerOptions{MinRetryTime: 200 * time.Millisecond}
  349. peerManager, err := p2p.NewPeerManager(selfID, dbm.NewMemDB(), options)
  350. require.NoError(t, err)
  351. a := p2p.NodeAddress{Protocol: "memory", NodeID: p2p.NodeID(strings.Repeat("a", 40))}
  352. // Add a, dial it, and mark it a failure. This will start a retry timer.
  353. added, err := peerManager.Add(a)
  354. require.NoError(t, err)
  355. require.True(t, added)
  356. dial, err := peerManager.TryDialNext()
  357. require.NoError(t, err)
  358. require.Equal(t, a, dial)
  359. require.NoError(t, peerManager.DialFailed(dial))
  360. failed := time.Now()
  361. // The retry timer should unblock DialNext and make a available again after
  362. // the retry time passes.
  363. ctx, cancel := context.WithTimeout(ctx, 3*time.Second)
  364. defer cancel()
  365. dial, err = peerManager.DialNext(ctx)
  366. require.NoError(t, err)
  367. require.Equal(t, a, dial)
  368. require.GreaterOrEqual(t, time.Since(failed), options.MinRetryTime)
  369. }
  370. func TestPeerManager_DialNext_WakeOnDisconnected(t *testing.T) {
  371. a := p2p.NodeAddress{Protocol: "memory", NodeID: p2p.NodeID(strings.Repeat("a", 40))}
  372. peerManager, err := p2p.NewPeerManager(selfID, dbm.NewMemDB(), p2p.PeerManagerOptions{})
  373. require.NoError(t, err)
  374. added, err := peerManager.Add(a)
  375. require.NoError(t, err)
  376. require.True(t, added)
  377. err = peerManager.Accepted(a.NodeID)
  378. require.NoError(t, err)
  379. dial, err := peerManager.TryDialNext()
  380. require.NoError(t, err)
  381. require.Zero(t, dial)
  382. go func() {
  383. time.Sleep(200 * time.Millisecond)
  384. require.NoError(t, peerManager.Disconnected(a.NodeID))
  385. }()
  386. ctx, cancel := context.WithTimeout(ctx, 3*time.Second)
  387. defer cancel()
  388. dial, err = peerManager.DialNext(ctx)
  389. require.NoError(t, err)
  390. require.Equal(t, a, dial)
  391. }
  392. func TestPeerManager_TryDialNext_MaxConnected(t *testing.T) {
  393. a := p2p.NodeAddress{Protocol: "memory", NodeID: p2p.NodeID(strings.Repeat("a", 40))}
  394. b := p2p.NodeAddress{Protocol: "memory", NodeID: p2p.NodeID(strings.Repeat("b", 40))}
  395. c := p2p.NodeAddress{Protocol: "memory", NodeID: p2p.NodeID(strings.Repeat("c", 40))}
  396. peerManager, err := p2p.NewPeerManager(selfID, dbm.NewMemDB(), p2p.PeerManagerOptions{
  397. MaxConnected: 2,
  398. })
  399. require.NoError(t, err)
  400. // Add a and connect to it.
  401. added, err := peerManager.Add(a)
  402. require.NoError(t, err)
  403. require.True(t, added)
  404. dial, err := peerManager.TryDialNext()
  405. require.NoError(t, err)
  406. require.Equal(t, a, dial)
  407. require.NoError(t, peerManager.Dialed(a))
  408. // Add b and start dialing it.
  409. added, err = peerManager.Add(b)
  410. require.NoError(t, err)
  411. require.True(t, added)
  412. dial, err = peerManager.TryDialNext()
  413. require.NoError(t, err)
  414. require.Equal(t, b, dial)
  415. // At this point, adding c will not allow dialing it.
  416. added, err = peerManager.Add(c)
  417. require.NoError(t, err)
  418. require.True(t, added)
  419. dial, err = peerManager.TryDialNext()
  420. require.NoError(t, err)
  421. require.Zero(t, dial)
  422. }
  423. func TestPeerManager_TryDialNext_MaxConnectedUpgrade(t *testing.T) {
  424. a := p2p.NodeAddress{Protocol: "memory", NodeID: p2p.NodeID(strings.Repeat("a", 40))}
  425. b := p2p.NodeAddress{Protocol: "memory", NodeID: p2p.NodeID(strings.Repeat("b", 40))}
  426. c := p2p.NodeAddress{Protocol: "memory", NodeID: p2p.NodeID(strings.Repeat("c", 40))}
  427. d := p2p.NodeAddress{Protocol: "memory", NodeID: p2p.NodeID(strings.Repeat("d", 40))}
  428. e := p2p.NodeAddress{Protocol: "memory", NodeID: p2p.NodeID(strings.Repeat("e", 40))}
  429. peerManager, err := p2p.NewPeerManager(selfID, dbm.NewMemDB(), p2p.PeerManagerOptions{
  430. PeerScores: map[p2p.NodeID]p2p.PeerScore{
  431. a.NodeID: 0,
  432. b.NodeID: 1,
  433. c.NodeID: 2,
  434. d.NodeID: 3,
  435. e.NodeID: 0,
  436. },
  437. PersistentPeers: []p2p.NodeID{c.NodeID, d.NodeID},
  438. MaxConnected: 2,
  439. MaxConnectedUpgrade: 1,
  440. })
  441. require.NoError(t, err)
  442. // Add a and connect to it.
  443. added, err := peerManager.Add(a)
  444. require.NoError(t, err)
  445. require.True(t, added)
  446. dial, err := peerManager.TryDialNext()
  447. require.NoError(t, err)
  448. require.Equal(t, a, dial)
  449. require.NoError(t, peerManager.Dialed(a))
  450. // Add b and start dialing it.
  451. added, err = peerManager.Add(b)
  452. require.NoError(t, err)
  453. require.True(t, added)
  454. dial, err = peerManager.TryDialNext()
  455. require.NoError(t, err)
  456. require.Equal(t, b, dial)
  457. // Even though we are at capacity, we should be allowed to dial c for an
  458. // upgrade of a, since it's higher-scored.
  459. added, err = peerManager.Add(c)
  460. require.NoError(t, err)
  461. require.True(t, added)
  462. dial, err = peerManager.TryDialNext()
  463. require.NoError(t, err)
  464. require.Equal(t, c, dial)
  465. // However, since we're using all upgrade slots now, we can't add and dial
  466. // d, even though it's also higher-scored.
  467. added, err = peerManager.Add(d)
  468. require.NoError(t, err)
  469. require.True(t, added)
  470. dial, err = peerManager.TryDialNext()
  471. require.NoError(t, err)
  472. require.Zero(t, dial)
  473. // We go through with c's upgrade.
  474. require.NoError(t, peerManager.Dialed(c))
  475. // Still can't dial d.
  476. dial, err = peerManager.TryDialNext()
  477. require.NoError(t, err)
  478. require.Zero(t, dial)
  479. // Now, if we disconnect a, we should be allowed to dial d because we have a
  480. // free upgrade slot.
  481. require.NoError(t, peerManager.Disconnected(a.NodeID))
  482. dial, err = peerManager.TryDialNext()
  483. require.NoError(t, err)
  484. require.Equal(t, d, dial)
  485. require.NoError(t, peerManager.Dialed(d))
  486. // However, if we disconnect b (such that only c and d are connected), we
  487. // should not be allowed to dial e even though there are upgrade slots,
  488. // because there are no lower-scored nodes that can be upgraded.
  489. require.NoError(t, peerManager.Disconnected(b.NodeID))
  490. added, err = peerManager.Add(e)
  491. require.NoError(t, err)
  492. require.True(t, added)
  493. dial, err = peerManager.TryDialNext()
  494. require.NoError(t, err)
  495. require.Zero(t, dial)
  496. }
  497. func TestPeerManager_TryDialNext_UpgradeReservesPeer(t *testing.T) {
  498. a := p2p.NodeAddress{Protocol: "memory", NodeID: p2p.NodeID(strings.Repeat("a", 40))}
  499. b := p2p.NodeAddress{Protocol: "memory", NodeID: p2p.NodeID(strings.Repeat("b", 40))}
  500. c := p2p.NodeAddress{Protocol: "memory", NodeID: p2p.NodeID(strings.Repeat("c", 40))}
  501. peerManager, err := p2p.NewPeerManager(selfID, dbm.NewMemDB(), p2p.PeerManagerOptions{
  502. PeerScores: map[p2p.NodeID]p2p.PeerScore{b.NodeID: 1, c.NodeID: 1},
  503. MaxConnected: 1,
  504. MaxConnectedUpgrade: 2,
  505. })
  506. require.NoError(t, err)
  507. // Add a and connect to it.
  508. added, err := peerManager.Add(a)
  509. require.NoError(t, err)
  510. require.True(t, added)
  511. dial, err := peerManager.TryDialNext()
  512. require.NoError(t, err)
  513. require.Equal(t, a, dial)
  514. require.NoError(t, peerManager.Dialed(a))
  515. // Add b and start dialing it. This will claim a for upgrading.
  516. added, err = peerManager.Add(b)
  517. require.NoError(t, err)
  518. require.True(t, added)
  519. dial, err = peerManager.TryDialNext()
  520. require.NoError(t, err)
  521. require.Equal(t, b, dial)
  522. // Adding c and dialing it will fail, because a is the only connected
  523. // peer that can be upgraded, and b is already trying to upgrade it.
  524. added, err = peerManager.Add(c)
  525. require.NoError(t, err)
  526. require.True(t, added)
  527. dial, err = peerManager.TryDialNext()
  528. require.NoError(t, err)
  529. require.Empty(t, dial)
  530. }
  531. func TestPeerManager_TryDialNext_DialingConnected(t *testing.T) {
  532. aID := p2p.NodeID(strings.Repeat("a", 40))
  533. a := p2p.NodeAddress{Protocol: "memory", NodeID: aID}
  534. aTCP := p2p.NodeAddress{Protocol: "tcp", NodeID: aID, Hostname: "localhost"}
  535. bID := p2p.NodeID(strings.Repeat("b", 40))
  536. b := p2p.NodeAddress{Protocol: "memory", NodeID: bID}
  537. peerManager, err := p2p.NewPeerManager(selfID, dbm.NewMemDB(), p2p.PeerManagerOptions{
  538. MaxConnected: 2,
  539. })
  540. require.NoError(t, err)
  541. // Add a and dial it.
  542. added, err := peerManager.Add(a)
  543. require.NoError(t, err)
  544. require.True(t, added)
  545. dial, err := peerManager.TryDialNext()
  546. require.NoError(t, err)
  547. require.Equal(t, a, dial)
  548. // Adding a's TCP address will not dispense a, since it's already dialing.
  549. added, err = peerManager.Add(aTCP)
  550. require.NoError(t, err)
  551. require.True(t, added)
  552. dial, err = peerManager.TryDialNext()
  553. require.NoError(t, err)
  554. require.Zero(t, dial)
  555. // Marking a as dialed will still not dispense it.
  556. require.NoError(t, peerManager.Dialed(a))
  557. dial, err = peerManager.TryDialNext()
  558. require.NoError(t, err)
  559. require.Zero(t, dial)
  560. // Adding b and accepting a connection from it will not dispense it either.
  561. added, err = peerManager.Add(b)
  562. require.NoError(t, err)
  563. require.True(t, added)
  564. require.NoError(t, peerManager.Accepted(bID))
  565. dial, err = peerManager.TryDialNext()
  566. require.NoError(t, err)
  567. require.Zero(t, dial)
  568. }
  569. func TestPeerManager_TryDialNext_Multiple(t *testing.T) {
  570. aID := p2p.NodeID(strings.Repeat("a", 40))
  571. bID := p2p.NodeID(strings.Repeat("b", 40))
  572. addresses := []p2p.NodeAddress{
  573. {Protocol: "memory", NodeID: aID},
  574. {Protocol: "memory", NodeID: bID},
  575. {Protocol: "tcp", NodeID: aID, Hostname: "127.0.0.1"},
  576. {Protocol: "tcp", NodeID: bID, Hostname: "::1"},
  577. }
  578. peerManager, err := p2p.NewPeerManager(selfID, dbm.NewMemDB(), p2p.PeerManagerOptions{})
  579. require.NoError(t, err)
  580. for _, address := range addresses {
  581. added, err := peerManager.Add(address)
  582. require.NoError(t, err)
  583. require.True(t, added)
  584. }
  585. // All addresses should be dispensed as long as dialing them has failed.
  586. dial := []p2p.NodeAddress{}
  587. for range addresses {
  588. address, err := peerManager.TryDialNext()
  589. require.NoError(t, err)
  590. require.NotZero(t, address)
  591. require.NoError(t, peerManager.DialFailed(address))
  592. dial = append(dial, address)
  593. }
  594. require.ElementsMatch(t, dial, addresses)
  595. address, err := peerManager.TryDialNext()
  596. require.NoError(t, err)
  597. require.Zero(t, address)
  598. }
  599. func TestPeerManager_DialFailed(t *testing.T) {
  600. // DialFailed is tested through other tests, we'll just check a few basic
  601. // things here, e.g. reporting unknown addresses.
  602. aID := p2p.NodeID(strings.Repeat("a", 40))
  603. a := p2p.NodeAddress{Protocol: "memory", NodeID: aID}
  604. bID := p2p.NodeID(strings.Repeat("b", 40))
  605. b := p2p.NodeAddress{Protocol: "memory", NodeID: bID}
  606. peerManager, err := p2p.NewPeerManager(selfID, dbm.NewMemDB(), p2p.PeerManagerOptions{})
  607. require.NoError(t, err)
  608. added, err := peerManager.Add(a)
  609. require.NoError(t, err)
  610. require.True(t, added)
  611. // Dialing and then calling DialFailed with a different address (same
  612. // NodeID) should unmark as dialing and allow us to dial the other address
  613. // again, but not register the failed address.
  614. dial, err := peerManager.TryDialNext()
  615. require.NoError(t, err)
  616. require.Equal(t, a, dial)
  617. require.NoError(t, peerManager.DialFailed(p2p.NodeAddress{
  618. Protocol: "tcp", NodeID: aID, Hostname: "localhost"}))
  619. require.Equal(t, []p2p.NodeAddress{a}, peerManager.Addresses(aID))
  620. dial, err = peerManager.TryDialNext()
  621. require.NoError(t, err)
  622. require.Equal(t, a, dial)
  623. // Calling DialFailed on same address twice should be fine.
  624. require.NoError(t, peerManager.DialFailed(a))
  625. require.NoError(t, peerManager.DialFailed(a))
  626. // DialFailed on an unknown peer shouldn't error or add it.
  627. require.NoError(t, peerManager.DialFailed(b))
  628. require.Equal(t, []p2p.NodeID{aID}, peerManager.Peers())
  629. }
  630. func TestPeerManager_DialFailed_UnreservePeer(t *testing.T) {
  631. a := p2p.NodeAddress{Protocol: "memory", NodeID: p2p.NodeID(strings.Repeat("a", 40))}
  632. b := p2p.NodeAddress{Protocol: "memory", NodeID: p2p.NodeID(strings.Repeat("b", 40))}
  633. c := p2p.NodeAddress{Protocol: "memory", NodeID: p2p.NodeID(strings.Repeat("c", 40))}
  634. peerManager, err := p2p.NewPeerManager(selfID, dbm.NewMemDB(), p2p.PeerManagerOptions{
  635. PeerScores: map[p2p.NodeID]p2p.PeerScore{b.NodeID: 1, c.NodeID: 1},
  636. MaxConnected: 1,
  637. MaxConnectedUpgrade: 2,
  638. })
  639. require.NoError(t, err)
  640. // Add a and connect to it.
  641. added, err := peerManager.Add(a)
  642. require.NoError(t, err)
  643. require.True(t, added)
  644. dial, err := peerManager.TryDialNext()
  645. require.NoError(t, err)
  646. require.Equal(t, a, dial)
  647. require.NoError(t, peerManager.Dialed(a))
  648. // Add b and start dialing it. This will claim a for upgrading.
  649. added, err = peerManager.Add(b)
  650. require.NoError(t, err)
  651. require.True(t, added)
  652. dial, err = peerManager.TryDialNext()
  653. require.NoError(t, err)
  654. require.Equal(t, b, dial)
  655. // Adding c and dialing it will fail, even though it could upgrade a and we
  656. // have free upgrade slots, because a is the only connected peer that can be
  657. // upgraded and b is already trying to upgrade it.
  658. added, err = peerManager.Add(c)
  659. require.NoError(t, err)
  660. require.True(t, added)
  661. dial, err = peerManager.TryDialNext()
  662. require.NoError(t, err)
  663. require.Empty(t, dial)
  664. // Failing b's dial will now make c available for dialing.
  665. require.NoError(t, peerManager.DialFailed(b))
  666. dial, err = peerManager.TryDialNext()
  667. require.NoError(t, err)
  668. require.Equal(t, c, dial)
  669. }
  670. func TestPeerManager_Dialed_Connected(t *testing.T) {
  671. a := p2p.NodeAddress{Protocol: "memory", NodeID: p2p.NodeID(strings.Repeat("a", 40))}
  672. b := p2p.NodeAddress{Protocol: "memory", NodeID: p2p.NodeID(strings.Repeat("b", 40))}
  673. peerManager, err := p2p.NewPeerManager(selfID, dbm.NewMemDB(), p2p.PeerManagerOptions{})
  674. require.NoError(t, err)
  675. // Marking a as dialed twice should error.
  676. added, err := peerManager.Add(a)
  677. require.NoError(t, err)
  678. require.True(t, added)
  679. dial, err := peerManager.TryDialNext()
  680. require.NoError(t, err)
  681. require.Equal(t, a, dial)
  682. require.NoError(t, peerManager.Dialed(a))
  683. require.Error(t, peerManager.Dialed(a))
  684. // Accepting a connection from b and then trying to mark it as dialed should fail.
  685. added, err = peerManager.Add(b)
  686. require.NoError(t, err)
  687. require.True(t, added)
  688. dial, err = peerManager.TryDialNext()
  689. require.NoError(t, err)
  690. require.Equal(t, b, dial)
  691. require.NoError(t, peerManager.Accepted(b.NodeID))
  692. require.Error(t, peerManager.Dialed(b))
  693. }
  694. func TestPeerManager_Dialed_Self(t *testing.T) {
  695. peerManager, err := p2p.NewPeerManager(selfID, dbm.NewMemDB(), p2p.PeerManagerOptions{})
  696. require.NoError(t, err)
  697. // Dialing self should error.
  698. _, err = peerManager.Add(p2p.NodeAddress{Protocol: "memory", NodeID: selfID})
  699. require.Error(t, err)
  700. }
  701. func TestPeerManager_Dialed_MaxConnected(t *testing.T) {
  702. a := p2p.NodeAddress{Protocol: "memory", NodeID: p2p.NodeID(strings.Repeat("a", 40))}
  703. b := p2p.NodeAddress{Protocol: "memory", NodeID: p2p.NodeID(strings.Repeat("b", 40))}
  704. peerManager, err := p2p.NewPeerManager(selfID, dbm.NewMemDB(), p2p.PeerManagerOptions{
  705. MaxConnected: 1,
  706. })
  707. require.NoError(t, err)
  708. // Start to dial a.
  709. added, err := peerManager.Add(a)
  710. require.NoError(t, err)
  711. require.True(t, added)
  712. dial, err := peerManager.TryDialNext()
  713. require.NoError(t, err)
  714. require.Equal(t, a, dial)
  715. // Marking b as dialed in the meanwhile (even without TryDialNext)
  716. // should be fine.
  717. added, err = peerManager.Add(b)
  718. require.NoError(t, err)
  719. require.True(t, added)
  720. require.NoError(t, peerManager.Dialed(b))
  721. // Completing the dial for a should now error.
  722. require.Error(t, peerManager.Dialed(a))
  723. }
  724. func TestPeerManager_Dialed_MaxConnectedUpgrade(t *testing.T) {
  725. a := p2p.NodeAddress{Protocol: "memory", NodeID: p2p.NodeID(strings.Repeat("a", 40))}
  726. b := p2p.NodeAddress{Protocol: "memory", NodeID: p2p.NodeID(strings.Repeat("b", 40))}
  727. c := p2p.NodeAddress{Protocol: "memory", NodeID: p2p.NodeID(strings.Repeat("c", 40))}
  728. d := p2p.NodeAddress{Protocol: "memory", NodeID: p2p.NodeID(strings.Repeat("d", 40))}
  729. peerManager, err := p2p.NewPeerManager(selfID, dbm.NewMemDB(), p2p.PeerManagerOptions{
  730. MaxConnected: 2,
  731. MaxConnectedUpgrade: 1,
  732. PeerScores: map[p2p.NodeID]p2p.PeerScore{c.NodeID: 1, d.NodeID: 1},
  733. })
  734. require.NoError(t, err)
  735. // Dialing a and b is fine.
  736. added, err := peerManager.Add(a)
  737. require.NoError(t, err)
  738. require.True(t, added)
  739. require.NoError(t, peerManager.Dialed(a))
  740. added, err = peerManager.Add(b)
  741. require.NoError(t, err)
  742. require.True(t, added)
  743. require.NoError(t, peerManager.Dialed(b))
  744. // Starting an upgrade of c should be fine.
  745. added, err = peerManager.Add(c)
  746. require.NoError(t, err)
  747. require.True(t, added)
  748. dial, err := peerManager.TryDialNext()
  749. require.NoError(t, err)
  750. require.Equal(t, c, dial)
  751. require.NoError(t, peerManager.Dialed(c))
  752. // Trying to mark d dialed should fail, since there are no more upgrade
  753. // slots and a/b haven't been evicted yet.
  754. added, err = peerManager.Add(d)
  755. require.NoError(t, err)
  756. require.True(t, added)
  757. require.Error(t, peerManager.Dialed(d))
  758. }
  759. func TestPeerManager_Dialed_Unknown(t *testing.T) {
  760. a := p2p.NodeAddress{Protocol: "memory", NodeID: p2p.NodeID(strings.Repeat("a", 40))}
  761. peerManager, err := p2p.NewPeerManager(selfID, dbm.NewMemDB(), p2p.PeerManagerOptions{})
  762. require.NoError(t, err)
  763. // Marking an unknown node as dialed should error.
  764. require.Error(t, peerManager.Dialed(a))
  765. }
  766. func TestPeerManager_Dialed_Upgrade(t *testing.T) {
  767. a := p2p.NodeAddress{Protocol: "memory", NodeID: p2p.NodeID(strings.Repeat("a", 40))}
  768. b := p2p.NodeAddress{Protocol: "memory", NodeID: p2p.NodeID(strings.Repeat("b", 40))}
  769. c := p2p.NodeAddress{Protocol: "memory", NodeID: p2p.NodeID(strings.Repeat("c", 40))}
  770. peerManager, err := p2p.NewPeerManager(selfID, dbm.NewMemDB(), p2p.PeerManagerOptions{
  771. MaxConnected: 1,
  772. MaxConnectedUpgrade: 2,
  773. PeerScores: map[p2p.NodeID]p2p.PeerScore{b.NodeID: 1, c.NodeID: 1},
  774. })
  775. require.NoError(t, err)
  776. // Dialing a is fine.
  777. added, err := peerManager.Add(a)
  778. require.NoError(t, err)
  779. require.True(t, added)
  780. require.NoError(t, peerManager.Dialed(a))
  781. // Upgrading it with b should work, since b has a higher score.
  782. added, err = peerManager.Add(b)
  783. require.NoError(t, err)
  784. require.True(t, added)
  785. dial, err := peerManager.TryDialNext()
  786. require.NoError(t, err)
  787. require.Equal(t, b, dial)
  788. require.NoError(t, peerManager.Dialed(b))
  789. // a hasn't been evicted yet, but c shouldn't be allowed to upgrade anyway
  790. // since it's about to be evicted.
  791. added, err = peerManager.Add(c)
  792. require.NoError(t, err)
  793. require.True(t, added)
  794. dial, err = peerManager.TryDialNext()
  795. require.NoError(t, err)
  796. require.Empty(t, dial)
  797. // a should now be evicted.
  798. evict, err := peerManager.TryEvictNext()
  799. require.NoError(t, err)
  800. require.Equal(t, a.NodeID, evict)
  801. }
  802. func TestPeerManager_Dialed_UpgradeEvenLower(t *testing.T) {
  803. a := p2p.NodeAddress{Protocol: "memory", NodeID: p2p.NodeID(strings.Repeat("a", 40))}
  804. b := p2p.NodeAddress{Protocol: "memory", NodeID: p2p.NodeID(strings.Repeat("b", 40))}
  805. c := p2p.NodeAddress{Protocol: "memory", NodeID: p2p.NodeID(strings.Repeat("c", 40))}
  806. d := p2p.NodeAddress{Protocol: "memory", NodeID: p2p.NodeID(strings.Repeat("d", 40))}
  807. peerManager, err := p2p.NewPeerManager(selfID, dbm.NewMemDB(), p2p.PeerManagerOptions{
  808. MaxConnected: 2,
  809. MaxConnectedUpgrade: 1,
  810. PeerScores: map[p2p.NodeID]p2p.PeerScore{
  811. a.NodeID: 3,
  812. b.NodeID: 2,
  813. c.NodeID: 10,
  814. d.NodeID: 1,
  815. },
  816. })
  817. require.NoError(t, err)
  818. // Connect to a and b.
  819. added, err := peerManager.Add(a)
  820. require.NoError(t, err)
  821. require.True(t, added)
  822. require.NoError(t, peerManager.Dialed(a))
  823. added, err = peerManager.Add(b)
  824. require.NoError(t, err)
  825. require.True(t, added)
  826. require.NoError(t, peerManager.Dialed(b))
  827. // Start an upgrade with c, which should pick b to upgrade (since it
  828. // has score 2).
  829. added, err = peerManager.Add(c)
  830. require.NoError(t, err)
  831. require.True(t, added)
  832. dial, err := peerManager.TryDialNext()
  833. require.NoError(t, err)
  834. require.Equal(t, c, dial)
  835. // In the meanwhile, a disconnects and d connects. d is even lower-scored
  836. // than b (1 vs 2), which is currently being upgraded.
  837. require.NoError(t, peerManager.Disconnected(a.NodeID))
  838. added, err = peerManager.Add(d)
  839. require.NoError(t, err)
  840. require.True(t, added)
  841. require.NoError(t, peerManager.Accepted(d.NodeID))
  842. // Once c completes the upgrade of b, it should instead evict d,
  843. // since it has en even lower score.
  844. require.NoError(t, peerManager.Dialed(c))
  845. evict, err := peerManager.TryEvictNext()
  846. require.NoError(t, err)
  847. require.Equal(t, d.NodeID, evict)
  848. }
  849. func TestPeerManager_Dialed_UpgradeNoEvict(t *testing.T) {
  850. a := p2p.NodeAddress{Protocol: "memory", NodeID: p2p.NodeID(strings.Repeat("a", 40))}
  851. b := p2p.NodeAddress{Protocol: "memory", NodeID: p2p.NodeID(strings.Repeat("b", 40))}
  852. c := p2p.NodeAddress{Protocol: "memory", NodeID: p2p.NodeID(strings.Repeat("c", 40))}
  853. peerManager, err := p2p.NewPeerManager(selfID, dbm.NewMemDB(), p2p.PeerManagerOptions{
  854. MaxConnected: 2,
  855. MaxConnectedUpgrade: 1,
  856. PeerScores: map[p2p.NodeID]p2p.PeerScore{
  857. a.NodeID: 1,
  858. b.NodeID: 2,
  859. c.NodeID: 3,
  860. },
  861. })
  862. require.NoError(t, err)
  863. // Connect to a and b.
  864. added, err := peerManager.Add(a)
  865. require.NoError(t, err)
  866. require.True(t, added)
  867. require.NoError(t, peerManager.Dialed(a))
  868. added, err = peerManager.Add(b)
  869. require.NoError(t, err)
  870. require.True(t, added)
  871. require.NoError(t, peerManager.Dialed(b))
  872. // Start an upgrade with c, which should pick a to upgrade.
  873. added, err = peerManager.Add(c)
  874. require.NoError(t, err)
  875. require.True(t, added)
  876. dial, err := peerManager.TryDialNext()
  877. require.NoError(t, err)
  878. require.Equal(t, c, dial)
  879. // In the meanwhile, b disconnects.
  880. require.NoError(t, peerManager.Disconnected(b.NodeID))
  881. // Once c completes the upgrade of b, there is no longer a need to
  882. // evict anything since we're at capacity.
  883. // since it has en even lower score.
  884. require.NoError(t, peerManager.Dialed(c))
  885. evict, err := peerManager.TryEvictNext()
  886. require.NoError(t, err)
  887. require.Zero(t, evict)
  888. }
  889. func TestPeerManager_Accepted(t *testing.T) {
  890. a := p2p.NodeAddress{Protocol: "memory", NodeID: p2p.NodeID(strings.Repeat("a", 40))}
  891. b := p2p.NodeAddress{Protocol: "memory", NodeID: p2p.NodeID(strings.Repeat("b", 40))}
  892. c := p2p.NodeAddress{Protocol: "memory", NodeID: p2p.NodeID(strings.Repeat("c", 40))}
  893. d := p2p.NodeAddress{Protocol: "memory", NodeID: p2p.NodeID(strings.Repeat("d", 40))}
  894. peerManager, err := p2p.NewPeerManager(selfID, dbm.NewMemDB(), p2p.PeerManagerOptions{})
  895. require.NoError(t, err)
  896. // Accepting a connection from self should error.
  897. require.Error(t, peerManager.Accepted(selfID))
  898. // Accepting a connection from a known peer should work.
  899. added, err := peerManager.Add(a)
  900. require.NoError(t, err)
  901. require.True(t, added)
  902. require.NoError(t, peerManager.Accepted(a.NodeID))
  903. // Accepting a connection from an already accepted peer should error.
  904. require.Error(t, peerManager.Accepted(a.NodeID))
  905. // Accepting a connection from an unknown peer should work and register it.
  906. require.NoError(t, peerManager.Accepted(b.NodeID))
  907. require.ElementsMatch(t, []p2p.NodeID{a.NodeID, b.NodeID}, peerManager.Peers())
  908. // Accepting a connection from a peer that's being dialed should work, and
  909. // should cause the dial to fail.
  910. added, err = peerManager.Add(c)
  911. require.NoError(t, err)
  912. require.True(t, added)
  913. dial, err := peerManager.TryDialNext()
  914. require.NoError(t, err)
  915. require.Equal(t, c, dial)
  916. require.NoError(t, peerManager.Accepted(c.NodeID))
  917. require.Error(t, peerManager.Dialed(c))
  918. // Accepting a connection from a peer that's been dialed should fail.
  919. added, err = peerManager.Add(d)
  920. require.NoError(t, err)
  921. require.True(t, added)
  922. dial, err = peerManager.TryDialNext()
  923. require.NoError(t, err)
  924. require.Equal(t, d, dial)
  925. require.NoError(t, peerManager.Dialed(d))
  926. require.Error(t, peerManager.Accepted(d.NodeID))
  927. }
  928. func TestPeerManager_Accepted_MaxConnected(t *testing.T) {
  929. a := p2p.NodeAddress{Protocol: "memory", NodeID: p2p.NodeID(strings.Repeat("a", 40))}
  930. b := p2p.NodeAddress{Protocol: "memory", NodeID: p2p.NodeID(strings.Repeat("b", 40))}
  931. c := p2p.NodeAddress{Protocol: "memory", NodeID: p2p.NodeID(strings.Repeat("c", 40))}
  932. peerManager, err := p2p.NewPeerManager(selfID, dbm.NewMemDB(), p2p.PeerManagerOptions{
  933. MaxConnected: 2,
  934. })
  935. require.NoError(t, err)
  936. // Connect to a and b.
  937. added, err := peerManager.Add(a)
  938. require.NoError(t, err)
  939. require.True(t, added)
  940. require.NoError(t, peerManager.Dialed(a))
  941. added, err = peerManager.Add(b)
  942. require.NoError(t, err)
  943. require.True(t, added)
  944. require.NoError(t, peerManager.Accepted(b.NodeID))
  945. // Accepting c should now fail.
  946. added, err = peerManager.Add(c)
  947. require.NoError(t, err)
  948. require.True(t, added)
  949. require.Error(t, peerManager.Accepted(c.NodeID))
  950. }
  951. func TestPeerManager_Accepted_MaxConnectedUpgrade(t *testing.T) {
  952. a := p2p.NodeAddress{Protocol: "memory", NodeID: p2p.NodeID(strings.Repeat("a", 40))}
  953. b := p2p.NodeAddress{Protocol: "memory", NodeID: p2p.NodeID(strings.Repeat("b", 40))}
  954. c := p2p.NodeAddress{Protocol: "memory", NodeID: p2p.NodeID(strings.Repeat("c", 40))}
  955. d := p2p.NodeAddress{Protocol: "memory", NodeID: p2p.NodeID(strings.Repeat("d", 40))}
  956. peerManager, err := p2p.NewPeerManager(selfID, dbm.NewMemDB(), p2p.PeerManagerOptions{
  957. PeerScores: map[p2p.NodeID]p2p.PeerScore{
  958. c.NodeID: 1,
  959. d.NodeID: 2,
  960. },
  961. MaxConnected: 1,
  962. MaxConnectedUpgrade: 1,
  963. })
  964. require.NoError(t, err)
  965. // Dial a.
  966. added, err := peerManager.Add(a)
  967. require.NoError(t, err)
  968. require.True(t, added)
  969. require.NoError(t, peerManager.Dialed(a))
  970. // Accepting b should fail, since it's not an upgrade over a.
  971. added, err = peerManager.Add(b)
  972. require.NoError(t, err)
  973. require.True(t, added)
  974. require.Error(t, peerManager.Accepted(b.NodeID))
  975. // Accepting c should work, since it upgrades a.
  976. added, err = peerManager.Add(c)
  977. require.NoError(t, err)
  978. require.True(t, added)
  979. require.NoError(t, peerManager.Accepted(c.NodeID))
  980. // a still hasn't been evicted, so accepting b should still fail.
  981. _, err = peerManager.Add(b)
  982. require.NoError(t, err)
  983. require.Error(t, peerManager.Accepted(b.NodeID))
  984. // Also, accepting d should fail, since all upgrade slots are full.
  985. added, err = peerManager.Add(d)
  986. require.NoError(t, err)
  987. require.True(t, added)
  988. require.Error(t, peerManager.Accepted(d.NodeID))
  989. }
  990. func TestPeerManager_Accepted_Upgrade(t *testing.T) {
  991. a := p2p.NodeAddress{Protocol: "memory", NodeID: p2p.NodeID(strings.Repeat("a", 40))}
  992. b := p2p.NodeAddress{Protocol: "memory", NodeID: p2p.NodeID(strings.Repeat("b", 40))}
  993. c := p2p.NodeAddress{Protocol: "memory", NodeID: p2p.NodeID(strings.Repeat("c", 40))}
  994. peerManager, err := p2p.NewPeerManager(selfID, dbm.NewMemDB(), p2p.PeerManagerOptions{
  995. PeerScores: map[p2p.NodeID]p2p.PeerScore{
  996. b.NodeID: 1,
  997. c.NodeID: 1,
  998. },
  999. MaxConnected: 1,
  1000. MaxConnectedUpgrade: 2,
  1001. })
  1002. require.NoError(t, err)
  1003. // Accept a.
  1004. added, err := peerManager.Add(a)
  1005. require.NoError(t, err)
  1006. require.True(t, added)
  1007. require.NoError(t, peerManager.Accepted(a.NodeID))
  1008. // Accepting b should work, since it upgrades a.
  1009. added, err = peerManager.Add(b)
  1010. require.NoError(t, err)
  1011. require.True(t, added)
  1012. require.NoError(t, peerManager.Accepted(b.NodeID))
  1013. // c cannot get accepted, since a has been upgraded by b.
  1014. added, err = peerManager.Add(c)
  1015. require.NoError(t, err)
  1016. require.True(t, added)
  1017. require.Error(t, peerManager.Accepted(c.NodeID))
  1018. // This should cause a to get evicted.
  1019. evict, err := peerManager.TryEvictNext()
  1020. require.NoError(t, err)
  1021. require.Equal(t, a.NodeID, evict)
  1022. require.NoError(t, peerManager.Disconnected(a.NodeID))
  1023. // c still cannot get accepted, since it's not scored above b.
  1024. require.Error(t, peerManager.Accepted(c.NodeID))
  1025. }
  1026. func TestPeerManager_Accepted_UpgradeDialing(t *testing.T) {
  1027. a := p2p.NodeAddress{Protocol: "memory", NodeID: p2p.NodeID(strings.Repeat("a", 40))}
  1028. b := p2p.NodeAddress{Protocol: "memory", NodeID: p2p.NodeID(strings.Repeat("b", 40))}
  1029. c := p2p.NodeAddress{Protocol: "memory", NodeID: p2p.NodeID(strings.Repeat("c", 40))}
  1030. peerManager, err := p2p.NewPeerManager(selfID, dbm.NewMemDB(), p2p.PeerManagerOptions{
  1031. PeerScores: map[p2p.NodeID]p2p.PeerScore{
  1032. b.NodeID: 1,
  1033. c.NodeID: 1,
  1034. },
  1035. MaxConnected: 1,
  1036. MaxConnectedUpgrade: 2,
  1037. })
  1038. require.NoError(t, err)
  1039. // Accept a.
  1040. added, err := peerManager.Add(a)
  1041. require.NoError(t, err)
  1042. require.True(t, added)
  1043. require.NoError(t, peerManager.Accepted(a.NodeID))
  1044. // Start dial upgrade from a to b.
  1045. added, err = peerManager.Add(b)
  1046. require.NoError(t, err)
  1047. require.True(t, added)
  1048. dial, err := peerManager.TryDialNext()
  1049. require.NoError(t, err)
  1050. require.Equal(t, b, dial)
  1051. // a has already been claimed as an upgrade of a, so accepting
  1052. // c should fail since there's noone else to upgrade.
  1053. added, err = peerManager.Add(c)
  1054. require.NoError(t, err)
  1055. require.True(t, added)
  1056. require.Error(t, peerManager.Accepted(c.NodeID))
  1057. // However, if b connects to us while we're also trying to upgrade to it via
  1058. // dialing, then we accept the incoming connection as an upgrade.
  1059. require.NoError(t, peerManager.Accepted(b.NodeID))
  1060. // This should cause a to get evicted, and the dial upgrade to fail.
  1061. evict, err := peerManager.TryEvictNext()
  1062. require.NoError(t, err)
  1063. require.Equal(t, a.NodeID, evict)
  1064. require.Error(t, peerManager.Dialed(b))
  1065. }
  1066. func TestPeerManager_Ready(t *testing.T) {
  1067. a := p2p.NodeAddress{Protocol: "memory", NodeID: p2p.NodeID(strings.Repeat("a", 40))}
  1068. b := p2p.NodeAddress{Protocol: "memory", NodeID: p2p.NodeID(strings.Repeat("b", 40))}
  1069. peerManager, err := p2p.NewPeerManager(selfID, dbm.NewMemDB(), p2p.PeerManagerOptions{})
  1070. require.NoError(t, err)
  1071. sub := peerManager.Subscribe()
  1072. defer sub.Close()
  1073. // Connecting to a should still have it as status down.
  1074. added, err := peerManager.Add(a)
  1075. require.NoError(t, err)
  1076. require.True(t, added)
  1077. require.NoError(t, peerManager.Accepted(a.NodeID))
  1078. require.Equal(t, p2p.PeerStatusDown, peerManager.Status(a.NodeID))
  1079. // Marking a as ready should transition it to PeerStatusUp and send an update.
  1080. require.NoError(t, peerManager.Ready(a.NodeID))
  1081. require.Equal(t, p2p.PeerStatusUp, peerManager.Status(a.NodeID))
  1082. require.Equal(t, p2p.PeerUpdate{
  1083. NodeID: a.NodeID,
  1084. Status: p2p.PeerStatusUp,
  1085. }, <-sub.Updates())
  1086. // Marking an unconnected peer as ready should do nothing.
  1087. added, err = peerManager.Add(b)
  1088. require.NoError(t, err)
  1089. require.True(t, added)
  1090. require.Equal(t, p2p.PeerStatusDown, peerManager.Status(b.NodeID))
  1091. require.NoError(t, peerManager.Ready(b.NodeID))
  1092. require.Equal(t, p2p.PeerStatusDown, peerManager.Status(b.NodeID))
  1093. require.Empty(t, sub.Updates())
  1094. }
  1095. // See TryEvictNext for most tests, this just tests blocking behavior.
  1096. func TestPeerManager_EvictNext(t *testing.T) {
  1097. a := p2p.NodeAddress{Protocol: "memory", NodeID: p2p.NodeID(strings.Repeat("a", 40))}
  1098. peerManager, err := p2p.NewPeerManager(selfID, dbm.NewMemDB(), p2p.PeerManagerOptions{})
  1099. require.NoError(t, err)
  1100. added, err := peerManager.Add(a)
  1101. require.NoError(t, err)
  1102. require.True(t, added)
  1103. require.NoError(t, peerManager.Accepted(a.NodeID))
  1104. require.NoError(t, peerManager.Ready(a.NodeID))
  1105. // Since there are no peers to evict, EvictNext should block until timeout.
  1106. timeoutCtx, cancel := context.WithTimeout(ctx, 100*time.Millisecond)
  1107. defer cancel()
  1108. _, err = peerManager.EvictNext(timeoutCtx)
  1109. require.Error(t, err)
  1110. require.Equal(t, context.DeadlineExceeded, err)
  1111. // Erroring the peer will return it from EvictNext().
  1112. require.NoError(t, peerManager.Errored(a.NodeID, errors.New("foo")))
  1113. evict, err := peerManager.EvictNext(timeoutCtx)
  1114. require.NoError(t, err)
  1115. require.Equal(t, a.NodeID, evict)
  1116. // Since there are no more peers to evict, the next call should block.
  1117. timeoutCtx, cancel = context.WithTimeout(ctx, 100*time.Millisecond)
  1118. defer cancel()
  1119. _, err = peerManager.EvictNext(timeoutCtx)
  1120. require.Error(t, err)
  1121. require.Equal(t, context.DeadlineExceeded, err)
  1122. }
  1123. func TestPeerManager_EvictNext_WakeOnError(t *testing.T) {
  1124. a := p2p.NodeAddress{Protocol: "memory", NodeID: p2p.NodeID(strings.Repeat("a", 40))}
  1125. peerManager, err := p2p.NewPeerManager(selfID, dbm.NewMemDB(), p2p.PeerManagerOptions{})
  1126. require.NoError(t, err)
  1127. added, err := peerManager.Add(a)
  1128. require.NoError(t, err)
  1129. require.True(t, added)
  1130. require.NoError(t, peerManager.Accepted(a.NodeID))
  1131. require.NoError(t, peerManager.Ready(a.NodeID))
  1132. // Spawn a goroutine to error a peer after a delay.
  1133. go func() {
  1134. time.Sleep(200 * time.Millisecond)
  1135. require.NoError(t, peerManager.Errored(a.NodeID, errors.New("foo")))
  1136. }()
  1137. // This will block until peer errors above.
  1138. ctx, cancel := context.WithTimeout(ctx, 3*time.Second)
  1139. defer cancel()
  1140. evict, err := peerManager.EvictNext(ctx)
  1141. require.NoError(t, err)
  1142. require.Equal(t, a.NodeID, evict)
  1143. }
  1144. func TestPeerManager_EvictNext_WakeOnUpgradeDialed(t *testing.T) {
  1145. a := p2p.NodeAddress{Protocol: "memory", NodeID: p2p.NodeID(strings.Repeat("a", 40))}
  1146. b := p2p.NodeAddress{Protocol: "memory", NodeID: p2p.NodeID(strings.Repeat("b", 40))}
  1147. peerManager, err := p2p.NewPeerManager(selfID, dbm.NewMemDB(), p2p.PeerManagerOptions{
  1148. MaxConnected: 1,
  1149. MaxConnectedUpgrade: 1,
  1150. PeerScores: map[p2p.NodeID]p2p.PeerScore{b.NodeID: 1},
  1151. })
  1152. require.NoError(t, err)
  1153. // Connect a.
  1154. added, err := peerManager.Add(a)
  1155. require.NoError(t, err)
  1156. require.True(t, added)
  1157. require.NoError(t, peerManager.Accepted(a.NodeID))
  1158. require.NoError(t, peerManager.Ready(a.NodeID))
  1159. // Spawn a goroutine to upgrade to b with a delay.
  1160. go func() {
  1161. time.Sleep(200 * time.Millisecond)
  1162. added, err := peerManager.Add(b)
  1163. require.NoError(t, err)
  1164. require.True(t, added)
  1165. dial, err := peerManager.TryDialNext()
  1166. require.NoError(t, err)
  1167. require.Equal(t, b, dial)
  1168. require.NoError(t, peerManager.Dialed(b))
  1169. }()
  1170. // This will block until peer is upgraded above.
  1171. ctx, cancel := context.WithTimeout(ctx, 3*time.Second)
  1172. defer cancel()
  1173. evict, err := peerManager.EvictNext(ctx)
  1174. require.NoError(t, err)
  1175. require.Equal(t, a.NodeID, evict)
  1176. }
  1177. func TestPeerManager_EvictNext_WakeOnUpgradeAccepted(t *testing.T) {
  1178. a := p2p.NodeAddress{Protocol: "memory", NodeID: p2p.NodeID(strings.Repeat("a", 40))}
  1179. b := p2p.NodeAddress{Protocol: "memory", NodeID: p2p.NodeID(strings.Repeat("b", 40))}
  1180. peerManager, err := p2p.NewPeerManager(selfID, dbm.NewMemDB(), p2p.PeerManagerOptions{
  1181. MaxConnected: 1,
  1182. MaxConnectedUpgrade: 1,
  1183. PeerScores: map[p2p.NodeID]p2p.PeerScore{b.NodeID: 1},
  1184. })
  1185. require.NoError(t, err)
  1186. // Connect a.
  1187. added, err := peerManager.Add(a)
  1188. require.NoError(t, err)
  1189. require.True(t, added)
  1190. require.NoError(t, peerManager.Accepted(a.NodeID))
  1191. require.NoError(t, peerManager.Ready(a.NodeID))
  1192. // Spawn a goroutine to upgrade b with a delay.
  1193. go func() {
  1194. time.Sleep(200 * time.Millisecond)
  1195. require.NoError(t, peerManager.Accepted(b.NodeID))
  1196. }()
  1197. // This will block until peer is upgraded above.
  1198. ctx, cancel := context.WithTimeout(ctx, 3*time.Second)
  1199. defer cancel()
  1200. evict, err := peerManager.EvictNext(ctx)
  1201. require.NoError(t, err)
  1202. require.Equal(t, a.NodeID, evict)
  1203. }
  1204. func TestPeerManager_TryEvictNext(t *testing.T) {
  1205. a := p2p.NodeAddress{Protocol: "memory", NodeID: p2p.NodeID(strings.Repeat("a", 40))}
  1206. peerManager, err := p2p.NewPeerManager(selfID, dbm.NewMemDB(), p2p.PeerManagerOptions{})
  1207. require.NoError(t, err)
  1208. added, err := peerManager.Add(a)
  1209. require.NoError(t, err)
  1210. require.True(t, added)
  1211. // Nothing is evicted with no peers connected.
  1212. evict, err := peerManager.TryEvictNext()
  1213. require.NoError(t, err)
  1214. require.Zero(t, evict)
  1215. // Connecting to a won't evict anything either.
  1216. require.NoError(t, peerManager.Accepted(a.NodeID))
  1217. require.NoError(t, peerManager.Ready(a.NodeID))
  1218. // But if a errors it should be evicted.
  1219. require.NoError(t, peerManager.Errored(a.NodeID, errors.New("foo")))
  1220. evict, err = peerManager.TryEvictNext()
  1221. require.NoError(t, err)
  1222. require.Equal(t, a.NodeID, evict)
  1223. // While a is being evicted (before disconnect), it shouldn't get evicted again.
  1224. evict, err = peerManager.TryEvictNext()
  1225. require.NoError(t, err)
  1226. require.Zero(t, evict)
  1227. require.NoError(t, peerManager.Errored(a.NodeID, errors.New("foo")))
  1228. evict, err = peerManager.TryEvictNext()
  1229. require.NoError(t, err)
  1230. require.Zero(t, evict)
  1231. }
  1232. func TestPeerManager_Disconnected(t *testing.T) {
  1233. a := p2p.NodeAddress{Protocol: "memory", NodeID: p2p.NodeID(strings.Repeat("a", 40))}
  1234. peerManager, err := p2p.NewPeerManager(selfID, dbm.NewMemDB(), p2p.PeerManagerOptions{})
  1235. require.NoError(t, err)
  1236. sub := peerManager.Subscribe()
  1237. defer sub.Close()
  1238. // Disconnecting an unknown peer does nothing.
  1239. require.NoError(t, peerManager.Disconnected(a.NodeID))
  1240. require.Empty(t, peerManager.Peers())
  1241. require.Empty(t, sub.Updates())
  1242. // Disconnecting an accepted non-ready peer does not send a status update.
  1243. added, err := peerManager.Add(a)
  1244. require.NoError(t, err)
  1245. require.True(t, added)
  1246. require.NoError(t, peerManager.Accepted(a.NodeID))
  1247. require.NoError(t, peerManager.Disconnected(a.NodeID))
  1248. require.Empty(t, sub.Updates())
  1249. // Disconnecting a ready peer sends a status update.
  1250. _, err = peerManager.Add(a)
  1251. require.NoError(t, err)
  1252. require.NoError(t, peerManager.Accepted(a.NodeID))
  1253. require.NoError(t, peerManager.Ready(a.NodeID))
  1254. require.Equal(t, p2p.PeerStatusUp, peerManager.Status(a.NodeID))
  1255. require.NotEmpty(t, sub.Updates())
  1256. require.Equal(t, p2p.PeerUpdate{
  1257. NodeID: a.NodeID,
  1258. Status: p2p.PeerStatusUp,
  1259. }, <-sub.Updates())
  1260. require.NoError(t, peerManager.Disconnected(a.NodeID))
  1261. require.Equal(t, p2p.PeerStatusDown, peerManager.Status(a.NodeID))
  1262. require.NotEmpty(t, sub.Updates())
  1263. require.Equal(t, p2p.PeerUpdate{
  1264. NodeID: a.NodeID,
  1265. Status: p2p.PeerStatusDown,
  1266. }, <-sub.Updates())
  1267. // Disconnecting a dialing peer does not unmark it as dialing, to avoid
  1268. // dialing it multiple times in parallel.
  1269. dial, err := peerManager.TryDialNext()
  1270. require.NoError(t, err)
  1271. require.Equal(t, a, dial)
  1272. require.NoError(t, peerManager.Disconnected(a.NodeID))
  1273. dial, err = peerManager.TryDialNext()
  1274. require.NoError(t, err)
  1275. require.Zero(t, dial)
  1276. }
  1277. func TestPeerManager_Errored(t *testing.T) {
  1278. a := p2p.NodeAddress{Protocol: "memory", NodeID: p2p.NodeID(strings.Repeat("a", 40))}
  1279. peerManager, err := p2p.NewPeerManager(selfID, dbm.NewMemDB(), p2p.PeerManagerOptions{})
  1280. require.NoError(t, err)
  1281. // Erroring an unknown peer does nothing.
  1282. require.NoError(t, peerManager.Errored(a.NodeID, errors.New("foo")))
  1283. require.Empty(t, peerManager.Peers())
  1284. evict, err := peerManager.TryEvictNext()
  1285. require.NoError(t, err)
  1286. require.Zero(t, evict)
  1287. // Erroring a known peer does nothing, and won't evict it later,
  1288. // even when it connects.
  1289. added, err := peerManager.Add(a)
  1290. require.NoError(t, err)
  1291. require.True(t, added)
  1292. require.NoError(t, peerManager.Errored(a.NodeID, errors.New("foo")))
  1293. evict, err = peerManager.TryEvictNext()
  1294. require.NoError(t, err)
  1295. require.Zero(t, evict)
  1296. require.NoError(t, peerManager.Accepted(a.NodeID))
  1297. require.NoError(t, peerManager.Ready(a.NodeID))
  1298. evict, err = peerManager.TryEvictNext()
  1299. require.NoError(t, err)
  1300. require.Zero(t, evict)
  1301. // However, erroring once connected will evict it.
  1302. require.NoError(t, peerManager.Errored(a.NodeID, errors.New("foo")))
  1303. evict, err = peerManager.TryEvictNext()
  1304. require.NoError(t, err)
  1305. require.Equal(t, a.NodeID, evict)
  1306. }
  1307. func TestPeerManager_Subscribe(t *testing.T) {
  1308. a := p2p.NodeAddress{Protocol: "memory", NodeID: p2p.NodeID(strings.Repeat("a", 40))}
  1309. peerManager, err := p2p.NewPeerManager(selfID, dbm.NewMemDB(), p2p.PeerManagerOptions{})
  1310. require.NoError(t, err)
  1311. // This tests all subscription events for full peer lifecycles.
  1312. sub := peerManager.Subscribe()
  1313. defer sub.Close()
  1314. added, err := peerManager.Add(a)
  1315. require.NoError(t, err)
  1316. require.True(t, added)
  1317. require.Empty(t, sub.Updates())
  1318. // Inbound connection.
  1319. require.NoError(t, peerManager.Accepted(a.NodeID))
  1320. require.Empty(t, sub.Updates())
  1321. require.NoError(t, peerManager.Ready(a.NodeID))
  1322. require.NotEmpty(t, sub.Updates())
  1323. require.Equal(t, p2p.PeerUpdate{NodeID: a.NodeID, Status: p2p.PeerStatusUp}, <-sub.Updates())
  1324. require.NoError(t, peerManager.Disconnected(a.NodeID))
  1325. require.NotEmpty(t, sub.Updates())
  1326. require.Equal(t, p2p.PeerUpdate{NodeID: a.NodeID, Status: p2p.PeerStatusDown}, <-sub.Updates())
  1327. // Outbound connection with peer error and eviction.
  1328. dial, err := peerManager.TryDialNext()
  1329. require.NoError(t, err)
  1330. require.Equal(t, a, dial)
  1331. require.Empty(t, sub.Updates())
  1332. require.NoError(t, peerManager.Dialed(a))
  1333. require.Empty(t, sub.Updates())
  1334. require.NoError(t, peerManager.Ready(a.NodeID))
  1335. require.NotEmpty(t, sub.Updates())
  1336. require.Equal(t, p2p.PeerUpdate{NodeID: a.NodeID, Status: p2p.PeerStatusUp}, <-sub.Updates())
  1337. require.NoError(t, peerManager.Errored(a.NodeID, errors.New("foo")))
  1338. require.Empty(t, sub.Updates())
  1339. evict, err := peerManager.TryEvictNext()
  1340. require.NoError(t, err)
  1341. require.Equal(t, a.NodeID, evict)
  1342. require.NoError(t, peerManager.Disconnected(a.NodeID))
  1343. require.NotEmpty(t, sub.Updates())
  1344. require.Equal(t, p2p.PeerUpdate{NodeID: a.NodeID, Status: p2p.PeerStatusDown}, <-sub.Updates())
  1345. // Outbound connection with dial failure.
  1346. dial, err = peerManager.TryDialNext()
  1347. require.NoError(t, err)
  1348. require.Equal(t, a, dial)
  1349. require.Empty(t, sub.Updates())
  1350. require.NoError(t, peerManager.DialFailed(a))
  1351. require.Empty(t, sub.Updates())
  1352. }
  1353. func TestPeerManager_Subscribe_Close(t *testing.T) {
  1354. a := p2p.NodeAddress{Protocol: "memory", NodeID: p2p.NodeID(strings.Repeat("a", 40))}
  1355. peerManager, err := p2p.NewPeerManager(selfID, dbm.NewMemDB(), p2p.PeerManagerOptions{})
  1356. require.NoError(t, err)
  1357. sub := peerManager.Subscribe()
  1358. defer sub.Close()
  1359. added, err := peerManager.Add(a)
  1360. require.NoError(t, err)
  1361. require.True(t, added)
  1362. require.NoError(t, peerManager.Accepted(a.NodeID))
  1363. require.Empty(t, sub.Updates())
  1364. require.NoError(t, peerManager.Ready(a.NodeID))
  1365. require.NotEmpty(t, sub.Updates())
  1366. require.Equal(t, p2p.PeerUpdate{NodeID: a.NodeID, Status: p2p.PeerStatusUp}, <-sub.Updates())
  1367. // Closing the subscription should not send us the disconnected update.
  1368. sub.Close()
  1369. require.NoError(t, peerManager.Disconnected(a.NodeID))
  1370. require.Empty(t, sub.Updates())
  1371. }
  1372. func TestPeerManager_Subscribe_Broadcast(t *testing.T) {
  1373. t.Cleanup(leaktest.Check(t))
  1374. a := p2p.NodeAddress{Protocol: "memory", NodeID: p2p.NodeID(strings.Repeat("a", 40))}
  1375. peerManager, err := p2p.NewPeerManager(selfID, dbm.NewMemDB(), p2p.PeerManagerOptions{})
  1376. require.NoError(t, err)
  1377. s1 := peerManager.Subscribe()
  1378. defer s1.Close()
  1379. s2 := peerManager.Subscribe()
  1380. defer s2.Close()
  1381. s3 := peerManager.Subscribe()
  1382. defer s3.Close()
  1383. // Connecting to a peer should send updates on all subscriptions.
  1384. added, err := peerManager.Add(a)
  1385. require.NoError(t, err)
  1386. require.True(t, added)
  1387. require.NoError(t, peerManager.Accepted(a.NodeID))
  1388. require.NoError(t, peerManager.Ready(a.NodeID))
  1389. expectUp := p2p.PeerUpdate{NodeID: a.NodeID, Status: p2p.PeerStatusUp}
  1390. require.NotEmpty(t, s1)
  1391. require.Equal(t, expectUp, <-s1.Updates())
  1392. require.NotEmpty(t, s2)
  1393. require.Equal(t, expectUp, <-s2.Updates())
  1394. require.NotEmpty(t, s3)
  1395. require.Equal(t, expectUp, <-s3.Updates())
  1396. // We now close s2. Disconnecting the peer should only send updates
  1397. // on s1 and s3.
  1398. s2.Close()
  1399. require.NoError(t, peerManager.Disconnected(a.NodeID))
  1400. expectDown := p2p.PeerUpdate{NodeID: a.NodeID, Status: p2p.PeerStatusDown}
  1401. require.NotEmpty(t, s1)
  1402. require.Equal(t, expectDown, <-s1.Updates())
  1403. require.Empty(t, s2.Updates())
  1404. require.NotEmpty(t, s3)
  1405. require.Equal(t, expectDown, <-s3.Updates())
  1406. }
  1407. func TestPeerManager_Close(t *testing.T) {
  1408. // leaktest will check that spawned goroutines are closed.
  1409. t.Cleanup(leaktest.CheckTimeout(t, 1*time.Second))
  1410. a := p2p.NodeAddress{Protocol: "memory", NodeID: p2p.NodeID(strings.Repeat("a", 40))}
  1411. peerManager, err := p2p.NewPeerManager(selfID, dbm.NewMemDB(), p2p.PeerManagerOptions{
  1412. MinRetryTime: 10 * time.Second,
  1413. })
  1414. require.NoError(t, err)
  1415. // This subscription isn't closed, but PeerManager.Close()
  1416. // should reap the spawned goroutine.
  1417. _ = peerManager.Subscribe()
  1418. // This dial failure will start a retry timer for 10 seconds, which
  1419. // should be reaped.
  1420. added, err := peerManager.Add(a)
  1421. require.NoError(t, err)
  1422. require.True(t, added)
  1423. dial, err := peerManager.TryDialNext()
  1424. require.NoError(t, err)
  1425. require.Equal(t, a, dial)
  1426. require.NoError(t, peerManager.DialFailed(a))
  1427. // This should clean up the goroutines.
  1428. peerManager.Close()
  1429. }
  1430. func TestPeerManager_Advertise(t *testing.T) {
  1431. aID := p2p.NodeID(strings.Repeat("a", 40))
  1432. aTCP := p2p.NodeAddress{Protocol: "tcp", NodeID: aID, Hostname: "127.0.0.1", Port: 26657, Path: "/path"}
  1433. aMem := p2p.NodeAddress{Protocol: "memory", NodeID: aID}
  1434. bID := p2p.NodeID(strings.Repeat("b", 40))
  1435. bTCP := p2p.NodeAddress{Protocol: "tcp", NodeID: bID, Hostname: "b10c::1", Port: 26657, Path: "/path"}
  1436. bMem := p2p.NodeAddress{Protocol: "memory", NodeID: bID}
  1437. cID := p2p.NodeID(strings.Repeat("c", 40))
  1438. cTCP := p2p.NodeAddress{Protocol: "tcp", NodeID: cID, Hostname: "host.domain", Port: 80}
  1439. cMem := p2p.NodeAddress{Protocol: "memory", NodeID: cID}
  1440. dID := p2p.NodeID(strings.Repeat("d", 40))
  1441. // Create an initial peer manager and add the peers.
  1442. peerManager, err := p2p.NewPeerManager(selfID, dbm.NewMemDB(), p2p.PeerManagerOptions{
  1443. PeerScores: map[p2p.NodeID]p2p.PeerScore{aID: 3, bID: 2, cID: 1},
  1444. })
  1445. require.NoError(t, err)
  1446. defer peerManager.Close()
  1447. added, err := peerManager.Add(aTCP)
  1448. require.NoError(t, err)
  1449. require.True(t, added)
  1450. added, err = peerManager.Add(aMem)
  1451. require.NoError(t, err)
  1452. require.True(t, added)
  1453. added, err = peerManager.Add(bTCP)
  1454. require.NoError(t, err)
  1455. require.True(t, added)
  1456. added, err = peerManager.Add(bMem)
  1457. require.NoError(t, err)
  1458. require.True(t, added)
  1459. added, err = peerManager.Add(cTCP)
  1460. require.NoError(t, err)
  1461. require.True(t, added)
  1462. added, err = peerManager.Add(cMem)
  1463. require.NoError(t, err)
  1464. require.True(t, added)
  1465. // d should get all addresses.
  1466. require.ElementsMatch(t, []p2p.NodeAddress{
  1467. aTCP, aMem, bTCP, bMem, cTCP, cMem,
  1468. }, peerManager.Advertise(dID, 100))
  1469. // a should not get its own addresses.
  1470. require.ElementsMatch(t, []p2p.NodeAddress{
  1471. bTCP, bMem, cTCP, cMem,
  1472. }, peerManager.Advertise(aID, 100))
  1473. // Asking for 0 addresses should return, well, 0.
  1474. require.Empty(t, peerManager.Advertise(aID, 0))
  1475. // Asking for 2 addresses should get the highest-rated ones, i.e. a.
  1476. require.ElementsMatch(t, []p2p.NodeAddress{
  1477. aTCP, aMem,
  1478. }, peerManager.Advertise(dID, 2))
  1479. }
  1480. func TestPeerManager_SetHeight_GetHeight(t *testing.T) {
  1481. a := p2p.NodeAddress{Protocol: "memory", NodeID: p2p.NodeID(strings.Repeat("a", 40))}
  1482. b := p2p.NodeAddress{Protocol: "memory", NodeID: p2p.NodeID(strings.Repeat("b", 40))}
  1483. db := dbm.NewMemDB()
  1484. peerManager, err := p2p.NewPeerManager(selfID, db, p2p.PeerManagerOptions{})
  1485. require.NoError(t, err)
  1486. // Getting a height should default to 0, for unknown peers and
  1487. // for known peers without height.
  1488. added, err := peerManager.Add(a)
  1489. require.NoError(t, err)
  1490. require.True(t, added)
  1491. require.EqualValues(t, 0, peerManager.GetHeight(a.NodeID))
  1492. require.EqualValues(t, 0, peerManager.GetHeight(b.NodeID))
  1493. // Setting a height should work for a known node.
  1494. require.NoError(t, peerManager.SetHeight(a.NodeID, 3))
  1495. require.EqualValues(t, 3, peerManager.GetHeight(a.NodeID))
  1496. // Setting a height should add an unknown node.
  1497. require.Equal(t, []p2p.NodeID{a.NodeID}, peerManager.Peers())
  1498. require.NoError(t, peerManager.SetHeight(b.NodeID, 7))
  1499. require.EqualValues(t, 7, peerManager.GetHeight(b.NodeID))
  1500. require.ElementsMatch(t, []p2p.NodeID{a.NodeID, b.NodeID}, peerManager.Peers())
  1501. // The heights should not be persisted.
  1502. peerManager.Close()
  1503. peerManager, err = p2p.NewPeerManager(selfID, db, p2p.PeerManagerOptions{})
  1504. require.NoError(t, err)
  1505. require.ElementsMatch(t, []p2p.NodeID{a.NodeID, b.NodeID}, peerManager.Peers())
  1506. require.Zero(t, peerManager.GetHeight(a.NodeID))
  1507. require.Zero(t, peerManager.GetHeight(b.NodeID))
  1508. }