diff --git a/docs/USING.md b/docs/USING.md index 4ce07eb..39915a8 100644 --- a/docs/USING.md +++ b/docs/USING.md @@ -304,6 +304,34 @@ Application work and scheduler load can add delay. The timer does not extend the original request deadline. A zero delay still yields through the scheduler. +### Per-request deadlines + +`timeout_ms` is the default deadline for every request cycle. +A long event stream needs a longer deadline than an ordinary request. +Set `max_timeout_ms` to allow a callback to select a longer deadline. +Then call `try context.setRequestTimeout(ms)` during any callback of that request. + +The new deadline is `ms` after the request's first byte arrived. +The scheduler adopts the value when the callback returns or flushes. +A value above `max_timeout_ms`, or zero, returns `InvalidRequestTimeout`. +Without `max_timeout_ms`, the method returns `RequestTimeoutUnavailable`. +Earlier unsent pipelined responses keep their deadline until their batch completes. +The next request on the connection uses `timeout_ms` again. + +### Idle connections + +A connection is idle while no request byte is buffered or active. +This state lasts from accept to the first byte, and between keep-alive requests. +`idle_timeout_ms` bounds that wait; zero selects `timeout_ms`. +The first received byte ends the idle state and starts the request deadline. +Idle time therefore never shortens the next request's deadline. +Proxies such as `tailscale serve` reuse idle backend connections, so this matters for long streams. +A slow client can use at most the idle bound plus the request bound per request. + +A long deadline holds one admitted connection for longer. +It reserves no additional memory. +Prefer `wait` or `waitNotification` for long streams, because they release the worker between events. + Call `wait` before beginning a response, or after a completed flush with an empty output snapshot. The method rejects pending output, reservations, borrowed payloads, and cancellation. The scheduler drains earlier finished responses before waiting on a subsequent pipeline request. @@ -409,7 +437,9 @@ The [architecture guide](ARCHITECTURE.md#resource-boundaries) explains which res | `response_batch_limit` | 128 | Bound retained response snapshots; worker execution uses one. | | `max_response_bytes` | 16 MiB | Bound logical response bytes across flushes. | | `callbacks_per_turn` | 0, automatic | Bound inline callbacks per owner turn. | -| `timeout_ms` | 5000 | Bound a request cycle while the owner can execute. | +| `timeout_ms` | 5000 | Bound a request from its first received byte while the owner can execute. | +| `idle_timeout_ms` | 0, same as `timeout_ms` | Bound the wait for a request's first byte, after accept and between keep-alive requests. | +| `max_timeout_ms` | 0, disabled | Bound the deadline a callback can select with `setRequestTimeout`; at least `timeout_ms`, at most one day. | | `shutdown_ms` | 5000 | Bound shutdown coordination and drain waits. | | `memory_budget_bytes` | 512 MiB | Bound requested framework heap and startup stack reservations. | | `worker_stack_bytes` | 1 MiB | Reserve requested stack space for each worker and secondary owner. | diff --git a/src/api.zig b/src/api.zig index f63d88f..1b06e7d 100644 --- a/src/api.zig +++ b/src/api.zig @@ -79,6 +79,9 @@ pub const Context = struct { resume_after_ns: ?u64 = null, await_notification: bool = false, notification_cell: ?*Notification.Cell = null, + /// Scheduler-owned target for setRequestTimeout; null when Config.max_timeout_ms is zero. + requested_timeout_ms: ?*u32 = null, + max_timeout_ms: u32 = 0, /// Request one cancellation callback if this callback returns flush or wait. /// Cancellation runs on the configured application executor after kernel borrows end. @@ -101,6 +104,18 @@ pub const Context = struct { return .wait; } + /// Replace this request's deadline with `timeout_ms` after the request started. + /// Config.max_timeout_ms bounds the value and must be nonzero to allow overrides. + /// The server adopts the value when this callback returns or flushes. + /// A later call in the same request replaces an earlier one. The next request + /// on the connection uses Config.timeout_ms again. + pub fn setRequestTimeout(self: *Context, timeout_ms: u32) error{ RequestTimeoutUnavailable, InvalidRequestTimeout, Cancelled }!void { + if (self.cancelled.load(.acquire)) return error.Cancelled; + const target = self.requested_timeout_ms orelse return error.RequestTimeoutUnavailable; + if (timeout_ms == 0 or timeout_ms > self.max_timeout_ms) return error.InvalidRequestTimeout; + target.* = timeout_ms; + } + /// Obtain a handle without retaining the callback context or request storage. pub fn notification(self: *Context) error{ NotificationUnavailable, Cancelled }!Notification { if (self.cancelled.load(.acquire)) return error.Cancelled; @@ -613,6 +628,37 @@ test "blocking flush rejects unavailable, cancelled and unpublished states witho try std.testing.expectError(error.InvalidState, context.flushAndWait()); } +test "request timeout overrides require a target, stay bounded and stop after cancellation" { + var arena: [1024]u8 = undefined; + const cache = testCache(); + var writer = Writer.init(&arena, &cache, 0); + var request: http.Request = undefined; + var state: [8]usize = @splat(0); + var cancelled: std.atomic.Value(bool) = .init(false); + var context: Context = .{ + .request = &request, + .writer = &writer, + .event = .request, + .state = &state, + .application = null, + .cancelled = &cancelled, + }; + try std.testing.expectError(error.RequestTimeoutUnavailable, context.setRequestTimeout(1000)); + var target: u32 = 0; + context.requested_timeout_ms = ⌖ + context.max_timeout_ms = 60_000; + try std.testing.expectError(error.InvalidRequestTimeout, context.setRequestTimeout(0)); + try std.testing.expectError(error.InvalidRequestTimeout, context.setRequestTimeout(60_001)); + try std.testing.expectEqual(@as(u32, 0), target); + try context.setRequestTimeout(60_000); + try std.testing.expectEqual(@as(u32, 60_000), target); + try context.setRequestTimeout(1); + try std.testing.expectEqual(@as(u32, 1), target); + cancelled.store(true, .release); + try std.testing.expectError(error.Cancelled, context.setRequestTimeout(2)); + try std.testing.expectEqual(@as(u32, 1), target); +} + test "begin writes the head into the arena and flush keeps the snapshot" { var arena: [1024]u8 = undefined; const cache = testCache(); diff --git a/src/main.zig b/src/main.zig index 5a3b953..11b8fb5 100644 --- a/src/main.zig +++ b/src/main.zig @@ -41,6 +41,7 @@ const Demo = struct { html: []const u8, stall_ms: u32, execution: framework.Execution, + max_timeout_ms: u32 = 0, continuation_started: std.atomic.Value(u64) = .init(0), continuation_finished: std.atomic.Value(u64) = .init(0), continuation_cancelled: std.atomic.Value(u64) = .init(0), @@ -59,7 +60,7 @@ pub fn main(init: std.process.Init) !void { if (std.mem.eql(u8, flag, "--help")) { std.debug.print("bounded-http: bounded experimental Linux io_uring / macOS kqueue / Windows IOCP HTTP/1.1\n" ++ "--bind-address A.B.C.D --port N --connections N --execution workers|inline --workers N --max-body N --max-header N\n" ++ - "--timeout-ms N --duration-ms N --send-chunk N --gather-send 0|1 --stall-ms N\n" ++ + "--timeout-ms N --max-timeout-ms N --idle-timeout-ms N --duration-ms N --send-chunk N --gather-send 0|1 --stall-ms N\n" ++ "--response-batch-limit N --socket-send-buffer N --output-bytes N --max-response N --memory-budget N --index FILE\n" ++ "--borrow-copy-threshold N --callbacks-per-turn N --callback-timing 0|1 --deadline-sweep-ms N --shards N\n", .{}); return; @@ -82,6 +83,10 @@ pub fn main(init: std.process.Init) !void { config.max_header = try std.fmt.parseInt(u32, value, 10); } else if (std.mem.eql(u8, flag, "--timeout-ms")) { config.timeout_ms = try std.fmt.parseInt(u32, value, 10); + } else if (std.mem.eql(u8, flag, "--max-timeout-ms")) { + config.max_timeout_ms = try std.fmt.parseInt(u32, value, 10); + } else if (std.mem.eql(u8, flag, "--idle-timeout-ms")) { + config.idle_timeout_ms = try std.fmt.parseInt(u32, value, 10); } else if (std.mem.eql(u8, flag, "--duration-ms")) { config.duration_ms = try std.fmt.parseInt(u32, value, 10); } else if (std.mem.eql(u8, flag, "--send-chunk")) { @@ -125,7 +130,7 @@ pub fn main(init: std.process.Init) !void { const shards = framework.Cluster.resolveShards(config); const html = try std.Io.Dir.cwd().readFileAlloc(init.io, index_path, init.gpa, .limited(65536)); defer init.gpa.free(html); - var demo: Demo = .{ .html = html, .stall_ms = stall_ms, .execution = config.execution }; + var demo: Demo = .{ .html = html, .stall_ms = stall_ms, .execution = config.execution, .max_timeout_ms = config.max_timeout_ms }; var budget: framework.Budget = .{ .upstream = init.gpa, .limit_bytes = config.memory_budget_bytes - try framework.Cluster.stackBytes(config) }; defer std.debug.assert(budget.live_bytes == 0); const cluster = try framework.Cluster.init(budget.allocator(), config, handler, &demo); @@ -221,6 +226,7 @@ fn handle(context: *api.Context) !api.Action { if (std.mem.startsWith(u8, path, "/notify") or std.mem.startsWith(u8, path, "/signal/")) return notificationContinuation(context, demo, path); if (std.mem.eql(u8, path, "/timed-chunks") or std.mem.eql(u8, path, "/wait-only") or + std.mem.eql(u8, path, "/extended-wait-only") or std.mem.eql(u8, path, "/empty-timer") or std.mem.eql(u8, path, "/abort-after-flush")) return timedContinuation(context, demo, path); if (std.mem.eql(u8, context.request.method, "CONNECT")) { @@ -391,6 +397,11 @@ fn timedContinuation(context: *api.Context, demo: *Demo, path: []const u8) !api. _ = demo.continuation_live.fetchAdd(1, .monotonic); _ = demo.continuation_started.fetchAdd(1, .release); context.requestCancellation(); + if (std.mem.eql(u8, path, "/extended-wait-only")) { + // Select the configured maximum deadline for this request only. + try context.setRequestTimeout(demo.max_timeout_ms); + return context.wait(@as(u64, demo.stall_ms) * std.time.ns_per_ms); + } if (std.mem.eql(u8, path, "/wait-only")) return context.wait(@as(u64, demo.stall_ms) * std.time.ns_per_ms); try writer.begin(200, "text/plain", null); if (!std.mem.eql(u8, path, "/empty-timer")) { @@ -405,7 +416,7 @@ fn timedContinuation(context: *api.Context, demo: *Demo, path: []const u8) !api. return context.wait(if (std.mem.eql(u8, path, "/empty-timer")) 0 else @as(u64, demo.stall_ms) * std.time.ns_per_ms); } std.debug.assert(context.event == .timer); - if (std.mem.eql(u8, path, "/wait-only")) { + if (std.mem.eql(u8, path, "/wait-only") or std.mem.eql(u8, path, "/extended-wait-only")) { try writer.begin(200, "text/plain", 4); try writer.borrow("done"); } else if (std.mem.eql(u8, path, "/timed-chunks")) { diff --git a/src/server.zig b/src/server.zig index 4623ba6..ecd8841 100644 --- a/src/server.zig +++ b/src/server.zig @@ -70,6 +70,14 @@ pub const Config = struct { shard_affinity: bool = false, max_response_bytes: usize = 16 * 1024 * 1024, timeout_ms: u32 = 5000, + /// Largest request deadline a callback may select with + /// Context.setRequestTimeout, such as for a long event stream. + /// Zero disables the override. Otherwise at least timeout_ms, at most one day. + max_timeout_ms: u32 = 0, + /// Bound the wait for the first byte of a request: after accept, and + /// between keep-alive requests. Zero selects timeout_ms, the previous + /// single-deadline behavior. The request deadline starts at that first byte. + idle_timeout_ms: u32 = 0, shutdown_ms: u32 = 5000, duration_ms: u32 = 0, send_chunk: u32 = 65536, @@ -80,6 +88,19 @@ pub const Config = struct { reuse_port: bool = false, pub const max_callbacks_per_turn_auto: u32 = 8192; + /// Upper bound for max_timeout_ms: one day. + pub const max_timeout_limit_ms: u32 = 24 * 60 * 60 * 1000; + + /// The request deadline that applies without a callback override. + pub fn defaultTimeoutNs(self: Config) u64 { + return @as(u64, self.timeout_ms) * std.time.ns_per_ms; + } + + /// The wait allowed for the first byte of the next request. + pub fn idleTimeoutNs(self: Config) u64 { + const ms = if (self.idle_timeout_ms == 0) self.timeout_ms else self.idle_timeout_ms; + return @as(u64, ms) * std.time.ns_per_ms; + } /// Parse four decimal IPv4 octets at startup. No allocation or name lookup. pub fn parseBindAddress(text: []const u8) error{InvalidBindAddress}![4]u8 { @@ -148,6 +169,8 @@ pub const Config = struct { self.response_batch_limit == 0 or self.response_batch_limit > 511 or self.borrow_copy_threshold > 4096 or self.callbacks_per_turn > 1 << 20 or self.deadline_sweep_ms == 0 or self.deadline_sweep_ms > 1000 or self.shards > 64 or + (self.max_timeout_ms != 0 and (self.max_timeout_ms < self.timeout_ms or self.max_timeout_ms > max_timeout_limit_ms)) or + self.idle_timeout_ms > max_timeout_limit_ms or self.submit_batch > 4096) return error.InvalidConfiguration; assert(self.maxSendParts() <= transport.max_vectors); @@ -302,6 +325,15 @@ const Slot = struct { batch_borrows_input: bool = false, closing: bool = false, deadline: u64 = 0, + /// Deadline length of the current request: the configured default, or + /// a callback's bounded Context.setRequestTimeout value. + timeout_ns: u64 = 0, + /// Written by the running callback, read by the owner after the phase + /// release that publishes the callback's result or stream snapshot. Zero is none. + requested_timeout_ms: u32 = 0, + /// No request byte is buffered or active: the idle deadline applies, and + /// the next received byte starts the request deadline. + idle: bool = false, request_started: u64 = 0, queued_at: u64 = 0, queue_ns: u64 = 0, @@ -524,6 +556,7 @@ pub const Server = struct { .max_target_bytes = 8192, }), .writer = api.Writer.init(arena, &self.header_cache, config.borrow_copy_threshold), + .timeout_ns = config.defaultTimeoutNs(), }; @memset(slot.cells, .{}); // Higher slots go in first so accept pops the lowest free index. @@ -801,6 +834,7 @@ pub const Server = struct { // The worker still owns its stack, request and application state. // Only the frozen writer snapshot transfers to the I/O owner. slot.phase.store(.stream_wait, .release); + self.applyRequestedTimeout(slot); if (slot.closing or self.stop_requested.load(.acquire) or self.now >= slot.deadline) { if (!slot.closing and !self.stop_requested.load(.acquire)) self.stats.timeouts += 1; try self.beginClose(slot); @@ -815,6 +849,7 @@ pub const Server = struct { self.stats.max_queue_ns = @max(self.stats.max_queue_ns, slot.queue_ns); } slot.phase.store(.io, .release); + self.applyRequestedTimeout(slot); if (slot.closing or slot.action == .close) { try self.beginClose(slot); } else if (self.stop_requested.load(.acquire) or self.now >= slot.deadline) { @@ -912,7 +947,7 @@ pub const Server = struct { // Refresh once per adoption, including during a full queue drain. // Request callbacks continue to use the existing turn-clock policy. self.sampleClock(); - const deadline = value.accepted_at + @as(u64, self.config.timeout_ms) * 1_000_000; + const deadline = value.accepted_at + self.config.idleTimeoutNs(); self.stats.max_handoff_delay_ns = @max(self.stats.max_handoff_delay_ns, self.now - value.accepted_at); if (self.stop_requested.load(.acquire) or self.now >= deadline) { value.socket.close(); @@ -994,7 +1029,10 @@ pub const Server = struct { slot.parser.reset(); slot.interim_sent = false; slot.logical_written = 0; - slot.deadline = accepted_at + @as(u64, self.config.timeout_ms) * 1_000_000; + slot.timeout_ns = self.config.defaultTimeoutNs(); + slot.requested_timeout_ms = 0; + slot.idle = true; + slot.deadline = accepted_at + self.config.idleTimeoutNs(); slot.request_started = accepted_at; self.stats.accepted += 1; self.stats.live_connections += 1; @@ -1089,6 +1127,13 @@ pub const Server = struct { assert(transferred <= slot.input.len - slot.received); slot.received += transferred; self.stats.bytes_received += transferred; + if (slot.idle) { + // The first byte of a request ends the idle wait and + // starts that request's own deadline. + slot.idle = false; + slot.request_started = self.now; + slot.deadline = self.now + slot.timeout_ns; + } // Bytes that arrive while a batch is still being sent, or while // a flushed request waits to resume, are parsed after that batch. if (!slot.send_pending and slot.batch_count == 0 and !slot.request_active) @@ -1222,6 +1267,20 @@ pub const Server = struct { try self.receive(index); } + /// Adopt a callback's Context.setRequestTimeout value after its result or + /// stream snapshot is published. The new deadline counts from the request + /// start. Earlier unsent pipelined responses keep their deadline; the new + /// one applies when their batch completes. + fn applyRequestedTimeout(self: *Server, slot: *Slot) void { + const ms = slot.requested_timeout_ms; + if (ms == 0) return; + slot.requested_timeout_ms = 0; + assert(self.config.max_timeout_ms != 0 and ms <= self.config.max_timeout_ms); + if (!slot.request_active) return; + slot.timeout_ns = @as(u64, ms) * std.time.ns_per_ms; + if (slot.batch_count == 0) slot.deadline = slot.request_started + slot.timeout_ns; + } + fn dispatch(self: *Server, index: usize) void { const slot = &self.slots[index]; assert(slot.phase.load(.acquire) == .io and !slot.closing and !slot.send_pending); @@ -1285,6 +1344,8 @@ pub const Server = struct { .cancelled = &slot.cancelled, .notify_cancel = slot.notify_cancel, .notification_cell = &slot.notification, + .requested_timeout_ms = if (self.config.max_timeout_ms != 0) &slot.requested_timeout_ms else null, + .max_timeout_ms = self.config.max_timeout_ms, .blocking_flush = if (self.config.execution == .workers and slot.event != .cancelled) .{ .context = &streaming, .flush = WorkerFlush.flush, @@ -1427,6 +1488,8 @@ pub const Server = struct { } assert(slot.action == .finish); slot.request_active = false; + // The next request and the idle keep-alive cycle use the default again. + slot.timeout_ns = self.config.defaultTimeoutNs(); assert(slot.request.consumed <= slot.received - slot.input_cursor); slot.input_cursor += slot.request.consumed; if (!slot.request.keep_alive) return self.drainBatch(index, .close); @@ -1568,9 +1631,13 @@ pub const Server = struct { slot.arena_used = 0; // A fresh idle cycle starts after the preceding send finishes. Already // buffered partial/current requests retain their earlier deadline. - if (slot.batch_next == .parse and !slot.request_active and slot.input_cursor == slot.received) + if (slot.batch_next == .parse and !slot.request_active and slot.input_cursor == slot.received) { + slot.idle = true; slot.request_started = completed_at; - slot.deadline = slot.request_started + @as(u64, self.config.timeout_ms) * 1_000_000; + slot.deadline = completed_at + self.config.idleTimeoutNs(); + } else { + slot.deadline = slot.request_started + slot.timeout_ns; + } switch (slot.batch_next) { .resume_flush => { assert(slot.request_active); @@ -1856,6 +1923,99 @@ pub const Server = struct { try std.testing.expect(!slot.in_use); } + test "a callback's request timeout replaces the deadline and the next request uses the default" { + const server = try Server.init(std.testing.allocator, .{ + .execution = .inline_event_loop, + .workers = 0, + .connections = 1, + .port = 0, + .timeout_ms = 1000, + .max_timeout_ms = 60_000, + }, struct { + fn handler(ctx: *api.Context) api.Action { + if (ctx.event == .cancelled) return .close; + ctx.requestCancellation(); + ctx.setRequestTimeout(60_000) catch return .close; + return ctx.wait(std.time.ns_per_s) catch .close; + } + }.handler, null); + defer server.deinit(); + const slot = &server.slots[0]; + slot.writer.open(0, true, false); + slot.request_active = true; + slot.in_use = true; + slot.request_started = nowNs(); + slot.timeout_ns = server.config.defaultTimeoutNs(); + slot.deadline = slot.request_started + slot.timeout_ns; + server.free_count = 0; + server.stats.live_connections = 1; + server.now = slot.request_started; + slot.phase.store(.running, .release); + server.invokeHandler(slot); + try std.testing.expectEqual(@as(u32, 60_000), slot.requested_timeout_ms); + try server.serviceSlot(0); + try std.testing.expectEqual(@as(u32, 0), slot.requested_timeout_ms); + try std.testing.expectEqual(60 * std.time.ns_per_s, slot.timeout_ns); + try std.testing.expectEqual(slot.request_started + 60 * std.time.ns_per_s, slot.deadline); + try std.testing.expect(slot.resume_at != null and slot.in_use); + // The default deadline has passed, but the extended one still holds the wait. + const deadline = slot.deadline; + slot.resume_at = deadline + 5; + server.now = slot.request_started + 2 * std.time.ns_per_s; + server.next_timer = server.now; + try server.serviceTimers(); + try std.testing.expectEqual(Phase.io, slot.phase.load(.acquire)); + try std.testing.expect(!slot.closing and slot.resume_at != null); + try std.testing.expectEqual(deadline + 5, server.next_timer.?); + // The extended deadline cancels the wait. + server.now = deadline + 5; + try server.serviceTimers(); + try std.testing.expectEqual(api.Event.cancelled, slot.event); + _ = server.popReady(); + slot.phase.store(.running, .release); + server.invokeHandler(slot); + slot.phase.store(.io, .release); + server.maybeFree(slot); + try std.testing.expect(!slot.in_use); + } + + test "an unsent pipelined response keeps its deadline when a later callback extends it" { + const server = try Server.init(std.testing.allocator, .{ + .execution = .inline_event_loop, + .workers = 0, + .connections = 1, + .port = 0, + .timeout_ms = 1000, + .max_timeout_ms = 60_000, + }, struct { + fn handler(_: *api.Context) api.Action { + unreachable; + } + }.handler, null); + defer server.deinit(); + const slot = &server.slots[0]; + slot.request_active = true; + slot.request_started = 1000; + slot.timeout_ns = server.config.defaultTimeoutNs(); + slot.deadline = 1000 + slot.timeout_ns; + slot.batch_count = 1; + slot.requested_timeout_ms = 60_000; + server.applyRequestedTimeout(slot); + try std.testing.expectEqual(1000 + server.config.defaultTimeoutNs(), slot.deadline); + try std.testing.expectEqual(60 * std.time.ns_per_s, slot.timeout_ns); + // With no earlier output in the arena, the new deadline applies at once. + slot.batch_count = 0; + slot.requested_timeout_ms = 30_000; + server.applyRequestedTimeout(slot); + try std.testing.expectEqual(1000 + 30 * std.time.ns_per_s, slot.deadline); + // A finished request cannot extend the next idle cycle. + slot.request_active = false; + slot.requested_timeout_ms = 60_000; + server.applyRequestedTimeout(slot); + try std.testing.expectEqual(1000 + 30 * std.time.ns_per_s, slot.deadline); + slot.batch_count = 0; + } + test "a due timer waits for its preceding frozen batch and rearms after release" { const server = try Server.init(std.testing.allocator, .{ .execution = .inline_event_loop, @@ -2745,6 +2905,21 @@ test "configuration rejects combined resource overcommit and impossible worker l try std.testing.expectEqual(Config.max_callbacks_per_turn_auto, (Config{ .connections = 4096 }).effectiveCallbacksPerTurn()); } +test "request timeout bound is disabled by default or covers the default deadline within one day" { + try (Config{ .max_timeout_ms = 0 }).validate(); + try (Config{ .timeout_ms = 5000, .max_timeout_ms = 5000 }).validate(); + try (Config{ .max_timeout_ms = Config.max_timeout_limit_ms }).validate(); + try std.testing.expectError(error.InvalidConfiguration, (Config{ .timeout_ms = 5000, .max_timeout_ms = 4999 }).validate()); + try std.testing.expectError(error.InvalidConfiguration, (Config{ .max_timeout_ms = Config.max_timeout_limit_ms + 1 }).validate()); +} + +test "idle timeout defaults to the request timeout and is bounded" { + try std.testing.expectEqual(@as(u64, 5000) * std.time.ns_per_ms, (Config{}).idleTimeoutNs()); + try std.testing.expectEqual(@as(u64, 200) * std.time.ns_per_ms, (Config{ .idle_timeout_ms = 200 }).idleTimeoutNs()); + try (Config{ .idle_timeout_ms = Config.max_timeout_limit_ms }).validate(); + try std.testing.expectError(error.InvalidConfiguration, (Config{ .idle_timeout_ms = Config.max_timeout_limit_ms + 1 }).validate()); +} + test "cluster signal stop touches every atomic flag without accessing transports" { // Only stop flags exist in this witness. Backend and lifecycle fields remain undefined. var first: Server = undefined; diff --git a/tests/continuation_integration.py b/tests/continuation_integration.py index 7d83a26..77e934d 100644 --- a/tests/continuation_integration.py +++ b/tests/continuation_integration.py @@ -136,6 +136,90 @@ def timeout_and_shutdown(mode): require(current['cancelled'] == 13 and current['finished'] == 0, 'shutdown failed to drain cancellation callbacks') +def extended_request_deadline(mode): + """A callback-selected deadline outlives the default, stays bounded and resets per request.""" + options = shared_options(mode) + options['timeout_ms'] = 150 + options['max_timeout_ms'] = 2000 + options['deadline_sweep_ms'] = 5 + with Server(**options, stall_ms=600) as server: + with server.connect(timeout=5) as client: + reader = ResponseReader(client) + started = time.monotonic() + client.sendall(request('/extended-wait-only')) + status, _, body = reader.response() + elapsed = time.monotonic() - started + require(status == 200 and body == b'done', 'extended deadline did not keep the wait alive') + require(elapsed >= .55, f'extended wait finished too early: {elapsed:.3f}s') + # The next request on the same connection uses the default deadline again. + started = time.monotonic() + client.sendall(request('/wait-only')) + require(client.recv(1) == b'', 'default deadline did not return for the next request') + require(time.monotonic() - started < .5, 'next request kept the extended deadline') + with Server(**dict(options, max_timeout_ms=300), stall_ms=1500) as server: + with server.connect(timeout=5) as client: + started = time.monotonic() + client.sendall(request('/extended-wait-only')) + require(client.recv(1) == b'', 'extended deadline did not cancel the wait') + elapsed = time.monotonic() - started + require(.25 <= elapsed < 1.2, f'bounded extended deadline cancelled after {elapsed:.3f}s') + current = await_count(server, 'cancelled', 1) + require(current['live'] == 0, 'extended timeout cleanup did not run') + final_counts(server) + + +def idle_keep_alive_then_request(mode): + """Idle keep-alive time does not count toward the next request's deadline.""" + options = shared_options(mode) + options['timeout_ms'] = 500 + options['max_timeout_ms'] = 1000 + options['deadline_sweep_ms'] = 5 + with Server(**options, stall_ms=350) as server: + with server.connect(timeout=5) as client: + reader = ResponseReader(client) + client.sendall(request('/plaintext')) + require(reader.response()[0] == 200, 'first keep-alive request failed') + time.sleep(.3) + # Before the fix, the deadline counted from the idle start: 0.3 s + 0.35 s > 0.5 s. + client.sendall(request('/wait-only')) + status, _, body = reader.response() + require(status == 200 and body == b'done', 'idle time shortened the next request deadline') + # An extended deadline also counts from the request's first byte: 0.4 s + 0.7 s > 1 s. + with Server(**options, stall_ms=700) as server: + with server.connect(timeout=5) as client: + reader = ResponseReader(client) + client.sendall(request('/plaintext')) + require(reader.response()[0] == 200, 'first keep-alive request failed') + time.sleep(.4) + client.sendall(request('/extended-wait-only')) + status, _, body = reader.response() + require(status == 200 and body == b'done', 'idle time shortened the extended request deadline') + final_counts(server) + + +def idle_timeout_is_separate(mode): + """idle_timeout_ms closes a quiet connection but never cuts an active request.""" + options = shared_options(mode) + options['timeout_ms'] = 1000 + options['idle_timeout_ms'] = 200 + options['deadline_sweep_ms'] = 5 + with Server(**options, stall_ms=500) as server: + with server.connect(timeout=5) as client: + started = time.monotonic() + require(client.recv(1) == b'', 'idle connection outlived idle_timeout_ms') + elapsed = time.monotonic() - started + require(.15 <= elapsed < .8, f'idle close after {elapsed:.3f}s') + with server.connect(timeout=5) as client: + reader = ResponseReader(client) + client.sendall(request('/wait-only')) + status, _, body = reader.response() + require(status == 200 and body == b'done', 'idle timeout cut an active request') + started = time.monotonic() + require(client.recv(1) == b'', 'kept-alive connection outlived idle_timeout_ms') + require(time.monotonic() - started < .8, 'idle timeout did not apply after a response') + final_counts(server) + + def disconnect_after_flush(mode): options = shared_options(mode) options['connections'] = 1 @@ -280,7 +364,9 @@ def main(): SERVER_BINARY = args.server.resolve() results = [] for mode in ('inline', 'workers'): - for test in (framing_and_reuse, many_waits, timeout_and_shutdown, disconnect_after_flush, + for test in (framing_and_reuse, many_waits, timeout_and_shutdown, extended_request_deadline, + idle_keep_alive_then_request, idle_timeout_is_separate, + disconnect_after_flush, notification_ordering_and_reuse, notification_many_waits, notification_pipeline_and_flush, notification_cancellation): started = time.monotonic() diff --git a/tools/verify_windows.py b/tools/verify_windows.py index 6ff7d05..6d14f7e 100644 --- a/tools/verify_windows.py +++ b/tools/verify_windows.py @@ -126,7 +126,7 @@ def run(name, arguments, timeout): [sys.executable, 'tests/continuation_integration.py', '--server', str(binary), '--json', str(packet / 'continuation-integration.json')], 150) continuation = json.loads((packet / 'continuation-integration.json').read_text()) - require(continuation.get('ok') is True and continuation.get('groups') == 18, + require(continuation.get('ok') is True and continuation.get('groups') == 24, 'Continuation timer/cancellation receipt mismatch') run('windows-shards-integration', [sys.executable, 'tests/windows_shards_integration.py', '--server', str(binary),