laminar_core/cluster/control/topology/
target_preparation.rs

1//! Historical target preparation observations under the still-held parent cut.
2
3use serde::{Deserialize, Serialize};
4
5use super::{TopologyAdmissionPhase, TopologyAdmissionStatus, TopologyError};
6use crate::checkpoint::{CheckpointParticipant, MAX_CHECKPOINT_PARTICIPANTS};
7
8/// Every reporting process must implement target restore and observed parent retirement.
9pub const TOPOLOGY_TARGET_PREPARATION_PROTOCOL_VERSION: u16 = 3;
10
11/// One process restored the bound target and observed all its parent actors terminal.
12/// The operation's immutable plan/root/cut bind the observation. It is historical evidence,
13/// not proof that an image is still resident, receivers/sinks are installed or intake may open.
14#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
15#[serde(deny_unknown_fields)]
16pub struct TopologyTargetPreparationReceipt {
17    /// Exact frozen owner/evidence participant, including its boot incarnation.
18    pub participant: CheckpointParticipant,
19    /// Exact process term from its original candidate certificate.
20    pub process_term: u64,
21    /// Exact target preparation protocol understood by this process.
22    pub protocol_version: u16,
23    /// First immutable shared authority append carrying this observation.
24    pub authority_sequence: u64,
25}
26
27impl TopologyAdmissionStatus {
28    /// Whether the still-prepared operation has observations from every frozen participant.
29    /// This does not authorize a target Commit, installation or Release. Commit must revalidate
30    /// current processes and assignment; installation must restore/retain its own target image.
31    #[must_use]
32    pub fn target_preparation_complete(&self) -> bool {
33        self.phase == TopologyAdmissionPhase::CutPrepared
34            && self.target_preparation_roster_complete()
35    }
36
37    pub(crate) fn target_preparation_roster_complete(&self) -> bool {
38        self.migration_root.is_some()
39            && self.cut.as_ref().is_some_and(|cut| {
40                cut.inventory
41                    .assignment_fence
42                    .as_ref()
43                    .is_some_and(|fence| {
44                        self.target_preparations.len() == fence.participants.len()
45                            && self
46                                .target_preparations
47                                .iter()
48                                .zip(&fence.participants)
49                                .all(|(receipt, participant)| receipt.participant == *participant)
50                    })
51            })
52    }
53
54    // Only receipt/status progress can change while an image is retained. Every field defining
55    // its restore authority remains exact. Destructuring forces future fields to be considered.
56    pub(crate) fn same_restore_binding(&self, other: &Self) -> bool {
57        (self.phase == other.phase || (self.has_target_commit() && other.has_target_commit()))
58            && self.commit == other.commit
59            && self.same_migration_binding(other)
60    }
61
62    // A Commit preserves these historical requirements while changing phase/catalog authority.
63    pub(crate) fn same_migration_binding(&self, other: &Self) -> bool {
64        let Self {
65            operation_id,
66            plan,
67            admitted_by,
68            admitted_sequence,
69            status_sequence: _,
70            phase: _,
71            cut,
72            preparation,
73            migration_root,
74            target_preparations: _,
75            commit: _,
76            activation: _,
77        } = self;
78        *operation_id == other.operation_id
79            && *plan == other.plan
80            && *admitted_by == other.admitted_by
81            && *admitted_sequence == other.admitted_sequence
82            && *cut == other.cut
83            && *preparation == other.preparation
84            && *migration_root == other.migration_root
85    }
86
87    pub(crate) fn validate_target_preparations(&self, head: u64) -> Result<(), TopologyError> {
88        if self.target_preparations.is_empty() {
89            return Ok(());
90        }
91        let root = self.migration_root.as_ref().ok_or_else(|| {
92            TopologyError::Invalid("target preparation requires a published root".into())
93        })?;
94        let preparation = self.preparation.as_ref().ok_or_else(|| {
95            TopologyError::Invalid("target preparation requires candidate certificates".into())
96        })?;
97        if !matches!(
98            self.phase,
99            TopologyAdmissionPhase::CutPrepared
100                | TopologyAdmissionPhase::Committed
101                | TopologyAdmissionPhase::Activating
102                | TopologyAdmissionPhase::Active
103                | TopologyAdmissionPhase::Aborted { .. }
104        ) || self.target_preparations.len() > MAX_CHECKPOINT_PARTICIPANTS
105            || !self
106                .target_preparations
107                .windows(2)
108                .all(|pair| pair[0].participant.node_id < pair[1].participant.node_id)
109            || self.target_preparations.iter().any(|receipt| {
110                !matches!(
111                    receipt.protocol_version,
112                    TOPOLOGY_TARGET_PREPARATION_PROTOCOL_VERSION
113                        | super::TOPOLOGY_COMMIT_PROTOCOL_VERSION
114                ) || receipt.authority_sequence <= root.authority_sequence
115                    || receipt.authority_sequence > self.status_sequence
116                    || receipt.authority_sequence > head
117                    || !preparation.certificates.iter().any(|certificate| {
118                        certificate.participant == receipt.participant
119                            && certificate.process_term == receipt.process_term
120                    })
121            })
122        {
123            return Err(TopologyError::Invalid(
124                "invalid target preparation observations".into(),
125            ));
126        }
127        Ok(())
128    }
129
130    pub(crate) fn validate_target_preparation_successor(
131        &self,
132        after: &Self,
133        sequence: u64,
134    ) -> Result<(), TopologyError> {
135        if self.target_preparations == after.target_preparations {
136            return Ok(());
137        }
138        after.validate_target_preparations(sequence)?;
139        if self.phase != TopologyAdmissionPhase::CutPrepared
140            || self.migration_root.is_none()
141            || !self.same_restore_binding(after)
142            || after.target_preparations.len() != self.target_preparations.len() + 1
143            || self
144                .target_preparations
145                .iter()
146                .any(|receipt| !after.target_preparations.contains(receipt))
147            || after
148                .target_preparations
149                .iter()
150                .filter(|receipt| !self.target_preparations.contains(receipt))
151                .any(|receipt| receipt.authority_sequence != sequence)
152        {
153            return Err(TopologyError::Invalid(
154                "authority cannot rewrite, forget or append stale target preparation observations"
155                    .into(),
156            ));
157        }
158        Ok(())
159    }
160}
laminar_core/cluster/control/topology/target_preparation.rs

laminar_core/cluster/control/topology/
target_preparation.rs
