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.

661 lines
25 KiB

  1. package statesync
  2. import (
  3. "errors"
  4. "testing"
  5. "time"
  6. "github.com/stretchr/testify/assert"
  7. "github.com/stretchr/testify/mock"
  8. "github.com/stretchr/testify/require"
  9. abci "github.com/tendermint/tendermint/abci/types"
  10. "github.com/tendermint/tendermint/libs/log"
  11. tmsync "github.com/tendermint/tendermint/libs/sync"
  12. "github.com/tendermint/tendermint/p2p"
  13. p2pmocks "github.com/tendermint/tendermint/p2p/mocks"
  14. tmstate "github.com/tendermint/tendermint/proto/tendermint/state"
  15. ssproto "github.com/tendermint/tendermint/proto/tendermint/statesync"
  16. tmversion "github.com/tendermint/tendermint/proto/tendermint/version"
  17. "github.com/tendermint/tendermint/proxy"
  18. proxymocks "github.com/tendermint/tendermint/proxy/mocks"
  19. sm "github.com/tendermint/tendermint/state"
  20. "github.com/tendermint/tendermint/statesync/mocks"
  21. "github.com/tendermint/tendermint/types"
  22. "github.com/tendermint/tendermint/version"
  23. )
  24. // Sets up a basic syncer that can be used to test OfferSnapshot requests
  25. func setupOfferSyncer(t *testing.T) (*syncer, *proxymocks.AppConnSnapshot) {
  26. connQuery := &proxymocks.AppConnQuery{}
  27. connSnapshot := &proxymocks.AppConnSnapshot{}
  28. stateProvider := &mocks.StateProvider{}
  29. stateProvider.On("AppHash", mock.Anything).Return([]byte("app_hash"), nil)
  30. syncer := newSyncer(log.NewNopLogger(), connSnapshot, connQuery, stateProvider, "")
  31. return syncer, connSnapshot
  32. }
  33. // Sets up a simple peer mock with an ID
  34. func simplePeer(id string) *p2pmocks.Peer {
  35. peer := &p2pmocks.Peer{}
  36. peer.On("ID").Return(p2p.ID(id))
  37. return peer
  38. }
  39. func TestSyncer_SyncAny(t *testing.T) {
  40. state := sm.State{
  41. ChainID: "chain",
  42. Version: tmstate.Version{
  43. Consensus: tmversion.Consensus{
  44. Block: version.BlockProtocol,
  45. App: 0,
  46. },
  47. Software: version.TMCoreSemVer,
  48. },
  49. LastBlockHeight: 1,
  50. LastBlockID: types.BlockID{Hash: []byte("blockhash")},
  51. LastBlockTime: time.Now(),
  52. LastResultsHash: []byte("last_results_hash"),
  53. AppHash: []byte("app_hash"),
  54. LastValidators: &types.ValidatorSet{Proposer: &types.Validator{Address: []byte("val1")}},
  55. Validators: &types.ValidatorSet{Proposer: &types.Validator{Address: []byte("val2")}},
  56. NextValidators: &types.ValidatorSet{Proposer: &types.Validator{Address: []byte("val3")}},
  57. ConsensusParams: *types.DefaultConsensusParams(),
  58. LastHeightConsensusParamsChanged: 1,
  59. }
  60. commit := &types.Commit{BlockID: types.BlockID{Hash: []byte("blockhash")}}
  61. chunks := []*chunk{
  62. {Height: 1, Format: 1, Index: 0, Chunk: []byte{1, 1, 0}},
  63. {Height: 1, Format: 1, Index: 1, Chunk: []byte{1, 1, 1}},
  64. {Height: 1, Format: 1, Index: 2, Chunk: []byte{1, 1, 2}},
  65. }
  66. s := &snapshot{Height: 1, Format: 1, Chunks: 3, Hash: []byte{1, 2, 3}}
  67. stateProvider := &mocks.StateProvider{}
  68. stateProvider.On("AppHash", uint64(1)).Return(state.AppHash, nil)
  69. stateProvider.On("AppHash", uint64(2)).Return([]byte("app_hash_2"), nil)
  70. stateProvider.On("Commit", uint64(1)).Return(commit, nil)
  71. stateProvider.On("State", uint64(1)).Return(state, nil)
  72. connSnapshot := &proxymocks.AppConnSnapshot{}
  73. connQuery := &proxymocks.AppConnQuery{}
  74. syncer := newSyncer(log.NewNopLogger(), connSnapshot, connQuery, stateProvider, "")
  75. // Adding a chunk should error when no sync is in progress
  76. _, err := syncer.AddChunk(&chunk{Height: 1, Format: 1, Index: 0, Chunk: []byte{1}})
  77. require.Error(t, err)
  78. // Adding a couple of peers should trigger snapshot discovery messages
  79. peerA := &p2pmocks.Peer{}
  80. peerA.On("ID").Return(p2p.ID("a"))
  81. peerA.On("Send", SnapshotChannel, mustEncodeMsg(&ssproto.SnapshotsRequest{})).Return(true)
  82. syncer.AddPeer(peerA)
  83. peerA.AssertExpectations(t)
  84. peerB := &p2pmocks.Peer{}
  85. peerB.On("ID").Return(p2p.ID("b"))
  86. peerB.On("Send", SnapshotChannel, mustEncodeMsg(&ssproto.SnapshotsRequest{})).Return(true)
  87. syncer.AddPeer(peerB)
  88. peerB.AssertExpectations(t)
  89. // Both peers report back with snapshots. One of them also returns a snapshot we don't want, in
  90. // format 2, which will be rejected by the ABCI application.
  91. new, err := syncer.AddSnapshot(peerA, s)
  92. require.NoError(t, err)
  93. assert.True(t, new)
  94. new, err = syncer.AddSnapshot(peerB, s)
  95. require.NoError(t, err)
  96. assert.False(t, new)
  97. new, err = syncer.AddSnapshot(peerB, &snapshot{Height: 2, Format: 2, Chunks: 3, Hash: []byte{1}})
  98. require.NoError(t, err)
  99. assert.True(t, new)
  100. // We start a sync, with peers sending back chunks when requested. We first reject the snapshot
  101. // with height 2 format 2, and accept the snapshot at height 1.
  102. connSnapshot.On("OfferSnapshotSync", abci.RequestOfferSnapshot{
  103. Snapshot: &abci.Snapshot{
  104. Height: 2,
  105. Format: 2,
  106. Chunks: 3,
  107. Hash: []byte{1},
  108. },
  109. AppHash: []byte("app_hash_2"),
  110. }).Return(&abci.ResponseOfferSnapshot{Result: abci.ResponseOfferSnapshot_REJECT_FORMAT}, nil)
  111. connSnapshot.On("OfferSnapshotSync", abci.RequestOfferSnapshot{
  112. Snapshot: &abci.Snapshot{
  113. Height: s.Height,
  114. Format: s.Format,
  115. Chunks: s.Chunks,
  116. Hash: s.Hash,
  117. Metadata: s.Metadata,
  118. },
  119. AppHash: []byte("app_hash"),
  120. }).Times(2).Return(&abci.ResponseOfferSnapshot{Result: abci.ResponseOfferSnapshot_ACCEPT}, nil)
  121. chunkRequests := make(map[uint32]int)
  122. chunkRequestsMtx := tmsync.Mutex{}
  123. onChunkRequest := func(args mock.Arguments) {
  124. pb, err := decodeMsg(args[1].([]byte))
  125. require.NoError(t, err)
  126. msg := pb.(*ssproto.ChunkRequest)
  127. require.EqualValues(t, 1, msg.Height)
  128. require.EqualValues(t, 1, msg.Format)
  129. require.LessOrEqual(t, msg.Index, uint32(len(chunks)))
  130. added, err := syncer.AddChunk(chunks[msg.Index])
  131. require.NoError(t, err)
  132. assert.True(t, added)
  133. chunkRequestsMtx.Lock()
  134. chunkRequests[msg.Index]++
  135. chunkRequestsMtx.Unlock()
  136. }
  137. peerA.On("Send", ChunkChannel, mock.Anything).Maybe().Run(onChunkRequest).Return(true)
  138. peerB.On("Send", ChunkChannel, mock.Anything).Maybe().Run(onChunkRequest).Return(true)
  139. // The first time we're applying chunk 2 we tell it to retry the snapshot and discard chunk 1,
  140. // which should cause it to keep the existing chunk 0 and 2, and restart restoration from
  141. // beginning. We also wait for a little while, to exercise the retry logic in fetchChunks().
  142. connSnapshot.On("ApplySnapshotChunkSync", abci.RequestApplySnapshotChunk{
  143. Index: 2, Chunk: []byte{1, 1, 2},
  144. }).Once().Run(func(args mock.Arguments) { time.Sleep(2 * time.Second) }).Return(
  145. &abci.ResponseApplySnapshotChunk{
  146. Result: abci.ResponseApplySnapshotChunk_RETRY_SNAPSHOT,
  147. RefetchChunks: []uint32{1},
  148. }, nil)
  149. connSnapshot.On("ApplySnapshotChunkSync", abci.RequestApplySnapshotChunk{
  150. Index: 0, Chunk: []byte{1, 1, 0},
  151. }).Times(2).Return(&abci.ResponseApplySnapshotChunk{Result: abci.ResponseApplySnapshotChunk_ACCEPT}, nil)
  152. connSnapshot.On("ApplySnapshotChunkSync", abci.RequestApplySnapshotChunk{
  153. Index: 1, Chunk: []byte{1, 1, 1},
  154. }).Times(2).Return(&abci.ResponseApplySnapshotChunk{Result: abci.ResponseApplySnapshotChunk_ACCEPT}, nil)
  155. connSnapshot.On("ApplySnapshotChunkSync", abci.RequestApplySnapshotChunk{
  156. Index: 2, Chunk: []byte{1, 1, 2},
  157. }).Once().Return(&abci.ResponseApplySnapshotChunk{Result: abci.ResponseApplySnapshotChunk_ACCEPT}, nil)
  158. connQuery.On("InfoSync", proxy.RequestInfo).Return(&abci.ResponseInfo{
  159. AppVersion: 9,
  160. LastBlockHeight: 1,
  161. LastBlockAppHash: []byte("app_hash"),
  162. }, nil)
  163. newState, lastCommit, err := syncer.SyncAny(0)
  164. require.NoError(t, err)
  165. time.Sleep(50 * time.Millisecond) // wait for peers to receive requests
  166. chunkRequestsMtx.Lock()
  167. assert.Equal(t, map[uint32]int{0: 1, 1: 2, 2: 1}, chunkRequests)
  168. chunkRequestsMtx.Unlock()
  169. // The syncer should have updated the state app version from the ABCI info response.
  170. expectState := state
  171. expectState.Version.Consensus.App = 9
  172. assert.Equal(t, expectState, newState)
  173. assert.Equal(t, commit, lastCommit)
  174. connSnapshot.AssertExpectations(t)
  175. connQuery.AssertExpectations(t)
  176. peerA.AssertExpectations(t)
  177. peerB.AssertExpectations(t)
  178. }
  179. func TestSyncer_SyncAny_noSnapshots(t *testing.T) {
  180. syncer, _ := setupOfferSyncer(t)
  181. _, _, err := syncer.SyncAny(0)
  182. assert.Equal(t, errNoSnapshots, err)
  183. }
  184. func TestSyncer_SyncAny_abort(t *testing.T) {
  185. syncer, connSnapshot := setupOfferSyncer(t)
  186. s := &snapshot{Height: 1, Format: 1, Chunks: 3, Hash: []byte{1, 2, 3}}
  187. _, err := syncer.AddSnapshot(simplePeer("id"), s)
  188. require.NoError(t, err)
  189. connSnapshot.On("OfferSnapshotSync", abci.RequestOfferSnapshot{
  190. Snapshot: toABCI(s), AppHash: []byte("app_hash"),
  191. }).Once().Return(&abci.ResponseOfferSnapshot{Result: abci.ResponseOfferSnapshot_ABORT}, nil)
  192. _, _, err = syncer.SyncAny(0)
  193. assert.Equal(t, errAbort, err)
  194. connSnapshot.AssertExpectations(t)
  195. }
  196. func TestSyncer_SyncAny_reject(t *testing.T) {
  197. syncer, connSnapshot := setupOfferSyncer(t)
  198. // s22 is tried first, then s12, then s11, then errNoSnapshots
  199. s22 := &snapshot{Height: 2, Format: 2, Chunks: 3, Hash: []byte{1, 2, 3}}
  200. s12 := &snapshot{Height: 1, Format: 2, Chunks: 3, Hash: []byte{1, 2, 3}}
  201. s11 := &snapshot{Height: 1, Format: 1, Chunks: 3, Hash: []byte{1, 2, 3}}
  202. _, err := syncer.AddSnapshot(simplePeer("id"), s22)
  203. require.NoError(t, err)
  204. _, err = syncer.AddSnapshot(simplePeer("id"), s12)
  205. require.NoError(t, err)
  206. _, err = syncer.AddSnapshot(simplePeer("id"), s11)
  207. require.NoError(t, err)
  208. connSnapshot.On("OfferSnapshotSync", abci.RequestOfferSnapshot{
  209. Snapshot: toABCI(s22), AppHash: []byte("app_hash"),
  210. }).Once().Return(&abci.ResponseOfferSnapshot{Result: abci.ResponseOfferSnapshot_REJECT}, nil)
  211. connSnapshot.On("OfferSnapshotSync", abci.RequestOfferSnapshot{
  212. Snapshot: toABCI(s12), AppHash: []byte("app_hash"),
  213. }).Once().Return(&abci.ResponseOfferSnapshot{Result: abci.ResponseOfferSnapshot_REJECT}, nil)
  214. connSnapshot.On("OfferSnapshotSync", abci.RequestOfferSnapshot{
  215. Snapshot: toABCI(s11), AppHash: []byte("app_hash"),
  216. }).Once().Return(&abci.ResponseOfferSnapshot{Result: abci.ResponseOfferSnapshot_REJECT}, nil)
  217. _, _, err = syncer.SyncAny(0)
  218. assert.Equal(t, errNoSnapshots, err)
  219. connSnapshot.AssertExpectations(t)
  220. }
  221. func TestSyncer_SyncAny_reject_format(t *testing.T) {
  222. syncer, connSnapshot := setupOfferSyncer(t)
  223. // s22 is tried first, which reject s22 and s12, then s11 will abort.
  224. s22 := &snapshot{Height: 2, Format: 2, Chunks: 3, Hash: []byte{1, 2, 3}}
  225. s12 := &snapshot{Height: 1, Format: 2, Chunks: 3, Hash: []byte{1, 2, 3}}
  226. s11 := &snapshot{Height: 1, Format: 1, Chunks: 3, Hash: []byte{1, 2, 3}}
  227. _, err := syncer.AddSnapshot(simplePeer("id"), s22)
  228. require.NoError(t, err)
  229. _, err = syncer.AddSnapshot(simplePeer("id"), s12)
  230. require.NoError(t, err)
  231. _, err = syncer.AddSnapshot(simplePeer("id"), s11)
  232. require.NoError(t, err)
  233. connSnapshot.On("OfferSnapshotSync", abci.RequestOfferSnapshot{
  234. Snapshot: toABCI(s22), AppHash: []byte("app_hash"),
  235. }).Once().Return(&abci.ResponseOfferSnapshot{Result: abci.ResponseOfferSnapshot_REJECT_FORMAT}, nil)
  236. connSnapshot.On("OfferSnapshotSync", abci.RequestOfferSnapshot{
  237. Snapshot: toABCI(s11), AppHash: []byte("app_hash"),
  238. }).Once().Return(&abci.ResponseOfferSnapshot{Result: abci.ResponseOfferSnapshot_ABORT}, nil)
  239. _, _, err = syncer.SyncAny(0)
  240. assert.Equal(t, errAbort, err)
  241. connSnapshot.AssertExpectations(t)
  242. }
  243. func TestSyncer_SyncAny_reject_sender(t *testing.T) {
  244. syncer, connSnapshot := setupOfferSyncer(t)
  245. peerA := simplePeer("a")
  246. peerB := simplePeer("b")
  247. peerC := simplePeer("c")
  248. // sbc will be offered first, which will be rejected with reject_sender, causing all snapshots
  249. // submitted by both b and c (i.e. sb, sc, sbc) to be rejected. Finally, sa will reject and
  250. // errNoSnapshots is returned.
  251. sa := &snapshot{Height: 1, Format: 1, Chunks: 3, Hash: []byte{1, 2, 3}}
  252. sb := &snapshot{Height: 2, Format: 1, Chunks: 3, Hash: []byte{1, 2, 3}}
  253. sc := &snapshot{Height: 3, Format: 1, Chunks: 3, Hash: []byte{1, 2, 3}}
  254. sbc := &snapshot{Height: 4, Format: 1, Chunks: 3, Hash: []byte{1, 2, 3}}
  255. _, err := syncer.AddSnapshot(peerA, sa)
  256. require.NoError(t, err)
  257. _, err = syncer.AddSnapshot(peerB, sb)
  258. require.NoError(t, err)
  259. _, err = syncer.AddSnapshot(peerC, sc)
  260. require.NoError(t, err)
  261. _, err = syncer.AddSnapshot(peerB, sbc)
  262. require.NoError(t, err)
  263. _, err = syncer.AddSnapshot(peerC, sbc)
  264. require.NoError(t, err)
  265. connSnapshot.On("OfferSnapshotSync", abci.RequestOfferSnapshot{
  266. Snapshot: toABCI(sbc), AppHash: []byte("app_hash"),
  267. }).Once().Return(&abci.ResponseOfferSnapshot{Result: abci.ResponseOfferSnapshot_REJECT_SENDER}, nil)
  268. connSnapshot.On("OfferSnapshotSync", abci.RequestOfferSnapshot{
  269. Snapshot: toABCI(sa), AppHash: []byte("app_hash"),
  270. }).Once().Return(&abci.ResponseOfferSnapshot{Result: abci.ResponseOfferSnapshot_REJECT}, nil)
  271. _, _, err = syncer.SyncAny(0)
  272. assert.Equal(t, errNoSnapshots, err)
  273. connSnapshot.AssertExpectations(t)
  274. }
  275. func TestSyncer_SyncAny_abciError(t *testing.T) {
  276. syncer, connSnapshot := setupOfferSyncer(t)
  277. errBoom := errors.New("boom")
  278. s := &snapshot{Height: 1, Format: 1, Chunks: 3, Hash: []byte{1, 2, 3}}
  279. _, err := syncer.AddSnapshot(simplePeer("id"), s)
  280. require.NoError(t, err)
  281. connSnapshot.On("OfferSnapshotSync", abci.RequestOfferSnapshot{
  282. Snapshot: toABCI(s), AppHash: []byte("app_hash"),
  283. }).Once().Return(nil, errBoom)
  284. _, _, err = syncer.SyncAny(0)
  285. assert.True(t, errors.Is(err, errBoom))
  286. connSnapshot.AssertExpectations(t)
  287. }
  288. func TestSyncer_offerSnapshot(t *testing.T) {
  289. unknownErr := errors.New("unknown error")
  290. boom := errors.New("boom")
  291. testcases := map[string]struct {
  292. result abci.ResponseOfferSnapshot_Result
  293. err error
  294. expectErr error
  295. }{
  296. "accept": {abci.ResponseOfferSnapshot_ACCEPT, nil, nil},
  297. "abort": {abci.ResponseOfferSnapshot_ABORT, nil, errAbort},
  298. "reject": {abci.ResponseOfferSnapshot_REJECT, nil, errRejectSnapshot},
  299. "reject_format": {abci.ResponseOfferSnapshot_REJECT_FORMAT, nil, errRejectFormat},
  300. "reject_sender": {abci.ResponseOfferSnapshot_REJECT_SENDER, nil, errRejectSender},
  301. "unknown": {abci.ResponseOfferSnapshot_UNKNOWN, nil, unknownErr},
  302. "error": {0, boom, boom},
  303. "unknown non-zero": {9, nil, unknownErr},
  304. }
  305. for name, tc := range testcases {
  306. tc := tc
  307. t.Run(name, func(t *testing.T) {
  308. syncer, connSnapshot := setupOfferSyncer(t)
  309. s := &snapshot{Height: 1, Format: 1, Chunks: 3, Hash: []byte{1, 2, 3}, trustedAppHash: []byte("app_hash")}
  310. connSnapshot.On("OfferSnapshotSync", abci.RequestOfferSnapshot{
  311. Snapshot: toABCI(s),
  312. AppHash: []byte("app_hash"),
  313. }).Return(&abci.ResponseOfferSnapshot{Result: tc.result}, tc.err)
  314. err := syncer.offerSnapshot(s)
  315. if tc.expectErr == unknownErr {
  316. require.Error(t, err)
  317. } else {
  318. unwrapped := errors.Unwrap(err)
  319. if unwrapped != nil {
  320. err = unwrapped
  321. }
  322. assert.Equal(t, tc.expectErr, err)
  323. }
  324. })
  325. }
  326. }
  327. func TestSyncer_applyChunks_Results(t *testing.T) {
  328. unknownErr := errors.New("unknown error")
  329. boom := errors.New("boom")
  330. testcases := map[string]struct {
  331. result abci.ResponseApplySnapshotChunk_Result
  332. err error
  333. expectErr error
  334. }{
  335. "accept": {abci.ResponseApplySnapshotChunk_ACCEPT, nil, nil},
  336. "abort": {abci.ResponseApplySnapshotChunk_ABORT, nil, errAbort},
  337. "retry": {abci.ResponseApplySnapshotChunk_RETRY, nil, nil},
  338. "retry_snapshot": {abci.ResponseApplySnapshotChunk_RETRY_SNAPSHOT, nil, errRetrySnapshot},
  339. "reject_snapshot": {abci.ResponseApplySnapshotChunk_REJECT_SNAPSHOT, nil, errRejectSnapshot},
  340. "unknown": {abci.ResponseApplySnapshotChunk_UNKNOWN, nil, unknownErr},
  341. "error": {0, boom, boom},
  342. "unknown non-zero": {9, nil, unknownErr},
  343. }
  344. for name, tc := range testcases {
  345. tc := tc
  346. t.Run(name, func(t *testing.T) {
  347. connQuery := &proxymocks.AppConnQuery{}
  348. connSnapshot := &proxymocks.AppConnSnapshot{}
  349. stateProvider := &mocks.StateProvider{}
  350. stateProvider.On("AppHash", mock.Anything).Return([]byte("app_hash"), nil)
  351. syncer := newSyncer(log.NewNopLogger(), connSnapshot, connQuery, stateProvider, "")
  352. body := []byte{1, 2, 3}
  353. chunks, err := newChunkQueue(&snapshot{Height: 1, Format: 1, Chunks: 1}, "")
  354. require.NoError(t, err)
  355. _, err = chunks.Add(&chunk{Height: 1, Format: 1, Index: 0, Chunk: body})
  356. require.NoError(t, err)
  357. connSnapshot.On("ApplySnapshotChunkSync", abci.RequestApplySnapshotChunk{
  358. Index: 0, Chunk: body,
  359. }).Once().Return(&abci.ResponseApplySnapshotChunk{Result: tc.result}, tc.err)
  360. if tc.result == abci.ResponseApplySnapshotChunk_RETRY {
  361. connSnapshot.On("ApplySnapshotChunkSync", abci.RequestApplySnapshotChunk{
  362. Index: 0, Chunk: body,
  363. }).Once().Return(&abci.ResponseApplySnapshotChunk{
  364. Result: abci.ResponseApplySnapshotChunk_ACCEPT}, nil)
  365. }
  366. err = syncer.applyChunks(chunks)
  367. if tc.expectErr == unknownErr {
  368. require.Error(t, err)
  369. } else {
  370. unwrapped := errors.Unwrap(err)
  371. if unwrapped != nil {
  372. err = unwrapped
  373. }
  374. assert.Equal(t, tc.expectErr, err)
  375. }
  376. connSnapshot.AssertExpectations(t)
  377. })
  378. }
  379. }
  380. func TestSyncer_applyChunks_RefetchChunks(t *testing.T) {
  381. // Discarding chunks via refetch_chunks should work the same for all results
  382. testcases := map[string]struct {
  383. result abci.ResponseApplySnapshotChunk_Result
  384. }{
  385. "accept": {abci.ResponseApplySnapshotChunk_ACCEPT},
  386. "abort": {abci.ResponseApplySnapshotChunk_ABORT},
  387. "retry": {abci.ResponseApplySnapshotChunk_RETRY},
  388. "retry_snapshot": {abci.ResponseApplySnapshotChunk_RETRY_SNAPSHOT},
  389. "reject_snapshot": {abci.ResponseApplySnapshotChunk_REJECT_SNAPSHOT},
  390. }
  391. for name, tc := range testcases {
  392. tc := tc
  393. t.Run(name, func(t *testing.T) {
  394. connQuery := &proxymocks.AppConnQuery{}
  395. connSnapshot := &proxymocks.AppConnSnapshot{}
  396. stateProvider := &mocks.StateProvider{}
  397. stateProvider.On("AppHash", mock.Anything).Return([]byte("app_hash"), nil)
  398. syncer := newSyncer(log.NewNopLogger(), connSnapshot, connQuery, stateProvider, "")
  399. chunks, err := newChunkQueue(&snapshot{Height: 1, Format: 1, Chunks: 3}, "")
  400. require.NoError(t, err)
  401. added, err := chunks.Add(&chunk{Height: 1, Format: 1, Index: 0, Chunk: []byte{0}})
  402. require.True(t, added)
  403. require.NoError(t, err)
  404. added, err = chunks.Add(&chunk{Height: 1, Format: 1, Index: 1, Chunk: []byte{1}})
  405. require.True(t, added)
  406. require.NoError(t, err)
  407. added, err = chunks.Add(&chunk{Height: 1, Format: 1, Index: 2, Chunk: []byte{2}})
  408. require.True(t, added)
  409. require.NoError(t, err)
  410. // The first two chunks are accepted, before the last one asks for 1 to be refetched
  411. connSnapshot.On("ApplySnapshotChunkSync", abci.RequestApplySnapshotChunk{
  412. Index: 0, Chunk: []byte{0},
  413. }).Once().Return(&abci.ResponseApplySnapshotChunk{Result: abci.ResponseApplySnapshotChunk_ACCEPT}, nil)
  414. connSnapshot.On("ApplySnapshotChunkSync", abci.RequestApplySnapshotChunk{
  415. Index: 1, Chunk: []byte{1},
  416. }).Once().Return(&abci.ResponseApplySnapshotChunk{Result: abci.ResponseApplySnapshotChunk_ACCEPT}, nil)
  417. connSnapshot.On("ApplySnapshotChunkSync", abci.RequestApplySnapshotChunk{
  418. Index: 2, Chunk: []byte{2},
  419. }).Once().Return(&abci.ResponseApplySnapshotChunk{
  420. Result: tc.result,
  421. RefetchChunks: []uint32{1},
  422. }, nil)
  423. // Since removing the chunk will cause Next() to block, we spawn a goroutine, then
  424. // check the queue contents, and finally close the queue to end the goroutine.
  425. // We don't really care about the result of applyChunks, since it has separate test.
  426. go func() {
  427. syncer.applyChunks(chunks) //nolint:errcheck // purposefully ignore error
  428. }()
  429. time.Sleep(50 * time.Millisecond)
  430. assert.True(t, chunks.Has(0))
  431. assert.False(t, chunks.Has(1))
  432. assert.True(t, chunks.Has(2))
  433. err = chunks.Close()
  434. require.NoError(t, err)
  435. })
  436. }
  437. }
  438. func TestSyncer_applyChunks_RejectSenders(t *testing.T) {
  439. // Banning chunks senders via ban_chunk_senders should work the same for all results
  440. testcases := map[string]struct {
  441. result abci.ResponseApplySnapshotChunk_Result
  442. }{
  443. "accept": {abci.ResponseApplySnapshotChunk_ACCEPT},
  444. "abort": {abci.ResponseApplySnapshotChunk_ABORT},
  445. "retry": {abci.ResponseApplySnapshotChunk_RETRY},
  446. "retry_snapshot": {abci.ResponseApplySnapshotChunk_RETRY_SNAPSHOT},
  447. "reject_snapshot": {abci.ResponseApplySnapshotChunk_REJECT_SNAPSHOT},
  448. }
  449. for name, tc := range testcases {
  450. tc := tc
  451. t.Run(name, func(t *testing.T) {
  452. connQuery := &proxymocks.AppConnQuery{}
  453. connSnapshot := &proxymocks.AppConnSnapshot{}
  454. stateProvider := &mocks.StateProvider{}
  455. stateProvider.On("AppHash", mock.Anything).Return([]byte("app_hash"), nil)
  456. syncer := newSyncer(log.NewNopLogger(), connSnapshot, connQuery, stateProvider, "")
  457. // Set up three peers across two snapshots, and ask for one of them to be banned.
  458. // It should be banned from all snapshots.
  459. peerA := simplePeer("a")
  460. peerB := simplePeer("b")
  461. peerC := simplePeer("c")
  462. s1 := &snapshot{Height: 1, Format: 1, Chunks: 3}
  463. s2 := &snapshot{Height: 2, Format: 1, Chunks: 3}
  464. _, err := syncer.AddSnapshot(peerA, s1)
  465. require.NoError(t, err)
  466. _, err = syncer.AddSnapshot(peerA, s2)
  467. require.NoError(t, err)
  468. _, err = syncer.AddSnapshot(peerB, s1)
  469. require.NoError(t, err)
  470. _, err = syncer.AddSnapshot(peerB, s2)
  471. require.NoError(t, err)
  472. _, err = syncer.AddSnapshot(peerC, s1)
  473. require.NoError(t, err)
  474. _, err = syncer.AddSnapshot(peerC, s2)
  475. require.NoError(t, err)
  476. chunks, err := newChunkQueue(s1, "")
  477. require.NoError(t, err)
  478. added, err := chunks.Add(&chunk{Height: 1, Format: 1, Index: 0, Chunk: []byte{0}, Sender: peerA.ID()})
  479. require.True(t, added)
  480. require.NoError(t, err)
  481. added, err = chunks.Add(&chunk{Height: 1, Format: 1, Index: 1, Chunk: []byte{1}, Sender: peerB.ID()})
  482. require.True(t, added)
  483. require.NoError(t, err)
  484. added, err = chunks.Add(&chunk{Height: 1, Format: 1, Index: 2, Chunk: []byte{2}, Sender: peerC.ID()})
  485. require.True(t, added)
  486. require.NoError(t, err)
  487. // The first two chunks are accepted, before the last one asks for b sender to be rejected
  488. connSnapshot.On("ApplySnapshotChunkSync", abci.RequestApplySnapshotChunk{
  489. Index: 0, Chunk: []byte{0}, Sender: "a",
  490. }).Once().Return(&abci.ResponseApplySnapshotChunk{Result: abci.ResponseApplySnapshotChunk_ACCEPT}, nil)
  491. connSnapshot.On("ApplySnapshotChunkSync", abci.RequestApplySnapshotChunk{
  492. Index: 1, Chunk: []byte{1}, Sender: "b",
  493. }).Once().Return(&abci.ResponseApplySnapshotChunk{Result: abci.ResponseApplySnapshotChunk_ACCEPT}, nil)
  494. connSnapshot.On("ApplySnapshotChunkSync", abci.RequestApplySnapshotChunk{
  495. Index: 2, Chunk: []byte{2}, Sender: "c",
  496. }).Once().Return(&abci.ResponseApplySnapshotChunk{
  497. Result: tc.result,
  498. RejectSenders: []string{string(peerB.ID())},
  499. }, nil)
  500. // On retry, the last chunk will be tried again, so we just accept it then.
  501. if tc.result == abci.ResponseApplySnapshotChunk_RETRY {
  502. connSnapshot.On("ApplySnapshotChunkSync", abci.RequestApplySnapshotChunk{
  503. Index: 2, Chunk: []byte{2}, Sender: "c",
  504. }).Once().Return(&abci.ResponseApplySnapshotChunk{Result: abci.ResponseApplySnapshotChunk_ACCEPT}, nil)
  505. }
  506. // We don't really care about the result of applyChunks, since it has separate test.
  507. // However, it will block on e.g. retry result, so we spawn a goroutine that will
  508. // be shut down when the chunk queue closes.
  509. go func() {
  510. syncer.applyChunks(chunks) //nolint:errcheck // purposefully ignore error
  511. }()
  512. time.Sleep(50 * time.Millisecond)
  513. s1peers := syncer.snapshots.GetPeers(s1)
  514. assert.Len(t, s1peers, 2)
  515. assert.EqualValues(t, "a", s1peers[0].ID())
  516. assert.EqualValues(t, "c", s1peers[1].ID())
  517. syncer.snapshots.GetPeers(s1)
  518. assert.Len(t, s1peers, 2)
  519. assert.EqualValues(t, "a", s1peers[0].ID())
  520. assert.EqualValues(t, "c", s1peers[1].ID())
  521. err = chunks.Close()
  522. require.NoError(t, err)
  523. })
  524. }
  525. }
  526. func TestSyncer_verifyApp(t *testing.T) {
  527. boom := errors.New("boom")
  528. s := &snapshot{Height: 3, Format: 1, Chunks: 5, Hash: []byte{1, 2, 3}, trustedAppHash: []byte("app_hash")}
  529. testcases := map[string]struct {
  530. response *abci.ResponseInfo
  531. err error
  532. expectErr error
  533. }{
  534. "verified": {&abci.ResponseInfo{
  535. LastBlockHeight: 3,
  536. LastBlockAppHash: []byte("app_hash"),
  537. AppVersion: 9,
  538. }, nil, nil},
  539. "invalid height": {&abci.ResponseInfo{
  540. LastBlockHeight: 5,
  541. LastBlockAppHash: []byte("app_hash"),
  542. AppVersion: 9,
  543. }, nil, errVerifyFailed},
  544. "invalid hash": {&abci.ResponseInfo{
  545. LastBlockHeight: 3,
  546. LastBlockAppHash: []byte("xxx"),
  547. AppVersion: 9,
  548. }, nil, errVerifyFailed},
  549. "error": {nil, boom, boom},
  550. }
  551. for name, tc := range testcases {
  552. tc := tc
  553. t.Run(name, func(t *testing.T) {
  554. connQuery := &proxymocks.AppConnQuery{}
  555. connSnapshot := &proxymocks.AppConnSnapshot{}
  556. stateProvider := &mocks.StateProvider{}
  557. syncer := newSyncer(log.NewNopLogger(), connSnapshot, connQuery, stateProvider, "")
  558. connQuery.On("InfoSync", proxy.RequestInfo).Return(tc.response, tc.err)
  559. version, err := syncer.verifyApp(s)
  560. unwrapped := errors.Unwrap(err)
  561. if unwrapped != nil {
  562. err = unwrapped
  563. }
  564. assert.Equal(t, tc.expectErr, err)
  565. if err == nil {
  566. assert.Equal(t, tc.response.AppVersion, version)
  567. }
  568. })
  569. }
  570. }
  571. func toABCI(s *snapshot) *abci.Snapshot {
  572. return &abci.Snapshot{
  573. Height: s.Height,
  574. Format: s.Format,
  575. Chunks: s.Chunks,
  576. Hash: s.Hash,
  577. Metadata: s.Metadata,
  578. }
  579. }