Merge branch 'cal/vote-extensions-1' into cal/vote-extensions-2

state/execution.go

@@ -26,6 +26,9 @@ type BlockExecutor struct {
 	// save state, validators, consensus params, abci responses here
 	store Store
 
+	// use blockstore for the pruning functions.
+	blockStore BlockStore
+
 	// execute the app against this
 	proxyApp proxy.AppConnConsensus
 
@@ -58,16 +61,18 @@ func NewBlockExecutor(
 	proxyApp proxy.AppConnConsensus,
 	mempool mempool.Mempool,
 	evpool EvidencePool,
+	blockStore BlockStore,
 	options ...BlockExecutorOption,
 ) *BlockExecutor {
 	res := &BlockExecutor{
-		store:    stateStore,
-		proxyApp: proxyApp,
-		eventBus: types.NopEventBus{},
-		mempool:  mempool,
-		evpool:   evpool,
-		logger:   logger,
-		metrics:  NopMetrics(),
+		store:      stateStore,
+		proxyApp:   proxyApp,
+		eventBus:   types.NopEventBus{},
+		mempool:    mempool,
+		evpool:     evpool,
+		logger:     logger,
+		metrics:    NopMetrics(),
+		blockStore: blockStore,
 	}
 
 	for _, option := range options {
@@ -183,16 +188,16 @@ func (blockExec *BlockExecutor) ValidateBlock(state State, block *types.Block) e
 
 // ApplyBlock validates the block against the state, executes it against the app,
 // fires the relevant events, commits the app, and saves the new state and responses.
-// It returns the new state and the block height to retain (pruning older blocks).
+// It returns the new state.
 // It's the only function that needs to be called
 // from outside this package to process and commit an entire block.
 // It takes a blockID to avoid recomputing the parts hash.
 func (blockExec *BlockExecutor) ApplyBlock(
 	state State, blockID types.BlockID, block *types.Block,
-) (State, int64, error) {
+) (State, error) {
 
 	if err := validateBlock(state, block); err != nil {
-		return state, 0, ErrInvalidBlock(err)
+		return state, ErrInvalidBlock(err)
 	}
 
 	startTime := time.Now().UnixNano()
@@ -202,14 +207,14 @@ func (blockExec *BlockExecutor) ApplyBlock(
 	endTime := time.Now().UnixNano()
 	blockExec.metrics.BlockProcessingTime.Observe(float64(endTime-startTime) / 1000000)
 	if err != nil {
-		return state, 0, ErrProxyAppConn(err)
+		return state, ErrProxyAppConn(err)
 	}
 
 	fail.Fail() // XXX
 
 	// Save the results before we commit.
 	if err := blockExec.store.SaveFinalizeBlockResponse(block.Height, abciResponse); err != nil {
-		return state, 0, err
+		return state, err
 	}
 
 	fail.Fail() // XXX
@@ -217,12 +222,12 @@ func (blockExec *BlockExecutor) ApplyBlock(
 	// validate the validator updates and convert to tendermint types
 	err = validateValidatorUpdates(abciResponse.ValidatorUpdates, state.ConsensusParams.Validator)
 	if err != nil {
-		return state, 0, fmt.Errorf("error in validator updates: %w", err)
+		return state, fmt.Errorf("error in validator updates: %v", err)
 	}
 
 	validatorUpdates, err := types.PB2TM.ValidatorUpdates(abciResponse.ValidatorUpdates)
 	if err != nil {
-		return state, 0, err
+		return state, err
 	}
 	if len(validatorUpdates) > 0 {
 		blockExec.logger.Debug("updates to validators", "updates", types.ValidatorListString(validatorUpdates))
@@ -235,13 +240,13 @@ func (blockExec *BlockExecutor) ApplyBlock(
 	// Update the state with the block and responses.
 	state, err = updateState(state, blockID, &block.Header, abciResponse, validatorUpdates)
 	if err != nil {
-		return state, 0, fmt.Errorf("failed to update state: %w", err)
+		return state, fmt.Errorf("commit failed for application: %v", err)
 	}
 
 	// Lock mempool, commit app state, update mempoool.
 	retainHeight, err := blockExec.Commit(state, block, abciResponse)
 	if err != nil {
-		return state, 0, fmt.Errorf("commit failed for application: %w", err)
+		return state, fmt.Errorf("commit failed for application: %v", err)
 	}
 
 	// Update evpool with the latest state.
@@ -252,16 +257,26 @@ func (blockExec *BlockExecutor) ApplyBlock(
 	// Update the app hash and save the state.
 	state.AppHash = abciResponse.AgreedAppData
 	if err := blockExec.store.Save(state); err != nil {
-		return state, 0, err
+		return state, err
 	}
 
 	fail.Fail() // XXX
 
+	// Prune old heights, if requested by ABCI app.
+	if retainHeight > 0 {
+		pruned, err := blockExec.pruneBlocks(retainHeight, state)
+		if err != nil {
+			blockExec.logger.Error("failed to prune blocks", "retain_height", retainHeight, "err", err)
+		} else {
+			blockExec.logger.Debug("pruned blocks", "pruned", pruned, "retain_height", retainHeight)
+		}
+	}
+
 	// Events are fired after everything else.
 	// NOTE: if we crash between Commit and Save, events wont be fired during replay
 	fireEvents(blockExec.logger, blockExec.eventBus, block, blockID, abciResponse, validatorUpdates)
 
-	return state, retainHeight, nil
+	return state, nil
 }
 
 // Commit locks the mempool, runs the ABCI Commit message, and updates the
@@ -671,3 +686,21 @@ func ExecCommitBlock(
 
 	return resp.AgreedAppData, nil
 }
+
+func (blockExec *BlockExecutor) pruneBlocks(retainHeight int64, state State) (uint64, error) {
+	base := blockExec.blockStore.Base()
+	if retainHeight <= base {
+		return 0, nil
+	}
+
+	amountPruned, prunedHeaderHeight, err := blockExec.blockStore.PruneBlocks(retainHeight, state)
+	if err != nil {
+		return 0, fmt.Errorf("failed to prune block store: %w", err)
+	}
+
+	err = blockExec.Store().PruneStates(base, retainHeight, prunedHeaderHeight)
+	if err != nil {
+		return 0, fmt.Errorf("failed to prune state store: %w", err)
+	}
+	return amountPruned, nil
+}

state/execution_test.go

@@ -25,9 +25,11 @@ import (
 	pmocks "github.com/tendermint/tendermint/proxy/mocks"
 	sm "github.com/tendermint/tendermint/state"
 	"github.com/tendermint/tendermint/state/mocks"
+	"github.com/tendermint/tendermint/store"
 	"github.com/tendermint/tendermint/types"
 	tmtime "github.com/tendermint/tendermint/types/time"
 	"github.com/tendermint/tendermint/version"
+	dbm "github.com/tendermint/tm-db"
 )
 
 var (
@@ -47,6 +49,8 @@ func TestApplyBlock(t *testing.T) {
 	stateStore := sm.NewStore(stateDB, sm.StoreOptions{
 		DiscardFinalizeBlockResponses: false,
 	})
+	blockStore := store.NewBlockStore(dbm.NewMemDB())
+
 	mp := &mpmocks.Mempool{}
 	mp.On("Lock").Return()
 	mp.On("Unlock").Return()
@@ -59,16 +63,15 @@ func TestApplyBlock(t *testing.T) {
 		mock.Anything,
 		mock.Anything).Return(nil)
 	blockExec := sm.NewBlockExecutor(stateStore, log.TestingLogger(), proxyApp.Consensus(),
-		mp, sm.EmptyEvidencePool{})
+		mp, sm.EmptyEvidencePool{}, blockStore)
 
 	block := makeBlock(state, 1, new(types.Commit))
 	bps, err := block.MakePartSet(testPartSize)
 	require.NoError(t, err)
 	blockID := types.BlockID{Hash: block.Hash(), PartSetHeader: bps.Header()}
 
-	state, retainHeight, err := blockExec.ApplyBlock(state, blockID, block)
+	state, err = blockExec.ApplyBlock(state, blockID, block)
 	require.Nil(t, err)
-	assert.EqualValues(t, retainHeight, 1)
 
 	// TODO check state and mempool
 	assert.EqualValues(t, 1, state.Version.Consensus.App, "App version wasn't updated")
@@ -80,7 +83,7 @@ func TestFinalizeBlockValidators(t *testing.T) {
 	cc := proxy.NewLocalClientCreator(app)
 	proxyApp := proxy.NewAppConns(cc, proxy.NopMetrics())
 	err := proxyApp.Start()
-	require.Nil(t, err)
+	require.NoError(t, err)
 	defer proxyApp.Stop() //nolint:errcheck // no need to check error again
 
 	state, stateDB, _ := makeState(2, 2)
@@ -139,13 +142,13 @@ func TestFinalizeBlockValidators(t *testing.T) {
 	}
 }
 
-// TestFinalizeBlockByzantineValidators ensures we send byzantine validators list.
-func TestFinalizeBlockByzantineValidators(t *testing.T) {
+// TestFinalizeBlockMisbehavior ensures we send misbehavior list.
+func TestFinalizeBlockMisbehavior(t *testing.T) {
 	app := &testApp{}
 	cc := proxy.NewLocalClientCreator(app)
 	proxyApp := proxy.NewAppConns(cc, proxy.NopMetrics())
 	err := proxyApp.Start()
-	require.Nil(t, err)
+	require.NoError(t, err)
 	defer proxyApp.Stop() //nolint:errcheck // ignore for tests
 
 	state, stateDB, privVals := makeState(1, 1)
@@ -231,8 +234,10 @@ func TestFinalizeBlockByzantineValidators(t *testing.T) {
 		mock.Anything,
 		mock.Anything).Return(nil)
 
+	blockStore := store.NewBlockStore(dbm.NewMemDB())
+
 	blockExec := sm.NewBlockExecutor(stateStore, log.TestingLogger(), proxyApp.Consensus(),
-		mp, evpool)
+		mp, evpool, blockStore)
 
 	block := makeBlock(state, 1, new(types.Commit))
 	block.Evidence = types.EvidenceData{Evidence: ev}
@@ -242,9 +247,8 @@ func TestFinalizeBlockByzantineValidators(t *testing.T) {
 
 	blockID = types.BlockID{Hash: block.Hash(), PartSetHeader: bps.Header()}
 
-	state, retainHeight, err := blockExec.ApplyBlock(state, blockID, block)
-	require.Nil(t, err)
-	assert.EqualValues(t, retainHeight, 1)
+	_, err = blockExec.ApplyBlock(state, blockID, block)
+	require.NoError(t, err)
 
 	// TODO check state and mempool
 	assert.Equal(t, abciMb, app.Misbehavior)
@@ -268,7 +272,7 @@ func TestProcessProposal(t *testing.T) {
 	stateStore := sm.NewStore(stateDB, sm.StoreOptions{
 		DiscardFinalizeBlockResponses: false,
 	})
-
+	blockStore := store.NewBlockStore(dbm.NewMemDB())
 	eventBus := types.NewEventBus()
 	err = eventBus.Start()
 	require.NoError(t, err)
@@ -279,6 +283,7 @@ func TestProcessProposal(t *testing.T) {
 		proxyApp.Consensus(),
 		new(mpmocks.Mempool),
 		sm.EmptyEvidencePool{},
+		blockStore,
 	)
 
 	block0 := makeBlock(state, height-1, new(types.Commit))
@@ -462,13 +467,13 @@ func TestUpdateValidators(t *testing.T) {
 	}
 }
 
-// TestEndBlockValidatorUpdates ensures we update validator set and send an event.
-func TestEndBlockValidatorUpdates(t *testing.T) {
+// TestFinalizeBlockValidatorUpdates ensures we update validator set and send an event.
+func TestFinalizeBlockValidatorUpdates(t *testing.T) {
 	app := &testApp{}
 	cc := proxy.NewLocalClientCreator(app)
 	proxyApp := proxy.NewAppConns(cc, proxy.NopMetrics())
 	err := proxyApp.Start()
-	require.Nil(t, err)
+	require.NoError(t, err)
 	defer proxyApp.Stop() //nolint:errcheck // ignore for tests
 
 	state, stateDB, _ := makeState(1, 1)
@@ -488,12 +493,14 @@ func TestEndBlockValidatorUpdates(t *testing.T) {
 		mock.Anything).Return(nil)
 	mp.On("ReapMaxBytesMaxGas", mock.Anything, mock.Anything).Return(types.Txs{})
 
+	blockStore := store.NewBlockStore(dbm.NewMemDB())
 	blockExec := sm.NewBlockExecutor(
 		stateStore,
 		log.TestingLogger(),
 		proxyApp.Consensus(),
 		mp,
 		sm.EmptyEvidencePool{},
+		blockStore,
 	)
 
 	eventBus := types.NewEventBus()
@@ -522,8 +529,8 @@ func TestEndBlockValidatorUpdates(t *testing.T) {
 		{PubKey: pk, Power: 10},
 	}
 
-	state, _, err = blockExec.ApplyBlock(state, blockID, block)
-	require.Nil(t, err)
+	state, err = blockExec.ApplyBlock(state, blockID, block)
+	require.NoError(t, err)
 	// test new validator was added to NextValidators
 	if assert.Equal(t, state.Validators.Size()+1, state.NextValidators.Size()) {
 		idx, _ := state.NextValidators.GetByAddress(pubkey.Address())
@@ -548,26 +555,28 @@ func TestEndBlockValidatorUpdates(t *testing.T) {
 	}
 }
 
-// TestEndBlockValidatorUpdatesResultingInEmptySet checks that processing validator updates that
+// TestFinalizeBlockValidatorUpdatesResultingInEmptySet checks that processing validator updates that
 // would result in empty set causes no panic, an error is raised and NextValidators is not updated
-func TestEndBlockValidatorUpdatesResultingInEmptySet(t *testing.T) {
+func TestFinalizeBlockValidatorUpdatesResultingInEmptySet(t *testing.T) {
 	app := &testApp{}
 	cc := proxy.NewLocalClientCreator(app)
 	proxyApp := proxy.NewAppConns(cc, proxy.NopMetrics())
 	err := proxyApp.Start()
-	require.Nil(t, err)
+	require.NoError(t, err)
 	defer proxyApp.Stop() //nolint:errcheck // ignore for tests
 
 	state, stateDB, _ := makeState(1, 1)
 	stateStore := sm.NewStore(stateDB, sm.StoreOptions{
 		DiscardFinalizeBlockResponses: false,
 	})
+	blockStore := store.NewBlockStore(dbm.NewMemDB())
 	blockExec := sm.NewBlockExecutor(
 		stateStore,
 		log.TestingLogger(),
 		proxyApp.Consensus(),
 		new(mpmocks.Mempool),
 		sm.EmptyEvidencePool{},
+		blockStore,
 	)
 
 	block := makeBlock(state, 1, new(types.Commit))
@@ -582,8 +591,8 @@ func TestEndBlockValidatorUpdatesResultingInEmptySet(t *testing.T) {
 		{PubKey: vp, Power: 0},
 	}
 
-	assert.NotPanics(t, func() { state, _, err = blockExec.ApplyBlock(state, blockID, block) })
-	assert.NotNil(t, err)
+	assert.NotPanics(t, func() { state, err = blockExec.ApplyBlock(state, blockID, block) })
+	assert.Error(t, err)
 	assert.NotEmpty(t, state.NextValidators.Validators)
 }
 
@@ -614,12 +623,14 @@ func TestEmptyPrepareProposal(t *testing.T) {
 		mock.Anything).Return(nil)
 	mp.On("ReapMaxBytesMaxGas", mock.Anything, mock.Anything).Return(types.Txs{})
 
+	blockStore := store.NewBlockStore(dbm.NewMemDB())
 	blockExec := sm.NewBlockExecutor(
 		stateStore,
 		log.TestingLogger(),
 		proxyApp.Consensus(),
 		mp,
 		sm.EmptyEvidencePool{},
+		blockStore,
 	)
 	pa, _ := state.Validators.GetByIndex(0)
 	commit, _ := makeValidExtendedCommit(t, height, types.BlockID{}, state.Validators, privVals)
@@ -654,12 +665,14 @@ func TestPrepareProposalTxsAllIncluded(t *testing.T) {
 	require.NoError(t, err)
 	defer proxyApp.Stop() //nolint:errcheck // ignore for tests
 
+	blockStore := store.NewBlockStore(dbm.NewMemDB())
 	blockExec := sm.NewBlockExecutor(
 		stateStore,
 		log.TestingLogger(),
 		proxyApp.Consensus(),
 		mp,
 		evpool,
+		blockStore,
 	)
 	pa, _ := state.Validators.GetByIndex(0)
 	commit, _ := makeValidExtendedCommit(t, height, types.BlockID{}, state.Validators, privVals)
@@ -704,12 +717,14 @@ func TestPrepareProposalReorderTxs(t *testing.T) {
 	require.NoError(t, err)
 	defer proxyApp.Stop() //nolint:errcheck // ignore for tests
 
+	blockStore := store.NewBlockStore(dbm.NewMemDB())
 	blockExec := sm.NewBlockExecutor(
 		stateStore,
 		log.TestingLogger(),
 		proxyApp.Consensus(),
 		mp,
 		evpool,
+		blockStore,
 	)
 	pa, _ := state.Validators.GetByIndex(0)
 	commit, _ := makeValidExtendedCommit(t, height, types.BlockID{}, state.Validators, privVals)
@@ -756,12 +771,14 @@ func TestPrepareProposalErrorOnTooManyTxs(t *testing.T) {
 	require.NoError(t, err)
 	defer proxyApp.Stop() //nolint:errcheck // ignore for tests
 
+	blockStore := store.NewBlockStore(dbm.NewMemDB())
 	blockExec := sm.NewBlockExecutor(
 		stateStore,
 		log.NewNopLogger(),
 		proxyApp.Consensus(),
 		mp,
 		evpool,
+		blockStore,
 	)
 	pa, _ := state.Validators.GetByIndex(0)
 	commit, _ := makeValidExtendedCommit(t, height, types.BlockID{}, state.Validators, privVals)
@@ -803,12 +820,14 @@ func TestPrepareProposalErrorOnPrepareProposalError(t *testing.T) {
 	require.NoError(t, err)
 	defer proxyApp.Stop() //nolint:errcheck // ignore for tests
 
+	blockStore := store.NewBlockStore(dbm.NewMemDB())
 	blockExec := sm.NewBlockExecutor(
 		stateStore,
 		log.NewNopLogger(),
 		proxyApp.Consensus(),
 		mp,
 		evpool,
+		blockStore,
 	)
 	pa, _ := state.Validators.GetByIndex(0)
 	commit, _ := makeValidExtendedCommit(t, height, types.BlockID{}, state.Validators, privVals)

state/helpers_test.go

@@ -67,7 +67,7 @@ func makeAndApplyGoodBlock(state sm.State, height int64, lastCommit *types.Commi
 	}
 	blockID := types.BlockID{Hash: block.Hash(),
 		PartSetHeader: partSet.Header()}
-	state, _, err = blockExec.ApplyBlock(state, blockID, block)
+	state, err = blockExec.ApplyBlock(state, blockID, block)
 	if err != nil {
 		return state, types.BlockID{}, err
 	}

state/indexer/block/kv/kv.go

@@ -13,6 +13,7 @@ import (
 
 	abci "github.com/tendermint/tendermint/abci/types"
 	"github.com/tendermint/tendermint/libs/pubsub/query"
+	"github.com/tendermint/tendermint/libs/pubsub/query/syntax"
 	"github.com/tendermint/tendermint/state/indexer"
 	"github.com/tendermint/tendermint/types"
 )
@@ -86,10 +87,7 @@ func (idx *BlockerIndexer) Search(ctx context.Context, q *query.Query) ([]int64,
 	default:
 	}
 
-	conditions, err := q.Conditions()
-	if err != nil {
-		return nil, fmt.Errorf("failed to parse query conditions: %w", err)
-	}
+	conditions := q.Syntax()
 
 	// If there is an exact height query, return the result immediately
 	// (if it exists).
@@ -153,7 +151,7 @@ func (idx *BlockerIndexer) Search(ctx context.Context, q *query.Query) ([]int64,
 			continue
 		}
 
-		startKey, err := orderedcode.Append(nil, c.CompositeKey, fmt.Sprintf("%v", c.Operand))
+		startKey, err := orderedcode.Append(nil, c.Tag, c.Arg.Value())
 		if err != nil {
 			return nil, err
 		}
@@ -321,7 +319,7 @@ LOOP:
 // matched.
 func (idx *BlockerIndexer) match(
 	ctx context.Context,
-	c query.Condition,
+	c syntax.Condition,
 	startKeyBz []byte,
 	filteredHeights map[string][]byte,
 	firstRun bool,
@@ -336,7 +334,7 @@ func (idx *BlockerIndexer) match(
 	tmpHeights := make(map[string][]byte)
 
 	switch {
-	case c.Op == query.OpEqual:
+	case c.Op == syntax.TEq:
 		it, err := dbm.IteratePrefix(idx.store, startKeyBz)
 		if err != nil {
 			return nil, fmt.Errorf("failed to create prefix iterator: %w", err)
@@ -355,8 +353,8 @@ func (idx *BlockerIndexer) match(
 			return nil, err
 		}
 
-	case c.Op == query.OpExists:
-		prefix, err := orderedcode.Append(nil, c.CompositeKey)
+	case c.Op == syntax.TExists:
+		prefix, err := orderedcode.Append(nil, c.Tag)
 		if err != nil {
 			return nil, err
 		}
@@ -382,8 +380,8 @@ func (idx *BlockerIndexer) match(
 			return nil, err
 		}
 
-	case c.Op == query.OpContains:
-		prefix, err := orderedcode.Append(nil, c.CompositeKey)
+	case c.Op == syntax.TContains:
+		prefix, err := orderedcode.Append(nil, c.Tag)
 		if err != nil {
 			return nil, err
 		}
@@ -400,7 +398,7 @@ func (idx *BlockerIndexer) match(
 				continue
 			}
 
-			if strings.Contains(eventValue, c.Operand.(string)) {
+			if strings.Contains(eventValue, c.Arg.Value()) {
 				tmpHeights[string(it.Value())] = it.Value()
 			}
 

state/indexer/block/kv/kv_test.go

@@ -82,39 +82,39 @@ func TestBlockIndexer(t *testing.T) {
 		results []int64
 	}{
 		"block.height = 100": {
-			q:       query.MustParse("block.height = 100"),
+			q:       query.MustCompile(`block.height = 100`),
 			results: []int64{},
 		},
 		"block.height = 5": {
-			q:       query.MustParse("block.height = 5"),
+			q:       query.MustCompile(`block.height = 5`),
 			results: []int64{5},
 		},
 		"begin_event.key1 = 'value1'": {
-			q:       query.MustParse("begin_event.key1 = 'value1'"),
+			q:       query.MustCompile(`begin_event.key1 = 'value1'`),
 			results: []int64{},
 		},
 		"begin_event.proposer = 'FCAA001'": {
-			q:       query.MustParse("begin_event.proposer = 'FCAA001'"),
+			q:       query.MustCompile(`begin_event.proposer = 'FCAA001'`),
 			results: []int64{1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11},
 		},
 		"end_event.foo <= 5": {
-			q:       query.MustParse("end_event.foo <= 5"),
+			q:       query.MustCompile(`end_event.foo <= 5`),
 			results: []int64{2, 4},
 		},
 		"end_event.foo >= 100": {
-			q:       query.MustParse("end_event.foo >= 100"),
+			q:       query.MustCompile(`end_event.foo >= 100`),
 			results: []int64{1},
 		},
 		"block.height > 2 AND end_event.foo <= 8": {
-			q:       query.MustParse("block.height > 2 AND end_event.foo <= 8"),
+			q:       query.MustCompile(`block.height > 2 AND end_event.foo <= 8`),
 			results: []int64{4, 6, 8},
 		},
 		"begin_event.proposer CONTAINS 'FFFFFFF'": {
-			q:       query.MustParse("begin_event.proposer CONTAINS 'FFFFFFF'"),
+			q:       query.MustCompile(`begin_event.proposer CONTAINS 'FFFFFFF'`),
 			results: []int64{},
 		},
 		"begin_event.proposer CONTAINS 'FCAA001'": {
-			q:       query.MustParse("begin_event.proposer CONTAINS 'FCAA001'"),
+			q:       query.MustCompile(`begin_event.proposer CONTAINS 'FCAA001'`),
 			results: []int64{1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11},
 		},
 	}

state/indexer/block/kv/util.go

@@ -6,8 +6,7 @@ import (
 	"strconv"
 
 	"github.com/google/orderedcode"
-
-	"github.com/tendermint/tendermint/libs/pubsub/query"
+	"github.com/tendermint/tendermint/libs/pubsub/query/syntax"
 	"github.com/tendermint/tendermint/types"
 )
 
@@ -86,10 +85,10 @@ func parseValueFromEventKey(key []byte) (string, error) {
 	return eventValue, nil
 }
 
-func lookForHeight(conditions []query.Condition) (int64, bool) {
+func lookForHeight(conditions []syntax.Condition) (int64, bool) {
 	for _, c := range conditions {
-		if c.CompositeKey == types.BlockHeightKey && c.Op == query.OpEqual {
-			return c.Operand.(int64), true
+		if c.Tag == types.BlockHeightKey && c.Op == syntax.TEq {
+			return int64(c.Arg.Number()), true
 		}
 	}
 

state/indexer/query_range.go

@@ -3,7 +3,7 @@ package indexer
 import (
 	"time"
 
-	"github.com/tendermint/tendermint/libs/pubsub/query"
+	"github.com/tendermint/tendermint/libs/pubsub/query/syntax"
 )
 
 // QueryRanges defines a mapping between a composite event key and a QueryRange.
@@ -77,32 +77,32 @@ func (qr QueryRange) UpperBoundValue() interface{} {
 
 // LookForRanges returns a mapping of QueryRanges and the matching indexes in
 // the provided query conditions.
-func LookForRanges(conditions []query.Condition) (ranges QueryRanges, indexes []int) {
+func LookForRanges(conditions []syntax.Condition) (ranges QueryRanges, indexes []int) {
 	ranges = make(QueryRanges)
 	for i, c := range conditions {
 		if IsRangeOperation(c.Op) {
-			r, ok := ranges[c.CompositeKey]
+			r, ok := ranges[c.Tag]
 			if !ok {
-				r = QueryRange{Key: c.CompositeKey}
+				r = QueryRange{Key: c.Tag}
 			}
 
 			switch c.Op {
-			case query.OpGreater:
-				r.LowerBound = c.Operand
+			case syntax.TGt:
+				r.LowerBound = conditionArg(c)
 
-			case query.OpGreaterEqual:
+			case syntax.TGeq:
 				r.IncludeLowerBound = true
-				r.LowerBound = c.Operand
+				r.LowerBound = conditionArg(c)
 
-			case query.OpLess:
-				r.UpperBound = c.Operand
+			case syntax.TLt:
+				r.UpperBound = conditionArg(c)
 
-			case query.OpLessEqual:
+			case syntax.TLeq:
 				r.IncludeUpperBound = true
-				r.UpperBound = c.Operand
+				r.UpperBound = conditionArg(c)
 			}
 
-			ranges[c.CompositeKey] = r
+			ranges[c.Tag] = r
 			indexes = append(indexes, i)
 		}
 	}
@@ -112,12 +112,26 @@ func LookForRanges(conditions []query.Condition) (ranges QueryRanges, indexes []
 
 // IsRangeOperation returns a boolean signifying if a query Operator is a range
 // operation or not.
-func IsRangeOperation(op query.Operator) bool {
+func IsRangeOperation(op syntax.Token) bool {
 	switch op {
-	case query.OpGreater, query.OpGreaterEqual, query.OpLess, query.OpLessEqual:
+	case syntax.TGt, syntax.TGeq, syntax.TLt, syntax.TLeq:
 		return true
 
 	default:
 		return false
 	}
 }
+
+func conditionArg(c syntax.Condition) interface{} {
+	if c.Arg == nil {
+		return nil
+	}
+	switch c.Arg.Type {
+	case syntax.TNumber:
+		return int64(c.Arg.Number())
+	case syntax.TTime, syntax.TDate:
+		return c.Arg.Time()
+	default:
+		return c.Arg.Value() // string
+	}
+}

state/indexer/sink/psql/psql_test.go

@@ -19,6 +19,8 @@ import (
 	"github.com/stretchr/testify/require"
 
 	abci "github.com/tendermint/tendermint/abci/types"
+	tmlog "github.com/tendermint/tendermint/libs/log"
+	"github.com/tendermint/tendermint/state/txindex"
 	"github.com/tendermint/tendermint/types"
 
 	// Register the Postgres database driver.
@@ -196,6 +198,56 @@ func TestIndexing(t *testing.T) {
 		err = indexer.IndexTxEvents([]*abci.TxResult{txResult})
 		require.NoError(t, err)
 	})
+
+	t.Run("IndexerService", func(t *testing.T) {
+		indexer := &EventSink{store: testDB(), chainID: chainID}
+
+		// event bus
+		eventBus := types.NewEventBus()
+		err := eventBus.Start()
+		require.NoError(t, err)
+		t.Cleanup(func() {
+			if err := eventBus.Stop(); err != nil {
+				t.Error(err)
+			}
+		})
+
+		service := txindex.NewIndexerService(indexer.TxIndexer(), indexer.BlockIndexer(), eventBus, true)
+		service.SetLogger(tmlog.TestingLogger())
+		err = service.Start()
+		require.NoError(t, err)
+		t.Cleanup(func() {
+			if err := service.Stop(); err != nil {
+				t.Error(err)
+			}
+		})
+
+		// publish block with txs
+		err = eventBus.PublishEventNewBlockEvents(types.EventDataNewBlockEvents{
+			Height: 1,
+			NumTxs: 2,
+		})
+		require.NoError(t, err)
+		txResult1 := &abci.TxResult{
+			Height: 1,
+			Index:  uint32(0),
+			Tx:     types.Tx("foo"),
+			Result: abci.ExecTxResult{Code: 0},
+		}
+		err = eventBus.PublishEventTx(types.EventDataTx{TxResult: *txResult1})
+		require.NoError(t, err)
+		txResult2 := &abci.TxResult{
+			Height: 1,
+			Index:  uint32(1),
+			Tx:     types.Tx("bar"),
+			Result: abci.ExecTxResult{Code: 1},
+		}
+		err = eventBus.PublishEventTx(types.EventDataTx{TxResult: *txResult2})
+		require.NoError(t, err)
+
+		time.Sleep(100 * time.Millisecond)
+		require.True(t, service.IsRunning())
+	})
 }
 
 func TestStop(t *testing.T) {

state/mocks/block_store.go

@@ -4,6 +4,7 @@ package mocks
 
 import (
 	mock "github.com/stretchr/testify/mock"
+	state "github.com/tendermint/tendermint/state"
 
 	types "github.com/tendermint/tendermint/types"
 )
@@ -27,6 +28,20 @@ func (_m *BlockStore) Base() int64 {
 	return r0
 }
 
+// DeleteLatestBlock provides a mock function with given fields:
+func (_m *BlockStore) DeleteLatestBlock() 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
+}
+
 // Height provides a mock function with given fields:
 func (_m *BlockStore) Height() int64 {
 	ret := _m.Called()
@@ -185,25 +200,32 @@ func (_m *BlockStore) LoadSeenCommit(height int64) *types.Commit {
 	return r0
 }
 
-// PruneBlocks provides a mock function with given fields: height
-func (_m *BlockStore) PruneBlocks(height int64) (uint64, error) {
-	ret := _m.Called(height)
+// PruneBlocks provides a mock function with given fields: height, _a1
+func (_m *BlockStore) PruneBlocks(height int64, _a1 state.State) (uint64, int64, error) {
+	ret := _m.Called(height, _a1)
 
 	var r0 uint64
-	if rf, ok := ret.Get(0).(func(int64) uint64); ok {
-		r0 = rf(height)
+	if rf, ok := ret.Get(0).(func(int64, state.State) uint64); ok {
+		r0 = rf(height, _a1)
 	} else {
 		r0 = ret.Get(0).(uint64)
 	}
 
-	var r1 error
-	if rf, ok := ret.Get(1).(func(int64) error); ok {
-		r1 = rf(height)
+	var r1 int64
+	if rf, ok := ret.Get(1).(func(int64, state.State) int64); ok {
+		r1 = rf(height, _a1)
 	} else {
-		r1 = ret.Error(1)
+		r1 = ret.Get(1).(int64)
 	}
 
-	return r0, r1
+	var r2 error
+	if rf, ok := ret.Get(2).(func(int64, state.State) error); ok {
+		r2 = rf(height, _a1)
+	} else {
+		r2 = ret.Error(2)
+	}
+
+	return r0, r1, r2
 }
 
 // SaveBlock provides a mock function with given fields: block, blockParts, seenCommit

state/mocks/store.go

@@ -197,13 +197,13 @@ func (_m *Store) LoadValidators(_a0 int64) (*types.ValidatorSet, error) {
 	return r0, r1
 }
 
-// PruneStates provides a mock function with given fields: _a0, _a1
-func (_m *Store) PruneStates(_a0 int64, _a1 int64) error {
-	ret := _m.Called(_a0, _a1)
+// PruneStates provides a mock function with given fields: _a0, _a1, _a2
+func (_m *Store) PruneStates(_a0 int64, _a1 int64, _a2 int64) error {
+	ret := _m.Called(_a0, _a1, _a2)
 
 	var r0 error
-	if rf, ok := ret.Get(0).(func(int64, int64) error); ok {
-		r0 = rf(_a0, _a1)
+	if rf, ok := ret.Get(0).(func(int64, int64, int64) error); ok {
+		r0 = rf(_a0, _a1, _a2)
 	} else {
 		r0 = ret.Error(0)
 	}

state/rollback.go

@@ -12,7 +12,7 @@ import (
 // Rollback overwrites the current Tendermint state (height n) with the most
 // recent previous state (height n - 1).
 // Note that this function does not affect application state.
-func Rollback(bs BlockStore, ss Store) (int64, []byte, error) {
+func Rollback(bs BlockStore, ss Store, removeBlock bool) (int64, []byte, error) {
 	invalidState, err := ss.Load()
 	if err != nil {
 		return -1, nil, err
@@ -24,9 +24,14 @@ func Rollback(bs BlockStore, ss Store) (int64, []byte, error) {
 	height := bs.Height()
 
 	// NOTE: persistence of state and blocks don't happen atomically. Therefore it is possible that
-	// when the user stopped the node the state wasn't updated but the blockstore was. In this situation
-	// we don't need to rollback any state and can just return early
+	// when the user stopped the node the state wasn't updated but the blockstore was. Discard the
+	// pending block before continuing.
 	if height == invalidState.LastBlockHeight+1 {
+		if removeBlock {
+			if err := bs.DeleteLatestBlock(); err != nil {
+				return -1, nil, fmt.Errorf("failed to remove final block from blockstore: %w", err)
+			}
+		}
 		return invalidState.LastBlockHeight, invalidState.AppHash, nil
 	}
 
@@ -108,5 +113,13 @@ func Rollback(bs BlockStore, ss Store) (int64, []byte, error) {
 		return -1, nil, fmt.Errorf("failed to save rolled back state: %w", err)
 	}
 
+	// If removeBlock is true then also remove the block associated with the previous state.
+	// This will mean both the last state and last block height is equal to n - 1
+	if removeBlock {
+		if err := bs.DeleteLatestBlock(); err != nil {
+			return -1, nil, fmt.Errorf("failed to remove final block from blockstore: %w", err)
+		}
+	}
+
 	return rolledBackState.LastBlockHeight, rolledBackState.AppHash, nil
 }

state/rollback_test.go

@@ -3,6 +3,7 @@ package state_test
 import (
 	"crypto/rand"
 	"testing"
+	"time"
 
 	"github.com/stretchr/testify/require"
 	dbm "github.com/tendermint/tm-db"
@@ -13,6 +14,7 @@ import (
 	tmversion "github.com/tendermint/tendermint/proto/tendermint/version"
 	"github.com/tendermint/tendermint/state"
 	"github.com/tendermint/tendermint/state/mocks"
+	"github.com/tendermint/tendermint/store"
 	"github.com/tendermint/tendermint/types"
 	"github.com/tendermint/tendermint/version"
 )
@@ -50,6 +52,7 @@ func TestRollback(t *testing.T) {
 		BlockID: initialState.LastBlockID,
 		Header: types.Header{
 			Height:          initialState.LastBlockHeight,
+			Time:            initialState.LastBlockTime,
 			AppHash:         crypto.CRandBytes(tmhash.Size),
 			LastBlockID:     makeBlockIDRandom(),
 			LastResultsHash: initialState.LastResultsHash,
@@ -61,6 +64,7 @@ func TestRollback(t *testing.T) {
 			Height:          nextState.LastBlockHeight,
 			AppHash:         initialState.AppHash,
 			LastBlockID:     block.BlockID,
+			Time:            nextState.LastBlockTime,
 			LastResultsHash: nextState.LastResultsHash,
 		},
 	}
@@ -69,7 +73,7 @@ func TestRollback(t *testing.T) {
 	blockStore.On("Height").Return(nextHeight)
 
 	// rollback the state
-	rollbackHeight, rollbackHash, err := state.Rollback(blockStore, stateStore)
+	rollbackHeight, rollbackHash, err := state.Rollback(blockStore, stateStore, false)
 	require.NoError(t, err)
 	require.EqualValues(t, height, rollbackHeight)
 	require.EqualValues(t, initialState.AppHash, rollbackHash)
@@ -81,6 +85,122 @@ func TestRollback(t *testing.T) {
 	require.EqualValues(t, initialState, loadedState)
 }
 
+func TestRollbackHard(t *testing.T) {
+	const height int64 = 100
+	blockStore := store.NewBlockStore(dbm.NewMemDB())
+	stateStore := state.NewStore(dbm.NewMemDB(), state.StoreOptions{DiscardFinalizeBlockResponses: false})
+
+	valSet, _ := types.RandValidatorSet(5, 10)
+
+	params := types.DefaultConsensusParams()
+	params.Version.App = 10
+	now := time.Date(2020, 1, 1, 0, 0, 0, 0, time.UTC)
+
+	block := &types.Block{
+		Header: types.Header{
+			Version:            tmversion.Consensus{Block: version.BlockProtocol, App: 1},
+			ChainID:            "test-chain",
+			Time:               now,
+			Height:             height,
+			AppHash:            crypto.CRandBytes(tmhash.Size),
+			LastBlockID:        makeBlockIDRandom(),
+			LastCommitHash:     crypto.CRandBytes(tmhash.Size),
+			DataHash:           crypto.CRandBytes(tmhash.Size),
+			ValidatorsHash:     valSet.Hash(),
+			NextValidatorsHash: valSet.CopyIncrementProposerPriority(1).Hash(),
+			ConsensusHash:      params.Hash(),
+			LastResultsHash:    crypto.CRandBytes(tmhash.Size),
+			EvidenceHash:       crypto.CRandBytes(tmhash.Size),
+			ProposerAddress:    crypto.CRandBytes(crypto.AddressSize),
+		},
+		LastCommit: &types.Commit{Height: height - 1},
+	}
+
+	partSet, err := block.MakePartSet(types.BlockPartSizeBytes)
+	require.NoError(t, err)
+	blockStore.SaveBlock(block, partSet, &types.Commit{Height: block.Height})
+
+	currState := state.State{
+		Version: tmstate.Version{
+			Consensus: block.Header.Version,
+			Software:  version.TMCoreSemVer,
+		},
+		LastBlockHeight:                  block.Height,
+		LastBlockTime:                    block.Time,
+		AppHash:                          crypto.CRandBytes(tmhash.Size),
+		LastValidators:                   valSet,
+		Validators:                       valSet.CopyIncrementProposerPriority(1),
+		NextValidators:                   valSet.CopyIncrementProposerPriority(2),
+		ConsensusParams:                  *params,
+		LastHeightConsensusParamsChanged: height + 1,
+		LastHeightValidatorsChanged:      height + 1,
+		LastResultsHash:                  crypto.CRandBytes(tmhash.Size),
+	}
+	require.NoError(t, stateStore.Bootstrap(currState))
+
+	nextBlock := &types.Block{
+		Header: types.Header{
+			Version:            tmversion.Consensus{Block: version.BlockProtocol, App: 1},
+			ChainID:            block.ChainID,
+			Time:               block.Time,
+			Height:             currState.LastBlockHeight + 1,
+			AppHash:            currState.AppHash,
+			LastBlockID:        types.BlockID{Hash: block.Hash(), PartSetHeader: partSet.Header()},
+			LastCommitHash:     crypto.CRandBytes(tmhash.Size),
+			DataHash:           crypto.CRandBytes(tmhash.Size),
+			ValidatorsHash:     valSet.CopyIncrementProposerPriority(1).Hash(),
+			NextValidatorsHash: valSet.CopyIncrementProposerPriority(2).Hash(),
+			ConsensusHash:      params.Hash(),
+			LastResultsHash:    currState.LastResultsHash,
+			EvidenceHash:       crypto.CRandBytes(tmhash.Size),
+			ProposerAddress:    crypto.CRandBytes(crypto.AddressSize),
+		},
+		LastCommit: &types.Commit{Height: currState.LastBlockHeight},
+	}
+
+	nextPartSet, err := nextBlock.MakePartSet(types.BlockPartSizeBytes)
+	require.NoError(t, err)
+	blockStore.SaveBlock(nextBlock, nextPartSet, &types.Commit{Height: nextBlock.Height})
+
+	rollbackHeight, rollbackHash, err := state.Rollback(blockStore, stateStore, true)
+	require.NoError(t, err)
+	require.Equal(t, rollbackHeight, currState.LastBlockHeight)
+	require.Equal(t, rollbackHash, currState.AppHash)
+
+	// state should not have been changed
+	loadedState, err := stateStore.Load()
+	require.NoError(t, err)
+	require.Equal(t, currState, loadedState)
+
+	// resave the same block
+	blockStore.SaveBlock(nextBlock, nextPartSet, &types.Commit{Height: nextBlock.Height})
+
+	params.Version.App = 11
+
+	nextState := state.State{
+		Version: tmstate.Version{
+			Consensus: block.Header.Version,
+			Software:  version.TMCoreSemVer,
+		},
+		LastBlockHeight:                  nextBlock.Height,
+		LastBlockTime:                    nextBlock.Time,
+		AppHash:                          crypto.CRandBytes(tmhash.Size),
+		LastValidators:                   valSet.CopyIncrementProposerPriority(1),
+		Validators:                       valSet.CopyIncrementProposerPriority(2),
+		NextValidators:                   valSet.CopyIncrementProposerPriority(3),
+		ConsensusParams:                  *params,
+		LastHeightConsensusParamsChanged: nextBlock.Height + 1,
+		LastHeightValidatorsChanged:      nextBlock.Height + 1,
+		LastResultsHash:                  crypto.CRandBytes(tmhash.Size),
+	}
+	require.NoError(t, stateStore.Save(nextState))
+
+	rollbackHeight, rollbackHash, err = state.Rollback(blockStore, stateStore, true)
+	require.NoError(t, err)
+	require.Equal(t, rollbackHeight, currState.LastBlockHeight)
+	require.Equal(t, rollbackHash, currState.AppHash)
+}
+
 func TestRollbackNoState(t *testing.T) {
 	stateStore := state.NewStore(dbm.NewMemDB(),
 		state.StoreOptions{
@@ -88,7 +208,7 @@ func TestRollbackNoState(t *testing.T) {
 		})
 	blockStore := &mocks.BlockStore{}
 
-	_, _, err := state.Rollback(blockStore, stateStore)
+	_, _, err := state.Rollback(blockStore, stateStore, false)
 	require.Error(t, err)
 	require.Contains(t, err.Error(), "no state found")
 }
@@ -101,7 +221,7 @@ func TestRollbackNoBlocks(t *testing.T) {
 	blockStore.On("LoadBlockMeta", height).Return(nil)
 	blockStore.On("LoadBlockMeta", height-1).Return(nil)
 
-	_, _, err := state.Rollback(blockStore, stateStore)
+	_, _, err := state.Rollback(blockStore, stateStore, false)
 	require.Error(t, err)
 	require.Contains(t, err.Error(), "block at height 99 not found")
 }
@@ -112,7 +232,7 @@ func TestRollbackDifferentStateHeight(t *testing.T) {
 	blockStore := &mocks.BlockStore{}
 	blockStore.On("Height").Return(height + 2)
 
-	_, _, err := state.Rollback(blockStore, stateStore)
+	_, _, err := state.Rollback(blockStore, stateStore, false)
 	require.Error(t, err)
 	require.Equal(t, err.Error(), "statestore height (100) is not one below or equal to blockstore height (102)")
 }
@@ -138,6 +258,7 @@ func setupStateStore(t *testing.T, height int64) state.Store {
 		AppHash:                          tmhash.Sum([]byte("app_hash")),
 		LastResultsHash:                  tmhash.Sum([]byte("last_results_hash")),
 		LastBlockHeight:                  height,
+		LastBlockTime:                    time.Now(),
 		LastValidators:                   valSet,
 		Validators:                       valSet.CopyIncrementProposerPriority(1),
 		NextValidators:                   valSet.CopyIncrementProposerPriority(2),

state/services.go

@@ -27,7 +27,7 @@ type BlockStore interface {
 	SaveBlock(block *types.Block, blockParts *types.PartSet, seenCommit *types.Commit)
 	SaveBlockWithExtendedCommit(block *types.Block, blockParts *types.PartSet, seenExtendedCommit *types.ExtendedCommit)
 
-	PruneBlocks(height int64) (uint64, error)
+	PruneBlocks(height int64, state State) (uint64, int64, error)
 
 	LoadBlockByHash(hash []byte) *types.Block
 	LoadBlockMetaByHash(hash []byte) *types.BlockMeta
@@ -36,6 +36,8 @@ type BlockStore interface {
 	LoadBlockCommit(height int64) *types.Commit
 	LoadSeenCommit(height int64) *types.Commit
 	LoadBlockExtendedCommit(height int64) *types.ExtendedCommit
+
+	DeleteLatestBlock() error
 }
 
 //-----------------------------------------------------------------------------

state/state.go

@@ -317,7 +317,7 @@ func MakeGenesisDocFromFile(genDocFile string) (*types.GenesisDoc, error) {
 func MakeGenesisState(genDoc *types.GenesisDoc) (State, error) {
 	err := genDoc.ValidateAndComplete()
 	if err != nil {
-		return State{}, fmt.Errorf("error in genesis file: %v", err)
+		return State{}, fmt.Errorf("error in genesis doc: %w", err)
 	}
 
 	var validatorSet, nextValidatorSet *types.ValidatorSet

state/state_test.go

@@ -14,9 +14,9 @@ import (
 	dbm "github.com/tendermint/tm-db"
 
 	abci "github.com/tendermint/tendermint/abci/types"
-	cfg "github.com/tendermint/tendermint/config"
 	"github.com/tendermint/tendermint/crypto/ed25519"
 	cryptoenc "github.com/tendermint/tendermint/crypto/encoding"
+	"github.com/tendermint/tendermint/internal/test"
 	tmrand "github.com/tendermint/tendermint/libs/rand"
 	sm "github.com/tendermint/tendermint/state"
 	"github.com/tendermint/tendermint/types"
@@ -24,7 +24,7 @@ import (
 
 // setupTestCase does setup common to all test cases.
 func setupTestCase(t *testing.T) (func(t *testing.T), dbm.DB, sm.State) {
-	config := cfg.ResetTestRoot("state_")
+	config := test.ResetTestRoot("state_")
 	dbType := dbm.BackendType(config.DBBackend)
 	stateDB, err := dbm.NewDB("state", dbType, config.DBDir())
 	stateStore := sm.NewStore(stateDB, sm.StoreOptions{

state/store.go

@@ -70,8 +70,8 @@ type Store interface {
 	SaveFinalizeBlockResponse(int64, *abci.ResponseFinalizeBlock) error
 	// Bootstrap is used for bootstrapping state when not starting from a initial height.
 	Bootstrap(State) error
-	// PruneStates takes the height from which to start prning and which height stop at
-	PruneStates(int64, int64) error
+	// PruneStates takes the height from which to start pruning and which height stop at
+	PruneStates(int64, int64, int64) error
 	// Close closes the connection with the database
 	Close() error
 }
@@ -237,14 +237,15 @@ func (store dbStore) Bootstrap(state State) error {
 // encoding not preserving ordering: https://github.com/tendermint/tendermint/issues/4567
 // This will cause some old states to be left behind when doing incremental partial prunes,
 // specifically older checkpoints and LastHeightChanged targets.
-func (store dbStore) PruneStates(from int64, to int64) error {
+func (store dbStore) PruneStates(from int64, to int64, evidenceThresholdHeight int64) error {
 	if from <= 0 || to <= 0 {
 		return fmt.Errorf("from height %v and to height %v must be greater than 0", from, to)
 	}
 	if from >= to {
 		return fmt.Errorf("from height %v must be lower than to height %v", from, to)
 	}
-	valInfo, err := loadValidatorsInfo(store.db, to)
+
+	valInfo, err := loadValidatorsInfo(store.db, min(to, evidenceThresholdHeight))
 	if err != nil {
 		return fmt.Errorf("validators at height %v not found: %w", to, err)
 	}
@@ -298,12 +299,14 @@ func (store dbStore) PruneStates(from int64, to int64) error {
 					return err
 				}
 			}
-		} else {
+		} else if h < evidenceThresholdHeight {
 			err = batch.Delete(calcValidatorsKey(h))
 			if err != nil {
 				return err
 			}
 		}
+		// else we keep the validator set because we might need
+		// it later on for evidence verification
 
 		if keepParams[h] {
 			p, err := store.loadConsensusParamsInfo(h)
@@ -547,7 +550,7 @@ func loadValidatorsInfo(db dbm.DB, height int64) (*tmstate.ValidatorsInfo, error
 	}
 
 	if len(buf) == 0 {
-		return nil, errors.New("no last ABCI response has been persisted")
+		return nil, errors.New("value retrieved from db is empty")
 	}
 
 	v := new(tmstate.ValidatorsInfo)
@@ -635,7 +638,7 @@ func (store dbStore) loadConsensusParamsInfo(height int64) (*tmstate.ConsensusPa
 		return nil, err
 	}
 	if len(buf) == 0 {
-		return nil, errors.New("no last ABCI response has been persisted")
+		return nil, errors.New("value retrieved from db is empty")
 	}
 
 	paramsInfo := new(tmstate.ConsensusParamsInfo)
@@ -677,3 +680,10 @@ func (store dbStore) saveConsensusParamsInfo(nextHeight, changeHeight int64, par
 func (store dbStore) Close() error {
 	return store.db.Close()
 }
+
+func min(a int64, b int64) int64 {
+	if a < b {
+		return a
+	}
+	return b
+}

state/store_test.go

@@ -11,9 +11,9 @@ import (
 	dbm "github.com/tendermint/tm-db"
 
 	abci "github.com/tendermint/tendermint/abci/types"
-	cfg "github.com/tendermint/tendermint/config"
 	"github.com/tendermint/tendermint/crypto"
 	"github.com/tendermint/tendermint/crypto/ed25519"
+	"github.com/tendermint/tendermint/internal/test"
 	tmrand "github.com/tendermint/tendermint/libs/rand"
 	sm "github.com/tendermint/tendermint/state"
 	"github.com/tendermint/tendermint/types"
@@ -49,7 +49,7 @@ func TestStoreLoadValidators(t *testing.T) {
 func BenchmarkLoadValidators(b *testing.B) {
 	const valSetSize = 100
 
-	config := cfg.ResetTestRoot("state_")
+	config := test.ResetTestRoot("state_")
 	defer os.RemoveAll(config.RootDir)
 	dbType := dbm.BackendType(config.DBBackend)
 	stateDB, err := dbm.NewDB("state", dbType, config.DBDir())
@@ -87,23 +87,25 @@ func BenchmarkLoadValidators(b *testing.B) {
 
 func TestPruneStates(t *testing.T) {
 	testcases := map[string]struct {
-		makeHeights  int64
-		pruneFrom    int64
-		pruneTo      int64
-		expectErr    bool
-		expectVals   []int64
-		expectParams []int64
-		expectABCI   []int64
+		makeHeights             int64
+		pruneFrom               int64
+		pruneTo                 int64
+		evidenceThresholdHeight int64
+		expectErr               bool
+		expectVals              []int64
+		expectParams            []int64
+		expectABCI              []int64
 	}{
-		"error on pruning from 0":      {100, 0, 5, true, nil, nil, nil},
-		"error when from > to":         {100, 3, 2, true, nil, nil, nil},
-		"error when from == to":        {100, 3, 3, true, nil, nil, nil},
-		"error when to does not exist": {100, 1, 101, true, nil, nil, nil},
-		"prune all":                    {100, 1, 100, false, []int64{93, 100}, []int64{95, 100}, []int64{100}},
-		"prune some": {10, 2, 8, false, []int64{1, 3, 8, 9, 10},
+		"error on pruning from 0":      {100, 0, 5, 100, true, nil, nil, nil},
+		"error when from > to":         {100, 3, 2, 2, true, nil, nil, nil},
+		"error when from == to":        {100, 3, 3, 3, true, nil, nil, nil},
+		"error when to does not exist": {100, 1, 101, 101, true, nil, nil, nil},
+		"prune all":                    {100, 1, 100, 100, false, []int64{93, 100}, []int64{95, 100}, []int64{100}},
+		"prune some": {10, 2, 8, 8, false, []int64{1, 3, 8, 9, 10},
 			[]int64{1, 5, 8, 9, 10}, []int64{1, 8, 9, 10}},
-		"prune across checkpoint": {100001, 1, 100001, false, []int64{99993, 100000, 100001},
+		"prune across checkpoint": {100001, 1, 100001, 100001, false, []int64{99993, 100000, 100001},
 			[]int64{99995, 100001}, []int64{100001}},
+		"prune when evidence height < height": {20, 1, 18, 17, false, []int64{13, 17, 18, 19, 20}, []int64{15, 18, 19, 20}, []int64{18, 19, 20}},
 	}
 	for name, tc := range testcases {
 		tc := tc
@@ -162,7 +164,7 @@ func TestPruneStates(t *testing.T) {
 			}
 
 			// Test assertions
-			err := stateStore.PruneStates(tc.pruneFrom, tc.pruneTo)
+			err := stateStore.PruneStates(tc.pruneFrom, tc.pruneTo, tc.evidenceThresholdHeight)
 			if tc.expectErr {
 				require.Error(t, err)
 				return

state/txindex/indexer_service.go

@@ -3,7 +3,6 @@ package txindex
 import (
 	"context"
 
-	abci "github.com/tendermint/tendermint/abci/types"
 	"github.com/tendermint/tendermint/libs/service"
 	"github.com/tendermint/tendermint/state/indexer"
 	"github.com/tendermint/tendermint/types"
@@ -20,9 +19,10 @@ const (
 type IndexerService struct {
 	service.BaseService
 
-	txIdxr    TxIndexer
-	blockIdxr indexer.BlockIndexer
-	eventBus  *types.EventBus
+	txIdxr           TxIndexer
+	blockIdxr        indexer.BlockIndexer
+	eventBus         *types.EventBus
+	terminateOnError bool
 }
 
 // NewIndexerService returns a new service instance.
@@ -30,9 +30,10 @@ func NewIndexerService(
 	txIdxr TxIndexer,
 	blockIdxr indexer.BlockIndexer,
 	eventBus *types.EventBus,
+	terminateOnError bool,
 ) *IndexerService {
 
-	is := &IndexerService{txIdxr: txIdxr, blockIdxr: blockIdxr, eventBus: eventBus}
+	is := &IndexerService{txIdxr: txIdxr, blockIdxr: blockIdxr, eventBus: eventBus, terminateOnError: terminateOnError}
 	is.BaseService = *service.NewBaseService(nil, "IndexerService", is)
 	return is
 }
@@ -76,22 +77,36 @@ func (is *IndexerService) OnStart() error {
 						"index", txResult.Index,
 						"err", err,
 					)
+
+					if is.terminateOnError {
+						if err := is.Stop(); err != nil {
+							is.Logger.Error("failed to stop", "err", err)
+						}
+						return
+					}
 				}
 			}
 
 			if err := is.blockIdxr.Index(eventNewBlockEvents); err != nil {
 				is.Logger.Error("failed to index block", "height", height, "err", err)
+				if is.terminateOnError {
+					if err := is.Stop(); err != nil {
+						is.Logger.Error("failed to stop", "err", err)
+					}
+					return
+				}
 			} else {
-				is.Logger.Info("indexed block", "height", height)
-			}
-
-			batch.Ops, err = DeduplicateBatch(batch.Ops, is.txIdxr)
-			if err != nil {
-				is.Logger.Error("deduplicate batch", "height", height)
+				is.Logger.Info("indexed block exents", "height", height)
 			}
 
 			if err = is.txIdxr.AddBatch(batch); err != nil {
 				is.Logger.Error("failed to index block txs", "height", height, "err", err)
+				if is.terminateOnError {
+					if err := is.Stop(); err != nil {
+						is.Logger.Error("failed to stop", "err", err)
+					}
+					return
+				}
 			} else {
 				is.Logger.Debug("indexed block txs", "height", height, "num_txs", numTxs)
 			}
@@ -106,45 +121,3 @@ func (is *IndexerService) OnStop() {
 		_ = is.eventBus.UnsubscribeAll(context.Background(), subscriber)
 	}
 }
-
-// DeduplicateBatch consider the case of duplicate txs.
-// if the current one under investigation is NOT OK, then we need to check
-// whether there's a previously indexed tx.
-// SKIP the current tx if the previously indexed record is found and successful.
-func DeduplicateBatch(ops []*abci.TxResult, txIdxr TxIndexer) ([]*abci.TxResult, error) {
-	result := make([]*abci.TxResult, 0, len(ops))
-
-	// keep track of successful txs in this block in order to suppress latter ones being indexed.
-	var successfulTxsInThisBlock = make(map[string]struct{})
-
-	for _, txResult := range ops {
-		hash := types.Tx(txResult.Tx).Hash()
-
-		if txResult.Result.IsOK() {
-			successfulTxsInThisBlock[string(hash)] = struct{}{}
-		} else {
-			// if it already appeared in current block and was successful, skip.
-			if _, found := successfulTxsInThisBlock[string(hash)]; found {
-				continue
-			}
-
-			// check if this tx hash is already indexed
-			old, err := txIdxr.Get(hash)
-
-			// if db op errored
-			// Not found is not an error
-			if err != nil {
-				return nil, err
-			}
-
-			// if it's already indexed in an older block and was successful, skip.
-			if old != nil && old.Result.Code == abci.CodeTypeOK {
-				continue
-			}
-		}
-
-		result = append(result, txResult)
-	}
-
-	return result, nil
-}

state/txindex/indexer_service_test.go

@@ -32,7 +32,7 @@ func TestIndexerServiceIndexesBlocks(t *testing.T) {
 	txIndexer := kv.NewTxIndex(store)
 	blockIndexer := blockidxkv.New(db.NewPrefixDB(store, []byte("block_events")))
 
-	service := txindex.NewIndexerService(txIndexer, blockIndexer, eventBus)
+	service := txindex.NewIndexerService(txIndexer, blockIndexer, eventBus, false)
 	service.SetLogger(log.TestingLogger())
 	err = service.Start()
 	require.NoError(t, err)
@@ -91,164 +91,3 @@ func TestIndexerServiceIndexesBlocks(t *testing.T) {
 	require.NoError(t, err)
 	require.Equal(t, txResult2, res)
 }
-
-func TestTxIndexDuplicatePreviouslySuccessful(t *testing.T) {
-	var mockTx = types.Tx("MOCK_TX_HASH")
-
-	testCases := []struct {
-		name    string
-		tx1     abci.TxResult
-		tx2     abci.TxResult
-		expSkip bool // do we expect the second tx to be skipped by tx indexer
-	}{
-		{"skip, previously successful",
-			abci.TxResult{
-				Height: 1,
-				Index:  0,
-				Tx:     mockTx,
-				Result: abci.ExecTxResult{
-					Code: abci.CodeTypeOK,
-				},
-			},
-			abci.TxResult{
-				Height: 2,
-				Index:  0,
-				Tx:     mockTx,
-				Result: abci.ExecTxResult{
-					Code: abci.CodeTypeOK + 1,
-				},
-			},
-			true,
-		},
-		{"not skip, previously unsuccessful",
-			abci.TxResult{
-				Height: 1,
-				Index:  0,
-				Tx:     mockTx,
-				Result: abci.ExecTxResult{
-					Code: abci.CodeTypeOK + 1,
-				},
-			},
-			abci.TxResult{
-				Height: 2,
-				Index:  0,
-				Tx:     mockTx,
-				Result: abci.ExecTxResult{
-					Code: abci.CodeTypeOK + 1,
-				},
-			},
-			false,
-		},
-		{"not skip, both successful",
-			abci.TxResult{
-				Height: 1,
-				Index:  0,
-				Tx:     mockTx,
-				Result: abci.ExecTxResult{
-					Code: abci.CodeTypeOK,
-				},
-			},
-			abci.TxResult{
-				Height: 2,
-				Index:  0,
-				Tx:     mockTx,
-				Result: abci.ExecTxResult{
-					Code: abci.CodeTypeOK,
-				},
-			},
-			false,
-		},
-		{"not skip, both unsuccessful",
-			abci.TxResult{
-				Height: 1,
-				Index:  0,
-				Tx:     mockTx,
-				Result: abci.ExecTxResult{
-					Code: abci.CodeTypeOK + 1,
-				},
-			},
-			abci.TxResult{
-				Height: 2,
-				Index:  0,
-				Tx:     mockTx,
-				Result: abci.ExecTxResult{
-					Code: abci.CodeTypeOK + 1,
-				},
-			},
-			false,
-		},
-		{"skip, same block, previously successful",
-			abci.TxResult{
-				Height: 1,
-				Index:  0,
-				Tx:     mockTx,
-				Result: abci.ExecTxResult{
-					Code: abci.CodeTypeOK,
-				},
-			},
-			abci.TxResult{
-				Height: 1,
-				Index:  0,
-				Tx:     mockTx,
-				Result: abci.ExecTxResult{
-					Code: abci.CodeTypeOK + 1,
-				},
-			},
-			true,
-		},
-		{"not skip, same block, previously unsuccessful",
-			abci.TxResult{
-				Height: 1,
-				Index:  0,
-				Tx:     mockTx,
-				Result: abci.ExecTxResult{
-					Code: abci.CodeTypeOK + 1,
-				},
-			},
-			abci.TxResult{
-				Height: 1,
-				Index:  0,
-				Tx:     mockTx,
-				Result: abci.ExecTxResult{
-					Code: abci.CodeTypeOK,
-				},
-			},
-			false,
-		},
-	}
-
-	for _, tc := range testCases {
-		t.Run(tc.name, func(t *testing.T) {
-			indexer := kv.NewTxIndex(db.NewMemDB())
-
-			if tc.tx1.Height != tc.tx2.Height {
-				// index the first tx
-				err := indexer.AddBatch(&txindex.Batch{
-					Ops: []*abci.TxResult{&tc.tx1},
-				})
-				require.NoError(t, err)
-
-				// check if the second one should be skipped.
-				ops, err := txindex.DeduplicateBatch([]*abci.TxResult{&tc.tx2}, indexer)
-				require.NoError(t, err)
-
-				if tc.expSkip {
-					require.Empty(t, ops)
-				} else {
-					require.Equal(t, []*abci.TxResult{&tc.tx2}, ops)
-				}
-			} else {
-				// same block
-				ops := []*abci.TxResult{&tc.tx1, &tc.tx2}
-				ops, err := txindex.DeduplicateBatch(ops, indexer)
-				require.NoError(t, err)
-				if tc.expSkip {
-					// the second one is skipped
-					require.Equal(t, []*abci.TxResult{&tc.tx1}, ops)
-				} else {
-					require.Equal(t, []*abci.TxResult{&tc.tx1, &tc.tx2}, ops)
-				}
-			}
-		})
-	}
-}

state/txindex/kv/kv.go

@@ -13,6 +13,7 @@ import (
 
 	abci "github.com/tendermint/tendermint/abci/types"
 	"github.com/tendermint/tendermint/libs/pubsub/query"
+	"github.com/tendermint/tendermint/libs/pubsub/query/syntax"
 	"github.com/tendermint/tendermint/state/indexer"
 	"github.com/tendermint/tendermint/state/txindex"
 	"github.com/tendermint/tendermint/types"
@@ -101,12 +102,30 @@ func (txi *TxIndex) AddBatch(b *txindex.Batch) error {
 // that indexed from the tx's events is a composite of the event type and the
 // respective attribute's key delimited by a "." (eg. "account.number").
 // Any event with an empty type is not indexed.
+//
+// If a transaction is indexed with the same hash as a previous transaction, it will
+// be overwritten unless the tx result was NOT OK and the prior result was OK i.e.
+// more transactions that successfully executed overwrite transactions that failed
+// or successful yet older transactions.
 func (txi *TxIndex) Index(result *abci.TxResult) error {
 	b := txi.store.NewBatch()
 	defer b.Close()
 
 	hash := types.Tx(result.Tx).Hash()
 
+	if !result.Result.IsOK() {
+		oldResult, err := txi.Get(hash)
+		if err != nil {
+			return err
+		}
+
+		// if the new transaction failed and it's already indexed in an older block and was successful
+		// we skip it as we want users to get the older successful transaction when they query.
+		if oldResult != nil && oldResult.Result.Code == abci.CodeTypeOK {
+			return nil
+		}
+	}
+
 	// index tx by events
 	err := txi.indexEvents(result, hash, b)
 	if err != nil {
@@ -185,10 +204,7 @@ func (txi *TxIndex) Search(ctx context.Context, q *query.Query) ([]*abci.TxResul
 	filteredHashes := make(map[string][]byte)
 
 	// get a list of conditions (like "tx.height > 5")
-	conditions, err := q.Conditions()
-	if err != nil {
-		return nil, fmt.Errorf("error during parsing conditions from query: %w", err)
-	}
+	conditions := q.Syntax()
 
 	// if there is a hash condition, return the result immediately
 	hash, ok, err := lookForHash(conditions)
@@ -275,10 +291,10 @@ RESULTS_LOOP:
 	return results, nil
 }
 
-func lookForHash(conditions []query.Condition) (hash []byte, ok bool, err error) {
+func lookForHash(conditions []syntax.Condition) (hash []byte, ok bool, err error) {
 	for _, c := range conditions {
-		if c.CompositeKey == types.TxHashKey {
-			decoded, err := hex.DecodeString(c.Operand.(string))
+		if c.Tag == types.TxHashKey {
+			decoded, err := hex.DecodeString(c.Arg.Value())
 			return decoded, true, err
 		}
 	}
@@ -286,10 +302,10 @@ func lookForHash(conditions []query.Condition) (hash []byte, ok bool, err error)
 }
 
 // lookForHeight returns a height if there is an "height=X" condition.
-func lookForHeight(conditions []query.Condition) (height int64) {
+func lookForHeight(conditions []syntax.Condition) (height int64) {
 	for _, c := range conditions {
-		if c.CompositeKey == types.TxHeightKey && c.Op == query.OpEqual {
-			return c.Operand.(int64)
+		if c.Tag == types.TxHeightKey && c.Op == syntax.TEq {
+			return int64(c.Arg.Number())
 		}
 	}
 	return 0
@@ -302,7 +318,7 @@ func lookForHeight(conditions []query.Condition) (height int64) {
 // NOTE: filteredHashes may be empty if no previous condition has matched.
 func (txi *TxIndex) match(
 	ctx context.Context,
-	c query.Condition,
+	c syntax.Condition,
 	startKeyBz []byte,
 	filteredHashes map[string][]byte,
 	firstRun bool,
@@ -315,8 +331,8 @@ func (txi *TxIndex) match(
 
 	tmpHashes := make(map[string][]byte)
 
-	switch c.Op {
-	case query.OpEqual:
+	switch {
+	case c.Op == syntax.TEq:
 		it, err := dbm.IteratePrefix(txi.store, startKeyBz)
 		if err != nil {
 			panic(err)
@@ -338,10 +354,10 @@ func (txi *TxIndex) match(
 			panic(err)
 		}
 
-	case query.OpExists:
+	case c.Op == syntax.TExists:
 		// XXX: can't use startKeyBz here because c.Operand is nil
 		// (e.g. "account.owner/<nil>/" won't match w/ a single row)
-		it, err := dbm.IteratePrefix(txi.store, startKey(c.CompositeKey))
+		it, err := dbm.IteratePrefix(txi.store, startKey(c.Tag))
 		if err != nil {
 			panic(err)
 		}
@@ -362,11 +378,11 @@ func (txi *TxIndex) match(
 			panic(err)
 		}
 
-	case query.OpContains:
+	case c.Op == syntax.TContains:
 		// XXX: startKey does not apply here.
 		// For example, if startKey = "account.owner/an/" and search query = "account.owner CONTAINS an"
 		// we can't iterate with prefix "account.owner/an/" because we might miss keys like "account.owner/Ulan/"
-		it, err := dbm.IteratePrefix(txi.store, startKey(c.CompositeKey))
+		it, err := dbm.IteratePrefix(txi.store, startKey(c.Tag))
 		if err != nil {
 			panic(err)
 		}
@@ -377,8 +393,7 @@ func (txi *TxIndex) match(
 			if !isTagKey(it.Key()) {
 				continue
 			}
-
-			if strings.Contains(extractValueFromKey(it.Key()), c.Operand.(string)) {
+			if strings.Contains(extractValueFromKey(it.Key()), c.Arg.Value()) {
 				tmpHashes[string(it.Value())] = it.Value()
 			}
 
@@ -557,11 +572,11 @@ func keyForHeight(result *abci.TxResult) []byte {
 	))
 }
 
-func startKeyForCondition(c query.Condition, height int64) []byte {
+func startKeyForCondition(c syntax.Condition, height int64) []byte {
 	if height > 0 {
-		return startKey(c.CompositeKey, c.Operand, height)
+		return startKey(c.Tag, c.Arg.Value(), height)
 	}
-	return startKey(c.CompositeKey, c.Operand)
+	return startKey(c.Tag, c.Arg.Value())
 }
 
 func startKey(fields ...interface{}) []byte {

state/txindex/kv/kv_bench_test.go

@@ -60,7 +60,7 @@ func BenchmarkTxSearch(b *testing.B) {
 		}
 	}
 
-	txQuery := query.MustParse("transfer.address = 'address_43' AND transfer.amount = 50")
+	txQuery := query.MustCompile(`transfer.address = 'address_43' AND transfer.amount = 50`)
 
 	b.ResetTimer()
 

state/txindex/kv/kv_test.go

@@ -126,7 +126,7 @@ func TestTxSearch(t *testing.T) {
 	for _, tc := range testCases {
 		tc := tc
 		t.Run(tc.q, func(t *testing.T) {
-			results, err := indexer.Search(ctx, query.MustParse(tc.q))
+			results, err := indexer.Search(ctx, query.MustCompile(tc.q))
 			assert.NoError(t, err)
 
 			assert.Len(t, results, tc.resultsLength)
@@ -152,7 +152,7 @@ func TestTxSearchWithCancelation(t *testing.T) {
 
 	ctx, cancel := context.WithCancel(context.Background())
 	cancel()
-	results, err := indexer.Search(ctx, query.MustParse("account.number = 1"))
+	results, err := indexer.Search(ctx, query.MustCompile(`account.number = 1`))
 	assert.NoError(t, err)
 	assert.Empty(t, results)
 }
@@ -225,7 +225,7 @@ func TestTxSearchDeprecatedIndexing(t *testing.T) {
 	for _, tc := range testCases {
 		tc := tc
 		t.Run(tc.q, func(t *testing.T) {
-			results, err := indexer.Search(ctx, query.MustParse(tc.q))
+			results, err := indexer.Search(ctx, query.MustCompile(tc.q))
 			require.NoError(t, err)
 			for _, txr := range results {
 				for _, tr := range tc.results {
@@ -249,7 +249,7 @@ func TestTxSearchOneTxWithMultipleSameTagsButDifferentValues(t *testing.T) {
 
 	ctx := context.Background()
 
-	results, err := indexer.Search(ctx, query.MustParse("account.number >= 1"))
+	results, err := indexer.Search(ctx, query.MustCompile(`account.number >= 1`))
 	assert.NoError(t, err)
 
 	assert.Len(t, results, 1)
@@ -258,6 +258,103 @@ func TestTxSearchOneTxWithMultipleSameTagsButDifferentValues(t *testing.T) {
 	}
 }
 
+func TestTxIndexDuplicatePreviouslySuccessful(t *testing.T) {
+	var mockTx = types.Tx("MOCK_TX_HASH")
+
+	testCases := []struct {
+		name         string
+		tx1          *abci.TxResult
+		tx2          *abci.TxResult
+		expOverwrite bool // do we expect the second tx to overwrite the first tx
+	}{
+		{
+			"don't overwrite as a non-zero code was returned and the previous tx was successful",
+			&abci.TxResult{
+				Height: 1,
+				Index:  0,
+				Tx:     mockTx,
+				Result: abci.ExecTxResult{
+					Code: abci.CodeTypeOK,
+				},
+			},
+			&abci.TxResult{
+				Height: 2,
+				Index:  0,
+				Tx:     mockTx,
+				Result: abci.ExecTxResult{
+					Code: abci.CodeTypeOK + 1,
+				},
+			},
+			false,
+		},
+		{
+			"overwrite as the previous tx was also unsuccessful",
+			&abci.TxResult{
+				Height: 1,
+				Index:  0,
+				Tx:     mockTx,
+				Result: abci.ExecTxResult{
+					Code: abci.CodeTypeOK + 1,
+				},
+			},
+			&abci.TxResult{
+				Height: 2,
+				Index:  0,
+				Tx:     mockTx,
+				Result: abci.ExecTxResult{
+					Code: abci.CodeTypeOK + 1,
+				},
+			},
+			true,
+		},
+		{
+			"overwrite as the most recent tx was successful",
+			&abci.TxResult{
+				Height: 1,
+				Index:  0,
+				Tx:     mockTx,
+				Result: abci.ExecTxResult{
+					Code: abci.CodeTypeOK,
+				},
+			},
+			&abci.TxResult{
+				Height: 2,
+				Index:  0,
+				Tx:     mockTx,
+				Result: abci.ExecTxResult{
+					Code: abci.CodeTypeOK,
+				},
+			},
+			true,
+		},
+	}
+
+	hash := mockTx.Hash()
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			indexer := NewTxIndex(db.NewMemDB())
+
+			// index the first tx
+			err := indexer.Index(tc.tx1)
+			require.NoError(t, err)
+
+			// index the same tx with different results
+			err = indexer.Index(tc.tx2)
+			require.NoError(t, err)
+
+			res, err := indexer.Get(hash)
+			require.NoError(t, err)
+
+			if tc.expOverwrite {
+				require.Equal(t, tc.tx2, res)
+			} else {
+				require.Equal(t, tc.tx1, res)
+			}
+		})
+	}
+}
+
 func TestTxSearchMultipleTxs(t *testing.T) {
 	indexer := NewTxIndex(db.NewMemDB())
 
@@ -306,7 +403,7 @@ func TestTxSearchMultipleTxs(t *testing.T) {
 
 	ctx := context.Background()
 
-	results, err := indexer.Search(ctx, query.MustParse("account.number >= 1"))
+	results, err := indexer.Search(ctx, query.MustCompile(`account.number >= 1`))
 	assert.NoError(t, err)
 
 	require.Len(t, results, 3)

state/validation_test.go

@@ -17,8 +17,10 @@ import (
 	tmproto "github.com/tendermint/tendermint/proto/tendermint/types"
 	sm "github.com/tendermint/tendermint/state"
 	"github.com/tendermint/tendermint/state/mocks"
+	"github.com/tendermint/tendermint/store"
 	"github.com/tendermint/tendermint/types"
 	tmtime "github.com/tendermint/tendermint/types/time"
+	dbm "github.com/tendermint/tm-db"
 )
 
 const validationTestsStopHeight int64 = 10
@@ -44,12 +46,15 @@ func TestValidateBlockHeader(t *testing.T) {
 		mock.Anything,
 		mock.Anything).Return(nil)
 
+	blockStore := store.NewBlockStore(dbm.NewMemDB())
+
 	blockExec := sm.NewBlockExecutor(
 		stateStore,
 		log.TestingLogger(),
 		proxyApp.Consensus(),
 		mp,
 		sm.EmptyEvidencePool{},
+		blockStore,
 	)
 	lastCommit := types.NewCommit(0, 0, types.BlockID{}, nil)
 	var lastExtCommit *types.ExtendedCommit
@@ -136,12 +141,15 @@ func TestValidateBlockCommit(t *testing.T) {
 		mock.Anything,
 		mock.Anything).Return(nil)
 
+	blockStore := store.NewBlockStore(dbm.NewMemDB())
+
 	blockExec := sm.NewBlockExecutor(
 		stateStore,
 		log.TestingLogger(),
 		proxyApp.Consensus(),
 		mp,
 		sm.EmptyEvidencePool{},
+		blockStore,
 	)
 	lastCommit := types.NewCommit(0, 0, types.BlockID{}, nil)
 	var lastExtCommit *types.ExtendedCommit
@@ -277,12 +285,15 @@ func TestValidateBlockEvidence(t *testing.T) {
 		mock.Anything,
 		mock.Anything).Return(nil)
 	state.ConsensusParams.Evidence.MaxBytes = 1000
+	blockStore := store.NewBlockStore(dbm.NewMemDB())
+
 	blockExec := sm.NewBlockExecutor(
 		stateStore,
 		log.TestingLogger(),
 		proxyApp.Consensus(),
 		mp,
 		evpool,
+		blockStore,
 	)
 	lastCommit := types.NewCommit(0, 0, types.BlockID{}, nil)
 	var lastExtCommit *types.ExtendedCommit