laminar_core/cluster/control/namespace_proof/
mod.rs

1use std::sync::Arc;
2use std::time::Duration;
3
4use futures::StreamExt;
5use object_store::{ObjectStore, ObjectStoreExt};
6use serde::{Deserialize, Serialize};
7use sha2::{Digest, Sha256};
8
9use super::ClusterKv;
10use crate::checkpoint::{CheckpointParticipant, MAX_CHECKPOINT_PARTICIPANTS};
11use crate::cluster::discovery::NodeId;
12
13const NAMESPACE_PROOF_KEY: &str = "control:shared-checkpoint-namespace-proof-v1";
14const NAMESPACE_PROOF_VERSION: u8 = 1;
15const NAMESPACE_PROOF_MAX_RECORD_BYTES: usize = 512;
16const NAMESPACE_PROOF_MAX_SENTINEL_BYTES: u64 = 512;
17const NAMESPACE_PROOF_RETRY_INTERVAL: Duration = Duration::from_millis(100);
18const NAMESPACE_PROOF_READ_CONCURRENCY: usize = 16;
19
20/// Maximum time allowed for shared checkpoint namespace verification.
21pub const MAX_SHARED_NAMESPACE_PROOF_TIMEOUT: Duration = Duration::from_secs(60);
22
23/// Exact checkpoint object-store handle admitted by a successful cluster namespace proof.
24#[derive(Clone)]
25pub struct VerifiedClusterNamespaces {
26    checkpoint: Arc<dyn ObjectStore>,
27    local: CheckpointParticipant,
28}
29
30impl std::fmt::Debug for VerifiedClusterNamespaces {
31    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
32        formatter
33            .debug_struct("VerifiedClusterNamespaces")
34            .finish_non_exhaustive()
35    }
36}
37
38impl VerifiedClusterNamespaces {
39    /// Return the exact checkpoint store handle covered by the proof.
40    #[must_use]
41    pub fn checkpoint_store(&self) -> Arc<dyn ObjectStore> {
42        Arc::clone(&self.checkpoint)
43    }
44
45    /// Return the process identity that produced this proof.
46    #[must_use]
47    pub const fn local_participant(&self) -> CheckpointParticipant {
48        self.local
49    }
50}
51
52/// Failure to establish one checkpoint namespace across the startup roster.
53#[derive(Debug, thiserror::Error)]
54#[error("{message}")]
55pub struct NamespaceProofError {
56    message: String,
57}
58
59impl NamespaceProofError {
60    fn configuration(message: &'static str) -> Self {
61        Self {
62            message: message.to_string(),
63        }
64    }
65
66    fn verification(message: &str) -> Self {
67        Self {
68            message: format!("shared checkpoint namespace proof failed: {message}"),
69        }
70    }
71
72    fn timeout(timeout: Duration) -> Self {
73        Self {
74            message: format!("shared checkpoint namespace proof exceeded {timeout:?}"),
75        }
76    }
77}
78
79#[derive(Deserialize, Serialize)]
80#[serde(deny_unknown_fields)]
81struct NamespaceProofRecord {
82    version: u8,
83    node_id: u64,
84    boot_incarnation: uuid::Uuid,
85    nonce: uuid::Uuid,
86    roster_sha256: String,
87}
88
89impl NamespaceProofRecord {
90    fn validate_identity(&self, participant: CheckpointParticipant) -> Result<(), String> {
91        if self.version != NAMESPACE_PROOF_VERSION
92            || self.node_id != participant.node_id
93            || self.boot_incarnation != participant.boot_incarnation
94            || self.nonce.is_nil()
95            || self.roster_sha256.len() != 64
96            || !self
97                .roster_sha256
98                .bytes()
99                .all(|byte| byte.is_ascii_digit() || (b'a'..=b'f').contains(&byte))
100        {
101            return Err(format!(
102                "node {} published a stale or mismatched shared checkpoint namespace proof",
103                participant.node_id
104            ));
105        }
106        Ok(())
107    }
108}
109
110fn namespace_proof_roster_sha256(participants: &[CheckpointParticipant]) -> String {
111    const HEX: &[u8; 16] = b"0123456789abcdef";
112
113    let mut hash = Sha256::new();
114    hash.update(b"LAMINAR_SHARED_CHECKPOINT_NAMESPACE_ROSTER_V1\0");
115    hash.update(
116        u64::try_from(participants.len())
117            .unwrap_or(u64::MAX)
118            .to_be_bytes(),
119    );
120    for participant in participants {
121        hash.update(participant.node_id.to_be_bytes());
122        hash.update(participant.boot_incarnation.as_bytes());
123    }
124    let digest = hash.finalize();
125    let mut encoded = String::with_capacity(digest.len() * 2);
126    for byte in digest {
127        encoded.push(HEX[(byte >> 4) as usize] as char);
128        encoded.push(HEX[(byte & 0x0f) as usize] as char);
129    }
130    encoded
131}
132
133fn namespace_proof_path(node_id: u64) -> object_store::path::Path {
134    object_store::path::Path::from(format!(
135        "cluster-checkpoint-namespace-proof/v1/node={node_id}/sentinel"
136    ))
137}
138
139fn namespace_proof_sentinel(record: &NamespaceProofRecord) -> bytes::Bytes {
140    bytes::Bytes::from(format!(
141        "LAMINAR_SHARED_CHECKPOINT_NAMESPACE_V1\n{}\n{}\n{}\n{}\n",
142        record.node_id, record.boot_incarnation, record.nonce, record.roster_sha256
143    ))
144}
145
146async fn write_namespace_proof_sentinel(
147    object_store: &Arc<dyn ObjectStore>,
148    record: &NamespaceProofRecord,
149) -> Result<(), String> {
150    let payload = namespace_proof_sentinel(record);
151    if u64::try_from(payload.len()).unwrap_or(u64::MAX) > NAMESPACE_PROOF_MAX_SENTINEL_BYTES {
152        return Err("shared checkpoint namespace sentinel exceeds its fixed size bound".into());
153    }
154    object_store
155        .put(
156            &namespace_proof_path(record.node_id),
157            object_store::PutPayload::from(payload),
158        )
159        .await
160        .map(|_| ())
161        .map_err(|error| format!("write checkpoint namespace sentinel: {error}"))
162}
163
164async fn read_namespace_proof_sentinel(
165    object_store: &Arc<dyn ObjectStore>,
166    record: &NamespaceProofRecord,
167) -> Result<(), String> {
168    let result = object_store
169        .get(&namespace_proof_path(record.node_id))
170        .await
171        .map_err(|error| {
172            format!(
173                "read node {} checkpoint namespace sentinel: {error}",
174                record.node_id
175            )
176        })?;
177    if result.meta.size == 0 || result.meta.size > NAMESPACE_PROOF_MAX_SENTINEL_BYTES {
178        return Err(format!(
179            "node {} checkpoint namespace sentinel is {} bytes; maximum is {}",
180            record.node_id, result.meta.size, NAMESPACE_PROOF_MAX_SENTINEL_BYTES
181        ));
182    }
183    let bytes = result.bytes().await.map_err(|error| {
184        format!(
185            "read node {} checkpoint namespace sentinel body: {error}",
186            record.node_id
187        )
188    })?;
189    if bytes != namespace_proof_sentinel(record) {
190        return Err(format!(
191            "node {} checkpoint namespace sentinel does not match its boot proof",
192            record.node_id
193        ));
194    }
195    Ok(())
196}
197
198async fn verify_namespace_proof_visibility(
199    control: &Arc<dyn ClusterKv>,
200    checkpoint_store: &Arc<dyn ObjectStore>,
201    participants: &[CheckpointParticipant],
202    local: CheckpointParticipant,
203    roster_sha256: &str,
204) -> Result<(), String> {
205    let checks = futures::stream::iter(participants.iter().copied())
206        .map(|participant| async move {
207            let encoded = control
208                .read_from_checked(NodeId(participant.node_id), NAMESPACE_PROOF_KEY)
209                .await?
210                .ok_or_else(|| {
211                    format!(
212                        "node {} has not published its shared checkpoint namespace proof",
213                        participant.node_id
214                    )
215                })?;
216            if encoded.len() > NAMESPACE_PROOF_MAX_RECORD_BYTES {
217                return Err(format!(
218                    "node {} shared checkpoint namespace proof exceeds {} bytes",
219                    participant.node_id, NAMESPACE_PROOF_MAX_RECORD_BYTES
220                ));
221            }
222            let record: NamespaceProofRecord = serde_json::from_str(&encoded).map_err(|error| {
223                format!(
224                    "decode node {} shared checkpoint namespace proof: {error}",
225                    participant.node_id
226                )
227            })?;
228            record.validate_identity(participant)?;
229            if participant == local && record.roster_sha256 != roster_sha256 {
230                return Err(
231                    "local shared checkpoint namespace proof has the wrong startup roster".into(),
232                );
233            }
234            read_namespace_proof_sentinel(checkpoint_store, &record).await?;
235            Ok::<_, String>(())
236        })
237        .buffer_unordered(NAMESPACE_PROOF_READ_CONCURRENCY);
238    let results: Vec<Result<(), String>> = checks.collect().await;
239    for result in results {
240        result?;
241    }
242    Ok(())
243}
244
245async fn wait_for_namespace_proof_visibility(
246    control: &Arc<dyn ClusterKv>,
247    checkpoint_store: &Arc<dyn ObjectStore>,
248    participants: &[CheckpointParticipant],
249    local: CheckpointParticipant,
250    roster_sha256: &str,
251    deadline: tokio::time::Instant,
252) -> Result<(), String> {
253    let mut last_failure = "no verification attempt completed".to_string();
254    loop {
255        let remaining = deadline.saturating_duration_since(tokio::time::Instant::now());
256        if remaining.is_zero() {
257            return Err(format!(
258                "shared checkpoint namespace peer visibility timed out: {last_failure}"
259            ));
260        }
261        match tokio::time::timeout(
262            remaining,
263            verify_namespace_proof_visibility(
264                control,
265                checkpoint_store,
266                participants,
267                local,
268                roster_sha256,
269            ),
270        )
271        .await
272        {
273            Ok(Ok(())) => return Ok(()),
274            Ok(Err(error)) => last_failure = error,
275            Err(_) => {
276                return Err(format!(
277                    "shared checkpoint namespace peer visibility timed out: {last_failure}"
278                ));
279            }
280        }
281        tokio::time::sleep(
282            NAMESPACE_PROOF_RETRY_INTERVAL
283                .min(deadline.saturating_duration_since(tokio::time::Instant::now())),
284        )
285        .await;
286    }
287}
288
289/// Prove that the local process can read every startup participant's retained checkpoint
290/// sentinel through its candidate object-store handle.
291///
292/// # Errors
293///
294/// Returns an error for an invalid roster, a visibility failure, or a proof timeout.
295pub async fn prove_shared_object_store_namespaces(
296    local: CheckpointParticipant,
297    participants: &[CheckpointParticipant],
298    control: Arc<dyn ClusterKv>,
299    checkpoint_store: Arc<dyn ObjectStore>,
300    timeout: Duration,
301) -> Result<VerifiedClusterNamespaces, NamespaceProofError> {
302    if participants.is_empty()
303        || participants.len() > MAX_CHECKPOINT_PARTICIPANTS
304        || participants
305            .windows(2)
306            .any(|pair| pair[0].node_id >= pair[1].node_id)
307        || participants
308            .iter()
309            .any(|participant| participant.node_id == 0 || participant.boot_incarnation.is_nil())
310        || participants
311            .iter()
312            .filter(|participant| **participant == local)
313            .count()
314            != 1
315    {
316        return Err(NamespaceProofError::configuration(
317            "shared checkpoint namespace proof requires one canonical exact startup roster",
318        ));
319    }
320    let timeout = timeout.min(MAX_SHARED_NAMESPACE_PROOF_TIMEOUT);
321    if timeout.is_zero() {
322        return Err(NamespaceProofError::configuration(
323            "shared checkpoint namespace proof timeout is zero",
324        ));
325    }
326    let roster_sha256 = namespace_proof_roster_sha256(participants);
327    let record = NamespaceProofRecord {
328        version: NAMESPACE_PROOF_VERSION,
329        node_id: local.node_id,
330        boot_incarnation: local.boot_incarnation,
331        nonce: uuid::Uuid::new_v4(),
332        roster_sha256: roster_sha256.clone(),
333    };
334    let deadline = tokio::time::Instant::now() + timeout;
335    let proof = async {
336        write_namespace_proof_sentinel(&checkpoint_store, &record).await?;
337        let encoded = serde_json::to_string(&record).map_err(|error| error.to_string())?;
338        if encoded.len() > NAMESPACE_PROOF_MAX_RECORD_BYTES {
339            return Err(
340                "local shared checkpoint namespace proof exceeds its size bound".to_string(),
341            );
342        }
343        control.write_checked(NAMESPACE_PROOF_KEY, encoded).await?;
344        wait_for_namespace_proof_visibility(
345            &control,
346            &checkpoint_store,
347            participants,
348            local,
349            &roster_sha256,
350            deadline,
351        )
352        .await
353    };
354    match tokio::time::timeout(timeout, proof).await {
355        Ok(Ok(())) => Ok(VerifiedClusterNamespaces {
356            checkpoint: checkpoint_store,
357            local,
358        }),
359        Ok(Err(error)) => Err(NamespaceProofError::verification(&error)),
360        Err(_) => Err(NamespaceProofError::timeout(timeout)),
361    }
362}
363
364#[cfg(test)]
365mod tests;
laminar_core/cluster/control/namespace_proof/mod.rs

laminar_core/cluster/control/namespace_proof/
mod.rs
