aster_forge_cloud_files_core/
store.rs

1//! Product-neutral durable-store ports for checkpoints, mutations, cache writes, uploads, and eviction.
2
3use std::fmt;
4
5use async_trait::async_trait;
6
7use crate::{
8    ByteRange, ChangeBatch, ChangeCursor, CloudScope, ContentCacheKey, ContentCacheWriteIntent,
9    ContentCacheWriteOperationId, ContentCacheWriteRecord, ContentEvictionBegin,
10    ContentEvictionIntent, ContentEvictionOperationId, ContentEvictionPhysicalEffect,
11    ContentEvictionRecord, ContentLeaseId, ContentLeaseKind, ContentStorageEntry,
12    ContentUploadIntent, ContentUploadRecord, ContentUploadSession, LocalContentGeneration,
13    LocalContentSnapshot, MutationIntent, MutationRecord, MutationRemoteOutcome, OperationId,
14    PlatformMaterializationState, SessionGeneration, SessionState,
15};
16
17/// Result returned by cloud-files durable-store ports.
18pub type StoreResult<T> = std::result::Result<T, CloudFilesStoreError>;
19
20/// One bounded, deterministically ordered recovery page.
21#[derive(Debug, Clone, PartialEq, Eq)]
22pub struct RecoveryPage<T> {
23    items: Vec<T>,
24    has_more: bool,
25}
26
27impl<T> RecoveryPage<T> {
28    /// Creates a page from at most `limit` records and an explicit continuation flag.
29    #[must_use]
30    pub fn new(items: Vec<T>, has_more: bool) -> Self {
31        Self { items, has_more }
32    }
33
34    /// Returns the records in this page.
35    #[must_use]
36    pub fn items(&self) -> &[T] {
37        &self.items
38    }
39
40    /// Returns whether another page exists after this one.
41    #[must_use]
42    pub const fn has_more(&self) -> bool {
43        self.has_more
44    }
45
46    /// Consumes the page and returns its records.
47    #[must_use]
48    pub fn into_items(self) -> Vec<T> {
49        self.items
50    }
51}
52
53/// Stable classification of a persistence-layer failure.
54#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
55pub enum CloudFilesStoreErrorKind {
56    /// A requested checkpoint, batch, session, content entry, or operation record does not exist.
57    NotFound,
58    /// Existing durable state conflicts with the requested write.
59    Conflict,
60    /// The requested state transition violates the durable protocol.
61    InvalidTransition,
62    /// The persistence implementation did not durably complete the requested operation.
63    PersistenceFailure,
64}
65
66/// Product-neutral store failure with adapter-owned diagnostic context.
67#[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)]
68#[error("cloud-files store {kind:?}: {context}")]
69pub struct CloudFilesStoreError {
70    kind: CloudFilesStoreErrorKind,
71    context: String,
72}
73
74impl CloudFilesStoreError {
75    /// Creates a classified persistence error.
76    pub fn new(kind: CloudFilesStoreErrorKind, context: impl Into<String>) -> Self {
77        Self {
78            kind,
79            context: context.into(),
80        }
81    }
82
83    /// Returns the stable error classification.
84    #[must_use]
85    pub const fn kind(&self) -> CloudFilesStoreErrorKind {
86        self.kind
87    }
88
89    /// Returns implementation diagnostic context. Product layers map user-visible text.
90    #[must_use]
91    pub fn context(&self) -> &str {
92        &self.context
93    }
94}
95
96/// Stable identity of one persisted change batch.
97#[derive(Clone, PartialEq, Eq, Hash)]
98pub struct ChangeBatchId(String);
99
100impl ChangeBatchId {
101    /// Creates a non-empty batch identity.
102    /// # Errors
103    ///
104    /// Returns an error when validation fails or an underlying backend, store, or platform
105    /// operation fails.
106    pub fn new(value: impl Into<String>) -> crate::Result<Self> {
107        let value = value.into();
108        if value.is_empty() {
109            return Err(crate::CloudFilesCoreError::empty("change batch id"));
110        }
111        Ok(Self(value))
112    }
113
114    /// Returns the opaque batch identity.
115    #[must_use]
116    pub fn as_str(&self) -> &str {
117        &self.0
118    }
119}
120
121impl fmt::Debug for ChangeBatchId {
122    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
123        formatter
124            .debug_struct("ChangeBatchId")
125            .field("byte_len", &self.0.len())
126            .finish()
127    }
128}
129
130/// Durable state of one pending change batch.
131#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
132pub enum PersistedChangeBatchState {
133    /// Batch bytes and base checkpoint are durable, but effects may not yet be replayable.
134    Recorded,
135    /// All effects needed after restart are durable and may be replayed idempotently.
136    EffectsReplayable,
137}
138
139/// Change batch persisted before its following cursor becomes active.
140#[derive(Debug, Clone, PartialEq, Eq)]
141pub struct PersistedChangeBatch {
142    id: ChangeBatchId,
143    scope: CloudScope,
144    base_cursor: Option<ChangeCursor>,
145    batch: ChangeBatch,
146    state: PersistedChangeBatchState,
147}
148
149impl PersistedChangeBatch {
150    /// Creates the first durable form of a fetched change batch.
151    #[must_use]
152    pub const fn new(
153        id: ChangeBatchId,
154        scope: CloudScope,
155        base_cursor: Option<ChangeCursor>,
156        batch: ChangeBatch,
157    ) -> Self {
158        Self {
159            id,
160            scope,
161            base_cursor,
162            batch,
163            state: PersistedChangeBatchState::Recorded,
164        }
165    }
166
167    /// Returns the batch identity.
168    #[must_use]
169    pub const fn id(&self) -> &ChangeBatchId {
170        &self.id
171    }
172
173    /// Returns the namespace/root checkpoint scope.
174    #[must_use]
175    pub const fn scope(&self) -> &CloudScope {
176        &self.scope
177    }
178
179    /// Returns the active cursor from which this batch was fetched.
180    #[must_use]
181    pub const fn base_cursor(&self) -> Option<&ChangeCursor> {
182        self.base_cursor.as_ref()
183    }
184
185    /// Returns the durable backend batch.
186    #[must_use]
187    pub const fn batch(&self) -> &ChangeBatch {
188        &self.batch
189    }
190
191    /// Returns whether replay prerequisites are durable.
192    #[must_use]
193    pub const fn state(&self) -> PersistedChangeBatchState {
194        self.state
195    }
196
197    /// Marks the batch effects replayable. The operation is idempotent.
198    pub fn mark_effects_replayable(&mut self) -> bool {
199        if self.state == PersistedChangeBatchState::EffectsReplayable {
200            return false;
201        }
202        self.state = PersistedChangeBatchState::EffectsReplayable;
203        true
204    }
205}
206
207/// Durable checkpoint snapshot for one namespace/root scope.
208#[derive(Debug, Clone, PartialEq, Eq)]
209pub struct ChangeCursorCheckpoint {
210    scope: CloudScope,
211    active_cursor: Option<ChangeCursor>,
212    pending_batch: Option<PersistedChangeBatch>,
213}
214
215impl ChangeCursorCheckpoint {
216    /// Creates an empty checkpoint at the initial backend position.
217    #[must_use]
218    pub const fn initial(scope: CloudScope) -> Self {
219        Self {
220            scope,
221            active_cursor: None,
222            pending_batch: None,
223        }
224    }
225
226    /// Creates a checkpoint snapshot returned by a store implementation.
227    #[must_use]
228    pub const fn new(
229        scope: CloudScope,
230        active_cursor: Option<ChangeCursor>,
231        pending_batch: Option<PersistedChangeBatch>,
232    ) -> Self {
233        Self {
234            scope,
235            active_cursor,
236            pending_batch,
237        }
238    }
239
240    /// Returns the namespace/root scope.
241    #[must_use]
242    pub const fn scope(&self) -> &CloudScope {
243        &self.scope
244    }
245
246    /// Returns the cursor that is safe to use for the next backend fetch.
247    #[must_use]
248    pub const fn active_cursor(&self) -> Option<&ChangeCursor> {
249        self.active_cursor.as_ref()
250    }
251
252    /// Returns the batch that must be replayed or completed before another fetch.
253    #[must_use]
254    pub const fn pending_batch(&self) -> Option<&PersistedChangeBatch> {
255        self.pending_batch.as_ref()
256    }
257}
258
259/// Result of an idempotent store write.
260#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
261pub enum StoreWriteStatus {
262    /// Durable state changed.
263    Applied,
264    /// Equivalent durable state already existed.
265    AlreadyApplied,
266    /// The active session generation or lifecycle fence rejected this write.
267    Fenced,
268}
269
270/// Durable two-phase change-checkpoint protocol.
271#[async_trait]
272pub trait ChangeCheckpointStore: Send + Sync {
273    /// Loads the active cursor and any pending replayable batch for one scope.
274    async fn load_change_checkpoint(
275        &self,
276        scope: &CloudScope,
277    ) -> StoreResult<ChangeCursorCheckpoint>;
278
279    /// Persists a batch against the current active cursor without advancing that cursor.
280    async fn record_change_batch(
281        &self,
282        batch: PersistedChangeBatch,
283    ) -> StoreResult<StoreWriteStatus>;
284
285    /// Marks all post-crash effects for the pending batch durably replayable.
286    async fn mark_change_effects_replayable(
287        &self,
288        scope: &CloudScope,
289        batch_id: &ChangeBatchId,
290    ) -> StoreResult<StoreWriteStatus>;
291
292    /// Atomically promotes a replayable batch cursor and clears its pending slot.
293    ///
294    /// Repeating a successful commit for the same batch identity must be idempotent.
295    async fn commit_change_cursor(
296        &self,
297        scope: &CloudScope,
298        batch_id: &ChangeBatchId,
299    ) -> StoreResult<StoreWriteStatus>;
300}
301
302/// Durable content metadata, runtime lease, and eviction-recovery protocol.
303///
304/// Implementations may compose a database-backed metadata repository with an in-process lease
305/// registry. The port requires their eviction reservation decision to be serialized so a lease
306/// cannot start between guard evaluation and intent persistence.
307#[async_trait]
308pub trait ContentStorageStore: Send + Sync {
309    /// Creates a revision-bound content entry without replacing a different existing snapshot.
310    async fn create_content_entry(
311        &self,
312        entry: ContentStorageEntry,
313    ) -> StoreResult<StoreWriteStatus>;
314
315    /// Loads one content storage snapshot including current runtime lease counts.
316    async fn load_content_entry(
317        &self,
318        key: &ContentCacheKey,
319    ) -> StoreResult<Option<ContentStorageEntry>>;
320
321    /// Records provider-owned cached range coverage.
322    async fn record_provider_cached_range(
323        &self,
324        key: &ContentCacheKey,
325        range: ByteRange,
326    ) -> StoreResult<StoreWriteStatus>;
327
328    /// Records platform-observed materialization without writing provider-cache coverage.
329    async fn observe_platform_materialization(
330        &self,
331        key: &ContentCacheKey,
332        state: PlatformMaterializationState,
333    ) -> StoreResult<StoreWriteStatus>;
334
335    /// Updates the pin guard.
336    async fn set_content_pinned(
337        &self,
338        key: &ContentCacheKey,
339        pinned: bool,
340    ) -> StoreResult<StoreWriteStatus>;
341
342    /// Records an immutable dirty snapshot, fencing an older local generation.
343    async fn mark_content_dirty(
344        &self,
345        key: &ContentCacheKey,
346        snapshot: LocalContentSnapshot,
347    ) -> StoreResult<StoreWriteStatus>;
348
349    /// Clears dirty state only if the completed upload still owns the active generation.
350    async fn clear_content_dirty(
351        &self,
352        key: &ContentCacheKey,
353        generation: LocalContentGeneration,
354    ) -> StoreResult<StoreWriteStatus>;
355
356    /// Acquires an idempotently identified runtime lease, unless eviction already reserved entry.
357    async fn acquire_content_lease(
358        &self,
359        key: &ContentCacheKey,
360        lease_id: ContentLeaseId,
361        kind: ContentLeaseKind,
362    ) -> StoreResult<StoreWriteStatus>;
363
364    /// Releases one runtime lease.
365    async fn release_content_lease(
366        &self,
367        key: &ContentCacheKey,
368        lease_id: &ContentLeaseId,
369    ) -> StoreResult<StoreWriteStatus>;
370
371    /// Atomically checks guards, reserves the entry, and persists an eviction intent.
372    async fn begin_content_eviction(
373        &self,
374        intent: ContentEvictionIntent,
375    ) -> StoreResult<ContentEvictionBegin>;
376
377    /// Loads one durable eviction record.
378    async fn load_content_eviction(
379        &self,
380        operation_id: &ContentEvictionOperationId,
381    ) -> StoreResult<Option<ContentEvictionRecord>>;
382
383    /// Loads one bounded eviction-recovery page.
384    ///
385    /// Implementations must apply `limit` at the storage boundary with stable operation-id order.
386    async fn recoverable_content_evictions_page(
387        &self,
388        scope: &CloudScope,
389        after_operation_id: Option<&str>,
390        limit: usize,
391    ) -> StoreResult<RecoveryPage<ContentEvictionRecord>>;
392
393    /// Records one idempotent physical effect observed by cache/platform reconciliation.
394    async fn record_content_eviction_effect(
395        &self,
396        operation_id: &ContentEvictionOperationId,
397        effect: ContentEvictionPhysicalEffect,
398    ) -> StoreResult<StoreWriteStatus>;
399
400    /// Atomically applies the physical result to content metadata.
401    async fn reconcile_content_eviction_metadata(
402        &self,
403        operation_id: &ContentEvictionOperationId,
404    ) -> StoreResult<StoreWriteStatus>;
405
406    /// Marks the eviction terminal and releases its entry reservation.
407    async fn complete_content_eviction(
408        &self,
409        operation_id: &ContentEvictionOperationId,
410    ) -> StoreResult<StoreWriteStatus>;
411}
412
413/// Durable provider-cache byte installation and sparse-coverage reconciliation protocol.
414///
415/// Implementations atomically reserve the target entry with intent persistence, and atomically
416/// update provider coverage with the durable record's coverage transition.
417#[async_trait]
418pub trait ContentCacheWriteStore: Send + Sync {
419    /// Atomically reserves the entry and persists a write intent before physical bytes are changed.
420    async fn begin_content_cache_write(
421        &self,
422        intent: ContentCacheWriteIntent,
423    ) -> StoreResult<StoreWriteStatus>;
424
425    /// Loads one durable cache-write record.
426    async fn load_content_cache_write(
427        &self,
428        operation_id: &ContentCacheWriteOperationId,
429    ) -> StoreResult<Option<ContentCacheWriteRecord>>;
430
431    /// Loads one bounded cache-write recovery page.
432    async fn recoverable_content_cache_writes_page(
433        &self,
434        scope: &CloudScope,
435        after_operation_id: Option<&str>,
436        limit: usize,
437    ) -> StoreResult<RecoveryPage<ContentCacheWriteRecord>>;
438
439    /// Records that the intended bytes are physically committed and observable.
440    async fn record_content_cache_write_physical_commit(
441        &self,
442        operation_id: &ContentCacheWriteOperationId,
443    ) -> StoreResult<StoreWriteStatus>;
444
445    /// Atomically records provider range coverage for the committed physical bytes.
446    async fn reconcile_content_cache_write_coverage(
447        &self,
448        operation_id: &ContentCacheWriteOperationId,
449    ) -> StoreResult<StoreWriteStatus>;
450
451    /// Marks the write terminal and releases its entry reservation.
452    async fn complete_content_cache_write(
453        &self,
454        operation_id: &ContentCacheWriteOperationId,
455    ) -> StoreResult<StoreWriteStatus>;
456}
457
458/// Durable resumable-upload checkpoint and dirty-generation reconciliation protocol.
459///
460/// Intent persistence atomically validates the active immutable dirty snapshot and acquires its
461/// operation-owned upload lease. Metadata reconciliation clears dirty state only when the upload's
462/// local generation is still current; a newer generation fences stale completion. Every mutating
463/// transition also receives the executor's platform session generation. Implementations must
464/// compare it with the active generation in the same transaction as the requested transition.
465#[async_trait]
466pub trait ContentUploadStore: Send + Sync {
467    /// Atomically validates dirty state and generation, acquires the lease, and persists intent.
468    async fn persist_content_upload_intent(
469        &self,
470        intent: ContentUploadIntent,
471        execution_generation: SessionGeneration,
472    ) -> StoreResult<StoreWriteStatus>;
473
474    /// Loads one durable upload record.
475    async fn load_content_upload(
476        &self,
477        operation_id: &OperationId,
478    ) -> StoreResult<Option<ContentUploadRecord>>;
479
480    /// Loads one bounded upload-recovery page.
481    async fn recoverable_content_uploads_page(
482        &self,
483        scope: &CloudScope,
484        after_operation_id: Option<&str>,
485        limit: usize,
486    ) -> StoreResult<RecoveryPage<ContentUploadRecord>>;
487
488    /// Records a backend session or monotonically advances its accepted offset when unfenced.
489    async fn record_content_upload_session(
490        &self,
491        operation_id: &OperationId,
492        session: ContentUploadSession,
493        execution_generation: SessionGeneration,
494    ) -> StoreResult<StoreWriteStatus>;
495
496    /// Marks remote commit started after all immutable bytes were accepted when unfenced.
497    async fn begin_content_upload_remote_commit(
498        &self,
499        operation_id: &OperationId,
500        execution_generation: SessionGeneration,
501    ) -> StoreResult<StoreWriteStatus>;
502
503    /// Records a known or unknown remote upload outcome when the executor remains active.
504    async fn record_content_upload_remote_outcome(
505        &self,
506        operation_id: &OperationId,
507        outcome: MutationRemoteOutcome,
508        execution_generation: SessionGeneration,
509    ) -> StoreResult<StoreWriteStatus>;
510
511    /// Reconciles the outcome and conditionally clears dirty state when the executor is active.
512    async fn reconcile_content_upload_metadata(
513        &self,
514        operation_id: &OperationId,
515        execution_generation: SessionGeneration,
516    ) -> StoreResult<StoreWriteStatus>;
517
518    /// Marks the upload terminal and releases its lease when the executor remains active.
519    async fn complete_content_upload(
520        &self,
521        operation_id: &OperationId,
522        execution_generation: SessionGeneration,
523    ) -> StoreResult<StoreWriteStatus>;
524}
525
526/// Durable mutation journal and active-session fence protocol.
527#[async_trait]
528pub trait MutationJournalStore: Send + Sync {
529    /// Activates a generation in `Accepting`, fencing all lower generations for the same scope.
530    async fn activate_session(
531        &self,
532        scope: &CloudScope,
533        generation: SessionGeneration,
534    ) -> StoreResult<StoreWriteStatus>;
535
536    /// Loads the active generation and lifecycle state for one scope.
537    async fn load_session(
538        &self,
539        scope: &CloudScope,
540    ) -> StoreResult<Option<(SessionGeneration, SessionState)>>;
541
542    /// Advances one active generation through closing, draining, and closed.
543    async fn transition_session(
544        &self,
545        scope: &CloudScope,
546        generation: SessionGeneration,
547        next: SessionState,
548    ) -> StoreResult<StoreWriteStatus>;
549
550    /// Inserts the first recoverable mutation state before ingress acknowledgement.
551    async fn persist_mutation_intent(
552        &self,
553        intent: MutationIntent,
554    ) -> StoreResult<StoreWriteStatus>;
555
556    /// Loads one durable mutation record.
557    async fn load_mutation(
558        &self,
559        operation_id: &OperationId,
560    ) -> StoreResult<Option<MutationRecord>>;
561
562    /// Loads one bounded mutation-recovery page.
563    async fn recoverable_mutations_page(
564        &self,
565        scope: &CloudScope,
566        after_operation_id: Option<&str>,
567        limit: usize,
568    ) -> StoreResult<RecoveryPage<MutationRecord>>;
569
570    /// Marks remote application started under the active generation.
571    ///
572    /// A newer generation may resume an older persisted intent; a completion still carrying an
573    /// inactive generation is fenced.
574    async fn begin_remote_apply(
575        &self,
576        operation_id: &OperationId,
577        generation: SessionGeneration,
578    ) -> StoreResult<StoreWriteStatus>;
579
580    /// Records a remote outcome or reconciles a previous unknown outcome under the active session.
581    async fn record_remote_outcome(
582        &self,
583        operation_id: &OperationId,
584        generation: SessionGeneration,
585        outcome: MutationRemoteOutcome,
586    ) -> StoreResult<StoreWriteStatus>;
587
588    /// Durably reconciles required product/local platform state for a still-active generation.
589    ///
590    /// Implementations must generation-check and complete, or durably enqueue, the required local
591    /// metadata, namespace mapping, and platform reconciliation in the same transaction as this
592    /// marker. This method must not write a marker that falsely claims reconciliation while the
593    /// corresponding product/platform effect remains neither durable nor replayable.
594    async fn mark_platform_reconciled(
595        &self,
596        operation_id: &OperationId,
597        generation: SessionGeneration,
598    ) -> StoreResult<StoreWriteStatus>;
599
600    /// Marks a reconciled mutation terminal without repeating its remote effect.
601    async fn complete_mutation(
602        &self,
603        operation_id: &OperationId,
604        generation: SessionGeneration,
605    ) -> StoreResult<StoreWriteStatus>;
606}
aster_forge_cloud_files_core/store.rs

aster_forge_cloud_files_core/
store.rs
