Something went wrong. Try again.
jetstream v2 in zig stream.waow.tech
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610//! jss segment format v1 — writer, part 1: block encode + active file//!//! encodes events into the columnar block layout, compresses each block//! into a zstd frame (content checksum on, matching upstream's encoder),//! and appends length-prefixed frames after a 256-byte active header//! (magic + zeros; checksum 0 marks the file unsealed).//!//! sealing (docs/jss-seal-spec.md): walks the file's own frames like//! upstream's walkActiveFrames, builds the footer (block index, gloom DID//! blooms, collection index with first-seen ids + $-sentinels), and//! finalizes the header with the file xxh3. per-block zstd frame CRCs are//! written and verified through the vendored libzstd implementation.
const std = @import("std");const footer_mod = @import("segment_footer.zig");const gloom = @import("gloom.zig");const segment = @import("segment.zig");const zstd = @import("zstd.zig");
const Allocator = std.mem.Allocator;
pub const default_max_events_per_block = 4096;
pub const Error = error{ OutOfMemory, CompressFailed, FieldTooLong, InvalidEvent,};
/// Uncompressed, self-contained block detached from an active writer. The/// pending-event arena can be reused immediately while compression happens on/// another worker.pub const PreparedBlock = struct { bytes: []u8, max_seq: u64,
pub fn deinit(self: *PreparedBlock, allocator: Allocator) void { allocator.free(self.bytes); self.* = undefined; }};
/// validate upstream's column-width limits; mirrors segment.ValidateEvent./// spec-legal-but-unrepresentable fields (e.g. 256-byte rkeys) must be/// dropped by the caller with a distinct metric, never truncated.pub fn validateEvent(ev: segment.Event) Error!void { if (ev.seq == 0) return error.InvalidEvent; // 0 is the "before beginning" sentinel try validateFields(ev);}
/// column-width checks only — for rows validated before a seq is assigned/// (the backfill collector; the archive assigns seqs at append)pub fn validateFields(ev: segment.Event) Error!void { if (ev.did.len > std.math.maxInt(u16)) return error.FieldTooLong; if (ev.collection.len > std.math.maxInt(u8)) return error.FieldTooLong; if (ev.rkey.len > std.math.maxInt(u8)) return error.FieldTooLong; if (ev.rev.len > std.math.maxInt(u8)) return error.FieldTooLong; if (ev.payload.len > std.math.maxInt(u32)) return error.FieldTooLong;}
/// encode events into one uncompressed columnar block.pub fn encodeBlock(allocator: Allocator, events: []const segment.Event) Error![]u8 { if (events.len == 0) { // empty compaction block: exactly the 4 count bytes const out = try allocator.alloc(u8, 4); @memset(out, 0); return out; }
var blob_len: usize = 0; for (events) |ev| { try validateEvent(ev); blob_len += ev.collection.len + ev.did.len + ev.rkey.len + ev.rev.len + ev.payload.len; } const n = events.len; const total = 4 + n * (8 + 8 + 8 + 1 + 1 + 2 + 1 + 1 + 4) + blob_len; const out = try allocator.alloc(u8, total); errdefer allocator.free(out);
std.mem.writeInt(u32, out[0..4], @intCast(n), .little); var off: usize = 4; for (events, 0..) |ev, i| std.mem.writeInt(u64, out[off + i * 8 ..][0..8], ev.seq, .little); off += n * 8; for (events, 0..) |ev, i| std.mem.writeInt(i64, out[off + i * 8 ..][0..8], ev.witnessed_at, .little); off += n * 8; for (events, 0..) |ev, i| std.mem.writeInt(i64, out[off + i * 8 ..][0..8], ev.indexed_at, .little); off += n * 8; for (events, 0..) |ev, i| out[off + i] = @backingInt(ev.kind); off += n; for (events, 0..) |ev, i| out[off + i] = @intCast(ev.collection.len); off += n; for (events, 0..) |ev, i| std.mem.writeInt(u16, out[off + i * 2 ..][0..2], @intCast(ev.did.len), .little); off += n * 2; for (events, 0..) |ev, i| out[off + i] = @intCast(ev.rkey.len); off += n; for (events, 0..) |ev, i| out[off + i] = @intCast(ev.rev.len); off += n; for (events, 0..) |ev, i| std.mem.writeInt(u32, out[off + i * 4 ..][0..4], @intCast(ev.payload.len), .little); off += n * 4;
inline for (.{ "collection", "did", "rkey", "rev", "payload" }) |field| { for (events) |ev| { const v = @field(ev, field); @memcpy(out[off..][0..v.len], v); off += v.len; } } std.debug.assert(off == total); return out;}
/// an in-memory active segment file under construction. pure format code:/// the caller owns durability (writing bytes to disk, fsync policy).pub const ActiveWriter = struct { allocator: Allocator, buf: std.ArrayList(u8) = .empty, pending: std.ArrayList(segment.Event) = .empty, /// borrowed strings in `pending` point into these owned copies pending_arena: std.heap.ArenaAllocator, max_events_per_block: usize = default_max_events_per_block, event_count: u64 = 0, block_count: u32 = 0, /// Footer state accumulated as blocks flush, so seal() is O(footer) /// instead of decompressing and re-walking the whole segment while the /// archive mutex starves ingest (and with it every live subscriber). /// Heap-allocated so the arena's inner pointer survives the writer /// being moved by value (recover() constructs then assigns). index_arena: *std.heap.ArenaAllocator, builder: footer_mod.Builder, env: footer_mod.Envelope = .{}, /// blocks represented in `builder`; seal() takes the incremental path /// only when this matches block_count, else it falls back to the walk indexed_blocks: u32 = 0,
pub fn init(allocator: Allocator) Error!ActiveWriter { const index_arena = try allocator.create(std.heap.ArenaAllocator); errdefer allocator.destroy(index_arena); index_arena.* = std.heap.ArenaAllocator.init(allocator); errdefer index_arena.deinit(); var w: ActiveWriter = .{ .allocator = allocator, .pending_arena = std.heap.ArenaAllocator.init(allocator), .index_arena = index_arena, .builder = footer_mod.Builder.init(index_arena.allocator()), }; try w.buf.appendSlice(allocator, segment.magic); try w.buf.appendNTimes(allocator, 0, segment.header_size - segment.magic.len); return w; }
pub fn deinit(self: *ActiveWriter) void { self.buf.deinit(self.allocator); self.pending.deinit(self.allocator); self.pending_arena.deinit(); self.index_arena.deinit(); self.allocator.destroy(self.index_arena); }
/// stage one event; flushes a block when the threshold is reached. pub fn append(self: *ActiveWriter, ev: segment.Event) Error!void { try self.stage(ev); if (self.pending.items.len >= self.max_events_per_block) try self.flush(); }
/// Stage one event and detach a complete block without compressing it. /// Used by the bootstrap-only async compression pipeline. pub fn appendPrepared(self: *ActiveWriter, ev: segment.Event) Error!?PreparedBlock { try self.stage(ev); if (self.pending.items.len < self.max_events_per_block) return null; return try self.prepareFlush(); }
fn stage(self: *ActiveWriter, ev: segment.Event) Error!void { try validateEvent(ev); const alloc = self.pending_arena.allocator(); try self.pending.append(self.allocator, .{ .seq = ev.seq, .witnessed_at = ev.witnessed_at, .indexed_at = ev.indexed_at, .kind = ev.kind, .did = try alloc.dupe(u8, ev.did), .collection = try alloc.dupe(u8, ev.collection), .rkey = try alloc.dupe(u8, ev.rkey), .rev = try alloc.dupe(u8, ev.rev), .payload = try alloc.dupe(u8, ev.payload), }); self.event_count += 1; }
/// Encode and detach staged rows. No borrowed event bytes escape: the /// returned columnar block owns a complete copy. pub fn prepareFlush(self: *ActiveWriter) Error!?PreparedBlock { if (self.pending.items.len == 0) return null; const max_seq = self.pending.items[self.pending.items.len - 1].seq; const block = try encodeBlock(self.allocator, self.pending.items); try self.noteBlock(.{ .offset = 0, // patched with the file position in commitPrepared .compressed_size = 0, // ditto, once the frame size is known .uncompressed_size = @intCast(block.len), .event_count = 0, .min_seq = 0, .max_seq = 0, .min_witnessed_at = 0, .max_witnessed_at = 0, }, self.pending.items); self.pending.clearRetainingCapacity(); _ = self.pending_arena.reset(.retain_capacity); return .{ .bytes = block, .max_seq = max_seq }; }
/// Record one block into the incremental footer state. `info` needs only /// sizes/offset populated; envelope fields are derived from `events`. fn noteBlock(self: *ActiveWriter, info_in: segment.BlockIndexEntry, events: []const segment.Event) Error!void { var info = info_in; info.event_count = @intCast(events.len); info.min_seq = events[0].seq; info.max_seq = events[0].seq; info.min_witnessed_at = events[0].witnessed_at; info.max_witnessed_at = events[0].witnessed_at; for (events) |ev| { info.min_seq = @min(info.min_seq, ev.seq); info.max_seq = @max(info.max_seq, ev.seq); info.min_witnessed_at = @min(info.min_witnessed_at, ev.witnessed_at); info.max_witnessed_at = @max(info.max_witnessed_at, ev.witnessed_at); } if (self.env.event_count == 0) { self.env.min_seq = info.min_seq; self.env.max_seq = info.max_seq; self.env.min_witnessed_at = info.min_witnessed_at; self.env.max_witnessed_at = info.max_witnessed_at; } else { self.env.min_seq = @min(self.env.min_seq, info.min_seq); self.env.max_seq = @max(self.env.max_seq, info.max_seq); self.env.min_witnessed_at = @min(self.env.min_witnessed_at, info.min_witnessed_at); self.env.max_witnessed_at = @max(self.env.max_witnessed_at, info.max_witnessed_at); } self.env.event_count += events.len; try self.builder.addBlock(info, events); self.indexed_blocks += 1; }
/// Commit a previously compressed block in preparation order. pub fn commitPrepared(self: *ActiveWriter, frame: []const u8) Error!void { if (self.block_count < self.builder.infos.items.len) { const info = &self.builder.infos.items[self.block_count]; info.offset = self.buf.items.len; info.compressed_size = @intCast(frame.len); } var len_prefix: [8]u8 = undefined; std.mem.writeInt(u64, &len_prefix, frame.len, .little); try self.buf.appendSlice(self.allocator, &len_prefix); try self.buf.appendSlice(self.allocator, frame);
self.block_count += 1; }
/// compress staged events into a frame and append it to the file image. pub fn flush(self: *ActiveWriter) Error!void { var prepared = (try self.prepareFlush()) orelse return; defer prepared.deinit(self.allocator); const frame = try compressPrepared(self.allocator, &prepared); defer self.allocator.free(frame); try self.commitPrepared(frame); }
/// the active-file bytes so far (valid jss active segment). pub fn bytes(self: *const ActiveWriter) []const u8 { return self.buf.items; }
/// flush pending events, walk our frames, append the footer, and /// finalize the header. after this, bytes() is a sealed jss v1 file. pub fn seal(self: *ActiveWriter) !void { try self.flush();
// Incremental path: every block was indexed as it flushed (or by // rebuildIndex after recovery), so the footer needs no re-walk. The // walk below survives only as the fallback for a writer whose index // fell out of lockstep — it decompresses the entire segment and is // what used to stall ingest ~50s per seal under the archive mutex. if (self.indexed_blocks == self.block_count) { try self.builder.finish(self.allocator, &self.buf, self.env); return; }
// Footer metadata must live until finish(), but decoded block payloads // must not. A single arena for both retained every uncompressed block // in the segment and made recovery of a 268 MiB compressed segment // exceed the canary's 2 GiB cgroup limit. Keep the builder's compact // indexes in one arena and recycle the bulky decode arena per block. var builder_arena = std.heap.ArenaAllocator.init(self.allocator); defer builder_arena.deinit(); const scratch = builder_arena.allocator();
// walk our own frames (mirrors upstream walkActiveFrames) var builder = footer_mod.Builder.init(scratch); var env: footer_mod.Envelope = .{}; var saw_any = false;
var it: ActiveIterator = .{ .bytes = self.buf.items }; while (true) { var block_arena = std.heap.ArenaAllocator.init(self.allocator); defer block_arena.deinit(); const before = it.offset; const block = (try it.next(block_arena.allocator())) orelse break; if (block.events.len == 0) break; // upstream: empty block ends the walk
var info: segment.BlockIndexEntry = .{ .offset = before, .compressed_size = @intCast(it.offset - before - 8), .uncompressed_size = @intCast(block.buffer.len), .event_count = @intCast(block.events.len), .min_seq = block.events[0].seq, .max_seq = block.events[0].seq, .min_witnessed_at = block.events[0].witnessed_at, .max_witnessed_at = block.events[0].witnessed_at, }; for (block.events) |ev| { info.min_seq = @min(info.min_seq, ev.seq); info.max_seq = @max(info.max_seq, ev.seq); info.min_witnessed_at = @min(info.min_witnessed_at, ev.witnessed_at); info.max_witnessed_at = @max(info.max_witnessed_at, ev.witnessed_at); } env.event_count += block.events.len; if (!saw_any) { env.min_seq = info.min_seq; env.max_seq = info.max_seq; env.min_witnessed_at = info.min_witnessed_at; env.max_witnessed_at = info.max_witnessed_at; saw_any = true; } else { env.min_seq = @min(env.min_seq, info.min_seq); env.max_seq = @max(env.max_seq, info.max_seq); env.min_witnessed_at = @min(env.min_witnessed_at, info.min_witnessed_at); env.max_witnessed_at = @max(env.max_witnessed_at, info.max_witnessed_at); } try builder.addBlock(info, block.events); }
try builder.finish(self.allocator, &self.buf, env); }
/// Re-derive the incremental footer index from a resumed file image /// (recovery), paying the one full decompression walk at startup so the /// eventual seal stays off the ingest hot path. pub fn rebuildIndex(self: *ActiveWriter) !void { var it: ActiveIterator = .{ .bytes = self.buf.items }; while (true) { var block_arena = std.heap.ArenaAllocator.init(self.allocator); defer block_arena.deinit(); const before = it.offset; const block = (try it.next(block_arena.allocator())) orelse break; if (block.events.len == 0) break; try self.noteBlock(.{ .offset = before, .compressed_size = @intCast(it.offset - before - 8), .uncompressed_size = @intCast(block.buffer.len), .event_count = 0, .min_seq = 0, .max_seq = 0, .min_witnessed_at = 0, .max_witnessed_at = 0, }, block.events); } }};
pub fn compressPrepared(allocator: Allocator, prepared: *const PreparedBlock) Error![]u8 { return zstd.compress(allocator, prepared.bytes) catch |err| switch (err) { error.OutOfMemory => return error.OutOfMemory, else => return error.CompressFailed, };}
/// walk an active file's frames sequentially (no footer). returns decoded/// blocks one at a time via the iterator.pub const ActiveIterator = struct { bytes: []const u8, offset: usize = segment.header_size,
pub fn next(self: *ActiveIterator, allocator: Allocator) !?segment.Block { if (self.offset == self.bytes.len) return null; if (self.offset + 8 > self.bytes.len) return error.Truncated; const frame_len = std.mem.readInt(u64, self.bytes[self.offset..][0..8], .little); if (self.offset + 8 + frame_len > self.bytes.len) return error.Truncated; const frame = self.bytes[self.offset + 8 ..][0..frame_len]; self.offset += 8 + frame_len;
var in: std.Io.Reader = .fixed(frame); var stream: std.compress.zstd.Decompress = .init(&in, &.{}, .{}); var out: std.Io.Writer.Allocating = .init(allocator); errdefer out.deinit(); _ = stream.reader.streamRemaining(&out.writer) catch return error.DecompressFailed; const buffer = try out.toOwnedSlice(); return try segment.decodeBlock(allocator, buffer); }};
// === tests ===
const testing = std.testing;
fn sampleEvent(seq: u64) segment.Event { return .{ .seq = seq, .witnessed_at = @intCast(1_700_000_000_000_000 + seq), .indexed_at = 0, .kind = .create, .did = "did:plc:sampleaccount123", .collection = "app.bsky.feed.post", .rkey = "3l3qo2vuowo2b", .rev = "3l3qo2vutsw2b", .payload = "\xa1\x64\x74\x65\x78\x74\x62\x68\x69", // {"text": "hi"} };}
test "block encode/decode round-trip" { var events: [10]segment.Event = undefined; for (&events, 0..) |*e, i| e.* = sampleEvent(i + 1); const encoded = try encodeBlock(testing.allocator, &events); var block = try segment.decodeBlock(testing.allocator, encoded); defer block.deinit(testing.allocator); try testing.expectEqual(@as(usize, 10), block.events.len); for (block.events, 0..) |ev, i| { try testing.expectEqual(@as(u64, i + 1), ev.seq); try testing.expectEqualStrings("did:plc:sampleaccount123", ev.did); try testing.expectEqualStrings("app.bsky.feed.post", ev.collection); try testing.expectEqual(segment.Kind.create, ev.kind); try testing.expectEqualStrings("\xa1\x64\x74\x65\x78\x74\x62\x68\x69", ev.payload); }}
test "validateEvent enforces column widths" { var ev = sampleEvent(1); ev.rkey = &@as([256]u8, @splat('x')); try testing.expectError(error.FieldTooLong, validateEvent(ev)); ev = sampleEvent(0); try testing.expectError(error.InvalidEvent, validateEvent(ev));}
test "seal: incremental footer is byte-identical to the walk fallback" { // varied dids/collections/kinds so the blooms, interning table, and // $-sentinels all carry real content into the footer var dids: [13][32]u8 = undefined; var cols: [5][32]u8 = undefined; for (&dids, 0..) |*d, i| _ = std.fmt.bufPrint(d, "did:plc:writer{d:0>17}", .{i}) catch unreachable; for (&cols, 0..) |*c, i| _ = std.fmt.bufPrint(c, "app.test.collection{d:0>3}", .{i}) catch unreachable;
var a = try ActiveWriter.init(testing.allocator); defer a.deinit(); a.max_events_per_block = 100; var seq: u64 = 1; while (seq <= 250) : (seq += 1) { var ev = sampleEvent(seq); ev.did = dids[seq % 13][0..24]; ev.collection = cols[seq % 5][0..22]; if (seq % 11 == 0) { ev.kind = .account; ev.collection = ""; ev.rkey = ""; } try a.append(ev); } try a.flush(); try testing.expectEqual(a.block_count, a.indexed_blocks);
// resumed cold with no index: seal must take the walk fallback var b = try ActiveWriter.init(testing.allocator); defer b.deinit(); b.buf.clearRetainingCapacity(); try b.buf.appendSlice(testing.allocator, a.bytes()); b.event_count = a.event_count; b.block_count = a.block_count;
// resumed the way recovery does: rebuildIndex restores the fast path var c = try ActiveWriter.init(testing.allocator); defer c.deinit(); c.buf.clearRetainingCapacity(); try c.buf.appendSlice(testing.allocator, a.bytes()); c.event_count = a.event_count; c.block_count = a.block_count; try c.rebuildIndex(); try testing.expectEqual(c.block_count, c.indexed_blocks);
try testing.expectEqual(@as(u32, 0), b.indexed_blocks); // proves b walks try a.seal(); try b.seal(); try c.seal(); try testing.expectEqualSlices(u8, b.bytes(), a.bytes()); try testing.expectEqualSlices(u8, b.bytes(), c.bytes());
// and the sealed image parses var sealed = try segment.Sealed.parse(testing.allocator, a.bytes()); defer sealed.deinit(testing.allocator); try testing.expectEqual(@as(u64, 250), sealed.header.event_count);}
test "active file round-trips through our reader's block walk" { var w = try ActiveWriter.init(testing.allocator); defer w.deinit(); w.max_events_per_block = 100;
var seq: u64 = 1; while (seq <= 250) : (seq += 1) try w.append(sampleEvent(seq)); try w.flush(); // tail block (50 events)
try testing.expectEqual(@as(u32, 3), w.block_count); const file = w.bytes();
// header must read as active try testing.expectError(error.ActiveSegment, segment.Header.decode(file));
var it: ActiveIterator = .{ .bytes = file }; var total: usize = 0; var expect_seq: u64 = 1; while (try it.next(testing.allocator)) |*block_const| { var block = block_const.*; defer block.deinit(testing.allocator); for (block.events) |ev| { try testing.expectEqual(expect_seq, ev.seq); expect_seq += 1; } total += block.events.len; } try testing.expectEqual(@as(usize, 250), total);}
test "seal: our reader accepts our sealed file" { var w = try ActiveWriter.init(testing.allocator); defer w.deinit(); w.max_events_per_block = 100;
var seq: u64 = 1; while (seq <= 250) : (seq += 1) { var ev = sampleEvent(seq); if (seq % 25 == 0) { ev.kind = .identity; ev.collection = ""; ev.rkey = ""; ev.rev = ""; ev.payload = "\xa0"; } else if (seq % 3 == 0) { ev.collection = "app.bsky.feed.like"; } try w.append(ev); } try w.seal();
var sealed = try segment.Sealed.parse(testing.allocator, w.bytes()); defer sealed.deinit(testing.allocator); try testing.expectEqual(@as(u32, 3), sealed.header.block_count); try testing.expectEqual(@as(u32, 250), sealed.header.event_count); try testing.expectEqual(@as(u32, 1), sealed.header.unique_did_count); try testing.expectEqual(@as(u64, 1), sealed.header.min_seq); try testing.expectEqual(@as(u64, 250), sealed.header.max_seq);
var total: usize = 0; for (0..sealed.block_index.len) |i| { var block = try sealed.readBlock(testing.allocator, i); defer block.deinit(testing.allocator); total += block.events.len; } try testing.expectEqual(@as(usize, 250), total);
// collection table: first-seen order with $identity sentinel interleaved var ci = try sealed.readCollectionIndex(testing.allocator); defer ci.deinit(testing.allocator); try testing.expectEqual(@as(usize, 3), ci.entries.len); try testing.expectEqualStrings("app.bsky.feed.post", ci.entries[0].nsid); try testing.expectEqualStrings("app.bsky.feed.like", ci.entries[1].nsid); try testing.expectEqualStrings("$identity", ci.entries[2].nsid); try testing.expectEqual(@as(u32, 0), ci.entries[2].count); // sentinels never count try testing.expectEqual(@as(u32, 250 - 10), ci.entries[0].count + ci.entries[1].count);}
test "seal: per-block blooms are sized to actual DID cardinality, not the block cap" { var w = try ActiveWriter.init(testing.allocator); defer w.deinit(); w.max_events_per_block = 100;
// three blocks, every event the same DID: max per-block unique-DID // cardinality is 1. A fixed-capacity bloom (block cap 100, let alone // the production 4096) marshals kilobytes; a right-sized one is tiny. var seq: u64 = 1; while (seq <= 250) : (seq += 1) try w.append(sampleEvent(seq)); try w.seal();
var sealed = try segment.Sealed.parse(testing.allocator, w.bytes()); defer sealed.deinit(testing.allocator);
const blobs = try sealed.blockBloomBlobs(testing.allocator); defer testing.allocator.free(blobs); try testing.expectEqual(@as(usize, 3), blobs.len);
// the pre-fix sizing used the production block cap (4096); right-sized // blooms for cardinality-1 blocks must be far below it var reference = try gloom.Filter.init(testing.allocator, default_max_events_per_block, 0.001); defer reference.deinit(testing.allocator); const fixed_size = (try reference.marshal(testing.allocator)); defer testing.allocator.free(fixed_size); try testing.expect(blobs[0].len * 10 < fixed_size.len);
// right-sizing must not change lookup behavior for (blobs) |blob| { try testing.expect(try gloom.containsMarshaled(blob, "did:plc:sampleaccount123")); try testing.expect(!try gloom.containsMarshaled(blob, "did:plc:someoneelse")); }}