laminar_core/cluster/control/leader_lease/
topology_target_preparation.rs

1//! Exact-root target preparation observations use the existing authority append.
2
3use super::topology_admission::{CONTROL_TIMEOUT, MAX_ADMISSION_ATTEMPTS};
4use super::{
5    read_authority_record, AssignmentSnapshotStore, AuthorityCreateOutcome, LeaderLeaseStore,
6    LeaseError, TOPOLOGY_COMMIT_RECORD_VERSION, TOPOLOGY_TARGET_PREPARATION_RECORD_VERSION,
7};
8use crate::cluster::control::{
9    ProcessLeaseAuthority, TopologyAdmissionPhase, TopologyAdmissionStatus, TopologyError,
10    TopologyRestoreInput, TopologyTargetPreparationReceipt, TOPOLOGY_COMMIT_PROTOCOL_VERSION,
11    TOPOLOGY_TARGET_PREPARATION_PROTOCOL_VERSION,
12};
13
14impl LeaderLeaseStore {
15    /// Record one participant after it restored the exact private target and observed its old
16    /// compute/source/sink actors and connector children terminal. The runtime owns that proof;
17    /// call through its configured controller. This historical observation grants no install,
18    /// Commit, output or intake release, and does not certify that an image remains resident.
19    ///
20    /// # Errors
21    /// Rejects missing root/cut, changed requirements, old protocol, stale leader/process/assignment
22    /// or recovery. The budget is 16 CAS attempts within 15 seconds. A cancelled/uncertain write
23    /// may have appended: read operation status and retry the same input; never invent a new ID.
24    pub async fn certify_topology_target_preparation(
25        &self,
26        assignments: &AssignmentSnapshotStore,
27        processes: &ProcessLeaseAuthority,
28        input: &TopologyRestoreInput,
29        protocol_version: u16,
30    ) -> Result<TopologyAdmissionStatus, TopologyError> {
31        if !matches!(
32            protocol_version,
33            TOPOLOGY_TARGET_PREPARATION_PROTOCOL_VERSION | TOPOLOGY_COMMIT_PROTOCOL_VERSION
34        ) {
35            return Err(TopologyError::Protocol(
36                "participant lacks target preparation protocol three".into(),
37            ));
38        }
39        tokio::time::timeout(CONTROL_TIMEOUT, async {
40            for _ in 0..MAX_ADMISSION_ATTEMPTS {
41                let published = self
42                    .load_published_authority_head()
43                    .await?
44                    .ok_or(TopologyError::Fenced)?;
45                let current = &published.record;
46                let index = current
47                    .topology_operations
48                    .iter()
49                    .position(|entry| entry.operation_id == input.operation().operation_id)
50                    .ok_or_else(|| TopologyError::Conflict("unknown topology operation".into()))?;
51                let operation = &current.topology_operations[index];
52                if operation.phase != TopologyAdmissionPhase::CutPrepared
53                    || !current.lease.matches_proof(&operation.admitted_by)
54                    || !operation.same_restore_binding(input.operation())
55                {
56                    return Err(TopologyError::Fenced);
57                }
58                let fresh = self
59                    .topology_restore_input(
60                        assignments,
61                        processes,
62                        operation.operation_id,
63                        input.process(),
64                    )
65                    .await?;
66                if !fresh.same_restore_requirements(input) {
67                    return Err(TopologyError::Fenced);
68                }
69                if let Some(existing) =
70                    fresh
71                        .operation()
72                        .target_preparations
73                        .iter()
74                        .find(|receipt| {
75                            receipt.participant.node_id == input.process().participant.node_id
76                        })
77                {
78                    if existing.participant != input.process().participant
79                        || existing.process_term != input.process().process_term
80                        || existing.protocol_version != protocol_version
81                    {
82                        return Err(TopologyError::Fenced);
83                    }
84                    return Ok(fresh.operation().clone());
85                }
86                self.require_topology_process(processes, input.process())
87                    .await?;
88                let mut lease = current.lease.clone();
89                lease.seq = lease
90                    .seq
91                    .checked_add(1)
92                    .ok_or_else(|| TopologyError::Invalid("authority sequence exhausted".into()))?;
93                let sequence = lease.seq;
94                let mut next = current.preserve_with_lease(lease);
95                next.version =
96                    next.version
97                        .max(if protocol_version == TOPOLOGY_COMMIT_PROTOCOL_VERSION {
98                            TOPOLOGY_COMMIT_RECORD_VERSION
99                        } else {
100                            TOPOLOGY_TARGET_PREPARATION_RECORD_VERSION
101                        });
102                let operation = &mut next.topology_operations[index];
103                operation.status_sequence = sequence;
104                operation
105                    .target_preparations
106                    .push(TopologyTargetPreparationReceipt {
107                        participant: input.process().participant,
108                        process_term: input.process().process_term,
109                        protocol_version,
110                        authority_sequence: sequence,
111                    });
112                operation
113                    .target_preparations
114                    .sort_by_key(|receipt| receipt.participant.node_id);
115                match self
116                    .create_authority_record(Some(&published), &next)
117                    .await?
118                {
119                    AuthorityCreateOutcome::Created | AuthorityCreateOutcome::ExistingIdentical => {
120                        let after = self
121                            .topology_restore_input(
122                                assignments,
123                                processes,
124                                input.operation().operation_id,
125                                input.process(),
126                            )
127                            .await?;
128                        if !after.same_restore_requirements(input) {
129                            return Err(TopologyError::Fenced);
130                        }
131                        return Ok(after.operation().clone());
132                    }
133                    AuthorityCreateOutcome::Contended(_) => tokio::task::yield_now().await,
134                }
135            }
136            Err(TopologyError::Contended)
137        })
138        .await
139        .map_err(|_| TopologyError::Contended)?
140    }
141
142    pub(super) async fn audit_topology_target_preparations(
143        &self,
144        operation: &TopologyAdmissionStatus,
145    ) -> Result<(), LeaseError> {
146        for receipt in &operation.target_preparations {
147            let record = read_authority_record(self.store.as_ref(), receipt.authority_sequence)
148                .await?
149                .ok_or_else(|| {
150                    LeaseError::Invalid("target preparation authority anchor is missing".into())
151                })?;
152            if record.version < TOPOLOGY_TARGET_PREPARATION_RECORD_VERSION
153                || (receipt.protocol_version == TOPOLOGY_COMMIT_PROTOCOL_VERSION
154                    && record.version < TOPOLOGY_COMMIT_RECORD_VERSION)
155                || record
156                    .topology_operations
157                    .iter()
158                    .find(|entry| entry.operation_id == operation.operation_id)
159                    .is_none_or(|anchored| {
160                        // An abort retains the historical prepared observation under the old term.
161                        anchored.phase != TopologyAdmissionPhase::CutPrepared
162                            || anchored.operation_id != operation.operation_id
163                            || anchored.plan != operation.plan
164                            || anchored.admitted_by != operation.admitted_by
165                            || anchored.admitted_sequence != operation.admitted_sequence
166                            || anchored.cut != operation.cut
167                            || anchored.preparation != operation.preparation
168                            || anchored.migration_root != operation.migration_root
169                            || anchored.status_sequence != receipt.authority_sequence
170                            || !anchored.target_preparations.contains(receipt)
171                    })
172            {
173                return Err(LeaseError::Invalid(
174                    "target preparation differs from its immutable authority append".into(),
175                ));
176            }
177        }
178        Ok(())
179    }
180}
laminar_core/cluster/control/leader_lease/topology_target_preparation.rs

laminar_core/cluster/control/leader_lease/
topology_target_preparation.rs
