Merge pull request #5492 from ethereum-optimism/inphi/fpp-comms

op-program: host-client interaction via I/O pipes

op-e2e/system_fpp_test.go

@@ -9,6 +9,7 @@ import (
 	"github.com/ethereum-optimism/optimism/op-node/client"
 	"github.com/ethereum-optimism/optimism/op-node/sources"
 	"github.com/ethereum-optimism/optimism/op-node/testlog"
+	oppcl "github.com/ethereum-optimism/optimism/op-program/client"
 	opp "github.com/ethereum-optimism/optimism/op-program/host"
 	oppconf "github.com/ethereum-optimism/optimism/op-program/host/config"
 	"github.com/ethereum/go-ethereum/accounts/abi/bind"
@@ -118,7 +119,7 @@ func TestVerifyL2OutputRoot(t *testing.T) {
 	t.Log("Running fault proof with invalid claim")
 	fppConfig.L2Claim = common.Hash{0xaa}
 	err = opp.FaultProofProgram(log, fppConfig)
-	require.ErrorIs(t, err, opp.ErrClaimNotValid)
+	require.ErrorIs(t, err, oppcl.ErrClaimNotValid)
 }
 
 func waitForSafeHead(ctx context.Context, safeBlockNum uint64, rollupClient *sources.RollupClient) error {

op-program/client/program.go

+package client
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"io"
+
+	"github.com/ethereum/go-ethereum/common"
+	"github.com/ethereum/go-ethereum/log"
+	"github.com/ethereum/go-ethereum/params"
+
+	"github.com/ethereum-optimism/optimism/op-node/eth"
+	"github.com/ethereum-optimism/optimism/op-node/rollup"
+	cldr "github.com/ethereum-optimism/optimism/op-program/client/driver"
+	"github.com/ethereum-optimism/optimism/op-program/client/l1"
+	"github.com/ethereum-optimism/optimism/op-program/client/l2"
+	"github.com/ethereum-optimism/optimism/op-program/preimage"
+)
+
+var (
+	ErrClaimNotValid = errors.New("invalid claim")
+)
+
+// ClientProgram executes the Program, while attached to an IO based pre-image oracle, to be served by a host.
+func ClientProgram(
+	logger log.Logger,
+	cfg *rollup.Config,
+	l2Cfg *params.ChainConfig,
+	l1Head common.Hash,
+	l2Head common.Hash,
+	l2Claim common.Hash,
+	l2ClaimBlockNumber uint64,
+	preimageOracle io.ReadWriter,
+	preimageHinter io.ReadWriter,
+) error {
+	pClient := preimage.NewOracleClient(preimageOracle)
+	hClient := preimage.NewHintWriter(preimageHinter)
+	l1PreimageOracle := l1.NewPreimageOracle(pClient, hClient)
+	l2PreimageOracle := l2.NewPreimageOracle(pClient, hClient)
+	return Program(logger, cfg, l2Cfg, l1Head, l2Head, l2Claim, l2ClaimBlockNumber, l1PreimageOracle, l2PreimageOracle)
+}
+
+// Program executes the L2 state transition, given a minimal interface to retrieve data.
+func Program(logger log.Logger, cfg *rollup.Config, l2Cfg *params.ChainConfig, l1Head common.Hash, l2Head common.Hash, l2Claim common.Hash, l2ClaimBlockNum uint64, l1Oracle l1.Oracle, l2Oracle l2.Oracle) error {
+	l1Source := l1.NewOracleL1Client(logger, l1Oracle, l1Head)
+	engineBackend, err := l2.NewOracleBackedL2Chain(logger, l2Oracle, l2Cfg, l2Head)
+	if err != nil {
+		return fmt.Errorf("failed to create oracle-backed L2 chain: %w", err)
+	}
+	l2Source := l2.NewOracleEngine(cfg, logger, engineBackend)
+
+	logger.Info("Starting derivation")
+	d := cldr.NewDriver(logger, cfg, l1Source, l2Source, l2ClaimBlockNum)
+	for {
+		if err = d.Step(context.Background()); errors.Is(err, io.EOF) {
+			break
+		} else if err != nil {
+			return err
+		}
+	}
+	if !d.ValidateClaim(eth.Bytes32(l2Claim)) {
+		return ErrClaimNotValid
+	}
+	logger.Info("Derivation complete", "head", d.SafeHead())
+	return nil
+}

op-program/host/host.go

@@ -9,23 +9,16 @@ import (
 
 	"github.com/ethereum-optimism/optimism/op-node/chaincfg"
 	"github.com/ethereum-optimism/optimism/op-node/client"
-	"github.com/ethereum-optimism/optimism/op-node/eth"
 	"github.com/ethereum-optimism/optimism/op-node/sources"
-	cldr "github.com/ethereum-optimism/optimism/op-program/client/driver"
+	cl "github.com/ethereum-optimism/optimism/op-program/client"
 	"github.com/ethereum-optimism/optimism/op-program/host/config"
 	"github.com/ethereum-optimism/optimism/op-program/host/kvstore"
-	"github.com/ethereum-optimism/optimism/op-program/host/l1"
-	"github.com/ethereum-optimism/optimism/op-program/host/l2"
 	"github.com/ethereum-optimism/optimism/op-program/host/prefetcher"
 	"github.com/ethereum-optimism/optimism/op-program/preimage"
 	"github.com/ethereum/go-ethereum/common"
 	"github.com/ethereum/go-ethereum/log"
 )
 
-var (
-	ErrClaimNotValid = errors.New("invalid claim")
-)
-
 type L2Source struct {
 	*sources.L2Client
 	*sources.DebugClient
@@ -51,8 +44,8 @@ func FaultProofProgram(logger log.Logger, cfg *config.Config) error {
 		kv = kvstore.NewDiskKV(cfg.DataDir)
 	}
 
-	var preimageOracle preimage.OracleFn
-	var hinter preimage.HinterFn
+	var getPreimage func(key common.Hash) ([]byte, error)
+	var hinter func(hint string) error
 	if cfg.FetchingEnabled() {
 		logger.Info("Connecting to L1 node", "l1", cfg.L1URL)
 		l1RPC, err := client.NewRPC(ctx, logger, cfg.L1URL)
@@ -80,54 +73,85 @@ func FaultProofProgram(logger log.Logger, cfg *config.Config) error {
 
 		logger.Info("Setting up pre-fetcher")
 		prefetch := prefetcher.NewPrefetcher(logger, l1Cl, l2DebugCl, kv)
-		preimageOracle = asOracleFn(func(key common.Hash) ([]byte, error) {
-			return prefetch.GetPreimage(ctx, key)
-		})
-		hinter = asHinter(prefetch.Hint)
+		getPreimage = func(key common.Hash) ([]byte, error) { return prefetch.GetPreimage(ctx, key) }
+		hinter = prefetch.Hint
 	} else {
 		logger.Info("Using offline mode. All required pre-images must be pre-populated.")
-		preimageOracle = asOracleFn(kv.Get)
-		hinter = func(v preimage.Hint) {
-			logger.Debug("ignoring prefetch hint", "hint", v)
+		getPreimage = kv.Get
+		hinter = func(hint string) error {
+			logger.Debug("ignoring prefetch hint", "hint", hint)
+			return nil
 		}
 	}
-	l1Source := l1.NewSource(logger, preimageOracle, hinter, cfg.L1Head)
 
-	l2Source, err := l2.NewEngine(logger, preimageOracle, hinter, cfg)
-	if err != nil {
-		return fmt.Errorf("connect l2 oracle: %w", err)
-	}
+	// Setup pipe for preimage oracle interaction
+	pClientRW, pHostRW := bidirectionalPipe()
+	oracleServer := preimage.NewOracleServer(pHostRW)
+	// Setup pipe for hint comms
+	hClientRW, hHostRW := bidirectionalPipe()
+	hHost := preimage.NewHintReader(hHostRW)
+	defer pHostRW.Close()
+	defer hHostRW.Close()
+	routeHints(logger, hHost, hinter)
+	launchOracleServer(logger, oracleServer, getPreimage)
 
-	logger.Info("Starting derivation")
-	d := cldr.NewDriver(logger, cfg.Rollup, l1Source, l2Source, cfg.L2ClaimBlockNumber)
-	for {
-		if err = d.Step(ctx); errors.Is(err, io.EOF) {
-			break
-		} else if err != nil {
-			return err
-		}
-	}
-	if !d.ValidateClaim(eth.Bytes32(cfg.L2Claim)) {
-		return ErrClaimNotValid
+	return cl.ClientProgram(
+		logger,
+		cfg.Rollup,
+		cfg.L2ChainConfig,
+		cfg.L1Head,
+		cfg.L2Head,
+		cfg.L2Claim,
+		cfg.L2ClaimBlockNumber,
+		pClientRW,
+		hClientRW,
+	)
+}
+
+type readWritePair struct {
+	io.ReadCloser
+	io.WriteCloser
+}
+
+func (rw *readWritePair) Close() error {
+	if err := rw.ReadCloser.Close(); err != nil {
+		return err
 	}
-	return nil
+	return rw.WriteCloser.Close()
 }
 
-func asOracleFn(getter func(key common.Hash) ([]byte, error)) preimage.OracleFn {
-	return func(key preimage.Key) []byte {
-		pre, err := getter(key.PreimageKey())
-		if err != nil {
-			panic(fmt.Errorf("preimage unavailable for key %v: %w", key, err))
+func bidirectionalPipe() (a, b io.ReadWriteCloser) {
+	ar, bw := io.Pipe()
+	br, aw := io.Pipe()
+	return &readWritePair{ReadCloser: ar, WriteCloser: aw}, &readWritePair{ReadCloser: br, WriteCloser: bw}
+}
+
+func routeHints(logger log.Logger, hintReader *preimage.HintReader, hinter func(hint string) error) {
+	go func() {
+		for {
+			if err := hintReader.NextHint(hinter); err != nil {
+				if err == io.EOF || errors.Is(err, io.ErrClosedPipe) {
+					logger.Info("closing pre-image hint handler")
+					return
+				}
+				logger.Error("pre-image hint router error", "err", err)
+				return
+			}
 		}
-		return pre
-	}
+	}()
 }
 
-func asHinter(hint func(hint string) error) preimage.HinterFn {
-	return func(v preimage.Hint) {
-		err := hint(v.Hint())
-		if err != nil {
-			panic(fmt.Errorf("hint rejected %v: %w", v, err))
+func launchOracleServer(logger log.Logger, server *preimage.OracleServer, getter func(key common.Hash) ([]byte, error)) {
+	go func() {
+		for {
+			if err := server.NextPreimageRequest(getter); err != nil {
+				if err == io.EOF || errors.Is(err, io.ErrClosedPipe) {
+					logger.Info("closing pre-image server")
+					return
+				}
+				logger.Error("pre-image server error", "error", err)
+				return
+			}
 		}
-	}
+	}()
 }

op-program/preimage/hints.go

@@ -9,13 +9,13 @@ import (
 // HintWriter writes hints to an io.Writer (e.g. a special file descriptor, or a debug log),
 // for a pre-image oracle service to prepare specific pre-images.
 type HintWriter struct {
-	w io.Writer
+	rw io.ReadWriter
 }
 
 var _ Hinter = (*HintWriter)(nil)
 
-func NewHintWriter(w io.Writer) *HintWriter {
-	return &HintWriter{w: w}
+func NewHintWriter(rw io.ReadWriter) *HintWriter {
+	return &HintWriter{rw: rw}
 }
 
 func (hw *HintWriter) Hint(v Hint) {
@@ -23,26 +23,29 @@ func (hw *HintWriter) Hint(v Hint) {
 	var hintBytes []byte
 	hintBytes = binary.BigEndian.AppendUint32(hintBytes, uint32(len(hint)))
 	hintBytes = append(hintBytes, []byte(hint)...)
-	hintBytes = append(hintBytes, 0) // to block writing on
-	_, err := hw.w.Write(hintBytes)
+	_, err := hw.rw.Write(hintBytes)
 	if err != nil {
 		panic(fmt.Errorf("failed to write pre-image hint: %w", err))
 	}
+	_, err = hw.rw.Read([]byte{0})
+	if err != nil {
+		panic(fmt.Errorf("failed to read pre-image hint ack: %w", err))
+	}
 }
 
 // HintReader reads the hints of HintWriter and passes them to a router for preparation of the requested pre-images.
 // Onchain the written hints are no-op.
 type HintReader struct {
-	r io.Reader
+	rw io.ReadWriter
 }
 
-func NewHintReader(r io.Reader) *HintReader {
-	return &HintReader{r: r}
+func NewHintReader(rw io.ReadWriter) *HintReader {
+	return &HintReader{rw: rw}
 }
 
 func (hr *HintReader) NextHint(router func(hint string) error) error {
 	var length uint32
-	if err := binary.Read(hr.r, binary.BigEndian, &length); err != nil {
+	if err := binary.Read(hr.rw, binary.BigEndian, &length); err != nil {
 		if err == io.EOF {
 			return io.EOF
 		}
@@ -50,17 +53,17 @@ func (hr *HintReader) NextHint(router func(hint string) error) error {
 	}
 	payload := make([]byte, length)
 	if length > 0 {
-		if _, err := io.ReadFull(hr.r, payload); err != nil {
+		if _, err := io.ReadFull(hr.rw, payload); err != nil {
 			return fmt.Errorf("failed to read hint payload (length %d): %w", length, err)
 		}
 	}
 	if err := router(string(payload)); err != nil {
-		// stream recovery
-		_, _ = hr.r.Read([]byte{0})
+		// write back on error to unblock the HintWriter
+		_, _ = hr.rw.Write([]byte{0})
 		return fmt.Errorf("failed to handle hint: %w", err)
 	}
-	if _, err := hr.r.Read([]byte{0}); err != nil {
-		return fmt.Errorf("failed to read trailing no-op byte to unblock hint writer: %w", err)
+	if _, err := hr.rw.Write([]byte{0}); err != nil {
+		return fmt.Errorf("failed to write trailing no-op byte to unblock hint writer: %w", err)
 	}
 	return nil
 }

op-program/preimage/hints_test.go

@@ -5,7 +5,9 @@ import (
 	"crypto/rand"
 	"errors"
 	"io"
+	"sync"
 	"testing"
+	"time"
 
 	"github.com/stretchr/testify/require"
 )
@@ -20,26 +22,40 @@ func TestHints(t *testing.T) {
 	// Note: pretty much every string is valid communication:
 	// length, payload, 0. Worst case you run out of data, or allocate too much.
 	testHint := func(hints ...string) {
-		var buf bytes.Buffer
-		hw := NewHintWriter(&buf)
-		for _, h := range hints {
-			hw.Hint(rawHint(h))
-		}
-		hr := NewHintReader(&buf)
-		var got []string
-		for i := 0; i < 100; i++ { // sanity limit
-			err := hr.NextHint(func(hint string) error {
-				got = append(got, hint)
-				return nil
-			})
-			if err == io.EOF {
-				break
+		a, b := bidirectionalPipe()
+		var wg sync.WaitGroup
+		wg.Add(2)
+
+		go func() {
+			hw := NewHintWriter(a)
+			for _, h := range hints {
+				hw.Hint(rawHint(h))
 			}
-			require.NoError(t, err)
+			wg.Done()
+		}()
+
+		got := make(chan string, len(hints))
+		go func() {
+			defer wg.Done()
+			hr := NewHintReader(b)
+			for i := 0; i < len(hints); i++ {
+				err := hr.NextHint(func(hint string) error {
+					got <- hint
+					return nil
+				})
+				if err == io.EOF {
+					break
+				}
+				require.NoError(t, err)
+			}
+		}()
+		if waitTimeout(&wg) {
+			t.Error("hint read/write stuck")
 		}
+
 		require.Equal(t, len(hints), len(got), "got all hints")
-		for i, h := range hints {
-			require.Equal(t, h, got[i], "hints match")
+		for _, h := range hints {
+			require.Equal(t, h, <-got, "hints match")
 		}
 	}
 
@@ -73,20 +89,47 @@ func TestHints(t *testing.T) {
 		require.ErrorIs(t, err, io.ErrUnexpectedEOF)
 	})
 	t.Run("cb error", func(t *testing.T) {
-		var buf bytes.Buffer
-		hw := NewHintWriter(&buf)
-		hw.Hint(rawHint("one"))
-		hw.Hint(rawHint("two"))
-		hr := NewHintReader(&buf)
-		cbErr := errors.New("fail")
-		err := hr.NextHint(func(hint string) error { return cbErr })
-		require.ErrorIs(t, err, cbErr)
-		var readHint string
-		err = hr.NextHint(func(hint string) error {
-			readHint = hint
-			return nil
-		})
-		require.NoError(t, err)
-		require.Equal(t, readHint, "two")
+		a, b := bidirectionalPipe()
+		var wg sync.WaitGroup
+		wg.Add(2)
+
+		go func() {
+			hw := NewHintWriter(a)
+			hw.Hint(rawHint("one"))
+			hw.Hint(rawHint("two"))
+			wg.Done()
+		}()
+		go func() {
+			defer wg.Done()
+			hr := NewHintReader(b)
+			cbErr := errors.New("fail")
+			err := hr.NextHint(func(hint string) error { return cbErr })
+			require.ErrorIs(t, err, cbErr)
+			var readHint string
+			err = hr.NextHint(func(hint string) error {
+				readHint = hint
+				return nil
+			})
+			require.NoError(t, err)
+			require.Equal(t, readHint, "two")
+		}()
+		if waitTimeout(&wg) {
+			t.Error("read/write hint stuck")
+		}
 	})
 }
+
+// waitTimeout returns true iff wg.Wait timed out
+func waitTimeout(wg *sync.WaitGroup) bool {
+	done := make(chan struct{})
+	go func() {
+		wg.Wait()
+		close(done)
+	}()
+	select {
+	case <-time.After(time.Second * 30):
+		return true
+	case <-done:
+		return false
+	}
+}