op-batcher: introduce ClearAllStateMetrics() and call from channelManager.Clear()

op-batcher/batcher/channel_manager.go

@@ -83,7 +83,10 @@ func (s *channelManager) Clear(l1OriginLastSubmittedChannel eth.BlockID) {
 	s.tip = common.Hash{}
 	s.currentChannel = nil
 	s.channelQueue = nil
-	s.metr.RecordChannelQueueLength(0)
+
+	// This is particularly important because pendingDABytes metric controls throttling:
+	s.metr.ClearAllStateMetrics()
+
 	s.txChannels = make(map[string]*channel)
 }
 

op-batcher/batcher/channel_manager_test.go

@@ -122,7 +122,7 @@ func ChannelManager_Clear(t *testing.T, batchType uint) {
 	// clearing confirmed transactions, and resetting the pendingChannels map
 	cfg.ChannelTimeout = 10
 	cfg.InitRatioCompressor(1, derive.Zlib)
-	m := NewChannelManager(log, metrics.NoopMetrics, cfg, defaultTestRollupConfig)
+	m := NewChannelManager(log, &metrics.TestMetrics{}, cfg, defaultTestRollupConfig)
 
 	// Channel Manager state should be empty by default
 	require.Empty(m.blocks)
@@ -150,7 +150,6 @@ func ChannelManager_Clear(t *testing.T, batchType uint) {
 
 	// Process the blocks
 	// We should have a pending channel with 1 frame
-
 	require.NoError(m.processBlocks())
 	require.NoError(m.currentChannel.channelBuilder.co.Flush())
 	require.NoError(m.outputFrames())
@@ -174,6 +173,11 @@ func ChannelManager_Clear(t *testing.T, batchType uint) {
 	safeL1Origin := eth.BlockID{
 		Number: 123,
 	}
+
+	// Artificially pump up some metrics which need to be cleared
+	m.metr.RecordL2BlockInPendingQueue(a)
+	require.NotZero(m.metr.PendingDABytes())
+
 	// Clear the channel manager
 	m.Clear(safeL1Origin)
 
@@ -184,6 +188,7 @@ func ChannelManager_Clear(t *testing.T, batchType uint) {
 	require.Nil(m.currentChannel)
 	require.Empty(m.channelQueue)
 	require.Empty(m.txChannels)
+	require.Zero(m.metr.PendingDABytes())
 }
 
 func ChannelManager_TxResend(t *testing.T, batchType uint) {

op-batcher/metrics/metrics.go

@@ -44,6 +44,10 @@ type Metricer interface {
 	RecordChannelTimedOut(id derive.ChannelID)
 	RecordChannelQueueLength(len int)
 
+	// ClearAllStateMetrics resets any metrics that track current ChannelManager state
+	// It should be called when clearing the ChannelManager state.
+	ClearAllStateMetrics()
+
 	RecordBatchTxSubmitted()
 	RecordBatchTxSuccess()
 	RecordBatchTxFailed()
@@ -349,6 +353,13 @@ func (m *Metrics) RecordChannelQueueLength(len int) {
 	m.channelQueueLength.Set(float64(len))
 }
 
+// ClearAllStateMetrics clears all state metrics.
+func (m *Metrics) ClearAllStateMetrics() {
+	m.RecordChannelQueueLength(0)
+	atomic.StoreInt64(&m.pendingDABytes, 0)
+	m.pendingBlocksBytesCurrent.Set(0)
+}
+
 // estimateBatchSize returns the estimated size of the block in a batch both with compression ('daSize') and without
 // ('rawSize').
 func estimateBatchSize(block *types.Block) (daSize, rawSize uint64) {

op-batcher/metrics/noop.go

@@ -61,3 +61,5 @@ type ThrottlingMetrics struct {
 func (nm *ThrottlingMetrics) PendingDABytes() float64 {
 	return math.MaxFloat64
 }
+
+func (*noopMetrics) ClearAllStateMetrics() {}

op-batcher/metrics/test.go

@@ -8,19 +8,28 @@ type TestMetrics struct {
 	noopMetrics
 	PendingBlocksBytesCurrent float64
 	ChannelQueueLength        int
+	pendingDABytes            float64
 }
 
 var _ Metricer = new(TestMetrics)
 
 func (m *TestMetrics) RecordL2BlockInPendingQueue(block *types.Block) {
-	_, rawSize := estimateBatchSize(block)
+	daSize, rawSize := estimateBatchSize(block)
 	m.PendingBlocksBytesCurrent += float64(rawSize)
-
+	m.pendingDABytes += float64(daSize)
 }
 func (m *TestMetrics) RecordL2BlockInChannel(block *types.Block) {
-	_, rawSize := estimateBatchSize(block)
+	daSize, rawSize := estimateBatchSize(block)
 	m.PendingBlocksBytesCurrent -= float64(rawSize)
+	m.pendingDABytes -= float64(daSize)
 }
 func (m *TestMetrics) RecordChannelQueueLength(l int) {
 	m.ChannelQueueLength = l
 }
+func (m *TestMetrics) PendingDABytes() float64 {
+	return m.pendingDABytes
+}
+func (m *TestMetrics) ClearAllStateMetrics() {
+	m.PendingBlocksBytesCurrent = 0
+	m.ChannelQueueLength = 0
+}