//! A hardened HTTP client over std.http.Client.
//!
//! `Client.fetch` runs one request under a response-head deadline, a body
//! stall guard, and dead-connection replay for idempotent requests, and
//! identifies itself with a real User-Agent. Requires std.Io (zig 0.16+).

const std = @import("std");
const build_options = @import("build_options");

const log = std.log.scoped(.zttp);

/// Sent unless a caller overrides `user_agent`. std.http's default is
/// `zig/<version> (std.http)`, which tells a server operator nothing about who
/// is making the request or how to reach them. Anything calling other people's
/// servers at scale should say what it is: operators seeing unexplained load
/// have no way to ask about it otherwise.
pub const default_user_agent = "zttp/" ++ build_options.version ++ " (+https://tangled.org/zat.dev/zttp)";

/// Failures this client adds on top of std.http's own. `fetch` returns these
/// or a std.http / std.Io error; callers classifying errors can switch on
/// this set and treat the rest as transport failures.
pub const Error = error{
    /// No response head within `head_deadline_ns`. Nothing was received, so
    /// an idempotent request is safe to retry elsewhere.
    ResponseHeadTimeout,
    /// No body bytes for `StallGuard.idle_ns`.
    TransferStalled,
    /// Body rate under `StallGuard.min_bytes_per_s` after the grace period.
    TransferTooSlow,
    /// `FetchOptions.deadline_ns` elapsed. The request may have taken effect.
    RequestDeadlineExceeded,
    /// Body exceeded `max_response_size` or `StreamOptions.max_bytes`.
    ResponseTooLarge,
    /// `Response.stream` was called twice.
    BodyAlreadyTaken,
    /// `accept` plus `extra_headers` exceed the header buffer.
    TooManyHeaders,
    /// A payload was given without a `content_type`.
    MissingContentType,
    /// The URL contains a control character. std would write it to the
    /// request line verbatim, which is header injection when any part of
    /// the URL came from outside.
    InvalidUrl,
    /// A header name or value contains CR, LF, or (for names) a colon.
    InvalidHeader,
    /// `resolved_connection.logical_host` does not match the URL host.
    ResolvedHostMismatch,
    /// More hops than `RedirectPolicy.max_hops`.
    TooManyRedirects,
    /// A redirect target was already visited in this chain.
    RedirectLoop,
    /// An https origin redirected to http and downgrades are not allowed.
    SchemeDowngrade,
    /// A 3xx arrived and `on_redirect` is `.refuse`.
    RedirectRefused,
    /// A 3xx status this client does not know how to follow.
    HttpRedirectUnsupported,
    HttpRedirectLocationMissing,
    HttpRedirectLocationInvalid,
};

/// Concurrent `fetch` calls on one client are fine: the connection pool and
/// CA bundle take their own locks. The configuration fields are plain
/// values; set them before sharing the client, and use the per-request
/// overrides on `FetchOptions` after that.
pub const Client = struct {
    allocator: std.mem.Allocator,
    io: std.Io,
    http_client: std.http.Client,
    tls_connect_lock: std.Io.Mutex = .init,
    keep_alive: bool = true,
    /// Default cap on a buffered response body; `FetchOptions.max_response_size`
    /// overrides it per request. null is unbounded.
    max_response_size: ?usize = null,
    /// Guards against a peer that accepts the connection and then trickles or
    /// stops. Without these a single slow origin holds a worker for as long as
    /// the caller's own deadline allows, and a pool of workers can sit "busy"
    /// while moving almost no bytes. Set any field to 0 to disable that check.
    stall: StallGuard = .{},
    /// Bounds connect, TLS, request send, and the wait for the response head —
    /// the phases StallGuard cannot see, because no body bytes exist yet. A
    /// peer that accepts and never answers, or a SYN that goes unanswered,
    /// otherwise blocks the caller until the kernel gives up (~2 minutes) or
    /// forever on an established socket. 0 disables.
    head_deadline_ns: u64 = 10 * std.time.ns_per_s,
    /// A pooled connection the origin closed while it sat idle looks identical
    /// to a request that failed, but nothing was ever delivered — so replaying
    /// an idempotent request on a fresh connection is safe, and skipping the
    /// replay turns routine housekeeping into a caller-visible error.
    dead_connection_attempts: usize = 3,
    /// How `fetch` follows 3xx responses. `authorization` never crosses an
    /// origin boundary regardless of these settings.
    redirects: RedirectPolicy = .{},
    /// Identify this client to the servers it calls. Applications should set
    /// their own; the default names this library so an operator can at least
    /// tell what is calling.
    user_agent: []const u8 = default_user_agent,

    pub fn init(io: std.Io, allocator: std.mem.Allocator) Client {
        return .{
            .allocator = allocator,
            .io = io,
            .http_client = .{ .allocator = allocator, .io = io },
        };
    }

    /// Same as `init`, naming the calling application rather than this library.
    pub fn initWithUserAgent(io: std.Io, allocator: std.mem.Allocator, user_agent: []const u8) Client {
        var transport = init(io, allocator);
        if (user_agent.len > 0) transport.user_agent = user_agent;
        return transport;
    }

    pub fn deinit(self: *Client) void {
        self.http_client.deinit();
    }

    /// fetch a URL and write response to provided writer
    /// Open, stream the whole body into memory, close. The one-call shape
    /// for bounded responses; `open` is the same request with the body left
    /// to the caller.
    pub fn fetch(self: *Client, options: FetchOptions) !FetchResult {
        const response = try self.open(options);
        defer response.deinit();

        var aw: std.Io.Writer.Allocating = .init(self.allocator);
        defer aw.deinit();
        _ = try response.stream(&aw.writer, .{
            .max_bytes = options.max_response_size orelse self.max_response_size,
            .stall = options.stall,
        });

        const body = try self.allocator.dupe(u8, aw.written());
        var result: FetchResult = .{
            .status = response.status,
            .body = body,
            .rate_limit = response.rate_limit,
            .headers = response.headers,
            .final_url = response.final_url,
        };
        _ = &result;
        // Ownership of the captured headers and url moves to the result.
        response.headers = .{};
        response.final_url = null;
        return result;
    }

    /// Send the request and return once the response head is in. Every
    /// guard on the request phase applies: the head deadline, dead-connection
    /// replay, the redirect policy, input validation. The body is untouched
    /// until `Response.stream`; `Response.deinit` closes it either way.
    pub fn open(self: *Client, options: FetchOptions) !*Response {
        if (options.payload != null and options.content_type == null) return error.MissingContentType;
        for (options.url) |c| if (c < 0x20 or c == 0x7f) return error.InvalidUrl;
        if (options.authorization) |v| try validateHeaderValue(v);
        if (options.accept) |v| try validateHeaderValue(v);
        if (options.accept_encoding) |v| try validateHeaderValue(v);
        if (options.content_type) |v| try validateHeaderValue(v);
        if (options.extra_headers) |hdrs| for (hdrs) |h| {
            if (h.name.len == 0) return error.InvalidHeader;
            for (h.name) |c| if (c == ':' or c == '\r' or c == '\n') return error.InvalidHeader;
            try validateHeaderValue(h.value);
        };

        var extra_buf: [8]std.http.Header = undefined;
        var extra_count: usize = 0;

        if (options.accept) |accept| {
            extra_buf[extra_count] = .{ .name = "accept", .value = accept };
            extra_count += 1;
        }

        if (options.extra_headers) |hdrs| {
            if (extra_count + hdrs.len > extra_buf.len) return error.TooManyHeaders;
            for (hdrs) |h| {
                extra_buf[extra_count] = h;
                extra_count += 1;
            }
        }

        const extra = extra_buf[0..extra_count];
        var attempt: usize = 0;
        while (true) {
            attempt += 1;
            const result = self.openDeadlined(options, extra);

            return result catch |err| {
                if (attempt < self.dead_connection_attempts and
                    isDeadConnection(err) and
                    isReplayable(options)) continue;
                return err;
            };
        }
    }

    const HeadProgress = struct {
        done: std.atomic.Value(bool) = .init(false),
    };

    /// Run one attempt under the head deadline. The attempt runs as a task;
    /// the watcher cancels it if the head has not arrived in time. Whatever
    /// the request deadline has left is handed to the response for the body.
    fn openDeadlined(
        self: *Client,
        options: FetchOptions,
        extra_headers: []const std.http.Header,
    ) !*Response {
        var progress: HeadProgress = .{};
        const head_deadline_ns = options.head_deadline_ns orelse self.head_deadline_ns;
        const total_deadline_ns = options.deadline_ns orelse 0;
        const deadline_ns = if (head_deadline_ns == 0) total_deadline_ns else if (total_deadline_ns == 0) head_deadline_ns else @min(head_deadline_ns, total_deadline_ns);
        if (deadline_ns == 0)
            return self.openOnce(options, extra_headers, &progress);

        var future = self.io.concurrent(openOnce, .{ self, options, extra_headers, &progress }) catch |err| switch (err) {
            error.ConcurrencyUnavailable => {
                warnUnguarded();
                const response = try self.openOnce(options, extra_headers, &progress);
                response.deadline_remaining_ns = total_deadline_ns;
                return response;
            },
        };

        const tick_ns: u64 = 10 * std.time.ns_per_ms;
        var elapsed_ns: u64 = 0;
        while (!progress.done.load(.acquire)) {
            if (elapsed_ns >= deadline_ns) {
                if (future.cancel(self.io)) |completed| completed.deinit() else |_| {}
                return if (total_deadline_ns != 0 and elapsed_ns >= total_deadline_ns)
                    error.RequestDeadlineExceeded
                else
                    error.ResponseHeadTimeout;
            }
            std.Io.Clock.Duration.sleep(
                .{ .raw = .fromNanoseconds(tick_ns), .clock = .awake },
                self.io,
            ) catch {
                if (future.cancel(self.io)) |completed| completed.deinit() else |_| {}
                return error.Canceled;
            };
            elapsed_ns += tick_ns;
        }

        const response = try future.await(self.io);
        if (total_deadline_ns != 0) response.deadline_remaining_ns = total_deadline_ns -| elapsed_ns;
        return response;
    }

    fn openOnce(
        self: *Client,
        options: FetchOptions,
        extra_headers: []const std.http.Header,
        progress: *HeadProgress,
    ) !*Response {
        defer progress.done.store(true, .release);
        const response = try self.allocator.create(Response);
        errdefer self.allocator.destroy(response);
        response.* = .{ .client = self, .request = undefined, .inner = undefined };
        if (options.resolved_connection) |resolved|
            try self.openResolved(response, options, resolved, extra_headers)
        else
            try self.openUrl(response, options, extra_headers);
        return response;
    }

    /// One hop of a request: everything about it that a redirect may rewrite.
    const Hop = struct {
        url: []const u8,
        method: std.http.Method,
        payload: ?[]const u8,
        content_type: ?[]const u8,
        authorization: ?[]const u8,
        accept_encoding: ?[]const u8,

        fn fromOptions(options: FetchOptions) Hop {
            return .{
                .url = options.url,
                .method = options.method,
                .payload = options.payload,
                .content_type = options.content_type,
                .authorization = options.authorization,
                .accept_encoding = options.accept_encoding,
            };
        }
    };

    fn requestHeaders(self: *Client, hop: Hop) std.http.Client.Request.Headers {
        return .{
            .content_type = if (hop.content_type) |ct| .{ .override = ct } else .default,
            .authorization = if (hop.authorization) |auth| .{ .override = auth } else .default,
            .accept_encoding = if (hop.accept_encoding) |enc| .{ .override = enc } else .default,
            .user_agent = .{ .override = self.user_agent },
        };
    }

    fn createRequest(
        self: *Client,
        method: std.http.Method,
        uri: std.Uri,
        options: std.http.Client.RequestOptions,
        resolved: ?std.http.Client.ConnectTcpOptions,
    ) !std.http.Client.Request {
        const protocol = std.http.Client.Protocol.fromUri(uri) orelse return error.UnsupportedUriScheme;
        const tls = !std.http.Client.disable_tls and protocol == .tls;
        if (tls) try self.tls_connect_lock.lock(self.io);
        defer if (tls) self.tls_connect_lock.unlock(self.io);

        if (!std.http.Client.disable_tls and protocol == .tls) {
            const now = std.Io.Clock.real.now(self.io);
            if (self.http_client.now == null) {
                self.http_client.ca_bundle.rescan(self.allocator, self.io, now) catch |err| switch (err) {
                    error.Canceled => return error.Canceled,
                    else => return error.CertificateBundleLoadFailure,
                };
            }
            @import("trust.zig").distrust(&self.http_client.ca_bundle);
            self.http_client.now = now;
        }

        var request_options = options;
        if (resolved) |dial| {
            const connection = try self.http_client.connectTcpOptions(dial);
            errdefer self.http_client.connection_pool.release(connection, self.io);
            request_options.connection = connection;
            return self.http_client.request(method, uri, request_options);
        }
        return self.http_client.request(method, uri, request_options);
    }

    fn sendRequest(request: *std.http.Client.Request, payload: ?[]const u8) !void {
        if (payload) |bytes| {
            request.transfer_encoding = .{ .content_length = bytes.len };
            var body = try request.sendBodyUnflushed(&.{});
            try body.writer.writeAll(bytes);
            try body.end();
            try request.connection.?.flush();
        } else {
            try request.sendBodiless();
        }
    }

    /// std reports every failed socket read as `error.ReadFailed` and keeps the
    /// cause on the connection. Surface the cause, so a reset on a pooled
    /// connection is recognised as a dead connection and replayed.
    fn receiveHead(request: *std.http.Client.Request) !std.http.Client.Response {
        return request.receiveHead(&.{}) catch |err| switch (err) {
            error.ReadFailed => request.connection.?.getReadError() orelse err,
            else => err,
        };
    }

    /// Redirects are handled here rather than by std: std's own loop keeps
    /// `authorization` across hosts (its `privileged_headers` are stripped on
    /// a domain change but never written to the wire in 0.16), has no loop
    /// detection, and will happily step from https down to http.
    ///
    /// The request lives in `response` so that std's internal pointers into
    /// it stay valid once the head is in; each hop rebuilds it in place.
    fn openUrl(
        self: *Client,
        response: *Response,
        options: FetchOptions,
        extra_headers: []const std.http.Header,
    ) !void {
        var hop: Hop = .fromOptions(options);
        var owned_url: ?[]u8 = null;
        defer if (owned_url) |u| self.allocator.free(u);
        var visited: std.ArrayList([]u8) = .empty;
        defer {
            for (visited.items) |u| self.allocator.free(u);
            visited.deinit(self.allocator);
        }
        var hops: u8 = 0;

        while (true) {
            const uri = try std.Uri.parse(hop.url);
            var status: std.http.Status = undefined;
            const next = blk: {
                response.request = try self.createRequest(hop.method, uri, .{
                    .headers = self.requestHeaders(hop),
                    .extra_headers = extra_headers,
                    .keep_alive = self.keep_alive,
                    .redirect_behavior = .unhandled,
                }, null);
                errdefer response.request.deinit();
                try sendRequest(&response.request, hop.payload);

                response.inner = try receiveHead(&response.request);
                if (response.inner.head.status.class() != .redirect or options.on_redirect == .surface) {
                    try response.captureHead(options);
                    response.final_url = owned_url;
                    owned_url = null;
                    return;
                }
                if (options.on_redirect == .refuse) return error.RedirectRefused;
                if (hops >= self.redirects.max_hops) return error.TooManyRedirects;
                hops += 1;

                const next = try self.resolveRedirect(uri, hop.url, &response.inner);
                status = response.inner.head.status;
                var transfer_buf: [4096]u8 = undefined;
                const body = response.inner.reader(&transfer_buf);
                _ = body.discardRemaining() catch {};
                break :blk next;
            };
            response.request.deinit();
            errdefer self.allocator.free(next);

            if (std.ascii.eqlIgnoreCase(hop.url, next)) return error.RedirectLoop;
            for (visited.items) |seen| if (std.ascii.eqlIgnoreCase(seen, next)) return error.RedirectLoop;
            try visited.append(self.allocator, try self.allocator.dupe(u8, hop.url));

            const next_uri = try std.Uri.parse(next);
            if (!self.redirects.allow_scheme_downgrade and
                std.ascii.eqlIgnoreCase(uri.scheme, "https") and
                std.ascii.eqlIgnoreCase(next_uri.scheme, "http"))
                return error.SchemeDowngrade;
            if (!try sameOrigin(uri, next_uri)) hop.authorization = null;

            switch (status) {
                .see_other => {
                    hop.method = .GET;
                    hop.payload = null;
                    hop.content_type = null;
                },
                .moved_permanently, .found => if (hop.method == .POST) {
                    hop.method = .GET;
                    hop.payload = null;
                    hop.content_type = null;
                },
                .temporary_redirect, .permanent_redirect => {},
                else => return error.HttpRedirectUnsupported,
            }

            if (owned_url) |u| self.allocator.free(u);
            owned_url = next;
            hop.url = next;
        }
    }

    /// Resolve the response's `Location` against the request URI into an
    /// owned absolute URL. Must run before the body is read: reading
    /// invalidates the head's strings.
    fn resolveRedirect(self: *Client, base: std.Uri, base_url: []const u8, response: *std.http.Client.Response) ![]u8 {
        const location = response.head.location orelse return error.HttpRedirectLocationMissing;
        // resolveInPlace wants the location at the front of a buffer with
        // room for a merged path after it.
        var aux = try self.allocator.alloc(u8, location.len + base_url.len + location.len + 2);
        defer self.allocator.free(aux);
        @memcpy(aux[0..location.len], location);
        var aux_slice: []u8 = aux;
        const resolved = base.resolveInPlace(location.len, &aux_slice) catch return error.HttpRedirectLocationInvalid;
        if (resolved.host == null) return error.HttpRedirectLocationInvalid;
        return try std.fmt.allocPrint(self.allocator, "{f}", .{resolved});
    }

    /// A connection the caller resolved is never redirected away from: the
    /// caller vetted that address, and a `Location` pointing anywhere else
    /// would be dialed unvetted. A 3xx is surfaced instead.
    fn openResolved(
        self: *Client,
        response: *Response,
        options: FetchOptions,
        resolved: ResolvedConnection,
        extra_headers: []const std.http.Header,
    ) !void {
        const uri = try std.Uri.parse(options.url);
        const protocol = std.http.Client.Protocol.fromUri(uri) orelse return error.UnsupportedUriScheme;

        var uri_host_buf: [std.Io.net.HostName.max_len]u8 = undefined;
        const uri_host = try std.Io.net.HostName.fromUri(uri, &uri_host_buf);
        if (!std.ascii.eqlIgnoreCase(uri_host.bytes, resolved.logical_host)) {
            return error.ResolvedHostMismatch;
        }

        const dial_host = try std.Io.net.HostName.init(resolved.dial_host);
        const logical_host = try std.Io.net.HostName.init(resolved.logical_host);
        response.request = try self.createRequest(options.method, uri, .{
            .headers = self.requestHeaders(.fromOptions(options)),
            .extra_headers = extra_headers,
            .keep_alive = self.keep_alive,
            .redirect_behavior = .unhandled,
        }, .{
            .host = dial_host,
            .port = uri.port orelse defaultPort(protocol),
            .protocol = protocol,
            .proxied_host = logical_host,
        });
        errdefer response.request.deinit();
        try sendRequest(&response.request, options.payload);

        response.inner = try receiveHead(&response.request);
        if (response.inner.head.status.class() == .redirect and options.on_redirect == .refuse) return error.RedirectRefused;
        try response.captureHead(options);
    }

    /// A request whose head has arrived. The body is still on the wire:
    /// `stream` it into any writer, or `deinit` to drop it.
    pub const Response = struct {
        client: *Client,
        request: std.http.Client.Request,
        inner: std.http.Client.Response,
        status: std.http.Status = .ok,
        rate_limit: RateLimitHeaders = .{},
        headers: CapturedHeaders = .{},
        /// Where the response actually came from when redirects were
        /// followed; null when it came from the requested URL.
        final_url: ?[]u8 = null,
        /// What `FetchOptions.deadline_ns` has left for the body. 0 is no bound.
        deadline_remaining_ns: u64 = 0,
        body_taken: bool = false,

        fn captureHead(self: *Response, options: FetchOptions) !void {
            self.status = self.inner.head.status;
            self.rate_limit = RateLimitHeaders.fromResponseHead(self.inner.head);
            self.headers = try CapturedHeaders.fromResponseHead(self.client.allocator, self.inner.head, options.capture_headers);
        }

        pub const StreamOptions = struct {
            /// Hard cap; `error.ResponseTooLarge` past it. A writer that
            /// truncates instead is the caller's choice of `sink`.
            max_bytes: ?usize = null,
            /// Overrides the client's stall guard for this body.
            stall: ?StallGuard = null,
            /// Byte and completion counters the caller may watch from
            /// another task, for its own reporting or watchdog.
            progress: ?*Progress = null,
        };

        /// Decode and drain the body into `sink` under the stall guard and
        /// whatever the request deadline has left. Returns the decoded byte
        /// count. Call once.
        pub fn stream(self: *Response, sink: *std.Io.Writer, options: StreamOptions) !u64 {
            if (self.body_taken) return error.BodyAlreadyTaken;
            self.body_taken = true;
            const client = self.client;
            const dbuf = try client.allocator.alloc(u8, decompressBufferLen(self.inner.head.content_encoding));
            defer client.allocator.free(dbuf);
            var own_progress: Progress = .{};
            const progress = options.progress orelse &own_progress;
            try streamResponseBody(
                client.io,
                &self.inner,
                sink,
                options.stall orelse client.stall,
                options.max_bytes,
                self.deadline_remaining_ns,
                dbuf,
                progress,
            );
            return progress.bytes.load(.acquire);
        }

        pub fn deinit(self: *Response) void {
            const allocator = self.client.allocator;
            self.request.deinit();
            self.headers.deinit(allocator);
            if (self.final_url) |u| allocator.free(u);
            allocator.destroy(self);
        }
    };

    pub const FetchOptions = struct {
        url: []const u8,
        method: std.http.Method = .GET,
        payload: ?[]const u8 = null,
        authorization: ?[]const u8 = null,
        accept: ?[]const u8 = null,
        /// Replaces std's default `gzip, deflate`. `identity` asks for an
        /// uncompressed body; a server that compresses anyway is still
        /// decoded.
        accept_encoding: ?[]const u8 = null,
        content_type: ?[]const u8 = null,
        extra_headers: ?[]const std.http.Header = null,
        /// Overrides `Client.max_response_size` for this request.
        max_response_size: ?usize = null,
        /// Overrides `Client.head_deadline_ns` for this request.
        head_deadline_ns: ?u64 = null,
        /// Overrides `Client.stall` for this request.
        stall: ?StallGuard = null,
        /// Bounds the whole request, head and body, including every redirect
        /// hop and dead-connection replay of this attempt. Fires as
        /// `error.RequestDeadlineExceeded`. null is unbounded.
        deadline_ns: ?u64 = null,
        on_redirect: OnRedirect = .follow,
        resolved_connection: ?ResolvedConnection = null,
        /// Response header names (case-insensitive) to copy into
        /// `FetchResult.headers`. Everything else in the head is dropped once
        /// the body has been read.
        capture_headers: []const []const u8 = &.{},
    };

    pub const OnRedirect = enum {
        /// Follow under `Client.redirects`. Bodies are resent only for 307 and
        /// 308; 303, and 301/302 on POST, become GET without a body.
        follow,
        /// Return the 3xx response itself.
        surface,
        /// Fail with `error.RedirectRefused`.
        refuse,
    };

    pub const RedirectPolicy = struct {
        /// Hops followed before `error.TooManyRedirects`.
        max_hops: u8 = 3,
        /// Whether an https origin may send the request on to plain http.
        allow_scheme_downgrade: bool = false,
    };

    pub const StallGuard = struct {
        /// Abort if no body bytes arrive for this long.
        idle_ns: u64 = 20 * std.time.ns_per_s,
        /// After `grace_ns`, abort if the body rate over the last `window_ns`
        /// is below this. Catches a trickle, which an idle timeout alone
        /// never fires on.
        min_bytes_per_s: u64 = 4 * 1024,
        /// Slow starts are normal; do not judge the rate before this.
        grace_ns: u64 = 10 * std.time.ns_per_s,
        /// The rate is measured over this trailing window, so a transfer
        /// that ran fast and then trickles is judged on the trickle. A
        /// lifetime average would let the fast start mask it for hours.
        /// 0 measures over the whole transfer. Capped at 60s.
        window_ns: u64 = 10 * std.time.ns_per_s,
    };

    pub const ResolvedConnection = struct {
        /// Checked address to dial. Currently IPv4 text, which std.Io.net.HostName accepts.
        dial_host: []const u8,
        /// Original URL host, used by std.http for TLS/SNI and connection identity.
        logical_host: []const u8,
    };

    pub const FetchResult = struct {
        status: std.http.Status,
        body: []u8,
        rate_limit: RateLimitHeaders = .{},
        headers: CapturedHeaders = .{},
        /// Where the response actually came from when redirects were
        /// followed; null when it came from the requested URL.
        final_url: ?[]u8 = null,

        pub fn deinit(self: *FetchResult, allocator: std.mem.Allocator) void {
            allocator.free(self.body);
            self.headers.deinit(allocator);
            if (self.final_url) |u| allocator.free(u);
        }
    };

    /// The subset of response headers a caller asked for, owned copies. When
    /// a header repeats, the first occurrence wins.
    pub const CapturedHeaders = struct {
        entries: []std.http.Header = &.{},

        pub fn fromResponseHead(
            allocator: std.mem.Allocator,
            head: std.http.Client.Response.Head,
            names: []const []const u8,
        ) !CapturedHeaders {
            if (names.len == 0) return .{};
            const entries = try allocator.alloc(std.http.Header, names.len);
            var count: usize = 0;
            errdefer {
                for (entries[0..count]) |e| allocator.free(e.value);
                allocator.free(entries);
            }
            var it = head.iterateHeaders();
            while (it.next()) |header| {
                for (names) |name| {
                    if (!std.ascii.eqlIgnoreCase(header.name, name)) continue;
                    if (hasName(entries[0..count], name)) break;
                    entries[count] = .{ .name = name, .value = try allocator.dupe(u8, header.value) };
                    count += 1;
                    break;
                }
            }
            if (count == 0) {
                allocator.free(entries);
                return .{};
            }
            return .{ .entries = try shrink(allocator, entries, count) };
        }

        fn shrink(allocator: std.mem.Allocator, entries: []std.http.Header, count: usize) ![]std.http.Header {
            if (allocator.resize(entries, count)) return entries[0..count];
            const exact = try allocator.dupe(std.http.Header, entries[0..count]);
            allocator.free(entries);
            return exact;
        }

        fn hasName(entries: []const std.http.Header, name: []const u8) bool {
            for (entries) |e| if (std.ascii.eqlIgnoreCase(e.name, name)) return true;
            return false;
        }

        pub fn get(self: CapturedHeaders, name: []const u8) ?[]const u8 {
            for (self.entries) |e| if (std.ascii.eqlIgnoreCase(e.name, name)) return e.value;
            return null;
        }

        pub fn deinit(self: *CapturedHeaders, allocator: std.mem.Allocator) void {
            for (self.entries) |e| allocator.free(e.value);
            if (self.entries.len > 0) allocator.free(self.entries);
            self.* = .{};
        }
    };

    pub const RateLimitHeaders = struct {
        limit: ?u64 = null,
        remaining: ?u64 = null,
        /// `RateLimit-Reset` as sent: delta seconds per the IETF draft, but
        /// some servers send an absolute Unix time. `delaySeconds` tells them
        /// apart. Signed values preserve pre-epoch timestamps for protocol
        /// policies (such as ATProto) that give Reset precedence over Retry-After.
        reset: ?i64 = null,
        /// `Retry-After` in its delta-seconds form.
        retry_after: ?u64 = null,
        /// `Retry-After` in its HTTP-date form, as Unix seconds.
        retry_after_at: ?i64 = null,

        pub fn fromResponseHead(head: std.http.Client.Response.Head) RateLimitHeaders {
            var result: RateLimitHeaders = .{};
            var it = head.iterateHeaders();
            while (it.next()) |header| {
                if (std.ascii.eqlIgnoreCase(header.name, "ratelimit-limit")) {
                    result.limit = parseHeaderInt(header.value);
                } else if (std.ascii.eqlIgnoreCase(header.name, "ratelimit-remaining")) {
                    result.remaining = parseHeaderInt(header.value);
                } else if (std.ascii.eqlIgnoreCase(header.name, "ratelimit-reset")) {
                    result.reset = std.fmt.parseInt(i64, std.mem.trim(u8, header.value, " \t"), 10) catch null;
                } else if (std.ascii.eqlIgnoreCase(header.name, "retry-after")) {
                    result.retry_after = parseHeaderInt(header.value);
                    if (result.retry_after == null) result.retry_after_at = parseHttpDate(header.value);
                }
            }
            return result;
        }

        pub fn isEmpty(self: RateLimitHeaders) bool {
            return self.limit == null and self.remaining == null and self.reset == null and
                self.retry_after == null and self.retry_after_at == null;
        }

        /// Seconds the server asked us to wait, given the current Unix time.
        /// `Retry-After` wins over `RateLimit-Reset`; a reset at or past
        /// `epoch_threshold` is read as an absolute time. Null when no
        /// header said anything usable; a hint already in the past is 0.
        pub fn delaySeconds(self: RateLimitHeaders, now_s: i64) ?u64 {
            if (self.retry_after) |secs| return secs;
            if (self.retry_after_at) |at| return secondsUntil(at, now_s);
            if (self.reset) |reset| {
                if (reset <= 0) return 0;
                if (reset >= epoch_threshold) return secondsUntil(@intCast(reset), now_s);
                return @intCast(reset);
            }
            return null;
        }

        /// A `RateLimit-Reset` this large is a Unix time (2001-09-09), not
        /// a wait: no server asks for a 31-year pause.
        pub const epoch_threshold: u64 = 1_000_000_000;

        fn secondsUntil(at: i64, now_s: i64) u64 {
            return if (at > now_s) @intCast(at - now_s) else 0;
        }
    };
};

/// HTTP-date: IMF-fixdate, obsolete RFC 850, or ANSI C asctime.
/// RFC 850 two-digit years use Go time.Parse’s 69/68 pivot. Returns Unix seconds.
fn parseHttpDate(value: []const u8) ?i64 {
    const input = std.mem.trim(u8, value, " \t");
    var normalized: [29]u8 = undefined;
    const v = normalizeHttpDate(input, &normalized) orelse return null;
    // Sun, 06 Nov 1994 08:49:37 GMT
    if (v.len != 29) return null;
    if (v[3] != ',' or v[4] != ' ' or v[7] != ' ' or v[11] != ' ' or v[16] != ' ' or v[19] != ':' or v[22] != ':' or v[25] != ' ') return null;
    if (!std.mem.eql(u8, v[26..29], "GMT")) return null;
    const day = std.fmt.parseInt(u32, v[5..7], 10) catch return null;
    const month = monthNumber(v[8..11]) orelse return null;
    const year = std.fmt.parseInt(i64, v[12..16], 10) catch return null;
    const hour = std.fmt.parseInt(u32, v[17..19], 10) catch return null;
    const minute = std.fmt.parseInt(u32, v[20..22], 10) catch return null;
    const second = std.fmt.parseInt(u32, v[23..25], 10) catch return null;
    const weekdays = [_][]const u8{ "Mon", "Tue", "Wed", "Thu", "Fri", "Sat", "Sun" };
    var valid_weekday = false;
    for (weekdays) |name| valid_weekday = valid_weekday or std.mem.eql(u8, v[0..3], name);
    if (!valid_weekday) return null;
    const leap = @mod(year, 4) == 0 and (@mod(year, 100) != 0 or @mod(year, 400) == 0);
    const days = [_]u8{ 31, if (leap) 29 else 28, 31, 30, 31, 30, 31, 31, 30, 31, 30, 31 };
    if (day == 0 or day > days[month - 1] or hour > 23 or minute > 59 or second > 59) return null;
    return daysFromCivil(year, month, day) * 86_400 + @as(i64, hour) * 3600 + @as(i64, minute) * 60 + second;
}

fn normalizeHttpDate(input: []const u8, buf: *[29]u8) ?[]const u8 {
    // Canonical dates require no copy.
    if (input.len == 29 and input[3] == ',') return input;
    if (std.mem.indexOf(u8, input, ", ")) |comma| {
        const names = [_][]const u8{ "Monday", "Tuesday", "Wednesday", "Thursday", "Friday", "Saturday", "Sunday" };
        var valid = false;
        for (names) |name| valid = valid or std.mem.eql(u8, input[0..comma], name);
        if (!valid) return null;
        const tail = input[comma + 2 ..];
        if (tail.len != 22 or tail[2] != '-' or tail[6] != '-' or tail[9] != ' ') return null;
        const short_year = std.fmt.parseInt(u8, tail[7..9], 10) catch return null;
        const year: u16 = @as(u16, short_year) + @as(u16, if (short_year >= 69) 1900 else 2000);
        return std.fmt.bufPrint(buf, "{s}, {s} {s} {d} {s}", .{ input[0..3], tail[0..2], tail[3..6], year, tail[10..] }) catch null;
    }
    if (input.len != 24 or input[3] != ' ' or input[7] != ' ' or input[10] != ' ' or input[19] != ' ') return null;
    const day = std.fmt.parseInt(u8, std.mem.trimStart(u8, input[8..10], " "), 10) catch return null;
    return std.fmt.bufPrint(buf, "{s}, {d:0>2} {s} {s} {s} GMT", .{ input[0..3], day, input[4..7], input[20..24], input[11..19] }) catch null;
}

fn monthNumber(name: []const u8) ?u32 {
    const names = [_][]const u8{ "Jan", "Feb", "Mar", "Apr", "May", "Jun", "Jul", "Aug", "Sep", "Oct", "Nov", "Dec" };
    for (names, 1..) |n, i| if (std.ascii.eqlIgnoreCase(n, name)) return @intCast(i);
    return null;
}

/// Days since 1970-01-01 for a proleptic Gregorian date (Howard Hinnant's
/// days_from_civil).
fn daysFromCivil(year: i64, month: u32, day: u32) i64 {
    const y: i64 = if (month <= 2) year - 1 else year;
    const era: i64 = @divFloor(y, 400);
    const yoe: i64 = y - era * 400;
    const m: i64 = month;
    const mp: i64 = @mod(m + 9, 12);
    const doy: i64 = @divFloor(153 * mp + 2, 5) + @as(i64, day) - 1;
    const doe: i64 = yoe * 365 + @divFloor(yoe, 4) - @divFloor(yoe, 100) + doy;
    return era * 146_097 + doe - 719_468;
}

/// std asserts on CR/LF in headers, which is a crash in safe builds and
/// injection in fast ones. Refuse up front instead.
fn validateHeaderValue(value: []const u8) error{InvalidHeader}!void {
    for (value) |c| if (c == '\r' or c == '\n') return error.InvalidHeader;
}

fn parseHeaderInt(value: []const u8) ?u64 {
    const trimmed = std.mem.trim(u8, value, " \t");
    return std.fmt.parseInt(u64, trimmed, 10) catch null;
}

var warned_unguarded: std.atomic.Value(bool) = .init(false);

/// A watcher only bounds work that runs beside it. An Io with no concurrency
/// to give leaves the request to run on the caller's own stack, where nothing
/// can interrupt it.
fn warnUnguarded() void {
    if (warned_unguarded.swap(true, .monotonic)) return;
    log.warn("the Io cannot run tasks concurrently; requests are running without deadline or stall guards", .{});
}

/// Body transfer counters, readable from another task while `stream` runs.
pub const Progress = struct {
    bytes: std.atomic.Value(usize) = .init(0),
    done: std.atomic.Value(bool) = .init(false),
};

fn streamBody(
    response: *std.http.Client.Response,
    writer: *std.Io.Writer,
    limit: ?usize,
    decompress_buffer: []u8,
    progress: *Progress,
) !void {
    defer progress.done.store(true, .release);

    var transfer_buffer: [8 * 1024]u8 = undefined;
    var decompress: std.http.Decompress = undefined;
    const reader = response.readerDecompressing(&transfer_buffer, &decompress, decompress_buffer);

    var chunk: [64 * 1024]u8 = undefined;
    while (true) {
        var sink = std.Io.Writer.fixed(&chunk);
        const n = reader.stream(&sink, .limited(chunk.len)) catch |err| switch (err) {
            error.EndOfStream => return,
            // Null when the read was canceled rather than failing on its own.
            error.ReadFailed => return response.bodyErr() orelse error.ReadFailed,
            else => |e| return e,
        };
        if (n == 0) continue;
        if (limit) |max| if (progress.bytes.load(.monotonic) + n > max) return error.ResponseTooLarge;
        try writer.writeAll(chunk[0..n]);
        _ = progress.bytes.fetchAdd(n, .release);
    }
}

/// A wall-clock check placed *after* a read cannot fire, because a peer that
/// simply stops sending leaves that read blocked in the kernel forever. So the
/// read runs as a task and the caller watches its byte counter, canceling when
/// progress dies.
fn streamResponseBody(
    io: std.Io,
    response: *std.http.Client.Response,
    writer: *std.Io.Writer,
    guard: Client.StallGuard,
    limit: ?usize,
    deadline_ns: u64,
    decompress_buffer: []u8,
    progress: *Progress,
) !void {
    if (guard.idle_ns == 0 and guard.min_bytes_per_s == 0 and deadline_ns == 0)
        return streamBody(response, writer, limit, decompress_buffer, progress);

    var future = io.concurrent(streamBody, .{ response, writer, limit, decompress_buffer, progress }) catch |err| switch (err) {
        error.ConcurrencyUnavailable => {
            warnUnguarded();
            return streamBody(response, writer, limit, decompress_buffer, progress);
        },
    };

    // Poll often so a body that lands at once is not charged a whole sample
    // period; sample the rate on a coarser grid so the window stays cheap.
    const tick_ns: u64 = 10 * std.time.ns_per_ms;
    const sample_ns: u64 = 100 * std.time.ns_per_ms;
    var window: RateWindow = .init(guard.window_ns, sample_ns);
    var idle_ns: u64 = 0;
    var elapsed_ns: u64 = 0;
    var since_sample_ns: u64 = 0;
    var last_bytes: usize = 0;

    while (!progress.done.load(.acquire)) {
        std.Io.Clock.Duration.sleep(
            .{ .raw = .fromNanoseconds(tick_ns), .clock = .awake },
            io,
        ) catch {
            // Cancelled from above: take the reader down with us rather
            // than awaiting a read that may never return.
            future.cancel(io) catch {};
            return error.Canceled;
        };
        elapsed_ns += tick_ns;
        since_sample_ns += tick_ns;

        if (deadline_ns != 0 and elapsed_ns >= deadline_ns) {
            future.cancel(io) catch {};
            return error.RequestDeadlineExceeded;
        }

        const seen = progress.bytes.load(.acquire);
        if (seen != last_bytes) {
            idle_ns = 0;
        } else {
            idle_ns += tick_ns;
        }
        if (guard.idle_ns != 0 and idle_ns >= guard.idle_ns) {
            future.cancel(io) catch {};
            return error.TransferStalled;
        }

        if (since_sample_ns >= sample_ns) {
            since_sample_ns = 0;
            window.push(seen - last_bytes);
            if (guard.min_bytes_per_s != 0 and elapsed_ns >= guard.grace_ns) {
                if (window.bytesPerSecond(seen, elapsed_ns)) |rate| {
                    if (rate < guard.min_bytes_per_s) {
                        future.cancel(io) catch {};
                        return error.TransferTooSlow;
                    }
                }
            }
        }
        last_bytes = seen;
    }

    return future.await(io);
}

/// Bytes moved per sample over a trailing window. With a zero window the
/// rate is the lifetime average.
const RateWindow = struct {
    samples: [max_samples]usize = @splat(0),
    len: usize,
    next: usize = 0,
    filled: usize = 0,
    sample_ns: u64,

    const max_samples = 600;

    fn init(window_ns: u64, sample_ns: u64) RateWindow {
        const wanted = window_ns / sample_ns;
        return .{ .len = @min(wanted, max_samples), .sample_ns = sample_ns };
    }

    fn push(self: *RateWindow, delta: usize) void {
        if (self.len == 0) return;
        self.samples[self.next] = delta;
        self.next = (self.next + 1) % self.len;
        if (self.filled < self.len) self.filled += 1;
    }

    /// Null until the window has enough samples to judge.
    fn bytesPerSecond(self: *const RateWindow, total: usize, elapsed_ns: u64) ?u64 {
        if (self.len == 0) {
            const secs = elapsed_ns / std.time.ns_per_s;
            if (secs == 0) return null;
            return total / secs;
        }
        if (self.filled < self.len) return null;
        var sum: usize = 0;
        for (self.samples[0..self.len]) |n| sum += n;
        const window_ns = self.sample_ns * self.len;
        return @intCast((@as(u128, sum) * std.time.ns_per_s) / window_ns);
    }
};

/// The peer hung up without answering. Distinct from a mid-body failure: no
/// part of a response was seen, so the request provably did not take effect.
fn isDeadConnection(err: anyerror) bool {
    return switch (err) {
        error.ConnectionResetByPeer,
        error.BrokenPipe,
        error.EndOfStream,
        error.UnexpectedWriteFailure,
        error.HttpConnectionClosing,
        => true,
        else => false,
    };
}

fn isReplayable(options: Client.FetchOptions) bool {
    if (options.payload != null) return false;
    return switch (options.method) {
        .GET, .HEAD, .OPTIONS => true,
        else => false,
    };
}

/// A zero-length buffer works only while every response is `identity`; gzip
/// needs a real window. Sizing from the negotiated encoding keeps the identity
/// path allocation-free in practice (0 bytes) without capping compressed ones.
fn decompressBufferLen(encoding: std.http.ContentEncoding) usize {
    return encoding.minBufferCapacity();
}

/// Scheme, host, and effective port all equal. A redirect that changes any
/// of them is a different party and must not inherit credentials.
fn sameOrigin(a: std.Uri, b: std.Uri) !bool {
    if (!std.ascii.eqlIgnoreCase(a.scheme, b.scheme)) return false;
    var a_buf: [std.Io.net.HostName.max_len]u8 = undefined;
    var b_buf: [std.Io.net.HostName.max_len]u8 = undefined;
    const a_host = try std.Io.net.HostName.fromUri(a, &a_buf);
    const b_host = try std.Io.net.HostName.fromUri(b, &b_buf);
    if (!std.ascii.eqlIgnoreCase(a_host.bytes, b_host.bytes)) return false;
    const a_proto = std.http.Client.Protocol.fromUri(a) orelse return error.UnsupportedUriScheme;
    const b_proto = std.http.Client.Protocol.fromUri(b) orelse return error.UnsupportedUriScheme;
    return (a.port orelse defaultPort(a_proto)) == (b.port orelse defaultPort(b_proto));
}

fn defaultPort(protocol: std.http.Client.Protocol) u16 {
    return switch (protocol) {
        .plain => 80,
        .tls => 443,
    };
}

// === tests ===

test "client init/deinit" {
    const io = std.Options.debug_io;
    var transport = Client.init(io, std.testing.allocator);
    defer transport.deinit();
}

test "resolved transport rejects mismatched logical host" {
    const io = std.Options.debug_io;
    var transport = Client.init(io, std.testing.allocator);
    defer transport.deinit();

    // debug_io has no concurrency to give, so this request runs unguarded.
    const log_level = std.testing.log_level;
    std.testing.log_level = .err;
    defer std.testing.log_level = log_level;

    try std.testing.expectError(error.ResolvedHostMismatch, transport.fetch(.{
        .url = "http://example.com/",
        .resolved_connection = .{
            .dial_host = "127.0.0.1",
            .logical_host = "other.example",
        },
    }));
}

test "client parses rate limit headers" {
    const response_bytes = "HTTP/1.1 429 Too Many Requests\r\n" ++
        "RateLimit-Limit: 3000\r\n" ++
        "ratelimit-remaining: 0\r\n" ++
        "RateLimit-Reset: 1710000000\r\n" ++
        "Retry-After: 2\r\n\r\n";

    const head = try std.http.Client.Response.Head.parse(response_bytes);
    const headers = Client.RateLimitHeaders.fromResponseHead(head);

    try std.testing.expectEqual(@as(?u64, 3000), headers.limit);
    try std.testing.expectEqual(@as(?u64, 0), headers.remaining);
    try std.testing.expectEqual(@as(?i64, 1710000000), headers.reset);
    try std.testing.expectEqual(@as(?u64, 2), headers.retry_after);
    try std.testing.expect(!headers.isEmpty());
}

test "captured headers are the ones asked for, case-insensitively, first occurrence wins" {
    const response_bytes = "HTTP/1.1 401 Unauthorized\r\n" ++
        "Content-Type: application/json; charset=utf-8\r\n" ++
        "X-Nonce: first\r\n" ++
        "x-nonce: second\r\n" ++
        "WWW-Authenticate: Bearer realm=\"x\"\r\n\r\n";

    const head = try std.http.Client.Response.Head.parse(response_bytes);
    var headers = try Client.CapturedHeaders.fromResponseHead(
        std.testing.allocator,
        head,
        &.{ "content-type", "X-NONCE", "x-missing" },
    );
    defer headers.deinit(std.testing.allocator);

    try std.testing.expectEqualStrings("application/json; charset=utf-8", headers.get("Content-Type").?);
    try std.testing.expectEqualStrings("first", headers.get("x-nonce").?);
    try std.testing.expectEqual(@as(?[]const u8, null), headers.get("www-authenticate"));
    try std.testing.expectEqual(@as(?[]const u8, null), headers.get("x-missing"));
    try std.testing.expectEqual(@as(usize, 2), headers.entries.len);
}

test "asking for no headers captures nothing and owns nothing" {
    const head = try std.http.Client.Response.Head.parse("HTTP/1.1 200 OK\r\nX-A: 1\r\n\r\n");
    var headers = try Client.CapturedHeaders.fromResponseHead(std.testing.allocator, head, &.{});
    defer headers.deinit(std.testing.allocator);
    try std.testing.expectEqual(@as(usize, 0), headers.entries.len);
}

test "requests identify the client, and applications can name themselves" {
    // A server operator seeing load should be able to tell what is calling
    // and where to ask about it. std.http's default says neither.
    const testing = std.testing;
    var threaded: std.Io.Threaded = .init(testing.allocator, .{});
    defer threaded.deinit();
    const io = threaded.io();

    var default_transport = Client.init(io, testing.allocator);
    defer default_transport.deinit();
    try testing.expectEqualStrings(default_user_agent, default_transport.user_agent);
    try testing.expect(std.mem.indexOf(u8, default_transport.user_agent, "http") != null);
    try testing.expect(std.mem.startsWith(u8, default_transport.user_agent, "zttp/" ++ build_options.version ++ " "));

    var named = Client.initWithUserAgent(io, testing.allocator, "stream/1.2.3 (+https://example.invalid)");
    defer named.deinit();
    try testing.expectEqualStrings("stream/1.2.3 (+https://example.invalid)", named.user_agent);

    // An empty override must not silently erase the identity.
    var empty = Client.initWithUserAgent(io, testing.allocator, "");
    defer empty.deinit();
    try testing.expectEqualStrings(default_user_agent, empty.user_agent);
}

test "extra headers past the buffer are an error, not silently dropped" {
    const testing = std.testing;
    var threaded: std.Io.Threaded = .init(testing.allocator, .{});
    defer threaded.deinit();

    var transport = Client.init(threaded.io(), testing.allocator);
    defer transport.deinit();

    const header: std.http.Header = .{ .name = "x-zttp-test", .value = "1" };
    const too_many: [8]std.http.Header = @splat(header);
    try testing.expectError(error.TooManyHeaders, transport.fetch(.{
        .url = "http://127.0.0.1:1/never-reached",
        .accept = "application/json",
        .extra_headers = &too_many,
    }));
}

// A loopback origin for transport tests. Each canned response is served on its
// own accepted connection; `serve` runs on a thread so the client can drive the
// exchange from the test body.
const TestOrigin = struct {
    listener: std.Io.net.Server,
    io: std.Io,

    fn init(io: std.Io) !TestOrigin {
        return .{
            .listener = try (std.Io.net.IpAddress{ .ip4 = .loopback(0) }).listen(io, .{ .reuse_address = true }),
            .io = io,
        };
    }
    fn deinit(self: *TestOrigin) void {
        self.listener.deinit(self.io);
    }
    fn port(self: *const TestOrigin) u16 {
        return self.listener.socket.address.getPort();
    }

    /// Accept one connection, read the request head, write `response` verbatim.
    /// Answers with headers promising a long body, dribbles a few bytes, then
    /// goes quiet — the shape of a slow origin that pins a worker.
    fn serveStalling(self: *TestOrigin) !void {
        var conn = try self.listener.accept(self.io);
        defer conn.close(self.io);
        var rbuf: [4096]u8 = undefined;
        var reader = conn.reader(self.io, &rbuf);
        var scratch: [2048]u8 = undefined;
        var slices = [_][]u8{&scratch};
        _ = reader.interface.readVec(&slices) catch {};
        var wbuf: [4096]u8 = undefined;
        var writer = conn.writer(self.io, &wbuf);
        try writer.interface.writeAll(
            "HTTP/1.1 200 OK\r\nContent-Length: 1048576\r\n\r\nhello",
        );
        try writer.interface.flush();
        // Send nothing further and block until the client hangs up. A guarded
        // client returns quickly; an unguarded one waits here forever.
        var sink: [1]u8 = undefined;
        var tail = [_][]u8{&sink};
        _ = reader.interface.readVec(&tail) catch {};
    }

    /// Accepts, reads the request, then never answers: connection
    /// established, response head never arrives, nothing for StallGuard to
    /// measure.
    fn serveSilent(self: *TestOrigin) !void {
        var conn = try self.listener.accept(self.io);
        defer conn.close(self.io);
        var rbuf: [4096]u8 = undefined;
        var reader = conn.reader(self.io, &rbuf);
        var scratch: [2048]u8 = undefined;
        var slices = [_][]u8{&scratch};
        _ = reader.interface.readVec(&slices) catch {};
        // Say nothing and block until the client hangs up.
        var sink: [1]u8 = undefined;
        var tail = [_][]u8{&sink};
        _ = reader.interface.readVec(&tail) catch {};
    }

    /// Accepts, reads the request, optionally writes `preamble`, then holds
    /// the socket open and silent until `release` is set. Gives up on its own
    /// after a few seconds so a client that never lets go fails its test
    /// instead of hanging the suite.
    fn serveHeldOpen(
        self: *TestOrigin,
        preamble: []const u8,
        got_request: *std.atomic.Value(bool),
        release: *std.atomic.Value(bool),
    ) !void {
        var conn = try self.listener.accept(self.io);
        defer conn.close(self.io);
        var rbuf: [4096]u8 = undefined;
        var reader = conn.reader(self.io, &rbuf);
        var scratch: [2048]u8 = undefined;
        var slices = [_][]u8{&scratch};
        _ = reader.interface.readVec(&slices) catch {};
        if (preamble.len > 0) {
            var wbuf: [4096]u8 = undefined;
            var writer = conn.writer(self.io, &wbuf);
            try writer.interface.writeAll(preamble);
            try writer.interface.flush();
        }
        got_request.store(true, .release);
        var waited_ms: u64 = 0;
        while (!release.load(.acquire) and waited_ms < 3000) : (waited_ms += 10)
            try self.io.sleep(.fromMilliseconds(10), .awake);
    }

    /// Accepts, reads the request, then hangs up without answering — what a
    /// pooled connection reaped by the origin looks like on next reuse.
    fn serveDeadThenOk(self: *TestOrigin, response: []const u8) !void {
        {
            var dead = try self.listener.accept(self.io);
            var rbuf: [4096]u8 = undefined;
            var reader = dead.reader(self.io, &rbuf);
            var scratch: [2048]u8 = undefined;
            var slices = [_][]u8{&scratch};
            _ = reader.interface.readVec(&slices) catch {};
            dead.close(self.io);
        }
        try self.serveOnce(response);
    }

    /// One keep-alive connection: answers the first request, reads the second
    /// and resets the connection instead of answering. The client's read then
    /// fails with a reset rather than a clean end of stream. A fresh connection
    /// is served normally.
    fn serveOkThenReset(self: *TestOrigin, response: []const u8) !void {
        {
            var conn = try self.listener.accept(self.io);
            defer conn.close(self.io);
            var rbuf: [4096]u8 = undefined;
            var reader = conn.reader(self.io, &rbuf);
            var scratch: [2048]u8 = undefined;
            var slices = [_][]u8{&scratch};
            _ = reader.interface.readVec(&slices) catch {};
            var wbuf: [4096]u8 = undefined;
            var writer = conn.writer(self.io, &wbuf);
            try writer.interface.writeAll(response);
            try writer.interface.flush();
            var again = [_][]u8{&scratch};
            _ = reader.interface.readVec(&again) catch {};
            const reset: std.posix.linger = .{ .onoff = 1, .linger = 0 };
            try std.posix.setsockopt(conn.socket.handle, std.posix.SOL.SOCKET, std.posix.SO.LINGER, std.mem.asBytes(&reset));
        }
        try self.serveOnce(response);
    }

    /// One keep-alive connection: answers the first request, then reads the
    /// second and goes silent with the socket held open — the shape of a
    /// pooled connection a middlebox has blackholed while it sat idle (no FIN
    /// or RST ever reaches the client). After the client abandons that socket,
    /// accepts a fresh connection and serves normally.
    fn serveOkThenBlackhole(
        self: *TestOrigin,
        response: []const u8,
        reused: *std.atomic.Value(bool),
    ) !void {
        {
            var conn = try self.listener.accept(self.io);
            defer conn.close(self.io);
            var rbuf: [4096]u8 = undefined;
            var reader = conn.reader(self.io, &rbuf);
            var scratch: [2048]u8 = undefined;
            var slices = [_][]u8{&scratch};
            _ = reader.interface.readVec(&slices) catch {};
            var wbuf: [4096]u8 = undefined;
            var writer = conn.writer(self.io, &wbuf);
            try writer.interface.writeAll(response);
            try writer.interface.flush();
            // Second request arrives on the same pooled connection; read it
            // and say nothing, holding the socket open until the client
            // gives up and hangs up.
            var again = [_][]u8{&scratch};
            const n = reader.interface.readVec(&again) catch 0;
            if (n > 0) reused.store(true, .release);
            var sink: [1]u8 = undefined;
            var tail = [_][]u8{&sink};
            _ = reader.interface.readVec(&tail) catch {};
        }
        try self.serveOnce(response);
    }

    /// Accept one connection, remember whether the request carried an
    /// `authorization` header, and answer `response`.
    fn serveRecordingAuth(self: *TestOrigin, response: []const u8, saw_auth: *std.atomic.Value(bool)) !void {
        var conn = try self.listener.accept(self.io);
        defer conn.close(self.io);
        var rbuf: [4096]u8 = undefined;
        var reader = conn.reader(self.io, &rbuf);
        var scratch: [2048]u8 = undefined;
        var slices = [_][]u8{&scratch};
        const n = reader.interface.readVec(&slices) catch 0;
        if (std.ascii.findIgnoreCase(scratch[0..n], "\r\nauthorization:") != null) saw_auth.store(true, .release);
        var wbuf: [4096]u8 = undefined;
        var writer = conn.writer(self.io, &wbuf);
        try writer.interface.writeAll(response);
        try writer.interface.flush();
    }

    /// Accept one connection, record the request line and whether a body
    /// followed the head, and answer `response`.
    fn serveRecordingRequest(self: *TestOrigin, response: []const u8, request_line: *[128]u8, request_line_len: *usize, saw_body: *std.atomic.Value(bool)) !void {
        var conn = try self.listener.accept(self.io);
        defer conn.close(self.io);
        var rbuf: [4096]u8 = undefined;
        var reader = conn.reader(self.io, &rbuf);
        var scratch: [4096]u8 = undefined;
        var slices = [_][]u8{&scratch};
        const n = reader.interface.readVec(&slices) catch 0;
        const seen = scratch[0..n];
        const eol = std.mem.indexOf(u8, seen, "\r\n") orelse seen.len;
        const line_len = @min(eol, request_line.len);
        @memcpy(request_line[0..line_len], seen[0..line_len]);
        request_line_len.* = line_len;
        if (std.mem.indexOf(u8, seen, "\r\n\r\n")) |head_end| {
            if (head_end + 4 < seen.len) saw_body.store(true, .release);
        }
        var wbuf: [4096]u8 = undefined;
        var writer = conn.writer(self.io, &wbuf);
        try writer.interface.writeAll(response);
        try writer.interface.flush();
    }

    /// Answers with a long Content-Length, sends `burst` bytes at once, then
    /// goes quiet: a transfer that ran fast and is now trickling nothing.
    fn serveBurstThenStall(self: *TestOrigin, burst: usize) !void {
        var conn = try self.listener.accept(self.io);
        defer conn.close(self.io);
        var rbuf: [4096]u8 = undefined;
        var reader = conn.reader(self.io, &rbuf);
        var scratch: [2048]u8 = undefined;
        var slices = [_][]u8{&scratch};
        _ = reader.interface.readVec(&slices) catch {};
        var wbuf: [64 * 1024]u8 = undefined;
        var writer = conn.writer(self.io, &wbuf);
        try writer.interface.print("HTTP/1.1 200 OK\r\nContent-Length: {d}\r\n\r\n", .{burst * 4});
        const chunk: [4096]u8 = @splat('x');
        var sent: usize = 0;
        while (sent < burst) : (sent += chunk.len) try writer.interface.writeAll(&chunk);
        try writer.interface.flush();
        var sink: [1]u8 = undefined;
        var tail = [_][]u8{&sink};
        _ = reader.interface.readVec(&tail) catch {};
    }

    /// Accept one connection, copy the request head into `head`, answer.
    fn serveRecordingHead(self: *TestOrigin, response: []const u8, head: *[4096]u8, head_len: *usize) !void {
        var conn = try self.listener.accept(self.io);
        defer conn.close(self.io);
        var rbuf: [4096]u8 = undefined;
        var reader = conn.reader(self.io, &rbuf);
        var slices = [_][]u8{head};
        head_len.* = reader.interface.readVec(&slices) catch 0;
        var wbuf: [4096]u8 = undefined;
        var writer = conn.writer(self.io, &wbuf);
        try writer.interface.writeAll(response);
        try writer.interface.flush();
    }

    fn serveOnce(self: *TestOrigin, response: []const u8) !void {
        var conn = try self.listener.accept(self.io);
        defer conn.close(self.io);
        var rbuf: [4096]u8 = undefined;
        var reader = conn.reader(self.io, &rbuf);
        // Read whatever the client sent; the request content is irrelevant here.
        var scratch: [2048]u8 = undefined;
        var slices = [_][]u8{&scratch};
        _ = reader.interface.readVec(&slices) catch {};
        var wbuf: [4096]u8 = undefined;
        var writer = conn.writer(self.io, &wbuf);
        try writer.interface.writeAll(response);
        try writer.interface.flush();
    }
};

test "gzip responses decode: the identity override was covering an empty decompress buffer" {
    const testing = std.testing;
    var threaded: std.Io.Threaded = .init(testing.allocator, .{});
    defer threaded.deinit();
    const io = threaded.io();

    var origin = try TestOrigin.init(io);
    defer origin.deinit();

    // gzip of "hello zttp" produced at comptime would need a compressor; instead
    // assert the *identity* path still works through readerDecompressing with a
    // real buffer, which is the change under test. A gzip body is exercised by
    // the live network suite.
    const body = "hello zttp";
    var resp_buf: [256]u8 = undefined;
    const response = try std.fmt.bufPrint(&resp_buf, "HTTP/1.1 200 OK\r\nContent-Length: {d}\r\nConnection: close\r\n\r\n{s}", .{ body.len, body });

    const T = struct {
        fn run(o: *TestOrigin, r: []const u8) void {
            o.serveOnce(r) catch {};
        }
    };
    var th = try std.Thread.spawn(.{}, T.run, .{ &origin, response });
    defer th.join();

    var transport = Client.init(io, testing.allocator);
    defer transport.deinit();
    transport.keep_alive = false;

    var url_buf: [64]u8 = undefined;
    const url = try std.fmt.bufPrint(&url_buf, "http://127.0.0.1:{d}/", .{origin.port()});

    var res = try transport.fetch(.{ .url = url });
    defer res.deinit(testing.allocator);
    try testing.expectEqual(std.http.Status.ok, res.status);
    try testing.expectEqualStrings(body, res.body);
}

test "a stalled transfer is abandoned instead of pinning the caller" {
    const allocator = std.testing.allocator;
    var threaded: std.Io.Threaded = .init(allocator, .{});
    defer threaded.deinit();
    const io = threaded.io();

    var origin = try TestOrigin.init(io);
    defer origin.deinit();
    const url = try std.fmt.allocPrint(allocator, "http://127.0.0.1:{d}/stall", .{origin.port()});
    defer allocator.free(url);

    var server = try std.Thread.spawn(.{}, struct {
        fn run(o: *TestOrigin) void {
            o.serveStalling() catch {};
        }
    }.run, .{&origin});
    defer server.join();

    var transport = Client.init(io, allocator);
    defer transport.deinit();
    transport.stall = .{
        .idle_ns = 400 * std.time.ns_per_ms,
        .min_bytes_per_s = 0,
        .grace_ns = 0,
    };

    // Unguarded, this call never returns: the origin holds the connection open
    // and simply stops sending.
    try std.testing.expectError(error.TransferStalled, transport.fetch(.{ .url = url }));
}

test "a gzip response decodes: the identity override was covering an empty decompress buffer" {
    const allocator = std.testing.allocator;
    var threaded: std.Io.Threaded = .init(allocator, .{});
    defer threaded.deinit();
    const io = threaded.io();

    const repeated: [512][4]u8 = @splat("zttp".*);
    const payload = std.mem.asBytes(&repeated);
    var gz_buf: [64 * 1024]u8 = undefined;
    var gz = std.Io.Writer.fixed(&gz_buf);
    {
        var flate_buf: [std.compress.flate.max_window_len]u8 = undefined;
        var comp: std.compress.flate.Compress = try .init(&gz, &flate_buf, .gzip, .default);
        try comp.writer.writeAll(payload);
        try comp.finish();
    }

    var origin = try TestOrigin.init(io);
    defer origin.deinit();
    const url = try std.fmt.allocPrint(allocator, "http://127.0.0.1:{d}/gz", .{origin.port()});
    defer allocator.free(url);

    const response = try std.fmt.allocPrint(
        allocator,
        "HTTP/1.1 200 OK\r\nContent-Encoding: gzip\r\nContent-Length: {d}\r\n\r\n{s}",
        .{ gz.buffered().len, gz.buffered() },
    );
    defer allocator.free(response);

    var server = try std.Thread.spawn(.{}, struct {
        fn run(o: *TestOrigin, r: []const u8) void {
            o.serveOnce(r) catch {};
        }
    }.run, .{ &origin, response });
    defer server.join();

    var transport = Client.init(io, allocator);
    defer transport.deinit();
    var res = transport.fetch(.{ .url = url }) catch |err| {
        std.debug.print("fetch failed: {t}\n", .{err});
        return err;
    };
    defer res.deinit(allocator);

    try std.testing.expectEqualStrings(payload, res.body);
}

test "a peer that accepts and never answers is abandoned at the head deadline" {
    const allocator = std.testing.allocator;
    var threaded: std.Io.Threaded = .init(allocator, .{});
    defer threaded.deinit();
    const io = threaded.io();

    var origin = try TestOrigin.init(io);
    defer origin.deinit();
    const url = try std.fmt.allocPrint(allocator, "http://127.0.0.1:{d}/silent", .{origin.port()});
    defer allocator.free(url);

    var server = try std.Thread.spawn(.{}, struct {
        fn run(o: *TestOrigin) void {
            o.serveSilent() catch {};
        }
    }.run, .{&origin});
    defer server.join();

    var transport = Client.init(io, allocator);
    defer transport.deinit();
    transport.head_deadline_ns = 300 * std.time.ns_per_ms;

    // Unguarded, this call sits in receiveHead until the peer or kernel gives
    // up — for a wedged keep-alive socket, never.
    try std.testing.expectError(error.ResponseHeadTimeout, transport.fetch(.{ .url = url }));
}

test "canceling a fetch parked on a silent peer returns error.Canceled promptly" {
    const allocator = std.testing.allocator;
    var threaded: std.Io.Threaded = .init(allocator, .{});
    defer threaded.deinit();
    const io = threaded.io();

    var origin = try TestOrigin.init(io);
    defer origin.deinit();
    const url = try std.fmt.allocPrint(allocator, "http://127.0.0.1:{d}/silent", .{origin.port()});
    defer allocator.free(url);

    var got_request: std.atomic.Value(bool) = .init(false);
    var release: std.atomic.Value(bool) = .init(false);
    var server = try std.Thread.spawn(.{}, struct {
        fn run(o: *TestOrigin, got: *std.atomic.Value(bool), rel: *std.atomic.Value(bool)) void {
            o.serveHeldOpen("", got, rel) catch {};
        }
    }.run, .{ &origin, &got_request, &release });
    defer server.join();
    defer release.store(true, .release);

    var transport = Client.init(io, allocator);
    defer transport.deinit();
    transport.head_deadline_ns = 30 * std.time.ns_per_s;
    transport.dead_connection_attempts = 1;

    var fetching = try io.concurrent(struct {
        fn run(c: *Client, u: []const u8) !void {
            var res = try c.fetch(.{ .url = u });
            res.deinit(c.allocator);
        }
    }.run, .{ &transport, url });
    var waited_ms: u64 = 0;
    while (!got_request.load(.acquire) and waited_ms < 3000) : (waited_ms += 5)
        io.sleep(.fromMilliseconds(5), .awake) catch {};
    io.sleep(.fromMilliseconds(50), .awake) catch {};

    const started = std.Io.Timestamp.now(io, .awake).nanoseconds;
    try std.testing.expectError(error.Canceled, fetching.cancel(io));
    const elapsed_ns = std.Io.Timestamp.now(io, .awake).nanoseconds - started;
    try std.testing.expect(elapsed_ns < std.time.ns_per_s);
}

test "the head deadline fires when the Io would run an async task inline" {
    const allocator = std.testing.allocator;
    var threaded: std.Io.Threaded = .init(allocator, .{ .async_limit = .nothing });
    defer threaded.deinit();
    const io = threaded.io();

    var origin = try TestOrigin.init(io);
    defer origin.deinit();
    const url = try std.fmt.allocPrint(allocator, "http://127.0.0.1:{d}/silent", .{origin.port()});
    defer allocator.free(url);

    var got_request: std.atomic.Value(bool) = .init(false);
    var release: std.atomic.Value(bool) = .init(false);
    var server = try std.Thread.spawn(.{}, struct {
        fn run(o: *TestOrigin, got: *std.atomic.Value(bool), rel: *std.atomic.Value(bool)) void {
            o.serveHeldOpen("", got, rel) catch {};
        }
    }.run, .{ &origin, &got_request, &release });
    defer server.join();
    defer release.store(true, .release);

    var transport = Client.init(io, allocator);
    defer transport.deinit();
    transport.dead_connection_attempts = 1;
    transport.head_deadline_ns = 300 * std.time.ns_per_ms;

    const started = std.Io.Timestamp.now(io, .awake).nanoseconds;
    try std.testing.expectError(error.ResponseHeadTimeout, transport.fetch(.{ .url = url }));
    const elapsed_ns = std.Io.Timestamp.now(io, .awake).nanoseconds - started;
    try std.testing.expect(elapsed_ns < 2 * std.time.ns_per_s);
}

test "the stall guard fires when the Io would run an async task inline" {
    const allocator = std.testing.allocator;
    var threaded: std.Io.Threaded = .init(allocator, .{ .async_limit = .nothing });
    defer threaded.deinit();
    const io = threaded.io();

    var origin = try TestOrigin.init(io);
    defer origin.deinit();
    const url = try std.fmt.allocPrint(allocator, "http://127.0.0.1:{d}/stall", .{origin.port()});
    defer allocator.free(url);

    var got_request: std.atomic.Value(bool) = .init(false);
    var release: std.atomic.Value(bool) = .init(false);
    var server = try std.Thread.spawn(.{}, struct {
        fn run(o: *TestOrigin, got: *std.atomic.Value(bool), rel: *std.atomic.Value(bool)) void {
            o.serveHeldOpen("HTTP/1.1 200 OK\r\nContent-Length: 1048576\r\n\r\nhello", got, rel) catch {};
        }
    }.run, .{ &origin, &got_request, &release });
    defer server.join();
    defer release.store(true, .release);

    var transport = Client.init(io, allocator);
    defer transport.deinit();
    transport.dead_connection_attempts = 1;
    transport.stall = .{ .idle_ns = 300 * std.time.ns_per_ms, .min_bytes_per_s = 0, .grace_ns = 0 };

    const started = std.Io.Timestamp.now(io, .awake).nanoseconds;
    try std.testing.expectError(error.TransferStalled, transport.fetch(.{ .url = url }));
    const elapsed_ns = std.Io.Timestamp.now(io, .awake).nanoseconds - started;
    try std.testing.expect(elapsed_ns < 2 * std.time.ns_per_s);
}

test "without concurrency a request still completes, unguarded" {
    const allocator = std.testing.allocator;
    var threaded: std.Io.Threaded = .init(allocator, .{ .async_limit = .nothing, .concurrent_limit = .nothing });
    defer threaded.deinit();
    const io = threaded.io();

    var origin = try TestOrigin.init(io);
    defer origin.deinit();
    const url = try std.fmt.allocPrint(allocator, "http://127.0.0.1:{d}/x", .{origin.port()});
    defer allocator.free(url);

    var server = try std.Thread.spawn(.{}, struct {
        fn run(o: *TestOrigin) void {
            o.serveOnce("HTTP/1.1 200 OK\r\nContent-Length: 2\r\n\r\nok") catch {};
        }
    }.run, .{&origin});
    defer server.join();

    var transport = Client.init(io, allocator);
    defer transport.deinit();

    const log_level = std.testing.log_level;
    std.testing.log_level = .err;
    defer std.testing.log_level = log_level;

    var res = try transport.fetch(.{ .url = url, .deadline_ns = 5 * std.time.ns_per_s });
    defer res.deinit(allocator);
    try std.testing.expectEqual(std.http.Status.ok, res.status);
    try std.testing.expectEqualStrings("ok", res.body);
}

test "a connection closed before it answers is replayed, not surfaced" {
    const allocator = std.testing.allocator;
    var threaded: std.Io.Threaded = .init(allocator, .{});
    defer threaded.deinit();
    const io = threaded.io();

    var origin = try TestOrigin.init(io);
    defer origin.deinit();
    const url = try std.fmt.allocPrint(allocator, "http://127.0.0.1:{d}/x", .{origin.port()});
    defer allocator.free(url);

    var server = try std.Thread.spawn(.{}, struct {
        fn run(o: *TestOrigin) void {
            o.serveDeadThenOk("HTTP/1.1 200 OK\r\nContent-Length: 2\r\n\r\nok") catch {};
        }
    }.run, .{&origin});
    defer server.join();

    var transport = Client.init(io, allocator);
    defer transport.deinit();

    var res = try transport.fetch(.{ .url = url });
    defer res.deinit(allocator);
    try std.testing.expectEqual(std.http.Status.ok, res.status);
    try std.testing.expectEqualStrings("ok", res.body);
}

test "a pooled connection reset instead of answered is replayed, not surfaced" {
    const allocator = std.testing.allocator;
    var threaded: std.Io.Threaded = .init(allocator, .{});
    defer threaded.deinit();
    const io = threaded.io();

    var origin = try TestOrigin.init(io);
    defer origin.deinit();
    const url = try std.fmt.allocPrint(allocator, "http://127.0.0.1:{d}/x", .{origin.port()});
    defer allocator.free(url);

    var server = try std.Thread.spawn(.{}, struct {
        fn run(o: *TestOrigin) void {
            o.serveOkThenReset("HTTP/1.1 200 OK\r\nContent-Length: 2\r\n\r\nok") catch {};
        }
    }.run, .{&origin});
    defer server.join();

    var transport = Client.init(io, allocator);
    defer transport.deinit();

    var first = try transport.fetch(.{ .url = url });
    defer first.deinit(allocator);
    try std.testing.expectEqualStrings("ok", first.body);

    var second = try transport.fetch(.{ .url = url });
    defer second.deinit(allocator);
    try std.testing.expectEqualStrings("ok", second.body);
}

test "a pooled connection silently blackholed while idle costs one head-deadline stall, then self-clears" {
    const allocator = std.testing.allocator;
    var threaded: std.Io.Threaded = .init(allocator, .{});
    defer threaded.deinit();
    const io = threaded.io();

    var origin = try TestOrigin.init(io);
    defer origin.deinit();
    const url = try std.fmt.allocPrint(allocator, "http://127.0.0.1:{d}/x", .{origin.port()});
    defer allocator.free(url);

    var reused: std.atomic.Value(bool) = .init(false);
    var server = try std.Thread.spawn(.{}, struct {
        fn run(o: *TestOrigin, flag: *std.atomic.Value(bool)) void {
            o.serveOkThenBlackhole("HTTP/1.1 200 OK\r\nContent-Length: 2\r\n\r\nok", flag) catch {};
        }
    }.run, .{ &origin, &reused });
    defer server.join();

    var transport = Client.init(io, allocator);
    defer transport.deinit();
    transport.head_deadline_ns = 300 * std.time.ns_per_ms;

    var first = try transport.fetch(.{ .url = url });
    defer first.deinit(allocator);
    try std.testing.expectEqual(std.http.Status.ok, first.status);

    // The second fetch reuses the pooled keep-alive connection. The origin
    // reads the request and never answers — no error is ever delivered, so
    // only the head deadline catches it, and nothing replays it: this is the
    // caller-visible failure (what a proxy in front of it would surface as a
    // 502 at exactly the deadline).
    try std.testing.expectError(error.ResponseHeadTimeout, transport.fetch(.{ .url = url }));
    try std.testing.expect(reused.load(.acquire));

    // The abandoned connection must not be re-pooled: the next fetch dials
    // fresh and succeeds, so each stale socket costs exactly one stall.
    var third = try transport.fetch(.{ .url = url });
    defer third.deinit(allocator);
    try std.testing.expectEqual(std.http.Status.ok, third.status);
    try std.testing.expectEqualStrings("ok", third.body);
}

test "authorization does not follow a redirect to another host" {
    const allocator = std.testing.allocator;
    var threaded: std.Io.Threaded = .init(allocator, .{});
    defer threaded.deinit();
    const io = threaded.io();

    var first = try TestOrigin.init(io);
    defer first.deinit();
    var second = try TestOrigin.init(io);
    defer second.deinit();

    // Both listen on loopback, but `localhost` and `127.0.0.1` are different
    // hosts as far as a redirect is concerned, which is the cross-origin case.
    const redirect = try std.fmt.allocPrint(
        allocator,
        "HTTP/1.1 302 Found\r\nLocation: http://127.0.0.1:{d}/elsewhere\r\nContent-Length: 0\r\n\r\n",
        .{second.port()},
    );
    defer allocator.free(redirect);
    const url = try std.fmt.allocPrint(allocator, "http://localhost:{d}/start", .{first.port()});
    defer allocator.free(url);

    var first_saw_auth: std.atomic.Value(bool) = .init(false);
    var second_saw_auth: std.atomic.Value(bool) = .init(false);
    const Server = struct {
        fn run(o: *TestOrigin, response: []const u8, flag: *std.atomic.Value(bool)) void {
            o.serveRecordingAuth(response, flag) catch {};
        }
    };
    var t1 = try std.Thread.spawn(.{}, Server.run, .{ &first, redirect, &first_saw_auth });
    defer t1.join();
    var t2 = try std.Thread.spawn(.{}, Server.run, .{ &second, "HTTP/1.1 200 OK\r\nContent-Length: 2\r\n\r\nok", &second_saw_auth });
    defer t2.join();

    var client = Client.init(io, allocator);
    defer client.deinit();
    var res = try client.fetch(.{ .url = url, .authorization = "Bearer secret" });
    defer res.deinit(allocator);

    try std.testing.expectEqual(std.http.Status.ok, res.status);
    try std.testing.expect(first_saw_auth.load(.acquire));
    try std.testing.expect(!second_saw_auth.load(.acquire));
}

fn redirectResponse(allocator: std.mem.Allocator, status: []const u8, location: []const u8) ![]u8 {
    return std.fmt.allocPrint(allocator, "HTTP/1.1 {s}\r\nLocation: {s}\r\nContent-Length: 0\r\n\r\n", .{ status, location });
}

const ok_response = "HTTP/1.1 200 OK\r\nContent-Length: 2\r\n\r\nok";

test "a redirect back to a visited url is a loop, not a hang" {
    const allocator = std.testing.allocator;
    var threaded: std.Io.Threaded = .init(allocator, .{});
    defer threaded.deinit();
    const io = threaded.io();

    var origin = try TestOrigin.init(io);
    defer origin.deinit();
    const url = try std.fmt.allocPrint(allocator, "http://127.0.0.1:{d}/self", .{origin.port()});
    defer allocator.free(url);
    const response = try redirectResponse(allocator, "302 Found", "/self");
    defer allocator.free(response);

    var server = try std.Thread.spawn(.{}, struct {
        fn run(o: *TestOrigin, r: []const u8) void {
            o.serveOnce(r) catch {};
        }
    }.run, .{ &origin, response });
    defer server.join();

    var client = Client.init(io, allocator);
    defer client.deinit();
    try std.testing.expectError(error.RedirectLoop, client.fetch(.{ .url = url }));
}

test "hops past the policy are an error; the policy is per client" {
    const allocator = std.testing.allocator;
    var threaded: std.Io.Threaded = .init(allocator, .{});
    defer threaded.deinit();
    const io = threaded.io();

    var origin = try TestOrigin.init(io);
    defer origin.deinit();
    const url = try std.fmt.allocPrint(allocator, "http://127.0.0.1:{d}/0", .{origin.port()});
    defer allocator.free(url);

    // /0 -> /1 -> /2: two hops, each a distinct url.
    const r0 = try redirectResponse(allocator, "302 Found", "/1");
    defer allocator.free(r0);
    const r1 = try redirectResponse(allocator, "302 Found", "/2");
    defer allocator.free(r1);

    var server = try std.Thread.spawn(.{}, struct {
        fn run(o: *TestOrigin, a: []const u8, b: []const u8) void {
            o.serveOnce(a) catch {};
            o.serveOnce(b) catch {};
        }
    }.run, .{ &origin, r0, r1 });
    defer server.join();

    var client = Client.init(io, allocator);
    defer client.deinit();
    client.keep_alive = false;
    client.redirects.max_hops = 1;
    try std.testing.expectError(error.TooManyRedirects, client.fetch(.{ .url = url }));
}

test "surface hands the 3xx back; refuse makes it an error" {
    const allocator = std.testing.allocator;
    var threaded: std.Io.Threaded = .init(allocator, .{});
    defer threaded.deinit();
    const io = threaded.io();

    var origin = try TestOrigin.init(io);
    defer origin.deinit();
    const url = try std.fmt.allocPrint(allocator, "http://127.0.0.1:{d}/x", .{origin.port()});
    defer allocator.free(url);
    const response = try redirectResponse(allocator, "301 Moved Permanently", "http://example.invalid/never-dialed");
    defer allocator.free(response);

    var server = try std.Thread.spawn(.{}, struct {
        fn run(o: *TestOrigin, r: []const u8) void {
            o.serveOnce(r) catch {};
            o.serveOnce(r) catch {};
        }
    }.run, .{ &origin, response });
    defer server.join();

    var client = Client.init(io, allocator);
    defer client.deinit();

    var surfaced = try client.fetch(.{ .url = url, .on_redirect = .surface });
    defer surfaced.deinit(allocator);
    try std.testing.expectEqual(std.http.Status.moved_permanently, surfaced.status);

    try std.testing.expectError(error.RedirectRefused, client.fetch(.{ .url = url, .on_redirect = .refuse }));
}

test "https never steps down to http unless the policy allows it" {
    // No TLS in the loopback fixture, so build the check's inputs directly.
    const https = try std.Uri.parse("https://a.example/x");
    const http = try std.Uri.parse("http://a.example/x");
    try std.testing.expect(try sameOrigin(https, https));
    try std.testing.expect(!try sameOrigin(https, http));
    try std.testing.expect(try sameOrigin(try std.Uri.parse("http://A.example:80/"), try std.Uri.parse("http://a.example/")));
    try std.testing.expect(!try sameOrigin(try std.Uri.parse("http://a.example:8080/"), try std.Uri.parse("http://a.example/")));
}

test "303 and 301-on-POST become a bodiless GET; 307 resends the body" {
    const allocator = std.testing.allocator;
    var threaded: std.Io.Threaded = .init(allocator, .{});
    defer threaded.deinit();
    const io = threaded.io();

    var origin = try TestOrigin.init(io);
    defer origin.deinit();
    const url = try std.fmt.allocPrint(allocator, "http://127.0.0.1:{d}/post", .{origin.port()});
    defer allocator.free(url);

    const Recorded = struct {
        line: [128]u8 = undefined,
        len: usize = 0,
        body: std.atomic.Value(bool) = .init(false),
        fn text(self: *const @This()) []const u8 {
            return self.line[0..self.len];
        }
    };
    var after_303: Recorded = .{};
    var after_301: Recorded = .{};
    var after_307: Recorded = .{};

    const r303 = try redirectResponse(allocator, "303 See Other", "/after-303");
    defer allocator.free(r303);
    const r301 = try redirectResponse(allocator, "301 Moved Permanently", "/after-301");
    defer allocator.free(r301);
    const r307 = try redirectResponse(allocator, "307 Temporary Redirect", "/after-307");
    defer allocator.free(r307);

    var server = try std.Thread.spawn(.{}, struct {
        fn run(o: *TestOrigin, a: []const u8, b: []const u8, c: []const u8, ra: *Recorded, rb: *Recorded, rc: *Recorded) void {
            o.serveOnce(a) catch {};
            o.serveRecordingRequest(ok_response, &ra.line, &ra.len, &ra.body) catch {};
            o.serveOnce(b) catch {};
            o.serveRecordingRequest(ok_response, &rb.line, &rb.len, &rb.body) catch {};
            o.serveOnce(c) catch {};
            o.serveRecordingRequest(ok_response, &rc.line, &rc.len, &rc.body) catch {};
        }
    }.run, .{ &origin, r303, r301, r307, &after_303, &after_301, &after_307 });
    defer server.join();

    var client = Client.init(io, allocator);
    defer client.deinit();
    client.keep_alive = false;
    const post: Client.FetchOptions = .{ .url = url, .method = .POST, .payload = "{}", .content_type = "application/json" };

    var a = try client.fetch(post);
    a.deinit(allocator);
    try std.testing.expectEqualStrings("GET /after-303 HTTP/1.1", after_303.text());
    try std.testing.expect(!after_303.body.load(.acquire));

    var b = try client.fetch(post);
    b.deinit(allocator);
    try std.testing.expectEqualStrings("GET /after-301 HTTP/1.1", after_301.text());
    try std.testing.expect(!after_301.body.load(.acquire));

    var c = try client.fetch(post);
    c.deinit(allocator);
    try std.testing.expectEqualStrings("POST /after-307 HTTP/1.1", after_307.text());
    try std.testing.expect(after_307.body.load(.acquire));
}

test "a payload without a content type is refused up front" {
    const io = std.Options.debug_io;
    var client = Client.init(io, std.testing.allocator);
    defer client.deinit();
    try std.testing.expectError(error.MissingContentType, client.fetch(.{ .url = "http://127.0.0.1:1/", .method = .POST, .payload = "{}" }));
}

test "retry-after accepts the http-date form and delaySeconds ranks the hints" {
    const H = Client.RateLimitHeaders;
    // Sun, 06 Nov 1994 08:49:37 GMT is 784111777.
    const head = try std.http.Client.Response.Head.parse("HTTP/1.1 503 x\r\nRetry-After: Sun, 06 Nov 1994 08:49:37 GMT\r\n\r\n");
    const dated = H.fromResponseHead(head);
    try std.testing.expectEqual(@as(?i64, 784_111_777), dated.retry_after_at);
    try std.testing.expectEqual(@as(?u64, 7), dated.delaySeconds(784_111_770));
    try std.testing.expectEqual(@as(?u64, 0), dated.delaySeconds(784_111_800));

    try std.testing.expectEqual(@as(?u64, 5), (H{ .retry_after = 5, .reset = 1_800_000_000 }).delaySeconds(1_700_000_000));
    try std.testing.expectEqual(@as(?u64, 100_000_000), (H{ .reset = 1_800_000_000 }).delaySeconds(1_700_000_000));
    try std.testing.expectEqual(@as(?u64, 30), (H{ .reset = 30 }).delaySeconds(1_700_000_000));
    try std.testing.expectEqual(@as(?u64, null), (H{ .limit = 10 }).delaySeconds(0));

    try std.testing.expectEqual(@as(?i64, 784_111_777), parseHttpDate("Sunday, 06-Nov-94 08:49:37 GMT"));
    try std.testing.expectEqual(@as(?i64, 0), parseHttpDate("Thu, 01 Jan 1970 00:00:00 GMT"));
    try std.testing.expectEqual(@as(?i64, 951_782_400), parseHttpDate("Tue, 29 Feb 2000 00:00:00 GMT"));
}

test "a transfer that started fast and now trickles is judged on the trickle" {
    const allocator = std.testing.allocator;
    var threaded: std.Io.Threaded = .init(allocator, .{});
    defer threaded.deinit();
    const io = threaded.io();

    var origin = try TestOrigin.init(io);
    defer origin.deinit();
    const url = try std.fmt.allocPrint(allocator, "http://127.0.0.1:{d}/big", .{origin.port()});
    defer allocator.free(url);

    // 4 MiB lands at once. A lifetime average would read as megabytes per
    // second for a long time; the trailing window sees nothing moving.
    var server = try std.Thread.spawn(.{}, struct {
        fn run(o: *TestOrigin) void {
            o.serveBurstThenStall(4 * 1024 * 1024) catch {};
        }
    }.run, .{&origin});
    defer server.join();

    var client = Client.init(io, allocator);
    defer client.deinit();
    client.stall = .{
        .idle_ns = 0,
        .min_bytes_per_s = 1024 * 1024,
        .grace_ns = 400 * std.time.ns_per_ms,
        .window_ns = 300 * std.time.ns_per_ms,
    };
    const started = std.Io.Timestamp.now(io, .awake).nanoseconds;
    try std.testing.expectError(error.TransferTooSlow, client.fetch(.{ .url = url }));
    const elapsed_ns = std.Io.Timestamp.now(io, .awake).nanoseconds - started;
    try std.testing.expect(elapsed_ns < 3 * std.time.ns_per_s);
}

test "the request deadline bounds a dribbling body, with the stall guard on or off" {
    const allocator = std.testing.allocator;
    var threaded: std.Io.Threaded = .init(allocator, .{});
    defer threaded.deinit();
    const io = threaded.io();

    var origin = try TestOrigin.init(io);
    defer origin.deinit();
    const url = try std.fmt.allocPrint(allocator, "http://127.0.0.1:{d}/slow", .{origin.port()});
    defer allocator.free(url);

    var server = try std.Thread.spawn(.{}, struct {
        fn run(o: *TestOrigin) void {
            o.serveStalling() catch {};
            o.serveStalling() catch {};
        }
    }.run, .{&origin});
    defer server.join();

    var client = Client.init(io, allocator);
    defer client.deinit();
    client.keep_alive = false;

    // Guards on: the deadline has to cancel through the stall watcher and
    // its reader task.
    const started = std.Io.Timestamp.now(io, .awake).nanoseconds;
    try std.testing.expectError(error.RequestDeadlineExceeded, client.fetch(.{
        .url = url,
        .deadline_ns = 300 * std.time.ns_per_ms,
    }));
    // Guards off: the deadline cancels a plain blocking read.
    try std.testing.expectError(error.RequestDeadlineExceeded, client.fetch(.{
        .url = url,
        .deadline_ns = 300 * std.time.ns_per_ms,
        .stall = .{ .idle_ns = 0, .min_bytes_per_s = 0 },
    }));
    const elapsed_ns = std.Io.Timestamp.now(io, .awake).nanoseconds - started;
    try std.testing.expect(elapsed_ns < 3 * std.time.ns_per_s);
}

test "a followed redirect reports where the response came from" {
    const allocator = std.testing.allocator;
    var threaded: std.Io.Threaded = .init(allocator, .{});
    defer threaded.deinit();
    const io = threaded.io();

    var origin = try TestOrigin.init(io);
    defer origin.deinit();
    const url = try std.fmt.allocPrint(allocator, "http://127.0.0.1:{d}/start", .{origin.port()});
    defer allocator.free(url);
    const expected = try std.fmt.allocPrint(allocator, "http://127.0.0.1:{d}/final", .{origin.port()});
    defer allocator.free(expected);
    const response = try redirectResponse(allocator, "302 Found", "/final");
    defer allocator.free(response);

    var server = try std.Thread.spawn(.{}, struct {
        fn run(o: *TestOrigin, r: []const u8) void {
            o.serveOnce(r) catch {};
            o.serveOnce(ok_response) catch {};
            o.serveOnce(ok_response) catch {};
        }
    }.run, .{ &origin, response });
    defer server.join();

    var client = Client.init(io, allocator);
    defer client.deinit();
    client.keep_alive = false;
    var res = try client.fetch(.{ .url = url });
    defer res.deinit(allocator);
    try std.testing.expectEqualStrings(expected, res.final_url.?);

    var direct = try client.fetch(.{ .url = expected });
    defer direct.deinit(allocator);
    try std.testing.expectEqual(@as(?[]u8, null), direct.final_url);
}

test "control characters in a url or header are refused before anything is sent" {
    const io = std.Options.debug_io;
    var client = Client.init(io, std.testing.allocator);
    defer client.deinit();
    try std.testing.expectError(error.InvalidUrl, client.fetch(.{ .url = "http://127.0.0.1:1/x\r\nx-injected: 1\r\n" }));
    try std.testing.expectError(error.InvalidHeader, client.fetch(.{ .url = "http://127.0.0.1:1/", .authorization = "Bearer a\r\nx-injected: 1" }));
    try std.testing.expectError(error.InvalidHeader, client.fetch(.{
        .url = "http://127.0.0.1:1/",
        .extra_headers = &.{.{ .name = "x-a", .value = "1\nx-b: 2" }},
    }));
    try std.testing.expectError(error.InvalidHeader, client.fetch(.{
        .url = "http://127.0.0.1:1/",
        .extra_headers = &.{.{ .name = "x-a: b", .value = "1" }},
    }));
}

test "the response cap is enforced while streaming, not by preallocating it" {
    const allocator = std.testing.allocator;
    var threaded: std.Io.Threaded = .init(allocator, .{});
    defer threaded.deinit();
    const io = threaded.io();

    var origin = try TestOrigin.init(io);
    defer origin.deinit();
    const url = try std.fmt.allocPrint(allocator, "http://127.0.0.1:{d}/x", .{origin.port()});
    defer allocator.free(url);

    var server = try std.Thread.spawn(.{}, struct {
        fn run(o: *TestOrigin) void {
            o.serveOnce("HTTP/1.1 200 OK\r\nContent-Length: 5\r\n\r\nhello") catch {};
            o.serveOnce("HTTP/1.1 200 OK\r\nContent-Length: 5\r\n\r\nhello") catch {};
        }
    }.run, .{&origin});
    defer server.join();

    var client = Client.init(io, allocator);
    defer client.deinit();
    client.keep_alive = false;
    client.max_response_size = 4;
    try std.testing.expectError(error.ResponseTooLarge, client.fetch(.{ .url = url }));

    var res = try client.fetch(.{ .url = url, .max_response_size = 1 << 30 });
    defer res.deinit(allocator);
    try std.testing.expectEqualStrings("hello", res.body);
}

test "open then stream: the sink is chosen after the head, and the body never touches the heap" {
    const allocator = std.testing.allocator;
    var threaded: std.Io.Threaded = .init(allocator, .{});
    defer threaded.deinit();
    const io = threaded.io();

    var origin = try TestOrigin.init(io);
    defer origin.deinit();
    const url = try std.fmt.allocPrint(allocator, "http://127.0.0.1:{d}/repo", .{origin.port()});
    defer allocator.free(url);

    // A 64 KiB body, then a 404 with a json body: the two shapes a bulk
    // downloader sees, wanting a file for one and a few bytes for the other.
    const big = try allocator.alloc(u8, 64 * 1024);
    defer allocator.free(big);
    @memset(big, 'z');
    const big_response = try std.fmt.allocPrint(allocator, "HTTP/1.1 200 OK\r\nContent-Length: {d}\r\n\r\n{s}", .{ big.len, big });
    defer allocator.free(big_response);
    const not_found = "HTTP/1.1 404 Not Found\r\nContent-Type: application/json\r\nContent-Length: 24\r\n\r\n{\"error\":\"RepoNotFound\"}";

    var server = try std.Thread.spawn(.{}, struct {
        fn run(o: *TestOrigin, a: []const u8, b: []const u8) void {
            o.serveOnce(a) catch {};
            o.serveOnce(b) catch {};
        }
    }.run, .{ &origin, big_response, not_found });
    defer server.join();

    var client = Client.init(io, allocator);
    defer client.deinit();
    client.keep_alive = false;

    var tmp = std.testing.tmpDir(.{});
    defer tmp.cleanup();

    {
        const response = try client.open(.{ .url = url });
        defer response.deinit();
        try std.testing.expectEqual(std.http.Status.ok, response.status);

        var file = try tmp.dir.createFile(io, "repo.car", .{ .read = true });
        defer file.close(io);
        var wbuf: [8 * 1024]u8 = undefined;
        var fw = std.Io.File.Writer.initStreaming(file, io, &wbuf);
        var progress: Progress = .{};
        const n = try response.stream(&fw.interface, .{ .progress = &progress });
        try fw.interface.flush();
        try std.testing.expectEqual(big.len, n);
        try std.testing.expectEqual(big.len, progress.bytes.load(.acquire));
        try std.testing.expect(progress.done.load(.acquire));
        try std.testing.expectEqual(big.len, try file.length(io));
        try std.testing.expectError(error.BodyAlreadyTaken, response.stream(&fw.interface, .{}));
    }

    {
        const response = try client.open(.{ .url = url, .capture_headers = &.{"content-type"} });
        defer response.deinit();
        try std.testing.expectEqual(std.http.Status.not_found, response.status);
        try std.testing.expectEqualStrings("application/json", response.headers.get("content-type").?);
        var small: [64]u8 = undefined;
        var fixed = std.Io.Writer.fixed(&small);
        _ = try response.stream(&fixed, .{ .max_bytes = small.len });
        try std.testing.expectEqualStrings("{\"error\":\"RepoNotFound\"}", fixed.buffered());
    }
}

test "a response dropped without streaming leaves the client usable" {
    const allocator = std.testing.allocator;
    var threaded: std.Io.Threaded = .init(allocator, .{});
    defer threaded.deinit();
    const io = threaded.io();

    var origin = try TestOrigin.init(io);
    defer origin.deinit();
    const url = try std.fmt.allocPrint(allocator, "http://127.0.0.1:{d}/x", .{origin.port()});
    defer allocator.free(url);

    var server = try std.Thread.spawn(.{}, struct {
        fn run(o: *TestOrigin) void {
            o.serveOnce(ok_response) catch {};
            o.serveOnce(ok_response) catch {};
        }
    }.run, .{&origin});
    defer server.join();

    var client = Client.init(io, allocator);
    defer client.deinit();

    const dropped = try client.open(.{ .url = url });
    dropped.deinit();

    var res = try client.fetch(.{ .url = url });
    defer res.deinit(allocator);
    try std.testing.expectEqualStrings("ok", res.body);
}

test "what the request deadline has left after the head bounds the body" {
    const allocator = std.testing.allocator;
    var threaded: std.Io.Threaded = .init(allocator, .{});
    defer threaded.deinit();
    const io = threaded.io();

    var origin = try TestOrigin.init(io);
    defer origin.deinit();
    const url = try std.fmt.allocPrint(allocator, "http://127.0.0.1:{d}/slow", .{origin.port()});
    defer allocator.free(url);

    var server = try std.Thread.spawn(.{}, struct {
        fn run(o: *TestOrigin) void {
            o.serveStalling() catch {};
        }
    }.run, .{&origin});
    defer server.join();

    var client = Client.init(io, allocator);
    defer client.deinit();

    const response = try client.open(.{ .url = url, .deadline_ns = 300 * std.time.ns_per_ms });
    defer response.deinit();
    try std.testing.expect(response.deadline_remaining_ns > 0);
    try std.testing.expect(response.deadline_remaining_ns <= 300 * std.time.ns_per_ms);

    var aw: std.Io.Writer.Allocating = .init(allocator);
    defer aw.deinit();
    const started = std.Io.Timestamp.now(io, .awake).nanoseconds;
    try std.testing.expectError(error.RequestDeadlineExceeded, response.stream(&aw.writer, .{
        .stall = .{ .idle_ns = 0, .min_bytes_per_s = 0 },
    }));
    try std.testing.expect(std.Io.Timestamp.now(io, .awake).nanoseconds - started < 2 * std.time.ns_per_s);
}

test "accept_encoding replaces std's default instead of adding a second header" {
    const allocator = std.testing.allocator;
    var threaded: std.Io.Threaded = .init(allocator, .{});
    defer threaded.deinit();
    const io = threaded.io();

    var origin = try TestOrigin.init(io);
    defer origin.deinit();
    const url = try std.fmt.allocPrint(allocator, "http://127.0.0.1:{d}/x", .{origin.port()});
    defer allocator.free(url);

    var default_head: [4096]u8 = undefined;
    var default_len: usize = 0;
    var identity_head: [4096]u8 = undefined;
    var identity_len: usize = 0;
    var server = try std.Thread.spawn(.{}, struct {
        fn run(o: *TestOrigin, a: *[4096]u8, al: *usize, b: *[4096]u8, bl: *usize) void {
            o.serveRecordingHead(ok_response, a, al) catch {};
            o.serveRecordingHead(ok_response, b, bl) catch {};
        }
    }.run, .{ &origin, &default_head, &default_len, &identity_head, &identity_len });
    defer server.join();

    var client = Client.init(io, allocator);
    defer client.deinit();
    client.keep_alive = false;

    var a = try client.fetch(.{ .url = url });
    a.deinit(allocator);
    var b = try client.fetch(.{ .url = url, .accept_encoding = "identity" });
    b.deinit(allocator);

    try std.testing.expect(std.ascii.findIgnoreCase(default_head[0..default_len], "\r\naccept-encoding: gzip") != null);
    const identity = identity_head[0..identity_len];
    try std.testing.expect(std.ascii.findIgnoreCase(identity, "\r\naccept-encoding: identity\r\n") != null);
    try std.testing.expectEqual(@as(usize, 1), std.mem.count(u8, identity, "ccept-encoding"));
}

test "a request that may have taken effect is never replayed" {
    try std.testing.expect(!isReplayable(.{ .url = "http://x/", .method = .POST }));
    try std.testing.expect(!isReplayable(.{ .url = "http://x/", .method = .GET, .payload = "b" }));
    try std.testing.expect(isReplayable(.{ .url = "http://x/", .method = .GET }));
}

test "HTTP dates accept legacy formats and reject invalid calendars" {
    const H = Client.RateLimitHeaders;
    inline for (.{ "Sunday, 06-Nov-94 08:49:37 GMT", "Sun Nov  6 08:49:37 1994", "Sun Nov 06 08:49:37 1994" }) |date| {
        const head = try std.http.Client.Response.Head.parse("HTTP/1.1 503 x\r\nRetry-After: " ++ date ++ "\r\n\r\n");
        const hints = H.fromResponseHead(head);
        try std.testing.expectEqual(@as(?i64, 784_111_777), hints.retry_after_at);
        try std.testing.expectEqual(@as(?u64, 7), hints.delaySeconds(784_111_770));
        try std.testing.expectEqual(@as(?u64, 0), hints.delaySeconds(784_111_800));
    }
    for ([_][]const u8{
        "Sun, 31 Feb 1994 08:49:37 GMT",
        "Sunday, 29-Feb-99 08:49:37 GMT",
        "Sun Feb 29 08:49:37 1900",
        "Sun Nov  6 24:49:37 1994",
        "Sun, 06 Nov 1994 08:49:60 GMT",
        "Nop Nov  6 08:49:37 1994",
        "Sunday, 06-Nov-94 08:49:37 GMT trailing",
        "",
    }) |date| try std.testing.expectEqual(@as(?i64, null), parseHttpDate(date));
    try std.testing.expectEqual(@as(?i64, 951_782_400), parseHttpDate("Tuesday, 29-Feb-00 00:00:00 GMT"));
}

test "rate limit reset retains signed timestamps for protocol retry policy" {
    const head = try std.http.Client.Response.Head.parse("HTTP/1.1 429 Too Many Requests\r\nRateLimit-Reset: -1\r\nRetry-After: 9\r\n\r\n");
    const headers = Client.RateLimitHeaders.fromResponseHead(head);
    try std.testing.expect(headers.reset != null);
    try std.testing.expectEqual(@as(i128, -1), @as(i128, headers.reset.?));
    try std.testing.expectEqual(@as(?u64, 9), headers.retry_after);
}

test "signed reset rejects overflow and generic delay clamps expired values" {
    inline for (.{ "9223372036854775808", "-9223372036854775809", "bad" }) |value| {
        const head = try std.http.Client.Response.Head.parse("HTTP/1.1 429 Too Many Requests\r\nRateLimit-Reset: " ++ value ++ "\r\n\r\n");
        try std.testing.expectEqual(@as(?i64, null), Client.RateLimitHeaders.fromResponseHead(head).reset);
    }
    try std.testing.expectEqual(@as(?u64, 0), (Client.RateLimitHeaders{ .reset = -1 }).delaySeconds(100));
    try std.testing.expectEqual(@as(?u64, 9), (Client.RateLimitHeaders{ .reset = -1, .retry_after = 9 }).delaySeconds(100));
}

test "stream delivers buffered body before waiting for the next network read" {
    const allocator = std.testing.allocator;
    var threaded: std.Io.Threaded = .init(allocator, .{});
    defer threaded.deinit();
    const io = threaded.io();
    var origin = try TestOrigin.init(io);
    defer origin.deinit();
    var progress: Progress = .{};
    var observed: std.atomic.Value(bool) = .init(false);
    const server = try std.Thread.spawn(.{}, struct {
        fn run(o: *TestOrigin, p: *Progress, seen: *std.atomic.Value(bool)) void {
            var conn = o.listener.accept(o.io) catch return;
            defer conn.close(o.io);
            var rb: [4096]u8 = undefined;
            var r = conn.reader(o.io, &rb);
            _ = r.interface.takeDelimiterInclusive('\n') catch return;
            var w = conn.writer(o.io, &.{});
            w.interface.writeAll("HTTP/1.1 200 OK\r\nContent-Type: text/event-stream\r\nConnection: close\r\n\r\ndata: first\n\n") catch return;
            w.interface.flush() catch return;
            for (0..50) |_| {
                if (p.bytes.load(.acquire) > 0) {
                    seen.store(true, .release);
                    break;
                }
                std.Io.Clock.Duration.sleep(.{ .raw = .fromMilliseconds(10), .clock = .awake }, o.io) catch return;
            }
            w.interface.writeAll("data: [DONE]\n\n") catch return;
            w.interface.flush() catch return;
        }
    }.run, .{ &origin, &progress, &observed });
    defer server.join();
    var client = Client.init(io, allocator);
    defer client.deinit();
    const url = try std.fmt.allocPrint(allocator, "http://127.0.0.1:{d}/stream", .{origin.port()});
    defer allocator.free(url);
    var response = try client.open(.{ .url = url });
    defer response.deinit();
    var body: std.Io.Writer.Allocating = .init(allocator);
    defer body.deinit();
    _ = try response.stream(&body.writer, .{ .progress = &progress });
    try std.testing.expect(observed.load(.acquire));
    try std.testing.expectEqualStrings("data: first\n\ndata: [DONE]\n\n", body.written());
}