laminar_core/checkpoint_decision/
deployment.rs

1//! Durable deployment identity creation and validation.
2
3use super::{
4    CheckpointDecisionStore, DecisionError, DecisionStoreUpdateMode, DeploymentIdentity,
5    ObjectStore, PutMode, PutOptions, PutPayload, UpdateVersion, VersionedCheckpointIdHead,
6    DEPLOYMENT_IDENTITY_MAX_BYTES, DEPLOYMENT_IDENTITY_VERSION,
7};
8
9impl CheckpointDecisionStore {
10    fn validate_deployment_identity(
11        identity: &DeploymentIdentity,
12    ) -> Result<String, DecisionError> {
13        if identity.version != DEPLOYMENT_IDENTITY_VERSION {
14            return Err(DecisionError::Conflict(format!(
15                "deployment identity version {} is unsupported (expected \
16                 {DEPLOYMENT_IDENTITY_VERSION})",
17                identity.version
18            )));
19        }
20        let parsed = uuid::Uuid::parse_str(&identity.id).map_err(|error| {
21            DecisionError::Conflict(format!("deployment identity is not a UUID: {error}"))
22        })?;
23        let canonical = parsed.to_string();
24        if canonical != identity.id || parsed.is_nil() {
25            return Err(DecisionError::Conflict(
26                "deployment identity must be a canonical non-nil UUID".into(),
27            ));
28        }
29        let allocation_id = uuid::Uuid::parse_str(&identity.allocation_id).map_err(|error| {
30            DecisionError::Conflict(format!(
31                "deployment identity has invalid allocation identity: {error}"
32            ))
33        })?;
34        if allocation_id.is_nil() || allocation_id.to_string() != identity.allocation_id {
35            return Err(DecisionError::Conflict(
36                "deployment identity must use a canonical non-nil allocation identity".into(),
37            ));
38        }
39        Ok(canonical)
40    }
41
42    pub(super) async fn read_deployment_identity(
43        &self,
44    ) -> Result<Option<VersionedCheckpointIdHead>, DecisionError> {
45        let Some(result) = self
46            .get_control_record(
47                &Self::deployment_identity_path(),
48                "deployment identity",
49                DEPLOYMENT_IDENTITY_MAX_BYTES,
50            )
51            .await?
52        else {
53            return Ok(None);
54        };
55        let update_version = UpdateVersion {
56            e_tag: result.meta.e_tag.clone(),
57            version: result.meta.version.clone(),
58        };
59        self.require_native_cas_token("deployment identity", &update_version)?;
60        let bytes = Self::read_control_record_bytes(
61            result,
62            "deployment identity",
63            DEPLOYMENT_IDENTITY_MAX_BYTES,
64            None,
65        )
66        .await?;
67        let identity: DeploymentIdentity = serde_json::from_slice(&bytes)
68            .map_err(|error| DecisionError::Conflict(format!("deployment identity: {error}")))?;
69        Self::validate_deployment_identity(&identity)?;
70        if identity.allocator_mode != self.update_mode {
71            return Err(DecisionError::Conflict(format!(
72                "deployment identity allocator mode {:?} cannot be opened as {:?}",
73                identity.allocator_mode, self.update_mode
74            )));
75        }
76        let canonical = serde_json::to_vec(&identity)
77            .map_err(|error| DecisionError::Conflict(error.to_string()))?;
78        if canonical.as_slice() != bytes.as_ref() {
79            return Err(DecisionError::Conflict(
80                "deployment identity does not use its canonical body".into(),
81            ));
82        }
83        Ok(Some(VersionedCheckpointIdHead {
84            head: identity,
85            update_version,
86        }))
87    }
88
89    /// Load the checkpoint namespace's create-once deployment incarnation, creating it when the
90    /// durable store is empty. Concurrent cluster members converge through object-store CAS.
91    ///
92    /// # Errors
93    /// Object-store I/O or a malformed/conflicting persisted identity.
94    pub async fn load_or_create_deployment_id(&self) -> Result<String, DecisionError> {
95        if let Some(identity) = self.deployment_id.get() {
96            return Ok(identity.clone());
97        }
98        let _guard = self.metadata_write_lock.lock().await;
99        if let Some(identity) = self.deployment_id.get() {
100            return Ok(identity.clone());
101        }
102        if let Some(stored) = self.read_deployment_identity().await? {
103            let identity = stored.head.id.clone();
104            self.cache_checkpoint_id_head(Some(stored));
105            let _ = self.deployment_id.set(identity.clone());
106            return Ok(identity);
107        }
108
109        let identity = DeploymentIdentity {
110            version: DEPLOYMENT_IDENTITY_VERSION,
111            id: uuid::Uuid::now_v7().to_string(),
112            allocator_mode: self.update_mode,
113            checkpoint_id: 0,
114            allocation_id: uuid::Uuid::now_v7().to_string(),
115        };
116        let payload = Self::encode_control_record(
117            "deployment identity",
118            &identity,
119            DEPLOYMENT_IDENTITY_MAX_BYTES,
120        )?;
121        let options = PutOptions {
122            mode: PutMode::Create,
123            ..PutOptions::default()
124        };
125        match self
126            .store
127            .put_opts(
128                &Self::deployment_identity_path(),
129                PutPayload::from(payload),
130                options,
131            )
132            .await
133        {
134            Ok(put_result) => {
135                let update_version: UpdateVersion = put_result.into();
136                if self.update_mode == DecisionStoreUpdateMode::NativeCas
137                    && update_version.e_tag.is_none()
138                    && update_version.version.is_none()
139                {
140                    self.cache_checkpoint_id_head(None);
141                } else {
142                    self.cache_checkpoint_id_head(Some(VersionedCheckpointIdHead {
143                        head: identity.clone(),
144                        update_version,
145                    }));
146                }
147                let _ = self.deployment_id.set(identity.id.clone());
148                Ok(identity.id)
149            }
150            Err(
151                object_store::Error::Precondition { .. }
152                | object_store::Error::AlreadyExists { .. },
153            ) => {
154                let stored = self.read_deployment_identity().await?.ok_or_else(|| {
155                    DecisionError::Conflict(
156                        "deployment identity disappeared after create conflict".into(),
157                    )
158                })?;
159                let identity = stored.head.id.clone();
160                self.cache_checkpoint_id_head(Some(stored));
161                let _ = self.deployment_id.set(identity.clone());
162                Ok(identity)
163            }
164            Err(error) => match self.read_deployment_identity().await? {
165                Some(stored) => {
166                    let identity = stored.head.id.clone();
167                    self.cache_checkpoint_id_head(Some(stored));
168                    let _ = self.deployment_id.set(identity.clone());
169                    Ok(identity)
170                }
171                None => Err(DecisionError::Io(error.to_string())),
172            },
173        }
174    }
175}
laminar_core/checkpoint_decision/deployment.rs

laminar_core/checkpoint_decision/
deployment.rs
