coven/storage/local/
traits.rs

1//! Local managed blob storage: plaintext files at content-addressed paths.
2use crate::storage::local::storage_path;
3use thiserror::Error;
4
5#[derive(Error, Debug)]
6pub enum StorageError {
7    #[error("IO error: {0}")]
8    Io(#[from] std::io::Error),
9    #[error("Storage not configured")]
10    NotConfigured,
11    #[error("Cloud storage error: {0}")]
12    Cloud(String),
13    #[error("Database error: {0}")]
14    Database(String),
15    /// The file id does not form a safe content-addressed path (see
16    /// [`crate::store_dir::PathTokenError`]). For local managed storage the id is
17    /// device-generated, so this is a programmer error surfaced loudly, never a
18    /// silent mis-shard.
19    #[error("invalid blob id: {0}")]
20    InvalidId(#[from] crate::store_dir::PathTokenError),
21}
22
23/// Progress callback type: (bytes_written, total_bytes)
24pub(super) type ProgressCallback = Box<dyn Fn(usize, usize) + Send + Sync>;
25
26/// Storage implementation for managed local storage.
27///
28/// Writes files to `store_dir/storage/ab/cd/{file_id}` as plaintext.
29/// Local files are never encrypted -- encryption only happens when uploading
30/// to the cloud home.
31#[derive(Clone)]
32pub struct BlobStore {
33    store_dir: crate::store_dir::StoreDir,
34}
35
36impl BlobStore {
37    /// Create storage for managed local blobs.
38    pub fn new_local(store_dir: crate::store_dir::StoreDir) -> Self {
39        Self { store_dir }
40    }
41
42    /// Write bytes to local storage without creating a DB record.
43    ///
44    /// Uses the given `file_id` for the hash-based storage path.
45    pub async fn store_bytes(
46        &self,
47        file_id: &str,
48        data: &[u8],
49        on_progress: ProgressCallback,
50    ) -> Result<(), StorageError> {
51        use tokio::io::AsyncWriteExt;
52
53        let total_bytes = data.len();
54        on_progress(0, total_bytes);
55
56        let rel_path = storage_path(file_id)?;
57        let path = self.store_dir.join(&rel_path);
58
59        if let Some(parent) = path.parent() {
60            tokio::fs::create_dir_all(parent).await?;
61        }
62
63        let batch_size = 1_048_576;
64        let mut file = tokio::fs::File::create(&path).await?;
65        let mut bytes_written = 0usize;
66
67        for chunk in data.chunks(batch_size) {
68            file.write_all(chunk).await?;
69            bytes_written += chunk.len();
70            on_progress(bytes_written.min(total_bytes), total_bytes);
71        }
72
73        file.flush().await?;
74
75        Ok(())
76    }
77
78    /// Stream a source file from disk into local storage without buffering the
79    /// whole thing in memory. Progress is reported in 1 MiB batches to match
80    /// the cadence of `store_bytes`.
81    pub async fn store_from_path(
82        &self,
83        file_id: &str,
84        source: &std::path::Path,
85        on_progress: ProgressCallback,
86    ) -> Result<(), StorageError> {
87        use tokio::io::{AsyncReadExt, AsyncWriteExt};
88
89        let total_bytes = tokio::fs::metadata(source).await?.len() as usize;
90        on_progress(0, total_bytes);
91
92        let rel_path = storage_path(file_id)?;
93        let path = self.store_dir.join(&rel_path);
94
95        if let Some(parent) = path.parent() {
96            tokio::fs::create_dir_all(parent).await?;
97        }
98
99        let batch_size = 1_048_576;
100        let mut reader = tokio::fs::File::open(source).await?;
101        let mut dest = tokio::fs::File::create(&path).await?;
102        let mut buf = vec![0u8; batch_size];
103        let mut bytes_written = 0usize;
104
105        // Fill `buf` up to `batch_size` per iteration so progress fires once per
106        // full batch. A single `read` on `tokio::fs::File` (even via `BufReader`)
107        // can return far less than the requested length, so without the inner
108        // fill loop we'd report progress on every short read.
109        loop {
110            let mut filled = 0usize;
111            while filled < batch_size {
112                let n = reader.read(&mut buf[filled..]).await?;
113                if n == 0 {
114                    break;
115                }
116                filled += n;
117            }
118            if filled == 0 {
119                break;
120            }
121            dest.write_all(&buf[..filled]).await?;
122            bytes_written += filled;
123            on_progress(bytes_written.min(total_bytes), total_bytes);
124        }
125
126        dest.flush().await?;
127
128        Ok(())
129    }
130}
131
132#[cfg(test)]
133mod tests {
134    use super::*;
135    use std::sync::{Arc, Mutex};
136    use tempfile::TempDir;
137
138    #[tokio::test]
139    async fn store_from_path_copies_bytes_and_reports_1mib_cadence() {
140        let temp = TempDir::new().unwrap();
141        let store_dir = crate::store_dir::StoreDir::new(temp.path());
142        let storage = BlobStore::new_local(store_dir);
143
144        // 2.5 MiB — two full batches plus a partial tail.
145        let total = 2_621_440usize;
146        let source_bytes: Vec<u8> = (0..total).map(|i| (i % 256) as u8).collect();
147        let source_path = temp.path().join("source.bin");
148        tokio::fs::write(&source_path, &source_bytes).await.unwrap();
149
150        let calls: Arc<Mutex<Vec<(usize, usize)>>> = Arc::new(Mutex::new(Vec::new()));
151        let calls_clone = calls.clone();
152        let file_id = "abcdef1234567890";
153        storage
154            .store_from_path(
155                file_id,
156                &source_path,
157                Box::new(move |written, total| {
158                    calls_clone.lock().unwrap().push((written, total));
159                }),
160            )
161            .await
162            .unwrap();
163
164        let dest_path = temp.path().join(storage_path(file_id).expect("valid id"));
165        let dest_bytes = tokio::fs::read(&dest_path).await.unwrap();
166        assert_eq!(dest_bytes, source_bytes, "destination equals source");
167
168        let calls = calls.lock().unwrap();
169        assert_eq!(
170            &*calls,
171            &[
172                (0, total),
173                (1_048_576, total),
174                (2_097_152, total),
175                (total, total),
176            ],
177            "progress fires once per 1 MiB batch (plus initial and final partial)",
178        );
179    }
180}
coven/storage/local/traits.rs

coven/storage/local/
traits.rs
