Port monitoring to resubscriptions.

op-challenger/config/config.go

@@ -78,6 +78,7 @@ func ValidTraceType(value TraceType) bool {
 }
 
 const (
+	DefaultPollInterval       = time.Second * 12
 	DefaultCannonSnapshotFreq = uint(1_000_000_000)
 	DefaultCannonInfoFreq     = uint(10_000_000)
 	// DefaultGameWindow is the default maximum time duration in the past
@@ -98,6 +99,7 @@ type Config struct {
 	AgreeWithProposedOutput bool             // Temporary config if we agree or disagree with the posted output
 	Datadir                 string           // Data Directory
 	MaxConcurrency          uint             // Maximum number of threads to use when progressing games
+	PollInterval            time.Duration    // Polling interval for latest-block subscription when using an HTTP RPC provider
 
 	TraceType TraceType // Type of trace
 
@@ -131,6 +133,7 @@ func NewConfig(
 		L1EthRpc:           l1EthRpc,
 		GameFactoryAddress: gameFactoryAddress,
 		MaxConcurrency:     uint(runtime.NumCPU()),
+		PollInterval:       DefaultPollInterval,
 
 		AgreeWithProposedOutput: agreeWithProposedOutput,
 

op-challenger/flags/flags.go

@@ -70,6 +70,12 @@ var (
 		EnvVars: prefixEnvVars("MAX_CONCURRENCY"),
 		Value:   uint(runtime.NumCPU()),
 	}
+	HTTPPollInterval = &cli.DurationFlag{
+		Name:    "http-poll-interval",
+		Usage:   "Polling interval for latest-block subscription when using an HTTP RPC provider.",
+		EnvVars: prefixEnvVars("HTTP_POLL_INTERVAL"),
+		Value:   config.DefaultPollInterval,
+	}
 	AlphabetFlag = &cli.StringFlag{
 		Name:    "alphabet",
 		Usage:   "Correct Alphabet Trace (alphabet trace type only)",
@@ -142,6 +148,7 @@ var requiredFlags = []cli.Flag{
 // optionalFlags is a list of unchecked cli flags
 var optionalFlags = []cli.Flag{
 	MaxConcurrencyFlag,
+	HTTPPollInterval,
 	AlphabetFlag,
 	GameAllowlistFlag,
 	CannonNetworkFlag,
@@ -247,6 +254,7 @@ func NewConfigFromCLI(ctx *cli.Context) (*config.Config, error) {
 		GameAllowlist:           allowedGames,
 		GameWindow:              ctx.Duration(GameWindowFlag.Name),
 		MaxConcurrency:          maxConcurrency,
+		PollInterval:            ctx.Duration(HTTPPollInterval.Name),
 		AlphabetTrace:           ctx.String(AlphabetFlag.Name),
 		CannonNetwork:           ctx.String(CannonNetworkFlag.Name),
 		CannonRollupConfigPath:  ctx.String(CannonRollupConfigFlag.Name),

op-challenger/game/monitor.go

@@ -9,7 +9,12 @@ import (
 
 	"github.com/ethereum-optimism/optimism/op-challenger/game/scheduler"
 	"github.com/ethereum-optimism/optimism/op-service/clock"
+	"github.com/ethereum-optimism/optimism/op-service/eth"
+
+	"github.com/ethereum/go-ethereum"
 	"github.com/ethereum/go-ethereum/common"
+	ethTypes "github.com/ethereum/go-ethereum/core/types"
+	"github.com/ethereum/go-ethereum/event"
 	"github.com/ethereum/go-ethereum/log"
 )
 
@@ -32,6 +37,20 @@ type gameMonitor struct {
 	gameWindow       time.Duration
 	fetchBlockNumber blockNumberFetcher
 	allowedGames     []common.Address
+	l1HeadsSub       ethereum.Subscription
+	l1Source         *headSource
+}
+
+type MinimalSubscriber interface {
+	EthSubscribe(ctx context.Context, channel interface{}, args ...interface{}) (ethereum.Subscription, error)
+}
+
+type headSource struct {
+	inner MinimalSubscriber
+}
+
+func (s *headSource) SubscribeNewHead(ctx context.Context, ch chan<- *ethTypes.Header) (ethereum.Subscription, error) {
+	return s.inner.EthSubscribe(ctx, ch, "newHeads")
 }
 
 func newGameMonitor(
@@ -42,6 +61,7 @@ func newGameMonitor(
 	gameWindow time.Duration,
 	fetchBlockNumber blockNumberFetcher,
 	allowedGames []common.Address,
+	l1Source MinimalSubscriber,
 ) *gameMonitor {
 	return &gameMonitor{
 		logger:           logger,
@@ -51,6 +71,7 @@ func newGameMonitor(
 		gameWindow:       gameWindow,
 		fetchBlockNumber: fetchBlockNumber,
 		allowedGames:     allowedGames,
+		l1Source:         &headSource{inner: l1Source},
 	}
 }
 
@@ -99,29 +120,32 @@ func (m *gameMonitor) progressGames(ctx context.Context, blockNum uint64) error
 	return nil
 }
 
-func (m *gameMonitor) MonitorGames(ctx context.Context) error {
-	m.logger.Info("Monitoring fault dispute games")
+func (m *gameMonitor) onNewL1Head(ctx context.Context, sig eth.L1BlockRef) {
+	if err := m.progressGames(ctx, sig.Number); err != nil {
+		m.logger.Error("Failed to progress games", "err", err)
+	}
+}
+
+func (m *gameMonitor) resubscribeFunction(ctx context.Context) event.ResubscribeErrFunc {
+	return func(innerCtx context.Context, err error) (event.Subscription, error) {
+		if err != nil {
+			m.logger.Warn("resubscribing after failed L1 subscription", "err", err)
+		}
+		return eth.WatchHeadChanges(ctx, m.l1Source, m.onNewL1Head)
+	}
+}
 
-	blockNum := uint64(0)
+func (m *gameMonitor) MonitorGames(ctx context.Context) error {
+	m.l1HeadsSub = event.ResubscribeErr(time.Second*10, m.resubscribeFunction(ctx))
 	for {
 		select {
 		case <-ctx.Done():
-			return ctx.Err()
-		default:
-			nextBlockNum, err := m.fetchBlockNumber(ctx)
-			if err != nil {
-				m.logger.Error("Failed to load current block number", "err", err)
-				continue
-			}
-			if nextBlockNum > blockNum {
-				blockNum = nextBlockNum
-				if err := m.progressGames(ctx, nextBlockNum); err != nil {
-					m.logger.Error("Failed to progress games", "err", err)
-				}
-			}
-			if err := m.clock.SleepCtx(ctx, time.Second); err != nil {
+			return nil
+		case err, ok := <-m.l1HeadsSub.Err():
+			if !ok {
 				return err
 			}
+			m.logger.Error("L1 subscription error", "err", err)
 		}
 	}
 }

op-challenger/game/monitor_test.go

@@ -2,35 +2,40 @@ package game
 
 import (
 	"context"
+	"fmt"
 	"math/big"
 	"testing"
 	"time"
 
-	"github.com/ethereum-optimism/optimism/op-node/testlog"
-	"github.com/ethereum-optimism/optimism/op-service/clock"
+	"github.com/ethereum/go-ethereum"
 	"github.com/ethereum/go-ethereum/common"
+	ethtypes "github.com/ethereum/go-ethereum/core/types"
 	"github.com/ethereum/go-ethereum/log"
 	"github.com/stretchr/testify/require"
+
+	"github.com/ethereum-optimism/optimism/op-e2e/e2eutils/wait"
+	"github.com/ethereum-optimism/optimism/op-node/testlog"
+	"github.com/ethereum-optimism/optimism/op-service/clock"
 )
 
 func TestMonitorMinGameTimestamp(t *testing.T) {
 	t.Parallel()
 
 	t.Run("zero game window returns zero", func(t *testing.T) {
-		monitor, _, _ := setupMonitorTest(t, []common.Address{})
+		monitor, _, _, _ := setupMonitorTest(t, []common.Address{})
 		monitor.gameWindow = time.Duration(0)
 		require.Equal(t, monitor.minGameTimestamp(), uint64(0))
 	})
 
 	t.Run("non-zero game window with zero clock", func(t *testing.T) {
-		monitor, _, _ := setupMonitorTest(t, []common.Address{})
+		monitor, _, _, _ := setupMonitorTest(t, []common.Address{})
 		monitor.gameWindow = time.Minute
 		monitor.clock = clock.NewDeterministicClock(time.Unix(0, 0))
 		require.Equal(t, monitor.minGameTimestamp(), uint64(0))
 	})
 
 	t.Run("minimum computed correctly", func(t *testing.T) {
-		monitor, _, _ := setupMonitorTest(t, []common.Address{})
+		monitor, _, _, _ := setupMonitorTest(t, []common.Address{})
 		monitor.gameWindow = time.Minute
 		frozen := time.Unix(int64(time.Hour.Seconds()), 0)
 		monitor.clock = clock.NewDeterministicClock(frozen)
@@ -39,29 +44,95 @@ func TestMonitorMinGameTimestamp(t *testing.T) {
 	})
 }
 
-func TestMonitorExitsWhenContextDone(t *testing.T) {
-	monitor, _, _ := setupMonitorTest(t, []common.Address{{}})
-	ctx, cancel := context.WithCancel(context.Background())
-	cancel()
-	err := monitor.MonitorGames(ctx)
-	require.ErrorIs(t, err, context.Canceled)
+// TestMonitorGames tests that the monitor can handle a new head event
+// and resubscribe to new heads if the subscription errors.
+func TestMonitorGames(t *testing.T) {
+	t.Run("Schedules games", func(t *testing.T) {
+		addr1 := common.Address{0xaa}
+		addr2 := common.Address{0xbb}
+		monitor, source, sched, mockHeadSource := setupMonitorTest(t, []common.Address{})
+		source.games = []FaultDisputeGame{newFDG(addr1, 9999), newFDG(addr2, 9999)}
+
+		ctx, cancel := context.WithCancel(context.Background())
+		defer cancel()
+
+		go func() {
+			headerNotSent := true
+			waitErr := wait.For(context.Background(), 100*time.Millisecond, func() (bool, error) {
+				if len(sched.scheduled) >= 1 {
+					return true, nil
+				}
+				if mockHeadSource.sub == nil {
+					return false, nil
+				}
+				if headerNotSent {
+					mockHeadSource.sub.headers <- &ethtypes.Header{
+						Number: big.NewInt(1),
+					}
+					headerNotSent = false
+				}
+				return false, nil
+			})
+			require.NoError(t, waitErr)
+			mockHeadSource.err = fmt.Errorf("eth subscribe test error")
+			cancel()
+		}()
+
+		err := monitor.MonitorGames(ctx)
+		require.NoError(t, err)
+		require.Len(t, sched.scheduled, 1)
+		require.Equal(t, []common.Address{addr1, addr2}, sched.scheduled[0])
+	})
+
+	t.Run("Resubscribes on error", func(t *testing.T) {
+		addr1 := common.Address{0xaa}
+		addr2 := common.Address{0xbb}
+		monitor, source, sched, mockHeadSource := setupMonitorTest(t, []common.Address{})
+		source.games = []FaultDisputeGame{newFDG(addr1, 9999), newFDG(addr2, 9999)}
+
+		ctx, cancel := context.WithCancel(context.Background())
+		defer cancel()
+
+		go func() {
+			headerNotSent := true
+			waitErr := wait.For(context.Background(), 100*time.Millisecond, func() (bool, error) {
+				return mockHeadSource.sub != nil, nil
+			})
+			require.NoError(t, waitErr)
+			mockHeadSource.sub.errChan <- fmt.Errorf("test error")
+			waitErr = wait.For(context.Background(), 100*time.Millisecond, func() (bool, error) {
+				if len(sched.scheduled) >= 1 {
+					return true, nil
+				}
+				if mockHeadSource.sub == nil {
+					return false, nil
+				}
+				if headerNotSent {
+					mockHeadSource.sub.headers <- &ethtypes.Header{
+						Number: big.NewInt(1),
+					}
+					headerNotSent = false
+				}
+				return false, nil
+			})
+			require.NoError(t, waitErr)
+			mockHeadSource.err = fmt.Errorf("eth subscribe test error")
+			cancel()
+		}()
+
+		err := monitor.MonitorGames(ctx)
+		require.NoError(t, err)
+		require.Len(t, sched.scheduled, 1)
+		require.Equal(t, []common.Address{addr1, addr2}, sched.scheduled[0])
+	})
 }
 
 func TestMonitorCreateAndProgressGameAgents(t *testing.T) {
-	monitor, source, sched := setupMonitorTest(t, []common.Address{})
+	monitor, source, sched, _ := setupMonitorTest(t, []common.Address{})
 
 	addr1 := common.Address{0xaa}
 	addr2 := common.Address{0xbb}
-	source.games = []FaultDisputeGame{
-		{
-			Proxy:     addr1,
-			Timestamp: 9999,
-		},
-		{
-			Proxy:     addr2,
-			Timestamp: 9999,
-		},
-	}
+	source.games = []FaultDisputeGame{newFDG(addr1, 9999), newFDG(addr2, 9999)}
 
 	require.NoError(t, monitor.progressGames(context.Background(), uint64(1)))
 
@@ -72,18 +143,8 @@ func TestMonitorCreateAndProgressGameAgents(t *testing.T) {
 func TestMonitorOnlyScheduleSpecifiedGame(t *testing.T) {
 	addr1 := common.Address{0xaa}
 	addr2 := common.Address{0xbb}
-	monitor, source, sched := setupMonitorTest(t, []common.Address{addr2})
-
-	source.games = []FaultDisputeGame{
-		{
-			Proxy:     addr1,
-			Timestamp: 9999,
-		},
-		{
-			Proxy:     addr2,
-			Timestamp: 9999,
-		},
-	}
+	monitor, source, sched, _ := setupMonitorTest(t, []common.Address{addr2})
+	source.games = []FaultDisputeGame{newFDG(addr1, 9999), newFDG(addr2, 9999)}
 
 	require.NoError(t, monitor.progressGames(context.Background(), uint64(1)))
 
@@ -91,7 +152,17 @@ func TestMonitorOnlyScheduleSpecifiedGame(t *testing.T) {
 	require.Equal(t, []common.Address{addr2}, sched.scheduled[0])
 }
 
-func setupMonitorTest(t *testing.T, allowedGames []common.Address) (*gameMonitor, *stubGameSource, *stubScheduler) {
+func newFDG(proxy common.Address, timestamp uint64) FaultDisputeGame {
+	return FaultDisputeGame{
+		Proxy:     proxy,
+		Timestamp: timestamp,
+	}
+}
+
+func setupMonitorTest(
+	t *testing.T,
+	allowedGames []common.Address,
+) (*gameMonitor, *stubGameSource, *stubScheduler, *mockNewHeadSource) {
 	logger := testlog.Logger(t, log.LvlDebug)
 	source := &stubGameSource{}
 	i := uint64(1)
@@ -100,15 +171,58 @@ func setupMonitorTest(t *testing.T, allowedGames []common.Address) (*gameMonitor
 		return i, nil
 	}
 	sched := &stubScheduler{}
-	monitor := newGameMonitor(logger, clock.SystemClock, source, sched, time.Duration(0), fetchBlockNum, allowedGames)
-	return monitor, source, sched
+	mockHeadSource := &mockNewHeadSource{}
+	monitor := newGameMonitor(
+		logger,
+		clock.SystemClock,
+		source,
+		sched,
+		time.Duration(0),
+		fetchBlockNum,
+		allowedGames,
+		mockHeadSource,
+	)
+	return monitor, source, sched, mockHeadSource
+}
+
+type mockNewHeadSource struct {
+	sub *mockSubscription
+	err error
+}
+
+func (m *mockNewHeadSource) EthSubscribe(
+	ctx context.Context,
+	ch any,
+	args ...any,
+) (ethereum.Subscription, error) {
+	errChan := make(chan error)
+	m.sub = &mockSubscription{errChan, (ch).(chan<- *ethtypes.Header)}
+	if m.err != nil {
+		return nil, m.err
+	}
+	return m.sub, nil
+}
+
+type mockSubscription struct {
+	errChan chan error
+	headers chan<- *ethtypes.Header
+}
+
+func (m *mockSubscription) Unsubscribe() {}
+
+func (m *mockSubscription) Err() <-chan error {
+	return m.errChan
 }
 
 type stubGameSource struct {
 	games []FaultDisputeGame
 }
 
-func (s *stubGameSource) FetchAllGamesAtBlock(ctx context.Context, earliest uint64, blockNumber *big.Int) ([]FaultDisputeGame, error) {
+func (s *stubGameSource) FetchAllGamesAtBlock(
+	ctx context.Context,
+	earliest uint64,
+	blockNumber *big.Int,
+) ([]FaultDisputeGame, error) {
 	return s.games, nil
 }
 

op-challenger/game/service.go

@@ -34,7 +34,7 @@ func NewService(ctx context.Context, logger log.Logger, cfg *config.Config) (*Se
 		return nil, fmt.Errorf("failed to create the transaction manager: %w", err)
 	}
 
-	client, err := client.DialEthClientWithTimeout(client.DefaultDialTimeout, logger, cfg.L1EthRpc)
+	l1Client, err := client.DialEthClientWithTimeout(client.DefaultDialTimeout, logger, cfg.L1EthRpc)
 	if err != nil {
 		return nil, fmt.Errorf("failed to dial L1: %w", err)
 	}
@@ -57,10 +57,10 @@ func NewService(ctx context.Context, logger log.Logger, cfg *config.Config) (*Se
 				logger.Error("error starting metrics server", "err", err)
 			}
 		}()
-		m.StartBalanceMetrics(ctx, logger, client, txMgr.From())
+		m.StartBalanceMetrics(ctx, logger, l1Client, txMgr.From())
 	}
 
-	factory, err := bindings.NewDisputeGameFactory(cfg.GameFactoryAddress, client)
+	factory, err := bindings.NewDisputeGameFactory(cfg.GameFactoryAddress, l1Client)
 	if err != nil {
 		return nil, fmt.Errorf("failed to bind the fault dispute game factory contract: %w", err)
 	}
@@ -73,10 +73,15 @@ func NewService(ctx context.Context, logger log.Logger, cfg *config.Config) (*Se
 		disk,
 		cfg.MaxConcurrency,
 		func(addr common.Address, dir string) (scheduler.GamePlayer, error) {
-			return fault.NewGamePlayer(ctx, logger, m, cfg, dir, addr, txMgr, client)
+			return fault.NewGamePlayer(ctx, logger, m, cfg, dir, addr, txMgr, l1Client)
 		})
 
-	monitor := newGameMonitor(logger, cl, loader, sched, cfg.GameWindow, client.BlockNumber, cfg.GameAllowlist)
+	polledClient, err := client.DialPolledClientWithTimeout(ctx, client.DefaultDialTimeout, logger, cfg.L1EthRpc, cfg.PollInterval)
+	if err != nil {
+		return nil, fmt.Errorf("failed to dial L1: %w", err)
+	}
+
+	monitor := newGameMonitor(logger, cl, loader, sched, cfg.GameWindow, l1Client.BlockNumber, cfg.GameAllowlist, polledClient)
 
 	m.RecordInfo(version.SimpleWithMeta)
 	m.RecordUp()

op-e2e/e2eutils/challenger/helper.go

@@ -145,6 +145,10 @@ func NewChallengerConfig(t *testing.T, l1Endpoint string, options ...Option) *co
 		_, err := os.Stat(cfg.CannonAbsolutePreState)
 		require.NoError(t, err, "cannon pre-state should be built. Make sure you've run make cannon-prestate")
 	}
+	if cfg.PollInterval == 0 {
+		cfg.PollInterval = 2 * time.Second
+	}
+
 	return &cfg
 }
 

op-e2e/e2eutils/disputegame/game_helper.go

@@ -65,7 +65,7 @@ func (g *FaultGameHelper) MaxDepth(ctx context.Context) int64 {
 }
 
 func (g *FaultGameHelper) waitForClaim(ctx context.Context, errorMsg string, predicate func(claim ContractClaim) bool) {
-	timedCtx, cancel := context.WithTimeout(ctx, time.Minute)
+	timedCtx, cancel := context.WithTimeout(ctx, 2*time.Minute)
 	defer cancel()
 	err := wait.For(timedCtx, time.Second, func() (bool, error) {
 		count, err := g.game.ClaimDataLen(&bind.CallOpts{Context: timedCtx})

op-e2e/e2eutils/disputegame/helper.go

@@ -100,7 +100,7 @@ func (h *FactoryHelper) StartAlphabetGame(ctx context.Context, claimedAlphabet s
 	l2BlockNumber := h.waitForProposals(ctx)
 	l1Head := h.checkpointL1Block(ctx)
 
-	ctx, cancel := context.WithTimeout(ctx, 1*time.Minute)
+	ctx, cancel := context.WithTimeout(ctx, 2*time.Minute)
 	defer cancel()
 
 	trace := alphabet.NewTraceProvider(claimedAlphabet, alphabetGameDepth)

op-service/client/dial.go

@@ -13,6 +13,9 @@ import (
 	"github.com/ethereum/go-ethereum/rpc"
 )
 
+// RPC is re-exported from op-node/client.
+type RPC client.RPC
+
 // DefaultDialTimeout is a default timeout for dialing a client.
 const DefaultDialTimeout = 1 * time.Minute
 const defaultRetryCount = 30
@@ -47,15 +50,25 @@ func DialRollupClientWithTimeout(timeout time.Duration, log log.Logger, url stri
 	return sources.NewRollupClient(client.NewBaseRPCClient(rpcCl)), nil
 }
 
+// DialPolledClientWithTimeout attempts to dial the RPC provider using the provided URL.
+// If the dial doesn't complete within timeout seconds, this method will return an error.
+func DialPolledClientWithTimeout(ctx context.Context, timeout time.Duration, log log.Logger, url string, pollInterval time.Duration) (client.RPC, error) {
+	opts := []client.RPCOption{
+		client.WithHttpPollInterval(pollInterval),
+		client.WithDialBackoff(defaultRetryCount),
+	}
+	return client.NewRPC(ctx, log, url, opts...)
+}
+
 // Dials a JSON-RPC endpoint repeatedly, with a backoff, until a client connection is established. Auth is optional.
-func dialRPCClientWithBackoff(ctx context.Context, log log.Logger, addr string) (*rpc.Client, error) {
+func dialRPCClientWithBackoff(ctx context.Context, log log.Logger, addr string, opts ...rpc.ClientOption) (*rpc.Client, error) {
 	bOff := retry.Fixed(defaultRetryTime)
 	return retry.Do(ctx, defaultRetryCount, bOff, func() (*rpc.Client, error) {
 		if !client.IsURLAvailable(addr) {
 			log.Warn("failed to dial address, but may connect later", "addr", addr)
 			return nil, fmt.Errorf("address unavailable (%s)", addr)
 		}
-		client, err := rpc.DialOptions(ctx, addr)
+		client, err := rpc.DialOptions(ctx, addr, opts...)
 		if err != nil {
 			return nil, fmt.Errorf("failed to dial address (%s): %w", addr, err)
 		}