atproto relay in zig zlay.waow.tech
relay zig atproto
Something went wrong. Try again.
Zig
1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586878889909192939495969798991001011021031041051061071081091101111121131141151161171181191201211221231241251261271281291301311321331341351361371381391401411421431441451461471481491501511521531541551561571581591601611621631641651661671681691701711721731741751761771781791801811821831841851861871881891901911921931941951961971981992002012022032042052062072082092102112122132142152162172182192202212222232242252262272282292302312322332342352362372382392402412422432442452462472482492502512522532542552562572582592602612622632642652662672682692702712722732742752762772782792802812822832842852862872882892902912922932942952962972982993003013023033043053063073083093103113123133143153163173183193203213223233243253263273283293303313323333343353363373383393403413423433443453463473483493503513523533543553563573583593603613623633643653663673683693703713723733743753763773783793803813823833843853863873883893903913923933943953963973983994004014024034044054064074084094104114124134144154164174184194204214224234244254264274284294304314324334344354364374384394404414424434444454464474484494504514524534544554564574584594604614624634644654664674684694704714724734744754764774784794804814824834844854864874884894904914924934944954964974984995005015025035045055065075085095105115125135145155165175185195205215225235245255265275285295305315325335345355365375385395405415425435445455465475485495505515525535545555565575585595605615625635645655665675685695705715725735745755765775785795805815825835845855865875885895905915925935945955965975985996006016026036046056066076086096106116126136146156166176186196206216226236246256266276286296306316326336346356366376386396406416426436446456466476486496506516526536546556566576586596606616626636646656666676686696706716726736746756766776786796806816826836846856866876886896906916926936946956966976986997007017027037047057067077087097107117127137147157167177187197207217227237247257267277287297307317327337347357367377387397407417427437447457467477487497507517527537547557567577587597607617627637647657667677687697707717727737747757767777787797807817827837847857867877887897907917927937947957967977987998008018028038048058068078088098108118128138148158168178188198208218228238248258268278288298308318328338348358368378388398408418428438448458468478488498508518528538548558568578588598608618628638648658668678688698708718728738748758768778788798808818828838848858868878888898908918928938948958968978988999009019029039049059069079089099109119129139149159169179189199209219229239249259269279289299309319329339349359369379389399409419429439449459469479489499509519529539549559569579589599609619629639649659669679689699709719729739749759769779789799809819829839849859869879889899909919929939949959969979989991000100110021003100410051006100710081009101010111012101310141015101610171018101910201021102210231024102510261027102810291030103110321033103410351036103710381039104010411042104310441045104610471048104910501051105210531054105510561057105810591060106110621063106410651066106710681069107010711072107310741075107610771078107910801081108210831084108510861087108810891090109110921093109410951096109710981099110011011102110311041105110611071108110911101111111211131114111511161117111811191120112111221123112411251126112711281129113011311132113311341135113611371138113911401141114211431144114511461147114811491150115111521153115411551156115711581159116011611162116311641165116611671168116911701171117211731174117511761177117811791180118111821183118411851186118711881189119011911192119311941195119611971198119912001201120212031204120512061207120812091210121112121213121412151216121712181219122012211222122312241225122612271228122912301231123212331234123512361237123812391240124112421243124412451246124712481249125012511252125312541255125612571258125912601261126212631264126512661267126812691270127112721273127412751276127712781279128012811282128312841285128612871288128912901291129212931294129512961297129812991300130113021303130413051306130713081309131013111312131313141315131613171318131913201321132213231324132513261327132813291330133113321333133413351336133713381339134013411342134313441345134613471348134913501351135213531354135513561357135813591360136113621363136413651366136713681369137013711372137313741375137613771378137913801381138213831384138513861387138813891390139113921393139413951396139713981399140014011402140314041405140614071408140914101411141214131414141514161417141814191420142114221423142414251426142714281429143014311432143314341435143614371438143914401441144214431444144514461447144814491450145114521453145414551456145714581459146014611462146314641465146614671468146914701471147214731474147514761477147814791480148114821483148414851486148714881489149014911492149314941495149614971498149915001501150215031504150515061507150815091510151115121513151415151516151715181519152015211522152315241525152615271528152915301531153215331534153515361537153815391540154115421543154415451546154715481549155015511552155315541555155615571558155915601561156215631564156515661567156815691570157115721573157415751576157715781579158015811582158315841585158615871588158915901591159215931594159515961597159815991600160116021603160416051606160716081609161016111612161316141615161616171618161916201621162216231624162516261627162816291630163116321633163416351636163716381639164016411642164316441645164616471648164916501651165216531654165516561657165816591660166116621663166416651666166716681669167016711672167316741675167616771678167916801681168216831684168516861687168816891690169116921693169416951696169716981699170017011702170317041705170617071708170917101711171217131714171517161717171817191720172117221723172417251726172717281729173017311732173317341735173617371738173917401741174217431744174517461747174817491750175117521753175417551756175717581759176017611762176317641765176617671768176917701771177217731774177517761777177817791780178117821783178417851786178717881789179017911792179317941795179617971798179918001801180218031804180518061807180818091810181118121813181418151816181718181819182018211822182318241825182618271828182918301831183218331834183518361837183818391840184118421843184418451846184718481849185018511852185318541855185618571858185918601861186218631864186518661867186818691870187118721873187418751876187718781879188018811882188318841885188618871888188918901891189218931894189518961897189818991900190119021903190419051906190719081909191019111912191319141915191619171918191919201921192219231924192519261927192819291930193119321933193419351936193719381939194019411942194319441945194619471948194919501951195219531954195519561957195819591960196119621963196419651966//! relay broadcaster — fan-out firehose frames to downstream consumers//!//! features://! - ref-counted shared frames (one copy per broadcast, not per consumer)//! - ConsumerTooSlow CBOR error frame before disconnect//! - ping/pong keepalive (30s idle, 5s deadline)//! - non-blocking broadcast via per-consumer ring buffers + write threads//! - health/stats HTTP endpoints via websocket handshake routing
const std = @import("std");const Io = std.Io;const websocket = @import("websocket");const zat = @import("zat");const ring_buffer = @import("util/ring_buffer.zig");const event_log_mod = @import("event_log.zig");
const builtin = @import("builtin");const build_options = @import("build_options");const util = @import("util/util.zig");const Allocator = std.mem.Allocator;const log = std.log.scoped(.relay);
const timestamp = util.timestamp;
// --- stats ---
pub const Stats = struct { seq: std.atomic.Value(u64) = .{ .raw = 0 }, relay_seq: std.atomic.Value(u64) = .{ .raw = 0 }, consumer_count: std.atomic.Value(u64) = .{ .raw = 0 }, frames_in: std.atomic.Value(u64) = .{ .raw = 0 }, frames_out: std.atomic.Value(u64) = .{ .raw = 0 }, validated: std.atomic.Value(u64) = .{ .raw = 0 }, failed: std.atomic.Value(u64) = .{ .raw = 0 }, // per-reason failure breakdown (subsets of `failed`) failed_bad_did: std.atomic.Value(u64) = .{ .raw = 0 }, failed_bad_rev: std.atomic.Value(u64) = .{ .raw = 0 }, failed_missing_blocks: std.atomic.Value(u64) = .{ .raw = 0 }, failed_oversized_blocks: std.atomic.Value(u64) = .{ .raw = 0 }, failed_bad_structure: std.atomic.Value(u64) = .{ .raw = 0 }, failed_signature: std.atomic.Value(u64) = .{ .raw = 0 }, failed_commit_integrity: std.atomic.Value(u64) = .{ .raw = 0 }, // Proof failures split by cause, because they mean opposite things: a // sender that omitted prevData is behaving as expected for a pre-Sync-1.1 // PDS, while a proof that fails to invert means the commit or our // verification is wrong. Counting both as sync_1_1 hid a zat MST bug that // was dropping valid commits at 5% of validated traffic. failed_sync_1_1: std.atomic.Value(u64) = .{ .raw = 0 }, failed_prev_data_missing: std.atomic.Value(u64) = .{ .raw = 0 }, failed_host_authority: std.atomic.Value(u64) = .{ .raw = 0 }, failed_future_rev: std.atomic.Value(u64) = .{ .raw = 0 }, skipped: std.atomic.Value(u64) = .{ .raw = 0 }, rate_limited: std.atomic.Value(u64) = .{ .raw = 0 }, rate_limited_sec: std.atomic.Value(u64) = .{ .raw = 0 }, rate_limited_hour: std.atomic.Value(u64) = .{ .raw = 0 }, rate_limited_day: std.atomic.Value(u64) = .{ .raw = 0 }, decode_errors: std.atomic.Value(u64) = .{ .raw = 0 }, cache_hits: std.atomic.Value(u64) = .{ .raw = 0 }, cache_misses: std.atomic.Value(u64) = .{ .raw = 0 }, slow_consumers: std.atomic.Value(u64) = .{ .raw = 0 }, connected_inbound: std.atomic.Value(u64) = .{ .raw = 0 }, cache_evictions: std.atomic.Value(u64) = .{ .raw = 0 }, // chain continuity failures by reason (since vs stored rev, prevData vs stored data CID) chain_breaks_since: std.atomic.Value(u64) = .{ .raw = 0 }, chain_breaks_prev_data: std.atomic.Value(u64) = .{ .raw = 0 }, pool_backpressure: std.atomic.Value(u64) = .{ .raw = 0 }, // keepalive pinger (ws_ping.zig) — pings sent, connections torn down for // unanswered pings, and ping writes that themselves failed ws_pings_sent: std.atomic.Value(u64) = .{ .raw = 0 }, ws_ping_timeout_closes: std.atomic.Value(u64) = .{ .raw = 0 }, ws_ping_write_failures: std.atomic.Value(u64) = .{ .raw = 0 }, // host authority resolution metrics host_authority_checks: std.atomic.Value(u64) = .{ .raw = 0 }, host_authority_is_new: std.atomic.Value(u64) = .{ .raw = 0 }, host_authority_host_changed: std.atomic.Value(u64) = .{ .raw = 0 }, host_authority_time_us: std.atomic.Value(u64) = .{ .raw = 0 }, host_authority_cache_hits: std.atomic.Value(u64) = .{ .raw = 0 }, host_authority_network_checks: std.atomic.Value(u64) = .{ .raw = 0 }, host_authority_network_time_us: std.atomic.Value(u64) = .{ .raw = 0 }, host_authority_pool_wait_us: std.atomic.Value(u64) = .{ .raw = 0 }, // per-branch reject breakdown (subsets of failed_host_authority). // lets us tell whether rejects come from stale cached forgeries // (host_mismatch), genuinely forged DIDs (resolve/no_endpoint), // or a mismatch between DID doc endpoint and our host table. host_authority_reject_parse_did: std.atomic.Value(u64) = .{ .raw = 0 }, host_authority_reject_resolve: std.atomic.Value(u64) = .{ .raw = 0 }, host_authority_reject_no_endpoint: std.atomic.Value(u64) = .{ .raw = 0 }, host_authority_reject_bad_url: std.atomic.Value(u64) = .{ .raw = 0 }, host_authority_reject_unknown_host: std.atomic.Value(u64) = .{ .raw = 0 }, host_authority_reject_host_mismatch: std.atomic.Value(u64) = .{ .raw = 0 }, // subscriber reconnect-cycle disconnect breakdown. only handshake-or-earlier // failures accumulate to failed_attempts (reset_failures fires after // ws_handshake succeeds, before read_loop). 15 consecutive of these → // host moves to 'exhausted'. read_loop disconnects do not directly // exhaust unless the next handshake also fails. subscriber_disconnect_dns_connect: std.atomic.Value(u64) = .{ .raw = 0 }, subscriber_disconnect_tls_init: std.atomic.Value(u64) = .{ .raw = 0 }, subscriber_disconnect_ws_handshake: std.atomic.Value(u64) = .{ .raw = 0 }, subscriber_disconnect_read_loop: std.atomic.Value(u64) = .{ .raw = 0 }, // frame pool memory pressure pool_queued_bytes: std.atomic.Value(u64) = .{ .raw = 0 }, // persist/broadcast pipeline contention persist_order_spins: std.atomic.Value(u64) = .{ .raw = 0 }, broadcast_queue_push_lock_spins: std.atomic.Value(u64) = .{ .raw = 0 }, broadcast_queue_full: std.atomic.Value(u64) = .{ .raw = 0 }, broadcast_queue_depth_hwm: std.atomic.Value(u32) = .{ .raw = 0 }, broadcast_no_consumers: std.atomic.Value(u64) = .{ .raw = 0 }, start_time: i64 = 0,};
// --- shared frame (ref-counted) ---
pub const SharedFrame = struct { data: []const u8, ref_count: std.atomic.Value(u32), allocator: Allocator,
pub fn create(allocator: Allocator, data: []const u8) !*SharedFrame { const frame = try allocator.create(SharedFrame); const duped = try allocator.dupe(u8, data); frame.* = .{ .data = duped, .ref_count = .{ .raw = 1 }, .allocator = allocator, }; return frame; }
pub fn acquire(self: *SharedFrame) void { _ = self.ref_count.fetchAdd(1, .monotonic); }
pub fn release(self: *SharedFrame) void { if (self.ref_count.fetchSub(1, .acq_rel) == 1) { self.allocator.free(self.data); self.allocator.destroy(self); } }};
// --- ConsumerTooSlow error frame ---
/// build the ConsumerTooSlow CBOR error frame using the SDK encoder./// header: {op: -1}, payload: {error: "ConsumerTooSlow", message: "..."}fn buildErrorFrame(allocator: Allocator) ?[]const u8 { const cbor = zat.cbor;
// header: {op: -1} — spec says no `t` field on error frames const header: cbor.Value = .{ .map = &.{ .{ .key = "op", .value = .{ .negative = -1 } }, } };
// payload: {error: "ConsumerTooSlow", message: "consumer buffer full"} const payload: cbor.Value = .{ .map = &.{ .{ .key = "error", .value = .{ .text = "ConsumerTooSlow" } }, .{ .key = "message", .value = .{ .text = "consumer buffer full" } }, } };
const header_bytes = cbor.encodeAlloc(allocator, header) catch return null; const payload_bytes = cbor.encodeAlloc(allocator, payload) catch { allocator.free(header_bytes); return null; };
var frame = allocator.alloc(u8, header_bytes.len + payload_bytes.len) catch { allocator.free(header_bytes); allocator.free(payload_bytes); return null; }; @memcpy(frame[0..header_bytes.len], header_bytes); @memcpy(frame[header_bytes.len..], payload_bytes);
allocator.free(header_bytes); allocator.free(payload_bytes);
return frame;}
/// encode an error frame into caller-provided buffer./// used for InvalidCursor / FutureCursor error frames./// header: {op: -1}, payload: {error: <name>, message: <msg>}fn encodeErrorFrame(error_name: []const u8, message: []const u8, out: []u8) ?[]const u8 { const cbor = zat.cbor; const alloc = std.heap.c_allocator;
const header: cbor.Value = .{ .map = &.{ .{ .key = "op", .value = .{ .negative = -1 } }, } }; const payload: cbor.Value = .{ .map = &.{ .{ .key = "error", .value = .{ .text = error_name } }, .{ .key = "message", .value = .{ .text = message } }, } };
const header_bytes = cbor.encodeAlloc(alloc, header) catch return null; defer alloc.free(header_bytes); const payload_bytes = cbor.encodeAlloc(alloc, payload) catch return null; defer alloc.free(payload_bytes);
const total = header_bytes.len + payload_bytes.len; if (total > out.len) return null; @memcpy(out[0..header_bytes.len], header_bytes); @memcpy(out[header_bytes.len..][0..payload_bytes.len], payload_bytes);
return out[0..total];}
/// encode an #info message frame into caller-provided buffer./// used for OutdatedCursor. this is a message frame (op: 1), not an error frame./// header: {op: 1, t: "#info"}, payload: {name: <name>, message: <msg>}fn encodeInfoMessage(name: []const u8, message: []const u8, out: []u8) ?[]const u8 { const cbor = zat.cbor; const alloc = std.heap.c_allocator;
const header: cbor.Value = .{ .map = &.{ .{ .key = "op", .value = .{ .unsigned = 1 } }, .{ .key = "t", .value = .{ .text = "#info" } }, } }; const payload: cbor.Value = .{ .map = &.{ .{ .key = "name", .value = .{ .text = name } }, .{ .key = "message", .value = .{ .text = message } }, } };
const header_bytes = cbor.encodeAlloc(alloc, header) catch return null; defer alloc.free(header_bytes); const payload_bytes = cbor.encodeAlloc(alloc, payload) catch return null; defer alloc.free(payload_bytes);
const total = header_bytes.len + payload_bytes.len; if (total > out.len) return null; @memcpy(out[0..header_bytes.len], header_bytes); @memcpy(out[header_bytes.len..][0..payload_bytes.len], payload_bytes);
return out[0..total];}
// --- frame re-sequencing ---
/// resequence a firehose frame: replace the payload's "seq" with relay_seq./// header bytes are preserved as-is, only the payload is re-encoded./// returns null on decode failure; caller owns returned memory via allocator.pub fn resequenceFrame(allocator: Allocator, data: []const u8, relay_seq: u64) ?[]u8 { const cbor = zat.cbor;
// split frame: [header CBOR] [payload CBOR] const header_result = cbor.decode(allocator, data) catch return null; const header_end = header_result.consumed; const payload_data = data[header_end..];
// check header type — commit frames need tooBig normalization const is_commit = blk: { const header_val = switch (header_result.value) { .map => |entries| entries, else => break :blk false, }; for (header_val) |entry| { if (std.mem.eql(u8, entry.key, "t")) { const t = switch (entry.value) { .text => |s| s, else => break :blk false, }; break :blk std.mem.eql(u8, t, "#commit"); } } break :blk false; };
// decode payload map const payload = cbor.decodeAll(allocator, payload_data) catch return null; const old_entries = switch (payload) { .map => |entries| entries, else => return null, };
// rebuild map entries with seq replaced var new_entries: std.ArrayListUnmanaged(cbor.Value.MapEntry) = .empty; var found_seq = false; var found_too_big = false; for (old_entries) |entry| { if (std.mem.eql(u8, entry.key, "seq")) { new_entries.append(allocator, .{ .key = "seq", .value = .{ .unsigned = relay_seq } }) catch return null; found_seq = true; } else { if (std.mem.eql(u8, entry.key, "tooBig")) found_too_big = true; new_entries.append(allocator, entry) catch return null; } } if (!found_seq) { new_entries.append(allocator, .{ .key = "seq", .value = .{ .unsigned = relay_seq } }) catch return null; } // lexicon requires tooBig on #commit — some PDSes omit it, hydrant et al. reject the frame if (is_commit and !found_too_big) { new_entries.append(allocator, .{ .key = "tooBig", .value = .{ .boolean = false } }) catch return null; }
const new_payload: cbor.Value = .{ .map = new_entries.items }; const new_payload_bytes = cbor.encodeAlloc(allocator, new_payload) catch return null;
// combine: original header bytes + resequenced payload var result = allocator.alloc(u8, header_end + new_payload_bytes.len) catch return null; @memcpy(result[0..header_end], data[0..header_end]); @memcpy(result[header_end..], new_payload_bytes);
return result;}
// --- broadcast queue (worker → fiber handoff) ---
/// item produced by frame workers, consumed by the broadcaster fiber./// data is heap-allocated by the worker; the broadcaster fiber frees it after broadcast.pub const BroadcastItem = struct { seq: u64, data: []const u8,};
/// MPSC ring buffer: multiple frame worker threads push, one broadcaster fiber pops./// uses an atomic spinlock for push (no Io.Mutex — works from any execution context)./// pop is single-consumer (broadcaster fiber only), no locking needed.pub const BroadcastQueue = struct { const CAPACITY = 8192;
items: [CAPACITY]BroadcastItem = undefined, head: std.atomic.Value(u32) = .{ .raw = 0 }, tail: std.atomic.Value(u32) = .{ .raw = 0 }, push_lock: std.atomic.Value(u32) = .{ .raw = 0 }, allocator: Allocator,
pub fn init(allocator: Allocator) BroadcastQueue { return .{ .allocator = allocator }; }
/// approximate queue depth (may be slightly stale — for metrics only) pub fn depth(self: *const BroadcastQueue) u32 { const tail = self.tail.load(.monotonic); const head = self.head.load(.monotonic); return if (tail >= head) tail - head else CAPACITY - head + tail; }
/// push an item (called by worker threads). spins until space is available. /// matches Indigo semantics: every persisted event reaches live broadcast. /// the broadcaster fiber drains at memory speed (no I/O), so spin is brief. pub fn push(self: *BroadcastQueue, seq: u64, data: []const u8, stats: *Stats) void { var full_spins: u32 = 0; while (true) { // acquire spinlock var lock_spins: u32 = 0; while (self.push_lock.cmpxchgWeak(0, 1, .acquire, .monotonic) != null) { lock_spins += 1; std.atomic.spinLoopHint(); } if (lock_spins > 0) { _ = stats.broadcast_queue_push_lock_spins.fetchAdd(lock_spins, .monotonic); }
const tail = self.tail.load(.monotonic); const next_tail = (tail + 1) % CAPACITY; if (next_tail == self.head.load(.acquire)) { // full — release lock, yield, retry self.push_lock.store(0, .release); // saturate: this is a diagnostic counter, and the queue staying // full is exactly when it must not take the process down. a // stalled drain spun this past 2^32 and panicked the worker // under ReleaseSafe, turning "consumers are slow" into a crash. full_spins +|= 1; std.atomic.spinLoopHint(); continue; }
self.items[tail] = .{ .seq = seq, .data = data }; self.tail.store(next_tail, .release); self.push_lock.store(0, .release);
if (full_spins > 0) { _ = stats.broadcast_queue_full.fetchAdd(full_spins, .monotonic); } // update high-water mark (racy but directionally correct) const d = self.depth(); const hwm = stats.broadcast_queue_depth_hwm.load(.monotonic); if (d > hwm) { _ = stats.broadcast_queue_depth_hwm.cmpxchgWeak(hwm, d, .monotonic, .monotonic); } return; } }
/// peek the seq at the head without removing it (broadcaster fiber only). /// used to gate emission on the durable watermark. null if empty. pub fn peekSeq(self: *BroadcastQueue) ?u64 { const head = self.head.load(.monotonic); if (head == self.tail.load(.acquire)) return null; // empty return self.items[head].seq; }
/// pop an item (called by broadcaster fiber only). returns null if empty. pub fn pop(self: *BroadcastQueue) ?BroadcastItem { const head = self.head.load(.monotonic); if (head == self.tail.load(.acquire)) return null; // empty
const item = self.items[head]; self.head.store((head + 1) % CAPACITY, .release); return item; }};
// --- consumer ---
const ping_interval_ns: u64 = 30 * std.time.ns_per_s;const ping_timeout_ns: u64 = 5 * std.time.ns_per_s;
pub const Consumer = struct { const BUFFER_CAP = 65536;
conn: *websocket.Conn, allocator: Allocator, buf: [BUFFER_CAP]*SharedFrame = undefined, write_pos: usize = 0, read_pos: usize = 0, buf_len: usize = 0, alive: std.atomic.Value(bool) = .{ .raw = true }, mutex: Io.Mutex = Io.Mutex.init, cond: Io.Condition = Io.Condition.init, future: ?Io.Future(void) = null, last_send_time: i128 = 0, io: Io,
/// push a shared frame to this consumer's send buffer. /// acquires a reference. returns false if full (consumer too slow) or /// dead — a disconnected consumer must not keep accumulating frames /// (replay for a flapping consumer compounded into an OOM otherwise). pub fn enqueue(self: *Consumer, frame: *SharedFrame) bool { self.mutex.lockUncancelable(self.io); defer self.mutex.unlock(self.io);
if (!self.alive.load(.acquire)) return false; if (self.buf_len == BUFFER_CAP) return false;
frame.acquire(); self.buf[self.write_pos] = frame; self.write_pos = (self.write_pos + 1) % BUFFER_CAP; self.buf_len += 1; self.cond.signal(self.io); return true; }
/// push raw bytes (for replay, error frames). allocates a new SharedFrame. pub fn enqueueRaw(self: *Consumer, data: []const u8) bool { const frame = SharedFrame.create(self.allocator, data) catch return false; // enqueue will acquire, so we start with refcount 1 (for our ownership) // then release our ownership after enqueue if (self.enqueue(frame)) { frame.release(); // drop our ref, consumer write loop holds one return true; } frame.release(); return false; }
fn dequeue(self: *Consumer) ?*SharedFrame { // caller must hold mutex if (self.buf_len == 0) return null; const frame = self.buf[self.read_pos]; self.read_pos = (self.read_pos + 1) % BUFFER_CAP; self.buf_len -= 1; return frame; }
fn writeLoop(self: *Consumer) void { self.last_send_time = Io.Timestamp.now(self.io, .real).nanoseconds;
while (self.alive.load(.acquire)) { var frame: ?*SharedFrame = null; { self.mutex.lockUncancelable(self.io); defer self.mutex.unlock(self.io); if (self.buf_len == 0) { // no data: poll briefly instead of blocking on cond // (Io.Condition has no timedWait, so we poll to allow periodic ping checks) self.mutex.unlock(self.io); self.io.sleep(Io.Duration.fromMilliseconds(100), .awake) catch {}; self.mutex.lockUncancelable(self.io); } frame = self.dequeue(); } if (frame) |f| { defer f.release(); self.conn.writeBin(f.data) catch { self.alive.store(false, .release); break; }; self.last_send_time = Io.Timestamp.now(self.io, .real).nanoseconds; } else { // no data — check if we should send a ping self.maybePing(); } } // drain remaining buffered frames while (true) { self.mutex.lockUncancelable(self.io); const frame = self.dequeue(); self.mutex.unlock(self.io); if (frame) |f| { f.release(); } else break; } // close the connection from the writeLoop thread to unblock readLoop. // this MUST happen here, not from the broadcast thread — closing from // broadcast races with readLoop's setsockopt/read calls on the socket, // causing EBADF → unreachable panic under ReleaseSafe. self.conn.close(.{}) catch {}; }
fn maybePing(self: *Consumer) void { const now = Io.Timestamp.now(self.io, .real).nanoseconds; const elapsed: u64 = @intCast(@max(0, now - self.last_send_time)); if (elapsed >= ping_interval_ns) { // send ping self.conn.writePing(&.{}) catch { self.alive.store(false, .release); return; }; self.last_send_time = now; } }
pub fn shutdown(self: *Consumer) void { self.alive.store(false, .release); self.cond.signal(self.io); if (self.future) |*f| f.cancel(self.io); self.future = null; }};
// --- broadcaster ---
/// per-chunk bound for disk replay: caps the transient allocation of one/// consumer connect at ~16 MiB regardless of how much backlog its cursor/// spans. worst-case per-frame size (5 MiB max_message_size) is still/// covered: max_bytes is checked before each read, so a chunk holds at most/// max_bytes + one frame.const replay_chunk_limits: event_log_mod.PlaybackLimits = .{ .max_entries = 4096, .max_bytes = 16 * 1024 * 1024,};
const history_capacity = 50_000;const FrameHistory = ring_buffer.RingBuffer(history_capacity);
/// function pointer for HTTP fallback handler (set by main.zig)pub const HttpFallbackFn = *const fn ( *websocket.Conn, []const u8, []const u8, []const u8, *const websocket.Handshake.KeyValue, ?*anyopaque,) void;
pub const Broadcaster = struct { allocator: Allocator, consumers: std.ArrayListUnmanaged(*Consumer) = .empty, consumers_mutex: Io.Mutex = Io.Mutex.init, /// ordering spinlock for persist → queue push atomicity. /// atomic spinlock (not Io.Mutex) so both Threaded workers and Evented admin /// can participate in the same ordered publication path. persist_order: std.atomic.Value(u32) = .{ .raw = 0 }, /// worker → fiber handoff: frame workers push here, broadcaster fiber pops. broadcast_queue: BroadcastQueue, history: FrameHistory, persist: ?*event_log_mod.DiskPersist = null, db_queue: ?*event_log_mod.DbRequestQueue = null, stats: Stats = .{}, error_frame: ?[]const u8 = null, http_fallback: ?HttpFallbackFn = null, http_fallback_ctx: ?*anyopaque = null, io: Io, shutdown: *std.atomic.Value(bool), producers_done: std.atomic.Value(bool) = .{ .raw = false },
pub fn init(allocator: Allocator, io: Io, shutdown: *std.atomic.Value(bool)) Broadcaster { return .{ .allocator = allocator, .broadcast_queue = BroadcastQueue.init(allocator), .history = FrameHistory.init(allocator, io), .stats = .{ .start_time = timestamp(io) }, .error_frame = buildErrorFrame(allocator), .io = io, .shutdown = shutdown, }; }
pub fn deinit(self: *Broadcaster) void { self.consumers_mutex.lockUncancelable(self.io); defer self.consumers_mutex.unlock(self.io); for (self.consumers.items) |consumer| { consumer.shutdown(); self.allocator.destroy(consumer); } self.consumers.deinit(self.allocator); self.history.deinit(); if (self.error_frame) |ef| self.allocator.free(ef); while (self.broadcast_queue.pop()) |item| self.allocator.free(@constCast(item.data)); }
pub fn addConsumer(self: *Broadcaster, conn: *websocket.Conn) !*Consumer { const consumer = try self.allocator.create(Consumer); consumer.* = .{ .conn = conn, .allocator = self.allocator, .io = self.io, }; consumer.future = self.io.concurrent(Consumer.writeLoop, .{consumer}) catch { self.allocator.destroy(consumer); return error.ThreadSpawnFailed; };
self.consumers_mutex.lockUncancelable(self.io); defer self.consumers_mutex.unlock(self.io); self.consumers.append(self.allocator, consumer) catch { consumer.shutdown(); self.allocator.destroy(consumer); return error.OutOfMemory; }; _ = self.stats.consumer_count.fetchAdd(1, .monotonic); log.info("consumer connected ({d} total)", .{self.consumers.items.len}); return consumer; }
pub fn removeConsumer(self: *Broadcaster, consumer: *Consumer) void { { self.consumers_mutex.lockUncancelable(self.io); defer self.consumers_mutex.unlock(self.io); for (self.consumers.items, 0..) |c, i| { if (c == consumer) { _ = self.consumers.swapRemove(i); // only decrement if we found it — dropSlowConsumer already // decremented for slow consumers removed during broadcast() _ = self.stats.consumer_count.fetchSub(1, .monotonic); break; } } } consumer.shutdown(); log.info("consumer disconnected ({d} remaining)", .{self.consumers.items.len}); self.allocator.destroy(consumer); }
/// broadcast a frame to all consumers. non-blocking — just enqueues. /// drops slow consumers whose buffers are full after sending ConsumerTooSlow. pub fn broadcast(self: *Broadcaster, seq: u64, data: []const u8) void { self.stats.seq.store(seq, .release);
// add to history for cursor replay _ = self.history.push(seq, data);
// fast path: skip SharedFrame allocation when nobody is listening if (self.stats.consumer_count.load(.acquire) == 0) { _ = self.stats.broadcast_no_consumers.fetchAdd(1, .monotonic); return; }
// create one shared frame for all consumers const frame = SharedFrame.create(self.allocator, data) catch return; defer frame.release(); // release broadcaster's reference
self.consumers_mutex.lockUncancelable(self.io); defer self.consumers_mutex.unlock(self.io);
var frames_sent: u64 = 0; var i: usize = 0; while (i < self.consumers.items.len) { const consumer = self.consumers.items[i]; if (!consumer.alive.load(.acquire)) { // remove from broadcast list but don't destroy — // Handler.close() → removeConsumer() owns cleanup _ = self.consumers.swapRemove(i); _ = self.stats.consumer_count.fetchSub(1, .monotonic); continue; } if (consumer.enqueue(frame)) { frames_sent += 1; } else { // buffer full — send ConsumerTooSlow error, then drop self.dropSlowConsumer(consumer); _ = self.consumers.swapRemove(i); _ = self.stats.consumer_count.fetchSub(1, .monotonic); _ = self.stats.slow_consumers.fetchAdd(1, .monotonic); continue; } i += 1; } _ = self.stats.frames_out.fetchAdd(frames_sent, .monotonic); }
fn dropSlowConsumer(self: *Broadcaster, consumer: *Consumer) void { log.warn("dropping slow consumer (ConsumerTooSlow)", .{}); // try to send error frame directly (bypass buffer since it's full) if (self.error_frame) |ef| { consumer.conn.writeBin(ef) catch {}; } // signal consumer to stop — writeLoop checks alive and exits, then // Handler.close → removeConsumer handles shutdown + destroy. // do NOT close the socket here: readLoop (websocket server thread) may // be calling setsockopt/read on it concurrently. closing from a third // thread causes EBADF → unreachable panic in posix.setsockopt under // ReleaseSafe. let writeLoop's exit trigger the natural close sequence. consumer.alive.store(false, .release); consumer.cond.signal(consumer.io); // consumer is removed from list by broadcast() caller (swapRemove). // cleanup deferred to removeConsumer to avoid double-destroy. }
/// two-phase cursor replay: disk (diskpersist) first, then in-memory ring buffer. /// the consumer is already in the live broadcast list, so frames arriving /// during replay are buffered — no gap possible. /// /// disk playback runs on pool_io (Threaded) via request/reply — playback() /// holds the DiskPersist mutex and reads files, all of which require Threaded Io. pub fn replayTo(self: *Broadcaster, consumer: *Consumer, cursor: u64) void { // phase 1: disk replay from diskpersist via request/reply, in bounded // chunks. materializing the whole backlog per connect is an // O(retention) allocation — a flapping consumer stacking those // replays OOMed the canary (2026-08-06). each chunk is capped, the // consumer's liveness is re-checked between chunks, and a dead or // slow consumer aborts the replay entirely. if (self.persist) |dp| { var since = cursor; var replayed: usize = 0; while (consumer.alive.load(.acquire)) { var req: event_log_mod.PlaybackRequest = .{ .since = since, .allocator = self.allocator, .limits = replay_chunk_limits, }; dp.enqueuePlayback(&req);
// poll until pool_io worker completes the request. // SAFETY: req is stack-local — we MUST wait for the worker to finish before // returning, otherwise the stack frame unwinds while the worker still holds &req. // if sleep fails (shutdown/io error), fall back to spin-wait. the host_ops worker // drains all pending playback requests before exiting, so this is bounded. while (!req.done.load(.acquire)) { self.io.sleep(Io.Duration.fromMicroseconds(100), .awake) catch { while (!req.done.load(.acquire)) std.atomic.spinLoopHint(); break; }; }
defer { for (req.entries.items) |e| self.allocator.free(e.data); req.entries.deinit(self.allocator); }
if (req.err) |err| { log.warn("disk replay failed: {s}, falling back to memory", .{@errorName(err)}); self.replayFromMemory(consumer, since); return; }
if (req.entries.items.len == 0) break; // caught up
var reseq_arena = std.heap.ArenaAllocator.init(self.allocator); defer reseq_arena.deinit();
for (req.entries.items) |entry| { // resequence: replace upstream seq in CBOR with relay-assigned seq const frame_data = resequenceFrame(reseq_arena.allocator(), entry.data, entry.seq) orelse entry.data; if (!consumer.enqueueRaw(frame_data)) { log.warn("replay aborted after {d} frames (consumer dead or buffer full, cursor={d})", .{ replayed, cursor }); return; } replayed += 1; since = entry.seq; } } if (replayed > 0) { log.info("replayed {d} frames from disk (cursor={d})", .{ replayed, cursor }); } return; }
// phase 2 fallback: in-memory ring buffer (no persist configured) self.replayFromMemory(consumer, cursor); }
fn replayFromMemory(self: *Broadcaster, consumer: *Consumer, cursor: u64) void { self.history.mutex.lockUncancelable(self.io); defer self.history.mutex.unlock(self.io);
var i: usize = 0; while (i < self.history.len) : (i += 1) { const idx = (self.history.read_pos + i) % history_capacity; const entry = self.history.entries[idx]; if (entry.seq > cursor) { if (!consumer.enqueueRaw(entry.data)) break; } } }
pub fn consumerCount(self: *Broadcaster) usize { self.consumers_mutex.lockUncancelable(self.io); defer self.consumers_mutex.unlock(self.io); return self.consumers.items.len; }
/// broadcast loop — drains the broadcast queue, fans out to consumers /// and calls broadcast() for each item. this is the only path that touches /// consumers_mutex / consumer.mutex / consumer.cond. pub fn runBroadcastLoop(self: *Broadcaster) void { log.info("broadcaster fiber started", .{}); while (true) { const finished = self.producers_done.load(.acquire); // only emit frames that are durably on disk: hold any whose seq is // past the persister's committed watermark until the next flush // advances it. without a persister (degenerate no-persist mode), // nothing is gated. see docs/incident-2026-05-31-seq-rewind.md. const committed: u64 = if (self.persist) |dp| dp.committed_seq.load(.acquire) else std.math.maxInt(u64);
var drained: usize = 0; while (self.broadcast_queue.peekSeq()) |seq| { if (seq > committed) break; // not durable yet — hold const item = self.broadcast_queue.pop() orelse break; self.broadcast(item.seq, item.data); self.allocator.free(@constCast(item.data)); drained += 1; } if (finished) break; if (drained == 0) { // nothing emittable — yield briefly so other fibers run self.io.sleep(Io.Duration.fromMilliseconds(1), .awake) catch return; } } while (self.broadcast_queue.pop()) |item| { self.allocator.free(@constCast(item.data)); } log.info("broadcaster fiber stopped", .{}); }};
// --- websocket handler ---
pub const Handler = struct { consumer: ?*Consumer, broadcaster: *Broadcaster, conn: *websocket.Conn, cursor: ?u64, invalid_cursor: bool = false,
/// called during handshake — validate path, parse cursor. /// do NOT start the write thread here (handshake reply hasn't been sent yet). pub fn init(handshake: *const websocket.Handshake, conn: *websocket.Conn, ctx: *Broadcaster) !Handler { const url = handshake.url; const path = if (std.mem.indexOfScalar(u8, url, '?')) |qi| url[0..qi] else url;
if (!std.mem.eql(u8, path, "/xrpc/com.atproto.sync.subscribeRepos")) { return error.NotFound; }
// parse cursor (deferred until afterInit when connection is fully upgraded) var cursor: ?u64 = null; var invalid_cursor = false; if (std.mem.indexOf(u8, url, "cursor=")) |cursor_start| { const value_start = cursor_start + "cursor=".len; const value_end = std.mem.indexOfScalarPos(u8, url, value_start, '&') orelse url.len; const cursor_str = url[value_start..value_end]; cursor = std.fmt.parseInt(u64, cursor_str, 10) catch blk: { invalid_cursor = true; break :blk null; }; }
return .{ .consumer = null, .broadcaster = ctx, .conn = conn, .cursor = cursor, .invalid_cursor = invalid_cursor }; }
/// called AFTER the websocket handshake reply has been sent to the client. /// safe to start the consumer write thread and begin sending frames. pub fn afterInit(self: *Handler, ctx: *Broadcaster) !void { const consumer = ctx.addConsumer(self.conn) catch return error.Unexpected; self.consumer = consumer;
if (self.invalid_cursor) { // spec: send error frame with InvalidCursor, then close var buf: [256]u8 = undefined; if (encodeErrorFrame("InvalidCursor", "cursor parameter is not a valid integer", &buf)) |frame| { consumer.conn.writeBin(frame) catch {}; } consumer.conn.close(.{}) catch {}; return; }
if (self.cursor) |cursor| { const latest = ctx.stats.relay_seq.load(.acquire);
// FutureCursor: cursor ahead of current seq — error, close if (latest > 0 and cursor > latest) { var buf: [256]u8 = undefined; if (encodeErrorFrame("FutureCursor", "Cursor in the future.", &buf)) |frame| { consumer.conn.writeBin(frame) catch {}; } consumer.conn.close(.{}) catch {}; return; }
// OutdatedCursor: cursor older than oldest available — info, continue const oldest = blk: { if (ctx.persist != null and ctx.db_queue != null) { const FirstSeqReq = struct { base: event_log_mod.DbRequest = .{ .callback = &execute }, result: ?u64 = null,
fn execute(b: *event_log_mod.DbRequest, dp: *event_log_mod.DiskPersist) void { const s: *@This() = @fieldParentPtr("base", b); s.result = dp.firstSeq(); } }; var first_req: FirstSeqReq = .{}; ctx.db_queue.?.push(&first_req.base); first_req.base.wait(ctx.io, ctx.shutdown); if (first_req.result) |s| break :blk s; } break :blk ctx.history.oldestSeq() orelse 0; }; if (oldest > 0 and cursor < oldest) { var buf: [256]u8 = undefined; if (encodeInfoMessage("OutdatedCursor", "Requested cursor exceeded limit. Possibly missing events", &buf)) |frame| { consumer.conn.writeBin(frame) catch {}; } // don't close — fall through to replayTo which replays from oldest available }
ctx.replayTo(consumer, cursor); } }
pub fn clientMessage(_: *Handler, _: []const u8) !void { // relay consumers are read-only }
/// handle client-initiated close frame. without this, the websocket /// library tries to write a close reply which can panic if the socket /// is already dead (unreachable in std/net.zig drain). pub fn clientClose(self: *Handler, _: []const u8) !void { if (self.consumer) |c| { c.alive.store(false, .release); c.cond.signal(c.io); } self.conn.close(.{}) catch {}; }
pub fn close(self: *Handler) void { if (self.consumer) |c| self.broadcaster.removeConsumer(c); }
/// called by websocket server for non-WebSocket HTTP requests pub fn httpFallback( conn: *websocket.Conn, method: []const u8, url: []const u8, body: []const u8, headers: *const websocket.Handshake.KeyValue, ctx: *Broadcaster, ) void { if (ctx.http_fallback) |f| { f(conn, method, url, body, headers, ctx.http_fallback_ctx); } }};
pub const AttributionMetrics = struct { history_entries: usize = 0, evtbuf_entries: usize = 0, did_cache_entries: usize = 0, host_authority_cache_entries: usize = 0, resolve_queue_len: usize = 0, resolve_queued_set_count: usize = 0, // memory attribution — internal capacities of key data structures validator_cache_map_cap: usize = 0, did_cache_map_cap: usize = 0, queued_set_map_cap: usize = 0, evtbuf_cap: usize = 0, outbuf_cap: usize = 0, workers_count: usize = 0, // db_queue liveness — handled stops advancing while depth > 0 = workers wedged db_queue_depth: u32 = 0, db_queue_handled_total: u64 = 0,};
pub fn formatPrometheusMetrics(stats: *const Stats, cache_entries: usize, attribution: AttributionMetrics, data_dir: []const u8, buf: []u8, io: Io) []const u8 { const uptime: i64 = timestamp(io) - stats.start_time; var w: Io.Writer = .fixed(buf);
// canary metric — proves what binary is running w.print( \\# TYPE relay_build_info gauge \\# HELP relay_build_info build provenance \\relay_build_info{{git_sha="{s}",optimize="{s}"}} 1 \\ , .{ build_options.git_sha, build_options.optimize }) catch {};
w.print( \\# TYPE relay_frames_received_total counter \\relay_frames_received_total {d} \\ \\# TYPE relay_frames_broadcast_total counter \\relay_frames_broadcast_total {d} \\ \\# TYPE relay_consumers_active gauge \\relay_consumers_active {d} \\ \\# TYPE relay_validation_total counter \\relay_validation_total{{result="validated"}} {d} \\relay_validation_total{{result="failed"}} {d} \\relay_validation_total{{result="skipped"}} {d} \\ \\# TYPE relay_rate_limited_total counter \\relay_rate_limited_total {d} \\ \\# TYPE relay_rate_limited_by_tier counter \\relay_rate_limited_by_tier{{tier="sec"}} {d} \\relay_rate_limited_by_tier{{tier="hour"}} {d} \\relay_rate_limited_by_tier{{tier="day"}} {d} \\ \\# TYPE relay_pool_backpressure_total counter \\# HELP relay_pool_backpressure_total times pool submit blocked on full queue \\relay_pool_backpressure_total {d} \\ \\# TYPE relay_decode_errors_total counter \\relay_decode_errors_total {d} \\ \\# TYPE relay_cache_total counter \\relay_cache_total{{result="hit"}} {d} \\relay_cache_total{{result="miss"}} {d} \\ \\# TYPE relay_slow_consumers_total counter \\relay_slow_consumers_total {d} \\ \\# TYPE relay_connected_inbound gauge \\relay_connected_inbound {d} \\ \\# TYPE relay_upstream_seq gauge \\relay_upstream_seq {d} \\ \\# TYPE relay_seq gauge \\relay_seq {d} \\ \\# TYPE relay_uptime_seconds gauge \\relay_uptime_seconds {d} \\ \\# TYPE relay_validator_cache_entries gauge \\relay_validator_cache_entries {d} \\ \\# TYPE relay_validator_cache_evictions_total counter \\relay_validator_cache_evictions_total {d} \\ \\# TYPE relay_history_entries gauge \\# HELP relay_history_entries in-memory frame history ring buffer entries \\relay_history_entries {d} \\ \\# TYPE relay_evtbuf_entries gauge \\# HELP relay_evtbuf_entries pending flush buffer entries \\relay_evtbuf_entries {d} \\ \\# TYPE relay_did_cache_entries gauge \\# HELP relay_did_cache_entries DID-to-UID cache entries \\relay_did_cache_entries {d} \\ \\# TYPE relay_resolve_queue_len gauge \\# HELP relay_resolve_queue_len DID resolve queue length \\relay_resolve_queue_len {d} \\ \\# TYPE relay_resolve_queued_set_count gauge \\# HELP relay_resolve_queued_set_count DID resolve dedup set size \\relay_resolve_queued_set_count {d} \\ \\# TYPE relay_host_authority_checks_total counter \\# HELP relay_host_authority_checks_total resolveHostAuthority calls on frame workers \\relay_host_authority_checks_total {d} \\ \\# TYPE relay_host_authority_trigger counter \\# HELP relay_host_authority_trigger host authority check triggers by reason \\relay_host_authority_trigger{{reason="is_new"}} {d} \\relay_host_authority_trigger{{reason="host_changed"}} {d} \\ \\# TYPE relay_host_authority_time_us_total counter \\# HELP relay_host_authority_time_us_total cumulative microseconds in resolveHostAuthority \\relay_host_authority_time_us_total {d} \\ \\# TYPE relay_pool_queued_bytes gauge \\# HELP relay_pool_queued_bytes estimated bytes of frame data queued in worker pool \\relay_pool_queued_bytes {d} \\ , .{ stats.frames_in.load(.acquire), stats.frames_out.load(.acquire), stats.consumer_count.load(.acquire), stats.validated.load(.acquire), stats.failed.load(.acquire), stats.skipped.load(.acquire), stats.rate_limited.load(.acquire), stats.rate_limited_sec.load(.acquire), stats.rate_limited_hour.load(.acquire), stats.rate_limited_day.load(.acquire), stats.pool_backpressure.load(.acquire), stats.decode_errors.load(.acquire), stats.cache_hits.load(.acquire), stats.cache_misses.load(.acquire), stats.slow_consumers.load(.acquire), stats.connected_inbound.load(.acquire), stats.seq.load(.acquire), stats.relay_seq.load(.acquire), uptime, cache_entries, stats.cache_evictions.load(.acquire), attribution.history_entries, attribution.evtbuf_entries, attribution.did_cache_entries, attribution.resolve_queue_len, attribution.resolve_queued_set_count, stats.host_authority_checks.load(.acquire), stats.host_authority_is_new.load(.acquire), stats.host_authority_host_changed.load(.acquire), stats.host_authority_time_us.load(.acquire), stats.pool_queued_bytes.load(.acquire), }) catch return w.buffered();
// keepalive pinger metrics (separate print to stay under 32-arg limit) w.print( \\# TYPE relay_ws_pings_sent_total counter \\# HELP relay_ws_pings_sent_total keepalive pings sent to silent upstream connections \\relay_ws_pings_sent_total {d} \\ \\# TYPE relay_ws_ping_timeout_closes_total counter \\# HELP relay_ws_ping_timeout_closes_total connections closed as half-open after max unanswered keepalive pings \\relay_ws_ping_timeout_closes_total {d} \\ \\# TYPE relay_ws_ping_write_failures_total counter \\# HELP relay_ws_ping_write_failures_total keepalive ping writes that failed (connection torn down) \\relay_ws_ping_write_failures_total {d} \\ , .{ stats.ws_pings_sent.load(.acquire), stats.ws_ping_timeout_closes.load(.acquire), stats.ws_ping_write_failures.load(.acquire), }) catch return w.buffered();
w.print( \\# TYPE relay_host_authority_cache_hits_total counter \\# HELP relay_host_authority_cache_hits_total confirmed host mismatches rejected without network resolution \\relay_host_authority_cache_hits_total {d} \\ \\# TYPE relay_host_authority_network_checks_total counter \\# HELP relay_host_authority_network_checks_total host authority decisions that performed network resolution \\relay_host_authority_network_checks_total {d} \\ \\# TYPE relay_host_authority_network_time_us_total counter \\# HELP relay_host_authority_network_time_us_total worker microseconds spent waiting on host authority network decisions \\relay_host_authority_network_time_us_total {d} \\ \\# TYPE relay_host_authority_pool_wait_us_total counter \\# HELP relay_host_authority_pool_wait_us_total worker microseconds spent waiting for a host resolver pool slot \\relay_host_authority_pool_wait_us_total {d} \\ \\# TYPE relay_host_authority_cache_entries gauge \\# HELP relay_host_authority_cache_entries confirmed DID and incoming host mismatches cached \\relay_host_authority_cache_entries {d} \\ , .{ stats.host_authority_cache_hits.load(.acquire), stats.host_authority_network_checks.load(.acquire), stats.host_authority_network_time_us.load(.acquire), stats.host_authority_pool_wait_us.load(.acquire), attribution.host_authority_cache_entries, }) catch return w.buffered();
// pipeline contention metrics (separate print to stay under 32-arg limit) w.print( \\# TYPE relay_chain_breaks_total counter \\# HELP relay_chain_breaks_total chain continuity failures by reason \\relay_chain_breaks_total{{reason="since"}} {d} \\relay_chain_breaks_total{{reason="prev_data"}} {d} \\ \\# TYPE relay_persist_order_spins_total counter \\# HELP relay_persist_order_spins_total spin iterations waiting for persist ordering lock \\relay_persist_order_spins_total {d} \\ \\# TYPE relay_broadcast_queue_push_lock_spins_total counter \\# HELP relay_broadcast_queue_push_lock_spins_total spin iterations waiting for broadcast queue push lock \\relay_broadcast_queue_push_lock_spins_total {d} \\ \\# TYPE relay_broadcast_queue_full_total counter \\# HELP relay_broadcast_queue_full_total spin iterations on full broadcast queue (ring capacity) \\relay_broadcast_queue_full_total {d} \\ \\# TYPE relay_broadcast_queue_depth_hwm gauge \\# HELP relay_broadcast_queue_depth_hwm high-water mark of broadcast queue depth \\relay_broadcast_queue_depth_hwm {d} \\ \\# TYPE relay_broadcast_no_consumers_total counter \\# HELP relay_broadcast_no_consumers_total frames skipped broadcast (no consumers) \\relay_broadcast_no_consumers_total {d} \\ , .{ stats.chain_breaks_since.load(.acquire), stats.chain_breaks_prev_data.load(.acquire), stats.persist_order_spins.load(.acquire), stats.broadcast_queue_push_lock_spins.load(.acquire), stats.broadcast_queue_full.load(.acquire), stats.broadcast_queue_depth_hwm.load(.acquire), stats.broadcast_no_consumers.load(.acquire), }) catch {};
// validation failure breakdown by reason w.print( \\# TYPE relay_validation_failed counter \\# HELP relay_validation_failed validation failures by reason \\relay_validation_failed{{reason="bad_did"}} {d} \\relay_validation_failed{{reason="bad_rev"}} {d} \\relay_validation_failed{{reason="missing_blocks"}} {d} \\relay_validation_failed{{reason="oversized_blocks"}} {d} \\relay_validation_failed{{reason="bad_structure"}} {d} \\relay_validation_failed{{reason="signature"}} {d} \\relay_validation_failed{{reason="commit_integrity"}} {d} \\relay_validation_failed{{reason="sync_1_1"}} {d} \\relay_validation_failed{{reason="prev_data_missing"}} {d} \\relay_validation_failed{{reason="host_authority"}} {d} \\relay_validation_failed{{reason="future_rev"}} {d} \\ \\# TYPE relay_host_authority_reject counter \\# HELP relay_host_authority_reject host authority reject breakdown by branch \\relay_host_authority_reject{{branch="parse_did"}} {d} \\relay_host_authority_reject{{branch="resolve"}} {d} \\relay_host_authority_reject{{branch="no_endpoint"}} {d} \\relay_host_authority_reject{{branch="bad_url"}} {d} \\relay_host_authority_reject{{branch="unknown_host"}} {d} \\relay_host_authority_reject{{branch="host_mismatch"}} {d} \\ \\# TYPE relay_subscriber_disconnect_total counter \\# HELP relay_subscriber_disconnect_total subscriber reconnect-cycle disconnects by phase \\relay_subscriber_disconnect_total{{phase="dns_connect"}} {d} \\relay_subscriber_disconnect_total{{phase="tls_init"}} {d} \\relay_subscriber_disconnect_total{{phase="ws_handshake"}} {d} \\relay_subscriber_disconnect_total{{phase="read_loop"}} {d} \\ , .{ stats.failed_bad_did.load(.acquire), stats.failed_bad_rev.load(.acquire), stats.failed_missing_blocks.load(.acquire), stats.failed_oversized_blocks.load(.acquire), stats.failed_bad_structure.load(.acquire), stats.failed_signature.load(.acquire), stats.failed_commit_integrity.load(.acquire), stats.failed_sync_1_1.load(.acquire), stats.failed_prev_data_missing.load(.acquire), stats.failed_host_authority.load(.acquire), stats.failed_future_rev.load(.acquire), stats.host_authority_reject_parse_did.load(.acquire), stats.host_authority_reject_resolve.load(.acquire), stats.host_authority_reject_no_endpoint.load(.acquire), stats.host_authority_reject_bad_url.load(.acquire), stats.host_authority_reject_unknown_host.load(.acquire), stats.host_authority_reject_host_mismatch.load(.acquire), stats.subscriber_disconnect_dns_connect.load(.acquire), stats.subscriber_disconnect_tls_init.load(.acquire), stats.subscriber_disconnect_ws_handshake.load(.acquire), stats.subscriber_disconnect_read_loop.load(.acquire), }) catch return w.buffered();
// memory attribution — internal capacities help identify what's consuming RSS w.print( \\# TYPE relay_validator_cache_map_cap gauge \\# HELP relay_validator_cache_map_cap hashmap backing capacity of signing key LRU \\relay_validator_cache_map_cap {d} \\ \\# TYPE relay_did_cache_map_cap gauge \\# HELP relay_did_cache_map_cap hashmap backing capacity of DID-to-UID LRU \\relay_did_cache_map_cap {d} \\ \\# TYPE relay_queued_set_map_cap gauge \\# HELP relay_queued_set_map_cap hashmap backing capacity of resolver dedup set \\relay_queued_set_map_cap {d} \\ \\# TYPE relay_evtbuf_cap gauge \\# HELP relay_evtbuf_cap allocated capacity of event flush buffer (jobs) \\relay_evtbuf_cap {d} \\ \\# TYPE relay_outbuf_cap gauge \\# HELP relay_outbuf_cap allocated capacity of output byte buffer \\relay_outbuf_cap {d} \\ \\# TYPE relay_workers_count gauge \\# HELP relay_workers_count active subscriber worker threads \\relay_workers_count {d} \\ \\# TYPE relay_db_queue_depth gauge \\# HELP relay_db_queue_depth current depth of the DbRequestQueue (tail - head). non-zero with stagnant relay_db_queue_handled_total = workers wedged \\relay_db_queue_depth {d} \\ \\# TYPE relay_db_queue_handled_total counter \\# HELP relay_db_queue_handled_total total DbRequest callbacks executed across worker threads \\relay_db_queue_handled_total {d} \\ , .{ attribution.validator_cache_map_cap, attribution.did_cache_map_cap, attribution.queued_set_map_cap, attribution.evtbuf_cap, attribution.outbuf_cap, attribution.workers_count, attribution.db_queue_depth, attribution.db_queue_handled_total, }) catch {};
// linux-only process metrics from /proc if (comptime builtin.os.tag == .linux) { appendProcMetrics(&w, io); appendDiskMetrics(&w, data_dir); }
return w.buffered();}
fn appendProcMetrics(w: *Io.Writer, io: Io) void { // RSS from /proc/self/statm (field[1] * page_size) if (Io.Dir.openFileAbsolute(io, "/proc/self/statm", .{})) |f| { defer f.close(io); var statm_buf: [256]u8 = undefined; const n = f.readPositionalAll(io, &statm_buf, 0) catch 0; if (n > 0) { const line = statm_buf[0..n]; var iter = std.mem.splitScalar(u8, line, ' '); _ = iter.next(); // skip total pages if (iter.next()) |rss_pages| { if (std.fmt.parseInt(u64, rss_pages, 10)) |pages| { w.print( \\# TYPE relay_process_rss_bytes gauge \\relay_process_rss_bytes {d} \\ , .{pages * 4096}) catch {}; } else |_| {} } } } else |_| {}
// thread count, VmHWM, RssAnon from /proc/self/status if (Io.Dir.openFileAbsolute(io, "/proc/self/status", .{})) |f| { defer f.close(io); var status_buf: [4096]u8 = undefined; const n = f.readPositionalAll(io, &status_buf, 0) catch 0; if (n > 0) { const content = status_buf[0..n]; const fields = .{ .{ "Threads:", "relay_threads_total" }, .{ "VmHWM:", "relay_vm_hwm_kb" }, .{ "RssAnon:", "relay_rss_anon_kb" }, }; inline for (fields) |entry| { if (std.mem.indexOf(u8, content, entry[0])) |pos| { const rest = content[pos + entry[0].len ..]; const trimmed = std.mem.trimStart(u8, rest, " \t"); const end = std.mem.indexOfScalar(u8, trimmed, ' ') orelse (std.mem.indexOfScalar(u8, trimmed, '\n') orelse trimmed.len); if (std.fmt.parseInt(u64, trimmed[0..end], 10)) |val| { w.print( \\# TYPE {s} gauge \\{s} {d} \\ , .{ entry[1], entry[1], val }) catch {}; } else |_| {} } } } } else |_| {}
// glibc malloc stats — mallinfo fields are c_int, bitcast to u32 to extend range to 4 GiB. // note: mallinfo only reports the main arena, not per-thread arenas. // for accurate total heap picture, rely on RssAnon from /proc/self/status above. const mi = mallinfo(); const arena: u64 = @as(u32, @bitCast(mi.arena)); const in_use: u64 = @as(u32, @bitCast(mi.uordblks)); const free_bytes: u64 = @as(u32, @bitCast(mi.fordblks)); const mmap_bytes: u64 = @as(u32, @bitCast(mi.hblkhd)); w.print( \\# TYPE relay_malloc_arena_bytes gauge \\relay_malloc_arena_bytes {d} \\ \\# TYPE relay_malloc_in_use_bytes gauge \\relay_malloc_in_use_bytes {d} \\ \\# TYPE relay_malloc_free_bytes gauge \\relay_malloc_free_bytes {d} \\ \\# TYPE relay_malloc_mmap_bytes gauge \\relay_malloc_mmap_bytes {d} \\ , .{ arena, in_use, free_bytes, mmap_bytes }) catch {};
// all-arena heap picture via malloc_info() (mallinfo above is main-arena only). // with ~2670 worker threads glibc spreads allocations across many per-thread // arenas; post-burst fragmentation lives there, invisible to mallinfo. these // gauges sum every arena so a rising free_total / arena_count is observable. if (readMallocInfo()) |info| { w.print( \\# TYPE relay_malloc_system_bytes gauge \\# HELP relay_malloc_system_bytes total bytes obtained from the OS across all arenas \\relay_malloc_system_bytes {d} \\ \\# TYPE relay_malloc_free_total_bytes gauge \\# HELP relay_malloc_free_total_bytes free bytes held in all arenas (fast+rest), not returned to OS \\relay_malloc_free_total_bytes {d} \\ \\# TYPE relay_malloc_arena_count gauge \\# HELP relay_malloc_arena_count number of glibc malloc arenas (tune via MALLOC_ARENA_MAX) \\relay_malloc_arena_count {d} \\ , .{ info.system_bytes, info.free_bytes, info.arena_count }) catch {}; }}
// glibc mallinfo struct — fields are c_int (32-bit), we bitcast to u32 for range.const MallInfo = extern struct { arena: c_int, ordblks: c_int, smblks: c_int, hblks: c_int, hblkhd: c_int, usmblks: c_int, fsmblks: c_int, uordblks: c_int, fordblks: c_int, keepcost: c_int,};
const mallinfo = @extern(*const fn () callconv(.c) MallInfo, .{ .name = "mallinfo" });
// all-arena heap stats, parsed from glibc malloc_info() XML. unlike mallinfo()// these cover every per-thread arena, not just the main one.const MallocInfo = struct { system_bytes: u64, free_bytes: u64, arena_count: u32,};
const CFile = opaque {};const open_memstream = @extern(*const fn (ptr: *[*c]u8, size: *usize) callconv(.c) ?*CFile, .{ .name = "open_memstream" });const fclose = @extern(*const fn (*CFile) callconv(.c) c_int, .{ .name = "fclose" });const c_free = @extern(*const fn (?*anyopaque) callconv(.c) void, .{ .name = "free" });const malloc_info_fn = @extern(*const fn (options: c_int, stream: *CFile) callconv(.c) c_int, .{ .name = "malloc_info" });
fn readMallocInfo() ?MallocInfo { // comptime if prunes the linux-only externs from analysis on other hosts, // so `zig build test` on macOS doesn't try to link malloc_info/open_memstream. return if (comptime builtin.os.tag == .linux) readMallocInfoLinux() else null;}
fn readMallocInfoLinux() ?MallocInfo { var buf_ptr: [*c]u8 = null; var buf_size: usize = 0; const stream = open_memstream(&buf_ptr, &buf_size) orelse return null; _ = malloc_info_fn(0, stream); _ = fclose(stream); // flushes and populates buf_ptr/buf_size const p = buf_ptr orelse return null; defer c_free(p); const xml = p[0..buf_size]; return MallocInfo{ // top-level totals are the LAST occurrence (each <heap> emits its own). .system_bytes = lastSizeAttr(xml, "system type=\"current\"") orelse 0, .free_bytes = (lastSizeAttr(xml, "total type=\"fast\"") orelse 0) + (lastSizeAttr(xml, "total type=\"rest\"") orelse 0), .arena_count = countOccurrences(xml, "<heap nr="), };}
// find the last occurrence of `marker`, then parse the first `size="N"` after it.fn lastSizeAttr(xml: []const u8, marker: []const u8) ?u64 { const at = std.mem.lastIndexOf(u8, xml, marker) orelse return null; const rest = xml[at + marker.len ..]; const key = "size=\""; const kpos = std.mem.indexOf(u8, rest, key) orelse return null; const num = rest[kpos + key.len ..]; const end = std.mem.indexOfScalar(u8, num, '"') orelse return null; return std.fmt.parseInt(u64, num[0..end], 10) catch null;}
fn countOccurrences(haystack: []const u8, needle: []const u8) u32 { var n: u32 = 0; var i: usize = 0; while (std.mem.indexOfPos(u8, haystack, i, needle)) |pos| { n += 1; i = pos + needle.len; } return n;}
test "readMallocInfo parses all-arena totals" { if (comptime builtin.os.tag != .linux) return error.SkipZigTest; // force some allocation so the arena is non-trivial const a = std.heap.c_allocator; var kept: [64][]u8 = undefined; for (&kept) |*slot| slot.* = try a.alloc(u8, 4096); defer for (kept) |s| a.free(s);
const info = readMallocInfo() orelse return error.MallocInfoUnavailable; try std.testing.expect(info.system_bytes > 0); try std.testing.expect(info.arena_count >= 1);}
test "lastSizeAttr / countOccurrences against sample XML" { const xml = \\<malloc version="1"> \\<heap nr="0"> \\<total type="fast" count="1" size="100"/> \\<total type="rest" count="2" size="200"/> \\<system type="current" size="9999"/> \\</heap> \\<heap nr="1"> \\<total type="fast" count="3" size="111"/> \\<total type="rest" count="4" size="222"/> \\<system type="current" size="8888"/> \\</heap> \\<total type="fast" count="4" size="211"/> \\<total type="rest" count="6" size="422"/> \\<system type="current" size="18887"/> \\</malloc> ; try std.testing.expectEqual(@as(?u64, 18887), lastSizeAttr(xml, "system type=\"current\"")); try std.testing.expectEqual(@as(?u64, 211), lastSizeAttr(xml, "total type=\"fast\"")); try std.testing.expectEqual(@as(?u64, 422), lastSizeAttr(xml, "total type=\"rest\"")); try std.testing.expectEqual(@as(u32, 2), countOccurrences(xml, "<heap nr="));}
// statvfs — only the fields we use (f_frsize, f_blocks, f_bavail)const StatVfs = extern struct { f_bsize: c_ulong, f_frsize: c_ulong, f_blocks: c_ulong, f_bfree: c_ulong, f_bavail: c_ulong, f_files: c_ulong, f_ffree: c_ulong, f_favail: c_ulong, f_fsid: c_ulong, f_flag: c_ulong, f_namemax: c_ulong,};
const statvfs_fn = @extern(*const fn ([*:0]const u8, *StatVfs) callconv(.c) c_int, .{ .name = "statvfs" });
fn appendDiskMetrics(w: *Io.Writer, data_dir: []const u8) void { // statvfs needs a null-terminated path var path_buf: [4096]u8 = undefined; if (data_dir.len >= path_buf.len) return; @memcpy(path_buf[0..data_dir.len], data_dir); path_buf[data_dir.len] = 0;
var stat: StatVfs = undefined; if (statvfs_fn(@ptrCast(&path_buf), &stat) != 0) return;
const block_size: u64 = stat.f_frsize; const total = stat.f_blocks * block_size; const available = stat.f_bavail * block_size;
w.print( \\# TYPE relay_disk_total_bytes gauge \\relay_disk_total_bytes {d} \\ \\# TYPE relay_disk_available_bytes gauge \\relay_disk_available_bytes {d} \\ , .{ total, available }) catch {};}
pub fn formatStatsResponse(stats: *const Stats, buf: []u8, io: Io) []const u8 { return std.fmt.bufPrint(buf, \\{{"seq":{d},"relay_seq":{d},"consumers":{d},"connected_inbound":{d},"frames_in":{d},"frames_out":{d},"validated":{d},"failed":{d},"skipped":{d},"decode_errors":{d},"cache_hits":{d},"cache_misses":{d},"slow_consumers":{d},"chain_breaks":{d},"pool_backpressure":{d},"uptime_seconds":{d}}} , .{ stats.seq.load(.acquire), stats.relay_seq.load(.acquire), stats.consumer_count.load(.acquire), stats.connected_inbound.load(.acquire), stats.frames_in.load(.acquire), stats.frames_out.load(.acquire), stats.validated.load(.acquire), stats.failed.load(.acquire), stats.skipped.load(.acquire), stats.decode_errors.load(.acquire), stats.cache_hits.load(.acquire), stats.cache_misses.load(.acquire), stats.slow_consumers.load(.acquire), stats.chain_breaks_since.load(.acquire) + stats.chain_breaks_prev_data.load(.acquire), stats.pool_backpressure.load(.acquire), timestamp(io) - stats.start_time, }) catch "";}
// --- tests ---
var test_shutdown: std.atomic.Value(bool) = .{ .raw = false };
test "broadcaster add and remove consumer" { var b = Broadcaster.init(std.testing.allocator, std.testing.io, &test_shutdown); defer b.deinit();
try std.testing.expectEqual(@as(u64, 0), b.stats.seq.load(.acquire)); try std.testing.expectEqual(@as(usize, 0), b.consumerCount());}
test "broadcast updates stats and history" { var b = Broadcaster.init(std.testing.allocator, std.testing.io, &test_shutdown); defer b.deinit();
b.broadcast(1, "frame1"); b.broadcast(2, "frame2"); b.broadcast(3, "frame3");
try std.testing.expectEqual(@as(u64, 3), b.stats.seq.load(.acquire)); // frames_in is now incremented in subscriber, not broadcast try std.testing.expectEqual(@as(u64, 0), b.stats.frames_in.load(.acquire)); try std.testing.expectEqual(@as(u64, 1), b.history.oldestSeq().?); try std.testing.expectEqual(@as(u64, 3), b.history.newestSeq().?);}
test "frame history supports cursor replay" { var b = Broadcaster.init(std.testing.allocator, std.testing.io, &test_shutdown); defer b.deinit();
for (1..6) |i| { var buf: [32]u8 = undefined; const data = std.fmt.bufPrint(&buf, "frame{d}", .{i}) catch unreachable; b.broadcast(@intCast(i), data); }
const frames = try b.history.framesSince(std.testing.allocator, 3); defer { for (frames) |f| std.testing.allocator.free(f.data); std.testing.allocator.free(frames); }
try std.testing.expectEqual(@as(usize, 2), frames.len); try std.testing.expectEqual(@as(u64, 4), frames[0].seq); try std.testing.expectEqual(@as(u64, 5), frames[1].seq);}
test "enqueue refuses frames for a dead consumer" { // regression: replay for an abruptly-disconnected consumer kept // accumulating frames, compounding into the 2026-08-06 canary OOM. var consumer = Consumer{ .conn = undefined, // enqueue never touches the connection .allocator = std.testing.allocator, .io = std.testing.io, };
try std.testing.expect(consumer.enqueueRaw("frame1"));
consumer.alive.store(false, .release); try std.testing.expect(!consumer.enqueueRaw("frame2"));
// drain the accepted frame consumer.mutex.lockUncancelable(consumer.io); while (consumer.dequeue()) |f| f.release(); consumer.mutex.unlock(consumer.io);}
test "shared frame ref counting" { const frame = try SharedFrame.create(std.testing.allocator, "hello"); frame.acquire(); // ref=2 frame.acquire(); // ref=3 frame.release(); // ref=2 frame.release(); // ref=1 frame.release(); // ref=0 → freed}
test "error frame is valid CBOR" { const cbor = zat.cbor; var arena = std.heap.ArenaAllocator.init(std.testing.allocator); defer arena.deinit(); const alloc = arena.allocator();
const frame = buildErrorFrame(alloc) orelse return error.FrameBuildFailed;
// decode header const h_result = try cbor.decode(alloc, frame); const h = h_result.value; try std.testing.expectEqual(@as(i64, -1), h.getInt("op").?); // spec: error frames have {op: -1} only, no `t` field try std.testing.expect(h.getString("t") == null);
// decode payload const p = try cbor.decodeAll(alloc, frame[h_result.consumed..]); try std.testing.expectEqualStrings("ConsumerTooSlow", p.getString("error").?); try std.testing.expectEqualStrings("consumer buffer full", p.getString("message").?);}
test "formatPrometheusMetrics produces valid output" { var stats = Stats{ .start_time = timestamp(std.testing.io) - 60 }; stats.seq.store(99999, .release); stats.relay_seq.store(12345, .release); stats.consumer_count.store(3, .release); stats.frames_in.store(10000, .release); stats.frames_out.store(9000, .release); stats.validated.store(500, .release); stats.failed.store(2, .release); stats.skipped.store(9498, .release); stats.cache_hits.store(400, .release); stats.cache_misses.store(100, .release); stats.host_authority_cache_hits.store(321, .release); stats.host_authority_network_checks.store(12, .release); stats.host_authority_network_time_us.store(456_000, .release); stats.host_authority_pool_wait_us.store(789_000, .release);
var buf: [65536]u8 = undefined; const output = formatPrometheusMetrics(&stats, 42, .{ .host_authority_cache_entries = 9, }, "/tmp", &buf, std.testing.io);
try std.testing.expect(std.mem.indexOf(u8, output, "relay_frames_received_total 10000") != null); try std.testing.expect(std.mem.indexOf(u8, output, "relay_frames_broadcast_total 9000") != null); try std.testing.expect(std.mem.indexOf(u8, output, "relay_consumers_active 3") != null); try std.testing.expect(std.mem.indexOf(u8, output, "relay_validation_total{result=\"validated\"} 500") != null); try std.testing.expect(std.mem.indexOf(u8, output, "relay_validation_total{result=\"failed\"} 2") != null); try std.testing.expect(std.mem.indexOf(u8, output, "relay_connected_inbound 0") != null); try std.testing.expect(std.mem.indexOf(u8, output, "relay_upstream_seq 99999") != null); try std.testing.expect(std.mem.indexOf(u8, output, "relay_seq 12345") != null); try std.testing.expect(std.mem.indexOf(u8, output, "# TYPE relay_uptime_seconds gauge") != null); try std.testing.expect(std.mem.indexOf(u8, output, "relay_validator_cache_entries 42") != null); try std.testing.expect(std.mem.indexOf(u8, output, "relay_validator_cache_evictions_total 0") != null); try std.testing.expect(std.mem.indexOf(u8, output, "relay_host_authority_cache_hits_total 321") != null); try std.testing.expect(std.mem.indexOf(u8, output, "relay_host_authority_network_checks_total 12") != null); try std.testing.expect(std.mem.indexOf(u8, output, "relay_host_authority_network_time_us_total 456000") != null); try std.testing.expect(std.mem.indexOf(u8, output, "relay_host_authority_pool_wait_us_total 789000") != null); try std.testing.expect(std.mem.indexOf(u8, output, "relay_host_authority_cache_entries 9") != null);}
test "ws ping counters render on their own lines with no cross-wiring" { var stats = Stats{ .start_time = timestamp(std.testing.io) }; stats.ws_pings_sent.store(111_111, .release); stats.ws_ping_timeout_closes.store(222, .release); stats.ws_ping_write_failures.store(3, .release);
var buf: [65536]u8 = undefined; const output = formatPrometheusMetrics(&stats, 0, .{ .host_authority_cache_entries = 0, }, "/tmp", &buf, std.testing.io);
try std.testing.expect(std.mem.indexOf(u8, output, "relay_ws_pings_sent_total 111111\n") != null); try std.testing.expect(std.mem.indexOf(u8, output, "relay_ws_ping_timeout_closes_total 222\n") != null); try std.testing.expect(std.mem.indexOf(u8, output, "relay_ws_ping_write_failures_total 3\n") != null);}
test "formatStatsResponse produces valid JSON" { var stats = Stats{ .start_time = timestamp(std.testing.io) - 42 }; stats.seq.store(100, .release); stats.frames_in.store(200, .release); stats.consumer_count.store(5, .release);
var buf: [4096]u8 = undefined; const response = formatStatsResponse(&stats, &buf, std.testing.io);
try std.testing.expect(std.mem.startsWith(u8, response, "{")); try std.testing.expect(std.mem.indexOf(u8, response, "\"seq\":100") != null); try std.testing.expect(std.mem.indexOf(u8, response, "\"consumers\":5") != null); try std.testing.expect(std.mem.indexOf(u8, response, "\"frames_in\":200") != null); try std.testing.expect(std.mem.indexOf(u8, response, "\"slow_consumers\":0") != null);}
test "resequenceFrame replaces seq in payload" { const cbor = zat.cbor; var arena = std.heap.ArenaAllocator.init(std.testing.allocator); defer arena.deinit(); const alloc = arena.allocator();
// build a frame with upstream seq=12345 const header: cbor.Value = .{ .map = &.{ .{ .key = "op", .value = .{ .unsigned = 1 } }, .{ .key = "t", .value = .{ .text = "#commit" } }, } }; const payload: cbor.Value = .{ .map = &.{ .{ .key = "repo", .value = .{ .text = "did:plc:test123" } }, .{ .key = "seq", .value = .{ .unsigned = 12345 } }, .{ .key = "rev", .value = .{ .text = "3k2abc000000" } }, } };
const header_bytes = try cbor.encodeAlloc(alloc, header); const payload_bytes = try cbor.encodeAlloc(alloc, payload); const frame = try alloc.alloc(u8, header_bytes.len + payload_bytes.len); @memcpy(frame[0..header_bytes.len], header_bytes); @memcpy(frame[header_bytes.len..], payload_bytes);
// resequence to relay seq 99999 const result = resequenceFrame(alloc, frame, 99999) orelse return error.ResequenceFailed;
// decode result and verify const h_result = try cbor.decode(alloc, result); const p = try cbor.decodeAll(alloc, result[h_result.consumed..]);
// header preserved try std.testing.expectEqualStrings("#commit", h_result.value.getString("t").?); try std.testing.expectEqual(@as(i64, 1), h_result.value.getInt("op").?); // seq replaced with relay seq try std.testing.expectEqual(@as(u64, 99999), p.getUint("seq").?); // other fields preserved try std.testing.expectEqualStrings("did:plc:test123", p.getString("repo").?); try std.testing.expectEqualStrings("3k2abc000000", p.getString("rev").?); // tooBig injected for commit frames missing it try std.testing.expectEqual(false, p.getBool("tooBig").?);}
test "resequenceFrame preserves identity frame fields" { const cbor = zat.cbor; var arena = std.heap.ArenaAllocator.init(std.testing.allocator); defer arena.deinit(); const alloc = arena.allocator();
const header: cbor.Value = .{ .map = &.{ .{ .key = "op", .value = .{ .unsigned = 1 } }, .{ .key = "t", .value = .{ .text = "#identity" } }, } }; const payload: cbor.Value = .{ .map = &.{ .{ .key = "did", .value = .{ .text = "did:plc:alice" } }, .{ .key = "seq", .value = .{ .unsigned = 500 } }, .{ .key = "time", .value = .{ .text = "2024-01-15T10:30:00Z" } }, } };
const header_bytes = try cbor.encodeAlloc(alloc, header); const payload_bytes = try cbor.encodeAlloc(alloc, payload); const frame = try alloc.alloc(u8, header_bytes.len + payload_bytes.len); @memcpy(frame[0..header_bytes.len], header_bytes); @memcpy(frame[header_bytes.len..], payload_bytes);
const result = resequenceFrame(alloc, frame, 42) orelse return error.ResequenceFailed;
const h_result = try cbor.decode(alloc, result); const p = try cbor.decodeAll(alloc, result[h_result.consumed..]);
try std.testing.expectEqualStrings("#identity", h_result.value.getString("t").?); try std.testing.expectEqual(@as(u64, 42), p.getUint("seq").?); try std.testing.expectEqualStrings("did:plc:alice", p.getString("did").?); try std.testing.expectEqualStrings("2024-01-15T10:30:00Z", p.getString("time").?); // tooBig NOT injected for non-commit frames try std.testing.expectEqual(@as(?bool, null), p.getBool("tooBig"));}
test "encodeErrorFrame produces valid error frame CBOR" { const cbor = zat.cbor; var arena = std.heap.ArenaAllocator.init(std.testing.allocator); defer arena.deinit(); const alloc = arena.allocator();
var buf: [256]u8 = undefined; const frame = encodeErrorFrame("FutureCursor", "Cursor in the future.", &buf) orelse return error.EncodeFailed;
// decode header — must be {op: -1} const h_result = try cbor.decode(alloc, frame); const h = h_result.value; try std.testing.expectEqual(@as(i64, -1), h.getInt("op").?); try std.testing.expect(h.getString("t") == null); // no t field on error frames
// decode payload — must have error and message fields const p = try cbor.decodeAll(alloc, frame[h_result.consumed..]); try std.testing.expectEqualStrings("FutureCursor", p.getString("error").?); try std.testing.expectEqualStrings("Cursor in the future.", p.getString("message").?);}
test "encodeInfoMessage produces valid #info message frame CBOR" { const cbor = zat.cbor; var arena = std.heap.ArenaAllocator.init(std.testing.allocator); defer arena.deinit(); const alloc = arena.allocator();
var buf: [256]u8 = undefined; const frame = encodeInfoMessage("OutdatedCursor", "Requested cursor exceeded limit. Possibly missing events", &buf) orelse return error.EncodeFailed;
// decode header — must be {op: 1, t: "#info"} const h_result = try cbor.decode(alloc, frame); const h = h_result.value; try std.testing.expectEqual(@as(i64, 1), h.getInt("op").?); try std.testing.expectEqualStrings("#info", h.getString("t").?);
// decode payload — must have name and message fields (not error) const p = try cbor.decodeAll(alloc, frame[h_result.consumed..]); try std.testing.expectEqualStrings("OutdatedCursor", p.getString("name").?); try std.testing.expectEqualStrings("Requested cursor exceeded limit. Possibly missing events", p.getString("message").?); // info messages must NOT have an error field try std.testing.expect(p.getString("error") == null);}
test "concurrent broadcast queue + drain produces monotonic sequences" { // regression test: multiple worker threads push to the broadcast queue // under persist_order lock, then the broadcaster fiber drains and broadcasts. // the ring buffer history must be strictly monotonic.
var bc = Broadcaster.init(std.testing.allocator, std.testing.io, &test_shutdown); defer bc.deinit();
const num_threads = 8; const frames_per_thread = 500; const total_frames = num_threads * frames_per_thread; var seq_counter = std.atomic.Value(u64).init(0);
const Worker = struct { fn run(bc_ptr: *Broadcaster, counter: *std.atomic.Value(u64)) void { for (0..frames_per_thread) |_| { while (bc_ptr.persist_order.cmpxchgWeak(0, 1, .acquire, .monotonic) != null) { std.atomic.spinLoopHint(); } const seq = counter.fetchAdd(1, .monotonic) + 1; const data = std.testing.allocator.dupe(u8, "x") catch { bc_ptr.persist_order.store(0, .release); continue; }; bc_ptr.broadcast_queue.push(seq, data, &bc_ptr.stats); bc_ptr.persist_order.store(0, .release); } } };
var threads: [num_threads]std.Thread = undefined; for (&threads) |*t| { t.* = std.Thread.spawn(.{}, Worker.run, .{ &bc, &seq_counter }) catch unreachable; } for (&threads) |*t| t.join();
// drain the queue (simulating the broadcaster fiber) while (bc.broadcast_queue.pop()) |item| { bc.broadcast(item.seq, item.data); std.testing.allocator.free(@constCast(item.data)); }
// verify: ring buffer history has strictly monotonic sequences const frames = try bc.history.framesSince(std.testing.allocator, 0); defer { for (frames) |f| std.testing.allocator.free(f.data); std.testing.allocator.free(frames); }
// history is capped at 50K, we pushed 4K — all should be present try std.testing.expectEqual(total_frames, frames.len);
var prev_seq: u64 = 0; for (frames) |f| { if (f.seq <= prev_seq) { std.debug.print("monotonicity violation: seq {d} after {d}\n", .{ f.seq, prev_seq }); return error.NonMonotonicSequence; } prev_seq = f.seq; }}
test "shutdown keeps draining until the last producer finishes" { var shutdown = std.atomic.Value(bool){ .raw = true }; var bc = Broadcaster.init(std.testing.allocator, std.testing.io, &shutdown); defer bc.deinit(); const drain = try std.Thread.spawn(.{}, Broadcaster.runBroadcastLoop, .{&bc}); defer drain.join(); defer bc.producers_done.store(true, .release); for (1..10_001) |seq| { const data = try std.testing.allocator.dupe(u8, "event"); bc.broadcast_queue.push(seq, data, &bc.stats); } bc.producers_done.store(true, .release); while (bc.stats.seq.load(.acquire) != 10_000) { try std.testing.io.sleep(Io.Duration.fromMilliseconds(1), .awake); } try std.testing.expectEqual(@as(u32, 0), bc.broadcast_queue.depth());}