restore pipelined fallback-auth submission

Author: lukeiannucci

op-batcher/batcher/driver.go

@@ -136,6 +136,15 @@ type BatchSubmitter struct {
 	throttleController *throttler.ThrottleController
 
 	publishSignal chan pubInfo
+
+	// authGroup tracks the fallback batcher's receipt-watcher goroutines (one
+	// per auth+batch pair) so the publishing loop can drain them via
+	// waitForAuthGroup before closing receiptsCh. New watchers are back-pressured
+	// (not hard-bounded) by the txmgr Queue: queue.Send blocks at
+	// MaxPendingTransactions, so watchers are created no faster than txs drain,
+	// though a slow receipts loop can briefly leave more than that parked on their
+	// final receiptsCh send.
+	authGroup sync.WaitGroup
 }
 
 // NewBatchSubmitter initializes the BatchSubmitter driver from a preconfigured DriverSetup
@@ -528,6 +537,12 @@ func (l *BatchSubmitter) publishingLoop(ctx context.Context, wg *sync.WaitGroup,
 		}
 	}
 
+	// Wait for all in-flight fallback-auth submissions to complete to prevent
+	// new transactions being queued. No-op when the rollup is not configured
+	// with a BatchAuthenticator or when the EspressoTime hardfork has not
+	// activated.
+	l.waitForAuthGroup()
+
 	// We _must_ wait for all senders on receiptsCh to finish before we can close it.
 	if err := txQueue.Wait(); err != nil {
 		if !errors.Is(err, context.Canceled) {

op-batcher/batcher/espresso_driver.go

@@ -6,11 +6,20 @@ import (
 	"github.com/ethereum-optimism/optimism/op-service/txmgr"
 )
 
+// waitForAuthGroup blocks until all in-flight fallback-auth watcher goroutines
+// have finished. publishingLoop calls it before closing receiptsCh: each watcher
+// is a sender on receiptsCh, so the receipts loop must still be draining
+// receiptsCh at this point or a watcher's final send would block forever. Each
+// watcher always terminates because the txmgr Queue emits exactly one receipt per
+// Send, even on context cancellation.
+func (l *BatchSubmitter) waitForAuthGroup() {
+	l.authGroup.Wait()
+}
+
 // dispatchAuthenticatedSendTx routes sendTx through the fallback-batcher
-// post-fork auth path, returning true when the tx has been fully handled
-// (sendTxWithFallbackAuth blocks until the pair is resolved and a receipt has
-// been forwarded). Returns false to mean "fall through to the upstream
-// queue.Send path" — pre-fork operation and any cancel tx.
+// post-fork auth path, returning true when the tx has been handed off to
+// authGroup. Returns false to mean "fall through to the upstream queue.Send
+// path" — pre-fork operation and any cancel tx.
 //
 // The fallback batcher consults isFallbackAuthRequired to gate authentication
 // behind the EspressoTime hardfork: pre-fork the verifier accepts plain

op-batcher/batcher/fallback_auth.go

@@ -86,17 +86,43 @@ func (l *BatchSubmitter) sendTxWithFallbackAuth(txdata txData, isCancel bool, ca
 		To:     &l.RollupConfig.BatchAuthenticatorAddress,
 	}
 
+	// Private buffered channels: queue.Send forwards exactly one receipt to each, so the watcher
+	// reads exactly once per channel (even on context cancellation, the queue still emits a
+	// ctx-error receipt). These never reach handleReceipt; only the synthetic receipt below does.
+	authReceiptCh := make(chan txmgr.TxReceipt[txRef], 1)
+	batchReceiptCh := make(chan txmgr.TxReceipt[txRef], 1)
+
 	l.Log.Debug(
 		"Sending fallback authenticateBatchInfo transaction",
 		"txRef", transactionReference,
 		"commitment", hexutil.Encode(commitment[:]),
 		"address", l.RollupConfig.BatchAuthenticatorAddress.String(),
 	)
-	// Submit the auth tx and wait for its receipt, then send the batch tx. Each Send
-	// blocks here when the queue is at its MaxPendingTransactions limit.
-	authReceiptCh := make(chan txmgr.TxReceipt[txRef], 1)
+	// Submit the auth tx then the batch tx, in order, on the publishing-loop goroutine so their
+	// nonces are assigned in submission order (auth = N, batch = N+1), guaranteeing the auth tx
+	// precedes the batch tx on chain. Each Send blocks here when the queue is at its
+	// MaxPendingTransactions limit, so the pair pipelines with subsequent pairs rather than
+	// stalling the loop for a full confirmation cycle. The auth tx uses authReference so its
+	// failures are typed as calldata (see above).
 	queue.Send(authReference, verifyCandidate, authReceiptCh)
+	queue.Send(transactionReference, *candidate, batchReceiptCh)
+
+	l.authGroup.Add(1)
+	go func() {
+		defer l.authGroup.Done()
+		l.watchFallbackAuthReceipts(transactionReference, authReference, authReceiptCh, batchReceiptCh, receiptsCh)
+	}()
+}
+
+// watchFallbackAuthReceipts collects the auth and batch receipts for a fallback-auth pair,
+// validates that the batch tx landed within the lookback window of the auth tx, and forwards a
+// single synthetic receipt keyed to the batch txData onto receiptsCh. Any failure produces an
+// error receipt so the channel manager rewinds and resubmits the frame set. Auth failures are
+// reported under authReference so cancelBlockingTx targets the calldata pool, while the batch
+// txData id is preserved so the right frames are requeued.
+func (l *BatchSubmitter) watchFallbackAuthReceipts(transactionReference, authReference txRef, authReceiptCh, batchReceiptCh chan txmgr.TxReceipt[txRef], receiptsCh chan txmgr.TxReceipt[txRef]) {
 	authResult := <-authReceiptCh
+	batchResult := <-batchReceiptCh
 
 	if authResult.Err != nil {
 		l.Log.Error("Failed to send fallback authenticateBatchInfo transaction", "txRef", transactionReference, "err", authResult.Err)
@@ -121,11 +147,6 @@ func (l *BatchSubmitter) sendTxWithFallbackAuth(txdata txData, isCancel bool, ca
 		return
 	}
 
-	// The auth tx is confirmed on L1. Now send the batch tx
-	batchReceiptCh := make(chan txmgr.TxReceipt[txRef], 1)
-	queue.Send(transactionReference, *candidate, batchReceiptCh)
-	batchResult := <-batchReceiptCh
-
 	if batchResult.Err != nil {
 		l.Log.Error("Failed to send batch inbox transaction", "txRef", transactionReference, "err", batchResult.Err)
 		receiptsCh <- txmgr.TxReceipt[txRef]{

op-batcher/batcher/fallback_auth_test.go

@@ -104,14 +104,20 @@ func TestFallbackAuth_OrderingAndSuccess(t *testing.T) {
 	require.Equal(t, txdata.ID().String(), got.ID.id.String())
 }
 
+// TestFallbackAuth_AuthFailureRetried verifies that an auth-tx send failure produces an error
+// receipt keyed to the batch txData so the frames are re-queued. The pair is submitted
+// back-to-back (pipelined), so the batch tx is also sent; if its auth never lands the batch is
+// simply dropped by derivation — its commitment has no matching auth event in the lookback
+// window — so the orphaned send is harmless to safety.
 func TestFallbackAuth_AuthFailureRetried(t *testing.T) {
 	l := newFallbackAuthSubmitter(t)
 	txdata := testFallbackTxData(t)
 	candidate := &txmgr.TxCandidate{TxData: []byte("batch-calldata")}
 
 	queue := &fakeTxSender{
 		responses: []txmgr.TxReceipt[txRef]{
-			{Err: errSendFailed}, // auth fails. No batch response, it must never be sent
+			{Err: errSendFailed},             // auth fails
+			{Receipt: receiptWithBlock(101)}, // batch send (result discarded on auth failure)
 		},
 	}
 	receiptsCh := make(chan txmgr.TxReceipt[txRef], 1)
@@ -121,9 +127,8 @@ func TestFallbackAuth_AuthFailureRetried(t *testing.T) {
 	got := <-receiptsCh
 	require.Error(t, got.Err)
 	require.Equal(t, txdata.ID().String(), got.ID.id.String())
-	// The batch tx must not be sent after an auth failure: it would be crafted at the
-	// next nonce while the failed auth tx resets the txmgr nonce, leaving a nonce gap.
-	require.Len(t, queue.sends, 1)
+	// Both txs are submitted back-to-back; the auth failure does not gate the batch send.
+	require.Len(t, queue.sends, 2)
 }
 
 // TestFallbackAuth_AuthFailureTxRefType verifies that an auth-tx failure is
@@ -140,7 +145,8 @@ func TestFallbackAuth_AuthFailureTxRefType(t *testing.T) {
 
 	queue := &fakeTxSender{
 		responses: []txmgr.TxReceipt[txRef]{
-			{Err: errSendFailed}, // auth fails. No batch response, it must never be sent
+			{Err: errSendFailed},             // auth fails
+			{Receipt: receiptWithBlock(101)}, // batch send (result discarded on auth failure)
 		},
 	}
 	receiptsCh := make(chan txmgr.TxReceipt[txRef], 1)
@@ -150,7 +156,7 @@ func TestFallbackAuth_AuthFailureTxRefType(t *testing.T) {
 	got := <-receiptsCh
 	require.Error(t, got.Err)
 	require.Equal(t, txdata.ID().String(), got.ID.id.String())
-	require.Len(t, queue.sends, 1)
+	require.Len(t, queue.sends, 2)
 	require.False(t, got.ID.isBlob)
 	require.Equal(t, DaTypeCalldata, got.ID.daType)
 }
@@ -211,7 +217,8 @@ func TestFallbackAuth_AuthRevertedRetried(t *testing.T) {
 
 	queue := &fakeTxSender{
 		responses: []txmgr.TxReceipt[txRef]{
-			{Receipt: revertedReceiptWithBlock(100)}, // auth mined but reverted. Batch must never be sent
+			{Receipt: revertedReceiptWithBlock(100)}, // auth mined but reverted
+			{Receipt: receiptWithBlock(101)},         // batch send (result discarded on auth revert)
 		},
 	}
 	receiptsCh := make(chan txmgr.TxReceipt[txRef], 1)
@@ -221,9 +228,10 @@ func TestFallbackAuth_AuthRevertedRetried(t *testing.T) {
 	got := <-receiptsCh
 	require.Error(t, got.Err)
 	require.Equal(t, txdata.ID().String(), got.ID.id.String())
-	// A reverted auth tx emits no BatchInfoAuthenticated event, so the batch data would be
-	// unverifiable so the batch tx must not be submitted at all.
-	require.Len(t, queue.sends, 1)
+	// A reverted auth tx emits no BatchInfoAuthenticated event, so the batch is unverifiable and
+	// an error receipt is forwarded to re-queue the frames. Both txs are still submitted
+	// (pipelined); the orphaned batch is dropped by derivation for lack of a matching auth event.
+	require.Len(t, queue.sends, 2)
 }
 
 // TestFallbackAuth_WindowViolationRetried verifies that a batch tx landing