From 9868e983beba57c7a379de439e90f6b426667e85 Mon Sep 17 00:00:00 2001 From: zzstoatzz Date: Wed, 8 Jul 2026 22:47:59 -0500 Subject: [PATCH] =?UTF-8?q?segment:=20writer=20part=201=20=E2=80=94=20bloc?= =?UTF-8?q?k=20encode=20+=20active=20files=20(libzstd)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit columnar block encoder (validateEvent mirroring upstream's column widths — oversized fields are the caller's drop, never truncated), minimal libzstd bindings (compress2, content checksum on, level 3 to match upstream's encoder), and ActiveWriter producing valid active jss files: 256B active header + length-prefixed zstd frames. pure format code — callers own durability. cross-implementation verification both ways: our reader parses upstream's sealed fixture (previous commit), and upstream's inspect-segment now walks a file written by us — 3 blocks, 2500 events, correct seq ranges/collections (zig build write-sample). part 2 (sealing) needs xxh3-128 for gloom DID blooms plus the footer walk; zig std only has xxh3-64. Co-Authored-By: Claude Fable 5 --- build.zig | 24 +++ src/internal/segment.zig | 5 +- src/internal/segment_writer.zig | 259 ++++++++++++++++++++++++++++++++ src/internal/zstd.zig | 56 +++++++ src/tests.zig | 2 + src/write_sample.zig | 42 ++++++ 6 files changed, 387 insertions(+), 1 deletion(-) create mode 100644 src/internal/segment_writer.zig create mode 100644 src/internal/zstd.zig create mode 100644 src/write_sample.zig diff --git a/build.zig b/build.zig index 337801f..de5bbb0 100644 --- a/build.zig +++ b/build.zig @@ -1,5 +1,15 @@ const std = @import("std"); +fn linkZstd(mod: *std.Build.Module, b: *std.Build) void { + // homebrew installs outside the default search path on macos + if (@import("builtin").os.tag == .macos) { + mod.addLibraryPath(.{ .cwd_relative = "/opt/homebrew/lib" }); + mod.addIncludePath(.{ .cwd_relative = "/opt/homebrew/include" }); + } + mod.linkSystemLibrary("zstd", .{}); + _ = b; +} + pub fn build(b: *std.Build) void { const target = b.standardTargetOptions(.{}); const optimize = b.standardOptimizeOption(.{}); @@ -39,6 +49,7 @@ pub fn build(b: *std.Build) void { }); exe_mod.addImport("build_options", build_options.createModule()); exe_mod.link_libc = true; + linkZstd(exe_mod, b); const exe = b.addExecutable(.{ .name = "stream", .root_module = exe_mod, @@ -50,6 +61,18 @@ pub fn build(b: *std.Build) void { const run_step = b.step("run", "run stream"); run_step.dependOn(&run_exe.step); + const sample_mod = b.createModule(.{ + .root_source_file = b.path("src/write_sample.zig"), + .target = target, + .optimize = optimize, + }); + sample_mod.link_libc = true; + linkZstd(sample_mod, b); + const write_sample = b.addExecutable(.{ .name = "write-sample-segment", .root_module = sample_mod }); + const run_sample = b.addRunArtifact(write_sample); + if (b.args) |args| run_sample.addArgs(args); + b.step("write-sample", "write a sample active segment (writer dev tool)").dependOn(&run_sample.step); + const test_step = b.step("test", "run unit tests"); const test_mod = b.createModule(.{ .root_source_file = b.path("src/tests.zig"), @@ -59,6 +82,7 @@ pub fn build(b: *std.Build) void { }); test_mod.addImport("build_options", build_options.createModule()); test_mod.link_libc = true; + linkZstd(test_mod, b); const t = b.addTest(.{ .root_module = test_mod, }); diff --git a/src/internal/segment.zig b/src/internal/segment.zig index f2aecb7..8beb165 100644 --- a/src/internal/segment.zig +++ b/src/internal/segment.zig @@ -160,7 +160,10 @@ pub const CollectionIndex = struct { } pub fn blockHasCollection(self: *const CollectionIndex, block: usize, collection_id: usize) bool { - const row = self.bitmasks[block * self.bitmask_len ..][0..self.bitmask_len]; + if (collection_id >= self.entries.len) return false; + const at = block * self.bitmask_len; + if (at + self.bitmask_len > self.bitmasks.len) return false; + const row = self.bitmasks[at..][0..self.bitmask_len]; return (row[collection_id / 8] >> @intCast(collection_id % 8)) & 1 == 1; } }; diff --git a/src/internal/segment_writer.zig b/src/internal/segment_writer.zig new file mode 100644 index 0000000..7d5b27b --- /dev/null +++ b/src/internal/segment_writer.zig @@ -0,0 +1,259 @@ +//! 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). +//! +//! not yet (part 2): sealing — per-block stats walk, gloom DID blooms +//! (needs xxh3-128), collection index, footer, finalized header + xxh3. +//! until then upstream's inspect-segment reads our files in "active +//! (unsealed; block walk)" mode. + +const std = @import("std"); +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, +}; + +/// 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 + 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] = @intFromEnum(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, + + pub fn init(allocator: Allocator) Error!ActiveWriter { + var w: ActiveWriter = .{ + .allocator = allocator, + .pending_arena = std.heap.ArenaAllocator.init(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(); + } + + /// stage one event; flushes a block when the threshold is reached. + pub fn append(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; + if (self.pending.items.len >= self.max_events_per_block) try self.flush(); + } + + /// compress staged events into a frame and append it to the file image. + pub fn flush(self: *ActiveWriter) Error!void { + if (self.pending.items.len == 0) return; + const block = try encodeBlock(self.allocator, self.pending.items); + defer self.allocator.free(block); + const frame = zstd.compress(self.allocator, block) catch |err| switch (err) { + error.OutOfMemory => return error.OutOfMemory, + else => return error.CompressFailed, + }; + defer self.allocator.free(frame); + + 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; + self.pending.clearRetainingCapacity(); + _ = self.pending_arena.reset(.retain_capacity); + } + + /// the active-file bytes so far (valid jss active segment). + pub fn bytes(self: *const ActiveWriter) []const u8 { + return self.buf.items; + } +}; + +/// 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 = "x" ** 256; + try testing.expectError(error.FieldTooLong, validateEvent(ev)); + ev = sampleEvent(0); + try testing.expectError(error.InvalidEvent, validateEvent(ev)); +} + +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); +} diff --git a/src/internal/zstd.zig b/src/internal/zstd.zig new file mode 100644 index 0000000..d04ef96 --- /dev/null +++ b/src/internal/zstd.zig @@ -0,0 +1,56 @@ +//! minimal libzstd bindings — compression only +//! +//! decompression stays on std.compress.zstd; the writer needs a compressor +//! (absent from zig std) producing frames with content checksums, matching +//! upstream's encoder settings (SpeedDefault ≈ level 3, WithEncoderCRC). + +const std = @import("std"); + +const c = struct { + extern fn ZSTD_createCCtx() ?*anyopaque; + extern fn ZSTD_freeCCtx(cctx: ?*anyopaque) usize; + extern fn ZSTD_CCtx_setParameter(cctx: ?*anyopaque, param: c_int, value: c_int) usize; + extern fn ZSTD_compress2(cctx: ?*anyopaque, dst: [*]u8, dst_cap: usize, src: [*]const u8, src_len: usize) usize; + extern fn ZSTD_compressBound(src_len: usize) usize; + extern fn ZSTD_isError(code: usize) c_uint; + + // ZSTD_cParameter values (zstd.h, stable API) + const ZSTD_c_compressionLevel: c_int = 100; + const ZSTD_c_checksumFlag: c_int = 201; +}; + +pub const level_default: c_int = 3; + +/// compress `src` into one zstd frame with a content checksum. +/// returned slice is allocated from `allocator`. +pub fn compress(allocator: std.mem.Allocator, src: []const u8) ![]u8 { + const cctx = c.ZSTD_createCCtx() orelse return error.OutOfMemory; + defer _ = c.ZSTD_freeCCtx(cctx); + if (c.ZSTD_isError(c.ZSTD_CCtx_setParameter(cctx, c.ZSTD_c_compressionLevel, level_default)) != 0) + return error.CompressFailed; + if (c.ZSTD_isError(c.ZSTD_CCtx_setParameter(cctx, c.ZSTD_c_checksumFlag, 1)) != 0) + return error.CompressFailed; + + const bound = c.ZSTD_compressBound(src.len); + const dst = try allocator.alloc(u8, bound); + errdefer allocator.free(dst); + const written = c.ZSTD_compress2(cctx, dst.ptr, dst.len, src.ptr, src.len); + if (c.ZSTD_isError(written) != 0) return error.CompressFailed; + return allocator.realloc(dst, written); +} + +// === tests === + +test "round-trip through std decompress" { + const src = "the quick brown fox jumps over the lazy dog" ** 100; + const frame = try compress(std.testing.allocator, src); + defer std.testing.allocator.free(frame); + try std.testing.expect(frame.len < src.len); + + var in: std.Io.Reader = .fixed(frame); + var stream: std.compress.zstd.Decompress = .init(&in, &.{}, .{}); + var out: std.Io.Writer.Allocating = .init(std.testing.allocator); + defer out.deinit(); + _ = try stream.reader.streamRemaining(&out.writer); + try std.testing.expectEqualStrings(src, out.written()); +} diff --git a/src/tests.zig b/src/tests.zig index 452d8a0..35b4557 100644 --- a/src/tests.zig +++ b/src/tests.zig @@ -4,6 +4,8 @@ comptime { _ = @import("internal/cursor.zig"); _ = @import("internal/metrics.zig"); _ = @import("internal/segment.zig"); + _ = @import("internal/zstd.zig"); + _ = @import("internal/segment_writer.zig"); _ = @import("internal/segment_fixture_test.zig"); _ = @import("internal/verify.zig"); _ = @import("internal/wire.zig"); diff --git a/src/write_sample.zig b/src/write_sample.zig new file mode 100644 index 0000000..e4af6b6 --- /dev/null +++ b/src/write_sample.zig @@ -0,0 +1,42 @@ +//! dev tool: write a sample active segment to the path in argv[1] +//! (used to differentially verify the writer against upstream's +//! inspect-segment; see docs/jss-format-v1.md) + +const std = @import("std"); +const segment = @import("internal/segment.zig"); +const writer_mod = @import("internal/segment_writer.zig"); + +pub fn main(init: std.process.Init.Minimal) !void { + const allocator = std.heap.c_allocator; + var threaded: std.Io.Threaded = .init(allocator, .{}); + defer threaded.deinit(); + const io = threaded.io(); + + var args = init.args.iterate(); + _ = args.next(); + const path = args.next() orelse return error.MissingPath; + + var w = try writer_mod.ActiveWriter.init(allocator); + defer w.deinit(); + w.max_events_per_block = 1000; + + var seq: u64 = 1; + while (seq <= 2500) : (seq += 1) { + try w.append(.{ + .seq = seq, + .witnessed_at = @intCast(1_700_000_000_000_000 + seq * 1000), + .indexed_at = 0, + .kind = if (seq % 10 == 0) .delete else .create, + .did = "did:plc:streamsampleaccount", + .collection = "app.bsky.feed.post", + .rkey = "3l3qo2vuowo2b", + .rev = "3l3qo2vutsw2b", + .payload = if (seq % 10 == 0) "" else "\xa1\x64\x74\x65\x78\x74\x62\x68\x69", + }); + } + try w.flush(); + + var f = try std.Io.Dir.cwd().createFile(io, path, .{}); + defer f.close(io); + try f.writeStreamingAll(io, w.bytes()); +} -- 2.51.2