Something went wrong. Try again.
jetstream v2 in zig stream.waow.tech
Something went wrong. Try again.
12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182838485868788899091929394959697989910010110210310410510610710810911011111211311411511611711811912012112212312412512612712812913013113213313413513613713813914014114214314414514614714814915015115215315415515615715815916016116216316416516616716816917017117217317417517617717817918018118218318418518618718818919019119219319419519619719819920020120220320420520620720820921021121221321421521621721821922022122222322422522622722822923023123223323423523623723823924024124224324424524624724824925025125225325425525625725825926026126226326426526626726826927027127227327427527627727827928028128228328428528628728828929029129229329429529629729829930030130230330430530630730830931031131231331431531631731831932032132232332432532632732832933033133233333433533633733833934034134234334434534634734834935035135235335435535635735835936036136236336436536636736836937037137237337437537637737837938038138238338438538638738838939039139239339439539639739839940040140240340440540640740840941041141241341441541641741841942042142242342442542642742842943043143243343443543643743843944044144244344444544644744844945045145245345445545645745845946046146246346446546646746846947047147247347447547647747847948048148248348448548648748848949049149249349449549649749849950050150250350450550650750850951051151251351451551651751851952052152252352452552652752852953053153253353453553653753853954054154254354454554654754854955055155255355455555655755855956056156256356456556656756856957057157257357457557657757857958058158258358458558658758858959059159259359459559659759859960060160260360460560660760860961061161261361461561661761861962062162262362462562662762862963063163263363463563663763863964064164264364464564664764864965065165265365465565665765865966066166266366466566666766866967067167267367467567667767867968068168268368468568668768868969069169269369469569669769869970070170270370470570670770870971071171271371471571671771871972072172272372472572672772872973073173273373473573673773873974074174274374474574674774874975075175275375475575675775875976076176276376476576676776876977077177277377477577677777877978078178278378478578678778878979079179279379479579679779879980080180280380480580680780880981081181281381481581681781881982082182282382482582682782882983083183283383483583683783883984084184284384484584684784884985085185285385485585685785885986086186286386486586686786886987087187287387487587687787887988088188288388488588688788888989089189289389489589689789889990090190290390490590690790890991091191291391491591691791891992092192292392492592692792892993093193293393493593693793893994094194294394494594694794894995095195295395495595695795895996096196296396496596696796896997097197297397497597697797897998098198298398498598698798898999099199299399499599699799899910001001100210031004100510061007100810091010101110121013101410151016101710181019102010211022102310241025102610271028102910301031103210331034103510361037103810391040104110421043104410451046104710481049105010511052105310541055105610571058105910601061106210631064106510661067106810691070107110721073107410751076107710781079108010811082108310841085108610871088108910901091109210931094109510961097109810991100110111021103110411051106110711081109111011111112111311141115111611171118111911201121112211231124112511261127112811291130113111321133113411351136113711381139114011411142114311441145114611471148114911501151115211531154115511561157115811591160116111621163116411651166116711681169117011711172117311741175117611771178117911801181118211831184118511861187118811891190119111921193119411951196119711981199120012011202120312041205120612071208120912101211121212131214121512161217121812191220122112221223122412251226122712281229123012311232123312341235123612371238123912401241124212431244124512461247124812491250125112521253125412551256125712581259126012611262126312641265126612671268126912701271127212731274127512761277127812791280128112821283128412851286128712881289129012911292129312941295129612971298129913001301130213031304130513061307130813091310131113121313131413151316131713181319132013211322132313241325132613271328132913301331133213331334133513361337133813391340134113421343134413451346134713481349135013511352135313541355135613571358135913601361136213631364136513661367136813691370137113721373137413751376137713781379138013811382138313841385138613871388138913901391139213931394139513961397139813991400140114021403140414051406140714081409141014111412//! upstream ingest — firehose events feeding the hot tail//!//! transport (connect, reconnect/backoff, heartbeat, frame decode) is//! zat.FirehoseClient's job; this module is just the event handler://! convert to v1 frames, append to the shared tail.
const std = @import("std");const zat = @import("zat");const archive_mod = @import("../storage/archive.zig");const cold = @import("../storage/cold.zig");const convert = @import("convert.zig");const segment = @import("../storage/segment.zig");const cursor_mod = @import("cursor.zig");const metrics = @import("../runtime/metrics.zig");const meta_store = @import("../storage/meta_store.zig");const observability = @import("../runtime/observability.zig");const pipeline_mod = @import("pipeline.zig");const repair = @import("repair.zig");const resync = @import("backfill/resync.zig");const repos = @import("backfill/repos.zig");const tail_mod = @import("../serve/tail.zig");const verify = @import("verify.zig");
const Io = std.Io;const Allocator = std.mem.Allocator;const log = std.log.scoped(.stream);
const StreamErrorCode = enum { future_cursor, consumer_too_slow, other, missing };const InspectedFrame = union(enum) { message: struct { seq: ?i64 }, unknown: struct { seq: ?i64 }, stream_error: StreamErrorCode, malformed,};
fn inspectFrame(allocator: Allocator, data: []const u8) InspectedFrame { var arena = std.heap.ArenaAllocator.init(allocator); defer arena.deinit(); // Firehose envelopes are schema-ordered CBOR, not repository DAG-CBOR. // Future lexicon producers may use any legal map order; inspection must // classify those frames without weakening strict repo/block decoding. const header = zat.cbor.decodeLax(arena.allocator(), data) catch return .malformed; const op = header.value.getInt("op") orelse return .malformed; const payload = zat.cbor.decodeAllLax(arena.allocator(), data[header.consumed..]) catch return .malformed; if (op == -1) { const code = payload.getString("error") orelse return .{ .stream_error = .missing }; if (std.mem.eql(u8, code, "FutureCursor")) return .{ .stream_error = .future_cursor }; if (std.mem.eql(u8, code, "ConsumerTooSlow")) return .{ .stream_error = .consumer_too_slow }; return .{ .stream_error = .other }; } const seq = payload.getInt("seq"); if (op != 1) return .{ .unknown = .{ .seq = seq } }; const tag = header.value.getString("t") orelse return .malformed; const known = std.mem.eql(u8, tag, "#commit") or std.mem.eql(u8, tag, "#sync") or std.mem.eql(u8, tag, "#identity") or std.mem.eql(u8, tag, "#account") or std.mem.eql(u8, tag, "#info"); return if (known) .{ .message = .{ .seq = seq } } else .{ .unknown = .{ .seq = seq } };}
fn testFrame(allocator: Allocator, header: zat.cbor.Value, payload: zat.cbor.Value) ![]u8 { const header_bytes = try zat.cbor.encodeAlloc(allocator, header); defer allocator.free(header_bytes); const payload_bytes = try zat.cbor.encodeAlloc(allocator, payload); defer allocator.free(payload_bytes); return std.mem.concat(allocator, u8, &.{ header_bytes, payload_bytes });}
/// Reconnect trigger for an upstream that keeps the socket busy while/// delivering far below the network's rate. relay1.us-east.bsky.network served/// ~18 frames/s for 80 minutes on 2026-10-01: enough to keep the idle timeout/// quiet, and replayable only because someone restarted the process. Windows/// tumble per connection; each slow verdict doubles the next window so a/// persistently slow upstream is not reconnected to once a minute forever.pub const SlowUpstream = struct { /// frames/second; non-positive disables the check min_rate: f64 = 0, base_window_ns: u64 = 60 * std.time.ns_per_s, backoff: u3 = 0, window_start_ns: ?i96 = null, frames: u64 = 0,
const max_backoff: u3 = 4;
pub fn onConnect(self: *SlowUpstream) void { self.window_start_ns = null; self.frames = 0; }
/// Returns the measured rate when the window that just closed was slow. pub fn observe(self: *SlowUpstream, now_ns: i96) ?f64 { if (self.min_rate <= 0) return null; const start = self.window_start_ns orelse { self.window_start_ns = now_ns; return null; }; self.frames += 1; const elapsed_ns = now_ns - start; if (elapsed_ns < @as(i96, self.base_window_ns) << self.backoff) return null; const rate = @as(f64, @floatFromInt(self.frames)) * std.time.ns_per_s / @as(f64, @floatFromInt(elapsed_ns)); self.window_start_ns = now_ns; self.frames = 0; if (rate >= self.min_rate) { self.backoff = 0; return null; } self.backoff = @min(self.backoff + 1, max_backoff); return rate; }};
pub const Consumer = struct { allocator: Allocator, io: Io, tail: *tail_mod.Tail, /// full ws(s):// URL, e.g. "ws://localhost:7777" or "wss://bsky.network" upstream: []const u8, cursor: ?i64 = null, to_stdout: bool = false, durable_upstream_seq: i64 = 0, verifier: ?*verify.Verifier = null, pipeline: ?*pipeline_mod.Pipeline = null, cursor_store: ?*cursor_mod.Store = null, archive: ?*archive_mod.Archive = null, stats: ?*metrics.Stats = null, /// Process shutdown gate. The signal handler only flips this atomic; the /// reconnect loop observes it through shouldStop at a safe point. shutdown: ?*const std.atomic.Value(bool) = null, drops: convert.Drops = .initFill(0), pending_cursors: std.ArrayList(CursorPoint) = .empty, pending_chains: std.ArrayList(ChainPoint) = .empty, pending_ratchets: std.ArrayList(RatchetPoint) = .empty, pending_hosting: std.ArrayList(HostingPoint) = .empty, /// Bootstrap-live owns a temporary local sequence space. It persists the /// shared relay cursor and verifier state, while lifecycle publishes the /// merged destination's sequence metadata at cutover. write_archive_sequence_metadata: bool = true, emit_failure_logs: bool = true, fatal_mu: Io.Mutex = .init, fatal_error: ?anyerror = null, last_wire_seq: ?i64 = null, slow_upstream: SlowUpstream = .{},
/// blocks forever; zat handles reconnect with backoff and host rotation. pub fn run(self: *Consumer) !void { var client = zat.FirehoseClient.init(self.io, self.allocator, .{ .hosts = &.{self.upstream}, .cursor = self.cursor, .max_message_size = 2 * 1024 * 1024, // the relay stalled silently for minutes (2026-08-11 12:02Z, // 16:16Z) before closing us itself; a live relay is never this // quiet, so bound the wait instead of eating the whole gap .idle_timeout_ms = 20_000, }); defer client.deinit(); try client.subscribe(self); if (self.pipeline) |pipeline| if (pipeline.fatalError()) |err| return err; if (self.fatalError()) |err| return err; }
/// Raw frames go through the Atmos-compatible scheduler. Its writer calls /// emitVerified in completion order across DIDs and FIFO order per DID. pub fn onRawFrame(self: *Consumer, data: []const u8) !void { // Failing before submit leaves this frame unacknowledged, so the // reconnect resumes from the cursor and the relay redelivers it. if (self.slow_upstream.observe(Io.Timestamp.now(self.io, .awake).toNanoseconds())) |rate| { if (self.stats) |stats| _ = stats.upstream_slow_reconnects_total.fetchAdd(1, .monotonic); if (self.emit_failure_logs) log.warn("upstream delivered {d:.1} frames/s, below --upstream-slow-min-rate={d}", .{ rate, self.slow_upstream.min_rate }); return error.UpstreamTooSlow; } const pipe = self.pipeline orelse return; switch (inspectFrame(self.allocator, data)) { .message => |message| self.observeWireSeq(message.seq), .unknown => |unknown| { self.observeWireSeq(unknown.seq); if (self.stats) |stats| _ = stats.livestream_unknown_events_total.fetchAdd(1, .monotonic); return; }, .stream_error => |code| { self.observeStreamError(code); return; }, .malformed => {}, } pipe.submit(data) catch |err| { // Cancellation can interrupt a producer waiting for one of the // 32 active-DID worker slots during intentional shutdown. A // closed pipeline without a shutdown request is different: its // fatal cause must escape to the websocket reader so that frame // is not acknowledged as reconnect-safe. if ((err == error.Canceled or err == error.Closed) and self.shutdown != null and self.shutdown.?.load(.acquire)) return err; const failure = pipe.fatalError() orelse err; if (self.emit_failure_logs) log.err("pipeline rejected firehose frame: {s}", .{@errorName(failure)}); self.setFatal(failure); return failure; }; }
pub fn onError(self: *Consumer, err: anyerror) void { if (self.emit_failure_logs) log.err("firehose error: {s}, reconnecting", .{@errorName(err)}); }
pub fn onConnect(self: *Consumer, _: []const u8) void { self.slow_upstream.onConnect(); }
pub fn onReconnect(self: *Consumer) void { if (self.stats) |stats| _ = stats.livestream_reconnects_total.fetchAdd(1, .monotonic); }
fn observeWireSeq(self: *Consumer, maybe_seq: ?i64) void { const seq = maybe_seq orelse return; const previous = self.last_wire_seq orelse self.cursor; if (previous) |last| { const delta = std.math.sub(i64, seq, last) catch std.math.maxInt(i64); if (delta > 1) if (self.stats) |stats| { _ = stats.livestream_sequence_gaps_total.fetchAdd(1, .monotonic); _ = stats.livestream_sequence_gap_missed_seqs_total.fetchAdd(@intCast(delta - 1), .monotonic); }; if (seq <= last) return; } self.last_wire_seq = seq; }
fn observeStreamError(self: *Consumer, code: StreamErrorCode) void { const stats = self.stats orelse return; const counter = switch (code) { .future_cursor => &stats.livestream_stream_error_future_cursor_total, .consumer_too_slow => &stats.livestream_stream_error_consumer_too_slow_total, .other => &stats.livestream_stream_error_other_total, .missing => &stats.livestream_stream_error_missing_total, }; _ = counter.fetchAdd(1, .monotonic); }
pub fn shouldStop(self: *Consumer) bool { if (self.shutdown) |flag| if (flag.load(.acquire)) return true; return self.fatalError() != null; }
/// pipeline writer callback: the event was decoded once by the verify /// worker; its backing arena stays alive for the duration of this call. pub fn emitVerified( ctx: *anyopaque, event: zat.FirehoseEvent, verdict: verify.Result, trace_context: observability.Context, ) !void { const self: *Consumer = @ptrCast(@alignCast(ctx)); self.handleEvent(event, verdict, false, trace_context) catch |err| { log.err("fatal event handling failure: {s}", .{@errorName(err)}); self.setFatal(err); return err; }; }
pub fn emitRepair(ctx: *anyopaque, completion: *repair.Completion, trace_context: observability.Context) !void { const self: *Consumer = @ptrCast(@alignCast(ctx)); const archive = self.archive orelse return error.RepairRequiresArchive; const attributes = [_]observability.Attribute{ .{ .key = "did", .value = .{ .string = completion.did } }, }; var trace = observability.Span.start(self.allocator, "ingest/backfill", "handleRepo", trace_context, &attributes); defer trace.deinit(); defer trace.succeed(); try trace.recordError(self.applyRepair(archive, completion, trace.context())); }
fn applyRepair( self: *Consumer, archive: *archive_mod.Archive, completion: *repair.Completion, trace_context: observability.Context, ) !void { var arena = std.heap.ArenaAllocator.init(self.allocator); defer arena.deinit(); const alloc = arena.allocator(); const now_us = completion.witnessed_at;
const Sink = struct { consumer: *Consumer, archive: *archive_mod.Archive, allocator: Allocator, now_us: i64, batch: std.ArrayList(segment.Event) = .empty, last_seq: u64 = 0, fetched_chain: verify.ChainUpdate, did: []const u8, upstream_seq: ?i64, trace_context: observability.Context, stage_chain_on_flush: bool = false,
fn stageChain(hook_ctx: *anyopaque, _: u64, last: u64) !void { const sink: *@This() = @ptrCast(@alignCast(hook_ctx)); const owned_did = try sink.consumer.allocator.dupe(u8, sink.did); errdefer sink.consumer.allocator.free(owned_did); try sink.consumer.pending_chains.append(sink.consumer.allocator, .{ .local_seq = last, .did = owned_did, .state = sink.fetched_chain, }); if (sink.upstream_seq) |seq| { if (sink.consumer.stats) |stats| stats.upstream_seq.store(seq, .monotonic); try sink.consumer.pending_cursors.append(sink.consumer.allocator, .{ .local_seq = last, .upstream_seq = seq, }); } }
fn flush(sink: *@This()) !void { if (sink.batch.items.len == 0) return; const first = (if (sink.stage_chain_on_flush) try sink.archive.appendBatchBeforeDurableTraced( sink.batch.items, sink.now_us, sink, stageChain, sink.trace_context, ) else try sink.archive.appendBatchTraced(sink.batch.items, sink.now_us, sink.trace_context)) orelse return; for (sink.batch.items, 0..) |row, i| { var stamped = row; stamped.seq = first + i; try appendArchivedTailRow(sink.consumer, stamped); } sink.last_seq = first + sink.batch.items.len - 1; sink.batch.clearRetainingCapacity(); }
fn emit(sink: *@This(), row: segment.Event) anyerror!void { if (sink.batch.items.len == 1024) try sink.flush(); try sink.batch.append(sink.allocator, row); } }; var sink: Sink = .{ .consumer = self, .archive = archive, .allocator = alloc, .now_us = now_us, .fetched_chain = completion.fetched_chain, .did = completion.did, .upstream_seq = completion.upstream_seq, .trace_context = trace_context, }; defer sink.batch.deinit(alloc); const sync_event = if (completion.sync_payload) |payload| try resync.syncTombstoneFromPayload(alloc, completion.did, payload, now_us) else try resync.syncTombstoneEvent(alloc, completion.did, completion.prepared.repo.rev, now_us);
// A malformed #sync rev is downstream-gate-owned in upstream. The // authoritative fetch still repairs and durably advances verifier // state, but the original directive and every replacement op are // rejected as one live invalid_rev event. if (completion.sync_payload != null and zat.Tid.parse(sync_event.rev) == null) { const cs = self.cursor_store orelse return error.RepairRequiresCursorStore; var writes = cs.meta.batch(); defer writes.deinit(); try verify.stageChain(&writes, completion.did, completion.fetched_chain); try writes.commitDurable(); self.drops.getPtr(.invalid_rev).* += 1; if (self.stats) |stats| _ = stats.drops.getPtr(.invalid_rev).fetchAdd(1, .monotonic); for (completion.pending.items) |item| try self.handleEvent(item.event, item.verdict, true, trace_context); try archive.flushBlockTraced(trace_context); return; }
try sink.emit(sync_event); var drops: repos.DropCounts = .{}; _ = try repos.emitPrepared( completion.prepared_arena.allocator(), &completion.prepared.repo, completion.did, now_us, .create_resync, &drops, &sink, Sink.emit, ); self.drops.getPtr(.invalid_collection).* += drops.invalid_collection; self.drops.getPtr(.invalid_rkey).* += drops.invalid_rkey; self.drops.getPtr(.field_too_long).* += drops.field_too_long; if (self.stats) |stats| { _ = stats.drops.getPtr(.invalid_collection).fetchAdd(drops.invalid_collection, .monotonic); _ = stats.drops.getPtr(.invalid_rkey).fetchAdd(drops.invalid_rkey, .monotonic); _ = stats.drops.getPtr(.field_too_long).fetchAdd(drops.field_too_long, .monotonic); } sink.stage_chain_on_flush = true; try sink.flush(); for (completion.pending.items) |item| try self.handleEvent(item.event, item.verdict, true, trace_context); try archive.flushBlockTraced(trace_context); }
/// Atmos advances its cursor only after a non-empty delivery batch /// returns. The pipeline supplies that smallest-inflight watermark here; /// it is staged against the latest local archive row, never an individual /// event's possibly-out-of-order relay seq. pub fn emitCursor(ctx: *anyopaque, watermark: i64) !void { const self: *Consumer = @ptrCast(@alignCast(ctx)); const now_us = Io.Timestamp.now(self.io, .real).toMicroseconds(); try self.noteWatermark(watermark, now_us, if (self.archive) |archive| archive.lastSeq() else 0); }
pub fn fatalError(self: *Consumer) ?anyerror { self.fatal_mu.lockUncancelable(self.io); const own_error = self.fatal_error; self.fatal_mu.unlock(self.io); if (own_error) |err| return err;
// A pipeline can fail between websocket frames. In that schedule the // reconnect loop observes shouldStop before onRawFrame gets another // opportunity to copy the cause into fatal_error. Always consult the // pipeline after releasing fatal_mu so bootstrap reports the original // persistence/repair failure rather than the generic capture-ended // sentinel. return if (self.pipeline) |pipeline| pipeline.fatalError() else null; }
fn setFatal(self: *Consumer, err: anyerror) void { self.fatal_mu.lockUncancelable(self.io); defer self.fatal_mu.unlock(self.io); if (self.fatal_error == null) self.fatal_error = err; }
fn handleEvent( self: *Consumer, event: zat.FirehoseEvent, verdict: verify.Result, replayed_after_repair: bool, trace_context: observability.Context, ) !void { var arena = std.heap.ArenaAllocator.init(self.allocator); defer arena.deinit(); const alloc = arena.allocator();
const time_us: i64 = Io.Timestamp.now(self.io, .real).toMicroseconds(); if (!replayed_after_repair) { if (self.stats) |st| { _ = st.events_total.fetchAdd(1, .monotonic); st.livestream_last_seen_upstream_event_timestamp_seconds.store( @divFloor(time_us, std.time.us_per_s), .monotonic, ); } }
switch (event) { .commit => |c| switch (verdict.outcome) { .invalid_signature, .invalid_commit, .unverified => { // Never contaminate the archive with a commit we could // not prove. These failures bypass authoritative repair. log.warn("rejected unverified commit: {s} seq={d} reason={s}", .{ c.repo, c.seq, @tagName(verdict.outcome) }); return; }, .needs_resync => { // The coordinator owns this DID now. The trigger is // replaced by its authoritative full repo; later commits // are already in the bounded pending FIFO. log.debug("commit withheld for repair: {s} seq={d}", .{ c.repo, c.seq }); return; }, .valid => {}, .replay => { return; }, }, .sync => switch (verdict.outcome) { .valid => {}, .replay, .invalid_signature, .invalid_commit, .unverified, .needs_resync => { return; }, }, .identity => |identity| { if (identity.seq <= try self.appliedSeq(.identity, identity.did)) { if (self.stats) |stats| _ = stats.replayed_identity_events_dropped_total.fetchAdd(1, .monotonic); if (!replayed_after_repair) try self.noteSeq(event, time_us, if (self.archive) |a| a.lastSeq() else 0); return; } // rotation signal: next commit re-resolves the key if (self.verifier) |v| v.evict(identity.did); }, .account => |account| if (account.seq <= try self.appliedSeq(.account, account.did)) { if (self.stats) |stats| _ = stats.replayed_account_events_dropped_total.fetchAdd(1, .monotonic); if (!replayed_after_repair) try self.noteSeq(event, time_us, if (self.archive) |a| a.lastSeq() else 0); return; }, else => {}, } var persisted: Persisted = .{}; const owns_cursor = !replayed_after_repair and self.pipeline == null; if (self.archive) |a| persisted = self.persist(alloc, a, event, verdict, owns_cursor, trace_context) catch |err| { // our own storage failing is fatal-worthy, not drop-worthy log.err("archive append failed: {s}", .{@errorName(err)}); return err; };
var frames: std.ArrayList(convert.Frame) = .empty; const drops_before = self.drops; try convert.convertStamped( alloc, event, time_us, persisted.first, if (persisted.rows.len == 0) null else persisted.rows, &frames, &self.drops, ); logNewDrops(event, &drops_before, &self.drops); if (self.stats) |st| { st.addDrops(&drops_before, &self.drops); _ = st.frames_total.fetchAdd(frames.items.len, .monotonic); }
for (frames.items) |frame| { try self.tail.append(frame.kind, frame.did, frame.collection, frame.time_us, frame.seq, frame.json, frame.json_v2, if (frame.kind == .sync) .sync else .none); }
if (persisted.last == 0) { if (self.archive) |a| { const barrier = a.lastSeq(); try self.stageEventBoundary(event, verdict, barrier, owns_cursor); try self.commitDurableMetadata( a.committed_seq.load(.acquire), a.committed_seq.load(.acquire) + 1, if (owns_cursor) event.seq() else null, ); } else if (owns_cursor) try self.noteSeq(event, time_us, 0); }
if (self.to_stdout and frames.items.len > 0) { var stdout_buf: [64 * 1024]u8 = undefined; var stdout = std.Io.File.stdout().writer(self.io, &stdout_buf); for (frames.items) |frame| { stdout.interface.print("{s}\n", .{frame.json}) catch return error.WriteFailed; } stdout.interface.flush() catch return error.WriteFailed; } }
fn noteSeq(self: *Consumer, event: zat.FirehoseEvent, time_us: i64, local_seq: u64) !void { if (self.pipeline != null) return; const seq = event.seq() orelse return; try self.noteWatermark(seq, time_us, local_seq); }
fn noteWatermark(self: *Consumer, seq: i64, time_us: i64, local_seq: u64) !void { if (self.stats) |st| st.upstream_seq.store(seq, .monotonic); if (self.archive) |a| { if (a.committed_seq.load(.acquire) >= local_seq) { try self.commitDurableMetadata(a.committed_seq.load(.acquire), a.committed_seq.load(.acquire) + 1, seq); } else try self.pending_cursors.append(self.allocator, .{ .local_seq = local_seq, .upstream_seq = seq }); return; } if (self.cursor_store) |cs| cs.update(self.io, seq, time_us); }
pub fn onArchiveDurable(ctx: *anyopaque, durable_seq: u64, next_seq: u64) anyerror!void { const self: *Consumer = @ptrCast(@alignCast(ctx)); var upstream_seq: ?i64 = null; var consumed: usize = 0; for (self.pending_cursors.items) |point| { if (point.local_seq > durable_seq) break; upstream_seq = point.upstream_seq; consumed += 1; } try self.commitDurableMetadata(durable_seq, next_seq, upstream_seq); if (consumed > 0) self.pending_cursors.replaceRange(self.allocator, 0, consumed, &.{}) catch unreachable; // Segment bytes and their cursor metadata are now durable; the tail // may release the corresponding pinned prefix to eviction. self.tail.publishDurable(durable_seq); }
fn commitDurableMetadata(self: *Consumer, durable_seq: u64, next_seq: u64, upstream_seq: ?i64) !void { const cs = self.cursor_store orelse return; var writes = cs.meta.batch(); defer writes.deinit(); if (self.write_archive_sequence_metadata) { var next_bytes: [8]u8 = undefined; std.mem.writeInt(u64, &next_bytes, next_seq, .little); writes.put("seq/next", &next_bytes); var durable_bytes: [8]u8 = undefined; std.mem.writeInt(u64, &durable_bytes, durable_seq, .little); writes.put("seq/durable", &durable_bytes); } if (upstream_seq) |seq| try cursor_mod.Store.stage(&writes, seq); var durable_chains: usize = 0; for (self.pending_chains.items) |point| { if (point.local_seq > durable_seq) break; try verify.stageChain(&writes, point.did, point.state); durable_chains += 1; } var durable_ratchets: usize = 0; for (self.pending_ratchets.items) |point| { if (point.local_seq > durable_seq) break; var key_buf: [512]u8 = undefined; const key = try ratchetKey(&key_buf, point.kind, point.did); var seq_bytes: [8]u8 = undefined; std.mem.writeInt(i64, &seq_bytes, point.upstream_seq, .little); writes.put(key, &seq_bytes); durable_ratchets += 1; } var durable_hosting: usize = 0; for (self.pending_hosting.items) |point| { if (point.local_seq > durable_seq) break; try verify.stageHosting(&writes, point.did, point.state); durable_hosting += 1; } try writes.commitDurable(); if (upstream_seq) |seq| self.durable_upstream_seq = seq; for (self.pending_chains.items[0..durable_chains]) |point| self.allocator.free(point.did); if (durable_chains > 0) self.pending_chains.replaceRange(self.allocator, 0, durable_chains, &.{}) catch unreachable; for (self.pending_ratchets.items[0..durable_ratchets]) |point| self.allocator.free(point.did); if (durable_ratchets > 0) self.pending_ratchets.replaceRange(self.allocator, 0, durable_ratchets, &.{}) catch unreachable; for (self.pending_hosting.items[0..durable_hosting]) |point| self.allocator.free(point.did); if (durable_hosting > 0) self.pending_hosting.replaceRange(self.allocator, 0, durable_hosting, &.{}) catch unreachable; }
fn appliedSeq(self: *Consumer, kind: RatchetKind, did: []const u8) !i64 { var applied: i64 = 0; for (self.pending_ratchets.items) |point| { if (point.kind == kind and std.mem.eql(u8, point.did, did)) applied = @max(applied, point.upstream_seq); } var key_buf: [512]u8 = undefined; const key = try ratchetKey(&key_buf, kind, did); const cs = self.cursor_store orelse return applied; const bytes = try cs.meta.getAlloc(self.allocator, key) orelse return applied; defer self.allocator.free(bytes); if (bytes.len != 8) return error.InvalidVerifierState; return @max(applied, std.mem.readInt(i64, bytes[0..8], .little)); }
fn stageRatchet(self: *Consumer, kind: RatchetKind, did: []const u8, upstream_seq: i64, local_seq: u64) !void { const owned = try self.allocator.dupe(u8, did); errdefer self.allocator.free(owned); try self.pending_ratchets.append(self.allocator, .{ .kind = kind, .did = owned, .upstream_seq = upstream_seq, .local_seq = local_seq, }); }
/// persist a firehose event into the archive as segment rows: one per /// commit op (payload = re-encoded record DAG-CBOR), one per /// identity/account (payload = the envelope re-encoded as CBOR). /// returns the local seq of the FIRST row appended (0 if none); rows /// for one upstream event are contiguous, so frame i has seq first+i fn persist( self: *Consumer, alloc: Allocator, a: *archive_mod.Archive, event: zat.FirehoseEvent, verdict: verify.Result, stage_cursor: bool, trace_context: observability.Context, ) !Persisted { const now_us: i64 = Io.Timestamp.now(self.io, .real).toMicroseconds();
var rows: std.ArrayList(segment.Event) = .empty; defer rows.deinit(alloc); switch (event) { .commit => |c| { if (zat.Tid.parse(c.rev) == null) return .{}; for (c.ops) |op| { if (zat.Nsid.parse(op.collection) == null) continue; if (zat.Rkey.parse(op.rkey) == null) continue; const is_delete = op.action == .delete; if (!is_delete and (op.record == null or op.cid == null)) continue; const payload: []const u8 = if (is_delete) "" else try zat.cbor.encodeAlloc(alloc, op.record.?); if (!convert.representableFields(c.repo, op.collection, op.rkey, c.rev, payload.len)) continue; try rows.append(alloc, .{ .seq = 0, .witnessed_at = now_us, .indexed_at = 0, .kind = switch (op.action) { .create => .create, .update => .update, .delete => .delete, }, .did = c.repo, .collection = op.collection, .rkey = op.rkey, .rev = c.rev, .payload = payload, }); } }, .identity => |i| try rows.append(alloc, .{ .seq = 0, .witnessed_at = now_us, .indexed_at = 0, .kind = .identity, .did = i.did, .collection = "", .rkey = "", .rev = "", .payload = try identityEnvelopeCbor(alloc, i), }), .account => |acct| try rows.append(alloc, .{ .seq = 0, .witnessed_at = now_us, .indexed_at = 0, .kind = .account, .did = acct.did, .collection = "", .rkey = "", .rev = "", .payload = try accountEnvelopeCbor(alloc, acct), }), // #sync IS archived (kind 6) so v2 can serve it; payload is the // envelope re-encoded as CBOR (blocks = raw CAR bytes) .sync => |sy| { if (zat.Tid.parse(sy.rev) != null) { try rows.append(alloc, .{ .seq = 0, .witnessed_at = now_us, .indexed_at = 0, .kind = .sync, .did = sy.did, .collection = "", .rkey = "", .rev = sy.rev, .payload = try syncEnvelopeCbor(alloc, sy), }); } }, .info => {}, } const Boundary = struct { consumer: *Consumer, event: zat.FirehoseEvent, verdict: verify.Result, stage_cursor: bool,
fn stage(ctx: *anyopaque, _: u64, last: u64) !void { const boundary: *@This() = @ptrCast(@alignCast(ctx)); try boundary.consumer.stageEventBoundary(boundary.event, boundary.verdict, last, boundary.stage_cursor); } }; var boundary: Boundary = .{ .consumer = self, .event = event, .verdict = verdict, // Repair replay is synthetic archive work. Its original relay // frame was a verifier silent-drop, so (matching Atmos) it does // not advance the cursor by itself; a later normally emitted // upstream event advances the watermark beyond it. .stage_cursor = stage_cursor and event.seq() != null, }; const first = (try a.appendBatchBeforeDurableTraced( rows.items, now_us, &boundary, Boundary.stage, trace_context, )) orelse return .{}; return .{ .first = first, .last = first + rows.items.len - 1, .rows = try rows.toOwnedSlice(alloc), }; }
fn stageEventBoundary( self: *Consumer, event: zat.FirehoseEvent, verdict: verify.Result, local_seq: u64, stage_cursor: bool, ) !void { if (verdict.chain) |chain| if (event == .commit or event == .sync) { const did = switch (event) { .commit => |value| value.repo, .sync => |value| value.did, else => unreachable, }; const owned_did = try self.allocator.dupe(u8, did); errdefer self.allocator.free(owned_did); try self.pending_chains.append(self.allocator, .{ .local_seq = local_seq, .did = owned_did, .state = chain }); }; if (verdict.hosting) |hosting| if (event == .account) { const owned_did = try self.allocator.dupe(u8, event.account.did); errdefer self.allocator.free(owned_did); try self.pending_hosting.append(self.allocator, .{ .local_seq = local_seq, .did = owned_did, .state = hosting }); }; switch (event) { .identity => |identity| try self.stageRatchet(.identity, identity.did, identity.seq, local_seq), .account => |account| try self.stageRatchet(.account, account.did, account.seq, local_seq), else => {}, } if (stage_cursor) if (event.seq()) |seq| { if (self.stats) |stats| stats.upstream_seq.store(seq, .monotonic); try self.pending_cursors.append(self.allocator, .{ .local_seq = local_seq, .upstream_seq = seq }); }; }
pub fn deinit(self: *Consumer) void { self.pending_cursors.deinit(self.allocator); for (self.pending_chains.items) |point| self.allocator.free(point.did); self.pending_chains.deinit(self.allocator); for (self.pending_ratchets.items) |point| self.allocator.free(point.did); self.pending_ratchets.deinit(self.allocator); for (self.pending_hosting.items) |point| self.allocator.free(point.did); self.pending_hosting.deinit(self.allocator); }
pub fn close(_: *Consumer) void { log.info("upstream closed connection", .{}); }};
/// Publish one already-archived row into the hot tail. Archive payloads are/// opaque bytes; a repository can therefore contain a record that a/// subscriber encoder cannot render as AT Protocol JSON. Upstream memoizes/// that encoding error on the row and each subscriber skips it. Preserve the/// durable row and publish an empty body for the affected wire instead of/// turning remote record bytes into a process-fatal repair failure.fn appendArchivedTailRow(consumer: *Consumer, row: segment.Event) !void { // Tail.append deep-copies both wire encodings. Allocate their complete // CBOR/CID scratch graph in a per-row arena; a repair-lifetime arena must // not retain two whole-repo JSON copies on top of the CAR and MST. var wire_arena = std.heap.ArenaAllocator.init(consumer.allocator); defer wire_arena.deinit(); const alloc = wire_arena.allocator(); const json = cold.encodeEvent(alloc, row) catch |err| blk: { log.warn("hot-tail v1 encode skipped row seq={d}: {s}", .{ row.seq, @errorName(err) }); if (consumer.stats) |stats| _ = stats.subscribe_encode_errors_total.fetchAdd(1, .monotonic); break :blk ""; } orelse ""; const json_v2 = cold.encodeEventV2(alloc, row) catch |err| blk: { log.warn("hot-tail v2 encode skipped row seq={d}: {s}", .{ row.seq, @errorName(err) }); if (consumer.stats) |stats| _ = stats.subscribe_encode_errors_total.fetchAdd(1, .monotonic); break :blk ""; } orelse ""; const kind: @import("../serve/wire.zig").Kind = switch (row.kind) { .create, .update, .delete, .create_resync => .commit, .identity => .identity, .account => .account, .sync => .sync, }; const skip_v1: tail_mod.SkipV1 = switch (row.kind) { .sync => .sync, .create_resync => .resync_replacement, else => .none, }; try consumer.tail.append( kind, row.did, row.collection, row.displayTime(), row.seq, json, json_v2, skip_v1, );}
const CursorPoint = struct { local_seq: u64, upstream_seq: i64 };const ChainPoint = struct { local_seq: u64, did: []u8, state: verify.ChainUpdate };const Persisted = struct { first: u64 = 0, last: u64 = 0, rows: []segment.Event = &.{},};const RatchetKind = enum { identity, account };const RatchetPoint = struct { local_seq: u64, did: []u8, upstream_seq: i64, kind: RatchetKind };const HostingPoint = struct { local_seq: u64, did: []u8, state: verify.HostingUpdate };
fn ratchetKey(buf: []u8, kind: RatchetKind, did: []const u8) ![]const u8 { return std.fmt.bufPrint(buf, "verify/{s}/{s}", .{ @tagName(kind), did });}
fn accountEnvelopeCbor(allocator: std.mem.Allocator, acct: anytype) ![]u8 { // keys length-first: did(3) seq(3) time(4) active(6) status(6) var entries: std.ArrayList(zat.cbor.Value.MapEntry) = .empty; defer entries.deinit(allocator); try entries.append(allocator, .{ .key = "did", .value = .{ .text = acct.did } }); try entries.append(allocator, .{ .key = "seq", .value = .{ .unsigned = @intCast(@max(acct.seq, 0)) } }); try entries.append(allocator, .{ .key = "time", .value = .{ .text = acct.time } }); try entries.append(allocator, .{ .key = "active", .value = .{ .boolean = acct.active } }); if (acct.status) |st| try entries.append(allocator, .{ .key = "status", .value = .{ .text = @tagName(st) } }); return zat.cbor.encodeAlloc(allocator, .{ .map = entries.items });}
fn syncEnvelopeCbor(allocator: std.mem.Allocator, sy: anytype) ![]u8 { // keys length-first: did(3) rev(3) seq(3) time(4) blocks(6) const entries = [_]zat.cbor.Value.MapEntry{ .{ .key = "did", .value = .{ .text = sy.did } }, .{ .key = "rev", .value = .{ .text = sy.rev } }, .{ .key = "seq", .value = .{ .unsigned = @intCast(@max(sy.seq, 0)) } }, .{ .key = "time", .value = .{ .text = sy.time } }, .{ .key = "blocks", .value = .{ .bytes = sy.blocks } }, }; return zat.cbor.encodeAlloc(allocator, .{ .map = &entries });}
fn identityEnvelopeCbor(allocator: std.mem.Allocator, identity: anytype) ![]u8 { // Canonical length-first order: did(3), seq(3), time(4), handle(6). // `handle` is optional on the firehose, but when present it is part of the // durable upstream envelope. Dropping it makes replay observably different // and loses a real identity transition. var entries: std.ArrayList(zat.cbor.Value.MapEntry) = .empty; defer entries.deinit(allocator); try entries.append(allocator, .{ .key = "did", .value = .{ .text = identity.did } }); try entries.append(allocator, .{ .key = "seq", .value = .{ .unsigned = @intCast(@max(identity.seq, 0)) } }); try entries.append(allocator, .{ .key = "time", .value = .{ .text = identity.time } }); if (identity.handle) |handle| try entries.append(allocator, .{ .key = "handle", .value = .{ .text = handle } }); return zat.cbor.encodeAlloc(allocator, .{ .map = entries.items });}
test "identity durable envelope preserves optional handle" { const testing = std.testing; var arena = std.heap.ArenaAllocator.init(testing.allocator); defer arena.deinit(); const alloc = arena.allocator(); const with_handle = try identityEnvelopeCbor(alloc, zat.firehose.IdentityEvent{ .seq = 7, .did = "did:plc:identity-envelope", .time = "2024-09-09T19:46:02.102Z", .handle = "alice.test", }); const decoded = try zat.cbor.decodeAll(alloc, with_handle); try testing.expectEqualStrings("alice.test", decoded.getString("handle").?);
const absent = try identityEnvelopeCbor(alloc, zat.firehose.IdentityEvent{ .seq = 8, .did = "did:plc:identity-envelope", .time = "2024-09-09T19:46:03.102Z", }); const decoded_absent = try zat.cbor.decodeAll(alloc, absent); try testing.expect(decoded_absent.getString("handle") == null);}
test "repair hot tail skips an undecodable archived record without failing ingest" { const testing = std.testing; var threaded: Io.Threaded = .init(testing.allocator, .{}); defer threaded.deinit(); const io = threaded.io(); var tail = tail_mod.Tail.init(testing.allocator, io, 1 << 20); defer tail.deinit(); var stats: metrics.Stats = .{}; var consumer: Consumer = .{ .allocator = testing.allocator, .io = io, .tail = &tail, .upstream = "ws://offline.invalid", .stats = &stats, }; defer consumer.deinit();
// A complete repository can contain opaque record bytes outside // DAG-CBOR. This exact float decoder error killed the experiment while a // verifier repair was filling the hot tail from its fetched repository. const float_record = "\xa1\x61x\xfb\x3f\xf0\x00\x00\x00\x00\x00\x00"; try appendArchivedTailRow(&consumer, .{ .seq = 42, .witnessed_at = 100, .indexed_at = 0, .kind = .create_resync, .did = "did:plc:float-record", .collection = "app.bsky.feed.post", .rkey = "3l3qo2vuowo2b", .rev = "3l3qo2vutsw2b", .payload = float_record, });
try testing.expectEqual(@as(u64, 1), stats.subscribe_encode_errors_total.load(.monotonic)); var arena = std.heap.ArenaAllocator.init(testing.allocator); defer arena.deinit(); var index: u64 = 0; var stop: std.atomic.Value(bool) = .init(false); const result = try tail.readBatch(arena.allocator(), &index, &stop, true, 1); try testing.expectEqual(@as(usize, 1), result.batch.entries.len); try testing.expectEqual(@as(u64, 42), result.batch.entries[0].seq); try testing.expectEqual(@as(usize, 0), result.batch.entries[0].json.len); try testing.expectEqual(@as(u64, 1), index);}
/// Mirrors upstream's processBatch arm (internal/ingest/live/consumer.go):/// one warn per affected event for missing-block drops only, carrying/// seq/count/did so the originating repo is identifiable. Every other/// reason is counter-only by contract — hostile input must not drive log/// volume (docs/upstream-harness.md).fn logNewDrops(event: zat.firehose.Event, before: *const convert.Drops, after: *const convert.Drops) void { const line = missingBlockLogLine(event, before, after) orelse return; log.warn("dropped ops with missing record blocks: seq={d} count={d} did={s}", .{ line.seq, line.count, line.did });}
const MissingBlockLogLine = struct { count: u64, seq: i64, did: []const u8 };
fn missingBlockLogLine(event: zat.firehose.Event, before: *const convert.Drops, after: *const convert.Drops) ?MissingBlockLogLine { const delta = after.get(.missing_block) - before.get(.missing_block); if (delta == 0) return null; return .{ .count = delta, .seq = event.seq() orelse 0, .did = switch (event) { .commit => |c| c.repo, else => "", }, };}
// Partial CARs are spec-permitted upstream omissions, not local corruption// (upstream's events.go documents the no-op-update and non-canonical-PDS// cases). The warn must identify the originating repo, and validation drops// must stay counter-only so hostile input cannot drive log volume.test "only missing-block drops earn a log line, and it names the repo" { const testing_local = std.testing; const event: zat.firehose.Event = .{ .commit = .{ .seq = 4242, .repo = "did:plc:partialcar", .rev = "3l3qo2vutsw2b", .time = "t", .ops = &.{}, } }; var before: convert.Drops = .initFill(0); var after: convert.Drops = .initFill(0);
// validation drops: counter-only, no line after.getPtr(.invalid_rev).* = 3; after.getPtr(.invalid_collection).* = 2; try testing_local.expectEqual(null, missingBlockLogLine(event, &before, &after));
// a missing-block drop earns exactly the seq/count/did upstream logs after.getPtr(.missing_block).* = 2; const line = missingBlockLogLine(event, &before, &after).?; try testing_local.expectEqual(@as(u64, 2), line.count); try testing_local.expectEqual(@as(i64, 4242), line.seq); try testing_local.expectEqualStrings("did:plc:partialcar", line.did);
// deltas, not totals: an already-counted drop does not re-log before = after; try testing_local.expectEqual(null, missingBlockLogLine(event, &before, &after));}
test "applied identity and account ratchets commit with durable boundary" { const testing = std.testing; var threaded: Io.Threaded = .init(testing.allocator, .{}); defer threaded.deinit(); const io = threaded.io(); var tmp = testing.tmpDir(.{}); defer tmp.cleanup(); var path_buf: [std.fs.max_path_bytes]u8 = undefined; const path_len = try tmp.dir.realPath(io, &path_buf); var meta = try meta_store.Store.open(testing.allocator, path_buf[0..path_len]); defer meta.deinit(); var cursor = cursor_mod.Store.init(&meta); defer cursor.deinit(io); var tail = tail_mod.Tail.init(testing.allocator, io, 1 << 20); defer tail.deinit();
var consumer: Consumer = .{ .allocator = testing.allocator, .io = io, .tail = &tail, .upstream = "ws://unused", .cursor_store = &cursor, }; defer consumer.deinit(); try consumer.stageRatchet(.identity, "did:plc:test", 10, 1); try consumer.stageRatchet(.account, "did:plc:test", 12, 2); try testing.expectEqual(@as(i64, 10), try consumer.appliedSeq(.identity, "did:plc:test")); try consumer.commitDurableMetadata(1, 2, null); try testing.expectEqual(@as(i64, 10), try consumer.appliedSeq(.identity, "did:plc:test")); try testing.expectEqual(@as(i64, 12), try consumer.appliedSeq(.account, "did:plc:test")); try consumer.commitDurableMetadata(2, 3, null);
// Clear the in-memory promoted view: both values must reload from DB. consumer.deinit(); consumer = .{ .allocator = testing.allocator, .io = io, .tail = &tail, .upstream = "ws://unused", .cursor_store = &cursor, }; try testing.expectEqual(@as(i64, 10), try consumer.appliedSeq(.identity, "did:plc:test")); try testing.expectEqual(@as(i64, 12), try consumer.appliedSeq(.account, "did:plc:test"));}
test "verifier silent drop waits for a later emitted cursor watermark" { const testing = std.testing; var threaded: Io.Threaded = .init(testing.allocator, .{}); defer threaded.deinit(); const io = threaded.io(); var tmp = testing.tmpDir(.{}); defer tmp.cleanup(); var path_buf: [std.fs.max_path_bytes]u8 = undefined; const path_len = try tmp.dir.realPath(io, &path_buf); const root = path_buf[0..path_len]; var meta = try meta_store.Store.open(testing.allocator, root); defer meta.deinit(); var cursor = cursor_mod.Store.init(&meta); var archive = try archive_mod.Archive.init(testing.allocator, io, root); defer archive.deinit(); var tail = tail_mod.Tail.init(testing.allocator, io, 1 << 20); defer tail.deinit(); tail.enableDurabilityGate(0); var consumer: Consumer = .{ .allocator = testing.allocator, .io = io, .tail = &tail, .upstream = "ws://offline.invalid", .cursor_store = &cursor, .archive = &archive, }; defer consumer.deinit(); archive.on_durable_ctx = &consumer; archive.on_durable = Consumer.onArchiveDurable;
const trigger: zat.FirehoseEvent = .{ .commit = .{ .seq = 11, .repo = "did:plc:repair-cursor", .rev = "3l3qo2vutsw2b", .time = "2024-09-09T19:46:02.102Z", .ops = &.{}, } }; try consumer.handleEvent(trigger, .{ .outcome = .needs_resync }, false, observability.root_context); try testing.expectEqual(@as(?i64, null), try cursor.load(io));
const later: zat.FirehoseEvent = .{ .identity = .{ .seq = 12, .did = "did:plc:later-event", .time = "2024-09-09T19:46:03.102Z", .handle = "later.test", } }; try consumer.handleEvent(later, .{}, false, observability.root_context); try archive.flushBlock(); try testing.expectEqual(@as(?i64, 12), try cursor.load(io));}
test "raw firehose inspection classifies unknown and bounded error frames" { const testing = std.testing; const unknown = try testFrame(testing.allocator, .{ .map = &.{ .{ .key = "op", .value = .{ .unsigned = 1 } }, .{ .key = "t", .value = .{ .text = "#futureThing" } }, } }, .{ .map = &.{.{ .key = "seq", .value = .{ .unsigned = 2 } }} }); defer testing.allocator.free(unknown); try testing.expectEqual(@as(?i64, 2), inspectFrame(testing.allocator, unknown).unknown.seq);
// Exact schema-ordered shape used by upstream's future-frame oracle: // payload keys `seq`, then `did` are legal CBOR but not DAG-CBOR order. const schema_ordered_unknown = "\xa2\x62op\x01\x61t\x6c#futureThing" ++ "\xa2\x63seq\x03\x63did\x61x"; try testing.expectEqual(@as(?i64, 3), inspectFrame(testing.allocator, schema_ordered_unknown).unknown.seq);
const future = try testFrame(testing.allocator, .{ .map = &.{ .{ .key = "op", .value = .{ .negative = -1 } }, } }, .{ .map = &.{.{ .key = "error", .value = .{ .text = "FutureCursor" } }} }); defer testing.allocator.free(future); try testing.expectEqual(StreamErrorCode.future_cursor, inspectFrame(testing.allocator, future).stream_error);
const hostile = try testFrame(testing.allocator, .{ .map = &.{ .{ .key = "op", .value = .{ .negative = -1 } }, } }, .{ .map = &.{.{ .key = "error", .value = .{ .text = "unbounded-attacker-value" } }} }); defer testing.allocator.free(hostile); try testing.expectEqual(StreamErrorCode.other, inspectFrame(testing.allocator, hostile).stream_error);}
// Production twice exited 1 after `firehose connection closed` /// HttpConnectionClosing: a transient upstream disconnect was process-fatal.// A relay dropping us is ordinary, and reconnect is the whole point of the// loop, so run() must survive repeated abrupt disconnects and return cleanly// only once shutdown is actually requested.test "abrupt upstream disconnects reconnect instead of ending the consumer" { const testing = std.testing; var threaded: Io.Threaded = .init(testing.allocator, .{}); defer threaded.deinit(); const io = threaded.io();
var listener = try (Io.net.IpAddress{ .ip4 = .loopback(0) }).listen(io, .{ .reuse_address = true }); defer listener.deinit(io); const port = listener.socket.address.getPort();
var shutdown: std.atomic.Value(bool) = .init(false); var accepted: std.atomic.Value(u32) = .init(0);
// Accept and immediately drop: the upstream vanishing mid-handshake is // exactly the shape that killed the process. The server itself requests // shutdown once it has dropped the consumer twice, so stopping keys off a // definite event rather than a poll of a counter. const Hostile = struct { fn run( l: *Io.net.Server, loop_io: Io, seen: *std.atomic.Value(u32), stop: *std.atomic.Value(bool), ) void { while (true) { var conn = l.accept(loop_io) catch return; conn.close(loop_io); if (seen.fetchAdd(1, .release) + 1 >= 3) { stop.store(true, .release); return; } } } }; var server = try io.concurrent(Hostile.run, .{ &listener, io, &accepted, &shutdown }); defer _ = server.cancel(io);
var upstream_buf: [64]u8 = undefined; const upstream = try std.fmt.bufPrint(&upstream_buf, "ws://127.0.0.1:{d}", .{port});
var stats: metrics.Stats = .{}; var tail_instance: tail_mod.Tail = .init(testing.allocator, io, 1 << 16); defer tail_instance.deinit(); var consumer: Consumer = .{ .allocator = testing.allocator, .io = io, .tail = &tail_instance, .upstream = upstream, .cursor = null, .to_stdout = false, .stats = &stats, .shutdown = &shutdown, .emit_failure_logs = false, }; defer consumer.deinit();
// The invariant: a dropped upstream is never an error out of run(). try consumer.run(); try testing.expectEqual(@as(u32, 3), accepted.load(.acquire)); // Three connections dropped means the loop reconnected after each of the // first two rather than treating the disconnect as terminal. try testing.expect(stats.livestream_reconnects_total.load(.monotonic) >= 2);}
test "raw firehose callback propagates the pipeline fatal cause" { const testing = std.testing; var threaded: Io.Threaded = .init(testing.allocator, .{}); defer threaded.deinit(); const io = threaded.io(); const frame = try testFrame(testing.allocator, .{ .map = &.{ .{ .key = "op", .value = .{ .unsigned = 1 } }, .{ .key = "t", .value = .{ .text = "#identity" } }, } }, .{ .map = &.{ .{ .key = "did", .value = .{ .text = "did:plc:pipeline-fatal" } }, .{ .key = "seq", .value = .{ .unsigned = 42 } }, .{ .key = "time", .value = .{ .text = "2024-09-09T19:46:02.102Z" } }, } }); defer testing.allocator.free(frame);
const FailingEmitter = struct { fn emit(_: *anyopaque, _: zat.FirehoseEvent, _: verify.Result, _: observability.Context) !void { return error.TestPipelineFailure; } }; var pipe: pipeline_mod.Pipeline = .{ .allocator = testing.allocator, .io = io, .verifier = null, .emit_ctx = undefined, .emit = FailingEmitter.emit, }; try pipe.start(); defer pipe.stop(); try pipe.submit(frame); for (0..5_000) |_| { if (pipe.fatalError() != null) break; try io.sleep(.fromMilliseconds(1), .awake); } try testing.expectEqual(error.TestPipelineFailure, pipe.fatalError().?);
var tail = tail_mod.Tail.init(testing.allocator, io, 1024); defer tail.deinit(); var consumer: Consumer = .{ .allocator = testing.allocator, .io = io, .tail = &tail, .upstream = "ws://offline.invalid", .pipeline = &pipe, .emit_failure_logs = false, }; defer consumer.deinit(); try testing.expectError(error.TestPipelineFailure, consumer.onRawFrame(frame)); try testing.expectEqual(error.TestPipelineFailure, consumer.fatalError().?); try testing.expect(consumer.shouldStop());}
test "slow upstream detector trips on a trickle and backs its window off" { const testing = std.testing; const s = std.time.ns_per_s; var slow: SlowUpstream = .{ .min_rate = 50, .base_window_ns = 60 * s };
// 2026-10-01 shape: ~18 frames/s. No verdict until the window closes. var now: i96 = 0; try testing.expectEqual(null, slow.observe(now)); while (now < 59 * s) { now += s / 18; try testing.expectEqual(null, slow.observe(now)); } var verdict: ?f64 = null; while (verdict == null) { now += s / 18; verdict = slow.observe(now); } try testing.expectApproxEqAbs(@as(f64, 18), verdict.?, 0.5); try testing.expectEqual(@as(u3, 1), slow.backoff);
// The reconnect starts a fresh window, now twice as long. slow.onConnect(); const reconnected = now; try testing.expectEqual(null, slow.observe(now)); verdict = null; while (verdict == null) { now += s / 18; verdict = slow.observe(now); } try testing.expect(now - reconnected >= 120 * s); try testing.expectEqual(@as(u3, 2), slow.backoff);
// A healthy window clears the backoff. slow.onConnect(); try testing.expectEqual(null, slow.observe(now)); const healthy_until = now + 240 * s + s; while (now < healthy_until) { now += s / 400; try testing.expectEqual(null, slow.observe(now)); } try testing.expectEqual(@as(u3, 0), slow.backoff);
// The backoff saturates instead of overflowing. for (0..10) |_| { slow.onConnect(); _ = slow.observe(now); verdict = null; while (verdict == null) { now += s; verdict = slow.observe(now); } } try testing.expectEqual(SlowUpstream.max_backoff, slow.backoff);}
test "slow upstream detector is inert when unconfigured" { var slow: SlowUpstream = .{}; var now: i96 = 0; for (0..10) |_| { now += 600 * std.time.ns_per_s; try std.testing.expectEqual(null, slow.observe(now)); }}
test "slow upstream rejects the frame so the reconnect redelivers it" { const testing = std.testing; var threaded: Io.Threaded = .init(testing.allocator, .{}); defer threaded.deinit(); const io = threaded.io();
var stats: metrics.Stats = .{}; var tail = tail_mod.Tail.init(testing.allocator, io, 1024); defer tail.deinit(); var consumer: Consumer = .{ .allocator = testing.allocator, .io = io, .tail = &tail, .upstream = "ws://offline.invalid", .stats = &stats, .emit_failure_logs = false, .slow_upstream = .{ .min_rate = 1_000_000, .base_window_ns = std.time.ns_per_ms }, }; defer consumer.deinit();
consumer.onConnect("ws://offline.invalid"); try consumer.onRawFrame(""); try io.sleep(.fromMilliseconds(5), .awake); try testing.expectError(error.UpstreamTooSlow, consumer.onRawFrame("")); try testing.expectEqual(@as(u64, 1), stats.upstream_slow_reconnects_total.load(.monotonic)); // Slow is a reconnect, never a process-fatal condition. try testing.expectEqual(null, consumer.fatalError()); try testing.expect(!consumer.shouldStop());}
test "wire sequence accounting ignores replay and sizes forward gaps" { const testing = std.testing; var threaded: Io.Threaded = .init(testing.allocator, .{}); defer threaded.deinit(); var tail = tail_mod.Tail.init(testing.allocator, threaded.io(), 1024); defer tail.deinit(); var stats: metrics.Stats = .{}; var consumer: Consumer = .{ .allocator = testing.allocator, .io = threaded.io(), .tail = &tail, .upstream = "ws://offline.invalid", .cursor = 10, .stats = &stats, }; consumer.observeWireSeq(11); consumer.observeWireSeq(9); consumer.observeWireSeq(15); consumer.onReconnect();
try testing.expectEqual(@as(u64, 1), stats.livestream_sequence_gaps_total.load(.monotonic)); try testing.expectEqual(@as(u64, 3), stats.livestream_sequence_gap_missed_seqs_total.load(.monotonic)); try testing.expectEqual(@as(u64, 1), stats.livestream_reconnects_total.load(.monotonic)); try testing.expectEqual(@as(?i64, 15), consumer.last_wire_seq);}