feat: fill transaction index gap on enabling automatic backfill

chain/ethhashlookup/eth_transaction_hash_lookup.go

@@ -33,6 +33,7 @@ const (
 	getCidFromHash  = `SELECT cid FROM eth_tx_hashes WHERE hash = ?`
 	getHashFromCid  = `SELECT hash FROM eth_tx_hashes WHERE cid = ?`
 	deleteOlderThan = `DELETE FROM eth_tx_hashes WHERE insertion_time < datetime('now', ?);`
+	getLastTxHash   = `SELECT hash, cid FROM eth_tx_hashes ORDER BY insertion_time DESC LIMIT 1`
 )
 
 type EthTxHashLookup struct {
@@ -42,6 +43,7 @@ type EthTxHashLookup struct {
 	stmtGetCidFromHash  *sql.Stmt
 	stmtGetHashFromCid  *sql.Stmt
 	stmtDeleteOlderThan *sql.Stmt
+	stmtGetLastTxHash   *sql.Stmt
 }
 
 func NewTransactionHashLookup(ctx context.Context, path string) (*EthTxHashLookup, error) {
@@ -82,6 +84,10 @@ func (ei *EthTxHashLookup) initStatements() (err error) {
 	if err != nil {
 		return xerrors.Errorf("prepare stmtDeleteOlderThan: %w", err)
 	}
+	ei.stmtGetLastTxHash, err = ei.db.Prepare(getLastTxHash)
+	if err != nil {
+		return xerrors.Errorf("prepare stmtGetMostRecentTxHash: %w", err)
+	}
 	return nil
 }
 
@@ -128,6 +134,22 @@ func (ei *EthTxHashLookup) GetHashFromCid(c cid.Cid) (ethtypes.EthHash, error) {
 	return ethtypes.ParseEthHash(hashString)
 }
 
+// GetLastTransaction returns the most recent transaction hash and cid from the database based on insertion_time
+func (ei *EthTxHashLookup) GetLastTransaction() (string, cid.Cid, error) {
+	if ei.db == nil {
+		return "", cid.Undef, xerrors.New("db closed")
+	}
+
+	var hashStr, cidStr string
+	err := ei.stmtGetLastTxHash.QueryRow().Scan(&hashStr, &cidStr)
+	if err != nil {
+		return "", cid.Undef, err
+	}
+
+	c, err := cid.Decode(cidStr)
+	return hashStr, c, err
+}
+
 func (ei *EthTxHashLookup) DeleteEntriesOlderThan(days int) (int64, error) {
 	if ei.db == nil {
 		return 0, xerrors.New("db closed")

node/builder_chain.go

@@ -258,7 +258,7 @@ func ConfigFullNode(c interface{}) Option {
 
 			If(cfg.Fevm.EnableEthRPC,
 				Override(new(*full.EthEventHandler), modules.EthEventHandler(cfg.Events, cfg.Fevm.EnableEthRPC)),
-				Override(new(full.EthModuleAPI), modules.EthModuleAPI(cfg.Fevm)),
+				Override(new(full.EthModuleAPI), modules.EthModuleAPI(cfg.Fevm, cfg.Index.EnableAutomaticBackFill)),
 				Override(new(full.EthEventAPI), From(new(*full.EthEventHandler))),
 			),
 			If(!cfg.Fevm.EnableEthRPC,

node/config/types.go

@@ -71,7 +71,7 @@ type ApisConfig struct {
 }
 
 type JournalConfig struct {
-	//Events of the form: "system1:event1,system1:event2[,...]"
+	// Events of the form: "system1:event1,system1:event2[,...]"
 	DisabledEvents string
 }
 
@@ -621,6 +621,9 @@ type IndexConfig struct {
 	// EXPERIMENTAL FEATURE. USE WITH CAUTION
 	// EnableMsgIndex enables indexing of messages on chain.
 	EnableMsgIndex bool
+
+	// EnableAutomaticBackFill enables automatic index back-filling
+	EnableAutomaticBackFill bool
 }
 
 type HarmonyDB struct {

node/impl/full/txhashmanager.go

@@ -23,6 +23,49 @@ func (m *EthTxHashManager) Revert(ctx context.Context, from, to *types.TipSet) e
 	return nil
 }
 
+// FillIndexGap back-fills the gap between the current head and the last populated transaction index
+func (m *EthTxHashManager) FillIndexGap(ctx context.Context) error {
+	lastTxHash, lastTxCid, err := m.TransactionHashLookup.GetLastTransaction()
+	if err != nil {
+		return err
+	}
+
+	log.Infof("Start filling transaction index gap from: %s", lastTxHash)
+
+	totalProcessedBlock := 0
+	ts := m.StateAPI.Chain.GetHeaviestTipSet()
+	for _, block := range ts.Blocks() {
+		msgs, err := m.StateAPI.Chain.SecpkMessagesForBlock(ctx, block)
+		if err != nil {
+			// If we can't find the messages, we've either imported from snapshot or pruned the store
+			log.Debug("exiting message mapping population at epoch ", ts.Height())
+			return nil
+		}
+
+		for _, msg := range msgs {
+			if msg.Cid() == lastTxCid {
+				// We've reached the last indexed transaction
+				return nil
+			}
+
+			// process the message
+			m.ProcessSignedMessage(ctx, msg)
+		}
+
+		totalProcessedBlock++
+	}
+
+	ts, err = m.StateAPI.Chain.GetTipSetFromKey(ctx, ts.Parents())
+	if err != nil {
+		return err
+	}
+
+	log.Infof("Finished filling transaction index gap. Total processed block: %d", totalProcessedBlock)
+
+	return nil
+}
+
+// PopulateExistingMappings walk back from the current head to the minimum height and populate the eth transaction hash lookup database
 func (m *EthTxHashManager) PopulateExistingMappings(ctx context.Context, minHeight abi.ChainEpoch) error {
 	if minHeight < buildconstants.UpgradeHyggeHeight {
 		minHeight = buildconstants.UpgradeHyggeHeight

node/modules/ethmodule.go

@@ -7,6 +7,7 @@ import (
 	"time"
 
 	"go.uber.org/fx"
+	"golang.org/x/xerrors"
 
 	"github.com/filecoin-project/go-state-types/abi"
 
@@ -21,7 +22,7 @@ import (
 	"github.com/filecoin-project/lotus/node/repo"
 )
 
-func EthModuleAPI(cfg config.FevmConfig) func(helpers.MetricsCtx, repo.LockedRepo, fx.Lifecycle, *store.ChainStore, *stmgr.StateManager, EventHelperAPI, *messagepool.MessagePool, full.StateAPI, full.ChainAPI, full.MpoolAPI, full.SyncAPI, *full.EthEventHandler) (*full.EthModule, error) {
+func EthModuleAPI(cfg config.FevmConfig, enableAutomaticBackFill bool) func(helpers.MetricsCtx, repo.LockedRepo, fx.Lifecycle, *store.ChainStore, *stmgr.StateManager, EventHelperAPI, *messagepool.MessagePool, full.StateAPI, full.ChainAPI, full.MpoolAPI, full.SyncAPI, *full.EthEventHandler) (*full.EthModule, error) {
 	return func(mctx helpers.MetricsCtx, r repo.LockedRepo, lc fx.Lifecycle, cs *store.ChainStore, sm *stmgr.StateManager, evapi EventHelperAPI, mp *messagepool.MessagePool, stateapi full.StateAPI, chainapi full.ChainAPI, mpoolapi full.MpoolAPI, syncapi full.SyncAPI, ethEventHandler *full.EthEventHandler) (*full.EthModule, error) {
 		ctx := helpers.LifecycleCtx(mctx, lc)
 
@@ -57,6 +58,11 @@ func EthModuleAPI(cfg config.FevmConfig) func(helpers.MetricsCtx, repo.LockedRep
 			if err != nil {
 				return nil, err
 			}
+		} else if enableAutomaticBackFill { // If the db exists and back-fill is enabled, we'll back-fill missing entries
+			err = ethTxHashManager.FillIndexGap(mctx)
+			if err != nil {
+				return nil, xerrors.Errorf("error when back-filling transaction index gap: %w", err)
+			}
 		}
 
 		// prefill the whole skiplist cache maintained internally by the GetTipsetByHeight
@@ -78,7 +84,7 @@ func EthModuleAPI(cfg config.FevmConfig) func(helpers.MetricsCtx, repo.LockedRep
 				}
 
 				// Tipset listener
-				_ = ev.Observe(&ethTxHashManager)
+				ev.Observe(&ethTxHashManager)
 
 				ch, err := mp.Updates(ctx)
 				if err != nil {