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    /// Read the checkpoint namespace's deployment incarnation without creating or allocating.
90    ///
91    /// # Errors
92    /// Fails for storage I/O or malformed/conflicting persisted identity.
93    pub async fn load_deployment_id(&self) -> Result<Option<String>, DecisionError> {
94        self.read_deployment_identity()
95            .await
96            .map(|identity| identity.map(|stored| stored.head.id))
97    }
98
99    /// Load the checkpoint namespace's create-once deployment incarnation, creating it when the
100    /// durable store is empty. Concurrent cluster members converge through object-store CAS.
101    ///
102    /// # Errors
103    /// Object-store I/O or a malformed/conflicting persisted identity.
104    pub async fn load_or_create_deployment_id(&self) -> Result<String, DecisionError> {
105        if let Some(identity) = self.deployment_id.get() {
106            return Ok(identity.clone());
107        }
108        let _guard = self.metadata_write_lock.lock().await;
109        if let Some(identity) = self.deployment_id.get() {
110            return Ok(identity.clone());
111        }
112        if let Some(stored) = self.read_deployment_identity().await? {
113            let identity = stored.head.id.clone();
114            self.cache_checkpoint_id_head(Some(stored));
115            let _ = self.deployment_id.set(identity.clone());
116            return Ok(identity);
117        }
118
119        let identity = DeploymentIdentity {
120            version: DEPLOYMENT_IDENTITY_VERSION,
121            id: uuid::Uuid::now_v7().to_string(),
122            allocator_mode: self.update_mode,
123            checkpoint_id: 0,
124            allocation_id: uuid::Uuid::now_v7().to_string(),
125        };
126        let payload = Self::encode_control_record(
127            "deployment identity",
128            &identity,
129            DEPLOYMENT_IDENTITY_MAX_BYTES,
130        )?;
131        let options = PutOptions {
132            mode: PutMode::Create,
133            ..PutOptions::default()
134        };
135        match self
136            .store
137            .put_opts(
138                &Self::deployment_identity_path(),
139                PutPayload::from(payload),
140                options,
141            )
142            .await
143        {
144            Ok(put_result) => {
145                let update_version: UpdateVersion = put_result.into();
146                if self.update_mode == DecisionStoreUpdateMode::NativeCas
147                    && update_version.e_tag.is_none()
148                    && update_version.version.is_none()
149                {
150                    self.cache_checkpoint_id_head(None);
151                } else {
152                    self.cache_checkpoint_id_head(Some(VersionedCheckpointIdHead {
153                        head: identity.clone(),
154                        update_version,
155                    }));
156                }
157                let _ = self.deployment_id.set(identity.id.clone());
158                Ok(identity.id)
159            }
160            Err(
161                object_store::Error::Precondition { .. }
162                | object_store::Error::AlreadyExists { .. },
163            ) => {
164                let stored = self.read_deployment_identity().await?.ok_or_else(|| {
165                    DecisionError::Conflict(
166                        "deployment identity disappeared after create conflict".into(),
167                    )
168                })?;
169                let identity = stored.head.id.clone();
170                self.cache_checkpoint_id_head(Some(stored));
171                let _ = self.deployment_id.set(identity.clone());
172                Ok(identity)
173            }
174            Err(error) => match self.read_deployment_identity().await? {
175                Some(stored) => {
176                    let identity = stored.head.id.clone();
177                    self.cache_checkpoint_id_head(Some(stored));
178                    let _ = self.deployment_id.set(identity.clone());
179                    Ok(identity)
180                }
181                None => Err(DecisionError::Io(error.to_string())),
182            },
183        }
184    }
185}
laminar_core/checkpoint_decision/deployment.rs

laminar_core/checkpoint_decision/
deployment.rs
