op-batcher: introduce moreComing and conditionally register L1 block

Author: nonsense

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

@@ -37,12 +37,14 @@ func TestBatcherFullChannelsAfterDowntime(gt *testing.T) {
 	l.Info("Latest unsafe block after stopping the L2 sequencer", "latestUnsafe", latestUnsafe_A)
 
 	parent := latestUnsafe_A
+	nonce := uint64(0)
 	for j := 0; j < 200; j++ {
 		l1Origin := sys.L1EL.BlockRefByLabel(eth.Unsafe).Hash
 
 		for i := 0; i < 5; i++ {
 			l.Debug("Sequencing L2 block", "iteration", i, "parent", parent)
-			sequenceBlockWithL1Origin(t, ts_L2, parent, l1Origin, alice, cathrine)
+			sequenceBlockWithL1Origin(t, ts_L2, parent, l1Origin, alice, cathrine, nonce)
+			nonce++
 
 			parent = sys.L2CL.HeadBlockRef(types.LocalUnsafe).Hash
 
@@ -66,7 +68,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
@@ -77,16 +79,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 {
@@ -109,13 +103,13 @@ func sequenceBlock(t devtest.T, ts apis.TestSequencerControlAPI, parent common.H
 	require.NoError(t, ts.Next(t.Ctx()))
 }
 
-func sequenceBlockWithL1Origin(t devtest.T, ts apis.TestSequencerControlAPI, parent common.Hash, l1Origin common.Hash, alice *dsl.EOA, cathrine *dsl.EOA) {
+func sequenceBlockWithL1Origin(t devtest.T, ts apis.TestSequencerControlAPI, parent common.Hash, l1Origin common.Hash, alice *dsl.EOA, cathrine *dsl.EOA, nonce uint64) {
 	require.NoError(t, ts.New(t.Ctx(), seqtypes.BuildOpts{Parent: parent, L1Origin: &l1Origin}))
 
 	// include simple transfer tx in opened block
 	{
 		to := cathrine.PlanTransfer(alice.Address(), eth.OneWei)
-		opt := txplan.Combine(to)
+		opt := txplan.Combine(to, txplan.WithStaticNonce(nonce))
 		ptx := txplan.NewPlannedTx(opt)
 		signed_tx, err := ptx.Signed.Eval(t.Ctx())
 		require.NoError(t, err, "Expected to be able to evaluate a planned transaction on op-test-sequencer, but got error")

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

@@ -512,7 +512,7 @@ func (l *BatchSubmitter) publishingLoop(ctx context.Context) {
 
 	for pi := range l.publishSignal {
 		l.Log.Debug("publishing loop received signal", "force_publish", pi.forcePublish)
-		l.publishStateToL1(ctx, txQueue, daGroup, pi.forcePublish)
+		l.publishStateToL1(ctx, txQueue, daGroup, pi)
 	}
 
 	// First wait for all DA requests to finish to prevent new transactions being queued
@@ -779,7 +779,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], daGroup *errgroup.Group, forcePublish bool) {
+func (l *BatchSubmitter) publishStateToL1(ctx context.Context, queue *txmgr.Queue[txRef], daGroup *errgroup.Group, pi pubInfo) {
 	for {
 		select {
 		case <-ctx.Done():
@@ -796,7 +796,7 @@ func (l *BatchSubmitter) publishStateToL1(ctx context.Context, queue *txmgr.Queu
 			return
 		}
 
-		err := l.publishTxToL1(ctx, queue, daGroup, forcePublish)
+		err := l.publishTxToL1(ctx, queue, daGroup, pi)
 		if err != nil {
 			if err != io.EOF {
 				l.Log.Error("Error publishing tx to l1", "err", err)
@@ -850,7 +850,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], daGroup *errgroup.Group, forcePublish bool) error {
+func (l *BatchSubmitter) publishTxToL1(ctx context.Context, queue *txmgr.Queue[txRef], daGroup *errgroup.Group, pi pubInfo) error {
 	// send all available transactions
 	l1tip, isPectra, err := l.l1Tip(ctx)
 	if err != nil {
@@ -863,7 +863,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 {