diff --git a/bench/README.md b/bench/README.md index 0074b86..1339631 100644 --- a/bench/README.md +++ b/bench/README.md @@ -65,14 +65,14 @@ The operation classes match Tranquil's `metastore` bench: apply/write one record, get one record CID/record, and list records from a seeded repo. The implementations are not identical internally. Tranquil measures its metastore handler pool and eventlog path; ZDS measures the local store API, including repo -commit construction and the current process-wide store mutex. +commit construction, per-DID write lanes, and the shared SQLite connection. | operation | callers | ops | zds | tranquil | |---|---:|---:|---:|---:| | apply/write | 1 | 1000 | 206 ops/s, p95 11.9 ms | not measured at this count | | apply/write | 1 | 5000 | not rerun | 292 ops/s, p95 5.8 ms | -| apply/write | 10 | 10000 | 106 ops/s, p95 119 ms | 436 ops/s, p95 36.0 ms | -| apply/write | 100 | 20000 | 125 ops/s, p95 110 ms | 944 ops/s, p95 119 ms | +| apply/write | 10 | 10000 | 303 ops/s, p95 88 ms | 436 ops/s, p95 36.0 ms | +| apply/write | 100 | 20000 | 1287 ops/s, p95 72 ms | 944 ops/s, p95 119 ms | | apply/write | 1000 | 50000 | not rerun | 1174 ops/s, then read backpressure | | get record | 1 | 1000 | 55k ops/s, p95 25 us | not measured at this count | | get record | 10 | 10000 | 56k ops/s, p95 64 us | 571k ops/s, p95 31 us | @@ -86,19 +86,18 @@ phase failed with `metastore handler backpressure` on this machine. ## write profile -`write-profile` shows that the current ZDS write path is dominated by serialized -time around the store mutex, not SQLite alone: +`write-profile` tracks where concurrent writes spend time: | callers | ops | throughput | lock wait | load repo | build commit | sqlite/event | |---:|---:|---:|---:|---:|---:|---:| -| 10 | 500 | 2386 ops/s | 83% | 3% | 8% | 6% | -| 10 | 5000 | 353 ops/s | 82% | 13% | 2% | 3% | -| 100 | 2000 | 1310 ops/s | 98% | 1% | 1% | 1% | - -The next write-path work should shrink or restructure the serialized section: -keep validation and request parsing outside the store lock, move toward a -Tranquil-style write worker/queue for repo mutation, and avoid rebuilding more -repo state than a single write needs. +| 10 | 5000 | 708 ops/s | 25% | 26% | 3% | 46% | +| 100 | 2000 | 3683 ops/s | 64% | 10% | 1% | 25% | + +ZDS now separates repo ordering from database serialization: writes take a +per-DID lane before loading and mutating the repo, then use the DB mutex for the +shared SQLite connection. The next write-path work should continue toward the +Tranquil shape: bounded write queues with explicit backpressure, less full-repo +state loading, and a narrower SQLite persistence phase. ## comparison work diff --git a/docs/operations.md b/docs/operations.md index b4f3341..93363ee 100644 --- a/docs/operations.md +++ b/docs/operations.md @@ -24,6 +24,8 @@ just bench write just bench read just bench repo just bench blob +just bench metastore 10 1000 +just bench write-profile 10 500 just bench run --records 1000 ``` diff --git a/src/internal/sharded_locks.zig b/src/internal/sharded_locks.zig new file mode 100644 index 0000000..73c044f --- /dev/null +++ b/src/internal/sharded_locks.zig @@ -0,0 +1,38 @@ +const std = @import("std"); + +const Io = std.Io; + +pub fn ShardedLocks(comptime count: usize) type { + if (count == 0) @compileError("ShardedLocks requires at least one shard"); + + return struct { + const Self = @This(); + + locks: [count]Io.Mutex = [_]Io.Mutex{.init} ** count, + + pub const Guard = struct { + owner: *Self, + io: Io, + index: usize, + + pub fn unlock(self: Guard) void { + self.owner.locks[self.index].unlock(self.io); + } + }; + + pub fn lock(self: *Self, io: Io, key: []const u8) Guard { + const index = shardIndex(key); + self.locks[index].lockUncancelable(io); + return .{ .owner = self, .io = io, .index = index }; + } + + pub fn shardIndex(key: []const u8) usize { + return @intCast(std.hash.Wyhash.hash(0, key) % count); + } + }; +} + +test "same key maps to same shard" { + const Locks = ShardedLocks(16); + try std.testing.expectEqual(Locks.shardIndex("did:plc:abc"), Locks.shardIndex("did:plc:abc")); +} diff --git a/src/root.zig b/src/root.zig index d28d866..95a9db3 100644 --- a/src/root.zig +++ b/src/root.zig @@ -31,6 +31,7 @@ pub const http = struct { pub const internal = struct { pub const cli = @import("internal/cli.zig"); + pub const sharded_locks = @import("internal/sharded_locks.zig"); }; pub const storage = struct { diff --git a/src/storage/store.zig b/src/storage/store.zig index 8693ff5..e4136c5 100644 --- a/src/storage/store.zig +++ b/src/storage/store.zig @@ -3,6 +3,7 @@ const atid = @import("../core/atid.zig"); const auth = @import("../auth/tokens.zig"); const blobstore = @import("blobstore.zig"); const eventlog = @import("eventlog.zig"); +const sharded_locks = @import("../internal/sharded_locks.zig"); const zat = @import("zat"); const zqlite = @import("zqlite"); const Io = std.Io; @@ -169,7 +170,9 @@ const BlobRef = struct { var conn: zqlite.Conn = undefined; var initialized = false; var store_io: Io = undefined; -var mutex: Io.Mutex = .init; +var db_mutex: Io.Mutex = .init; +var write_lanes: sharded_locks.ShardedLocks(32) = .{}; +var next_seq: std.atomic.Value(u64) = .init(1); pub fn init(io: Io, path: []const u8) !void { if (initialized) return; @@ -187,6 +190,7 @@ pub fn init(io: Io, path: []const u8) !void { try conn.execNoArgs("PRAGMA journal_mode=WAL"); try conn.execNoArgs("PRAGMA foreign_keys=ON"); try migrate(); + next_seq.store(try loadNextSeqLocked(), .release); } pub fn close() void { @@ -220,16 +224,16 @@ pub fn resolveRepo(repo: []const u8) ?auth.Account { } pub fn findAccount(allocator: std.mem.Allocator, identifier: []const u8) !?auth.Account { - mutex.lockUncancelable(store_io); - defer mutex.unlock(store_io); + db_mutex.lockUncancelable(store_io); + defer db_mutex.unlock(store_io); try requireInitialized(); return try findAccountLocked(allocator, identifier); } pub fn searchAccounts(allocator: std.mem.Allocator, query: []const u8, limit: usize) ![]auth.Account { - mutex.lockUncancelable(store_io); - defer mutex.unlock(store_io); + db_mutex.lockUncancelable(store_io); + defer db_mutex.unlock(store_io); try requireInitialized(); const actual_limit = if (limit == 0) 25 else limit; @@ -260,8 +264,8 @@ pub fn searchAccounts(allocator: std.mem.Allocator, query: []const u8, limit: us } pub fn listResidents(allocator: std.mem.Allocator, limit: usize) ![]Resident { - mutex.lockUncancelable(store_io); - defer mutex.unlock(store_io); + db_mutex.lockUncancelable(store_io); + defer db_mutex.unlock(store_io); try requireInitialized(); const actual_limit = if (limit == 0) 24 else @min(limit, 100); @@ -300,8 +304,8 @@ pub fn listResidents(allocator: std.mem.Allocator, limit: usize) ![]Resident { } pub fn profileAvatarCid(allocator: std.mem.Allocator, did: []const u8) !?[]const u8 { - mutex.lockUncancelable(store_io); - defer mutex.unlock(store_io); + db_mutex.lockUncancelable(store_io); + defer db_mutex.unlock(store_io); try requireInitialized(); const row = try conn.row( @@ -334,8 +338,8 @@ pub fn profileAvatarCid(allocator: std.mem.Allocator, did: []const u8) !?[]const } pub fn listCollectionSummaries(allocator: std.mem.Allocator, limit: usize) ![]CollectionSummary { - mutex.lockUncancelable(store_io); - defer mutex.unlock(store_io); + db_mutex.lockUncancelable(store_io); + defer db_mutex.unlock(store_io); try requireInitialized(); const actual_limit = if (limit == 0) 12 else @min(limit, 50); @@ -362,8 +366,8 @@ pub fn listCollectionSummaries(allocator: std.mem.Allocator, limit: usize) ![]Co } pub fn listLandingRecentRecords(allocator: std.mem.Allocator, limit: usize) ![]RecentRecord { - mutex.lockUncancelable(store_io); - defer mutex.unlock(store_io); + db_mutex.lockUncancelable(store_io); + defer db_mutex.unlock(store_io); try requireInitialized(); const actual_limit = if (limit == 0) 8 else @min(limit, 24); @@ -434,8 +438,8 @@ pub fn createAccountWithSigningKey( activated: bool, signing_key: [32]u8, ) !auth.Account { - mutex.lockUncancelable(store_io); - defer mutex.unlock(store_io); + db_mutex.lockUncancelable(store_io); + defer db_mutex.unlock(store_io); try requireInitialized(); var salt: [16]u8 = undefined; @@ -462,22 +466,22 @@ pub fn createAccountWithSigningKey( } pub fn generateAccountSigningKey() ![32]u8 { - mutex.lockUncancelable(store_io); - defer mutex.unlock(store_io); + db_mutex.lockUncancelable(store_io); + defer db_mutex.unlock(store_io); try requireInitialized(); return generateSigningKey(); } pub fn signingKeypair(did: []const u8) !zat.Keypair { - mutex.lockUncancelable(store_io); - defer mutex.unlock(store_io); + db_mutex.lockUncancelable(store_io); + defer db_mutex.unlock(store_io); try requireInitialized(); return signingKeypairLocked(did); } pub fn setAccountActive(did: []const u8, active: bool) !void { - mutex.lockUncancelable(store_io); - defer mutex.unlock(store_io); + db_mutex.lockUncancelable(store_io); + defer db_mutex.unlock(store_io); try requireInitialized(); if (active) { try conn.exec( @@ -496,8 +500,8 @@ pub fn setAccountActive(did: []const u8, active: bool) !void { } pub fn sequenceAccountEvent(allocator: std.mem.Allocator, did: []const u8, active: bool) !void { - mutex.lockUncancelable(store_io); - defer mutex.unlock(store_io); + db_mutex.lockUncancelable(store_io); + defer db_mutex.unlock(store_io); try requireInitialized(); const seq = try nextSeqLocked(); const frame = try accountEventFrame(allocator, seq, did, active); @@ -506,8 +510,8 @@ pub fn sequenceAccountEvent(allocator: std.mem.Allocator, did: []const u8, activ } pub fn sequenceIdentityEvent(allocator: std.mem.Allocator, did: []const u8, handle: []const u8) !void { - mutex.lockUncancelable(store_io); - defer mutex.unlock(store_io); + db_mutex.lockUncancelable(store_io); + defer db_mutex.unlock(store_io); try requireInitialized(); const seq = try nextSeqLocked(); const frame = try identityEventFrame(allocator, seq, did, handle); @@ -516,8 +520,8 @@ pub fn sequenceIdentityEvent(allocator: std.mem.Allocator, did: []const u8, hand } pub fn sequenceSyncEvent(allocator: std.mem.Allocator, did: []const u8) !void { - mutex.lockUncancelable(store_io); - defer mutex.unlock(store_io); + db_mutex.lockUncancelable(store_io); + defer db_mutex.unlock(store_io); try requireInitialized(); const seq = try nextSeqLocked(); const commit = try latestCommitRawLocked(allocator, did) orelse return Error.RepoNotFound; @@ -527,8 +531,8 @@ pub fn sequenceSyncEvent(allocator: std.mem.Allocator, did: []const u8) !void { } pub fn isAccountActive(did: []const u8) bool { - mutex.lockUncancelable(store_io); - defer mutex.unlock(store_io); + db_mutex.lockUncancelable(store_io); + defer db_mutex.unlock(store_io); requireInitialized() catch return false; return accountActiveLocked(did) catch false; } @@ -563,8 +567,8 @@ pub fn putOAuthRequest( dpop_jkt: ?[]const u8, expires_at: i64, ) !void { - mutex.lockUncancelable(store_io); - defer mutex.unlock(store_io); + db_mutex.lockUncancelable(store_io); + defer db_mutex.unlock(store_io); try requireInitialized(); try conn.exec( \\INSERT INTO oauth_requests @@ -574,8 +578,8 @@ pub fn putOAuthRequest( } pub fn getOAuthRequest(allocator: std.mem.Allocator, request_id: []const u8) !?OAuthRequest { - mutex.lockUncancelable(store_io); - defer mutex.unlock(store_io); + db_mutex.lockUncancelable(store_io); + defer db_mutex.unlock(store_io); try requireInitialized(); const row = try conn.row( \\SELECT request_id, client_id, redirect_uri, scope, state, code_challenge, code_challenge_method, response_mode, login_hint, dpop_jkt, expires_at, sub, code @@ -588,8 +592,8 @@ pub fn getOAuthRequest(allocator: std.mem.Allocator, request_id: []const u8) !?O } pub fn authorizeOAuthRequest(request_id: []const u8, did: []const u8, code: []const u8) !void { - mutex.lockUncancelable(store_io); - defer mutex.unlock(store_io); + db_mutex.lockUncancelable(store_io); + defer db_mutex.unlock(store_io); try requireInitialized(); try conn.exec( \\UPDATE oauth_requests @@ -599,8 +603,8 @@ pub fn authorizeOAuthRequest(request_id: []const u8, did: []const u8, code: []co } pub fn consumeOAuthCode(allocator: std.mem.Allocator, code: []const u8) !?OAuthRequest { - mutex.lockUncancelable(store_io); - defer mutex.unlock(store_io); + db_mutex.lockUncancelable(store_io); + defer db_mutex.unlock(store_io); try requireInitialized(); const row = try conn.row( \\SELECT request_id, client_id, redirect_uri, scope, state, code_challenge, code_challenge_method, response_mode, login_hint, dpop_jkt, expires_at, sub, code @@ -622,8 +626,8 @@ pub fn putOAuthToken( refresh_token: []const u8, expires_at: i64, ) !void { - mutex.lockUncancelable(store_io); - defer mutex.unlock(store_io); + db_mutex.lockUncancelable(store_io); + defer db_mutex.unlock(store_io); try requireInitialized(); try conn.exec( \\INSERT INTO oauth_tokens (access_token, refresh_token, did, client_id, scope, expires_at) @@ -632,8 +636,8 @@ pub fn putOAuthToken( } pub fn getOAuthToken(allocator: std.mem.Allocator, token: []const u8) !?OAuthToken { - mutex.lockUncancelable(store_io); - defer mutex.unlock(store_io); + db_mutex.lockUncancelable(store_io); + defer db_mutex.unlock(store_io); try requireInitialized(); const row = try conn.row( \\SELECT did, client_id, scope, access_token, refresh_token, expires_at, revoked_at @@ -656,8 +660,8 @@ pub fn getOAuthToken(allocator: std.mem.Allocator, token: []const u8) !?OAuthTok } pub fn revokeOAuthToken(token: []const u8) !void { - mutex.lockUncancelable(store_io); - defer mutex.unlock(store_io); + db_mutex.lockUncancelable(store_io); + defer db_mutex.unlock(store_io); try requireInitialized(); try conn.exec( \\UPDATE oauth_tokens @@ -723,16 +727,31 @@ fn applyWritesMeasured(allocator: std.mem.Allocator, account: auth.Account, ops: addProfile(profile, .validation_ns, elapsedNs(validation_start)); const lock_start = monotonicNs(); - mutex.lockUncancelable(store_io); + const lane = write_lanes.lock(store_io, account.did); + defer lane.unlock(); addProfile(profile, .lock_wait_ns, elapsedNs(lock_start)); - defer mutex.unlock(store_io); - try requireInitialized(); - const load_start = monotonicNs(); + try requireInitialized(); const seq = try nextSeqLocked(); - const current = try latestCommitRawLocked(allocator, account.did); - const rev = try revForSeq(allocator, seq, if (current) |root| root.rev else null); - const repo_car = try readRepoCarLocked(allocator, account.did); + const load_start = monotonicNs(); + const loaded = blk: { + db_mutex.lockUncancelable(store_io); + defer db_mutex.unlock(store_io); + const current = try latestCommitRawLocked(allocator, account.did); + break :blk .{ + .current = current, + .rev = try revForSeq(allocator, seq, if (current) |root| root.rev else null), + .repo_car = try readRepoCarLocked(allocator, account.did), + .keypair = try signingKeypairLocked(account.did), + .ops = try resolveWriteOpsLocked(allocator, ops), + }; + }; + const current = loaded.current; + const rev = loaded.rev; + const repo_car = loaded.repo_car; + const keypair = loaded.keypair; + const resolved_ops = loaded.ops; + var tree = if (current) |root| try zat.mst.Mst.loadFromBlocks(allocator, repo_car, root.data_cid_raw) else @@ -745,9 +764,9 @@ fn applyWritesMeasured(allocator: std.mem.Allocator, account: auth.Account, ops: var blob_refs: std.ArrayList(BlobRef) = .empty; const stage_start = monotonicNs(); - for (ops) |op| switch (op) { + for (resolved_ops) |op| switch (op) { .create => |create_op| { - const rkey = create_op.rkey orelse try nextRkeyLocked(allocator); + const rkey = create_op.rkey orelse return Error.InvalidRecordKey; const record = try stageRecordWrite(allocator, &tree, account, create_op.collection, rkey, create_op.value, rev, seq, &record_blocks, &blob_refs); try records.append(allocator, record); }, @@ -765,10 +784,12 @@ fn applyWritesMeasured(allocator: std.mem.Allocator, account: auth.Account, ops: const build_start = monotonicNs(); const data_cid = try tree.rootCid(); try writeMstBlocks(allocator, &tree, &mst_blocks); - const commit = try signedCommit(allocator, account.did, rev, data_cid, if (current) |root| root.commit_cid_raw else null); + const commit = try signedCommitWithKeypair(allocator, account.did, rev, data_cid, if (current) |root| root.commit_cid_raw else null, &keypair); addProfile(profile, .build_commit_ns, elapsedNs(build_start)); const sql_start = monotonicNs(); + db_mutex.lockUncancelable(store_io); + defer db_mutex.unlock(store_io); try conn.exclusiveTransaction(); errdefer conn.rollback(); for (record_blocks.items) |block| { @@ -791,7 +812,7 @@ fn applyWritesMeasured(allocator: std.mem.Allocator, account: auth.Account, ops: \\ON CONFLICT(did, cid) DO UPDATE SET data = excluded.data , .{ account.did, commit.cid, zqlite.blob(commit.data) }); - for (ops) |op| switch (op) { + for (resolved_ops) |op| switch (op) { .delete => |delete_op| { const uri = try std.fmt.allocPrint(allocator, "at://{s}/{s}/{s}", .{ account.did, delete_op.collection, delete_op.rkey }); try conn.exec("DELETE FROM expected_blobs WHERE record_uri = ?", .{uri}); @@ -834,7 +855,7 @@ fn applyWritesMeasured(allocator: std.mem.Allocator, account: auth.Account, ops: commit.data, record_blocks.items, mst_blocks.items, - ops, + resolved_ops, records.items, ); try conn.exec( @@ -858,7 +879,6 @@ fn applyWritesMeasured(allocator: std.mem.Allocator, account: auth.Account, ops: .records = try records.toOwnedSlice(allocator), }; } - const ProfileField = enum { validation_ns, lock_wait_ns, @@ -920,8 +940,8 @@ fn nowMicros() u64 { } pub fn get(did: []const u8, collection: []const u8, rkey: []const u8) ?Record { - mutex.lockUncancelable(store_io); - defer mutex.unlock(store_io); + db_mutex.lockUncancelable(store_io); + defer db_mutex.unlock(store_io); requireInitialized() catch return null; const row = conn.row( @@ -935,15 +955,15 @@ pub fn get(did: []const u8, collection: []const u8, rkey: []const u8) ?Record { } pub fn getByUri(uri: []const u8) ?Record { - mutex.lockUncancelable(store_io); - defer mutex.unlock(store_io); + db_mutex.lockUncancelable(store_io); + defer db_mutex.unlock(store_io); requireInitialized() catch return null; return getByUriLocked(uri); } pub fn listRecentRecords(allocator: std.mem.Allocator, limit: usize) ![]Record { - mutex.lockUncancelable(store_io); - defer mutex.unlock(store_io); + db_mutex.lockUncancelable(store_io); + defer db_mutex.unlock(store_io); try requireInitialized(); var rows = try conn.rows( @@ -968,8 +988,8 @@ pub fn listRecords( collection: []const u8, limit: usize, ) ![]Record { - mutex.lockUncancelable(store_io); - defer mutex.unlock(store_io); + db_mutex.lockUncancelable(store_io); + defer db_mutex.unlock(store_io); try requireInitialized(); var rows = try conn.rows( @@ -995,8 +1015,8 @@ pub fn listRecordsContaining( needle: []const u8, limit: usize, ) ![]Record { - mutex.lockUncancelable(store_io); - defer mutex.unlock(store_io); + db_mutex.lockUncancelable(store_io); + defer db_mutex.unlock(store_io); try requireInitialized(); var rows = try conn.rows( @@ -1023,8 +1043,8 @@ pub fn listRecordsByDidContaining( needle: []const u8, limit: usize, ) ![]Record { - mutex.lockUncancelable(store_io); - defer mutex.unlock(store_io); + db_mutex.lockUncancelable(store_io); + defer db_mutex.unlock(store_io); try requireInitialized(); var rows = try conn.rows( @@ -1068,8 +1088,8 @@ fn getByStoredUriLocked(uri: []const u8) ?Record { } pub fn count(did: []const u8, collection: []const u8) usize { - mutex.lockUncancelable(store_io); - defer mutex.unlock(store_io); + db_mutex.lockUncancelable(store_io); + defer db_mutex.unlock(store_io); requireInitialized() catch return 0; const row = conn.row("SELECT count(*) FROM records WHERE did = ? AND collection = ?", .{ did, collection }) catch return 0; if (row == null) return 0; @@ -1078,8 +1098,8 @@ pub fn count(did: []const u8, collection: []const u8) usize { } pub fn countSubject(collection: []const u8, subject_did: []const u8) usize { - mutex.lockUncancelable(store_io); - defer mutex.unlock(store_io); + db_mutex.lockUncancelable(store_io); + defer db_mutex.unlock(store_io); requireInitialized() catch return 0; const row = conn.row("SELECT count(*) FROM records WHERE collection = ? AND instr(value_json, ?) > 0", .{ collection, subject_did }) catch return 0; if (row == null) return 0; @@ -1088,8 +1108,8 @@ pub fn countSubject(collection: []const u8, subject_did: []const u8) usize { } pub fn listCollectionsJson(allocator: std.mem.Allocator, did: []const u8) ![]const u8 { - mutex.lockUncancelable(store_io); - defer mutex.unlock(store_io); + db_mutex.lockUncancelable(store_io); + defer db_mutex.unlock(store_io); try requireInitialized(); var rows = try conn.rows( @@ -1115,8 +1135,8 @@ pub fn listCollectionsJson(allocator: std.mem.Allocator, did: []const u8) ![]con } pub fn getAppPreferences(allocator: std.mem.Allocator, did: []const u8, namespace: []const u8) ![]AppPreference { - mutex.lockUncancelable(store_io); - defer mutex.unlock(store_io); + db_mutex.lockUncancelable(store_io); + defer db_mutex.unlock(store_io); try requireInitialized(); const like_pattern = try std.fmt.allocPrint(allocator, "{s}.%", .{namespace}); @@ -1145,8 +1165,8 @@ pub fn replaceAppPreferences( namespace: []const u8, preferences: []const AppPreference, ) !void { - mutex.lockUncancelable(store_io); - defer mutex.unlock(store_io); + db_mutex.lockUncancelable(store_io); + defer db_mutex.unlock(store_io); try requireInitialized(); const like_pattern = try std.fmt.allocPrint(allocator, "{s}.%", .{namespace}); try conn.exclusiveTransaction(); @@ -1173,8 +1193,8 @@ pub fn putBlob( ) ![]const u8 { const cid = try cidForBlob(allocator, data); - mutex.lockUncancelable(store_io); - defer mutex.unlock(store_io); + db_mutex.lockUncancelable(store_io); + defer db_mutex.unlock(store_io); try requireInitialized(); try blobstore.put(allocator, io, account.did, cid, data); try conn.exec( @@ -1192,8 +1212,8 @@ pub fn getBlob( did: []const u8, cid: []const u8, ) ?BlobRecord { - mutex.lockUncancelable(store_io); - defer mutex.unlock(store_io); + db_mutex.lockUncancelable(store_io); + defer db_mutex.unlock(store_io); requireInitialized() catch return null; const row = conn.row( \\SELECT mime_type, size @@ -1210,8 +1230,8 @@ pub fn getBlob( } pub fn writeBlobListJson(allocator: std.mem.Allocator, did: []const u8, limit: usize) ![]const u8 { - mutex.lockUncancelable(store_io); - defer mutex.unlock(store_io); + db_mutex.lockUncancelable(store_io); + defer db_mutex.unlock(store_io); try requireInitialized(); var rows = try conn.rows( @@ -1243,8 +1263,8 @@ pub fn writeBlobListJson(allocator: std.mem.Allocator, did: []const u8, limit: u } pub fn writeMissingBlobsJson(allocator: std.mem.Allocator, did: []const u8, limit: usize) ![]const u8 { - mutex.lockUncancelable(store_io); - defer mutex.unlock(store_io); + db_mutex.lockUncancelable(store_io); + defer db_mutex.unlock(store_io); try requireInitialized(); var rows = try conn.rows( @@ -1281,8 +1301,8 @@ pub fn writeMissingBlobsJson(allocator: std.mem.Allocator, did: []const u8, limi } pub fn writeAccountStatusJson(allocator: std.mem.Allocator, did: []const u8) ![]const u8 { - mutex.lockUncancelable(store_io); - defer mutex.unlock(store_io); + db_mutex.lockUncancelable(store_io); + defer db_mutex.unlock(store_io); try requireInitialized(); const record_count = try scalarCountLocked( @@ -1332,8 +1352,8 @@ pub fn importRepo( records: []const ImportedRecord, blocks: []const ImportedBlock, ) !void { - mutex.lockUncancelable(store_io); - defer mutex.unlock(store_io); + db_mutex.lockUncancelable(store_io); + defer db_mutex.unlock(store_io); try requireInitialized(); const seq = try nextSeqLocked(); @@ -1384,8 +1404,8 @@ pub fn importRepo( } pub fn writeRepoCar(allocator: std.mem.Allocator, did: []const u8) ![]const u8 { - mutex.lockUncancelable(store_io); - defer mutex.unlock(store_io); + db_mutex.lockUncancelable(store_io); + defer db_mutex.unlock(store_io); try requireInitialized(); const root = try latestRootLocked(allocator, did); @@ -1419,8 +1439,8 @@ pub fn writeRepoCar(allocator: std.mem.Allocator, did: []const u8) ![]const u8 { } pub fn writeRepoListJson(allocator: std.mem.Allocator, limit: usize) ![]const u8 { - mutex.lockUncancelable(store_io); - defer mutex.unlock(store_io); + db_mutex.lockUncancelable(store_io); + defer db_mutex.unlock(store_io); try requireInitialized(); const actual_limit = if (limit == 0) 500 else @min(limit, 1000); @@ -1466,8 +1486,8 @@ pub fn writeRepoListJson(allocator: std.mem.Allocator, limit: usize) ![]const u8 } pub fn writeLatestCommitJson(allocator: std.mem.Allocator, did: []const u8) ![]const u8 { - mutex.lockUncancelable(store_io); - defer mutex.unlock(store_io); + db_mutex.lockUncancelable(store_io); + defer db_mutex.unlock(store_io); try requireInitialized(); const root = try latestRootLocked(allocator, did); return std.fmt.allocPrint( @@ -1478,8 +1498,8 @@ pub fn writeLatestCommitJson(allocator: std.mem.Allocator, did: []const u8) ![]c } pub fn listSeqEvents(allocator: std.mem.Allocator, cursor: u64, limit: usize) ![]SeqEvent { - mutex.lockUncancelable(store_io); - defer mutex.unlock(store_io); + db_mutex.lockUncancelable(store_io); + defer db_mutex.unlock(store_io); try requireInitialized(); try backfillSeqEventsLocked(allocator); @@ -1515,8 +1535,8 @@ fn insertSeqEventLocked(seq: u64, did: []const u8, commit_cid: []const u8, frame } pub fn writeRepoStatusJson(allocator: std.mem.Allocator, did: []const u8) ![]const u8 { - mutex.lockUncancelable(store_io); - defer mutex.unlock(store_io); + db_mutex.lockUncancelable(store_io); + defer db_mutex.unlock(store_io); try requireInitialized(); const root = try latestRootLocked(allocator, did); const active = try accountActiveLocked(did); @@ -1547,8 +1567,8 @@ pub fn writeListJson( collection: []const u8, limit: usize, ) ![]const u8 { - mutex.lockUncancelable(store_io); - defer mutex.unlock(store_io); + db_mutex.lockUncancelable(store_io); + defer db_mutex.unlock(store_io); try requireInitialized(); var rows = try conn.rows( @@ -1710,8 +1730,8 @@ fn blobTableHasDataColumn() !bool { } pub fn getEmailInfo(allocator: std.mem.Allocator, did: []const u8) ?EmailInfo { - mutex.lockUncancelable(store_io); - defer mutex.unlock(store_io); + db_mutex.lockUncancelable(store_io); + defer db_mutex.unlock(store_io); requireInitialized() catch return null; const row = conn.row( \\SELECT email, email_confirmed_at, auth_code, auth_code_expires_at, pending_email @@ -1730,8 +1750,8 @@ pub fn getEmailInfo(allocator: std.mem.Allocator, did: []const u8) ?EmailInfo { } pub fn setAuthCode(did: []const u8, code: []const u8, expires_at_ms: i64) !void { - mutex.lockUncancelable(store_io); - defer mutex.unlock(store_io); + db_mutex.lockUncancelable(store_io); + defer db_mutex.unlock(store_io); try requireInitialized(); try conn.exec( \\UPDATE accounts @@ -1741,8 +1761,8 @@ pub fn setAuthCode(did: []const u8, code: []const u8, expires_at_ms: i64) !void } pub fn setPendingEmail(did: []const u8, pending_email: []const u8, code: []const u8, expires_at_ms: i64) !void { - mutex.lockUncancelable(store_io); - defer mutex.unlock(store_io); + db_mutex.lockUncancelable(store_io); + defer db_mutex.unlock(store_io); try requireInitialized(); try conn.exec( \\UPDATE accounts @@ -1752,8 +1772,8 @@ pub fn setPendingEmail(did: []const u8, pending_email: []const u8, code: []const } pub fn validateAuthCode(did: []const u8, code: []const u8, now_ms: i64) CodeStatus { - mutex.lockUncancelable(store_io); - defer mutex.unlock(store_io); + db_mutex.lockUncancelable(store_io); + defer db_mutex.unlock(store_io); requireInitialized() catch return .invalid; const row = conn.row( \\SELECT auth_code, auth_code_expires_at @@ -1770,8 +1790,8 @@ pub fn validateAuthCode(did: []const u8, code: []const u8, now_ms: i64) CodeStat } pub fn clearAuthCode(did: []const u8) !void { - mutex.lockUncancelable(store_io); - defer mutex.unlock(store_io); + db_mutex.lockUncancelable(store_io); + defer db_mutex.unlock(store_io); try requireInitialized(); try conn.exec( \\UPDATE accounts @@ -1782,8 +1802,8 @@ pub fn clearAuthCode(did: []const u8) !void { } pub fn confirmEmail(did: []const u8) !void { - mutex.lockUncancelable(store_io); - defer mutex.unlock(store_io); + db_mutex.lockUncancelable(store_io); + defer db_mutex.unlock(store_io); try requireInitialized(); try conn.exec( \\UPDATE accounts @@ -1795,8 +1815,8 @@ pub fn confirmEmail(did: []const u8) !void { } pub fn updateEmail(did: []const u8, email: []const u8) !void { - mutex.lockUncancelable(store_io); - defer mutex.unlock(store_io); + db_mutex.lockUncancelable(store_io); + defer db_mutex.unlock(store_io); try requireInitialized(); try conn.exec( \\UPDATE accounts @@ -1809,7 +1829,7 @@ pub fn updateEmail(did: []const u8, email: []const u8) !void { , .{ email, did }); } -fn nextSeqLocked() !u64 { +fn loadNextSeqLocked() !u64 { const row = try conn.row( \\SELECT COALESCE(MAX(seq), 0) + 1 \\FROM ( @@ -1823,9 +1843,14 @@ fn nextSeqLocked() !u64 { return @intCast(row.?.int(0)); } +fn nextSeqLocked() !u64 { + _ = try requireInitialized(); + return next_seq.fetchAdd(1, .monotonic); +} + fn nextRkey(allocator: std.mem.Allocator) ![]const u8 { - mutex.lockUncancelable(store_io); - defer mutex.unlock(store_io); + db_mutex.lockUncancelable(store_io); + defer db_mutex.unlock(store_io); try requireInitialized(); return nextRkeyLocked(allocator); } @@ -1836,6 +1861,22 @@ fn nextRkeyLocked(allocator: std.mem.Allocator) ![]const u8 { return allocator.dupe(u8, &tid); } +fn resolveWriteOpsLocked(allocator: std.mem.Allocator, ops: []const WriteOp) ![]WriteOp { + var resolved = try allocator.alloc(WriteOp, ops.len); + for (ops, 0..) |op, i| { + resolved[i] = switch (op) { + .create => |create_op| .{ .create = .{ + .collection = create_op.collection, + .rkey = create_op.rkey orelse try nextRkeyLocked(allocator), + .value = create_op.value, + } }, + .update => |update_op| .{ .update = update_op }, + .delete => |delete_op| .{ .delete = delete_op }, + }; + } + return resolved; +} + fn validateRecordForWrite(collection: []const u8, rkey: ?[]const u8, value: std.json.Value) Error!void { const object = switch (value) { .object => |object| object, @@ -2104,6 +2145,18 @@ fn signedCommit( rev: []const u8, data_cid: zat.cbor.Cid, prev_cid_raw: ?[]const u8, +) !SignedCommit { + var keypair = try signingKeypairLocked(did); + return signedCommitWithKeypair(allocator, did, rev, data_cid, prev_cid_raw, &keypair); +} + +fn signedCommitWithKeypair( + allocator: std.mem.Allocator, + did: []const u8, + rev: []const u8, + data_cid: zat.cbor.Cid, + prev_cid_raw: ?[]const u8, + keypair: *const zat.Keypair, ) !SignedCommit { var unsigned_entries = try allocator.alloc(zat.cbor.Value.MapEntry, 5); unsigned_entries[0] = .{ .key = "did", .value = .{ .text = did } }; @@ -2113,7 +2166,6 @@ fn signedCommit( unsigned_entries[4] = .{ .key = "prev", .value = if (prev_cid_raw) |raw| .{ .cid = .{ .raw = raw } } else .null }; const unsigned_value: zat.cbor.Value = .{ .map = unsigned_entries }; const unsigned_bytes = try zat.cbor.encodeAlloc(allocator, unsigned_value); - var keypair = try signingKeypairLocked(did); const sig = try keypair.sign(unsigned_bytes); var signed_entries = try allocator.alloc(zat.cbor.Value.MapEntry, 6);