diff --git a/pkgs/node/src/chain.zig b/pkgs/node/src/chain.zig index e62124636..1ba11fd83 100644 --- a/pkgs/node/src/chain.zig +++ b/pkgs/node/src/chain.zig @@ -158,11 +158,6 @@ pub const BeamChain = struct { prune_cached_blocks_fn: ?PruneCachedBlocksFn = null, // Queue for blocks that arrived before forkchoice had ticked to their slot. - // When a peer gossips a block for the current slot before our local interval - // timer fires, the forkchoice rejects it with FutureSlot. We hold such - // blocks here and replay them in onInterval once the clock has caught up. - pending_blocks: std.ArrayList(types.SignedBlock), - pub const PruneCachedBlocksFn = *const fn (ptr: *anyopaque, finalized: types.Checkpoint) usize; const Self = @This(); @@ -209,7 +204,6 @@ pub const BeamChain = struct { .is_aggregator_enabled = std.atomic.Value(bool).init(opts.is_aggregator), .public_key_cache = xmss.PublicKeyCache.init(allocator), .root_to_slot_cache = types.RootToSlotCache.init(allocator), - .pending_blocks = .empty, .thread_pool = opts.thread_pool, }; // Initialize cache with anchor block root and any post-finalized entries from state @@ -245,11 +239,6 @@ pub const BeamChain = struct { // Clean up root to slot cache self.root_to_slot_cache.deinit(); - // Clean up any blocks that were queued waiting for the forkchoice clock - for (self.pending_blocks.items) |*block| { - block.deinit(); - } - self.pending_blocks.deinit(self.allocator); // assume the allocator of config is same as self.allocator self.config.deinit(self.allocator); @@ -285,51 +274,6 @@ pub const BeamChain = struct { zeam_metrics.metrics.lean_validators_count.set(self.registered_validator_ids.len); } - /// Replay blocks that were queued because the forkchoice clock hadn't yet - /// reached their slot. Called from onInterval after advancing the clock. - /// Returns a slice of all missing attestation roots encountered while - /// processing queued blocks; the caller owns and must free the slice. - pub fn processPendingBlocks(self: *Self) []types.Root { - var all_missing_roots: std.ArrayListUnmanaged(types.Root) = .empty; - const fc_time = self.forkChoice.fcStore.slot_clock.time.load(.monotonic); - var i: usize = 0; - while (i < self.pending_blocks.items.len) { - const queued_slot = self.pending_blocks.items[i].block.slot; - if (queued_slot * constants.INTERVALS_PER_SLOT <= fc_time) { - // Remove from queue (ownership transferred to local var). - var queued_block = self.pending_blocks.orderedRemove(i); - defer queued_block.deinit(); - - var block_root: types.Root = undefined; - zeam_utils.hashTreeRoot(types.BeamBlock, queued_block.block, &block_root, self.allocator) catch |err| { - self.logger.err("queued block slot={d}: failed to compute block root: {any}", .{ queued_slot, err }); - continue; - }; - - self.logger.info( - "replaying queued block slot={d} blockroot=0x{x} (fc_time now={d})", - .{ queued_slot, &block_root, fc_time }, - ); - - const missing_roots = self.onBlock(queued_block, .{ - .blockRoot = block_root, - }) catch |err| { - self.logger.err("queued block slot={d} root=0x{x}: processing failed: {any}", .{ queued_slot, &block_root, err }); - continue; - }; - defer self.allocator.free(missing_roots); - - self.onBlockFollowup(true, &queued_block); - - // Accumulate missing roots so the caller can fetch them. - all_missing_roots.appendSlice(self.allocator, missing_roots) catch {}; - } else { - i += 1; - } - } - return all_missing_roots.toOwnedSlice(self.allocator) catch &.{}; - } - pub fn onInterval(self: *Self, time_intervals: usize) !void { // see if the node has a proposal this slot to properly tick // forkchoice head @@ -549,13 +493,7 @@ pub const BeamChain = struct { } }; - // 4. Advance fork choice to this block's slot so the block is not rejected as FutureSlot - // PS: this isn't required because forkchoice is already ticked before validator's oninterval is called - // which then leads to block production call - // - // try self.forkChoice.onInterval(block.slot * constants.INTERVALS_PER_SLOT, false); - - // 5. Add the block to directly forkchoice as this proposer will next need to construct its vote + // 4. Add the block directly to forkchoice as this proposer will next need to construct its vote // note - attestations packed in the block are already in the knownVotes so we don't need to re-import // them in the forkchoice _ = try self.forkChoice.onBlock(block, post_state, .{ @@ -717,49 +655,14 @@ pub const BeamChain = struct { }); if (!hasBlock) { - // Validation errors propagate to node.zig for context-aware logging + // Validation errors propagate to node.zig for context-aware logging. + // validateBlock checks parent / signature / STF preconditions; STF + // and signature verification are the gating layers for any block. try self.validateBlock(block, true); - // If the forkchoice clock hasn't yet ticked to this block's slot, - // onBlock would reject it with FutureSlot. Queue the block and - // replay it from onInterval once the clock has advanced. - if (block.slot * constants.INTERVALS_PER_SLOT > self.forkChoice.fcStore.slot_clock.time.load(.monotonic)) { - self.logger.debug( - "queuing gossip block slot={d} blockroot=0x{x}: forkchoice time={d} < slot_start={d}", - .{ block.slot, &block_root, self.forkChoice.fcStore.slot_clock.time.load(.monotonic), block.slot * constants.INTERVALS_PER_SLOT }, - ); - var cloned: types.SignedBlock = undefined; - try types.sszClone(self.allocator, types.SignedBlock, signed_block, &cloned); - - // TODO: in beam sim, it seems to have queued after the oninterval fires even if block arrives pre on interval - // because of race conditions between competing threads as the above sszClone aparently takes too much time - // currently managing this by checking condition again but ideally fix it by identifying chain entrypoints and - // holding mutex between then for chain modification sections - if (block.slot * constants.INTERVALS_PER_SLOT > self.forkChoice.fcStore.slot_clock.time.load(.monotonic)) { - try self.pending_blocks.append(self.allocator, cloned); - - self.logger.info( - "queued gossip block slot={d} blockroot=0x{x}: forkchoice time={d} < slot_start={d}", - .{ block.slot, &block_root, self.forkChoice.fcStore.slot_clock.time.load(.monotonic), block.slot * constants.INTERVALS_PER_SLOT }, - ); - return .{}; - } else { - self.logger.debug( - // - "chain already ticked while cloning block for queuing, skipping queuing and directly processing slot={d} blockroot=0x{x}: forkchoice time={d} < slot_start={d}", - // - .{ block.slot, &block_root, self.forkChoice.fcStore.slot_clock.time.load(.monotonic), block.slot * constants.INTERVALS_PER_SLOT }); - // by the time we cloned, chain ticked, so we can directly add and deinit clone - cloned.deinit(); - } - } - const missing_roots = self.onBlock(signed_block, .{ .blockRoot = block_root, }) catch |err| { - // we will not catch and enqueue block for FutureSlot error because this error here means - // that the block's slot is 2 ahead of the local because we have tolerance of 1 in case of - // clock skew or race between oninterval and block arrival self.logger.err("error processing block for slot={d} root=0x{x}: {any}", .{ block.slot, &block_root, @@ -1348,9 +1251,35 @@ pub const BeamChain = struct { pruned_count, }); - // 5 Rebase forkchouce - if (pruneForkchoice) - try self.forkChoice.rebase(latestFinalized.root, &canonical_view); + // 5 Rebase forkchoice — lazy prune with node-count threshold. + // + // Eager rebase drops pre-finalized ancestors from proto-array and + // remaps attestation-tracker indices. In-flight attestations whose + // source/target/head still points at one of those ancestors then + // fail the existence checks in validateAttestationData with + // Unknown{Source,Target,Head}Block, and the node burns bandwidth + // re-fetching blocks that will never come back. The grace window + // is measured in proto-array node count — see + // constants.PRUNE_NODE_THRESHOLD for the sizing rationale. + if (pruneForkchoice) { + if (self.forkChoice.getProtoNodeIndex(latestFinalized.root)) |finalized_idx| { + if (finalized_idx >= constants.PRUNE_NODE_THRESHOLD) { + try self.forkChoice.rebase(latestFinalized.root, &canonical_view); + } + // else: threshold not reached; keep pre-finalized ancestors + // in proto-array so in-flight attestations still resolve. + } else { + // Shouldn't happen: getCanonicalViewAndAnalysis already resolved + // latestFinalized.root via protoArray.indices. If it ever does, + // proto-array has fallen out of sync with fcStore — log loudly so + // the invariant violation is visible, and keep skipping the rebase + // (which also drops pruning) rather than dereferencing a stale node. + self.logger.warn( + "forkchoice: finalized root 0x{x} missing from proto-array; skipping rebase (invariant violation)", + .{&latestFinalized.root}, + ); + } + } // TODO: // 6. Remove orphaned blocks from database and cleanup unfinalized indices of there are any @@ -1408,31 +1337,19 @@ pub const BeamChain = struct { /// flooding the node with invalid blocks. /// /// Validations performed: - /// 1. Future slot check: block.slot must not be too far in the future - /// 2. Pre-finalized slot check: block.slot must be >= finalized_slot - /// 3. Proposer index bounds check: proposer_index must be < validator_count - /// 4. Parent existence check: parent_root must be known - /// 5. Slot ordering check: block.slot must be > parent.slot + /// 1. Pre-finalized slot check: block.slot must be >= finalized_slot + /// 2. Proposer index bounds check: proposer_index must be < validator_count + /// 3. Parent existence check: parent_root must be known + /// 4. Slot ordering check: block.slot must be > parent.slot + /// + /// Block admission is gated by parent / signature / STF; the slot is only + /// checked relative to the finalized boundary and to the parent block. pub fn validateBlock(self: *Self, block: types.BeamBlock, is_from_gossip: bool) !void { _ = is_from_gossip; - const current_slot = self.forkChoice.fcStore.slot_clock.timeSlots.load(.monotonic); const finalized_slot = self.forkChoice.fcStore.latest_finalized.slot; - // 1. Future slot check - reject blocks too far in the future - // Allow a small tolerance for clock skew, but reject clearly invalid future slots - // this can also happen because of race conditions between oninterval and block arrival - const max_future_tolerance: types.Slot = constants.MAX_FUTURE_SLOT_TOLERANCE; - if (block.slot > current_slot + max_future_tolerance) { - self.logger.debug("block validation failed: future slot {d} > max allowed {d} time(intervals)={d}", .{ - block.slot, - current_slot + max_future_tolerance, - self.forkChoice.fcStore.slot_clock.time.load(.monotonic), - }); - return BlockValidationError.FutureSlot; - } - - // 2. Pre-finalized slot check - reject blocks before finalized slot + // 1. Pre-finalized slot check - reject blocks before finalized slot if (block.slot < finalized_slot) { self.logger.debug("block validation failed: pre-finalized slot {d} < finalized {d}", .{ block.slot, @@ -1441,7 +1358,7 @@ pub const BeamChain = struct { return BlockValidationError.PreFinalizedSlot; } - // 3. Proposer index bounds check - sanity check against registry limit + // 2. Proposer index bounds check - sanity check against registry limit // This is a fast pre-check; actual proposer validity is verified during signature verification // We use VALIDATOR_REGISTRY_LIMIT as the upper bound since the validator set can grow beyond genesis if (block.proposer_index >= params.VALIDATOR_REGISTRY_LIMIT) { @@ -1452,14 +1369,14 @@ pub const BeamChain = struct { return BlockValidationError.InvalidProposerIndex; } - // 4. Parent existence check + // 3. Parent existence check const parent_block = self.forkChoice.getBlock(block.parent_root); if (parent_block == null) { // Log decision moved to node.zig where we can check if parent is already being fetched return BlockValidationError.UnknownParentBlock; } - // 5. Slot ordering check - block slot must be greater than parent slot + // 4. Slot ordering check - block slot must be greater than parent slot if (block.slot <= parent_block.?.slot) { self.logger.debug("block validation failed: slot {d} <= parent slot {d}", .{ block.slot, @@ -1471,11 +1388,13 @@ pub const BeamChain = struct { /// Validate incoming attestation before processing. /// - /// is_from_block: true if attestation came from a block, false if from network gossip + /// The time check applies only to the gossip path: admit a vote iff + /// `data.slot * INTERVALS_PER_SLOT <= store.time + GOSSIP_DISPARITY_INTERVALS`. + /// The bound is in intervals, not slots: a whole-slot margin would let an + /// adversary pre-publish next-slot aggregates ahead of any honest validator. /// - /// Per leanSpec: - /// - Gossip attestations (is_from_block=false): attestation.slot <= current_slot (no future tolerance) - /// - Block attestations (is_from_block=true): attestation.slot <= current_slot + 1 (lenient) + /// Block-included attestations skip the time check; they are trusted under + /// the block's own validation. `is_from_block` is retained as a log marker. pub fn validateAttestationData(self: *Self, data: types.AttestationData, is_from_block: bool) !void { const timer = zeam_metrics.lean_attestation_validation_time_seconds.start(); defer _ = timer.observe(); @@ -1503,7 +1422,6 @@ pub const BeamChain = struct { }); return AttestationValidationError.UnknownHeadBlock; }; - _ = head_block; // Will be used in future validations // 2. Validate slot relationships if (source_block.slot > target_block.slot) { @@ -1523,6 +1441,15 @@ pub const BeamChain = struct { return AttestationValidationError.SourceCheckpointExceedsTarget; } + // This corresponds to leanSpec's: assert data.head.slot >= data.target.slot + if (data.head.slot < data.target.slot) { + self.logger.debug("attestation validation failed: head slot {d} < target slot {d}", .{ + data.head.slot, + data.target.slot, + }); + return AttestationValidationError.HeadOlderThanTarget; + } + // 3. Validate checkpoint slots match block slots if (source_block.slot != data.source.slot) { self.logger.debug("attestation validation failed: source block slot {d} != source checkpoint slot {d}", .{ @@ -1541,24 +1468,34 @@ pub const BeamChain = struct { return AttestationValidationError.TargetCheckpointSlotMismatch; } - // 4. Validate attestation is not too far in the future - // - // Gossip attestations must be for current or past slots only. Validators attest - // in interval 1 of the current slot, so they cannot attest for future slots. - // Block attestations can be more lenient since the block itself was validated. - const current_slot = self.forkChoice.getCurrentSlot(); - const max_allowed_slot = if (is_from_block) - current_slot + constants.MAX_FUTURE_SLOT_TOLERANCE // Block attestations: allow +1 - else - current_slot; // Gossip attestations: no future slots allowed - - if (data.slot > max_allowed_slot) { - self.logger.debug("attestation validation failed: attestation slot {d} > max allowed slot {d} (is_from_block={any})", .{ - data.slot, - max_allowed_slot, - is_from_block, + // This corresponds to leanSpec's: assert head_block.slot == attestation.head.slot + if (head_block.slot != data.head.slot) { + self.logger.debug("attestation validation failed: head block slot {d} != head checkpoint slot {d}", .{ + head_block.slot, + data.head.slot, }); - return AttestationValidationError.AttestationTooFarInFuture; + return AttestationValidationError.HeadCheckpointSlotMismatch; + } + + // 4. Validate gossip attestation is not too far in the future. + // + // Bound is in intervals, not slots, and only applies to the gossip + // path. Block-included attestations are trusted under the block's + // own validation (matching leanSpec on_block, which doesn't run + // validate_attestation on block-body attestations at all). + if (!is_from_block) { + const current_time = self.forkChoice.fcStore.slot_clock.time.load(.monotonic); + const attestation_start_interval = data.slot * constants.INTERVALS_PER_SLOT; + const max_allowed_interval = current_time + constants.GOSSIP_DISPARITY_INTERVALS; + if (attestation_start_interval > max_allowed_interval) { + self.logger.debug("attestation validation failed: gossip attestation start interval {d} > max allowed interval {d} (slot={d}, time={d})", .{ + attestation_start_interval, + max_allowed_interval, + data.slot, + current_time, + }); + return AttestationValidationError.AttestationTooFarInFuture; + } } self.logger.debug("attestation validation passed: slot={d} source={d} target={d} is_from_block={any}", .{ data.slot, @@ -1586,7 +1523,6 @@ pub const BeamChain = struct { } pub fn onGossipAggregatedAttestation(self: *Self, signedAggregation: types.SignedAggregatedAttestation) !void { - // Validate the attestation data first (same rules as individual gossip attestations) try self.validateAttestationData(signedAggregation.data, false); try self.verifyAggregatedAttestation(signedAggregation); @@ -1834,12 +1770,12 @@ const AttestationValidationError = error{ SourceCheckpointExceedsTarget, SourceCheckpointSlotMismatch, TargetCheckpointSlotMismatch, + HeadCheckpointSlotMismatch, + HeadOlderThanTarget, AttestationTooFarInFuture, }; pub const BlockValidationError = error{ UnknownParentBlock, - /// Block slot is too far in the future - FutureSlot, /// Block slot is before the finalized slot PreFinalizedSlot, /// Block proposer_index exceeds validator count @@ -2354,12 +2290,17 @@ test "attestation validation - comprehensive" { try std.testing.expectError(error.TargetCheckpointSlotMismatch, beam_chain.validateAttestationData(invalid_attestation.message, false)); } - // Test 9: Attestation too far in future (for gossip) + // Test 9: Attestation too far in future (gossip path) + // + // Setup ended at time = 2 * INTERVALS_PER_SLOT = 10 (slot 2, interval 0). + // Gossip bound: data.slot * INTERVALS_PER_SLOT <= time + GOSSIP_DISPARITY_INTERVALS + // → max admitted slot here is ⌊(10 + 1) / 5⌋ = 2. + // slot 4 → start interval 20, well beyond 11. Rejected. { const future_attestation: types.SignedAttestation = .{ .validator_id = 0, .message = .{ - .slot = 3, // Future slot (current is 2) + .slot = 4, .head = types.Checkpoint{ .root = mock_chain.blockRoots[2], .slot = 2, @@ -2381,10 +2322,21 @@ test "attestation validation - comprehensive" { // TODO: Enable and update this test once the keymanager file-reading PR is added // JSON parsing for chain config needs to support validator_attestation_pubkeys instead of num_validators -test "attestation validation - gossip vs block future slot handling" { - // Test that gossip and block attestations have different future slot tolerances - // Gossip: must be <= current_slot - // Block: can be <= current_slot + 1 +test "attestation validation - gossip future-slot bound" { + // Gossip path is interval-grained: + // + // data.slot * INTERVALS_PER_SLOT <= time + GOSSIP_DISPARITY_INTERVALS + // + // Block-included attestations skip the time check entirely. + // + // Scenario: + // - Setup leaves the chain at slot 1, time = 5 (slot 1, interval 0). + // - A slot-2 vote at time = 5: gossip rejects (10 > 5 + 1 = 6). + // - Tick to time = 9 (slot 1, interval 4 — disparity boundary): + // gossip accepts (10 <= 9 + 1 = 10). + // - A slot-3 vote at time = 9: gossip rejects (15 > 9 + 1 = 10). + // - Block-included path admits the slot-3 vote at every tick. + var arena_allocator = std.heap.ArenaAllocator.init(std.testing.allocator); defer arena_allocator.deinit(); const allocator = arena_allocator.allocator(); @@ -2426,13 +2378,12 @@ test "attestation validation - gossip vs block future slot handling" { var beam_chain = try BeamChain.init(allocator, ChainOpts{ .config = chain_config, .anchorState = &beam_state, .nodeId = 0, .logger_config = &zeam_logger_config, .db = db, .node_registry = test_registry }, connected_peers); defer beam_chain.deinit(); - // Add one block (slot 1) + // Add one block (slot 1). Forkchoice ticks to time = INTERVALS_PER_SLOT (slot 1, interval 0). const block = mock_chain.blocks[1]; try beam_chain.forkChoice.onInterval(block.block.slot * constants.INTERVALS_PER_SLOT, false); const missing_roots = try beam_chain.onBlock(block, .{}); allocator.free(missing_roots); - // Current time is at slot 1, create attestation for slot 2 (next slot) const next_slot_attestation: types.SignedAttestation = .{ .validator_id = 0, .message = .{ @@ -2453,17 +2404,27 @@ test "attestation validation - gossip vs block future slot handling" { .signature = ZERO_SIGBYTES, }; - // Gossip attestations: should FAIL for next slot (current + 1) - // Per spec store.py:177: assert attestation.slot <= time_slots - try std.testing.expectError(error.AttestationTooFarInFuture, beam_chain.validateAttestationData(next_slot_attestation.message, false)); + // At time = 5 the gossip path rejects a slot-2 vote (3 intervals shy of boundary). + try std.testing.expectError( + error.AttestationTooFarInFuture, + beam_chain.validateAttestationData(next_slot_attestation.message, false), + ); - // Block attestations: should PASS for next slot (current + 1) - // Per spec store.py:140: assert attestation.slot <= Slot(current_slot + Slot(1)) + // Block-included attestations skip the time check. try beam_chain.validateAttestationData(next_slot_attestation.message, true); + + // Tick to the gossip disparity boundary: time = 2 * INTERVALS_PER_SLOT - GOSSIP_DISPARITY_INTERVALS = 9. + const boundary_time = 2 * constants.INTERVALS_PER_SLOT - constants.GOSSIP_DISPARITY_INTERVALS; + try beam_chain.forkChoice.onInterval(boundary_time, false); + + // At the boundary the gossip path admits the same vote. + try beam_chain.validateAttestationData(next_slot_attestation.message, false); + + // A slot-3 vote stays beyond the boundary on the gossip path but still admitted on the block path. const too_far_attestation: types.SignedAttestation = .{ .validator_id = 0, .message = .{ - .slot = 3, // Too far in future + .slot = 3, .head = types.Checkpoint{ .root = mock_chain.blockRoots[1], .slot = 1, @@ -2479,9 +2440,8 @@ test "attestation validation - gossip vs block future slot handling" { }, .signature = ZERO_SIGBYTES, }; - // Both should fail for slot 3 when current is slot 1 try std.testing.expectError(error.AttestationTooFarInFuture, beam_chain.validateAttestationData(too_far_attestation.message, false)); - try std.testing.expectError(error.AttestationTooFarInFuture, beam_chain.validateAttestationData(too_far_attestation.message, true)); + try beam_chain.validateAttestationData(too_far_attestation.message, true); } // TODO: Enable and update this test once the keymanager file-reading PR is added // JSON parsing for chain config needs to support validator_attestation_pubkeys instead of num_validators @@ -2711,3 +2671,95 @@ test "produceBlock - greedy selection by latest slot is suboptimal when attestat try std.testing.expect(unseen_count == 0); try std.testing.expect(known_count > 0); } + +test "processFinalizationAdvancement: below PRUNE_NODE_THRESHOLD keeps pre-finalized ancestors" { + // Regression: eager ProtoArray.rebase dropped pre-finalized ancestors on + // every finalization advance, so in-flight attestations whose source / + // target / head referenced those ancestors failed the existence check in + // validateAttestationData with Unknown{Source,Target,Head}Block. + // + // With the threshold gate, rebase is skipped while the finalized node's + // index in protoArray is below PRUNE_NODE_THRESHOLD. A short mock chain + // never crosses the threshold, so all recently-finalized ancestors must + // still be addressable after the chain finalizes. + var arena_allocator = std.heap.ArenaAllocator.init(std.testing.allocator); + defer arena_allocator.deinit(); + const allocator = arena_allocator.allocator(); + + const mock_chain = try stf.genMockChain(allocator, 5, null); + const spec_name = try allocator.dupe(u8, "beamdev"); + const fork_digest = try allocator.dupe(u8, "12345678"); + const chain_config = configs.ChainConfig{ + .id = configs.Chain.custom, + .genesis = mock_chain.genesis_config, + .spec = .{ + .preset = params.Preset.mainnet, + .name = spec_name, + .fork_digest = fork_digest, + .attestation_committee_count = 1, + .max_attestations_data = 16, + }, + }; + var beam_state = mock_chain.genesis_state; + var zeam_logger_config = zeam_utils.getTestLoggerConfig(); + + var tmp_dir = std.testing.tmpDir(.{}); + defer tmp_dir.cleanup(); + const data_dir = try tmp_dir.dir.realpathAlloc(allocator, "."); + defer allocator.free(data_dir); + + var db = try database.Db.open(allocator, zeam_logger_config.logger(.database_test), data_dir); + defer db.deinit(); + + const connected_peers = try allocator.create(std.StringHashMap(PeerInfo)); + connected_peers.* = std.StringHashMap(PeerInfo).init(allocator); + + const test_registry = try allocator.create(NodeNameRegistry); + defer allocator.destroy(test_registry); + test_registry.* = NodeNameRegistry.init(allocator); + defer test_registry.deinit(); + + var beam_chain = try BeamChain.init( + allocator, + ChainOpts{ + .config = chain_config, + .anchorState = &beam_state, + .nodeId = 7, + .logger_config = &zeam_logger_config, + .db = db, + .node_registry = test_registry, + }, + connected_peers, + ); + defer beam_chain.deinit(); + + // Drive the chain through the same entrypoint node.zig uses so the gate + // (fired inside processFinalizationAdvancement → onBlockFollowup) is + // actually exercised. Calling onBlock alone never triggers the rebase + // path, so the test would pass even without the threshold fix. + for (1..mock_chain.blocks.len) |i| { + const signed_block = mock_chain.blocks[i]; + const current_slot = signed_block.block.slot; + try beam_chain.forkChoice.onInterval(current_slot * constants.INTERVALS_PER_SLOT, false); + const missing_roots = try beam_chain.onBlock(signed_block, .{}); + allocator.free(missing_roots); + beam_chain.onBlockFollowup(true, &signed_block); + } + + // Sanity-check: mock chain must actually advance finalization for this + // regression to mean anything. + try std.testing.expect(beam_chain.forkChoice.getLatestFinalized().slot > 0); + + // The finalized node's index in protoArray should be well under the + // threshold (5 blocks total, threshold = 64). + const finalized_idx = beam_chain.forkChoice.getProtoNodeIndex(beam_chain.forkChoice.getLatestFinalized().root); + try std.testing.expect(finalized_idx != null); + try std.testing.expect(finalized_idx.? < constants.PRUNE_NODE_THRESHOLD); + + // All processed block roots — including the pre-finalized ones — must + // still resolve through the fork-choice API that validateAttestationData + // consults. Pre-gate, rebase would have dropped the below-finalized ones. + for (1..mock_chain.blocks.len) |i| { + try std.testing.expect(beam_chain.forkChoice.getProtoNode(mock_chain.blockRoots[i]) != null); + } +} diff --git a/pkgs/node/src/constants.zig b/pkgs/node/src/constants.zig index 105b771c3..5916fc354 100644 --- a/pkgs/node/src/constants.zig +++ b/pkgs/node/src/constants.zig @@ -5,9 +5,20 @@ const params = @import("@zeam/params"); pub const INTERVALS_PER_SLOT = 5; pub const SECONDS_PER_INTERVAL_MS: isize = @divFloor(params.SECONDS_PER_SLOT * std.time.ms_per_s, INTERVALS_PER_SLOT); -// Maximum number of slots in the future that an attestation is allowed to reference -// This prevents accepting attestations that are too far ahead of the current slot -pub const MAX_FUTURE_SLOT_TOLERANCE = 1; +// Future-slot tolerance for gossip attestations, measured in intervals: +// +// data.slot * INTERVALS_PER_SLOT <= store.time + GOSSIP_DISPARITY_INTERVALS +// +// where store.time is in intervals. One interval is roughly 800 ms at +// SECONDS_PER_SLOT=4 / INTERVALS_PER_SLOT=5. +// +// A whole-slot tolerance would let an adversary pre-publish next-slot +// aggregates ahead of any honest validator (~800 ms head start at 4 s +// slots); tightening to one interval bounds that head start to NTP drift. +// +// Block-included attestations skip this check entirely; they are trusted +// under the block's own validation. +pub const GOSSIP_DISPARITY_INTERVALS = 1; // Maximum depth for recursive block fetching // When fetching parent blocks, we stop after this many levels to avoid infinite loops @@ -21,6 +32,27 @@ pub const MAX_CACHED_BLOCKS = 1024; // Set to 7200 slots (approximately 8 hours in Lean, assuming 4 seconds per slot) pub const FORKCHOICE_PRUNING_INTERVAL_SLOTS: u64 = 7200; +// Grace window before proto-array rebuild fires on finalization advance, +// measured in *proto-array node count*, not slot distance. +// +// Eager rebase drops the just-finalized block's pre-finalized ancestors from +// proto-array and remaps attestation-tracker indices. In-flight attestations +// whose source / target / head still references one of those dropped blocks +// then fail existence checks with Unknown{Source,Target,Head}Block even +// though they were valid at sign time. 3SF-mini's fast finalization cadence +// makes this race fire across normal gossip delay. +// +// Gate rebase on the finalized node's index inside proto-array: only rebuild +// once at least PRUNE_NODE_THRESHOLD pre-finalized nodes sit before the +// finalized anchor. Below that, leave the prefix in place so in-flight +// attestations still resolve their references. On the canonical chain 64 +// nodes corresponds to ~64 slots (≈256 s at SECONDS_PER_SLOT=4), but because +// the index counts every node in proto-array — including fork siblings — +// the wall-clock grace can be shorter under heavy forking. The bound on +// memory is directly in nodes (bounded, small) and is what this threshold +// is sizing for. +pub const PRUNE_NODE_THRESHOLD: usize = 64; + // Forkchoice visualization constants pub const MAX_FC_DISPLAY_DEPTH = 100; pub const MAX_FC_DISPLAY_BRANCH = 10; diff --git a/pkgs/node/src/forkchoice.zig b/pkgs/node/src/forkchoice.zig index 60a5dfb9f..f6ea5b011 100644 --- a/pkgs/node/src/forkchoice.zig +++ b/pkgs/node/src/forkchoice.zig @@ -1647,9 +1647,10 @@ pub const ForkChoice = struct { // we will use parent block later as per the finalization gadget _ = parent_block; - if (slot * constants.INTERVALS_PER_SLOT > self.fcStore.slot_clock.time.load(.monotonic)) { - return ForkChoiceError.FutureSlot; - } else if (slot < self.fcStore.latest_finalized.slot) { + // Block admission only requires a known parent and a slot above + // the finalized boundary; STF and signature verification are the + // gating layers. + if (slot < self.fcStore.latest_finalized.slot) { return ForkChoiceError.PreFinalizedSlot; } @@ -1919,6 +1920,15 @@ pub const ForkChoice = struct { return self.protoArray.nodes.items[idx]; } + /// Get a ProtoNode's index in the underlying nodes array. Callers use + /// this to measure how many nodes precede a given block (e.g. to gate + /// proto-array rebase on a grace-window threshold). + pub fn getProtoNodeIndex(self: *Self, blockRoot: types.Root) ?usize { + self.mutex.lockShared(); + defer self.mutex.unlockShared(); + return self.protoArray.indices.get(blockRoot); + } + /// Get the current number of nodes in the forkchoice tree pub fn getNodeCount(self: *Self) usize { self.mutex.lockShared(); @@ -1930,7 +1940,6 @@ pub const ForkChoice = struct { pub const ForkChoiceError = error{ NotImplemented, UnknownParent, - FutureSlot, InvalidFutureAttestation, InvalidOnChainAttestation, PreFinalizedSlot, @@ -1993,11 +2002,8 @@ test "forkchoice block tree" { const block = signed_block.block; try stf.apply_transition(allocator, &beam_state, block, .{ .logger = module_logger }); - // shouldn't accept a future slot - const current_slot = block.slot; - try std.testing.expectError(error.FutureSlot, fork_choice.onBlock(block, &beam_state, .{ .currentSlot = current_slot, .blockDelayMs = 0, .confirmed = true })); - - try fork_choice.onInterval(current_slot * constants.INTERVALS_PER_SLOT, false); + // onBlock only requires a known parent and a non-pre-finalized slot. + try fork_choice.onInterval(block.slot * constants.INTERVALS_PER_SLOT, false); _ = try fork_choice.onBlock(block, &beam_state, .{ .currentSlot = block.slot, .blockDelayMs = 0, .confirmed = true }); try std.testing.expect(fork_choice.protoArray.nodes.items.len == i + 1); try std.testing.expect(std.mem.eql(u8, &mock_chain.blockRoots[i], &fork_choice.protoArray.nodes.items[i].blockRoot)); diff --git a/pkgs/node/src/node.zig b/pkgs/node/src/node.zig index 2319c5614..446e02b86 100644 --- a/pkgs/node/src/node.zig +++ b/pkgs/node/src/node.zig @@ -367,34 +367,6 @@ pub const BeamNode = struct { } return; }, - // Block arrived too early for local clock - cache and retry later. - error.FutureSlot => { - if (data.* == .block) { - const signed_block = data.block; - // Reuse the root we computed before taking the lock (issue #786). - const block_root = precomputed_block_root; - if (self.cacheFutureBlock(block_root, signed_block)) |_| { - self.logger.debug( - "cached future gossip block 0x{s} at slot {d}", - .{ std.fmt.bytesToHex(block_root, .lower)[0..], signed_block.block.slot }, - ); - } else |cache_err| { - if (cache_err == CacheBlockError.PreFinalized) { - self.logger.info( - "future gossip block 0x{s} is pre-finalized (slot={d}), pruning cached descendants", - .{ std.fmt.bytesToHex(block_root, .lower)[0..], signed_block.block.slot }, - ); - _ = self.network.pruneCachedBlocks(block_root, null); - } else { - self.logger.warn("failed to cache future gossip block 0x{s}: {any}", .{ - std.fmt.bytesToHex(block_root, .lower)[0..], - cache_err, - }); - } - } - } - return; - }, // Attestation/aggregation validation failed due to missing head/source/target block - // downgrade to debug when the missing block is already being fetched. error.UnknownHeadBlock, error.UnknownSourceBlock, error.UnknownTargetBlock => { @@ -541,12 +513,6 @@ pub const BeamNode = struct { "Cached block 0x{x} still missing parent, keeping in cache", .{&descendant_root}, ); - } else if (err == error.FutureSlot) { - // Block is still in the future, keep it cached - self.logger.debug( - "Cached block 0x{s} still in future slot, keeping in cache", - .{std.fmt.bytesToHex(descendant_root, .lower)[0..]}, - ); } else if (err == forkchoice.ForkChoiceError.PreFinalizedSlot) { // This block is now before finalized (finalization advanced while it was cached). // Prune this block and all its cached descendants; they are no longer useful. @@ -696,60 +662,6 @@ pub const BeamNode = struct { return parent_root; } - fn cacheFutureBlock( - self: *Self, - block_root: types.Root, - signed_block: types.SignedBlock, - ) CacheBlockError!void { - const finalized_slot = self.chain.forkChoice.fcStore.latest_finalized.slot; - const block_slot = signed_block.block.slot; - - if (block_slot <= finalized_slot) { - return CacheBlockError.PreFinalized; - } - - if (self.network.hasFetchedBlock(block_root)) { - return CacheBlockError.AlreadyCached; - } - - if (self.network.fetched_blocks.count() >= constants.MAX_CACHED_BLOCKS) { - self.logger.warn("Cache full ({d} blocks), rejecting future block 0x{s} at slot {d}", .{ - self.network.fetched_blocks.count(), - std.fmt.bytesToHex(block_root, .lower)[0..], - block_slot, - }); - return CacheBlockError.CachingFailed; - } - - const block_ptr = self.allocator.create(types.SignedBlock) catch { - return CacheBlockError.AllocationFailed; - }; - var block_owned = true; - errdefer if (block_owned) self.allocator.destroy(block_ptr); - - // Clone the block and capture its SSZ bytes in one pass. - // sszCloneAndGetBytes serializes the original block once (read-only on `signed_block`), - // then deserializes into the clone. The returned bytes are stored alongside the cached - // block so that onBlock never needs to re-serialize a live SignedBlock, which has been - // observed to cause memory corruption on the next cached block's processing. - const ssz_bytes = types.sszCloneAndGetBytes(self.allocator, types.SignedBlock, signed_block, block_ptr) catch { - return CacheBlockError.CloneFailed; - }; - errdefer if (block_owned) block_ptr.deinit(); - errdefer self.allocator.free(ssz_bytes); - - self.network.cacheFetchedBlock(block_root, block_ptr) catch { - return CacheBlockError.CachingFailed; - }; - block_owned = false; - - // Store the SSZ bytes after caching; ignore store failure (block is already cached, - // onBlock will fall back to fresh serialization if bytes are unavailable). - self.network.storeFetchedBlockSsz(block_root, ssz_bytes) catch { - self.allocator.free(ssz_bytes); - }; - } - fn processBlockByRootChunk(self: *Self, block_ctx: *const BlockByRootContext, signed_block: *const types.SignedBlock) !void { var block_root: types.Root = undefined; if (zeam_utils.hashTreeRoot(types.BeamBlock, signed_block.block, &block_root, self.allocator)) |_| { @@ -1292,19 +1204,6 @@ pub const BeamNode = struct { return e; }; - // Replay blocks that were queued waiting for the forkchoice clock to advance, - // then fetch any attestation head roots that were missing during replay. - const pending_missing_roots = self.chain.processPendingBlocks(); - defer self.allocator.free(pending_missing_roots); - if (pending_missing_roots.len > 0) { - self.fetchBlockByRoots(pending_missing_roots, 0) catch |err| { - self.logger.warn( - "failed to fetch {d} missing block(s) from pending blocks: {any}", - .{ pending_missing_roots.len, err }, - ); - }; - } - // Sweep timed-out RPC requests to prevent sync stalls from non-responsive peers. self.sweepTimedOutRequests(); @@ -1567,7 +1466,7 @@ pub const BeamNode = struct { pub fn run(self: *Self) !void { // Catch up fork choice time to current interval before processing any requests. - // This prevents FutureSlot errors when receiving blocks via RPC immediately after starting. + // Keeps validator duties and aggregation timing aligned with the local clock. const current_interval = self.clock.current_interval; if (current_interval > 0) { try self.chain.forkChoice.onInterval(@intCast(current_interval), false);