Something went wrong. Try again.
atproto relay in zig zlay.waow.tech
relay zig atproto
Something went wrong. Try again.
Zig
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937const std = @import("std");const rocks = @import("rocksdb");const c = @import("rocksdb_c");const zkv = @import("zkv");const index = @import("internal/collection_index/index.zig");const gpa = std.heap.c_allocator;const Database = zkv.Db(zkv.DirStore);const families = [_]rocks.ColumnFamilyDescription{ .{ .name = "default" }, .{ .name = "rbc" }, .{ .name = "cbr" } };
const Rocks = struct { db: rocks.DB, handles: []const rocks.ColumnFamily, fn open(path: []const u8, read_only: bool) !Rocks { const name = try gpa.dupeSentinel(u8, path, 0); defer gpa.free(name); var err: ?rocks.Data = null; defer if (err) |e| e.deinit(); const db, const handles = try rocks.DB.open(gpa, name, .{ .create_if_missing = !read_only, .create_missing_column_families = !read_only }, &families, read_only, &err); return .{ .db = db, .handles = handles }; } fn view(self: *const Rocks) View { return .{ .db = self.db.db, .handles = self.handles }; } fn close(self: *Rocks) void { self.db.deinit(); gpa.free(self.handles); } fn write(self: *Rocks, batch: rocks.WriteBatch) !void { var err: ?rocks.Data = null; defer if (err) |e| e.deinit(); try self.db.writeWithOptions(batch, .{ .sync = true }, &err); }};
const View = struct { db: *c.rocksdb_t, handles: []const rocks.ColumnFamily, fn iterator(self: View, handle: rocks.ColumnFamilyHandle, direction: rocks.IteratorDirection, start: ?[]const u8) rocks.Iterator { std.debug.assert(direction == .forward and start == null); const options = c.rocksdb_readoptions_create().?; defer c.rocksdb_readoptions_destroy(options); const raw = rocks.RawIterator{ .inner = c.rocksdb_create_iterator_cf(self.db, options, handle).? }; raw.seekToFirst(); return .{ .raw = raw, .direction = .forward, .done = false }; }};
const Secondary = struct { db: *c.rocksdb_t, handles: [3]rocks.ColumnFamily,
fn open(source: []const u8, scratch: []const u8) !Secondary { const source_z = try gpa.dupeSentinel(u8, source, 0); defer gpa.free(source_z); const scratch_z = try gpa.dupeSentinel(u8, scratch, 0); defer gpa.free(scratch_z); const options = c.rocksdb_options_create().?; defer c.rocksdb_options_destroy(options); c.rocksdb_options_set_max_open_files(options, -1); const names = [_][*c]const u8{ "default", "rbc", "cbr" }; const cf_options: [3]?*const c.rocksdb_options_t = @splat(options); var handles: [3]?*c.rocksdb_column_family_handle_t = @splat(null); var message: [*c]u8 = null; defer if (message != null) c.rocksdb_free(message); const db = c.rocksdb_open_as_secondary_column_families(options, source_z, scratch_z, 3, &names, &cf_options, &handles, &message) orelse return error.SecondaryOpen; var result: Secondary = .{ .db = db, .handles = undefined }; for (&result.handles, handles, families) |*slot, handle, family| slot.* = .{ .name = family.name, .handle = handle.? }; return result; }
fn view(self: *const Secondary) View { return .{ .db = self.db, .handles = &self.handles }; }
fn sequence(self: *const Secondary) u64 { return c.rocksdb_get_latest_sequence_number(self.db); }
fn close(self: *Secondary) void { for (self.handles) |family| c.rocksdb_column_family_handle_destroy(family.handle); c.rocksdb_close(self.db); }};
fn liveRehearsal(io: std.Io, source: []const u8, scratch: []const u8, inspect_only: bool) !void { try std.Io.Dir.cwd().createDir(io, scratch, .default_dir); const secondary_dir = try std.fs.path.join(gpa, &.{ scratch, "secondary" }); defer gpa.free(secondary_dir); const target = try std.fs.path.join(gpa, &.{ scratch, "zkv-target" }); defer gpa.free(target); const before = std.Io.Clock.awake.now(io); var secondary = try Secondary.open(source, secondary_dir); defer secondary.close(); const sequence = secondary.sequence(); const opened = std.Io.Clock.awake.now(io); var estimates: [2]u64 = undefined; for (secondary.handles[1..], &estimates) |family, *estimate| { if (c.rocksdb_property_int_cf(secondary.db, family.handle, "rocksdb.estimate-num-keys", estimate) != 0) return error.MissingSourceEstimate; } std.debug.print("live_rehearsal_open sequence={d} estimated_rbc={d} estimated_cbr={d} open_ms={d}\n", .{ sequence, estimates[0], estimates[1], before.durationTo(opened).toMilliseconds() }); if (inspect_only) return; const count = try forwardView(io, secondary.view(), target); const converted = std.Io.Clock.awake.now(io); if (try verifyView(io, secondary.view(), target) != count or secondary.sequence() != sequence) return error.InconsistentSecondary; std.debug.print("live_rehearsal sequence={d} entries={d} open_ms={d} forward_ms={d} verify_ms={d}\n", .{ sequence, count, before.durationTo(opened).toMilliseconds(), opened.durationTo(converted).toMilliseconds(), converted.durationTo(std.Io.Clock.awake.now(io)).toMilliseconds() });}
fn pair(batch: *rocks.WriteBatch, handles: []const rocks.ColumnFamily, i: usize, remove: bool) !void { var did_buf: [64]u8 = undefined; var col_buf: [64]u8 = undefined; var key_buf: [160]u8 = undefined; const did = try std.fmt.bufPrint(&did_buf, "did:plc:offline{d:0>16}", .{i / 2}); const col = try std.fmt.bufPrint(&col_buf, "dev.offline.collection{d:0>6}", .{i % 1024}); for ([_][2][]const u8{ .{ col, did }, .{ did, col } }, 1..) |parts, n| { const key = try std.fmt.bufPrint(&key_buf, "{s}\x00{s}", .{ parts[0], parts[1] }); if (remove) batch.delete(handles[n].handle, key) else batch.put(handles[n].handle, key, ""); }}
fn maintain(db: *Database) !void { if (db.stagedBytes() >= 256 << 10) try db.commit(); if (db.memtableBytes() >= 8 << 20) { try db.flush(); while (try db.compact()) {} try db.collect(); }}
fn maintainDelta(db: *Database) !void { if (db.stagedBytes() >= 256 << 10) try db.commit(); if (db.memtableBytes() >= 8 << 20) try db.flush();}
fn finish(db: *Database) !void { try db.commit(); try db.flush(); while (try db.compact()) {} try db.collect();}
fn seed(path: []const u8, count: usize) !void { var db = try Rocks.open(path, false); defer db.close(); var batch = rocks.WriteBatch.init(); defer batch.deinit(); for (0..count) |i| { try pair(&batch, db.handles, i, false); if (i % 4096 == 4095) { try db.write(batch); batch.deinit(); batch = rocks.WriteBatch.init(); } } try db.write(batch);}
fn forward(io: std.Io, source: []const u8, target: []const u8) !u64 { var old = try Rocks.open(source, true); defer old.close(); return forwardView(io, old.view(), target);}
fn forwardView(io: std.Io, view: View, target: []const u8) !u64 { var dir = try std.Io.Dir.cwd().createDirPathOpen(io, target, .{}); defer dir.close(io); var store = zkv.DirStore.init(gpa, io, dir); defer store.deinit(); const db = try Database.open(gpa, &store, .{ .memtable_bytes = 8 << 20 }); defer db.close(); var count: u64 = 0; var key: std.ArrayList(u8) = .empty; defer key.deinit(gpa); var err: ?rocks.Data = null; defer if (err) |e| e.deinit(); for (view.handles, 0..) |family, n| { var it = view.iterator(family.handle, .forward, null); defer it.deinit(); while (try it.next(&err)) |entry| { if (n == 0) return error.NonEmptyDefaultFamily; if (entry[1].data.len != 0) return error.NonEmptyIndexValue; if (std.mem.indexOfScalar(u8, entry[0].data, 0) == null) return error.InvalidIndexKey; key.clearRetainingCapacity(); try key.append(gpa, @intCast(n)); try key.appendSlice(gpa, entry[0].data); try db.put(key.items, entry[1].data); count += 1; try maintain(db); } } try finish(db); return count;}
fn mutateSource(source: []const u8, count: usize, changes: usize) !void { var old = try Rocks.open(source, false); defer old.close(); var batch = rocks.WriteBatch.init(); defer batch.deinit(); for (0..changes) |i| { try pair(&batch, old.handles, i, true); try pair(&batch, old.handles, count + i, false); } try old.write(batch);}
fn reconcile(io: std.Io, source: []const u8, target: []const u8) !u64 { var old = try Rocks.open(source, true); defer old.close(); return reconcileView(io, old.view(), target);}
fn reconcileView(io: std.Io, view: View, target: []const u8) !u64 { var dir = try std.Io.Dir.cwd().openDir(io, target, .{}); defer dir.close(io); var store = zkv.DirStore.init(gpa, io, dir); defer store.deinit(); const db = try Database.open(gpa, &store, .{ .memtable_bytes = 8 << 20 }); defer db.close(); var changed: u64 = 0; { var snapshot = try db.scan(.{}); defer snapshot.deinit(); var next = try snapshot.next(); var err: ?rocks.Data = null; defer if (err) |e| e.deinit(); var key: std.ArrayList(u8) = .empty; defer key.deinit(gpa); for (view.handles, 0..) |family, n| { var it = view.iterator(family.handle, .forward, null); defer it.deinit(); while (try it.next(&err)) |entry| { if (n == 0) return error.NonEmptyDefaultFamily; key.clearRetainingCapacity(); try key.append(gpa, @intCast(n)); try key.appendSlice(gpa, entry[0].data); while (next) |e| { if (std.mem.order(u8, e.key, key.items) != .lt) break; try db.del(e.key); changed += 1; try maintainDelta(db); next = try snapshot.next(); } if (next != null and std.mem.eql(u8, next.?.key, key.items)) { if (!std.mem.eql(u8, next.?.value, entry[1].data)) { try db.put(key.items, entry[1].data); changed += 1; } next = try snapshot.next(); } else { try db.put(key.items, entry[1].data); changed += 1; } try maintainDelta(db); } } while (next) |e| { try db.del(e.key); changed += 1; try maintainDelta(db); next = try snapshot.next(); } } try db.commit(); try db.flush(); try db.collect(); std.debug.print("reconciliation_state l0_tables={d} unflushed={d}\n", .{ db.version.tables[0].len, db.unflushedSegments() }); return changed;}
fn reverse(io: std.Io, source: []const u8, target: []const u8) !void { var dir = try std.Io.Dir.cwd().openDir(io, source, .{}); defer dir.close(io); var store = zkv.DirStore.init(gpa, io, dir); defer store.deinit(); const db = try Database.openReader(gpa, &store, .{}); defer db.close(); var old = try Rocks.open(target, false); defer old.close(); var batch = rocks.WriteBatch.init(); defer batch.deinit(); var it = try db.scan(.{}); defer it.deinit(); var count: usize = 0; while (try it.next()) |e| { if (e.key.len < 2 or e.key[0] < 1 or e.key[0] > 2) return error.InvalidNamespace; batch.put(old.handles[e.key[0]].handle, e.key[1..], e.value); count += 1; if (count % 4096 == 0) { try old.write(batch); batch.deinit(); batch = rocks.WriteBatch.init(); } } try old.write(batch);}
fn verify(io: std.Io, source: []const u8, target: []const u8) !u64 { var old = try Rocks.open(source, true); defer old.close(); return verifyView(io, old.view(), target);}
fn verifyView(io: std.Io, view: View, target: []const u8) !u64 { var dir = try std.Io.Dir.cwd().openDir(io, target, .{}); defer dir.close(io); var store = zkv.DirStore.init(gpa, io, dir); defer store.deinit(); const db = try Database.openReader(gpa, &store, .{}); defer db.close(); var scan = try db.scan(.{}); defer scan.deinit(); var count: u64 = 0; var err: ?rocks.Data = null; defer if (err) |e| e.deinit(); for (view.handles, 0..) |family, n| { var it = view.iterator(family.handle, .forward, null); defer it.deinit(); while (try it.next(&err)) |e| { if (n == 0) return error.NonEmptyDefaultFamily; const next = (try scan.next()) orelse return error.MissingEntry; if (next.key.len != e[0].data.len + 1 or next.key[0] != n or !std.mem.eql(u8, next.key[1..], e[0].data) or !std.mem.eql(u8, next.value, e[1].data)) return error.MismatchedEntry; count += 1; } } if (try scan.next() != null) return error.ExtraEntry; db.assertIntegrity(); return count;}
const VerifyPart = struct { io: std.Io, view: View, target: []const u8, family: usize, result: anyerror!u64 = error.NotStarted,
fn run(self: *VerifyPart) void { self.result = self.verify(); }
fn verify(self: *const VerifyPart) !u64 { var dir = try std.Io.Dir.cwd().openDir(self.io, self.target, .{}); defer dir.close(self.io); var store = zkv.DirStore.init(gpa, self.io, dir); defer store.deinit(); const db = try Database.openReader(gpa, &store, .{}); defer db.close(); var scan = try db.scan(if (self.family == 1) .{ .end = "\x02" } else .{ .start = "\x02" }); defer scan.deinit(); var it = self.view.iterator(self.view.handles[self.family].handle, .forward, null); defer it.deinit(); var err: ?rocks.Data = null; defer if (err) |e| e.deinit(); var count: u64 = 0; while (try it.next(&err)) |e| { const next = (try scan.next()) orelse return error.MissingEntry; if (next.key.len != e[0].data.len + 1 or next.key[0] != self.family or !std.mem.eql(u8, next.key[1..], e[0].data) or !std.mem.eql(u8, next.value, e[1].data)) return error.MismatchedEntry; count += 1; } if (try scan.next() != null) return error.ExtraEntry; db.assertIntegrity(); return count; }};
fn verifyParallel(io: std.Io, view: View, target: []const u8) !u64 { var err: ?rocks.Data = null; defer if (err) |e| e.deinit(); var default = view.iterator(view.handles[0].handle, .forward, null); defer default.deinit(); if (try default.next(&err) != null) return error.NonEmptyDefaultFamily; var parts = [_]VerifyPart{ .{ .io = io, .view = view, .target = target, .family = 1 }, .{ .io = io, .view = view, .target = target, .family = 2 }, }; const first = try std.Thread.spawn(.{}, VerifyPart.run, .{&parts[0]}); const second = std.Thread.spawn(.{}, VerifyPart.run, .{&parts[1]}) catch |e| { first.join(); return e; }; first.join(); second.join(); return (try parts[0].result) + (try parts[1].result);}
fn verifyParallelCommand(io: std.Io, source: []const u8, target: []const u8) !void { var old = try Rocks.open(source, true); defer old.close(); const before = std.Io.Clock.awake.now(io); const count = try verifyParallel(io, old.view(), target); std.debug.print("verify_parallel entries={d} verify_ms={d}\n", .{ count, before.durationTo(std.Io.Clock.awake.now(io)).toMilliseconds() });}
const Reader = struct { store: zkv.DirStore, db: *Database,
fn open(io: std.Io, path: []const u8) !*Reader { const self = try gpa.create(Reader); errdefer gpa.destroy(self); const dir = try std.Io.Dir.cwd().openDir(io, path, .{}); errdefer dir.close(io); self.store = .init(gpa, io, dir); errdefer self.store.deinit(); self.db = try Database.openReader(gpa, &self.store, .{}); return self; }
fn close(self: *Reader, io: std.Io) void { self.db.close(); self.store.deinit(); self.store.dir.close(io); gpa.destroy(self); }};
const SharedWriter = struct { db: *Database, io: std.Io, mutex: std.Io.Mutex = .init,
fn apply(self: *SharedWriter, key: []const u8, value: ?[]const u8) !void { self.mutex.lockUncancelable(self.io); defer self.mutex.unlock(self.io); if (value) |bytes| try self.db.put(key, bytes) else try self.db.del(key); try maintainDelta(self.db); }};
const DiffPart = struct { view: View, reader: *Database, family: usize, destination: union(enum) { zkv: *SharedWriter, rocks: *Rocks }, result: anyerror!u64 = error.NotStarted, batch: ?rocks.WriteBatch = null, pending: usize = 0,
fn run(self: *DiffPart) void { self.result = self.compare(); }
fn apply(self: *DiffPart, key: []const u8, value: ?[]const u8) !void { switch (self.destination) { .zkv => |writer| try writer.apply(key, value), .rocks => |writer| { if (value) |bytes| self.batch.?.put(writer.handles[self.family].handle, key[1..], bytes) else self.batch.?.delete(writer.handles[self.family].handle, key[1..]); self.pending += 1; if (self.pending >= 4096) { try writer.write(self.batch.?); self.batch.?.deinit(); self.batch = rocks.WriteBatch.init(); self.pending = 0; } }, } }
fn compare(self: *DiffPart) !u64 { if (self.destination == .rocks) self.batch = rocks.WriteBatch.init(); defer if (self.batch) |batch| batch.deinit(); var scan = try self.reader.scan(if (self.family == 1) .{ .end = "\x02" } else .{ .start = "\x02" }); defer scan.deinit(); var next = try scan.next(); var it = self.view.iterator(self.view.handles[self.family].handle, .forward, null); defer it.deinit(); var err: ?rocks.Data = null; defer if (err) |e| e.deinit(); var entry = try it.next(&err); var key: std.ArrayList(u8) = .empty; defer key.deinit(gpa); var changed: u64 = 0; while (entry != null or next != null) { if (next) |e| if (e.key.len < 2 or e.key[0] != self.family) return error.InvalidNamespace; key.clearRetainingCapacity(); if (entry) |e| { try key.append(gpa, @intCast(self.family)); try key.appendSlice(gpa, e[0].data); } const order = if (entry == null) std.math.Order.gt else if (next == null) std.math.Order.lt else std.mem.order(u8, key.items, next.?.key); if (order == .lt) { try self.apply(key.items, if (self.destination == .zkv) entry.?[1].data else null); changed += 1; entry = try it.next(&err); } else if (order == .gt) { try self.apply(next.?.key, if (self.destination == .rocks) next.?.value else null); changed += 1; next = try scan.next(); } else { if (!std.mem.eql(u8, entry.?[1].data, next.?.value)) { try self.apply(key.items, if (self.destination == .zkv) entry.?[1].data else next.?.value); changed += 1; } entry = try it.next(&err); next = try scan.next(); } } if (self.destination == .rocks and self.pending > 0) try self.destination.rocks.write(self.batch.?); return changed; }};
fn compareParallel(io: std.Io, view: View, target: []const u8, destination: @FieldType(DiffPart, "destination")) !u64 { var err: ?rocks.Data = null; defer if (err) |e| e.deinit(); var default = view.iterator(view.handles[0].handle, .forward, null); defer default.deinit(); if (try default.next(&err) != null) return error.NonEmptyDefaultFamily; const first_reader = try Reader.open(io, target); defer first_reader.close(io); const second_reader = try Reader.open(io, target); defer second_reader.close(io); if (first_reader.db.etag != second_reader.db.etag or first_reader.db.seq != second_reader.db.seq) return error.SourceChanged; if (destination == .zkv and (first_reader.db.etag != destination.zkv.db.etag or first_reader.db.seq != destination.zkv.db.seq)) return error.SourceChanged; var parts = [_]DiffPart{ .{ .view = view, .reader = first_reader.db, .family = 1, .destination = destination }, .{ .view = view, .reader = second_reader.db, .family = 2, .destination = destination }, }; const first = try std.Thread.spawn(.{}, DiffPart.run, .{&parts[0]}); const second = std.Thread.spawn(.{}, DiffPart.run, .{&parts[1]}) catch |e| { first.join(); return e; }; first.join(); second.join(); return (try parts[0].result) + (try parts[1].result);}
fn finalizeParallel(io: std.Io, source: []const u8, target: []const u8) !void { var old = try Rocks.open(source, true); defer old.close(); return finalizeParallelView(io, old.view(), target);}
fn finalizeParallelView(io: std.Io, view: View, target: []const u8) !void { const before = std.Io.Clock.awake.now(io); var changed: u64 = undefined; { var dir = try std.Io.Dir.cwd().openDir(io, target, .{}); defer dir.close(io); var store = zkv.DirStore.init(gpa, io, dir); defer store.deinit(); const db = try Database.open(gpa, &store, .{ .memtable_bytes = 8 << 20 }); defer db.close(); var writer: SharedWriter = .{ .db = db, .io = io }; changed = try compareParallel(io, view, target, .{ .zkv = &writer }); try db.commit(); try db.flush(); try db.collect(); std.debug.print("reconciliation_state l0_tables={d} unflushed={d}\n", .{ db.version.tables[0].len, db.unflushedSegments() }); } const reconciled = std.Io.Clock.awake.now(io); const entries = try verifyParallel(io, view, target); std.debug.print("finalize_parallel entries={d} changes={d} reconcile_ms={d} verify_ms={d}\n", .{ entries, changed, before.durationTo(reconciled).toMilliseconds(), reconciled.durationTo(std.Io.Clock.awake.now(io)).toMilliseconds() });}
fn rollbackParallel(io: std.Io, source: []const u8, target: []const u8) !void { const before = std.Io.Clock.awake.now(io); var dir = try std.Io.Dir.cwd().openDir(io, target, .{}); defer dir.close(io); try dir.access(io, "CURRENT", .{}); var changed: u64 = undefined; { var old = try Rocks.open(target, false); defer old.close(); changed = try compareParallel(io, old.view(), source, .{ .rocks = &old }); } const reconciled = std.Io.Clock.awake.now(io); var old = try Rocks.open(target, true); defer old.close(); const entries = try verifyParallel(io, old.view(), source); std.debug.print("rollback_parallel entries={d} changes={d} reconcile_ms={d} verify_ms={d}\n", .{ entries, changed, before.durationTo(reconciled).toMilliseconds(), reconciled.durationTo(std.Io.Clock.awake.now(io)).toMilliseconds() });}
fn rehearseParallel(io: std.Io, source: []const u8, root: []const u8, rollback: []const u8) !void { const scratch = try std.fs.path.join(gpa, &.{ root, "secondary-parallel" }); defer gpa.free(scratch); try std.Io.Dir.cwd().createDir(io, scratch, .default_dir); const target = try std.fs.path.join(gpa, &.{ root, "zkv-target" }); defer gpa.free(target); var secondary = try Secondary.open(source, scratch); defer secondary.close(); const sequence = secondary.sequence(); std.debug.print("parallel_rehearsal_open sequence={d}\n", .{sequence}); try finalizeParallelView(io, secondary.view(), target); try rollbackParallel(io, target, rollback); if (secondary.sequence() != sequence) return error.InconsistentSecondary; std.debug.print("parallel_rehearsal_done sequence={d}\n", .{sequence});}
fn deltaRehearsal(io: std.Io, source: []const u8, root: []const u8) !void { const target = try std.fs.path.join(gpa, &.{ root, "zkv-target" }); defer gpa.free(target); const scratch = try std.fs.path.join(gpa, &.{ root, "secondary-delta" }); defer gpa.free(scratch); const rollback = try std.fs.path.join(gpa, &.{ root, "rocks-rollback" }); defer gpa.free(rollback); try std.Io.Dir.cwd().createDir(io, rollback, .default_dir); var secondary = try Secondary.open(source, scratch); defer secondary.close(); const sequence = secondary.sequence(); const before = std.Io.Clock.awake.now(io); std.debug.print("delta_rehearsal_open sequence={d}\n", .{sequence}); const changed = try reconcileView(io, secondary.view(), target); const reconciled = std.Io.Clock.awake.now(io); const entries = try verifyView(io, secondary.view(), target); const verified = std.Io.Clock.awake.now(io); std.debug.print("delta_rehearsal_verified sequence={d} entries={d} changes={d} reconcile_ms={d} verify_ms={d}\n", .{ sequence, entries, changed, before.durationTo(reconciled).toMilliseconds(), reconciled.durationTo(verified).toMilliseconds() }); try reverse(io, target, rollback); const reversed = std.Io.Clock.awake.now(io); if (try verify(io, rollback, target) != entries or secondary.sequence() != sequence) return error.IncorrectRollback; std.debug.print("delta_rehearsal_done entries={d} reverse_ms={d} reverse_verify_ms={d}\n", .{ entries, verified.durationTo(reversed).toMilliseconds(), reversed.durationTo(std.Io.Clock.awake.now(io)).toMilliseconds() });}
fn finalize(io: std.Io, source: []const u8, target: []const u8) !void { var old = try Rocks.open(source, true); defer old.close(); const before = std.Io.Clock.awake.now(io); const changed = try reconcileView(io, old.view(), target); const reconciled = std.Io.Clock.awake.now(io); const entries = try verifyView(io, old.view(), target); std.debug.print("finalize entries={d} changes={d} reconcile_ms={d} verify_ms={d}\n", .{ entries, changed, before.durationTo(reconciled).toMilliseconds(), reconciled.durationTo(std.Io.Clock.awake.now(io)).toMilliseconds() });}
fn rollbackVerified(io: std.Io, source: []const u8, target: []const u8) !void { const before = std.Io.Clock.awake.now(io); var changed: u64 = 0; var written: u64 = 0; var target_dir = try std.Io.Dir.cwd().openDir(io, target, .{}); defer target_dir.close(io); try target_dir.access(io, "CURRENT", .{}); { var dir = try std.Io.Dir.cwd().openDir(io, source, .{}); defer dir.close(io); var store = zkv.DirStore.init(gpa, io, dir); defer store.deinit(); const db = try Database.openReader(gpa, &store, .{}); defer db.close(); var scan = try db.scan(.{}); defer scan.deinit(); var next = try scan.next(); if (next) |e| if (e.key.len < 2 or e.key[0] < 1 or e.key[0] > 2) return error.InvalidNamespace; var old = try Rocks.open(target, false); defer old.close(); var batch = rocks.WriteBatch.init(); defer batch.deinit(); var err: ?rocks.Data = null; defer if (err) |e| e.deinit(); for (old.handles, 0..) |family, n| { var it = old.view().iterator(family.handle, .forward, null); defer it.deinit(); var entry = try it.next(&err); if (n == 0) { if (entry != null) return error.NonEmptyDefaultFamily; continue; } while (entry != null or (next != null and next.?.key[0] == n)) { const fresh = if (next != null and next.?.key[0] == n) next else null; const order = if (entry == null) std.math.Order.gt else if (fresh == null) std.math.Order.lt else std.mem.order(u8, entry.?[0].data, fresh.?.key[1..]); if (order == .lt) { batch.delete(family.handle, entry.?[0].data); changed += 1; entry = try it.next(&err); } else { if (order == .gt or !std.mem.eql(u8, entry.?[1].data, fresh.?.value)) { batch.put(family.handle, fresh.?.key[1..], fresh.?.value); changed += 1; } if (order == .eq) entry = try it.next(&err); next = try scan.next(); if (next) |e| if (e.key.len < 2 or e.key[0] < 1 or e.key[0] > 2) return error.InvalidNamespace; } if (changed - written >= 4096) { try old.write(batch); batch.deinit(); batch = rocks.WriteBatch.init(); written = changed; } } } if (next != null) return error.InvalidNamespace; try old.write(batch); } const reconciled = std.Io.Clock.awake.now(io); const entries = try verify(io, target, source); std.debug.print("rollback_verified entries={d} changes={d} reconcile_ms={d} verify_ms={d}\n", .{ entries, changed, before.durationTo(reconciled).toMilliseconds(), reconciled.durationTo(std.Io.Clock.awake.now(io)).toMilliseconds() });}
fn reverseVerified(io: std.Io, source: []const u8, target: []const u8) !void { try std.Io.Dir.cwd().createDir(io, target, .default_dir); const before = std.Io.Clock.awake.now(io); try reverse(io, source, target); const reversed = std.Io.Clock.awake.now(io); const entries = try verify(io, target, source); std.debug.print("reverse_verified entries={d} reverse_ms={d} verify_ms={d}\n", .{ entries, before.durationTo(reversed).toMilliseconds(), reversed.durationTo(std.Io.Clock.awake.now(io)).toMilliseconds() });}
pub fn main(init: std.process.Init) !void { const args = try init.minimal.args.toSlice(init.arena.allocator()); if (args.len == 5 and std.mem.eql(u8, args[1], "rehearse-parallel")) return rehearseParallel(init.io, args[2], args[3], args[4]); if (args.len == 4 and (std.mem.eql(u8, args[1], "live-rehearsal") or std.mem.eql(u8, args[1], "inspect-source"))) return liveRehearsal(init.io, args[2], args[3], std.mem.eql(u8, args[1], "inspect-source")); if (args.len == 4 and std.mem.eql(u8, args[1], "delta-rehearsal")) return deltaRehearsal(init.io, args[2], args[3]); if (args.len == 4 and std.mem.eql(u8, args[1], "rollback-verified")) return rollbackVerified(init.io, args[2], args[3]); if (args.len == 4 and std.mem.eql(u8, args[1], "verify-parallel")) return verifyParallelCommand(init.io, args[2], args[3]); if (args.len == 4 and std.mem.eql(u8, args[1], "finalize-parallel")) return finalizeParallel(init.io, args[2], args[3]); if (args.len == 4 and std.mem.eql(u8, args[1], "rollback-parallel")) return rollbackParallel(init.io, args[2], args[3]); if (args.len == 4 and std.mem.eql(u8, args[1], "finalize")) return finalize(init.io, args[2], args[3]); if (args.len == 4 and std.mem.eql(u8, args[1], "reverse-verified")) return reverseVerified(init.io, args[2], args[3]); if (args.len != 3) return error.ExpectedNewRehearsalDirectoryAndPairCount; const count = try std.fmt.parseInt(usize, args[2], 10); if (count == 0 or count > 1000000) return error.InvalidRehearsalSize; try std.Io.Dir.cwd().createDir(init.io, args[1], .default_dir); const source = try std.fs.path.join(gpa, &.{ args[1], "rocks-source" }); defer gpa.free(source); const target = try std.fs.path.join(gpa, &.{ args[1], "zkv-target" }); defer gpa.free(target); const rollback = try std.fs.path.join(gpa, &.{ args[1], "rocks-roundtrip" }); defer gpa.free(rollback); try seed(source, count); const start = std.Io.Clock.awake.now(init.io); const entries = try forward(init.io, source, target); const converted = std.Io.Clock.awake.now(init.io); if (entries != 2 * count or try verify(init.io, source, target) != entries) return error.IncorrectCount; const checked = std.Io.Clock.awake.now(init.io); const changes: usize = @min(count / 10, 1000); try mutateSource(source, count, changes); const before_delta = std.Io.Clock.awake.now(init.io); if (try reconcile(init.io, source, target) != 4 * changes) return error.IncorrectDelta; const reconciled = std.Io.Clock.awake.now(init.io); if (try verify(init.io, source, target) != entries) return error.IncorrectReconciledCount; const delta_checked = std.Io.Clock.awake.now(init.io); { var ci = try index.CollectionIndex.open(gpa, init.io, target); defer ci.deinit(); try ci.addCollection("did:plc:postcutover", "dev.offline.after"); var buffer: [1024]u8 = undefined; var offsets: [10]index.Span = undefined; const page = try ci.listReposByCollection("dev.offline.after", 10, null, &buffer, &offsets); if (page.count != 1 or !std.mem.eql(u8, page.getDid(0), "did:plc:postcutover") or !ci.hasCollections("did:plc:postcutover")) return error.BadPostCutoverMembership; } const before_reverse = std.Io.Clock.awake.now(init.io); try reverse(init.io, target, rollback); const reversed = std.Io.Clock.awake.now(init.io); if (try verify(init.io, rollback, target) != entries + 2) return error.IncorrectRollback; std.debug.print("migration pairs={d} entries={d} forward_ms={d} verify_ms={d} delta_ops={d} reconcile_ms={d} delta_verify_ms={d} reverse_ms={d} reverse_verify_ms={d}\n", .{ count, entries, start.durationTo(converted).toMilliseconds(), converted.durationTo(checked).toMilliseconds(), 4 * changes, before_delta.durationTo(reconciled).toMilliseconds(), reconciled.durationTo(delta_checked).toMilliseconds(), before_reverse.durationTo(reversed).toMilliseconds(), reversed.durationTo(std.Io.Clock.awake.now(init.io)).toMilliseconds() });}
test "migration verification detects changed values and extra entries after exact conversion" { const t = std.testing; var tmp = t.tmpDir(.{}); defer tmp.cleanup(); const base = try std.fmt.allocPrint(gpa, ".zig-cache/tmp/{s}", .{@as([]const u8, &tmp.sub_path)}); defer gpa.free(base); const source = try std.fs.path.join(gpa, &.{ base, "source" }); defer gpa.free(source); const target = try std.fs.path.join(gpa, &.{ base, "target" }); defer gpa.free(target); try seed(source, 32); try t.expectEqual(@as(u64, 64), try forward(t.io, source, target)); try t.expectEqual(@as(u64, 64), try verify(t.io, source, target)); var dir = try std.Io.Dir.cwd().openDir(t.io, target, .{}); defer dir.close(t.io); var store = zkv.DirStore.init(gpa, t.io, dir); defer store.deinit(); const db = try Database.open(gpa, &store, .{}); defer db.close(); var scan = try db.scan(.{}); const key = try gpa.dupe(u8, (try scan.next()).?.key); defer gpa.free(key); scan.deinit(); try db.put(key, "unexpected"); try db.commit(); try t.expectError(error.MismatchedEntry, verify(t.io, source, target)); try db.put(key, ""); try db.put("\x03extra", ""); try db.commit(); try t.expectError(error.ExtraEntry, verify(t.io, source, target));}
test "secondary stays frozen while primary writes and compacts both column families" { const t = std.testing; var tmp = t.tmpDir(.{}); defer tmp.cleanup(); const base = try std.fmt.allocPrint(gpa, ".zig-cache/tmp/{s}", .{@as([]const u8, &tmp.sub_path)}); defer gpa.free(base); const source = try std.fs.path.join(gpa, &.{ base, "source" }); defer gpa.free(source); const scratch = try std.fs.path.join(gpa, &.{ base, "secondary" }); defer gpa.free(scratch); const target = try std.fs.path.join(gpa, &.{ base, "target" }); defer gpa.free(target); try seed(source, 128); var primary = try Rocks.open(source, false); defer primary.close(); var secondary = try Secondary.open(source, scratch); defer secondary.close(); const sequence = secondary.sequence(); try t.expectEqual(@as(u64, 256), try forwardView(t.io, secondary.view(), target)); var batch = rocks.WriteBatch.init(); defer batch.deinit(); for (0..64) |i| { try pair(&batch, primary.handles, i, true); try pair(&batch, primary.handles, 128 + i, false); } try primary.write(batch); var err: ?rocks.Data = null; defer if (err) |e| e.deinit(); for (primary.handles) |family| { try primary.db.flush(family.handle, &err); c.rocksdb_compact_range_cf(primary.db.db, family.handle, null, 0, null, 0); } try t.expectEqual(sequence, secondary.sequence()); try t.expectEqual(@as(u64, 256), try verifyView(t.io, secondary.view(), target)); try t.expectError(error.MismatchedEntry, verifyView(t.io, primary.view(), target));}
test "reconciliation flushes durable deltas without forcing an eligible compaction" { const t = std.testing; var tmp = t.tmpDir(.{}); defer tmp.cleanup(); const base = try std.fmt.allocPrint(gpa, ".zig-cache/tmp/{s}", .{@as([]const u8, &tmp.sub_path)}); defer gpa.free(base); const source = try std.fs.path.join(gpa, &.{ base, "source" }); defer gpa.free(source); const target = try std.fs.path.join(gpa, &.{ base, "target" }); defer gpa.free(target); try seed(source, 32); _ = try forward(t.io, source, target); for (0..4) |i| { try mutateSource(source, 32 + i * 4, 4); _ = try reconcile(t.io, source, target); _ = try verify(t.io, source, target); } var dir = try std.Io.Dir.cwd().openDir(t.io, target, .{}); defer dir.close(t.io); var store = zkv.DirStore.init(gpa, t.io, dir); defer store.deinit(); const db = try Database.open(gpa, &store, .{}); defer db.close(); try t.expectEqual(@as(usize, 5), db.version.tables[0].len); try t.expectEqual(@as(usize, 0), db.memtableBytes());}
test "cutover commands reconcile turnover and preserve later writes through reverse conversion" { const t = std.testing; var tmp = t.tmpDir(.{}); defer tmp.cleanup(); const base = try std.fmt.allocPrint(gpa, ".zig-cache/tmp/{s}", .{@as([]const u8, &tmp.sub_path)}); defer gpa.free(base); const source = try std.fs.path.join(gpa, &.{ base, "source" }); defer gpa.free(source); const target = try std.fs.path.join(gpa, &.{ base, "target" }); defer gpa.free(target); const rollback = try std.fs.path.join(gpa, &.{ base, "rollback" }); defer gpa.free(rollback); try seed(source, 32); _ = try forward(t.io, source, target); try mutateSource(source, 32, 8); try finalize(t.io, source, target); { var ci = try index.CollectionIndex.open(gpa, t.io, target); defer ci.deinit(); try ci.addCollection("did:plc:aftercutover", "dev.offline.new"); } try reverseVerified(t.io, target, rollback); try t.expectEqual(@as(u64, 66), try verify(t.io, rollback, target)); try t.expectError(error.PathAlreadyExists, reverseVerified(t.io, target, rollback)); try mutateSource(source, 100, 4); try rollbackVerified(t.io, target, source); try t.expectEqual(@as(u64, 66), try verify(t.io, source, target));}
test "parallel verification compares both namespaces and rejects keys outside them" { const t = std.testing; var tmp = t.tmpDir(.{}); defer tmp.cleanup(); const base = try std.fmt.allocPrint(gpa, ".zig-cache/tmp/{s}", .{@as([]const u8, &tmp.sub_path)}); defer gpa.free(base); const source = try std.fs.path.join(gpa, &.{ base, "source" }); defer gpa.free(source); const target = try std.fs.path.join(gpa, &.{ base, "target" }); defer gpa.free(target); try seed(source, 32); _ = try forward(t.io, source, target); try verifyParallelCommand(t.io, source, target); var dir = try std.Io.Dir.cwd().openDir(t.io, target, .{}); defer dir.close(t.io); var store = zkv.DirStore.init(gpa, t.io, dir); defer store.deinit(); const db = try Database.open(gpa, &store, .{}); defer db.close(); for ([_][]const u8{ "\x01", "\x02" }) |prefix| { var scan = try db.scan(.{ .start = prefix }); const key = try gpa.dupe(u8, (try scan.next()).?.key); defer gpa.free(key); scan.deinit(); try db.put(key, "changed"); try db.commit(); try t.expectError(error.MismatchedEntry, verifyParallelCommand(t.io, source, target)); try db.put(key, ""); try db.commit(); } for ([_][]const u8{ "\x00invalid", "\x03invalid" }) |key| { try db.put(key, ""); try db.commit(); try t.expectError(if (key[0] == 0) error.MismatchedEntry else error.ExtraEntry, verifyParallelCommand(t.io, source, target)); try db.del(key); try db.commit(); }}
test "parallel migration reconciles both directions against independent serial verification" { const t = std.testing; var tmp = t.tmpDir(.{}); defer tmp.cleanup(); const base = try std.fmt.allocPrint(gpa, ".zig-cache/tmp/{s}", .{@as([]const u8, &tmp.sub_path)}); defer gpa.free(base); const source = try std.fs.path.join(gpa, &.{ base, "source" }); defer gpa.free(source); const target = try std.fs.path.join(gpa, &.{ base, "target" }); defer gpa.free(target); try seed(source, 10000); _ = try forward(t.io, source, target); try mutateSource(source, 10000, 5000); try finalizeParallel(t.io, source, target); try t.expectEqual(@as(u64, 20000), try verify(t.io, source, target)); { var ci = try index.CollectionIndex.open(gpa, t.io, target); defer ci.deinit(); try ci.removeAll("did:plc:offline0000000000002500"); try ci.addCollection("did:plc:postcutover", "dev.offline.new"); } try mutateSource(source, 20000, 5000); try rollbackParallel(t.io, target, source); try t.expectEqual(@as(u64, 19998), try verify(t.io, source, target));}