op-node,op-batcher: implement span channel out block count limit

op-batcher/batcher/channel_builder.go

@@ -94,7 +94,10 @@ func NewChannelBuilder(cfg ChannelConfig, rollupCfg rollup.Config, latestL1Origi
 	chainSpec := rollup.NewChainSpec(&rollupCfg)
 	var co derive.ChannelOut
 	if cfg.BatchType == derive.SpanBatchType {
-		co, err = derive.NewSpanChannelOut(rollupCfg.Genesis.L2Time, rollupCfg.L2ChainID, cfg.CompressorConfig.TargetOutputSize, cfg.CompressorConfig.CompressionAlgo, chainSpec)
+		co, err = derive.NewSpanChannelOut(
+			rollupCfg.Genesis.L2Time, rollupCfg.L2ChainID,
+			cfg.CompressorConfig.TargetOutputSize, cfg.CompressorConfig.CompressionAlgo,
+			chainSpec, derive.SCOWithMaxBlocksPerSpanBatch(cfg.MaxBlocksPerSpanBatch))
 	} else {
 		co, err = derive.NewSingularChannelOut(c, chainSpec)
 	}

op-batcher/batcher/channel_config.go

@@ -29,6 +29,9 @@ type ChannelConfig struct {
 	SubSafetyMargin uint64
 	// The maximum byte-size a frame can have.
 	MaxFrameSize uint64
+	// MaxBlocksPerSpanBatch is the maximum number of blocks to add to a span batch.
+	// A value of 0 disables a maximum.
+	MaxBlocksPerSpanBatch int
 
 	// Target number of frames to create per channel.
 	// For blob transactions, this controls the number of blobs to target adding

op-batcher/batcher/config.go

@@ -56,6 +56,9 @@ type CLIConfig struct {
 	// If using blobs, this setting is ignored and the max blob size is used.
 	MaxL1TxSize uint64
 
+	// Maximum number of blocks to add to a span batch. Default is 0 - no maximum.
+	MaxBlocksPerSpanBatch int
+
 	// The target number of frames to create per channel. Controls number of blobs
 	// per blob tx, if using Blob DA.
 	TargetNumFrames int
@@ -173,6 +176,7 @@ func NewConfig(ctx *cli.Context) *CLIConfig {
 		MaxPendingTransactions:       ctx.Uint64(flags.MaxPendingTransactionsFlag.Name),
 		MaxChannelDuration:           ctx.Uint64(flags.MaxChannelDurationFlag.Name),
 		MaxL1TxSize:                  ctx.Uint64(flags.MaxL1TxSizeBytesFlag.Name),
+		MaxBlocksPerSpanBatch:        ctx.Int(flags.MaxBlocksPerSpanBatch.Name),
 		TargetNumFrames:              ctx.Int(flags.TargetNumFramesFlag.Name),
 		ApproxComprRatio:             ctx.Float64(flags.ApproxComprRatioFlag.Name),
 		Compressor:                   ctx.String(flags.CompressorFlag.Name),

op-batcher/batcher/service.go

@@ -192,13 +192,14 @@ func (bs *BatcherService) initChannelConfig(cfg *CLIConfig) error {
 		channelTimeout = bs.RollupConfig.ChannelTimeoutGranite
 	}
 	cc := ChannelConfig{
-		SeqWindowSize:      bs.RollupConfig.SeqWindowSize,
-		ChannelTimeout:     channelTimeout,
-		MaxChannelDuration: cfg.MaxChannelDuration,
-		MaxFrameSize:       cfg.MaxL1TxSize - 1, // account for version byte prefix; reset for blobs
-		TargetNumFrames:    cfg.TargetNumFrames,
-		SubSafetyMargin:    cfg.SubSafetyMargin,
-		BatchType:          cfg.BatchType,
+		SeqWindowSize:         bs.RollupConfig.SeqWindowSize,
+		ChannelTimeout:        channelTimeout,
+		MaxChannelDuration:    cfg.MaxChannelDuration,
+		MaxFrameSize:          cfg.MaxL1TxSize - 1, // account for version byte prefix; reset for blobs
+		MaxBlocksPerSpanBatch: cfg.MaxBlocksPerSpanBatch,
+		TargetNumFrames:       cfg.TargetNumFrames,
+		SubSafetyMargin:       cfg.SubSafetyMargin,
+		BatchType:             cfg.BatchType,
 	}
 
 	switch cfg.DataAvailabilityType {

op-batcher/flags/flags.go

@@ -76,6 +76,11 @@ var (
 		Value:   120_000, // will be overwritten to max for blob da-type
 		EnvVars: prefixEnvVars("MAX_L1_TX_SIZE_BYTES"),
 	}
+	MaxBlocksPerSpanBatch = &cli.IntFlag{
+		Name:    "max-blocks-per-span-batch",
+		Usage:   "Maximum number of blocks to add to a span batch. Default is 0 - no maximum.",
+		EnvVars: prefixEnvVars("MAX_BLOCKS_PER_SPAN_BATCH"),
+	}
 	TargetNumFramesFlag = &cli.IntFlag{
 		Name:    "target-num-frames",
 		Usage:   "The target number of frames to create per channel. Controls number of blobs per blob tx, if using Blob DA.",
@@ -169,6 +174,7 @@ var optionalFlags = []cli.Flag{
 	MaxPendingTransactionsFlag,
 	MaxChannelDurationFlag,
 	MaxL1TxSizeBytesFlag,
+	MaxBlocksPerSpanBatch,
 	TargetNumFramesFlag,
 	ApproxComprRatioFlag,
 	CompressorFlag,

op-e2e/setup.go

@@ -288,6 +288,12 @@ type SystemConfig struct {
 	// whether to actually use BatcherMaxL1TxSizeBytes for blobs, insteaf of max blob size
 	BatcherUseMaxTxSizeForBlobs bool
 
+	// Singular (0) or span batches (1)
+	BatcherBatchType uint
+
+	// If 0, limits the number of blocks per span batch
+	BatcherMaxBlocksPerSpanBatch int
+
 	// SupportL1TimeTravel determines if the L1 node supports quickly skipping forward in time
 	SupportL1TimeTravel bool
 
@@ -884,10 +890,6 @@ func (cfg SystemConfig) Start(t *testing.T, _opts ...SystemConfigOption) (*Syste
 	}
 	sys.L2OutputSubmitter = proposer
 
-	var batchType uint = derive.SingularBatchType
-	if cfg.DeployConfig.L2GenesisDeltaTimeOffset != nil && *cfg.DeployConfig.L2GenesisDeltaTimeOffset == hexutil.Uint64(0) {
-		batchType = derive.SpanBatchType
-	}
 	// batcher defaults if unset
 	batcherMaxL1TxSizeBytes := cfg.BatcherMaxL1TxSizeBytes
 	if batcherMaxL1TxSizeBytes == 0 {
@@ -921,10 +923,11 @@ func (cfg SystemConfig) Start(t *testing.T, _opts ...SystemConfigOption) (*Syste
 			Level:  log.LevelInfo,
 			Format: oplog.FormatText,
 		},
-		Stopped:              sys.Cfg.DisableBatcher, // Batch submitter may be enabled later
-		BatchType:            batchType,
-		DataAvailabilityType: sys.Cfg.DataAvailabilityType,
-		CompressionAlgo:      compressionAlgo,
+		Stopped:               sys.Cfg.DisableBatcher, // Batch submitter may be enabled later
+		BatchType:             cfg.BatcherBatchType,
+		MaxBlocksPerSpanBatch: cfg.BatcherMaxBlocksPerSpanBatch,
+		DataAvailabilityType:  sys.Cfg.DataAvailabilityType,
+		CompressionAlgo:       compressionAlgo,
 	}
 	// Batch Submitter
 	batcher, err := bss.BatcherServiceFromCLIConfig(context.Background(), "0.0.1", batcherCLIConfig, sys.Cfg.Loggers["batcher"])

op-e2e/system_test.go

@@ -55,22 +55,27 @@ import (
 func TestSystemBatchType(t *testing.T) {
 	tests := []struct {
 		name string
-		f    func(gt *testing.T, deltaTimeOffset *hexutil.Uint64)
+		f    func(*testing.T, func(*SystemConfig))
 	}{
 		{"StopStartBatcher", StopStartBatcher},
 	}
 	for _, test := range tests {
 		test := test
 		t.Run(test.name+"_SingularBatch", func(t *testing.T) {
-			test.f(t, nil)
+			test.f(t, func(sc *SystemConfig) {
+				sc.BatcherBatchType = derive.SingularBatchType
+			})
 		})
-	}
-
-	deltaTimeOffset := hexutil.Uint64(0)
-	for _, test := range tests {
-		test := test
 		t.Run(test.name+"_SpanBatch", func(t *testing.T) {
-			test.f(t, &deltaTimeOffset)
+			test.f(t, func(sc *SystemConfig) {
+				sc.BatcherBatchType = derive.SpanBatchType
+			})
+		})
+		t.Run(test.name+"_SpanBatchMaxBlocks", func(t *testing.T) {
+			test.f(t, func(sc *SystemConfig) {
+				sc.BatcherBatchType = derive.SpanBatchType
+				sc.BatcherMaxBlocksPerSpanBatch = 2
+			})
 		})
 	}
 }
@@ -1292,10 +1297,11 @@ func testFees(t *testing.T, cfg SystemConfig) {
 	require.Equal(t, balanceDiff, totalFee, "balances should add up")
 }
 
-func StopStartBatcher(t *testing.T, deltaTimeOffset *hexutil.Uint64) {
+func StopStartBatcher(t *testing.T, cfgMod func(*SystemConfig)) {
 	InitParallel(t)
 
-	cfg := DeltaSystemConfig(t, deltaTimeOffset)
+	cfg := DefaultSystemConfig(t)
+	cfgMod(&cfg)
 	sys, err := cfg.Start(t)
 	require.NoError(t, err, "Error starting up system")
 	defer sys.Close()