Something went wrong. Try again.
jetstream v2 in zig stream.waow.tech
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498//! Crash-resumable timestamp-import execution over the real archive.//!//! A fresh run durably activates append rules, force-seals the active segment,//! buckets CSV offsets, then records `bucketed=true`. Phase C rewrites sealed//! files under Archive's rewrite lock and checkpoints each segment only after//! its atomic patch is durable. A restarted process resumes from those exact//! files and RocksDB markers.
const std = @import("std");const archive_mod = @import("../storage/archive.zig");const apply = @import("apply.zig");const bucket = @import("bucket.zig");const jobs_mod = @import("jobs.zig");const parse = @import("parse.zig");const rules_mod = @import("rules.zig");const metrics = @import("../runtime/metrics.zig");const observability = @import("../runtime/observability.zig");
const Allocator = std.mem.Allocator;const Io = std.Io;const log = std.log.scoped(.stream_timestamp_import);
pub const Config = struct { allocator: Allocator, io: Io, scratch_root: []const u8, archive: *archive_mod.Archive, rules: *rules_mod.Store, jobs: *jobs_mod.Store, stats: ?*metrics.Stats = null, cancelled: ?*const std.atomic.Value(bool) = null, on_segment_applied_ctx: ?*anyopaque = null, on_segment_applied: ?*const fn (?*anyopaque, u64) void = null, trace_context: observability.Context = observability.root_context,};
const Touched = struct { index: u64, segment_name: []const u8,};
pub fn run(config: Config, id: []const u8) !void { var trace = observability.Span.start( config.allocator, "ingest/orchestrator", "RunImport", config.trace_context, &.{}, ); defer trace.deinit(); defer trace.succeed(); try trace.recordError(runJob(config, id, trace.context()));}
fn runJob(config: Config, id: []const u8, trace_context: observability.Context) !void { const started_us = Io.Timestamp.now(config.io, .real).toMicroseconds(); defer if (config.stats) |stats| stats.import_phase.store(0, .release); var parsed = (try config.jobs.get(id)) orelse return error.ImportJobNotFound; defer parsed.deinit(); var record = parsed.value; if (record.state.terminal()) return error.ImportJobTerminal;
runInner(config, &record, trace_context) catch |err| { if (err == error.ImportCancelled) { try config.jobs.put(record); return @errorCast(err); } record.state = .failed; record.error_message = @errorName(err); record.finished_at_us = Io.Timestamp.now(config.io, .real).toMicroseconds(); try config.jobs.put(record); observeTerminal(config, false, started_us, record.finished_at_us); cleanup(config, record.id) catch {}; return @errorCast(err); };
record.state = .complete; record.phase = .apply; record.error_message = ""; record.finished_at_us = Io.Timestamp.now(config.io, .real).toMicroseconds(); try config.jobs.put(record); observeTerminal(config, true, started_us, record.finished_at_us); cleanup(config, record.id) catch {};}
fn runInner(config: Config, record: *jobs_mod.Record, trace_context: observability.Context) !void { const job_path = try std.fs.path.join(config.allocator, &.{ config.scratch_root, record.id }); defer config.allocator.free(job_path);
// The catalog is a point-in-time view. Keep compaction out from the fresh // force-seal through the last patch; live append uses a different mutex. config.archive.lockRewrites(); defer config.archive.unlockRewrites();
if (!record.bucketed) { if (config.stats) |stats| stats.import_phase.store(1, .release); try std.Io.Dir.cwd().deleteTree(config.io, job_path); try config.jobs.dropDone(record.id); try std.Io.Dir.cwd().createDirPath(config.io, job_path);
const rule_scratch = try std.fs.path.join(config.allocator, &.{ job_path, "rules" }); defer config.allocator.free(rule_scratch); var rule_result: rules_mod.ImportResult = .{}; defer rule_result.deinit(config.allocator); var rule_error: ?anyerror = null; config.rules.importCsv(.{ .csv_path = record.csv_path, .scratch_dir = rule_scratch, .cancelled = config.cancelled, }, &rule_result) catch |err| { rule_error = err; }; record.rows_total = rule_result.parse.rows_total; record.rows_valid = rule_result.parse.rows_valid; record.rows_rejected = rule_result.parse.rows_rejected; record.rejects_by_reason = rule_result.parse.rejects_by_reason; var bucket_parse_observed = false; defer if (!bucket_parse_observed) observeParse(config, &rule_result.parse); if (rule_error) |err| return err; log.info("timestamp import {s}: rules committed", .{record.id});
// Rules are live before this boundary. Pre-existing active rows become // sealed and enter the catalog; later rows are stamped at append time. config.archive.rotateTraced(trace_context) catch |err| { log.err("timestamp import {s}: archive rotate failed: {s}", .{ record.id, @errorName(err) }); return err; }; log.info("timestamp import {s}: archive rotated", .{record.id}); var catalog = bucket.Catalog.open(config.allocator, config.io, config.archive.dir) catch |err| { log.err("timestamp import {s}: catalog open failed: {s}", .{ record.id, @errorName(err) }); return err; }; defer catalog.deinit(); var bucketer = bucket.Bucketer.init(config.allocator, config.io, &catalog, job_path, .{}) catch |err| { log.err("timestamp import {s}: bucketer init failed: {s}", .{ record.id, @errorName(err) }); return err; }; defer bucketer.deinit(); log.info("timestamp import {s}: bucketer initialized", .{record.id}); var csv = try std.Io.Dir.cwd().openFile(config.io, record.csv_path, .{}); defer csv.close(config.io); var read_buffer: [64 * 1024]u8 = undefined; var reader = csv.reader(config.io, &read_buffer); var stats: parse.Stats = .{}; defer stats.deinit(config.allocator); const Route = struct { config: Config, bucketer: *bucket.Bucketer,
fn row(raw: ?*anyopaque, item: parse.Row) !void { const self: *@This() = @ptrCast(@alignCast(raw.?)); if (isCancelled(self.config)) return error.ImportCancelled; try self.bucketer.route(item); } }; var route: Route = .{ .config = config, .bucketer = &bucketer }; var parse_error: ?anyerror = null; parse.parseInto(config.allocator, &reader.interface, .{ .context = &route, .on_row = Route.row, }, &stats) catch |err| { parse_error = err; }; var close_error: ?anyerror = null; bucketer.close() catch |err| { close_error = err; };
record.rows_total = stats.rows_total; record.rows_valid = stats.rows_valid; record.rows_rejected = stats.rows_rejected; record.rejects_by_reason = stats.rejects_by_reason; observeParse(config, &stats); bucket_parse_observed = true; if (config.stats) |metric_stats| _ = metric_stats.import_dids_matched_total.fetchAdd( bucketer.statsSnapshot().rows_routed, .monotonic, ); if (parse_error) |err| { log.err("timestamp import {s}: CSV bucketing failed: {s}", .{ record.id, @errorName(err) }); return err; } if (close_error) |err| { log.err("timestamp import {s}: bucket durability close failed: {s}", .{ record.id, @errorName(err) }); return err; } record.bucketed = true; record.phase = .apply; record.segments_to_apply = bucketer.statsSnapshot().segments_touched; try config.jobs.put(record.*); log.info("timestamp import {s}: bucketing committed ({d} segments)", .{ record.id, record.segments_to_apply }); }
if (isCancelled(config)) return error.ImportCancelled; if (config.stats) |stats| stats.import_phase.store(2, .release); var catalog = try bucket.Catalog.open(config.allocator, config.io, config.archive.dir); defer catalog.deinit(); var done = try config.jobs.loadDoneFor(record.id); defer done.deinit(config.allocator); record.segments_applied = done.count();
const touched = try touchedSegments(config, &catalog, job_path, &done); defer config.allocator.free(touched); record.segments_to_apply = record.segments_applied + touched.len; record.phase = .apply; try config.jobs.put(record.*);
var row_reader = try parse.RowReader.open(config.io, record.csv_path); defer row_reader.close(); for (touched) |item| { if (isCancelled(config)) return error.ImportCancelled; var offset_name_buffer: [64]u8 = undefined; const offset_path = try std.fs.path.join(config.allocator, &.{ job_path, try bucket.offsetFileName(&offset_name_buffer, item.index), }); defer config.allocator.free(offset_path); var plan = try apply.buildPlan(config.allocator, config.io, offset_path, &row_reader); defer plan.deinit(); const patch = try apply.applyFile( config.allocator, config.io, config.archive.dir, item.segment_name, &plan, ); log.info("timestamp import {s}: segment {d} patch completed (patched={})", .{ record.id, item.index, patch.patched }); const plan_stats = plan.stats(); const applied = plan.applied(); record.segments_examined += 1; record.rows_corrupt_offset += plan_stats.rows_corrupt; record.rows_matched_all_versions += applied.rows_matched_all_versions; record.rows_matched_specific += applied.rows_matched_specific; record.specific_cids_unmatched += applied.specific_cids_unmatched; var rewritten_bytes: u64 = 0; if (patch.patched) { try config.archive.manifest.refresh(config.archive.dir, item.index, true); record.segments_patched += 1; record.rows_mutated += patch.rows_mutated; var file = try config.archive.dir.openFile(config.io, item.segment_name, .{}); defer file.close(config.io); rewritten_bytes = (try file.stat(config.io)).size; record.bytes_rewritten += rewritten_bytes; } if (config.stats) |stats| { _ = stats.import_segments_examined_total.fetchAdd(1, .monotonic); _ = stats.import_rows_corrupt_offset_total.fetchAdd(plan_stats.rows_corrupt, .monotonic); _ = stats.import_rows_matched_all_total.fetchAdd(applied.rows_matched_all_versions, .monotonic); _ = stats.import_rows_matched_specific_total.fetchAdd(applied.rows_matched_specific, .monotonic); if (patch.patched) { _ = stats.import_segments_patched_total.fetchAdd(1, .monotonic); _ = stats.import_rows_mutated_total.fetchAdd(patch.rows_mutated, .monotonic); _ = stats.import_bytes_rewritten_total.fetchAdd(rewritten_bytes, .monotonic); } }
// The file rename and directory sync precede this synchronous marker. try config.jobs.checkpointSegment(record.id, item.index); try done.put(config.allocator, item.index, {}); record.segments_applied = done.count(); try config.jobs.put(record.*); if (config.on_segment_applied) |notify| notify(config.on_segment_applied_ctx, item.index); }}
fn touchedSegments( config: Config, catalog: *const bucket.Catalog, job_path: []const u8, done: *const std.AutoHashMapUnmanaged(u64, void),) ![]Touched { var result: std.ArrayList(Touched) = .empty; errdefer result.deinit(config.allocator); var dir = try std.Io.Dir.cwd().openDir(config.io, job_path, .{}); defer dir.close(config.io); for (catalog.segments.items) |item| { if (done.contains(item.index)) continue; var name_buffer: [64]u8 = undefined; const name = try bucket.offsetFileName(&name_buffer, item.index); var file = dir.openFile(config.io, name, .{}) catch |err| switch (err) { error.FileNotFound => continue, else => return err, }; file.close(config.io); try result.append(config.allocator, .{ .index = item.index, .segment_name = item.name }); } return result.toOwnedSlice(config.allocator);}
fn isCancelled(config: Config) bool { return if (config.cancelled) |flag| flag.load(.acquire) else false;}
fn observeParse(config: Config, parsed: *const parse.Stats) void { const stats = config.stats orelse return; _ = stats.import_rows_parsed_total.fetchAdd(parsed.rows_total, .monotonic); inline for (comptime std.enums.values(parse.RejectReason)) |reason| { _ = stats.import_rows_rejected.getPtr(reason).fetchAdd(parsed.rejected(reason), .monotonic); }}
fn observeTerminal(config: Config, success: bool, started_us: i64, finished_us: i64) void { const stats = config.stats orelse return; if (success) _ = stats.import_jobs_ok_total.fetchAdd(1, .monotonic) else _ = stats.import_jobs_error_total.fetchAdd(1, .monotonic); const duration_us: u64 = @intCast(@max(finished_us - started_us, 0)); observeDuration(stats, duration_us);}
fn observeDuration(stats: *metrics.Stats, duration_us: u64) void { _ = stats.import_job_duration_count.fetchAdd(1, .monotonic); _ = stats.import_job_duration_sum_us.fetchAdd(duration_us, .monotonic); for (0..16) |bucket_index| { const upper_us = (@as(u64, 1) << @intCast(bucket_index)) * std.time.us_per_s; if (duration_us <= upper_us) _ = stats.import_job_duration_buckets[bucket_index].fetchAdd(1, .monotonic); }}
test "duration histogram compares fractional seconds without flooring" { var stats: metrics.Stats = .{}; observeDuration(&stats, 1_500_000);
try std.testing.expectEqual(@as(u64, 0), stats.import_job_duration_buckets[0].load(.monotonic)); try std.testing.expectEqual(@as(u64, 1), stats.import_job_duration_buckets[1].load(.monotonic)); try std.testing.expectEqual(@as(u64, 1), stats.import_job_duration_buckets[15].load(.monotonic)); try std.testing.expectEqual(@as(u64, 1), stats.import_job_duration_count.load(.monotonic)); try std.testing.expectEqual(@as(u64, 1_500_000), stats.import_job_duration_sum_us.load(.monotonic));}
fn cleanup(config: Config, id: []const u8) !void { const job_path = try std.fs.path.join(config.allocator, &.{ config.scratch_root, id }); defer config.allocator.free(job_path); try std.Io.Dir.cwd().deleteTree(config.io, job_path); try config.jobs.dropDone(id);}
test "real import pauses after durable patch and resumes after full reopen" { const testing = std.testing; const meta_store = @import("../storage/meta_store.zig"); const segment = @import("../storage/segment.zig"); 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 scratch = try std.fs.path.join(testing.allocator, &.{ root, "timestamp-jobs" }); defer testing.allocator.free(scratch); 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 = "uri,timestamp\n" ++ "at://did:plc:resume/app.bsky.feed.post/one,2024-01-02T03:04:05Z\n", });
const Pause = struct { cancelled: *std.atomic.Value(bool), seen: usize = 0,
fn applied(raw: ?*anyopaque, _: u64) void { const self: *@This() = @ptrCast(@alignCast(raw.?)); self.seen += 1; if (self.seen == 1) self.cancelled.store(true, .release); } }; var cancelled: std.atomic.Value(bool) = .init(false); var pause: Pause = .{ .cancelled = &cancelled };
// First process: create two archive generations, durably patch one, then // stop at the same between-segment cancellation boundary production uses. { var meta = try meta_store.Store.open(testing.allocator, root); defer meta.deinit(); var job_store = jobs_mod.Store.init(testing.allocator, &meta); var rules = try rules_mod.Store.open(testing.allocator, io, root); defer rules.deinit(); var archive = try archive_mod.Archive.init(testing.allocator, io, root); defer archive.deinit(); archive.stamper_ctx = &rules; archive.stamper = rules_mod.Store.stampOpaque; _ = try archive.append(.{ .seq = 0, .witnessed_at = 10, .indexed_at = 0, .kind = .create, .did = "did:plc:resume", .collection = "app.bsky.feed.post", .rkey = "one", .rev = "3lone", .payload = "first", }, 10); try archive.rotate(); _ = try archive.append(.{ .seq = 0, .witnessed_at = 20, .indexed_at = 0, .kind = .update, .did = "did:plc:resume", .collection = "app.bsky.feed.post", .rkey = "one", .rev = "3ltwo", .payload = "second", }, 20); try job_store.begin(.{ .id = "resume-job", .csv_path = csv_path, .submitted_at_us = 1, }); try testing.expectError(error.ImportCancelled, run(.{ .allocator = testing.allocator, .io = io, .scratch_root = scratch, .archive = &archive, .rules = &rules, .jobs = &job_store, .cancelled = &cancelled, .on_segment_applied_ctx = &pause, .on_segment_applied = Pause.applied, }, "resume-job")); var saved = (try job_store.get("resume-job")).?; defer saved.deinit(); try testing.expectEqual(jobs_mod.State.running, saved.value.state); try testing.expect(saved.value.bucketed); try testing.expectEqual(@as(u64, 1), saved.value.segments_applied); var done = try job_store.loadDoneFor("resume-job"); defer done.deinit(testing.allocator); try testing.expectEqual(@as(usize, 1), done.count()); }
// Second process: RocksDB, rule store and archive are all reopened. The // durable bucket flag and done-set skip completed work, then cleanup drops // both the offset directory and checkpoint keys. cancelled.store(false, .release); { var meta = try meta_store.Store.open(testing.allocator, root); defer meta.deinit(); var job_store = jobs_mod.Store.init(testing.allocator, &meta); var rules = try rules_mod.Store.open(testing.allocator, io, root); defer rules.deinit(); var archive = try archive_mod.Archive.init(testing.allocator, io, root); defer archive.deinit(); archive.stamper_ctx = &rules; archive.stamper = rules_mod.Store.stampOpaque; try run(.{ .allocator = testing.allocator, .io = io, .scratch_root = scratch, .archive = &archive, .rules = &rules, .jobs = &job_store, .cancelled = &cancelled, }, "resume-job"); var finished = (try job_store.get("resume-job")).?; defer finished.deinit(); try testing.expectEqual(jobs_mod.State.complete, finished.value.state); try testing.expectEqual(@as(u64, 2), finished.value.segments_applied); try testing.expectEqual(@as(u64, 2), finished.value.rows_mutated); var done = try job_store.loadDoneFor("resume-job"); defer done.deinit(testing.allocator); try testing.expectEqual(@as(usize, 0), done.count()); const finished_scratch = try std.fs.path.join(testing.allocator, &.{ scratch, "resume-job" }); defer testing.allocator.free(finished_scratch); try testing.expectError(error.FileNotFound, Io.Dir.accessAbsolute(io, finished_scratch, .{}));
const expected: i64 = 1_704_164_645_000_000; var seen: usize = 0; var iterator = archive.dir.iterate(); while (try iterator.next(io)) |entry| { const bytes = try archive.dir.readFileAlloc(io, entry.name, testing.allocator, .limited(1 << 20)); defer testing.allocator.free(bytes); var sealed = segment.Sealed.parse(testing.allocator, bytes) catch continue; defer sealed.deinit(testing.allocator); const segment_index = archive_mod.parseSegmentIndex(entry.name).?; const resident = archive.manifest.segmentByIndex(segment_index).?; try testing.expectEqual(sealed.header.checksum, resident.header.checksum); var resident_blocks = try archive.manifest.blockIndex(segment_index); defer resident_blocks.deinit(); try testing.expectEqual(sealed.block_index.len, resident_blocks.entries.len); 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) |event| { try testing.expectEqual(expected, event.indexed_at); seen += 1; } } } try testing.expectEqual(@as(usize, 2), seen); }}