diff --git a/README.md b/README.md index cb2f7c2..b79b4d5 100644 --- a/README.md +++ b/README.md @@ -130,7 +130,10 @@ just docker-publish-release v0.1.1 - Known record schemas are validated before writes are accepted. - Unknown record schemas are accepted with `validationStatus: "unknown"`, matching the official PDS behavior when validation is not explicitly required. -- Uploaded blob bytes are stored in the configured disk blobstore. +- Uploaded blob bytes are stored in the configured disk blobstore. Untethered + uploads have a 24-hour grace period; blobs are retained while referenced by + public or permissioned records and collected after their final reference is + removed. - Invite codes use the official PDS table shape: code metadata plus recorded uses. When invites are required, `createAccount` rejects missing, disabled, or exhausted codes. diff --git a/bench/README.md b/bench/README.md index b9a6ba8..2f9624d 100644 --- a/bench/README.md +++ b/bench/README.md @@ -16,6 +16,7 @@ just bench write just bench read just bench repo just bench blob +just bench blob-gc 512 65536 just bench metastore 10 1000 just bench get-cid 10 1000 just bench get-block 10 1000 @@ -45,6 +46,7 @@ just bench run --scenario write --records 10000 - `read`: repeated `listRecords` queries against one seeded repo. - `repo`: full repo CAR materialization through `writeRepoCar`. - `blob`: blob write/read against disk blobstore plus SQLite metadata. +- `blob-gc`: bounded deletion of untethered blob metadata and disk files. - `metastore`: Tranquil-shaped apply/get/list benchmark with caller counts and latency percentiles. - `get-cid`: CID-only record lookup, matching Tranquil's `get_record_cid` diff --git a/bench/fly/target-entrypoint.sh b/bench/fly/target-entrypoint.sh index d275d9a..d4194d3 100755 --- a/bench/fly/target-entrypoint.sh +++ b/bench/fly/target-entrypoint.sh @@ -45,10 +45,12 @@ COMMIT; SQL printf 'timestamp,memory_bytes,cpu_usage_usec,db_bytes,blob_bytes\n' >/tmp/zds-bench-resources.csv +server_pid="$$" ( while true; do timestamp="$(date +%s)" - memory_bytes="$(cat /sys/fs/cgroup/memory.current 2>/dev/null || printf 0)" + memory_kib="$(awk '/^VmRSS:/ { print $2; exit }' "/proc/$server_pid/status" 2>/dev/null || printf 0)" + memory_bytes="$((memory_kib * 1024))" cpu_usage_usec="$(awk '/usage_usec/ { print $2; exit }' /sys/fs/cgroup/cpu.stat 2>/dev/null || printf 0)" db_bytes="$(stat -c %s "$db" 2>/dev/null || printf 0)" blob_bytes="$(du -sb "${ZDS_BLOBSTORE_PATH:-/data/blobs}" 2>/dev/null | awk '{ print $1 }' || printf 0)" diff --git a/bench/justfile b/bench/justfile index b27e9e4..b48ac6f 100644 --- a/bench/justfile +++ b/bench/justfile @@ -25,6 +25,10 @@ read: blob: {{zig}} build bench -Doptimize=ReleaseFast -- --scenario blob --blobs 512 --blob-size 65536 +# benchmark reclaiming untethered blob metadata and disk files +blob-gc blobs="512" blob_size="65536": + {{zig}} build bench -Doptimize=ReleaseFast -- --scenario blob-gc --blobs {{blobs}} --blob-size {{blob_size}} + # benchmark full repo CAR materialization repo: {{zig}} build bench -Doptimize=ReleaseFast -- --scenario repo --records {{scale}} diff --git a/bench/main.zig b/bench/main.zig index b412278..08eeb36 100644 --- a/bench/main.zig +++ b/bench/main.zig @@ -6,6 +6,7 @@ const Scenario = enum { write, read, blob, + blob_gc, repo, metastore, get_cid, @@ -118,6 +119,11 @@ pub fn main(init: std.process.Init) !void { defer state.deinit(); (try benchBlob(allocator, state.account, options.blobs, options.blob_size)).print(); }, + .blob_gc => { + var state = try initBench(allocator); + defer state.deinit(); + (try benchBlobGc(allocator, state.account, options.blobs, options.blob_size)).print(); + }, .metastore => try benchMetastore(allocator, options), .get_cid => try benchGetCid(allocator, options), .get_block => try benchGetBlock(allocator, options), @@ -228,6 +234,23 @@ fn benchBlob(allocator: std.mem.Allocator, account: zds.auth.tokens.Account, blo return .{ .name = "blob put+get", .ops = blobs, .bytes = blobs * blob_size * 2, .elapsed_ns = nowNs() - start }; } +fn benchBlobGc(allocator: std.mem.Allocator, account: zds.auth.tokens.Account, blobs: usize, blob_size: usize) !BenchResult { + var arena = std.heap.ArenaAllocator.init(allocator); + defer arena.deinit(); + const a = arena.allocator(); + const payload = try a.alloc(u8, @max(blob_size, @sizeOf(u64))); + @memset(payload, 0xa5); + for (0..blobs) |i| { + std.mem.writeInt(u64, payload[0..@sizeOf(u64)], i, .little); + _ = try zds.storage.store.putBlob(a, std.Options.debug_io, account, payload, "application/octet-stream"); + } + + const start = nowNs(); + const result = try zds.storage.store.collectBlobGarbage(a, 0, blobs); + if (result.deleted != blobs or result.failed != 0) return error.UnexpectedBlobGcResult; + return .{ .name = "blob gc", .ops = result.deleted, .bytes = @intCast(result.bytes), .elapsed_ns = nowNs() - start }; +} + const ConcurrentStats = struct { p50: u64, p95: u64, @@ -1487,6 +1510,7 @@ fn parseScenario(value: []const u8) !Scenario { if (std.mem.eql(u8, value, "write")) return .write; if (std.mem.eql(u8, value, "read")) return .read; if (std.mem.eql(u8, value, "blob")) return .blob; + if (std.mem.eql(u8, value, "blob-gc")) return .blob_gc; if (std.mem.eql(u8, value, "repo")) return .repo; if (std.mem.eql(u8, value, "metastore")) return .metastore; if (std.mem.eql(u8, value, "get-cid")) return .get_cid; @@ -1506,7 +1530,7 @@ fn parseScenario(value: []const u8) !Scenario { fn usage() void { std.debug.print( - \\usage: zds-bench [--scenario all|write|read|blob|repo|metastore|get-cid|get-block|decode-record|render-record|get-record|list-records|repo-size|get-repo|firehose|publish-sync|space|write-profile] [--records N] [--blobs N] [--blob-size BYTES] [--callers N --ops-per-caller N] + \\usage: zds-bench [--scenario all|write|read|blob|blob-gc|repo|metastore|get-cid|get-block|decode-record|render-record|get-record|list-records|repo-size|get-repo|firehose|publish-sync|space|write-profile] [--records N] [--blobs N] [--blob-size BYTES] [--callers N --ops-per-caller N] \\ , .{}); } diff --git a/docs/operations.md b/docs/operations.md index b749e51..727cbb9 100644 --- a/docs/operations.md +++ b/docs/operations.md @@ -33,6 +33,7 @@ just bench write just bench read just bench repo just bench blob +just bench blob-gc 512 65536 just bench http just bench metastore 10 1000 just bench get-cid 10 1000 @@ -41,10 +42,27 @@ just bench run --records 1000 ``` These measure repo writes, indexed reads, full repo CAR materialization, blob -disk IO, and selected HTTP route overhead against temporary SQLite/blobstore +disk IO and collection, and selected HTTP route overhead against temporary SQLite/blobstore state. Treat the numbers as local trend data; compare runs on the same machine and optimize against the shape of the curve, not a single absolute value. +## blob lifecycle + +Uploaded blobs begin untethered. ZDS gives an untethered upload 24 hours to be +referenced by either a public repo record or a permissioned-space record. Once +a blob is old enough, removing its final record reference makes it collectable. +A reference from either record system keeps the blob. + +Collection is bounded and automatic. A background worker starts with the +service and reclaims at most 256 eligible blobs per minute. Upload bodies are +streamed through a fixed 64 KiB buffer into unique temporary files without +holding a repo-write lane; only the +atomic promotion and metadata update are serialized with collection for the +same CID. Collection rechecks both record systems under the account write lane +before deleting metadata and disk bytes. The service logs deleted counts, +bytes, and failures. `just bench blob-gc` measures the deletion path against +temporary state. + Keep experimental permissioned-data probes separate from this baseline. Their first job is to catch obviously bad access patterns in `com.atproto.space.*`, not to expand the normal benchmark gate. diff --git a/src/atproto/repo.zig b/src/atproto/repo.zig index ec62338..10a9312 100644 --- a/src/atproto/repo.zig +++ b/src/atproto/repo.zig @@ -428,14 +428,16 @@ pub fn uploadBlob(io: std.Io, request: *http_api.Request) !void { if (auth_ctx.oauth_scope) |scope| { try requireBlobScope(request, scope, mime_type); } - const body = http_api.readBodyAlloc(request, allocator, config.blobUploadLimit()) catch |err| switch (err) { - error.BodyTooLarge => return http_api.xrpcError(request, .payload_too_large, "InvalidRequest", "blob is too large"), - else => return err, - }; - const cid = store.putBlob(allocator, io, account, body, mime_type) catch |err| { - return http_api.xrpcError(request, .internal_server_error, "InternalServerError", @errorName(err)); + const max_size = config.blobUploadLimit(); + if (request.body_len > max_size) { + return http_api.xrpcError(request, .payload_too_large, "InvalidRequest", "blob is too large"); + } + var body_reader = try request.reader(30_000); + const blob = store.putBlobReader(allocator, io, account, &body_reader.interface, request.body_len, max_size, mime_type) catch |err| switch (err) { + error.BlobTooLarge => return http_api.xrpcError(request, .payload_too_large, "InvalidRequest", "blob is too large"), + else => return http_api.xrpcError(request, .internal_server_error, "InternalServerError", @errorName(err)), }; - const body_out = try uploadBlobResponseJson(allocator, cid, mime_type, body.len); + const body_out = try uploadBlobResponseJson(allocator, blob.cid, mime_type, blob.size); return http_api.json(request, .ok, body_out); } diff --git a/src/http/server.zig b/src/http/server.zig index 909c8bf..72e3d4a 100644 --- a/src/http/server.zig +++ b/src/http/server.zig @@ -170,7 +170,7 @@ pub fn listen(io: std.Io, options: Options) !void { .address = .{ .ip = .{ .ip4 = try std.Io.net.Ip4Address.parse(options.host, options.port) } }, .request = .{ .max_body_size = 128 * 1024 * 1024, - .lazy_read_size = 16 * 1024 * 1024, + .lazy_read_size = 64 * 1024, }, .workers = .{ .large_buffer_count = 4, diff --git a/src/main.zig b/src/main.zig index ae5d5d2..4bab668 100644 --- a/src/main.zig +++ b/src/main.zig @@ -60,6 +60,7 @@ pub fn main(init: std.process.Init) !void { zds.storage.blobstore.init(io, zds.core.config.blobstorePath()); zds.storage.eventlog.init(io); try zds.storage.store.init(io, options.db_path); + zds.storage.store.startBlobGarbageCollector(); if (zds.core.config.inviteRequired()) { if (try zds.storage.store.ensureBootstrapInviteCode(allocator, zds.core.config.publicUrl())) |code| { zds.core.log.info("bootstrap invite code: {s}\n", .{code}); diff --git a/src/storage/blobstore.zig b/src/storage/blobstore.zig index 249c1e1..bbd96dc 100644 --- a/src/storage/blobstore.zig +++ b/src/storage/blobstore.zig @@ -12,11 +12,75 @@ pub fn init(io: Io, path: []const u8) void { } pub fn put(allocator: std.mem.Allocator, io: Io, did: []const u8, cid: []const u8, data: []const u8) !void { + const temp_path = try stage(allocator, io, did, cid, data); + errdefer discard(temp_path); + try promote(allocator, temp_path, did, cid); +} + +pub fn stage(allocator: std.mem.Allocator, io: Io, did: []const u8, cid: []const u8, data: []const u8) ![]const u8 { try requireInitialized(); const dir_path = try actorDirPath(allocator, did); try mkdirPath(allocator, dir_path); const path = try blobPath(allocator, did, cid); - try writeFileAtomicC(allocator, io, path, data); + return writeTempFileC(allocator, io, path, data); +} + +pub const StagedUpload = struct { + temp_path: []const u8, + digest: [std.crypto.hash.sha2.Sha256.digest_length]u8, + size: usize, +}; + +pub fn stageUpload( + allocator: std.mem.Allocator, + io: Io, + did: []const u8, + reader: *std.Io.Reader, + expected_size: usize, + max_size: usize, +) !StagedUpload { + if (expected_size > max_size) return error.BlobTooLarge; + try requireInitialized(); + const dir_path = try actorDirPath(allocator, did); + try mkdirPath(allocator, dir_path); + const temp_path = try uniqueTempPath(allocator, io, dir_path); + const temp_path_z = try allocator.dupeZ(u8, temp_path); + errdefer Io.Dir.cwd().deleteFile(io, temp_path) catch {}; + + const file = std.c.fopen(temp_path_z.ptr, "wb") orelse return error.OpenBlobFailed; + var closed = false; + defer { + if (!closed) _ = std.c.fclose(file); + } + + const Sha256 = std.crypto.hash.sha2.Sha256; + var hasher = Sha256.init(.{}); + var size: usize = 0; + var buf: [64 * 1024]u8 = undefined; + while (size < expected_size) { + const read_len = @min(buf.len, expected_size - size); + reader.readSliceAll(buf[0..read_len]) catch return error.IncompleteBlob; + const chunk = buf[0..read_len]; + if (std.c.fwrite(chunk.ptr, 1, chunk.len, file) != chunk.len) return error.WriteBlobFailed; + hasher.update(chunk); + size += chunk.len; + } + if (std.c.fclose(file) != 0) return error.WriteBlobFailed; + closed = true; + + var digest: [Sha256.digest_length]u8 = undefined; + hasher.final(&digest); + return .{ .temp_path = temp_path, .digest = digest, .size = size }; +} + +pub fn promote(allocator: std.mem.Allocator, temp_path: []const u8, did: []const u8, cid: []const u8) !void { + try requireInitialized(); + const path = try blobPath(allocator, did, cid); + try Io.Dir.cwd().rename(temp_path, Io.Dir.cwd(), path, blob_io); +} + +pub fn discard(temp_path: []const u8) void { + Io.Dir.cwd().deleteFile(blob_io, temp_path) catch {}; } pub fn get(allocator: std.mem.Allocator, did: []const u8, cid: []const u8, limit: usize) ![]u8 { @@ -25,9 +89,13 @@ pub fn get(allocator: std.mem.Allocator, did: []const u8, cid: []const u8, limit return Io.Dir.cwd().readFileAlloc(blob_io, path, allocator, .limited(limit)); } -pub fn delete(allocator: std.mem.Allocator, did: []const u8, cid: []const u8) void { - const path = blobPath(allocator, did, cid) catch return; - Io.Dir.cwd().deleteFile(blob_io, path) catch {}; +pub fn delete(allocator: std.mem.Allocator, did: []const u8, cid: []const u8) !void { + try requireInitialized(); + const path = try blobPath(allocator, did, cid); + Io.Dir.cwd().deleteFile(blob_io, path) catch |err| switch (err) { + error.FileNotFound => {}, + else => return err, + }; } fn actorDirPath(allocator: std.mem.Allocator, did: []const u8) ![]const u8 { @@ -65,7 +133,7 @@ fn mkdirOne(allocator: std.mem.Allocator, path: []const u8) !void { return error.CreateDirFailed; } -fn writeFileAtomicC(allocator: std.mem.Allocator, io: Io, path: []const u8, data: []const u8) !void { +fn writeTempFileC(allocator: std.mem.Allocator, io: Io, path: []const u8, data: []const u8) ![]const u8 { var random_bytes: [8]u8 = undefined; io.random(&random_bytes); const nonce = std.fmt.bytesToHex(random_bytes, .lower); @@ -83,7 +151,14 @@ fn writeFileAtomicC(allocator: std.mem.Allocator, io: Io, path: []const u8, data } if (std.c.fclose(file) != 0) return error.WriteBlobFailed; closed = true; - try Io.Dir.cwd().rename(temp_path, Io.Dir.cwd(), path, io); + return temp_path; +} + +fn uniqueTempPath(allocator: std.mem.Allocator, io: Io, dir_path: []const u8) ![]const u8 { + var random_bytes: [8]u8 = undefined; + io.random(&random_bytes); + const nonce = std.fmt.bytesToHex(random_bytes, .lower); + return std.fmt.allocPrint(allocator, "{s}/.upload-{s}.tmp", .{ dir_path, &nonce }); } test "disk blobstore writes and reads account blob bytes" { @@ -122,3 +197,21 @@ test "disk blobstore replaces existing bytes atomically" { const data = try get(a, "did:plc:test", "bafytest", 1024); try std.testing.expectEqualStrings("second", data); } + +test "streamed blob upload enforces its size limit" { + const allocator = std.testing.allocator; + var tmp = std.testing.tmpDir(.{}); + defer tmp.cleanup(); + + var path_buf: [std.fs.max_path_bytes]u8 = undefined; + const path_len = try tmp.dir.realPath(std.testing.io, &path_buf); + init(std.Options.debug_io, path_buf[0..path_len]); + + var arena = std.heap.ArenaAllocator.init(allocator); + defer arena.deinit(); + var reader: std.Io.Reader = .fixed("too large"); + try std.testing.expectError( + error.BlobTooLarge, + stageUpload(arena.allocator(), std.Options.debug_io, "did:plc:test", &reader, 9, 3), + ); +} diff --git a/src/storage/store.zig b/src/storage/store.zig index 06cacfa..4fe75c5 100644 --- a/src/storage/store.zig +++ b/src/storage/store.zig @@ -538,7 +538,19 @@ const DatabaseMutex = struct { var db_mutex: DatabaseMutex = .{}; var write_lanes: sharded_locks.ShardedLocks(32) = .{}; +var blob_lanes: sharded_locks.ShardedLocks(64) = .{}; var next_seq: std.atomic.Value(u64) = .init(1); +var blob_gc_worker_started: std.atomic.Value(bool) = .init(false); + +const blob_gc_grace_seconds: i64 = 24 * 60 * 60; +const blob_gc_batch: usize = 256; +const blob_gc_interval_seconds: i64 = 60; + +pub const BlobGcResult = struct { + deleted: usize = 0, + bytes: u64 = 0, + failed: usize = 0, +}; pub fn init(io: Io, path: []const u8) !void { if (initialized) return; @@ -2080,7 +2092,8 @@ fn applyWritesMeasured(allocator: std.mem.Allocator, account: auth.Account, ops: const sql_start = monotonicNs(); db_mutex.lockUncancelable(store_io); - defer db_mutex.unlock(store_io); + var db_locked = true; + defer if (db_locked) db_mutex.unlock(store_io); try conn.exclusiveTransaction(); errdefer conn.rollback(); for (record_blocks.items) |block| { @@ -2165,6 +2178,8 @@ fn applyWritesMeasured(allocator: std.mem.Allocator, account: auth.Account, ops: \\ evt = excluded.evt , .{ @as(i64, @intCast(seq)), account.did, commit.cid, zqlite.blob(event_frame) }); try conn.commit(); + db_mutex.unlock(store_io); + db_locked = false; addProfile(profile, .sql_ns, elapsedNs(sql_start)); const publish_start = monotonicNs(); @@ -2554,21 +2569,219 @@ pub fn putBlob( data: []const u8, mime_type: []const u8, ) ![]const u8 { - const cid = try cidForBlob(allocator, data); - try requireInitialized(); - try blobstore.put(allocator, io, account.did, cid, data); + var reader: std.Io.Reader = .fixed(data); + return (try putBlobReader(allocator, io, account, &reader, data.len, data.len, mime_type)).cid; +} - db_mutex.lockUncancelable(store_io); - defer db_mutex.unlock(store_io); +pub const StoredBlob = struct { + cid: []const u8, + size: usize, +}; + +pub fn putBlobReader( + allocator: std.mem.Allocator, + io: Io, + account: auth.Account, + reader: *std.Io.Reader, + expected_size: usize, + max_size: usize, + mime_type: []const u8, +) !StoredBlob { try requireInitialized(); - try conn.exec( - \\INSERT INTO blobs (cid, did, mime_type, size) - \\VALUES (?, ?, ?, ?) - \\ON CONFLICT(did, cid) DO UPDATE SET - \\ mime_type = excluded.mime_type, - \\ size = excluded.size - , .{ cid, account.did, mime_type, @as(i64, @intCast(data.len)) }); - return cid; + const staged = try blobstore.stageUpload(allocator, io, account.did, reader, expected_size, max_size); + errdefer blobstore.discard(staged.temp_path); + const cid = try cidForBlobDigest(allocator, staged.digest); + { + const lane = blob_lanes.lock(store_io, cid); + defer lane.unlock(); + try blobstore.promote(allocator, staged.temp_path, account.did, cid); + + db_mutex.lockUncancelable(store_io); + defer db_mutex.unlock(store_io); + try requireInitialized(); + try conn.exec( + \\INSERT INTO blobs (cid, did, mime_type, size) + \\VALUES (?, ?, ?, ?) + \\ON CONFLICT(did, cid) DO UPDATE SET + \\ mime_type = excluded.mime_type, + \\ size = excluded.size, + \\ created_at = unixepoch() + , .{ cid, account.did, mime_type, @as(i64, @intCast(staged.size)) }); + } + return .{ .cid = cid, .size = staged.size }; +} + +const BlobGcCandidate = struct { cid: []const u8 }; + +pub fn collectBlobGarbage( + allocator: std.mem.Allocator, + grace_seconds: i64, + limit: usize, +) !BlobGcResult { + if (limit == 0) return .{}; + + var dids: std.ArrayList([]const u8) = .empty; + { + db_mutex.lockUncancelable(store_io); + defer db_mutex.unlock(store_io); + try requireInitialized(); + var rows = try conn.rows( + \\SELECT DISTINCT b.did + \\FROM blobs b + \\WHERE NOT EXISTS ( + \\ SELECT 1 FROM expected_blobs eb + \\ JOIN records r ON r.uri = eb.record_uri + \\ WHERE r.did = b.did AND eb.blob_cid = b.cid + \\) + \\AND NOT EXISTS ( + \\ SELECT 1 FROM permissioned_space_record_blobs psrb + \\ WHERE psrb.repo_did = b.did AND psrb.blob_cid = b.cid + \\) + \\AND b.created_at <= unixepoch() - ? + \\ORDER BY b.did + , .{grace_seconds}); + defer rows.deinit(); + while (rows.next()) |row| try dids.append(allocator, try allocator.dupe(u8, row.text(0))); + if (rows.err) |err| return err; + } + + var result: BlobGcResult = .{}; + for (dids.items) |did| { + if (result.deleted + result.failed >= limit) break; + { + const lane = write_lanes.lock(store_io, did); + defer lane.unlock(); + const account_result = try collectBlobGarbageForDidInLane( + allocator, + did, + grace_seconds, + limit - result.deleted - result.failed, + ); + result.deleted += account_result.deleted; + result.bytes += account_result.bytes; + result.failed += account_result.failed; + } + } + return result; +} + +fn collectBlobGarbageForDidInLane( + allocator: std.mem.Allocator, + did: []const u8, + grace_seconds: i64, + limit: usize, +) !BlobGcResult { + if (limit == 0) return .{}; + + var candidates: std.ArrayList(BlobGcCandidate) = .empty; + { + db_mutex.lockUncancelable(store_io); + defer db_mutex.unlock(store_io); + try requireInitialized(); + var rows = try conn.rows( + \\SELECT b.cid + \\FROM blobs b + \\WHERE b.did = ? + \\AND NOT EXISTS ( + \\ SELECT 1 FROM expected_blobs eb + \\ JOIN records r ON r.uri = eb.record_uri + \\ WHERE r.did = b.did AND eb.blob_cid = b.cid + \\) + \\AND NOT EXISTS ( + \\ SELECT 1 FROM permissioned_space_record_blobs psrb + \\ WHERE psrb.repo_did = b.did AND psrb.blob_cid = b.cid + \\) + \\AND b.created_at <= unixepoch() - ? + \\ORDER BY b.created_at, b.cid + \\LIMIT ? + , .{ did, grace_seconds, @as(i64, @intCast(limit)) }); + defer rows.deinit(); + while (rows.next()) |row| { + try candidates.append(allocator, .{ + .cid = try allocator.dupe(u8, row.text(0)), + }); + } + if (rows.err) |err| return err; + } + + var result: BlobGcResult = .{}; + for (candidates.items) |candidate| { + const blob_lane = blob_lanes.lock(store_io, candidate.cid); + defer blob_lane.unlock(); + const size = blk: { + db_mutex.lockUncancelable(store_io); + defer db_mutex.unlock(store_io); + const row = try conn.row( + \\SELECT b.size + \\FROM blobs b + \\WHERE b.did = ? AND b.cid = ? + \\AND NOT EXISTS ( + \\ SELECT 1 FROM expected_blobs eb + \\ JOIN records r ON r.uri = eb.record_uri + \\ WHERE r.did = b.did AND eb.blob_cid = b.cid + \\) + \\AND NOT EXISTS ( + \\ SELECT 1 FROM permissioned_space_record_blobs psrb + \\ WHERE psrb.repo_did = b.did AND psrb.blob_cid = b.cid + \\) + \\AND b.created_at <= unixepoch() - ? + , .{ did, candidate.cid, grace_seconds }); + const found = row orelse continue; + defer found.deinit(); + break :blk @as(u64, @intCast(found.int(0))); + }; + blobstore.delete(allocator, did, candidate.cid) catch { + result.failed += 1; + continue; + }; + { + db_mutex.lockUncancelable(store_io); + defer db_mutex.unlock(store_io); + try conn.exec( + \\DELETE FROM blobs + \\WHERE did = ? AND cid = ? + \\AND NOT EXISTS ( + \\ SELECT 1 FROM expected_blobs eb + \\ JOIN records r ON r.uri = eb.record_uri + \\ WHERE r.did = blobs.did AND eb.blob_cid = blobs.cid + \\) + \\AND NOT EXISTS ( + \\ SELECT 1 FROM permissioned_space_record_blobs psrb + \\ WHERE psrb.repo_did = blobs.did AND psrb.blob_cid = blobs.cid + \\) + , .{ did, candidate.cid }); + } + result.deleted += 1; + result.bytes += size; + } + return result; +} + +pub fn startBlobGarbageCollector() void { + if (blob_gc_worker_started.cmpxchgStrong(false, true, .acq_rel, .acquire) != null) return; + const thread = std.Thread.spawn(.{}, collectBlobGarbageThread, .{}) catch |err| { + blob_gc_worker_started.store(false, .release); + log.err("blob gc spawn failed error={s}\n", .{@errorName(err)}); + return; + }; + thread.detach(); +} + +fn collectBlobGarbageThread() void { + while (true) { + var arena = std.heap.ArenaAllocator.init(std.heap.page_allocator); + const result = collectBlobGarbage(arena.allocator(), blob_gc_grace_seconds, blob_gc_batch) catch |err| blk: { + log.err("blob gc failed error={s}\n", .{@errorName(err)}); + break :blk null; + }; + if (result) |summary| { + if (summary.deleted > 0 or summary.failed > 0) { + log.info("blob gc deleted={d} bytes={d} failed={d}\n", .{ summary.deleted, summary.bytes, summary.failed }); + } + } + arena.deinit(); + store_io.sleep(Io.Duration.fromSeconds(blob_gc_interval_seconds), .awake) catch {}; + } } pub fn getBlob( @@ -2826,11 +3039,16 @@ pub fn importRepo( records: []const ImportedRecord, blocks: []const ImportedBlock, ) !void { + const lane = write_lanes.lock(store_io, account.did); + defer lane.unlock(); + var publish_event = false; + var seq: u64 = 0; db_mutex.lockUncancelable(store_io); - defer db_mutex.unlock(store_io); + var db_locked = true; + defer if (db_locked) db_mutex.unlock(store_io); try requireInitialized(); - const seq = try nextSeqLocked(); + seq = try nextSeqLocked(); try conn.exclusiveTransaction(); errdefer conn.rollback(); try conn.exec("DELETE FROM expected_blobs WHERE record_uri IN (SELECT uri FROM records WHERE did = ?)", .{account.did}); @@ -2866,7 +3084,7 @@ pub fn importRepo( \\INSERT INTO commits (seq, did, cid, rev, prev) \\VALUES (?, ?, ?, ?, NULL) , .{ @as(i64, @intCast(seq)), account.did, commit_cid, rev }); - const publish_event = try accountActiveLocked(account.did); + publish_event = try accountActiveLocked(account.did); if (publish_event) { const event_frame = try importedCommitEventFrame(allocator, seq, account.did, commit_cid, rev, blocks); try conn.exec( @@ -2879,6 +3097,8 @@ pub fn importRepo( , .{ @as(i64, @intCast(seq)), account.did, commit_cid, zqlite.blob(event_frame) }); } try conn.commit(); + db_mutex.unlock(store_io); + db_locked = false; if (publish_event) eventlog.publish(seq); } @@ -3909,8 +4129,13 @@ pub fn applySpaceWrites( }, }; + const lane = write_lanes.lock(store_io, repo_did); + defer lane.unlock(); + + var results: std.ArrayList(SpaceWriteResult) = .empty; db_mutex.lockUncancelable(store_io); - defer db_mutex.unlock(store_io); + var db_locked = true; + defer if (db_locked) db_mutex.unlock(store_io); try requireInitialized(); const space_config = (try getSpaceConfigLocked(allocator, space)) orelse return Error.RepoNotFound; @@ -3921,7 +4146,6 @@ pub fn applySpaceWrites( var set_hash = if (state.set_hash) |bytes| try permissioned.LtHash.fromBytes(bytes) else permissioned.LtHash{}; const rev = try nextRkeyLocked(allocator); - var results: std.ArrayList(SpaceWriteResult) = .empty; var idx: i64 = 0; for (ops) |op| { switch (op) { @@ -3991,6 +4215,8 @@ pub fn applySpaceWrites( , .{ space, repo_did, rev, zqlite.blob(&digest) }); } try conn.commit(); + db_mutex.unlock(store_io); + db_locked = false; return results.toOwnedSlice(allocator); } @@ -5990,6 +6216,16 @@ fn cidForBlob(allocator: std.mem.Allocator, data: []const u8) ![]const u8 { return zat.multibase.base32lower.encode(allocator, cid.raw); } +fn cidForBlobDigest( + allocator: std.mem.Allocator, + digest: [std.crypto.hash.sha2.Sha256.digest_length]u8, +) ![]const u8 { + var raw: [4 + std.crypto.hash.sha2.Sha256.digest_length]u8 = undefined; + raw[0..4].* = .{ 1, 0x55, 0x12, std.crypto.hash.sha2.Sha256.digest_length }; + @memcpy(raw[4..], &digest); + return zat.multibase.base32lower.encode(allocator, &raw); +} + fn requireInitialized() !void { if (!initialized) return Error.StoreNotInitialized; } @@ -6417,6 +6653,7 @@ const post_schema_statements = [_][*:0]const u8{ \\ PRIMARY KEY (blob_cid, record_uri) \\) , + "CREATE INDEX IF NOT EXISTS blobs_gc_idx ON blobs (created_at, did, cid)", "CREATE UNIQUE INDEX IF NOT EXISTS oauth_tokens_family_id_idx ON oauth_tokens (family_id)", "CREATE UNIQUE INDEX IF NOT EXISTS oauth_tokens_previous_refresh_token_idx ON oauth_tokens (previous_refresh_token) WHERE previous_refresh_token IS NOT NULL", \\CREATE TABLE IF NOT EXISTS oauth_used_refresh_tokens ( @@ -7462,6 +7699,8 @@ test "stores larger blob metadata in sqlite and bytes in disk blobstore" { @memset(data, 0xaa); const cid = try putBlob(allocator, std.Options.debug_io, account, data, "image/jpeg"); + const expected_cid = try cidForBlob(allocator, data); + try std.testing.expectEqualStrings(expected_cid, cid); const row = try conn.row("SELECT mime_type, size FROM blobs WHERE did = ? AND cid = ?", .{ account.did, cid }); try std.testing.expect(row != null); defer row.?.deinit(); @@ -7473,6 +7712,131 @@ test "stores larger blob metadata in sqlite and bytes in disk blobstore" { try std.testing.expectEqualSlices(u8, data, blob.data); } +test "blob gc preserves fresh uploads through the untethered grace period" { + 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 path_len = try tmp.dir.realPath(std.testing.io, &path_buf); + blobstore.init(std.Options.debug_io, path_buf[0..path_len]); + + try init(std.Options.debug_io, ":memory:"); + defer close(); + const account = try createAccount(allocator, "gc-grace.test", "gc-grace@test.com", "password", "did:plc:gcgrace", true); + const cid = try putBlob(allocator, std.Options.debug_io, account, "untethered", "text/plain"); + + const protected = try collectBlobGarbage(allocator, 60, 10); + try std.testing.expectEqual(@as(usize, 0), protected.deleted); + try std.testing.expect(getBlob(allocator, account.did, cid) != null); + + const expired = try collectBlobGarbage(allocator, 0, 10); + try std.testing.expectEqual(@as(usize, 1), expired.deleted); + try std.testing.expectEqual(@as(u64, 10), expired.bytes); + try std.testing.expect(getBlob(allocator, account.did, cid) == null); +} + +test "blob gc waits for the last public record reference" { + 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 path_len = try tmp.dir.realPath(std.testing.io, &path_buf); + blobstore.init(std.Options.debug_io, path_buf[0..path_len]); + + try init(std.Options.debug_io, ":memory:"); + defer close(); + const account = try createAccount(allocator, "gc-public.test", "gc-public@test.com", "password", "did:plc:gcpublic", true); + const cid = try putBlob(allocator, std.Options.debug_io, account, "public", "text/plain"); + const json = try std.fmt.allocPrint( + allocator, + "{{\"$type\":\"fm.example.blob\",\"media\":{{\"$type\":\"blob\",\"ref\":{{\"$link\":\"{s}\"}},\"mimeType\":\"text/plain\",\"size\":6}}}}", + .{cid}, + ); + const parsed = try std.json.parseFromSlice(std.json.Value, allocator, json, .{}); + defer parsed.deinit(); + _ = try create(allocator, account, "fm.example.blob", "one", parsed.value); + _ = try create(allocator, account, "fm.example.blob", "two", parsed.value); + + _ = try delete(allocator, account, "fm.example.blob", "one"); + try std.testing.expect(getBlob(allocator, account.did, cid) != null); + _ = try delete(allocator, account, "fm.example.blob", "two"); + const gc = try collectBlobGarbage(allocator, 0, 10); + try std.testing.expectEqual(@as(usize, 1), gc.deleted); + try std.testing.expect(getBlob(allocator, account.did, cid) == null); +} + +test "blob gc follows permissioned record references" { + 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 path_len = try tmp.dir.realPath(std.testing.io, &path_buf); + blobstore.init(std.Options.debug_io, path_buf[0..path_len]); + + try init(std.Options.debug_io, ":memory:"); + defer close(); + const account = try createAccount(allocator, "gc-space.test", "gc-space@test.com", "password", "did:plc:gcspace", true); + const cid = try putBlob(allocator, std.Options.debug_io, account, "private", "audio/mpeg"); + const space = try createSpace(allocator, .{ + .actor_did = account.did, + .authority_did = account.did, + .space_type = "fm.example.private", + .skey = "self", + .is_authority = true, + .managing_app = null, + .policy = "member-list", + .app_access_json = "{\"type\":\"open\"}", + }); + const json = try std.fmt.allocPrint( + allocator, + "{{\"$type\":\"fm.example.privateBlob\",\"media\":{{\"$type\":\"blob\",\"ref\":{{\"$link\":\"{s}\"}},\"mimeType\":\"audio/mpeg\",\"size\":7}}}}", + .{cid}, + ); + const parsed = try std.json.parseFromSlice(std.json.Value, allocator, json, .{}); + defer parsed.deinit(); + const prepared = try prepareRecordValue(allocator, "fm.example.privateBlob", "one", parsed.value); + _ = try putSpaceRecord(allocator, space.uri, account.did, "fm.example.privateBlob", "one", prepared); + try std.testing.expect(getBlob(allocator, account.did, cid) != null); + + try deleteSpaceRecord(allocator, space.uri, account.did, "fm.example.privateBlob", "one"); + const gc = try collectBlobGarbage(allocator, 0, 10); + try std.testing.expectEqual(@as(usize, 1), gc.deleted); + try std.testing.expect(getBlob(allocator, account.did, cid) == null); +} + +test "re-upload refreshes the untethered grace period" { + 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 path_len = try tmp.dir.realPath(std.testing.io, &path_buf); + blobstore.init(std.Options.debug_io, path_buf[0..path_len]); + + try init(std.Options.debug_io, ":memory:"); + defer close(); + const account = try createAccount(allocator, "gc-reupload.test", "gc-reupload@test.com", "password", "did:plc:gcreupload", true); + const cid = try putBlob(allocator, std.Options.debug_io, account, "re-uploaded", "text/plain"); + try conn.exec("UPDATE blobs SET created_at = unixepoch() - 120 WHERE did = ? AND cid = ?", .{ account.did, cid }); + const same_cid = try putBlob(allocator, std.Options.debug_io, account, "re-uploaded", "text/plain"); + try std.testing.expectEqualStrings(cid, same_cid); + + const gc = try collectBlobGarbage(allocator, 60, 10); + try std.testing.expectEqual(@as(usize, 0), gc.deleted); + try std.testing.expect(getBlob(allocator, account.did, cid) != null); +} + test "invite codes gate account creation and record uses" { var arena = std.heap.ArenaAllocator.init(std.testing.allocator); defer arena.deinit();