fxcp_core/
chunk_transport.rs

1// SPDX-License-Identifier: GPL-2.0-or-later
2// Copyright (C) 2025 Joel Wirāmu Pauling <aenertia@aenertia.net>
3//! Chunk-level I/O abstraction for delta upsert across transport underlays.
4
5use async_trait::async_trait;
6use crate::error::Result;
7
8/// Chunk-level I/O abstraction for delta upsert across transport underlays.
9///
10/// Implementations: [`PosixChunkTransport`] (local/USB/SMB), `S3ChunkTransport` (S3/MinIO).
11/// All methods use BLAKE3 hex strings as chunk identifiers.
12#[async_trait]
13pub trait ChunkTransport: Send + Sync {
14    /// Upload a chunk. Idempotent  --  re-uploading same hash+data is a no-op for most backends.
15    async fn put_chunk(&self, hash: &str, data: &[u8]) -> Result<()>;
16
17    /// Download a chunk by BLAKE3 hash.
18    async fn get_chunk(&self, hash: &str) -> Result<Vec<u8>>;
19
20    /// Check if a chunk exists without downloading it.
21    async fn has_chunk(&self, hash: &str) -> Result<bool>;
22
23    /// Upload opaque manifest bytes (replaces any existing manifest).
24    async fn put_manifest(&self, data: &[u8]) -> Result<()>;
25
26    /// Download manifest bytes. Returns `None` if no manifest stored yet.
27    async fn get_manifest(&self) -> Result<Option<Vec<u8>>>;
28
29    /// Verify transport is accessible and writable.
30    async fn health_check(&self) -> Result<()>;
31}
32
33use std::path::PathBuf;
34
35const CAS_DIR: &str = ".foxing_cas";
36const CHUNKS_DIR: &str = "chunks";
37const TMP_DIR: &str = "chunks/tmp";
38const MANIFEST_FILE: &str = "manifest.bin";
39
40/// Crash-safe POSIX filesystem [`ChunkTransport`] for local/USB/SMB mounts.
41///
42/// Chunk layout: `{root}/.foxing_cas/chunks/{hash[0..2]}/{full_hash}` (2-char shard).
43/// All writes use tmp->fsync->rename->dir_fsync for crash safety.
44#[derive(Debug)]
45pub struct PosixChunkTransport {
46    root: PathBuf,
47}
48
49impl PosixChunkTransport {
50    pub fn new(root: PathBuf) -> Self {
51        Self { root }
52    }
53
54    fn chunk_path(&self, hash: &str) -> PathBuf {
55        let prefix = &hash[..std::cmp::min(2, hash.len())];
56        self.root.join(CAS_DIR).join(CHUNKS_DIR).join(prefix).join(hash)
57    }
58
59    fn tmp_dir(&self) -> PathBuf {
60        self.root.join(CAS_DIR).join(TMP_DIR)
61    }
62
63    fn manifest_path(&self) -> PathBuf {
64        self.root.join(CAS_DIR).join(MANIFEST_FILE)
65    }
66
67    fn tmp_name() -> String {
68        uuid::Uuid::new_v4().to_string()
69    }
70}
71
72#[async_trait]
73impl ChunkTransport for PosixChunkTransport {
74    async fn put_chunk(&self, hash: &str, data: &[u8]) -> Result<()> {
75        use std::io::Write;
76
77        let final_path = self.chunk_path(hash);
78        if final_path.exists() {
79            return Ok(());
80        }
81
82        if let Some(parent) = final_path.parent() {
83            std::fs::create_dir_all(parent)?;
84        }
85
86        let tmp_dir = self.tmp_dir();
87        std::fs::create_dir_all(&tmp_dir)?;
88
89        // Crash-safe: tmp -> fsync -> rename -> dir fsync
90        let tmp_path = tmp_dir.join(Self::tmp_name());
91        {
92            let mut f = std::fs::File::create(&tmp_path)?;
93            f.write_all(data)?;
94            f.sync_all()?;
95        }
96        std::fs::rename(&tmp_path, &final_path)?;
97        if let Some(parent) = final_path.parent()
98            && let Ok(dir) = std::fs::File::open(parent) {
99                let _ = dir.sync_all();
100            }
101
102        Ok(())
103    }
104
105    async fn get_chunk(&self, hash: &str) -> Result<Vec<u8>> {
106        let path = self.chunk_path(hash);
107        std::fs::read(&path).map_err(|e| {
108            if e.kind() == std::io::ErrorKind::NotFound {
109                crate::error::FxcpError::ChunkMissing(hash.to_string())
110            } else {
111                crate::error::FxcpError::Io(e)
112            }
113        })
114    }
115
116    async fn has_chunk(&self, hash: &str) -> Result<bool> {
117        Ok(self.chunk_path(hash).exists())
118    }
119
120    async fn put_manifest(&self, data: &[u8]) -> Result<()> {
121        use std::io::Write;
122
123        let final_path = self.manifest_path();
124        if let Some(parent) = final_path.parent() {
125            std::fs::create_dir_all(parent)?;
126        }
127
128        let tmp_dir = self.tmp_dir();
129        std::fs::create_dir_all(&tmp_dir)?;
130
131        // Crash-safe: tmp -> fsync -> rename -> dir fsync
132        let tmp_path = tmp_dir.join(Self::tmp_name());
133        {
134            let mut f = std::fs::File::create(&tmp_path)?;
135            f.write_all(data)?;
136            f.sync_all()?;
137        }
138        std::fs::rename(&tmp_path, &final_path)?;
139        if let Some(parent) = final_path.parent()
140            && let Ok(dir) = std::fs::File::open(parent) {
141                let _ = dir.sync_all();
142            }
143
144        Ok(())
145    }
146
147    async fn get_manifest(&self) -> Result<Option<Vec<u8>>> {
148        let path = self.manifest_path();
149        match std::fs::read(&path) {
150            Ok(data) => Ok(Some(data)),
151            Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(None),
152            Err(e) => Err(crate::error::FxcpError::Io(e)),
153        }
154    }
155
156    async fn health_check(&self) -> Result<()> {
157        let cas_dir = self.root.join(CAS_DIR);
158        std::fs::create_dir_all(&cas_dir)?;
159
160        let probe = cas_dir.join(format!(".health_probe_{}", Self::tmp_name()));
161        std::fs::write(&probe, b"ok").map_err(|e| {
162            crate::error::FxcpError::CasStoreError(format!("health check write failed: {}", e))
163        })?;
164        std::fs::remove_file(&probe).map_err(|e| {
165            crate::error::FxcpError::CasStoreError(format!("health check cleanup failed: {}", e))
166        })?;
167
168        Ok(())
169    }
170}
171
172#[cfg(feature = "cloud")]
173mod s3_transport {
174    use super::ChunkTransport;
175    use crate::error::{FxcpError, Result};
176    use async_trait::async_trait;
177    use object_store::{ObjectStore, PutPayload, path::Path as ObjPath};
178    use std::sync::Arc;
179
180    /// S3/MinIO [`ChunkTransport`] using the same 4-char shard layout
181    /// (`{prefix}/.foxing_cas/chunks/{hash[0..4]}/{hash}`) as
182    /// [`S3CasStore`](crate::s3_cas::S3CasStore).
183    pub struct S3ChunkTransport {
184        store: Arc<dyn ObjectStore>,
185        prefix: String,
186    }
187
188    impl S3ChunkTransport {
189        pub fn new(store: Arc<dyn ObjectStore>, prefix: String) -> Self {
190            Self { store, prefix }
191        }
192
193        fn chunk_path(&self, hash: &str) -> ObjPath {
194            let shard = &hash[..4.min(hash.len())];
195            ObjPath::from(format!(
196                "{}/.foxing_cas/chunks/{}/{}",
197                self.prefix, shard, hash
198            ))
199        }
200
201        fn manifest_path(&self) -> ObjPath {
202            ObjPath::from(format!(
203                "{}/.foxing_cas/transport_manifest.bin",
204                self.prefix
205            ))
206        }
207    }
208
209    #[async_trait]
210    impl ChunkTransport for S3ChunkTransport {
211        async fn put_chunk(&self, hash: &str, data: &[u8]) -> Result<()> {
212            let path = self.chunk_path(hash);
213            self.store
214                .put(&path, PutPayload::from(data.to_vec()))
215                .await
216                .map_err(|e| FxcpError::CasStoreError(e.to_string()))?;
217            Ok(())
218        }
219
220        async fn get_chunk(&self, hash: &str) -> Result<Vec<u8>> {
221            let path = self.chunk_path(hash);
222            let result = self.store.get(&path).await.map_err(|e| match e {
223                object_store::Error::NotFound { .. } => {
224                    FxcpError::ChunkMissing(hash.to_string())
225                }
226                other => FxcpError::CasStoreError(other.to_string()),
227            })?;
228            let bytes = result
229                .bytes()
230                .await
231                .map_err(|e| FxcpError::CasStoreError(e.to_string()))?;
232            Ok(bytes.to_vec())
233        }
234
235        async fn has_chunk(&self, hash: &str) -> Result<bool> {
236            let path = self.chunk_path(hash);
237            Ok(self.store.head(&path).await.is_ok())
238        }
239
240        async fn put_manifest(&self, data: &[u8]) -> Result<()> {
241            let path = self.manifest_path();
242            self.store
243                .put(&path, PutPayload::from(data.to_vec()))
244                .await
245                .map_err(|e| FxcpError::CasStoreError(e.to_string()))?;
246            Ok(())
247        }
248
249        async fn get_manifest(&self) -> Result<Option<Vec<u8>>> {
250            let path = self.manifest_path();
251            match self.store.get(&path).await {
252                Ok(result) => {
253                    let bytes = result
254                        .bytes()
255                        .await
256                        .map_err(|e| FxcpError::CasStoreError(e.to_string()))?;
257                    Ok(Some(bytes.to_vec()))
258                }
259                Err(object_store::Error::NotFound { .. }) => Ok(None),
260                Err(e) => Err(FxcpError::CasStoreError(e.to_string())),
261            }
262        }
263
264        async fn health_check(&self) -> Result<()> {
265            let probe_path = ObjPath::from(format!(
266                "{}/.foxing_cas/_health_probe",
267                self.prefix
268            ));
269            self.store
270                .put(&probe_path, PutPayload::from(b"ok".to_vec()))
271                .await
272                .map_err(|e| FxcpError::CasStoreError(e.to_string()))?;
273            self.store
274                .delete(&probe_path)
275                .await
276                .map_err(|e| FxcpError::CasStoreError(e.to_string()))?;
277            Ok(())
278        }
279    }
280}
281
282#[cfg(feature = "cloud")]
283pub use s3_transport::S3ChunkTransport;
284
285#[cfg(test)]
286mod tests {
287    #![allow(clippy::unwrap_used, clippy::expect_used, clippy::panic)]
288    use super::*;
289    use std::collections::HashMap;
290    use std::sync::Mutex;
291
292    struct InMemoryChunkTransport {
293        chunks: Mutex<HashMap<String, Vec<u8>>>,
294        manifest: Mutex<Option<Vec<u8>>>,
295    }
296
297    impl InMemoryChunkTransport {
298        fn new() -> Self {
299            Self {
300                chunks: Mutex::new(HashMap::new()),
301                manifest: Mutex::new(None),
302            }
303        }
304    }
305
306    #[async_trait::async_trait]
307    impl ChunkTransport for InMemoryChunkTransport {
308        async fn put_chunk(&self, hash: &str, data: &[u8]) -> Result<()> {
309            self.chunks.lock().unwrap().insert(hash.to_string(), data.to_vec());
310            Ok(())
311        }
312        async fn get_chunk(&self, hash: &str) -> Result<Vec<u8>> {
313            self.chunks.lock().unwrap().get(hash)
314                .cloned()
315                .ok_or_else(|| crate::error::FxcpError::ChunkMissing(hash.to_string()))
316        }
317        async fn has_chunk(&self, hash: &str) -> Result<bool> {
318            Ok(self.chunks.lock().unwrap().contains_key(hash))
319        }
320        async fn put_manifest(&self, data: &[u8]) -> Result<()> {
321            *self.manifest.lock().unwrap() = Some(data.to_vec());
322            Ok(())
323        }
324        async fn get_manifest(&self) -> Result<Option<Vec<u8>>> {
325            Ok(self.manifest.lock().unwrap().clone())
326        }
327        async fn health_check(&self) -> Result<()> {
328            Ok(())
329        }
330    }
331
332    #[tokio::test]
333    async fn test_put_get_round_trip() {
334        let t = InMemoryChunkTransport::new();
335        t.put_chunk("abc", b"hello world").await.unwrap();
336        let got = t.get_chunk("abc").await.unwrap();
337        assert_eq!(got, b"hello world");
338    }
339
340    #[tokio::test]
341    async fn test_get_missing_returns_chunk_missing_error() {
342        let t = InMemoryChunkTransport::new();
343        let err = t.get_chunk("nonexistent").await;
344        assert!(err.is_err());
345        match err.unwrap_err() {
346            crate::error::FxcpError::ChunkMissing(h) => assert_eq!(h, "nonexistent"),
347            other => panic!("expected ChunkMissing, got {:?}", other),
348        }
349    }
350
351    #[tokio::test]
352    async fn test_has_chunk_before_and_after_put() {
353        let t = InMemoryChunkTransport::new();
354        assert!(!t.has_chunk("xyz").await.unwrap());
355        t.put_chunk("xyz", b"data").await.unwrap();
356        assert!(t.has_chunk("xyz").await.unwrap());
357    }
358
359    #[tokio::test]
360    async fn test_manifest_round_trip() {
361        let t = InMemoryChunkTransport::new();
362        assert!(t.get_manifest().await.unwrap().is_none());
363        t.put_manifest(b"manifest data").await.unwrap();
364        let got = t.get_manifest().await.unwrap();
365        assert_eq!(got.as_deref(), Some(b"manifest data".as_slice()));
366    }
367
368    #[tokio::test]
369    async fn test_health_check() {
370        let t = InMemoryChunkTransport::new();
371        t.health_check().await.unwrap();
372    }
373
374    #[tokio::test]
375    async fn test_posix_put_get_roundtrip() {
376        let dir = tempfile::tempdir().unwrap();
377        let t = super::PosixChunkTransport::new(dir.path().to_path_buf());
378        t.put_chunk("abc123", b"hello world").await.unwrap();
379        let got = t.get_chunk("abc123").await.unwrap();
380        assert_eq!(got, b"hello world");
381    }
382
383    #[tokio::test]
384    async fn test_posix_get_missing_returns_chunk_missing() {
385        let dir = tempfile::tempdir().unwrap();
386        let t = super::PosixChunkTransport::new(dir.path().to_path_buf());
387        let err = t.get_chunk("nonexistent").await;
388        assert!(err.is_err());
389        match err.unwrap_err() {
390            crate::error::FxcpError::ChunkMissing(h) => assert_eq!(h, "nonexistent"),
391            other => panic!("expected ChunkMissing, got {:?}", other),
392        }
393    }
394
395    #[tokio::test]
396    async fn test_posix_has_chunk_before_and_after() {
397        let dir = tempfile::tempdir().unwrap();
398        let t = super::PosixChunkTransport::new(dir.path().to_path_buf());
399        assert!(!t.has_chunk("xyz789").await.unwrap());
400        t.put_chunk("xyz789", b"data").await.unwrap();
401        assert!(t.has_chunk("xyz789").await.unwrap());
402    }
403
404    #[tokio::test]
405    async fn test_posix_manifest_roundtrip() {
406        let dir = tempfile::tempdir().unwrap();
407        let t = super::PosixChunkTransport::new(dir.path().to_path_buf());
408        assert!(t.get_manifest().await.unwrap().is_none());
409        t.put_manifest(b"manifest payload").await.unwrap();
410        let got = t.get_manifest().await.unwrap();
411        assert_eq!(got.as_deref(), Some(b"manifest payload".as_slice()));
412    }
413
414    #[tokio::test]
415    async fn test_posix_health_check_writable() {
416        let dir = tempfile::tempdir().unwrap();
417        let t = super::PosixChunkTransport::new(dir.path().to_path_buf());
418        t.health_check().await.unwrap();
419    }
420
421    #[tokio::test]
422    async fn test_posix_no_tmp_files_after_put() {
423        let dir = tempfile::tempdir().unwrap();
424        let t = super::PosixChunkTransport::new(dir.path().to_path_buf());
425        t.put_chunk("aabbcc", b"some data").await.unwrap();
426
427        let tmp_dir = dir.path().join(".foxing_cas").join("chunks").join("tmp");
428        if tmp_dir.exists() {
429            let entries: Vec<_> = std::fs::read_dir(&tmp_dir).unwrap().collect();
430            assert!(entries.is_empty(), "tmp directory should be empty after put_chunk, found {} files", entries.len());
431        }
432    }
433
434    #[tokio::test]
435    async fn test_posix_shard_path_is_two_char() {
436        let dir = tempfile::tempdir().unwrap();
437        let t = super::PosixChunkTransport::new(dir.path().to_path_buf());
438        t.put_chunk("deadbeef0123", b"shard test").await.unwrap();
439
440        let expected_shard = dir.path().join(".foxing_cas").join("chunks").join("de");
441        assert!(expected_shard.is_dir(), "shard directory 'de' should exist");
442
443        let chunk_file = expected_shard.join("deadbeef0123");
444        assert!(chunk_file.exists(), "chunk file should be in 2-char shard directory");
445    }
446}
447
448#[cfg(all(test, feature = "cloud"))]
449mod s3_tests {
450    use super::*;
451    use object_store::memory::InMemory;
452    use object_store::{ObjectStore, PutPayload, path::Path as ObjPath};
453    use std::sync::Arc;
454
455    fn make_transport() -> S3ChunkTransport {
456        S3ChunkTransport::new(Arc::new(InMemory::new()), "test".to_string())
457    }
458
459    #[tokio::test]
460    async fn test_s3_put_get_roundtrip() {
461        let t = make_transport();
462        t.put_chunk("abc", b"hello").await.unwrap();
463        let got = t.get_chunk("abc").await.unwrap();
464        assert_eq!(got, b"hello");
465    }
466
467    #[tokio::test]
468    async fn test_s3_get_missing_returns_chunk_missing() {
469        let t = make_transport();
470        let err = t.get_chunk("nonexistent").await.unwrap_err();
471        match err {
472            crate::error::FxcpError::ChunkMissing(h) => assert_eq!(h, "nonexistent"),
473            other => panic!("expected ChunkMissing, got {other:?}"),
474        }
475    }
476
477    #[tokio::test]
478    async fn test_s3_has_chunk_true_false() {
479        let t = make_transport();
480        assert!(!t.has_chunk("xyz").await.unwrap());
481        t.put_chunk("xyz", b"data").await.unwrap();
482        assert!(t.has_chunk("xyz").await.unwrap());
483    }
484
485    #[tokio::test]
486    async fn test_s3_manifest_roundtrip() {
487        let t = make_transport();
488        assert!(t.get_manifest().await.unwrap().is_none());
489        t.put_manifest(b"manifest bytes").await.unwrap();
490        let got = t.get_manifest().await.unwrap();
491        assert_eq!(got.as_deref(), Some(b"manifest bytes".as_slice()));
492    }
493
494    #[tokio::test]
495    async fn test_s3_health_check() {
496        let t = make_transport();
497        t.health_check().await.unwrap();
498    }
499
500    #[tokio::test]
501    async fn test_s3_put_chunk_error_propagates() {
502        #[derive(Debug)]
503        struct FailingPutStore;
504
505        impl std::fmt::Display for FailingPutStore {
506            fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
507                write!(f, "FailingPutStore")
508            }
509        }
510
511        #[async_trait::async_trait]
512        impl ObjectStore for FailingPutStore {
513            async fn put_opts(
514                &self,
515                _location: &ObjPath,
516                _payload: PutPayload,
517                _opts: object_store::PutOptions,
518            ) -> object_store::Result<object_store::PutResult> {
519                Err(object_store::Error::Generic {
520                    store: "FailingPutStore",
521                    source: "simulated failure".into(),
522                })
523            }
524
525            async fn put_multipart_opts(
526                &self,
527                _location: &ObjPath,
528                _opts: object_store::PutMultipartOpts,
529            ) -> object_store::Result<Box<dyn object_store::MultipartUpload>> {
530                Err(object_store::Error::Generic {
531                    store: "FailingPutStore",
532                    source: "simulated failure".into(),
533                })
534            }
535
536            async fn get_opts(
537                &self,
538                _location: &ObjPath,
539                _options: object_store::GetOptions,
540            ) -> object_store::Result<object_store::GetResult> {
541                Err(object_store::Error::Generic {
542                    store: "FailingPutStore",
543                    source: "simulated failure".into(),
544                })
545            }
546
547            async fn delete(&self, _location: &ObjPath) -> object_store::Result<()> {
548                Ok(())
549            }
550
551            fn list(
552                &self,
553                _prefix: Option<&ObjPath>,
554            ) -> futures::stream::BoxStream<'_, object_store::Result<object_store::ObjectMeta>>
555            {
556                futures::stream::empty().boxed()
557            }
558
559            async fn list_with_delimiter(
560                &self,
561                _prefix: Option<&ObjPath>,
562            ) -> object_store::Result<object_store::ListResult> {
563                Ok(object_store::ListResult {
564                    common_prefixes: vec![],
565                    objects: vec![],
566                })
567            }
568
569            async fn copy(
570                &self,
571                _from: &ObjPath,
572                _to: &ObjPath,
573            ) -> object_store::Result<()> {
574                Ok(())
575            }
576
577            async fn copy_if_not_exists(
578                &self,
579                _from: &ObjPath,
580                _to: &ObjPath,
581            ) -> object_store::Result<()> {
582                Ok(())
583            }
584        }
585
586        use futures::StreamExt;
587
588        let t = S3ChunkTransport::new(Arc::new(FailingPutStore), "test".to_string());
589        let err = t.put_chunk("abc", b"data").await;
590        assert!(err.is_err());
591        match err.unwrap_err() {
592            crate::error::FxcpError::CasStoreError(_) => {}
593            other => panic!("expected CasStoreError, got {other:?}"),
594        }
595    }
596
597    #[tokio::test]
598    async fn test_s3_chunk_path_uses_four_char_shard() {
599        let store: Arc<dyn ObjectStore> = Arc::new(InMemory::new());
600        let t = S3ChunkTransport::new(Arc::clone(&store), "repo".to_string());
601
602        t.put_chunk("abcdef0123456789", b"payload").await.unwrap();
603
604        let expected = ObjPath::from("repo/.foxing_cas/chunks/abcd/abcdef0123456789");
605        let meta = store.head(&expected).await;
606        assert!(meta.is_ok(), "chunk not found at expected 4-char shard path");
607    }
608}
fxcp_core/chunk_transport.rs

fxcp_core/
chunk_transport.rs
