Tests

Author: axelKingsley

op-batcher/batcher/channel_builder.go

@@ -78,6 +78,7 @@ type ChannelBuilder struct {
 
 // newChannelBuilder creates a new channel builder or returns an error if the
 // channel out could not be created.
+// it acts as a factory for either a span or singular channel out
 func NewChannelBuilder(cfg ChannelConfig, rollupCfg rollup.Config, latestL1OriginBlockNum uint64) (*ChannelBuilder, error) {
 	c, err := cfg.CompressorConfig.NewCompressor()
 	if err != nil {
@@ -90,7 +91,7 @@ func NewChannelBuilder(cfg ChannelConfig, rollupCfg rollup.Config, latestL1Origi
 			return nil, err
 		}
 	} else {
-		co, err = derive.NewChannelOut(c)
+		co, err = derive.NewSingularChannelOut(c)
 		if err != nil {
 			return nil, err
 		}

op-batcher/batcher/channel_builder_test.go

@@ -297,6 +297,7 @@ func TestChannelBuilderBatchType(t *testing.T) {
 		{"ChannelBuilder_PendingFrames_TotalFrames", ChannelBuilder_PendingFrames_TotalFrames},
 		{"ChannelBuilder_InputBytes", ChannelBuilder_InputBytes},
 		{"ChannelBuilder_OutputBytes", ChannelBuilder_OutputBytes},
+		{"ChannelBuilder_OutputWrongFramePanic", ChannelBuilder_OutputWrongFramePanic},
 	}
 	for _, test := range tests {
 		test := test
@@ -354,8 +355,9 @@ func TestChannelBuilder_NextFrame(t *testing.T) {
 }
 
 // TestChannelBuilder_OutputWrongFramePanic tests that a panic is thrown when a frame is pushed with an invalid frame id
-func TestChannelBuilder_OutputWrongFramePanic(t *testing.T) {
+func ChannelBuilder_OutputWrongFramePanic(t *testing.T, batchType uint) {
 	channelConfig := defaultTestChannelConfig()
+	channelConfig.BatchType = batchType
 
 	// Construct a channel builder
 	cb, err := NewChannelBuilder(channelConfig, defaultTestRollupConfig, latestL1BlockOrigin)
@@ -365,7 +367,9 @@ func TestChannelBuilder_OutputWrongFramePanic(t *testing.T) {
 	// to construct a single frame
 	c, err := channelConfig.CompressorConfig.NewCompressor()
 	require.NoError(t, err)
-	co, err := derive.NewChannelOut(c)
+	// co is only created to supply a random ID
+	co, err := derive.NewSingularChannelOut(c)
+
 	require.NoError(t, err)
 	var buf bytes.Buffer
 	fn, err := co.OutputFrame(&buf, channelConfig.MaxFrameSize)

op-e2e/actions/l2_batcher.go

@@ -199,7 +199,7 @@ func (s *L2Batcher) Buffer(t Testing) error {
 				t.Fatalf("ForceSubmitSingularBatch and ForceSubmitSpanBatch cannot be set to true at the same time")
 			} else if s.l2BatcherCfg.ForceSubmitSingularBatch {
 				// use SingularBatchType
-				ch, err = derive.NewChannelOut(c)
+				ch, err = derive.NewSingularChannelOut(c)
 			} else if s.l2BatcherCfg.ForceSubmitSpanBatch || s.rollupCfg.IsDelta(block.Time()) {
 				// If both ForceSubmitSingularBatch and ForceSubmitSpanbatch are false, use SpanBatch automatically if Delta HF is activated.
 				ch, err = derive.NewSpanChannelOut(s.rollupCfg.Genesis.L2Time, s.rollupCfg.L2ChainID, target)

op-e2e/testoutput.txt

@@ -59,7 +59,7 @@ type compressorAndTarget struct {
 func channelOutByType(batchType uint, compKey string) (derive.ChannelOut, error) {
 	chainID := big.NewInt(333)
 	if batchType == derive.SingularBatchType {
-		return derive.NewChannelOut(compressors[compKey].compressor)
+		return derive.NewSingularChannelOut(compressors[compKey].compressor)
 	}
 	if batchType == derive.SpanBatchType {
 		return derive.NewSpanChannelOut(0, chainID, compressors[compKey].targetOutput)

op-node/benchmarks/batchbuilding_test.go

@@ -63,10 +63,6 @@ type ChannelOut interface {
 	OutputFrame(*bytes.Buffer, uint64) (uint16, error)
 }
 
-func NewChannelOut(compress Compressor) (ChannelOut, error) {
-	return NewSingularChannelOut(compress)
-}
-
 type SingularChannelOut struct {
 	id ChannelID
 	// Frame ID of the next frame to emit. Increment after emitting

op-node/rollup/derive/channel_out.go

@@ -2,6 +2,7 @@ package derive
 
 import (
 	"bytes"
+	"fmt"
 	"io"
 	"math/big"
 	"math/rand"
@@ -36,63 +37,93 @@ func (s *nonCompressor) FullErr() error {
 	return nil
 }
 
-func (t *nonCompressor) TargetOutputSize() uint64 {
-	return 0
+// channelTypes allows tests to run against different channel types
+var channelTypes = []struct {
+	ChannelOut func(t *testing.T) ChannelOut
+	Name       string
+}{
+	{
+		Name: "Singular",
+		ChannelOut: func(t *testing.T) ChannelOut {
+			cout, err := NewSingularChannelOut(&nonCompressor{})
+			require.NoError(t, err)
+			return cout
+		},
+	},
+	{
+		Name: "Span",
+		ChannelOut: func(t *testing.T) ChannelOut {
+			cout, err := NewSpanChannelOut(0, big.NewInt(0), 128_000)
+			require.NoError(t, err)
+			return cout
+		},
+	},
 }
 
 func TestChannelOutAddBlock(t *testing.T) {
-	cout, err := NewChannelOut(&nonCompressor{})
-	require.NoError(t, err)
-
-	t.Run("returns err if first tx is not an l1info tx", func(t *testing.T) {
-		header := &types.Header{Number: big.NewInt(1), Difficulty: big.NewInt(100)}
-		block := types.NewBlockWithHeader(header).WithBody(
-			[]*types.Transaction{
-				types.NewTx(&types.DynamicFeeTx{}),
-			},
-			nil,
-		)
-		_, err := cout.AddBlock(&rollupCfg, block)
-		require.Error(t, err)
-		require.Equal(t, ErrNotDepositTx, err)
-	})
+	for _, tcase := range channelTypes {
+		t.Run(fmt.Sprintf("%s - returns err if first tx is not an l1info tx", tcase.Name), func(t *testing.T) {
+			cout := tcase.ChannelOut(t)
+			header := &types.Header{Number: big.NewInt(1), Difficulty: big.NewInt(100)}
+			block := types.NewBlockWithHeader(header).WithBody(
+				[]*types.Transaction{
+					types.NewTx(&types.DynamicFeeTx{}),
+				},
+				nil,
+			)
+			_, err := cout.AddBlock(&rollupCfg, block)
+			require.Error(t, err)
+			require.Equal(t, ErrNotDepositTx, err)
+		})
+	}
 }
 
 // TestOutputFrameSmallMaxSize tests that calling [OutputFrame] with a small
 // max size that is below the fixed frame size overhead of 23, will return
 // an error.
 func TestOutputFrameSmallMaxSize(t *testing.T) {
-	cout, err := NewChannelOut(&nonCompressor{})
-	require.NoError(t, err)
-
-	// Call OutputFrame with the range of small max size values that err
-	var w bytes.Buffer
-	for i := 0; i < 23; i++ {
-		fid, err := cout.OutputFrame(&w, uint64(i))
-		require.ErrorIs(t, err, ErrMaxFrameSizeTooSmall)
-		require.Zero(t, fid)
+	for _, tcase := range channelTypes {
+		t.Run(fmt.Sprintf("%s - returns ErrMaxFrameSizeTooSmall", tcase.Name), func(t *testing.T) {
+			cout := tcase.ChannelOut(t)
+			// Call OutputFrame with the range of small max size values that err
+			var w bytes.Buffer
+			for i := 0; i < 23; i++ {
+				fid, err := cout.OutputFrame(&w, uint64(i))
+				require.ErrorIs(t, err, ErrMaxFrameSizeTooSmall)
+				require.Zero(t, fid)
+			}
+		})
 	}
 }
 
 func TestOutputFrameNoEmptyLastFrame(t *testing.T) {
-	cout, err := NewChannelOut(&nonCompressor{})
-	require.NoError(t, err)
+	for _, tcase := range channelTypes {
+		t.Run(fmt.Sprintf("%s - no empty last frame", tcase.Name), func(t *testing.T) {
+			cout := tcase.ChannelOut(t)
 
-	rng := rand.New(rand.NewSource(0x543331))
-	chainID := big.NewInt(rng.Int63n(1000))
-	txCount := 1
-	singularBatch := RandomSingularBatch(rng, txCount, chainID)
+			rng := rand.New(rand.NewSource(0x543331))
+			chainID := big.NewInt(0)
+			txCount := 1
+			singularBatch := RandomSingularBatch(rng, txCount, chainID)
 
-	written, err := cout.AddSingularBatch(singularBatch, 0)
-	require.NoError(t, err)
+			written, err := cout.AddSingularBatch(singularBatch, 0)
+			require.NoError(t, err)
 
-	require.NoError(t, cout.Close())
-
-	var buf bytes.Buffer
-	// Output a frame which needs exactly `written` bytes. This frame is expected to be the last frame.
-	_, err = cout.OutputFrame(&buf, written+FrameV0OverHeadSize)
-	require.ErrorIs(t, err, io.EOF)
+			require.NoError(t, cout.Close())
+			// span batches return the length of the RLP structure, not the compressed length
+			// so we need to collect the compressed length from the span batch
+			// no production code relies on the written value from either type of batch
+			// (span batches don't always compress their data, and can't always determine the compressed length for this reason)
+			if span, ok := cout.(*SpanChannelOut); ok {
+				written = uint64(span.compressed.Len())
+			}
 
+			var buf bytes.Buffer
+			// Output a frame which needs exactly `written` bytes. This frame is expected to be the last frame.
+			_, err = cout.OutputFrame(&buf, written+FrameV0OverHeadSize)
+			require.ErrorIs(t, err, io.EOF)
+		})
+	}
 }
 
 // TestRLPByteLimit ensures that stream encoder is properly limiting the length.
@@ -188,3 +219,94 @@ func TestBlockToBatchValidity(t *testing.T) {
 	_, _, err := BlockToSingularBatch(&rollupCfg, block)
 	require.ErrorContains(t, err, "has no transactions")
 }
+
+// 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 TestSpanChannelOutCompressionOnlyOneBatch(t *testing.T) {
+	// target is larger than one batch, but smaller than two batches
+	target := uint64(300)
+	rng := rand.New(rand.NewSource(0x543331))
+	chainID := big.NewInt(rng.Int63n(1000))
+	txCount := 1
+	cout, err := NewSpanChannelOut(0, chainID, target)
+	require.NoError(t, err)
+
+	compressedLens := []int{}
+
+	// adding the first batch should not cause an error
+	singularBatch := RandomSingularBatch(rng, txCount, chainID)
+	_, err = cout.AddSingularBatch(singularBatch, 0)
+	compressedLens = append(compressedLens, cout.compressed.Len())
+	require.NoError(t, err)
+
+	// confirm compression was not skipped
+	require.Greater(t, compressedLens[0], 0)
+	// confirm the channel is full
+	require.ErrorIs(t, cout.FullErr(), ErrCompressorFull)
+}
+
+// TestSpanChannelOutCompressionUndo tests that the SpanChannelOut compression works as expected to not accept a second batch
+// if that batch would cause the channel to exceed the target size. The first batch should be compressed only
+// the channel should be compressed, full, and should not have the second batch
+func TestSpanChannelOutCompressionUndo(t *testing.T) {
+	// target is larger than one batch, but smaller than two batches
+	target := uint64(750)
+	rng := rand.New(rand.NewSource(0x543331))
+	chainID := big.NewInt(rng.Int63n(1000))
+	txCount := 1
+	cout, err := NewSpanChannelOut(0, chainID, target)
+	require.NoError(t, err)
+
+	compressedLens := []int{}
+	rlpLens := []int{}
+
+	// adding the first batch should not cause an error
+	singularBatch := RandomSingularBatch(rng, txCount, chainID)
+	_, err = cout.AddSingularBatch(singularBatch, 0)
+	compressedLens = append(compressedLens, cout.compressed.Len())
+	rlpLens = append(rlpLens, cout.activeRLP().Len())
+	require.NoError(t, err)
+
+	// adding the second batch should cause an error
+	singularBatch = RandomSingularBatch(rng, txCount, chainID)
+	_, err = cout.AddSingularBatch(singularBatch, 0)
+	compressedLens = append(compressedLens, cout.compressed.Len())
+	rlpLens = append(rlpLens, cout.activeRLP().Len())
+	require.ErrorIs(t, err, ErrCompressorFull)
+
+	// confirm that the first compression was skipped
+	require.Equal(t, 0, compressedLens[0])
+	// confirm that the second compression was not skipped
+	require.Greater(t, compressedLens[1], 0)
+	// confirm that the second rlp is tht same size as the first (because the second batch was not added)
+	require.Equal(t, rlpLens[0], rlpLens[1])
+}
+
+// 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 TestSpanChannelOutClose(t *testing.T) {
+	// target is larger than one batch
+	target := uint64(600)
+	rng := rand.New(rand.NewSource(0x543331))
+	chainID := big.NewInt(rng.Int63n(1000))
+	txCount := 1
+	cout, err := NewSpanChannelOut(0, chainID, target)
+	require.NoError(t, err)
+
+	singularBatch := RandomSingularBatch(rng, txCount, chainID)
+	_, err = cout.AddSingularBatch(singularBatch, 0)
+	require.NoError(t, err)
+
+	rlpLen := cout.activeRLP().Len()
+
+	// confirm no compression has happened yet
+	require.Equal(t, 0, cout.compressed.Len())
+	require.Less(t, uint64(rlpLen), target)
+
+	// close the channel
+	require.NoError(t, cout.Close())
+
+	// confirm that the only batch was compressed, and that the RLP did not change
+	require.Greater(t, cout.compressed.Len(), 0)
+	require.Equal(t, rlpLen, cout.activeRLP().Len())
+}

op-node/rollup/derive/channel_out_test.go

@@ -115,8 +115,8 @@ func (co *SpanChannelOut) AddBlock(rollupCfg *rollup.Config, block *types.Block)
 // and the compression happens on the previous RLP buffer instead
 // if the input is too small to need compression, data is accumulated but not compressed
 func (co *SpanChannelOut) AddSingularBatch(batch *SingularBatch, seqNum uint64) (uint64, error) {
-	// sentinel error for closed channel
-	if co.closed {
+	// sentinel error for closed or full channel
+	if co.closed || co.FullErr() != nil {
 		return 0, ErrChannelOutAlreadyClosed
 	}
 
@@ -204,9 +204,10 @@ func (co *SpanChannelOut) ReadyBytes() int {
 	return 0
 }
 
-// Flush implements the Channel Out interface by passing the flush call to the compressor
+// Flush implements the Channel Out
+// Span Channel Out manages the flushing of the compressor internally, so this is a no-op
 func (co *SpanChannelOut) Flush() error {
-	return co.compressor.Flush()
+	return nil
 }
 
 // checkFull sets the full error if the compressed data is over the target size.
@@ -230,10 +231,14 @@ func (co *SpanChannelOut) Close() error {
 		return ErrChannelOutAlreadyClosed
 	}
 	co.closed = true
-	if err := co.Flush(); err != nil {
-		return err
+	// if the channel was already full,
+	// the compressor is already flushed and closed
+	if co.FullErr() != nil {
+		return nil
 	}
-	return co.compressor.Close()
+	// if this channel is not full, we need to compress the last batch
+	// this also flushes/closes the compressor
+	return co.compress()
 }
 
 // OutputFrame writes a frame to w with a given max size and returns the frame