aster_forge_cloud_files_core/
mutation.rs

1//! Durable product-neutral mutation identities, intents, outcomes, and recovery states.
2
3use std::{fmt, num::NonZeroU64, time::SystemTime};
4
5use async_trait::async_trait;
6
7use crate::{
8    BackendResult, CloudBackendError, CloudFilesCoreError, CloudFilesStoreError, CloudItem,
9    CloudItemId, CloudItemKey, CloudItemKind, CloudScope, ContentRevision, LocalContentSnapshot,
10    MetadataRevision, MutationJournalStore, Result, StoreWriteStatus,
11};
12
13macro_rules! opaque_string {
14    ($name:ident, $field:literal, $docs:literal) => {
15        #[doc = $docs]
16        #[derive(Clone, PartialEq, Eq, Hash)]
17        pub struct $name(String);
18
19        impl $name {
20            /// Creates a non-empty opaque value and preserves it exactly.
21            /// # Errors
22            ///
23            /// Returns an error when validation fails or an underlying backend, store, or platform
24            /// operation fails.
25            pub fn new(value: impl Into<String>) -> Result<Self> {
26                let value = value.into();
27                if value.is_empty() {
28                    return Err(CloudFilesCoreError::empty($field));
29                }
30                Ok(Self(value))
31            }
32
33            /// Returns the opaque value.
34            pub fn as_str(&self) -> &str {
35                &self.0
36            }
37
38            /// Consumes the wrapper and returns the opaque value.
39            pub fn into_string(self) -> String {
40                self.0
41            }
42        }
43
44        impl fmt::Debug for $name {
45            fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
46                formatter
47                    .debug_struct(stringify!($name))
48                    .field("byte_len", &self.0.len())
49                    .finish()
50            }
51        }
52    };
53}
54
55opaque_string!(
56    OperationId,
57    "operation id",
58    "Stable identity of one durable mutation journal operation."
59);
60opaque_string!(
61    IdempotencyKey,
62    "idempotency key",
63    "Opaque backend reconciliation key used to detect an already committed mutation."
64);
65opaque_string!(
66    PlatformRequestCorrelation,
67    "platform request correlation",
68    "Owned correlation token for a platform request; native pointers and handles are excluded."
69);
70opaque_string!(
71    RemoteReconciliationKey,
72    "remote reconciliation key",
73    "Opaque key used by status queries or change-feed reconciliation."
74);
75
76/// Monotonic fence for one platform connection, extension instance, or mount session.
77#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash)]
78pub struct SessionGeneration(NonZeroU64);
79
80impl SessionGeneration {
81    /// Creates a non-zero session generation.
82    /// # Errors
83    ///
84    /// Returns an error when validation fails or an underlying backend, store, or platform
85    /// operation fails.
86    pub const fn new(value: u64) -> Result<Self> {
87        match NonZeroU64::new(value) {
88            Some(value) => Ok(Self(value)),
89            None => Err(CloudFilesCoreError::InvalidSessionGeneration),
90        }
91    }
92
93    /// Returns the numeric fence value.
94    #[must_use]
95    pub const fn get(self) -> u64 {
96        self.0.get()
97    }
98}
99
100/// Lifecycle of one active platform session generation.
101#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
102pub enum SessionState {
103    /// New requests may be accepted.
104    Accepting,
105    /// New requests are rejected while accepted work is being classified for drain or cancel.
106    Closing,
107    /// Accepted work is being drained, cancelled, or explicitly failed.
108    Draining,
109    /// The generation no longer accepts completions that mutate active session state.
110    Closed,
111}
112
113impl SessionState {
114    /// Returns whether moving from this state to `next` is a valid idempotent lifecycle step.
115    #[must_use]
116    pub const fn can_transition_to(self, next: Self) -> bool {
117        matches!(
118            (self, next),
119            (Self::Accepting, Self::Accepting | Self::Closing)
120                | (Self::Closing, Self::Closing | Self::Draining)
121                | (Self::Draining, Self::Draining | Self::Closed)
122                | (Self::Closed, Self::Closed)
123        )
124    }
125}
126
127/// How a local or remote effect entered the shared mutation/reconciliation mechanism.
128#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
129pub enum MutationOrigin {
130    /// A native operation waits for durable acceptance and the configured remote outcome.
131    PlatformCommand,
132    /// A native operation requires an allow, deny, or acknowledgement before local completion.
133    PlatformPreflight,
134    /// The local or platform effect already happened and must be journaled and reconciled.
135    PlatformObserved,
136    /// A durable remote change is being reconciled into platform state.
137    RemoteChange,
138}
139
140/// Conditional revisions captured when a mutation intent is created.
141#[derive(Debug, Clone, Default, PartialEq, Eq)]
142pub struct MutationPreconditions {
143    metadata_revision: Option<MetadataRevision>,
144    content_revision: Option<ContentRevision>,
145}
146
147impl MutationPreconditions {
148    /// Creates independent metadata and content revision preconditions.
149    #[must_use]
150    pub const fn new(
151        metadata_revision: Option<MetadataRevision>,
152        content_revision: Option<ContentRevision>,
153    ) -> Self {
154        Self {
155            metadata_revision,
156            content_revision,
157        }
158    }
159
160    /// Returns the expected metadata revision, when one is required.
161    #[must_use]
162    pub const fn metadata_revision(&self) -> Option<&MetadataRevision> {
163        self.metadata_revision.as_ref()
164    }
165
166    /// Returns the expected content revision, when one is required.
167    #[must_use]
168    pub const fn content_revision(&self) -> Option<&ContentRevision> {
169        self.content_revision.as_ref()
170    }
171}
172
173/// Product-neutral desired effect of a durable mutation.
174#[derive(Debug, Clone, PartialEq, Eq)]
175pub enum DesiredMutation {
176    /// Creates a child below an existing directory.
177    Create {
178        /// Namespace/root scope in which the child will be created.
179        scope: CloudScope,
180        /// Stable identity of the desired parent.
181        parent_id: CloudItemId,
182        /// Desired child name, preserved without product normalization.
183        name: String,
184        /// Desired item kind.
185        kind: CloudItemKind,
186    },
187    /// Replaces local content for an existing file.
188    ModifyContent {
189        /// Stable identity of the file being modified.
190        key: CloudItemKey,
191    },
192    /// Changes the parent, name, or both for an existing item.
193    ModifyMetadata {
194        /// Stable identity preserved by the metadata mutation.
195        key: CloudItemKey,
196        /// Desired parent when this mutation moves the item.
197        parent_id: Option<CloudItemId>,
198        /// Desired name when this mutation renames the item.
199        name: Option<String>,
200    },
201    /// Deletes an existing item.
202    Delete {
203        /// Stable identity of the item being deleted.
204        key: CloudItemKey,
205    },
206}
207
208impl DesiredMutation {
209    /// Creates a child intent with a non-empty opaque name.
210    /// # Errors
211    ///
212    /// Returns an error when validation fails or an underlying backend, store, or platform
213    /// operation fails.
214    pub fn create(
215        scope: CloudScope,
216        parent_id: CloudItemId,
217        name: impl Into<String>,
218        kind: CloudItemKind,
219    ) -> Result<Self> {
220        let name = name.into();
221        if name.is_empty() {
222            return Err(CloudFilesCoreError::empty("mutation create name"));
223        }
224        Ok(Self::Create {
225            scope,
226            parent_id,
227            name,
228            kind,
229        })
230    }
231
232    /// Creates a content replacement intent.
233    #[must_use]
234    pub const fn modify_content(key: CloudItemKey) -> Self {
235        Self::ModifyContent { key }
236    }
237
238    /// Creates a metadata intent that changes the parent, name, or both.
239    /// # Errors
240    ///
241    /// Returns an error when validation fails or an underlying backend, store, or platform
242    /// operation fails.
243    pub fn modify_metadata(
244        key: CloudItemKey,
245        parent_id: Option<CloudItemId>,
246        name: Option<String>,
247    ) -> Result<Self> {
248        if parent_id.is_none() && name.is_none() {
249            return Err(CloudFilesCoreError::invalid_mutation_intent(
250                "metadata mutation must change parent or name",
251            ));
252        }
253        if name.as_ref().is_some_and(String::is_empty) {
254            return Err(CloudFilesCoreError::empty("mutation metadata name"));
255        }
256        Ok(Self::ModifyMetadata {
257            key,
258            parent_id,
259            name,
260        })
261    }
262
263    /// Creates a delete intent.
264    #[must_use]
265    pub const fn delete(key: CloudItemKey) -> Self {
266        Self::Delete { key }
267    }
268
269    /// Returns the namespace/root scope affected by this mutation.
270    #[must_use]
271    pub const fn scope(&self) -> &CloudScope {
272        match self {
273            Self::Create { scope, .. } => scope,
274            Self::ModifyContent { key }
275            | Self::ModifyMetadata { key, .. }
276            | Self::Delete { key } => key.scope(),
277        }
278    }
279
280    /// Returns the existing item key when the mutation targets an already identified item.
281    #[must_use]
282    pub const fn existing_item_key(&self) -> Option<&CloudItemKey> {
283        match self {
284            Self::Create { .. } => None,
285            Self::ModifyContent { key }
286            | Self::ModifyMetadata { key, .. }
287            | Self::Delete { key } => Some(key),
288        }
289    }
290
291    const fn requires_local_content(&self) -> bool {
292        matches!(self, Self::ModifyContent { .. })
293    }
294}
295
296/// Retry metadata persisted with a mutation intent.
297#[derive(Debug, Clone, PartialEq, Eq)]
298pub struct MutationRetryMetadata {
299    attempt: u32,
300    not_before: Option<SystemTime>,
301}
302
303impl MutationRetryMetadata {
304    /// Creates retry metadata from an attempt count and optional absolute retry time.
305    #[must_use]
306    pub const fn new(attempt: u32, not_before: Option<SystemTime>) -> Self {
307        Self {
308            attempt,
309            not_before,
310        }
311    }
312
313    /// Returns how many remote attempts have already started.
314    #[must_use]
315    pub const fn attempt(&self) -> u32 {
316        self.attempt
317    }
318
319    /// Returns the earliest retry time, when one is scheduled.
320    #[must_use]
321    pub const fn not_before(&self) -> Option<SystemTime> {
322        self.not_before
323    }
324}
325
326impl Default for MutationRetryMetadata {
327    fn default() -> Self {
328        Self::new(0, None)
329    }
330}
331
332/// In-memory intent that must be durably inserted before acknowledgement.
333#[derive(Debug, Clone, PartialEq, Eq)]
334pub struct MutationIntent {
335    operation_id: OperationId,
336    idempotency_key: IdempotencyKey,
337    origin: MutationOrigin,
338    session_generation: SessionGeneration,
339    desired: DesiredMutation,
340    preconditions: MutationPreconditions,
341    local_content: Option<LocalContentSnapshot>,
342    platform_correlation: Option<PlatformRequestCorrelation>,
343    remote_reconciliation_key: Option<RemoteReconciliationKey>,
344    retry: MutationRetryMetadata,
345}
346
347impl MutationIntent {
348    /// Creates an intent. Call [`Self::validate_for_persistence`] before durable insertion.
349    #[must_use]
350    pub const fn new(
351        operation_id: OperationId,
352        idempotency_key: IdempotencyKey,
353        origin: MutationOrigin,
354        session_generation: SessionGeneration,
355        desired: DesiredMutation,
356        preconditions: MutationPreconditions,
357    ) -> Self {
358        Self {
359            operation_id,
360            idempotency_key,
361            origin,
362            session_generation,
363            desired,
364            preconditions,
365            local_content: None,
366            platform_correlation: None,
367            remote_reconciliation_key: None,
368            retry: MutationRetryMetadata::new(0, None),
369        }
370    }
371
372    /// Attaches an owned local-content reference.
373    #[must_use]
374    pub fn with_local_content(mut self, local_content: LocalContentSnapshot) -> Self {
375        self.local_content = Some(local_content);
376        self
377    }
378
379    /// Attaches an owned platform correlation token.
380    #[must_use]
381    pub fn with_platform_correlation(
382        mut self,
383        platform_correlation: PlatformRequestCorrelation,
384    ) -> Self {
385        self.platform_correlation = Some(platform_correlation);
386        self
387    }
388
389    /// Attaches a backend status/change-feed reconciliation key.
390    #[must_use]
391    pub fn with_remote_reconciliation_key(
392        mut self,
393        remote_reconciliation_key: RemoteReconciliationKey,
394    ) -> Self {
395        self.remote_reconciliation_key = Some(remote_reconciliation_key);
396        self
397    }
398
399    /// Replaces persisted retry metadata.
400    #[must_use]
401    pub fn with_retry(mut self, retry: MutationRetryMetadata) -> Self {
402        self.retry = retry;
403        self
404    }
405
406    /// Validates invariants needed before the intent becomes recoverable.
407    /// # Errors
408    ///
409    /// Returns an error when validation fails or an underlying backend, store, or platform
410    /// operation fails.
411    pub fn validate_for_persistence(&self) -> Result<()> {
412        if self.desired.requires_local_content() && self.local_content.is_none() {
413            return Err(CloudFilesCoreError::invalid_mutation_intent(
414                "content mutation requires a local content reference",
415            ));
416        }
417        if !self.desired.requires_local_content() && self.local_content.is_some() {
418            return Err(CloudFilesCoreError::invalid_mutation_intent(
419                "local content snapshot is only valid for content mutation",
420            ));
421        }
422        if let (Some(key), Some(snapshot)) = (
423            self.desired.existing_item_key(),
424            self.local_content.as_ref(),
425        ) && snapshot.item_key() != key
426        {
427            return Err(CloudFilesCoreError::invalid_mutation_intent(
428                "local content snapshot item does not match the mutation target",
429            ));
430        }
431        Ok(())
432    }
433
434    /// Returns the operation identity.
435    #[must_use]
436    pub const fn operation_id(&self) -> &OperationId {
437        &self.operation_id
438    }
439
440    /// Returns the backend idempotency key.
441    #[must_use]
442    pub const fn idempotency_key(&self) -> &IdempotencyKey {
443        &self.idempotency_key
444    }
445
446    /// Returns how the mutation entered the shared mechanism.
447    #[must_use]
448    pub const fn origin(&self) -> MutationOrigin {
449        self.origin
450    }
451
452    /// Returns the platform session generation captured by the intent.
453    #[must_use]
454    pub const fn session_generation(&self) -> SessionGeneration {
455        self.session_generation
456    }
457
458    /// Returns the desired product-neutral mutation.
459    #[must_use]
460    pub const fn desired(&self) -> &DesiredMutation {
461        &self.desired
462    }
463
464    /// Returns independent metadata/content revision preconditions.
465    #[must_use]
466    pub const fn preconditions(&self) -> &MutationPreconditions {
467        &self.preconditions
468    }
469
470    /// Returns the owned local-content reference, when needed.
471    #[must_use]
472    pub const fn local_content(&self) -> Option<&LocalContentSnapshot> {
473        self.local_content.as_ref()
474    }
475
476    /// Returns the owned platform request correlation token.
477    #[must_use]
478    pub const fn platform_correlation(&self) -> Option<&PlatformRequestCorrelation> {
479        self.platform_correlation.as_ref()
480    }
481
482    /// Returns the remote status/change-feed reconciliation key.
483    #[must_use]
484    pub const fn remote_reconciliation_key(&self) -> Option<&RemoteReconciliationKey> {
485        self.remote_reconciliation_key.as_ref()
486    }
487
488    /// Returns persisted retry metadata.
489    #[must_use]
490    pub const fn retry(&self) -> &MutationRetryMetadata {
491        &self.retry
492    }
493}
494
495/// Durable journal state. `Detected` is deliberately absent because it is not recoverable.
496#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
497pub enum MutationState {
498    /// The complete intent is durable and may be acknowledged according to ingress policy.
499    IntentPersisted,
500    /// A remote effect is in flight.
501    RemoteApplying,
502    /// Transport completion did not prove whether the backend committed the effect.
503    RemoteOutcomeUnknown,
504    /// A committed, already-committed, or precondition-failed outcome is durable.
505    RemoteOutcomeKnown,
506    /// Required platform state has been reconciled with the durable remote outcome.
507    PlatformReconciled,
508    /// The operation is terminal and replay must not repeat its remote effect.
509    Completed,
510}
511
512/// Product-neutral result of one conditional remote mutation.
513#[derive(Debug, Clone, PartialEq, Eq)]
514pub enum MutationRemoteOutcome {
515    /// The backend committed the requested effect during this attempt.
516    Committed {
517        /// Current item state when the operation has one; delete may return `None`.
518        item: Option<CloudItem>,
519    },
520    /// Reconciliation proved that an earlier attempt already committed the effect.
521    AlreadyCommitted {
522        /// Current item state when the operation has one; delete may return `None`.
523        item: Option<CloudItem>,
524    },
525    /// One or more revision preconditions no longer match remote state.
526    PreconditionFailed {
527        /// Current metadata revision when known.
528        metadata_revision: Option<MetadataRevision>,
529        /// Current content revision when known.
530        content_revision: Option<ContentRevision>,
531    },
532    /// The transport result does not prove whether the backend committed the effect.
533    RemoteOutcomeUnknown,
534}
535
536impl MutationRemoteOutcome {
537    const fn durable_state(&self) -> MutationState {
538        match self {
539            Self::RemoteOutcomeUnknown => MutationState::RemoteOutcomeUnknown,
540            Self::Committed { .. }
541            | Self::AlreadyCommitted { .. }
542            | Self::PreconditionFailed { .. } => MutationState::RemoteOutcomeKnown,
543        }
544    }
545
546    fn same_committed_effect(&self, other: &Self) -> bool {
547        match (self, other) {
548            (
549                Self::Committed { item: left } | Self::AlreadyCommitted { item: left },
550                Self::Committed { item: right } | Self::AlreadyCommitted { item: right },
551            ) => left == right,
552            _ => false,
553        }
554    }
555}
556
557/// Product-owned adapter for applying and reconciling durable mutations.
558///
559/// Implementations own transport, authentication, product DTOs, stable/provisional identity
560/// mapping, and backend-specific status queries. [`Self::apply_mutation`] must use the intent's
561/// idempotency identity. A transport result that does not prove whether the effect committed must
562/// return [`MutationRemoteOutcome::RemoteOutcomeUnknown`] rather than an ordinary backend error.
563#[async_trait]
564pub trait CloudMutationBackend: Send + Sync {
565    /// Applies one idempotently identified mutation to the product backend.
566    async fn apply_mutation(&self, intent: &MutationIntent)
567    -> BackendResult<MutationRemoteOutcome>;
568
569    /// Resolves an explicitly unknown remote outcome through status or change reconciliation.
570    async fn reconcile_mutation(
571        &self,
572        intent: &MutationIntent,
573    ) -> BackendResult<MutationRemoteOutcome>;
574}
575
576/// Result returned by one mutation-runner invocation.
577pub type MutationRunResult<T> = std::result::Result<T, MutationRunError>;
578
579/// Product-neutral failure while driving one durable mutation record.
580#[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)]
581pub enum MutationRunError {
582    /// The product-owned remote mutation adapter failed.
583    #[error(transparent)]
584    Backend(#[from] CloudBackendError),
585    /// The durable mutation journal failed.
586    #[error(transparent)]
587    Store(#[from] CloudFilesStoreError),
588    /// A backend outcome or durable record violated the mutation contract.
589    #[error(transparent)]
590    Contract(#[from] CloudFilesCoreError),
591    /// The requested operation has no durable mutation record.
592    #[error("mutation record was not found")]
593    RecordNotFound,
594}
595
596/// Stable outcome of one mutation-runner invocation.
597#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
598pub enum MutationRunOutcome {
599    /// Remote outcome, product/platform reconciliation, and completion are durable.
600    Completed,
601    /// Remote application may have committed, but reconciliation still has no known outcome.
602    RemoteOutcomePending,
603    /// A newer active platform session fenced this executor before its next durable transition.
604    Fenced,
605}
606
607/// Runtime-neutral driver for one durable product-neutral mutation.
608///
609/// The runner does not allocate identities, spawn tasks, sleep, select retry delays, call native
610/// platform APIs, or own product transport. One invocation advances a durable record until it
611/// completes, reaches an unknown remote outcome, is fenced, or returns the first backend, store,
612/// or contract failure.
613#[derive(Debug, Clone, Copy, Default)]
614pub struct MutationRunner;
615
616impl MutationRunner {
617    /// Persists a caller-owned intent, then advances its durable record.
618    /// # Errors
619    ///
620    /// Returns an error when validation fails or an underlying backend, store, or platform
621    /// operation fails.
622    pub async fn submit(
623        &self,
624        intent: MutationIntent,
625        execution_generation: SessionGeneration,
626        store: &dyn MutationJournalStore,
627        backend: &dyn CloudMutationBackend,
628    ) -> MutationRunResult<MutationRunOutcome> {
629        if intent.session_generation() != execution_generation {
630            return Err(CloudFilesCoreError::invalid_mutation_intent(
631                "submitted mutation generation must match the executor generation",
632            )
633            .into());
634        }
635        let operation_id = intent.operation_id().clone();
636        if store.persist_mutation_intent(intent).await? == StoreWriteStatus::Fenced {
637            return Ok(MutationRunOutcome::Fenced);
638        }
639        self.resume(&operation_id, execution_generation, store, backend)
640            .await
641    }
642
643    /// Advances one already-persisted mutation using the active executor generation.
644    /// # Errors
645    ///
646    /// Returns an error when validation fails or an underlying backend, store, or platform
647    /// operation fails.
648    pub async fn resume(
649        &self,
650        operation_id: &OperationId,
651        execution_generation: SessionGeneration,
652        store: &dyn MutationJournalStore,
653        backend: &dyn CloudMutationBackend,
654    ) -> MutationRunResult<MutationRunOutcome> {
655        loop {
656            let record = store
657                .load_mutation(operation_id)
658                .await?
659                .ok_or(MutationRunError::RecordNotFound)?;
660            match record.state() {
661                MutationState::IntentPersisted => {
662                    let status = store
663                        .begin_remote_apply(operation_id, execution_generation)
664                        .await?;
665                    if observe_mutation_transition(status, &record, operation_id, store, false)
666                        .await?
667                        .is_none()
668                    {
669                        return Ok(MutationRunOutcome::Fenced);
670                    }
671                }
672                MutationState::RemoteApplying => {
673                    let outcome = backend.apply_mutation(record.intent()).await?;
674                    validate_remote_outcome(record.intent(), &outcome)?;
675                    let pending = matches!(outcome, MutationRemoteOutcome::RemoteOutcomeUnknown);
676                    let status = store
677                        .record_remote_outcome(operation_id, execution_generation, outcome)
678                        .await?;
679                    let Some(current) =
680                        observe_mutation_transition(status, &record, operation_id, store, false)
681                            .await?
682                    else {
683                        return Ok(MutationRunOutcome::Fenced);
684                    };
685                    if pending && current.state() == MutationState::RemoteOutcomeUnknown {
686                        return Ok(MutationRunOutcome::RemoteOutcomePending);
687                    }
688                }
689                MutationState::RemoteOutcomeUnknown => {
690                    let outcome = backend.reconcile_mutation(record.intent()).await?;
691                    validate_remote_outcome(record.intent(), &outcome)?;
692                    let pending = matches!(outcome, MutationRemoteOutcome::RemoteOutcomeUnknown);
693                    let status = store
694                        .record_remote_outcome(operation_id, execution_generation, outcome)
695                        .await?;
696                    let Some(current) =
697                        observe_mutation_transition(status, &record, operation_id, store, pending)
698                            .await?
699                    else {
700                        return Ok(MutationRunOutcome::Fenced);
701                    };
702                    if pending && current.state() == MutationState::RemoteOutcomeUnknown {
703                        return Ok(MutationRunOutcome::RemoteOutcomePending);
704                    }
705                }
706                MutationState::RemoteOutcomeKnown => {
707                    validate_recorded_outcome(&record)?;
708                    let status = store
709                        .mark_platform_reconciled(operation_id, execution_generation)
710                        .await?;
711                    if observe_mutation_transition(status, &record, operation_id, store, false)
712                        .await?
713                        .is_none()
714                    {
715                        return Ok(MutationRunOutcome::Fenced);
716                    }
717                }
718                MutationState::PlatformReconciled => {
719                    validate_recorded_outcome(&record)?;
720                    let status = store
721                        .complete_mutation(operation_id, execution_generation)
722                        .await?;
723                    if observe_mutation_transition(status, &record, operation_id, store, false)
724                        .await?
725                        .is_none()
726                    {
727                        return Ok(MutationRunOutcome::Fenced);
728                    }
729                }
730                MutationState::Completed => {
731                    validate_recorded_outcome(&record)?;
732                    return Ok(MutationRunOutcome::Completed);
733                }
734            }
735        }
736    }
737}
738
739fn validate_recorded_outcome(record: &MutationRecord) -> Result<()> {
740    let outcome = record.remote_outcome().ok_or_else(|| {
741        CloudFilesCoreError::invalid_mutation_transition(
742            "known, reconciled, or completed mutation omitted its remote outcome",
743        )
744    })?;
745    if matches!(outcome, MutationRemoteOutcome::RemoteOutcomeUnknown) {
746        return Err(CloudFilesCoreError::invalid_mutation_transition(
747            "known, reconciled, or completed mutation retained an unknown remote outcome",
748        ));
749    }
750    validate_remote_outcome(record.intent(), outcome)
751}
752
753fn validate_remote_outcome(intent: &MutationIntent, outcome: &MutationRemoteOutcome) -> Result<()> {
754    let item = match outcome {
755        MutationRemoteOutcome::Committed { item }
756        | MutationRemoteOutcome::AlreadyCommitted { item } => item.as_ref(),
757        MutationRemoteOutcome::PreconditionFailed { .. }
758        | MutationRemoteOutcome::RemoteOutcomeUnknown => return Ok(()),
759    };
760
761    match intent.desired() {
762        DesiredMutation::Create {
763            scope,
764            parent_id,
765            name,
766            kind,
767        } => {
768            let item = item.ok_or_else(|| {
769                CloudFilesCoreError::invalid_mutation_transition(
770                    "committed create outcome requires current item metadata",
771                )
772            })?;
773            if item.key().scope() != scope {
774                return Err(CloudFilesCoreError::invalid_mutation_transition(
775                    "committed create outcome belongs to another scope",
776                ));
777            }
778            if item.parent_id() != Some(parent_id) {
779                return Err(CloudFilesCoreError::invalid_mutation_transition(
780                    "committed create outcome has another parent",
781                ));
782            }
783            if item.name() != name {
784                return Err(CloudFilesCoreError::invalid_mutation_transition(
785                    "committed create outcome has another name",
786                ));
787            }
788            if item.kind() != *kind {
789                return Err(CloudFilesCoreError::invalid_mutation_transition(
790                    "committed create outcome has another item kind",
791                ));
792            }
793        }
794        DesiredMutation::ModifyContent { key } => {
795            let item = required_existing_item(
796                item,
797                key,
798                "committed content mutation outcome requires current item metadata",
799                "committed content mutation outcome belongs to another item",
800            )?;
801            if item.kind() != CloudItemKind::File {
802                return Err(CloudFilesCoreError::invalid_mutation_transition(
803                    "committed content mutation outcome must describe a file",
804                ));
805            }
806        }
807        DesiredMutation::ModifyMetadata {
808            key,
809            parent_id,
810            name,
811        } => {
812            let item = required_existing_item(
813                item,
814                key,
815                "committed metadata mutation outcome requires current item metadata",
816                "committed metadata mutation outcome belongs to another item",
817            )?;
818            if parent_id
819                .as_ref()
820                .is_some_and(|parent_id| item.parent_id() != Some(parent_id))
821            {
822                return Err(CloudFilesCoreError::invalid_mutation_transition(
823                    "committed metadata mutation outcome did not apply the requested parent",
824                ));
825            }
826            if name
827                .as_ref()
828                .is_some_and(|name| item.name() != name.as_str())
829            {
830                return Err(CloudFilesCoreError::invalid_mutation_transition(
831                    "committed metadata mutation outcome did not apply the requested name",
832                ));
833            }
834        }
835        DesiredMutation::Delete { key } => {
836            if let Some(item) = item
837                && item.key() != key
838            {
839                return Err(CloudFilesCoreError::invalid_mutation_transition(
840                    "committed delete outcome belongs to another item",
841                ));
842            }
843        }
844    }
845    Ok(())
846}
847
848fn required_existing_item<'a>(
849    item: Option<&'a CloudItem>,
850    key: &CloudItemKey,
851    missing_reason: &'static str,
852    mismatch_reason: &'static str,
853) -> Result<&'a CloudItem> {
854    let item =
855        item.ok_or_else(|| CloudFilesCoreError::invalid_mutation_transition(missing_reason))?;
856    if item.key() != key {
857        return Err(CloudFilesCoreError::invalid_mutation_transition(
858            mismatch_reason,
859        ));
860    }
861    Ok(item)
862}
863
864async fn observe_mutation_transition(
865    status: StoreWriteStatus,
866    previous: &MutationRecord,
867    operation_id: &OperationId,
868    store: &dyn MutationJournalStore,
869    allow_stable_unknown: bool,
870) -> MutationRunResult<Option<MutationRecord>> {
871    let current = store
872        .load_mutation(operation_id)
873        .await?
874        .ok_or(MutationRunError::RecordNotFound)?;
875    if &current != previous {
876        return Ok(Some(current));
877    }
878    if status == StoreWriteStatus::Fenced {
879        return Ok(None);
880    }
881    if allow_stable_unknown
882        && status == StoreWriteStatus::AlreadyApplied
883        && current.state() == MutationState::RemoteOutcomeUnknown
884    {
885        return Ok(Some(current));
886    }
887    Err(CloudFilesCoreError::invalid_mutation_transition(
888        "mutation store reported a transition without durable progress",
889    )
890    .into())
891}
892
893/// Result of applying an idempotent mutation record transition.
894#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
895pub enum MutationRecordTransition {
896    /// The durable record changed state.
897    Applied,
898    /// The record already contained the same transition result.
899    AlreadyApplied,
900}
901
902/// Recoverable durable mutation journal record.
903#[derive(Debug, Clone, PartialEq, Eq)]
904pub struct MutationRecord {
905    intent: MutationIntent,
906    state: MutationState,
907    remote_outcome: Option<MutationRemoteOutcome>,
908}
909
910impl MutationRecord {
911    /// Converts a validated in-memory intent into its first recoverable journal state.
912    /// # Errors
913    ///
914    /// Returns an error when validation fails or an underlying backend, store, or platform
915    /// operation fails.
916    pub fn persist(intent: MutationIntent) -> Result<Self> {
917        intent.validate_for_persistence()?;
918        Ok(Self {
919            intent,
920            state: MutationState::IntentPersisted,
921            remote_outcome: None,
922        })
923    }
924
925    /// Returns the immutable persisted intent.
926    #[must_use]
927    pub const fn intent(&self) -> &MutationIntent {
928        &self.intent
929    }
930
931    /// Returns the current durable state.
932    #[must_use]
933    pub const fn state(&self) -> MutationState {
934        self.state
935    }
936
937    /// Returns the durable remote outcome, when one has been recorded.
938    #[must_use]
939    pub const fn remote_outcome(&self) -> Option<&MutationRemoteOutcome> {
940        self.remote_outcome.as_ref()
941    }
942
943    /// Marks that a remote effect has started.
944    /// # Errors
945    ///
946    /// Returns an error when validation fails or an underlying backend, store, or platform
947    /// operation fails.
948    pub fn begin_remote_apply(&mut self) -> Result<MutationRecordTransition> {
949        match self.state {
950            MutationState::IntentPersisted => {
951                self.state = MutationState::RemoteApplying;
952                Ok(MutationRecordTransition::Applied)
953            }
954            MutationState::RemoteApplying
955            | MutationState::RemoteOutcomeUnknown
956            | MutationState::RemoteOutcomeKnown
957            | MutationState::PlatformReconciled
958            | MutationState::Completed => Ok(MutationRecordTransition::AlreadyApplied),
959        }
960    }
961
962    /// Durably records a known or unknown remote outcome.
963    /// # Errors
964    ///
965    /// Returns an error when validation fails or an underlying backend, store, or platform
966    /// operation fails.
967    pub fn record_remote_outcome(
968        &mut self,
969        outcome: MutationRemoteOutcome,
970    ) -> Result<MutationRecordTransition> {
971        let next_state = outcome.durable_state();
972        if let Some(current) = self.remote_outcome.as_ref()
973            && (current == &outcome || current.same_committed_effect(&outcome))
974        {
975            return Ok(MutationRecordTransition::AlreadyApplied);
976        }
977        if !matches!(
978            self.state,
979            MutationState::RemoteApplying | MutationState::RemoteOutcomeUnknown
980        ) {
981            return Err(CloudFilesCoreError::invalid_mutation_transition(
982                "remote outcome must follow remote apply or reconcile an unknown outcome",
983            ));
984        }
985        self.state = next_state;
986        self.remote_outcome = Some(outcome);
987        Ok(MutationRecordTransition::Applied)
988    }
989
990    /// Marks required platform effects as reconciled.
991    /// # Errors
992    ///
993    /// Returns an error when validation fails or an underlying backend, store, or platform
994    /// operation fails.
995    pub fn mark_platform_reconciled(&mut self) -> Result<MutationRecordTransition> {
996        match self.state {
997            MutationState::RemoteOutcomeKnown => {
998                self.state = MutationState::PlatformReconciled;
999                Ok(MutationRecordTransition::Applied)
1000            }
1001            MutationState::PlatformReconciled | MutationState::Completed => {
1002                Ok(MutationRecordTransition::AlreadyApplied)
1003            }
1004            _ => Err(CloudFilesCoreError::invalid_mutation_transition(
1005                "platform reconciliation requires a known remote outcome",
1006            )),
1007        }
1008    }
1009
1010    /// Marks the operation complete after platform reconciliation.
1011    /// # Errors
1012    ///
1013    /// Returns an error when validation fails or an underlying backend, store, or platform
1014    /// operation fails.
1015    pub fn complete(&mut self) -> Result<MutationRecordTransition> {
1016        match self.state {
1017            MutationState::PlatformReconciled => {
1018                self.state = MutationState::Completed;
1019                Ok(MutationRecordTransition::Applied)
1020            }
1021            MutationState::Completed => Ok(MutationRecordTransition::AlreadyApplied),
1022            _ => Err(CloudFilesCoreError::invalid_mutation_transition(
1023                "completion requires platform reconciliation",
1024            )),
1025        }
1026    }
1027}
1028
1029#[cfg(test)]
1030mod tests {
1031    use super::*;
1032    use crate::{CloudItemId, CloudNamespaceId, CloudRootId, CloudScope};
1033
1034    fn intent() -> MutationIntent {
1035        let scope = CloudScope::new(
1036            CloudNamespaceId::new("namespace").expect("namespace fixture should be valid"),
1037            CloudRootId::new("root").expect("root fixture should be valid"),
1038        );
1039        let key = CloudItemKey::new(
1040            scope,
1041            CloudItemId::new("item").expect("item fixture should be valid"),
1042        );
1043        MutationIntent::new(
1044            OperationId::new("operation").expect("operation fixture should be valid"),
1045            IdempotencyKey::new("idempotency").expect("idempotency fixture should be valid"),
1046            MutationOrigin::PlatformCommand,
1047            SessionGeneration::new(1).expect("generation fixture should be valid"),
1048            DesiredMutation::delete(key),
1049            MutationPreconditions::default(),
1050        )
1051    }
1052
1053    #[test]
1054    fn corrupted_known_record_requires_a_non_unknown_remote_outcome() {
1055        let mut record = MutationRecord::persist(intent()).expect("intent should persist");
1056        record.state = MutationState::RemoteOutcomeKnown;
1057        assert_eq!(
1058            validate_recorded_outcome(&record),
1059            Err(CloudFilesCoreError::InvalidMutationTransition {
1060                reason: "known, reconciled, or completed mutation omitted its remote outcome",
1061            })
1062        );
1063
1064        record.remote_outcome = Some(MutationRemoteOutcome::RemoteOutcomeUnknown);
1065        assert_eq!(
1066            validate_recorded_outcome(&record),
1067            Err(CloudFilesCoreError::InvalidMutationTransition {
1068                reason: "known, reconciled, or completed mutation retained an unknown remote outcome",
1069            })
1070        );
1071    }
1072}
aster_forge_cloud_files_core/mutation.rs

aster_forge_cloud_files_core/
mutation.rs
