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 }