lib/sql/src/history/refs.zig
daab053ee43316e1809a84551d573ddd1e5bf3d2
1 const std = @import("std");
2 const simd = @import("simd");
3 const sql = @import("../root.zig");
4 const record_mod = @import("record.zig");
5 const validate_mod = @import("validate.zig");
6
7 const Bytes = simd.ScalableTag(u8);
8 const version = sql.version;
9
10 const Allocator = std.mem.Allocator;
11
12 pub const Error = Allocator.Error;
13 pub const ReadError = Error || std.Io.Dir.ReadFileAllocError;
14
15 pub const suffix = ".refs";
16 pub const magic: u32 = 0x7473_7266;
17 pub const format_version: u32 = 2;
18 pub const max_snapshot_bytes: usize = 4 * 1024 * 1024;
19 pub const max_entries: u32 = 65_536;
20 pub const max_name_bytes: u32 = 4_096;
21 pub const path_bytes_max: usize = 1_024;
22 pub const repair_metadata_record_bytes_max: usize = max_snapshot_bytes;
23
24 const repair_stream_bytes: usize = 64 * 1024;
25 const repair_dependencies_max: usize = repair_metadata_record_bytes_max /
26 version.hash_bytes;
27 const repair_conflict_bytes_max: usize = max_snapshot_bytes;
28 const repair_limits = validate_mod.Limits{
29 .metadata_bytes_max = repair_metadata_record_bytes_max,
30 .dependencies_max = repair_dependencies_max,
31 .conflict_bytes_max = repair_conflict_bytes_max,
32 };
33
34 const header_size: usize = 4 + 4 + 8 + 4;
35 const entry_fixed_size: usize = 4 + 3 * version.hash_bytes;
36 const digest_size: usize = version.hash_bytes;
37 const write_suffix = ".next";
38 const write_path_bytes_max = path_bytes_max + write_suffix.len;
39
40 pub const Entry = struct {
41 name: []const u8,
42 head: version.Hash,
43 root: version.Hash,
44 conflicts: version.Hash,
45 };
46
47 pub const Snapshot = struct {
48 allocator: Allocator,
49 covered_length: u64,
50 entries: []Entry,
51 names: []u8,
52
53 pub fn deinit(self: *Snapshot) void {
54 self.allocator.free(self.entries);
55 self.allocator.free(self.names);
56 self.* = undefined;
57 }
58
59 pub fn find(self: *const Snapshot, name: []const u8) ?Entry {
60 std.debug.assert(self.entries.len <= max_entries);
61 for (self.entries) |entry| {
62 if (std.mem.eql(u8, entry.name, name)) return entry;
63 }
64 return null;
65 }
66 };
67
68 pub fn pathFor(buffer: *[path_bytes_max]u8, history_path: []const u8) ?[]const u8 {
69 if (history_path.len + suffix.len > buffer.len) return null;
70 @memcpy(buffer[0..history_path.len], history_path);
71 @memcpy(buffer[history_path.len..][0..suffix.len], suffix);
72 return buffer[0 .. history_path.len + suffix.len];
73 }
74
75 pub fn load(
76 allocator: Allocator,
77 io: std.Io,
78 dir: std.Io.Dir,
79 history_path: []const u8,
80 ) Error!?Snapshot {
81 return loadExisting(allocator, io, dir, history_path) catch |err| switch (err) {
82 error.OutOfMemory => return error.OutOfMemory,
83 else => return null,
84 };
85 }
86
87 pub fn loadExisting(
88 allocator: Allocator,
89 io: std.Io,
90 dir: std.Io.Dir,
91 history_path: []const u8,
92 ) ReadError!?Snapshot {
93 var path_buffer: [path_bytes_max]u8 = undefined;
94 const path = pathFor(&path_buffer, history_path) orelse return null;
95 const bytes = dir.readFileAlloc(io, path, allocator, .limited(max_snapshot_bytes)) catch |err| switch (err) {
96 error.StreamTooLong => return null,
97 else => return err,
98 };
99 defer allocator.free(bytes);
100 return try decode(allocator, bytes);
101 }
102
103 pub fn store(
104 allocator: Allocator,
105 io: std.Io,
106 dir: std.Io.Dir,
107 history_path: []const u8,
108 covered_length: u64,
109 entries: []const Entry,
110 ) (Error || std.Io.File.OpenError || std.Io.File.WritePositionalError ||
111 std.Io.Dir.RenameError)!void {
112 std.debug.assert(entries.len <= max_entries);
113 var path_buffer: [path_bytes_max]u8 = undefined;
114 const path = pathFor(&path_buffer, history_path) orelse return;
115 var write_buffer: [write_path_bytes_max]u8 = undefined;
116 @memcpy(write_buffer[0..path.len], path);
117 @memcpy(write_buffer[path.len..][0..write_suffix.len], write_suffix);
118 const write_path = write_buffer[0 .. path.len + write_suffix.len];
119
120 const bytes = try encode(allocator, covered_length, entries);
121 defer allocator.free(bytes);
122
123 {
124 var file = try dir.createFile(io, write_path, .{ .read = true, .truncate = true });
125 defer file.close(io);
126 try file.writePositionalAll(io, bytes, 0);
127 }
128 try dir.rename(write_path, dir, path, io);
129 }
130
131 fn encode(allocator: Allocator, covered_length: u64, entries: []const Entry) Error![]u8 {
132 std.debug.assert(entries.len <= max_entries);
133 var total: usize = header_size;
134 for (entries) |entry| {
135 std.debug.assert(entry.name.len <= max_name_bytes);
136 total += entry_fixed_size + entry.name.len;
137 }
138 total += digest_size;
139 std.debug.assert(total <= max_snapshot_bytes);
140
141 const bytes = try allocator.alloc(u8, total);
142 errdefer allocator.free(bytes);
143 var cursor: usize = 0;
144 writeU32(bytes, &cursor, magic);
145 writeU32(bytes, &cursor, format_version);
146 writeU64(bytes, &cursor, covered_length);
147 writeU32(bytes, &cursor, @intCast(entries.len));
148 for (entries) |entry| {
149 writeU32(bytes, &cursor, @intCast(entry.name.len));
150 @memcpy(bytes[cursor..][0..entry.name.len], entry.name);
151 cursor += entry.name.len;
152 @memcpy(bytes[cursor..][0..version.hash_bytes], entry.head[0..]);
153 cursor += version.hash_bytes;
154 @memcpy(bytes[cursor..][0..version.hash_bytes], entry.root[0..]);
155 cursor += version.hash_bytes;
156 @memcpy(bytes[cursor..][0..version.hash_bytes], entry.conflicts[0..]);
157 cursor += version.hash_bytes;
158 }
159 std.debug.assert(cursor + digest_size == total);
160 var digest: version.Hash = undefined;
161 std.crypto.hash.sha2.Sha256.hash(bytes[0..cursor], &digest, .{});
162 @memcpy(bytes[cursor..][0..digest_size], digest[0..]);
163 return bytes;
164 }
165
166 fn decode(allocator: Allocator, bytes: []const u8) Error!?Snapshot {
167 if (bytes.len < header_size + digest_size) return null;
168 if (bytes.len > max_snapshot_bytes) return null;
169 const body = bytes[0 .. bytes.len - digest_size];
170 var digest: version.Hash = undefined;
171 std.crypto.hash.sha2.Sha256.hash(body, &digest, .{});
172 if (!std.mem.eql(u8, digest[0..], bytes[body.len..])) return null;
173
174 var cursor: usize = 0;
175 if (readU32(body, &cursor) != magic) return null;
176 if (readU32(body, &cursor) != format_version) return null;
177 const covered_length = readU64(body, &cursor);
178 const entry_count = readU32(body, &cursor);
179 if (entry_count > max_entries) return null;
180
181 var name_bytes: usize = 0;
182 {
183 var scan = cursor;
184 var index: u32 = 0;
185 while (index < entry_count) : (index += 1) {
186 if (body.len - scan < 4) return null;
187 var peek = scan;
188 const name_len = readU32(body, &peek);
189 if (name_len == 0 or name_len > max_name_bytes) return null;
190 if (body.len - peek < name_len + 3 * version.hash_bytes) return null;
191 name_bytes += name_len;
192 scan = peek + name_len + 3 * version.hash_bytes;
193 }
194 if (scan != body.len) return null;
195 }
196
197 const entries = try allocator.alloc(Entry, entry_count);
198 errdefer allocator.free(entries);
199 const names = try allocator.alloc(u8, name_bytes);
200 errdefer allocator.free(names);
201
202 var name_cursor: usize = 0;
203 var index: u32 = 0;
204 while (index < entry_count) : (index += 1) {
205 const name_len = readU32(body, &cursor);
206 const name = names[name_cursor..][0..name_len];
207 @memcpy(name, body[cursor..][0..name_len]);
208 cursor += name_len;
209 name_cursor += name_len;
210 var head: version.Hash = undefined;
211 @memcpy(head[0..], body[cursor..][0..version.hash_bytes]);
212 cursor += version.hash_bytes;
213 var root: version.Hash = undefined;
214 @memcpy(root[0..], body[cursor..][0..version.hash_bytes]);
215 cursor += version.hash_bytes;
216 var conflicts: version.Hash = undefined;
217 @memcpy(conflicts[0..], body[cursor..][0..version.hash_bytes]);
218 cursor += version.hash_bytes;
219 entries[index] = .{
220 .name = name,
221 .head = head,
222 .root = root,
223 .conflicts = conflicts,
224 };
225 }
226 std.debug.assert(cursor == body.len);
227 std.debug.assert(name_cursor == names.len);
228
229 return .{
230 .allocator = allocator,
231 .covered_length = covered_length,
232 .entries = entries,
233 .names = names,
234 };
235 }
236
237 fn writeU32(bytes: []u8, cursor: *usize, value: u32) void {
238 std.mem.writeInt(u32, bytes[cursor.*..][0..4], value, .big);
239 cursor.* += 4;
240 }
241
242 fn writeU64(bytes: []u8, cursor: *usize, value: u64) void {
243 std.mem.writeInt(u64, bytes[cursor.*..][0..8], value, .big);
244 cursor.* += 8;
245 }
246
247 fn readU32(bytes: []const u8, cursor: *usize) u32 {
248 const value = std.mem.readInt(u32, bytes[cursor.*..][0..4], .big);
249 cursor.* += 4;
250 return value;
251 }
252
253 fn readU64(bytes: []const u8, cursor: *usize) u64 {
254 const value = std.mem.readInt(u64, bytes[cursor.*..][0..8], .big);
255 cursor.* += 8;
256 return value;
257 }
258
259 const RepairStage = enum {
260 current,
261 head,
262 root,
263 };
264
265 const RepairEntry = struct {
266 name: []u8,
267 head: version.Hash,
268 root: version.Hash,
269 conflicts: version.Hash,
270 stage: RepairStage,
271 };
272
273 const FastForwardDecision = enum {
274 pending,
275 baseline,
276 target,
277 };
278
279 const FastForward = struct {
280 id: version.Hash,
281 name: []u8,
282 expected: version.Hash,
283 target: version.Hash,
284 decision: FastForwardDecision = .pending,
285 };
286
287 const RepairState = struct {
288 allocator: Allocator,
289 entries: std.ArrayList(RepairEntry) = .empty,
290 encoded_bytes: usize = header_size + digest_size,
291 fast_forward: ?FastForward = null,
292
293 fn init(allocator: Allocator, snapshot: *const Snapshot) !RepairState {
294 var state = RepairState{ .allocator = allocator };
295 errdefer state.deinit();
296 try state.entries.ensureTotalCapacity(allocator, snapshot.entries.len);
297 for (snapshot.entries) |entry| {
298 const name = try allocator.dupe(u8, entry.name);
299 state.entries.appendAssumeCapacity(.{
300 .name = name,
301 .head = entry.head,
302 .root = entry.root,
303 .conflicts = entry.conflicts,
304 .stage = .current,
305 });
306 state.encoded_bytes += entry_fixed_size + name.len;
307 }
308 std.debug.assert(state.entries.items.len <= max_entries);
309 std.debug.assert(state.encoded_bytes <= max_snapshot_bytes);
310 return state;
311 }
312
313 fn deinit(self: *RepairState) void {
314 if (self.fast_forward) |pending| self.allocator.free(pending.name);
315 for (self.entries.items) |entry| self.allocator.free(entry.name);
316 self.entries.deinit(self.allocator);
317 self.* = undefined;
318 }
319
320 fn find(self: *RepairState, name: []const u8) ?usize {
321 std.debug.assert(self.entries.items.len <= max_entries);
322 for (self.entries.items, 0..) |entry, index| {
323 if (std.mem.eql(u8, entry.name, name)) return index;
324 }
325 return null;
326 }
327
328 fn setRef(self: *RepairState, name: []const u8, head: version.Hash) !void {
329 if (name.len == 0 or name.len > max_name_bytes) return error.InvalidHistory;
330 if (self.find(name)) |index| {
331 if (version.same(self.entries.items[index].head, head)) return;
332 const metadata = self.metadataForHead(head);
333 self.entries.items[index].head = head;
334 if (metadata) |entry| {
335 self.entries.items[index].root = entry.root;
336 self.entries.items[index].conflicts = entry.conflicts;
337 self.entries.items[index].stage = .current;
338 } else {
339 self.entries.items[index].stage = .head;
340 }
341 return;
342 }
343 if (self.entries.items.len >= max_entries) return error.StreamTooLong;
344 const next_bytes = std.math.add(
345 usize,
346 self.encoded_bytes,
347 entry_fixed_size + name.len,
348 ) catch return error.StreamTooLong;
349 if (next_bytes > max_snapshot_bytes) return error.StreamTooLong;
350 const owned_name = try self.allocator.dupe(u8, name);
351 errdefer self.allocator.free(owned_name);
352 try self.entries.append(self.allocator, .{
353 .name = owned_name,
354 .head = head,
355 .root = undefined,
356 .conflicts = undefined,
357 .stage = .head,
358 });
359 self.encoded_bytes = next_bytes;
360 }
361
362 fn metadataForHead(self: *const RepairState, head: version.Hash) ?RepairEntry {
363 std.debug.assert(self.entries.items.len <= max_entries);
364 for (self.entries.items) |entry| {
365 if (entry.stage == .current and version.same(entry.head, head)) return entry;
366 }
367 return null;
368 }
369
370 fn deleteRef(self: *RepairState, name: []const u8) void {
371 const index = self.find(name) orelse return;
372 const removed = self.entries.orderedRemove(index);
373 self.encoded_bytes -= entry_fixed_size + removed.name.len;
374 self.allocator.free(removed.name);
375 }
376
377 fn finish(self: *RepairState, covered_length: u64) !Snapshot {
378 if (self.fast_forward != null) return error.InvalidHistory;
379 for (self.entries.items) |entry| {
380 if (entry.stage != .current) return error.InvalidHistory;
381 }
382 const entries = try self.allocator.alloc(Entry, self.entries.items.len);
383 errdefer self.allocator.free(entries);
384 var names_bytes: usize = 0;
385 for (self.entries.items) |entry| names_bytes += entry.name.len;
386 const names = try self.allocator.alloc(u8, names_bytes);
387 errdefer self.allocator.free(names);
388 var cursor: usize = 0;
389 for (self.entries.items, entries) |source, *target| {
390 const name = names[cursor..][0..source.name.len];
391 @memcpy(name, source.name);
392 cursor += name.len;
393 target.* = .{
394 .name = name,
395 .head = source.head,
396 .root = source.root,
397 .conflicts = source.conflicts,
398 };
399 }
400 std.debug.assert(cursor == names.len);
401 return .{
402 .allocator = self.allocator,
403 .covered_length = covered_length,
404 .entries = entries,
405 .names = names,
406 };
407 }
408 };
409
410 const RecordView = struct {
411 kind: record_mod.RecordKind,
412 expected: version.Hash,
413 payload: []const u8,
414 };
415
416 const ScanMode = enum {
417 refs,
418 commits,
419 roots,
420
421 fn materializes(self: ScanMode, kind: record_mod.RecordKind) bool {
422 return switch (self) {
423 .refs => switch (kind) {
424 .ref,
425 .ref_delete,
426 .fast_forward_prepare,
427 .fast_forward_commit,
428 .fast_forward_abort,
429 .fast_forward_complete,
430 => true,
431 else => false,
432 },
433 .commits => kind == .commit,
434 .roots => kind == .database_root,
435 };
436 }
437 };
438
439 const RecordScanner = struct {
440 allocator: Allocator,
441 io: std.Io,
442 file: std.Io.File,
443 limit: usize,
444 offset: usize,
445 mode: ScanMode,
446 control: sql.wal.Control,
447 payload: std.ArrayList(u8) = .empty,
448
449 fn init(
450 allocator: Allocator,
451 io: std.Io,
452 file: std.Io.File,
453 start: usize,
454 end: usize,
455 mode: ScanMode,
456 control: sql.wal.Control,
457 ) RecordScanner {
458 std.debug.assert(start <= end);
459 return .{
460 .allocator = allocator,
461 .io = io,
462 .file = file,
463 .limit = end,
464 .offset = start,
465 .mode = mode,
466 .control = control,
467 };
468 }
469
470 fn deinit(self: *RecordScanner) void {
471 self.payload.deinit(self.allocator);
472 self.* = undefined;
473 }
474
475 fn next(self: *RecordScanner) !?RecordView {
476 try self.control.check();
477 if (self.offset == self.limit) return null;
478 if (self.limit - self.offset < record_mod.record_header_size) {
479 return error.InvalidHistory;
480 }
481 var header: [record_mod.record_header_size]u8 = undefined;
482 try self.readExact(&header, self.offset);
483 if (record_mod.readIntU32(header[0..4]) != record_mod.magic or
484 record_mod.readIntU32(header[4..8]) != record_mod.format_version)
485 {
486 return error.InvalidHistory;
487 }
488 const kind_value = record_mod.readIntU32(header[8..12]);
489 const kind = record_mod.recordKind(kind_value) orelse return error.InvalidHistory;
490 const payload_len: usize = record_mod.readIntU32(header[12..16]);
491 if (kind == .row_chunk and payload_len < version.hash_bytes) {
492 return error.InvalidHistory;
493 }
494 const payload_offset = std.math.add(
495 usize,
496 self.offset,
497 record_mod.record_header_size,
498 ) catch return error.InvalidHistory;
499 const payload_end = std.math.add(usize, payload_offset, payload_len) catch
500 return error.InvalidHistory;
501 if (payload_end > self.limit) return error.InvalidHistory;
502 const materialize = self.mode.materializes(kind);
503 if (materialize and payload_len > repair_metadata_record_bytes_max) {
504 return error.StreamTooLong;
505 }
506 self.payload.clearRetainingCapacity();
507 if (materialize) try self.payload.resize(self.allocator, payload_len);
508 var hasher = std.crypto.hash.sha2.Sha256.init(.{});
509 record_mod.writeHashU32(&hasher, record_mod.magic);
510 record_mod.writeHashU32(&hasher, record_mod.format_version);
511 record_mod.writeHashU32(&hasher, kind_value);
512 record_mod.writeHashU32(&hasher, @intCast(payload_len));
513 if (materialize) {
514 try self.readExact(self.payload.items, payload_offset);
515 hasher.update(self.payload.items);
516 } else {
517 var scratch: [repair_stream_bytes]u8 = undefined;
518 var cursor = payload_offset;
519 while (cursor < payload_end) {
520 try self.control.check();
521 const chunk_len = @min(scratch.len, payload_end - cursor);
522 try self.readExact(scratch[0..chunk_len], cursor);
523 hasher.update(scratch[0..chunk_len]);
524 cursor += chunk_len;
525 }
526 }
527 var actual: version.Hash = undefined;
528 hasher.final(&actual);
529 const expected = header[16..][0..version.hash_bytes].*;
530 if (!version.same(actual, expected)) return error.InvalidHistory;
531 self.offset = payload_end;
532 return .{
533 .kind = kind,
534 .expected = expected,
535 .payload = self.payload.items,
536 };
537 }
538
539 fn readExact(self: *RecordScanner, target: []u8, start: usize) !void {
540 var filled: usize = 0;
541 while (filled < target.len) {
542 try self.control.check();
543 const count = try self.file.readPositionalAll(
544 self.io,
545 target[filled..],
546 start + filled,
547 );
548 if (count == 0) return error.InvalidHistory;
549 filled += count;
550 }
551 }
552 };
553
554 pub fn advance(
555 allocator: Allocator,
556 io: std.Io,
557 dir: std.Io.Dir,
558 history_path: []const u8,
559 snapshot: *const Snapshot,
560 history_length: u64,
561 control: sql.wal.Control,
562 ) !Snapshot {
563 try control.check();
564 if (snapshot.covered_length > history_length) return error.InvalidHistory;
565 const start = std.math.cast(usize, snapshot.covered_length) orelse
566 return error.StreamTooLong;
567 const end = std.math.cast(usize, history_length) orelse return error.StreamTooLong;
568 var file = try dir.openFile(io, history_path, .{});
569 defer file.close(io);
570 try validate_mod.validateSuffix(
571 allocator,
572 io,
573 file,
574 start,
575 end,
576 snapshot.entries,
577 repair_limits,
578 control,
579 );
580 var state = try RepairState.init(allocator, snapshot);
581 defer state.deinit();
582 try applySuffix(&state, allocator, io, file, start, end, control);
583 try resolveHeads(&state, allocator, io, file, start, end, control);
584 if (hasStage(&state, .head) and start != 0) {
585 try resolveHeads(&state, allocator, io, file, 0, start, control);
586 }
587 try resolveRoots(&state, allocator, io, file, start, end, control);
588 if (hasStage(&state, .root) and start != 0) {
589 try resolveRoots(&state, allocator, io, file, 0, start, control);
590 }
591 return try state.finish(history_length);
592 }
593
594 fn applySuffix(
595 state: *RepairState,
596 allocator: Allocator,
597 io: std.Io,
598 file: std.Io.File,
599 start: usize,
600 end: usize,
601 control: sql.wal.Control,
602 ) !void {
603 var scanner = RecordScanner.init(allocator, io, file, start, end, .refs, control);
604 defer scanner.deinit();
605 while (try scanner.next()) |record| {
606 if (state.fast_forward != null and !fastForwardKind(record.kind)) {
607 return error.InvalidHistory;
608 }
609 switch (record.kind) {
610 .ref => {
611 try applyRef(state, record.payload);
612 },
613 .ref_delete => try applyRefDelete(state, record.payload),
614 .fast_forward_prepare => try applyFastForwardPrepare(state, record),
615 .fast_forward_commit => try applyFastForwardDecision(state, record.payload, .target),
616 .fast_forward_abort => try applyFastForwardDecision(state, record.payload, .baseline),
617 .fast_forward_complete => try applyFastForwardComplete(state, record.payload),
618 else => {},
619 }
620 }
621 }
622
623 fn fastForwardKind(kind: record_mod.RecordKind) bool {
624 return switch (kind) {
625 .fast_forward_prepare,
626 .fast_forward_commit,
627 .fast_forward_abort,
628 .fast_forward_complete,
629 => true,
630 else => false,
631 };
632 }
633
634 fn applyRef(state: *RepairState, payload: []const u8) !void {
635 var reader = record_mod.PayloadReader.init(payload);
636 const target = try reader.hash();
637 const name = try reader.readBytes();
638 try reader.finish();
639 try state.setRef(name, target);
640 }
641
642 fn applyRefDelete(state: *RepairState, payload: []const u8) !void {
643 var reader = record_mod.PayloadReader.init(payload);
644 const name = try reader.readBytes();
645 try reader.finish();
646 state.deleteRef(name);
647 }
648
649 fn applyFastForwardPrepare(
650 state: *RepairState,
651 record: RecordView,
652 ) !void {
653 if (state.fast_forward != null) return error.InvalidHistory;
654 var reader = record_mod.PayloadReader.init(record.payload);
655 const name = try reader.readBytes();
656 const expected = try reader.hash();
657 const target = try reader.hash();
658 try reader.finish();
659 const index = state.find(name) orelse return error.InvalidHistory;
660 if (!version.same(state.entries.items[index].head, expected)) return error.InvalidHistory;
661 state.fast_forward = .{
662 .id = record.expected,
663 .name = try state.allocator.dupe(u8, name),
664 .expected = expected,
665 .target = target,
666 };
667 }
668
669 fn applyFastForwardDecision(
670 state: *RepairState,
671 payload: []const u8,
672 decision: FastForwardDecision,
673 ) !void {
674 var reader = record_mod.PayloadReader.init(payload);
675 const id = try reader.hash();
676 try reader.finish();
677 const pending = if (state.fast_forward) |*value| value else return error.InvalidHistory;
678 if (!version.same(pending.id, id) or pending.decision != .pending) {
679 return error.InvalidHistory;
680 }
681 if (decision == .target) try state.setRef(pending.name, pending.target);
682 pending.decision = decision;
683 }
684
685 fn applyFastForwardComplete(state: *RepairState, payload: []const u8) !void {
686 var reader = record_mod.PayloadReader.init(payload);
687 const id = try reader.hash();
688 try reader.finish();
689 const pending = state.fast_forward orelse return error.InvalidHistory;
690 if (!version.same(pending.id, id) or pending.decision == .pending) {
691 return error.InvalidHistory;
692 }
693 const selected = if (pending.decision == .target) pending.target else pending.expected;
694 const index = state.find(pending.name) orelse return error.InvalidHistory;
695 if (!version.same(state.entries.items[index].head, selected)) return error.InvalidHistory;
696 state.allocator.free(pending.name);
697 state.fast_forward = null;
698 }
699
700 fn hasStage(state: *const RepairState, stage: RepairStage) bool {
701 std.debug.assert(state.entries.items.len <= max_entries);
702 for (state.entries.items) |entry| if (entry.stage == stage) return true;
703 return false;
704 }
705
706 fn resolveHeads(
707 state: *RepairState,
708 allocator: Allocator,
709 io: std.Io,
710 file: std.Io.File,
711 start: usize,
712 end: usize,
713 control: sql.wal.Control,
714 ) !void {
715 if (!hasStage(state, .head) or start == end) return;
716 var scanner = RecordScanner.init(allocator, io, file, start, end, .commits, control);
717 defer scanner.deinit();
718 while (try scanner.next()) |record| {
719 if (record.kind != .commit) continue;
720 const identity = try commitIdentity(allocator, record.payload);
721 for (state.entries.items) |*entry| {
722 if (entry.stage != .head or !version.same(entry.head, identity.head)) continue;
723 entry.root = identity.root;
724 entry.stage = .root;
725 }
726 if (!hasStage(state, .head)) break;
727 }
728 }
729
730 const CommitIdentity = struct {
731 head: version.Hash,
732 root: version.Hash,
733 };
734
735 fn commitIdentity(allocator: Allocator, payload: []const u8) !CommitIdentity {
736 var reader = record_mod.PayloadReader.init(payload);
737 const root = try reader.hash();
738 const parent_count = try reader.readU32();
739 const parent_bytes = std.math.mul(usize, parent_count, version.hash_bytes) catch
740 return error.InvalidHistory;
741 if (parent_bytes != reader.remaining()) return error.InvalidHistory;
742 const parents = try allocator.alloc(version.Hash, parent_count);
743 defer allocator.free(parents);
744 for (parents) |*parent| parent.* = try reader.hash();
745 try reader.finish();
746 return .{ .head = version.Commit.init(root, parents).hash, .root = root };
747 }
748
749 fn resolveRoots(
750 state: *RepairState,
751 allocator: Allocator,
752 io: std.Io,
753 file: std.Io.File,
754 start: usize,
755 end: usize,
756 control: sql.wal.Control,
757 ) !void {
758 if (!hasStage(state, .root) or start == end) return;
759 var scanner = RecordScanner.init(allocator, io, file, start, end, .roots, control);
760 defer scanner.deinit();
761 while (try scanner.next()) |record| {
762 if (record.kind != .database_root) continue;
763 const identity = try databaseRootIdentity(allocator, record.payload);
764 for (state.entries.items) |*entry| {
765 if (entry.stage != .root or !version.same(entry.root, identity.root)) continue;
766 entry.conflicts = identity.conflicts;
767 entry.stage = .current;
768 }
769 if (!hasStage(state, .root)) break;
770 }
771 }
772
773 const DatabaseRootIdentity = struct {
774 root: version.Hash,
775 conflicts: version.Hash,
776 };
777
778 fn databaseRootIdentity(allocator: Allocator, payload: []const u8) !DatabaseRootIdentity {
779 var reader = record_mod.PayloadReader.init(payload);
780 const conflicts = try reader.hash();
781 const entry_count = try reader.readU32();
782 const fixed_bytes = std.math.mul(usize, entry_count, 4 + version.hash_bytes) catch
783 return error.InvalidHistory;
784 if (fixed_bytes > reader.remaining()) return error.InvalidHistory;
785 const entries = try allocator.alloc(version.RelationEntry, entry_count);
786 defer allocator.free(entries);
787 for (entries) |*entry| {
788 entry.* = .{ .name = try reader.readBytes(), .hash = try reader.hash() };
789 }
790 try reader.finish();
791 std.mem.sort(version.RelationEntry, entries, {}, relationEntryLessThan);
792 const root = version.DatabaseRoot.init(entries, .{ .hash = conflicts });
793 return .{ .root = root.hash, .conflicts = conflicts };
794 }
795
796 fn relationEntryLessThan(
797 _: void,
798 left: version.RelationEntry,
799 right: version.RelationEntry,
800 ) bool {
801 return simd.order(Bytes, left.name, right.name) == .lt;
802 }
803
804 const testing_io = std.Options.debug_io;
805
806 test "refs snapshot round-trips entries and covered length" {
807 var tmp = std.testing.tmpDir(.{});
808 defer tmp.cleanup();
809
810 const head = @as([version.hash_bytes]u8, @splat(0xaa));
811 const root = @as([version.hash_bytes]u8, @splat(0xbb));
812 const conflicts = @as([version.hash_bytes]u8, @splat(0xcc));
813 const entries = [_]Entry{
814 .{ .name = "main", .head = head, .root = root, .conflicts = conflicts },
815 .{ .name = "feature/wide", .head = root, .root = head, .conflicts = root },
816 };
817 try store(std.testing.allocator, testing_io, tmp.dir, "log.history", 12_345, entries[0..]);
818
819 var loaded = (try load(std.testing.allocator, testing_io, tmp.dir, "log.history")) orelse
820 return error.SnapshotMissing;
821 defer loaded.deinit();
822
823 try std.testing.expectEqual(@as(u64, 12_345), loaded.covered_length);
824 try std.testing.expectEqual(@as(usize, 2), loaded.entries.len);
825 const found = loaded.find("main") orelse return error.RefMissing;
826 try std.testing.expectEqualSlices(u8, head[0..], found.head[0..]);
827 try std.testing.expectEqualSlices(u8, root[0..], found.root[0..]);
828 try std.testing.expectEqualSlices(u8, conflicts[0..], found.conflicts[0..]);
829 try std.testing.expect(loaded.find("missing") == null);
830 }
831
832 test "refs snapshot load rejects a corrupted digest" {
833 var tmp = std.testing.tmpDir(.{});
834 defer tmp.cleanup();
835
836 const head = @as([version.hash_bytes]u8, @splat(0x11));
837 const entries = [_]Entry{.{
838 .name = "main",
839 .head = head,
840 .root = head,
841 .conflicts = head,
842 }};
843 try store(std.testing.allocator, testing_io, tmp.dir, "log.history", 7, entries[0..]);
844
845 {
846 var file = try tmp.dir.openFile(testing_io, "log.history" ++ suffix, .{ .mode = .read_write });
847 defer file.close(testing_io);
848 try file.writePositionalAll(testing_io, "?", 9);
849 }
850
851 try std.testing.expect((try load(std.testing.allocator, testing_io, tmp.dir, "log.history")) == null);
852 }
853
854 test "refs snapshot load returns null when the sidecar is missing" {
855 var tmp = std.testing.tmpDir(.{});
856 defer tmp.cleanup();
857 try std.testing.expect((try load(std.testing.allocator, testing_io, tmp.dir, "absent.history")) == null);
858 }
859
860 test "refs snapshot rejects an empty branch name" {
861 var tmp = std.testing.tmpDir(.{});
862 defer tmp.cleanup();
863 const bytes = try encode(std.testing.allocator, 3, &.{});
864 defer std.testing.allocator.free(bytes);
865 var snapshot = (try decode(std.testing.allocator, bytes)) orelse return error.SnapshotMissing;
866 defer snapshot.deinit();
867 try std.testing.expectEqual(@as(usize, 0), snapshot.entries.len);
868 try std.testing.expectEqual(@as(u64, 3), snapshot.covered_length);
869 }
870
871 test "refs advance resolves stale suffix and pre-boundary targets" {
872 var tmp = std.testing.tmpDir(.{});
873 defer tmp.cleanup();
874
875 const history_path = "repair.history";
876 const conflicts = version.ConflictRoot.empty();
877 const first_root = version.DatabaseRoot.init(&.{}, conflicts);
878 const first_commit = version.Commit.init(first_root.hash, &.{});
879 const second_root = version.DatabaseRoot.init(&.{.{
880 .name = "items",
881 .hash = version.emptyHash("repair.items"),
882 }}, conflicts);
883 var second_parents = [_]version.Hash{first_commit.hash};
884 const second_commit = version.Commit.init(second_root.hash, &second_parents);
885
886 var history = try sql.History.open(std.testing.allocator, tmp.dir, .{
887 .path = history_path,
888 .recovery = .reject,
889 });
890 defer history.deinit();
891 try history.putDatabaseRoot(first_root);
892 try history.putCommit(first_commit);
893 try history.putRef(.{ .name = "main", .target = first_commit.hash });
894
895 var stale = (try loadExisting(
896 std.testing.allocator,
897 testing_io,
898 tmp.dir,
899 history_path,
900 )).?;
901 defer stale.deinit();
902
903 try history.putDatabaseRoot(second_root);
904 try history.putCommit(second_commit);
905 try history.putRef(.{ .name = "main", .target = second_commit.hash });
906 try history.putRef(.{ .name = "archive", .target = first_commit.hash });
907 const history_stat = try tmp.dir.statFile(testing_io, history_path, .{});
908 try std.testing.expect(stale.covered_length < history_stat.size);
909
910 var repaired = try advance(
911 std.testing.allocator,
912 testing_io,
913 tmp.dir,
914 history_path,
915 &stale,
916 history_stat.size,
917 .{},
918 );
919 defer repaired.deinit();
920 try std.testing.expectEqual(history_stat.size, repaired.covered_length);
921 const main = repaired.find("main") orelse return error.RefMissing;
922 try std.testing.expect(version.same(second_commit.hash, main.head));
923 try std.testing.expect(version.same(second_root.hash, main.root));
924 try std.testing.expect(version.same(second_root.conflicts, main.conflicts));
925 const archive = repaired.find("archive") orelse return error.RefMissing;
926 try std.testing.expect(version.same(first_commit.hash, archive.head));
927 try std.testing.expect(version.same(first_root.hash, archive.root));
928 try std.testing.expect(version.same(first_root.conflicts, archive.conflicts));
929 }
930
931 test "refs advance rejects a ref before its target commit" {
932 var tmp = std.testing.tmpDir(.{});
933 defer tmp.cleanup();
934
935 const history_path = "future-ref.history";
936 const conflicts = version.ConflictRoot.empty();
937 const first_root = version.DatabaseRoot.init(&.{}, conflicts);
938 const first_commit = version.Commit.init(first_root.hash, &.{});
939 const future_root = version.DatabaseRoot.init(&.{.{
940 .name = "future",
941 .hash = version.emptyHash("future-ref.relation"),
942 }}, conflicts);
943 var parents = [_]version.Hash{first_commit.hash};
944 const future_commit = version.Commit.init(future_root.hash, &parents);
945 var stale: Snapshot = undefined;
946 {
947 var history = try sql.History.open(std.testing.allocator, tmp.dir, .{
948 .path = history_path,
949 .recovery = .reject,
950 });
951 defer history.deinit();
952 try history.putDatabaseRoot(first_root);
953 try history.putCommit(first_commit);
954 try history.putRef(.{ .name = "main", .target = first_commit.hash });
955 stale = (try loadExisting(
956 std.testing.allocator,
957 testing_io,
958 tmp.dir,
959 history_path,
960 )).?;
961 try history.putDatabaseRoot(future_root);
962 var payload: std.ArrayList(u8) = .empty;
963 defer payload.deinit(std.testing.allocator);
964 try record_mod.appendHash(std.testing.allocator, &payload, future_commit.hash);
965 try record_mod.appendBytes(std.testing.allocator, &payload, "main");
966 try history.appendRecord(.ref, payload.items);
967 try history.putCommit(future_commit);
968 }
969 defer stale.deinit();
970 const history_stat = try tmp.dir.statFile(testing_io, history_path, .{});
971 try std.testing.expectError(
972 error.InvalidHistory,
973 advance(
974 std.testing.allocator,
975 testing_io,
976 tmp.dir,
977 history_path,
978 &stale,
979 history_stat.size,
980 .{},
981 ),
982 );
983 try std.testing.expectError(
984 error.TruncatedHistory,
985 sql.History.open(std.testing.allocator, tmp.dir, .{
986 .path = history_path,
987 .recovery = .reject,
988 }),
989 );
990 }
991
992 test "refs advance rejects a hash-valid malformed non-ref suffix" {
993 var tmp = std.testing.tmpDir(.{});
994 defer tmp.cleanup();
995
996 const history_path = "malformed-conflict.history";
997 const conflicts = version.ConflictRoot.empty();
998 const root = version.DatabaseRoot.init(&.{}, conflicts);
999 const commit = version.Commit.init(root.hash, &.{});
1000 var stale: Snapshot = undefined;
1001 {
1002 var history = try sql.History.open(std.testing.allocator, tmp.dir, .{
1003 .path = history_path,
1004 .recovery = .reject,
1005 });
1006 defer history.deinit();
1007 try history.putDatabaseRoot(root);
1008 try history.putCommit(commit);
1009 try history.putRef(.{ .name = "main", .target = commit.hash });
1010 stale = (try loadExisting(
1011 std.testing.allocator,
1012 testing_io,
1013 tmp.dir,
1014 history_path,
1015 )).?;
1016 try history.appendRecord(.conflict, &.{});
1017 }
1018 defer stale.deinit();
1019 const refs_before = try tmp.dir.readFileAlloc(
1020 testing_io,
1021 "malformed-conflict.history.refs",
1022 std.testing.allocator,
1023 .unlimited,
1024 );
1025 defer std.testing.allocator.free(refs_before);
1026 const history_stat = try tmp.dir.statFile(testing_io, history_path, .{});
1027 try std.testing.expectError(
1028 error.InvalidHistory,
1029 advance(
1030 std.testing.allocator,
1031 testing_io,
1032 tmp.dir,
1033 history_path,
1034 &stale,
1035 history_stat.size,
1036 .{},
1037 ),
1038 );
1039 const refs_after = try tmp.dir.readFileAlloc(
1040 testing_io,
1041 "malformed-conflict.history.refs",
1042 std.testing.allocator,
1043 .unlimited,
1044 );
1045 defer std.testing.allocator.free(refs_after);
1046 try std.testing.expectEqualSlices(u8, refs_before, refs_after);
1047 try std.testing.expectError(
1048 error.TruncatedHistory,
1049 sql.History.open(std.testing.allocator, tmp.dir, .{
1050 .path = history_path,
1051 .recovery = .reject,
1052 }),
1053 );
1054 }
1055
1056 test "refs advance streams opaque records above the metadata bound" {
1057 var tmp = std.testing.tmpDir(.{});
1058 defer tmp.cleanup();
1059
1060 const history_path = "opaque.history";
1061 const payload = try std.testing.allocator.alloc(u8, repair_metadata_record_bytes_max);
1062 defer std.testing.allocator.free(payload);
1063 @memset(payload, 0);
1064 const kind = record_mod.RecordKind.row_chunk;
1065 const expected = record_mod.recordHash(@backingInt(kind), payload);
1066 var header: [record_mod.record_header_size]u8 = @splat(0);
1067 std.mem.writeInt(u32, header[0..4], record_mod.magic, .big);
1068 std.mem.writeInt(u32, header[4..8], record_mod.format_version, .big);
1069 std.mem.writeInt(u32, header[8..12], @backingInt(kind), .big);
1070 std.mem.writeInt(u32, header[12..16], @intCast(payload.len), .big);
1071 @memcpy(header[16..][0..version.hash_bytes], &expected);
1072 var file = try tmp.dir.createFile(testing_io, history_path, .{ .read = true });
1073 defer file.close(testing_io);
1074 try file.writePositionalAll(testing_io, &header, 0);
1075 try file.writePositionalAll(testing_io, payload, header.len);
1076 const history_length = header.len + payload.len;
1077
1078 var empty = Snapshot{
1079 .allocator = std.testing.allocator,
1080 .covered_length = 0,
1081 .entries = try std.testing.allocator.alloc(Entry, 0),
1082 .names = try std.testing.allocator.alloc(u8, 0),
1083 };
1084 defer empty.deinit();
1085 var repaired = try advance(
1086 std.testing.allocator,
1087 testing_io,
1088 tmp.dir,
1089 history_path,
1090 &empty,
1091 history_length,
1092 .{},
1093 );
1094 defer repaired.deinit();
1095 try std.testing.expectEqual(@as(usize, 0), repaired.entries.len);
1096 try std.testing.expectEqual(@as(u64, history_length), repaired.covered_length);
1097 }
1098
1099 test "refs advance rejects metadata above the repair bound" {
1100 var tmp = std.testing.tmpDir(.{});
1101 defer tmp.cleanup();
1102
1103 const history_path = "oversized.history";
1104 var header: [record_mod.record_header_size]u8 = @splat(0);
1105 std.mem.writeInt(u32, header[0..4], record_mod.magic, .big);
1106 std.mem.writeInt(u32, header[4..8], record_mod.format_version, .big);
1107 std.mem.writeInt(u32, header[8..12], @backingInt(record_mod.RecordKind.commit), .big);
1108 const oversized_payload_bytes = repair_metadata_record_bytes_max + 1;
1109 std.mem.writeInt(u32, header[12..16], oversized_payload_bytes, .big);
1110 var file = try tmp.dir.createFile(testing_io, history_path, .{ .read = true });
1111 defer file.close(testing_io);
1112 try file.writePositionalAll(testing_io, &header, 0);
1113 const history_length = record_mod.record_header_size + oversized_payload_bytes;
1114 try file.setLength(testing_io, history_length);
1115
1116 var empty = Snapshot{
1117 .allocator = std.testing.allocator,
1118 .covered_length = 0,
1119 .entries = try std.testing.allocator.alloc(Entry, 0),
1120 .names = try std.testing.allocator.alloc(u8, 0),
1121 };
1122 defer empty.deinit();
1123 try std.testing.expectError(
1124 error.StreamTooLong,
1125 advance(
1126 std.testing.allocator,
1127 testing_io,
1128 tmp.dir,
1129 history_path,
1130 &empty,
1131 history_length,
1132 .{},
1133 ),
1134 );
1135 }
1136
1137 test "refs snapshot rejects the retired format version" {
1138 const head = @as([version.hash_bytes]u8, @splat(0x33));
1139 const entries = [_]Entry{.{
1140 .name = "main",
1141 .head = head,
1142 .root = head,
1143 .conflicts = head,
1144 }};
1145 const bytes = try encode(std.testing.allocator, 9, &entries);
1146 defer std.testing.allocator.free(bytes);
1147 std.mem.writeInt(u32, bytes[4..8], 1, .big);
1148 std.crypto.hash.sha2.Sha256.hash(
1149 bytes[0 .. bytes.len - digest_size],
1150 bytes[bytes.len - digest_size ..][0..digest_size],
1151 .{},
1152 );
1153 try std.testing.expect((try decode(std.testing.allocator, bytes)) == null);
1154 }