Something went wrong. Try again.
jetstream v2 in zig stream.waow.tech
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716//! in-memory hot tail — bounded deque of encoded v1 frames//!//! pull-based fan-out, following upstream's readable-log shape//! (docs/upstream-harness.md): the ingest thread appends encode-once frames;//! each subscriber thread reads at its own absolute index and blocks on a//! condvar at the tip. no per-subscriber queues, so a slow reader costs//! nothing but its own lag. eviction is by byte budget; a reader that falls//! below the floor resumes from the oldest retained frame (there is no cold//! store yet — that comes with the archive).
const std = @import("std");const wire = @import("wire.zig");
const Io = std.Io;const Allocator = std.mem.Allocator;
pub const SkipV1 = enum { none, sync, resync_replacement };
pub const Entry = struct { kind: wire.Kind, did: []u8, collection: []u8, time_us: i64, /// local archive seq (0 when the row wasn't archived) seq: u64, /// v1 wire; empty for kinds v1 never emits (#sync) json: []u8, /// v2 superset wire json_v2: []u8, skip_v1: SkipV1,
fn deinit(self: Entry, allocator: Allocator) void { allocator.free(self.did); allocator.free(self.collection); allocator.free(self.json); allocator.free(self.json_v2); }};
/// One immutable copied entry in a subscriber pull batch. The caller supplies/// a short-lived arena, so all slices remain valid until that batch has been/// processed without pinning the writer's deque against eviction.pub const BatchEntry = struct { kind: wire.Kind, did: []const u8, collection: []const u8, seq: u64, json: []const u8, skip_v1: SkipV1,};
pub const Batch = struct { entries: []BatchEntry,};
pub const ReadResult = union(enum) { batch: Batch, fell_behind, closed,};
pub const Tail = struct { allocator: Allocator, io: Io, mutex: Io.Mutex = .init, cond: Io.Condition = .init, entries: std.ArrayList(Entry) = .empty, /// Prefix slots are retired in O(1); compactLocked amortizes their removal /// instead of shifting the whole retained log for every eviction. head: usize = 0, /// absolute index of entries.items[head] base: u64 = 0, cur_bytes: usize = 0, /// Bytes belonging to seq-bearing rows newer than published_seq. Kept as /// an accumulator so a Prometheus scrape stays O(1) under the tail lock. pinned_bytes: usize = 0, max_bytes: usize, closed: bool = false, /// When enabled, rows newer than the durable watermark are pinned against /// eviction (upstream ReadableLog: the hot log is the only copy until the /// archive fsyncs them). Visibility is not affected: readers always see /// every appended row, matching upstream ReadFrom. durability_gated: bool = false, published_seq: u64 = 0, /// Highest nonzero seq ever appended (monotone; survives eviction). /// O(1) upper bound for cursor resolution: a cursor above it cannot be /// resident, so indexForSeq and publishedNextSeq never scan for it. The /// 2026-08-09 ingest convoy came from exactly that scan: every seq-cursor /// subscriber walked all ~165k resident entries under this mutex on every /// read-loop pass once the deque reached full retention. max_appended_seq: u64 = 0,
pub fn init(allocator: Allocator, io: Io, max_bytes: usize) Tail { return .{ .allocator = allocator, .io = io, .max_bytes = max_bytes }; }
pub fn enableDurabilityGate(self: *Tail, durable_seq: u64) void { self.mutex.lockUncancelable(self.io); defer self.mutex.unlock(self.io); self.durability_gated = true; self.published_seq = durable_seq; self.pinned_bytes = 0; for (self.liveItems()) |entry| { if (entry.seq != 0 and entry.seq > durable_seq) self.pinned_bytes += entryBytes(entry); } self.evictLocked(); }
/// Advance the durable watermark after Archive's fsync + metadata /// transaction. Unpins the newly durable rows so eviction may reclaim /// them; does not change reader visibility (upstream advanceDurable). pub fn publishDurable(self: *Tail, durable_seq: u64) void { self.mutex.lockUncancelable(self.io); defer self.mutex.unlock(self.io); const old_durable = self.published_seq; if (durable_seq > old_durable) { // Seq-bearing rows are ordered. Walk only the undurable suffix, // not the entire retained log, then stop at the prior watermark. const live = self.liveItems(); var i = live.len; while (i > 0) { i -= 1; const entry = live[i]; if (entry.seq == 0 or entry.seq > durable_seq) continue; if (entry.seq <= old_durable) break; self.pinned_bytes -= entryBytes(entry); } self.published_seq = durable_seq; } self.evictLocked(); }
pub fn deinit(self: *Tail) void { for (self.liveItems()) |e| e.deinit(self.allocator); self.entries.deinit(self.allocator); }
/// append a frame; copies all slices. wakes blocked readers. pub fn append(self: *Tail, kind: wire.Kind, did: []const u8, collection: []const u8, time_us: i64, seq: u64, json: []const u8, json_v2: []const u8, skip_v1: SkipV1) !void { const entry: Entry = .{ .kind = kind, .did = try self.allocator.dupe(u8, did), .collection = try self.allocator.dupe(u8, collection), .time_us = time_us, .seq = seq, .json = try self.allocator.dupe(u8, json), .json_v2 = try self.allocator.dupe(u8, json_v2), .skip_v1 = skip_v1, }; self.mutex.lockUncancelable(self.io); defer self.mutex.unlock(self.io); try self.entries.append(self.allocator, entry); if (entry.seq != 0) self.max_appended_seq = @max(self.max_appended_seq, entry.seq); const entry_bytes = entryBytes(entry); self.cur_bytes += entry_bytes; if (self.durability_gated and entry.seq != 0 and entry.seq > self.published_seq) self.pinned_bytes += entry_bytes; self.evictLocked(); self.cond.broadcast(self.io); }
fn evictLocked(self: *Tail) void { while (self.cur_bytes > self.max_bytes and self.head < self.entries.items.len) { const oldest = self.entries.items[self.head]; // Rows newer than the archive watermark are the only copy a hot // subscriber can read. Match upstream's ReadableLog: exceed the // retention budget rather than evicting an undurable row. if (self.durability_gated and oldest.seq != 0 and oldest.seq > self.published_seq) break; const evicted = oldest; self.head += 1; self.cur_bytes -= entryBytes(evicted); evicted.deinit(self.allocator); self.base += 1; } self.compactLocked(); }
fn liveItems(self: *const Tail) []Entry { return self.entries.items[self.head..]; }
fn compactLocked(self: *Tail) void { if (self.head == 0) return; const live_len = self.entries.items.len - self.head; if (live_len == 0) { self.entries.clearRetainingCapacity(); self.head = 0; return; } // Retire a large prefix in one move. Requiring both an absolute and a // relative threshold avoids churn for tiny tails while keeping stale // slots bounded for long-running append/evict workloads. if (self.head < 4096 or self.head < live_len) return; std.mem.copyForwards(Entry, self.entries.items[0..live_len], self.entries.items[self.head..]); self.entries.shrinkRetainingCapacity(live_len); self.head = 0; }
fn entryBytes(e: Entry) usize { return e.json.len + e.json_v2.len + e.did.len + e.collection.len + 64; }
pub const Gauges = struct { entries: usize, bytes: usize, pinned_bytes: usize, pinned_overrun_bytes: usize, base: u64, tip: u64, };
pub fn gauges(self: *Tail) Gauges { self.mutex.lockUncancelable(self.io); defer self.mutex.unlock(self.io); return .{ .entries = self.liveItems().len, .bytes = self.cur_bytes, .pinned_bytes = self.pinned_bytes, .pinned_overrun_bytes = self.pinned_bytes -| self.max_bytes, .base = self.base, .tip = self.base + self.liveItems().len, }; }
/// index of the next frame to be appended (start here for live tail). pub fn tip(self: *Tail) u64 { self.mutex.lockUncancelable(self.io); defer self.mutex.unlock(self.io); return self.base + self.liveItems().len; }
/// Next index after the retained log. A new live subscriber starts here, /// matching upstream ReadFrom's at-tip semantics. pub fn publishedTip(self: *Tail) u64 { self.mutex.lockUncancelable(self.io); defer self.mutex.unlock(self.io); return self.base + self.liveItems().len; }
/// Next local archive seq after the resident hot rows. Zero means no /// seq-bearing row is resident yet, matching upstream's unwarmed writer /// sentinel used by cursor resolution. pub fn publishedNextSeq(self: *Tail) u64 { self.mutex.lockUncancelable(self.io); defer self.mutex.unlock(self.io); if (self.max_appended_seq == 0) return 0; return self.max_appended_seq +| 1; }
/// First sequence still resident in the hot log. Cold readers stop /// immediately before this sequence, then resolve it back to a hot /// index. A moving floor may force another cold batch; it must never turn /// into a skipped range. pub fn publishedFloorSeq(self: *Tail) ?u64 { self.mutex.lockUncancelable(self.io); defer self.mutex.unlock(self.io); for (self.liveItems()) |entry| { if (entry.seq != 0) return entry.seq; } return null; }
/// time of the oldest retained entry, or null when empty. pub fn oldestTime(self: *Tail) ?i64 { self.mutex.lockUncancelable(self.io); defer self.mutex.unlock(self.io); const live = self.liveItems(); if (live.len == 0) return null; return live[0].time_us; }
/// smallest index whose entry has time_us >= cursor; tip if none. /// v1 cursor semantics: a below-floor cursor clamps to the oldest /// retained frame, never an error. pub fn indexForCursor(self: *Tail, cursor: i64) u64 { self.mutex.lockUncancelable(self.io); defer self.mutex.unlock(self.io); const live = self.liveItems(); for (live, 0..) |e, i| { if (e.time_us >= cursor) return self.base + i; } return self.base + live.len; }
/// copy the next frame at/after absolute index `idx` that matches `pred`, /// allocated from `allocator`. blocks at the tip until append or close. /// returns null when the tail is closed. pub fn next( self: *Tail, allocator: Allocator, idx: *u64, stop: *const std.atomic.Value(bool), v2: bool, ctx: anytype, comptime pred: fn (@TypeOf(ctx), wire.Kind, []const u8, []const u8, SkipV1) bool, ) !?[]u8 { self.mutex.lockUncancelable(self.io); defer self.mutex.unlock(self.io); while (true) { if (self.closed or stop.load(.acquire)) return null; if (idx.* < self.base) idx.* = self.base; // fell below the floor const rel: usize = @intCast(idx.* - self.base); const live = self.liveItems(); if (rel >= live.len) { self.cond.wait(self.io, &self.mutex) catch return null; continue; } const e = live[rel]; idx.* += 1; if (!v2 and e.skip_v1 != .none) { _ = pred(ctx, e.kind, e.did, e.collection, e.skip_v1); continue; } const json = if (v2) e.json_v2 else e.json; if (json.len == 0) continue; // kind not on this wire (#sync on v1) if (!pred(ctx, e.kind, e.did, e.collection, .none)) continue; return try allocator.dupe(u8, json); } }
/// Pull at most `max` RAW log entries. Filtering and v1 skip rules happen /// after this boundary in the subscriber, matching upstream ReadFrom: /// selective clients cannot turn a bounded scan into an unbounded one. pub fn readBatch( self: *Tail, allocator: Allocator, idx: *u64, stop: *const std.atomic.Value(bool), v2: bool, max: usize, ) !ReadResult { self.mutex.lockUncancelable(self.io); defer self.mutex.unlock(self.io); const limit = @max(max, 1); while (true) { if (self.closed or stop.load(.acquire)) return .closed; if (idx.* < self.base) return .fell_behind; const rel: usize = @intCast(idx.* - self.base); const live = self.liveItems(); if (rel >= live.len) { self.cond.wait(self.io, &self.mutex) catch return .closed; continue; }
const visible = @min(live.len - rel, limit); const result = try allocator.alloc(BatchEntry, visible); var initialized: usize = 0; errdefer { for (result[0..initialized]) |entry| { allocator.free(entry.did); allocator.free(entry.collection); allocator.free(entry.json); } allocator.free(result); } for (result, live[rel..][0..visible]) |*out, entry| { const did = try allocator.dupe(u8, entry.did); errdefer allocator.free(did); const collection = try allocator.dupe(u8, entry.collection); errdefer allocator.free(collection); const json = try allocator.dupe(u8, if (v2) entry.json_v2 else entry.json); out.* = .{ .kind = entry.kind, .did = did, .collection = collection, .seq = entry.seq, .json = json, .skip_v1 = entry.skip_v1 }; initialized += 1; } idx.* += visible; return .{ .batch = .{ .entries = result } }; } }
/// smallest index whose entry has seq >= cursor (v2 seq cursors); /// null when the cursor is below the retained floor. pub fn indexForSeq(self: *Tail, cursor: u64) ?u64 { self.mutex.lockUncancelable(self.io); defer self.mutex.unlock(self.io); const live = self.liveItems(); if (live.len == 0) return null; // A cursor above every seq ever appended cannot be resident: answer // at-tip without scanning, exactly what the full walk below returns // for it. At-tip and future cursors are the steady state of every // caught-up subscriber, so this bound keeps the read loop O(1). if (self.max_appended_seq != 0 and cursor > self.max_appended_seq) return self.base + live.len; // Non-archived diagnostic/control rows (seq=0) do not define the // retained sequence floor, and a tail of only such rows has no floor // at all: the caller must consult the archive instead. (Upstream's // read log is dense — append panics off-by-one — so it resolves // cursors with O(1) arithmetic; carrying seq=0 repair rows is our // divergence, and this lookup accommodates it in O(log n) instead of // the O(n) walk that convoyed ingest appends under this mutex — the // second scan from that family, after max_appended_seq above.) var first_nonzero: usize = 0; while (first_nonzero < live.len and live[first_nonzero].seq == 0) first_nonzero += 1; if (first_nonzero == live.len) return null; if (live[first_nonzero].seq > cursor and cursor > 0) return null; // leftmost nonzero-seq entry with seq >= cursor; zero rows are // transparent (probed past within the current half) var lo: usize = first_nonzero; var hi: usize = live.len; while (lo < hi) { const mid = lo + (hi - lo) / 2; var probe = mid; while (probe < hi and live[probe].seq == 0) probe += 1; if (probe == hi) { // [mid, hi) is all control rows: no candidate there hi = mid; continue; } if (live[probe].seq >= cursor) hi = probe else lo = probe + 1; } // the search can terminate on a control row; the linear scan's answer // is the first nonzero entry at or after it (identical by ordering) while (lo < live.len and live[lo].seq == 0) lo += 1; return self.base + lo; }
/// wake all blocked readers without closing (e.g. so a reader can /// re-check its own liveness flag). pub fn wake(self: *Tail) void { self.mutex.lockUncancelable(self.io); defer self.mutex.unlock(self.io); self.cond.broadcast(self.io); }
/// wake all blocked readers and make them return null. pub fn close(self: *Tail) void { self.mutex.lockUncancelable(self.io); defer self.mutex.unlock(self.io); self.closed = true; self.cond.broadcast(self.io); }};
// === tests ===
const testing = std.testing;
fn acceptAll(_: void, _: wire.Kind, _: []const u8, _: []const u8, _: SkipV1) bool { return true;}
test "append, read, evict, cursor clamp" { var threaded: Io.Threaded = .init(testing.allocator, .{}); defer threaded.deinit(); var t = Tail.init(testing.allocator, threaded.io(), 300); defer t.deinit();
try t.append(.commit, "did:plc:a", "app.bsky.feed.post", 100, 1, "{\"n\":1}", "{\"v2\":1}", .none); try t.append(.commit, "did:plc:a", "app.bsky.feed.post", 200, 2, "{\"n\":2}", "{\"v2\":2}", .none);
var idx: u64 = 0; var stop: std.atomic.Value(bool) = .init(false); const j1 = (try t.next(testing.allocator, &idx, &stop, false, {}, acceptAll)).?; defer testing.allocator.free(j1); try testing.expectEqualStrings("{\"n\":1}", j1);
// small budget: appending more evicts the oldest try t.append(.commit, "did:plc:a", "app.bsky.feed.post", 300, 3, "{\"n\":3}", "{\"v2\":3}", .none); try t.append(.commit, "did:plc:a", "app.bsky.feed.post", 400, 4, "{\"n\":4}", "{\"v2\":4}", .none); try testing.expect(t.base > 0);
try testing.expectEqual(t.base, t.indexForCursor(0)); try testing.expectEqual(t.tip(), t.indexForCursor(999_999));}
test "future and at-tip cursors resolve O(1): the appended-seq bound answers without a scan" { // 2026-08-09 ingest convoy regression: every caught-up seq-cursor // subscriber walked the whole resident deque (and publishedNextSeq walked // it again via tooSlow) under the tail mutex on every read-loop pass. // The monotone max-appended bound must answer both queries without a // scan, and must survive eviction so the answers stay stable. var threaded: Io.Threaded = .init(testing.allocator, .{}); defer threaded.deinit(); var t = Tail.init(testing.allocator, threaded.io(), 300); defer t.deinit();
try t.append(.commit, "did:plc:a", "app.bsky.feed.post", 100, 10, "{\"n\":1}", "{\"v2\":1}", .none); try t.append(.commit, "did:plc:a", "app.bsky.feed.post", 200, 11, "{\"n\":2}", "{\"v2\":2}", .none);
// at-tip (next expected) and far-future (e.g. a relay-space seq handed to // a local-seq cursor) both resolve to the tip, matching the full walk try testing.expectEqual(@as(?u64, t.tip()), t.indexForSeq(12)); try testing.expectEqual(@as(?u64, t.tip()), t.indexForSeq(32_581_690_274)); try testing.expectEqual(@as(u64, 12), t.publishedNextSeq());
// evicting the seq-bearing rows must not reset the bound: the next-seq // anchor is "highest ever appended", not "highest still resident" try t.append(.commit, "did:plc:a", "app.bsky.feed.post", 300, 12, "{\"n\":3}", "{\"v2\":3}", .none); try t.append(.commit, "did:plc:a", "app.bsky.feed.post", 400, 13, "{\"n\":4}", "{\"v2\":4}", .none); try testing.expect(t.base > 0); // small budget forced eviction try testing.expectEqual(@as(u64, 14), t.publishedNextSeq()); try testing.expectEqual(@as(?u64, t.tip()), t.indexForSeq(14));
// a resident cursor still resolves exactly (no early-out short circuit) try testing.expectEqual(@as(?u64, t.tip() - 1), t.indexForSeq(13));}
test "read batch bounds raw scanned entries before subscriber filtering" { var threaded: Io.Threaded = .init(testing.allocator, .{}); defer threaded.deinit(); var t = Tail.init(testing.allocator, threaded.io(), 1 << 20); defer t.deinit(); try t.append(.sync, "did:plc:a", "", 100, 1, "", "{\"seq\":1}", .sync); try t.append(.commit, "did:plc:a", "app.bsky.feed.post", 200, 2, "{\"n\":2}", "{\"seq\":2}", .none); try t.append(.commit, "did:plc:b", "app.bsky.feed.like", 300, 3, "{\"n\":3}", "{\"seq\":3}", .none);
var idx: u64 = 0; var stop: std.atomic.Value(bool) = .init(false); var first_arena = std.heap.ArenaAllocator.init(testing.allocator); defer first_arena.deinit(); const first = try t.readBatch(first_arena.allocator(), &idx, &stop, false, 2); try testing.expectEqual(@as(usize, 2), first.batch.entries.len); try testing.expectEqual(@as(u64, 2), idx); try testing.expectEqual(SkipV1.sync, first.batch.entries[0].skip_v1); try testing.expectEqual(@as(usize, 0), first.batch.entries[0].json.len);
var second_arena = std.heap.ArenaAllocator.init(testing.allocator); defer second_arena.deinit(); const second = try t.readBatch(second_arena.allocator(), &idx, &stop, false, 2); try testing.expectEqual(@as(usize, 1), second.batch.entries.len); try testing.expectEqual(@as(u64, 3), idx); try testing.expectEqualStrings("{\"n\":3}", second.batch.entries[0].json);}
test "large eviction burst retires a prefix without per-entry shifts" { var threaded: Io.Threaded = .init(testing.allocator, .{}); defer threaded.deinit(); var t = Tail.init(testing.allocator, threaded.io(), 64 * 1024 * 1024); defer t.deinit();
for (0..20_000) |i| { try t.append(.commit, "did:plc:burst", "app.bsky.feed.post", @intCast(i), i + 1, "{}", "{}", .none); } try testing.expectEqual(@as(usize, 20_000), t.gauges().entries);
// A large repair can push a previously full tail over a newly relevant // byte floor in one writer turn. This used to orderedRemove(0) twenty // thousand times, shifting the retained array on every row. t.max_bytes = 1024; try t.append(.commit, "did:plc:burst", "app.bsky.feed.post", 20_001, 20_001, "{}", "{}", .none); const gauges = t.gauges(); try testing.expect(gauges.bytes <= t.max_bytes); try testing.expect(gauges.base > 19_000); try testing.expect(gauges.entries < 20); try testing.expect(t.entries.items.len < 20);}
test "appended frames are readable before durability publication" { var threaded: Io.Threaded = .init(testing.allocator, .{}); defer threaded.deinit(); var t = Tail.init(testing.allocator, threaded.io(), 1 << 20); defer t.deinit(); t.enableDurabilityGate(0);
// Upstream ReadFrom serves to the appended tip; durability only pins // rows against eviction. A live reader must see the row immediately. try t.append(.commit, "did:plc:a", "app.bsky.feed.post", 100, 1, "{\"n\":1}", "{\"v2\":1}", .none); try testing.expectEqual(@as(u64, 1), t.tip()); try testing.expectEqual(@as(u64, 1), t.publishedTip()); try testing.expectEqual(@as(?u64, 0), t.indexForSeq(1));
var idx: u64 = 0; var stop: std.atomic.Value(bool) = .init(false); const json = (try t.next(testing.allocator, &idx, &stop, false, {}, acceptAll)).?; defer testing.allocator.free(json); try testing.expectEqualStrings("{\"n\":1}", json); try testing.expectEqual(t.gauges().bytes, t.gauges().pinned_bytes);
t.publishDurable(1); try testing.expectEqual(@as(usize, 0), t.gauges().pinned_bytes);}
test "seq cursor below the hot floor ignores leading unarchived rows" { var threaded: Io.Threaded = .init(testing.allocator, .{}); defer threaded.deinit(); var t = Tail.init(testing.allocator, threaded.io(), 1 << 20); defer t.deinit();
try t.append(.identity, "did:plc:a", "", 100, 0, "{}", "{}", .none); try t.append(.commit, "did:plc:a", "app.bsky.feed.post", 200, 42, "{}", "{}", .none);
try testing.expectEqual(@as(?u64, null), t.indexForSeq(1)); try testing.expectEqual(@as(?u64, 1), t.indexForSeq(42));}
test "indexForSeq binary search matches the linear scan across control-row layouts" { var threaded: Io.Threaded = .init(testing.allocator, .{}); defer threaded.deinit(); var t = Tail.init(testing.allocator, threaded.io(), 1 << 24); defer t.deinit();
// interleave archived rows (seqs advancing with gaps) and seq=0 control // rows in runs: leading, interior singles, an interior run, and trailing var seq: u64 = 10; var prng = std.Random.DefaultPrng.init(7); const random = prng.random(); try t.append(.identity, "did:plc:z", "", 1, 0, "{}", "{}", .none); try t.append(.identity, "did:plc:z", "", 1, 0, "{}", "{}", .none); for (0..500) |_| { seq += 1 + random.uintLessThan(u64, 5); try t.append(.commit, "did:plc:a", "app.bsky.feed.post", 1, seq, "{}", "{}", .none); if (random.uintLessThan(u8, 4) == 0) { for (0..1 + random.uintLessThan(usize, 3)) |_| try t.append(.identity, "did:plc:z", "", 1, 0, "{}", "{}", .none); } } try t.append(.identity, "did:plc:z", "", 1, 0, "{}", "{}", .none);
// linear reference: the pre-binary-search behavior, verbatim const Ref = struct { fn indexForSeq(tl: *Tail, cursor: u64) ?u64 { const live = tl.entries.items[tl.head..]; if (live.len == 0) return null; if (tl.max_appended_seq != 0 and cursor > tl.max_appended_seq) return tl.base + live.len; var saw_seq = false; for (live, 0..) |e, i| { if (e.seq == 0) continue; if (!saw_seq) { saw_seq = true; if (e.seq > cursor and cursor > 0) return null; } if (e.seq >= cursor) return tl.base + i; } if (!saw_seq) return null; return tl.base + live.len; } };
// sweep: below floor, every present seq, every gap neighbor, at/past tip var cursor: u64 = 0; while (cursor <= seq + 3) : (cursor += 1) { try testing.expectEqual(Ref.indexForSeq(&t, cursor), t.indexForSeq(cursor)); }}
test "undurable rows pin readable log beyond budget until publication" { var threaded: Io.Threaded = .init(testing.allocator, .{}); defer threaded.deinit(); var t = Tail.init(testing.allocator, threaded.io(), 1); defer t.deinit(); t.enableDurabilityGate(0);
try t.append(.commit, "did:plc:a", "app.bsky.feed.post", 100, 1, "{\"n\":1}", "{\"v2\":1}", .none); try t.append(.commit, "did:plc:a", "app.bsky.feed.post", 200, 2, "{\"n\":2}", "{\"v2\":2}", .none); const pinned = t.gauges(); try testing.expectEqual(@as(usize, 2), pinned.entries); try testing.expectEqual(pinned.bytes, pinned.pinned_bytes); try testing.expectEqual(pinned.pinned_bytes - 1, pinned.pinned_overrun_bytes); try testing.expectEqual(@as(u64, 0), pinned.base);
t.publishDurable(1); const one_pinned = t.gauges(); try testing.expectEqual(@as(usize, 1), one_pinned.entries); try testing.expectEqual(@as(u64, 1), one_pinned.base); try testing.expectEqual(one_pinned.bytes, one_pinned.pinned_bytes);
t.publishDurable(2); const drained = t.gauges(); try testing.expectEqual(@as(usize, 0), drained.entries); try testing.expectEqual(@as(usize, 0), drained.bytes); try testing.expectEqual(@as(usize, 0), drained.pinned_bytes); try testing.expectEqual(@as(usize, 0), drained.pinned_overrun_bytes); try testing.expectEqual(@as(u64, 2), drained.base); try testing.expectEqual(@as(u64, 2), drained.tip);}
test "v1 observes skipped rows while v2 receives them" { var threaded: Io.Threaded = .init(testing.allocator, .{}); defer threaded.deinit(); var t = Tail.init(testing.allocator, threaded.io(), 1 << 20); defer t.deinit(); try t.append(.sync, "did:plc:a", "", 100, 1, "", "{\"kind\":\"sync\"}", .sync); try t.append(.commit, "did:plc:a", "app.bsky.feed.post", 200, 2, "{\"kind\":\"commit\"}", "{\"kind\":\"commit\",\"seq\":2}", .none);
const Observer = struct { skipped: usize = 0, fn wants(self: *@This(), _: wire.Kind, _: []const u8, _: []const u8, skip: SkipV1) bool { if (skip != .none) self.skipped += 1; return true; } }; var stop: std.atomic.Value(bool) = .init(false); var v1_observer: Observer = .{}; var v1_idx: u64 = 0; const v1 = (try t.next(testing.allocator, &v1_idx, &stop, false, &v1_observer, Observer.wants)).?; defer testing.allocator.free(v1); try testing.expectEqual(@as(usize, 1), v1_observer.skipped); try testing.expectEqualStrings("{\"kind\":\"commit\"}", v1);
var v2_observer: Observer = .{}; var v2_idx: u64 = 0; const v2 = (try t.next(testing.allocator, &v2_idx, &stop, true, &v2_observer, Observer.wants)).?; defer testing.allocator.free(v2); try testing.expectEqual(@as(usize, 0), v2_observer.skipped); try testing.expectEqualStrings("{\"kind\":\"sync\"}", v2);}
test "close unblocks readers" { var threaded: Io.Threaded = .init(testing.allocator, .{}); defer threaded.deinit(); var t = Tail.init(testing.allocator, threaded.io(), 1 << 20); defer t.deinit(); const Reader = struct { fn run(tl: *Tail) void { var idx: u64 = 0; var stop: std.atomic.Value(bool) = .init(false); const got = tl.next(testing.allocator, &idx, &stop, false, {}, acceptAll) catch unreachable; std.debug.assert(got == null); } }; const th = try std.Thread.spawn(.{}, Reader.run, .{&t}); t.close(); th.join();}