acyclic_fs/
async_storage.rs

1//! Runtime-neutral asynchronous storage contracts for browser and remote use.
2
3use crate::cancellation::CancellationToken;
4use crate::foundation::{
5    AuthorityId, Epoch, GenerationId, Head, OperationId, ProposedCommit, Sequence,
6};
7use crate::performance::WorkBudget;
8use crate::storage::{
9    AppendOutcome, AuthorityResult, AuthorityStore, CreateAuthorityOutcome, FenceOutcome,
10    GuardedAppend, ObjectId, ObjectRead, ObjectReadRequest, ObjectResult, ObjectStore, ObjectWrite,
11    PublicationPermit, PublicationReservation, ReplayLimit, ReservationOutcome,
12};
13use bytes::Bytes;
14use std::any::{Any, TypeId};
15use std::future::Future;
16use std::mem::size_of;
17use std::sync::Arc;
18use std::task::{Context, Poll, Waker};
19
20/// Runtime-appropriate bound for storage futures.
21///
22/// Native service futures must move between executor workers. Browser futures
23/// may retain thread-affine JavaScript transaction handles.
24#[cfg(not(target_arch = "wasm32"))]
25pub trait StorageFuture: Send {}
26#[cfg(not(target_arch = "wasm32"))]
27impl<T: Send> StorageFuture for T {}
28
29/// Runtime-appropriate bound for storage futures.
30#[cfg(target_arch = "wasm32")]
31pub trait StorageFuture {}
32#[cfg(target_arch = "wasm32")]
33impl<T> StorageFuture for T {}
34
35/// Heap allocated future with the platform's storage executor bound.
36#[cfg(not(target_arch = "wasm32"))]
37pub(crate) type BoxStorageFuture<'a, T> = std::pin::Pin<Box<dyn Future<Output = T> + Send + 'a>>;
38/// Heap allocated future with the platform's storage executor bound.
39#[cfg(target_arch = "wasm32")]
40pub(crate) type BoxStorageFuture<'a, T> = std::pin::Pin<Box<dyn Future<Output = T> + 'a>>;
41
42/// Runtime-appropriate provider ownership bound.
43#[cfg(not(target_arch = "wasm32"))]
44pub trait StorageProvider: Send + Sync {}
45#[cfg(not(target_arch = "wasm32"))]
46impl<T: Send + Sync + ?Sized> StorageProvider for T {}
47
48/// Runtime-appropriate provider ownership bound.
49#[cfg(target_arch = "wasm32")]
50pub trait StorageProvider {}
51#[cfg(target_arch = "wasm32")]
52impl<T: ?Sized> StorageProvider for T {}
53
54/// Exact identity of one authenticated decoded-object representation.
55#[derive(Clone, Copy, Debug, Eq, Ord, PartialEq, PartialOrd)]
56#[doc(hidden)]
57pub struct DecodedCacheKey {
58    pub(crate) object_id: ObjectId,
59    decoder: TypeId,
60    limits: crate::kernel::DecodeLimits,
61}
62
63impl DecodedCacheKey {
64    pub(crate) fn new<T: 'static>(
65        object_id: ObjectId,
66        limits: crate::kernel::DecodeLimits,
67    ) -> Self {
68        Self {
69            object_id,
70            decoder: TypeId::of::<T>(),
71            limits,
72        }
73    }
74}
75
76/// Type-erased immutable decoded value retained by a disposable accelerator.
77#[derive(Clone)]
78#[doc(hidden)]
79pub struct DecodedCacheValue {
80    pub(crate) value: Arc<dyn Any + Send + Sync>,
81    pub(crate) logical_bytes: u64,
82}
83
84/// Whether a decoded value became shared cache state or remains operation-owned.
85#[doc(hidden)]
86pub enum DecodedCacheAdmission {
87    Uncached(DecodedCacheValue),
88    Shared(DecodedCacheValue),
89}
90
91/// Polls a future exactly once for adapters whose contract guarantees that no
92/// asynchronous suspension is possible.
93///
94/// This keeps synchronous adapters free of executors while
95/// failing closed if an implementation unexpectedly blocks.
96pub(crate) fn poll_ready<F: Future>(future: F) -> Option<F::Output> {
97    let mut future = std::pin::pin!(future);
98    let waker = Waker::noop();
99    let mut context = Context::from_waker(waker);
100    match future.as_mut().poll(&mut context) {
101        Poll::Ready(value) => Some(value),
102        Poll::Pending => None,
103    }
104}
105
106/// Drives an [`ImmediateObjectStore`] adapter exactly once.
107///
108/// Every such adapter is supplied by this module's blanket implementation and
109/// contains no suspension point, so pending is an internal contract violation
110/// rather than a recoverable filesystem outcome.
111#[allow(
112    clippy::expect_used,
113    reason = "only crate-owned ImmediateObjectStore adapter futures reach this boundary; they contain no suspension point"
114)]
115pub(crate) fn poll_immediate<F: Future>(future: F) -> F::Output {
116    poll_ready(future).expect("immediate storage adapter suspended")
117}
118
119/// Authority lineage semantics for one workspace fork.
120#[derive(Clone, Copy, Debug, Eq, PartialEq)]
121pub enum GenerationFork {
122    /// Reuse the source authority's durable lineage through the selected generation.
123    PublishedPrefix,
124    /// Start independent authority history from an authenticated unpublished generation.
125    Independent,
126}
127
128/// Exact source and lineage semantics for one authority fork.
129#[derive(Clone, Copy, Debug, Eq, PartialEq)]
130pub struct GenerationForkSource {
131    /// Source authority containing the selected generation.
132    pub authority: AuthorityId,
133    /// Exact authenticated source generation.
134    pub generation: GenerationId,
135    /// Whether durable source lineage is reused.
136    pub lineage: GenerationFork,
137}
138
139/// Nonblocking authority-store contract. Futures are sendable on native
140/// targets and may remain JavaScript-thread-affine in browsers.
141pub trait AsyncAuthorityStore: StorageProvider {
142    /// Whether this backend can retain an immutable published-generation
143    /// lineage prefix when it creates a child authority.
144    fn supports_generation_lineage_prefix(&self) -> bool {
145        false
146    }
147
148    /// Creates a destination authority whose canonical generation lineage is
149    /// either an immutable native prefix of the source or an explicit
150    /// independent lineage rooted at an unpublished authenticated generation.
151    fn fork_generation_authority(
152        &self,
153        source: GenerationForkSource,
154        destination_authority: AuthorityId,
155        operation_id: OperationId,
156        budget: WorkBudget,
157        cancellation: &CancellationToken,
158    ) -> impl Future<Output = AuthorityResult<CreateAuthorityOutcome>> + StorageFuture;
159
160    /// Asynchronously creates one authority.
161    fn create_authority(
162        &self,
163        authority_id: AuthorityId,
164        genesis_epoch: Epoch,
165        budget: WorkBudget,
166        cancellation: &CancellationToken,
167    ) -> impl Future<Output = AuthorityResult<CreateAuthorityOutcome>> + StorageFuture;
168
169    /// Asynchronously reads the linearizable head.
170    fn head(
171        &self,
172        authority_id: AuthorityId,
173        budget: WorkBudget,
174        cancellation: &CancellationToken,
175    ) -> impl Future<Output = AuthorityResult<Head>> + StorageFuture;
176
177    /// Asynchronously compares and appends one durable operation.
178    fn compare_and_append(
179        &self,
180        authority_id: AuthorityId,
181        epoch: Epoch,
182        expected: Head,
183        commit: ProposedCommit,
184        budget: WorkBudget,
185        cancellation: &CancellationToken,
186    ) -> impl Future<Output = AuthorityResult<AppendOutcome>> + StorageFuture;
187
188    /// Asynchronously compares and appends under an optional operation-window
189    /// permit. Backends that cannot atomically evaluate lease gates reject
190    /// managed permits rather than using a read-then-write approximation.
191    fn compare_and_append_guarded(
192        &self,
193        request: GuardedAppend,
194        budget: WorkBudget,
195        cancellation: &CancellationToken,
196    ) -> impl Future<Output = AuthorityResult<AppendOutcome>> + StorageFuture {
197        async move {
198            if request.permit != PublicationPermit::Unrestricted {
199                return Err(crate::storage::AuthorityFailure::before_work(
200                    crate::storage::AuthorityStoreError::Rejected(
201                        "authority backend cannot atomically evaluate operation leases".to_owned(),
202                    ),
203                ));
204            }
205            self.compare_and_append(
206                request.authority_id,
207                request.epoch,
208                request.expected,
209                request.commit,
210                budget,
211                cancellation,
212            )
213            .await
214        }
215    }
216
217    /// Atomically acquires the exclusive publication gate at an exact head.
218    fn reserve_publication(
219        &self,
220        _authority_id: AuthorityId,
221        _expected: Head,
222        _operation_id: OperationId,
223        _budget: WorkBudget,
224        _cancellation: &CancellationToken,
225    ) -> impl Future<Output = AuthorityResult<ReservationOutcome>> + StorageFuture {
226        async move {
227            Err(crate::storage::AuthorityFailure::before_work(
228                crate::storage::AuthorityStoreError::Rejected(
229                    "authority backend does not support durable publication reservations"
230                        .to_owned(),
231                ),
232            ))
233        }
234    }
235
236    /// Releases an exact publication reservation. Exact retries are idempotent.
237    fn release_publication(
238        &self,
239        _reservation: PublicationReservation,
240        _budget: WorkBudget,
241        _cancellation: &CancellationToken,
242    ) -> impl Future<Output = AuthorityResult<()>> + StorageFuture {
243        async move {
244            Err(crate::storage::AuthorityFailure::before_work(
245                crate::storage::AuthorityStoreError::Rejected(
246                    "authority backend does not support durable publication reservations"
247                        .to_owned(),
248                ),
249            ))
250        }
251    }
252
253    /// Asynchronously replays one bounded contiguous page.
254    fn replay(
255        &self,
256        authority_id: AuthorityId,
257        after: Sequence,
258        limit: ReplayLimit,
259        budget: WorkBudget,
260        cancellation: &CancellationToken,
261    ) -> impl Future<Output = AuthorityResult<Vec<crate::foundation::DurableCommit>>> + StorageFuture;
262
263    /// Asynchronously advances a writer fence.
264    fn fence(
265        &self,
266        authority_id: AuthorityId,
267        expected: Head,
268        budget: WorkBudget,
269        cancellation: &CancellationToken,
270    ) -> impl Future<Output = AuthorityResult<FenceOutcome>> + StorageFuture;
271
272    /// Asynchronously resolves one idempotent operation identity.
273    fn find_operation(
274        &self,
275        authority_id: AuthorityId,
276        operation_id: OperationId,
277        budget: WorkBudget,
278        cancellation: &CancellationToken,
279    ) -> impl Future<Output = AuthorityResult<Option<crate::foundation::DurableCommit>>> + StorageFuture;
280}
281
282/// Explicit opt-in for synchronous stores that are safe to complete inline.
283///
284/// Implementing this marker makes the synchronous authority methods satisfy
285/// [`AsyncAuthorityStore`] without yielding. Blocking native backends must use
286/// a provider-owned bounded blocking executor instead.
287pub trait ImmediateAuthorityStore: AuthorityStore {}
288
289impl<T: ImmediateAuthorityStore + ?Sized> ImmediateAuthorityStore for Arc<T> {}
290
291/// Nonblocking immutable-object contract suitable for `IndexedDB` and remote I/O.
292pub trait AsyncObjectStore: StorageProvider {
293    /// Returns one exact decoded immutable representation when resident.
294    ///
295    /// # Errors
296    ///
297    /// Returns a storage error when disposable accelerator state is corrupt.
298    fn decoded_cache_get(
299        &self,
300        _key: DecodedCacheKey,
301    ) -> Result<Option<DecodedCacheValue>, crate::storage::ObjectStoreError> {
302        Ok(None)
303    }
304
305    /// Offers one authenticated decoded representation to a disposable cache.
306    ///
307    /// # Errors
308    ///
309    /// Returns a storage error when the accelerator cannot safely admit the
310    /// supplied immutable representation.
311    fn decoded_cache_admit(
312        &self,
313        _key: DecodedCacheKey,
314        value: DecodedCacheValue,
315    ) -> Result<DecodedCacheAdmission, crate::storage::ObjectStoreError> {
316        Ok(DecodedCacheAdmission::Uncached(value))
317    }
318
319    /// Asynchronously admits one verified immutable object.
320    fn put(
321        &self,
322        object_id: ObjectId,
323        bytes: Bytes,
324        budget: WorkBudget,
325        cancellation: &CancellationToken,
326    ) -> impl Future<Output = ObjectResult<()>> + StorageFuture;
327
328    /// Asynchronously admits an ordered bounded group of verified immutable objects.
329    ///
330    /// Implementations with a real batch primitive override this method. A
331    /// backend without one rejects the operation instead of hiding repeated
332    /// single-object operations behind the batch contract.
333    fn put_many(
334        &self,
335        writes: &[ObjectWrite],
336        budget: WorkBudget,
337        cancellation: &CancellationToken,
338    ) -> impl Future<Output = ObjectResult<()>> + StorageFuture {
339        let _ = (writes, budget, cancellation);
340        async {
341            Err(crate::storage::ObjectFailure::before_work(
342                crate::storage::ObjectStoreError::Rejected(
343                    "object backend has no batch-write primitive".to_owned(),
344                ),
345            ))
346        }
347    }
348
349    /// Makes every previously admitted immutable object crash-durable before
350    /// an authority record may reference it. Ordinary stores already provide
351    /// that guarantee from `put`/`put_many`; a bounded staging adapter overrides
352    /// this boundary to group physical writes without changing publication.
353    fn flush_before_publish(
354        &self,
355        _budget: WorkBudget,
356        _cancellation: &CancellationToken,
357    ) -> impl Future<Output = ObjectResult<()>> + StorageFuture {
358        async {
359            Ok(crate::storage::ObjectReceipt {
360                value: (),
361                work: crate::WorkCounters::default(),
362            })
363        }
364    }
365
366    /// Asynchronously reads one complete bounded object.
367    fn read(
368        &self,
369        object_id: ObjectId,
370        maximum_bytes: u64,
371        budget: WorkBudget,
372        cancellation: &CancellationToken,
373    ) -> impl Future<Output = ObjectResult<ObjectRead>> + StorageFuture;
374
375    /// Asynchronously reads an ordered batch of complete bounded objects.
376    ///
377    /// Backends with transactional multi-get or batched I/O should override
378    /// this method. The default is cancellation-aware and retains exact
379    /// per-object work rather than pretending sequential execution was one
380    /// physical operation.
381    fn read_many(
382        &self,
383        requests: &[ObjectReadRequest],
384        budget: WorkBudget,
385        cancellation: &CancellationToken,
386    ) -> impl Future<Output = ObjectResult<Vec<ObjectRead>>> + StorageFuture;
387
388    /// Asynchronously probes one exact object identity.
389    fn contains(
390        &self,
391        object_id: ObjectId,
392        budget: WorkBudget,
393        cancellation: &CancellationToken,
394    ) -> impl Future<Output = ObjectResult<bool>> + StorageFuture;
395}
396
397/// Explicit opt-in for synchronous stores that are safe to complete inline.
398///
399/// This marker also admits the synchronous kernel convenience functions.
400/// Blocking native backends must own bounded async dispatch internally.
401pub trait ImmediateObjectStore: ObjectStore {}
402
403impl<T: ImmediateObjectStore + ?Sized> ImmediateObjectStore for Arc<T> {}
404
405/// Explicit sequential implementation for adapters whose physical backend has
406/// no multi-get primitive. Backends must opt into this implementation rather
407/// than silently inheriting it.
408///
409/// # Errors
410///
411/// Returns an empty-batch rejection, cancellation, budget failure, allocation
412/// failure, or the first exact object-read failure with all spent work.
413pub async fn read_many_sequential_async<S: AsyncObjectStore + ?Sized>(
414    store: &S,
415    requests: &[ObjectReadRequest],
416    budget: WorkBudget,
417    cancellation: &CancellationToken,
418) -> ObjectResult<Vec<ObjectRead>> {
419    cancellation.check().map_err(|_| {
420        crate::storage::ObjectFailure::before_work(crate::storage::ObjectStoreError::Cancelled)
421    })?;
422    if requests.is_empty() {
423        return Err(crate::storage::ObjectFailure::before_work(
424            crate::storage::ObjectStoreError::Rejected("object read batch is empty".to_owned()),
425        ));
426    }
427    let item_count = u64::try_from(requests.len()).unwrap_or(u64::MAX);
428    let vector_bytes = u64::try_from(requests.len())
429        .unwrap_or(u64::MAX)
430        .saturating_mul(u64::try_from(size_of::<ObjectRead>()).unwrap_or(u64::MAX));
431    let mut work = crate::performance::WorkCounters {
432        items_examined: item_count,
433        allocation_operations: u64::from(!requests.is_empty()),
434        peak_allocation_bytes: vector_bytes,
435        ..crate::performance::WorkCounters::default()
436    };
437    let mut admission = work;
438    admission.items_returned = item_count;
439    admission
440        .verify(budget)
441        .map_err(|error| crate::storage::ObjectFailure::before_work(error.into()))?;
442    let mut values = Vec::new();
443    values.try_reserve_exact(requests.len()).map_err(|_| {
444        crate::storage::ObjectFailure::before_work(crate::storage::ObjectStoreError::Rejected(
445            "object batch result allocation failed".to_owned(),
446        ))
447    })?;
448    let mut retained_bytes = vector_bytes;
449    for request in requests {
450        cancellation.check().map_err(|_| {
451            crate::storage::ObjectFailure::new(crate::storage::ObjectStoreError::Cancelled, work)
452        })?;
453        let remaining = work
454            .remaining(budget)
455            .map_err(|error| crate::storage::ObjectFailure::new(error.into(), work))?;
456        let receipt = store
457            .read(
458                request.object_id,
459                request.maximum_bytes,
460                remaining,
461                cancellation,
462            )
463            .await
464            .map_err(|failure| {
465                let nested = *failure.work;
466                let peak = work
467                    .peak_allocation_bytes
468                    .max(retained_bytes.saturating_add(nested.peak_allocation_bytes));
469                match work.checked_add(nested) {
470                    Ok(mut combined) => {
471                        combined.peak_allocation_bytes = peak;
472                        crate::storage::ObjectFailure::new(failure.error, combined)
473                    }
474                    Err(error) => crate::storage::ObjectFailure::new(error.into(), work),
475                }
476            })?;
477        let nested_peak = retained_bytes.saturating_add(receipt.work.peak_allocation_bytes);
478        work = work
479            .checked_add(receipt.work)
480            .map_err(|error| crate::storage::ObjectFailure::new(error.into(), work))?;
481        work.peak_allocation_bytes = work.peak_allocation_bytes.max(nested_peak);
482        retained_bytes = retained_bytes.saturating_add(match receipt.value.retention {
483            crate::storage::ObjectReadRetention::Shared => 0,
484            crate::storage::ObjectReadRetention::Owned { logical_bytes } => logical_bytes,
485        });
486        work.peak_allocation_bytes = work.peak_allocation_bytes.max(retained_bytes);
487        work.verify(budget)
488            .map_err(|error| crate::storage::ObjectFailure::new(error.into(), work))?;
489        values.push(receipt.value);
490    }
491    work.items_returned = item_count;
492    Ok(crate::storage::ObjectReceipt {
493        value: values,
494        work,
495    })
496}
497
498impl<T: ImmediateAuthorityStore + ?Sized> AsyncAuthorityStore for T {
499    async fn fork_generation_authority(
500        &self,
501        source: GenerationForkSource,
502        destination_authority: AuthorityId,
503        _operation_id: OperationId,
504        budget: WorkBudget,
505        cancellation: &CancellationToken,
506    ) -> AuthorityResult<CreateAuthorityOutcome> {
507        cancellation.check().map_err(|_| {
508            crate::storage::AuthorityFailure::before_work(
509                crate::storage::AuthorityStoreError::Cancelled,
510            )
511        })?;
512        if source.lineage == GenerationFork::PublishedPrefix {
513            return Err(crate::storage::AuthorityFailure::before_work(
514                crate::storage::AuthorityStoreError::Rejected(
515                    "immediate authority stores do not retain generation lineage prefixes"
516                        .to_owned(),
517                ),
518            ));
519        }
520        AuthorityStore::create_authority(self, destination_authority, Epoch::GENESIS, budget)
521    }
522
523    async fn create_authority(
524        &self,
525        authority_id: AuthorityId,
526        genesis_epoch: Epoch,
527        budget: WorkBudget,
528        cancellation: &CancellationToken,
529    ) -> AuthorityResult<CreateAuthorityOutcome> {
530        cancellation.check().map_err(|_| {
531            crate::storage::AuthorityFailure::before_work(
532                crate::storage::AuthorityStoreError::Cancelled,
533            )
534        })?;
535        AuthorityStore::create_authority(self, authority_id, genesis_epoch, budget)
536    }
537
538    async fn head(
539        &self,
540        authority_id: AuthorityId,
541        budget: WorkBudget,
542        cancellation: &CancellationToken,
543    ) -> AuthorityResult<Head> {
544        cancellation.check().map_err(|_| {
545            crate::storage::AuthorityFailure::before_work(
546                crate::storage::AuthorityStoreError::Cancelled,
547            )
548        })?;
549        AuthorityStore::head(self, authority_id, budget)
550    }
551
552    async fn compare_and_append(
553        &self,
554        authority_id: AuthorityId,
555        epoch: Epoch,
556        expected: Head,
557        commit: ProposedCommit,
558        budget: WorkBudget,
559        cancellation: &CancellationToken,
560    ) -> AuthorityResult<AppendOutcome> {
561        cancellation.check().map_err(|_| {
562            crate::storage::AuthorityFailure::before_work(
563                crate::storage::AuthorityStoreError::Cancelled,
564            )
565        })?;
566        AuthorityStore::compare_and_append(self, authority_id, epoch, expected, commit, budget)
567    }
568
569    async fn replay(
570        &self,
571        authority_id: AuthorityId,
572        after: Sequence,
573        limit: ReplayLimit,
574        budget: WorkBudget,
575        cancellation: &CancellationToken,
576    ) -> AuthorityResult<Vec<crate::foundation::DurableCommit>> {
577        cancellation.check().map_err(|_| {
578            crate::storage::AuthorityFailure::before_work(
579                crate::storage::AuthorityStoreError::Cancelled,
580            )
581        })?;
582        AuthorityStore::replay(self, authority_id, after, limit, budget)
583    }
584
585    async fn fence(
586        &self,
587        authority_id: AuthorityId,
588        expected: Head,
589        budget: WorkBudget,
590        cancellation: &CancellationToken,
591    ) -> AuthorityResult<FenceOutcome> {
592        cancellation.check().map_err(|_| {
593            crate::storage::AuthorityFailure::before_work(
594                crate::storage::AuthorityStoreError::Cancelled,
595            )
596        })?;
597        AuthorityStore::fence(self, authority_id, expected, budget)
598    }
599
600    async fn find_operation(
601        &self,
602        authority_id: AuthorityId,
603        operation_id: OperationId,
604        budget: WorkBudget,
605        cancellation: &CancellationToken,
606    ) -> AuthorityResult<Option<crate::foundation::DurableCommit>> {
607        cancellation.check().map_err(|_| {
608            crate::storage::AuthorityFailure::before_work(
609                crate::storage::AuthorityStoreError::Cancelled,
610            )
611        })?;
612        AuthorityStore::find_operation(self, authority_id, operation_id, budget)
613    }
614}
615
616impl<T: ImmediateObjectStore + ?Sized> AsyncObjectStore for T {
617    async fn put(
618        &self,
619        object_id: ObjectId,
620        bytes: Bytes,
621        budget: WorkBudget,
622        cancellation: &CancellationToken,
623    ) -> ObjectResult<()> {
624        cancellation.check().map_err(|_| {
625            crate::storage::ObjectFailure::before_work(crate::storage::ObjectStoreError::Cancelled)
626        })?;
627        ObjectStore::put(self, object_id, bytes, budget)
628    }
629
630    async fn put_many(
631        &self,
632        writes: &[ObjectWrite],
633        budget: WorkBudget,
634        cancellation: &CancellationToken,
635    ) -> ObjectResult<()> {
636        cancellation.check().map_err(|_| {
637            crate::storage::ObjectFailure::before_work(crate::storage::ObjectStoreError::Cancelled)
638        })?;
639        ObjectStore::put_many(self, writes, budget)
640    }
641
642    async fn read(
643        &self,
644        object_id: ObjectId,
645        maximum_bytes: u64,
646        budget: WorkBudget,
647        cancellation: &CancellationToken,
648    ) -> ObjectResult<ObjectRead> {
649        cancellation.check().map_err(|_| {
650            crate::storage::ObjectFailure::before_work(crate::storage::ObjectStoreError::Cancelled)
651        })?;
652        ObjectStore::read(self, object_id, maximum_bytes, budget)
653    }
654
655    async fn read_many(
656        &self,
657        requests: &[ObjectReadRequest],
658        budget: WorkBudget,
659        cancellation: &CancellationToken,
660    ) -> ObjectResult<Vec<ObjectRead>> {
661        cancellation.check().map_err(|_| {
662            crate::storage::ObjectFailure::before_work(crate::storage::ObjectStoreError::Cancelled)
663        })?;
664        ObjectStore::read_many(self, requests, budget)
665    }
666
667    async fn contains(
668        &self,
669        object_id: ObjectId,
670        budget: WorkBudget,
671        cancellation: &CancellationToken,
672    ) -> ObjectResult<bool> {
673        cancellation.check().map_err(|_| {
674            crate::storage::ObjectFailure::before_work(crate::storage::ObjectStoreError::Cancelled)
675        })?;
676        ObjectStore::contains(self, object_id, budget)
677    }
678}
679
680#[cfg(all(test, feature = "memory"))]
681#[path = "tests/async_storage.rs"]
682mod tests;
acyclic_fs/async_storage.rs

acyclic_fs/
async_storage.rs
