Cleanup

op-batcher/batcher/driver.go

@@ -300,6 +300,14 @@ func (l *BatchSubmitter) loop() {
 	publishTicker := time.NewTicker(100 * time.Millisecond)
 	defer publishTicker.Stop()
 	receiptsCh := make(chan txReceipt)
+	receiptHandler := func(r txReceipt) {
+		// Record TX Status
+		if r.err != nil {
+			l.recordFailedTx(r.id, r.err)
+		} else {
+			l.recordConfirmedTx(r.id, r.receipt)
+		}
+	}
 
 	for {
 		select {
@@ -307,21 +315,16 @@ func (l *BatchSubmitter) loop() {
 			l.loadBlocksIntoState(l.shutdownCtx)
 		case <-publishTicker.C:
 			_ = l.publishStateToL1(l.killCtx, receiptsCh)
-		case res := <-receiptsCh:
-			// Record TX Status
-			if res.err != nil {
-				l.recordFailedTx(res.id, res.err)
-			} else {
-				l.recordConfirmedTx(res.id, res.receipt)
-			}
+		case r := <-receiptsCh:
+			receiptHandler(r)
 		case <-l.shutdownCtx.Done():
-			l.drainState(receiptsCh)
+			l.drainState(receiptsCh, receiptHandler)
 			return
 		}
 	}
 }
 
-func (l *BatchSubmitter) drainState(receiptsCh chan txReceipt) {
+func (l *BatchSubmitter) drainState(receiptsCh chan txReceipt, receiptHandler func(res txReceipt)) {
 	err := l.state.Close()
 	if err != nil {
 		l.log.Error("error closing the channel manager", "err", err)
@@ -343,27 +346,23 @@ func (l *BatchSubmitter) drainState(receiptsCh chan txReceipt) {
 			}
 		}
 	}()
-	var receipts []txReceipt
-	receiptsDone := make(chan struct{})
+
+	txDone := make(chan struct{})
 	go func() {
-		for {
-			select {
-			case res := <-receiptsCh:
-				receipts = append(receipts, res)
-			case <-receiptsDone:
-				return
-			}
-		}
+		// wait for all transactions to complete
+		l.txWg.Wait()
+		close(txDone)
 	}()
-	// wait for all transactions to complete
-	l.txWg.Wait()
-	close(receiptsDone)
-	// process the receipts
-	for _, res := range receipts {
-		if res.err != nil {
-			l.recordFailedTx(res.id, res.err)
-		} else {
-			l.recordConfirmedTx(res.id, res.receipt)
+	// handle the remaining receipts
+	for {
+		select {
+		case r := <-receiptsCh:
+			receiptHandler(r)
+		case <-txDone:
+			for len(receiptsCh) > 0 {
+				receiptHandler(<-receiptsCh)
+			}
+			return
 		}
 	}
 }

op-service/txmgr/txmgr.go

@@ -159,7 +159,7 @@ func (m *SimpleTxManager) reset() {
 		return
 	}
 
-	close(m.resetChannel())
+	close(m.getResetChannel())
 	m.wg.Wait()
 
 	m.lock.Lock()
@@ -170,9 +170,9 @@ func (m *SimpleTxManager) reset() {
 	m.resetting.Store(false)
 }
 
-// resetChannel is a thread-safe getter for the channel that is closed upon
+// getResetChannel is a thread-safe getter for the channel that is closed upon
 // transaction manager resets.
-func (m *SimpleTxManager) resetChannel() chan struct{} {
+func (m *SimpleTxManager) getResetChannel() chan struct{} {
 	m.lock.RLock()
 	defer m.lock.RUnlock()
 	return m.resetC
@@ -294,7 +294,7 @@ func (m *SimpleTxManager) sendTx(ctx context.Context, tx *types.Transaction) (*t
 
 	ticker := time.NewTicker(m.cfg.ResubmissionTimeout)
 	defer ticker.Stop()
-	resetChan := m.resetChannel()
+	resetChan := m.getResetChannel()
 
 	bumpCounter := 0
 	for {