feat(driver): add blob datasource (#688)

Co-authored-by: maskpp <[email protected]>

cmd/flags/driver.go

@@ -35,6 +35,12 @@ var (
 		Value:    0,
 		Category: driverCategory,
 	}
+	// blob server endpoint
+	BlobServerEndpoint = &cli.StringFlag{
+		Name:     "blob.server",
+		Usage:    "Blob sidecar storage server",
+		Category: driverCategory,
+	}
 )
 
 // DriverFlags All driver flags.
@@ -47,4 +53,5 @@ var DriverFlags = MergeFlags(CommonFlags, []cli.Flag{
 	P2PSyncTimeout,
 	CheckPointSyncURL,
 	MaxExponent,
+	BlobServerEndpoint,
 })

driver/chain_syncer/calldata/syncer.go

@@ -5,27 +5,28 @@ import (
 	"errors"
 	"fmt"
 	"math/big"
+	"net/url"
 	"time"
 
+	"github.com/ethereum/go-ethereum"
 	"github.com/ethereum/go-ethereum/accounts/abi/bind"
 	"github.com/ethereum/go-ethereum/beacon/engine"
 	"github.com/ethereum/go-ethereum/common"
 	"github.com/ethereum/go-ethereum/core/rawdb"
 	"github.com/ethereum/go-ethereum/core/types"
 	"github.com/ethereum/go-ethereum/log"
 	"github.com/ethereum/go-ethereum/rlp"
-
-	"github.com/ethereum/go-ethereum"
 	"github.com/taikoxyz/taiko-client/bindings"
 	"github.com/taikoxyz/taiko-client/bindings/encoding"
-	anchorTxConstructor "github.com/taikoxyz/taiko-client/driver/anchor_tx_constructor"
 	"github.com/taikoxyz/taiko-client/driver/chain_syncer/beaconsync"
 	"github.com/taikoxyz/taiko-client/driver/state"
-	txlistfetcher "github.com/taikoxyz/taiko-client/driver/txlist_fetcher"
 	"github.com/taikoxyz/taiko-client/internal/metrics"
 	"github.com/taikoxyz/taiko-client/internal/utils"
-	eventIterator "github.com/taikoxyz/taiko-client/pkg/chain_iterator/event_iterator"
 	"github.com/taikoxyz/taiko-client/pkg/rpc"
+
+	anchorTxConstructor "github.com/taikoxyz/taiko-client/driver/anchor_tx_constructor"
+	txlistfetcher "github.com/taikoxyz/taiko-client/driver/txlist_fetcher"
+	eventIterator "github.com/taikoxyz/taiko-client/pkg/chain_iterator/event_iterator"
 	txListValidator "github.com/taikoxyz/taiko-client/pkg/txlist_validator"
 )
 
@@ -42,6 +43,7 @@ type Syncer struct {
 	lastInsertedBlockID *big.Int
 	reorgDetectedFlag   bool
 	maxRetrieveExponent uint64
+	blobDatasource      *rpc.BlobDataSource
 }
 
 // NewSyncer creates a new syncer instance.
@@ -51,6 +53,7 @@ func NewSyncer(
 	state *state.State,
 	progressTracker *beaconsync.SyncProgressTracker,
 	maxRetrieveExponent uint64,
+	blobServerEndpoint *url.URL,
 ) (*Syncer, error) {
 	configs, err := client.TaikoL1.GetConfig(&bind.CallOpts{Context: ctx})
 	if err != nil {
@@ -74,6 +77,11 @@ func NewSyncer(
 			client.L2.ChainID,
 		),
 		maxRetrieveExponent: maxRetrieveExponent,
+		blobDatasource: rpc.NewBlobDataSource(
+			ctx,
+			client,
+			blobServerEndpoint,
+		),
 	}, nil
 }
 
@@ -239,7 +247,7 @@ func (s *Syncer) onBlockProposed(
 	// Decode transactions list.
 	var txListDecoder txlistfetcher.TxListFetcher
 	if event.Meta.BlobUsed {
-		txListDecoder = txlistfetcher.NewBlobTxListFetcher(s.rpc.L1Beacon)
+		txListDecoder = txlistfetcher.NewBlobTxListFetcher(s.rpc.L1Beacon, s.blobDatasource)
 	} else {
 		txListDecoder = new(txlistfetcher.CalldataFetcher)
 	}

driver/chain_syncer/calldata/syncer_test.go

@@ -41,6 +41,7 @@ func (s *CalldataSyncerTestSuite) SetupTest() {
 		state2,
 		beaconsync.NewSyncProgressTracker(s.RPCClient.L2, 1*time.Hour),
 		0,
+		nil,
 	)
 	s.Nil(err)
 	s.s = syncer
@@ -56,6 +57,7 @@ func (s *CalldataSyncerTestSuite) TestCancelNewSyncer() {
 		s.s.state,
 		s.s.progressTracker,
 		0,
+		nil,
 	)
 	s.Nil(syncer)
 	s.NotNil(err)

driver/chain_syncer/chain_syncer.go

@@ -3,6 +3,7 @@ package chainsyncer
 import (
 	"context"
 	"fmt"
+	"net/url"
 	"time"
 
 	"github.com/ethereum/go-ethereum/accounts/abi/bind"
@@ -45,6 +46,8 @@ func New(
 	p2pSyncVerifiedBlocks bool,
 	p2pSyncTimeout time.Duration,
 	maxRetrieveExponent uint64,
+	blobServerEndpoint *url.URL,
+
 ) (*L2ChainSyncer, error) {
 	tracker := beaconsync.NewSyncProgressTracker(rpc.L2, p2pSyncTimeout)
 	go tracker.Track(ctx)
@@ -54,7 +57,7 @@ func New(
 		return nil, err
 	}
 	beaconSyncer := beaconsync.NewSyncer(ctx, rpc, state, syncMode, tracker)
-	calldataSyncer, err := calldata.NewSyncer(ctx, rpc, state, tracker, maxRetrieveExponent)
+	calldataSyncer, err := calldata.NewSyncer(ctx, rpc, state, tracker, maxRetrieveExponent, blobServerEndpoint)
 	if err != nil {
 		return nil, err
 	}

driver/chain_syncer/chain_syncer_test.go

@@ -41,6 +41,7 @@ func (s *ChainSyncerTestSuite) SetupTest() {
 		false,
 		1*time.Hour,
 		0,
+		nil,
 	)
 	s.Nil(err)
 	s.s = syncer

driver/config.go

@@ -3,6 +3,7 @@ package driver
 import (
 	"errors"
 	"fmt"
+	"net/url"
 	"time"
 
 	"github.com/ethereum/go-ethereum/common"
@@ -21,6 +22,7 @@ type Config struct {
 	RPCTimeout            time.Duration
 	RetryInterval         time.Duration
 	MaxExponent           uint64
+	BlobServerEndpoint    *url.URL
 }
 
 // NewConfigFromCliContext creates a new config instance from
@@ -44,6 +46,15 @@ func NewConfigFromCliContext(c *cli.Context) (*Config, error) {
 		return nil, errors.New("empty L1 beacon endpoint")
 	}
 
+	var blobServerEndpoint *url.URL
+	if c.IsSet(flags.BlobServerEndpoint.Name) {
+		if blobServerEndpoint, err = url.Parse(
+			c.String(flags.BlobServerEndpoint.Name),
+		); err != nil {
+			return nil, err
+		}
+	}
+
 	var timeout = c.Duration(flags.RPCTimeout.Name)
 	return &Config{
 		ClientConfig: &rpc.ClientConfig{
@@ -62,5 +73,6 @@ func NewConfigFromCliContext(c *cli.Context) (*Config, error) {
 		P2PSyncTimeout:        c.Duration(flags.P2PSyncTimeout.Name),
 		RPCTimeout:            timeout,
 		MaxExponent:           c.Uint64(flags.MaxExponent.Name),
+		BlobServerEndpoint:    blobServerEndpoint,
 	}, nil
 }

driver/driver.go

@@ -80,6 +80,7 @@ func (d *Driver) InitFromConfig(ctx context.Context, cfg *Config) (err error) {
 		cfg.P2PSyncVerifiedBlocks,
 		cfg.P2PSyncTimeout,
 		cfg.MaxExponent,
+		cfg.BlobServerEndpoint,
 	); err != nil {
 		return err
 	}

driver/txlist_fetcher/blob.go

@@ -11,17 +11,19 @@ import (
 	"github.com/ethereum/go-ethereum/log"
 
 	"github.com/taikoxyz/taiko-client/bindings"
+	"github.com/taikoxyz/taiko-client/pkg"
 	"github.com/taikoxyz/taiko-client/pkg/rpc"
 )
 
 // BlobFetcher is responsible for fetching the txList blob from the L1 block sidecar.
 type BlobFetcher struct {
 	l1Beacon *rpc.BeaconClient
+	ds       *rpc.BlobDataSource
 }
 
 // NewBlobTxListFetcher creates a new BlobFetcher instance based on the given rpc client.
-func NewBlobTxListFetcher(l1Beacon *rpc.BeaconClient) *BlobFetcher {
-	return &BlobFetcher{l1Beacon}
+func NewBlobTxListFetcher(l1Beacon *rpc.BeaconClient, ds *rpc.BlobDataSource) *BlobFetcher {
+	return &BlobFetcher{l1Beacon, ds}
 }
 
 // Fetch implements the TxListFetcher interface.
@@ -31,11 +33,11 @@ func (d *BlobFetcher) Fetch(
 	meta *bindings.TaikoDataBlockMetadata,
 ) ([]byte, error) {
 	if !meta.BlobUsed {
-		return nil, errBlobUnused
+		return nil, pkg.ErrBlobUsed
 	}
 
 	// Fetch the L1 block sidecars.
-	sidecars, err := d.l1Beacon.GetBlobs(ctx, meta.Timestamp)
+	sidecars, err := d.ds.GetBlobs(ctx, meta)
 	if err != nil {
 		return nil, err
 	}
@@ -61,5 +63,5 @@ func (d *BlobFetcher) Fetch(
 		}
 	}
 
-	return nil, errSidecarNotFound
+	return nil, pkg.ErrSidecarNotFound
 }

driver/txlist_fetcher/calldata.go

@@ -6,6 +6,7 @@ import (
 	"github.com/ethereum/go-ethereum/core/types"
 	"github.com/taikoxyz/taiko-client/bindings"
 	"github.com/taikoxyz/taiko-client/bindings/encoding"
+	"github.com/taikoxyz/taiko-client/pkg"
 )
 
 // CalldataFetcher is responsible for fetching the txList bytes from the transaction's calldata.
@@ -18,7 +19,7 @@ func (d *CalldataFetcher) Fetch(
 	meta *bindings.TaikoDataBlockMetadata,
 ) ([]byte, error) {
 	if meta.BlobUsed {
-		return nil, errBlobUsed
+		return nil, pkg.ErrBlobUsed
 	}
 
 	return encoding.UnpackTxListBytes(tx.Data())

driver/txlist_fetcher/interface.go

@@ -2,18 +2,11 @@ package txlistdecoder
 
 import (
 	"context"
-	"errors"
 
 	"github.com/ethereum/go-ethereum/core/types"
 	"github.com/taikoxyz/taiko-client/bindings"
 )
 
-var (
-	errBlobUsed        = errors.New("blob is used")
-	errBlobUnused      = errors.New("blob is not used")
-	errSidecarNotFound = errors.New("sidecar not found")
-)
-
 // TxListFetcher is responsible for fetching the L2 txList bytes from L1
 type TxListFetcher interface {
 	Fetch(ctx context.Context, tx *types.Transaction, meta *bindings.TaikoDataBlockMetadata) ([]byte, error)

internal/testutils/helper.go

@@ -287,7 +287,7 @@ func RandomBytes(size int) (b []byte) {
 func RandomPort() int {
 	port, err := freeport.GetFreePort()
 	if err != nil {
-		log.Crit("Failed to get local free random port", "err", err)
+		log.Crit("Failed to get local free random port", "error", err)
 	}
 	return port
 }
@@ -298,7 +298,7 @@ func LocalRandomProverEndpoint() *url.URL {
 
 	proverEndpoint, err := url.Parse(fmt.Sprintf("http://localhost:%v", port))
 	if err != nil {
-		log.Crit("Failed to parse local prover endpoint", "err", err)
+		log.Crit("Failed to parse local prover endpoint", "error", err)
 	}
 
 	return proverEndpoint

internal/utils/utils.go

@@ -22,14 +22,14 @@ import (
 func LoadEnv() {
 	currentPath, err := os.Getwd()
 	if err != nil {
-		log.Debug("Failed to get current path", "err", err)
+		log.Debug("Failed to get current path", "error", err)
 	}
 	path := strings.Split(currentPath, "/taiko-client")
 	if len(path) == 0 {
 		log.Debug("Not a taiko-client repo")
 	}
 	if godotenv.Load(fmt.Sprintf("%s/taiko-client/integration_test/.env", path[0])) != nil {
-		log.Debug("Failed to load test env", "current path", currentPath, "err", err)
+		log.Debug("Failed to load test env", "current path", currentPath, "error", err)
 	}
 }
 

pkg/error.go

@@ -0,0 +1,12 @@
+package pkg
+
+import (
+	"errors"
+)
+
+var (
+	ErrBlobUsed        = errors.New("blob is used")
+	ErrBlobUnused      = errors.New("blob is not used")
+	ErrSidecarNotFound = errors.New("sidecar not found")
+	ErrBeaconNotFound  = errors.New("beacon client not found")
+)