lib/choir/src/product/revision/receipt.zig
daab053ee43316e1809a84551d573ddd1e5bf3d2
1 const std = @import("std");
2 const record = @import("root.zig").record;
3
4 pub const accounting_version: u32 = 1;
5
6 pub const WorkVector = struct {
7 input_bytes: u64 = 0,
8 output_bytes: u64 = 0,
9 structural_visits: u64 = 0,
10 analysis_computations: u64 = 0,
11 rewrite_attempts: u64 = 0,
12 allocation_capacity: u64 = 0,
13
14 pub const Component: type = std.meta.FieldEnum(WorkVector);
15
16 pub fn add(self: WorkVector, other: WorkVector) error{WorkOverflow}!WorkVector {
17 var result: WorkVector = .{};
18 inline for (@typeInfo(WorkVector).@"struct".field_names) |field| {
19 @field(result, field) = std.math.add(
20 u64,
21 @field(self, field),
22 @field(other, field),
23 ) catch return error.WorkOverflow;
24 }
25 return result;
26 }
27
28 pub fn exceeded(self: WorkVector, allowance: WorkVector) ?Component {
29 inline for (@typeInfo(WorkVector).@"struct".field_names) |field| {
30 if (@field(self, field) > @field(allowance, field)) {
31 return @field(Component, field);
32 }
33 }
34 return null;
35 }
36
37 pub fn uniform(value: u64) WorkVector {
38 var result: WorkVector = .{};
39 inline for (@typeInfo(WorkVector).@"struct".field_names) |field| {
40 @field(result, field) = value;
41 }
42 return result;
43 }
44 };
45
46 /// Physical observations are informational and never determine admission.
47 pub const Counters = struct {
48 pass_runs: u64 = 0,
49 passes_modified: u64 = 0,
50 analysis_misses: u64 = 0,
51 analysis_hits: u64 = 0,
52 successful_rewrites: u64 = 0,
53 rewrite_iterations: u64 = 0,
54
55 fn add(self: Counters, other: Counters) !Counters {
56 var result: Counters = .{};
57 inline for (@typeInfo(Counters).@"struct".field_names) |field| {
58 @field(result, field) = std.math.add(
59 u64,
60 @field(self, field),
61 @field(other, field),
62 ) catch return error.WorkOverflow;
63 }
64 return result;
65 }
66 };
67
68 pub const Outcome = enum(u8) { running, success, rejected, exhausted, cancelled };
69 fn failureOutcome(err: anyerror) Outcome {
70 return switch (err) {
71 error.WorkExhausted, error.WorkOverflow, error.WorkTraceLimit => .exhausted,
72 else => .rejected,
73 };
74 }
75
76 pub const Phase = enum(u8) {
77 transport,
78 reservation,
79 input,
80 decode,
81 verification,
82 pass,
83 analysis,
84 capture,
85 gate,
86 output,
87 };
88
89 pub const Limits = struct {
90 allowance: WorkVector,
91 workspace: u64,
92 events: u32,
93 };
94
95 pub const Contract = struct {
96 identity: record.Version,
97 work: WorkVector,
98 workspace: u64 = 0,
99 retained_storage: u64 = 0,
100 };
101
102 pub const Executed = struct {
103 work: WorkVector = .{},
104 counters: Counters = .{},
105 };
106
107 pub const Event = struct {
108 name: [96]u8 = @splat(0),
109 name_length: u8,
110 version: u32,
111 phase: Phase,
112 occurrence: u32,
113 parent: ?u32,
114 charged: WorkVector,
115 executed: Executed = .{},
116 workspace: u64,
117 retained_storage: u64,
118 outcome: Outcome = .running,
119 admitted: bool = false,
120
121 pub fn identity(self: *const Event) record.Version {
122 return .{ .name = self.name[0..self.name_length], .version = self.version };
123 }
124 };
125
126 pub const WorkReceiptV1 = struct {
127 version: u32 = accounting_version,
128 limits: Limits,
129 charged: WorkVector,
130 executed: Executed,
131 events: []const Event,
132 /// Conservative normative ceiling, not an observed allocator or RSS peak.
133 maximum_live_storage: u64,
134 outcome: Outcome,
135 missing_work_contract: bool,
136 exceeded: ?WorkVector.Component,
137 };
138
139 /// A job owns this bounded ledger. Only admitted obligations may execute.
140 pub const AccountingV1 = opaque {
141 pub fn create(
142 allocator: std.mem.Allocator,
143 limits: Limits,
144 pipeline: []const record.Version,
145 ) !*AccountingV1 {
146 return Ledger.create(allocator, limits, pipeline);
147 }
148
149 pub fn destroy(self: *AccountingV1) void {
150 ledger(self).destroy();
151 }
152
153 pub fn begin(self: *AccountingV1, phase: Phase, contract: Contract) !u32 {
154 return ledger(self).begin(phase, contract);
155 }
156
157 pub fn finish(self: *AccountingV1, token: u32, outcome: Outcome, executed: Executed) !void {
158 try ledger(self).finish(token, outcome, executed);
159 }
160
161 /// Record physical screening separately from the fixed transport reservation.
162 pub fn observeTransport(self: *AccountingV1, bytes: u64) !void {
163 const state = ledger(self);
164 if (state.outcome != .running) return error.TerminalWorkOutcome;
165 if (state.count != 4 or state.events[0].phase != .transport) {
166 return error.CompilerWorkAlreadyStarted;
167 }
168 const executed = WorkVector{ .input_bytes = bytes };
169 state.events[0].executed.work = try state.events[0].executed.work.add(executed);
170 state.executed.work = try state.executed.work.add(executed);
171 }
172
173 /// Physical counters do not alter admission or the normative charge trace.
174 /// Closing observations may follow failure; successful completion is immutable.
175 pub fn observeCounters(self: *AccountingV1, counters: Counters) !void {
176 const state = ledger(self);
177 if (state.outcome == .success) return error.TerminalWorkOutcome;
178 const total = state.executed.counters.add(counters) catch {
179 return state.reject(error.WorkOverflow);
180 };
181 if (state.current) |token| {
182 const event = &state.events[token];
183 event.executed.counters = event.executed.counters.add(counters) catch {
184 return state.reject(error.WorkOverflow);
185 };
186 }
187 state.executed.counters = total;
188 }
189
190 pub fn missingContract(self: *AccountingV1) void {
191 std.debug.assert(ledger(self).outcome == .running);
192 ledger(self).missing_work_contract = true;
193 }
194
195 pub fn fail(self: *AccountingV1, outcome: Outcome) void {
196 std.debug.assert(outcome != .running);
197 std.debug.assert(outcome != .success);
198 if (ledger(self).outcome == .running) ledger(self).outcome = outcome;
199 }
200
201 pub fn producersComplete(self: *AccountingV1) !void {
202 try ledger(self).producersComplete();
203 }
204
205 pub fn complete(self: *AccountingV1) !void {
206 const state = ledger(self);
207 try state.producersComplete();
208 const required = [_]Phase{ .transport, .reservation, .input, .capture, .gate, .output };
209 for (required) |phase| {
210 if (!state.completed.contains(phase)) {
211 state.missing_work_contract = true;
212 return error.MissingWorkContract;
213 }
214 }
215 state.outcome = .success;
216 }
217
218 pub fn view(self: *const AccountingV1) WorkReceiptV1 {
219 const state: *const Ledger = @ptrCast(@alignCast(self));
220 return state.view();
221 }
222
223 /// Only a locally sealed store may treat a replay as reuse entitlement.
224 pub fn replay(self: *AccountingV1, cold: WorkReceiptV1, skip: u32) !void {
225 if (cold.version != accounting_version) return error.AccountingVersionMismatch;
226 if (cold.outcome != .success or cold.missing_work_contract) {
227 return error.MissingWorkContract;
228 }
229 if (skip > cold.events.len) return error.InvalidWorkTrace;
230 const state = ledger(self);
231 for (cold.events[skip..]) |*event| {
232 if (!event.admitted or event.outcome != .success) return error.InvalidWorkTrace;
233 while (state.current != event.parent) {
234 const token = state.current orelse return error.InvalidWorkTrace;
235 try self.finish(token, .success, .{});
236 }
237 if (event.occurrence != state.count) return error.InvalidWorkTrace;
238 _ = try self.begin(event.phase, .{
239 .identity = event.identity(),
240 .work = event.charged,
241 .workspace = event.workspace,
242 .retained_storage = event.retained_storage,
243 });
244 }
245 while (state.current) |token| try self.finish(token, .success, .{});
246 }
247 };
248
249 const Ledger = struct {
250 allocator: std.mem.Allocator,
251 limits: Limits,
252 events: []Event,
253 count: u32 = 0,
254 current: ?u32 = null,
255 pipeline: []record.Version,
256 next_pass: usize = 0,
257 charged: WorkVector = .{},
258 executed: Executed = .{},
259 maximum_live_storage: u64 = 0,
260 retained_storage: u64 = 0,
261 outcome: Outcome = .running,
262 missing_work_contract: bool = false,
263 exceeded: ?WorkVector.Component = null,
264 completed: std.EnumSet(Phase) = .{},
265
266 fn create(
267 allocator: std.mem.Allocator,
268 limits: Limits,
269 pipeline: []const record.Version,
270 ) !*AccountingV1 {
271 if (limits.events == 0) return error.WorkTraceLimit;
272 const control_bytes = try storageCapacity(limits, pipeline);
273 const self = try allocator.create(Ledger);
274 errdefer allocator.destroy(self);
275 const events = try allocator.alloc(Event, limits.events);
276 errdefer allocator.free(events);
277 const owned_pipeline = try clonePipeline(allocator, pipeline);
278 self.* = .{
279 .allocator = allocator,
280 .limits = limits,
281 .events = events,
282 .pipeline = owned_pipeline,
283 .retained_storage = control_bytes,
284 .maximum_live_storage = control_bytes,
285 };
286 return @ptrCast(self);
287 }
288
289 fn destroy(self: *Ledger) void {
290 const allocator = self.allocator;
291 for (self.pipeline) |version| allocator.free(version.name);
292 allocator.free(self.pipeline);
293 allocator.free(self.events);
294 allocator.destroy(self);
295 }
296
297 fn begin(self: *Ledger, phase: Phase, contract: Contract) !u32 {
298 if (self.outcome != .running) return error.TerminalWorkOutcome;
299 if (self.count == self.events.len) return self.reject(error.WorkTraceLimit);
300 if (contract.identity.name.len > 96) return self.reject(error.ObligationNameTooLong);
301 if (contract.identity.name.len == 0 or contract.identity.version == 0) {
302 return self.reject(error.MissingWorkContract);
303 }
304 if (phase == .pass) try self.admitPass(contract.identity);
305 const token = self.count;
306 const event = &self.events[token];
307 event.* = .{
308 .name_length = @intCast(contract.identity.name.len),
309 .version = contract.identity.version,
310 .phase = phase,
311 .occurrence = token,
312 .parent = self.current,
313 .charged = contract.work,
314 .workspace = contract.workspace,
315 .retained_storage = contract.retained_storage,
316 };
317 @memcpy(event.name[0..event.name_length], contract.identity.name);
318 self.count += 1;
319 self.charged = self.charged.add(contract.work) catch {
320 event.outcome = .exhausted;
321 self.outcome = .exhausted;
322 return error.WorkOverflow;
323 };
324 try self.admitStorage(contract, event);
325 event.admitted = true;
326 self.current = token;
327 return token;
328 }
329
330 fn admitStorage(self: *Ledger, contract: Contract, event: *Event) !void {
331 const live = std.math.add(
332 u64,
333 self.retained_storage,
334 @max(contract.workspace, contract.retained_storage),
335 ) catch {
336 event.outcome = .exhausted;
337 return self.reject(error.WorkOverflow);
338 };
339 self.maximum_live_storage = @max(self.maximum_live_storage, live);
340 self.exceeded = self.charged.exceeded(self.limits.allowance);
341 if (self.exceeded != null or contract.workspace > self.limits.workspace) {
342 event.outcome = .exhausted;
343 self.outcome = .exhausted;
344 return error.WorkExhausted;
345 }
346 self.retained_storage = std.math.add(
347 u64,
348 self.retained_storage,
349 contract.retained_storage,
350 ) catch {
351 event.outcome = .exhausted;
352 return self.reject(error.WorkOverflow);
353 };
354 }
355
356 fn admitPass(self: *Ledger, identity: record.Version) !void {
357 if (self.next_pass == self.pipeline.len) return self.reject(error.MissingWorkContract);
358 if (!identity.eql(self.pipeline[self.next_pass])) {
359 return self.reject(error.WorkContractMismatch);
360 }
361 self.next_pass += 1;
362 }
363
364 fn finish(self: *Ledger, token: u32, outcome: Outcome, executed: Executed) !void {
365 std.debug.assert(token < self.count);
366 std.debug.assert(self.current == token);
367 std.debug.assert(outcome != .running);
368 const event = &self.events[token];
369 std.debug.assert(event.outcome == .running);
370 const final_outcome = if (self.outcome == .running) outcome else self.outcome;
371 event.outcome = final_outcome;
372 self.current = event.parent;
373 event.executed.work = executed.work;
374 event.executed.counters = event.executed.counters.add(executed.counters) catch {
375 event.outcome = .exhausted;
376 return self.reject(error.WorkOverflow);
377 };
378 self.executed.work = self.executed.work.add(executed.work) catch {
379 return self.reject(error.WorkOverflow);
380 };
381 self.executed.counters = self.executed.counters.add(executed.counters) catch {
382 return self.reject(error.WorkOverflow);
383 };
384 if (final_outcome == .success) {
385 self.completed.insert(event.phase);
386 } else if (self.outcome == .running) {
387 self.outcome = final_outcome;
388 }
389 }
390
391 fn producersComplete(self: *Ledger) !void {
392 if (self.outcome == .exhausted) return error.WorkExhausted;
393 if (self.outcome != .running) return error.TerminalWorkOutcome;
394 if (self.current != null or self.missing_work_contract or
395 self.next_pass != self.pipeline.len)
396 {
397 self.missing_work_contract = true;
398 return error.MissingWorkContract;
399 }
400 }
401
402 fn reject(self: *Ledger, err: anyerror) anyerror {
403 if (self.outcome == .running) self.outcome = failureOutcome(err);
404 if (err == error.MissingWorkContract) self.missing_work_contract = true;
405 return err;
406 }
407
408 fn view(self: *const Ledger) WorkReceiptV1 {
409 return .{
410 .limits = self.limits,
411 .charged = self.charged,
412 .executed = self.executed,
413 .events = self.events[0..self.count],
414 .maximum_live_storage = self.maximum_live_storage,
415 .outcome = self.outcome,
416 .missing_work_contract = self.missing_work_contract,
417 .exceeded = self.exceeded,
418 };
419 }
420 };
421
422 fn ledger(handle: *AccountingV1) *Ledger {
423 return @ptrCast(@alignCast(handle));
424 }
425
426 fn storageCapacity(limits: Limits, pipeline: []const record.Version) !u64 {
427 var size = std.math.mul(u64, limits.events, @sizeOf(Event)) catch
428 return error.WorkOverflow;
429 size = std.math.add(u64, size, @sizeOf(Ledger)) catch return error.WorkOverflow;
430 const entries = std.math.mul(u64, pipeline.len, @sizeOf(record.Version)) catch
431 return error.WorkOverflow;
432 size = std.math.add(u64, size, entries) catch return error.WorkOverflow;
433 for (pipeline) |version| {
434 size = std.math.add(u64, size, version.name.len) catch return error.WorkOverflow;
435 }
436 return size;
437 }
438
439 fn clonePipeline(allocator: std.mem.Allocator, pipeline: []const record.Version) ![]record.Version {
440 const copy = try allocator.alloc(record.Version, pipeline.len);
441 errdefer allocator.free(copy);
442 var initialized: usize = 0;
443 errdefer for (copy[0..initialized]) |version| allocator.free(version.name);
444 for (pipeline, copy) |version, *target| {
445 target.* = .{
446 .name = try allocator.dupe(u8, version.name),
447 .version = version.version,
448 };
449 initialized += 1;
450 }
451 return copy;
452 }
453
454 test "revision accounting refuses a producer before execution and retains prior charges" {
455 const pipeline = [_]record.Version{.{ .name = "producer", .version = 1 }};
456 const job = try AccountingV1.create(std.testing.allocator, .{
457 .allowance = .{ .structural_visits = 4 },
458 .workspace = 20,
459 .events = 8,
460 }, &pipeline);
461 defer job.destroy();
462 const input = try job.begin(.input, .{
463 .identity = .{ .name = "input", .version = 1 },
464 .work = .{ .structural_visits = 2 },
465 });
466 try job.finish(input, .success, .{ .work = .{ .structural_visits = 2 } });
467 try std.testing.expectError(error.WorkExhausted, job.begin(.pass, .{
468 .identity = pipeline[0],
469 .work = .{ .structural_visits = 3 },
470 .workspace = 20,
471 }));
472 const view = job.view();
473 try std.testing.expectEqual(5, view.charged.structural_visits);
474 try std.testing.expectEqual(2, view.executed.work.structural_visits);
475 try std.testing.expectEqual(0, view.executed.counters.pass_runs);
476 try std.testing.expectEqual(.exhausted, view.outcome);
477 try std.testing.expectError(error.TerminalWorkOutcome, job.begin(.input, .{
478 .identity = .{ .name = "retry", .version = 1 },
479 .work = .{},
480 }));
481 }
482
483 test "revision accounting rejects omitted contracts and incomplete programs" {
484 const pipeline = [_]record.Version{.{ .name = "required", .version = 1 }};
485 const job = try AccountingV1.create(std.testing.allocator, .{
486 .allowance = WorkVector.uniform(100),
487 .workspace = 100,
488 .events = 8,
489 }, &pipeline);
490 defer job.destroy();
491 try std.testing.expectError(error.MissingWorkContract, job.producersComplete());
492 job.missingContract();
493 const token = try job.begin(.pass, .{ .identity = pipeline[0], .work = .{} });
494 try job.finish(token, .success, .{ .counters = .{ .pass_runs = 1 } });
495 try std.testing.expectError(error.MissingWorkContract, job.producersComplete());
496 try std.testing.expect(job.view().missing_work_contract);
497 }
498
499 test "revision accounting overflow is terminal and unknown versions cannot replay" {
500 const job = try AccountingV1.create(std.testing.allocator, .{
501 .allowance = WorkVector.uniform(std.math.maxInt(u64)),
502 .workspace = 0,
503 .events = 4,
504 }, &.{});
505 defer job.destroy();
506 const token = try job.begin(.input, .{
507 .identity = .{ .name = "input", .version = 1 },
508 .work = .{ .input_bytes = std.math.maxInt(u64) },
509 });
510 try job.finish(token, .success, .{});
511 var imported = job.view();
512 imported.version += 1;
513 try std.testing.expectError(error.AccountingVersionMismatch, job.replay(imported, 0));
514 try std.testing.expectError(error.WorkOverflow, job.begin(.decode, .{
515 .identity = .{ .name = "decode", .version = 1 },
516 .work = .{ .input_bytes = 1 },
517 }));
518 try std.testing.expectEqual(.exhausted, job.view().outcome);
519 try std.testing.expectEqual(std.math.maxInt(u64), job.view().charged.input_bytes);
520 try std.testing.expectEqual(1, job.view().events[1].charged.input_bytes);
521 try std.testing.expect(!job.view().events[1].admitted);
522 }
523
524 test "revision accounting includes pre-execution control storage in its live ceiling" {
525 var observed = std.testing.FailingAllocator.init(std.testing.allocator, .{});
526 const pipeline = [_]record.Version{.{ .name = "producer", .version = 1 }};
527 const work = try AccountingV1.create(observed.allocator(), .{
528 .allowance = WorkVector.uniform(100),
529 .workspace = 100,
530 .events = 8,
531 }, &pipeline);
532 defer work.destroy();
533 try std.testing.expectEqual(observed.allocated_bytes, work.view().maximum_live_storage);
534 try std.testing.expectEqual(0, work.view().events.len);
535 try std.testing.expectEqualDeep(WorkVector{}, work.view().charged);
536 }
537
538 test "revision accounting retains its first failure while closing physical observations" {
539 const job = try AccountingV1.create(std.testing.allocator, .{
540 .allowance = .{},
541 .workspace = 0,
542 .events = 1,
543 }, &.{});
544 defer job.destroy();
545 job.missingContract();
546 job.fail(.cancelled);
547 try job.observeCounters(.{ .pass_runs = 1 });
548 try std.testing.expectEqual(1, job.view().executed.counters.pass_runs);
549 try std.testing.expectEqual(.cancelled, job.view().outcome);
550 try std.testing.expectError(error.WorkOverflow, job.observeCounters(.{
551 .pass_runs = std.math.maxInt(u64),
552 }));
553 try std.testing.expectEqual(.cancelled, job.view().outcome);
554 try std.testing.expectEqual(1, job.view().executed.counters.pass_runs);
555 }