lib/sql/src/publication.zig
daab053ee43316e1809a84551d573ddd1e5bf3d2
1 const std = @import("std");
2
3 const Sha256 = std.crypto.hash.sha2.Sha256;
4 const path_bytes_max: usize = 1024;
5 const copy_buffer_bytes: usize = 64 * 1024;
6 const magic = "SQLPUB01".*;
7 const version: u32 = 1;
8 const seal_bytes: usize = 136;
9
10 pub const Token = u64;
11
12 pub const Error =
13 std.Io.File.OpenError ||
14 std.Io.File.StatError ||
15 std.Io.File.ReadPositionalError ||
16 std.Io.File.WritePositionalError ||
17 std.Io.File.SetLengthError ||
18 std.Io.File.SyncError ||
19 std.Io.Dir.RenameError ||
20 std.Io.Dir.DeleteFileError ||
21 error{InvalidPublication};
22
23 pub const Pair = struct {
24 database: []const u8,
25 wal: []const u8,
26 };
27
28 const Name = struct {
29 storage: [path_bytes_max]u8 = undefined,
30 len: u16 = 0,
31
32 fn init(base: []const u8, suffix: []const u8) Error!Name {
33 var value = Name{};
34 const written = std.fmt.bufPrint(&value.storage, "{s}{s}", .{ base, suffix }) catch
35 return error.InvalidPublication;
36 value.len = std.math.cast(u16, written.len) orelse
37 return error.InvalidPublication;
38 return value;
39 }
40
41 fn initJoined(
42 base: []const u8,
43 suffix: []const u8,
44 extra: []const u8,
45 ) Error!Name {
46 var value = Name{};
47 const written = std.fmt.bufPrint(
48 &value.storage,
49 "{s}{s}{s}",
50 .{ base, suffix, extra },
51 ) catch return error.InvalidPublication;
52 value.len = std.math.cast(u16, written.len) orelse
53 return error.InvalidPublication;
54 return value;
55 }
56
57 fn bytes(self: *const Name) []const u8 {
58 return self.storage[0..self.len];
59 }
60 };
61
62 pub const Lane = struct {
63 index: u1,
64 database: Name,
65 wal: Name,
66 seal: Name,
67 database_next: Name,
68 wal_next: Name,
69 seal_next: Name,
70
71 pub fn init(base: Pair, index: u1) Error!Lane {
72 var suffix_storage: [32]u8 = undefined;
73 const suffix = std.fmt.bufPrint(
74 &suffix_storage,
75 ".publication-{d}",
76 .{index},
77 ) catch return error.InvalidPublication;
78 var seal_suffix_storage: [40]u8 = undefined;
79 const seal_suffix = std.fmt.bufPrint(
80 &seal_suffix_storage,
81 "{s}.seal",
82 .{suffix},
83 ) catch return error.InvalidPublication;
84 return .{
85 .index = index,
86 .database = try Name.init(base.database, suffix),
87 .wal = try Name.init(base.wal, suffix),
88 .seal = try Name.init(base.database, seal_suffix),
89 .database_next = try Name.initJoined(base.database, suffix, ".next"),
90 .wal_next = try Name.initJoined(base.wal, suffix, ".next"),
91 .seal_next = try Name.initJoined(base.database, seal_suffix, ".next"),
92 };
93 }
94
95 pub fn pair(self: *const Lane) Pair {
96 return .{
97 .database = self.database.bytes(),
98 .wal = self.wal.bytes(),
99 };
100 }
101
102 pub fn artifact(self: *const Lane, index: usize) ?[]const u8 {
103 return switch (index) {
104 0 => self.database.bytes(),
105 1 => self.wal.bytes(),
106 2 => self.seal.bytes(),
107 3 => self.database_next.bytes(),
108 4 => self.wal_next.bytes(),
109 5 => self.seal_next.bytes(),
110 else => null,
111 };
112 }
113 };
114
115 pub const Inspect = enum {
116 missing,
117 ready,
118 };
119
120 pub const Opened = struct {
121 lane: u1,
122 database: std.Io.File,
123 wal: std.Io.File,
124
125 pub fn deinit(self: *Opened, io: std.Io) void {
126 self.database.close(io);
127 self.wal.close(io);
128 self.* = undefined;
129 }
130 };
131
132 const FileIdentity = struct {
133 bytes: u64,
134 digest: [Sha256.digest_length]u8,
135 };
136
137 const Seal = struct {
138 lane: u1,
139 token: Token,
140 database: FileIdentity,
141 wal: FileIdentity,
142 };
143
144 pub fn prepare(
145 io: std.Io,
146 dir: std.Io.Dir,
147 source: Pair,
148 destination: *const Lane,
149 ) Error!void {
150 try invalidate(io, dir, destination);
151 try copyFile(io, dir, source.database, destination.database_next.bytes());
152 errdefer removeIfPresent(io, dir, destination.database_next.bytes()) catch {};
153 try copyFile(io, dir, source.wal, destination.wal_next.bytes());
154 errdefer removeIfPresent(io, dir, destination.wal_next.bytes()) catch {};
155 try dir.rename(
156 destination.database_next.bytes(),
157 dir,
158 destination.database.bytes(),
159 io,
160 );
161 try dir.rename(
162 destination.wal_next.bytes(),
163 dir,
164 destination.wal.bytes(),
165 io,
166 );
167 try syncDirectory(io, dir);
168 }
169
170 pub fn prepareOpened(
171 io: std.Io,
172 dir: std.Io.Dir,
173 source: *const Opened,
174 destination: *const Lane,
175 ) Error!void {
176 try invalidate(io, dir, destination);
177 try copyOpenFile(io, dir, source.database, destination.database_next.bytes());
178 errdefer removeIfPresent(io, dir, destination.database_next.bytes()) catch {};
179 try copyOpenFile(io, dir, source.wal, destination.wal_next.bytes());
180 errdefer removeIfPresent(io, dir, destination.wal_next.bytes()) catch {};
181 try dir.rename(
182 destination.database_next.bytes(),
183 dir,
184 destination.database.bytes(),
185 io,
186 );
187 try dir.rename(
188 destination.wal_next.bytes(),
189 dir,
190 destination.wal.bytes(),
191 io,
192 );
193 try syncDirectory(io, dir);
194 }
195
196 pub fn invalidate(io: std.Io, dir: std.Io.Dir, lane: *const Lane) Error!void {
197 try removeIfPresent(io, dir, lane.seal.bytes());
198 try removeIfPresent(io, dir, lane.seal_next.bytes());
199 try syncDirectory(io, dir);
200 }
201
202 pub fn seal(
203 io: std.Io,
204 dir: std.Io.Dir,
205 lane: *const Lane,
206 token: Token,
207 ) Error!void {
208 if (token == 0) return error.InvalidPublication;
209 const database = try hashFile(io, dir, lane.database.bytes());
210 const wal = try hashFile(io, dir, lane.wal.bytes());
211 const value = Seal{
212 .lane = lane.index,
213 .token = token,
214 .database = database,
215 .wal = wal,
216 };
217 var encoded: [seal_bytes]u8 = undefined;
218 encodeSeal(&encoded, value);
219 try writeFile(io, dir, lane.seal_next.bytes(), &encoded);
220 errdefer removeIfPresent(io, dir, lane.seal_next.bytes()) catch {};
221 try dir.rename(lane.seal_next.bytes(), dir, lane.seal.bytes(), io);
222 try syncDirectory(io, dir);
223 }
224
225 pub fn inspect(
226 io: std.Io,
227 dir: std.Io.Dir,
228 lane: *const Lane,
229 token: Token,
230 ) Error!Inspect {
231 const value = (try readSeal(io, dir, lane.seal.bytes())) orelse
232 return .missing;
233 if (value.lane != lane.index or value.token != token) {
234 return error.InvalidPublication;
235 }
236 const database = try hashFile(io, dir, lane.database.bytes());
237 const wal = try hashFile(io, dir, lane.wal.bytes());
238 if (!sameIdentity(value.database, database) or
239 !sameIdentity(value.wal, wal))
240 {
241 return error.InvalidPublication;
242 }
243 return .ready;
244 }
245
246 pub fn locate(
247 io: std.Io,
248 dir: std.Io.Dir,
249 base: Pair,
250 token: Token,
251 ) Error!Lane {
252 var opened = try openSelected(io, dir, base, token);
253 defer opened.deinit(io);
254 return try Lane.init(base, opened.lane);
255 }
256
257 pub fn openSelected(
258 io: std.Io,
259 dir: std.Io.Dir,
260 base: Pair,
261 token: Token,
262 ) Error!Opened {
263 if (token == 0) return error.InvalidPublication;
264 var found: ?Opened = null;
265 errdefer if (found) |*opened| opened.deinit(io);
266 for (0..2) |index| {
267 const lane = try Lane.init(base, @intCast(index));
268 const opened = (try openLane(io, dir, &lane, token)) orelse continue;
269 if (found != null) {
270 var duplicate = opened;
271 duplicate.deinit(io);
272 return error.InvalidPublication;
273 }
274 found = opened;
275 }
276 const opened = found orelse return error.InvalidPublication;
277 found = null;
278 return opened;
279 }
280
281 pub fn reset(io: std.Io, dir: std.Io.Dir, base: Pair) Error!void {
282 for (0..2) |index| {
283 const lane = try Lane.init(base, @intCast(index));
284 var artifact_index: usize = 0;
285 while (lane.artifact(artifact_index)) |name| : (artifact_index += 1) {
286 try removeIfPresent(io, dir, name);
287 }
288 }
289 try syncDirectory(io, dir);
290 }
291
292 fn encodeSeal(out: *[seal_bytes]u8, value: Seal) void {
293 @memcpy(out[0..8], &magic);
294 std.mem.writeInt(u32, out[8..12], version, .little);
295 std.mem.writeInt(u32, out[12..16], value.lane, .little);
296 std.mem.writeInt(u64, out[16..24], value.token, .little);
297 std.mem.writeInt(u64, out[24..32], value.database.bytes, .little);
298 std.mem.writeInt(u64, out[32..40], value.wal.bytes, .little);
299 @memcpy(out[40..72], &value.database.digest);
300 @memcpy(out[72..104], &value.wal.digest);
301 Sha256.hash(out[0..104], out[104..136], .{});
302 }
303
304 fn decodeSeal(encoded: *const [seal_bytes]u8) ?Seal {
305 if (!std.mem.eql(u8, encoded[0..8], &magic)) return null;
306 if (std.mem.readInt(u32, encoded[8..12], .little) != version) return null;
307 const lane_value = std.mem.readInt(u32, encoded[12..16], .little);
308 if (lane_value > 1) return null;
309 const token = std.mem.readInt(u64, encoded[16..24], .little);
310 if (token == 0) return null;
311 var digest: [Sha256.digest_length]u8 = undefined;
312 Sha256.hash(encoded[0..104], &digest, .{});
313 if (!std.mem.eql(u8, encoded[104..136], &digest)) return null;
314 return .{
315 .lane = @intCast(lane_value),
316 .token = token,
317 .database = .{
318 .bytes = std.mem.readInt(u64, encoded[24..32], .little),
319 .digest = encoded[40..72].*,
320 },
321 .wal = .{
322 .bytes = std.mem.readInt(u64, encoded[32..40], .little),
323 .digest = encoded[72..104].*,
324 },
325 };
326 }
327
328 fn readSeal(
329 io: std.Io,
330 dir: std.Io.Dir,
331 path: []const u8,
332 ) Error!?Seal {
333 return readSealMaybe(io, dir, path);
334 }
335
336 fn readSealMaybe(
337 io: std.Io,
338 dir: std.Io.Dir,
339 path: []const u8,
340 ) Error!?Seal {
341 var file = dir.openFile(io, path, .{}) catch |err| switch (err) {
342 error.FileNotFound => return null,
343 else => return err,
344 };
345 defer file.close(io);
346 var encoded: [seal_bytes + 1]u8 = undefined;
347 const count = try file.readPositionalAll(io, &encoded, 0);
348 if (count != seal_bytes) return error.InvalidPublication;
349 return decodeSeal(encoded[0..seal_bytes]) orelse
350 return error.InvalidPublication;
351 }
352
353 fn copyFile(
354 io: std.Io,
355 dir: std.Io.Dir,
356 source_path: []const u8,
357 target_path: []const u8,
358 ) Error!void {
359 var source = try dir.openFile(io, source_path, .{});
360 defer source.close(io);
361 try copyOpenFile(io, dir, source, target_path);
362 }
363
364 fn copyOpenFile(
365 io: std.Io,
366 dir: std.Io.Dir,
367 source: std.Io.File,
368 target_path: []const u8,
369 ) Error!void {
370 var target = try dir.createFile(io, target_path, .{
371 .read = true,
372 .truncate = true,
373 });
374 defer target.close(io);
375 const length = try source.length(io);
376 var buffer: [copy_buffer_bytes]u8 = undefined;
377 var offset: u64 = 0;
378 while (offset < length) {
379 const remaining = length - offset;
380 const count: usize = @intCast(@min(remaining, buffer.len));
381 const read = try source.readPositionalAll(io, buffer[0..count], offset);
382 if (read != count) return error.InvalidPublication;
383 try target.writePositionalAll(io, buffer[0..count], offset);
384 offset += count;
385 }
386 try target.setLength(io, length);
387 try target.sync(io);
388 }
389
390 fn hashFile(
391 io: std.Io,
392 dir: std.Io.Dir,
393 path: []const u8,
394 ) Error!FileIdentity {
395 var file = try dir.openFile(io, path, .{});
396 defer file.close(io);
397 return hashOpenFile(io, file);
398 }
399
400 fn hashOpenFile(io: std.Io, file: std.Io.File) Error!FileIdentity {
401 const length = try file.length(io);
402 var hash = Sha256.init(.{});
403 var buffer: [copy_buffer_bytes]u8 = undefined;
404 var offset: u64 = 0;
405 while (offset < length) {
406 const remaining = length - offset;
407 const count: usize = @intCast(@min(remaining, buffer.len));
408 const read = try file.readPositionalAll(io, buffer[0..count], offset);
409 if (read != count) return error.InvalidPublication;
410 hash.update(buffer[0..count]);
411 offset += count;
412 }
413 var digest: [Sha256.digest_length]u8 = undefined;
414 hash.final(&digest);
415 return .{ .bytes = length, .digest = digest };
416 }
417
418 fn openLane(
419 io: std.Io,
420 dir: std.Io.Dir,
421 lane: *const Lane,
422 token: Token,
423 ) Error!?Opened {
424 const value = readSealMaybe(io, dir, lane.seal.bytes()) catch |err| switch (err) {
425 error.InvalidPublication => return null,
426 else => return err,
427 } orelse return null;
428 if (value.token != token) return null;
429 if (value.lane != lane.index) return error.InvalidPublication;
430 var database = dir.openFile(io, lane.database.bytes(), .{}) catch |err| switch (err) {
431 error.FileNotFound => return error.InvalidPublication,
432 else => return err,
433 };
434 errdefer database.close(io);
435 var wal_file = dir.openFile(io, lane.wal.bytes(), .{}) catch |err| switch (err) {
436 error.FileNotFound => return error.InvalidPublication,
437 else => return err,
438 };
439 errdefer wal_file.close(io);
440 const database_identity = try hashOpenFile(io, database);
441 const wal_identity = try hashOpenFile(io, wal_file);
442 if (!sameIdentity(value.database, database_identity) or
443 !sameIdentity(value.wal, wal_identity))
444 {
445 return error.InvalidPublication;
446 }
447 return .{ .lane = lane.index, .database = database, .wal = wal_file };
448 }
449
450 fn writeFile(
451 io: std.Io,
452 dir: std.Io.Dir,
453 path: []const u8,
454 bytes: []const u8,
455 ) Error!void {
456 var file = try dir.createFile(io, path, .{ .read = true, .truncate = true });
457 defer file.close(io);
458 try file.writePositionalAll(io, bytes, 0);
459 try file.setLength(io, bytes.len);
460 try file.sync(io);
461 }
462
463 fn removeIfPresent(io: std.Io, dir: std.Io.Dir, path: []const u8) Error!void {
464 dir.deleteFile(io, path) catch |err| switch (err) {
465 error.FileNotFound => {},
466 else => return err,
467 };
468 }
469
470 fn syncDirectory(io: std.Io, dir: std.Io.Dir) Error!void {
471 var file = try dir.openFile(io, ".", .{
472 .mode = .read_only,
473 .allow_directory = true,
474 });
475 defer file.close(io);
476 try file.sync(io);
477 }
478
479 fn sameIdentity(left: FileIdentity, right: FileIdentity) bool {
480 return left.bytes == right.bytes and
481 std.mem.eql(u8, &left.digest, &right.digest);
482 }
483
484 test "publication prepares and authenticates alternating physical lanes" {
485 const testing = std.testing;
486 const io = std.Options.debug_io;
487 var tmp = testing.tmpDir(.{});
488 defer tmp.cleanup();
489 try writeFile(io, tmp.dir, "working.db", "database-one");
490 try writeFile(io, tmp.dir, "working.wal", "wal-one");
491 const base = Pair{ .database = "working.db", .wal = "working.wal" };
492 const first = try Lane.init(base, 0);
493 try prepare(io, tmp.dir, base, &first);
494 try testing.expectEqual(Inspect.missing, try inspect(io, tmp.dir, &first, 1));
495 try seal(io, tmp.dir, &first, 1);
496 try testing.expectEqual(Inspect.ready, try inspect(io, tmp.dir, &first, 1));
497
498 try writeFile(io, tmp.dir, "working.db", "database-two");
499 try writeFile(io, tmp.dir, "working.wal", "wal-two");
500 const second = try Lane.init(base, 1);
501 try prepare(io, tmp.dir, base, &second);
502 try seal(io, tmp.dir, &second, 2);
503 try testing.expectEqual(Inspect.ready, try inspect(io, tmp.dir, &first, 1));
504 try testing.expectEqual(Inspect.ready, try inspect(io, tmp.dir, &second, 2));
505 try testing.expectError(error.InvalidPublication, inspect(io, tmp.dir, &second, 1));
506 }
507
508 test "publication rejects a seal after physical lane mutation" {
509 const testing = std.testing;
510 const io = std.Options.debug_io;
511 var tmp = testing.tmpDir(.{});
512 defer tmp.cleanup();
513 try writeFile(io, tmp.dir, "working.db", "database");
514 try writeFile(io, tmp.dir, "working.wal", "wal");
515 const base = Pair{ .database = "working.db", .wal = "working.wal" };
516 const lane = try Lane.init(base, 0);
517 try prepare(io, tmp.dir, base, &lane);
518 try seal(io, tmp.dir, &lane, 1);
519 try writeFile(io, tmp.dir, lane.database.bytes(), "changed");
520 try testing.expectError(error.InvalidPublication, inspect(io, tmp.dir, &lane, 1));
521 }