Something went wrong. Try again.
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948949950951952953954955956957958959960961962963964965966967968969970971972973974975976977978979980981982983984985986987988989990991992993994995996997998999100010011002100310041005100610071008100910101011101210131014101510161017101810191020102110221023102410251026102710281029103010311032103310341035103610371038103910401041104210431044104510461047104810491050105110521053105410551056105710581059106010611062106310641065106610671068106910701071107210731074107510761077107810791080108110821083108410851086108710881089109010911092109310941095109610971098109911001101110211031104110511061107110811091110111111121113111411151116111711181119112011211122112311241125112611271128112911301131113211331134113511361137113811391140114111421143114411451146114711481149115011511152115311541155115611571158115911601161116211631164116511661167116811691170117111721173117411751176117711781179118011811182118311841185118611871188118911901191119211931194119511961197119811991200120112021203120412051206120712081209121012111212121312141215121612171218121912201221122212231224122512261227122812291230123112321233123412351236123712381239124012411242124312441245124612471248124912501251125212531254125512561257125812591260126112621263126412651266126712681269127012711272127312741275127612771278127912801281128212831284128512861287128812891290129112921293129412951296129712981299130013011302130313041305130613071308130913101311131213131314131513161317131813191320132113221323132413251326132713281329133013311332133313341335133613371338133913401341134213431344134513461347134813491350135113521353135413551356135713581359136013611362136313641365136613671368136913701371137213731374137513761377137813791380138113821383138413851386138713881389139013911392139313941395139613971398139914001401140214031404140514061407140814091410141114121413141414151416141714181419142014211422142314241425142614271428142914301431143214331434143514361437143814391440144114421443144414451446144714481449145014511452145314541455145614571458145914601461146214631464146514661467146814691470147114721473147414751476147714781479148014811482148314841485148614871488148914901491149214931494149514961497149814991500150115021503150415051506150715081509151015111512151315141515151615171518151915201521152215231524152515261527152815291530153115321533153415351536153715381539154015411542154315441545154615471548154915501551155215531554155515561557155815591560156115621563156415651566156715681569157015711572157315741575157615771578157915801581158215831584158515861587158815891590159115921593159415951596159715981599160016011602160316041605160616071608160916101611161216131614161516161617161816191620162116221623162416251626162716281629163016311632163316341635163616371638163916401641164216431644164516461647164816491650165116521653165416551656165716581659166016611662166316641665166616671668166916701671167216731674167516761677167816791680168116821683168416851686168716881689169016911692169316941695169616971698169917001701170217031704170517061707170817091710171117121713171417151716171717181719172017211722172317241725172617271728172917301731173217331734173517361737173817391740174117421743174417451746174717481749175017511752175317541755175617571758175917601761176217631764176517661767176817691770177117721773177417751776177717781779178017811782178317841785178617871788178917901791179217931794179517961797179817991800180118021803180418051806180718081809181018111812181318141815181618171818181918201821182218231824182518261827182818291830183118321833183418351836183718381839184018411842184318441845184618471848184918501851185218531854185518561857185818591860186118621863186418651866186718681869187018711872187318741875187618771878187918801881188218831884188518861887188818891890189118921893189418951896189718981899190019011902190319041905190619071908190919101911191219131914191519161917191819191920192119221923192419251926192719281929193019311932193319341935193619371938193919401941194219431944194519461947194819491950195119521953195419551956195719581959196019611962196319641965196619671968196919701971197219731974197519761977197819791980198119821983198419851986198719881989199019911992199319941995199619971998199920002001200220032004200520062007200820092010201120122013201420152016201720182019202020212022202320242025202620272028202920302031203220332034203520362037203820392040204120422043204420452046204720482049205020512052205320542055205620572058205920602061206220632064206520662067206820692070207120722073207420752076207720782079208020812082208320842085208620872088208920902091209220932094209520962097209820992100210121022103210421052106210721082109211021112112211321142115211621172118211921202121212221232124212521262127212821292130213121322133213421352136213721382139214021412142214321442145214621472148214921502151215221532154215521562157215821592160216121622163216421652166216721682169217021712172217321742175217621772178217921802181218221832184218521862187218821892190219121922193219421952196219721982199220022012202220322042205220622072208220922102211221222132214221522162217221822192220222122222223222422252226222722282229223022312232223322342235223622372238223922402241224222432244224522462247224822492250225122522253225422552256225722582259226022612262226322642265226622672268226922702271227222732274227522762277227822792280228122822283228422852286228722882289229022912292229322942295229622972298229923002301230223032304230523062307230823092310231123122313231423152316231723182319232023212322232323242325232623272328232923302331233223332334233523362337233823392340234123422343234423452346234723482349235023512352235323542355235623572358235923602361236223632364236523662367236823692370237123722373237423752376237723782379238023812382238323842385238623872388238923902391239223932394239523962397239823992400240124022403240424052406240724082409241024112412241324142415241624172418241924202421242224232424242524262427242824292430243124322433243424352436243724382439244024412442244324442445244624472448244924502451245224532454245524562457245824592460//! A hardened HTTP client over std.http.Client.//!//! `Client.fetch` runs one request under a response-head deadline, a body//! stall guard, and dead-connection replay for idempotent requests, and//! identifies itself with a real User-Agent. Requires std.Io (zig 0.16+).
const std = @import("std");const build_options = @import("build_options");
const log = std.log.scoped(.zttp);
/// Sent unless a caller overrides `user_agent`. std.http's default is/// `zig/<version> (std.http)`, which tells a server operator nothing about who/// is making the request or how to reach them. Anything calling other people's/// servers at scale should say what it is: operators seeing unexplained load/// have no way to ask about it otherwise.pub const default_user_agent = "zttp/" ++ build_options.version ++ " (+https://tangled.org/zat.dev/zttp)";
/// Failures this client adds on top of std.http's own. `fetch` returns these/// or a std.http / std.Io error; callers classifying errors can switch on/// this set and treat the rest as transport failures.pub const Error = error{ /// No response head within `head_deadline_ns`. Nothing was received, so /// an idempotent request is safe to retry elsewhere. ResponseHeadTimeout, /// No body bytes for `StallGuard.idle_ns`. TransferStalled, /// Body rate under `StallGuard.min_bytes_per_s` after the grace period. TransferTooSlow, /// `FetchOptions.deadline_ns` elapsed. The request may have taken effect. RequestDeadlineExceeded, /// Body exceeded `max_response_size` or `StreamOptions.max_bytes`. ResponseTooLarge, /// `Response.stream` was called twice. BodyAlreadyTaken, /// `accept` plus `extra_headers` exceed the header buffer. TooManyHeaders, /// A payload was given without a `content_type`. MissingContentType, /// The URL contains a control character. std would write it to the /// request line verbatim, which is header injection when any part of /// the URL came from outside. InvalidUrl, /// A header name or value contains CR, LF, or (for names) a colon. InvalidHeader, /// `resolved_connection.logical_host` does not match the URL host. ResolvedHostMismatch, /// More hops than `RedirectPolicy.max_hops`. TooManyRedirects, /// A redirect target was already visited in this chain. RedirectLoop, /// An https origin redirected to http and downgrades are not allowed. SchemeDowngrade, /// A 3xx arrived and `on_redirect` is `.refuse`. RedirectRefused, /// A 3xx status this client does not know how to follow. HttpRedirectUnsupported, HttpRedirectLocationMissing, HttpRedirectLocationInvalid,};
/// Concurrent `fetch` calls on one client are fine: the connection pool and/// CA bundle take their own locks. The configuration fields are plain/// values; set them before sharing the client, and use the per-request/// overrides on `FetchOptions` after that.pub const Client = struct { allocator: std.mem.Allocator, io: std.Io, http_client: std.http.Client, tls_connect_lock: std.Io.Mutex = .init, keep_alive: bool = true, /// Default cap on a buffered response body; `FetchOptions.max_response_size` /// overrides it per request. null is unbounded. max_response_size: ?usize = null, /// Guards against a peer that accepts the connection and then trickles or /// stops. Without these a single slow origin holds a worker for as long as /// the caller's own deadline allows, and a pool of workers can sit "busy" /// while moving almost no bytes. Set any field to 0 to disable that check. stall: StallGuard = .{}, /// Bounds connect, TLS, request send, and the wait for the response head — /// the phases StallGuard cannot see, because no body bytes exist yet. A /// peer that accepts and never answers, or a SYN that goes unanswered, /// otherwise blocks the caller until the kernel gives up (~2 minutes) or /// forever on an established socket. 0 disables. head_deadline_ns: u64 = 10 * std.time.ns_per_s, /// A pooled connection the origin closed while it sat idle looks identical /// to a request that failed, but nothing was ever delivered — so replaying /// an idempotent request on a fresh connection is safe, and skipping the /// replay turns routine housekeeping into a caller-visible error. dead_connection_attempts: usize = 3, /// How `fetch` follows 3xx responses. `authorization` never crosses an /// origin boundary regardless of these settings. redirects: RedirectPolicy = .{}, /// Identify this client to the servers it calls. Applications should set /// their own; the default names this library so an operator can at least /// tell what is calling. user_agent: []const u8 = default_user_agent,
pub fn init(io: std.Io, allocator: std.mem.Allocator) Client { return .{ .allocator = allocator, .io = io, .http_client = .{ .allocator = allocator, .io = io }, }; }
/// Same as `init`, naming the calling application rather than this library. pub fn initWithUserAgent(io: std.Io, allocator: std.mem.Allocator, user_agent: []const u8) Client { var transport = init(io, allocator); if (user_agent.len > 0) transport.user_agent = user_agent; return transport; }
pub fn deinit(self: *Client) void { self.http_client.deinit(); }
/// fetch a URL and write response to provided writer /// Open, stream the whole body into memory, close. The one-call shape /// for bounded responses; `open` is the same request with the body left /// to the caller. pub fn fetch(self: *Client, options: FetchOptions) !FetchResult { const response = try self.open(options); defer response.deinit();
var aw: std.Io.Writer.Allocating = .init(self.allocator); defer aw.deinit(); _ = try response.stream(&aw.writer, .{ .max_bytes = options.max_response_size orelse self.max_response_size, .stall = options.stall, });
const body = try self.allocator.dupe(u8, aw.written()); var result: FetchResult = .{ .status = response.status, .body = body, .rate_limit = response.rate_limit, .headers = response.headers, .final_url = response.final_url, }; _ = &result; // Ownership of the captured headers and url moves to the result. response.headers = .{}; response.final_url = null; return result; }
/// Send the request and return once the response head is in. Every /// guard on the request phase applies: the head deadline, dead-connection /// replay, the redirect policy, input validation. The body is untouched /// until `Response.stream`; `Response.deinit` closes it either way. pub fn open(self: *Client, options: FetchOptions) !*Response { if (options.payload != null and options.content_type == null) return error.MissingContentType; for (options.url) |c| if (c < 0x20 or c == 0x7f) return error.InvalidUrl; if (options.authorization) |v| try validateHeaderValue(v); if (options.accept) |v| try validateHeaderValue(v); if (options.accept_encoding) |v| try validateHeaderValue(v); if (options.content_type) |v| try validateHeaderValue(v); if (options.extra_headers) |hdrs| for (hdrs) |h| { if (h.name.len == 0) return error.InvalidHeader; for (h.name) |c| if (c == ':' or c == '\r' or c == '\n') return error.InvalidHeader; try validateHeaderValue(h.value); };
var extra_buf: [8]std.http.Header = undefined; var extra_count: usize = 0;
if (options.accept) |accept| { extra_buf[extra_count] = .{ .name = "accept", .value = accept }; extra_count += 1; }
if (options.extra_headers) |hdrs| { if (extra_count + hdrs.len > extra_buf.len) return error.TooManyHeaders; for (hdrs) |h| { extra_buf[extra_count] = h; extra_count += 1; } }
const extra = extra_buf[0..extra_count]; var attempt: usize = 0; while (true) { attempt += 1; const result = self.openDeadlined(options, extra);
return result catch |err| { if (attempt < self.dead_connection_attempts and isDeadConnection(err) and isReplayable(options)) continue; return err; }; } }
const HeadProgress = struct { done: std.atomic.Value(bool) = .init(false), };
/// Run one attempt under the head deadline. The attempt runs as a task; /// the watcher cancels it if the head has not arrived in time. Whatever /// the request deadline has left is handed to the response for the body. fn openDeadlined( self: *Client, options: FetchOptions, extra_headers: []const std.http.Header, ) !*Response { var progress: HeadProgress = .{}; const head_deadline_ns = options.head_deadline_ns orelse self.head_deadline_ns; const total_deadline_ns = options.deadline_ns orelse 0; const deadline_ns = if (head_deadline_ns == 0) total_deadline_ns else if (total_deadline_ns == 0) head_deadline_ns else @min(head_deadline_ns, total_deadline_ns); if (deadline_ns == 0) return self.openOnce(options, extra_headers, &progress);
var future = self.io.concurrent(openOnce, .{ self, options, extra_headers, &progress }) catch |err| switch (err) { error.ConcurrencyUnavailable => { warnUnguarded(); const response = try self.openOnce(options, extra_headers, &progress); response.deadline_remaining_ns = total_deadline_ns; return response; }, };
const tick_ns: u64 = 10 * std.time.ns_per_ms; var elapsed_ns: u64 = 0; while (!progress.done.load(.acquire)) { if (elapsed_ns >= deadline_ns) { if (future.cancel(self.io)) |completed| completed.deinit() else |_| {} return if (total_deadline_ns != 0 and elapsed_ns >= total_deadline_ns) error.RequestDeadlineExceeded else error.ResponseHeadTimeout; } std.Io.Clock.Duration.sleep( .{ .raw = .fromNanoseconds(tick_ns), .clock = .awake }, self.io, ) catch { if (future.cancel(self.io)) |completed| completed.deinit() else |_| {} return error.Canceled; }; elapsed_ns += tick_ns; }
const response = try future.await(self.io); if (total_deadline_ns != 0) response.deadline_remaining_ns = total_deadline_ns -| elapsed_ns; return response; }
fn openOnce( self: *Client, options: FetchOptions, extra_headers: []const std.http.Header, progress: *HeadProgress, ) !*Response { defer progress.done.store(true, .release); const response = try self.allocator.create(Response); errdefer self.allocator.destroy(response); response.* = .{ .client = self, .request = undefined, .inner = undefined }; if (options.resolved_connection) |resolved| try self.openResolved(response, options, resolved, extra_headers) else try self.openUrl(response, options, extra_headers); return response; }
/// One hop of a request: everything about it that a redirect may rewrite. const Hop = struct { url: []const u8, method: std.http.Method, payload: ?[]const u8, content_type: ?[]const u8, authorization: ?[]const u8, accept_encoding: ?[]const u8,
fn fromOptions(options: FetchOptions) Hop { return .{ .url = options.url, .method = options.method, .payload = options.payload, .content_type = options.content_type, .authorization = options.authorization, .accept_encoding = options.accept_encoding, }; } };
fn requestHeaders(self: *Client, hop: Hop) std.http.Client.Request.Headers { return .{ .content_type = if (hop.content_type) |ct| .{ .override = ct } else .default, .authorization = if (hop.authorization) |auth| .{ .override = auth } else .default, .accept_encoding = if (hop.accept_encoding) |enc| .{ .override = enc } else .default, .user_agent = .{ .override = self.user_agent }, }; }
fn createRequest( self: *Client, method: std.http.Method, uri: std.Uri, options: std.http.Client.RequestOptions, resolved: ?std.http.Client.ConnectTcpOptions, ) !std.http.Client.Request { const protocol = std.http.Client.Protocol.fromUri(uri) orelse return error.UnsupportedUriScheme; const tls = !std.http.Client.disable_tls and protocol == .tls; if (tls) try self.tls_connect_lock.lock(self.io); defer if (tls) self.tls_connect_lock.unlock(self.io);
if (!std.http.Client.disable_tls and protocol == .tls) { const now = std.Io.Clock.real.now(self.io); if (self.http_client.now == null) { self.http_client.ca_bundle.rescan(self.allocator, self.io, now) catch |err| switch (err) { error.Canceled => return error.Canceled, else => return error.CertificateBundleLoadFailure, }; } @import("trust.zig").distrust(&self.http_client.ca_bundle); self.http_client.now = now; }
var request_options = options; if (resolved) |dial| { const connection = try self.http_client.connectTcpOptions(dial); errdefer self.http_client.connection_pool.release(connection, self.io); request_options.connection = connection; return self.http_client.request(method, uri, request_options); } return self.http_client.request(method, uri, request_options); }
fn sendRequest(request: *std.http.Client.Request, payload: ?[]const u8) !void { if (payload) |bytes| { request.transfer_encoding = .{ .content_length = bytes.len }; var body = try request.sendBodyUnflushed(&.{}); try body.writer.writeAll(bytes); try body.end(); try request.connection.?.flush(); } else { try request.sendBodiless(); } }
/// std reports every failed socket read as `error.ReadFailed` and keeps the /// cause on the connection. Surface the cause, so a reset on a pooled /// connection is recognised as a dead connection and replayed. fn receiveHead(request: *std.http.Client.Request) !std.http.Client.Response { return request.receiveHead(&.{}) catch |err| switch (err) { error.ReadFailed => request.connection.?.getReadError() orelse err, else => err, }; }
/// Redirects are handled here rather than by std: std's own loop keeps /// `authorization` across hosts (its `privileged_headers` are stripped on /// a domain change but never written to the wire in 0.16), has no loop /// detection, and will happily step from https down to http. /// /// The request lives in `response` so that std's internal pointers into /// it stay valid once the head is in; each hop rebuilds it in place. fn openUrl( self: *Client, response: *Response, options: FetchOptions, extra_headers: []const std.http.Header, ) !void { var hop: Hop = .fromOptions(options); var owned_url: ?[]u8 = null; defer if (owned_url) |u| self.allocator.free(u); var visited: std.ArrayList([]u8) = .empty; defer { for (visited.items) |u| self.allocator.free(u); visited.deinit(self.allocator); } var hops: u8 = 0;
while (true) { const uri = try std.Uri.parse(hop.url); var status: std.http.Status = undefined; const next = blk: { response.request = try self.createRequest(hop.method, uri, .{ .headers = self.requestHeaders(hop), .extra_headers = extra_headers, .keep_alive = self.keep_alive, .redirect_behavior = .unhandled, }, null); errdefer response.request.deinit(); try sendRequest(&response.request, hop.payload);
response.inner = try receiveHead(&response.request); if (response.inner.head.status.class() != .redirect or options.on_redirect == .surface) { try response.captureHead(options); response.final_url = owned_url; owned_url = null; return; } if (options.on_redirect == .refuse) return error.RedirectRefused; if (hops >= self.redirects.max_hops) return error.TooManyRedirects; hops += 1;
const next = try self.resolveRedirect(uri, hop.url, &response.inner); status = response.inner.head.status; var transfer_buf: [4096]u8 = undefined; const body = response.inner.reader(&transfer_buf); _ = body.discardRemaining() catch {}; break :blk next; }; response.request.deinit(); errdefer self.allocator.free(next);
if (std.ascii.eqlIgnoreCase(hop.url, next)) return error.RedirectLoop; for (visited.items) |seen| if (std.ascii.eqlIgnoreCase(seen, next)) return error.RedirectLoop; try visited.append(self.allocator, try self.allocator.dupe(u8, hop.url));
const next_uri = try std.Uri.parse(next); if (!self.redirects.allow_scheme_downgrade and std.ascii.eqlIgnoreCase(uri.scheme, "https") and std.ascii.eqlIgnoreCase(next_uri.scheme, "http")) return error.SchemeDowngrade; if (!try sameOrigin(uri, next_uri)) hop.authorization = null;
switch (status) { .see_other => { hop.method = .GET; hop.payload = null; hop.content_type = null; }, .moved_permanently, .found => if (hop.method == .POST) { hop.method = .GET; hop.payload = null; hop.content_type = null; }, .temporary_redirect, .permanent_redirect => {}, else => return error.HttpRedirectUnsupported, }
if (owned_url) |u| self.allocator.free(u); owned_url = next; hop.url = next; } }
/// Resolve the response's `Location` against the request URI into an /// owned absolute URL. Must run before the body is read: reading /// invalidates the head's strings. fn resolveRedirect(self: *Client, base: std.Uri, base_url: []const u8, response: *std.http.Client.Response) ![]u8 { const location = response.head.location orelse return error.HttpRedirectLocationMissing; // resolveInPlace wants the location at the front of a buffer with // room for a merged path after it. var aux = try self.allocator.alloc(u8, location.len + base_url.len + location.len + 2); defer self.allocator.free(aux); @memcpy(aux[0..location.len], location); var aux_slice: []u8 = aux; const resolved = base.resolveInPlace(location.len, &aux_slice) catch return error.HttpRedirectLocationInvalid; if (resolved.host == null) return error.HttpRedirectLocationInvalid; return try std.fmt.allocPrint(self.allocator, "{f}", .{resolved}); }
/// A connection the caller resolved is never redirected away from: the /// caller vetted that address, and a `Location` pointing anywhere else /// would be dialed unvetted. A 3xx is surfaced instead. fn openResolved( self: *Client, response: *Response, options: FetchOptions, resolved: ResolvedConnection, extra_headers: []const std.http.Header, ) !void { const uri = try std.Uri.parse(options.url); const protocol = std.http.Client.Protocol.fromUri(uri) orelse return error.UnsupportedUriScheme;
var uri_host_buf: [std.Io.net.HostName.max_len]u8 = undefined; const uri_host = try std.Io.net.HostName.fromUri(uri, &uri_host_buf); if (!std.ascii.eqlIgnoreCase(uri_host.bytes, resolved.logical_host)) { return error.ResolvedHostMismatch; }
const dial_host = try std.Io.net.HostName.init(resolved.dial_host); const logical_host = try std.Io.net.HostName.init(resolved.logical_host); response.request = try self.createRequest(options.method, uri, .{ .headers = self.requestHeaders(.fromOptions(options)), .extra_headers = extra_headers, .keep_alive = self.keep_alive, .redirect_behavior = .unhandled, }, .{ .host = dial_host, .port = uri.port orelse defaultPort(protocol), .protocol = protocol, .proxied_host = logical_host, }); errdefer response.request.deinit(); try sendRequest(&response.request, options.payload);
response.inner = try receiveHead(&response.request); if (response.inner.head.status.class() == .redirect and options.on_redirect == .refuse) return error.RedirectRefused; try response.captureHead(options); }
/// A request whose head has arrived. The body is still on the wire: /// `stream` it into any writer, or `deinit` to drop it. pub const Response = struct { client: *Client, request: std.http.Client.Request, inner: std.http.Client.Response, status: std.http.Status = .ok, rate_limit: RateLimitHeaders = .{}, headers: CapturedHeaders = .{}, /// Where the response actually came from when redirects were /// followed; null when it came from the requested URL. final_url: ?[]u8 = null, /// What `FetchOptions.deadline_ns` has left for the body. 0 is no bound. deadline_remaining_ns: u64 = 0, body_taken: bool = false,
fn captureHead(self: *Response, options: FetchOptions) !void { self.status = self.inner.head.status; self.rate_limit = RateLimitHeaders.fromResponseHead(self.inner.head); self.headers = try CapturedHeaders.fromResponseHead(self.client.allocator, self.inner.head, options.capture_headers); }
pub const StreamOptions = struct { /// Hard cap; `error.ResponseTooLarge` past it. A writer that /// truncates instead is the caller's choice of `sink`. max_bytes: ?usize = null, /// Overrides the client's stall guard for this body. stall: ?StallGuard = null, /// Byte and completion counters the caller may watch from /// another task, for its own reporting or watchdog. progress: ?*Progress = null, };
/// Decode and drain the body into `sink` under the stall guard and /// whatever the request deadline has left. Returns the decoded byte /// count. Call once. pub fn stream(self: *Response, sink: *std.Io.Writer, options: StreamOptions) !u64 { if (self.body_taken) return error.BodyAlreadyTaken; self.body_taken = true; const client = self.client; const dbuf = try client.allocator.alloc(u8, decompressBufferLen(self.inner.head.content_encoding)); defer client.allocator.free(dbuf); var own_progress: Progress = .{}; const progress = options.progress orelse &own_progress; try streamResponseBody( client.io, &self.inner, sink, options.stall orelse client.stall, options.max_bytes, self.deadline_remaining_ns, dbuf, progress, ); return progress.bytes.load(.acquire); }
pub fn deinit(self: *Response) void { const allocator = self.client.allocator; self.request.deinit(); self.headers.deinit(allocator); if (self.final_url) |u| allocator.free(u); allocator.destroy(self); } };
pub const FetchOptions = struct { url: []const u8, method: std.http.Method = .GET, payload: ?[]const u8 = null, authorization: ?[]const u8 = null, accept: ?[]const u8 = null, /// Replaces std's default `gzip, deflate`. `identity` asks for an /// uncompressed body; a server that compresses anyway is still /// decoded. accept_encoding: ?[]const u8 = null, content_type: ?[]const u8 = null, extra_headers: ?[]const std.http.Header = null, /// Overrides `Client.max_response_size` for this request. max_response_size: ?usize = null, /// Overrides `Client.head_deadline_ns` for this request. head_deadline_ns: ?u64 = null, /// Overrides `Client.stall` for this request. stall: ?StallGuard = null, /// Bounds the whole request, head and body, including every redirect /// hop and dead-connection replay of this attempt. Fires as /// `error.RequestDeadlineExceeded`. null is unbounded. deadline_ns: ?u64 = null, on_redirect: OnRedirect = .follow, resolved_connection: ?ResolvedConnection = null, /// Response header names (case-insensitive) to copy into /// `FetchResult.headers`. Everything else in the head is dropped once /// the body has been read. capture_headers: []const []const u8 = &.{}, };
pub const OnRedirect = enum { /// Follow under `Client.redirects`. Bodies are resent only for 307 and /// 308; 303, and 301/302 on POST, become GET without a body. follow, /// Return the 3xx response itself. surface, /// Fail with `error.RedirectRefused`. refuse, };
pub const RedirectPolicy = struct { /// Hops followed before `error.TooManyRedirects`. max_hops: u8 = 3, /// Whether an https origin may send the request on to plain http. allow_scheme_downgrade: bool = false, };
pub const StallGuard = struct { /// Abort if no body bytes arrive for this long. idle_ns: u64 = 20 * std.time.ns_per_s, /// After `grace_ns`, abort if the body rate over the last `window_ns` /// is below this. Catches a trickle, which an idle timeout alone /// never fires on. min_bytes_per_s: u64 = 4 * 1024, /// Slow starts are normal; do not judge the rate before this. grace_ns: u64 = 10 * std.time.ns_per_s, /// The rate is measured over this trailing window, so a transfer /// that ran fast and then trickles is judged on the trickle. A /// lifetime average would let the fast start mask it for hours. /// 0 measures over the whole transfer. Capped at 60s. window_ns: u64 = 10 * std.time.ns_per_s, };
pub const ResolvedConnection = struct { /// Checked address to dial. Currently IPv4 text, which std.Io.net.HostName accepts. dial_host: []const u8, /// Original URL host, used by std.http for TLS/SNI and connection identity. logical_host: []const u8, };
pub const FetchResult = struct { status: std.http.Status, body: []u8, rate_limit: RateLimitHeaders = .{}, headers: CapturedHeaders = .{}, /// Where the response actually came from when redirects were /// followed; null when it came from the requested URL. final_url: ?[]u8 = null,
pub fn deinit(self: *FetchResult, allocator: std.mem.Allocator) void { allocator.free(self.body); self.headers.deinit(allocator); if (self.final_url) |u| allocator.free(u); } };
/// The subset of response headers a caller asked for, owned copies. When /// a header repeats, the first occurrence wins. pub const CapturedHeaders = struct { entries: []std.http.Header = &.{},
pub fn fromResponseHead( allocator: std.mem.Allocator, head: std.http.Client.Response.Head, names: []const []const u8, ) !CapturedHeaders { if (names.len == 0) return .{}; const entries = try allocator.alloc(std.http.Header, names.len); var count: usize = 0; errdefer { for (entries[0..count]) |e| allocator.free(e.value); allocator.free(entries); } var it = head.iterateHeaders(); while (it.next()) |header| { for (names) |name| { if (!std.ascii.eqlIgnoreCase(header.name, name)) continue; if (hasName(entries[0..count], name)) break; entries[count] = .{ .name = name, .value = try allocator.dupe(u8, header.value) }; count += 1; break; } } if (count == 0) { allocator.free(entries); return .{}; } return .{ .entries = try shrink(allocator, entries, count) }; }
fn shrink(allocator: std.mem.Allocator, entries: []std.http.Header, count: usize) ![]std.http.Header { if (allocator.resize(entries, count)) return entries[0..count]; const exact = try allocator.dupe(std.http.Header, entries[0..count]); allocator.free(entries); return exact; }
fn hasName(entries: []const std.http.Header, name: []const u8) bool { for (entries) |e| if (std.ascii.eqlIgnoreCase(e.name, name)) return true; return false; }
pub fn get(self: CapturedHeaders, name: []const u8) ?[]const u8 { for (self.entries) |e| if (std.ascii.eqlIgnoreCase(e.name, name)) return e.value; return null; }
pub fn deinit(self: *CapturedHeaders, allocator: std.mem.Allocator) void { for (self.entries) |e| allocator.free(e.value); if (self.entries.len > 0) allocator.free(self.entries); self.* = .{}; } };
pub const RateLimitHeaders = struct { limit: ?u64 = null, remaining: ?u64 = null, /// `RateLimit-Reset` as sent: delta seconds per the IETF draft, but /// some servers send an absolute Unix time. `delaySeconds` tells them /// apart. Signed values preserve pre-epoch timestamps for protocol /// policies (such as ATProto) that give Reset precedence over Retry-After. reset: ?i64 = null, /// `Retry-After` in its delta-seconds form. retry_after: ?u64 = null, /// `Retry-After` in its HTTP-date form, as Unix seconds. retry_after_at: ?i64 = null,
pub fn fromResponseHead(head: std.http.Client.Response.Head) RateLimitHeaders { var result: RateLimitHeaders = .{}; var it = head.iterateHeaders(); while (it.next()) |header| { if (std.ascii.eqlIgnoreCase(header.name, "ratelimit-limit")) { result.limit = parseHeaderInt(header.value); } else if (std.ascii.eqlIgnoreCase(header.name, "ratelimit-remaining")) { result.remaining = parseHeaderInt(header.value); } else if (std.ascii.eqlIgnoreCase(header.name, "ratelimit-reset")) { result.reset = std.fmt.parseInt(i64, std.mem.trim(u8, header.value, " \t"), 10) catch null; } else if (std.ascii.eqlIgnoreCase(header.name, "retry-after")) { result.retry_after = parseHeaderInt(header.value); if (result.retry_after == null) result.retry_after_at = parseHttpDate(header.value); } } return result; }
pub fn isEmpty(self: RateLimitHeaders) bool { return self.limit == null and self.remaining == null and self.reset == null and self.retry_after == null and self.retry_after_at == null; }
/// Seconds the server asked us to wait, given the current Unix time. /// `Retry-After` wins over `RateLimit-Reset`; a reset at or past /// `epoch_threshold` is read as an absolute time. Null when no /// header said anything usable; a hint already in the past is 0. pub fn delaySeconds(self: RateLimitHeaders, now_s: i64) ?u64 { if (self.retry_after) |secs| return secs; if (self.retry_after_at) |at| return secondsUntil(at, now_s); if (self.reset) |reset| { if (reset <= 0) return 0; if (reset >= epoch_threshold) return secondsUntil(@intCast(reset), now_s); return @intCast(reset); } return null; }
/// A `RateLimit-Reset` this large is a Unix time (2001-09-09), not /// a wait: no server asks for a 31-year pause. pub const epoch_threshold: u64 = 1_000_000_000;
fn secondsUntil(at: i64, now_s: i64) u64 { return if (at > now_s) @intCast(at - now_s) else 0; } };};
/// HTTP-date: IMF-fixdate, obsolete RFC 850, or ANSI C asctime./// RFC 850 two-digit years use Go time.Parse’s 69/68 pivot. Returns Unix seconds.fn parseHttpDate(value: []const u8) ?i64 { const input = std.mem.trim(u8, value, " \t"); var normalized: [29]u8 = undefined; const v = normalizeHttpDate(input, &normalized) orelse return null; // Sun, 06 Nov 1994 08:49:37 GMT if (v.len != 29) return null; if (v[3] != ',' or v[4] != ' ' or v[7] != ' ' or v[11] != ' ' or v[16] != ' ' or v[19] != ':' or v[22] != ':' or v[25] != ' ') return null; if (!std.mem.eql(u8, v[26..29], "GMT")) return null; const day = std.fmt.parseInt(u32, v[5..7], 10) catch return null; const month = monthNumber(v[8..11]) orelse return null; const year = std.fmt.parseInt(i64, v[12..16], 10) catch return null; const hour = std.fmt.parseInt(u32, v[17..19], 10) catch return null; const minute = std.fmt.parseInt(u32, v[20..22], 10) catch return null; const second = std.fmt.parseInt(u32, v[23..25], 10) catch return null; const weekdays = [_][]const u8{ "Mon", "Tue", "Wed", "Thu", "Fri", "Sat", "Sun" }; var valid_weekday = false; for (weekdays) |name| valid_weekday = valid_weekday or std.mem.eql(u8, v[0..3], name); if (!valid_weekday) return null; const leap = @mod(year, 4) == 0 and (@mod(year, 100) != 0 or @mod(year, 400) == 0); const days = [_]u8{ 31, if (leap) 29 else 28, 31, 30, 31, 30, 31, 31, 30, 31, 30, 31 }; if (day == 0 or day > days[month - 1] or hour > 23 or minute > 59 or second > 59) return null; return daysFromCivil(year, month, day) * 86_400 + @as(i64, hour) * 3600 + @as(i64, minute) * 60 + second;}
fn normalizeHttpDate(input: []const u8, buf: *[29]u8) ?[]const u8 { // Canonical dates require no copy. if (input.len == 29 and input[3] == ',') return input; if (std.mem.indexOf(u8, input, ", ")) |comma| { const names = [_][]const u8{ "Monday", "Tuesday", "Wednesday", "Thursday", "Friday", "Saturday", "Sunday" }; var valid = false; for (names) |name| valid = valid or std.mem.eql(u8, input[0..comma], name); if (!valid) return null; const tail = input[comma + 2 ..]; if (tail.len != 22 or tail[2] != '-' or tail[6] != '-' or tail[9] != ' ') return null; const short_year = std.fmt.parseInt(u8, tail[7..9], 10) catch return null; const year: u16 = @as(u16, short_year) + @as(u16, if (short_year >= 69) 1900 else 2000); return std.fmt.bufPrint(buf, "{s}, {s} {s} {d} {s}", .{ input[0..3], tail[0..2], tail[3..6], year, tail[10..] }) catch null; } if (input.len != 24 or input[3] != ' ' or input[7] != ' ' or input[10] != ' ' or input[19] != ' ') return null; const day = std.fmt.parseInt(u8, std.mem.trimStart(u8, input[8..10], " "), 10) catch return null; return std.fmt.bufPrint(buf, "{s}, {d:0>2} {s} {s} {s} GMT", .{ input[0..3], day, input[4..7], input[20..24], input[11..19] }) catch null;}
fn monthNumber(name: []const u8) ?u32 { const names = [_][]const u8{ "Jan", "Feb", "Mar", "Apr", "May", "Jun", "Jul", "Aug", "Sep", "Oct", "Nov", "Dec" }; for (names, 1..) |n, i| if (std.ascii.eqlIgnoreCase(n, name)) return @intCast(i); return null;}
/// Days since 1970-01-01 for a proleptic Gregorian date (Howard Hinnant's/// days_from_civil).fn daysFromCivil(year: i64, month: u32, day: u32) i64 { const y: i64 = if (month <= 2) year - 1 else year; const era: i64 = @divFloor(y, 400); const yoe: i64 = y - era * 400; const m: i64 = month; const mp: i64 = @mod(m + 9, 12); const doy: i64 = @divFloor(153 * mp + 2, 5) + @as(i64, day) - 1; const doe: i64 = yoe * 365 + @divFloor(yoe, 4) - @divFloor(yoe, 100) + doy; return era * 146_097 + doe - 719_468;}
/// std asserts on CR/LF in headers, which is a crash in safe builds and/// injection in fast ones. Refuse up front instead.fn validateHeaderValue(value: []const u8) error{InvalidHeader}!void { for (value) |c| if (c == '\r' or c == '\n') return error.InvalidHeader;}
fn parseHeaderInt(value: []const u8) ?u64 { const trimmed = std.mem.trim(u8, value, " \t"); return std.fmt.parseInt(u64, trimmed, 10) catch null;}
var warned_unguarded: std.atomic.Value(bool) = .init(false);
/// A watcher only bounds work that runs beside it. An Io with no concurrency/// to give leaves the request to run on the caller's own stack, where nothing/// can interrupt it.fn warnUnguarded() void { if (warned_unguarded.swap(true, .monotonic)) return; log.warn("the Io cannot run tasks concurrently; requests are running without deadline or stall guards", .{});}
/// Body transfer counters, readable from another task while `stream` runs.pub const Progress = struct { bytes: std.atomic.Value(usize) = .init(0), done: std.atomic.Value(bool) = .init(false),};
fn streamBody( response: *std.http.Client.Response, writer: *std.Io.Writer, limit: ?usize, decompress_buffer: []u8, progress: *Progress,) !void { defer progress.done.store(true, .release);
var transfer_buffer: [8 * 1024]u8 = undefined; var decompress: std.http.Decompress = undefined; const reader = response.readerDecompressing(&transfer_buffer, &decompress, decompress_buffer);
var chunk: [64 * 1024]u8 = undefined; while (true) { var sink = std.Io.Writer.fixed(&chunk); const n = reader.stream(&sink, .limited(chunk.len)) catch |err| switch (err) { error.EndOfStream => return, // Null when the read was canceled rather than failing on its own. error.ReadFailed => return response.bodyErr() orelse error.ReadFailed, else => |e| return e, }; if (n == 0) continue; if (limit) |max| if (progress.bytes.load(.monotonic) + n > max) return error.ResponseTooLarge; try writer.writeAll(chunk[0..n]); _ = progress.bytes.fetchAdd(n, .release); }}
/// A wall-clock check placed *after* a read cannot fire, because a peer that/// simply stops sending leaves that read blocked in the kernel forever. So the/// read runs as a task and the caller watches its byte counter, canceling when/// progress dies.fn streamResponseBody( io: std.Io, response: *std.http.Client.Response, writer: *std.Io.Writer, guard: Client.StallGuard, limit: ?usize, deadline_ns: u64, decompress_buffer: []u8, progress: *Progress,) !void { if (guard.idle_ns == 0 and guard.min_bytes_per_s == 0 and deadline_ns == 0) return streamBody(response, writer, limit, decompress_buffer, progress);
var future = io.concurrent(streamBody, .{ response, writer, limit, decompress_buffer, progress }) catch |err| switch (err) { error.ConcurrencyUnavailable => { warnUnguarded(); return streamBody(response, writer, limit, decompress_buffer, progress); }, };
// Poll often so a body that lands at once is not charged a whole sample // period; sample the rate on a coarser grid so the window stays cheap. const tick_ns: u64 = 10 * std.time.ns_per_ms; const sample_ns: u64 = 100 * std.time.ns_per_ms; var window: RateWindow = .init(guard.window_ns, sample_ns); var idle_ns: u64 = 0; var elapsed_ns: u64 = 0; var since_sample_ns: u64 = 0; var last_bytes: usize = 0;
while (!progress.done.load(.acquire)) { std.Io.Clock.Duration.sleep( .{ .raw = .fromNanoseconds(tick_ns), .clock = .awake }, io, ) catch { // Cancelled from above: take the reader down with us rather // than awaiting a read that may never return. future.cancel(io) catch {}; return error.Canceled; }; elapsed_ns += tick_ns; since_sample_ns += tick_ns;
if (deadline_ns != 0 and elapsed_ns >= deadline_ns) { future.cancel(io) catch {}; return error.RequestDeadlineExceeded; }
const seen = progress.bytes.load(.acquire); if (seen != last_bytes) { idle_ns = 0; } else { idle_ns += tick_ns; } if (guard.idle_ns != 0 and idle_ns >= guard.idle_ns) { future.cancel(io) catch {}; return error.TransferStalled; }
if (since_sample_ns >= sample_ns) { since_sample_ns = 0; window.push(seen - last_bytes); if (guard.min_bytes_per_s != 0 and elapsed_ns >= guard.grace_ns) { if (window.bytesPerSecond(seen, elapsed_ns)) |rate| { if (rate < guard.min_bytes_per_s) { future.cancel(io) catch {}; return error.TransferTooSlow; } } } } last_bytes = seen; }
return future.await(io);}
/// Bytes moved per sample over a trailing window. With a zero window the/// rate is the lifetime average.const RateWindow = struct { samples: [max_samples]usize = @splat(0), len: usize, next: usize = 0, filled: usize = 0, sample_ns: u64,
const max_samples = 600;
fn init(window_ns: u64, sample_ns: u64) RateWindow { const wanted = window_ns / sample_ns; return .{ .len = @min(wanted, max_samples), .sample_ns = sample_ns }; }
fn push(self: *RateWindow, delta: usize) void { if (self.len == 0) return; self.samples[self.next] = delta; self.next = (self.next + 1) % self.len; if (self.filled < self.len) self.filled += 1; }
/// Null until the window has enough samples to judge. fn bytesPerSecond(self: *const RateWindow, total: usize, elapsed_ns: u64) ?u64 { if (self.len == 0) { const secs = elapsed_ns / std.time.ns_per_s; if (secs == 0) return null; return total / secs; } if (self.filled < self.len) return null; var sum: usize = 0; for (self.samples[0..self.len]) |n| sum += n; const window_ns = self.sample_ns * self.len; return @intCast((@as(u128, sum) * std.time.ns_per_s) / window_ns); }};
/// The peer hung up without answering. Distinct from a mid-body failure: no/// part of a response was seen, so the request provably did not take effect.fn isDeadConnection(err: anyerror) bool { return switch (err) { error.ConnectionResetByPeer, error.BrokenPipe, error.EndOfStream, error.UnexpectedWriteFailure, error.HttpConnectionClosing, => true, else => false, };}
fn isReplayable(options: Client.FetchOptions) bool { if (options.payload != null) return false; return switch (options.method) { .GET, .HEAD, .OPTIONS => true, else => false, };}
/// A zero-length buffer works only while every response is `identity`; gzip/// needs a real window. Sizing from the negotiated encoding keeps the identity/// path allocation-free in practice (0 bytes) without capping compressed ones.fn decompressBufferLen(encoding: std.http.ContentEncoding) usize { return encoding.minBufferCapacity();}
/// Scheme, host, and effective port all equal. A redirect that changes any/// of them is a different party and must not inherit credentials.fn sameOrigin(a: std.Uri, b: std.Uri) !bool { if (!std.ascii.eqlIgnoreCase(a.scheme, b.scheme)) return false; var a_buf: [std.Io.net.HostName.max_len]u8 = undefined; var b_buf: [std.Io.net.HostName.max_len]u8 = undefined; const a_host = try std.Io.net.HostName.fromUri(a, &a_buf); const b_host = try std.Io.net.HostName.fromUri(b, &b_buf); if (!std.ascii.eqlIgnoreCase(a_host.bytes, b_host.bytes)) return false; const a_proto = std.http.Client.Protocol.fromUri(a) orelse return error.UnsupportedUriScheme; const b_proto = std.http.Client.Protocol.fromUri(b) orelse return error.UnsupportedUriScheme; return (a.port orelse defaultPort(a_proto)) == (b.port orelse defaultPort(b_proto));}
fn defaultPort(protocol: std.http.Client.Protocol) u16 { return switch (protocol) { .plain => 80, .tls => 443, };}
// === tests ===
test "client init/deinit" { const io = std.Options.debug_io; var transport = Client.init(io, std.testing.allocator); defer transport.deinit();}
test "resolved transport rejects mismatched logical host" { const io = std.Options.debug_io; var transport = Client.init(io, std.testing.allocator); defer transport.deinit();
// debug_io has no concurrency to give, so this request runs unguarded. const log_level = std.testing.log_level; std.testing.log_level = .err; defer std.testing.log_level = log_level;
try std.testing.expectError(error.ResolvedHostMismatch, transport.fetch(.{ .url = "http://example.com/", .resolved_connection = .{ .dial_host = "127.0.0.1", .logical_host = "other.example", }, }));}
test "client parses rate limit headers" { const response_bytes = "HTTP/1.1 429 Too Many Requests\r\n" ++ "RateLimit-Limit: 3000\r\n" ++ "ratelimit-remaining: 0\r\n" ++ "RateLimit-Reset: 1710000000\r\n" ++ "Retry-After: 2\r\n\r\n";
const head = try std.http.Client.Response.Head.parse(response_bytes); const headers = Client.RateLimitHeaders.fromResponseHead(head);
try std.testing.expectEqual(@as(?u64, 3000), headers.limit); try std.testing.expectEqual(@as(?u64, 0), headers.remaining); try std.testing.expectEqual(@as(?i64, 1710000000), headers.reset); try std.testing.expectEqual(@as(?u64, 2), headers.retry_after); try std.testing.expect(!headers.isEmpty());}
test "captured headers are the ones asked for, case-insensitively, first occurrence wins" { const response_bytes = "HTTP/1.1 401 Unauthorized\r\n" ++ "Content-Type: application/json; charset=utf-8\r\n" ++ "X-Nonce: first\r\n" ++ "x-nonce: second\r\n" ++ "WWW-Authenticate: Bearer realm=\"x\"\r\n\r\n";
const head = try std.http.Client.Response.Head.parse(response_bytes); var headers = try Client.CapturedHeaders.fromResponseHead( std.testing.allocator, head, &.{ "content-type", "X-NONCE", "x-missing" }, ); defer headers.deinit(std.testing.allocator);
try std.testing.expectEqualStrings("application/json; charset=utf-8", headers.get("Content-Type").?); try std.testing.expectEqualStrings("first", headers.get("x-nonce").?); try std.testing.expectEqual(@as(?[]const u8, null), headers.get("www-authenticate")); try std.testing.expectEqual(@as(?[]const u8, null), headers.get("x-missing")); try std.testing.expectEqual(@as(usize, 2), headers.entries.len);}
test "asking for no headers captures nothing and owns nothing" { const head = try std.http.Client.Response.Head.parse("HTTP/1.1 200 OK\r\nX-A: 1\r\n\r\n"); var headers = try Client.CapturedHeaders.fromResponseHead(std.testing.allocator, head, &.{}); defer headers.deinit(std.testing.allocator); try std.testing.expectEqual(@as(usize, 0), headers.entries.len);}
test "requests identify the client, and applications can name themselves" { // A server operator seeing load should be able to tell what is calling // and where to ask about it. std.http's default says neither. const testing = std.testing; var threaded: std.Io.Threaded = .init(testing.allocator, .{}); defer threaded.deinit(); const io = threaded.io();
var default_transport = Client.init(io, testing.allocator); defer default_transport.deinit(); try testing.expectEqualStrings(default_user_agent, default_transport.user_agent); try testing.expect(std.mem.indexOf(u8, default_transport.user_agent, "http") != null); try testing.expect(std.mem.startsWith(u8, default_transport.user_agent, "zttp/" ++ build_options.version ++ " "));
var named = Client.initWithUserAgent(io, testing.allocator, "stream/1.2.3 (+https://example.invalid)"); defer named.deinit(); try testing.expectEqualStrings("stream/1.2.3 (+https://example.invalid)", named.user_agent);
// An empty override must not silently erase the identity. var empty = Client.initWithUserAgent(io, testing.allocator, ""); defer empty.deinit(); try testing.expectEqualStrings(default_user_agent, empty.user_agent);}
test "extra headers past the buffer are an error, not silently dropped" { const testing = std.testing; var threaded: std.Io.Threaded = .init(testing.allocator, .{}); defer threaded.deinit();
var transport = Client.init(threaded.io(), testing.allocator); defer transport.deinit();
const header: std.http.Header = .{ .name = "x-zttp-test", .value = "1" }; const too_many: [8]std.http.Header = @splat(header); try testing.expectError(error.TooManyHeaders, transport.fetch(.{ .url = "http://127.0.0.1:1/never-reached", .accept = "application/json", .extra_headers = &too_many, }));}
// A loopback origin for transport tests. Each canned response is served on its// own accepted connection; `serve` runs on a thread so the client can drive the// exchange from the test body.const TestOrigin = struct { listener: std.Io.net.Server, io: std.Io,
fn init(io: std.Io) !TestOrigin { return .{ .listener = try (std.Io.net.IpAddress{ .ip4 = .loopback(0) }).listen(io, .{ .reuse_address = true }), .io = io, }; } fn deinit(self: *TestOrigin) void { self.listener.deinit(self.io); } fn port(self: *const TestOrigin) u16 { return self.listener.socket.address.getPort(); }
/// Accept one connection, read the request head, write `response` verbatim. /// Answers with headers promising a long body, dribbles a few bytes, then /// goes quiet — the shape of a slow origin that pins a worker. fn serveStalling(self: *TestOrigin) !void { var conn = try self.listener.accept(self.io); defer conn.close(self.io); var rbuf: [4096]u8 = undefined; var reader = conn.reader(self.io, &rbuf); var scratch: [2048]u8 = undefined; var slices = [_][]u8{&scratch}; _ = reader.interface.readVec(&slices) catch {}; var wbuf: [4096]u8 = undefined; var writer = conn.writer(self.io, &wbuf); try writer.interface.writeAll( "HTTP/1.1 200 OK\r\nContent-Length: 1048576\r\n\r\nhello", ); try writer.interface.flush(); // Send nothing further and block until the client hangs up. A guarded // client returns quickly; an unguarded one waits here forever. var sink: [1]u8 = undefined; var tail = [_][]u8{&sink}; _ = reader.interface.readVec(&tail) catch {}; }
/// Accepts, reads the request, then never answers: connection /// established, response head never arrives, nothing for StallGuard to /// measure. fn serveSilent(self: *TestOrigin) !void { var conn = try self.listener.accept(self.io); defer conn.close(self.io); var rbuf: [4096]u8 = undefined; var reader = conn.reader(self.io, &rbuf); var scratch: [2048]u8 = undefined; var slices = [_][]u8{&scratch}; _ = reader.interface.readVec(&slices) catch {}; // Say nothing and block until the client hangs up. var sink: [1]u8 = undefined; var tail = [_][]u8{&sink}; _ = reader.interface.readVec(&tail) catch {}; }
/// Accepts, reads the request, optionally writes `preamble`, then holds /// the socket open and silent until `release` is set. Gives up on its own /// after a few seconds so a client that never lets go fails its test /// instead of hanging the suite. fn serveHeldOpen( self: *TestOrigin, preamble: []const u8, got_request: *std.atomic.Value(bool), release: *std.atomic.Value(bool), ) !void { var conn = try self.listener.accept(self.io); defer conn.close(self.io); var rbuf: [4096]u8 = undefined; var reader = conn.reader(self.io, &rbuf); var scratch: [2048]u8 = undefined; var slices = [_][]u8{&scratch}; _ = reader.interface.readVec(&slices) catch {}; if (preamble.len > 0) { var wbuf: [4096]u8 = undefined; var writer = conn.writer(self.io, &wbuf); try writer.interface.writeAll(preamble); try writer.interface.flush(); } got_request.store(true, .release); var waited_ms: u64 = 0; while (!release.load(.acquire) and waited_ms < 3000) : (waited_ms += 10) try self.io.sleep(.fromMilliseconds(10), .awake); }
/// Accepts, reads the request, then hangs up without answering — what a /// pooled connection reaped by the origin looks like on next reuse. fn serveDeadThenOk(self: *TestOrigin, response: []const u8) !void { { var dead = try self.listener.accept(self.io); var rbuf: [4096]u8 = undefined; var reader = dead.reader(self.io, &rbuf); var scratch: [2048]u8 = undefined; var slices = [_][]u8{&scratch}; _ = reader.interface.readVec(&slices) catch {}; dead.close(self.io); } try self.serveOnce(response); }
/// One keep-alive connection: answers the first request, reads the second /// and resets the connection instead of answering. The client's read then /// fails with a reset rather than a clean end of stream. A fresh connection /// is served normally. fn serveOkThenReset(self: *TestOrigin, response: []const u8) !void { { var conn = try self.listener.accept(self.io); defer conn.close(self.io); var rbuf: [4096]u8 = undefined; var reader = conn.reader(self.io, &rbuf); var scratch: [2048]u8 = undefined; var slices = [_][]u8{&scratch}; _ = reader.interface.readVec(&slices) catch {}; var wbuf: [4096]u8 = undefined; var writer = conn.writer(self.io, &wbuf); try writer.interface.writeAll(response); try writer.interface.flush(); var again = [_][]u8{&scratch}; _ = reader.interface.readVec(&again) catch {}; const reset: std.posix.linger = .{ .onoff = 1, .linger = 0 }; try std.posix.setsockopt(conn.socket.handle, std.posix.SOL.SOCKET, std.posix.SO.LINGER, std.mem.asBytes(&reset)); } try self.serveOnce(response); }
/// One keep-alive connection: answers the first request, then reads the /// second and goes silent with the socket held open — the shape of a /// pooled connection a middlebox has blackholed while it sat idle (no FIN /// or RST ever reaches the client). After the client abandons that socket, /// accepts a fresh connection and serves normally. fn serveOkThenBlackhole( self: *TestOrigin, response: []const u8, reused: *std.atomic.Value(bool), ) !void { { var conn = try self.listener.accept(self.io); defer conn.close(self.io); var rbuf: [4096]u8 = undefined; var reader = conn.reader(self.io, &rbuf); var scratch: [2048]u8 = undefined; var slices = [_][]u8{&scratch}; _ = reader.interface.readVec(&slices) catch {}; var wbuf: [4096]u8 = undefined; var writer = conn.writer(self.io, &wbuf); try writer.interface.writeAll(response); try writer.interface.flush(); // Second request arrives on the same pooled connection; read it // and say nothing, holding the socket open until the client // gives up and hangs up. var again = [_][]u8{&scratch}; const n = reader.interface.readVec(&again) catch 0; if (n > 0) reused.store(true, .release); var sink: [1]u8 = undefined; var tail = [_][]u8{&sink}; _ = reader.interface.readVec(&tail) catch {}; } try self.serveOnce(response); }
/// Accept one connection, remember whether the request carried an /// `authorization` header, and answer `response`. fn serveRecordingAuth(self: *TestOrigin, response: []const u8, saw_auth: *std.atomic.Value(bool)) !void { var conn = try self.listener.accept(self.io); defer conn.close(self.io); var rbuf: [4096]u8 = undefined; var reader = conn.reader(self.io, &rbuf); var scratch: [2048]u8 = undefined; var slices = [_][]u8{&scratch}; const n = reader.interface.readVec(&slices) catch 0; if (std.ascii.findIgnoreCase(scratch[0..n], "\r\nauthorization:") != null) saw_auth.store(true, .release); var wbuf: [4096]u8 = undefined; var writer = conn.writer(self.io, &wbuf); try writer.interface.writeAll(response); try writer.interface.flush(); }
/// Accept one connection, record the request line and whether a body /// followed the head, and answer `response`. fn serveRecordingRequest(self: *TestOrigin, response: []const u8, request_line: *[128]u8, request_line_len: *usize, saw_body: *std.atomic.Value(bool)) !void { var conn = try self.listener.accept(self.io); defer conn.close(self.io); var rbuf: [4096]u8 = undefined; var reader = conn.reader(self.io, &rbuf); var scratch: [4096]u8 = undefined; var slices = [_][]u8{&scratch}; const n = reader.interface.readVec(&slices) catch 0; const seen = scratch[0..n]; const eol = std.mem.indexOf(u8, seen, "\r\n") orelse seen.len; const line_len = @min(eol, request_line.len); @memcpy(request_line[0..line_len], seen[0..line_len]); request_line_len.* = line_len; if (std.mem.indexOf(u8, seen, "\r\n\r\n")) |head_end| { if (head_end + 4 < seen.len) saw_body.store(true, .release); } var wbuf: [4096]u8 = undefined; var writer = conn.writer(self.io, &wbuf); try writer.interface.writeAll(response); try writer.interface.flush(); }
/// Answers with a long Content-Length, sends `burst` bytes at once, then /// goes quiet: a transfer that ran fast and is now trickling nothing. fn serveBurstThenStall(self: *TestOrigin, burst: usize) !void { var conn = try self.listener.accept(self.io); defer conn.close(self.io); var rbuf: [4096]u8 = undefined; var reader = conn.reader(self.io, &rbuf); var scratch: [2048]u8 = undefined; var slices = [_][]u8{&scratch}; _ = reader.interface.readVec(&slices) catch {}; var wbuf: [64 * 1024]u8 = undefined; var writer = conn.writer(self.io, &wbuf); try writer.interface.print("HTTP/1.1 200 OK\r\nContent-Length: {d}\r\n\r\n", .{burst * 4}); const chunk: [4096]u8 = @splat('x'); var sent: usize = 0; while (sent < burst) : (sent += chunk.len) try writer.interface.writeAll(&chunk); try writer.interface.flush(); var sink: [1]u8 = undefined; var tail = [_][]u8{&sink}; _ = reader.interface.readVec(&tail) catch {}; }
/// Accept one connection, copy the request head into `head`, answer. fn serveRecordingHead(self: *TestOrigin, response: []const u8, head: *[4096]u8, head_len: *usize) !void { var conn = try self.listener.accept(self.io); defer conn.close(self.io); var rbuf: [4096]u8 = undefined; var reader = conn.reader(self.io, &rbuf); var slices = [_][]u8{head}; head_len.* = reader.interface.readVec(&slices) catch 0; var wbuf: [4096]u8 = undefined; var writer = conn.writer(self.io, &wbuf); try writer.interface.writeAll(response); try writer.interface.flush(); }
fn serveOnce(self: *TestOrigin, response: []const u8) !void { var conn = try self.listener.accept(self.io); defer conn.close(self.io); var rbuf: [4096]u8 = undefined; var reader = conn.reader(self.io, &rbuf); // Read whatever the client sent; the request content is irrelevant here. var scratch: [2048]u8 = undefined; var slices = [_][]u8{&scratch}; _ = reader.interface.readVec(&slices) catch {}; var wbuf: [4096]u8 = undefined; var writer = conn.writer(self.io, &wbuf); try writer.interface.writeAll(response); try writer.interface.flush(); }};
test "gzip responses decode: the identity override was covering an empty decompress buffer" { const testing = std.testing; var threaded: std.Io.Threaded = .init(testing.allocator, .{}); defer threaded.deinit(); const io = threaded.io();
var origin = try TestOrigin.init(io); defer origin.deinit();
// gzip of "hello zttp" produced at comptime would need a compressor; instead // assert the *identity* path still works through readerDecompressing with a // real buffer, which is the change under test. A gzip body is exercised by // the live network suite. const body = "hello zttp"; var resp_buf: [256]u8 = undefined; const response = try std.fmt.bufPrint(&resp_buf, "HTTP/1.1 200 OK\r\nContent-Length: {d}\r\nConnection: close\r\n\r\n{s}", .{ body.len, body });
const T = struct { fn run(o: *TestOrigin, r: []const u8) void { o.serveOnce(r) catch {}; } }; var th = try std.Thread.spawn(.{}, T.run, .{ &origin, response }); defer th.join();
var transport = Client.init(io, testing.allocator); defer transport.deinit(); transport.keep_alive = false;
var url_buf: [64]u8 = undefined; const url = try std.fmt.bufPrint(&url_buf, "http://127.0.0.1:{d}/", .{origin.port()});
var res = try transport.fetch(.{ .url = url }); defer res.deinit(testing.allocator); try testing.expectEqual(std.http.Status.ok, res.status); try testing.expectEqualStrings(body, res.body);}
test "a stalled transfer is abandoned instead of pinning the caller" { const allocator = std.testing.allocator; var threaded: std.Io.Threaded = .init(allocator, .{}); defer threaded.deinit(); const io = threaded.io();
var origin = try TestOrigin.init(io); defer origin.deinit(); const url = try std.fmt.allocPrint(allocator, "http://127.0.0.1:{d}/stall", .{origin.port()}); defer allocator.free(url);
var server = try std.Thread.spawn(.{}, struct { fn run(o: *TestOrigin) void { o.serveStalling() catch {}; } }.run, .{&origin}); defer server.join();
var transport = Client.init(io, allocator); defer transport.deinit(); transport.stall = .{ .idle_ns = 400 * std.time.ns_per_ms, .min_bytes_per_s = 0, .grace_ns = 0, };
// Unguarded, this call never returns: the origin holds the connection open // and simply stops sending. try std.testing.expectError(error.TransferStalled, transport.fetch(.{ .url = url }));}
test "a gzip response decodes: the identity override was covering an empty decompress buffer" { const allocator = std.testing.allocator; var threaded: std.Io.Threaded = .init(allocator, .{}); defer threaded.deinit(); const io = threaded.io();
const repeated: [512][4]u8 = @splat("zttp".*); const payload = std.mem.asBytes(&repeated); var gz_buf: [64 * 1024]u8 = undefined; var gz = std.Io.Writer.fixed(&gz_buf); { var flate_buf: [std.compress.flate.max_window_len]u8 = undefined; var comp: std.compress.flate.Compress = try .init(&gz, &flate_buf, .gzip, .default); try comp.writer.writeAll(payload); try comp.finish(); }
var origin = try TestOrigin.init(io); defer origin.deinit(); const url = try std.fmt.allocPrint(allocator, "http://127.0.0.1:{d}/gz", .{origin.port()}); defer allocator.free(url);
const response = try std.fmt.allocPrint( allocator, "HTTP/1.1 200 OK\r\nContent-Encoding: gzip\r\nContent-Length: {d}\r\n\r\n{s}", .{ gz.buffered().len, gz.buffered() }, ); defer allocator.free(response);
var server = try std.Thread.spawn(.{}, struct { fn run(o: *TestOrigin, r: []const u8) void { o.serveOnce(r) catch {}; } }.run, .{ &origin, response }); defer server.join();
var transport = Client.init(io, allocator); defer transport.deinit(); var res = transport.fetch(.{ .url = url }) catch |err| { std.debug.print("fetch failed: {t}\n", .{err}); return err; }; defer res.deinit(allocator);
try std.testing.expectEqualStrings(payload, res.body);}
test "a peer that accepts and never answers is abandoned at the head deadline" { const allocator = std.testing.allocator; var threaded: std.Io.Threaded = .init(allocator, .{}); defer threaded.deinit(); const io = threaded.io();
var origin = try TestOrigin.init(io); defer origin.deinit(); const url = try std.fmt.allocPrint(allocator, "http://127.0.0.1:{d}/silent", .{origin.port()}); defer allocator.free(url);
var server = try std.Thread.spawn(.{}, struct { fn run(o: *TestOrigin) void { o.serveSilent() catch {}; } }.run, .{&origin}); defer server.join();
var transport = Client.init(io, allocator); defer transport.deinit(); transport.head_deadline_ns = 300 * std.time.ns_per_ms;
// Unguarded, this call sits in receiveHead until the peer or kernel gives // up — for a wedged keep-alive socket, never. try std.testing.expectError(error.ResponseHeadTimeout, transport.fetch(.{ .url = url }));}
test "canceling a fetch parked on a silent peer returns error.Canceled promptly" { const allocator = std.testing.allocator; var threaded: std.Io.Threaded = .init(allocator, .{}); defer threaded.deinit(); const io = threaded.io();
var origin = try TestOrigin.init(io); defer origin.deinit(); const url = try std.fmt.allocPrint(allocator, "http://127.0.0.1:{d}/silent", .{origin.port()}); defer allocator.free(url);
var got_request: std.atomic.Value(bool) = .init(false); var release: std.atomic.Value(bool) = .init(false); var server = try std.Thread.spawn(.{}, struct { fn run(o: *TestOrigin, got: *std.atomic.Value(bool), rel: *std.atomic.Value(bool)) void { o.serveHeldOpen("", got, rel) catch {}; } }.run, .{ &origin, &got_request, &release }); defer server.join(); defer release.store(true, .release);
var transport = Client.init(io, allocator); defer transport.deinit(); transport.head_deadline_ns = 30 * std.time.ns_per_s; transport.dead_connection_attempts = 1;
var fetching = try io.concurrent(struct { fn run(c: *Client, u: []const u8) !void { var res = try c.fetch(.{ .url = u }); res.deinit(c.allocator); } }.run, .{ &transport, url }); var waited_ms: u64 = 0; while (!got_request.load(.acquire) and waited_ms < 3000) : (waited_ms += 5) io.sleep(.fromMilliseconds(5), .awake) catch {}; io.sleep(.fromMilliseconds(50), .awake) catch {};
const started = std.Io.Timestamp.now(io, .awake).nanoseconds; try std.testing.expectError(error.Canceled, fetching.cancel(io)); const elapsed_ns = std.Io.Timestamp.now(io, .awake).nanoseconds - started; try std.testing.expect(elapsed_ns < std.time.ns_per_s);}
test "the head deadline fires when the Io would run an async task inline" { const allocator = std.testing.allocator; var threaded: std.Io.Threaded = .init(allocator, .{ .async_limit = .nothing }); defer threaded.deinit(); const io = threaded.io();
var origin = try TestOrigin.init(io); defer origin.deinit(); const url = try std.fmt.allocPrint(allocator, "http://127.0.0.1:{d}/silent", .{origin.port()}); defer allocator.free(url);
var got_request: std.atomic.Value(bool) = .init(false); var release: std.atomic.Value(bool) = .init(false); var server = try std.Thread.spawn(.{}, struct { fn run(o: *TestOrigin, got: *std.atomic.Value(bool), rel: *std.atomic.Value(bool)) void { o.serveHeldOpen("", got, rel) catch {}; } }.run, .{ &origin, &got_request, &release }); defer server.join(); defer release.store(true, .release);
var transport = Client.init(io, allocator); defer transport.deinit(); transport.dead_connection_attempts = 1; transport.head_deadline_ns = 300 * std.time.ns_per_ms;
const started = std.Io.Timestamp.now(io, .awake).nanoseconds; try std.testing.expectError(error.ResponseHeadTimeout, transport.fetch(.{ .url = url })); const elapsed_ns = std.Io.Timestamp.now(io, .awake).nanoseconds - started; try std.testing.expect(elapsed_ns < 2 * std.time.ns_per_s);}
test "the stall guard fires when the Io would run an async task inline" { const allocator = std.testing.allocator; var threaded: std.Io.Threaded = .init(allocator, .{ .async_limit = .nothing }); defer threaded.deinit(); const io = threaded.io();
var origin = try TestOrigin.init(io); defer origin.deinit(); const url = try std.fmt.allocPrint(allocator, "http://127.0.0.1:{d}/stall", .{origin.port()}); defer allocator.free(url);
var got_request: std.atomic.Value(bool) = .init(false); var release: std.atomic.Value(bool) = .init(false); var server = try std.Thread.spawn(.{}, struct { fn run(o: *TestOrigin, got: *std.atomic.Value(bool), rel: *std.atomic.Value(bool)) void { o.serveHeldOpen("HTTP/1.1 200 OK\r\nContent-Length: 1048576\r\n\r\nhello", got, rel) catch {}; } }.run, .{ &origin, &got_request, &release }); defer server.join(); defer release.store(true, .release);
var transport = Client.init(io, allocator); defer transport.deinit(); transport.dead_connection_attempts = 1; transport.stall = .{ .idle_ns = 300 * std.time.ns_per_ms, .min_bytes_per_s = 0, .grace_ns = 0 };
const started = std.Io.Timestamp.now(io, .awake).nanoseconds; try std.testing.expectError(error.TransferStalled, transport.fetch(.{ .url = url })); const elapsed_ns = std.Io.Timestamp.now(io, .awake).nanoseconds - started; try std.testing.expect(elapsed_ns < 2 * std.time.ns_per_s);}
test "without concurrency a request still completes, unguarded" { const allocator = std.testing.allocator; var threaded: std.Io.Threaded = .init(allocator, .{ .async_limit = .nothing, .concurrent_limit = .nothing }); defer threaded.deinit(); const io = threaded.io();
var origin = try TestOrigin.init(io); defer origin.deinit(); const url = try std.fmt.allocPrint(allocator, "http://127.0.0.1:{d}/x", .{origin.port()}); defer allocator.free(url);
var server = try std.Thread.spawn(.{}, struct { fn run(o: *TestOrigin) void { o.serveOnce("HTTP/1.1 200 OK\r\nContent-Length: 2\r\n\r\nok") catch {}; } }.run, .{&origin}); defer server.join();
var transport = Client.init(io, allocator); defer transport.deinit();
const log_level = std.testing.log_level; std.testing.log_level = .err; defer std.testing.log_level = log_level;
var res = try transport.fetch(.{ .url = url, .deadline_ns = 5 * std.time.ns_per_s }); defer res.deinit(allocator); try std.testing.expectEqual(std.http.Status.ok, res.status); try std.testing.expectEqualStrings("ok", res.body);}
test "a connection closed before it answers is replayed, not surfaced" { const allocator = std.testing.allocator; var threaded: std.Io.Threaded = .init(allocator, .{}); defer threaded.deinit(); const io = threaded.io();
var origin = try TestOrigin.init(io); defer origin.deinit(); const url = try std.fmt.allocPrint(allocator, "http://127.0.0.1:{d}/x", .{origin.port()}); defer allocator.free(url);
var server = try std.Thread.spawn(.{}, struct { fn run(o: *TestOrigin) void { o.serveDeadThenOk("HTTP/1.1 200 OK\r\nContent-Length: 2\r\n\r\nok") catch {}; } }.run, .{&origin}); defer server.join();
var transport = Client.init(io, allocator); defer transport.deinit();
var res = try transport.fetch(.{ .url = url }); defer res.deinit(allocator); try std.testing.expectEqual(std.http.Status.ok, res.status); try std.testing.expectEqualStrings("ok", res.body);}
test "a pooled connection reset instead of answered is replayed, not surfaced" { const allocator = std.testing.allocator; var threaded: std.Io.Threaded = .init(allocator, .{}); defer threaded.deinit(); const io = threaded.io();
var origin = try TestOrigin.init(io); defer origin.deinit(); const url = try std.fmt.allocPrint(allocator, "http://127.0.0.1:{d}/x", .{origin.port()}); defer allocator.free(url);
var server = try std.Thread.spawn(.{}, struct { fn run(o: *TestOrigin) void { o.serveOkThenReset("HTTP/1.1 200 OK\r\nContent-Length: 2\r\n\r\nok") catch {}; } }.run, .{&origin}); defer server.join();
var transport = Client.init(io, allocator); defer transport.deinit();
var first = try transport.fetch(.{ .url = url }); defer first.deinit(allocator); try std.testing.expectEqualStrings("ok", first.body);
var second = try transport.fetch(.{ .url = url }); defer second.deinit(allocator); try std.testing.expectEqualStrings("ok", second.body);}
test "a pooled connection silently blackholed while idle costs one head-deadline stall, then self-clears" { const allocator = std.testing.allocator; var threaded: std.Io.Threaded = .init(allocator, .{}); defer threaded.deinit(); const io = threaded.io();
var origin = try TestOrigin.init(io); defer origin.deinit(); const url = try std.fmt.allocPrint(allocator, "http://127.0.0.1:{d}/x", .{origin.port()}); defer allocator.free(url);
var reused: std.atomic.Value(bool) = .init(false); var server = try std.Thread.spawn(.{}, struct { fn run(o: *TestOrigin, flag: *std.atomic.Value(bool)) void { o.serveOkThenBlackhole("HTTP/1.1 200 OK\r\nContent-Length: 2\r\n\r\nok", flag) catch {}; } }.run, .{ &origin, &reused }); defer server.join();
var transport = Client.init(io, allocator); defer transport.deinit(); transport.head_deadline_ns = 300 * std.time.ns_per_ms;
var first = try transport.fetch(.{ .url = url }); defer first.deinit(allocator); try std.testing.expectEqual(std.http.Status.ok, first.status);
// The second fetch reuses the pooled keep-alive connection. The origin // reads the request and never answers — no error is ever delivered, so // only the head deadline catches it, and nothing replays it: this is the // caller-visible failure (what a proxy in front of it would surface as a // 502 at exactly the deadline). try std.testing.expectError(error.ResponseHeadTimeout, transport.fetch(.{ .url = url })); try std.testing.expect(reused.load(.acquire));
// The abandoned connection must not be re-pooled: the next fetch dials // fresh and succeeds, so each stale socket costs exactly one stall. var third = try transport.fetch(.{ .url = url }); defer third.deinit(allocator); try std.testing.expectEqual(std.http.Status.ok, third.status); try std.testing.expectEqualStrings("ok", third.body);}
test "authorization does not follow a redirect to another host" { const allocator = std.testing.allocator; var threaded: std.Io.Threaded = .init(allocator, .{}); defer threaded.deinit(); const io = threaded.io();
var first = try TestOrigin.init(io); defer first.deinit(); var second = try TestOrigin.init(io); defer second.deinit();
// Both listen on loopback, but `localhost` and `127.0.0.1` are different // hosts as far as a redirect is concerned, which is the cross-origin case. const redirect = try std.fmt.allocPrint( allocator, "HTTP/1.1 302 Found\r\nLocation: http://127.0.0.1:{d}/elsewhere\r\nContent-Length: 0\r\n\r\n", .{second.port()}, ); defer allocator.free(redirect); const url = try std.fmt.allocPrint(allocator, "http://localhost:{d}/start", .{first.port()}); defer allocator.free(url);
var first_saw_auth: std.atomic.Value(bool) = .init(false); var second_saw_auth: std.atomic.Value(bool) = .init(false); const Server = struct { fn run(o: *TestOrigin, response: []const u8, flag: *std.atomic.Value(bool)) void { o.serveRecordingAuth(response, flag) catch {}; } }; var t1 = try std.Thread.spawn(.{}, Server.run, .{ &first, redirect, &first_saw_auth }); defer t1.join(); var t2 = try std.Thread.spawn(.{}, Server.run, .{ &second, "HTTP/1.1 200 OK\r\nContent-Length: 2\r\n\r\nok", &second_saw_auth }); defer t2.join();
var client = Client.init(io, allocator); defer client.deinit(); var res = try client.fetch(.{ .url = url, .authorization = "Bearer secret" }); defer res.deinit(allocator);
try std.testing.expectEqual(std.http.Status.ok, res.status); try std.testing.expect(first_saw_auth.load(.acquire)); try std.testing.expect(!second_saw_auth.load(.acquire));}
fn redirectResponse(allocator: std.mem.Allocator, status: []const u8, location: []const u8) ![]u8 { return std.fmt.allocPrint(allocator, "HTTP/1.1 {s}\r\nLocation: {s}\r\nContent-Length: 0\r\n\r\n", .{ status, location });}
const ok_response = "HTTP/1.1 200 OK\r\nContent-Length: 2\r\n\r\nok";
test "a redirect back to a visited url is a loop, not a hang" { const allocator = std.testing.allocator; var threaded: std.Io.Threaded = .init(allocator, .{}); defer threaded.deinit(); const io = threaded.io();
var origin = try TestOrigin.init(io); defer origin.deinit(); const url = try std.fmt.allocPrint(allocator, "http://127.0.0.1:{d}/self", .{origin.port()}); defer allocator.free(url); const response = try redirectResponse(allocator, "302 Found", "/self"); defer allocator.free(response);
var server = try std.Thread.spawn(.{}, struct { fn run(o: *TestOrigin, r: []const u8) void { o.serveOnce(r) catch {}; } }.run, .{ &origin, response }); defer server.join();
var client = Client.init(io, allocator); defer client.deinit(); try std.testing.expectError(error.RedirectLoop, client.fetch(.{ .url = url }));}
test "hops past the policy are an error; the policy is per client" { const allocator = std.testing.allocator; var threaded: std.Io.Threaded = .init(allocator, .{}); defer threaded.deinit(); const io = threaded.io();
var origin = try TestOrigin.init(io); defer origin.deinit(); const url = try std.fmt.allocPrint(allocator, "http://127.0.0.1:{d}/0", .{origin.port()}); defer allocator.free(url);
// /0 -> /1 -> /2: two hops, each a distinct url. const r0 = try redirectResponse(allocator, "302 Found", "/1"); defer allocator.free(r0); const r1 = try redirectResponse(allocator, "302 Found", "/2"); defer allocator.free(r1);
var server = try std.Thread.spawn(.{}, struct { fn run(o: *TestOrigin, a: []const u8, b: []const u8) void { o.serveOnce(a) catch {}; o.serveOnce(b) catch {}; } }.run, .{ &origin, r0, r1 }); defer server.join();
var client = Client.init(io, allocator); defer client.deinit(); client.keep_alive = false; client.redirects.max_hops = 1; try std.testing.expectError(error.TooManyRedirects, client.fetch(.{ .url = url }));}
test "surface hands the 3xx back; refuse makes it an error" { const allocator = std.testing.allocator; var threaded: std.Io.Threaded = .init(allocator, .{}); defer threaded.deinit(); const io = threaded.io();
var origin = try TestOrigin.init(io); defer origin.deinit(); const url = try std.fmt.allocPrint(allocator, "http://127.0.0.1:{d}/x", .{origin.port()}); defer allocator.free(url); const response = try redirectResponse(allocator, "301 Moved Permanently", "http://example.invalid/never-dialed"); defer allocator.free(response);
var server = try std.Thread.spawn(.{}, struct { fn run(o: *TestOrigin, r: []const u8) void { o.serveOnce(r) catch {}; o.serveOnce(r) catch {}; } }.run, .{ &origin, response }); defer server.join();
var client = Client.init(io, allocator); defer client.deinit();
var surfaced = try client.fetch(.{ .url = url, .on_redirect = .surface }); defer surfaced.deinit(allocator); try std.testing.expectEqual(std.http.Status.moved_permanently, surfaced.status);
try std.testing.expectError(error.RedirectRefused, client.fetch(.{ .url = url, .on_redirect = .refuse }));}
test "https never steps down to http unless the policy allows it" { // No TLS in the loopback fixture, so build the check's inputs directly. const https = try std.Uri.parse("https://a.example/x"); const http = try std.Uri.parse("http://a.example/x"); try std.testing.expect(try sameOrigin(https, https)); try std.testing.expect(!try sameOrigin(https, http)); try std.testing.expect(try sameOrigin(try std.Uri.parse("http://A.example:80/"), try std.Uri.parse("http://a.example/"))); try std.testing.expect(!try sameOrigin(try std.Uri.parse("http://a.example:8080/"), try std.Uri.parse("http://a.example/")));}
test "303 and 301-on-POST become a bodiless GET; 307 resends the body" { const allocator = std.testing.allocator; var threaded: std.Io.Threaded = .init(allocator, .{}); defer threaded.deinit(); const io = threaded.io();
var origin = try TestOrigin.init(io); defer origin.deinit(); const url = try std.fmt.allocPrint(allocator, "http://127.0.0.1:{d}/post", .{origin.port()}); defer allocator.free(url);
const Recorded = struct { line: [128]u8 = undefined, len: usize = 0, body: std.atomic.Value(bool) = .init(false), fn text(self: *const @This()) []const u8 { return self.line[0..self.len]; } }; var after_303: Recorded = .{}; var after_301: Recorded = .{}; var after_307: Recorded = .{};
const r303 = try redirectResponse(allocator, "303 See Other", "/after-303"); defer allocator.free(r303); const r301 = try redirectResponse(allocator, "301 Moved Permanently", "/after-301"); defer allocator.free(r301); const r307 = try redirectResponse(allocator, "307 Temporary Redirect", "/after-307"); defer allocator.free(r307);
var server = try std.Thread.spawn(.{}, struct { fn run(o: *TestOrigin, a: []const u8, b: []const u8, c: []const u8, ra: *Recorded, rb: *Recorded, rc: *Recorded) void { o.serveOnce(a) catch {}; o.serveRecordingRequest(ok_response, &ra.line, &ra.len, &ra.body) catch {}; o.serveOnce(b) catch {}; o.serveRecordingRequest(ok_response, &rb.line, &rb.len, &rb.body) catch {}; o.serveOnce(c) catch {}; o.serveRecordingRequest(ok_response, &rc.line, &rc.len, &rc.body) catch {}; } }.run, .{ &origin, r303, r301, r307, &after_303, &after_301, &after_307 }); defer server.join();
var client = Client.init(io, allocator); defer client.deinit(); client.keep_alive = false; const post: Client.FetchOptions = .{ .url = url, .method = .POST, .payload = "{}", .content_type = "application/json" };
var a = try client.fetch(post); a.deinit(allocator); try std.testing.expectEqualStrings("GET /after-303 HTTP/1.1", after_303.text()); try std.testing.expect(!after_303.body.load(.acquire));
var b = try client.fetch(post); b.deinit(allocator); try std.testing.expectEqualStrings("GET /after-301 HTTP/1.1", after_301.text()); try std.testing.expect(!after_301.body.load(.acquire));
var c = try client.fetch(post); c.deinit(allocator); try std.testing.expectEqualStrings("POST /after-307 HTTP/1.1", after_307.text()); try std.testing.expect(after_307.body.load(.acquire));}
test "a payload without a content type is refused up front" { const io = std.Options.debug_io; var client = Client.init(io, std.testing.allocator); defer client.deinit(); try std.testing.expectError(error.MissingContentType, client.fetch(.{ .url = "http://127.0.0.1:1/", .method = .POST, .payload = "{}" }));}
test "retry-after accepts the http-date form and delaySeconds ranks the hints" { const H = Client.RateLimitHeaders; // Sun, 06 Nov 1994 08:49:37 GMT is 784111777. const head = try std.http.Client.Response.Head.parse("HTTP/1.1 503 x\r\nRetry-After: Sun, 06 Nov 1994 08:49:37 GMT\r\n\r\n"); const dated = H.fromResponseHead(head); try std.testing.expectEqual(@as(?i64, 784_111_777), dated.retry_after_at); try std.testing.expectEqual(@as(?u64, 7), dated.delaySeconds(784_111_770)); try std.testing.expectEqual(@as(?u64, 0), dated.delaySeconds(784_111_800));
try std.testing.expectEqual(@as(?u64, 5), (H{ .retry_after = 5, .reset = 1_800_000_000 }).delaySeconds(1_700_000_000)); try std.testing.expectEqual(@as(?u64, 100_000_000), (H{ .reset = 1_800_000_000 }).delaySeconds(1_700_000_000)); try std.testing.expectEqual(@as(?u64, 30), (H{ .reset = 30 }).delaySeconds(1_700_000_000)); try std.testing.expectEqual(@as(?u64, null), (H{ .limit = 10 }).delaySeconds(0));
try std.testing.expectEqual(@as(?i64, 784_111_777), parseHttpDate("Sunday, 06-Nov-94 08:49:37 GMT")); try std.testing.expectEqual(@as(?i64, 0), parseHttpDate("Thu, 01 Jan 1970 00:00:00 GMT")); try std.testing.expectEqual(@as(?i64, 951_782_400), parseHttpDate("Tue, 29 Feb 2000 00:00:00 GMT"));}
test "a transfer that started fast and now trickles is judged on the trickle" { const allocator = std.testing.allocator; var threaded: std.Io.Threaded = .init(allocator, .{}); defer threaded.deinit(); const io = threaded.io();
var origin = try TestOrigin.init(io); defer origin.deinit(); const url = try std.fmt.allocPrint(allocator, "http://127.0.0.1:{d}/big", .{origin.port()}); defer allocator.free(url);
// 4 MiB lands at once. A lifetime average would read as megabytes per // second for a long time; the trailing window sees nothing moving. var server = try std.Thread.spawn(.{}, struct { fn run(o: *TestOrigin) void { o.serveBurstThenStall(4 * 1024 * 1024) catch {}; } }.run, .{&origin}); defer server.join();
var client = Client.init(io, allocator); defer client.deinit(); client.stall = .{ .idle_ns = 0, .min_bytes_per_s = 1024 * 1024, .grace_ns = 400 * std.time.ns_per_ms, .window_ns = 300 * std.time.ns_per_ms, }; const started = std.Io.Timestamp.now(io, .awake).nanoseconds; try std.testing.expectError(error.TransferTooSlow, client.fetch(.{ .url = url })); const elapsed_ns = std.Io.Timestamp.now(io, .awake).nanoseconds - started; try std.testing.expect(elapsed_ns < 3 * std.time.ns_per_s);}
test "the request deadline bounds a dribbling body, with the stall guard on or off" { const allocator = std.testing.allocator; var threaded: std.Io.Threaded = .init(allocator, .{}); defer threaded.deinit(); const io = threaded.io();
var origin = try TestOrigin.init(io); defer origin.deinit(); const url = try std.fmt.allocPrint(allocator, "http://127.0.0.1:{d}/slow", .{origin.port()}); defer allocator.free(url);
var server = try std.Thread.spawn(.{}, struct { fn run(o: *TestOrigin) void { o.serveStalling() catch {}; o.serveStalling() catch {}; } }.run, .{&origin}); defer server.join();
var client = Client.init(io, allocator); defer client.deinit(); client.keep_alive = false;
// Guards on: the deadline has to cancel through the stall watcher and // its reader task. const started = std.Io.Timestamp.now(io, .awake).nanoseconds; try std.testing.expectError(error.RequestDeadlineExceeded, client.fetch(.{ .url = url, .deadline_ns = 300 * std.time.ns_per_ms, })); // Guards off: the deadline cancels a plain blocking read. try std.testing.expectError(error.RequestDeadlineExceeded, client.fetch(.{ .url = url, .deadline_ns = 300 * std.time.ns_per_ms, .stall = .{ .idle_ns = 0, .min_bytes_per_s = 0 }, })); const elapsed_ns = std.Io.Timestamp.now(io, .awake).nanoseconds - started; try std.testing.expect(elapsed_ns < 3 * std.time.ns_per_s);}
test "a followed redirect reports where the response came from" { const allocator = std.testing.allocator; var threaded: std.Io.Threaded = .init(allocator, .{}); defer threaded.deinit(); const io = threaded.io();
var origin = try TestOrigin.init(io); defer origin.deinit(); const url = try std.fmt.allocPrint(allocator, "http://127.0.0.1:{d}/start", .{origin.port()}); defer allocator.free(url); const expected = try std.fmt.allocPrint(allocator, "http://127.0.0.1:{d}/final", .{origin.port()}); defer allocator.free(expected); const response = try redirectResponse(allocator, "302 Found", "/final"); defer allocator.free(response);
var server = try std.Thread.spawn(.{}, struct { fn run(o: *TestOrigin, r: []const u8) void { o.serveOnce(r) catch {}; o.serveOnce(ok_response) catch {}; o.serveOnce(ok_response) catch {}; } }.run, .{ &origin, response }); defer server.join();
var client = Client.init(io, allocator); defer client.deinit(); client.keep_alive = false; var res = try client.fetch(.{ .url = url }); defer res.deinit(allocator); try std.testing.expectEqualStrings(expected, res.final_url.?);
var direct = try client.fetch(.{ .url = expected }); defer direct.deinit(allocator); try std.testing.expectEqual(@as(?[]u8, null), direct.final_url);}
test "control characters in a url or header are refused before anything is sent" { const io = std.Options.debug_io; var client = Client.init(io, std.testing.allocator); defer client.deinit(); try std.testing.expectError(error.InvalidUrl, client.fetch(.{ .url = "http://127.0.0.1:1/x\r\nx-injected: 1\r\n" })); try std.testing.expectError(error.InvalidHeader, client.fetch(.{ .url = "http://127.0.0.1:1/", .authorization = "Bearer a\r\nx-injected: 1" })); try std.testing.expectError(error.InvalidHeader, client.fetch(.{ .url = "http://127.0.0.1:1/", .extra_headers = &.{.{ .name = "x-a", .value = "1\nx-b: 2" }}, })); try std.testing.expectError(error.InvalidHeader, client.fetch(.{ .url = "http://127.0.0.1:1/", .extra_headers = &.{.{ .name = "x-a: b", .value = "1" }}, }));}
test "the response cap is enforced while streaming, not by preallocating it" { const allocator = std.testing.allocator; var threaded: std.Io.Threaded = .init(allocator, .{}); defer threaded.deinit(); const io = threaded.io();
var origin = try TestOrigin.init(io); defer origin.deinit(); const url = try std.fmt.allocPrint(allocator, "http://127.0.0.1:{d}/x", .{origin.port()}); defer allocator.free(url);
var server = try std.Thread.spawn(.{}, struct { fn run(o: *TestOrigin) void { o.serveOnce("HTTP/1.1 200 OK\r\nContent-Length: 5\r\n\r\nhello") catch {}; o.serveOnce("HTTP/1.1 200 OK\r\nContent-Length: 5\r\n\r\nhello") catch {}; } }.run, .{&origin}); defer server.join();
var client = Client.init(io, allocator); defer client.deinit(); client.keep_alive = false; client.max_response_size = 4; try std.testing.expectError(error.ResponseTooLarge, client.fetch(.{ .url = url }));
var res = try client.fetch(.{ .url = url, .max_response_size = 1 << 30 }); defer res.deinit(allocator); try std.testing.expectEqualStrings("hello", res.body);}
test "open then stream: the sink is chosen after the head, and the body never touches the heap" { const allocator = std.testing.allocator; var threaded: std.Io.Threaded = .init(allocator, .{}); defer threaded.deinit(); const io = threaded.io();
var origin = try TestOrigin.init(io); defer origin.deinit(); const url = try std.fmt.allocPrint(allocator, "http://127.0.0.1:{d}/repo", .{origin.port()}); defer allocator.free(url);
// A 64 KiB body, then a 404 with a json body: the two shapes a bulk // downloader sees, wanting a file for one and a few bytes for the other. const big = try allocator.alloc(u8, 64 * 1024); defer allocator.free(big); @memset(big, 'z'); const big_response = try std.fmt.allocPrint(allocator, "HTTP/1.1 200 OK\r\nContent-Length: {d}\r\n\r\n{s}", .{ big.len, big }); defer allocator.free(big_response); const not_found = "HTTP/1.1 404 Not Found\r\nContent-Type: application/json\r\nContent-Length: 24\r\n\r\n{\"error\":\"RepoNotFound\"}";
var server = try std.Thread.spawn(.{}, struct { fn run(o: *TestOrigin, a: []const u8, b: []const u8) void { o.serveOnce(a) catch {}; o.serveOnce(b) catch {}; } }.run, .{ &origin, big_response, not_found }); defer server.join();
var client = Client.init(io, allocator); defer client.deinit(); client.keep_alive = false;
var tmp = std.testing.tmpDir(.{}); defer tmp.cleanup();
{ const response = try client.open(.{ .url = url }); defer response.deinit(); try std.testing.expectEqual(std.http.Status.ok, response.status);
var file = try tmp.dir.createFile(io, "repo.car", .{ .read = true }); defer file.close(io); var wbuf: [8 * 1024]u8 = undefined; var fw = std.Io.File.Writer.initStreaming(file, io, &wbuf); var progress: Progress = .{}; const n = try response.stream(&fw.interface, .{ .progress = &progress }); try fw.interface.flush(); try std.testing.expectEqual(big.len, n); try std.testing.expectEqual(big.len, progress.bytes.load(.acquire)); try std.testing.expect(progress.done.load(.acquire)); try std.testing.expectEqual(big.len, try file.length(io)); try std.testing.expectError(error.BodyAlreadyTaken, response.stream(&fw.interface, .{})); }
{ const response = try client.open(.{ .url = url, .capture_headers = &.{"content-type"} }); defer response.deinit(); try std.testing.expectEqual(std.http.Status.not_found, response.status); try std.testing.expectEqualStrings("application/json", response.headers.get("content-type").?); var small: [64]u8 = undefined; var fixed = std.Io.Writer.fixed(&small); _ = try response.stream(&fixed, .{ .max_bytes = small.len }); try std.testing.expectEqualStrings("{\"error\":\"RepoNotFound\"}", fixed.buffered()); }}
test "a response dropped without streaming leaves the client usable" { const allocator = std.testing.allocator; var threaded: std.Io.Threaded = .init(allocator, .{}); defer threaded.deinit(); const io = threaded.io();
var origin = try TestOrigin.init(io); defer origin.deinit(); const url = try std.fmt.allocPrint(allocator, "http://127.0.0.1:{d}/x", .{origin.port()}); defer allocator.free(url);
var server = try std.Thread.spawn(.{}, struct { fn run(o: *TestOrigin) void { o.serveOnce(ok_response) catch {}; o.serveOnce(ok_response) catch {}; } }.run, .{&origin}); defer server.join();
var client = Client.init(io, allocator); defer client.deinit();
const dropped = try client.open(.{ .url = url }); dropped.deinit();
var res = try client.fetch(.{ .url = url }); defer res.deinit(allocator); try std.testing.expectEqualStrings("ok", res.body);}
test "what the request deadline has left after the head bounds the body" { const allocator = std.testing.allocator; var threaded: std.Io.Threaded = .init(allocator, .{}); defer threaded.deinit(); const io = threaded.io();
var origin = try TestOrigin.init(io); defer origin.deinit(); const url = try std.fmt.allocPrint(allocator, "http://127.0.0.1:{d}/slow", .{origin.port()}); defer allocator.free(url);
var server = try std.Thread.spawn(.{}, struct { fn run(o: *TestOrigin) void { o.serveStalling() catch {}; } }.run, .{&origin}); defer server.join();
var client = Client.init(io, allocator); defer client.deinit();
const response = try client.open(.{ .url = url, .deadline_ns = 300 * std.time.ns_per_ms }); defer response.deinit(); try std.testing.expect(response.deadline_remaining_ns > 0); try std.testing.expect(response.deadline_remaining_ns <= 300 * std.time.ns_per_ms);
var aw: std.Io.Writer.Allocating = .init(allocator); defer aw.deinit(); const started = std.Io.Timestamp.now(io, .awake).nanoseconds; try std.testing.expectError(error.RequestDeadlineExceeded, response.stream(&aw.writer, .{ .stall = .{ .idle_ns = 0, .min_bytes_per_s = 0 }, })); try std.testing.expect(std.Io.Timestamp.now(io, .awake).nanoseconds - started < 2 * std.time.ns_per_s);}
test "accept_encoding replaces std's default instead of adding a second header" { const allocator = std.testing.allocator; var threaded: std.Io.Threaded = .init(allocator, .{}); defer threaded.deinit(); const io = threaded.io();
var origin = try TestOrigin.init(io); defer origin.deinit(); const url = try std.fmt.allocPrint(allocator, "http://127.0.0.1:{d}/x", .{origin.port()}); defer allocator.free(url);
var default_head: [4096]u8 = undefined; var default_len: usize = 0; var identity_head: [4096]u8 = undefined; var identity_len: usize = 0; var server = try std.Thread.spawn(.{}, struct { fn run(o: *TestOrigin, a: *[4096]u8, al: *usize, b: *[4096]u8, bl: *usize) void { o.serveRecordingHead(ok_response, a, al) catch {}; o.serveRecordingHead(ok_response, b, bl) catch {}; } }.run, .{ &origin, &default_head, &default_len, &identity_head, &identity_len }); defer server.join();
var client = Client.init(io, allocator); defer client.deinit(); client.keep_alive = false;
var a = try client.fetch(.{ .url = url }); a.deinit(allocator); var b = try client.fetch(.{ .url = url, .accept_encoding = "identity" }); b.deinit(allocator);
try std.testing.expect(std.ascii.findIgnoreCase(default_head[0..default_len], "\r\naccept-encoding: gzip") != null); const identity = identity_head[0..identity_len]; try std.testing.expect(std.ascii.findIgnoreCase(identity, "\r\naccept-encoding: identity\r\n") != null); try std.testing.expectEqual(@as(usize, 1), std.mem.count(u8, identity, "ccept-encoding"));}
test "a request that may have taken effect is never replayed" { try std.testing.expect(!isReplayable(.{ .url = "http://x/", .method = .POST })); try std.testing.expect(!isReplayable(.{ .url = "http://x/", .method = .GET, .payload = "b" })); try std.testing.expect(isReplayable(.{ .url = "http://x/", .method = .GET }));}
test "HTTP dates accept legacy formats and reject invalid calendars" { const H = Client.RateLimitHeaders; inline for (.{ "Sunday, 06-Nov-94 08:49:37 GMT", "Sun Nov 6 08:49:37 1994", "Sun Nov 06 08:49:37 1994" }) |date| { const head = try std.http.Client.Response.Head.parse("HTTP/1.1 503 x\r\nRetry-After: " ++ date ++ "\r\n\r\n"); const hints = H.fromResponseHead(head); try std.testing.expectEqual(@as(?i64, 784_111_777), hints.retry_after_at); try std.testing.expectEqual(@as(?u64, 7), hints.delaySeconds(784_111_770)); try std.testing.expectEqual(@as(?u64, 0), hints.delaySeconds(784_111_800)); } for ([_][]const u8{ "Sun, 31 Feb 1994 08:49:37 GMT", "Sunday, 29-Feb-99 08:49:37 GMT", "Sun Feb 29 08:49:37 1900", "Sun Nov 6 24:49:37 1994", "Sun, 06 Nov 1994 08:49:60 GMT", "Nop Nov 6 08:49:37 1994", "Sunday, 06-Nov-94 08:49:37 GMT trailing", "", }) |date| try std.testing.expectEqual(@as(?i64, null), parseHttpDate(date)); try std.testing.expectEqual(@as(?i64, 951_782_400), parseHttpDate("Tuesday, 29-Feb-00 00:00:00 GMT"));}
test "rate limit reset retains signed timestamps for protocol retry policy" { const head = try std.http.Client.Response.Head.parse("HTTP/1.1 429 Too Many Requests\r\nRateLimit-Reset: -1\r\nRetry-After: 9\r\n\r\n"); const headers = Client.RateLimitHeaders.fromResponseHead(head); try std.testing.expect(headers.reset != null); try std.testing.expectEqual(@as(i128, -1), @as(i128, headers.reset.?)); try std.testing.expectEqual(@as(?u64, 9), headers.retry_after);}
test "signed reset rejects overflow and generic delay clamps expired values" { inline for (.{ "9223372036854775808", "-9223372036854775809", "bad" }) |value| { const head = try std.http.Client.Response.Head.parse("HTTP/1.1 429 Too Many Requests\r\nRateLimit-Reset: " ++ value ++ "\r\n\r\n"); try std.testing.expectEqual(@as(?i64, null), Client.RateLimitHeaders.fromResponseHead(head).reset); } try std.testing.expectEqual(@as(?u64, 0), (Client.RateLimitHeaders{ .reset = -1 }).delaySeconds(100)); try std.testing.expectEqual(@as(?u64, 9), (Client.RateLimitHeaders{ .reset = -1, .retry_after = 9 }).delaySeconds(100));}
test "stream delivers buffered body before waiting for the next network read" { const allocator = std.testing.allocator; var threaded: std.Io.Threaded = .init(allocator, .{}); defer threaded.deinit(); const io = threaded.io(); var origin = try TestOrigin.init(io); defer origin.deinit(); var progress: Progress = .{}; var observed: std.atomic.Value(bool) = .init(false); const server = try std.Thread.spawn(.{}, struct { fn run(o: *TestOrigin, p: *Progress, seen: *std.atomic.Value(bool)) void { var conn = o.listener.accept(o.io) catch return; defer conn.close(o.io); var rb: [4096]u8 = undefined; var r = conn.reader(o.io, &rb); _ = r.interface.takeDelimiterInclusive('\n') catch return; var w = conn.writer(o.io, &.{}); w.interface.writeAll("HTTP/1.1 200 OK\r\nContent-Type: text/event-stream\r\nConnection: close\r\n\r\ndata: first\n\n") catch return; w.interface.flush() catch return; for (0..50) |_| { if (p.bytes.load(.acquire) > 0) { seen.store(true, .release); break; } std.Io.Clock.Duration.sleep(.{ .raw = .fromMilliseconds(10), .clock = .awake }, o.io) catch return; } w.interface.writeAll("data: [DONE]\n\n") catch return; w.interface.flush() catch return; } }.run, .{ &origin, &progress, &observed }); defer server.join(); var client = Client.init(io, allocator); defer client.deinit(); const url = try std.fmt.allocPrint(allocator, "http://127.0.0.1:{d}/stream", .{origin.port()}); defer allocator.free(url); var response = try client.open(.{ .url = url }); defer response.deinit(); var body: std.Io.Writer.Allocating = .init(allocator); defer body.deinit(); _ = try response.stream(&body.writer, .{ .progress = &progress }); try std.testing.expect(observed.load(.acquire)); try std.testing.expectEqualStrings("data: first\n\ndata: [DONE]\n\n", body.written());}