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, CollectionTransferArtifact, Digest, IndexTransferArtifact,
9    Operation, OperationState, PortabilityStorageCapability, PrepareCollectionExportRequest,
10    PrepareCollectionExportRetirementRequest, PrepareCollectionRestoreRequest, PrepareCollectionTransferExportRequest,
11    PrepareCollectionTransferRestoreRequest, PrepareIndexExportRequest, PrepareIndexImportRequest,
12    ResolveCollectionExportRequest, ResolveCollectionExportRetirementRequest, ResolveCollectionRestoreRequest,
13    ResolveCollectionTransferExportRequest, ResolveCollectionTransferRestoreRequest, ResolveIndexExportRequest,
14    ResolveIndexImportRequest, Resource, StartCollectionExportRequest, StartCollectionExportRetirementRequest,
15    StartCollectionRestoreRequest, StartCollectionTransferExportRequest, StartCollectionTransferRestoreRequest,
16    StartIndexExportRequest, StartIndexImportRequest,
17};
18use serde::Serialize;
19
20pub const PORTABILITY_WORKFLOW_FORMAT_VERSION: u16 = 4;
21const PORTABILITY_WAIT_ATTEMPTS: u32 = 32;
22const PORTABILITY_WAIT_INTERVAL_MILLIS: u64 = 250;
23
24#[derive(Clone, Debug, PartialEq, Serialize)]
25pub struct CollectionExportExecutionV1 {
26    pub operation: Operation,
27    pub artifact: Option<CollectionExportArtifact>,
28}
29
30#[derive(Clone, Debug, PartialEq, Serialize)]
31pub struct CollectionRestoreExecutionV1 {
32    pub operation: Operation,
33}
34
35#[derive(Clone, Debug, PartialEq, Serialize)]
36pub struct CollectionTransferExportExecutionV1 {
37    pub operation: Operation,
38    pub artifact: Option<CollectionTransferArtifact>,
39}
40
41#[derive(Clone, Debug, PartialEq, Serialize)]
42pub struct CollectionTransferRestoreExecutionV1 {
43    pub operation: Operation,
44    pub verified_content_digest: Option<Digest>,
45}
46
47#[derive(Clone, Debug, PartialEq, Serialize)]
48pub struct CollectionExportRetirementExecutionV1 {
49    pub operation: Operation,
50    pub retirement: Option<CollectionExportRetirement>,
51}
52
53#[derive(Clone, Debug, PartialEq, Serialize)]
54pub struct IndexExportExecutionV1 {
55    pub operation: Operation,
56    pub artifact: Option<IndexTransferArtifact>,
57}
58
59#[derive(Clone, Debug, PartialEq, Serialize)]
60pub struct IndexImportExecutionV1 {
61    pub operation: Operation,
62    pub destination: Option<Resource>,
63}
64
65#[derive(Debug)]
66pub enum PortabilityWorkflowError {
67    Client(ClientError),
68    ExpiredUnknown,
69    MissingOperation,
70    MissingArtifact,
71    MissingRetirement,
72    MissingDestination,
73    MissingVerification,
74    VerificationMismatch,
75    Wait(OperationWaitError),
76}
77
78impl Display for PortabilityWorkflowError {
79    fn fmt(&self, formatter: &mut Formatter<'_>) -> fmt::Result {
80        formatter.write_str(match self {
81            Self::Client(_) => "portability API call failed",
82            Self::ExpiredUnknown => "portability outcome retired before resolution",
83            Self::MissingOperation => "portability resolver omitted the operation",
84            Self::MissingArtifact => "successful portability export omitted its artifact",
85            Self::MissingRetirement => "successful Collection export retirement omitted its absence proof",
86            Self::MissingDestination => "successful index import omitted its destination",
87            Self::MissingVerification => "successful Collection transfer restore omitted its content verification",
88            Self::VerificationMismatch => {
89                "successful Collection transfer restore returned another content verification"
90            }
91            Self::Wait(_) => "portability operation wait failed",
92        })
93    }
94}
95
96impl Error for PortabilityWorkflowError {
97    fn source(&self) -> Option<&(dyn Error + 'static)> {
98        match self {
99            Self::Client(error) => Some(error),
100            Self::Wait(error) => Some(error),
101            Self::ExpiredUnknown
102            | Self::MissingOperation
103            | Self::MissingArtifact
104            | Self::MissingRetirement
105            | Self::MissingDestination
106            | Self::MissingVerification
107            | Self::VerificationMismatch => None,
108        }
109    }
110}
111
112impl From<ClientError> for PortabilityWorkflowError {
113    fn from(error: ClientError) -> Self {
114        Self::Client(error)
115    }
116}
117
118fn resolved_operation(
119    operation: Option<Operation>,
120    expired_unknown: bool,
121) -> Result<Operation, PortabilityWorkflowError> {
122    operation.ok_or(if expired_unknown {
123        PortabilityWorkflowError::ExpiredUnknown
124    } else {
125        PortabilityWorkflowError::MissingOperation
126    })
127}
128
129impl Client {
130    /// Prepares, submits, waits once through the bounded native waiter, and
131    /// resolves a Collection export. An indeterminate start response is never
132    /// replayed blindly; the caller action identity is resolved instead.
133    pub async fn export_collection(
134        &self,
135        action_id: Digest,
136        storage: PortabilityStorageCapability,
137        options: CallOptions,
138    ) -> Result<CollectionExportExecutionV1, PortabilityWorkflowError> {
139        let preparation = self
140            .prepare_collection_export(
141                PrepareCollectionExportRequest {
142                    action_id: action_id.clone(),
143                    storage: storage.clone(),
144                },
145                options.clone(),
146            )
147            .await?;
148        let start = self
149            .start_collection_export(
150                StartCollectionExportRequest {
151                    action_id: action_id.clone(),
152                    semantic_digest: preparation.semantic_digest.clone(),
153                    storage: storage.clone(),
154                },
155                options.clone(),
156            )
157            .await;
158        if let Err(error) = start
159            && error.submission_disposition() == SubmissionDisposition::NotSubmitted
160        {
161            return Err(PortabilityWorkflowError::Client(error));
162        }
163        let request = ResolveCollectionExportRequest {
164            action_id,
165            semantic_digest: preparation.semantic_digest,
166            storage,
167        };
168        let mut resolved = self.resolve_collection_export(request.clone(), options.clone()).await?;
169        let mut operation = resolved_operation(resolved.operation, resolved.expired_unknown)?;
170        if operation.current_state == OperationState::Pending {
171            let _waited = self.wait_portability_operation(&operation, options.clone()).await?;
172            resolved = self.resolve_collection_export(request, options).await?;
173            operation = resolved_operation(resolved.operation, resolved.expired_unknown)?;
174        }
175        if operation.current_state == OperationState::Succeeded && resolved.artifact.is_none() {
176            return Err(PortabilityWorkflowError::MissingArtifact);
177        }
178        Ok(CollectionExportExecutionV1 {
179            operation,
180            artifact: resolved.artifact,
181        })
182    }
183
184    /// Retires one exact Collection export and resolves the authoritative
185    /// absence proof. A pending resolver outcome is followed by one exact
186    /// replay of the same action identity and semantic digest; no fresh
187    /// deletion identity is invented.
188    ///
189    /// Callers that require recovery across their own process loss must durably
190    /// retain the prepare result and complete request before invoking start,
191    /// then use the generated resolve/start methods with those exact values.
192    pub async fn retire_collection_export(
193        &self,
194        action_id: Digest,
195        artifact: CollectionExportArtifact,
196        storage: PortabilityStorageCapability,
197        options: CallOptions,
198    ) -> Result<CollectionExportRetirementExecutionV1, PortabilityWorkflowError> {
199        let preparation = self
200            .prepare_collection_export_retirement(
201                PrepareCollectionExportRetirementRequest {
202                    action_id: action_id.clone(),
203                    artifact: artifact.clone(),
204                    storage: storage.clone(),
205                },
206                options.clone(),
207            )
208            .await?;
209        let start_request = StartCollectionExportRetirementRequest {
210            action_id: action_id.clone(),
211            semantic_digest: preparation.semantic_digest.clone(),
212            artifact: artifact.clone(),
213            storage: storage.clone(),
214        };
215        let start = self
216            .start_collection_export_retirement(start_request.clone(), options.clone())
217            .await;
218        if let Err(error) = start
219            && error.submission_disposition() == SubmissionDisposition::NotSubmitted
220        {
221            return Err(PortabilityWorkflowError::Client(error));
222        }
223        let resolve_request = ResolveCollectionExportRetirementRequest {
224            action_id,
225            semantic_digest: preparation.semantic_digest,
226            artifact,
227            storage,
228        };
229        let mut resolved = self
230            .resolve_collection_export_retirement(resolve_request.clone(), options.clone())
231            .await?;
232        let mut operation = resolved_operation(resolved.operation, resolved.expired_unknown)?;
233        if operation.current_state == OperationState::Pending {
234            // Recovery is permitted only after the read-only resolver proves
235            // that this exact retained operation is still pending.
236            let _replayed = self
237                .start_collection_export_retirement(start_request, options.clone())
238                .await;
239            resolved = self
240                .resolve_collection_export_retirement(resolve_request.clone(), options.clone())
241                .await?;
242            operation = resolved_operation(resolved.operation, resolved.expired_unknown)?;
243        }
244        if operation.current_state == OperationState::Pending {
245            let _waited = self.wait_portability_operation(&operation, options.clone()).await?;
246            resolved = self
247                .resolve_collection_export_retirement(resolve_request, options)
248                .await?;
249            operation = resolved_operation(resolved.operation, resolved.expired_unknown)?;
250        }
251        if operation.current_state == OperationState::Succeeded && resolved.retirement.is_none() {
252            return Err(PortabilityWorkflowError::MissingRetirement);
253        }
254        Ok(CollectionExportRetirementExecutionV1 {
255            operation,
256            retirement: resolved.retirement,
257        })
258    }
259
260    /// Prepares, submits, waits once through the bounded native waiter, and
261    /// resolves a restore into the caller's fresh destination Collection.
262    pub async fn restore_collection(
263        &self,
264        action_id: Digest,
265        artifact: CollectionExportArtifact,
266        storage: PortabilityStorageCapability,
267        options: CallOptions,
268    ) -> Result<CollectionRestoreExecutionV1, PortabilityWorkflowError> {
269        let preparation = self
270            .prepare_collection_restore(
271                PrepareCollectionRestoreRequest {
272                    action_id: action_id.clone(),
273                    artifact: artifact.clone(),
274                    storage: storage.clone(),
275                },
276                options.clone(),
277            )
278            .await?;
279        let start = self
280            .start_collection_restore(
281                StartCollectionRestoreRequest {
282                    action_id: action_id.clone(),
283                    semantic_digest: preparation.semantic_digest.clone(),
284                    artifact: artifact.clone(),
285                    storage: storage.clone(),
286                },
287                options.clone(),
288            )
289            .await;
290        if let Err(error) = start
291            && error.submission_disposition() == SubmissionDisposition::NotSubmitted
292        {
293            return Err(PortabilityWorkflowError::Client(error));
294        }
295        let request = ResolveCollectionRestoreRequest {
296            action_id,
297            semantic_digest: preparation.semantic_digest,
298            artifact,
299            storage,
300        };
301        let mut resolved = self
302            .resolve_collection_restore(request.clone(), options.clone())
303            .await?;
304        let mut operation = resolved_operation(resolved.operation, resolved.expired_unknown)?;
305        if operation.current_state == OperationState::Pending {
306            let _waited = self.wait_portability_operation(&operation, options.clone()).await?;
307            resolved = self.resolve_collection_restore(request, options).await?;
308            operation = resolved_operation(resolved.operation, resolved.expired_unknown)?;
309        }
310        Ok(CollectionRestoreExecutionV1 { operation })
311    }
312
313    /// Captures one bounded live Collection through the Core-owned admission
314    /// fence and resolves its immutable base-plus-delta artifact. The complete
315    /// prepare request is the caller's durable recovery identity.
316    pub async fn export_collection_transfer(
317        &self,
318        request: PrepareCollectionTransferExportRequest,
319        options: CallOptions,
320    ) -> Result<CollectionTransferExportExecutionV1, PortabilityWorkflowError> {
321        let preparation = self
322            .prepare_collection_transfer_export(request.clone(), options.clone())
323            .await?;
324        let start = self
325            .start_collection_transfer_export(
326                StartCollectionTransferExportRequest {
327                    action_id: request.action_id.clone(),
328                    semantic_digest: preparation.semantic_digest.clone(),
329                    expected_security_generation: request.expected_security_generation.clone(),
330                    storage: request.storage.clone(),
331                },
332                options.clone(),
333            )
334            .await;
335        if let Err(error) = start
336            && error.submission_disposition() == SubmissionDisposition::NotSubmitted
337        {
338            return Err(PortabilityWorkflowError::Client(error));
339        }
340        let resolve_request = ResolveCollectionTransferExportRequest {
341            action_id: request.action_id,
342            semantic_digest: preparation.semantic_digest,
343            expected_security_generation: request.expected_security_generation,
344            storage: request.storage,
345        };
346        let mut resolved = self
347            .resolve_collection_transfer_export(resolve_request.clone(), options.clone())
348            .await?;
349        let mut operation = resolved_operation(resolved.operation, resolved.expired_unknown)?;
350        if operation.current_state == OperationState::Pending {
351            let _waited = self.wait_portability_operation(&operation, options.clone()).await?;
352            resolved = self
353                .resolve_collection_transfer_export(resolve_request, options)
354                .await?;
355            operation = resolved_operation(resolved.operation, resolved.expired_unknown)?;
356        }
357        if operation.current_state == OperationState::Succeeded && resolved.artifact.is_none() {
358            return Err(PortabilityWorkflowError::MissingArtifact);
359        }
360        Ok(CollectionTransferExportExecutionV1 {
361            operation,
362            artifact: resolved.artifact,
363        })
364    }
365
366    /// Restores and verifies one exact fenced Collection transfer artifact.
367    /// Submitted ambiguity is resolved only with the original action,
368    /// artifact and storage capability.
369    pub async fn restore_collection_transfer(
370        &self,
371        action_id: Digest,
372        artifact: CollectionTransferArtifact,
373        storage: PortabilityStorageCapability,
374        options: CallOptions,
375    ) -> Result<CollectionTransferRestoreExecutionV1, PortabilityWorkflowError> {
376        let preparation = self
377            .prepare_collection_transfer_restore(
378                PrepareCollectionTransferRestoreRequest {
379                    action_id: action_id.clone(),
380                    artifact: artifact.clone(),
381                    storage: storage.clone(),
382                },
383                options.clone(),
384            )
385            .await?;
386        let expected_final_content_digest = preparation.final_content_digest.clone();
387        let start = self
388            .start_collection_transfer_restore(
389                StartCollectionTransferRestoreRequest {
390                    action_id: action_id.clone(),
391                    semantic_digest: preparation.semantic_digest.clone(),
392                    artifact: artifact.clone(),
393                    storage: storage.clone(),
394                },
395                options.clone(),
396            )
397            .await;
398        if let Err(error) = start
399            && error.submission_disposition() == SubmissionDisposition::NotSubmitted
400        {
401            return Err(PortabilityWorkflowError::Client(error));
402        }
403        let resolve_request = ResolveCollectionTransferRestoreRequest {
404            action_id,
405            semantic_digest: preparation.semantic_digest,
406            artifact,
407            storage,
408        };
409        let mut resolved = self
410            .resolve_collection_transfer_restore(resolve_request.clone(), options.clone())
411            .await?;
412        let mut operation = resolved_operation(resolved.operation, resolved.expired_unknown)?;
413        if operation.current_state == OperationState::Pending {
414            let _waited = self.wait_portability_operation(&operation, options.clone()).await?;
415            resolved = self
416                .resolve_collection_transfer_restore(resolve_request, options)
417                .await?;
418            operation = resolved_operation(resolved.operation, resolved.expired_unknown)?;
419        }
420        if operation.current_state == OperationState::Succeeded && resolved.verified_content_digest.is_none() {
421            return Err(PortabilityWorkflowError::MissingVerification);
422        }
423        if operation.current_state == OperationState::Succeeded
424            && resolved.verified_content_digest.as_ref() != Some(&expected_final_content_digest)
425        {
426            return Err(PortabilityWorkflowError::VerificationMismatch);
427        }
428        Ok(CollectionTransferRestoreExecutionV1 {
429            operation,
430            verified_content_digest: resolved.verified_content_digest,
431        })
432    }
433
434    /// Exports one exact source index into bounded, independently verifiable
435    /// transfer pages. The original resource identity and caller-selected
436    /// bounds remain part of every recovery request.
437    pub async fn export_index(
438        &self,
439        request: PrepareIndexExportRequest,
440        options: CallOptions,
441    ) -> Result<IndexExportExecutionV1, PortabilityWorkflowError> {
442        let preparation = self.prepare_index_export(request.clone(), options.clone()).await?;
443        let start = self
444            .start_index_export(
445                StartIndexExportRequest {
446                    action_id: request.action_id.clone(),
447                    semantic_digest: preparation.semantic_digest.clone(),
448                    index: request.index.clone(),
449                    storage: request.storage.clone(),
450                    maximum_records: request.maximum_records,
451                    maximum_pages: request.maximum_pages,
452                    maximum_page_bytes: request.maximum_page_bytes.clone(),
453                },
454                options.clone(),
455            )
456            .await;
457        if let Err(error) = start
458            && error.submission_disposition() == SubmissionDisposition::NotSubmitted
459        {
460            return Err(PortabilityWorkflowError::Client(error));
461        }
462        let resolve_request = ResolveIndexExportRequest {
463            action_id: request.action_id,
464            semantic_digest: preparation.semantic_digest,
465            index: request.index,
466            storage: request.storage,
467            maximum_records: request.maximum_records,
468            maximum_pages: request.maximum_pages,
469            maximum_page_bytes: request.maximum_page_bytes,
470        };
471        let mut resolved = self
472            .resolve_index_export(resolve_request.clone(), options.clone())
473            .await?;
474        let mut operation = resolved_operation(resolved.operation, resolved.expired_unknown)?;
475        if operation.current_state == OperationState::Pending {
476            let _waited = self.wait_portability_operation(&operation, options.clone()).await?;
477            resolved = self.resolve_index_export(resolve_request, options).await?;
478            operation = resolved_operation(resolved.operation, resolved.expired_unknown)?;
479        }
480        if operation.current_state == OperationState::Succeeded && resolved.artifact.is_none() {
481            return Err(PortabilityWorkflowError::MissingArtifact);
482        }
483        Ok(IndexExportExecutionV1 {
484            operation,
485            artifact: resolved.artifact,
486        })
487    }
488
489    /// Imports one verified transfer artifact into a fresh destination name.
490    /// Destination policy, topology, and storage choices are caller-owned and
491    /// remain bound to the same recovery identity.
492    pub async fn import_index(
493        &self,
494        request: PrepareIndexImportRequest,
495        options: CallOptions,
496    ) -> Result<IndexImportExecutionV1, PortabilityWorkflowError> {
497        let preparation = self.prepare_index_import(request.clone(), options.clone()).await?;
498        let start = self
499            .start_index_import(
500                StartIndexImportRequest {
501                    action_id: request.action_id.clone(),
502                    semantic_digest: preparation.semantic_digest.clone(),
503                    artifact: request.artifact.clone(),
504                    storage: request.storage.clone(),
505                    destination_name: request.destination_name.clone(),
506                    expected_name_absent: request.expected_name_absent,
507                    destination_policies: request.destination_policies.clone(),
508                    destination_resource_spec: request.destination_resource_spec.clone(),
509                },
510                options.clone(),
511            )
512            .await;
513        if let Err(error) = start
514            && error.submission_disposition() == SubmissionDisposition::NotSubmitted
515        {
516            return Err(PortabilityWorkflowError::Client(error));
517        }
518        let resolve_request = ResolveIndexImportRequest {
519            action_id: request.action_id,
520            semantic_digest: preparation.semantic_digest,
521            artifact: request.artifact,
522            storage: request.storage,
523            destination_name: request.destination_name,
524            expected_name_absent: request.expected_name_absent,
525            destination_policies: request.destination_policies,
526            destination_resource_spec: request.destination_resource_spec,
527        };
528        let mut resolved = self
529            .resolve_index_import(resolve_request.clone(), options.clone())
530            .await?;
531        let mut operation = resolved_operation(resolved.operation, resolved.expired_unknown)?;
532        if operation.current_state == OperationState::Pending {
533            let _waited = self.wait_portability_operation(&operation, options.clone()).await?;
534            resolved = self.resolve_index_import(resolve_request, options).await?;
535            operation = resolved_operation(resolved.operation, resolved.expired_unknown)?;
536        }
537        if operation.current_state == OperationState::Succeeded && resolved.destination.is_none() {
538            return Err(PortabilityWorkflowError::MissingDestination);
539        }
540        Ok(IndexImportExecutionV1 {
541            operation,
542            destination: resolved.destination,
543        })
544    }
545
546    async fn wait_portability_operation(
547        &self,
548        operation: &Operation,
549        options: CallOptions,
550    ) -> Result<Operation, PortabilityWorkflowError> {
551        Ok(self
552            .wait_for_operation(
553                operation.locator.clone(),
554                operation.authority_revision.clone(),
555                OperationWaitOptions {
556                    maximum_attempts: PORTABILITY_WAIT_ATTEMPTS,
557                    interval_millis: PORTABILITY_WAIT_INTERVAL_MILLIS,
558                },
559                options,
560            )
561            .await
562            .map_err(PortabilityWorkflowError::Wait)?
563            .operation)
564    }
565}
566
567#[cfg(test)]
568mod tests {
569    use super::*;
570
571    #[test]
572    fn portability_wait_contract_is_bounded() {
573        assert!((1..=64).contains(&PORTABILITY_WAIT_ATTEMPTS));
574        assert!((1..=60_000).contains(&PORTABILITY_WAIT_INTERVAL_MILLIS));
575        assert_eq!(PORTABILITY_WORKFLOW_FORMAT_VERSION, 4);
576    }
577}
lodestone_client/portability.rs

lodestone_client/
portability.rs
