Merge pull request #10184 from filecoin-project/asr/merge-release-into-master

chore: merge release/v1.20.0 into master
2023-02-09 16:41:58 +01:00
parent 8066833b3a 693f230fbb
commit cb8e2836bb
104 changed files with 9790 additions and 2452 deletions
@@ -127,6 +127,8 @@ const (

 	SetApiEndpointKey

+	StoreEventsKey
+
 	_nInvokes // keep this last
 )

@@ -219,6 +219,11 @@ func ConfigFullNode(c interface{}) Option {
 			Override(SetupFallbackBlockstoresKey, modules.InitFallbackBlockstores),
 		),

+		// If the Eth JSON-RPC is enabled, enable storing events at the ChainStore.
+		// This is the case even if real-time and historic filtering are disabled,
+		// as it enables us to serve logs in eth_getTransactionReceipt.
+		If(cfg.Fevm.EnableEthRPC, Override(StoreEventsKey, modules.EnableStoringEvents)),
+
 		Override(new(dtypes.ClientImportMgr), modules.ClientImportMgr),

 		Override(new(dtypes.ClientBlockstore), modules.ClientBlockstore),
@@ -258,11 +263,18 @@ func ConfigFullNode(c interface{}) Option {

 		// Actor event filtering support
 		Override(new(events.EventAPI), From(new(modules.EventAPI))),
-		// in lite-mode Eth event api is provided by gateway
-		ApplyIf(isFullNode, Override(new(full.EthEventAPI), modules.EthEventAPI(cfg.Fevm))),

-		If(cfg.Fevm.EnableEthRPC, Override(new(full.EthModuleAPI), modules.EthModuleAPI(cfg.Fevm))),
-		If(!cfg.Fevm.EnableEthRPC, Override(new(full.EthModuleAPI), &full.EthModuleDummy{})),
+		// in lite-mode Eth api is provided by gateway
+		ApplyIf(isFullNode,
+			If(cfg.Fevm.EnableEthRPC,
+				Override(new(full.EthModuleAPI), modules.EthModuleAPI(cfg.Fevm)),
+				Override(new(full.EthEventAPI), modules.EthEventAPI(cfg.Fevm)),
+			),
+			If(!cfg.Fevm.EnableEthRPC,
+				Override(new(full.EthModuleAPI), &full.EthModuleDummy{}),
+				Override(new(full.EthEventAPI), &full.EthModuleDummy{}),
+			),
+		),
 	)
 }

@@ -6,6 +6,8 @@ import (

 	"github.com/ipfs/go-cid"

+	"github.com/filecoin-project/go-jsonrpc"
+
 	"github.com/filecoin-project/lotus/api"
 	"github.com/filecoin-project/lotus/chain/types/ethtypes"
 )
@@ -118,4 +120,45 @@ func (e *EthModuleDummy) EthSendRawTransaction(ctx context.Context, rawTx ethtyp
 	return ethtypes.EthHash{}, ErrModuleDisabled
 }

+func (e *EthModuleDummy) Web3ClientVersion(ctx context.Context) (string, error) {
+	return "", ErrModuleDisabled
+}
+
+func (e *EthModuleDummy) EthGetLogs(ctx context.Context, filter *ethtypes.EthFilterSpec) (*ethtypes.EthFilterResult, error) {
+	return &ethtypes.EthFilterResult{}, ErrModuleDisabled
+}
+
+func (e *EthModuleDummy) EthGetFilterChanges(ctx context.Context, id ethtypes.EthFilterID) (*ethtypes.EthFilterResult, error) {
+	return &ethtypes.EthFilterResult{}, ErrModuleDisabled
+}
+
+func (e *EthModuleDummy) EthGetFilterLogs(ctx context.Context, id ethtypes.EthFilterID) (*ethtypes.EthFilterResult, error) {
+	return &ethtypes.EthFilterResult{}, ErrModuleDisabled
+}
+
+func (e *EthModuleDummy) EthNewFilter(ctx context.Context, filter *ethtypes.EthFilterSpec) (ethtypes.EthFilterID, error) {
+	return ethtypes.EthFilterID{}, ErrModuleDisabled
+}
+
+func (e *EthModuleDummy) EthNewBlockFilter(ctx context.Context) (ethtypes.EthFilterID, error) {
+	return ethtypes.EthFilterID{}, ErrModuleDisabled
+}
+
+func (e *EthModuleDummy) EthNewPendingTransactionFilter(ctx context.Context) (ethtypes.EthFilterID, error) {
+	return ethtypes.EthFilterID{}, ErrModuleDisabled
+}
+
+func (e *EthModuleDummy) EthUninstallFilter(ctx context.Context, id ethtypes.EthFilterID) (bool, error) {
+	return false, ErrModuleDisabled
+}
+
+func (e *EthModuleDummy) EthSubscribe(ctx context.Context, params jsonrpc.RawParams) (ethtypes.EthSubscriptionID, error) {
+	return ethtypes.EthSubscriptionID{}, ErrModuleDisabled
+}
+
+func (e *EthModuleDummy) EthUnsubscribe(ctx context.Context, id ethtypes.EthSubscriptionID) (bool, error) {
+	return false, ErrModuleDisabled
+}
+
 var _ EthModuleAPI = &EthModuleDummy{}
+var _ EthEventAPI = &EthModuleDummy{}
@@ -3,6 +3,7 @@ package full
 import (
 	"bytes"
 	"context"
+	"encoding/json"
 	"errors"
 	"fmt"
 	"strconv"
@@ -16,12 +17,14 @@ import (
 	"golang.org/x/xerrors"

 	"github.com/filecoin-project/go-address"
+	"github.com/filecoin-project/go-jsonrpc"
 	"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"
 	"github.com/filecoin-project/lotus/build"
@@ -64,6 +67,7 @@ type EthModuleAPI interface {
 	EthCall(ctx context.Context, tx ethtypes.EthCall, blkParam string) (ethtypes.EthBytes, error)
 	EthMaxPriorityFeePerGas(ctx context.Context) (ethtypes.EthBigInt, error)
 	EthSendRawTransaction(ctx context.Context, rawTx ethtypes.EthBytes) (ethtypes.EthHash, error)
+	Web3ClientVersion(ctx context.Context) (string, error)
 }

 type EthEventAPI interface {
@@ -74,13 +78,15 @@ type EthEventAPI interface {
 	EthNewBlockFilter(ctx context.Context) (ethtypes.EthFilterID, error)
 	EthNewPendingTransactionFilter(ctx context.Context) (ethtypes.EthFilterID, error)
 	EthUninstallFilter(ctx context.Context, id ethtypes.EthFilterID) (bool, error)
-	EthSubscribe(ctx context.Context, eventType string, params *ethtypes.EthSubscriptionParams) (<-chan ethtypes.EthSubscriptionResponse, error)
+	EthSubscribe(ctx context.Context, params jsonrpc.RawParams) (ethtypes.EthSubscriptionID, error)
 	EthUnsubscribe(ctx context.Context, id ethtypes.EthSubscriptionID) (bool, error)
 }

 var (
 	_ EthModuleAPI = *new(api.FullNode)
 	_ EthEventAPI  = *new(api.FullNode)
+
+	_ EthModuleAPI = *new(api.Gateway)
 )

 // EthModule provides the default implementation of the standard Ethereum JSON-RPC API.
@@ -131,6 +137,7 @@ type EthEvent struct {
 	FilterStore          filter.FilterStore
 	SubManager           *EthSubscriptionManager
 	MaxFilterHeightRange abi.ChainEpoch
+	SubscribtionCtx      context.Context
 }

 var _ EthEventAPI = (*EthEvent)(nil)
@@ -211,7 +218,7 @@ func (a *EthModule) EthGetBlockByHash(ctx context.Context, blkHash ethtypes.EthH
 	if err != nil {
 		return ethtypes.EthBlock{}, xerrors.Errorf("error loading tipset %s: %w", ts, err)
 	}
-	return newEthBlockFromFilecoinTipSet(ctx, ts, fullTxInfo, a.Chain, a.ChainAPI, a.StateAPI)
+	return newEthBlockFromFilecoinTipSet(ctx, ts, fullTxInfo, a.Chain, a.StateAPI)
 }

 func (a *EthModule) parseBlkParam(ctx context.Context, blkParam string) (tipset *types.TipSet, err error) {
@@ -248,7 +255,7 @@ func (a *EthModule) EthGetBlockByNumber(ctx context.Context, blkParam string, fu
 	if err != nil {
 		return ethtypes.EthBlock{}, err
 	}
-	return newEthBlockFromFilecoinTipSet(ctx, ts, fullTxInfo, a.Chain, a.ChainAPI, a.StateAPI)
+	return newEthBlockFromFilecoinTipSet(ctx, ts, fullTxInfo, a.Chain, a.StateAPI)
 }

 func (a *EthModule) EthGetTransactionByHash(ctx context.Context, txHash *ethtypes.EthHash) (*ethtypes.EthTx, error) {
@@ -269,8 +276,8 @@ func (a *EthModule) EthGetTransactionByHash(ctx context.Context, txHash *ethtype

 	// first, try to get the cid from mined transactions
 	msgLookup, err := a.StateAPI.StateSearchMsg(ctx, types.EmptyTSK, c, api.LookbackNoLimit, true)
-	if err == nil {
-		tx, err := newEthTxFromFilecoinMessageLookup(ctx, msgLookup, -1, a.Chain, a.StateAPI)
+	if err == nil && msgLookup != nil {
+		tx, err := newEthTxFromMessageLookup(ctx, msgLookup, -1, a.Chain, a.StateAPI)
 		if err == nil {
 			return &tx, nil
 		}
@@ -286,7 +293,7 @@ func (a *EthModule) EthGetTransactionByHash(ctx context.Context, txHash *ethtype

 	for _, p := range pending {
 		if p.Cid() == c {
-			tx, err := NewEthTxFromFilecoinMessage(ctx, p, a.StateAPI)
+			tx, err := newEthTxFromSignedMessage(ctx, p, a.StateAPI)
 			if err != nil {
 				return nil, fmt.Errorf("could not convert Filecoin message into tx: %s", err)
 			}
@@ -335,7 +342,7 @@ func (a *EthModule) EthGetMessageCidByTransactionHash(ctx context.Context, txHas
 }

 func (a *EthModule) EthGetTransactionHashByCid(ctx context.Context, cid cid.Cid) (*ethtypes.EthHash, error) {
-	hash, err := EthTxHashFromFilecoinMessageCid(ctx, cid, a.StateAPI)
+	hash, err := EthTxHashFromMessageCid(ctx, cid, a.StateAPI)
 	if hash == ethtypes.EmptyEthHash {
 		// not found
 		return nil, nil
@@ -378,7 +385,7 @@ func (a *EthModule) EthGetTransactionReceipt(ctx context.Context, txHash ethtype
 		return nil, nil
 	}

-	tx, err := newEthTxFromFilecoinMessageLookup(ctx, msgLookup, -1, a.Chain, a.StateAPI)
+	tx, err := newEthTxFromMessageLookup(ctx, msgLookup, -1, a.Chain, a.StateAPI)
 	if err != nil {
 		return nil, nil
 	}
@@ -440,6 +447,11 @@ func (a *EthModule) EthGetCode(ctx context.Context, ethAddr ethtypes.EthAddress,
 		return nil, xerrors.Errorf("cannot parse block param: %s", blkParam)
 	}

+	// StateManager.Call will panic if there is no parent
+	if ts.Height() == 0 {
+		return nil, xerrors.Errorf("block param must not specify genesis block")
+	}
+
 	// Try calling until we find a height with no migration.
 	var res *api.InvocResult
 	for {
@@ -500,18 +512,8 @@ func (a *EthModule) EthGetStorageAt(ctx context.Context, ethAddr ethtypes.EthAdd
 		return nil, fmt.Errorf("failed to construct system sender address: %w", err)
 	}

-	// TODO super duper hack (raulk). The EVM runtime actor uses the U256 parameter type in
-	//  GetStorageAtParams, which serializes as a hex-encoded string. It should serialize
-	//  as bytes. We didn't get to fix in time for Iron, so for now we just pass
-	//  through the hex-encoded value passed through the Eth JSON-RPC API, by remarshalling it.
-	//  We don't fix this at origin (builtin-actors) because we are not updating the bundle
-	//  for Iron.
-	tmp, err := position.MarshalJSON()
-	if err != nil {
-		panic(err)
-	}
 	params, err := actors.SerializeParams(&evm.GetStorageAtParams{
-		StorageKey: tmp[1 : len(tmp)-1], // TODO strip the JSON-encoding quotes -- yuck
+		StorageKey: position,
 	})
 	if err != nil {
 		return nil, fmt.Errorf("failed to serialize parameters: %w", err)
@@ -614,7 +616,7 @@ func (a *EthModule) EthFeeHistory(ctx context.Context, blkCount ethtypes.EthUint
 	for ts.Height() >= abi.ChainEpoch(oldestBlkHeight) {
 		// Unfortunately we need to rebuild the full message view so we can
 		// totalize gas used in the tipset.
-		block, err := newEthBlockFromFilecoinTipSet(ctx, ts, false, a.Chain, a.ChainAPI, a.StateAPI)
+		block, err := newEthBlockFromFilecoinTipSet(ctx, ts, false, a.Chain, a.StateAPI)
 		if err != nil {
 			return ethtypes.EthFeeHistory{}, fmt.Errorf("cannot create eth block: %v", err)
 		}
@@ -640,7 +642,7 @@ func (a *EthModule) EthFeeHistory(ctx context.Context, blkCount ethtypes.EthUint
 	}

 	return ethtypes.EthFeeHistory{
-		OldestBlock:   oldestBlkHeight,
+		OldestBlock:   ethtypes.EthUint64(oldestBlkHeight),
 		BaseFeePerGas: baseFeeArray,
 		GasUsedRatio:  gasUsedRatioArray,
 	}, nil
@@ -699,13 +701,6 @@ func (a *EthModule) EthSendRawTransaction(ctx context.Context, rawTx ethtypes.Et
 		return ethtypes.EmptyEthHash, err
 	}

-	_, err = a.StateAPI.StateGetActor(ctx, smsg.Message.To, types.EmptyTSK)
-	if err != nil {
-		// if actor does not exist on chain yet, set the method to 0 because
-		// placeholders only implement method 0
-		smsg.Message.Method = builtinactors.MethodSend
-	}
-
 	_, err = a.MpoolAPI.MpoolPush(ctx, smsg)
 	if err != nil {
 		return ethtypes.EmptyEthHash, err
@@ -714,6 +709,10 @@ func (a *EthModule) EthSendRawTransaction(ctx context.Context, rawTx ethtypes.Et
 	return ethtypes.EthHashFromTxBytes(rawTx), nil
 }

+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{}) {
@@ -816,19 +815,145 @@ func (a *EthModule) EthEstimateGas(ctx context.Context, tx ethtypes.EthCall) (et
 	// gas estimation actually run.
 	msg.GasLimit = 0

-	msg, err = a.GasAPI.GasEstimateMessageGas(ctx, msg, nil, types.EmptyTSK)
+	ts := a.Chain.GetHeaviestTipSet()
+	msg, err = a.GasAPI.GasEstimateMessageGas(ctx, msg, nil, ts.Key())
 	if err != nil {
-		return ethtypes.EthUint64(0), err
+		return ethtypes.EthUint64(0), xerrors.Errorf("failed to estimate gas: %w", err)
 	}

-	return ethtypes.EthUint64(msg.GasLimit), nil
+	expectedGas, err := ethGasSearch(ctx, a.Chain, a.Stmgr, a.Mpool, msg, ts)
+	if err != nil {
+		log.Errorw("expected gas", "err", err)
+	}
+
+	return ethtypes.EthUint64(expectedGas), nil
+}
+
+// gasSearch does an exponential search to find a gas value to execute the
+// message with. It first finds a high gas limit that allows the message to execute
+// by doubling the previous gas limit until it succeeds then does a binary
+// search till it gets within a range of 1%
+func gasSearch(
+	ctx context.Context,
+	smgr *stmgr.StateManager,
+	msgIn *types.Message,
+	priorMsgs []types.ChainMsg,
+	ts *types.TipSet,
+) (int64, error) {
+	msg := *msgIn
+
+	high := msg.GasLimit
+	low := msg.GasLimit
+
+	canSucceed := func(limit int64) (bool, error) {
+		msg.GasLimit = limit
+
+		res, err := smgr.CallWithGas(ctx, &msg, priorMsgs, ts)
+		if err != nil {
+			return false, xerrors.Errorf("CallWithGas failed: %w", err)
+		}
+
+		if res.MsgRct.ExitCode.IsSuccess() {
+			return true, nil
+		}
+
+		return false, nil
+	}
+
+	for {
+		ok, err := canSucceed(high)
+		if err != nil {
+			return -1, xerrors.Errorf("searching for high gas limit failed: %w", err)
+		}
+		if ok {
+			break
+		}
+
+		low = high
+		high = high * 2
+
+		if high > build.BlockGasLimit {
+			high = build.BlockGasLimit
+			break
+		}
+	}
+
+	checkThreshold := high / 100
+	for (high - low) > checkThreshold {
+		median := (low + high) / 2
+		ok, err := canSucceed(median)
+		if err != nil {
+			return -1, xerrors.Errorf("searching for optimal gas limit failed: %w", err)
+		}
+
+		if ok {
+			high = median
+		} else {
+			low = median
+		}
+
+		checkThreshold = median / 100
+	}
+
+	return high, nil
+}
+
+func traceContainsExitCode(et types.ExecutionTrace, ex exitcode.ExitCode) bool {
+	if et.MsgRct.ExitCode == ex {
+		return true
+	}
+
+	for _, et := range et.Subcalls {
+		if traceContainsExitCode(et, ex) {
+			return true
+		}
+	}
+
+	return false
+}
+
+// ethGasSearch executes a message for gas estimation using the previously estimated gas.
+// If the message fails due to an out of gas error then a gas search is performed.
+// See gasSearch.
+func ethGasSearch(
+	ctx context.Context,
+	cstore *store.ChainStore,
+	smgr *stmgr.StateManager,
+	mpool *messagepool.MessagePool,
+	msgIn *types.Message,
+	ts *types.TipSet,
+) (int64, error) {
+	msg := *msgIn
+	currTs := ts
+
+	res, priorMsgs, ts, err := gasEstimateCallWithGas(ctx, cstore, smgr, mpool, &msg, currTs)
+	if err != nil {
+		return -1, xerrors.Errorf("gas estimation failed: %w", err)
+	}
+
+	if res.MsgRct.ExitCode.IsSuccess() {
+		return msg.GasLimit, nil
+	}
+
+	if traceContainsExitCode(res.ExecutionTrace, exitcode.SysErrOutOfGas) {
+		ret, err := gasSearch(ctx, smgr, &msg, priorMsgs, ts)
+		if err != nil {
+			return -1, xerrors.Errorf("gas estimation search failed: %w", err)
+		}
+
+		ret = int64(float64(ret) * mpool.GetConfig().GasLimitOverestimation)
+		return ret, nil
+	}
+
+	return -1, xerrors.Errorf("message execution failed: exit %s, reason: %s", res.MsgRct.ExitCode, res.Error)
 }

 func (a *EthModule) EthCall(ctx context.Context, tx ethtypes.EthCall, blkParam string) (ethtypes.EthBytes, error) {
 	msg, err := a.ethCallToFilecoinMessage(ctx, tx)
 	if err != nil {
-		return nil, err
+		return nil, xerrors.Errorf("failed to convert ethcall to filecoin message: %w", err)
 	}
+
 	ts, err := a.parseBlkParam(ctx, blkParam)
 	if err != nil {
 		return nil, xerrors.Errorf("cannot parse block param: %s", blkParam)
@@ -836,11 +961,16 @@ func (a *EthModule) EthCall(ctx context.Context, tx ethtypes.EthCall, blkParam s

 	invokeResult, err := a.applyMessage(ctx, msg, ts.Key())
 	if err != nil {
-		return nil, err
+		return nil, xerrors.Errorf("failed to apply message: %w", err)
 	}
-	if len(invokeResult.MsgRct.Return) > 0 {
+
+	if msg.To == builtintypes.EthereumAddressManagerActorAddr {
+		// As far as I can tell, the Eth API always returns empty on contract deployment
+		return ethtypes.EthBytes{}, nil
+	} else if len(invokeResult.MsgRct.Return) > 0 {
 		return cbg.ReadByteArray(bytes.NewReader(invokeResult.MsgRct.Return), uint64(len(invokeResult.MsgRct.Return)))
 	}
+
 	return ethtypes.EthBytes{}, nil
 }

@@ -980,17 +1110,9 @@ func (e *EthEvent) installEthFilterSpec(ctx context.Context, filterSpec *ethtype
 		addresses = append(addresses, a)
 	}

-	for idx, vals := range filterSpec.Topics {
-		if len(vals) == 0 {
-			continue
-		}
-		// Ethereum topics are emitted using `LOG{0..4}` opcodes resulting in topics1..4
-		key := fmt.Sprintf("topic%d", idx+1)
-		for _, v := range vals {
-			buf := make([]byte, len(v[:]))
-			copy(buf, v[:])
-			keys[key] = append(keys[key], buf)
-		}
+	keys, err := parseEthTopics(filterSpec.Topics)
+	if err != nil {
+		return nil, err
 	}

 	return e.EventFilterManager.Install(ctx, minHeight, maxHeight, tipsetCid, addresses, keys)
@@ -1015,7 +1137,6 @@ func (e *EthEvent) EthNewFilter(ctx context.Context, filterSpec *ethtypes.EthFil

 		return ethtypes.EthFilterID{}, err
 	}
-
 	return ethtypes.EthFilterID(f.ID()), nil
 }

@@ -1114,54 +1235,71 @@ const (
 	EthSubscribeEventTypeLogs  = "logs"
 )

-func (e *EthEvent) EthSubscribe(ctx context.Context, eventType string, params *ethtypes.EthSubscriptionParams) (<-chan ethtypes.EthSubscriptionResponse, error) {
-	if e.SubManager == nil {
-		return nil, api.ErrNotSupported
-	}
-	// Note that go-jsonrpc will set the method field of the response to "xrpc.ch.val" but the ethereum api expects the name of the
-	// method to be "eth_subscription". This probably doesn't matter in practice.
-
-	sub, err := e.SubManager.StartSubscription(ctx)
+func (e *EthEvent) EthSubscribe(ctx context.Context, p jsonrpc.RawParams) (ethtypes.EthSubscriptionID, error) {
+	params, err := jsonrpc.DecodeParams[ethtypes.EthSubscribeParams](p)
 	if err != nil {
-		return nil, err
+		return ethtypes.EthSubscriptionID{}, xerrors.Errorf("decoding params: %w", err)
 	}

-	switch eventType {
+	if e.SubManager == nil {
+		return ethtypes.EthSubscriptionID{}, api.ErrNotSupported
+	}
+
+	ethCb, ok := jsonrpc.ExtractReverseClient[api.EthSubscriberMethods](ctx)
+	if !ok {
+		return ethtypes.EthSubscriptionID{}, xerrors.Errorf("connection doesn't support callbacks")
+	}
+
+	sub, err := e.SubManager.StartSubscription(e.SubscribtionCtx, ethCb.EthSubscription)
+	if err != nil {
+		return ethtypes.EthSubscriptionID{}, err
+	}
+
+	switch params.EventType {
 	case EthSubscribeEventTypeHeads:
 		f, err := e.TipSetFilterManager.Install(ctx)
 		if err != nil {
 			// clean up any previous filters added and stop the sub
 			_, _ = e.EthUnsubscribe(ctx, sub.id)
-			return nil, err
+			return ethtypes.EthSubscriptionID{}, err
 		}
 		sub.addFilter(ctx, f)

 	case EthSubscribeEventTypeLogs:
 		keys := map[string][][]byte{}
-		if params != nil {
-			for idx, vals := range params.Topics {
-				// Ethereum topics are emitted using `LOG{0..4}` opcodes resulting in topics1..4
-				key := fmt.Sprintf("topic%d", idx+1)
-				keyvals := make([][]byte, len(vals))
-				for i, v := range vals {
-					keyvals[i] = v[:]
-				}
-				keys[key] = keyvals
+		if params.Params != nil {
+			var err error
+			keys, err = parseEthTopics(params.Params.Topics)
+			if err != nil {
+				// clean up any previous filters added and stop the sub
+				_, _ = e.EthUnsubscribe(ctx, sub.id)
+				return ethtypes.EthSubscriptionID{}, err
 			}
 		}

-		f, err := e.EventFilterManager.Install(ctx, -1, -1, cid.Undef, []address.Address{}, keys)
+		var addresses []address.Address
+		if params.Params != nil {
+			for _, ea := range params.Params.Address {
+				a, err := ea.ToFilecoinAddress()
+				if err != nil {
+					return ethtypes.EthSubscriptionID{}, xerrors.Errorf("invalid address %x", ea)
+				}
+				addresses = append(addresses, a)
+			}
+		}
+
+		f, err := e.EventFilterManager.Install(ctx, -1, -1, cid.Undef, addresses, keys)
 		if err != nil {
 			// clean up any previous filters added and stop the sub
 			_, _ = e.EthUnsubscribe(ctx, sub.id)
-			return nil, err
+			return ethtypes.EthSubscriptionID{}, err
 		}
 		sub.addFilter(ctx, f)
 	default:
-		return nil, xerrors.Errorf("unsupported event type: %s", eventType)
+		return ethtypes.EthSubscriptionID{}, xerrors.Errorf("unsupported event type: %s", params.EventType)
 	}

-	return sub.out, nil
+	return sub.id, nil
 }

 func (e *EthEvent) EthUnsubscribe(ctx context.Context, id ethtypes.EthSubscriptionID) (bool, error) {
@@ -1233,7 +1371,10 @@ func ethFilterResultFromEvents(evs []*filter.CollectedEvent, sa StateAPI) (*etht
 		var err error

 		for _, entry := range ev.Entries {
-			value := ethtypes.EthBytes(leftpad32(entry.Value)) // value has already been cbor-decoded but see https://github.com/filecoin-project/ref-fvm/issues/1345
+			value, err := cborDecodeTopicValue(entry.Value)
+			if err != nil {
+				return nil, err
+			}
 			if entry.Key == ethtypes.EthTopic1 || entry.Key == ethtypes.EthTopic2 || entry.Key == ethtypes.EthTopic3 || entry.Key == ethtypes.EthTopic4 {
 				log.Topics = append(log.Topics, value)
 			} else {
@@ -1246,7 +1387,7 @@ func ethFilterResultFromEvents(evs []*filter.CollectedEvent, sa StateAPI) (*etht
 			return nil, err
 		}

-		log.TransactionHash, err = EthTxHashFromFilecoinMessageCid(context.TODO(), ev.MsgCid, sa)
+		log.TransactionHash, err = EthTxHashFromMessageCid(context.TODO(), ev.MsgCid, sa)
 		if err != nil {
 			return nil, err
 		}
@@ -1288,7 +1429,7 @@ func ethFilterResultFromMessages(cs []*types.SignedMessage, sa StateAPI) (*ethty
 	res := &ethtypes.EthFilterResult{}

 	for _, c := range cs {
-		hash, err := EthTxHashFromSignedFilecoinMessage(context.TODO(), c, sa)
+		hash, err := EthTxHashFromSignedMessage(context.TODO(), c, sa)
 		if err != nil {
 			return nil, err
 		}
@@ -1307,7 +1448,7 @@ type EthSubscriptionManager struct {
 	subs     map[ethtypes.EthSubscriptionID]*ethSubscription
 }

-func (e *EthSubscriptionManager) StartSubscription(ctx context.Context) (*ethSubscription, error) { // nolint
+func (e *EthSubscriptionManager) StartSubscription(ctx context.Context, out ethSubscriptionCallback) (*ethSubscription, error) { // nolint
 	rawid, err := uuid.NewRandom()
 	if err != nil {
 		return nil, xerrors.Errorf("new uuid: %w", err)
@@ -1323,7 +1464,7 @@ func (e *EthSubscriptionManager) StartSubscription(ctx context.Context) (*ethSub
 		ChainAPI: e.ChainAPI,
 		id:       id,
 		in:       make(chan interface{}, 200),
-		out:      make(chan ethtypes.EthSubscriptionResponse, 20),
+		out:      out,
 		quit:     quit,
 	}

@@ -1353,13 +1494,15 @@ func (e *EthSubscriptionManager) StopSubscription(ctx context.Context, id ethtyp
 	return sub.filters, nil
 }

+type ethSubscriptionCallback func(context.Context, jsonrpc.RawParams) error
+
 type ethSubscription struct {
 	Chain    *store.ChainStore
 	StateAPI StateAPI
 	ChainAPI ChainAPI
 	id       ethtypes.EthSubscriptionID
 	in       chan interface{}
-	out      chan ethtypes.EthSubscriptionResponse
+	out      ethSubscriptionCallback

 	mu      sync.Mutex
 	filters []filter.Filter
@@ -1389,7 +1532,7 @@ func (e *ethSubscription) start(ctx context.Context) {
 			case *filter.CollectedEvent:
 				resp.Result, err = ethFilterResultFromEvents([]*filter.CollectedEvent{vt}, e.StateAPI)
 			case *types.TipSet:
-				eb, err := newEthBlockFromFilecoinTipSet(ctx, vt, true, e.Chain, e.ChainAPI, e.StateAPI)
+				eb, err := newEthBlockFromFilecoinTipSet(ctx, vt, true, e.Chain, e.StateAPI)
 				if err != nil {
 					break
 				}
@@ -1403,10 +1546,15 @@ func (e *ethSubscription) start(ctx context.Context) {
 				continue
 			}

-			select {
-			case e.out <- resp:
-			default:
-				// Skip if client is not reading responses
+			outParam, err := json.Marshal(resp)
+			if err != nil {
+				log.Warnw("marshaling subscription response", "sub", e.id, "error", err)
+				continue
+			}
+
+			if err := e.out(ctx, outParam); err != nil {
+				log.Warnw("sending subscription response", "sub", e.id, "error", err)
+				continue
 			}
 		}
 	}
@@ -1418,12 +1566,11 @@ func (e *ethSubscription) stop() {

 	if e.quit != nil {
 		e.quit()
-		close(e.out)
 		e.quit = nil
 	}
 }

-func newEthBlockFromFilecoinTipSet(ctx context.Context, ts *types.TipSet, fullTxInfo bool, cs *store.ChainStore, ca ChainAPI, sa StateAPI) (ethtypes.EthBlock, error) {
+func newEthBlockFromFilecoinTipSet(ctx context.Context, ts *types.TipSet, fullTxInfo bool, cs *store.ChainStore, sa StateAPI) (ethtypes.EthBlock, error) {
 	parent, err := cs.LoadTipSet(ctx, ts.Parents())
 	if err != nil {
 		return ethtypes.EthBlock{}, err
@@ -1451,7 +1598,7 @@ func newEthBlockFromFilecoinTipSet(ctx context.Context, ts *types.TipSet, fullTx
 		return ethtypes.EthBlock{}, xerrors.Errorf("error loading messages for tipset: %v: %w", ts, err)
 	}

-	block := ethtypes.NewEthBlock()
+	block := ethtypes.NewEthBlock(len(msgs) > 0)

 	// this seems to be a very expensive way to get gasUsed of the block. may need to find an efficient way to do it
 	gasUsed := int64(0)
@@ -1462,7 +1609,7 @@ func newEthBlockFromFilecoinTipSet(ctx context.Context, ts *types.TipSet, fullTx
 		}
 		gasUsed += msgLookup.Receipt.GasUsed

-		tx, err := newEthTxFromFilecoinMessageLookup(ctx, msgLookup, txIdx, cs, sa)
+		tx, err := newEthTxFromMessageLookup(ctx, msgLookup, txIdx, cs, sa)
 		if err != nil {
 			return ethtypes.EthBlock{}, nil
 		}
@@ -1522,11 +1669,11 @@ func lookupEthAddress(ctx context.Context, addr address.Address, sa StateAPI) (e
 	return ethtypes.EthAddressFromFilecoinAddress(idAddr)
 }

-func EthTxHashFromFilecoinMessageCid(ctx context.Context, c cid.Cid, sa StateAPI) (ethtypes.EthHash, error) {
+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 EthTxHashFromSignedFilecoinMessage(ctx, smsg, sa)
+		return EthTxHashFromSignedMessage(ctx, smsg, sa)
 	}

 	_, err = sa.Chain.GetMessage(ctx, c)
@@ -1538,93 +1685,51 @@ func EthTxHashFromFilecoinMessageCid(ctx context.Context, c cid.Cid, sa StateAPI
 	return ethtypes.EmptyEthHash, nil
 }

-func EthTxHashFromSignedFilecoinMessage(ctx context.Context, smsg *types.SignedMessage, sa StateAPI) (ethtypes.EthHash, error) {
+func EthTxHashFromSignedMessage(ctx context.Context, smsg *types.SignedMessage, sa StateAPI) (ethtypes.EthHash, error) {
 	if smsg.Signature.Type == crypto.SigTypeDelegated {
-		ethTx, err := NewEthTxFromFilecoinMessage(ctx, smsg, sa)
+		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())
 	}
-
-	return ethtypes.EthHashFromCid(smsg.Cid())
 }

-func NewEthTxFromFilecoinMessage(ctx context.Context, smsg *types.SignedMessage, sa StateAPI) (ethtypes.EthTx, error) {
-	// Ignore errors here so we can still parse non-eth messages
-	fromEthAddr, _ := lookupEthAddress(ctx, smsg.Message.From, sa)
-	toEthAddr, _ := lookupEthAddress(ctx, smsg.Message.To, sa)
-
-	toAddr := &toEthAddr
-	input := smsg.Message.Params
+func newEthTxFromSignedMessage(ctx context.Context, smsg *types.SignedMessage, sa StateAPI) (ethtypes.EthTx, error) {
+	var tx ethtypes.EthTx
 	var err error
-	// Check to see if we need to decode as contract deployment.
-	// We don't need to resolve the to address, because there's only one form (an ID).
-	if smsg.Message.To == builtintypes.EthereumAddressManagerActorAddr {
-		switch smsg.Message.Method {
-		case builtintypes.MethodsEAM.Create:
-			toAddr = nil
-			var params eam.CreateParams
-			err = params.UnmarshalCBOR(bytes.NewReader(smsg.Message.Params))
-			input = params.Initcode
-		case builtintypes.MethodsEAM.Create2:
-			toAddr = nil
-			var params eam.Create2Params
-			err = params.UnmarshalCBOR(bytes.NewReader(smsg.Message.Params))
-			input = params.Initcode
-		case builtintypes.MethodsEAM.CreateExternal:
-			toAddr = nil
-			var params abi.CborBytes
-			err = params.UnmarshalCBOR(bytes.NewReader(smsg.Message.Params))
-			input = []byte(params)
-		}
-		if err != nil {
-			return ethtypes.EthTx{}, err
-		}
-	}
-	// Otherwise, try to decode as a cbor byte array.
-	// TODO: Actually check if this is an ethereum call. This code will work for demo purposes, but is not correct.
-	if toAddr != nil {
-		if decodedParams, err := cbg.ReadByteArray(bytes.NewReader(smsg.Message.Params), uint64(len(smsg.Message.Params))); err == nil {
-			input = decodedParams
-		}
-	}
-
-	r, s, v, err := ethtypes.RecoverSignature(smsg.Signature)
-	if err != nil {
-		// we don't want to return error if the message is not an Eth tx
-		r, s, v = ethtypes.EthBigIntZero, ethtypes.EthBigIntZero, ethtypes.EthBigIntZero
-	}
-
-	tx := ethtypes.EthTx{
-		Nonce:                ethtypes.EthUint64(smsg.Message.Nonce),
-		ChainID:              ethtypes.EthUint64(build.Eip155ChainId),
-		From:                 fromEthAddr,
-		To:                   toAddr,
-		Value:                ethtypes.EthBigInt(smsg.Message.Value),
-		Type:                 ethtypes.EthUint64(2),
-		Input:                input,
-		Gas:                  ethtypes.EthUint64(smsg.Message.GasLimit),
-		MaxFeePerGas:         ethtypes.EthBigInt(smsg.Message.GasFeeCap),
-		MaxPriorityFeePerGas: ethtypes.EthBigInt(smsg.Message.GasPremium),
-		V:                    v,
-		R:                    r,
-		S:                    s,
-	}

 	// 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 tx, err
+			return ethtypes.EthTx{}, xerrors.Errorf("failed to calculate hash for ethTx: %w", err)
 		}
-	} else if smsg.Signature.Type == crypto.SigTypeUnknown { // BLS Filecoin message
-		tx.Hash, err = ethtypes.EthHashFromCid(smsg.Message.Cid())
+
+		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 { // Secp Filecoin Message
-		tx.Hash, err = ethtypes.EthHashFromCid(smsg.Cid())
+	} else { // BLS Filecoin message
+		tx = ethTxFromNativeMessage(ctx, smsg.VMMessage(), sa)
+		tx.Hash, err = ethtypes.EthHashFromCid(smsg.Message.Cid())
 		if err != nil {
 			return tx, err
 		}
@@ -1633,14 +1738,32 @@ func NewEthTxFromFilecoinMessage(ctx context.Context, smsg *types.SignedMessage,
 	return tx, nil
 }

-// newEthTxFromFilecoinMessageLookup creates an ethereum transaction from filecoin message lookup. If a negative txIdx is passed
-// into the function, it looksup the transaction index of the message in the tipset, otherwise it uses the txIdx passed into the
-// function
-func newEthTxFromFilecoinMessageLookup(ctx context.Context, msgLookup *api.MsgLookup, txIdx int, cs *store.ChainStore, sa StateAPI) (ethtypes.EthTx, error) {
-	if msgLookup == nil {
-		return ethtypes.EthTx{}, fmt.Errorf("msg does not exist")
+// 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),
 	}
+}

+// 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
@@ -1689,13 +1812,13 @@ func newEthTxFromFilecoinMessageLookup(ctx context.Context, msgLookup *api.MsgLo
 		smsg = &types.SignedMessage{
 			Message: *msg,
 			Signature: crypto.Signature{
-				Type: crypto.SigTypeUnknown,
+				Type: crypto.SigTypeBLS,
 				Data: nil,
 			},
 		}
 	}

-	tx, err := NewEthTxFromFilecoinMessage(ctx, smsg, sa)
+	tx, err := newEthTxFromSignedMessage(ctx, smsg, sa)
 	if err != nil {
 		return ethtypes.EthTx{}, err
 	}
@@ -1741,16 +1864,6 @@ func newEthTxReceipt(ctx context.Context, tx ethtypes.EthTx, lookup *api.MsgLook
 		LogsBloom:        ethtypes.EmptyEthBloom[:],
 	}

-	if receipt.To == nil && lookup.Receipt.ExitCode.IsSuccess() {
-		// Create and Create2 return the same things.
-		var ret eam.CreateReturn
-		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 lookup.Receipt.ExitCode.IsSuccess() {
 		receipt.Status = 1
 	}
@@ -1758,6 +1871,24 @@ func newEthTxReceipt(ctx context.Context, tx ethtypes.EthTx, lookup *api.MsgLook
 		receipt.Status = 0
 	}

+	receipt.GasUsed = ethtypes.EthUint64(lookup.Receipt.GasUsed)
+
+	// TODO: handle CumulativeGasUsed
+	receipt.CumulativeGasUsed = ethtypes.EmptyEthInt
+
+	effectiveGasPrice := big.Div(replay.GasCost.TotalCost, 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 {
 		// TODO return a dummy non-zero bloom to signal that there are logs
 		//  need to figure out how worth it is to populate with a real bloom
@@ -1776,7 +1907,10 @@ func newEthTxReceipt(ctx context.Context, tx ethtypes.EthTx, lookup *api.MsgLook
 			}

 			for _, entry := range evt.Entries {
-				value := ethtypes.EthBytes(leftpad32(entry.Value)) // value has already been cbor-decoded but see https://github.com/filecoin-project/ref-fvm/issues/1345
+				value, err := cborDecodeTopicValue(entry.Value)
+				if err != nil {
+					return api.EthTxReceipt{}, xerrors.Errorf("failed to decode event log value: %w", err)
+				}
 				if entry.Key == ethtypes.EthTopic1 || entry.Key == ethtypes.EthTopic2 || entry.Key == ethtypes.EthTopic3 || entry.Key == ethtypes.EthTopic4 {
 					l.Topics = append(l.Topics, value)
 				} else {
@@ -1798,14 +1932,6 @@ func newEthTxReceipt(ctx context.Context, tx ethtypes.EthTx, lookup *api.MsgLook
 		}
 	}

-	receipt.GasUsed = ethtypes.EthUint64(lookup.Receipt.GasUsed)
-
-	// TODO: handle CumulativeGasUsed
-	receipt.CumulativeGasUsed = ethtypes.EmptyEthInt
-
-	effectiveGasPrice := big.Div(replay.GasCost.TotalCost, big.NewInt(lookup.Receipt.GasUsed))
-	receipt.EffectiveGasPrice = ethtypes.EthBigInt(effectiveGasPrice)
-
 	return receipt, nil
 }

@@ -1821,7 +1947,7 @@ func (m *EthTxHashManager) Apply(ctx context.Context, from, to *types.TipSet) er
 				continue
 			}

-			hash, err := EthTxHashFromSignedFilecoinMessage(ctx, smsg, m.StateAPI)
+			hash, err := EthTxHashFromSignedMessage(ctx, smsg, m.StateAPI)
 			if err != nil {
 				return err
 			}
@@ -1858,7 +1984,7 @@ func WaitForMpoolUpdates(ctx context.Context, ch <-chan api.MpoolUpdate, manager
 				continue
 			}

-			ethTx, err := NewEthTxFromFilecoinMessage(ctx, u.Message, manager.StateAPI)
+			ethTx, err := newEthTxFromSignedMessage(ctx, u.Message, manager.StateAPI)
 			if err != nil {
 				log.Errorf("error converting filecoin message to eth tx: %s", err)
 			}
@@ -1887,10 +2013,6 @@ func EthTxHashGC(ctx context.Context, retentionDays int, manager *EthTxHashManag
 	}
 }

-// TODO we could also emit full EVM words from the EVM runtime, but not doing so
-// makes the contract slightly cheaper (and saves storage bytes), at the expense
-// of having to left pad in the API, which is a pretty acceptable tradeoff at
-// face value. There may be other protocol implications to consider.
 func leftpad32(orig []byte) []byte {
 	needed := 32 - len(orig)
 	if needed <= 0 {
@@ -1900,3 +2022,51 @@ func leftpad32(orig []byte) []byte {
 	copy(ret[needed:], orig)
 	return ret
 }
+
+func trimLeadingZeros(b []byte) []byte {
+	for i := range b {
+		if b[i] != 0 {
+			return b[i:]
+		}
+	}
+	return []byte{}
+}
+
+func cborEncodeTopicValue(orig []byte) ([]byte, error) {
+	var buf bytes.Buffer
+	err := cbg.WriteByteArray(&buf, trimLeadingZeros(orig))
+	if err != nil {
+		return nil, err
+	}
+	return buf.Bytes(), nil
+}
+
+func cborDecodeTopicValue(orig []byte) ([]byte, error) {
+	if len(orig) == 0 {
+		return orig, nil
+	}
+	decoded, err := cbg.ReadByteArray(bytes.NewReader(orig), uint64(len(orig)))
+	if err != nil {
+		return nil, err
+	}
+	return leftpad32(decoded), nil
+}
+
+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 {
+			encodedVal, err := cborEncodeTopicValue(v[:])
+			if err != nil {
+				return nil, xerrors.Errorf("failed to encode topic value")
+			}
+			keys[key] = append(keys[key], encodedVal)
+		}
+	}
+	return keys, nil
+}
@@ -248,22 +248,23 @@ func (m *GasModule) GasEstimateGasLimit(ctx context.Context, msgIn *types.Messag
 	}
 	return gasEstimateGasLimit(ctx, m.Chain, m.Stmgr, m.Mpool, msgIn, ts)
 }
-func gasEstimateGasLimit(
+
+// gasEstimateCallWithGas invokes a message "msgIn" on the earliest available tipset with pending
+// messages in the message pool. The function returns the result of the message invocation, the
+// pending messages, the tipset used for the invocation, and an error if occurred.
+// The returned information can be used to make subsequent calls to CallWithGas with the same parameters.
+func gasEstimateCallWithGas(
 	ctx context.Context,
 	cstore *store.ChainStore,
 	smgr *stmgr.StateManager,
 	mpool *messagepool.MessagePool,
 	msgIn *types.Message,
 	currTs *types.TipSet,
-) (int64, error) {
+) (*api.InvocResult, []types.ChainMsg, *types.TipSet, error) {
 	msg := *msgIn
-	msg.GasLimit = build.BlockGasLimit
-	msg.GasFeeCap = big.Zero()
-	msg.GasPremium = big.Zero()
-
 	fromA, err := smgr.ResolveToDeterministicAddress(ctx, msgIn.From, currTs)
 	if err != nil {
-		return -1, xerrors.Errorf("getting key address: %w", err)
+		return nil, []types.ChainMsg{}, nil, xerrors.Errorf("getting key address: %w", err)
 	}

 	pending, ts := mpool.PendingFor(ctx, fromA)
@@ -284,12 +285,34 @@ func gasEstimateGasLimit(
 		}
 		ts, err = cstore.GetTipSetFromKey(ctx, ts.Parents())
 		if err != nil {
-			return -1, xerrors.Errorf("getting parent tipset: %w", err)
+			return nil, []types.ChainMsg{}, nil, xerrors.Errorf("getting parent tipset: %w", err)
 		}
 	}
 	if err != nil {
-		return -1, xerrors.Errorf("CallWithGas failed: %w", err)
+		return nil, []types.ChainMsg{}, nil, xerrors.Errorf("CallWithGas failed: %w", err)
 	}
+
+	return res, priorMsgs, ts, nil
+}
+
+func gasEstimateGasLimit(
+	ctx context.Context,
+	cstore *store.ChainStore,
+	smgr *stmgr.StateManager,
+	mpool *messagepool.MessagePool,
+	msgIn *types.Message,
+	currTs *types.TipSet,
+) (int64, error) {
+	msg := *msgIn
+	msg.GasLimit = build.BlockGasLimit
+	msg.GasFeeCap = big.Zero()
+	msg.GasPremium = big.Zero()
+
+	res, _, ts, err := gasEstimateCallWithGas(ctx, cstore, smgr, mpool, &msg, currTs)
+	if err != nil {
+		return -1, xerrors.Errorf("gas estimation failed: %w", err)
+	}
+
 	if res.MsgRct.ExitCode == exitcode.SysErrOutOfGas {
 		return -1, &api.ErrOutOfGas{}
 	}
@@ -300,6 +323,8 @@ func gasEstimateGasLimit(

 	ret := res.MsgRct.GasUsed

+	log.Infow("GasEstimateMessageGas CallWithGas Result", "GasUsed", ret, "ExitCode", res.MsgRct.ExitCode)
+
 	transitionalMulti := 1.0
 	// Overestimate gas around the upgrade
 	if ts.Height() <= build.UpgradeSkyrHeight && (build.UpgradeSkyrHeight-ts.Height() <= 20) {
@@ -40,6 +40,7 @@ func EthEventAPI(cfg config.FevmConfig) func(helpers.MetricsCtx, repo.LockedRepo
 		ee := &full.EthEvent{
 			Chain:                cs,
 			MaxFilterHeightRange: abi.ChainEpoch(cfg.Events.MaxFilterHeightRange),
+			SubscribtionCtx:      ctx,
 		}

 		if !cfg.EnableEthRPC || cfg.Events.DisableRealTimeFilterAPI {
@@ -181,3 +181,7 @@ func NewSlashFilter(ds dtypes.MetadataDS) *slashfilter.SlashFilter {
 func UpgradeSchedule() stmgr.UpgradeSchedule {
 	return filcns.DefaultUpgradeSchedule()
 }
+
+func EnableStoringEvents(cs *store.ChainStore) {
+	cs.StoreEvents(true)
+}
@@ -75,7 +75,7 @@ func FullNodeHandler(a v1api.FullNode, permissioned bool, opts ...jsonrpc.Server
 	m := mux.NewRouter()

 	serveRpc := func(path string, hnd interface{}) {
-		rpcServer := jsonrpc.NewServer(append(opts, jsonrpc.WithServerErrors(api.RPCErrors))...)
+		rpcServer := jsonrpc.NewServer(append(opts, jsonrpc.WithReverseClient[api.EthSubscriberMethods]("Filecoin"), jsonrpc.WithServerErrors(api.RPCErrors))...)
 		rpcServer.Register("Filecoin", hnd)
 		rpcServer.AliasMethod("rpc.discover", "Filecoin.Discover")