coven_storage/
local_file.rs

1//! Storage-side adapters over the foundation staged-file machinery in
2//! [`coven_foundation::local_file`]: typed plaintext chunk sources whose remote errors
3//! stay classified, and exact-fact reads expressed in protocol hashes.
4
5use std::path::Path;
6
7use async_trait::async_trait;
8use sha2::{Digest, Sha256};
9use tokio::io::AsyncReadExt;
10
11pub struct PlaintextReader(
12    Box<dyn coven_foundation::local_file::PlaintextChunkReader<Error = PlaintextChunkError>>,
13);
14
15impl PlaintextReader {
16    pub(crate) async fn next_chunk(&mut self, max: usize) -> Result<Vec<u8>, PlaintextChunkError> {
17        self.0.next_chunk(max).await
18    }
19
20    #[cfg(test)]
21    pub(crate) fn from_test_reader(
22        reader: impl coven_foundation::local_file::PlaintextChunkReader<Error = PlaintextChunkError>
23            + 'static,
24    ) -> Self {
25        Self(Box::new(reader))
26    }
27}
28
29#[derive(Debug, thiserror::Error)]
30pub(crate) enum PlaintextChunkError {
31    #[error(transparent)]
32    Remote(#[from] coven_protocol::objects::StorageError),
33    #[error("invalid remote content: {0}")]
34    InvalidContent(String),
35    #[error("decrypt remote content {context}: {source}")]
36    Decryption {
37        context: String,
38        #[source]
39        source: coven_keys::encryption::EncryptionError,
40    },
41    #[error("local plaintext source: {0}")]
42    Local(coven_foundation::atomic_file::FileError),
43}
44
45impl From<PlaintextChunkError> for coven_protocol::objects::StorageError {
46    fn from(error: PlaintextChunkError) -> Self {
47        match error {
48            PlaintextChunkError::Remote(error) => error,
49            PlaintextChunkError::InvalidContent(message) => Self::InvalidContent(message),
50            PlaintextChunkError::Decryption { context, source } => {
51                Self::Decryption { context, source }
52            }
53            PlaintextChunkError::Local(error) => Self::LocalFilesystem(error),
54        }
55    }
56}
57
58pub(crate) async fn open_reader(
59    path: &Path,
60) -> Result<PlaintextReader, coven_foundation::atomic_file::FileError> {
61    let file = tokio::fs::File::open(path).await.map_err(|source| {
62        coven_foundation::atomic_file::FileError::Path {
63            operation: "open local blob for streaming",
64            path: path.to_path_buf(),
65            source,
66        }
67    })?;
68    Ok(PlaintextReader(Box::new(FilePlaintextReader {
69        file,
70        path: path.to_path_buf(),
71        exact: None,
72    })))
73}
74
75pub(crate) async fn open_exact_reader(
76    path: &Path,
77    expected_size: u64,
78    expected_hash: coven_protocol::store_commit::ObjectHash,
79    progress: crate::cloud::PreparationProgress,
80) -> Result<PlaintextReader, coven_foundation::atomic_file::FileError> {
81    let file = tokio::fs::File::open(path).await.map_err(|source| {
82        coven_foundation::atomic_file::FileError::Path {
83            operation: "open local blob for streaming",
84            path: path.to_path_buf(),
85            source,
86        }
87    })?;
88    Ok(PlaintextReader(Box::new(FilePlaintextReader {
89        file,
90        path: path.to_path_buf(),
91        exact: Some(ExactPlaintextRead {
92            expected_size,
93            expected_hash,
94            size: 0,
95            hasher: Some(Sha256::new()),
96            progress,
97        }),
98    })))
99}
100
101/// Stream the exact stored size and SHA-256 identity of a local file.
102pub(crate) async fn exact_file_facts(
103    path: &Path,
104) -> Result<(u64, coven_protocol::store_commit::ObjectHash), coven_foundation::atomic_file::FileError>
105{
106    let (size, digest) = coven_foundation::local_file::file_facts(path).await?;
107    Ok((
108        size,
109        coven_protocol::store_commit::ObjectHash::from_digest(digest),
110    ))
111}
112
113struct FilePlaintextReader {
114    file: tokio::fs::File,
115    path: std::path::PathBuf,
116    exact: Option<ExactPlaintextRead>,
117}
118
119struct ExactPlaintextRead {
120    expected_size: u64,
121    expected_hash: coven_protocol::store_commit::ObjectHash,
122    size: u64,
123    hasher: Option<Sha256>,
124    progress: crate::cloud::PreparationProgress,
125}
126
127#[async_trait]
128impl coven_foundation::local_file::PlaintextChunkReader for FilePlaintextReader {
129    type Error = PlaintextChunkError;
130
131    async fn next_chunk(&mut self, max: usize) -> Result<Vec<u8>, PlaintextChunkError> {
132        debug_assert!(max > 0, "next_chunk max must be positive");
133        let mut buf = vec![0u8; max];
134        let mut filled = 0;
135        while filled < max {
136            let read = self.file.read(&mut buf[filled..]).await.map_err(|source| {
137                PlaintextChunkError::Local(coven_foundation::atomic_file::FileError::Path {
138                    operation: "read local blob",
139                    path: self.path.clone(),
140                    source,
141                })
142            })?;
143            if read == 0 {
144                break;
145            }
146            if let Some(exact) = &mut self.exact {
147                exact.size = exact.size.checked_add(read as u64).ok_or_else(|| {
148                    PlaintextChunkError::InvalidContent(
149                        "local blob size overflow while preparing upload".to_string(),
150                    )
151                })?;
152                if exact.size > exact.expected_size {
153                    return Err(PlaintextChunkError::InvalidContent(format!(
154                        "local blob grew past its declared {} bytes while preparing upload",
155                        exact.expected_size
156                    )));
157                }
158                exact
159                    .hasher
160                    .as_mut()
161                    .expect("exact plaintext hash is unfinished")
162                    .update(&buf[filled..filled + read]);
163                (exact.progress)(exact.size);
164            }
165            filled += read;
166        }
167        buf.truncate(filled);
168        if filled == 0 {
169            if let Some(exact) = &mut self.exact {
170                let digest = exact
171                    .hasher
172                    .take()
173                    .expect("exact plaintext reader reaches EOF once")
174                    .finalize();
175                let actual_hash =
176                    coven_protocol::store_commit::ObjectHash::from_digest(digest.into());
177                if exact.size != exact.expected_size || actual_hash != exact.expected_hash {
178                    return Err(PlaintextChunkError::InvalidContent(format!(
179                        "local blob source differs from its declared size/hash: expected {} bytes/{}, read {} bytes/{}",
180                        exact.expected_size, exact.expected_hash, exact.size, actual_hash
181                    )));
182                }
183            }
184        }
185        Ok(buf)
186    }
187}
coven_storage/local_file.rs

coven_storage/
local_file.rs
