Immediately flush channel builder on channel manager close

op-batcher/batcher/channel_manager.go

@@ -187,8 +187,8 @@ func (s *channelManager) TxData(l1Head eth.BlockID) (txData, error) {
 	dataPending := s.pendingChannel != nil && s.pendingChannel.HasFrame()
 	s.log.Debug("Requested tx data", "l1Head", l1Head, "data_pending", dataPending, "blocks_pending", len(s.blocks))
 
-	// Short circuit if there is a pending frame.
-	if dataPending {
+	// Short circuit if there is a pending frame or the channel manager is closed.
+	if dataPending || s.closed {
 		return s.nextTxData()
 	}
 
@@ -203,11 +203,8 @@ func (s *channelManager) TxData(l1Head eth.BlockID) (txData, error) {
 		return txData{}, err
 	}
 
-	// Avoid processing blocks if the channel manager has been explicitly closed.
-	if !s.closed {
-		if err := s.processBlocks(); err != nil {
-			return txData{}, err
-		}
+	if err := s.processBlocks(); err != nil {
+		return txData{}, err
 	}
 
 	// Register current L1 head only after all pending blocks have been
@@ -356,14 +353,12 @@ func l2BlockRefFromBlockAndL1Info(block *types.Block, l1info derive.L1BlockInfo)
 	}
 }
 
-// Close closes the current pending channel, if one exists, and prevents the
-// creation of any new channels.
-// This ensures that no new blocks will be added to the channel, but there may be any number
-// of frames still produced by calling `OutputFrames()`, which flushes the compression buffer.
-// These frames still need to be published.
-func (s *channelManager) Close() {
+// Close closes the current pending channel, if one exists, outputs any remaining frames,
+// and prevents the creation of any new channels.
+// Any outputted frames still need to be published.
+func (s *channelManager) Close() error {
 	if s.closed {
-		return
+		return nil
 	}
 
 	s.closed = true
@@ -374,8 +369,10 @@ func (s *channelManager) Close() {
 	}
 
 	if s.pendingChannel == nil {
-		return
+		return nil
 	}
 
 	s.pendingChannel.Close()
+
+	return s.outputFrames()
 }

op-batcher/batcher/driver.go

@@ -317,9 +317,15 @@ func (l *BatchSubmitter) publishStateToL1(ctx context.Context) {
 		// produced. Any remaining frames must still be published to the L1 to prevent stalling.
 		select {
 		case <-ctx.Done():
-			l.state.Close()
+			err := l.state.Close()
+			if err != nil {
+				l.log.Error("error closing the channel manager", "err", err)
+			}
 		case <-l.shutdownCtx.Done():
-			l.state.Close()
+			err := l.state.Close()
+			if err != nil {
+				l.log.Error("error closing the channel manager", "err", err)
+			}
 		default:
 		}
 

op-e2e/migration_test.go

@@ -352,7 +352,9 @@ func TestMigration(t *testing.T) {
 	}, lgr.New("module", "batcher"), batchermetrics.NoopMetrics)
 	require.NoError(t, err)
 	t.Cleanup(func() {
-		batcher.StopIfRunning(context.Background())
+		ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+		defer cancel()
+		batcher.StopIfRunning(ctx)
 	})
 
 	proposer, err := l2os.NewL2OutputSubmitterFromCLIConfig(l2os.CLIConfig{