//! 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 L799-804) — 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 repeat = @import("zit").repeat;
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;
/// The request header a trusted front uses to hand this server the
/// branch globs whose membership the push may not change. Honoured only by a
/// `fronted` responder. Loopback plumbing between a front and its child; it is
/// not on the fossil wire and no client is expected to send it.
pub const protect_header = "x-zit-protect";
/// Bound on the header's value. A glob list, not a cookie line.
pub const protect_header_max: usize = 4096;
/// The request header a trusted front uses to allow a divert for this push:
/// the login it authenticated. Honoured only by a `fronted` responder, like
/// `protect_header`, so only a front can name the account a move is kept for.
pub const divert_login_header = "x-zit-divert-login";
pub const divert_login_max: usize = 256;
/// The response header that tells the front which check-ins a round kept on
/// attempt branches, so it never parses message text: one header per moved
/// run, `HASH BRANCH PROTECTED`, each field fossil-encoded as on the sync
/// wire. Only a fronted responder sends it.
pub const diverted_header = "x-zit-diverted";
/// Bound on the `x-zit-diverted` headers one reply carries.
pub const max_diverted_headers: usize = 64;
/// What a trusted front said about one request. Both fields are null unless
/// the responder is `fronted` and the front sent the header.
pub const Front = struct {
protect: ?[]const u8 = null,
divert_login: ?[]const u8 = null,
};
const net = std.Io.net;
/// §4 L1222-1228 / 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 L1225-1228 — 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 L1225-1228).
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.
/// `ServerBusy`: another writer held the repository's lock past the wait.
/// Answered 503, which tells the client to try again, not that the server
/// is broken.
pub const AnswerError = XferError || error{ ServerStorage, ServerBusy };
/// 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,
/// sync §10.3 r33 — the sub-manifest settings an AUTHORIZED pushed config
/// card asked the served repository to adopt, carried out of the sans-IO
/// responder for the caller that owns the repository's config to apply.
/// All-null on every other reply, including a rejected push.
config_apply: sync.SubmanifestConfig = .{},
/// The `x-zit-diverted` values for the front, one per line, or null when
/// nothing was diverted. Caller-owned, freed with `body`.
diverted: ?[]u8 = null,
};
/// 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, cookie: ?[]const u8, front: Front, request_compressed: bool) AnswerError!XferReply,
/// This server sits behind a trusted front on loopback and
/// honours that front's per-request `X-Zit-Protect` header (the branch
/// globs whose membership this push may not change). Off, the header is
/// ignored outright, so a `zit serve` reachable directly is the server it
/// always was: nothing a stranger can put on the wire reaches the rule.
fronted: bool = false,
/// Optional between-connections maintenance. When present, `serve` calls it
/// after each connection has closed and before the next accept — the one
/// point in the loop where no connection is open, so any O(store) work it does
/// (re-priming a snapshot a push left stale) falls outside every connection's
/// read deadline (the backlog mechanics that keep the next client waiting
/// meanwhile are in `listen`'s doc). Default null: a responder with no cache to
/// refresh (the tests, `driveHttp`) leaves it unset and the loop skips the call.
maintainFn: ?*const fn (ctx: *anyopaque, gpa: std.mem.Allocator) void = null,
/// 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, cookie: ?[]const u8, front: Front, request_compressed: bool) AnswerError!XferReply {
return self.answerFn(self.ctx, gpa, request_container, cookie, front, request_compressed);
}
};
// ---------------------------------------------------------------------------
// 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,
cookie: ?[]const u8,
request_compressed: bool,
) XferError!XferReply {
// §4 L1225-1228 — 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;
// x-fossil-debug carries the cards uncompressed; the reply is framed the
// same either way, and the reference client picks the reply's framing off
// the reply's own content type.
const request = if (request_compressed) 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,
} else gpa.dupe(u8, request_container) catch return error.OutOfMemory;
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, cookie, &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, .config_apply = reply.config_apply };
}
// 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, .config_apply = reply.config_apply };
}
// ---------------------------------------------------------------------------
// 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 L799-804 — 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 L799-804 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 L847), 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 L802-803 / `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 L800), 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 L799-804 — dispatch is path-independent.
if (method != .POST) return .method_not_allowed;
const ct = content_type orelse return .unsupported_media_type;
// The media type (parameters ignored, see sync.parseMediaType) must name
// one this server can process on a request: §4 r2/r3's x-fossil (a §4.1
// zlib container) or x-fossil-debug (the same cards, uncompressed). The
// reference client sends the debug type under --httptrace, the first
// thing anyone reaches for when a sync is misbehaving, so refusing it
// took the diagnostic away exactly when it was needed. Whether the body
// is a container is the media type's `outer_compressed`, which the
// exchange hands to `handleXfer`. x-fossil-uncompressed is reply-only.
const media = sync.parseMediaType(ct) orelse return .unsupported_media_type;
const props = sync.media_props.get(media);
if (!props.valid_request_type) return .unsupported_media_type;
assert(method == .POST and props.valid_request_type);
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.
// ---------------------------------------------------------------------------
/// 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,
};
/// Open the listener on `address` — the socket-bring-up half of serving, split
/// from the accept loop so the caller can do work between the two. The kernel
/// begins queueing inbound connections onto this listener's accept backlog the
/// moment it returns, before `serve` is ever called: a client that connects
/// during that window completes its TCP handshake against the backlog and simply
/// waits to be accepted. That is the whole point of the split — `runServe` binds,
/// truthfully reports "listening" (only reachable once the bind actually
/// succeeded), then does its unbudgeted O(store) prime while clients queue, and
/// only then enters `serve`. The returned `net.Server` is caller-owned; the
/// caller must `deinit` it. `reuse_address` matches the old inline bind. Named
/// for `listen`, not `bind`: the operation the fix's correctness rests on is the
/// listen that opens the accept backlog (and the error it returns is `Listen`),
/// not the address bind that precedes it.
pub fn listen(io: std.Io, address: net.IpAddress) error{Listen}!net.Server {
return address.listen(io, .{ .reuse_address = true }) catch error.Listen;
}
/// Serve sync requests single-threaded over an already-bound `listener`: accept,
/// answer one request per connection via `responder`, close, then — if the
/// responder has a maintenance hook — run it before the next accept, and repeat.
/// `respond` being stateless, no state is shared across connections; the only
/// cross-connection work is the optional `maintainFn`, run only when no
/// connection is open (see `Responder.maintainFn`). `listener` is borrowed, not
/// owned: the caller opened it via `listen` and closes it.
pub fn serve(
io: std.Io,
gpa: std.mem.Allocator,
listener: *net.Server,
responder: Responder,
opts: ServeOptions,
) error{Accept}!void {
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);
// The connection is now closed (its reply flushed, its deadline retired).
// This is the only instant in the loop with no open connection, so any
// O(store) maintenance runs here — its cost is never inside a connection's
// read budget (the accept backlog that holds the next client meanwhile
// is described in `listen`'s doc).
if (responder.maintainFn) |maintainFn| maintainFn(responder.ctx, gpa);
}
}
/// 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);
// THE DEADLINE BOUNDS TIME BLOCKED ON THE PEER, NOT OUR OWN WORK. It used
// to race the whole exchange -- read, answer and write as one unit -- which
// conflated "this peer has stalled" with "this answer is taking a while".
// A clone round that legitimately took longer than the deadline was
// cancelled mid-answer, and because the cancellation landed on an in-flight
// pread the peer was told `error.Storage`: the server timed out its own
// compute and reported it as a damaged repository.
//
// So the phases are raced separately. Reading the request and writing the
// reply are the points where a slow-loris can pin the single-threaded
// accept loop, and each gets the full deadline. Computing the answer is
// bounded by the store, not by the peer, and runs unraced -- no peer can
// extend it, so no peer can exploit it.
var st: Exchange = .{ .gpa = gpa, .server = &server, .responder = responder };
if (!racePhase(io, timeout_ns, readPhase, &st)) return; // stalled or gone
if (!st.have_request) return; // malformed head, or already answered a status
answerPhase(&st); // our own work: deliberately unbounded
const reply_len: usize = if (st.reply) |r| r.body.len else 0;
_ = racePhase(io, writeDeadlineNs(timeout_ns, reply_len), writePhase, &st);
if (st.reply) |r| {
gpa.free(r.body);
if (r.diverted) |d| gpa.free(d);
}
}
/// One request's state, threaded across the three phases so the read and write
/// halves can be raced against the deadline independently while the answer
/// between them is not.
const Exchange = struct {
gpa: std.mem.Allocator,
server: *http.Server,
responder: Responder,
request: http.Server.Request = undefined,
have_request: bool = false,
container: ?[]u8 = null,
reply: ?XferReply = null,
/// `receiveHead` failed — the head did not parse. The socket path treats
/// this as a clean close; `driveHttp` surfaces it, because its callers (the
/// fuzz lane) distinguish "rejected" from "answered".
head_failed: bool = false,
/// §6.5 — the request's x-f-l-c cookie value, copied out of the
/// head buffer; zero-length means the header was absent. Bounded by the
/// sync line ceiling, which also bounds every well-formed cookie value.
cookie_buf: [sync.max_line_len]u8 = undefined,
cookie_len: usize = 0,
/// The front's `X-Zit-Protect` value, lifted only when the
/// responder is `fronted`. A glob list, so far shorter than a cookie line.
protect_buf: [protect_header_max]u8 = undefined,
protect_len: usize = 0,
protect_set: bool = false,
/// The front's `X-Zit-Divert-Login`, lifted on the same terms.
divert_buf: [divert_login_max]u8 = undefined,
divert_len: usize = 0,
divert_set: bool = false,
/// The request media type's framing: false for x-fossil-debug, whose
/// cards arrive uncompressed.
request_compressed: bool = true,
};
/// One cookie's value out of a `Cookie:` header line (RFC 6265 §4.2.1
/// `cookie-pair *( "; " cookie-pair )`), or null when `name` is absent.
/// Matching is exact on the name; surrounding OWS is tolerated because real
/// agents emit it.
fn extractCookie(header: []const u8, name: []const u8) ?[]const u8 {
var it = std.mem.splitScalar(u8, header, ';');
while (it.next()) |raw_pair| {
const pair = std.mem.trim(u8, raw_pair, " \t");
const eq = std.mem.indexOfScalar(u8, pair, '=') orelse continue;
if (std.mem.eql(u8, pair[0..eq], name)) return pair[eq + 1 ..];
}
return null;
}
/// Run `phase` against the deadline. Returns false when the deadline won (the
/// peer stalled) or the phase could not be spawned concurrently, true when the
/// phase finished. A false return means the connection is not worth continuing.
fn racePhase(io: std.Io, timeout_ns: u64, comptime phase: fn (*Exchange) void, st: *Exchange) bool {
const Outcome = union(enum) { done: void, timed_out: void };
var slots: [2]Outcome = undefined;
var race = std.Io.Select(Outcome).init(io, &slots);
race.concurrent(.done, phase, .{st}) catch {
// No concurrency available: run it without the deadline bound, exactly
// as the previous single-race version degraded.
phase(st);
return true;
};
race.concurrent(.timed_out, sleepDeadline, .{ io, timeout_ns }) catch {
_ = race.await() catch {};
race.cancelDiscard();
return true;
};
const winner = race.await() catch {
race.cancelDiscard();
return false;
};
race.cancelDiscard();
return winner == .done;
}
/// The slowest peer the WRITE deadline tolerates. The flat deadline bounds a
/// peer that STALLS, but a peer legitimately draining a large reply needs
/// time proportional to the reply's size: §8.2's byte-budgeted clone batches
/// made replies ~4 MiB, and racing that against the flat 5 s alone imposed a
/// ≥6.4 Mbps floor on every healthy client — the same shape (a healthy
/// answer cancelled mid-flight) reintroduced from the write side. 256 KiB/s
/// (~2 Mbps) admits DSL-class links, while the worst pin a deliberately slow
/// reader can inflict on the single-threaded accept loop stays bounded by
/// the server's own chosen reply size (a 4 MiB batch: 5 s + 16 s).
pub const min_drain_bytes_per_sec: u64 = 256 * 1024;
/// The write-phase deadline: the flat stall bound plus drain time for
/// `reply_len` bytes at the minimum tolerated rate.
fn writeDeadlineNs(base_ns: u64, reply_len: usize) u64 {
const drain_ns = (@as(u64, reply_len) * std.time.ns_per_s) / min_drain_bytes_per_sec;
return base_ns + drain_ns;
}
/// 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 L799-804, 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).
/// Phase 1, raced against the deadline: receive the head and read the body.
/// Every blocking point here waits on the PEER, which is what the deadline is
/// for. A request rejected on classification or framing is answered here and
/// leaves `have_request` false so no later phase runs.
fn readPhase(st: *Exchange) void {
st.request = st.server.receiveHead() catch {
st.head_failed = true;
return;
};
const head = st.request.head;
// §6.5 — lift the x-f-l-c login cookie, the credential channel a
// ≥22700 client uses on every round. Copied out here because the head
// buffer's lifetime is the request's; absent stays null and the request
// authenticates by body login or not at all.
st.cookie_len = 0;
st.protect_len = 0;
st.protect_set = false;
var headers = st.request.iterateHeaders();
while (headers.next()) |h| {
if (std.ascii.eqlIgnoreCase(h.name, "cookie")) {
if (extractCookie(h.value, "x-f-l-c")) |value| {
if (value.len <= st.cookie_buf.len) {
@memcpy(st.cookie_buf[0..value.len], value);
st.cookie_len = value.len;
}
}
continue;
}
// The front's membership rule for this push. Read only when
// the responder declared a front; otherwise the header is any other
// header a stranger might send, and stays unread. An over-long value
// is a front bug and is refused rather than truncated: a truncated
// glob list protects less than the front asked for, silently.
if (st.responder.fronted and std.ascii.eqlIgnoreCase(h.name, protect_header)) {
if (h.value.len > st.protect_buf.len) {
st.request.respond("", .{ .status = .request_header_fields_too_large, .keep_alive = false }) catch {};
return;
}
@memcpy(st.protect_buf[0..h.value.len], h.value);
st.protect_len = h.value.len;
st.protect_set = true;
}
if (st.responder.fronted and std.ascii.eqlIgnoreCase(h.name, divert_login_header)) {
if (h.value.len > st.divert_buf.len) {
st.request.respond("", .{ .status = .request_header_fields_too_large, .keep_alive = false }) catch {};
return;
}
@memcpy(st.divert_buf[0..h.value.len], h.value);
st.divert_len = h.value.len;
st.divert_set = true;
}
}
const disposition = classifyXfer(head.method, head.target, head.content_type);
if (disposition != .ok) {
st.request.respond("", .{ .status = statusFor(disposition), .keep_alive = false }) catch {};
return;
}
// `.ok` proved the header names a request-valid media type.
st.request_compressed = sync.media_props.get(sync.parseMediaType(head.content_type.?).?).outer_compressed;
var body_buf: [16 * 1024]u8 = undefined;
const body_reader = st.request.readerExpectContinue(&body_buf) catch {
st.request.respond("", .{ .status = .expectation_failed, .keep_alive = false }) catch {};
return;
};
const container = body_reader.allocRemaining(st.gpa, .limited(max_request_body)) catch |err| switch (err) {
error.OutOfMemory => return respondStatus(&st.request, .internal_server_error),
error.StreamTooLong => return respondStatus(&st.request, .payload_too_large), // §4 L1225-1228
error.ReadFailed => return, // client vanished mid-body
};
// Postcondition of the `.limited` read: an accepted body never exceeds the
// §4 L1225-1228 ceiling — `StreamTooLong` already rejected an over-cap send.
assert(container.len <= max_request_body);
st.container = container;
st.have_request = true;
}
/// Phase 2, NOT raced: compute the answer. This is the server's own work over
/// its own store — no peer can make it longer, so bounding it by a
/// peer-stall deadline was only ever a way to fail healthy requests.
fn answerPhase(st: *Exchange) void {
const container = st.container orelse return;
defer st.gpa.free(container);
st.container = null;
const cookie: ?[]const u8 = if (st.cookie_len != 0) st.cookie_buf[0..st.cookie_len] else null;
const front: Front = .{
.protect = if (st.protect_set) st.protect_buf[0..st.protect_len] else null,
.divert_login = if (st.divert_set) st.divert_buf[0..st.divert_len] else null,
};
st.reply = st.responder.answer(st.gpa, container, cookie, front, st.request_compressed) catch |err| switch (err) {
error.RequestTooLarge => return respondStatus(&st.request, .payload_too_large),
error.RequestMalformed => return respondStatus(&st.request, .bad_request),
error.ResponseTooLarge, error.OutOfMemory, error.ServerStorage => return respondStatus(&st.request, .internal_server_error),
error.ServerBusy => return respondStatus(&st.request, .service_unavailable),
};
}
/// Phase 3, raced against the deadline: write the reply. Blocks on the peer
/// draining the socket, so a peer that stops reading cannot pin the loop.
///
/// §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`).
fn writePhase(st: *Exchange) void {
const reply = st.reply orelse return;
// One header per diverted run, bounded by the reply's own size cap on
// moves; past `max_diverted_headers` the rest are dropped rather than the
// reply refused, since the front can still read the store.
var headers: [1 + max_diverted_headers]http.Header = undefined;
headers[0] = .{ .name = "content-type", .value = reply.content_type };
var n: usize = 1;
if (reply.diverted) |d| {
var it = std.mem.splitScalar(u8, d, '\n');
while (it.next()) |line| {
if (n == headers.len) break;
headers[n] = .{ .name = diverted_header, .value = line };
n += 1;
}
}
st.request.respond(reply.body, .{
.status = .ok,
.keep_alive = false,
.extra_headers = headers[0..n],
}) 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 = repeat("deadbeef", 8);
const held_content = "hello from zit\n";
/// draft-fossil-submanifests-00 §7.2 — "A server whose enable-submanifests
/// setting (Section 8.1) is false sends the pragma avoid-submanifests, in the
/// same reply positions as avoid-delta-manifests." zit's §8.1 default is false,
/// so EVERY reply below opens with this card. Named once here so the tests that
/// assert whole reply bodies say why the card is there rather than carrying an
/// unexplained literal, and so turning the setting on in a future test is a
/// one-line change at each assertion instead of a re-transcription.
const avoid_submanifests_card = "pragma avoid-submanifests\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: an uncompressed x-fossil-debug request is answered exactly like a compressed one" {
// The reference client sends its cards uncompressed under --httptrace,
// typed application/x-fossil-debug. Refusing that type with 415 took the
// wire diagnostic away in the one situation it exists for.
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 plain = "pragma client-version 20000\ngimme " ++ held_hash ++ "\n";
const container = try framed(gpa, plain);
defer gpa.free(container);
const compressed = try handleXfer(gpa, &server, ms.store(), container, null, true);
defer gpa.free(compressed.body);
const debug = try handleXfer(gpa, &server, ms.store(), plain, null, false);
defer gpa.free(debug.body);
try testing.expectEqualStrings(compressed.content_type, debug.content_type);
try testing.expectEqualSlices(u8, compressed.body, debug.body);
try testing.expectEqual(Disposition.ok, classifyXfer(.POST, "/xfer", "application/x-fossil-debug"));
try testing.expectEqual(Disposition.ok, classifyXfer(.POST, "/xfer", "application/x-fossil"));
try testing.expectEqual(Disposition.unsupported_media_type, classifyXfer(.POST, "/xfer", "application/x-fossil-uncompressed"));
}
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, null, true);
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 L569-572 — `file HASH LEN\n` + content, in request order, behind the
// submanifests §7.2 pragma this server's §8.1 setting (false) requires.
const expected = avoid_submanifests_card ++ "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, null, true);
defer gpa.free(first.body);
const second = try handleXfer(gpa, &server, ms.store(), request, null, true);
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, null, true);
defer gpa.free(reply.body);
// §8.1 L1517-1518 / §4.1 L885-886 — 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 L885-887 — no outer compression: the body is the raw card stream, so
// its cards are readable 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. The clone sequence opens after the
// submanifests §7.2 pragma (§8.1 default false).
try testing.expect(std.mem.startsWith(u8, reply.body, avoid_submanifests_card));
const cards = reply.body[avoid_submanifests_card.len..];
try testing.expect(std.mem.startsWith(u8, cards, "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(cards, 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(cards, 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 }, null, true));
}
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, null, true));
}
test "serve §4 L1225-1228: 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, null, true));
}
test "serve §4 L799-804: sync dispatches on method + Content-Type regardless of path" {
// §4 L799-804 — 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 L801-802: application/x-fossil-debug is request-valid, and the
// reference client sends it under --httptrace; its body is uncompressed,
// which `handleXfer` is told by the exchange.
try testing.expectEqual(Disposition.ok, 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 L802-803 — 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], null, true)) |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, cookie: ?[]const u8, front: Front, request_compressed: bool) AnswerError!XferReply {
_ = cookie;
_ = request_compressed;
const self: *const TestResponder = @ptrCast(@alignCast(ctx));
// A lifted header becomes this round's rule, over a resolver
// that knows only the main branch: enough for a root check-in to
// resolve, which is all the header tests below need.
var round = self.server.*;
if (front.protect) |globs| {
round.push_policy.protect_globs = globs;
round.branch_of = main_only_resolver;
}
return handleXfer(gpa, &round, self.store, request_container, null, true);
}
};
/// Test seam: resolves nothing held, names `trunk` as main.
const main_only_resolver = sync.BranchResolver{
.ctx = @ptrCast(@constCast(&main_only_token)),
.branchOfFn = struct {
fn f(_: *anyopaque, _: []const u8) error{Unavailable}!?[]const u8 {
return null;
}
}.f,
.mainBranchFn = struct {
fn f(_: *anyopaque) error{Unavailable}![]const u8 {
return "trunk";
}
}.f,
.tagsOfFn = struct {
fn f(_: *anyopaque, _: std.mem.Allocator, _: []const u8) error{ Unavailable, OutOfMemory }!?[]repostate.RepoState.TagEntry {
return null;
}
}.f,
};
const main_only_token: u8 = 0;
/// 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);
// The same three phases the socket path runs, in order and unraced: this
// seam has no peer to stall, so there is no deadline to apply.
var st: Exchange = .{ .gpa = gpa, .server = &server, .responder = responder };
readPhase(&st);
if (st.head_failed) return error.ReadFailed;
if (st.have_request) {
answerPhase(&st);
writePhase(&st);
}
if (st.reply) |r| {
gpa.free(r.body);
if (r.diverted) |d| gpa.free(d);
}
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);
// The submanifests §7.2 pragma leads the reply (§8.1 default false).
try testing.expectEqualStrings(avoid_submanifests_card ++ "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"));
}
// --- socket-bring-up: the `listen`/`serve` split and the accept-loop maintenance
// hook. These are the deliberate IO-bearing exceptions to serve.zig's
// otherwise sans-IO unit surface (`driveHttp` above drives the request path over
// in-memory buffers). Each opens a real loopback listener, so they exercise the
// two seams the request-path smoke tests cannot: that `listen` opens the socket
// independently of — and before — the accept loop (so a client that connects
// during `runServe`'s pre-serve O(store) prime is queued, not refused, the original
// symptom), and that the accept loop invokes the responder's `maintainFn` between
// connections. They are single-threaded: a client connection is queued into the
// kernel accept backlog *before* `serve` runs, then a bounded `serve`
// (`max_requests`) drains it — no cross-thread coordination, and each binds an
// ephemeral port (0) read back from the socket, so there is no fixed port to
// flake on. ---------------------------------------------------------------------
/// The raw HTTP `POST /xfer` bytes for a framed `gimme` of `held_content` — a
/// complete, well-formed request the responder answers with a §4.1 file card.
/// Sending it in full (head + framed body) means `serveConnection` reads a whole
/// request and answers at once, never waiting on the per-connection deadline, so
/// these tests finish in milliseconds. Caller owns the returned bytes.
fn xferGimmeRequest(gpa: std.mem.Allocator) ![]u8 {
const body = try framed(gpa, "pragma client-version 20000\ngimme " ++ held_hash ++ "\n");
defer gpa.free(body);
const head = 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(head);
return std.mem.concat(gpa, u8, &.{ head, body });
}
/// Connect a loopback client to `port` and send `request` in full. The listener
/// must already be bound; on a bound listener this completes into the kernel
/// accept backlog even though nothing has accepted yet. Caller owns the returned
/// stream and must `close` it.
fn queueClientRequest(io: std.Io, port: u16, request: []const u8) !net.Stream {
const addr = net.IpAddress.parseIp4("127.0.0.1", port) catch return error.BadTestAddress;
const client = try addr.connect(io, .{ .mode = .stream });
errdefer client.close(io);
var wbuf: [1024]u8 = undefined;
var w = client.writer(io, &wbuf);
try w.interface.writeAll(request);
try w.interface.flush();
return client;
}
test "serve: listen opens the accept backlog before serve, so a connection queued during the prime window is served" {
const gpa = testing.allocator;
const io = std.testing.io;
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 };
// Bind to an ephemeral port (0): no fixed port to collide with another
// process, a prior run's lingering socket, or a concurrent test binary.
const addr = try net.IpAddress.parse("127.0.0.1", 0);
// The seam under test. `listen` opens the listening socket; its accept backlog
// starts queueing connections the instant it returns, BEFORE `serve` (the
// accept loop) is ever entered. In `runServe` the interval between this listen
// and entering `serve` is exactly where the O(store) prime runs — so a
// client that connects during the prime must complete its handshake into the
// backlog rather than be refused. Before the fix the socket opened only after the
// prime finished, and a large-repo probe that connected during it timed out
// with "server did not come up".
var listener = try listen(io, addr);
defer listener.deinit(io);
// The kernel's chosen port, read back from the bound socket.
const port = listener.socket.address.getPort();
// Connect and send a COMPLETE request NOW — before `serve` is entered. This
// is the connection that, in the bug, arrives during the prime. It completes
// into the backlog against the already-open socket; nothing has accepted it.
const request = try xferGimmeRequest(gpa);
defer gpa.free(request);
const client = try queueClientRequest(io, port, request);
defer client.close(io);
// Only now enter the accept loop. It must drain the connection already waiting
// in the backlog — proof the socket was open and queueing before the loop ran.
// `max_requests = 1` returns the loop after that single connection.
try serve(io, gpa, &listener, responder, .{ .max_requests = 1 });
// The connection queued before `serve` got a real §4.1 reply: a 200 with the
// file card. (The reply bytes were written into the socket before the server
// closed its side; the client reads them here.)
var rbuf: [1024]u8 = undefined;
var r = client.reader(io, &rbuf);
const status = try r.interface.take(12);
try testing.expectEqualStrings("HTTP/1.1 200", status);
}
/// A `Responder` that also counts how many times the accept loop invoked its
/// between-connections maintenance hook — the observable for the loop's
/// `maintainFn` wiring (serve.zig, the call after `serveConnection`).
const CountingResponder = struct {
server: *const sync.Server,
store: repostate.ArtifactStore,
maintain_calls: u64 = 0,
fn answer(ctx: *anyopaque, gpa: std.mem.Allocator, request_container: []const u8, cookie: ?[]const u8, front: Front, request_compressed: bool) AnswerError!XferReply {
_ = cookie;
_ = front;
const self: *CountingResponder = @ptrCast(@alignCast(ctx));
return handleXfer(gpa, self.server, self.store, request_container, null, request_compressed);
}
fn maintain(ctx: *anyopaque, gpa: std.mem.Allocator) void {
_ = gpa;
const self: *CountingResponder = @ptrCast(@alignCast(ctx));
self.maintain_calls += 1;
}
};
test "serve: the accept loop runs the responder's maintainFn between connections" {
const gpa = testing.allocator;
const io = std.testing.io;
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 = CountingResponder{ .server = &server, .store = ms.store() };
const responder = Responder{
.ctx = &ctx,
.answerFn = CountingResponder.answer,
.maintainFn = CountingResponder.maintain,
};
// Ephemeral port (0), read back from the bound socket — no fixed port to flake on.
const addr = try net.IpAddress.parse("127.0.0.1", 0);
var listener = try listen(io, addr);
defer listener.deinit(io);
const port = listener.socket.address.getPort();
// Queue two complete requests into the backlog before the loop runs, so the
// bounded `serve` accepts and answers exactly two connections back to back.
const request = try xferGimmeRequest(gpa);
defer gpa.free(request);
const c0 = try queueClientRequest(io, port, request);
defer c0.close(io);
const c1 = try queueClientRequest(io, port, request);
defer c1.close(io);
try serve(io, gpa, &listener, responder, .{ .max_requests = 2 });
// The loop calls `maintainFn` after each connection closes and before the
// next accept — the one instant with no open connection, where the O(store)
// rebuild is safe. Two served connections ⇒ two maintenance calls. Without the
// loop's `if (responder.maintainFn) |m| m(...)` this stays 0: the seam that
// connects `ServeCtx.maintain` to the running server would be dead.
try testing.expectEqual(@as(u64, 2), ctx.maintain_calls);
}
/// A `Responder` whose repository stayed locked by another writer.
const BusyResponder = struct {
fn answer(ctx: *anyopaque, gpa: std.mem.Allocator, request_container: []const u8, cookie: ?[]const u8, front: Front, request_compressed: bool) AnswerError!XferReply {
_ = .{ ctx, gpa, request_container, cookie, front, request_compressed };
return error.ServerBusy;
}
};
test "serve: a repository held by another writer is answered 503, not 500" {
const gpa = testing.allocator;
const io = std.testing.io;
var ctx: u8 = 0;
const responder = Responder{ .ctx = &ctx, .answerFn = BusyResponder.answer };
const addr = try net.IpAddress.parse("127.0.0.1", 0);
var listener = try listen(io, addr);
defer listener.deinit(io);
const port = listener.socket.address.getPort();
const request = try xferGimmeRequest(gpa);
defer gpa.free(request);
const client = try queueClientRequest(io, port, request);
defer client.close(io);
try serve(io, gpa, &listener, responder, .{ .max_requests = 1 });
// 503 tells the client the push can be tried again; 500 said the server
// was broken.
var rbuf: [1024]u8 = undefined;
var r = client.reader(io, &rbuf);
const status = try r.interface.take(12);
try testing.expectEqualStrings("HTTP/1.1 503", status);
}
// The header is a front's channel, not a client's.
fn protectHttpRequest(gpa: std.mem.Allocator, body: []const u8, extra_header: []const u8) ![]u8 {
return std.fmt.allocPrint(
gpa,
"POST /xfer HTTP/1.1\r\nHost: z\r\nContent-Type: application/x-fossil\r\nContent-Length: {d}\r\n{s}Connection: close\r\n\r\n{s}",
.{ body.len, extra_header, body },
);
}
const protect_artifact = zit.artifact;
const protect_root_body = "C root\nD 2026-08-26T00:00:00\nF a.txt 0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef\nU u\n";
/// `protect_root_body` sealed with its Z card, which the import parser requires.
const protect_root = protect_root_body ++ "Z " ++ md5Hex(protect_root_body) ++ "\n";
fn md5Hex(comptime body: []const u8) [32]u8 {
var digest: [16]u8 = undefined;
std.crypto.hash.Md5.hash(body, &digest, .{});
var hex: [32]u8 = undefined;
const lower = "0123456789abcdef";
for (digest, 0..) |b, i| {
hex[i * 2] = lower[b >> 4];
hex[i * 2 + 1] = lower[b & 0x0f];
}
return hex;
}
/// A responder that only remembers what the front said, and answers with a
/// fixed body carrying an `x-zit-diverted` value so the header's way out is
/// exercised too.
const FrontEcho = struct {
login_buf: [divert_login_max]u8 = undefined,
login: ?[]const u8 = null,
fn answer(ctx: *anyopaque, gpa: std.mem.Allocator, request_container: []const u8, cookie: ?[]const u8, front: Front, request_compressed: bool) AnswerError!XferReply {
_ = .{ request_container, cookie, request_compressed };
const self: *FrontEcho = @ptrCast(@alignCast(ctx));
if (front.divert_login) |l| {
@memcpy(self.login_buf[0..l.len], l);
self.login = self.login_buf[0..l.len];
}
const body = codec.encode(gpa, "", .default) catch return error.OutOfMemory;
errdefer gpa.free(body);
const diverted = gpa.dupe(u8, "h attempt/jan/h trunk") catch return error.OutOfMemory;
return .{ .body = body, .content_type = xfer_media_type, .diverted = diverted };
}
};
test "serve: X-Zit-Divert-Login reaches only a fronted responder, an over-long one is 431, and the divert header goes back" {
const gpa = testing.allocator;
const body = try framed(gpa, "pull a b\n");
defer gpa.free(body);
const sent = "X-Zit-Divert-Login: jan\r\n";
{
var echo: FrontEcho = .{};
const responder = Responder{ .ctx = &echo, .answerFn = FrontEcho.answer, .fronted = true };
const req = try protectHttpRequest(gpa, body, sent);
defer gpa.free(req);
const response = try driveHttp(gpa, req, responder);
defer gpa.free(response);
try testing.expect(std.mem.startsWith(u8, response, "HTTP/1.1 200"));
try testing.expectEqualStrings("jan", echo.login.?);
try testing.expect(std.ascii.findIgnoreCase(response, "x-zit-diverted: h attempt/jan/h trunk\r\n") != null);
}
{
var echo: FrontEcho = .{};
const responder = Responder{ .ctx = &echo, .answerFn = FrontEcho.answer };
const req = try protectHttpRequest(gpa, body, sent);
defer gpa.free(req);
const response = try driveHttp(gpa, req, responder);
defer gpa.free(response);
try testing.expect(echo.login == null);
}
{
var echo: FrontEcho = .{};
const responder = Responder{ .ctx = &echo, .answerFn = FrontEcho.answer, .fronted = true };
const long = try gpa.alloc(u8, divert_login_max + 1);
defer gpa.free(long);
@memset(long, 'x');
const header = try std.fmt.allocPrint(gpa, "X-Zit-Divert-Login: {s}\r\n", .{long});
defer gpa.free(header);
const req = try protectHttpRequest(gpa, body, header);
defer gpa.free(req);
const response = try driveHttp(gpa, req, responder);
defer gpa.free(response);
try testing.expect(std.mem.startsWith(u8, response, "HTTP/1.1 431"));
try testing.expect(echo.login == null);
}
}
test "serve: X-Zit-Protect is honoured only by a fronted responder, and an over-long value is 431" {
const gpa = testing.allocator;
var hb: [64]u8 = undefined;
const root_hash = protect_artifact.artifactName(.sha3_256, protect_root, &hb);
const cards = try std.fmt.allocPrint(gpa, "push {s} {s}\nfile {s} {d}\n{s}", .{ repeat("a", 40), repeat("b", 40), root_hash, protect_root.len, protect_root });
defer gpa.free(cards);
const body = try framed(gpa, cards);
defer gpa.free(body);
const server = sync.Server{ .version = server_version };
// Fronted + header: the root would join trunk; refused, store unchanged,
// and the reply is the ordinary error card naming the branch.
{
var ms = repostate.MemStore.init(gpa);
defer ms.deinit();
var ctx = TestResponder{ .server = &server, .store = ms.store() };
const responder = Responder{ .ctx = &ctx, .answerFn = TestResponder.answer, .fronted = true };
const req = try protectHttpRequest(gpa, body, "X-Zit-Protect: trunk\r\n");
defer gpa.free(req);
const response = try driveHttp(gpa, req, responder);
defer gpa.free(response);
try testing.expect(std.mem.startsWith(u8, response, "HTTP/1.1 200"));
const reply = try codec.decode(gpa, httpBody(response), sync.reply_body_cap);
defer gpa.free(reply);
try testing.expect(std.mem.startsWith(u8, reply, "error "));
try testing.expect(std.mem.indexOf(u8, reply, "'trunk'") != null);
try testing.expectEqual(@as(usize, 0), ms.store().count());
}
// Not fronted + the same header: ignored. The push lands.
{
var ms = repostate.MemStore.init(gpa);
defer ms.deinit();
var ctx = TestResponder{ .server = &server, .store = ms.store() };
const responder = Responder{ .ctx = &ctx, .answerFn = TestResponder.answer };
const req = try protectHttpRequest(gpa, body, "X-Zit-Protect: trunk\r\n");
defer gpa.free(req);
const response = try driveHttp(gpa, req, responder);
defer gpa.free(response);
try testing.expect(std.mem.startsWith(u8, response, "HTTP/1.1 200"));
try testing.expect(ms.store().lookup(root_hash) != null);
}
// Fronted, no header: nothing to enforce. The push lands.
{
var ms = repostate.MemStore.init(gpa);
defer ms.deinit();
var ctx = TestResponder{ .server = &server, .store = ms.store() };
const responder = Responder{ .ctx = &ctx, .answerFn = TestResponder.answer, .fronted = true };
const req = try protectHttpRequest(gpa, body, "");
defer gpa.free(req);
const response = try driveHttp(gpa, req, responder);
defer gpa.free(response);
try testing.expect(std.mem.startsWith(u8, response, "HTTP/1.1 200"));
try testing.expect(ms.store().lookup(root_hash) != null);
}
// Fronted + a value past the bound: refused outright, never truncated.
{
var ms = repostate.MemStore.init(gpa);
defer ms.deinit();
var ctx = TestResponder{ .server = &server, .store = ms.store() };
const responder = Responder{ .ctx = &ctx, .answerFn = TestResponder.answer, .fronted = true };
const long = try gpa.alloc(u8, protect_header_max + 1);
defer gpa.free(long);
@memset(long, 'x');
const header = try std.fmt.allocPrint(gpa, "X-Zit-Protect: {s}\r\n", .{long});
defer gpa.free(header);
const req = try protectHttpRequest(gpa, body, header);
defer gpa.free(req);
const response = try driveHttp(gpa, req, responder);
defer gpa.free(response);
try testing.expect(std.mem.startsWith(u8, response, "HTTP/1.1 431"));
try testing.expectEqual(@as(usize, 0), ms.store().count());
}
}