Create an abstraction for channel compression

op-batcher/batcher/channel_builder.go

@@ -13,7 +13,6 @@ import (
 
 var (
 	ErrInvalidChannelTimeout = errors.New("channel timeout is less than the safety margin")
-	ErrInputTargetReached    = errors.New("target amount of input data reached")
 	ErrMaxFrameIndex         = errors.New("max frame index reached (uint16)")
 	ErrMaxDurationReached    = errors.New("max channel duration reached")
 	ErrChannelTimeoutClose   = errors.New("close to channel timeout")
@@ -55,19 +54,8 @@ type ChannelConfig struct {
 	SubSafetyMargin uint64
 	// The maximum byte-size a frame can have.
 	MaxFrameSize uint64
-	// The target number of frames to create per channel. Note that if the
-	// realized compression ratio is worse than the approximate, more frames may
-	// actually be created. This also depends on how close TargetFrameSize is to
-	// MaxFrameSize.
-	TargetFrameSize uint64
-	// The target number of frames to create in this channel. If the realized
-	// compression ratio is worse than approxComprRatio, additional leftover
-	// frame(s) might get created.
-	TargetNumFrames int
-	// Approximated compression ratio to assume. Should be slightly smaller than
-	// average from experiments to avoid the chances of creating a small
-	// additional leftover frame.
-	ApproxComprRatio float64
+	// Compressor to use to compress frame data.
+	Compressor derive.Compressor
 }
 
 // Check validates the [ChannelConfig] parameters.
@@ -93,13 +81,12 @@ func (cc *ChannelConfig) Check() error {
 		return fmt.Errorf("max frame size %d is less than the minimum 23", cc.MaxFrameSize)
 	}
 
-	return nil
-}
+	// Compressor must be set
+	if cc.Compressor == nil {
+		return errors.New("compressor cannot be nil")
+	}
 
-// InputThreshold calculates the input data threshold in bytes from the given
-// parameters.
-func (c ChannelConfig) InputThreshold() uint64 {
-	return uint64(float64(c.TargetNumFrames) * float64(c.TargetFrameSize) / c.ApproxComprRatio)
+	return nil
 }
 
 type frameID struct {
@@ -142,7 +129,7 @@ type channelBuilder struct {
 // newChannelBuilder creates a new channel builder or returns an error if the
 // channel out could not be created.
 func newChannelBuilder(cfg ChannelConfig) (*channelBuilder, error) {
-	co, err := derive.NewChannelOut()
+	co, err := derive.NewChannelOut(cfg.Compressor)
 	if err != nil {
 		return nil, err
 	}
@@ -218,8 +205,8 @@ func (c *channelBuilder) AddBlock(block *types.Block) (derive.L1BlockInfo, error
 	c.blocks = append(c.blocks, block)
 	c.updateSwTimeout(batch)
 
-	if c.inputTargetReached() {
-		c.setFullErr(ErrInputTargetReached)
+	if err = c.cfg.Compressor.FullErr(); err != nil {
+		c.setFullErr(err)
 		// Adding this block still worked, so don't return error, just mark as full
 	}
 
@@ -293,12 +280,6 @@ func (c *channelBuilder) TimedOut(blockNum uint64) bool {
 	return c.timeout != 0 && blockNum >= c.timeout
 }
 
-// inputTargetReached says whether the target amount of input data has been
-// reached in this channel builder. No more blocks can be added afterwards.
-func (c *channelBuilder) inputTargetReached() bool {
-	return uint64(c.co.InputBytes()) >= c.cfg.InputThreshold()
-}
-
 // IsFull returns whether the channel is full.
 // FullErr returns the reason for the channel being full.
 func (c *channelBuilder) IsFull() bool {
@@ -310,7 +291,7 @@ func (c *channelBuilder) IsFull() bool {
 //
 // It returns a ChannelFullError wrapping one of the following possible reasons
 // for the channel being full:
-//   - ErrInputTargetReached if the target amount of input data has been reached,
+//   - derive.CompressorFullErr if the compressor target has been reached.
 //   - derive.MaxRLPBytesPerChannel if the general maximum amount of input data
 //     would have been exceeded by the latest AddBlock call,
 //   - ErrMaxFrameIndex if the maximum number of frames has been generated

op-batcher/batcher/channel_manager_test.go

@@ -98,9 +98,8 @@ func TestChannelManagerReturnsErrReorgWhenDrained(t *testing.T) {
 	log := testlog.Logger(t, log.LvlCrit)
 	m := NewChannelManager(log, metrics.NoopMetrics,
 		ChannelConfig{
-			TargetFrameSize:  0,
 			MaxFrameSize: 120_000,
-			ApproxComprRatio: 1.0,
+			Compressor:   newCompressor(t, 1, 1, 1),
 		})
 
 	a := newMiniL2Block(0)
@@ -171,8 +170,7 @@ func TestChannelManager_Clear(t *testing.T) {
 		// Have to set the max frame size here otherwise the channel builder would not
 		// be able to output any frames
 		MaxFrameSize: 24,
-		TargetFrameSize:  24,
-		ApproxComprRatio: 1.0,
+		Compressor:   newCompressor(t, 24, 1, 1),
 	})
 
 	// Channel Manager state should be empty by default
@@ -331,9 +329,8 @@ func TestChannelManager_TxResend(t *testing.T) {
 	log := testlog.Logger(t, log.LvlError)
 	m := NewChannelManager(log, metrics.NoopMetrics,
 		ChannelConfig{
-			TargetFrameSize:  0,
 			MaxFrameSize: 120_000,
-			ApproxComprRatio: 1.0,
+			Compressor:   newCompressor(t, 1, 1, 1),
 		})
 
 	a, _ := derivetest.RandomL2Block(rng, 4)
@@ -372,9 +369,8 @@ func TestChannelManagerCloseBeforeFirstUse(t *testing.T) {
 	log := testlog.Logger(t, log.LvlCrit)
 	m := NewChannelManager(log, metrics.NoopMetrics,
 		ChannelConfig{
-			TargetFrameSize:  0,
 			MaxFrameSize:   100,
-			ApproxComprRatio: 1.0,
+			Compressor:     newCompressor(t, 0, 1, 1),
 			ChannelTimeout: 1000,
 		})
 
@@ -397,9 +393,8 @@ func TestChannelManagerCloseNoPendingChannel(t *testing.T) {
 	log := testlog.Logger(t, log.LvlCrit)
 	m := NewChannelManager(log, metrics.NoopMetrics,
 		ChannelConfig{
-			TargetFrameSize:  0,
-			MaxFrameSize:     100,
-			ApproxComprRatio: 1.0,
+			MaxFrameSize:   1000,
+			Compressor:     newCompressor(t, 1, 1, 1),
 			ChannelTimeout: 1000,
 		})
 	a := newMiniL2Block(0)
@@ -433,10 +428,8 @@ func TestChannelManagerClosePendingChannel(t *testing.T) {
 	log := testlog.Logger(t, log.LvlCrit)
 	m := NewChannelManager(log, metrics.NoopMetrics,
 		ChannelConfig{
-			TargetNumFrames:  100,
-			TargetFrameSize:  1000,
 			MaxFrameSize:   1000,
-			ApproxComprRatio: 1.0,
+			Compressor:     newCompressor(t, 1000, 100, 1),
 			ChannelTimeout: 1000,
 		})
 
@@ -476,10 +469,8 @@ func TestChannelManagerCloseAllTxsFailed(t *testing.T) {
 	log := testlog.Logger(t, log.LvlCrit)
 	m := NewChannelManager(log, metrics.NoopMetrics,
 		ChannelConfig{
-			TargetNumFrames:  100,
-			TargetFrameSize:  1000,
 			MaxFrameSize:   1000,
-			ApproxComprRatio: 1.0,
+			Compressor:     newCompressor(t, 1000, 100, 1),
 			ChannelTimeout: 1000,
 		})
 

op-batcher/batcher/driver.go

@@ -75,6 +75,15 @@ func NewBatchSubmitterFromCLIConfig(cfg CLIConfig, l log.Logger, m metrics.Metri
 		return nil, err
 	}
 
+	compressor, err := NewTargetSizeCompressor(
+		cfg.TargetL1TxSize-1, // subtract 1 byte for version
+		cfg.TargetNumFrames,
+		cfg.ApproxComprRatio,
+	)
+	if err != nil {
+		return nil, err
+	}
+
 	batcherCfg := Config{
 		L1Client:               l1Client,
 		L2Client:               l2Client,
@@ -90,9 +99,7 @@ func NewBatchSubmitterFromCLIConfig(cfg CLIConfig, l log.Logger, m metrics.Metri
 			MaxChannelDuration: cfg.MaxChannelDuration,
 			SubSafetyMargin:    cfg.SubSafetyMargin,
 			MaxFrameSize:       cfg.MaxL1TxSize - 1, // subtract 1 byte for version
-			TargetFrameSize:    cfg.TargetL1TxSize - 1, // subtract 1 byte for version
-			TargetNumFrames:    cfg.TargetNumFrames,
-			ApproxComprRatio:   cfg.ApproxComprRatio,
+			Compressor:         compressor,
 		},
 	}
 

op-batcher/batcher/target_size_compressor.go

+package batcher
+
+import (
+	"bytes"
+	"compress/zlib"
+	"github.com/ethereum-optimism/optimism/op-node/rollup/derive"
+)
+
+type TargetSizeCompressor struct {
+	// The target number of frames to create per channel. Note that if the
+	// realized compression ratio is worse than the approximate, more frames may
+	// actually be created. This also depends on how close TargetFrameSize is to
+	// MaxFrameSize.
+	TargetFrameSize uint64
+	// The target number of frames to create in this channel. If the realized
+	// compression ratio is worse than approxComprRatio, additional leftover
+	// frame(s) might get created.
+	TargetNumFrames int
+	// Approximated compression ratio to assume. Should be slightly smaller than
+	// average from experiments to avoid the chances of creating a small
+	// additional leftover frame.
+	ApproxComprRatio float64
+
+	inputBytes int
+	buf        bytes.Buffer
+	compress   *zlib.Writer
+}
+
+func NewTargetSizeCompressor(targetFrameSize uint64, targetNumFrames int, approxCompRatio float64) (derive.Compressor, error) {
+	c := &TargetSizeCompressor{
+		TargetFrameSize:  targetFrameSize,
+		TargetNumFrames:  targetNumFrames,
+		ApproxComprRatio: approxCompRatio,
+	}
+
+	compress, err := zlib.NewWriterLevel(&c.buf, zlib.BestCompression)
+	if err != nil {
+		return nil, err
+	}
+	c.compress = compress
+
+	return c, nil
+}
+
+func (t *TargetSizeCompressor) Write(p []byte) (int, error) {
+	if err := t.FullErr(); err != nil {
+		return 0, err
+	}
+	t.inputBytes += len(p)
+	return t.compress.Write(p)
+}
+
+func (t *TargetSizeCompressor) Close() error {
+	return t.compress.Close()
+}
+
+func (t *TargetSizeCompressor) Read(p []byte) (int, error) {
+	return t.buf.Read(p)
+}
+
+func (t *TargetSizeCompressor) Reset() {
+	t.buf.Reset()
+	t.compress.Reset(&t.buf)
+	t.inputBytes = 0
+}
+
+func (t *TargetSizeCompressor) Len() int {
+	return t.buf.Len()
+}
+
+func (t *TargetSizeCompressor) Flush() error {
+	return t.compress.Flush()
+}
+
+func (t *TargetSizeCompressor) FullErr() error {
+	if t.inputTargetReached() {
+		return derive.CompressorFullErr
+	}
+	return nil
+}
+
+// InputThreshold calculates the input data threshold in bytes from the given
+// parameters.
+func (t *TargetSizeCompressor) InputThreshold() uint64 {
+	return uint64(float64(t.TargetNumFrames) * float64(t.TargetFrameSize) / t.ApproxComprRatio)
+}
+
+// inputTargetReached says whether the target amount of input data has been
+// reached in this channel builder. No more blocks can be added afterwards.
+func (t *TargetSizeCompressor) inputTargetReached() bool {
+	return uint64(t.inputBytes) >= t.InputThreshold()
+}

op-batcher/batcher/channel_config_test.go → op-batcher/batcher/target_size_compressor_test.go

@@ -118,12 +118,13 @@ func TestInputThreshold(t *testing.T) {
 
 	// Validate each test case
 	for _, tt := range tests {
-		config := batcher.ChannelConfig{
-			TargetFrameSize:  tt.input.TargetFrameSize,
-			TargetNumFrames:  tt.input.TargetNumFrames,
-			ApproxComprRatio: tt.input.ApproxComprRatio,
-		}
-		got := config.InputThreshold()
+		compressor, err := batcher.NewTargetSizeCompressor(
+			tt.input.TargetFrameSize,
+			tt.input.TargetNumFrames,
+			tt.input.ApproxComprRatio,
+		)
+		require.NoError(t, err)
+		got := compressor.(*batcher.TargetSizeCompressor).InputThreshold()
 		tt.assertion(got)
 	}
 }

op-e2e/actions/l2_batcher.go

@@ -16,6 +16,7 @@ import (
 	"github.com/ethereum/go-ethereum/params"
 	"github.com/stretchr/testify/require"
 
+	"github.com/ethereum-optimism/optimism/op-batcher/batcher"
 	"github.com/ethereum-optimism/optimism/op-node/eth"
 	"github.com/ethereum-optimism/optimism/op-node/rollup"
 	"github.com/ethereum-optimism/optimism/op-node/rollup/derive"
@@ -133,7 +134,9 @@ func (s *L2Batcher) Buffer(t Testing) error {
 		if s.l2BatcherCfg.GarbageCfg != nil {
 			ch, err = NewGarbageChannelOut(s.l2BatcherCfg.GarbageCfg)
 		} else {
-			ch, err = derive.NewChannelOut()
+			c, e := batcher.NewTargetSizeCompressor(s.l2BatcherCfg.MaxL1TxSize, 1, 1)
+			require.NoError(t, e, "failed to create compressor")
+			ch, err = derive.NewChannelOut(c)
 		}
 		require.NoError(t, err, "failed to create channel")
 		s.l2ChannelOut = ch

op-node/rollup/derive/channel_out.go

@@ -2,7 +2,6 @@ package derive
 
 import (
 	"bytes"
-	"compress/zlib"
 	"crypto/rand"
 	"errors"
 	"fmt"
@@ -25,6 +24,30 @@ var ErrTooManyRLPBytes = errors.New("batch would cause RLP bytes to go over limi
 // [Frame Format]: https://github.com/ethereum-optimism/optimism/blob/develop/specs/derivation.md#frame-format
 const FrameV0OverHeadSize = 23
 
+var CompressorFullErr = errors.New("compressor is full")
+
+type Compressor interface {
+	// Writer is used to write uncompressed data which will be compressed. Should return
+	// CompressorFullErr if the compressor is full and no more data should be written.
+	io.Writer
+	// Closer Close function should be called before reading any data.
+	io.Closer
+	// Reader is used to Read compressed data; should only be called after Close.
+	io.Reader
+	// Reset will reset all written data
+	Reset()
+	// Len returns an estimate of the current length of the compressed data; calling Flush will
+	// increase the accuracy at the expense of a poorer compression ratio.
+	Len() int
+	// Flush flushes any uncompressed data to the compression buffer. This will result in a
+	// non-optimal compression ratio.
+	Flush() error
+	// FullErr returns CompressorFullErr if the compressor is known to be full. Note that
+	// calls to Write will fail if an error is returned from this method, but calls to Write
+	// can still return CompressorFullErr even if this does not.
+	FullErr() error
+}
+
 type ChannelOut struct {
 	id ChannelID
 	// Frame ID of the next frame to emit. Increment after emitting
@@ -33,9 +56,7 @@ type ChannelOut struct {
 	rlpLength int
 
 	// Compressor stage. Write input data to it
-	compress *zlib.Writer
-	// post compression buffer
-	buf bytes.Buffer
+	compress Compressor
 
 	closed bool
 }
@@ -44,23 +65,18 @@ func (co *ChannelOut) ID() ChannelID {
 	return co.id
 }
 
-func NewChannelOut() (*ChannelOut, error) {
+func NewChannelOut(compress Compressor) (*ChannelOut, error) {
 	c := &ChannelOut{
 		id:        ChannelID{}, // TODO: use GUID here instead of fully random data
 		frame:     0,
 		rlpLength: 0,
+		compress:  compress,
 	}
 	_, err := rand.Read(c.id[:])
 	if err != nil {
 		return nil, err
 	}
 
-	compress, err := zlib.NewWriterLevel(&c.buf, zlib.BestCompression)
-	if err != nil {
-		return nil, err
-	}
-	c.compress = compress
-
 	return c, nil
 }
 
@@ -68,8 +84,7 @@ func NewChannelOut() (*ChannelOut, error) {
 func (co *ChannelOut) Reset() error {
 	co.frame = 0
 	co.rlpLength = 0
-	co.buf.Reset()
-	co.compress.Reset(&co.buf)
+	co.compress.Reset()
 	co.closed = false
 	_, err := rand.Read(co.id[:])
 	return err
@@ -116,7 +131,8 @@ func (co *ChannelOut) AddBatch(batch *BatchData) (uint64, error) {
 	}
 	co.rlpLength += buf.Len()
 
-	written, err := io.Copy(co.compress, &buf)
+	// avoid using io.Copy here, because we need all or nothing
+	written, err := co.compress.Write(buf.Bytes())
 	return uint64(written), err
 }
 
@@ -129,7 +145,7 @@ func (co *ChannelOut) InputBytes() int {
 // Use `Flush` or `Close` to move data from the compression buffer into the ready buffer if more bytes
 // are needed. Add blocks may add to the ready buffer, but it is not guaranteed due to the compression stage.
 func (co *ChannelOut) ReadyBytes() int {
-	return co.buf.Len()
+	return co.compress.Len()
 }
 
 // Flush flushes the internal compression stage to the ready buffer. It enables pulling a larger & more
@@ -166,8 +182,8 @@ func (co *ChannelOut) OutputFrame(w *bytes.Buffer, maxSize uint64) (uint16, erro
 
 	// Copy data from the local buffer into the frame data buffer
 	maxDataSize := maxSize - FrameV0OverHeadSize
-	if maxDataSize > uint64(co.buf.Len()) {
-		maxDataSize = uint64(co.buf.Len())
+	if maxDataSize > uint64(co.compress.Len()) {
+		maxDataSize = uint64(co.compress.Len())
 		// If we are closed & will not spill past the current frame
 		// mark it is the final frame of the channel.
 		if co.closed {
@@ -176,7 +192,7 @@ func (co *ChannelOut) OutputFrame(w *bytes.Buffer, maxSize uint64) (uint16, erro
 	}
 	f.Data = make([]byte, maxDataSize)
 
-	if _, err := io.ReadFull(&co.buf, f.Data); err != nil {
+	if _, err := io.ReadFull(co.compress, f.Data); err != nil {
 		return 0, err
 	}
 

op-node/rollup/derive/channel_out_test.go

@@ -11,8 +11,21 @@ import (
 	"github.com/stretchr/testify/require"
 )
 
+// basic implementation of the Compressor interface that does no compression
+type nonCompressor struct {
+	bytes.Buffer
+}
+
+func (s *nonCompressor) Flush() error {
+	return nil
+}
+
+func (s *nonCompressor) Close() error {
+	return nil
+}
+
 func TestChannelOutAddBlock(t *testing.T) {
-	cout, err := NewChannelOut()
+	cout, err := NewChannelOut(&nonCompressor{})
 	require.NoError(t, err)
 
 	t.Run("returns err if first tx is not an l1info tx", func(t *testing.T) {
@@ -33,7 +46,7 @@ func TestChannelOutAddBlock(t *testing.T) {
 // max size that is below the fixed frame size overhead of 23, will return
 // an error.
 func TestOutputFrameSmallMaxSize(t *testing.T) {
-	cout, err := NewChannelOut()
+	cout, err := NewChannelOut(&nonCompressor{})
 	require.NoError(t, err)
 
 	// Call OutputFrame with the range of small max size values that err