feat(op-program): 4844 blobs fetcher in host (#9105)

* Add blob preimage type

* rebase

* Start 4844 host changes

* implement hint route

* fix prefetcher test

* Hook up prefetcher / flags / mocks

* Test

* tidy

* @protolambda review

* fix fmt

* flags

* @refcell review

* rebase

go.mod

@@ -6,6 +6,8 @@ require (
 	github.com/BurntSushi/toml v1.3.2
 	github.com/btcsuite/btcd v0.24.0
 	github.com/btcsuite/btcd/chaincfg/chainhash v1.1.0
+	github.com/consensys/gnark-crypto v0.12.1
+	github.com/crate-crypto/go-kzg-4844 v0.7.0
 	github.com/decred/dcrd/dcrec/secp256k1/v4 v4.2.0
 	github.com/ethereum-optimism/go-ethereum-hdwallet v0.1.3
 	github.com/ethereum-optimism/superchain-registry/superchain v0.0.0-20240131175747-1300b1825140
@@ -70,12 +72,10 @@ require (
 	github.com/cockroachdb/redact v1.1.5 // indirect
 	github.com/cockroachdb/tokenbucket v0.0.0-20230807174530-cc333fc44b06 // indirect
 	github.com/consensys/bavard v0.1.13 // indirect
-	github.com/consensys/gnark-crypto v0.12.1 // indirect
 	github.com/containerd/cgroups v1.1.0 // indirect
 	github.com/coreos/go-systemd/v22 v22.5.0 // indirect
 	github.com/cpuguy83/go-md2man/v2 v2.0.2 // indirect
 	github.com/crate-crypto/go-ipa v0.0.0-20230601170251-1830d0757c80 // indirect
-	github.com/crate-crypto/go-kzg-4844 v0.7.0 // indirect
 	github.com/davecgh/go-spew v1.1.1 // indirect
 	github.com/davidlazar/go-crypto v0.0.0-20200604182044-b73af7476f6c // indirect
 	github.com/deckarep/golang-set/v2 v2.1.0 // indirect

op-program/host/config/config.go

@@ -35,12 +35,13 @@ var (
 type Config struct {
 	Rollup *rollup.Config
 	// DataDir is the directory to read/write pre-image data from/to.
-	//If not set, an in-memory key-value store is used and fetching data must be enabled
+	// If not set, an in-memory key-value store is used and fetching data must be enabled
 	DataDir string
 
 	// L1Head is the block has of the L1 chain head block
 	L1Head      common.Hash
 	L1URL       string
+	L1BeaconURL string
 	L1TrustRPC  bool
 	L1RPCKind   sources.RPCProviderKind
 
@@ -104,6 +105,7 @@ func (c *Config) Check() error {
 }
 
 func (c *Config) FetchingEnabled() bool {
+	// TODO: Include Beacon URL once cancun is active on all chains we fault prove.
 	return c.L1URL != "" && c.L2URL != ""
 }
 
@@ -193,6 +195,7 @@ func NewConfigFromCLI(log log.Logger, ctx *cli.Context) (*Config, error) {
 		L2ClaimBlockNumber:  l2ClaimBlockNum,
 		L1Head:              l1Head,
 		L1URL:               ctx.String(flags.L1NodeAddr.Name),
+		L1BeaconURL:         ctx.String(flags.L1BeaconAddr.Name),
 		L1TrustRPC:          ctx.Bool(flags.L1TrustRPC.Name),
 		L1RPCKind:           sources.RPCProviderKind(ctx.String(flags.L1RPCProviderKind.Name)),
 		ExecCmd:             ctx.String(flags.Exec.Name),

op-program/host/flags/flags.go

@@ -75,6 +75,12 @@ var (
 		Usage:   "Address of L1 JSON-RPC endpoint to use (eth namespace required)",
 		EnvVars: prefixEnvVars("L1_RPC"),
 	}
+	L1BeaconAddr = &cli.StringFlag{
+		Name:    "l1.beacon",
+		Usage:   "Address of L1 Beacon API endpoint to use",
+		EnvVars: prefixEnvVars("L1_BEACON_API"),
+		Hidden:  true,
+	}
 	L1TrustRPC = &cli.BoolFlag{
 		Name:    "l1.trustrpc",
 		Usage:   "Trust the L1 RPC, sync faster at risk of malicious/buggy RPC providing bad or inconsistent L1 data",
@@ -112,6 +118,7 @@ var requiredFlags = []cli.Flag{
 	L2Claim,
 	L2BlockNumber,
 }
+
 var programFlags = []cli.Flag{
 	RollupConfig,
 	Network,
@@ -119,6 +126,7 @@ var programFlags = []cli.Flag{
 	L2NodeAddr,
 	L2GenesisPath,
 	L1NodeAddr,
+	L1BeaconAddr,
 	L1TrustRPC,
 	L1RPCProviderKind,
 	Exec,

op-program/host/host.go

@@ -201,12 +201,14 @@ func makePrefetcher(ctx context.Context, logger log.Logger, kv kvstore.KV, cfg *
 	if err != nil {
 		return nil, fmt.Errorf("failed to create L1 client: %w", err)
 	}
+	l1Beacon := client.NewBasicHTTPClient(cfg.L1BeaconURL, logger)
+	l1BlobFetcher := sources.NewL1BeaconClient(l1Beacon, sources.L1BeaconClientConfig{FetchAllSidecars: false})
 	l2Cl, err := NewL2Client(l2RPC, logger, nil, &L2ClientConfig{L2ClientConfig: l2ClCfg, L2Head: cfg.L2Head})
 	if err != nil {
 		return nil, fmt.Errorf("failed to create L2 client: %w", err)
 	}
 	l2DebugCl := &L2Source{L2Client: l2Cl, DebugClient: sources.NewDebugClient(l2RPC.CallContext)}
-	return prefetcher.NewPrefetcher(logger, l1Cl, l2DebugCl, kv), nil
+	return prefetcher.NewPrefetcher(logger, l1Cl, l1BlobFetcher, l2DebugCl, kv), nil
 }
 
 func routeHints(logger log.Logger, hHostRW io.ReadWriter, hinter preimage.HintHandler) chan error {

op-program/host/prefetcher/prefetcher.go

@@ -2,6 +2,7 @@ package prefetcher
 
 import (
 	"context"
+	"encoding/binary"
 	"errors"
 	"fmt"
 	"strings"
@@ -17,6 +18,7 @@ import (
 	"github.com/ethereum/go-ethereum/core/types"
 	"github.com/ethereum/go-ethereum/crypto"
 	"github.com/ethereum/go-ethereum/log"
+	"github.com/ethereum/go-ethereum/params"
 )
 
 type L1Source interface {
@@ -25,6 +27,11 @@ type L1Source interface {
 	FetchReceipts(ctx context.Context, blockHash common.Hash) (eth.BlockInfo, types.Receipts, error)
 }
 
+type L1BlobSource interface {
+	GetBlobSidecars(ctx context.Context, ref eth.L1BlockRef, hashes []eth.IndexedBlobHash) ([]*eth.BlobSidecar, error)
+	GetBlobs(ctx context.Context, ref eth.L1BlockRef, hashes []eth.IndexedBlobHash) ([]*eth.Blob, error)
+}
+
 type L2Source interface {
 	InfoAndTxsByHash(ctx context.Context, blockHash common.Hash) (eth.BlockInfo, types.Transactions, error)
 	NodeByHash(ctx context.Context, hash common.Hash) ([]byte, error)
@@ -35,15 +42,17 @@ type L2Source interface {
 type Prefetcher struct {
 	logger        log.Logger
 	l1Fetcher     L1Source
+	l1BlobFetcher L1BlobSource
 	l2Fetcher     L2Source
 	lastHint      string
 	kvStore       kvstore.KV
 }
 
-func NewPrefetcher(logger log.Logger, l1Fetcher L1Source, l2Fetcher L2Source, kvStore kvstore.KV) *Prefetcher {
+func NewPrefetcher(logger log.Logger, l1Fetcher L1Source, l1BlobFetcher L1BlobSource, l2Fetcher L2Source, kvStore kvstore.KV) *Prefetcher {
 	return &Prefetcher{
 		logger:        logger,
 		l1Fetcher:     NewRetryingL1Source(logger, l1Fetcher),
+		l1BlobFetcher: NewRetryingL1BlobSource(logger, l1BlobFetcher),
 		l2Fetcher:     NewRetryingL2Source(logger, l2Fetcher),
 		kvStore:       kvStore,
 	}
@@ -75,13 +84,17 @@ func (p *Prefetcher) GetPreimage(ctx context.Context, key common.Hash) ([]byte,
 }
 
 func (p *Prefetcher) prefetch(ctx context.Context, hint string) error {
-	hintType, hash, err := parseHint(hint)
+	hintType, hintBytes, err := parseHint(hint)
 	if err != nil {
 		return err
 	}
-	p.logger.Debug("Prefetching", "type", hintType, "hash", hash)
+	p.logger.Debug("Prefetching", "type", hintType, "bytes", hexutil.Bytes(hintBytes))
 	switch hintType {
 	case l1.HintL1BlockHeader:
+		if len(hintBytes) != 32 {
+			return fmt.Errorf("invalid L1 block hint: %x", hint)
+		}
+		hash := common.Hash(hintBytes)
 		header, err := p.l1Fetcher.InfoByHash(ctx, hash)
 		if err != nil {
 			return fmt.Errorf("failed to fetch L1 block %s header: %w", hash, err)
@@ -92,18 +105,68 @@ func (p *Prefetcher) prefetch(ctx context.Context, hint string) error {
 		}
 		return p.kvStore.Put(preimage.Keccak256Key(hash).PreimageKey(), data)
 	case l1.HintL1Transactions:
+		if len(hintBytes) != 32 {
+			return fmt.Errorf("invalid L1 transactions hint: %x", hint)
+		}
+		hash := common.Hash(hintBytes)
 		_, txs, err := p.l1Fetcher.InfoAndTxsByHash(ctx, hash)
 		if err != nil {
 			return fmt.Errorf("failed to fetch L1 block %s txs: %w", hash, err)
 		}
 		return p.storeTransactions(txs)
 	case l1.HintL1Receipts:
+		if len(hintBytes) != 32 {
+			return fmt.Errorf("invalid L1 receipts hint: %x", hint)
+		}
+		hash := common.Hash(hintBytes)
 		_, receipts, err := p.l1Fetcher.FetchReceipts(ctx, hash)
 		if err != nil {
 			return fmt.Errorf("failed to fetch L1 block %s receipts: %w", hash, err)
 		}
 		return p.storeReceipts(receipts)
+	case l1.HintL1Blob:
+		if len(hintBytes) != 48 {
+			return fmt.Errorf("invalid blob hint: %x", hint)
+		}
+
+		blobVersionHash := common.Hash(hintBytes[:32])
+		blobHashIndex := binary.BigEndian.Uint64(hintBytes[32:40])
+		refTimestamp := binary.BigEndian.Uint64(hintBytes[40:48])
+
+		// Fetch the blob sidecar for the indexed blob hash passed in the hint.
+		indexedBlobHash := eth.IndexedBlobHash{
+			Hash:  blobVersionHash,
+			Index: blobHashIndex,
+		}
+		// We pass an `eth.L1BlockRef`, but `GetBlobSidecars` only uses the timestamp, which we received in the hint.
+		sidecars, err := p.l1BlobFetcher.GetBlobSidecars(ctx, eth.L1BlockRef{Time: refTimestamp}, []eth.IndexedBlobHash{indexedBlobHash})
+		if err != nil || len(sidecars) != 1 {
+			return fmt.Errorf("failed to fetch blob sidecars for %s %d: %w", blobVersionHash, blobHashIndex, err)
+		}
+		sidecar := sidecars[0]
+
+		// Put the preimage for the versioned hash into the kv store
+		if err = p.kvStore.Put(preimage.Sha256Key(blobVersionHash).PreimageKey(), sidecar.KZGCommitment[:]); err != nil {
+			return err
+		}
+
+		// Put all of the blob's field elements into the kv store. There should be 4096. The preimage oracle key for
+		// each field element is the keccak256 hash of `abi.encodePacked(sidecar.KZGCommitment, uint256(i))`
+		blobKey := make([]byte, 80)
+		copy(blobKey[:48], sidecar.KZGCommitment[:])
+		for i := 0; i < params.BlobTxFieldElementsPerBlob; i++ {
+			binary.BigEndian.PutUint64(blobKey[72:], uint64(i))
+			blobKeyHash := crypto.Keccak256Hash(blobKey)
+			if err = p.kvStore.Put(preimage.BlobKey(blobKeyHash).PreimageKey(), sidecar.Blob[i<<5:(i+1)<<5]); err != nil {
+				return err
+			}
+		}
+		return nil
 	case l2.HintL2BlockHeader, l2.HintL2Transactions:
+		if len(hintBytes) != 32 {
+			return fmt.Errorf("invalid L2 header/tx hint: %x", hint)
+		}
+		hash := common.Hash(hintBytes)
 		header, txs, err := p.l2Fetcher.InfoAndTxsByHash(ctx, hash)
 		if err != nil {
 			return fmt.Errorf("failed to fetch L2 block %s: %w", hash, err)
@@ -118,18 +181,30 @@ func (p *Prefetcher) prefetch(ctx context.Context, hint string) error {
 		}
 		return p.storeTransactions(txs)
 	case l2.HintL2StateNode:
+		if len(hintBytes) != 32 {
+			return fmt.Errorf("invalid L2 state node hint: %x", hint)
+		}
+		hash := common.Hash(hintBytes)
 		node, err := p.l2Fetcher.NodeByHash(ctx, hash)
 		if err != nil {
 			return fmt.Errorf("failed to fetch L2 state node %s: %w", hash, err)
 		}
 		return p.kvStore.Put(preimage.Keccak256Key(hash).PreimageKey(), node)
 	case l2.HintL2Code:
+		if len(hintBytes) != 32 {
+			return fmt.Errorf("invalid L2 code hint: %x", hint)
+		}
+		hash := common.Hash(hintBytes)
 		code, err := p.l2Fetcher.CodeByHash(ctx, hash)
 		if err != nil {
 			return fmt.Errorf("failed to fetch L2 contract code %s: %w", hash, err)
 		}
 		return p.kvStore.Put(preimage.Keccak256Key(hash).PreimageKey(), code)
 	case l2.HintL2Output:
+		if len(hintBytes) != 32 {
+			return fmt.Errorf("invalid L2 output hint: %x", hint)
+		}
+		hash := common.Hash(hintBytes)
 		output, err := p.l2Fetcher.OutputByRoot(ctx, hash)
 		if err != nil {
 			return fmt.Errorf("failed to fetch L2 output root %s: %w", hash, err)
@@ -167,14 +242,15 @@ func (p *Prefetcher) storeTrieNodes(values []hexutil.Bytes) error {
 }
 
 // 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, " ")
+func parseHint(hint string) (string, []byte, error) {
+	hintType, bytesStr, found := strings.Cut(hint, " ")
 	if !found {
-		return "", common.Hash{}, fmt.Errorf("unsupported hint: %s", hint)
+		return "", nil, fmt.Errorf("unsupported hint: %s", hint)
 	}
-	hash := common.HexToHash(hashStr)
-	if hash == (common.Hash{}) {
-		return "", common.Hash{}, fmt.Errorf("invalid hash: %s", hashStr)
+
+	hintBytes, err := hexutil.Decode(bytesStr)
+	if err != nil {
+		return "", make([]byte, 0), fmt.Errorf("invalid bytes: %s", bytesStr)
 	}
-	return hintType, hash, nil
+	return hintType, hintBytes, nil
 }

op-program/host/prefetcher/prefetcher_test.go

@@ -2,13 +2,18 @@ package prefetcher
 
 import (
 	"context"
+	"crypto/sha256"
+	"encoding/binary"
 	"math/rand"
 	"testing"
 
+	"github.com/consensys/gnark-crypto/ecc/bls12-381/fr"
+	gokzg4844 "github.com/crate-crypto/go-kzg-4844"
 	"github.com/ethereum/go-ethereum/common"
 	"github.com/ethereum/go-ethereum/core/types"
 	"github.com/ethereum/go-ethereum/crypto"
 	"github.com/ethereum/go-ethereum/log"
+	"github.com/ethereum/go-ethereum/params"
 	"github.com/ethereum/go-ethereum/rlp"
 	"github.com/stretchr/testify/require"
 
@@ -24,14 +29,14 @@ import (
 
 func TestNoHint(t *testing.T) {
 	t.Run("NotFound", func(t *testing.T) {
-		prefetcher, _, _, _ := createPrefetcher(t)
+		prefetcher, _, _, _, _ := createPrefetcher(t)
 		res, err := prefetcher.GetPreimage(context.Background(), common.Hash{0xab})
 		require.ErrorIs(t, err, kvstore.ErrNotFound)
 		require.Nil(t, res)
 	})
 
 	t.Run("Exists", func(t *testing.T) {
-		prefetcher, _, _, kv := createPrefetcher(t)
+		prefetcher, _, _, _, kv := createPrefetcher(t)
 		data := []byte{1, 2, 3}
 		hash := crypto.Keccak256Hash(data)
 		require.NoError(t, kv.Put(hash, data))
@@ -51,7 +56,7 @@ func TestFetchL1BlockHeader(t *testing.T) {
 	require.NoError(t, err)
 
 	t.Run("AlreadyKnown", func(t *testing.T) {
-		prefetcher, _, _, kv := createPrefetcher(t)
+		prefetcher, _, _, _, kv := createPrefetcher(t)
 		storeBlock(t, kv, block, rcpts)
 
 		oracle := l1.NewPreimageOracle(asOracleFn(t, prefetcher), asHinter(t, prefetcher))
@@ -60,7 +65,7 @@ func TestFetchL1BlockHeader(t *testing.T) {
 	})
 
 	t.Run("Unknown", func(t *testing.T) {
-		prefetcher, l1Cl, _, _ := createPrefetcher(t)
+		prefetcher, l1Cl, _, _, _ := createPrefetcher(t)
 		l1Cl.ExpectInfoByHash(hash, eth.HeaderBlockInfo(block.Header()), nil)
 		defer l1Cl.AssertExpectations(t)
 
@@ -77,7 +82,7 @@ func TestFetchL1Transactions(t *testing.T) {
 	hash := block.Hash()
 
 	t.Run("AlreadyKnown", func(t *testing.T) {
-		prefetcher, _, _, kv := createPrefetcher(t)
+		prefetcher, _, _, _, kv := createPrefetcher(t)
 
 		storeBlock(t, kv, block, rcpts)
 
@@ -89,7 +94,7 @@ func TestFetchL1Transactions(t *testing.T) {
 	})
 
 	t.Run("Unknown", func(t *testing.T) {
-		prefetcher, l1Cl, _, _ := createPrefetcher(t)
+		prefetcher, l1Cl, _, _, _ := createPrefetcher(t)
 		l1Cl.ExpectInfoByHash(hash, eth.BlockToInfo(block), nil)
 		l1Cl.ExpectInfoAndTxsByHash(hash, eth.BlockToInfo(block), block.Transactions(), nil)
 		defer l1Cl.AssertExpectations(t)
@@ -107,7 +112,7 @@ func TestFetchL1Receipts(t *testing.T) {
 	hash := block.Hash()
 
 	t.Run("AlreadyKnown", func(t *testing.T) {
-		prefetcher, _, _, kv := createPrefetcher(t)
+		prefetcher, _, _, _, kv := createPrefetcher(t)
 		storeBlock(t, kv, block, receipts)
 
 		// Check the data is available (note the oracle does not know about the block, only the kvstore does)
@@ -118,7 +123,7 @@ func TestFetchL1Receipts(t *testing.T) {
 	})
 
 	t.Run("Unknown", func(t *testing.T) {
-		prefetcher, l1Cl, _, _ := createPrefetcher(t)
+		prefetcher, l1Cl, _, _, _ := createPrefetcher(t)
 		l1Cl.ExpectInfoByHash(hash, eth.BlockToInfo(block), nil)
 		l1Cl.ExpectInfoAndTxsByHash(hash, eth.BlockToInfo(block), block.Transactions(), nil)
 		l1Cl.ExpectFetchReceipts(hash, eth.BlockToInfo(block), receipts, nil)
@@ -133,7 +138,7 @@ func TestFetchL1Receipts(t *testing.T) {
 	// Blocks may have identical RLP receipts for different transactions.
 	// Check that the node already existing is handled
 	t.Run("CommonTrieNodes", func(t *testing.T) {
-		prefetcher, l1Cl, _, kv := createPrefetcher(t)
+		prefetcher, l1Cl, _, _, kv := createPrefetcher(t)
 		l1Cl.ExpectInfoByHash(hash, eth.BlockToInfo(block), nil)
 		l1Cl.ExpectInfoAndTxsByHash(hash, eth.BlockToInfo(block), block.Transactions(), nil)
 		l1Cl.ExpectFetchReceipts(hash, eth.BlockToInfo(block), receipts, nil)
@@ -153,13 +158,80 @@ func TestFetchL1Receipts(t *testing.T) {
 	})
 }
 
+// Globally initialize a kzgCtx for blob tests.
+var kzgCtx, _ = gokzg4844.NewContext4096Secure()
+
+// Returns a serialized random field element in big-endian
+func GetRandFieldElement(seed int64) [32]byte {
+	var r fr.Element
+	_, _ = r.SetRandom()
+	return gokzg4844.SerializeScalar(r)
+}
+
+func GetRandBlob(seed int64) gokzg4844.Blob {
+	var blob gokzg4844.Blob
+	bytesPerBlob := gokzg4844.ScalarsPerBlob * gokzg4844.SerializedScalarSize
+	for i := 0; i < bytesPerBlob; i += gokzg4844.SerializedScalarSize {
+		fieldElementBytes := GetRandFieldElement(seed + int64(i))
+		copy(blob[i:i+gokzg4844.SerializedScalarSize], fieldElementBytes[:])
+	}
+	return blob
+}
+
+func TestFetchL1Blob(t *testing.T) {
+	blob := GetRandBlob(0xf00f00)
+	commitment, err := kzgCtx.BlobToKZGCommitment(blob, 0)
+	require.NoError(t, err)
+	versionedHash := sha256.Sum256(commitment[:])
+	versionedHash[0] = params.BlobTxHashVersion
+	blobHash := eth.IndexedBlobHash{Hash: versionedHash, Index: 0xFACADE}
+	l1Ref := eth.L1BlockRef{Time: 0}
+
+	t.Run("AlreadyKnown", func(t *testing.T) {
+		prefetcher, _, blobFetcher, _, kv := createPrefetcher(t)
+		storeBlob(t, kv, (eth.Bytes48)(commitment), (*eth.Blob)(&blob))
+
+		oracle := l1.NewPreimageOracle(asOracleFn(t, prefetcher), asHinter(t, prefetcher))
+		blobFetcher.ExpectOnGetBlobSidecars(
+			context.Background(),
+			l1Ref,
+			[]eth.IndexedBlobHash{blobHash},
+			(eth.Bytes48)(commitment),
+			[]*eth.Blob{(*eth.Blob)(&blob)},
+			nil,
+		)
+		defer blobFetcher.AssertExpectations(t)
+
+		blobs := oracle.GetBlob(l1Ref, blobHash)
+		require.EqualValues(t, blobs[:], blob[:])
+	})
+
+	t.Run("Unknown", func(t *testing.T) {
+		prefetcher, _, blobFetcher, _, _ := createPrefetcher(t)
+
+		oracle := l1.NewPreimageOracle(asOracleFn(t, prefetcher), asHinter(t, prefetcher))
+		blobFetcher.ExpectOnGetBlobSidecars(
+			context.Background(),
+			l1Ref,
+			[]eth.IndexedBlobHash{blobHash},
+			(eth.Bytes48)(commitment),
+			[]*eth.Blob{(*eth.Blob)(&blob)},
+			nil,
+		)
+		defer blobFetcher.AssertExpectations(t)
+
+		blobs := oracle.GetBlob(l1Ref, blobHash)
+		require.EqualValues(t, blobs[:], blob[:])
+	})
+}
+
 func TestFetchL2Block(t *testing.T) {
 	rng := rand.New(rand.NewSource(123))
 	block, rcpts := testutils.RandomBlock(rng, 10)
 	hash := block.Hash()
 
 	t.Run("AlreadyKnown", func(t *testing.T) {
-		prefetcher, _, _, kv := createPrefetcher(t)
+		prefetcher, _, _, _, kv := createPrefetcher(t)
 		storeBlock(t, kv, block, rcpts)
 
 		oracle := l2.NewPreimageOracle(asOracleFn(t, prefetcher), asHinter(t, prefetcher))
@@ -169,7 +241,7 @@ func TestFetchL2Block(t *testing.T) {
 	})
 
 	t.Run("Unknown", func(t *testing.T) {
-		prefetcher, _, l2Cl, _ := createPrefetcher(t)
+		prefetcher, _, _, l2Cl, _ := createPrefetcher(t)
 		l2Cl.ExpectInfoAndTxsByHash(hash, eth.BlockToInfo(block), block.Transactions(), nil)
 		defer l2Cl.MockL2Client.AssertExpectations(t)
 
@@ -186,7 +258,7 @@ func TestFetchL2Transactions(t *testing.T) {
 	hash := block.Hash()
 
 	t.Run("AlreadyKnown", func(t *testing.T) {
-		prefetcher, _, _, kv := createPrefetcher(t)
+		prefetcher, _, _, _, kv := createPrefetcher(t)
 		storeBlock(t, kv, block, rcpts)
 
 		oracle := l2.NewPreimageOracle(asOracleFn(t, prefetcher), asHinter(t, prefetcher))
@@ -195,7 +267,7 @@ func TestFetchL2Transactions(t *testing.T) {
 	})
 
 	t.Run("Unknown", func(t *testing.T) {
-		prefetcher, _, l2Cl, _ := createPrefetcher(t)
+		prefetcher, _, _, l2Cl, _ := createPrefetcher(t)
 		l2Cl.ExpectInfoAndTxsByHash(hash, eth.BlockToInfo(block), block.Transactions(), nil)
 		defer l2Cl.MockL2Client.AssertExpectations(t)
 
@@ -212,7 +284,7 @@ func TestFetchL2Node(t *testing.T) {
 	key := preimage.Keccak256Key(hash).PreimageKey()
 
 	t.Run("AlreadyKnown", func(t *testing.T) {
-		prefetcher, _, _, kv := createPrefetcher(t)
+		prefetcher, _, _, _, kv := createPrefetcher(t)
 		require.NoError(t, kv.Put(key, node))
 
 		oracle := l2.NewPreimageOracle(asOracleFn(t, prefetcher), asHinter(t, prefetcher))
@@ -221,7 +293,7 @@ func TestFetchL2Node(t *testing.T) {
 	})
 
 	t.Run("Unknown", func(t *testing.T) {
-		prefetcher, _, l2Cl, _ := createPrefetcher(t)
+		prefetcher, _, _, l2Cl, _ := createPrefetcher(t)
 		l2Cl.ExpectNodeByHash(hash, node, nil)
 		defer l2Cl.MockDebugClient.AssertExpectations(t)
 
@@ -238,7 +310,7 @@ func TestFetchL2Code(t *testing.T) {
 	key := preimage.Keccak256Key(hash).PreimageKey()
 
 	t.Run("AlreadyKnown", func(t *testing.T) {
-		prefetcher, _, _, kv := createPrefetcher(t)
+		prefetcher, _, _, _, kv := createPrefetcher(t)
 		require.NoError(t, kv.Put(key, code))
 
 		oracle := l2.NewPreimageOracle(asOracleFn(t, prefetcher), asHinter(t, prefetcher))
@@ -247,7 +319,7 @@ func TestFetchL2Code(t *testing.T) {
 	})
 
 	t.Run("Unknown", func(t *testing.T) {
-		prefetcher, _, l2Cl, _ := createPrefetcher(t)
+		prefetcher, _, _, l2Cl, _ := createPrefetcher(t)
 		l2Cl.ExpectCodeByHash(hash, code, nil)
 		defer l2Cl.MockDebugClient.AssertExpectations(t)
 
@@ -258,7 +330,7 @@ func TestFetchL2Code(t *testing.T) {
 }
 
 func TestBadHints(t *testing.T) {
-	prefetcher, _, _, kv := createPrefetcher(t)
+	prefetcher, _, _, _, kv := createPrefetcher(t)
 	hash := common.Hash{0xad}
 
 	t.Run("NoSpace", func(t *testing.T) {
@@ -277,7 +349,7 @@ func TestBadHints(t *testing.T) {
 
 		// But it will fail to prefetch when the pre-image isn't available
 		pre, err := prefetcher.GetPreimage(context.Background(), hash)
-		require.ErrorContains(t, err, "invalid hash")
+		require.ErrorContains(t, err, "invalid bytes")
 		require.Nil(t, pre)
 	})
 
@@ -311,10 +383,10 @@ func TestRetryWhenNotAvailableAfterPrefetching(t *testing.T) {
 	node := testutils.RandomData(rng, 30)
 	hash := crypto.Keccak256Hash(node)
 
-	_, l1Source, l2Cl, kv := createPrefetcher(t)
+	_, l1Source, l1BlobSource, l2Cl, kv := createPrefetcher(t)
 	putsToIgnore := 2
 	kv = &unreliableKvStore{KV: kv, putsToIgnore: putsToIgnore}
-	prefetcher := NewPrefetcher(testlog.Logger(t, log.LvlInfo), l1Source, l2Cl, kv)
+	prefetcher := NewPrefetcher(testlog.Logger(t, log.LvlInfo), l1Source, l1BlobSource, l2Cl, kv)
 
 	// Expect one call for each ignored put, plus one more request for when the put succeeds
 	for i := 0; i < putsToIgnore+1; i++ {
@@ -355,18 +427,19 @@ func (m *l2Client) ExpectOutputByRoot(root common.Hash, output eth.Output, err e
 	m.Mock.On("OutputByRoot", root).Once().Return(output, &err)
 }
 
-func createPrefetcher(t *testing.T) (*Prefetcher, *testutils.MockL1Source, *l2Client, kvstore.KV) {
+func createPrefetcher(t *testing.T) (*Prefetcher, *testutils.MockL1Source, *testutils.MockBlobsFetcher, *l2Client, kvstore.KV) {
 	logger := testlog.Logger(t, log.LvlDebug)
 	kv := kvstore.NewMemKV()
 
 	l1Source := new(testutils.MockL1Source)
+	l1BlobSource := new(testutils.MockBlobsFetcher)
 	l2Source := &l2Client{
 		MockL2Client:    new(testutils.MockL2Client),
 		MockDebugClient: new(testutils.MockDebugClient),
 	}
 
-	prefetcher := NewPrefetcher(logger, l1Source, l2Source, kv)
-	return prefetcher, l1Source, l2Source, kv
+	prefetcher := NewPrefetcher(logger, l1Source, l1BlobSource, l2Source, kv)
+	return prefetcher, l1Source, l1BlobSource, l2Source, kv
 }
 
 func storeBlock(t *testing.T, kv kvstore.KV, block *types.Block, receipts types.Receipts) {
@@ -392,6 +465,23 @@ func storeBlock(t *testing.T, kv kvstore.KV, block *types.Block, receipts types.
 	require.NoError(t, kv.Put(preimage.Keccak256Key(block.Hash()).PreimageKey(), headerRlp))
 }
 
+func storeBlob(t *testing.T, kv kvstore.KV, commitment eth.Bytes48, blob *eth.Blob) {
+	// Pre-store versioned hash preimage (commitment)
+	err := kv.Put(preimage.Sha256Key(sha256.Sum256(commitment[:])).PreimageKey(), commitment[:])
+	require.NoError(t, err, "Failed to store versioned hash preimage in kvstore")
+
+	// Pre-store blob field elements
+	blobKeyBuf := make([]byte, 80)
+	copy(blobKeyBuf[:48], commitment[:])
+	for i := 0; i < params.BlobTxFieldElementsPerBlob; i++ {
+		binary.BigEndian.PutUint64(blobKeyBuf[:72], uint64(i))
+		feKey := crypto.Keccak256Hash(blobKeyBuf)
+
+		err = kv.Put(preimage.BlobKey(feKey).PreimageKey(), blob[i<<5:(i+1)<<5])
+		require.NoError(t, err, "Failed to store field element preimage in kvstore")
+	}
+}
+
 func asOracleFn(t *testing.T, prefetcher *Prefetcher) preimage.OracleFn {
 	return func(key preimage.Key) []byte {
 		pre, err := prefetcher.GetPreimage(context.Background(), key.PreimageKey())

op-program/host/prefetcher/retry.go

@@ -59,6 +59,42 @@ func (s *RetryingL1Source) FetchReceipts(ctx context.Context, blockHash common.H
 
 var _ L1Source = (*RetryingL1Source)(nil)
 
+type RetryingL1BlobSource struct {
+	logger   log.Logger
+	source   L1BlobSource
+	strategy retry.Strategy
+}
+
+func NewRetryingL1BlobSource(logger log.Logger, source L1BlobSource) *RetryingL1BlobSource {
+	return &RetryingL1BlobSource{
+		logger:   logger,
+		source:   source,
+		strategy: retry.Exponential(),
+	}
+}
+
+func (s *RetryingL1BlobSource) GetBlobSidecars(ctx context.Context, ref eth.L1BlockRef, hashes []eth.IndexedBlobHash) ([]*eth.BlobSidecar, error) {
+	return retry.Do(ctx, maxAttempts, s.strategy, func() ([]*eth.BlobSidecar, error) {
+		sidecars, err := s.source.GetBlobSidecars(ctx, ref, hashes)
+		if err != nil {
+			s.logger.Warn("Failed to retrieve blob sidecars", "ref", ref, "err", err)
+		}
+		return sidecars, err
+	})
+}
+
+func (s *RetryingL1BlobSource) GetBlobs(ctx context.Context, ref eth.L1BlockRef, hashes []eth.IndexedBlobHash) ([]*eth.Blob, error) {
+	return retry.Do(ctx, maxAttempts, s.strategy, func() ([]*eth.Blob, error) {
+		blobs, err := s.source.GetBlobs(ctx, ref, hashes)
+		if err != nil {
+			s.logger.Warn("Failed to retrieve blobs", "ref", ref, "err", err)
+		}
+		return blobs, err
+	})
+}
+
+var _ L1BlobSource = (*RetryingL1BlobSource)(nil)
+
 type RetryingL2Source struct {
 	logger   log.Logger
 	source   L2Source

op-program/host/prefetcher/retry_test.go

@@ -2,6 +2,7 @@ package prefetcher
 
 import (
 	"context"
+	"crypto/sha256"
 	"errors"
 	"testing"
 
@@ -11,6 +12,7 @@ import (
 	"github.com/ethereum/go-ethereum/common"
 	"github.com/ethereum/go-ethereum/core/types"
 	"github.com/ethereum/go-ethereum/log"
+	"github.com/ethereum/go-ethereum/params"
 
 	"github.com/ethereum-optimism/optimism/op-service/eth"
 	"github.com/ethereum-optimism/optimism/op-service/retry"
@@ -111,6 +113,117 @@ func createL1Source(t *testing.T) (*RetryingL1Source, *testutils.MockL1Source) {
 	return source, mock
 }
 
+func TestRetryingL1BlobSource(t *testing.T) {
+	ctx := context.Background()
+	blob := GetRandBlob(0xFACADE)
+	commitment, err := kzgCtx.BlobToKZGCommitment(blob, 0)
+	require.NoError(t, err)
+	versionedHash := sha256.Sum256(commitment[:])
+	versionedHash[0] = params.BlobTxHashVersion
+	blobHash := eth.IndexedBlobHash{Hash: versionedHash, Index: 0xFACADE}
+	l1BlockRef := eth.L1BlockRef{Time: 0}
+
+	t.Run("GetBlobs Success", func(t *testing.T) {
+		source, mock := createL1BlobSource(t)
+		defer mock.AssertExpectations(t)
+		mock.ExpectOnGetBlobs(
+			ctx,
+			l1BlockRef,
+			[]eth.IndexedBlobHash{blobHash},
+			[]*eth.Blob{(*eth.Blob)(&blob)},
+			nil,
+		)
+
+		result, err := source.GetBlobs(ctx, l1BlockRef, []eth.IndexedBlobHash{blobHash})
+		require.NoError(t, err)
+		require.Equal(t, len(result), 1)
+		require.Equal(t, blob[:], result[0][:])
+	})
+
+	t.Run("GetBlobs Error", func(t *testing.T) {
+		source, mock := createL1BlobSource(t)
+		defer mock.AssertExpectations(t)
+		expectedErr := errors.New("boom")
+		mock.ExpectOnGetBlobs(
+			ctx,
+			l1BlockRef,
+			[]eth.IndexedBlobHash{blobHash},
+			nil,
+			expectedErr,
+		)
+		mock.ExpectOnGetBlobs(
+			ctx,
+			l1BlockRef,
+			[]eth.IndexedBlobHash{blobHash},
+			[]*eth.Blob{(*eth.Blob)(&blob)},
+			nil,
+		)
+
+		result, err := source.GetBlobs(ctx, l1BlockRef, []eth.IndexedBlobHash{blobHash})
+		require.NoError(t, err)
+		require.Equal(t, len(result), 1)
+		require.Equal(t, blob[:], result[0][:])
+	})
+
+	t.Run("GetBlobSidecars Success", func(t *testing.T) {
+		source, mock := createL1BlobSource(t)
+		defer mock.AssertExpectations(t)
+		mock.ExpectOnGetBlobSidecars(
+			ctx,
+			l1BlockRef,
+			[]eth.IndexedBlobHash{blobHash},
+			(eth.Bytes48)(commitment),
+			[]*eth.Blob{(*eth.Blob)(&blob)},
+			nil,
+		)
+
+		result, err := source.GetBlobSidecars(ctx, l1BlockRef, []eth.IndexedBlobHash{blobHash})
+		require.NoError(t, err)
+		require.Equal(t, len(result), 1)
+		require.Equal(t, blob[:], result[0].Blob[:])
+		require.Equal(t, blobHash.Index, uint64(result[0].Index))
+		require.Equal(t, (eth.Bytes48)(commitment), result[0].KZGCommitment)
+	})
+
+	t.Run("GetBlobSidecars Error", func(t *testing.T) {
+		source, mock := createL1BlobSource(t)
+		defer mock.AssertExpectations(t)
+		expectedErr := errors.New("boom")
+		mock.ExpectOnGetBlobSidecars(
+			ctx,
+			l1BlockRef,
+			[]eth.IndexedBlobHash{blobHash},
+			(eth.Bytes48)(commitment),
+			[]*eth.Blob{(*eth.Blob)(&blob)},
+			expectedErr,
+		)
+		mock.ExpectOnGetBlobSidecars(
+			ctx,
+			l1BlockRef,
+			[]eth.IndexedBlobHash{blobHash},
+			(eth.Bytes48)(commitment),
+			[]*eth.Blob{(*eth.Blob)(&blob)},
+			nil,
+		)
+
+		result, err := source.GetBlobSidecars(ctx, l1BlockRef, []eth.IndexedBlobHash{blobHash})
+		require.NoError(t, err)
+		require.Equal(t, len(result), 1)
+		require.Equal(t, blob[:], result[0].Blob[:])
+		require.Equal(t, blobHash.Index, uint64(result[0].Index))
+		require.Equal(t, (eth.Bytes48)(commitment), result[0].KZGCommitment)
+	})
+}
+
+func createL1BlobSource(t *testing.T) (*RetryingL1BlobSource, *testutils.MockBlobsFetcher) {
+	logger := testlog.Logger(t, log.LvlDebug)
+	mock := &testutils.MockBlobsFetcher{}
+	source := NewRetryingL1BlobSource(logger, mock)
+	// Avoid sleeping in tests by using a fixed retry strategy with no delay
+	source.strategy = retry.Fixed(0)
+	return source, mock
+}
+
 func TestRetryingL2Source(t *testing.T) {
 	ctx := context.Background()
 	hash := common.Hash{0xab}

op-service/testutils/mock_blobs_fetcher.go

+package testutils
+
+import (
+	"context"
+
+	"github.com/ethereum-optimism/optimism/op-service/eth"
+	"github.com/stretchr/testify/mock"
+)
+
+type MockBlobsFetcher struct {
+	mock.Mock
+}
+
+func (cl *MockBlobsFetcher) GetBlobs(ctx context.Context, ref eth.L1BlockRef, hashes []eth.IndexedBlobHash) ([]*eth.Blob, error) {
+	out := cl.Mock.MethodCalled("GetBlobs", ref, hashes)
+	return out.Get(0).([]*eth.Blob), out.Error(1)
+}
+
+func (cl *MockBlobsFetcher) ExpectOnGetBlobs(ctx context.Context, ref eth.L1BlockRef, hashes []eth.IndexedBlobHash, blobs []*eth.Blob, err error) {
+	cl.Mock.On("GetBlobs", ref, hashes).Once().Return(blobs, err)
+}
+
+func (cl *MockBlobsFetcher) GetBlobSidecars(ctx context.Context, ref eth.L1BlockRef, hashes []eth.IndexedBlobHash) ([]*eth.BlobSidecar, error) {
+	out := cl.Mock.MethodCalled("GetBlobSidecars", ref, hashes)
+	return out.Get(0).([]*eth.BlobSidecar), out.Error(1)
+}
+
+func (cl *MockBlobsFetcher) ExpectOnGetBlobSidecars(ctx context.Context, ref eth.L1BlockRef, hashes []eth.IndexedBlobHash, commitment eth.Bytes48, blobs []*eth.Blob, err error) {
+	cl.Mock.On("GetBlobSidecars", ref, hashes).Once().Return([]*eth.BlobSidecar{{
+		Blob:          *blobs[0],
+		Index:         eth.Uint64String(hashes[0].Index),
+		KZGCommitment: commitment,
+	}}, err)
+}