aster_forge_cloud_files_core/
upload.rs

1//! Product-neutral resumable content-upload requests, backend port, and durable recovery model.
2
3use std::{fmt, num::NonZeroUsize};
4
5use async_trait::async_trait;
6use bytes::Bytes;
7
8use crate::{
9    BackendResult, CloudBackendError, CloudBackendErrorKind, CloudFilesCoreError,
10    CloudFilesStoreError, ContentCacheKey, ContentLeaseId, ContentUploadStore, IdempotencyKey,
11    LocalContentGeneration, LocalContentSnapshot, MutationRemoteOutcome, OperationId, Result,
12    SessionGeneration, StoreResult, StoreWriteStatus,
13};
14
15/// Opaque backend upload-session identity.
16#[derive(Clone, PartialEq, Eq, Hash)]
17pub struct ContentUploadSessionId(String);
18
19impl ContentUploadSessionId {
20    /// Creates a non-empty backend session identity.
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("content upload session id"));
29        }
30        Ok(Self(value))
31    }
32
33    /// Returns the opaque backend session identity.
34    #[must_use]
35    pub fn as_str(&self) -> &str {
36        &self.0
37    }
38}
39
40impl fmt::Debug for ContentUploadSessionId {
41    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
42        formatter
43            .debug_struct("ContentUploadSessionId")
44            .field("byte_len", &self.0.len())
45            .finish()
46    }
47}
48
49/// Immutable upload intent persisted before acquiring or resuming a backend upload session.
50#[derive(Debug, Clone, PartialEq, Eq)]
51pub struct ContentUploadIntent {
52    operation_id: OperationId,
53    idempotency_key: IdempotencyKey,
54    base_cache_key: ContentCacheKey,
55    snapshot: LocalContentSnapshot,
56    upload_lease_id: ContentLeaseId,
57    session_generation: SessionGeneration,
58}
59
60impl ContentUploadIntent {
61    /// Creates an upload bound to one base remote revision and immutable local generation.
62    /// # Errors
63    ///
64    /// Returns an error when validation fails or an underlying backend, store, or platform
65    /// operation fails.
66    pub fn new(
67        operation_id: OperationId,
68        idempotency_key: IdempotencyKey,
69        base_cache_key: ContentCacheKey,
70        snapshot: LocalContentSnapshot,
71        upload_lease_id: ContentLeaseId,
72        session_generation: SessionGeneration,
73    ) -> Result<Self> {
74        if base_cache_key.item_key() != snapshot.item_key() {
75            return Err(CloudFilesCoreError::invalid_content_upload(
76                "upload snapshot item does not match the base cache key",
77            ));
78        }
79        Ok(Self {
80            operation_id,
81            idempotency_key,
82            base_cache_key,
83            snapshot,
84            upload_lease_id,
85            session_generation,
86        })
87    }
88
89    /// Returns the durable operation identity shared with mutation reconciliation.
90    #[must_use]
91    pub const fn operation_id(&self) -> &OperationId {
92        &self.operation_id
93    }
94
95    /// Returns the backend idempotency/status reconciliation key.
96    #[must_use]
97    pub const fn idempotency_key(&self) -> &IdempotencyKey {
98        &self.idempotency_key
99    }
100
101    /// Returns the exact remote content revision used as the conditional upload base.
102    #[must_use]
103    pub const fn base_cache_key(&self) -> &ContentCacheKey {
104        &self.base_cache_key
105    }
106
107    /// Returns the immutable local source snapshot.
108    #[must_use]
109    pub const fn snapshot(&self) -> &LocalContentSnapshot {
110        &self.snapshot
111    }
112
113    /// Returns the operation-owned upload lease identity.
114    #[must_use]
115    pub const fn upload_lease_id(&self) -> &ContentLeaseId {
116        &self.upload_lease_id
117    }
118
119    /// Returns the platform session generation that accepted the upload.
120    #[must_use]
121    pub const fn session_generation(&self) -> SessionGeneration {
122        self.session_generation
123    }
124}
125
126/// Backend upload session and the next byte offset that remains to be sent.
127#[derive(Debug, Clone, PartialEq, Eq)]
128pub struct ContentUploadSession {
129    id: ContentUploadSessionId,
130    accepted_offset: u64,
131}
132
133impl ContentUploadSession {
134    /// Creates a backend upload-session checkpoint.
135    #[must_use]
136    pub const fn new(id: ContentUploadSessionId, accepted_offset: u64) -> Self {
137        Self {
138            id,
139            accepted_offset,
140        }
141    }
142
143    /// Returns the backend upload-session identity.
144    #[must_use]
145    pub const fn id(&self) -> &ContentUploadSessionId {
146        &self.id
147    }
148
149    /// Returns the first local byte not yet accepted by the backend.
150    #[must_use]
151    pub const fn accepted_offset(&self) -> u64 {
152        self.accepted_offset
153    }
154
155    /// Validates that the backend checkpoint remains within the immutable snapshot.
156    /// # Errors
157    ///
158    /// Returns an error when validation fails or an underlying backend, store, or platform
159    /// operation fails.
160    pub fn validate_for(&self, intent: &ContentUploadIntent) -> Result<()> {
161        if self.accepted_offset > intent.snapshot().size() {
162            return Err(CloudFilesCoreError::invalid_content_upload(
163                "backend upload offset exceeds the immutable snapshot size",
164            ));
165        }
166        Ok(())
167    }
168}
169
170/// One immutable local byte chunk submitted to a backend upload session.
171#[derive(Debug, Clone, PartialEq, Eq)]
172pub struct ContentUploadChunk {
173    session_id: ContentUploadSessionId,
174    local_generation: LocalContentGeneration,
175    offset: u64,
176    bytes: Bytes,
177    byte_len: u64,
178    total_size: u64,
179}
180
181impl ContentUploadChunk {
182    /// Creates a non-empty chunk within the immutable snapshot size.
183    /// # Errors
184    ///
185    /// Returns an error when validation fails or an underlying backend, store, or platform
186    /// operation fails.
187    pub fn new(
188        session_id: ContentUploadSessionId,
189        local_generation: LocalContentGeneration,
190        offset: u64,
191        bytes: Bytes,
192        total_size: u64,
193    ) -> Result<Self> {
194        let byte_len = bytes.len() as u64;
195        if byte_len == 0 {
196            return Err(CloudFilesCoreError::invalid_content_upload(
197                "upload chunks must contain at least one byte",
198            ));
199        }
200        let end = offset.checked_add(byte_len).ok_or_else(|| {
201            CloudFilesCoreError::invalid_content_upload("upload chunk end exceeds u64")
202        })?;
203        if end > total_size {
204            return Err(CloudFilesCoreError::invalid_content_upload(
205                "upload chunk exceeds the immutable snapshot size",
206            ));
207        }
208        Ok(Self::from_validated_parts(
209            session_id,
210            local_generation,
211            offset,
212            bytes,
213            byte_len,
214            total_size,
215        ))
216    }
217
218    fn from_validated_parts(
219        session_id: ContentUploadSessionId,
220        local_generation: LocalContentGeneration,
221        offset: u64,
222        bytes: Bytes,
223        byte_len: u64,
224        total_size: u64,
225    ) -> Self {
226        Self {
227            session_id,
228            local_generation,
229            offset,
230            bytes,
231            byte_len,
232            total_size,
233        }
234    }
235
236    /// Returns the backend upload-session identity.
237    pub const fn session_id(&self) -> &ContentUploadSessionId {
238        &self.session_id
239    }
240
241    /// Returns the immutable local generation whose bytes are carried.
242    pub const fn local_generation(&self) -> LocalContentGeneration {
243        self.local_generation
244    }
245
246    /// Returns the first byte offset in the immutable snapshot.
247    pub const fn offset(&self) -> u64 {
248        self.offset
249    }
250
251    /// Returns the owned upload bytes.
252    pub const fn bytes(&self) -> &Bytes {
253        &self.bytes
254    }
255
256    /// Returns the number of bytes in this chunk.
257    pub const fn byte_len(&self) -> u64 {
258        self.byte_len
259    }
260
261    /// Returns the exclusive end acknowledged when this chunk commits exactly.
262    pub const fn end_exclusive(&self) -> u64 {
263        self.offset + self.byte_len
264    }
265
266    /// Returns the complete immutable snapshot size.
267    pub const fn total_size(&self) -> u64 {
268        self.total_size
269    }
270
271    /// Validates that this chunk continues the exact durable session and immutable snapshot.
272    /// # Errors
273    ///
274    /// Returns an error when validation fails or an underlying backend, store, or platform
275    /// operation fails.
276    pub fn validate_for(
277        &self,
278        intent: &ContentUploadIntent,
279        session: &ContentUploadSession,
280    ) -> Result<()> {
281        if &self.session_id != session.id() {
282            return Err(CloudFilesCoreError::invalid_content_upload(
283                "upload chunk belongs to another backend session",
284            ));
285        }
286        if self.local_generation != intent.snapshot().generation() {
287            return Err(CloudFilesCoreError::invalid_content_upload(
288                "upload chunk belongs to another local generation",
289            ));
290        }
291        if self.total_size != intent.snapshot().size() {
292            return Err(CloudFilesCoreError::invalid_content_upload(
293                "upload chunk size does not match the immutable snapshot",
294            ));
295        }
296        if self.offset != session.accepted_offset() {
297            return Err(CloudFilesCoreError::invalid_content_upload(
298                "upload chunk does not continue the durable accepted offset",
299            ));
300        }
301        Ok(())
302    }
303}
304
305/// Backend acknowledgement of one exact upload chunk.
306#[derive(Debug, Clone, PartialEq, Eq)]
307pub struct ContentUploadChunkAck {
308    session_id: ContentUploadSessionId,
309    accepted_offset: u64,
310}
311
312impl ContentUploadChunkAck {
313    /// Creates an acknowledgement whose offset is validated by the submitted chunk.
314    #[must_use]
315    pub const fn new(session_id: ContentUploadSessionId, accepted_offset: u64) -> Self {
316        Self {
317            session_id,
318            accepted_offset,
319        }
320    }
321
322    /// Returns the backend upload-session identity.
323    #[must_use]
324    pub const fn session_id(&self) -> &ContentUploadSessionId {
325        &self.session_id
326    }
327
328    /// Returns the first byte not yet accepted by the backend.
329    #[must_use]
330    pub const fn accepted_offset(&self) -> u64 {
331        self.accepted_offset
332    }
333
334    /// Validates exact, monotonic acknowledgement of the submitted chunk.
335    /// # Errors
336    ///
337    /// Returns an error when validation fails or an underlying backend, store, or platform
338    /// operation fails.
339    pub fn validate_for(&self, chunk: &ContentUploadChunk) -> Result<()> {
340        if self.session_id != chunk.session_id {
341            return Err(CloudFilesCoreError::invalid_content_upload(
342                "upload acknowledgement belongs to another session",
343            ));
344        }
345        if self.accepted_offset != chunk.end_exclusive() {
346            return Err(CloudFilesCoreError::invalid_content_upload(
347                "upload acknowledgement does not match the submitted chunk end",
348            ));
349        }
350        Ok(())
351    }
352}
353
354/// Product-owned resumable upload transport.
355#[async_trait]
356pub trait CloudContentUploadBackend: Send + Sync {
357    /// Starts or idempotently resumes a backend session for one immutable upload intent.
358    async fn start_upload(
359        &self,
360        intent: &ContentUploadIntent,
361    ) -> BackendResult<ContentUploadSession>;
362
363    /// Uploads one exact immutable chunk.
364    async fn upload_chunk(
365        &self,
366        intent: &ContentUploadIntent,
367        chunk: &ContentUploadChunk,
368    ) -> BackendResult<ContentUploadChunkAck>;
369
370    /// Reconciles a chunk that may already have been accepted before its durable checkpoint.
371    ///
372    /// Backends with strict offset semantics return the current accepted offset for the exact
373    /// `(session, offset, bytes)` replay.
374    async fn reconcile_upload_chunk(
375        &self,
376        intent: &ContentUploadIntent,
377        session: &ContentUploadSession,
378        chunk: &ContentUploadChunk,
379    ) -> BackendResult<ContentUploadChunkAck>;
380
381    /// Conditionally and idempotently commits the fully uploaded content.
382    ///
383    /// The runner may call this again from a durable `RemoteCommitting` record. If transport
384    /// completion leaves the remote effect ambiguous, return
385    /// [`MutationRemoteOutcome::RemoteOutcomeUnknown`] instead of a backend error so recovery uses
386    /// [`Self::reconcile_upload`].
387    async fn commit_upload(
388        &self,
389        intent: &ContentUploadIntent,
390        session: &ContentUploadSession,
391    ) -> BackendResult<MutationRemoteOutcome>;
392
393    /// Reconciles a commit whose transport outcome did not prove whether it succeeded.
394    async fn reconcile_upload(
395        &self,
396        intent: &ContentUploadIntent,
397    ) -> BackendResult<MutationRemoteOutcome>;
398}
399
400/// Product-owned reader for immutable local snapshot bytes.
401///
402/// The reader must resolve the exact opaque [`LocalContentSnapshot::reference`] and return the
403/// complete requested range from that immutable generation. Filesystem paths, database rows, and
404/// provider-cache layouts remain private to the product implementation.
405#[async_trait]
406pub trait LocalContentSnapshotReader: Send + Sync {
407    /// Reads one non-empty range from an immutable local snapshot.
408    async fn read_snapshot(
409        &self,
410        snapshot: &LocalContentSnapshot,
411        offset: u64,
412        length: u64,
413    ) -> StoreResult<Bytes>;
414}
415
416/// Result returned by one upload-runner invocation.
417pub type ContentUploadRunResult<T> = std::result::Result<T, ContentUploadRunError>;
418
419/// Product-neutral failure while driving one durable upload record.
420#[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)]
421pub enum ContentUploadRunError {
422    /// The product-owned remote upload adapter failed.
423    #[error(transparent)]
424    Backend(#[from] CloudBackendError),
425    /// The durable upload store or immutable snapshot reader failed.
426    #[error(transparent)]
427    Store(#[from] CloudFilesStoreError),
428    /// A source, backend, or durable record violated the upload contract.
429    #[error(transparent)]
430    Contract(#[from] CloudFilesCoreError),
431    /// The requested operation has no durable upload record.
432    #[error("content upload record was not found")]
433    RecordNotFound,
434}
435
436/// Stable outcome of one upload-runner invocation.
437#[derive(Debug, Clone, PartialEq, Eq)]
438pub enum ContentUploadRunOutcome {
439    /// The upload, metadata reconciliation, and upload-lease release are durable.
440    Completed,
441    /// Remote commit may have happened, but reconciliation still cannot prove its outcome.
442    RemoteOutcomePending,
443    /// A newer active platform session fenced this executor before its next durable transition.
444    Fenced,
445}
446
447/// Runtime-neutral driver for resumable immutable-snapshot uploads.
448///
449/// The runner does not spawn tasks, sleep, choose retry delays, allocate operation identities, or
450/// own product transport. One invocation advances a durable record until it completes, reaches an
451/// unknown remote outcome, is fenced, or returns the first backend/store/contract failure.
452#[derive(Debug, Clone, Copy)]
453pub struct ContentUploadRunner {
454    chunk_size: NonZeroUsize,
455}
456
457impl ContentUploadRunner {
458    /// Creates a runner with a non-zero in-memory chunk size.
459    /// # Errors
460    ///
461    /// Returns an error when validation fails or an underlying backend, store, or platform
462    /// operation fails.
463    pub fn new(chunk_size: usize) -> Result<Self> {
464        let chunk_size = NonZeroUsize::new(chunk_size).ok_or_else(|| {
465            CloudFilesCoreError::invalid_content_upload("upload runner chunk size must be non-zero")
466        })?;
467        Ok(Self { chunk_size })
468    }
469
470    /// Returns the maximum immutable byte count submitted in one backend chunk.
471    #[must_use]
472    pub const fn chunk_size(&self) -> usize {
473        self.chunk_size.get()
474    }
475
476    /// Persists a caller-owned upload intent, then advances its durable record.
477    /// # Errors
478    ///
479    /// Returns an error when validation fails or an underlying backend, store, or platform
480    /// operation fails.
481    pub async fn submit(
482        &self,
483        intent: ContentUploadIntent,
484        execution_generation: SessionGeneration,
485        store: &dyn ContentUploadStore,
486        backend: &dyn CloudContentUploadBackend,
487        source: &dyn LocalContentSnapshotReader,
488    ) -> ContentUploadRunResult<ContentUploadRunOutcome> {
489        if store
490            .persist_content_upload_intent(intent.clone(), execution_generation)
491            .await?
492            == StoreWriteStatus::Fenced
493        {
494            return Ok(ContentUploadRunOutcome::Fenced);
495        }
496        self.resume(
497            intent.operation_id(),
498            execution_generation,
499            store,
500            backend,
501            source,
502        )
503        .await
504    }
505
506    /// Advances one already-persisted upload using the active executor generation.
507    /// # Errors
508    ///
509    /// Returns an error when validation fails or an underlying backend, store, or platform
510    /// operation fails.
511    #[expect(
512        clippy::too_many_lines,
513        reason = "the upload recovery loop keeps every durable state transition in one auditable match"
514    )]
515    pub async fn resume(
516        &self,
517        operation_id: &OperationId,
518        execution_generation: SessionGeneration,
519        store: &dyn ContentUploadStore,
520        backend: &dyn CloudContentUploadBackend,
521        source: &dyn LocalContentSnapshotReader,
522    ) -> ContentUploadRunResult<ContentUploadRunOutcome> {
523        loop {
524            let record = store
525                .load_content_upload(operation_id)
526                .await?
527                .ok_or(ContentUploadRunError::RecordNotFound)?;
528            match record.state() {
529                ContentUploadState::IntentPersisted => {
530                    let session = backend.start_upload(record.intent()).await?;
531                    session.validate_for(record.intent())?;
532                    if observe_transition(
533                        store
534                            .record_content_upload_session(
535                                operation_id,
536                                session,
537                                execution_generation,
538                            )
539                            .await?,
540                        &record,
541                        operation_id,
542                        store,
543                    )
544                    .await?
545                    {
546                        return Ok(ContentUploadRunOutcome::Fenced);
547                    }
548                }
549                ContentUploadState::Uploading => {
550                    let session = required_upload_session(
551                        &record,
552                        "uploading record omitted its backend session",
553                    )?;
554                    session.validate_for(record.intent())?;
555                    if session.accepted_offset() < record.intent().snapshot().size() {
556                        let remaining =
557                            record.intent().snapshot().size() - session.accepted_offset();
558                        let length = remaining.min(self.chunk_size.get() as u64);
559                        let bytes = source
560                            .read_snapshot(
561                                record.intent().snapshot(),
562                                session.accepted_offset(),
563                                length,
564                            )
565                            .await?;
566                        validate_source_bytes(&bytes, length)?;
567                        let chunk = ContentUploadChunk::from_validated_parts(
568                            session.id().clone(),
569                            record.intent().snapshot().generation(),
570                            session.accepted_offset(),
571                            bytes,
572                            length,
573                            record.intent().snapshot().size(),
574                        );
575                        let acknowledgement =
576                            match backend.upload_chunk(record.intent(), &chunk).await {
577                                Ok(acknowledgement) => acknowledgement,
578                                Err(error)
579                                    if matches!(
580                                        error.kind(),
581                                        CloudBackendErrorKind::Conflict
582                                            | CloudBackendErrorKind::PreconditionFailed
583                                    ) =>
584                                {
585                                    backend
586                                        .reconcile_upload_chunk(record.intent(), session, &chunk)
587                                        .await?
588                                }
589                                Err(error) => return Err(error.into()),
590                            };
591                        acknowledgement.validate_for(&chunk)?;
592                        let checkpoint = ContentUploadSession::new(
593                            acknowledgement.session_id().clone(),
594                            acknowledgement.accepted_offset(),
595                        );
596                        if observe_transition(
597                            store
598                                .record_content_upload_session(
599                                    operation_id,
600                                    checkpoint,
601                                    execution_generation,
602                                )
603                                .await?,
604                            &record,
605                            operation_id,
606                            store,
607                        )
608                        .await?
609                        {
610                            return Ok(ContentUploadRunOutcome::Fenced);
611                        }
612                    } else if observe_transition(
613                        store
614                            .begin_content_upload_remote_commit(operation_id, execution_generation)
615                            .await?,
616                        &record,
617                        operation_id,
618                        store,
619                    )
620                    .await?
621                    {
622                        return Ok(ContentUploadRunOutcome::Fenced);
623                    }
624                }
625                ContentUploadState::RemoteCommitting => {
626                    let session = required_upload_session(
627                        &record,
628                        "remote-committing record omitted its backend session",
629                    )?;
630                    let outcome = backend.commit_upload(record.intent(), session).await?;
631                    let pending = matches!(outcome, MutationRemoteOutcome::RemoteOutcomeUnknown);
632                    if observe_transition(
633                        store
634                            .record_content_upload_remote_outcome(
635                                operation_id,
636                                outcome,
637                                execution_generation,
638                            )
639                            .await?,
640                        &record,
641                        operation_id,
642                        store,
643                    )
644                    .await?
645                    {
646                        return Ok(ContentUploadRunOutcome::Fenced);
647                    }
648                    if pending {
649                        return Ok(ContentUploadRunOutcome::RemoteOutcomePending);
650                    }
651                }
652                ContentUploadState::RemoteOutcomeUnknown => {
653                    let outcome = backend.reconcile_upload(record.intent()).await?;
654                    let pending = matches!(outcome, MutationRemoteOutcome::RemoteOutcomeUnknown);
655                    let status = store
656                        .record_content_upload_remote_outcome(
657                            operation_id,
658                            outcome,
659                            execution_generation,
660                        )
661                        .await?;
662                    if status == StoreWriteStatus::Fenced {
663                        return Ok(ContentUploadRunOutcome::Fenced);
664                    }
665                    let current = store
666                        .load_content_upload(operation_id)
667                        .await?
668                        .ok_or(ContentUploadRunError::RecordNotFound)?;
669                    if current == record {
670                        if pending
671                            && status == StoreWriteStatus::AlreadyApplied
672                            && current.state() == ContentUploadState::RemoteOutcomeUnknown
673                        {
674                            return Ok(ContentUploadRunOutcome::RemoteOutcomePending);
675                        }
676                        return Err(CloudFilesCoreError::invalid_content_upload(
677                            "upload store reported a transition without durable progress",
678                        )
679                        .into());
680                    }
681                }
682                ContentUploadState::RemoteOutcomeKnown => {
683                    if observe_transition(
684                        store
685                            .reconcile_content_upload_metadata(operation_id, execution_generation)
686                            .await?,
687                        &record,
688                        operation_id,
689                        store,
690                    )
691                    .await?
692                    {
693                        return Ok(ContentUploadRunOutcome::Fenced);
694                    }
695                }
696                ContentUploadState::MetadataReconciled => {
697                    if observe_transition(
698                        store
699                            .complete_content_upload(operation_id, execution_generation)
700                            .await?,
701                        &record,
702                        operation_id,
703                        store,
704                    )
705                    .await?
706                    {
707                        return Ok(ContentUploadRunOutcome::Fenced);
708                    }
709                }
710                ContentUploadState::Completed => {
711                    validate_completed_upload(&record)?;
712                    return Ok(ContentUploadRunOutcome::Completed);
713                }
714            }
715        }
716    }
717}
718
719fn required_upload_session<'a>(
720    record: &'a ContentUploadRecord,
721    missing_reason: &'static str,
722) -> Result<&'a ContentUploadSession> {
723    record
724        .session()
725        .ok_or_else(|| CloudFilesCoreError::invalid_content_upload(missing_reason))
726}
727
728fn validate_completed_upload(record: &ContentUploadRecord) -> Result<()> {
729    record
730        .remote_outcome()
731        .ok_or_else(|| {
732            CloudFilesCoreError::invalid_content_upload(
733                "completed upload record omitted its remote outcome",
734            )
735        })
736        .map(|_| ())
737}
738
739fn validate_source_bytes(bytes: &Bytes, expected: u64) -> Result<()> {
740    let actual = bytes.len() as u64;
741    if actual != expected {
742        return Err(CloudFilesCoreError::invalid_content_upload(
743            "local upload source returned a partial or oversized range",
744        ));
745    }
746    Ok(())
747}
748
749async fn observe_transition(
750    status: StoreWriteStatus,
751    previous: &ContentUploadRecord,
752    operation_id: &OperationId,
753    store: &dyn ContentUploadStore,
754) -> ContentUploadRunResult<bool> {
755    let current = store
756        .load_content_upload(operation_id)
757        .await?
758        .ok_or(ContentUploadRunError::RecordNotFound)?;
759    if &current != previous {
760        return Ok(false);
761    }
762    if status == StoreWriteStatus::Fenced {
763        return Ok(true);
764    }
765    Err(CloudFilesCoreError::invalid_content_upload(
766        "upload store reported a transition without durable progress",
767    )
768    .into())
769}
770
771/// Durable upload state.
772#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
773pub enum ContentUploadState {
774    /// Intent, dirty generation, and upload lease are durable.
775    IntentPersisted,
776    /// A backend session and resumable accepted offset are durable.
777    Uploading,
778    /// Every byte was accepted and remote commit has started.
779    RemoteCommitting,
780    /// Transport completion did not prove whether remote commit happened.
781    RemoteOutcomeUnknown,
782    /// A committed, already-committed, or precondition-failed outcome is durable.
783    RemoteOutcomeKnown,
784    /// Local metadata acknowledged the outcome without clearing a newer dirty generation.
785    MetadataReconciled,
786    /// The operation is terminal and its upload lease was released.
787    Completed,
788}
789
790/// Result of an idempotent upload-record transition.
791#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
792pub enum ContentUploadRecordTransition {
793    /// Durable state changed.
794    Applied,
795    /// Equivalent or later durable state already contains the transition.
796    AlreadyApplied,
797    /// A newer durable offset or local generation superseded the transition.
798    Fenced,
799}
800
801/// Recoverable resumable upload journal record.
802#[derive(Debug, Clone, PartialEq, Eq)]
803pub struct ContentUploadRecord {
804    intent: ContentUploadIntent,
805    state: ContentUploadState,
806    session: Option<ContentUploadSession>,
807    remote_outcome: Option<MutationRemoteOutcome>,
808}
809
810impl ContentUploadRecord {
811    /// Creates the first recoverable upload state.
812    #[must_use]
813    pub const fn persist(intent: ContentUploadIntent) -> Self {
814        Self {
815            intent,
816            state: ContentUploadState::IntentPersisted,
817            session: None,
818            remote_outcome: None,
819        }
820    }
821
822    /// Returns the immutable upload intent.
823    #[must_use]
824    pub const fn intent(&self) -> &ContentUploadIntent {
825        &self.intent
826    }
827
828    /// Returns the durable upload state.
829    #[must_use]
830    pub const fn state(&self) -> ContentUploadState {
831        self.state
832    }
833
834    /// Returns the durable backend session checkpoint.
835    #[must_use]
836    pub const fn session(&self) -> Option<&ContentUploadSession> {
837        self.session.as_ref()
838    }
839
840    /// Returns the durable remote outcome.
841    #[must_use]
842    pub const fn remote_outcome(&self) -> Option<&MutationRemoteOutcome> {
843        self.remote_outcome.as_ref()
844    }
845
846    /// Records or advances the backend session's accepted offset.
847    /// # Errors
848    ///
849    /// Returns an error when validation fails or an underlying backend, store, or platform
850    /// operation fails.
851    pub fn record_session(
852        &mut self,
853        session: ContentUploadSession,
854    ) -> Result<ContentUploadRecordTransition> {
855        session.validate_for(&self.intent)?;
856        if !matches!(
857            self.state,
858            ContentUploadState::IntentPersisted | ContentUploadState::Uploading
859        ) {
860            if self.session.as_ref() == Some(&session) {
861                return Ok(ContentUploadRecordTransition::AlreadyApplied);
862            }
863            return Err(CloudFilesCoreError::invalid_content_upload(
864                "upload session checkpoint must precede remote commit",
865            ));
866        }
867        let Some(current) = self.session.as_ref() else {
868            self.state = ContentUploadState::Uploading;
869            self.session = Some(session);
870            return Ok(ContentUploadRecordTransition::Applied);
871        };
872        if current.id() != session.id() {
873            return Err(CloudFilesCoreError::invalid_content_upload(
874                "one upload operation cannot switch backend session identity",
875            ));
876        }
877        if session.accepted_offset() < current.accepted_offset() {
878            return Ok(ContentUploadRecordTransition::Fenced);
879        }
880        if session.accepted_offset() == current.accepted_offset() {
881            return Ok(ContentUploadRecordTransition::AlreadyApplied);
882        }
883        self.session = Some(session);
884        Ok(ContentUploadRecordTransition::Applied)
885    }
886
887    /// Marks remote commit started after all immutable bytes were accepted.
888    /// # Errors
889    ///
890    /// Returns an error when validation fails or an underlying backend, store, or platform
891    /// operation fails.
892    pub fn begin_remote_commit(&mut self) -> Result<ContentUploadRecordTransition> {
893        match self.state {
894            ContentUploadState::Uploading => {
895                let complete = self.session.as_ref().is_some_and(|session| {
896                    session.accepted_offset() == self.intent.snapshot().size()
897                });
898                if !complete {
899                    return Err(CloudFilesCoreError::invalid_content_upload(
900                        "remote commit requires every immutable byte to be accepted",
901                    ));
902                }
903                self.state = ContentUploadState::RemoteCommitting;
904                Ok(ContentUploadRecordTransition::Applied)
905            }
906            ContentUploadState::RemoteCommitting
907            | ContentUploadState::RemoteOutcomeUnknown
908            | ContentUploadState::RemoteOutcomeKnown
909            | ContentUploadState::MetadataReconciled
910            | ContentUploadState::Completed => Ok(ContentUploadRecordTransition::AlreadyApplied),
911            ContentUploadState::IntentPersisted => {
912                Err(CloudFilesCoreError::invalid_content_upload(
913                    "remote commit requires a backend upload session",
914                ))
915            }
916        }
917    }
918
919    /// Records a known or unknown conditional remote outcome.
920    /// # Errors
921    ///
922    /// Returns an error when validation fails or an underlying backend, store, or platform
923    /// operation fails.
924    pub fn record_remote_outcome(
925        &mut self,
926        outcome: MutationRemoteOutcome,
927    ) -> Result<ContentUploadRecordTransition> {
928        if let MutationRemoteOutcome::Committed { item }
929        | MutationRemoteOutcome::AlreadyCommitted { item } = &outcome
930        {
931            let Some(item) = item.as_ref() else {
932                return Err(CloudFilesCoreError::invalid_content_upload(
933                    "committed upload outcome requires current item metadata",
934                ));
935            };
936            if item.key() != self.intent.base_cache_key().item_key() {
937                return Err(CloudFilesCoreError::invalid_content_upload(
938                    "committed upload outcome belongs to another item",
939                ));
940            }
941            let Some(content) = item.content() else {
942                return Err(CloudFilesCoreError::invalid_content_upload(
943                    "committed upload outcome must describe file content",
944                ));
945            };
946            if content.size() != self.intent.snapshot().size() {
947                return Err(CloudFilesCoreError::invalid_content_upload(
948                    "committed upload size does not match the immutable snapshot",
949                ));
950            }
951        }
952        let next_state = if matches!(outcome, MutationRemoteOutcome::RemoteOutcomeUnknown) {
953            ContentUploadState::RemoteOutcomeUnknown
954        } else {
955            ContentUploadState::RemoteOutcomeKnown
956        };
957        let same_outcome = self.remote_outcome.as_ref() == Some(&outcome);
958        let outcome_is_already_durable = matches!(
959            self.state,
960            ContentUploadState::RemoteOutcomeUnknown
961                | ContentUploadState::RemoteOutcomeKnown
962                | ContentUploadState::MetadataReconciled
963                | ContentUploadState::Completed
964        );
965        if same_outcome && outcome_is_already_durable {
966            return Ok(ContentUploadRecordTransition::AlreadyApplied);
967        }
968        if !matches!(
969            self.state,
970            ContentUploadState::RemoteCommitting | ContentUploadState::RemoteOutcomeUnknown
971        ) {
972            return Err(CloudFilesCoreError::invalid_content_upload(
973                "remote upload outcome must follow commit or reconcile an unknown outcome",
974            ));
975        }
976        self.state = next_state;
977        self.remote_outcome = Some(outcome);
978        Ok(ContentUploadRecordTransition::Applied)
979    }
980
981    /// Marks local metadata reconciled after a known remote outcome.
982    /// # Errors
983    ///
984    /// Returns an error when validation fails or an underlying backend, store, or platform
985    /// operation fails.
986    pub fn mark_metadata_reconciled(&mut self) -> Result<ContentUploadRecordTransition> {
987        match self.state {
988            ContentUploadState::RemoteOutcomeKnown => {
989                self.state = ContentUploadState::MetadataReconciled;
990                Ok(ContentUploadRecordTransition::Applied)
991            }
992            ContentUploadState::MetadataReconciled | ContentUploadState::Completed => {
993                Ok(ContentUploadRecordTransition::AlreadyApplied)
994            }
995            _ => Err(CloudFilesCoreError::invalid_content_upload(
996                "upload metadata reconciliation requires a known remote outcome",
997            )),
998        }
999    }
1000
1001    /// Marks the upload terminal after metadata reconciliation.
1002    /// # Errors
1003    ///
1004    /// Returns an error when validation fails or an underlying backend, store, or platform
1005    /// operation fails.
1006    pub fn complete(&mut self) -> Result<ContentUploadRecordTransition> {
1007        match self.state {
1008            ContentUploadState::MetadataReconciled => {
1009                self.state = ContentUploadState::Completed;
1010                Ok(ContentUploadRecordTransition::Applied)
1011            }
1012            ContentUploadState::Completed => Ok(ContentUploadRecordTransition::AlreadyApplied),
1013            _ => Err(CloudFilesCoreError::invalid_content_upload(
1014                "upload completion requires metadata reconciliation",
1015            )),
1016        }
1017    }
1018}
1019
1020#[cfg(test)]
1021mod tests {
1022    use super::*;
1023    use crate::{
1024        CloudItemId, CloudItemKey, CloudNamespaceId, CloudRootId, CloudScope, LocalContentReference,
1025    };
1026
1027    fn intent() -> ContentUploadIntent {
1028        let item_key = CloudItemKey::new(
1029            CloudScope::new(
1030                CloudNamespaceId::new("namespace").expect("namespace fixture should be valid"),
1031                CloudRootId::new("root").expect("root fixture should be valid"),
1032            ),
1033            CloudItemId::new("item").expect("item fixture should be valid"),
1034        );
1035        let snapshot = LocalContentSnapshot::new(
1036            item_key.clone(),
1037            LocalContentGeneration::new(1).expect("local generation fixture should be valid"),
1038            LocalContentReference::new("local-reference")
1039                .expect("local reference fixture should be valid"),
1040            4,
1041            None,
1042        );
1043        ContentUploadIntent::new(
1044            OperationId::new("operation").expect("operation fixture should be valid"),
1045            IdempotencyKey::new("idempotency").expect("idempotency fixture should be valid"),
1046            ContentCacheKey::new(
1047                item_key,
1048                crate::ContentRevision::from_slice(b"content-v1")
1049                    .expect("content revision fixture should be valid"),
1050            ),
1051            snapshot,
1052            ContentLeaseId::new("upload-lease").expect("lease fixture should be valid"),
1053            SessionGeneration::new(1).expect("generation fixture should be valid"),
1054        )
1055        .expect("upload intent fixture should be valid")
1056    }
1057
1058    #[test]
1059    fn corrupted_runner_records_require_sessions_and_completed_outcomes() {
1060        let mut record = ContentUploadRecord::persist(intent());
1061        record.state = ContentUploadState::Uploading;
1062        assert_eq!(
1063            required_upload_session(&record, "uploading record omitted its backend session"),
1064            Err(CloudFilesCoreError::InvalidContentUploadTransition {
1065                reason: "uploading record omitted its backend session",
1066            })
1067        );
1068
1069        record.state = ContentUploadState::RemoteCommitting;
1070        assert_eq!(
1071            required_upload_session(
1072                &record,
1073                "remote-committing record omitted its backend session",
1074            ),
1075            Err(CloudFilesCoreError::InvalidContentUploadTransition {
1076                reason: "remote-committing record omitted its backend session",
1077            })
1078        );
1079
1080        record.state = ContentUploadState::Completed;
1081        assert_eq!(
1082            validate_completed_upload(&record),
1083            Err(CloudFilesCoreError::InvalidContentUploadTransition {
1084                reason: "completed upload record omitted its remote outcome",
1085            })
1086        );
1087    }
1088}
1089
1090#[cfg(test)]
1091#[path = "upload_runner_tests.rs"]
1092mod runner_tests;
aster_forge_cloud_files_core/upload.rs

aster_forge_cloud_files_core/
upload.rs
