Add `GarbageChannelOut` test harness; Invalid compression & malformed block RLP tests

op-e2e/actions/garbage_channel_out.go

+package actions
+
+import (
+	"bytes"
+	"compress/gzip"
+	"compress/zlib"
+	"crypto/rand"
+	"errors"
+	"fmt"
+	"io"
+
+	"github.com/ethereum-optimism/optimism/op-node/rollup"
+	"github.com/ethereum-optimism/optimism/op-node/rollup/derive"
+	"github.com/ethereum/go-ethereum/common/hexutil"
+	"github.com/ethereum/go-ethereum/core/types"
+	"github.com/ethereum/go-ethereum/rlp"
+)
+
+var ErrNotDepositTx = errors.New("first transaction in block is not a deposit tx")
+var ErrTooManyRLPBytes = errors.New("batch would cause RLP bytes to go over limit")
+
+// Interface shared between `zlib.Writer` and `gzip.Writer`
+type WriterApi interface {
+	Close() error
+	Flush() error
+	Reset(io.Writer)
+	Write([]byte) (int, error)
+}
+
+// Modified `derive.ChannelOut` that can be configured to behave differently than the original
+type GarbageChannelOut struct {
+	id derive.ChannelID
+	// Frame ID of the next frame to emit. Increment after emitting
+	frame uint64
+	// rlpLength is the uncompressed size of the channel. Must be less than MAX_RLP_BYTES_PER_CHANNEL
+	rlpLength int
+
+	// Compressor stage. Write input data to it
+	compress WriterApi
+	// post compression buffer
+	buf bytes.Buffer
+
+	closed bool
+
+	// Garbage channel configuration
+	cfg *GarbageChannelCfg
+}
+
+func (co *GarbageChannelOut) ID() derive.ChannelID {
+	return co.id
+}
+
+func NewGarbageChannelOut(cfg *GarbageChannelCfg) (*GarbageChannelOut, error) {
+	c := &GarbageChannelOut{
+		id:        derive.ChannelID{}, // TODO: use GUID here instead of fully random data
+		frame:     0,
+		rlpLength: 0,
+		cfg:       cfg,
+	}
+	_, err := rand.Read(c.id[:])
+	if err != nil {
+		return nil, err
+	}
+
+	// Optionally use zlib or gzip compression
+	var compress WriterApi
+	if cfg.useInvalidCompression {
+		compress, err = gzip.NewWriterLevel(&c.buf, gzip.BestCompression)
+	} else {
+		compress, err = zlib.NewWriterLevel(&c.buf, zlib.BestCompression)
+	}
+	if err != nil {
+		return nil, err
+	}
+	c.compress = compress
+
+	return c, nil
+}
+
+// TODO: reuse ChannelOut for performance
+func (co *GarbageChannelOut) Reset() error {
+	co.frame = 0
+	co.rlpLength = 0
+	co.buf.Reset()
+	co.compress.Reset(&co.buf)
+	co.closed = false
+	_, err := rand.Read(co.id[:])
+	if err != nil {
+		return err
+	}
+	return nil
+}
+
+// AddBlock adds a block to the channel. It returns an error
+// if there is a problem adding the block. The only sentinel
+// 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(block *types.Block) error {
+	if co.closed {
+		return errors.New("already closed")
+	}
+	batch, err := blockToBatch(block)
+	if err != nil {
+		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 err
+	}
+	if co.cfg.malformRLP {
+		// Malform the RLP by incrementing the length prefix by 1.
+		bufBytes := buf.Bytes()
+		bufBytes[0] += 1
+		buf.Reset()
+		buf.Write(bufBytes)
+	}
+	if co.rlpLength+buf.Len() > derive.MaxRLPBytesPerChannel {
+		return fmt.Errorf("could not add %d bytes to channel of %d bytes, max is %d. err: %w",
+			buf.Len(), co.rlpLength, derive.MaxRLPBytesPerChannel, ErrTooManyRLPBytes)
+	}
+	co.rlpLength += buf.Len()
+
+	_, err = io.Copy(co.compress, &buf)
+	return err
+}
+
+// ReadyBytes returns the number of bytes that the channel out can immediately output into a frame.
+// 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 *GarbageChannelOut) ReadyBytes() int {
+	return co.buf.Len()
+}
+
+// Flush flushes the internal compression stage to the ready buffer. It enables pulling a larger & more
+// complete frame. It reduces the compression efficiency.
+func (co *GarbageChannelOut) Flush() error {
+	return co.compress.Flush()
+}
+
+func (co *GarbageChannelOut) Close() error {
+	if co.closed {
+		return errors.New("already closed")
+	}
+	co.closed = true
+	return co.compress.Close()
+}
+
+// OutputFrame writes a frame to w with a given max size
+// Use `ReadyBytes`, `Flush`, and `Close` to modify the ready buffer.
+// Returns io.EOF when the channel is closed & there are no more frames
+// Returns nil if there is still more buffered data.
+// Returns and error if it ran into an error during processing.
+func (co *GarbageChannelOut) OutputFrame(w *bytes.Buffer, maxSize uint64) error {
+	f := derive.Frame{
+		ID:          co.id,
+		FrameNumber: uint16(co.frame),
+	}
+
+	// Copy data from the local buffer into the frame data buffer
+	// Don't go past the maxSize with the fixed frame overhead.
+	// Fixed overhead: 32 + 8 + 2 + 4 + 1  = 47 bytes.
+	// Add one extra byte for the version byte (for the entire L1 tx though)
+	maxDataSize := maxSize - 47 - 1
+	if maxDataSize > uint64(co.buf.Len()) {
+		maxDataSize = uint64(co.buf.Len())
+		// If we are closed & will not spill past the current frame
+		// mark it is the final frame of the channel.
+		if co.closed {
+			f.IsLast = true
+		}
+	}
+	f.Data = make([]byte, maxDataSize)
+
+	if _, err := io.ReadFull(&co.buf, f.Data); err != nil {
+		return err
+	}
+
+	if err := f.MarshalBinary(w); err != nil {
+		return err
+	}
+
+	co.frame += 1
+	if f.IsLast {
+		return io.EOF
+	} else {
+		return nil
+	}
+}
+
+// blockToBatch transforms a block into a batch object that can easily be RLP encoded.
+func blockToBatch(block *types.Block) (*derive.BatchData, error) {
+	opaqueTxs := make([]hexutil.Bytes, 0, len(block.Transactions()))
+	for i, tx := range block.Transactions() {
+		if tx.Type() == types.DepositTxType {
+			continue
+		}
+		otx, err := tx.MarshalBinary()
+		if err != nil {
+			return nil, fmt.Errorf("could not encode tx %v in block %v: %w", i, tx.Hash(), err)
+		}
+		opaqueTxs = append(opaqueTxs, otx)
+	}
+	l1InfoTx := block.Transactions()[0]
+	if l1InfoTx.Type() != types.DepositTxType {
+		return nil, ErrNotDepositTx
+	}
+	l1Info, err := derive.L1InfoDepositTxData(l1InfoTx.Data())
+	if err != nil {
+		return nil, fmt.Errorf("could not parse the L1 Info deposit: %w", err)
+	}
+
+	return &derive.BatchData{
+		derive.BatchV1{
+			ParentHash:   block.ParentHash(),
+			EpochNum:     rollup.Epoch(l1Info.Number),
+			EpochHash:    l1Info.BlockHash,
+			Timestamp:    block.Time(),
+			Transactions: opaqueTxs,
+		},
+	}, nil
+}

op-e2e/actions/l2_batcher.go

@@ -35,12 +35,24 @@ type L1TxAPI interface {
 	SendTransaction(ctx context.Context, tx *types.Transaction) error
 }
 
+type ChannelOutApi interface {
+	ID() derive.ChannelID
+	Reset() error
+	AddBlock(block *types.Block) error
+	ReadyBytes() int
+	Flush() error
+	Close() error
+	OutputFrame(w *bytes.Buffer, maxSize uint64) error
+}
+
 type BatcherCfg struct {
 	// Limit the size of txs
 	MinL1TxSize uint64
 	MaxL1TxSize uint64
 
 	BatcherKey *ecdsa.PrivateKey
+
+	GarbageCfg *GarbageChannelCfg
 }
 
 // L2Batcher buffers and submits L2 batches to L1.
@@ -59,7 +71,7 @@ type L2Batcher struct {
 
 	l1Signer types.Signer
 
-	l2ChannelOut     *derive.ChannelOut
+	l2ChannelOut     ChannelOutApi
 	l2Submitting     bool // when the channel out is being submitted, and not safe to write to without resetting
 	l2BufferedBlock  eth.BlockID
 	l2SubmittedBlock eth.BlockID
@@ -123,7 +135,12 @@ func (s *L2Batcher) ActL2BatchBuffer(t Testing) {
 	}
 	// Create channel if we don't have one yet
 	if s.l2ChannelOut == nil {
-		ch, err := derive.NewChannelOut()
+		var ch ChannelOutApi
+		if s.l2BatcherCfg.GarbageCfg != nil {
+			ch, err = NewGarbageChannelOut(s.l2BatcherCfg.GarbageCfg)
+		} else {
+			ch, err = derive.NewChannelOut()
+		}
 		require.NoError(t, err, "failed to create channel")
 		s.l2ChannelOut = ch
 	}
@@ -196,25 +213,45 @@ func (s *L2Batcher) ActL2BatchSubmit(t Testing) {
 	require.NoError(t, err, "need to send tx")
 }
 
+func (s *L2Batcher) ActBufferAll(t Testing) {
+	stat, err := s.syncStatusAPI.SyncStatus(t.Ctx())
+	require.NoError(t, err)
+	for s.l2BufferedBlock.Number < stat.UnsafeL2.Number {
+		s.ActL2BatchBuffer(t)
+	}
+}
+
+func (s *L2Batcher) ActSubmitAll(t Testing) {
+	s.ActBufferAll(t)
+	s.ActL2ChannelClose(t)
+	s.ActL2BatchSubmit(t)
+}
+
 // TODO: Move this to a better location
 type GarbageKind int64
 
 const (
 	STRIP_VERSION GarbageKind = iota
 	RANDOM
-	MALFORM_RLP
-	INVALID_COMPRESSION
 	TRUNCATE_END
 	DIRTY_APPEND
+	INVALID_COMPRESSION
+	MALFORM_RLP
 )
 
 var GarbageKinds = []GarbageKind{
 	STRIP_VERSION,
 	RANDOM,
-	// MALFORM_RLP,
-	// INVALID_COMPRESSION,
 	TRUNCATE_END,
 	DIRTY_APPEND,
+	INVALID_COMPRESSION,
+	MALFORM_RLP,
+}
+
+// Configuration for a garbage channel (`ChannelOut` in the `actions` pkg)
+type GarbageChannelCfg struct {
+	useInvalidCompression bool
+	malformRLP            bool
 }
 
 // ActL2BatchSubmit constructs a malformed channel frame and submits it to the
@@ -226,6 +263,7 @@ func (s *L2Batcher) ActL2BatchSubmitGarbage(t Testing, kind GarbageKind) {
 		t.InvalidAction("need to buffer data first, cannot batch submit with empty buffer")
 		return
 	}
+
 	// Collect the output frame
 	data := new(bytes.Buffer)
 	data.WriteByte(derive.DerivationVersion0)
@@ -260,6 +298,14 @@ func (s *L2Batcher) ActL2BatchSubmitGarbage(t Testing, kind GarbageKind) {
 	// Append 4 garbage bytes to the end of the output frame
 	case DIRTY_APPEND:
 		outputFrame = append(outputFrame, []byte{0xBA, 0xD0, 0xC0, 0xDE}...)
+	case INVALID_COMPRESSION:
+		// Do nothing post frame encoding- the `GarbageChannelOut` used for this case is modified to
+		// use gzip compression rather than zlib, which is invalid.
+		break
+	case MALFORM_RLP:
+		// Do nothing post frame encoding- the `GarbageChannelOut` used for this case is modified to
+		// write malformed RLP each time a block is added to the channel.
+		break
 	default:
 		t.Fatalf("Unexpected garbage kind: %v", kind)
 	}
@@ -290,17 +336,3 @@ func (s *L2Batcher) ActL2BatchSubmitGarbage(t Testing, kind GarbageKind) {
 	err = s.l1.SendTransaction(t.Ctx(), tx)
 	require.NoError(t, err, "need to send tx")
 }
-
-func (s *L2Batcher) ActBufferAll(t Testing) {
-	stat, err := s.syncStatusAPI.SyncStatus(t.Ctx())
-	require.NoError(t, err)
-	for s.l2BufferedBlock.Number < stat.UnsafeL2.Number {
-		s.ActL2BatchBuffer(t)
-	}
-}
-
-func (s *L2Batcher) ActSubmitAll(t Testing) {
-	s.ActBufferAll(t)
-	s.ActL2ChannelClose(t)
-	s.ActL2BatchSubmit(t)
-}

op-e2e/actions/l2_batcher_test.go

@@ -209,11 +209,22 @@ func TestGarbageBatch(gt *testing.T) {
 
 		_, verifier := setupVerifier(t, sd, log, miner.L1Client(t, sd.RollupCfg))
 
-		batcher := NewL2Batcher(log, sd.RollupCfg, &BatcherCfg{
+		batcherCfg := &BatcherCfg{
 			MinL1TxSize: 0,
 			MaxL1TxSize: 128_000,
 			BatcherKey:  dp.Secrets.Batcher,
-		}, sequencer.RollupClient(), miner.EthClient(), engine.EthClient())
+		}
+
+		if garbageKind == MALFORM_RLP || garbageKind == INVALID_COMPRESSION {
+			// If the garbage kind is `INVALID_COMPRESSION` or `MALFORM_RLP`, use the `actions` packages
+			// modified `ChannelOut`.
+			batcherCfg.GarbageCfg = &GarbageChannelCfg{
+				useInvalidCompression: garbageKind == INVALID_COMPRESSION,
+				malformRLP:            garbageKind == MALFORM_RLP,
+			}
+		}
+
+		batcher := NewL2Batcher(log, sd.RollupCfg, batcherCfg, sequencer.RollupClient(), miner.EthClient(), engine.EthClient())
 
 		sequencer.ActL2PipelineFull(t)
 		verifier.ActL2PipelineFull(t)