coven_storage/
connection.rs

1//! One connected provider, shared by sync steps, operations and file transfers.
2
3use crate::*;
4use coven_foundation::id_source::DeviceId;
5use std::sync::Arc;
6
7/// Applies the storage contract around native provider calls (E5). Every
8/// consumer uses this capability and waits for the same provider cooldown.
9pub struct StorageConnection<P: ?Sized = dyn ProviderOps> {
10    provider: Arc<P>,
11    /// The device this connection writes for: what it stages sits in that
12    /// device's reserved folder (D10).
13    device: DeviceId,
14    publications: Arc<dyn PublicationGate>,
15    cooldown: Arc<dyn Cooldown>,
16    transfers: Arc<TransferBudget>,
17    /// The store's clock, which times this connection's requests and its
18    /// waits.
19    clock: coven_foundation::clock::ClockRef,
20    /// Every request this connection sent, timed by that clock.
21    requests: crate::request_tally::RequestTally,
22    /// For the connection a setup call makes for its own requests, what the
23    /// call has left of its minute: rate-limited requests are sent again in
24    /// place (§20.8).
25    setup: Option<SetupWaits>,
26}
27
28impl<P: ProviderOps + ?Sized> StorageConnection<P> {
29    /// Compose once when connecting, before distributing the connection to owners.
30    #[cfg(test)]
31    pub(crate) fn from_provider(provider: Arc<P>) -> Self {
32        Self::with_gate(
33            provider,
34            crate::test_utils::DEVICE,
35            Arc::new(crate::test_utils::UnrestrictedPublications),
36            Arc::new(NoCooldown),
37            Arc::new(TransferBudget::new(TransferLimits::default())),
38            Arc::new(coven_foundation::clock::FixedClock::new(std::time::UNIX_EPOCH)),
39        )
40    }
41
42    pub(crate) fn with_gate(
43        provider: Arc<P>,
44        device: DeviceId,
45        publications: Arc<dyn PublicationGate>,
46        cooldown: Arc<dyn Cooldown>,
47        transfers: Arc<TransferBudget>,
48        clock: coven_foundation::clock::ClockRef,
49    ) -> Self {
50        Self {
51            provider,
52            device,
53            publications,
54            cooldown,
55            transfers,
56            requests: crate::request_tally::RequestTally::new(clock.clone()),
57            clock,
58            setup: None,
59        }
60    }
61
62    /// A connection beside this one for the setup call spending `waits`,
63    /// and only for that call's own requests: one client to the provider
64    /// with this one, through the same provider, cooldown, publications and
65    /// transfer limits, but its reads, listings, deletions, creates and
66    /// replacements refused for rate limiting are sent again until the
67    /// call's minute is spent. Upload session steps and sharing requests are
68    /// not: a session's owner settles its steps through the recorded session
69    /// (§20.6). The call drops it, and its waits, when it ends.
70    pub fn during_setup(&self, waits: SetupWaits) -> Self {
71        Self {
72            provider: self.provider.clone(),
73            device: self.device,
74            publications: self.publications.clone(),
75            cooldown: self.cooldown.clone(),
76            transfers: self.transfers.clone(),
77            requests: crate::request_tally::RequestTally::new(self.clock.clone()),
78            clock: self.clock.clone(),
79            setup: Some(waits),
80        }
81    }
82
83    /// Every request this connection has sent so far, by kind.
84    pub fn requests(&self) -> crate::RequestCounts {
85        self.requests.counts()
86    }
87
88    /// Whether to send the request that failed with `error` again, having
89    /// waited: only on a setup call's own connection.
90    pub(crate) async fn resend_after(&self, error: &StorageError) -> bool {
91        match &self.setup {
92            Some(waits) => waits.wait_to_resend(self.clock.as_ref(), error).await,
93            None => false,
94        }
95    }
96
97    /// Send `request` once, or, during setup, again after each rate-limited
98    /// refusal the call can wait out. Only for a request whose refusal
99    /// stored nothing.
100    pub(crate) async fn resent<T, F: std::future::Future<Output = Result<T, StorageError>>>(
101        &self,
102        mut request: impl FnMut() -> F,
103    ) -> Result<T, StorageError> {
104        match &self.setup {
105            Some(waits) => waits.resend(self.clock.as_ref(), request).await,
106            None => request().await,
107        }
108    }
109
110    /// Count a read's bytes that arrived with its response.
111    pub(crate) fn received(&self, bytes: u64) {
112        self.requests.received(bytes);
113    }
114
115    /// Send one native request of `kind`, counting it.
116    pub(crate) async fn send<T>(
117        &self,
118        kind: crate::RequestKind,
119        bytes: u64,
120        request: impl std::future::Future<Output = T>,
121    ) -> T {
122        self.requests.send(kind, bytes, request).await
123    }
124
125    /// Refuse, without sending, while a provider cooldown is live. A refused
126    /// request has not reached storage.
127    pub(crate) fn cooling(&self) -> Result<(), StorageError> {
128        crate::cooldown::cooling(self.cooldown.as_ref())
129    }
130
131    fn publication(&self, path: &ObjectPath) -> Result<PublicationRequest, StorageError> {
132        let request = PublicationRequest::new(self.publications.clone(), path.clone());
133        request.check()?;
134        Ok(request)
135    }
136
137    /// The publication a session's next request goes under, once the session
138    /// is confirmed to name this location: none for a session already
139    /// complete, which has no request left to send.
140    fn unfinished_publication(
141        &self,
142        session: &UploadSession,
143    ) -> Result<Option<PublicationRequest>, StorageError> {
144        session.check(&self.config())?;
145        if session.is_complete() {
146            return Ok(None);
147        }
148        self.publication(session.path()).map(Some)
149    }
150
151    /// An account change: grants and revocations wait out a provider's
152    /// cooldown and are counted, except on S3, whose answers are console
153    /// instructions that make no request.
154    async fn account_change<T>(
155        &self,
156        change: impl std::future::Future<Output = Result<T, StorageError>>,
157    ) -> Result<T, StorageError> {
158        if self.config().provider() == CloudProvider::S3 {
159            return change.await;
160        }
161        self.cooling()?;
162        let result = self.send(RequestKind::Account, 0, change).await;
163        self.observed(result).await
164    }
165
166    /// Record a requested cooldown before any consumer sees the response
167    /// that asked for it (§6.7).
168    pub(crate) async fn observed<T>(
169        &self,
170        result: Result<T, StorageError>,
171    ) -> Result<T, StorageError> {
172        crate::cooldown::observed(self.cooldown.as_ref(), result).await
173    }
174    /// End each upload a listing found, counting each request.
175    async fn end_left(&self, left: &[LeftUpload]) -> Result<(), StorageError> {
176        for upload in left {
177            self.cooling()?;
178            self.observed(
179                self.send(upload.request(), 0, self.provider.end_left_upload(upload)).await,
180            )
181            .await?;
182        }
183        Ok(())
184    }
185    /// Abort the unfinished multipart uploads S3 detection run `run` left
186    /// in `store`'s `.coven-check` prefix, then delete its probes (§20.7),
187    /// as listing the run's prefix finds them. One already gone succeeds.
188    pub async fn end_detection_run(
189        &self,
190        store: coven_foundation::id_source::StoreId,
191        run: uuid::Uuid,
192    ) -> Result<(), StorageError> {
193        self.cooling()?;
194        let left = self
195            .observed(
196                self.send(RequestKind::List, 0, self.provider.detection_leftovers(store, run))
197                    .await,
198            )
199            .await?;
200        self.end_left(&left).await
201    }
202    async fn list_checked(&self, prefix: &ObjectPrefix) -> Result<Vec<StoredObject>, StorageError> {
203        let mut listing = ObjectListing::new(prefix.clone(), self.config().provider());
204        self.send(RequestKind::List, 0, self.provider.list(&mut listing)).await?;
205        Ok(listing.finish())
206    }
207
208    /// Replace `previous` with `bytes`. During setup a resend refused as a
209    /// conflict may have met its own earlier attempt: a copy publication can
210    /// store the object, then have its temporary's deletion refused (§20.6).
211    /// The object holding exactly these sealed bytes settles it as replaced.
212    pub(crate) async fn replace_object(
213        &self,
214        previous: &StoredObject,
215        bytes: &[u8],
216    ) -> Result<(), StorageError> {
217        let mut resent = false;
218        loop {
219            let replaced = self.replace_once(previous, bytes).await;
220            if let Err(error) = &replaced {
221                if resent && error.own_failure() == Some(StorageFailure::Conflict) {
222                    let stored = self.read(&previous.path).await?;
223                    return if stored == bytes { Ok(()) } else { replaced };
224                }
225                if self.resend_after(error).await {
226                    resent = true;
227                    continue;
228                }
229            }
230            return replaced;
231        }
232    }
233
234    async fn create_once(
235        &self,
236        path: &ObjectPath,
237        bytes: &[u8],
238        sent: &ByteProgress,
239    ) -> Result<Option<StoredObject>, StorageError> {
240        self.cooling()?;
241        let publication = self.publication(path)?;
242        let created = crate::transfer::upload_bytes(self, path, bytes, sent, async {
243            let object = self
244                .send(
245                    RequestKind::Create,
246                    bytes.len() as u64,
247                    self.provider.create(&publication, path, bytes, sent, self.device),
248                )
249                .await?;
250            if let Some(object) = &object {
251                object.check(path)?;
252            }
253            Ok(object)
254        })
255        .await;
256        self.observed(created).await
257    }
258
259    async fn stream_once(
260        &self,
261        path: &ObjectPath,
262        range: Option<ByteRange>,
263        received: &ByteProgress,
264    ) -> Result<ObjectStream<'_>, StorageError> {
265        self.cooling()?;
266        // The download limit counts file downloads. The sequence's own reads,
267        // its slots, snapshots and floor, are bounded by their one reader
268        // instead (§21.1), so they never wait behind a pin's files.
269        let download = match path.object_id() {
270            Some(_) => Some(self.transfers.download().await),
271            None => None,
272        };
273        self.cooling()?;
274        // A failed response records its cooldown before another download
275        // can acquire the capacity it releases.
276        let stream = self
277            .observed(
278                self.send(RequestKind::Read, 0, self.provider.stream_object(path, range, received))
279                    .await,
280            )
281            .await?;
282        stream.check(path, range)?;
283        let start = range.map_or(0, ByteRange::start);
284        let stream = match download {
285            Some(lease) => stream.with_download(lease),
286            None => stream,
287        };
288        Ok(stream.heard_by(received.starting_at(start)).counted(&self.requests))
289    }
290
291    async fn replace_once(
292        &self,
293        previous: &StoredObject,
294        bytes: &[u8],
295    ) -> Result<(), StorageError> {
296        self.cooling()?;
297        previous.check(&previous.path)?;
298        crate::transfer::check_single_request(bytes.len() as u64, self.single_request_limit())?;
299        let publication = self.publication(&previous.path)?;
300        let result = self
301            .send(
302                RequestKind::Replace,
303                bytes.len() as u64,
304                self.provider.replace(&publication, previous, bytes, self.device),
305            )
306            .await;
307        self.observed(result.map_err(StorageError::replacement)).await
308    }
309
310    pub(crate) async fn begin_upload_checked(
311        &self,
312        target: &UploadTarget,
313        total: u64,
314    ) -> Result<UploadSession, StorageError> {
315        target.check()?;
316        if total == 0 {
317            return Err(StorageFailure::InvalidPart.into());
318        }
319        let publication = self.publication(target.path())?;
320        self.send(
321            RequestKind::Upload,
322            0,
323            self.provider.begin_upload(&publication, target, total, self.device),
324        )
325        .await
326        .map_err(|error| target.error(error))
327    }
328    /// Admit `bytes` as `session`'s next part, sending nothing: the session
329    /// names this location, the part fits it, and its destination can be
330    /// published now. A planned session's first part marks its plan.
331    pub(crate) fn part_permit<'a>(
332        &self,
333        session: &mut UploadSession,
334        bytes: &'a [u8],
335    ) -> Result<PartPermit<'a>, StorageError> {
336        let config = self.config();
337        // Preserve native refusal order for a wrong location and an invalid part.
338        let part = match config.provider() {
339            CloudProvider::S3 | CloudProvider::CloudKit => {
340                let part = session.next_part(bytes.len())?;
341                session.check(&config)?;
342                part
343            }
344            CloudProvider::Dropbox | CloudProvider::OneDrive => {
345                session.check(&config)?;
346                session.next_part(bytes.len())?
347            }
348        };
349        let publication = self.publication(session.path())?;
350        Ok(PartPermit::admit(session, bytes, part, publication))
351    }
352    pub(crate) async fn send_permitted(
353        &self,
354        permit: PartPermit<'_>,
355        session: &mut UploadSession,
356        sent: &ByteProgress,
357    ) -> Result<(), StorageError> {
358        let PartPermit { bytes, part, publication, .. } = permit;
359        // The part's bytes are heard from where it starts in the object.
360        let sent = sent.starting_at(part.start);
361        let result = self
362            .send(
363                RequestKind::Upload,
364                bytes.len() as u64,
365                self.provider.upload_part(&publication, session, bytes, part.end, &sent),
366            )
367            .await;
368        result.map_err(|error| session.target.error(error))
369    }
370    pub(crate) async fn finish_upload_checked(
371        &self,
372        session: &mut UploadSession,
373    ) -> Result<(), StorageError> {
374        let Some(publication) = self.unfinished_publication(session)? else {
375            return Ok(());
376        };
377        let result = self
378            .send(RequestKind::Upload, 0, self.provider.finish_upload(&publication, session))
379            .await;
380        result.map_err(|error| session.target.error(error))
381    }
382    pub(crate) async fn abort_upload_checked(
383        &self,
384        session: &UploadSession,
385    ) -> Result<(), StorageError> {
386        let Some(publication) = self.unfinished_publication(session)? else {
387            return Ok(());
388        };
389        self.send(RequestKind::Upload, 0, self.provider.abort_upload(&publication, session)).await
390    }
391}
392
393#[async_trait::async_trait]
394impl<P: ProviderOps + ?Sized> Storage for StorageConnection<P> {
395    fn config(&self) -> StorageConfig {
396        self.provider.config()
397    }
398    fn single_request_limit(&self) -> u64 {
399        self.provider.single_request_limit()
400    }
401    fn read_part_size(&self) -> Option<std::num::NonZeroU64> {
402        self.provider.read_part_size()
403    }
404    async fn set_s3_credentials(&self, credentials: S3Credentials) -> Result<(), StorageError> {
405        self.provider.set_s3_credentials(credentials).await
406    }
407    async fn set_oauth_tokens(&self, tokens: OAuthTokens) -> Result<(), StorageError> {
408        ConnectionCredentials {
409            location: self.config(),
410            credentials: StorageCredentials::OAuth(tokens.clone()),
411        }
412        .validate()?;
413        self.provider.set_oauth_tokens(tokens).await
414    }
415    async fn account(&self) -> Result<String, StorageError> {
416        self.resent(move || async move {
417            self.cooling()?;
418            self.observed(self.send(RequestKind::Account, 0, self.provider.account()).await).await
419        })
420        .await
421    }
422    /// During setup a resend refused as an occupied path may have met its
423    /// own earlier attempt: a copy publication can store the object, then
424    /// have its temporary's deletion refused (§20.6). The object holding
425    /// exactly these sealed bytes settles it as created.
426    async fn create_reporting(
427        &self,
428        path: &ObjectPath,
429        bytes: &[u8],
430        sent: &ByteProgress,
431    ) -> Result<Option<StoredObject>, StorageError> {
432        let mut resent = false;
433        loop {
434            let created = self.create_once(path, bytes, sent).await;
435            if let Err(error) = &created {
436                if resent && error.own_failure() == Some(StorageFailure::AlreadyExists) {
437                    let stored = self.read_object(path, None).await?;
438                    return if stored.bytes == bytes { Ok(Some(stored.metadata)) } else { created };
439                }
440                if self.resend_after(error).await {
441                    resent = true;
442                    continue;
443                }
444            }
445            return created;
446        }
447    }
448    async fn replace(&self, previous: &StoredObject, bytes: &[u8]) -> Result<(), StorageError> {
449        if !previous.path.is_replaceable() {
450            return Err(StorageFailure::InvalidPath.into());
451        }
452        self.replace_object(previous, bytes).await
453    }
454    async fn status(&self, path: &ObjectPath) -> Result<Option<StoredObject>, StorageError> {
455        self.resent(move || async move {
456            self.cooling()?;
457            self.observed(
458                async {
459                    let object =
460                        self.send(RequestKind::Status, 0, self.provider.status(path)).await?;
461                    if let Some(object) = &object {
462                        object.check(path)?;
463                    }
464                    Ok(object)
465                }
466                .await,
467            )
468            .await
469        })
470        .await
471    }
472    async fn stream_object_reporting(
473        &self,
474        path: &ObjectPath,
475        range: Option<ByteRange>,
476        received: &ByteProgress,
477    ) -> Result<ObjectStream<'_>, StorageError> {
478        loop {
479            match self.stream_once(path, range, received).await {
480                Err(error) if !self.resend_after(&error).await => return Err(error),
481                Err(_) => {}
482                stream => return stream,
483            }
484        }
485    }
486    async fn list(&self, prefix: &ObjectPrefix) -> Result<Vec<StoredObject>, StorageError> {
487        self.resent(move || async move {
488            self.cooling()?;
489            self.observed(self.list_checked(prefix).await).await
490        })
491        .await
492    }
493
494    async fn delete(&self, path: &ObjectPath) -> Result<(), StorageError> {
495        self.resent(move || async move {
496            self.cooling()?;
497            self.observed(self.send(RequestKind::Delete, 0, self.provider.delete(path)).await).await
498        })
499        .await
500    }
501    async fn end_superseded_floor_sessions(
502        &self,
503        store: coven_foundation::id_source::StoreId,
504    ) -> Result<(), StorageError> {
505        if !self.provider.dates_floor_sessions() {
506            return Ok(());
507        }
508        let Some(floor) = self.status(&ObjectPath::floor(store)).await? else {
509            return Ok(());
510        };
511        self.cooling()?;
512        let left = self
513            .observed(
514                self.send(RequestKind::List, 0, self.provider.superseded_floor_uploads(&floor))
515                    .await,
516            )
517            .await?;
518        self.end_left(&left).await
519    }
520    async fn end_settled_claims(
521        &self,
522        store: coven_foundation::id_source::StoreId,
523        through: u64,
524    ) -> Result<(), StorageError> {
525        self.cooling()?;
526        let left = self
527            .observed(
528                self.send(
529                    RequestKind::List,
530                    0,
531                    self.provider.settled_claim_uploads(store, through),
532                )
533                .await,
534            )
535            .await?;
536        self.end_left(&left).await
537    }
538    async fn end_device_uploads(
539        &self,
540        store: coven_foundation::id_source::StoreId,
541        device: DeviceId,
542    ) -> Result<(), StorageError> {
543        self.cooling()?;
544        let left = self
545            .observed(
546                self.send(RequestKind::List, 0, self.provider.device_uploads(store, device)).await,
547            )
548            .await?;
549        self.end_left(&left).await
550    }
551    async fn grant_access(&self, account: &str) -> Result<AccessGrant, StorageError> {
552        self.account_change(self.provider.grant_access(account)).await
553    }
554    async fn revoke_access(&self, account: &str) -> Result<Vec<RetainedAccess>, StorageError> {
555        self.account_change(self.provider.revoke_access(account)).await
556    }
557    async fn join(
558        &self,
559        store: coven_foundation::id_source::StoreId,
560        invitation: &StorageInvitation,
561    ) -> Result<(), StorageError> {
562        self.cooling()?;
563        let result = async {
564            invitation.check(&self.config())?;
565            self.send(RequestKind::Account, 0, self.provider.join(invitation)).await?;
566            self.check_location_claim(store).await?;
567            self.status(&ObjectPath::floor(store)).await?;
568            Ok(())
569        }
570        .await;
571        self.observed(result).await
572    }
573    async fn leave(&self, invitation: &StorageInvitation) -> Result<(), StorageError> {
574        self.cooling()?;
575        let result = async {
576            invitation.check(&self.config())?;
577            self.send(RequestKind::Account, 0, self.provider.leave(invitation)).await
578        }
579        .await;
580        self.observed(result).await
581    }
582    async fn begin_upload(
583        &self,
584        path: &ObjectPath,
585        total: u64,
586    ) -> Result<UploadSession, StorageError> {
587        self.cooling()?;
588        self.observed(self.begin_upload_checked(&UploadTarget::Create(path.clone()), total).await)
589            .await
590    }
591    async fn begin_replace(
592        &self,
593        previous: &StoredObject,
594        total: u64,
595    ) -> Result<UploadSession, StorageError> {
596        self.cooling()?;
597        self.observed(
598            self.begin_upload_checked(&UploadTarget::Replace(previous.clone()), total).await,
599        )
600        .await
601    }
602    async fn restart_upload(&self, expired: &UploadSession) -> Result<UploadSession, StorageError> {
603        self.cooling()?;
604        let result = async {
605            expired.check(&self.config())?;
606            if expired.is_complete() {
607                return Err(StorageFailure::InvalidPart.into());
608            }
609            if matches!(expired.state, crate::session::SessionState::S3Restart { .. }) {
610                self.abort_upload_checked(expired).await?;
611            }
612            self.begin_upload_checked(expired.target(), expired.total_bytes()).await
613        }
614        .await;
615        self.observed(result).await
616    }
617    async fn resume_upload(&self, session: &mut UploadSession) -> Result<(), StorageError> {
618        self.cooling()?;
619        let result = async {
620            let Some(publication) = self.unfinished_publication(session)? else {
621                return Ok(());
622            };
623            let result = self
624                .send(RequestKind::Upload, 0, self.provider.resume_upload(&publication, session))
625                .await;
626            result.map_err(|error| session.target.error(error))
627        }
628        .await;
629        self.observed(result).await
630    }
631    fn admit_part<'a>(
632        &self,
633        session: &mut UploadSession,
634        bytes: &'a [u8],
635    ) -> Result<PartPermit<'a>, StorageError> {
636        self.cooling()?;
637        self.part_permit(session, bytes)
638    }
639    async fn send_part(
640        &self,
641        permit: PartPermit<'_>,
642        session: &mut UploadSession,
643        sent: &ByteProgress,
644    ) -> Result<(), StorageError> {
645        self.observed(self.send_permitted(permit, session, sent).await).await
646    }
647    async fn send_independent_part(
648        &self,
649        session: &UploadSession,
650        offset: u64,
651        bytes: &[u8],
652        sent: &ByteProgress,
653    ) -> Result<ConfirmedPart, StorageError> {
654        self.cooling()?;
655        if !session.parts_independent() {
656            return Err(StorageFailure::InvalidPart.into());
657        }
658        // An S3 part is refused for its size before its location, as an
659        // admitted part is.
660        session.part_end(offset, bytes.len())?;
661        session.check(&self.config())?;
662        let publication = self.publication(session.path())?;
663        // The part's bytes are heard from where it starts in the object.
664        let sent = sent.starting_at(offset);
665        let result = self
666            .send(
667                RequestKind::Upload,
668                bytes.len() as u64,
669                self.provider.send_independent_part(&publication, session, offset, bytes, &sent),
670            )
671            .await;
672        self.observed(result.map_err(|error| session.target.error(error))).await
673    }
674    async fn finish_upload(&self, session: &mut UploadSession) -> Result<(), StorageError> {
675        self.cooling()?;
676        self.observed(self.finish_upload_checked(session).await).await
677    }
678    async fn abort_upload(&self, session: &UploadSession) -> Result<(), StorageError> {
679        self.cooling()?;
680        self.observed(self.abort_upload_checked(session).await).await
681    }
682}
683
684/// What a provider's tests reach beneath a connection: the provider itself,
685/// and the device and gate it writes for.
686#[cfg(test)]
687impl<P: ?Sized> StorageConnection<P> {
688    pub(crate) fn provider(&self) -> &Arc<P> {
689        &self.provider
690    }
691
692    pub(crate) fn provider_mut(&mut self) -> &mut Arc<P> {
693        &mut self.provider
694    }
695
696    pub(crate) fn set_device(&mut self, device: DeviceId) {
697        self.device = device;
698    }
699
700    pub(crate) fn set_publications(&mut self, publications: Arc<dyn PublicationGate>) {
701        self.publications = publications;
702    }
703}
704
705// The connection's location claim and folder label requests and the
706// in-memory connection's own calls read its private parts, so they live
707// beneath it.
708#[path = "folder_label.rs"]
709mod folder_label;
710#[path = "location_claim.rs"]
711pub(crate) mod location_claim;
712#[cfg(any(test, feature = "test-utils"))]
713#[path = "test_utils.rs"]
714pub mod test_utils;
715
716#[cfg(test)]
717#[path = "connection_tests.rs"]
718mod tests;
coven_storage/connection.rs

coven_storage/
connection.rs
