feat(l1): [3] add eth_subscribe support to the ws transport

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)
+}

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 }

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
+}