coven_storage/cloud/
exact_upload.rs

1use super::{BlobBody, CloudHomeError};
2use coven_protocol::objects::{ExactObjectRef, StorageError};
3use std::path::Path;
4
5pub(crate) fn accept_unchecked_create_response(
6    created_response_was_observed: bool,
7    object: &ExactObjectRef,
8) -> Result<(), CloudHomeError> {
9    if created_response_was_observed {
10        Ok(())
11    } else {
12        Err(CloudHomeError::AlreadyExists(
13            object.slot().logical_key().to_string(),
14        ))
15    }
16}
17
18pub(crate) async fn settle_exact_create<Verify, Verification>(
19    operation: Result<(), CloudHomeError>,
20    verify: Verify,
21) -> Result<super::ExactCreateOutcome, CloudHomeError>
22where
23    Verify: FnOnce(bool) -> Verification,
24    Verification: std::future::Future<Output = Result<(), CloudHomeError>>,
25{
26    match operation {
27        Ok(()) => {
28            verify(true).await?;
29            Ok(super::ExactCreateOutcome::Created)
30        }
31        Err(CloudHomeError::AlreadyExists(_)) => {
32            verify(false).await?;
33            Ok(super::ExactCreateOutcome::AlreadyPresent)
34        }
35        Err(operation) => match verify(false).await {
36            Ok(()) => Ok(super::ExactCreateOutcome::AlreadyPresent),
37            Err(CloudHomeError::NotFound(_)) => Err(operation),
38            Err(collision @ CloudHomeError::SlotCollision(_)) => Err(collision),
39            Err(settlement) => Err(CloudHomeError::UnresolvedOutcome {
40                operation: Box::new(operation),
41                settlement: Box::new(settlement),
42            }),
43        },
44    }
45}
46
47/// One immutable exact object together with a replayable source for its final
48/// stored bytes. Provider adapters derive their own native checksums from this
49/// source and can reopen it after an ambiguous create response.
50#[derive(Clone, Copy)]
51pub struct ExactUpload<'source> {
52    object: &'source ExactObjectRef,
53    source: ExactUploadSource<'source>,
54}
55
56#[derive(Clone, Copy)]
57pub enum ExactUploadSource<'source> {
58    Bytes(&'source [u8]),
59    File(&'source Path),
60}
61
62impl<'source> ExactUpload<'source> {
63    pub fn from_bytes(
64        object: &'source ExactObjectRef,
65        bytes: &'source [u8],
66    ) -> Result<Self, StorageError> {
67        object.verify(bytes)?;
68        Ok(Self {
69            object,
70            source: ExactUploadSource::Bytes(bytes),
71        })
72    }
73
74    pub async fn from_file(
75        object: &'source ExactObjectRef,
76        path: &'source Path,
77    ) -> Result<Self, StorageError> {
78        let (size, digest) = coven_foundation::local_file::file_facts(path)
79            .await
80            .map_err(StorageError::LocalFilesystem)?;
81        object.verify_stored_facts(
82            path,
83            size,
84            coven_protocol::store_commit::ObjectHash::from_digest(digest),
85        )?;
86        Ok(Self {
87            object,
88            source: ExactUploadSource::File(path),
89        })
90    }
91
92    pub fn object(&self) -> &ExactObjectRef {
93        self.object
94    }
95
96    pub fn source(&self) -> ExactUploadSource<'source> {
97        self.source
98    }
99
100    pub fn verify_stored_bytes(&self, bytes: &[u8]) -> Result<(), CloudHomeError> {
101        self.object.verify(bytes).map_err(|_| {
102            CloudHomeError::SlotCollision(self.object.slot().logical_key().to_string())
103        })
104    }
105
106    pub async fn body(&self) -> Result<BlobBody, CloudHomeError> {
107        match self.source {
108            ExactUploadSource::Bytes(bytes) => Ok(BlobBody::from_bytes(bytes.to_vec())),
109            ExactUploadSource::File(path) => BlobBody::from_file(path)
110                .await
111                .map_err(CloudHomeError::Local),
112        }
113    }
114}
115
116#[cfg(test)]
117mod tests {
118    use super::*;
119    use coven_protocol::objects::ObjectSlot;
120    use coven_protocol::store_commit::ObjectHash;
121
122    fn object() -> ExactObjectRef {
123        ExactObjectRef::new(
124            ObjectSlot::logical("unchecked/object".to_string()).expect("logical slot"),
125            5,
126            ObjectHash::digest(b"bytes"),
127        )
128    }
129
130    #[test]
131    fn unchecked_accepts_only_a_witnessed_successful_create() {
132        let object = object();
133
134        assert!(accept_unchecked_create_response(true, &object).is_ok());
135        assert!(matches!(
136            accept_unchecked_create_response(false, &object),
137            Err(CloudHomeError::AlreadyExists(key)) if key == "unchecked/object"
138        ));
139    }
140
141    #[tokio::test]
142    async fn unchecked_does_not_turn_occupied_or_ambiguous_results_into_success() {
143        let object = object();
144        let occupied = settle_exact_create(
145            Err(CloudHomeError::AlreadyExists(
146                "unchecked/object".to_string(),
147            )),
148            |_| async { accept_unchecked_create_response(false, &object) },
149        )
150        .await;
151        assert!(matches!(
152            occupied,
153            Err(CloudHomeError::AlreadyExists(key)) if key == "unchecked/object"
154        ));
155
156        let ambiguous = settle_exact_create(
157            Err(CloudHomeError::Transport("lost response".to_string())),
158            |_| async { accept_unchecked_create_response(false, &object) },
159        )
160        .await;
161        assert!(matches!(
162            ambiguous,
163            Err(CloudHomeError::UnresolvedOutcome { operation, settlement })
164                if matches!(*operation, CloudHomeError::Transport(_))
165                    && matches!(*settlement, CloudHomeError::AlreadyExists(_))
166        ));
167    }
168}
coven_storage/cloud/exact_upload.rs

coven_storage/cloud/
exact_upload.rs
