Something went wrong. Try again.
jetstream v2 in zig stream.waow.tech
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432//! Phase C timestamp-import planning and sealed-segment application.//!//! Packed Phase-B offsets are replayed against the original seekable CSV and//! independently revalidated. Rules fold last-write-wins into a bounded-to-one//! segment plan. Applying the plan changes only materialization `indexed_at`;//! a matching specific-version DAG-CBOR CID wins over an all-versions rule.
const std = @import("std");const bucket = @import("bucket.zig");const parse = @import("parse.zig");const segment = @import("../storage/segment.zig");const patch_mod = @import("../storage/segment_patch.zig");
const Allocator = std.mem.Allocator;const Io = std.Io;
const Specific = struct { timestamp: i64, matched: bool = false,};
const Target = struct { did: []u8, collection: []u8, rkey: []u8, all_versions: i64 = 0, specific: std.AutoHashMapUnmanaged(parse.Cid, Specific) = .empty,};
const TargetList = std.ArrayListUnmanaged(Target);
pub const PlanStats = struct { rows_planned: u64 = 0, rows_corrupt: u64 = 0, distinct_paths: usize = 0, paths_all_versions: usize = 0, paths_specific: usize = 0,};
pub const ApplyStats = struct { rows_matched_all_versions: u64 = 0, rows_matched_specific: u64 = 0, specific_cids_unmatched: u64 = 0,};
pub const Plan = struct { allocator: Allocator, paths: std.AutoHashMapUnmanaged(u64, TargetList) = .empty, plan_stats: PlanStats = .{}, apply_stats: ApplyStats = .{},
pub fn init(allocator: Allocator) Plan { return .{ .allocator = allocator }; }
pub fn deinit(self: *Plan) void { var iterator = self.paths.valueIterator(); while (iterator.next()) |list| { for (list.items) |*target| { self.allocator.free(target.did); self.allocator.free(target.collection); self.allocator.free(target.rkey); target.specific.deinit(self.allocator); } list.deinit(self.allocator); } self.paths.deinit(self.allocator); self.* = undefined; }
pub fn empty(self: *const Plan) bool { return self.plan_stats.distinct_paths == 0; }
pub fn stats(self: *const Plan) PlanStats { return self.plan_stats; }
pub fn applied(self: *const Plan) ApplyStats { var result = self.apply_stats; var iterator = self.paths.valueIterator(); while (iterator.next()) |list| for (list.items) |*target| { var specifics = target.specific.valueIterator(); while (specifics.next()) |specific| { if (!specific.matched) result.specific_cids_unmatched += 1; } }; return result; }
fn add(self: *Plan, row: parse.Row) !void { self.plan_stats.rows_planned += 1; const hash = pathHash(row.did, row.collection, row.rkey); const result = try self.paths.getOrPut(self.allocator, hash); if (!result.found_existing) result.value_ptr.* = .empty; var target: *Target = for (result.value_ptr.items) |*candidate| { if (pathEqual(candidate, row.did, row.collection, row.rkey)) break candidate; } else create: { const did = try self.allocator.dupe(u8, row.did); errdefer self.allocator.free(did); const collection = try self.allocator.dupe(u8, row.collection); errdefer self.allocator.free(collection); const rkey = try self.allocator.dupe(u8, row.rkey); errdefer self.allocator.free(rkey); try result.value_ptr.append(self.allocator, .{ .did = did, .collection = collection, .rkey = rkey, }); self.plan_stats.distinct_paths += 1; break :create &result.value_ptr.items[result.value_ptr.items.len - 1]; };
switch (row.scope) { .all_versions => { if (target.all_versions == 0) self.plan_stats.paths_all_versions += 1; target.all_versions = row.timestamp_micros; }, .specific_version => { if (target.specific.count() == 0) self.plan_stats.paths_specific += 1; try target.specific.put(self.allocator, row.cid.?, .{ .timestamp = row.timestamp_micros }); }, } }
pub fn candidateDIDs(self: *Plan, allocator: Allocator) ![][]const u8 { var seen: std.StringHashMapUnmanaged(void) = .empty; defer seen.deinit(allocator); var result: std.ArrayList([]const u8) = .empty; errdefer result.deinit(allocator); var iterator = self.paths.valueIterator(); while (iterator.next()) |list| for (list.items) |target| { const entry = try seen.getOrPut(allocator, target.did); if (!entry.found_existing) try result.append(allocator, target.did); }; return result.toOwnedSlice(allocator); }
pub fn mutateOpaque(self: *Plan, event: *segment.Event) void { if (!isMaterialization(event.kind)) return; const list = self.paths.getPtr(pathHash(event.did, event.collection, event.rkey)) orelse return; const target: *Target = for (list.items) |*candidate| { if (pathEqual(candidate, event.did, event.collection, event.rkey)) break candidate; } else return;
var timestamp: i64 = 0; var specific_match = false; if (target.specific.count() > 0) { var digest: [32]u8 = undefined; std.crypto.hash.sha2.Sha256.hash(event.payload, &digest, .{}); if (target.specific.getPtr(.{ .codec = 0x71, .hash = digest })) |specific| { specific.matched = true; timestamp = specific.timestamp; specific_match = true; } } if (timestamp == 0) timestamp = target.all_versions; if (timestamp == 0 or event.indexed_at == timestamp) return; event.indexed_at = timestamp; if (specific_match) self.apply_stats.rows_matched_specific += 1 else self.apply_stats.rows_matched_all_versions += 1; }};
pub fn buildPlan( allocator: Allocator, io: Io, offset_path: []const u8, row_reader: *parse.RowReader,) !Plan { var plan = Plan.init(allocator); errdefer plan.deinit(); var file = try Io.Dir.cwd().openFile(io, offset_path, .{}); defer file.close(io); const stat = try file.stat(io); if (stat.size % bucket.offset_record_size != 0) return error.PartialOffsetRecord; var position: u64 = 0; var encoded: [bucket.offset_record_size]u8 = undefined; while (position < stat.size) : (position += encoded.len) { if (try file.readPositionalAll(io, &encoded, position) != encoded.len) return error.PartialOffsetRecord; const offset = std.mem.readInt(u64, &encoded, .little); const row = row_reader.read(offset) catch |err| switch (err) { error.CorruptOffset => { plan.plan_stats.rows_corrupt += 1; continue; }, else => return err, }; try plan.add(row); } return plan;}
pub fn applyFile( allocator: Allocator, io: Io, segments_dir: Io.Dir, segment_name: []const u8, plan: *Plan,) !patch_mod.Result { if (plan.empty()) return .{}; const dids = try plan.candidateDIDs(allocator); defer allocator.free(dids); return patch_mod.patchFile(allocator, io, segments_dir, segment_name, dids, plan, Plan.mutateOpaque);}
fn pathHash(did: []const u8, collection: []const u8, rkey: []const u8) u64 { var hasher = std.hash.XxHash3.init(0); var length: [8]u8 = undefined; inline for (.{ did, collection, rkey }) |part| { std.mem.writeInt(u64, &length, part.len, .little); hasher.update(&length); hasher.update(part); } return hasher.final();}
fn pathEqual(target: *const Target, did: []const u8, collection: []const u8, rkey: []const u8) bool { return std.mem.eql(u8, target.did, did) and std.mem.eql(u8, target.collection, collection) and std.mem.eql(u8, target.rkey, rkey);}
fn isMaterialization(kind: segment.Kind) bool { return switch (kind) { .create, .update, .create_resync => true, else => false, };}
// === offline disk-to-disk tests ===
const testing = std.testing;const writer = @import("../storage/segment_writer.zig");const zat = @import("zat");
fn makeEvent(seq: u64, kind: segment.Kind, did: []const u8, rkey: []const u8, payload: []const u8) segment.Event { return .{ .seq = seq, .witnessed_at = @intCast(10_000 + seq), .indexed_at = 0, .kind = kind, .did = did, .collection = "app.bsky.feed.post", .rkey = rkey, .rev = "v1", .payload = payload, };}
fn writeSegment(io: Io, dir: Io.Dir, name: []const u8, events: []const segment.Event) !void { var active = try writer.ActiveWriter.init(testing.allocator); defer active.deinit(); active.max_events_per_block = 2; for (events) |item| try active.append(item); try active.seal(); var file = try dir.createFile(io, name, .{}); defer file.close(io); try file.writeStreamingAll(io, active.bytes()); try file.sync(io);}
test "parse bucket seek plan and patch preserve upstream timestamp semantics" { var threaded: Io.Threaded = .init(testing.allocator, .{}); defer threaded.deinit(); const io = threaded.io(); var tmp = testing.tmpDir(.{}); defer tmp.cleanup(); var root_buffer: [std.fs.max_path_bytes]u8 = undefined; const root_len = try tmp.dir.realPath(io, &root_buffer); const root = root_buffer[0..root_len]; var segments_dir = try tmp.dir.createDirPathOpen(io, "segments", .{ .open_options = .{ .iterate = true } }); defer segments_dir.close(io);
const payload_a = "\xa1\x61\x78\x01"; const payload_b = "\xa1\x61\x78\x02"; const first = [_]segment.Event{ makeEvent(1, .create, "did:plc:a", "r1", payload_a), makeEvent(2, .update, "did:plc:a", "r1", payload_b), makeEvent(3, .delete, "did:plc:a", "r1", ""), makeEvent(4, .create, "did:plc:b", "r2", payload_a), }; const second = [_]segment.Event{ makeEvent(5, .create_resync, "did:plc:a", "r1", payload_a), }; try writeSegment(io, segments_dir, "seg_0000000000.jss", &first); try writeSegment(io, segments_dir, "seg_0000000001.jss", &second);
const cid_a_raw = try zat.cbor.Cid.forDagCbor(testing.allocator, payload_a); defer testing.allocator.free(@constCast(cid_a_raw.raw)); const cid_a = try @import("../serve/wire.zig").cidString(testing.allocator, cid_a_raw); defer testing.allocator.free(cid_a); const unmatched_raw = try zat.cbor.Cid.forDagCbor(testing.allocator, "unmatched"); defer testing.allocator.free(@constCast(unmatched_raw.raw)); const unmatched = try @import("../serve/wire.zig").cidString(testing.allocator, unmatched_raw); defer testing.allocator.free(unmatched); const csv = try std.fmt.allocPrint(testing.allocator, "uri,timestamp,scope,cid\n" ++ "at://did:plc:a/app.bsky.feed.post/r1,2020-01-01T00:00:00Z,all_versions,\n" ++ "at://did:plc:a/app.bsky.feed.post/r1,2022-01-02T03:04:05Z,specific_version,{s}\n" ++ "at://did:plc:a/app.bsky.feed.post/r1,2024-01-02T03:04:05Z,specific_version,{s}\n" ++ "at://did:plc:b/app.bsky.feed.post/r2,2021-01-01T00:00:00Z,all_versions,\n" ++ "at://did:plc:a/app.bsky.feed.post/r1,2023-01-02T03:04:05Z,all_versions,\n" ++ "at://did:plc:not-resident/app.bsky.feed.post/r9,2025-01-01T00:00:00Z,all_versions,\n", .{ cid_a, unmatched }); defer testing.allocator.free(csv); const csv_path = try std.fs.path.join(testing.allocator, &.{ root, "import.csv" }); defer testing.allocator.free(csv_path); try Io.Dir.cwd().writeFile(io, .{ .sub_path = csv_path, .data = csv }); const job_path = try std.fs.path.join(testing.allocator, &.{ root, "job" }); defer testing.allocator.free(job_path);
var catalog = try bucket.Catalog.open(testing.allocator, io, segments_dir); defer catalog.deinit(); try testing.expectEqual(@as(usize, 2), catalog.segments.items.len); var bucketer = try bucket.Bucketer.init(testing.allocator, io, &catalog, job_path, .{ .did_cache_size = 1, .open_file_limit = 1, }); defer bucketer.deinit(); var csv_file = try Io.Dir.cwd().openFile(io, csv_path, .{}); defer csv_file.close(io); var read_buffer: [4096]u8 = undefined; var csv_reader = csv_file.reader(io, &read_buffer); var parse_stats = try parse.parse(testing.allocator, &csv_reader.interface, .{ .context = &bucketer, .on_row = bucket.Bucketer.routeOpaque, .reject_sample_limit = parse.Options.no_reject_sample, }); defer parse_stats.deinit(testing.allocator); try bucketer.close(); try testing.expectEqual(@as(u64, 6), parse_stats.rows_valid); const bucket_stats = bucketer.statsSnapshot(); try testing.expectEqual(@as(u64, 5), bucket_stats.rows_routed); try testing.expectEqual(@as(u64, 1), bucket_stats.rows_no_candidate); try testing.expectEqual(@as(u64, 9), bucket_stats.offsets_written); try testing.expectEqual(@as(usize, 2), bucket_stats.segments_touched);
var row_reader = try parse.RowReader.open(io, csv_path); defer row_reader.close(); var offset_name: [64]u8 = undefined; const offset_zero = try std.fs.path.join(testing.allocator, &.{ job_path, try bucket.offsetFileName(&offset_name, 0) }); defer testing.allocator.free(offset_zero); var plan_zero = try buildPlan(testing.allocator, io, offset_zero, &row_reader); defer plan_zero.deinit(); try testing.expectEqual(@as(u64, 5), plan_zero.stats().rows_planned); try testing.expectEqual(@as(usize, 2), plan_zero.stats().distinct_paths); const result_zero = try applyFile(testing.allocator, io, segments_dir, "seg_0000000000.jss", &plan_zero); try testing.expect(result_zero.patched); try testing.expectEqual(@as(u64, 3), result_zero.rows_mutated); const applied_zero = plan_zero.applied(); try testing.expectEqual(@as(u64, 1), applied_zero.rows_matched_specific); try testing.expectEqual(@as(u64, 2), applied_zero.rows_matched_all_versions); try testing.expectEqual(@as(u64, 1), applied_zero.specific_cids_unmatched);
const offset_one = try std.fs.path.join(testing.allocator, &.{ job_path, try bucket.offsetFileName(&offset_name, 1) }); defer testing.allocator.free(offset_one); var plan_one = try buildPlan(testing.allocator, io, offset_one, &row_reader); defer plan_one.deinit(); try testing.expectEqual(@as(u64, 4), plan_one.stats().rows_planned); const result_one = try applyFile(testing.allocator, io, segments_dir, "seg_0000000001.jss", &plan_one); try testing.expect(result_one.patched); try testing.expectEqual(@as(u64, 1), result_one.rows_mutated);
const expected_specific: i64 = 1_641_092_645_000_000; const expected_all: i64 = 1_672_628_645_000_000; const expected_b: i64 = 1_609_459_200_000_000; for ([_][]const u8{ "seg_0000000000.jss", "seg_0000000001.jss" }) |name| { const bytes = try segments_dir.readFileAlloc(io, name, testing.allocator, .limited(1 << 20)); defer testing.allocator.free(bytes); var sealed = try segment.Sealed.parse(testing.allocator, bytes); defer sealed.deinit(testing.allocator); for (0..sealed.block_index.len) |block_index| { var block_data = try sealed.readBlock(testing.allocator, block_index); defer block_data.deinit(testing.allocator); for (block_data.events) |stored| { const wanted: i64 = if (stored.kind == .delete) 0 else if (std.mem.eql(u8, stored.did, "did:plc:b")) expected_b else if (std.mem.eql(u8, stored.payload, payload_a)) expected_specific else expected_all; try testing.expectEqual(wanted, stored.indexed_at); } } }
// Re-applying the same plan is a genuine no-op at the file layer. const again = try applyFile(testing.allocator, io, segments_dir, "seg_0000000000.jss", &plan_zero); try testing.expect(!again.patched);}
test "row reader and plan reject stale mid-row and torn offsets" { var threaded: Io.Threaded = .init(testing.allocator, .{}); defer threaded.deinit(); const io = threaded.io(); var tmp = testing.tmpDir(.{}); defer tmp.cleanup(); var root_buffer: [std.fs.max_path_bytes]u8 = undefined; const root_len = try tmp.dir.realPath(io, &root_buffer); const root = root_buffer[0..root_len]; const csv_path = try std.fs.path.join(testing.allocator, &.{ root, "import.csv" }); defer testing.allocator.free(csv_path); const csv = "uri,timestamp\n" ++ "at://did:plc:a/app.bsky.feed.post/r1,2022-01-02T03:04:05Z\n"; try Io.Dir.cwd().writeFile(io, .{ .sub_path = csv_path, .data = csv }); var reader = try parse.RowReader.open(io, csv_path); defer reader.close(); const valid_offset = "uri,timestamp\n".len; const valid = try reader.read(valid_offset); try testing.expectEqualStrings("did:plc:a", valid.did); try testing.expectError(error.CorruptOffset, reader.read(valid_offset + 1)); try testing.expectError(error.CorruptOffset, reader.read(0));
const offsets_path = try std.fs.path.join(testing.allocator, &.{ root, "offsets.bin" }); defer testing.allocator.free(offsets_path); var encoded: [16]u8 = undefined; std.mem.writeInt(u64, encoded[0..8], valid_offset, .little); std.mem.writeInt(u64, encoded[8..16], valid_offset + 1, .little); try Io.Dir.cwd().writeFile(io, .{ .sub_path = offsets_path, .data = &encoded }); var plan = try buildPlan(testing.allocator, io, offsets_path, &reader); defer plan.deinit(); try testing.expectEqual(@as(u64, 1), plan.stats().rows_planned); try testing.expectEqual(@as(u64, 1), plan.stats().rows_corrupt);
try Io.Dir.cwd().writeFile(io, .{ .sub_path = offsets_path, .data = encoded[0..9] }); try testing.expectError(error.PartialOffsetRecord, buildPlan(testing.allocator, io, offsets_path, &reader));}