diff --git a/l1/eth/client/filter_query.go b/l1/eth/client/filter_query.go new file mode 100644 index 0000000000..addef346c1 --- /dev/null +++ b/l1/eth/client/filter_query.go @@ -0,0 +1,58 @@ +package client + +import ( + "encoding/json" + "strconv" + + "github.com/NethermindEth/juno/l1/eth" +) + +// FilterQuery selects logs by inclusive block range, contract address, and +// topics. A nil FromBlock/ToBlock is omitted from the wire: geth treats an +// explicit toBlock as a bounded historical filter, which would break live +// eth_subscribe subscriptions. +type FilterQuery struct { + FromBlock *uint64 + ToBlock *uint64 + Addresses []eth.Address + // Topics is position-major: Topics[i] is the allowed-set at topic + // position i (OR'd together); empty means "any value at that position". + Topics [][]eth.Hash +} + +type filterQueryWire struct { + FromBlock string `json:"fromBlock,omitempty"` + ToBlock string `json:"toBlock,omitempty"` + Address []eth.Address `json:"address,omitempty"` + Topics []any `json:"topics,omitempty"` +} + +func quantityHex(n uint64) string { + return "0x" + strconv.FormatUint(n, 16) +} + +func (q FilterQuery) MarshalJSON() ([]byte, error) { + wire := filterQueryWire{ + Address: q.Addresses, + } + if q.FromBlock != nil { + wire.FromBlock = quantityHex(*q.FromBlock) + } + if q.ToBlock != nil { + wire.ToBlock = quantityHex(*q.ToBlock) + } + if len(q.Topics) > 0 { + wire.Topics = make([]any, len(q.Topics)) + for i, ts := range q.Topics { + switch len(ts) { + case 0: + wire.Topics[i] = nil + case 1: + wire.Topics[i] = ts[0] + default: + wire.Topics[i] = ts + } + } + } + return json.Marshal(wire) +} diff --git a/l1/eth/client/filter_query_test.go b/l1/eth/client/filter_query_test.go new file mode 100644 index 0000000000..65b1bff85e --- /dev/null +++ b/l1/eth/client/filter_query_test.go @@ -0,0 +1,113 @@ +package client_test + +import ( + "encoding/json" + "testing" + + "github.com/NethermindEth/juno/l1/eth" + "github.com/NethermindEth/juno/l1/eth/client" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestFilterQuery_MarshalShapes(t *testing.T) { + addr := eth.AddressFromString("0x000000000000000000000000000000000000beef") + hash1 := eth.HashFromString("0x" + repeatHex("11", 32)) + hash2 := eth.HashFromString("0x" + repeatHex("22", 32)) + + cases := []struct { + name string + q client.FilterQuery + assert func(t *testing.T, sent map[string]any) + }{ + { + // Unset FromBlock/ToBlock must not hit the wire: geth reads an + // explicit toBlock=0 as a bounded filter ending at block 0. + name: "unset block range omits keys", + q: client.FilterQuery{}, + assert: func(t *testing.T, sent map[string]any) { + _, hasFrom := sent["fromBlock"] + assert.False(t, hasFrom, "fromBlock must be omitted when unset") + _, hasTo := sent["toBlock"] + assert.False(t, hasTo, "toBlock must be omitted when unset") + _, hasAddr := sent["address"] + assert.False(t, hasAddr) + _, hasTopics := sent["topics"] + assert.False(t, hasTopics) + }, + }, + { + // Explicit zero is distinct from unset and still expressible + // (e.g. eth_getLogs from genesis). + name: "explicit block zero", + q: client.FilterQuery{FromBlock: ptr(uint64(0)), ToBlock: ptr(uint64(0))}, + assert: func(t *testing.T, sent map[string]any) { + assert.Equal(t, "0x0", sent["fromBlock"]) + assert.Equal(t, "0x0", sent["toBlock"]) + }, + }, + { + name: "single topic", + q: client.FilterQuery{Topics: [][]eth.Hash{{hash1}}}, + assert: func(t *testing.T, sent map[string]any) { + topics := sent["topics"].([]any) + require.Len(t, topics, 1) + _, isString := topics[0].(string) + assert.True(t, isString) + }, + }, + { + name: "any-at-position-0 then exact-at-1", + q: client.FilterQuery{Topics: [][]eth.Hash{nil, {hash1}}}, + assert: func(t *testing.T, sent map[string]any) { + topics := sent["topics"].([]any) + require.Len(t, topics, 2) + assert.Nil(t, topics[0]) + _, isString := topics[1].(string) + assert.True(t, isString) + }, + }, + { + name: "OR-list at position 0", + q: client.FilterQuery{Topics: [][]eth.Hash{{hash1, hash2}}}, + assert: func(t *testing.T, sent map[string]any) { + topics := sent["topics"].([]any) + _, isArr := topics[0].([]any) + assert.True(t, isArr) + }, + }, + { + name: "addresses", + q: client.FilterQuery{Addresses: []eth.Address{addr}}, + assert: func(t *testing.T, sent map[string]any) { + addrs := sent["address"].([]any) + require.Len(t, addrs, 1) + assert.Equal(t, addrHex(addr), addrs[0]) + }, + }, + } + for _, c := range cases { + t.Run(c.name, func(t *testing.T) { + raw, err := json.Marshal(c.q) + require.NoError(t, err) + var sent map[string]any + require.NoError(t, json.Unmarshal(raw, &sent)) + c.assert(t, sent) + }) + } +} + +func repeatHex(unit string, repeat int) string { + out := make([]byte, 0, len(unit)*repeat) + for range repeat { + out = append(out, unit...) + } + return string(out) +} + +func addrHex(a eth.Address) string { + b, _ := a.MarshalText() + return string(b) +} + +func ptr[T any](v T) *T { return &v } diff --git a/l1/eth/client/subscribe.go b/l1/eth/client/subscribe.go new file mode 100644 index 0000000000..fb8fa8e2a6 --- /dev/null +++ b/l1/eth/client/subscribe.go @@ -0,0 +1,125 @@ +package client + +import ( + "context" + "encoding/json" + "fmt" + "sync" + + "github.com/NethermindEth/juno/l1/eth" + "go.uber.org/zap" +) + +// Subscription mirrors go-ethereum's event.Subscription, as a drop-in for +// callers migrating off that package. +type Subscription interface { + Err() <-chan error + Unsubscribe() +} + +type wsLogSub struct { + id string // server-assigned; set during the subscribe handshake + transport *wsTransport + sink chan<- *eth.Log + + // cancelled is set under transport.mu by cancelPending when the caller's ctx fires; + // registerSub checks it to avoid orphaning a server-side sub when the reply races + // the cancellation. + cancelled bool + + // logCh decouples the shared reader goroutine from this sub's decode+deliver work. + logCh chan json.RawMessage + + // errCh is closed when the subscription terminates; a non-nil cause is sent before close. + errCh chan error + + closed chan struct{} + closeOnce sync.Once +} + +func (s *wsLogSub) Err() <-chan error { return s.errCh } + +func (s *wsLogSub) Unsubscribe() { + s.fail(nil) + s.transport.mu.Lock() + id := s.id + s.id = "" + if s.transport.subs != nil && id != "" { + delete(s.transport.subs, id) + } + s.transport.mu.Unlock() + + if id == "" { + return + } + ctx, cancel := context.WithTimeout(context.Background(), wsUnsubscribeTimeout) + defer cancel() + if _, err := s.transport.call(ctx, "eth_unsubscribe", id); err != nil { + s.transport.logger.Trace( + "eth_unsubscribe failed", + zap.String("subscription", id), + zap.Error(err), + ) + } +} + +// fail(nil) is a clean shutdown (Unsubscribe); a non-nil cause is surfaced on Err(). +func (s *wsLogSub) fail(cause error) { + s.closeOnce.Do(func() { + close(s.closed) + if cause != nil { + select { + case s.errCh <- cause: + default: + } + } + close(s.errCh) + }) +} + +func (s *wsLogSub) dispatch() { + for { + select { + case raw := <-s.logCh: + var log eth.Log + if err := json.Unmarshal(raw, &log); err != nil { + s.fail(fmt.Errorf("decoding log: %w", err)) + s.transport.removeSub(s) + return + } + select { + case s.sink <- &log: + case <-s.closed: + return + } + case <-s.closed: + return + } + } +} + +func (t *wsTransport) subscribeLogs( + ctx context.Context, + q FilterQuery, + sink chan<- *eth.Log, +) (*wsLogSub, error) { + sub := &wsLogSub{ + transport: t, + sink: sink, + logCh: make(chan json.RawMessage, wsLogSubBuffer), + errCh: make(chan error, 1), + closed: make(chan struct{}), + } + + // The sub becomes routable mid-handshake (registerSub), so the drain + // goroutine must already be running or a notification burst could + // overflow logCh before anyone reads it. + go sub.dispatch() + + if _, err := t.callWithSubReg(ctx, "eth_subscribe", sub, "logs", q); err != nil { + sub.closeOnce.Do(func() { close(sub.closed); close(sub.errCh) }) + return nil, fmt.Errorf("subscribing to logs: %w", err) + } + + return sub, nil +} diff --git a/l1/eth/client/transport_ws.go b/l1/eth/client/transport_ws.go index 092072565d..4da4719ac2 100644 --- a/l1/eth/client/transport_ws.go +++ b/l1/eth/client/transport_ws.go @@ -19,11 +19,24 @@ import ( var ErrTransportClosed = errors.New("transport closed") +// ErrSubscriptionQueueOverflow fails a subscription whose sink is not drained +// fast enough, matching go-ethereum's behaviour. +var ErrSubscriptionQueueOverflow = errors.New("subscription queue overflow (slow subscriber)") + const ( // wsReadLimit (16 MiB) is far above any real payload; it only stops a // malicious server from forcing unbounded allocations. wsReadLimit = 16 << 20 + wsLogSubBuffer = 64 + + // maxOrphanedSubs bounds the orphaned-subscribe tracking against a server + // that never answers; past the cap a late reply leaks its server-side sub, + // which such a server was never going to honour anyway. + maxOrphanedSubs = 1024 + + wsUnsubscribeTimeout = 2 * time.Second + wsPingInterval = 30 * time.Second wsPingTimeout = 10 * time.Second wsDialTimeout = time.Minute @@ -49,7 +62,8 @@ func (e rpcError) Error() string { return fmt.Sprintf("jsonrpc %d: %s", e.err.Code, e.err.Message) } -// wsTransport multiplexes unary calls over one conn, routed by request id. +// wsTransport multiplexes unary calls and eth_subscribe notifications over +// one conn, routed by request id / subscription id. type wsTransport struct { conn *websocket.Conn nextID atomic.Uint64 @@ -57,6 +71,13 @@ type wsTransport struct { mu sync.Mutex pending map[uint64]chan rpcReply // by request id + // pendingSubs: subscribe calls awaiting the server-assigned sub id, keyed by request id. + pendingSubs map[uint64]*wsLogSub + // orphanedSubs: request ids of cancelled subscribes whose reply never arrived. A late + // successful reply still triggers a best-effort eth_unsubscribe, without retaining the + // full wsLogSub. + orphanedSubs map[uint64]struct{} + subs map[string]*wsLogSub // active subscriptions, by sub id pingInterval time.Duration pingTimeout time.Duration @@ -98,6 +119,9 @@ func dialWS(ctx context.Context, rawURL string, opts options) (*wsTransport, err conn: conn, logger: opts.logger, pending: make(map[uint64]chan rpcReply), + pendingSubs: make(map[uint64]*wsLogSub), + orphanedSubs: make(map[uint64]struct{}), + subs: make(map[string]*wsLogSub), pingInterval: opts.pingInterval, pingTimeout: opts.pingTimeout, closed: make(chan struct{}), @@ -160,6 +184,8 @@ func (t *wsTransport) dispatch(data []byte) { return } switch { + case probe.Method == "eth_subscription": + t.dispatchNotification(data) case len(probe.ID) > 0 && !isJSONNull(probe.ID): t.dispatchResponse(data, probe.ID) default: @@ -194,10 +220,20 @@ func (t *wsTransport) dispatchResponse(data []byte, rawID json.RawMessage) { t.mu.Lock() ch, hasPending := t.pending[id] delete(t.pending, id) + pendingSub, isSubscribe := t.pendingSubs[id] + delete(t.pendingSubs, id) + _, isOrphaned := t.orphanedSubs[id] + delete(t.orphanedSubs, id) t.mu.Unlock() if !hasPending { - // Caller's ctx fired before the reply landed, or an unsolicited reply. + // Caller's ctx fired before the reply landed (or an unsolicited reply). A successful + // subscribe orphaned a server-side sub the caller will never own — release it. + if (isSubscribe || isOrphaned) && resp.Error == nil { + if subID, derr := decodeSubID(rawResult); derr == nil { + t.unsubscribeInBackground(subID) + } + } t.logger.Trace( "drop response (no pending caller)", zap.Uint64("id", id), @@ -206,9 +242,20 @@ func (t *wsTransport) dispatchResponse(data []byte, rawID json.RawMessage) { } reply := rpcReply{} - if resp.Error != nil { + switch { + case resp.Error != nil: reply.err = rpcError{resp.Error} - } else { + case isSubscribe: + // Register the sub before waking the caller, else a notification could race in ahead of it. + if subID, derr := decodeSubID(rawResult); derr != nil { + reply.err = derr + } else if t.registerSub(pendingSub, subID) { + reply.result = rawResult + } else { + // Transport closed or caller cancelled; registerSub released any orphaned server-side sub. + reply.err = ErrTransportClosed + } + default: reply.result = rawResult } @@ -219,6 +266,56 @@ func (t *wsTransport) dispatchResponse(data []byte, rawID json.RawMessage) { } } +func (t *wsTransport) dispatchNotification(data []byte) { + var notif struct { + Method string `json:"method"` + Params struct { + Subscription string `json:"subscription"` + Result json.RawMessage `json:"result"` + } `json:"params"` + } + if err := json.Unmarshal(data, ¬if); err != nil { + t.logger.Trace( + "drop notification (decode failed)", + zap.Int("bytes", len(data)), + zap.Error(err), + ) + return + } + t.mu.Lock() + sub := t.subs[notif.Params.Subscription] + t.mu.Unlock() + if sub == nil { + // Late notification between our eth_unsubscribe and the server processing it. Harmless. + t.logger.Trace( + "drop notification for unknown subscription", + zap.String("subscription", notif.Params.Subscription), + ) + return + } + select { + case sub.logCh <- notif.Params.Result: + default: + // Full buffer: fail the slow subscription rather than block the shared + // readLoop, which would stall every unary call on the connection. + sub.fail(ErrSubscriptionQueueOverflow) + t.removeSub(sub) + } +} + +func (t *wsTransport) removeSub(s *wsLogSub) { + t.mu.Lock() + id := s.id + s.id = "" + if t.subs != nil && id != "" { + delete(t.subs, id) + } + t.mu.Unlock() + if id != "" { + t.unsubscribeInBackground(id) + } +} + // shutdown is the single termination path. The cause is normalised so // errors.Is(err, ErrTransportClosed) holds for every observer, including the // in-flight call that races the disconnect and must redial. @@ -232,8 +329,11 @@ func (t *wsTransport) shutdown(cause error) { cause = fmt.Errorf("%w: %w", ErrTransportClosed, cause) } t.mu.Lock() - pending := t.pending + pending, pendingSubs, subs := t.pending, t.pendingSubs, t.subs t.pending = nil + t.pendingSubs = nil + t.orphanedSubs = nil + t.subs = nil t.closeErr = cause t.mu.Unlock() close(t.closed) @@ -244,6 +344,12 @@ func (t *wsTransport) shutdown(cause error) { default: } } + for _, sub := range pendingSubs { + sub.fail(cause) + } + for _, sub := range subs { + sub.fail(cause) + } if t.cancelLoops != nil { t.cancelLoops() } @@ -274,6 +380,17 @@ func (t *wsTransport) call( ctx context.Context, method string, params ...any, +) (json.RawMessage, error) { + return t.callWithSubReg(ctx, method, nil, params...) +} + +// callWithSubReg registers a non-nil pendingSub atomically with the reply +// delivery, so no notification can arrive before the sub is routable. +func (t *wsTransport) callWithSubReg( + ctx context.Context, + method string, + pendingSub *wsLogSub, + params ...any, ) (json.RawMessage, error) { if params == nil { params = []any{} @@ -287,12 +404,16 @@ func (t *wsTransport) call( return nil, ErrTransportClosed } t.pending[id] = ch + if pendingSub != nil { + t.pendingSubs[id] = pendingSub + } t.mu.Unlock() deregister := func() { t.mu.Lock() if t.pending != nil { delete(t.pending, id) + delete(t.pendingSubs, id) } t.mu.Unlock() } @@ -319,7 +440,7 @@ func (t *wsTransport) call( } return reply.result, nil case <-ctx.Done(): - deregister() + t.cancelPending(id, pendingSub) return nil, ctx.Err() case <-t.closed: deregister() @@ -332,6 +453,81 @@ func (t *wsTransport) call( } } +// registerSub returns false (releasing the server-side sub) if the transport +// closed or the caller cancelled. Atomic against cancelPending under t.mu: +// whoever locks first wins. +func (t *wsTransport) registerSub(pendingSub *wsLogSub, subID string) bool { + t.mu.Lock() + switch { + case t.subs == nil: + t.mu.Unlock() + return false + case pendingSub.cancelled: + t.mu.Unlock() + t.unsubscribeInBackground(subID) + return false + default: + pendingSub.id = subID + t.subs[subID] = pendingSub + t.mu.Unlock() + return true + } +} + +// cancelPending tears down a pending call whose caller's ctx fired, releasing +// any server-side sub it would otherwise leak. +func (t *wsTransport) cancelPending(id uint64, pendingSub *wsLogSub) { + var leakedSubID string + t.mu.Lock() + if t.pending != nil { + delete(t.pending, id) + } + if pendingSub != nil { + pendingSub.cancelled = true + if _, awaitingReply := t.pendingSubs[id]; awaitingReply { + delete(t.pendingSubs, id) + if len(t.orphanedSubs) < maxOrphanedSubs { + t.orphanedSubs[id] = struct{}{} + } + } + if pendingSub.id != "" && t.subs != nil { + leakedSubID = pendingSub.id + delete(t.subs, leakedSubID) + } + } + t.mu.Unlock() + if leakedSubID == "" { + return + } + t.unsubscribeInBackground(leakedSubID) +} + +// unsubscribeInBackground never delays the caller; bounded by wsUnsubscribeTimeout. +func (t *wsTransport) unsubscribeInBackground(subID string) { + go func() { + ctx, cancel := context.WithTimeout(context.Background(), wsUnsubscribeTimeout) + defer cancel() + if _, err := t.call(ctx, "eth_unsubscribe", subID); err != nil { + t.logger.Trace( + "best-effort eth_unsubscribe failed", + zap.String("subscription", subID), + zap.Error(err), + ) + } + }() +} + +func decodeSubID(raw json.RawMessage) (string, error) { + var subID string + if err := json.Unmarshal(raw, &subID); err != nil { + return "", fmt.Errorf("decoding subscription id: %w", err) + } + if subID == "" { + return "", errors.New("empty subscription id") + } + return subID, nil +} + func isJSONNull(raw json.RawMessage) bool { trimmed := bytes.TrimSpace(raw) return len(trimmed) == 0 || string(trimmed) == "null" diff --git a/l1/eth/client/transport_ws_internal_test.go b/l1/eth/client/transport_ws_internal_test.go index fc521deb92..6634d4d88a 100644 --- a/l1/eth/client/transport_ws_internal_test.go +++ b/l1/eth/client/transport_ws_internal_test.go @@ -8,6 +8,7 @@ import ( "context" "testing" + "github.com/NethermindEth/juno/l1/eth" "github.com/NethermindEth/juno/l1/internal/clienttest" "github.com/coder/websocket" "github.com/stretchr/testify/require" @@ -39,3 +40,60 @@ func TestWS_WriteFailureClassifiedAsTransportClosed(t *testing.T) { t.Fatal("a failed write must shut the transport down, not leave it half-alive") } } + +func TestWS_OrphanedSubsIsBounded(t *testing.T) { + gate := make(chan struct{}) + t.Cleanup(func() { close(gate) }) + srv := clienttest.NewTestServer(t) + srv.SetHandler(func(_ clienttest.TestRequest) (any, *clienttest.TestRPCError) { + <-gate // never reply + return nil, nil + }) + + tr := newTestTransport(t, srv) + + // A pre-cancelled ctx makes every subscribe write its frame and then + // immediately abandon the call, orphaning the request id. + ctx, cancel := context.WithCancel(t.Context()) + cancel() + sink := make(chan *eth.Log, 1) + for range maxOrphanedSubs + 1 { + _, err := tr.subscribeLogs(ctx, FilterQuery{}, sink) + require.ErrorIs(t, err, context.Canceled) + } + + tr.mu.Lock() + orphans := len(tr.orphanedSubs) + tr.mu.Unlock() + require.LessOrEqual(t, orphans, maxOrphanedSubs, + "orphanedSubs must not grow beyond its cap against an unresponsive server") +} + +func TestWS_CancelledSubscribeDoesNotRetainPendingSub(t *testing.T) { + received := make(chan struct{}) + gate := make(chan struct{}) + t.Cleanup(func() { close(gate) }) + srv := clienttest.NewTestServer(t) + srv.SetHandler(func(req clienttest.TestRequest) (any, *clienttest.TestRPCError) { + close(received) + <-gate // hold the reply past the caller's cancellation + return "0xfeed", nil + }) + + tr := newTestTransport(t, srv) + + ctx, cancel := context.WithCancel(t.Context()) + go func() { + <-received + cancel() + }() + sink := make(chan *eth.Log, 1) + _, err := tr.subscribeLogs(ctx, FilterQuery{}, sink) + require.ErrorIs(t, err, context.Canceled) + + tr.mu.Lock() + retained := len(tr.pendingSubs) + tr.mu.Unlock() + require.Zero(t, retained, + "cancelled subscribe must not retain its wsLogSub in pendingSubs") +} diff --git a/l1/eth/client/transport_ws_test.go b/l1/eth/client/transport_ws_test.go index 4fb9768c73..21c10cfd39 100644 --- a/l1/eth/client/transport_ws_test.go +++ b/l1/eth/client/transport_ws_test.go @@ -6,9 +6,11 @@ import ( "errors" "fmt" "sync" + "sync/atomic" "testing" "time" + "github.com/NethermindEth/juno/l1/eth" "github.com/NethermindEth/juno/l1/internal/clienttest" "github.com/NethermindEth/juno/utils/log" "github.com/stretchr/testify/assert" @@ -223,3 +225,379 @@ func TestWS_CloseRacesInFlightCalls(t *testing.T) { } } } + +func TestWS_SubscribeReceivesLogs(t *testing.T) { + const subID = "0x1a2b3c" + srv := clienttest.NewTestServer(t) + srv.SetHandler(func(req clienttest.TestRequest) (any, *clienttest.TestRPCError) { + switch req.Method { + case "eth_subscribe": + require.GreaterOrEqual(t, len(req.Params), 1) + return subID, nil + case "eth_unsubscribe": + return true, nil + } + return nil, &clienttest.TestRPCError{Code: -32601, Message: req.Method} + }) + + tr := newTestTransport(t, srv) + + sink := make(chan *eth.Log, 4) + q := FilterQuery{Topics: [][]eth.Hash{{eth.HashFromString( + "0xdb80dd488acf86d17c747445b0eabb5d57c541d3bd7b6b87af987858e5066b2b", + )}}} + sub, err := tr.subscribeLogs(t.Context(), q, sink) + require.NoError(t, err) + defer sub.Unsubscribe() + + for _, bnHex := range []string{"0x10", "0x11"} { + require.NoError(t, srv.PushNotification(t.Context(), subID, map[string]any{ + "topics": []string{"0xdb80dd488acf86d17c747445b0eabb5d57c541d3bd7b6b87af987858e5066b2b"}, + "data": "0x", + "blockNumber": bnHex, + "removed": false, + })) + } + + got := receiveLogs(t, sink, 2, 2*time.Second) + require.Len(t, got, 2) + assert.Equal(t, uint64(0x10), uint64(got[0].BlockNumber)) + assert.Equal(t, uint64(0x11), uint64(got[1].BlockNumber)) + + select { + case err, open := <-sub.Err(): + t.Fatalf("Err() fired unexpectedly: err=%v open=%v", err, open) + default: + } +} + +func TestWS_SubscribeServerError(t *testing.T) { + srv := clienttest.NewTestServer(t) + srv.SetHandler(func(req clienttest.TestRequest) (any, *clienttest.TestRPCError) { + return nil, &clienttest.TestRPCError{Code: -32601, Message: "method not supported"} + }) + tr := newTestTransport(t, srv) + + sink := make(chan *eth.Log, 1) + _, err := tr.subscribeLogs(t.Context(), FilterQuery{}, sink) + require.Error(t, err) + assert.Contains(t, err.Error(), "subscribing to logs") + assert.Contains(t, err.Error(), "method not supported") +} + +func TestWS_ServerKillsConnection(t *testing.T) { + const subID = "0xdeadbeef" + srv := clienttest.NewTestServer(t) + srv.SetHandler(func(req clienttest.TestRequest) (any, *clienttest.TestRPCError) { + require.Equal(t, "eth_subscribe", req.Method) + return subID, nil + }) + + tr := newTestTransport(t, srv) + + sink := make(chan *eth.Log, 1) + sub, err := tr.subscribeLogs(t.Context(), FilterQuery{}, sink) + require.NoError(t, err) + + srv.KillWSConns() + + select { + case err := <-sub.Err(): + assert.Error(t, err) + case <-time.After(2 * time.Second): + t.Fatal("Err() did not fire after server killed the connection") + } +} + +func TestWS_UnsubscribeIssuesCall(t *testing.T) { + const subID = "0xabc" + var sawUnsub atomic.Bool + srv := clienttest.NewTestServer(t) + srv.SetHandler(func(req clienttest.TestRequest) (any, *clienttest.TestRPCError) { + switch req.Method { + case "eth_subscribe": + return subID, nil + case "eth_unsubscribe": + sawUnsub.Store(true) + return true, nil + } + return nil, &clienttest.TestRPCError{Code: -32601, Message: req.Method} + }) + + tr := newTestTransport(t, srv) + + sink := make(chan *eth.Log, 1) + sub, err := tr.subscribeLogs(t.Context(), FilterQuery{}, sink) + require.NoError(t, err) + + sub.Unsubscribe() + + select { + case _, open := <-sub.Err(): + assert.False(t, open, "Err() should be closed after Unsubscribe") + case <-time.After(time.Second): + t.Fatal("Err() did not close after Unsubscribe") + } + require.Eventually(t, sawUnsub.Load, 2*time.Second, 10*time.Millisecond, + "server never received eth_unsubscribe") +} + +func TestWS_ClientCloseFailsActiveSubscriptions(t *testing.T) { + srv := clienttest.NewTestServer(t) + srv.SetHandler(func(req clienttest.TestRequest) (any, *clienttest.TestRPCError) { + return "0xfeed", nil + }) + + tr := newTestTransport(t, srv) + + sink := make(chan *eth.Log, 1) + sub, err := tr.subscribeLogs(t.Context(), FilterQuery{}, sink) + require.NoError(t, err) + + tr.close() + + select { + case err := <-sub.Err(): + assert.ErrorIs(t, err, ErrTransportClosed) + case <-time.After(2 * time.Second): + t.Fatal("Err() did not fire after client.Close") + } +} + +func TestWS_SubscribeOmitsBlockRange(t *testing.T) { + const subID = "0xfeed" + type capturedSub struct { + params []json.RawMessage + } + var captured atomic.Pointer[capturedSub] + + srv := clienttest.NewTestServer(t) + srv.SetHandler(func(req clienttest.TestRequest) (any, *clienttest.TestRPCError) { + if req.Method == "eth_subscribe" { + captured.Store(&capturedSub{params: req.Params}) + return subID, nil + } + if req.Method == "eth_unsubscribe" { + return true, nil + } + return nil, &clienttest.TestRPCError{Code: -32601, Message: req.Method} + }) + + tr := newTestTransport(t, srv) + + sink := make(chan *eth.Log, 1) + sub, err := tr.subscribeLogs(t.Context(), FilterQuery{}, sink) + require.NoError(t, err) + defer sub.Unsubscribe() + + got := captured.Load() + require.NotNil(t, got, "eth_subscribe was never received by the test server") + require.Len(t, got.params, 2, `expected ["logs", ] params`) + + var filter map[string]any + require.NoError(t, json.Unmarshal(got.params[1], &filter)) + _, hasFrom := filter["fromBlock"] + assert.False( + t, hasFrom, + `eth_subscribe filter must omit "fromBlock" for a live-logs subscription; got %v`, + filter, + ) + _, hasTo := filter["toBlock"] + assert.False( + t, hasTo, + `eth_subscribe filter must omit "toBlock" for a live-logs subscription; got %v`, + filter, + ) +} + +func TestWS_SubscribeCtxCancelBeforeReplyReleasesServerSub(t *testing.T) { + const subID = "0xabc" + release := make(chan struct{}) + gotSubscribe := make(chan struct{}, 1) + gotUnsubscribe := make(chan string, 1) + + srv := clienttest.NewTestServer(t) + srv.SetHandler(func(req clienttest.TestRequest) (any, *clienttest.TestRPCError) { + switch req.Method { + case "eth_subscribe": + gotSubscribe <- struct{}{} + <-release // hold the reply until the test cancels the ctx + return subID, nil + case "eth_unsubscribe": + var id string + _ = json.Unmarshal(req.Params[0], &id) + gotUnsubscribe <- id + return true, nil + } + return nil, &clienttest.TestRPCError{Code: -32601, Message: req.Method} + }) + + tr := newTestTransport(t, srv) + + ctx, cancel := context.WithCancel(context.Background()) + sink := make(chan *eth.Log, 1) + subErr := make(chan error, 1) + go func() { + _, e := tr.subscribeLogs(ctx, FilterQuery{}, sink) + subErr <- e + }() + + // Cancel while the server holds the reply, forcing the caller onto ctx.Done(). + <-gotSubscribe + cancel() + require.ErrorIs(t, <-subErr, context.Canceled) + + // Releasing the abandoned reply must trigger the unsubscribe. + close(release) + select { + case id := <-gotUnsubscribe: + require.Equal(t, subID, id) + case <-time.After(2 * time.Second): + t.Fatal("transport did not release the orphaned server-side subscription") + } +} + +func TestWS_UnsubscribeToleratesServerRejection(t *testing.T) { + const subID = "0xrej" + var unsubAttempted atomic.Bool + + srv := clienttest.NewTestServer(t) + srv.SetHandler(func(req clienttest.TestRequest) (any, *clienttest.TestRPCError) { + switch req.Method { + case "eth_subscribe": + return subID, nil + case "eth_unsubscribe": + unsubAttempted.Store(true) + return nil, &clienttest.TestRPCError{Code: -32000, Message: "subscription not found"} + } + return nil, &clienttest.TestRPCError{Code: -32601, Message: req.Method} + }) + + tr := newTestTransport(t, srv) + + sink := make(chan *eth.Log, 1) + sub, err := tr.subscribeLogs(t.Context(), FilterQuery{}, sink) + require.NoError(t, err) + + sub.Unsubscribe() + require.True(t, unsubAttempted.Load(), "Unsubscribe must attempt eth_unsubscribe") +} + +func TestWS_SubscribeDispatchDecodeFailure(t *testing.T) { + const subID = "0xc0de" + srv := clienttest.NewTestServer(t) + srv.SetHandler(func(req clienttest.TestRequest) (any, *clienttest.TestRPCError) { + switch req.Method { + case "eth_subscribe": + return subID, nil + case "eth_unsubscribe": + return true, nil + } + return nil, &clienttest.TestRPCError{Code: -32601, Message: req.Method} + }) + + tr := newTestTransport(t, srv) + + sink := make(chan *eth.Log, 1) + sub, err := tr.subscribeLogs(t.Context(), FilterQuery{}, sink) + require.NoError(t, err) + defer sub.Unsubscribe() + + // topics expects an array of hex strings; a string forces the unmarshal to fail. + require.NoError(t, srv.PushNotification(t.Context(), subID, map[string]any{ + "topics": "not-an-array", + })) + + select { + case errOut, open := <-sub.Err(): + assert.True(t, open, "Err() must deliver the cause before closing") + require.Error(t, errOut) + assert.Contains(t, errOut.Error(), "decoding log") + case <-time.After(2 * time.Second): + t.Fatal("Err() did not fire on undecodable notification payload") + } +} + +func TestWS_CallReturnsCtxErrAfterCancellation(t *testing.T) { + received := make(chan struct{}) + gate := make(chan struct{}) + t.Cleanup(func() { close(gate) }) + srv := clienttest.NewTestServer(t) + srv.SetHandler(func(_ clienttest.TestRequest) (any, *clienttest.TestRPCError) { + close(received) + <-gate + return "0xfeed", nil + }) + + tr := newTestTransport(t, srv) + + ctx, cancel := context.WithCancel(t.Context()) + go func() { + <-received + cancel() + }() + sink := make(chan *eth.Log, 1) + _, err := tr.subscribeLogs(ctx, FilterQuery{}, sink) + require.Error(t, err) + assert.True(t, errors.Is(err, context.Canceled), "expected ctx.Canceled, got %v", err) +} + +func TestWS_SlowSubscriberFailsInsteadOfStallingConn(t *testing.T) { + const subID = "0x5109" + srv := clienttest.NewTestServer(t) + srv.SetHandler(func(req clienttest.TestRequest) (any, *clienttest.TestRPCError) { + switch req.Method { + case "eth_subscribe": + return subID, nil + case "eth_unsubscribe": + return true, nil + case "eth_chainId": + return "0x1", nil + } + return nil, &clienttest.TestRPCError{Code: -32601, Message: req.Method} + }) + + tr := newTestTransport(t, srv) + + sink := make(chan *eth.Log) + sub, err := tr.subscribeLogs(t.Context(), FilterQuery{}, sink) + require.NoError(t, err) + + // 2x the internal buffer guarantees overflow. + for range 128 { + require.NoError(t, srv.PushNotification(t.Context(), subID, map[string]any{ + "topics": []string{"0xdb80dd488acf86d17c747445b0eabb5d57c541d3bd7b6b87af987858e5066b2b"}, + "data": "0x", + "blockNumber": "0x10", + "removed": false, + })) + } + + select { + case err := <-sub.Err(): + require.ErrorIs(t, err, ErrSubscriptionQueueOverflow) + case <-time.After(2 * time.Second): + t.Fatal("subscription Err() did not fire on queue overflow") + } + + ctx, cancel := context.WithTimeout(t.Context(), 2*time.Second) + defer cancel() + raw, err := tr.call(ctx, "eth_chainId") + require.NoError(t, err, "unary call stalled by a slow subscriber") + assert.Equal(t, `"0x1"`, string(raw)) +} + +func receiveLogs(t *testing.T, sink <-chan *eth.Log, n int, timeout time.Duration) []*eth.Log { + t.Helper() + deadline := time.After(timeout) + out := make([]*eth.Log, 0, n) + for len(out) < n { + select { + case log := <-sink: + out = append(out, log) + case <-deadline: + return out + } + } + return out +}