op-batcher: Properly drain state on L2 reorgs

When a L2 reorg occurs the batcher should close out the pending channel
and submit everything that it has before clearing the channel builder.
The reason for this is that even though it may be submitting incorrect
data it should not leave dangling channels. After the pending state is
drained it can begin tracking the new L2 chain.

op-batcher/batcher/driver.go

@@ -187,13 +187,15 @@ func (l *BatchSubmitter) Stop(ctx context.Context) error {
 // 2. Check if the sync status is valid or if we are all the way up to date
 // 3. Check if it needs to initialize state OR it is lagging (todo: lagging just means race condition?)
 // 4. Load all new blocks into the local state.
-func (l *BatchSubmitter) loadBlocksIntoState(ctx context.Context) {
+// If there is a reorg, it will reset the last stored block but not clear the internal state so
+// the state can be flushed to L1.
+func (l *BatchSubmitter) loadBlocksIntoState(ctx context.Context) error {
 	start, end, err := l.calculateL2BlockRangeToStore(ctx)
 	if err != nil {
 		l.log.Warn("Error calculating L2 block range", "err", err)
-		return
+		return err
 	} else if start.Number >= end.Number {
-		return
+		return errors.New("start number is >= end number")
 	}
 
 	var latestBlock *types.Block
@@ -202,12 +204,11 @@ func (l *BatchSubmitter) loadBlocksIntoState(ctx context.Context) {
 		block, err := l.loadBlockIntoState(ctx, i)
 		if errors.Is(err, ErrReorg) {
 			l.log.Warn("Found L2 reorg", "block_number", i)
-			l.state.Clear()
 			l.lastStoredBlock = eth.BlockID{}
-			return
+			return err
 		} else if err != nil {
 			l.log.Warn("failed to load block into state", "err", err)
-			return
+			return err
 		}
 		l.lastStoredBlock = eth.ToBlockID(block)
 		latestBlock = block
@@ -216,10 +217,11 @@ func (l *BatchSubmitter) loadBlocksIntoState(ctx context.Context) {
 	l2ref, err := derive.L2BlockToBlockRef(latestBlock, &l.Rollup.Genesis)
 	if err != nil {
 		l.log.Warn("Invalid L2 block loaded into state", "err", err)
-		return
+		return err
 	}
 
 	l.metr.RecordL2BlocksLoaded(l2ref)
+	return nil
 }
 
 // loadBlockIntoState fetches & stores a single block into `state`. It returns the block it loaded.
@@ -294,7 +296,15 @@ func (l *BatchSubmitter) loop() {
 	for {
 		select {
 		case <-ticker.C:
-			l.loadBlocksIntoState(l.shutdownCtx)
+			if err := l.loadBlocksIntoState(l.shutdownCtx); errors.Is(err, ErrReorg) {
+				err := l.state.Close()
+				if err != nil {
+					l.log.Error("error closing the channel manager to handle a L2 reorg", "err", err)
+				}
+				l.publishStateToL1(queue, receiptsCh, true)
+				l.state.Clear()
+				continue
+			}
 			l.publishStateToL1(queue, receiptsCh, false)
 		case r := <-receiptsCh:
 			l.handleReceipt(r)