feat: implement asynchronous response handling and lock-free RPC serving

pkgs/cli/src/main.zig

@@ -673,6 +673,7 @@ fn mainInner(init: std.process.Init) !void {
                 .node_registry = registry_1,
                 .is_aggregator = beamcmd.@"is-aggregator",
                 .thread_pool = thread_pool,
+                .loop = loop,
             });
 
             if (api_server_handle) |handle| {
@@ -694,6 +695,7 @@ fn mainInner(init: std.process.Init) !void {
                 .node_registry = registry_2,
                 .is_aggregator = false,
                 .thread_pool = thread_pool,
+                .loop = loop,
             });
 
             // Node 3 setup - delayed start for initial sync testing
@@ -713,6 +715,7 @@ fn mainInner(init: std.process.Init) !void {
                 .node_registry = registry_3,
                 .is_aggregator = false,
                 .thread_pool = thread_pool,
+                .loop = loop,
             });
 
             // Delayed runner - starts both network3 and node3 together
@@ -730,7 +733,7 @@ fn mainInner(init: std.process.Init) !void {
                     if (self.started) return;
 
                     // Wait until finalization has advanced beyond genesis on the reference node
-                    const finalized_slot = self.reference_node.chain.forkChoice.fcStore.latest_finalized.slot;
+                    const finalized_slot = self.reference_node.chain.forkChoice.getLatestFinalized().slot;
                     if (finalized_slot == 0) return;
 
                     std.debug.print("\n=== STARTING NODE 3 (delayed sync node) at interval {d} ===\n", .{interval});

pkgs/cli/src/node.zig

@@ -447,6 +447,7 @@ pub const Node = struct {
             .aggregation_subnet_ids = options.aggregation_subnet_ids,
             .thread_pool = self.thread_pool,
             .chain_worker_enabled = options.chain_worker_enabled,
+            .loop = &self.loop,
         });
         errdefer self.beam_node.deinit();
 

pkgs/cli/test/integration.zig

@@ -7,6 +7,11 @@ const constants = @import("cli_constants");
 const error_handler = @import("error_handler");
 const ErrorHandler = error_handler.ErrorHandler;
 
+const BeamSimProcess = struct {
+    child: *process.Child,
+    run_dir: []u8,
+};
+
 /// Verify that the Zeam executable exists and return its path
 /// Includes detailed debugging output if the executable is not found
 fn getZeamExecutable() ![]const u8 {
@@ -53,19 +58,33 @@ fn getZeamExecutable() ![]const u8 {
 /// Helper function to start a beam simulation node and wait for it to be ready
 /// Handles the complete process lifecycle: creation, spawning, and waiting for readiness
 /// Returns the process handle for cleanup, or error if startup fails
-fn spinBeamSimNode(allocator: std.mem.Allocator, exe_path: []const u8) !*process.Child {
+fn spinBeamSimNode(allocator: std.mem.Allocator, exe_path: []const u8) !BeamSimProcess {
     const io = std.testing.io;
 
+    const run_dir = try std.fmt.allocPrint(allocator, ".zig-cache/integration-run-{d}", .{zeam_utils.monotonicTimestampNs()});
+    errdefer allocator.free(run_dir);
+    try std.Io.Dir.cwd().createDirPath(io, run_dir);
+    errdefer std.Io.Dir.cwd().deleteTree(io, run_dir) catch {};
+
+    const cwd = try std.process.currentPathAlloc(io, allocator);
+    defer allocator.free(cwd);
+    const child_exe_path = if (std.fs.path.isAbsolute(exe_path))
+        exe_path
+    else
+        try std.fmt.allocPrint(allocator, "{s}/{s}", .{ cwd, exe_path });
+    defer if (!std.fs.path.isAbsolute(exe_path)) allocator.free(child_exe_path);
+
     // Set up process with beam command and mock network
-    const args = [_][]const u8{ exe_path, "beam", "--mockNetwork", "true", "--is-aggregator", "true" };
+    const args = [_][]const u8{ child_exe_path, "beam", "--mockNetwork", "true", "--is-aggregator", "true" };
     const cli_process = try allocator.create(process.Child);
+    errdefer allocator.destroy(cli_process);
 
     // Start the process
     cli_process.* = process.spawn(io, .{
         .argv = &args,
+        .cwd = .{ .path = run_dir },
     }) catch |err| {
         std.debug.print("ERROR: Failed to spawn process: {}\n", .{err});
-        allocator.destroy(cli_process);
         return err;
     };
 
@@ -163,11 +182,15 @@ fn spinBeamSimNode(allocator: std.mem.Allocator, exe_path: []const u8) !*process
         std.debug.print("INFO: Terminated process after startup timeout\n", .{});
 
         // Server not ready, cleanup and return error
+        std.Io.Dir.cwd().deleteTree(io, run_dir) catch {};
         allocator.destroy(cli_process);
         return error.ServerStartupTimeout;
     }
 
-    return cli_process;
+    return .{
+        .child = cli_process,
+        .run_dir = run_dir,
+    };
 }
 
 /// Wait for node to start and be ready for activity
@@ -581,11 +604,13 @@ const SSEClient = struct {
 };
 
 /// Clean up a process created by spinBeamSimNode
-fn cleanupProcess(allocator: std.mem.Allocator, cli_process: *process.Child) void {
+fn cleanupProcess(allocator: std.mem.Allocator, sim_process: BeamSimProcess) void {
     const io = std.testing.io;
-    cli_process.kill(io);
+    sim_process.child.kill(io);
     // cli_process.wait(io) catch {};
-    allocator.destroy(cli_process);
+    allocator.destroy(sim_process.child);
+    std.Io.Dir.cwd().deleteTree(io, sim_process.run_dir) catch {};
+    allocator.free(sim_process.run_dir);
 }
 
 test "CLI beam command with mock network - complete integration test" {
@@ -633,7 +658,7 @@ test "admin aggregator endpoint - GET returns seed, POST toggles at runtime" {
     // The API server comes up before the chain is wired in (503 until
     // `setChain` is called inside main.zig after validator key generation).
     // Poll until the chain is ready, then assert the baseline.
-    const chain_ready_deadline_ms: i64 = 60_000;
+    const chain_ready_deadline_ms: i64 = 180_000;
     const poll_start = zeam_utils.unixTimestampMillis();
     var get_before = try zeam_request.getAggregator();
     while (get_before.status != .ok) {
@@ -700,16 +725,16 @@ test "SSE events integration test - wait for justification and finalization" {
     const cli_process = try spinBeamSimNode(allocator, exe_path);
     defer cleanupProcess(allocator, cli_process);
 
-    // Wait for node to be fully active
-    waitForNodeStart();
-
     // Create SSE client
     var sse_client = try SSEClient.init(allocator);
     defer sse_client.deinit();
 
     // Connect to SSE endpoint
     try sse_client.connect();
 
+    // Wait for node activity after subscribing so one-shot chain events are not missed.
+    waitForNodeStart();
+
     std.debug.print("INFO: Connected to SSE endpoint, waiting for events...\n", .{});
 
     // Read events until justification, any finalization, AND explicit node3 finalization sync are verified, or timeout.

pkgs/node/src/chain.zig

@@ -350,6 +350,10 @@ pub const BeamChain = struct {
     // during `processPendingBlocks` (drain path) so the queue self-cleans.
     pending_blocks: std.ArrayList(PendingBlockEntry),
 
+    /// Cached req/resp status snapshot, updated when head/finalization move.
+    cached_status_mutex: zeam_utils.SyncMutex = .{},
+    cached_status: types.Status,
+
     // Per-resource locks (slice a-2 of #803). See
     // `docs/threading_refactor_slice_a.md` for the lock-hierarchy contract:
     //   tier 3: states_lock
@@ -557,6 +561,12 @@ pub const BeamChain = struct {
             .public_key_cache = try xmss.PublicKeyCache.init(allocator, @intCast(opts.config.genesis.numValidators())),
             .root_to_slot_cache = types.RootToSlotCache.init(allocator),
             .thread_pool = opts.thread_pool,
+            .cached_status = .{
+                .finalized_root = fork_choice.fcStore.latest_finalized.root,
+                .finalized_slot = fork_choice.fcStore.latest_finalized.slot,
+                .head_root = fork_choice.head.blockRoot,
+                .head_slot = fork_choice.head.slot,
+            },
             // pending_blocks is the future-slot queue (issue #788). It's an
             // unmanaged ArrayList, so default-init to `.empty`; the lock
             // below guards mutation. Required field — without it the
@@ -2363,6 +2373,12 @@ pub const BeamChain = struct {
         // Only forkchoice tick failure means the chain clock did not advance.
         try self.forkChoice.onInterval(time_intervals, has_proposal);
 
+        {
+            const head = self.forkChoice.getHead();
+            const finalized = self.forkChoice.getLatestFinalized();
+            self.updateCachedStatus(head, finalized);
+        }
+
         if (interval == 1) {
             // interval to attest so we should put out the chain status information to the user along with
             // latest head which most likely should be the new block received and processed
@@ -3595,6 +3611,8 @@ pub const BeamChain = struct {
         const latest_justified = self.forkChoice.getLatestJustified();
         const latest_finalized = self.forkChoice.getLatestFinalized();
 
+        self.updateCachedStatus(new_head, latest_finalized);
+
         // 8. Asap emit justification/finalization events based on forkchoice store.
         //    `events_lock` (tier 5c) covers the read-modify-write of
         //    `last_emitted_justified`, `last_emitted_finalized`, and (later)
@@ -4401,10 +4419,15 @@ pub const BeamChain = struct {
     }
 
     pub fn getStatus(self: *Self) types.Status {
-        const finalized = self.forkChoice.getLatestFinalized();
-        const head = self.forkChoice.getHead();
+        self.cached_status_mutex.lock();
+        defer self.cached_status_mutex.unlock();
+        return self.cached_status;
+    }
 
-        return .{
+    fn updateCachedStatus(self: *Self, head: types.ProtoBlock, finalized: types.Checkpoint) void {
+        self.cached_status_mutex.lock();
+        defer self.cached_status_mutex.unlock();
+        self.cached_status = .{
             .finalized_root = finalized.root,
             .finalized_slot = finalized.slot,
             .head_root = head.blockRoot,

pkgs/node/src/clock.zig

@@ -36,6 +36,7 @@ pub const Clock = struct {
     allocator: Allocator,
 
     timer: xev.Timer,
+    tick_completion: xev.Completion = .{},
     logger: zeam_utils.ModuleLogger,
 
     const Self = @This();
@@ -123,41 +124,45 @@ pub const Clock = struct {
             self.current_interval += 1;
         }
 
-        const next_interval_time_ms: isize = self.current_interval_time_ms + constants.SECONDS_PER_INTERVAL_MS;
-        const time_to_next_interval_ms: usize = @intCast(next_interval_time_ms - time_now_ms);
-
         for (0..self.on_interval_cbs.items.len) |i| {
             const cbWrapper = self.on_interval_cbs.items[i];
-            cbWrapper.interval = self.current_interval + 1;
+            cbWrapper.interval = self.current_interval;
+            cbWrapper.onInterval() catch |err| {
+                self.logger.err("failed to call onInterval subscriber: {any}", .{err});
+            };
+        }
+
+        self.scheduleNextTick(time_now_ms);
+    }
 
-            self.timer.run(
-                self.events.loop,
-                &cbWrapper.c,
-                time_to_next_interval_ms,
-                OnIntervalCbWrapper,
-                cbWrapper,
-                (struct {
-                    fn callback(
-                        ud: ?*OnIntervalCbWrapper,
-                        _: *xev.Loop,
-                        _: *xev.Completion,
-                        r: xev.Timer.RunError!void,
-                    ) xev.CallbackAction {
-                        r catch |err| {
-                            // Canceled is expected when tickInterval re-arms a still-pending
-                            // completion (the old fire arrives with Canceled).  Swallow it
-                            // silently; the new timer is already scheduled.
-                            if (err != error.Canceled) std.debug.panic("unexpected xev timer error: {}", .{err});
-                            return .disarm;
-                        };
-                        if (ud) |cb_wrapper| {
-                            _ = cb_wrapper.onInterval() catch void;
-                        }
+    fn scheduleNextTick(self: *Self, time_now_ms: isize) void {
+        const next_interval_time_ms: isize = self.current_interval_time_ms + constants.SECONDS_PER_INTERVAL_MS;
+        const time_to_next_interval_ms: usize = @intCast(next_interval_time_ms - time_now_ms);
+
+        self.timer.run(
+            self.events.loop,
+            &self.tick_completion,
+            time_to_next_interval_ms,
+            Self,
+            self,
+            (struct {
+                fn callback(
+                    ud: ?*Self,
+                    _: *xev.Loop,
+                    _: *xev.Completion,
+                    r: xev.Timer.RunError!void,
+                ) xev.CallbackAction {
+                    r catch |err| {
+                        if (err != error.Canceled) std.debug.panic("unexpected xev timer error: {}", .{err});
                         return .disarm;
+                    };
+                    if (ud) |clock| {
+                        clock.tickInterval();
                     }
-                }).callback,
-            );
-        }
+                    return .disarm;
+                }
+            }).callback,
+        );
     }
 
     pub fn run(self: *Self) !void {