From 4a950c2cb8aa72ebd8f595d124f449705469a074 Mon Sep 17 00:00:00 2001 From: zzstoatzz Date: Thu, 30 Jul 2026 19:50:44 +0000 Subject: [PATCH] Avoid global store stalls during blob IO --- src/storage/blobstore.zig | 48 +++++++++++++++++++++++++++++++++++++++--------- src/storage/store.zig | 69 ++++++++++++++++++++++++++++++++++++++++++--------------------------- 2 file(s) changed, 81 insertion(s)(+), 36 deletion(s)(-) diff --git a/src/storage/blobstore.zig b/src/storage/blobstore.zig --- a/src/storage/blobstore.zig +++ b/src/storage/blobstore.zig @@ -12,12 +12,11 @@ } pub fn put(allocator: std.mem.Allocator, io: Io, did: []const u8, cid: []const u8, data: []const u8) !void { - _ = io; try requireInitialized(); const dir_path = try actorDirPath(allocator, did); try mkdirPath(allocator, dir_path); const path = try blobPath(allocator, did, cid); - try writeFileC(allocator, path, data); + try writeFileAtomicC(allocator, io, path, data); } pub fn get(allocator: std.mem.Allocator, did: []const u8, cid: []const u8, limit: usize) ![]u8 { @@ -66,13 +65,25 @@ return error.CreateDirFailed; } -fn writeFileC(allocator: std.mem.Allocator, path: []const u8, data: []const u8) !void { - const path_z = try allocator.dupeZ(u8, path); - const file = std.c.fopen(path_z.ptr, "wb") orelse return error.OpenBlobFailed; - defer _ = std.c.fclose(file); - if (data.len == 0) return; - const written = std.c.fwrite(data.ptr, 1, data.len, file); - if (written != data.len) return error.WriteBlobFailed; +fn writeFileAtomicC(allocator: std.mem.Allocator, io: Io, path: []const u8, data: []const u8) !void { + var random_bytes: [8]u8 = undefined; + io.random(&random_bytes); + const nonce = std.fmt.bytesToHex(random_bytes, .lower); + const temp_path = try std.fmt.allocPrint(allocator, "{s}.tmp-{s}", .{ path, &nonce }); + 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); + } + if (data.len > 0 and std.c.fwrite(data.ptr, 1, data.len, file) != data.len) { + return error.WriteBlobFailed; + } + if (std.c.fclose(file) != 0) return error.WriteBlobFailed; + closed = true; + try Io.Dir.cwd().rename(temp_path, Io.Dir.cwd(), path, io); } test "disk blobstore writes and reads account blob bytes" { @@ -91,4 +102,23 @@ try put(a, std.Options.debug_io, "did:plc:test", "bafytest", "hello blobstore"); const data = try get(a, "did:plc:test", "bafytest", 1024); try std.testing.expectEqualStrings("hello blobstore", data); +} + +test "disk blobstore replaces existing bytes atomically" { + 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); + const path = path_buf[0..path_len]; + init(std.Options.debug_io, path); + + var arena = std.heap.ArenaAllocator.init(allocator); + defer arena.deinit(); + const a = arena.allocator(); + try put(a, std.Options.debug_io, "did:plc:test", "bafytest", "first"); + try put(a, std.Options.debug_io, "did:plc:test", "bafytest", "second"); + const data = try get(a, "did:plc:test", "bafytest", 1024); + try std.testing.expectEqualStrings("second", data); } diff --git a/src/storage/store.zig b/src/storage/store.zig --- a/src/storage/store.zig +++ b/src/storage/store.zig @@ -2498,11 +2498,12 @@ mime_type: []const u8, ) ![]const u8 { const cid = try cidForBlob(allocator, data); + try requireInitialized(); + try blobstore.put(allocator, io, account.did, cid, data); 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( \\INSERT INTO blobs (cid, did, mime_type, size) \\VALUES (?, ?, ?, ?) @@ -2518,20 +2519,10 @@ did: []const u8, cid: []const u8, ) ?BlobRecord { - db_mutex.lockUncancelable(store_io); - defer db_mutex.unlock(store_io); - requireInitialized() catch return null; - const row = conn.row( - \\SELECT mime_type, size - \\FROM blobs - \\WHERE did = ? AND cid = ? - , .{ did, cid }) catch return null; - const found = row orelse return null; - defer found.deinit(); - const size: usize = @intCast(found.int(1)); + const metadata = blobMetadata(allocator, did, cid, false) orelse return null; return .{ - .mime_type = allocator.dupe(u8, found.text(0)) catch return null, - .data = blobstore.get(allocator, did, cid, size + 1) catch return null, + .mime_type = metadata.mime_type, + .data = blobstore.get(allocator, did, cid, metadata.size + 1) catch return null, }; } @@ -2540,26 +2531,50 @@ did: []const u8, cid: []const u8, ) ?BlobRecord { + const metadata = blobMetadata(allocator, did, cid, true) orelse return null; + return .{ + .mime_type = metadata.mime_type, + .data = blobstore.get(allocator, did, cid, metadata.size + 1) catch return null, + }; +} + +const BlobMetadata = struct { + mime_type: []const u8, + size: usize, +}; + +fn blobMetadata( + allocator: std.mem.Allocator, + did: []const u8, + cid: []const u8, + require_public_reference: bool, +) ?BlobMetadata { db_mutex.lockUncancelable(store_io); defer db_mutex.unlock(store_io); requireInitialized() catch return null; - const row = conn.row( - \\SELECT b.mime_type, b.size - \\FROM blobs b - \\WHERE b.did = ? AND b.cid = ? - \\ AND 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 - \\ ) - , .{ did, cid }) catch return null; + const row = if (require_public_reference) + conn.row( + \\SELECT b.mime_type, b.size + \\FROM blobs b + \\WHERE b.did = ? AND b.cid = ? + \\ AND 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 + \\ ) + , .{ did, cid }) catch return null + else + conn.row( + \\SELECT mime_type, size + \\FROM blobs + \\WHERE did = ? AND cid = ? + , .{ did, cid }) catch return null; const found = row orelse return null; defer found.deinit(); - const size: usize = @intCast(found.int(1)); return .{ .mime_type = allocator.dupe(u8, found.text(0)) catch return null, - .data = blobstore.get(allocator, did, cid, size + 1) catch return null, + .size = @intCast(found.int(1)), }; } -- tangled.sh