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.

574 lines
17 KiB

  1. package evidence_test
  2. import (
  3. "encoding/hex"
  4. "math/rand"
  5. "sync"
  6. "testing"
  7. "time"
  8. "github.com/fortytw2/leaktest"
  9. "github.com/stretchr/testify/assert"
  10. "github.com/stretchr/testify/mock"
  11. "github.com/stretchr/testify/require"
  12. dbm "github.com/tendermint/tm-db"
  13. "github.com/tendermint/tendermint/crypto"
  14. "github.com/tendermint/tendermint/crypto/tmhash"
  15. "github.com/tendermint/tendermint/evidence"
  16. "github.com/tendermint/tendermint/evidence/mocks"
  17. "github.com/tendermint/tendermint/libs/log"
  18. "github.com/tendermint/tendermint/p2p"
  19. "github.com/tendermint/tendermint/p2p/p2ptest"
  20. tmproto "github.com/tendermint/tendermint/proto/tendermint/types"
  21. sm "github.com/tendermint/tendermint/state"
  22. "github.com/tendermint/tendermint/types"
  23. )
  24. var (
  25. numEvidence = 10
  26. rng = rand.New(rand.NewSource(time.Now().UnixNano()))
  27. )
  28. type reactorTestSuite struct {
  29. network *p2ptest.Network
  30. logger log.Logger
  31. reactors map[p2p.NodeID]*evidence.Reactor
  32. pools map[p2p.NodeID]*evidence.Pool
  33. evidenceChannels map[p2p.NodeID]*p2p.Channel
  34. peerUpdates map[p2p.NodeID]*p2p.PeerUpdates
  35. peerChans map[p2p.NodeID]chan p2p.PeerUpdate
  36. nodes []*p2ptest.Node
  37. numStateStores int
  38. }
  39. func setup(t *testing.T, stateStores []sm.Store, chBuf uint) *reactorTestSuite {
  40. t.Helper()
  41. pID := make([]byte, 16)
  42. _, err := rng.Read(pID)
  43. require.NoError(t, err)
  44. numStateStores := len(stateStores)
  45. rts := &reactorTestSuite{
  46. numStateStores: numStateStores,
  47. logger: log.TestingLogger().With("testCase", t.Name()),
  48. network: p2ptest.MakeNetwork(t, p2ptest.NetworkOptions{NumNodes: numStateStores}),
  49. reactors: make(map[p2p.NodeID]*evidence.Reactor, numStateStores),
  50. pools: make(map[p2p.NodeID]*evidence.Pool, numStateStores),
  51. peerUpdates: make(map[p2p.NodeID]*p2p.PeerUpdates, numStateStores),
  52. peerChans: make(map[p2p.NodeID]chan p2p.PeerUpdate, numStateStores),
  53. }
  54. rts.evidenceChannels = rts.network.MakeChannelsNoCleanup(t,
  55. evidence.EvidenceChannel,
  56. new(tmproto.EvidenceList),
  57. int(chBuf))
  58. require.Len(t, rts.network.RandomNode().PeerManager.Peers(), 0)
  59. idx := 0
  60. evidenceTime := time.Date(2019, 1, 1, 0, 0, 0, 0, time.UTC)
  61. for nodeID := range rts.network.Nodes {
  62. logger := rts.logger.With("validator", idx)
  63. evidenceDB := dbm.NewMemDB()
  64. blockStore := &mocks.BlockStore{}
  65. state, _ := stateStores[idx].Load()
  66. blockStore.On("LoadBlockMeta", mock.AnythingOfType("int64")).Return(func(h int64) *types.BlockMeta {
  67. if h <= state.LastBlockHeight {
  68. return &types.BlockMeta{Header: types.Header{Time: evidenceTime}}
  69. }
  70. return nil
  71. })
  72. rts.pools[nodeID], err = evidence.NewPool(logger, evidenceDB, stateStores[idx], blockStore)
  73. require.NoError(t, err)
  74. rts.peerChans[nodeID] = make(chan p2p.PeerUpdate)
  75. rts.peerUpdates[nodeID] = p2p.NewPeerUpdates(rts.peerChans[nodeID])
  76. rts.network.Nodes[nodeID].PeerManager.Register(rts.peerUpdates[nodeID])
  77. rts.nodes = append(rts.nodes, rts.network.Nodes[nodeID])
  78. rts.reactors[nodeID] = evidence.NewReactor(logger,
  79. rts.evidenceChannels[nodeID],
  80. rts.peerUpdates[nodeID],
  81. rts.pools[nodeID])
  82. require.NoError(t, rts.reactors[nodeID].Start())
  83. require.True(t, rts.reactors[nodeID].IsRunning())
  84. idx++
  85. }
  86. t.Cleanup(func() {
  87. for _, r := range rts.reactors {
  88. if r.IsRunning() {
  89. require.NoError(t, r.Stop())
  90. require.False(t, r.IsRunning())
  91. }
  92. }
  93. leaktest.Check(t)
  94. })
  95. return rts
  96. }
  97. func (rts *reactorTestSuite) start(t *testing.T) {
  98. rts.network.Start(t)
  99. require.Len(t,
  100. rts.network.RandomNode().PeerManager.Peers(),
  101. rts.numStateStores-1,
  102. "network does not have expected number of nodes")
  103. }
  104. func (rts *reactorTestSuite) waitForEvidence(t *testing.T, evList types.EvidenceList, ids ...p2p.NodeID) {
  105. t.Helper()
  106. fn := func(pool *evidence.Pool) {
  107. var (
  108. localEvList []types.Evidence
  109. size int64
  110. loops int
  111. )
  112. // wait till we have at least the amount of evidence
  113. // that we expect. if there's more local evidence then
  114. // it doesn't make sense to wait longer and a
  115. // different assertion should catch the resulting error
  116. for len(localEvList) < len(evList) {
  117. // each evidence should not be more than 500 bytes
  118. localEvList, size = pool.PendingEvidence(int64(len(evList) * 500))
  119. if loops == 100 {
  120. t.Log("current wait status:", "|",
  121. "local", len(localEvList), "|",
  122. "waitlist", len(evList), "|",
  123. "size", size)
  124. }
  125. loops++
  126. }
  127. // put the reaped evidence in a map so we can quickly check we got everything
  128. evMap := make(map[string]types.Evidence)
  129. for _, e := range localEvList {
  130. evMap[string(e.Hash())] = e
  131. }
  132. for i, expectedEv := range evList {
  133. gotEv := evMap[string(expectedEv.Hash())]
  134. require.Equalf(
  135. t,
  136. expectedEv,
  137. gotEv,
  138. "evidence for pool %d in pool does not match; got: %v, expected: %v", i, gotEv, expectedEv,
  139. )
  140. }
  141. }
  142. if len(ids) == 1 {
  143. // special case waiting once, just to avoid the extra
  144. // goroutine, in the case that this hits a timeout,
  145. // the stack will be clearer.
  146. fn(rts.pools[ids[0]])
  147. return
  148. }
  149. wg := sync.WaitGroup{}
  150. for id := range rts.pools {
  151. if len(ids) > 0 && !p2ptest.NodeInSlice(id, ids) {
  152. // if an ID list is specified, then we only
  153. // want to wait for those pools that are
  154. // specified in the list, otherwise, wait for
  155. // all pools.
  156. continue
  157. }
  158. wg.Add(1)
  159. go func(id p2p.NodeID) { defer wg.Done(); fn(rts.pools[id]) }(id)
  160. }
  161. wg.Wait()
  162. }
  163. func (rts *reactorTestSuite) assertEvidenceChannelsEmpty(t *testing.T) {
  164. t.Helper()
  165. for id, r := range rts.reactors {
  166. require.NoError(t, r.Stop(), "stopping reactor #%s", id)
  167. r.Wait()
  168. require.False(t, r.IsRunning(), "reactor #%d did not stop", id)
  169. }
  170. for id, ech := range rts.evidenceChannels {
  171. require.Empty(t, ech.Out, "checking channel #%q", id)
  172. }
  173. }
  174. func createEvidenceList(
  175. t *testing.T,
  176. pool *evidence.Pool,
  177. val types.PrivValidator,
  178. numEvidence int,
  179. ) types.EvidenceList {
  180. t.Helper()
  181. evList := make([]types.Evidence, numEvidence)
  182. for i := 0; i < numEvidence; i++ {
  183. ev := types.NewMockDuplicateVoteEvidenceWithValidator(
  184. int64(i+1),
  185. time.Date(2019, 1, 1, 0, 0, 0, 0, time.UTC),
  186. val,
  187. evidenceChainID,
  188. )
  189. require.NoError(t, pool.AddEvidence(ev),
  190. "adding evidence it#%d of %d to pool with height %d",
  191. i, numEvidence, pool.State().LastBlockHeight)
  192. evList[i] = ev
  193. }
  194. return evList
  195. }
  196. func TestReactorMultiDisconnect(t *testing.T) {
  197. val := types.NewMockPV()
  198. height := int64(numEvidence) + 10
  199. stateDB1 := initializeValidatorState(t, val, height)
  200. stateDB2 := initializeValidatorState(t, val, height)
  201. rts := setup(t, []sm.Store{stateDB1, stateDB2}, 20)
  202. primary := rts.nodes[0]
  203. secondary := rts.nodes[1]
  204. _ = createEvidenceList(t, rts.pools[primary.NodeID], val, numEvidence)
  205. require.Equal(t, primary.PeerManager.Status(secondary.NodeID), p2p.PeerStatusDown)
  206. rts.start(t)
  207. require.Equal(t, primary.PeerManager.Status(secondary.NodeID), p2p.PeerStatusUp)
  208. // Ensure "disconnecting" the secondary peer from the primary more than once
  209. // is handled gracefully.
  210. require.NoError(t, primary.PeerManager.Disconnected(secondary.NodeID))
  211. require.Equal(t, primary.PeerManager.Status(secondary.NodeID), p2p.PeerStatusDown)
  212. _, err := primary.PeerManager.TryEvictNext()
  213. require.NoError(t, err)
  214. require.NoError(t, primary.PeerManager.Disconnected(secondary.NodeID))
  215. require.Equal(t, primary.PeerManager.Status(secondary.NodeID), p2p.PeerStatusDown)
  216. require.Equal(t, secondary.PeerManager.Status(primary.NodeID), p2p.PeerStatusUp)
  217. }
  218. // TestReactorBroadcastEvidence creates an environment of multiple peers that
  219. // are all at the same height. One peer, designated as a primary, gossips all
  220. // evidence to the remaining peers.
  221. func TestReactorBroadcastEvidence(t *testing.T) {
  222. numPeers := 7
  223. // create a stateDB for all test suites (nodes)
  224. stateDBs := make([]sm.Store, numPeers)
  225. val := types.NewMockPV()
  226. // We need all validators saved for heights at least as high as we have
  227. // evidence for.
  228. height := int64(numEvidence) + 10
  229. for i := 0; i < numPeers; i++ {
  230. stateDBs[i] = initializeValidatorState(t, val, height)
  231. }
  232. rts := setup(t, stateDBs, 0)
  233. rts.start(t)
  234. // Create a series of fixtures where each suite contains a reactor and
  235. // evidence pool. In addition, we mark a primary suite and the rest are
  236. // secondaries where each secondary is added as a peer via a PeerUpdate to the
  237. // primary. As a result, the primary will gossip all evidence to each secondary.
  238. primary := rts.network.RandomNode()
  239. secondaries := make([]*p2ptest.Node, 0, len(rts.network.NodeIDs())-1)
  240. secondaryIDs := make([]p2p.NodeID, 0, cap(secondaries))
  241. for id := range rts.network.Nodes {
  242. if id == primary.NodeID {
  243. continue
  244. }
  245. secondaries = append(secondaries, rts.network.Nodes[id])
  246. secondaryIDs = append(secondaryIDs, id)
  247. }
  248. evList := createEvidenceList(t, rts.pools[primary.NodeID], val, numEvidence)
  249. // Add each secondary suite (node) as a peer to the primary suite (node). This
  250. // will cause the primary to gossip all evidence to the secondaries.
  251. for _, suite := range secondaries {
  252. rts.peerChans[primary.NodeID] <- p2p.PeerUpdate{
  253. Status: p2p.PeerStatusUp,
  254. NodeID: suite.NodeID,
  255. }
  256. }
  257. // Wait till all secondary suites (reactor) received all evidence from the
  258. // primary suite (node).
  259. rts.waitForEvidence(t, evList, secondaryIDs...)
  260. for _, pool := range rts.pools {
  261. require.Equal(t, numEvidence, int(pool.Size()))
  262. }
  263. rts.assertEvidenceChannelsEmpty(t)
  264. }
  265. // TestReactorSelectiveBroadcast tests a context where we have two reactors
  266. // connected to one another but are at different heights. Reactor 1 which is
  267. // ahead receives a list of evidence.
  268. func TestReactorBroadcastEvidence_Lagging(t *testing.T) {
  269. val := types.NewMockPV()
  270. height1 := int64(numEvidence) + 10
  271. height2 := int64(numEvidence) / 2
  272. // stateDB1 is ahead of stateDB2, where stateDB1 has all heights (1-20) and
  273. // stateDB2 only has heights 1-5.
  274. stateDB1 := initializeValidatorState(t, val, height1)
  275. stateDB2 := initializeValidatorState(t, val, height2)
  276. rts := setup(t, []sm.Store{stateDB1, stateDB2}, 100)
  277. rts.start(t)
  278. primary := rts.nodes[0]
  279. secondary := rts.nodes[1]
  280. // Send a list of valid evidence to the first reactor's, the one that is ahead,
  281. // evidence pool.
  282. evList := createEvidenceList(t, rts.pools[primary.NodeID], val, numEvidence)
  283. // Add each secondary suite (node) as a peer to the primary suite (node). This
  284. // will cause the primary to gossip all evidence to the secondaries.
  285. rts.peerChans[primary.NodeID] <- p2p.PeerUpdate{
  286. Status: p2p.PeerStatusUp,
  287. NodeID: secondary.NodeID,
  288. }
  289. // only ones less than the peers height should make it through
  290. rts.waitForEvidence(t, evList[:height2], secondary.NodeID)
  291. require.Equal(t, numEvidence, int(rts.pools[primary.NodeID].Size()))
  292. require.Equal(t, int(height2), int(rts.pools[secondary.NodeID].Size()))
  293. rts.assertEvidenceChannelsEmpty(t)
  294. }
  295. func TestReactorBroadcastEvidence_Pending(t *testing.T) {
  296. val := types.NewMockPV()
  297. height := int64(10)
  298. stateDB1 := initializeValidatorState(t, val, height)
  299. stateDB2 := initializeValidatorState(t, val, height)
  300. rts := setup(t, []sm.Store{stateDB1, stateDB2}, 100)
  301. primary := rts.nodes[0]
  302. secondary := rts.nodes[1]
  303. evList := createEvidenceList(t, rts.pools[primary.NodeID], val, numEvidence)
  304. // Manually add half the evidence to the secondary which will mark them as
  305. // pending.
  306. for i := 0; i < numEvidence/2; i++ {
  307. require.NoError(t, rts.pools[secondary.NodeID].AddEvidence(evList[i]))
  308. }
  309. // the secondary should have half the evidence as pending
  310. require.Equal(t, numEvidence/2, int(rts.pools[secondary.NodeID].Size()))
  311. rts.start(t)
  312. // The secondary reactor should have received all the evidence ignoring the
  313. // already pending evidence.
  314. rts.waitForEvidence(t, evList, secondary.NodeID)
  315. // check to make sure that all of the evidence has
  316. // propogated
  317. require.Len(t, rts.pools, 2)
  318. assert.EqualValues(t, numEvidence, rts.pools[primary.NodeID].Size(),
  319. "primary node should have all the evidence")
  320. if assert.EqualValues(t, numEvidence, rts.pools[secondary.NodeID].Size(),
  321. "secondary nodes should have caught up") {
  322. rts.assertEvidenceChannelsEmpty(t)
  323. }
  324. }
  325. func TestReactorBroadcastEvidence_Committed(t *testing.T) {
  326. val := types.NewMockPV()
  327. height := int64(10)
  328. stateDB1 := initializeValidatorState(t, val, height)
  329. stateDB2 := initializeValidatorState(t, val, height)
  330. rts := setup(t, []sm.Store{stateDB1, stateDB2}, 0)
  331. primary := rts.nodes[0]
  332. secondary := rts.nodes[1]
  333. // add all evidence to the primary reactor
  334. evList := createEvidenceList(t, rts.pools[primary.NodeID], val, numEvidence)
  335. // Manually add half the evidence to the secondary which will mark them as
  336. // pending.
  337. for i := 0; i < numEvidence/2; i++ {
  338. require.NoError(t, rts.pools[secondary.NodeID].AddEvidence(evList[i]))
  339. }
  340. // the secondary should have half the evidence as pending
  341. require.Equal(t, numEvidence/2, int(rts.pools[secondary.NodeID].Size()))
  342. state, err := stateDB2.Load()
  343. require.NoError(t, err)
  344. // update the secondary's pool such that all pending evidence is committed
  345. state.LastBlockHeight++
  346. rts.pools[secondary.NodeID].Update(state, evList[:numEvidence/2])
  347. // the secondary should have half the evidence as committed
  348. require.Equal(t, 0, int(rts.pools[secondary.NodeID].Size()))
  349. // start the network and ensure it's configured
  350. rts.start(t)
  351. // without the following sleep the test consistently fails;
  352. // likely because the sleep forces a context switch that lets
  353. // the router process other operations.
  354. time.Sleep(2 * time.Millisecond)
  355. // The secondary reactor should have received all the evidence ignoring the
  356. // already committed evidence.
  357. rts.waitForEvidence(t, evList[numEvidence/2:], secondary.NodeID)
  358. require.Len(t, rts.pools, 2)
  359. assert.EqualValues(t, numEvidence, rts.pools[primary.NodeID].Size(),
  360. "primary node should have all the evidence")
  361. if assert.EqualValues(t, numEvidence/2, rts.pools[secondary.NodeID].Size(),
  362. "secondary nodes should have caught up") {
  363. rts.assertEvidenceChannelsEmpty(t)
  364. }
  365. }
  366. func TestReactorBroadcastEvidence_FullyConnected(t *testing.T) {
  367. numPeers := 7
  368. // create a stateDB for all test suites (nodes)
  369. stateDBs := make([]sm.Store, numPeers)
  370. val := types.NewMockPV()
  371. // We need all validators saved for heights at least as high as we have
  372. // evidence for.
  373. height := int64(numEvidence) + 10
  374. for i := 0; i < numPeers; i++ {
  375. stateDBs[i] = initializeValidatorState(t, val, height)
  376. }
  377. rts := setup(t, stateDBs, 0)
  378. rts.start(t)
  379. evList := createEvidenceList(t, rts.pools[rts.network.RandomNode().NodeID], val, numEvidence)
  380. // every suite (reactor) connects to every other suite (reactor)
  381. for outerID, outerChan := range rts.peerChans {
  382. for innerID := range rts.peerChans {
  383. if outerID != innerID {
  384. outerChan <- p2p.PeerUpdate{
  385. Status: p2p.PeerStatusUp,
  386. NodeID: innerID,
  387. }
  388. }
  389. }
  390. }
  391. // wait till all suites (reactors) received all evidence from other suites (reactors)
  392. rts.waitForEvidence(t, evList)
  393. for _, pool := range rts.pools {
  394. require.Equal(t, numEvidence, int(pool.Size()))
  395. // commit state so we do not continue to repeat gossiping the same evidence
  396. state := pool.State()
  397. state.LastBlockHeight++
  398. pool.Update(state, evList)
  399. }
  400. }
  401. // nolint:lll
  402. func TestEvidenceListSerialization(t *testing.T) {
  403. exampleVote := func(msgType byte) *types.Vote {
  404. var stamp, err = time.Parse(types.TimeFormat, "2017-12-25T03:00:01.234Z")
  405. require.NoError(t, err)
  406. return &types.Vote{
  407. Type: tmproto.SignedMsgType(msgType),
  408. Height: 3,
  409. Round: 2,
  410. Timestamp: stamp,
  411. BlockID: types.BlockID{
  412. Hash: tmhash.Sum([]byte("blockID_hash")),
  413. PartSetHeader: types.PartSetHeader{
  414. Total: 1000000,
  415. Hash: tmhash.Sum([]byte("blockID_part_set_header_hash")),
  416. },
  417. },
  418. ValidatorAddress: crypto.AddressHash([]byte("validator_address")),
  419. ValidatorIndex: 56789,
  420. }
  421. }
  422. val := &types.Validator{
  423. Address: crypto.AddressHash([]byte("validator_address")),
  424. VotingPower: 10,
  425. }
  426. valSet := types.NewValidatorSet([]*types.Validator{val})
  427. dupl := types.NewDuplicateVoteEvidence(
  428. exampleVote(1),
  429. exampleVote(2),
  430. defaultEvidenceTime,
  431. valSet,
  432. )
  433. testCases := map[string]struct {
  434. evidenceList []types.Evidence
  435. expBytes string
  436. }{
  437. "DuplicateVoteEvidence": {
  438. []types.Evidence{dupl},
  439. "0a85020a82020a79080210031802224a0a208b01023386c371778ecb6368573e539afc3cc860ec3a2f614e54fe5652f4fc80122608c0843d122072db3d959635dff1bb567bedaa70573392c5159666a3f8caf11e413aac52207a2a0b08b1d381d20510809dca6f32146af1f4111082efb388211bc72c55bcd61e9ac3d538d5bb031279080110031802224a0a208b01023386c371778ecb6368573e539afc3cc860ec3a2f614e54fe5652f4fc80122608c0843d122072db3d959635dff1bb567bedaa70573392c5159666a3f8caf11e413aac52207a2a0b08b1d381d20510809dca6f32146af1f4111082efb388211bc72c55bcd61e9ac3d538d5bb03180a200a2a060880dbaae105",
  440. },
  441. }
  442. for name, tc := range testCases {
  443. tc := tc
  444. t.Run(name, func(t *testing.T) {
  445. protoEv := make([]tmproto.Evidence, len(tc.evidenceList))
  446. for i := 0; i < len(tc.evidenceList); i++ {
  447. ev, err := types.EvidenceToProto(tc.evidenceList[i])
  448. require.NoError(t, err)
  449. protoEv[i] = *ev
  450. }
  451. epl := tmproto.EvidenceList{
  452. Evidence: protoEv,
  453. }
  454. bz, err := epl.Marshal()
  455. require.NoError(t, err)
  456. require.Equal(t, tc.expBytes, hex.EncodeToString(bz))
  457. })
  458. }
  459. }