lib/sql/src/history/repair.zig
daab053ee43316e1809a84551d573ddd1e5bf3d2
1 const std = @import("std");
2 const sql = @import("../root.zig");
3 const record = @import("record.zig");
4 const materialize = @import("materialize.zig");
5 const Allocator = std.mem.Allocator;
6 const Hash = sql.version.Hash;
7 const State = sql.lattice.State;
8 const Frame = sql.sync.PackFrame;
9 const Reader = record.PayloadReader;
10
11 pub const Candidate = struct { relation: []const u8, rowid: i64 };
12 pub const Summary = struct { roots_repaired: usize = 0, rows_dropped: usize = 0 };
13
14 const Aggregate = struct {
15 states: []State,
16 dropped: ?[]State = null,
17 count: u64 = 0,
18 bytes: u64 = 0,
19 drop_bytes: u64 = 0,
20 first: ?i64 = null,
21 last: ?i64 = null,
22
23 fn init(allocator: Allocator, count: usize) !Aggregate {
24 const states = try allocator.alloc(State, count);
25 @memset(states, State.empty);
26 return .{ .states = states };
27 }
28
29 fn deinit(self: *Aggregate, allocator: Allocator) void {
30 allocator.free(self.states);
31 if (self.dropped) |states| allocator.free(states);
32 }
33
34 fn add(self: *Aggregate, allocator: Allocator, other: *const Aggregate) !void {
35 if (self.last) |last| if (other.first) |first| {
36 if (first <= last) return error.UnprovablePackRepair;
37 };
38 if (self.first == null) self.first = other.first;
39 if (other.last != null) self.last = other.last;
40 self.count = try std.math.add(u64, self.count, other.count);
41 self.bytes = try std.math.add(u64, self.bytes, other.bytes);
42 for (self.states, other.states) |*state, *part| state.add(part);
43 if (other.dropped) |states| {
44 if (self.dropped != null) return error.UnprovablePackRepair;
45 self.dropped = try allocator.dupe(State, states);
46 self.drop_bytes = other.drop_bytes;
47 }
48 }
49
50 fn removeCandidate(self: *Aggregate) !void {
51 const dropped = self.dropped orelse return error.UnprovablePackRepair;
52 self.count = try std.math.sub(u64, self.count, 1);
53 self.bytes = try std.math.sub(u64, self.bytes, self.drop_bytes);
54 for (self.states, dropped) |*state, *part| state.subtract(part);
55 }
56 };
57
58 const Engine = struct {
59 allocator: Allocator,
60 pack: *sql.sync.Pack,
61 candidate: Candidate,
62 objects: [6]std.AutoHashMapUnmanaged(Hash, usize) = @splat(.empty),
63 chunks: std.AutoHashMapUnmanaged([64]u8, Aggregate) = .empty,
64 pages: std.AutoHashMapUnmanaged([64]u8, Aggregate) = .empty,
65 replacements: std.AutoHashMapUnmanaged(Hash, Hash) = .empty,
66 additions: std.ArrayList(Frame) = .empty,
67 added: std.AutoHashMapUnmanaged(Hash, void) = .empty,
68 summary: Summary = .{},
69
70 fn deinit(self: *Engine) void {
71 for (&self.objects) |*map| map.deinit(self.allocator);
72 for ([_]*std.AutoHashMapUnmanaged([64]u8, Aggregate){ &self.chunks, &self.pages }) |cache| {
73 var values = cache.valueIterator();
74 while (values.next()) |value| value.deinit(self.allocator);
75 cache.deinit(self.allocator);
76 }
77 self.replacements.deinit(self.allocator);
78 self.added.deinit(self.allocator);
79 for (self.additions.items) |frame| self.allocator.free(frame.payload);
80 self.additions.deinit(self.allocator);
81 }
82
83 fn index(self: *Engine) !void {
84 for (self.pack.frames, 0..) |frame, i| {
85 const kind = @backingInt(frame.kind);
86 if (kind > 5) continue;
87 var reader = Reader.init(frame.payload);
88 const hash = if (kind == 5) blk: {
89 if (frame.payload.len < 32) return error.InvalidPack;
90 break :blk frame.payload[frame.payload.len - 32 ..][0..32].*;
91 } else try reader.hash();
92 const slot = try self.objects[kind].getOrPut(self.allocator, hash);
93 if (slot.found_existing) return error.InvalidPack;
94 slot.value_ptr.* = i;
95 }
96 }
97
98 fn payload(self: *const Engine, kind: usize, hash: Hash) ![]const u8 {
99 const i = self.objects[kind].get(hash) orelse return error.InvalidPack;
100 return self.pack.frames[i].payload;
101 }
102
103 fn validateLinks(self: *Engine) !void {
104 var roots: std.AutoHashMapUnmanaged(Hash, void) = .empty;
105 defer roots.deinit(self.allocator);
106 var commits: std.AutoHashMapUnmanaged(Hash, void) = .empty;
107 defer commits.deinit(self.allocator);
108 for (self.pack.frames) |frame| {
109 if (frame.kind != .database_root) continue;
110 var reader = Reader.init(frame.payload);
111 const conflicts = try reader.hash();
112 const count = try reader.readU32();
113 if (count > reader.remaining() / 36) return error.InvalidPack;
114 const entries = try self.allocator.alloc(sql.version.RelationEntry, count);
115 defer self.allocator.free(entries);
116 for (entries) |*entry| {
117 entry.* = .{ .name = try reader.readBytes(), .hash = try reader.hash() };
118 if (!self.objects[5].contains(entry.hash) or
119 !self.objects[4].contains(entry.hash)) return error.InvalidPack;
120 }
121 try reader.finish();
122 var root = try sql.version.DatabaseRoot.initSorted(
123 self.allocator,
124 entries,
125 .{ .hash = conflicts },
126 );
127 defer root.deinit();
128 try roots.put(self.allocator, root.hash, {});
129 }
130 for (self.pack.commits) |commit| {
131 try commits.put(self.allocator, commit.hash, {});
132 }
133 for (self.pack.commits) |commit| {
134 if (!roots.contains(commit.root)) return error.InvalidPack;
135 for (commit.parents) |parent| {
136 if (!commits.contains(parent)) return error.InvalidPack;
137 }
138 }
139 for (self.pack.refs) |ref| {
140 if (!commits.contains(ref.target)) return error.InvalidPack;
141 }
142 var row_roots = self.objects[4].keyIterator();
143 while (row_roots.next()) |hash| {
144 if (!self.objects[5].contains(hash.*)) return error.InvalidPack;
145 }
146 }
147
148 fn inspect(self: *Engine) !void {
149 for (self.pack.frames) |frame| {
150 if (frame.kind != .relation_root) continue;
151 var reader = Reader.init(frame.payload);
152 var decoded = try materialize.readRelationRootShallow(self.allocator, &reader);
153 defer decoded.root.deinit();
154 defer self.allocator.free(decoded.index_keys);
155 try reader.finish();
156 try self.inspectRoot(&decoded.root);
157 }
158 }
159
160 fn inspectRoot(self: *Engine, root: *const sql.version.RelationRoot) !void {
161 var aggregate = try Aggregate.init(self.allocator, root.indexes.len + 1);
162 defer aggregate.deinit(self.allocator);
163 var reader = Reader.init(try self.payload(4, root.hash));
164 _ = try reader.hash();
165 const count = try reader.readU32();
166 for (0..count) |_| {
167 const page = try self.aggregatePage(root, try reader.hash());
168 try aggregate.add(self.allocator, page);
169 }
170 try reader.finish();
171 if (try matches(root, &aggregate)) return;
172 if (!std.mem.eql(u8, root.name, self.candidate.relation)) {
173 return error.UnprovablePackRepair;
174 }
175 try aggregate.removeCandidate();
176 if (!try matches(root, &aggregate)) return error.UnprovablePackRepair;
177 try self.rewriteRows(root.hash);
178 self.summary.roots_repaired += 1;
179 self.summary.rows_dropped += 1;
180 }
181
182 fn aggregatePage(
183 self: *Engine,
184 root: *const sql.version.RelationRoot,
185 hash: Hash,
186 ) anyerror!*const Aggregate {
187 const key = cacheKey(root, hash);
188 if (self.pages.getPtr(key)) |found| return found;
189 var aggregate = try Aggregate.init(self.allocator, root.indexes.len + 1);
190 errdefer aggregate.deinit(self.allocator);
191 const payload_bytes = try self.payload(2, hash);
192 const hashes = try hashList(self.allocator, payload_bytes, true);
193 defer self.allocator.free(hashes);
194 for (hashes) |chunk| {
195 const part = try self.aggregateChunk(root, chunk);
196 try aggregate.add(self.allocator, part);
197 }
198 try self.pages.put(self.allocator, key, aggregate);
199 return self.pages.getPtr(key).?;
200 }
201
202 fn aggregateChunk(
203 self: *Engine,
204 root: *const sql.version.RelationRoot,
205 hash: Hash,
206 ) !*const Aggregate {
207 const key = cacheKey(root, hash);
208 if (self.chunks.getPtr(key)) |found| return found;
209 var aggregate = try Aggregate.init(self.allocator, root.indexes.len + 1);
210 errdefer aggregate.deinit(self.allocator);
211 const rows = try self.chunkRows(hash);
212 defer sql.version.freeRelationRows(self.allocator, rows);
213 for (rows) |row| {
214 if (aggregate.last) |last| {
215 if (row.rowid <= last) return error.UnprovablePackRepair;
216 }
217 if (aggregate.first == null) aggregate.first = row.rowid;
218 aggregate.last = row.rowid;
219 aggregate.count += 1;
220 aggregate.bytes = try std.math.add(u64, aggregate.bytes, row.bytes.len);
221 const states = try rowStates(self.allocator, root, row);
222 defer self.allocator.free(states);
223 for (aggregate.states, states) |*state, *part| state.add(part);
224 if (row.rowid == self.candidate.rowid and
225 std.mem.eql(u8, root.name, self.candidate.relation))
226 {
227 aggregate.dropped = try self.allocator.dupe(State, states);
228 aggregate.drop_bytes = row.bytes.len;
229 }
230 }
231 try self.chunks.put(self.allocator, key, aggregate);
232 return self.chunks.getPtr(key).?;
233 }
234
235 fn chunkRows(self: *Engine, hash: Hash) ![]sql.version.RelationRow {
236 var reader = Reader.init(try self.payload(1, hash));
237 _ = try reader.hash();
238 const rows = try record.readRelationRows(self.allocator, &reader);
239 errdefer sql.version.freeRelationRows(self.allocator, rows);
240 try reader.finish();
241 if (!sql.version.same(hash, sql.chunk.digest(rows))) return error.InvalidPack;
242 return rows;
243 }
244
245 fn rewriteRows(self: *Engine, hash: Hash) !void {
246 const i = self.objects[4].get(hash) orelse return error.InvalidPack;
247 const pages = try hashList(self.allocator, self.pack.frames[i].payload, false);
248 defer self.allocator.free(pages);
249 for (pages) |*page| page.* = try self.rewritePage(page.*);
250 const payload_bytes = try encodeHashes(self.allocator, hash, pages);
251 self.allocator.free(self.pack.frames[i].payload);
252 self.pack.frames[i].payload = payload_bytes;
253 }
254
255 fn rewritePage(self: *Engine, hash: Hash) !Hash {
256 if (self.replacements.get(hash)) |found| return found;
257 const chunks = try hashList(self.allocator, try self.payload(2, hash), true);
258 defer self.allocator.free(chunks);
259 var changed = false;
260 for (chunks) |*chunk| {
261 const replacement = try self.rewriteChunk(chunk.*);
262 changed = changed or !sql.version.same(chunk.*, replacement);
263 chunk.* = replacement;
264 }
265 const result = if (changed) sql.chunk.pageDigest(chunks) else hash;
266 if (changed) try self.addFrame(.{
267 .kind = .chunk_index_page,
268 .payload = try encodeHashes(self.allocator, result, chunks),
269 });
270 try self.replacements.put(self.allocator, hash, result);
271 return result;
272 }
273
274 fn rewriteChunk(self: *Engine, hash: Hash) !Hash {
275 if (self.replacements.get(hash)) |found| return found;
276 const rows = try self.chunkRows(hash);
277 defer sql.version.freeRelationRows(self.allocator, rows);
278 var retained: std.ArrayList(sql.version.RelationRow) = .empty;
279 defer retained.deinit(self.allocator);
280 for (rows) |row| {
281 if (row.rowid != self.candidate.rowid) try retained.append(self.allocator, row);
282 }
283 const changed = retained.items.len != rows.len;
284 const result = if (changed) sql.chunk.digest(retained.items) else hash;
285 if (changed) {
286 var payload_bytes: std.ArrayList(u8) = .empty;
287 errdefer payload_bytes.deinit(self.allocator);
288 try record.appendHash(self.allocator, &payload_bytes, result);
289 try record.appendRelationRows(self.allocator, &payload_bytes, retained.items);
290 try self.addFrame(.{
291 .kind = .row_chunk,
292 .payload = try payload_bytes.toOwnedSlice(self.allocator),
293 });
294 }
295 try self.replacements.put(self.allocator, hash, result);
296 return result;
297 }
298
299 fn addFrame(self: *Engine, frame: Frame) !void {
300 errdefer self.allocator.free(frame.payload);
301 const hash = frame.payload[0..32].*;
302 if (self.objects[@backingInt(frame.kind)].contains(hash) or self.added.contains(hash)) {
303 self.allocator.free(frame.payload);
304 return;
305 }
306 try self.added.put(self.allocator, hash, {});
307 try self.additions.append(self.allocator, frame);
308 }
309
310 fn finish(self: *Engine) !void {
311 if (self.additions.items.len == 0) return;
312 const length = try std.math.add(usize, self.pack.frames.len, self.additions.items.len);
313 const frames = try self.allocator.alloc(Frame, length);
314 var cursor: usize = 0;
315 for ([_]record.PackRecordKind{ .row_chunk, .chunk_index_page }) |kind| {
316 for (self.additions.items) |frame| {
317 if (frame.kind != kind) continue;
318 frames[cursor] = frame;
319 cursor += 1;
320 }
321 for (self.pack.frames) |frame| {
322 if (frame.kind != kind) continue;
323 frames[cursor] = frame;
324 cursor += 1;
325 }
326 }
327 for (self.pack.frames) |frame| {
328 if (frame.kind == .row_chunk or frame.kind == .chunk_index_page) continue;
329 frames[cursor] = frame;
330 cursor += 1;
331 }
332 std.debug.assert(cursor == length);
333 self.allocator.free(self.pack.frames);
334 self.pack.frames = frames;
335 self.additions.clearRetainingCapacity();
336 }
337 };
338
339 /// Removes a row an interrupted deletion left behind when an operator repairs a
340 /// backup through the `pack repair` command, rewriting one pack in place to
341 /// repair its row topology around a single candidate row, named by its relation
342 /// and its rowid, and returning how many roots it repaired and how many rows it
343 /// dropped. The function drops the candidate only where every committed
344 /// relation of that history rebuilds to the hash the commit recorded, and
345 /// reports an error when any of them does not, while a relation whose roots
346 /// already agree keeps the row, so the candidate names a row that may be
347 /// dropped. The repair leaves the commits, the database roots, the relation
348 /// roots, and the refs exactly as they were, and replaces only the chunk and
349 /// page objects that disagreed, so a pack that already agrees comes back byte
350 /// for byte and repeating the same repair on the same candidate changes nothing
351 /// further. The caller supplies the allocator, and a failed repair frees what
352 /// it took.
353 pub fn repair(allocator: Allocator, pack: *sql.sync.Pack, candidate: Candidate) !Summary {
354 var engine = Engine{ .allocator = allocator, .pack = pack, .candidate = candidate };
355 defer engine.deinit();
356 try engine.index();
357 try engine.validateLinks();
358 try engine.inspect();
359 try engine.finish();
360 return engine.summary;
361 }
362
363 fn cacheKey(root: *const sql.version.RelationRoot, hash: Hash) [64]u8 {
364 var context = std.crypto.hash.sha2.Sha256.init(.{});
365 context.update(root.name);
366 context.update(&root.schema);
367 var result: [64]u8 = undefined;
368 result[0..32].* = hash;
369 context.final(result[32..64]);
370 return result;
371 }
372
373 fn hashList(allocator: Allocator, payload_bytes: []const u8, page: bool) ![]Hash {
374 var reader = Reader.init(payload_bytes);
375 const expected = try reader.hash();
376 const count = try reader.readU32();
377 if (count > reader.remaining() / 32) return error.InvalidPack;
378 const hashes = try allocator.alloc(Hash, count);
379 errdefer allocator.free(hashes);
380 for (hashes) |*hash| hash.* = try reader.hash();
381 try reader.finish();
382 if (page and !sql.version.same(expected, sql.chunk.pageDigest(hashes))) {
383 return error.InvalidPack;
384 }
385 return hashes;
386 }
387
388 fn encodeHashes(allocator: Allocator, hash: Hash, hashes: []const Hash) ![]u8 {
389 var payload: std.ArrayList(u8) = .empty;
390 errdefer payload.deinit(allocator);
391 try record.appendHash(allocator, &payload, hash);
392 try record.appendU32(allocator, &payload, std.math.cast(u32, hashes.len) orelse
393 return error.InvalidPack);
394 for (hashes) |item| try record.appendHash(allocator, &payload, item);
395 return try payload.toOwnedSlice(allocator);
396 }
397
398 fn rowStates(
399 allocator: Allocator,
400 root: *const sql.version.RelationRoot,
401 row: sql.version.RelationRow,
402 ) ![]State {
403 const result = try allocator.alloc(State, root.indexes.len + 1);
404 errdefer allocator.free(result);
405 var key: [sql.page.size]u8 = undefined;
406 const table_key = try sql.key.encodeRowId(&key, row.rowid);
407 result[0] = sql.lattice.entryState(table_key, row.bytes);
408 const view = try sql.row.View.init(row.bytes);
409 if (root.indexes.len != root.schema_descriptor.indexes.len) return error.InvalidPack;
410 for (root.schema_descriptor.indexes, result[1..]) |index, *state| {
411 if (index.fields.len > sql.relation.max_index_fields) return error.InvalidPack;
412 var values: [sql.relation.max_index_fields]sql.row.Value = undefined;
413 const projected = try view.project(index.fields, &values);
414 const index_key = try sql.key.encodeIndex(&key, projected, index.columns, row.rowid);
415 state.* = sql.lattice.entryState(index_key, "");
416 }
417 return result;
418 }
419
420 const Hasher = struct {
421 value: std.crypto.hash.sha2.Sha256,
422
423 fn init(tag: []const u8) Hasher {
424 var self = Hasher{ .value = std.crypto.hash.sha2.Sha256.init(.{}) };
425 self.bytes(tag);
426 return self;
427 }
428
429 fn integer(self: *Hasher, comptime T: type, value: T) void {
430 var buffer: [@sizeOf(T)]u8 = undefined;
431 std.mem.writeInt(T, &buffer, value, .big);
432 self.value.update(&buffer);
433 }
434
435 fn bytes(self: *Hasher, value: []const u8) void {
436 self.integer(u64, value.len);
437 self.value.update(value);
438 }
439
440 fn finish(self: *Hasher) Hash {
441 var hash: Hash = undefined;
442 self.value.final(&hash);
443 return hash;
444 }
445 };
446
447 fn matches(root: *const sql.version.RelationRoot, aggregate: *const Aggregate) !bool {
448 if (root.format != sql.version.format_version) return error.InvalidPack;
449 const schema = sql.version.schemaHash(
450 root.schema_descriptor.columns,
451 root.schema_descriptor.indexes,
452 );
453 if (!sql.version.same(schema, root.schema)) return error.InvalidPack;
454 if (aggregate.count != root.table.summary.entries or
455 aggregate.bytes != root.table.summary.value_bytes) return false;
456 const table_hash = aggregate.states[0].digest(aggregate.count);
457 if (!sql.version.same(table_hash, root.table.hash)) return false;
458 var hash = Hasher.init("sql.relation");
459 hash.integer(u32, root.format);
460 hash.bytes(root.name);
461 hash.value.update(&schema);
462 hash.value.update(&root.table.hash);
463 hash.value.update(&root.stats.hash);
464 hash.integer(u64, root.indexes.len);
465 for (
466 root.indexes,
467 root.schema_descriptor.indexes,
468 aggregate.states[1..],
469 ) |index, definition, *state| {
470 if (!sql.version.same(state.digest(aggregate.count), index.map.hash)) return false;
471 var fields = Hasher.init("sql.index.fields");
472 fields.integer(u64, definition.fields.len);
473 for (definition.fields) |field| fields.integer(u64, field);
474 fields.integer(u64, definition.columns.len);
475 for (definition.columns) |column| fields.integer(u8, @backingInt(column.collation));
476 const fields_hash = fields.finish();
477 if (!sql.version.same(fields_hash, index.fields)) return error.InvalidPack;
478 var item = Hasher.init("sql.index");
479 item.value.update(&fields_hash);
480 item.value.update(&index.map.hash);
481 item.value.update(&index.stats);
482 const index_hash = item.finish();
483 if (!sql.version.same(index_hash, index.hash)) return error.InvalidPack;
484 hash.value.update(&index_hash);
485 }
486 return sql.version.same(hash.finish(), root.hash);
487 }
488
489 fn execute(connection: *sql.Connection, statement: []const u8) !void {
490 var result = try connection.execute(
491 std.testing.allocator,
492 statement,
493 .{ .durability = .buffered },
494 );
495 defer result.deinit(std.testing.allocator);
496 }
497
498 fn fixture(database: *sql.FileDatabase, history: *sql.History, corrupt: bool) !void {
499 const allocator = std.testing.allocator;
500 var connection = try sql.Connection.create(allocator, database, history, .{});
501 defer connection.deinit();
502 try execute(&connection, "CREATE TABLE items (value, INDEX by_value (value))");
503 for (1..514) |i| {
504 var buffer: [128]u8 = undefined;
505 const statement = try std.fmt.bufPrint(
506 &buffer,
507 "INSERT INTO items (rowid, value) VALUES ({d}, 'value-{d}')",
508 .{ i, i },
509 );
510 try execute(&connection, statement);
511 }
512 try connection.stage();
513 _ = try connection.commit(history);
514 if (!corrupt) return;
515 var base = try connection.materializedWorkingValue(allocator);
516 defer base.deinit();
517 try execute(&connection, "DELETE FROM items WHERE rowid = 1");
518 try execute(&connection, "INSERT INTO items (rowid, value) VALUES (514, 'new')");
519 try publishCorruptFixture(&connection, history, &base);
520 }
521
522 fn publishCorruptFixture(
523 connection: *sql.Connection,
524 history: *sql.History,
525 base: *const sql.version.DatabaseValue,
526 ) !void {
527 const allocator = std.testing.allocator;
528 var value = try connection.materializedWorkingValue(allocator);
529 defer value.deinit();
530 const before = base.findRelation("items").?;
531 const after = value.findRelation("items").?;
532 const rows = try allocator.alloc(sql.version.RelationRow, after.rows.len + 1);
533 defer allocator.free(rows);
534 rows[0] = before.rows[0];
535 @memcpy(rows[1..], after.rows);
536 try history.putRelationRoot(after.root);
537 try history.putRelationRows(after.root.hash, rows);
538 try history.putDatabaseRoot(value.root);
539 _ = try history.commitBranch("main", value.root.hash);
540 }
541
542 /// Builds the pack that the repair tests and the command-line tests run
543 /// against, so they both start from this pack and exercise the same topology,
544 /// by writing two commits over a small database and history in a temporary
545 /// directory, then exporting the whole history as a pack. With `corrupt` true,
546 /// the pack carries the stale row topology an interrupted deletion leaves,
547 /// while with it false, the pack already agrees. The function compiles only in
548 /// test builds, and the caller frees the returned pack.
549 pub fn fixturePack(corrupt: bool) !sql.sync.Pack {
550 if (!@import("builtin").is_test) @compileError("fixturePack is available only in tests");
551 var tmp = std.testing.tmpDir(.{});
552 defer tmp.cleanup();
553 var database = try sql.FileDatabase.openForTesting(std.testing.allocator, tmp.dir, .{
554 .paths = .{ .database = "source.db", .wal = "source.wal" },
555 .header = .{ .sequence = 1, .salt = .{ .first = 17, .second = 19 } },
556 });
557 defer database.deinit();
558 try database.reserve(.{ .wal_frames = 8192 });
559 var history = try sql.History.open(std.testing.allocator, tmp.dir, .{
560 .path = "source.history",
561 .recovery = .reject,
562 });
563 defer history.deinit();
564 try fixture(&database, &history, corrupt);
565 return try sql.exportHistoryPack(std.testing.allocator, &history);
566 }
567
568 test "pack repair proves interrupted deletion and imports unchanged committed identities" {
569 const allocator = std.testing.allocator;
570 var pack = try fixturePack(true);
571 defer pack.deinit();
572 const head = pack.refs[0].target;
573 const commits = pack.commits.len;
574 const result = try repair(allocator, &pack, .{ .relation = "items", .rowid = 1 });
575 try std.testing.expectEqual(@as(usize, 1), result.roots_repaired);
576 try std.testing.expectEqual(@as(usize, 1), result.rows_dropped);
577 try std.testing.expectEqualSlices(u8, &head, &pack.refs[0].target);
578 try std.testing.expectEqual(commits, pack.commits.len);
579 try verifyImport(&pack);
580 const repaired_bytes = try sql.encodeHistoryPack(allocator, &pack);
581 defer allocator.free(repaired_bytes);
582 const repeated = try repair(allocator, &pack, .{ .relation = "items", .rowid = 1 });
583 try std.testing.expectEqual(@as(usize, 0), repeated.roots_repaired);
584 const repeated_bytes = try sql.encodeHistoryPack(allocator, &pack);
585 defer allocator.free(repeated_bytes);
586 try std.testing.expectEqualSlices(u8, repaired_bytes, repeated_bytes);
587 }
588
589 fn verifyImport(pack: *const sql.sync.Pack) !void {
590 const allocator = std.testing.allocator;
591 var tmp = std.testing.tmpDir(.{});
592 defer tmp.cleanup();
593 var database = try sql.FileDatabase.openForTesting(allocator, tmp.dir, .{
594 .paths = .{ .database = "target.db", .wal = "target.wal" },
595 .header = .{ .sequence = 2, .salt = .{ .first = 23, .second = 29 } },
596 });
597 defer database.deinit();
598 try database.reserve(.{ .wal_frames = 8192 });
599 var history = try sql.History.open(allocator, tmp.dir, .{
600 .path = "target.history",
601 .recovery = .reject,
602 });
603 defer history.deinit();
604 _ = try sql.importHistoryPack(&history, pack);
605 var connection = try sql.Connection.open(allocator, &database, &history, .{});
606 defer connection.deinit();
607 try connection.checkoutBranch(allocator, &history, "main");
608 for (pack.commits) |commit| {
609 var value = try history.databaseValue(allocator, commit.root);
610 defer value.deinit();
611 for (value.relations) |relation| {
612 var rebuilt = try sql.version.relationRootFromRows(
613 allocator,
614 &relation.root,
615 relation.rows,
616 );
617 defer rebuilt.deinit();
618 try std.testing.expect(sql.version.same(rebuilt.hash, relation.root.hash));
619 }
620 }
621 }
622
623 test "pack repair preserves consistent bytes and refuses an unproved deletion" {
624 const allocator = std.testing.allocator;
625 var good = try fixturePack(false);
626 defer good.deinit();
627 const before = try sql.encodeHistoryPack(allocator, &good);
628 defer allocator.free(before);
629 const summary = try repair(allocator, &good, .{ .relation = "items", .rowid = 1 });
630 try std.testing.expectEqual(@as(usize, 0), summary.roots_repaired);
631 const after = try sql.encodeHistoryPack(allocator, &good);
632 defer allocator.free(after);
633 try std.testing.expectEqualSlices(u8, before, after);
634 var bad = try fixturePack(true);
635 defer bad.deinit();
636 try std.testing.expectError(
637 error.UnprovablePackRepair,
638 repair(allocator, &bad, .{ .relation = "items", .rowid = 2 }),
639 );
640 }