diff --git a/.github/workflows/native.yml b/.github/workflows/native.yml new file mode 100644 index 0000000..df8c11a --- /dev/null +++ b/.github/workflows/native.yml @@ -0,0 +1,77 @@ +name: Native HTTP verification + +on: + push: + pull_request: + workflow_dispatch: + +permissions: + contents: read + +concurrency: + group: bounded-http-native-${{ github.ref }} + cancel-in-progress: true + +jobs: + native: + name: Native ${{ matrix.os }} + strategy: + fail-fast: false + matrix: + os: [ubuntu-latest, macos-latest] + runs-on: ${{ matrix.os }} + timeout-minutes: 25 + env: + PYTHONDONTWRITEBYTECODE: "1" + steps: + - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1 + + - name: Install exact checksum-verified Zig + shell: bash + run: | + python3 - <<'PYTHON' + import hashlib, json, os, pathlib, platform, tarfile, urllib.request + version = pathlib.Path('.zig-version').read_text().strip() + assert version == '0.16.0', 'Expected exact Zig 0.16.0' + architecture = {'x86_64': 'x86_64', 'arm64': 'aarch64', 'aarch64': 'aarch64'}[platform.machine()] + system = {'Linux': 'linux', 'Darwin': 'macos'}[platform.system()] + with urllib.request.urlopen('https://ziglang.org/download/index.json', timeout=30) as response: + artifact = json.load(response)[version][f'{architecture}-{system}'] + assert artifact['tarball'].startswith('https://ziglang.org/') + directory = pathlib.Path(os.environ['RUNNER_TEMP']) / 'bounded-http-zig' + directory.mkdir() + archive = directory / 'zig.tar.xz' + with urllib.request.urlopen(artifact['tarball'], timeout=180) as response: + archive.write_bytes(response.read()) + assert hashlib.sha256(archive.read_bytes()).hexdigest() == artifact['shasum'] + with tarfile.open(archive) as package: + package.extractall(directory, filter='data') + compiler, = directory.glob('*/zig') + with open(os.environ['GITHUB_PATH'], 'a') as path: + path.write(str(compiler.parent) + '\n') + PYTHON + + - name: Verify and exercise HTTP lifecycle + shell: bash + run: | + set -euo pipefail + packet="$RUNNER_TEMP/bounded-http-native" + mkdir -p "$packet" + { git rev-parse HEAD; uname -a; zig version; python3 --version; } > "$packet/environment.log" + zig build verify -Doptimize=Debug -j2 --summary all 2>&1 | tee "$packet/debug.log" + zig build verify -Doptimize=ReleaseSafe -j2 --summary all 2>&1 | tee "$packet/release-safe.log" + zig build -Doptimize=ReleaseSafe -j2 2>&1 | tee "$packet/install.log" + python3 tests/test_compare.py -v 2>&1 | tee "$packet/comparator.log" + for suite in integration inline_integration gather_integration batch_integration arena_lifecycle_integration; do + python3 "tests/$suite.py" --timeout 120 --json "$packet/$suite.json" 2>&1 | tee "$packet/$suite.log" + done + python3 tests/continuation_integration.py --json "$packet/continuation.json" 2>&1 | tee "$packet/continuation.log" + + - name: Upload verification logs + if: always() + uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a # v7.0.1 + with: + name: bounded-http-${{ matrix.os }}-${{ github.run_id }}-${{ github.run_attempt }} + path: ${{ runner.temp }}/bounded-http-native + if-no-files-found: warn + retention-days: 14 diff --git a/README.md b/README.md index bbfc59b..18be668 100644 --- a/README.md +++ b/README.md @@ -146,6 +146,11 @@ means the local socket accepted the bytes; it does not prove peer receipt. `return writer.finish()` sends the remaining bytes and completes HTTP framing, without another application callback. No later writes belong to that response. +`return try context.wait(delay_ns)` releases the callback until a monotonic timer expires. +The handler resumes with `.timer`; the original request deadline still applies. +Applications can [request cancellation callbacks](docs/USING.md#timers-and-cancellation-callbacks) to release retained state after abandoned waits or flushes. +These callbacks run on the configured application executor after kernel borrows return. + `borrow` requires request-owned input or immutable server-lifetime storage. `begin` copies its `content_type` argument during the call. There is no dynamic lease-release callback diff --git a/docs/USING.md b/docs/USING.md index 19eaa1b..5951066 100644 --- a/docs/USING.md +++ b/docs/USING.md @@ -293,6 +293,41 @@ Return that result immediately. The owner completes response framing without another callback for that response. Finished responses can remain buffered within a batch before transmission. +### Timers and cancellation callbacks + +`return try context.wait(delay_ns)` ends the current callback and releases its worker. +The scheduler uses its monotonic clock and fixed slot storage. +The handler receives `.timer` after the requested delay. +The scheduler can deliver a timer late; the existing poll interval is at most 10 milliseconds. +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. + +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. +The application must retain its state between callbacks. +The application must not retain callback Context pointers. + +Call `context.requestCancellation()` before returning a flush or wait action that retains application state. +Cancellation then schedules one `.cancelled` callback after application and kernel borrows return. +The callback runs on the configured worker or inline executor. +The request and eight state words remain available during cleanup. +The writer is frozen, and blocking flush is unavailable. +The callback must release retained state and return `.close`. +The scheduler ignores other actions from this cleanup callback. +Shutdown drains these callbacks before stopping workers or freeing slots. + +A handler must release its own state before returning `.finish` or `.close`. +Those actions never receive an additional cancellation callback, including cancellation races during result publication. +State, locals, and recycled application buffers remain ineligible for output borrowing. +This API does not provide a dynamic output lease or a successful-send finalizer. + +`tests/continuation_integration.py` checks timers, partial output, depth-128 pipelines, cancellation, slot reuse, and shutdown. +The worker fixture admits 32 waiting requests with one worker. +The multi-owner fixture runs on Linux and Windows; macOS supports one owner. +These fixtures establish correctness, not a performance comparison. + ### Flush within a worker callback `Context.supportsBlockingFlush()` reports whether the callback runs on a fixed application worker. diff --git a/src/api.zig b/src/api.zig index 85f467e..53c5fe1 100644 --- a/src/api.zig +++ b/src/api.zig @@ -2,8 +2,8 @@ const std = @import("std"); pub const http = @import("http.zig"); const assert = std.debug.assert; -pub const Action = enum { flush, finish, close }; -pub const Event = enum { request, flushed }; +pub const Action = enum { flush, finish, close, wait }; +pub const Event = enum { request, flushed, timer, cancelled }; pub const Handler = *const fn (*Context) Action; pub const FlushError = error{ BlockingFlushUnavailable, InvalidState, Cancelled }; @@ -17,6 +17,29 @@ pub const Context = struct { cancelled: *const std.atomic.Value(bool), /// The scheduler supplies this hook only for an existing application worker. blocking_flush: ?BlockingFlush = null, + /// Internal callback result metadata. Use requestCancellation() and wait(). + notify_cancel: bool = false, + resume_after_ns: ?u64 = null, + + /// Request one cancellation callback if this callback returns flush or wait. + /// Cancellation runs on the configured application executor after kernel borrows end. + /// Its writer is unavailable. Return close after releasing application state. + /// Finish and close require local cleanup and never receive this callback. + pub fn requestCancellation(self: *Context) void { + self.notify_cancel = true; + } + + /// Return this action to release the worker until a monotonic timer expires. + /// The original request deadline still applies. Wakeups can arrive late. + /// Publish pending response bytes with flush before waiting. + pub fn wait(self: *Context, delay_ns: u64) error{ InvalidState, Cancelled }!Action { + if (self.cancelled.load(.acquire)) return error.Cancelled; + if (self.writer.frozen or self.writer.reserved != 0 or self.writer.borrowed != null or + (self.writer.began and (!self.writer.headers_committed or self.writer.bodyBytes() != 0))) + return error.InvalidState; + self.resume_after_ns = delay_ns; + return .wait; + } pub const BlockingFlush = struct { context: *anyopaque, @@ -714,3 +737,32 @@ test "draft header aliases copy before storage is reused and preserve repeated f try std.testing.expect(std.mem.indexOf(u8, arena[0..writer.body_start], "Content-Type: application/example\r\n") != null); try std.testing.expect(std.mem.endsWith(u8, arena[0..writer.body_start], fields ++ "\r\n")); } + +test "timer waits reject output ownership and cancellation without publication" { + var arena: [1024]u8 = undefined; + const cache = testCache(); + var writer = Writer.init(&arena, &cache, 0); + writer.open(0, true, false); + 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 }; + context.requestCancellation(); + try std.testing.expect(context.notify_cancel); + try std.testing.expectEqual(Action.wait, try context.wait(0)); + try std.testing.expectEqual(@as(?u64, 0), context.resume_after_ns); + try writer.begin(200, "text/plain", null); + try std.testing.expectError(error.InvalidState, context.wait(1)); + _ = writer.flush(); + try std.testing.expectError(error.InvalidState, context.wait(1)); + writer.headers_committed = true; + writer.release(); + writer.resumeSnapshot(0); + try std.testing.expectEqual(Action.wait, try context.wait(std.math.maxInt(u64))); + _ = try writer.reserve(1); + try std.testing.expectError(error.InvalidState, context.wait(1)); + writer.commit(1); + try std.testing.expectError(error.InvalidState, context.wait(1)); + cancelled.store(true, .release); + try std.testing.expectError(error.Cancelled, context.wait(1)); +} diff --git a/src/main.zig b/src/main.zig index 5a840d0..0d63356 100644 --- a/src/main.zig +++ b/src/main.zig @@ -30,7 +30,15 @@ fn requestControlStop() void { if (active_cluster.load(.seq_cst)) |cluster| cluster.requestStopFromSignal(); } -const Demo = struct { html: []const u8, stall_ms: u32, execution: framework.Execution }; +const Demo = struct { + html: []const u8, + stall_ms: u32, + execution: framework.Execution, + continuation_started: std.atomic.Value(u64) = .init(0), + continuation_finished: std.atomic.Value(u64) = .init(0), + continuation_cancelled: std.atomic.Value(u64) = .init(0), + continuation_live: std.atomic.Value(u64) = .init(0), +}; pub fn main(init: std.process.Init) !void { var config: framework.Config = .{}; @@ -164,14 +172,21 @@ pub fn main(init: std.process.Init) !void { const stats = try std.json.Stringify.valueAlloc(init.gpa, merged, .{}); defer init.gpa.free(stats); std.debug.print("STATS {s}\n", .{stats}); + std.debug.print("CONTINUATIONS started={d} finished={d} cancelled={d} live={d}\n", .{ + demo.continuation_started.load(.acquire), demo.continuation_finished.load(.acquire), + demo.continuation_cancelled.load(.acquire), demo.continuation_live.load(.acquire), + }); } fn handler(context: *api.Context) api.Action { - return handle(context) catch .close; + return handle(context) catch { + continuationDone(context, false); + return .close; + }; } fn handle(context: *api.Context) !api.Action { - const demo: *const Demo = @ptrCast(@alignCast(context.application.?)); + const demo: *Demo = @ptrCast(@alignCast(context.application.?)); const writer = context.writer; // Exact-target fast path for the measured route; every other form takes // the general route resolution below. @@ -181,6 +196,21 @@ fn handle(context: *api.Context) !api.Action { return writer.finish(); } const path = routePath(context.request.target); + if (std.mem.eql(u8, path, "/continuation-counts")) { + var buffer: [192]u8 = undefined; + const body = try std.fmt.bufPrint(&buffer, "{{\"started\":{d},\"finished\":{d},\"cancelled\":{d},\"live\":{d}}}", .{ + demo.continuation_started.load(.acquire), demo.continuation_finished.load(.acquire), + demo.continuation_cancelled.load(.acquire), demo.continuation_live.load(.acquire), + }); + try writer.begin(200, "application/json", body.len); + const out = try writer.reserve(body.len); + @memcpy(out, body); + writer.commit(body.len); + return writer.finish(); + } + if (std.mem.eql(u8, path, "/timed-chunks") or std.mem.eql(u8, path, "/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")) { try writer.begin(501, "text/plain", 0); return writer.finish(); @@ -280,6 +310,64 @@ fn handle(context: *api.Context) !api.Action { return writer.finish(); } +/// Lifecycle fixtures use fixed state and atomics; they never sleep on workers. +fn timedContinuation(context: *api.Context, demo: *Demo, path: []const u8) !api.Action { + const writer = context.writer; + if (context.event == .cancelled) { + std.debug.assert(writer.frozen and !context.supportsBlockingFlush()); + std.debug.assert(context.state[7] == 1); + continuationDone(context, true); + return .close; + } + if (context.event == .request) { + context.state[7] = 1; + _ = demo.continuation_live.fetchAdd(1, .monotonic); + _ = demo.continuation_started.fetchAdd(1, .release); + context.requestCancellation(); + 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")) { + const output = try writer.reserve(6); + @memcpy(output, "first "); + writer.commit(6); + } + return writer.flush(); + } + if (context.event == .flushed) { + if (std.mem.eql(u8, path, "/abort-after-flush")) return error.FixtureAfterFlush; + 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")) { + try writer.begin(200, "text/plain", 4); + try writer.borrow("done"); + } else if (std.mem.eql(u8, path, "/timed-chunks")) { + if (context.state[0] == 0) { + context.state[0] = 1; + try writer.borrow("second "); + return writer.flush(); + } + const output = try writer.reserve(5); + @memcpy(output, "third"); + writer.commit(5); + } + continuationDone(context, false); + return writer.finish(); +} + +fn continuationDone(context: *api.Context, cancelled: bool) void { + if (context.state[7] == 0) return; + context.state[7] = 0; + const demo: *Demo = @ptrCast(@alignCast(context.application.?)); + const previous = demo.continuation_live.fetchSub(1, .monotonic); + std.debug.assert(previous > 0); + if (cancelled) { + _ = demo.continuation_cancelled.fetchAdd(1, .release); + } else { + _ = demo.continuation_finished.fetchAdd(1, .release); + } +} + fn routePath(target: []const u8) []const u8 { var path = target; // Origin-form targets start with '/'; only other forms can carry a scheme. diff --git a/src/server.zig b/src/server.zig index ed2ee27..e78809f 100644 --- a/src/server.zig +++ b/src/server.zig @@ -227,7 +227,7 @@ const Kind = enum(u8) { accept = 1, recv, send, cancel_recv, cancel_send, cancel const accept_token: u64 = @intFromEnum(Kind.accept); const cancel_accept_token: u64 = @intFromEnum(Kind.cancel_accept); -const BatchNext = enum { parse, resume_flush, close }; +const BatchNext = enum { parse, resume_flush, resume_timer, close }; /// One finished or flushed response snapshot: a range of the connection's /// arena, with an optional borrowed span logically inserted at `borrow_at`. @@ -267,6 +267,9 @@ const Slot = struct { state: [8]usize = @splat(0), action: api.Action = .close, event: api.Event = .request, + notify_cancel: bool = false, + resume_at: ?u64 = null, + wait_delay_ns: ?u64 = null, /// A receive into the free input tail and a send of the frozen batch may /// be in flight together; each has its own cell and cancel cell. recv_pending: bool = false, @@ -415,6 +418,7 @@ pub const Server = struct { now: u64 = 0, clock_budget: u32 = 0, last_sweep: u64 = 0, + next_timer: ?u64 = null, header_cache: api.HeaderCache = .{}, date: [29]u8 = undefined, date_second: u64 = std.math.maxInt(u64), @@ -653,6 +657,7 @@ pub const Server = struct { } } } + try self.serviceTimers(); switch (self.config.execution) { .inline_event_loop => { // One pass over the slots that were ready when the turn @@ -696,6 +701,29 @@ pub const Server = struct { self.safe_to_destroy = true; } + /// Timer storage belongs to slots. A due timer causes one bounded slot scan. + fn serviceTimers(self: *Server) !void { + const next = self.next_timer orelse return; + if (self.now < next) return; + self.next_timer = null; + for (self.slots, 0..) |*slot, index| { + const at = slot.resume_at orelse continue; + assert(slot.in_use and slot.request_active and slot.phase.load(.acquire) == .io); + if (slot.closing or self.stopping or self.now >= slot.deadline) { + if (!slot.closing and !self.stopping) self.stats.timeouts += 1; + try self.beginClose(slot); + } else if (slot.send_pending) { + // The preceding batch still owns output. Completion rearms the timer. + } else if (at <= self.now) { + slot.resume_at = null; + slot.event = .timer; + self.dispatch(index); + } else { + self.next_timer = @min(self.next_timer orelse at, at); + } + } + } + fn anyResult(self: *Server) bool { for (self.slots) |*slot| { if (!slot.in_use) continue; @@ -753,6 +781,20 @@ pub const Server = struct { } else if (self.stop_requested.load(.acquire) or self.now >= slot.deadline) { if (!self.stop_requested.load(.acquire)) self.stats.timeouts += 1; try self.beginClose(slot); + } else if (slot.action == .wait) { + const delay = slot.wait_delay_ns orelse { + try self.beginClose(slot); + break; + }; + const at = nowNs() +| delay; + slot.resume_at = at; + self.next_timer = @min(self.next_timer orelse at, at); + if (slot.batch_count != 0) { + // Initial waits cannot retain earlier finished responses in the arena. + assert(!slot.writer.began); + slot.writer.frozen = true; + try self.drainBatch(index, .resume_timer); + } } else { try self.prepareResponse(index, iteration + 1 < batch_limit); } @@ -1089,6 +1131,8 @@ pub const Server = struct { assert(slot.arena.len - slot.arena_used >= self.config.callback_output_reserve); slot.writer.open(slot.arena_used, slot.request.keep_alive, slot.request.head_only); slot.state = @splat(0); + slot.notify_cancel = false; + slot.resume_at = null; slot.event = .request; slot.logical_written = 0; self.dispatch(index); @@ -1186,7 +1230,7 @@ pub const Server = struct { const timing = self.config.callback_timing; const started = if (timing) nowNs() else 0; if (timing) slot.queue_ns = started - slot.queued_at; - if (slot.cancelled.load(.acquire)) { + if (slot.cancelled.load(.acquire) and slot.event != .cancelled) { slot.action = .close; } else { var streaming: WorkerFlush = .{ .server = self, .slot = slot }; @@ -1197,13 +1241,23 @@ pub const Server = struct { .state = &slot.state, .application = self.application, .cancelled = &slot.cancelled, - .blocking_flush = if (self.config.execution == .workers) .{ + .notify_cancel = slot.notify_cancel, + .blocking_flush = if (self.config.execution == .workers and slot.event != .cancelled) .{ .context = &streaming, .flush = WorkerFlush.flush, } else null, }; slot.action = self.handler(&context); - if (slot.action != .close) assert(slot.writer.frozen); + if (slot.event == .cancelled) { + // Cleanup cannot publish more bytes or suspend again. + slot.action = .close; + slot.notify_cancel = false; + slot.wait_delay_ns = null; + } else { + slot.notify_cancel = context.notify_cancel and (slot.action == .flush or slot.action == .wait); + slot.wait_delay_ns = if (slot.action == .wait) context.resume_after_ns else null; + if (slot.action != .close and slot.action != .wait) assert(slot.writer.frozen); + } } if (timing) slot.handler_ns = nowNs() - started; assert(slot.phase.load(.acquire) == .running); @@ -1484,6 +1538,12 @@ pub const Server = struct { self.dispatch(index); } }, + .resume_timer => { + assert(slot.request_active and !slot.writer.began); + slot.writer.open(0, slot.request.keep_alive, slot.request.head_only); + const at = slot.resume_at.?; + self.next_timer = @min(self.next_timer orelse at, at); + }, .close => try self.beginClose(slot), .parse => { assert(!slot.request_active and slot.input_cursor <= slot.received); @@ -1508,6 +1568,7 @@ pub const Server = struct { fn beginClose(self: *Server, slot: *Slot) !void { const index = (@intFromPtr(slot) - @intFromPtr(self.slots.ptr)) / @sizeOf(Slot); slot.closing = true; + slot.resume_at = null; slot.cancelled.store(true, .release); if (slot.fd >= 0) self.backend.shutdown(slot.fd); if (slot.recv_pending and !slot.recv_cancel_pending) { @@ -1550,6 +1611,19 @@ pub const Server = struct { return; } if (phase != .io) return; + if (slot.notify_cancel) { + // All transport borrows ended. A final application callback still owns state. + slot.notify_cancel = false; + slot.event = .cancelled; + slot.writer.frozen = true; + if (self.config.callback_timing) slot.queued_at = nowNs(); + slot.phase.store(.ready, .release); + if (self.config.execution == .workers) { + self.stats.worker_dispatches += 1; + self.workers[index % self.workers.len].wake(); + } else self.pushReady(index); + return; + } if (slot.fd >= 0) { self.backend.close(self.recvCell(index), slot.fd); slot.fd = -1; @@ -1570,6 +1644,252 @@ pub const Server = struct { self.free_count += 1; } + test "cancellation callback waits for transport release and retains its application executor" { + for ([_]Execution{ .inline_event_loop, .workers }) |execution| { + var calls: usize = 0; + const server = try Server.init(std.testing.allocator, .{ + .execution = execution, + .workers = if (execution == .workers) 1 else 0, + .connections = 1, + .port = 0, + }, struct { + fn handler(ctx: *api.Context) api.Action { + const count: *usize = @ptrCast(@alignCast(ctx.application.?)); + assert(ctx.event == .cancelled and ctx.cancelled.load(.acquire)); + assert(ctx.writer.frozen and !ctx.supportsBlockingFlush()); + assert(ctx.state[0] == 42); + count.* += 1; + return .wait; // The scheduler ignores invalid cleanup actions. + } + }.handler, &calls); + defer server.deinit(); + const slot = &server.slots[0]; + slot.in_use = true; + slot.closing = true; + slot.cancelled.store(true, .release); + slot.request_active = true; + slot.state[0] = 42; + slot.notify_cancel = true; + server.stats.live_connections = 1; + server.free_count = 0; + slot.send_pending = true; + server.maybeFree(slot); + try std.testing.expect(slot.notify_cancel and slot.phase.load(.acquire) == .io); + slot.send_pending = false; + slot.send_cancel_pending = true; + server.maybeFree(slot); + try std.testing.expect(slot.notify_cancel); + slot.send_cancel_pending = false; + server.maybeFree(slot); + try std.testing.expect(slot.in_use and slot.phase.load(.acquire) == .ready); + try std.testing.expect(!slot.notify_cancel); + server.maybeFree(slot); + try std.testing.expectEqual(@as(usize, 0), calls); + if (execution == .inline_event_loop) _ = server.popReady(); + slot.phase.store(.running, .release); + server.invokeHandler(slot); + try std.testing.expectEqual(@as(usize, 1), calls); + try std.testing.expectEqual(api.Action.close, slot.action); + try std.testing.expect(slot.in_use); + slot.phase.store(.io, .release); + server.maybeFree(slot); + try std.testing.expect(!slot.in_use and server.free_count == 1); + try std.testing.expectEqual(@as(u64, 0), server.stats.live_connections); + } + } + + test "terminal results disarm cancellation even when cancellation races their publication" { + const Shared = struct { + calls: usize = 0, + finish: bool, + cancel_inside: bool, + fn handler(ctx: *api.Context) api.Action { + const shared: *@This() = @ptrCast(@alignCast(ctx.application.?)); + assert(ctx.event != .cancelled); + shared.calls += 1; + ctx.requestCancellation(); + if (shared.cancel_inside) @constCast(ctx.cancelled).store(true, .release); + ctx.state[0] = 0; // The application released its continuation lease. + if (!shared.finish) return .close; + ctx.writer.begin(200, "text/plain", 0) catch unreachable; + return ctx.writer.finish(); + } + }; + for ([_]bool{ false, true }) |finish| { + for ([_]bool{ false, true }) |cancel_inside| { + var shared: Shared = .{ .finish = finish, .cancel_inside = cancel_inside }; + const server = try Server.init(std.testing.allocator, .{ + .execution = .workers, + .workers = 1, + .connections = 1, + .port = 0, + }, Shared.handler, &shared); + defer server.deinit(); + const slot = &server.slots[0]; + slot.in_use = true; + slot.request_active = true; + slot.state[0] = 42; + slot.notify_cancel = true; + slot.writer.open(0, true, false); + server.header_cache.refresh("Sat, 05 Sep 2026 12:34:56 GMT"); + server.stats.live_connections = 1; + server.free_count = 0; + slot.phase.store(.running, .release); + server.invokeHandler(slot); + try std.testing.expect(!slot.notify_cancel); + try std.testing.expectEqual(Phase.result, slot.phase.load(.acquire)); + try server.beginClose(slot); + try std.testing.expect(slot.in_use); + try server.serviceSlot(0); + try std.testing.expect(!slot.in_use); + try std.testing.expectEqual(@as(usize, 1), shared.calls); + try std.testing.expectEqual(@as(usize, 0), slot.state[0]); + } + } + } + + test "timers release the callback, preserve deadlines, and cancel instead of extending requests" { + const server = try Server.init(std.testing.allocator, .{ + .execution = .inline_event_loop, + .workers = 0, + .connections = 1, + .port = 0, + }, struct { + fn handler(ctx: *api.Context) api.Action { + if (ctx.event == .request) { + ctx.requestCancellation(); + return ctx.wait(50) catch .close; + } + assert(ctx.event == .cancelled); + return .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.deadline = nowNs() + std.time.ns_per_s; + const deadline = slot.deadline; + server.free_count = 0; + server.stats.live_connections = 1; + server.now = 1; // Deliberately stale turn clock must not shorten a new wait. + slot.phase.store(.running, .release); + server.invokeHandler(slot); + const before = nowNs(); + try server.serviceSlot(0); + const after = nowNs(); + const at = slot.resume_at.?; + try std.testing.expect(at >= before + 50 and at <= after + 50); + try std.testing.expectEqual(Phase.io, slot.phase.load(.acquire)); + server.now = at - 1; + try server.serviceTimers(); + try std.testing.expectEqual(Phase.io, slot.phase.load(.acquire)); + server.now = at; + try server.serviceTimers(); + try std.testing.expectEqual(api.Event.timer, slot.event); + try std.testing.expectEqual(Phase.ready, slot.phase.load(.acquire)); + try std.testing.expectEqual(deadline, slot.deadline); + _ = server.popReady(); + slot.phase.store(.io, .release); + slot.resume_at = deadline + 5; + server.next_timer = deadline + 5; + server.now = deadline + 5; + try server.serviceTimers(); + try std.testing.expectEqual(api.Event.cancelled, slot.event); + try std.testing.expect(slot.resume_at == null and slot.in_use); + _ = 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 "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, + .workers = 0, + .connections = 1, + .port = 0, + }, struct { + fn handler(_: *api.Context) api.Action { + unreachable; + } + }.handler, null); + defer server.deinit(); + const slot = &server.slots[0]; + slot.in_use = true; + slot.request_active = true; + slot.request.keep_alive = true; + slot.request.head_only = false; + slot.deadline = 1000; + slot.resume_at = 123; + slot.writer.open(0, true, false); + slot.writer.frozen = true; + slot.batch_count = 1; + slot.cells[0] = .{ .finished = true }; + slot.batch_next = .resume_timer; + slot.send_pending = true; + server.stats.live_connections = 1; + server.free_count = 0; + server.now = 123; + server.next_timer = 123; + try server.serviceTimers(); + try std.testing.expectEqual(@as(?u64, 123), slot.resume_at); + try std.testing.expect(server.next_timer == null and slot.writer.frozen); + try std.testing.expectEqual(Phase.io, slot.phase.load(.acquire)); + slot.send_pending = false; + try server.completeBatch(0); + try std.testing.expectEqual(@as(?u64, 123), server.next_timer); + try std.testing.expect(!slot.writer.frozen and !slot.writer.began); + try server.serviceTimers(); + try std.testing.expectEqual(api.Event.timer, slot.event); + try std.testing.expect(slot.resume_at == null); + _ = server.popReady(); + slot.phase.store(.io, .release); + try server.beginClose(slot); + try std.testing.expect(!slot.in_use); + } + + test "cancelling a queued resume preserves one cancellation callback" { + var calls: usize = 0; + const server = try Server.init(std.testing.allocator, .{ + .execution = .workers, + .workers = 1, + .connections = 1, + .port = 0, + }, struct { + fn handler(ctx: *api.Context) api.Action { + assert(ctx.event == .cancelled); + const count: *usize = @ptrCast(@alignCast(ctx.application.?)); + count.* += 1; + return .close; + } + }.handler, &calls); + defer server.deinit(); + const slot = &server.slots[0]; + slot.in_use = true; + slot.request_active = true; + slot.event = .timer; + slot.notify_cancel = true; + slot.cancelled.store(true, .release); + server.stats.live_connections = 1; + server.free_count = 0; + slot.phase.store(.running, .release); + server.invokeHandler(slot); + try std.testing.expectEqual(@as(usize, 0), calls); + try std.testing.expect(slot.notify_cancel); + try server.serviceSlot(0); + try std.testing.expectEqual(api.Event.cancelled, slot.event); + try std.testing.expectEqual(Phase.ready, slot.phase.load(.acquire)); + slot.phase.store(.running, .release); + server.invokeHandler(slot); + try server.serviceSlot(0); + try std.testing.expectEqual(@as(usize, 1), calls); + try std.testing.expect(!slot.in_use); + } + test "blocking flush resumes the same stack and preserves the request deadline" { const server = try Server.init(std.testing.allocator, .{ .execution = .workers, @@ -1637,6 +1957,7 @@ pub const Server = struct { } slot.in_use = true; slot.closing = true; + slot.resume_at = null; slot.cancelled.store(true, .release); slot.phase.store(.stream_wait, .release); server.stats.live_connections = 1; diff --git a/tests/continuation_integration.py b/tests/continuation_integration.py new file mode 100644 index 0000000..883ec39 --- /dev/null +++ b/tests/continuation_integration.py @@ -0,0 +1,217 @@ +#!/usr/bin/env python3 +"""Finite correctness gates for callback-returning timers and cancellation.""" +import argparse +import contextlib +import json +import platform +from pathlib import Path +import re +import socket +import signal +import os +import time + +from integration import Server as BaseServer, ResponseReader, require, SERVER_BINARY + + +def Server(**options): + return BaseServer(SERVER_BINARY, **options) + + +def request(path, close=False): + return (f'GET {path} HTTP/1.1\r\nHost: localhost\r\n' + + ('Connection: close\r\n' if close else '') + '\r\n').encode() + + +def fetch(server, path): + with server.connect() as client: + client.sendall(request(path, True)) + return ResponseReader(client).response() + + +def counts(server): + status, _, body = fetch(server, '/continuation-counts') + require(status == 200, 'counter route failed') + return json.loads(body) + + +def await_count(server, key, minimum): + deadline = time.monotonic() + 2 + current = {} + while time.monotonic() < deadline: + try: + current = counts(server) + except (EOFError, ConnectionResetError, ConnectionAbortedError, ConnectionRefusedError): + require(server.process.poll() is None, 'server exited while awaiting terminal state') + time.sleep(.01) + continue + if current[key] >= minimum: + return current + time.sleep(.01) + raise AssertionError(f'{key} did not reach {minimum}: {current}') + + +def final_counts(server): + for line in server.lines: + match = re.fullmatch(r'CONTINUATIONS started=(\d+) finished=(\d+) cancelled=(\d+) live=(\d+)', line) + if match: + result = dict(zip(('started', 'finished', 'cancelled', 'live'), map(int, match.groups()))) + require(result['live'] == 0, 'application state remained after shutdown') + require(result['started'] == result['finished'] + result['cancelled'], 'terminal callback count mismatch') + return result + raise AssertionError('missing final continuation counters') + + +def shared_options(mode): + return dict(shards=1, execution=mode, workers=1 if mode == 'workers' else 0, + connections=64, send_chunk=3, timeout_ms=4000, + output_bytes=2048, max_response=4096) + + +def framing_and_reuse(mode): + with Server(**shared_options(mode), stall_ms=1) as server: + with server.connect(timeout=8) as client: + client.sendall(b''.join(request('/timed-chunks') for _ in range(128))) + reader = ResponseReader(client) + for _ in range(128): + status, headers, body = reader.response() + require(status == 200 and headers.get(b'transfer-encoding') == b'chunked', 'stream framing failed') + require(body == b'first second third', 'resume lost or replayed body bytes') + require(fetch(server, '/empty-timer')[2] == b'', 'empty flush/timer changed body') + current = counts(server) + require(current == dict(started=129, finished=129, cancelled=0, live=0), 'handler replay or leaked finished state') + with server.connect() as client: + client.sendall(request('/abort-after-flush')) + try: + ResponseReader(client).response() + except (EOFError, ConnectionResetError): + pass + else: + raise AssertionError('post-flush failure published a completed response') + current = await_count(server, 'finished', 130) + require(current['started'] == 130 and current['live'] == 0, 'error cleanup repeated or leaked') + final_counts(server) + + +def many_waits(mode): + with Server(**shared_options(mode), stall_ms=1200) as server: + with contextlib.ExitStack() as stack: + clients = [stack.enter_context(server.connect()) for _ in range(32)] + for client in clients: + client.sendall(request('/wait-only')) + current = await_count(server, 'started', 32) + require(current['live'] == 32, 'waiting handlers retained the worker or timer expired before admission') + require(fetch(server, '/plaintext')[2] == b'Hello, World!', 'unrelated request could not progress during waits') + for client in clients: + require(ResponseReader(client).response()[2] == b'done', 'timer response mismatch') + with server.connect() as client: + client.sendall(request('/plaintext') + request('/wait-only')) + reader = ResponseReader(client) + client.settimeout(.7) + require(reader.response()[2] == b'Hello, World!', 'timer retained an earlier batched response') + client.settimeout(3) + require(reader.response()[2] == b'done', 'timer failed after preceding batch drained') + current = counts(server) + require(current == dict(started=33, finished=33, cancelled=0, live=0), 'timer finish count mismatch') + final_counts(server) + + +def timeout_and_shutdown(mode): + options = shared_options(mode) + options['timeout_ms'] = 150 + options['deadline_sweep_ms'] = 5 + with Server(**options, stall_ms=10000) as server: + with server.connect() as client: + client.sendall(request('/wait-only')) + require(client.recv(1) == b'', 'sticky request deadline did not cancel timer') + current = await_count(server, 'cancelled', 1) + require(current['live'] == 0 and current['started'] == 1, 'timeout cleanup did not run exactly once') + with contextlib.ExitStack() as stack: + clients = [stack.enter_context(server.connect()) for _ in range(12)] + for client in clients: + client.sendall(request('/wait-only')) + await_count(server, 'started', 13) + server.request_stop() + current = final_counts(server) + require(current['cancelled'] == 13 and current['finished'] == 0, 'shutdown failed to drain cancellation callbacks') + + +def disconnect_after_flush(mode): + options = shared_options(mode) + options['connections'] = 1 + with Server(**options, stall_ms=300) as server: + with server.connect() as client: + client.sendall(request('/timed-chunks')) + reader = ResponseReader(client) + require(reader._line().startswith(b'HTTP/1.1 200 '), 'initial flush was not visible') + while reader._line(): + pass + size = int(reader._line(), 16) + require(reader._take(size) == b'first ', 'first chunk changed') + require(reader._take(2) == b'\r\n', 'first chunk terminator changed') + import struct + client.setsockopt(socket.SOL_SOCKET, socket.SO_LINGER, + struct.pack('hh' if __import__('os').name == 'nt' else 'ii', 1, 0)) + current = await_count(server, 'cancelled', 1) + require(current['live'] == 0, 'disconnect retained application state') + require(fetch(server, '/empty-timer')[0] == 200, 'slot reuse after cancellation failed') + final_counts(server) + + +def sharded_timers(): + options = shared_options('inline') + options['shards'] = 3 + with Server(**options, stall_ms=100) as server: + with contextlib.ExitStack() as stack: + clients = [stack.enter_context(server.connect()) for _ in range(24)] + for client in clients: + client.sendall(request('/timed-chunks')) + for client in clients: + require(ResponseReader(client).response()[2] == b'first second third', 'sharded timer body mismatch') + require(counts(server) == dict(started=24, finished=24, cancelled=0, live=0), 'sharded state accounting mismatch') + final_counts(server) + require(server.stats['shards'] == 3, 'configured shard count changed') + + +def main(): + parser = argparse.ArgumentParser() + parser.add_argument('--server', type=Path) + parser.add_argument('--json', type=Path) + args = parser.parse_args() + if args.server: + global SERVER_BINARY + 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): + started = time.monotonic() + test(mode) + result = dict(name=f'{test.__name__}/{mode}', passed=True, seconds=round(time.monotonic() - started, 3)) + results.append(result) + print(f"PASS {result['name']}", flush=True) + started = time.monotonic() + if platform.system() == 'Darwin': + results.append(dict(name='sharded_timers/inline', skipped='macOS engine supports one owner')) + print('SKIP sharded_timers/inline: macOS engine supports one owner', flush=True) + else: + sharded_timers() + results.append(dict(name='sharded_timers/inline', passed=True, seconds=round(time.monotonic() - started, 3))) + print('PASS sharded_timers/inline', flush=True) + packet = dict(ok=True, groups=len(results), results=results) + if args.json: + args.json.write_text(json.dumps(packet, indent=2) + '\n') + print(json.dumps(packet)) + + +if __name__ == '__main__': + # Windows runs under the existing process-tree supervisor in verify_windows.py. + if os.name == 'posix': + def watchdog(signum, frame): + raise TimeoutError('continuation suite exceeded its 120-second watchdog') + signal.signal(signal.SIGALRM, watchdog) + signal.alarm(120) + try: + main() + finally: + if os.name == 'posix': + signal.alarm(0) diff --git a/tools/verify_windows.py b/tools/verify_windows.py index ec746d9..bb74245 100644 --- a/tools/verify_windows.py +++ b/tools/verify_windows.py @@ -122,6 +122,12 @@ def run(name, arguments, timeout): '--timeout', '120', '--json', str(packet / (name + '.json'))], 150) wire = json.loads((packet / (name + '.json')).read_text()) require(wire.get('ok') is True, name + ' did not produce a passing receipt') + run('continuation-integration', + [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') == 9, + 'Continuation timer/cancellation receipt mismatch') run('windows-shards-integration', [sys.executable, 'tests/windows_shards_integration.py', '--server', str(binary), '--timeout', '120', '--json', str(packet / 'windows-shards-integration.json')], 150)