Skip to content

fix(spectest): promote gossip attestations before head selection in fork-choice runner #999

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
zclawz wants to merge 4 commits into
base: main
Choose a base branch
from
Open

fix(spectest): promote gossip attestations before head selection in fork-choice runner #999

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
8d009ac
node: move proposal signing to validator client
zclawz Jun 12, 2026
52917f1
fix(spectest): promote gossip attestations before head selection in f…
zclawz Jun 15, 2026
f7f541f
fix(spectest): revert gossipAggregatedAttestation to updateHead() — n…
Jun 15, 2026
1af41c6
fix: promoteGossipVotes — separate vote promotion from aggregated pay…
Jun 15, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
53 changes: 9 additions & 44 deletions pkgs/node/src/chain.zig
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -5085,9 +5085,9 @@ pub const BeamChain = struct {
}

/// Worker body for off-loop block production (runs on a shared `ThreadPool` worker).
/// produceBlock advances local fork-choice; buildBlockProof does the recursive-STARK Type-2
/// merge; publishBlock verifies + persists + gossips. The SignedBlock is freed here — onBlock
/// consumes it as input and does not retain its allocations (see publishBlock).
/// produceBlock advances local fork-choice, then validator_client signs/builds
/// the recursive-STARK Type-2 proof and returns a gossip output. BeamNode
/// drains that output on the interval tick and remains the sole publisher.
fn proposeImpl(chain: *Self, node: *@import("./node.zig").BeamNode, slot: usize, proposer_id: usize) void {
defer _ = chain.propose_inflight.fetchSub(1, .acq_rel);
const worker_timer = zeam_metrics.lean_block_building_time_seconds.start();
Expand Down Expand Up @@ -5115,50 +5115,15 @@ pub const BeamChain = struct {

chain.logger.info("produced block for slot={d} proposer={d} root={x}", .{ slot, proposer_id, &produced_block.blockRoot });

// TODO - since signing needs to be architectrually seggregated from the nodes, move following section needs to move to validator_client
// 1. signing by the validator to get proposer signature
// 2. building block proof as done underneath
// 3. returning the signed block as array of gossip objects like in mayBeDoAttestation which will make node handle the publish in its onInterval
//
// All this needs to happen from the mayBeDoProposal which will first call node to produce and return the block with signtaures
// This also frees up the node from building the merge proof in the architecture where validator client runs separately from node
// alleviating node from the hardwork so that it can also serve validators altruistically
const proposer_signature = validator.key_manager.signBlockRoot(proposer_id, &produced_block.blockRoot, @intCast(slot)) catch |e| {
chain.logger.err("propose worker: signBlockRoot failed slot={d}: {any}", .{ slot, e });
return;
};

chain.logger.info("produced block signed, building block proof for slot={d} proposer={d} root={x}", .{
slot,
proposer_id,
&produced_block.blockRoot,
});
var proof = types.MultiMessageAggregate.init(chain.allocator) catch return;
var proof_owned = true;
defer if (proof_owned) proof.deinit();
chain.buildBlockProof(&produced_block, &proposer_signature, &proof) catch |e| {
chain.logger.err("propose worker: buildBlockProof failed slot={d}: {any}", .{ slot, e });
return;
};

// The Type-1 list is now folded into the Type-2 proof; free it. The block moves into the
// SignedBlock (which we free after publishing).
for (produced_block.attestation_signatures.slice()) |*t1| t1.deinit();
produced_block.attestation_signatures.deinit();

var signed_block = types.SignedBlock{
.block = produced_block.block,
.proof = proof,
};
produced_owned = false; // block now owned by signed_block
proof_owned = false; // proof now owned by signed_block
defer signed_block.deinit();
var output = validator.buildProposalOutput(&produced_block, proposer_id, slot) catch return;
produced_owned = false; // block now owned by output's SignedBlock
defer output.deinit();

node.publishBlock(signed_block) catch |e| {
chain.logger.err("propose worker: publishBlock failed slot={d}: {any}", .{ slot, e });
node.enqueueValidatorOutput(&output) catch |e| {
chain.logger.err("propose worker: enqueue validator output failed slot={d}: {any}", .{ slot, e });
return;
};
chain.logger.info("published block for slot={d} root={x}", .{ slot, &produced_block.blockRoot });
chain.logger.info("queued signed block for validator publish slot={d} root={x}", .{ slot, &produced_block.blockRoot });
}

/// Find the subnet of the first set participant in `participants`.
Expand Down
30 changes: 30 additions & 0 deletions pkgs/node/src/forkchoice.zig
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2875,6 +2875,36 @@ pub const ForkChoice = struct {
return self.acceptNewAttestationsUnlocked();
}

/// Promote per-validator gossip votes (latestNew → latestKnown) and recompute head.
///
/// Unlike `acceptNewAttestations`, this does NOT migrate `latest_new_aggregated_payloads`
/// into `latest_known_aggregated_payloads`. That migration only belongs at the periodic
/// acceptance tick (slot_interval == 4 in production, or an explicit TickStep that crosses
/// interval 4 in the spec-test runner).
///
/// Use this from the spec-test runner's `processAttestationStep` when processing individual
/// gossip attestations: the vote becomes visible to head selection immediately (matching the
/// production `acceptNewAttestations` path), but the aggregated-payload pools remain in the
/// state the fixture expects until the next proper acceptance tick.
pub fn promoteGossipVotes(self: *Self) !ProtoBlock {
self.mutex.lock();
defer self.mutex.unlock();

// Mirror the tracker-promotion half of acceptNewAttestationsUnlocked:
// keep the fresher of latestNew / latestKnown, never drop a known vote.
for (0..self.config.genesis.numValidators()) |validator_id| {
var tracker = self.attestations.get(validator_id) orelse continue;
const new_vote = tracker.latestNew orelse continue;
const known_slot = (tracker.latestKnown orelse ProtoAttestation{}).slot;
if (tracker.latestKnown == null or new_vote.slot >= known_slot) {
tracker.latestKnown = new_vote;
try self.attestations.put(validator_id, tracker);
}
}

return self.updateHeadUnlocked();
}

// SAFE GETTERS FOR SHARED STATE
// These provide thread-safe access to internal state

Expand Down
113 changes: 77 additions & 36 deletions pkgs/node/src/node.zig
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -161,6 +161,11 @@ pub const BeamNode = struct {
unserved_block_retries: std.AutoHashMap(types.Root, UnservedBlockRetry),
unserved_block_retries_lock: zeam_utils.SyncMutex = .{},

/// Validator outputs produced by off-loop proposal workers and drained by
/// the libxev interval tick so BeamNode remains the only publisher.
pending_validator_gossip: std.ArrayList(networks.GossipMessage),
pending_validator_gossip_lock: zeam_utils.SyncMutex = .{},

/// Test-only failure injection for `onInterval` catch-and-continue paths.
test_inject_validator_error_at_intervals: []const usize = &.{},
test_inject_aggregator_error_at_intervals: []const usize = &.{},
Expand Down Expand Up @@ -293,6 +298,7 @@ pub const BeamNode = struct {
.orphan_dependents = std.AutoHashMap(types.Root, std.ArrayList(types.Root)).init(allocator),
.range_async_chunk_imports = std.AutoHashMap(types.Root, u64).init(allocator),
.unserved_block_retries = std.AutoHashMap(types.Root, UnservedBlockRetry).init(allocator),
.pending_validator_gossip = .empty,
};

chain.setPruneCachedBlocksCallback(self, pruneCachedBlocksCallback);
Expand Down Expand Up @@ -336,9 +342,56 @@ pub const BeamNode = struct {
self.orphan_dependents.deinit();
self.range_async_chunk_imports.deinit();
self.unserved_block_retries.deinit();
{
var output = validatorClient.ValidatorClientOutput{
.allocator = self.allocator,
.gossip_messages = self.pending_validator_gossip,
};
output.deinit();
}
self.network.deinit();
}

pub fn enqueueValidatorOutput(self: *Self, output: *validatorClient.ValidatorClientOutput) !void {
self.pending_validator_gossip_lock.lock();
defer self.pending_validator_gossip_lock.unlock();
try self.pending_validator_gossip.appendSlice(self.allocator, output.gossip_messages.items);
output.gossip_messages.clearRetainingCapacity();
}

fn drainPendingValidatorOutput(self: *Self) validatorClient.ValidatorClientOutput {
var output = validatorClient.ValidatorClientOutput.init(self.allocator);
self.pending_validator_gossip_lock.lock();
std.mem.swap(std.ArrayList(networks.GossipMessage), &output.gossip_messages, &self.pending_validator_gossip);
self.pending_validator_gossip_lock.unlock();
return output;
}

fn publishValidatorOutput(self: *Self, output: *validatorClient.ValidatorClientOutput, slot: types.Slot, interval: usize) void {
for (output.gossip_messages.items) |gossip_msg| {
switch (gossip_msg) {
.block => |signed_block| {
self.publishBlock(signed_block) catch |e| {
self.logger.err("error publishing block from validator at slot={d} interval={d}: {any} (continuing tick)", .{ slot, interval, e });
zeam_metrics.metrics.lean_node_interval_error_total.incr(.{ .site = "publishBlock" }) catch |me| self.logger.warn("metric incr failed: {any}", .{me});
};
},
.attestation => |signed_attestation| {
self.publishAttestation(signed_attestation) catch |e| {
self.logger.err("error publishing attestation from validator at slot={d} interval={d}: {any} (continuing tick)", .{ slot, interval, e });
zeam_metrics.metrics.lean_node_interval_error_total.incr(.{ .site = "publishAttestation" }) catch |me| self.logger.warn("metric incr failed: {any}", .{me});
};
},
.aggregation => |signed_aggregation| {
self.publishAggregation(signed_aggregation) catch |e| {
self.logger.err("error publishing aggregation from validator at slot={d} interval={d}: {any} (continuing tick)", .{ slot, interval, e });
zeam_metrics.metrics.lean_node_interval_error_total.incr(.{ .site = "publishAggregation" }) catch |me| self.logger.warn("metric incr failed: {any}", .{me});
};
},
}
}
}

fn recordRangeSyncOutcome(_: *Self, outcome: []const u8) void {
zeam_metrics.metrics.zeam_blocks_by_range_sync_total.incr(.{ .outcome = outcome }) catch {};
}
Expand Down Expand Up @@ -2957,44 +3010,32 @@ pub const BeamNode = struct {
}

// Application-layer failures are logged and counted, not returned.
if (self.test_inject_validator_error_at_intervals.len > 0 and
std.mem.indexOfScalar(usize, self.test_inject_validator_error_at_intervals, interval) != null)
{
self.logger.err("error ticking validator to time(intervals)={d} err=error.TestInjected (continuing tick)", .{interval});
zeam_metrics.metrics.lean_node_interval_error_total.incr(.{ .site = "validator.onInterval" }) catch |me| self.logger.warn("metric incr failed: {any}", .{me});
} else if (self.validator) |*validator| {
// we also tick validator per interval in case it would
// need to sync its future duties when its an independent validator
var maybe_validator_output = validator.onInterval(self, interval) catch |e| blk: {
self.logger.err("error ticking validator to time(intervals)={d} err={any} (continuing tick)", .{ interval, e });
if (self.validator) |*validator| {
var queued_validator_output = self.drainPendingValidatorOutput();
if (queued_validator_output.gossip_messages.items.len > 0) {
defer queued_validator_output.deinit();
self.publishValidatorOutput(&queued_validator_output, slot, interval);
} else {
queued_validator_output.deinit();
}

if (self.test_inject_validator_error_at_intervals.len > 0 and
std.mem.indexOfScalar(usize, self.test_inject_validator_error_at_intervals, interval) != null)
{
self.logger.err("error ticking validator to time(intervals)={d} err=error.TestInjected (continuing tick)", .{interval});
zeam_metrics.metrics.lean_node_interval_error_total.incr(.{ .site = "validator.onInterval" }) catch |me| self.logger.warn("metric incr failed: {any}", .{me});
break :blk null;
};
} else {
// we also tick validator per interval in case it would
// need to sync its future duties when its an independent validator
var maybe_validator_output = validator.onInterval(self, interval) catch |e| blk: {
self.logger.err("error ticking validator to time(intervals)={d} err={any} (continuing tick)", .{ interval, e });
zeam_metrics.metrics.lean_node_interval_error_total.incr(.{ .site = "validator.onInterval" }) catch |me| self.logger.warn("metric incr failed: {any}", .{me});
break :blk null;
};

if (maybe_validator_output) |*output| {
defer output.deinit();
for (output.gossip_messages.items) |gossip_msg| {
// Process based on message type
switch (gossip_msg) {
.block => |signed_block| {
self.publishBlock(signed_block) catch |e| {
self.logger.err("error publishing block from validator at slot={d} interval={d}: {any} (continuing tick)", .{ slot, interval, e });
zeam_metrics.metrics.lean_node_interval_error_total.incr(.{ .site = "publishBlock" }) catch |me| self.logger.warn("metric incr failed: {any}", .{me});
};
},
.attestation => |signed_attestation| {
self.publishAttestation(signed_attestation) catch |e| {
self.logger.err("error publishing attestation from validator at slot={d} interval={d}: {any} (continuing tick)", .{ slot, interval, e });
zeam_metrics.metrics.lean_node_interval_error_total.incr(.{ .site = "publishAttestation" }) catch |me| self.logger.warn("metric incr failed: {any}", .{me});
};
},
.aggregation => |signed_aggregation| {
self.publishAggregation(signed_aggregation) catch |e| {
self.logger.err("error publishing aggregation from validator at slot={d} interval={d}: {any} (continuing tick)", .{ slot, interval, e });
zeam_metrics.metrics.lean_node_interval_error_total.incr(.{ .site = "publishAggregation" }) catch |me| self.logger.warn("metric incr failed: {any}", .{me});
};
},
}
if (maybe_validator_output) |*output| {
defer output.deinit();
self.publishValidatorOutput(output, slot, interval);
}
}
}
Expand Down
58 changes: 51 additions & 7 deletions pkgs/node/src/validator_client.zig
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -30,6 +30,7 @@ pub const ValidatorClientOutput = struct {
pub fn deinit(self: *Self) void {
for (self.gossip_messages.items) |*gossip_msg| {
switch (gossip_msg.*) {
.block => |*signed_block| signed_block.deinit(),
.aggregation => |*signed_aggregation| signed_aggregation.deinit(),
else => {},
}
Expand Down Expand Up @@ -117,13 +118,56 @@ pub const ValidatorClient = struct {
}
}

// Block production moved off the slot loop to submitPropose / proposeImpl (a
// thread_pool worker). The old on-loop maybeDoProposal — which built the multi-second Type-2
// merge inline and would freeze gossip/tick handling — was removed in favour of that single
// off-loop path. main's improvements to the old maybeDoProposal (the
// .behind_peers → .peers_materially_ahead rename, logBehindPeersDebug, behind_peer_count, and
// the submitBlockBuildOnInterval/finalizeProposalIfReady build/sign split) now live in the
// off-loop chain.proposeImpl path and in mayBeDoAttestation below.
// Block production still runs off the slot loop, but validator-owned signing
// and proof assembly live here rather than in chain.zig. The chain worker
// produces the unsigned block, calls this method to sign/build the
// SignedBlock, and queues the resulting gossip output back to BeamNode for
// publication on the next interval tick.
pub fn buildProposalOutput(
self: *Self,
produced_block: *chains.ProducedBlock,
proposer_id: usize,
slot: usize,
) !ValidatorClientOutput {
var result = ValidatorClientOutput.init(self.allocator);
errdefer result.deinit();
try result.gossip_messages.ensureTotalCapacity(self.allocator, 1);

const proposer_signature = self.key_manager.signBlockRoot(proposer_id, &produced_block.blockRoot, @intCast(slot)) catch |e| {
self.logger.err("propose worker: signBlockRoot failed slot={d}: {any}", .{ slot, e });
return e;
};

self.logger.info("produced block signed, building block proof for slot={d} proposer={d} root={x}", .{
slot,
proposer_id,
&produced_block.blockRoot,
});

var proof = types.MultiMessageAggregate.init(self.allocator) catch |e| {
self.logger.err("propose worker: init block proof failed slot={d}: {any}", .{ slot, e });
return e;
};
errdefer proof.deinit();

self.chain.buildBlockProof(produced_block, &proposer_signature, &proof) catch |e| {
self.logger.err("propose worker: buildBlockProof failed slot={d}: {any}", .{ slot, e });
return e;
};

// The Type-1 list is now folded into the Type-2 proof; free it. The
// produced block itself moves into the SignedBlock below, which is
// owned by the ValidatorClientOutput until BeamNode publishes it.
for (produced_block.attestation_signatures.slice()) |*t1| t1.deinit();
produced_block.attestation_signatures.deinit();

const signed_block = types.SignedBlock{
.block = produced_block.block,
.proof = proof,
};
result.gossip_messages.appendAssumeCapacity(.{ .block = signed_block });
return result;
}

pub fn mayBeDoAttestation(self: *Self, slot: usize) !?ValidatorClientOutput {
if (self.ids.len == 0) return null;
Expand Down
Loading
Loading