integrate es with op

op-node/flags/flags.go

@@ -43,6 +43,12 @@ var (
 		Destination: new(string),
 	}
 	/* Optional Flags */
+	ESNodeAddr = &cli.StringFlag{
+		Name:     "esnode",
+		Usage:    "Address of es-node HTTP endpoint to use.",
+		Required: false,
+		EnvVars:  prefixEnvVars("ES_NODE"),
+	}
 	BeaconAddr = &cli.StringFlag{
 		Name:     "l1.beacon",
 		Usage:    "Address of L1 Beacon-node HTTP endpoint to use.",

op-node/node/client.go

@@ -10,6 +10,7 @@ import (
 	"github.com/ethereum-optimism/optimism/op-service/client"
 	"github.com/ethereum-optimism/optimism/op-service/sources"
 
+	"github.com/ethereum/go-ethereum/common"
 	"github.com/ethereum/go-ethereum/log"
 	gn "github.com/ethereum/go-ethereum/node"
 	"github.com/ethereum/go-ethereum/rpc"
@@ -37,6 +38,10 @@ type L1BeaconEndpointSetup interface {
 	Check() error
 }
 
+type ESNodeEndpointSetup interface {
+	Setup(ctx context.Context, batcherAddr common.Address, log log.Logger) (cl *sources.ESClient, err error)
+}
+
 type L2EndpointConfig struct {
 	// L2EngineAddr is the address of the L2 Engine JSON-RPC endpoint to use. The engine and eth
 	// namespaces must be enabled by the endpoint.
@@ -201,3 +206,13 @@ func (cfg *L1BeaconEndpointConfig) ShouldIgnoreBeaconCheck() bool {
 func (cfg *L1BeaconEndpointConfig) ShouldFetchAllSidecars() bool {
 	return cfg.BeaconFetchAllSidecars
 }
+
+var _ ESNodeEndpointSetup = (*ESNodeEndpointConfig)(nil)
+
+type ESNodeEndpointConfig struct {
+	ESNodeAddr string
+}
+
+func (cfg *ESNodeEndpointConfig) Setup(ctx context.Context, batchInboxAddr common.Address, log log.Logger) (cl *sources.ESClient, err error) {
+	return sources.NewESClient(cfg.ESNodeAddr, batchInboxAddr, log)
+}

op-node/node/config.go

@@ -21,6 +21,7 @@ type Config struct {
 	L2 L2EndpointSetup
 
 	Beacon L1BeaconEndpointSetup
+	ESNode ESNodeEndpointSetup
 
 	Driver driver.Config
 

op-node/node/node.go

@@ -306,8 +306,17 @@ func (n *OpNode) initL1BeaconAPI(ctx context.Context, cfg *Config) error {
 	if err != nil {
 		return fmt.Errorf("failed to setup L1 Beacon API client: %w", err)
 	}
+	var esClient *sources.ESClient
+	if cfg.ESNode != nil {
+		esClient, err = cfg.ESNode.Setup(ctx, cfg.Rollup.BatchInboxAddress, n.log)
+		if err != nil {
+			return fmt.Errorf("failed to setup ES client: %w", err)
+		}
+	}
+
 	beaconCfg := sources.L1BeaconClientConfig{
 		FetchAllSidecars: cfg.Beacon.ShouldFetchAllSidecars(),
+		ESClient:         esClient,
 	}
 	n.beacon = sources.NewL1BeaconClient(httpClient, beaconCfg)
 

op-node/service.go

@@ -79,6 +79,7 @@ func NewConfig(ctx *cli.Context, log log.Logger) (*node.Config, error) {
 		Rollup: *rollupConfig,
 		Driver: *driverConfig,
 		Beacon: NewBeaconEndpointConfig(ctx),
+		ESNode: NewConditionalESNodeConfig(ctx),
 		RPC: node.RPCConfig{
 			ListenAddr:  ctx.String(flags.RPCListenAddr.Name),
 			ListenPort:  ctx.Int(flags.RPCListenPort.Name),
@@ -132,6 +133,16 @@ func NewBeaconEndpointConfig(ctx *cli.Context) node.L1BeaconEndpointSetup {
 	}
 }
 
+func NewConditionalESNodeConfig(ctx *cli.Context) node.ESNodeEndpointSetup {
+	if ctx.String(flags.ESNodeAddr.Name) == "" {
+		return nil
+	}
+
+	return &node.ESNodeEndpointConfig{
+		ESNodeAddr: ctx.String(flags.ESNodeAddr.Name),
+	}
+}
+
 func NewL1EndpointConfig(ctx *cli.Context) *node.L1EndpointConfig {
 	return &node.L1EndpointConfig{
 		L1NodeAddr:       ctx.String(flags.L1NodeAddr.Name),

op-service/sources/es_client.go

@@ -0,0 +1,91 @@
+package sources
+
+import (
+	"context"
+	"fmt"
+	"math/big"
+	"strings"
+
+	"github.com/ethereum-optimism/optimism/op-service/eth"
+	"github.com/ethereum/go-ethereum/accounts/abi"
+	"github.com/ethereum/go-ethereum/common"
+	"github.com/ethereum/go-ethereum/crypto/kzg4844"
+	"github.com/ethereum/go-ethereum/ethclient"
+	"github.com/ethereum/go-ethereum/log"
+	"github.com/zhiqiangxu/multicall"
+)
+
+type ESClient struct {
+	logger         log.Logger
+	client         *ethclient.Client
+	abi            abi.ABI
+	batchInboxAddr common.Address
+}
+
+func NewESClient(nodeAddr string, batchInboxAddr common.Address, logger log.Logger) (*ESClient, error) {
+	client, err := ethclient.Dial(nodeAddr)
+	if err != nil {
+		return nil, err
+	}
+
+	abi, err := abi.JSON(strings.NewReader(abiDecentralizedKV))
+	if err != nil {
+		return nil, err
+	}
+	return &ESClient{logger: logger, client: client, abi: abi, batchInboxAddr: batchInboxAddr}, nil
+}
+
+func (s *ESClient) GetBlobs(blobHashes []eth.IndexedBlobHash) ([]*eth.BlobSidecar, error) {
+
+	invokes := make([]multicall.Invoke, 0, len(blobHashes))
+	for _, blobHash := range blobHashes {
+		invokes = append(invokes, multicall.Invoke{
+			Contract: common.HexToAddress("0x804C520d3c084C805E37A35E90057Ac32831F96f"), // TODO make it configurable
+			Name:     "get",
+			Args:     []interface{}{blobHash.Hash, uint8(0), big.NewInt(0), big.NewInt(blobSize)},
+		})
+	}
+
+	output := make([][]byte, len(invokes))
+	err := multicall.DoFrom(context.Background(), s.client, &s.abi, invokes, output, s.batchInboxAddr)
+	if err != nil {
+		return nil, err
+	}
+
+	blobSideCars := make([]*eth.BlobSidecar, 0, len(output))
+	for i, blobBytes := range output {
+		if len(blobBytes) != blobSize {
+			return nil, fmt.Errorf("invalid blob size:%d, index:%d", len(blobBytes), blobHashes[i].Index)
+		}
+		var blob eth.Blob
+		copy(blob[:], blobBytes)
+		blobSideCar, err := makeBlobSideCar(blob, blobHashes[i].Index)
+		if err != nil {
+			return nil, err
+		}
+		blobSideCars = append(blobSideCars, blobSideCar)
+	}
+
+	return blobSideCars, nil
+}
+
+func makeBlobSideCar(blob eth.Blob, blobIndex uint64) (*eth.BlobSidecar, error) {
+
+	rawBlob := *blob.KZGBlob()
+	commitment, err := kzg4844.BlobToCommitment(rawBlob)
+	if err != nil {
+		return nil, fmt.Errorf("cannot compute KZG commitment of blob %d in tx candidate: %w", blobIndex, err)
+	}
+
+	proof, err := kzg4844.ComputeBlobProof(rawBlob, commitment)
+	if err != nil {
+		return nil, fmt.Errorf("cannot compute KZG proof for fast commitment verification of blob %d in tx candidate: %w", blobIndex, err)
+	}
+
+	return &eth.BlobSidecar{Blob: blob, Index: eth.Uint64String(blobIndex), KZGCommitment: eth.Bytes48(commitment), KZGProof: eth.Bytes48(proof)}, nil
+}
+
+const (
+	abiDecentralizedKV = "[{\"inputs\":[],\"name\":\"InvalidInitialization\",\"type\":\"error\"},{\"inputs\":[],\"name\":\"NotInitializing\",\"type\":\"error\"},{\"inputs\":[{\"internalType\":\"address\",\"name\":\"owner\",\"type\":\"address\"}],\"name\":\"OwnableInvalidOwner\",\"type\":\"error\"},{\"inputs\":[{\"internalType\":\"address\",\"name\":\"account\",\"type\":\"address\"}],\"name\":\"OwnableUnauthorizedAccount\",\"type\":\"error\"},{\"anonymous\":false,\"inputs\":[{\"indexed\":false,\"internalType\":\"uint64\",\"name\":\"version\",\"type\":\"uint64\"}],\"name\":\"Initialized\",\"type\":\"event\"},{\"anonymous\":false,\"inputs\":[{\"indexed\":true,\"internalType\":\"address\",\"name\":\"previousOwner\",\"type\":\"address\"},{\"indexed\":true,\"internalType\":\"address\",\"name\":\"newOwner\",\"type\":\"address\"}],\"name\":\"OwnershipTransferred\",\"type\":\"event\"},{\"anonymous\":false,\"inputs\":[{\"indexed\":true,\"internalType\":\"uint256\",\"name\":\"kvIdx\",\"type\":\"uint256\"},{\"indexed\":true,\"internalType\":\"uint256\",\"name\":\"lastKvIdx\",\"type\":\"uint256\"}],\"name\":\"Remove\",\"type\":\"event\"},{\"inputs\":[{\"internalType\":\"uint256\",\"name\":\"_maxKvSize\",\"type\":\"uint256\"},{\"internalType\":\"uint256\",\"name\":\"_startTime\",\"type\":\"uint256\"},{\"internalType\":\"uint256\",\"name\":\"_storageCost\",\"type\":\"uint256\"},{\"internalType\":\"uint256\",\"name\":\"_dcfFactor\",\"type\":\"uint256\"},{\"internalType\":\"address\",\"name\":\"_owner\",\"type\":\"address\"}],\"name\":\"__init_KV\",\"outputs\":[],\"stateMutability\":\"nonpayable\",\"type\":\"function\"},{\"inputs\":[],\"name\":\"dcfFactor\",\"outputs\":[{\"internalType\":\"uint256\",\"name\":\"\",\"type\":\"uint256\"}],\"stateMutability\":\"view\",\"type\":\"function\"},{\"inputs\":[{\"internalType\":\"bytes32\",\"name\":\"key\",\"type\":\"bytes32\"}],\"name\":\"exist\",\"outputs\":[{\"internalType\":\"bool\",\"name\":\"\",\"type\":\"bool\"}],\"stateMutability\":\"view\",\"type\":\"function\"},{\"inputs\":[{\"internalType\":\"bytes32\",\"name\":\"key\",\"type\":\"bytes32\"},{\"internalType\":\"enumDecentralizedKV.DecodeType\",\"name\":\"decodeType\",\"type\":\"uint8\"},{\"internalType\":\"uint256\",\"name\":\"off\",\"type\":\"uint256\"},{\"internalType\":\"uint256\",\"name\":\"len\",\"type\":\"uint256\"}],\"name\":\"get\",\"outputs\":[{\"internalType\":\"bytes\",\"name\":\"\",\"type\":\"bytes\"}],\"stateMutability\":\"view\",\"type\":\"function\"},{\"inputs\":[{\"internalType\":\"uint256[]\",\"name\":\"kvIndices\",\"type\":\"uint256[]\"}],\"name\":\"getKvMetas\",\"outputs\":[{\"internalType\":\"bytes32[]\",\"name\":\"\",\"type\":\"bytes32[]\"}],\"stateMutability\":\"view\",\"type\":\"function\"},{\"inputs\":[{\"internalType\":\"bytes32\",\"name\":\"key\",\"type\":\"bytes32\"}],\"name\":\"hash\",\"outputs\":[{\"internalType\":\"bytes24\",\"name\":\"\",\"type\":\"bytes24\"}],\"stateMutability\":\"view\",\"type\":\"function\"},{\"inputs\":[],\"name\":\"lastKvIdx\",\"outputs\":[{\"internalType\":\"uint40\",\"name\":\"\",\"type\":\"uint40\"}],\"stateMutability\":\"view\",\"type\":\"function\"},{\"inputs\":[],\"name\":\"maxKvSize\",\"outputs\":[{\"internalType\":\"uint256\",\"name\":\"\",\"type\":\"uint256\"}],\"stateMutability\":\"view\",\"type\":\"function\"},{\"inputs\":[],\"name\":\"owner\",\"outputs\":[{\"internalType\":\"address\",\"name\":\"\",\"type\":\"address\"}],\"stateMutability\":\"view\",\"type\":\"function\"},{\"inputs\":[{\"internalType\":\"bytes32\",\"name\":\"key\",\"type\":\"bytes32\"}],\"name\":\"remove\",\"outputs\":[],\"stateMutability\":\"nonpayable\",\"type\":\"function\"},{\"inputs\":[{\"internalType\":\"bytes32\",\"name\":\"key\",\"type\":\"bytes32\"},{\"internalType\":\"address\",\"name\":\"to\",\"type\":\"address\"}],\"name\":\"removeTo\",\"outputs\":[],\"stateMutability\":\"nonpayable\",\"type\":\"function\"},{\"inputs\":[],\"name\":\"renounceOwnership\",\"outputs\":[],\"stateMutability\":\"nonpayable\",\"type\":\"function\"},{\"inputs\":[{\"internalType\":\"bytes32\",\"name\":\"key\",\"type\":\"bytes32\"}],\"name\":\"size\",\"outputs\":[{\"internalType\":\"uint256\",\"name\":\"\",\"type\":\"uint256\"}],\"stateMutability\":\"view\",\"type\":\"function\"},{\"inputs\":[],\"name\":\"startTime\",\"outputs\":[{\"internalType\":\"uint256\",\"name\":\"\",\"type\":\"uint256\"}],\"stateMutability\":\"view\",\"type\":\"function\"},{\"inputs\":[],\"name\":\"storageCost\",\"outputs\":[{\"internalType\":\"uint256\",\"name\":\"\",\"type\":\"uint256\"}],\"stateMutability\":\"view\",\"type\":\"function\"},{\"inputs\":[{\"internalType\":\"address\",\"name\":\"newOwner\",\"type\":\"address\"}],\"name\":\"transferOwnership\",\"outputs\":[],\"stateMutability\":\"nonpayable\",\"type\":\"function\"},{\"inputs\":[],\"name\":\"upfrontPayment\",\"outputs\":[{\"internalType\":\"uint256\",\"name\":\"\",\"type\":\"uint256\"}],\"stateMutability\":\"view\",\"type\":\"function\"}]"
+	blobSize           = 32 * 4096
+)

op-service/sources/l1_beacon_client.go

@@ -26,6 +26,7 @@ const (
 
 type L1BeaconClientConfig struct {
 	FetchAllSidecars bool
+	ESClient         *ESClient
 }
 
 type L1BeaconClient struct {
@@ -130,22 +131,44 @@ func (cl *L1BeaconClient) GetBlobSidecars(ctx context.Context, ref eth.L1BlockRe
 
 	apiscs := make([]*eth.APIBlobSidecar, 0, len(hashes))
 	// filter and order by hashes
+	var missingHashes []eth.IndexedBlobHash
 	for _, h := range hashes {
+		found := false
 		for _, apisc := range resp.Data {
 			if h.Index == uint64(apisc.Index) {
+				found = true
 				apiscs = append(apiscs, apisc)
 				break
 			}
 		}
+		if !found {
+			missingHashes = append(missingHashes, h)
+			// nil as a place holder
+			apiscs = append(apiscs, nil)
+		}
 	}
 
-	if len(hashes) != len(apiscs) {
-		return nil, fmt.Errorf("expected %v sidecars but got %v", len(hashes), len(apiscs))
+	var fetchedFromES []*eth.BlobSidecar
+	if len(missingHashes) > 0 && cl.cfg.ESClient != nil {
+		fetchedFromES, err = cl.cfg.ESClient.GetBlobs(missingHashes)
+		if err != nil {
+			return nil, fmt.Errorf("failed to fetch missing blobs from es client:%v", err)
+		}
+		missingHashes = nil
+	}
+
+	if len(missingHashes) > 0 {
+		return nil, fmt.Errorf("expected %v sidecars but got %v", len(hashes), len(hashes)-len(missingHashes))
 	}
 
 	bscs := make([]*eth.BlobSidecar, 0, len(hashes))
 	for _, apisc := range apiscs {
-		bscs = append(bscs, apisc.BlobSidecar())
+		if apisc != nil {
+			bscs = append(bscs, apisc.BlobSidecar())
+		} else {
+			bscs = append(bscs, fetchedFromES[0])
+			fetchedFromES = fetchedFromES[1:]
+		}
 	}
 
 	return bscs, nil