krabka_kraft_core/
simulation_support.rs

1//! Shared deterministic mechanics for the browser and storage simulation harnesses.
2
3use {
4    crate::{
5        Action, Epoch, Event, LogEnd, NodeId, QuorumStateMachine, Role, SimInstant, SuccessorRank,
6    },
7    std::collections::{BTreeSet, VecDeque},
8};
9
10/// The endpoints of one queued message, used to clear messages crossing a
11/// simulator partition.
12pub trait PartitionMessage<Node> {
13    fn touches(&self, node: Node) -> bool;
14}
15
16/// Isolate one node and remove messages already in flight to or from it.
17pub fn partition_node<Node, Message>(
18    partitioned: &mut BTreeSet<Node>,
19    queue: &mut VecDeque<Message>,
20    node: Node,
21) where
22    Node: Copy + Ord,
23    Message: PartitionMessage<Node>,
24{
25    partitioned.insert(node);
26    queue.retain(|message| !message.touches(node));
27}
28
29/// Stop after two timer rounds leave the cluster fingerprint unchanged.
30pub struct StableFingerprint<F> {
31    previous: F,
32    unchanged: u32,
33}
34
35impl<F: PartialEq> StableFingerprint<F> {
36    #[must_use]
37    pub fn new(previous: F) -> Self {
38        Self {
39            previous,
40            unchanged: 0,
41        }
42    }
43
44    /// Observe this round and report whether two consecutive rounds were stable.
45    pub fn observe(&mut self, fingerprint: F) -> bool {
46        if fingerprint == self.previous {
47            self.unchanged = self.unchanged.saturating_add(1);
48        } else {
49            self.unchanged = 0;
50            self.previous = fingerprint;
51        }
52        self.unchanged >= 2
53    }
54}
55
56/// Static simulator voters have no directory identity or network endpoints.
57#[must_use]
58pub fn voter_set(ids: &[NodeId]) -> krabka_voters::VoterSet {
59    krabka_voters::VoterSet::from_voters(ids.iter().map(|&id| krabka_voters::Voter {
60        id,
61        directory_id: uuid::Uuid::nil(),
62        endpoints: Vec::new(),
63        kraft_version: krabka_voters::KRaftVersionRange::default(),
64    }))
65}
66
67/// Translate the consensus actions that only send messages.
68/// `None` leaves local log and timer effects to the harness.
69#[must_use]
70pub fn action_messages(
71    source: NodeId,
72    action: &Action,
73    voters: &[NodeId],
74    peers: &[NodeId],
75    log_end: LogEnd,
76) -> Option<Vec<(NodeId, Event)>> {
77    let recipients = match action {
78        Action::SendVoteRequest { .. } => voters,
79        Action::SendBeginQuorumEpoch { .. } | Action::SendEndQuorumEpoch { .. } => peers,
80        Action::ReplyVote { to, epoch, granted } => {
81            return Some(vec![(*to, Event::ReceiveVoteResponse {
82                from: source,
83                epoch: *epoch,
84                vote_granted: *granted,
85            })]);
86        }
87        _ => return None,
88    };
89    Some(
90        recipients
91            .iter()
92            .copied()
93            .filter(|peer| *peer != source)
94            .map(|peer| {
95                let event = match action {
96                    Action::SendVoteRequest { epoch, pre_vote } => Event::ReceiveVoteRequest {
97                        from: source,
98                        cluster_id: None,
99                        voter_id: peer,
100                        voter_directory_id: uuid::Uuid::nil(),
101                        candidate_epoch: *epoch,
102                        candidate: source,
103                        candidate_directory_id: uuid::Uuid::nil(),
104                        candidate_log_end: log_end,
105                        pre_vote: *pre_vote,
106                    },
107                    Action::SendBeginQuorumEpoch { epoch } => Event::ReceiveBeginQuorumEpoch {
108                        leader_id: source,
109                        leader_epoch: *epoch,
110                    },
111                    Action::SendEndQuorumEpoch { epoch, .. } => Event::ReceiveEndQuorumEpoch {
112                        leader_id: source,
113                        leader_epoch: *epoch,
114                        successor_rank: SuccessorRank::default(),
115                    },
116                    _ => unreachable!("message action selected above"),
117                };
118                (peer, event)
119            })
120            .collect(),
121    )
122}
123
124/// Answer a leader-side fetch only when data or divergence can advance the follower.
125#[must_use]
126pub fn fetch_response(
127    leader: NodeId,
128    epoch: Epoch,
129    actions: &[Action],
130    leader_end: i64,
131    follower_end: i64,
132) -> Option<Event> {
133    let diverging = actions.iter().find_map(|action| match action {
134        Action::ReplyDivergingEpoch(point) => Some(*point),
135        _ => None,
136    });
137    (diverging.is_some() || follower_end < leader_end).then_some(Event::ReceiveFetchResponse {
138        leader_id: leader,
139        leader_epoch: epoch,
140        diverging,
141    })
142}
143
144/// The timer wheel includes a leader heartbeat in addition to core timers.
145#[derive(Debug, Clone, Copy, PartialEq, Eq)]
146pub enum SimulationTimer {
147    Election,
148    Fetch,
149    Heartbeat,
150    CheckQuorum,
151}
152
153/// Earliest timer, retaining the input node order and timer order on equal deadlines.
154#[must_use]
155pub fn earliest_timer(
156    nodes: impl Iterator<Item = (NodeId, [Option<SimInstant>; 4])>,
157) -> Option<(SimInstant, NodeId, SimulationTimer)> {
158    let mut best = None;
159    for (id, deadlines) in nodes {
160        for (deadline, kind) in deadlines.into_iter().zip([
161            SimulationTimer::Election,
162            SimulationTimer::Fetch,
163            SimulationTimer::Heartbeat,
164            SimulationTimer::CheckQuorum,
165        ]) {
166            if let Some(deadline) = deadline
167                && best.is_none_or(|(current, _, _)| deadline < current)
168            {
169                best = Some((deadline, id, kind));
170            }
171        }
172    }
173    best
174}
175
176/// Disarm the timer that just fired before applying its event.
177pub fn clear_timer(
178    kind: SimulationTimer,
179    election: &mut Option<SimInstant>,
180    fetch: &mut Option<SimInstant>,
181    heartbeat: &mut Option<SimInstant>,
182    quorum: &mut Option<SimInstant>,
183) {
184    *match kind {
185        SimulationTimer::Election => election,
186        SimulationTimer::Fetch => fetch,
187        SimulationTimer::Heartbeat => heartbeat,
188        SimulationTimer::CheckQuorum => quorum,
189    } = None;
190}
191
192/// Enforce per-role timer ownership, preserving existing leader heartbeats.
193pub fn reconcile_timers(
194    role: &Role,
195    election: &mut Option<SimInstant>,
196    fetch: &mut Option<SimInstant>,
197    heartbeat: &mut Option<SimInstant>,
198    quorum: &mut Option<SimInstant>,
199    heartbeat_deadline: SimInstant,
200) {
201    match role {
202        Role::Leader { .. } => {
203            *election = None;
204            *fetch = None;
205            heartbeat.get_or_insert(heartbeat_deadline);
206        }
207        Role::Follower { .. } | Role::Observer { .. } => {
208            *election = None;
209            *heartbeat = None;
210            *quorum = None;
211        }
212        _ => {
213            *fetch = None;
214            *heartbeat = None;
215            *quorum = None;
216        }
217    }
218}
219
220/// A fetch watchdog can poll again while the node's leader is reachable.
221#[must_use]
222pub fn reachable_leader(
223    role: &Role,
224    follower: NodeId,
225    partitioned: &BTreeSet<NodeId>,
226    is_leader: impl FnOnce(NodeId) -> bool,
227) -> Option<NodeId> {
228    let leader = match role {
229        Role::Follower { leader_id, .. }
230        | Role::Observer {
231            leader_id: Some(leader_id),
232            ..
233        } => *leader_id,
234        _ => return None,
235    };
236    (!partitioned.contains(&follower) && !partitioned.contains(&leader) && is_leader(leader))
237        .then_some(leader)
238}
239
240/// What one node contributes to the cluster fingerprint.
241#[derive(Debug, Clone, PartialEq, Eq)]
242pub struct NodeFingerprint {
243    pub node: NodeId,
244    pub role: &'static str,
245    pub epoch: Epoch,
246    pub records: usize,
247    pub high_watermark: i64,
248}
249
250/// A node fingerprint uses the core's leader watermark when it leads.
251#[must_use]
252pub fn fingerprint(
253    id: NodeId,
254    machine: &QuorumStateMachine,
255    records: usize,
256    hwm: i64,
257) -> NodeFingerprint {
258    let hwm = match machine.role() {
259        Role::Leader { high_watermark, .. } => *high_watermark,
260        _ => hwm,
261    };
262    NodeFingerprint {
263        node: id,
264        role: machine.role().name(),
265        epoch: machine.quorum_state().leader_epoch,
266        records,
267        high_watermark: hwm,
268    }
269}
270
271/// Append one leader epoch per simulated record.
272pub fn append_epochs(epochs: &mut Vec<Epoch>, epoch: Epoch, count: usize) {
273    for _ in 0..count {
274        epochs.push(epoch);
275    }
276}
277
278/// Clamp negative cuts to zero; a cut beyond the log preserves it.
279pub fn truncate_epochs(epochs: &mut Vec<Epoch>, offset: i64) {
280    epochs.truncate(usize::try_from(offset.max(0)).unwrap_or(usize::MAX));
281}
282
283/// Return the leader suffix a follower log of `follower_len` records lacks.
284///
285/// The follower appends it to grow without overwriting its matching prefix. A
286/// follower at or beyond the leader's length gets an empty suffix.
287#[must_use]
288pub fn epochs_to_replicate(follower_len: usize, leader_epochs: &[Epoch]) -> &[Epoch] {
289    leader_epochs.get(follower_len..).unwrap_or_default()
290}
krabka_kraft_core/simulation_support.rs

krabka_kraft_core/
simulation_support.rs
