coven_storage/
transfer.rs

1use crate::{ObjectPath, Storage, StorageError, StorageFailure};
2use std::future::Future;
3
4/// Most bytes a caller collects in memory for one request body, matching
5/// [`crate::BUFFERED_READ_LIMIT`] for response bodies. A single request
6/// carries its whole body, so no object goes in one request past this, even
7/// on a provider accepting more: a large object goes one part at a time and
8/// never has to fit in memory (§17.1).
9pub const BUFFERED_REQUEST_LIMIT: u64 = 32 * 1024 * 1024;
10
11/// Whether an object of `size` bytes goes up in parts rather than in one
12/// request: past the provider's single-request limit or
13/// [`BUFFERED_REQUEST_LIMIT`], whichever is less (§20.6). The 64-KiB
14/// encryption chunk plays no part. Every upload routes by this alone: a
15/// create of complete bytes, and a caller sending a repeatable source, which
16/// reads it whole only for one request and otherwise records a session.
17pub fn goes_in_parts(storage: &(impl Storage + ?Sized), size: u64) -> bool {
18    size > storage.single_request_limit().min(BUFFERED_REQUEST_LIMIT)
19}
20
21/// Send complete bytes in the provider's single request or, when
22/// [`goes_in_parts`] says so, through its session operations, telling `sent`
23/// how far through them the parts reach. Callers needing crash continuation
24/// use the session operations directly and persist each returned recording.
25pub(crate) async fn upload_bytes(
26    storage: &crate::StorageConnection<impl crate::ProviderOps + ?Sized>,
27    path: &ObjectPath,
28    bytes: &[u8],
29    sent: &crate::ByteProgress,
30    single_request: impl Future<Output = Result<Option<crate::StoredObject>, StorageError>> + Send,
31) -> Result<Option<crate::StoredObject>, StorageError> {
32    if !goes_in_parts(storage, bytes.len() as u64) {
33        return single_request.await;
34    }
35    let mut session = storage
36        .begin_upload_checked(&crate::UploadTarget::Create(path.clone()), bytes.len() as u64)
37        .await?;
38    let upload = async {
39        while let Some(part) = session.parts_to_send().first().cloned() {
40            let confirmed = session.confirmed_bytes();
41            let span = usize::try_from(part.start)
42                .and_then(|start| Ok(start..usize::try_from(part.end)?))
43                .map_err(|error| StorageFailure::InvalidPart.with_source(error))?;
44            let part_bytes = bytes.get(span).ok_or(StorageFailure::InvalidPart)?;
45            let permit = storage.part_permit(&mut session, part_bytes)?;
46            storage.send_permitted(permit, &mut session, sent).await?;
47            if session.confirmed_bytes() <= confirmed {
48                return Err(StorageFailure::Protocol.with_source("upload did not advance"));
49            }
50        }
51        storage.finish_upload_checked(&mut session).await
52    }
53    .await;
54    match upload {
55        Ok(()) => Ok(None),
56        Err(operation) => {
57            with_cleanup(Err(operation), storage.abort_upload_checked(&session).await)
58        }
59    }
60}
61
62pub(crate) fn with_cleanup<T>(
63    operation: Result<T, StorageError>,
64    cleanup: Result<(), StorageError>,
65) -> Result<T, StorageError> {
66    match (operation, cleanup) {
67        (Ok(value), Ok(())) => Ok(value),
68        (Err(error), Ok(())) | (Ok(_), Err(error)) => Err(error),
69        (Err(operation), Err(cleanup)) => Err(StorageError::Cleanup {
70            operation: Box::new(operation),
71            cleanup: Box::new(cleanup),
72        }),
73    }
74}
75
76pub(crate) fn check_single_request(size: u64, limit: u64) -> Result<(), StorageError> {
77    if size > limit {
78        return Err(StorageFailure::SingleRequestTooLarge { size, limit }.into());
79    }
80    Ok(())
81}
82
83#[cfg(test)]
84#[path = "transfer_tests.rs"]
85mod tests;
coven_storage/transfer.rs

coven_storage/
transfer.rs
