jetstream client
atproto jetstream client
Something went wrong. Try again.
1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586878889909192939495969798991001011021031041051061071081091101111121131141151161171181191201211221231241251261271281291301311321331341351361371381391401411421431441451461471481491501511521531541551561571581591601611621631641651661671681691701711721731741751761771781791801811821831841851861871881891901911921931941951961971981992002012022032042052062072082092102112122132142152162172182192202212222232242252262272282292302312322332342352362372382392402412422432442452462472482492502512522532542552562572582592602612622632642652662672682692702712722732742752762772782792802812822832842852862872882892902912922932942952962972982993003013023033043053063073083093103113123133143153163173183193203213223233243253263273283293303313323333343353363373383393403413423433443453463473483493503513523533543553563573583593603613623633643653663673683693703713723733743753763773783793803813823833843853863873883893903913923933943953963973983994004014024034044054064074084094104114124134144154164174184194204214224234244254264274284294304314324334344354364374384394404414424434444454464474484494504514524534544554564574584594604614624634644654664674684694704714724734744754764774784794804814824834844854864874884894904914924934944954964974984995005015025035045055065075085095105115125135145155165175185195205215225235245255265275285295305315325335345355365375385395405415425435445455465475485495505515525535545555565575585595605615625635645655665675685695705715725735745755765775785795805815825835845855865875885895905915925935945955965975985996006016026036046056066076086096106116126136146156166176186196206216226236246256266276286296306316326336346356366376386396406416426436446456466476486496506516526536546556566576586596606616626636646656666676686696706716726736746756766776786796806816826836846856866876886896906916926936946956966976986997007017027037047057067077087097107117127137147157167177187197207217227237247257267277287297307317327337347357367377387397407417427437447457467477487497507517527537547557567577587597607617627637647657667677687697707717727737747757767777787797807817827837847857867877887897907917927937947957967977987998008018028038048058068078088098108118128138148158168178188198208218228238248258268278288298308318328338348358368378388398408418428438448458468478488498508518528538548558568578588598608618628638648658668678688698708718728738748758768778788798808818828838848858868878888898908918928938948958968978988999009019029039049059069079089099109119129139149159169179189199209219229239249259269279289299309319329339349359369379389399409419429439449459469479489499509519529539549559569579589599609619629639649659669679689699709719729739749759769779789799809819829839849859869879889899909919929939949959969979989991000100110021003100410051006100710081009101010111012101310141015101610171018101910201021102210231024102510261027102810291030103110321033103410351036103710381039104010411042104310441045104610471048104910501051105210531054105510561057105810591060106110621063106410651066106710681069107010711072107310741075107610771078107910801081108210831084108510861087108810891090109110921093109410951096109710981099110011011102110311041105110611071108110911101111111211131114111511161117111811191120112111221123112411251126112711281129113011311132113311341135113611371138113911401141114211431144114511461147114811491150115111521153115411551156115711581159116011611162116311641165116611671168116911701171117211731174117511761177117811791180118111821183118411851186118711881189119011911192119311941195119611971198119912001201120212031204120512061207120812091210121112121213121412151216121712181219122012211222122312241225122612271228122912301231123212331234123512361237123812391240124112421243124412451246124712481249125012511252125312541255125612571258125912601261126212631264126512661267126812691270127112721273127412751276127712781279128012811282128312841285128612871288128912901291129212931294129512961297129812991300130113021303130413051306130713081309131013111312131313141315131613171318131913201321132213231324132513261327132813291330133113321333133413351336133713381339134013411342134313441345134613471348134913501351135213531354135513561357135813591360136113621363136413651366136713681369137013711372137313741375137613771378137913801381138213831384138513861387138813891390139113921393139413951396139713981399140014011402140314041405140614071408140914101411141214131414141514161417141814191420142114221423142414251426142714281429143014311432143314341435143614371438143914401441144214431444144514461447144814491450145114521453145414551456145714581459146014611462146314641465146614671468146914701471147214731474147514761477147814791480148114821483148414851486148714881489149014911492149314941495149614971498149915001501150215031504150515061507150815091510151115121513151415151516151715181519152015211522152315241525152615271528152915301531153215331534153515361537153815391540154115421543154415451546154715481549155015511552155315541555155615571558155915601561156215631564156515661567156815691570157115721573157415751576157715781579158015811582158315841585158615871588158915901591159215931594159515961597159815991600160116021603160416051606160716081609161016111612161316141615161616171618161916201621162216231624162516261627162816291630163116321633163416351636163716381639164016411642164316441645164616471648164916501651165216531654165516561657165816591660166116621663166416651666166716681669167016711672//! archive backfill — replay a jetstream archive through the live-tail handler//!//! consumes the network.bsky.jetstream archive API (planSnapshot / getBlock ///! getSegment; planSnapshot was named planBackfill before jetstream's//! proposal-0015 rename) served by archiving jetstream instances such as//! stream.waow.tech: plan → fetch sealed jss v1 segment blocks → decompress//! (plain zstd, no dictionary) → columnar decode → deliver each matching row//! as a jetstream.Event through the same onEvent handler the live client uses.//!//! after `run` returns, connect the live client with a cursor at or before//! the returned `last_time_us` (or the plan boundary) — overlap is fine for//! idempotent consumers; delivery is at-least-once. the plan covers SEALED//! segments only: the newest ~one segment of events is unsealed, so a//! complete drain is always archive pass + live replay from the boundary.//!//! scope (measured 2026-08-07): retrieval is block-granular, so an UNBOUNDED//! plan over a sparse collection reads ~1,300x more than it keeps//! (streamplace chat: 40 GB fetched for ~30 MB of rows; one block per ~4096//! events regardless of how few match). two ways that cost disappears://! - bound the plan: `after_seq`/`before_seq` (see fetchSeqBounds for//! mapping a witnessed-time window to bounds — a two-hour window planned//! ~25 blocks when measured 2026-08-12)//! - or, for a sparse collection's CURRENT records only, use a collection//! directory (lightrail.microcosm.blue) + per-repo listRecords — though//! that cannot express deletes or event ordering.//! whole-segment mode buffers one full segment in memory (~277 MB);//! block mode buffers one block.//!//! jss v1 format: https://tangled.org/zat.dev/stream → docs/jss-format-v1.md
const std = @import("std");const zat = @import("zat");const jetstream = zat.jetstream;const sync = zat.firehose; // re-exports sync.CommitActionconst cbor = zat.cbor;const zstd = @import("zstd.zig");const livedecode = @import("livedecode.zig");const filter = @import("filter.zig");const multibase = zat.multibase;const HttpTransport = zat.HttpTransport;
const mem = std.mem;const json = std.json;const Allocator = mem.Allocator;const Io = std.Io;const log = std.log.scoped(.zat);
pub const Event = jetstream.Event;
pub const Options = struct { /// base URL of an archiving jetstream instance, e.g. "https://stream.waow.tech" host: []const u8, /// collections to replay; empty = all collections: []const []const u8 = &.{}, /// plan-level kind pruning (upstream WithKinds -> plan request /// "kinds"). without it the plan admits marker/sentinel blocks — /// measured 83x more blocks for a network-wide collection slice. /// empty = all kinds. kinds: []const livedecode.Kind = &.{}, /// dids to replay; empty = all dids: []const []const u8 = &.{}, /// bearer credential for token-gated archives (Bluesky's hosted /// instances 401 archive endpoints without one; the live tail never /// needs it). mirrors the Go client's WithAPIKey. sent as /// "Authorization: Bearer <key>" on plan/list/segment/block requests. api_key: ?[]const u8 = null, /// exclusive lower seq bound for the plan; null = from the beginning. /// derive from a time window with fetchSeqBounds. after_seq: ?u64 = null, /// inclusive upper seq bound for the plan; null = to the sealed tip before_seq: ?u64 = null, /// stop after decoding this many blocks. a full plan can span tens of GB; /// use this to bound a partial or exploratory run. when the limit stops /// the run early, the plan is not fully covered — do not treat /// `planned_through_seq` as a completed-backfill boundary. max_blocks: ?u64 = null, /// resume a prior run: skip everything up to and including this position /// (take it from the prior Result.position). segment indices are stable /// for a given archive, so a re-plan covers at least the same range. start_after: ?Position = null, /// Decode each commit's record CBOR into Event.record. Like upstream's /// raw-record mode, false skips decoding and leaves record null — for consumers /// that only want (seq, did, collection, rkey, op), it avoids the cost /// and means one undecodable stored payload cannot stop the sweep. decode_records: bool = true, /// verify each whole-mode segment's jss self-checksum (xxh3 over header /// and footer, cross-checked against the plan's checksum) before decoding. /// off by default. per-block zstd content checksums are always verified /// by libzstd regardless of this flag. verify_checksums: bool = false, /// block fetches kept in flight. serial fetch is round-trip-bound (~0.2 /// MB/s at 900ms RTT measured 2026-08-08; 8 in flight was 7.2x faster), /// so WAN consumers want 4-8. delivery order is unaffected: events /// always arrive in plan order. stay modest — each slot holds its own /// connection, and politeness is the caller's job. clamped to [1, 16]. concurrency: usize = 1,};
/// a block's place in the plan; Result.position reports the last decoded onepub const Position = struct { segment_index: u32, block_index: u32,};
pub const Result = struct { /// archive coverage boundary from the plan. resume the live client with a /// cursor at or before the timestamp of the last delivered event. planned_through_seq: u64, sealed_tip_seq: u64, /// true when an extended row handler (onRow) returned false mid-sweep; /// the plan was abandoned cleanly at that point stopped: bool = false, events_delivered: u64 = 0, blocks_decoded: u64 = 0, /// compressed bytes fetched over the wire (getBlock frames and /// getSegment bodies), for IO accounting bytes_fetched: u64 = 0, /// true when Options.max_blocks stopped the run before the plan was exhausted truncated: bool = false, /// last decoded block; feed to Options.start_after to resume position: ?Position = null, /// witnessed_at of the last delivered event, in the same µs domain as the /// live client's cursor. null if nothing matched. last_time_us: ?i64 = null,};
/// replay the archive through `handler` (same contract as JetstreamClient:/// `fn onEvent(*H, Event) void`). event slices are only valid during the/// onEvent call. Pages through the sealed tip observed on the first plan,/// unless the handler stops or max_blocks is reached. PlanStalled means/// the server returned a continuation that cannot advance the sweep./// Optional onError(anyerror) bool accepts a recoverable archive error with/// true or stops cleanly with false. Without it, the error is returned.pub fn run(io: Io, allocator: Allocator, options: Options, handler: anytype) !Result { var page_options = options; var pinned_tip: ?u64 = null; var result: Result = .{ .planned_through_seq = options.after_seq orelse 0, .sealed_tip_seq = 0 }; while (true) { if (options.max_blocks) |max| { if (result.blocks_decoded >= max) { result.truncated = true; return result; } page_options.max_blocks = max - result.blocks_decoded; } const page = try runPage(io, allocator, page_options, handler); if (pinned_tip == null) pinned_tip = page.sealed_tip_seq; result.sealed_tip_seq = pinned_tip.?; result.planned_through_seq = page.planned_through_seq; result.events_delivered += page.events_delivered; result.blocks_decoded += page.blocks_decoded; result.bytes_fetched += page.bytes_fetched; if (page.position) |position| result.position = position; if (page.last_time_us) |time| result.last_time_us = time; result.stopped = page.stopped; result.truncated = page.truncated; if (page.stopped or page.truncated or page.planned_through_seq >= pinned_tip.?) return result; if (page.planned_through_seq <= (page_options.after_seq orelse 0)) return error.PlanStalled; page_options.after_seq = page.planned_through_seq; // Snapshot membership is fixed by page one, even if the server seals // more data while downloading. The live tail covers that later data. page_options.before_seq = pinned_tip; }}
fn runPage(io: Io, allocator: Allocator, options: Options, handler: anytype) !Result { var transport = HttpTransport.init(io, allocator); defer transport.deinit();
const authorization = try bearerHeader(allocator, options.api_key); defer if (authorization) |a| allocator.free(a);
var plan_arena = std.heap.ArenaAllocator.init(allocator); defer plan_arena.deinit();
const plan = try fetchPlan(plan_arena.allocator(), &transport, options, authorization);
var result: Result = .{ .planned_through_seq = plan.planned_through_seq, .sealed_tip_seq = plan.sealed_tip_seq, };
const concurrency = @min(@max(options.concurrency, 1), max_concurrency); var window = try FetchWindow.init(io, allocator, concurrency); defer window.deinit(); window.authorization = authorization;
for (plan.segments) |segment| { if (segment.blocks) |ranges| { for (ranges) |range| { var bi = range.first; while (bi <= range.last) : (bi += 1) { if (skippedByResume(options.start_after, segment.index, bi)) continue; if (limitReached(options, &result)) { window.discardAll(); return result; } if (window.isFull()) window.deliverOldest(allocator, options, handler, &result) catch |err| { window.discardAll(); if (err == error.Stopped) { result.stopped = true; return result; } return err; }; window.start(options, segment.name, .{ .segment_index = segment.index, .block_index = bi, }) catch |err| { window.discardAll(); return err; }; } } } else { if (options.start_after) |sa| if (segment.index < sa.segment_index) continue; // whole-segment fetches are serial; flush pending blocks first so // delivery stays in plan order window.deliverAll(allocator, options, handler, &result) catch |err| { window.discardAll(); if (err == error.Stopped) { result.stopped = true; return result; } return err; }; if (limitReached(options, &result)) return result; const url = try std.fmt.allocPrint( allocator, "{s}/xrpc/network.bsky.jetstream.getSegment?name={s}", .{ options.host, segment.name }, ); defer allocator.free(url); var fetched = fetchWithRetry(io, &transport, url, authorization, .segment) catch |err| { reportArchiveError(handler, err) catch |reported| { if (reported != error.Stopped) return reported; result.stopped = true; return result; }; continue; }; defer fetched.deinit(allocator); result.bytes_fetched += fetched.body.len; deliverSegment(allocator, fetched.body, options, segment, handler, &result) catch |err| { reportArchiveError(handler, err) catch |reported| { if (reported != error.Stopped) return reported; result.stopped = true; return result; }; }; } }
window.deliverAll(allocator, options, handler, &result) catch |err| { window.discardAll(); if (err == error.Stopped) { result.stopped = true; return result; } return err; }; return result;}
const max_concurrency = 16;
const FetchError = anyerror;
/// No callback means no permission to skip data. Cancellation and allocator/// failure are not corrupt archive entries and must never become recoverable.fn reportArchiveError(handler: anytype, err: anyerror) !void { switch (err) { error.Canceled, error.OutOfMemory, error.Stopped => return err, else => {}, } if (comptime @hasDecl(@TypeOf(handler.*), "onError")) { const decision = handler.onError(err); // Legacy archive handlers only log errors. Keep them source-compatible // without treating a notification as permission to skip an entry. if (@TypeOf(decision) == void) return err; const accepted = if (@TypeOf(decision) == bool) decision else try decision; if (!accepted) return error.Stopped; } else return err;}
fn fetchJob(io: Io, transport: *HttpTransport, url: []const u8, authorization: ?[]const u8) FetchError!HttpTransport.FetchResult { return fetchWithRetry(io, transport, url, authorization, .xrpc);}
/// a FIFO window of in-flight block fetches. starting is out-of-order I/O;/// delivery always awaits the oldest first, so handler order equals plan/// order regardless of concurrency. each slot owns a transport (one/// connection) reused as the window slides.const FetchWindow = struct { io: Io, allocator: Allocator, transports: []HttpTransport, slots: []Inflight, /// borrowed from run(); outlives every in-flight fetch authorization: ?[]const u8 = null, head: usize = 0, count: usize = 0, failed_segment: ?u32 = null,
const Inflight = struct { future: std.Io.Future(FetchError!HttpTransport.FetchResult), url: []u8, position: Position, };
fn init(io: Io, allocator: Allocator, concurrency: usize) !FetchWindow { const transports = try allocator.alloc(HttpTransport, concurrency); errdefer allocator.free(transports); const slots = try allocator.alloc(Inflight, concurrency); for (transports) |*t| t.* = HttpTransport.init(io, allocator); return .{ .io = io, .allocator = allocator, .transports = transports, .slots = slots }; }
fn deinit(self: *FetchWindow) void { self.discardAll(); for (self.transports) |*t| t.deinit(); self.allocator.free(self.transports); self.allocator.free(self.slots); }
fn isFull(self: *const FetchWindow) bool { return self.count == self.slots.len; }
fn start(self: *FetchWindow, options: Options, segment_name: []const u8, position: Position) !void { if (self.failed_segment == position.segment_index) return; std.debug.assert(!self.isFull()); const url = try std.fmt.allocPrint( self.allocator, "{s}/xrpc/network.bsky.jetstream.getBlock?segment={s}&blockIndex={d}", .{ options.host, segment_name, position.block_index }, ); errdefer self.allocator.free(url); const slot = (self.head + self.count) % self.slots.len; self.slots[slot] = .{ .future = self.io.async(fetchJob, .{ self.io, &self.transports[slot], url, self.authorization }), .url = url, .position = position, }; self.count += 1; }
fn deliverOldest(self: *FetchWindow, allocator: Allocator, options: Options, handler: anytype, result: *Result) !void { std.debug.assert(self.count > 0); const slot = &self.slots[self.head]; self.head = (self.head + 1) % self.slots.len; self.count -= 1; defer self.allocator.free(slot.url); if (self.failed_segment == slot.position.segment_index) { if (slot.future.cancel(self.io)) |fetched| { var discarded = fetched; discarded.deinit(allocator); } else |_| {} return; } var fetched = slot.future.await(self.io) catch |err| { self.failed_segment = slot.position.segment_index; return reportArchiveError(handler, err); }; defer fetched.deinit(allocator); result.bytes_fetched += fetched.body.len; deliverFrame(allocator, fetched.body, options, slot.position, handler, result) catch |err| { try reportArchiveError(handler, err); }; }
/// drain in order, stopping (and discarding the rest) if max_blocks hits fn deliverAll(self: *FetchWindow, allocator: Allocator, options: Options, handler: anytype, result: *Result) !void { while (self.count > 0) { if (limitReached(options, result)) { self.discardAll(); return; } try self.deliverOldest(allocator, options, handler, result); } }
/// Cancel and release pending requests when the caller stops or a budget /// is reached; cleanup must not wait through unused retry schedules. fn discardAll(self: *FetchWindow) void { while (self.count > 0) { const slot = &self.slots[self.head]; self.head = (self.head + 1) % self.slots.len; self.count -= 1; if (slot.future.cancel(self.io)) |fetched| { var discarded = fetched; discarded.deinit(self.allocator); } else |_| {} self.allocator.free(slot.url); } }};
/// decode every block of a sealed in-memory jss segment through `handler` —/// the offline counterpart of `run` for a segment already on disk (local/// archive copies, getSegment bodies fetched out-of-band). same delivery/// contract; Options.host is unused. honors collections/dids/max_blocks//// verify_checksums.pub fn replaySegment(allocator: Allocator, body: []const u8, options: Options, handler: anytype) !Result { var result: Result = .{ .planned_through_seq = 0, .sealed_tip_seq = 0 }; try deliverSegment(allocator, body, options, .{ .name = "", .index = 0, .checksum = null, .blocks = null, }, handler, &result); return result;}
/// resume skip: everything at or before start_after is already deliveredfn skippedByResume(start_after: ?Position, segment_index: u32, block_index: u32) bool { const sa = start_after orelse return false; if (segment_index < sa.segment_index) return true; return segment_index == sa.segment_index and block_index <= sa.block_index;}
const RetryKind = enum { xrpc, segment };fn retryDelay(kind: RetryKind, attempt: u32, hint_ns: u64, jitter: f64) u64 { const cap = 30 * std.time.ns_per_s; const base = @as(u64, 500 * std.time.ns_per_ms) << @as(u6, @intCast(@min(attempt - 1, 6))); const scheduled: u64 = @min(cap, if (kind == .xrpc) @as(u64, @intFromFloat(@as(f64, @floatFromInt(base)) * (1 + 0.2 * jitter))) else base); if (kind == .xrpc and hint_ns > 0) return @min(cap, hint_ns); return @min(cap, @max(scheduled, hint_ns));}
fn fillRandom(io: *Io, bytes: []u8) void { io.random(bytes);}
const fetch_attempts = 3;
/// Match the Go bulk-segment and XRPC retry policies separately.fn fetchWithRetry(io: Io, transport: *HttpTransport, url: []const u8, authorization: ?[]const u8, kind: RetryKind) !HttpTransport.FetchResult { const previous_attempts = transport.dead_connection_attempts; transport.dead_connection_attempts = 1; defer transport.dead_connection_attempts = previous_attempts; var random_io = io; const random = std.Random.init(&random_io, fillRandom); var attempt: u32 = 0; while (true) { attempt += 1; var hint_ns: u64 = 0; if (transport.fetch(.{ .url = url, .authorization = authorization })) |result| { if (result.status == .ok) return result; var discarded = result; const code = @backingInt(result.status); hint_ns = retryHint(result.rate_limit, Io.Clock.real.now(io).toNanoseconds()); discarded.deinit(transport.allocator); const retryable = if (kind == .segment) code >= 500 or code == 429 else code == 429 or code == 500 or code == 502 or code == 503 or code == 504; if (!retryable) return error.FetchFailed; if (attempt >= fetch_attempts) return error.FetchRetriesExhausted; } else |err| { if (err == error.Canceled or err == error.OutOfMemory) return err; if (attempt >= fetch_attempts) return err; } const jitter = if (kind == .xrpc) random.float(f64) else 0; const delay = retryDelay(kind, attempt, hint_ns, jitter); try io.sleep(Io.Duration.fromNanoseconds(@intCast(delay)), .awake); }}
fn retryHint(headers: HttpTransport.RateLimitHeaders, now_ns: i96) u64 { const nanoseconds: i128 = if (headers.reset) |reset| @as(i128, reset) * std.time.ns_per_s - now_ns else if (headers.retry_after) |delta| @as(i128, delta) * std.time.ns_per_s else if (headers.retry_after_at) |date| @as(i128, date) * std.time.ns_per_s - now_ns else 0; return @intCast(@min(@max(nanoseconds, 0), 30 * std.time.ns_per_s));}
fn limitReached(options: Options, result: *Result) bool { const max = options.max_blocks orelse return false; if (result.blocks_decoded < max) return false; result.truncated = true; return true;}
// === plan ===
const BlockRange = struct { first: u32, last: u32 };
const PlannedSegment = struct { name: []const u8, index: u32, /// jss self-checksum from the plan (16 hex chars on the wire); null if absent checksum: ?u64, /// null = whole-segment mode blocks: ?[]const BlockRange,};
const Plan = struct { planned_through_seq: u64, sealed_tip_seq: u64, segments: []const PlannedSegment,};
fn planRequestBody(arena: Allocator, options: Options) ![]const u8 { var body: std.Io.Writer.Allocating = .init(arena); var stringify: json.Stringify = .{ .writer = &body.writer }; try stringify.beginObject(); try stringify.objectField("collections"); try stringify.beginArray(); for (options.collections) |c| try stringify.write(c); try stringify.endArray(); try stringify.objectField("dids"); try stringify.beginArray(); for (options.dids) |d| try stringify.write(d); try stringify.endArray(); if (options.after_seq) |seq| { try stringify.objectField("afterSeq"); try stringify.write(seq); } if (options.before_seq) |seq| { try stringify.objectField("beforeSeq"); try stringify.write(seq); } if (options.kinds.len > 0) { try stringify.objectField("kinds"); try stringify.beginArray(); for (options.kinds) |k| try stringify.write(@tagName(k)); try stringify.endArray(); } try stringify.endObject(); return body.written();}
fn bearerHeader(allocator: Allocator, api_key: ?[]const u8) !?[]u8 { const key = api_key orelse return null; return try std.fmt.allocPrint(allocator, "Bearer {s}", .{key});}
fn fetchPlan(arena: Allocator, transport: *HttpTransport, options: Options, authorization: ?[]const u8) !Plan { const body = try planRequestBody(arena, options); const url = try std.fmt.allocPrint(arena, "{s}/xrpc/network.bsky.jetstream.planSnapshot", .{options.host}); var fetched = try transport.fetch(.{ .url = url, .method = .POST, .payload = body, .content_type = "application/json", .authorization = authorization, }); defer fetched.deinit(transport.allocator); if (fetched.status != .ok) { log.warn("planSnapshot failed: {d} {s}", .{ @backingInt(fetched.status), fetched.body }); return error.PlanFailed; } return parsePlan(arena, fetched.body);}
// === time window → seq bounds ===
/// one sealed segment's metadata row from listSegments, reduced to what/// bounds derivation needspub const SegmentSpan = struct { min_seq: u64, max_seq: u64, min_witnessed_us: i64, max_witnessed_us: i64,};
pub const SeqBounds = struct { after_seq: ?u64, before_seq: ?u64, /// false when no sealed segment's witnessed window intersects the /// requested window — the window predates the archive's live capture or /// lies entirely in the unsealed tail. plan nothing; go straight to the /// live client (or drop the bounds for a full historical scan). covered: bool,};
/// reduce a witnessed-time window to plan seq bounds: the bounds admit every/// segment whose witnessed span intersects [start_us, end_us]. sound for/// live-captured rows (an event is witnessed at or after it happened);/// bootstrap-merged rows carry bootstrap-era witnessed times, so windows/// older than the archive's live capture report covered=false rather than/// pretending.pub fn seqBoundsForWindow(spans: []const SegmentSpan, start_us: i64, end_us: ?i64) SeqBounds { var after: ?u64 = null; var before: ?u64 = null; var covered = false; for (spans) |span| { const ends_before = span.max_witnessed_us < start_us; const starts_after = if (end_us) |e| span.min_witnessed_us > e else false; if (ends_before or starts_after) continue; covered = true; if (after == null or span.min_seq -| 1 < after.?) after = span.min_seq -| 1; if (before == null or span.max_seq > before.?) before = span.max_seq; } // an unbounded end means "through the sealed tip": drop the upper bound if (end_us == null) before = null; return .{ .after_seq = after, .before_seq = before, .covered = covered };}
/// re-anchor point for entering a DIFFERENT instance's seq space from a/// witnessed-time floor (cross-instance failover: seqs don't transfer,/// witnessed time approximately does). `after_seq` is where the sweep/// starts on the new host; rows older than the floor are the caller's to/// trim (segment granularity is coarse — hours — so the seq bound decides/// what to FETCH and the time floor decides what to DELIVER).pub const Anchor = struct { after_seq: ?u64, /// false when no sealed segment reaches the floor: the window lives in /// the unsealed tail (after_seq = sealed tip, live cutover covers the /// rest) or the host has no sealed archive at all (after_seq = null, /// start live from the tip). covered: bool,};
/// reduce sealed spans to a failover anchor for `floor_us`: the sweep must/// start at or before the first row witnessed >= floor. covered → one seq/// below the earliest intersecting segment; not covered but spans exist →/// the sealed tip (the floor lies in the unsealed tail; the live cutover's/// server-side cold replay covers it); no spans → null (live from tip).pub fn anchorForFloor(spans: []const SegmentSpan, floor_us: i64) Anchor { const bounds = seqBoundsForWindow(spans, floor_us, null); if (bounds.covered) return .{ .after_seq = bounds.after_seq, .covered = true }; var tip: ?u64 = null; for (spans) |span| { if (tip == null or span.max_seq > tip.?) tip = span.max_seq; } return .{ .after_seq = tip, .covered = false };}
/// fetch every sealed segment's metadata from `host` into `arena`pub fn fetchSegmentSpans(io: Io, allocator: Allocator, arena: Allocator, host: []const u8, api_key: ?[]const u8) ![]SegmentSpan { var transport = HttpTransport.init(io, allocator); defer transport.deinit(); const authorization = try bearerHeader(allocator, api_key); defer if (authorization) |a| allocator.free(a);
var spans: std.ArrayList(SegmentSpan) = .empty; var cursor: ?[]const u8 = null; while (true) { const url = if (cursor) |c| try std.fmt.allocPrint(arena, "{s}/xrpc/network.bsky.jetstream.listSegments?limit=1000&cursor={s}", .{ host, c }) else try std.fmt.allocPrint(arena, "{s}/xrpc/network.bsky.jetstream.listSegments?limit=1000", .{host}); var fetched = try fetchWithRetry(io, &transport, url, authorization, .xrpc); defer fetched.deinit(allocator); const parsed = try json.parseFromSliceLeaky(json.Value, arena, fetched.body, .{}); const root = switch (parsed) { .object => |o| o, else => return error.MalformedSegmentList, }; const segments = switch (root.get("segments") orelse return error.MalformedSegmentList) { .array => |a| a.items, else => return error.MalformedSegmentList, }; if (segments.len == 0) break; for (segments) |seg| { try spans.append(arena, .{ .min_seq = try getJsonU64(seg, "minSeq"), .max_seq = try getJsonU64(seg, "maxSeq"), .min_witnessed_us = @intCast(try getJsonU64(seg, "minWitnessedAt")), .max_witnessed_us = @intCast(try getJsonU64(seg, "maxWitnessedAt")), }); } // a short page is the end even if a cursor is present if (segments.len < 1000) break; const next = root.get("cursor") orelse break; cursor = switch (next) { .string => |s| try arena.dupe(u8, s), .integer => |i| try std.fmt.allocPrint(arena, "{d}", .{i}), else => break, }; } return spans.items;}
/// fetch every sealed segment's metadata from `host` and derive the seq/// bounds covering the witnessed window [start_us, end_us]. end_us == null/// means "through the sealed tip".pub fn fetchSeqBounds(io: Io, allocator: Allocator, host: []const u8, start_us: i64, end_us: ?i64, api_key: ?[]const u8) !SeqBounds { var arena_state = std.heap.ArenaAllocator.init(allocator); defer arena_state.deinit(); const spans = try fetchSegmentSpans(io, allocator, arena_state.allocator(), host, api_key); return seqBoundsForWindow(spans, start_us, end_us);}
/// fetch segment metadata from `host` and derive the failover anchor for/// `floor_us` (see anchorForFloor)pub fn fetchAnchor(io: Io, allocator: Allocator, host: []const u8, floor_us: i64, api_key: ?[]const u8) !Anchor { var arena_state = std.heap.ArenaAllocator.init(allocator); defer arena_state.deinit(); const spans = try fetchSegmentSpans(io, allocator, arena_state.allocator(), host, api_key); return anchorForFloor(spans, floor_us);}
fn parsePlan(arena: Allocator, body: []const u8) !Plan { const parsed = try json.parseFromSliceLeaky(json.Value, arena, body, .{}); const root = switch (parsed) { .object => |o| o, else => return error.MalformedPlan, };
const segments_val = root.get("segments") orelse return error.MalformedPlan; const segments_json = switch (segments_val) { .array => |a| a.items, else => return error.MalformedPlan, };
const segments = try arena.alloc(PlannedSegment, segments_json.len); for (segments_json, segments) |seg_val, *out| { const seg = switch (seg_val) { .object => |o| o, else => return error.MalformedPlan, }; const name = switch (seg.get("name") orelse return error.MalformedPlan) { .string => |s| s, else => return error.MalformedPlan, }; const mode = switch (seg.get("mode") orelse return error.MalformedPlan) { .string => |s| s, else => return error.MalformedPlan, }; var blocks: ?[]const BlockRange = null; if (mem.eql(u8, mode, "blocks")) { const ranges_json = switch (seg.get("blocks") orelse return error.MalformedPlan) { .array => |a| a.items, else => return error.MalformedPlan, }; const ranges = try arena.alloc(BlockRange, ranges_json.len); for (ranges_json, ranges) |range_val, *r| { r.* = .{ .first = try getJsonU32(range_val, "first"), .last = try getJsonU32(range_val, "last"), }; } blocks = ranges; } const checksum: ?u64 = if (seg.get("checksum")) |v| switch (v) { .string => |s| std.fmt.parseInt(u64, s, 16) catch null, else => null, } else null; out.* = .{ .name = name, .index = try getJsonU32(seg_val, "index"), .checksum = checksum, .blocks = blocks, }; }
return .{ .planned_through_seq = try getJsonU64(.{ .object = root }, "plannedThroughSeq"), .sealed_tip_seq = try getJsonU64(.{ .object = root }, "sealedTipSeq"), .segments = segments, };}
fn getJsonU64(val: json.Value, key: []const u8) !u64 { const obj = switch (val) { .object => |o| o, else => return error.MalformedPlan, }; return switch (obj.get(key) orelse return error.MalformedPlan) { .integer => |i| if (i < 0) error.MalformedPlan else @intCast(i), else => error.MalformedPlan, };}
fn getJsonU32(val: json.Value, key: []const u8) !u32 { return std.math.cast(u32, try getJsonU64(val, key)) orelse error.MalformedPlan;}
// === jss v1 segment walk ===
const header_size = 256;const block_index_entry_size = 52;const max_block_event_count = 1 << 18;const max_block_count = 1 << 20;
fn deliverSegment(allocator: Allocator, body: []const u8, options: Options, segment: PlannedSegment, handler: anytype, result: *Result) !void { if (body.len < header_size) return error.MalformedSegment; if (!mem.eql(u8, body[0..4], "jss0")) return error.MalformedSegment; const block_count = mem.readInt(u32, body[14..18], .little); const block_index_offset = mem.readInt(u64, body[90..98], .little); if (block_count > max_block_count) return error.MalformedSegment;
if (options.verify_checksums) try verifySegment(body, segment.checksum);
for (0..block_count) |bi| { if (skippedByResume(options.start_after, segment.index, @intCast(bi))) continue; if (limitReached(options, result)) return; const entry_off = block_index_offset + bi * block_index_entry_size; if (entry_off + block_index_entry_size > body.len) return error.MalformedSegment; const offset = mem.readInt(u64, body[entry_off..][0..8], .little); const compressed_size = mem.readInt(u32, body[entry_off + 8 ..][0..4], .little); const frame_start = offset + 8; if (frame_start + compressed_size > body.len) return error.MalformedSegment; deliverFrame(allocator, body[frame_start..][0..compressed_size], options, .{ .segment_index = segment.index, .block_index = @intCast(bi), }, handler, result) catch |err| { try reportArchiveError(handler, err); }; }}
/// verify a sealed segment's self-checksum: xxh3_64(seed 0) over/// header[12..256] ++ file[footer_offset..EOF], stored at header offset 4./// also cross-checks the plan's checksum for the segment when present.pub fn verifySegment(body: []const u8, plan_checksum: ?u64) !void { if (body.len < header_size) return error.MalformedSegment; if (!mem.eql(u8, body[0..4], "jss0")) return error.MalformedSegment; const stored = mem.readInt(u64, body[4..12], .little); if (stored == 0) return error.SegmentUnsealed; if (plan_checksum) |expected| { if (expected != stored) return error.ChecksumMismatch; } const footer_offset = mem.readInt(u64, body[58..66], .little); if (footer_offset < header_size or footer_offset > body.len) return error.MalformedSegment; var hasher = std.hash.XxHash3.init(0); hasher.update(body[12..header_size]); hasher.update(body[footer_offset..]); if (hasher.final() != stored) return error.ChecksumMismatch;}
fn deliverFrame(allocator: Allocator, compressed: []const u8, options: Options, position: Position, handler: anytype, result: *Result) !void { var arena_state = std.heap.ArenaAllocator.init(allocator); defer arena_state.deinit(); const arena = arena_state.allocator();
const raw = zstd.decompressAlloc(arena, compressed) catch |err| switch (err) { error.OutOfMemory => return error.OutOfMemory, else => return error.BlockDecompressFailed, };
try deliverBlock(arena, raw, options, handler, result); result.blocks_decoded += 1; result.position = position; // a bounded-but-large plan (a session inside the merged bootstrap // region can touch thousands of blocks) must not look like a hang if (result.blocks_decoded % 256 == 0) log.info("archive backfill: {d} blocks decoded, {d} events delivered", .{ result.blocks_decoded, result.events_delivered });}
// jss event kinds (1..7)const kind_create = 1;const kind_update = 2;const kind_delete = 3;const kind_create_resync = 7;const kind_identity = 4;const kind_account = 5;const kind_sync = 6;
/// decode one decompressed columnar block and deliver matching rows./// `arena` backs per-row record decoding; the caller resets it per block.fn deliverBlock(arena: Allocator, raw: []const u8, options: Options, handler: anytype, result: *Result) !void { if (raw.len < 4) return error.MalformedBlock; const n: usize = mem.readInt(u32, raw[0..4], .little); if (n == 0) { if (raw.len != 4) return error.MalformedBlock; return; } if (n > max_block_event_count) return error.MalformedBlock;
var o: usize = 4; const seq_base = o; o += 8 * n; // seq const wit_base = o; o += 8 * n; // witnessed_at o += 8 * n; // indexed_at (unused) const kind_base = o; o += n; const col_len_base = o; o += n; const did_len_base = o; o += 2 * n; const rkey_len_base = o; o += n; const rev_len_base = o; o += n; const pay_len_base = o; o += 4 * n; if (o > raw.len) return error.MalformedBlock;
// blob section bases from length totals var col_total: usize = 0; var did_total: usize = 0; var rkey_total: usize = 0; var rev_total: usize = 0; var pay_total: usize = 0; for (0..n) |i| { col_total += raw[col_len_base + i]; did_total += mem.readInt(u16, raw[did_len_base + 2 * i ..][0..2], .little); rkey_total += raw[rkey_len_base + i]; rev_total += raw[rev_len_base + i]; pay_total += mem.readInt(u32, raw[pay_len_base + 4 * i ..][0..4], .little); } if (o + col_total + did_total + rkey_total + rev_total + pay_total != raw.len) return error.MalformedBlock;
var col_off = o; var did_off = col_off + col_total; var rkey_off = did_off + did_total; var rev_off = rkey_off + rkey_total; var pay_off = rev_off + rev_total; var record_error: ?anyerror = null;
for (0..n) |i| { const col_len = raw[col_len_base + i]; const did_len = mem.readInt(u16, raw[did_len_base + 2 * i ..][0..2], .little); const rkey_len = raw[rkey_len_base + i]; const rev_len = raw[rev_len_base + i]; const pay_len = mem.readInt(u32, raw[pay_len_base + 4 * i ..][0..4], .little);
const collection = raw[col_off..][0..col_len]; const did = raw[did_off..][0..did_len]; const rkey = raw[rkey_off..][0..rkey_len]; const rev = raw[rev_off..][0..rev_len]; const payload = raw[pay_off..][0..pay_len]; col_off += col_len; did_off += did_len; rkey_off += rkey_len; rev_off += rev_len; pay_off += pay_len;
const kind_byte = raw[kind_base + i]; const witnessed_at_row = mem.readInt(i64, raw[wit_base + 8 * i ..][0..8], .little); const seq_row = mem.readInt(u64, raw[seq_base + 8 * i ..][0..8], .little);
// the plan prunes whole segments/blocks; the exact (after_seq, // before_seq] window is the client's to enforce per row, exactly // like upstream's matcher (filter.go: "applied on the client") if (options.after_seq) |a| if (seq_row <= a) continue; if (options.before_seq) |b| if (seq_row > b) continue;
// extended row handler (the unified client's archive half): every // matching row is delivered as a v2 event WITH its seq — including // identity/account/sync marker rows, which ride inline exactly like // upstream's backfill (a folding consumer needs the #sync tombstone // to know to refold). the classic onEvent handler below keeps its // original commits-only, seq-less contract. if (comptime @hasDecl(@TypeOf(handler.*), "onRow")) { if (options.dids.len > 0 and !containsString(options.dids, did)) continue; const row_event: ?livedecode.Event = switch (kind_byte) { kind_create, kind_create_resync, kind_update, kind_delete => blk: { if (!filter.collectionMatches(options.collections, collection)) break :blk null; const op: livedecode.Operation = switch (kind_byte) { kind_update => .update, kind_delete => .delete, else => .create, }; const rec: ?json.Value = if (options.decode_records and op != .delete and payload.len > 0) rec: { const decoded = cbor.decode(arena, payload) catch |err| { if (err == error.OutOfMemory) return err; record_error = error.MalformedRecord; continue; }; break :rec try livedecode.cborToJson(arena, decoded.value); } else null; break :blk .{ .seq = seq_row, .time_us = witnessed_at_row, .did = did, .payload = .{ .commit = .{ .operation = op, .collection = collection, .rkey = rkey, .rev = rev, .record = rec, // upstream Event.RecordCBOR on the archive path: // the stored payload verbatim, aliasing the block // buffer (same lifetime as the string fields) .record_cbor = if (op != .delete) payload else "", }, }, }; }, kind_identity, kind_account, kind_sync => blk: { // marker payloads are the archived upstream envelope as a // CBOR map; fields legitimately absent on synthetic rows // (e.g. an async-resync #sync) default to zero values, // matching upstream's generated decoder const env: ?cbor.Value = if (payload.len > 0) (cbor.decode(arena, payload) catch |err| { if (err == error.OutOfMemory) return err; record_error = error.MalformedRecord; continue; }).value else null; break :blk switch (kind_byte) { kind_identity => .{ .seq = seq_row, .time_us = witnessed_at_row, .did = did, .payload = .{ .identity = .{ .did = did, .handle = if (env) |e| e.getString("handle") else null, .seq = if (env) |e| e.getInt("seq") else null, .time = if (env) |e| e.getString("time") else null, } } }, kind_account => .{ .seq = seq_row, .time_us = witnessed_at_row, .did = did, .payload = .{ .account = .{ .did = did, .active = if (env) |e| e.getBool("active") orelse false else false, .status = if (env) |e| e.getString("status") else null, .seq = if (env) |e| e.getInt("seq") else null, .time = if (env) |e| e.getString("time") else null, } } }, else => .{ .seq = seq_row, .time_us = witnessed_at_row, .did = did, .payload = .{ .sync = .{ .did = did, .rev = if (env) |e| e.getString("rev") orelse rev else rev, .seq = if (env) |e| e.getInt("seq") else null, .time = if (env) |e| e.getString("time") else null, } } }, }; }, else => null, // unknown jss kind from a newer writer: skip }; if (row_event) |ev| { if (!handler.onRow(ev)) return error.Stopped; result.events_delivered += 1; result.last_time_us = witnessed_at_row; } continue; }
const operation: sync.CommitAction = switch (kind_byte) { kind_create, kind_create_resync => .create, kind_update => .update, kind_delete => .delete, // identity/account/sync marker rows carry no record; the live-tail // pass re-covers them, so the archive pass skips them. else => continue, };
// planned blocks can contain other collections' rows — always filter client-side if (!filter.collectionMatches(options.collections, collection)) continue; if (options.dids.len > 0 and !containsString(options.dids, did)) continue;
const record: ?json.Value = if (options.decode_records and operation != .delete and payload.len > 0) blk: { const decoded = cbor.decode(arena, payload) catch |err| { if (err == error.OutOfMemory) return err; record_error = error.MalformedRecord; continue; }; break :blk try livedecode.cborToJson(arena, decoded.value); } else null;
const witnessed_at = witnessed_at_row;
handler.onEvent(.{ .commit = .{ .did = did, .time_us = witnessed_at, .rev = if (rev_len > 0) rev else null, .operation = operation, .collection = collection, .rkey = rkey, .record = record, } }); result.events_delivered += 1; result.last_time_us = witnessed_at; } if (record_error) |err| try reportArchiveError(handler, err);}
fn containsString(haystack: []const []const u8, needle: []const u8) bool { for (haystack) |s| if (mem.eql(u8, s, needle)) return true; return false;}
// === tests ===
const testing = std.testing;
const CollectingHandler = struct { arena: Allocator, events: std.array_list.Managed(jetstream.CommitEvent),
fn init(arena: Allocator) CollectingHandler { return .{ .arena = arena, .events = .init(arena) }; }
pub fn onEvent(self: *CollectingHandler, event: Event) void { // copy string fields — block memory dies after deliverBlock returns var commit = event.commit; commit.did = self.arena.dupe(u8, commit.did) catch unreachable; commit.collection = self.arena.dupe(u8, commit.collection) catch unreachable; commit.rkey = self.arena.dupe(u8, commit.rkey) catch unreachable; commit.record = null; self.events.append(commit) catch unreachable; }};
/// build a columnar jss block from rows for tests (shared with the/// loopback e2e fixtures)pub fn buildTestBlock(arena: Allocator, rows: []const struct { seq: u64, witnessed_at: i64, kind: u8, collection: []const u8, did: []const u8, rkey: []const u8, rev: []const u8, payload: []const u8,}) ![]u8 { var out: std.Io.Writer.Allocating = .init(arena); const w = &out.writer; try w.writeInt(u32, @intCast(rows.len), .little); for (rows) |r| try w.writeInt(u64, r.seq, .little); for (rows) |r| try w.writeInt(i64, r.witnessed_at, .little); for (rows) |_| try w.writeInt(i64, 0, .little); // indexed_at for (rows) |r| try w.writeByte(r.kind); for (rows) |r| try w.writeByte(@intCast(r.collection.len)); for (rows) |r| try w.writeInt(u16, @intCast(r.did.len), .little); for (rows) |r| try w.writeByte(@intCast(r.rkey.len)); for (rows) |r| try w.writeByte(@intCast(r.rev.len)); for (rows) |r| try w.writeInt(u32, @intCast(r.payload.len), .little); for (rows) |r| try w.writeAll(r.collection); for (rows) |r| try w.writeAll(r.did); for (rows) |r| try w.writeAll(r.rkey); for (rows) |r| try w.writeAll(r.rev); for (rows) |r| try w.writeAll(r.payload); return out.written();}
// {"$type": "tech.waow.pollz.vote", "answer": 2}pub const test_record_cbor = [_]u8{ 0xa2, // map(2) 0x65, '$', 't', 'y', 'p', 'e', // "$type" 0x74, 't', 'e', 'c', 'h', '.', 'w', 'a', 'o', 'w', '.', 'p', 'o', 'l', 'l', 'z', '.', 'v', 'o', 't', 'e', 0x66, 'a', 'n', 's', 'w', 'e', 'r', // "answer" 0x02,};
test "deliverBlock decodes rows, filters collections, maps kinds" { var arena_state = std.heap.ArenaAllocator.init(testing.allocator); defer arena_state.deinit(); const arena = arena_state.allocator();
const raw = try buildTestBlock(arena, &.{ .{ .seq = 10, .witnessed_at = 1000, .kind = kind_create, .collection = "tech.waow.pollz.vote", .did = "did:plc:alice", .rkey = "r1", .rev = "rev1", .payload = &test_record_cbor }, .{ .seq = 11, .witnessed_at = 1001, .kind = kind_create, .collection = "app.bsky.feed.post", .did = "did:plc:bob", .rkey = "r2", .rev = "rev2", .payload = &test_record_cbor }, .{ .seq = 12, .witnessed_at = 1002, .kind = kind_delete, .collection = "tech.waow.pollz.vote", .did = "did:plc:carol", .rkey = "r3", .rev = "", .payload = "" }, .{ .seq = 13, .witnessed_at = 1003, .kind = kind_create_resync, .collection = "tech.waow.pollz.vote", .did = "did:plc:dave", .rkey = "r4", .rev = "rev4", .payload = &test_record_cbor }, .{ .seq = 14, .witnessed_at = 1004, .kind = 5, .collection = "$account", .did = "did:plc:eve", .rkey = "", .rev = "", .payload = "" }, });
var handler = CollectingHandler.init(arena); var result: Result = .{ .planned_through_seq = 0, .sealed_tip_seq = 0 }; try deliverBlock(arena, raw, .{ .host = "https://example.test", .collections = &.{"tech.waow.pollz.vote"}, }, &handler, &result);
try testing.expectEqual(@as(u64, 3), result.events_delivered); try testing.expectEqual(@as(i64, 1003), result.last_time_us.?);
try testing.expectEqualStrings("did:plc:alice", handler.events.items[0].did); try testing.expectEqual(sync.CommitAction.create, handler.events.items[0].operation); try testing.expectEqual(@as(i64, 1000), handler.events.items[0].time_us);
try testing.expectEqual(sync.CommitAction.delete, handler.events.items[1].operation); try testing.expectEqualStrings("did:plc:carol", handler.events.items[1].did);
// create_resync maps to create try testing.expectEqual(sync.CommitAction.create, handler.events.items[2].operation); try testing.expectEqualStrings("did:plc:dave", handler.events.items[2].did);}
test "deliverBlock decodes DAG-CBOR record into json.Value" { var arena_state = std.heap.ArenaAllocator.init(testing.allocator); defer arena_state.deinit(); const arena = arena_state.allocator();
const raw = try buildTestBlock(arena, &.{ .{ .seq = 1, .witnessed_at = 42, .kind = kind_create, .collection = "tech.waow.pollz.vote", .did = "did:plc:alice", .rkey = "r1", .rev = "rev1", .payload = &test_record_cbor }, });
const RecordHandler = struct { answer: ?i64 = null, type_ok: bool = false,
pub fn onEvent(self: *@This(), event: Event) void { const record = event.commit.record orelse return; const obj = record.object; self.type_ok = mem.eql(u8, obj.get("$type").?.string, "tech.waow.pollz.vote"); self.answer = obj.get("answer").?.integer; } };
var handler = RecordHandler{}; var result: Result = .{ .planned_through_seq = 0, .sealed_tip_seq = 0 }; try deliverBlock(arena, raw, .{ .host = "https://example.test" }, &handler, &result);
try testing.expect(handler.type_ok); try testing.expectEqual(@as(i64, 2), handler.answer.?);}
test "deliverBlock rejects trailing bytes and truncation" { var arena_state = std.heap.ArenaAllocator.init(testing.allocator); defer arena_state.deinit(); const arena = arena_state.allocator();
const raw = try buildTestBlock(arena, &.{ .{ .seq = 1, .witnessed_at = 42, .kind = kind_create, .collection = "c", .did = "d", .rkey = "r", .rev = "", .payload = "" }, });
const NoopHandler = struct { pub fn onEvent(_: *@This(), _: Event) void {} }; var handler = NoopHandler{}; var result: Result = .{ .planned_through_seq = 0, .sealed_tip_seq = 0 }; const opts = Options{ .host = "https://example.test" };
const with_trailing = try mem.concat(arena, u8, &.{ raw, "x" }); try testing.expectError(error.MalformedBlock, deliverBlock(arena, with_trailing, opts, &handler, &result)); try testing.expectError(error.MalformedBlock, deliverBlock(arena, raw[0 .. raw.len - 1], opts, &handler, &result));}
test "deliverBlock accepts empty compacted block" { const NoopHandler = struct { pub fn onEvent(_: *@This(), _: Event) void {} }; var handler = NoopHandler{}; var result: Result = .{ .planned_through_seq = 0, .sealed_tip_seq = 0 }; try deliverBlock(testing.allocator, &.{ 0, 0, 0, 0 }, .{ .host = "https://example.test" }, &handler, &result); try testing.expectEqual(@as(u64, 0), result.events_delivered);}
test "deliverFrame decompresses a plain zstd frame" { var arena_state = std.heap.ArenaAllocator.init(testing.allocator); defer arena_state.deinit(); const arena = arena_state.allocator();
// empty block (event_count 0) wrapped in a hand-built zstd frame: // magic, FHD (single-segment, 1-byte FCS), content size 4, // block header (last, raw, size 4), payload const frame = [_]u8{ 0x28, 0xb5, 0x2f, 0xfd, 0x20, 0x04, 0x21, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00 };
const NoopHandler = struct { pub fn onEvent(_: *@This(), _: Event) void {} }; var handler = NoopHandler{}; var result: Result = .{ .planned_through_seq = 0, .sealed_tip_seq = 0 }; try deliverFrame(arena, &frame, .{ .host = "https://example.test" }, .{ .segment_index = 7, .block_index = 3 }, &handler, &result); try testing.expectEqual(@as(u64, 0), result.events_delivered); try testing.expectEqual(@as(u64, 1), result.blocks_decoded); try testing.expectEqual(@as(u32, 7), result.position.?.segment_index); try testing.expectEqual(@as(u32, 3), result.position.?.block_index);}
test "cborToJson converts bytes and cids to lex-JSON conventions" { var arena_state = std.heap.ArenaAllocator.init(testing.allocator); defer arena_state.deinit(); const arena = arena_state.allocator();
const bytes_json = try livedecode.cborToJson(arena, .{ .bytes = "hi" }); try testing.expectEqualStrings("aGk", bytes_json.object.get("$bytes").?.string);
const cid_raw = [_]u8{ 0x01, 0x71, 0x12, 0x04, 0xde, 0xad, 0xbe, 0xef }; const cid_json = try livedecode.cborToJson(arena, .{ .cid = .{ .raw = &cid_raw } }); const link = cid_json.object.get("$link").?.string; try testing.expect(link.len > 1 and link[0] == 'b');
const nested = try livedecode.cborToJson(arena, .{ .array = &.{ .{ .unsigned = 7 }, .null, .{ .boolean = true } } }); try testing.expectEqual(@as(i64, 7), nested.array.items[0].integer); try testing.expectEqual(json.Value.null, nested.array.items[1]); try testing.expectEqual(true, nested.array.items[2].bool);}
test "plan request carries kinds only when set" { var arena_state = std.heap.ArenaAllocator.init(testing.allocator); defer arena_state.deinit(); const arena = arena_state.allocator(); const with_kinds = try planRequestBody(arena, .{ .host = "h", .kinds = &.{ .commit, .sync } }); try testing.expect(std.mem.indexOf(u8, with_kinds, "\"kinds\":[\"commit\",\"sync\"]") != null); const without = try planRequestBody(arena, .{ .host = "h" }); try testing.expect(std.mem.indexOf(u8, without, "kinds") == null);}
test "plan request carries seq bounds only when set" { var arena_state = std.heap.ArenaAllocator.init(std.testing.allocator); defer arena_state.deinit(); const arena = arena_state.allocator();
const unbounded = try planRequestBody(arena, .{ .host = "", .collections = &.{"a.b.c"} }); try std.testing.expect(mem.indexOf(u8, unbounded, "afterSeq") == null); try std.testing.expect(mem.indexOf(u8, unbounded, "beforeSeq") == null);
const bounded = try planRequestBody(arena, .{ .host = "", .collections = &.{"a.b.c"}, .after_seq = 100, .before_seq = 200, }); try std.testing.expect(mem.indexOf(u8, bounded, "\"afterSeq\":100") != null); try std.testing.expect(mem.indexOf(u8, bounded, "\"beforeSeq\":200") != null);}
test "anchorForFloor: covered floor, unsealed-tail floor, empty archive" { const spans = [_]SegmentSpan{ .{ .min_seq = 1, .max_seq = 100, .min_witnessed_us = 1000, .max_witnessed_us = 2000 }, .{ .min_seq = 101, .max_seq = 200, .min_witnessed_us = 1900, .max_witnessed_us = 3000 }, };
// floor inside the sealed range: sweep from one below the earliest // intersecting segment (the time floor trims the over-fetch) const covered = anchorForFloor(&spans, 2500); try std.testing.expect(covered.covered); try std.testing.expectEqual(@as(?u64, 100), covered.after_seq);
// floor past every sealed segment: the window lives in the unsealed // tail — anchor at the sealed tip and let the live cutover's cold // replay cover the rest const tail = anchorForFloor(&spans, 9000); try std.testing.expect(!tail.covered); try std.testing.expectEqual(@as(?u64, 200), tail.after_seq);
// no sealed archive at all: nothing to anchor; live from the tip const empty = anchorForFloor(&.{}, 2500); try std.testing.expect(!empty.covered); try std.testing.expectEqual(@as(?u64, null), empty.after_seq);}
test "seqBoundsForWindow admits intersecting segments, reports uncovered windows" { // three sealed segments; witnessed spans overlap at the edges the way // real segments do (per-connection skew) const spans = [_]SegmentSpan{ .{ .min_seq = 1, .max_seq = 100, .min_witnessed_us = 1000, .max_witnessed_us = 2000 }, .{ .min_seq = 101, .max_seq = 200, .min_witnessed_us = 1900, .max_witnessed_us = 3000 }, .{ .min_seq = 201, .max_seq = 300, .min_witnessed_us = 2900, .max_witnessed_us = 4000 }, };
// a window inside the middle segment admits it plus the overlapping edge const mid = seqBoundsForWindow(&spans, 2500, 2600); try std.testing.expect(mid.covered); try std.testing.expectEqual(@as(?u64, 100), mid.after_seq); try std.testing.expectEqual(@as(?u64, 200), mid.before_seq);
// an open end means "through the sealed tip": no upper bound const open = seqBoundsForWindow(&spans, 3500, null); try std.testing.expect(open.covered); try std.testing.expectEqual(@as(?u64, 200), open.after_seq); try std.testing.expectEqual(@as(?u64, null), open.before_seq);
// a window before all witnessed spans (bootstrap-era session) is honest const before_archive = seqBoundsForWindow(&spans, 10, 20); try std.testing.expect(!before_archive.covered);
// straddling windows admit everything they touch const straddle = seqBoundsForWindow(&spans, 1500, 3500); try std.testing.expect(straddle.covered); try std.testing.expectEqual(@as(?u64, 0), straddle.after_seq); try std.testing.expectEqual(@as(?u64, 300), straddle.before_seq);}
test "parsePlan handles blocks and whole-segment modes" { var arena_state = std.heap.ArenaAllocator.init(testing.allocator); defer arena_state.deinit(); const arena = arena_state.allocator();
const plan = try parsePlan(arena, \\{"plannedThroughSeq": 12345, "sealedTipSeq": 12000, "segments": [ \\ {"name": "seg_0000000001", "index": 0, "checksum": "abc", "minSeq": 1, "maxSeq": 100, \\ "mode": "blocks", "blocks": [{"first": 0, "last": 2}, {"first": 5, "last": 5}]}, \\ {"name": "seg_0000000002", "index": 1, "checksum": "def", "minSeq": 101, "maxSeq": 200, \\ "mode": "segment"} \\]} );
try testing.expectEqual(@as(u64, 12345), plan.planned_through_seq); try testing.expectEqual(@as(u64, 12000), plan.sealed_tip_seq); try testing.expectEqual(@as(usize, 2), plan.segments.len); try testing.expectEqualStrings("seg_0000000001", plan.segments[0].name); try testing.expectEqual(@as(u32, 0), plan.segments[0].index); try testing.expectEqual(@as(u64, 0xabc), plan.segments[0].checksum.?); try testing.expectEqual(@as(u32, 1), plan.segments[1].index); const ranges = plan.segments[0].blocks.?; try testing.expectEqual(@as(u32, 0), ranges[0].first); try testing.expectEqual(@as(u32, 2), ranges[0].last); try testing.expectEqual(@as(u32, 5), ranges[1].first); try testing.expect(plan.segments[1].blocks == null);}
test "deliverSegment walks header and block index" { var arena_state = std.heap.ArenaAllocator.init(testing.allocator); defer arena_state.deinit(); const arena = arena_state.allocator();
// one empty block, hand-built zstd frame as in the frame test const frame = [_]u8{ 0x28, 0xb5, 0x2f, 0xfd, 0x20, 0x04, 0x21, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00 };
var out: std.Io.Writer.Allocating = .init(arena); const w = &out.writer; try w.writeAll("jss0"); try w.splatByteAll(0, 10); // checksum u64 + version u16 try w.writeInt(u32, 1, .little); // block_count @14 try w.splatByteAll(0, 72); // through offset 90 const frame_offset: u64 = header_size; const block_index_offset: u64 = header_size + 8 + frame.len; try w.writeInt(u64, block_index_offset, .little); // block_index_offset @90 try w.splatByteAll(0, 158); // reserved → header complete @256 try w.writeInt(u64, frame.len, .little); // block length prefix try w.writeAll(&frame); // block index entry try w.writeInt(u64, frame_offset, .little); try w.writeInt(u32, frame.len, .little); try w.writeInt(u32, 4, .little); // uncompressed_size try w.writeInt(u32, 0, .little); // event_count try w.splatByteAll(0, 32); // seq/witnessed bounds
const NoopHandler = struct { pub fn onEvent(_: *@This(), _: Event) void {} }; var handler = NoopHandler{}; var result: Result = .{ .planned_through_seq = 0, .sealed_tip_seq = 0 }; const seg = PlannedSegment{ .name = "seg_test", .index = 0, .checksum = null, .blocks = null }; try deliverSegment(arena, out.written(), .{ .host = "https://example.test" }, seg, &handler, &result); try testing.expectEqual(@as(u64, 0), result.events_delivered); try testing.expectEqual(@as(u64, 1), result.blocks_decoded);}
test "skippedByResume skips through the resume position, nothing after" { const sa = Position{ .segment_index = 5, .block_index = 10 }; try testing.expect(skippedByResume(sa, 4, 999)); try testing.expect(skippedByResume(sa, 5, 9)); try testing.expect(skippedByResume(sa, 5, 10)); try testing.expect(!skippedByResume(sa, 5, 11)); try testing.expect(!skippedByResume(sa, 6, 0)); try testing.expect(!skippedByResume(null, 0, 0));}
test "verifySegment checks the jss self-checksum and the plan's" { var arena_state = std.heap.ArenaAllocator.init(testing.allocator); defer arena_state.deinit(); const arena = arena_state.allocator();
// minimal sealed segment: 256B header + 4-byte footer var out: std.Io.Writer.Allocating = .init(arena); const w = &out.writer; try w.writeAll("jss0"); try w.splatByteAll(0, 54); // checksum + version..offset 58 try w.writeInt(u64, header_size, .little); // footer_offset @58 try w.splatByteAll(0, 190); // rest of header try w.writeAll("foot"); const body = out.written();
var hasher = std.hash.XxHash3.init(0); hasher.update(body[12..header_size]); hasher.update(body[header_size..]); const checksum = hasher.final(); mem.writeInt(u64, body[4..12], checksum, .little);
try verifySegment(body, checksum); try verifySegment(body, null); try testing.expectError(error.ChecksumMismatch, verifySegment(body, checksum + 1));
body[body.len - 1] ^= 1; // corrupt the footer try testing.expectError(error.ChecksumMismatch, verifySegment(body, checksum)); body[body.len - 1] ^= 1;
mem.writeInt(u64, body[4..12], 0, .little); // unsealed try testing.expectError(error.SegmentUnsealed, verifySegment(body, null));}
test "deliverSegment rejects bad magic" { const NoopHandler = struct { pub fn onEvent(_: *@This(), _: Event) void {} }; var handler = NoopHandler{}; var result: Result = .{ .planned_through_seq = 0, .sealed_tip_seq = 0 }; const bogus: [header_size]u8 = @splat(0); try testing.expectError(error.MalformedSegment, deliverSegment( testing.allocator, &bogus, .{ .host = "https://example.test" }, .{ .name = "seg_test", .index = 0, .checksum = null, .blocks = null }, &handler, &result, ));}
test "deliverBlock enforces the exact (after_seq, before_seq] window per row" { // upstream matcher parity (filter.go): the plan prunes whole blocks, // but the exact seq window is the client's to apply on each row — // found live by the cutover smoke, which received ~2 whole blocks of // rows below the requested resume point var arena_state = std.heap.ArenaAllocator.init(testing.allocator); defer arena_state.deinit(); const arena = arena_state.allocator();
const raw = try buildTestBlock(arena, &.{ .{ .seq = 10, .witnessed_at = 1000, .kind = kind_create, .collection = "c", .did = "did:plc:a", .rkey = "r1", .rev = "v1", .payload = &test_record_cbor }, .{ .seq = 11, .witnessed_at = 1001, .kind = kind_create, .collection = "c", .did = "did:plc:a", .rkey = "r2", .rev = "v2", .payload = &test_record_cbor }, .{ .seq = 12, .witnessed_at = 1002, .kind = kind_create, .collection = "c", .did = "did:plc:a", .rkey = "r3", .rev = "v3", .payload = &test_record_cbor }, .{ .seq = 13, .witnessed_at = 1003, .kind = kind_create, .collection = "c", .did = "did:plc:a", .rkey = "r4", .rev = "v4", .payload = &test_record_cbor }, });
const RowCollector = struct { seqs: std.array_list.Managed(u64), pub fn onRow(self: *@This(), event: livedecode.Event) bool { self.seqs.append(event.seq) catch unreachable; return true; } }; var handler = RowCollector{ .seqs = .init(arena) }; var result: Result = .{ .planned_through_seq = 0, .sealed_tip_seq = 0 }; try deliverBlock(arena, raw, .{ .host = "https://example.test", .after_seq = 10, .before_seq = 12, }, &handler, &result); try testing.expectEqualSlices(u64, &.{ 11, 12 }, handler.seqs.items);}
// Upstream filter_test.go: TestMatcherCollectionExactAndWildcard,// TestMatcherWildcardBoundary, and marker/DID filter contracts.test "conformance: archive wildcards preserve namespace boundaries and markers" { var arena_state = std.heap.ArenaAllocator.init(testing.allocator); defer arena_state.deinit(); const arena = arena_state.allocator(); const raw = try buildTestBlock(arena, &.{ .{ .seq = 1, .witnessed_at = 1, .kind = kind_create, .collection = "app.bsky.feed.post", .did = "did:plc:a", .rkey = "r", .rev = "r", .payload = &test_record_cbor }, .{ .seq = 2, .witnessed_at = 2, .kind = kind_create, .collection = "app.bsky.graph.follow", .did = "did:plc:a", .rkey = "r", .rev = "r", .payload = &test_record_cbor }, .{ .seq = 3, .witnessed_at = 3, .kind = kind_create, .collection = "app.bsky.graphient.thing", .did = "did:plc:a", .rkey = "r", .rev = "r", .payload = &test_record_cbor }, .{ .seq = 4, .witnessed_at = 4, .kind = kind_account, .collection = "$account", .did = "did:plc:a", .rkey = "", .rev = "", .payload = "" }, .{ .seq = 5, .witnessed_at = 5, .kind = kind_identity, .collection = "$identity", .did = "did:plc:other", .rkey = "", .rev = "", .payload = "" }, }); const Handler = struct { seqs: std.ArrayList(u64) = .empty, allocator: Allocator, pub fn onRow(self: *@This(), event: livedecode.Event) bool { self.seqs.append(self.allocator, event.seq) catch @panic("out of memory"); return true; } }; var handler = Handler{ .allocator = arena }; var result: Result = .{ .planned_through_seq = 0, .sealed_tip_seq = 0 }; try deliverBlock(arena, raw, .{ .host = "", .collections = &.{ "app.bsky.feed.post", "app.bsky.graph.*" }, .dids = &.{"did:plc:a"}, }, &handler, &result); try testing.expectEqualSlices(u64, &.{ 1, 2, 4 }, handler.seqs.items);}
// Upstream WithRawRecords/decodeCommitInto: skip map materialization,// preserve raw bytes for creates, and provide no record for deletes.test "conformance: raw archive records skip decoding and preserve bytes" { var arena_state = std.heap.ArenaAllocator.init(testing.allocator); defer arena_state.deinit(); const arena = arena_state.allocator(); const invalid_cbor = "\xff"; const raw = try buildTestBlock(arena, &.{ .{ .seq = 1, .witnessed_at = 1, .kind = kind_create, .collection = "app.bsky.feed.post", .did = "did:plc:a", .rkey = "r", .rev = "r", .payload = invalid_cbor }, .{ .seq = 2, .witnessed_at = 2, .kind = kind_delete, .collection = "app.bsky.feed.post", .did = "did:plc:a", .rkey = "r", .rev = "r", .payload = "" }, }); const Handler = struct { count: usize = 0, valid: bool = true, pub fn onRow(self: *@This(), event: livedecode.Event) bool { const commit = event.payload.commit; self.valid = self.valid and commit.record == null and mem.eql(u8, if (commit.operation == .delete) "" else invalid_cbor, commit.record_cbor); self.count += 1; return true; } }; var handler = Handler{}; var result: Result = .{ .planned_through_seq = 0, .sealed_tip_seq = 0 }; try deliverBlock(arena, raw, .{ .host = "", .decode_records = false }, &handler, &result); try testing.expectEqual(@as(usize, 2), handler.count); try testing.expect(handler.valid); try testing.expectError(error.MalformedRecord, deliverBlock(arena, raw, .{ .host = "" }, &handler, &result));}
// Upstream TestDownloadTransformKeepsPayloadWithMalformedRecordError:// valid rows from a mixed block survive and the decode error is reported.test "archive malformed records preserve valid rows before reporting the error" { var arena_state = std.heap.ArenaAllocator.init(testing.allocator); defer arena_state.deinit(); const arena = arena_state.allocator(); const raw = try buildTestBlock(arena, &.{ .{ .seq = 1, .witnessed_at = 1, .kind = kind_create, .collection = "app.bsky.feed.post", .did = "did:plc:a", .rkey = "r", .rev = "r", .payload = &test_record_cbor }, .{ .seq = 2, .witnessed_at = 2, .kind = kind_create, .collection = "app.bsky.feed.post", .did = "did:plc:a", .rkey = "r", .rev = "r", .payload = "\xff" }, .{ .seq = 3, .witnessed_at = 3, .kind = kind_create, .collection = "app.bsky.feed.post", .did = "did:plc:a", .rkey = "r", .rev = "r", .payload = &test_record_cbor }, }); const Handler = struct { items: [4]u64 = undefined, len: usize = 0, err: ?anyerror = null, pub fn onRow(self: *@This(), event: livedecode.Event) bool { self.items[self.len] = event.seq; self.len += 1; return true; } pub fn onError(self: *@This(), err: anyerror) bool { self.err = err; self.items[self.len] = 0; self.len += 1; return true; } }; var handler = Handler{}; var result: Result = .{ .planned_through_seq = 0, .sealed_tip_seq = 0 }; try deliverBlock(arena, raw, .{ .host = "" }, &handler, &result); try testing.expectEqualSlices(u64, &.{ 1, 3, 0 }, handler.items[0..handler.len]); try testing.expectEqual(@as(?anyerror, error.MalformedRecord), handler.err);}
test "archive recovery never swallows cancellation allocation failure or absent callbacks" { const Handler = struct { errors: usize = 0, pub fn onError(self: *@This(), _: anyerror) bool { self.errors += 1; return true; } }; var handler = Handler{}; try testing.expectError(error.Canceled, reportArchiveError(&handler, error.Canceled)); try testing.expectError(error.OutOfMemory, reportArchiveError(&handler, error.OutOfMemory)); try testing.expectEqual(@as(usize, 0), handler.errors); var no_callback: struct {} = .{}; try testing.expectError(error.FetchFailed, reportArchiveError(&no_callback, error.FetchFailed)); const NotifyOnly = struct { errors: usize = 0, pub fn onError(self: *@This(), _: anyerror) void { self.errors += 1; } }; var notify_only = NotifyOnly{}; try testing.expectError(error.FetchFailed, reportArchiveError(¬ify_only, error.FetchFailed)); try testing.expectEqual(@as(usize, 1), notify_only.errors);}
test "download retries use upstream's three attempt default" { try std.testing.expectEqual(@as(u32, 3), fetch_attempts);}
test "retry timing matches upstream segment and XRPC policies" { const ms = std.time.ns_per_ms; const sec = std.time.ns_per_s; try std.testing.expectEqual(@as(u64, 500 * ms), retryDelay(.segment, 1, 0, 0)); try std.testing.expectEqual(@as(u64, sec), retryDelay(.segment, 2, 0, 0)); try std.testing.expectEqual(@as(u64, 30 * sec), retryDelay(.segment, 100, 0, 0)); try std.testing.expectEqual(@as(u64, 5 * sec), retryDelay(.segment, 1, 5 * sec, 0)); try std.testing.expectEqual(@as(u64, sec), retryDelay(.segment, 2, 100 * ms, 0)); try std.testing.expectEqual(@as(u64, 500 * ms), retryDelay(.xrpc, 1, 0, 0)); try std.testing.expect(retryDelay(.xrpc, 1, 0, 0.999999) < 600 * ms); try std.testing.expectEqual(@as(u64, 550 * ms), retryDelay(.xrpc, 1, 0, 0.5)); try std.testing.expectEqual(@as(u64, 100 * ms), retryDelay(.xrpc, 2, 100 * ms, 0)); try std.testing.expectEqual(@as(u64, 30 * sec), retryDelay(.xrpc, 1, 60 * sec, 0));}
test "retry hints prefer absolute reset and bound missing expired and large values" { const sec = std.time.ns_per_s; try std.testing.expectEqual(@as(u64, 0), retryHint(.{}, 100 * sec)); try std.testing.expectEqual(@as(u64, 250 * std.time.ns_per_ms), retryHint(.{ .reset = 101 }, 100 * sec + 750 * std.time.ns_per_ms)); try std.testing.expectEqual(@as(u64, 5 * sec), retryHint(.{ .retry_after = 5 }, 100 * sec)); try std.testing.expectEqual(@as(u64, 3 * sec), retryHint(.{ .reset = 103, .retry_after = 9 }, 100 * sec)); try std.testing.expectEqual(@as(u64, 0), retryHint(.{ .reset = 99, .retry_after = 9 }, 100 * sec)); try std.testing.expectEqual(@as(u64, 2 * sec), retryHint(.{ .retry_after_at = 102 }, 100 * sec)); try std.testing.expectEqual(@as(u64, 0), retryHint(.{ .retry_after_at = 90 }, 100 * sec)); try std.testing.expectEqual(@as(u64, 30 * sec), retryHint(.{ .retry_after = std.math.maxInt(u64) }, 100 * sec));}
test "negative reset takes precedence over retry after" { try testing.expectEqual(@as(u64, 0), retryHint(.{ .reset = -1, .retry_after = 9 }, 100 * std.time.ns_per_s));}