lodestone_client/
portability.rs

1//! Recovery-aware Collection 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, Operation, OperationState, PortabilityStorageCapability,
9    PrepareCollectionExportRequest, PrepareCollectionRestoreRequest, ResolveCollectionExportRequest,
10    ResolveCollectionRestoreRequest, StartCollectionExportRequest, StartCollectionRestoreRequest,
11};
12use serde::Serialize;
13
14pub const PORTABILITY_WORKFLOW_FORMAT_VERSION: u16 = 1;
15const PORTABILITY_WAIT_ATTEMPTS: u32 = 32;
16const PORTABILITY_WAIT_INTERVAL_MILLIS: u64 = 250;
17
18#[derive(Clone, Debug, PartialEq, Serialize)]
19pub struct CollectionExportExecutionV1 {
20    pub operation: Operation,
21    pub artifact: Option<CollectionExportArtifact>,
22}
23
24#[derive(Clone, Debug, PartialEq, Serialize)]
25pub struct CollectionRestoreExecutionV1 {
26    pub operation: Operation,
27}
28
29#[derive(Debug)]
30pub enum PortabilityWorkflowError {
31    Client(ClientError),
32    ExpiredUnknown,
33    MissingOperation,
34    MissingArtifact,
35    Wait(OperationWaitError),
36}
37
38impl Display for PortabilityWorkflowError {
39    fn fmt(&self, formatter: &mut Formatter<'_>) -> fmt::Result {
40        formatter.write_str(match self {
41            Self::Client(_) => "Collection portability API call failed",
42            Self::ExpiredUnknown => "Collection portability outcome retired before resolution",
43            Self::MissingOperation => "Collection portability resolver omitted the operation",
44            Self::MissingArtifact => "successful Collection export omitted its artifact",
45            Self::Wait(_) => "Collection portability operation wait failed",
46        })
47    }
48}
49
50impl Error for PortabilityWorkflowError {
51    fn source(&self) -> Option<&(dyn Error + 'static)> {
52        match self {
53            Self::Client(error) => Some(error),
54            Self::Wait(error) => Some(error),
55            Self::ExpiredUnknown | Self::MissingOperation | Self::MissingArtifact => None,
56        }
57    }
58}
59
60impl From<ClientError> for PortabilityWorkflowError {
61    fn from(error: ClientError) -> Self {
62        Self::Client(error)
63    }
64}
65
66fn resolved_operation(
67    operation: Option<Operation>,
68    expired_unknown: bool,
69) -> Result<Operation, PortabilityWorkflowError> {
70    operation.ok_or(if expired_unknown {
71        PortabilityWorkflowError::ExpiredUnknown
72    } else {
73        PortabilityWorkflowError::MissingOperation
74    })
75}
76
77impl Client {
78    /// Prepares, submits, waits once through the bounded native waiter, and
79    /// resolves a Collection export. An indeterminate start response is never
80    /// replayed blindly; the caller action identity is resolved instead.
81    pub async fn export_collection(
82        &self,
83        action_id: Digest,
84        storage: PortabilityStorageCapability,
85        options: CallOptions,
86    ) -> Result<CollectionExportExecutionV1, PortabilityWorkflowError> {
87        let preparation = self
88            .prepare_collection_export(
89                PrepareCollectionExportRequest {
90                    action_id: action_id.clone(),
91                    storage: storage.clone(),
92                },
93                options.clone(),
94            )
95            .await?;
96        let start = self
97            .start_collection_export(
98                StartCollectionExportRequest {
99                    action_id: action_id.clone(),
100                    semantic_digest: preparation.semantic_digest.clone(),
101                    storage: storage.clone(),
102                },
103                options.clone(),
104            )
105            .await;
106        if let Err(error) = start
107            && error.submission_disposition() == SubmissionDisposition::NotSubmitted
108        {
109            return Err(PortabilityWorkflowError::Client(error));
110        }
111        let request = ResolveCollectionExportRequest {
112            action_id,
113            semantic_digest: preparation.semantic_digest,
114            storage,
115        };
116        let mut resolved = self.resolve_collection_export(request.clone(), options.clone()).await?;
117        let mut operation = resolved_operation(resolved.operation, resolved.expired_unknown)?;
118        if operation.current_state == OperationState::Pending {
119            let _waited = self.wait_portability_operation(&operation, options.clone()).await?;
120            resolved = self.resolve_collection_export(request, options).await?;
121            operation = resolved_operation(resolved.operation, resolved.expired_unknown)?;
122        }
123        if operation.current_state == OperationState::Succeeded && resolved.artifact.is_none() {
124            return Err(PortabilityWorkflowError::MissingArtifact);
125        }
126        Ok(CollectionExportExecutionV1 {
127            operation,
128            artifact: resolved.artifact,
129        })
130    }
131
132    /// Prepares, submits, waits once through the bounded native waiter, and
133    /// resolves a restore into the caller's fresh destination Collection.
134    pub async fn restore_collection(
135        &self,
136        action_id: Digest,
137        artifact: CollectionExportArtifact,
138        storage: PortabilityStorageCapability,
139        options: CallOptions,
140    ) -> Result<CollectionRestoreExecutionV1, PortabilityWorkflowError> {
141        let preparation = self
142            .prepare_collection_restore(
143                PrepareCollectionRestoreRequest {
144                    action_id: action_id.clone(),
145                    artifact: artifact.clone(),
146                    storage: storage.clone(),
147                },
148                options.clone(),
149            )
150            .await?;
151        let start = self
152            .start_collection_restore(
153                StartCollectionRestoreRequest {
154                    action_id: action_id.clone(),
155                    semantic_digest: preparation.semantic_digest.clone(),
156                    artifact: artifact.clone(),
157                    storage: storage.clone(),
158                },
159                options.clone(),
160            )
161            .await;
162        if let Err(error) = start
163            && error.submission_disposition() == SubmissionDisposition::NotSubmitted
164        {
165            return Err(PortabilityWorkflowError::Client(error));
166        }
167        let request = ResolveCollectionRestoreRequest {
168            action_id,
169            semantic_digest: preparation.semantic_digest,
170            artifact,
171            storage,
172        };
173        let mut resolved = self
174            .resolve_collection_restore(request.clone(), options.clone())
175            .await?;
176        let mut operation = resolved_operation(resolved.operation, resolved.expired_unknown)?;
177        if operation.current_state == OperationState::Pending {
178            let _waited = self.wait_portability_operation(&operation, options.clone()).await?;
179            resolved = self.resolve_collection_restore(request, options).await?;
180            operation = resolved_operation(resolved.operation, resolved.expired_unknown)?;
181        }
182        Ok(CollectionRestoreExecutionV1 { operation })
183    }
184
185    async fn wait_portability_operation(
186        &self,
187        operation: &Operation,
188        options: CallOptions,
189    ) -> Result<Operation, PortabilityWorkflowError> {
190        Ok(self
191            .wait_for_operation(
192                operation.locator.clone(),
193                operation.authority_revision.clone(),
194                OperationWaitOptions {
195                    maximum_attempts: PORTABILITY_WAIT_ATTEMPTS,
196                    interval_millis: PORTABILITY_WAIT_INTERVAL_MILLIS,
197                },
198                options,
199            )
200            .await
201            .map_err(PortabilityWorkflowError::Wait)?
202            .operation)
203    }
204}
205
206#[cfg(test)]
207mod tests {
208    use super::*;
209
210    #[test]
211    fn portability_wait_contract_is_bounded() {
212        assert!((1..=64).contains(&PORTABILITY_WAIT_ATTEMPTS));
213        assert!((1..=60_000).contains(&PORTABILITY_WAIT_INTERVAL_MILLIS));
214        assert_eq!(PORTABILITY_WORKFLOW_FORMAT_VERSION, 1);
215    }
216}
lodestone_client/portability.rs

lodestone_client/
portability.rs
