TXManager: add IsClosed to TxMgr and use check in BatchSubmitter (#9470)

* add IsClosed to TxMgr and use check in BatchSubmitter

* add generated mock

* move isClosed above publish in sending loop

* more IsClosed

op-batcher/batcher/driver.go

@@ -280,6 +280,12 @@ func (l *BatchSubmitter) loop() {
 		case r := <-receiptsCh:
 			l.handleReceipt(r)
 		case <-l.shutdownCtx.Done():
+			// if the txmgr is closed, we stop the transaction sending
+			// don't even bother draining the queue, as all sending will fail
+			if l.Txmgr.IsClosed() {
+				l.Log.Info("Txmgr is closed, remaining channel data won't be sent")
+				return
+			}
 			// This removes any never-submitted pending channels, so these do not have to be drained with transactions.
 			// Any remaining unfinished channel is terminated, so its data gets submitted.
 			err := l.state.Close()
@@ -304,13 +310,19 @@ func (l *BatchSubmitter) publishStateToL1(queue *txmgr.Queue[txData], receiptsCh
 	// send/wait and receipt reading must be on a separate goroutines to avoid deadlocks
 	go func() {
 		defer func() {
-			if drain {
-				// if draining, we wait for all transactions to complete
+			// if draining, we wait for all transactions to complete
+			// if the txmgr is closed, there is no need to wait as all transactions will fail
+			if drain && !l.Txmgr.IsClosed() {
 				queue.Wait()
 			}
 			close(txDone)
 		}()
 		for {
+			// if the txmgr is closed, we stop the transaction sending
+			if l.Txmgr.IsClosed() {
+				l.Log.Info("Txmgr is closed, no further receipts expected")
+				return
+			}
 			err := l.publishTxToL1(l.killCtx, queue, receiptsCh)
 			if err != nil {
 				if drain && err != io.EOF {

op-challenger/sender/sender_test.go

@@ -93,6 +93,10 @@ type stubTxMgr struct {
 	sending map[byte]chan *types.Receipt
 }
 
+func (s *stubTxMgr) IsClosed() bool {
+	return false
+}
+
 func (s *stubTxMgr) Send(ctx context.Context, candidate txmgr.TxCandidate) (*types.Receipt, error) {
 	ch := s.recordTx(candidate)
 	return <-ch, nil

op-e2e/actions/l2_proposer.go

@@ -68,6 +68,10 @@ func (f fakeTxMgr) Send(_ context.Context, _ txmgr.TxCandidate) (*types.Receipt,
 func (f fakeTxMgr) Close() {
 }
 
+func (f fakeTxMgr) IsClosed() bool {
+	return false
+}
+
 func NewL2Proposer(t Testing, log log.Logger, cfg *ProposerCfg, l1 *ethclient.Client, rollupCl *sources.RollupClient) *L2Proposer {
 	proposerConfig := proposer.ProposerConfig{
 		PollInterval:           time.Second,

op-service/txmgr/mocks/TxManager.go

@@ -64,6 +64,20 @@ func (_m *TxManager) From() common.Address {
 	return r0
 }
 
+// IsClosed provides a mock function with given fields:
+func (_m *TxManager) IsClosed() bool {
+	ret := _m.Called()
+
+	var r0 bool
+	if rf, ok := ret.Get(0).(func() bool); ok {
+		r0 = rf()
+	} else {
+		r0 = ret.Get(0).(bool)
+	}
+
+	return r0
+}
+
 // Send provides a mock function with given fields: ctx, candidate
 func (_m *TxManager) Send(ctx context.Context, candidate txmgr.TxCandidate) (*types.Receipt, error) {
 	ret := _m.Called(ctx, candidate)

op-service/txmgr/txmgr.go

@@ -74,6 +74,7 @@ type TxManager interface {
 
 	// Close the underlying connection
 	Close()
+	IsClosed() bool
 }
 
 // ETHBackend is the set of methods that the transaction manager uses to resubmit gas & determine
@@ -795,6 +796,11 @@ func (m *SimpleTxManager) checkBlobFeeLimits(blobBaseFee, bumpedBlobFee *big.Int
 	return nil
 }
 
+// IsClosed returns true if the tx manager is closed.
+func (m *SimpleTxManager) IsClosed() bool {
+	return m.closed.Load()
+}
+
 // calcThresholdValue returns ceil(x * priceBumpPercent / 100) for non-blob txs, or
 // ceil(x * blobPriceBumpPercent / 100) for blob txs.
 // It guarantees that x is increased by at least 1