Store mapping from hashes for Ethereum transactions to Filecoin Messa…

…ge Cids

.circleci/config.yml

@@ -746,6 +746,11 @@ workflows:
           suite: itest-eth_filter
           target: "./itests/eth_filter_test.go"
 
+      - test:
+          name: test-itest-eth_hash_lookup
+          suite: itest-eth_hash_lookup
+          target: "./itests/eth_hash_lookup_test.go"
+
       - test:
           name: test-itest-eth_transactions
           suite: itest-eth_transactions

api/api_full.go

@@ -778,6 +778,7 @@ type FullNode interface {
 	EthGetBlockByHash(ctx context.Context, blkHash ethtypes.EthHash, fullTxInfo bool) (ethtypes.EthBlock, error)                               //perm:read
 	EthGetBlockByNumber(ctx context.Context, blkNum string, fullTxInfo bool) (ethtypes.EthBlock, error)                                        //perm:read
 	EthGetTransactionByHash(ctx context.Context, txHash *ethtypes.EthHash) (*ethtypes.EthTx, error)                                            //perm:read
+	EthGetTransactionHashByCid(ctx context.Context, cid cid.Cid) (*ethtypes.EthHash, error)                                                    //perm:read
 	EthGetTransactionCount(ctx context.Context, sender ethtypes.EthAddress, blkOpt string) (ethtypes.EthUint64, error)                         //perm:read
 	EthGetTransactionReceipt(ctx context.Context, txHash ethtypes.EthHash) (*EthTxReceipt, error)                                              //perm:read
 	EthGetTransactionByBlockHashAndIndex(ctx context.Context, blkHash ethtypes.EthHash, txIndex ethtypes.EthUint64) (ethtypes.EthTx, error)    //perm:read

chain/eth_transaction_hash_lookup.go

@@ -0,0 +1,162 @@
+package chain
+
+import (
+	"database/sql"
+	"math"
+
+	"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 pragmas = []string{
+	"PRAGMA synchronous = normal",
+	"PRAGMA temp_store = memory",
+	"PRAGMA mmap_size = 30000000000",
+	"PRAGMA page_size = 32768",
+	"PRAGMA auto_vacuum = NONE",
+	"PRAGMA automatic_index = OFF",
+	"PRAGMA journal_mode = WAL",
+	"PRAGMA read_uncommitted = ON",
+}
+
+var ddls = []string{
+	`CREATE TABLE IF NOT EXISTS tx_hash_lookup (
+		hash TEXT PRIMARY KEY,
+		cid TEXT NOT NULL,
+		epoch INT NOT NULL
+	)`,
+
+	// metadata containing version of schema
+	`CREATE TABLE IF NOT EXISTS _meta (
+    	version UINT64 NOT NULL UNIQUE
+	)`,
+
+	// version 1.
+	`INSERT OR IGNORE INTO _meta (version) VALUES (1)`,
+}
+
+const schemaVersion = 1
+const MemPoolEpoch = math.MaxInt64
+
+const (
+	insertTxHash = `INSERT INTO tx_hash_lookup
+	(hash, cid, epoch)
+	VALUES(?, ?, ?)
+	ON CONFLICT (hash) DO UPDATE SET epoch = EXCLUDED.epoch
+	WHERE epoch > EXCLUDED.epoch`
+)
+
+type TransactionHashLookup struct {
+	db *sql.DB
+}
+
+func (ei *TransactionHashLookup) InsertTxHash(txHash ethtypes.EthHash, c cid.Cid, epoch int64) 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)
+	return err
+}
+
+func (ei *TransactionHashLookup) LookupCidFromTxHash(txHash ethtypes.EthHash) (cid.Cid, error) {
+	q, err := ei.db.Query("SELECT cid FROM tx_hash_lookup 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())
+	}
+	err = q.Scan(&c)
+	if err != nil {
+		return cid.Undef, err
+	}
+	return cid.Decode(c)
+}
+
+func (ei *TransactionHashLookup) LookupTxHashFromCid(c cid.Cid) (ethtypes.EthHash, error) {
+	q, err := ei.db.Query("SELECT hash FROM tx_hash_lookup 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())
+	}
+	err = q.Scan(&hashString)
+	if err != nil {
+		return ethtypes.EmptyEthHash, err
+	}
+	return ethtypes.ParseEthHash(hashString)
+}
+
+func (ei *TransactionHashLookup) RemoveEntriesOlderThan(epoch abi.ChainEpoch) (int64, error) {
+	res, err := ei.db.Exec("DELETE FROM tx_hash_lookup WHERE epoch < :epoch;", sql.Named("epoch", epoch))
+	if err != nil {
+		return 0, err
+	}
+
+	return res.RowsAffected()
+}
+
+func NewTransactionHashLookup(path string) (*TransactionHashLookup, error) {
+	db, err := sql.Open("sqlite3", path+"?mode=rwc")
+	if err != nil {
+		return nil, xerrors.Errorf("open sqlite3 database: %w", err)
+	}
+
+	for _, pragma := range pragmas {
+		if _, err := db.Exec(pragma); err != nil {
+			_ = db.Close()
+			return nil, xerrors.Errorf("exec pragma %q: %w", pragma, err)
+		}
+	}
+
+	q, err := db.Query("SELECT name FROM sqlite_master WHERE type='table' AND name='_meta';")
+	if err == sql.ErrNoRows || !q.Next() {
+		// empty database, create the schema
+		for _, ddl := range ddls {
+			if _, err := db.Exec(ddl); err != nil {
+				_ = db.Close()
+				return nil, xerrors.Errorf("exec ddl %q: %w", ddl, err)
+			}
+		}
+	} else if err != nil {
+		_ = db.Close()
+		return nil, xerrors.Errorf("looking for _meta table: %w", err)
+	} else {
+		// Ensure we don't open a database from a different schema version
+
+		row := db.QueryRow("SELECT max(version) FROM _meta")
+		var version int
+		err := row.Scan(&version)
+		if err != nil {
+			_ = db.Close()
+			return nil, xerrors.Errorf("invalid database version: no version found")
+		}
+		if version != schemaVersion {
+			_ = db.Close()
+			return nil, xerrors.Errorf("invalid database version: got %d, expected %d", version, schemaVersion)
+		}
+	}
+
+	return &TransactionHashLookup{
+		db: db,
+	}, nil
+}
+
+func (ei *TransactionHashLookup) Close() error {
+	if ei.db == nil {
+		return nil
+	}
+	return ei.db.Close()
+}

chain/events/filter/mempool.go

@@ -5,7 +5,6 @@ import (
 	"sync"
 	"time"
 
-	"github.com/ipfs/go-cid"
 	"golang.org/x/xerrors"
 
 	"github.com/filecoin-project/lotus/api"
@@ -18,7 +17,7 @@ type MemPoolFilter struct {
 	ch         chan<- interface{}
 
 	mu        sync.Mutex
-	collected []cid.Cid
+	collected []*types.SignedMessage
 	lastTaken time.Time
 }
 
@@ -55,10 +54,10 @@ func (f *MemPoolFilter) CollectMessage(ctx context.Context, msg *types.SignedMes
 		copy(f.collected, f.collected[1:])
 		f.collected = f.collected[:len(f.collected)-1]
 	}
-	f.collected = append(f.collected, msg.Cid())
+	f.collected = append(f.collected, msg)
 }
 
-func (f *MemPoolFilter) TakeCollectedMessages(context.Context) []cid.Cid {
+func (f *MemPoolFilter) TakeCollectedMessages(context.Context) []*types.SignedMessage {
 	f.mu.Lock()
 	collected := f.collected
 	f.collected = nil

chain/types/ethtypes/eth_transactions.go

@@ -208,6 +208,36 @@ func (tx *EthTxArgs) ToRlpUnsignedMsg() ([]byte, error) {
 	return append([]byte{0x02}, encoded...), nil
 }
 
+func (tx *EthTx) ToEthTxArgs() EthTxArgs {
+	return EthTxArgs{
+		ChainID:              int(tx.ChainID),
+		Nonce:                int(tx.Nonce),
+		To:                   tx.To,
+		Value:                big.Int(tx.Value),
+		MaxFeePerGas:         big.Int(tx.MaxFeePerGas),
+		MaxPriorityFeePerGas: big.Int(tx.MaxPriorityFeePerGas),
+		GasLimit:             int(tx.Gas),
+		Input:                tx.Input,
+		V:                    big.Int(tx.V),
+		R:                    big.Int(tx.R),
+		S:                    big.Int(tx.S),
+	}
+}
+
+func (tx *EthTx) TxHash() (EthHash, error) {
+	ethTxArgs := tx.ToEthTxArgs()
+	return (&ethTxArgs).TxHash()
+}
+
+func (tx *EthTxArgs) TxHash() (EthHash, error) {
+	rlp, err := tx.ToRlpSignedMsg()
+	if err != nil {
+		return EmptyEthHash, err
+	}
+
+	return EthHashFromTxBytes(rlp), nil
+}
+
 func (tx *EthTxArgs) ToRlpSignedMsg() ([]byte, error) {
 	packed1, err := tx.packTxFields()
 	if err != nil {

chain/types/ethtypes/eth_types.go

@@ -409,10 +409,21 @@ func ParseEthHash(s string) (EthHash, error) {
 	return h, nil
 }
 
+func EthHashFromTxBytes(b []byte) EthHash {
+	hasher := sha3.NewLegacyKeccak256()
+	hasher.Write(b)
+	hash := hasher.Sum(nil)
+
+	var ethHash EthHash
+	copy(ethHash[:], hash)
+	return ethHash
+}
+
 func (h EthHash) String() string {
 	return "0x" + hex.EncodeToString(h[:])
 }
 
+// Should ONLY be used for blocks and Filecoin messages. Eth transactions expect a different hashing scheme.
 func (h EthHash) ToCid() cid.Cid {
 	// err is always nil
 	mh, _ := multihash.EncodeName(h[:], "blake2b-256")
@@ -597,7 +608,7 @@ type EthLog struct {
 	// The index corresponds to the sequence of messages produced by ChainGetParentMessages
 	TransactionIndex EthUint64 `json:"transactionIndex"`
 
-	// TransactionHash is the cid of the message that produced the event log.
+	// TransactionHash is the hash of the RLP message that produced the event log.
 	TransactionHash EthHash `json:"transactionHash"`
 
 	// BlockHash is the hash of the tipset containing the message that produced the log.

documentation/en/api-v1-unstable-methods.md

@@ -88,6 +88,7 @@
   * [EthGetTransactionByBlockNumberAndIndex](#EthGetTransactionByBlockNumberAndIndex)
   * [EthGetTransactionByHash](#EthGetTransactionByHash)
   * [EthGetTransactionCount](#EthGetTransactionCount)
+  * [EthGetTransactionHashByCid](#EthGetTransactionHashByCid)
   * [EthGetTransactionReceipt](#EthGetTransactionReceipt)
   * [EthMaxPriorityFeePerGas](#EthMaxPriorityFeePerGas)
   * [EthNewBlockFilter](#EthNewBlockFilter)
@@ -2778,6 +2779,22 @@ Inputs:
 
 Response: `"0x5"`
 
+### EthGetTransactionHashByCid
+
+
+Perms: read
+
+Inputs:
+```json
+[
+  {
+    "/": "bafy2bzacea3wsdh6y3a36tb3skempjoxqpuyompjbmfeyf34fi3uy6uue42v4"
+  }
+]
+```
+
+Response: `"0x37690cfec6c1bf4c3b9288c7a5d783e98731e90b0a4c177c2a374c7a9427355e"`
+
 ### EthGetTransactionReceipt
 
 

documentation/en/default-lotus-config.toml

@@ -342,3 +342,11 @@
   #ActorEventDatabasePath = ""
 
 
+[EthTxHashConfig]
+  # EnableEthHashToFilecoinCidMapping enables storing a mapping of eth transaction hashes to filecoin message Cids
+  #
+  # type: bool
+  # env var: LOTUS_ETHTXHASHCONFIG_ENABLEETHHASHTOFILECOINCIDMAPPING
+  #EnableEthHashToFilecoinCidMapping = true
+
+

itests/eth_deploy_test.go

@@ -41,6 +41,7 @@ func TestDeployment(t *testing.T) {
 			cfg.ActorEvent.EnableRealTimeFilterAPI = true
 			return nil
 		}),
+		kit.EthTxHashLookup(),
 	)
 	ens.InterconnectAll().BeginMining(blockTime)