diff --git a/pkgs/cli/src/main.zig b/pkgs/cli/src/main.zig index d58281e35..22e3d27d9 100644 --- a/pkgs/cli/src/main.zig +++ b/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}); diff --git a/pkgs/cli/src/node.zig b/pkgs/cli/src/node.zig index 66480ed1c..1a1dde723 100644 --- a/pkgs/cli/src/node.zig +++ b/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(); diff --git a/pkgs/cli/test/integration.zig b/pkgs/cli/test/integration.zig index 621565e2f..87430d8d2 100644 --- a/pkgs/cli/test/integration.zig +++ b/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,9 +725,6 @@ 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(); @@ -710,6 +732,9 @@ test "SSE events integration test - wait for justification and finalization" { // 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. diff --git a/pkgs/node/src/chain.zig b/pkgs/node/src/chain.zig index ad53ed609..e1e098484 100644 --- a/pkgs/node/src/chain.zig +++ b/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, diff --git a/pkgs/node/src/clock.zig b/pkgs/node/src/clock.zig index 46690851d..e0dca4a56 100644 --- a/pkgs/node/src/clock.zig +++ b/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 { diff --git a/pkgs/node/src/node.zig b/pkgs/node/src/node.zig index f8cffce1f..252c4383b 100644 --- a/pkgs/node/src/node.zig +++ b/pkgs/node/src/node.zig @@ -59,6 +59,9 @@ const NodeOpts = struct { /// `--chain-worker` (bool); `--chain-worker false` is the /// kill-switch for the legacy synchronous path. chain_worker_enabled: bool = true, + /// Event loop for registering the xev.Async watcher that wakes the main + /// loop when the Rust bridge thread enqueues a ReqRespResponseEvent. + loop: ?*xev.Loop = null, }; pub const BeamNode = struct { @@ -94,6 +97,13 @@ pub const BeamNode = struct { batch_pending_parent_roots: std.AutoHashMap(types.Root, u32), batch_pending_parent_roots_lock: zeam_utils.SyncMutex = .{}, + // The Rust bridge thread enqueues ReqRespResponseEvent values here and + // wakes the main libxev loop through async_notifier. + resp_queue: RespQueue = .{}, + resp_queue_mutex: zeam_utils.SyncMutex = .{}, + async_notifier: ?xev.Async = null, + async_completion: xev.Completion = .{}, + /// Range chunks handed to the chain-worker before `onBlock` completes (#893). /// Maps block_root → blocks_by_range request_id for post-import accounting. range_async_chunk_imports: std.AutoHashMap(types.Root, u64), @@ -114,6 +124,35 @@ pub const BeamNode = struct { /// that preserves that invariant. sync_refresh_pending: std.atomic.Value(bool) = std.atomic.Value(bool).init(false), + const RespQueue = BoundedQueue(networks.ReqRespResponseEvent, 256); + + fn BoundedQueue(comptime T: type, comptime capacity: usize) type { + return struct { + buf: [capacity]T = undefined, + head: usize = 0, + tail: usize = 0, + len: usize = 0, + + const BQ = @This(); + + fn push(self: *BQ, item: T) bool { + if (self.len == capacity) return false; + self.buf[self.tail] = item; + self.tail = (self.tail + 1) % capacity; + self.len += 1; + return true; + } + + fn pop(self: *BQ) ?T { + if (self.len == 0) return null; + const item = self.buf[self.head]; + self.head = (self.head + 1) % capacity; + self.len -= 1; + return item; + } + }; + } + const Self = @This(); pub fn init(self: *Self, allocator: Allocator, opts: NodeOpts) !void { @@ -206,10 +245,27 @@ pub const BeamNode = struct { chain.setImportedBlockCallback(self, handleChainImportedBlock); chain.setRejectedBlockCallback(self, handleChainRejectedBlock); + if (opts.loop) |loop| { + var notifier = xev.Async.init() catch |err| { + opts.logger_config.logger(.node).err("failed to init xev.Async notifier: {any}", .{err}); + return err; + }; + notifier.wait(loop, &self.async_completion, Self, self, drainRespQueueCb); + self.async_notifier = notifier; + } + network_init_cleanup = false; } pub fn deinit(self: *Self) void { + if (self.async_notifier) |*notifier| notifier.deinit(); + self.resp_queue_mutex.lock(); + while (self.resp_queue.pop()) |event| { + var owned_event = event; + owned_event.deinit(self.allocator); + } + self.resp_queue_mutex.unlock(); + // Order matters under #890. `chain.deinit()` is what stops/ // joins the chain-worker thread, so any state the worker // callbacks (`handleChainImportedBlock`, @@ -1983,8 +2039,79 @@ pub const BeamNode = struct { } } + fn cloneRespEvent(self: *Self, event: *const networks.ReqRespResponseEvent) !networks.ReqRespResponseEvent { + switch (event.payload) { + .success => |resp| switch (resp) { + .blocks_by_root => |block| { + var cloned: networks.ReqRespResponse = .{ .blocks_by_root = undefined }; + try types.sszClone(self.allocator, types.SignedBlock, block, &cloned.blocks_by_root); + return .{ + .method = event.method, + .request_id = event.request_id, + .payload = .{ .success = cloned }, + }; + }, + .blocks_by_range => |block| { + var cloned: networks.ReqRespResponse = .{ .blocks_by_range = undefined }; + try types.sszClone(self.allocator, types.SignedBlock, block, &cloned.blocks_by_range); + return .{ + .method = event.method, + .request_id = event.request_id, + .payload = .{ .success = cloned }, + }; + }, + .status => |status| { + return .{ + .method = event.method, + .request_id = event.request_id, + .payload = .{ .success = .{ .status = status } }, + }; + }, + }, + .failure => |err_payload| { + const owned_msg = try self.allocator.dupe(u8, err_payload.message); + return .{ + .method = event.method, + .request_id = event.request_id, + .payload = .{ .failure = .{ .code = err_payload.code, .message = owned_msg } }, + }; + }, + .completed => { + return .{ + .method = event.method, + .request_id = event.request_id, + .payload = .completed, + }; + }, + } + } + pub fn onReqRespResponse(ptr: *anyopaque, event: *const networks.ReqRespResponseEvent) anyerror!void { const self: *Self = @ptrCast(@alignCast(ptr)); + + if (self.async_notifier != null) { + const cloned = self.cloneRespEvent(event) catch |err| { + self.logger.warn("failed to clone ReqRespResponseEvent for async queue: {any}", .{err}); + return err; + }; + + self.resp_queue_mutex.lock(); + const pushed = self.resp_queue.push(cloned); + self.resp_queue_mutex.unlock(); + + if (!pushed) { + self.logger.warn("resp_queue full (256), dropping ReqRespResponseEvent request_id={d}", .{event.request_id}); + var to_free = cloned; + to_free.deinit(self.allocator); + return; + } + + self.async_notifier.?.notify() catch |err| { + self.logger.warn("xev.Async.notify() failed: {any}", .{err}); + }; + return; + } + // Slice (a-3): no outer mutex. `handleReqRespResponse` snapshots // the pending request entry under the pending_rpc_requests lock, // then calls `chain.onBlock` (per-resource locks) for the @@ -1993,6 +2120,40 @@ pub const BeamNode = struct { try self.handleReqRespResponse(event); } + fn drainRespQueueCb( + ud: ?*Self, + _: *xev.Loop, + _: *xev.Completion, + r: xev.Async.WaitError!void, + ) xev.CallbackAction { + r catch |err| { + if (ud) |self| { + self.logger.err("xev.Async wait error: {any}", .{err}); + } + return .rearm; + }; + + const self = ud orelse return .rearm; + + while (true) { + self.resp_queue_mutex.lock(); + const maybe_event = self.resp_queue.pop(); + self.resp_queue_mutex.unlock(); + + if (maybe_event) |ev| { + var event = ev; + defer event.deinit(self.allocator); + self.handleReqRespResponse(&event) catch |err| { + self.logger.warn("drainRespQueue: handleReqRespResponse failed: {any}", .{err}); + }; + } else break; + } + + self.flushPendingParentFetches(); + + return .rearm; + } + pub fn getOnGossipCbHandler(self: *Self) !networks.OnGossipCbHandler { return .{ .ptr = self, diff --git a/pkgs/types/src/state.zig b/pkgs/types/src/state.zig index 3f5af102f..08389df63 100644 --- a/pkgs/types/src/state.zig +++ b/pkgs/types/src/state.zig @@ -495,7 +495,9 @@ pub const BeamState = struct { const end_slot_usize: usize = @intCast(target_slot); for (start_slot_usize..end_slot_usize) |slot_usize| { const slot: Slot = @intCast(slot_usize); - if (try utils.IsJustifiableSlot(self.latest_finalized.slot, slot)) { + if (try utils.IsJustifiableSlot(self.latest_finalized.slot, slot) and + !(try utils.isSlotJustified(finalized_slot, &self.justified_slots, slot))) + { can_target_finalize = false; break; }