op-service: RPC subscribe interface fix and utils (#13476)

* op-service: RPC subscribe interface fix and utils

* op-service: fix subscribe args pass through

op-challenger/game/monitor.go

@@ -56,7 +56,7 @@ type gameMonitor struct {
 }
 
 type MinimalSubscriber interface {
-	EthSubscribe(ctx context.Context, channel interface{}, args ...interface{}) (ethereum.Subscription, error)
+	Subscribe(ctx context.Context, namespace string, channel interface{}, args ...interface{}) (ethereum.Subscription, error)
 }
 
 type headSource struct {
@@ -64,7 +64,7 @@ type headSource struct {
 }
 
 func (s *headSource) SubscribeNewHead(ctx context.Context, ch chan<- *ethTypes.Header) (ethereum.Subscription, error) {
-	return s.inner.EthSubscribe(ctx, ch, "newHeads")
+	return s.inner.Subscribe(ctx, "eth", ch, "newHeads")
 }
 
 func newGameMonitor(

op-challenger/game/monitor_test.go

@@ -197,13 +197,17 @@ func (m *mockNewHeadSource) SetErr(err error) {
 	m.err = err
 }
 
-func (m *mockNewHeadSource) EthSubscribe(
+func (m *mockNewHeadSource) Subscribe(
 	_ context.Context,
+	namespace string,
 	ch any,
 	_ ...any,
 ) (ethereum.Subscription, error) {
 	m.Lock()
 	defer m.Unlock()
+	if namespace != "eth" {
+		return nil, fmt.Errorf("only support eth RPC subscription, got %q", namespace)
+	}
 	errChan := make(chan error)
 	m.sub = &mockSubscription{errChan, (ch).(chan<- *ethtypes.Header)}
 	if m.err != nil {

op-service/client/lazy_dial.go

@@ -76,9 +76,9 @@ func (l *lazyRPC) BatchCallContext(ctx context.Context, b []rpc.BatchElem) error
 	return l.inner.BatchCallContext(ctx, b)
 }
 
-func (l *lazyRPC) EthSubscribe(ctx context.Context, channel any, args ...any) (ethereum.Subscription, error) {
+func (l *lazyRPC) Subscribe(ctx context.Context, namespace string, channel any, args ...any) (ethereum.Subscription, error) {
 	if err := l.dial(ctx); err != nil {
 		return nil, err
 	}
-	return l.inner.EthSubscribe(ctx, channel, args...)
+	return l.inner.Subscribe(ctx, namespace, channel, args...)
 }

op-service/client/polling.go

@@ -3,6 +3,7 @@ package client
 import (
 	"context"
 	"errors"
+	"fmt"
 	"sync"
 	"time"
 
@@ -86,11 +87,15 @@ func (w *PollingClient) BatchCallContext(ctx context.Context, b []rpc.BatchElem)
 	return w.c.BatchCallContext(ctx, b)
 }
 
-// EthSubscribe creates a new newHeads subscription. It takes identical arguments
-// to Geth's native EthSubscribe method. It will return an error, however, if the
+// Subscribe supports eth_subscribe of block headers,
+// by creating a new newHeads subscription. It takes identical arguments
+// to Geth's native Subscribe method. It will return an error, however, if the
 // passed in channel is not a *types.Headers channel or the subscription type is not
-// newHeads.
-func (w *PollingClient) EthSubscribe(ctx context.Context, channel any, args ...any) (ethereum.Subscription, error) {
+// newHeads. Or if the namespace is not "eth".
+func (w *PollingClient) Subscribe(ctx context.Context, namespace string, channel any, args ...any) (ethereum.Subscription, error) {
+	if namespace != "eth" {
+		return nil, fmt.Errorf("polling fallback is only supported for eth_subscribe, not for namespace %q", namespace)
+	}
 	select {
 	case <-w.ctx.Done():
 		return nil, ErrSubscriberClosed

op-service/client/polling_test.go

@@ -61,8 +61,8 @@ func (m *MockRPC) BatchCallContext(ctx context.Context, b []rpc.BatchElem) error
 	return nil
 }
 
-func (m *MockRPC) EthSubscribe(ctx context.Context, channel any, args ...any) (ethereum.Subscription, error) {
-	m.t.Fatal("EthSubscribe should not be called")
+func (m *MockRPC) Subscribe(ctx context.Context, namespace string, channel any, args ...any) (ethereum.Subscription, error) {
+	m.t.Fatal("Subscribe should not be called")
 	return nil, nil
 }
 
@@ -202,5 +202,5 @@ func requireChansEqual(t *testing.T, chans []chan *types.Header, root common.Has
 }
 
 func doSubscribe(client RPC, ch chan<- *types.Header) (ethereum.Subscription, error) {
-	return client.EthSubscribe(context.Background(), ch, "newHeads")
+	return client.Subscribe(context.Background(), "eth", ch, "newHeads")
 }

op-service/client/rate_limited.go

@@ -44,9 +44,9 @@ func (b *RateLimitingClient) BatchCallContext(ctx context.Context, batch []rpc.B
 	return b.c.BatchCallContext(cCtx, batch)
 }
 
-func (b *RateLimitingClient) EthSubscribe(ctx context.Context, channel any, args ...any) (ethereum.Subscription, error) {
+func (b *RateLimitingClient) Subscribe(ctx context.Context, namespace string, channel any, args ...any) (ethereum.Subscription, error) {
 	if err := b.rl.Wait(ctx); err != nil {
 		return nil, err
 	}
-	return b.c.EthSubscribe(ctx, channel, args...)
+	return b.c.Subscribe(ctx, namespace, channel, args...)
 }

op-service/client/rpc.go

@@ -25,7 +25,7 @@ type RPC interface {
 	Close()
 	CallContext(ctx context.Context, result any, method string, args ...any) error
 	BatchCallContext(ctx context.Context, b []rpc.BatchElem) error
-	EthSubscribe(ctx context.Context, channel any, args ...any) (ethereum.Subscription, error)
+	Subscribe(ctx context.Context, namespace string, channel any, args ...any) (ethereum.Subscription, error)
 }
 
 type rpcConfig struct {
@@ -234,8 +234,8 @@ func (b *BaseRPCClient) BatchCallContext(ctx context.Context, batch []rpc.BatchE
 	return b.c.BatchCallContext(cCtx, batch)
 }
 
-func (b *BaseRPCClient) EthSubscribe(ctx context.Context, channel any, args ...any) (ethereum.Subscription, error) {
-	return b.c.EthSubscribe(ctx, channel, args...)
+func (b *BaseRPCClient) Subscribe(ctx context.Context, namespace string, channel any, args ...any) (ethereum.Subscription, error) {
+	return b.c.Subscribe(ctx, namespace, channel, args...)
 }
 
 // InstrumentedRPCClient is an RPC client that tracks
@@ -269,8 +269,8 @@ func (ic *InstrumentedRPCClient) BatchCallContext(ctx context.Context, b []rpc.B
 	}, b)
 }
 
-func (ic *InstrumentedRPCClient) EthSubscribe(ctx context.Context, channel any, args ...any) (ethereum.Subscription, error) {
-	return ic.c.EthSubscribe(ctx, channel, args...)
+func (ic *InstrumentedRPCClient) Subscribe(ctx context.Context, namespace string, channel any, args ...any) (ethereum.Subscription, error) {
+	return ic.c.Subscribe(ctx, namespace, channel, args...)
 }
 
 // instrumentBatch handles metrics for batch calls. Request metrics are

op-service/rpc/subscription.go

+package rpc
+
+import (
+	"context"
+
+	"github.com/ethereum/go-ethereum/event"
+	"github.com/ethereum/go-ethereum/log"
+	gethrpc "github.com/ethereum/go-ethereum/rpc"
+)
+
+func SubscribeRPC[T any](ctx context.Context, logger log.Logger, feed *event.FeedOf[T]) (*gethrpc.Subscription, error) {
+	notifier, supported := gethrpc.NotifierFromContext(ctx)
+	if !supported {
+		return &gethrpc.Subscription{}, gethrpc.ErrNotificationsUnsupported
+	}
+	logger.Info("Opening subscription via RPC")
+
+	rpcSub := notifier.CreateSubscription()
+	ch := make(chan T, 10)
+	feedSub := feed.Subscribe(ch)
+
+	go func() {
+		defer logger.Info("Closing RPC subscription")
+		defer feedSub.Unsubscribe()
+
+		for {
+			select {
+			case v := <-ch:
+				if err := notifier.Notify(rpcSub.ID, v); err != nil {
+					logger.Warn("Failed to notify RPC subscription", "err", err)
+					return
+				}
+			case err, ok := <-rpcSub.Err():
+				if !ok {
+					logger.Debug("Exiting subscription")
+					return
+				}
+				logger.Warn("RPC subscription failed", "err", err)
+				return
+			}
+		}
+	}()
+
+	return rpcSub, nil
+}

op-service/rpc/subscription_test.go

+package rpc
+
+import (
+	"context"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+
+	gethevent "github.com/ethereum/go-ethereum/event"
+	"github.com/ethereum/go-ethereum/log"
+	"github.com/ethereum/go-ethereum/rpc"
+
+	"github.com/ethereum-optimism/optimism/op-service/testlog"
+)
+
+type testSubscribeAPI struct {
+	log log.Logger
+
+	foo gethevent.FeedOf[int]
+	bar gethevent.FeedOf[string]
+}
+
+func (api *testSubscribeAPI) Foo(ctx context.Context) (*rpc.Subscription, error) {
+	return SubscribeRPC(ctx, api.log, &api.foo)
+}
+
+func (api *testSubscribeAPI) Bar(ctx context.Context) (*rpc.Subscription, error) {
+	return SubscribeRPC(ctx, api.log, &api.bar)
+}
+
+func (api *testSubscribeAPI) GreetName(ctx context.Context, name string) (*rpc.Subscription, error) {
+	return nil, &rpc.JsonError{
+		Code:    -100_000,
+		Message: "hello " + name,
+		Data:    nil,
+	}
+}
+
+func TestSubscribeRPC(t *testing.T) {
+	logger := testlog.Logger(t, log.LevelDebug)
+	server := rpc.NewServer()
+	api := &testSubscribeAPI{
+		log: logger,
+	}
+	require.NoError(t, server.RegisterName("custom", api))
+
+	cl := rpc.DialInProc(server)
+
+	// Set up Foo subscription
+	ctx, fooCancel := context.WithCancel(context.Background())
+	fooCh := make(chan int, 10)
+	fooSub, err := cl.Subscribe(ctx, "custom", fooCh, "foo")
+	require.NoError(t, err)
+	api.foo.Send(123)
+	api.foo.Send(42)
+	api.bar.Send("x") // will be missed, we are not yet subscribed to "bar"
+	api.foo.Send(10)
+	for _, v := range []int{123, 42, 10} {
+		select {
+		case x := <-fooCh:
+			require.Equal(t, v, x)
+		case err := <-fooSub.Err():
+			require.NoError(t, err)
+		}
+	}
+
+	// setup Bar subscription
+	ctx, barCancel := context.WithCancel(context.Background())
+	barCh := make(chan string, 10)
+	barSub, err := cl.Subscribe(ctx, "custom", barCh, "bar")
+	require.NoError(t, err)
+	api.bar.Send("a")
+	api.foo.Send(20) // must not interfere
+	api.bar.Send("b")
+	api.bar.Send("c")
+	// "x" was sent before this subscription became active
+	for _, v := range []string{"a", "b", "c"} {
+		select {
+		case x := <-barCh:
+			require.Equal(t, v, x)
+		case err := <-barSub.Err():
+			require.NoError(t, err)
+		}
+	}
+
+	// pick up the item from Foo that we ignored in Bar
+	select {
+	case x := <-fooCh:
+		require.Equal(t, 20, x)
+	case err := <-fooSub.Err():
+		require.NoError(t, err)
+	}
+
+	barSub.Unsubscribe()
+	barCancel()
+
+	fooSub.Unsubscribe()
+	fooCancel()
+
+	// Try with a function argument, verify the function arg works
+	ctx, greetCancel := context.WithCancel(context.Background())
+	greetCh := make(chan string, 10)
+	_, err = cl.Subscribe(ctx, "custom", greetCh, "greetName", "alice")
+	var x rpc.Error
+	require.ErrorAs(t, err, &x)
+	require.Equal(t, x.Error(), "hello alice")
+
+	greetCancel()
+}

op-service/sources/eth_client.go

@@ -151,7 +151,7 @@ func NewEthClient(client client.RPC, log log.Logger, metrics caching.Metrics, co
 func (s *EthClient) SubscribeNewHead(ctx context.Context, ch chan<- *types.Header) (ethereum.Subscription, error) {
 	// Note that *types.Header does not cache the block hash unlike *HeaderInfo, it always recomputes.
 	// Inefficient if used poorly, but no trust issue.
-	return s.client.EthSubscribe(ctx, ch, "newHeads")
+	return s.client.Subscribe(ctx, "eth", ch, "newHeads")
 }
 
 // rpcBlockID is an internal type to enforce header and block call results match the requested identifier

op-service/sources/eth_client_test.go

@@ -34,8 +34,8 @@ func (m *mockRPC) CallContext(ctx context.Context, result any, method string, ar
 	return m.MethodCalled("CallContext", ctx, result, method, args).Get(0).([]error)[0]
 }
 
-func (m *mockRPC) EthSubscribe(ctx context.Context, channel any, args ...any) (ethereum.Subscription, error) {
-	called := m.MethodCalled("EthSubscribe", channel, args)
+func (m *mockRPC) Subscribe(ctx context.Context, namespace string, channel any, args ...any) (ethereum.Subscription, error) {
+	called := m.MethodCalled("Subscribe", namespace, channel, args)
 	return called.Get(0).(*rpc.ClientSubscription), called.Get(1).([]error)[0]
 }
 

op-service/sources/limit.go

@@ -65,13 +65,13 @@ func (lc *limitClient) CallContext(ctx context.Context, result any, method strin
 	return lc.c.CallContext(ctx, result, method, args...)
 }
 
-func (lc *limitClient) EthSubscribe(ctx context.Context, channel any, args ...any) (ethereum.Subscription, error) {
+func (lc *limitClient) Subscribe(ctx context.Context, namespace string, channel any, args ...any) (ethereum.Subscription, error) {
 	if !lc.joinWaitGroup() {
 		return nil, net.ErrClosed
 	}
 	defer lc.wg.Done()
 	// subscription doesn't count towards request limit
-	return lc.c.EthSubscribe(ctx, channel, args...)
+	return lc.c.Subscribe(ctx, namespace, channel, args...)
 }
 
 func (lc *limitClient) Close() {

op-service/sources/limit_test.go

@@ -34,8 +34,8 @@ func (m *MockRPC) BatchCallContext(ctx context.Context, b []rpc.BatchElem) error
 	return <-m.errC
 }
 
-func (m *MockRPC) EthSubscribe(ctx context.Context, channel any, args ...any) (ethereum.Subscription, error) {
-	m.t.Fatal("EthSubscribe should not be called")
+func (m *MockRPC) Subscribe(ctx context.Context, namespace string, channel any, args ...any) (ethereum.Subscription, error) {
+	m.t.Fatal("Subscribe should not be called")
 	return nil, nil
 }
 

op-service/testutils/mock_rpc.go

@@ -59,11 +59,11 @@ func (m *MockRPC) ExpectBatchCallContext(b []rpc.BatchElem, err error) {
 	}).Return(err)
 }
 
-func (m *MockRPC) EthSubscribe(ctx context.Context, channel any, args ...any) (ethereum.Subscription, error) {
-	out := m.Mock.Called(ctx, channel, args)
+func (m *MockRPC) Subscribe(ctx context.Context, namespace string, channel any, args ...any) (ethereum.Subscription, error) {
+	out := m.Mock.Called(ctx, namespace, channel, args)
 	return out.Get(0).(ethereum.Subscription), out.Error(1)
 }
 
-func (m *MockRPC) ExpectEthSubscribe(channel any, args []any, sub ethereum.Subscription, err error) {
-	m.Mock.On("EthSubscribe", mock.Anything, channel, args).Once().Return(sub, err)
+func (m *MockRPC) ExpectSubscribe(namespace string, channel any, args []any, sub ethereum.Subscription, err error) {
+	m.Mock.On("Subscribe", mock.Anything, namespace, channel, args).Once().Return(sub, err)
 }

op-service/testutils/rpc_err_faker.go

@@ -44,13 +44,13 @@ func (r RPCErrFaker) BatchCallContext(ctx context.Context, b []rpc.BatchElem) er
 	return r.RPC.BatchCallContext(ctx, b)
 }
 
-func (r RPCErrFaker) EthSubscribe(ctx context.Context, channel any, args ...any) (ethereum.Subscription, error) {
+func (r RPCErrFaker) Subscribe(ctx context.Context, namespace string, channel any, args ...any) (ethereum.Subscription, error) {
 	if r.ErrFn != nil {
 		if err := r.ErrFn(nil); err != nil {
 			return nil, err
 		}
 	}
-	return r.RPC.EthSubscribe(ctx, channel, args...)
+	return r.RPC.Subscribe(ctx, namespace, channel, args...)
 }
 
 var _ client.RPC = (*RPCErrFaker)(nil)