Merge pull request #5536 from ethereum-optimism/jg/handle_l2_reorgs_cleanly

op-batcher: Properly drain state on L2 reorgs

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)