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

lodestone_client/
portability.rs
