lib/sql/src/history/resolver.zig
daab053ee43316e1809a84551d573ddd1e5bf3d2
1 const std = @import("std");
2 const sql = @import("../root.zig");
3 const identity = @import("identity.zig");
4 const record_mod = @import("record.zig");
5
6 const version = sql.version;
7
8 pub const Error = std.Io.File.ReadPositionalError || error{
9 CapacityExceeded,
10 CapacityOverflow,
11 InvalidHistory,
12 WrongObject,
13 };
14
15 pub const Limits = struct {
16 metadata_bytes_max: usize = 4 * 1024 * 1024,
17 stream_bytes: usize = 64 * 1024,
18 relation_entries_max: usize = 1024,
19 conflict_entries_max: usize = 1024,
20 };
21
22 pub const Capacity = struct {
23 bytes: usize,
24 relation_entries: usize,
25 conflict_entries: usize,
26
27 pub fn derive(limits: Limits) error{CapacityOverflow}!Capacity {
28 if (limits.stream_bytes < version.hash_bytes) return error.CapacityOverflow;
29 if (limits.metadata_bytes_max < version.hash_bytes) return error.CapacityOverflow;
30 if (limits.metadata_bytes_max > std.math.maxInt(u32)) {
31 return error.CapacityOverflow;
32 }
33 const bytes = std.math.add(
34 usize,
35 limits.metadata_bytes_max,
36 limits.stream_bytes,
37 ) catch return error.CapacityOverflow;
38 _ = std.math.mul(
39 usize,
40 limits.relation_entries_max,
41 @sizeOf(version.RelationEntry),
42 ) catch return error.CapacityOverflow;
43 _ = std.math.mul(
44 usize,
45 limits.conflict_entries_max,
46 @sizeOf(version.ConflictEntry),
47 ) catch return error.CapacityOverflow;
48 return .{
49 .bytes = bytes,
50 .relation_entries = limits.relation_entries_max,
51 .conflict_entries = limits.conflict_entries_max,
52 };
53 }
54 };
55
56 pub const Storage = struct {
57 bytes: []u8,
58 relations: []version.RelationEntry,
59 conflicts: []version.ConflictEntry,
60 };
61
62 pub const Exhaustion = struct {
63 name: []const u8,
64 requested: usize,
65 available: usize,
66 };
67
68 pub const Workspace = struct {
69 metadata: []u8,
70 stream: []u8,
71 relations: []version.RelationEntry,
72 conflicts: []version.ConflictEntry,
73 exhaustion: ?Exhaustion = null,
74
75 /// Activation borrows all storage before any record is read.
76 pub fn activate(limits: Limits, storage: Storage) Error!Workspace {
77 const capacity = try Capacity.derive(limits);
78 if (storage.bytes.len < capacity.bytes) return error.CapacityExceeded;
79 if (storage.relations.len < capacity.relation_entries) return error.CapacityExceeded;
80 if (storage.conflicts.len < capacity.conflict_entries) return error.CapacityExceeded;
81 return .{
82 .metadata = storage.bytes[0..limits.metadata_bytes_max],
83 .stream = storage.bytes[limits.metadata_bytes_max..capacity.bytes],
84 .relations = storage.relations[0..limits.relation_entries_max],
85 .conflicts = storage.conflicts[0..limits.conflict_entries_max],
86 };
87 }
88
89 pub fn scratch(self: *Workspace) identity.Scratch {
90 return .{ .relations = self.relations, .conflicts = self.conflicts };
91 }
92
93 pub fn each(
94 self: *Workspace,
95 kind: record_mod.RecordKind,
96 payload: []const u8,
97 location: identity.Location,
98 context: anytype,
99 comptime accept: fn (@TypeOf(context), identity.Entry) anyerror!void,
100 ) anyerror!void {
101 identity.each(kind, payload, location, self.scratch(), context, accept) catch |err| {
102 if (err != error.KeyScratchExhausted) return err;
103 const count_offset: usize = switch (kind) {
104 .database_root => version.hash_bytes,
105 .conflict_root => 0,
106 else => unreachable,
107 };
108 if (payload.len -| count_offset < 4) return error.InvalidHistory;
109 const requested = record_mod.readIntU32(payload[count_offset..][0..4]);
110 const available = switch (kind) {
111 .database_root => self.relations.len,
112 .conflict_root => self.conflicts.len,
113 else => unreachable,
114 };
115 try self.require("key_entries_max", requested, available);
116 unreachable;
117 };
118 }
119
120 fn require(self: *Workspace, name: []const u8, requested: usize, available: usize) Error!void {
121 if (requested <= available) return;
122 self.exhaustion = .{
123 .name = name,
124 .requested = requested,
125 .available = available,
126 };
127 return error.CapacityExceeded;
128 }
129 };
130
131 pub const Scanned = struct {
132 kind: record_mod.RecordKind,
133 location: identity.Location,
134 payload: []const u8,
135 end: u64,
136 };
137
138 /// A file-order scanner retains only one metadata record or stream chunk.
139 pub const Scanner = struct {
140 io: std.Io,
141 file: std.Io.File,
142 workspace: *Workspace,
143 offset: u64,
144 end: u64,
145 chunk_key: version.Hash = undefined,
146
147 pub fn init(
148 io: std.Io,
149 file: std.Io.File,
150 workspace: *Workspace,
151 start: u64,
152 end: u64,
153 ) Error!Scanner {
154 if (start > end) return error.InvalidHistory;
155 return .{ .io = io, .file = file, .workspace = workspace, .offset = start, .end = end };
156 }
157
158 pub fn next(self: *Scanner) Error!?Scanned {
159 if (self.offset == self.end) return null;
160 if (self.end - self.offset < record_mod.record_header_size) return error.InvalidHistory;
161 var header_bytes: [record_mod.record_header_size]u8 = undefined;
162 try readExact(self.io, self.file, &header_bytes, self.offset);
163 const header = try record_mod.Header.decode(&header_bytes);
164 const payload_start = std.math.add(
165 u64,
166 self.offset,
167 record_mod.record_header_size,
168 ) catch return error.InvalidHistory;
169 const record_end = std.math.add(
170 u64,
171 payload_start,
172 header.payload_len,
173 ) catch return error.InvalidHistory;
174 if (record_end > self.end) return error.InvalidHistory;
175 const location = identity.Location{
176 .record_offset = self.offset,
177 .payload_len = header.payload_len,
178 .envelope_hash = header.envelope_hash,
179 };
180 var hasher = record_mod.envelopeHasher(
181 @backingInt(header.kind),
182 header.payload_len,
183 );
184 const payload: []const u8 = if (header.kind == .row_chunk) chunk: {
185 var cursor = payload_start;
186 var copied: usize = 0;
187 while (cursor < record_end) {
188 const count: usize = @intCast(@min(self.workspace.stream.len, record_end - cursor));
189 try readExact(self.io, self.file, self.workspace.stream[0..count], cursor);
190 if (copied < version.hash_bytes) {
191 const take = @min(version.hash_bytes - copied, count);
192 @memcpy(self.chunk_key[copied..][0..take], self.workspace.stream[0..take]);
193 copied += take;
194 }
195 hasher.update(self.workspace.stream[0..count]);
196 cursor += count;
197 }
198 break :chunk &self.chunk_key;
199 } else metadata: {
200 try self.workspace.require(
201 "metadata_bytes_max",
202 header.payload_len,
203 self.workspace.metadata.len,
204 );
205 const bytes = self.workspace.metadata[0..header.payload_len];
206 try readExact(self.io, self.file, bytes, payload_start);
207 hasher.update(bytes);
208 break :metadata bytes;
209 };
210 var actual: version.Hash = undefined;
211 hasher.final(&actual);
212 if (!version.same(actual, header.envelope_hash)) return error.InvalidHistory;
213 const result = Scanned{
214 .kind = header.kind,
215 .location = location,
216 .payload = payload,
217 .end = record_end,
218 };
219 self.offset = record_end;
220 return result;
221 }
222 };
223
224 pub const Resolver = struct {
225 io: std.Io,
226 file: std.Io.File,
227 indexed_eof: u64,
228 workspace: *Workspace,
229
230 /// The caller owns payload storage and keeps the returned slice alive.
231 pub fn read(
232 self: *Resolver,
233 key: identity.Key,
234 location: identity.Location,
235 output: []u8,
236 ) Error![]const u8 {
237 const end = std.math.add(
238 u64,
239 location.record_offset,
240 location.encodedLen(),
241 ) catch return error.InvalidHistory;
242 if (end > self.indexed_eof) return error.InvalidHistory;
243 try self.workspace.require("record_payload", location.payload_len, output.len);
244 var header_bytes: [record_mod.record_header_size]u8 = undefined;
245 try readExact(self.io, self.file, &header_bytes, location.record_offset);
246 const header = try record_mod.Header.decode(&header_bytes);
247 if (header.payload_len != location.payload_len) return error.InvalidHistory;
248 if (!version.same(header.envelope_hash, location.envelope_hash)) {
249 return error.InvalidHistory;
250 }
251 try readExact(
252 self.io,
253 self.file,
254 output[0..location.payload_len],
255 location.record_offset + record_mod.record_header_size,
256 );
257 const payload = output[0..location.payload_len];
258 if (!version.same(
259 record_mod.recordHash(@backingInt(header.kind), payload),
260 header.envelope_hash,
261 )) return error.InvalidHistory;
262 var found = false;
263 const Context = struct {
264 wanted: identity.Key,
265 position: u32,
266 found: *bool,
267
268 fn accept(context: *@This(), entry: identity.Entry) Error!void {
269 if (entry.key.kind != context.wanted.kind) return;
270 if (entry.location.node_position != context.position) return;
271 context.found.* = version.same(entry.key.bytes, context.wanted.bytes);
272 }
273 };
274 var context = Context{
275 .wanted = key,
276 .position = location.node_position,
277 .found = &found,
278 };
279 self.workspace.each(
280 header.kind,
281 payload,
282 location,
283 &context,
284 Context.accept,
285 ) catch |err| switch (err) {
286 error.CapacityExceeded => return error.CapacityExceeded,
287 else => return error.InvalidHistory,
288 };
289 if (!found) return error.WrongObject;
290 return payload;
291 }
292 };
293
294 fn readExact(io: std.Io, file: std.Io.File, output: []u8, offset: u64) Error!void {
295 var filled: usize = 0;
296 while (filled < output.len) {
297 const position = std.math.add(u64, offset, filled) catch return error.InvalidHistory;
298 const count = try file.readPositionalAll(io, output[filled..], position);
299 if (count == 0) return error.InvalidHistory;
300 filled += count;
301 }
302 }
303
304 test "history resolver keeps the first typed commit location" {
305 const store_mod = @import("store.zig");
306 const allocator = std.testing.allocator;
307 const io = std.Options.debug_io;
308 var tmp = std.testing.tmpDir(.{});
309 defer tmp.cleanup();
310 var file = try tmp.dir.createFile(io, "objects.history", .{ .read = true });
311 defer file.close(io);
312 var payload: std.ArrayList(u8) = .empty;
313 defer payload.deinit(allocator);
314 const root = version.emptyHash("resolver-commit");
315 try record_mod.appendHash(allocator, &payload, root);
316 try record_mod.appendU32(allocator, &payload, 0);
317 const first_end = try store_mod.writeTestingRecord(file, 0, .commit, payload.items);
318 const end = try store_mod.writeTestingRecord(file, first_end, .commit, payload.items);
319 const limits = Limits{
320 .metadata_bytes_max = 128,
321 .stream_bytes = 64,
322 .relation_entries_max = 2,
323 .conflict_entries_max = 2,
324 };
325 const capacity = try Capacity.derive(limits);
326 const bytes = try allocator.alloc(u8, capacity.bytes);
327 defer allocator.free(bytes);
328 const relations = try allocator.alloc(version.RelationEntry, capacity.relation_entries);
329 defer allocator.free(relations);
330 const conflicts = try allocator.alloc(version.ConflictEntry, capacity.conflict_entries);
331 defer allocator.free(conflicts);
332 var workspace = try Workspace.activate(limits, .{
333 .bytes = bytes,
334 .relations = relations,
335 .conflicts = conflicts,
336 });
337 var scanner = try Scanner.init(io, file, &workspace, 0, end);
338 const first = (try scanner.next()).?;
339 try std.testing.expectEqual(@as(u64, 0), first.location.record_offset);
340 const expected = version.Commit.init(root, &.{}).hash;
341 var captured: ?identity.Entry = null;
342 const Capture = struct {
343 fn accept(target: *?identity.Entry, entry: identity.Entry) !void {
344 target.* = entry;
345 }
346 };
347 try workspace.each(
348 first.kind,
349 first.payload,
350 first.location,
351 &captured,
352 Capture.accept,
353 );
354 try std.testing.expect(version.same(expected, captured.?.key.bytes));
355 const second = (try scanner.next()).?;
356 try std.testing.expectEqual(@as(u64, first_end), second.location.record_offset);
357 try std.testing.expect((try scanner.next()) == null);
358 var resolver = Resolver{ .io = io, .file = file, .indexed_eof = end, .workspace = &workspace };
359 var output: [64]u8 = undefined;
360 const read = try resolver.read(captured.?.key, captured.?.location, &output);
361 try std.testing.expectEqualSlices(u8, payload.items, read);
362 }
363
364 test "history resolver typed keys agree with replay across object kinds" {
365 const store_mod = @import("store.zig");
366 const allocator = std.testing.allocator;
367 const io = std.Options.debug_io;
368 var tmp = std.testing.tmpDir(.{});
369 defer tmp.cleanup();
370 const path = "typed.history";
371 var relation = try store_mod.testingMerkleRelationRoot(
372 allocator,
373 .{ .branches = 1, .leaves = 2 },
374 null,
375 "typed",
376 );
377 defer relation.deinit();
378 const rows = try store_mod.testingChunkRows(allocator, 300, 42);
379 defer version.freeRelationRows(allocator, rows);
380 const artifact = version.ConflictArtifact.init("metrics", 7, "old", "ours", "theirs");
381 var database_root: version.DatabaseRoot = undefined;
382 var commit: version.Commit = undefined;
383 {
384 var history = try store_mod.History.open(allocator, tmp.dir, .{
385 .path = path,
386 .recovery = .reject,
387 });
388 defer history.deinit();
389 try history.putRelationRoot(relation);
390 try history.putRelationRows(relation.hash, rows);
391 try history.putConflict(artifact);
392 const conflicts = try history.putConflictRoot(&.{artifact.entry()});
393 database_root = try version.DatabaseRoot.initSorted(allocator, &.{
394 .{ .name = "metrics", .hash = relation.hash },
395 }, conflicts);
396 try history.putDatabaseRoot(database_root);
397 commit = version.Commit.init(database_root.hash, &.{});
398 try history.putCommit(commit);
399 try history.putRef(.{ .name = "main", .target = commit.hash });
400 try history.putRef(.{ .name = "temporary", .target = commit.hash });
401 try history.deleteRef("temporary");
402 const chunk = history.row_chunks.items[0].payload;
403 const duplicate = try allocator.alloc(u8, chunk.len);
404 defer allocator.free(duplicate);
405 try std.testing.expectEqual(
406 chunk.len,
407 try history.file.?.readPositionalAll(io, duplicate, chunk.offset),
408 );
409 try history.appendRecord(.row_chunk, duplicate);
410 try history.flushSync();
411 }
412 defer database_root.deinit();
413 var replay = try store_mod.History.open(allocator, tmp.dir, .{
414 .path = path,
415 .recovery = .reject,
416 .read_only = true,
417 });
418 defer replay.deinit();
419 const verify_mod = @import("verify.zig");
420 const verify_limits = verify_mod.Limits{
421 .metadata_bytes_max = 128 * 1024,
422 .dependencies_max = 4096,
423 .conflict_bytes_max = 4096,
424 .working_bytes_max = 8 * 1024 * 1024,
425 };
426 const verify_capacity = try verify_mod.Capacity.derive(verify_limits);
427 const verify_bytes = try allocator.alloc(u8, verify_capacity.bytes);
428 defer allocator.free(verify_bytes);
429 var verify_workspace = try verify_mod.Workspace.activate(
430 verify_limits,
431 .{ .bytes = verify_bytes },
432 );
433 const report = try verify_mod.verifyFile(
434 &verify_workspace,
435 io,
436 replay.file.?,
437 replay.len(),
438 .{},
439 );
440 try std.testing.expect(report.bad == null);
441 try std.testing.expectEqual(replay.len(), report.last_valid_boundary);
442 const limits = Limits{
443 .metadata_bytes_max = 128 * 1024,
444 .stream_bytes = 4096,
445 .relation_entries_max = 16,
446 .conflict_entries_max = 16,
447 };
448 const capacity = try Capacity.derive(limits);
449 const bytes = try allocator.alloc(u8, capacity.bytes);
450 defer allocator.free(bytes);
451 const relations = try allocator.alloc(version.RelationEntry, capacity.relation_entries);
452 defer allocator.free(relations);
453 const conflicts = try allocator.alloc(version.ConflictEntry, capacity.conflict_entries);
454 defer allocator.free(conflicts);
455 var workspace = try Workspace.activate(limits, .{
456 .bytes = bytes,
457 .relations = relations,
458 .conflicts = conflicts,
459 });
460 var entries: std.ArrayList(identity.Entry) = .empty;
461 defer entries.deinit(allocator);
462 var duplicates: usize = 0;
463 const Collect = struct {
464 allocator: std.mem.Allocator,
465 entries: *std.ArrayList(identity.Entry),
466 duplicates: *usize,
467
468 fn accept(context: *@This(), entry: identity.Entry) !void {
469 for (context.entries.items) |existing| {
470 if (existing.key.kind != entry.key.kind) continue;
471 if (!version.same(existing.key.bytes, entry.key.bytes)) continue;
472 context.duplicates.* += 1;
473 return;
474 }
475 try context.entries.append(context.allocator, entry);
476 }
477 };
478 var collect = Collect{
479 .allocator = allocator,
480 .entries = &entries,
481 .duplicates = &duplicates,
482 };
483 var scanner = try Scanner.init(io, replay.file.?, &workspace, 0, replay.len());
484 while (try scanner.next()) |record| {
485 try workspace.each(
486 record.kind,
487 record.payload,
488 record.location,
489 &collect,
490 Collect.accept,
491 );
492 }
493 try std.testing.expectEqual(@as(usize, 1), duplicates);
494 const replay_count = replay.commits.items.len + replay.conflicts.items.len +
495 replay.database_roots.items.len + replay.relation_roots.items.len +
496 replay.relation_rows.items.len + replay.conflict_roots.items.len +
497 replay.row_chunks.items.len + replay.index_pages.items.len +
498 replay.tree_nodes.items.len + replay.relation_spans.items.len;
499 try std.testing.expectEqual(replay_count, entries.items.len);
500 var resolver = Resolver{
501 .io = io,
502 .file = replay.file.?,
503 .indexed_eof = replay.len(),
504 .workspace = &workspace,
505 };
506 const output = try allocator.alloc(u8, 128 * 1024);
507 defer allocator.free(output);
508 for (entries.items) |entry| {
509 const key = entry.key.bytes;
510 const present = switch (entry.key.kind) {
511 .commit => replay.findCommit(key) != null,
512 .conflict => replay.findConflict(key) != null,
513 .database_root => replay.findDatabaseRoot(key) != null,
514 .relation_root => replay.findRelationRoot(key) != null,
515 .relation_rows => replay.hasRelationRows(key),
516 .conflict_root => replay.findConflictRoot(key) != null,
517 .row_chunk => replay.findRowChunk(key) != null,
518 .chunk_index_page => replay.hasIndexPage(key),
519 .tree_node => replay.findTreeNode(key) != null,
520 .relation_spans => replay.lookup.relation_spans.get(key) != null,
521 };
522 try std.testing.expect(present);
523 _ = try resolver.read(entry.key, entry.location, output);
524 }
525 }