types, node: parallelize per-att_data aggregate FFI (#907)

build.zig

@@ -214,6 +214,7 @@ pub fn build(b: *Builder) !void {
     zeam_types.addImport("@zeam/utils", zeam_utils);
     zeam_types.addImport("@zeam/metrics", zeam_metrics);
     zeam_types.addImport("@zeam/xmss", zeam_xmss);
+    zeam_types.addImport("@zeam/thread-pool", zeam_thread_pool);
 
     // add zeam-types
     const zeam_configs = b.addModule("@zeam/configs", .{

pkgs/cli/src/node.zig

@@ -178,6 +178,70 @@ pub const Node = struct {
 
     const Self = @This();
 
+    /// Thread roles configured at startup. Counts are logged once init
+    /// completes so operators can verify `--rayon-threads` / aggregator
+    /// auto-tune and compare against `cpu_count`.
+    const StartupThreadBudget = struct {
+        cpu_count: usize,
+        zig_worker_pool: usize,
+        rayon: usize,
+        aggregate_max_inflight: u32,
+        chain_worker: usize,
+        metrics_server: usize,
+        api_server: usize,
+        rayon_source: []const u8,
+
+        fn estimatedTotal(self: @This()) usize {
+            // main (xev slot-driver) + Io.Threaded + libp2p + zig pool + rayon
+            // + chain-worker + metrics + api + slot watchdog (run()).
+            return 1 + 1 + 1 + self.zig_worker_pool + self.rayon + self.chain_worker +
+                self.metrics_server + self.api_server + 1;
+        }
+    };
+
+    fn logStartupThreadBudget(self: *Self, budget: StartupThreadBudget) void {
+        self.logger.info(
+            "startup thread budget: cpu_count={d} estimated_os_threads≈{d}",
+            .{ budget.cpu_count, budget.estimatedTotal() },
+        );
+        self.logger.info(
+            "  main/xev slot-driver: 1 (this thread runs xev.Loop via clock.run)",
+            .{},
+        );
+        self.logger.info(
+            "  std.Io.Threaded:      1 (blocking I/O for Zig pool + disk)",
+            .{},
+        );
+        self.logger.info(
+            "  libp2p rust bridge:   1 (spawned in network.run)",
+            .{},
+        );
+        self.logger.info(
+            "  zig worker pool:      {d} (aggregate_max_inflight={d})",
+            .{ budget.zig_worker_pool, budget.aggregate_max_inflight },
+        );
+        self.logger.info(
+            "  rayon (XMSS prover):  {d}{s}",
+            .{ budget.rayon, budget.rayon_source },
+        );
+        self.logger.info(
+            "  chain-worker:         {d}{s}",
+            .{ budget.chain_worker, if (budget.chain_worker == 0) " (disabled)" else "" },
+        );
+        self.logger.info(
+            "  metrics HTTP server:  {d}",
+            .{budget.metrics_server},
+        );
+        self.logger.info(
+            "  api HTTP server:      {d}",
+            .{budget.api_server},
+        );
+        self.logger.info(
+            "  slot-driver watchdog: 1 (spawned at run() start)",
+            .{},
+        );
+    }
+
     /// Closes the current database, wipes the on-disk rocksdb directory, and
     /// reopens a fresh database at the same path.
     ///
@@ -420,8 +484,16 @@ pub const Node = struct {
         const cpu_count = std.Thread.getCpuCount() catch 2;
         const reserved_system_threads: usize = 4; // main, p2p, api server, metrics server
         const desired_workers = @max(@as(usize, 1), cpu_count -| reserved_system_threads);
-        const zig_worker_budget = @max(@as(usize, 1), (desired_workers + 1) / 2);
-        const worker_count = @min(zig_worker_budget, @as(usize, ThreadPool.max_thread_count));
+
+        // Aggregators: XMSS recursive prove (rayon) is the per-slot bottleneck. Keep a
+        // small Zig pool for capped in-flight aggregate workers (#907).
+        const worker_count = if (options.is_aggregator) blk: {
+            const aggregator_zig_workers = @max(@as(usize, 2), desired_workers / 4);
+            break :blk @min(@as(usize, ThreadPool.max_thread_count), aggregator_zig_workers);
+        } else blk: {
+            const zig_worker_budget = @max(@as(usize, 1), (desired_workers + 1) / 2);
+            break :blk @min(zig_worker_budget, @as(usize, ThreadPool.max_thread_count));
+        };
         self.thread_pool = try ThreadPool.init(.{
             .allocator = allocator,
             .io = std.Io.Threaded.global_single_threaded.io(),
@@ -447,17 +519,23 @@ pub const Node = struct {
         // pool is initialized lazily on first use.
         const rayon_threads = if (options.rayon_threads) |override|
             @max(@as(usize, 1), @as(usize, override))
+        else if (options.is_aggregator)
+            desired_workers
         else
             @max(@as(usize, 1), desired_workers -| worker_count);
-        self.logger.info(
-            "thread pools: cpu_count={d} zig_workers={d} rayon_threads={d}{s}",
-            .{
-                cpu_count,
-                worker_count,
-                rayon_threads,
-                if (options.rayon_threads != null) " (rayon override via --rayon-threads)" else "",
-            },
-        );
+        // One outer aggregate worker at a time on aggregators. Parallelize the
+        // ~11s STARK work per att_data inside that worker
+        // (`computeAggregatedSignatures` thread pool), not across independent
+        // interval workers. Multiple in-flight outer workers can snapshot the
+        // same gossip sigs before either commits and publish duplicate
+        // aggregates (PR #920 review).
+        const aggregate_max_inflight: u32 = if (options.is_aggregator) 1 else 4;
+        const rayon_source: []const u8 = if (options.rayon_threads != null)
+            " (--rayon-threads override)"
+        else if (options.is_aggregator)
+            " (aggregator auto-tune: full post-system budget)"
+        else
+            " (non-aggregator auto-split)";
         // Operator-typo guard for --rayon-threads (review feedback on #903).
         // Rayon tolerates over-subscription, but values like `--rayon-threads 160`
         // on a 4-vCPU box silently degrade throughput. Warn (don't reject) so the
@@ -474,6 +552,12 @@ pub const Node = struct {
         }
         xmss.setRayonThreads(rayon_threads);
 
+        if (options.is_aggregator) {
+            xmss.setupProver() catch |err| {
+                self.logger.warn("xmss prover setup failed: {any}; aggregation may be unavailable", .{err});
+            };
+        }
+
         // 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
@@ -516,6 +600,7 @@ pub const Node = struct {
             .thread_pool = self.thread_pool,
             .chain_worker_enabled = options.chain_worker_enabled,
             .min_aggregation_inputs = options.min_aggregation_inputs,
+            .aggregate_max_inflight = aggregate_max_inflight,
         });
         errdefer self.beam_node.deinit();
 
@@ -563,6 +648,17 @@ pub const Node = struct {
             zeam_metrics.metrics.lean_node_info.set(.{ .name = "zeam", .version = build_options.version }, 1) catch {};
         }
 
+        self.logStartupThreadBudget(.{
+            .cpu_count = cpu_count,
+            .zig_worker_pool = worker_count,
+            .rayon = rayon_threads,
+            .aggregate_max_inflight = aggregate_max_inflight,
+            .chain_worker = if (options.chain_worker_enabled) 1 else 0,
+            .metrics_server = if (options.metrics_enable) 1 else 0,
+            .api_server = if (options.metrics_enable) 1 else 0,
+            .rayon_source = rayon_source,
+        });
+
         self.logger = options.logger_config.logger(.node);
     }
 

pkgs/cli/src/test_driver.zig

@@ -466,7 +466,7 @@ pub fn initForkChoiceDriver(
     zeam_utils.hashTreeRoot(types.BeamBlock, anchor_block, &anchor_block_root, driver_allocator) catch
         return error.HashFailed;
 
-    var test_thread_pool = try node.testing.initTestThreadPool(driver_allocator);
+    var test_thread_pool = try @import("@zeam/node").testing.initTestThreadPool(driver_allocator);
     errdefer test_thread_pool.deinit();
 
     // Init fork choice (uses anchor_state_ptr for anchorState)

pkgs/node/src/chain.zig

@@ -386,8 +386,9 @@ pub const BeamChain = struct {
     /// In-flight aggregate worker count (issue #907). `submitAggregateOnInterval`
     /// atomically checks against `aggregate_max_inflight` before spawning so the
     /// libxev interval tick never blocks. `aggregateImpl` decrements on exit.
-    /// Concurrent runs are safe by construction — see the three-phase
-    /// snapshot/compute/merge contract on `forkchoice.aggregateUnlocked`.
+    /// Aggregators keep this at 1; per-att_data FFI parallelism lives inside
+    /// `aggregateUnlocked`. Values > 1 rely on commit-time duplicate suppression
+    /// in `forkchoice.aggregateUnlocked` (see PR #920 review).
     aggregate_inflight: std.atomic.Value(u32) = .init(0),
 
     /// Joined at `deinit` so aggregate workers cannot outlive chain state.

pkgs/node/src/forkchoice.zig

@@ -2106,12 +2106,12 @@ pub const ForkChoice = struct {
     ///      pairs that existed at snapshot time (entries added during
     ///      phase 2 stay so the next aggregator pass can consume them).
     ///
-    /// Concurrent callers are safe by construction: the three-phase contract
-    /// above (snapshot under `signatures_mutex` → lock-free compute on owned
-    /// clones → merge under `signatures_mutex`) ensures two aggregations
-    /// cannot race on the live maps. `submitAggregateOnInterval`'s
-    /// `aggregate_max_inflight` cap is a CPU-budget knob layered on top, not
-    /// a correctness mechanism.
+    /// The three-phase contract (snapshot under `signatures_mutex` → lock-free
+    /// compute on owned clones → merge under `signatures_mutex`) prevents map
+    /// races but not duplicate publishes when two workers snapshot the same
+    /// gossip sigs before either commits. Aggregators keep
+    /// `aggregate_max_inflight` at 1; commit also suppresses merge/publish for
+    /// att_data whose snapshot gossip vids are already gone from the live map.
     fn aggregateUnlocked(self: *Self, state_opt: ?*const types.BeamState, slot_filter: ?[]const types.Slot) ![]types.SignedAggregatedAttestation {
         const state = state_opt orelse return try self.allocator.alloc(types.SignedAggregatedAttestation, 0);
         const agg_timer = zeam_metrics.lean_committee_signatures_aggregation_time_seconds.start();
@@ -2171,6 +2171,7 @@ pub const ForkChoice = struct {
             &snap.new_payloads,
             &snap.known_payloads,
             slot_filter,
+            self.thread_pool,
         );
         observeAggregateBuildPhase("compute_ffi", compute_start_ns);
 
@@ -2253,6 +2254,31 @@ pub const ForkChoice = struct {
             self.signatures_mutex.lock();
             defer self.signatures_mutex.unlock();
 
+            // Drop att_data whose snapshot gossip inputs were already consumed
+            // by an earlier in-flight worker (defense when aggregate_max_inflight > 1).
+            var results_write_idx: usize = 0;
+            var results_idx: usize = 0;
+            while (results_idx < results.items.len) : (results_idx += 1) {
+                const att_data = results.items[results_idx].data;
+                if (shouldSuppressDuplicateAggregateCommit(&snap, att_data, &self.attestation_signatures)) {
+                    results.items[results_idx].deinit();
+                    if (new_payloads_local.fetchRemove(att_data)) |kv| {
+                        var list = kv.value;
+                        deinitAggregatedPayloadsList(self.allocator, &list);
+                    }
+                    self.logger.debug(
+                        "suppress duplicate aggregate commit att_data slot={d} (snapshot gossip already consumed)",
+                        .{att_data.slot},
+                    );
+                } else {
+                    if (results_write_idx != results_idx) {
+                        results.items[results_write_idx] = results.items[results_idx];
+                    }
+                    results_write_idx += 1;
+                }
+            }
+            results.shrinkRetainingCapacity(results_write_idx);
+
             // MERGE (NOT replace). The previous shape did
             //   deinitAggregatedPayloadsMap(allocator, &self.latest_new_aggregated_payloads);
             //   self.latest_new_aggregated_payloads = new_payloads_local;
@@ -3795,6 +3821,95 @@ test "isAggregatorTrivialInput higher thresholds skip more no-children inputs" {
     try std.testing.expect(!isAggregatorTrivialInput(0, 3, 3));
 }
 
+/// Returns true when this worker's snapshot gossip vids for `att_data` were
+/// already consumed by an earlier aggregate commit, so merging/publishing this
+/// pass would duplicate aggregates (PR #920 review).
+fn shouldSuppressDuplicateAggregateCommit(
+    snap: *const AggregateSnapshot,
+    att_data: types.AttestationData,
+    live_sigs: *const types.SignaturesMap,
+) bool {
+    const snap_inner = snap.signatures.get(att_data) orelse return false;
+    if (snap_inner.count() == 0) return false;
+    const live_inner = live_sigs.get(att_data) orelse return true;
+    var it = snap_inner.iterator();
+    while (it.next()) |entry| {
+        if (live_inner.contains(entry.key_ptr.*)) return false;
+    }
+    return true;
+}
+
+test "shouldSuppressDuplicateAggregateCommit: no snap gossip never suppresses" {
+    const allocator = std.testing.allocator;
+    const zero_root = std.mem.zeroes(types.Root);
+    const att_data = types.AttestationData{
+        .slot = 1,
+        .head = .{ .root = zero_root, .slot = 0 },
+        .target = .{ .root = zero_root, .slot = 0 },
+        .source = .{ .root = zero_root, .slot = 0 },
+    };
+    var snap = AggregateSnapshot{
+        .signatures = types.SignaturesMap.init(allocator),
+        .new_payloads = AggregatedPayloadsMap.init(allocator),
+        .known_payloads = AggregatedPayloadsMap.init(allocator),
+    };
+    defer snap.deinit(allocator);
+    var live = types.SignaturesMap.init(allocator);
+    defer live.deinit();
+    try std.testing.expect(!shouldSuppressDuplicateAggregateCommit(&snap, att_data, &live));
+}
+
+test "shouldSuppressDuplicateAggregateCommit: snap gossip still live does not suppress" {
+    const allocator = std.testing.allocator;
+    const zero_root = std.mem.zeroes(types.Root);
+    const att_data = types.AttestationData{
+        .slot = 1,
+        .head = .{ .root = zero_root, .slot = 0 },
+        .target = .{ .root = zero_root, .slot = 0 },
+        .source = .{ .root = zero_root, .slot = 0 },
+    };
+    const stored_sig = types.StoredSignature{ .slot = 1, .signature = std.mem.zeroes(@TypeOf(@as(types.StoredSignature, undefined).signature)) };
+    var snap = AggregateSnapshot{
+        .signatures = types.SignaturesMap.init(allocator),
+        .new_payloads = AggregatedPayloadsMap.init(allocator),
+        .known_payloads = AggregatedPayloadsMap.init(allocator),
+    };
+    defer snap.deinit(allocator);
+    var snap_inner = types.SignaturesMap.InnerMap.init(allocator);
+    try snap_inner.put(0, stored_sig);
+    try snap_inner.put(1, stored_sig);
+    try snap.signatures.put(att_data, snap_inner);
+    var live = types.SignaturesMap.init(allocator);
+    defer live.deinit();
+    try live.addSignature(att_data, 0, stored_sig);
+    try std.testing.expect(!shouldSuppressDuplicateAggregateCommit(&snap, att_data, &live));
+}
+
+test "shouldSuppressDuplicateAggregateCommit: snap gossip fully consumed suppresses" {
+    const allocator = std.testing.allocator;
+    const zero_root = std.mem.zeroes(types.Root);
+    const att_data = types.AttestationData{
+        .slot = 1,
+        .head = .{ .root = zero_root, .slot = 0 },
+        .target = .{ .root = zero_root, .slot = 0 },
+        .source = .{ .root = zero_root, .slot = 0 },
+    };
+    const stored_sig = types.StoredSignature{ .slot = 1, .signature = std.mem.zeroes(@TypeOf(@as(types.StoredSignature, undefined).signature)) };
+    var snap = AggregateSnapshot{
+        .signatures = types.SignaturesMap.init(allocator),
+        .new_payloads = AggregatedPayloadsMap.init(allocator),
+        .known_payloads = AggregatedPayloadsMap.init(allocator),
+    };
+    defer snap.deinit(allocator);
+    var snap_inner = types.SignaturesMap.InnerMap.init(allocator);
+    try snap_inner.put(0, stored_sig);
+    try snap_inner.put(1, stored_sig);
+    try snap.signatures.put(att_data, snap_inner);
+    var live = types.SignaturesMap.init(allocator);
+    defer live.deinit();
+    try std.testing.expect(shouldSuppressDuplicateAggregateCommit(&snap, att_data, &live));
+}
+
 /// 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.

pkgs/node/src/node.zig

@@ -62,6 +62,8 @@ const NodeOpts = struct {
     /// aggregation threshold; see
     /// `pkgs/types/src/block.zig:default_min_aggregation_inputs`.
     min_aggregation_inputs: u32 = types.default_min_aggregation_inputs,
+    /// Soft cap on concurrent aggregate workers (`BeamChain.aggregate_inflight`).
+    aggregate_max_inflight: u32 = 4,
 };
 
 pub const BeamNode = struct {
@@ -142,6 +144,7 @@ pub const BeamNode = struct {
                 .is_aggregator = opts.is_aggregator,
                 .thread_pool = opts.thread_pool,
                 .min_aggregation_inputs = opts.min_aggregation_inputs,
+                .aggregate_max_inflight = opts.aggregate_max_inflight,
             },
             network.connected_peers,
         ) catch |init_err| {