Something went wrong. Try again.
jetstream v2 in zig stream.waow.tech
Something went wrong. Try again.
1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586878889909192939495969798991001011021031041051061071081091101111121131141151161171181191201211221231241251261271281291301311321331341351361371381391401411421431441451461471481491501511521531541551561571581591601611621631641651661671681691701711721731741751761771781791801811821831841851861871881891901911921931941951961971981992002012022032042052062072082092102112122132142152162172182192202212222232242252262272282292302312322332342352362372382392402412422432442452462472482492502512522532542552562572582592602612622632642652662672682692702712722732742752762772782792802812822832842852862872882892902912922932942952962972982993003013023033043053063073083093103113123133143153163173183193203213223233243253263273283293303313323333343353363373383393403413423433443453463473483493503513523533543553563573583593603613623633643653663673683693703713723733743753763773783793803813823833843853863873883893903913923933943953963973983994004014024034044054064074084094104114124134144154164174184194204214224234244254264274284294304314324334344354364374384394404414424434444454464474484494504514524534544554564574584594604614624634644654664674684694704714724734744754764774784794804814824834844854864874884894904914924934944954964974984995005015025035045055065075085095105115125135145155165175185195205215225235245255265275285295305315325335345355365375385395405415425435445455465475485495505515525535545555565575585595605615625635645655665675685695705715725735745755765775785795805815825835845855865875885895905915925935945955965975985996006016026036046056066076086096106116126136146156166176186196206216226236246256266276286296306316326336346356366376386396406416426436446456466476486496506516526536546556566576586596606616626636646656666676686696706716726736746756766776786796806816826836846856866876886896906916926936946956966976986997007017027037047057067077087097107117127137147157167177187197207217227237247257267277287297307317327337347357367377387397407417427437447457467477487497507517527537547557567577587597607617627637647657667677687697707717727737747757767777787797807817827837847857867877887897907917927937947957967977987998008018028038048058068078088098108118128138148158168178188198208218228238248258268278288298308318328338348358368378388398408418428438448458468478488498508518528538548558568578588598608618628638648658668678688698708718728738748758768778788798808818828838848858868878888898908918928938948958968978988999009019029039049059069079089099109119129139149159169179189199209219229239249259269279289299309319329339349359369379389399409419429439449459469479489499509519529539549559569579589599609619629639649659669679689699709719729739749759769779789799809819829839849859869879889899909919929939949959969979989991000100110021003100410051006100710081009101010111012101310141015101610171018101910201021102210231024102510261027102810291030103110321033103410351036103710381039104010411042104310441045104610471048104910501051105210531054105510561057105810591060106110621063106410651066106710681069107010711072107310741075107610771078107910801081108210831084108510861087108810891090109110921093109410951096109710981099110011011102110311041105110611071108110911101111111211131114111511161117111811191120112111221123112411251126112711281129113011311132113311341135113611371138113911401141114211431144114511461147114811491150115111521153115411551156115711581159116011611162116311641165116611671168116911701171117211731174117511761177117811791180118111821183118411851186118711881189119011911192119311941195119611971198119912001201120212031204120512061207120812091210121112121213121412151216121712181219122012211222122312241225122612271228122912301231123212331234123512361237123812391240124112421243124412451246124712481249125012511252125312541255125612571258125912601261126212631264126512661267126812691270127112721273127412751276127712781279128012811282128312841285128612871288128912901291129212931294129512961297129812991300130113021303130413051306130713081309131013111312131313141315131613171318131913201321132213231324132513261327132813291330133113321333133413351336133713381339134013411342134313441345134613471348134913501351135213531354135513561357135813591360136113621363136413651366136713681369137013711372137313741375137613771378137913801381138213831384138513861387138813891390139113921393139413951396139713981399140014011402140314041405140614071408140914101411141214131414141514161417141814191420142114221423142414251426142714281429143014311432143314341435143614371438143914401441144214431444144514461447144814491450145114521453145414551456145714581459146014611462146314641465146614671468146914701471147214731474147514761477147814791480148114821483148414851486148714881489149014911492149314941495149614971498149915001501150215031504150515061507150815091510151115121513151415151516151715181519152015211522152315241525152615271528152915301531153215331534153515361537153815391540154115421543154415451546154715481549155015511552155315541555155615571558155915601561156215631564156515661567156815691570157115721573157415751576157715781579158015811582158315841585158615871588158915901591159215931594159515961597159815991600160116021603160416051606160716081609161016111612161316141615161616171618161916201621162216231624162516261627162816291630163116321633163416351636163716381639164016411642164316441645164616471648164916501651165216531654165516561657165816591660166116621663166416651666166716681669167016711672167316741675167616771678167916801681168216831684168516861687168816891690169116921693169416951696169716981699170017011702170317041705170617071708170917101711171217131714171517161717171817191720172117221723172417251726172717281729173017311732173317341735173617371738173917401741//! cold reader — replay durable disk history onto the subscriber wire//!//! serves /subscribe cursors that fall below the hot tail's floor: walks//! sealed segment files in index order, skips whole segments/blocks via//! their witnessed_at envelopes, re-encodes matching rows as v1 JSON//! (commit payloads are the stored record DAG-CBOR; the record CID is//! recomputed from those bytes, which is how upstream serves it too), and//! includes the fsynced active range selected by sequence. Each server pass//! reselects hot or cold from the advanced cursor. Overlap is at-least-once.//!//! A replay snapshot atomically pairs resident sealed metadata with the active//! segment index and fsynced byte boundary. This closes the rotation seam//! without holding the archive writer lock during decode or socket writes.
const std = @import("std");const zat = @import("zat");const filter_mod = @import("../serve/filter.zig");const segment = @import("segment.zig");const archive_mod = @import("archive.zig");const rewrite_mod = @import("segment_rewrite.zig");const cold_cache = @import("cold_cache.zig");const wire = @import("../serve/wire.zig");const tail = @import("../serve/tail.zig");const zstd = @import("zstd.zig");
const Io = std.Io;const Allocator = std.mem.Allocator;const log = std.log.scoped(.stream);
pub const ReplayResult = enum { sent, filtered, oversize, skipped_sync, skipped_resync, encode_error };pub const WriteResult = enum { sent, oversize, encode_error };pub const default_block_cache_bytes = cold_cache.default_max_bytes;
pub const CachedFrame = struct { /// Logical JSON body used for maxMessageSizeBytes and encoded-byte metrics. body: []const u8, /// Wire payload; either body or a memoized dictionary zstd frame. payload: []const u8, compressed: bool,};
pub const BatchResult = struct { frames: u64, scanned: usize, next_seq: u64, exhausted: bool,};
pub const ColdReader = struct { cache: cold_cache.Cache,
pub fn init(allocator: Allocator, io: Io, max_bytes: usize) ColdReader { return .{ .cache = .init(allocator, io, max_bytes) }; }
pub fn deinit(self: *ColdReader) void { self.cache.deinit(); }
pub fn invalidateSegment(self: *ColdReader, segment_index: u64) !void { try self.cache.invalidateSegment(segment_index); }
pub fn invalidateOpaque(ctx: ?*anyopaque, segment_index: u64) anyerror!void { const self: *ColdReader = @ptrCast(@alignCast(ctx.?)); try self.invalidateSegment(segment_index); }
/// Cached sealed-block replay used by production subscribers. Decoded /// blocks plus v1/v2 JSON and dictionary frames are shared across every /// concurrent reader while the cache entry remains resident. pub fn readSeqBatch( self: *ColdReader, archive: *archive_mod.Archive, filter: anytype, comptime wants: fn (@TypeOf(filter), wire.Kind, []const u8, []const u8, tail.SkipV1) bool, sink: anytype, comptime write: fn (@TypeOf(sink), CachedFrame) anyerror!WriteResult, comptime observe: fn (@TypeOf(sink), u64, ReplayResult) anyerror!void, v2: bool, compressed: bool, from_seq: u64, stop_seq: u64, max: usize, ) !BatchResult { const allocator = self.cache.allocator; const io = self.cache.io; var snapshot = try archive.coldSnapshot(allocator, from_seq); defer snapshot.deinit();
var frames: u64 = 0; var scanned: usize = 0; var next_seq = from_seq; const limit = @max(max, 1); for (snapshot.summaries) |summary| { if (summary.header.max_seq < from_seq) continue; var blocks = try archive.manifest.blockIndex(summary.idx); defer blocks.deinit(); var name_buffer: [64]u8 = undefined; const name = archive_mod.formatSegmentName(&name_buffer, summary.idx); var file = archive.dir.openFile(io, name, .{}) catch |err| { // A manifest-listed segment we cannot open is a hole, not an // empty range. Skipping it would advance the cursor past // durable events that were never delivered. log.warn("cold: open sealed segment {s}: {s}", .{ name, @errorName(err) }); return err; }; defer file.close(io); const fresh_header = try segment.readHeaderFile(io, &file); if (@as(usize, fresh_header.block_count) != blocks.entries.len) return error.BlockTopologyChanged;
for (blocks.entries, 0..) |info, block_index| { if (info.max_seq < from_seq) continue; const fresh_entry = try segment.readBlockIndexFile(io, &file, fresh_header, block_index); const Decode = struct { io: Io, file: *Io.File, entry: segment.BlockIndexEntry,
fn run(ctx: @This(), alloc: Allocator) anyerror!segment.Block { return segment.readBlockFile(alloc, ctx.io, ctx.file, ctx.entry); } }; var handle = try self.cache.acquire( self.cache.key(summary.idx, fresh_header.checksum, block_index), Decode{ .io = io, .file = &file, .entry = fresh_entry }, Decode.run, ); defer handle.deinit(); for (handle.events(), 0..) |ev, event_index| { if (ev.seq < from_seq) continue; if (stop_seq != 0 and ev.seq >= stop_seq) return .{ .frames = frames, .scanned = scanned, .next_seq = @max(next_seq, stop_seq), .exhausted = true, }; next_seq = ev.seq +| 1; scanned += 1; const kind: wire.Kind = switch (ev.kind) { .create, .update, .delete, .create_resync => .commit, .identity => .identity, .account => .account, .sync => .sync, }; if (!v2) switch (ev.kind) { .sync => { try observe(sink, ev.seq, .skipped_sync); if (scanned >= limit) return .{ .frames = frames, .scanned = scanned, .next_seq = next_seq, .exhausted = false }; continue; }, .create_resync => { try observe(sink, ev.seq, .skipped_resync); if (scanned >= limit) return .{ .frames = frames, .scanned = scanned, .next_seq = next_seq, .exhausted = false }; continue; }, else => {}, }; if (!wants(filter, kind, ev.did, ev.collection, .none)) { try observe(sink, ev.seq, .filtered); if (scanned >= limit) return .{ .frames = frames, .scanned = scanned, .next_seq = next_seq, .exhausted = false }; continue; } const cached_frame = self.frame(&handle, event_index, v2, compressed) catch { try observe(sink, ev.seq, .encode_error); if (scanned >= limit) return .{ .frames = frames, .scanned = scanned, .next_seq = next_seq, .exhausted = false }; continue; } orelse { try observe(sink, ev.seq, .encode_error); if (scanned >= limit) return .{ .frames = frames, .scanned = scanned, .next_seq = next_seq, .exhausted = false }; continue; }; switch (try write(sink, cached_frame)) { .sent => { try observe(sink, ev.seq, .sent); frames += 1; }, .oversize => try observe(sink, ev.seq, .oversize), .encode_error => try observe(sink, ev.seq, .encode_error), } if (scanned >= limit) return .{ .frames = frames, .scanned = scanned, .next_seq = next_seq, .exhausted = false, }; } } }
// The hot floor can sit above rows in the current active segment. // Only its fsynced prefix appears in the snapshot; pending rows stay // exclusively in the durability-gated hot tail. var active_name_buf: [64]u8 = undefined; const active_name = archive_mod.formatSegmentName(&active_name_buf, snapshot.active_index); var active_file = try archive.dir.openFile(io, active_name, .{}); defer active_file.close(io); var active_offset = snapshot.active_start; while (active_offset < snapshot.active_length) { var block_arena = std.heap.ArenaAllocator.init(allocator); defer block_arena.deinit(); if (active_offset + 8 > snapshot.active_length) return error.TruncatedActiveBlock; var frame_len_bytes: [8]u8 = undefined; _ = try active_file.readPositionalAll(io, &frame_len_bytes, active_offset); const frame_len = std.mem.readInt(u64, &frame_len_bytes, .little); if (frame_len > snapshot.active_length - active_offset - 8) return error.TruncatedActiveBlock; const compressed_frame = try block_arena.allocator().alloc(u8, @intCast(frame_len)); _ = try active_file.readPositionalAll(io, compressed_frame, active_offset + 8); active_offset += 8 + @as(usize, @intCast(frame_len)); var frame_reader: std.Io.Reader = .fixed(compressed_frame); var zstd_stream: std.compress.zstd.Decompress = .init(&frame_reader, &.{}, .{}); var decoded: std.Io.Writer.Allocating = .init(block_arena.allocator()); _ = zstd_stream.reader.streamRemaining(&decoded.writer) catch return error.DecompressFailed; const block_bytes = try decoded.toOwnedSlice(); const block = try segment.decodeBlock(block_arena.allocator(), block_bytes); for (block.events) |ev| { if (ev.seq < from_seq) continue; if (stop_seq != 0 and ev.seq >= stop_seq) return .{ .frames = frames, .scanned = scanned, .next_seq = @max(next_seq, stop_seq), .exhausted = true, }; next_seq = ev.seq +| 1; scanned += 1; const kind: wire.Kind = switch (ev.kind) { .create, .update, .delete, .create_resync => .commit, .identity => .identity, .account => .account, .sync => .sync, }; if (!v2) switch (ev.kind) { .sync => { try observe(sink, ev.seq, .skipped_sync); if (scanned >= limit) return .{ .frames = frames, .scanned = scanned, .next_seq = next_seq, .exhausted = false }; continue; }, .create_resync => { try observe(sink, ev.seq, .skipped_resync); if (scanned >= limit) return .{ .frames = frames, .scanned = scanned, .next_seq = next_seq, .exhausted = false }; continue; }, else => {}, }; if (!wants(filter, kind, ev.did, ev.collection, .none)) { try observe(sink, ev.seq, .filtered); if (scanned >= limit) return .{ .frames = frames, .scanned = scanned, .next_seq = next_seq, .exhausted = false }; continue; }
var frame_arena = std.heap.ArenaAllocator.init(allocator); defer frame_arena.deinit(); const frame_allocator = frame_arena.allocator(); const body = ((if (v2) try encodeEventV2(frame_allocator, ev) else try encodeEvent(frame_allocator, ev)) orelse { try observe(sink, ev.seq, .encode_error); if (scanned >= limit) return .{ .frames = frames, .scanned = scanned, .next_seq = next_seq, .exhausted = false }; continue; }); var active_frame: CachedFrame = .{ .body = body, .payload = body, .compressed = false }; if (compressed) { var compressor = if (v2) try zstd.DictCompressor.initLevel(&zstd.v2_dictionary, 1) else try zstd.DictCompressor.init(zstd.v1_dictionary); defer compressor.deinit(); active_frame.payload = compressor.compress(frame_allocator, body) catch { try observe(sink, ev.seq, .encode_error); if (scanned >= limit) return .{ .frames = frames, .scanned = scanned, .next_seq = next_seq, .exhausted = false }; continue; }; active_frame.compressed = true; } switch (try write(sink, active_frame)) { .sent => { try observe(sink, ev.seq, .sent); frames += 1; }, .oversize => try observe(sink, ev.seq, .oversize), .encode_error => try observe(sink, ev.seq, .encode_error), } if (scanned >= limit) return .{ .frames = frames, .scanned = scanned, .next_seq = next_seq, .exhausted = false }; } } if (next_seq < stop_seq) return error.ColdReplayGap; return .{ .frames = frames, .scanned = scanned, .next_seq = next_seq, .exhausted = true }; }
pub fn replaySeq( self: *ColdReader, archive: *archive_mod.Archive, filter: anytype, comptime wants: fn (@TypeOf(filter), wire.Kind, []const u8, []const u8, tail.SkipV1) bool, sink: anytype, comptime write: fn (@TypeOf(sink), CachedFrame) anyerror!WriteResult, comptime observe: fn (@TypeOf(sink), u64, ReplayResult) anyerror!void, v2: bool, compressed: bool, from_seq: u64, ) !u64 { const result = try self.readSeqBatch( archive, filter, wants, sink, write, observe, v2, compressed, from_seq, 0, std.math.maxInt(usize), ); return result.frames; }
fn frame(self: *ColdReader, handle: *cold_cache.Handle, event_index: usize, v2: bool, compressed: bool) !?CachedFrame { _ = self; const Encoder = struct { fn v1(_: void, allocator: Allocator, event: segment.Event) anyerror!?[]u8 { var arena = std.heap.ArenaAllocator.init(allocator); defer arena.deinit(); const body = (try encodeEvent(arena.allocator(), event)) orelse return null; return try allocator.dupe(u8, body); } fn encodeV2(_: void, allocator: Allocator, event: segment.Event) anyerror!?[]u8 { var arena = std.heap.ArenaAllocator.init(allocator); defer arena.deinit(); const body = (try encodeEventV2(arena.allocator(), event)) orelse return null; return try allocator.dupe(u8, body); } fn zstdV1(body: []const u8, allocator: Allocator, _: segment.Event) anyerror!?[]u8 { var compressor = try zstd.DictCompressor.init(zstd.v1_dictionary); defer compressor.deinit(); return try compressor.compress(allocator, body); } fn zstdV2(body: []const u8, allocator: Allocator, _: segment.Event) anyerror!?[]u8 { var compressor = try zstd.DictCompressor.initLevel(&zstd.v2_dictionary, 1); defer compressor.deinit(); return try compressor.compress(allocator, body); } }; const body = (if (v2) try handle.memoize(event_index, .v2, {}, Encoder.encodeV2) else try handle.memoize(event_index, .v1, {}, Encoder.v1)) orelse return null; if (!compressed) return .{ .body = body, .payload = body, .compressed = false }; const payload = (if (v2) try handle.memoize(event_index, .zstd_v2, body, Encoder.zstdV2) else try handle.memoize(event_index, .zstd_v1, body, Encoder.zstdV1)) orelse return error.CompressFailed; return .{ .body = body, .payload = payload, .compressed = true }; }};
pub const ResolvedCursor = struct { seq: u64, clamped: bool = false,};
/// Translate a witnessed-at timestamp to the first archived sequence at or/// after it. Segment bounds and block indexes are resident; only the selected/// compressed block is read from disk.pub fn resolveTimeToSeq( allocator: Allocator, io: Io, archive: *archive_mod.Archive, time_us: i64,) !ResolvedCursor { const metas = try archive.manifest.all(allocator); defer allocator.free(metas);
if (metas.len == 0) return .{ .seq = 1, .clamped = true }; const first = metas[0]; if (time_us <= first.header.min_witnessed_at) return .{ .seq = @max(first.header.min_seq, 1), .clamped = true };
var candidate: ?@TypeOf(first) = null; for (metas) |meta| { if (meta.header.max_witnessed_at >= time_us) { candidate = meta; break; } } if (candidate == null) { // The time falls in the not-yet-sealed window. Upstream anchors at // sealed_max+1 and its replay walks the active segment efficiently; // our cold walk re-reads the active region on every pass, so a // subscriber anchored up to a whole open segment back lands on a // treadmill that cannot outrun live ingest (2026-08-08 boot-window // consumers). Resolve INTO the active region's fsynced blocks: one // sequential walk, at connect time only. return resolveTimeInActive(allocator, io, archive, time_us, metas[metas.len - 1].header.max_seq +| 1); }
const meta = candidate.?; var name_buffer: [64]u8 = undefined; const name = archive_mod.formatSegmentName(&name_buffer, meta.idx); var file = try archive.dir.openFile(io, name, .{}); defer file.close(io); var blocks = try archive.manifest.blockIndex(meta.idx); defer blocks.deinit(); const fresh_header = try segment.readHeaderFile(io, &file); if (@as(usize, fresh_header.block_count) != blocks.entries.len) return error.BlockTopologyChanged; var lo: u32 = 0; var hi: u32 = @intCast(blocks.entries.len); while (lo < hi) { const mid = lo + (hi - lo) / 2; const entry = blocks.entries[@intCast(mid)]; if (entry.max_witnessed_at >= time_us) hi = mid else lo = mid + 1; } if (lo == blocks.entries.len) return .{ .seq = @max(meta.header.min_seq, 1) }; const entry = blocks.entries[@intCast(lo)]; const fresh_entry = try segment.readBlockIndexFile(io, &file, fresh_header, @intCast(lo)); var block = try segment.readBlockFile(allocator, io, &file, fresh_entry); defer block.deinit(allocator); for (block.events) |event| { if (event.witnessed_at >= time_us) return .{ .seq = @max(event.seq, 1) }; } return .{ .seq = @max(entry.max_seq, 1) };}
/// First fsynced active-region sequence witnessed at/after `time_us`./// Falls back to `sealed_fallback` when the active region cannot answer/// (no active file, no fsynced frames); when every fsynced row is older/// than the requested time, the answer is the next not-yet-durable seq —/// the true near-now anchor.fn resolveTimeInActive( allocator: Allocator, io: Io, archive: *archive_mod.Archive, time_us: i64, sealed_fallback: u64,) !ResolvedCursor { var snapshot = try archive.coldSnapshot(allocator, 0); defer snapshot.deinit(); var name_buf: [64]u8 = undefined; const name = archive_mod.formatSegmentName(&name_buf, snapshot.active_index); var file = archive.dir.openFile(io, name, .{}) catch return .{ .seq = sealed_fallback }; defer file.close(io);
// Walk only the 8-byte length prefixes to learn block boundaries (no // decompression), then binary-search with one decompressed probe per // step. The previous sequential decompress-everything walk took tens of // seconds on a mid-fill active segment and ran inside the websocket // handshake: any time cursor aged past the hot tail but not yet sealed // (a band hours wide) timed the connect out entirely (2026-08-11 coral // wedge). Sealed segments already resolve this way via their block index. var offsets: std.ArrayList(u64) = .empty; defer offsets.deinit(allocator); { var offset: usize = segment.header_size; while (offset < snapshot.active_length) { if (offset + 8 > snapshot.active_length) break; var frame_len_bytes: [8]u8 = undefined; _ = try file.readPositionalAll(io, &frame_len_bytes, offset); const frame_len = std.mem.readInt(u64, &frame_len_bytes, .little); if (frame_len > snapshot.active_length - offset - 8) break; try offsets.append(allocator, offset); offset += 8 + @as(usize, @intCast(frame_len)); } } if (offsets.items.len == 0) return .{ .seq = @max(sealed_fallback, 1) };
const Probe = struct { fn load(alloc: Allocator, io_: Io, f: *Io.File, off: u64, end: u64) !segment.Block { var frame_len_bytes: [8]u8 = undefined; _ = try f.readPositionalAll(io_, &frame_len_bytes, @intCast(off)); const frame_len = std.mem.readInt(u64, &frame_len_bytes, .little); if (frame_len > end - off - 8) return error.Truncated; const compressed = try alloc.alloc(u8, @intCast(frame_len)); _ = try f.readPositionalAll(io_, compressed, @intCast(off + 8)); var frame_reader: std.Io.Reader = .fixed(compressed); var zstd_stream: std.compress.zstd.Decompress = .init(&frame_reader, &.{}, .{}); var decoded: std.Io.Writer.Allocating = .init(alloc); _ = zstd_stream.reader.streamRemaining(&decoded.writer) catch return error.Truncated; return segment.decodeBlock(alloc, try decoded.toOwnedSlice()); } };
// leftmost block whose last event is witnessed at/after time_us var next_after: u64 = sealed_fallback; var lo: usize = 0; var hi: usize = offsets.items.len; while (lo < hi) { const mid = lo + (hi - lo) / 2; var block_arena = std.heap.ArenaAllocator.init(allocator); defer block_arena.deinit(); const block = Probe.load(block_arena.allocator(), io, &file, offsets.items[mid], snapshot.active_length) catch break; if (block.events.len == 0) { // empty block cannot answer; treat as older so the search moves on lo = mid + 1; continue; } const last = block.events[block.events.len - 1]; if (last.witnessed_at >= time_us) { hi = mid; } else { next_after = last.seq +| 1; lo = mid + 1; } } if (lo < offsets.items.len) { var block_arena = std.heap.ArenaAllocator.init(allocator); defer block_arena.deinit(); if (Probe.load(block_arena.allocator(), io, &file, offsets.items[lo], snapshot.active_length)) |block| { for (block.events) |event| { if (event.witnessed_at >= time_us) return .{ .seq = @max(event.seq, 1) }; } } else |_| {} } return .{ .seq = @max(next_after, 1) };}
/// stream all events with witnessed_at in [from_us, until_us) matching/// `filter` to `sink` (fn (ctx, json: []const u8) !void). returns the/// number of frames written.pub fn replay( allocator: Allocator, io: Io, archive: *archive_mod.Archive, filter: anytype, // *Subscriber-shaped: wantsPred via filter fn below comptime wants: fn (@TypeOf(filter), wire.Kind, []const u8, []const u8, tail.SkipV1) bool, sink: anytype, comptime write: fn (@TypeOf(sink), []const u8) anyerror!WriteResult, comptime observe: fn (@TypeOf(sink), u64, ReplayResult) anyerror!void, from_us: i64, until_us: i64,) !u64 { const summaries = try archive.manifest.all(allocator); defer allocator.free(summaries);
var frames: u64 = 0; for (summaries) |summary| { if (summary.header.max_witnessed_at < from_us) continue; if (summary.header.min_witnessed_at >= until_us) break; var blocks = try archive.manifest.blockIndex(summary.idx); defer blocks.deinit(); var name_buffer: [64]u8 = undefined; const name = archive_mod.formatSegmentName(&name_buffer, summary.idx); var file = archive.dir.openFile(io, name, .{}) catch |err| { log.warn("cold: open resident segment {s}: {s}", .{ name, @errorName(err) }); return err; }; defer file.close(io); const fresh_header = try segment.readHeaderFile(io, &file); if (@as(usize, fresh_header.block_count) != blocks.entries.len) return error.BlockTopologyChanged;
for (blocks.entries, 0..) |info, block_index| { if (info.max_witnessed_at < from_us) continue; if (info.min_witnessed_at >= until_us) break; const fresh_entry = try segment.readBlockIndexFile(io, &file, fresh_header, block_index); var block = try segment.readBlockFile(allocator, io, &file, fresh_entry); defer block.deinit(allocator); for (block.events) |ev| { if (ev.witnessed_at < from_us or ev.witnessed_at >= until_us) continue; const kind: wire.Kind = switch (ev.kind) { .create, .update, .delete, .create_resync => .commit, .identity => .identity, .account => .account, .sync => .sync, }; const skip: tail.SkipV1 = switch (ev.kind) { .sync => .sync, .create_resync => .resync_replacement, else => .none, }; if (skip != .none) { try observe(sink, ev.seq, if (skip == .sync) .skipped_sync else .skipped_resync); continue; } if (!wants(filter, kind, ev.did, ev.collection, .none)) { try observe(sink, ev.seq, .filtered); continue; } var arena = std.heap.ArenaAllocator.init(allocator); defer arena.deinit(); const json = encodeEvent(arena.allocator(), ev) catch |err| { log.warn("cold: undecodable row seq={d}: {s}", .{ ev.seq, @errorName(err) }); try observe(sink, ev.seq, .encode_error); continue; } orelse { try observe(sink, ev.seq, .encode_error); continue; }; switch (try write(sink, json)) { .sent => { try observe(sink, ev.seq, .sent); frames += 1; }, .oversize => try observe(sink, ev.seq, .oversize), .encode_error => try observe(sink, ev.seq, .encode_error), } } } } return frames;}
/// stream events with seq >= from_seq up to the hot tail's seq floor on the/// requested endpoint wire. Both v1 and v2 accept seq cursors; only v2 emits/// sync/resync-replacement rows and the v2 envelope fields.pub fn replaySeq( allocator: Allocator, io: Io, archive: *archive_mod.Archive, filter: anytype, comptime wants: fn (@TypeOf(filter), wire.Kind, []const u8, []const u8, tail.SkipV1) bool, sink: anytype, comptime write: fn (@TypeOf(sink), []const u8) anyerror!WriteResult, comptime observe: fn (@TypeOf(sink), u64, ReplayResult) anyerror!void, v2: bool, from_seq: u64,) !u64 { const summaries = try archive.manifest.all(allocator); defer allocator.free(summaries);
var frames: u64 = 0; for (summaries) |summary| { if (summary.header.max_seq < from_seq) continue; var blocks = try archive.manifest.blockIndex(summary.idx); defer blocks.deinit(); var name_buffer: [64]u8 = undefined; const name = archive_mod.formatSegmentName(&name_buffer, summary.idx); var file = archive.dir.openFile(io, name, .{}) catch |err| { log.warn("cold: open sealed segment {s}: {s}", .{ name, @errorName(err) }); return err; }; defer file.close(io); const fresh_header = try segment.readHeaderFile(io, &file); if (@as(usize, fresh_header.block_count) != blocks.entries.len) return error.BlockTopologyChanged;
for (blocks.entries, 0..) |info, block_index| { if (info.max_seq < from_seq) continue; const fresh_entry = try segment.readBlockIndexFile(io, &file, fresh_header, block_index); var block = try segment.readBlockFile(allocator, io, &file, fresh_entry); defer block.deinit(allocator); for (block.events) |ev| { if (ev.seq < from_seq) continue; const kind: wire.Kind = switch (ev.kind) { .create, .update, .delete, .create_resync => .commit, .identity => .identity, .account => .account, .sync => .sync, }; if (!v2) switch (ev.kind) { .sync => { try observe(sink, ev.seq, .skipped_sync); continue; }, .create_resync => { try observe(sink, ev.seq, .skipped_resync); continue; }, else => {}, }; if (!wants(filter, kind, ev.did, ev.collection, .none)) { try observe(sink, ev.seq, .filtered); continue; } var arena = std.heap.ArenaAllocator.init(allocator); defer arena.deinit(); const json = (if (v2) encodeEventV2(arena.allocator(), ev) else encodeEvent(arena.allocator(), ev)) catch { try observe(sink, ev.seq, .encode_error); continue; } orelse { try observe(sink, ev.seq, .encode_error); continue; }; switch (try write(sink, json)) { .sent => { try observe(sink, ev.seq, .sent); frames += 1; }, .oversize => try observe(sink, ev.seq, .oversize), .encode_error => try observe(sink, ev.seq, .encode_error), } } } } return frames;}
/// one archived row → one v2 wire framepub fn encodeEventV2(alloc: Allocator, ev: segment.Event) !?[]u8 { var out: Io.Writer.Allocating = .init(alloc); switch (ev.kind) { .create, .update, .delete, .create_resync => { const is_delete = ev.kind == .delete; var record: ?zat.cbor.Value = null; var cid: ?zat.cbor.Cid = null; if (!is_delete) { record = try zat.cbor.decodeAll(alloc, ev.payload); cid = try zat.cbor.Cid.forDagCbor(alloc, ev.payload); } try wire.encodeCommitV2(alloc, &out.writer, .{ .did = ev.did, .time_us = ev.displayTime(), .rev = ev.rev, .operation = if (ev.kind == .create_resync) "create" else @tagName(ev.kind), .collection = ev.collection, .rkey = ev.rkey, .record = record, .cid = cid, }, ev.seq); }, .identity, .account => { const env = try zat.cbor.decodeAll(alloc, ev.payload); const seq: i64 = env.getInt("seq") orelse @intCast(env.getUint("seq") orelse 0); const time = env.getString("time") orelse ""; if (ev.kind == .identity) { try wire.encodeIdentityV2(&out.writer, .{ .did = ev.did, .time_us = ev.displayTime(), .seq = seq, .time = time, .handle = env.getString("handle"), }, ev.seq); } else { try wire.encodeAccountV2(&out.writer, .{ .did = ev.did, .time_us = ev.displayTime(), .seq = seq, .time = time, .active = env.getBool("active") orelse true, .status = env.getString("status"), }, ev.seq); } }, .sync => { const env = try zat.cbor.decodeAll(alloc, ev.payload); try wire.encodeSyncV2(&out.writer, .{ .did = ev.did, .time_us = ev.displayTime(), .upstream_seq = env.getInt("seq") orelse 0, .rev = ev.rev, .time = env.getString("time") orelse "", .blocks = env.getBytes("blocks") orelse "", }, ev.seq); }, } return try out.toOwnedSlice();}
/// one archived row → one v1 wire frame (null for rows v1 never emits)pub fn encodeEvent(alloc: Allocator, ev: segment.Event) !?[]u8 { var out: Io.Writer.Allocating = .init(alloc); switch (ev.kind) { .create, .update, .delete => { const is_delete = ev.kind == .delete; var record: ?zat.cbor.Value = null; var cid: ?zat.cbor.Cid = null; if (!is_delete) { record = try zat.cbor.decodeAll(alloc, ev.payload); cid = try zat.cbor.Cid.forDagCbor(alloc, ev.payload); } try wire.encodeCommit(alloc, &out.writer, .{ .did = ev.did, .time_us = ev.displayTime(), .rev = ev.rev, .operation = @tagName(ev.kind), .collection = ev.collection, .rkey = ev.rkey, .record = record, .cid = cid, }); }, .identity, .account => { const env = try zat.cbor.decodeAll(alloc, ev.payload); const seq: i64 = env.getInt("seq") orelse @intCast(env.getUint("seq") orelse 0); const time = env.getString("time") orelse ""; if (ev.kind == .identity) { try wire.encodeIdentity(&out.writer, .{ .did = ev.did, .time_us = ev.displayTime(), .seq = seq, .time = time, .handle = env.getString("handle"), }); } else { try wire.encodeAccount(&out.writer, .{ .did = ev.did, .time_us = ev.displayTime(), .seq = seq, .time = time, .active = env.getBool("active") orelse true, .status = env.getString("status"), }); } }, .sync, .create_resync => return null, // never on the v1 wire } return try out.toOwnedSlice();}
// === tests ===
test "indexed timestamp overrides wire display time without changing witness" { const testing = std.testing; var arena = std.heap.ArenaAllocator.init(testing.allocator); defer arena.deinit(); const allocator = arena.allocator(); const base: segment.Event = .{ .seq = 42, .witnessed_at = 1_700_000_000_000_000, .indexed_at = 1_600_000_000_000_000, .kind = .create, .did = "did:plc:timestampfixture", .collection = "app.bsky.feed.post", .rkey = "3k2abcdefghij", .rev = "3k2abcdefghij", .payload = "\xa1\x64text\x62hi", }; const v1 = (try encodeEvent(allocator, base)).?; const v2 = (try encodeEventV2(allocator, base)).?; try testing.expect(std.mem.indexOf(u8, v1, "\"time_us\":1600000000000000") != null); try testing.expect(std.mem.indexOf(u8, v2, "\"time\":\"2020-09-13T12:26:40.000000Z\"") != null); try testing.expectEqual(@as(i64, 1_700_000_000_000_000), base.witnessed_at);
var fallback = base; fallback.indexed_at = 0; const legacy = (try encodeEvent(allocator, fallback)).?; try testing.expect(std.mem.indexOf(u8, legacy, "\"time_us\":1700000000000000") != null);}
test "offline seq replay uses the selected endpoint wire" { 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: [Io.Dir.max_path_bytes]u8 = undefined; const path_len = try tmp.dir.realPath(io, &path_buf); var archive = try archive_mod.Archive.init(testing.allocator, io, path_buf[0..path_len]); defer archive.deinit(); archive.max_events_per_block = 1; archive.writer.max_events_per_block = 1; _ = try archive.append(.{ .seq = 0, .witnessed_at = 1_700_000_000_000_000, .indexed_at = 0, .kind = .create, .did = "did:plc:coldwirefixture", .collection = "app.bsky.feed.post", .rkey = "3k2abcdefghij", .rev = "3k2abcdefghij", .payload = "\xa1\x64text\x62hi", }, 1_700_000_000_000_000); try archive.rotate();
const Sink = struct { allocator: Allocator, frames: std.ArrayList([]u8) = .empty,
fn deinit(self: *@This()) void { for (self.frames.items) |frame| self.allocator.free(frame); self.frames.deinit(self.allocator); } fn wants(_: *@This(), _: wire.Kind, _: []const u8, _: []const u8, _: tail.SkipV1) bool { return true; } fn write(self: *@This(), json: []const u8) anyerror!WriteResult { try self.frames.append(self.allocator, try self.allocator.dupe(u8, json)); return .sent; } fn observe(_: *@This(), _: u64, _: ReplayResult) anyerror!void {} };
var v1: Sink = .{ .allocator = testing.allocator }; defer v1.deinit(); try testing.expectEqual(@as(u64, 1), try replaySeq(testing.allocator, io, &archive, &v1, Sink.wants, &v1, Sink.write, Sink.observe, false, 1)); try testing.expect(std.mem.indexOf(u8, v1.frames.items[0], "\"seq\":1") == null); try testing.expect(std.mem.indexOf(u8, v1.frames.items[0], "record_cbor") == null);
var v2: Sink = .{ .allocator = testing.allocator }; defer v2.deinit(); try testing.expectEqual(@as(u64, 1), try replaySeq(testing.allocator, io, &archive, &v2, Sink.wants, &v2, Sink.write, Sink.observe, true, 1)); try testing.expect(std.mem.indexOf(u8, v2.frames.items[0], "\"seq\":1") != null); try testing.expect(std.mem.indexOf(u8, v2.frames.items[0], "subscribeEvents#commit") != null); try testing.expect(std.mem.indexOf(u8, v2.frames.items[0], "record_cbor") == null);}
test "cached cold batches bound raw scans before filtering and stop at hot floor" { 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: [Io.Dir.max_path_bytes]u8 = undefined; const path_len = try tmp.dir.realPath(io, &path_buf); var archive = try archive_mod.Archive.init(testing.allocator, io, path_buf[0..path_len]); defer archive.deinit(); archive.max_events_per_block = 2; archive.writer.max_events_per_block = 2; for (0..5) |i| { _ = try archive.append(.{ .seq = 0, .witnessed_at = 1_700_000_000_000_000 + @as(i64, @intCast(i)), .indexed_at = 0, .kind = .delete, .did = "did:plc:coldbatchfixture", .collection = "app.bsky.feed.post", .rkey = "3k2abcdefghij", .rev = "3k2abcdefghij", .payload = "", }, 1_700_000_000_000_000); } try archive.rotate();
var reader = ColdReader.init(testing.allocator, io, 1 << 20); defer reader.deinit(); const Sink = struct { observed: usize = 0,
fn wants(_: *@This(), _: wire.Kind, _: []const u8, _: []const u8, _: tail.SkipV1) bool { return false; } fn write(_: *@This(), _: CachedFrame) anyerror!WriteResult { return error.UnexpectedWrite; } fn observe(self: *@This(), _: u64, result: ReplayResult) anyerror!void { try testing.expectEqual(ReplayResult.filtered, result); self.observed += 1; } }; var sink: Sink = .{}; const first = try reader.readSeqBatch(&archive, &sink, Sink.wants, &sink, Sink.write, Sink.observe, false, false, 1, 5, 2); try testing.expectEqual(@as(usize, 2), first.scanned); try testing.expectEqual(@as(u64, 3), first.next_seq); try testing.expectEqual(@as(u64, 0), first.frames); try testing.expect(!first.exhausted);
const second = try reader.readSeqBatch(&archive, &sink, Sink.wants, &sink, Sink.write, Sink.observe, false, false, first.next_seq, 5, 2); try testing.expectEqual(@as(usize, 2), second.scanned); try testing.expectEqual(@as(u64, 5), second.next_seq); try testing.expectEqual(@as(u64, 0), second.frames); try testing.expect(!second.exhausted); try testing.expectEqual(@as(usize, 4), sink.observed);
const handoff = try reader.readSeqBatch(&archive, &sink, Sink.wants, &sink, Sink.write, Sink.observe, false, false, second.next_seq, 5, 2); try testing.expectEqual(@as(usize, 0), handoff.scanned); try testing.expectEqual(@as(u64, 5), handoff.next_seq); try testing.expect(handoff.exhausted);}
test "cold batches include the active segment fsynced prefix" { 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: [Io.Dir.max_path_bytes]u8 = undefined; const path_len = try tmp.dir.realPath(io, &path_buf); var archive = try archive_mod.Archive.init(testing.allocator, io, path_buf[0..path_len]); defer archive.deinit(); archive.max_events_per_block = 2; archive.writer.max_events_per_block = 2; archive.flush_interval_us = std.math.maxInt(i64); for (0..4) |i| { _ = try archive.append(.{ .seq = 0, .witnessed_at = 1_700_000_000_000_000 + @as(i64, @intCast(i)), .indexed_at = 0, .kind = .delete, .did = "did:plc:activebatchfixture", .collection = "app.bsky.feed.post", .rkey = "3k2abcdefghij", .rev = "3k2abcdefghij", .payload = "", }, 1); } const summaries = try archive.manifest.all(testing.allocator); defer testing.allocator.free(summaries); try testing.expectEqual(@as(usize, 0), summaries.len); try testing.expectEqual(@as(u64, 4), archive.committed_seq.load(.acquire));
var reader = ColdReader.init(testing.allocator, io, 1 << 20); defer reader.deinit(); const Sink = struct { seqs: [4]u64 = undefined, count: usize = 0,
fn wants(_: *@This(), _: wire.Kind, _: []const u8, _: []const u8, _: tail.SkipV1) bool { return true; } fn write(_: *@This(), _: CachedFrame) anyerror!WriteResult { return .sent; } fn observe(self: *@This(), seq: u64, result: ReplayResult) anyerror!void { if (result != .sent) return; self.seqs[self.count] = seq; self.count += 1; } }; var sink: Sink = .{}; const first = try reader.readSeqBatch(&archive, &sink, Sink.wants, &sink, Sink.write, Sink.observe, false, false, 1, 5, 3); try testing.expectEqual(@as(usize, 3), first.scanned); try testing.expectEqual(@as(u64, 4), first.next_seq); const second = try reader.readSeqBatch(&archive, &sink, Sink.wants, &sink, Sink.write, Sink.observe, false, false, first.next_seq, 5, 3); try testing.expectEqual(@as(usize, 1), second.scanned); try testing.expectEqual(@as(u64, 5), second.next_seq); try testing.expectEqualSlices(u64, &.{ 1, 2, 3, 4 }, sink.seqs[0..sink.count]);}
// Mirrors TestActiveFlushedRangeExcludesPendingAndAdvancesByBlock; poisoned// prefix frames make forbidden earlier reads fail, including after resume.test "late active batches seek past unreadable prefix and survive rotation" { 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: [Io.Dir.max_path_bytes]u8 = undefined; const path_len = try tmp.dir.realPath(io, &path_buf); var archive = try archive_mod.Archive.init(testing.allocator, io, path_buf[0..path_len]); defer archive.deinit(); archive.writer.max_events_per_block = 2; archive.flush_interval_us = std.math.maxInt(i64); for (0..2051) |i| { _ = try archive.append(.{ .seq = 0, .witnessed_at = @intCast(i + 1), .indexed_at = 0, .kind = .delete, .did = "did:plc:activeseekfixture", .collection = "app.bsky.feed.post", .rkey = "3k2abcdefghij", .rev = "3k2abcdefghij", .payload = "", }, 1); } var pending = try archive.coldSnapshot(testing.allocator, 2051); defer pending.deinit(); try testing.expectEqual(pending.active_length, pending.active_start); try testing.expectEqual(@as(u64, 2050), archive.committed_seq.load(.acquire)); // Resume rebuilds the same block index from the durable file image. archive.deinit(); archive = try archive_mod.Archive.init(testing.allocator, io, path_buf[0..path_len]); var snapshot = try archive.coldSnapshot(testing.allocator, 2048); defer snapshot.deinit(); const infos = archive.writer.builder.infos.items; try testing.expectEqual(@as(usize, @intCast(infos[1023].offset)), snapshot.active_start); for (infos[0..1023]) |info| { try @import("segment_io.zig").writePositionalAll(&archive.file, io, &(@as([8]u8, @splat(255))), info.offset); } var reader = ColdReader.init(testing.allocator, io, 1 << 20); defer reader.deinit(); const Sink = struct { next: u64 = 2048, fn wants(_: *@This(), _: wire.Kind, _: []const u8, _: []const u8, _: tail.SkipV1) bool { return false; } fn write(_: *@This(), _: CachedFrame) anyerror!WriteResult { return error.UnexpectedWrite; } fn observe(self: *@This(), seq: u64, result: ReplayResult) anyerror!void { try testing.expectEqual(ReplayResult.filtered, result); try testing.expectEqual(self.next, seq); self.next += 1; } }; var sink: Sink = .{}; const first = try reader.readSeqBatch(&archive, &sink, Sink.wants, &sink, Sink.write, Sink.observe, false, false, 2048, 0, 1); try testing.expectEqual(@as(usize, 1), first.scanned); try testing.expectEqual(@as(u64, 2049), first.next_seq); try archive.rotate(); // The captured generation still names the same byte-stable prefix. var name_buf: [64]u8 = undefined; var old_file = try archive.dir.openFile(io, archive_mod.formatSegmentName(&name_buf, snapshot.active_index), .{}); defer old_file.close(io); var prefix: [8]u8 = undefined; try testing.expectEqual(@as(usize, 8), try old_file.readPositionalAll(io, &prefix, snapshot.active_start)); try testing.expect(std.mem.readInt(u64, &prefix, .little) < snapshot.active_length - snapshot.active_start); const second = try reader.readSeqBatch(&archive, &sink, Sink.wants, &sink, Sink.write, Sink.observe, false, false, first.next_seq, 2051, 1); try testing.expectEqual(@as(usize, 1), second.scanned); try testing.expectEqual(@as(u64, 2050), second.next_seq); const third = try reader.readSeqBatch(&archive, &sink, Sink.wants, &sink, Sink.write, Sink.observe, false, false, second.next_seq, 2051, 8); try testing.expectEqual(@as(usize, 1), third.scanned); try testing.expectEqual(@as(u64, 2051), third.next_seq); try testing.expectEqual(@as(u64, 2051), sink.next);}
// Upstream's walkSealedSegment returns `open seg %d: %w` and leaves the// replay cursor where it was. Skipping an unreadable manifest-listed segment// would emit the following segment's events and advance the cursor past// durable events the subscriber never received -- an undetectable hole.test "cold replay refuses to skip an unreadable sealed segment" { 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: [Io.Dir.max_path_bytes]u8 = undefined; const path_len = try tmp.dir.realPath(io, &path_buf); var archive = try archive_mod.Archive.init(testing.allocator, io, path_buf[0..path_len]); defer archive.deinit();
for (0..4) |i| { _ = try archive.append(.{ .seq = 0, .witnessed_at = 1_700_000_000_000_000 + @as(i64, @intCast(i)), .indexed_at = 0, .kind = .create, .did = "did:plc:coldholefixture", .collection = "app.bsky.feed.post", .rkey = "3k2abcdefghij", .rev = "3k2abcdefghij", .payload = "\xa1\x64text\x62hi", }, 1_700_000_000_000_000 + @as(i64, @intCast(i))); if (i == 1) try archive.rotate(); } try archive.rotate();
const summaries = try archive.manifest.all(testing.allocator); defer testing.allocator.free(summaries); try testing.expectEqual(@as(usize, 2), summaries.len);
// The manifest still lists segment 0; the file is gone underneath it. try archive.dir.deleteFile(io, "seg_0000000000.jss");
var reader = ColdReader.init(testing.allocator, io, 1 << 20); defer reader.deinit(); const Sink = struct { delivered: usize = 0,
fn wants(_: *@This(), _: wire.Kind, _: []const u8, _: []const u8, _: tail.SkipV1) bool { return true; } fn write(_: *@This(), _: CachedFrame) anyerror!WriteResult { return .sent; } fn observe(self: *@This(), _: u64, result: ReplayResult) anyerror!void { if (result == .sent) self.delivered += 1; } }; var sink: Sink = .{}; try testing.expectError(error.FileNotFound, reader.readSeqBatch( &archive, &sink, Sink.wants, &sink, Sink.write, Sink.observe, false, false, 1, 5, 16, )); // Nothing from the intact later segment may be emitted over the hole. try testing.expectEqual(@as(usize, 0), sink.delivered);}
// Same invariant on the second cold walker: replaySeq must not step over an// unreadable manifest-listed segment to reach the intact one behind it.test "replaySeq refuses to skip an unreadable sealed segment" { 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: [Io.Dir.max_path_bytes]u8 = undefined; const path_len = try tmp.dir.realPath(io, &path_buf); var archive = try archive_mod.Archive.init(testing.allocator, io, path_buf[0..path_len]); defer archive.deinit();
for (0..4) |i| { _ = try archive.append(.{ .seq = 0, .witnessed_at = 1_700_000_000_000_000 + @as(i64, @intCast(i)), .indexed_at = 0, .kind = .create, .did = "did:plc:replayseqhole", .collection = "app.bsky.feed.post", .rkey = "3k2abcdefghij", .rev = "3k2abcdefghij", .payload = "\xa1\x64text\x62hi", }, 1_700_000_000_000_000 + @as(i64, @intCast(i))); if (i == 1) try archive.rotate(); } try archive.rotate(); try archive.dir.deleteFile(io, "seg_0000000000.jss");
const Sink = struct { delivered: usize = 0,
fn wants(_: *@This(), _: wire.Kind, _: []const u8, _: []const u8, _: tail.SkipV1) bool { return true; } fn write(_: *@This(), _: []const u8) anyerror!WriteResult { return .sent; } fn observe(self: *@This(), _: u64, result: ReplayResult) anyerror!void { if (result == .sent) self.delivered += 1; } }; var sink: Sink = .{}; try testing.expectError(error.FileNotFound, replaySeq( testing.allocator, io, &archive, &sink, Sink.wants, &sink, Sink.write, Sink.observe, false, 1, )); try testing.expectEqual(@as(usize, 0), sink.delivered);}
test "cold cache single-flights real block decode and shares wire memos" { 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: [Io.Dir.max_path_bytes]u8 = undefined; const path_len = try tmp.dir.realPath(io, &path_buf); var archive = try archive_mod.Archive.init(testing.allocator, io, path_buf[0..path_len]); defer archive.deinit(); archive.max_events_per_block = 1; archive.writer.max_events_per_block = 1; _ = try archive.append(.{ .seq = 0, .witnessed_at = 1_700_000_000_000_000, .indexed_at = 0, .kind = .create, .did = "did:plc:coldcachefixture", .collection = "app.bsky.feed.post", .rkey = "3k2abcdefghij", .rev = "3k2abcdefghij", .payload = "\xa1\x64text\x62hi", }, 1_700_000_000_000_000); try archive.rotate();
var reader = ColdReader.init(testing.allocator, io, 1 << 20); defer reader.deinit(); const Sink = struct { body: []const u8 = "", payload: []const u8 = "",
fn wants(_: *@This(), _: wire.Kind, _: []const u8, _: []const u8, _: tail.SkipV1) bool { return true; } fn write(self: *@This(), frame_value: CachedFrame) anyerror!WriteResult { self.body = frame_value.body; self.payload = frame_value.payload; try testing.expect(frame_value.compressed); return .sent; } fn observe(_: *@This(), _: u64, _: ReplayResult) anyerror!void {} }; const Worker = struct { start: *Io.Event, reader: *ColdReader, archive: *archive_mod.Archive, sink: Sink = .{}, frames: u64 = 0, failed: bool = false,
fn run(self: *@This()) void { self.start.waitUncancelable(self.reader.cache.io); self.frames = self.reader.replaySeq( self.archive, &self.sink, Sink.wants, &self.sink, Sink.write, Sink.observe, true, true, 1, ) catch { self.failed = true; return; }; } };
var start: Io.Event = .unset; var workers: [8]Worker = undefined; var threads: [8]std.Thread = undefined; for (&workers, &threads) |*worker, *thread| { worker.* = .{ .start = &start, .reader = &reader, .archive = &archive }; thread.* = try std.Thread.spawn(.{ .stack_size = 8 * 1024 * 1024 }, Worker.run, .{worker}); } start.set(io); for (&threads) |*thread| thread.join();
const body_ptr = @intFromPtr(workers[0].sink.body.ptr); const payload_ptr = @intFromPtr(workers[0].sink.payload.ptr); try testing.expect(body_ptr != 0 and payload_ptr != 0 and body_ptr != payload_ptr); for (workers) |worker| { try testing.expect(!worker.failed); try testing.expectEqual(@as(u64, 1), worker.frames); try testing.expectEqual(body_ptr, @intFromPtr(worker.sink.body.ptr)); try testing.expectEqual(payload_ptr, @intFromPtr(worker.sink.payload.ptr)); } try testing.expectEqual(@as(u64, 1), reader.cache.decode_count.load(.monotonic)); var reference = try zstd.DictCompressor.initLevel(&zstd.v2_dictionary, 1); defer reference.deinit(); const reference_frame = try reference.compress(testing.allocator, workers[0].sink.body); defer testing.allocator.free(reference_frame); try testing.expectEqualSlices(u8, reference_frame, workers[0].sink.payload);
// The cached v2 JSON+zstd bodies are resident. Leave one byte of budget // headroom, then materialize the independent v1 memo: that growth must // evict the item rather than silently exceed the configured budget. reader.cache.max_bytes = reader.cache.bytes() + 1; const PlainSink = struct { fn wants(_: *@This(), _: wire.Kind, _: []const u8, _: []const u8, _: tail.SkipV1) bool { return true; } fn write(_: *@This(), frame_value: CachedFrame) anyerror!WriteResult { try testing.expect(!frame_value.compressed); return .sent; } fn observe(_: *@This(), _: u64, _: ReplayResult) anyerror!void {} }; var plain: PlainSink = .{}; try testing.expectEqual(@as(u64, 1), try reader.replaySeq(&archive, &plain, PlainSink.wants, &plain, PlainSink.write, PlainSink.observe, false, false, 1)); try testing.expectEqual(@as(usize, 0), reader.cache.bytes());}
test "manifest refresh invalidates the production cold cache" { 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: [Io.Dir.max_path_bytes]u8 = undefined; const path_len = try tmp.dir.realPath(io, &path_buf); var archive = try archive_mod.Archive.init(testing.allocator, io, path_buf[0..path_len]); defer archive.deinit(); archive.max_events_per_block = 1; archive.writer.max_events_per_block = 1; _ = try archive.append(.{ .seq = 0, .witnessed_at = 1_700_000_000_000_000, .indexed_at = 0, .kind = .delete, .did = "did:plc:coldrefreshfixture", .collection = "app.bsky.feed.post", .rkey = "3k2abcdefghij", .rev = "3k2abcdefghij", .payload = "", }, 1_700_000_000_000_000); try archive.rotate();
var reader = ColdReader.init(testing.allocator, io, 1 << 20); defer reader.deinit(); archive.manifest.setInvalidator(&reader, ColdReader.invalidateOpaque); defer archive.manifest.setInvalidator(null, null); const Sink = struct { fn wants(_: *@This(), _: wire.Kind, _: []const u8, _: []const u8, _: tail.SkipV1) bool { return true; } fn write(_: *@This(), _: CachedFrame) anyerror!WriteResult { return .sent; } fn observe(_: *@This(), _: u64, _: ReplayResult) anyerror!void {} }; var sink: Sink = .{}; try testing.expectEqual(@as(u64, 1), try reader.replaySeq(&archive, &sink, Sink.wants, &sink, Sink.write, Sink.observe, false, false, 1)); try testing.expect(reader.cache.bytes() > 0); try testing.expectEqual(@as(u64, 1), reader.cache.decode_count.load(.monotonic));
// Refresh the exact same real file. The callback must still invalidate // its generation: a rewrite can retain topology while changing bytes. try archive.manifest.refresh(archive.dir, 0, true); try testing.expectEqual(@as(usize, 0), reader.cache.bytes()); try testing.expectEqual(@as(u64, 1), try reader.replaySeq(&archive, &sink, Sink.wants, &sink, Sink.write, Sink.observe, false, false, 1)); try testing.expectEqual(@as(u64, 2), reader.cache.decode_count.load(.monotonic));}
test "segment invalidation prevents a real in-flight decode from reinserting" { 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: [Io.Dir.max_path_bytes]u8 = undefined; const path_len = try tmp.dir.realPath(io, &path_buf); var archive = try archive_mod.Archive.init(testing.allocator, io, path_buf[0..path_len]); defer archive.deinit(); archive.max_events_per_block = 1; archive.writer.max_events_per_block = 1; _ = try archive.append(.{ .seq = 0, .witnessed_at = 1_700_000_000_000_000, .indexed_at = 0, .kind = .delete, .did = "did:plc:coldinflightfixture", .collection = "app.bsky.feed.post", .rkey = "3k2abcdefghij", .rev = "3k2abcdefghij", .payload = "", }, 1_700_000_000_000_000); try archive.rotate();
var file = try archive.dir.openFile(io, "seg_0000000000.jss", .{}); defer file.close(io); const header = try segment.readHeaderFile(io, &file); const entry = try segment.readBlockIndexFile(io, &file, header, 0); var cache = cold_cache.Cache.init(testing.allocator, io, 1 << 20); defer cache.deinit(); const key_value = cache.key(0, header.checksum, 0); var started: Io.Event = .unset; var release: Io.Event = .unset; const Decode = struct { io: Io, file: *Io.File, entry: segment.BlockIndexEntry, started: *Io.Event, release: *Io.Event,
fn run(ctx: @This(), allocator: Allocator) anyerror!segment.Block { ctx.started.set(ctx.io); ctx.release.waitUncancelable(ctx.io); return segment.readBlockFile(allocator, ctx.io, ctx.file, ctx.entry); } fn direct(ctx: @This(), allocator: Allocator) anyerror!segment.Block { return segment.readBlockFile(allocator, ctx.io, ctx.file, ctx.entry); } }; const Worker = struct { cache: *cold_cache.Cache, key_value: cold_cache.Key, decode: Decode, failed: bool = false,
fn run(self: *@This()) void { var handle = self.cache.acquire(self.key_value, self.decode, Decode.run) catch { self.failed = true; return; }; handle.deinit(); } }; const decode: Decode = .{ .io = io, .file = &file, .entry = entry, .started = &started, .release = &release }; var worker: Worker = .{ .cache = &cache, .key_value = key_value, .decode = decode }; var thread = try std.Thread.spawn(.{ .stack_size = 8 * 1024 * 1024 }, Worker.run, .{&worker}); started.waitUncancelable(io); try cache.invalidateSegment(0); release.set(io); thread.join(); try testing.expect(!worker.failed); try testing.expectEqual(@as(usize, 0), cache.bytes()); try testing.expectEqual(@as(u64, 1), cache.decode_count.load(.monotonic));
// A post-invalidation acquisition uses the new generation and performs a // second real zstd/file decode; the stale in-flight result never became a // resident hit. var current = try cache.acquire(cache.key(0, header.checksum, 0), decode, Decode.direct); defer current.deinit(); try testing.expect(cache.bytes() > 0); try testing.expectEqual(@as(u64, 2), cache.decode_count.load(.monotonic));}
test "timestamp cursor resolves exact seq from real sealed blocks" { 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: [Io.Dir.max_path_bytes]u8 = undefined; const path_len = try tmp.dir.realPath(io, &path_buf); var archive = try archive_mod.Archive.init(testing.allocator, io, path_buf[0..path_len]); defer archive.deinit(); archive.max_events_per_block = 2; archive.writer.max_events_per_block = 2;
for (0..4) |i| { _ = try archive.append(.{ .seq = 0, .witnessed_at = 1_700_000_000_001_000 + @as(i64, @intCast(i)) * 1_000, .indexed_at = 0, .kind = .delete, .did = "did:plc:cursorfixture", .collection = "app.bsky.feed.post", .rkey = "3k2abcdefghij", .rev = "3k2abcdefghij", .payload = "", }, 1_700_000_000_001_000); } try archive.rotate();
const old = try resolveTimeToSeq(testing.allocator, io, &archive, 1_700_000_000_000_000); try testing.expectEqual(@as(u64, 1), old.seq); try testing.expect(old.clamped); const middle = try resolveTimeToSeq(testing.allocator, io, &archive, 1_700_000_000_002_500); try testing.expectEqual(@as(u64, 3), middle.seq); try testing.expect(!middle.clamped); const newer = try resolveTimeToSeq(testing.allocator, io, &archive, 1_700_000_000_006_000); try testing.expectEqual(@as(u64, 5), newer.seq); try testing.expect(!newer.clamped);}
test "timestamp cursor resolves into the active region, not the sealed boundary" { // the 2026-08-08 treadmill: a near-now cursor resolved to sealed_max+1 — // up to a whole open segment back — and the cold walk could not outrun // live ingest. times inside the fsynced active region must resolve there. 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: [Io.Dir.max_path_bytes]u8 = undefined; const path_len = try tmp.dir.realPath(io, &path_buf); var archive = try archive_mod.Archive.init(testing.allocator, io, path_buf[0..path_len]); defer archive.deinit(); archive.max_events_per_block = 2; archive.writer.max_events_per_block = 2;
const base: i64 = 1_700_000_000_000_000; for (0..4) |i| { _ = try archive.append(.{ .seq = 0, .witnessed_at = base + 1_000 + @as(i64, @intCast(i)) * 1_000, .indexed_at = 0, .kind = .delete, .did = "did:plc:activefixture", .collection = "app.bsky.feed.post", .rkey = "3k2abcdefghij", .rev = "3k2abcdefghij", .payload = "", }, base + 1_000); } try archive.rotate(); // two rows into the fresh active segment (one full flushed block) for (0..2) |i| { _ = try archive.append(.{ .seq = 0, .witnessed_at = base + 5_000 + @as(i64, @intCast(i)) * 1_000, .indexed_at = 0, .kind = .delete, .did = "did:plc:activefixture", .collection = "app.bsky.feed.post", .rkey = "3k2abcdefghij", .rev = "3k2abcdefghij", .payload = "", }, base + 5_000); }
// between sealed max and the first active row: first active row const boundary = try resolveTimeToSeq(testing.allocator, io, &archive, base + 4_500); try testing.expectEqual(@as(u64, 5), boundary.seq); // inside the active region: the exact row, not sealed_max+1 const inside = try resolveTimeToSeq(testing.allocator, io, &archive, base + 5_800); try testing.expectEqual(@as(u64, 6), inside.seq); // newer than every fsynced row: the next not-yet-durable seq const ahead = try resolveTimeToSeq(testing.allocator, io, &archive, base + 9_000); try testing.expectEqual(@as(u64, 7), ahead.seq);}
test "active-region time resolution: binary search sweeps every boundary like the linear walk" { // 2026-08-11 coral wedge regression: the sequential decompress walk this // replaced was so slow mid-fill that connects timed out. correctness of // the replacement: every time between/at/past rows must resolve to the // same seq the exhaustive definition gives — first fsynced active row // witnessed at/after the time, else the next not-yet-durable seq. 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: [Io.Dir.max_path_bytes]u8 = undefined; const path_len = try tmp.dir.realPath(io, &path_buf); var archive = try archive_mod.Archive.init(testing.allocator, io, path_buf[0..path_len]); defer archive.deinit(); archive.max_events_per_block = 2; archive.writer.max_events_per_block = 2;
const base: i64 = 1_700_000_000_000_000; _ = try archive.append(.{ .seq = 0, .witnessed_at = base, .indexed_at = 0, .kind = .delete, .did = "did:plc:sweepfixture", .collection = "app.bsky.feed.post", .rkey = "3k2abcdefghij", .rev = "3k2abcdefghij", .payload = "", }, base); try archive.rotate(); // 21 rows -> ten flushed 2-row blocks plus one pending (not fsynced) row var times: [21]i64 = undefined; for (0..21) |i| { times[i] = base + 1_000 + @as(i64, @intCast(i)) * 137; _ = try archive.append(.{ .seq = 0, .witnessed_at = times[i], .indexed_at = 0, .kind = .delete, .did = "did:plc:sweepfixture", .collection = "app.bsky.feed.post", .rkey = "3k2abcdefghij", .rev = "3k2abcdefghij", .payload = "", }, times[i]); }
// rows 1..20 are seqs 2..21; the fsynced prefix covers seqs 2..21 (ten // blocks); the 21st appended row (seq 22) is pending. exhaustive check: var t = times[0] - 1; while (t <= times[20] + 200) : (t += 1) { var expected: u64 = 22; // next not-yet-durable for (times[0..20], 0..) |w, i| { if (w >= t) { expected = 2 + @as(u64, @intCast(i)); break; } } const got = try resolveTimeToSeq(testing.allocator, io, &archive, t); try testing.expectEqual(expected, got.seq); }}
test "timestamp cursor surfaces corrupt selected block" { 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: [Io.Dir.max_path_bytes]u8 = undefined; const path_len = try tmp.dir.realPath(io, &path_buf); var archive = try archive_mod.Archive.init(testing.allocator, io, path_buf[0..path_len]); defer archive.deinit(); archive.max_events_per_block = 1; archive.writer.max_events_per_block = 1;
for (0..2) |i| { _ = try archive.append(.{ .seq = 0, .witnessed_at = 1_700_000_000_001_000 + @as(i64, @intCast(i)) * 1_000, .indexed_at = 0, .kind = .delete, .did = "did:plc:corruptcursorfixture", .collection = "app.bsky.feed.post", .rkey = "3k2abcdefghij", .rev = "3k2abcdefghij", .payload = "", }, 1_700_000_000_001_000); } try archive.rotate();
var file = try archive.dir.openFile(io, "seg_0000000000.jss", .{ .mode = .read_write }); var header_buf: [segment.header_size]u8 = undefined; _ = try file.readPositionalAll(io, &header_buf, 0); const header = try segment.Header.decode(&header_buf); var entry_buf: [segment.BlockIndexEntry.encoded_size]u8 = undefined; _ = try file.readPositionalAll(io, &entry_buf, header.block_index_offset + segment.BlockIndexEntry.encoded_size); const entry = segment.BlockIndexEntry.decodePublic(&entry_buf); try file.writePositionalAll(io, &@as([8]u8, @splat(0)), entry.offset); try file.sync(io); file.close(io);
try testing.expectError( error.BadBlockFrame, resolveTimeToSeq(testing.allocator, io, &archive, 1_700_000_000_002_000), );}
test "cursor combines resident envelopes with fresh generation offsets" { 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: [Io.Dir.max_path_bytes]u8 = undefined; const path_len = try tmp.dir.realPath(io, &path_buf); var archive = try archive_mod.Archive.init(testing.allocator, io, path_buf[0..path_len]); defer archive.deinit(); archive.max_events_per_block = 1; archive.writer.max_events_per_block = 1;
for (0..3) |index| { _ = try archive.append(.{ .seq = 0, .witnessed_at = 1_700_000_000_001_000 + @as(i64, @intCast(index)) * 1_000, .indexed_at = 0, .kind = .create, .did = "did:plc:generationfixture", .collection = "app.bsky.feed.post", .rkey = "3k2abcdefghij", .rev = "3k2abcdefghij", .payload = "\xa1\x64text\x62hi", }, 1_700_000_000_001_000); } try archive.rotate(); var resident = try archive.manifest.blockIndex(0); defer resident.deinit(); const old_offset = resident.entries[2].offset;
const Keep = struct { fn row(_: void, event: segment.Event) bool { return event.seq >= 2; } }; const rewritten = try rewrite_mod.rewriteFile( testing.allocator, io, archive.dir, "seg_0000000000.jss", .{}, {}, Keep.row, ); try testing.expect(rewritten.rewritten);
var file = try archive.dir.openFile(io, "seg_0000000000.jss", .{}); defer file.close(io); const fresh_header = try segment.readHeaderFile(io, &file); const fresh_third = try segment.readBlockIndexFile(io, &file, fresh_header, 2); try testing.expect(old_offset != fresh_third.offset);
// Deliberately do not refresh the manifest: this models an in-flight // reader that retained the old immutable snapshot across the rename. const resolved = try resolveTimeToSeq(testing.allocator, io, &archive, 1_700_000_000_003_000); try testing.expectEqual(@as(u64, 3), resolved.seq);}