diff --git a/pkgs/cli/src/main.zig b/pkgs/cli/src/main.zig index b30dda4ea..8706656be 100644 --- a/pkgs/cli/src/main.zig +++ b/pkgs/cli/src/main.zig @@ -89,6 +89,14 @@ pub const NodeCommand = struct { /// aggregator nodes. Aggregators on CPU-rich hosts can pass a value here /// to give the prover more parallelism without rebuilding (#899). @"rayon-threads": ?u32 = null, + /// Minimum (children + gossip-sig) inputs required before the aggregator + /// invokes the recursive STARK prover for an `AttestationData`. Default + /// `2` (post-#908) skips the no-children + single-gossip-sig case where + /// the prover would produce a 1-validator aggregate of zero consensus + /// value. `1` reverts to pre-#908 behavior (always aggregate ≥1 sig). + /// Higher values trade slot latency for fewer sub-threshold aggregates + /// on chatty subnets. See issue #907 finding 4. + @"min-aggregation-inputs": u32 = types.default_min_aggregation_inputs, pub const __shorts__ = .{ .help = .h, @@ -115,6 +123,7 @@ pub const NodeCommand = struct { .@"chain-spec" = "Path to the chain specification file, if unspecified falls back to the default setting", .@"chain-worker" = "Route gossip block + attestation handlers through the dedicated chain-worker thread. On by default; pass `--chain-worker false` to fall back to the legacy synchronous path as a kill-switch.", .@"rayon-threads" = "Override the rayon worker count used by the multisig aggregate prover. If unset, half of the post-system-thread budget goes to the Zig pool and half to rayon. Aggregators in CPU-rich environments benefit from a higher value (e.g. 12 on a 16-vCPU host); non-aggregators can leave it unset.", + .@"min-aggregation-inputs" = "Minimum (children + gossip-sig) inputs required before the aggregator invokes the recursive STARK prover for an AttestationData. Default 2 skips the trivial 'no children + 1 local sig' case (the lone sig is already on the gossip topic, so peers can fold it in directly; building a 1-validator aggregate spends the full prover budget for zero consensus signal). Set 1 to revert to pre-#908 behavior. Higher values trade slot latency for fewer sub-threshold aggregates on chatty subnets.", .help = "Show help information for the node command", }; }; @@ -851,6 +860,7 @@ fn mainInner(init: std.process.Init) !void { .node_registry = node_registry, .db_backend = leancmd.@"db-backend", .rayon_threads = leancmd.@"rayon-threads", + .min_aggregation_inputs = leancmd.@"min-aggregation-inputs", }; defer start_options.deinit(allocator); diff --git a/pkgs/cli/src/node.zig b/pkgs/cli/src/node.zig index be8f4669c..fb7ad898e 100644 --- a/pkgs/cli/src/node.zig +++ b/pkgs/cli/src/node.zig @@ -110,6 +110,17 @@ pub const NodeOptions = struct { /// parallelism without rebuilding (#899). Surfaced as `--rayon-threads` /// on the `zeam node` CLI. rayon_threads: ?u32 = null, + /// Minimum (children + gossip-sig) inputs required before the + /// aggregator pre-filter lets an `AttestationData` reach the FFI. + /// Threaded through to `ForkChoice.min_aggregation_inputs` and + /// applied by `pruneTrivialFromAggregateSnapshot` in + /// `pkgs/node/src/forkchoice.zig` BEFORE + /// `computeAggregatedSignatures` runs (the FFI itself stays + /// spec-pure). Surfaced as `--min-aggregation-inputs` on the + /// `zeam node` CLI; default is + /// `pkgs/types/src/block.zig:default_min_aggregation_inputs`. See + /// issue #907 finding 4. + min_aggregation_inputs: u32 = types.default_min_aggregation_inputs, pub fn deinit(self: *NodeOptions, allocator: std.mem.Allocator) void { for (self.bootnodes) |b| allocator.free(b); @@ -463,6 +474,22 @@ pub const Node = struct { } xmss.setRayonThreads(rayon_threads); + // Log the aggregator threshold on startup so operators can see + // exactly how `--min-aggregation-inputs` was resolved (default vs + // override). The threshold is enforced by the aggregator-side + // pre-filter in `forkchoice.zig:pruneTrivialFromAggregateSnapshot`, + // not inside the spec-pure `computeAggregatedSignatures` FFI. + self.logger.info( + "aggregator threshold: min_aggregation_inputs={d}{s}", + .{ + options.min_aggregation_inputs, + if (options.min_aggregation_inputs != types.default_min_aggregation_inputs) + " (override via --min-aggregation-inputs)" + else + "", + }, + ); + // Pre-warm the XMSS verifier on the main thread before any worker can // call `verifyAggregatedPayload`. The Rust-side verifier setup is // documented as idempotent but is not hardened against first-time-init @@ -488,6 +515,7 @@ pub const Node = struct { .aggregation_subnet_ids = options.aggregation_subnet_ids, .thread_pool = self.thread_pool, .chain_worker_enabled = options.chain_worker_enabled, + .min_aggregation_inputs = options.min_aggregation_inputs, }); errdefer self.beam_node.deinit(); diff --git a/pkgs/node/src/chain.zig b/pkgs/node/src/chain.zig index 859ccadaa..49a23ba52 100644 --- a/pkgs/node/src/chain.zig +++ b/pkgs/node/src/chain.zig @@ -66,6 +66,11 @@ pub const ChainOpts = struct { // Optional shared worker pool for CPU-bound work (signature verification). // When null, the chain falls back to the serial code paths. thread_pool: ?*ThreadPool = null, + /// Surfaces the `--min-aggregation-inputs` CLI flag to the + /// per-`ForkChoice` aggregation threshold. See + /// `pkgs/types/src/block.zig:default_min_aggregation_inputs` for the + /// default and `isTrivialAggregationInput` for the predicate semantics. + min_aggregation_inputs: u32 = types.default_min_aggregation_inputs, }; pub const CachedProcessedBlockInfo = struct { @@ -502,6 +507,7 @@ pub const BeamChain = struct { .anchorState = opts.anchorState, .logger = logger_config.logger(.forkchoice), .thread_pool = opts.thread_pool, + .min_aggregation_inputs = opts.min_aggregation_inputs, }); var states = std.AutoHashMap(types.Root, *RcBeamState).init(allocator); diff --git a/pkgs/node/src/forkchoice.zig b/pkgs/node/src/forkchoice.zig index 1d7c525b5..12be3fe48 100644 --- a/pkgs/node/src/forkchoice.zig +++ b/pkgs/node/src/forkchoice.zig @@ -279,6 +279,13 @@ pub const ForkChoiceParams = struct { anchorState: *const types.BeamState, logger: zeam_utils.ModuleLogger, thread_pool: ?*ThreadPool = null, + /// Minimum (children + gossip-sig) inputs the aggregator pre-filter + /// requires before letting an `att_data` reach + /// `computeAggregatedSignatures`. Threaded through from the + /// `--min-aggregation-inputs` CLI flag and consumed by + /// `pruneTrivialFromAggregateSnapshot` (NOT by the FFI). See + /// `isAggregatorTrivialInput` for the predicate semantics. + min_aggregation_inputs: u32 = types.default_min_aggregation_inputs, }; // Use shared signature map types from types package @@ -334,6 +341,13 @@ pub const ForkChoice = struct { status: ForkChoiceStatus, // Optional shared worker pool used for CPU-heavy attestation compaction. thread_pool: ?*ThreadPool = null, + /// Threshold consumed by `pruneTrivialFromAggregateSnapshot` on + /// every `aggregate*` call to decide which `att_data` entries to + /// drop before invoking `computeAggregatedSignatures`. See + /// `ForkChoiceParams.min_aggregation_inputs`. Defaulted at the field + /// level so test struct literals that don't care about the + /// threshold can omit it. + min_aggregation_inputs: u32 = types.default_min_aggregation_inputs, last_node_tick_time_ms: ?i64, const Self = @This(); @@ -426,6 +440,7 @@ pub const ForkChoice = struct { // checkpoint is observed through block processing. .status = if (opts.anchorState.slot == 0) .ready else .initing, .thread_pool = opts.thread_pool, + .min_aggregation_inputs = opts.min_aggregation_inputs, .last_node_tick_time_ms = null, }; if (fc.status == .initing) { @@ -2107,6 +2122,26 @@ pub const ForkChoice = struct { agg.attestation_signatures.deinit(); }; + // Aggregator-only policy filter: drop `att_data` entries from the + // owned snapshot whose only input is a single (or sub-threshold) + // local gossip sig and which have no peer payload to fold into. + // This MUST live here in the aggregator wrapper, NOT inside + // `computeAggregatedSignatures`: the same FFI function is the + // primitive a future block proposer would use to aggregate the + // exact `att_data` set it chose to include in a block, and a + // proposer must aggregate every chosen `att_data` (blocks carry + // aggregated proofs, not raw sigs) — even a 1-validator one. By + // filtering the snapshot before the FFI runs, the function + // remains spec-pure (aggregate whatever you are given) and the + // policy is opt-in per call site. See review on PR #908 and + // issue #907 finding 4. + pruneTrivialFromAggregateSnapshot( + self.allocator, + &snap, + self.min_aggregation_inputs, + self.logger, + ); + const compute_start_ns = zeam_utils.monotonicTimestampNs(); try agg.computeAggregatedSignatures( &state.validators, @@ -2982,17 +3017,25 @@ test "aggregate prunes attestation signatures" { .slot = 0, }, }; - const attestation = types.Attestation{ - .validator_id = 0, - .data = attestation_data, - }; - const signature = try key_manager.signAttestation(&attestation, allocator); - - try fork_choice.onSignedAttestation(.{ - .validator_id = 0, - .message = attestation_data, - .signature = signature, - }); + // Two signatures from distinct validators on the same AttestationData. + // A single sig with no peer payloads is now dropped from the snapshot + // by the aggregator pre-filter (issue #907 finding 4 — see + // `pruneTrivialFromAggregateSnapshot`) and is left in + // `attestation_signatures` for a future non-trivial pass. Two sigs is + // the minimum non-trivial shape that exercises the aggregation + + // pruning path this test is asserting. + inline for ([_]u32{ 0, 1 }) |validator_id| { + const attestation = types.Attestation{ + .validator_id = validator_id, + .data = attestation_data, + }; + const signature = try key_manager.signAttestation(&attestation, allocator); + try fork_choice.onSignedAttestation(.{ + .validator_id = validator_id, + .message = attestation_data, + .signature = signature, + }); + } const aggregations = try fork_choice.aggregate(&mock_chain.genesis_state); defer { @@ -3072,22 +3115,29 @@ test "aggregate (#890): does not acquire forkchoice main mutex" { }); defer fork_choice.deinit(); - // Seed one signature so `aggregate` has work to do (otherwise an - // empty-input return path could pass without ever touching the - // mutex even before the fix). + // Seed two signatures so `aggregate` has work to do (otherwise an + // empty-input or trivial-input return path could pass without ever + // touching the mutex even before the fix). Two distinct validators + // is the minimum non-trivial shape after the issue #907 finding 4 + // aggregator pre-filter was added — a single sig with no peer + // payloads is now dropped from the snapshot before + // `computeAggregatedSignatures` runs, so it would not exercise the + // FFI-bearing path this mutex-contract test is guarding. const attestation_data = types.AttestationData{ .slot = 0, .head = .{ .root = fork_choice.head.blockRoot, .slot = 0 }, .target = .{ .root = fork_choice.head.blockRoot, .slot = 0 }, .source = .{ .root = fork_choice.head.blockRoot, .slot = 0 }, }; - const attestation = types.Attestation{ .validator_id = 0, .data = attestation_data }; - const signature = try key_manager.signAttestation(&attestation, allocator); - try fork_choice.onSignedAttestation(.{ - .validator_id = 0, - .message = attestation_data, - .signature = signature, - }); + inline for ([_]u32{ 0, 1 }) |validator_id| { + const attestation = types.Attestation{ .validator_id = validator_id, .data = attestation_data }; + const signature = try key_manager.signAttestation(&attestation, allocator); + try fork_choice.onSignedAttestation(.{ + .validator_id = validator_id, + .message = attestation_data, + .signature = signature, + }); + } // Hold the forkchoice main mutex exclusive on the test thread for // the entire aggregator-thread lifetime. If `aggregate` (still) @@ -3690,6 +3740,114 @@ const AggregateSnapshot = struct { } }; +/// Returns true when an `att_data` in the aggregator's snapshot is +/// trivial under the operator's threshold and should be dropped before +/// `computeAggregatedSignatures` runs: +/// - no child payloads (neither `new` nor `known`) for this `att_data`, AND +/// - fewer than `min_inputs` local gossip sigs. +/// +/// This is the AGGREGATOR-side policy. The FFI function +/// `computeAggregatedSignatures` itself is spec-pure and has no notion +/// of this threshold — block proposers would call it with no +/// pre-filtering so a chosen `att_data` with a lone gossip sig still +/// produces a valid (1-validator) aggregate for inclusion in the block. +/// +/// Pure / unit-testable: takes counts, not maps. +fn isAggregatorTrivialInput(num_children: usize, num_gossip_sigs: usize, min_inputs: u32) bool { + if (num_children > 0) return false; + return num_gossip_sigs < @as(usize, min_inputs); +} + +test "isAggregatorTrivialInput: any child makes it non-trivial" { + try std.testing.expect(!isAggregatorTrivialInput(1, 0, 2)); + try std.testing.expect(!isAggregatorTrivialInput(1, 1, 2)); + try std.testing.expect(!isAggregatorTrivialInput(2, 5, 8)); +} + +test "isAggregatorTrivialInput default threshold (2): 0 or 1 sig is trivial, 2+ is not" { + try std.testing.expect(isAggregatorTrivialInput(0, 0, 2)); + try std.testing.expect(isAggregatorTrivialInput(0, 1, 2)); + try std.testing.expect(!isAggregatorTrivialInput(0, 2, 2)); + try std.testing.expect(!isAggregatorTrivialInput(0, 8, 2)); +} + +test "isAggregatorTrivialInput threshold=1 reverts to pre-#908 (only 0+0 trivial)" { + try std.testing.expect(isAggregatorTrivialInput(0, 0, 1)); + try std.testing.expect(!isAggregatorTrivialInput(0, 1, 1)); + try std.testing.expect(!isAggregatorTrivialInput(0, 2, 1)); +} + +test "isAggregatorTrivialInput higher thresholds skip more no-children inputs" { + try std.testing.expect(isAggregatorTrivialInput(0, 1, 3)); + try std.testing.expect(isAggregatorTrivialInput(0, 2, 3)); + try std.testing.expect(!isAggregatorTrivialInput(0, 3, 3)); +} + +/// Walk the owned aggregator snapshot and drop every `att_data` entry +/// from `signatures` that is trivial under `min_inputs` (see +/// `isAggregatorTrivialInput`). The corresponding inner map is freed. +/// +/// Untouched gossip sigs stay in the live `attestation_signatures` map +/// (the snapshot is a clone — pruning the snapshot does NOT touch the +/// live map) and naturally feed the next aggregation pass once any +/// peer payload arrives or another local sig accumulates on the same +/// `att_data`. +/// +/// Logs a single debug line summarising the count + threshold so the +/// startup `aggregator threshold:` info line stays the only visible +/// boot-time mention of this knob; per-pass detail is at debug level. +fn pruneTrivialFromAggregateSnapshot( + allocator: Allocator, + snap: *AggregateSnapshot, + min_inputs: u32, + logger: zeam_utils.ModuleLogger, +) void { + // `min_inputs <= 1` reduces to the pre-existing `0 sigs + 0 + // children` skip already inside `computeAggregatedSignatures`, so + // the pre-filter is a no-op and we can return early. + if (min_inputs <= 1) return; + + // Collect keys to remove first — std.AutoHashMap does not support + // removal during iteration. + var to_drop: std.ArrayList(types.AttestationData) = .empty; + defer to_drop.deinit(allocator); + + var it = snap.signatures.iterator(); + while (it.next()) |entry| { + const att_data = entry.key_ptr.*; + const num_sigs: usize = @intCast(entry.value_ptr.count()); + + const num_children: usize = blk: { + var n: usize = 0; + if (snap.new_payloads.get(att_data)) |list| n += list.items.len; + if (snap.known_payloads.get(att_data)) |list| n += list.items.len; + break :blk n; + }; + + if (isAggregatorTrivialInput(num_children, num_sigs, min_inputs)) { + to_drop.append(allocator, att_data) catch { + // Best-effort filtering: on OOM we skip the rest of the + // sweep and let `computeAggregatedSignatures` aggregate + // whatever's still there. Correctness is preserved + // (over-aggregation, never under), and OOM here is + // exceedingly unlikely given how small the key list is. + return; + }; + } + } + + if (to_drop.items.len == 0) return; + + for (to_drop.items) |att_data| { + snap.signatures.removeAndDeinit(att_data); + } + + logger.debug( + "aggregator pre-filter: dropped {d} trivial att_data (threshold min_aggregation_inputs={d})", + .{ to_drop.items.len, min_inputs }, + ); +} + fn collectCoverageFromPayloads( map: *AggregatedPayloadsMap, slot: types.Slot, diff --git a/pkgs/node/src/node.zig b/pkgs/node/src/node.zig index f8cffce1f..1f0027314 100644 --- a/pkgs/node/src/node.zig +++ b/pkgs/node/src/node.zig @@ -59,6 +59,10 @@ const NodeOpts = struct { /// `--chain-worker` (bool); `--chain-worker false` is the /// kill-switch for the legacy synchronous path. chain_worker_enabled: bool = true, + /// CLI knob (`--min-aggregation-inputs`) for the per-att_data + /// aggregation threshold; see + /// `pkgs/types/src/block.zig:default_min_aggregation_inputs`. + min_aggregation_inputs: u32 = types.default_min_aggregation_inputs, }; pub const BeamNode = struct { @@ -138,6 +142,7 @@ pub const BeamNode = struct { .node_registry = opts.node_registry, .is_aggregator = opts.is_aggregator, .thread_pool = opts.thread_pool, + .min_aggregation_inputs = opts.min_aggregation_inputs, }, network.connected_peers, ) catch |init_err| { diff --git a/pkgs/types/src/block.zig b/pkgs/types/src/block.zig index c954cb21f..843bbdbb4 100644 --- a/pkgs/types/src/block.zig +++ b/pkgs/types/src/block.zig @@ -33,6 +33,20 @@ const json = std.json; const freeJsonValue = utils.freeJsonValue; +/// Default `min_aggregation_inputs` for the aggregator-role pre-filter. +/// +/// Surfaced on the CLI as `--min-aggregation-inputs` (see +/// `pkgs/cli/src/main.zig`) and consumed by the aggregator wrapper in +/// `pkgs/node/src/forkchoice.zig` (NOT by `computeAggregatedSignatures`, +/// which is spec-pure and aggregates whatever it is given). Default +/// `2`: aggregator skips publishing when an `att_data` has only a +/// single local gossip sig and no peer payload, since the raw sig is +/// already on the gossip topic and a 1-validator "aggregate" carries +/// no consensus signal (#907 finding 4). `1` reverts to pre-#908 +/// behavior (always aggregate ≥1 sig). Higher values trade slot +/// latency for fewer sub-threshold aggregates on chatty subnets. +pub const default_min_aggregation_inputs: u32 = 2; + // signatures_map types for aggregation /// Stored signatures_map entry: per-validator signature + slot metadata. @@ -479,7 +493,18 @@ pub const AggregatedAttestationsResult = struct { /// optionally restricted to the supplied attestation slots. /// Step 2: Greedy child proof selection — new_payloads first, then known_payloads as helpers /// Step 3: Collect individual gossip signatures not covered by children - /// Step 4: Recursive aggregate — combine selected children + remaining gossip sigs + /// Step 4: Lone-child clone fast path: `0 gossip + 1 child` clones the + /// lone child as the result without invoking the prover. + /// Step 5: Recursive aggregate — combine selected children + remaining gossip sigs. + /// + /// Spec-pure: this function aggregates whatever it is given. Callers + /// that want to skip trivially-shaped inputs (e.g. the aggregator + /// role wanting to avoid spending the full STARK prover budget on a + /// single-validator "aggregate" that carries no consensus signal — + /// see issue #907 finding 4) must filter the inputs they pass in. + /// Block proposers, by contrast, MUST aggregate every `att_data` + /// they choose to include in a block, even if its only input is a + /// single gossip signature. pub fn computeAggregatedSignatures( self: *Self, validators: *const Validators, diff --git a/pkgs/types/src/lib.zig b/pkgs/types/src/lib.zig index a1d799fbd..ea32eb1d4 100644 --- a/pkgs/types/src/lib.zig +++ b/pkgs/types/src/lib.zig @@ -35,6 +35,7 @@ pub const StoredAggregatedPayload = block.StoredAggregatedPayload; pub const AggregatedPayloadsList = block.AggregatedPayloadsList; pub const AggregatedPayloadsMap = block.AggregatedPayloadsMap; pub const compactAttestations = block.compactAttestations; +pub const default_min_aggregation_inputs = block.default_min_aggregation_inputs; const state = @import("./state.zig"); pub const BeamStateConfig = state.BeamStateConfig;