blockstore: fix problem with seen commit (#6782)

abci/client/mocks/client.go

@@ -1,4 +1,4 @@
-// Code generated by mockery v0.0.0-dev. DO NOT EDIT.
+// Code generated by mockery 2.9.0. DO NOT EDIT.
 
 package mocks
 

internal/consensus/mocks/cons_sync_reactor.go

@@ -1,4 +1,4 @@
-// Code generated by mockery 2.7.5. DO NOT EDIT.
+// Code generated by mockery 2.9.0. DO NOT EDIT.
 
 package mocks
 

internal/consensus/replay_test.go

@@ -1203,8 +1203,8 @@ func (bs *mockBlockStore) SaveBlock(block *types.Block, blockParts *types.PartSe
 func (bs *mockBlockStore) LoadBlockCommit(height int64) *types.Commit {
 	return bs.commits[height-1]
 }
-func (bs *mockBlockStore) LoadSeenCommit(height int64) *types.Commit {
-	return bs.commits[height-1]
+func (bs *mockBlockStore) LoadSeenCommit() *types.Commit {
+	return bs.commits[len(bs.commits)-1]
 }
 
 func (bs *mockBlockStore) PruneBlocks(height int64) (uint64, error) {

internal/consensus/state.go

@@ -314,7 +314,14 @@ func (cs *State) LoadCommit(height int64) *types.Commit {
 	defer cs.mtx.RUnlock()
 
 	if height == cs.blockStore.Height() {
-		return cs.blockStore.LoadSeenCommit(height)
+		commit := cs.blockStore.LoadSeenCommit()
+		// NOTE: Retrieving the height of the most recent block and retrieving
+		// the most recent commit does not currently occur as an atomic
+		// operation. We check the height and commit here in case a more recent
+		// commit has arrived since retrieving the latest height.
+		if commit != nil && commit.Height == height {
+			return commit
+		}
 	}
 
 	return cs.blockStore.LoadBlockCommit(height)
@@ -594,15 +601,19 @@ func (cs *State) sendInternalMessage(mi msgInfo) {
 // Reconstruct LastCommit from SeenCommit, which we saved along with the block,
 // (which happens even before saving the state)
 func (cs *State) reconstructLastCommit(state sm.State) {
-	seenCommit := cs.blockStore.LoadSeenCommit(state.LastBlockHeight)
-	if seenCommit == nil {
+	commit := cs.blockStore.LoadSeenCommit()
+	if commit == nil || commit.Height != state.LastBlockHeight {
+		commit = cs.blockStore.LoadBlockCommit(state.LastBlockHeight)
+	}
+
+	if commit == nil {
 		panic(fmt.Sprintf(
-			"failed to reconstruct last commit; seen commit for height %v not found",
+			"failed to reconstruct last commit; commit for height %v not found",
 			state.LastBlockHeight,
 		))
 	}
 
-	lastPrecommits := types.CommitToVoteSet(state.ChainID, seenCommit, state.LastValidators)
+	lastPrecommits := types.CommitToVoteSet(state.ChainID, commit, state.LastValidators)
 	if !lastPrecommits.HasTwoThirdsMajority() {
 		panic("failed to reconstruct last commit; does not have +2/3 maj")
 	}
@@ -2318,7 +2329,7 @@ func (cs *State) checkDoubleSigningRisk(height int64) error {
 		}
 
 		for i := int64(1); i < doubleSignCheckHeight; i++ {
-			lastCommit := cs.blockStore.LoadSeenCommit(height - i)
+			lastCommit := cs.LoadCommit(height - i)
 			if lastCommit != nil {
 				for sigIdx, s := range lastCommit.Signatures {
 					if s.BlockIDFlag == types.BlockIDFlagCommit && bytes.Equal(s.ValidatorAddress, valAddr) {

internal/evidence/mocks/block_store.go

@@ -1,4 +1,4 @@
-// Code generated by mockery v0.0.0-dev. DO NOT EDIT.
+// Code generated by mockery 2.9.0. DO NOT EDIT.
 
 package mocks
 

internal/p2p/mocks/connection.go

@@ -1,4 +1,4 @@
-// Code generated by mockery v0.0.0-dev. DO NOT EDIT.
+// Code generated by mockery 2.9.0. DO NOT EDIT.
 
 package mocks
 

internal/p2p/mocks/peer.go

@@ -1,4 +1,4 @@
-// Code generated by mockery v0.0.0-dev. DO NOT EDIT.
+// Code generated by mockery 2.9.0. DO NOT EDIT.
 
 package mocks
 

internal/p2p/mocks/transport.go

@@ -1,4 +1,4 @@
-// Code generated by mockery v0.0.0-dev. DO NOT EDIT.
+// Code generated by mockery 2.9.0. DO NOT EDIT.
 
 package mocks
 

internal/statesync/mocks/state_provider.go

@@ -1,4 +1,4 @@
-// Code generated by mockery v0.0.0-dev. DO NOT EDIT.
+// Code generated by mockery 2.9.0. DO NOT EDIT.
 
 package mocks
 

light/provider/mocks/provider.go

@@ -1,4 +1,4 @@
-// Code generated by mockery v0.0.0-dev. DO NOT EDIT.
+// Code generated by mockery 2.9.0. DO NOT EDIT.
 
 package mocks
 

light/rpc/mocks/light_client.go

@@ -1,4 +1,4 @@
-// Code generated by mockery v0.0.0-dev. DO NOT EDIT.
+// Code generated by mockery 2.9.0. DO NOT EDIT.
 
 package mocks
 

proxy/mocks/app_conn_consensus.go

@@ -1,4 +1,4 @@
-// Code generated by mockery v0.0.0-dev. DO NOT EDIT.
+// Code generated by mockery 2.9.0. DO NOT EDIT.
 
 package mocks
 

proxy/mocks/app_conn_mempool.go

@@ -1,4 +1,4 @@
-// Code generated by mockery v0.0.0-dev. DO NOT EDIT.
+// Code generated by mockery 2.9.0. DO NOT EDIT.
 
 package mocks
 

proxy/mocks/app_conn_query.go

@@ -1,4 +1,4 @@
-// Code generated by mockery v0.0.0-dev. DO NOT EDIT.
+// Code generated by mockery 2.9.0. DO NOT EDIT.
 
 package mocks
 

proxy/mocks/app_conn_snapshot.go

@@ -1,4 +1,4 @@
-// Code generated by mockery v0.0.0-dev. DO NOT EDIT.
+// Code generated by mockery 2.9.0. DO NOT EDIT.
 
 package mocks
 

rpc/client/evidence_test.go

@@ -116,12 +116,6 @@ func TestBroadcastEvidence_DuplicateVoteEvidence(t *testing.T) {
 	defer cancel()
 
 	n, config := NodeSuite(t)
-
-	// previous versions of this test used a shared fixture with
-	// other tests, and in this version we give it a little time
-	// for the node to make progress before running the test
-	time.Sleep(100 * time.Millisecond)
-
 	chainID := config.ChainID()
 
 	pv, err := privval.LoadOrGenFilePV(config.PrivValidator.KeyFile(), config.PrivValidator.StateFile())
@@ -131,6 +125,9 @@ func TestBroadcastEvidence_DuplicateVoteEvidence(t *testing.T) {
 		correct, fakes := makeEvidences(t, pv, chainID)
 		t.Logf("client %d", i)
 
+		// make sure that the node has produced enough blocks
+		waitForBlock(ctx, t, c, 2)
+
 		result, err := c.BroadcastEvidence(ctx, correct)
 		require.NoError(t, err, "BroadcastEvidence(%s) failed", correct)
 		assert.Equal(t, correct.Hash(), result.Hash, "expected result hash to match evidence hash")
@@ -171,3 +168,21 @@ func TestBroadcastEmptyEvidence(t *testing.T) {
 		assert.Error(t, err)
 	}
 }
+
+func waitForBlock(ctx context.Context, t *testing.T, c client.Client, height int64) {
+	timer := time.NewTimer(0 * time.Millisecond)
+	defer timer.Stop()
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case <-timer.C:
+			status, err := c.Status(ctx)
+			require.NoError(t, err)
+			if status.SyncInfo.LatestBlockHeight >= height {
+				return
+			}
+			timer.Reset(200 * time.Millisecond)
+		}
+	}
+}

rpc/client/mocks/client.go

@@ -1,4 +1,4 @@
-// Code generated by mockery v0.0.0-dev. DO NOT EDIT.
+// Code generated by mockery 2.9.0. DO NOT EDIT.
 
 package mocks
 

rpc/core/blocks.go

@@ -135,12 +135,19 @@ func (env *Environment) Commit(ctx *rpctypes.Context, heightPtr *int64) (*ctypes
 	// If the next block has not been committed yet,
 	// use a non-canonical commit
 	if height == env.BlockStore.Height() {
-		commit := env.BlockStore.LoadSeenCommit(height)
-		return ctypes.NewResultCommit(&header, commit, false), nil
+		commit := env.BlockStore.LoadSeenCommit()
+		// NOTE: we can't yet ensure atomicity of operations in asserting
+		// whether this is the latest height and retrieving the seen commit
+		if commit != nil && commit.Height == height {
+			return ctypes.NewResultCommit(&header, commit, false), nil
+		}
 	}
 
 	// Return the canonical commit (comes from the block at height+1)
 	commit := env.BlockStore.LoadBlockCommit(height)
+	if commit == nil {
+		return nil, nil
+	}
 	return ctypes.NewResultCommit(&header, commit, true), nil
 }
 

rpc/core/blocks_test.go

@@ -129,7 +129,7 @@ func (mockBlockStore) LoadBlock(height int64) *types.Block               { retur
 func (mockBlockStore) LoadBlockByHash(hash []byte) *types.Block          { return nil }
 func (mockBlockStore) LoadBlockPart(height int64, index int) *types.Part { return nil }
 func (mockBlockStore) LoadBlockCommit(height int64) *types.Commit        { return nil }
-func (mockBlockStore) LoadSeenCommit(height int64) *types.Commit         { return nil }
+func (mockBlockStore) LoadSeenCommit() *types.Commit                     { return nil }
 func (mockBlockStore) PruneBlocks(height int64) (uint64, error)          { return 0, nil }
 func (mockBlockStore) SaveBlock(block *types.Block, blockParts *types.PartSet, seenCommit *types.Commit) {
 }

state/indexer/mocks/event_sink.go

@@ -0,0 +1,167 @@
+// Code generated by mockery 2.9.0. DO NOT EDIT.
+
+package mocks
+
+import (
+	context "context"
+
+	mock "github.com/stretchr/testify/mock"
+	indexer "github.com/tendermint/tendermint/state/indexer"
+
+	query "github.com/tendermint/tendermint/libs/pubsub/query"
+
+	tenderminttypes "github.com/tendermint/tendermint/types"
+
+	types "github.com/tendermint/tendermint/abci/types"
+)
+
+// EventSink is an autogenerated mock type for the EventSink type
+type EventSink struct {
+	mock.Mock
+}
+
+// GetTxByHash provides a mock function with given fields: _a0
+func (_m *EventSink) GetTxByHash(_a0 []byte) (*types.TxResult, error) {
+	ret := _m.Called(_a0)
+
+	var r0 *types.TxResult
+	if rf, ok := ret.Get(0).(func([]byte) *types.TxResult); ok {
+		r0 = rf(_a0)
+	} else {
+		if ret.Get(0) != nil {
+			r0 = ret.Get(0).(*types.TxResult)
+		}
+	}
+
+	var r1 error
+	if rf, ok := ret.Get(1).(func([]byte) error); ok {
+		r1 = rf(_a0)
+	} else {
+		r1 = ret.Error(1)
+	}
+
+	return r0, r1
+}
+
+// HasBlock provides a mock function with given fields: _a0
+func (_m *EventSink) HasBlock(_a0 int64) (bool, error) {
+	ret := _m.Called(_a0)
+
+	var r0 bool
+	if rf, ok := ret.Get(0).(func(int64) bool); ok {
+		r0 = rf(_a0)
+	} else {
+		r0 = ret.Get(0).(bool)
+	}
+
+	var r1 error
+	if rf, ok := ret.Get(1).(func(int64) error); ok {
+		r1 = rf(_a0)
+	} else {
+		r1 = ret.Error(1)
+	}
+
+	return r0, r1
+}
+
+// IndexBlockEvents provides a mock function with given fields: _a0
+func (_m *EventSink) IndexBlockEvents(_a0 tenderminttypes.EventDataNewBlockHeader) error {
+	ret := _m.Called(_a0)
+
+	var r0 error
+	if rf, ok := ret.Get(0).(func(tenderminttypes.EventDataNewBlockHeader) error); ok {
+		r0 = rf(_a0)
+	} else {
+		r0 = ret.Error(0)
+	}
+
+	return r0
+}
+
+// IndexTxEvents provides a mock function with given fields: _a0
+func (_m *EventSink) IndexTxEvents(_a0 []*types.TxResult) error {
+	ret := _m.Called(_a0)
+
+	var r0 error
+	if rf, ok := ret.Get(0).(func([]*types.TxResult) error); ok {
+		r0 = rf(_a0)
+	} else {
+		r0 = ret.Error(0)
+	}
+
+	return r0
+}
+
+// SearchBlockEvents provides a mock function with given fields: _a0, _a1
+func (_m *EventSink) SearchBlockEvents(_a0 context.Context, _a1 *query.Query) ([]int64, error) {
+	ret := _m.Called(_a0, _a1)
+
+	var r0 []int64
+	if rf, ok := ret.Get(0).(func(context.Context, *query.Query) []int64); ok {
+		r0 = rf(_a0, _a1)
+	} else {
+		if ret.Get(0) != nil {
+			r0 = ret.Get(0).([]int64)
+		}
+	}
+
+	var r1 error
+	if rf, ok := ret.Get(1).(func(context.Context, *query.Query) error); ok {
+		r1 = rf(_a0, _a1)
+	} else {
+		r1 = ret.Error(1)
+	}
+
+	return r0, r1
+}
+
+// SearchTxEvents provides a mock function with given fields: _a0, _a1
+func (_m *EventSink) SearchTxEvents(_a0 context.Context, _a1 *query.Query) ([]*types.TxResult, error) {
+	ret := _m.Called(_a0, _a1)
+
+	var r0 []*types.TxResult
+	if rf, ok := ret.Get(0).(func(context.Context, *query.Query) []*types.TxResult); ok {
+		r0 = rf(_a0, _a1)
+	} else {
+		if ret.Get(0) != nil {
+			r0 = ret.Get(0).([]*types.TxResult)
+		}
+	}
+
+	var r1 error
+	if rf, ok := ret.Get(1).(func(context.Context, *query.Query) error); ok {
+		r1 = rf(_a0, _a1)
+	} else {
+		r1 = ret.Error(1)
+	}
+
+	return r0, r1
+}
+
+// Stop provides a mock function with given fields:
+func (_m *EventSink) Stop() error {
+	ret := _m.Called()
+
+	var r0 error
+	if rf, ok := ret.Get(0).(func() error); ok {
+		r0 = rf()
+	} else {
+		r0 = ret.Error(0)
+	}
+
+	return r0
+}
+
+// Type provides a mock function with given fields:
+func (_m *EventSink) Type() indexer.EventSinkType {
+	ret := _m.Called()
+
+	var r0 indexer.EventSinkType
+	if rf, ok := ret.Get(0).(func() indexer.EventSinkType); ok {
+		r0 = rf()
+	} else {
+		r0 = ret.Get(0).(indexer.EventSinkType)
+	}
+
+	return r0
+}

state/mocks/block_store.go

@@ -1,4 +1,4 @@
-// Code generated by mockery 2.7.5. DO NOT EDIT.
+// Code generated by mockery 2.9.0. DO NOT EDIT.
 
 package mocks
 
@@ -137,13 +137,13 @@ func (_m *BlockStore) LoadBlockPart(height int64, index int) *types.Part {
 	return r0
 }
 
-// LoadSeenCommit provides a mock function with given fields: height
-func (_m *BlockStore) LoadSeenCommit(height int64) *types.Commit {
-	ret := _m.Called(height)
+// LoadSeenCommit provides a mock function with given fields:
+func (_m *BlockStore) LoadSeenCommit() *types.Commit {
+	ret := _m.Called()
 
 	var r0 *types.Commit
-	if rf, ok := ret.Get(0).(func(int64) *types.Commit); ok {
-		r0 = rf(height)
+	if rf, ok := ret.Get(0).(func() *types.Commit); ok {
+		r0 = rf()
 	} else {
 		if ret.Get(0) != nil {
 			r0 = ret.Get(0).(*types.Commit)
@@ -191,4 +191,4 @@ func (_m *BlockStore) Size() int64 {
 	}
 
 	return r0
-}
+}

state/mocks/evidence_pool.go

@@ -1,4 +1,4 @@
-// Code generated by mockery v0.0.0-dev. DO NOT EDIT.
+// Code generated by mockery 2.9.0. DO NOT EDIT.
 
 package mocks
 

state/mocks/store.go

@@ -1,4 +1,4 @@
-// Code generated by mockery v0.0.0-dev. DO NOT EDIT.
+// Code generated by mockery 2.9.0. DO NOT EDIT.
 
 package mocks
 

state/services.go

@@ -32,7 +32,7 @@ type BlockStore interface {
 	LoadBlockPart(height int64, index int) *types.Part
 
 	LoadBlockCommit(height int64) *types.Commit
-	LoadSeenCommit(height int64) *types.Commit
+	LoadSeenCommit() *types.Commit
 }
 
 //-----------------------------------------------------------------------------

store/store.go

@@ -258,12 +258,13 @@ func (bs *BlockStore) LoadBlockCommit(height int64) *types.Commit {
 	return commit
 }
 
-// LoadSeenCommit returns the locally seen Commit for the given height.
-// This is useful when we've seen a commit, but there has not yet been
-// a new block at `height + 1` that includes this commit in its block.LastCommit.
-func (bs *BlockStore) LoadSeenCommit(height int64) *types.Commit {
+// LoadSeenCommit returns the last locally seen Commit before being
+// cannonicalized. This is useful when we've seen a commit, but there
+// has not yet been a new block at `height + 1` that includes this
+// commit in its block.LastCommit.
+func (bs *BlockStore) LoadSeenCommit() *types.Commit {
 	var pbc = new(tmproto.Commit)
-	bz, err := bs.db.Get(seenCommitKey(height))
+	bz, err := bs.db.Get(seenCommitKey())
 	if err != nil {
 		panic(err)
 	}
@@ -329,10 +330,6 @@ func (bs *BlockStore) PruneBlocks(height int64) (uint64, error) {
 		return pruned, err
 	}
 
-	if _, err := bs.pruneRange(seenCommitKey(0), seenCommitKey(height), nil); err != nil {
-		return pruned, err
-	}
-
 	return pruned, nil
 }
 
@@ -479,13 +476,7 @@ func (bs *BlockStore) SaveBlock(block *types.Block, blockParts *types.PartSet, s
 	// Save seen commit (seen +2/3 precommits for block)
 	pbsc := seenCommit.ToProto()
 	seenCommitBytes := mustEncode(pbsc)
-	if err := batch.Set(seenCommitKey(height), seenCommitBytes); err != nil {
-		panic(err)
-	}
-
-	// remove the previous seen commit that we have just replaced with the
-	// canonical commit
-	if err := batch.Delete(seenCommitKey(height - 1)); err != nil {
+	if err := batch.Set(seenCommitKey(), seenCommitBytes); err != nil {
 		panic(err)
 	}
 
@@ -516,7 +507,7 @@ func (bs *BlockStore) SaveSeenCommit(height int64, seenCommit *types.Commit) err
 	if err != nil {
 		return fmt.Errorf("unable to marshal commit: %w", err)
 	}
-	return bs.db.Set(seenCommitKey(height), seenCommitBytes)
+	return bs.db.Set(seenCommitKey(), seenCommitBytes)
 }
 
 func (bs *BlockStore) SaveSignedHeader(sh *types.SignedHeader, blockID types.BlockID) error {
@@ -612,8 +603,8 @@ func blockCommitKey(height int64) []byte {
 	return key
 }
 
-func seenCommitKey(height int64) []byte {
-	key, err := orderedcode.Append(nil, prefixSeenCommit, height)
+func seenCommitKey() []byte {
+	key, err := orderedcode.Append(nil, prefixSeenCommit)
 	if err != nil {
 		panic(err)
 	}

store/store_test.go

@@ -234,14 +234,14 @@ func TestBlockStoreSaveLoadBlock(t *testing.T) {
 			bBlockMeta := bs.LoadBlockMeta(tuple.block.Height)
 
 			if tuple.eraseSeenCommitInDB {
-				err := db.Delete(seenCommitKey(tuple.block.Height))
+				err := db.Delete(seenCommitKey())
 				require.NoError(t, err)
 			}
 			if tuple.corruptSeenCommitInDB {
-				err := db.Set(seenCommitKey(tuple.block.Height), []byte("bogus-seen-commit"))
+				err := db.Set(seenCommitKey(), []byte("bogus-seen-commit"))
 				require.NoError(t, err)
 			}
-			bSeenCommit := bs.LoadSeenCommit(tuple.block.Height)
+			bSeenCommit := bs.LoadSeenCommit()
 
 			commitHeight := tuple.block.Height - 1
 			if tuple.eraseCommitInDB {
@@ -494,9 +494,8 @@ func TestBlockFetchAtHeight(t *testing.T) {
 
 func TestSeenAndCanonicalCommit(t *testing.T) {
 	bs, _ := freshBlockStore()
-	height := int64(2)
 	loadCommit := func() (interface{}, error) {
-		meta := bs.LoadSeenCommit(height)
+		meta := bs.LoadSeenCommit()
 		return meta, nil
 	}
 
@@ -509,20 +508,15 @@ func TestSeenAndCanonicalCommit(t *testing.T) {
 	// produce a few blocks and check that the correct seen and cannoncial commits
 	// are persisted.
 	for h := int64(3); h <= 5; h++ {
-		c1 := bs.LoadSeenCommit(h)
-		require.Nil(t, c1)
-		c2 := bs.LoadBlockCommit(h - 1)
-		require.Nil(t, c2)
 		blockCommit := makeTestCommit(h-1, tmtime.Now())
 		block := factory.MakeBlock(state, h, blockCommit)
 		partSet := block.MakePartSet(2)
 		seenCommit := makeTestCommit(h, tmtime.Now())
 		bs.SaveBlock(block, partSet, seenCommit)
-		c3 := bs.LoadSeenCommit(h)
+		c3 := bs.LoadSeenCommit()
+		require.NotNil(t, c3)
+		require.Equal(t, h, c3.Height)
 		require.Equal(t, seenCommit.Hash(), c3.Hash())
-		// the previous seen commit should be removed
-		c4 := bs.LoadSeenCommit(h - 1)
-		require.Nil(t, c4)
 		c5 := bs.LoadBlockCommit(h)
 		require.Nil(t, c5)
 		c6 := bs.LoadBlockCommit(h - 1)