lib/machine/src/fabric/owner.zig
daab053ee43316e1809a84551d573ddd1e5bf3d2
1 const admission = @import("../admission/root.zig");
2 const canon = @import("canon.zig");
3 const fault = @import("../fault/root.zig");
4 const instance = @import("../instance/root.zig");
5 const os = @import("os");
6 const profile = @import("../profile/root.zig");
7 const std = @import("std");
8 const types = @import("types.zig");
9
10 const FabricError = error{
11 CapacityStateMismatch,
12 EffectRequestRequired,
13 EntropyCapacityExceeded,
14 EntrySequenceMismatch,
15 EntryTypeMismatch,
16 FaultAlternativesEqual,
17 FaultEffectMismatch,
18 FaultFrontierExhausted,
19 FaultRequiresAdmission,
20 FaultStatusInvalid,
21 InvalidQuiescenceReceipt,
22 LinkPartitioned,
23 MachineBasisMismatch,
24 NoTurnPending,
25 NodeUnavailable,
26 PacketDestinationMismatch,
27 PacketDelayed,
28 PacketIdExhausted,
29 PacketNotReady,
30 PacketOrderInvalid,
31 PacketUnknown,
32 PartitionStateMismatch,
33 ProcessIdentityInvalid,
34 ReceiptMismatch,
35 TurnPending,
36 UnknownNode,
37 VirtualTimeRegression,
38 WorldMismatch,
39 };
40
41 pub const Error = canon.Error || profile.Error || instance.BasisError || FabricError;
42
43 pub const Fabric = struct {
44 state: types.State,
45
46 pub fn init(input: types.Input) Error!Fabric {
47 try profile.validate(input.profile);
48 if (input.nodes.len == 0 or input.nodes.len > types.node_limit or
49 input.nodes.len > input.profile.contract.instance_limit)
50 {
51 return error.NodeCapacityExceeded;
52 }
53 if (os.abi.wire.allZero(&input.world)) return error.InvalidWorld;
54 const contract = try profile.contractFingerprint(input.profile);
55 var state: types.State = .{
56 .contract = contract,
57 .world = input.world,
58 .node_count = @intCast(input.nodes.len),
59 .nodes = undefined,
60 .virtual_time_tick = input.virtual_time_tick,
61 .entropy_frontier = 0,
62 .entropy_bytes = 0,
63 .packet_next_id = 0,
64 .packet_count = 0,
65 .packets = undefined,
66 .partition_bits = 0,
67 .capacity_bits = 0,
68 .pending = null,
69 .entry_frontier = 0,
70 .admission_frontier = 0,
71 .fault_frontier = 0,
72 .admission_limit = types.admission_limit,
73 .ledger_digest = @splat(0),
74 .fault_digest = @splat(1),
75 .root = .{
76 .digest = @splat(1),
77 .dialect = .ordered_effect_fabric_v3,
78 .machine_contract = contract,
79 .entry_frontier = 0,
80 .admission_frontier = 0,
81 .fault_frontier = 0,
82 },
83 };
84 for (input.nodes, 0..) |node, index| {
85 state.nodes[index] = .{
86 .id = node.id,
87 .basis = node.basis,
88 .machine = .{
89 .kind = .basis,
90 .digest = canon.basisDigest(node.basis),
91 },
92 .available = true,
93 };
94 }
95 try canon.initialize(&state);
96 return .{ .state = state };
97 }
98
99 pub fn root(self: *const Fabric) types.Root {
100 return self.state.root;
101 }
102
103 pub fn cut(self: *const Fabric) Error!types.Cut {
104 try canon.validate(self.state);
105 if (self.state.pending != null) return error.TurnPending;
106 var result: types.Cut = .{
107 .root = self.state.root,
108 .node_count = self.state.node_count,
109 .nodes = @splat(emptyNodeBoundary()),
110 };
111 for (self.state.nodes[0..self.state.node_count], 0..) |node, index| {
112 result.nodes[index] = .{
113 .id = node.id,
114 .available = node.available,
115 .machine = node.machine,
116 };
117 }
118 return result;
119 }
120
121 pub fn turnPending(self: *const Fabric) bool {
122 return self.state.pending != null;
123 }
124
125 pub fn packetCount(self: *const Fabric) u8 {
126 return self.state.packet_count;
127 }
128
129 pub fn packet(self: *const Fabric, id: u64) Error!types.Packet {
130 try canon.validate(self.state);
131 const index = packetIndex(self.state, id) orelse return error.PacketUnknown;
132 return self.state.packets[index].packet;
133 }
134
135 pub fn nodeAvailable(self: *const Fabric, node: types.NodeId) Error!bool {
136 try canon.validate(self.state);
137 const index = nodeIndex(self.state, node) orelse return error.UnknownNode;
138 return self.state.nodes[index].available;
139 }
140
141 pub fn partitioned(
142 self: *const Fabric,
143 first: types.NodeId,
144 second: types.NodeId,
145 ) Error!bool {
146 try canon.validate(self.state);
147 return linkPartitioned(self.state, first, second);
148 }
149
150 pub fn applyFault(
151 self: *Fabric,
152 input: types.FaultInput,
153 ) Error!types.Faulted {
154 try canon.validate(self.state);
155 if (self.state.pending != null) return error.TurnPending;
156 if (self.state.admission_frontier == self.state.admission_limit) {
157 return error.AdmissionCapacityExceeded;
158 }
159 try validateDirectFault(self.state, input.effect);
160 const point = try canon.faultPoint(self.state.root, input.effect);
161 const decision = try fault.decide(point, input.choice);
162 const entry = try nextEntry(self.state, .{ .fault = .{
163 .effect = input.effect,
164 .decision = decision,
165 } });
166 var candidate = self.*;
167 try applyEntry(&candidate.state, entry);
168 self.* = candidate;
169 return .{
170 .entry = entry,
171 .root = self.state.root,
172 .decision = decision,
173 };
174 }
175
176 pub fn terminal(
177 self: *Fabric,
178 node: types.NodeId,
179 machine: *const instance.Instance,
180 fence: os.abi.ActivationFence,
181 bytes: []const u8,
182 ) Error!types.Admitted {
183 const basis = try self.liveBasis(node, machine);
184 const record = try admission.terminal(
185 basis.frontiers.terminal_input_offset,
186 bytes,
187 );
188 return self.admit(node, basis, fence, record, .direct);
189 }
190
191 pub fn advanceTime(
192 self: *Fabric,
193 node: types.NodeId,
194 machine: *const instance.Instance,
195 fence: os.abi.ActivationFence,
196 to_tick: u64,
197 ) Error!types.Admitted {
198 const basis = try self.liveBasis(node, machine);
199 if (to_tick < self.state.virtual_time_tick) {
200 return error.VirtualTimeRegression;
201 }
202 const record = try admission.virtualTime(
203 basis.frontiers.virtual_time_tick,
204 to_tick,
205 );
206 return self.admit(node, basis, fence, record, .direct);
207 }
208
209 pub fn injectEntropy(
210 self: *Fabric,
211 node: types.NodeId,
212 machine: *const instance.Instance,
213 fence: os.abi.ActivationFence,
214 bytes: []const u8,
215 ) Error!types.Admitted {
216 const basis = try self.liveBasis(node, machine);
217 const generation = std.math.add(
218 u64,
219 basis.frontiers.entropy_generation,
220 1,
221 ) catch return error.EntropyCapacityExceeded;
222 const record = try admission.entropy(generation, bytes);
223 return self.admit(node, basis, fence, record, .direct);
224 }
225
226 pub fn chooseEntropy(
227 self: *Fabric,
228 node: types.NodeId,
229 machine: *const instance.Instance,
230 fence: os.abi.ActivationFence,
231 bypass_bytes: []const u8,
232 inject_bytes: []const u8,
233 choice: fault.Choice,
234 ) Error!types.Admitted {
235 const basis = try self.liveBasis(node, machine);
236 const generation = std.math.add(
237 u64,
238 basis.frontiers.entropy_generation,
239 1,
240 ) catch return error.EntropyCapacityExceeded;
241 const bypass_record = try admission.entropy(generation, bypass_bytes);
242 const inject_record = try admission.entropy(generation, inject_bytes);
243 return self.admitAlternatives(
244 node,
245 basis,
246 fence,
247 bypass_record,
248 inject_record,
249 .entropy_choice,
250 choice,
251 );
252 }
253
254 pub fn settleService(
255 self: *Fabric,
256 node: types.NodeId,
257 machine: *const instance.Instance,
258 fence: os.abi.ActivationFence,
259 result: types.ServiceResult,
260 ) Error!types.Admitted {
261 const basis = try self.liveBasis(node, machine);
262 const request = basis.outstanding_effect orelse
263 return error.EffectRequestRequired;
264 const record = try admission.effectResult(
265 request.receipt.digest,
266 request.correlation,
267 result.status,
268 result.output_root,
269 result.bytes,
270 );
271 return self.admit(node, basis, fence, record, .direct);
272 }
273
274 pub fn chooseService(
275 self: *Fabric,
276 node: types.NodeId,
277 machine: *const instance.Instance,
278 fence: os.abi.ActivationFence,
279 kind: fault.Kind,
280 bypass: types.ServiceResult,
281 injected: types.ServiceResult,
282 choice: fault.Choice,
283 ) Error!types.Admitted {
284 const basis = try self.liveBasis(node, machine);
285 const request = basis.outstanding_effect orelse
286 return error.EffectRequestRequired;
287 try validateServiceFault(kind, injected.status);
288 const bypass_record = try serviceRecord(request, bypass);
289 const inject_record = try serviceRecord(request, injected);
290 return self.admitAlternatives(
291 node,
292 basis,
293 fence,
294 bypass_record,
295 inject_record,
296 kind,
297 choice,
298 );
299 }
300
301 pub fn sendPacket(
302 self: *Fabric,
303 node: types.NodeId,
304 machine: *const instance.Instance,
305 fence: os.abi.ActivationFence,
306 input: types.PacketInput,
307 ) Error!types.Admitted {
308 const basis = try self.liveBasis(node, machine);
309 const request = basis.outstanding_effect orelse
310 return error.EffectRequestRequired;
311 if (self.state.packet_count == types.packet_limit or
312 capacityActive(self.state, .packet))
313 {
314 return error.PacketCapacityExceeded;
315 }
316 _ = try availableNodeIndex(self.state, input.destination);
317 if (try linkPartitioned(self.state, node, input.destination)) {
318 return error.LinkPartitioned;
319 }
320 const packet_id = std.math.add(
321 u64,
322 self.state.packet_next_id,
323 1,
324 ) catch return error.PacketIdExhausted;
325 const packet_value = try canon.makePacket(packet_id, node, request, input);
326 const record = try admission.effectResult(
327 request.receipt.digest,
328 request.correlation,
329 .ok,
330 packet_value.digest,
331 "",
332 );
333 return self.admit(
334 node,
335 basis,
336 fence,
337 record,
338 .{ .packet_send = packet_value },
339 );
340 }
341
342 pub fn receivePacket(
343 self: *Fabric,
344 node: types.NodeId,
345 machine: *const instance.Instance,
346 fence: os.abi.ActivationFence,
347 packet_id: u64,
348 ) Error!types.Admitted {
349 const basis = try self.liveBasis(node, machine);
350 const request = basis.outstanding_effect orelse
351 return error.EffectRequestRequired;
352 const index = packetIndex(self.state, packet_id) orelse
353 return error.PacketUnknown;
354 const queued = self.state.packets[index];
355 if (!queued.ready) return error.PacketNotReady;
356 if (queued.ready_at_tick > self.state.virtual_time_tick) {
357 return error.PacketDelayed;
358 }
359 if (!canon.sameNode(queued.packet.destination, node)) {
360 return error.PacketDestinationMismatch;
361 }
362 if (try linkPartitioned(self.state, queued.packet.source, node)) {
363 return error.LinkPartitioned;
364 }
365 const record = try admission.effectResult(
366 request.receipt.digest,
367 request.correlation,
368 .ok,
369 queued.packet.digest,
370 queued.packet.bytes(),
371 );
372 return self.admit(
373 node,
374 basis,
375 fence,
376 record,
377 .{ .packet_delivery = packet_id },
378 );
379 }
380
381 pub fn receiveNextPacket(
382 self: *Fabric,
383 node: types.NodeId,
384 machine: *const instance.Instance,
385 fence: os.abi.ActivationFence,
386 ) Error!types.Admitted {
387 try canon.validate(self.state);
388 const packet_id = (try nextPacketId(self.state, node)) orelse
389 return error.PacketUnknown;
390 return self.receivePacket(node, machine, fence, packet_id);
391 }
392
393 pub fn pendingDelivery(
394 self: *const Fabric,
395 node: types.NodeId,
396 fence: os.abi.ActivationFence,
397 ) Error!admission.Delivery {
398 try canon.validate(self.state);
399 const pending = self.state.pending orelse return error.NoTurnPending;
400 if (!canon.sameNode(pending.node, node)) return error.UnknownNode;
401 try validateFence(self.state.world, fence);
402 return admission.bindDelivery(
403 pending.admission,
404 pending.delivery_root,
405 fence,
406 );
407 }
408
409 pub fn settle(
410 self: *Fabric,
411 node: types.NodeId,
412 receipt: instance.QuiescenceReceipt,
413 ) Error!types.Settled {
414 try canon.validate(self.state);
415 const pending = self.state.pending orelse return error.NoTurnPending;
416 if (!canon.sameNode(pending.node, node)) return error.UnknownNode;
417 try instance.verifyQuiescenceReceipt(receipt, receipt.fence);
418 try validateFence(self.state.world, receipt.fence);
419 const delivery = try admission.bindDelivery(
420 pending.admission,
421 pending.delivery_root,
422 receipt.fence,
423 );
424 if (!std.meta.eql(delivery.receipt, receipt.delivery)) {
425 return error.ReceiptMismatch;
426 }
427 const semantic = try instance.projectSemanticReceipt(receipt);
428 const entry = try nextEntry(self.state, .{ .settlement = .{
429 .node = node,
430 .receipt = semantic,
431 } });
432 var candidate = self.*;
433 try applyEntry(&candidate.state, entry);
434 self.* = candidate;
435 return .{ .entry = entry, .root = self.state.root };
436 }
437
438 pub fn replay(self: *Fabric, entry: types.Entry) Error!types.Root {
439 var candidate = self.*;
440 try applyEntry(&candidate.state, entry);
441 self.* = candidate;
442 return self.state.root;
443 }
444
445 fn liveBasis(
446 self: *const Fabric,
447 node: types.NodeId,
448 machine: *const instance.Instance,
449 ) Error!admission.Basis {
450 try canon.validate(self.state);
451 if (self.state.pending != null) return error.TurnPending;
452 const index = nodeIndex(self.state, node) orelse return error.UnknownNode;
453 if (!self.state.nodes[index].available) return error.NodeUnavailable;
454 if (capacityActive(self.state, .admission)) {
455 return error.AdmissionCapacityExceeded;
456 }
457 const basis = try machine.admissionBasis();
458 if (!std.meta.eql(basis, self.state.nodes[index].basis)) {
459 return error.MachineBasisMismatch;
460 }
461 return basis;
462 }
463
464 fn admit(
465 self: *Fabric,
466 node: types.NodeId,
467 basis: admission.Basis,
468 fence: os.abi.ActivationFence,
469 record: admission.Record,
470 effect: types.AdmissionEffect,
471 ) Error!types.Admitted {
472 const prepared = try admission.prepare(basis, record);
473 return self.admitPrepared(node, fence, prepared, effect, null);
474 }
475
476 fn admitAlternatives(
477 self: *Fabric,
478 node: types.NodeId,
479 basis: admission.Basis,
480 fence: os.abi.ActivationFence,
481 bypass_record: admission.Record,
482 inject_record: admission.Record,
483 kind: fault.Kind,
484 choice: fault.Choice,
485 ) Error!types.Admitted {
486 const bypassed = try admission.prepare(basis, bypass_record);
487 const injected = try admission.prepare(basis, inject_record);
488 if (std.meta.eql(bypassed.receipt, injected.receipt)) {
489 return error.FaultAlternativesEqual;
490 }
491 const alternatives: types.FaultAlternatives = .{
492 .node = node,
493 .bypass = bypassed.receipt.digest,
494 .inject = injected.receipt.digest,
495 };
496 const effect: types.FaultEffect = switch (kind) {
497 .entropy_choice => .{ .entropy_choice = alternatives },
498 .io_error => .{ .io_error = alternatives },
499 .host_service_failure => .{ .host_service_failure = alternatives },
500 else => return error.FaultEffectMismatch,
501 };
502 const point = try canon.faultPoint(self.state.root, effect);
503 const decision = try fault.decide(point, choice);
504 const selected = switch (choice) {
505 .bypass => bypassed,
506 .inject => injected,
507 };
508 return self.admitPrepared(node, fence, selected, .direct, .{
509 .effect = effect,
510 .decision = decision,
511 });
512 }
513
514 fn admitPrepared(
515 self: *Fabric,
516 node: types.NodeId,
517 fence: os.abi.ActivationFence,
518 prepared: admission.Admission,
519 effect: types.AdmissionEffect,
520 admission_fault: ?types.AdmissionFault,
521 ) Error!types.Admitted {
522 try validateFence(self.state.world, fence);
523 const entry = try nextEntry(self.state, .{ .admission = .{
524 .node = node,
525 .admission = prepared,
526 .effect = effect,
527 .fault = admission_fault,
528 } });
529 var candidate = self.*;
530 try applyEntry(&candidate.state, entry);
531 const delivery = try candidate.pendingDelivery(node, fence);
532 self.* = candidate;
533 return .{
534 .entry = entry,
535 .root = self.state.root,
536 .delivery = delivery,
537 };
538 }
539 };
540
541 fn applyEntry(state: *types.State, entry: types.Entry) Error!void {
542 try canon.validate(state.*);
543 if (!std.meta.eql(entry.previous, state.root)) return error.RootMismatch;
544 const expected = std.math.add(u64, state.entry_frontier, 1) catch
545 return error.EntrySequenceExhausted;
546 if (entry.sequence != expected) return error.EntrySequenceMismatch;
547 switch (entry.value) {
548 .admission => |value| try applyAdmission(state, entry, value),
549 .settlement => |value| try applySettlement(state, entry, value),
550 .fault => |value| try applyFaultEntry(state, entry, value),
551 }
552 try canon.validate(state.*);
553 }
554
555 fn applyAdmission(
556 state: *types.State,
557 entry: types.Entry,
558 value: types.AdmissionEntry,
559 ) Error!void {
560 if (state.pending != null) return error.TurnPending;
561 if (state.admission_frontier == state.admission_limit) {
562 return error.AdmissionCapacityExceeded;
563 }
564 if (capacityActive(state.*, .admission)) {
565 return error.AdmissionCapacityExceeded;
566 }
567 const index = nodeIndex(state.*, value.node) orelse return error.UnknownNode;
568 if (!state.nodes[index].available) return error.NodeUnavailable;
569 try admission.verify(state.nodes[index].basis, &value.admission);
570 if (value.fault) |admission_fault| {
571 try applyAdmissionFault(state, entry.previous, value, admission_fault);
572 }
573 const pending_effect = try applyAdmissionEffect(state, value);
574 try applyInputAccounting(state, value.admission.record);
575 state.pending = .{
576 .node = value.node,
577 .admission = value.admission,
578 .effect = pending_effect,
579 .delivery_root = @splat(0),
580 };
581 state.entry_frontier = entry.sequence;
582 state.admission_frontier += 1;
583 state.ledger_digest = try canon.advanceLedger(state.ledger_digest, entry);
584 state.root = canon.rootFor(state.*);
585 state.pending.?.delivery_root = state.root.digest;
586 }
587
588 fn applyFaultEntry(
589 state: *types.State,
590 entry: types.Entry,
591 value: types.FaultEntry,
592 ) Error!void {
593 if (state.pending != null) return error.TurnPending;
594 if (state.admission_frontier == state.admission_limit) {
595 return error.AdmissionCapacityExceeded;
596 }
597 try validateDirectFault(state.*, value.effect);
598 try canon.verifyFaultDecision(entry.previous, value.effect, value.decision);
599 if (value.decision.choice == .inject) {
600 try applyDirectFault(state, value.effect);
601 }
602 try commitFaultDecision(state, value.decision);
603 state.entry_frontier = entry.sequence;
604 state.admission_frontier += 1;
605 state.ledger_digest = try canon.advanceLedger(state.ledger_digest, entry);
606 state.root = canon.rootFor(state.*);
607 }
608
609 fn applyAdmissionFault(
610 state: *types.State,
611 previous: types.Root,
612 value: types.AdmissionEntry,
613 admission_fault: types.AdmissionFault,
614 ) Error!void {
615 if (std.meta.activeTag(value.effect) != .direct) {
616 return error.FaultEffectMismatch;
617 }
618 try canon.verifyFaultDecision(
619 previous,
620 admission_fault.effect,
621 admission_fault.decision,
622 );
623 const alternatives = try admissionAlternatives(value, admission_fault.effect);
624 const selected = switch (admission_fault.decision.choice) {
625 .bypass => alternatives.bypass,
626 .inject => alternatives.inject,
627 };
628 if (!std.mem.eql(u8, &selected, &value.admission.receipt.digest)) {
629 return error.FaultEffectMismatch;
630 }
631 try validateAdmissionFaultRecord(value, admission_fault);
632 try commitFaultDecision(state, admission_fault.decision);
633 }
634
635 fn validateDirectFault(state: types.State, effect: types.FaultEffect) Error!void {
636 switch (effect) {
637 .machine_crash => |node| _ = try availableNodeIndex(state, node),
638 .process_crash => |value| {
639 _ = try availableNodeIndex(state, value.node);
640 if (value.process == 0) return error.ProcessIdentityInvalid;
641 },
642 .io_error, .entropy_choice, .host_service_failure => return error.FaultRequiresAdmission,
643 .packet_loss => |packet_id| {
644 const index = packetIndex(state, packet_id) orelse
645 return error.PacketUnknown;
646 if (!state.packets[index].ready) return error.PacketNotReady;
647 },
648 .packet_delay => |value| try validatePacketDelay(state, value),
649 .packet_reorder => |value| try validatePacketReorder(state, value),
650 .partition => |value| try validatePartition(state, value),
651 .clock_jump => |to_tick| {
652 if (to_tick <= state.virtual_time_tick) {
653 return error.VirtualTimeRegression;
654 }
655 },
656 .capacity_exhaustion => |value| {
657 if (capacityActive(state, value.resource) == value.active) {
658 return error.CapacityStateMismatch;
659 }
660 },
661 }
662 }
663
664 fn applyDirectFault(state: *types.State, effect: types.FaultEffect) Error!void {
665 switch (effect) {
666 .machine_crash => |node| {
667 const index = try availableNodeIndex(state.*, node);
668 state.nodes[index].available = false;
669 },
670 .process_crash => {},
671 .io_error, .entropy_choice, .host_service_failure => return error.FaultRequiresAdmission,
672 .packet_loss => |packet_id| {
673 const index = packetIndex(state.*, packet_id) orelse
674 return error.PacketUnknown;
675 removePacket(state, index);
676 },
677 .packet_delay => |value| {
678 const index = packetIndex(state.*, value.packet).?;
679 state.packets[index].ready_at_tick = value.until_tick;
680 },
681 .packet_reorder => |value| {
682 const first = packetIndex(state.*, value.first).?;
683 const second = packetIndex(state.*, value.second).?;
684 const order = state.packets[first].order;
685 state.packets[first].order = state.packets[second].order;
686 state.packets[second].order = order;
687 },
688 .partition => |value| try setPartition(
689 state,
690 value.first,
691 value.second,
692 value.active,
693 ),
694 .clock_jump => |to_tick| state.virtual_time_tick = to_tick,
695 .capacity_exhaustion => |value| setCapacity(
696 state,
697 value.resource,
698 value.active,
699 ),
700 }
701 }
702
703 fn admissionAlternatives(
704 value: types.AdmissionEntry,
705 effect: types.FaultEffect,
706 ) Error!types.FaultAlternatives {
707 const alternatives = switch (effect) {
708 .entropy_choice => |selected| selected,
709 .io_error => |selected| selected,
710 .host_service_failure => |selected| selected,
711 else => return error.FaultEffectMismatch,
712 };
713 if (!canon.sameNode(alternatives.node, value.node)) {
714 return error.FaultEffectMismatch;
715 }
716 if (std.mem.eql(u8, &alternatives.bypass, &alternatives.inject)) {
717 return error.FaultAlternativesEqual;
718 }
719 os.abi.wire.validateDigest(alternatives.bypass) catch
720 return error.FaultEffectMismatch;
721 os.abi.wire.validateDigest(alternatives.inject) catch
722 return error.FaultEffectMismatch;
723 return alternatives;
724 }
725
726 fn validateAdmissionFaultRecord(
727 value: types.AdmissionEntry,
728 admission_fault: types.AdmissionFault,
729 ) Error!void {
730 switch (admission_fault.effect) {
731 .entropy_choice => {
732 if (std.meta.activeTag(value.admission.record) != .entropy) {
733 return error.FaultEffectMismatch;
734 }
735 },
736 .io_error => try expectFaultStatus(
737 value.admission.record,
738 admission_fault.decision.choice,
739 .failed,
740 ),
741 .host_service_failure => try expectFaultStatus(
742 value.admission.record,
743 admission_fault.decision.choice,
744 .unavailable,
745 ),
746 else => return error.FaultEffectMismatch,
747 }
748 }
749
750 fn expectFaultStatus(
751 record: admission.Record,
752 choice: fault.Choice,
753 injected_status: os.abi.EffectStatus,
754 ) Error!void {
755 if (std.meta.activeTag(record) != .effect_result) {
756 return error.FaultEffectMismatch;
757 }
758 if (choice == .inject and record.effect_result.status != injected_status) {
759 return error.FaultStatusInvalid;
760 }
761 }
762
763 fn commitFaultDecision(
764 state: *types.State,
765 decision: fault.Decision,
766 ) Error!void {
767 state.fault_frontier = std.math.add(
768 u64,
769 state.fault_frontier,
770 1,
771 ) catch return error.FaultFrontierExhausted;
772 state.fault_digest = try canon.advanceFault(state.fault_digest, decision);
773 }
774
775 fn applySettlement(
776 state: *types.State,
777 entry: types.Entry,
778 value: types.SettlementEntry,
779 ) Error!void {
780 const pending = state.pending orelse return error.NoTurnPending;
781 if (!canon.sameNode(pending.node, value.node)) return error.UnknownNode;
782 try instance.verifySemanticReceipt(value.receipt);
783 if (!std.meta.eql(value.receipt.admission_receipt, pending.admission.receipt) or
784 !std.meta.eql(value.receipt.basis.contract, state.contract) or
785 !std.mem.eql(
786 u8,
787 &value.receipt.basis.source_root,
788 &pending.delivery_root,
789 ) or
790 !std.meta.eql(
791 value.receipt.basis.frontiers,
792 pending.admission.expected,
793 ) or
794 !std.meta.eql(
795 value.receipt.basis.outstanding_effect,
796 pending.admission.expected_outstanding_effect,
797 ))
798 {
799 return error.ReceiptMismatch;
800 }
801 const index = nodeIndex(state.*, value.node) orelse return error.UnknownNode;
802 state.nodes[index].basis = value.receipt.basis;
803 state.nodes[index].machine = .{
804 .kind = .semantic,
805 .digest = try instance.semanticReceiptDigest(value.receipt),
806 };
807 switch (pending.effect) {
808 .direct, .packet_delivery => {},
809 .packet_send => |packet_id| try markPacketReady(state, packet_id),
810 }
811 state.pending = null;
812 state.entry_frontier = entry.sequence;
813 state.ledger_digest = try canon.advanceLedger(state.ledger_digest, entry);
814 state.root = canon.rootFor(state.*);
815 }
816
817 fn applyAdmissionEffect(
818 state: *types.State,
819 value: types.AdmissionEntry,
820 ) Error!types.PendingEffect {
821 return switch (value.effect) {
822 .direct => .direct,
823 .packet_send => |packet_value| try stagePacket(state, value, packet_value),
824 .packet_delivery => |packet_id| try reservePacket(state, value, packet_id),
825 };
826 }
827
828 fn stagePacket(
829 state: *types.State,
830 value: types.AdmissionEntry,
831 packet_value: types.Packet,
832 ) Error!types.PendingEffect {
833 if (state.packet_count == types.packet_limit or
834 capacityActive(state.*, .packet))
835 {
836 return error.PacketCapacityExceeded;
837 }
838 const node = state.nodes[nodeIndex(state.*, value.node).?];
839 const request = node.basis.outstanding_effect orelse
840 return error.EffectRequestRequired;
841 const expected_id = std.math.add(u64, state.packet_next_id, 1) catch
842 return error.PacketIdExhausted;
843 const destination_index = nodeIndex(state.*, packet_value.destination) orelse
844 return error.UnknownNode;
845 if (!state.nodes[destination_index].available) return error.NodeUnavailable;
846 if (try linkPartitioned(state.*, value.node, packet_value.destination)) {
847 return error.LinkPartitioned;
848 }
849 if (packet_value.id != expected_id or
850 !canon.sameNode(packet_value.source, value.node) or
851 !std.meta.eql(packet_value.request, request) or
852 !std.mem.eql(
853 u8,
854 &packet_value.digest,
855 &canon.packetDigest(packet_value),
856 ))
857 {
858 return error.InvalidPacket;
859 }
860 try expectPacketResult(value.admission.record, packet_value, false);
861 state.packets[state.packet_count] = .{
862 .packet = packet_value,
863 .ready = false,
864 .ready_at_tick = 0,
865 .order = packet_value.id,
866 };
867 state.packet_count += 1;
868 state.packet_next_id = packet_value.id;
869 return .{ .packet_send = packet_value.id };
870 }
871
872 fn reservePacket(
873 state: *types.State,
874 value: types.AdmissionEntry,
875 packet_id: u64,
876 ) Error!types.PendingEffect {
877 const index = packetIndex(state.*, packet_id) orelse return error.PacketUnknown;
878 const queued = state.packets[index];
879 if (!queued.ready) return error.PacketNotReady;
880 if (queued.ready_at_tick > state.virtual_time_tick) return error.PacketDelayed;
881 if (!canon.sameNode(queued.packet.destination, value.node)) {
882 return error.PacketDestinationMismatch;
883 }
884 if (try linkPartitioned(state.*, queued.packet.source, value.node)) {
885 return error.LinkPartitioned;
886 }
887 try expectPacketResult(value.admission.record, queued.packet, true);
888 removePacket(state, index);
889 return .{ .packet_delivery = packet_id };
890 }
891
892 fn expectPacketResult(
893 record: admission.Record,
894 packet_value: types.Packet,
895 payload: bool,
896 ) Error!void {
897 if (std.meta.activeTag(record) != .effect_result) {
898 return error.EntryTypeMismatch;
899 }
900 const result = record.effect_result;
901 const expected_bytes = if (payload) packet_value.bytes() else "";
902 if ((!payload and !std.meta.eql(result.request, packet_value.request)) or
903 result.status != .ok or
904 !std.mem.eql(u8, &result.output_root, &packet_value.digest) or
905 !std.mem.eql(u8, result.storage[0..result.length], expected_bytes))
906 {
907 return error.InvalidPacket;
908 }
909 }
910
911 fn serviceRecord(
912 request: admission.EffectRequest,
913 result: types.ServiceResult,
914 ) admission.Error!admission.Record {
915 return admission.effectResult(
916 request.receipt.digest,
917 request.correlation,
918 result.status,
919 result.output_root,
920 result.bytes,
921 );
922 }
923
924 fn validateServiceFault(
925 kind: fault.Kind,
926 status: os.abi.EffectStatus,
927 ) Error!void {
928 switch (kind) {
929 .io_error => if (status != .failed) return error.FaultStatusInvalid,
930 .host_service_failure => if (status != .unavailable) {
931 return error.FaultStatusInvalid;
932 },
933 else => return error.FaultEffectMismatch,
934 }
935 }
936
937 fn applyInputAccounting(state: *types.State, record: admission.Record) Error!void {
938 switch (record) {
939 .virtual_time => |value| {
940 if (value.to_tick < state.virtual_time_tick) {
941 return error.VirtualTimeRegression;
942 }
943 state.virtual_time_tick = value.to_tick;
944 },
945 .entropy => |value| {
946 state.entropy_frontier = std.math.add(
947 u64,
948 state.entropy_frontier,
949 1,
950 ) catch return error.EntropyCapacityExceeded;
951 state.entropy_bytes = std.math.add(
952 u64,
953 state.entropy_bytes,
954 value.length,
955 ) catch return error.EntropyCapacityExceeded;
956 },
957 .terminal, .effect_result => {},
958 }
959 }
960
961 fn markPacketReady(state: *types.State, packet_id: u64) Error!void {
962 const index = packetIndex(state.*, packet_id) orelse return error.PacketUnknown;
963 if (state.packets[index].ready) return error.InvalidPacket;
964 state.packets[index].ready = true;
965 state.packets[index].ready_at_tick = state.virtual_time_tick;
966 }
967
968 fn removePacket(state: *types.State, index: usize) void {
969 std.debug.assert(index < state.packet_count);
970 var cursor = index;
971 while (cursor + 1 < state.packet_count) : (cursor += 1) {
972 state.packets[cursor] = state.packets[cursor + 1];
973 }
974 state.packet_count -= 1;
975 }
976
977 fn validateFence(world: [16]u8, fence: os.abi.ActivationFence) Error!void {
978 os.abi.wire.validateFence(fence) catch return error.WorldMismatch;
979 if (!std.mem.eql(u8, &world, &fence.world)) return error.WorldMismatch;
980 }
981
982 fn nextEntry(state: types.State, value: types.EntryValue) Error!types.Entry {
983 const sequence = std.math.add(u64, state.entry_frontier, 1) catch
984 return error.EntrySequenceExhausted;
985 return .{ .sequence = sequence, .previous = state.root, .value = value };
986 }
987
988 fn nodeIndex(state: types.State, id: types.NodeId) ?usize {
989 for (state.nodes[0..state.node_count], 0..) |node, index| {
990 if (canon.sameNode(node.id, id)) return index;
991 }
992 return null;
993 }
994
995 fn packetIndex(state: types.State, id: u64) ?usize {
996 for (state.packets[0..state.packet_count], 0..) |packet_value, index| {
997 if (packet_value.packet.id == id) return index;
998 }
999 return null;
1000 }
1001
1002 fn availableNodeIndex(state: types.State, node: types.NodeId) Error!usize {
1003 const index = nodeIndex(state, node) orelse return error.UnknownNode;
1004 if (!state.nodes[index].available) return error.NodeUnavailable;
1005 return index;
1006 }
1007
1008 fn validatePacketDelay(state: types.State, value: types.PacketDelay) Error!void {
1009 const index = packetIndex(state, value.packet) orelse return error.PacketUnknown;
1010 const queued = state.packets[index];
1011 if (!queued.ready) return error.PacketNotReady;
1012 if (value.until_tick <= state.virtual_time_tick or
1013 value.until_tick == queued.ready_at_tick)
1014 {
1015 return error.PacketDelayed;
1016 }
1017 }
1018
1019 fn validatePacketReorder(
1020 state: types.State,
1021 value: types.PacketReorder,
1022 ) Error!void {
1023 if (value.first >= value.second) return error.PacketOrderInvalid;
1024 const first = packetIndex(state, value.first) orelse return error.PacketUnknown;
1025 const second = packetIndex(state, value.second) orelse return error.PacketUnknown;
1026 const first_packet = state.packets[first];
1027 const second_packet = state.packets[second];
1028 if (!first_packet.ready or !second_packet.ready or
1029 first_packet.ready_at_tick > state.virtual_time_tick or
1030 second_packet.ready_at_tick > state.virtual_time_tick or
1031 !canon.sameNode(
1032 first_packet.packet.destination,
1033 second_packet.packet.destination,
1034 ))
1035 {
1036 return error.PacketOrderInvalid;
1037 }
1038 }
1039
1040 fn validatePartition(state: types.State, value: types.Partition) Error!void {
1041 if (!canon.lessNode(value.first, value.second)) {
1042 return error.InvalidNode;
1043 }
1044 if (try linkPartitioned(state, value.first, value.second) == value.active) {
1045 return error.PartitionStateMismatch;
1046 }
1047 }
1048
1049 fn nextPacketId(state: types.State, node: types.NodeId) Error!?u64 {
1050 _ = try availableNodeIndex(state, node);
1051 var selected: ?types.QueuedPacket = null;
1052 for (state.packets[0..state.packet_count]) |queued| {
1053 if (!queued.ready or queued.ready_at_tick > state.virtual_time_tick or
1054 !canon.sameNode(queued.packet.destination, node) or
1055 try linkPartitioned(state, queued.packet.source, node))
1056 {
1057 continue;
1058 }
1059 if (selected == null or queued.order < selected.?.order) selected = queued;
1060 }
1061 return if (selected) |queued| queued.packet.id else null;
1062 }
1063
1064 fn linkPartitioned(
1065 state: types.State,
1066 first: types.NodeId,
1067 second: types.NodeId,
1068 ) Error!bool {
1069 const bit = try linkBit(state, first, second);
1070 return state.partition_bits & bit != 0;
1071 }
1072
1073 fn setPartition(
1074 state: *types.State,
1075 first: types.NodeId,
1076 second: types.NodeId,
1077 active: bool,
1078 ) Error!void {
1079 const bit = try linkBit(state.*, first, second);
1080 if (active) {
1081 state.partition_bits |= bit;
1082 } else {
1083 state.partition_bits &= ~bit;
1084 }
1085 }
1086
1087 fn linkBit(
1088 state: types.State,
1089 first: types.NodeId,
1090 second: types.NodeId,
1091 ) Error!u8 {
1092 const first_index = nodeIndex(state, first) orelse return error.UnknownNode;
1093 const second_index = nodeIndex(state, second) orelse return error.UnknownNode;
1094 if (first_index == second_index) return error.InvalidNode;
1095 const low = @min(first_index, second_index);
1096 const high = @max(first_index, second_index);
1097 var bit_index: usize = 0;
1098 for (0..state.node_count) |left| {
1099 for (left + 1..state.node_count) |right| {
1100 if (left == low and right == high) {
1101 return @as(u8, 1) << @intCast(bit_index);
1102 }
1103 bit_index += 1;
1104 }
1105 }
1106 unreachable;
1107 }
1108
1109 fn capacityActive(state: types.State, resource: types.CapacityResource) bool {
1110 return state.capacity_bits & capacityBit(resource) != 0;
1111 }
1112
1113 fn setCapacity(
1114 state: *types.State,
1115 resource: types.CapacityResource,
1116 active: bool,
1117 ) void {
1118 const bit = capacityBit(resource);
1119 if (active) {
1120 state.capacity_bits |= bit;
1121 } else {
1122 state.capacity_bits &= ~bit;
1123 }
1124 }
1125
1126 fn capacityBit(resource: types.CapacityResource) u8 {
1127 return switch (resource) {
1128 .admission => 0b01,
1129 .packet => 0b10,
1130 };
1131 }
1132
1133 fn emptyNodeBoundary() types.NodeBoundary {
1134 return .{
1135 .id = .{ .bytes = @splat(0) },
1136 .available = false,
1137 .machine = .{ .kind = .basis, .digest = @splat(0) },
1138 };
1139 }