lib/machine/src/world/fault.zig
daab053ee43316e1809a84551d573ddd1e5bf3d2
1 const fabric = @import("../fabric/root.zig");
2 const fault = @import("../fault/root.zig");
3 const moment = @import("moment.zig");
4 const restore = @import("restore.zig");
5 const source = @import("source.zig");
6 const std = @import("std");
7
8 const FaultOwnerError = error{
9 CapabilitySetMismatch,
10 FaultTransitionMismatch,
11 FaultTransitionExpected,
12 MachineNodeUnavailable,
13 ProcessFaultProviderUnavailable,
14 };
15
16 pub const Error = fabric.Error ||
17 fabric.transition.Error ||
18 restore.StableError ||
19 source.Error ||
20 FaultOwnerError;
21
22 /// The record left behind by an applied fault: the ledger transition holding its
23 /// record, the decision that was taken, and the moment the world reaches afterwards.
24 /// A caller reads this to see what the fault did to the ledger and where the world
25 /// now stands.
26 pub const Applied = struct {
27 transition: fabric.Transition,
28 decision: fault.Decision,
29 moment: moment.Moment,
30 };
31
32 /// A named failure carrying the node involved whenever the code can name one. A
33 /// caller reads this to learn which node the failure concerns.
34 pub const Rejection = struct {
35 node: ?fabric.NodeId,
36 failure: Error,
37 };
38
39 /// The fault outcome. A world that rejected the call still holds the state it had
40 /// before and stays usable.
41 pub const Result = union(enum) {
42 applied: Applied,
43 rejected: Rejection,
44 };
45
46 const CommitPlan = struct {
47 crash_index: ?usize,
48 decision: fault.Decision,
49 next_moment: moment.Moment,
50 };
51
52 /// Applies one fault input to a candidate ledger, then commits that ledger and the
53 /// moment together. Injecting a machine crash destroys the live instance on that
54 /// node and marks the node unavailable. A process fault requires a provider that
55 /// this owner never holds, so the call rejects it. The candidate's node set must
56 /// match the restored world's, node for node, before it commits.
57 pub fn applyFault(
58 restored: *restore.Restored,
59 input: fabric.FaultInput,
60 ) Result {
61 validateCapabilities(restored) catch |failure|
62 return reject(null, failure);
63 if (std.meta.activeTag(input.effect) == .process_crash) {
64 return reject(
65 input.effect.process_crash.node,
66 error.ProcessFaultProviderUnavailable,
67 );
68 }
69 var candidate = restored.fabric.*;
70 const faulted = candidate.applyFault(input) catch |failure|
71 return reject(faultNode(input.effect), failure);
72 return commit(
73 restored,
74 candidate,
75 .{ .entry = faulted.entry, .root = faulted.root },
76 );
77 }
78
79 /// Reads one recorded fault transition back from its wire bytes so a caller proves
80 /// the recording reproduces the same ledger position, and demands that the replay
81 /// land on the root the record declares. The reader is checked against the world's
82 /// live memory before anything is read. The call accepts a fault entry alone, and
83 /// it rejects a recorded process crash. The call commits exactly like `applyFault`.
84 pub fn replayFault(
85 restored: *restore.Restored,
86 reader: *std.Io.Reader,
87 expected_root: fabric.Root,
88 ) Result {
89 source.validate(restored, &.{reader}) catch |failure|
90 return reject(null, failure);
91 validateCapabilities(restored) catch |failure|
92 return reject(null, failure);
93 var candidate = restored.fabric.*;
94 const transition = fabric.transition.replayDisjoint(
95 &candidate,
96 reader,
97 expected_root,
98 ) catch |failure| return reject(null, failure);
99 const entry = switch (transition.entry.value) {
100 .fault => |value| value,
101 else => return reject(null, error.FaultTransitionExpected),
102 };
103 if (std.meta.activeTag(entry.effect) == .process_crash) {
104 return reject(
105 entry.effect.process_crash.node,
106 error.ProcessFaultProviderUnavailable,
107 );
108 }
109 return commit(restored, candidate, transition);
110 }
111
112 fn commit(
113 restored: *restore.Restored,
114 candidate: fabric.Fabric,
115 transition: fabric.Transition,
116 ) Result {
117 const node = transitionNode(transition);
118 const plan = prepareCommit(restored, &candidate, transition) catch |failure|
119 return reject(node, failure);
120 if (plan.crash_index) |index| {
121 const machine = if (restored.nodes[index].machine) |*value|
122 value
123 else
124 unreachable;
125 machine.deinit();
126 restored.nodes[index].machine = null;
127 }
128 restored.fabric.* = candidate;
129 restored.moment = plan.next_moment;
130 std.debug.assert(std.meta.eql(restored.fabric.root(), transition.root));
131 return .{ .applied = .{
132 .transition = transition,
133 .decision = plan.decision,
134 .moment = plan.next_moment,
135 } };
136 }
137
138 fn prepareCommit(
139 restored: *restore.Restored,
140 candidate: *const fabric.Fabric,
141 transition: fabric.Transition,
142 ) Error!CommitPlan {
143 const entry = switch (transition.entry.value) {
144 .fault => |value| value,
145 else => return error.FaultTransitionExpected,
146 };
147 if (!std.meta.eql(candidate.root(), transition.root)) {
148 return error.FaultTransitionMismatch;
149 }
150 const crash_index = switch (entry.effect) {
151 .machine_crash => |node| if (entry.decision.choice == .inject)
152 try liveNodeIndex(restored, node)
153 else blk: {
154 _ = try liveNodeIndex(restored, node);
155 break :blk null;
156 },
157 .process_crash => return error.ProcessFaultProviderUnavailable,
158 else => null,
159 };
160 const cut = try candidate.cut();
161 try validateCandidate(restored, cut, crash_index);
162 const next_moment = try moment.prepare(restored.root, candidate.root());
163 return .{
164 .crash_index = crash_index,
165 .decision = entry.decision,
166 .next_moment = next_moment,
167 };
168 }
169
170 fn validateCapabilities(restored: *restore.Restored) Error!void {
171 _ = try restore.validateStable(restored);
172 }
173
174 fn validateCandidate(
175 restored: *restore.Restored,
176 cut: fabric.Cut,
177 crash_index: ?usize,
178 ) Error!void {
179 if (cut.node_count != restored.node_count) {
180 return error.CapabilitySetMismatch;
181 }
182 for (cut.nodes[0..cut.node_count], 0..) |boundary, index| {
183 const node = &restored.nodes[index];
184 if (!std.meta.eql(boundary.id, node.id)) {
185 return error.CapabilitySetMismatch;
186 }
187 if (crash_index != null and crash_index.? == index) {
188 if (boundary.available) return error.CapabilitySetMismatch;
189 const machine = if (node.machine) |*value|
190 value
191 else
192 return error.MachineNodeUnavailable;
193 if (!machine.active()) return error.MachineNodeUnavailable;
194 } else if (boundary.available) {
195 const machine = if (node.machine) |*value|
196 value
197 else
198 return error.CapabilitySetMismatch;
199 if (!machine.active()) return error.CapabilitySetMismatch;
200 } else if (node.machine != null) {
201 return error.CapabilitySetMismatch;
202 }
203 }
204 }
205
206 fn liveNodeIndex(
207 restored: *restore.Restored,
208 id: fabric.NodeId,
209 ) Error!usize {
210 for (restored.nodes[0..restored.node_count], 0..) |*node, index| {
211 if (!std.meta.eql(node.id, id)) continue;
212 const machine = if (node.machine) |*value|
213 value
214 else
215 return error.MachineNodeUnavailable;
216 if (!machine.active()) return error.MachineNodeUnavailable;
217 return index;
218 }
219 return error.MachineNodeUnavailable;
220 }
221
222 fn faultNode(effect: fabric.FaultEffect) ?fabric.NodeId {
223 return switch (effect) {
224 .machine_crash => |node| node,
225 .process_crash => |value| value.node,
226 else => null,
227 };
228 }
229
230 fn transitionNode(transition: fabric.Transition) ?fabric.NodeId {
231 return switch (transition.entry.value) {
232 .fault => |entry| faultNode(entry.effect),
233 else => null,
234 };
235 }
236
237 fn reject(node: ?fabric.NodeId, failure: Error) Result {
238 return .{ .rejected = .{ .node = node, .failure = failure } };
239 }