From 384f60d9dad706365b96310dac075899ed485947 Mon Sep 17 00:00:00 2001 From: Alex Russell Date: Sun, 13 Sep 2026 12:32:37 +0200 Subject: [PATCH] fix(dispatch): add timeouts to worker dispatch and health probes A worker that accepts a connection and then never answers blocks the caller forever. std.http.Client has no timeout: FetchOptions carries none, and the only `timeout` field in the http client (ConnectTcpOptions.timeout) is never read. Both callers sit on long-lived loops. probeWorker runs on the engine's health pass and dispatchStep runs on the engine and tracker ticks, so one silent worker stalls the loop that is supposed to notice it. The circuit breaker cannot trip, because the health check is the thing that is stuck. With the tracker enabled the tick holds state.mutex across the call, so the whole HTTP API stops answering. Each request now runs on a dedicated threaded Io via Io.concurrent and is cancelled once its deadline passes. Io.concurrent is used rather than Io.async because async may run the call inline when the pool is saturated (on a 1-CPU host its default limit is 0), which would silently disable the deadline. The process-wide Io is left alone. Two EngineConfig knobs, 0 disables either: worker_probe_timeout_ms 5000 a worker that cannot answer a bare GET promptly is not healthy default_timeout_ms 1800000 already declared but never read; wired up here. The default is raised from 300000 because agent turns legitimately run for minutes, and a deadline shorter than a real turn would break working dispatches. Tested against a worker that accepts and never responds. Before: failures stay at 0 and the worker stays active indefinitely, and with the tracker on, /health, /workers and /tracker/status stop responding. After: the worker reaches 'dead' after three probe failures and the API keeps answering. Adds a unit test covering the probe path. Co-Authored-By: Claude Opus 5.5 --- src/config.zig | 7 +- src/dispatch.zig | 190 ++++++++++++++++++++++++++++++++++++++++++----- src/main.zig | 6 ++ 3 files changed, 182 insertions(+), 21 deletions(-) diff --git a/src/config.zig b/src/config.zig index 413f563..c78f399 100644 --- a/src/config.zig +++ b/src/config.zig @@ -17,9 +17,14 @@ pub const WorkerConfig = struct { pub const EngineConfig = struct { poll_interval_ms: u32 = 500, - default_timeout_ms: u32 = 300000, + /// Deadline for one dispatch to a worker, in ms. 0 disables it. + /// Agent turns run for minutes, so this is a backstop against a wedged + /// socket, not a latency budget. + default_timeout_ms: u32 = 1800000, default_max_attempts: u32 = 1, health_check_interval_ms: u32 = 30000, + /// Deadline for one worker health probe, in ms. 0 disables it. + worker_probe_timeout_ms: u32 = 5000, worker_failure_threshold: u32 = 3, worker_circuit_breaker_ms: u32 = 60000, retry_base_delay_ms: u32 = 1000, diff --git a/src/dispatch.zig b/src/dispatch.zig index 107b978..76a3c69 100644 --- a/src/dispatch.zig +++ b/src/dispatch.zig @@ -9,6 +9,131 @@ const worker_response = @import("worker_response.zig"); const redis_client = @import("redis_client.zig"); const mqtt_client = @import("mqtt_client.zig"); +// ── Outbound HTTP timeouts ──────────────────────────────────────────── +// +// std.http.Client has no timeout: FetchOptions carries none, and the only +// `timeout` field in the http client (ConnectTcpOptions.timeout) is never read. +// A worker that accepts a connection and then goes silent therefore blocks the +// caller forever. Both callers here run on long-lived loops — probeWorker on +// the engine's health-check pass, dispatchStep on the engine and tracker ticks — +// so one silent worker stalls the loop that is supposed to notice it, and the +// circuit breaker can never trip. +// +// The fix runs each request under a deadline on a dedicated threaded Io and +// cancels it if the deadline passes. The process-wide Io is left alone: it is +// single-threaded, where `async` runs the call inline and cancellation is +// unreachable. + +var outbound_io_state: std.Io.Threaded = undefined; +var outbound_io_ready: std.atomic.Value(bool) = .init(false); +var outbound_io_mutex: std_compat.sync.Mutex = .{}; + +fn outboundIo() std.Io { + if (!outbound_io_ready.load(.acquire)) { + outbound_io_mutex.lock(); + defer outbound_io_mutex.unlock(); + if (!outbound_io_ready.load(.acquire)) { + outbound_io_state = .init(std.heap.smp_allocator, .{}); + outbound_io_ready.store(true, .release); + } + } + return outbound_io_state.io(); +} + +/// Deadline for a dispatch to a worker. Agent turns legitimately run for +/// minutes, so this is a backstop against a wedged socket rather than a +/// latency budget; a value below a real turn would break working dispatches. +var dispatch_timeout_ms: u64 = 1_800_000; + +/// Deadline for a health probe. Short: a worker that cannot answer a bare GET +/// promptly is not healthy, and this is the signal the circuit breaker needs. +var probe_timeout_ms: u64 = 5_000; + +/// Called once at startup from the loaded config. +pub fn configureTimeouts(dispatch_ms: u64, probe_ms: u64) void { + dispatch_timeout_ms = dispatch_ms; + probe_timeout_ms = probe_ms; +} + +const FetchCall = struct { + io: std.Io, + allocator: std.mem.Allocator, + url: []const u8, + method: std.http.Method, + payload: ?[]const u8, + extra_headers: []const std.http.Header, + content_type_json: bool, + response_writer: *std.Io.Writer, + + /// Set by the worker task on every exit path. The future's own return value + /// is not trusted after a cancel; this flag is what distinguishes a + /// completed request from an abandoned one. + finished: std.atomic.Value(bool) = .init(false), + status: u16 = 0, + failed: bool = false, +}; + +fn fetchTask(call: *FetchCall) void { + defer call.finished.store(true, .release); + var client: std.http.Client = .{ .allocator = call.allocator, .io = call.io }; + defer client.deinit(); + const result = client.fetch(.{ + .location = .{ .url = call.url }, + .method = call.method, + .payload = call.payload, + .response_writer = call.response_writer, + .extra_headers = call.extra_headers, + .headers = .{ + .content_type = if (call.content_type_json) + .{ .override = "application/json" } + else + .default, + }, + }) catch { + call.failed = true; + return; + }; + call.status = @intFromEnum(result.status); +} + +const FetchTimeoutError = error{ Timeout, RequestFailed }; + +/// Runs one HTTP request with a deadline. Returns the status code, or +/// error.Timeout if the deadline passed before the request finished. +/// +/// The request runs via `Io.concurrent`, not `Io.async`: `async` is allowed to +/// run the call inline when the pool's async limit is reached (on a 1-CPU host +/// that limit is 0), which would silently disable the deadline. +/// +/// `Io.Threaded.cancel` joins the task before returning, so `call` is dead by +/// the time this function returns and may live on the caller's stack. +fn fetchWithDeadline(call: *FetchCall, timeout_ms: u64) FetchTimeoutError!u16 { + const io = call.io; + + if (timeout_ms == 0) { + // 0 disables the deadline; run the request inline as before. + fetchTask(call); + } else if (io.concurrent(fetchTask, .{call})) |future_val| { + var future = future_val; + const deadline_ns: u64 = timeout_ms * std.time.ns_per_ms; + const step_ns: u64 = @min(deadline_ns, 25 * std.time.ns_per_ms); + var waited_ns: u64 = 0; + while (!call.finished.load(.acquire) and waited_ns < deadline_ns) { + std.Io.sleep(io, .fromNanoseconds(@intCast(step_ns)), .awake) catch break; + waited_ns += step_ns; + } + _ = future.cancel(io); + } else |_| { + // No thread available to enforce the deadline; fall back to an + // unbounded inline request rather than failing the dispatch. + fetchTask(call); + } + + if (!call.finished.load(.acquire)) return error.Timeout; + if (call.failed) return error.RequestFailed; + return call.status; +} + // ── Types ───────────────────────────────────────────────────────────── pub const WorkerInfo = struct { @@ -192,9 +317,6 @@ pub fn dispatchStepWithOpts( } defer if (auth_header) |ah| allocator.free(ah); - var client: std.http.Client = .{ .allocator = allocator, .io = std_compat.io() }; - defer client.deinit(); - var response_body: std.Io.Writer.Allocating = .init(allocator); defer response_body.deinit(); @@ -204,24 +326,27 @@ pub fn dispatchStepWithOpts( break :blk headers_buf[0..1]; } else &.{}; - const result = client.fetch(.{ - .location = .{ .url = url }, + var call: FetchCall = .{ + .io = outboundIo(), + .allocator = allocator, + .url = url, .method = .POST, .payload = body, - .response_writer = &response_body.writer, .extra_headers = extra_headers, - .headers = .{ - .content_type = .{ .override = "application/json" }, - }, - }) catch { + .content_type_json = true, + .response_writer = &response_body.writer, + }; + + const status_code = fetchWithDeadline(&call, dispatch_timeout_ms) catch |err| { return DispatchResult{ .output = "", .success = false, - .error_text = "HTTP request failed", + .error_text = switch (err) { + error.Timeout => "worker did not respond within the dispatch timeout", + error.RequestFailed => "HTTP request failed", + }, }; }; - - const status_code = @intFromEnum(result.status); if (status_code < 200 or status_code >= 300) { const err_msg = try std.fmt.allocPrint(allocator, "HTTP {d}", .{status_code}); return DispatchResult{ @@ -254,19 +379,24 @@ pub fn probeWorker( const url = worker_protocol.buildRequestUrl(allocator, worker_url, protocol) catch return false; defer allocator.free(url); - var client: std.http.Client = .{ .allocator = allocator, .io = std_compat.io() }; - defer client.deinit(); - var response_body: std.Io.Writer.Allocating = .init(allocator); defer response_body.deinit(); - const result = client.fetch(.{ - .location = .{ .url = url }, + var call: FetchCall = .{ + .io = outboundIo(), + .allocator = allocator, + .url = url, .method = .GET, + .payload = null, + .extra_headers = &.{}, + .content_type_json = false, .response_writer = &response_body.writer, - }) catch return false; + }; - const status_code = @intFromEnum(result.status); + // A worker that does not answer the probe within the deadline is unhealthy, + // which is the whole point: without this the probe blocks and the failure + // is never counted. + const status_code = fetchWithDeadline(&call, probe_timeout_ms) catch return false; return status_code < 500; } @@ -1131,3 +1261,23 @@ test "parseA2aResponse: handles invalid JSON" { try std.testing.expect(!result.success); try std.testing.expectEqualStrings("A2A: invalid JSON response", result.error_text.?); } + +test "probeWorker gives up on a worker that accepts but never responds" { + if (@import("builtin").os.tag == .windows) return error.SkipZigTest; + + // A listening socket that is never accepted from: the kernel completes the + // TCP handshake, then the request goes unanswered. + const addr: std.Io.net.IpAddress = .{ .ip4 = std.Io.net.Ip4Address.loopback(0) }; + var server = try addr.listen(std_compat.io(), .{ .reuse_address = true }); + defer server.deinit(std_compat.io()); + + const url = try std.fmt.allocPrint(std.testing.allocator, "http://127.0.0.1:{d}/webhook", .{server.socket.address.ip4.port}); + defer std.testing.allocator.free(url); + + const saved_dispatch = dispatch_timeout_ms; + const saved_probe = probe_timeout_ms; + defer configureTimeouts(saved_dispatch, saved_probe); + configureTimeouts(saved_dispatch, 200); + + try std.testing.expect(!probeWorker(std.testing.allocator, url, "webhook")); +} diff --git a/src/main.zig b/src/main.zig index 36b371e..5e5afe3 100644 --- a/src/main.zig +++ b/src/main.zig @@ -10,6 +10,7 @@ const ids = @import("ids.zig"); const metrics_mod = @import("metrics.zig"); const worker_protocol = @import("worker_protocol.zig"); const async_dispatch = @import("async_dispatch.zig"); +const dispatch_mod = @import("dispatch.zig"); const redis_client = @import("redis_client.zig"); const mqtt_client = @import("mqtt_client.zig"); const tracker_mod = @import("tracker.zig"); @@ -262,6 +263,11 @@ pub fn main(init: std.process.Init) !void { } } + dispatch_mod.configureTimeouts( + @as(u64, cfg.engine.default_timeout_ms), + @as(u64, cfg.engine.worker_probe_timeout_ms), + ); + var engine = engine_mod.Engine.init(&store, allocator, poll_ms); engine.configure(.{ .health_check_interval_ms = @as(i64, @intCast(cfg.engine.health_check_interval_ms)),