op-program: Implement prefetcher to use hints to populate the preimage oracle

op-program now uses an in-memory key-value store as the backing for all data with the same key-based retrieval and deserialization as in non-fetching mode.

op-program/client/l1/hints.go

@@ -6,12 +6,18 @@ import (
 	"github.com/ethereum-optimism/optimism/op-program/preimage"
 )
 
+const (
+	HintL1BlockHeader  = "l1-block-header"
+	HintL1Transactions = "l1-transactions"
+	HintL1Receipts     = "l1-receipts"
+)
+
 type BlockHeaderHint common.Hash
 
 var _ preimage.Hint = BlockHeaderHint{}
 
 func (l BlockHeaderHint) Hint() string {
-	return "l1-block-header " + (common.Hash)(l).String()
+	return HintL1BlockHeader + " " + (common.Hash)(l).String()
 }
 
 type TransactionsHint common.Hash
@@ -19,7 +25,7 @@ type TransactionsHint common.Hash
 var _ preimage.Hint = TransactionsHint{}
 
 func (l TransactionsHint) Hint() string {
-	return "l1-transactions " + (common.Hash)(l).String()
+	return HintL1Transactions + " " + (common.Hash)(l).String()
 }
 
 type ReceiptsHint common.Hash
@@ -27,5 +33,5 @@ type ReceiptsHint common.Hash
 var _ preimage.Hint = ReceiptsHint{}
 
 func (l ReceiptsHint) Hint() string {
-	return "l1-receipts " + (common.Hash)(l).String()
+	return HintL1Receipts + " " + (common.Hash)(l).String()
 }

op-program/client/l2/hints.go

@@ -6,12 +6,19 @@ import (
 	"github.com/ethereum-optimism/optimism/op-program/preimage"
 )
 
+const (
+	HintL2BlockHeader  = "l2-block-header"
+	HintL2Transactions = "l2-transactions"
+	HintL2Code         = "l2-code"
+	HintL2StateNode    = "l2-state-node"
+)
+
 type BlockHeaderHint common.Hash
 
 var _ preimage.Hint = BlockHeaderHint{}
 
 func (l BlockHeaderHint) Hint() string {
-	return "l2-block-header " + (common.Hash)(l).String()
+	return HintL2BlockHeader + " " + (common.Hash)(l).String()
 }
 
 type TransactionsHint common.Hash
@@ -19,7 +26,7 @@ type TransactionsHint common.Hash
 var _ preimage.Hint = TransactionsHint{}
 
 func (l TransactionsHint) Hint() string {
-	return "l2-transactions " + (common.Hash)(l).String()
+	return HintL2Transactions + " " + (common.Hash)(l).String()
 }
 
 type CodeHint common.Hash
@@ -27,7 +34,7 @@ type CodeHint common.Hash
 var _ preimage.Hint = CodeHint{}
 
 func (l CodeHint) Hint() string {
-	return "l2-code " + (common.Hash)(l).String()
+	return HintL2Code + " " + (common.Hash)(l).String()
 }
 
 type StateNodeHint common.Hash
@@ -35,5 +42,5 @@ type StateNodeHint common.Hash
 var _ preimage.Hint = StateNodeHint{}
 
 func (l StateNodeHint) Hint() string {
-	return "l2-state-node " + (common.Hash)(l).String()
+	return HintL2StateNode + " " + (common.Hash)(l).String()
 }

op-program/host/cmd/main.go

@@ -6,17 +6,18 @@ import (
 	"fmt"
 	"io"
 	"os"
-	"time"
 
 	"github.com/ethereum-optimism/optimism/op-node/chaincfg"
 	"github.com/ethereum-optimism/optimism/op-node/eth"
-	"github.com/ethereum-optimism/optimism/op-node/rollup/derive"
 	cldr "github.com/ethereum-optimism/optimism/op-program/client/driver"
 	"github.com/ethereum-optimism/optimism/op-program/host/config"
 	"github.com/ethereum-optimism/optimism/op-program/host/flags"
+	"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/host/version"
+	"github.com/ethereum-optimism/optimism/op-program/preimage"
 	oplog "github.com/ethereum-optimism/optimism/op-service/log"
 	"github.com/ethereum/go-ethereum/log"
 	"github.com/urfave/cli"
@@ -104,27 +105,35 @@ func FaultProofProgram(logger log.Logger, cfg *config.Config) error {
 	}
 
 	ctx := context.Background()
+	kv := kvstore.NewMemKV()
+
 	logger.Info("Connecting to L1 node", "l1", cfg.L1URL)
-	l1Source, err := l1.NewFetchingL1(ctx, logger, cfg)
+	l1Fetcher, err := l1.NewFetchingOracle(ctx, logger, cfg)
+	if err != nil {
+		return fmt.Errorf("connect l1 fetcher: %w", err)
+	}
+
+	logger.Info("Connecting to L2 node", "l2", cfg.L2URL)
+	l2Fetcher, err := l2.NewFetchingOracle(ctx, logger, cfg)
 	if err != nil {
-		return fmt.Errorf("connect l1 oracle: %w", err)
+		return fmt.Errorf("connect l2 fetcher: %w", err)
 	}
+	prefetch := prefetcher.NewPrefetcher(l1Fetcher, l2Fetcher, kv)
+	preimageOracle := asOracleFn(prefetch)
+	hinter := asHinter(prefetch)
+	l1Source := l1.NewSource(logger, preimageOracle, hinter, cfg.L1Head)
 
 	logger.Info("Connecting to L2 node", "l2", cfg.L2URL)
-	l2Source, err := l2.NewFetchingEngine(ctx, logger, cfg)
+	l2Source, err := l2.NewEngine(logger, preimageOracle, hinter, cfg)
 	if err != nil {
 		return fmt.Errorf("connect l2 oracle: %w", err)
 	}
 
+	logger.Info("Starting derivation")
 	d := cldr.NewDriver(logger, cfg.Rollup, 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
 		}
@@ -135,3 +144,22 @@ func FaultProofProgram(logger log.Logger, cfg *config.Config) error {
 	}
 	return nil
 }
+
+func asOracleFn(prefetcher *prefetcher.Prefetcher) preimage.OracleFn {
+	return func(key preimage.Key) []byte {
+		pre, err := prefetcher.GetPreimage(key.PreimageKey())
+		if err != nil {
+			panic(fmt.Errorf("preimage unavailable for key %v: %w", key, err))
+		}
+		return pre
+	}
+}
+
+func asHinter(prefetcher *prefetcher.Prefetcher) preimage.HinterFn {
+	return func(v preimage.Hint) {
+		err := prefetcher.Hint(v.Hint())
+		if err != nil {
+			panic(fmt.Errorf("hint rejected %v: %w", v, err))
+		}
+	}
+}

op-program/host/l1/l1.go

@@ -8,10 +8,12 @@ import (
 	"github.com/ethereum-optimism/optimism/op-node/sources"
 	cll1 "github.com/ethereum-optimism/optimism/op-program/client/l1"
 	"github.com/ethereum-optimism/optimism/op-program/host/config"
+	"github.com/ethereum-optimism/optimism/op-program/preimage"
+	"github.com/ethereum/go-ethereum/common"
 	"github.com/ethereum/go-ethereum/log"
 )
 
-func NewFetchingL1(ctx context.Context, logger log.Logger, cfg *config.Config) (derive.L1Fetcher, error) {
+func NewFetchingOracle(ctx context.Context, logger log.Logger, cfg *config.Config) (cll1.Oracle, error) {
 	rpc, err := client.NewRPC(ctx, logger, cfg.L1URL)
 	if err != nil {
 		return nil, err
@@ -21,6 +23,10 @@ func NewFetchingL1(ctx context.Context, logger log.Logger, cfg *config.Config) (
 	if err != nil {
 		return nil, err
 	}
-	oracle := cll1.NewCachingOracle(NewFetchingL1Oracle(ctx, logger, source))
-	return cll1.NewOracleL1Client(logger, oracle, cfg.L1Head), err
+	return NewFetchingL1Oracle(ctx, logger, source), nil
+}
+
+func NewSource(logger log.Logger, oracle preimage.Oracle, hint preimage.Hinter, l1Head common.Hash) derive.L1Fetcher {
+	l1Oracle := cll1.NewCachingOracle(cll1.NewPreimageOracle(oracle, hint))
+	return cll1.NewOracleL1Client(logger, l1Oracle, l1Head)
 }

op-program/host/l2/l2.go

@@ -8,22 +8,18 @@ import (
 
 	cll2 "github.com/ethereum-optimism/optimism/op-program/client/l2"
 	"github.com/ethereum-optimism/optimism/op-program/host/config"
+	"github.com/ethereum-optimism/optimism/op-program/preimage"
 	"github.com/ethereum/go-ethereum/core"
 	"github.com/ethereum/go-ethereum/log"
 	"github.com/ethereum/go-ethereum/params"
 )
 
-func NewFetchingEngine(ctx context.Context, logger log.Logger, cfg *config.Config) (*cll2.OracleEngine, error) {
+func NewEngine(logger log.Logger, pre preimage.Oracle, hint preimage.Hinter, cfg *config.Config) (*cll2.OracleEngine, error) {
+	oracle := cll2.NewCachingOracle(cll2.NewPreimageOracle(pre, hint))
 	genesis, err := loadL2Genesis(cfg)
 	if err != nil {
 		return nil, err
 	}
-	fetcher, err := NewFetchingL2Oracle(ctx, logger, cfg.L2URL, cfg.L2Head)
-	if err != nil {
-		return nil, fmt.Errorf("connect l2 oracle: %w", err)
-	}
-	oracle := cll2.NewCachingOracle(fetcher)
-
 	engineBackend, err := cll2.NewOracleBackedL2Chain(logger, oracle, genesis, cfg.L2Head)
 	if err != nil {
 		return nil, fmt.Errorf("create l2 chain: %w", err)
@@ -31,6 +27,14 @@ func NewFetchingEngine(ctx context.Context, logger log.Logger, cfg *config.Confi
 	return cll2.NewOracleEngine(cfg.Rollup, logger, engineBackend), nil
 }
 
+func NewFetchingOracle(ctx context.Context, logger log.Logger, cfg *config.Config) (cll2.Oracle, error) {
+	oracle, err := NewFetchingL2Oracle(ctx, logger, cfg.L2URL, cfg.L2Head)
+	if err != nil {
+		return nil, fmt.Errorf("connect l2 oracle: %w", err)
+	}
+	return oracle, nil
+}
+
 func loadL2Genesis(cfg *config.Config) (*params.ChainConfig, error) {
 	data, err := os.ReadFile(cfg.L2GenesisPath)
 	if err != nil {

op-program/host/prefetcher/prefetcher.go

+package prefetcher
+
+import (
+	"errors"
+	"fmt"
+	"strings"
+
+	"github.com/ethereum-optimism/optimism/op-node/eth"
+	"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/client/mpt"
+	"github.com/ethereum-optimism/optimism/op-program/host/kvstore"
+	"github.com/ethereum-optimism/optimism/op-program/preimage"
+	"github.com/ethereum/go-ethereum/common"
+	"github.com/ethereum/go-ethereum/common/hexutil"
+	"github.com/ethereum/go-ethereum/core/types"
+	"github.com/ethereum/go-ethereum/crypto"
+	"github.com/ethereum/go-ethereum/rlp"
+)
+
+type Prefetcher struct {
+	l1Fetcher l1.Oracle
+	l2Fetcher l2.Oracle
+	lastHint  string
+	kvStore   kvstore.KV
+}
+
+func NewPrefetcher(l1Fetcher l1.Oracle, l2Fetcher l2.Oracle, kvStore kvstore.KV) *Prefetcher {
+	return &Prefetcher{
+		l1Fetcher: l1Fetcher,
+		l2Fetcher: l2Fetcher,
+		kvStore:   kvStore,
+	}
+}
+
+func (p *Prefetcher) Hint(hint string) error {
+	p.lastHint = hint
+	return nil
+}
+
+func (p *Prefetcher) GetPreimage(key common.Hash) ([]byte, error) {
+	pre, err := p.kvStore.Get(key)
+	if errors.Is(err, kvstore.ErrNotFound) && p.lastHint != "" {
+		hint := p.lastHint
+		p.lastHint = ""
+		if err := p.prefetch(hint); err != nil {
+			return nil, fmt.Errorf("prefetch failed: %w", err)
+		}
+		// Should now be available
+		return p.kvStore.Get(key)
+	}
+	return pre, err
+}
+
+func (p *Prefetcher) prefetch(hint string) error {
+	hintType, hash, err := parseHint(hint)
+	if err != nil {
+		return err
+	}
+	switch hintType {
+	case l1.HintL1BlockHeader:
+		header := p.l1Fetcher.HeaderByBlockHash(hash)
+		data, err := header.HeaderRLP()
+		if err != nil {
+			return fmt.Errorf("marshall header: %w", err)
+		}
+
+		return p.kvStore.Put(preimage.Keccak256Key(hash).PreimageKey(), data)
+	case l1.HintL1Transactions:
+		_, txs := p.l1Fetcher.TransactionsByBlockHash(hash)
+		return p.storeTransactions(txs)
+	case l1.HintL1Receipts:
+		_, rcpts := p.l1Fetcher.ReceiptsByBlockHash(hash)
+		opaqueRcpts, err := eth.EncodeReceipts(rcpts)
+		if err != nil {
+			return err
+		}
+		return p.storeTrieNodes(opaqueRcpts)
+	case l2.HintL2BlockHeader:
+		// Pre-fetch both block and transactions
+		block := p.l2Fetcher.BlockByHash(hash)
+		data, err := rlp.EncodeToBytes(block.Header())
+		if err != nil {
+			return fmt.Errorf("marshall header: %w", err)
+		}
+		err = p.kvStore.Put(preimage.Keccak256Key(hash).PreimageKey(), data)
+		if err != nil {
+			return err
+		}
+		return p.storeTransactions(block.Transactions())
+	case l2.HintL2StateNode:
+		node := p.l2Fetcher.NodeByHash(hash)
+		return p.kvStore.Put(preimage.Keccak256Key(hash).PreimageKey(), node)
+	case l2.HintL2Code:
+		code := p.l2Fetcher.CodeByHash(hash)
+		return p.kvStore.Put(preimage.Keccak256Key(hash).PreimageKey(), code)
+	}
+	return fmt.Errorf("unknown hint type: %v", hintType)
+}
+
+func (p *Prefetcher) storeTransactions(txs types.Transactions) error {
+	opaqueTxs, err := eth.EncodeTransactions(txs)
+	if err != nil {
+		return err
+	}
+	return p.storeTrieNodes(opaqueTxs)
+}
+
+func (p *Prefetcher) storeTrieNodes(values []hexutil.Bytes) error {
+	_, nodes := mpt.WriteTrie(values)
+	for _, node := range nodes {
+		err := p.kvStore.Put(preimage.Keccak256Key(crypto.Keccak256Hash(node)).PreimageKey(), node)
+		if err != nil {
+			return fmt.Errorf("failed to store node: %w", err)
+		}
+	}
+	return nil
+}
+
+// parseHint parses a hint string in wire protocol. Returns the hint type, requested hash and error (if any).
+func parseHint(hint string) (string, common.Hash, error) {
+	hintType, hashStr, found := strings.Cut(hint, " ")
+	if !found {
+		return "", common.Hash{}, fmt.Errorf("unsupported hint: %s", hint)
+	}
+	hash := common.HexToHash(hashStr)
+	if hash == (common.Hash{}) {
+		return "", common.Hash{}, fmt.Errorf("invalid hash: %s", hashStr)
+	}
+	return hintType, hash, nil
+}