fix(stage_batches): only unwind if the stream block is ahead

turbo/stages/stageloop.go

@@ -102,13 +102,9 @@ func StageLoop(
 			if errors.Is(err, libcommon.ErrStopped) || errors.Is(err, context.Canceled) {
 				return
 			}
-
 			if !errors.Is(err, zk.ErrLimboState) {
 				log.Error("Staged Sync", "err", err)
 			}
-			if recoveryErr := hd.RecoverFromDb(db); recoveryErr != nil {
-				log.Error("Failed to recover header sentriesClient", "err", recoveryErr)
-			}
 			time.Sleep(500 * time.Millisecond) // just to avoid too much similar errors in logs
 			continue
 		}

zk/stages/stage_batches.go

@@ -295,17 +295,19 @@ LOOP:
 
 				// skip if we already have this block
 				if entry.L2BlockNumber < lastBlockHeight+1 {
-					log.Warn(fmt.Sprintf("[%s] Unwinding to block %d", logPrefix, entry.L2BlockNumber))
+					log.Info(fmt.Sprintf("[%s] Skipping block %d, already processed", logPrefix, entry.L2BlockNumber))
+					continue
+				}
+
+				// if the block number isn't last + 1 then we've skipped AHEAD (see above logic to continue if we're behind)
+				if entry.L2BlockNumber != lastBlockHeight+1 {
+					log.Warn(fmt.Sprintf("[%s] Stream ahead of node, unwinding to block %d", logPrefix, entry.L2BlockNumber))
 					badBlock, err := eriDb.ReadCanonicalHash(entry.L2BlockNumber)
 					if err != nil {
 						return fmt.Errorf("failed to get bad block: %v", err)
 					}
 					u.UnwindTo(entry.L2BlockNumber, badBlock)
-				}
-
-				// check for sequential block numbers
-				if entry.L2BlockNumber != lastBlockHeight+1 {
-					return fmt.Errorf("block number is not sequential, expected %d, got %d", lastBlockHeight+1, entry.L2BlockNumber)
+					return nil
 				}
 
 				// batch boundary - record the highest hashable block number (last block in last full batch)
@@ -424,16 +426,19 @@ LOOP:
 				return err
 			}
 
-			if err := tx.Commit(); err != nil {
-				return fmt.Errorf("failed to commit tx, %w", err)
-			}
+			if freshTx {
+				if err := tx.Commit(); err != nil {
+					return fmt.Errorf("failed to commit tx, %w", err)
+				}
 
-			tx, err = cfg.db.BeginRw(ctx)
-			if err != nil {
-				return fmt.Errorf("failed to open tx, %w", err)
+				tx, err = cfg.db.BeginRw(ctx)
+				if err != nil {
+					return fmt.Errorf("failed to open tx, %w", err)
+				}
+				hermezDb = hermez_db.NewHermezDb(tx)
+				eriDb = erigon_db.NewErigonDb(tx)
 			}
-			hermezDb = hermez_db.NewHermezDb(tx)
-			eriDb = erigon_db.NewErigonDb(tx)
+
 			prevAmountBlocksWritten = blocksWritten
 		}