Merge pull request #10027 from filecoin-project/feat/correct-eth-sub

feat: rpc: correct `eth_subscribe` implementation

api/api_full.go

@@ -16,6 +16,7 @@ import (
 	datatransfer "github.com/filecoin-project/go-data-transfer"
 	"github.com/filecoin-project/go-fil-markets/retrievalmarket"
 	"github.com/filecoin-project/go-fil-markets/storagemarket"
+	"github.com/filecoin-project/go-jsonrpc"
 	"github.com/filecoin-project/go-state-types/abi"
 	"github.com/filecoin-project/go-state-types/big"
 	"github.com/filecoin-project/go-state-types/builtin/v8/paych"
@@ -831,7 +832,7 @@ type FullNode interface {
 	//  - logs: notify new event logs that match a criteria
 	// params contains additional parameters used with the log event type
 	// The client will receive a stream of EthSubscriptionResponse values until EthUnsubscribe is called.
-	EthSubscribe(ctx context.Context, eventType string, params *ethtypes.EthSubscriptionParams) (<-chan ethtypes.EthSubscriptionResponse, error) //perm:write
+	EthSubscribe(ctx context.Context, params jsonrpc.RawParams) (ethtypes.EthSubscriptionID, error) //perm:write
 
 	// Unsubscribe from a websocket subscription
 	EthUnsubscribe(ctx context.Context, id ethtypes.EthSubscriptionID) (bool, error) //perm:write
@@ -849,6 +850,12 @@ type FullNode interface {
 	RaftLeader(ctx context.Context) (peer.ID, error)       //perm:read
 }
 
+// reverse interface to the client, called after EthSubscribe
+type EthSubscriber interface {
+	// note: the parameter is ethtypes.EthSubscriptionResponse serialized as json object
+	EthSubscription(ctx context.Context, r jsonrpc.RawParams) error //rpc_method:eth_subscription notify:true
+}
+
 type StorageAsk struct {
 	Response *storagemarket.StorageAsk
 

api/api_gateway.go

@@ -7,6 +7,7 @@ import (
 	"github.com/ipfs/go-cid"
 
 	"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/builtin/v9/miner"
 	"github.com/filecoin-project/go-state-types/dline"
@@ -102,6 +103,6 @@ type Gateway 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)
 }

api/client/client.go

@@ -35,10 +35,10 @@ func NewFullNodeRPCV0(ctx context.Context, addr string, requestHeader http.Heade
 }
 
 // NewFullNodeRPCV1 creates a new http jsonrpc client.
-func NewFullNodeRPCV1(ctx context.Context, addr string, requestHeader http.Header) (api.FullNode, jsonrpc.ClientCloser, error) {
+func NewFullNodeRPCV1(ctx context.Context, addr string, requestHeader http.Header, opts ...jsonrpc.Option) (api.FullNode, jsonrpc.ClientCloser, error) {
 	var res v1api.FullNodeStruct
 	closer, err := jsonrpc.NewMergeClient(ctx, addr, "Filecoin",
-		api.GetInternalStructs(&res), requestHeader, jsonrpc.WithErrors(api.RPCErrors))
+		api.GetInternalStructs(&res), requestHeader, append([]jsonrpc.Option{jsonrpc.WithErrors(api.RPCErrors)}, opts...)...)
 
 	return &res, closer, err
 }

api/mocks/mock_full.go

@@ -23,6 +23,7 @@ import (
 	bitfield "github.com/filecoin-project/go-bitfield"
 	datatransfer "github.com/filecoin-project/go-data-transfer"
 	retrievalmarket "github.com/filecoin-project/go-fil-markets/retrievalmarket"
+	jsonrpc "github.com/filecoin-project/go-jsonrpc"
 	auth "github.com/filecoin-project/go-jsonrpc/auth"
 	abi "github.com/filecoin-project/go-state-types/abi"
 	big "github.com/filecoin-project/go-state-types/big"
@@ -1388,18 +1389,18 @@ func (mr *MockFullNodeMockRecorder) EthSendRawTransaction(arg0, arg1 interface{}
 }
 
 // EthSubscribe mocks base method.
-func (m *MockFullNode) EthSubscribe(arg0 context.Context, arg1 string, arg2 *ethtypes.EthSubscriptionParams) (<-chan ethtypes.EthSubscriptionResponse, error) {
+func (m *MockFullNode) EthSubscribe(arg0 context.Context, arg1 jsonrpc.RawParams) (ethtypes.EthSubscriptionID, error) {
 	m.ctrl.T.Helper()
-	ret := m.ctrl.Call(m, "EthSubscribe", arg0, arg1, arg2)
+	ret := m.ctrl.Call(m, "EthSubscribe", arg0, arg1)
-	ret0, _ := ret[0].(<-chan ethtypes.EthSubscriptionResponse)
+	ret0, _ := ret[0].(ethtypes.EthSubscriptionID)
 	ret1, _ := ret[1].(error)
 	return ret0, ret1
 }
 
 // EthSubscribe indicates an expected call of EthSubscribe.
-func (mr *MockFullNodeMockRecorder) EthSubscribe(arg0, arg1, arg2 interface{}) *gomock.Call {
+func (mr *MockFullNodeMockRecorder) EthSubscribe(arg0, arg1 interface{}) *gomock.Call {
 	mr.mock.ctrl.T.Helper()
-	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "EthSubscribe", reflect.TypeOf((*MockFullNode)(nil).EthSubscribe), arg0, arg1, arg2)
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "EthSubscribe", reflect.TypeOf((*MockFullNode)(nil).EthSubscribe), arg0, arg1)
 }
 
 // EthUninstallFilter mocks base method.

api/proxy_gen.go

@@ -22,6 +22,7 @@ import (
 	"github.com/filecoin-project/go-fil-markets/piecestore"
 	"github.com/filecoin-project/go-fil-markets/retrievalmarket"
 	"github.com/filecoin-project/go-fil-markets/storagemarket"
+	"github.com/filecoin-project/go-jsonrpc"
 	"github.com/filecoin-project/go-jsonrpc/auth"
 	"github.com/filecoin-project/go-state-types/abi"
 	"github.com/filecoin-project/go-state-types/builtin/v8/paych"
@@ -49,20 +50,25 @@ import (
 var ErrNotSupported = xerrors.New("method not supported")
 
 type ChainIOStruct struct {
-	Internal struct {
+	Internal ChainIOMethods
+}
+
+type ChainIOMethods struct {
 	ChainHasObj func(p0 context.Context, p1 cid.Cid) (bool, error) ``
 
 	ChainPutObj func(p0 context.Context, p1 blocks.Block) error ``
 
 	ChainReadObj func(p0 context.Context, p1 cid.Cid) ([]byte, error) ``
-	}
 }
 
 type ChainIOStub struct {
 }
 
 type CommonStruct struct {
-	Internal struct {
+	Internal CommonMethods
+}
+
+type CommonMethods struct {
 	AuthNew func(p0 context.Context, p1 []auth.Permission) ([]byte, error) `perm:"admin"`
 
 	AuthVerify func(p0 context.Context, p1 string) ([]auth.Permission, error) `perm:"read"`
@@ -84,7 +90,6 @@ type CommonStruct struct {
 	StartTime func(p0 context.Context) (time.Time, error) `perm:"read"`
 
 	Version func(p0 context.Context) (APIVersion, error) `perm:"read"`
-	}
 }
 
 type CommonStub struct {
@@ -95,8 +100,10 @@ type CommonNetStruct struct {
 
 	NetStruct
 
-	Internal struct {
+	Internal CommonNetMethods
-	}
+}
+
+type CommonNetMethods struct {
 }
 
 type CommonNetStub struct {
@@ -105,12 +112,26 @@ type CommonNetStub struct {
 	NetStub
 }
 
+type EthSubscriberStruct struct {
+	Internal EthSubscriberMethods
+}
+
+type EthSubscriberMethods struct {
+	EthSubscription func(p0 context.Context, p1 jsonrpc.RawParams) error `notify:"true"rpc_method:"eth_subscription"`
+}
+
+type EthSubscriberStub struct {
+}
+
 type FullNodeStruct struct {
 	CommonStruct
 
 	NetStruct
 
-	Internal struct {
+	Internal FullNodeMethods
+}
+
+type FullNodeMethods struct {
 	ChainBlockstoreInfo func(p0 context.Context) (map[string]interface{}, error) `perm:"read"`
 
 	ChainCheckBlockstore func(p0 context.Context) error `perm:"admin"`
@@ -281,7 +302,7 @@ type FullNodeStruct struct {
 
 	EthSendRawTransaction func(p0 context.Context, p1 ethtypes.EthBytes) (ethtypes.EthHash, error) `perm:"read"`
 
-		EthSubscribe func(p0 context.Context, p1 string, p2 *ethtypes.EthSubscriptionParams) (<-chan ethtypes.EthSubscriptionResponse, error) `perm:"write"`
+	EthSubscribe func(p0 context.Context, p1 jsonrpc.RawParams) (ethtypes.EthSubscriptionID, error) `perm:"write"`
 
 	EthUninstallFilter func(p0 context.Context, p1 ethtypes.EthFilterID) (bool, error) `perm:"write"`
 
@@ -584,7 +605,6 @@ type FullNodeStruct struct {
 	WalletVerify func(p0 context.Context, p1 address.Address, p2 []byte, p3 *crypto.Signature) (bool, error) `perm:"read"`
 
 	Web3ClientVersion func(p0 context.Context) (string, error) `perm:"read"`
-	}
 }
 
 type FullNodeStub struct {
@@ -594,7 +614,10 @@ type FullNodeStub struct {
 }
 
 type GatewayStruct struct {
-	Internal struct {
+	Internal GatewayMethods
+}
+
+type GatewayMethods struct {
 	ChainGetBlockMessages func(p0 context.Context, p1 cid.Cid) (*BlockMessages, error) ``
 
 	ChainGetGenesis func(p0 context.Context) (*types.TipSet, error) ``
@@ -681,7 +704,7 @@ type GatewayStruct struct {
 
 	EthSendRawTransaction func(p0 context.Context, p1 ethtypes.EthBytes) (ethtypes.EthHash, error) ``
 
-		EthSubscribe func(p0 context.Context, p1 string, p2 *ethtypes.EthSubscriptionParams) (<-chan ethtypes.EthSubscriptionResponse, error) ``
+	EthSubscribe func(p0 context.Context, p1 jsonrpc.RawParams) (ethtypes.EthSubscriptionID, error) ``
 
 	EthUninstallFilter func(p0 context.Context, p1 ethtypes.EthFilterID) (bool, error) ``
 
@@ -738,14 +761,16 @@ type GatewayStruct struct {
 	Version func(p0 context.Context) (APIVersion, error) ``
 
 	WalletBalance func(p0 context.Context, p1 address.Address) (types.BigInt, error) ``
-	}
 }
 
 type GatewayStub struct {
 }
 
 type NetStruct struct {
-	Internal struct {
+	Internal NetMethods
+}
+
+type NetMethods struct {
 	ID func(p0 context.Context) (peer.ID, error) `perm:"read"`
 
 	NetAddrsListen func(p0 context.Context) (peer.AddrInfo, error) `perm:"read"`
@@ -793,16 +818,17 @@ type NetStruct struct {
 	NetSetLimit func(p0 context.Context, p1 string, p2 NetLimit) error `perm:"admin"`
 
 	NetStat func(p0 context.Context, p1 string) (NetStat, error) `perm:"read"`
-	}
 }
 
 type NetStub struct {
 }
 
 type SignableStruct struct {
-	Internal struct {
+	Internal SignableMethods
+}
+
+type SignableMethods struct {
 	Sign func(p0 context.Context, p1 SignFunc) error ``
-	}
 }
 
 type SignableStub struct {
@@ -813,7 +839,10 @@ type StorageMinerStruct struct {
 
 	NetStruct
 
-	Internal struct {
+	Internal StorageMinerMethods
+}
+
+type StorageMinerMethods struct {
 	ActorAddress func(p0 context.Context) (address.Address, error) `perm:"read"`
 
 	ActorAddressConfig func(p0 context.Context) (AddressConfig, error) `perm:"read"`
@@ -1077,7 +1106,6 @@ type StorageMinerStruct struct {
 	WorkerJobs func(p0 context.Context) (map[uuid.UUID][]storiface.WorkerJob, error) `perm:"admin"`
 
 	WorkerStats func(p0 context.Context) (map[uuid.UUID]storiface.WorkerStats, error) `perm:"admin"`
-	}
 }
 
 type StorageMinerStub struct {
@@ -1087,7 +1115,10 @@ type StorageMinerStub struct {
 }
 
 type WalletStruct struct {
-	Internal struct {
+	Internal WalletMethods
+}
+
+type WalletMethods struct {
 	WalletDelete func(p0 context.Context, p1 address.Address) error `perm:"admin"`
 
 	WalletExport func(p0 context.Context, p1 address.Address) (*types.KeyInfo, error) `perm:"admin"`
@@ -1101,14 +1132,16 @@ type WalletStruct struct {
 	WalletNew func(p0 context.Context, p1 types.KeyType) (address.Address, error) `perm:"admin"`
 
 	WalletSign func(p0 context.Context, p1 address.Address, p2 []byte, p3 MsgMeta) (*crypto.Signature, error) `perm:"admin"`
-	}
 }
 
 type WalletStub struct {
 }
 
 type WorkerStruct struct {
-	Internal struct {
+	Internal WorkerMethods
+}
+
+type WorkerMethods struct {
 	AddPiece func(p0 context.Context, p1 storiface.SectorRef, p2 []abi.UnpaddedPieceSize, p3 abi.UnpaddedPieceSize, p4 storiface.Data) (storiface.CallID, error) `perm:"admin"`
 
 	DataCid func(p0 context.Context, p1 abi.UnpaddedPieceSize, p2 storiface.Data) (storiface.CallID, error) `perm:"admin"`
@@ -1182,7 +1215,6 @@ type WorkerStruct struct {
 	Version func(p0 context.Context) (Version, error) `perm:"admin"`
 
 	WaitQuiet func(p0 context.Context) error `perm:"admin"`
-	}
 }
 
 type WorkerStub struct {
@@ -1342,6 +1374,17 @@ func (s *CommonStub) Version(p0 context.Context) (APIVersion, error) {
 	return *new(APIVersion), ErrNotSupported
 }
 
+func (s *EthSubscriberStruct) EthSubscription(p0 context.Context, p1 jsonrpc.RawParams) error {
+	if s.Internal.EthSubscription == nil {
+		return ErrNotSupported
+	}
+	return s.Internal.EthSubscription(p0, p1)
+}
+
+func (s *EthSubscriberStub) EthSubscription(p0 context.Context, p1 jsonrpc.RawParams) error {
+	return ErrNotSupported
+}
+
 func (s *FullNodeStruct) ChainBlockstoreInfo(p0 context.Context) (map[string]interface{}, error) {
 	if s.Internal.ChainBlockstoreInfo == nil {
 		return *new(map[string]interface{}), ErrNotSupported
@@ -2277,15 +2320,15 @@ func (s *FullNodeStub) EthSendRawTransaction(p0 context.Context, p1 ethtypes.Eth
 	return *new(ethtypes.EthHash), ErrNotSupported
 }
 
-func (s *FullNodeStruct) EthSubscribe(p0 context.Context, p1 string, p2 *ethtypes.EthSubscriptionParams) (<-chan ethtypes.EthSubscriptionResponse, error) {
+func (s *FullNodeStruct) EthSubscribe(p0 context.Context, p1 jsonrpc.RawParams) (ethtypes.EthSubscriptionID, error) {
 	if s.Internal.EthSubscribe == nil {
-		return nil, ErrNotSupported
+		return *new(ethtypes.EthSubscriptionID), ErrNotSupported
 	}
-	return s.Internal.EthSubscribe(p0, p1, p2)
+	return s.Internal.EthSubscribe(p0, p1)
 }
 
-func (s *FullNodeStub) EthSubscribe(p0 context.Context, p1 string, p2 *ethtypes.EthSubscriptionParams) (<-chan ethtypes.EthSubscriptionResponse, error) {
+func (s *FullNodeStub) EthSubscribe(p0 context.Context, p1 jsonrpc.RawParams) (ethtypes.EthSubscriptionID, error) {
-	return nil, ErrNotSupported
+	return *new(ethtypes.EthSubscriptionID), ErrNotSupported
 }
 
 func (s *FullNodeStruct) EthUninstallFilter(p0 context.Context, p1 ethtypes.EthFilterID) (bool, error) {
@@ -4422,15 +4465,15 @@ func (s *GatewayStub) EthSendRawTransaction(p0 context.Context, p1 ethtypes.EthB
 	return *new(ethtypes.EthHash), ErrNotSupported
 }
 
-func (s *GatewayStruct) EthSubscribe(p0 context.Context, p1 string, p2 *ethtypes.EthSubscriptionParams) (<-chan ethtypes.EthSubscriptionResponse, error) {
+func (s *GatewayStruct) EthSubscribe(p0 context.Context, p1 jsonrpc.RawParams) (ethtypes.EthSubscriptionID, error) {
 	if s.Internal.EthSubscribe == nil {
-		return nil, ErrNotSupported
+		return *new(ethtypes.EthSubscriptionID), ErrNotSupported
 	}
-	return s.Internal.EthSubscribe(p0, p1, p2)
+	return s.Internal.EthSubscribe(p0, p1)
 }
 
-func (s *GatewayStub) EthSubscribe(p0 context.Context, p1 string, p2 *ethtypes.EthSubscriptionParams) (<-chan ethtypes.EthSubscriptionResponse, error) {
+func (s *GatewayStub) EthSubscribe(p0 context.Context, p1 jsonrpc.RawParams) (ethtypes.EthSubscriptionID, error) {
-	return nil, ErrNotSupported
+	return *new(ethtypes.EthSubscriptionID), ErrNotSupported
 }
 
 func (s *GatewayStruct) EthUninstallFilter(p0 context.Context, p1 ethtypes.EthFilterID) (bool, error) {
@@ -6955,6 +6998,7 @@ func (s *WorkerStub) WaitQuiet(p0 context.Context) error {
 var _ ChainIO = new(ChainIOStruct)
 var _ Common = new(CommonStruct)
 var _ CommonNet = new(CommonNetStruct)
+var _ EthSubscriber = new(EthSubscriberStruct)
 var _ FullNode = new(FullNodeStruct)
 var _ Gateway = new(GatewayStruct)
 var _ Net = new(NetStruct)

api/v0api/proxy_gen.go

@@ -39,7 +39,10 @@ type FullNodeStruct struct {
 
 	NetStruct
 
-	Internal struct {
+	Internal FullNodeMethods
+}
+
+type FullNodeMethods struct {
 	BeaconGetEntry func(p0 context.Context, p1 abi.ChainEpoch) (*types.BeaconEntry, error) `perm:"read"`
 
 	ChainDeleteObj func(p0 context.Context, p1 cid.Cid) error `perm:"admin"`
@@ -415,7 +418,6 @@ type FullNodeStruct struct {
 	WalletValidateAddress func(p0 context.Context, p1 string) (address.Address, error) `perm:"read"`
 
 	WalletVerify func(p0 context.Context, p1 address.Address, p2 []byte, p3 *crypto.Signature) (bool, error) `perm:"read"`
-	}
 }
 
 type FullNodeStub struct {
@@ -425,7 +427,10 @@ type FullNodeStub struct {
 }
 
 type GatewayStruct struct {
-	Internal struct {
+	Internal GatewayMethods
+}
+
+type GatewayMethods struct {
 	ChainGetBlockMessages func(p0 context.Context, p1 cid.Cid) (*api.BlockMessages, error) ``
 
 	ChainGetMessage func(p0 context.Context, p1 cid.Cid) (*types.Message, error) ``
@@ -489,7 +494,6 @@ type GatewayStruct struct {
 	Version func(p0 context.Context) (api.APIVersion, error) ``
 
 	WalletBalance func(p0 context.Context, p1 address.Address) (types.BigInt, error) ``
-	}
 }
 
 type GatewayStub struct {

chain/types/ethtypes/eth_types.go

@@ -363,6 +363,18 @@ func (h *EthHash) UnmarshalJSON(b []byte) error {
 	return nil
 }
 
+func (h EthHash) String() string {
+	return "0x" + hex.EncodeToString(h[:])
+}
+
+// Should ONLY be used for blocks and Filecoin messages. Eth transactions expect a different hashing scheme.
+func (h EthHash) ToCid() cid.Cid {
+	// err is always nil
+	mh, _ := multihash.EncodeName(h[:], "blake2b-256")
+
+	return cid.NewCidV1(cid.DagCBOR, mh)
+}
+
 func decodeHexString(s string, expectedLen int) ([]byte, error) {
 	s = handleHexStringPrefix(s)
 	if len(s) != expectedLen*2 {
@@ -420,18 +432,6 @@ func EthHashFromTxBytes(b []byte) EthHash {
 	return ethHash
 }
 
-func (h EthHash) String() string {
-	return "0x" + hex.EncodeToString(h[:])
-}
-
-// Should ONLY be used for blocks and Filecoin messages. Eth transactions expect a different hashing scheme.
-func (h EthHash) ToCid() cid.Cid {
-	// err is always nil
-	mh, _ := multihash.EncodeName(h[:], "blake2b-256")
-
-	return cid.NewCidV1(cid.DagCBOR, mh)
-}
-
 type EthFeeHistory struct {
 	OldestBlock   uint64         `json:"oldestBlock"`
 	BaseFeePerGas []EthBigInt    `json:"baseFeePerGas"`
@@ -441,9 +441,33 @@ type EthFeeHistory struct {
 
 type EthFilterID EthHash
 
+func (h EthFilterID) MarshalJSON() ([]byte, error) {
+	return (EthHash)(h).MarshalJSON()
+}
+
+func (h *EthFilterID) UnmarshalJSON(b []byte) error {
+	return (*EthHash)(h).UnmarshalJSON(b)
+}
+
+func (h EthFilterID) String() string {
+	return (EthHash)(h).String()
+}
+
 // An opaque identifier generated by the Lotus node to refer to an active subscription.
 type EthSubscriptionID EthHash
 
+func (h EthSubscriptionID) MarshalJSON() ([]byte, error) {
+	return (EthHash)(h).MarshalJSON()
+}
+
+func (h *EthSubscriptionID) UnmarshalJSON(b []byte) error {
+	return (*EthHash)(h).UnmarshalJSON(b)
+}
+
+func (h EthSubscriptionID) String() string {
+	return (EthHash)(h).String()
+}
+
 type EthFilterSpec struct {
 	// Interpreted as an epoch or one of "latest" for last mined block, "earliest" for first,
 	// "pending" for not yet committed messages.
@@ -595,6 +619,43 @@ type EthLog struct {
 	BlockNumber EthUint64 `json:"blockNumber"`
 }
 
+// EthSubscribeParams handles raw jsonrpc params for eth_subscribe
+type EthSubscribeParams struct {
+	EventType string
+	Params    *EthSubscriptionParams
+}
+
+func (e *EthSubscribeParams) UnmarshalJSON(b []byte) error {
+	var params []json.RawMessage
+	err := json.Unmarshal(b, &params)
+	if err != nil {
+		return err
+	}
+	switch len(params) {
+	case 2:
+		err = json.Unmarshal(params[1], &e.Params)
+		if err != nil {
+			return err
+		}
+		fallthrough
+	case 1:
+		err = json.Unmarshal(params[0], &e.EventType)
+		if err != nil {
+			return err
+		}
+	default:
+		return xerrors.Errorf("expected 1 or 2 params, got %d", len(params))
+	}
+	return nil
+}
+
+func (e EthSubscribeParams) MarshalJSON() ([]byte, error) {
+	if e.Params != nil {
+		return json.Marshal([]interface{}{e.EventType, e.Params})
+	}
+	return json.Marshal([]interface{}{e.EventType})
+}
+
 type EthSubscriptionParams struct {
 	// List of topics to be matched.
 	// Optional, default: empty list

chain/types/ethtypes/eth_types_test.go

@@ -93,6 +93,48 @@ func TestEthHash(t *testing.T) {
 		h1, err := EthHashFromCid(c)
 		require.Nil(t, err)
 		require.Equal(t, h, h1)
+
+		jm, err := json.Marshal(h)
+		require.NoError(t, err)
+		require.Equal(t, hash, string(jm))
+	}
+}
+
+func TestEthFilterID(t *testing.T) {
+	testcases := []string{
+		`"0x013dbb9442ca9667baccc6230fcd5c1c4b2d4d2870f4bd20681d4d47cfd15184"`,
+		`"0xab8653edf9f51785664a643b47605a7ba3d917b5339a0724e7642c114d0e4738"`,
+	}
+
+	for _, hash := range testcases {
+		var h EthFilterID
+		err := h.UnmarshalJSON([]byte(hash))
+
+		require.Nil(t, err)
+		require.Equal(t, h.String(), strings.Replace(hash, `"`, "", -1))
+
+		jm, err := json.Marshal(h)
+		require.NoError(t, err)
+		require.Equal(t, hash, string(jm))
+	}
+}
+
+func TestEthSubscriptionID(t *testing.T) {
+	testcases := []string{
+		`"0x013dbb9442ca9667baccc6230fcd5c1c4b2d4d2870f4bd20681d4d47cfd15184"`,
+		`"0xab8653edf9f51785664a643b47605a7ba3d917b5339a0724e7642c114d0e4738"`,
+	}
+
+	for _, hash := range testcases {
+		var h EthSubscriptionID
+		err := h.UnmarshalJSON([]byte(hash))
+
+		require.Nil(t, err)
+		require.Equal(t, h.String(), strings.Replace(hash, `"`, "", -1))
+
+		jm, err := json.Marshal(h)
+		require.NoError(t, err)
+		require.Equal(t, hash, string(jm))
 	}
 }
 

cli/util/api.go

@@ -319,11 +319,33 @@ func GetFullNodeAPIV1Single(ctx *cli.Context) (v1api.FullNode, jsonrpc.ClientClo
 	return v1API, closer, nil
 }
 
-func GetFullNodeAPIV1(ctx *cli.Context) (v1api.FullNode, jsonrpc.ClientCloser, error) {
+type GetFullNodeOptions struct {
+	ethSubHandler api.EthSubscriber
+}
+
+type GetFullNodeOption func(*GetFullNodeOptions)
+
+func FullNodeWithEthSubscribtionHandler(sh api.EthSubscriber) GetFullNodeOption {
+	return func(opts *GetFullNodeOptions) {
+		opts.ethSubHandler = sh
+	}
+}
+
+func GetFullNodeAPIV1(ctx *cli.Context, opts ...GetFullNodeOption) (v1api.FullNode, jsonrpc.ClientCloser, error) {
 	if tn, ok := ctx.App.Metadata["testnode-full"]; ok {
 		return tn.(v1api.FullNode), func() {}, nil
 	}
 
+	var options GetFullNodeOptions
+	for _, opt := range opts {
+		opt(&options)
+	}
+
+	var rpcOpts []jsonrpc.Option
+	if options.ethSubHandler != nil {
+		rpcOpts = append(rpcOpts, jsonrpc.WithClientHandler("Filecoin", options.ethSubHandler), jsonrpc.WithClientHandlerAlias("eth_subscription", "Filecoin.EthSubscription"))
+	}
+
 	heads, err := GetRawAPIMulti(ctx, repo.FullNode, "v1")
 	if err != nil {
 		return nil, nil, err
@@ -337,7 +359,7 @@ func GetFullNodeAPIV1(ctx *cli.Context) (v1api.FullNode, jsonrpc.ClientCloser, e
 	var closers []jsonrpc.ClientCloser
 
 	for _, head := range heads {
-		v1api, closer, err := client.NewFullNodeRPCV1(ctx.Context, head.addr, head.header)
+		v1api, closer, err := client.NewFullNodeRPCV1(ctx.Context, head.addr, head.header, rpcOpts...)
 		if err != nil {
 			log.Warnf("Not able to establish connection to node with addr: ", head.addr)
 			continue

cmd/lotus-gateway/main.go

@@ -162,7 +162,9 @@ var runCmd = &cli.Command{
 			log.Fatalf("Cannot register the view: %v", err)
 		}
 
-		api, closer, err := lcli.GetFullNodeAPIV1(cctx)
+		subHnd := gateway.NewEthSubHandler()
+
+		api, closer, err := lcli.GetFullNodeAPIV1(cctx, cliutil.FullNodeWithEthSubscribtionHandler(subHnd))
 		if err != nil {
 			return err
 		}
@@ -195,7 +197,7 @@ var runCmd = &cli.Command{
 			return xerrors.Errorf("failed to convert endpoint address to multiaddr: %w", err)
 		}
 
-		gwapi := gateway.NewNode(api, lookbackCap, waitLookback, rateLimit, rateLimitTimeout)
+		gwapi := gateway.NewNode(api, subHnd, lookbackCap, waitLookback, rateLimit, rateLimitTimeout)
 		h, err := gateway.Handler(gwapi, api, perConnRateLimit, connPerMinute, serverOptions...)
 		if err != nil {
 			return xerrors.Errorf("failed to set up gateway HTTP handler")

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

@@ -2520,40 +2520,7 @@ Perms: write
 Inputs:
 ```json
 [
-  [
+  "0x37690cfec6c1bf4c3b9288c7a5d783e98731e90b0a4c177c2a374c7a9427355e"
-    55,
-    105,
-    12,
-    254,
-    198,
-    193,
-    191,
-    76,
-    59,
-    146,
-    136,
-    199,
-    165,
-    215,
-    131,
-    233,
-    135,
-    49,
-    233,
-    11,
-    10,
-    76,
-    23,
-    124,
-    42,
-    55,
-    76,
-    122,
-    148,
-    39,
-    53,
-    94
-  ]
 ]
 ```
 
@@ -2574,40 +2541,7 @@ Perms: write
 Inputs:
 ```json
 [
-  [
+  "0x37690cfec6c1bf4c3b9288c7a5d783e98731e90b0a4c177c2a374c7a9427355e"
-    55,
-    105,
-    12,
-    254,
-    198,
-    193,
-    191,
-    76,
-    59,
-    146,
-    136,
-    199,
-    165,
-    215,
-    131,
-    233,
-    135,
-    49,
-    233,
-    11,
-    10,
-    76,
-    23,
-    124,
-    42,
-    55,
-    76,
-    122,
-    148,
-    39,
-    53,
-    94
-  ]
 ]
 ```
 
@@ -2881,43 +2815,7 @@ Perms: write
 
 Inputs: `null`
 
-Response:
+Response: `"0x37690cfec6c1bf4c3b9288c7a5d783e98731e90b0a4c177c2a374c7a9427355e"`
-```json
-[
-  55,
-  105,
-  12,
-  254,
-  198,
-  193,
-  191,
-  76,
-  59,
-  146,
-  136,
-  199,
-  165,
-  215,
-  131,
-  233,
-  135,
-  49,
-  233,
-  11,
-  10,
-  76,
-  23,
-  124,
-  42,
-  55,
-  76,
-  122,
-  148,
-  39,
-  53,
-  94
-]
-```
 
 ### EthNewFilter
 Installs a persistent filter based on given filter spec.
@@ -2938,43 +2836,7 @@ Inputs:
 ]
 ```
 
-Response:
+Response: `"0x37690cfec6c1bf4c3b9288c7a5d783e98731e90b0a4c177c2a374c7a9427355e"`
-```json
-[
-  55,
-  105,
-  12,
-  254,
-  198,
-  193,
-  191,
-  76,
-  59,
-  146,
-  136,
-  199,
-  165,
-  215,
-  131,
-  233,
-  135,
-  49,
-  233,
-  11,
-  10,
-  76,
-  23,
-  124,
-  42,
-  55,
-  76,
-  122,
-  148,
-  39,
-  53,
-  94
-]
-```
 
 ### EthNewPendingTransactionFilter
 Installs a persistent filter to notify when new messages arrive in the message pool.
@@ -2984,43 +2846,7 @@ Perms: write
 
 Inputs: `null`
 
-Response:
+Response: `"0x37690cfec6c1bf4c3b9288c7a5d783e98731e90b0a4c177c2a374c7a9427355e"`
-```json
-[
-  55,
-  105,
-  12,
-  254,
-  198,
-  193,
-  191,
-  76,
-  59,
-  146,
-  136,
-  199,
-  165,
-  215,
-  131,
-  233,
-  135,
-  49,
-  233,
-  11,
-  10,
-  76,
-  23,
-  124,
-  42,
-  55,
-  76,
-  122,
-  148,
-  39,
-  53,
-  94
-]
-```
 
 ### EthProtocolVersion
 
@@ -3060,57 +2886,11 @@ Perms: write
 Inputs:
 ```json
 [
-  "string value",
+  "Bw=="
-  {
-    "topics": [
-      [
-        "0x37690cfec6c1bf4c3b9288c7a5d783e98731e90b0a4c177c2a374c7a9427355e"
-      ]
-    ]
-  }
 ]
 ```
 
-Response:
+Response: `"0x37690cfec6c1bf4c3b9288c7a5d783e98731e90b0a4c177c2a374c7a9427355e"`
-```json
-{
-  "subscription": [
-    55,
-    105,
-    12,
-    254,
-    198,
-    193,
-    191,
-    76,
-    59,
-    146,
-    136,
-    199,
-    165,
-    215,
-    131,
-    233,
-    135,
-    49,
-    233,
-    11,
-    10,
-    76,
-    23,
-    124,
-    42,
-    55,
-    76,
-    122,
-    148,
-    39,
-    53,
-    94
-  ],
-  "result": {}
-}
-```
 
 ### EthUninstallFilter
 Uninstalls a filter with given id.
@@ -3121,40 +2901,7 @@ Perms: write
 Inputs:
 ```json
 [
-  [
+  "0x37690cfec6c1bf4c3b9288c7a5d783e98731e90b0a4c177c2a374c7a9427355e"
-    55,
-    105,
-    12,
-    254,
-    198,
-    193,
-    191,
-    76,
-    59,
-    146,
-    136,
-    199,
-    165,
-    215,
-    131,
-    233,
-    135,
-    49,
-    233,
-    11,
-    10,
-    76,
-    23,
-    124,
-    42,
-    55,
-    76,
-    122,
-    148,
-    39,
-    53,
-    94
-  ]
 ]
 ```
 
@@ -3169,40 +2916,7 @@ Perms: write
 Inputs:
 ```json
 [
-  [
+  "0x37690cfec6c1bf4c3b9288c7a5d783e98731e90b0a4c177c2a374c7a9427355e"
-    55,
-    105,
-    12,
-    254,
-    198,
-    193,
-    191,
-    76,
-    59,
-    146,
-    136,
-    199,
-    165,
-    215,
-    131,
-    233,
-    135,
-    49,
-    233,
-    11,
-    10,
-    76,
-    23,
-    124,
-    42,
-    55,
-    76,
-    122,
-    148,
-    39,
-    53,
-    94
-  ]
 ]
 ```
 

gateway/eth_sub.go

@@ -0,0 +1,71 @@
+package gateway
+
+import (
+	"context"
+	"sync"
+
+	"github.com/filecoin-project/go-jsonrpc"
+
+	"github.com/filecoin-project/lotus/api"
+	"github.com/filecoin-project/lotus/chain/types/ethtypes"
+)
+
+type EthSubHandler struct {
+	queued map[ethtypes.EthSubscriptionID][]ethtypes.EthSubscriptionResponse
+	sinks  map[ethtypes.EthSubscriptionID]func(context.Context, *ethtypes.EthSubscriptionResponse) error
+
+	lk sync.Mutex
+}
+
+func NewEthSubHandler() *EthSubHandler {
+	return &EthSubHandler{
+		queued: make(map[ethtypes.EthSubscriptionID][]ethtypes.EthSubscriptionResponse),
+		sinks:  make(map[ethtypes.EthSubscriptionID]func(context.Context, *ethtypes.EthSubscriptionResponse) error),
+	}
+}
+
+func (e *EthSubHandler) AddSub(ctx context.Context, id ethtypes.EthSubscriptionID, sink func(context.Context, *ethtypes.EthSubscriptionResponse) error) error {
+	e.lk.Lock()
+	defer e.lk.Unlock()
+
+	for _, p := range e.queued[id] {
+		p := p // copy
+		if err := sink(ctx, &p); err != nil {
+			return err
+		}
+	}
+	delete(e.queued, id)
+	e.sinks[id] = sink
+	return nil
+}
+
+func (e *EthSubHandler) RemoveSub(id ethtypes.EthSubscriptionID) {
+	e.lk.Lock()
+	defer e.lk.Unlock()
+
+	delete(e.sinks, id)
+	delete(e.queued, id)
+}
+
+func (e *EthSubHandler) EthSubscription(ctx context.Context, r jsonrpc.RawParams) error {
+	p, err := jsonrpc.DecodeParams[ethtypes.EthSubscriptionResponse](r)
+	if err != nil {
+		return err
+	}
+
+	e.lk.Lock()
+
+	sink := e.sinks[p.SubscriptionID]
+
+	if sink == nil {
+		e.queued[p.SubscriptionID] = append(e.queued[p.SubscriptionID], p)
+		e.lk.Unlock()
+		return nil
+	}
+
+	e.lk.Unlock()
+
+	return sink(ctx, &p) // todo track errors and auto-unsubscribe on rpc conn close?
+}
+
+var _ api.EthSubscriber = (*EthSubHandler)(nil)

gateway/handler.go

@@ -27,14 +27,14 @@ const perConnLimiterKey perConnLimiterKeyType = "limiter"
 
 type filterTrackerKeyType string
 
-const filterTrackerKey filterTrackerKeyType = "filterTracker"
+const statefulCallTrackerKey filterTrackerKeyType = "statefulCallTracker"
 
 // Handler returns a gateway http.Handler, to be mounted as-is on the server.
 func Handler(gwapi lapi.Gateway, api lapi.FullNode, rateLimit int64, connPerMinute int64, opts ...jsonrpc.ServerOption) (http.Handler, error) {
 	m := mux.NewRouter()
 
 	serveRpc := func(path string, hnd interface{}) {
-		rpcServer := jsonrpc.NewServer(append(opts, jsonrpc.WithServerErrors(lapi.RPCErrors))...)
+		rpcServer := jsonrpc.NewServer(append(opts, jsonrpc.WithReverseClient[lapi.EthSubscriberMethods]("Filecoin"), jsonrpc.WithServerErrors(lapi.RPCErrors))...)
 		rpcServer.Register("Filecoin", hnd)
 		rpcServer.AliasMethod("rpc.discover", "Filecoin.Discover")
 
@@ -90,7 +90,7 @@ func (h RateLimiterHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 	r = r.WithContext(context.WithValue(r.Context(), perConnLimiterKey, h.limiter))
 
 	// also add a filter tracker to the context
-	r = r.WithContext(context.WithValue(r.Context(), filterTrackerKey, newFilterTracker()))
+	r = r.WithContext(context.WithValue(r.Context(), statefulCallTrackerKey, newStatefulCallTracker()))
 
 	h.handler.ServeHTTP(w, r)
 }

gateway/node.go

@@ -12,6 +12,7 @@ import (
 
 	"github.com/filecoin-project/go-address"
 	"github.com/filecoin-project/go-bitfield"
+	"github.com/filecoin-project/go-jsonrpc"
 	"github.com/filecoin-project/go-state-types/abi"
 	"github.com/filecoin-project/go-state-types/dline"
 	"github.com/filecoin-project/go-state-types/network"
@@ -117,7 +118,7 @@ type TargetAPI 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)
 }
 
@@ -125,6 +126,7 @@ var _ TargetAPI = *new(api.FullNode) // gateway depends on latest
 
 type Node struct {
 	target                 TargetAPI
+	subHnd                 *EthSubHandler
 	lookbackCap            time.Duration
 	stateWaitLookbackLimit abi.ChainEpoch
 	rateLimiter            *rate.Limiter
@@ -141,7 +143,7 @@ var (
 )
 
 // NewNode creates a new gateway node.
-func NewNode(api TargetAPI, lookbackCap time.Duration, stateWaitLookbackLimit abi.ChainEpoch, rateLimit int64, rateLimitTimeout time.Duration) *Node {
+func NewNode(api TargetAPI, sHnd *EthSubHandler, lookbackCap time.Duration, stateWaitLookbackLimit abi.ChainEpoch, rateLimit int64, rateLimitTimeout time.Duration) *Node {
 	var limit rate.Limit
 	if rateLimit == 0 {
 		limit = rate.Inf
@@ -150,6 +152,7 @@ func NewNode(api TargetAPI, lookbackCap time.Duration, stateWaitLookbackLimit ab
 	}
 	return &Node{
 		target:                 api,
+		subHnd:                 sHnd,
 		lookbackCap:            lookbackCap,
 		stateWaitLookbackLimit: stateWaitLookbackLimit,
 		rateLimiter:            rate.NewLimiter(limit, stateRateLimitTokens),

gateway/node_test.go

@@ -89,7 +89,7 @@ func TestGatewayAPIChainGetTipSetByHeight(t *testing.T) {
 		tt := tt
 		t.Run(tt.name, func(t *testing.T) {
 			mock := &mockGatewayDepsAPI{}
-			a := NewNode(mock, DefaultLookbackCap, DefaultStateWaitLookbackLimit, 0, time.Minute)
+			a := NewNode(mock, nil, DefaultLookbackCap, DefaultStateWaitLookbackLimit, 0, time.Minute)
 
 			// Create tipsets from genesis up to tskh and return the highest
 			ts := mock.createTipSets(tt.args.tskh, tt.args.genesisTS)
@@ -245,7 +245,7 @@ func TestGatewayVersion(t *testing.T) {
 	//stm: @GATEWAY_NODE_GET_VERSION_001
 	ctx := context.Background()
 	mock := &mockGatewayDepsAPI{}
-	a := NewNode(mock, DefaultLookbackCap, DefaultStateWaitLookbackLimit, 0, time.Minute)
+	a := NewNode(mock, nil, DefaultLookbackCap, DefaultStateWaitLookbackLimit, 0, time.Minute)
 
 	v, err := a.Version(ctx)
 	require.NoError(t, err)
@@ -256,7 +256,7 @@ func TestGatewayLimitTokensAvailable(t *testing.T) {
 	ctx := context.Background()
 	mock := &mockGatewayDepsAPI{}
 	tokens := 3
-	a := NewNode(mock, DefaultLookbackCap, DefaultStateWaitLookbackLimit, int64(tokens), time.Minute)
+	a := NewNode(mock, nil, DefaultLookbackCap, DefaultStateWaitLookbackLimit, int64(tokens), time.Minute)
 	require.NoError(t, a.limit(ctx, tokens), "requests should not be limited when there are enough tokens available")
 }
 
@@ -264,7 +264,7 @@ func TestGatewayLimitTokensNotAvailable(t *testing.T) {
 	ctx := context.Background()
 	mock := &mockGatewayDepsAPI{}
 	tokens := 3
-	a := NewNode(mock, DefaultLookbackCap, DefaultStateWaitLookbackLimit, int64(1), time.Millisecond)
+	a := NewNode(mock, nil, DefaultLookbackCap, DefaultStateWaitLookbackLimit, int64(1), time.Millisecond)
 	var err error
 	// try to be rate limited
 	for i := 0; i <= 1000; i++ {

gateway/proxy_eth.go

@@ -3,12 +3,14 @@ package gateway
 import (
 	"bytes"
 	"context"
+	"encoding/json"
 	"fmt"
 	"sync"
 	"time"
 
 	"golang.org/x/xerrors"
 
+	"github.com/filecoin-project/go-jsonrpc"
 	"github.com/filecoin-project/go-state-types/abi"
 	"github.com/filecoin-project/go-state-types/big"
 
@@ -352,7 +354,7 @@ func (gw *Node) EthGetFilterChanges(ctx context.Context, id ethtypes.EthFilterID
 		return nil, err
 	}
 
-	ft := filterTrackerFromContext(ctx)
+	ft := statefulCallFromContext(ctx)
 	ft.lk.Lock()
 	_, ok := ft.userFilters[id]
 	ft.lk.Unlock()
@@ -369,7 +371,7 @@ func (gw *Node) EthGetFilterLogs(ctx context.Context, id ethtypes.EthFilterID) (
 		return nil, err
 	}
 
-	ft := filterTrackerFromContext(ctx)
+	ft := statefulCallFromContext(ctx)
 	ft.lk.Lock()
 	_, ok := ft.userFilters[id]
 	ft.lk.Unlock()
@@ -417,7 +419,7 @@ func (gw *Node) EthUninstallFilter(ctx context.Context, id ethtypes.EthFilterID)
 	}
 
 	// check if the filter belongs to this connection
-	ft := filterTrackerFromContext(ctx)
+	ft := statefulCallFromContext(ctx)
 	ft.lk.Lock()
 	defer ft.lk.Unlock()
 
@@ -434,18 +436,88 @@ func (gw *Node) EthUninstallFilter(ctx context.Context, id ethtypes.EthFilterID)
 	return ok, nil
 }
 
-func (gw *Node) EthSubscribe(ctx context.Context, eventType string, params *ethtypes.EthSubscriptionParams) (<-chan ethtypes.EthSubscriptionResponse, error) {
+func (gw *Node) EthSubscribe(ctx context.Context, p jsonrpc.RawParams) (ethtypes.EthSubscriptionID, error) {
-	return nil, xerrors.Errorf("not implemented")
+	// validate params
+	_, err := jsonrpc.DecodeParams[ethtypes.EthSubscribeParams](p)
+	if err != nil {
+		return ethtypes.EthSubscriptionID{}, xerrors.Errorf("decoding params: %w", err)
+	}
+
+	if err := gw.limit(ctx, stateRateLimitTokens); err != nil {
+		return ethtypes.EthSubscriptionID{}, err
+	}
+
+	if gw.subHnd == nil {
+		return ethtypes.EthSubscriptionID{}, xerrors.New("subscription support not enabled")
+	}
+
+	ethCb, ok := jsonrpc.ExtractReverseClient[api.EthSubscriberMethods](ctx)
+	if !ok {
+		return ethtypes.EthSubscriptionID{}, xerrors.Errorf("connection doesn't support callbacks")
+	}
+
+	ft := statefulCallFromContext(ctx)
+	ft.lk.Lock()
+	defer ft.lk.Unlock()
+
+	if len(ft.userSubscriptions) >= EthMaxFiltersPerConn {
+		return ethtypes.EthSubscriptionID{}, fmt.Errorf("too many subscriptions")
+	}
+
+	sub, err := gw.target.EthSubscribe(ctx, p)
+	if err != nil {
+		return ethtypes.EthSubscriptionID{}, err
+	}
+
+	err = gw.subHnd.AddSub(ctx, sub, func(ctx context.Context, response *ethtypes.EthSubscriptionResponse) error {
+		outParam, err := json.Marshal(response)
+		if err != nil {
+			return err
+		}
+
+		return ethCb.EthSubscription(ctx, outParam)
+	})
+	if err != nil {
+		return ethtypes.EthSubscriptionID{}, err
+	}
+
+	ft.userSubscriptions[sub] = time.Now()
+
+	return sub, err
 }
 
 func (gw *Node) EthUnsubscribe(ctx context.Context, id ethtypes.EthSubscriptionID) (bool, error) {
-	return false, xerrors.Errorf("not implemented")
+	if err := gw.limit(ctx, stateRateLimitTokens); err != nil {
+		return false, err
+	}
+
+	// check if the filter belongs to this connection
+	ft := statefulCallFromContext(ctx)
+	ft.lk.Lock()
+	defer ft.lk.Unlock()
+
+	if _, ok := ft.userSubscriptions[id]; !ok {
+		return false, nil
+	}
+
+	ok, err := gw.target.EthUnsubscribe(ctx, id)
+	if err != nil {
+		return false, err
+	}
+
+	delete(ft.userSubscriptions, id)
+
+	if gw.subHnd != nil {
+		gw.subHnd.RemoveSub(id)
+	}
+
+	return ok, nil
 }
 
 var EthMaxFiltersPerConn = 16 // todo make this configurable
 
 func addUserFilterLimited(ctx context.Context, cb func() (ethtypes.EthFilterID, error)) (ethtypes.EthFilterID, error) {
-	ft := filterTrackerFromContext(ctx)
+	ft := statefulCallFromContext(ctx)
 	ft.lk.Lock()
 	defer ft.lk.Unlock()
 
@@ -463,19 +535,21 @@ func addUserFilterLimited(ctx context.Context, cb func() (ethtypes.EthFilterID,
 	return id, nil
 }
 
-func filterTrackerFromContext(ctx context.Context) *filterTracker {
+func statefulCallFromContext(ctx context.Context) *statefulCallTracker {
-	return ctx.Value(filterTrackerKey).(*filterTracker)
+	return ctx.Value(statefulCallTrackerKey).(*statefulCallTracker)
 }
 
-type filterTracker struct {
+type statefulCallTracker struct {
 	lk sync.Mutex
 
 	userFilters       map[ethtypes.EthFilterID]time.Time
+	userSubscriptions map[ethtypes.EthSubscriptionID]time.Time
 }
 
 // called per request (ws connection)
-func newFilterTracker() *filterTracker {
+func newStatefulCallTracker() *statefulCallTracker {
-	return &filterTracker{
+	return &statefulCallTracker{
 		userFilters:       make(map[ethtypes.EthFilterID]time.Time),
+		userSubscriptions: make(map[ethtypes.EthSubscriptionID]time.Time),
 	}
 }

gen/api/proxygen.go

@@ -259,7 +259,7 @@ func generate(path, pkg, outpkg, outfile string) error {
 						if len(tf) != 2 {
 							continue
 						}
-						if tf[0] != "perm" { // todo: allow more tag types
+						if tf[0] != "perm" && tf[0] != "rpc_method" && tf[0] != "notify" { // todo: allow more tag types
 							continue
 						}
 						info.Methods[mname].Tags[tf[0]] = tf
@@ -302,12 +302,14 @@ type {{.Num}}Struct struct {
 {{range .Include}}
 	{{.}}Struct
 {{end}}
-	Internal struct {
+	Internal {{.Num}}Methods
+}
+
+type {{.Num}}Methods struct {
 {{range .Methods}}
 		{{.Num}} func({{.NamedParams}}) ({{.Results}}) `+"`"+`{{range .Tags}}{{index . 0}}:"{{index . 1}}"{{end}}`+"`"+`
 {{end}}
 	}
-}
 
 type {{.Num}}Stub struct {
 {{range .Include}}

go.mod

@@ -40,7 +40,7 @@ require (
 	github.com/filecoin-project/go-fil-commcid v0.1.0
 	github.com/filecoin-project/go-fil-commp-hashhash v0.1.0
 	github.com/filecoin-project/go-fil-markets v1.25.2
-	github.com/filecoin-project/go-jsonrpc v0.1.9
+	github.com/filecoin-project/go-jsonrpc v0.2.0
 	github.com/filecoin-project/go-legs v0.4.4
 	github.com/filecoin-project/go-padreader v0.0.1
 	github.com/filecoin-project/go-paramfetch v0.0.4

go.sum

@@ -340,8 +340,8 @@ github.com/filecoin-project/go-hamt-ipld/v2 v2.0.0/go.mod h1:7aWZdaQ1b16BVoQUYR+
 github.com/filecoin-project/go-hamt-ipld/v3 v3.0.1/go.mod h1:gXpNmr3oQx8l3o7qkGyDjJjYSRX7hp/FGOStdqrWyDI=
 github.com/filecoin-project/go-hamt-ipld/v3 v3.1.0 h1:rVVNq0x6RGQIzCo1iiJlGFm9AGIZzeifggxtKMU7zmI=
 github.com/filecoin-project/go-hamt-ipld/v3 v3.1.0/go.mod h1:bxmzgT8tmeVQA1/gvBwFmYdT8SOFUwB3ovSUfG1Ux0g=
-github.com/filecoin-project/go-jsonrpc v0.1.9 h1:HRWLxo7HAWzI3xZGeFG4LZJoYpms+Q+8kwmMTLnyS3A=
+github.com/filecoin-project/go-jsonrpc v0.2.0 h1:lhcu0Oa7xxwdclU4EyBPgnZihiZOaTFiAnW6P3NRqko=
-github.com/filecoin-project/go-jsonrpc v0.1.9/go.mod h1:XBBpuKIMaXIIzeqzO1iucq4GvbF8CxmXRFoezRh+Cx4=
+github.com/filecoin-project/go-jsonrpc v0.2.0/go.mod h1:jBSvPTl8V1N7gSTuCR4bis8wnQnIjHbRPpROol6iQKM=
 github.com/filecoin-project/go-legs v0.4.4 h1:mpMmAOOnamaz0CV9rgeKhEWA8j9kMC+f+UGCGrxKaZo=
 github.com/filecoin-project/go-legs v0.4.4/go.mod h1:JQ3hA6xpJdbR8euZ2rO0jkxaMxeidXf0LDnVuqPAe9s=
 github.com/filecoin-project/go-padreader v0.0.0-20200903213702-ed5fae088b20/go.mod h1:mPn+LRRd5gEKNAtc+r3ScpW2JRU/pj4NBKdADYWHiak=
@@ -824,7 +824,6 @@ github.com/ipfs/go-log/v2 v2.0.1/go.mod h1:O7P1lJt27vWHhOwQmcFEvlmo49ry2VY2+JfBW
 github.com/ipfs/go-log/v2 v2.0.2/go.mod h1:O7P1lJt27vWHhOwQmcFEvlmo49ry2VY2+JfBWFaa9+0=
 github.com/ipfs/go-log/v2 v2.0.3/go.mod h1:O7P1lJt27vWHhOwQmcFEvlmo49ry2VY2+JfBWFaa9+0=
 github.com/ipfs/go-log/v2 v2.0.5/go.mod h1:eZs4Xt4ZUJQFM3DlanGhy7TkwwawCZcSByscwkWG+dw=
-github.com/ipfs/go-log/v2 v2.0.8/go.mod h1:eZs4Xt4ZUJQFM3DlanGhy7TkwwawCZcSByscwkWG+dw=
 github.com/ipfs/go-log/v2 v2.1.1/go.mod h1:2v2nsGfZsvvAJz13SyFzf9ObaqwHiHxsPLEHntrv9KM=
 github.com/ipfs/go-log/v2 v2.1.2-0.20200626104915-0016c0b4b3e4/go.mod h1:2v2nsGfZsvvAJz13SyFzf9ObaqwHiHxsPLEHntrv9KM=
 github.com/ipfs/go-log/v2 v2.1.2/go.mod h1:2v2nsGfZsvvAJz13SyFzf9ObaqwHiHxsPLEHntrv9KM=

itests/eth_filter_test.go

@@ -6,6 +6,7 @@ import (
 	"context"
 	"encoding/binary"
 	"encoding/hex"
+	"encoding/json"
 	"fmt"
 	"os"
 	"sort"
@@ -21,6 +22,7 @@ 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"
 
@@ -29,6 +31,7 @@ import (
 	"github.com/filecoin-project/lotus/chain/types"
 	"github.com/filecoin-project/lotus/chain/types/ethtypes"
 	"github.com/filecoin-project/lotus/itests/kit"
+	res "github.com/filecoin-project/lotus/lib/result"
 )
 
 // SolidityContractDef holds information about one of the test contracts
@@ -438,15 +441,15 @@ func TestEthSubscribeLogsNoTopicSpec(t *testing.T) {
 	t.Logf("actor ID address is %s", idAddr)
 
 	// install filter
-	respCh, err := client.EthSubscribe(ctx, "logs", nil)
+	subId, err := client.EthSubscribe(ctx, res.Wrap[jsonrpc.RawParams](json.Marshal(ethtypes.EthSubscribeParams{EventType: "logs"})).Assert(require.NoError))
 	require.NoError(err)
 
-	subResponses := []ethtypes.EthSubscriptionResponse{}
+	var subResponses []ethtypes.EthSubscriptionResponse
-	go func() {
+	err = client.EthSubRouter.AddSub(ctx, subId, func(ctx context.Context, resp *ethtypes.EthSubscriptionResponse) error {
-		for resp := range respCh {
+		subResponses = append(subResponses, *resp)
-			subResponses = append(subResponses, resp)
+		return nil
-		}
+	})
-	}()
+	require.NoError(err)
 
 	const iterations = 10
 	ethContractAddr, messages := invokeLogFourData(t, client, iterations)
@@ -567,44 +570,33 @@ func TestEthSubscribeLogs(t *testing.T) {
 
 	testResponses := map[string]chan ethtypes.EthSubscriptionResponse{}
 
-	// quit is used to signal that we're ready to start testing collected results
-	quit := make(chan struct{})
-
 	// Create all the filters
 	for _, tc := range testCases {
-
 		// subscribe to topics in filter
-		subCh, err := client.EthSubscribe(ctx, "logs", &ethtypes.EthSubscriptionParams{Topics: tc.spec.Topics})
+		subParam, err := json.Marshal(ethtypes.EthSubscribeParams{
+			EventType: "logs",
+			Params:    &ethtypes.EthSubscriptionParams{Topics: tc.spec.Topics},
+		})
+		require.NoError(err)
+
+		subId, err := client.EthSubscribe(ctx, subParam)
 		require.NoError(err)
 
 		responseCh := make(chan ethtypes.EthSubscriptionResponse, len(invocations))
 		testResponses[tc.name] = responseCh
 
-		// start a goroutine to forward responses from subscription to a buffered channel with guaranteed capacity
+		err = client.EthSubRouter.AddSub(ctx, subId, func(ctx context.Context, resp *ethtypes.EthSubscriptionResponse) error {
-		go func(subCh <-chan ethtypes.EthSubscriptionResponse, responseCh chan<- ethtypes.EthSubscriptionResponse, quit chan struct{}) {
+			responseCh <- *resp
-			defer func() {
+			return nil
-				close(responseCh)
+		})
-			}()
+		require.NoError(err)
-			for {
-				select {
-				case resp := <-subCh:
-					responseCh <- resp
-				case <-quit:
-					return
-				case <-ctx.Done():
-					return
-				}
-			}
-		}(subCh, responseCh, quit)
-
 	}
 
 	// Perform all the invocations
 	messages := invokeAndWaitUntilAllOnChain(t, client, invocations)
 
-	// wait a little for subscriptions to gather results and then tell all the goroutines to stop
+	// wait a little for subscriptions to gather results
 	time.Sleep(blockTime * 6)
-	close(quit)
 
 	for _, tc := range testCases {
 		tc := tc // appease the lint despot
@@ -612,6 +604,9 @@ func TestEthSubscribeLogs(t *testing.T) {
 			responseCh, ok := testResponses[tc.name]
 			require.True(ok)
 
+			// don't expect any more responses
+			close(responseCh)
+
 			var elogs []*ethtypes.EthLog
 			for resp := range responseCh {
 				rlist, ok := resp.Result.([]interface{})

itests/gateway_test.go

@@ -290,7 +290,7 @@ func startNodes(
 	ens.InterconnectAll().BeginMining(blocktime)
 
 	// Create a gateway server in front of the full node
-	gwapi := gateway.NewNode(full, lookbackCap, stateWaitLookbackLimit, 0, time.Minute)
+	gwapi := gateway.NewNode(full, nil, lookbackCap, stateWaitLookbackLimit, 0, time.Minute)
 	handler, err := gateway.Handler(gwapi, full, 0, 0)
 	require.NoError(t, err)
 

itests/kit/ensemble.go

@@ -47,6 +47,7 @@ import (
 	"github.com/filecoin-project/lotus/chain/wallet/key"
 	"github.com/filecoin-project/lotus/cmd/lotus-seed/seed"
 	"github.com/filecoin-project/lotus/cmd/lotus-worker/sealworker"
+	"github.com/filecoin-project/lotus/gateway"
 	"github.com/filecoin-project/lotus/genesis"
 	"github.com/filecoin-project/lotus/markets/idxprov"
 	"github.com/filecoin-project/lotus/markets/idxprov/idxprov_test"
@@ -210,7 +211,7 @@ func (n *Ensemble) FullNode(full *TestFullNode, opts ...NodeOpt) *Ensemble {
 		n.genesis.accounts = append(n.genesis.accounts, genacc)
 	}
 
-	*full = TestFullNode{t: n.t, options: options, DefaultKey: key}
+	*full = TestFullNode{t: n.t, options: options, DefaultKey: key, EthSubRouter: gateway.NewEthSubHandler()}
 
 	n.inactive.fullnodes = append(n.inactive.fullnodes, full)
 	return n

itests/kit/node_full.go

@@ -22,6 +22,7 @@ import (
 	"github.com/filecoin-project/lotus/chain/types"
 	"github.com/filecoin-project/lotus/chain/wallet/key"
 	cliutil "github.com/filecoin-project/lotus/cli/util"
+	"github.com/filecoin-project/lotus/gateway"
 	"github.com/filecoin-project/lotus/node"
 )
 
@@ -46,6 +47,10 @@ type TestFullNode struct {
 
 	Stop node.StopFunc
 
+	// gateway handler makes it convenient to register callbalks per topic, so we
+	// also use it for tests
+	EthSubRouter *gateway.EthSubHandler
+
 	options nodeOpts
 }
 

itests/kit/rpc.go

@@ -13,6 +13,8 @@ import (
 	manet "github.com/multiformats/go-multiaddr/net"
 	"github.com/stretchr/testify/require"
 
+	"github.com/filecoin-project/go-jsonrpc"
+
 	"github.com/filecoin-project/lotus/api/client"
 	"github.com/filecoin-project/lotus/cmd/lotus-worker/sealworker"
 	"github.com/filecoin-project/lotus/node"
@@ -52,7 +54,12 @@ func fullRpc(t *testing.T, f *TestFullNode) (*TestFullNode, Closer) {
 	fmt.Printf("FULLNODE RPC ENV FOR CLI DEBUGGING `export FULLNODE_API_INFO=%s`\n", "ws://"+srv.Listener.Addr().String())
 	sendItestdNotif("FULLNODE_API_INFO", t.Name(), "ws://"+srv.Listener.Addr().String())
 
-	cl, stop, err := client.NewFullNodeRPCV1(context.Background(), "ws://"+srv.Listener.Addr().String()+"/rpc/v1", nil)
+	rpcOpts := []jsonrpc.Option{
+		jsonrpc.WithClientHandler("Filecoin", f.EthSubRouter),
+		jsonrpc.WithClientHandlerAlias("eth_subscription", "Filecoin.EthSubscription"),
+	}
+
+	cl, stop, err := client.NewFullNodeRPCV1(context.Background(), "ws://"+srv.Listener.Addr().String()+"/rpc/v1", nil, rpcOpts...)
 	require.NoError(t, err)
 	f.ListenAddr, f.ListenURL, f.FullNode = maddr, srv.URL, cl
 

lib/result/result.go

@@ -30,6 +30,12 @@ func Wrap[T any](value T, err error) Result[T] {
 	}
 }
 
-func (r *Result[T]) Unwrap() (T, error) {
+func (r Result[T]) Unwrap() (T, error) {
 	return r.Value, r.Error
 }
+
+func (r Result[T]) Assert(noErrFn func(err error, msgAndArgs ...interface{})) T {
+	noErrFn(r.Error)
+
+	return r.Value
+}

node/impl/full/eth.go

@@ -3,6 +3,7 @@ package full
 import (
 	"bytes"
 	"context"
+	"encoding/json"
 	"errors"
 	"fmt"
 	"strconv"
@@ -16,6 +17,7 @@ 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"
@@ -75,7 +77,7 @@ 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)
 }
 
@@ -132,6 +134,7 @@ type EthEvent struct {
 	FilterStore          filter.FilterStore
 	SubManager           *EthSubscriptionManager
 	MaxFilterHeightRange abi.ChainEpoch
+	SubscribtionCtx      context.Context
 }
 
 var _ EthEventAPI = (*EthEvent)(nil)
@@ -1100,37 +1103,45 @@ const (
 	EthSubscribeEventTypeLogs  = "logs"
 )
 
-func (e *EthEvent) EthSubscribe(ctx context.Context, eventType string, params *ethtypes.EthSubscriptionParams) (<-chan ethtypes.EthSubscriptionResponse, error) {
+func (e *EthEvent) EthSubscribe(ctx context.Context, p jsonrpc.RawParams) (ethtypes.EthSubscriptionID, error) {
-	if e.SubManager == nil {
+	params, err := jsonrpc.DecodeParams[ethtypes.EthSubscribeParams](p)
-		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)
 	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 {
+		if params.Params != nil {
 			var err error
-			keys, err = parseEthTopics(params.Topics)
+			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 nil, err
+				return ethtypes.EthSubscriptionID{}, err
 			}
 		}
 
@@ -1138,14 +1149,14 @@ func (e *EthEvent) EthSubscribe(ctx context.Context, eventType string, params *e
 		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) {
@@ -1294,7 +1305,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)
@@ -1310,7 +1321,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,
 	}
 
@@ -1340,13 +1351,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
@@ -1390,10 +1403,15 @@ func (e *ethSubscription) start(ctx context.Context) {
 				continue
 			}
 
-			select {
+			outParam, err := json.Marshal(resp)
-			case e.out <- resp:
+			if err != nil {
-			default:
+				log.Warnw("marshaling subscription response", "sub", e.id, "error", err)
-				// Skip if client is not reading responses
+				continue
+			}
+
+			if err := e.out(ctx, outParam); err != nil {
+				log.Warnw("sending subscription response", "sub", e.id, "error", err)
+				continue
 			}
 		}
 	}
@@ -1405,7 +1423,6 @@ func (e *ethSubscription) stop() {
 
 	if e.quit != nil {
 		e.quit()
-		close(e.out)
 		e.quit = nil
 	}
 }

node/modules/actorevent.go

@@ -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 {

node/rpc.go

@@ -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")