lib/machine/src/explore/distributed/state.zig

daab053ee43316e1809a84551d573ddd1e5bf3d2

 1 const fabric = @import("../../fabric/root.zig");
 2 const queue = @import("queue.zig");
 3 const std = @import("std");
 4 const types = @import("types.zig");
 5 
 6 /// One position of the three-node protocol. The state records the ledger root
 7 /// reached so far, the pending messages, what each node accepted, the
 8 /// coordinator's round, proposal, and acknowledgement count, the commit, which
 9 /// node is hung and for how long, the simulated clock, and which of the two
10 /// facts the planted defect needs have happened. The driver and the replayer
11 /// hand one to each workload step and read it back afterward. Only a workload
12 /// step changes the state. Each step that advances folds the previous ledger
13 /// digest, the decision, the protocol fields, and the pending messages into the
14 /// next digest, so two equal histories reach equal *ledger roots* (the digest
15 /// and counters naming the position of the ledger, the ordered record of every
16 /// admitted input and fault decision). `init` starts on a given ledger root at
17 /// tick 1, with no messages, no *round* (one attempt by the coordinator, node
18 /// 0, to get a value accepted), and no commit. `backing` counts the nodes that
19 /// accepted the current round's proposal. `votes` counts the coordinator's own
20 /// vote plus every *acknowledgement* (a node's reply that it accepted a round's
21 /// proposal) it counted in the current round. `suspecting` holds while a round
22 /// is open and a node other than the coordinator has stayed hung for at least
23 /// two fault steps.
24 pub const State = struct {
25     fabric: fabric.Root,
26     messages: queue.Queue = .{},
27     nodes: [types.node_count]types.Node = @splat(types.Node{}),
28     round: u16 = 0,
29     open: bool = false,
30     proposal: u64 = 0,
31     acks: u8 = 0,
32     ack_mask: u8 = 0,
33     committed: bool = false,
34     committed_round: u16 = 0,
35     committed_value: u64 = 0,
36     hang_streak: u8 = 0,
37     hung: ?u8 = null,
38     delivery_blocked: bool = false,
39     turn_node: ?u8 = null,
40     request: u64 = 0,
41     now: u64 = types.start_tick,
42     conjunct: types.Conjunct = .{},
43 
44     pub fn init(root: fabric.Root) State {
45         const value: State = .{ .fabric = root };
46         value.assertValid();
47         return value;
48     }
49 
50     pub fn backing(self: *const State) u8 {
51         self.assertValid();
52         var count: u8 = 0;
53         for (self.nodes) |node| {
54             if (!node.accepted) continue;
55             if (node.accepted_round != self.round) continue;
56             if (node.accepted_value != self.proposal) continue;
57             count += 1;
58         }
59         std.debug.assert(count <= types.node_count);
60         return count;
61     }
62 
63     pub fn votes(self: *const State) u8 {
64         self.assertValid();
65         return 1 + self.acks;
66     }
67 
68     pub fn suspecting(self: *const State) bool {
69         self.assertValid();
70         if (!self.open) return false;
71         if (self.hang_streak < types.suspicion_steps) return false;
72         const hung = self.hung orelse return false;
73         return hung != types.coordinator;
74     }
75 
76     pub fn assertValid(self: *const State) void {
77         std.debug.assert(self.acks <= types.depth);
78         std.debug.assert(self.committed_round <= self.round);
79         std.debug.assert(self.now >= types.start_tick);
80         std.debug.assert(self.now <= types.end_tick);
81         if (self.hung) |node| std.debug.assert(node < types.node_count);
82         if (self.turn_node) |node| std.debug.assert(node < types.node_count);
83         for (self.nodes) |node| std.debug.assert(node.accepted_round <= self.round);
84     }
85 };