From 04f5ace264ac70dd9a9d5975c5fe8a7d04df049d Mon Sep 17 00:00:00 2001 From: Chen Kai <281165273grape@gmail.com> Date: Thu, 11 Jun 2026 14:29:46 +0800 Subject: [PATCH] feat(node): bound the block-proof merge by a publish budget (attestation-group cap) The proposal deadline (proposal_deadline_pct) bounded only the gathering/ compaction phase; the Type-2 merge itself ran unbounded, so a busy pool let the block-proof merge push gossip publication multiple intervals late. The merge cost scales with its component count (one per attestation group + the proposer), so the proposer now derives a group cap from a publish budget: publish budget (proposal_publish_target_intervals, default 2 intervals) - gathering budget - signing margin = merge budget merge budget / per-component cost estimate - 1 (proposer part) = group cap The per-component estimate is an EMA of measured merges, seeded at 800ms and updated only from multi-part merges (a proposer-only merge attributes its whole fixed cost to one part and would ratchet the cap down). The cap floors at one group: attestations only reach the state through blocks, so capping to zero across the network would stall justification outright; the thin-vs-late trade applies only above the floor. Deferred groups stay in the pool for later proposers. Candidates are already sorted most-valuable-first, so truncation keeps the highest-coverage groups. No CLI flag yet: the arg struct already sits at zigcli's comptime eval-branch-quota cliff (any new field fails the build); the knob is plumbed through NodeOptions/ChainOpts for programmatic configuration. --- pkgs/cli/src/node.zig | 2 + pkgs/node/src/chain.zig | 117 +++++++++++++++++++++++++++++++++-- pkgs/node/src/forkchoice.zig | 26 +++++++- pkgs/node/src/lib.zig | 1 + pkgs/node/src/node.zig | 2 + 5 files changed, 140 insertions(+), 8 deletions(-) diff --git a/pkgs/cli/src/node.zig b/pkgs/cli/src/node.zig index 91cae7bd5..de98412f8 100644 --- a/pkgs/cli/src/node.zig +++ b/pkgs/cli/src/node.zig @@ -120,6 +120,7 @@ pub const NodeOptions = struct { /// Percentage of the proposal interval allocated as the build-worker /// deadline budget. proposal_deadline_pct: u32 = node_lib.default_proposal_deadline_pct, + proposal_publish_target_intervals: u32 = node_lib.default_proposal_publish_target_intervals, /// Cap on the number of child STARK proofs merged with raw signatures /// by the aggregator-worker path. Threaded through to /// `ForkChoice.max_aggregation_children` and applied by @@ -607,6 +608,7 @@ pub const Node = struct { .max_aggregation_children = options.max_aggregation_children, .aggregate_max_inflight = aggregate_max_inflight, .proposal_deadline_pct = options.proposal_deadline_pct, + .proposal_publish_target_intervals = options.proposal_publish_target_intervals, }); errdefer self.beam_node.deinit(); diff --git a/pkgs/node/src/chain.zig b/pkgs/node/src/chain.zig index 3395d1413..77ea0a5f1 100644 --- a/pkgs/node/src/chain.zig +++ b/pkgs/node/src/chain.zig @@ -62,6 +62,14 @@ pub const BlockProductionParams = struct { /// rest of the interval for block signing and the Type-2 multi-message merge. deadline_ns: ?i64 = null, + /// Optional cap on the number of distinct attestation groups selected into + /// the block (each group is one component of the Type-2 block proof). + /// `proposeImpl` derives this from the publish-target budget and the + /// observed per-component merge cost so the merge finishes in time for the + /// block to reach gossip by `proposal_publish_target_intervals`. `null` = + /// spec cap only. + max_attestation_groups: ?usize = null, + pub fn format(self: BlockProductionParams, writer: anytype) !void { try writer.print("BlockProductionParams{{ slot={d}, proposer_index={d} }}", .{ self.slot, self.proposer_index }); } @@ -113,11 +121,29 @@ pub const ChainOpts = struct { /// Default 90: ~720ms aggregation budget + ~80ms finalize/sign budget at /// the default 800ms interval. proposal_deadline_pct: u32 = default_proposal_deadline_pct, + proposal_publish_target_intervals: u32 = default_proposal_publish_target_intervals, }; /// Default value for `ChainOpts.proposal_deadline_pct` pub const default_proposal_deadline_pct: u32 = 50; +/// Default value for `ChainOpts.proposal_publish_target_intervals`: the block +/// (including its Type-2 proof) should reach gossip before this many proposal +/// intervals have elapsed since the propose trigger. This bounds how LATE the +/// block lands, trading block fullness for timeliness; it does not make +/// same-slot attestation (interval 1) reachable — at current prover speeds a +/// merge with any attestation group overruns interval 1 regardless, and +/// attesters vote on the parent. 2 keeps the block well ahead of the next +/// slot's proposer instead of drifting toward intervals 3-4 unbounded. +pub const default_proposal_publish_target_intervals: u32 = 2; + +/// Seed for the per-component merge-cost estimate (EMA, nanoseconds) used to +/// derive the attestation-group cap from the publish budget. Calibrated from +/// observed Type-2 merge times (~0.8-1.5s for 1-2 components on devnet-class +/// hardware); the estimate self-corrects from measured merges after the first +/// proposal. +pub const default_merge_per_part_ns: u64 = 800 * std.time.ns_per_ms; + pub const CachedProcessedBlockInfo = struct { postState: ?*types.BeamState = null, blockRoot: ?types.Root = null, @@ -481,6 +507,17 @@ pub const BeamChain = struct { /// it (this percent of the proposal interval). proposal_deadline_pct: u32, + /// See `ChainOpts.proposal_publish_target_intervals` (CLI + /// `--proposal-publish-target-intervals`): the publish budget, in proposal + /// intervals, from which `proposeImpl` derives the attestation-group cap. + proposal_publish_target_intervals: u32, + + /// Exponential moving average of the observed per-component Type-2 merge + /// cost in nanoseconds. Written only by the propose worker (single-flight), + /// read at the next proposal to size the group cap. Atomic for cross-thread + /// visibility, not for contention. + merge_per_part_ns: std.atomic.Value(u64) = .init(default_merge_per_part_ns), + /// Optional chain-worker thread (slice c-2b commit 3 of #803). /// When non-null, `BeamChain` exposes the `submit*` family of /// methods which enqueue work onto the worker's queues; the @@ -665,6 +702,7 @@ pub const BeamChain = struct { .aggregate_wg = .{}, .aggregate_max_inflight = opts.aggregate_max_inflight, .proposal_deadline_pct = opts.proposal_deadline_pct, + .proposal_publish_target_intervals = opts.proposal_publish_target_intervals, // Pending attestation / aggregated-attestation buffers — empty // at init; FIFO-bounded by constants.MAX_PENDING_ATTESTATIONS in // the enqueue helpers. @@ -2720,7 +2758,7 @@ pub const BeamChain = struct { const payload_agg_timer = zeam_metrics.lean_block_building_payload_aggregation_time_seconds.start(); // FFI call against the owned snapshot — no lock held during this // window. - const proposal_atts = try self.forkChoice.getProposalAttestations(pre_snapshot, opts.slot, opts.proposer_index, parent_root, opts.deadline_ns); + const proposal_atts = try self.forkChoice.getProposalAttestations(pre_snapshot, opts.slot, opts.proposer_index, parent_root, opts.deadline_ns, opts.max_attestation_groups); _ = payload_agg_timer.observe(); var agg_attestations = proposal_atts.attestations; @@ -5069,14 +5107,33 @@ pub const BeamChain = struct { // Cap single-message attestation aggregation at `proposal_deadline_pct`% of the // proposal interval, reserving the remainder for block signing and the Type-2 // multi-message merge. A timely block with fewer attestation_data beats a late - // block carrying more — the off-loop merge is the dominant cost, so we bound the - // controllable gathering phase rather than the merge. Clamp to [0, 99] so the - // sign/merge step always keeps headroom (same form as main's produceBlockWorker). + // block carrying more. The merge itself is bounded separately below via the + // publish-budget group cap. Clamp to [0, 99] so the sign/merge step always + // keeps headroom (same form as main's produceBlockWorker). const pct_clamped: i64 = @intCast(@min(chain.proposal_deadline_pct, 99)); const budget_ms: i64 = @divFloor(@as(i64, constants.SECONDS_PER_INTERVAL_MS) * pct_clamped, 100); const deadline_ns: i64 = @intCast(zeam_utils.monotonicTimestampNs() + @as(i128, budget_ms) * @as(i128, std.time.ns_per_ms)); - var produced_block = chain.produceBlock(.{ .slot = slot, .proposer_index = proposer_id, .deadline_ns = deadline_ns }) catch |e| { + // Merge-budget group cap: the Type-2 merge cost scales with its component + // count (one per attestation group + the proposer), so cap the number of + // groups such that the merge fits in what remains of the publish budget + // after gathering and signing. The per-component cost estimate is an EMA + // of measured merges (seeded by default_merge_per_part_ns). Degrades to a + // proposer-only (empty-attestation) block when nothing fits: a thin block + // on time beats a full block that misses the attestation interval, and + // deferred groups stay in the pool for later proposers. + const publish_budget_ns: u64 = @as(u64, chain.proposal_publish_target_intervals) * + @as(u64, constants.SECONDS_PER_INTERVAL_MS) * std.time.ns_per_ms; + const sign_margin_ns: u64 = 100 * std.time.ns_per_ms; + const merge_budget_ns = publish_budget_ns -| @as(u64, @intCast(budget_ms)) * std.time.ns_per_ms -| sign_margin_ns; + const max_groups = computeMergeGroupCap(merge_budget_ns, chain.merge_per_part_ns.load(.monotonic)); + + var produced_block = chain.produceBlock(.{ + .slot = slot, + .proposer_index = proposer_id, + .deadline_ns = deadline_ns, + .max_attestation_groups = max_groups, + }) catch |e| { chain.logger.err("propose worker: produceBlock failed slot={d}: {any}", .{ slot, e }); return; }; @@ -5094,10 +5151,30 @@ pub const BeamChain = struct { var proof = types.MultiMessageAggregate.init(chain.allocator) catch return; var proof_owned = true; defer if (proof_owned) proof.deinit(); + const merge_start_ns = zeam_utils.monotonicTimestampNs(); chain.buildBlockProof(&produced_block, &proposer_signature, &proof) catch |e| { chain.logger.err("propose worker: buildBlockProof failed slot={d}: {any}", .{ slot, e }); return; }; + // Update the per-component merge-cost EMA from the measured merge + // (components = attestation groups + the proposer part). Weight 3:1 + // toward history so a single outlier cannot whipsaw the next cap. + // Only multi-part merges update the estimate: a proposer-only merge + // attributes its entire fixed cost (state snapshot, proposer prove, + // gate wait) to a single part and carries no marginal signal — feeding + // those samples back would inflate the estimate and ratchet the cap + // down. The measured window still includes the fixed overhead, so the + // estimate is an upper bound on the marginal cost; the liveness floor + // in computeMergeGroupCap keeps that conservatism harmless. + { + const merge_ns: u64 = @intCast(@max(zeam_utils.monotonicTimestampNs() - merge_start_ns, 0)); + const parts: u64 = @intCast(produced_block.attestation_signatures.len() + 1); + if (parts >= 2) { + const per_part = merge_ns / parts; + const old = chain.merge_per_part_ns.load(.monotonic); + chain.merge_per_part_ns.store((old * 3 + per_part) / 4, .monotonic); + } + } // 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). @@ -5119,6 +5196,21 @@ pub const BeamChain = struct { chain.logger.info("published block for slot={d} root={x}", .{ slot, &produced_block.blockRoot }); } + /// Pure decision: how many attestation groups fit the merge budget, given + /// the per-component cost estimate. One component is always reserved for + /// the proposer's own signature; returns 0 when even a 2-component merge + /// does not fit (proposer-only block). + pub fn computeMergeGroupCap(merge_budget_ns: u64, per_part_ns: u64) usize { + const per = @max(per_part_ns, 1); + const max_parts = merge_budget_ns / per; + // Liveness floor: never cap below one attestation group. Attestations + // only reach the state through blocks, so an all-zero cap across the + // network would stall justification outright; one group per block is + // the minimum that keeps the chain advancing, even when it overruns + // the publish budget. The thin-vs-late trade applies only above this. + return @max(@as(usize, @intCast(max_parts -| 1)), 1); + } + /// Find the subnet of the first set participant in `participants`. /// All participants in a single aggregated attestation come from the same /// committee subnet, so the first one is representative. Returns null if @@ -9575,3 +9667,18 @@ test "chain.statesGet under chain_worker enabled does not block exclusive writer // the map: refcount kept the underlying state alive. try std.testing.expectEqual(mock_chain.genesis_state.slot, borrow.state.slot); } + +test "computeMergeGroupCap reserves the proposer part and floors at one group" { + const ms = std.time.ns_per_ms; + // 2200ms budget at 700ms/part -> 3 parts fit -> 2 attestation groups. + try std.testing.expectEqual(@as(usize, 2), BeamChain.computeMergeGroupCap(2200 * ms, 700 * ms)); + // 1500ms budget at 700ms/part -> 2 parts fit -> 1 group. + try std.testing.expectEqual(@as(usize, 1), BeamChain.computeMergeGroupCap(1500 * ms, 700 * ms)); + // Budget too small for any group -> liveness floor keeps 1 group anyway. + try std.testing.expectEqual(@as(usize, 1), BeamChain.computeMergeGroupCap(700 * ms, 700 * ms)); + try std.testing.expectEqual(@as(usize, 1), BeamChain.computeMergeGroupCap(0, 700 * ms)); + // Large budget at the default seed -> plenty of room (spec cap applies later). + try std.testing.expect(BeamChain.computeMergeGroupCap(8000 * ms, default_merge_per_part_ns) >= 8); + // Degenerate estimate of 0 must not divide-by-zero. + _ = BeamChain.computeMergeGroupCap(1000 * ms, 0); +} diff --git a/pkgs/node/src/forkchoice.zig b/pkgs/node/src/forkchoice.zig index a2f120423..bfb4bda1d 100644 --- a/pkgs/node/src/forkchoice.zig +++ b/pkgs/node/src/forkchoice.zig @@ -1166,6 +1166,7 @@ pub const ForkChoice = struct { proposer_index: types.ValidatorIndex, parent_root: [32]u8, deadline_ns: ?i64, + max_groups: ?usize, ) !ProposalAttestationsResult { var agg_attestations = try types.AggregatedAttestations.init(self.allocator); var agg_att_cleanup = true; @@ -1275,8 +1276,25 @@ pub const ForkChoice = struct { const found_entries = sorted_entries.items.len > 0; for (sorted_entries.items) |map_entry| { - // Limit the number of distinct AttestationData entries per block. - if (processed_att_data.count() >= self.config.spec.max_attestations_data) break; + // Limit the number of distinct AttestationData entries per block: + // the spec cap, further tightened by the caller's merge-budget cap + // (each distinct entry becomes one component of the Type-2 block + // proof, so the part count directly sets the merge duration; the + // entries are sorted most-valuable-first, so truncation defers the + // least valuable groups to a later proposer). + const group_cap: usize = @min( + @as(usize, self.config.spec.max_attestations_data), + max_groups orelse std.math.maxInt(usize), + ); + if (processed_att_data.count() >= group_cap) { + if (group_cap < self.config.spec.max_attestations_data) { + self.logger.info( + "proposal slot={d}: merge budget caps attestation groups at {d}, remaining candidates deferred", + .{ slot, group_cap }, + ); + } + break; + } try processed_att_data.put(map_entry.att_data.*, {}); @@ -2755,10 +2773,11 @@ pub const ForkChoice = struct { proposer_index: types.ValidatorIndex, parent_root: [32]u8, deadline_ns: ?i64, + max_groups: ?usize, ) !ProposalAttestationsResult { self.mutex.lockShared(); defer self.mutex.unlockShared(); - return self.getProposalAttestationsUnlocked(pre_state, slot, proposer_index, parent_root, deadline_ns); + return self.getProposalAttestationsUnlocked(pre_state, slot, proposer_index, parent_root, deadline_ns, max_groups); } pub fn getAttestationTarget(self: *Self) !types.Checkpoint { @@ -5363,6 +5382,7 @@ test "getProposalAttestations: high-target keys survive cap with stale low-targe 0, parent_root, elapsed_deadline, + null, ); defer { for (result.attestations.slice()) |*att| att.deinit(); diff --git a/pkgs/node/src/lib.zig b/pkgs/node/src/lib.zig index 6d26bbd5a..6810a8c0f 100644 --- a/pkgs/node/src/lib.zig +++ b/pkgs/node/src/lib.zig @@ -10,6 +10,7 @@ pub const BeamNode = nodeFactory.BeamNode; const chainFactory = @import("./chain.zig"); pub const BeamChain = chainFactory.BeamChain; pub const default_proposal_deadline_pct = chainFactory.default_proposal_deadline_pct; +pub const default_proposal_publish_target_intervals = chainFactory.default_proposal_publish_target_intervals; pub const fcFactory = @import("./forkchoice.zig"); pub const testing = @import("./testing.zig"); diff --git a/pkgs/node/src/node.zig b/pkgs/node/src/node.zig index 0632f63a4..7fa4dc97d 100644 --- a/pkgs/node/src/node.zig +++ b/pkgs/node/src/node.zig @@ -69,6 +69,7 @@ const NodeOpts = struct { aggregate_max_inflight: u32 = 4, /// See `chainFactory.ChainOpts.proposal_deadline_pct`. proposal_deadline_pct: u32 = chainFactory.default_proposal_deadline_pct, + proposal_publish_target_intervals: u32 = chainFactory.default_proposal_publish_target_intervals, }; /// blocks_by_root retry backoff (interop storm fix). A requested block root that no peer can @@ -229,6 +230,7 @@ pub const BeamNode = struct { .max_aggregation_children = opts.max_aggregation_children, .aggregate_max_inflight = opts.aggregate_max_inflight, .proposal_deadline_pct = opts.proposal_deadline_pct, + .proposal_publish_target_intervals = opts.proposal_publish_target_intervals, }, network.connected_peers, ) catch |init_err| {