op-batcher: introduce moreComing and conditionally register L1 block

Author: nonsense

op-acceptance-tests/tests/batcher/batcher_test.go

@@ -49,6 +49,7 @@ func TestBatcherFullChannelsAfterDowntime(gt *testing.T) {
 			parent = sys.L2CL.HeadBlockRef(types.LocalUnsafe).Hash
 
 			sys.AdvanceTime(time.Second * 2)
+			time.Sleep(20 * time.Millisecond) // failed to force-include tx: type: 2 sender; err: nonce too high
 		}
 
 		l.Debug("Sequencing L1 block", "iteration_j", j)
@@ -68,7 +69,7 @@ func TestBatcherFullChannelsAfterDowntime(gt *testing.T) {
 			l.Info("Channel details", "channelID", c.String(), "frameCount", len(channelFrames[c]), "dataLength_frame0", len(channelFrames[c][0].Data))
 		}
 
-		require.Equal(t, 10, len(channels)) // we expect a total of 10 channels, due to existing MaxChannelDuration bug filed at: https://github.com/ethereum-optimism/optimism/issues/18092
+		require.Equal(t, 2, len(channels)) // we expect a total of 2 channels
 
 		// values are dependent on:
 		// - MaxPendingTransactions
@@ -79,16 +80,8 @@ func TestBatcherFullChannelsAfterDowntime(gt *testing.T) {
 			max  int
 			note string
 		}{
-			{min: 7_000, max: 10_000, note: "channel 0 - the first 100 blocks"},
-			{min: 100, max: 1000, note: "channel 1 - only a few blocks due to racy behavior, nowhere near the channel capacity"},
-			{min: 100, max: 1000, note: "channel 2 - only a few blocks due to racy behavior, nowhere near the channel capacity"},
-			{min: 100, max: 1000, note: "channel 3 - only a few blocks due to racy behavior, nowhere near the channel capacity"},
-			{min: 100, max: 1000, note: "channel 4 - only a few blocks due to racy behavior, nowhere near the channel capacity"},
-			{min: 100, max: 1000, note: "channel 5 - only a few blocks due to racy behavior, nowhere near the channel capacity"},
-			{min: 100, max: 1000, note: "channel 6 - only a few blocks due to racy behavior, nowhere near the channel capacity"},
-			{min: 100, max: 1000, note: "channel 7 - only a few blocks due to racy behavior, nowhere near the channel capacity"},
-			{min: 20_000, max: 40_000, note: "channel 8 - filled to the max capacity, due to blocking behavior because of MaxPendingTransactions limit reached"},
-			{min: 20_000, max: 40_000, note: "channel 9 - filled to the max capacity, due to blocking behavior because of MaxPendingTransactions limit reached"},
+			{min: 30_000, max: 40_000, note: "channel 0 - filled to the max capacity"},
+			{min: 30_000, max: 40_000, note: "channel 1 - remaining data, filling channel close to max capacity"},
 		}
 
 		for i, entry := range sizeRanges {

op-batcher/batcher/channel_manager.go

@@ -224,8 +224,8 @@ func (s *channelManager) nextTxData(channel *channel) (txData, error) {
 // It will decide whether to switch DA type automatically.
 // When switching DA type, the channelManager state will be rebuilt
 // with a new ChannelConfig.
-func (s *channelManager) TxData(l1Head eth.BlockID, isPectra bool, isThrottling bool, forcePublish pubInfo) (txData, error) {
-	channel, err := s.getReadyChannel(l1Head, forcePublish)
+func (s *channelManager) TxData(l1Head eth.BlockID, isPectra bool, isThrottling bool, pi pubInfo) (txData, error) {
+	channel, err := s.getReadyChannel(l1Head, pi)
 	if err != nil {
 		return emptyTxData, err
 	}
@@ -259,7 +259,7 @@ func (s *channelManager) TxData(l1Head eth.BlockID, isPectra bool, isThrottling
 	s.defaultCfg = newCfg
 
 	// Try again to get data to send on chain.
-	channel, err = s.getReadyChannel(l1Head, forcePublish)
+	channel, err = s.getReadyChannel(l1Head, pi)
 	if err != nil {
 		return emptyTxData, err
 	}
@@ -318,7 +318,12 @@ func (s *channelManager) getReadyChannel(l1Head eth.BlockID, pi pubInfo) (*chann
 		return nil, err
 	}
 
-	s.registerL1Block(l1Head)
+	if !pi.moreComing {
+		// Register current L1 head only after all pending blocks have been
+		// processed. Even if a timeout will be triggered now, it is better to have
+		// all pending blocks be included in this channel for submission.
+		s.registerL1Block(l1Head)
+	}
 
 	if err := s.outputFrames(); err != nil {
 		return nil, err

op-batcher/batcher/driver.go

@@ -501,7 +501,7 @@ func (l *BatchSubmitter) publishingLoop(ctx context.Context, wg *sync.WaitGroup,
 
 	for pi := range l.publishSignal {
 		l.Log.Debug("publishing loop received signal", "force_publish", pi.forcePublish)
-		l.publishStateToL1(ctx, txQueue, receiptsCh, daGroup, pi.forcePublish)
+		l.publishStateToL1(ctx, txQueue, receiptsCh, daGroup, pi)
 	}
 
 	// First wait for all DA requests to finish to prevent new transactions being queued
@@ -771,7 +771,7 @@ func (l *BatchSubmitter) waitNodeSync() error {
 
 // publishStateToL1 queues up all pending TxData to be published to the L1, returning when there is no more data to
 // queue for publishing or if there was an error queuing the data.
-func (l *BatchSubmitter) publishStateToL1(ctx context.Context, queue *txmgr.Queue[txRef], receiptsCh chan txmgr.TxReceipt[txRef], daGroup *errgroup.Group, forcePublish bool) {
+func (l *BatchSubmitter) publishStateToL1(ctx context.Context, queue *txmgr.Queue[txRef], receiptsCh chan txmgr.TxReceipt[txRef], daGroup *errgroup.Group, pi pubInfo) {
 	for {
 		select {
 		case <-ctx.Done():
@@ -788,7 +788,7 @@ func (l *BatchSubmitter) publishStateToL1(ctx context.Context, queue *txmgr.Queu
 			return
 		}
 
-		err := l.publishTxToL1(ctx, queue, receiptsCh, daGroup, forcePublish)
+		err := l.publishTxToL1(ctx, queue, receiptsCh, daGroup, pi)
 		if err != nil {
 			if err != io.EOF {
 				l.Log.Error("Error publishing tx to l1", "err", err)
@@ -842,7 +842,7 @@ func (l *BatchSubmitter) clearState(ctx context.Context) {
 }
 
 // publishTxToL1 submits a single state tx to the L1
-func (l *BatchSubmitter) publishTxToL1(ctx context.Context, queue *txmgr.Queue[txRef], receiptsCh chan txmgr.TxReceipt[txRef], daGroup *errgroup.Group, forcePublish bool) error {
+func (l *BatchSubmitter) publishTxToL1(ctx context.Context, queue *txmgr.Queue[txRef], receiptsCh chan txmgr.TxReceipt[txRef], daGroup *errgroup.Group, pi pubInfo) error {
 	// send all available transactions
 	l1tip, isPectra, err := l.l1Tip(ctx)
 	if err != nil {
@@ -855,7 +855,7 @@ func (l *BatchSubmitter) publishTxToL1(ctx context.Context, queue *txmgr.Queue[t
 	// Collect next transaction data. This pulls data out of the channel, so we need to make sure
 	// to put it back if ever da or txmgr requests fail, by calling l.recordFailedDARequest/recordFailedTx.
 	l.channelMgrMutex.Lock()
-	txdata, err := l.channelMgr.TxData(l1tip.ID(), isPectra, params.IsThrottling(), pubInfo{forcePublish: forcePublish})
+	txdata, err := l.channelMgr.TxData(l1tip.ID(), isPectra, params.IsThrottling(), pi)
 	l.channelMgrMutex.Unlock()
 
 	if err == io.EOF {