lib/sql/src/history/store.zig
daab053ee43316e1809a84551d573ddd1e5bf3d2
1 const std = @import("std");
2 const sql = @import("../root.zig");
3 const materialize_mod = @import("materialize.zig");
4 const instrumentation = @import("instrumentation.zig");
5 const pack_mod = @import("pack.zig");
6 const recover_mod = @import("recover.zig");
7 const conflict_mod = @import("conflict.zig");
8 const record_mod = @import("record.zig");
9 const refs_mod = @import("refs.zig");
10 const branch = sql.branch;
11 const catalog_mod = sql.catalog;
12 const chunk_mod = sql.chunk;
13 const diff_mod = sql.diff;
14 const file_mod = sql.file;
15 const key_mod = sql.key;
16 const row = sql.row;
17 const tree = sql.tree;
18 const version = sql.version;
19
20 const Allocator = std.mem.Allocator;
21 const testing_io = std.Options.debug_io;
22
23 pub const Error = Allocator.Error || branch.Error || std.Io.File.OpenError || std.Io.File.StatError || std.Io.File.ReadPositionalError || std.Io.File.WritePositionalError || std.Io.File.SetLengthError || std.Io.File.SyncError || std.Io.Dir.ReadFileAllocError || error{
24 Interrupted,
25 InvalidHistory,
26 TruncatedHistory,
27 HistoryNotFound,
28 DatabaseRootNotFound,
29 RelationRootNotFound,
30 RelationRowsNotFound,
31 RefExists,
32 RefNotFound,
33 RefChanged,
34 RecoveryRequired,
35 ReadOnly,
36 ConflictRootNotFound,
37 ConflictArtifactNotFound,
38 };
39
40 pub const Options = struct {
41 io: std.Io = std.Options.debug_io,
42 path: []const u8 = "tiny.sql.history",
43 create: bool = true,
44 read_only: bool = false,
45 recovery: RecoveryPolicy,
46 replay_instrumentation: ?*instrumentation.Recorder = null,
47 control: sql.wal.Control = .{},
48 };
49
50 pub const RecoveryPolicy = enum {
51 reject,
52 truncate,
53 };
54
55 pub const Truncation = struct {
56 original_length: usize,
57 valid_length: usize,
58 };
59
60 pub const Recovery = union(enum) {
61 clean,
62 truncated: Truncation,
63 };
64
65 pub const WriteIo = struct {
66 writes: usize = 0,
67 /// Truncations of the history file after open. An append ends the file
68 /// where its write stops, so only a failed append truncates it.
69 resizes: usize = 0,
70 syncs: usize = 0,
71 };
72
73 pub const FastForwardDecision = enum {
74 pending,
75 baseline,
76 target,
77 };
78
79 pub const FastForwardRecovery = struct {
80 id: version.Hash,
81 name: []const u8,
82 expected: version.Hash,
83 target: version.Hash,
84 decision: FastForwardDecision = .pending,
85 };
86
87 pub const FastForwardUpdate = struct {
88 history: *History,
89 id: version.Hash,
90
91 pub fn commit(self: *FastForwardUpdate) Error!void {
92 try self.history.decideFastForward(self.id, .target);
93 }
94
95 pub fn abort(self: *FastForwardUpdate) Error!void {
96 try self.history.decideFastForward(self.id, .baseline);
97 }
98
99 pub fn complete(self: *FastForwardUpdate) Error!void {
100 try self.history.completeFastForward(self.id);
101 }
102 };
103
104 pub fn freeRefList(allocator: Allocator, refs: []version.Ref) void {
105 for (refs) |ref_value| allocator.free(ref_value.name);
106 if (refs.len != 0) allocator.free(refs);
107 }
108
109 pub const DatabaseRootView = struct {
110 conflicts: version.Hash,
111 entries: []const version.RelationEntry,
112 };
113
114 pub const RelationKeysView = struct {
115 allocator: Allocator,
116 owned: ?[]?version.Hash = null,
117 table_key: ?version.Hash,
118 index_keys: []const ?version.Hash,
119
120 pub fn deinit(self: *RelationKeysView) void {
121 if (self.owned) |owned| self.allocator.free(owned);
122 self.* = undefined;
123 }
124 };
125
126 pub const History = struct {
127 pub const ByteRange = struct {
128 pub const Count = u64;
129 pub const unit_precision_bytes: Count = 1;
130 pub const maximum_units_per_second: Count = 512 * 1024 * 1024;
131 pub const maximum_seconds_per_year: Count = 366 * 24 * 60 * 60;
132 pub const service_lifetime_years: Count = 1_000;
133 pub const service_lifetime_seconds: Count =
134 maximum_seconds_per_year * service_lifetime_years;
135 pub const maximum_append_bytes: Count =
136 @as(Count, record_mod.record_header_size) + std.math.maxInt(u32);
137 pub const budget_bytes: Count = total(
138 maximum_units_per_second,
139 service_lifetime_seconds,
140 ).?;
141
142 pub fn total(
143 units_per_second: Count,
144 lifetime_seconds: Count,
145 ) ?Count {
146 const bytes_per_second = std.math.mul(
147 Count,
148 unit_precision_bytes,
149 units_per_second,
150 ) catch return null;
151 return std.math.mul(Count, bytes_per_second, lifetime_seconds) catch null;
152 }
153 };
154
155 comptime {
156 std.debug.assert(
157 @as(u128, ByteRange.budget_bytes) +
158 @as(u128, ByteRange.maximum_append_bytes) <=
159 std.math.maxInt(ByteRange.Count),
160 );
161 std.debug.assert(@bitSizeOf(usize) >= @bitSizeOf(ByteRange.Count));
162 }
163
164 allocator: Allocator,
165 io: std.Io,
166 file: ?std.Io.File = null,
167 read_only: bool = false,
168 sidecar_dir: ?std.Io.Dir = null,
169 sidecar_path: ?[]u8 = null,
170 /// The length of the history file once pending records are written.
171 /// The file holds every byte before the pending records and nothing
172 /// after them, so writing them ends the file where the write stops.
173 bytes_written: usize = 0,
174 recovery: Recovery = .clean,
175 needs_sync: bool = false,
176 write_batch_depth: usize = 0,
177 pending_write: std.ArrayList(u8) = .empty,
178 write_io: WriteIo = .{},
179 fast_forward: ?FastForwardRecovery = null,
180 recovery_required: bool = false,
181 database_roots: std.ArrayList(record_mod.DatabaseRootRecord) = .empty,
182 relation_roots: std.ArrayList(record_mod.RelationRootRecord) = .empty,
183 materialized_relation_roots: std.ArrayList(record_mod.MaterializedRelationRoot) = .empty,
184 relation_rows: std.ArrayList(record_mod.RelationRowsRecord) = .empty,
185 relation_spans: std.ArrayList(record_mod.RelationSpansRecord) = .empty,
186 row_chunks: std.ArrayList(record_mod.RowChunkRecord) = .empty,
187 index_pages: std.ArrayList(record_mod.IndexPageRecord) = .empty,
188 tree_nodes: std.ArrayList(record_mod.TreeNodeRecord) = .empty,
189 commits: std.ArrayList(record_mod.CommitRecord) = .empty,
190 refs: std.ArrayList(record_mod.RefRecord) = .empty,
191 conflicts: std.ArrayList(record_mod.ConflictRecord) = .empty,
192 conflict_roots: std.ArrayList(record_mod.ConflictRootRecord) = .empty,
193 lookup: RecordLookup = .{},
194
195 pub const SuffixPlan = struct {
196 reused: usize,
197 boundary: i64,
198 };
199
200 pub const RowsNeed = union(enum) {
201 none,
202 full,
203 suffix: SuffixPlan,
204 };
205
206 pub fn open(allocator: Allocator, dir: std.Io.Dir, options: Options) Error!History {
207 var history = History{
208 .allocator = allocator,
209 .io = options.io,
210 .read_only = options.read_only,
211 };
212 errdefer history.deinit();
213
214 const recorded_length: ?usize = if (dir.statFile(options.io, options.path, .{})) |stat|
215 @intCast(stat.size)
216 else |err| switch (err) {
217 error.FileNotFound => if (options.create) null else return error.HistoryNotFound,
218 else => return err,
219 };
220
221 history.file = if (options.read_only)
222 try dir.openFile(options.io, options.path, .{})
223 else
224 try dir.createFile(options.io, options.path, .{ .read = true, .truncate = false });
225 if (!options.read_only) {
226 if (dir.openDir(options.io, ".", .{})) |owned_dir| {
227 history.sidecar_dir = owned_dir;
228 history.sidecar_path = try allocator.dupe(u8, options.path);
229 } else |_| {}
230 }
231 if (recorded_length) |length| {
232 history.bytes_written = try recover_mod.replayFile(
233 &history,
234 length,
235 options.replay_instrumentation,
236 options.control,
237 );
238 if (length != history.bytes_written) {
239 if (options.read_only or options.recovery == .reject) return error.TruncatedHistory;
240 try history.file.?.setLength(options.io, history.bytes_written);
241 try history.file.?.sync(options.io);
242 history.recovery = .{ .truncated = .{
243 .original_length = length,
244 .valid_length = history.bytes_written,
245 } };
246 }
247 history.refreshRefsSidecar();
248 }
249 return history;
250 }
251
252 pub fn deinit(self: *History) void {
253 if (self.file) |*file| {
254 if (!self.recovery_required) {
255 if (self.pending_write.items.len != 0) self.flushPendingWrite() catch {};
256 if (self.needs_sync) file.sync(self.io) catch {};
257 }
258 file.close(self.io);
259 }
260 if (self.sidecar_dir) |*dir| dir.close(self.io);
261 if (self.sidecar_path) |path| self.allocator.free(path);
262 for (self.database_roots.items) |*record| record.deinit();
263 for (self.materialized_relation_roots.items) |*record| record.deinit();
264 for (self.relation_rows.items) |*record| record.deinit(self.allocator);
265 for (self.relation_spans.items) |*record| record.deinit(self.allocator);
266 for (self.index_pages.items) |*record| record.deinit(self.allocator);
267 for (self.commits.items) |*record| record.deinit(self.allocator);
268 for (self.refs.items) |*record| record.deinit(self.allocator);
269 for (self.conflicts.items) |*record| record.deinit(self.allocator);
270 for (self.conflict_roots.items) |*record| record.deinit(self.allocator);
271 self.database_roots.deinit(self.allocator);
272 self.relation_roots.deinit(self.allocator);
273 self.materialized_relation_roots.deinit(self.allocator);
274 self.relation_rows.deinit(self.allocator);
275 self.relation_spans.deinit(self.allocator);
276 self.row_chunks.deinit(self.allocator);
277 self.index_pages.deinit(self.allocator);
278 self.tree_nodes.deinit(self.allocator);
279 self.commits.deinit(self.allocator);
280 self.refs.deinit(self.allocator);
281 self.conflicts.deinit(self.allocator);
282 self.conflict_roots.deinit(self.allocator);
283 self.lookup.deinit(self.allocator);
284 self.pending_write.deinit(self.allocator);
285 self.* = undefined;
286 }
287
288 pub fn putDatabaseRoot(self: *History, root: version.DatabaseRoot) Error!void {
289 try self.ensureUsable();
290 if (self.findDatabaseRoot(root.hash) != null) return;
291 var owned = try root.clone(self.allocator);
292 errdefer owned.deinit();
293 try self.database_roots.ensureUnusedCapacity(self.allocator, 1);
294 try self.lookup.database_roots.ensureUnusedCapacity(self.allocator, 1);
295 try self.appendDatabaseRootRecord(owned);
296 self.lookup.database_roots.putAssumeCapacity(owned.hash, self.database_roots.items.len);
297 self.database_roots.appendAssumeCapacity(.{ .root = owned });
298 try self.flushSync();
299 }
300
301 pub fn putRelationRoot(self: *History, root: version.RelationRoot) Error!void {
302 try self.ensureUsable();
303 if (self.hasRelationRoot(root.hash)) return;
304 var owned = try root.clone(self.allocator);
305 var owned_transferred = false;
306 errdefer if (!owned_transferred) owned.deinit();
307 var stage = NodeStage.init(self);
308 defer stage.deinit();
309 const index_keys = try self.allocator.alloc(?version.Hash, owned.indexes.len);
310 var index_keys_owned = true;
311 errdefer if (index_keys_owned) self.allocator.free(index_keys);
312 const table_key = try stage.mapKey(&owned.table);
313 for (owned.indexes, index_keys) |*index, *index_key| index_key.* = try stage.mapKey(&index.map);
314 try stage.flush();
315 try self.relation_roots.ensureUnusedCapacity(self.allocator, 1);
316 try self.materialized_relation_roots.ensureUnusedCapacity(self.allocator, 1);
317 try self.lookup.relation_roots.ensureUnusedCapacity(self.allocator, 1);
318 try self.appendRelationRootRecord(owned, table_key, index_keys);
319 materialize_mod.shrinkRelationRootMaps(self.allocator, &owned);
320 self.lookup.relation_roots.putAssumeCapacity(owned.hash, self.relation_roots.items.len);
321 self.relation_roots.appendAssumeCapacity(.{
322 .hash = owned.hash,
323 .storage = .{ .materialized = self.materialized_relation_roots.items.len },
324 });
325 self.materialized_relation_roots.appendAssumeCapacity(.{
326 .root = owned,
327 .table_key = table_key,
328 .index_keys = index_keys,
329 });
330 owned_transferred = true;
331 index_keys_owned = false;
332 try self.flushSync();
333 }
334
335 pub fn putRelationRows(self: *History, root: version.Hash, rows: []const version.RelationRow) Error!void {
336 try self.ensureUsable();
337 if (self.hasRelationRows(root)) return;
338 const sorted = try self.allocator.dupe(version.RelationRow, rows);
339 defer if (sorted.len != 0) self.allocator.free(sorted);
340 std.mem.sort(version.RelationRow, sorted, {}, version.relationRowLessThan);
341 var writer = try self.beginRelationRows(root);
342 defer writer.deinit();
343 for (sorted) |row_value| try writer.append(row_value.rowid, row_value.bytes);
344 try writer.finish();
345 }
346
347 pub fn beginRelationRows(self: *History, root: version.Hash) Error!RelationRowsWriter {
348 try self.ensureMutable();
349 std.debug.assert(!self.hasRelationRows(root));
350 return .{ .history = self, .root = root, .boundary = null };
351 }
352
353 pub fn beginRelationRowsSuffix(
354 self: *History,
355 root: version.Hash,
356 base: IncrementalBase,
357 plan: SuffixPlan,
358 ) Error!RelationRowsWriter {
359 try self.ensureUsable();
360 std.debug.assert(!self.hasRelationRows(root));
361 var base_pages = (try self.relationRowsPages(self.allocator, base.root)) orelse return error.InvalidHistory;
362 defer base_pages.deinit();
363 var base_spans = (try self.relationSpans(self.allocator, base.root)) orelse return error.InvalidHistory;
364 defer base_spans.deinit();
365 if (plan.reused == 0 or plan.reused >= base_spans.items.len) return error.InvalidHistory;
366 if (base_spans.items[plan.reused - 1].last != plan.boundary) return error.InvalidHistory;
367
368 const base_chunks = try self.baseChunkList(self.allocator, base_pages.items);
369 defer if (base_chunks.len != 0) self.allocator.free(base_chunks);
370 if (base_chunks.len != base_spans.items.len) return error.InvalidHistory;
371
372 var writer = RelationRowsWriter{ .history = self, .root = root, .boundary = plan.boundary };
373 errdefer writer.deinit();
374 try writer.chunks.appendSlice(self.allocator, base_chunks[0..plan.reused]);
375 try writer.spans.appendSlice(self.allocator, base_spans.items[0..plan.reused]);
376 return writer;
377 }
378
379 pub fn relationRowsNeed(self: *const History, allocator: Allocator, root: version.Hash, base: ?IncrementalBase) Error!RowsNeed {
380 try self.ensureUsable();
381 if (self.hasRelationRows(root)) return .none;
382 const info = base orelse return .full;
383 var base_pages = (try self.relationRowsPages(allocator, info.root)) orelse return .full;
384 defer base_pages.deinit();
385 var base_spans = (try self.relationSpans(allocator, info.root)) orelse return .full;
386 defer base_spans.deinit();
387 if (base_spans.items.len == 0) return .full;
388 const base_chunks = try self.baseChunkList(allocator, base_pages.items);
389 defer allocator.free(base_chunks);
390 if (base_chunks.len != base_spans.items.len) return .full;
391
392 const first_affected = firstAffectedSpan(base_spans.items, info.edited);
393 if (first_affected == 0) return .full;
394 return .{ .suffix = .{
395 .reused = first_affected,
396 .boundary = base_spans.items[first_affected - 1].last,
397 } };
398 }
399
400 pub fn putRelationRowsSuffix(self: *History, root: version.Hash, base: IncrementalBase, plan: SuffixPlan, tail: []const version.RelationRow) Error!void {
401 try self.ensureUsable();
402 if (self.hasRelationRows(root)) return;
403 if (!relationRowsSorted(tail)) return error.InvalidHistory;
404 var writer = try self.beginRelationRowsSuffix(root, base, plan);
405 defer writer.deinit();
406 for (tail) |row_value| try writer.append(row_value.rowid, row_value.bytes);
407 try writer.finish();
408 }
409
410 pub fn finishRelationRows(self: *History, root: version.Hash, chunks: []const version.Hash, spans: []const record_mod.ChunkSpan) Error!void {
411 try self.ensureUsable();
412 const page_bounds = try chunk_mod.pageBoundaries(self.allocator, chunks);
413 defer if (page_bounds.len != 0) self.allocator.free(page_bounds);
414 const pages = try self.allocator.alloc(version.Hash, page_bounds.len);
415 errdefer if (pages.len != 0) self.allocator.free(pages);
416 for (page_bounds, pages) |bound, *digest| {
417 const slice = chunks[bound.start..bound.end];
418 digest.* = chunk_mod.pageDigest(slice);
419 try self.putIndexPage(digest.*, slice);
420 }
421 try self.relation_rows.ensureUnusedCapacity(self.allocator, 1);
422 try self.lookup.relation_rows.ensureUnusedCapacity(self.allocator, 1);
423 try self.appendRelationRowsRecord(root, pages);
424 self.lookup.relation_rows.putAssumeCapacity(root, self.relation_rows.items.len);
425 self.relation_rows.appendAssumeCapacity(.{
426 .root = root,
427 .storage = .{ .materialized = pages },
428 });
429 try self.putRelationSpans(root, spans);
430 try self.flushSync();
431 }
432
433 pub fn relationSpans(self: *const History, allocator: Allocator, hash: version.Hash) Error!?record_mod.ChunkSpanView {
434 try self.ensureUsable();
435 const index = self.lookup.relation_spans.get(hash) orelse return null;
436 const record = &self.relation_spans.items[index];
437 return try materialize_mod.readRelationSpansRecord(self, allocator, record.root, record.storage);
438 }
439
440 pub fn putRelationSpans(self: *History, root: version.Hash, spans: []const record_mod.ChunkSpan) Error!void {
441 try self.ensureUsable();
442 if (self.lookup.relation_spans.get(root) != null) return;
443 const owned = try self.allocator.dupe(record_mod.ChunkSpan, spans);
444 errdefer self.allocator.free(owned);
445 try self.relation_spans.ensureUnusedCapacity(self.allocator, 1);
446 try self.lookup.relation_spans.ensureUnusedCapacity(self.allocator, 1);
447 try self.appendRelationSpansRecord(root, owned);
448 self.lookup.relation_spans.putAssumeCapacity(root, self.relation_spans.items.len);
449 self.relation_spans.appendAssumeCapacity(.{
450 .root = root,
451 .storage = .{ .materialized = owned },
452 });
453 }
454
455 pub fn appendRelationSpansRecord(self: *History, root: version.Hash, spans: []const record_mod.ChunkSpan) Error!void {
456 try self.ensureUsable();
457 if (spans.len > std.math.maxInt(u32)) return error.InvalidHistory;
458 var payload: std.ArrayList(u8) = .empty;
459 defer payload.deinit(self.allocator);
460 try record_mod.appendHash(self.allocator, &payload, root);
461 try record_mod.appendU32(self.allocator, &payload, @intCast(spans.len));
462 for (spans) |span| {
463 try record_mod.appendU64(self.allocator, &payload, @as(u64, @bitCast(span.first)));
464 try record_mod.appendU64(self.allocator, &payload, @as(u64, @bitCast(span.last)));
465 }
466 try self.appendRecord(.relation_spans, payload.items);
467 }
468
469 pub fn baseChunkList(self: *const History, allocator: Allocator, pages: []const version.Hash) Error![]version.Hash {
470 try self.ensureUsable();
471 var chunks: std.ArrayList(version.Hash) = .empty;
472 errdefer chunks.deinit(allocator);
473 for (pages) |page_digest| {
474 var page = (try self.indexPageChunks(allocator, page_digest)) orelse return error.InvalidHistory;
475 defer page.deinit();
476 try chunks.appendSlice(allocator, page.items);
477 }
478 return try chunks.toOwnedSlice(allocator);
479 }
480
481 pub fn putRowChunk(self: *History, digest: version.Hash, rows: []const version.RelationRow) Error!void {
482 try self.ensureUsable();
483 if (self.findRowChunk(digest) != null) return;
484 var payload: std.ArrayList(u8) = .empty;
485 defer payload.deinit(self.allocator);
486 try record_mod.appendHash(self.allocator, &payload, digest);
487 try record_mod.appendRelationRows(self.allocator, &payload, rows);
488 try self.row_chunks.ensureUnusedCapacity(self.allocator, 1);
489 try self.lookup.row_chunks.ensureUnusedCapacity(self.allocator, 1);
490 const payload_offset = self.bytes_written + record_mod.record_header_size;
491 const expected = record_mod.recordHash(@backingInt(record_mod.RecordKind.row_chunk), payload.items);
492 try self.appendRecord(.row_chunk, payload.items);
493 self.lookup.row_chunks.putAssumeCapacity(digest, self.row_chunks.items.len);
494 self.row_chunks.appendAssumeCapacity(.{
495 .digest = digest,
496 .payload = .{
497 .expected = expected,
498 .offset = payload_offset,
499 .len = payload.items.len,
500 },
501 });
502 }
503
504 pub fn putIndexPage(self: *History, digest: version.Hash, chunks: []const version.Hash) Error!void {
505 try self.ensureUsable();
506 if (self.hasIndexPage(digest)) return;
507 const owned = try self.allocator.dupe(version.Hash, chunks);
508 errdefer self.allocator.free(owned);
509 try self.index_pages.ensureUnusedCapacity(self.allocator, 1);
510 try self.lookup.index_pages.ensureUnusedCapacity(self.allocator, 1);
511 try self.appendChunkIndexPageRecord(digest, owned);
512 self.lookup.index_pages.putAssumeCapacity(digest, self.index_pages.items.len);
513 self.index_pages.appendAssumeCapacity(.{
514 .digest = digest,
515 .storage = .{ .materialized = owned },
516 });
517 }
518
519 pub fn putDatabaseValue(self: *History, value: *const version.DatabaseValue) Error!void {
520 try self.ensureUsable();
521 var batch = try self.beginWriteBatch();
522 errdefer batch.deinit();
523 for (value.relations) |relation| {
524 try self.putRelationRoot(relation.root);
525 try self.putRelationRows(relation.root.hash, relation.rows);
526 }
527 try self.putDatabaseRoot(value.root);
528 try batch.finish();
529 }
530
531 pub fn putCommit(self: *History, commit: version.Commit) Error!void {
532 try self.ensureUsable();
533 const canonical = version.Commit.init(commit.root, commit.parents);
534 if (self.findCommit(canonical.hash) != null) return;
535 const parents = try self.allocator.dupe(version.Hash, commit.parents);
536 errdefer self.allocator.free(parents);
537 try self.commits.ensureUnusedCapacity(self.allocator, 1);
538 try self.lookup.commits.ensureUnusedCapacity(self.allocator, 1);
539 try self.appendCommitRecord(canonical.root, parents);
540 try self.flushSync();
541 self.lookup.commits.putAssumeCapacity(canonical.hash, self.commits.items.len);
542 self.commits.appendAssumeCapacity(.{
543 .parents = parents,
544 .commit = .{
545 .root = canonical.root,
546 .parents = parents,
547 .hash = canonical.hash,
548 },
549 });
550 }
551
552 pub fn putRef(self: *History, ref_value: version.Ref) Error!void {
553 try self.ensureUsable();
554 if (self.findCommit(ref_value.target) == null) return error.CommitNotFound;
555 if (self.findRef(ref_value.name)) |record| {
556 try self.appendRefRecord(ref_value.name, ref_value.target);
557 try self.flushSync();
558 record.ref.target = ref_value.target;
559 self.refreshRefsSidecar();
560 return;
561 }
562
563 const name = try self.allocator.dupe(u8, ref_value.name);
564 errdefer self.allocator.free(name);
565 try self.refs.ensureUnusedCapacity(self.allocator, 1);
566 try self.appendRefRecord(name, ref_value.target);
567 try self.flushSync();
568 self.refs.appendAssumeCapacity(.{
569 .name = name,
570 .ref = .{
571 .name = name,
572 .target = ref_value.target,
573 },
574 });
575 self.refreshRefsSidecar();
576 }
577
578 pub fn putRefIfMatches(self: *History, ref_value: version.Ref, expected: ?version.Hash) Error!void {
579 try self.ensureUsable();
580 if (self.findCommit(ref_value.target) == null) return error.CommitNotFound;
581 if (self.findRef(ref_value.name)) |record| {
582 const expected_hash = expected orelse return error.RefChanged;
583 if (!version.same(record.ref.target, expected_hash)) return error.RefChanged;
584 } else if (expected != null) {
585 return error.RefChanged;
586 }
587 try self.putRef(ref_value);
588 }
589
590 pub fn beginFastForward(
591 self: *History,
592 name: []const u8,
593 expected: version.Hash,
594 target: version.Hash,
595 ) Error!FastForwardUpdate {
596 try self.ensureUsable();
597 if (self.fast_forward != null) return error.RecoveryRequired;
598 if (self.pending_write.items.len != 0 or
599 self.needs_sync or
600 self.write_batch_depth != 0)
601 {
602 self.poison();
603 return error.RecoveryRequired;
604 }
605 const ref_record = self.findRef(name) orelse return error.RefNotFound;
606 if (!version.same(ref_record.ref.target, expected)) return error.RefChanged;
607 if (self.findCommit(expected) == null or self.findCommit(target) == null) {
608 return error.CommitNotFound;
609 }
610 var payload: std.ArrayList(u8) = .empty;
611 defer payload.deinit(self.allocator);
612 try record_mod.appendBytes(self.allocator, &payload, name);
613 try record_mod.appendHash(self.allocator, &payload, expected);
614 try record_mod.appendHash(self.allocator, &payload, target);
615 const id = record_mod.recordHash(
616 @backingInt(record_mod.RecordKind.fast_forward_prepare),
617 payload.items,
618 );
619 try self.appendCoordinatorRecord(.fast_forward_prepare, payload.items);
620 self.fast_forward = .{
621 .id = id,
622 .name = ref_record.name,
623 .expected = expected,
624 .target = target,
625 };
626 return .{ .history = self, .id = id };
627 }
628
629 pub fn fastForwardRecovery(self: *const History) ?*const FastForwardRecovery {
630 return if (self.fast_forward) |*recovery| recovery else null;
631 }
632
633 pub fn poison(self: *History) void {
634 self.recovery_required = true;
635 }
636
637 pub fn requiresRecovery(self: *const History) bool {
638 return self.recovery_required;
639 }
640
641 fn decideFastForward(
642 self: *History,
643 id: version.Hash,
644 decision: FastForwardDecision,
645 ) Error!void {
646 try self.ensureUsable();
647 const recovery = if (self.fast_forward) |*active| active else return error.InvalidHistory;
648 if (!version.same(recovery.id, id) or recovery.decision != .pending) {
649 return error.InvalidHistory;
650 }
651 const ref_record = self.findRef(recovery.name) orelse return error.InvalidHistory;
652 if (!version.same(ref_record.ref.target, recovery.expected)) return error.InvalidHistory;
653 const kind: record_mod.RecordKind = switch (decision) {
654 .pending => return error.InvalidHistory,
655 .baseline => .fast_forward_abort,
656 .target => .fast_forward_commit,
657 };
658 try self.appendCoordinatorHashRecord(kind, id);
659 if (decision == .target) ref_record.ref.target = recovery.target;
660 recovery.decision = decision;
661 }
662
663 fn completeFastForward(self: *History, id: version.Hash) Error!void {
664 try self.ensureUsable();
665 const recovery = if (self.fast_forward) |*active| active else return error.InvalidHistory;
666 if (!version.same(recovery.id, id) or recovery.decision == .pending) {
667 return error.InvalidHistory;
668 }
669 const ref_record = self.findRef(recovery.name) orelse return error.InvalidHistory;
670 const selected = switch (recovery.decision) {
671 .pending => unreachable,
672 .baseline => recovery.expected,
673 .target => recovery.target,
674 };
675 if (!version.same(ref_record.ref.target, selected)) return error.InvalidHistory;
676 try self.appendCoordinatorHashRecord(.fast_forward_complete, id);
677 self.fast_forward = null;
678 self.refreshRefsSidecar();
679 }
680
681 fn ensureUsable(self: *const History) Error!void {
682 if (self.recovery_required) return error.RecoveryRequired;
683 }
684
685 fn ensureMutable(self: *const History) Error!void {
686 try self.ensureUsable();
687 if (self.read_only) return error.ReadOnly;
688 if (self.fast_forward != null) return error.RecoveryRequired;
689 }
690
691 pub fn deleteRef(self: *History, name: []const u8) Error!void {
692 try self.ensureUsable();
693 const index = self.findRefIndex(name) orelse return error.RefNotFound;
694 try self.appendRefDeleteRecord(name);
695 try self.flushSync();
696 var removed = self.refs.orderedRemove(index);
697 removed.deinit(self.allocator);
698 self.refreshRefsSidecar();
699 }
700
701 pub fn createBranch(self: *History, name: []const u8, target: version.Hash) Error!version.Ref {
702 try self.ensureUsable();
703 if (self.findRef(name) != null) return error.RefExists;
704 if (self.findCommit(target) == null) return error.CommitNotFound;
705 const ref_value = version.Ref{
706 .name = name,
707 .target = target,
708 };
709 try self.putRef(ref_value);
710 return (try self.ref(name)).?;
711 }
712
713 pub fn checkoutBranch(self: *const History, name: []const u8) Error!branch.Checkout {
714 try self.ensureUsable();
715 const record = self.findRef(name) orelse return error.RefNotFound;
716 const commit = self.findCommit(record.ref.target) orelse return error.CommitNotFound;
717 return branch.checkout(record.ref, commit.commit.root);
718 }
719
720 pub fn fastForwardBranch(self: *History, allocator: Allocator, name: []const u8, target: version.Hash) Error!void {
721 try self.ensureUsable();
722 const record = self.findRef(name) orelse return error.RefNotFound;
723 var ref_value = record.ref;
724 const entries = try self.commitEntries(allocator);
725 defer allocator.free(entries);
726 try branch.fastForwardRef(allocator, entries, &ref_value, target);
727 try self.putRef(ref_value);
728 }
729
730 pub fn commitBranch(self: *History, name: []const u8, root: version.Hash) Error!version.Hash {
731 try self.ensureUsable();
732 const record = self.findRef(name) orelse return error.RefNotFound;
733 if (self.findCommit(record.ref.target) == null) return error.CommitNotFound;
734 var parents = [_]version.Hash{record.ref.target};
735 const commit = version.Commit.init(root, parents[0..]);
736 try self.putCommitAndRef(commit, record, commit.hash);
737 return commit.hash;
738 }
739
740 pub fn mergeCommitBranch(self: *History, name: []const u8, root: version.Hash, theirs: version.Hash) Error!version.Hash {
741 try self.ensureUsable();
742 const record = self.findRef(name) orelse return error.RefNotFound;
743 if (self.findCommit(record.ref.target) == null) return error.CommitNotFound;
744 if (self.findCommit(theirs) == null) return error.CommitNotFound;
745 var parents = [_]version.Hash{ record.ref.target, theirs };
746 const commit = version.Commit.init(root, parents[0..]);
747 try self.putCommitAndRef(commit, record, commit.hash);
748 return commit.hash;
749 }
750
751 pub fn putCommitAndRef(self: *History, commit: version.Commit, ref_record: *record_mod.RefRecord, target: version.Hash) Error!void {
752 try self.ensureUsable();
753 const canonical = version.Commit.init(commit.root, commit.parents);
754 const append_commit = self.findCommit(canonical.hash) == null;
755 var parents: []version.Hash = &.{};
756 if (append_commit) parents = try self.allocator.dupe(version.Hash, canonical.parents);
757 var parents_live = append_commit;
758 errdefer if (parents_live) self.allocator.free(parents);
759 if (append_commit) {
760 try self.commits.ensureUnusedCapacity(self.allocator, 1);
761 try self.lookup.commits.ensureUnusedCapacity(self.allocator, 1);
762 try self.appendCommitRecord(canonical.root, parents);
763 }
764 try self.appendRefRecord(ref_record.name, target);
765 try self.flushSync();
766 if (append_commit) {
767 self.lookup.commits.putAssumeCapacity(canonical.hash, self.commits.items.len);
768 self.commits.appendAssumeCapacity(.{
769 .parents = parents,
770 .commit = .{
771 .root = canonical.root,
772 .parents = parents,
773 .hash = canonical.hash,
774 },
775 });
776 parents_live = false;
777 }
778 ref_record.ref.target = target;
779 self.refreshRefsSidecar();
780 }
781
782 pub fn putConflict(self: *History, artifact: version.ConflictArtifact) Error!void {
783 try self.ensureUsable();
784 const canonical = conflict_mod.canonicalConflictArtifact(artifact);
785 if (self.findConflict(canonical.hash) != null) return;
786 var owned = try conflict_mod.cloneConflictArtifact(self.allocator, artifact);
787 errdefer conflict_mod.deinitConflictArtifact(self.allocator, &owned);
788 try self.conflicts.ensureUnusedCapacity(self.allocator, 1);
789 try self.lookup.conflicts.ensureUnusedCapacity(self.allocator, 1);
790 try self.appendConflictRecord(canonical);
791 try self.flushSync();
792 self.lookup.conflicts.putAssumeCapacity(canonical.hash, self.conflicts.items.len);
793 self.conflicts.appendAssumeCapacity(.{ .artifact = owned });
794 }
795
796 pub fn putConflictRoot(self: *History, entries: []const version.ConflictEntry) Error!version.ConflictRoot {
797 try self.ensureUsable();
798 if (entries.len == 0) return version.ConflictRoot.empty();
799 const owned = try conflict_mod.cloneConflictEntriesSorted(self.allocator, entries);
800 var owned_live = true;
801 errdefer if (owned_live) conflict_mod.deinitConflictEntries(self.allocator, owned);
802 try self.validateConflictEntries(owned);
803 const root = version.ConflictRoot.init(owned);
804 if (self.findConflictRoot(root.hash) != null) {
805 try self.validateConflictRoot(root.hash);
806 conflict_mod.deinitConflictEntries(self.allocator, owned);
807 owned_live = false;
808 return root;
809 }
810 try self.conflict_roots.ensureUnusedCapacity(self.allocator, 1);
811 try self.lookup.conflict_roots.ensureUnusedCapacity(self.allocator, 1);
812 try self.appendConflictRootRecord(owned);
813 try self.flushSync();
814 self.lookup.conflict_roots.putAssumeCapacity(root.hash, self.conflict_roots.items.len);
815 self.conflict_roots.appendAssumeCapacity(.{
816 .root = root,
817 .entries = owned,
818 });
819 owned_live = false;
820 return root;
821 }
822
823 pub fn ref(self: *const History, name: []const u8) Error!?version.Ref {
824 try self.ensureUsable();
825 for (self.refs.items) |record| {
826 if (std.mem.eql(u8, record.name, name)) return record.ref;
827 }
828 return null;
829 }
830
831 pub fn refList(self: *const History, allocator: Allocator) Error![]version.Ref {
832 try self.ensureUsable();
833 const refs = try allocator.alloc(version.Ref, self.refs.items.len);
834 var count: usize = 0;
835 errdefer freeRefList(allocator, refs[0..count]);
836 for (self.refs.items, refs) |record, *target| {
837 const name = try allocator.dupe(u8, record.name);
838 target.* = .{
839 .name = name,
840 .target = record.ref.target,
841 };
842 count += 1;
843 }
844 return refs;
845 }
846
847 pub fn conflict(self: *const History, hash: version.Hash) ?version.ConflictArtifact {
848 if (self.findConflict(hash)) |record| return record.artifact;
849 return null;
850 }
851
852 pub fn hasCommit(self: *const History, hash: version.Hash) bool {
853 return self.findCommit(hash) != null;
854 }
855
856 pub fn hasDatabaseRoot(self: *const History, hash: version.Hash) bool {
857 return self.findDatabaseRoot(hash) != null;
858 }
859
860 pub fn hasConflict(self: *const History, hash: version.Hash) bool {
861 return self.findConflict(hash) != null;
862 }
863
864 pub fn hasConflictRoot(self: *const History, hash: version.Hash) Error!bool {
865 self.validateConflictRoot(hash) catch |err| switch (err) {
866 error.ConflictRootNotFound => return false,
867 else => return err,
868 };
869 return true;
870 }
871
872 pub fn validateConflictRoot(self: *const History, hash: version.Hash) Error!void {
873 try self.ensureUsable();
874 if (version.same(hash, version.ConflictRoot.empty().hash)) return;
875 const record = self.findConflictRoot(hash) orelse return error.ConflictRootNotFound;
876 const derived = version.ConflictRoot.init(record.entries);
877 if (!version.same(record.root.hash, hash) or
878 !version.same(derived.hash, hash) or
879 derived.count != record.root.count)
880 {
881 return error.InvalidHistory;
882 }
883 try self.validateConflictEntries(record.entries);
884 }
885
886 pub fn validateConflictEntries(
887 self: *const History,
888 entries: []const version.ConflictEntry,
889 ) error{ InvalidHistory, ConflictArtifactNotFound }!void {
890 var previous: ?version.ConflictEntry = null;
891 for (entries) |entry| {
892 if (previous) |prior| {
893 if (prior.sameSlot(entry) or
894 !version.ConflictEntry.lessThan({}, prior, entry))
895 {
896 return error.InvalidHistory;
897 }
898 }
899 const artifact_record = self.findConflict(entry.hash) orelse {
900 return error.ConflictArtifactNotFound;
901 };
902 if (!entry.eql(artifact_record.artifact.entry())) {
903 return error.InvalidHistory;
904 }
905 previous = entry;
906 }
907 }
908
909 pub fn conflictEntries(self: *const History, allocator: Allocator, root: version.Hash) Error!conflict_mod.ConflictEntries {
910 try self.ensureUsable();
911 if (version.same(root, version.ConflictRoot.empty().hash)) {
912 return .{
913 .allocator = allocator,
914 .root = version.ConflictRoot.empty(),
915 .entries = &.{},
916 };
917 }
918 try self.validateConflictRoot(root);
919 const record = self.findConflictRoot(root) orelse return error.ConflictRootNotFound;
920 const entries = try conflict_mod.cloneConflictEntries(allocator, record.entries);
921 errdefer conflict_mod.deinitConflictEntries(allocator, entries);
922 return .{
923 .allocator = allocator,
924 .root = record.root,
925 .entries = entries,
926 };
927 }
928
929 pub fn conflictArtifacts(self: *const History, allocator: Allocator, root: version.Hash) Error!conflict_mod.ConflictArtifacts {
930 try self.ensureUsable();
931 var entries = try self.conflictEntries(allocator, root);
932 defer entries.deinit();
933 if (entries.entries.len == 0) {
934 return .{
935 .allocator = allocator,
936 .root = entries.root,
937 .artifacts = &.{},
938 };
939 }
940
941 const artifacts = try allocator.alloc(version.ConflictArtifact, entries.entries.len);
942 var count: usize = 0;
943 errdefer {
944 for (artifacts[0..count]) |*artifact| conflict_mod.deinitConflictArtifact(allocator, artifact);
945 allocator.free(artifacts);
946 }
947 for (entries.entries, artifacts) |entry, *target| {
948 const record = self.findConflict(entry.hash) orelse return error.ConflictArtifactNotFound;
949 target.* = try conflict_mod.cloneConflictArtifact(allocator, record.artifact);
950 count += 1;
951 }
952 return .{
953 .allocator = allocator,
954 .root = entries.root,
955 .artifacts = artifacts,
956 };
957 }
958
959 pub fn commitValue(self: *const History, hash: version.Hash) Error!version.Commit {
960 try self.ensureUsable();
961 const record = self.findCommit(hash) orelse return error.CommitNotFound;
962 return record.commit;
963 }
964
965 pub fn databaseRoot(self: *const History, allocator: Allocator, root: version.Hash) Error!version.DatabaseRoot {
966 try self.ensureUsable();
967 const record = self.findDatabaseRoot(root) orelse return error.DatabaseRootNotFound;
968 return try record.root.clone(allocator);
969 }
970
971 pub fn databaseValue(self: *const History, allocator: Allocator, root: version.Hash) Error!version.DatabaseValue {
972 try self.ensureUsable();
973 var database_root = try self.databaseRoot(allocator, root);
974 defer database_root.deinit();
975
976 const relations = try allocator.alloc(version.RelationValue, database_root.entries.len);
977 var relation_count: usize = 0;
978 var relations_owned = true;
979 errdefer {
980 if (relations_owned) {
981 for (relations[0..relation_count]) |*relation| relation.deinit(allocator);
982 if (relations.len != 0) allocator.free(relations);
983 }
984 }
985
986 for (database_root.entries, relations) |entry, *relation| {
987 var relation_root = try self.relationRoot(allocator, entry.hash);
988 errdefer relation_root.deinit();
989 const rows = try self.relationRows(allocator, entry.hash);
990 errdefer version.freeRelationRows(allocator, rows);
991 relation.* = .{
992 .root = relation_root,
993 .rows = rows,
994 };
995 relation_count += 1;
996 }
997
998 var value = try version.databaseValueFromOwnedRelations(allocator, relations, database_root.conflicts);
999 relations_owned = false;
1000 errdefer value.deinit();
1001 if (!version.same(value.root.hash, database_root.hash)) return error.InvalidHistory;
1002 return value;
1003 }
1004
1005 pub fn relationRoot(self: *const History, allocator: Allocator, root: version.Hash) Error!version.RelationRoot {
1006 try self.ensureUsable();
1007 const record = self.findRelationRoot(root) orelse return error.RelationRootNotFound;
1008 var out: version.RelationRoot = undefined;
1009 var table_key: ?version.Hash = null;
1010 var index_keys: []const ?version.Hash = &.{};
1011 var owned_index_keys: ?[]?version.Hash = null;
1012 defer if (owned_index_keys) |owned| allocator.free(owned);
1013 switch (record.storage) {
1014 .materialized => |index| {
1015 const materialized = &self.materialized_relation_roots.items[index];
1016 out = try materialized.root.clone(allocator);
1017 table_key = materialized.table_key;
1018 index_keys = materialized.index_keys;
1019 },
1020 .indexed => |location| {
1021 const decoded = try materialize_mod.readIndexedRelationRoot(self, allocator, record.hash, location);
1022 out = decoded.root;
1023 table_key = decoded.table_key;
1024 index_keys = decoded.index_keys;
1025 owned_index_keys = decoded.index_keys;
1026 },
1027 }
1028 errdefer out.deinit();
1029 const table = try materialize_mod.merkleMapRoot(self, allocator, out.table.summary, out.table.hash, out.table.subtree, table_key);
1030 var table_swapped = out.table;
1031 out.table = table;
1032 table_swapped.deinit();
1033 for (out.indexes, index_keys) |*index, index_key| {
1034 const map = try materialize_mod.merkleMapRoot(self, allocator, index.map.summary, index.map.hash, index.map.subtree, index_key);
1035 var map_swapped = index.map;
1036 index.map = map;
1037 map_swapped.deinit();
1038 }
1039 return out;
1040 }
1041
1042 pub fn relationRows(self: *const History, allocator: Allocator, root: version.Hash) Error![]version.RelationRow {
1043 try self.ensureUsable();
1044 var pages = (try self.relationRowsPages(allocator, root)) orelse return error.RelationRowsNotFound;
1045 defer pages.deinit();
1046 var rows: std.ArrayList(version.RelationRow) = .empty;
1047 errdefer {
1048 for (rows.items) |row_value| allocator.free(row_value.bytes);
1049 rows.deinit(allocator);
1050 }
1051 for (pages.items) |page_digest| {
1052 var chunks = (try self.indexPageChunks(allocator, page_digest)) orelse return error.RelationRowsNotFound;
1053 defer chunks.deinit();
1054 for (chunks.items) |digest| {
1055 const chunk_record = self.findRowChunk(digest) orelse return error.RelationRowsNotFound;
1056 try self.chunkRowsInto(allocator, chunk_record, &rows);
1057 }
1058 }
1059 return try rows.toOwnedSlice(allocator);
1060 }
1061
1062 pub fn chunkRowsInto(self: *const History, allocator: Allocator, record: *const record_mod.RowChunkRecord, rows: *std.ArrayList(version.RelationRow)) Error!void {
1063 try self.ensureUsable();
1064 const payload = try materialize_mod.readPayload(self, allocator, .row_chunk, record.payload);
1065 defer allocator.free(payload);
1066 var reader = record_mod.PayloadReader.init(payload);
1067 const digest = try reader.hash();
1068 if (!version.same(digest, record.digest)) return error.InvalidHistory;
1069 const row_count = try reader.readU32();
1070 try rows.ensureUnusedCapacity(allocator, row_count);
1071 var read_rows: u32 = 0;
1072 while (read_rows < row_count) : (read_rows += 1) {
1073 const rowid = try reader.readI64();
1074 const bytes = try allocator.dupe(u8, try reader.readBytes());
1075 rows.appendAssumeCapacity(.{
1076 .rowid = rowid,
1077 .bytes = bytes,
1078 });
1079 }
1080 try reader.finish();
1081 }
1082
1083 pub fn commitDatabaseRoot(self: *const History, allocator: Allocator, commit_hash: version.Hash) Error!version.DatabaseRoot {
1084 try self.ensureUsable();
1085 const commit_value = try self.commitValue(commit_hash);
1086 return try self.databaseRoot(allocator, commit_value.root);
1087 }
1088
1089 pub fn commitEntries(self: *const History, allocator: Allocator) Error![]branch.CommitEntry {
1090 try self.ensureUsable();
1091 const entries = try allocator.alloc(branch.CommitEntry, self.commits.items.len);
1092 for (self.commits.items, entries) |record, *entry| {
1093 entry.* = branch.commitEntry(record.commit);
1094 }
1095 return entries;
1096 }
1097
1098 pub fn len(self: *const History) usize {
1099 return self.bytes_written;
1100 }
1101
1102 pub fn hasRelationRoot(self: *const History, hash: version.Hash) bool {
1103 return self.lookup.relation_roots.get(hash) != null;
1104 }
1105
1106 pub fn hasRelationRows(self: *const History, root: version.Hash) bool {
1107 return self.lookup.relation_rows.get(root) != null;
1108 }
1109
1110 pub fn hasIndexPage(self: *const History, digest: version.Hash) bool {
1111 return self.lookup.index_pages.get(digest) != null;
1112 }
1113
1114 pub fn hasRowChunk(self: *const History, digest: version.Hash) bool {
1115 return self.findRowChunk(digest) != null;
1116 }
1117
1118 pub fn hasTreeNode(self: *const History, key: version.Hash) bool {
1119 return self.findTreeNode(key) != null;
1120 }
1121
1122 pub fn databaseRootView(self: *const History, hash: version.Hash) ?DatabaseRootView {
1123 const record = self.findDatabaseRoot(hash) orelse return null;
1124 return .{
1125 .conflicts = record.root.conflicts,
1126 .entries = record.root.entries,
1127 };
1128 }
1129
1130 pub fn relationKeysView(self: *const History, allocator: Allocator, hash: version.Hash) Error!?RelationKeysView {
1131 try self.ensureUsable();
1132 const record = self.findRelationRoot(hash) orelse return null;
1133 switch (record.storage) {
1134 .materialized => |index| {
1135 const materialized = &self.materialized_relation_roots.items[index];
1136 return .{
1137 .allocator = allocator,
1138 .table_key = materialized.table_key,
1139 .index_keys = materialized.index_keys,
1140 };
1141 },
1142 .indexed => |location| {
1143 var decoded = try materialize_mod.readIndexedRelationRoot(self, allocator, record.hash, location);
1144 defer decoded.root.deinit();
1145 return .{
1146 .allocator = allocator,
1147 .owned = decoded.index_keys,
1148 .table_key = decoded.table_key,
1149 .index_keys = decoded.index_keys,
1150 };
1151 },
1152 }
1153 }
1154
1155 pub fn relationRowsPages(self: *const History, allocator: Allocator, root: version.Hash) Error!?record_mod.HashListView {
1156 try self.ensureUsable();
1157 const index = self.lookup.relation_rows.get(root) orelse return null;
1158 const record = &self.relation_rows.items[index];
1159 return try materialize_mod.readHashListRecord(self, allocator, .relation_rows, record.root, record.storage);
1160 }
1161
1162 pub fn indexPageChunks(self: *const History, allocator: Allocator, digest: version.Hash) Error!?record_mod.HashListView {
1163 try self.ensureUsable();
1164 const index = self.lookup.index_pages.get(digest) orelse return null;
1165 const record = &self.index_pages.items[index];
1166 return try materialize_mod.readHashListRecord(self, allocator, .chunk_index_page, record.digest, record.storage);
1167 }
1168
1169 pub fn treeNodeChildren(self: *const History, allocator: Allocator, key: version.Hash) Error![]version.Hash {
1170 try self.ensureUsable();
1171 const index = self.lookup.tree_nodes.get(key) orelse return error.InvalidHistory;
1172 const node = try materialize_mod.readTreeNodeRecord(self, allocator, &self.tree_nodes.items[index]);
1173 allocator.free(node.node.lower);
1174 if (node.node.upper) |upper| allocator.free(upper);
1175 return node.children;
1176 }
1177
1178 pub fn appendPackRecordPayload(self: *const History, allocator: Allocator, target: *std.ArrayList(u8), kind: record_mod.PackRecordKind, hash: version.Hash) Error!void {
1179 try self.ensureUsable();
1180 switch (kind) {
1181 .row_chunk => try pack_mod.appendRowChunkPackPayload(self, allocator, target, hash),
1182 .chunk_index_page => try pack_mod.appendChunkIndexPagePackPayload(self, allocator, target, hash),
1183 .tree_node => try pack_mod.appendTreeNodePackPayload(self, allocator, target, hash),
1184 .relation_rows => try pack_mod.appendRelationRowsPackPayload(self, allocator, target, hash),
1185 .relation_root => try pack_mod.appendRelationRootPackPayload(self, allocator, target, hash),
1186 .database_root => try pack_mod.appendDatabaseRootPackPayload(self, allocator, target, hash),
1187 .conflict => try pack_mod.appendConflictPackPayload(self, allocator, target, hash),
1188 .conflict_root => try pack_mod.appendConflictRootPackPayload(self, allocator, target, hash),
1189 }
1190 }
1191
1192 pub fn packRecordPresent(self: *const History, kind: record_mod.PackRecordKind, payload: []const u8) Error!bool {
1193 try self.ensureUsable();
1194 var reader = record_mod.PayloadReader.init(payload);
1195 switch (kind) {
1196 .row_chunk => return self.findRowChunk(try reader.hash()) != null,
1197 .chunk_index_page => return self.hasIndexPage(try reader.hash()),
1198 .tree_node => {
1199 if (try reader.readU32() != 1) return error.InvalidHistory;
1200 return self.findTreeNode(try reader.hash()) != null;
1201 },
1202 .relation_rows => return self.hasRelationRows(try reader.hash()),
1203 .relation_root => {
1204 var decoded = try materialize_mod.readRelationRootShallow(self.allocator, &reader);
1205 defer decoded.root.deinit();
1206 defer self.allocator.free(decoded.index_keys);
1207 return self.findRelationRoot(decoded.root.hash) != null;
1208 },
1209 .database_root => {
1210 const conflicts = try reader.hash();
1211 const entry_count = try reader.readU32();
1212 const entries = try self.allocator.alloc(version.RelationEntry, entry_count);
1213 defer self.allocator.free(entries);
1214 for (entries) |*entry| {
1215 const name = try reader.readBytes();
1216 const hash = try reader.hash();
1217 entry.* = .{
1218 .name = name,
1219 .hash = hash,
1220 };
1221 }
1222 var root = try version.DatabaseRoot.initSorted(self.allocator, entries, .{ .hash = conflicts });
1223 const hash = root.hash;
1224 root.deinit();
1225 return self.findDatabaseRoot(hash) != null;
1226 },
1227 .conflict => {
1228 const decoded = try conflict_mod.decodeConflictArtifactPayload(&reader);
1229 return self.findConflict(decoded.hash) != null;
1230 },
1231 .conflict_root => {
1232 const entries = try conflict_mod.readConflictEntries(self.allocator, &reader);
1233 defer conflict_mod.deinitConflictEntries(self.allocator, entries);
1234 try reader.finish();
1235 self.validateConflictEntries(entries) catch {
1236 return error.InvalidHistory;
1237 };
1238 const root = version.ConflictRoot.init(entries);
1239 if (self.findConflictRoot(root.hash) == null) return false;
1240 try self.validateConflictRoot(root.hash);
1241 return true;
1242 },
1243 }
1244 }
1245
1246 pub fn importPackRecord(self: *History, kind: record_mod.PackRecordKind, payload: []const u8) Error!bool {
1247 try self.ensureUsable();
1248 return switch (kind) {
1249 .row_chunk => try pack_mod.importRowChunkPayload(self, payload),
1250 .chunk_index_page => try pack_mod.importChunkIndexPagePayload(self, payload),
1251 .tree_node => try pack_mod.importTreeNodePayload(self, payload),
1252 .relation_rows => try pack_mod.importRelationRowsPayload(self, payload),
1253 .relation_root => try pack_mod.importRelationRootPayload(self, payload),
1254 .database_root => try pack_mod.importDatabaseRootPayload(self, payload),
1255 .conflict => try pack_mod.importConflictPayload(self, payload),
1256 .conflict_root => try pack_mod.importConflictRootPayload(self, payload),
1257 };
1258 }
1259
1260 pub fn importPackCommit(self: *History, root: version.Hash, parents: []const version.Hash) Error!bool {
1261 try self.ensureUsable();
1262 const canonical = version.Commit.init(root, parents);
1263 if (self.findCommit(canonical.hash) != null) return false;
1264 if (self.findDatabaseRoot(root) == null) return error.InvalidHistory;
1265 const owned = try self.allocator.dupe(version.Hash, parents);
1266 errdefer self.allocator.free(owned);
1267 try self.commits.ensureUnusedCapacity(self.allocator, 1);
1268 try self.lookup.commits.ensureUnusedCapacity(self.allocator, 1);
1269 try self.appendCommitRecord(canonical.root, owned);
1270 self.lookup.commits.putAssumeCapacity(canonical.hash, self.commits.items.len);
1271 self.commits.appendAssumeCapacity(.{
1272 .parents = owned,
1273 .commit = .{
1274 .root = canonical.root,
1275 .parents = owned,
1276 .hash = canonical.hash,
1277 },
1278 });
1279 return true;
1280 }
1281
1282 fn appendDatabaseRootRecord(self: *History, root: version.DatabaseRoot) Error!void {
1283 var payload: std.ArrayList(u8) = .empty;
1284 defer payload.deinit(self.allocator);
1285 try record_mod.appendDatabaseRootValue(self.allocator, &payload, root);
1286 try self.appendRecord(.database_root, payload.items);
1287 }
1288
1289 fn appendRelationRootRecord(self: *History, root: version.RelationRoot, table_key: ?version.Hash, index_keys: []const ?version.Hash) Error!void {
1290 var payload: std.ArrayList(u8) = .empty;
1291 defer payload.deinit(self.allocator);
1292 try record_mod.appendRelationRootMerkle(self.allocator, &payload, root, table_key, index_keys);
1293 try self.appendRecord(.relation_root, payload.items);
1294 }
1295
1296 fn appendRelationRowsRecord(self: *History, root: version.Hash, pages: []const version.Hash) Error!void {
1297 if (pages.len > std.math.maxInt(u32)) return error.InvalidHistory;
1298 var payload: std.ArrayList(u8) = .empty;
1299 defer payload.deinit(self.allocator);
1300 try record_mod.appendHash(self.allocator, &payload, root);
1301 try record_mod.appendU32(self.allocator, &payload, @intCast(pages.len));
1302 for (pages) |digest| try record_mod.appendHash(self.allocator, &payload, digest);
1303 try self.appendRecord(.relation_rows, payload.items);
1304 }
1305
1306 fn appendChunkIndexPageRecord(self: *History, digest: version.Hash, chunks: []const version.Hash) Error!void {
1307 if (chunks.len > std.math.maxInt(u32)) return error.InvalidHistory;
1308 var payload: std.ArrayList(u8) = .empty;
1309 defer payload.deinit(self.allocator);
1310 try record_mod.appendHash(self.allocator, &payload, digest);
1311 try record_mod.appendU32(self.allocator, &payload, @intCast(chunks.len));
1312 for (chunks) |chunk_digest| try record_mod.appendHash(self.allocator, &payload, chunk_digest);
1313 try self.appendRecord(.chunk_index_page, payload.items);
1314 }
1315
1316 fn appendCommitRecord(self: *History, root: version.Hash, parents: []const version.Hash) Error!void {
1317 if (parents.len > std.math.maxInt(u32)) return error.InvalidHistory;
1318 var payload: std.ArrayList(u8) = .empty;
1319 defer payload.deinit(self.allocator);
1320 try record_mod.appendHash(self.allocator, &payload, root);
1321 try record_mod.appendU32(self.allocator, &payload, @intCast(parents.len));
1322 for (parents) |parent| try record_mod.appendHash(self.allocator, &payload, parent);
1323 try self.appendRecord(.commit, payload.items);
1324 }
1325
1326 fn appendRefRecord(self: *History, name: []const u8, target: version.Hash) Error!void {
1327 var payload: std.ArrayList(u8) = .empty;
1328 defer payload.deinit(self.allocator);
1329 try record_mod.appendHash(self.allocator, &payload, target);
1330 try record_mod.appendBytes(self.allocator, &payload, name);
1331 try self.appendRecord(.ref, payload.items);
1332 }
1333
1334 fn appendRefDeleteRecord(self: *History, name: []const u8) Error!void {
1335 var payload: std.ArrayList(u8) = .empty;
1336 defer payload.deinit(self.allocator);
1337 try record_mod.appendBytes(self.allocator, &payload, name);
1338 try self.appendRecord(.ref_delete, payload.items);
1339 }
1340
1341 fn appendConflictRecord(self: *History, artifact: version.ConflictArtifact) Error!void {
1342 var payload: std.ArrayList(u8) = .empty;
1343 defer payload.deinit(self.allocator);
1344 try conflict_mod.appendConflictArtifactRecordPayload(self.allocator, &payload, artifact);
1345 try self.appendRecord(.conflict, payload.items);
1346 }
1347
1348 fn appendConflictRootRecord(self: *History, entries: []const version.ConflictEntry) Error!void {
1349 var payload: std.ArrayList(u8) = .empty;
1350 defer payload.deinit(self.allocator);
1351 try conflict_mod.appendConflictEntries(self.allocator, &payload, entries);
1352 try self.appendRecord(.conflict_root, payload.items);
1353 }
1354
1355 pub fn appendRecord(self: *History, kind: record_mod.RecordKind, payload: []const u8) Error!void {
1356 try self.ensureMutable();
1357 const record = try encodeRecord(self.allocator, kind, payload);
1358 defer self.allocator.free(record);
1359
1360 const next_bytes_written = std.math.add(
1361 usize,
1362 self.bytes_written,
1363 record.len,
1364 ) catch return error.InvalidHistory;
1365 try self.pending_write.appendSlice(self.allocator, record);
1366 self.bytes_written = next_bytes_written;
1367 self.needs_sync = true;
1368 if (self.write_batch_depth == 0) {
1369 try self.flushPendingWrite();
1370 try self.flushSync();
1371 }
1372 }
1373
1374 fn appendCoordinatorRecord(self: *History, kind: record_mod.RecordKind, payload: []const u8) Error!void {
1375 if (self.write_batch_depth != 0 or
1376 self.pending_write.items.len != 0 or
1377 self.needs_sync)
1378 {
1379 return error.InvalidHistory;
1380 }
1381 const record = try encodeRecord(self.allocator, kind, payload);
1382 defer self.allocator.free(record);
1383 try self.pending_write.ensureUnusedCapacity(self.allocator, record.len);
1384 try self.persistCoordinatorRecord(record);
1385 }
1386
1387 fn appendCoordinatorHashRecord(self: *History, kind: record_mod.RecordKind, id: version.Hash) Error!void {
1388 if (self.write_batch_depth != 0 or
1389 self.pending_write.items.len != 0 or
1390 self.needs_sync)
1391 {
1392 return error.InvalidHistory;
1393 }
1394 var record: [record_mod.record_header_size + version.hash_bytes]u8 = undefined;
1395 std.mem.writeInt(u32, record[0..4], record_mod.magic, .big);
1396 std.mem.writeInt(u32, record[4..8], record_mod.format_version, .big);
1397 std.mem.writeInt(u32, record[8..12], @backingInt(kind), .big);
1398 std.mem.writeInt(u32, record[12..16], version.hash_bytes, .big);
1399 const digest = record_mod.recordHash(@backingInt(kind), id[0..]);
1400 @memcpy(record[16..record_mod.record_header_size], digest[0..]);
1401 @memcpy(record[record_mod.record_header_size..], id[0..]);
1402 try self.pending_write.ensureUnusedCapacity(self.allocator, record.len);
1403 try self.persistCoordinatorRecord(record[0..]);
1404 }
1405
1406 fn persistCoordinatorRecord(self: *History, record: []const u8) Error!void {
1407 if (self.read_only) return error.ReadOnly;
1408 const start = self.bytes_written;
1409 self.pending_write.appendSliceAssumeCapacity(record);
1410 self.bytes_written = std.math.add(usize, start, record.len) catch {
1411 self.pending_write.clearRetainingCapacity();
1412 return error.InvalidHistory;
1413 };
1414 self.needs_sync = true;
1415 self.flushPendingWrite() catch |err| {
1416 self.restorePreparedAppend(start) catch {
1417 self.poison();
1418 return error.RecoveryRequired;
1419 };
1420 return err;
1421 };
1422 self.flushSync() catch |err| {
1423 self.restorePreparedAppend(start) catch {
1424 self.poison();
1425 return error.RecoveryRequired;
1426 };
1427 return err;
1428 };
1429 }
1430
1431 pub fn flushSync(self: *History) Error!void {
1432 try self.ensureUsable();
1433 if (!self.needs_sync) return;
1434 if (self.write_batch_depth != 0) return;
1435 try self.flushPendingWrite();
1436 const file = self.file orelse return error.InvalidHistory;
1437 try file.sync(self.io);
1438 self.write_io.syncs += 1;
1439 self.needs_sync = false;
1440 }
1441
1442 fn refreshRefsSidecar(self: *History) void {
1443 if (self.fast_forward != null or self.recovery_required) return;
1444 const dir = self.sidecar_dir orelse return;
1445 const path = self.sidecar_path orelse return;
1446 if (self.needs_sync or self.write_batch_depth != 0) return;
1447 if (self.pending_write.items.len != 0) return;
1448 if (self.refs.items.len > refs_mod.max_entries) return;
1449 const entries = self.allocator.alloc(refs_mod.Entry, self.refs.items.len) catch return;
1450 defer self.allocator.free(entries);
1451 for (self.refs.items, 0..) |record, index| {
1452 std.debug.assert(record.ref.name.len != 0);
1453 const commit = self.findCommit(record.ref.target) orelse return;
1454 const root = self.findDatabaseRoot(commit.commit.root) orelse return;
1455 entries[index] = .{
1456 .name = record.ref.name,
1457 .head = record.ref.target,
1458 .root = commit.commit.root,
1459 .conflicts = root.root.conflicts,
1460 };
1461 }
1462 refs_mod.store(self.allocator, self.io, dir, path, self.bytes_written, entries) catch return;
1463 }
1464
1465 fn flushPendingWrite(self: *History) Error!void {
1466 if (self.pending_write.items.len == 0) return;
1467 const file = self.file orelse return error.InvalidHistory;
1468 const start = self.bytes_written - self.pending_write.items.len;
1469 errdefer if (file.setLength(self.io, start)) {
1470 self.write_io.resizes += 1;
1471 } else |_| {};
1472 try file.writePositionalAll(self.io, self.pending_write.items, start);
1473 self.write_io.writes += 1;
1474 self.pending_write.clearRetainingCapacity();
1475 }
1476
1477 fn restorePreparedAppend(self: *History, bytes_written: usize) Error!void {
1478 const file = self.file orelse return error.InvalidHistory;
1479 try file.setLength(self.io, bytes_written);
1480 self.write_io.resizes += 1;
1481 self.pending_write.clearRetainingCapacity();
1482 self.bytes_written = bytes_written;
1483 self.needs_sync = false;
1484 try file.sync(self.io);
1485 }
1486
1487 pub fn beginWriteBatch(self: *History) Error!WriteBatch {
1488 try self.ensureMutable();
1489 self.write_batch_depth += 1;
1490 return .{ .history = self };
1491 }
1492
1493 pub fn findCommit(self: *const History, hash: version.Hash) ?*record_mod.CommitRecord {
1494 const index = self.lookup.commits.get(hash) orelse return null;
1495 return &self.commits.items[index];
1496 }
1497
1498 pub fn findDatabaseRoot(self: *const History, hash: version.Hash) ?*record_mod.DatabaseRootRecord {
1499 const index = self.lookup.database_roots.get(hash) orelse return null;
1500 return &self.database_roots.items[index];
1501 }
1502
1503 pub fn findRelationRoot(self: *const History, hash: version.Hash) ?*record_mod.RelationRootRecord {
1504 const index = self.lookup.relation_roots.get(hash) orelse return null;
1505 return &self.relation_roots.items[index];
1506 }
1507
1508 pub fn findRowChunk(self: *const History, digest: version.Hash) ?*record_mod.RowChunkRecord {
1509 const index = self.lookup.row_chunks.get(digest) orelse return null;
1510 return &self.row_chunks.items[index];
1511 }
1512
1513 pub fn findTreeNode(self: *const History, key: version.Hash) ?*record_mod.TreeNodeRecord {
1514 const index = self.lookup.tree_nodes.get(key) orelse return null;
1515 return &self.tree_nodes.items[index];
1516 }
1517
1518 pub fn findRef(self: *const History, name: []const u8) ?*record_mod.RefRecord {
1519 for (self.refs.items) |*record| {
1520 if (std.mem.eql(u8, record.name, name)) return record;
1521 }
1522 return null;
1523 }
1524
1525 pub fn findRefIndex(self: *const History, name: []const u8) ?usize {
1526 for (self.refs.items, 0..) |record, index| {
1527 if (std.mem.eql(u8, record.name, name)) return index;
1528 }
1529 return null;
1530 }
1531
1532 pub fn findConflict(self: *const History, hash: version.Hash) ?*record_mod.ConflictRecord {
1533 const index = self.lookup.conflicts.get(hash) orelse return null;
1534 return &self.conflicts.items[index];
1535 }
1536
1537 pub fn findConflictRoot(self: *const History, hash: version.Hash) ?*record_mod.ConflictRootRecord {
1538 const index = self.lookup.conflict_roots.get(hash) orelse return null;
1539 return &self.conflict_roots.items[index];
1540 }
1541 };
1542
1543 pub const RelationRowsWriter = struct {
1544 history: *History,
1545 root: version.Hash,
1546 boundary: ?i64,
1547 previous: ?i64 = null,
1548 chunker: chunk_mod.Chunker = .{},
1549 chunks: std.ArrayList(version.Hash) = .empty,
1550 spans: std.ArrayList(record_mod.ChunkSpan) = .empty,
1551 finished: bool = false,
1552
1553 pub fn append(self: *RelationRowsWriter, rowid: i64, bytes: []const u8) Error!void {
1554 std.debug.assert(!self.finished);
1555 if (self.boundary) |boundary| {
1556 if (rowid <= boundary) return error.InvalidHistory;
1557 }
1558 if (self.previous) |previous| {
1559 if (rowid <= previous) return error.InvalidHistory;
1560 }
1561 self.previous = rowid;
1562 if (try self.chunker.append(self.history.allocator, rowid, bytes)) try self.emit();
1563 }
1564
1565 pub fn finish(self: *RelationRowsWriter) Error!void {
1566 std.debug.assert(!self.finished);
1567 if (self.chunker.pending() != 0) try self.emit();
1568 try self.history.finishRelationRows(self.root, self.chunks.items, self.spans.items);
1569 self.finished = true;
1570 }
1571
1572 pub fn deinit(self: *RelationRowsWriter) void {
1573 const allocator = self.history.allocator;
1574 self.chunker.deinit(allocator);
1575 self.chunks.deinit(allocator);
1576 self.spans.deinit(allocator);
1577 self.* = undefined;
1578 }
1579
1580 fn emit(self: *RelationRowsWriter) Error!void {
1581 var rows_scratch: [chunk_mod.max_rows]version.RelationRow = undefined;
1582 const rows = self.chunker.view(&rows_scratch);
1583 std.debug.assert(rows.len > 0);
1584 const digest = chunk_mod.digest(rows);
1585 const allocator = self.history.allocator;
1586 try self.chunks.ensureUnusedCapacity(allocator, 1);
1587 try self.spans.ensureUnusedCapacity(allocator, 1);
1588 try self.history.putRowChunk(digest, rows);
1589 self.chunks.appendAssumeCapacity(digest);
1590 self.spans.appendAssumeCapacity(.{ .first = rows[0].rowid, .last = rows[rows.len - 1].rowid });
1591 self.chunker.reset();
1592 }
1593 };
1594
1595 fn encodeRecord(allocator: Allocator, kind: record_mod.RecordKind, payload: []const u8) Error![]u8 {
1596 if (payload.len > std.math.maxInt(u32)) return error.InvalidHistory;
1597 var record: std.ArrayList(u8) = .empty;
1598 errdefer record.deinit(allocator);
1599 try record_mod.appendU32(allocator, &record, record_mod.magic);
1600 try record_mod.appendU32(allocator, &record, record_mod.format_version);
1601 try record_mod.appendU32(allocator, &record, @backingInt(kind));
1602 try record_mod.appendU32(allocator, &record, @intCast(payload.len));
1603 try record_mod.appendHash(
1604 allocator,
1605 &record,
1606 record_mod.recordHash(@backingInt(kind), payload),
1607 );
1608 try record.appendSlice(allocator, payload);
1609 return try record.toOwnedSlice(allocator);
1610 }
1611
1612 pub const HashIndex = std.AutoArrayHashMapUnmanaged(version.Hash, usize);
1613
1614 pub const ShallowRelationRoot = struct {
1615 root: version.RelationRoot,
1616 table_key: ?version.Hash,
1617 index_keys: []?version.Hash,
1618 };
1619
1620 pub const MapRootHeader = struct {
1621 summary: tree.Summary,
1622 hash: version.Hash,
1623 subtree: version.Hash,
1624 root_key: ?version.Hash,
1625 };
1626
1627 pub const WriteBatch = struct {
1628 history: *History,
1629 active: bool = true,
1630
1631 pub fn finish(self: *WriteBatch) Error!void {
1632 if (!self.active) return;
1633 try self.history.ensureUsable();
1634 std.debug.assert(self.history.write_batch_depth > 0);
1635 if (self.history.write_batch_depth > 1) {
1636 self.history.write_batch_depth -= 1;
1637 self.active = false;
1638 return;
1639 }
1640 try self.history.flushPendingWrite();
1641 self.history.write_batch_depth = 0;
1642 errdefer self.history.write_batch_depth = 1;
1643 try self.history.flushSync();
1644 self.active = false;
1645 }
1646
1647 pub fn deinit(self: *WriteBatch) void {
1648 if (!self.active) return;
1649 std.debug.assert(self.history.write_batch_depth > 0);
1650 self.history.write_batch_depth -= 1;
1651 self.active = false;
1652 }
1653 };
1654
1655 pub const RecordLookup = struct {
1656 database_roots: HashIndex = .empty,
1657 relation_roots: HashIndex = .empty,
1658 relation_rows: HashIndex = .empty,
1659 relation_spans: HashIndex = .empty,
1660 row_chunks: HashIndex = .empty,
1661 index_pages: HashIndex = .empty,
1662 tree_nodes: HashIndex = .empty,
1663 commits: HashIndex = .empty,
1664 conflicts: HashIndex = .empty,
1665 conflict_roots: HashIndex = .empty,
1666
1667 fn deinit(self: *RecordLookup, allocator: Allocator) void {
1668 self.database_roots.deinit(allocator);
1669 self.relation_roots.deinit(allocator);
1670 self.relation_rows.deinit(allocator);
1671 self.relation_spans.deinit(allocator);
1672 self.row_chunks.deinit(allocator);
1673 self.index_pages.deinit(allocator);
1674 self.tree_nodes.deinit(allocator);
1675 self.commits.deinit(allocator);
1676 self.conflicts.deinit(allocator);
1677 self.conflict_roots.deinit(allocator);
1678 self.* = undefined;
1679 }
1680 };
1681
1682 pub const NodeStage = struct {
1683 history: *History,
1684 staged: HashIndex = .empty,
1685 batch: std.ArrayList(u8) = .empty,
1686 records: std.ArrayList(version.Hash) = .empty,
1687
1688 fn init(history: *History) NodeStage {
1689 return .{ .history = history };
1690 }
1691
1692 fn deinit(self: *NodeStage) void {
1693 const allocator = self.history.allocator;
1694 self.staged.deinit(allocator);
1695 self.batch.deinit(allocator);
1696 self.records.deinit(allocator);
1697 self.* = undefined;
1698 }
1699
1700 fn mapKey(self: *NodeStage, map: *const version.MapRoot) Error!?version.Hash {
1701 if (map.nodes.len == 0) return null;
1702 const keys = try self.history.allocator.alloc(?version.Hash, map.nodes.len);
1703 defer self.history.allocator.free(keys);
1704 @memset(keys, null);
1705 return try self.nodeKey(map, keys, 0);
1706 }
1707
1708 fn nodeKey(self: *NodeStage, map: *const version.MapRoot, keys: []?version.Hash, index: usize) Error!version.Hash {
1709 if (keys[index]) |existing| return existing;
1710 const allocator = self.history.allocator;
1711 const node = &map.nodes[index];
1712 const children = try allocator.alloc(version.Hash, node.children_len);
1713 defer if (children.len != 0) allocator.free(children);
1714 for (map.childIndexes(node), children) |child_index, *child_key| {
1715 child_key.* = try self.nodeKey(map, keys, child_index);
1716 }
1717 var content: std.ArrayList(u8) = .empty;
1718 defer content.deinit(allocator);
1719 try record_mod.appendTreeNodeContent(allocator, &content, node, children);
1720 const key = record_mod.treeNodeKey(content.items);
1721 keys[index] = key;
1722 if (self.history.findTreeNode(key) == null and !self.staged.contains(key)) try self.stage(key, content.items);
1723 return key;
1724 }
1725
1726 fn stage(self: *NodeStage, key: version.Hash, content: []const u8) Error!void {
1727 const allocator = self.history.allocator;
1728 try self.staged.ensureUnusedCapacity(allocator, 1);
1729 try self.records.ensureUnusedCapacity(allocator, 1);
1730 try self.batch.ensureUnusedCapacity(allocator, version.hash_bytes + content.len);
1731 self.batch.appendSliceAssumeCapacity(key[0..]);
1732 self.batch.appendSliceAssumeCapacity(content);
1733 self.staged.putAssumeCapacity(key, self.records.items.len);
1734 self.records.appendAssumeCapacity(key);
1735 if (self.records.items.len >= record_mod.node_batch_max) try self.flush();
1736 }
1737
1738 fn flush(self: *NodeStage) Error!void {
1739 if (self.records.items.len == 0) return;
1740 const history = self.history;
1741 const allocator = history.allocator;
1742 if (self.records.items.len > std.math.maxInt(u32)) return error.InvalidHistory;
1743 var payload: std.ArrayList(u8) = .empty;
1744 defer payload.deinit(allocator);
1745 try record_mod.appendU32(allocator, &payload, @intCast(self.records.items.len));
1746 try payload.appendSlice(allocator, self.batch.items);
1747 try history.tree_nodes.ensureUnusedCapacity(allocator, self.records.items.len);
1748 try history.lookup.tree_nodes.ensureUnusedCapacity(allocator, self.records.items.len);
1749 const expected = record_mod.recordHash(@backingInt(record_mod.RecordKind.tree_nodes), payload.items);
1750 const payload_offset = history.bytes_written + record_mod.record_header_size;
1751 const payload_len = payload.items.len;
1752 try history.appendRecord(.tree_nodes, payload.items);
1753 for (self.records.items) |key| {
1754 history.lookup.tree_nodes.putAssumeCapacity(key, history.tree_nodes.items.len);
1755 history.tree_nodes.appendAssumeCapacity(.{
1756 .key = key,
1757 .payload = .{
1758 .expected = expected,
1759 .offset = payload_offset,
1760 .len = payload_len,
1761 },
1762 });
1763 }
1764 self.records.clearRetainingCapacity();
1765 self.batch.clearRetainingCapacity();
1766 }
1767 };
1768
1769 pub const IncrementalBase = struct {
1770 root: version.Hash,
1771 edited: []const i64,
1772 };
1773
1774 pub fn relationRowsSorted(rows: []const version.RelationRow) bool {
1775 var index: usize = 1;
1776 while (index < rows.len) : (index += 1) {
1777 if (rows[index - 1].rowid >= rows[index].rowid) return false;
1778 }
1779 return true;
1780 }
1781
1782 /// Returns the first base span that holds an edited rowid, or the last span
1783 /// when no earlier span holds one. A span holds the rowids above the last
1784 /// rowid of the span before it, or above `minInt(i64)` for the first span,
1785 /// through its own last rowid. Every span before the first one whose last
1786 /// rowid reaches the lowest edited rowid ends below every edited rowid, and
1787 /// that span holds the lowest one, so one pass over the edits and one over
1788 /// the spans find it.
1789 fn firstAffectedSpan(spans: []const record_mod.ChunkSpan, edited: []const i64) usize {
1790 std.debug.assert(spans.len != 0);
1791 const last_index = spans.len - 1;
1792 var lowest: ?i64 = null;
1793 for (edited) |rowid| {
1794 if (rowid == std.math.minInt(i64)) continue;
1795 lowest = if (lowest) |current| @min(current, rowid) else rowid;
1796 }
1797 const bound = lowest orelse return last_index;
1798 for (spans[0..last_index], 0..) |span, index| {
1799 if (span.last >= bound) return index;
1800 }
1801 return last_index;
1802 }
1803
1804 pub const MaterializedTreeSource = struct {
1805 children: []version.Hash,
1806 };
1807
1808 pub const MaterializedTreeNode = struct {
1809 node: tree.Node,
1810 children: []version.Hash,
1811
1812 pub fn deinit(self: *MaterializedTreeNode, allocator: Allocator) void {
1813 allocator.free(self.node.lower);
1814 if (self.node.upper) |upper| allocator.free(upper);
1815 allocator.free(self.children);
1816 self.* = undefined;
1817 }
1818 };
1819
1820 pub fn testingRelationRoot(allocator: Allocator) !version.RelationRoot {
1821 const name = try allocator.dupe(u8, "items");
1822 errdefer allocator.free(name);
1823 var schema_descriptor = try testingRelationSchema(allocator);
1824 errdefer schema_descriptor.deinit();
1825 var table = try testingMapRoot(allocator, "history.table");
1826 errdefer table.deinit();
1827 const indexes = try allocator.alloc(version.IndexRoot, 1);
1828 var index_count: usize = 0;
1829 errdefer {
1830 for (indexes[0..index_count]) |*index| index.deinit();
1831 allocator.free(indexes);
1832 }
1833 var map = try testingMapRoot(allocator, "history.index");
1834 errdefer map.deinit();
1835 indexes[0] = .{
1836 .fields = version.emptyHash("history.index.fields"),
1837 .map = map,
1838 .stats = version.emptyHash("history.index.stats"),
1839 .hash = version.emptyHash("history.index.root"),
1840 };
1841 index_count += 1;
1842 return .{
1843 .allocator = allocator,
1844 .format = version.format_version,
1845 .name = name,
1846 .catalog = .{ .version = 7 },
1847 .schema = version.schemaHash(schema_descriptor.columns, schema_descriptor.indexes),
1848 .schema_descriptor = schema_descriptor,
1849 .table = table,
1850 .indexes = indexes,
1851 .stats = .{
1852 .table = table.summary,
1853 .indexes = 1,
1854 .hash = version.emptyHash("history.stats"),
1855 },
1856 .hash = version.emptyHash("history.relation"),
1857 };
1858 }
1859
1860 pub fn testingRelationSchema(allocator: Allocator) !version.RelationSchema {
1861 const fields = [_]usize{0};
1862 const index_columns = [_]row.Column{.{ .collation = .nocase }};
1863 const columns = [_]catalog_mod.ColumnDefinition{.{
1864 .name = "name",
1865 .column = .{ .collation = .nocase },
1866 .default = .{ .text = "missing" },
1867 }};
1868 const indexes = [_]catalog_mod.IndexDefinition{.{
1869 .name = "items_name",
1870 .fields = fields[0..],
1871 .columns = index_columns[0..],
1872 }};
1873 return try version.RelationSchema.init(allocator, columns[0..], indexes[0..]);
1874 }
1875
1876 pub fn testingMapRoot(allocator: Allocator, label: []const u8) !version.MapRoot {
1877 const lower = try allocator.dupe(u8, label);
1878 errdefer allocator.free(lower);
1879 const upper = try allocator.dupe(u8, "upper");
1880 errdefer allocator.free(upper);
1881 const nodes = try allocator.alloc(tree.Node, 1);
1882 errdefer allocator.free(nodes);
1883 const edges = try allocator.alloc(usize, 0);
1884 errdefer allocator.free(edges);
1885 const summary = tree.Summary{
1886 .leaf_pages = 1,
1887 .entries = 1,
1888 .key_bytes = label.len,
1889 .record_bytes = label.len + 1,
1890 .value_bytes = label.len + 2,
1891 };
1892 nodes[0] = .{
1893 .kind = .leaf,
1894 .lower = lower,
1895 .upper = upper,
1896 .depth = 0,
1897 .summary = summary,
1898 .hash = version.emptyHash(label),
1899 .children_start = 0,
1900 .children_len = 0,
1901 };
1902 return .{
1903 .allocator = allocator,
1904 .summary = summary,
1905 .hash = version.emptyHash(label),
1906 .subtree = version.emptyHash(label),
1907 .nodes = nodes,
1908 .edges = edges,
1909 };
1910 }
1911
1912 pub const TestingTreeShape = struct {
1913 branches: usize,
1914 leaves: usize,
1915 step: i64 = 16,
1916 };
1917
1918 pub fn testingTreeBound(allocator: Allocator, rowid: i64) ![]u8 {
1919 const bytes = try allocator.alloc(u8, key_mod.rowid_size);
1920 errdefer allocator.free(bytes);
1921 _ = try key_mod.encodeRowId(bytes, rowid);
1922 return bytes;
1923 }
1924
1925 pub fn testingMerkleTree(allocator: Allocator, shape: TestingTreeShape, changed: ?usize, generation: []const u8) !version.MapRoot {
1926 const leaf_count = shape.branches * shape.leaves;
1927 const node_count = 1 + shape.branches + leaf_count;
1928 const nodes = try allocator.alloc(tree.Node, node_count);
1929 var built: usize = 0;
1930 errdefer {
1931 for (nodes[0..built]) |node| {
1932 allocator.free(node.lower);
1933 if (node.upper) |upper| allocator.free(upper);
1934 }
1935 allocator.free(nodes);
1936 }
1937 const edges = try allocator.alloc(usize, shape.branches + leaf_count);
1938 errdefer allocator.free(edges);
1939 for (edges[0..shape.branches], 0..) |*edge, index| edge.* = 1 + index;
1940 for (edges[shape.branches..], 0..) |*edge, index| edge.* = 1 + shape.branches + index;
1941
1942 var label_buffer: [64]u8 = undefined;
1943 const root_lower = try allocator.dupe(u8, "");
1944 var root_lower_assigned = false;
1945 errdefer if (!root_lower_assigned) allocator.free(root_lower);
1946 nodes[0] = .{
1947 .kind = .branch,
1948 .lower = root_lower,
1949 .upper = null,
1950 .depth = 0,
1951 .summary = .{
1952 .branch_pages = 1 + shape.branches,
1953 .leaf_pages = leaf_count,
1954 .entries = leaf_count * @as(usize, @intCast(shape.step)),
1955 .max_depth = 2,
1956 },
1957 .hash = version.emptyHash(try std.fmt.bufPrint(&label_buffer, "merkle.root.{s}", .{generation})),
1958 .children_start = 0,
1959 .children_len = shape.branches,
1960 };
1961 root_lower_assigned = true;
1962 built = 1;
1963
1964 for (nodes[1 .. 1 + shape.branches], 0..) |*node, index| {
1965 const start = @as(i64, @intCast(index * shape.leaves)) * shape.step;
1966 const end = @as(i64, @intCast((index + 1) * shape.leaves)) * shape.step;
1967 const lower = if (index == 0) try allocator.dupe(u8, "") else try testingTreeBound(allocator, start);
1968 var lower_assigned = false;
1969 errdefer if (!lower_assigned) allocator.free(lower);
1970 const upper = if (index + 1 == shape.branches) null else try testingTreeBound(allocator, end);
1971 var upper_assigned = upper == null;
1972 errdefer if (!upper_assigned) allocator.free(upper.?);
1973 const touched = changed != null and changed.? / shape.leaves == index;
1974 const label = try std.fmt.bufPrint(&label_buffer, "merkle.branch.{d}.{s}", .{ index, if (touched) generation else "base" });
1975 node.* = .{
1976 .kind = .branch,
1977 .lower = lower,
1978 .upper = upper,
1979 .depth = 1,
1980 .summary = .{
1981 .branch_pages = 1,
1982 .leaf_pages = shape.leaves,
1983 .entries = shape.leaves * @as(usize, @intCast(shape.step)),
1984 .max_depth = 2,
1985 },
1986 .hash = version.emptyHash(label),
1987 .children_start = shape.branches + index * shape.leaves,
1988 .children_len = shape.leaves,
1989 };
1990 lower_assigned = true;
1991 upper_assigned = true;
1992 built += 1;
1993 }
1994
1995 for (nodes[1 + shape.branches ..], 0..) |*node, index| {
1996 const start = @as(i64, @intCast(index)) * shape.step;
1997 const end = @as(i64, @intCast(index + 1)) * shape.step;
1998 const lower = if (index == 0) try allocator.dupe(u8, "") else try testingTreeBound(allocator, start);
1999 var lower_assigned = false;
2000 errdefer if (!lower_assigned) allocator.free(lower);
2001 const upper = if (index + 1 == leaf_count) null else try testingTreeBound(allocator, end);
2002 var upper_assigned = upper == null;
2003 errdefer if (!upper_assigned) allocator.free(upper.?);
2004 const touched = changed != null and changed.? == index;
2005 const label = try std.fmt.bufPrint(&label_buffer, "merkle.leaf.{d}.{s}", .{ index, if (touched) generation else "base" });
2006 node.* = .{
2007 .kind = .leaf,
2008 .lower = lower,
2009 .upper = upper,
2010 .depth = 2,
2011 .summary = .{
2012 .leaf_pages = 1,
2013 .entries = @intCast(shape.step),
2014 .max_depth = 2,
2015 },
2016 .hash = version.emptyHash(label),
2017 };
2018 lower_assigned = true;
2019 upper_assigned = true;
2020 built += 1;
2021 }
2022
2023 return .{
2024 .allocator = allocator,
2025 .summary = nodes[0].summary,
2026 .hash = version.emptyHash(try std.fmt.bufPrint(&label_buffer, "merkle.map.{s}", .{generation})),
2027 .subtree = nodes[0].hash,
2028 .nodes = nodes,
2029 .edges = edges,
2030 };
2031 }
2032
2033 pub fn testingMerkleRelationRoot(allocator: Allocator, shape: TestingTreeShape, changed: ?usize, generation: []const u8) !version.RelationRoot {
2034 const name = try allocator.dupe(u8, "metrics");
2035 errdefer allocator.free(name);
2036 var schema_descriptor = try testingRelationSchema(allocator);
2037 errdefer schema_descriptor.deinit();
2038 var table = try testingMerkleTree(allocator, shape, changed, generation);
2039 errdefer table.deinit();
2040 const indexes = try allocator.alloc(version.IndexRoot, 0);
2041 errdefer allocator.free(indexes);
2042 var label_buffer: [64]u8 = undefined;
2043 const label = try std.fmt.bufPrint(&label_buffer, "merkle.relation.{s}", .{generation});
2044 return .{
2045 .allocator = allocator,
2046 .format = version.format_version,
2047 .name = name,
2048 .catalog = .{ .version = 3 },
2049 .schema = version.schemaHash(schema_descriptor.columns, schema_descriptor.indexes),
2050 .schema_descriptor = schema_descriptor,
2051 .table = table,
2052 .indexes = indexes,
2053 .stats = .{
2054 .table = table.summary,
2055 .indexes = 0,
2056 .hash = version.emptyHash("merkle.stats"),
2057 },
2058 .hash = version.emptyHash(label),
2059 };
2060 }
2061
2062 pub fn expectSameTree(expected: *const version.MapRoot, actual: *const version.MapRoot) !void {
2063 try std.testing.expect(version.same(expected.hash, actual.hash));
2064 try std.testing.expect(version.same(expected.subtree, actual.subtree));
2065 try std.testing.expectEqual(expected.summary, actual.summary);
2066 try std.testing.expectEqual(expected.nodes.len, actual.nodes.len);
2067 try std.testing.expectEqualSlices(usize, expected.edges, actual.edges);
2068 for (expected.nodes, actual.nodes) |expected_node, actual_node| {
2069 try std.testing.expectEqual(expected_node.kind, actual_node.kind);
2070 try std.testing.expectEqualSlices(u8, expected_node.lower, actual_node.lower);
2071 if (expected_node.upper) |expected_upper| {
2072 try std.testing.expectEqualSlices(u8, expected_upper, actual_node.upper.?);
2073 } else {
2074 try std.testing.expect(actual_node.upper == null);
2075 }
2076 try std.testing.expectEqual(expected_node.depth, actual_node.depth);
2077 try std.testing.expectEqual(expected_node.summary, actual_node.summary);
2078 try std.testing.expect(version.same(expected_node.hash, actual_node.hash));
2079 try std.testing.expectEqual(expected_node.children_start, actual_node.children_start);
2080 try std.testing.expectEqual(expected_node.children_len, actual_node.children_len);
2081 }
2082 }
2083
2084 pub fn expectSameDiff(expected: *const diff_mod.RelationDiff, actual: *const diff_mod.RelationDiff) !void {
2085 try std.testing.expectEqual(expected.schema_changed, actual.schema_changed);
2086 try std.testing.expectEqual(expected.skipped_ranges, actual.skipped_ranges);
2087 try std.testing.expectEqual(expected.changes.len, actual.changes.len);
2088 for (expected.changes, actual.changes) |expected_change, actual_change| {
2089 try std.testing.expectEqual(expected_change.kind, actual_change.kind);
2090 try std.testing.expectEqual(expected_change.rowid, actual_change.rowid);
2091 if (expected_change.from) |from| {
2092 try std.testing.expectEqualSlices(u8, from, actual_change.from.?);
2093 } else {
2094 try std.testing.expect(actual_change.from == null);
2095 }
2096 if (expected_change.to) |to| {
2097 try std.testing.expectEqualSlices(u8, to, actual_change.to.?);
2098 } else {
2099 try std.testing.expect(actual_change.to == null);
2100 }
2101 }
2102 }
2103
2104 pub fn writeTestingRecord(file: std.Io.File, offset: usize, kind: record_mod.RecordKind, payload: []const u8) !usize {
2105 var record: std.ArrayList(u8) = .empty;
2106 defer record.deinit(std.testing.allocator);
2107 try record_mod.appendU32(std.testing.allocator, &record, record_mod.magic);
2108 try record_mod.appendU32(std.testing.allocator, &record, record_mod.format_version);
2109 try record_mod.appendU32(std.testing.allocator, &record, @backingInt(kind));
2110 try record_mod.appendU32(std.testing.allocator, &record, @intCast(payload.len));
2111 try record_mod.appendHash(std.testing.allocator, &record, record_mod.recordHash(@backingInt(kind), payload));
2112 try record.appendSlice(std.testing.allocator, payload);
2113 try file.writePositionalAll(testing_io, record.items, offset);
2114 return offset + record.items.len;
2115 }
2116
2117 fn expectTruncation(recovery: Recovery, original_length: usize, valid_length: usize) !void {
2118 switch (recovery) {
2119 .clean => return error.TestExpectedEqual,
2120 .truncated => |truncation| {
2121 try std.testing.expectEqual(original_length, truncation.original_length);
2122 try std.testing.expectEqual(valid_length, truncation.valid_length);
2123 },
2124 }
2125 }
2126
2127 test "history merkle relation roots preserve diff behavior across reopen" {
2128 var tmp = std.testing.tmpDir(.{});
2129 defer tmp.cleanup();
2130
2131 var database = try file_mod.Database.openForTesting(std.testing.allocator, tmp.dir, .{
2132 .paths = .{ .database = "history-diff.db", .wal = "history-diff.wal" },
2133 .header = .{
2134 .sequence = 4242,
2135 .salt = .{ .first = 0x1234_5678, .second = 0x9abc_def0 },
2136 },
2137 });
2138 defer database.deinit();
2139 try database.reserve(.{ .wal_frames = 768 });
2140
2141 var catalog = try catalog_mod.Catalog.open(&database, .{});
2142 _ = try catalog.createRelation(std.testing.allocator, .{
2143 .name = "left",
2144 .columns = &.{.{ .name = "value" }},
2145 }, .{ .durability = .buffered });
2146 _ = try catalog.createRelation(std.testing.allocator, .{
2147 .name = "right",
2148 .columns = &.{.{ .name = "value" }},
2149 }, .{ .durability = .buffered });
2150
2151 var left = try catalog.openRelation(std.testing.allocator, "left");
2152 defer left.deinit();
2153 var right = try catalog.openRelation(std.testing.allocator, "right");
2154 defer right.deinit();
2155
2156 var rowid: i64 = 0;
2157 while (rowid < 260) : (rowid += 1) {
2158 var value_buffer: [16]u8 = undefined;
2159 const value = try std.fmt.bufPrint(&value_buffer, "v{d:0>8}", .{rowid});
2160 _ = try left.relation.put(std.testing.allocator, rowid, &.{.{ .text = value }}, .{ .durability = .buffered });
2161 _ = try right.relation.put(std.testing.allocator, rowid, &.{.{ .text = value }}, .{ .durability = .buffered });
2162 }
2163 _ = try right.relation.put(std.testing.allocator, 259, &.{.{ .text = "changed" }}, .{ .durability = .buffered });
2164
2165 const schema = try catalog.schemaState(std.testing.allocator);
2166 var left_root = try version.relationRoot(std.testing.allocator, "left", schema, &left, null);
2167 defer left_root.deinit();
2168 var right_root = try version.relationRoot(std.testing.allocator, "right", schema, &right, null);
2169 defer right_root.deinit();
2170 const left_rows = try version.relationRows(std.testing.allocator, &left);
2171 defer version.freeRelationRows(std.testing.allocator, left_rows);
2172 const right_rows = try version.relationRows(std.testing.allocator, &right);
2173 defer version.freeRelationRows(std.testing.allocator, right_rows);
2174
2175 {
2176 var history = try History.open(std.testing.allocator, tmp.dir, .{ .recovery = .reject });
2177 defer history.deinit();
2178 try history.putRelationRoot(left_root);
2179 try history.putRelationRoot(right_root);
2180 try std.testing.expect(history.tree_nodes.items.len > 2);
2181 try std.testing.expect(history.tree_nodes.items.len < left_root.table.nodes.len + right_root.table.nodes.len);
2182 }
2183
2184 var recovered_history = try History.open(std.testing.allocator, tmp.dir, .{ .recovery = .reject });
2185 defer recovered_history.deinit();
2186 var recovered_left = try recovered_history.relationRoot(std.testing.allocator, left_root.hash);
2187 defer recovered_left.deinit();
2188 var recovered_right = try recovered_history.relationRoot(std.testing.allocator, right_root.hash);
2189 defer recovered_right.deinit();
2190 try expectSameTree(&left_root.table, &recovered_left.table);
2191 try expectSameTree(&right_root.table, &recovered_right.table);
2192
2193 var expected = try diff_mod.relation(std.testing.allocator, .{
2194 .root = &left_root,
2195 .rows = .{ .materialized = left_rows },
2196 }, .{
2197 .root = &right_root,
2198 .rows = .{ .materialized = right_rows },
2199 });
2200 defer expected.deinit();
2201 var actual = try diff_mod.relation(std.testing.allocator, .{
2202 .root = &recovered_left,
2203 .rows = .{ .materialized = left_rows },
2204 }, .{
2205 .root = &recovered_right,
2206 .rows = .{ .materialized = right_rows },
2207 });
2208 defer actual.deinit();
2209
2210 try std.testing.expect(expected.skipped_ranges > 0);
2211 try std.testing.expectEqual(@as(usize, 1), expected.changes.len);
2212 try std.testing.expectEqual(diff_mod.ChangeKind.modified, expected.changes[0].kind);
2213 try expectSameDiff(&expected, &actual);
2214 }
2215
2216 test "history shares tree nodes across relation root versions" {
2217 var tmp = std.testing.tmpDir(.{});
2218 defer tmp.cleanup();
2219
2220 const shape = TestingTreeShape{ .branches = 16, .leaves = 16 };
2221 var base = try testingMerkleRelationRoot(std.testing.allocator, shape, null, "base");
2222 defer base.deinit();
2223 var next = try testingMerkleRelationRoot(std.testing.allocator, shape, 120, "next");
2224 defer next.deinit();
2225
2226 var first_len: usize = 0;
2227 var growth: usize = 0;
2228 {
2229 var history = try History.open(std.testing.allocator, tmp.dir, .{ .recovery = .reject });
2230 defer history.deinit();
2231 try history.putRelationRoot(base);
2232 first_len = history.len();
2233 try history.putRelationRoot(next);
2234 growth = history.len() - first_len;
2235 try std.testing.expectEqual(base.table.nodes.len + 3, history.tree_nodes.items.len);
2236 }
2237 try std.testing.expect(growth > 0);
2238 try std.testing.expect(growth < first_len / 16);
2239
2240 var recovered_history = try History.open(std.testing.allocator, tmp.dir, .{ .recovery = .reject });
2241 defer recovered_history.deinit();
2242 var recovered_base = try recovered_history.relationRoot(std.testing.allocator, base.hash);
2243 defer recovered_base.deinit();
2244 var recovered_next = try recovered_history.relationRoot(std.testing.allocator, next.hash);
2245 defer recovered_next.deinit();
2246 try expectSameTree(&base.table, &recovered_base.table);
2247 try expectSameTree(&next.table, &recovered_next.table);
2248
2249 var expected = try diff_mod.relation(std.testing.allocator, .{
2250 .root = &base,
2251 .rows = .{ .materialized = &.{} },
2252 }, .{
2253 .root = &next,
2254 .rows = .{ .materialized = &.{} },
2255 });
2256 defer expected.deinit();
2257 var actual = try diff_mod.relation(std.testing.allocator, .{
2258 .root = &recovered_base,
2259 .rows = .{ .materialized = &.{} },
2260 }, .{
2261 .root = &recovered_next,
2262 .rows = .{ .materialized = &.{} },
2263 });
2264 defer actual.deinit();
2265 try std.testing.expect(expected.skipped_ranges > 0);
2266 try expectSameDiff(&expected, &actual);
2267 }
2268
2269 test "history verifies tree node batches during open" {
2270 var tmp = std.testing.tmpDir(.{});
2271 defer tmp.cleanup();
2272
2273 const shape = TestingTreeShape{ .branches = 8, .leaves = 8 };
2274 var root = try testingMerkleRelationRoot(std.testing.allocator, shape, null, "lazy");
2275 defer root.deinit();
2276 const commit = version.Commit.init(version.emptyHash("tree-node-prefix"), &.{});
2277 var clean_len: usize = 0;
2278 var record_offset: usize = 0;
2279 var record_len: usize = 0;
2280 {
2281 var history = try History.open(std.testing.allocator, tmp.dir, .{ .recovery = .reject });
2282 defer history.deinit();
2283 try history.putCommit(commit);
2284 clean_len = history.len();
2285 try history.putRelationRoot(root);
2286 const record = history.tree_nodes.items[0];
2287 record_offset = record.payload.offset;
2288 record_len = record.payload.len;
2289 }
2290
2291 {
2292 var file = try tmp.dir.createFile(testing_io, "tiny.sql.history", .{ .read = true, .truncate = false });
2293 defer file.close(testing_io);
2294 const payload = try std.testing.allocator.alloc(u8, record_len);
2295 defer std.testing.allocator.free(payload);
2296 try std.testing.expectEqual(record_len, try file.readPositionalAll(testing_io, payload, record_offset));
2297 var reader = record_mod.PayloadReader.init(payload);
2298 _ = try reader.readU32();
2299 _ = try reader.hash();
2300 _ = try record_mod.nodeKind(try reader.readU8());
2301 _ = try reader.readBytes();
2302 _ = try reader.optionalBytes();
2303 _ = try record_mod.readUsize(&reader);
2304 _ = try record_mod.readSummary(&reader);
2305 const corrupt_offset = record_offset + reader.cursor;
2306 var byte = [_]u8{payload[reader.cursor] ^ 0xff};
2307 try file.writePositionalAll(testing_io, byte[0..], corrupt_offset);
2308 }
2309
2310 const corrupted_len: usize = @intCast((try tmp.dir.statFile(testing_io, "tiny.sql.history", .{})).size);
2311 try std.testing.expectError(error.TruncatedHistory, History.open(std.testing.allocator, tmp.dir, .{ .recovery = .reject }));
2312 var recovered = try History.open(std.testing.allocator, tmp.dir, .{ .recovery = .truncate });
2313 defer recovered.deinit();
2314 try expectTruncation(recovered.recovery, corrupted_len, clean_len);
2315 try std.testing.expect(recovered.hasCommit(commit.hash));
2316 try std.testing.expectError(error.RelationRootNotFound, recovered.relationRoot(std.testing.allocator, root.hash));
2317 }
2318
2319 test "history rejects relation roots referencing missing tree nodes" {
2320 var tmp = std.testing.tmpDir(.{});
2321 defer tmp.cleanup();
2322
2323 var root = try testingRelationRoot(std.testing.allocator);
2324 defer root.deinit();
2325 var payload: std.ArrayList(u8) = .empty;
2326 defer payload.deinit(std.testing.allocator);
2327 const index_keys = [_]?version.Hash{version.emptyHash("missing.index.node")};
2328 try record_mod.appendRelationRootMerkle(std.testing.allocator, &payload, root, version.emptyHash("missing.table.node"), index_keys[0..]);
2329
2330 {
2331 var file = try tmp.dir.createFile(testing_io, "missing.history", .{});
2332 const end = try writeTestingRecord(file, 0, .relation_root, payload.items);
2333 try file.setLength(testing_io, end);
2334 file.close(testing_io);
2335 try std.testing.expectError(error.InvalidHistory, History.open(std.testing.allocator, tmp.dir, .{ .path = "missing.history", .recovery = .reject }));
2336 }
2337
2338 payload.clearRetainingCapacity();
2339 try record_mod.appendRelationRootMerkle(std.testing.allocator, &payload, root, null, index_keys[0..]);
2340 {
2341 var file = try tmp.dir.createFile(testing_io, "missing.index.history", .{});
2342 const end = try writeTestingRecord(file, 0, .relation_root, payload.items);
2343 try file.setLength(testing_io, end);
2344 file.close(testing_io);
2345 try std.testing.expectError(error.InvalidHistory, History.open(std.testing.allocator, tmp.dir, .{ .path = "missing.index.history", .recovery = .reject }));
2346 }
2347
2348 {
2349 var content: std.ArrayList(u8) = .empty;
2350 defer content.deinit(std.testing.allocator);
2351 const bogus_children = [_]version.Hash{version.emptyHash("missing.child")};
2352 try record_mod.appendTreeNodeContent(std.testing.allocator, &content, &root.table.nodes[0], bogus_children[0..]);
2353 var node_payload: std.ArrayList(u8) = .empty;
2354 defer node_payload.deinit(std.testing.allocator);
2355 try record_mod.appendU32(std.testing.allocator, &node_payload, 1);
2356 try record_mod.appendHash(std.testing.allocator, &node_payload, record_mod.treeNodeKey(content.items));
2357 try node_payload.appendSlice(std.testing.allocator, content.items);
2358 var file = try tmp.dir.createFile(testing_io, "missing.children.history", .{});
2359 const end = try writeTestingRecord(file, 0, .tree_nodes, node_payload.items);
2360 try file.setLength(testing_io, end);
2361 file.close(testing_io);
2362 try std.testing.expectError(error.InvalidHistory, History.open(std.testing.allocator, tmp.dir, .{ .path = "missing.children.history", .recovery = .reject }));
2363 }
2364
2365 const commit = version.Commit.init(version.emptyHash("missing.base"), &.{});
2366 var clean_len: usize = 0;
2367 {
2368 var history = try History.open(std.testing.allocator, tmp.dir, .{ .recovery = .reject });
2369 defer history.deinit();
2370 try history.putCommit(commit);
2371 clean_len = history.len();
2372 }
2373 {
2374 var file = try tmp.dir.createFile(testing_io, "tiny.sql.history", .{ .read = true, .truncate = false });
2375 const end = try writeTestingRecord(file, clean_len, .relation_root, payload.items);
2376 try file.setLength(testing_io, end);
2377 file.close(testing_io);
2378 }
2379 const corrupted_len: usize = @intCast((try tmp.dir.statFile(testing_io, "tiny.sql.history", .{})).size);
2380 try std.testing.expectError(error.TruncatedHistory, History.open(std.testing.allocator, tmp.dir, .{ .recovery = .reject }));
2381 var recovered = try History.open(std.testing.allocator, tmp.dir, .{ .recovery = .truncate });
2382 defer recovered.deinit();
2383 try expectTruncation(recovered.recovery, corrupted_len, clean_len);
2384 try std.testing.expectEqual(clean_len, recovered.len());
2385 try std.testing.expect(recovered.hasCommit(commit.hash));
2386 try std.testing.expectError(error.RelationRootNotFound, recovered.relationRoot(std.testing.allocator, root.hash));
2387 }
2388
2389 test "history open without create requires an existing file" {
2390 var tmp = std.testing.tmpDir(.{});
2391 defer tmp.cleanup();
2392
2393 try std.testing.expectError(error.HistoryNotFound, History.open(std.testing.allocator, tmp.dir, .{ .path = "absent.history", .create = false, .recovery = .reject }));
2394 try std.testing.expectError(error.FileNotFound, tmp.dir.readFileAlloc(testing_io, "absent.history", std.testing.allocator, .unlimited));
2395
2396 {
2397 var created = try History.open(std.testing.allocator, tmp.dir, .{ .path = "present.history", .recovery = .reject });
2398 created.deinit();
2399 }
2400 var reopened = try History.open(std.testing.allocator, tmp.dir, .{ .path = "present.history", .create = false, .recovery = .reject });
2401 reopened.deinit();
2402 }
2403
2404 test "history recovers commits refs and conflicts" {
2405 var tmp = std.testing.tmpDir(.{});
2406 defer tmp.cleanup();
2407
2408 const root_commit = version.Commit.init(version.emptyHash("root"), &.{});
2409 var child_parents = [_]version.Hash{root_commit.hash};
2410 const child_commit = version.Commit.init(version.emptyHash("child"), child_parents[0..]);
2411 const artifact = version.ConflictArtifact.init("items", 9, "base", "ours", "theirs");
2412 const ours_root = version.emptyHash("conflict.relation.ours");
2413 const theirs_root = version.emptyHash("conflict.relation.theirs");
2414 const relation_artifact = version.ConflictArtifact.initRelation("schema", null, ours_root, theirs_root);
2415 const expected_root = try version.ConflictRoot.initSorted(std.testing.allocator, &.{ artifact.entry(), relation_artifact.entry() });
2416
2417 {
2418 var history = try History.open(std.testing.allocator, tmp.dir, .{ .recovery = .reject });
2419 defer history.deinit();
2420 try history.putCommit(root_commit);
2421 try history.putCommit(child_commit);
2422 try history.putRef(.{ .name = "main", .target = child_commit.hash });
2423 try history.putConflict(artifact);
2424 try history.putConflict(relation_artifact);
2425 const stored_root = try history.putConflictRoot(&.{ relation_artifact.entry(), artifact.entry() });
2426 try std.testing.expect(version.same(expected_root.hash, stored_root.hash));
2427 try std.testing.expect(history.len() > 0);
2428 }
2429
2430 var recovered = try History.open(std.testing.allocator, tmp.dir, .{ .recovery = .reject });
2431 defer recovered.deinit();
2432 const main = (try recovered.ref("main")).?;
2433 try std.testing.expect(version.same(main.target, child_commit.hash));
2434
2435 const entries = try recovered.commitEntries(std.testing.allocator);
2436 defer std.testing.allocator.free(entries);
2437 try std.testing.expect(try branch.canFastForward(std.testing.allocator, entries, root_commit.hash, child_commit.hash));
2438
2439 const conflict = recovered.conflict(artifact.hash).?;
2440 try std.testing.expectEqual(@as(i64, 9), conflict.rowid);
2441 try std.testing.expectEqualStrings("items", conflict.relation);
2442 try std.testing.expectEqualStrings("ours", conflict.ours.?.row);
2443 const relation_conflict = recovered.conflict(relation_artifact.hash).?;
2444 try std.testing.expectEqual(version.ConflictKind.relation, relation_conflict.kind);
2445 try std.testing.expectEqualStrings("schema", relation_conflict.relation);
2446 try std.testing.expect(version.same(ours_root, relation_conflict.ours.?.relation));
2447 try std.testing.expect(version.same(theirs_root, relation_conflict.theirs.?.relation));
2448
2449 var recovered_root = try recovered.conflictEntries(std.testing.allocator, expected_root.hash);
2450 defer recovered_root.deinit();
2451 try std.testing.expect(version.same(expected_root.hash, recovered_root.root.hash));
2452 try std.testing.expectEqual(@as(usize, 2), recovered_root.entries.len);
2453 try std.testing.expectEqualStrings("items", recovered_root.entries[0].relation);
2454 try std.testing.expectEqualStrings("schema", recovered_root.entries[1].relation);
2455 var empty_root = try recovered.conflictEntries(std.testing.allocator, version.ConflictRoot.empty().hash);
2456 defer empty_root.deinit();
2457 try std.testing.expectEqual(@as(usize, 0), empty_root.entries.len);
2458
2459 var recovered_artifacts = try recovered.conflictArtifacts(std.testing.allocator, expected_root.hash);
2460 defer recovered_artifacts.deinit();
2461 try std.testing.expect(version.same(expected_root.hash, recovered_artifacts.root.hash));
2462 try std.testing.expectEqual(@as(usize, 2), recovered_artifacts.artifacts.len);
2463 try std.testing.expectEqualStrings("items", recovered_artifacts.artifacts[0].relation);
2464 try std.testing.expectEqual(@as(i64, 9), recovered_artifacts.artifacts[0].rowid);
2465 try std.testing.expectEqualStrings("ours", recovered_artifacts.artifacts[0].ours.?.row);
2466 try std.testing.expectEqual(version.ConflictKind.relation, recovered_artifacts.artifacts[1].kind);
2467 try std.testing.expectEqualStrings("schema", recovered_artifacts.artifacts[1].relation);
2468 try std.testing.expect(version.same(ours_root, recovered_artifacts.artifacts[1].ours.?.relation));
2469 var empty_artifacts = try recovered.conflictArtifacts(std.testing.allocator, version.ConflictRoot.empty().hash);
2470 defer empty_artifacts.deinit();
2471 try std.testing.expectEqual(@as(usize, 0), empty_artifacts.artifacts.len);
2472
2473 {
2474 var broken = try History.open(std.testing.allocator, tmp.dir, .{ .path = "broken.history", .recovery = .reject });
2475 defer broken.deinit();
2476 try std.testing.expectError(
2477 error.ConflictArtifactNotFound,
2478 broken.putConflictRoot(&.{artifact.entry()}),
2479 );
2480 }
2481 }
2482
2483 test "history validates exact conflict root closure" {
2484 var tmp = std.testing.tmpDir(.{});
2485 defer tmp.cleanup();
2486
2487 var history = try History.open(std.testing.allocator, tmp.dir, .{
2488 .path = "conflict-closure.history",
2489 .recovery = .reject,
2490 });
2491 defer history.deinit();
2492
2493 try history.validateConflictRoot(version.ConflictRoot.empty().hash);
2494 try std.testing.expectError(
2495 error.ConflictRootNotFound,
2496 history.validateConflictRoot(version.emptyHash("unknown.conflict.root")),
2497 );
2498
2499 const artifact = version.ConflictArtifact.init("items", 7, "base", "ours", "theirs");
2500 try std.testing.expectError(
2501 error.ConflictArtifactNotFound,
2502 history.putConflictRoot(&.{artifact.entry()}),
2503 );
2504
2505 try history.putConflict(artifact);
2506
2507 var wrong_entry = artifact.entry();
2508 wrong_entry.rowid += 1;
2509 try std.testing.expectError(
2510 error.InvalidHistory,
2511 history.putConflictRoot(&.{wrong_entry}),
2512 );
2513
2514 const valid = try history.putConflictRoot(&.{artifact.entry()});
2515 try history.validateConflictRoot(valid.hash);
2516
2517 const record = history.findConflictRoot(valid.hash).?;
2518 record.root.count += 1;
2519 try std.testing.expectError(error.InvalidHistory, history.validateConflictRoot(valid.hash));
2520 try std.testing.expectError(
2521 error.InvalidHistory,
2522 history.conflictEntries(std.testing.allocator, valid.hash),
2523 );
2524 try std.testing.expectError(error.InvalidHistory, history.hasConflictRoot(valid.hash));
2525 record.root.count -= 1;
2526 record.root.hash = version.emptyHash("wrong.conflict.root");
2527 try std.testing.expectError(error.InvalidHistory, history.validateConflictRoot(valid.hash));
2528 }
2529
2530 test "history rejects duplicate conflict slots before persistence" {
2531 var tmp = std.testing.tmpDir(.{});
2532 defer tmp.cleanup();
2533
2534 var history = try History.open(std.testing.allocator, tmp.dir, .{
2535 .path = "duplicate-conflict-root.history",
2536 .recovery = .reject,
2537 });
2538 defer history.deinit();
2539
2540 const first = version.ConflictArtifact.init(
2541 "items",
2542 7,
2543 "base",
2544 "ours",
2545 "theirs",
2546 );
2547 const second = version.ConflictArtifact.init(
2548 "items",
2549 7,
2550 "base",
2551 "other",
2552 "theirs",
2553 );
2554 try history.putConflict(first);
2555 try history.putConflict(second);
2556 const baseline = history.len();
2557
2558 try std.testing.expectError(
2559 error.InvalidHistory,
2560 history.putConflictRoot(&.{ first.entry(), first.entry() }),
2561 );
2562 try std.testing.expectEqual(baseline, history.len());
2563 try std.testing.expectError(
2564 error.InvalidHistory,
2565 history.putConflictRoot(&.{ first.entry(), second.entry() }),
2566 );
2567 try std.testing.expectEqual(baseline, history.len());
2568 }
2569
2570 test "history pack rejects noncanonical conflict roots before append" {
2571 var tmp = std.testing.tmpDir(.{});
2572 defer tmp.cleanup();
2573
2574 var history = try History.open(std.testing.allocator, tmp.dir, .{
2575 .path = "noncanonical-pack.history",
2576 .recovery = .reject,
2577 });
2578 defer history.deinit();
2579
2580 const first = version.ConflictArtifact.init("a", 1, null, "ours", null);
2581 const second = version.ConflictArtifact.init("b", 2, null, "ours", null);
2582 try history.putConflict(first);
2583 try history.putConflict(second);
2584 var payload: std.ArrayList(u8) = .empty;
2585 defer payload.deinit(std.testing.allocator);
2586 try conflict_mod.appendConflictEntries(
2587 std.testing.allocator,
2588 &payload,
2589 &.{ second.entry(), first.entry() },
2590 );
2591 const baseline = history.len();
2592
2593 try std.testing.expectError(
2594 error.InvalidHistory,
2595 history.packRecordPresent(.conflict_root, payload.items),
2596 );
2597 try std.testing.expectError(
2598 error.InvalidHistory,
2599 history.importPackRecord(.conflict_root, payload.items),
2600 );
2601 try std.testing.expectEqual(baseline, history.len());
2602 }
2603
2604 test "history replay truncates noncanonical conflict roots" {
2605 var tmp = std.testing.tmpDir(.{});
2606 defer tmp.cleanup();
2607
2608 var valid_len: usize = 0;
2609 var corrupt_len: usize = 0;
2610 {
2611 var history = try History.open(std.testing.allocator, tmp.dir, .{
2612 .path = "noncanonical-replay.history",
2613 .recovery = .reject,
2614 });
2615 defer history.deinit();
2616
2617 const first = version.ConflictArtifact.init("a", 1, null, "ours", null);
2618 const second = version.ConflictArtifact.init("b", 2, null, "ours", null);
2619 try history.putConflict(first);
2620 try history.putConflict(second);
2621 valid_len = history.len();
2622 var payload: std.ArrayList(u8) = .empty;
2623 defer payload.deinit(std.testing.allocator);
2624 try conflict_mod.appendConflictEntries(
2625 std.testing.allocator,
2626 &payload,
2627 &.{ second.entry(), first.entry() },
2628 );
2629 try history.appendRecord(.conflict_root, payload.items);
2630 corrupt_len = history.len();
2631 }
2632
2633 try std.testing.expectError(
2634 error.TruncatedHistory,
2635 History.open(std.testing.allocator, tmp.dir, .{
2636 .path = "noncanonical-replay.history",
2637 .recovery = .reject,
2638 }),
2639 );
2640 var recovered = try History.open(std.testing.allocator, tmp.dir, .{
2641 .path = "noncanonical-replay.history",
2642 .recovery = .truncate,
2643 });
2644 defer recovered.deinit();
2645 try expectTruncation(recovered.recovery, corrupt_len, valid_len);
2646 try std.testing.expectEqual(valid_len, recovered.len());
2647 }
2648
2649 test "history write batch emits one write and one sync" {
2650 var tmp = std.testing.tmpDir(.{});
2651 defer tmp.cleanup();
2652
2653 const root_commit = version.Commit.init(version.emptyHash("batch.root"), &.{});
2654 var child_parents = [_]version.Hash{root_commit.hash};
2655 const child_commit = version.Commit.init(version.emptyHash("batch.child"), child_parents[0..]);
2656
2657 {
2658 var history = try History.open(std.testing.allocator, tmp.dir, .{ .recovery = .reject });
2659 defer history.deinit();
2660 var batch = try history.beginWriteBatch();
2661 try history.putCommit(root_commit);
2662 try history.putCommit(child_commit);
2663 try history.putRef(.{ .name = "main", .target = child_commit.hash });
2664 try std.testing.expect(history.needs_sync);
2665 try std.testing.expectEqual(@as(usize, 1), history.write_batch_depth);
2666 try std.testing.expectEqual(@as(usize, 0), try history.file.?.length(testing_io));
2667 try batch.finish();
2668 try std.testing.expect(!history.needs_sync);
2669 try std.testing.expectEqual(@as(usize, 0), history.write_batch_depth);
2670 try std.testing.expectEqual(history.bytes_written, try history.file.?.length(testing_io));
2671 try std.testing.expectEqual(@as(usize, 1), history.write_io.writes);
2672 try std.testing.expectEqual(@as(usize, 0), history.write_io.resizes);
2673 try std.testing.expectEqual(@as(usize, 1), history.write_io.syncs);
2674 }
2675
2676 var recovered = try History.open(std.testing.allocator, tmp.dir, .{ .recovery = .reject });
2677 defer recovered.deinit();
2678 try std.testing.expect(version.same(child_commit.hash, (try recovered.ref("main")).?.target));
2679 try std.testing.expect(recovered.hasCommit(root_commit.hash));
2680 try std.testing.expect(recovered.hasCommit(child_commit.hash));
2681 }
2682
2683 test "fast forward coordinator fences history until completion" {
2684 var tmp = std.testing.tmpDir(.{});
2685 defer tmp.cleanup();
2686
2687 const first = version.Commit.init(version.emptyHash("prepared.ref.first"), &.{});
2688 var second_parents = [_]version.Hash{first.hash};
2689 const second = version.Commit.init(
2690 version.emptyHash("prepared.ref.second"),
2691 second_parents[0..],
2692 );
2693 var third_parents = [_]version.Hash{second.hash};
2694 const third = version.Commit.init(
2695 version.emptyHash("prepared.ref.third"),
2696 third_parents[0..],
2697 );
2698
2699 {
2700 var history = try History.open(std.testing.allocator, tmp.dir, .{
2701 .path = "fast-forward.history",
2702 .recovery = .reject,
2703 });
2704 defer history.deinit();
2705 try history.putCommit(first);
2706 try history.putCommit(second);
2707 try history.putRef(.{ .name = "main", .target = first.hash });
2708 var update = try history.beginFastForward("main", first.hash, second.hash);
2709 try std.testing.expectError(error.RecoveryRequired, history.putCommit(third));
2710 try std.testing.expectError(error.RecoveryRequired, history.beginWriteBatch());
2711 try std.testing.expectError(
2712 error.RecoveryRequired,
2713 history.beginRelationRows(version.emptyHash("prepared.rows")),
2714 );
2715 try update.commit();
2716 try std.testing.expect(version.same(second.hash, (try history.ref("main")).?.target));
2717 try std.testing.expectEqual(FastForwardDecision.target, history.fastForwardRecovery().?.decision);
2718 try update.complete();
2719 try std.testing.expect(history.fastForwardRecovery() == null);
2720 try history.putCommit(third);
2721 try std.testing.expect(history.hasCommit(third.hash));
2722 }
2723
2724 var reopened = try History.open(std.testing.allocator, tmp.dir, .{
2725 .path = "fast-forward.history",
2726 .recovery = .reject,
2727 });
2728 defer reopened.deinit();
2729 try std.testing.expect(version.same(second.hash, (try reopened.ref("main")).?.target));
2730 try std.testing.expect(reopened.hasCommit(third.hash));
2731 }
2732
2733 test "fast forward poisons an ambiguous preexisting write batch" {
2734 var tmp = std.testing.tmpDir(.{});
2735 defer tmp.cleanup();
2736
2737 const first = version.Commit.init(version.emptyHash("batch.prepare.first"), &.{});
2738 const second_parents = [_]version.Hash{first.hash};
2739 const second = version.Commit.init(version.emptyHash("batch.prepare.second"), &second_parents);
2740 const third_parents = [_]version.Hash{second.hash};
2741 const third = version.Commit.init(version.emptyHash("batch.prepare.third"), &third_parents);
2742
2743 {
2744 var history = try History.open(std.testing.allocator, tmp.dir, .{
2745 .path = "batch-prepare.history",
2746 .recovery = .reject,
2747 });
2748 defer history.deinit();
2749 try history.putCommit(first);
2750 try history.putCommit(second);
2751 try history.putRef(.{ .name = "main", .target = first.hash });
2752 var batch = try history.beginWriteBatch();
2753 defer batch.deinit();
2754 try history.putCommit(third);
2755 try std.testing.expectError(
2756 error.RecoveryRequired,
2757 history.beginFastForward("main", first.hash, second.hash),
2758 );
2759 try std.testing.expect(history.requiresRecovery());
2760 try std.testing.expectError(error.RecoveryRequired, batch.finish());
2761 try std.testing.expectError(error.RecoveryRequired, history.ref("main"));
2762 }
2763
2764 var reopened = try History.open(std.testing.allocator, tmp.dir, .{
2765 .path = "batch-prepare.history",
2766 .recovery = .reject,
2767 });
2768 defer reopened.deinit();
2769 try std.testing.expect(!reopened.hasCommit(third.hash));
2770 try std.testing.expect(version.same(first.hash, (try reopened.ref("main")).?.target));
2771 }
2772
2773 test "history replays pending fast forward through abort completion" {
2774 var tmp = std.testing.tmpDir(.{});
2775 defer tmp.cleanup();
2776
2777 const first = version.Commit.init(version.emptyHash("fast.forward.abort.first"), &.{});
2778 const parents = [_]version.Hash{first.hash};
2779 const second = version.Commit.init(
2780 version.emptyHash("fast.forward.abort.second"),
2781 &parents,
2782 );
2783 var id: version.Hash = undefined;
2784 {
2785 var history = try History.open(std.testing.allocator, tmp.dir, .{
2786 .path = "fast-forward-abort.history",
2787 .recovery = .reject,
2788 });
2789 defer history.deinit();
2790 try history.putCommit(first);
2791 try history.putCommit(second);
2792 try history.putRef(.{ .name = "main", .target = first.hash });
2793 const update = try history.beginFastForward("main", first.hash, second.hash);
2794 id = update.id;
2795 }
2796
2797 {
2798 var recovered = try History.open(std.testing.allocator, tmp.dir, .{
2799 .path = "fast-forward-abort.history",
2800 .recovery = .reject,
2801 });
2802 defer recovered.deinit();
2803 const active = recovered.fastForwardRecovery().?;
2804 try std.testing.expect(version.same(id, active.id));
2805 try std.testing.expectEqual(FastForwardDecision.pending, active.decision);
2806 try std.testing.expect(version.same(first.hash, (try recovered.ref("main")).?.target));
2807 var update = FastForwardUpdate{ .history = &recovered, .id = id };
2808 try update.abort();
2809 try std.testing.expectEqual(
2810 FastForwardDecision.baseline,
2811 recovered.fastForwardRecovery().?.decision,
2812 );
2813 try update.complete();
2814 }
2815
2816 var complete = try History.open(std.testing.allocator, tmp.dir, .{
2817 .path = "fast-forward-abort.history",
2818 .recovery = .reject,
2819 });
2820 defer complete.deinit();
2821 try std.testing.expect(complete.fastForwardRecovery() == null);
2822 try std.testing.expect(version.same(first.hash, (try complete.ref("main")).?.target));
2823 }
2824
2825 test "history replays committed fast forward until completion" {
2826 var tmp = std.testing.tmpDir(.{});
2827 defer tmp.cleanup();
2828
2829 const first = version.Commit.init(version.emptyHash("fast.forward.commit.first"), &.{});
2830 const parents = [_]version.Hash{first.hash};
2831 const second = version.Commit.init(
2832 version.emptyHash("fast.forward.commit.second"),
2833 &parents,
2834 );
2835 var id: version.Hash = undefined;
2836 {
2837 var history = try History.open(std.testing.allocator, tmp.dir, .{
2838 .path = "fast-forward-commit.history",
2839 .recovery = .reject,
2840 });
2841 defer history.deinit();
2842 try history.putCommit(first);
2843 try history.putCommit(second);
2844 try history.putRef(.{ .name = "main", .target = first.hash });
2845 var update = try history.beginFastForward("main", first.hash, second.hash);
2846 id = update.id;
2847 try update.commit();
2848 }
2849
2850 {
2851 var recovered = try History.open(std.testing.allocator, tmp.dir, .{
2852 .path = "fast-forward-commit.history",
2853 .recovery = .reject,
2854 });
2855 defer recovered.deinit();
2856 const active = recovered.fastForwardRecovery().?;
2857 try std.testing.expectEqual(FastForwardDecision.target, active.decision);
2858 try std.testing.expect(version.same(second.hash, (try recovered.ref("main")).?.target));
2859 var update = FastForwardUpdate{ .history = &recovered, .id = id };
2860 try update.complete();
2861 }
2862
2863 var complete = try History.open(std.testing.allocator, tmp.dir, .{
2864 .path = "fast-forward-commit.history",
2865 .recovery = .reject,
2866 });
2867 defer complete.deinit();
2868 try std.testing.expect(complete.fastForwardRecovery() == null);
2869 try std.testing.expect(version.same(second.hash, (try complete.ref("main")).?.target));
2870 }
2871
2872 test "history truncates a torn fast forward decision to pending prepare" {
2873 var tmp = std.testing.tmpDir(.{});
2874 defer tmp.cleanup();
2875
2876 const first = version.Commit.init(version.emptyHash("fast.forward.torn.first"), &.{});
2877 const parents = [_]version.Hash{first.hash};
2878 const second = version.Commit.init(
2879 version.emptyHash("fast.forward.torn.second"),
2880 &parents,
2881 );
2882 var id: version.Hash = undefined;
2883 var prepare_length: usize = 0;
2884 {
2885 var history = try History.open(std.testing.allocator, tmp.dir, .{
2886 .path = "fast-forward-torn.history",
2887 .recovery = .reject,
2888 });
2889 defer history.deinit();
2890 try history.putCommit(first);
2891 try history.putCommit(second);
2892 try history.putRef(.{ .name = "main", .target = first.hash });
2893 const update = try history.beginFastForward("main", first.hash, second.hash);
2894 id = update.id;
2895 prepare_length = history.bytes_written;
2896 }
2897
2898 var payload: std.ArrayList(u8) = .empty;
2899 defer payload.deinit(std.testing.allocator);
2900 try record_mod.appendHash(std.testing.allocator, &payload, id);
2901 const decision = try encodeRecord(
2902 std.testing.allocator,
2903 .fast_forward_commit,
2904 payload.items,
2905 );
2906 defer std.testing.allocator.free(decision);
2907 var file = try tmp.dir.createFile(testing_io, "fast-forward-torn.history", .{
2908 .read = true,
2909 .truncate = false,
2910 });
2911 try file.writePositionalAll(testing_io, decision[0 .. decision.len - 1], prepare_length);
2912 try file.setLength(testing_io, prepare_length + decision.len - 1);
2913 file.close(testing_io);
2914
2915 var recovered = try History.open(std.testing.allocator, tmp.dir, .{
2916 .path = "fast-forward-torn.history",
2917 .recovery = .truncate,
2918 });
2919 defer recovered.deinit();
2920 try expectTruncation(
2921 recovered.recovery,
2922 prepare_length + decision.len - 1,
2923 prepare_length,
2924 );
2925 try std.testing.expectEqual(
2926 FastForwardDecision.pending,
2927 recovered.fastForwardRecovery().?.decision,
2928 );
2929 try std.testing.expect(version.same(first.hash, (try recovered.ref("main")).?.target));
2930 }
2931
2932 test "history preserves semantically invalid coordinator evidence" {
2933 var tmp = std.testing.tmpDir(.{});
2934 defer tmp.cleanup();
2935
2936 const first = version.Commit.init(version.emptyHash("fast.forward.invalid.first"), &.{});
2937 const parents = [_]version.Hash{first.hash};
2938 const second = version.Commit.init(
2939 version.emptyHash("fast.forward.invalid.second"),
2940 &parents,
2941 );
2942 var offset: usize = 0;
2943 {
2944 var history = try History.open(std.testing.allocator, tmp.dir, .{
2945 .path = "fast-forward-invalid.history",
2946 .recovery = .reject,
2947 });
2948 defer history.deinit();
2949 try history.putCommit(first);
2950 try history.putCommit(second);
2951 try history.putRef(.{ .name = "main", .target = first.hash });
2952 _ = try history.beginFastForward("main", first.hash, second.hash);
2953 offset = history.bytes_written;
2954 }
2955
2956 var payload: std.ArrayList(u8) = .empty;
2957 defer payload.deinit(std.testing.allocator);
2958 try record_mod.appendBytes(std.testing.allocator, &payload, "main");
2959 try record_mod.appendHash(std.testing.allocator, &payload, first.hash);
2960 try record_mod.appendHash(std.testing.allocator, &payload, second.hash);
2961 var file = try tmp.dir.createFile(testing_io, "fast-forward-invalid.history", .{
2962 .read = true,
2963 .truncate = false,
2964 });
2965 const length = try writeTestingRecord(
2966 file,
2967 offset,
2968 .fast_forward_prepare,
2969 payload.items,
2970 );
2971 try file.setLength(testing_io, length);
2972 file.close(testing_io);
2973
2974 try std.testing.expectError(
2975 error.RecoveryRequired,
2976 History.open(std.testing.allocator, tmp.dir, .{
2977 .path = "fast-forward-invalid.history",
2978 .recovery = .truncate,
2979 }),
2980 );
2981 const stat = try tmp.dir.statFile(testing_io, "fast-forward-invalid.history", .{});
2982 try std.testing.expectEqual(@as(u64, @intCast(length)), stat.size);
2983 }
2984
2985 test "history write batch preserves completed records after allocation failure" {
2986 var tmp = std.testing.tmpDir(.{});
2987 defer tmp.cleanup();
2988
2989 const stable = version.Commit.init(version.emptyHash("batch.failure.stable"), &.{});
2990 const rejected = version.Commit.init(version.emptyHash("batch.failure.rejected"), &.{});
2991 var failing = std.testing.FailingAllocator.init(std.testing.allocator, .{});
2992 {
2993 var history = try History.open(failing.allocator(), tmp.dir, .{ .recovery = .reject });
2994 defer history.deinit();
2995 var batch = try history.beginWriteBatch();
2996 defer batch.deinit();
2997
2998 try history.putCommit(stable);
2999 failing.fail_index = failing.alloc_index;
3000 failing.resize_fail_index = failing.resize_index;
3001 try std.testing.expectError(error.OutOfMemory, history.putCommit(rejected));
3002 failing.fail_index = std.math.maxInt(usize);
3003 failing.resize_fail_index = std.math.maxInt(usize);
3004 }
3005
3006 var recovered = try History.open(std.testing.allocator, tmp.dir, .{ .recovery = .reject });
3007 defer recovered.deinit();
3008 try std.testing.expect(recovered.hasCommit(stable.hash));
3009 try std.testing.expect(!recovered.hasCommit(rejected.hash));
3010 }
3011
3012 test "history recovers database roots by commit" {
3013 var tmp = std.testing.tmpDir(.{});
3014 defer tmp.cleanup();
3015
3016 const items_hash = version.emptyHash("items");
3017 const users_hash = version.emptyHash("users");
3018 var database_root = try version.DatabaseRoot.initSorted(std.testing.allocator, &.{
3019 .{ .name = "users", .hash = users_hash },
3020 .{ .name = "items", .hash = items_hash },
3021 }, version.ConflictRoot.empty());
3022 defer database_root.deinit();
3023 const commit = version.Commit.init(database_root.hash, &.{});
3024
3025 {
3026 var history = try History.open(std.testing.allocator, tmp.dir, .{ .recovery = .reject });
3027 defer history.deinit();
3028 try history.putDatabaseRoot(database_root);
3029 try history.putCommit(commit);
3030 _ = try history.createBranch("main", commit.hash);
3031 }
3032
3033 var recovered = try History.open(std.testing.allocator, tmp.dir, .{ .recovery = .reject });
3034 defer recovered.deinit();
3035 var root = try recovered.commitDatabaseRoot(std.testing.allocator, commit.hash);
3036 defer root.deinit();
3037 try std.testing.expect(version.same(database_root.hash, root.hash));
3038 try std.testing.expectEqual(@as(usize, 2), root.entries.len);
3039 try std.testing.expectEqualStrings("items", root.entries[0].name);
3040 try std.testing.expectEqualStrings("users", root.entries[1].name);
3041 try std.testing.expectError(error.DatabaseRootNotFound, recovered.databaseRoot(std.testing.allocator, version.emptyHash("missing")));
3042 }
3043
3044 test "history recovers relation roots by hash" {
3045 var tmp = std.testing.tmpDir(.{});
3046 defer tmp.cleanup();
3047
3048 var root = try testingRelationRoot(std.testing.allocator);
3049 defer root.deinit();
3050 var expected_payload: std.ArrayList(u8) = .empty;
3051 defer expected_payload.deinit(std.testing.allocator);
3052 try record_mod.appendRelationRoot(std.testing.allocator, &expected_payload, root);
3053
3054 {
3055 var history = try History.open(std.testing.allocator, tmp.dir, .{ .recovery = .reject });
3056 defer history.deinit();
3057 try history.putRelationRoot(root);
3058 switch (history.relation_roots.items[0].storage) {
3059 .materialized => {},
3060 .indexed => return error.TestUnexpectedResult,
3061 }
3062 root.catalog.version += 1;
3063 var immediate = try history.relationRoot(std.testing.allocator, root.hash);
3064 defer immediate.deinit();
3065 try std.testing.expectEqual(@as(u64, 7), immediate.catalog.version);
3066 var immediate_payload: std.ArrayList(u8) = .empty;
3067 defer immediate_payload.deinit(std.testing.allocator);
3068 try record_mod.appendRelationRoot(std.testing.allocator, &immediate_payload, immediate);
3069 try std.testing.expectEqualSlices(u8, expected_payload.items, immediate_payload.items);
3070 root.catalog.version -= 1;
3071 }
3072
3073 var recovered_history = try History.open(std.testing.allocator, tmp.dir, .{ .recovery = .reject });
3074 defer recovered_history.deinit();
3075 switch (recovered_history.relation_roots.items[0].storage) {
3076 .indexed => {},
3077 .materialized => return error.TestUnexpectedResult,
3078 }
3079 var recovered = try recovered_history.relationRoot(std.testing.allocator, root.hash);
3080 defer recovered.deinit();
3081 try std.testing.expect(version.same(root.hash, recovered.hash));
3082 try std.testing.expectEqualStrings("items", recovered.name);
3083 try std.testing.expectEqual(@as(u64, 7), recovered.catalog.version);
3084 try std.testing.expect(version.same(root.schema, recovered.schema));
3085 try std.testing.expectEqual(@as(usize, 1), recovered.schema_descriptor.columns.len);
3086 try std.testing.expectEqualStrings("name", recovered.schema_descriptor.columns[0].name);
3087 try std.testing.expectEqual(row.Collation.nocase, recovered.schema_descriptor.columns[0].column.collation);
3088 try std.testing.expectEqualStrings("missing", recovered.schema_descriptor.columns[0].default.text);
3089 try std.testing.expectEqual(@as(usize, 1), recovered.schema_descriptor.indexes.len);
3090 try std.testing.expectEqualStrings("items_name", recovered.schema_descriptor.indexes[0].name);
3091 try std.testing.expectEqual(@as(usize, 1), recovered.schema_descriptor.indexes[0].fields.len);
3092 try std.testing.expectEqual(@as(usize, 0), recovered.schema_descriptor.indexes[0].fields[0]);
3093 try std.testing.expect(version.same(root.table.hash, recovered.table.hash));
3094 try std.testing.expectEqual(@as(usize, 1), recovered.table.summary.entries);
3095 try std.testing.expectEqual(@as(usize, 1), recovered.indexes.len);
3096 try std.testing.expect(version.same(root.indexes[0].hash, recovered.indexes[0].hash));
3097 try std.testing.expect(version.same(root.indexes[0].map.hash, recovered.indexes[0].map.hash));
3098 try std.testing.expect(version.same(root.stats.hash, recovered.stats.hash));
3099 var recovered_payload: std.ArrayList(u8) = .empty;
3100 defer recovered_payload.deinit(std.testing.allocator);
3101 try record_mod.appendRelationRoot(std.testing.allocator, &recovered_payload, recovered);
3102 try std.testing.expectEqualSlices(u8, expected_payload.items, recovered_payload.items);
3103 try std.testing.expectError(error.RelationRootNotFound, recovered_history.relationRoot(std.testing.allocator, version.emptyHash("missing")));
3104 }
3105
3106 test "history relation root duplicates keep the first valid record" {
3107 var tmp = std.testing.tmpDir(.{});
3108 defer tmp.cleanup();
3109
3110 var root = try testingRelationRoot(std.testing.allocator);
3111 defer root.deinit();
3112 var payload: std.ArrayList(u8) = .empty;
3113 defer payload.deinit(std.testing.allocator);
3114 var append_offset: usize = 0;
3115 {
3116 var history = try History.open(std.testing.allocator, tmp.dir, .{ .recovery = .reject });
3117 defer history.deinit();
3118 try history.putRelationRoot(root);
3119 var keys = (try history.relationKeysView(std.testing.allocator, root.hash)).?;
3120 defer keys.deinit();
3121 root.catalog.version += 1;
3122 try record_mod.appendRelationRootMerkle(std.testing.allocator, &payload, root, keys.table_key, keys.index_keys);
3123 root.catalog.version -= 1;
3124 append_offset = history.len();
3125 }
3126
3127 var file = try tmp.dir.createFile(testing_io, "tiny.sql.history", .{ .read = true, .truncate = false });
3128 const valid_end = try writeTestingRecord(file, append_offset, .relation_root, payload.items);
3129 try payload.append(std.testing.allocator, 0xff);
3130 const corrupted_end = try writeTestingRecord(file, valid_end, .relation_root, payload.items);
3131 file.close(testing_io);
3132
3133 try std.testing.expectError(error.TruncatedHistory, History.open(std.testing.allocator, tmp.dir, .{ .recovery = .reject }));
3134 var recovered_history = try History.open(std.testing.allocator, tmp.dir, .{ .recovery = .truncate });
3135 defer recovered_history.deinit();
3136 try expectTruncation(recovered_history.recovery, corrupted_end, valid_end);
3137 var recovered = try recovered_history.relationRoot(std.testing.allocator, root.hash);
3138 defer recovered.deinit();
3139 try std.testing.expectEqual(root.catalog.version, recovered.catalog.version);
3140 }
3141
3142 test "history rechecks indexed relation roots during access" {
3143 var tmp = std.testing.tmpDir(.{});
3144 defer tmp.cleanup();
3145
3146 var root = try testingRelationRoot(std.testing.allocator);
3147 defer root.deinit();
3148 {
3149 var history = try History.open(std.testing.allocator, tmp.dir, .{ .recovery = .reject });
3150 defer history.deinit();
3151 try history.putRelationRoot(root);
3152 }
3153
3154 var recovered = try History.open(std.testing.allocator, tmp.dir, .{ .recovery = .reject });
3155 defer recovered.deinit();
3156 const location = switch (recovered.relation_roots.items[0].storage) {
3157 .indexed => |location| location,
3158 .materialized => return error.TestUnexpectedResult,
3159 };
3160 var file = try tmp.dir.createFile(testing_io, "tiny.sql.history", .{ .read = true, .truncate = false });
3161 var byte: [1]u8 = undefined;
3162 const corrupt_offset = location.offset + location.len - 1;
3163 try std.testing.expectEqual(@as(usize, 1), try file.readPositionalAll(testing_io, byte[0..], corrupt_offset));
3164 byte[0] ^= 0xff;
3165 try file.writePositionalAll(testing_io, byte[0..], corrupt_offset);
3166 file.close(testing_io);
3167
3168 try std.testing.expectError(error.InvalidHistory, recovered.relationRoot(std.testing.allocator, root.hash));
3169 try std.testing.expectError(error.InvalidHistory, recovered.relationKeysView(std.testing.allocator, root.hash));
3170 var pack_payload: std.ArrayList(u8) = .empty;
3171 defer pack_payload.deinit(std.testing.allocator);
3172 try std.testing.expectError(error.InvalidHistory, recovered.appendPackRecordPayload(std.testing.allocator, &pack_payload, .relation_root, root.hash));
3173 }
3174
3175 test "history recovers relation rows by root hash" {
3176 var tmp = std.testing.tmpDir(.{});
3177 defer tmp.cleanup();
3178
3179 const root = version.emptyHash("history.rows.root");
3180 var first_bytes = [_]u8{ 1, 2, 3 };
3181 var second_bytes = [_]u8{ 4, 5 };
3182 const rows = [_]version.RelationRow{
3183 .{ .rowid = -1, .bytes = first_bytes[0..] },
3184 .{ .rowid = 3, .bytes = second_bytes[0..] },
3185 };
3186
3187 {
3188 var history = try History.open(std.testing.allocator, tmp.dir, .{ .recovery = .reject });
3189 defer history.deinit();
3190 try history.putRelationRows(root, rows[0..]);
3191 }
3192
3193 var recovered_history = try History.open(std.testing.allocator, tmp.dir, .{ .recovery = .reject });
3194 defer recovered_history.deinit();
3195 const recovered = try recovered_history.relationRows(std.testing.allocator, root);
3196 defer version.freeRelationRows(std.testing.allocator, recovered);
3197 try std.testing.expectEqual(@as(usize, 2), recovered.len);
3198 try std.testing.expectEqual(@as(i64, -1), recovered[0].rowid);
3199 try std.testing.expectEqualSlices(u8, first_bytes[0..], recovered[0].bytes);
3200 try std.testing.expectEqual(@as(i64, 3), recovered[1].rowid);
3201 try std.testing.expectEqualSlices(u8, second_bytes[0..], recovered[1].bytes);
3202 try std.testing.expectError(error.RelationRowsNotFound, recovered_history.relationRows(std.testing.allocator, version.emptyHash("missing")));
3203 }
3204
3205 test "history reads indexed row topology for incremental access" {
3206 var tmp = std.testing.tmpDir(.{});
3207 defer tmp.cleanup();
3208
3209 const root = version.emptyHash("history.spans.root");
3210 const next_root = version.emptyHash("history.spans.next");
3211 const rows = try testingChunkRows(std.testing.allocator, 300, 41);
3212 defer version.freeRelationRows(std.testing.allocator, rows);
3213
3214 {
3215 var history = try History.open(std.testing.allocator, tmp.dir, .{ .recovery = .reject });
3216 defer history.deinit();
3217 try history.putRelationRows(root, rows);
3218 }
3219
3220 {
3221 var recovered = try History.open(std.testing.allocator, tmp.dir, .{ .recovery = .reject });
3222 defer recovered.deinit();
3223 try std.testing.expectEqual(@as(usize, 1), recovered.relation_spans.items.len);
3224 for (recovered.relation_rows.items) |record| switch (record.storage) {
3225 .indexed => {},
3226 .materialized => return error.TestUnexpectedResult,
3227 };
3228 for (recovered.index_pages.items) |record| switch (record.storage) {
3229 .indexed => {},
3230 .materialized => return error.TestUnexpectedResult,
3231 };
3232 for (recovered.relation_spans.items) |record| switch (record.storage) {
3233 .indexed => {},
3234 .materialized => return error.TestUnexpectedResult,
3235 };
3236
3237 const edited = [_]i64{rows[rows.len - 1].rowid};
3238 const base = IncrementalBase{ .root = root, .edited = edited[0..] };
3239 const need = try recovered.relationRowsNeed(std.testing.allocator, next_root, base);
3240 const plan = switch (need) {
3241 .suffix => |plan| plan,
3242 else => return error.TestUnexpectedResult,
3243 };
3244 try std.testing.expect(plan.reused > 0);
3245 const tail_start: usize = @intCast(plan.boundary);
3246 try recovered.putRelationRowsSuffix(next_root, base, plan, rows[tail_start..]);
3247 const actual = try recovered.relationRows(std.testing.allocator, next_root);
3248 defer version.freeRelationRows(std.testing.allocator, actual);
3249 try expectRelationRowsEqual(rows, actual);
3250 }
3251
3252 var reopened = try History.open(std.testing.allocator, tmp.dir, .{ .recovery = .reject });
3253 defer reopened.deinit();
3254 const actual = try reopened.relationRows(std.testing.allocator, next_root);
3255 defer version.freeRelationRows(std.testing.allocator, actual);
3256 try expectRelationRowsEqual(rows, actual);
3257 }
3258
3259 fn expectRelationRowsEqual(expected: []const version.RelationRow, actual: []const version.RelationRow) !void {
3260 try std.testing.expectEqual(expected.len, actual.len);
3261 for (expected, actual) |expected_row, actual_row| {
3262 try std.testing.expectEqual(expected_row.rowid, actual_row.rowid);
3263 try std.testing.expectEqualSlices(u8, expected_row.bytes, actual_row.bytes);
3264 }
3265 }
3266
3267 test "first affected span matches a scan of every span and edited rowid" {
3268 var prng = std.Random.DefaultPrng.init(0x5a4e_1d17);
3269 const random = prng.random();
3270 var spans_storage: [24]record_mod.ChunkSpan = undefined;
3271 var edited_storage: [12]i64 = undefined;
3272 for (0..4096) |_| {
3273 const spans = spans_storage[0..random.intRangeAtMost(usize, 1, spans_storage.len)];
3274 const edited = edited_storage[0..random.uintAtMost(usize, edited_storage.len)];
3275 const sorted = random.boolean();
3276 var last: i64 = random.intRangeAtMost(i64, -40, 0);
3277 for (spans) |*span| {
3278 last = if (sorted)
3279 last + random.intRangeAtMost(i64, 1, 6)
3280 else
3281 drawSpanRowid(random);
3282 span.* = .{ .first = last, .last = last };
3283 }
3284 for (edited) |*rowid| rowid.* = drawSpanRowid(random);
3285 const expected = scanFirstAffectedSpan(spans, edited);
3286 try std.testing.expectEqual(expected, firstAffectedSpan(spans, edited));
3287 }
3288 }
3289
3290 /// Draws a rowid near the generated spans, or one of the extreme rowids.
3291 fn drawSpanRowid(random: std.Random) i64 {
3292 const extremes = [_]i64{
3293 std.math.minInt(i64),
3294 std.math.minInt(i64) + 1,
3295 0,
3296 std.math.maxInt(i64) - 1,
3297 std.math.maxInt(i64),
3298 };
3299 if (random.uintLessThan(u8, 8) == 0) {
3300 return extremes[random.uintLessThan(usize, extremes.len)];
3301 }
3302 return random.intRangeAtMost(i64, -40, 120);
3303 }
3304
3305 /// Tests each span in order against every edited rowid, the search that
3306 /// `relationRowsNeed` ran before `firstAffectedSpan`.
3307 fn scanFirstAffectedSpan(spans: []const record_mod.ChunkSpan, edited: []const i64) usize {
3308 const last_index = spans.len - 1;
3309 for (spans[0..last_index], 0..) |span, index| {
3310 const lower: i64 = if (index == 0) std.math.minInt(i64) else spans[index - 1].last;
3311 for (edited) |rowid| {
3312 if (rowid > lower and rowid <= span.last) return index;
3313 }
3314 }
3315 return last_index;
3316 }
3317
3318 test "history row topology duplicates keep the first records" {
3319 var tmp = std.testing.tmpDir(.{});
3320 defer tmp.cleanup();
3321
3322 const root = version.emptyHash("history.topology.duplicate");
3323 const rows = try testingChunkRows(std.testing.allocator, 300, 43);
3324 defer version.freeRelationRows(std.testing.allocator, rows);
3325 var original_count: usize = 0;
3326 var original_boundary: i64 = 0;
3327 var original_page_count: usize = 0;
3328 var original_page: version.Hash = undefined;
3329 var original_chunk_count: usize = 0;
3330 var original_chunk: version.Hash = undefined;
3331 var append_offset: usize = 0;
3332 {
3333 var history = try History.open(std.testing.allocator, tmp.dir, .{ .recovery = .reject });
3334 defer history.deinit();
3335 try history.putRelationRows(root, rows);
3336 var spans = (try history.relationSpans(std.testing.allocator, root)).?;
3337 defer spans.deinit();
3338 try std.testing.expect(spans.items.len > 1);
3339 original_count = spans.items.len;
3340 original_boundary = spans.items[spans.items.len - 2].last;
3341 var pages = (try history.relationRowsPages(std.testing.allocator, root)).?;
3342 defer pages.deinit();
3343 try std.testing.expect(pages.items.len > 0);
3344 original_page_count = pages.items.len;
3345 original_page = pages.items[0];
3346 var chunks = (try history.indexPageChunks(std.testing.allocator, original_page)).?;
3347 defer chunks.deinit();
3348 try std.testing.expect(chunks.items.len > 0);
3349 original_chunk_count = chunks.items.len;
3350 original_chunk = chunks.items[0];
3351 append_offset = history.len();
3352 }
3353
3354 var payload: std.ArrayList(u8) = .empty;
3355 defer payload.deinit(std.testing.allocator);
3356 try record_mod.appendHash(std.testing.allocator, &payload, root);
3357 try record_mod.appendU32(std.testing.allocator, &payload, 1);
3358 try record_mod.appendU64(std.testing.allocator, &payload, @as(u64, @bitCast(@as(i64, std.math.minInt(i64)))));
3359 try record_mod.appendU64(std.testing.allocator, &payload, @as(u64, @bitCast(@as(i64, std.math.maxInt(i64)))));
3360 var file = try tmp.dir.createFile(testing_io, "tiny.sql.history", .{ .read = true, .truncate = false });
3361 append_offset = try writeTestingRecord(file, append_offset, .relation_spans, payload.items);
3362 payload.clearRetainingCapacity();
3363 try record_mod.appendHash(std.testing.allocator, &payload, original_page);
3364 try record_mod.appendU32(std.testing.allocator, &payload, 0);
3365 append_offset = try writeTestingRecord(file, append_offset, .chunk_index_page, payload.items);
3366 payload.clearRetainingCapacity();
3367 try record_mod.appendHash(std.testing.allocator, &payload, root);
3368 try record_mod.appendU32(std.testing.allocator, &payload, 0);
3369 _ = try writeTestingRecord(file, append_offset, .relation_rows, payload.items);
3370 file.close(testing_io);
3371
3372 var recovered = try History.open(std.testing.allocator, tmp.dir, .{ .recovery = .reject });
3373 defer recovered.deinit();
3374 var spans = (try recovered.relationSpans(std.testing.allocator, root)).?;
3375 defer spans.deinit();
3376 try std.testing.expectEqual(original_count, spans.items.len);
3377 try std.testing.expectEqual(original_boundary, spans.items[spans.items.len - 2].last);
3378 var pages = (try recovered.relationRowsPages(std.testing.allocator, root)).?;
3379 defer pages.deinit();
3380 try std.testing.expectEqual(original_page_count, pages.items.len);
3381 try std.testing.expect(version.same(original_page, pages.items[0]));
3382 var chunks = (try recovered.indexPageChunks(std.testing.allocator, original_page)).?;
3383 defer chunks.deinit();
3384 try std.testing.expectEqual(original_chunk_count, chunks.items.len);
3385 try std.testing.expect(version.same(original_chunk, chunks.items[0]));
3386 const actual = try recovered.relationRows(std.testing.allocator, root);
3387 defer version.freeRelationRows(std.testing.allocator, actual);
3388 try expectRelationRowsEqual(rows, actual);
3389 }
3390
3391 test "history validates indexed row topology during open" {
3392 var tmp = std.testing.tmpDir(.{});
3393 defer tmp.cleanup();
3394
3395 const commit = version.Commit.init(version.emptyHash("history.topology.prefix"), &.{});
3396 var clean_len: usize = 0;
3397 {
3398 var history = try History.open(std.testing.allocator, tmp.dir, .{ .recovery = .reject });
3399 defer history.deinit();
3400 try history.putCommit(commit);
3401 clean_len = history.len();
3402 }
3403
3404 var payload: std.ArrayList(u8) = .empty;
3405 defer payload.deinit(std.testing.allocator);
3406 try record_mod.appendHash(std.testing.allocator, &payload, version.emptyHash("history.topology.invalid"));
3407 try record_mod.appendU32(std.testing.allocator, &payload, 2);
3408 try record_mod.appendU64(std.testing.allocator, &payload, 1);
3409 try record_mod.appendU64(std.testing.allocator, &payload, 2);
3410 var file = try tmp.dir.createFile(testing_io, "tiny.sql.history", .{ .read = true, .truncate = false });
3411 const invalid_len = try writeTestingRecord(file, clean_len, .relation_spans, payload.items);
3412 file.close(testing_io);
3413
3414 try std.testing.expectError(error.TruncatedHistory, History.open(std.testing.allocator, tmp.dir, .{ .recovery = .reject }));
3415 var recovered = try History.open(std.testing.allocator, tmp.dir, .{ .recovery = .truncate });
3416 defer recovered.deinit();
3417 try expectTruncation(recovered.recovery, invalid_len, clean_len);
3418 try std.testing.expect(recovered.hasCommit(commit.hash));
3419 }
3420
3421 test "history rechecks indexed row topology during access" {
3422 var tmp = std.testing.tmpDir(.{});
3423 defer tmp.cleanup();
3424
3425 const root = version.emptyHash("history.topology.access");
3426 var bytes = [_]u8{ 1, 2, 3 };
3427 const rows = [_]version.RelationRow{.{ .rowid = 1, .bytes = bytes[0..] }};
3428 {
3429 var history = try History.open(std.testing.allocator, tmp.dir, .{ .recovery = .reject });
3430 defer history.deinit();
3431 try history.putRelationRows(root, rows[0..]);
3432 }
3433
3434 var recovered = try History.open(std.testing.allocator, tmp.dir, .{ .recovery = .reject });
3435 defer recovered.deinit();
3436 const location = switch (recovered.relation_rows.items[0].storage) {
3437 .indexed => |location| location,
3438 .materialized => return error.TestUnexpectedResult,
3439 };
3440 var file = try tmp.dir.createFile(testing_io, "tiny.sql.history", .{ .read = true, .truncate = false });
3441 var byte: [1]u8 = undefined;
3442 const corrupt_offset = location.offset + location.len - 1;
3443 try std.testing.expectEqual(@as(usize, 1), try file.readPositionalAll(testing_io, byte[0..], corrupt_offset));
3444 byte[0] ^= 0xff;
3445 try file.writePositionalAll(testing_io, byte[0..], corrupt_offset);
3446 file.close(testing_io);
3447
3448 try std.testing.expectError(error.InvalidHistory, recovered.relationRows(std.testing.allocator, root));
3449 }
3450
3451 test "history reopens logs larger than sixteen mebibytes" {
3452 var tmp = std.testing.tmpDir(.{});
3453 defer tmp.cleanup();
3454
3455 const payload = try std.testing.allocator.alloc(u8, 6 * 1024 * 1024);
3456 defer std.testing.allocator.free(payload);
3457 @memset(payload, 0xa5);
3458
3459 {
3460 var history = try History.open(std.testing.allocator, tmp.dir, .{ .recovery = .reject });
3461 defer history.deinit();
3462 var index: usize = 0;
3463 while (index < 3) : (index += 1) {
3464 var name_buffer: [32]u8 = undefined;
3465 const name = try std.fmt.bufPrint(&name_buffer, "history.large.{d}", .{index});
3466 const rows = [_]version.RelationRow{.{ .rowid = @intCast(index), .bytes = payload }};
3467 try history.putRelationRows(version.emptyHash(name), rows[0..]);
3468 }
3469 try std.testing.expect(history.len() > 16 * 1024 * 1024);
3470 }
3471
3472 var recovered = try History.open(std.testing.allocator, tmp.dir, .{ .recovery = .reject });
3473 defer recovered.deinit();
3474 const rows = try recovered.relationRows(std.testing.allocator, version.emptyHash("history.large.2"));
3475 defer version.freeRelationRows(std.testing.allocator, rows);
3476 try std.testing.expectEqual(@as(usize, 1), rows.len);
3477 try std.testing.expectEqualSlices(u8, payload, rows[0].bytes);
3478 }
3479
3480 test "history creates checkouts and advances branches durably" {
3481 var tmp = std.testing.tmpDir(.{});
3482 defer tmp.cleanup();
3483
3484 const root_commit = version.Commit.init(version.emptyHash("root"), &.{});
3485 var history = try History.open(std.testing.allocator, tmp.dir, .{ .recovery = .reject });
3486 defer history.deinit();
3487
3488 try history.putCommit(root_commit);
3489 const main = try history.createBranch("main", root_commit.hash);
3490 try std.testing.expectEqualStrings("main", main.name);
3491 try std.testing.expect(version.same(root_commit.hash, main.target));
3492 try std.testing.expectError(error.RefExists, history.createBranch("main", root_commit.hash));
3493 try std.testing.expectError(error.CommitNotFound, history.createBranch("missing", version.emptyHash("missing")));
3494
3495 const checkout = try history.checkoutBranch("main");
3496 try std.testing.expectEqualStrings("main", checkout.name);
3497 try std.testing.expect(version.same(root_commit.hash, checkout.head));
3498 try std.testing.expect(version.same(root_commit.root, checkout.working.base));
3499 try std.testing.expect(!checkout.working.dirty());
3500
3501 const next_root = version.emptyHash("next-root");
3502 const next_hash = try history.commitBranch("main", next_root);
3503 const advanced = try history.checkoutBranch("main");
3504 try std.testing.expect(version.same(next_hash, advanced.head));
3505 try std.testing.expect(version.same(next_root, advanced.working.base));
3506 try std.testing.expect(!advanced.working.dirty());
3507
3508 const side_root = version.emptyHash("side-root");
3509 const side_hash = try history.commitBranch("main", side_root);
3510 try history.fastForwardBranch(std.testing.allocator, "main", side_hash);
3511 try std.testing.expect(version.same(side_hash, (try history.ref("main")).?.target));
3512 try std.testing.expectError(error.NonFastForward, history.fastForwardBranch(std.testing.allocator, "main", next_hash));
3513 try std.testing.expectError(error.RefNotFound, history.checkoutBranch("missing"));
3514 }
3515
3516 test "history updates refs only when expected target matches" {
3517 var tmp = std.testing.tmpDir(.{});
3518 defer tmp.cleanup();
3519
3520 const root_commit = version.Commit.init(version.emptyHash("root"), &.{});
3521 var next_parents = [_]version.Hash{root_commit.hash};
3522 const next_commit = version.Commit.init(version.emptyHash("next"), next_parents[0..]);
3523 var other_parents = [_]version.Hash{root_commit.hash};
3524 const other_commit = version.Commit.init(version.emptyHash("other"), other_parents[0..]);
3525
3526 var history = try History.open(std.testing.allocator, tmp.dir, .{ .recovery = .reject });
3527 defer history.deinit();
3528 try history.putCommit(root_commit);
3529 try history.putCommit(next_commit);
3530 try history.putCommit(other_commit);
3531
3532 try history.putRefIfMatches(.{ .name = "main", .target = root_commit.hash }, null);
3533 try std.testing.expect(version.same(root_commit.hash, (try history.ref("main")).?.target));
3534 try history.putRefIfMatches(.{ .name = "main", .target = next_commit.hash }, root_commit.hash);
3535 try std.testing.expect(version.same(next_commit.hash, (try history.ref("main")).?.target));
3536 try std.testing.expectError(error.RefChanged, history.putRefIfMatches(.{ .name = "main", .target = other_commit.hash }, root_commit.hash));
3537 try std.testing.expect(version.same(next_commit.hash, (try history.ref("main")).?.target));
3538 try std.testing.expectError(error.RefChanged, history.putRefIfMatches(.{ .name = "side", .target = other_commit.hash }, root_commit.hash));
3539 try std.testing.expectError(error.CommitNotFound, history.putRefIfMatches(.{ .name = "missing", .target = version.emptyHash("missing") }, null));
3540 }
3541
3542 test "history recovers branch commits and merge commits" {
3543 var tmp = std.testing.tmpDir(.{});
3544 defer tmp.cleanup();
3545
3546 const root_commit = version.Commit.init(version.emptyHash("root"), &.{});
3547 const main_root = version.emptyHash("main-root");
3548 const side_root = version.emptyHash("side-root");
3549 const merge_root = version.emptyHash("merge-root");
3550 var main_hash: version.Hash = undefined;
3551 var side_hash: version.Hash = undefined;
3552 var merge_hash: version.Hash = undefined;
3553 {
3554 var history = try History.open(std.testing.allocator, tmp.dir, .{ .recovery = .reject });
3555 defer history.deinit();
3556 try history.putCommit(root_commit);
3557 _ = try history.createBranch("main", root_commit.hash);
3558 main_hash = try history.commitBranch("main", main_root);
3559
3560 var side_parents = [_]version.Hash{root_commit.hash};
3561 const side_commit = version.Commit.init(side_root, side_parents[0..]);
3562 side_hash = side_commit.hash;
3563 try history.putCommit(side_commit);
3564 _ = try history.createBranch("side", side_commit.hash);
3565 merge_hash = try history.mergeCommitBranch("main", merge_root, side_commit.hash);
3566 }
3567
3568 var recovered = try History.open(std.testing.allocator, tmp.dir, .{ .recovery = .reject });
3569 defer recovered.deinit();
3570 try std.testing.expect(version.same(merge_hash, (try recovered.ref("main")).?.target));
3571 try std.testing.expect(version.same(side_hash, (try recovered.ref("side")).?.target));
3572 try std.testing.expect(version.same(side_root, (try recovered.checkoutBranch("side")).working.base));
3573
3574 const entries = try recovered.commitEntries(std.testing.allocator);
3575 defer std.testing.allocator.free(entries);
3576 const base = (try branch.mergeBase(std.testing.allocator, entries, merge_hash, (try recovered.ref("side")).?.target)).?;
3577 try std.testing.expect(try branch.canFastForward(std.testing.allocator, entries, main_hash, merge_hash));
3578 try std.testing.expect(try branch.canFastForward(std.testing.allocator, entries, side_hash, merge_hash));
3579 try std.testing.expect(version.same(side_hash, base));
3580 }
3581
3582 test "history refs recover the latest target" {
3583 var tmp = std.testing.tmpDir(.{});
3584 defer tmp.cleanup();
3585
3586 const first = version.Commit.init(version.emptyHash("first"), &.{});
3587 var second_parents = [_]version.Hash{first.hash};
3588 const second = version.Commit.init(version.emptyHash("second"), second_parents[0..]);
3589
3590 {
3591 var history = try History.open(std.testing.allocator, tmp.dir, .{ .recovery = .reject });
3592 defer history.deinit();
3593 try history.putCommit(first);
3594 try history.putCommit(second);
3595 try history.putRef(.{ .name = "main", .target = first.hash });
3596 try history.putRef(.{ .name = "main", .target = second.hash });
3597 }
3598
3599 var recovered = try History.open(std.testing.allocator, tmp.dir, .{ .recovery = .reject });
3600 defer recovered.deinit();
3601 try std.testing.expect(version.same(second.hash, (try recovered.ref("main")).?.target));
3602 }
3603
3604 test "history recovery truncates refs with missing commits" {
3605 var tmp = std.testing.tmpDir(.{});
3606 defer tmp.cleanup();
3607
3608 const first = version.Commit.init(version.emptyHash("valid-ref"), &.{});
3609 var clean_len: usize = 0;
3610 {
3611 var history = try History.open(std.testing.allocator, tmp.dir, .{ .recovery = .reject });
3612 defer history.deinit();
3613 try history.putCommit(first);
3614 try history.putRef(.{ .name = "main", .target = first.hash });
3615 clean_len = history.len();
3616 try std.testing.expectError(error.CommitNotFound, history.putRef(.{ .name = "broken", .target = version.emptyHash("missing-ref") }));
3617 try history.appendRefRecord("main", version.emptyHash("missing-ref"));
3618 try history.flushSync();
3619 try std.testing.expect(history.len() > clean_len);
3620 }
3621
3622 const corrupted_len: usize = @intCast((try tmp.dir.statFile(testing_io, "tiny.sql.history", .{})).size);
3623 try std.testing.expectError(error.TruncatedHistory, History.open(std.testing.allocator, tmp.dir, .{ .recovery = .reject }));
3624 var recovered = try History.open(std.testing.allocator, tmp.dir, .{ .recovery = .truncate });
3625 defer recovered.deinit();
3626 try expectTruncation(recovered.recovery, corrupted_len, clean_len);
3627 try std.testing.expectEqual(clean_len, recovered.len());
3628 try std.testing.expect(version.same(first.hash, (try recovered.ref("main")).?.target));
3629 }
3630
3631 test "history deletes refs durably" {
3632 var tmp = std.testing.tmpDir(.{});
3633 defer tmp.cleanup();
3634
3635 const first = version.Commit.init(version.emptyHash("ref-delete-first"), &.{});
3636 var second_parents = [_]version.Hash{first.hash};
3637 const second = version.Commit.init(version.emptyHash("ref-delete-second"), second_parents[0..]);
3638
3639 {
3640 var history = try History.open(std.testing.allocator, tmp.dir, .{ .recovery = .reject });
3641 defer history.deinit();
3642 try history.putCommit(first);
3643 try history.putCommit(second);
3644 try history.putRef(.{ .name = "main", .target = first.hash });
3645 try history.putRef(.{ .name = "refs/tags/v1", .target = second.hash });
3646 try history.deleteRef("refs/tags/v1");
3647 try std.testing.expect((try history.ref("refs/tags/v1")) == null);
3648 try std.testing.expect(version.same(first.hash, (try history.ref("main")).?.target));
3649 try std.testing.expectError(error.RefNotFound, history.deleteRef("refs/tags/missing"));
3650 }
3651
3652 var recovered = try History.open(std.testing.allocator, tmp.dir, .{ .recovery = .reject });
3653 defer recovered.deinit();
3654 try std.testing.expect((try recovered.ref("refs/tags/v1")) == null);
3655 try std.testing.expect(version.same(first.hash, (try recovered.ref("main")).?.target));
3656 const refs = try recovered.refList(std.testing.allocator);
3657 defer freeRefList(std.testing.allocator, refs);
3658 try std.testing.expectEqual(@as(usize, 1), refs.len);
3659 try std.testing.expectEqualStrings("main", refs[0].name);
3660 }
3661
3662 pub fn testingChunkRows(allocator: Allocator, count: usize, seed: u64) ![]version.RelationRow {
3663 var prng = std.Random.DefaultPrng.init(seed);
3664 const random = prng.random();
3665 const rows = try allocator.alloc(version.RelationRow, count);
3666 var built: usize = 0;
3667 errdefer version.freeRelationRows(allocator, rows[0..built]);
3668 for (rows, 0..) |*row_value, index| {
3669 const bytes = try allocator.alloc(u8, 24 + random.uintLessThan(usize, 40));
3670 random.bytes(bytes);
3671 row_value.* = .{
3672 .rowid = @intCast(index + 1),
3673 .bytes = bytes,
3674 };
3675 built += 1;
3676 }
3677 return rows;
3678 }
3679
3680 test "history shares index pages across large relation versions" {
3681 var tmp = std.testing.tmpDir(.{});
3682 defer tmp.cleanup();
3683
3684 const rows = try testingChunkRows(std.testing.allocator, 6_000, 53);
3685 defer version.freeRelationRows(std.testing.allocator, rows);
3686
3687 var history = try History.open(std.testing.allocator, tmp.dir, .{ .recovery = .reject });
3688 defer history.deinit();
3689
3690 try history.putRelationRows(version.emptyHash("pages.v1"), rows);
3691 const first_len = history.len();
3692 try std.testing.expect(history.index_pages.items.len >= 2);
3693
3694 const replaced = try std.testing.allocator.dupe(u8, "page sharing replacement row");
3695 defer std.testing.allocator.free(replaced);
3696 const original = rows[3_000].bytes;
3697 rows[3_000].bytes = replaced;
3698 defer rows[3_000].bytes = original;
3699 try history.putRelationRows(version.emptyHash("pages.v2"), rows);
3700 const growth = history.len() - first_len;
3701 try std.testing.expect(growth < first_len / 8);
3702
3703 var recovered = try History.open(std.testing.allocator, tmp.dir, .{ .recovery = .reject });
3704 defer recovered.deinit();
3705 const second = try recovered.relationRows(std.testing.allocator, version.emptyHash("pages.v2"));
3706 defer version.freeRelationRows(std.testing.allocator, second);
3707 try std.testing.expectEqual(rows.len, second.len);
3708 try std.testing.expectEqualSlices(u8, replaced, second[3_000].bytes);
3709 }
3710
3711 test "history shares row chunks across relation versions" {
3712 var tmp = std.testing.tmpDir(.{});
3713 defer tmp.cleanup();
3714
3715 const rows = try testingChunkRows(std.testing.allocator, 300, 41);
3716 defer version.freeRelationRows(std.testing.allocator, rows);
3717
3718 var history = try History.open(std.testing.allocator, tmp.dir, .{ .recovery = .reject });
3719 defer history.deinit();
3720
3721 try history.putRelationRows(version.emptyHash("chunks.v1"), rows);
3722 const first_len = history.len();
3723
3724 const replaced = try std.testing.allocator.dupe(u8, "replacement row payload");
3725 defer std.testing.allocator.free(replaced);
3726 const original = rows[150].bytes;
3727 rows[150].bytes = replaced;
3728 defer rows[150].bytes = original;
3729 try history.putRelationRows(version.emptyHash("chunks.v2"), rows);
3730 const growth = history.len() - first_len;
3731 try std.testing.expect(growth < first_len / 2);
3732
3733 var recovered = try History.open(std.testing.allocator, tmp.dir, .{ .recovery = .reject });
3734 defer recovered.deinit();
3735 const second = try recovered.relationRows(std.testing.allocator, version.emptyHash("chunks.v2"));
3736 defer version.freeRelationRows(std.testing.allocator, second);
3737 try std.testing.expectEqual(rows.len, second.len);
3738 for (rows, second) |expected, actual| {
3739 try std.testing.expectEqual(expected.rowid, actual.rowid);
3740 try std.testing.expectEqualSlices(u8, expected.bytes, actual.bytes);
3741 }
3742 const first = try recovered.relationRows(std.testing.allocator, version.emptyHash("chunks.v1"));
3743 defer version.freeRelationRows(std.testing.allocator, first);
3744 try std.testing.expectEqualSlices(u8, original, first[150].bytes);
3745 }
3746
3747 test "history verifies row chunk payloads during open" {
3748 var tmp = std.testing.tmpDir(.{});
3749 defer tmp.cleanup();
3750
3751 const commit = version.Commit.init(version.emptyHash("row-chunk-prefix"), &.{});
3752 const root = version.emptyHash("row-chunk-root");
3753 var first_bytes = [_]u8{ 'f', 'i', 'r', 's', 't' };
3754 var second_bytes = [_]u8{ 's', 'e', 'c', 'o', 'n', 'd' };
3755 const rows = [_]version.RelationRow{
3756 .{ .rowid = 1, .bytes = first_bytes[0..] },
3757 .{ .rowid = 2, .bytes = second_bytes[0..] },
3758 };
3759 var clean_len: usize = 0;
3760 var corrupt_offset: usize = 0;
3761 {
3762 var history = try History.open(std.testing.allocator, tmp.dir, .{ .recovery = .reject });
3763 defer history.deinit();
3764 try history.putCommit(commit);
3765 clean_len = history.len();
3766 try history.putRelationRows(root, rows[0..]);
3767 const chunk = history.row_chunks.items[0];
3768 corrupt_offset = chunk.payload.offset + chunk.payload.len - 1;
3769 }
3770
3771 {
3772 var chunk_file = try tmp.dir.createFile(testing_io, "tiny.sql.history", .{ .read = true, .truncate = false });
3773 var byte: [1]u8 = undefined;
3774 try std.testing.expectEqual(@as(usize, 1), try chunk_file.readPositionalAll(testing_io, byte[0..], corrupt_offset));
3775 byte[0] ^= 0xff;
3776 try chunk_file.writePositionalAll(testing_io, byte[0..], corrupt_offset);
3777 chunk_file.close(testing_io);
3778 }
3779
3780 const corrupted_len: usize = @intCast((try tmp.dir.statFile(testing_io, "tiny.sql.history", .{})).size);
3781 try std.testing.expectError(error.TruncatedHistory, History.open(std.testing.allocator, tmp.dir, .{ .recovery = .reject }));
3782 var recovered = try History.open(std.testing.allocator, tmp.dir, .{ .recovery = .truncate });
3783 defer recovered.deinit();
3784 try expectTruncation(recovered.recovery, corrupted_len, clean_len);
3785 try std.testing.expect(recovered.hasCommit(commit.hash));
3786 try std.testing.expectError(error.RelationRowsNotFound, recovered.relationRows(std.testing.allocator, root));
3787 }
3788
3789 test "history open rejects an unreadable first record" {
3790 var tmp = std.testing.tmpDir(.{});
3791 defer tmp.cleanup();
3792
3793 var file = try tmp.dir.createFile(testing_io, "tiny.sql.history", .{});
3794 var garbage: [96]u8 = undefined;
3795 @memset(garbage[0..], 0x5a);
3796 try file.writePositionalAll(testing_io, garbage[0..], 0);
3797 file.close(testing_io);
3798
3799 try std.testing.expectError(error.InvalidHistory, History.open(std.testing.allocator, tmp.dir, .{ .recovery = .reject }));
3800 }
3801
3802 test "history open rejects prior format records" {
3803 var tmp = std.testing.tmpDir(.{});
3804 defer tmp.cleanup();
3805
3806 {
3807 var history = try History.open(std.testing.allocator, tmp.dir, .{ .recovery = .reject });
3808 defer history.deinit();
3809 try history.putCommit(version.Commit.init(version.emptyHash("format"), &.{}));
3810 }
3811
3812 var file = try tmp.dir.createFile(testing_io, "tiny.sql.history", .{ .read = true, .truncate = false });
3813 var stale_version: [4]u8 = undefined;
3814 std.mem.writeInt(u32, stale_version[0..], record_mod.format_version - 1, .big);
3815 try file.writePositionalAll(testing_io, stale_version[0..], 4);
3816 file.close(testing_io);
3817
3818 try std.testing.expectError(error.InvalidHistory, History.open(std.testing.allocator, tmp.dir, .{ .recovery = .reject }));
3819 }
3820
3821 test "history recovery truncates corrupt tail" {
3822 var tmp = std.testing.tmpDir(.{});
3823 defer tmp.cleanup();
3824
3825 const commit = version.Commit.init(version.emptyHash("root"), &.{});
3826 var clean_len: usize = 0;
3827 {
3828 var history = try History.open(std.testing.allocator, tmp.dir, .{ .recovery = .reject });
3829 defer history.deinit();
3830 try history.putCommit(commit);
3831 clean_len = history.len();
3832 }
3833
3834 var file = try tmp.dir.createFile(testing_io, "tiny.sql.history", .{ .read = true, .truncate = false });
3835 try file.writePositionalAll(testing_io, "corrupt", clean_len);
3836 const corrupted_len = clean_len + "corrupt".len;
3837 try file.setLength(testing_io, corrupted_len);
3838 file.close(testing_io);
3839
3840 try std.testing.expectError(error.TruncatedHistory, History.open(std.testing.allocator, tmp.dir, .{ .recovery = .reject }));
3841 var recovered = try History.open(std.testing.allocator, tmp.dir, .{ .recovery = .truncate });
3842 defer recovered.deinit();
3843 try expectTruncation(recovered.recovery, corrupted_len, clean_len);
3844 try std.testing.expectEqual(clean_len, recovered.len());
3845 const entries = try recovered.commitEntries(std.testing.allocator);
3846 defer std.testing.allocator.free(entries);
3847 try std.testing.expectEqual(@as(usize, 1), entries.len);
3848 try std.testing.expect(version.same(commit.hash, entries[0].hash));
3849 }
3850
3851 test "history appends after a truncating recovery end the file where they stop" {
3852 var tmp = std.testing.tmpDir(.{});
3853 defer tmp.cleanup();
3854
3855 const first = version.Commit.init(version.emptyHash("first"), &.{});
3856 var parents = [_]version.Hash{first.hash};
3857 const second = version.Commit.init(version.emptyHash("second"), parents[0..]);
3858 var clean_len: usize = 0;
3859 {
3860 var history = try History.open(std.testing.allocator, tmp.dir, .{ .recovery = .reject });
3861 defer history.deinit();
3862 try history.putCommit(first);
3863 clean_len = history.len();
3864 }
3865
3866 const torn: [4096]u8 = @splat(0xa5);
3867 {
3868 var file = try tmp.dir.createFile(testing_io, "tiny.sql.history", .{
3869 .read = true,
3870 .truncate = false,
3871 });
3872 defer file.close(testing_io);
3873 try file.writePositionalAll(testing_io, torn[0..], clean_len);
3874 }
3875
3876 {
3877 var recovered = try History.open(std.testing.allocator, tmp.dir, .{ .recovery = .truncate });
3878 defer recovered.deinit();
3879 try expectTruncation(recovered.recovery, clean_len + torn.len, clean_len);
3880 try recovered.putCommit(second);
3881 try std.testing.expect(recovered.len() < clean_len + torn.len);
3882 try std.testing.expectEqual(recovered.len(), try recovered.file.?.length(testing_io));
3883 try std.testing.expectEqual(@as(usize, 0), recovered.write_io.resizes);
3884 }
3885
3886 var reopened = try History.open(std.testing.allocator, tmp.dir, .{ .recovery = .reject });
3887 defer reopened.deinit();
3888 const entries = try reopened.commitEntries(std.testing.allocator);
3889 defer std.testing.allocator.free(entries);
3890 try std.testing.expectEqual(@as(usize, 2), entries.len);
3891 }