op-program: Implement derivation driver

Hooks up the L1 and L2 sources with the derivation pipeline so the program can run successfully in fetching mode.

op-program/cmd/main.go

@@ -4,10 +4,14 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"io"
 	"os"
+	"time"
 
 	"github.com/ethereum-optimism/optimism/op-node/chaincfg"
+	"github.com/ethereum-optimism/optimism/op-node/rollup/derive"
 	"github.com/ethereum-optimism/optimism/op-program/config"
+	"github.com/ethereum-optimism/optimism/op-program/driver"
 	"github.com/ethereum-optimism/optimism/op-program/flags"
 	"github.com/ethereum-optimism/optimism/op-program/l1"
 	"github.com/ethereum-optimism/optimism/op-program/l2"
@@ -96,15 +100,30 @@ func FaultProofProgram(logger log.Logger, cfg *config.Config) error {
 
 	ctx := context.Background()
 	logger.Info("Connecting to L1 node", "l1", cfg.L1URL)
-	_, err := l1.NewFetchingL1(ctx, logger, cfg)
+	l1Source, err := l1.NewFetchingL1(ctx, logger, cfg)
 	if err != nil {
 		return fmt.Errorf("connect l1 oracle: %w", err)
 	}
 
 	logger.Info("Connecting to L2 node", "l2", cfg.L2URL)
-	_, err = l2.NewFetchingEngine(ctx, logger, cfg)
+	l2Source, err := l2.NewFetchingEngine(ctx, logger, cfg)
 	if err != nil {
 		return fmt.Errorf("connect l2 oracle: %w", err)
 	}
+
+	d := driver.NewDriver(logger, cfg, l1Source, l2Source)
+	for {
+		if err = d.Step(ctx); errors.Is(err, io.EOF) {
+			break
+		} else if cfg.FetchingEnabled() && errors.Is(err, derive.ErrTemporary) {
+			// When in fetching mode, recover from temporary errors to allow us to keep fetching data
+			// TODO(CLI-3780) Ideally the retry would happen in the fetcher so this is not needed
+			logger.Warn("Temporary error in pipeline", "err", err)
+			time.Sleep(5 * time.Second)
+		} else if err != nil {
+			return err
+		}
+	}
+	logger.Info("Derivation complete", "head", d.SafeHead())
 	return nil
 }

op-program/driver/driver.go

+package driver
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"io"
+
+	"github.com/ethereum-optimism/optimism/op-node/eth"
+	"github.com/ethereum-optimism/optimism/op-node/metrics"
+	"github.com/ethereum-optimism/optimism/op-node/rollup/derive"
+	"github.com/ethereum-optimism/optimism/op-program/config"
+	"github.com/ethereum/go-ethereum/log"
+)
+
+type Derivation interface {
+	Step(ctx context.Context) error
+	SafeL2Head() eth.L2BlockRef
+}
+
+type Driver struct {
+	logger   log.Logger
+	pipeline Derivation
+}
+
+func NewDriver(logger log.Logger, cfg *config.Config, l1Source derive.L1Fetcher, l2Source derive.Engine) *Driver {
+	pipeline := derive.NewDerivationPipeline(logger, cfg.Rollup, l1Source, l2Source, metrics.NoopMetrics)
+	pipeline.Reset()
+	return &Driver{
+		logger:   logger,
+		pipeline: pipeline,
+	}
+}
+
+// Step runs the next step of the derivation pipeline.
+// Returns nil if there are further steps to be performed
+// Returns io.EOF if the derivation completed successfully
+// Returns a non-EOF error if the derivation failed
+func (d *Driver) Step(ctx context.Context) error {
+	if err := d.pipeline.Step(ctx); errors.Is(err, io.EOF) {
+		return io.EOF
+	} else if errors.Is(err, derive.NotEnoughData) {
+		d.logger.Debug("Data is lacking")
+		return nil
+	} else if err != nil {
+		return fmt.Errorf("pipeline err: %w", err)
+	}
+	return nil
+}
+
+func (d *Driver) SafeHead() eth.L2BlockRef {
+	return d.pipeline.SafeL2Head()
+}

op-program/driver/driver_test.go

+package driver
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"io"
+	"testing"
+
+	"github.com/ethereum-optimism/optimism/op-node/eth"
+	"github.com/ethereum-optimism/optimism/op-node/rollup/derive"
+	"github.com/ethereum-optimism/optimism/op-node/testlog"
+	"github.com/ethereum/go-ethereum/log"
+	"github.com/stretchr/testify/require"
+)
+
+func TestDerivationComplete(t *testing.T) {
+	driver := createDriver(t, fmt.Errorf("derivation complete: %w", io.EOF))
+	err := driver.Step(context.Background())
+	require.ErrorIs(t, err, io.EOF)
+}
+
+func TestTemporaryError(t *testing.T) {
+	driver := createDriver(t, fmt.Errorf("whoopsie: %w", derive.ErrTemporary))
+	err := driver.Step(context.Background())
+	require.ErrorIs(t, err, derive.ErrTemporary)
+}
+
+func TestNotEnoughDataError(t *testing.T) {
+	driver := createDriver(t, fmt.Errorf("idk: %w", derive.NotEnoughData))
+	err := driver.Step(context.Background())
+	require.NoError(t, err)
+}
+
+func TestGenericError(t *testing.T) {
+	expected := errors.New("boom")
+	driver := createDriver(t, expected)
+	err := driver.Step(context.Background())
+	require.ErrorIs(t, err, expected)
+}
+
+func TestNoError(t *testing.T) {
+	driver := createDriver(t, nil)
+	err := driver.Step(context.Background())
+	require.NoError(t, err)
+}
+
+func createDriver(t *testing.T, derivationResult error) *Driver {
+	derivation := &stubDerivation{nextErr: derivationResult}
+	return &Driver{
+		logger:   testlog.Logger(t, log.LvlDebug),
+		pipeline: derivation,
+	}
+}
+
+type stubDerivation struct {
+	nextErr error
+}
+
+func (s stubDerivation) Step(ctx context.Context) error {
+	return s.nextErr
+}
+
+func (s stubDerivation) SafeL2Head() eth.L2BlockRef {
+	return eth.L2BlockRef{}
+}