op-service/dial: Add WaitRollupSync (#10116)

This is going to be used by the batcher and proposer in a follow up PR to wait at startup for the sequencer to sync to tip.

op-service/dial/rollup_sync.go

+package dial
+
+import (
+	"context"
+	"fmt"
+	"time"
+
+	"github.com/ethereum/go-ethereum/log"
+)
+
+func WaitRollupSync(
+	ctx context.Context,
+	lgr log.Logger,
+	rollup SyncStatusProvider,
+	l1BlockTarget uint64,
+	pollDuration time.Duration,
+) error {
+	for {
+		syncst, err := rollup.SyncStatus(ctx)
+		if err != nil {
+			// don't log assuming caller handles and logs errors
+			return fmt.Errorf("getting sync status: %w", err)
+		}
+
+		lgr := lgr.With("current_l1", syncst.CurrentL1, "target_l1", l1BlockTarget)
+		if syncst.CurrentL1.Number >= l1BlockTarget {
+			lgr.Info("rollup current L1 block target reached")
+			return nil
+		}
+
+		lgr.Info("rollup current L1 block still behind target, retrying")
+		timer := time.NewTimer(pollDuration)
+		select {
+		case <-timer.C: // next try
+		case <-ctx.Done():
+			lgr.Warn("waiting for rollup sync timed out")
+			timer.Stop()
+			return ctx.Err()
+		}
+	}
+}

op-service/dial/rollup_sync_test.go

+package dial
+
+import (
+	"context"
+	"errors"
+	"testing"
+	"time"
+
+	"github.com/ethereum-optimism/optimism/op-service/eth"
+	"github.com/ethereum-optimism/optimism/op-service/testlog"
+	"github.com/ethereum-optimism/optimism/op-service/testutils"
+	"github.com/stretchr/testify/require"
+	"golang.org/x/exp/slog"
+)
+
+func TestWaitRollupSync(t *testing.T) {
+	bctx := context.Background()
+	hasInfoLevel := testlog.NewLevelFilter(slog.LevelInfo)
+	const target = 42
+
+	t.Run("sync-error", func(t *testing.T) {
+		lgr, logs := testlog.CaptureLogger(t, slog.LevelInfo)
+		rollup := new(testutils.MockRollupClient)
+		syncErr := errors.New("test sync error")
+		rollup.ExpectSyncStatus(nil, syncErr)
+
+		err := WaitRollupSync(bctx, lgr, rollup, 0, 0)
+		require.ErrorIs(t, err, syncErr)
+		require.Nil(t, logs.FindLog(hasInfoLevel), "expected no logs")
+		rollup.AssertExpectations(t)
+	})
+
+	t.Run("at-target", func(t *testing.T) {
+		lgr, logs := testlog.CaptureLogger(t, slog.LevelDebug)
+		rollup := new(testutils.MockRollupClient)
+		rollup.ExpectSyncStatus(&eth.SyncStatus{
+			CurrentL1: eth.L1BlockRef{Number: target},
+		}, nil)
+
+		err := WaitRollupSync(bctx, lgr, rollup, target, 0)
+		require.NoError(t, err)
+		require.NotNil(t, logs.FindLog(hasInfoLevel,
+			testlog.NewMessageContainsFilter("target reached")))
+		rollup.AssertExpectations(t)
+	})
+
+	t.Run("beyond-target", func(t *testing.T) {
+		lgr, logs := testlog.CaptureLogger(t, slog.LevelDebug)
+		rollup := new(testutils.MockRollupClient)
+		rollup.ExpectSyncStatus(&eth.SyncStatus{
+			CurrentL1: eth.L1BlockRef{Number: target + 12},
+		}, nil)
+
+		err := WaitRollupSync(bctx, lgr, rollup, target, 0)
+		require.NoError(t, err)
+		require.NotNil(t, logs.FindLog(hasInfoLevel,
+			testlog.NewMessageContainsFilter("target reached")))
+		rollup.AssertExpectations(t)
+	})
+
+	t.Run("few-blocks-before-target", func(t *testing.T) {
+		lgr, logs := testlog.CaptureLogger(t, slog.LevelDebug)
+		rollup := new(testutils.MockRollupClient)
+		const gap = 7
+		for i := -gap; i <= 0; i++ {
+			rollup.ExpectSyncStatus(&eth.SyncStatus{
+				CurrentL1: eth.L1BlockRef{Number: uint64(target + i)},
+			}, nil)
+		}
+
+		err := WaitRollupSync(bctx, lgr, rollup, target, 0)
+		require.NoError(t, err)
+		require.NotNil(t, logs.FindLog(hasInfoLevel,
+			testlog.NewMessageContainsFilter("target reached")))
+		require.Len(t, logs.FindLogs(hasInfoLevel,
+			testlog.NewMessageContainsFilter("retrying")), gap)
+		rollup.AssertExpectations(t)
+	})
+
+	t.Run("ctx-timeout", func(t *testing.T) {
+		lgr, logs := testlog.CaptureLogger(t, slog.LevelDebug)
+		rollup := new(testutils.MockRollupClient)
+		rollup.ExpectSyncStatus(&eth.SyncStatus{
+			CurrentL1: eth.L1BlockRef{Number: uint64(target - 1)},
+		}, nil)
+
+		ctx, cancel := context.WithCancel(bctx)
+		// We can already cancel the context because the mock ignores the
+		// cancelled context.
+		cancel()
+		// need real duration or the timer races with the cancelled context
+		err := WaitRollupSync(ctx, lgr, rollup, target, time.Second)
+		require.ErrorIs(t, err, context.Canceled)
+		require.NotNil(t, logs.FindLogs(hasInfoLevel,
+			testlog.NewMessageContainsFilter("retrying")))
+		require.NotNil(t, logs.FindLog(
+			testlog.NewLevelFilter(slog.LevelWarn),
+			testlog.NewMessageContainsFilter("timed out")))
+		rollup.AssertExpectations(t)
+	})
+}

op-service/dial/rollupclient_interface.go

@@ -11,10 +11,16 @@ import (
 // RollupClientInterface is an interface for providing a RollupClient
 // It does not describe all of the functions a RollupClient has, only the ones used by the L2 Providers and their callers
 type RollupClientInterface interface {
+	SyncStatusProvider
 	OutputAtBlock(ctx context.Context, blockNum uint64) (*eth.OutputResponse, error)
-	SyncStatus(ctx context.Context) (*eth.SyncStatus, error)
 	RollupConfig(ctx context.Context) (*rollup.Config, error)
 	StartSequencer(ctx context.Context, unsafeHead common.Hash) error
 	SequencerActive(ctx context.Context) (bool, error)
 	Close()
 }
+
+// SyncStatusProvider is the interface of a rollup client from which its sync status
+// can be queried.
+type SyncStatusProvider interface {
+	SyncStatus(ctx context.Context) (*eth.SyncStatus, error)
+}