diff --git a/node/builder_chain.go b/node/builder_chain.go index 6f9cda9b356..fcdb26162a7 100644 --- a/node/builder_chain.go +++ b/node/builder_chain.go @@ -155,7 +155,6 @@ var ChainNode = Options( Override(new(stmgr.StateManagerAPI), rpcstmgr.NewRPCStateManager), Override(new(full.EthModuleAPI), From(new(api.Gateway))), Override(new(full.EthEventAPI), From(new(api.Gateway))), - Override(new(full.EthTraceAPI), From(new(api.Gateway))), ), // Full node API / service startup @@ -271,12 +270,10 @@ func ConfigFullNode(c interface{}) Option { If(cfg.Fevm.EnableEthRPC, Override(new(full.EthModuleAPI), modules.EthModuleAPI(cfg.Fevm)), Override(new(full.EthEventAPI), modules.EthEventAPI(cfg.Fevm)), - Override(new(full.EthTraceAPI), modules.EthTraceAPI()), ), If(!cfg.Fevm.EnableEthRPC, Override(new(full.EthModuleAPI), &full.EthModuleDummy{}), Override(new(full.EthEventAPI), &full.EthModuleDummy{}), - Override(new(full.EthTraceAPI), &full.EthModuleDummy{}), ), ), diff --git a/node/impl/full.go b/node/impl/full.go index 0f87cfe292c..affcc960e09 100644 --- a/node/impl/full.go +++ b/node/impl/full.go @@ -36,7 +36,6 @@ type FullNodeAPI struct { full.SyncAPI full.RaftAPI full.EthAPI - full.EthTraceAPI DS dtypes.MetadataDS NetworkName dtypes.NetworkName diff --git a/node/impl/full/dummy.go b/node/impl/full/dummy.go index 7412ed7549b..c06e5c084e5 100644 --- a/node/impl/full/dummy.go +++ b/node/impl/full/dummy.go @@ -188,4 +188,3 @@ func (e *EthModuleDummy) TraceReplayBlockTransactions(ctx context.Context, blkNu var _ EthModuleAPI = &EthModuleDummy{} var _ EthEventAPI = &EthModuleDummy{} -var _ EthTraceAPI = &EthModuleDummy{} diff --git a/node/impl/full/eth.go b/node/impl/full/eth.go index 424756f8140..a2f406b7c2d 100644 --- a/node/impl/full/eth.go +++ b/node/impl/full/eth.go @@ -3,20 +3,17 @@ package full import ( "bytes" "context" - "encoding/json" + "encoding/hex" "errors" "fmt" "os" "sort" "strconv" "strings" - "sync" "time" - "github.com/google/uuid" "github.com/ipfs/go-cid" cbg "github.com/whyrusleeping/cbor-gen" - "github.com/zyedidia/generic/queue" "go.uber.org/fx" "golang.org/x/xerrors" @@ -25,9 +22,7 @@ import ( "github.com/filecoin-project/go-state-types/abi" "github.com/filecoin-project/go-state-types/big" builtintypes "github.com/filecoin-project/go-state-types/builtin" - "github.com/filecoin-project/go-state-types/builtin/v10/eam" "github.com/filecoin-project/go-state-types/builtin/v10/evm" - "github.com/filecoin-project/go-state-types/crypto" "github.com/filecoin-project/go-state-types/exitcode" "github.com/filecoin-project/lotus/api" @@ -42,7 +37,6 @@ import ( "github.com/filecoin-project/lotus/chain/store" "github.com/filecoin-project/lotus/chain/types" "github.com/filecoin-project/lotus/chain/types/ethtypes" - "github.com/filecoin-project/lotus/chain/vm" "github.com/filecoin-project/lotus/node/modules/dtypes" ) @@ -77,6 +71,8 @@ type EthModuleAPI interface { EthMaxPriorityFeePerGas(ctx context.Context) (ethtypes.EthBigInt, error) EthSendRawTransaction(ctx context.Context, rawTx ethtypes.EthBytes) (ethtypes.EthHash, error) Web3ClientVersion(ctx context.Context) (string, error) + TraceBlock(ctx context.Context, blkNum string) (interface{}, error) + TraceReplayBlockTransactions(ctx context.Context, blkNum string, traceTypes []string) (interface{}, error) } type EthEventAPI interface { @@ -241,101 +237,8 @@ func (a *EthModule) EthGetBlockByHash(ctx context.Context, blkHash ethtypes.EthH return newEthBlockFromFilecoinTipSet(ctx, ts, fullTxInfo, a.Chain, a.StateAPI) } -func (a *EthModule) getTipsetByEthBlockNumberOrHash(ctx context.Context, blkParam ethtypes.EthBlockNumberOrHash) (*types.TipSet, error) { - head := a.Chain.GetHeaviestTipSet() - - predefined := blkParam.PredefinedBlock - if predefined != nil { - if *predefined == "earliest" { - return nil, fmt.Errorf("block param \"earliest\" is not supported") - } else if *predefined == "pending" { - return head, nil - } else if *predefined == "latest" { - parent, err := a.Chain.GetTipSetFromKey(ctx, head.Parents()) - if err != nil { - return nil, fmt.Errorf("cannot get parent tipset") - } - return parent, nil - } else { - return nil, fmt.Errorf("unknown predefined block %s", *predefined) - } - } - - if blkParam.BlockNumber != nil { - height := abi.ChainEpoch(*blkParam.BlockNumber) - if height > head.Height()-1 { - return nil, fmt.Errorf("requested a future epoch (beyond 'latest')") - } - ts, err := a.ChainAPI.ChainGetTipSetByHeight(ctx, height, head.Key()) - if err != nil { - return nil, fmt.Errorf("cannot get tipset at height: %v", height) - } - return ts, nil - } - - if blkParam.BlockHash != nil { - ts, err := a.Chain.GetTipSetByCid(ctx, blkParam.BlockHash.ToCid()) - if err != nil { - return nil, fmt.Errorf("cannot get tipset by hash: %v", err) - } - - // verify that the tipset is in the canonical chain - if blkParam.RequireCanonical { - // walk up the current chain (our head) until we reach ts.Height() - walkTs, err := a.ChainAPI.ChainGetTipSetByHeight(ctx, ts.Height(), head.Key()) - if err != nil { - return nil, fmt.Errorf("cannot get tipset at height: %v", ts.Height()) - } - - // verify that it equals the expected tipset - if !walkTs.Equals(ts) { - return nil, fmt.Errorf("tipset is not canonical") - } - } - - return ts, nil - } - - return nil, errors.New("invalid block param") -} - -func (a *EthModule) parseBlkParam(ctx context.Context, blkParam string, strict bool) (*types.TipSet, error) { - if blkParam == "earliest" { - return nil, fmt.Errorf("block param \"earliest\" is not supported") - } - - head := a.Chain.GetHeaviestTipSet() - switch blkParam { - case "pending": - return head, nil - case "latest": - parent, err := a.Chain.GetTipSetFromKey(ctx, head.Parents()) - if err != nil { - return nil, fmt.Errorf("cannot get parent tipset") - } - return parent, nil - default: - var num ethtypes.EthUint64 - err := num.UnmarshalJSON([]byte(`"` + blkParam + `"`)) - if err != nil { - return nil, fmt.Errorf("cannot parse block number: %v", err) - } - if abi.ChainEpoch(num) > head.Height()-1 { - return nil, fmt.Errorf("requested a future epoch (beyond 'latest')") - } - ts, err := a.ChainAPI.ChainGetTipSetByHeight(ctx, abi.ChainEpoch(num), head.Key()) - if err != nil { - return nil, fmt.Errorf("cannot get tipset at height: %v", num) - } - if strict && ts.Height() != abi.ChainEpoch(num) { - return nil, ErrNullRound - } - return ts, nil - } -} - func (a *EthModule) EthGetBlockByNumber(ctx context.Context, blkParam string, fullTxInfo bool) (ethtypes.EthBlock, error) { - ts, err := a.parseBlkParam(ctx, blkParam, true) + ts, err := getTipsetByBlockNr(ctx, a.Chain, blkParam, true) if err != nil { return ethtypes.EthBlock{}, err } @@ -431,7 +334,7 @@ func (a *EthModule) EthGetMessageCidByTransactionHash(ctx context.Context, txHas } func (a *EthModule) EthGetTransactionHashByCid(ctx context.Context, cid cid.Cid) (*ethtypes.EthHash, error) { - hash, err := EthTxHashFromMessageCid(ctx, cid, a.StateAPI) + hash, err := ethTxHashFromMessageCid(ctx, cid, a.StateAPI) if hash == ethtypes.EmptyEthHash { // not found return nil, nil @@ -446,7 +349,7 @@ func (a *EthModule) EthGetTransactionCount(ctx context.Context, sender ethtypes. return ethtypes.EthUint64(0), nil } - ts, err := a.getTipsetByEthBlockNumberOrHash(ctx, blkParam) + ts, err := getTipsetByEthBlockNumberOrHash(ctx, a.Chain, blkParam) if err != nil { return ethtypes.EthUint64(0), xerrors.Errorf("failed to process block param: %v; %w", blkParam, err) } @@ -534,7 +437,7 @@ func (a *EthModule) EthGetCode(ctx context.Context, ethAddr ethtypes.EthAddress, return nil, xerrors.Errorf("cannot get Filecoin address: %w", err) } - ts, err := a.getTipsetByEthBlockNumberOrHash(ctx, blkParam) + ts, err := getTipsetByEthBlockNumberOrHash(ctx, a.Chain, blkParam) if err != nil { return nil, xerrors.Errorf("failed to process block param: %v; %w", blkParam, err) } @@ -613,7 +516,7 @@ func (a *EthModule) EthGetCode(ctx context.Context, ethAddr ethtypes.EthAddress, } func (a *EthModule) EthGetStorageAt(ctx context.Context, ethAddr ethtypes.EthAddress, position ethtypes.EthBytes, blkParam ethtypes.EthBlockNumberOrHash) (ethtypes.EthBytes, error) { - ts, err := a.getTipsetByEthBlockNumberOrHash(ctx, blkParam) + ts, err := getTipsetByEthBlockNumberOrHash(ctx, a.Chain, blkParam) if err != nil { return nil, xerrors.Errorf("failed to process block param: %v; %w", blkParam, err) } @@ -709,7 +612,7 @@ func (a *EthModule) EthGetBalance(ctx context.Context, address ethtypes.EthAddre return ethtypes.EthBigInt{}, err } - ts, err := a.getTipsetByEthBlockNumberOrHash(ctx, blkParam) + ts, err := getTipsetByEthBlockNumberOrHash(ctx, a.Chain, blkParam) if err != nil { return ethtypes.EthBigInt{}, xerrors.Errorf("failed to process block param: %v; %w", blkParam, err) } @@ -790,7 +693,7 @@ func (a *EthModule) EthFeeHistory(ctx context.Context, p jsonrpc.RawParams) (eth } } - ts, err := a.parseBlkParam(ctx, params.NewestBlkNum, false) + ts, err := getTipsetByBlockNr(ctx, a.Chain, params.NewestBlkNum, false) if err != nil { return ethtypes.EthFeeHistory{}, fmt.Errorf("bad block parameter %s: %s", params.NewestBlkNum, err) } @@ -922,62 +825,135 @@ func (a *EthModule) Web3ClientVersion(ctx context.Context) (string, error) { return build.UserVersion(), nil } -func (a *EthModule) ethCallToFilecoinMessage(ctx context.Context, tx ethtypes.EthCall) (*types.Message, error) { - var from address.Address - if tx.From == nil || *tx.From == (ethtypes.EthAddress{}) { - // Send from the filecoin "system" address. - var err error - from, err = (ethtypes.EthAddress{}).ToFilecoinAddress() +func (e *EthModule) TraceBlock(ctx context.Context, blkNum string) (interface{}, error) { + ts, err := getTipsetByBlockNr(ctx, e.Chain, blkNum, false) + if err != nil { + return nil, err + } + + _, trace, err := e.StateManager.ExecutionTrace(ctx, ts) + if err != nil { + return nil, xerrors.Errorf("failed to compute base state: %w", err) + } + + tsParent, err := e.ChainAPI.ChainGetTipSetByHeight(ctx, ts.Height()+1, e.Chain.GetHeaviestTipSet().Key()) + if err != nil { + return nil, fmt.Errorf("cannot get tipset at height: %v", ts.Height()+1) + } + + msgs, err := e.ChainGetParentMessages(ctx, tsParent.Blocks()[0].Cid()) + if err != nil { + return nil, err + } + + cid, err := ts.Key().Cid() + if err != nil { + return nil, err + } + + blkHash, err := ethtypes.EthHashFromCid(cid) + if err != nil { + return nil, err + } + + allTraces := make([]*TraceBlock, 0, len(trace)) + for _, ir := range trace { + // ignore messages from f00 + if ir.Msg.From.String() == "f00" { + continue + } + + idx := -1 + for msgIdx, msg := range msgs { + if ir.Msg.From == msg.Message.From { + idx = msgIdx + break + } + } + if idx == -1 { + log.Warnf("cannot resolve message index for cid: %s", ir.MsgCid) + continue + } + + txHash, err := e.EthGetTransactionHashByCid(ctx, ir.MsgCid) if err != nil { - return nil, fmt.Errorf("failed to construct the ethereum system address: %w", err) + return nil, err } - } else { - // The from address must be translatable to an f4 address. - var err error - from, err = tx.From.ToFilecoinAddress() + if txHash == nil { + log.Warnf("cannot find transaction hash for cid %s", ir.MsgCid) + continue + } + + traces := []*Trace{} + err = buildTraces(&traces, nil, []int{}, ir.ExecutionTrace, int64(ts.Height())) if err != nil { - return nil, fmt.Errorf("failed to translate sender address (%s): %w", tx.From.String(), err) + return nil, xerrors.Errorf("failed when building traces: %w", err) } - if p := from.Protocol(); p != address.Delegated { - return nil, fmt.Errorf("expected a class 4 address, got: %d: %w", p, err) + + traceBlocks := make([]*TraceBlock, 0, len(trace)) + for _, trace := range traces { + traceBlocks = append(traceBlocks, &TraceBlock{ + Trace: trace, + BlockHash: blkHash, + BlockNumber: int64(ts.Height()), + TransactionHash: *txHash, + TransactionPosition: idx, + }) } + + allTraces = append(allTraces, traceBlocks...) + } + + return allTraces, nil +} + +func (e *EthModule) TraceReplayBlockTransactions(ctx context.Context, blkNum string, traceTypes []string) (interface{}, error) { + if len(traceTypes) != 1 || traceTypes[0] != "trace" { + return nil, fmt.Errorf("only 'trace' is supported") + } + + ts, err := getTipsetByBlockNr(ctx, e.Chain, blkNum, false) + if err != nil { + return nil, err + } + + _, trace, err := e.StateManager.ExecutionTrace(ctx, ts) + if err != nil { + return nil, xerrors.Errorf("failed when calling ExecutionTrace: %w", err) } - var params []byte - if len(tx.Data) > 0 { - initcode := abi.CborBytes(tx.Data) - params2, err := actors.SerializeParams(&initcode) + allTraces := make([]*TraceReplayBlockTransaction, 0, len(trace)) + for _, ir := range trace { + // ignore messages from f00 + if ir.Msg.From.String() == "f00" { + continue + } + + txHash, err := e.EthGetTransactionHashByCid(ctx, ir.MsgCid) if err != nil { - return nil, fmt.Errorf("failed to serialize params: %w", err) + return nil, err + } + if txHash == nil { + log.Warnf("cannot find transaction hash for cid %s", ir.MsgCid) + continue } - params = params2 - } - var to address.Address - var method abi.MethodNum - if tx.To == nil { - // this is a contract creation - to = builtintypes.EthereumAddressManagerActorAddr - method = builtintypes.MethodsEAM.CreateExternal - } else { - addr, err := tx.To.ToFilecoinAddress() + t := TraceReplayBlockTransaction{ + Output: hex.EncodeToString(ir.MsgRct.Return), + TransactionHash: *txHash, + StateDiff: nil, + VmTrace: nil, + } + + err = buildTraces(&t.Trace, nil, []int{}, ir.ExecutionTrace, int64(ts.Height())) if err != nil { - return nil, xerrors.Errorf("cannot get Filecoin address: %w", err) + return nil, xerrors.Errorf("failed when building traces: %w", err) } - to = addr - method = builtintypes.MethodsEVM.InvokeContract + + allTraces = append(allTraces, &t) } - return &types.Message{ - From: from, - To: to, - Value: big.Int(tx.Value), - Method: method, - Params: params, - GasLimit: build.BlockGasLimit, - GasFeeCap: big.Zero(), - GasPremium: big.Zero(), - }, nil + return allTraces, nil } func (a *EthModule) applyMessage(ctx context.Context, msg *types.Message, tsk types.TipSetKey) (res *api.InvocResult, err error) { @@ -1013,7 +989,7 @@ func (a *EthModule) applyMessage(ctx context.Context, msg *types.Message, tsk ty } func (a *EthModule) EthEstimateGas(ctx context.Context, tx ethtypes.EthCall) (ethtypes.EthUint64, error) { - msg, err := a.ethCallToFilecoinMessage(ctx, tx) + msg, err := ethCallToFilecoinMessage(ctx, tx) if err != nil { return ethtypes.EthUint64(0), err } @@ -1171,12 +1147,12 @@ func ethGasSearch( } func (a *EthModule) EthCall(ctx context.Context, tx ethtypes.EthCall, blkParam ethtypes.EthBlockNumberOrHash) (ethtypes.EthBytes, error) { - msg, err := a.ethCallToFilecoinMessage(ctx, tx) + msg, err := ethCallToFilecoinMessage(ctx, tx) if err != nil { return nil, xerrors.Errorf("failed to convert ethcall to filecoin message: %w", err) } - ts, err := a.getTipsetByEthBlockNumberOrHash(ctx, blkParam) + ts, err := getTipsetByEthBlockNumberOrHash(ctx, a.Chain, blkParam) if err != nil { return nil, xerrors.Errorf("failed to process block param: %v; %w", blkParam, err) } @@ -1577,1027 +1553,37 @@ func (e *EthEvent) GC(ctx context.Context, ttl time.Duration) { } } -type filterEventCollector interface { - TakeCollectedEvents(context.Context) []*filter.CollectedEvent -} - -type filterMessageCollector interface { - TakeCollectedMessages(context.Context) []*types.SignedMessage -} - -type filterTipSetCollector interface { - TakeCollectedTipSets(context.Context) []types.TipSetKey -} - -func ethLogFromEvent(entries []types.EventEntry) (data []byte, topics []ethtypes.EthHash, ok bool) { - var ( - topicsFound [4]bool - topicsFoundCount int - dataFound bool - ) - // Topics must be non-nil, even if empty. So we might as well pre-allocate for 4 (the max). - topics = make([]ethtypes.EthHash, 0, 4) - for _, entry := range entries { - // Drop events with non-raw topics to avoid mistakes. - if entry.Codec != cid.Raw { - log.Warnw("did not expect an event entry with a non-raw codec", "codec", entry.Codec, "key", entry.Key) - return nil, nil, false - } - // Check if the key is t1..t4 - if len(entry.Key) == 2 && "t1" <= entry.Key && entry.Key <= "t4" { - // '1' - '1' == 0, etc. - idx := int(entry.Key[1] - '1') - - // Drop events with mis-sized topics. - if len(entry.Value) != 32 { - log.Warnw("got an EVM event topic with an invalid size", "key", entry.Key, "size", len(entry.Value)) - return nil, nil, false - } - - // Drop events with duplicate topics. - if topicsFound[idx] { - log.Warnw("got a duplicate EVM event topic", "key", entry.Key) - return nil, nil, false - } - topicsFound[idx] = true - topicsFoundCount++ - - // Extend the topics array - for len(topics) <= idx { - topics = append(topics, ethtypes.EthHash{}) - } - copy(topics[idx][:], entry.Value) - } else if entry.Key == "d" { - // Drop events with duplicate data fields. - if dataFound { - log.Warnw("got duplicate EVM event data") - return nil, nil, false - } - - dataFound = true - data = entry.Value - } else { - // Skip entries we don't understand (makes it easier to extend things). - // But we warn for now because we don't expect them. - log.Warnw("unexpected event entry", "key", entry.Key) - } - - } - - // Drop events with skipped topics. - if len(topics) != topicsFoundCount { - log.Warnw("EVM event topic length mismatch", "expected", len(topics), "actual", topicsFoundCount) - return nil, nil, false - } - return data, topics, true -} - -func ethFilterResultFromEvents(evs []*filter.CollectedEvent, sa StateAPI) (*ethtypes.EthFilterResult, error) { - res := ðtypes.EthFilterResult{} - for _, ev := range evs { - log := ethtypes.EthLog{ - Removed: ev.Reverted, - LogIndex: ethtypes.EthUint64(ev.EventIdx), - TransactionIndex: ethtypes.EthUint64(ev.MsgIdx), - BlockNumber: ethtypes.EthUint64(ev.Height), - } - var ( - err error - ok bool - ) - - log.Data, log.Topics, ok = ethLogFromEvent(ev.Entries) - if !ok { - continue - } - - log.Address, err = ethtypes.EthAddressFromFilecoinAddress(ev.EmitterAddr) - if err != nil { - return nil, err - } - - log.TransactionHash, err = EthTxHashFromMessageCid(context.TODO(), ev.MsgCid, sa) - if err != nil { - return nil, err - } - c, err := ev.TipSetKey.Cid() - if err != nil { - return nil, err - } - log.BlockHash, err = ethtypes.EthHashFromCid(c) - if err != nil { - return nil, err - } - - res.Results = append(res.Results, log) - } - - return res, nil -} - -func ethFilterResultFromTipSets(tsks []types.TipSetKey) (*ethtypes.EthFilterResult, error) { - res := ðtypes.EthFilterResult{} - - for _, tsk := range tsks { - c, err := tsk.Cid() - if err != nil { - return nil, err - } - hash, err := ethtypes.EthHashFromCid(c) - if err != nil { - return nil, err - } - - res.Results = append(res.Results, hash) - } - - return res, nil -} - -func ethFilterResultFromMessages(cs []*types.SignedMessage, sa StateAPI) (*ethtypes.EthFilterResult, error) { - res := ðtypes.EthFilterResult{} - - for _, c := range cs { - hash, err := EthTxHashFromSignedMessage(context.TODO(), c, sa) - if err != nil { - return nil, err - } - - res.Results = append(res.Results, hash) - } - - return res, nil -} - -type EthSubscriptionManager struct { - Chain *store.ChainStore - StateAPI StateAPI - ChainAPI ChainAPI - mu sync.Mutex - subs map[ethtypes.EthSubscriptionID]*ethSubscription -} - -func (e *EthSubscriptionManager) StartSubscription(ctx context.Context, out ethSubscriptionCallback, dropFilter func(context.Context, filter.Filter) error) (*ethSubscription, error) { // nolint - rawid, err := uuid.NewRandom() - if err != nil { - return nil, xerrors.Errorf("new uuid: %w", err) - } - id := ethtypes.EthSubscriptionID{} - copy(id[:], rawid[:]) // uuid is 16 bytes - - ctx, quit := context.WithCancel(ctx) - - sub := ðSubscription{ - Chain: e.Chain, - StateAPI: e.StateAPI, - ChainAPI: e.ChainAPI, - uninstallFilter: dropFilter, - id: id, - in: make(chan interface{}, 200), - out: out, - quit: quit, - - toSend: queue.New[[]byte](), - sendCond: make(chan struct{}, 1), +func calculateRewardsAndGasUsed(rewardPercentiles []float64, txGasRewards gasRewardSorter) ([]ethtypes.EthBigInt, int64) { + var gasUsedTotal int64 + for _, tx := range txGasRewards { + gasUsedTotal += tx.gasUsed } - e.mu.Lock() - if e.subs == nil { - e.subs = make(map[ethtypes.EthSubscriptionID]*ethSubscription) + rewards := make([]ethtypes.EthBigInt, len(rewardPercentiles)) + for i := range rewards { + rewards[i] = ethtypes.EthBigInt(types.NewInt(MinGasPremium)) } - e.subs[sub.id] = sub - e.mu.Unlock() - - go sub.start(ctx) - go sub.startOut(ctx) - - return sub, nil -} - -func (e *EthSubscriptionManager) StopSubscription(ctx context.Context, id ethtypes.EthSubscriptionID) error { - e.mu.Lock() - defer e.mu.Unlock() - sub, ok := e.subs[id] - if !ok { - return xerrors.Errorf("subscription not found") + if len(txGasRewards) == 0 { + return rewards, gasUsedTotal } - sub.stop() - delete(e.subs, id) - - return nil -} - -type ethSubscriptionCallback func(context.Context, jsonrpc.RawParams) error - -const maxSendQueue = 20000 - -type ethSubscription struct { - Chain *store.ChainStore - StateAPI StateAPI - ChainAPI ChainAPI - uninstallFilter func(context.Context, filter.Filter) error - id ethtypes.EthSubscriptionID - in chan interface{} - out ethSubscriptionCallback - - mu sync.Mutex - filters []filter.Filter - quit func() - - sendLk sync.Mutex - sendQueueLen int - toSend *queue.Queue[[]byte] - sendCond chan struct{} -} - -func (e *ethSubscription) addFilter(ctx context.Context, f filter.Filter) { - e.mu.Lock() - defer e.mu.Unlock() - - f.SetSubChannel(e.in) - e.filters = append(e.filters, f) -} -// sendOut processes the final subscription queue. It's here in case the subscriber -// is slow, and we need to buffer the messages. -func (e *ethSubscription) startOut(ctx context.Context) { - for { - select { - case <-ctx.Done(): - return - case <-e.sendCond: - e.sendLk.Lock() - - for !e.toSend.Empty() { - front := e.toSend.Dequeue() - e.sendQueueLen-- - - e.sendLk.Unlock() - - if err := e.out(ctx, front); err != nil { - log.Warnw("error sending subscription response, killing subscription", "sub", e.id, "error", err) - e.stop() - return - } - - e.sendLk.Lock() - } + sort.Stable(txGasRewards) - e.sendLk.Unlock() + var idx int + var sum int64 + for i, percentile := range rewardPercentiles { + threshold := int64(float64(gasUsedTotal) * percentile / 100) + for sum < threshold && idx < len(txGasRewards)-1 { + sum += txGasRewards[idx].gasUsed + idx++ } - } -} - -func (e *ethSubscription) send(ctx context.Context, v interface{}) { - resp := ethtypes.EthSubscriptionResponse{ - SubscriptionID: e.id, - Result: v, - } - - outParam, err := json.Marshal(resp) - if err != nil { - log.Warnw("marshaling subscription response", "sub", e.id, "error", err) - return - } - - e.sendLk.Lock() - defer e.sendLk.Unlock() - - e.toSend.Enqueue(outParam) - - e.sendQueueLen++ - if e.sendQueueLen > maxSendQueue { - log.Warnw("subscription send queue full, killing subscription", "sub", e.id) - e.stop() - return - } - - select { - case e.sendCond <- struct{}{}: - default: // already signalled, and we're holding the lock so we know that the event will be processed - } -} - -func (e *ethSubscription) start(ctx context.Context) { - for { - select { - case <-ctx.Done(): - return - case v := <-e.in: - switch vt := v.(type) { - case *filter.CollectedEvent: - evs, err := ethFilterResultFromEvents([]*filter.CollectedEvent{vt}, e.StateAPI) - if err != nil { - continue - } - - for _, r := range evs.Results { - e.send(ctx, r) - } - case *types.TipSet: - ev, err := newEthBlockFromFilecoinTipSet(ctx, vt, true, e.Chain, e.StateAPI) - if err != nil { - break - } - - e.send(ctx, ev) - case *types.SignedMessage: // mpool txid - evs, err := ethFilterResultFromMessages([]*types.SignedMessage{vt}, e.StateAPI) - if err != nil { - continue - } - - for _, r := range evs.Results { - e.send(ctx, r) - } - default: - log.Warnf("unexpected subscription value type: %T", vt) - } - } - } -} - -func (e *ethSubscription) stop() { - e.mu.Lock() - if e.quit == nil { - e.mu.Unlock() - return - } - - if e.quit != nil { - e.quit() - e.quit = nil - e.mu.Unlock() - - for _, f := range e.filters { - // note: the context in actually unused in uninstallFilter - if err := e.uninstallFilter(context.TODO(), f); err != nil { - // this will leave the filter a zombie, collecting events up to the maximum allowed - log.Warnf("failed to remove filter when unsubscribing: %v", err) - } - } - } -} - -func newEthBlockFromFilecoinTipSet(ctx context.Context, ts *types.TipSet, fullTxInfo bool, cs *store.ChainStore, sa StateAPI) (ethtypes.EthBlock, error) { - parentKeyCid, err := ts.Parents().Cid() - if err != nil { - return ethtypes.EthBlock{}, err - } - parentBlkHash, err := ethtypes.EthHashFromCid(parentKeyCid) - if err != nil { - return ethtypes.EthBlock{}, err - } - - bn := ethtypes.EthUint64(ts.Height()) - - blkCid, err := ts.Key().Cid() - if err != nil { - return ethtypes.EthBlock{}, err - } - blkHash, err := ethtypes.EthHashFromCid(blkCid) - if err != nil { - return ethtypes.EthBlock{}, err - } - - msgs, rcpts, err := messagesAndReceipts(ctx, ts, cs, sa) - if err != nil { - return ethtypes.EthBlock{}, xerrors.Errorf("failed to retrieve messages and receipts: %w", err) - } - - block := ethtypes.NewEthBlock(len(msgs) > 0) - - gasUsed := int64(0) - for i, msg := range msgs { - rcpt := rcpts[i] - ti := ethtypes.EthUint64(i) - gasUsed += rcpt.GasUsed - var smsg *types.SignedMessage - switch msg := msg.(type) { - case *types.SignedMessage: - smsg = msg - case *types.Message: - smsg = &types.SignedMessage{ - Message: *msg, - Signature: crypto.Signature{ - Type: crypto.SigTypeBLS, - }, - } - default: - return ethtypes.EthBlock{}, xerrors.Errorf("failed to get signed msg %s: %w", msg.Cid(), err) - } - tx, err := newEthTxFromSignedMessage(ctx, smsg, sa) - if err != nil { - return ethtypes.EthBlock{}, xerrors.Errorf("failed to convert msg to ethTx: %w", err) - } - - tx.ChainID = ethtypes.EthUint64(build.Eip155ChainId) - tx.BlockHash = &blkHash - tx.BlockNumber = &bn - tx.TransactionIndex = &ti - - if fullTxInfo { - block.Transactions = append(block.Transactions, tx) - } else { - block.Transactions = append(block.Transactions, tx.Hash.String()) - } - } - - block.Hash = blkHash - block.Number = bn - block.ParentHash = parentBlkHash - block.Timestamp = ethtypes.EthUint64(ts.Blocks()[0].Timestamp) - block.BaseFeePerGas = ethtypes.EthBigInt{Int: ts.Blocks()[0].ParentBaseFee.Int} - block.GasUsed = ethtypes.EthUint64(gasUsed) - return block, nil -} - -func messagesAndReceipts(ctx context.Context, ts *types.TipSet, cs *store.ChainStore, sa StateAPI) ([]types.ChainMsg, []types.MessageReceipt, error) { - msgs, err := cs.MessagesForTipset(ctx, ts) - if err != nil { - return nil, nil, xerrors.Errorf("error loading messages for tipset: %v: %w", ts, err) - } - - _, rcptRoot, err := sa.StateManager.TipSetState(ctx, ts) - if err != nil { - return nil, nil, xerrors.Errorf("failed to compute state: %w", err) - } - - rcpts, err := cs.ReadReceipts(ctx, rcptRoot) - if err != nil { - return nil, nil, xerrors.Errorf("error loading receipts for tipset: %v: %w", ts, err) - } - - if len(msgs) != len(rcpts) { - return nil, nil, xerrors.Errorf("receipts and message array lengths didn't match for tipset: %v: %w", ts, err) - } - - return msgs, rcpts, nil -} - -// lookupEthAddress makes its best effort at finding the Ethereum address for a -// Filecoin address. It does the following: -// -// 1. If the supplied address is an f410 address, we return its payload as the EthAddress. -// 2. Otherwise (f0, f1, f2, f3), we look up the actor on the state tree. If it has a delegated address, we return it if it's f410 address. -// 3. Otherwise, we fall back to returning a masked ID Ethereum address. If the supplied address is an f0 address, we -// use that ID to form the masked ID address. -// 4. Otherwise, we fetch the actor's ID from the state tree and form the masked ID with it. -func lookupEthAddress(ctx context.Context, addr address.Address, sa StateAPI) (ethtypes.EthAddress, error) { - // BLOCK A: We are trying to get an actual Ethereum address from an f410 address. - // Attempt to convert directly, if it's an f4 address. - ethAddr, err := ethtypes.EthAddressFromFilecoinAddress(addr) - if err == nil && !ethAddr.IsMaskedID() { - return ethAddr, nil - } - - // Lookup on the target actor and try to get an f410 address. - if actor, err := sa.StateGetActor(ctx, addr, types.EmptyTSK); err != nil { - return ethtypes.EthAddress{}, err - } else if actor.Address != nil { - if ethAddr, err := ethtypes.EthAddressFromFilecoinAddress(*actor.Address); err == nil && !ethAddr.IsMaskedID() { - return ethAddr, nil - } - } - - // BLOCK B: We gave up on getting an actual Ethereum address and are falling back to a Masked ID address. - // Check if we already have an ID addr, and use it if possible. - if err == nil && ethAddr.IsMaskedID() { - return ethAddr, nil - } - - // Otherwise, resolve the ID addr. - idAddr, err := sa.StateLookupID(ctx, addr, types.EmptyTSK) - if err != nil { - return ethtypes.EthAddress{}, err - } - return ethtypes.EthAddressFromFilecoinAddress(idAddr) -} - -func EthTxHashFromMessageCid(ctx context.Context, c cid.Cid, sa StateAPI) (ethtypes.EthHash, error) { - smsg, err := sa.Chain.GetSignedMessage(ctx, c) - if err == nil { - // This is an Eth Tx, Secp message, Or BLS message in the mpool - return EthTxHashFromSignedMessage(ctx, smsg, sa) - } - - _, err = sa.Chain.GetMessage(ctx, c) - if err == nil { - // This is a BLS message - return ethtypes.EthHashFromCid(c) - } - - return ethtypes.EmptyEthHash, nil -} - -func EthTxHashFromSignedMessage(ctx context.Context, smsg *types.SignedMessage, sa StateAPI) (ethtypes.EthHash, error) { - if smsg.Signature.Type == crypto.SigTypeDelegated { - ethTx, err := newEthTxFromSignedMessage(ctx, smsg, sa) - if err != nil { - return ethtypes.EmptyEthHash, err - } - return ethTx.Hash, nil - } else if smsg.Signature.Type == crypto.SigTypeSecp256k1 { - return ethtypes.EthHashFromCid(smsg.Cid()) - } else { // BLS message - return ethtypes.EthHashFromCid(smsg.Message.Cid()) - } -} - -func newEthTxFromSignedMessage(ctx context.Context, smsg *types.SignedMessage, sa StateAPI) (ethtypes.EthTx, error) { - var tx ethtypes.EthTx - var err error - - // This is an eth tx - if smsg.Signature.Type == crypto.SigTypeDelegated { - tx, err = ethtypes.EthTxFromSignedEthMessage(smsg) - if err != nil { - return ethtypes.EthTx{}, xerrors.Errorf("failed to convert from signed message: %w", err) - } - - tx.Hash, err = tx.TxHash() - if err != nil { - return ethtypes.EthTx{}, xerrors.Errorf("failed to calculate hash for ethTx: %w", err) - } - - fromAddr, err := lookupEthAddress(ctx, smsg.Message.From, sa) - if err != nil { - return ethtypes.EthTx{}, xerrors.Errorf("failed to resolve Ethereum address: %w", err) - } - - tx.From = fromAddr - } else if smsg.Signature.Type == crypto.SigTypeSecp256k1 { // Secp Filecoin Message - tx = ethTxFromNativeMessage(ctx, smsg.VMMessage(), sa) - tx.Hash, err = ethtypes.EthHashFromCid(smsg.Cid()) - if err != nil { - return tx, err - } - } else { // BLS Filecoin message - tx = ethTxFromNativeMessage(ctx, smsg.VMMessage(), sa) - tx.Hash, err = ethtypes.EthHashFromCid(smsg.Message.Cid()) - if err != nil { - return tx, err - } - } - - return tx, nil -} - -// ethTxFromNativeMessage does NOT populate: -// - BlockHash -// - BlockNumber -// - TransactionIndex -// - Hash -func ethTxFromNativeMessage(ctx context.Context, msg *types.Message, sa StateAPI) ethtypes.EthTx { - // We don't care if we error here, conversion is best effort for non-eth transactions - from, _ := lookupEthAddress(ctx, msg.From, sa) - to, _ := lookupEthAddress(ctx, msg.To, sa) - return ethtypes.EthTx{ - To: &to, - From: from, - Nonce: ethtypes.EthUint64(msg.Nonce), - ChainID: ethtypes.EthUint64(build.Eip155ChainId), - Value: ethtypes.EthBigInt(msg.Value), - Type: ethtypes.Eip1559TxType, - Gas: ethtypes.EthUint64(msg.GasLimit), - MaxFeePerGas: ethtypes.EthBigInt(msg.GasFeeCap), - MaxPriorityFeePerGas: ethtypes.EthBigInt(msg.GasPremium), - AccessList: []ethtypes.EthHash{}, - } -} - -// newEthTxFromMessageLookup creates an ethereum transaction from filecoin message lookup. If a negative txIdx is passed -// into the function, it looks up the transaction index of the message in the tipset, otherwise it uses the txIdx passed into the -// function -func newEthTxFromMessageLookup(ctx context.Context, msgLookup *api.MsgLookup, txIdx int, cs *store.ChainStore, sa StateAPI) (ethtypes.EthTx, error) { - ts, err := cs.LoadTipSet(ctx, msgLookup.TipSet) - if err != nil { - return ethtypes.EthTx{}, err - } - - // This tx is located in the parent tipset - parentTs, err := cs.LoadTipSet(ctx, ts.Parents()) - if err != nil { - return ethtypes.EthTx{}, err - } - - parentTsCid, err := parentTs.Key().Cid() - if err != nil { - return ethtypes.EthTx{}, err - } - - // lookup the transactionIndex - if txIdx < 0 { - msgs, err := cs.MessagesForTipset(ctx, parentTs) - if err != nil { - return ethtypes.EthTx{}, err - } - for i, msg := range msgs { - if msg.Cid() == msgLookup.Message { - txIdx = i - break - } - } - if txIdx < 0 { - return ethtypes.EthTx{}, fmt.Errorf("cannot find the msg in the tipset") - } - } - - blkHash, err := ethtypes.EthHashFromCid(parentTsCid) - if err != nil { - return ethtypes.EthTx{}, err - } - - smsg, err := getSignedMessage(ctx, cs, msgLookup.Message) - if err != nil { - return ethtypes.EthTx{}, xerrors.Errorf("failed to get signed msg: %w", err) - } - - tx, err := newEthTxFromSignedMessage(ctx, smsg, sa) - if err != nil { - return ethtypes.EthTx{}, err - } - - var ( - bn = ethtypes.EthUint64(parentTs.Height()) - ti = ethtypes.EthUint64(txIdx) - ) - - tx.ChainID = ethtypes.EthUint64(build.Eip155ChainId) - tx.BlockHash = &blkHash - tx.BlockNumber = &bn - tx.TransactionIndex = &ti - return tx, nil -} - -func newEthTxReceipt(ctx context.Context, tx ethtypes.EthTx, lookup *api.MsgLookup, events []types.Event, cs *store.ChainStore, sa StateAPI) (api.EthTxReceipt, error) { - var ( - transactionIndex ethtypes.EthUint64 - blockHash ethtypes.EthHash - blockNumber ethtypes.EthUint64 - ) - - if tx.TransactionIndex != nil { - transactionIndex = *tx.TransactionIndex - } - if tx.BlockHash != nil { - blockHash = *tx.BlockHash - } - if tx.BlockNumber != nil { - blockNumber = *tx.BlockNumber - } - - receipt := api.EthTxReceipt{ - TransactionHash: tx.Hash, - From: tx.From, - To: tx.To, - TransactionIndex: transactionIndex, - BlockHash: blockHash, - BlockNumber: blockNumber, - Type: ethtypes.EthUint64(2), - Logs: []ethtypes.EthLog{}, // empty log array is compulsory when no logs, or libraries like ethers.js break - LogsBloom: ethtypes.EmptyEthBloom[:], - } - - if lookup.Receipt.ExitCode.IsSuccess() { - receipt.Status = 1 - } else { - receipt.Status = 0 - } - - receipt.GasUsed = ethtypes.EthUint64(lookup.Receipt.GasUsed) - - // TODO: handle CumulativeGasUsed - receipt.CumulativeGasUsed = ethtypes.EmptyEthInt - - // TODO: avoid loading the tipset twice (once here, once when we convert the message to a txn) - ts, err := cs.GetTipSetFromKey(ctx, lookup.TipSet) - if err != nil { - return api.EthTxReceipt{}, xerrors.Errorf("failed to lookup tipset %s when constructing the eth txn receipt: %w", lookup.TipSet, err) - } - - baseFee := ts.Blocks()[0].ParentBaseFee - gasOutputs := vm.ComputeGasOutputs(lookup.Receipt.GasUsed, int64(tx.Gas), baseFee, big.Int(tx.MaxFeePerGas), big.Int(tx.MaxPriorityFeePerGas), true) - totalSpent := big.Sum(gasOutputs.BaseFeeBurn, gasOutputs.MinerTip, gasOutputs.OverEstimationBurn) - - effectiveGasPrice := big.Zero() - if lookup.Receipt.GasUsed > 0 { - effectiveGasPrice = big.Div(totalSpent, big.NewInt(lookup.Receipt.GasUsed)) - } - receipt.EffectiveGasPrice = ethtypes.EthBigInt(effectiveGasPrice) - - if receipt.To == nil && lookup.Receipt.ExitCode.IsSuccess() { - // Create and Create2 return the same things. - var ret eam.CreateExternalReturn - if err := ret.UnmarshalCBOR(bytes.NewReader(lookup.Receipt.Return)); err != nil { - return api.EthTxReceipt{}, xerrors.Errorf("failed to parse contract creation result: %w", err) - } - addr := ethtypes.EthAddress(ret.EthAddress) - receipt.ContractAddress = &addr - } - - if len(events) > 0 { - receipt.Logs = make([]ethtypes.EthLog, 0, len(events)) - for i, evt := range events { - l := ethtypes.EthLog{ - Removed: false, - LogIndex: ethtypes.EthUint64(i), - TransactionHash: tx.Hash, - TransactionIndex: transactionIndex, - BlockHash: blockHash, - BlockNumber: blockNumber, - } - - data, topics, ok := ethLogFromEvent(evt.Entries) - if !ok { - // not an eth event. - continue - } - for _, topic := range topics { - ethtypes.EthBloomSet(receipt.LogsBloom, topic[:]) - } - l.Data = data - l.Topics = topics - - addr, err := address.NewIDAddress(uint64(evt.Emitter)) - if err != nil { - return api.EthTxReceipt{}, xerrors.Errorf("failed to create ID address: %w", err) - } - - l.Address, err = lookupEthAddress(ctx, addr, sa) - if err != nil { - return api.EthTxReceipt{}, xerrors.Errorf("failed to resolve Ethereum address: %w", err) - } - - ethtypes.EthBloomSet(receipt.LogsBloom, l.Address[:]) - receipt.Logs = append(receipt.Logs, l) - } - } - - return receipt, nil -} - -func (m *EthTxHashManager) Apply(ctx context.Context, from, to *types.TipSet) error { - for _, blk := range to.Blocks() { - _, smsgs, err := m.StateAPI.Chain.MessagesForBlock(ctx, blk) - if err != nil { - return err - } - - for _, smsg := range smsgs { - if smsg.Signature.Type != crypto.SigTypeDelegated { - continue - } - - hash, err := EthTxHashFromSignedMessage(ctx, smsg, m.StateAPI) - if err != nil { - return err - } - - err = m.TransactionHashLookup.UpsertHash(hash, smsg.Cid()) - if err != nil { - return err - } - } - } - - return nil -} - -type EthTxHashManager struct { - StateAPI StateAPI - TransactionHashLookup *ethhashlookup.EthTxHashLookup -} - -func (m *EthTxHashManager) Revert(ctx context.Context, from, to *types.TipSet) error { - return nil -} - -func (m *EthTxHashManager) PopulateExistingMappings(ctx context.Context, minHeight abi.ChainEpoch) error { - if minHeight < build.UpgradeHyggeHeight { - minHeight = build.UpgradeHyggeHeight - } - - ts := m.StateAPI.Chain.GetHeaviestTipSet() - for ts.Height() > minHeight { - 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 { - m.ProcessSignedMessage(ctx, msg) - } - } - - var err error - ts, err = m.StateAPI.Chain.GetTipSetFromKey(ctx, ts.Parents()) - if err != nil { - return err - } - } - - return nil -} - -func (m *EthTxHashManager) ProcessSignedMessage(ctx context.Context, msg *types.SignedMessage) { - if msg.Signature.Type != crypto.SigTypeDelegated { - return - } - - ethTx, err := newEthTxFromSignedMessage(ctx, msg, m.StateAPI) - if err != nil { - log.Errorf("error converting filecoin message to eth tx: %s", err) - return - } - - err = m.TransactionHashLookup.UpsertHash(ethTx.Hash, msg.Cid()) - if err != nil { - log.Errorf("error inserting tx mapping to db: %s", err) - return - } -} - -func WaitForMpoolUpdates(ctx context.Context, ch <-chan api.MpoolUpdate, manager *EthTxHashManager) { - for { - select { - case <-ctx.Done(): - return - case u := <-ch: - if u.Type != api.MpoolAdd { - continue - } - - manager.ProcessSignedMessage(ctx, u.Message) - } - } -} - -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) - } -} - -func parseEthTopics(topics ethtypes.EthTopicSpec) (map[string][][]byte, error) { - keys := map[string][][]byte{} - for idx, vals := range topics { - if len(vals) == 0 { - continue - } - // Ethereum topics are emitted using `LOG{0..4}` opcodes resulting in topics1..4 - key := fmt.Sprintf("t%d", idx+1) - for _, v := range vals { - v := v // copy the ethhash to avoid repeatedly referencing the same one. - keys[key] = append(keys[key], v[:]) - } - } - return keys, nil -} - -const errorFunctionSelector = "\x08\xc3\x79\xa0" // Error(string) -const panicFunctionSelector = "\x4e\x48\x7b\x71" // Panic(uint256) -// Eth ABI (solidity) panic codes. -var panicErrorCodes map[uint64]string = map[uint64]string{ - 0x00: "Panic()", - 0x01: "Assert()", - 0x11: "ArithmeticOverflow()", - 0x12: "DivideByZero()", - 0x21: "InvalidEnumVariant()", - 0x22: "InvalidStorageArray()", - 0x31: "PopEmptyArray()", - 0x32: "ArrayIndexOutOfBounds()", - 0x41: "OutOfMemory()", - 0x51: "CalledUninitializedFunction()", -} - -// Parse an ABI encoded revert reason. This reason should be encoded as if it were the parameters to -// an `Error(string)` function call. -// -// See https://docs.soliditylang.org/en/latest/control-structures.html#panic-via-assert-and-error-via-require -func parseEthRevert(ret []byte) string { - if len(ret) == 0 { - return "none" - } - var cbytes abi.CborBytes - if err := cbytes.UnmarshalCBOR(bytes.NewReader(ret)); err != nil { - return "ERROR: revert reason is not cbor encoded bytes" - } - if len(cbytes) == 0 { - return "none" - } - // If it's not long enough to contain an ABI encoded response, return immediately. - if len(cbytes) < 4+32 { - return ethtypes.EthBytes(cbytes).String() - } - switch string(cbytes[:4]) { - case panicFunctionSelector: - cbytes := cbytes[4 : 4+32] - // Read the and check the code. - code, err := ethtypes.EthUint64FromBytes(cbytes) - if err != nil { - // If it's too big, just return the raw value. - codeInt := big.PositiveFromUnsignedBytes(cbytes) - return fmt.Sprintf("Panic(%s)", ethtypes.EthBigInt(codeInt).String()) - } - if s, ok := panicErrorCodes[uint64(code)]; ok { - return s - } - return fmt.Sprintf("Panic(0x%x)", code) - case errorFunctionSelector: - cbytes := cbytes[4:] - cbytesLen := ethtypes.EthUint64(len(cbytes)) - // Read the and check the offset. - offset, err := ethtypes.EthUint64FromBytes(cbytes[:32]) - if err != nil { - break - } - if cbytesLen < offset { - break - } - - // Read and check the length. - if cbytesLen-offset < 32 { - break - } - start := offset + 32 - length, err := ethtypes.EthUint64FromBytes(cbytes[offset : offset+32]) - if err != nil { - break - } - if cbytesLen-start < length { - break - } - // Slice the error message. - return fmt.Sprintf("Error(%s)", cbytes[start:start+length]) - } - return ethtypes.EthBytes(cbytes).String() -} - -func calculateRewardsAndGasUsed(rewardPercentiles []float64, txGasRewards gasRewardSorter) ([]ethtypes.EthBigInt, int64) { - var gasUsedTotal int64 - for _, tx := range txGasRewards { - gasUsedTotal += tx.gasUsed - } - - rewards := make([]ethtypes.EthBigInt, len(rewardPercentiles)) - for i := range rewards { - rewards[i] = ethtypes.EthBigInt(types.NewInt(MinGasPremium)) - } - - if len(txGasRewards) == 0 { - return rewards, gasUsedTotal - } - - sort.Stable(txGasRewards) - - var idx int - var sum int64 - for i, percentile := range rewardPercentiles { - threshold := int64(float64(gasUsedTotal) * percentile / 100) - for sum < threshold && idx < len(txGasRewards)-1 { - sum += txGasRewards[idx].gasUsed - idx++ - } - rewards[i] = ethtypes.EthBigInt(txGasRewards[idx].premium) + rewards[i] = ethtypes.EthBigInt(txGasRewards[idx].premium) } return rewards, gasUsedTotal } -func getSignedMessage(ctx context.Context, cs *store.ChainStore, msgCid cid.Cid) (*types.SignedMessage, error) { - smsg, err := cs.GetSignedMessage(ctx, msgCid) - if err != nil { - // We couldn't find the signed message, it might be a BLS message, so search for a regular message. - msg, err := cs.GetMessage(ctx, msgCid) - if err != nil { - return nil, xerrors.Errorf("failed to find msg %s: %w", msgCid, err) - } - smsg = &types.SignedMessage{ - Message: *msg, - Signature: crypto.Signature{ - Type: crypto.SigTypeBLS, - }, - } - } - - return smsg, nil -} - type gasRewardTuple struct { gasUsed int64 premium abi.TokenAmount diff --git a/node/impl/full/eth_event.go b/node/impl/full/eth_event.go new file mode 100644 index 00000000000..69021e08aed --- /dev/null +++ b/node/impl/full/eth_event.go @@ -0,0 +1,382 @@ +package full + +import ( + "context" + "encoding/json" + "sync" + + "github.com/google/uuid" + "github.com/ipfs/go-cid" + "github.com/zyedidia/generic/queue" + "golang.org/x/xerrors" + + "github.com/filecoin-project/go-jsonrpc" + + "github.com/filecoin-project/lotus/chain/events/filter" + "github.com/filecoin-project/lotus/chain/store" + "github.com/filecoin-project/lotus/chain/types" + "github.com/filecoin-project/lotus/chain/types/ethtypes" +) + +type filterEventCollector interface { + TakeCollectedEvents(context.Context) []*filter.CollectedEvent +} + +type filterMessageCollector interface { + TakeCollectedMessages(context.Context) []*types.SignedMessage +} + +type filterTipSetCollector interface { + TakeCollectedTipSets(context.Context) []types.TipSetKey +} + +func ethLogFromEvent(entries []types.EventEntry) (data []byte, topics []ethtypes.EthHash, ok bool) { + var ( + topicsFound [4]bool + topicsFoundCount int + dataFound bool + ) + // Topics must be non-nil, even if empty. So we might as well pre-allocate for 4 (the max). + topics = make([]ethtypes.EthHash, 0, 4) + for _, entry := range entries { + // Drop events with non-raw topics to avoid mistakes. + if entry.Codec != cid.Raw { + log.Warnw("did not expect an event entry with a non-raw codec", "codec", entry.Codec, "key", entry.Key) + return nil, nil, false + } + // Check if the key is t1..t4 + if len(entry.Key) == 2 && "t1" <= entry.Key && entry.Key <= "t4" { + // '1' - '1' == 0, etc. + idx := int(entry.Key[1] - '1') + + // Drop events with mis-sized topics. + if len(entry.Value) != 32 { + log.Warnw("got an EVM event topic with an invalid size", "key", entry.Key, "size", len(entry.Value)) + return nil, nil, false + } + + // Drop events with duplicate topics. + if topicsFound[idx] { + log.Warnw("got a duplicate EVM event topic", "key", entry.Key) + return nil, nil, false + } + topicsFound[idx] = true + topicsFoundCount++ + + // Extend the topics array + for len(topics) <= idx { + topics = append(topics, ethtypes.EthHash{}) + } + copy(topics[idx][:], entry.Value) + } else if entry.Key == "d" { + // Drop events with duplicate data fields. + if dataFound { + log.Warnw("got duplicate EVM event data") + return nil, nil, false + } + + dataFound = true + data = entry.Value + } else { + // Skip entries we don't understand (makes it easier to extend things). + // But we warn for now because we don't expect them. + log.Warnw("unexpected event entry", "key", entry.Key) + } + + } + + // Drop events with skipped topics. + if len(topics) != topicsFoundCount { + log.Warnw("EVM event topic length mismatch", "expected", len(topics), "actual", topicsFoundCount) + return nil, nil, false + } + return data, topics, true +} + +func ethFilterResultFromEvents(evs []*filter.CollectedEvent, sa StateAPI) (*ethtypes.EthFilterResult, error) { + res := ðtypes.EthFilterResult{} + for _, ev := range evs { + log := ethtypes.EthLog{ + Removed: ev.Reverted, + LogIndex: ethtypes.EthUint64(ev.EventIdx), + TransactionIndex: ethtypes.EthUint64(ev.MsgIdx), + BlockNumber: ethtypes.EthUint64(ev.Height), + } + var ( + err error + ok bool + ) + + log.Data, log.Topics, ok = ethLogFromEvent(ev.Entries) + if !ok { + continue + } + + log.Address, err = ethtypes.EthAddressFromFilecoinAddress(ev.EmitterAddr) + if err != nil { + return nil, err + } + + log.TransactionHash, err = ethTxHashFromMessageCid(context.TODO(), ev.MsgCid, sa) + if err != nil { + return nil, err + } + c, err := ev.TipSetKey.Cid() + if err != nil { + return nil, err + } + log.BlockHash, err = ethtypes.EthHashFromCid(c) + if err != nil { + return nil, err + } + + res.Results = append(res.Results, log) + } + + return res, nil +} + +func ethFilterResultFromTipSets(tsks []types.TipSetKey) (*ethtypes.EthFilterResult, error) { + res := ðtypes.EthFilterResult{} + + for _, tsk := range tsks { + c, err := tsk.Cid() + if err != nil { + return nil, err + } + hash, err := ethtypes.EthHashFromCid(c) + if err != nil { + return nil, err + } + + res.Results = append(res.Results, hash) + } + + return res, nil +} + +func ethFilterResultFromMessages(cs []*types.SignedMessage, sa StateAPI) (*ethtypes.EthFilterResult, error) { + res := ðtypes.EthFilterResult{} + + for _, c := range cs { + hash, err := ethTxHashFromSignedMessage(context.TODO(), c, sa) + if err != nil { + return nil, err + } + + res.Results = append(res.Results, hash) + } + + return res, nil +} + +type EthSubscriptionManager struct { + Chain *store.ChainStore + StateAPI StateAPI + ChainAPI ChainAPI + mu sync.Mutex + subs map[ethtypes.EthSubscriptionID]*ethSubscription +} + +func (e *EthSubscriptionManager) StartSubscription(ctx context.Context, out ethSubscriptionCallback, dropFilter func(context.Context, filter.Filter) error) (*ethSubscription, error) { // nolint + rawid, err := uuid.NewRandom() + if err != nil { + return nil, xerrors.Errorf("new uuid: %w", err) + } + id := ethtypes.EthSubscriptionID{} + copy(id[:], rawid[:]) // uuid is 16 bytes + + ctx, quit := context.WithCancel(ctx) + + sub := ðSubscription{ + Chain: e.Chain, + StateAPI: e.StateAPI, + ChainAPI: e.ChainAPI, + uninstallFilter: dropFilter, + id: id, + in: make(chan interface{}, 200), + out: out, + quit: quit, + + toSend: queue.New[[]byte](), + sendCond: make(chan struct{}, 1), + } + + e.mu.Lock() + if e.subs == nil { + e.subs = make(map[ethtypes.EthSubscriptionID]*ethSubscription) + } + e.subs[sub.id] = sub + e.mu.Unlock() + + go sub.start(ctx) + go sub.startOut(ctx) + + return sub, nil +} + +func (e *EthSubscriptionManager) StopSubscription(ctx context.Context, id ethtypes.EthSubscriptionID) error { + e.mu.Lock() + defer e.mu.Unlock() + + sub, ok := e.subs[id] + if !ok { + return xerrors.Errorf("subscription not found") + } + sub.stop() + delete(e.subs, id) + + return nil +} + +type ethSubscriptionCallback func(context.Context, jsonrpc.RawParams) error + +const maxSendQueue = 20000 + +type ethSubscription struct { + Chain *store.ChainStore + StateAPI StateAPI + ChainAPI ChainAPI + uninstallFilter func(context.Context, filter.Filter) error + id ethtypes.EthSubscriptionID + in chan interface{} + out ethSubscriptionCallback + + mu sync.Mutex + filters []filter.Filter + quit func() + + sendLk sync.Mutex + sendQueueLen int + toSend *queue.Queue[[]byte] + sendCond chan struct{} +} + +func (e *ethSubscription) addFilter(ctx context.Context, f filter.Filter) { + e.mu.Lock() + defer e.mu.Unlock() + + f.SetSubChannel(e.in) + e.filters = append(e.filters, f) +} + +// sendOut processes the final subscription queue. It's here in case the subscriber +// is slow, and we need to buffer the messages. +func (e *ethSubscription) startOut(ctx context.Context) { + for { + select { + case <-ctx.Done(): + return + case <-e.sendCond: + e.sendLk.Lock() + + for !e.toSend.Empty() { + front := e.toSend.Dequeue() + e.sendQueueLen-- + + e.sendLk.Unlock() + + if err := e.out(ctx, front); err != nil { + log.Warnw("error sending subscription response, killing subscription", "sub", e.id, "error", err) + e.stop() + return + } + + e.sendLk.Lock() + } + + e.sendLk.Unlock() + } + } +} + +func (e *ethSubscription) send(ctx context.Context, v interface{}) { + resp := ethtypes.EthSubscriptionResponse{ + SubscriptionID: e.id, + Result: v, + } + + outParam, err := json.Marshal(resp) + if err != nil { + log.Warnw("marshaling subscription response", "sub", e.id, "error", err) + return + } + + e.sendLk.Lock() + defer e.sendLk.Unlock() + + e.toSend.Enqueue(outParam) + + e.sendQueueLen++ + if e.sendQueueLen > maxSendQueue { + log.Warnw("subscription send queue full, killing subscription", "sub", e.id) + e.stop() + return + } + + select { + case e.sendCond <- struct{}{}: + default: // already signalled, and we're holding the lock so we know that the event will be processed + } +} + +func (e *ethSubscription) start(ctx context.Context) { + for { + select { + case <-ctx.Done(): + return + case v := <-e.in: + switch vt := v.(type) { + case *filter.CollectedEvent: + evs, err := ethFilterResultFromEvents([]*filter.CollectedEvent{vt}, e.StateAPI) + if err != nil { + continue + } + + for _, r := range evs.Results { + e.send(ctx, r) + } + case *types.TipSet: + ev, err := newEthBlockFromFilecoinTipSet(ctx, vt, true, e.Chain, e.StateAPI) + if err != nil { + break + } + + e.send(ctx, ev) + case *types.SignedMessage: // mpool txid + evs, err := ethFilterResultFromMessages([]*types.SignedMessage{vt}, e.StateAPI) + if err != nil { + continue + } + + for _, r := range evs.Results { + e.send(ctx, r) + } + default: + log.Warnf("unexpected subscription value type: %T", vt) + } + } + } +} + +func (e *ethSubscription) stop() { + e.mu.Lock() + if e.quit == nil { + e.mu.Unlock() + return + } + + if e.quit != nil { + e.quit() + e.quit = nil + e.mu.Unlock() + + for _, f := range e.filters { + // note: the context in actually unused in uninstallFilter + if err := e.uninstallFilter(context.TODO(), f); err != nil { + // this will leave the filter a zombie, collecting events up to the maximum allowed + log.Warnf("failed to remove filter when unsubscribing: %v", err) + } + } + } +} diff --git a/node/impl/full/trace.go b/node/impl/full/eth_trace.go similarity index 68% rename from node/impl/full/trace.go rename to node/impl/full/eth_trace.go index 02ffdd46b9a..3e3b48c904f 100644 --- a/node/impl/full/trace.go +++ b/node/impl/full/eth_trace.go @@ -2,49 +2,22 @@ package full import ( "bytes" - "context" "encoding/binary" "encoding/hex" "fmt" "io" "github.com/ipfs/go-cid" - "go.uber.org/fx" - "golang.org/x/xerrors" "github.com/filecoin-project/go-state-types/abi" "github.com/filecoin-project/go-state-types/builtin" "github.com/filecoin-project/go-state-types/exitcode" - "github.com/filecoin-project/lotus/api" builtinactors "github.com/filecoin-project/lotus/chain/actors/builtin" - "github.com/filecoin-project/lotus/chain/stmgr" - "github.com/filecoin-project/lotus/chain/store" "github.com/filecoin-project/lotus/chain/types" "github.com/filecoin-project/lotus/chain/types/ethtypes" ) -type EthTraceAPI interface { - TraceBlock(ctx context.Context, blkNum string) (interface{}, error) - TraceReplayBlockTransactions(ctx context.Context, blkNum string, traceTypes []string) (interface{}, error) -} - -var ( - _ EthTraceAPI = *new(api.FullNode) -) - -type EthTrace struct { - fx.In - - Chain *store.ChainStore - StateManager *stmgr.StateManager - - ChainAPI - EthModuleAPI -} - -var _ EthTraceAPI = (*EthTrace)(nil) - type Trace struct { Action Action `json:"action"` Result Result `json:"result"` @@ -93,168 +66,6 @@ type Result struct { Output string `json:"output"` } -func (e *EthTrace) TraceBlock(ctx context.Context, blkNum string) (interface{}, error) { - ts, err := e.getTipsetByBlockNr(ctx, blkNum, false) - if err != nil { - return nil, err - } - - _, trace, err := e.StateManager.ExecutionTrace(ctx, ts) - if err != nil { - return nil, xerrors.Errorf("failed to compute base state: %w", err) - } - - tsParent, err := e.ChainAPI.ChainGetTipSetByHeight(ctx, ts.Height()+1, e.Chain.GetHeaviestTipSet().Key()) - if err != nil { - return nil, fmt.Errorf("cannot get tipset at height: %v", ts.Height()+1) - } - - msgs, err := e.ChainGetParentMessages(ctx, tsParent.Blocks()[0].Cid()) - if err != nil { - return nil, err - } - - cid, err := ts.Key().Cid() - if err != nil { - return nil, err - } - - blkHash, err := ethtypes.EthHashFromCid(cid) - if err != nil { - return nil, err - } - - allTraces := make([]*TraceBlock, 0, len(trace)) - for _, ir := range trace { - // ignore messages from f00 - if ir.Msg.From.String() == "f00" { - continue - } - - idx := -1 - for msgIdx, msg := range msgs { - if ir.Msg.From == msg.Message.From { - idx = msgIdx - break - } - } - if idx == -1 { - log.Warnf("cannot resolve message index for cid: %s", ir.MsgCid) - continue - } - - txHash, err := e.EthGetTransactionHashByCid(ctx, ir.MsgCid) - if err != nil { - return nil, err - } - if txHash == nil { - log.Warnf("cannot find transaction hash for cid %s", ir.MsgCid) - continue - } - - traces := []*Trace{} - err = buildTraces(&traces, nil, []int{}, ir.ExecutionTrace, int64(ts.Height())) - if err != nil { - return nil, xerrors.Errorf("failed when building traces: %w", err) - } - - traceBlocks := make([]*TraceBlock, 0, len(trace)) - for _, trace := range traces { - traceBlocks = append(traceBlocks, &TraceBlock{ - Trace: trace, - BlockHash: blkHash, - BlockNumber: int64(ts.Height()), - TransactionHash: *txHash, - TransactionPosition: idx, - }) - } - - allTraces = append(allTraces, traceBlocks...) - } - - return allTraces, nil -} - -func (e *EthTrace) TraceReplayBlockTransactions(ctx context.Context, blkNum string, traceTypes []string) (interface{}, error) { - if len(traceTypes) != 1 || traceTypes[0] != "trace" { - return nil, fmt.Errorf("only 'trace' is supported") - } - - ts, err := e.getTipsetByBlockNr(ctx, blkNum, false) - if err != nil { - return nil, err - } - - _, trace, err := e.StateManager.ExecutionTrace(ctx, ts) - if err != nil { - return nil, xerrors.Errorf("failed when calling ExecutionTrace: %w", err) - } - - allTraces := make([]*TraceReplayBlockTransaction, 0, len(trace)) - for _, ir := range trace { - // ignore messages from f00 - if ir.Msg.From.String() == "f00" { - continue - } - - txHash, err := e.EthGetTransactionHashByCid(ctx, ir.MsgCid) - if err != nil { - return nil, err - } - if txHash == nil { - log.Warnf("cannot find transaction hash for cid %s", ir.MsgCid) - continue - } - - t := TraceReplayBlockTransaction{ - Output: hex.EncodeToString(ir.MsgRct.Return), - TransactionHash: *txHash, - StateDiff: nil, - VmTrace: nil, - } - - err = buildTraces(&t.Trace, nil, []int{}, ir.ExecutionTrace, int64(ts.Height())) - if err != nil { - return nil, xerrors.Errorf("failed when building traces: %w", err) - } - - allTraces = append(allTraces, &t) - } - - return allTraces, nil -} - -func writePadded[T any](w io.Writer, data T, size int) error { - tmp := &bytes.Buffer{} - - // first write data to tmp buffer to get the size - err := binary.Write(tmp, binary.BigEndian, data) - if err != nil { - return fmt.Errorf("writePadded: failed writing tmp data to buffer: %w", err) - } - - if tmp.Len() > size { - return fmt.Errorf("writePadded: data is larger than size") - } - - // write tailing zeros to pad up to size - cnt := size - tmp.Len() - for i := 0; i < cnt; i++ { - err = binary.Write(w, binary.BigEndian, uint8(0)) - if err != nil { - return fmt.Errorf("writePadded: failed writing tailing zeros to buffer: %w", err) - } - } - - // finally write the actual value - err = binary.Write(w, binary.BigEndian, tmp.Bytes()) - if err != nil { - return fmt.Errorf("writePadded: failed writing data to buffer: %w", err) - } - - return nil -} - // buildTraces recursively builds the traces for a given ExecutionTrace by walking the subcalls func buildTraces(traces *[]*Trace, parent *Trace, addr []int, et types.ExecutionTrace, height int64) error { trace := &Trace{ @@ -416,6 +227,37 @@ func buildTraces(traces *[]*Trace, parent *Trace, addr []int, et types.Execution return nil } +func writePadded[T any](w io.Writer, data T, size int) error { + tmp := &bytes.Buffer{} + + // first write data to tmp buffer to get the size + err := binary.Write(tmp, binary.BigEndian, data) + if err != nil { + return fmt.Errorf("writePadded: failed writing tmp data to buffer: %w", err) + } + + if tmp.Len() > size { + return fmt.Errorf("writePadded: data is larger than size") + } + + // write tailing zeros to pad up to size + cnt := size - tmp.Len() + for i := 0; i < cnt; i++ { + err = binary.Write(w, binary.BigEndian, uint8(0)) + if err != nil { + return fmt.Errorf("writePadded: failed writing tailing zeros to buffer: %w", err) + } + } + + // finally write the actual value + err = binary.Write(w, binary.BigEndian, tmp.Bytes()) + if err != nil { + return fmt.Errorf("writePadded: failed writing data to buffer: %w", err) + } + + return nil +} + func handleFilecoinMethodInput(method abi.MethodNum, codec uint64, params []byte) ([]byte, error) { NATIVE_METHOD_SELECTOR := []byte{0x86, 0x8e, 0x10, 0xc4} EVM_WORD_SIZE := 32 @@ -477,39 +319,3 @@ func handleFilecoinMethodOutput(exitCode exitcode.ExitCode, codec uint64, data [ return w.Bytes(), nil } - -// TODO: refactor this to be shared code -func (e *EthTrace) getTipsetByBlockNr(ctx context.Context, blkParam string, strict bool) (*types.TipSet, error) { - if blkParam == "earliest" { - return nil, fmt.Errorf("block param \"earliest\" is not supported") - } - - head := e.Chain.GetHeaviestTipSet() - switch blkParam { - case "pending": - return head, nil - case "latest": - parent, err := e.Chain.GetTipSetFromKey(ctx, head.Parents()) - if err != nil { - return nil, fmt.Errorf("cannot get parent tipset") - } - return parent, nil - default: - var num ethtypes.EthUint64 - err := num.UnmarshalJSON([]byte(`"` + blkParam + `"`)) - if err != nil { - return nil, fmt.Errorf("cannot parse block number: %v", err) - } - if abi.ChainEpoch(num) > head.Height()-1 { - return nil, fmt.Errorf("requested a future epoch (beyond 'latest')") - } - ts, err := e.ChainAPI.ChainGetTipSetByHeight(ctx, abi.ChainEpoch(num), head.Key()) - if err != nil { - return nil, fmt.Errorf("cannot get tipset at height: %v", num) - } - if strict && ts.Height() != abi.ChainEpoch(num) { - return nil, ErrNullRound - } - return ts, nil - } -} diff --git a/node/impl/full/eth_utils.go b/node/impl/full/eth_utils.go new file mode 100644 index 00000000000..ab17d13b2c0 --- /dev/null +++ b/node/impl/full/eth_utils.go @@ -0,0 +1,689 @@ +package full + +import ( + "bytes" + "context" + "errors" + "fmt" + + "github.com/ipfs/go-cid" + "golang.org/x/xerrors" + + "github.com/filecoin-project/go-address" + "github.com/filecoin-project/go-state-types/abi" + "github.com/filecoin-project/go-state-types/big" + builtintypes "github.com/filecoin-project/go-state-types/builtin" + "github.com/filecoin-project/go-state-types/builtin/v10/eam" + "github.com/filecoin-project/go-state-types/crypto" + + "github.com/filecoin-project/lotus/api" + "github.com/filecoin-project/lotus/build" + "github.com/filecoin-project/lotus/chain/actors" + "github.com/filecoin-project/lotus/chain/store" + "github.com/filecoin-project/lotus/chain/types" + "github.com/filecoin-project/lotus/chain/types/ethtypes" + "github.com/filecoin-project/lotus/chain/vm" +) + +func getTipsetByBlockNr(ctx context.Context, chain *store.ChainStore, blkParam string, strict bool) (*types.TipSet, error) { + if blkParam == "earliest" { + return nil, fmt.Errorf("block param \"earliest\" is not supported") + } + + head := chain.GetHeaviestTipSet() + switch blkParam { + case "pending": + return head, nil + case "latest": + parent, err := chain.GetTipSetFromKey(ctx, head.Parents()) + if err != nil { + return nil, fmt.Errorf("cannot get parent tipset") + } + return parent, nil + default: + var num ethtypes.EthUint64 + err := num.UnmarshalJSON([]byte(`"` + blkParam + `"`)) + if err != nil { + return nil, fmt.Errorf("cannot parse block number: %v", err) + } + if abi.ChainEpoch(num) > head.Height()-1 { + return nil, fmt.Errorf("requested a future epoch (beyond 'latest')") + } + ts, err := chain.GetTipsetByHeight(ctx, abi.ChainEpoch(num), head, true) + if err != nil { + return nil, fmt.Errorf("cannot get tipset at height: %v", num) + } + if strict && ts.Height() != abi.ChainEpoch(num) { + return nil, ErrNullRound + } + return ts, nil + } +} + +func getTipsetByEthBlockNumberOrHash(ctx context.Context, chain *store.ChainStore, blkParam ethtypes.EthBlockNumberOrHash) (*types.TipSet, error) { + head := chain.GetHeaviestTipSet() + + predefined := blkParam.PredefinedBlock + if predefined != nil { + if *predefined == "earliest" { + return nil, fmt.Errorf("block param \"earliest\" is not supported") + } else if *predefined == "pending" { + return head, nil + } else if *predefined == "latest" { + parent, err := chain.GetTipSetFromKey(ctx, head.Parents()) + if err != nil { + return nil, fmt.Errorf("cannot get parent tipset") + } + return parent, nil + } else { + return nil, fmt.Errorf("unknown predefined block %s", *predefined) + } + } + + if blkParam.BlockNumber != nil { + height := abi.ChainEpoch(*blkParam.BlockNumber) + if height > head.Height()-1 { + return nil, fmt.Errorf("requested a future epoch (beyond 'latest')") + } + ts, err := chain.GetTipsetByHeight(ctx, height, head, true) + if err != nil { + return nil, fmt.Errorf("cannot get tipset at height: %v", height) + } + return ts, nil + } + + if blkParam.BlockHash != nil { + ts, err := chain.GetTipSetByCid(ctx, blkParam.BlockHash.ToCid()) + if err != nil { + return nil, fmt.Errorf("cannot get tipset by hash: %v", err) + } + + // verify that the tipset is in the canonical chain + if blkParam.RequireCanonical { + // walk up the current chain (our head) until we reach ts.Height() + walkTs, err := chain.GetTipsetByHeight(ctx, ts.Height(), head, true) + if err != nil { + return nil, fmt.Errorf("cannot get tipset at height: %v", ts.Height()) + } + + // verify that it equals the expected tipset + if !walkTs.Equals(ts) { + return nil, fmt.Errorf("tipset is not canonical") + } + } + + return ts, nil + } + + return nil, errors.New("invalid block param") +} + +func ethCallToFilecoinMessage(ctx context.Context, tx ethtypes.EthCall) (*types.Message, error) { + var from address.Address + if tx.From == nil || *tx.From == (ethtypes.EthAddress{}) { + // Send from the filecoin "system" address. + var err error + from, err = (ethtypes.EthAddress{}).ToFilecoinAddress() + if err != nil { + return nil, fmt.Errorf("failed to construct the ethereum system address: %w", err) + } + } else { + // The from address must be translatable to an f4 address. + var err error + from, err = tx.From.ToFilecoinAddress() + if err != nil { + return nil, fmt.Errorf("failed to translate sender address (%s): %w", tx.From.String(), err) + } + if p := from.Protocol(); p != address.Delegated { + return nil, fmt.Errorf("expected a class 4 address, got: %d: %w", p, err) + } + } + + var params []byte + if len(tx.Data) > 0 { + initcode := abi.CborBytes(tx.Data) + params2, err := actors.SerializeParams(&initcode) + if err != nil { + return nil, fmt.Errorf("failed to serialize params: %w", err) + } + params = params2 + } + + var to address.Address + var method abi.MethodNum + if tx.To == nil { + // this is a contract creation + to = builtintypes.EthereumAddressManagerActorAddr + method = builtintypes.MethodsEAM.CreateExternal + } else { + addr, err := tx.To.ToFilecoinAddress() + if err != nil { + return nil, xerrors.Errorf("cannot get Filecoin address: %w", err) + } + to = addr + method = builtintypes.MethodsEVM.InvokeContract + } + + return &types.Message{ + From: from, + To: to, + Value: big.Int(tx.Value), + Method: method, + Params: params, + GasLimit: build.BlockGasLimit, + GasFeeCap: big.Zero(), + GasPremium: big.Zero(), + }, nil +} + +func newEthBlockFromFilecoinTipSet(ctx context.Context, ts *types.TipSet, fullTxInfo bool, cs *store.ChainStore, sa StateAPI) (ethtypes.EthBlock, error) { + parentKeyCid, err := ts.Parents().Cid() + if err != nil { + return ethtypes.EthBlock{}, err + } + parentBlkHash, err := ethtypes.EthHashFromCid(parentKeyCid) + if err != nil { + return ethtypes.EthBlock{}, err + } + + bn := ethtypes.EthUint64(ts.Height()) + + blkCid, err := ts.Key().Cid() + if err != nil { + return ethtypes.EthBlock{}, err + } + blkHash, err := ethtypes.EthHashFromCid(blkCid) + if err != nil { + return ethtypes.EthBlock{}, err + } + + msgs, rcpts, err := messagesAndReceipts(ctx, ts, cs, sa) + if err != nil { + return ethtypes.EthBlock{}, xerrors.Errorf("failed to retrieve messages and receipts: %w", err) + } + + block := ethtypes.NewEthBlock(len(msgs) > 0) + + gasUsed := int64(0) + for i, msg := range msgs { + rcpt := rcpts[i] + ti := ethtypes.EthUint64(i) + gasUsed += rcpt.GasUsed + var smsg *types.SignedMessage + switch msg := msg.(type) { + case *types.SignedMessage: + smsg = msg + case *types.Message: + smsg = &types.SignedMessage{ + Message: *msg, + Signature: crypto.Signature{ + Type: crypto.SigTypeBLS, + }, + } + default: + return ethtypes.EthBlock{}, xerrors.Errorf("failed to get signed msg %s: %w", msg.Cid(), err) + } + tx, err := newEthTxFromSignedMessage(ctx, smsg, sa) + if err != nil { + return ethtypes.EthBlock{}, xerrors.Errorf("failed to convert msg to ethTx: %w", err) + } + + tx.ChainID = ethtypes.EthUint64(build.Eip155ChainId) + tx.BlockHash = &blkHash + tx.BlockNumber = &bn + tx.TransactionIndex = &ti + + if fullTxInfo { + block.Transactions = append(block.Transactions, tx) + } else { + block.Transactions = append(block.Transactions, tx.Hash.String()) + } + } + + block.Hash = blkHash + block.Number = bn + block.ParentHash = parentBlkHash + block.Timestamp = ethtypes.EthUint64(ts.Blocks()[0].Timestamp) + block.BaseFeePerGas = ethtypes.EthBigInt{Int: ts.Blocks()[0].ParentBaseFee.Int} + block.GasUsed = ethtypes.EthUint64(gasUsed) + return block, nil +} + +func messagesAndReceipts(ctx context.Context, ts *types.TipSet, cs *store.ChainStore, sa StateAPI) ([]types.ChainMsg, []types.MessageReceipt, error) { + msgs, err := cs.MessagesForTipset(ctx, ts) + if err != nil { + return nil, nil, xerrors.Errorf("error loading messages for tipset: %v: %w", ts, err) + } + + _, rcptRoot, err := sa.StateManager.TipSetState(ctx, ts) + if err != nil { + return nil, nil, xerrors.Errorf("failed to compute state: %w", err) + } + + rcpts, err := cs.ReadReceipts(ctx, rcptRoot) + if err != nil { + return nil, nil, xerrors.Errorf("error loading receipts for tipset: %v: %w", ts, err) + } + + if len(msgs) != len(rcpts) { + return nil, nil, xerrors.Errorf("receipts and message array lengths didn't match for tipset: %v: %w", ts, err) + } + + return msgs, rcpts, nil +} + +const errorFunctionSelector = "\x08\xc3\x79\xa0" // Error(string) +const panicFunctionSelector = "\x4e\x48\x7b\x71" // Panic(uint256) +// Eth ABI (solidity) panic codes. +var panicErrorCodes map[uint64]string = map[uint64]string{ + 0x00: "Panic()", + 0x01: "Assert()", + 0x11: "ArithmeticOverflow()", + 0x12: "DivideByZero()", + 0x21: "InvalidEnumVariant()", + 0x22: "InvalidStorageArray()", + 0x31: "PopEmptyArray()", + 0x32: "ArrayIndexOutOfBounds()", + 0x41: "OutOfMemory()", + 0x51: "CalledUninitializedFunction()", +} + +// Parse an ABI encoded revert reason. This reason should be encoded as if it were the parameters to +// an `Error(string)` function call. +// +// See https://docs.soliditylang.org/en/latest/control-structures.html#panic-via-assert-and-error-via-require +func parseEthRevert(ret []byte) string { + if len(ret) == 0 { + return "none" + } + var cbytes abi.CborBytes + if err := cbytes.UnmarshalCBOR(bytes.NewReader(ret)); err != nil { + return "ERROR: revert reason is not cbor encoded bytes" + } + if len(cbytes) == 0 { + return "none" + } + // If it's not long enough to contain an ABI encoded response, return immediately. + if len(cbytes) < 4+32 { + return ethtypes.EthBytes(cbytes).String() + } + switch string(cbytes[:4]) { + case panicFunctionSelector: + cbytes := cbytes[4 : 4+32] + // Read the and check the code. + code, err := ethtypes.EthUint64FromBytes(cbytes) + if err != nil { + // If it's too big, just return the raw value. + codeInt := big.PositiveFromUnsignedBytes(cbytes) + return fmt.Sprintf("Panic(%s)", ethtypes.EthBigInt(codeInt).String()) + } + if s, ok := panicErrorCodes[uint64(code)]; ok { + return s + } + return fmt.Sprintf("Panic(0x%x)", code) + case errorFunctionSelector: + cbytes := cbytes[4:] + cbytesLen := ethtypes.EthUint64(len(cbytes)) + // Read the and check the offset. + offset, err := ethtypes.EthUint64FromBytes(cbytes[:32]) + if err != nil { + break + } + if cbytesLen < offset { + break + } + + // Read and check the length. + if cbytesLen-offset < 32 { + break + } + start := offset + 32 + length, err := ethtypes.EthUint64FromBytes(cbytes[offset : offset+32]) + if err != nil { + break + } + if cbytesLen-start < length { + break + } + // Slice the error message. + return fmt.Sprintf("Error(%s)", cbytes[start:start+length]) + } + return ethtypes.EthBytes(cbytes).String() +} + +// lookupEthAddress makes its best effort at finding the Ethereum address for a +// Filecoin address. It does the following: +// +// 1. If the supplied address is an f410 address, we return its payload as the EthAddress. +// 2. Otherwise (f0, f1, f2, f3), we look up the actor on the state tree. If it has a delegated address, we return it if it's f410 address. +// 3. Otherwise, we fall back to returning a masked ID Ethereum address. If the supplied address is an f0 address, we +// use that ID to form the masked ID address. +// 4. Otherwise, we fetch the actor's ID from the state tree and form the masked ID with it. +func lookupEthAddress(ctx context.Context, addr address.Address, sa StateAPI) (ethtypes.EthAddress, error) { + // BLOCK A: We are trying to get an actual Ethereum address from an f410 address. + // Attempt to convert directly, if it's an f4 address. + ethAddr, err := ethtypes.EthAddressFromFilecoinAddress(addr) + if err == nil && !ethAddr.IsMaskedID() { + return ethAddr, nil + } + + // Lookup on the target actor and try to get an f410 address. + if actor, err := sa.StateGetActor(ctx, addr, types.EmptyTSK); err != nil { + return ethtypes.EthAddress{}, err + } else if actor.Address != nil { + if ethAddr, err := ethtypes.EthAddressFromFilecoinAddress(*actor.Address); err == nil && !ethAddr.IsMaskedID() { + return ethAddr, nil + } + } + + // BLOCK B: We gave up on getting an actual Ethereum address and are falling back to a Masked ID address. + // Check if we already have an ID addr, and use it if possible. + if err == nil && ethAddr.IsMaskedID() { + return ethAddr, nil + } + + // Otherwise, resolve the ID addr. + idAddr, err := sa.StateLookupID(ctx, addr, types.EmptyTSK) + if err != nil { + return ethtypes.EthAddress{}, err + } + return ethtypes.EthAddressFromFilecoinAddress(idAddr) +} + +func parseEthTopics(topics ethtypes.EthTopicSpec) (map[string][][]byte, error) { + keys := map[string][][]byte{} + for idx, vals := range topics { + if len(vals) == 0 { + continue + } + // Ethereum topics are emitted using `LOG{0..4}` opcodes resulting in topics1..4 + key := fmt.Sprintf("t%d", idx+1) + for _, v := range vals { + v := v // copy the ethhash to avoid repeatedly referencing the same one. + keys[key] = append(keys[key], v[:]) + } + } + return keys, nil +} + +func ethTxHashFromMessageCid(ctx context.Context, c cid.Cid, sa StateAPI) (ethtypes.EthHash, error) { + smsg, err := sa.Chain.GetSignedMessage(ctx, c) + if err == nil { + // This is an Eth Tx, Secp message, Or BLS message in the mpool + return ethTxHashFromSignedMessage(ctx, smsg, sa) + } + + _, err = sa.Chain.GetMessage(ctx, c) + if err == nil { + // This is a BLS message + return ethtypes.EthHashFromCid(c) + } + + return ethtypes.EmptyEthHash, nil +} + +func ethTxHashFromSignedMessage(ctx context.Context, smsg *types.SignedMessage, sa StateAPI) (ethtypes.EthHash, error) { + if smsg.Signature.Type == crypto.SigTypeDelegated { + ethTx, err := newEthTxFromSignedMessage(ctx, smsg, sa) + if err != nil { + return ethtypes.EmptyEthHash, err + } + return ethTx.Hash, nil + } else if smsg.Signature.Type == crypto.SigTypeSecp256k1 { + return ethtypes.EthHashFromCid(smsg.Cid()) + } else { // BLS message + return ethtypes.EthHashFromCid(smsg.Message.Cid()) + } +} + +func newEthTxFromSignedMessage(ctx context.Context, smsg *types.SignedMessage, sa StateAPI) (ethtypes.EthTx, error) { + var tx ethtypes.EthTx + var err error + + // This is an eth tx + if smsg.Signature.Type == crypto.SigTypeDelegated { + tx, err = ethtypes.EthTxFromSignedEthMessage(smsg) + if err != nil { + return ethtypes.EthTx{}, xerrors.Errorf("failed to convert from signed message: %w", err) + } + + tx.Hash, err = tx.TxHash() + if err != nil { + return ethtypes.EthTx{}, xerrors.Errorf("failed to calculate hash for ethTx: %w", err) + } + + fromAddr, err := lookupEthAddress(ctx, smsg.Message.From, sa) + if err != nil { + return ethtypes.EthTx{}, xerrors.Errorf("failed to resolve Ethereum address: %w", err) + } + + tx.From = fromAddr + } else if smsg.Signature.Type == crypto.SigTypeSecp256k1 { // Secp Filecoin Message + tx = ethTxFromNativeMessage(ctx, smsg.VMMessage(), sa) + tx.Hash, err = ethtypes.EthHashFromCid(smsg.Cid()) + if err != nil { + return tx, err + } + } else { // BLS Filecoin message + tx = ethTxFromNativeMessage(ctx, smsg.VMMessage(), sa) + tx.Hash, err = ethtypes.EthHashFromCid(smsg.Message.Cid()) + if err != nil { + return tx, err + } + } + + return tx, nil +} + +// ethTxFromNativeMessage does NOT populate: +// - BlockHash +// - BlockNumber +// - TransactionIndex +// - Hash +func ethTxFromNativeMessage(ctx context.Context, msg *types.Message, sa StateAPI) ethtypes.EthTx { + // We don't care if we error here, conversion is best effort for non-eth transactions + from, _ := lookupEthAddress(ctx, msg.From, sa) + to, _ := lookupEthAddress(ctx, msg.To, sa) + return ethtypes.EthTx{ + To: &to, + From: from, + Nonce: ethtypes.EthUint64(msg.Nonce), + ChainID: ethtypes.EthUint64(build.Eip155ChainId), + Value: ethtypes.EthBigInt(msg.Value), + Type: ethtypes.Eip1559TxType, + Gas: ethtypes.EthUint64(msg.GasLimit), + MaxFeePerGas: ethtypes.EthBigInt(msg.GasFeeCap), + MaxPriorityFeePerGas: ethtypes.EthBigInt(msg.GasPremium), + AccessList: []ethtypes.EthHash{}, + } +} + +func getSignedMessage(ctx context.Context, cs *store.ChainStore, msgCid cid.Cid) (*types.SignedMessage, error) { + smsg, err := cs.GetSignedMessage(ctx, msgCid) + if err != nil { + // We couldn't find the signed message, it might be a BLS message, so search for a regular message. + msg, err := cs.GetMessage(ctx, msgCid) + if err != nil { + return nil, xerrors.Errorf("failed to find msg %s: %w", msgCid, err) + } + smsg = &types.SignedMessage{ + Message: *msg, + Signature: crypto.Signature{ + Type: crypto.SigTypeBLS, + }, + } + } + + return smsg, nil +} + +// newEthTxFromMessageLookup creates an ethereum transaction from filecoin message lookup. If a negative txIdx is passed +// into the function, it looks up the transaction index of the message in the tipset, otherwise it uses the txIdx passed into the +// function +func newEthTxFromMessageLookup(ctx context.Context, msgLookup *api.MsgLookup, txIdx int, cs *store.ChainStore, sa StateAPI) (ethtypes.EthTx, error) { + ts, err := cs.LoadTipSet(ctx, msgLookup.TipSet) + if err != nil { + return ethtypes.EthTx{}, err + } + + // This tx is located in the parent tipset + parentTs, err := cs.LoadTipSet(ctx, ts.Parents()) + if err != nil { + return ethtypes.EthTx{}, err + } + + parentTsCid, err := parentTs.Key().Cid() + if err != nil { + return ethtypes.EthTx{}, err + } + + // lookup the transactionIndex + if txIdx < 0 { + msgs, err := cs.MessagesForTipset(ctx, parentTs) + if err != nil { + return ethtypes.EthTx{}, err + } + for i, msg := range msgs { + if msg.Cid() == msgLookup.Message { + txIdx = i + break + } + } + if txIdx < 0 { + return ethtypes.EthTx{}, fmt.Errorf("cannot find the msg in the tipset") + } + } + + blkHash, err := ethtypes.EthHashFromCid(parentTsCid) + if err != nil { + return ethtypes.EthTx{}, err + } + + smsg, err := getSignedMessage(ctx, cs, msgLookup.Message) + if err != nil { + return ethtypes.EthTx{}, xerrors.Errorf("failed to get signed msg: %w", err) + } + + tx, err := newEthTxFromSignedMessage(ctx, smsg, sa) + if err != nil { + return ethtypes.EthTx{}, err + } + + var ( + bn = ethtypes.EthUint64(parentTs.Height()) + ti = ethtypes.EthUint64(txIdx) + ) + + tx.ChainID = ethtypes.EthUint64(build.Eip155ChainId) + tx.BlockHash = &blkHash + tx.BlockNumber = &bn + tx.TransactionIndex = &ti + return tx, nil +} + +func newEthTxReceipt(ctx context.Context, tx ethtypes.EthTx, lookup *api.MsgLookup, events []types.Event, cs *store.ChainStore, sa StateAPI) (api.EthTxReceipt, error) { + var ( + transactionIndex ethtypes.EthUint64 + blockHash ethtypes.EthHash + blockNumber ethtypes.EthUint64 + ) + + if tx.TransactionIndex != nil { + transactionIndex = *tx.TransactionIndex + } + if tx.BlockHash != nil { + blockHash = *tx.BlockHash + } + if tx.BlockNumber != nil { + blockNumber = *tx.BlockNumber + } + + receipt := api.EthTxReceipt{ + TransactionHash: tx.Hash, + From: tx.From, + To: tx.To, + TransactionIndex: transactionIndex, + BlockHash: blockHash, + BlockNumber: blockNumber, + Type: ethtypes.EthUint64(2), + Logs: []ethtypes.EthLog{}, // empty log array is compulsory when no logs, or libraries like ethers.js break + LogsBloom: ethtypes.EmptyEthBloom[:], + } + + if lookup.Receipt.ExitCode.IsSuccess() { + receipt.Status = 1 + } else { + receipt.Status = 0 + } + + receipt.GasUsed = ethtypes.EthUint64(lookup.Receipt.GasUsed) + + // TODO: handle CumulativeGasUsed + receipt.CumulativeGasUsed = ethtypes.EmptyEthInt + + // TODO: avoid loading the tipset twice (once here, once when we convert the message to a txn) + ts, err := cs.GetTipSetFromKey(ctx, lookup.TipSet) + if err != nil { + return api.EthTxReceipt{}, xerrors.Errorf("failed to lookup tipset %s when constructing the eth txn receipt: %w", lookup.TipSet, err) + } + + baseFee := ts.Blocks()[0].ParentBaseFee + gasOutputs := vm.ComputeGasOutputs(lookup.Receipt.GasUsed, int64(tx.Gas), baseFee, big.Int(tx.MaxFeePerGas), big.Int(tx.MaxPriorityFeePerGas), true) + totalSpent := big.Sum(gasOutputs.BaseFeeBurn, gasOutputs.MinerTip, gasOutputs.OverEstimationBurn) + + effectiveGasPrice := big.Zero() + if lookup.Receipt.GasUsed > 0 { + effectiveGasPrice = big.Div(totalSpent, big.NewInt(lookup.Receipt.GasUsed)) + } + receipt.EffectiveGasPrice = ethtypes.EthBigInt(effectiveGasPrice) + + if receipt.To == nil && lookup.Receipt.ExitCode.IsSuccess() { + // Create and Create2 return the same things. + var ret eam.CreateExternalReturn + if err := ret.UnmarshalCBOR(bytes.NewReader(lookup.Receipt.Return)); err != nil { + return api.EthTxReceipt{}, xerrors.Errorf("failed to parse contract creation result: %w", err) + } + addr := ethtypes.EthAddress(ret.EthAddress) + receipt.ContractAddress = &addr + } + + if len(events) > 0 { + receipt.Logs = make([]ethtypes.EthLog, 0, len(events)) + for i, evt := range events { + l := ethtypes.EthLog{ + Removed: false, + LogIndex: ethtypes.EthUint64(i), + TransactionHash: tx.Hash, + TransactionIndex: transactionIndex, + BlockHash: blockHash, + BlockNumber: blockNumber, + } + + data, topics, ok := ethLogFromEvent(evt.Entries) + if !ok { + // not an eth event. + continue + } + for _, topic := range topics { + ethtypes.EthBloomSet(receipt.LogsBloom, topic[:]) + } + l.Data = data + l.Topics = topics + + addr, err := address.NewIDAddress(uint64(evt.Emitter)) + if err != nil { + return api.EthTxReceipt{}, xerrors.Errorf("failed to create ID address: %w", err) + } + + l.Address, err = lookupEthAddress(ctx, addr, sa) + if err != nil { + return api.EthTxReceipt{}, xerrors.Errorf("failed to resolve Ethereum address: %w", err) + } + + ethtypes.EthBloomSet(receipt.LogsBloom, l.Address[:]) + receipt.Logs = append(receipt.Logs, l) + } + } + + return receipt, nil +} diff --git a/node/impl/full/txhashmanager.go b/node/impl/full/txhashmanager.go new file mode 100644 index 00000000000..6757cc6dd92 --- /dev/null +++ b/node/impl/full/txhashmanager.go @@ -0,0 +1,129 @@ +package full + +import ( + "context" + "time" + + "github.com/filecoin-project/go-state-types/abi" + "github.com/filecoin-project/go-state-types/crypto" + + "github.com/filecoin-project/lotus/api" + "github.com/filecoin-project/lotus/build" + "github.com/filecoin-project/lotus/chain/ethhashlookup" + "github.com/filecoin-project/lotus/chain/types" +) + +type EthTxHashManager struct { + StateAPI StateAPI + TransactionHashLookup *ethhashlookup.EthTxHashLookup +} + +func (m *EthTxHashManager) Revert(ctx context.Context, from, to *types.TipSet) error { + return nil +} + +func (m *EthTxHashManager) PopulateExistingMappings(ctx context.Context, minHeight abi.ChainEpoch) error { + if minHeight < build.UpgradeHyggeHeight { + minHeight = build.UpgradeHyggeHeight + } + + ts := m.StateAPI.Chain.GetHeaviestTipSet() + for ts.Height() > minHeight { + 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 { + m.ProcessSignedMessage(ctx, msg) + } + } + + var err error + ts, err = m.StateAPI.Chain.GetTipSetFromKey(ctx, ts.Parents()) + if err != nil { + return err + } + } + + return nil +} + +func (m *EthTxHashManager) Apply(ctx context.Context, from, to *types.TipSet) error { + for _, blk := range to.Blocks() { + _, smsgs, err := m.StateAPI.Chain.MessagesForBlock(ctx, blk) + if err != nil { + return err + } + + for _, smsg := range smsgs { + if smsg.Signature.Type != crypto.SigTypeDelegated { + continue + } + + hash, err := ethTxHashFromSignedMessage(ctx, smsg, m.StateAPI) + if err != nil { + return err + } + + err = m.TransactionHashLookup.UpsertHash(hash, smsg.Cid()) + if err != nil { + return err + } + } + } + + return nil +} + +func (m *EthTxHashManager) ProcessSignedMessage(ctx context.Context, msg *types.SignedMessage) { + if msg.Signature.Type != crypto.SigTypeDelegated { + return + } + + ethTx, err := newEthTxFromSignedMessage(ctx, msg, m.StateAPI) + if err != nil { + log.Errorf("error converting filecoin message to eth tx: %s", err) + return + } + + err = m.TransactionHashLookup.UpsertHash(ethTx.Hash, msg.Cid()) + if err != nil { + log.Errorf("error inserting tx mapping to db: %s", err) + return + } +} + +func WaitForMpoolUpdates(ctx context.Context, ch <-chan api.MpoolUpdate, manager *EthTxHashManager) { + for { + select { + case <-ctx.Done(): + return + case u := <-ch: + if u.Type != api.MpoolAdd { + continue + } + + manager.ProcessSignedMessage(ctx, u.Message) + } + } +} + +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) + } +} diff --git a/node/modules/trace.go b/node/modules/trace.go deleted file mode 100644 index aea7fc02f72..00000000000 --- a/node/modules/trace.go +++ /dev/null @@ -1,19 +0,0 @@ -package modules - -import ( - "github.com/filecoin-project/lotus/chain/stmgr" - "github.com/filecoin-project/lotus/chain/store" - "github.com/filecoin-project/lotus/node/impl/full" -) - -func EthTraceAPI() func(*store.ChainStore, *stmgr.StateManager, full.EthModuleAPI, full.ChainAPI) (*full.EthTrace, error) { - return func(cs *store.ChainStore, sm *stmgr.StateManager, evapi full.EthModuleAPI, chainapi full.ChainAPI) (*full.EthTrace, error) { - return &full.EthTrace{ - Chain: cs, - StateManager: sm, - - ChainAPI: chainapi, - EthModuleAPI: evapi, - }, nil - } -}