op-node,op-batcher: implement span channel out block count limit (#11416)

* op-node: implement span channel out block count limit

* op-batcher: add max-blocks-per-span-batch flag

* op-e2e: test MaxBlocksPerSpanBatch in system test

* op-e2e: use span batches in 4844 test

* address Axel's review

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.WithMaxBlocksPerSpanBatch(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/eip4844_test.go

@@ -46,6 +46,7 @@ func testSystem4844E2E(t *testing.T, multiBlob bool, daType batcherFlags.DataAva
 
 	cfg := EcotoneSystemConfig(t, &genesisTime)
 	cfg.DataAvailabilityType = daType
+	cfg.BatcherBatchType = derive.SpanBatchType
 	cfg.DeployConfig.L1GenesisBlockBaseFeePerGas = (*hexutil.Big)(big.NewInt(7000))
 
 	const maxBlobs = 6

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()

op-node/rollup/derive/channel_out_test.go

@@ -5,6 +5,9 @@ import (
 	"io"
 	"math/big"
 	"math/rand"
+	"reflect"
+	"runtime"
+	"strconv"
 	"testing"
 
 	"github.com/ethereum/go-ethereum/common"
@@ -13,9 +16,17 @@ import (
 	"github.com/stretchr/testify/require"
 
 	"github.com/ethereum-optimism/optimism/op-node/rollup"
+	"github.com/ethereum-optimism/optimism/op-service/eth"
 )
 
-var rollupCfg rollup.Config
+var rollupCfg = rollup.Config{
+	Genesis: rollup.Genesis{
+		L2Time: uint64(1723618465),
+	},
+	BlockTime: 2,
+	L2ChainID: big.NewInt(420),
+	L1ChainID: big.NewInt(161),
+}
 
 // basic implementation of the Compressor interface that does no compression
 type nonCompressor struct {
@@ -204,7 +215,7 @@ func TestForceCloseTxData(t *testing.T) {
 	for i, test := range tests {
 		out, err := ForceCloseTxData(test.frames)
 		if test.errors {
-			require.NotNil(t, err, "Should error on tc %v", i)
+			require.Error(t, err, "Should error on tc %v", i)
 			require.Nil(t, out, "Should return no value in tc %v", i)
 		} else {
 			require.NoError(t, err, "Should not error on tc %v", i)
@@ -219,42 +230,46 @@ func TestBlockToBatchValidity(t *testing.T) {
 	require.ErrorContains(t, err, "has no transactions")
 }
 
-func SpanChannelAndBatches(t *testing.T, target uint64, len int, algo CompressionAlgo) (*SpanChannelOut, []*SingularBatch) {
+func SpanChannelAndBatches(t *testing.T, targetOutputSize uint64, numBatches int, algo CompressionAlgo, opts ...SpanChannelOutOption) (*SpanChannelOut, []*SingularBatch) {
 	// target is larger than one batch, but smaller than two batches
 	rng := rand.New(rand.NewSource(0x543331))
-	chainID := big.NewInt(rng.Int63n(1000))
+	chainID := rollupCfg.L2ChainID
 	txCount := 1
-	cout, err := NewSpanChannelOut(0, chainID, target, algo, rollup.NewChainSpec(&rollupCfg))
+	genesisTime := rollupCfg.Genesis.L2Time
+	cout, err := NewSpanChannelOut(genesisTime, chainID, targetOutputSize, algo, rollup.NewChainSpec(&rollupCfg), opts...)
 	require.NoError(t, err)
-	batches := make([]*SingularBatch, len)
+	batches := make([]*SingularBatch, 0, numBatches)
 	// adding the first batch should not cause an error
-	for i := 0; i < len; i++ {
+	for i := 0; i < numBatches; i++ {
 		singularBatch := RandomSingularBatch(rng, txCount, chainID)
-		batches[i] = singularBatch
+		// use default 2 sec block time
+		singularBatch.Timestamp = genesisTime + 420_000 + rollupCfg.BlockTime*uint64(i)
+		batches = append(batches, singularBatch)
 	}
 
 	return cout, batches
 }
 
 func TestSpanChannelOut(t *testing.T) {
-	tests := []struct {
-		name string
-		f    func(t *testing.T, algo CompressionAlgo)
-	}{
-		{"SpanChannelOutCompressionOnlyOneBatch", SpanChannelOutCompressionOnlyOneBatch},
-		{"SpanChannelOutCompressionUndo", SpanChannelOutCompressionUndo},
-		{"SpanChannelOutClose", SpanChannelOutClose},
+	tests := []func(t *testing.T, algo CompressionAlgo){
+		SpanChannelOutCompressionOnlyOneBatch,
+		SpanChannelOutCompressionUndo,
+		SpanChannelOutClose,
 	}
 	for _, test := range tests {
 		test := test
 		for _, algo := range CompressionAlgos {
-			t.Run(test.name+"_"+algo.String(), func(t *testing.T) {
-				test.f(t, algo)
+			t.Run(funcName(test)+"_"+algo.String(), func(t *testing.T) {
+				test(t, algo)
 			})
 		}
 	}
 }
 
+func funcName(fn any) string {
+	return runtime.FuncForPC(reflect.ValueOf(fn).Pointer()).Name()
+}
+
 // TestSpanChannelOutCompressionOnlyOneBatch tests that the SpanChannelOut compression works as expected when there is only one batch
 // and it is larger than the target size. The single batch should be compressed, and the channel should now be full
 func SpanChannelOutCompressionOnlyOneBatch(t *testing.T, algo CompressionAlgo) {
@@ -276,7 +291,7 @@ func SpanChannelOutCompressionOnlyOneBatch(t *testing.T, algo CompressionAlgo) {
 // TestSpanChannelOutCompressionUndo tests that the SpanChannelOut compression rejects a batch that would cause the channel to be overfull
 func SpanChannelOutCompressionUndo(t *testing.T, algo CompressionAlgo) {
 	// target is larger than one batch, but smaller than two batches
-	cout, singularBatches := SpanChannelAndBatches(t, 750, 2, algo)
+	cout, singularBatches := SpanChannelAndBatches(t, 1100, 2, algo)
 
 	err := cout.AddSingularBatch(singularBatches[0], 0)
 	require.NoError(t, err)
@@ -301,7 +316,7 @@ func SpanChannelOutCompressionUndo(t *testing.T, algo CompressionAlgo) {
 // TestSpanChannelOutClose tests that the SpanChannelOut compression works as expected when the channel is closed.
 // it should compress the batch even if it is smaller than the target size because the channel is closing
 func SpanChannelOutClose(t *testing.T, algo CompressionAlgo) {
-	target := uint64(600)
+	target := uint64(1100)
 	cout, singularBatches := SpanChannelAndBatches(t, target, 1, algo)
 
 	err := cout.AddSingularBatch(singularBatches[0], 0)
@@ -325,3 +340,150 @@ func SpanChannelOutClose(t *testing.T, algo CompressionAlgo) {
 	require.Greater(t, cout.compressor.Len(), 0)
 	require.Equal(t, rlpLen, cout.activeRLP().Len())
 }
+
+type maxBlocksTest struct {
+	outputSize        uint64
+	exactFull         bool // whether the outputSize is exactly hit by the last batch
+	numBatches        int  // the last batch should cause the compressor to be full
+	maxBlocks         int
+	expNumSpanBatches int
+	expLastNumBlocks  int
+}
+
+// This tests sets a max blocks per span batch and causes multiple span batches
+// within a single channel. It then does a full round trip, encoding and decoding
+// the channel, confirming that the expected batches were encoded.
+func TestSpanChannelOut_MaxBlocksPerSpanBatch(t *testing.T) {
+	for i, tt := range []maxBlocksTest{
+		{
+			outputSize:        10_751,
+			exactFull:         true,
+			numBatches:        15,
+			maxBlocks:         4,
+			expNumSpanBatches: 4,
+			expLastNumBlocks:  3,
+		},
+		{
+			outputSize:        11_000,
+			numBatches:        16,
+			maxBlocks:         4,
+			expNumSpanBatches: 4,
+			expLastNumBlocks:  3,
+		},
+		{
+			outputSize:        11_154,
+			exactFull:         true,
+			numBatches:        16,
+			maxBlocks:         4,
+			expNumSpanBatches: 4,
+			expLastNumBlocks:  4,
+		},
+		{
+			outputSize:        11_500,
+			numBatches:        17,
+			maxBlocks:         4,
+			expNumSpanBatches: 4,
+			expLastNumBlocks:  4,
+		},
+		{
+			outputSize:        11_801,
+			exactFull:         true,
+			numBatches:        17,
+			maxBlocks:         4,
+			expNumSpanBatches: 5,
+			expLastNumBlocks:  1,
+		},
+		{
+			outputSize:        12_000,
+			numBatches:        18,
+			maxBlocks:         4,
+			expNumSpanBatches: 5,
+			expLastNumBlocks:  1,
+		},
+	} {
+		t.Run("test-"+strconv.Itoa(i), func(t *testing.T) {
+			testSpanChannelOut_MaxBlocksPerSpanBatch(t, tt)
+		})
+	}
+}
+
+func testSpanChannelOut_MaxBlocksPerSpanBatch(t *testing.T, tt maxBlocksTest) {
+	l1Origin := eth.L1BlockRef{Number: rollupCfg.Genesis.L1.Number + 42_000, Hash: common.Hash{0xde, 0xad, 0x42}}
+	l2SafeHead := eth.L2BlockRef{Number: rollupCfg.Genesis.L2Time + 40_000}
+	cout, bs := SpanChannelAndBatches(t, tt.outputSize, tt.numBatches, Brotli, WithMaxBlocksPerSpanBatch(tt.maxBlocks))
+	for i, b := range bs {
+		b.EpochNum = rollup.Epoch(l1Origin.Number)
+		b.EpochHash = l1Origin.Hash
+		err := cout.AddSingularBatch(b, uint64(i))
+		if i != tt.numBatches-1 || tt.exactFull {
+			require.NoErrorf(t, err, "iteration %d", i)
+		} else {
+			// adding last batch should not succeed, if not making compressor exactly full
+			require.ErrorIs(t, err, ErrCompressorFull)
+			t.Logf("full compressor length: %d", cout.compressor.Len())
+		}
+
+	}
+	require.ErrorIs(t, cout.FullErr(), ErrCompressorFull)
+	expSpanBatchBlocks := tt.expLastNumBlocks
+	if !tt.exactFull {
+		// if we didn't fill up exactly, we expect that one more block got
+		// added to the current span batch to detect that the compressor is full
+		expSpanBatchBlocks = tt.expLastNumBlocks%tt.maxBlocks + 1
+	}
+	require.Equal(t, expSpanBatchBlocks, cout.spanBatch.GetBlockCount(),
+		"last block should still have been added to the span batch")
+	require.NoError(t, cout.Close())
+
+	// write cannel into a single frame
+	var frameBuf bytes.Buffer
+	fn, err := cout.OutputFrame(&frameBuf, tt.outputSize+FrameV0OverHeadSize)
+	require.Zero(t, fn)
+	require.ErrorIs(t, err, io.EOF)
+
+	// now roundtrip to decode the batches
+	var frame Frame
+	require.NoError(t, frame.UnmarshalBinary(&frameBuf))
+	require.True(t, frame.IsLast)
+	spec := rollup.NewChainSpec(&rollupCfg)
+	ch := NewChannel(frame.ID, l1Origin)
+	require.False(t, ch.IsReady())
+	require.NoError(t, ch.AddFrame(frame, l1Origin))
+	require.True(t, ch.IsReady())
+	br, err := BatchReader(ch.Reader(), spec.MaxRLPBytesPerChannel(0), true)
+	require.NoError(t, err)
+
+	sbs := make([]*SingularBatch, 0, tt.numBatches-1)
+	for i := 0; i < tt.expNumSpanBatches; i++ {
+		t.Logf("iteration %d", i)
+		expBlocks := tt.maxBlocks
+		if i == tt.expNumSpanBatches-1 {
+			// last span batch possibly contains less
+			expBlocks = tt.expLastNumBlocks
+		}
+
+		bd, err := br()
+		require.NoError(t, err)
+		require.EqualValues(t, SpanBatchType, bd.GetBatchType())
+		sb, err := DeriveSpanBatch(bd, rollupCfg.BlockTime, rollupCfg.Genesis.L2Time, cout.spanBatch.ChainID)
+		require.NoError(t, err)
+		require.Equal(t, expBlocks, sb.GetBlockCount())
+		sbs0, err := sb.GetSingularBatches([]eth.L1BlockRef{l1Origin}, l2SafeHead)
+		require.NoError(t, err)
+		// last span batch contains one less
+		require.Len(t, sbs0, expBlocks)
+		sbs = append(sbs, sbs0...)
+	}
+
+	// batch reader should be exhausted
+	_, err = br()
+	require.ErrorIs(t, err, io.EOF)
+
+	for i, batch := range sbs {
+		batch0 := bs[i]
+		// clear the expected parent hash, as GetSingularBatches doesn't set these yet
+		// we still compare timestamps and txs, which is enough
+		batch0.ParentHash = (common.Hash{})
+		require.Equalf(t, batch0, batch, "iteration %d", i)
+	}
+}

op-node/rollup/derive/span_channel_out.go

@@ -2,7 +2,6 @@ package derive
 
 import (
 	"bytes"
-
 	"crypto/rand"
 	"fmt"
 	"io"
@@ -37,6 +36,15 @@ type SpanChannelOut struct {
 	// spanBatch is the batch being built, which immutably holds genesis timestamp and chain ID, but otherwise can be reset
 	spanBatch *SpanBatch
 
+	// maxBlocksPerSpanBatch is an optional limit on the number of blocks per span batch.
+	// If non-zero, a new span batch will be started after the current span batch has
+	// reached this maximum.
+	maxBlocksPerSpanBatch int
+
+	// sealedRLPBytes stores the sealed number of input RLP bytes. This is used when maxBlocksPerSpanBatch is non-zero
+	// to seal full span batches (that have reached the max block count) in the rlp slices.
+	sealedRLPBytes int
+
 	chainSpec *rollup.ChainSpec
 }
 
@@ -49,7 +57,15 @@ func (co *SpanChannelOut) setRandomID() error {
 	return err
 }
 
-func NewSpanChannelOut(genesisTimestamp uint64, chainID *big.Int, targetOutputSize uint64, compressionAlgo CompressionAlgo, chainSpec *rollup.ChainSpec) (*SpanChannelOut, error) {
+type SpanChannelOutOption func(co *SpanChannelOut)
+
+func WithMaxBlocksPerSpanBatch(maxBlock int) SpanChannelOutOption {
+	return func(co *SpanChannelOut) {
+		co.maxBlocksPerSpanBatch = maxBlock
+	}
+}
+
+func NewSpanChannelOut(genesisTimestamp uint64, chainID *big.Int, targetOutputSize uint64, compressionAlgo CompressionAlgo, chainSpec *rollup.ChainSpec, opts ...SpanChannelOutOption) (*SpanChannelOut, error) {
 	c := &SpanChannelOut{
 		id:        ChannelID{},
 		frame:     0,
@@ -67,6 +83,10 @@ func NewSpanChannelOut(genesisTimestamp uint64, chainID *big.Int, targetOutputSi
 		return nil, err
 	}
 
+	for _, opt := range opts {
+		opt(c)
+	}
+
 	return c, nil
 }
 
@@ -74,15 +94,20 @@ func (co *SpanChannelOut) Reset() error {
 	co.closed = false
 	co.full = nil
 	co.frame = 0
+	co.sealedRLPBytes = 0
 	co.rlp[0].Reset()
 	co.rlp[1].Reset()
 	co.lastCompressedRLPSize = 0
 	co.compressor.Reset()
-	co.spanBatch = NewSpanBatch(co.spanBatch.GenesisTimestamp, co.spanBatch.ChainID)
+	co.resetSpanBatch()
 	// setting the new randomID is the only part of the reset that can fail
 	return co.setRandomID()
 }
 
+func (co *SpanChannelOut) resetSpanBatch() {
+	co.spanBatch = NewSpanBatch(co.spanBatch.GenesisTimestamp, co.spanBatch.ChainID)
+}
+
 // activeRLP returns the active RLP buffer using the current rlpIndex
 func (co *SpanChannelOut) activeRLP() *bytes.Buffer {
 	return co.rlp[co.rlpIndex]
@@ -127,6 +152,7 @@ func (co *SpanChannelOut) AddSingularBatch(batch *SingularBatch, seqNum uint64)
 		return err
 	}
 
+	co.ensureOpenSpanBatch()
 	// update the SpanBatch with the SingularBatch
 	if err := co.spanBatch.AppendSingularBatch(batch, seqNum); err != nil {
 		return fmt.Errorf("failed to append SingularBatch to SpanBatch: %w", err)
@@ -137,11 +163,13 @@ func (co *SpanChannelOut) AddSingularBatch(batch *SingularBatch, seqNum uint64)
 		return fmt.Errorf("failed to convert SpanBatch into RawSpanBatch: %w", err)
 	}
 
-	// switch to the other buffer and reset it for new use
-	// (the RLP buffer which is being made inactive holds the RLP encoded span batch just before the new batch was added)
+	// switch to the other buffer and truncate it for new use
+	// (the RLP buffer which is being made inactive holds the RLP encoded span batch(es)
+	// just before the new batch was added)
 	co.swapRLP()
-	co.activeRLP().Reset()
-	if err = rlp.Encode(co.activeRLP(), NewBatchData(rawSpanBatch)); err != nil {
+	active := co.activeRLP()
+	active.Truncate(co.sealedRLPBytes)
+	if err = rlp.Encode(active, NewBatchData(rawSpanBatch)); err != nil {
 		return fmt.Errorf("failed to encode RawSpanBatch into bytes: %w", err)
 	}
 
@@ -150,14 +178,14 @@ func (co *SpanChannelOut) AddSingularBatch(batch *SingularBatch, seqNum uint64)
 	// of the current batch guarantees that this channel will be included in an L1 block with a timestamp well after
 	// the Fjord activation.
 	maxRLPBytesPerChannel := co.chainSpec.MaxRLPBytesPerChannel(batch.Timestamp)
-	if co.activeRLP().Len() > int(maxRLPBytesPerChannel) {
+	if active.Len() > int(maxRLPBytesPerChannel) {
 		return fmt.Errorf("could not take %d bytes as replacement of channel of %d bytes, max is %d. err: %w",
-			co.activeRLP().Len(), co.inactiveRLP().Len(), maxRLPBytesPerChannel, ErrTooManyRLPBytes)
+			active.Len(), co.inactiveRLP().Len(), maxRLPBytesPerChannel, ErrTooManyRLPBytes)
 	}
 
 	// if the compressed data *plus* the new rlp data is under the target size, return early
 	// this optimizes out cases where the compressor will obviously come in under the target size
-	rlpGrowth := co.activeRLP().Len() - co.lastCompressedRLPSize
+	rlpGrowth := active.Len() - co.lastCompressedRLPSize
 	if uint64(co.compressor.Len()+rlpGrowth) < co.target {
 		return nil
 	}
@@ -166,20 +194,24 @@ func (co *SpanChannelOut) AddSingularBatch(batch *SingularBatch, seqNum uint64)
 	if err = co.compress(); err != nil {
 		return err
 	}
-	co.lastCompressedRLPSize = co.activeRLP().Len()
 
 	// if the channel is now full, either return the compressed data, or the compressed previous data
 	if err := co.FullErr(); err != nil {
-		// if there is only one batch in the channel, it *must* be returned
-		if len(co.spanBatch.Batches) == 1 {
+		// if it's the first singular batch/block of the channel, it *must* fit in
+		if co.sealedRLPBytes == 0 && co.spanBatch.GetBlockCount() == 1 {
+			return nil
+		}
+
+		// if we just perfectly filled up the channel, also return nil to retain block
+		if uint64(co.compressor.Len()) == co.target {
 			return nil
 		}
 
 		// if there is more than one batch in the channel, we revert the last batch
 		// by switching the RLP buffer and doing a fresh compression
 		co.swapRLP()
-		if err := co.compress(); err != nil {
-			return err
+		if cerr := co.compress(); cerr != nil {
+			return cerr
 		}
 		// return the full error
 		return err
@@ -188,13 +220,36 @@ func (co *SpanChannelOut) AddSingularBatch(batch *SingularBatch, seqNum uint64)
 	return nil
 }
 
+func (co *SpanChannelOut) ensureOpenSpanBatch() {
+	if co.maxBlocksPerSpanBatch == 0 || co.spanBatch.GetBlockCount() < co.maxBlocksPerSpanBatch {
+		return
+	}
+	// we assume that the full span batch has been written to the last active rlp buffer
+	active, inactive := co.activeRLP(), co.inactiveRLP()
+	if inactive.Len() > active.Len() {
+		panic("inactive rlp unexpectedly larger")
+	}
+	co.sealedRLPBytes = active.Len()
+	// Copy active to inactive rlp buffer so both have the same sealed state
+	// and resetting by truncation works as intended.
+	inactive.Reset()
+	// err is guaranteed to always be nil
+	_, _ = inactive.Write(active.Bytes())
+	co.resetSpanBatch()
+}
+
 // compress compresses the active RLP buffer and checks if the compressed data is over the target size.
 // it resets all the compression buffers because Span Batches aren't meant to be compressed incrementally.
 func (co *SpanChannelOut) compress() error {
 	co.compressor.Reset()
-	if _, err := co.compressor.Write(co.activeRLP().Bytes()); err != nil {
+	// we write Bytes() of the active RLP to the compressor, so the active RLP's
+	// buffer is not advanced as a ReadWriter, making it possible to later use
+	// Truncate().
+	rlpBytes := co.activeRLP().Bytes()
+	if _, err := co.compressor.Write(rlpBytes); err != nil {
 		return err
 	}
+	co.lastCompressedRLPSize = len(rlpBytes)
 	if err := co.compressor.Close(); err != nil {
 		return err
 	}