sync: recover from stale fork

pkgs/node/src/blocks_by_range_sync.zig

@@ -129,6 +129,52 @@ pub fn shouldCatchUpFromPeerStatus(
     return cappedSyncGapSlots(peer_head_slot, our_head_slot, wall_slot) > 0;
 }
 
+/// Recovery start for a `blocks_by_range` first-chunk fork mismatch.
+///
+/// A normal range starts at `our_head_slot + 1`, so the first block must extend
+/// our head-at-request. If the peer is on a heavier sibling fork, that check
+/// fails. Recovery can still use range sync when we have a recent common-ish
+/// anchor such as latest justified: request from `anchor_slot + 1` and validate
+/// the first chunk against `anchor_root` instead of the stale head.
+///
+/// Returns null when the anchor cannot improve on the failed request or is
+/// likely outside the peer's advertised recent-history window; callers should
+/// then fall back to the by-root parent walk from peer head.
+pub fn forkMismatchRecoveryStart(
+    failed_start_slot: types.Slot,
+    anchor_slot: types.Slot,
+    peer_head_slot: types.Slot,
+    min_slots_for_block_requests: u64,
+) ?types.Slot {
+    const recovery_start = anchor_slot +| 1;
+    if (recovery_start >= failed_start_slot) return null;
+    if (recovery_start > peer_head_slot) return null;
+
+    if (peer_head_slot >= min_slots_for_block_requests) {
+        const history_start = peer_head_slot - min_slots_for_block_requests;
+        if (recovery_start < history_start) return null;
+    }
+
+    return recovery_start;
+}
+
+pub fn shouldAttemptForkMismatchRangeRecovery(range_attempt: u8, max_attempts: u8) bool {
+    return range_attempt < max_attempts;
+}
+
+/// Proposal liveness guard for nodes that look "synced" by finalized-slot
+/// status but are clearly stale by wall-clock head lag.
+pub fn shouldSuppressProposalForHeadLag(
+    wall_head_lag_slots: u64,
+    max_proposal_head_lag_slots: u64,
+    latest_justified_slot: types.Slot,
+    has_fresher_peer_near_wall: bool,
+) bool {
+    if (latest_justified_slot == 0) return false; // preserve pre-justification cold start
+    if (!has_fresher_peer_near_wall) return false;
+    return wall_head_lag_slots > max_proposal_head_lag_slots;
+}
+
 /// Pure decider for the "stuck mesh cluster" recovery path.
 ///
 /// Fires when ALL of:
@@ -359,6 +405,43 @@ test "shouldCatchUpFromPeerStatus small gaps use by-root not threshold gate" {
     try std.testing.expect(shouldCatchUpFromPeerStatus(small_gap_peer, 0, 0, 0, 100));
 }
 
+test "forkMismatchRecoveryStart re-anchors to a recent justified ancestor" {
+    try std.testing.expectEqual(
+        @as(?types.Slot, 9340),
+        forkMismatchRecoveryStart(9649, 9339, 14735, constants.MIN_SLOTS_FOR_BLOCK_REQUESTS),
+    );
+}
+
+test "forkMismatchRecoveryStart falls back when anchor cannot improve failed range" {
+    try std.testing.expectEqual(
+        @as(?types.Slot, null),
+        forkMismatchRecoveryStart(9340, 9339, 14735, constants.MIN_SLOTS_FOR_BLOCK_REQUESTS),
+    );
+}
+
+test "forkMismatchRecoveryStart falls back when anchor is outside peer range history" {
+    try std.testing.expectEqual(
+        @as(?types.Slot, null),
+        forkMismatchRecoveryStart(15169, 9337, 15169, constants.MIN_SLOTS_FOR_BLOCK_REQUESTS),
+    );
+}
+
+test "shouldAttemptForkMismatchRangeRecovery stops before u8 overflow" {
+    try std.testing.expect(shouldAttemptForkMismatchRangeRecovery(1, constants.MAX_BLOCKS_BY_RANGE_SYNC_ATTEMPTS));
+    try std.testing.expect(!shouldAttemptForkMismatchRangeRecovery(
+        constants.MAX_BLOCKS_BY_RANGE_SYNC_ATTEMPTS,
+        constants.MAX_BLOCKS_BY_RANGE_SYNC_ATTEMPTS,
+    ));
+    try std.testing.expect(!shouldAttemptForkMismatchRangeRecovery(255, 255));
+}
+
+test "shouldSuppressProposalForHeadLag preserves cold start and blocks stale justified forks" {
+    try std.testing.expect(!shouldSuppressProposalForHeadLag(100, 4, 0, true));
+    try std.testing.expect(!shouldSuppressProposalForHeadLag(4, 4, 1, true));
+    try std.testing.expect(!shouldSuppressProposalForHeadLag(100, 4, 1, false));
+    try std.testing.expect(shouldSuppressProposalForHeadLag(5, 4, 1, true));
+}
+
 test "shouldForceFullPeerStatusRefresh fires when stuck behind a cluster of stale peers" {
     // The scenario the helper exists to recover from: wall is at
     // slot 300, best peer head zeam knows about is slot 50, no force-refresh has

pkgs/node/src/chain.zig

@@ -37,6 +37,7 @@ const RcBeamState = rc_beam_state.RcBeamState;
 const chain_worker = @import("./chain_worker.zig");
 const invalid_block_cache = @import("./invalid_block_cache.zig");
 const InvalidBlockSet = invalid_block_cache.InvalidBlockSet;
+const blocks_by_range_sync = @import("./blocks_by_range_sync.zig");
 
 /// Bound on the in-memory invalid-block-roots cache (see
 /// `BeamChain.invalid_block_roots`). 10 000 32-byte roots ≈ 320 KB plus
@@ -5272,6 +5273,21 @@ pub const BeamChain = struct {
     /// publishBlock) runs on a `thread_pool` worker so the multi-second
     /// prod-scheme merge never freezes the slot loop. At most one propose is in
     /// flight; a second trigger is dropped (the next slot re-proposes).
+    fn hasFresherPeerNearWall(self: *Self, our_head_slot: types.Slot, wall_head_lag_slots: u64) bool {
+        const wall_slot = our_head_slot +| wall_head_lag_slots;
+        var peer_guard = self.connected_peers.iterateLocked();
+        defer peer_guard.deinit();
+        var peer_iter = peer_guard.iter;
+        while (peer_iter.next()) |entry| {
+            const status = entry.value_ptr.latest_status orelse continue;
+            if (status.head_slot <= our_head_slot) continue;
+            if (status.head_slot +| constants.BLOCK_PROPOSAL_MAX_HEAD_LAG_SLOTS >= wall_slot) {
+                return true;
+            }
+        }
+        return false;
+    }
+
     pub fn submitPropose(self: *Self, node: *@import("./node.zig").BeamNode, slot: usize, proposer_id: usize) void {
         // Sync gating (the checks the old on-loop maybeDoProposal performed).
         switch (self.getSyncStatus()) {
@@ -5303,6 +5319,22 @@ pub const BeamChain = struct {
             },
         }
 
+        const wall_head_lag = self.wall_head_lag_slots.load(.monotonic);
+        const latest_justified_slot = self.forkChoice.getLatestJustified().slot;
+        const head = self.forkChoice.getHead();
+        if (blocks_by_range_sync.shouldSuppressProposalForHeadLag(
+            wall_head_lag,
+            constants.BLOCK_PROPOSAL_MAX_HEAD_LAG_SLOTS,
+            latest_justified_slot,
+            self.hasFresherPeerNearWall(head.slot, wall_head_lag),
+        )) {
+            self.logger.warn(
+                "skipping block production for slot={d} proposer={d}: local head is {d} wall-clock slots behind (head_slot={d}, latest_justified_slot={d})",
+                .{ slot, proposer_id, wall_head_lag, head.slot, latest_justified_slot },
+            );
+            return;
+        }
+
         // Single-flight: at most one propose at a time. fetchAdd-then-compare is a soft ceiling,
         // released on every early return below.
         const prev = self.propose_inflight.fetchAdd(1, .acq_rel);

pkgs/node/src/constants.zig

@@ -181,6 +181,12 @@ pub fn gossipStallThresholdMs() u64 {
 // batched `blocks_by_range` request, so anything beyond a few slots prefers range sync.
 pub const BLOCKS_BY_RANGE_SYNC_THRESHOLD: u64 = 4;
 
+// Refuse to produce a block when our local head is this many wall-clock slots
+// behind, once the chain has reached its first justification. This prevents a
+// recovered/stale minority fork from minting current-slot blocks on an ancient
+// parent while sync recovery is trying to import the heavier branch.
+pub const BLOCK_PROPOSAL_MAX_HEAD_LAG_SLOTS: u64 = 4;
+
 // Maximum `blocks_by_range` catch-up attempts (peer rotation + fallback) before
 // switching to head-by-root parent walk.
 pub const MAX_BLOCKS_BY_RANGE_SYNC_ATTEMPTS: u8 = 3;

pkgs/node/src/node.zig

@@ -1718,6 +1718,50 @@ pub const BeamNode = struct {
         };
     }
 
+    fn initiateForkMismatchRangeRecovery(
+        self: *Self,
+        snap: networkFactory.Network.PendingRequestSnapshot,
+    ) bool {
+        if (!blocks_by_range_sync.shouldAttemptForkMismatchRangeRecovery(
+            snap.range_attempt,
+            constants.MAX_BLOCKS_BY_RANGE_SYNC_ATTEMPTS,
+        )) return false;
+
+        const anchor = self.chain.forkChoice.getLatestJustified();
+        const recovery_start = blocks_by_range_sync.forkMismatchRecoveryStart(
+            snap.start_slot,
+            anchor.slot,
+            snap.peer_head_slot,
+            constants.MIN_SLOTS_FOR_BLOCK_REQUESTS,
+        ) orelse return false;
+
+        if (!self.network.peerSupportsBlocksByRange(snap.peer_id_copy)) return false;
+
+        const remaining = snap.peer_head_slot - recovery_start + 1;
+        const requested_count: u64 = @min(remaining, params.MAX_REQUEST_BLOCKS);
+        self.logger.warn(
+            "blocks_by_range fork recovery: re-anchoring peer {s}{f} from justified slot={d} start_slot={d} count={d} after failed start_slot={d}",
+            .{
+                snap.peer_id_copy,
+                self.node_registry.getNodeNameFromPeerId(snap.peer_id_copy),
+                anchor.slot,
+                recovery_start,
+                requested_count,
+                snap.start_slot,
+            },
+        );
+        self.initiateBlocksByRangeCatchUp(.{
+            .peer_id = snap.peer_id_copy,
+            .start_slot = recovery_start,
+            .count = requested_count,
+            .peer_head_slot = snap.peer_head_slot,
+            .peer_head_root = snap.peer_head_root,
+            .our_head_root_at_start = anchor.root,
+            .attempt = snap.range_attempt + 1,
+        });
+        return true;
+    }
+
     /// `blocks_by_root` catch-up: fetch the peer head and walk parents via the existing
     /// batched parent-fetch path (used when `blocks_by_range` is unavailable or gap is small).
     fn initiateCatchUpViaBlocksByRoot(self: *Self, status: CatchUpPeerStatus, our_head_slot: types.Slot) void {
@@ -1922,7 +1966,9 @@ pub const BeamNode = struct {
                     },
                 );
                 self.network.finalizePendingRequest(request_id);
-                self.syncFetchPeerHeadByRoot(snap.peer_id_copy, snap.peer_head_root);
+                if (action != .abort_fallback or !self.initiateForkMismatchRangeRecovery(snap)) {
+                    self.syncFetchPeerHeadByRoot(snap.peer_id_copy, snap.peer_head_root);
+                }
             },
             .pre_finalized_complete => {
                 self.recordRangeSyncOutcome("pre_finalized_noop");
@@ -2027,7 +2073,7 @@ pub const BeamNode = struct {
             !std.mem.eql(u8, &signed_block.block.parent_root, &view.our_head_root_at_start))
         {
             self.logger.warn(
-                "blocks_by_range: fork mismatch on first chunk slot={d} start_slot={d} (parent 0x{x} != our head-at-start 0x{x}); aborting range batch",
+                "blocks_by_range: fork mismatch on first chunk slot={d} start_slot={d} (parent 0x{x} != expected anchor 0x{x}); aborting range batch",
                 .{
                     signed_block.block.slot,
                     view.start_slot,