rewrite indexer to be a listener of eventBus

state/execution.go

@@ -8,7 +8,6 @@ import (
 	abci "github.com/tendermint/abci/types"
 	crypto "github.com/tendermint/go-crypto"
 	"github.com/tendermint/tendermint/proxy"
-	"github.com/tendermint/tendermint/state/txindex"
 	"github.com/tendermint/tendermint/types"
 	cmn "github.com/tendermint/tmlibs/common"
 	"github.com/tendermint/tmlibs/log"
@@ -54,47 +53,25 @@ func execBlockOnProxyApp(txEventPublisher types.TxEventPublisher, proxyAppConn p
 			// TODO: make use of this info
 			// Blocks may include invalid txs.
 			// reqDeliverTx := req.(abci.RequestDeliverTx)
-			txError := ""
 			txResult := r.DeliverTx
 			if txResult.Code == abci.CodeType_OK {
 				validTxs++
 			} else {
 				logger.Debug("Invalid tx", "code", txResult.Code, "log", txResult.Log)
 				invalidTxs++
-				txError = txResult.Code.String()
 			}
 
-			abciResponses.DeliverTx[txIndex] = txResult
-			txIndex++
-
 			// NOTE: if we count we can access the tx from the block instead of
 			// pulling it from the req
-			tx := types.Tx(req.GetDeliverTx().Tx)
+			txEventPublisher.PublishEventTx(types.EventDataTx{types.TxResult{
+				Height: uint64(block.Height),
+				Index:  uint32(txIndex),
+				Tx:     types.Tx(req.GetDeliverTx().Tx),
+				Result: *txResult,
+			}})
 
-			tags := make(map[string]interface{})
-			for _, t := range txResult.Tags {
-				// basic validation
-				if t.Key == "" {
-					logger.Info("Got tag with an empty key (skipping)", "tag", t, "tx", tx)
-					continue
-				}
-
-				if t.ValueString != "" {
-					tags[t.Key] = t.ValueString
-				} else {
-					tags[t.Key] = t.ValueInt
-				}
-			}
-
-			txEventPublisher.PublishEventTx(types.EventDataTx{
-				Height: block.Height,
-				Tx:     tx,
-				Data:   txResult.Data,
-				Code:   txResult.Code,
-				Log:    txResult.Log,
-				Tags:   tags,
-				Error:  txError,
-			})
+			abciResponses.DeliverTx[txIndex] = txResult
+			txIndex++
 		}
 	}
 	proxyAppConn.SetResponseCallback(proxyCb)
@@ -227,7 +204,6 @@ func (s *State) validateBlock(block *types.Block) error {
 //-----------------------------------------------------------------------------
 // ApplyBlock validates & executes the block, updates state w/ ABCI responses,
 // then commits and updates the mempool atomically, then saves state.
-// Transaction results are optionally indexed.
 
 // ApplyBlock validates the block against the state, executes it against the app,
 // commits it, and saves the block and state. It's the only function that needs to be called
@@ -242,9 +218,6 @@ func (s *State) ApplyBlock(txEventPublisher types.TxEventPublisher, proxyAppConn
 
 	fail.Fail() // XXX
 
-	// index txs. This could run in the background
-	s.indexTxs(abciResponses)
-
 	// save the results before we commit
 	s.SaveABCIResponses(abciResponses)
 
@@ -293,26 +266,6 @@ func (s *State) CommitStateUpdateMempool(proxyAppConn proxy.AppConnConsensus, bl
 	return mempool.Update(block.Height, block.Txs)
 }
 
-func (s *State) indexTxs(abciResponses *ABCIResponses) {
-	// save the tx results using the TxIndexer
-	// NOTE: these may be overwriting, but the values should be the same.
-	batch := txindex.NewBatch(len(abciResponses.DeliverTx))
-	for i, d := range abciResponses.DeliverTx {
-		tx := abciResponses.txs[i]
-		if err := batch.Add(types.TxResult{
-			Height: uint64(abciResponses.Height),
-			Index:  uint32(i),
-			Tx:     tx,
-			Result: *d,
-		}); err != nil {
-			s.logger.Error("Error with batch.Add", "err", err)
-		}
-	}
-	if err := s.TxIndexer.AddBatch(batch); err != nil {
-		s.logger.Error("Error adding batch", "err", err)
-	}
-}
-
 // ExecCommitBlock executes and commits a block on the proxyApp without validating or mutating the state.
 // It returns the application root hash (result of abci.Commit).
 func ExecCommitBlock(appConnConsensus proxy.AppConnConsensus, block *types.Block, logger log.Logger) ([]byte, error) {

state/execution_test.go

@@ -3,13 +3,11 @@ package state
 import (
 	"testing"
 
-	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 
 	"github.com/tendermint/abci/example/dummy"
 	crypto "github.com/tendermint/go-crypto"
 	"github.com/tendermint/tendermint/proxy"
-	"github.com/tendermint/tendermint/state/txindex"
 	"github.com/tendermint/tendermint/types"
 	dbm "github.com/tendermint/tmlibs/db"
 	"github.com/tendermint/tmlibs/log"
@@ -31,8 +29,6 @@ func TestApplyBlock(t *testing.T) {
 
 	state := state()
 	state.SetLogger(log.TestingLogger())
-	indexer := &dummyIndexer{0}
-	state.TxIndexer = indexer
 
 	// make block
 	block := makeBlock(1, state)
@@ -40,7 +36,6 @@ func TestApplyBlock(t *testing.T) {
 	err = state.ApplyBlock(types.NopEventBus{}, proxyApp.Consensus(), block, block.MakePartSet(testPartSize).Header(), types.MockMempool{})
 
 	require.Nil(t, err)
-	assert.Equal(t, nTxsPerBlock, indexer.Indexed) // test indexing works
 
 	// TODO check state and mempool
 }
@@ -75,16 +70,3 @@ func makeBlock(num int, state *State) *types.Block {
 		prevBlockID, valHash, state.AppHash, testPartSize)
 	return block
 }
-
-// dummyIndexer increments counter every time we index transaction.
-type dummyIndexer struct {
-	Indexed int
-}
-
-func (indexer *dummyIndexer) Get(hash []byte) (*types.TxResult, error) {
-	return nil, nil
-}
-func (indexer *dummyIndexer) AddBatch(batch *txindex.Batch) error {
-	indexer.Indexed += batch.Size()
-	return nil
-}

state/state.go

@@ -15,8 +15,6 @@ import (
 
 	wire "github.com/tendermint/go-wire"
 
-	"github.com/tendermint/tendermint/state/txindex"
-	"github.com/tendermint/tendermint/state/txindex/null"
 	"github.com/tendermint/tendermint/types"
 )
 
@@ -61,9 +59,6 @@ type State struct {
 	// AppHash is updated after Commit
 	AppHash []byte
 
-	// TxIndexer indexes transactions
-	TxIndexer txindex.TxIndexer `json:"-"`
-
 	logger log.Logger
 }
 
@@ -95,7 +90,7 @@ func loadState(db dbm.DB, key []byte) *State {
 		return nil
 	}
 
-	s := &State{db: db, TxIndexer: &null.TxIndex{}}
+	s := &State{db: db}
 	r, n, err := bytes.NewReader(buf), new(int), new(error)
 	wire.ReadBinaryPtr(&s, r, 0, n, err)
 	if *err != nil {
@@ -114,8 +109,6 @@ func (s *State) SetLogger(l log.Logger) {
 }
 
 // Copy makes a copy of the State for mutating.
-// NOTE: Does not create a copy of TxIndexer. It creates a new pointer that points to the same
-// underlying TxIndexer.
 func (s *State) Copy() *State {
 	return &State{
 		db:                          s.db,
@@ -125,7 +118,6 @@ func (s *State) Copy() *State {
 		Validators:                  s.Validators.Copy(),
 		LastValidators:              s.LastValidators.Copy(),
 		AppHash:                     s.AppHash,
-		TxIndexer:                   s.TxIndexer,
 		LastHeightValidatorsChanged: s.LastHeightValidatorsChanged,
 		logger:  s.logger,
 		ChainID: s.ChainID,
@@ -368,7 +360,6 @@ func MakeGenesisState(db dbm.DB, genDoc *types.GenesisDoc) (*State, error) {
 		}
 	}
 
-	// we do not need indexer during replay and in tests
 	return &State{
 		db: db,
 
@@ -381,7 +372,6 @@ func MakeGenesisState(db dbm.DB, genDoc *types.GenesisDoc) (*State, error) {
 		Validators:                  types.NewValidatorSet(validators),
 		LastValidators:              types.NewValidatorSet(nil),
 		AppHash:                     genDoc.AppHash,
-		TxIndexer:                   &null.TxIndex{},
 		LastHeightValidatorsChanged: 1,
 	}, nil
 }

state/txindex/indexer.go

@@ -9,12 +9,12 @@ import (
 // TxIndexer interface defines methods to index and search transactions.
 type TxIndexer interface {
 
-	// AddBatch analyzes, indexes or stores a batch of transactions.
-	// NOTE: We do not specify Index method for analyzing a single transaction
-	// here because it bears heavy performance losses. Almost all advanced indexers
-	// support batching.
+	// AddBatch analyzes, indexes and stores a batch of transactions.
 	AddBatch(b *Batch) error
 
+	// Index analyzes, indexes and stores a single transaction.
+	Index(result *types.TxResult) error
+
 	// Get returns the transaction specified by hash or nil if the transaction is not indexed
 	// or stored.
 	Get(hash []byte) (*types.TxResult, error)

state/txindex/kv/kv.go

@@ -4,7 +4,7 @@ import (
 	"bytes"
 	"fmt"
 
-	"github.com/tendermint/go-wire"
+	wire "github.com/tendermint/go-wire"
 
 	db "github.com/tendermint/tmlibs/db"
 
@@ -56,3 +56,10 @@ func (txi *TxIndex) AddBatch(b *txindex.Batch) error {
 	storeBatch.Write()
 	return nil
 }
+
+// Index writes a single transaction into the TxIndex storage.
+func (txi *TxIndex) Index(result *types.TxResult) error {
+	rawBytes := wire.BinaryBytes(result)
+	txi.store.Set(result.Tx.Hash(), rawBytes)
+	return nil
+}

state/txindex/kv/kv_test.go

@@ -30,6 +30,17 @@ func TestTxIndex(t *testing.T) {
 	loadedTxResult, err := indexer.Get(hash)
 	require.Nil(t, err)
 	assert.Equal(t, txResult, loadedTxResult)
+
+	tx2 := types.Tx("BYE BYE WORLD")
+	txResult2 := &types.TxResult{1, 0, tx2, abci.ResponseDeliverTx{Data: []byte{0}, Code: abci.CodeType_OK, Log: "", Tags: []*abci.KVPair{}}}
+	hash2 := tx2.Hash()
+
+	err = indexer.Index(txResult2)
+	require.Nil(t, err)
+
+	loadedTxResult2, err := indexer.Get(hash2)
+	require.Nil(t, err)
+	assert.Equal(t, txResult2, loadedTxResult2)
 }
 
 func benchmarkTxIndex(txsCount int, b *testing.B) {

state/txindex/null/null.go

@@ -19,3 +19,8 @@ func (txi *TxIndex) Get(hash []byte) (*types.TxResult, error) {
 func (txi *TxIndex) AddBatch(batch *txindex.Batch) error {
 	return nil
 }
+
+// Index is a noop and always returns nil.
+func (txi *TxIndex) Index(result *types.TxResult) error {
+	return nil
+}