bump batch size and make it a config

Author: tac0turtle

block/internal/da/client.go

@@ -32,29 +32,39 @@ type client struct {
 	da              coreda.DA
 	logger          zerolog.Logger
 	defaultTimeout  time.Duration
+	batchSize       int
 	namespaceBz     []byte
 	namespaceDataBz []byte
 }
 
+const (
+	defaultRetrieveBatchSize = 150
+)
+
 // Config contains configuration for the DA client.
 type Config struct {
-	DA             coreda.DA
-	Logger         zerolog.Logger
-	DefaultTimeout time.Duration
-	Namespace      string
-	DataNamespace  string
+	DA                coreda.DA
+	Logger            zerolog.Logger
+	DefaultTimeout    time.Duration
+	Namespace         string
+	DataNamespace     string
+	RetrieveBatchSize int
 }
 
 // NewClient creates a new DA client with pre-calculated namespace bytes.
 func NewClient(cfg Config) *client {
 	if cfg.DefaultTimeout == 0 {
 		cfg.DefaultTimeout = 30 * time.Second
 	}
+	if cfg.RetrieveBatchSize <= 0 {
+		cfg.RetrieveBatchSize = defaultRetrieveBatchSize
+	}
 
 	return &client{
 		da:              cfg.DA,
 		logger:          cfg.Logger.With().Str("component", "da_client").Logger(),
 		defaultTimeout:  cfg.DefaultTimeout,
+		batchSize:       cfg.RetrieveBatchSize,
 		namespaceBz:     coreda.NamespaceFromString(cfg.Namespace).Bytes(),
 		namespaceDataBz: coreda.NamespaceFromString(cfg.DataNamespace).Bytes(),
 	}
@@ -203,7 +213,8 @@ func (c *client) Retrieve(ctx context.Context, height uint64, namespace []byte)
 		}
 	}
 	// 2. Get Blobs using the retrieved IDs in batches
-	batchSize := 100
+	// Each batch has its own timeout while keeping the link to the parent context
+	batchSize := c.batchSize
 	blobs := make([][]byte, 0, len(idsResult.IDs))
 	for i := 0; i < len(idsResult.IDs); i += batchSize {
 		end := min(i+batchSize, len(idsResult.IDs))

block/public.go

@@ -1,6 +1,8 @@
 package block
 
 import (
+	"time"
+
 	"github.com/evstack/ev-node/block/internal/common"
 	"github.com/evstack/ev-node/block/internal/da"
 	coreda "github.com/evstack/ev-node/core/da"
@@ -39,9 +41,11 @@ func NewDAClient(
 	logger zerolog.Logger,
 ) DAClient {
 	return da.NewClient(da.Config{
-		DA:            daLayer,
-		Logger:        logger,
-		Namespace:     config.DA.GetNamespace(),
-		DataNamespace: config.DA.GetDataNamespace(),
+		DA:                daLayer,
+		Logger:            logger,
+		Namespace:         config.DA.GetNamespace(),
+		DefaultTimeout:    15 * time.Second,
+		DataNamespace:     config.DA.GetDataNamespace(),
+		RetrieveBatchSize: config.DA.RetrieveBatchSize,
 	})
 }

pkg/config/config.go

@@ -70,6 +70,8 @@ const (
 	FlagDAMempoolTTL = FlagPrefixEvnode + "da.mempool_ttl"
 	// FlagDAMaxSubmitAttempts is a flag for specifying the maximum DA submit attempts
 	FlagDAMaxSubmitAttempts = FlagPrefixEvnode + "da.max_submit_attempts"
+	// FlagDARetrieveBatchSize configures how many IDs are fetched per DA Get request
+	FlagDARetrieveBatchSize = FlagPrefixEvnode + "da.retrieve_batch_size"
 
 	// P2P configuration flags
 
@@ -162,6 +164,7 @@ type DAConfig struct {
 	BlockTime         DurationWrapper `mapstructure:"block_time" yaml:"block_time" comment:"Average block time of the DA chain (duration). Determines frequency of DA layer syncing, maximum backoff time for retries, and is multiplied by MempoolTTL to calculate transaction expiration. Examples: \"15s\", \"30s\", \"1m\", \"2m30s\", \"10m\"."`
 	MempoolTTL        uint64          `mapstructure:"mempool_ttl" yaml:"mempool_ttl" comment:"Number of DA blocks after which a transaction is considered expired and dropped from the mempool. Controls retry backoff timing."`
 	MaxSubmitAttempts int             `mapstructure:"max_submit_attempts" yaml:"max_submit_attempts" comment:"Maximum number of attempts to submit data to the DA layer before giving up. Higher values provide more resilience but can delay error reporting."`
+	RetrieveBatchSize int             `mapstructure:"retrieve_batch_size" yaml:"retrieve_batch_size" comment:"Number of IDs to request per DA Get call when retrieving blobs. Smaller batches lower per-request latency; larger batches reduce the number of RPC round trips. Default: 100."`
 }
 
 // GetNamespace returns the namespace for header submissions.
@@ -320,6 +323,7 @@ func AddFlags(cmd *cobra.Command) {
 	cmd.Flags().StringSlice(FlagDASigningAddresses, def.DA.SigningAddresses, "Comma-separated list of addresses for DA submissions (used in round-robin)")
 	cmd.Flags().Uint64(FlagDAMempoolTTL, def.DA.MempoolTTL, "number of DA blocks until transaction is dropped from the mempool")
 	cmd.Flags().Int(FlagDAMaxSubmitAttempts, def.DA.MaxSubmitAttempts, "maximum number of attempts to submit data to the DA layer before giving up")
+	cmd.Flags().Int(FlagDARetrieveBatchSize, def.DA.RetrieveBatchSize, "number of IDs to request per DA Get call when retrieving blobs")
 
 	// P2P configuration flags
 	cmd.Flags().String(FlagP2PListenAddress, def.P2P.ListenAddress, "P2P listen address (host:port)")

pkg/config/config_test.go

@@ -26,6 +26,7 @@ func TestDefaultConfig(t *testing.T) {
 	assert.Equal(t, "", def.DA.AuthToken)
 	assert.Equal(t, "", def.DA.SubmitOptions)
 	assert.NotEmpty(t, def.DA.Namespace)
+	assert.Equal(t, 150, def.DA.RetrieveBatchSize)
 	assert.Equal(t, 1*time.Second, def.Node.BlockTime.Duration)
 	assert.Equal(t, 6*time.Second, def.DA.BlockTime.Duration)
 	assert.Equal(t, uint64(0), def.DA.MempoolTTL)
@@ -70,6 +71,7 @@ func TestAddFlags(t *testing.T) {
 	assertFlagValue(t, flags, FlagDASigningAddresses, DefaultConfig().DA.SigningAddresses)
 	assertFlagValue(t, flags, FlagDAMempoolTTL, DefaultConfig().DA.MempoolTTL)
 	assertFlagValue(t, flags, FlagDAMaxSubmitAttempts, DefaultConfig().DA.MaxSubmitAttempts)
+	assertFlagValue(t, flags, FlagDARetrieveBatchSize, DefaultConfig().DA.RetrieveBatchSize)
 
 	// P2P flags
 	assertFlagValue(t, flags, FlagP2PListenAddress, DefaultConfig().P2P.ListenAddress)
@@ -99,7 +101,7 @@ func TestAddFlags(t *testing.T) {
 	assertFlagValue(t, flags, FlagRPCAddress, DefaultConfig().RPC.Address)
 
 	// Count the number of flags we're explicitly checking
-	expectedFlagCount := 37 // Update this number if you add more flag checks above
+	expectedFlagCount := 38 // Update this number if you add more flag checks above
 
 	// Get the actual number of flags (both regular and persistent)
 	actualFlagCount := 0

pkg/config/defaults.go

@@ -75,6 +75,7 @@ func DefaultConfig() Config {
 			MaxSubmitAttempts: 30,
 			Namespace:         randString(10),
 			DataNamespace:     "",
+			RetrieveBatchSize: 150,
 		},
 		Instrumentation: DefaultInstrumentationConfig(),
 		Log: LogConfig{