Artifact [21ca6c42ba]

Artifact 21ca6c42ba1da1eee45a2b96e16cc2a3749e8d7987e396809f50850853a756ec:

• File src/serve.zig — part of check-in [e87fb840d1] at 2026-07-20 04:34:32 on branch h11-serve-socket — serve: bound each connection so a stalled client cannot wedge the server

  zit serve accepts one connection to completion before the next (single-threaded, serve.zig), and receiveHead plus the body read blocked with no deadline. A loopback peer that dribbles a request head, half-sends a body, or connects idle and holds could pin the accept loop forever -- the socket read path the exchange fuzzers never reach, covered before only by happy-path interop.

  Add a whole-connection deadline: the request work runs on its own unit of concurrency raced against a timed sleep, and draining the loser cancels the blocked read, freeing the loop. A per-read SO_RCVTIMEO is unusable here (the Io backend treats a socket-timeout EAGAIN as a bug). serve binds 127.0.0.1 only, so a legitimate same-host request finishes in milliseconds even at the 512 MiB body ceiling; the 5 s bound never fires on a real client and only sheds a stalled one.

  Four end-to-end scenarios abuse a live server over loopback -- dribbled head, half-sent body, mid-response disconnect, idle-connection burst -- and assert with bounded waits that a fresh request is still answered and a clone still converges (sync 3.7). (user: dmestas, size: 34882) [annotate] [blame] [check-ins using] [more...]

//! zit — CLI-layer HTTP server for the Fossil sync protocol (sync §4).
//!
//! This module, with `transport.zig`, is one of only two places `std.http` and
//! real network I/O appear; `sync.zig` stays sans-IO. It puts an HTTP face on
//! the stateless responder `sync.Server.respond` (§3.2/§3.8): a POST carrying
//! an `application/x-fossil` body — to any URL path (§4 L759-764) — is decoded
//! from its §4.1 zlib
//! container to plaintext request cards, answered by `respond` over the
//! repository's `ArtifactStore`, and the reply cards are re-framed into a
//! container and returned as `application/x-fossil` — except a §8 clone reply,
//! which per §8.1/§4.1 is returned raw (no outer container) as
//! `application/x-fossil-uncompressed`, because its cfile payloads are already
//! stored compressed. `respondWithMedia` reports which of the two applies, so
//! the framing and the Content-Type are picked from one decision.
//!
//! Because `respond` is stateless — it takes `self: *const Server` and mutates
//! only the caller's store (§3.8) — the server holds no cross-request sync
//! state. Each request is answered independently over a freshly read view of
//! the store (`Responder`, supplied by the caller), so a killed server leaves
//! nothing partial.
//!
//! The §4.1 container codec is `codec.encode`/`codec.decode` (the leaf
//! `container.zig`) — the single framer `transport.zig` also wraps; no second framer is written
//! here. Every value that crosses the wire is `[]const u8` (Fossil cards embed
//! NUL, so no sentinel-terminated type is used).

const std = @import("std");
const zit = @import("zit");
const sync = zit.sync;
const codec = zit.container;
const repostate = zit.repostate;

const assert = std.debug.assert;
const http = std.http;
const net = std.Io.net;

/// §4 L1144-1147 / r15 — the production request/response media type. The body
/// under it is always the §4.1 zlib container (`outer_compressed = true`).
pub const xfer_media_type: []const u8 = "application/x-fossil";

/// §4 L1148-1149 — the message-body ceiling. Reused symmetrically as the bound
/// on an accepted request container (outer wire size) and on its decoded
/// plaintext, so a lying §4.1 length prefix cannot drive unbounded allocation
/// (`codec.decode` checks the declared length against this before
/// decompressing).
pub const max_request_body: usize = sync.reply_body_cap;

/// §10.1 — the release integer this server emits as `pragma server-version`.
/// A fixed build identity; carries a datetime so the full server-version form
/// is always available when a dated client-version asks for it (§10.1 r?/A.4).
pub const server_version: sync.VersionInfo = .{
    .version = 22800,
    .datetime = .{ .date = 20260715, .time = 0 },
};

/// §8.2/§9.1 — the SERVERCODE and PROJECTCODE fallbacks this server emits in
/// a clone reply's `push` card. `runServe` now reads the real per-repo
/// project code from the repository config and passes it in; these fixed
/// 16-hex values only cover a repo predating the stored code or an unreadable
/// config. Servercode is informational (§9.1); a well-formed, whitespace-free
/// pair keeps the `push` card three-token even on the fallback path.
pub const server_servercode = "0000000000000000";
pub const server_projectcode = "0000000000000000";

/// The closed error set of the per-request seam `handleXfer`. A request that is
/// too large, malformed, or would produce an over-cap reply is a client-visible
/// HTTP status, never a crash. Authentication failure is NOT here — `respond`
/// answers it in-band as an `error` card (§5.2/§6.4).
pub const XferError = error{
    /// The request container, or its declared plaintext length, exceeds
    /// `max_request_body` (§4 L1148-1149).
    RequestTooLarge,
    /// The request body is not a decodable §4.1 zlib container.
    RequestMalformed,
    /// `respond` produced a reply exceeding the §4 body ceiling.
    ResponseTooLarge,
    OutOfMemory,
};

/// The error set a `Responder` may surface: the seam errors plus a server-side
/// storage read failure (the repository could not be read to answer). Kept
/// distinct from `XferError` so the store-read failure maps to a 5xx, not a
/// client 4xx.
pub const AnswerError = XferError || error{ServerStorage};

/// A wire-ready reply and the Content-Type it MUST be served under. The two
/// travel together so a caller cannot serve the body under a Content-Type that
/// disagrees with its framing: a normal reply is the `application/x-fossil`
/// zlib container, a §8 clone reply is the raw card/cfile stream under
/// `application/x-fossil-uncompressed` (§8.1/§4.1). `body` is caller-owned.
pub const XferReply = struct {
    body: []u8,
    content_type: []const u8,
};

/// A capability that answers one framed sync request over the repository. The
/// HTTP layer holds no store knowledge: the caller (the CLI verb) supplies the
/// store-reading side here, so the server mechanism stays a deep module over a
/// one-method interface (§3.8 statelessness lives on the caller's side).
pub const Responder = struct {
    ctx: *anyopaque,
    answerFn: *const fn (ctx: *anyopaque, gpa: std.mem.Allocator, request_container: []const u8) AnswerError!XferReply,

    /// Answer `request_container` (a §4.1 container), returning the wire-ready
    /// reply body and its Content-Type; the caller owns and frees the body.
    pub fn answer(self: Responder, gpa: std.mem.Allocator, request_container: []const u8) AnswerError!XferReply {
        return self.answerFn(self.ctx, gpa, request_container);
    }
};

// ---------------------------------------------------------------------------
// The pure per-request seam — the deterministic, socket-free core.
// ---------------------------------------------------------------------------

/// Answer one framed sync request over `store`: decode the §4.1 container to
/// plaintext request cards, call the stateless `server.respondWithMedia` over
/// the store, and produce the wire-ready reply. An ordinary reply is re-framed
/// into an `application/x-fossil` container; a §8 clone reply is returned raw
/// (no outer container) as `application/x-fossil-uncompressed` (§8.1/§4.1),
/// because its cfile payloads are already stored compressed. Sans-IO and
/// deterministic: framed bytes in, wire bytes out. The returned body is
/// caller-owned.
pub fn handleXfer(
    gpa: std.mem.Allocator,
    server: *const sync.Server,
    store: repostate.ArtifactStore,
    request_container: []const u8,
) XferError!XferReply {
    // §4 L1148-1149 — bound the outer wire size before touching it (positive
    // space); the declared plaintext length is bounded independently below.
    if (request_container.len > max_request_body) return error.RequestTooLarge;

    const request = codec.decode(gpa, request_container, max_request_body) catch |err| switch (err) {
        error.OutOfMemory => return error.OutOfMemory,
        error.TruncatedHeader, error.InvalidStream => return error.RequestMalformed,
        error.OutputTooLarge => return error.RequestTooLarge,
    };
    defer gpa.free(request);
    assert(request.len <= max_request_body); // postcondition of the capped decode

    var out: std.ArrayListUnmanaged(u8) = .empty;
    defer out.deinit(gpa);
    const reply = server.respondWithMedia(gpa, store, request, &out) catch |err| switch (err) {
        error.OutOfMemory => return error.OutOfMemory,
        error.ResponseTooLarge => return error.ResponseTooLarge,
    };

    const props = sync.media_props.get(reply.media);
    if (!props.outer_compressed) {
        // §8.1/§4.1 — the clone reply is served with no outer container; its
        // cfile payloads are already stored compressed. `reply.body` aliases the
        // scratch `out`, so hand back an owned copy.
        assert(reply.media == .x_fossil_uncompressed); // only the clone reply is uncompressed
        const body = gpa.dupe(u8, reply.body) catch return error.OutOfMemory;
        return .{ .body = body, .content_type = props.wire };
    }

    // The uncompressed §8 clone reply took the branch above; every remaining
    // reply is the compressed production media, re-framed into the outer
    // container and served as application/x-fossil (§4.1).
    assert(reply.media != .x_fossil_uncompressed);
    assert(std.mem.eql(u8, props.wire, xfer_media_type));
    const container = codec.encode(gpa, reply.body, .default) catch |err| switch (err) {
        error.OutOfMemory => return error.OutOfMemory,
        error.InputTooLarge => return error.ResponseTooLarge,
    };
    assert(container.len >= 4); // §4.1 — every container carries the 4-byte length prefix
    return .{ .body = container, .content_type = props.wire };
}

// ---------------------------------------------------------------------------
// HTTP request classification — pure, so the accept/reject decision is tested
// without a socket.
// ---------------------------------------------------------------------------

/// The HTTP disposition of an incoming request against the sync contract: serve
/// it, or reject with the matching status. Dispatch is method → media type;
/// the URL path is NOT consulted (§4 L759-764 — sync dispatches "regardless of
/// path").
pub const Disposition = enum {
    ok,
    method_not_allowed, // 405 — non-POST (see classifyXfer note on §4 "redirected home")
    unsupported_media_type, // 415 — POST without a request-valid Content-Type
};

/// Classify a request line by method + Content-Type, independent of the URL
/// path. Pure — no allocation, no I/O.
///
/// §4 L759-764 is authoritative: "Sync uses HTTP POST to the repository URL
/// path unchanged; empty path becomes /. … Clients MUST NOT append /xfer.
/// application/x-fossil and application/x-fossil-debug dispatch to sync
/// regardless of path." So a POST bearing a processable sync media type is
/// sync at ANY path (`/`, `/xfer`, `/anything`); the `/xfer` string is only a
/// client-side redirect-normalization detail (§4 L791), never a server
/// endpoint requirement. `target` is accepted but deliberately unused. Of the
/// two request-valid types only the compressed x-fossil is processed today;
/// the uncompressed x-fossil-debug diagnostic transport is refused 415 by the
/// media-type gate below (see there).
///
/// Non-POST: §4 L762-763 / `sync.serverMethodDisposition` (r5) says such a
/// request is "redirected home." A redirect-home targets the repository's web
/// UI home page, which this minimal sync-only server does not serve — and a
/// 302 to "/" would loop, since dispatch is now path-independent and "/" is a
/// valid sync endpoint. A conformant sync client always POSTs (§4 L760), so it
/// never reaches this branch; a 405 therefore breaks no conformant client. We
/// keep 405 method_not_allowed rather than invent a web UI to redirect to.
pub fn classifyXfer(method: http.Method, target: []const u8, content_type: ?[]const u8) Disposition {
    _ = target; // §4 L759-764 — dispatch is path-independent.
    if (method != .POST) return .method_not_allowed;
    const ct = content_type orelse return .unsupported_media_type;
    // A charset/parameter suffix (`; charset=…`) is tolerated; the base type
    // must name a media type this server can actually process on a request.
    // §4 r2/r3 marks both x-fossil and x-fossil-debug request-valid, but
    // handleXfer only decodes a zlib-wrapped (§4.1 outer_compressed) body:
    // x-fossil-debug is an UNCOMPRESSED diagnostic transport (r16 — not for
    // production, never emitted by the reference client) that zit does not yet
    // process, and x-fossil-uncompressed is reply-only. Refuse both with 415
    // here rather than accept and then fail confusingly at the decode step.
    // Supporting the uncompressed debug body is a separate change (thread
    // outer_compressed through answer→handleXfer to skip codec.decode).
    const base = if (std.mem.indexOfScalar(u8, ct, ';')) |s| std.mem.trimEnd(u8, ct[0..s], " ") else ct;
    const media = sync.parseMediaType(base) orelse return .unsupported_media_type;
    const props = sync.media_props.get(media);
    if (!props.valid_request_type or !props.outer_compressed) return .unsupported_media_type;
    // A servable request is always a POST carrying a request-valid, outer-
    // compressed media type — the only shape `handleXfer` can decode (§4
    // L759-764, §4.1).
    assert(method == .POST and props.valid_request_type and props.outer_compressed);
    return .ok;
}

fn statusFor(d: Disposition) http.Status {
    return switch (d) {
        .ok => .ok,
        .method_not_allowed => .method_not_allowed,
        .unsupported_media_type => .unsupported_media_type,
    };
}

// ---------------------------------------------------------------------------
// The socket server — the sole IO-bearing part.
// ---------------------------------------------------------------------------

/// The closed error set of the accept loop. Per-connection faults are handled
/// inside `serveConnection` and never reach here; only bring-up (`listen`) and
/// a non-transient `accept` failure abort the server.
pub const ServeError = error{ Listen, Accept };

/// Per-connection read/answer budget. `zit serve` binds loopback only
/// (`runServe` → 127.0.0.1), so every legitimate peer is a same-host process:
/// its request head and body (bounded by `max_request_body`) cross the kernel
/// loopback path in well under a second even at the 512 MiB ceiling, and the
/// answer is computed from an already-open log. A connection still unfinished
/// after this bound is therefore stalled, not slow — a byte-dribbling
/// slow-loris, a half-sent body, or an idle hold — and the single-threaded
/// accept loop (one connection served to completion before the next `accept`)
/// would otherwise let one such peer pin the server forever, the availability
/// failure `ci.yml` names. This is a whole-connection deadline, not a per-read
/// idle timeout: a per-read `SO_RCVTIMEO` cannot be used because the Io backend
/// treats a socket-timeout `EAGAIN` as a bug, so the deadline is enforced by
/// racing the request against a timed sleep and cancelling the blocked read when
/// the sleep wins. 5 s is ~1000× a normal loopback request and several× the
/// worst-case max-size loopback push, so it never fires on a real client. (A
/// per-read idle timeout, or an operator-tunable value, is a later refinement.)
pub const default_conn_read_timeout_ns: u64 = 5_000_000_000;

/// Bounds and knobs for `serve`. `max_requests` caps the accept loop (the test
/// seam and an explicit TigerStyle bound); null runs until the process is
/// signalled — a stateless server has nothing to flush, so SIGINT termination
/// is a clean shutdown. `conn_read_timeout_ns` bounds any single connection so a
/// stalled peer cannot pin the single-threaded accept loop.
pub const ServeOptions = struct {
    max_requests: ?u64 = null,
    conn_read_timeout_ns: u64 = default_conn_read_timeout_ns,
};

/// Bind `address` and serve sync requests single-threaded: accept, answer one
/// request per connection via `responder`, close, repeat. `respond` being
/// stateless, no state is shared across connections.
pub fn serve(
    io: std.Io,
    gpa: std.mem.Allocator,
    address: net.IpAddress,
    responder: Responder,
    opts: ServeOptions,
) ServeError!void {
    var listener = address.listen(io, .{ .reuse_address = true }) catch return error.Listen;
    defer listener.deinit(io);

    var served: u64 = 0;
    while (opts.max_requests == null or served < opts.max_requests.?) {
        const stream = listener.accept(io) catch |err| switch (err) {
            error.Canceled => return, // cooperative shutdown
            error.ConnectionAborted, error.WouldBlock => continue, // transient; not a served request
            else => return error.Accept,
        };
        served += 1;
        // The accept loop honors its explicit request bound (a TigerStyle
        // termination knob): a bounded server serves at most `max_requests`.
        assert(opts.max_requests == null or served <= opts.max_requests.?);
        serveConnection(io, gpa, stream, responder, opts.conn_read_timeout_ns);
    }
}

/// Handle exactly one HTTP request on `stream`, then close it, under the bounded
/// per-connection deadline `timeout_ns` (see `default_conn_read_timeout_ns`).
/// The request work runs on its own unit of concurrency, raced against a timed
/// sleep: whichever finishes first wins, and draining the loser cancels it —
/// cancelling the request interrupts the blocked socket read of a stalled peer,
/// so the single-threaded accept loop is freed. A per-connection fault (a
/// truncated client, a write failure, or the deadline) ends this connection
/// only — nothing propagates out. If the Io backend cannot supply concurrency,
/// the request is served without the deadline (correctness preserved; the bound
/// is hardening, not a protocol requirement). The `recv`/`send` buffers and the
/// `http.Server` live on this frame for the whole race — the drain completes
/// before they leave scope and before `defer stream.close` runs, so the task
/// never touches freed storage or a closed socket.
fn serveConnection(
    io: std.Io,
    gpa: std.mem.Allocator,
    stream: net.Stream,
    responder: Responder,
    timeout_ns: u64,
) void {
    defer stream.close(io);

    var recv_buf: [64 * 1024]u8 = undefined; // header buffer bound (§4 headers are small)
    var send_buf: [64 * 1024]u8 = undefined;
    var reader = stream.reader(io, &recv_buf);
    var writer = stream.writer(io, &send_buf);
    var server = http.Server.init(&reader.interface, &writer.interface);

    const Outcome = union(enum) { served: void, timed_out: void };
    var slots: [2]Outcome = undefined;
    var race = std.Io.Select(Outcome).init(io, &slots);
    race.concurrent(.served, handleOneRequest, .{ gpa, &server, responder }) catch {
        // No concurrency available: serve without the deadline bound.
        handleOneRequest(gpa, &server, responder);
        return;
    };
    race.concurrent(.timed_out, sleepDeadline, .{ io, timeout_ns }) catch {
        // The request is already running concurrently but the deadline task
        // could not be spawned; await the request unbounded.
        _ = race.await() catch {};
        race.cancelDiscard();
        return;
    };
    _ = race.await() catch {};
    race.cancelDiscard();
}

/// The raced request body: read one HTTP head and answer it. A separate Select
/// task so a stalled read/write is a cancellation point the connection deadline
/// can interrupt.
fn handleOneRequest(gpa: std.mem.Allocator, server: *http.Server, responder: Responder) void {
    var request = server.receiveHead() catch return;
    handleHttpRequest(gpa, &request, responder);
}

/// The connection deadline: sleep `ns`, then report the timeout. Cancelled to a
/// no-op early return whenever the request wins the race first.
fn sleepDeadline(io: std.Io, ns: u64) void {
    io.sleep(.{ .nanoseconds = @intCast(ns) }, .awake) catch {};
}

/// Route one parsed HTTP request: reject a non-sync request (non-POST, or a
/// POST without a request-valid Content-Type — §4 L759-764, any path) with the
/// classified status, else read the body (bounded by `max_request_body`), answer it, and
/// write back the framed reply as `application/x-fossil`. All failures collapse
/// to a status write; a dead connection is swallowed (nothing to recover).
fn handleHttpRequest(gpa: std.mem.Allocator, request: *http.Server.Request, responder: Responder) void {
    const head = request.head;
    const disposition = classifyXfer(head.method, head.target, head.content_type);
    if (disposition != .ok) {
        request.respond("", .{ .status = statusFor(disposition), .keep_alive = false }) catch {};
        return;
    }

    var body_buf: [16 * 1024]u8 = undefined;
    const body_reader = request.readerExpectContinue(&body_buf) catch {
        request.respond("", .{ .status = .expectation_failed, .keep_alive = false }) catch {};
        return;
    };
    const container = body_reader.allocRemaining(gpa, .limited(max_request_body)) catch |err| switch (err) {
        error.OutOfMemory => return respondStatus(request, .internal_server_error),
        error.StreamTooLong => return respondStatus(request, .payload_too_large), // §4 L1148-1149
        error.ReadFailed => return, // client vanished mid-body
    };
    defer gpa.free(container);
    // Postcondition of the `.limited` read: an accepted body never exceeds the
    // §4 L1148-1149 ceiling — `StreamTooLong` already rejected an over-cap send.
    assert(container.len <= max_request_body);

    const reply = responder.answer(gpa, container) catch |err| switch (err) {
        error.RequestTooLarge => return respondStatus(request, .payload_too_large),
        error.RequestMalformed => return respondStatus(request, .bad_request),
        error.ResponseTooLarge, error.OutOfMemory, error.ServerStorage => return respondStatus(request, .internal_server_error),
    };
    defer gpa.free(reply.body);

    // §4.1 — serve the body under the Content-Type the answer chose so the
    // framing and media type agree (`application/x-fossil`, or the §8.1 clone
    // reply's `application/x-fossil-uncompressed`).
    request.respond(reply.body, .{
        .status = .ok,
        .keep_alive = false,
        .extra_headers = &.{.{ .name = "content-type", .value = reply.content_type }},
    }) catch {};
}

fn respondStatus(request: *http.Server.Request, status: http.Status) void {
    request.respond("", .{ .status = status, .keep_alive = false }) catch {};
}

// ===========================================================================
// Tests — expected values come from sync §4/§4.1/§3.x, never from running
// fossil; the RFC text and its Appendix A worked examples are the oracle.
// ===========================================================================

const testing = std.testing;

/// A 64-hex artifact name (SHA-256 width) and its content, seeded into a
/// server store so a `gimme` for it is satisfiable (§3.3).
const held_hash = "deadbeef" ** 8;
const held_content = "hello from zit\n";

fn framed(gpa: std.mem.Allocator, plaintext: []const u8) ![]u8 {
    return codec.encode(gpa, plaintext, .default);
}

fn unframed(gpa: std.mem.Allocator, container: []const u8) ![]u8 {
    return codec.decode(gpa, container, max_request_body);
}

test "serve §3.3/§4.1: a framed gimme for held content returns a framed file card" {
    const gpa = testing.allocator;
    var ms = repostate.MemStore.init(gpa);
    defer ms.deinit();
    _ = try ms.store().put(held_hash, held_content);

    const server = sync.Server{ .version = server_version };
    // §3.2 — a bare (undated) client-version emits no server-version, so the
    // reply is exactly the file card for the requested, held hash.
    const request = try framed(gpa, "pragma client-version 20000\ngimme " ++ held_hash ++ "\n");
    defer gpa.free(request);

    const reply_container = try handleXfer(gpa, &server, ms.store(), request);
    defer gpa.free(reply_container.body);

    // §4.1 — an ordinary (non-clone) reply is the compressed container.
    try testing.expectEqualStrings("application/x-fossil", reply_container.content_type);

    const reply = try unframed(gpa, reply_container.body);
    defer gpa.free(reply);

    // §3.3 L848-852 — `file HASH LEN\n` + content, in request order.
    const expected = "file " ++ held_hash ++ " 15\n" ++ held_content;
    try testing.expectEqualStrings(expected, reply);
}

test "serve §3.8: the same request over the same store yields byte-identical replies (stateless)" {
    const gpa = testing.allocator;
    var ms = repostate.MemStore.init(gpa);
    defer ms.deinit();
    _ = try ms.store().put(held_hash, held_content);

    const server = sync.Server{ .version = server_version };
    const request = try framed(gpa, "pragma client-version 20000\ngimme " ++ held_hash ++ "\n");
    defer gpa.free(request);

    const first = try handleXfer(gpa, &server, ms.store(), request);
    defer gpa.free(first.body);
    const second = try handleXfer(gpa, &server, ms.store(), request);
    defer gpa.free(second.body);

    try testing.expectEqualSlices(u8, first.body, second.body);
    // §3.8 — a gimme-only request ingests nothing, so the store is untouched.
    try testing.expectEqual(@as(usize, 1), ms.store().count());
}

test "serve §8.1/§4.1: a clone reply is emitted with no outer zlib container" {
    const gpa = testing.allocator;
    var ms = repostate.MemStore.init(gpa);
    defer ms.deinit();
    _ = try ms.store().put(held_hash, held_content);
    const server = sync.Server{
        .version = server_version,
        .self_servercode = server_servercode,
        .project_code = server_projectcode,
    };
    // §8 — an accepting `clone 3 1`; an undated client-version means the reply
    // opens directly on the first cfile (§8.2), so the absence of an outer zlib
    // container is directly observable.
    const request = try framed(gpa, "pragma client-version 20000\nclone 3 1\n");
    defer gpa.free(request);

    const reply = try handleXfer(gpa, &server, ms.store(), request);
    defer gpa.free(reply.body);

    // §8.1 L1365 / §4.1 L829 — a version-3 clone reply is served as
    // application/x-fossil-uncompressed, never the compressed default.
    try testing.expectEqualStrings("application/x-fossil-uncompressed", reply.content_type);

    // §4.1 L829-831 — no outer compression: the body is the raw card stream, so
    // it opens with the first `cfile` card verbatim, not a 4-byte length prefix
    // followed by a zlib header. Decoding it as an outer container would fail —
    // proof it is single-, not double-wrapped.
    try testing.expect(std.mem.startsWith(u8, reply.body, "cfile "));

    // §7.2 — the cfile payload is itself the only compression: its inner §4.1
    // container decodes straight to the stored content (a double-wrapped reply
    // would need the whole body deflated first).
    const line = try sync.readLine(reply.body, 0);
    const cc = sync.classifyCard(line.bytes);
    try testing.expectEqual(sync.Operator.cfile, cc.operator.?);
    const cfile = try sync.parseCfileCard(cc.tokens);
    try testing.expectEqualStrings(held_hash, cfile.hash);
    const payload = try sync.consumeContentPayload(reply.body, line.next, cfile.compressed_size);
    const content = try codec.decode(gpa, payload.bytes, sync.reply_body_cap);
    defer gpa.free(content);
    try testing.expectEqualStrings(held_content, content);
}

test "serve §4.1: a truncated container (shorter than its length prefix) is rejected" {
    const gpa = testing.allocator;
    var ms = repostate.MemStore.init(gpa);
    defer ms.deinit();
    const server = sync.Server{ .version = server_version };
    try testing.expectError(error.RequestMalformed, handleXfer(gpa, &server, ms.store(), &.{ 0x00, 0x00 }));
}

test "serve §4.1: a valid length prefix over a non-zlib stream is rejected as malformed" {
    const gpa = testing.allocator;
    var ms = repostate.MemStore.init(gpa);
    defer ms.deinit();
    const server = sync.Server{ .version = server_version };
    // Declared length 5, then five bytes that are not an RFC1950 stream.
    const bad = [_]u8{ 0x00, 0x00, 0x00, 0x05 } ++ "xxxxx".*;
    try testing.expectError(error.RequestMalformed, handleXfer(gpa, &server, ms.store(), &bad));
}

test "serve §4 L1148-1149: a container whose declared length exceeds the cap is rejected before allocating" {
    const gpa = testing.allocator;
    var ms = repostate.MemStore.init(gpa);
    defer ms.deinit();
    const server = sync.Server{ .version = server_version };
    // A 4-GiB declared length over the 512-MiB cap: rejected on the length
    // field, never decompressed.
    const lying = [_]u8{ 0xFF, 0xFF, 0xFF, 0xFF } ++ "zzzz".*;
    try testing.expectError(error.RequestTooLarge, handleXfer(gpa, &server, ms.store(), &lying));
}

test "serve §4 L759-764: sync dispatches on method + Content-Type regardless of path" {
    // §4 L759-764 — a POST bearing a request-valid media type is sync at ANY
    // path; the client sends the URL path unchanged and MUST NOT append /xfer.
    // POST + x-fossil is sync at "/", "/xfer", and any other path alike.
    try testing.expectEqual(Disposition.ok, classifyXfer(.POST, "/", "application/x-fossil"));
    try testing.expectEqual(Disposition.ok, classifyXfer(.POST, "/xfer", "application/x-fossil"));
    try testing.expectEqual(Disposition.ok, classifyXfer(.POST, "/random/path", "application/x-fossil"));
    // A charset/parameter suffix is tolerated; a query string is not the path.
    try testing.expectEqual(Disposition.ok, classifyXfer(.POST, "/?ic=1", "application/x-fossil; charset=binary"));
    // §4 L761-762 marks application/x-fossil-debug request-valid, but it is an
    // UNCOMPRESSED (§4.1) diagnostic transport handleXfer cannot decode (r16 —
    // not for production, never emitted by the reference client), so this server
    // refuses it 415 rather than accept-then-fail. See the media-type gate note.
    try testing.expectEqual(Disposition.unsupported_media_type, classifyXfer(.POST, "/", "application/x-fossil-debug"));
    // §4 r2/r3, §4.1 — a wrong or missing Content-Type on a POST is 415, and
    // the reply-only x-fossil-uncompressed is not a valid request type.
    try testing.expectEqual(Disposition.unsupported_media_type, classifyXfer(.POST, "/", null));
    try testing.expectEqual(Disposition.unsupported_media_type, classifyXfer(.POST, "/xfer", "text/plain"));
    try testing.expectEqual(Disposition.unsupported_media_type, classifyXfer(.POST, "/", "application/x-fossil-uncompressed"));
    // §4 L762-763 — a non-POST request is not sync (405; see classifyXfer note
    // on "redirected home").
    try testing.expectEqual(Disposition.method_not_allowed, classifyXfer(.GET, "/", "application/x-fossil"));
    try testing.expectEqual(Disposition.method_not_allowed, classifyXfer(.GET, "/index", "text/html"));
    try testing.expectEqual(Disposition.method_not_allowed, classifyXfer(.PUT, "/xfer", "application/x-fossil"));
}

test "serve §4.1 fuzz: handleXfer on arbitrary request bytes stays bounded and never crashes" {
    const gpa = testing.allocator;
    var ms = repostate.MemStore.init(gpa);
    defer ms.deinit();
    _ = try ms.store().put(held_hash, held_content);
    const server = sync.Server{ .version = server_version };

    var prng = std.Random.DefaultPrng.init(0x5e2fe);
    const rand = prng.random();
    var buf: [256]u8 = undefined;
    var i: usize = 0;
    while (i < 512) : (i += 1) {
        const n = rand.intRangeAtMost(usize, 0, buf.len);
        rand.bytes(buf[0..n]);
        if (handleXfer(gpa, &server, ms.store(), buf[0..n])) |reply| {
            gpa.free(reply.body); // a run of bytes that happened to decode is fine
        } else |err| switch (err) {
            error.RequestMalformed, error.RequestTooLarge, error.ResponseTooLarge, error.OutOfMemory => {},
        }
    }
}

// --- socket-level smoke: drive the real std.http.Server over an in-memory
// stream pair (deterministic, no loopback port to flake in CI) ---------------

const TestResponder = struct {
    server: *const sync.Server,
    store: repostate.ArtifactStore,

    fn answer(ctx: *anyopaque, gpa: std.mem.Allocator, request_container: []const u8) AnswerError!XferReply {
        const self: *const TestResponder = @ptrCast(@alignCast(ctx));
        return handleXfer(gpa, self.server, self.store, request_container);
    }
};

/// Drive one HTTP request through the real `std.http.Server` request path over
/// in-memory reader/writer buffers, returning the raw HTTP response bytes.
///
/// `pub` because it is the request-path seam two callers need and neither
/// should reimplement: the smoke tests below, and the `serve_request_garbage`
/// fuzz target (serve_fuzz.zig), which pushes raw bytes through this same
/// receiveHead -> handleHttpRequest path. In-memory on both sides, so it binds
/// no port and cannot flake in CI. Caller owns the returned bytes.
pub fn driveHttp(gpa: std.mem.Allocator, raw_request: []const u8, responder: Responder) ![]u8 {
    var in = std.Io.Reader.fixed(raw_request);
    var out = std.Io.Writer.Allocating.init(gpa);
    defer out.deinit();
    var server = http.Server.init(&in, &out.writer);
    var request = try server.receiveHead();
    handleHttpRequest(gpa, &request, responder);
    return gpa.dupe(u8, out.written());
}

fn httpBody(response: []const u8) []const u8 {
    const sep = std.mem.indexOf(u8, response, "\r\n\r\n") orelse return &.{};
    return response[sep + 4 ..];
}

test "serve §4 smoke: a POST /xfer application/x-fossil round-trips a framed reply over HTTP" {
    const gpa = testing.allocator;
    var ms = repostate.MemStore.init(gpa);
    defer ms.deinit();
    _ = try ms.store().put(held_hash, held_content);
    const server = sync.Server{ .version = server_version };
    var ctx = TestResponder{ .server = &server, .store = ms.store() };
    const responder = Responder{ .ctx = &ctx, .answerFn = TestResponder.answer };

    const body = try framed(gpa, "pragma client-version 20000\ngimme " ++ held_hash ++ "\n");
    defer gpa.free(body);
    const req = try std.fmt.allocPrint(
        gpa,
        "POST /xfer HTTP/1.1\r\nHost: z\r\nContent-Type: application/x-fossil\r\nContent-Length: {d}\r\nConnection: close\r\n\r\n",
        .{body.len},
    );
    defer gpa.free(req);
    const whole = try std.mem.concat(gpa, u8, &.{ req, body });
    defer gpa.free(whole);

    const response = try driveHttp(gpa, whole, responder);
    defer gpa.free(response);

    try testing.expect(std.mem.startsWith(u8, response, "HTTP/1.1 200"));
    try testing.expect(std.mem.indexOf(u8, response, "application/x-fossil") != null);

    const reply = try unframed(gpa, httpBody(response));
    defer gpa.free(reply);
    try testing.expectEqualStrings("file " ++ held_hash ++ " 15\n" ++ held_content, reply);
}

test "serve §4 smoke: a GET on /xfer is rejected 405 over HTTP" {
    const gpa = testing.allocator;
    var ms = repostate.MemStore.init(gpa);
    defer ms.deinit();
    const server = sync.Server{ .version = server_version };
    var ctx = TestResponder{ .server = &server, .store = ms.store() };
    const responder = Responder{ .ctx = &ctx, .answerFn = TestResponder.answer };

    const response = try driveHttp(gpa, "GET /xfer HTTP/1.1\r\nHost: z\r\nConnection: close\r\n\r\n", responder);
    defer gpa.free(response);
    try testing.expect(std.mem.startsWith(u8, response, "HTTP/1.1 405"));
}