Something went wrong. Try again.
jetstream v2 in zig stream.waow.tech
Something went wrong. Try again.
1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586878889909192939495969798991001011021031041051061071081091101111121131141151161171181191201211221231241251261271281291301311321331341351361371381391401411421431441451461471481491501511521531541551561571581591601611621631641651661671681691701711721731741751761771781791801811821831841851861871881891901911921931941951961971981992002012022032042052062072082092102112122132142152162172182192202212222232242252262272282292302312322332342352362372382392402412422432442452462472482492502512522532542552562572582592602612622632642652662672682692702712722732742752762772782792802812822832842852862872882892902912922932942952962972982993003013023033043053063073083093103113123133143153163173183193203213223233243253263273283293303313323333343353363373383393403413423433443453463473483493503513523533543553563573583593603613623633643653663673683693703713723733743753763773783793803813823833843853863873883893903913923933943953963973983994004014024034044054064074084094104114124134144154164174184194204214224234244254264274284294304314324334344354364374384394404414424434444454464474484494504514524534544554564574584594604614624634644654664674684694704714724734744754764774784794804814824834844854864874884894904914924934944954964974984995005015025035045055065075085095105115125135145155165175185195205215225235245255265275285295305315325335345355365375385395405415425435445455465475485495505515525535545555565575585595605615625635645655665675685695705715725735745755765775785795805815825835845855865875885895905915925935945955965975985996006016026036046056066076086096106116126136146156166176186196206216226236246256266276286296306316326336346356366376386396406416426436446456466476486496506516526536546556566576586596606616626636646656666676686696706716726736746756766776786796806816826836846856866876886896906916926936946956966976986997007017027037047057067077087097107117127137147157167177187197207217227237247257267277287297307317327337347357367377387397407417427437447457467477487497507517527537547557567577587597607617627637647657667677687697707717727737747757767777787797807817827837847857867877887897907917927937947957967977987998008018028038048058068078088098108118128138148158168178188198208218228238248258268278288298308318328338348358368378388398408418428438448458468478488498508518528538548558568578588598608618628638648658668678688698708718728738748758768778788798808818828838848858868878888898908918928938948958968978988999009019029039049059069079089099109119129139149159169179189199209219229239249259269279289299309319329339349359369379389399409419429439449459469479489499509519529539549559569579589599609619629639649659669679689699709719729739749759769779789799809819829839849859869879889899909919929939949959969979989991000100110021003100410051006100710081009101010111012101310141015101610171018101910201021102210231024102510261027102810291030103110321033103410351036103710381039104010411042104310441045104610471048104910501051105210531054105510561057105810591060106110621063106410651066106710681069107010711072107310741075107610771078107910801081108210831084108510861087108810891090109110921093109410951096109710981099110011011102110311041105110611071108110911101111111211131114111511161117111811191120112111221123112411251126112711281129113011311132113311341135113611371138113911401141114211431144114511461147114811491150115111521153115411551156115711581159116011611162116311641165116611671168116911701171117211731174117511761177117811791180118111821183118411851186118711881189119011911192119311941195119611971198119912001201120212031204120512061207120812091210121112121213121412151216121712181219122012211222122312241225122612271228122912301231123212331234123512361237123812391240124112421243124412451246124712481249125012511252125312541255125612571258125912601261126212631264126512661267126812691270127112721273127412751276127712781279128012811282128312841285128612871288128912901291129212931294129512961297129812991300130113021303130413051306130713081309131013111312131313141315131613171318131913201321132213231324132513261327132813291330133113321333133413351336133713381339134013411342134313441345134613471348134913501351135213531354135513561357135813591360136113621363136413651366136713681369137013711372137313741375137613771378137913801381138213831384138513861387138813891390139113921393139413951396139713981399140014011402140314041405140614071408140914101411141214131414141514161417141814191420142114221423142414251426142714281429143014311432143314341435143614371438143914401441144214431444144514461447144814491450145114521453145414551456145714581459146014611462146314641465146614671468146914701471147214731474147514761477147814791480148114821483148414851486148714881489149014911492149314941495149614971498149915001501150215031504150515061507150815091510151115121513151415151516151715181519152015211522152315241525152615271528152915301531153215331534153515361537153815391540154115421543154415451546154715481549155015511552155315541555155615571558155915601561156215631564156515661567156815691570157115721573157415751576157715781579158015811582158315841585158615871588158915901591159215931594159515961597159815991600160116021603160416051606160716081609161016111612161316141615161616171618161916201621162216231624162516261627162816291630163116321633163416351636163716381639164016411642164316441645164616471648164916501651165216531654165516561657165816591660166116621663166416651666166716681669167016711672167316741675167616771678167916801681168216831684168516861687168816891690169116921693169416951696169716981699170017011702170317041705170617071708170917101711171217131714171517161717171817191720172117221723172417251726172717281729173017311732173317341735173617371738173917401741174217431744174517461747174817491750175117521753175417551756175717581759176017611762176317641765176617671768176917701771177217731774177517761777177817791780178117821783178417851786178717881789179017911792179317941795179617971798179918001801180218031804180518061807180818091810181118121813181418151816181718181819182018211822182318241825182618271828182918301831183218331834183518361837183818391840184118421843184418451846184718481849185018511852185318541855185618571858185918601861186218631864186518661867186818691870187118721873187418751876187718781879188018811882188318841885188618871888188918901891189218931894189518961897189818991900190119021903190419051906190719081909191019111912191319141915191619171918191919201921192219231924192519261927192819291930193119321933193419351936193719381939194019411942194319441945194619471948194919501951195219531954195519561957195819591960196119621963196419651966196719681969197019711972197319741975197619771978197919801981198219831984198519861987198819891990199119921993199419951996199719981999200020012002200320042005200620072008200920102011201220132014201520162017201820192020202120222023202420252026202720282029203020312032203320342035203620372038203920402041204220432044204520462047204820492050205120522053205420552056205720582059206020612062206320642065206620672068206920702071207220732074207520762077207820792080208120822083208420852086208720882089209020912092209320942095209620972098209921002101210221032104210521062107210821092110211121122113211421152116211721182119212021212122212321242125212621272128212921302131213221332134213521362137213821392140214121422143214421452146214721482149215021512152215321542155215621572158215921602161216221632164216521662167216821692170217121722173217421752176217721782179218021812182218321842185218621872188218921902191219221932194219521962197219821992200220122022203220422052206220722082209221022112212221322142215221622172218221922202221222222232224222522262227222822292230223122322233223422352236223722382239224022412242224322442245224622472248224922502251225222532254225522562257225822592260226122622263226422652266226722682269227022712272227322742275227622772278227922802281228222832284228522862287228822892290229122922293229422952296229722982299230023012302230323042305230623072308230923102311231223132314231523162317231823192320232123222323232423252326232723282329233023312332233323342335233623372338233923402341234223432344234523462347234823492350235123522353235423552356235723582359236023612362236323642365236623672368236923702371237223732374237523762377237823792380238123822383238423852386238723882389239023912392239323942395239623972398239924002401240224032404240524062407240824092410241124122413241424152416241724182419242024212422242324242425242624272428242924302431243224332434243524362437243824392440244124422443244424452446244724482449245024512452245324542455245624572458245924602461246224632464246524662467246824692470247124722473247424752476247724782479248024812482248324842485248624872488248924902491249224932494249524962497249824992500250125022503250425052506250725082509251025112512251325142515251625172518251925202521252225232524252525262527252825292530253125322533253425352536253725382539254025412542254325442545254625472548254925502551255225532554255525562557255825592560256125622563256425652566256725682569257025712572257325742575257625772578257925802581258225832584258525862587258825892590259125922593259425952596259725982599260026012602260326042605260626072608260926102611261226132614261526162617261826192620262126222623262426252626262726282629263026312632263326342635263626372638263926402641264226432644264526462647264826492650265126522653265426552656265726582659266026612662266326642665266626672668266926702671267226732674267526762677267826792680268126822683268426852686268726882689269026912692269326942695//! /subscribe websocket server — legacy-compat live tail//!//! each subscriber runs a pull loop on its own thread, reading the shared//! hot tail at its own cursor (no per-client queues). filters apply at read//! time; options_update swaps them atomically. contract and quirks per//! docs/upstream-subscribe-protocol.md.//!//! v1 and v2 share the pull fanout while retaining their distinct wire,//! cursor-floor, and skipped-event contracts.
const std = @import("std");const websocket = @import("websocket");const zat = @import("zat");const archive_mod = @import("../storage/archive.zig");const cold = @import("../storage/cold.zig");const disk_space = @import("../storage/disk_space.zig");const endpoints = @import("endpoints.zig");const filter_mod = @import("filter.zig");const homepage = @import("homepage.zig");const http_time = @import("http_time.zig");const metrics = @import("../runtime/metrics.zig");const process_metrics = @import("../runtime/process_metrics.zig");const repo_action_limiter = @import("repo_action_limiter.zig");const repo_export = @import("repo_export.zig");const repo_store = @import("../ingest/backfill/repo_store.zig");const phase_mod = @import("../ingest/backfill/phase.zig");const status_page = @import("status_page.zig");const api_keys = @import("api_keys.zig");const tail_mod = @import("tail.zig");const wire = @import("wire.zig");const xrpcapi = @import("xrpcapi.zig");const zstd = @import("../storage/zstd.zig");
const Io = std.Io;const Allocator = std.mem.Allocator;const log = std.log.scoped(.stream);
/// 8 MB, matching the Io.Threaded stacks (docs/lessons-from-zlay.md #1)const subscriber_stack_size: usize = 8 * 1024 * 1024;const ping_stack_size: usize = subscriber_stack_size;const slow_lag_threshold: u64 = 100_000;pub const default_slow_window_ns: u64 = 60 * std.time.ns_per_s;pub const default_slow_min_rate: f64 = 5;pub const default_subscribe_read_batch: usize = 1024;const ping_interval: Io.Clock.Duration = .{ .raw = Io.Duration.fromSeconds(30), .clock = .awake };const frame_write_timeout_ms: u32 = 5_000;
const SlowDetector = struct { window_ns: u64 = default_slow_window_ns, min_rate: f64 = default_slow_min_rate, streak_start_ns: i96 = 0, streak_start_pos: u64 = 0, in_streak: bool = false,
fn observe(self: *SlowDetector, now_ns: i96, pos: u64, lag: u64) bool { if (lag <= slow_lag_threshold) { self.in_streak = false; return false; } if (!self.in_streak) { self.in_streak = true; self.streak_start_ns = now_ns; self.streak_start_pos = pos; return false; } const elapsed_ns = now_ns - self.streak_start_ns; if (elapsed_ns < self.window_ns) return false; const progress = pos -| self.streak_start_pos; const elapsed_seconds = @as(f64, @floatFromInt(elapsed_ns)) / std.time.ns_per_s; const rate = @as(f64, @floatFromInt(progress)) / elapsed_seconds; if (rate >= self.min_rate) { self.streak_start_ns = now_ns; self.streak_start_pos = pos; return false; } return true; }};
const CursorPlan = struct { /// null means live: anchor at the published tip when delivery starts, /// after requireHello if requested. Replay plans are fixed pre-upgrade. start_idx: ?u64 = null, /// First raw network sequence to scan. This advances even when filtering, /// v1 skip rules, or message caps suppress every frame in a batch. start_seq: ?u64 = null, mode: metrics.CursorMode = .live, /// v2 timestamp cursor clamped to the retention floor: the subscriber is /// owed one #info OutdatedCursor frame before events (lexicon contract) outdated_info: bool = false,};
pub const Hub = struct { allocator: Allocator, io: Io, tail: *tail_mod.Tail, stats: *metrics.Stats, archive: ?*archive_mod.Archive = null, cold_reader: ?*cold.ColdReader = null, xrpc: ?*xrpcapi.Api = null, repo_status: ?*repo_store.Store = null, data_dir: []const u8 = ".", /// Identity directory used by the public account diagnostics collector. /// Production always configures it, matching upstream's resolver-first /// handle semantics; tests may disable resolution explicitly. plc_url: []const u8 = "https://plc.directory", status_identity_resolution: bool = true, repo_action_rate_limits: bool = true, repo_action_limiter: repo_action_limiter.Limiter = .{}, active: std.atomic.Value(u32) = .init(0), /// admission caps, 0 = unbounded. a recorded divergence: upstream keeps /// connection admission at its CDN/proxy layer (docs/README.md §7); we /// have no CDN, and only the process knows a cursor is about to cost /// archive reads (docs/semantic-parity.md). max_subscribers: u32 = 0, max_cold_readers: u32 = 0, cold_active: std.atomic.Value(u32) = .init(0), bootstrap_enabled: bool = false, compaction_enabled: bool = false, retry_enabled: bool = false, /// Upstream cursor replay window. Zero makes both endpoints pure-live and /// classifies an otherwise-valid cursor as disabled. cursor_lookback_ns: u64 = 36 * std.time.ns_per_hour, subscribe_slow_window_ns: u64 = default_slow_window_ns, subscribe_slow_min_rate: f64 = default_slow_min_rate, subscribe_read_batch: usize = default_subscribe_read_batch, /// serving gate: false until the bootstrap lifecycle reaches /// steady_state — subscribe + xrpc return 503 (upstream contract); /// /healthz and /metrics stay reachable serving: std.atomic.Value(bool) = .init(true), /// Set before listener cancellation begins. An upgrade that was already /// accepted but has not reached afterInit observes this gate and sends the /// same 1001 shutdown close instead of escaping the drain snapshot. draining: std.atomic.Value(bool) = .init(false), /// True only after the public listener and the optional debug listener /// have both been bound and their accept loops started. This is the /// upstream /readyz contract; it deliberately says nothing about ingest /// health or bootstrap steady state. listeners_ready: std.atomic.Value(bool) = .init(false), /// False while main is still opening storage (active-segment recovery, /// archive manifest, rocksdb) behind an already-accepting listener. /// Upstream starts its server first and 503-gates data paths until ready /// (jetstreamd runtime.go: lifecycle.IsSteadyState on subscribe, the /// xrpcapi Ready hook on the archive surface); this is that gate. The /// release store in main publishes the hub's store pointers, so any path /// that observes true may dereference them. Defaults true so test hubs /// that publish pointers before serving are unaffected. storage_ready: std.atomic.Value(bool) = .init(true), /// per-key archive metering ring, rendered into /metrics when set key_ring: ?*api_keys.KeyRing = null,
/// Bounded cold-replay concurrency. fetchAdd-then-undo keeps the pair /// wait-free; transient overshoot is bounded by the number of racing /// subscriber threads and never admits past the cap. fn tryAcquireColdSlot(self: *Hub) bool { if (self.max_cold_readers == 0) return true; if (self.cold_active.fetchAdd(1, .acquire) >= self.max_cold_readers) { _ = self.cold_active.fetchSub(1, .release); return false; } return true; }
fn releaseColdSlot(self: *Hub) void { if (self.max_cold_readers == 0) return; _ = self.cold_active.fetchSub(1, .release); }
pub fn deinit(self: *Hub) void { self.repo_action_limiter.deinit(self.allocator); }};
pub const ListenerRole = enum { /// Historical Stream mode: protocol and operator routes share one port. legacy, /// Upstream public listener: protocol routes only. public, /// Upstream debug listener: health, readiness, and metrics only. debug,};
pub const ListenerContext = struct { hub: *Hub, role: ListenerRole,};
fn contextHub(context: anytype) *Hub { const T = @TypeOf(context); if (T == *Hub) return context; if (T == *ListenerContext) return context.hub; @compileError("Stream websocket listener context must be *Hub or *ListenerContext");}
fn contextRole(context: anytype) ListenerRole { return if (@TypeOf(context) == *Hub) .legacy else context.role;}
const Subscriber = struct { hub: *Hub, conn: *websocket.Conn, /// guarded by opts_mutex; swapped whole by options_update filter: filter_mod.Filter, opts_mutex: Io.Mutex = .init, /// uncompressed-size cap per event; 0 = no cap. quirk: garbage → 0. max_msg_size: std.atomic.Value(u64), idx: u64, cursor_seq: u64, alive: std.atomic.Value(bool) = .init(true), /// Serializes socket writes with close signaling. A normal peer close must /// wait for an in-flight write and prevent the next one before closing the /// descriptor; otherwise Threaded treats the resulting BADF as a bug. write_mutex: Io.Mutex = .init, /// inverse of alive, in the shape tail.next wants for its stop check stopped: std.atomic.Value(bool) = .init(false), /// Consecutive cold passes that made no progress while hot rows were /// resident: a gap between the archive and the retained hot range, which /// upstream treats as a contract violation and disconnects on. cold_stall_passes: u8 = 0, /// v1 zstd scheme: per-message dictionary frames as BINARY ws messages. /// the cap still compares the UNCOMPRESSED length (documented upstream /// divergence from original v1, kept deliberately). compressor: ?zstd.DictCompressor = null, /// network.bsky.jetstream.subscribeEvents: proposal-0015 wire (lexicon /// message frames, #sync, strict seq cursors) v2: bool = false, thread: ?std.Thread = null, ping_thread: ?std.Thread = null, /// Neither worker may touch the connection or trigger Handler.close until /// both join handles are installed and the subscriber is published. start_gate: Io.Event = .unset, ping_stop: Io.Event = .unset, slow: SlowDetector = .{},
/// the pull loop. cleanup happens in Handler.close, which joins this /// thread — the subscriber never frees itself, so the handler can always /// safely signal it. fn run(self: *Subscriber) void { self.start_gate.waitUncancelable(self.hub.io); if (!self.alive.load(.acquire)) return; defer self.workerDone(); while (self.alive.load(.acquire)) { // Upstream Tail.ReadFrom: the hot/cold boundary is decided per // read from the cursor alone — no sticky replay mode, so there is // no handoff to miss (the 2026-08-08 seam wedge). cursor_seq == 0 // means seq-unknown (live attach or a time-indexed start): follow // self.idx hot until the first delivered entry teaches us a seq. if (self.cursor_seq != 0) { const committed = if (self.hub.archive) |archive| archive.committed_seq.load(.acquire) else 0; if (self.hub.tail.indexForSeq(self.cursor_seq)) |idx| { self.idx = idx; // resident (or at tip, which blocks): hot self.cold_stall_passes = 0; } else if (self.cursor_seq > committed) { // nothing archived at/past the cursor: wait at the tip // for the next append (upstream guards the cold delegate // with cursor < nextSeq and blocks otherwise — cold is // never probed to discover exhaustion) self.cold_stall_passes = 0; self.idx = self.hub.tail.publishedTip(); } else switch (self.coldPass()) { .progressed => { self.cold_stall_passes = 0; continue; }, .empty => { // committed says rows exist at/past the cursor yet the // cold snapshot served none: a rotation seam. Upstream // retries once, then disconnects rather than spinning; // the client's reconnect resolves a fresh position. self.cold_stall_passes +|= 1; if (self.cold_stall_passes >= 2) { log.err("subscriber cursor {d} unserveable: cold exhausted below committed {d} (rotation seam invariant violated)", .{ self.cursor_seq, committed }); _ = self.hub.stats.subscribe_cold_stalls_total.fetchAdd(1, .monotonic); return self.noteExit(.cold_stall); } continue; }, .failed => return self.noteExit(.cold_failed), .contended => { // cold-reader slots full: brief pause and retry — // deliberate backpressure, not a stall or an error self.cold_stall_passes = 0; self.hub.io.sleep(Io.Duration.fromMilliseconds(20), .awake) catch return self.noteExit(.cold_failed); continue; }, } }
var arena = std.heap.ArenaAllocator.init(self.hub.allocator); defer arena.deinit(); const result = self.hub.tail.readBatch( arena.allocator(), &self.idx, &self.stopped, self.v2, self.hub.subscribe_read_batch, ) catch |err| { log.warn("subscriber read batch failed: {s}", .{@errorName(err)}); return self.noteExit(.read_batch_error); }; switch (result) { .closed => { // readBatch reports .closed for both "the tail is gone" // (shutdown) and "this subscriber was stopped" (its // handler saw the socket close). Different stories. return self.noteExit(if (self.stopped.load(.acquire)) .client_closed else .tail_closed); }, .fell_behind => { // evicted from under us: the next pass re-decides from // cursor_seq (seq-unknown subscribers clamp to the floor, // v1's documented below-floor semantics) if (self.cursor_seq == 0) self.idx = self.hub.tail.gauges().base; continue; }, .batch => |batch| { _ = self.hub.stats.subscribe_hot_reads_total.fetchAdd(1, .monotonic); for (batch.entries) |entry| { if (entry.seq != 0) self.cursor_seq = entry.seq +| 1; if (!self.wantsPred(entry.kind, entry.did, entry.collection, entry.skip_v1)) continue; const cap = self.max_msg_size.load(.monotonic); if (cap != 0 and entry.json.len > cap) { _ = self.hub.stats.subscribe_events_oversize_total.fetchAdd(1, .monotonic); continue; } const wrote = self.write(entry.json) catch |err| { if (err == error.Closed) return self.noteExit(.write_closed); log.warn("subscriber write failed: {s}", .{@errorName(err)}); return self.noteExit(.write_error); }; if (wrote == .encode_error) _ = self.hub.stats.subscribe_encode_errors_total.fetchAdd(1, .monotonic); } }, } if (self.tooSlow()) return self.noteExit(.too_slow); } }
fn noteExit(self: *Subscriber, reason: metrics.SubscriberExit) void { _ = self.hub.stats.subscribe_worker_exits.getPtr(reason).fetchAdd(1, .monotonic); }
fn pingLoop(self: *Subscriber) void { self.start_gate.waitUncancelable(self.hub.io); if (!self.alive.load(.acquire)) return; while (self.alive.load(.acquire)) { self.ping_stop.waitTimeout(self.hub.io, .{ .duration = ping_interval }) catch |err| switch (err) { error.Timeout => {}, error.Canceled => return, }; if (!self.alive.load(.acquire)) return; self.write_mutex.lockUncancelable(self.hub.io); if (!self.alive.load(.acquire)) { self.write_mutex.unlock(self.hub.io); return; } var payload: [0]u8 = .{}; self.conn.writePing(&payload) catch { self.noteExit(.ping_write_error); const interrupt = self.alive.swap(false, .acq_rel); self.stopped.store(true, .release); self.write_mutex.unlock(self.hub.io); self.ping_stop.set(self.hub.io); self.hub.tail.wake(); if (interrupt) self.conn.interruptRead(); return; }; self.write_mutex.unlock(self.hub.io); } }
/// A subscriber-worker failure must end the connection, not leave a /// ping-only zombie. Half-shutdown wakes the server reader; that reader /// remains the sole owner of final descriptor cleanup. fn workerDone(self: *Subscriber) void { self.write_mutex.lockUncancelable(self.hub.io); const interrupt = self.alive.swap(false, .acq_rel); self.stopped.store(true, .release); self.write_mutex.unlock(self.hub.io); self.ping_stop.set(self.hub.io); self.hub.tail.wake(); if (interrupt) self.conn.interruptRead(); }
fn write(self: *Subscriber, json: []const u8) !cold.WriteResult { self.write_mutex.lockUncancelable(self.hub.io); defer self.write_mutex.unlock(self.hub.io); if (!self.alive.load(.acquire)) return error.Closed; const scheme = self.compression(); if (self.compressor) |*comp| { const frame = comp.compress(self.hub.allocator, json) catch { return .encode_error; }; defer self.hub.allocator.free(frame); try self.conn.writeBin(frame); _ = self.hub.stats.subscribe_events_sent.getPtr(scheme).fetchAdd(1, .monotonic); _ = self.hub.stats.subscribe_bytes_sent.getPtr(scheme).fetchAdd(@intCast(frame.len), .monotonic); } else { try self.conn.writeText(json); _ = self.hub.stats.subscribe_events_sent.getPtr(scheme).fetchAdd(1, .monotonic); _ = self.hub.stats.subscribe_bytes_sent.getPtr(scheme).fetchAdd(@intCast(json.len), .monotonic); } _ = self.hub.stats.subscribe_bytes_encoded.getPtr(scheme).fetchAdd(@intCast(json.len), .monotonic); return .sent; }
fn compression(self: *const Subscriber) metrics.Compression { if (self.compressor != null) return .zstd; return if (self.conn.compression != null) .deflate else .none; }
const ColdPass = enum { progressed, empty, failed, contended };
/// One stateless cold read from cursor_seq (upstream Tail.cold). No stop /// floor: overlapping the hot range is harmless because the cursor is /// monotonic and the caller's next pass re-decides hot-vs-cold anyway. fn coldPass(self: *Subscriber) ColdPass { const hub = self.hub; const archive = hub.archive orelse return .failed; const reader = hub.cold_reader orelse return .failed; const from_seq = self.cursor_seq; if (!hub.tryAcquireColdSlot()) { _ = hub.stats.subscribe_cold_contended_total.fetchAdd(1, .monotonic); return .contended; } defer hub.releaseColdSlot(); _ = hub.stats.subscribe_cold_reads_total.fetchAdd(1, .monotonic); const batch = reader.readSeqBatch( archive, self, wantsPred, self, coldWriteCached, coldObserve, self.v2, self.compressor != null, from_seq, 0, hub.subscribe_read_batch, ) catch |err| { log.warn("seq cold replay failed: {s}", .{@errorName(err)}); return .failed; }; log.debug("seq cold batch scanned {d} entries and served {d} frames", .{ batch.scanned, batch.frames }); if (batch.next_seq <= from_seq) return .empty; self.cursor_seq = batch.next_seq; if (self.tooSlow()) return .failed; return .progressed; }
fn tooSlow(self: *Subscriber) bool { // seq-unknown (idx-anchored) subscribers have no measurable lag; a // 0 cursor would read as astronomically behind and drop idle live // connections after one quiet window. if (self.cursor_seq == 0) return false; var tip = self.hub.tail.publishedNextSeq(); if (self.hub.archive) |archive| tip = @max(tip, archive.committed_seq.load(.acquire) +| 1); const lag = tip -| self.cursor_seq; if (!self.slow.observe(Io.Timestamp.now(self.hub.io, .real).toNanoseconds(), self.cursor_seq, lag)) return false; _ = self.hub.stats.subscribe_adversarial_drops_total.fetchAdd(1, .monotonic); if (self.v2) { // proposal-0015: an error frame is terminal — best-effort send, // then the caller disconnects (upstream handler.go sendError) var err_buf: [256]u8 = undefined; var w: std.Io.Writer = .fixed(&err_buf); var msg_buf: [160]u8 = undefined; const msg = std.fmt.bufPrint(&msg_buf, "reading below the floor rate {d} events behind the tip; reconnect with cursor={d}", .{ lag, self.cursor_seq }) catch "reading below the floor rate"; wire.encodeErrorV2(&w, "ConsumerTooSlow", msg) catch {}; if (w.buffered().len > 0) _ = self.write(w.buffered()) catch {}; } return true; }
fn coldWriteCached(self: *Subscriber, frame: cold.CachedFrame) anyerror!cold.WriteResult { if (!self.alive.load(.acquire)) return error.Closed; const cap = self.max_msg_size.load(.monotonic); if (cap != 0 and frame.body.len > cap) { _ = self.hub.stats.subscribe_events_oversize_total.fetchAdd(1, .monotonic); return .oversize; } if (!frame.compressed) return self.write(frame.body);
self.write_mutex.lockUncancelable(self.hub.io); defer self.write_mutex.unlock(self.hub.io); if (!self.alive.load(.acquire)) return error.Closed; try self.conn.writeBin(frame.payload); _ = self.hub.stats.subscribe_events_sent.getPtr(.zstd).fetchAdd(1, .monotonic); _ = self.hub.stats.subscribe_bytes_sent.getPtr(.zstd).fetchAdd(@intCast(frame.payload.len), .monotonic); _ = self.hub.stats.subscribe_bytes_encoded.getPtr(.zstd).fetchAdd(@intCast(frame.body.len), .monotonic); return .sent; }
fn coldObserve(self: *Subscriber, _: u64, result: cold.ReplayResult) anyerror!void { switch (result) { .skipped_sync => _ = self.hub.stats.subscribe_events_skipped_sync_total.fetchAdd(1, .monotonic), .skipped_resync => _ = self.hub.stats.subscribe_events_skipped_resync_total.fetchAdd(1, .monotonic), .encode_error => _ = self.hub.stats.subscribe_encode_errors_total.fetchAdd(1, .monotonic), .sent, .filtered, .oversize => {}, } }
fn wantsPred(self: *Subscriber, kind: wire.Kind, did: []const u8, collection: []const u8, skip: tail_mod.SkipV1) bool { switch (skip) { .sync => { _ = self.hub.stats.subscribe_events_skipped_sync_total.fetchAdd(1, .monotonic); return false; }, .resync_replacement => { _ = self.hub.stats.subscribe_events_skipped_resync_total.fetchAdd(1, .monotonic); return false; }, .none => {}, } self.opts_mutex.lockUncancelable(self.hub.io); defer self.opts_mutex.unlock(self.hub.io); const wanted = self.filter.wants(kind, did, collection); if (!wanted) _ = self.hub.stats.subscribe_events_filtered_total.fetchAdd(1, .monotonic); return wanted; }
/// atomically replace the whole filter (options_update contract) fn updateOptions(self: *Subscriber, new_filter: filter_mod.Filter, max_msg_size: u64) void { self.opts_mutex.lockUncancelable(self.hub.io); var old = self.filter; self.filter = new_filter; self.opts_mutex.unlock(self.hub.io); old.deinit(); self.max_msg_size.store(max_msg_size, .monotonic); }};
const HttpObservation = struct { hub: *Hub, started: Io.Timestamp, handler: ?metrics.HttpHandler, method: metrics.HttpMethod, get_block: bool = false, code: u16 = 500, served_bytes: u64 = 0,
fn init(hub: *Hub, path: []const u8, method: []const u8) HttpObservation { return .{ .hub = hub, .started = Io.Timestamp.now(hub.io, .awake), .handler = metrics.httpHandler(path, method), .method = metrics.httpMethod(method), }; }
fn finish(self: *HttpObservation) void { const elapsed = self.started.durationTo(Io.Timestamp.now(self.hub.io, .awake)).toMicroseconds(); const duration_us: u64 = @intCast(@max(0, elapsed)); if (self.handler) |handler| self.hub.stats.observeHttp(handler, self.method, self.code, duration_us); if (self.get_block) self.hub.stats.observeGetBlock(self.code, self.served_bytes, duration_us); }};
pub const subscribe_events_path = "/xrpc/network.bsky.jetstream.subscribeEvents";/// the one subprotocol the v2 endpoint speaks (proposal 0015); also the/// lexicon-declared default, so negotiation is a pure header echopub const v2_subprotocol = "xrpc.v1.json";
pub const Handler = struct { hub: *Hub, conn: *websocket.Conn, subscriber: ?*Subscriber = null, /// options parsed from the query string, held until the subscriber /// starts (immediately, or on hello when requireHello) pending_filter: ?filter_mod.Filter, pending_max_msg_size: u64, cursor_plan: CursorPlan, awaiting_hello: bool, compress: bool, v2: bool, clean_disconnect_counted: bool = false, http_started: Io.Timestamp, http_handler: metrics.HttpHandler, http_observed: bool = false,
pub fn init(handshake: *const websocket.Handshake, conn: *websocket.Conn, context: anytype) !Handler { const hub = contextHub(context); const url = handshake.url; const qi = std.mem.indexOfScalar(u8, url, '?'); const query = if (qi) |i| url[i + 1 ..] else ""; const path = if (qi) |i| url[0..i] else url; const http_started = Io.Timestamp.now(hub.io, .awake); const http_handler = metrics.httpHandler(path, "GET"); const finishHttp = struct { fn call(h: *Hub, started: Io.Timestamp, handler: ?metrics.HttpHandler, code: u16) void { const label = handler orelse return; const elapsed = started.durationTo(Io.Timestamp.now(h.io, .awake)).toMicroseconds(); h.stats.observeHttp(label, .get, code, @intCast(@max(0, elapsed))); } }.call; if (contextRole(context) == .debug or (!std.mem.eql(u8, path, "/subscribe") and !std.mem.eql(u8, path, subscribe_events_path))) { respond(conn, "404 Not Found", "text/plain", "not found\n"); finishHttp(hub, http_started, http_handler, 404); return error.Close; } if (!hub.serving.load(.acquire)) { if (queryParam(query, "cursor")) |raw| { if (raw.len != 0) _ = hub.stats.subscribe_cursor_requests.getPtr(.unavailable).fetchAdd(1, .monotonic); } respond(conn, "503 Service Unavailable", "text/plain", "bootstrap in progress\n"); finishHttp(hub, http_started, http_handler, 503); return error.Close; // response already written; Close suppresses the 400 } const v2 = std.mem.eql(u8, path, subscribe_events_path); if (v2) { // proposal-0015 negotiation degenerates to a header echo: the // lexicon-declared default IS xrpc.v1.json and it is the only // token we speak, so an offer is echoed and a missing offer // changes nothing (unnegotiated connections get identical // framing). Unrecognized tokens are simply not echoed. if (handshake.headers.get("sec-websocket-protocol")) |offered| { var toks = std.mem.splitScalar(u8, offered, ','); while (toks.next()) |tok| { if (std.mem.eql(u8, std.mem.trim(u8, tok, " "), v2_subprotocol)) { handshake.res_headers.add("Sec-WebSocket-Protocol", v2_subprotocol); break; } } } // the legacy names are rejected with the replacement named, // never silently ignored (upstream docs README §5.2) if (queryParam(query, "wantedCollections") != null) { respond(conn, "400 Bad Request", "application/json", "{\"error\":\"InvalidRequest\",\"message\":\"wantedCollections is the legacy v1 parameter; use collections\"}\n"); finishHttp(hub, http_started, http_handler, 400); return error.Close; } if (queryParam(query, "wantedDids") != null) { respond(conn, "400 Bad Request", "application/json", "{\"error\":\"InvalidRequest\",\"message\":\"wantedDids is the legacy v1 parameter; use dids\"}\n"); finishHttp(hub, http_started, http_handler, 400); return error.Close; } }
var compress = false; const legacy_compress = if (queryParam(query, "compress")) |v| std.mem.eql(u8, v, "true") else false; const header_zstd = if (handshake.headers.get("socket-encoding")) |enc| std.mem.indexOf(u8, enc, "zstd") != null else false; if (v2) { if (legacy_compress or header_zstd) { respond(conn, "400 Bad Request", "text/plain", "compress=true / Socket-Encoding: zstd is the /subscribe (v1) opt-in; subscribeEvents uses zstdDictionary=<id> with the dictionary from getZstdDictionary\n"); finishHttp(hub, http_started, http_handler, 400); return error.Close; } if (queryParam(query, "zstdDictionary")) |raw| if (raw.len != 0) { const id = std.fmt.parseInt(u32, raw, 10) catch { respond(conn, "400 Bad Request", "text/plain", "zstdDictionary must be a positive integer zstd dictionary ID\n"); finishHttp(hub, http_started, http_handler, 400); return error.Close; }; if (id == 0) { respond(conn, "400 Bad Request", "text/plain", "zstdDictionary must be a positive integer zstd dictionary ID\n"); finishHttp(hub, http_started, http_handler, 400); return error.Close; } if (id != zstd.v2_dictionary_id) { var buf: [224]u8 = undefined; const body = std.fmt.bufPrint(&buf, "{{\"error\":\"UnknownZstdDictionary\",\"message\":\"dictionary {d} is no longer current; fetch dictionary {d} and reconnect\"}}\n", .{ id, zstd.v2_dictionary_id }) catch unreachable; respond(conn, "400 Bad Request", "application/json", body); finishHttp(hub, http_started, http_handler, 400); return error.Close; } compress = true; }; } else { compress = legacy_compress or header_zstd; if (compress) if (handshake.headers.get("sec-websocket-extensions")) |extensions| { if (offersPerMessageDeflate(extensions)) { respond(conn, "400 Bad Request", "text/plain", "choose one compression scheme: custom zstd (compress=true / Socket-Encoding: zstd) or RFC 7692 permessage-deflate, not both\n"); finishHttp(hub, http_started, http_handler, 400); return error.Close; } }; } // Upstream only negotiates RFC 7692 on ordinary v1 subscriptions. // V2 and either custom-zstd mode explicitly decline the extension. if (v2 or compress) conn.disableCompression();
var filter = (if (v2) filter_mod.Filter.parseQueryV2(hub.allocator, query) else filter_mod.Filter.parseQuery(hub.allocator, query)) catch |err| { log.debug("subscribe: invalid options: {s}", .{@errorName(err)}); if (v2 and err != error.OutOfMemory) { const msg = switch (err) { error.InvalidKind => "{\"error\":\"InvalidRequest\",\"message\":\"kinds values must be commit, identity, account, or sync\"}\n", error.CollectionsWithoutCommits => "{\"error\":\"InvalidRequest\",\"message\":\"collections can never apply when kinds excludes commit\"}\n", else => "{\"error\":\"InvalidRequest\",\"message\":\"invalid filter parameters\"}\n", }; respond(conn, "400 Bad Request", "application/json", msg); finishHttp(hub, http_started, http_handler, 400); return error.Close; } finishHttp(hub, http_started, http_handler, if (err == error.OutOfMemory) 500 else 400); return error.InvalidRequest; }; errdefer filter.deinit();
// Cursor resolution includes parsing and happens before upgrade, so // malformed and too-old v2 cursors remain ordinary HTTP failures. const resolve_started = Io.Timestamp.now(hub.io, .real).toMicroseconds(); const cursor_plan = resolveCursor(hub, v2, queryParam(query, "cursor")) catch |err| { const finished = Io.Timestamp.now(hub.io, .real).toMicroseconds(); hub.stats.observeCursorResolve(@intCast(@max(finished - resolve_started, 0))); if (err == error.CursorResolveFailed) { _ = hub.stats.subscribe_cursor_requests.getPtr(.resolve_failed).fetchAdd(1, .monotonic); respond(conn, "503 Service Unavailable", "text/plain", "service not ready: cursor resolution failed\n"); finishHttp(hub, http_started, http_handler, 503); return error.Close; } if (err == error.CursorTooOld) { _ = hub.stats.subscribe_cursor_requests.getPtr(.too_old).fetchAdd(1, .monotonic); const requested = std.fmt.parseInt(i64, queryParam(query, "cursor").?, 10) catch 0; const floor = if (hub.archive) |archive| archive.manifest.lookbackFloor(Io.Timestamp.now(hub.io, .real).toMicroseconds(), hub.cursor_lookback_ns).seq else 0; var body_buf: [256]u8 = undefined; const body = if (v2) std.fmt.bufPrint(&body_buf, "{{\"error\":\"CursorTooOld\",\"message\":\"cursor {d} is below the lookback floor {d}; re-backfill from your last seq via the archive\"}}\n", .{ requested, floor }) catch "{\"error\":\"CursorTooOld\"}\n" else std.fmt.bufPrint(&body_buf, "subscribe: cursor too old: cursor {d} below lookback floor {d}; re-backfill from your last seq\n", .{ requested, floor }) catch "subscribe: cursor too old\n"; respond(conn, "400 Bad Request", if (v2) "application/json" else "text/plain; charset=utf-8", body); finishHttp(hub, http_started, http_handler, 400); return error.Close; } finishHttp(hub, http_started, http_handler, 400); return error.InvalidRequest; }; const resolve_finished = Io.Timestamp.now(hub.io, .real).toMicroseconds(); hub.stats.observeCursorResolve(@intCast(@max(resolve_finished - resolve_started, 0))); _ = hub.stats.subscribe_cursor_requests.getPtr(cursor_plan.mode).fetchAdd(1, .monotonic);
// quirk: requireHello is true iff the value is exactly "true". // v2 is server-push only — the hello mechanism does not exist there. const awaiting_hello = if (!v2) if (queryParam(query, "requireHello")) |v| std.mem.eql(u8, v, "true") else false else false;
conn.writeTimeout(frame_write_timeout_ms) catch |err| { finishHttp(hub, http_started, http_handler, 500); return err; };
return .{ .hub = hub, .conn = conn, .pending_filter = filter, .pending_max_msg_size = parseMaxMsgSize(queryParam(query, "maxMessageSizeBytes")), .cursor_plan = cursor_plan, .awaiting_hello = awaiting_hello, .compress = compress, .v2 = v2, .http_started = http_started, .http_handler = http_handler orelse unreachable, }; }
/// runs after the handshake reply is on the wire — safe to start writing. /// with requireHello, delivery is deferred until the first valid /// options_update (no pre-hello buffering: the start index is resolved /// when the subscriber actually starts). pub fn afterInit(self: *Handler, context: anytype) !void { _ = context; if (self.hub.draining.load(.acquire)) { self.conn.writeClose(.{ .code = 1001, .reason = "server shutting down" }) catch {}; self.noteCleanDisconnect(); return error.ServerDraining; } if (self.awaiting_hello) return; try self.startSubscriber(); }
fn startSubscriber(self: *Handler) !void { const hub = self.hub; if (hub.max_subscribers > 0 and hub.active.load(.monotonic) >= hub.max_subscribers) { _ = hub.stats.subscribe_capacity_rejects_total.fetchAdd(1, .monotonic); self.conn.writeClose(.{ .code = 1013, .reason = "at capacity; retry later" }) catch {}; self.noteCleanDisconnect(); return error.AtCapacity; } const plan = self.cursor_plan; const start_idx = plan.start_idx orelse hub.tail.publishedTip(); // cursor_seq == 0 means seq-unknown: live and time-indexed plans are // idx-anchored (live must NOT replay the resident suffix) and learn // their seq from the first delivered entry. Seq-cursor plans dispatch // hot-vs-cold from cursor_seq on every read-loop pass. const start_seq = plan.start_seq orelse 0;
var compressor: ?zstd.DictCompressor = if (self.compress) if (self.v2) try zstd.DictCompressor.initLevel(&zstd.v2_dictionary, 1) else try zstd.DictCompressor.init(zstd.v1_dictionary) else null; errdefer if (compressor) |*comp| comp.deinit();
const sub = try hub.allocator.create(Subscriber); sub.* = .{ .hub = hub, .conn = self.conn, .filter = self.pending_filter.?, .max_msg_size = .init(self.pending_max_msg_size), .idx = start_idx, .cursor_seq = start_seq, .v2 = self.v2, .compressor = compressor, .slow = .{ .window_ns = hub.subscribe_slow_window_ns, .min_rate = hub.subscribe_slow_min_rate, }, }; self.pending_filter = null; // ownership moved to the subscriber if (self.v2 and plan.outdated_info) { // lexicon contract: a clamped timestamp cursor owes the client // one seq-less #info OutdatedCursor frame before any event var info_buf: [256]u8 = undefined; var w: std.Io.Writer = .fixed(&info_buf); wire.encodeInfoV2(&w, "OutdatedCursor", "requested timestamp was below retention; resumed at the floor") catch {}; if (w.buffered().len > 0) _ = sub.write(w.buffered()) catch {}; } const scheme = sub.compression(); sub.thread = std.Thread.spawn(.{ .stack_size = subscriber_stack_size }, Subscriber.run, .{sub}) catch |err| { sub.filter.deinit(); hub.allocator.destroy(sub); return err; }; sub.ping_thread = std.Thread.spawn(.{ .stack_size = ping_stack_size }, Subscriber.pingLoop, .{sub}) catch |err| { log.err("subscriber ping thread spawn failed: {s}", .{@errorName(err)}); sub.alive.store(false, .release); sub.stopped.store(true, .release); sub.start_gate.set(hub.io); sub.thread.?.join(); sub.filter.deinit(); hub.allocator.destroy(sub); return err; }; // Publish only after both join handles exist. A replay failure may // interrupt the server reader immediately after this release; close() // can now always join both workers exactly once. self.subscriber = sub; _ = hub.active.fetchAdd(1, .monotonic); _ = hub.stats.subscribe_active.getPtr(scheme).fetchAdd(1, .monotonic); sub.start_gate.set(hub.io); }
/// client → server messages: SubscriberSourcedMessage envelope. /// non-text frames are ignored; unknown types are ignored; malformed /// JSON closes 1007; a payload failing validation closes 1008. an /// invalid update while awaiting hello disconnects. pub fn clientMessage(self: *Handler, allocator: Allocator, data: []const u8, tpe: websocket.MessageTextType) !void { if (self.v2) { // server-push only: ANY client data frame closes the connection // (control frames are handled below this layer), matching the // atmos xrpcserver subscription contract return self.closeWith(1008, "network.bsky.jetstream.subscribeEvents is server-push only"); } if (tpe != .text) return;
const parsed = std.json.parseFromSlice(std.json.Value, allocator, data, .{}) catch { self.noteOptionsError(.bad_envelope_json); return self.closeWith(1007, "bad SubscriberSourcedMessage envelope"); }; defer parsed.deinit(); const root = parsed.value; if (root != .object) { self.noteOptionsError(.bad_envelope_json); return self.closeWith(1007, "bad SubscriberSourcedMessage envelope"); }
const type_val = root.object.get("type") orelse { self.noteOptionsError(.bad_envelope_json); return self.closeWith(1007, "bad SubscriberSourcedMessage envelope"); }; if (type_val != .string) { self.noteOptionsError(.bad_envelope_json); return self.closeWith(1007, "bad SubscriberSourcedMessage envelope"); } if (!std.mem.eql(u8, type_val.string, "options_update")) { // quirk: unknown types are logged and ignored, never fatal log.debug("ignoring client message type: {s}", .{type_val.string}); return; }
const payload = root.object.get("payload") orelse { self.noteOptionsError(.bad_payload_json); return self.closeWith(1007, "bad options_update payload"); }; if (payload != .object) { self.noteOptionsError(.bad_payload_json); return self.closeWith(1007, "bad options_update payload"); }
var new_filter = filter_mod.Filter.init(self.hub.allocator); var ok = true; var reason: []const u8 = "invalid options"; if (payload.object.get("wantedCollections")) |v| { ok = addAll(&new_filter, v, filter_mod.Filter.addCollection, &reason); } if (ok) if (payload.object.get("wantedDids")) |v| { ok = addAll(&new_filter, v, filter_mod.Filter.addDid, &reason); }; if (!ok) { new_filter.deinit(); self.noteOptionsError(.invalid_options); return self.closeWith(1008, reason); }
// maxMessageSizeBytes here is a JSON integer; negatives clamp to 0 var max_size: u64 = 0; if (payload.object.get("maxMessageSizeBytes")) |v| { if (v == .integer and v.integer > 0) max_size = @intCast(v.integer); }
if (self.subscriber) |sub| { sub.updateOptions(new_filter, max_size); } else { // hello: adopt the new options and start delivering if (self.pending_filter) |*f| f.deinit(); self.pending_filter = new_filter; self.pending_max_msg_size = max_size; self.awaiting_hello = false; self.startSubscriber() catch |err| { log.err("subscriber start failed: {s}", .{@errorName(err)}); return self.closeWith(1011, "internal error"); }; } _ = self.hub.stats.subscribe_options_updates_total.fetchAdd(1, .monotonic); }
fn noteOptionsError(self: *Handler, reason: metrics.OptionsErrorReason) void { _ = self.hub.stats.subscribe_options_update_errors.getPtr(reason).fetchAdd(1, .monotonic); }
/// Parser-level failures happen before clientMessage. The websocket /// dependency reports its actual size-limit error through this hook. pub fn clientError(self: *Handler, err: anyerror) void { if (err == error.TooLarge) self.noteOptionsError(.oversize); }
fn closeWith(self: *Handler, code: u16, reason: []const u8) !void { self.conn.close(.{ .code = code, .reason = reason }) catch {}; return error.Closed; }
pub fn clientClose(self: *Handler, _: []const u8) !void { if (self.subscriber) |sub| self.signalStop(sub); self.noteCleanDisconnect(); self.conn.close(.{}) catch {}; }
/// Called only by the websocket server's intentional shutdown path. pub fn serverClose(self: *Handler) void { if (self.subscriber) |sub| self.signalStop(sub); self.conn.writeClose(.{ .code = 1001, .reason = "server shutting down" }) catch {}; self.noteCleanDisconnect(); }
/// always called by the server when the connection ends — the single /// place the subscriber thread is joined and freed. pub fn close(self: *Handler) void { if (self.subscriber) |sub| { self.signalStop(sub); if (sub.thread) |t| t.join(); if (sub.ping_thread) |t| t.join(); if (sub.compressor) |*comp| comp.deinit(); sub.filter.deinit(); _ = self.hub.active.fetchSub(1, .monotonic); _ = self.hub.stats.subscribe_active.getPtr(sub.compression()).fetchSub(1, .monotonic); self.hub.allocator.destroy(sub); self.subscriber = null; } if (self.pending_filter) |*f| f.deinit(); if (!self.http_observed) { self.http_observed = true; const elapsed = self.http_started.durationTo(Io.Timestamp.now(self.hub.io, .awake)).toMicroseconds(); self.hub.stats.observeHttp(self.http_handler, .get, 200, @intCast(@max(0, elapsed))); } }
fn signalStop(self: *Handler, sub: *Subscriber) void { sub.write_mutex.lockUncancelable(self.hub.io); sub.alive.store(false, .release); sub.stopped.store(true, .release); sub.write_mutex.unlock(self.hub.io); sub.ping_stop.set(self.hub.io); self.hub.tail.wake(); }
fn noteCleanDisconnect(self: *Handler) void { if (self.clean_disconnect_counted) return; self.clean_disconnect_counted = true; _ = self.hub.stats.subscribe_clean_disconnects_total.fetchAdd(1, .monotonic); }
/// plain-HTTP requests on the websocket port: /metrics and /healthz pub fn httpFallback( conn: *websocket.Conn, method: []const u8, url: []const u8, body: []const u8, headers: *const websocket.Handshake.KeyValue, context: anytype, ) void { const hub = contextHub(context); const role = contextRole(context); const path = if (std.mem.indexOfScalar(u8, url, '?')) |i| url[0..i] else url; const query = if (std.mem.indexOfScalar(u8, url, '?')) |i| url[i + 1 ..] else ""; var observation = HttpObservation.init(hub, path, method); defer observation.finish(); const debug_path = std.mem.eql(u8, path, "/healthz") or std.mem.eql(u8, path, "/readyz") or std.mem.eql(u8, path, "/metrics"); if (role == .public and debug_path) { observation.code = 404; return respond(conn, "404 Not Found", "text/plain", "not found\n"); } if (role == .debug) { if (!debug_path) { observation.code = 404; return respond(conn, "404 Not Found", "text/plain", "not found\n"); } const head = std.mem.eql(u8, method, "HEAD"); if (!std.mem.eql(u8, method, "GET") and !head) { observation.code = 405; return respond(conn, "405 Method Not Allowed", "text/plain", "method not allowed\n"); } if (std.mem.eql(u8, path, "/healthz")) { observation.code = 200; return respondMaybeHead(conn, "200 OK", "text/plain", "ok\n", head); } if (std.mem.eql(u8, path, "/readyz")) { if (!hub.listeners_ready.load(.acquire)) { observation.code = 503; return respondMaybeHead(conn, "503 Service Unavailable", "text/plain", "not ready\n", head); } observation.code = 200; return respondMaybeHead(conn, "200 OK", "text/plain", "ok\n", head); } return respondMetrics(conn, hub, &observation, head); } // CORS preflight answers before ANY auth or method check: an OPTIONS // request never carries credentials, so the archive key gate must not // see it (the hosted edge's contract; ours 401'd preflights until a // browser consumer appeared the day after launch). if (std.mem.eql(u8, method, "OPTIONS")) { observation.code = 204; conn.writeFramed(cors_preflight) catch {}; return; } // Startup gate: the listener accepts before storage exists. Only // surfaces that touch no store serve until main flips storage_ready; // /status answers 200 with a minimal page (uptime + upstream seq in // the normal field format) so probes see "starting", not an outage. if (!hub.storage_ready.load(.acquire)) { if (std.mem.eql(u8, path, "/status")) { const now_s: i64 = @divTrunc(Io.Timestamp.now(hub.io, .real).toMicroseconds(), std.time.us_per_s); var starting_buf: [224]u8 = undefined; const starting_body = std.fmt.bufPrint( &starting_buf, "state starting (storage opening; data paths return 503)\nupstream seq {d}\nuptime {d}s\n", .{ hub.stats.upstream_seq.load(.monotonic), now_s - hub.stats.start_time_s }, ) catch "state starting\n"; observation.code = 200; return respond(conn, "200 OK", "text/plain; charset=utf-8", starting_body); } const storage_free = std.mem.eql(u8, path, "/healthz") or std.mem.eql(u8, path, "/readyz") or std.mem.eql(u8, path, "/metrics") or std.mem.eql(u8, path, "/") or std.mem.eql(u8, path, endpoints.json_path); if (!storage_free) { observation.code = 503; if (std.mem.startsWith(u8, path, "/xrpc/")) return respond(conn, "503 Service Unavailable", "application/json", "{\"error\":\"ServiceUnavailable\",\"message\":\"starting\"}"); return respond(conn, "503 Service Unavailable", "text/plain", "starting\n"); } } // the homepage stays up during bootstrap too (it says so) const import_ops = std.mem.eql(u8, path, "/xrpc/network.bsky.jetstream.importTimestamps") or std.mem.eql(u8, path, "/xrpc/network.bsky.jetstream.getImportStatus"); const dictionary_path = std.mem.eql(u8, path, "/xrpc/network.bsky.jetstream.getZstdDictionary"); const ops_path = std.mem.eql(u8, path, "/healthz") or std.mem.eql(u8, path, "/readyz") or std.mem.eql(u8, path, "/metrics") or std.mem.eql(u8, path, "/status") or std.mem.eql(u8, path, "/") or std.mem.eql(u8, path, endpoints.json_path) or import_ops or dictionary_path; if (!hub.serving.load(.acquire) and !ops_path) { observation.code = 503; if (std.mem.startsWith(u8, path, "/xrpc/")) return respond(conn, "503 Service Unavailable", "application/json", "{\"error\":\"ServiceUnavailable\",\"message\":\"bootstrap in progress\"}"); return respond(conn, "503 Service Unavailable", "text/plain", "bootstrap in progress\n"); } if (hub.xrpc) |api| { var responder: XrpcResponder = .{ .conn = conn, .observation = &observation }; observation.get_block = std.mem.eql(u8, method, "GET") and std.mem.eql(u8, path, "/xrpc/network.bsky.jetstream.getBlock"); if (api.handle( &responder, method, path, query, body, headers.get("authorization"), .{ .range = headers.get("range"), .if_match = headers.get("if-match"), .if_unmodified_since = headers.get("if-unmodified-since"), .if_none_match = headers.get("if-none-match"), .if_modified_since = headers.get("if-modified-since"), .if_range = headers.get("if-range"), }, )) return; } const status_head = std.mem.eql(u8, method, "HEAD") and std.mem.eql(u8, path, "/status"); if (!std.mem.eql(u8, method, "GET") and !status_head) { observation.code = 405; return respond(conn, "405 Method Not Allowed", "text/plain", "method not allowed\n"); } if (std.mem.eql(u8, path, "/healthz")) { observation.code = 200; return respond(conn, "200 OK", "text/plain", "ok\n"); } if (std.mem.eql(u8, path, "/readyz")) { if (!hub.listeners_ready.load(.acquire)) { observation.code = 503; return respond(conn, "503 Service Unavailable", "text/plain", "not ready\n"); } observation.code = 200; return respond(conn, "200 OK", "text/plain", "ok\n"); } if (std.mem.eql(u8, path, endpoints.json_path)) { observation.code = 200; return respond(conn, "200 OK", "application/json", endpoints.json); } if (std.mem.eql(u8, path, "/")) { observation.code = 200; const now_s: i64 = @divTrunc(Io.Timestamp.now(hub.io, .real).toMicroseconds(), std.time.us_per_s); const args: homepage.Args = .{ .seq = hub.stats.upstream_seq.load(.monotonic), .events_total = hub.stats.events_total.load(.monotonic), .subscribers = hub.active.load(.monotonic), .uptime_s = now_s - hub.stats.start_time_s, .serving = hub.serving.load(.acquire), }; var page_buf: [24 * 1024]u8 = undefined; const wants_html = if (headers.get("accept")) |a| std.mem.indexOf(u8, a, "text/html") != null else false; if (wants_html) return respond(conn, "200 OK", "text/html; charset=utf-8", homepage.renderHtml(&page_buf, args)); return respond(conn, "200 OK", "text/plain; charset=utf-8", homepage.render(&page_buf, args)); } if (std.mem.eql(u8, path, "/status")) { const wants_html = if (headers.get("accept")) |a| std.mem.indexOf(u8, a, "text/html") != null else false; const account_query = status_page.accountFromQueryAlloc(hub.allocator, query) catch { observation.code = 400; return respond(conn, "400 Bad Request", "text/plain", "invalid account query\n"); }; if (account_query) |raw_account| { defer hub.allocator.free(raw_account); const store = hub.repo_status orelse { observation.code = 500; return respond(conn, "500 Internal Server Error", "text/plain", "account diagnostics unavailable\n"); }; var arena = std.heap.ArenaAllocator.init(hub.allocator); defer arena.deinit(); const account = status_page.collectAccount( arena.allocator(), hub.io, store, hub.plc_url, raw_account, hub.status_identity_resolution, ) catch { observation.code = 500; return respond(conn, "500 Internal Server Error", "text/plain", "failed to read account diagnostics\n"); }; const now_us = Io.Timestamp.now(hub.io, .real).toMicroseconds(); var verification: ?repo_export.VerifyReport = null; var verification_error: []const u8 = ""; if (!status_head and account.found) if (hub.archive) |archive| verify: { if (hub.repo_action_rate_limits) { var peer_ip_buf: [64]u8 = undefined; const peer_ip = conn.peerIp(&peer_ip_buf) catch { verification_error = "unable to identify source IP for repo verification"; break :verify; }; const allowed = hub.repo_action_limiter.allow(hub.allocator, hub.io, peer_ip, now_us) catch { verification_error = "unable to apply repo action rate limit"; break :verify; }; if (!allowed) { verification_error = "repo action rate limit exceeded"; break :verify; } } verification = repo_export.verify( arena.allocator(), hub.io, archive, hub.data_dir, hub.plc_url, account.did, ) catch |err| { verification_error = std.fmt.allocPrint(arena.allocator(), "repoexport: verify {s}: {s}", .{ account.did, @errorName(err) }) catch "repo verification failed"; break :verify; }; }; const page = status_page.renderAccountAlloc(hub.allocator, account, now_us, wants_html, verification, verification_error) catch { observation.code = 500; return respond(conn, "500 Internal Server Error", "text/plain", "failed to render account diagnostics\n"); }; defer hub.allocator.free(page); observation.code = 200; return respondStatus(conn, if (wants_html) "text/html; charset=utf-8" else "text/plain; charset=utf-8", page, status_head, now_us); } if (status_page.segmentTabFromQuery(query)) |which| { const archive = hub.archive orelse { observation.code = 500; return respond(conn, "500 Internal Server Error", "text/plain", "archive diagnostics unavailable\n"); }; const now_us = Io.Timestamp.now(hub.io, .real).toMicroseconds(); const page = renderArchiveTab(hub, archive, which, wants_html) catch { observation.code = 500; return respond(conn, "500 Internal Server Error", "text/plain", "failed to read archive diagnostics\n"); }; defer hub.allocator.free(page); observation.code = 200; return respondStatus(conn, if (wants_html) "text/html; charset=utf-8" else "text/plain; charset=utf-8", page, status_head, now_us); } if (status_page.hostSortFromQuery(query)) |sort| { const store = hub.repo_status orelse { observation.code = 500; return respond(conn, "500 Internal Server Error", "text/plain", "host diagnostics unavailable\n"); }; const statuses = store.listHostStatuses(hub.allocator) catch { observation.code = 500; return respond(conn, "500 Internal Server Error", "text/plain", "failed to read host diagnostics\n"); }; defer store.freeHostStatuses(hub.allocator, statuses); const now_us = Io.Timestamp.now(hub.io, .real).toMicroseconds(); const page = status_page.renderHostsAlloc(hub.allocator, statuses, sort, now_us, wants_html) catch { observation.code = 500; return respond(conn, "500 Internal Server Error", "text/plain", "failed to render host diagnostics\n"); }; defer hub.allocator.free(page); observation.code = 200; return respondStatus(conn, if (wants_html) "text/html; charset=utf-8" else "text/plain; charset=utf-8", page, status_head, now_us); } const now_s: i64 = @divTrunc(Io.Timestamp.now(hub.io, .real).toMicroseconds(), std.time.us_per_s); const lifecycle_facts = readLifecycleFacts(hub, now_s); const repo_facts = readRepoFacts(hub); const retention = readRetention(hub); var import_buf: [64]u8 = undefined; const import_job = readImportJob(hub, &import_buf); const args: homepage.StatusArgs = .{ .upstream_seq = hub.stats.upstream_seq.load(.monotonic), .durable_seq = if (hub.archive) |archive| archive.committed_seq.load(.acquire) else 0, .process_events = hub.stats.events_total.load(.monotonic), .subscribers = hub.active.load(.monotonic), .uptime_s = now_s - hub.stats.start_time_s, .serving = hub.serving.load(.acquire), .bootstrap_enabled = hub.bootstrap_enabled, .compaction_enabled = hub.compaction_enabled, .retry_enabled = hub.retry_enabled, .verify_valid = hub.stats.verify_valid.load(.monotonic), .verify_invalid = hub.stats.verify_invalid.load(.monotonic), .phase = lifecycle_facts.phase, .phase_age_s = lifecycle_facts.age_s, .backfill_duration_us = lifecycle_facts.backfill_duration_us, .repos_total = repo_facts.total, .repos_complete = repo_facts.complete, .repos_pending = repo_facts.pending, .repos_failed = repo_facts.failed, .repos_unavailable = repo_facts.unavailable, .next_seq = if (hub.archive) |a| a.next_seq else 0, .lookback_s = hub.cursor_lookback_ns / std.time.ns_per_s, .oldest_retained_seq = retention.oldest_seq, .sealed_segments = retention.sealed_segments, .import_job = import_job, }; var page_buf: [32 * 1024]u8 = undefined; observation.code = 200; if (wants_html) return respondStatus(conn, "text/html; charset=utf-8", homepage.renderStatusHtml(&page_buf, args), status_head, now_s * std.time.us_per_s); return respondStatus(conn, "text/plain; charset=utf-8", homepage.renderStatus(&page_buf, args), status_head, now_s * std.time.us_per_s); } if (std.mem.eql(u8, path, "/metrics")) { return respondMetrics(conn, hub, &observation, false); } observation.code = 404; respond(conn, "404 Not Found", "text/plain", "not found\n"); }};
fn respondMetrics(conn: *websocket.Conn, hub: *Hub, observation: *HttpObservation, head: bool) void { observation.code = 200; const g = hub.tail.gauges(); const now_s: i64 = @divTrunc(Io.Timestamp.now(hub.io, .real).toMicroseconds(), std.time.us_per_s); var buf: [256 * 1024]u8 = undefined; const out = metrics.format(&buf, hub.stats, hub.active.load(.monotonic), .{ .entries = g.entries, .bytes = g.bytes, .pinned_bytes = g.pinned_bytes, .pinned_overrun_bytes = g.pinned_overrun_bytes, .base = g.base, .tip = g.tip, }, now_s, process_metrics.collect(hub.io), if (hub.archive) |archive| disk_space.freeBytes(archive.dir) else null, readDurableRepos(hub)); var key_buf: [16 * 1024]u8 = undefined; var key_writer: Io.Writer = .fixed(&key_buf); // sealed-archive totals are public by design (the site's ON DISK tile // reads them here; listSegments carries per-segment sizes but is // key-gated) if (hub.archive) |archive| { const totals = archive.manifest.sealedTotals(); key_writer.print("# TYPE stream_archive_sealed_bytes gauge\nstream_archive_sealed_bytes {d}\n# TYPE stream_archive_sealed_segments gauge\nstream_archive_sealed_segments {d}\n", .{ totals.bytes, totals.segments }) catch {}; } if (hub.key_ring) |ring| ring.renderMetrics(hub.io, &key_writer); respondPartsMaybeHead(conn, "200 OK", "text/plain; version=0.0.4", out, key_writer.buffered(), head);}
/// adapter the xrpc api writes through (json / raw+etag / xrpc error)const XrpcResponder = struct { conn: *websocket.Conn, observation: *HttpObservation, modified_s: ?i64 = null, cache_max_age_s: u64 = 0,
/// bytes actually served on this request (0 until a 200 body went out); /// the api key ring charges per-key budgets from this pub fn servedBytes(self: *const XrpcResponder) u64 { return self.observation.served_bytes; }
pub fn archiveModified(self: *XrpcResponder, modified_s: i64) void { // net/http treats the Unix epoch as an unspecified modification time. self.modified_s = if (modified_s == 0) null else modified_s; }
pub fn archiveCacheMaxAge(self: *XrpcResponder, max_age_s: u64) void { self.cache_max_age_s = max_age_s; }
fn cacheControl(self: *const XrpcResponder, buf: *[64]u8) []const u8 { if (self.cache_max_age_s == 0) return "public, no-cache"; return std.fmt.bufPrint(buf, "public, max-age={d}", .{self.cache_max_age_s}) catch "public, no-cache"; }
fn lastModifiedHeader(self: *const XrpcResponder, buf: *[64]u8) []const u8 { const modified_s = self.modified_s orelse return ""; var date_buf: [40]u8 = undefined; return std.fmt.bufPrint(buf, "Last-Modified: {s}\r\n", .{http_time.format(&date_buf, modified_s)}) catch ""; }
pub fn json(self: *XrpcResponder, status: u16, body: []const u8) void { self.observation.code = status; respond(self.conn, statusLine(status), "application/json", body); }
pub fn raw(self: *XrpcResponder, status: u16, body: []const u8, etag_hex: []const u8) void { var modified_buf: [64]u8 = undefined; const modified = self.lastModifiedHeader(&modified_buf); var cache_buf: [64]u8 = undefined; const cache = self.cacheControl(&cache_buf); var buf: [640]u8 = undefined; const header = std.fmt.bufPrint( &buf, "HTTP/1.1 {s}\r\n" ++ cors_headers ++ "Content-Type: application/octet-stream\r\nContent-Length: {d}\r\nAccept-Ranges: bytes\r\nETag: \"{s}\"\r\nCache-Control: {s}\r\n{s}Connection: close\r\nServer: stream\r\n\r\n", .{ statusLine(status), body.len, etag_hex, cache, modified }, ) catch return; self.observation.code = status; self.conn.writeFramed(header) catch return; if (body.len > 0) self.conn.writeFramed(body) catch return; if (status == 200) self.observation.served_bytes = body.len; }
pub fn dictionary(self: *XrpcResponder, body: []const u8, id: u32, etag: []const u8, not_modified: bool) void { var buf: [896]u8 = undefined; const header = if (not_modified) std.fmt.bufPrint( &buf, "HTTP/1.1 304 Not Modified\r\n" ++ cors_headers ++ "ETag: {s}\r\nCache-Control: public, max-age=31536000, immutable\r\nX-Zstd-Dictionary-Id: {d}\r\nContent-Length: 0\r\nConnection: close\r\nServer: stream\r\n\r\n", .{ etag, id }, ) catch return else std.fmt.bufPrint( &buf, "HTTP/1.1 200 OK\r\n" ++ cors_headers ++ "Content-Type: application/octet-stream\r\nContent-Length: {d}\r\nETag: {s}\r\nCache-Control: public, max-age=31536000, immutable\r\nX-Zstd-Dictionary-Id: {d}\r\nConnection: close\r\nServer: stream\r\n\r\n", .{ body.len, etag, id }, ) catch return; self.observation.code = if (not_modified) 304 else 200; self.conn.writeFramed(header) catch return; if (!not_modified) self.conn.writeFramed(body) catch return; }
pub fn partial(self: *XrpcResponder, body: []const u8, etag_hex: []const u8, start: u64, end: u64, total: u64) void { var modified_buf: [64]u8 = undefined; const modified = self.lastModifiedHeader(&modified_buf); var cache_buf: [64]u8 = undefined; const cache = self.cacheControl(&cache_buf); var buf: [720]u8 = undefined; const header = std.fmt.bufPrint( &buf, "HTTP/1.1 206 Partial Content\r\n" ++ cors_headers ++ "Content-Type: application/octet-stream\r\nContent-Length: {d}\r\nContent-Range: bytes {d}-{d}/{d}\r\nAccept-Ranges: bytes\r\nETag: \"{s}\"\r\nCache-Control: {s}\r\n{s}Connection: close\r\nServer: stream\r\n\r\n", .{ body.len, start, end, total, etag_hex, cache, modified }, ) catch return; self.observation.code = 206; self.conn.writeFramed(header) catch return; if (body.len > 0) self.conn.writeFramed(body) catch return; }
pub fn multipartBytes(self: *XrpcResponder, body: []const u8, etag_hex: []const u8, boundary: *const [60]u8, ranges: anytype) void { const encoded_len = multipartEncodedLength(boundary, ranges, body.len) orelse return self.err(500, "InternalError", "multipart response too large"); self.multipartHeader(etag_hex, boundary, encoded_len); if (self.observation.code != 206) return; for (ranges, 0..) |range, i| { var part_buf: [320]u8 = undefined; const part = multipartPartHeader(&part_buf, boundary, range.start, range.length, body.len, i == 0) orelse return; self.conn.writeFramed(part) catch return; const start: usize = @intCast(range.start); const end: usize = @intCast(range.start + range.length); if (end > start) self.conn.writeFramed(body[start..end]) catch return; } self.multipartClose(boundary); }
pub fn multipartFile(self: *XrpcResponder, io: Io, source: Io.File, etag_hex: []const u8, boundary: *const [60]u8, ranges: anytype, total: u64) void { const encoded_len = multipartEncodedLength(boundary, ranges, total) orelse return self.err(500, "InternalError", "multipart response too large"); self.multipartHeader(etag_hex, boundary, encoded_len); if (self.observation.code != 206) return; var copy_buf: [64 * 1024]u8 = undefined; for (ranges, 0..) |range, i| { var part_buf: [320]u8 = undefined; const part = multipartPartHeader(&part_buf, boundary, range.start, range.length, total, i == 0) orelse return; self.conn.writeFramed(part) catch return; var offset = range.start; var remaining = range.length; while (remaining > 0) { const want: usize = @intCast(@min(remaining, copy_buf.len)); const n = source.readPositional(io, &.{copy_buf[0..want]}, offset) catch return; if (n == 0) return; self.conn.writeFramed(copy_buf[0..n]) catch return; offset += n; remaining -= n; } } self.multipartClose(boundary); }
fn multipartHeader(self: *XrpcResponder, etag_hex: []const u8, boundary: *const [60]u8, body_len: u64) void { var modified_buf: [64]u8 = undefined; const modified = self.lastModifiedHeader(&modified_buf); var cache_buf: [64]u8 = undefined; const cache = self.cacheControl(&cache_buf); var buf: [832]u8 = undefined; const header = std.fmt.bufPrint( &buf, "HTTP/1.1 206 Partial Content\r\n" ++ cors_headers ++ "Content-Type: multipart/byteranges; boundary={s}\r\nContent-Length: {d}\r\nAccept-Ranges: bytes\r\nETag: \"{s}\"\r\nCache-Control: {s}\r\n{s}Connection: close\r\nServer: stream\r\n\r\n", .{ boundary, body_len, etag_hex, cache, modified }, ) catch return; self.observation.code = 206; self.conn.writeFramed(header) catch { self.observation.code = 500; return; }; }
fn multipartClose(self: *XrpcResponder, boundary: *const [60]u8) void { var buf: [72]u8 = undefined; const close = std.fmt.bufPrint(&buf, "\r\n--{s}--\r\n", .{boundary}) catch return; self.conn.writeFramed(close) catch return; }
/// Stream an immutable sealed file without retaining it per request. The /// upstream Go server reaches the same property through http.ServeContent /// and sendfile; chunking keeps Zig's memory bounded for slow clients. pub fn file(self: *XrpcResponder, io: Io, source: Io.File, etag_hex: []const u8, start: u64, end: u64, total: u64, range_requested: bool) void { // A satisfiable Range request is 206 even when it happens to span the // whole representation (e.g. bytes=0- or an oversized suffix). const partial_response = range_requested; const len = end - start + 1; var modified_buf: [64]u8 = undefined; const modified = self.lastModifiedHeader(&modified_buf); var cache_buf: [64]u8 = undefined; const cache = self.cacheControl(&cache_buf); var header_buf: [832]u8 = undefined; const header = if (partial_response) std.fmt.bufPrint( &header_buf, "HTTP/1.1 206 Partial Content\r\n" ++ cors_headers ++ "Content-Type: application/octet-stream\r\nContent-Length: {d}\r\nContent-Range: bytes {d}-{d}/{d}\r\nAccept-Ranges: bytes\r\nETag: \"{s}\"\r\nCache-Control: {s}\r\n{s}Connection: close\r\nServer: stream\r\n\r\n", .{ len, start, end, total, etag_hex, cache, modified }, ) catch return else std.fmt.bufPrint( &header_buf, "HTTP/1.1 200 OK\r\n" ++ cors_headers ++ "Content-Type: application/octet-stream\r\nContent-Length: {d}\r\nAccept-Ranges: bytes\r\nETag: \"{s}\"\r\nCache-Control: {s}\r\n{s}Connection: close\r\nServer: stream\r\n\r\n", .{ len, etag_hex, cache, modified }, ) catch return; self.observation.code = if (partial_response) 206 else 200; self.conn.writeFramed(header) catch return;
var offset = start; var remaining = len; var buf: [64 * 1024]u8 = undefined; while (remaining > 0) { const want: usize = @intCast(@min(remaining, buf.len)); const n = source.readPositional(io, &.{buf[0..want]}, offset) catch return; if (n == 0) return; self.conn.writeFramed(buf[0..n]) catch return; offset += n; remaining -= n; } if (!partial_response) self.observation.served_bytes = len; }
pub fn rangeNotSatisfiable(self: *XrpcResponder, total: u64, no_overlap: bool) void { const body = if (no_overlap) "invalid range: failed to overlap\n" else "invalid range\n"; var buf: [896]u8 = undefined; const header = if (no_overlap) std.fmt.bufPrint( &buf, "HTTP/1.1 416 Range Not Satisfiable\r\n" ++ cors_headers ++ "Content-Range: bytes */{d}\r\nContent-Type: text/plain; charset=utf-8\r\nX-Content-Type-Options: nosniff\r\nContent-Length: {d}\r\nConnection: close\r\nServer: stream\r\n\r\n", .{ total, body.len }, ) catch return else std.fmt.bufPrint( &buf, "HTTP/1.1 416 Range Not Satisfiable\r\n" ++ cors_headers ++ "Content-Type: text/plain; charset=utf-8\r\nX-Content-Type-Options: nosniff\r\nContent-Length: {d}\r\nConnection: close\r\nServer: stream\r\n\r\n", .{body.len}, ) catch return; self.observation.code = 416; self.conn.writeFramed(header) catch return; self.conn.writeFramed(body) catch return; }
pub fn preconditionFailed(self: *XrpcResponder, etag_hex: []const u8) void { var modified_buf: [64]u8 = undefined; const modified = self.lastModifiedHeader(&modified_buf); var cache_buf: [64]u8 = undefined; const cache = self.cacheControl(&cache_buf); var buf: [640]u8 = undefined; const header = std.fmt.bufPrint( &buf, "HTTP/1.1 412 Precondition Failed\r\n" ++ cors_headers ++ "Content-Type: application/octet-stream\r\nETag: \"{s}\"\r\nCache-Control: {s}\r\n{s}Content-Length: 0\r\nConnection: close\r\nServer: stream\r\n\r\n", .{ etag_hex, cache, modified }, ) catch return; self.observation.code = 412; self.conn.writeFramed(header) catch return; }
pub fn notModified(self: *XrpcResponder, etag_hex: []const u8) void { var cache_buf: [64]u8 = undefined; const cache = self.cacheControl(&cache_buf); var buf: [512]u8 = undefined; const header = std.fmt.bufPrint( &buf, "HTTP/1.1 304 Not Modified\r\n" ++ cors_headers ++ "ETag: \"{s}\"\r\nCache-Control: {s}\r\nConnection: close\r\nServer: stream\r\n\r\n", .{ etag_hex, cache }, ) catch return; self.observation.code = 304; self.conn.writeFramed(header) catch return; }
/// every status err() can be called with must have an arm here: the /// else-arm rewrites unknown statuses to 400, which silently breaks /// retry semantics (429 fell through to 400 on the key ring's first /// day in production — metered clients treated the throttle as fatal) fn errStatusLine(status: u16) []const u8 { return switch (status) { 401 => "401 Unauthorized", 404 => "404 Not Found", 405 => "405 Method Not Allowed", 409 => "409 Conflict", 429 => "429 Too Many Requests", 500 => "500 Internal Server Error", 501 => "501 Not Implemented", 503 => "503 Service Unavailable", else => "400 Bad Request", }; }
pub fn err(self: *XrpcResponder, status: u16, name: []const u8, message: []const u8) void { var buf: [512]u8 = undefined; const wire_name = if (status == 500) "InternalServerError" else name; const body = std.fmt.bufPrint(&buf, "{{\"error\":\"{s}\",\"message\":\"{s}\"}}", .{ wire_name, message }) catch return; const status_line = errStatusLine(status); self.observation.code = status; respond(self.conn, status_line, "application/json", body); }
/// The hosted Bluesky instances' archive-auth contract, byte for byte /// (probed 2026-08-13): 401, `WWW-Authenticate: Bearer`, and a bare /// {"error":"invalid bearer credential"} — NOT the XRPC error envelope. pub fn unauthorized(self: *XrpcResponder) void { self.observation.code = 401; const body = "{\"error\":\"invalid bearer credential\"}"; var buf: [512]u8 = undefined; const header = std.fmt.bufPrint( &buf, "HTTP/1.1 401 Unauthorized\r\n" ++ cors_headers ++ "Content-Type: application/json\r\nWWW-Authenticate: Bearer\r\nContent-Length: {d}\r\nConnection: close\r\nServer: stream\r\n\r\n", .{body.len}, ) catch return; self.conn.writeFramed(header) catch return; _ = self.conn.writeFramed(body) catch {}; }};
fn multipartPartHeader(buf: *[320]u8, boundary: *const [60]u8, start: u64, length: u64, total: u64, first: bool) ?[]const u8 { const end = if (length == 0) start - 1 else start + length - 1; return std.fmt.bufPrint( buf, "{s}--{s}\r\nContent-Range: bytes {d}-{d}/{d}\r\nContent-Type: application/octet-stream\r\n\r\n", .{ if (first) "" else "\r\n", boundary, start, end, total }, ) catch null;}
fn multipartEncodedLength(boundary: *const [60]u8, ranges: anytype, total: u64) ?u64 { var encoded: u64 = 0; for (ranges, 0..) |range, i| { var buf: [320]u8 = undefined; const part = multipartPartHeader(&buf, boundary, range.start, range.length, total, i == 0) orelse return null; encoded = std.math.add(u64, encoded, part.len) catch return null; encoded = std.math.add(u64, encoded, range.length) catch return null; } var close_buf: [72]u8 = undefined; const close = std.fmt.bufPrint(&close_buf, "\r\n--{s}--\r\n", .{boundary}) catch return null; return std.math.add(u64, encoded, close.len) catch null;}
fn statusLine(status: u16) []const u8 { return switch (status) { 200 => "200 OK", 201 => "201 Created", 400 => "400 Bad Request", 401 => "401 Unauthorized", 404 => "404 Not Found", 405 => "405 Method Not Allowed", 409 => "409 Conflict", 500 => "500 Internal Server Error", 503 => "503 Service Unavailable", else => "500 Internal Server Error", };}
/// Render the segments or collections view from resident manifest metadata./// Both walk sealed segments only; neither opens a segment file, so the cost/// is bounded by segment count rather than by archive size.fn renderArchiveTab( hub: *Hub, archive: *archive_mod.Archive, which: []const u8, html: bool,) ![]u8 { var arena = std.heap.ArenaAllocator.init(hub.allocator); defer arena.deinit(); const alloc = arena.allocator();
if (std.mem.eql(u8, which, "segments")) { const summaries = try archive.manifest.all(alloc); const rows = try alloc.alloc(status_page.SegmentRow, summaries.len); for (summaries, rows) |meta, *row| { row.* = .{ .index = meta.idx, .size_bytes = meta.size, .event_count = meta.header.event_count, .unique_did_count = meta.header.unique_did_count, .block_count = meta.header.block_count, .min_seq = meta.header.min_seq, .max_seq = meta.header.max_seq, .min_witnessed_at = meta.header.min_witnessed_at, .max_witnessed_at = meta.header.max_witnessed_at, }; } return status_page.renderSegmentsAlloc(hub.allocator, rows, html); }
// Collections: fold each sealed segment's resident collection index into // one archive-wide table. `$`-prefixed entries are DID-level marker // sentinels, not real collections, so they stay out of the operator view. const resident = try archive.manifest.planningSegments(alloc); defer for (resident) |*item| item.deinit(); var totals: std.StringHashMapUnmanaged(status_page.CollectionRow) = .empty; for (resident) |*sealed| { for (sealed.collection_index.entries) |entry| { if (entry.nsid.len == 0 or entry.nsid[0] == '$') continue; const gop = try totals.getOrPut(alloc, entry.nsid); if (gop.found_existing) { gop.value_ptr.event_count += entry.count; gop.value_ptr.segment_count += 1; } else { gop.key_ptr.* = try alloc.dupe(u8, entry.nsid); gop.value_ptr.* = .{ .nsid = gop.key_ptr.*, .event_count = entry.count, .segment_count = 1, }; } } } const rows = try alloc.alloc(status_page.CollectionRow, totals.count()); var it = totals.valueIterator(); var at: usize = 0; while (it.next()) |value| : (at += 1) rows[at] = value.*; status_page.sortCollections(rows); return status_page.renderCollectionsAlloc(hub.allocator, rows, html);}
/// Durable lifecycle facts for the status summary. A read failure is reported/// as absence here on purpose: this is a presentation surface, and refusing to/// render the whole page because one optional field could not be read would be/// worse than omitting the field. Every path that acts on these values reads/// them through the phase store, which fails closed.const LifecycleFacts = struct { phase: ?[]const u8 = null, age_s: ?i64 = null, backfill_duration_us: ?i64 = null,};
fn readLifecycleFacts(hub: *Hub, now_s: i64) LifecycleFacts { const store = hub.repo_status orelse return .{}; var phase_store = phase_mod.Store.init(store.meta); const phase = (phase_store.read(hub.io) catch return .{}) orelse return .{}; var facts: LifecycleFacts = .{ .phase = @tagName(phase) }; if (phase_store.readEnteredAt(hub.io) catch null) |entered_us| facts.age_s = now_s - @divTrunc(entered_us, std.time.us_per_s); const timing = phase_store.readBackfillTiming(hub.io) catch phase_mod.BackfillTiming{}; facts.backfill_duration_us = timing.durationUs(); return facts;}
/// Repository rollup for the status summary — upstream's BackfillStats. Null/// total means no repository state exists at all, which the page must not/// render as "zero of zero complete".const RepoFacts = struct { total: ?u32 = null, complete: u32 = 0, pending: u32 = 0, failed: u32 = 0, unavailable: u32 = 0,};
fn readRepoFacts(hub: *Hub) RepoFacts { const store = hub.repo_status orelse return .{}; const total = store.count(); if (total == 0) return .{}; return .{ .total = total, .complete = store.countByStatus(.complete), .pending = store.countByStatus(.pending), .failed = store.countByStatus(.failed), .unavailable = store.countByStatus(.unavailable), };}
/// Retention for the status summary — upstream's CursorLookbackStats.const Retention = struct { oldest_seq: ?u64 = null, sealed_segments: ?usize = null,};
fn readRetention(hub: *Hub) Retention { const archive = hub.archive orelse return .{}; const summaries = archive.manifest.all(hub.allocator) catch return .{}; defer hub.allocator.free(summaries); if (hub.cursor_lookback_ns == 0) return .{ .sealed_segments = summaries.len }; const now_us = Io.Timestamp.now(hub.io, .real).toMicroseconds(); return .{ .oldest_seq = archive.manifest.lookbackFloor(now_us, hub.cursor_lookback_ns).seq, .sealed_segments = summaries.len, };}
/// Current or most recent timestamp-import job — upstream's ImportInfo. Null/// when import has never run against this data dir.fn readImportJob(hub: *Hub, buf: []u8) ?[]const u8 { const api = hub.xrpc orelse return null; const manager = api.import orelse return null; var parsed = (manager.current() catch return null) orelse return null; defer parsed.deinit(); const record = parsed.value; return std.fmt.bufPrint(buf, "{s} ({s}, {s})", .{ record.id, @tagName(record.state), @tagName(record.phase), }) catch null;}
/// Durable repository counts for the progress gauges, read at scrape time.////// Null only when there is no repository store at all — that is genuinely/// unknown, and a panel should show a gap rather than invent a number. An/// empty store is different: zero complete of zero total is an accurate/// statement, and in a time series it reads as "nothing yet" rather than as a/// claim. The status page hides the section instead, because the same fact/// renders badly as prose; the distinction is medium, not truth.fn readDurableRepos(hub: *Hub) ?metrics.DurableRepos { const store = hub.repo_status orelse return null; return .{ .complete = store.countByStatus(.complete), .pending = store.countByStatus(.pending), .failed = store.countByStatus(.failed), .unavailable = store.countByStatus(.unavailable), .not_started = store.countByStatus(.not_started), };}
// The hosted Bluesky edge's CORS contract, probed 2026-08-14: allow-origin *// and these exposed headers on EVERY response (401s included) so browser// consumers work — a local-first sync engine hit this the day after launch.// Their Headwind-Quota-* names are omitted: we never send those headers.const cors_headers = "Access-Control-Allow-Origin: *\r\nAccess-Control-Expose-Headers: ETag, Content-Range, Accept-Ranges, Retry-After, WWW-Authenticate, X-Zstd-Dictionary-Id\r\n";const cors_preflight = "HTTP/1.1 204 No Content\r\nAccess-Control-Allow-Origin: *\r\nAccess-Control-Allow-Methods: GET, HEAD, POST, OPTIONS\r\nAccess-Control-Allow-Headers: Authorization, Content-Type, Range, If-Range\r\nAccess-Control-Max-Age: 600\r\nContent-Length: 0\r\nConnection: close\r\nServer: stream\r\n\r\n";
fn respond(conn: *websocket.Conn, status: []const u8, content_type: []const u8, resp_body: []const u8) void { respondMaybeHead(conn, status, content_type, resp_body, false);}
fn respondMaybeHead(conn: *websocket.Conn, status: []const u8, content_type: []const u8, resp_body: []const u8, head: bool) void { var buf: [512]u8 = undefined; const header = std.fmt.bufPrint( &buf, "HTTP/1.1 {s}\r\nContent-Type: {s}\r\n" ++ cors_headers ++ "Content-Length: {d}\r\nConnection: close\r\nServer: stream\r\n\r\n", .{ status, content_type, resp_body.len }, ) catch return; conn.writeFramed(header) catch return; if (!head and resp_body.len > 0) conn.writeFramed(resp_body) catch return;}
fn respondParts(conn: *websocket.Conn, status: []const u8, content_type: []const u8, first: []const u8, second: []const u8) void { respondPartsMaybeHead(conn, status, content_type, first, second, false);}
fn respondPartsMaybeHead(conn: *websocket.Conn, status: []const u8, content_type: []const u8, first: []const u8, second: []const u8, head: bool) void { var buf: [512]u8 = undefined; const header = std.fmt.bufPrint( &buf, "HTTP/1.1 {s}\r\n" ++ cors_headers ++ "Content-Type: {s}\r\nContent-Length: {d}\r\nConnection: close\r\nServer: stream\r\n\r\n", .{ status, content_type, first.len + second.len }, ) catch return; conn.writeFramed(header) catch return; if (!head and first.len > 0) conn.writeFramed(first) catch return; if (!head and second.len > 0) conn.writeFramed(second) catch return;}
fn respondStatus(conn: *websocket.Conn, content_type: []const u8, body: []const u8, head: bool, generated_us: i64) void { var time_buf: [40]u8 = undefined; const generated_at = formatRfc3339(&time_buf, generated_us); var header_buf: [768]u8 = undefined; const header = std.fmt.bufPrint( &header_buf, "HTTP/1.1 200 OK\r\n" ++ cors_headers ++ "Content-Type: {s}\r\nContent-Length: {d}\r\nCache-Control: no-store\r\nX-Status-Generated-At: {s}\r\nConnection: close\r\nServer: stream\r\n\r\n", .{ content_type, body.len, generated_at }, ) catch return; conn.writeFramed(header) catch return; if (!head) conn.writeFramed(body) catch {};}
fn formatRfc3339(buf: []u8, micros: i64) []const u8 { return zat.Datetime.fromMicros(micros).toString(buf);}
fn addAll( filter: *filter_mod.Filter, v: std.json.Value, comptime add: fn (*filter_mod.Filter, []const u8) filter_mod.ParseError!void, reason: *[]const u8,) bool { if (v != .array) { reason.* = "invalid options"; return false; } for (v.array.items) |item| { if (item != .string) { reason.* = "invalid options"; return false; } add(filter, item.string) catch |err| { reason.* = switch (err) { error.InvalidCollection => "invalid options: invalid collection", error.InvalidDid => "invalid options: invalid DID", error.InvalidEncoding => "invalid options: invalid encoding", error.TooManyCollections => "too many wanted collections", error.TooManyDids => "too many wanted DIDs", // v2 query-only errors; unreachable through options_update error.InvalidKind, error.CollectionsWithoutCommits => "invalid options", error.OutOfMemory => "internal error", }; return false; }; } return true;}
/// quirk: empty/malformed/negative/overflow → 0 ("no cap"), never an errorfn parseMaxMsgSize(raw: ?[]const u8) u64 { const s = raw orelse return 0; if (s.len == 0) return 0; const v = std.fmt.parseInt(i64, s, 10) catch return 0; if (v < 0) return 0; return @intCast(v);}
fn offersPerMessageDeflate(extensions: []const u8) bool { var it = std.mem.splitScalar(u8, extensions, ','); while (it.next()) |extension| { var params = std.mem.splitScalar(u8, extension, ';'); const name = std.mem.trim(u8, params.first(), " \t"); if (std.ascii.eqlIgnoreCase(name, "permessage-deflate")) return true; } return false;}
fn queryParam(query: []const u8, name: []const u8) ?[]const u8 { var it = std.mem.splitScalar(u8, query, '&'); while (it.next()) |pair| { const eq = std.mem.indexOfScalar(u8, pair, '=') orelse continue; if (std.mem.eql(u8, pair[0..eq], name)) return pair[eq + 1 ..]; } return null;}
fn resolveCursor(hub: *Hub, v2: bool, raw_param: ?[]const u8) !CursorPlan { const raw = raw_param orelse return .{}; if (raw.len == 0) return .{}; if (hub.cursor_lookback_ns == 0) return .{ .mode = .disabled }; const parsed = std.fmt.parseInt(i64, raw, 10) catch return error.InvalidCursor; if (parsed < 0) return error.InvalidCursor; const now_us = Io.Timestamp.now(hub.io, .real).toMicroseconds();
// Both endpoints use the magnitude split. V2 differs only by rejecting // a seq cursor below the retained floor; v1 preserves its clamp. if (parsed < 1_000_000_000_000_000) { var seq: u64 = @intCast(parsed); var next_seq = hub.tail.publishedNextSeq(); if (hub.archive) |archive| { const durable = archive.committed_seq.load(.acquire); if (durable != 0) next_seq = @max(next_seq, durable +| 1); } if (next_seq == 0 or seq >= next_seq) return .{ .mode = .clamped }; var clamped = false; if (seq == 0) { seq = 1; clamped = true; } if (hub.archive) |archive| { const floor = archive.manifest.lookbackFloor(now_us, hub.cursor_lookback_ns).seq; if (floor != 0 and seq < floor) { if (v2) return error.CursorTooOld; seq = floor; clamped = true; } } const gauges = hub.tail.gauges(); if (gauges.entries != 0) if (hub.tail.indexForSeq(seq)) |idx| return .{ .start_idx = idx, .start_seq = seq, .mode = if (clamped) .clamped else .seq }; if (hub.archive != null) return .{ .start_idx = gauges.base, .start_seq = seq, .mode = if (clamped) .clamped else .seq, }; if (v2) return error.CursorTooOld; return .{ .start_idx = gauges.base, .mode = .clamped }; }
if (parsed > now_us) return .{ .mode = .clamped }; // Hot-first: a time cursor within the retained hot window starts on the // hot tail directly. Resolving it through the archive instead lands at // the last sealed-segment boundary — up to a whole open segment behind — // and strands the subscriber on the cold reader, which delivers in // read-batch bursts at seal cadence (the "jerky resume" defect). if (hub.tail.oldestTime()) |oldest_hot_time| { if (parsed >= oldest_hot_time) return .{ .start_idx = hub.tail.indexForCursor(parsed), .mode = .time_us, }; } if (hub.archive) |archive| { const resolved = cold.resolveTimeToSeq(hub.allocator, hub.io, archive, parsed) catch |err| { log.err("timestamp cursor resolution failed for {d}: {s}", .{ parsed, @errorName(err) }); return error.CursorResolveFailed; }; var resolved_seq = resolved.seq; var clamped = resolved.clamped; const floor = archive.manifest.lookbackFloor(now_us, hub.cursor_lookback_ns).seq; if (floor != 0 and resolved_seq < floor) { resolved_seq = floor; clamped = true; } if (hub.tail.indexForSeq(resolved_seq)) |idx| return .{ .start_idx = idx, .start_seq = resolved_seq, .mode = if (clamped) .clamped else .time_us, .outdated_info = v2 and clamped, }; return .{ .start_idx = hub.tail.gauges().base, .start_seq = resolved_seq, .mode = if (clamped) .clamped else .time_us, .outdated_info = v2 and clamped, }; } var mode: metrics.CursorMode = .time_us; if (hub.tail.oldestTime()) |floor| { if (parsed < floor) mode = .clamped; } return .{ .start_idx = hub.tail.indexForCursor(parsed), .mode = mode, .outdated_info = v2 and mode == .clamped };}
// === tests ===
test "queryParam" { try std.testing.expectEqualStrings("42", queryParam("a=1&cursor=42&b=2", "cursor").?); try std.testing.expect(queryParam("a=1", "cursor") == null);}
test "parseMaxMsgSize quirks" { try std.testing.expectEqual(@as(u64, 0), parseMaxMsgSize(null)); try std.testing.expectEqual(@as(u64, 0), parseMaxMsgSize("")); try std.testing.expectEqual(@as(u64, 0), parseMaxMsgSize("garbage")); try std.testing.expectEqual(@as(u64, 0), parseMaxMsgSize("-5")); try std.testing.expectEqual(@as(u64, 0), parseMaxMsgSize("99999999999999999999")); try std.testing.expectEqual(@as(u64, 1024), parseMaxMsgSize("1024"));}
test "permessage-deflate offer detection" { try std.testing.expect(offersPerMessageDeflate("permessage-deflate")); try std.testing.expect(offersPerMessageDeflate("foo, permessage-deflate; client_max_window_bits")); try std.testing.expect(!offersPerMessageDeflate("x-webkit-deflate-frame"));}
test "slow detector matches upstream sustained-rate semantics" { var stalled: SlowDetector = .{}; var progressing: SlowDetector = .{}; var caught_up: SlowDetector = .{}; var stalled_pos: u64 = 0; var progressing_pos: u64 = 0; var dropped = false; for (1..121) |second| { const now_ns: i96 = @intCast(second * std.time.ns_per_s); stalled_pos += 1; progressing_pos += 100; if (stalled.observe(now_ns, stalled_pos, 1_000_000)) dropped = true; try std.testing.expect(!progressing.observe(now_ns, progressing_pos, 1_000_000)); try std.testing.expect(!caught_up.observe(now_ns, 0, 0)); } try std.testing.expect(dropped);}
test "slow detector recovery resets its full window" { var detector: SlowDetector = .{}; var pos: u64 = 0; for (1..31) |second| { pos += 1; try std.testing.expect(!detector.observe(@intCast(second * std.time.ns_per_s), pos, 1_000_000)); } pos += 1_000_000; try std.testing.expect(!detector.observe(31 * std.time.ns_per_s, pos, 100)); for (32..62) |second| { pos += 1; try std.testing.expect(!detector.observe(@intCast(second * std.time.ns_per_s), pos, 1_000_000)); }}
test "slow detector obeys configured fractional rate and window" { var below: SlowDetector = .{ .window_ns = 2 * std.time.ns_per_s, .min_rate = 2.5 }; try std.testing.expect(!below.observe(0, 100, 1_000_000)); try std.testing.expect(!below.observe(std.time.ns_per_s, 102, 1_000_000)); // Four events over two seconds is 2/s: below the configured floor. try std.testing.expect(below.observe(2 * std.time.ns_per_s, 104, 1_000_000));
var exact: SlowDetector = .{ .window_ns = 2 * std.time.ns_per_s, .min_rate = 2.5 }; try std.testing.expect(!exact.observe(0, 100, 1_000_000)); // Five events over two seconds is exactly the floor and re-anchors. try std.testing.expect(!exact.observe(2 * std.time.ns_per_s, 105, 1_000_000)); try std.testing.expectEqual(@as(i96, 2 * std.time.ns_per_s), exact.streak_start_ns); try std.testing.expectEqual(@as(u64, 105), exact.streak_start_pos);}
test "cursor resolution is pre-upgrade and preserves v1-v2 floor semantics" { var threaded: Io.Threaded = .init(std.testing.allocator, .{}); defer threaded.deinit(); var tail = tail_mod.Tail.init(std.testing.allocator, threaded.io(), 1 << 20); defer tail.deinit(); try tail.append(.commit, "did:plc:test", "app.bsky.feed.post", 100, 100, "{}", "{}", .none); var stats: metrics.Stats = .{}; var hub: Hub = .{ .allocator = std.testing.allocator, .io = threaded.io(), .tail = &tail, .stats = &stats, };
try std.testing.expectEqual(metrics.CursorMode.live, (try resolveCursor(&hub, false, null)).mode); try std.testing.expectEqual(metrics.CursorMode.clamped, (try resolveCursor(&hub, false, "50")).mode); try std.testing.expectError(error.CursorTooOld, resolveCursor(&hub, true, "50")); try std.testing.expectEqual(metrics.CursorMode.clamped, (try resolveCursor(&hub, true, "101")).mode); try std.testing.expectError(error.InvalidCursor, resolveCursor(&hub, true, "nope")); try std.testing.expectError(error.InvalidCursor, resolveCursor(&hub, true, "-1"));}
test "near-now time cursor within hot retention starts hot, never cold" { const testing = std.testing; var threaded: Io.Threaded = .init(testing.allocator, .{}); defer threaded.deinit(); const io = threaded.io(); var tail = tail_mod.Tail.init(testing.allocator, io, 1 << 20); defer tail.deinit(); try tail.append(.commit, "did:plc:test", "app.bsky.feed.post", 1_750_000_000_000_000, 100, "{}", "{}", .none); try tail.append(.commit, "did:plc:test", "app.bsky.feed.post", 1_750_000_000_500_000, 101, "{}", "{}", .none);
// The archive is present but knows nothing recent: resolving a hot-window // time cursor through it strands the subscriber on the cold reader (the // jerky-resume defect). Hot rows must win. var tmp = testing.tmpDir(.{}); defer tmp.cleanup(); var data_dir_buf: [Io.Dir.max_path_bytes]u8 = undefined; const data_dir_len = try tmp.dir.realPath(io, &data_dir_buf); var archive = try archive_mod.Archive.init(testing.allocator, io, data_dir_buf[0..data_dir_len]); defer archive.deinit();
var stats: metrics.Stats = .{}; var hub: Hub = .{ .allocator = testing.allocator, .io = io, .tail = &tail, .stats = &stats, .archive = &archive, };
const plan = try resolveCursor(&hub, false, "1750000000200000"); try testing.expectEqual(metrics.CursorMode.time_us, plan.mode); // hot plans are idx-anchored (seq-unknown): no start_seq, so the read // loop never routes them cold try testing.expectEqual(@as(?u64, null), plan.start_seq); // resumes at the first retained frame witnessed at/after the cursor try testing.expectEqual(tail.indexForSeq(101), plan.start_idx);
// a cursor predating hot retention still consults the archive: start_seq // set, and the read loop dispatches it cold until it is resident const old_plan = try resolveCursor(&hub, false, "1749999999999999"); try testing.expect(old_plan.start_seq != null);}
test "XRPC readiness gate returns the upstream JSON service error" { const testing = std.testing; var threaded: Io.Threaded = .init(testing.allocator, .{}); defer threaded.deinit(); const io = threaded.io(); var tail = tail_mod.Tail.init(testing.allocator, io, 1 << 20); defer tail.deinit(); var stats: metrics.Stats = .{}; var hub: Hub = .{ .allocator = testing.allocator, .io = io, .tail = &tail, .stats = &stats, .serving = .init(false), }; defer hub.deinit();
var listener = (Io.net.IpAddress{ .ip4 = .loopback(0) }).listen(io, .{ .reuse_address = true }) catch unreachable; defer listener.deinit(io); const port = listener.socket.address.getPort(); var ws_server = try websocket.Server(Handler).init(testing.allocator, io, .{ .port = port, .address = "127.0.0.1", .max_conn = 4, .handshake = .{ .max_size = 16 * 1024, .max_headers = 64 }, }); defer ws_server.deinit(); const Fixture = struct { fn run(server: *websocket.Server(Handler), server_listener: *Io.net.Server, fixture_hub: *Hub) !void { server.runIo(server_listener, fixture_hub); } }; var future = try io.concurrent(Fixture.run, .{ &ws_server, &listener, &hub }); defer _ = future.cancel(io) catch {};
var url_buf: [256]u8 = undefined; const url = try std.fmt.bufPrint(&url_buf, "http://127.0.0.1:{d}/xrpc/network.bsky.jetstream.listSegments", .{port}); var transport = zat.HttpTransport.init(io, testing.allocator); transport.keep_alive = false; defer transport.deinit(); var response = try transport.fetch(.{ .url = url, .max_response_size = 4096, .capture_headers = &.{"content-type"} }); defer response.deinit(testing.allocator); try testing.expectEqual(std.http.Status.service_unavailable, response.status); try testing.expectEqualStrings("application/json", response.headers.get("content-type").?); const parsed = try std.json.parseFromSlice(std.json.Value, testing.allocator, response.body, .{}); defer parsed.deinit(); try testing.expectEqualStrings("ServiceUnavailable", parsed.value.object.get("error").?.string); try testing.expect(std.mem.indexOf(u8, parsed.value.object.get("message").?.string, "bootstrap in progress") != null);}
test "xrpc error status lines: every emitted status has its own line" { // the key-ring 429 regression: unknown statuses rewrite to 400 and // break client retry classification try std.testing.expectEqualStrings("429 Too Many Requests", XrpcResponder.errStatusLine(429)); try std.testing.expectEqualStrings("401 Unauthorized", XrpcResponder.errStatusLine(401)); try std.testing.expectEqualStrings("503 Service Unavailable", XrpcResponder.errStatusLine(503)); try std.testing.expectEqualStrings("400 Bad Request", XrpcResponder.errStatusLine(400));}
test "startup gate: storage-free surfaces serve while everything else 503s as starting" { const testing = std.testing; var threaded: Io.Threaded = .init(testing.allocator, .{}); defer threaded.deinit(); const io = threaded.io(); var tail = tail_mod.Tail.init(testing.allocator, io, 1 << 20); defer tail.deinit(); var stats: metrics.Stats = .{}; var hub: Hub = .{ .allocator = testing.allocator, .io = io, .tail = &tail, .stats = &stats, .serving = .init(false), .storage_ready = .init(false), }; defer hub.deinit(); var context: ListenerContext = .{ .hub = &hub, .role = .public };
var listener = (Io.net.IpAddress{ .ip4 = .loopback(0) }).listen(io, .{ .reuse_address = true }) catch unreachable; defer listener.deinit(io); const port = listener.socket.address.getPort(); var ws_server = try websocket.Server(Handler).init(testing.allocator, io, .{ .port = port, .address = "127.0.0.1", .max_conn = 4, .handshake = .{ .max_size = 16 * 1024, .max_headers = 64 }, }); defer ws_server.deinit(); const Fixture = struct { fn run(server: *websocket.Server(Handler), server_listener: *Io.net.Server, listener_context: *ListenerContext) !void { server.runIo(server_listener, listener_context); } }; var future = try io.concurrent(Fixture.run, .{ &ws_server, &listener, &context }); defer _ = future.cancel(io) catch {};
var transport = zat.HttpTransport.init(io, testing.allocator); transport.keep_alive = false; defer transport.deinit(); var url_buf: [256]u8 = undefined;
// /status stays 200 with the probe-parseable minimal page: fleet-health // reads "upstream seq" and "uptime" from exactly these field names. var status = try transport.fetch(.{ .url = try std.fmt.bufPrint(&url_buf, "http://127.0.0.1:{d}/status", .{port}), .max_response_size = 1024 }); defer status.deinit(testing.allocator); try testing.expectEqual(std.http.Status.ok, status.status); try testing.expect(std.mem.indexOf(u8, status.body, "state starting") != null); try testing.expect(std.mem.indexOf(u8, status.body, "upstream seq 0") != null); try testing.expect(std.mem.indexOf(u8, status.body, "uptime ") != null);
// comptime endpoint surface needs no storage and serves during startup — // this is what keeps the homepage's httpInclude green through a deploy var eps = try transport.fetch(.{ .url = try std.fmt.bufPrint(&url_buf, "http://127.0.0.1:{d}{s}", .{ port, endpoints.json_path }), .max_response_size = 16384 }); defer eps.deinit(testing.allocator); try testing.expectEqual(std.http.Status.ok, eps.status);
var home = try transport.fetch(.{ .url = try std.fmt.bufPrint(&url_buf, "http://127.0.0.1:{d}/", .{port}), .max_response_size = 32768 }); defer home.deinit(testing.allocator); try testing.expectEqual(std.http.Status.ok, home.status);
// storage-backed paths 503 as starting: xrpc in the JSON error shape var xrpc_response = try transport.fetch(.{ .url = try std.fmt.bufPrint(&url_buf, "http://127.0.0.1:{d}/xrpc/network.bsky.jetstream.listSegments", .{port}), .max_response_size = 1024 }); defer xrpc_response.deinit(testing.allocator); try testing.expectEqual(std.http.Status.service_unavailable, xrpc_response.status); try testing.expect(std.mem.indexOf(u8, xrpc_response.body, "\"error\":\"ServiceUnavailable\"") != null); try testing.expect(std.mem.indexOf(u8, xrpc_response.body, "starting") != null);
var tab = try transport.fetch(.{ .url = try std.fmt.bufPrint(&url_buf, "http://127.0.0.1:{d}/status?tab=segments", .{port}), .max_response_size = 1024 }); defer tab.deinit(testing.allocator); try testing.expectEqual(std.http.Status.ok, tab.status); try testing.expect(std.mem.indexOf(u8, tab.body, "starting") != null);
// once main publishes the stores, the same connection path serves fully hub.storage_ready.store(true, .release); hub.serving.store(true, .release); var after = try transport.fetch(.{ .url = try std.fmt.bufPrint(&url_buf, "http://127.0.0.1:{d}/status", .{port}), .max_response_size = 8192 }); defer after.deinit(testing.allocator); try testing.expectEqual(std.http.Status.ok, after.status); try testing.expect(std.mem.indexOf(u8, after.body, "state starting") == null);}
test "debug readyz tracks listener readiness rather than ingest serving" { const testing = std.testing; var threaded: Io.Threaded = .init(testing.allocator, .{}); defer threaded.deinit(); const io = threaded.io(); var tail = tail_mod.Tail.init(testing.allocator, io, 1 << 20); defer tail.deinit(); var stats: metrics.Stats = .{}; var hub: Hub = .{ .allocator = testing.allocator, .io = io, .tail = &tail, .stats = &stats, .serving = .init(false), }; defer hub.deinit(); var context: ListenerContext = .{ .hub = &hub, .role = .debug };
var listener = (Io.net.IpAddress{ .ip4 = .loopback(0) }).listen(io, .{ .reuse_address = true }) catch unreachable; defer listener.deinit(io); const port = listener.socket.address.getPort(); var ws_server = try websocket.Server(Handler).init(testing.allocator, io, .{ .port = port, .address = "127.0.0.1", .max_conn = 4, .handshake = .{ .max_size = 16 * 1024, .max_headers = 64 }, }); defer ws_server.deinit(); const Fixture = struct { fn run(server: *websocket.Server(Handler), server_listener: *Io.net.Server, listener_context: *ListenerContext) !void { server.runIo(server_listener, listener_context); } }; var future = try io.concurrent(Fixture.run, .{ &ws_server, &listener, &context }); defer _ = future.cancel(io) catch {};
var url_buf: [128]u8 = undefined; const url = try std.fmt.bufPrint(&url_buf, "http://127.0.0.1:{d}/readyz", .{port}); var transport = zat.HttpTransport.init(io, testing.allocator); transport.keep_alive = false; defer transport.deinit(); var unavailable = try transport.fetch(.{ .url = url, .max_response_size = 256 }); defer unavailable.deinit(testing.allocator); try testing.expectEqual(std.http.Status.service_unavailable, unavailable.status); try testing.expectEqualStrings("not ready\n", unavailable.body);
hub.listeners_ready.store(true, .release); var ready = try transport.fetch(.{ .url = url, .max_response_size = 256 }); defer ready.deinit(testing.allocator); try testing.expectEqual(std.http.Status.ok, ready.status); try testing.expectEqualStrings("ok\n", ready.body); try testing.expect(!hub.serving.load(.acquire));}
test "accounts HTTP route verifies real archive state and rate limits by port-free peer IP" { const testing = std.testing; var threaded: Io.Threaded = .init(testing.allocator, .{}); defer threaded.deinit(); const io = threaded.io(); var fixture_arena = std.heap.ArenaAllocator.init(testing.allocator); defer fixture_arena.deinit(); const alloc = fixture_arena.allocator();
const did = "did:plc:statusverifyfixture"; const rev = "3k2abcdefghij"; const record = try zat.cbor.encodeAlloc(alloc, .{ .map = &.{ .{ .key = "$type", .value = .{ .text = "app.bsky.feed.post" } }, .{ .key = "text", .value = .{ .text = "status route verification" } }, } }); const record_cid = try zat.cbor.Cid.forDagCbor(alloc, record); var authoritative_tree = zat.mst.Mst.init(alloc); try authoritative_tree.put("app.bsky.feed.post/fixture", record_cid); const data_root = try authoritative_tree.rootCid(); const keypair = try zat.Keypair.fromSecretKey(.p256, @splat(43)); const signed = try zat.signCommit(alloc, .{ .did = did, .rev = rev, .data = data_root }, &keypair); const commit_cid_string = try zat.multibase.base32lower.encode(alloc, signed.cid.raw); const get_blocks_car = try zat.car.writeAlloc(alloc, .{ .roots = &.{}, .blocks = &.{.{ .cid_raw = signed.cid.raw, .data = signed.bytes }}, });
var identity_listener = (Io.net.IpAddress{ .ip4 = .loopback(0) }).listen(io, .{ .reuse_address = true }) catch unreachable; defer identity_listener.deinit(io); var identity_endpoint_buf: [96]u8 = undefined; const identity_endpoint = try std.fmt.bufPrint(&identity_endpoint_buf, "http://127.0.0.1:{d}", .{identity_listener.socket.address.getPort()}); const did_doc = try std.fmt.allocPrint(alloc, \\{{"id":"{s}","alsoKnownAs":["at://status.example"],"verificationMethod":[],"service":[{{"id":"#atproto_pds","type":"AtprotoPersonalDataServer","serviceEndpoint":"{s}"}}]}} , .{ did, identity_endpoint }); const latest_json = try std.fmt.allocPrint(alloc, "{{\"cid\":\"{s}\",\"rev\":\"{s}\"}}", .{ commit_cid_string, rev }); var identity_calls: std.atomic.Value(u32) = .init(0); const IdentityFixture = struct { fn run(server: *Io.net.Server, task_io: Io, calls: *std.atomic.Value(u32), doc: []const u8, latest: []const u8, blocks_car: []const u8) !void { while (true) { var stream = try server.accept(task_io); defer stream.close(task_io); var read_buf: [4096]u8 = undefined; var write_buf: [4096]u8 = undefined; var reader = stream.reader(task_io, &read_buf); var writer = stream.writer(task_io, &write_buf); var http = std.http.Server.init(&reader.interface, &writer.interface); var request = try http.receiveHead(); _ = calls.fetchAdd(1, .monotonic); const target = request.head.target; const response_body, const content_type = if (std.mem.indexOf(u8, target, "getLatestCommit") != null) .{ latest, "application/json" } else if (std.mem.indexOf(u8, target, "getBlocks") != null) .{ blocks_car, "application/vnd.ipld.car" } else .{ doc, "application/json" }; try request.respond(response_body, .{ .status = .ok, .keep_alive = false, .extra_headers = &.{.{ .name = "content-type", .value = content_type }}, }); } } }; var identity_future = try io.concurrent(IdentityFixture.run, .{ &identity_listener, io, &identity_calls, did_doc, latest_json, get_blocks_car }); defer _ = identity_future.cancel(io) catch {};
var tmp = testing.tmpDir(.{}); defer tmp.cleanup(); var data_dir_buf: [Io.Dir.max_path_bytes]u8 = undefined; const data_dir_len = try tmp.dir.realPath(io, &data_dir_buf); const data_dir = data_dir_buf[0..data_dir_len]; var archive = try archive_mod.Archive.init(testing.allocator, io, data_dir); defer archive.deinit(); _ = try archive.append(.{ .seq = 1, .witnessed_at = 1_750_000_000_000_000, .indexed_at = 0, .kind = .create, .did = did, .collection = "app.bsky.feed.post", .rkey = "fixture", .rev = rev, .payload = record, }, 1);
const meta_store = @import("../storage/meta_store.zig"); var meta = try meta_store.Store.open(testing.allocator, data_dir); defer meta.deinit(); var repo_status = try repo_store.Store.init(testing.allocator, io, &meta); defer repo_status.deinit(); try repo_status.put(did, .{ .status = .complete, .active = true, .rev = rev, .latest_rev = rev, .handle = "status.example", .pds = identity_endpoint, .host = "127.0.0.1", .record_count = 1, });
var tail = tail_mod.Tail.init(testing.allocator, io, 1 << 20); defer tail.deinit(); var stats: metrics.Stats = .{}; var hub: Hub = .{ .allocator = testing.allocator, .io = io, .tail = &tail, .stats = &stats, .archive = &archive, .repo_status = &repo_status, .data_dir = data_dir, .plc_url = identity_endpoint, }; defer hub.deinit();
var stream_listener = (Io.net.IpAddress{ .ip4 = .loopback(0) }).listen(io, .{ .reuse_address = true }) catch unreachable; defer stream_listener.deinit(io); const stream_port = stream_listener.socket.address.getPort(); var ws_server = try websocket.Server(Handler).init(testing.allocator, io, .{ .port = stream_port, .address = "127.0.0.1", .max_conn = 16, .handshake = .{ .max_size = 16 * 1024, .max_headers = 64 }, }); defer ws_server.deinit(); const StreamFixture = struct { fn run(server: *websocket.Server(Handler), listener: *Io.net.Server, fixture_hub: *Hub) !void { server.runIo(listener, fixture_hub); } }; var stream_future = try io.concurrent(StreamFixture.run, .{ &ws_server, &stream_listener, &hub }); defer _ = stream_future.cancel(io) catch {};
var status_url_buf: [256]u8 = undefined; const status_url = try std.fmt.bufPrint(&status_url_buf, "http://127.0.0.1:{d}/status?tab=accounts&account={s}", .{ stream_port, did }); var transport = zat.HttpTransport.init(io, testing.allocator); transport.keep_alive = false; defer transport.deinit(); for (0..4) |_| { const response = try transport.fetch(.{ .url = status_url, .accept = "text/html", .max_response_size = 256 * 1024 }); defer testing.allocator.free(response.body); try testing.expectEqual(std.http.Status.ok, response.status); try testing.expect(std.mem.indexOf(u8, response.body, "Repo verification") != null); try testing.expect(std.mem.indexOf(u8, response.body, "match-ok\">Match") != null); try testing.expect(std.mem.indexOf(u8, response.body, "repo action rate limit exceeded") == null); } try testing.expectEqual(@as(u32, 12), identity_calls.load(.monotonic));
const limited = try transport.fetch(.{ .url = status_url, .accept = "text/html", .max_response_size = 256 * 1024 }); defer testing.allocator.free(limited.body); try testing.expectEqual(std.http.Status.ok, limited.status); try testing.expect(std.mem.indexOf(u8, limited.body, "repo action rate limit exceeded") != null); try testing.expectEqual(@as(u32, 12), identity_calls.load(.monotonic));
const head = try transport.fetch(.{ .url = status_url, .method = .HEAD, .accept = "text/html", .max_response_size = 256 * 1024 }); defer testing.allocator.free(head.body); try testing.expectEqual(std.http.Status.ok, head.status); try testing.expectEqual(@as(usize, 0), head.body.len); try testing.expectEqual(@as(u32, 12), identity_calls.load(.monotonic));
hub.repo_action_rate_limits = false; const trusted = try transport.fetch(.{ .url = status_url, .accept = "text/html", .max_response_size = 256 * 1024 }); defer testing.allocator.free(trusted.body); try testing.expectEqual(std.http.Status.ok, trusted.status); try testing.expect(std.mem.indexOf(u8, trusted.body, "match-ok\">Match") != null); try testing.expect(std.mem.indexOf(u8, trusted.body, "repo action rate limit exceeded") == null); try testing.expectEqual(@as(u32, 15), identity_calls.load(.monotonic));}
test "cold-reader slots: cap admits, saturates, releases; zero is unbounded" { const testing = std.testing; var threaded: Io.Threaded = .init(testing.allocator, .{}); defer threaded.deinit(); const io = threaded.io(); var tail = tail_mod.Tail.init(testing.allocator, io, 1 << 20); defer tail.deinit(); var stats: metrics.Stats = .{}; var hub: Hub = .{ .allocator = testing.allocator, .io = io, .tail = &tail, .stats = &stats, .max_cold_readers = 2, }; try testing.expect(hub.tryAcquireColdSlot()); try testing.expect(hub.tryAcquireColdSlot()); try testing.expect(!hub.tryAcquireColdSlot()); // saturated hub.releaseColdSlot(); try testing.expect(hub.tryAcquireColdSlot()); // freed slot readmits hub.releaseColdSlot(); hub.releaseColdSlot(); try testing.expectEqual(@as(u32, 0), hub.cold_active.load(.monotonic));
hub.max_cold_readers = 0; // unbounded: never blocks, never counts try testing.expect(hub.tryAcquireColdSlot()); hub.releaseColdSlot(); try testing.expectEqual(@as(u32, 0), hub.cold_active.load(.monotonic));}
test "real-HTTP archive serving does not leak per request (prod RSS ratchet regression)" { // 2026-08-21: prod RSS grew ~575 B per archive request (anonymous heap, // smaps-verified). The Api layer alone is leak-free under this same // allocator, so this test drives the full socket -> dispatch -> responder // path; testing.allocator fails the test if any request leaks. const testing = std.testing; var threaded: Io.Threaded = .init(testing.allocator, .{}); defer threaded.deinit(); const io = threaded.io(); var tail = tail_mod.Tail.init(testing.allocator, io, 1 << 20); defer tail.deinit(); var stats: metrics.Stats = .{};
var tmp = testing.tmpDir(.{}); defer tmp.cleanup(); var path_buffer: [Io.Dir.max_path_bytes]u8 = undefined; const path_len = try tmp.dir.realPath(io, &path_buffer); var archive = try archive_mod.Archive.init(testing.allocator, io, path_buffer[0..path_len]); defer archive.deinit(); _ = try archive.append(.{ .seq = 1, .witnessed_at = 1, .indexed_at = 0, .kind = .create, .did = "did:plc:leakregression", .collection = "app.bsky.feed.post", .rkey = "3l3qo2vuowo2b", .rev = "3l3qo2vutsw2b", .payload = "\xa1\x64\x74\x65\x78\x74\x62\x68\x69", }, 1); try archive.rotate();
var xrpc_api: xrpcapi.Api = .{ .allocator = testing.allocator, .io = io, .archive = &archive }; var hub: Hub = .{ .allocator = testing.allocator, .io = io, .tail = &tail, .stats = &stats, .archive = &archive, .xrpc = &xrpc_api, }; defer hub.deinit();
var listener = (Io.net.IpAddress{ .ip4 = .loopback(0) }).listen(io, .{ .reuse_address = true }) catch unreachable; defer listener.deinit(io); const port = listener.socket.address.getPort(); var ws_server = try websocket.Server(Handler).init(testing.allocator, io, .{ .port = port, .address = "127.0.0.1", .max_conn = 4, .handshake = .{ .max_size = 16 * 1024, .max_headers = 64 }, }); defer ws_server.deinit(); const Fixture = struct { fn run(server: *websocket.Server(Handler), server_listener: *Io.net.Server, fixture_hub: *Hub) !void { server.runIo(server_listener, fixture_hub); } }; var future = try io.concurrent(Fixture.run, .{ &ws_server, &listener, &hub }); defer _ = future.cancel(io) catch {};
var url_buf: [256]u8 = undefined; const url = try std.fmt.bufPrint( &url_buf, "http://127.0.0.1:{d}/xrpc/network.bsky.jetstream.getBlock?segment=seg_0000000000.jss&blockIndex=0", .{port}, ); // fresh connection per request — the shape prod leaked under for (0..8) |_| { var transport = zat.HttpTransport.init(io, testing.allocator); transport.keep_alive = false; defer transport.deinit(); var response = try transport.fetch(.{ .url = url, .max_response_size = 1 << 20 }); defer response.deinit(testing.allocator); try testing.expectEqual(std.http.Status.ok, response.status); try testing.expect(response.body.len > 0); } // reused connection — prod leaked ~200 B/request here too var transport = zat.HttpTransport.init(io, testing.allocator); defer transport.deinit(); for (0..8) |_| { var response = try transport.fetch(.{ .url = url, .max_response_size = 1 << 20 }); defer response.deinit(testing.allocator); try testing.expectEqual(std.http.Status.ok, response.status); }}
test "obsolete live dictionary returns the upstream JSON rejection" { const testing = std.testing; var threaded: Io.Threaded = .init(testing.allocator, .{}); defer threaded.deinit(); const io = threaded.io(); var tail = tail_mod.Tail.init(testing.allocator, io, 1 << 20); defer tail.deinit(); var stats: metrics.Stats = .{}; var hub: Hub = .{ .allocator = testing.allocator, .io = io, .tail = &tail, .stats = &stats, .serving = .init(true), }; defer hub.deinit();
var listener = (Io.net.IpAddress{ .ip4 = .loopback(0) }).listen(io, .{ .reuse_address = true }) catch unreachable; defer listener.deinit(io); const port = listener.socket.address.getPort(); var ws_server = try websocket.Server(Handler).init(testing.allocator, io, .{ .port = port, .address = "127.0.0.1", .max_conn = 4, .handshake = .{ .max_size = 16 * 1024, .max_headers = 64 }, }); defer ws_server.deinit(); const Fixture = struct { fn run(server: *websocket.Server(Handler), server_listener: *Io.net.Server, fixture_hub: *Hub) !void { server.runIo(server_listener, fixture_hub); } }; var future = try io.concurrent(Fixture.run, .{ &ws_server, &listener, &hub }); defer _ = future.cancel(io) catch {};
const address: Io.net.IpAddress = .{ .ip4 = .loopback(port) }; var socket = try address.connect(io, .{ .mode = .stream }); defer socket.close(io); var write_buf: [1024]u8 = undefined; var writer = socket.writer(io, &write_buf); try writer.interface.print("GET /xrpc/network.bsky.jetstream.subscribeEvents?zstdDictionary={d} HTTP/1.1\r\nHost: localhost\r\nUpgrade: websocket\r\nConnection: Upgrade\r\nSec-WebSocket-Version: 13\r\nSec-WebSocket-Key: dGhlIHNhbXBsZSBub25jZQ==\r\nSec-WebSocket-Protocol: xrpc.v1.json\r\n\r\n", .{zstd.v2_dictionary_id ^ 1}); try writer.interface.flush(); var read_buf: [4096]u8 = undefined; var reader = socket.reader(io, &read_buf); const response = try reader.interface.allocRemaining(testing.allocator, .limited(4096)); defer testing.allocator.free(response); try testing.expect(std.mem.startsWith(u8, response, "HTTP/1.1 400 ")); const split = std.mem.indexOf(u8, response, "\r\n\r\n") orelse return error.MissingHeaders; try testing.expect(std.mem.indexOf(u8, response[0..split], "application/json") != null); const parsed = try std.json.parseFromSlice(std.json.Value, testing.allocator, response[split + 4 ..], .{}); defer parsed.deinit(); try testing.expectEqualStrings("UnknownZstdDictionary", parsed.value.object.get("error").?.string); try testing.expect(std.mem.indexOf(u8, parsed.value.object.get("message").?.string, "reconnect") != null);}