lodestone_client/
portability.rs

1//! Recovery-aware portability workflows over the generated API.
2
3use std::error::Error;
4use std::fmt::{self, Display, Formatter};
5
6use crate::{CallOptions, Client, ClientError, OperationWaitError, OperationWaitOptions, SubmissionDisposition};
7use lodestone_wire::{
8    CollectionExportArtifact, Digest, IndexTransferArtifact, Operation, OperationState, PortabilityStorageCapability,
9    PrepareCollectionExportRequest, PrepareCollectionRestoreRequest, PrepareIndexExportRequest,
10    PrepareIndexImportRequest, ResolveCollectionExportRequest, ResolveCollectionRestoreRequest,
11    ResolveIndexExportRequest, ResolveIndexImportRequest, Resource, StartCollectionExportRequest,
12    StartCollectionRestoreRequest, StartIndexExportRequest, StartIndexImportRequest,
13};
14use serde::Serialize;
15
16pub const PORTABILITY_WORKFLOW_FORMAT_VERSION: u16 = 2;
17const PORTABILITY_WAIT_ATTEMPTS: u32 = 32;
18const PORTABILITY_WAIT_INTERVAL_MILLIS: u64 = 250;
19
20#[derive(Clone, Debug, PartialEq, Serialize)]
21pub struct CollectionExportExecutionV1 {
22    pub operation: Operation,
23    pub artifact: Option<CollectionExportArtifact>,
24}
25
26#[derive(Clone, Debug, PartialEq, Serialize)]
27pub struct CollectionRestoreExecutionV1 {
28    pub operation: Operation,
29}
30
31#[derive(Clone, Debug, PartialEq, Serialize)]
32pub struct IndexExportExecutionV1 {
33    pub operation: Operation,
34    pub artifact: Option<IndexTransferArtifact>,
35}
36
37#[derive(Clone, Debug, PartialEq, Serialize)]
38pub struct IndexImportExecutionV1 {
39    pub operation: Operation,
40    pub destination: Option<Resource>,
41}
42
43#[derive(Debug)]
44pub enum PortabilityWorkflowError {
45    Client(ClientError),
46    ExpiredUnknown,
47    MissingOperation,
48    MissingArtifact,
49    MissingDestination,
50    Wait(OperationWaitError),
51}
52
53impl Display for PortabilityWorkflowError {
54    fn fmt(&self, formatter: &mut Formatter<'_>) -> fmt::Result {
55        formatter.write_str(match self {
56            Self::Client(_) => "portability API call failed",
57            Self::ExpiredUnknown => "portability outcome retired before resolution",
58            Self::MissingOperation => "portability resolver omitted the operation",
59            Self::MissingArtifact => "successful portability export omitted its artifact",
60            Self::MissingDestination => "successful index import omitted its destination",
61            Self::Wait(_) => "portability operation wait failed",
62        })
63    }
64}
65
66impl Error for PortabilityWorkflowError {
67    fn source(&self) -> Option<&(dyn Error + 'static)> {
68        match self {
69            Self::Client(error) => Some(error),
70            Self::Wait(error) => Some(error),
71            Self::ExpiredUnknown | Self::MissingOperation | Self::MissingArtifact | Self::MissingDestination => None,
72        }
73    }
74}
75
76impl From<ClientError> for PortabilityWorkflowError {
77    fn from(error: ClientError) -> Self {
78        Self::Client(error)
79    }
80}
81
82fn resolved_operation(
83    operation: Option<Operation>,
84    expired_unknown: bool,
85) -> Result<Operation, PortabilityWorkflowError> {
86    operation.ok_or(if expired_unknown {
87        PortabilityWorkflowError::ExpiredUnknown
88    } else {
89        PortabilityWorkflowError::MissingOperation
90    })
91}
92
93impl Client {
94    /// Prepares, submits, waits once through the bounded native waiter, and
95    /// resolves a Collection export. An indeterminate start response is never
96    /// replayed blindly; the caller action identity is resolved instead.
97    pub async fn export_collection(
98        &self,
99        action_id: Digest,
100        storage: PortabilityStorageCapability,
101        options: CallOptions,
102    ) -> Result<CollectionExportExecutionV1, PortabilityWorkflowError> {
103        let preparation = self
104            .prepare_collection_export(
105                PrepareCollectionExportRequest {
106                    action_id: action_id.clone(),
107                    storage: storage.clone(),
108                },
109                options.clone(),
110            )
111            .await?;
112        let start = self
113            .start_collection_export(
114                StartCollectionExportRequest {
115                    action_id: action_id.clone(),
116                    semantic_digest: preparation.semantic_digest.clone(),
117                    storage: storage.clone(),
118                },
119                options.clone(),
120            )
121            .await;
122        if let Err(error) = start
123            && error.submission_disposition() == SubmissionDisposition::NotSubmitted
124        {
125            return Err(PortabilityWorkflowError::Client(error));
126        }
127        let request = ResolveCollectionExportRequest {
128            action_id,
129            semantic_digest: preparation.semantic_digest,
130            storage,
131        };
132        let mut resolved = self.resolve_collection_export(request.clone(), options.clone()).await?;
133        let mut operation = resolved_operation(resolved.operation, resolved.expired_unknown)?;
134        if operation.current_state == OperationState::Pending {
135            let _waited = self.wait_portability_operation(&operation, options.clone()).await?;
136            resolved = self.resolve_collection_export(request, options).await?;
137            operation = resolved_operation(resolved.operation, resolved.expired_unknown)?;
138        }
139        if operation.current_state == OperationState::Succeeded && resolved.artifact.is_none() {
140            return Err(PortabilityWorkflowError::MissingArtifact);
141        }
142        Ok(CollectionExportExecutionV1 {
143            operation,
144            artifact: resolved.artifact,
145        })
146    }
147
148    /// Prepares, submits, waits once through the bounded native waiter, and
149    /// resolves a restore into the caller's fresh destination Collection.
150    pub async fn restore_collection(
151        &self,
152        action_id: Digest,
153        artifact: CollectionExportArtifact,
154        storage: PortabilityStorageCapability,
155        options: CallOptions,
156    ) -> Result<CollectionRestoreExecutionV1, PortabilityWorkflowError> {
157        let preparation = self
158            .prepare_collection_restore(
159                PrepareCollectionRestoreRequest {
160                    action_id: action_id.clone(),
161                    artifact: artifact.clone(),
162                    storage: storage.clone(),
163                },
164                options.clone(),
165            )
166            .await?;
167        let start = self
168            .start_collection_restore(
169                StartCollectionRestoreRequest {
170                    action_id: action_id.clone(),
171                    semantic_digest: preparation.semantic_digest.clone(),
172                    artifact: artifact.clone(),
173                    storage: storage.clone(),
174                },
175                options.clone(),
176            )
177            .await;
178        if let Err(error) = start
179            && error.submission_disposition() == SubmissionDisposition::NotSubmitted
180        {
181            return Err(PortabilityWorkflowError::Client(error));
182        }
183        let request = ResolveCollectionRestoreRequest {
184            action_id,
185            semantic_digest: preparation.semantic_digest,
186            artifact,
187            storage,
188        };
189        let mut resolved = self
190            .resolve_collection_restore(request.clone(), options.clone())
191            .await?;
192        let mut operation = resolved_operation(resolved.operation, resolved.expired_unknown)?;
193        if operation.current_state == OperationState::Pending {
194            let _waited = self.wait_portability_operation(&operation, options.clone()).await?;
195            resolved = self.resolve_collection_restore(request, options).await?;
196            operation = resolved_operation(resolved.operation, resolved.expired_unknown)?;
197        }
198        Ok(CollectionRestoreExecutionV1 { operation })
199    }
200
201    /// Exports one exact source index into bounded, independently verifiable
202    /// transfer pages. The original resource identity and caller-selected
203    /// bounds remain part of every recovery request.
204    pub async fn export_index(
205        &self,
206        request: PrepareIndexExportRequest,
207        options: CallOptions,
208    ) -> Result<IndexExportExecutionV1, PortabilityWorkflowError> {
209        let preparation = self.prepare_index_export(request.clone(), options.clone()).await?;
210        let start = self
211            .start_index_export(
212                StartIndexExportRequest {
213                    action_id: request.action_id.clone(),
214                    semantic_digest: preparation.semantic_digest.clone(),
215                    index: request.index.clone(),
216                    storage: request.storage.clone(),
217                    maximum_records: request.maximum_records,
218                    maximum_pages: request.maximum_pages,
219                    maximum_page_bytes: request.maximum_page_bytes.clone(),
220                },
221                options.clone(),
222            )
223            .await;
224        if let Err(error) = start
225            && error.submission_disposition() == SubmissionDisposition::NotSubmitted
226        {
227            return Err(PortabilityWorkflowError::Client(error));
228        }
229        let resolve_request = ResolveIndexExportRequest {
230            action_id: request.action_id,
231            semantic_digest: preparation.semantic_digest,
232            index: request.index,
233            storage: request.storage,
234            maximum_records: request.maximum_records,
235            maximum_pages: request.maximum_pages,
236            maximum_page_bytes: request.maximum_page_bytes,
237        };
238        let mut resolved = self
239            .resolve_index_export(resolve_request.clone(), options.clone())
240            .await?;
241        let mut operation = resolved_operation(resolved.operation, resolved.expired_unknown)?;
242        if operation.current_state == OperationState::Pending {
243            let _waited = self.wait_portability_operation(&operation, options.clone()).await?;
244            resolved = self.resolve_index_export(resolve_request, options).await?;
245            operation = resolved_operation(resolved.operation, resolved.expired_unknown)?;
246        }
247        if operation.current_state == OperationState::Succeeded && resolved.artifact.is_none() {
248            return Err(PortabilityWorkflowError::MissingArtifact);
249        }
250        Ok(IndexExportExecutionV1 {
251            operation,
252            artifact: resolved.artifact,
253        })
254    }
255
256    /// Imports one verified transfer artifact into a fresh destination name.
257    /// Destination policy, topology, and storage choices are caller-owned and
258    /// remain bound to the same recovery identity.
259    pub async fn import_index(
260        &self,
261        request: PrepareIndexImportRequest,
262        options: CallOptions,
263    ) -> Result<IndexImportExecutionV1, PortabilityWorkflowError> {
264        let preparation = self.prepare_index_import(request.clone(), options.clone()).await?;
265        let start = self
266            .start_index_import(
267                StartIndexImportRequest {
268                    action_id: request.action_id.clone(),
269                    semantic_digest: preparation.semantic_digest.clone(),
270                    artifact: request.artifact.clone(),
271                    storage: request.storage.clone(),
272                    destination_name: request.destination_name.clone(),
273                    expected_name_absent: request.expected_name_absent,
274                    destination_policies: request.destination_policies.clone(),
275                    destination_resource_spec: request.destination_resource_spec.clone(),
276                },
277                options.clone(),
278            )
279            .await;
280        if let Err(error) = start
281            && error.submission_disposition() == SubmissionDisposition::NotSubmitted
282        {
283            return Err(PortabilityWorkflowError::Client(error));
284        }
285        let resolve_request = ResolveIndexImportRequest {
286            action_id: request.action_id,
287            semantic_digest: preparation.semantic_digest,
288            artifact: request.artifact,
289            storage: request.storage,
290            destination_name: request.destination_name,
291            expected_name_absent: request.expected_name_absent,
292            destination_policies: request.destination_policies,
293            destination_resource_spec: request.destination_resource_spec,
294        };
295        let mut resolved = self
296            .resolve_index_import(resolve_request.clone(), options.clone())
297            .await?;
298        let mut operation = resolved_operation(resolved.operation, resolved.expired_unknown)?;
299        if operation.current_state == OperationState::Pending {
300            let _waited = self.wait_portability_operation(&operation, options.clone()).await?;
301            resolved = self.resolve_index_import(resolve_request, options).await?;
302            operation = resolved_operation(resolved.operation, resolved.expired_unknown)?;
303        }
304        if operation.current_state == OperationState::Succeeded && resolved.destination.is_none() {
305            return Err(PortabilityWorkflowError::MissingDestination);
306        }
307        Ok(IndexImportExecutionV1 {
308            operation,
309            destination: resolved.destination,
310        })
311    }
312
313    async fn wait_portability_operation(
314        &self,
315        operation: &Operation,
316        options: CallOptions,
317    ) -> Result<Operation, PortabilityWorkflowError> {
318        Ok(self
319            .wait_for_operation(
320                operation.locator.clone(),
321                operation.authority_revision.clone(),
322                OperationWaitOptions {
323                    maximum_attempts: PORTABILITY_WAIT_ATTEMPTS,
324                    interval_millis: PORTABILITY_WAIT_INTERVAL_MILLIS,
325                },
326                options,
327            )
328            .await
329            .map_err(PortabilityWorkflowError::Wait)?
330            .operation)
331    }
332}
333
334#[cfg(test)]
335mod tests {
336    use super::*;
337
338    #[test]
339    fn portability_wait_contract_is_bounded() {
340        assert!((1..=64).contains(&PORTABILITY_WAIT_ATTEMPTS));
341        assert!((1..=60_000).contains(&PORTABILITY_WAIT_INTERVAL_MILLIS));
342        assert_eq!(PORTABILITY_WORKFLOW_FORMAT_VERSION, 2);
343    }
344}
lodestone_client/portability.rs

lodestone_client/
portability.rs
