Something went wrong. Try again.
atproto relay in zig zlay.waow.tech
relay zig atproto
Something went wrong. Try again.
12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182838485868788899091929394959697989910010110210310410510610710810911011111211311411511611711811912012112212312412512612712812913013113213313413513613713813914014114214314414514614714814915015115215315415515615715815916016116216316416516616716816917017117217317417517617717817918018118218318418518618718818919019119219319419519619719819920020120220320420520620720820921021121221321421521621721821922022122222322422522622722822923023123223323423523623723823924024124224324424524624724824925025125225325425525625725825926026126226326426526626726826927027127227327427527627727827928028128228328428528628728828929029129229329429529629729829930030130230330430530630730830931031131231331431531631731831932032132232332432532632732832933033133233333433533633733833934034134234334434534634734834935035135235335435535635735835936036136236336436536636736836937037137237337437537637737837938038138238338438538638738838939039139239339439539639739839940040140240340440540640740840941041141241341441541641741841942042142242342442542642742842943043143243343443543643743843944044144244344444544644744844945045145245345445545645745845946046146246346446546646746846947047147247347447547647747847948048148248348448548648748848949049149249349449549649749849950050150250350450550650750850951051151251351451551651751851952052152252352452552652752852953053153253353453553653753853954054154254354454554654754854955055155255355455555655755855956056156256356456556656756856957057157257357457557657757857958058158258358458558658758858959059159259359459559659759859960060160260360460560660760860961061161261361461561661761861962062162262362462562662762862963063163263363463563663763863964064164264364464564664764864965065165265365465565665765865966066166266366466566666766866967067167267367467567667767867968068168268368468568668768868969069169269369469569669769869970070170270370470570670770870971071171271371471571671771871972072172272372472572672772872973073173273373473573673773873974074174274374474574674774874975075175275375475575675775875976076176276376476576676776876977077177277377477577677777877978078178278378478578678778878979079179279379479579679779879980080180280380480580680780880981081181281381481581681781881982082182282382482582682782882983083183283383483583683783883984084184284384484584684784884985085185285385485585685785885986086186286386486586686786886987087187287387487587687787887988088188288388488588688788888989089189289389489589689789889990090190290390490590690790890991091191291391491591691791891992092192292392492592692792892993093193293393493593693793893994094194294394494594694794894995095195295395495595695795895996096196296396496596696796896997097197297397497597697797897998098198298398498598698798898999099199299399499599699799899910001001100210031004100510061007100810091010101110121013101410151016101710181019102010211022102310241025102610271028102910301031103210331034103510361037103810391040104110421043104410451046104710481049105010511052105310541055105610571058105910601061106210631064106510661067106810691070107110721073107410751076107710781079108010811082108310841085108610871088108910901091109210931094109510961097109810991100110111021103110411051106110711081109111011111112111311141115111611171118111911201121112211231124112511261127112811291130113111321133113411351136113711381139114011411142114311441145114611471148114911501151115211531154115511561157115811591160116111621163116411651166116711681169117011711172117311741175117611771178117911801181118211831184118511861187118811891190119111921193119411951196119711981199120012011202120312041205120612071208120912101211121212131214121512161217121812191220122112221223122412251226122712281229123012311232123312341235123612371238123912401241124212431244124512461247124812491250125112521253125412551256125712581259126012611262126312641265126612671268126912701271127212731274127512761277127812791280128112821283128412851286128712881289129012911292129312941295129612971298129913001301130213031304130513061307130813091310131113121313131413151316131713181319132013211322132313241325132613271328132913301331133213331334133513361337133813391340134113421343134413451346134713481349135013511352135313541355135613571358135913601361136213631364136513661367136813691370137113721373137413751376137713781379138013811382138313841385138613871388138913901391139213931394139513961397139813991400140114021403140414051406140714081409141014111412141314141415141614171418141914201421142214231424142514261427142814291430143114321433143414351436143714381439144014411442144314441445144614471448144914501451145214531454145514561457145814591460146114621463146414651466146714681469147014711472147314741475147614771478147914801481148214831484148514861487148814891490149114921493149414951496149714981499150015011502150315041505150615071508150915101511151215131514151515161517151815191520152115221523152415251526152715281529153015311532153315341535153615371538153915401541154215431544154515461547154815491550155115521553155415551556155715581559156015611562156315641565156615671568156915701571157215731574157515761577157815791580158115821583158415851586158715881589159015911592159315941595159615971598159916001601160216031604160516061607160816091610161116121613161416151616161716181619162016211622162316241625162616271628162916301631163216331634163516361637163816391640164116421643164416451646164716481649165016511652165316541655165616571658165916601661166216631664166516661667166816691670167116721673167416751676167716781679168016811682168316841685168616871688168916901691169216931694169516961697169816991700170117021703170417051706170717081709171017111712171317141715171617171718171917201721172217231724172517261727172817291730173117321733173417351736173717381739174017411742174317441745174617471748174917501751175217531754175517561757175817591760176117621763176417651766176717681769177017711772177317741775177617771778177917801781178217831784178517861787178817891790179117921793179417951796179717981799180018011802180318041805180618071808180918101811181218131814181518161817181818191820182118221823182418251826182718281829183018311832183318341835183618371838183918401841184218431844184518461847184818491850185118521853185418551856185718581859186018611862186318641865186618671868186918701871187218731874187518761877187818791880188118821883188418851886188718881889189018911892189318941895189618971898189919001901190219031904190519061907190819091910191119121913191419151916191719181919192019211922192319241925192619271928192919301931193219331934193519361937193819391940194119421943194419451946194719481949195019511952195319541955195619571958195919601961196219631964196519661967196819691970197119721973197419751976197719781979198019811982198319841985198619871988198919901991199219931994199519961997199819992000200120022003200420052006200720082009201020112012201320142015201620172018201920202021202220232024202520262027202820292030203120322033203420352036203720382039204020412042204320442045204620472048204920502051205220532054205520562057205820592060206120622063206420652066206720682069207020712072207320742075207620772078207920802081208220832084208520862087208820892090209120922093209420952096209720982099210021012102210321042105210621072108210921102111211221132114211521162117211821192120212121222123212421252126212721282129213021312132213321342135213621372138213921402141214221432144214521462147214821492150215121522153215421552156215721582159216021612162216321642165216621672168216921702171217221732174217521762177217821792180218121822183218421852186218721882189219021912192219321942195219621972198219922002201220222032204220522062207220822092210221122122213221422152216221722182219222022212222222322242225222622272228222922302231223222332234223522362237223822392240224122422243224422452246224722482249225022512252225322542255225622572258225922602261226222632264226522662267226822692270227122722273227422752276227722782279228022812282228322842285228622872288228922902291229222932294229522962297229822992300230123022303230423052306230723082309231023112312231323142315231623172318231923202321232223232324232523262327232823292330233123322333233423352336233723382339234023412342234323442345234623472348234923502351235223532354235523562357235823592360236123622363236423652366236723682369237023712372237323742375237623772378237923802381238223832384238523862387238823892390239123922393239423952396239723982399240024012402240324042405240624072408240924102411241224132414241524162417241824192420242124222423242424252426242724282429243024312432243324342435243624372438243924402441244224432444//! 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 { replay_frame: @import("util/duration_profile.zig").DurationProfile = .{}, replay_validate: @import("util/duration_profile.zig").DurationProfile = .{}, replay_index: @import("util/duration_profile.zig").DurationProfile = .{}, replay_post_index: @import("util/duration_profile.zig").DurationProfile = .{}, 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 }, // backoff sleeps taken while the ring was full, the ordering lock was // held, or a replay waited for consumer buffer space. these advance while // the wait is in progress, so a stall is visible on /metrics as it happens. broadcast_queue_full_waits: std.atomic.Value(u64) = .{ .raw = 0 }, persist_order_waits: std.atomic.Value(u64) = .{ .raw = 0 }, replay_buffer_waits: std.atomic.Value(u64) = .{ .raw = 0 }, start_time: i64 = 0,};
/// spin briefly, then sleep between retries. `attempt` is the caller's retry/// count: the first `backoff_spin_budget` retries only hint the CPU, every/// later one sleeps so the waiter stops competing for CPU with whatever it is/// waiting for. under zio the sleep also yields the scheduler to other fibers.const backoff_spin_budget: u32 = 64;const backoff_sleep = Io.Duration.fromMilliseconds(1);
fn backoff(attempt: u32, waits: *std.atomic.Value(u64), io: Io) void { if (attempt <= backoff_spin_budget) { std.atomic.spinLoopHint(); return; } _ = waits.fetchAdd(1, .monotonic); io.sleep(backoff_sleep, .awake) catch std.atomic.spinLoopHint();}
// --- 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; var found_rebase = 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; if (std.mem.eql(u8, entry.key, "rebase")) found_rebase = 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; }
if (is_commit and !found_rebase) { new_entries.append(allocator, .{ .key = "rebase", .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). waits until space is available. /// matches Indigo semantics: every persisted event reaches live broadcast. /// a full ring means the broadcaster is not draining, and a producer that /// spins competes with it for the CPU it needs to drain (2026-10-18: 16 /// producers spun on 8 vCPUs for an hour). after a short spin budget the /// producer sleeps between retries instead. pub fn push(self: *BroadcastQueue, seq: u64, data: []const u8, stats: *Stats, io: Io) 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, back off, retry. the counter advances // per retry so the stall is visible while it lasts; saturating // add so a long stall can never panic the worker. self.push_lock.store(0, .release); _ = stats.broadcast_queue_full.fetchAdd(1, .monotonic); full_spins +|= 1; backoff(full_spins, &stats.broadcast_queue_full_waits, io); continue; }
self.items[tail] = .{ .seq = seq, .data = data }; self.tail.store(next_tail, .release); self.push_lock.store(0, .release);
// 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;/// how long a replay waits for the consumer to drain before retrying an enqueueconst replay_wait_sleep = Io.Duration.fromMilliseconds(10);
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, /// set while a cursor replay is in progress. broadcast() skips a replaying /// consumer entirely (no live enqueue, no slow-consumer kick): the replay /// delivers from disk at the consumer's own pace and hands off to live /// once it has caught up. see Broadcaster.replayTo. replaying: std.atomic.Value(bool) = .{ .raw = false }, /// replay → live handoff window: live frames are parked in `pending` with /// their seq until the final catch-up chunk is in `buf`, then moved over in /// order (dropping any the catch-up already delivered). guarded by `mutex`. staging: bool = false, pending: std.ArrayListUnmanaged(PendingFrame) = .empty, /// set by dropSlowConsumer. writeLoop sends ConsumerTooSlow on its way /// out (best effort) so the broadcaster never writes to a consumer socket. kicked: std.atomic.Value(bool) = .{ .raw = false }, error_frame: ?[]const u8 = null,
const PendingFrame = struct { seq: u64, frame: *SharedFrame };
/// 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; return self.enqueueLocked(frame); }
fn enqueueLocked(self: *Consumer, frame: *SharedFrame) bool { 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; }
/// live-path enqueue from broadcast(). during the replay → live handoff /// the frame is parked in `pending` instead of `buf`. pub fn enqueueLive(self: *Consumer, seq: u64, frame: *SharedFrame) bool { self.mutex.lockUncancelable(self.io); defer self.mutex.unlock(self.io);
if (!self.alive.load(.acquire)) return false; if (self.staging) { if (self.pending.items.len >= BUFFER_CAP) return false; self.pending.append(self.allocator, .{ .seq = seq, .frame = frame }) catch return false; frame.acquire(); return true; } return self.enqueueLocked(frame); }
/// replay enqueue: waits for buffer space instead of failing, so a /// consumer slower than disk playback gets a slower replay rather than a /// kick. returns false only when the consumer is dead or the relay is /// shutting down. pub fn enqueueRawWait(self: *Consumer, data: []const u8, shutdown_flag: *const std.atomic.Value(bool), waits: *std.atomic.Value(u64)) bool { const frame = SharedFrame.create(self.allocator, data) catch return false; defer frame.release(); while (self.alive.load(.acquire) and !shutdown_flag.load(.acquire)) { if (self.enqueue(frame)) return true; _ = waits.fetchAdd(1, .monotonic); self.io.sleep(replay_wait_sleep, .awake) catch return false; } return false; }
/// disk is drained: park live frames in `pending` while the final /// catch-up chunk is enqueued. fn beginStaging(self: *Consumer) void { self.mutex.lockUncancelable(self.io); defer self.mutex.unlock(self.io); self.staging = true; self.replaying.store(false, .release); }
/// move parked live frames newer than `since` (the last replayed seq) into /// `buf`, in order, and resume direct live delivery. returns false if `buf` /// cannot take them: the consumer is too slow for live and the caller kicks it. fn finishStaging(self: *Consumer, since: u64) bool { self.mutex.lockUncancelable(self.io); defer self.mutex.unlock(self.io); var ok = true; for (self.pending.items) |p| { defer p.frame.release(); if (p.seq <= since) continue; if (ok and !self.enqueueLocked(p.frame)) ok = false; } self.pending.clearAndFree(self.allocator); self.staging = false; return ok; }
/// drop parked frames (replay aborted, or the consumer is going away). fn abortStaging(self: *Consumer) void { self.mutex.lockUncancelable(self.io); defer self.mutex.unlock(self.io); self.releasePendingLocked(); }
fn releasePendingLocked(self: *Consumer) void { for (self.pending.items) |p| p.frame.release(); self.pending.clearAndFree(self.allocator); self.staging = false; }
pub fn deinit(self: *Consumer) void { self.mutex.lockUncancelable(self.io); defer self.mutex.unlock(self.io); self.releasePendingLocked(); }
/// 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; } self.abortStaging(); // a kicked consumer gets ConsumerTooSlow before the close, from this // fiber only: its socket is full by definition, and the broadcaster // must never park on it (2026-10-18). if (self.kicked.load(.acquire)) { if (self.error_frame) |ef| self.conn.writeBin(ef) catch {}; } // unblock readLoop, which is parked on this socket and owns its close. // this MUST happen here, not from the broadcast thread, which would // race readLoop's setsockopt/read calls on the socket. if (!self.conn.isClosed()) { self.conn.writeClose(.{}) catch {}; util.net.interrupt(self.io, self.conn.stream); } }
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, retain_ram_history: bool = true, 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(); consumer.deinit(); 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)); }
/// `replaying` marks a consumer that connected with a cursor: it is skipped /// by broadcast() until replayTo hands it off to live. pub fn addConsumer(self: *Broadcaster, conn: *websocket.Conn, replaying: bool) !*Consumer { const consumer = try self.allocator.create(Consumer); consumer.* = .{ .conn = conn, .allocator = self.allocator, .io = self.io, .replaying = .{ .raw = replaying }, .error_frame = self.error_frame, }; 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(); consumer.deinit(); 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 if (self.retain_ram_history and (self.persist == null or self.persist.?.history_enabled)) _ = 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.replaying.load(.acquire)) { // catching up from disk; replayTo delivers and hands off to live i += 1; continue; } if (consumer.enqueueLive(seq, 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 { _ = self; log.warn("dropping slow consumer (ConsumerTooSlow)", .{}); // never touch the consumer's socket from here. its buffer is full, so // its socket is full too, and a synchronous error-frame write parked // the broadcaster (holding consumers_mutex) until that peer read // (2026-10-18 wedge). writeLoop sends ConsumerTooSlow best effort on // its way out, then closes — closing from a third thread races // readLoop's setsockopt/read (EBADF → unreachable under ReleaseSafe). consumer.kicked.store(true, .release); 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. }
/// cursor replay. the consumer is in the live list but flagged `replaying`, /// so broadcast() leaves it alone while disk playback runs at the consumer's /// own pace (a slow reader gets a slow replay, not a kick). once a chunk /// comes back empty, live frames are parked (`staging`), one more disk pass /// collects anything broadcast meanwhile (every broadcast frame is on disk /// first, see runBroadcastLoop), and the parked frames newer than the last /// replayed seq follow in order. no gap, no reordering, duplicates only at /// the boundary. /// /// disk playback runs on pool_io (Threaded) via request/reply, in bounded /// chunks: materializing the whole backlog per connect is an O(retention) /// allocation, and a flapping consumer stacking those OOMed the canary /// (2026-08-06). pub fn replayTo(self: *Broadcaster, consumer: *Consumer, cursor: u64) void { if (self.persist) |dp| { var since = cursor; var replayed: usize = 0; if (!self.replayChunks(dp, consumer, &since, &replayed)) { self.abortReplay(consumer, cursor, replayed); return; } consumer.beginStaging(); if (!self.replayChunks(dp, consumer, &since, &replayed)) { self.abortReplay(consumer, cursor, replayed); return; } if (!consumer.finishStaging(since)) { log.warn("replay handoff: consumer too slow for live after {d} frames (cursor={d})", .{ replayed, cursor }); self.dropSlowConsumer(consumer); return; } if (replayed > 0) { log.info("replayed {d} frames from disk (cursor={d})", .{ replayed, cursor }); } return; }
// fallback: in-memory ring buffer (no persist configured). short and // synchronous, so live frames simply follow it. self.replayFromMemory(consumer, cursor); consumer.replaying.store(false, .release); }
fn abortReplay(self: *Broadcaster, consumer: *Consumer, cursor: u64, replayed: usize) void { _ = self; log.warn("replay aborted after {d} frames (consumer dead or shutting down, cursor={d})", .{ replayed, cursor }); consumer.abortStaging(); consumer.replaying.store(false, .release); }
/// play disk chunks after `since` to the consumer until one comes back /// empty. advances `since` and `replayed`. returns false when the consumer /// died or the relay is shutting down. fn replayChunks(self: *Broadcaster, dp: *event_log_mod.DiskPersist, consumer: *Consumer, since: *u64, replayed: *usize) bool { while (consumer.alive.load(.acquire) and !self.shutdown.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 true; }
if (req.entries.items.len == 0) return true; // 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.enqueueRawWait(frame_data, self.shutdown, &self.stats.replay_buffer_waits)) { return false; } replayed.* += 1; since.* = entry.seq; } } return false; }
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; }
/// take the persist → push ordering lock. spins briefly, then sleeps /// between retries: the holder may be blocked inside persist (disk mutex) /// or push (full ring), and spinning producers starve exactly the threads /// that would release it (2026-10-18: 15 workers spun on the lock while /// the 16th waited on a full ring, on an 8-vCPU node). pub fn lockPersistOrder(self: *Broadcaster, io: Io) void { var attempt: u32 = 0; while (self.persist_order.cmpxchgWeak(0, 1, .acquire, .monotonic) != null) { attempt +|= 1; _ = self.stats.persist_order_spins.fetchAdd(1, .monotonic); backoff(attempt, &self.stats.persist_order_waits, io); } }
pub fn unlockPersistOrder(self: *Broadcaster) void { self.persist_order.store(0, .release); }
/// 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 replaying = self.cursor != null and !self.invalid_cursor; const consumer = ctx.addConsumer(self.conn, replaying) 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| { if (ctx.persist) |dp| if (!dp.history_enabled) { var buf: [256]u8 = undefined; if (encodeErrorFrame("ReplayDisabled", "This relay serves live events only", &buf)) |frame| { consumer.conn.writeBin(frame) catch {}; } consumer.conn.close(.{}) catch {}; return; }; 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 retries waiting for the persist ordering lock \\relay_persist_order_spins_total {d} \\ \\# TYPE relay_persist_order_waits_total counter \\# HELP relay_persist_order_waits_total backoff sleeps waiting for the persist ordering lock (advances during a stall) \\relay_persist_order_waits_total {d} \\ \\# TYPE relay_broadcast_queue_full_waits_total counter \\# HELP relay_broadcast_queue_full_waits_total backoff sleeps on a full broadcast ring (advances during a stall) \\relay_broadcast_queue_full_waits_total {d} \\ \\# TYPE relay_replay_buffer_waits_total counter \\# HELP relay_replay_buffer_waits_total replay pauses waiting for a consumer to drain its buffer \\relay_replay_buffer_waits_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 retries on a 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.persist_order_waits.load(.acquire), stats.broadcast_queue_full_waits.load(.acquire), stats.replay_buffer_waits.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 {};
if (comptime @import("build_options").profile_stages) { inline for (.{ "frame", "validate", "index", "post_index" }) |stage| { const profile = &@field(stats, "replay_" ++ stage); w.print("relay_stage_count{{stage=\"{s}\"}} {d}\nrelay_stage_ns_total{{stage=\"{s}\"}} {d}\nrelay_stage_ns_max{{stage=\"{s}\"}} {d}\n", .{ stage, profile.count.load(.monotonic), stage, profile.total_ns.load(.monotonic), stage, profile.max_ns.load(.monotonic), }) 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").?); try std.testing.expectEqual(false, p.getBool("rebase").?);}
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")); try std.testing.expectEqual(@as(?bool, null), p.getBool("rebase"));}
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) |_| { bc_ptr.lockPersistOrder(std.testing.io); const seq = counter.fetchAdd(1, .monotonic) + 1; const data = std.testing.allocator.dupe(u8, "x") catch { bc_ptr.unlockPersistOrder(); continue; }; bc_ptr.broadcast_queue.push(seq, data, &bc_ptr.stats, std.testing.io); bc_ptr.unlockPersistOrder(); } } };
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, std.testing.io); } 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());}
test "push on a full ring sleeps instead of spinning, and completes once drained" { // regression: 2026-10-18 — a full ring turned every producer into a // busy spinner (7.5 cores on an 8-vCPU node) and starved the drain. var bc = Broadcaster.init(std.testing.allocator, std.testing.io, &test_shutdown); defer bc.deinit();
// fill the ring to capacity (CAPACITY - 1 usable slots) var seq: u64 = 1; while (bc.broadcast_queue.depth() < BroadcastQueue.CAPACITY - 1) : (seq += 1) { const data = try std.testing.allocator.dupe(u8, "x"); bc.broadcast_queue.push(seq, data, &bc.stats, std.testing.io); }
const Producer = struct { fn run(b: *Broadcaster, s: u64, done: *std.atomic.Value(bool)) void { const data = std.testing.allocator.dupe(u8, "y") catch unreachable; b.broadcast_queue.push(s, data, &b.stats, std.testing.io); done.store(true, .release); } }; var done = std.atomic.Value(bool){ .raw = false }; const t = try std.Thread.spawn(.{}, Producer.run, .{ &bc, seq, &done });
// give the producer time to exhaust its spin budget and start sleeping try std.testing.io.sleep(Io.Duration.fromMilliseconds(30), .awake); try std.testing.expect(!done.load(.acquire)); try std.testing.expect(bc.stats.broadcast_queue_full_waits.load(.acquire) > 0); try std.testing.expect(bc.stats.broadcast_queue_full.load(.acquire) > 0);
// one pop frees a slot; the producer must complete const item = bc.broadcast_queue.pop().?; std.testing.allocator.free(@constCast(item.data)); t.join(); try std.testing.expect(done.load(.acquire));
while (bc.broadcast_queue.pop()) |it| std.testing.allocator.free(@constCast(it.data));}
test "persist ordering lock backs off while held" { var bc = Broadcaster.init(std.testing.allocator, std.testing.io, &test_shutdown); defer bc.deinit();
bc.lockPersistOrder(std.testing.io); const Waiter = struct { fn run(b: *Broadcaster, acquired: *std.atomic.Value(bool)) void { b.lockPersistOrder(std.testing.io); acquired.store(true, .release); b.unlockPersistOrder(); } }; var acquired = std.atomic.Value(bool){ .raw = false }; const t = try std.Thread.spawn(.{}, Waiter.run, .{ &bc, &acquired }); try std.testing.io.sleep(Io.Duration.fromMilliseconds(30), .awake); try std.testing.expect(!acquired.load(.acquire)); try std.testing.expect(bc.stats.persist_order_waits.load(.acquire) > 0); bc.unlockPersistOrder(); t.join(); try std.testing.expect(acquired.load(.acquire));}
test "broadcast skips a replaying consumer and never writes to a kicked socket" { // the consumer's conn is undefined: any socket write from broadcast() // or dropSlowConsumer would fault. a replaying consumer gets nothing live; // a full consumer is kicked by flag only. var bc = Broadcaster.init(std.testing.allocator, std.testing.io, &test_shutdown); defer bc.deinit();
const consumer = try std.testing.allocator.create(Consumer); consumer.* = .{ .conn = undefined, .allocator = std.testing.allocator, .io = std.testing.io, .replaying = .{ .raw = true }, }; { bc.consumers_mutex.lockUncancelable(bc.io); defer bc.consumers_mutex.unlock(bc.io); try bc.consumers.append(std.testing.allocator, consumer); _ = bc.stats.consumer_count.fetchAdd(1, .monotonic); }
bc.broadcast(1, "live-1"); try std.testing.expectEqual(@as(usize, 0), consumer.buf_len); try std.testing.expectEqual(@as(u64, 0), bc.stats.frames_out.load(.acquire));
// hand off to live and fill the buffer: the next broadcast kicks by flag consumer.replaying.store(false, .release); while (consumer.buf_len < Consumer.BUFFER_CAP) { try std.testing.expect(consumer.enqueueRaw("fill")); } bc.broadcast(2, "live-2"); try std.testing.expect(consumer.kicked.load(.acquire)); try std.testing.expect(!consumer.alive.load(.acquire)); try std.testing.expectEqual(@as(u64, 1), bc.stats.slow_consumers.load(.acquire)); try std.testing.expectEqual(@as(usize, 0), bc.consumerCount());
consumer.mutex.lockUncancelable(consumer.io); while (consumer.dequeue()) |f| f.release(); consumer.mutex.unlock(consumer.io); consumer.deinit(); std.testing.allocator.destroy(consumer);}
test "replay handoff delivers staged live frames after the catch-up, in order, without duplicates" { var bc = Broadcaster.init(std.testing.allocator, std.testing.io, &test_shutdown); defer bc.deinit();
const consumer = try std.testing.allocator.create(Consumer); defer std.testing.allocator.destroy(consumer); defer consumer.deinit(); consumer.* = .{ .conn = undefined, .allocator = std.testing.allocator, .io = std.testing.io, .replaying = .{ .raw = true }, }; { bc.consumers_mutex.lockUncancelable(bc.io); defer bc.consumers_mutex.unlock(bc.io); try bc.consumers.append(std.testing.allocator, consumer); _ = bc.stats.consumer_count.fetchAdd(1, .monotonic); }
// replay delivers 1..3 while live 4 is skipped (it is on disk and the // catch-up pass will deliver it) try std.testing.expect(consumer.enqueueRaw("r1")); try std.testing.expect(consumer.enqueueRaw("r2")); bc.broadcast(4, "l4"); try std.testing.expect(consumer.enqueueRaw("r3"));
// handoff: live 5 and 6 are staged; the catch-up pass delivers 4 and 5 consumer.beginStaging(); bc.broadcast(5, "l5"); bc.broadcast(6, "l6"); try std.testing.expectEqual(@as(usize, 2), consumer.pending.items.len); try std.testing.expect(consumer.enqueueRaw("r4")); try std.testing.expect(consumer.enqueueRaw("r5")); try std.testing.expect(consumer.finishStaging(5));
// live 7 goes straight to the buffer bc.broadcast(7, "l7");
const expected = [_][]const u8{ "r1", "r2", "r3", "r4", "r5", "l6", "l7" }; consumer.mutex.lockUncancelable(consumer.io); defer consumer.mutex.unlock(consumer.io); for (expected) |want| { const f = consumer.dequeue() orelse return error.MissingFrame; defer f.release(); try std.testing.expectEqualStrings(want, f.data); } try std.testing.expectEqual(@as(usize, 0), consumer.buf_len); try std.testing.expect(!consumer.staging);
bc.consumers_mutex.lockUncancelable(bc.io); _ = bc.consumers.swapRemove(0); bc.consumers_mutex.unlock(bc.io);}
test "replay enqueue waits for buffer space instead of aborting" { var consumer = Consumer{ .conn = undefined, .allocator = std.testing.allocator, .io = std.testing.io, }; defer { consumer.mutex.lockUncancelable(consumer.io); while (consumer.dequeue()) |f| f.release(); consumer.mutex.unlock(consumer.io); } while (consumer.buf_len < Consumer.BUFFER_CAP) { try std.testing.expect(consumer.enqueueRaw("fill")); }
var waits = std.atomic.Value(u64){ .raw = 0 }; const Replayer = struct { fn run(c: *Consumer, w: *std.atomic.Value(u64), ok: *std.atomic.Value(bool)) void { ok.store(c.enqueueRawWait("late", &test_shutdown, w), .release); } }; var ok = std.atomic.Value(bool){ .raw = false }; const t = try std.Thread.spawn(.{}, Replayer.run, .{ &consumer, &waits, &ok }); try std.testing.io.sleep(Io.Duration.fromMilliseconds(40), .awake); try std.testing.expect(waits.load(.acquire) > 0);
consumer.mutex.lockUncancelable(consumer.io); const f = consumer.dequeue().?; consumer.mutex.unlock(consumer.io); f.release(); t.join(); try std.testing.expect(ok.load(.acquire));
// a dead consumer ends the wait with false consumer.alive.store(false, .release); try std.testing.expect(!consumer.enqueueRawWait("dead", &test_shutdown, &waits));}
test "resequenceFrame preserves explicit commit flags and signed blocks" { var arena = std.heap.ArenaAllocator.init(std.testing.allocator); defer arena.deinit(); const a = arena.allocator(); const cbor = zat.cbor; const header = try cbor.encodeAlloc(a, .{ .map = &.{ .{ .key = "op", .value = .{ .unsigned = 1 } }, .{ .key = "t", .value = .{ .text = "#commit" } }, } }); const payload = try cbor.encodeAlloc(a, .{ .map = &.{ .{ .key = "seq", .value = .{ .unsigned = 1 } }, .{ .key = "rebase", .value = .{ .boolean = true } }, .{ .key = "tooBig", .value = .{ .boolean = true } }, .{ .key = "blocks", .value = .{ .bytes = "signed CAR bytes" } }, } }); const frame = try std.mem.concat(a, u8, &.{ header, payload }); const result = resequenceFrame(a, frame, 42) orelse return error.ResequenceFailed; const h = try cbor.decode(a, result); const p = try cbor.decodeAll(a, result[h.consumed..]); try std.testing.expectEqual(true, p.getBool("rebase").?); try std.testing.expectEqual(true, p.getBool("tooBig").?); try std.testing.expectEqualStrings("signed CAR bytes", p.getBytes("blocks").?); try std.testing.expectEqual(@as(i64, 42), p.getInt("seq").?);}
test "live-only mode does not retain RAM replay frames" { var shutdown = std.atomic.Value(bool).init(false); var bc = Broadcaster.init(std.testing.allocator, std.testing.io, &shutdown); defer bc.deinit(); var dp: event_log_mod.DiskPersist = .{ .allocator = std.testing.allocator, .io = std.testing.io, .dir_path = "", .dir = undefined, .db = undefined, .did_cache = undefined, .host_id_cache = undefined, .history_enabled = false, }; bc.persist = &dp; bc.broadcast(1, "not retained"); try std.testing.expectEqual(@as(usize, 0), bc.history.count()); dp.history_enabled = true; bc.retain_ram_history = false; bc.broadcast(2, "disk replay without RAM retention"); try std.testing.expectEqual(@as(usize, 0), bc.history.count()); bc.retain_ram_history = true; bc.broadcast(3, "retained"); try std.testing.expectEqual(@as(usize, 1), bc.history.count());}