Let a callback select a bounded per-request deadline

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. |

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();

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")) {