op-batcher: Embed Zlib Compressor into Span Channel Out ; Compression…

… Avoidance Strategy (ethereum-optimism#10002)

* Add iterative batch building benchmark

* Embed Compressor Logic directly to Span Channel Out

* PR Comments

* Tests

* fix error handling

* remove errant comment

* PR Comments

op-batcher/batcher/channel_builder.go

@@ -78,18 +78,20 @@ 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 {
 		return nil, err
 	}
-	var spanBatch *derive.SpanBatch
+	var co derive.ChannelOut
 	if cfg.BatchType == derive.SpanBatchType {
-		spanBatch = derive.NewSpanBatch(rollupCfg.Genesis.L2Time, rollupCfg.L2ChainID)
+		co, err = derive.NewSpanChannelOut(rollupCfg.Genesis.L2Time, rollupCfg.L2ChainID, cfg.CompressorConfig.TargetOutputSize)
+	} else {
+		co, err = derive.NewSingularChannelOut(c)
 	}
-	co, err := derive.NewChannelOut(cfg.BatchType, c, spanBatch)
 	if err != nil {
-		return nil, err
+		return nil, fmt.Errorf("creating channel out: %w", err)
 	}
 
 	cb := &ChannelBuilder{
@@ -154,7 +156,7 @@ func (c *ChannelBuilder) AddBlock(block *types.Block) (*derive.L1BlockInfo, erro
 		return l1info, fmt.Errorf("converting block to batch: %w", err)
 	}
 
-	if _, err = c.co.AddSingularBatch(batch, l1info.SequenceNumber); errors.Is(err, derive.ErrTooManyRLPBytes) || errors.Is(err, derive.ErrCompressorFull) {
+	if err = c.co.AddSingularBatch(batch, l1info.SequenceNumber); errors.Is(err, derive.ErrTooManyRLPBytes) || errors.Is(err, derive.ErrCompressorFull) {
 		c.setFullErr(err)
 		return l1info, c.FullErr()
 	} else if err != nil {

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,18 +355,20 @@ 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)
 	require.NoError(t, err)
 
 	// Mock the internals of `ChannelBuilder.outputFrame`
 	// to construct a single frame
+	// the type of batch does not matter here because we are using it to construct a broken frame
 	c, err := channelConfig.CompressorConfig.NewCompressor()
 	require.NoError(t, err)
-	co, err := derive.NewChannelOut(derive.SingularBatchType, c, nil)
+	co, err := derive.NewSingularChannelOut(c)
 	require.NoError(t, err)
 	var buf bytes.Buffer
 	fn, err := co.OutputFrame(&buf, channelConfig.MaxFrameSize)

op-batcher/compressor/compressors.go

@@ -10,7 +10,6 @@ const (
 	RatioKind  = "ratio"
 	ShadowKind = "shadow"
 	NoneKind   = "none"
-	BlindKind  = "blind"
 
 	// CloseOverheadZlib is the number of final bytes a [zlib.Writer] call writes
 	// to the output buffer.
@@ -21,7 +20,6 @@ var Kinds = map[string]FactoryFunc{
 	RatioKind:  NewRatioCompressor,
 	ShadowKind: NewShadowCompressor,
 	NoneKind:   NewNonCompressor,
-	BlindKind:  NewBlindCompressor,
 }
 
 var KindKeys []string

op-e2e/actions/garbage_channel_out.go

@@ -54,7 +54,7 @@ type Writer interface {
 type ChannelOutIface interface {
 	ID() derive.ChannelID
 	Reset() error
-	AddBlock(rollupCfg *rollup.Config, block *types.Block) (uint64, error)
+	AddBlock(rollupCfg *rollup.Config, block *types.Block) error
 	ReadyBytes() int
 	Flush() error
 	Close() error
@@ -138,19 +138,19 @@ func (co *GarbageChannelOut) Reset() error {
 // error that it returns is ErrTooManyRLPBytes. If this error
 // is returned, the channel should be closed and a new one
 // should be made.
-func (co *GarbageChannelOut) AddBlock(rollupCfg *rollup.Config, block *types.Block) (uint64, error) {
+func (co *GarbageChannelOut) AddBlock(rollupCfg *rollup.Config, block *types.Block) error {
 	if co.closed {
-		return 0, errors.New("already closed")
+		return errors.New("already closed")
 	}
 	batch, err := blockToBatch(rollupCfg, block)
 	if err != nil {
-		return 0, err
+		return err
 	}
 	// We encode to a temporary buffer to determine the encoded length to
 	// ensure that the total size of all RLP elements is less than or equal to MAX_RLP_BYTES_PER_CHANNEL
 	var buf bytes.Buffer
 	if err := rlp.Encode(&buf, batch); err != nil {
-		return 0, err
+		return err
 	}
 	if co.cfg.malformRLP {
 		// Malform the RLP by incrementing the length prefix by 1.
@@ -160,13 +160,13 @@ func (co *GarbageChannelOut) AddBlock(rollupCfg *rollup.Config, block *types.Blo
 		buf.Write(bufBytes)
 	}
 	if co.rlpLength+buf.Len() > derive.MaxRLPBytesPerChannel {
-		return 0, fmt.Errorf("could not add %d bytes to channel of %d bytes, max is %d. err: %w",
+		return fmt.Errorf("could not add %d bytes to channel of %d bytes, max is %d. err: %w",
 			buf.Len(), co.rlpLength, derive.MaxRLPBytesPerChannel, derive.ErrTooManyRLPBytes)
 	}
 	co.rlpLength += buf.Len()
 
-	written, err := io.Copy(co.compress, &buf)
-	return uint64(written), err
+	_, err = io.Copy(co.compress, &buf)
+	return err
 }
 
 // ReadyBytes returns the number of bytes that the channel out can immediately output into a frame.

op-e2e/actions/l2_batcher.go

@@ -189,29 +189,28 @@ func (s *L2Batcher) Buffer(t Testing) error {
 		if s.l2BatcherCfg.GarbageCfg != nil {
 			ch, err = NewGarbageChannelOut(s.l2BatcherCfg.GarbageCfg)
 		} else {
-			c, e := compressor.NewBlindCompressor(compressor.Config{
-				TargetOutputSize: batcher.MaxDataSize(1, s.l2BatcherCfg.MaxL1TxSize),
+			target := batcher.MaxDataSize(1, s.l2BatcherCfg.MaxL1TxSize)
+			c, e := compressor.NewShadowCompressor(compressor.Config{
+				TargetOutputSize: target,
 			})
 			require.NoError(t, e, "failed to create compressor")
 
-			var batchType uint = derive.SingularBatchType
-			var spanBatch *derive.SpanBatch
-
 			if s.l2BatcherCfg.ForceSubmitSingularBatch && s.l2BatcherCfg.ForceSubmitSpanBatch {
 				t.Fatalf("ForceSubmitSingularBatch and ForceSubmitSpanBatch cannot be set to true at the same time")
-			} else if s.l2BatcherCfg.ForceSubmitSingularBatch {
-				// use SingularBatchType
-			} 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.
-				batchType = derive.SpanBatchType
-				spanBatch = derive.NewSpanBatch(s.rollupCfg.Genesis.L2Time, s.rollupCfg.L2ChainID)
+			} else {
+				// use span batch if we're forcing it or if we're at/beyond delta
+				if s.l2BatcherCfg.ForceSubmitSpanBatch || s.rollupCfg.IsDelta(block.Time()) {
+					ch, err = derive.NewSpanChannelOut(s.rollupCfg.Genesis.L2Time, s.rollupCfg.L2ChainID, target)
+					// use singular batches in all other cases
+				} else {
+					ch, err = derive.NewSingularChannelOut(c)
+				}
 			}
-			ch, err = derive.NewChannelOut(batchType, c, spanBatch)
 		}
 		require.NoError(t, err, "failed to create channel")
 		s.l2ChannelOut = ch
 	}
-	if _, err := s.l2ChannelOut.AddBlock(s.rollupCfg, block); err != nil { // should always succeed
+	if err := s.l2ChannelOut.AddBlock(s.rollupCfg, block); err != nil {
 		return err
 	}
 	ref, err := s.engCl.L2BlockRefByHash(t.Ctx(), block.Hash())

op-e2e/actions/sync_test.go

@@ -7,7 +7,6 @@ import (
 	"testing"
 	"time"
 
-	"github.com/ethereum-optimism/optimism/op-batcher/compressor"
 	"github.com/ethereum-optimism/optimism/op-e2e/e2eutils"
 	"github.com/ethereum-optimism/optimism/op-node/rollup/derive"
 	"github.com/ethereum-optimism/optimism/op-node/rollup/sync"
@@ -27,12 +26,7 @@ import (
 )
 
 func newSpanChannelOut(t StatefulTesting, e e2eutils.SetupData) derive.ChannelOut {
-	c, err := compressor.NewBlindCompressor(compressor.Config{
-		TargetOutputSize: 128_000,
-	})
-	require.NoError(t, err)
-	spanBatch := derive.NewSpanBatch(e.RollupCfg.Genesis.L2Time, e.RollupCfg.L2ChainID)
-	channelOut, err := derive.NewChannelOut(derive.SpanBatchType, c, spanBatch)
+	channelOut, err := derive.NewSpanChannelOut(e.RollupCfg.Genesis.L2Time, e.RollupCfg.L2ChainID, 128_000)
 	require.NoError(t, err)
 	return channelOut
 }
@@ -249,7 +243,7 @@ func TestBackupUnsafe(gt *testing.T) {
 			block = block.WithBody([]*types.Transaction{block.Transactions()[0], invalidTx}, []*types.Header{})
 		}
 		// Add A1, B2, B3, B4, B5 into the channel
-		_, err = channelOut.AddBlock(sd.RollupCfg, block)
+		err = channelOut.AddBlock(sd.RollupCfg, block)
 		require.NoError(t, err)
 	}
 
@@ -412,7 +406,7 @@ func TestBackupUnsafeReorgForkChoiceInputError(gt *testing.T) {
 			block = block.WithBody([]*types.Transaction{block.Transactions()[0], invalidTx}, []*types.Header{})
 		}
 		// Add A1, B2, B3, B4, B5 into the channel
-		_, err = channelOut.AddBlock(sd.RollupCfg, block)
+		err = channelOut.AddBlock(sd.RollupCfg, block)
 		require.NoError(t, err)
 	}
 
@@ -551,7 +545,7 @@ func TestBackupUnsafeReorgForkChoiceNotInputError(gt *testing.T) {
 			block = block.WithBody([]*types.Transaction{block.Transactions()[0], invalidTx}, []*types.Header{})
 		}
 		// Add A1, B2, B3, B4, B5 into the channel
-		_, err = channelOut.AddBlock(sd.RollupCfg, block)
+		err = channelOut.AddBlock(sd.RollupCfg, block)
 		require.NoError(t, err)
 	}
 
@@ -870,7 +864,7 @@ func TestInvalidPayloadInSpanBatch(gt *testing.T) {
 			block = block.WithBody([]*types.Transaction{block.Transactions()[0], invalidTx}, []*types.Header{})
 		}
 		// Add A1 ~ A12 into the channel
-		_, err = channelOut.AddBlock(sd.RollupCfg, block)
+		err = channelOut.AddBlock(sd.RollupCfg, block)
 		require.NoError(t, err)
 	}
 
@@ -919,7 +913,7 @@ func TestInvalidPayloadInSpanBatch(gt *testing.T) {
 			block = block.WithBody([]*types.Transaction{block.Transactions()[0], tx}, []*types.Header{})
 		}
 		// Add B1, A2 ~ A12 into the channel
-		_, err = channelOut.AddBlock(sd.RollupCfg, block)
+		err = channelOut.AddBlock(sd.RollupCfg, block)
 		require.NoError(t, err)
 	}
 	// Submit span batch(B1, A2, ... A12)