Something went wrong. Try again.
jetstream v2 in zig stream.waow.tech
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297//! Single-at-a-time operator lifecycle for timestamp imports.//!//! The durable current-job record is the cross-process authority. Submit//! confines the path, atomically creates the record/current pointer, and runs//! the real importer in a background thread. Startup adopts a nonterminal//! current job; shutdown cancels and joins it so the next process can resume.
const std = @import("std");const import_path = @import("import_path.zig");const jobs_mod = @import("jobs.zig");const runner = @import("runner.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, import_dir: []const u8, scratch_root: []const u8, jobs: *jobs_mod.Store, runner: runner.Config, ready: ?*const std.atomic.Value(bool) = null,};
pub const Manager = struct { allocator: Allocator, io: Io, root: import_path.Root, scratch_root: []u8, jobs: *jobs_mod.Store, ready: ?*const std.atomic.Value(bool), runner_config: runner.Config, mutex: Io.Mutex = .init, thread: ?std.Thread = null, active: bool = false, active_id: ?[]u8 = null, cancelled: std.atomic.Value(bool) = .init(false),
pub fn init(config: Config) !Manager { const root = try import_path.Root.open(config.allocator, config.io, config.import_dir); errdefer { var owned = root; owned.deinit(); } try Io.Dir.cwd().createDirPath(config.io, config.scratch_root); return .{ .allocator = config.allocator, .io = config.io, .root = root, .scratch_root = try config.allocator.dupe(u8, config.scratch_root), .jobs = config.jobs, .ready = config.ready, .runner_config = config.runner, }; }
pub fn deinit(self: *Manager) void { self.shutdown(); if (self.active_id) |id| self.allocator.free(id); self.allocator.free(self.scratch_root); self.root.deinit(); self.* = undefined; }
/// Returns an owned job id immediately after its durable admission. pub fn submit(self: *Manager, requested_path: []const u8) ![]u8 { const csv_path = try self.root.resolve(requested_path); defer self.allocator.free(csv_path); self.mutex.lockUncancelable(self.io); defer self.mutex.unlock(self.io); self.reapLocked(); try self.ensureIdleLocked(); if (self.ready) |ready| if (!ready.load(.acquire)) return error.ImportNotReady;
const id = try newJobId(self.allocator, self.io); defer self.allocator.free(id); try self.jobs.begin(.{ .id = id, .csv_path = csv_path, .submitted_at_us = Io.Timestamp.now(self.io, .real).toMicroseconds(), }); self.startLocked(id) catch |err| { var parsed = (try self.jobs.get(id)).?; defer parsed.deinit(); var failed = parsed.value; failed.state = .failed; failed.error_message = @errorName(err); failed.finished_at_us = Io.Timestamp.now(self.io, .real).toMicroseconds(); try self.jobs.put(failed); return err; }; return try self.allocator.dupe(u8, id); }
/// Adopt the persisted current job when it is nonterminal. pub fn resumeIncomplete(self: *Manager) !bool { self.mutex.lockUncancelable(self.io); defer self.mutex.unlock(self.io); self.reapLocked(); if (self.active) return false; const id = try self.jobs.current() orelse return false; defer self.allocator.free(id); var parsed = (try self.jobs.get(id)) orelse return error.ImportCurrentRecordMissing; defer parsed.deinit(); if (parsed.value.state.terminal()) return false; try self.startLocked(id); return true; }
pub fn status(self: *Manager, id: []const u8) !jobs_mod.ParsedRecord { return (try self.jobs.get(id)) orelse error.ImportJobNotFound; }
pub fn current(self: *Manager) !?jobs_mod.ParsedRecord { const id = try self.jobs.current() orelse return null; defer self.allocator.free(id); return (try self.jobs.get(id)) orelse error.ImportCurrentRecordMissing; }
pub fn wait(self: *Manager) void { self.mutex.lockUncancelable(self.io); const thread = self.thread; self.thread = null; self.mutex.unlock(self.io); if (thread) |item| item.join(); self.mutex.lockUncancelable(self.io); self.active = false; if (self.active_id) |id| self.allocator.free(id); self.active_id = null; self.mutex.unlock(self.io); }
pub fn shutdown(self: *Manager) void { self.cancelled.store(true, .release); self.wait(); }
pub fn setTraceContext(self: *Manager, trace_context: observability.Context) void { self.runner_config.trace_context = trace_context; }
fn ensureIdleLocked(self: *Manager) !void { if (self.active) return error.ImportJobInProgress; const current_id = try self.jobs.current() orelse return; defer self.allocator.free(current_id); var current_record = (try self.jobs.get(current_id)) orelse return error.ImportCurrentRecordMissing; defer current_record.deinit(); if (!current_record.value.state.terminal()) return error.ImportJobInProgress; }
fn startLocked(self: *Manager, id: []const u8) !void { if (self.active or self.thread != null) return error.ImportJobInProgress; self.cancelled.store(false, .release); self.active_id = try self.allocator.dupe(u8, id); errdefer { self.allocator.free(self.active_id.?); self.active_id = null; } self.active = true; errdefer self.active = false; self.thread = try std.Thread.spawn(.{}, worker, .{self}); }
fn worker(self: *Manager) void { var config = self.runner_config; config.scratch_root = self.scratch_root; config.cancelled = &self.cancelled; runner.run(config, self.active_id.?) catch |err| { log.err("timestamp import {s} stopped: {s}", .{ self.active_id.?, @errorName(err) }); }; self.mutex.lockUncancelable(self.io); self.active = false; self.mutex.unlock(self.io); }
fn reapLocked(self: *Manager) void { if (self.active or self.thread == null) return; self.thread.?.join(); self.thread = null; if (self.active_id) |id| self.allocator.free(id); self.active_id = null; }};
fn newJobId(allocator: Allocator, io: Io) ![]u8 { var random: [8]u8 = undefined; try io.randomSecure(&random); return std.fmt.allocPrint(allocator, "{d}-{x}", .{ Io.Timestamp.now(io, .real).toMicroseconds(), random, });}
test "manager submits real work and adopts a durable incomplete job" { const testing = std.testing; const archive_mod = @import("../storage/archive.zig"); const meta_store = @import("../storage/meta_store.zig"); const rules_mod = @import("rules.zig"); const metrics = @import("../runtime/metrics.zig"); var threaded: Io.Threaded = .init(testing.allocator, .{}); defer threaded.deinit(); const io = threaded.io(); var tmp = testing.tmpDir(.{}); defer tmp.cleanup(); try tmp.dir.createDirPath(io, "imports"); try tmp.dir.writeFile(io, .{ .sub_path = "imports/job.csv", .data = "uri,timestamp\n" ++ "at://did:plc:manager/app.bsky.feed.post/one,2024-01-02T03:04:05Z\n", }); 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 imports = try std.fs.path.join(testing.allocator, &.{ root, "imports" }); defer testing.allocator.free(imports); 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, &.{ imports, "job.csv" }); defer testing.allocator.free(csv_path);
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; var metric_stats: metrics.Stats = .{}; _ = try archive.append(.{ .seq = 0, .witnessed_at = 1, .indexed_at = 0, .kind = .create, .did = "did:plc:manager", .collection = "app.bsky.feed.post", .rkey = "one", .rev = "3lmanager", .payload = "payload", }, 1);
var manager = try Manager.init(.{ .allocator = testing.allocator, .io = io, .import_dir = imports, .scratch_root = scratch, .jobs = &job_store, .runner = .{ .allocator = testing.allocator, .io = io, .scratch_root = scratch, .archive = &archive, .rules = &rules, .jobs = &job_store, .stats = &metric_stats, }, }); defer manager.deinit();
const id = try manager.submit("job.csv"); defer testing.allocator.free(id); manager.wait(); var completed = try manager.status(id); defer completed.deinit(); try testing.expectEqual(jobs_mod.State.complete, completed.value.state); try testing.expectEqual(@as(u64, 1), completed.value.rows_mutated); var current = (try manager.current()).?; defer current.deinit(); try testing.expectEqualStrings(id, current.value.id);
// Model a hard process stop after admission: only the durable record and // current pointer exist. It blocks a new submit and startup adopts it. try job_store.begin(.{ .id = "adopt-me", .csv_path = csv_path, .submitted_at_us = 2, }); try testing.expectError(error.ImportJobInProgress, manager.submit("job.csv")); try testing.expect(try manager.resumeIncomplete()); manager.wait(); var adopted = try manager.status("adopt-me"); defer adopted.deinit(); try testing.expectEqual(jobs_mod.State.complete, adopted.value.state); try testing.expect(!(try manager.resumeIncomplete())); try testing.expectEqual(@as(u64, 2), metric_stats.import_jobs_ok_total.load(.monotonic)); try testing.expectEqual(@as(u64, 0), metric_stats.import_jobs_error_total.load(.monotonic)); try testing.expectEqual(@as(u64, 0), metric_stats.import_phase.load(.monotonic)); try testing.expectEqual(@as(u64, 2), metric_stats.import_rows_parsed_total.load(.monotonic)); try testing.expectEqual(@as(u64, 1), metric_stats.import_rows_mutated_total.load(.monotonic)); try testing.expectEqual(@as(u64, 2), metric_stats.import_segments_examined_total.load(.monotonic)); try testing.expect(metric_stats.import_bytes_rewritten_total.load(.monotonic) > 0);}