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, CollectionExportRetirement, Digest, IndexTransferArtifact, Operation, OperationState,
9    PortabilityStorageCapability, PrepareCollectionExportRequest, PrepareCollectionExportRetirementRequest,
10    PrepareCollectionRestoreRequest, PrepareIndexExportRequest, PrepareIndexImportRequest,
11    ResolveCollectionExportRequest, ResolveCollectionExportRetirementRequest, ResolveCollectionRestoreRequest,
12    ResolveIndexExportRequest, ResolveIndexImportRequest, Resource, StartCollectionExportRequest,
13    StartCollectionExportRetirementRequest, StartCollectionRestoreRequest, StartIndexExportRequest,
14    StartIndexImportRequest,
15};
16use serde::Serialize;
17
18pub const PORTABILITY_WORKFLOW_FORMAT_VERSION: u16 = 3;
19const PORTABILITY_WAIT_ATTEMPTS: u32 = 32;
20const PORTABILITY_WAIT_INTERVAL_MILLIS: u64 = 250;
21
22#[derive(Clone, Debug, PartialEq, Serialize)]
23pub struct CollectionExportExecutionV1 {
24    pub operation: Operation,
25    pub artifact: Option<CollectionExportArtifact>,
26}
27
28#[derive(Clone, Debug, PartialEq, Serialize)]
29pub struct CollectionRestoreExecutionV1 {
30    pub operation: Operation,
31}
32
33#[derive(Clone, Debug, PartialEq, Serialize)]
34pub struct CollectionExportRetirementExecutionV1 {
35    pub operation: Operation,
36    pub retirement: Option<CollectionExportRetirement>,
37}
38
39#[derive(Clone, Debug, PartialEq, Serialize)]
40pub struct IndexExportExecutionV1 {
41    pub operation: Operation,
42    pub artifact: Option<IndexTransferArtifact>,
43}
44
45#[derive(Clone, Debug, PartialEq, Serialize)]
46pub struct IndexImportExecutionV1 {
47    pub operation: Operation,
48    pub destination: Option<Resource>,
49}
50
51#[derive(Debug)]
52pub enum PortabilityWorkflowError {
53    Client(ClientError),
54    ExpiredUnknown,
55    MissingOperation,
56    MissingArtifact,
57    MissingRetirement,
58    MissingDestination,
59    Wait(OperationWaitError),
60}
61
62impl Display for PortabilityWorkflowError {
63    fn fmt(&self, formatter: &mut Formatter<'_>) -> fmt::Result {
64        formatter.write_str(match self {
65            Self::Client(_) => "portability API call failed",
66            Self::ExpiredUnknown => "portability outcome retired before resolution",
67            Self::MissingOperation => "portability resolver omitted the operation",
68            Self::MissingArtifact => "successful portability export omitted its artifact",
69            Self::MissingRetirement => "successful Collection export retirement omitted its absence proof",
70            Self::MissingDestination => "successful index import omitted its destination",
71            Self::Wait(_) => "portability operation wait failed",
72        })
73    }
74}
75
76impl Error for PortabilityWorkflowError {
77    fn source(&self) -> Option<&(dyn Error + 'static)> {
78        match self {
79            Self::Client(error) => Some(error),
80            Self::Wait(error) => Some(error),
81            Self::ExpiredUnknown
82            | Self::MissingOperation
83            | Self::MissingArtifact
84            | Self::MissingRetirement
85            | Self::MissingDestination => None,
86        }
87    }
88}
89
90impl From<ClientError> for PortabilityWorkflowError {
91    fn from(error: ClientError) -> Self {
92        Self::Client(error)
93    }
94}
95
96fn resolved_operation(
97    operation: Option<Operation>,
98    expired_unknown: bool,
99) -> Result<Operation, PortabilityWorkflowError> {
100    operation.ok_or(if expired_unknown {
101        PortabilityWorkflowError::ExpiredUnknown
102    } else {
103        PortabilityWorkflowError::MissingOperation
104    })
105}
106
107impl Client {
108    /// Prepares, submits, waits once through the bounded native waiter, and
109    /// resolves a Collection export. An indeterminate start response is never
110    /// replayed blindly; the caller action identity is resolved instead.
111    pub async fn export_collection(
112        &self,
113        action_id: Digest,
114        storage: PortabilityStorageCapability,
115        options: CallOptions,
116    ) -> Result<CollectionExportExecutionV1, PortabilityWorkflowError> {
117        let preparation = self
118            .prepare_collection_export(
119                PrepareCollectionExportRequest {
120                    action_id: action_id.clone(),
121                    storage: storage.clone(),
122                },
123                options.clone(),
124            )
125            .await?;
126        let start = self
127            .start_collection_export(
128                StartCollectionExportRequest {
129                    action_id: action_id.clone(),
130                    semantic_digest: preparation.semantic_digest.clone(),
131                    storage: storage.clone(),
132                },
133                options.clone(),
134            )
135            .await;
136        if let Err(error) = start
137            && error.submission_disposition() == SubmissionDisposition::NotSubmitted
138        {
139            return Err(PortabilityWorkflowError::Client(error));
140        }
141        let request = ResolveCollectionExportRequest {
142            action_id,
143            semantic_digest: preparation.semantic_digest,
144            storage,
145        };
146        let mut resolved = self.resolve_collection_export(request.clone(), options.clone()).await?;
147        let mut operation = resolved_operation(resolved.operation, resolved.expired_unknown)?;
148        if operation.current_state == OperationState::Pending {
149            let _waited = self.wait_portability_operation(&operation, options.clone()).await?;
150            resolved = self.resolve_collection_export(request, options).await?;
151            operation = resolved_operation(resolved.operation, resolved.expired_unknown)?;
152        }
153        if operation.current_state == OperationState::Succeeded && resolved.artifact.is_none() {
154            return Err(PortabilityWorkflowError::MissingArtifact);
155        }
156        Ok(CollectionExportExecutionV1 {
157            operation,
158            artifact: resolved.artifact,
159        })
160    }
161
162    /// Retires one exact Collection export and resolves the authoritative
163    /// absence proof. A pending resolver outcome is followed by one exact
164    /// replay of the same action identity and semantic digest; no fresh
165    /// deletion identity is invented.
166    ///
167    /// Callers that require recovery across their own process loss must durably
168    /// retain the prepare result and complete request before invoking start,
169    /// then use the generated resolve/start methods with those exact values.
170    pub async fn retire_collection_export(
171        &self,
172        action_id: Digest,
173        artifact: CollectionExportArtifact,
174        storage: PortabilityStorageCapability,
175        options: CallOptions,
176    ) -> Result<CollectionExportRetirementExecutionV1, PortabilityWorkflowError> {
177        let preparation = self
178            .prepare_collection_export_retirement(
179                PrepareCollectionExportRetirementRequest {
180                    action_id: action_id.clone(),
181                    artifact: artifact.clone(),
182                    storage: storage.clone(),
183                },
184                options.clone(),
185            )
186            .await?;
187        let start_request = StartCollectionExportRetirementRequest {
188            action_id: action_id.clone(),
189            semantic_digest: preparation.semantic_digest.clone(),
190            artifact: artifact.clone(),
191            storage: storage.clone(),
192        };
193        let start = self
194            .start_collection_export_retirement(start_request.clone(), options.clone())
195            .await;
196        if let Err(error) = start
197            && error.submission_disposition() == SubmissionDisposition::NotSubmitted
198        {
199            return Err(PortabilityWorkflowError::Client(error));
200        }
201        let resolve_request = ResolveCollectionExportRetirementRequest {
202            action_id,
203            semantic_digest: preparation.semantic_digest,
204            artifact,
205            storage,
206        };
207        let mut resolved = self
208            .resolve_collection_export_retirement(resolve_request.clone(), options.clone())
209            .await?;
210        let mut operation = resolved_operation(resolved.operation, resolved.expired_unknown)?;
211        if operation.current_state == OperationState::Pending {
212            // Recovery is permitted only after the read-only resolver proves
213            // that this exact retained operation is still pending.
214            let _replayed = self
215                .start_collection_export_retirement(start_request, options.clone())
216                .await;
217            resolved = self
218                .resolve_collection_export_retirement(resolve_request.clone(), options.clone())
219                .await?;
220            operation = resolved_operation(resolved.operation, resolved.expired_unknown)?;
221        }
222        if operation.current_state == OperationState::Pending {
223            let _waited = self.wait_portability_operation(&operation, options.clone()).await?;
224            resolved = self
225                .resolve_collection_export_retirement(resolve_request, options)
226                .await?;
227            operation = resolved_operation(resolved.operation, resolved.expired_unknown)?;
228        }
229        if operation.current_state == OperationState::Succeeded && resolved.retirement.is_none() {
230            return Err(PortabilityWorkflowError::MissingRetirement);
231        }
232        Ok(CollectionExportRetirementExecutionV1 {
233            operation,
234            retirement: resolved.retirement,
235        })
236    }
237
238    /// Prepares, submits, waits once through the bounded native waiter, and
239    /// resolves a restore into the caller's fresh destination Collection.
240    pub async fn restore_collection(
241        &self,
242        action_id: Digest,
243        artifact: CollectionExportArtifact,
244        storage: PortabilityStorageCapability,
245        options: CallOptions,
246    ) -> Result<CollectionRestoreExecutionV1, PortabilityWorkflowError> {
247        let preparation = self
248            .prepare_collection_restore(
249                PrepareCollectionRestoreRequest {
250                    action_id: action_id.clone(),
251                    artifact: artifact.clone(),
252                    storage: storage.clone(),
253                },
254                options.clone(),
255            )
256            .await?;
257        let start = self
258            .start_collection_restore(
259                StartCollectionRestoreRequest {
260                    action_id: action_id.clone(),
261                    semantic_digest: preparation.semantic_digest.clone(),
262                    artifact: artifact.clone(),
263                    storage: storage.clone(),
264                },
265                options.clone(),
266            )
267            .await;
268        if let Err(error) = start
269            && error.submission_disposition() == SubmissionDisposition::NotSubmitted
270        {
271            return Err(PortabilityWorkflowError::Client(error));
272        }
273        let request = ResolveCollectionRestoreRequest {
274            action_id,
275            semantic_digest: preparation.semantic_digest,
276            artifact,
277            storage,
278        };
279        let mut resolved = self
280            .resolve_collection_restore(request.clone(), options.clone())
281            .await?;
282        let mut operation = resolved_operation(resolved.operation, resolved.expired_unknown)?;
283        if operation.current_state == OperationState::Pending {
284            let _waited = self.wait_portability_operation(&operation, options.clone()).await?;
285            resolved = self.resolve_collection_restore(request, options).await?;
286            operation = resolved_operation(resolved.operation, resolved.expired_unknown)?;
287        }
288        Ok(CollectionRestoreExecutionV1 { operation })
289    }
290
291    /// Exports one exact source index into bounded, independently verifiable
292    /// transfer pages. The original resource identity and caller-selected
293    /// bounds remain part of every recovery request.
294    pub async fn export_index(
295        &self,
296        request: PrepareIndexExportRequest,
297        options: CallOptions,
298    ) -> Result<IndexExportExecutionV1, PortabilityWorkflowError> {
299        let preparation = self.prepare_index_export(request.clone(), options.clone()).await?;
300        let start = self
301            .start_index_export(
302                StartIndexExportRequest {
303                    action_id: request.action_id.clone(),
304                    semantic_digest: preparation.semantic_digest.clone(),
305                    index: request.index.clone(),
306                    storage: request.storage.clone(),
307                    maximum_records: request.maximum_records,
308                    maximum_pages: request.maximum_pages,
309                    maximum_page_bytes: request.maximum_page_bytes.clone(),
310                },
311                options.clone(),
312            )
313            .await;
314        if let Err(error) = start
315            && error.submission_disposition() == SubmissionDisposition::NotSubmitted
316        {
317            return Err(PortabilityWorkflowError::Client(error));
318        }
319        let resolve_request = ResolveIndexExportRequest {
320            action_id: request.action_id,
321            semantic_digest: preparation.semantic_digest,
322            index: request.index,
323            storage: request.storage,
324            maximum_records: request.maximum_records,
325            maximum_pages: request.maximum_pages,
326            maximum_page_bytes: request.maximum_page_bytes,
327        };
328        let mut resolved = self
329            .resolve_index_export(resolve_request.clone(), options.clone())
330            .await?;
331        let mut operation = resolved_operation(resolved.operation, resolved.expired_unknown)?;
332        if operation.current_state == OperationState::Pending {
333            let _waited = self.wait_portability_operation(&operation, options.clone()).await?;
334            resolved = self.resolve_index_export(resolve_request, options).await?;
335            operation = resolved_operation(resolved.operation, resolved.expired_unknown)?;
336        }
337        if operation.current_state == OperationState::Succeeded && resolved.artifact.is_none() {
338            return Err(PortabilityWorkflowError::MissingArtifact);
339        }
340        Ok(IndexExportExecutionV1 {
341            operation,
342            artifact: resolved.artifact,
343        })
344    }
345
346    /// Imports one verified transfer artifact into a fresh destination name.
347    /// Destination policy, topology, and storage choices are caller-owned and
348    /// remain bound to the same recovery identity.
349    pub async fn import_index(
350        &self,
351        request: PrepareIndexImportRequest,
352        options: CallOptions,
353    ) -> Result<IndexImportExecutionV1, PortabilityWorkflowError> {
354        let preparation = self.prepare_index_import(request.clone(), options.clone()).await?;
355        let start = self
356            .start_index_import(
357                StartIndexImportRequest {
358                    action_id: request.action_id.clone(),
359                    semantic_digest: preparation.semantic_digest.clone(),
360                    artifact: request.artifact.clone(),
361                    storage: request.storage.clone(),
362                    destination_name: request.destination_name.clone(),
363                    expected_name_absent: request.expected_name_absent,
364                    destination_policies: request.destination_policies.clone(),
365                    destination_resource_spec: request.destination_resource_spec.clone(),
366                },
367                options.clone(),
368            )
369            .await;
370        if let Err(error) = start
371            && error.submission_disposition() == SubmissionDisposition::NotSubmitted
372        {
373            return Err(PortabilityWorkflowError::Client(error));
374        }
375        let resolve_request = ResolveIndexImportRequest {
376            action_id: request.action_id,
377            semantic_digest: preparation.semantic_digest,
378            artifact: request.artifact,
379            storage: request.storage,
380            destination_name: request.destination_name,
381            expected_name_absent: request.expected_name_absent,
382            destination_policies: request.destination_policies,
383            destination_resource_spec: request.destination_resource_spec,
384        };
385        let mut resolved = self
386            .resolve_index_import(resolve_request.clone(), options.clone())
387            .await?;
388        let mut operation = resolved_operation(resolved.operation, resolved.expired_unknown)?;
389        if operation.current_state == OperationState::Pending {
390            let _waited = self.wait_portability_operation(&operation, options.clone()).await?;
391            resolved = self.resolve_index_import(resolve_request, options).await?;
392            operation = resolved_operation(resolved.operation, resolved.expired_unknown)?;
393        }
394        if operation.current_state == OperationState::Succeeded && resolved.destination.is_none() {
395            return Err(PortabilityWorkflowError::MissingDestination);
396        }
397        Ok(IndexImportExecutionV1 {
398            operation,
399            destination: resolved.destination,
400        })
401    }
402
403    async fn wait_portability_operation(
404        &self,
405        operation: &Operation,
406        options: CallOptions,
407    ) -> Result<Operation, PortabilityWorkflowError> {
408        Ok(self
409            .wait_for_operation(
410                operation.locator.clone(),
411                operation.authority_revision.clone(),
412                OperationWaitOptions {
413                    maximum_attempts: PORTABILITY_WAIT_ATTEMPTS,
414                    interval_millis: PORTABILITY_WAIT_INTERVAL_MILLIS,
415                },
416                options,
417            )
418            .await
419            .map_err(PortabilityWorkflowError::Wait)?
420            .operation)
421    }
422}
423
424#[cfg(test)]
425mod tests {
426    use super::*;
427
428    #[test]
429    fn portability_wait_contract_is_bounded() {
430        assert!((1..=64).contains(&PORTABILITY_WAIT_ATTEMPTS));
431        assert!((1..=60_000).contains(&PORTABILITY_WAIT_INTERVAL_MILLIS));
432        assert_eq!(PORTABILITY_WORKFLOW_FORMAT_VERSION, 3);
433    }
434}
lodestone_client/portability.rs

lodestone_client/
portability.rs
