Multiple pending channels

op-batcher/batcher/channel.go

+package batcher
+
+import (
+	"fmt"
+	"math"
+
+	"github.com/ethereum-optimism/optimism/op-batcher/metrics"
+	"github.com/ethereum-optimism/optimism/op-node/eth"
+	"github.com/ethereum-optimism/optimism/op-node/rollup/derive"
+	"github.com/ethereum/go-ethereum/core/types"
+	"github.com/ethereum/go-ethereum/log"
+)
+
+type channel struct {
+	log  log.Logger
+	metr metrics.Metricer
+	cfg  ChannelConfig
+
+	// pending channel builder
+	channelBuilder *channelBuilder
+	// Set of unconfirmed txID -> frame data. For tx resubmission
+	pendingTransactions map[txID]txData
+	// Set of confirmed txID -> inclusion block. For determining if the channel is timed out
+	confirmedTransactions map[txID]eth.BlockID
+}
+
+func newPendingChannel(log log.Logger, metr metrics.Metricer, cfg ChannelConfig) (*channel, error) {
+	cb, err := newChannelBuilder(cfg)
+	if err != nil {
+		return nil, fmt.Errorf("creating new channel: %w", err)
+	}
+	return &channel{
+		log:                   log,
+		metr:                  metr,
+		cfg:                   cfg,
+		channelBuilder:        cb,
+		pendingTransactions:   make(map[txID]txData),
+		confirmedTransactions: make(map[txID]eth.BlockID),
+	}, nil
+}
+
+// TxFailed records a transaction as failed. It will attempt to resubmit the data
+// in the failed transaction.
+func (s *channel) TxFailed(id txID) bool {
+	if data, ok := s.pendingTransactions[id]; ok {
+		s.log.Trace("marked transaction as failed", "id", id)
+		// Note: when the batcher is changed to send multiple frames per tx,
+		// this needs to be changed to iterate over all frames of the tx data
+		// and re-queue them.
+		s.channelBuilder.PushFrame(data.Frame())
+		delete(s.pendingTransactions, id)
+	} else {
+		s.log.Warn("unknown transaction marked as failed", "id", id)
+	}
+
+	s.metr.RecordBatchTxFailed()
+
+	return len(s.pendingTransactions) == 0 && len(s.confirmedTransactions) == 0
+}
+
+// TxConfirmed marks a transaction as confirmed on L1. Unfortunately even if all frames in
+// a channel have been marked as confirmed on L1 the channel may be invalid & need to be
+// resubmitted.
+// This function may reset the pending channel if the pending channel has timed out.
+func (s *channel) TxConfirmed(id txID, inclusionBlock eth.BlockID) (bool, []*types.Block) {
+	s.metr.RecordBatchTxSubmitted()
+	s.log.Debug("marked transaction as confirmed", "id", id, "block", inclusionBlock)
+	if _, ok := s.pendingTransactions[id]; !ok {
+		s.log.Warn("unknown transaction marked as confirmed", "id", id, "block", inclusionBlock)
+		// TODO: This can occur if we clear the channel while there are still pending transactions
+		// We need to keep track of stale transactions instead
+		return false, nil
+	}
+	delete(s.pendingTransactions, id)
+	s.confirmedTransactions[id] = inclusionBlock
+	s.channelBuilder.FramePublished(inclusionBlock.Number)
+
+	// If this channel timed out, put the pending blocks back into the local saved blocks
+	// and then reset this state so it can try to build a new channel.
+	if s.isTimedOut() {
+		s.metr.RecordChannelTimedOut(s.ID())
+		s.log.Warn("Channel timed out", "id", s.ID())
+		return true, s.channelBuilder.Blocks()
+	}
+	// If we are done with this channel, record that.
+	if s.isFullySubmitted() {
+		s.metr.RecordChannelFullySubmitted(s.ID())
+		s.log.Info("Channel is fully submitted", "id", s.ID())
+		return true, nil
+	}
+
+	return false, nil
+}
+
+// pendingChannelIsTimedOut returns true if submitted channel has timed out.
+// A channel has timed out if the difference in L1 Inclusion blocks between
+// the first & last included block is greater than or equal to the channel timeout.
+func (s *channel) isTimedOut() bool {
+	if len(s.confirmedTransactions) == 0 {
+		return false
+	}
+	// If there are confirmed transactions, find the first + last confirmed block numbers
+	min := uint64(math.MaxUint64)
+	max := uint64(0)
+	for _, inclusionBlock := range s.confirmedTransactions {
+		if inclusionBlock.Number < min {
+			min = inclusionBlock.Number
+		}
+		if inclusionBlock.Number > max {
+			max = inclusionBlock.Number
+		}
+	}
+	return max-min >= s.cfg.ChannelTimeout
+}
+
+// pendingChannelIsFullySubmitted returns true if the channel has been fully submitted.
+func (s *channel) isFullySubmitted() bool {
+	return s.IsFull() && len(s.pendingTransactions)+s.NumFrames() == 0
+}
+
+func (s *channel) NoneSubmitted() bool {
+	return len(s.confirmedTransactions) == 0 && len(s.pendingTransactions) == 0
+}
+
+func (s *channel) ID() derive.ChannelID {
+	return s.channelBuilder.ID()
+}
+
+func (s *channel) NextTxData() txData {
+	frame := s.channelBuilder.NextFrame()
+
+	txdata := txData{frame}
+	id := txdata.ID()
+
+	s.log.Trace("returning next tx data", "id", id)
+	s.pendingTransactions[id] = txdata
+
+	return txdata
+}
+
+func (s *channel) HasFrame() bool {
+	return s.channelBuilder.HasFrame()
+}
+
+func (s *channel) IsFull() bool {
+	return s.channelBuilder.IsFull()
+}
+
+func (s *channel) FullErr() error {
+	return s.channelBuilder.FullErr()
+}
+
+func (s *channel) RegisterL1Block(l1BlockNum uint64) {
+	s.channelBuilder.RegisterL1Block(l1BlockNum)
+}
+
+func (s *channel) AddBlock(block *types.Block) (derive.L1BlockInfo, error) {
+	return s.channelBuilder.AddBlock(block)
+}
+
+func (s *channel) InputBytes() int {
+	return s.channelBuilder.InputBytes()
+}
+
+func (s *channel) ReadyBytes() int {
+	return s.channelBuilder.ReadyBytes()
+}
+
+func (s *channel) OutputBytes() int {
+	return s.channelBuilder.OutputBytes()
+}
+
+func (s *channel) NumFrames() int {
+	return s.channelBuilder.NumFrames()
+}
+
+func (s *channel) OutputFrames() error {
+	return s.channelBuilder.OutputFrames()
+}
+
+func (s *channel) Close() {
+	s.channelBuilder.Close()
+}

op-batcher/batcher/channel_manager_test.go

@@ -19,50 +19,6 @@ import (
 	"github.com/stretchr/testify/require"
 )
 
-// TestPendingChannelTimeout tests that the channel manager
-// correctly identifies when a pending channel is timed out.
-func TestPendingChannelTimeout(t *testing.T) {
-	// Create a new channel manager with a ChannelTimeout
-	log := testlog.Logger(t, log.LvlCrit)
-	m := NewChannelManager(log, metrics.NoopMetrics, ChannelConfig{
-		ChannelTimeout: 100,
-	})
-
-	// Pending channel is nil so is cannot be timed out
-	timeout := m.pendingChannelIsTimedOut()
-	require.False(t, timeout)
-
-	// Set the pending channel
-	require.NoError(t, m.ensurePendingChannel(eth.BlockID{}))
-
-	// There are no confirmed transactions so
-	// the pending channel cannot be timed out
-	timeout = m.pendingChannelIsTimedOut()
-	require.False(t, timeout)
-
-	// Manually set a confirmed transactions
-	// To avoid other methods clearing state
-	m.confirmedTransactions[frameID{frameNumber: 0}] = eth.BlockID{Number: 0}
-	m.confirmedTransactions[frameID{frameNumber: 1}] = eth.BlockID{Number: 99}
-
-	// Since the ChannelTimeout is 100, the
-	// pending channel should not be timed out
-	timeout = m.pendingChannelIsTimedOut()
-	require.False(t, timeout)
-
-	// Add a confirmed transaction with a higher number
-	// than the ChannelTimeout
-	m.confirmedTransactions[frameID{
-		frameNumber: 2,
-	}] = eth.BlockID{
-		Number: 101,
-	}
-
-	// Now the pending channel should be timed out
-	timeout = m.pendingChannelIsTimedOut()
-	require.True(t, timeout)
-}
-
 // TestChannelManagerReturnsErrReorg ensures that the channel manager
 // detects a reorg when it has cached L1 blocks.
 func TestChannelManagerReturnsErrReorg(t *testing.T) {
@@ -101,7 +57,8 @@ func TestChannelManagerReturnsErrReorgWhenDrained(t *testing.T) {
 		ChannelConfig{
 			MaxFrameSize: 120_000,
 			CompressorConfig: compressor.Config{
-				TargetFrameSize:  0,
+				TargetFrameSize:  1,
+				TargetNumFrames:  1,
 				ApproxComprRatio: 1.0,
 			},
 		})
@@ -119,46 +76,6 @@ func TestChannelManagerReturnsErrReorgWhenDrained(t *testing.T) {
 	require.ErrorIs(t, m.AddL2Block(x), ErrReorg)
 }
 
-// TestChannelManagerNextTxData checks the nextTxData function.
-func TestChannelManagerNextTxData(t *testing.T) {
-	log := testlog.Logger(t, log.LvlCrit)
-	m := NewChannelManager(log, metrics.NoopMetrics, ChannelConfig{})
-
-	// Nil pending channel should return EOF
-	returnedTxData, err := m.nextTxData()
-	require.ErrorIs(t, err, io.EOF)
-	require.Equal(t, txData{}, returnedTxData)
-
-	// Set the pending channel
-	// The nextTxData function should still return EOF
-	// since the pending channel has no frames
-	require.NoError(t, m.ensurePendingChannel(eth.BlockID{}))
-	returnedTxData, err = m.nextTxData()
-	require.ErrorIs(t, err, io.EOF)
-	require.Equal(t, txData{}, returnedTxData)
-
-	// Manually push a frame into the pending channel
-	channelID := m.pendingChannel.ID()
-	frame := frameData{
-		data: []byte{},
-		id: frameID{
-			chID:        channelID,
-			frameNumber: uint16(0),
-		},
-	}
-	m.pendingChannel.PushFrame(frame)
-	require.Equal(t, 1, m.pendingChannel.NumFrames())
-
-	// Now the nextTxData function should return the frame
-	returnedTxData, err = m.nextTxData()
-	expectedTxData := txData{frame}
-	expectedChannelID := expectedTxData.ID()
-	require.NoError(t, err)
-	require.Equal(t, expectedTxData, returnedTxData)
-	require.Equal(t, 0, m.pendingChannel.NumFrames())
-	require.Equal(t, expectedTxData, m.pendingTransactions[expectedChannelID])
-}
-
 // TestChannelManager_Clear tests clearing the channel manager.
 func TestChannelManager_Clear(t *testing.T) {
 	require := require.New(t)
@@ -184,9 +101,9 @@ func TestChannelManager_Clear(t *testing.T) {
 	// Channel Manager state should be empty by default
 	require.Empty(m.blocks)
 	require.Equal(common.Hash{}, m.tip)
-	require.Nil(m.pendingChannel)
-	require.Empty(m.pendingTransactions)
-	require.Empty(m.confirmedTransactions)
+	require.Nil(m.currentChannel)
+	require.Empty(m.channelQueue)
+	require.Empty(m.txChannels)
 
 	// Add a block to the channel manager
 	a, _ := derivetest.RandomL2Block(rng, 4)
@@ -198,22 +115,22 @@ func TestChannelManager_Clear(t *testing.T) {
 	require.NoError(m.AddL2Block(a))
 
 	// Make sure there is a channel builder
-	require.NoError(m.ensurePendingChannel(l1BlockID))
-	require.NotNil(m.pendingChannel)
-	require.Len(m.confirmedTransactions, 0)
+	require.NoError(m.ensureChannelWithRoom(l1BlockID))
+	require.NotNil(m.currentChannel)
+	require.Len(m.currentChannel.confirmedTransactions, 0)
 
 	// Process the blocks
 	// We should have a pending channel with 1 frame
 	// and no more blocks since processBlocks consumes
 	// the list
 	require.NoError(m.processBlocks())
-	require.NoError(m.pendingChannel.co.Flush())
-	require.NoError(m.pendingChannel.OutputFrames())
-	_, err := m.nextTxData()
+	require.NoError(m.currentChannel.channelBuilder.co.Flush())
+	require.NoError(m.currentChannel.OutputFrames())
+	_, err := m.nextTxData(m.currentChannel)
 	require.NoError(err)
 	require.Len(m.blocks, 0)
 	require.Equal(newL1Tip, m.tip)
-	require.Len(m.pendingTransactions, 1)
+	require.Len(m.currentChannel.pendingTransactions, 1)
 
 	// Add a new block so we can test clearing
 	// the channel manager with a full state
@@ -231,104 +148,9 @@ func TestChannelManager_Clear(t *testing.T) {
 	// Check that the entire channel manager state cleared
 	require.Empty(m.blocks)
 	require.Equal(common.Hash{}, m.tip)
-	require.Nil(m.pendingChannel)
-	require.Empty(m.pendingTransactions)
-	require.Empty(m.confirmedTransactions)
-}
-
-// TestChannelManagerTxConfirmed checks the [ChannelManager.TxConfirmed] function.
-func TestChannelManagerTxConfirmed(t *testing.T) {
-	// Create a channel manager
-	log := testlog.Logger(t, log.LvlCrit)
-	m := NewChannelManager(log, metrics.NoopMetrics, ChannelConfig{
-		// Need to set the channel timeout here so we don't clear pending
-		// channels on confirmation. This would result in [TxConfirmed]
-		// clearing confirmed transactions, and reseting the pendingChannels map
-		ChannelTimeout: 10,
-	})
-
-	// Let's add a valid pending transaction to the channel manager
-	// So we can demonstrate that TxConfirmed's correctness
-	require.NoError(t, m.ensurePendingChannel(eth.BlockID{}))
-	channelID := m.pendingChannel.ID()
-	frame := frameData{
-		data: []byte{},
-		id: frameID{
-			chID:        channelID,
-			frameNumber: uint16(0),
-		},
-	}
-	m.pendingChannel.PushFrame(frame)
-	require.Equal(t, 1, m.pendingChannel.NumFrames())
-	returnedTxData, err := m.nextTxData()
-	expectedTxData := txData{frame}
-	expectedChannelID := expectedTxData.ID()
-	require.NoError(t, err)
-	require.Equal(t, expectedTxData, returnedTxData)
-	require.Equal(t, 0, m.pendingChannel.NumFrames())
-	require.Equal(t, expectedTxData, m.pendingTransactions[expectedChannelID])
-	require.Len(t, m.pendingTransactions, 1)
-
-	// An unknown pending transaction should not be marked as confirmed
-	// and should not be removed from the pending transactions map
-	actualChannelID := m.pendingChannel.ID()
-	unknownChannelID := derive.ChannelID([derive.ChannelIDLength]byte{0x69})
-	require.NotEqual(t, actualChannelID, unknownChannelID)
-	unknownTxID := frameID{chID: unknownChannelID, frameNumber: 0}
-	blockID := eth.BlockID{Number: 0, Hash: common.Hash{0x69}}
-	m.TxConfirmed(unknownTxID, blockID)
-	require.Empty(t, m.confirmedTransactions)
-	require.Len(t, m.pendingTransactions, 1)
-
-	// Now let's mark the pending transaction as confirmed
-	// and check that it is removed from the pending transactions map
-	// and added to the confirmed transactions map
-	m.TxConfirmed(expectedChannelID, blockID)
-	require.Empty(t, m.pendingTransactions)
-	require.Len(t, m.confirmedTransactions, 1)
-	require.Equal(t, blockID, m.confirmedTransactions[expectedChannelID])
-}
-
-// TestChannelManagerTxFailed checks the [ChannelManager.TxFailed] function.
-func TestChannelManagerTxFailed(t *testing.T) {
-	// Create a channel manager
-	log := testlog.Logger(t, log.LvlCrit)
-	m := NewChannelManager(log, metrics.NoopMetrics, ChannelConfig{})
-
-	// Let's add a valid pending transaction to the channel
-	// manager so we can demonstrate correctness
-	require.NoError(t, m.ensurePendingChannel(eth.BlockID{}))
-	channelID := m.pendingChannel.ID()
-	frame := frameData{
-		data: []byte{},
-		id: frameID{
-			chID:        channelID,
-			frameNumber: uint16(0),
-		},
-	}
-	m.pendingChannel.PushFrame(frame)
-	require.Equal(t, 1, m.pendingChannel.NumFrames())
-	returnedTxData, err := m.nextTxData()
-	expectedTxData := txData{frame}
-	expectedChannelID := expectedTxData.ID()
-	require.NoError(t, err)
-	require.Equal(t, expectedTxData, returnedTxData)
-	require.Equal(t, 0, m.pendingChannel.NumFrames())
-	require.Equal(t, expectedTxData, m.pendingTransactions[expectedChannelID])
-	require.Len(t, m.pendingTransactions, 1)
-
-	// Trying to mark an unknown pending transaction as failed
-	// shouldn't modify state
-	m.TxFailed(frameID{})
-	require.Equal(t, 0, m.pendingChannel.NumFrames())
-	require.Equal(t, expectedTxData, m.pendingTransactions[expectedChannelID])
-
-	// Now we still have a pending transaction
-	// Let's mark it as failed
-	m.TxFailed(expectedChannelID)
-	require.Empty(t, m.pendingTransactions)
-	// There should be a frame in the pending channel now
-	require.Equal(t, 1, m.pendingChannel.NumFrames())
+	require.Nil(m.currentChannel)
+	require.Empty(m.channelQueue)
+	require.Empty(m.txChannels)
 }
 
 func TestChannelManager_TxResend(t *testing.T) {
@@ -339,7 +161,8 @@ func TestChannelManager_TxResend(t *testing.T) {
 		ChannelConfig{
 			MaxFrameSize: 120_000,
 			CompressorConfig: compressor.Config{
-				TargetFrameSize:  0,
+				TargetFrameSize:  1,
+				TargetNumFrames:  1,
 				ApproxComprRatio: 1.0,
 			},
 		})

op-batcher/batcher/channel_test.go

+package batcher
+
+import (
+	"io"
+	"testing"
+
+	"github.com/ethereum-optimism/optimism/op-batcher/metrics"
+	"github.com/ethereum-optimism/optimism/op-node/eth"
+	"github.com/ethereum-optimism/optimism/op-node/rollup/derive"
+	"github.com/ethereum-optimism/optimism/op-node/testlog"
+	"github.com/ethereum/go-ethereum/common"
+	"github.com/ethereum/go-ethereum/log"
+	"github.com/stretchr/testify/require"
+)
+
+// TestChannelTimeout tests that the channel manager
+// correctly identifies when a pending channel is timed out.
+func TestChannelTimeout(t *testing.T) {
+	// Create a new channel manager with a ChannelTimeout
+	log := testlog.Logger(t, log.LvlCrit)
+	m := NewChannelManager(log, metrics.NoopMetrics, ChannelConfig{
+		ChannelTimeout: 100,
+	})
+
+	// Pending channel is nil so is cannot be timed out
+	require.Nil(t, m.currentChannel)
+
+	// Set the pending channel
+	require.NoError(t, m.ensureChannelWithRoom(eth.BlockID{}))
+	channel := m.currentChannel
+	require.NotNil(t, channel)
+
+	// There are no confirmed transactions so
+	// the pending channel cannot be timed out
+	timeout := channel.isTimedOut()
+	require.False(t, timeout)
+
+	// Manually set a confirmed transactions
+	// To avoid other methods clearing state
+	channel.confirmedTransactions[frameID{frameNumber: 0}] = eth.BlockID{Number: 0}
+	channel.confirmedTransactions[frameID{frameNumber: 1}] = eth.BlockID{Number: 99}
+
+	// Since the ChannelTimeout is 100, the
+	// pending channel should not be timed out
+	timeout = channel.isTimedOut()
+	require.False(t, timeout)
+
+	// Add a confirmed transaction with a higher number
+	// than the ChannelTimeout
+	channel.confirmedTransactions[frameID{
+		frameNumber: 2,
+	}] = eth.BlockID{
+		Number: 101,
+	}
+
+	// Now the pending channel should be timed out
+	timeout = channel.isTimedOut()
+	require.True(t, timeout)
+}
+
+// TestChannelNextTxData checks the nextTxData function.
+func TestChannelNextTxData(t *testing.T) {
+	log := testlog.Logger(t, log.LvlCrit)
+	m := NewChannelManager(log, metrics.NoopMetrics, ChannelConfig{})
+
+	// Nil pending channel should return EOF
+	returnedTxData, err := m.nextTxData(nil)
+	require.ErrorIs(t, err, io.EOF)
+	require.Equal(t, txData{}, returnedTxData)
+
+	// Set the pending channel
+	// The nextTxData function should still return EOF
+	// since the pending channel has no frames
+	require.NoError(t, m.ensureChannelWithRoom(eth.BlockID{}))
+	channel := m.currentChannel
+	require.NotNil(t, channel)
+	returnedTxData, err = m.nextTxData(channel)
+	require.ErrorIs(t, err, io.EOF)
+	require.Equal(t, txData{}, returnedTxData)
+
+	// Manually push a frame into the pending channel
+	channelID := channel.ID()
+	frame := frameData{
+		data: []byte{},
+		id: frameID{
+			chID:        channelID,
+			frameNumber: uint16(0),
+		},
+	}
+	channel.channelBuilder.PushFrame(frame)
+	require.Equal(t, 1, channel.NumFrames())
+
+	// Now the nextTxData function should return the frame
+	returnedTxData, err = m.nextTxData(channel)
+	expectedTxData := txData{frame}
+	expectedChannelID := expectedTxData.ID()
+	require.NoError(t, err)
+	require.Equal(t, expectedTxData, returnedTxData)
+	require.Equal(t, 0, channel.NumFrames())
+	require.Equal(t, expectedTxData, channel.pendingTransactions[expectedChannelID])
+}
+
+// TestChannelManagerTxConfirmed checks the [ChannelManager.TxConfirmed] function.
+func TestChannelManagerTxConfirmed(t *testing.T) {
+	// Create a channel manager
+	log := testlog.Logger(t, log.LvlCrit)
+	m := NewChannelManager(log, metrics.NoopMetrics, ChannelConfig{
+		// Need to set the channel timeout here so we don't clear pending
+		// channels on confirmation. This would result in [TxConfirmed]
+		// clearing confirmed transactions, and reseting the pendingChannels map
+		ChannelTimeout: 10,
+	})
+
+	// Let's add a valid pending transaction to the channel manager
+	// So we can demonstrate that TxConfirmed's correctness
+	require.NoError(t, m.ensureChannelWithRoom(eth.BlockID{}))
+	channelID := m.currentChannel.ID()
+	frame := frameData{
+		data: []byte{},
+		id: frameID{
+			chID:        channelID,
+			frameNumber: uint16(0),
+		},
+	}
+	m.currentChannel.channelBuilder.PushFrame(frame)
+	require.Equal(t, 1, m.currentChannel.NumFrames())
+	returnedTxData, err := m.nextTxData(m.currentChannel)
+	expectedTxData := txData{frame}
+	expectedChannelID := expectedTxData.ID()
+	require.NoError(t, err)
+	require.Equal(t, expectedTxData, returnedTxData)
+	require.Equal(t, 0, m.currentChannel.NumFrames())
+	require.Equal(t, expectedTxData, m.currentChannel.pendingTransactions[expectedChannelID])
+	require.Len(t, m.currentChannel.pendingTransactions, 1)
+
+	// An unknown pending transaction should not be marked as confirmed
+	// and should not be removed from the pending transactions map
+	actualChannelID := m.currentChannel.ID()
+	unknownChannelID := derive.ChannelID([derive.ChannelIDLength]byte{0x69})
+	require.NotEqual(t, actualChannelID, unknownChannelID)
+	unknownTxID := frameID{chID: unknownChannelID, frameNumber: 0}
+	blockID := eth.BlockID{Number: 0, Hash: common.Hash{0x69}}
+	m.TxConfirmed(unknownTxID, blockID)
+	require.Empty(t, m.currentChannel.confirmedTransactions)
+	require.Len(t, m.currentChannel.pendingTransactions, 1)
+
+	// Now let's mark the pending transaction as confirmed
+	// and check that it is removed from the pending transactions map
+	// and added to the confirmed transactions map
+	m.TxConfirmed(expectedChannelID, blockID)
+	require.Empty(t, m.currentChannel.pendingTransactions)
+	require.Len(t, m.currentChannel.confirmedTransactions, 1)
+	require.Equal(t, blockID, m.currentChannel.confirmedTransactions[expectedChannelID])
+}
+
+// TestChannelManagerTxFailed checks the [ChannelManager.TxFailed] function.
+func TestChannelManagerTxFailed(t *testing.T) {
+	// Create a channel manager
+	log := testlog.Logger(t, log.LvlCrit)
+	m := NewChannelManager(log, metrics.NoopMetrics, ChannelConfig{})
+
+	// Let's add a valid pending transaction to the channel
+	// manager so we can demonstrate correctness
+	require.NoError(t, m.ensureChannelWithRoom(eth.BlockID{}))
+	channelID := m.currentChannel.ID()
+	frame := frameData{
+		data: []byte{},
+		id: frameID{
+			chID:        channelID,
+			frameNumber: uint16(0),
+		},
+	}
+	m.currentChannel.channelBuilder.PushFrame(frame)
+	require.Equal(t, 1, m.currentChannel.NumFrames())
+	returnedTxData, err := m.nextTxData(m.currentChannel)
+	expectedTxData := txData{frame}
+	expectedChannelID := expectedTxData.ID()
+	require.NoError(t, err)
+	require.Equal(t, expectedTxData, returnedTxData)
+	require.Equal(t, 0, m.currentChannel.NumFrames())
+	require.Equal(t, expectedTxData, m.currentChannel.pendingTransactions[expectedChannelID])
+	require.Len(t, m.currentChannel.pendingTransactions, 1)
+
+	// Trying to mark an unknown pending transaction as failed
+	// shouldn't modify state
+	m.TxFailed(frameID{})
+	require.Equal(t, 0, m.currentChannel.NumFrames())
+	require.Equal(t, expectedTxData, m.currentChannel.pendingTransactions[expectedChannelID])
+
+	// Now we still have a pending transaction
+	// Let's mark it as failed
+	m.TxFailed(expectedChannelID)
+	require.Empty(t, m.currentChannel.pendingTransactions)
+	// There should be a frame in the pending channel now
+	require.Equal(t, 1, m.currentChannel.NumFrames())
+}

op-batcher/compressor/config.go

@@ -16,7 +16,8 @@ type Config struct {
 	// ApproxComprRatio to assume. Should be slightly smaller than average from
 	// experiments to avoid the chances of creating a small additional leftover frame.
 	ApproxComprRatio float64
-	// Kind of compressor to use. Must
+	// Kind of compressor to use. Must be one of KindKeys. If unset, NewCompressor
+	// will default to RatioKind.
 	Kind string
 }