From 9041e16f3c263dd04bf5adb2e099d97b109d2f65 Mon Sep 17 00:00:00 2001 From: zzstoatzz Date: Thu, 30 Jul 2026 15:31:22 -0500 Subject: [PATCH] Isolate repo exports from primary storage --- bench/README.md | 4 +- bench/main.zig | 109 +++++++++++++++++++++ docs/getrepo-notes.md | 22 +++++ src/storage/store.zig | 217 ++++++++++++++++++++++++++++++++++++++---- 4 files changed, 330 insertions(+), 22 deletions(-) diff --git a/bench/README.md b/bench/README.md index 7795124..2ff1c89 100644 --- a/bench/README.md +++ b/bench/README.md @@ -57,7 +57,9 @@ just bench run --scenario write --records 10000 runs. - `get-repo`: focused `com.atproto.sync.getRepo` storage-pressure probe. It measures full repo CAR construction, `since` diff CAR construction after one - additional commit, and concurrent full exports against one seeded repo. + additional commit, and concurrent full exports against one seeded repo. Its + isolation lane runs account lookups and writes during those exports so a + throughput improvement cannot hide resident-facing database stalls. - `space`: experimental permissioned-data storage probes for space discovery, private record writes, private record reads/lists, repo oplog catch-up, and blob readback. diff --git a/bench/main.zig b/bench/main.zig index 0bded7e..db58574 100644 --- a/bench/main.zig +++ b/bench/main.zig @@ -566,6 +566,7 @@ fn benchGetRepo(allocator: std.mem.Allocator, options: Options) !void { (try benchRepoCarForRepo(allocator, state.account, options.records + 2)).print(); (try benchRepoCarSince(allocator, state.account, since_rev)).print(); (try benchConcurrentRepoCar(allocator, state.account, level)).print(); + try benchRepoExportIsolation(allocator, state.account, options.records + 2, level); } fn benchRepoCarSince( @@ -615,6 +616,57 @@ fn benchConcurrentRepoCar( return concurrentResult("getRepo full", level, total_ops, nowNs() - start, latencies); } +fn benchRepoExportIsolation( + allocator: std.mem.Allocator, + account: zds.auth.tokens.Account, + first_write_index: usize, + level: ConcurrencyLevel, +) !void { + const export_ops = level.callers * level.ops_per_caller; + const probe_ops = @max(export_ops * 10, 20); + const write_ops = @max(export_ops, 4); + const export_latencies = try allocator.alloc(u64, export_ops); + defer allocator.free(export_latencies); + const probe_latencies = try allocator.alloc(u64, probe_ops); + defer allocator.free(probe_latencies); + const write_latencies = try allocator.alloc(u64, write_ops); + defer allocator.free(write_latencies); + const export_threads = try allocator.alloc(std.Thread, level.callers); + defer allocator.free(export_threads); + var start: std.atomic.Value(bool) = .init(false); + + for (export_threads, 0..) |*thread, i| { + thread.* = try std.Thread.spawn(.{}, repoCarWorkerStarting, .{ + account.did, + export_latencies[i * level.ops_per_caller ..][0..level.ops_per_caller], + &start, + }); + } + const probe_thread = try std.Thread.spawn(.{}, accountProbeWorker, .{ + account.did, + probe_latencies, + &start, + }); + const write_thread = try std.Thread.spawn(.{}, repoWriteWorker, .{ + account, + first_write_index, + write_latencies, + &start, + }); + + const started = nowNs(); + start.store(true, .release); + for (export_threads) |thread| thread.join(); + probe_thread.join(); + write_thread.join(); + const elapsed = nowNs() - started; + + std.debug.print("\n=== getRepo isolation benchmark ===\n", .{}); + concurrentResult("full exports", level, export_ops, elapsed, export_latencies).print(); + concurrentResult("account probes", .{ .callers = 1, .ops_per_caller = probe_ops }, probe_ops, elapsed, probe_latencies).print(); + concurrentResult("concurrent writes", .{ .callers = 1, .ops_per_caller = write_ops }, write_ops, elapsed, write_latencies).print(); +} + fn benchSpace(allocator: std.mem.Allocator, options: Options) !void { var state = try initBench(allocator); defer state.deinit(); @@ -979,6 +1031,63 @@ fn repoCarWorker(did: []const u8, latencies: []u64) !void { } } +fn repoCarWorkerStarting( + did: []const u8, + latencies: []u64, + start: *std.atomic.Value(bool), +) !void { + while (!start.load(.acquire)) std.atomic.spinLoopHint(); + return repoCarWorker(did, latencies); +} + +fn accountProbeWorker( + did: []const u8, + latencies: []u64, + start: *std.atomic.Value(bool), +) !void { + var arena = std.heap.ArenaAllocator.init(std.heap.page_allocator); + defer arena.deinit(); + while (!start.load(.acquire)) std.atomic.spinLoopHint(); + waitForExportsToEnterStore(); + for (latencies) |*latency| { + _ = arena.reset(.retain_capacity); + const started = nowNs(); + const account = try zds.storage.store.findAccount(arena.allocator(), did) orelse return error.MissingAccount; + if (!std.mem.eql(u8, account.did, did)) return error.MissingAccount; + latency.* = nowNs() - started; + } +} + +fn repoWriteWorker( + account: zds.auth.tokens.Account, + first_index: usize, + latencies: []u64, + start: *std.atomic.Value(bool), +) !void { + var arena = std.heap.ArenaAllocator.init(std.heap.page_allocator); + defer arena.deinit(); + while (!start.load(.acquire)) std.atomic.spinLoopHint(); + waitForExportsToEnterStore(); + for (latencies, 0..) |*latency, offset| { + _ = arena.reset(.retain_capacity); + const index = first_index + offset; + const value = try benchRecordValue(arena.allocator(), index); + const started = nowNs(); + const result = try zds.storage.store.applyWrites(arena.allocator(), account, &.{.{ .create = .{ + .collection = bench_collection, + .rkey = try std.fmt.allocPrint(arena.allocator(), "rec{d:0>8}", .{index}), + .value = value, + } }}); + if (result.records.len != 1) return error.UnexpectedWriteResult; + latency.* = nowNs() - started; + } +} + +fn waitForExportsToEnterStore() void { + const until = nowNs() + std.time.ns_per_ms; + while (nowNs() < until) std.atomic.spinLoopHint(); +} + const CpuPath = enum { decode, render }; fn benchRecordBlockCpu(allocator: std.mem.Allocator, options: Options, path: CpuPath) !void { diff --git a/docs/getrepo-notes.md b/docs/getrepo-notes.md index ed3b625..69e0488 100644 --- a/docs/getrepo-notes.md +++ b/docs/getrepo-notes.md @@ -10,6 +10,11 @@ Current shape: - As of the 2026-06-29 getRepo pass, full repo export starts at the latest commit block and uses `zat.mst.collectReachableBlocks` to include only current MST and record blocks reachable from the commit's `data` CID. - Incremental export with `since` filters by `repo_rev`. - Repo writes already use lazy MST loading backed by `repo_blocks`, so ZDS is closer to a Hubble-style block/CAR-serving shape than an eager full-repo rebuild path. +- File-backed exports use a dedicated read-only SQLite connection and WAL + transaction. The root and reachable blocks come from one coherent snapshot; + CAR encoding happens after that snapshot closes. Account, OAuth, and write + operations continue on the primary connection instead of waiting behind repo + traversal and encoding. - `subscribeRepos` has a connection cap. Full `getRepo` now has its own lightweight route-local concurrency cap so backup/backfill exports cannot consume unbounded handler slots. Operators can tune it with `ZDS_MAX_CONCURRENT_REPO_EXPORTS`; the default is `4`, matching Tranquil's current full-export default. Open questions: @@ -26,6 +31,23 @@ Done in this pass: 4. Added lightweight route backpressure for full `getRepo`, configurable with `ZDS_MAX_CONCURRENT_REPO_EXPORTS`. 5. Ran `just bench repo-size`; reachable full export measured about 1.4 ms for 100 records, 16.8 ms for 1,000 records, and 95.7 ms for 5,000 records on the local synthetic benchmark. 6. Added `just bench get-repo` as the focused storage-pressure lane for this route. It measures full export construction, one-commit `since` export construction, and concurrent full export latency against one seeded repo. +7. Added account-read and write latency to the focused benchmark after + production traffic demonstrated that export throughput alone missed + process-wide database contention. +8. Moved file-backed full and incremental exports to independent read + transactions. This follows the isolation boundary used by the reference + PDS's actor-store reader and Tranquil's separate block store while retaining + ZDS's SQLite/WAL architecture. + +Validation on 2026-07-30: + +- In the local 10k-record, four-export comparison, the previous implementation + delayed an account lookup by 325 ms and a write by 336 ms. The isolated + implementation kept their maxima to 0.08 ms and 1.92 ms respectively. +- On a disposable Fly `shared-cpu-1x` machine with 1 GB RAM, eight concurrent + exports of a 10k-record repo completed without failure. Account lookups + remained below 20 ms and writes below 44 ms even while the single CPU was + saturated. Suggested next pass: diff --git a/src/storage/store.zig b/src/storage/store.zig index 6cea527..3a9e144 100644 --- a/src/storage/store.zig +++ b/src/storage/store.zig @@ -444,6 +444,7 @@ const BlobRef = struct { const RepoBlockReader = struct { allocator: std.mem.Allocator, did: []const u8, + connection: ?zqlite.Conn = null, locked: bool = false, fn reader(self: *RepoBlockReader) zat.mst.BlockReader { @@ -457,6 +458,9 @@ const RepoBlockReader = struct { const self: *RepoBlockReader = @ptrCast(@alignCast(ctx)); const cid = try cidText(self.allocator, cid_raw); + if (self.connection) |read_conn| { + return repoBlockDataFrom(read_conn, self.allocator, self.did, cid); + } if (self.locked) { return repoBlockDataLocked(self.allocator, self.did, cid); } @@ -470,7 +474,16 @@ const RepoBlockReader = struct { }; fn repoBlockDataLocked(allocator: std.mem.Allocator, did: []const u8, cid: []const u8) !?[]const u8 { - const row = try conn.row( + return repoBlockDataFrom(conn, allocator, did, cid); +} + +fn repoBlockDataFrom( + read_conn: zqlite.Conn, + allocator: std.mem.Allocator, + did: []const u8, + cid: []const u8, +) !?[]const u8 { + const row = try read_conn.row( \\SELECT data \\FROM repo_blocks \\WHERE did = ? AND cid = ? @@ -484,6 +497,7 @@ fn repoBlockDataLocked(allocator: std.mem.Allocator, did: []const u8, cid: []con } var conn: zqlite.Conn = undefined; +var database_path: ?[:0]u8 = null; var initialized = false; var store_io: Io = undefined; var db_mutex: Io.Mutex = .init; @@ -498,7 +512,12 @@ pub fn init(io: Io, path: []const u8) !void { } const path_z = try std.heap.page_allocator.dupeZ(u8, path); + var path_owned = true; + defer if (path_owned) std.heap.page_allocator.free(path_z); + conn = try zqlite.open(path_z.ptr, zqlite.OpenFlags.Create | zqlite.OpenFlags.ReadWrite); + database_path = path_z; + path_owned = false; initialized = true; errdefer close(); @@ -512,6 +531,8 @@ pub fn init(io: Io, path: []const u8) !void { pub fn close() void { if (!initialized) return; conn.close(); + if (database_path) |path| std.heap.page_allocator.free(path); + database_path = null; initialized = false; } @@ -2815,16 +2836,43 @@ pub fn writeRepoCar(allocator: std.mem.Allocator, did: []const u8) ![]const u8 { pub fn writeRepoCarSince(allocator: std.mem.Allocator, did: []const u8, since: ?[]const u8) ![]const u8 { if (since == null) return writeRepoCarFull(allocator, did); - db_mutex.lockUncancelable(store_io); - defer db_mutex.unlock(store_io); - try requireInitialized(); + const path = database_path orelse return Error.StoreNotInitialized; + if (std.mem.eql(u8, path, ":memory:")) { + db_mutex.lockUncancelable(store_io); + defer db_mutex.unlock(store_io); + try requireInitialized(); + return writeRepoCarSinceFrom(allocator, conn, did, since.?); + } - const root = try latestRootLocked(allocator, did); + const read_conn = try openRepoReadConnection(path); + defer read_conn.close(); + try read_conn.transaction(); + errdefer read_conn.rollback(); + const data = try collectRepoCarSinceFrom(allocator, read_conn, did, since.?); + try read_conn.commit(); + return writeRepoCarData(allocator, data); +} + +fn writeRepoCarSinceFrom( + allocator: std.mem.Allocator, + read_conn: zqlite.Conn, + did: []const u8, + since_rev: []const u8, +) ![]const u8 { + const data = try collectRepoCarSinceFrom(allocator, read_conn, did, since_rev); + return writeRepoCarData(allocator, data); +} + +fn collectRepoCarSinceFrom( + allocator: std.mem.Allocator, + read_conn: zqlite.Conn, + did: []const u8, + since_rev: []const u8, +) !RepoCarData { + const root = try latestRootFrom(read_conn, allocator, did); const root_raw = try zat.multibase.base32lower.decode(allocator, root.cid[1..]); - const car_root = zat.cbor.Cid{ .raw = root_raw }; - const since_rev = since.?; - var rows = try conn.rows( + var rows = try read_conn.rows( \\SELECT cid, data \\FROM repo_blocks \\WHERE did = ? AND (repo_rev IS NULL OR repo_rev > ?) @@ -2843,19 +2891,43 @@ pub fn writeRepoCarSince(allocator: std.mem.Allocator, did: []const u8, since: ? }); } if (rows.err) |err| return err; - const c: zat.car.Car = .{ - .roots = &.{car_root}, - .blocks = blocks.items, + return .{ + .root_cid_raw = root_raw, + .blocks = try blocks.toOwnedSlice(allocator), }; - return zat.car.writeAlloc(allocator, c); } fn writeRepoCarFull(allocator: std.mem.Allocator, did: []const u8) ![]const u8 { - db_mutex.lockUncancelable(store_io); - defer db_mutex.unlock(store_io); - try requireInitialized(); + const path = database_path orelse return Error.StoreNotInitialized; + if (std.mem.eql(u8, path, ":memory:")) { + db_mutex.lockUncancelable(store_io); + defer db_mutex.unlock(store_io); + try requireInitialized(); + const data = try collectRepoCarFullFrom(allocator, conn, did); + return writeRepoCarData(allocator, data); + } - const root = try latestCommitRawLocked(allocator, did) orelse return Error.RepoNotFound; + const read_conn = try openRepoReadConnection(path); + defer read_conn.close(); + try read_conn.transaction(); + errdefer read_conn.rollback(); + + const data = try collectRepoCarFullFrom(allocator, read_conn, did); + try read_conn.commit(); + return writeRepoCarData(allocator, data); +} + +const RepoCarData = struct { + root_cid_raw: []const u8, + blocks: []zat.car.Block, +}; + +fn collectRepoCarFullFrom( + allocator: std.mem.Allocator, + read_conn: zqlite.Conn, + did: []const u8, +) !RepoCarData { + const root = try latestCommitRawFrom(read_conn, allocator, did) orelse return Error.RepoNotFound; var blocks: std.ArrayList(zat.car.Block) = .empty; try blocks.append(allocator, .{ .cid_raw = root.commit_cid_raw, @@ -2865,7 +2937,7 @@ fn writeRepoCarFull(allocator: std.mem.Allocator, did: []const u8) ![]const u8 { var repo_block_reader = RepoBlockReader{ .allocator = allocator, .did = did, - .locked = true, + .connection = read_conn, }; try zat.mst.collectReachableBlocks( allocator, @@ -2875,12 +2947,27 @@ fn writeRepoCarFull(allocator: std.mem.Allocator, did: []const u8) ![]const u8 { .{ .include_records = true }, ); + return .{ + .root_cid_raw = root.commit_cid_raw, + .blocks = try blocks.toOwnedSlice(allocator), + }; +} + +fn writeRepoCarData(allocator: std.mem.Allocator, data: RepoCarData) ![]const u8 { return zat.car.writeAlloc(allocator, .{ - .roots = &.{.{ .raw = root.commit_cid_raw }}, - .blocks = blocks.items, + .roots = &.{.{ .raw = data.root_cid_raw }}, + .blocks = data.blocks, }); } +fn openRepoReadConnection(path: [:0]const u8) !zqlite.Conn { + const read_conn = try zqlite.open(path.ptr, zqlite.OpenFlags.ReadOnly); + errdefer read_conn.close(); + try read_conn.busyTimeout(5000); + try read_conn.execNoArgs("PRAGMA query_only=ON"); + return read_conn; +} + pub fn writeRepoListJson(allocator: std.mem.Allocator, cursor: ?[]const u8, limit: usize) ![]const u8 { db_mutex.lockUncancelable(store_io); defer db_mutex.unlock(store_io); @@ -5066,7 +5153,15 @@ const CurrentCommit = struct { }; fn latestCommitRawLocked(allocator: std.mem.Allocator, did: []const u8) !?CurrentCommit { - const row = try conn.row( + return latestCommitRawFrom(conn, allocator, did); +} + +fn latestCommitRawFrom( + read_conn: zqlite.Conn, + allocator: std.mem.Allocator, + did: []const u8, +) !?CurrentCommit { + const row = try read_conn.row( \\SELECT c.cid, c.rev, rb.data \\FROM commits c \\JOIN repo_blocks rb ON rb.did = c.did AND rb.cid = c.cid @@ -5406,7 +5501,15 @@ const Root = struct { }; fn latestRootLocked(allocator: std.mem.Allocator, did: []const u8) !Root { - const row = try conn.row( + return latestRootFrom(conn, allocator, did); +} + +fn latestRootFrom( + read_conn: zqlite.Conn, + allocator: std.mem.Allocator, + did: []const u8, +) !Root { + const row = try read_conn.row( \\SELECT cid, rev \\FROM commits \\WHERE did = ? @@ -6665,6 +6768,78 @@ test "full getRepo exports only current reachable record blocks" { try std.testing.expect(!try carContainsCid(allocator, deleted_car, second_cid)); } +test "file-backed repo export snapshot does not block writes" { + var arena = std.heap.ArenaAllocator.init(std.testing.allocator); + defer arena.deinit(); + const allocator = arena.allocator(); + + var tmp = std.testing.tmpDir(.{}); + defer tmp.cleanup(); + var path_buf: [std.fs.max_path_bytes]u8 = undefined; + const root_len = try tmp.dir.realPath(std.testing.io, &path_buf); + const db_path = try std.fmt.allocPrint(allocator, "{s}/zds.sqlite3", .{path_buf[0..root_len]}); + + try init(std.Options.debug_io, db_path); + defer close(); + + const account = try createAccount( + allocator, + "export-snapshot.test", + "export-snapshot@test.com", + "password", + "did:plc:exportsnapshot", + true, + ); + const first = try std.json.parseFromSlice( + std.json.Value, + allocator, + "{\"$type\":\"app.bsky.feed.post\",\"text\":\"before snapshot\"}", + .{}, + ); + defer first.deinit(); + const first_rkey = "3jzfcijpj2z2g"; + _ = try create(allocator, account, "app.bsky.feed.post", first_rkey, first.value); + + const read_conn = try openRepoReadConnection(database_path.?); + defer read_conn.close(); + try read_conn.transaction(); + errdefer read_conn.rollback(); + const snapshot = try collectRepoCarFullFrom(allocator, read_conn, account.did); + + const second = try std.json.parseFromSlice( + std.json.Value, + allocator, + "{\"$type\":\"app.bsky.feed.post\",\"text\":\"during snapshot\"}", + .{}, + ); + defer second.deinit(); + const second_rkey = "3jzfcijpj2z2h"; + _ = try create(allocator, account, "app.bsky.feed.post", second_rkey, second.value); + + try read_conn.commit(); + const first_path = try std.fmt.allocPrint(allocator, "app.bsky.feed.post/{s}", .{first_rkey}); + const second_path = try std.fmt.allocPrint(allocator, "app.bsky.feed.post/{s}", .{second_rkey}); + + const snapshot_car = try writeRepoCarData(allocator, snapshot); + const loaded_snapshot = try zat.loadCommitFromCAR(allocator, snapshot_car); + var snapshot_tree = try zat.mst.Mst.loadFromBlocks( + allocator, + loaded_snapshot.repo_car, + loaded_snapshot.commit.data_cid, + ); + try std.testing.expect(snapshot_tree.get(first_path) != null); + try std.testing.expect(snapshot_tree.get(second_path) == null); + + const current_car = try writeRepoCar(allocator, account.did); + const loaded_current = try zat.loadCommitFromCAR(allocator, current_car); + var current_tree = try zat.mst.Mst.loadFromBlocks( + allocator, + loaded_current.repo_car, + loaded_current.commit.data_cid, + ); + try std.testing.expect(current_tree.get(second_path) != null); +} + test "permissioned spaces store self-owned records outside public repo" { var arena = std.heap.ArenaAllocator.init(std.testing.allocator); defer arena.deinit(); -- 2.51.2