node, types: skip recursive aggregation for 1-sig + 0-child trivial inputs (#907)

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);

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();
 

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);

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,

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| {