Something went wrong. Try again.
jetstream v2 in zig stream.waow.tech
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341//! tombstones — which rows a compaction pass may physically drop//!//! upstream internal/tombstone (docs/upstream-compaction-spec.md): two//! tombstone families, both keyed by seq with STRICT inequality://! - record: (did, collection, rkey) -> max superseding seq, produced by//! KindDelete and KindUpdate rows (update = delete + replace)//! - did: did -> seq, produced by KindSync rows (repo diverged) and//! KindAccount rows whose payload says active=false && status=="deleted"//!//! suppression drops ONLY materialization rows (create/update/create_resync)//! with seq strictly below the tombstone seq; marker rows (delete/account///! identity/sync) survive forever — clients fold the stream themselves,//! there is no read-time overlay. tombstones are never persisted separately://! they ARE the event rows, and snapshots are rebuilt by folding decoded//! rows. keys are cloned on insert — decoded events alias the decompressed//! block buffer (same hazard as upstream).
const std = @import("std");const zat = @import("zat");const segment = @import("segment.zig");
const Allocator = std.mem.Allocator;
pub const DidReason = enum { sync, account };pub const DropReason = enum { record, sync, account };
pub const DidTombstone = struct { seq: u64, reason: DidReason,};
pub const Counts = struct { record: u64 = 0, sync: u64 = 0, account: u64 = 0,};
/// a fold of tombstone rows over some seq window. also the live in-memory/// set (fed per append in steady state) — same structure, the live set is/// used only for the cap trigger, never for drop decisions.pub const Snapshot = struct { allocator: Allocator, /// both maps own their individually allocated keys so eviction releases /// real resident memory rather than retaining an arena high-water mark. records: std.StringHashMapUnmanaged(u64) = .empty, dids: std.StringHashMapUnmanaged(DidTombstone) = .empty, /// running total matching approxBytes' historical formula, maintained at /// insert/evict so approxBytes is O(1). The live set publishes it per /// archive append (Compactor.onAppend); the previous full-map walk cost /// ~80% of the writer core at 119k entries (perf, 2026-08-06). approx_bytes: usize = 0,
pub fn init(allocator: Allocator) Snapshot { return .{ .allocator = allocator }; }
pub fn deinit(self: *Snapshot) void { var records = self.records.keyIterator(); while (records.next()) |key| self.allocator.free(key.*); var dids = self.dids.keyIterator(); while (dids.next()) |key| self.allocator.free(key.*); self.records.deinit(self.allocator); self.dids.deinit(self.allocator); }
pub fn len(self: *const Snapshot) usize { return self.records.count() + self.dids.count(); }
/// observe one event row; per-key max seq wins. non-marker kinds are /// ignored, so callers can feed every row unconditionally. pub fn observe(self: *Snapshot, ev: segment.Event) !void { switch (ev.kind) { .delete, .update => { var key_buf: [1024]u8 = undefined; const key = recordKey(&key_buf, ev.did, ev.collection, ev.rkey) orelse return; const gop = try self.records.getOrPut(self.allocator, key); if (!gop.found_existing) { const owned = self.allocator.dupe(u8, key) catch |err| { _ = self.records.remove(key); return err; }; gop.key_ptr.* = owned; gop.value_ptr.* = ev.seq; self.approx_bytes += owned.len - 2 + 64; } else if (ev.seq > gop.value_ptr.*) { gop.value_ptr.* = ev.seq; } }, .sync => try self.observeDid(ev.did, ev.seq, .sync), .account => { if (try accountIsDelete(self.allocator, ev.payload)) try self.observeDid(ev.did, ev.seq, .account); }, .create, .create_resync, .identity => {}, } }
fn observeDid(self: *Snapshot, did: []const u8, seq: u64, reason: DidReason) !void { const gop = try self.dids.getOrPut(self.allocator, did); if (!gop.found_existing) { const owned = self.allocator.dupe(u8, did) catch |err| { _ = self.dids.remove(did); return err; }; gop.key_ptr.* = owned; gop.value_ptr.* = .{ .seq = seq, .reason = reason }; self.approx_bytes += owned.len + 64; } else if (seq > gop.value_ptr.seq) { gop.value_ptr.* = .{ .seq = seq, .reason = reason }; } }
/// true iff a compaction pass may physically drop this row: a /// materialization row strictly below a covering tombstone. marker rows /// are never dropped. pub fn shouldDrop(self: *const Snapshot, ev: segment.Event) bool { return self.dropReason(ev) != null; }
/// Upstream attributes each physically removed row to the covering /// tombstone. DID-wide tombstones win over record tombstones. pub fn dropReason(self: *const Snapshot, ev: segment.Event) ?DropReason { switch (ev.kind) { .create, .update, .create_resync => {}, else => return null, } if (self.dids.get(ev.did)) |t| { if (t.seq > ev.seq) return switch (t.reason) { .sync => .sync, .account => .account, }; } var key_buf: [1024]u8 = undefined; const key = recordKey(&key_buf, ev.did, ev.collection, ev.rkey) orelse return null; if (self.records.get(key)) |t_seq| { if (t_seq > ev.seq) return .record; } return null; }
/// Exact counterpart of upstream Snapshot's metric accounting. Counts /// are derived from the final map so a later account tombstone replacing /// a sync tombstone is attributed to account, not its insertion history. pub fn countsByReason(self: *const Snapshot) Counts { var result: Counts = .{ .record = self.records.count() }; var it = self.dids.iterator(); while (it.next()) |entry| switch (entry.value_ptr.reason) { .sync => result.sync += 1, .account => result.account += 1, }; return result; }
/// Key bytes plus upstream's conservative 64-byte map-entry overhead. pub fn approxBytes(self: *const Snapshot) usize { return self.approx_bytes; }
/// Distinct DIDs represented by this snapshot. Returns null once the /// upstream narrowing cutoff is exceeded, signaling a full scan. pub fn candidateDids(self: *const Snapshot, allocator: Allocator, max: usize) !?[][]const u8 { var set: std.StringHashMapUnmanaged(void) = .empty; defer set.deinit(allocator); var records = self.records.iterator(); while (records.next()) |entry| { const nul = std.mem.indexOfScalar(u8, entry.key_ptr.*, 0).?; try set.put(allocator, entry.key_ptr.*[0..nul], {}); if (set.count() > max) return null; } var dids = self.dids.iterator(); while (dids.next()) |entry| { try set.put(allocator, entry.key_ptr.*, {}); if (set.count() > max) return null; } const result = try allocator.alloc([]const u8, set.count()); var index: usize = 0; var it = set.keyIterator(); while (it.next()) |did| : (index += 1) result[index] = did.*; return result; }
/// drop entries with seq <= watermark (they cover only rows already /// physically compacted). live-set maintenance; snapshots are throwaway. pub fn evict(self: *Snapshot, watermark: u64) void { var rit = self.records.iterator(); while (rit.next()) |e| { if (e.value_ptr.* <= watermark) { const key = e.key_ptr.*; self.records.removeByPtr(e.key_ptr); self.approx_bytes -= key.len - 2 + 64; self.allocator.free(key); } } var dit = self.dids.iterator(); while (dit.next()) |e| { if (e.value_ptr.seq <= watermark) { const key = e.key_ptr.*; self.dids.removeByPtr(e.key_ptr); self.approx_bytes -= key.len + 64; self.allocator.free(key); } } }};
/// (did, collection, rkey) -> "did\x00collection\x00rkey". components never/// contain NUL (did/nsid/rkey syntax); null on overflow (caller drops the/// observation — such a row also fails segment validation upstream of here).fn recordKey(buf: []u8, did: []const u8, collection: []const u8, rkey: []const u8) ?[]const u8 { return std.fmt.bufPrint(buf, "{s}\x00{s}\x00{s}", .{ did, collection, rkey }) catch null;}
/// does an #account payload mark the account deleted? upstream decodes the/// envelope per fold; matches active == false && status == "deleted".pub fn accountIsDelete(allocator: Allocator, payload: []const u8) !bool { var arena = std.heap.ArenaAllocator.init(allocator); defer arena.deinit(); const env = try zat.cbor.decodeAll(arena.allocator(), payload); const active = env.getBool("active") orelse return false; if (active) return false; const status = env.getString("status") orelse return false; return std.mem.eql(u8, status, "deleted");}
// === tests ===
const testing = std.testing;
fn row(kind: segment.Kind, seq: u64, did: []const u8, coll: []const u8, rkey: []const u8, payload: []const u8) segment.Event { return .{ .seq = seq, .witnessed_at = 1, .indexed_at = 0, .kind = kind, .did = did, .collection = coll, .rkey = rkey, .rev = "3l3qo2vutsw2b", .payload = payload, };}
// {"active": false, "status": "deleted"} / {"active": true} as dag-cborconst account_deleted = "\xa2\x66\x61\x63\x74\x69\x76\x65\xf4\x66\x73\x74\x61\x74\x75\x73\x67\x64\x65\x6c\x65\x74\x65\x64";const account_active = "\xa1\x66\x61\x63\x74\x69\x76\x65\xf5";
test "record tombstones: strict inequality, max-seq wins, markers survive" { var snap = Snapshot.init(testing.allocator); defer snap.deinit();
try snap.observe(row(.delete, 10, "did:plc:a", "app.bsky.feed.post", "k1", "")); try snap.observe(row(.update, 7, "did:plc:a", "app.bsky.feed.post", "k1", "x")); // lower, ignored try testing.expectEqual(@as(usize, 1), snap.len());
// strictly below 10 → drop; at 10 (the marker's own seq) or above → keep try testing.expect(snap.shouldDrop(row(.create, 9, "did:plc:a", "app.bsky.feed.post", "k1", "x"))); try testing.expect(snap.shouldDrop(row(.update, 9, "did:plc:a", "app.bsky.feed.post", "k1", "x"))); try testing.expect(snap.shouldDrop(row(.create_resync, 9, "did:plc:a", "app.bsky.feed.post", "k1", "x"))); try testing.expect(!snap.shouldDrop(row(.create, 10, "did:plc:a", "app.bsky.feed.post", "k1", "x"))); try testing.expect(!snap.shouldDrop(row(.create, 11, "did:plc:a", "app.bsky.feed.post", "k1", "x"))); // different rkey untouched; marker kinds never drop try testing.expect(!snap.shouldDrop(row(.create, 9, "did:plc:a", "app.bsky.feed.post", "k2", "x"))); try testing.expect(!snap.shouldDrop(row(.delete, 9, "did:plc:a", "app.bsky.feed.post", "k1", ""))); try testing.expect(!snap.shouldDrop(row(.identity, 9, "did:plc:a", "", "", "x")));}
test "did tombstones: sync and account-delete cover the whole did" { var snap = Snapshot.init(testing.allocator); defer snap.deinit();
try snap.observe(row(.sync, 20, "did:plc:b", "", "", "")); try snap.observe(row(.account, 30, "did:plc:c", "", "", account_deleted)); try snap.observe(row(.account, 40, "did:plc:d", "", "", account_active)); // not a delete const counts = snap.countsByReason(); try testing.expectEqual(@as(u64, 1), counts.sync); try testing.expectEqual(@as(u64, 1), counts.account);
try testing.expect(snap.shouldDrop(row(.create, 19, "did:plc:b", "app.bsky.feed.post", "k", "x"))); try testing.expect(!snap.shouldDrop(row(.create, 20, "did:plc:b", "app.bsky.feed.post", "k", "x"))); try testing.expect(snap.shouldDrop(row(.create, 29, "did:plc:c", "app.bsky.feed.post", "k", "x"))); try testing.expect(!snap.shouldDrop(row(.create, 39, "did:plc:d", "app.bsky.feed.post", "k", "x")));}
test "evict removes entries at or below the watermark" { var snap = Snapshot.init(testing.allocator); defer snap.deinit(); try snap.observe(row(.delete, 5, "did:plc:a", "app.bsky.feed.post", "k1", "")); try snap.observe(row(.delete, 15, "did:plc:a", "app.bsky.feed.post", "k2", "")); try snap.observe(row(.sync, 10, "did:plc:b", "", "", "")); snap.evict(10); try testing.expectEqual(@as(usize, 1), snap.len()); try testing.expect(snap.shouldDrop(row(.create, 14, "did:plc:a", "app.bsky.feed.post", "k2", "x"))); try testing.expect(!snap.shouldDrop(row(.create, 4, "did:plc:a", "app.bsky.feed.post", "k1", "x")));}
test "account tombstones reject malformed CBOR and metrics reflect final reason" { var snap = Snapshot.init(testing.allocator); defer snap.deinit(); try testing.expectError(error.UnexpectedEof, snap.observe(row(.account, 1, "did:plc:a", "", "", "\xa1"))); try snap.observe(row(.sync, 2, "did:plc:a", "", "", "")); try snap.observe(row(.account, 3, "did:plc:a", "", "", account_deleted)); const counts = snap.countsByReason(); try testing.expectEqual(@as(u64, 0), counts.sync); try testing.expectEqual(@as(u64, 1), counts.account); try testing.expectEqual(@as(usize, "did:plc:a".len + 64), snap.approxBytes());}
test "approxBytes counter matches a full recomputation across observe and evict" { var snap = Snapshot.init(testing.allocator); defer snap.deinit();
const recompute = struct { fn walk(s: *const Snapshot) usize { var result: usize = 0; var rit = s.records.iterator(); while (rit.next()) |entry| result += entry.key_ptr.len - 2 + 64; var dit = s.dids.iterator(); while (dit.next()) |entry| result += entry.key_ptr.len + 64; return result; } }.walk;
try snap.observe(row(.delete, 10, "did:plc:aaa", "app.bsky.feed.post", "rkey1", "")); try snap.observe(row(.update, 12, "did:plc:aaa", "app.bsky.feed.post", "rkey2", "")); // duplicate key, higher seq: value updates, bytes must not change try snap.observe(row(.delete, 15, "did:plc:aaa", "app.bsky.feed.post", "rkey1", "")); try snap.observe(row(.sync, 20, "did:plc:bbb", "", "", "")); try snap.observe(row(.account, 25, "did:plc:ccc", "", "", account_deleted)); // non-marker kinds are ignored try snap.observe(row(.create, 30, "did:plc:ddd", "app.bsky.feed.like", "rkey3", "")); try testing.expectEqual(recompute(&snap), snap.approxBytes());
snap.evict(14); // removes rkey2 (12), keeps rkey1 (15), keeps dids (20, 25) try testing.expectEqual(recompute(&snap), snap.approxBytes());
snap.evict(std.math.maxInt(u64)); // removes everything try testing.expectEqual(@as(usize, 0), snap.approxBytes()); try testing.expectEqual(recompute(&snap), snap.approxBytes());}