laminar_core/cluster/control/
namespace_proof.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-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/state namespace verification.
21pub const MAX_SHARED_NAMESPACE_PROOF_TIMEOUT: Duration = Duration::from_secs(60);
22
23/// Exact object-store handles admitted by a successful cluster namespace proof.
24#[derive(Clone)]
25pub struct VerifiedClusterNamespaces {
26    checkpoint: Arc<dyn ObjectStore>,
27    state: Arc<dyn ObjectStore>,
28    local: CheckpointParticipant,
29}
30
31impl std::fmt::Debug for VerifiedClusterNamespaces {
32    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
33        formatter
34            .debug_struct("VerifiedClusterNamespaces")
35            .finish_non_exhaustive()
36    }
37}
38
39impl VerifiedClusterNamespaces {
40    /// Return the exact checkpoint store handle covered by the proof.
41    #[must_use]
42    pub fn checkpoint_store(&self) -> Arc<dyn ObjectStore> {
43        Arc::clone(&self.checkpoint)
44    }
45
46    /// Return the exact state store handle covered by the proof.
47    #[must_use]
48    pub fn state_store(&self) -> Arc<dyn ObjectStore> {
49        Arc::clone(&self.state)
50    }
51
52    /// Return the process identity that produced this proof.
53    #[must_use]
54    pub const fn local_participant(&self) -> CheckpointParticipant {
55        self.local
56    }
57}
58
59/// Failure to establish one checkpoint/state namespace across the startup roster.
60#[derive(Debug, thiserror::Error)]
61#[error("{message}")]
62pub struct NamespaceProofError {
63    message: String,
64}
65
66impl NamespaceProofError {
67    fn configuration(message: &'static str) -> Self {
68        Self {
69            message: message.to_string(),
70        }
71    }
72
73    fn verification(message: &str) -> Self {
74        Self {
75            message: format!("shared checkpoint/state namespace proof failed: {message}"),
76        }
77    }
78
79    fn timeout(timeout: Duration) -> Self {
80        Self {
81            message: format!("shared checkpoint/state namespace proof exceeded {timeout:?}"),
82        }
83    }
84}
85
86#[derive(Deserialize, Serialize)]
87#[serde(deny_unknown_fields)]
88struct NamespaceProofRecord {
89    version: u8,
90    node_id: u64,
91    boot_incarnation: uuid::Uuid,
92    nonce: uuid::Uuid,
93    roster_sha256: String,
94}
95
96impl NamespaceProofRecord {
97    fn validate_identity(&self, participant: CheckpointParticipant) -> Result<(), String> {
98        if self.version != NAMESPACE_PROOF_VERSION
99            || self.node_id != participant.node_id
100            || self.boot_incarnation != participant.boot_incarnation
101            || self.nonce.is_nil()
102            || self.roster_sha256.len() != 64
103            || !self
104                .roster_sha256
105                .bytes()
106                .all(|byte| byte.is_ascii_digit() || (b'a'..=b'f').contains(&byte))
107        {
108            return Err(format!(
109                "node {} published a stale or mismatched shared-namespace proof",
110                participant.node_id
111            ));
112        }
113        Ok(())
114    }
115}
116
117#[derive(Clone, Copy)]
118enum NamespaceProofStore {
119    Checkpoint,
120    State,
121}
122
123impl NamespaceProofStore {
124    const fn name(self) -> &'static str {
125        match self {
126            Self::Checkpoint => "checkpoint",
127            Self::State => "state",
128        }
129    }
130}
131
132fn namespace_proof_roster_sha256(participants: &[CheckpointParticipant]) -> String {
133    const HEX: &[u8; 16] = b"0123456789abcdef";
134
135    let mut hash = Sha256::new();
136    hash.update(b"LAMINAR_SHARED_NAMESPACE_ROSTER_V1\0");
137    hash.update(
138        u64::try_from(participants.len())
139            .unwrap_or(u64::MAX)
140            .to_be_bytes(),
141    );
142    for participant in participants {
143        hash.update(participant.node_id.to_be_bytes());
144        hash.update(participant.boot_incarnation.as_bytes());
145    }
146    let digest = hash.finalize();
147    let mut encoded = String::with_capacity(digest.len() * 2);
148    for byte in digest {
149        encoded.push(HEX[(byte >> 4) as usize] as char);
150        encoded.push(HEX[(byte & 0x0f) as usize] as char);
151    }
152    encoded
153}
154
155fn namespace_proof_path(store: NamespaceProofStore, node_id: u64) -> object_store::path::Path {
156    object_store::path::Path::from(format!(
157        "cluster-namespace-proof/v1/{}/node={node_id}/sentinel",
158        store.name()
159    ))
160}
161
162fn namespace_proof_sentinel(
163    store: NamespaceProofStore,
164    record: &NamespaceProofRecord,
165) -> bytes::Bytes {
166    bytes::Bytes::from(format!(
167        "LAMINAR_SHARED_NAMESPACE_V1\n{}\n{}\n{}\n{}\n{}\n",
168        store.name(),
169        record.node_id,
170        record.boot_incarnation,
171        record.nonce,
172        record.roster_sha256
173    ))
174}
175
176async fn write_namespace_proof_sentinel(
177    object_store: &Arc<dyn ObjectStore>,
178    store: NamespaceProofStore,
179    record: &NamespaceProofRecord,
180) -> Result<(), String> {
181    let payload = namespace_proof_sentinel(store, record);
182    if u64::try_from(payload.len()).unwrap_or(u64::MAX) > NAMESPACE_PROOF_MAX_SENTINEL_BYTES {
183        return Err("shared-namespace sentinel exceeds its fixed size bound".into());
184    }
185    object_store
186        .put(
187            &namespace_proof_path(store, record.node_id),
188            object_store::PutPayload::from(payload),
189        )
190        .await
191        .map(|_| ())
192        .map_err(|error| format!("write {} namespace sentinel: {error}", store.name()))
193}
194
195async fn read_namespace_proof_sentinel(
196    object_store: &Arc<dyn ObjectStore>,
197    store: NamespaceProofStore,
198    record: &NamespaceProofRecord,
199) -> Result<(), String> {
200    let result = object_store
201        .get(&namespace_proof_path(store, record.node_id))
202        .await
203        .map_err(|error| {
204            format!(
205                "read node {} {} namespace sentinel: {error}",
206                record.node_id,
207                store.name()
208            )
209        })?;
210    if result.meta.size == 0 || result.meta.size > NAMESPACE_PROOF_MAX_SENTINEL_BYTES {
211        return Err(format!(
212            "node {} {} namespace sentinel is {} bytes; maximum is {}",
213            record.node_id,
214            store.name(),
215            result.meta.size,
216            NAMESPACE_PROOF_MAX_SENTINEL_BYTES
217        ));
218    }
219    let bytes = result.bytes().await.map_err(|error| {
220        format!(
221            "read node {} {} namespace sentinel body: {error}",
222            record.node_id,
223            store.name()
224        )
225    })?;
226    if bytes != namespace_proof_sentinel(store, record) {
227        return Err(format!(
228            "node {} {} namespace sentinel does not match its boot proof",
229            record.node_id,
230            store.name()
231        ));
232    }
233    Ok(())
234}
235
236async fn verify_namespace_proof_visibility(
237    control: &Arc<dyn ClusterKv>,
238    checkpoint_store: &Arc<dyn ObjectStore>,
239    state_store: &Arc<dyn ObjectStore>,
240    participants: &[CheckpointParticipant],
241    local: CheckpointParticipant,
242    roster_sha256: &str,
243) -> Result<(), String> {
244    let checks = futures::stream::iter(participants.iter().copied())
245        .map(|participant| async move {
246            let encoded = control
247                .read_from_checked(NodeId(participant.node_id), NAMESPACE_PROOF_KEY)
248                .await?
249                .ok_or_else(|| {
250                    format!(
251                        "node {} has not published its shared-namespace proof",
252                        participant.node_id
253                    )
254                })?;
255            if encoded.len() > NAMESPACE_PROOF_MAX_RECORD_BYTES {
256                return Err(format!(
257                    "node {} shared-namespace proof exceeds {} bytes",
258                    participant.node_id, NAMESPACE_PROOF_MAX_RECORD_BYTES
259                ));
260            }
261            let record: NamespaceProofRecord = serde_json::from_str(&encoded).map_err(|error| {
262                format!(
263                    "decode node {} shared-namespace proof: {error}",
264                    participant.node_id
265                )
266            })?;
267            record.validate_identity(participant)?;
268            if participant == local && record.roster_sha256 != roster_sha256 {
269                return Err("local shared-namespace proof has the wrong startup roster".into());
270            }
271            tokio::try_join!(
272                read_namespace_proof_sentinel(
273                    checkpoint_store,
274                    NamespaceProofStore::Checkpoint,
275                    &record,
276                ),
277                read_namespace_proof_sentinel(state_store, NamespaceProofStore::State, &record),
278            )?;
279            Ok::<_, String>(())
280        })
281        .buffer_unordered(NAMESPACE_PROOF_READ_CONCURRENCY);
282    let results: Vec<Result<(), String>> = checks.collect().await;
283    for result in results {
284        result?;
285    }
286    Ok(())
287}
288
289async fn wait_for_namespace_proof_visibility(
290    control: &Arc<dyn ClusterKv>,
291    checkpoint_store: &Arc<dyn ObjectStore>,
292    state_store: &Arc<dyn ObjectStore>,
293    participants: &[CheckpointParticipant],
294    local: CheckpointParticipant,
295    roster_sha256: &str,
296    deadline: tokio::time::Instant,
297) -> Result<(), String> {
298    let mut last_failure = "no verification attempt completed".to_string();
299    loop {
300        let remaining = deadline.saturating_duration_since(tokio::time::Instant::now());
301        if remaining.is_zero() {
302            return Err(format!(
303                "shared-namespace peer visibility timed out: {last_failure}"
304            ));
305        }
306        match tokio::time::timeout(
307            remaining,
308            verify_namespace_proof_visibility(
309                control,
310                checkpoint_store,
311                state_store,
312                participants,
313                local,
314                roster_sha256,
315            ),
316        )
317        .await
318        {
319            Ok(Ok(())) => return Ok(()),
320            Ok(Err(error)) => last_failure = error,
321            Err(_) => {
322                return Err(format!(
323                    "shared-namespace peer visibility timed out: {last_failure}"
324                ));
325            }
326        }
327        tokio::time::sleep(
328            NAMESPACE_PROOF_RETRY_INTERVAL
329                .min(deadline.saturating_duration_since(tokio::time::Instant::now())),
330        )
331        .await;
332    }
333}
334
335/// Prove that the local process can read every startup participant's retained checkpoint and
336/// state sentinel through its candidate object-store handles.
337///
338/// # Errors
339///
340/// Returns an error for an invalid roster, a visibility failure, or a proof timeout.
341pub async fn prove_shared_object_store_namespaces(
342    local: CheckpointParticipant,
343    participants: &[CheckpointParticipant],
344    control: Arc<dyn ClusterKv>,
345    checkpoint_store: Arc<dyn ObjectStore>,
346    state_store: Arc<dyn ObjectStore>,
347    timeout: Duration,
348) -> Result<VerifiedClusterNamespaces, NamespaceProofError> {
349    if participants.is_empty()
350        || participants.len() > MAX_CHECKPOINT_PARTICIPANTS
351        || participants
352            .windows(2)
353            .any(|pair| pair[0].node_id >= pair[1].node_id)
354        || participants
355            .iter()
356            .any(|participant| participant.node_id == 0 || participant.boot_incarnation.is_nil())
357        || participants
358            .iter()
359            .filter(|participant| **participant == local)
360            .count()
361            != 1
362    {
363        return Err(NamespaceProofError::configuration(
364            "shared-namespace proof requires one canonical exact startup roster",
365        ));
366    }
367    let timeout = timeout.min(MAX_SHARED_NAMESPACE_PROOF_TIMEOUT);
368    if timeout.is_zero() {
369        return Err(NamespaceProofError::configuration(
370            "shared-namespace proof timeout is zero",
371        ));
372    }
373    let roster_sha256 = namespace_proof_roster_sha256(participants);
374    let record = NamespaceProofRecord {
375        version: NAMESPACE_PROOF_VERSION,
376        node_id: local.node_id,
377        boot_incarnation: local.boot_incarnation,
378        nonce: uuid::Uuid::new_v4(),
379        roster_sha256: roster_sha256.clone(),
380    };
381    let deadline = tokio::time::Instant::now() + timeout;
382    let proof = async {
383        tokio::try_join!(
384            write_namespace_proof_sentinel(
385                &checkpoint_store,
386                NamespaceProofStore::Checkpoint,
387                &record,
388            ),
389            write_namespace_proof_sentinel(&state_store, NamespaceProofStore::State, &record),
390        )?;
391        let encoded = serde_json::to_string(&record).map_err(|error| error.to_string())?;
392        if encoded.len() > NAMESPACE_PROOF_MAX_RECORD_BYTES {
393            return Err("local shared-namespace proof exceeds its size bound".to_string());
394        }
395        control.write_checked(NAMESPACE_PROOF_KEY, encoded).await?;
396        wait_for_namespace_proof_visibility(
397            &control,
398            &checkpoint_store,
399            &state_store,
400            participants,
401            local,
402            &roster_sha256,
403            deadline,
404        )
405        .await
406    };
407    match tokio::time::timeout(timeout, proof).await {
408        Ok(Ok(())) => Ok(VerifiedClusterNamespaces {
409            checkpoint: checkpoint_store,
410            state: state_store,
411            local,
412        }),
413        Ok(Err(error)) => Err(NamespaceProofError::verification(&error)),
414        Err(_) => Err(NamespaceProofError::timeout(timeout)),
415    }
416}
417
418#[cfg(test)]
419mod tests {
420    use std::collections::HashMap;
421
422    use async_trait::async_trait;
423    use parking_lot::Mutex;
424
425    use super::*;
426
427    #[derive(Clone)]
428    struct NamespaceProofTestKv {
429        local_id: NodeId,
430        values: Arc<Mutex<HashMap<(NodeId, String), String>>>,
431    }
432
433    #[async_trait]
434    impl ClusterKv for NamespaceProofTestKv {
435        async fn write(&self, key: &str, value: String) {
436            self.values
437                .lock()
438                .insert((self.local_id, key.to_string()), value);
439        }
440
441        async fn read_from(&self, who: NodeId, key: &str) -> Option<String> {
442            self.values.lock().get(&(who, key.to_string())).cloned()
443        }
444
445        async fn scan(&self, key: &str) -> Vec<(NodeId, String)> {
446            self.values
447                .lock()
448                .iter()
449                .filter(|((_, stored_key), _)| stored_key == key)
450                .map(|((node, _), value)| (*node, value.clone()))
451                .collect()
452        }
453    }
454
455    fn participants() -> [CheckpointParticipant; 2] {
456        [
457            CheckpointParticipant {
458                node_id: 1,
459                boot_incarnation: uuid::Uuid::from_u128(11),
460            },
461            CheckpointParticipant {
462                node_id: 2,
463                boot_incarnation: uuid::Uuid::from_u128(22),
464            },
465        ]
466    }
467
468    fn controls() -> [Arc<dyn ClusterKv>; 2] {
469        let values = Arc::new(Mutex::new(HashMap::new()));
470        [
471            Arc::new(NamespaceProofTestKv {
472                local_id: NodeId(1),
473                values: Arc::clone(&values),
474            }),
475            Arc::new(NamespaceProofTestKv {
476                local_id: NodeId(2),
477                values,
478            }),
479        ]
480    }
481
482    async fn run_two_node_proof(
483        participants: &[CheckpointParticipant; 2],
484        controls: &[Arc<dyn ClusterKv>; 2],
485        checkpoint_stores: [Arc<dyn ObjectStore>; 2],
486        state_stores: [Arc<dyn ObjectStore>; 2],
487        timeout: Duration,
488    ) -> [Result<VerifiedClusterNamespaces, NamespaceProofError>; 2] {
489        let first = prove_shared_object_store_namespaces(
490            participants[0],
491            participants,
492            Arc::clone(&controls[0]),
493            Arc::clone(&checkpoint_stores[0]),
494            Arc::clone(&state_stores[0]),
495            timeout,
496        );
497        let second = prove_shared_object_store_namespaces(
498            participants[1],
499            participants,
500            Arc::clone(&controls[1]),
501            Arc::clone(&checkpoint_stores[1]),
502            Arc::clone(&state_stores[1]),
503            timeout,
504        );
505        let (first, second) = tokio::join!(first, second);
506        [first, second]
507    }
508
509    async fn marker_count(store: &Arc<dyn ObjectStore>, role: NamespaceProofStore) -> usize {
510        let prefix =
511            object_store::path::Path::from(format!("cluster-namespace-proof/v1/{}/", role.name()));
512        let mut entries = store.list(Some(&prefix));
513        let mut count = 0;
514        while let Some(entry) = entries.next().await {
515            entry.unwrap();
516            count += 1;
517        }
518        count
519    }
520
521    #[tokio::test]
522    async fn proof_retains_bounded_markers_and_exact_handles() {
523        let checkpoint: Arc<dyn ObjectStore> = Arc::new(object_store::memory::InMemory::new());
524        let state: Arc<dyn ObjectStore> = Arc::new(object_store::memory::InMemory::new());
525        let participants = participants();
526        let controls = controls();
527        let results = run_two_node_proof(
528            &participants,
529            &controls,
530            [Arc::clone(&checkpoint), Arc::clone(&checkpoint)],
531            [Arc::clone(&state), Arc::clone(&state)],
532            Duration::from_secs(2),
533        )
534        .await;
535        assert!(results.iter().all(Result::is_ok));
536
537        let admitted = results[0].as_ref().unwrap();
538        assert!(Arc::ptr_eq(&admitted.checkpoint_store(), &checkpoint));
539        assert!(Arc::ptr_eq(&admitted.state_store(), &state));
540        assert_eq!(admitted.local_participant(), participants[0]);
541        for node_id in [1, 2] {
542            for (store, role) in [
543                (&checkpoint, NamespaceProofStore::Checkpoint),
544                (&state, NamespaceProofStore::State),
545            ] {
546                let marker = store
547                    .get(&namespace_proof_path(role, node_id))
548                    .await
549                    .expect("boot marker must remain available to rolling joiners");
550                assert!(marker.meta.size <= NAMESPACE_PROOF_MAX_SENTINEL_BYTES);
551            }
552        }
553    }
554
555    #[tokio::test]
556    async fn rolling_restart_uses_active_peers_retained_markers() {
557        let checkpoint: Arc<dyn ObjectStore> = Arc::new(object_store::memory::InMemory::new());
558        let state: Arc<dyn ObjectStore> = Arc::new(object_store::memory::InMemory::new());
559        let initial = participants();
560        let controls = controls();
561        let results = run_two_node_proof(
562            &initial,
563            &controls,
564            [Arc::clone(&checkpoint), Arc::clone(&checkpoint)],
565            [Arc::clone(&state), Arc::clone(&state)],
566            Duration::from_secs(2),
567        )
568        .await;
569        assert!(results.iter().all(Result::is_ok));
570        let active_peer_record = controls[1]
571            .read_from(NodeId(2), NAMESPACE_PROOF_KEY)
572            .await
573            .unwrap();
574
575        let restarted = [
576            CheckpointParticipant {
577                node_id: 1,
578                boot_incarnation: uuid::Uuid::from_u128(111),
579            },
580            initial[1],
581        ];
582        prove_shared_object_store_namespaces(
583            restarted[0],
584            &restarted,
585            Arc::clone(&controls[0]),
586            Arc::clone(&checkpoint),
587            Arc::clone(&state),
588            Duration::from_secs(1),
589        )
590        .await
591        .unwrap();
592        assert_eq!(
593            controls[1].read_from(NodeId(2), NAMESPACE_PROOF_KEY).await,
594            Some(active_peer_record),
595            "the active peer must not rerun startup for a rolling joiner"
596        );
597        assert_eq!(
598            marker_count(&checkpoint, NamespaceProofStore::Checkpoint).await,
599            2
600        );
601        assert_eq!(marker_count(&state, NamespaceProofStore::State).await, 2);
602    }
603
604    #[tokio::test]
605    async fn proof_rejects_split_state_namespaces() {
606        let checkpoint: Arc<dyn ObjectStore> = Arc::new(object_store::memory::InMemory::new());
607        let state_a: Arc<dyn ObjectStore> = Arc::new(object_store::memory::InMemory::new());
608        let state_b: Arc<dyn ObjectStore> = Arc::new(object_store::memory::InMemory::new());
609        let participants = participants();
610        let controls = controls();
611        let results = run_two_node_proof(
612            &participants,
613            &controls,
614            [Arc::clone(&checkpoint), checkpoint],
615            [state_a, state_b],
616            Duration::from_millis(250),
617        )
618        .await;
619        assert!(results.iter().all(Result::is_err));
620    }
621
622    #[tokio::test]
623    async fn proof_rejects_split_checkpoint_namespaces() {
624        let checkpoint_a: Arc<dyn ObjectStore> = Arc::new(object_store::memory::InMemory::new());
625        let checkpoint_b: Arc<dyn ObjectStore> = Arc::new(object_store::memory::InMemory::new());
626        let state: Arc<dyn ObjectStore> = Arc::new(object_store::memory::InMemory::new());
627        let participants = participants();
628        let controls = controls();
629        let results = run_two_node_proof(
630            &participants,
631            &controls,
632            [checkpoint_a, checkpoint_b],
633            [Arc::clone(&state), state],
634            Duration::from_millis(250),
635        )
636        .await;
637        assert!(results.iter().all(Result::is_err));
638    }
639}
laminar_core/cluster/control/namespace_proof.rs

laminar_core/cluster/control/
namespace_proof.rs
