Add gc for eth tx database

chain/ethhashlookup/eth_transaction_hash_lookup.go

@@ -2,17 +2,18 @@ package ethhashlookup
 
 import (
 	"database/sql"
-	"math"
+	"errors"
+	"strconv"
 
 	"github.com/ipfs/go-cid"
 	_ "github.com/mattn/go-sqlite3"
 	"golang.org/x/xerrors"
 
-	"github.com/filecoin-project/go-state-types/abi"
-
 	"github.com/filecoin-project/lotus/chain/types/ethtypes"
 )
 
+var ErrNotFound = errors.New("not found")
+
 var pragmas = []string{
 	"PRAGMA synchronous = normal",
 	"PRAGMA temp_store = memory",
@@ -28,10 +29,10 @@ var ddls = []string{
 	`CREATE TABLE IF NOT EXISTS eth_tx_hashes (
 		hash TEXT PRIMARY KEY NOT NULL,
 		cid TEXT NOT NULL UNIQUE,
-		epoch INT NOT NULL
+		insertion_time TIMESTAMP DEFAULT CURRENT_TIMESTAMP NOT NULL
 	)`,
 
-	`CREATE INDEX IF NOT EXISTS eth_tx_hashes_epoch_index ON eth_tx_hashes (epoch)`,
+	`CREATE INDEX IF NOT EXISTS insertion_time_index ON eth_tx_hashes (insertion_time)`,
 
 	// metadata containing version of schema
 	`CREATE TABLE IF NOT EXISTS _meta (
@@ -43,39 +44,37 @@ var ddls = []string{
 }
 
 const schemaVersion = 1
-const MemPoolEpoch = math.MaxInt64
 
 const (
 	insertTxHash = `INSERT INTO eth_tx_hashes
-	(hash, cid, epoch)
-	VALUES(?, ?, ?)
-	ON CONFLICT (hash) DO UPDATE SET epoch = EXCLUDED.epoch
-	WHERE epoch > EXCLUDED.epoch`
+	(hash, cid)
+	VALUES(?, ?)
+	ON CONFLICT (hash) DO UPDATE SET insertion_time = CURRENT_TIMESTAMP`
 )
 
-type TransactionHashLookup struct {
+type EthTxHashLookup struct {
 	db *sql.DB
 }
 
-func (ei *TransactionHashLookup) InsertTxHash(txHash ethtypes.EthHash, c cid.Cid, epoch int64) error {
+func (ei *EthTxHashLookup) UpsertHash(txHash ethtypes.EthHash, c cid.Cid) error {
 	hashEntry, err := ei.db.Prepare(insertTxHash)
 	if err != nil {
 		return xerrors.Errorf("prepare insert event: %w", err)
 	}
 
-	_, err = hashEntry.Exec(txHash.String(), c.String(), epoch)
+	_, err = hashEntry.Exec(txHash.String(), c.String())
 	return err
 }
 
-func (ei *TransactionHashLookup) LookupCidFromTxHash(txHash ethtypes.EthHash) (cid.Cid, error) {
+func (ei *EthTxHashLookup) GetCidFromHash(txHash ethtypes.EthHash) (cid.Cid, error) {
 	q, err := ei.db.Query("SELECT cid FROM eth_tx_hashes WHERE hash = :hash;", sql.Named("hash", txHash.String()))
 	if err != nil {
 		return cid.Undef, err
 	}
 
 	var c string
 	if !q.Next() {
-		return cid.Undef, xerrors.Errorf("transaction hash %s not found", txHash.String())
+		return cid.Undef, ErrNotFound
 	}
 	err = q.Scan(&c)
 	if err != nil {
@@ -84,15 +83,15 @@ func (ei *TransactionHashLookup) LookupCidFromTxHash(txHash ethtypes.EthHash) (c
 	return cid.Decode(c)
 }
 
-func (ei *TransactionHashLookup) LookupTxHashFromCid(c cid.Cid) (ethtypes.EthHash, error) {
+func (ei *EthTxHashLookup) GetHashFromCid(c cid.Cid) (ethtypes.EthHash, error) {
 	q, err := ei.db.Query("SELECT hash FROM eth_tx_hashes WHERE cid = :cid;", sql.Named("cid", c.String()))
 	if err != nil {
 		return ethtypes.EmptyEthHash, err
 	}
 
 	var hashString string
 	if !q.Next() {
-		return ethtypes.EmptyEthHash, xerrors.Errorf("transaction hash %s not found", c.String())
+		return ethtypes.EmptyEthHash, ErrNotFound
 	}
 	err = q.Scan(&hashString)
 	if err != nil {
@@ -101,16 +100,16 @@ func (ei *TransactionHashLookup) LookupTxHashFromCid(c cid.Cid) (ethtypes.EthHas
 	return ethtypes.ParseEthHash(hashString)
 }
 
-func (ei *TransactionHashLookup) RemoveEntriesOlderThan(epoch abi.ChainEpoch) (int64, error) {
-	res, err := ei.db.Exec("DELETE FROM eth_tx_hashes WHERE epoch < :epoch;", sql.Named("epoch", epoch))
+func (ei *EthTxHashLookup) DeleteEntriesOlderThan(days int) (int64, error) {
+	res, err := ei.db.Exec("DELETE FROM eth_tx_hashes WHERE insertion_time < datetime('now', ?);", "-"+strconv.Itoa(days)+" day")
 	if err != nil {
 		return 0, err
 	}
 
 	return res.RowsAffected()
 }
 
-func NewTransactionHashLookup(path string) (*TransactionHashLookup, error) {
+func NewTransactionHashLookup(path string) (*EthTxHashLookup, error) {
 	db, err := sql.Open("sqlite3", path+"?mode=rwc")
 	if err != nil {
 		return nil, xerrors.Errorf("open sqlite3 database: %w", err)
@@ -151,12 +150,12 @@ func NewTransactionHashLookup(path string) (*TransactionHashLookup, error) {
 		}
 	}
 
-	return &TransactionHashLookup{
+	return &EthTxHashLookup{
 		db: db,
 	}, nil
 }
 
-func (ei *TransactionHashLookup) Close() error {
+func (ei *EthTxHashLookup) Close() error {
 	if ei.db == nil {
 		return nil
 	}

documentation/en/default-lotus-config.toml

@@ -350,4 +350,11 @@
   # env var: LOTUS_FEVM_ENABLEETHHASHTOFILECOINCIDMAPPING
   #EnableEthHashToFilecoinCidMapping = false
 
+  # EthTxHashMappingLifetimeDays the transaction hash lookup database will delete mappings that have been stored for more than x days
+  # Set to 0 to keep all mappings
+  #
+  # type: int
+  # env var: LOTUS_FEVM_ETHTXHASHMAPPINGLIFETIMEDAYS
+  #EthTxHashMappingLifetimeDays = 0
+
 

itests/eth_transactions_test.go

@@ -106,7 +106,7 @@ func TestLegacyTransaction(t *testing.T) {
 func TestContractDeploymentValidSignature(t *testing.T) {
 
 	blockTime := 100 * time.Millisecond
-	client, _, ens := kit.EnsembleMinimal(t, kit.MockProofs(), kit.ThroughRPC())
+	client, _, ens := kit.EnsembleMinimal(t, kit.MockProofs(), kit.ThroughRPC(), kit.EthTxHashLookup())
 
 	ens.InterconnectAll().BeginMining(blockTime)
 
@@ -167,7 +167,7 @@ func TestContractDeploymentValidSignature(t *testing.T) {
 
 func TestContractInvocation(t *testing.T) {
 	blockTime := 100 * time.Millisecond
-	client, _, ens := kit.EnsembleMinimal(t, kit.MockProofs(), kit.ThroughRPC())
+	client, _, ens := kit.EnsembleMinimal(t, kit.MockProofs(), kit.ThroughRPC(), kit.EthTxHashLookup())
 
 	ens.InterconnectAll().BeginMining(blockTime)
 

itests/fevm_test.go

@@ -19,7 +19,7 @@ import (
 	"github.com/filecoin-project/lotus/itests/kit"
 )
 
-// TestFEVMBasic does a basic ethhash contract installation and invocation
+// TestFEVMBasic does a basic fevm contract installation and invocation
 func TestFEVMBasic(t *testing.T) {
 	// TODO the contract installation and invocation can be lifted into utility methods
 	// He who writes the second test, shall do that.

node/config/types.go

@@ -697,4 +697,7 @@ type ActorEventConfig struct {
 type FevmConfig struct {
 	// EnableEthHashToFilecoinCidMapping enables storing a mapping of eth transaction hashes to filecoin message Cids
 	EnableEthHashToFilecoinCidMapping bool
+	// EthTxHashMappingLifetimeDays the transaction hash lookup database will delete mappings that have been stored for more than x days
+	// Set to 0 to keep all mappings
+	EthTxHashMappingLifetimeDays int
 }

node/impl/full/eth.go

@@ -259,7 +259,7 @@ func (a *EthModule) EthGetTransactionByHash(ctx context.Context, txHash *ethtype
 	c := cid.Undef
 	if a.EthTxHashManager != nil {
 		var err error
-		c, err = a.EthTxHashManager.TransactionHashLookup.LookupCidFromTxHash(*txHash)
+		c, err = a.EthTxHashManager.TransactionHashLookup.GetCidFromHash(*txHash)
 		if err != nil {
 			log.Debug("could not find transaction hash %s in lookup table", txHash.String())
 		}
@@ -326,7 +326,7 @@ func (a *EthModule) EthGetTransactionReceipt(ctx context.Context, txHash ethtype
 	c := cid.Undef
 	if a.EthTxHashManager != nil {
 		var err error
-		c, err = a.EthTxHashManager.TransactionHashLookup.LookupCidFromTxHash(txHash)
+		c, err = a.EthTxHashManager.TransactionHashLookup.GetCidFromHash(txHash)
 		if err != nil {
 			log.Debug("could not find transaction hash %s in lookup table", txHash.String())
 		}
@@ -1777,7 +1777,7 @@ func (m *EthTxHashManager) Apply(ctx context.Context, from, to *types.TipSet) er
 				return err
 			}
 
-			err = m.TransactionHashLookup.InsertTxHash(hash, smsg.Cid(), int64(to.Height()))
+			err = m.TransactionHashLookup.UpsertHash(hash, smsg.Cid())
 			if err != nil {
 				return err
 			}
@@ -1789,7 +1789,7 @@ func (m *EthTxHashManager) Apply(ctx context.Context, from, to *types.TipSet) er
 
 type EthTxHashManager struct {
 	StateAPI              StateAPI
-	TransactionHashLookup *ethhashlookup.TransactionHashLookup
+	TransactionHashLookup *ethhashlookup.EthTxHashLookup
 }
 
 func (m *EthTxHashManager) Revert(ctx context.Context, from, to *types.TipSet) error {
@@ -1814,14 +1814,30 @@ func WaitForMpoolUpdates(ctx context.Context, ch <-chan api.MpoolUpdate, manager
 				log.Errorf("error converting filecoin message to eth tx: %s", err)
 			}
 
-			err = manager.TransactionHashLookup.InsertTxHash(ethTx.Hash, u.Message.Cid(), ethhashlookup.MemPoolEpoch)
+			err = manager.TransactionHashLookup.UpsertHash(ethTx.Hash, u.Message.Cid())
 			if err != nil {
 				log.Errorf("error inserting tx mapping to db: %s", err)
 			}
 		}
 	}
 }
 
+func EthTxHashGC(ctx context.Context, retentionDays int, manager *EthTxHashManager) {
+	if retentionDays == 0 {
+		return
+	}
+
+	gcPeriod := 1 * time.Hour
+	for {
+		entriesDeleted, err := manager.TransactionHashLookup.DeleteEntriesOlderThan(retentionDays)
+		if err != nil {
+			log.Errorf("error garbage collecting eth transaction hash database: %s", err)
+		}
+		log.Info("garbage collection run on eth transaction hash lookup database. %d entries deleted", entriesDeleted)
+		time.Sleep(gcPeriod)
+	}
+}
+
 // decodeLogBytes decodes a CBOR-serialized array into its original form.
 //
 // This function swallows errors and returns the original array if it failed

node/modules/ethmodule.go

@@ -74,6 +74,7 @@ func EthModuleAPI(cfg config.FevmConfig) func(helpers.MetricsCtx, repo.LockedRep
 					return err
 				}
 				go full.WaitForMpoolUpdates(ctx, ch, &ethTxHashManager)
+				go full.EthTxHashGC(ctx, cfg.EthTxHashMappingLifetimeDays, &ethTxHashManager)
 
 				return nil
 			},