laminar_core/cluster/control/topology/
recovery.rs

1//! Private recovery selection for an irreversible topology Commit.
2
3use super::{TopologyCommit, TopologyError, TopologyRestoreInput};
4use crate::checkpoint::CommittedCheckpointIndex;
5use crate::checkpoint_decision::CheckpointOutcome;
6
7/// Recovery rounds carrying a topology Commit require every exact participant to understand
8/// that binding. Older round decoders reject the additional field rather than recover a parent.
9pub const TOPOLOGY_RECOVERY_PROTOCOL_VERSION: u16 = 6;
10
11/// Immutable target of an existing coordinated recovery round. Its cut is selected only after
12/// the complete stopped quorum; binding the target earlier prevents a catalog/Prepare race.
13#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
14#[serde(deny_unknown_fields)]
15pub struct TopologyRecoveryBinding {
16    protocol_version: u16,
17    commit: TopologyCommit,
18    processes: Vec<crate::cluster::control::LocalProcessAuthorityIdentity>,
19}
20
21impl TopologyRecoveryBinding {
22    pub(crate) fn new(
23        commit: TopologyCommit,
24        processes: Vec<crate::cluster::control::LocalProcessAuthorityIdentity>,
25    ) -> Result<Self, TopologyError> {
26        let binding = Self {
27            protocol_version: TOPOLOGY_RECOVERY_PROTOCOL_VERSION,
28            commit,
29            processes,
30        };
31        binding.validate()?;
32        Ok(binding)
33    }
34
35    /// Exact catalog decision this round must restore and release.
36    #[must_use]
37    pub const fn commit(&self) -> &TopologyCommit {
38        &self.commit
39    }
40
41    /// Complete current owner/evidence process terms frozen into this recovery round.
42    #[must_use]
43    pub fn processes(&self) -> &[crate::cluster::control::LocalProcessAuthorityIdentity] {
44        &self.processes
45    }
46
47    pub(crate) fn validate(&self) -> Result<(), TopologyError> {
48        self.commit.validate()?;
49        if self.protocol_version != TOPOLOGY_RECOVERY_PROTOCOL_VERSION {
50            return Err(TopologyError::Protocol(
51                "topology-bound recovery requires protocol six".into(),
52            ));
53        }
54        if self.processes.is_empty()
55            || self.processes.len() > crate::checkpoint::MAX_CHECKPOINT_PARTICIPANTS
56            || self.processes.iter().any(|process| !process.is_canonical())
57            || self
58                .processes
59                .windows(2)
60                .any(|pair| pair[0].participant.node_id >= pair[1].participant.node_id)
61        {
62            return Err(TopologyError::Invalid(
63                "topology recovery process roster is not canonical".into(),
64            ));
65        }
66        Ok(())
67    }
68}
69
70/// Which exact durable cut supplies a committed target's private state image.
71#[derive(Debug, Clone, Copy, PartialEq, Eq)]
72pub enum TopologyRecoveryCut {
73    /// The authorized historical parent cut and its explicit target state mapping.
74    MigrationRoot,
75    /// A later checkpoint produced by the committed target itself.
76    TargetCheckpoint,
77}
78
79/// An audited choice of recovery cut, bound to the current target and local process adoption.
80/// Construction belongs to the shared authority. This grants private state reads only; actor
81/// installation and intake require a fresh coordinated recovery round and its complete quorum.
82#[derive(Debug, Clone, PartialEq, Eq)]
83pub struct TopologyRecoveryInput {
84    pub(crate) migration: TopologyRestoreInput,
85    pub(crate) target_checkpoint: Option<Box<(CheckpointOutcome, CommittedCheckpointIndex)>>,
86}
87
88impl TopologyRecoveryInput {
89    /// The current target, retained mapping and exact current owner/process roster.
90    #[must_use]
91    pub const fn migration(&self) -> &TopologyRestoreInput {
92        &self.migration
93    }
94
95    /// Selected state provenance. A target checkpoint never uses the parent state mapping.
96    #[must_use]
97    pub const fn cut(&self) -> TopologyRecoveryCut {
98        if self.target_checkpoint.is_some() {
99            TopologyRecoveryCut::TargetCheckpoint
100        } else {
101            TopologyRecoveryCut::MigrationRoot
102        }
103    }
104
105    /// Definitive Commit of the selected cut, retaining its original checkpoint identity.
106    #[must_use]
107    pub fn outcome(&self) -> &CheckpointOutcome {
108        self.target_checkpoint
109            .as_deref()
110            .map_or_else(|| self.migration.outcome(), |(outcome, _)| outcome)
111    }
112
113    /// Exact selected index. Root indexes retain the parent pipeline identity; later indexes
114    /// must have the target pipeline identity. Neither is relabelled during reconstruction.
115    #[must_use]
116    pub fn checkpoint(&self) -> &CommittedCheckpointIndex {
117        self.target_checkpoint
118            .as_deref()
119            .map_or_else(|| self.migration.checkpoint(), |(_, checkpoint)| checkpoint)
120    }
121
122    /// Compare two fresh authorizations, including the chosen cut and all local restore fences.
123    /// The caller must still recheck the current recovery round before actor installation.
124    #[must_use]
125    pub fn same_restore_requirements(&self, other: &Self) -> bool {
126        self.migration.same_restore_requirements(&other.migration)
127            && self.target_checkpoint == other.target_checkpoint
128    }
129}
laminar_core/cluster/control/topology/recovery.rs

laminar_core/cluster/control/topology/
recovery.rs
