lib/sql/src/history/verify.zig
daab053ee43316e1809a84551d573ddd1e5bf3d2
1 const std = @import("std");
2 const sys = @import("sys");
3 const sql = @import("../root.zig");
4 const record_mod = @import("record.zig");
5 const batch_mod = @import("batch.zig");
6 const validate_mod = @import("validate.zig");
7 const version = sql.version;
8
9 const max_lanes: usize = 4;
10
11 /// A complete scan uses caller storage and never opens the history for write.
12 pub const Limits = struct {
13 metadata_bytes_max: usize = 4 * 1024 * 1024,
14 dependencies_max: usize = 2 * 1024 * 1024,
15 conflict_bytes_max: usize = 4 * 1024 * 1024,
16 working_bytes_max: usize,
17 };
18
19 pub const Capacity = struct {
20 bytes: usize,
21
22 pub fn derive(limits: Limits) error{CapacityOverflow}!Capacity {
23 if (limits.metadata_bytes_max == 0 or limits.dependencies_max == 0 or
24 limits.conflict_bytes_max == 0 or limits.working_bytes_max == 0)
25 {
26 return error.CapacityOverflow;
27 }
28 if (limits.conflict_bytes_max > std.math.maxInt(u32)) {
29 return error.CapacityOverflow;
30 }
31 _ = std.math.add(
32 usize,
33 limits.metadata_bytes_max,
34 limits.conflict_bytes_max,
35 ) catch return error.CapacityOverflow;
36 return .{ .bytes = limits.working_bytes_max };
37 }
38 };
39
40 pub const Storage = struct {
41 bytes: []u8,
42 };
43
44 pub const Exhaustion = struct {
45 name: []const u8,
46 bytes_in_use: usize,
47 requested: usize,
48 available: usize,
49 };
50
51 pub const Workspace = struct {
52 storage: []u8,
53 fixed: std.heap.FixedBufferAllocator,
54 limits: Limits,
55 exhaustion: ?Exhaustion = null,
56
57 fn allocator(self: *Workspace) std.mem.Allocator {
58 return .{
59 .ptr = self,
60 .vtable = &.{
61 .alloc = alloc,
62 .resize = resize,
63 .remap = remap,
64 .free = free,
65 },
66 };
67 }
68
69 fn alloc(ctx: *anyopaque, len: usize, alignment: std.mem.Alignment, ra: usize) ?[*]u8 {
70 const self: *Workspace = @ptrCast(@alignCast(ctx));
71 const underlying = self.fixed.allocator();
72 const result = underlying.vtable.alloc(underlying.ptr, len, alignment, ra);
73 if (result == null) self.noteExhaustion(len);
74 return result;
75 }
76
77 fn resize(
78 ctx: *anyopaque,
79 memory: []u8,
80 alignment: std.mem.Alignment,
81 len: usize,
82 ra: usize,
83 ) bool {
84 const self: *Workspace = @ptrCast(@alignCast(ctx));
85 const underlying = self.fixed.allocator();
86 return underlying.vtable.resize(underlying.ptr, memory, alignment, len, ra);
87 }
88
89 fn remap(
90 ctx: *anyopaque,
91 memory: []u8,
92 alignment: std.mem.Alignment,
93 len: usize,
94 ra: usize,
95 ) ?[*]u8 {
96 const self: *Workspace = @ptrCast(@alignCast(ctx));
97 const underlying = self.fixed.allocator();
98 return underlying.vtable.remap(underlying.ptr, memory, alignment, len, ra);
99 }
100
101 fn free(ctx: *anyopaque, memory: []u8, alignment: std.mem.Alignment, ra: usize) void {
102 const self: *Workspace = @ptrCast(@alignCast(ctx));
103 const underlying = self.fixed.allocator();
104 underlying.vtable.free(underlying.ptr, memory, alignment, ra);
105 }
106
107 fn noteExhaustion(self: *Workspace, requested: usize) void {
108 const used = self.fixed.end_index;
109 self.exhaustion = .{
110 .name = "working_bytes_max",
111 .bytes_in_use = used,
112 .requested = requested,
113 .available = self.storage.len - used,
114 };
115 }
116
117 pub fn activate(limits: Limits, storage: Storage) error{
118 CapacityOverflow,
119 CapacityExceeded,
120 }!Workspace {
121 const capacity = try Capacity.derive(limits);
122 if (storage.bytes.len < capacity.bytes) return error.CapacityExceeded;
123 const bytes = storage.bytes[0..capacity.bytes];
124 return .{
125 .storage = bytes,
126 .fixed = std.heap.FixedBufferAllocator.init(bytes),
127 .limits = limits,
128 };
129 }
130
131 pub fn reset(self: *Workspace) void {
132 self.fixed.reset();
133 self.exhaustion = null;
134 }
135 };
136
137 pub const Report = validate_mod.Inspection;
138
139 /// Reports the earliest bad record while leaving the file untouched.
140 pub fn verifyFile(
141 workspace: *Workspace,
142 io: std.Io,
143 file: std.Io.File,
144 end: usize,
145 control: sql.wal.Control,
146 ) !Report {
147 workspace.reset();
148 return validate_mod.inspectWhole(
149 workspace.allocator(),
150 io,
151 file,
152 end,
153 .{
154 .metadata_bytes_max = workspace.limits.metadata_bytes_max,
155 .dependencies_max = workspace.limits.dependencies_max,
156 .conflict_bytes_max = workspace.limits.conflict_bytes_max,
157 },
158 control,
159 ) catch |err| switch (err) {
160 error.OutOfMemory => {
161 std.debug.assert(workspace.exhaustion != null);
162 return error.CapacityExceeded;
163 },
164 else => return err,
165 };
166 }
167
168 const Work = struct {
169 bytes: []const u8 = &.{},
170 records: []batch_mod.Record = &.{},
171 };
172
173 pub const Workers = struct {
174 handles: [max_lanes - 1]sys.thread.JoinHandle = undefined,
175 started: usize = 0,
176 mutex: sys.thread.Mutex = .{},
177 ready: sys.thread.Condition = .{},
178 done: sys.thread.Condition = .{},
179 generation: usize = 0,
180 completed: usize = 0,
181 shutdown: bool = false,
182 work: Work = .{},
183 next_index: std.atomic.Value(usize) = std.atomic.Value(usize).init(0),
184
185 pub fn start(self: *Workers, requested_lanes: usize) void {
186 if (comptime !sys.thread.threadsSupported()) return;
187 const background = @min(requested_lanes -| 1, self.handles.len);
188 while (self.started < background) {
189 self.handles[self.started] = sys.thread.spawn(workerLoop, .{self}) catch return;
190 self.started += 1;
191 }
192 }
193
194 pub fn deinit(self: *Workers) void {
195 self.mutex.lock();
196 self.shutdown = true;
197 self.ready.broadcast();
198 self.mutex.unlock();
199 for (self.handles[0..self.started]) |handle| handle.join();
200 self.* = undefined;
201 }
202
203 pub fn verify(self: *Workers, bytes: []const u8, records: []batch_mod.Record) void {
204 if (records.len == 0) return;
205 if (self.started == 0 or records.len == 1) {
206 verifySerial(bytes, records);
207 return;
208 }
209
210 self.mutex.lock();
211 self.work = .{ .bytes = bytes, .records = records };
212 self.next_index.store(0, .monotonic);
213 self.completed = 0;
214 self.generation +%= 1;
215 self.ready.broadcast();
216 self.mutex.unlock();
217
218 runWork(self, self.work);
219
220 self.mutex.lock();
221 while (self.completed != self.started) self.done.wait(&self.mutex);
222 self.mutex.unlock();
223 }
224
225 fn workerLoop(self: *Workers) void {
226 var observed_generation: usize = 0;
227 self.mutex.lock();
228 while (true) {
229 while (!self.shutdown and self.generation == observed_generation) self.ready.wait(&self.mutex);
230 if (self.shutdown) {
231 self.mutex.unlock();
232 return;
233 }
234 observed_generation = self.generation;
235 const work = self.work;
236 self.mutex.unlock();
237 runWork(self, work);
238 self.mutex.lock();
239 self.completed += 1;
240 if (self.completed == self.started) self.done.signal();
241 }
242 }
243 };
244
245 fn runWork(workers: *Workers, work: Work) void {
246 while (true) {
247 const index = workers.next_index.fetchAdd(1, .monotonic);
248 if (index >= work.records.len) return;
249 verifyRecord(work.bytes, &work.records[index]);
250 }
251 }
252
253 fn verifySerial(bytes: []const u8, records: []batch_mod.Record) void {
254 for (records) |*record| verifyRecord(bytes, record);
255 }
256
257 fn verifyRecord(bytes: []const u8, record: *batch_mod.Record) void {
258 const payload = bytes[record.payload_start..][0..record.payload_len];
259 record.valid_hash = version.same(record.expected, record_mod.recordHash(record.kind_value, payload));
260 }
261
262 test "history verification reports a missing dependency without writing" {
263 const store_mod = @import("store.zig");
264 const allocator = std.testing.allocator;
265 const io = std.Options.debug_io;
266 var tmp = std.testing.tmpDir(.{});
267 defer tmp.cleanup();
268 var file = try tmp.dir.createFile(io, "verify.history", .{ .read = true });
269 defer file.close(io);
270 var payload: std.ArrayList(u8) = .empty;
271 defer payload.deinit(allocator);
272 const root = version.emptyHash("verify-root");
273 try record_mod.appendHash(allocator, &payload, root);
274 try record_mod.appendU32(allocator, &payload, 0);
275 const boundary = try store_mod.writeTestingRecord(file, 0, .commit, payload.items);
276 payload.clearRetainingCapacity();
277 const missing = version.emptyHash("verify-missing");
278 try record_mod.appendHash(allocator, &payload, missing);
279 try record_mod.appendBytes(allocator, &payload, "main");
280 const end = try store_mod.writeTestingRecord(file, boundary, .ref, payload.items);
281 var before: [256]u8 = undefined;
282 try std.testing.expect(end <= before.len);
283 try std.testing.expectEqual(end, try file.readPositionalAll(io, before[0..end], 0));
284 const limits = Limits{
285 .metadata_bytes_max = 128,
286 .dependencies_max = 16,
287 .conflict_bytes_max = 128,
288 .working_bytes_max = 1024 * 1024,
289 };
290 const capacity = try Capacity.derive(limits);
291 const bytes = try allocator.alloc(u8, capacity.bytes);
292 defer allocator.free(bytes);
293 var workspace = try Workspace.activate(limits, .{ .bytes = bytes });
294 const report = try verifyFile(&workspace, io, file, end, .{});
295 try std.testing.expectEqual(boundary, report.last_valid_boundary);
296 try std.testing.expectEqual(boundary, report.bad.?.offset);
297 try std.testing.expectEqual(record_mod.RecordKind.ref, report.bad.?.kind.?);
298 try std.testing.expect(version.same(missing, report.bad.?.dependency.?.key));
299 var after: [256]u8 = undefined;
300 try std.testing.expectEqual(end, try file.readPositionalAll(io, after[0..end], 0));
301 try std.testing.expectEqualSlices(u8, before[0..end], after[0..end]);
302 after[boundary] ^= 0xff;
303 try file.writePositionalAll(io, after[boundary..][0..1], boundary);
304 const malformed = try verifyFile(&workspace, io, file, end, .{});
305 try std.testing.expectEqual(boundary, malformed.last_valid_boundary);
306 try std.testing.expectEqual(boundary, malformed.bad.?.offset);
307 try std.testing.expect(malformed.bad.?.kind == null);
308 var unchanged: [256]u8 = undefined;
309 try std.testing.expectEqual(end, try file.readPositionalAll(io, unchanged[0..end], 0));
310 try std.testing.expectEqualSlices(u8, after[0..end], unchanged[0..end]);
311
312 const small_limits = Limits{
313 .metadata_bytes_max = 128,
314 .dependencies_max = 16,
315 .conflict_bytes_max = 128,
316 .working_bytes_max = 64,
317 };
318 var small_storage: [64]u8 = undefined;
319 var small = try Workspace.activate(small_limits, .{ .bytes = &small_storage });
320 try std.testing.expectError(error.CapacityExceeded, verifyFile(&small, io, file, end, .{}));
321 const exhausted = small.exhaustion.?;
322 try std.testing.expectEqualStrings("working_bytes_max", exhausted.name);
323 try std.testing.expectEqual(small.fixed.end_index, exhausted.bytes_in_use);
324 try std.testing.expectEqual(small_storage.len - exhausted.bytes_in_use, exhausted.available);
325 try std.testing.expect(exhausted.requested > 0);
326 }