laminar_core/checkpoint/assignment/
mod.rs

1//! Feature-neutral checkpoint assignment certificate.
2
3use uuid::Uuid;
4
5use super::LeaderProof;
6use crate::state::{KeyGroupCount, PARTITIONING_ABI_VERSION};
7
8/// Maximum participants supported by the current all-to-all shuffle transport.
9///
10/// Every receiver admits one persistent stream from each of the other 128 participants; the
11/// receiver itself does not consume an inbound peer stream. Assignment certificates above this
12/// bound cannot establish a cluster-wide shuffle barrier and therefore fail admission.
13pub const MAX_CHECKPOINT_PARTICIPANTS: usize = 128 + 1;
14
15/// One exact process participating in a checkpoint cut.
16#[derive(Debug, Clone, Copy, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
17#[serde(deny_unknown_fields)]
18pub struct CheckpointParticipant {
19    /// Stable logical node identifier.
20    pub node_id: u64,
21    /// Boot-unique process identity. A same-node restart is a different participant.
22    pub boot_incarnation: Uuid,
23}
24
25/// One process's exact adopted assignment identity and local vnode-state readiness, published into
26/// its control-plane slot.
27#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
28#[serde(try_from = "UncheckedCheckpointAssignmentAdoption")]
29pub struct CheckpointAssignmentAdoption {
30    /// Reporting process.
31    pub participant: CheckpointParticipant,
32    /// Adopted assignment version.
33    pub assignment_version: u64,
34    /// Key encoding, hashing, and key-group mapping contract that was adopted.
35    pub partitioning_abi_version: u16,
36    /// Adopted vnode count.
37    pub vnode_count: u32,
38    /// Digest of the adopted partitioning ABI and ordered vnode-owner map.
39    pub assignment_digest: [u8; 32],
40    /// Whether every vnode transition for this assignment has completed semantic graph install.
41    /// Transport activation intentionally does not require this bit; assignment rotation does.
42    pub vnode_state_ready: bool,
43}
44
45#[derive(serde::Deserialize)]
46#[serde(deny_unknown_fields)]
47struct UncheckedCheckpointAssignmentAdoption {
48    participant: CheckpointParticipant,
49    assignment_version: u64,
50    partitioning_abi_version: u16,
51    vnode_count: u32,
52    assignment_digest: [u8; 32],
53    vnode_state_ready: bool,
54}
55
56impl TryFrom<UncheckedCheckpointAssignmentAdoption> for CheckpointAssignmentAdoption {
57    type Error = &'static str;
58
59    fn try_from(unchecked: UncheckedCheckpointAssignmentAdoption) -> Result<Self, Self::Error> {
60        let adoption = Self {
61            participant: unchecked.participant,
62            assignment_version: unchecked.assignment_version,
63            partitioning_abi_version: unchecked.partitioning_abi_version,
64            vnode_count: unchecked.vnode_count,
65            assignment_digest: unchecked.assignment_digest,
66            vnode_state_ready: unchecked.vnode_state_ready,
67        };
68        adoption
69            .is_canonical()
70            .then_some(adoption)
71            .ok_or("checkpoint assignment adoption is not canonical")
72    }
73}
74
75impl CheckpointAssignmentAdoption {
76    /// Whether every field has its canonical production shape.
77    #[must_use]
78    pub fn is_canonical(&self) -> bool {
79        self.participant.node_id != 0
80            && !self.participant.boot_incarnation.is_nil()
81            && self.assignment_version != 0
82            && self.partitioning_abi_version == PARTITIONING_ABI_VERSION
83            && KeyGroupCount::try_from(self.vnode_count).is_ok()
84            && self.assignment_digest != [0; 32]
85    }
86
87    /// Whether this report adopted the same map identity as `fence`.
88    #[must_use]
89    pub fn matches_fence(&self, fence: &CheckpointAssignmentFence) -> bool {
90        self.assignment_version == fence.assignment_version
91            && self.partitioning_abi_version == fence.partitioning_abi_version
92            && self.vnode_count == fence.vnode_count
93            && self.assignment_digest == fence.assignment_digest
94            && fence.participant_incarnation(self.participant.node_id)
95                == Some(self.participant.boot_incarnation)
96    }
97}
98
99/// Versioned proof of one exact vnode-owner map and its process-complete participant roster.
100#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
101#[serde(try_from = "UncheckedCheckpointAssignmentFence")]
102pub struct CheckpointAssignmentFence {
103    /// Exact vnode-assignment version covered by this proof.
104    pub assignment_version: u64,
105    /// Key encoding, hashing, and key-group mapping contract covered by this proof.
106    pub partitioning_abi_version: u16,
107    /// Number of vnodes in the ordered owner map.
108    pub vnode_count: u32,
109    /// SHA-256 of the partitioning ABI, `vnode_count`, and ordered vnode-to-owner map.
110    pub assignment_digest: [u8; 32],
111    /// Canonical participants, sorted by node id and bound to exact boot incarnations.
112    pub participants: Vec<CheckpointParticipant>,
113}
114
115#[derive(serde::Deserialize)]
116#[serde(deny_unknown_fields)]
117struct UncheckedCheckpointAssignmentFence {
118    assignment_version: u64,
119    partitioning_abi_version: u16,
120    vnode_count: u32,
121    assignment_digest: [u8; 32],
122    participants: Vec<CheckpointParticipant>,
123}
124
125impl TryFrom<UncheckedCheckpointAssignmentFence> for CheckpointAssignmentFence {
126    type Error = &'static str;
127
128    fn try_from(unchecked: UncheckedCheckpointAssignmentFence) -> Result<Self, Self::Error> {
129        let fence = Self {
130            assignment_version: unchecked.assignment_version,
131            partitioning_abi_version: unchecked.partitioning_abi_version,
132            vnode_count: unchecked.vnode_count,
133            assignment_digest: unchecked.assignment_digest,
134            participants: unchecked.participants,
135        };
136        fence
137            .is_canonical()
138            .then_some(fence)
139            .ok_or("checkpoint assignment certificate is not canonical")
140    }
141}
142
143impl CheckpointAssignmentFence {
144    /// Build a certificate from the exact ordered vnode-owner map.
145    ///
146    /// # Errors
147    /// Returns an error when the version, owner map, or participant roster is non-canonical.
148    pub fn from_owner_map(
149        assignment_version: u64,
150        owners: &[u64],
151        participants: Vec<CheckpointParticipant>,
152    ) -> Result<Self, String> {
153        if participants.len() > MAX_CHECKPOINT_PARTICIPANTS {
154            return Err(format!(
155                "checkpoint assignment has {} participants; maximum is {MAX_CHECKPOINT_PARTICIPANTS}",
156                participants.len()
157            ));
158        }
159        let vnode_count = u32::try_from(owners.len())
160            .map_err(|_| "checkpoint assignment has more than u32::MAX vnodes".to_string())?;
161        KeyGroupCount::try_from(vnode_count).map_err(|_| {
162            format!(
163                "checkpoint assignment key-group count must be between 1 and {}, got {vnode_count}",
164                crate::state::MAX_KEY_GROUP_COUNT
165            )
166        })?;
167        let fence = Self {
168            assignment_version,
169            partitioning_abi_version: PARTITIONING_ABI_VERSION,
170            vnode_count,
171            assignment_digest: Self::owner_map_digest(vnode_count, owners),
172            participants,
173        };
174        if !fence.is_canonical() {
175            return Err("checkpoint assignment certificate is not canonical".into());
176        }
177        let owner_ids: std::collections::BTreeSet<u64> = owners.iter().copied().collect();
178        if owner_ids.len() != fence.participants.len()
179            || fence
180                .participants
181                .iter()
182                .any(|participant| !owner_ids.contains(&participant.node_id))
183        {
184            return Err("checkpoint participants must be the exact vnode-owner set".into());
185        }
186        Ok(fence)
187    }
188
189    /// Stable digest of the current partitioning ABI and a canonical ordered vnode-owner map.
190    #[must_use]
191    pub fn owner_map_digest(vnode_count: u32, owners: &[u64]) -> [u8; 32] {
192        Self::owner_map_digest_iter(vnode_count, owners.iter().copied())
193    }
194
195    /// Allocation-free stable digest of the current partitioning ABI and an exact-size ordered
196    /// vnode-owner iterator.
197    #[must_use]
198    pub fn owner_map_digest_iter(
199        vnode_count: u32,
200        owners: impl ExactSizeIterator<Item = u64>,
201    ) -> [u8; 32] {
202        Self::owner_map_digest_iter_for_abi(PARTITIONING_ABI_VERSION, vnode_count, owners)
203    }
204
205    #[cfg(test)]
206    fn owner_map_digest_for_abi(
207        partitioning_abi_version: u16,
208        vnode_count: u32,
209        owners: &[u64],
210    ) -> [u8; 32] {
211        Self::owner_map_digest_iter_for_abi(
212            partitioning_abi_version,
213            vnode_count,
214            owners.iter().copied(),
215        )
216    }
217
218    fn owner_map_digest_iter_for_abi(
219        partitioning_abi_version: u16,
220        vnode_count: u32,
221        owners: impl ExactSizeIterator<Item = u64>,
222    ) -> [u8; 32] {
223        use sha2::{Digest, Sha256};
224
225        let owner_count = owners.len();
226        let mut hash = Sha256::new();
227        hash.update(b"laminardb-vnode-owner-map-v2\0");
228        hash.update(partitioning_abi_version.to_le_bytes());
229        hash.update(vnode_count.to_le_bytes());
230        hash.update(u64::try_from(owner_count).unwrap_or(u64::MAX).to_le_bytes());
231        for owner in owners {
232            hash.update(owner.to_le_bytes());
233        }
234        hash.finalize().into()
235    }
236
237    /// Whether every field has its canonical production shape.
238    #[must_use]
239    pub fn is_canonical(&self) -> bool {
240        self.assignment_version != 0
241            && self.partitioning_abi_version == PARTITIONING_ABI_VERSION
242            && KeyGroupCount::try_from(self.vnode_count).is_ok()
243            && self.assignment_digest != [0; 32]
244            && !self.participants.is_empty()
245            && self.participants.len() <= MAX_CHECKPOINT_PARTICIPANTS
246            && self
247                .participants
248                .windows(2)
249                .all(|pair| pair[0].node_id < pair[1].node_id)
250            && self.participants.iter().all(|participant| {
251                participant.node_id != 0 && !participant.boot_incarnation.is_nil()
252            })
253    }
254
255    /// Whether `node_id` belongs to the canonical participant roster.
256    #[must_use]
257    pub fn contains(&self, node_id: u64) -> bool {
258        self.participants
259            .binary_search_by_key(&node_id, |participant| participant.node_id)
260            .is_ok()
261    }
262
263    /// Exact boot identity certified for `node_id`.
264    #[must_use]
265    pub fn participant_incarnation(&self, node_id: u64) -> Option<Uuid> {
266        self.participants
267            .binary_search_by_key(&node_id, |participant| participant.node_id)
268            .ok()
269            .map(|index| self.participants[index].boot_incarnation)
270    }
271
272    /// Canonical participant node ids.
273    #[must_use]
274    pub fn participant_ids(&self) -> Vec<u64> {
275        self.participants
276            .iter()
277            .map(|participant| participant.node_id)
278            .collect()
279    }
280
281    /// Whether this certificate binds the supplied exact ordered owner map.
282    #[must_use]
283    pub fn matches_owner_map(&self, owners: &[u64]) -> bool {
284        self.partitioning_abi_version == PARTITIONING_ABI_VERSION
285            && usize::try_from(self.vnode_count).ok() == Some(owners.len())
286            && self.assignment_digest == Self::owner_map_digest(self.vnode_count, owners)
287            && {
288                let owner_ids: std::collections::BTreeSet<u64> = owners.iter().copied().collect();
289                owner_ids.len() == self.participants.len()
290                    && self
291                        .participants
292                        .iter()
293                        .all(|participant| owner_ids.contains(&participant.node_id))
294            }
295    }
296
297    /// Stable SHA-256 binding of every certificate dimension.
298    #[must_use]
299    pub fn digest(&self) -> [u8; 32] {
300        use sha2::{Digest, Sha256};
301
302        let mut hash = Sha256::new();
303        hash.update(b"laminardb-checkpoint-assignment-v3\0");
304        hash.update(self.assignment_version.to_le_bytes());
305        hash.update(self.partitioning_abi_version.to_le_bytes());
306        hash.update(self.vnode_count.to_le_bytes());
307        hash.update(self.assignment_digest);
308        hash.update(
309            u64::try_from(self.participants.len())
310                .unwrap_or(u64::MAX)
311                .to_le_bytes(),
312        );
313        for participant in &self.participants {
314            hash.update(participant.node_id.to_le_bytes());
315            hash.update(participant.boot_incarnation.as_bytes());
316        }
317        hash.finalize().into()
318    }
319}
320
321/// Compact identity of one exact assignment transition.
322#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, serde::Serialize, serde::Deserialize)]
323pub struct AssignmentDrainId {
324    /// Assignment version that remains authoritative while sources drain.
325    pub predecessor_version: u64,
326    /// Exact successor version sources are preparing to adopt.
327    pub target_version: u64,
328    /// SHA-256 binding both assignment certificates and the initiating leader term.
329    pub digest: [u8; 32],
330}
331
332impl AssignmentDrainId {
333    /// Whether every identity field has a canonical production value.
334    #[must_use]
335    pub fn is_canonical(self) -> bool {
336        self.predecessor_version != 0
337            && self.predecessor_version.checked_add(1) == Some(self.target_version)
338            && self.digest != [0; 32]
339    }
340}
341
342/// Exact predecessor-to-target transition at which every predecessor source must stop input.
343///
344/// Required drain participants always come from `predecessor`. A process that only joins the
345/// target has no predecessor input authority and therefore cannot block the drain quorum.
346#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
347#[serde(try_from = "UncheckedAssignmentDrainTransition")]
348pub struct AssignmentDrainTransition {
349    /// Assignment that remains authoritative through the pre-rotation checkpoint.
350    pub predecessor: CheckpointAssignmentFence,
351    /// Exact successor assignment to install after that checkpoint is durable.
352    pub target: CheckpointAssignmentFence,
353    /// Leader term that created the durable draining transition.
354    pub leader: LeaderProof,
355}
356
357#[derive(serde::Deserialize)]
358#[serde(deny_unknown_fields)]
359struct UncheckedAssignmentDrainTransition {
360    predecessor: CheckpointAssignmentFence,
361    target: CheckpointAssignmentFence,
362    leader: LeaderProof,
363}
364
365impl TryFrom<UncheckedAssignmentDrainTransition> for AssignmentDrainTransition {
366    type Error = &'static str;
367
368    fn try_from(unchecked: UncheckedAssignmentDrainTransition) -> Result<Self, Self::Error> {
369        Self::new(unchecked.predecessor, unchecked.target, unchecked.leader)
370            .map_err(|_| "assignment drain transition is not canonical")
371    }
372}
373
374impl AssignmentDrainTransition {
375    /// Construct and validate one exact successor transition.
376    ///
377    /// # Errors
378    /// Returns an error when either certificate, the version relation, vnode count, or captured
379    /// leader term is not canonical for the predecessor roster.
380    pub fn new(
381        predecessor: CheckpointAssignmentFence,
382        target: CheckpointAssignmentFence,
383        leader: LeaderProof,
384    ) -> Result<Self, String> {
385        let transition = Self {
386            predecessor,
387            target,
388            leader,
389        };
390        if !transition.is_canonical() {
391            return Err("assignment drain transition is not canonical".into());
392        }
393        Ok(transition)
394    }
395
396    /// Whether the transition is an exact, authority-bound successor of the predecessor.
397    #[must_use]
398    pub fn is_canonical(&self) -> bool {
399        self.predecessor.is_canonical()
400            && self.target.is_canonical()
401            && self.leader.is_canonical()
402            && self.predecessor.vnode_count == self.target.vnode_count
403            && self.predecessor.assignment_version.checked_add(1)
404                == Some(self.target.assignment_version)
405            && (self
406                .predecessor
407                .participant_incarnation(self.leader.owner.node_id)
408                == Some(self.leader.owner.boot_id)
409                || self
410                    .target
411                    .participant_incarnation(self.leader.owner.node_id)
412                    == Some(self.leader.owner.boot_id))
413    }
414
415    /// Compact identity carried by source commands and receipts.
416    #[must_use]
417    pub fn id(&self) -> AssignmentDrainId {
418        AssignmentDrainId {
419            predecessor_version: self.predecessor.assignment_version,
420            target_version: self.target.assignment_version,
421            digest: self.digest(),
422        }
423    }
424
425    /// Exact processes that must publish drain acknowledgements.
426    #[must_use]
427    pub fn required_participants(&self) -> &[CheckpointParticipant] {
428        &self.predecessor.participants
429    }
430
431    /// Stable SHA-256 binding both assignment maps, both process rosters, and the leader term.
432    #[must_use]
433    pub fn digest(&self) -> [u8; 32] {
434        use sha2::{Digest, Sha256};
435
436        let mut hash = Sha256::new();
437        hash.update(b"laminardb-assignment-drain-transition-v1\0");
438        hash.update(self.predecessor.digest());
439        hash.update(self.target.digest());
440        hash.update(self.leader.owner.node_id.to_le_bytes());
441        hash.update(self.leader.owner.boot_id.as_bytes());
442        hash.update(self.leader.owner.process_term.to_le_bytes());
443        hash.update(self.leader.fencing_token.to_le_bytes());
444        hash.finalize().into()
445    }
446}
447
448#[cfg(test)]
449mod tests;
laminar_core/checkpoint/assignment/mod.rs

laminar_core/checkpoint/assignment/
mod.rs
