Add bounded timer waits and cancellation callbacks

.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

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

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.

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));
+}

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.