edgestore_repl/
filesystem_remote_store.rs

1//! FilesystemRemoteStore — local-filesystem implementation of `RemoteStore`.
2//!
3//! Files are content-addressed: named `{hash_hex}.seg` where `hash_hex` is the
4//! 64-character lowercase hex encoding of the 32-byte BLAKE3 hash. Listing the
5//! directory is equivalent to listing stored segments.
6//!
7//! This is the Phase 4 implementation (Plan 04-04, D04). Real S3 (`S3RemoteStore`)
8//! is a future phase deliverable.
9
10use std::path::PathBuf;
11
12use edgestore::error::EdgestoreError;
13use edgestore::RemoteStore;
14
15/// Local-filesystem implementation of `RemoteStore`.
16///
17/// All operations are idempotent and atomic where applicable.
18/// `upload` uses a `.tmp` write + rename to prevent torn writes (T-04-10).
19pub struct FilesystemRemoteStore {
20    base_dir: PathBuf,
21}
22
23impl FilesystemRemoteStore {
24    /// Create a new `FilesystemRemoteStore` rooted at `base_dir`.
25    ///
26    /// Creates `base_dir` (and all parent directories) if it does not exist.
27    pub fn new(base_dir: PathBuf) -> Result<Self, EdgestoreError> {
28        std::fs::create_dir_all(&base_dir)
29            .map_err(|e| EdgestoreError::ReplicationError(e.to_string()))?;
30        Ok(Self { base_dir })
31    }
32
33    /// Encode a 32-byte hash as a 64-character lowercase hex string.
34    fn hash_hex(hash: &[u8; 32]) -> String {
35        hash.iter().map(|b| format!("{:02x}", b)).collect::<String>()
36    }
37
38    /// Return the path `{base_dir}/{hash_hex}.seg` for the given hash.
39    fn seg_path(&self, hash: &[u8; 32]) -> PathBuf {
40        self.base_dir.join(format!("{}.seg", Self::hash_hex(hash)))
41    }
42
43    fn aux_path(&self, hash: &[u8; 32], ext: &str) -> Result<PathBuf, EdgestoreError> {
44        if ext.is_empty() || !ext.bytes().all(|b| b.is_ascii_lowercase()) {
45            return Err(EdgestoreError::InvalidOperation(format!(
46                "invalid sidecar ext {:?}: only lowercase ASCII letters allowed",
47                ext
48            )));
49        }
50        Ok(self.base_dir.join(format!("{}.{}", Self::hash_hex(hash), ext)))
51    }
52}
53
54impl RemoteStore for FilesystemRemoteStore {
55    /// Store `data` under `hash`. Idempotent: if the file already exists, returns `Ok(())`.
56    ///
57    /// Writes to a `.tmp` file first, then renames atomically (T-04-10).
58    fn upload(&self, hash: &[u8; 32], data: &[u8]) -> Result<(), EdgestoreError> {
59        let dest = self.seg_path(hash);
60
61        // Content-addressed: if it already exists, nothing to do.
62        if dest.exists() {
63            return Ok(());
64        }
65
66        let tmp = self
67            .base_dir
68            .join(format!("{}.tmp", Self::hash_hex(hash)));
69
70        std::fs::write(&tmp, data)
71            .map_err(|e| EdgestoreError::ReplicationError(e.to_string()))?;
72
73        std::fs::rename(&tmp, &dest)
74            .map_err(|e| EdgestoreError::ReplicationError(e.to_string()))?;
75
76        Ok(())
77    }
78
79    /// Download the segment bytes for `hash`.
80    ///
81    /// Returns `EdgestoreError::ReplicationError` if the segment is not present.
82    fn download(&self, hash: &[u8; 32]) -> Result<Vec<u8>, EdgestoreError> {
83        let path = self.seg_path(hash);
84        std::fs::read(&path).map_err(|e| {
85            if e.kind() == std::io::ErrorKind::NotFound {
86                EdgestoreError::ReplicationError(format!(
87                    "segment not found: {}",
88                    Self::hash_hex(hash)
89                ))
90            } else {
91                EdgestoreError::ReplicationError(e.to_string())
92            }
93        })
94    }
95
96    /// List all stored segment hashes by scanning `{base_dir}/*.seg`.
97    ///
98    /// Filenames that are not exactly 64 lowercase hex characters followed by `.seg`
99    /// are silently skipped (T-04-12).
100    fn list(&self) -> Result<Vec<[u8; 32]>, EdgestoreError> {
101        let entries = std::fs::read_dir(&self.base_dir)
102            .map_err(|e| EdgestoreError::ReplicationError(e.to_string()))?;
103
104        let mut hashes = Vec::new();
105
106        for entry in entries.flatten() {
107            let file_name = entry.file_name();
108            let name = match file_name.to_str() {
109                Some(n) => n.to_owned(),
110                None => continue,
111            };
112
113            // Must end with ".seg"
114            if !name.ends_with(".seg") {
115                continue;
116            }
117
118            // Stem must be exactly 64 characters (32 bytes * 2 hex digits).
119            let stem = &name[..name.len() - 4]; // strip ".seg"
120            if stem.len() != 64 {
121                continue;
122            }
123
124            // Parse 64 hex chars → [u8; 32]
125            let parsed: Option<[u8; 32]> = (0..32)
126                .map(|i| u8::from_str_radix(&stem[i * 2..i * 2 + 2], 16).ok())
127                .collect::<Option<Vec<u8>>>()
128                .and_then(|v| v.try_into().ok());
129
130            if let Some(hash) = parsed {
131                hashes.push(hash);
132            }
133        }
134
135        Ok(hashes)
136    }
137
138    /// Remove the segment for `hash`. No-op if the segment does not exist (idempotent).
139    fn delete(&self, hash: &[u8; 32]) -> Result<(), EdgestoreError> {
140        let path = self.seg_path(hash);
141        match std::fs::remove_file(&path) {
142            Ok(()) => Ok(()),
143            Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(()),
144            Err(e) => Err(EdgestoreError::ReplicationError(e.to_string())),
145        }
146    }
147
148    fn upload_aux(
149        &self,
150        hash: &[u8; 32],
151        ext: &str,
152        data: &[u8],
153    ) -> Result<(), EdgestoreError> {
154        let dest = self.aux_path(hash, ext)?;
155        if dest.exists() {
156            return Ok(());
157        }
158        let tmp = self.base_dir.join(format!("{}.{}.tmp", Self::hash_hex(hash), ext));
159        std::fs::write(&tmp, data).map_err(|e| EdgestoreError::ReplicationError(e.to_string()))?;
160        std::fs::rename(&tmp, &dest).map_err(|e| EdgestoreError::ReplicationError(e.to_string()))
161    }
162
163    fn download_aux(&self, hash: &[u8; 32], ext: &str) -> Result<Vec<u8>, EdgestoreError> {
164        let path = self.aux_path(hash, ext)?;
165        std::fs::read(&path).map_err(|e| {
166            if e.kind() == std::io::ErrorKind::NotFound {
167                EdgestoreError::ReplicationError(format!(
168                    "sidecar not found: {}.{}",
169                    Self::hash_hex(hash),
170                    ext
171                ))
172            } else {
173                EdgestoreError::ReplicationError(e.to_string())
174            }
175        })
176    }
177}
178
179#[cfg(test)]
180mod tests {
181    use super::*;
182    use tempfile::TempDir;
183
184    fn make_store() -> (TempDir, FilesystemRemoteStore) {
185        let dir = TempDir::new().expect("tempdir");
186        let store = FilesystemRemoteStore::new(dir.path().to_path_buf())
187            .expect("FilesystemRemoteStore::new");
188        (dir, store)
189    }
190
191    #[test]
192    fn test_upload_download_roundtrip() {
193        let (_dir, store) = make_store();
194        let hash = [0x42u8; 32];
195        let data = b"hello edgestore";
196
197        store.upload(&hash, data).expect("upload");
198        let got = store.download(&hash).expect("download");
199        assert_eq!(got, data);
200    }
201
202    #[test]
203    fn test_upload_idempotent() {
204        let (_dir, store) = make_store();
205        let hash = [0x42u8; 32];
206        let data = b"original";
207
208        store.upload(&hash, data).expect("first upload");
209        // Second upload with same hash must succeed without error.
210        store.upload(&hash, b"different").expect("second upload (idempotent)");
211
212        // File should still contain the original data (idempotent — skipped overwrite).
213        let got = store.download(&hash).expect("download after idempotent upload");
214        assert_eq!(got, data);
215    }
216
217    #[test]
218    fn test_list_returns_uploaded_hashes() {
219        let (_dir, store) = make_store();
220        let hash1 = [0x01u8; 32];
221        let hash2 = [0x02u8; 32];
222        let hash3 = [0x03u8; 32];
223
224        store.upload(&hash1, b"a").expect("upload 1");
225        store.upload(&hash2, b"b").expect("upload 2");
226        store.upload(&hash3, b"c").expect("upload 3");
227
228        let mut listed = store.list().expect("list");
229        listed.sort();
230
231        let mut expected = vec![hash1, hash2, hash3];
232        expected.sort();
233
234        assert_eq!(listed, expected);
235    }
236
237    #[test]
238    fn test_delete_removes_file() {
239        let (_dir, store) = make_store();
240        let hash = [0x42u8; 32];
241
242        store.upload(&hash, b"segment data").expect("upload");
243        store.delete(&hash).expect("delete");
244
245        // Download must now fail.
246        let result = store.download(&hash);
247        assert!(result.is_err(), "download after delete should return Err");
248    }
249
250    #[test]
251    fn test_download_not_found() {
252        let (_dir, store) = make_store();
253        let hash = [0xFFu8; 32];
254
255        let result = store.download(&hash);
256        assert!(result.is_err(), "download of non-existent hash should return Err");
257    }
258
259    #[test]
260    fn test_upload_aux_download_aux_roundtrip() {
261        let (_dir, store) = make_store();
262        let hash = [0x77u8; 32];
263
264        store.upload_aux(&hash, "idx", b"index-data").expect("upload_aux idx");
265        store.upload_aux(&hash, "xf", b"filter-data").expect("upload_aux xf");
266        store.upload_aux(&hash, "meta", b"meta-data").expect("upload_aux meta");
267
268        assert_eq!(store.download_aux(&hash, "idx").unwrap(), b"index-data");
269        assert_eq!(store.download_aux(&hash, "xf").unwrap(), b"filter-data");
270        assert_eq!(store.download_aux(&hash, "meta").unwrap(), b"meta-data");
271    }
272
273    #[test]
274    fn test_upload_aux_idempotent() {
275        let (_dir, store) = make_store();
276        let hash = [0x88u8; 32];
277
278        store.upload_aux(&hash, "idx", b"v1").expect("first upload_aux");
279        store.upload_aux(&hash, "idx", b"v2").expect("second upload_aux idempotent");
280
281        // Content-addressed: first upload wins.
282        assert_eq!(store.download_aux(&hash, "idx").unwrap(), b"v1");
283    }
284
285    #[test]
286    fn test_download_aux_not_found() {
287        let (_dir, store) = make_store();
288        let hash = [0x99u8; 32];
289
290        let result = store.download_aux(&hash, "idx");
291        assert!(result.is_err(), "download_aux of non-existent sidecar should Err");
292    }
293
294    #[test]
295    fn test_upload_aux_rejects_path_traversal_ext() {
296        let (_dir, store) = make_store();
297        let hash = [0xAAu8; 32];
298
299        for bad_ext in &["../etc/passwd", "idx/../../shadow", "IDX", "idx.bak", "", "idx\0evil"] {
300            let result = store.upload_aux(&hash, bad_ext, b"data");
301            assert!(
302                result.is_err(),
303                "upload_aux must reject ext {:?}",
304                bad_ext
305            );
306        }
307    }
308}
edgestore_repl/filesystem_remote_store.rs

edgestore_repl/
filesystem_remote_store.rs
