+11 -12

bench/README.md

Reviewed

··· 65 65 record, get one record CID/record, and list records from a seeded repo. The 66 66 implementations are not identical internally. Tranquil measures its metastore 67 67 handler pool and eventlog path; ZDS measures the local store API, including repo 68 68 - commit construction and the current process-wide store mutex. 68 68 + commit construction, per-DID write lanes, and the shared SQLite connection. 69 69 70 70 | operation | callers | ops | zds | tranquil | 71 71 |---|---:|---:|---:|---:| 72 72 | apply/write | 1 | 1000 | 206 ops/s, p95 11.9 ms | not measured at this count | 73 73 | apply/write | 1 | 5000 | not rerun | 292 ops/s, p95 5.8 ms | 74 74 - | apply/write | 10 | 10000 | 106 ops/s, p95 119 ms | 436 ops/s, p95 36.0 ms | 75 75 - | apply/write | 100 | 20000 | 125 ops/s, p95 110 ms | 944 ops/s, p95 119 ms | 74 74 + | apply/write | 10 | 10000 | 303 ops/s, p95 88 ms | 436 ops/s, p95 36.0 ms | 75 75 + | apply/write | 100 | 20000 | 1287 ops/s, p95 72 ms | 944 ops/s, p95 119 ms | 76 76 | apply/write | 1000 | 50000 | not rerun | 1174 ops/s, then read backpressure | 77 77 | get record | 1 | 1000 | 55k ops/s, p95 25 us | not measured at this count | 78 78 | get record | 10 | 10000 | 56k ops/s, p95 64 us | 571k ops/s, p95 31 us | ··· 86 86 87 87 ## write profile 88 88 89 89 - `write-profile` shows that the current ZDS write path is dominated by serialized 90 90 - time around the store mutex, not SQLite alone: 89 89 + `write-profile` tracks where concurrent writes spend time: 91 90 92 91 | callers | ops | throughput | lock wait | load repo | build commit | sqlite/event | 93 92 |---:|---:|---:|---:|---:|---:|---:| 94 94 - | 10 | 500 | 2386 ops/s | 83% | 3% | 8% | 6% | 95 95 - | 10 | 5000 | 353 ops/s | 82% | 13% | 2% | 3% | 96 96 - | 100 | 2000 | 1310 ops/s | 98% | 1% | 1% | 1% | 93 93 + | 10 | 5000 | 708 ops/s | 25% | 26% | 3% | 46% | 94 94 + | 100 | 2000 | 3683 ops/s | 64% | 10% | 1% | 25% | 97 95 98 98 - The next write-path work should shrink or restructure the serialized section: 99 99 - keep validation and request parsing outside the store lock, move toward a 100 100 - Tranquil-style write worker/queue for repo mutation, and avoid rebuilding more 101 101 - repo state than a single write needs. 96 96 + ZDS now separates repo ordering from database serialization: writes take a 97 97 + per-DID lane before loading and mutating the repo, then use the DB mutex for the 98 98 + shared SQLite connection. The next write-path work should continue toward the 99 99 + Tranquil shape: bounded write queues with explicit backpressure, less full-repo 100 100 + state loading, and a narrower SQLite persistence phase. 102 101 103 102 ## comparison work 104 103

+2

docs/operations.md

Reviewed

··· 24 24 just bench read 25 25 just bench repo 26 26 just bench blob 27 27 + just bench metastore 10 1000 28 28 + just bench write-profile 10 500 27 29 just bench run --records 1000 28 30 ``` 29 31

+38

src/internal/sharded_locks.zig

Reviewed

··· 1 1 + const std = @import("std"); 2 2 + 3 3 + const Io = std.Io; 4 4 + 5 5 + pub fn ShardedLocks(comptime count: usize) type { 6 6 + if (count == 0) @compileError("ShardedLocks requires at least one shard"); 7 7 + 8 8 + return struct { 9 9 + const Self = @This(); 10 10 + 11 11 + locks: [count]Io.Mutex = [_]Io.Mutex{.init} ** count, 12 12 + 13 13 + pub const Guard = struct { 14 14 + owner: *Self, 15 15 + io: Io, 16 16 + index: usize, 17 17 + 18 18 + pub fn unlock(self: Guard) void { 19 19 + self.owner.locks[self.index].unlock(self.io); 20 20 + } 21 21 + }; 22 22 + 23 23 + pub fn lock(self: *Self, io: Io, key: []const u8) Guard { 24 24 + const index = shardIndex(key); 25 25 + self.locks[index].lockUncancelable(io); 26 26 + return .{ .owner = self, .io = io, .index = index }; 27 27 + } 28 28 + 29 29 + pub fn shardIndex(key: []const u8) usize { 30 30 + return @intCast(std.hash.Wyhash.hash(0, key) % count); 31 31 + } 32 32 + }; 33 33 + } 34 34 + 35 35 + test "same key maps to same shard" { 36 36 + const Locks = ShardedLocks(16); 37 37 + try std.testing.expectEqual(Locks.shardIndex("did:plc:abc"), Locks.shardIndex("did:plc:abc")); 38 38 + }

+1

src/root.zig

Reviewed

··· 31 31 32 32 pub const internal = struct { 33 33 pub const cli = @import("internal/cli.zig"); 34 34 + pub const sharded_locks = @import("internal/sharded_locks.zig"); 34 35 }; 35 36 36 37 pub const storage = struct {

+172 -120

src/storage/store.zig

Reviewed

··· 3 3 const auth = @import("../auth/tokens.zig"); 4 4 const blobstore = @import("blobstore.zig"); 5 5 const eventlog = @import("eventlog.zig"); 6 6 + const sharded_locks = @import("../internal/sharded_locks.zig"); 6 7 const zat = @import("zat"); 7 8 const zqlite = @import("zqlite"); 8 9 const Io = std.Io; ··· 169 170 var conn: zqlite.Conn = undefined; 170 171 var initialized = false; 171 172 var store_io: Io = undefined; 172 172 - var mutex: Io.Mutex = .init; 173 173 + var db_mutex: Io.Mutex = .init; 174 174 + var write_lanes: sharded_locks.ShardedLocks(32) = .{}; 175 175 + var next_seq: std.atomic.Value(u64) = .init(1); 173 176 174 177 pub fn init(io: Io, path: []const u8) !void { 175 178 if (initialized) return; ··· 187 190 try conn.execNoArgs("PRAGMA journal_mode=WAL"); 188 191 try conn.execNoArgs("PRAGMA foreign_keys=ON"); 189 192 try migrate(); 193 193 + next_seq.store(try loadNextSeqLocked(), .release); 190 194 } 191 195 192 196 pub fn close() void { ··· 220 224 } 221 225 222 226 pub fn findAccount(allocator: std.mem.Allocator, identifier: []const u8) !?auth.Account { 223 223 - mutex.lockUncancelable(store_io); 224 224 - defer mutex.unlock(store_io); 227 227 + db_mutex.lockUncancelable(store_io); 228 228 + defer db_mutex.unlock(store_io); 225 229 try requireInitialized(); 226 230 227 231 return try findAccountLocked(allocator, identifier); 228 232 } 229 233 230 234 pub fn searchAccounts(allocator: std.mem.Allocator, query: []const u8, limit: usize) ![]auth.Account { 231 231 - mutex.lockUncancelable(store_io); 232 232 - defer mutex.unlock(store_io); 235 235 + db_mutex.lockUncancelable(store_io); 236 236 + defer db_mutex.unlock(store_io); 233 237 try requireInitialized(); 234 238 235 239 const actual_limit = if (limit == 0) 25 else limit; ··· 260 264 } 261 265 262 266 pub fn listResidents(allocator: std.mem.Allocator, limit: usize) ![]Resident { 263 263 - mutex.lockUncancelable(store_io); 264 264 - defer mutex.unlock(store_io); 267 267 + db_mutex.lockUncancelable(store_io); 268 268 + defer db_mutex.unlock(store_io); 265 269 try requireInitialized(); 266 270 267 271 const actual_limit = if (limit == 0) 24 else @min(limit, 100); ··· 300 304 } 301 305 302 306 pub fn profileAvatarCid(allocator: std.mem.Allocator, did: []const u8) !?[]const u8 { 303 303 - mutex.lockUncancelable(store_io); 304 304 - defer mutex.unlock(store_io); 307 307 + db_mutex.lockUncancelable(store_io); 308 308 + defer db_mutex.unlock(store_io); 305 309 try requireInitialized(); 306 310 307 311 const row = try conn.row( ··· 334 338 } 335 339 336 340 pub fn listCollectionSummaries(allocator: std.mem.Allocator, limit: usize) ![]CollectionSummary { 337 337 - mutex.lockUncancelable(store_io); 338 338 - defer mutex.unlock(store_io); 341 341 + db_mutex.lockUncancelable(store_io); 342 342 + defer db_mutex.unlock(store_io); 339 343 try requireInitialized(); 340 344 341 345 const actual_limit = if (limit == 0) 12 else @min(limit, 50); ··· 362 366 } 363 367 364 368 pub fn listLandingRecentRecords(allocator: std.mem.Allocator, limit: usize) ![]RecentRecord { 365 365 - mutex.lockUncancelable(store_io); 366 366 - defer mutex.unlock(store_io); 369 369 + db_mutex.lockUncancelable(store_io); 370 370 + defer db_mutex.unlock(store_io); 367 371 try requireInitialized(); 368 372 369 373 const actual_limit = if (limit == 0) 8 else @min(limit, 24); ··· 434 438 activated: bool, 435 439 signing_key: [32]u8, 436 440 ) !auth.Account { 437 437 - mutex.lockUncancelable(store_io); 438 438 - defer mutex.unlock(store_io); 441 441 + db_mutex.lockUncancelable(store_io); 442 442 + defer db_mutex.unlock(store_io); 439 443 try requireInitialized(); 440 444 441 445 var salt: [16]u8 = undefined; ··· 462 466 } 463 467 464 468 pub fn generateAccountSigningKey() ![32]u8 { 465 465 - mutex.lockUncancelable(store_io); 466 466 - defer mutex.unlock(store_io); 469 469 + db_mutex.lockUncancelable(store_io); 470 470 + defer db_mutex.unlock(store_io); 467 471 try requireInitialized(); 468 472 return generateSigningKey(); 469 473 } 470 474 471 475 pub fn signingKeypair(did: []const u8) !zat.Keypair { 472 472 - mutex.lockUncancelable(store_io); 473 473 - defer mutex.unlock(store_io); 476 476 + db_mutex.lockUncancelable(store_io); 477 477 + defer db_mutex.unlock(store_io); 474 478 try requireInitialized(); 475 479 return signingKeypairLocked(did); 476 480 } 477 481 478 482 pub fn setAccountActive(did: []const u8, active: bool) !void { 479 479 - mutex.lockUncancelable(store_io); 480 480 - defer mutex.unlock(store_io); 483 483 + db_mutex.lockUncancelable(store_io); 484 484 + defer db_mutex.unlock(store_io); 481 485 try requireInitialized(); 482 486 if (active) { 483 487 try conn.exec( ··· 496 500 } 497 501 498 502 pub fn sequenceAccountEvent(allocator: std.mem.Allocator, did: []const u8, active: bool) !void { 499 499 - mutex.lockUncancelable(store_io); 500 500 - defer mutex.unlock(store_io); 503 503 + db_mutex.lockUncancelable(store_io); 504 504 + defer db_mutex.unlock(store_io); 501 505 try requireInitialized(); 502 506 const seq = try nextSeqLocked(); 503 507 const frame = try accountEventFrame(allocator, seq, did, active); ··· 506 510 } 507 511 508 512 pub fn sequenceIdentityEvent(allocator: std.mem.Allocator, did: []const u8, handle: []const u8) !void { 509 509 - mutex.lockUncancelable(store_io); 510 510 - defer mutex.unlock(store_io); 513 513 + db_mutex.lockUncancelable(store_io); 514 514 + defer db_mutex.unlock(store_io); 511 515 try requireInitialized(); 512 516 const seq = try nextSeqLocked(); 513 517 const frame = try identityEventFrame(allocator, seq, did, handle); ··· 516 520 } 517 521 518 522 pub fn sequenceSyncEvent(allocator: std.mem.Allocator, did: []const u8) !void { 519 519 - mutex.lockUncancelable(store_io); 520 520 - defer mutex.unlock(store_io); 523 523 + db_mutex.lockUncancelable(store_io); 524 524 + defer db_mutex.unlock(store_io); 521 525 try requireInitialized(); 522 526 const seq = try nextSeqLocked(); 523 527 const commit = try latestCommitRawLocked(allocator, did) orelse return Error.RepoNotFound; ··· 527 531 } 528 532 529 533 pub fn isAccountActive(did: []const u8) bool { 530 530 - mutex.lockUncancelable(store_io); 531 531 - defer mutex.unlock(store_io); 534 534 + db_mutex.lockUncancelable(store_io); 535 535 + defer db_mutex.unlock(store_io); 532 536 requireInitialized() catch return false; 533 537 return accountActiveLocked(did) catch false; 534 538 } ··· 563 567 dpop_jkt: ?[]const u8, 564 568 expires_at: i64, 565 569 ) !void { 566 566 - mutex.lockUncancelable(store_io); 567 567 - defer mutex.unlock(store_io); 570 570 + db_mutex.lockUncancelable(store_io); 571 571 + defer db_mutex.unlock(store_io); 568 572 try requireInitialized(); 569 573 try conn.exec( 570 574 \\INSERT INTO oauth_requests ··· 574 578 } 575 579 576 580 pub fn getOAuthRequest(allocator: std.mem.Allocator, request_id: []const u8) !?OAuthRequest { 577 577 - mutex.lockUncancelable(store_io); 578 578 - defer mutex.unlock(store_io); 581 581 + db_mutex.lockUncancelable(store_io); 582 582 + defer db_mutex.unlock(store_io); 579 583 try requireInitialized(); 580 584 const row = try conn.row( 581 585 \\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 592 } 589 593 590 594 pub fn authorizeOAuthRequest(request_id: []const u8, did: []const u8, code: []const u8) !void { 591 591 - mutex.lockUncancelable(store_io); 592 592 - defer mutex.unlock(store_io); 595 595 + db_mutex.lockUncancelable(store_io); 596 596 + defer db_mutex.unlock(store_io); 593 597 try requireInitialized(); 594 598 try conn.exec( 595 599 \\UPDATE oauth_requests ··· 599 603 } 600 604 601 605 pub fn consumeOAuthCode(allocator: std.mem.Allocator, code: []const u8) !?OAuthRequest { 602 602 - mutex.lockUncancelable(store_io); 603 603 - defer mutex.unlock(store_io); 606 606 + db_mutex.lockUncancelable(store_io); 607 607 + defer db_mutex.unlock(store_io); 604 608 try requireInitialized(); 605 609 const row = try conn.row( 606 610 \\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 626 refresh_token: []const u8, 623 627 expires_at: i64, 624 628 ) !void { 625 625 - mutex.lockUncancelable(store_io); 626 626 - defer mutex.unlock(store_io); 629 629 + db_mutex.lockUncancelable(store_io); 630 630 + defer db_mutex.unlock(store_io); 627 631 try requireInitialized(); 628 632 try conn.exec( 629 633 \\INSERT INTO oauth_tokens (access_token, refresh_token, did, client_id, scope, expires_at) ··· 632 636 } 633 637 634 638 pub fn getOAuthToken(allocator: std.mem.Allocator, token: []const u8) !?OAuthToken { 635 635 - mutex.lockUncancelable(store_io); 636 636 - defer mutex.unlock(store_io); 639 639 + db_mutex.lockUncancelable(store_io); 640 640 + defer db_mutex.unlock(store_io); 637 641 try requireInitialized(); 638 642 const row = try conn.row( 639 643 \\SELECT did, client_id, scope, access_token, refresh_token, expires_at, revoked_at ··· 656 660 } 657 661 658 662 pub fn revokeOAuthToken(token: []const u8) !void { 659 659 - mutex.lockUncancelable(store_io); 660 660 - defer mutex.unlock(store_io); 663 663 + db_mutex.lockUncancelable(store_io); 664 664 + defer db_mutex.unlock(store_io); 661 665 try requireInitialized(); 662 666 try conn.exec( 663 667 \\UPDATE oauth_tokens ··· 723 727 addProfile(profile, .validation_ns, elapsedNs(validation_start)); 724 728 725 729 const lock_start = monotonicNs(); 726 726 - mutex.lockUncancelable(store_io); 730 730 + const lane = write_lanes.lock(store_io, account.did); 731 731 + defer lane.unlock(); 727 732 addProfile(profile, .lock_wait_ns, elapsedNs(lock_start)); 728 728 - defer mutex.unlock(store_io); 733 733 + 729 734 try requireInitialized(); 735 735 + const seq = try nextSeqLocked(); 736 736 + const load_start = monotonicNs(); 737 737 + const loaded = blk: { 738 738 + db_mutex.lockUncancelable(store_io); 739 739 + defer db_mutex.unlock(store_io); 740 740 + const current = try latestCommitRawLocked(allocator, account.did); 741 741 + break :blk .{ 742 742 + .current = current, 743 743 + .rev = try revForSeq(allocator, seq, if (current) |root| root.rev else null), 744 744 + .repo_car = try readRepoCarLocked(allocator, account.did), 745 745 + .keypair = try signingKeypairLocked(account.did), 746 746 + .ops = try resolveWriteOpsLocked(allocator, ops), 747 747 + }; 748 748 + }; 749 749 + const current = loaded.current; 750 750 + const rev = loaded.rev; 751 751 + const repo_car = loaded.repo_car; 752 752 + const keypair = loaded.keypair; 753 753 + const resolved_ops = loaded.ops; 730 754 731 731 - const load_start = monotonicNs(); 732 732 - const seq = try nextSeqLocked(); 733 733 - const current = try latestCommitRawLocked(allocator, account.did); 734 734 - const rev = try revForSeq(allocator, seq, if (current) |root| root.rev else null); 735 735 - const repo_car = try readRepoCarLocked(allocator, account.did); 736 755 var tree = if (current) |root| 737 756 try zat.mst.Mst.loadFromBlocks(allocator, repo_car, root.data_cid_raw) 738 757 else ··· 745 764 var blob_refs: std.ArrayList(BlobRef) = .empty; 746 765 747 766 const stage_start = monotonicNs(); 748 748 - for (ops) |op| switch (op) { 767 767 + for (resolved_ops) |op| switch (op) { 749 768 .create => |create_op| { 750 750 - const rkey = create_op.rkey orelse try nextRkeyLocked(allocator); 769 769 + const rkey = create_op.rkey orelse return Error.InvalidRecordKey; 751 770 const record = try stageRecordWrite(allocator, &tree, account, create_op.collection, rkey, create_op.value, rev, seq, &record_blocks, &blob_refs); 752 771 try records.append(allocator, record); 753 772 }, ··· 765 784 const build_start = monotonicNs(); 766 785 const data_cid = try tree.rootCid(); 767 786 try writeMstBlocks(allocator, &tree, &mst_blocks); 768 768 - const commit = try signedCommit(allocator, account.did, rev, data_cid, if (current) |root| root.commit_cid_raw else null); 787 787 + const commit = try signedCommitWithKeypair(allocator, account.did, rev, data_cid, if (current) |root| root.commit_cid_raw else null, &keypair); 769 788 addProfile(profile, .build_commit_ns, elapsedNs(build_start)); 770 789 771 790 const sql_start = monotonicNs(); 791 791 + db_mutex.lockUncancelable(store_io); 792 792 + defer db_mutex.unlock(store_io); 772 793 try conn.exclusiveTransaction(); 773 794 errdefer conn.rollback(); 774 795 for (record_blocks.items) |block| { ··· 791 812 \\ON CONFLICT(did, cid) DO UPDATE SET data = excluded.data 792 813 , .{ account.did, commit.cid, zqlite.blob(commit.data) }); 793 814 794 794 - for (ops) |op| switch (op) { 815 815 + for (resolved_ops) |op| switch (op) { 795 816 .delete => |delete_op| { 796 817 const uri = try std.fmt.allocPrint(allocator, "at://{s}/{s}/{s}", .{ account.did, delete_op.collection, delete_op.rkey }); 797 818 try conn.exec("DELETE FROM expected_blobs WHERE record_uri = ?", .{uri}); ··· 834 855 commit.data, 835 856 record_blocks.items, 836 857 mst_blocks.items, 837 837 - ops, 858 858 + resolved_ops, 838 859 records.items, 839 860 ); 840 861 try conn.exec( ··· 858 879 .records = try records.toOwnedSlice(allocator), 859 880 }; 860 881 } 861 861 - 862 882 const ProfileField = enum { 863 883 validation_ns, 864 884 lock_wait_ns, ··· 920 940 } 921 941 922 942 pub fn get(did: []const u8, collection: []const u8, rkey: []const u8) ?Record { 923 923 - mutex.lockUncancelable(store_io); 924 924 - defer mutex.unlock(store_io); 943 943 + db_mutex.lockUncancelable(store_io); 944 944 + defer db_mutex.unlock(store_io); 925 945 requireInitialized() catch return null; 926 946 927 947 const row = conn.row( ··· 935 955 } 936 956 937 957 pub fn getByUri(uri: []const u8) ?Record { 938 938 - mutex.lockUncancelable(store_io); 939 939 - defer mutex.unlock(store_io); 958 958 + db_mutex.lockUncancelable(store_io); 959 959 + defer db_mutex.unlock(store_io); 940 960 requireInitialized() catch return null; 941 961 return getByUriLocked(uri); 942 962 } 943 963 944 964 pub fn listRecentRecords(allocator: std.mem.Allocator, limit: usize) ![]Record { 945 945 - mutex.lockUncancelable(store_io); 946 946 - defer mutex.unlock(store_io); 965 965 + db_mutex.lockUncancelable(store_io); 966 966 + defer db_mutex.unlock(store_io); 947 967 try requireInitialized(); 948 968 949 969 var rows = try conn.rows( ··· 968 988 collection: []const u8, 969 989 limit: usize, 970 990 ) ![]Record { 971 971 - mutex.lockUncancelable(store_io); 972 972 - defer mutex.unlock(store_io); 991 991 + db_mutex.lockUncancelable(store_io); 992 992 + defer db_mutex.unlock(store_io); 973 993 try requireInitialized(); 974 994 975 995 var rows = try conn.rows( ··· 995 1015 needle: []const u8, 996 1016 limit: usize, 997 1017 ) ![]Record { 998 998 - mutex.lockUncancelable(store_io); 999 999 - defer mutex.unlock(store_io); 1018 1018 + db_mutex.lockUncancelable(store_io); 1019 1019 + defer db_mutex.unlock(store_io); 1000 1020 try requireInitialized(); 1001 1021 1002 1022 var rows = try conn.rows( ··· 1023 1043 needle: []const u8, 1024 1044 limit: usize, 1025 1045 ) ![]Record { 1026 1026 - mutex.lockUncancelable(store_io); 1027 1027 - defer mutex.unlock(store_io); 1046 1046 + db_mutex.lockUncancelable(store_io); 1047 1047 + defer db_mutex.unlock(store_io); 1028 1048 try requireInitialized(); 1029 1049 1030 1050 var rows = try conn.rows( ··· 1068 1088 } 1069 1089 1070 1090 pub fn count(did: []const u8, collection: []const u8) usize { 1071 1071 - mutex.lockUncancelable(store_io); 1072 1072 - defer mutex.unlock(store_io); 1091 1091 + db_mutex.lockUncancelable(store_io); 1092 1092 + defer db_mutex.unlock(store_io); 1073 1093 requireInitialized() catch return 0; 1074 1094 const row = conn.row("SELECT count(*) FROM records WHERE did = ? AND collection = ?", .{ did, collection }) catch return 0; 1075 1095 if (row == null) return 0; ··· 1078 1098 } 1079 1099 1080 1100 pub fn countSubject(collection: []const u8, subject_did: []const u8) usize { 1081 1081 - mutex.lockUncancelable(store_io); 1082 1082 - defer mutex.unlock(store_io); 1101 1101 + db_mutex.lockUncancelable(store_io); 1102 1102 + defer db_mutex.unlock(store_io); 1083 1103 requireInitialized() catch return 0; 1084 1104 const row = conn.row("SELECT count(*) FROM records WHERE collection = ? AND instr(value_json, ?) > 0", .{ collection, subject_did }) catch return 0; 1085 1105 if (row == null) return 0; ··· 1088 1108 } 1089 1109 1090 1110 pub fn listCollectionsJson(allocator: std.mem.Allocator, did: []const u8) ![]const u8 { 1091 1091 - mutex.lockUncancelable(store_io); 1092 1092 - defer mutex.unlock(store_io); 1111 1111 + db_mutex.lockUncancelable(store_io); 1112 1112 + defer db_mutex.unlock(store_io); 1093 1113 try requireInitialized(); 1094 1114 1095 1115 var rows = try conn.rows( ··· 1115 1135 } 1116 1136 1117 1137 pub fn getAppPreferences(allocator: std.mem.Allocator, did: []const u8, namespace: []const u8) ![]AppPreference { 1118 1118 - mutex.lockUncancelable(store_io); 1119 1119 - defer mutex.unlock(store_io); 1138 1138 + db_mutex.lockUncancelable(store_io); 1139 1139 + defer db_mutex.unlock(store_io); 1120 1140 try requireInitialized(); 1121 1141 1122 1142 const like_pattern = try std.fmt.allocPrint(allocator, "{s}.%", .{namespace}); ··· 1145 1165 namespace: []const u8, 1146 1166 preferences: []const AppPreference, 1147 1167 ) !void { 1148 1148 - mutex.lockUncancelable(store_io); 1149 1149 - defer mutex.unlock(store_io); 1168 1168 + db_mutex.lockUncancelable(store_io); 1169 1169 + defer db_mutex.unlock(store_io); 1150 1170 try requireInitialized(); 1151 1171 const like_pattern = try std.fmt.allocPrint(allocator, "{s}.%", .{namespace}); 1152 1172 try conn.exclusiveTransaction(); ··· 1173 1193 ) ![]const u8 { 1174 1194 const cid = try cidForBlob(allocator, data); 1175 1195 1176 1176 - mutex.lockUncancelable(store_io); 1177 1177 - defer mutex.unlock(store_io); 1196 1196 + db_mutex.lockUncancelable(store_io); 1197 1197 + defer db_mutex.unlock(store_io); 1178 1198 try requireInitialized(); 1179 1199 try blobstore.put(allocator, io, account.did, cid, data); 1180 1200 try conn.exec( ··· 1192 1212 did: []const u8, 1193 1213 cid: []const u8, 1194 1214 ) ?BlobRecord { 1195 1195 - mutex.lockUncancelable(store_io); 1196 1196 - defer mutex.unlock(store_io); 1215 1215 + db_mutex.lockUncancelable(store_io); 1216 1216 + defer db_mutex.unlock(store_io); 1197 1217 requireInitialized() catch return null; 1198 1218 const row = conn.row( 1199 1219 \\SELECT mime_type, size ··· 1210 1230 } 1211 1231 1212 1232 pub fn writeBlobListJson(allocator: std.mem.Allocator, did: []const u8, limit: usize) ![]const u8 { 1213 1213 - mutex.lockUncancelable(store_io); 1214 1214 - defer mutex.unlock(store_io); 1233 1233 + db_mutex.lockUncancelable(store_io); 1234 1234 + defer db_mutex.unlock(store_io); 1215 1235 try requireInitialized(); 1216 1236 1217 1237 var rows = try conn.rows( ··· 1243 1263 } 1244 1264 1245 1265 pub fn writeMissingBlobsJson(allocator: std.mem.Allocator, did: []const u8, limit: usize) ![]const u8 { 1246 1246 - mutex.lockUncancelable(store_io); 1247 1247 - defer mutex.unlock(store_io); 1266 1266 + db_mutex.lockUncancelable(store_io); 1267 1267 + defer db_mutex.unlock(store_io); 1248 1268 try requireInitialized(); 1249 1269 1250 1270 var rows = try conn.rows( ··· 1281 1301 } 1282 1302 1283 1303 pub fn writeAccountStatusJson(allocator: std.mem.Allocator, did: []const u8) ![]const u8 { 1284 1284 - mutex.lockUncancelable(store_io); 1285 1285 - defer mutex.unlock(store_io); 1304 1304 + db_mutex.lockUncancelable(store_io); 1305 1305 + defer db_mutex.unlock(store_io); 1286 1306 try requireInitialized(); 1287 1307 1288 1308 const record_count = try scalarCountLocked( ··· 1332 1352 records: []const ImportedRecord, 1333 1353 blocks: []const ImportedBlock, 1334 1354 ) !void { 1335 1335 - mutex.lockUncancelable(store_io); 1336 1336 - defer mutex.unlock(store_io); 1355 1355 + db_mutex.lockUncancelable(store_io); 1356 1356 + defer db_mutex.unlock(store_io); 1337 1357 try requireInitialized(); 1338 1358 1339 1359 const seq = try nextSeqLocked(); ··· 1384 1404 } 1385 1405 1386 1406 pub fn writeRepoCar(allocator: std.mem.Allocator, did: []const u8) ![]const u8 { 1387 1387 - mutex.lockUncancelable(store_io); 1388 1388 - defer mutex.unlock(store_io); 1407 1407 + db_mutex.lockUncancelable(store_io); 1408 1408 + defer db_mutex.unlock(store_io); 1389 1409 try requireInitialized(); 1390 1410 1391 1411 const root = try latestRootLocked(allocator, did); ··· 1419 1439 } 1420 1440 1421 1441 pub fn writeRepoListJson(allocator: std.mem.Allocator, limit: usize) ![]const u8 { 1422 1422 - mutex.lockUncancelable(store_io); 1423 1423 - defer mutex.unlock(store_io); 1442 1442 + db_mutex.lockUncancelable(store_io); 1443 1443 + defer db_mutex.unlock(store_io); 1424 1444 try requireInitialized(); 1425 1445 1426 1446 const actual_limit = if (limit == 0) 500 else @min(limit, 1000); ··· 1466 1486 } 1467 1487 1468 1488 pub fn writeLatestCommitJson(allocator: std.mem.Allocator, did: []const u8) ![]const u8 { 1469 1469 - mutex.lockUncancelable(store_io); 1470 1470 - defer mutex.unlock(store_io); 1489 1489 + db_mutex.lockUncancelable(store_io); 1490 1490 + defer db_mutex.unlock(store_io); 1471 1491 try requireInitialized(); 1472 1492 const root = try latestRootLocked(allocator, did); 1473 1493 return std.fmt.allocPrint( ··· 1478 1498 } 1479 1499 1480 1500 pub fn listSeqEvents(allocator: std.mem.Allocator, cursor: u64, limit: usize) ![]SeqEvent { 1481 1481 - mutex.lockUncancelable(store_io); 1482 1482 - defer mutex.unlock(store_io); 1501 1501 + db_mutex.lockUncancelable(store_io); 1502 1502 + defer db_mutex.unlock(store_io); 1483 1503 try requireInitialized(); 1484 1504 try backfillSeqEventsLocked(allocator); 1485 1505 ··· 1515 1535 } 1516 1536 1517 1537 pub fn writeRepoStatusJson(allocator: std.mem.Allocator, did: []const u8) ![]const u8 { 1518 1518 - mutex.lockUncancelable(store_io); 1519 1519 - defer mutex.unlock(store_io); 1538 1538 + db_mutex.lockUncancelable(store_io); 1539 1539 + defer db_mutex.unlock(store_io); 1520 1540 try requireInitialized(); 1521 1541 const root = try latestRootLocked(allocator, did); 1522 1542 const active = try accountActiveLocked(did); ··· 1547 1567 collection: []const u8, 1548 1568 limit: usize, 1549 1569 ) ![]const u8 { 1550 1550 - mutex.lockUncancelable(store_io); 1551 1551 - defer mutex.unlock(store_io); 1570 1570 + db_mutex.lockUncancelable(store_io); 1571 1571 + defer db_mutex.unlock(store_io); 1552 1572 try requireInitialized(); 1553 1573 1554 1574 var rows = try conn.rows( ··· 1710 1730 } 1711 1731 1712 1732 pub fn getEmailInfo(allocator: std.mem.Allocator, did: []const u8) ?EmailInfo { 1713 1713 - mutex.lockUncancelable(store_io); 1714 1714 - defer mutex.unlock(store_io); 1733 1733 + db_mutex.lockUncancelable(store_io); 1734 1734 + defer db_mutex.unlock(store_io); 1715 1735 requireInitialized() catch return null; 1716 1736 const row = conn.row( 1717 1737 \\SELECT email, email_confirmed_at, auth_code, auth_code_expires_at, pending_email ··· 1730 1750 } 1731 1751 1732 1752 pub fn setAuthCode(did: []const u8, code: []const u8, expires_at_ms: i64) !void { 1733 1733 - mutex.lockUncancelable(store_io); 1734 1734 - defer mutex.unlock(store_io); 1753 1753 + db_mutex.lockUncancelable(store_io); 1754 1754 + defer db_mutex.unlock(store_io); 1735 1755 try requireInitialized(); 1736 1756 try conn.exec( 1737 1757 \\UPDATE accounts ··· 1741 1761 } 1742 1762 1743 1763 pub fn setPendingEmail(did: []const u8, pending_email: []const u8, code: []const u8, expires_at_ms: i64) !void { 1744 1744 - mutex.lockUncancelable(store_io); 1745 1745 - defer mutex.unlock(store_io); 1764 1764 + db_mutex.lockUncancelable(store_io); 1765 1765 + defer db_mutex.unlock(store_io); 1746 1766 try requireInitialized(); 1747 1767 try conn.exec( 1748 1768 \\UPDATE accounts ··· 1752 1772 } 1753 1773 1754 1774 pub fn validateAuthCode(did: []const u8, code: []const u8, now_ms: i64) CodeStatus { 1755 1755 - mutex.lockUncancelable(store_io); 1756 1756 - defer mutex.unlock(store_io); 1775 1775 + db_mutex.lockUncancelable(store_io); 1776 1776 + defer db_mutex.unlock(store_io); 1757 1777 requireInitialized() catch return .invalid; 1758 1778 const row = conn.row( 1759 1779 \\SELECT auth_code, auth_code_expires_at ··· 1770 1790 } 1771 1791 1772 1792 pub fn clearAuthCode(did: []const u8) !void { 1773 1773 - mutex.lockUncancelable(store_io); 1774 1774 - defer mutex.unlock(store_io); 1793 1793 + db_mutex.lockUncancelable(store_io); 1794 1794 + defer db_mutex.unlock(store_io); 1775 1795 try requireInitialized(); 1776 1796 try conn.exec( 1777 1797 \\UPDATE accounts ··· 1782 1802 } 1783 1803 1784 1804 pub fn confirmEmail(did: []const u8) !void { 1785 1785 - mutex.lockUncancelable(store_io); 1786 1786 - defer mutex.unlock(store_io); 1805 1805 + db_mutex.lockUncancelable(store_io); 1806 1806 + defer db_mutex.unlock(store_io); 1787 1807 try requireInitialized(); 1788 1808 try conn.exec( 1789 1809 \\UPDATE accounts ··· 1795 1815 } 1796 1816 1797 1817 pub fn updateEmail(did: []const u8, email: []const u8) !void { 1798 1798 - mutex.lockUncancelable(store_io); 1799 1799 - defer mutex.unlock(store_io); 1818 1818 + db_mutex.lockUncancelable(store_io); 1819 1819 + defer db_mutex.unlock(store_io); 1800 1820 try requireInitialized(); 1801 1821 try conn.exec( 1802 1822 \\UPDATE accounts ··· 1809 1829 , .{ email, did }); 1810 1830 } 1811 1831 1812 1812 - fn nextSeqLocked() !u64 { 1832 1832 + fn loadNextSeqLocked() !u64 { 1813 1833 const row = try conn.row( 1814 1834 \\SELECT COALESCE(MAX(seq), 0) + 1 1815 1835 \\FROM ( ··· 1823 1843 return @intCast(row.?.int(0)); 1824 1844 } 1825 1845 1846 1846 + fn nextSeqLocked() !u64 { 1847 1847 + _ = try requireInitialized(); 1848 1848 + return next_seq.fetchAdd(1, .monotonic); 1849 1849 + } 1850 1850 + 1826 1851 fn nextRkey(allocator: std.mem.Allocator) ![]const u8 { 1827 1827 - mutex.lockUncancelable(store_io); 1828 1828 - defer mutex.unlock(store_io); 1852 1852 + db_mutex.lockUncancelable(store_io); 1853 1853 + defer db_mutex.unlock(store_io); 1829 1854 try requireInitialized(); 1830 1855 return nextRkeyLocked(allocator); 1831 1856 } ··· 1834 1859 const seq = try nextSeqLocked(); 1835 1860 const tid = try atid.encode(nowMicros(), @intCast(seq % 1024)); 1836 1861 return allocator.dupe(u8, &tid); 1862 1862 + } 1863 1863 + 1864 1864 + fn resolveWriteOpsLocked(allocator: std.mem.Allocator, ops: []const WriteOp) ![]WriteOp { 1865 1865 + var resolved = try allocator.alloc(WriteOp, ops.len); 1866 1866 + for (ops, 0..) |op, i| { 1867 1867 + resolved[i] = switch (op) { 1868 1868 + .create => |create_op| .{ .create = .{ 1869 1869 + .collection = create_op.collection, 1870 1870 + .rkey = create_op.rkey orelse try nextRkeyLocked(allocator), 1871 1871 + .value = create_op.value, 1872 1872 + } }, 1873 1873 + .update => |update_op| .{ .update = update_op }, 1874 1874 + .delete => |delete_op| .{ .delete = delete_op }, 1875 1875 + }; 1876 1876 + } 1877 1877 + return resolved; 1837 1878 } 1838 1879 1839 1880 fn validateRecordForWrite(collection: []const u8, rkey: ?[]const u8, value: std.json.Value) Error!void { ··· 2105 2146 data_cid: zat.cbor.Cid, 2106 2147 prev_cid_raw: ?[]const u8, 2107 2148 ) !SignedCommit { 2149 2149 + var keypair = try signingKeypairLocked(did); 2150 2150 + return signedCommitWithKeypair(allocator, did, rev, data_cid, prev_cid_raw, &keypair); 2151 2151 + } 2152 2152 + 2153 2153 + fn signedCommitWithKeypair( 2154 2154 + allocator: std.mem.Allocator, 2155 2155 + did: []const u8, 2156 2156 + rev: []const u8, 2157 2157 + data_cid: zat.cbor.Cid, 2158 2158 + prev_cid_raw: ?[]const u8, 2159 2159 + keypair: *const zat.Keypair, 2160 2160 + ) !SignedCommit { 2108 2161 var unsigned_entries = try allocator.alloc(zat.cbor.Value.MapEntry, 5); 2109 2162 unsigned_entries[0] = .{ .key = "did", .value = .{ .text = did } }; 2110 2163 unsigned_entries[1] = .{ .key = "version", .value = .{ .unsigned = 3 } }; ··· 2113 2166 unsigned_entries[4] = .{ .key = "prev", .value = if (prev_cid_raw) |raw| .{ .cid = .{ .raw = raw } } else .null }; 2114 2167 const unsigned_value: zat.cbor.Value = .{ .map = unsigned_entries }; 2115 2168 const unsigned_bytes = try zat.cbor.encodeAlloc(allocator, unsigned_value); 2116 2116 - var keypair = try signingKeypairLocked(did); 2117 2169 const sig = try keypair.sign(unsigned_bytes); 2118 2170 2119 2171 var signed_entries = try allocator.alloc(zat.cbor.Value.MapEntry, 6);