Work around indexing problem for duplicate transactions (#8625)

Port the bug fix terra-money#76 to upstream. This is critical for ethermint json-rpc to work.

fix: prevent duplicate tx index if it succeeded before
fix: use CodeTypeOk instead of 0
fix: handle duplicate txs within the same block
Co-authored-by: jess jesse@soob.co

ref: #5281

state/txindex/indexer_service.go

@@ -3,6 +3,7 @@ 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"
@@ -82,6 +83,11 @@ func (is *IndexerService) OnStart() error {
 				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)
+			}
+
 			if err = is.txIdxr.AddBatch(batch); err != nil {
 				is.Logger.Error("failed to index block txs", "height", height, "err", err)
 			} else {
@@ -98,3 +104,45 @@ 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

@@ -79,3 +79,164 @@ 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.ResponseDeliverTx{
+					Code: abci.CodeTypeOK,
+				},
+			},
+			abci.TxResult{
+				Height: 2,
+				Index:  0,
+				Tx:     mockTx,
+				Result: abci.ResponseDeliverTx{
+					Code: abci.CodeTypeOK + 1,
+				},
+			},
+			true,
+		},
+		{"not skip, previously unsuccessful",
+			abci.TxResult{
+				Height: 1,
+				Index:  0,
+				Tx:     mockTx,
+				Result: abci.ResponseDeliverTx{
+					Code: abci.CodeTypeOK + 1,
+				},
+			},
+			abci.TxResult{
+				Height: 2,
+				Index:  0,
+				Tx:     mockTx,
+				Result: abci.ResponseDeliverTx{
+					Code: abci.CodeTypeOK + 1,
+				},
+			},
+			false,
+		},
+		{"not skip, both successful",
+			abci.TxResult{
+				Height: 1,
+				Index:  0,
+				Tx:     mockTx,
+				Result: abci.ResponseDeliverTx{
+					Code: abci.CodeTypeOK,
+				},
+			},
+			abci.TxResult{
+				Height: 2,
+				Index:  0,
+				Tx:     mockTx,
+				Result: abci.ResponseDeliverTx{
+					Code: abci.CodeTypeOK,
+				},
+			},
+			false,
+		},
+		{"not skip, both unsuccessful",
+			abci.TxResult{
+				Height: 1,
+				Index:  0,
+				Tx:     mockTx,
+				Result: abci.ResponseDeliverTx{
+					Code: abci.CodeTypeOK + 1,
+				},
+			},
+			abci.TxResult{
+				Height: 2,
+				Index:  0,
+				Tx:     mockTx,
+				Result: abci.ResponseDeliverTx{
+					Code: abci.CodeTypeOK + 1,
+				},
+			},
+			false,
+		},
+		{"skip, same block, previously successful",
+			abci.TxResult{
+				Height: 1,
+				Index:  0,
+				Tx:     mockTx,
+				Result: abci.ResponseDeliverTx{
+					Code: abci.CodeTypeOK,
+				},
+			},
+			abci.TxResult{
+				Height: 1,
+				Index:  0,
+				Tx:     mockTx,
+				Result: abci.ResponseDeliverTx{
+					Code: abci.CodeTypeOK + 1,
+				},
+			},
+			true,
+		},
+		{"not skip, same block, previously unsuccessful",
+			abci.TxResult{
+				Height: 1,
+				Index:  0,
+				Tx:     mockTx,
+				Result: abci.ResponseDeliverTx{
+					Code: abci.CodeTypeOK + 1,
+				},
+			},
+			abci.TxResult{
+				Height: 1,
+				Index:  0,
+				Tx:     mockTx,
+				Result: abci.ResponseDeliverTx{
+					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)
+				}
+			}
+		})
+	}
+}