coven_foundation/
store_dir.rs

1use std::ops::Deref;
2use std::path::{Path, PathBuf};
3use tracing::debug;
4
5use crate::atomic_file::FileError;
6
7/// Why a string is not a safe path token.
8///
9/// An untrusted string becomes a path component in several places: a blob's
10/// `id`/`namespace` (interpolated into its on-disk file path and cloud object
11/// key), and a `store_id`/`sid` from an untrusted device invitation or restore code (the
12/// name of a directory under `stores/`). All arrive from outside — an incoming
13/// changeset authored by any write-capable member, or a pasted code anyone can
14/// craft — so an unconstrained one could climb out of the directory it is joined
15/// onto (`..`, a path separator, an absolute leading slash) and make a pulling or
16/// joining device read, write, or recursively delete an arbitrary location, or —
17/// too short / not aligned to a char boundary — crash a blob's partition-prefix
18/// slice. A string that trips any of these is bad data, refused before a path is
19/// built or used.
20#[derive(Debug, Clone, PartialEq, Eq)]
21pub enum PathTokenError {
22    /// The token is empty — no file name to write, no key to form.
23    Empty,
24    /// The token contains a path separator (`/` or `\`), so joining it onto a
25    /// directory would descend into (or, with a leading separator, replace) the
26    /// path rather than name a single child.
27    Separator,
28    /// The token is exactly `..`, which names the parent of the directory it is
29    /// joined onto rather than a child. A trailing `..` component is normalized
30    /// away when the path is resolved, so the join lands on the parent.
31    ParentDir,
32    /// The token is exactly `.`, which names the directory it is joined onto
33    /// itself rather than a child. Like `..`, a trailing `.` component is
34    /// normalized away, so `stores/.` resolves to `stores`'s parent (the
35    /// data dir) — an escape just as `..` is.
36    CurDir,
37    /// The token contains a NUL byte, which truncates the path at the OS boundary.
38    NulByte,
39    /// The token contains a `:`, which on Windows names an alternate data stream
40    /// (`file:stream`) or a drive-relative reference (`c:dir`) rather than a child.
41    Colon,
42    /// The dash-stripped id is too short, or splits a multi-byte char, to take the
43    /// two leading byte-pairs the `{ab}/{cd}` partition prefix needs.
44    Unindexable,
45}
46
47#[derive(Debug, thiserror::Error)]
48pub enum RequiredLocalBlobPathError {
49    #[error("local blob path: {0}")]
50    Path(#[from] PathTokenError),
51    #[error("local blob {namespace}/{id} is absent")]
52    Missing { namespace: String, id: String },
53    #[error("local blob file: {0}")]
54    File(#[from] FileError),
55}
56
57#[derive(Debug, thiserror::Error)]
58pub enum CachedLocatorRemovalError {
59    #[error("blob cache path: {0}")]
60    Path(#[from] PathTokenError),
61    #[error("blob cache file: {0}")]
62    File(#[from] FileError),
63}
64
65#[derive(Debug, thiserror::Error)]
66pub enum LocalBlobRemovalError {
67    #[error("local blob path: {0}")]
68    Path(#[from] PathTokenError),
69    #[error("local blob file: {0}")]
70    File(#[from] FileError),
71}
72
73#[derive(Debug, thiserror::Error)]
74pub enum LocalBlobStoreError {
75    #[error("local blob path: {0}")]
76    Path(#[from] PathTokenError),
77    #[error("local blob file: {0}")]
78    File(#[from] FileError),
79    #[error("local blob {} has {actual_size} bytes, expected {expected_size}", path.display())]
80    SizeMismatch {
81        path: PathBuf,
82        expected_size: u64,
83        actual_size: u64,
84    },
85}
86
87#[derive(Debug, thiserror::Error)]
88pub enum StoreBlobFileError {
89    #[error("store blob path: {0}")]
90    Path(#[from] PathTokenError),
91    #[error("store blob file: {0}")]
92    File(#[from] FileError),
93    #[error("commit store blob: {0}")]
94    Commit(#[from] crate::local_file::CommitNewFileError),
95    #[error("store blob {} has size/hash {actual_size}/{actual_hash}, expected {expected_size}/{expected_hash}", path.display())]
96    Integrity {
97        path: PathBuf,
98        expected_size: u64,
99        actual_size: u64,
100        expected_hash: crate::object_hash::ObjectHash,
101        actual_hash: crate::object_hash::ObjectHash,
102    },
103}
104
105pub struct CachedBlobFile {
106    path: PathBuf,
107    recency: u64,
108    size: u64,
109}
110
111impl CachedBlobFile {
112    pub fn path(&self) -> &Path {
113        &self.path
114    }
115
116    pub fn recency(&self) -> u64 {
117        self.recency
118    }
119
120    pub fn size(&self) -> u64 {
121        self.size
122    }
123}
124
125impl std::fmt::Display for PathTokenError {
126    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
127        match self {
128            PathTokenError::Empty => write!(f, "path token is empty"),
129            PathTokenError::Separator => write!(f, "path token contains a path separator"),
130            PathTokenError::ParentDir => write!(f, "path token contains a parent reference"),
131            PathTokenError::CurDir => write!(f, "path token is a current-directory reference"),
132            PathTokenError::NulByte => write!(f, "path token contains a NUL byte"),
133            PathTokenError::Colon => write!(f, "path token contains a colon"),
134            PathTokenError::Unindexable => {
135                write!(
136                    f,
137                    "id is too short or misaligned to form a partition prefix"
138                )
139            }
140        }
141    }
142}
143
144impl std::error::Error for PathTokenError {}
145
146/// Reject a single untrusted path token (a blob `id`/`namespace`, or a
147/// `store_id`/`sid`) that could escape the directory it is joined onto. A safe
148/// token names exactly one child: no separator, no `..`, no `.`, no NUL, no `:` (a
149/// Windows stream/drive reference), non-empty. The single gate every path builder
150/// and every code decoder runs an untrusted token through, so traversal is refused
151/// before any on-disk or cloud path is formed — and a decoded id is a safe single
152/// component by the time any consumer joins it onto a directory.
153///
154/// Both `.` and `..` are refused: each is a directory-relative reference that a
155/// trailing path component normalizes away, so joining either onto `dir` resolves
156/// to `dir` itself or its parent rather than to a child of `dir`.
157pub fn validate_path_token(token: &str) -> Result<(), PathTokenError> {
158    if token.is_empty() {
159        return Err(PathTokenError::Empty);
160    }
161    if token.contains('\0') {
162        return Err(PathTokenError::NulByte);
163    }
164    if token.contains('/') || token.contains('\\') {
165        return Err(PathTokenError::Separator);
166    }
167    if token.contains(':') {
168        return Err(PathTokenError::Colon);
169    }
170    if token == ".." {
171        return Err(PathTokenError::ParentDir);
172    }
173    if token == "." {
174        return Err(PathTokenError::CurDir);
175    }
176    Ok(())
177}
178
179/// Reject an untrusted `cloud_path` (the consumer's readable object key under the
180/// plain scheme, e.g. `"Artist - Album/cover.jpg"`) that could escape its
181/// namespace prefix in the bucket. Unlike a path token, an interior `/` is
182/// legitimate — the readable path is nested — but every segment still has to be
183/// a canonical path token. Empty, `.`, `..`, colon/platform-prefix, backslash,
184/// and NUL forms are refused before an object key is built. The `cloud_path`
185/// never feeds a local file path, only the cloud object key, so this guards the
186/// keyspace, not the disk.
187pub fn validate_cloud_path(cloud_path: &str) -> Result<(), PathTokenError> {
188    if cloud_path.starts_with('/') {
189        return Err(PathTokenError::Separator);
190    }
191    for segment in cloud_path.split('/') {
192        validate_path_token(segment)?;
193    }
194    Ok(())
195}
196
197/// Default name of the parent directory a store lives under — overridden
198/// per host via [`StoreLayout::stores_dirname`].
199const DEFAULT_STORES_DIRNAME: &str = "stores";
200/// The name of a store's own database file.
201const DB_FILENAME: &str = "store.db";
202
203/// The host's on-disk layout for stores: which directory they live under.
204/// One rule shared by create, open, join, and restore, so a host that wants
205/// `libraries/<id>` instead of coven's default `stores/<id>` names it once
206/// here rather than each flow hardwiring (or working around) coven's own
207/// choice.
208#[derive(Clone, Debug)]
209pub struct StoreLayout {
210    app_dir: PathBuf,
211    stores_dirname: String,
212}
213
214impl StoreLayout {
215    pub fn new(app_dir: impl Into<PathBuf>) -> Self {
216        Self {
217            app_dir: app_dir.into(),
218            stores_dirname: DEFAULT_STORES_DIRNAME.to_string(),
219        }
220    }
221
222    pub fn stores_dirname(mut self, name: impl Into<String>) -> Self {
223        self.stores_dirname = name.into();
224        self
225    }
226
227    /// The stores parent dir (for host listing/discovery).
228    pub fn stores_root(&self) -> PathBuf {
229        self.app_dir.join(&self.stores_dirname)
230    }
231
232    pub fn pending_device_pairings_dir(&self) -> PathBuf {
233        self.app_dir.join("pending-device-pairings")
234    }
235
236    pub fn pending_device_pairing_path(&self, session_id: &str) -> Result<PathBuf, PathTokenError> {
237        validate_path_token(session_id)?;
238        Ok(self
239            .pending_device_pairings_dir()
240            .join(format!("{session_id}.json")))
241    }
242
243    /// Read every committed device-pairing journal entry. Atomic-write stages
244    /// are unpublished files and therefore never enter the durable record set.
245    pub fn pending_device_pairing_journals(&self) -> Result<Vec<(PathBuf, Vec<u8>)>, FileError> {
246        let directory = self.pending_device_pairings_dir();
247        let entries = match std::fs::read_dir(&directory) {
248            Ok(entries) => entries,
249            Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(Vec::new()),
250            Err(source) => {
251                return Err(FileError::at(
252                    "read pending device pairings",
253                    directory,
254                    source,
255                ))
256            }
257        };
258        let mut journals = Vec::new();
259        for entry in entries {
260            let entry = entry.map_err(|source| {
261                FileError::at(
262                    "read pending device pairing directory entry",
263                    &directory,
264                    source,
265                )
266            })?;
267            let path = entry.path();
268            if crate::atomic_file::is_atomic_staging_file_name(&entry.file_name()) {
269                debug!(path = %path.display(), "ignoring unpublished device-pairing journal stage");
270                continue;
271            }
272            if !entry
273                .file_type()
274                .map_err(|source| {
275                    FileError::at("read pending device pairing file type", &path, source)
276                })?
277                .is_file()
278            {
279                return Err(FileError::NotFile {
280                    subject: "pending device pairing",
281                    path,
282                });
283            }
284            let bytes = std::fs::read(&path)
285                .map_err(|source| FileError::at("read pending device pairing", &path, source))?;
286            journals.push((path, bytes));
287        }
288        Ok(journals)
289    }
290
291    /// The one `(app_dir, store_id) -> StoreDir` rule, named with this
292    /// layout's directory. Callers validate an untrusted `store_id`
293    /// ([`validate_path_token`]) BEFORE calling, as every
294    /// join/restore/create flow already does.
295    pub fn store_dir(&self, store_id: &str) -> StoreDir {
296        StoreDir {
297            path: self.stores_root().join(store_id),
298            file_sync: crate::atomic_file::FileSync::Enabled,
299        }
300    }
301}
302
303/// Typed wrapper for a store directory path.
304///
305/// Centralizes the on-disk layout so callers use methods instead of
306/// ad-hoc `path.join("images")` etc.
307#[derive(Clone, Debug)]
308pub struct StoreDir {
309    path: PathBuf,
310    file_sync: crate::atomic_file::FileSync,
311}
312
313impl PartialEq for StoreDir {
314    fn eq(&self, other: &Self) -> bool {
315        self.path == other.path
316    }
317}
318
319impl StoreDir {
320    pub fn new(path: impl Into<PathBuf>) -> Self {
321        Self {
322            path: path.into(),
323            file_sync: crate::atomic_file::FileSync::Enabled,
324        }
325    }
326
327    /// A store directory whose owning database is itself ephemeral. Atomic
328    /// visibility and rollback still run, but persistence barriers do not: no
329    /// file can outlive the durable state that names it.
330    pub fn new_ephemeral(path: impl Into<PathBuf>) -> Self {
331        Self {
332            path: path.into(),
333            file_sync: crate::atomic_file::FileSync::Disabled,
334        }
335    }
336
337    #[cfg(any(test, feature = "test-utils"))]
338    pub fn new_with_file_sync_observer_for_test(
339        path: impl Into<PathBuf>,
340    ) -> (Self, std::sync::Arc<std::sync::atomic::AtomicUsize>) {
341        let requests = std::sync::Arc::new(std::sync::atomic::AtomicUsize::new(0));
342        (
343            Self {
344                path: path.into(),
345                file_sync: crate::atomic_file::FileSync::ObservedDisabled(requests.clone()),
346            },
347            requests,
348        )
349    }
350
351    pub async fn stage_atomic_file(
352        &self,
353        destination: &Path,
354    ) -> Result<crate::local_file::AtomicStagedFile, FileError> {
355        crate::local_file::AtomicStagedFile::create_with_file_sync(
356            destination,
357            self.file_sync.clone(),
358        )
359        .await
360    }
361
362    pub fn create_payload_spool_stage(
363        &self,
364    ) -> Result<crate::atomic_file::AtomicFileStage, std::io::Error> {
365        crate::atomic_file::AtomicFileStage::create_in_with_file_sync(
366            &self.payload_spool_dir(),
367            self.file_sync.clone(),
368        )
369    }
370
371    pub async fn sync_parent_dir(&self, path: &Path) -> Result<(), FileError> {
372        self.file_sync.sync_parent(path).await
373    }
374
375    pub fn sync_parent_dir_blocking(&self, path: &Path) -> Result<(), FileError> {
376        self.file_sync.sync_parent_blocking(path)
377    }
378
379    pub fn db_path(&self) -> PathBuf {
380        self.path.join(DB_FILENAME)
381    }
382
383    pub fn config_path(&self) -> PathBuf {
384        self.path.join("config.yaml")
385    }
386
387    pub fn device_pairing_journal_path(&self) -> PathBuf {
388        self.path.join("device-pairing.json")
389    }
390
391    /// The two-level partition shard for `id`: `{ab}/{cd}/{id}`, where `{ab}`/`{cd}`
392    /// are the first two byte-pairs of the dash-stripped id. Cached and pinned
393    /// blobs use this shard with their locator hash as the id.
394    ///
395    /// `id` is validated as a single path token and must be long enough (and
396    /// char-boundary aligned) to take the two leading byte-pairs. An id that fails
397    /// is bad data — it could escape the directory or crash the slice — so this
398    /// returns [`PathTokenError`] rather than interpolating it or panicking; the
399    /// caller refuses the blob.
400    pub(crate) fn id_shard(id: &str) -> Result<String, PathTokenError> {
401        validate_path_token(id)?;
402        let hex = id.replace('-', "");
403        if !(hex.is_char_boundary(2) && hex.is_char_boundary(4)) {
404            return Err(PathTokenError::Unindexable);
405        }
406        Ok(format!("{}/{}/{id}", &hex[..2], &hex[2..4]))
407    }
408
409    pub fn storage_dir(&self) -> PathBuf {
410        self.path.join("storage")
411    }
412
413    /// Immutable stored bytes prepared for one blob locator. The locator hash is
414    /// the file name, so retries reopen the same exact spool rather than sealing
415    /// the plaintext again with fresh randomness.
416    pub fn outbound_blob_spool_path(
417        &self,
418        locator_hash: crate::object_hash::ObjectHash,
419    ) -> PathBuf {
420        self.storage_dir()
421            .join("outbound-blobs")
422            .join(locator_hash.to_string())
423    }
424
425    /// The directory holding every internal payload file.
426    pub fn payload_spool_dir(&self) -> PathBuf {
427        self.path.join("spool").join("payloads")
428    }
429
430    /// The file holding one internal payload — bytes a database row owns,
431    /// stored beside the database rather than inside it. The file is named for
432    /// the digest of the bytes it holds, so a retry of a failed insert rewrites
433    /// the same path with the same contents. Unlike a blob, a payload is never
434    /// leased, packaged for an audience, or evicted: it is deleted by the flow
435    /// that deletes the row referencing it.
436    pub fn payload_spool_path(&self, payload_hash: crate::object_hash::ObjectHash) -> PathBuf {
437        self.payload_spool_dir().join(payload_hash.to_string())
438    }
439
440    pub async fn remove_outbound_blob_spool(
441        &self,
442        locator_hash: crate::object_hash::ObjectHash,
443    ) -> Result<(), FileError> {
444        let path = self.outbound_blob_spool_path(locator_hash);
445        match tokio::fs::remove_file(&path).await {
446            Ok(()) => self.sync_parent_dir(&path).await,
447            Err(error) if error.kind() == std::io::ErrorKind::NotFound => Ok(()),
448            Err(source) => Err(FileError::at("remove exact blob spool", path, source)),
449        }
450    }
451
452    /// A kept (budget-exempt) cache copy of a **Remote** blob:
453    /// `storage/pinned/<namespace>/{ab}/{cd}/<locator-hash>`. The kept sibling of
454    /// [`Self::cache_blob_path`] — same per-namespace shard layout, in the `pinned`
455    /// folder instead of `cache`. The cache's truth is the folder a blob's file lives
456    /// in, not a table; a file here is a Remote blob's cache copy the user pinned for
457    /// offline (kept from eviction). `Err` if `namespace` is unsafe.
458    pub fn pinned_blob_path(
459        &self,
460        namespace: &str,
461        locator_hash: crate::object_hash::ObjectHash,
462    ) -> Result<PathBuf, PathTokenError> {
463        self.cache_folder_blob_path("pinned", namespace, &locator_hash.to_string())
464    }
465
466    pub async fn populate_pinned_blob_from_file(
467        &self,
468        namespace: &str,
469        locator_hash: crate::object_hash::ObjectHash,
470        expected_size: u64,
471        expected_hash: crate::object_hash::ObjectHash,
472        source: &Path,
473    ) -> Result<(), StoreBlobFileError> {
474        let destination = self
475            .pinned_blob_path(namespace, locator_hash)
476            .map_err(StoreBlobFileError::Path)?;
477        self.populate_exact_blob_from_file(destination, expected_size, expected_hash, source)
478            .await
479    }
480
481    pub async fn populate_cached_blob_from_file(
482        &self,
483        namespace: &str,
484        locator_hash: crate::object_hash::ObjectHash,
485        expected_size: u64,
486        expected_hash: crate::object_hash::ObjectHash,
487        source: &Path,
488    ) -> Result<PathBuf, StoreBlobFileError> {
489        let destination = self
490            .cache_blob_path(namespace, locator_hash)
491            .map_err(StoreBlobFileError::Path)?;
492        self.populate_exact_blob_from_file(
493            destination.clone(),
494            expected_size,
495            expected_hash,
496            source,
497        )
498        .await?;
499        Ok(destination)
500    }
501
502    async fn populate_exact_blob_from_file(
503        &self,
504        destination: PathBuf,
505        expected_size: u64,
506        expected_hash: crate::object_hash::ObjectHash,
507        source: &Path,
508    ) -> Result<(), StoreBlobFileError> {
509        let staged = self
510            .stage_atomic_file(&destination)
511            .await
512            .map_err(StoreBlobFileError::File)?;
513        let (staged, actual_size, actual_digest) = staged
514            .copy_from(source)
515            .await
516            .map_err(StoreBlobFileError::File)?;
517        let actual_hash = crate::object_hash::ObjectHash::from_digest(actual_digest);
518        if actual_size != expected_size || actual_hash != expected_hash {
519            return Err(StoreBlobFileError::Integrity {
520                path: source.to_path_buf(),
521                expected_size,
522                actual_size,
523                expected_hash,
524                actual_hash,
525            });
526        }
527        match staged.commit_new().await {
528            Ok(()) => Ok(()),
529            Err(crate::local_file::CommitNewFileError::DestinationExists(path)) => {
530                let (actual_size, actual_hash) = exact_file_facts(&path)
531                    .await
532                    .map_err(StoreBlobFileError::File)?;
533                if actual_size == expected_size && actual_hash == expected_hash {
534                    Ok(())
535                } else {
536                    Err(StoreBlobFileError::Integrity {
537                        path,
538                        expected_size,
539                        actual_size,
540                        expected_hash,
541                        actual_hash,
542                    })
543                }
544            }
545            Err(error) => Err(StoreBlobFileError::Commit(error)),
546        }
547    }
548
549    pub async fn pinned_blob_is_exact(
550        &self,
551        namespace: &str,
552        locator_hash: crate::object_hash::ObjectHash,
553        expected_size: u64,
554        expected_hash: crate::object_hash::ObjectHash,
555    ) -> Result<bool, StoreBlobFileError> {
556        let path = self
557            .pinned_blob_path(namespace, locator_hash)
558            .map_err(StoreBlobFileError::Path)?;
559        match file_exists(&path).await {
560            Ok(false) => Ok(false),
561            Err(error) => Err(StoreBlobFileError::File(error)),
562            Ok(true) => {
563                let (actual_size, actual_hash) = exact_file_facts(&path)
564                    .await
565                    .map_err(StoreBlobFileError::File)?;
566                if actual_size == expected_size && actual_hash == expected_hash {
567                    Ok(true)
568                } else {
569                    Err(StoreBlobFileError::Integrity {
570                        path,
571                        expected_size,
572                        actual_size,
573                        expected_hash,
574                        actual_hash,
575                    })
576                }
577            }
578        }
579    }
580
581    pub async fn remote_blob_is_exact(
582        &self,
583        namespace: &str,
584        locator_hash: crate::object_hash::ObjectHash,
585        expected_size: u64,
586        expected_hash: crate::object_hash::ObjectHash,
587    ) -> Result<bool, StoreBlobFileError> {
588        for path in [
589            self.pinned_blob_path(namespace, locator_hash)?,
590            self.cache_blob_path(namespace, locator_hash)?,
591        ] {
592            if file_is_exact(&path, expected_size, expected_hash).await? {
593                return Ok(true);
594            }
595        }
596        Ok(false)
597    }
598
599    pub async fn cached_blob_is_exact(
600        &self,
601        namespace: &str,
602        locator_hash: crate::object_hash::ObjectHash,
603        expected_size: u64,
604        expected_hash: crate::object_hash::ObjectHash,
605    ) -> Result<bool, StoreBlobFileError> {
606        let path = self.cache_blob_path(namespace, locator_hash)?;
607        file_is_exact(&path, expected_size, expected_hash).await
608    }
609
610    /// An opportunistic (evictable) cache copy of a **Remote** blob:
611    /// `storage/cache/<namespace>/{ab}/{cd}/<locator-hash>`. A file here is a cached-but-unpinned
612    /// blob — fetched on read or eagerly on pull, droppable by the budget sweep. The
613    /// folder it lives in, not a table, is what makes it evictable rather than kept.
614    /// Segmented by `namespace` so each namespace's budget evicts only its own
615    /// subtree, `storage/cache/<namespace>`. `Err` if `namespace` is unsafe.
616    pub fn cache_blob_path(
617        &self,
618        namespace: &str,
619        locator_hash: crate::object_hash::ObjectHash,
620    ) -> Result<PathBuf, PathTokenError> {
621        self.cache_folder_blob_path("cache", namespace, &locator_hash.to_string())
622    }
623
624    pub fn remote_blob_paths(
625        &self,
626        namespace: &str,
627        locator_hash: crate::object_hash::ObjectHash,
628    ) -> Result<(PathBuf, PathBuf), PathTokenError> {
629        Ok((
630            self.pinned_blob_path(namespace, locator_hash)?,
631            self.cache_blob_path(namespace, locator_hash)?,
632        ))
633    }
634
635    pub async fn remove_cached_locator(
636        &self,
637        namespace: &str,
638        locator_hash: crate::object_hash::ObjectHash,
639    ) -> Result<(), CachedLocatorRemovalError> {
640        for path in [
641            self.pinned_blob_path(namespace, locator_hash)
642                .map_err(CachedLocatorRemovalError::Path)?,
643            self.cache_blob_path(namespace, locator_hash)
644                .map_err(CachedLocatorRemovalError::Path)?,
645        ] {
646            remove_file(&path)
647                .await
648                .map_err(CachedLocatorRemovalError::File)?;
649        }
650        Ok(())
651    }
652
653    /// `storage/<folder>/<namespace>/{ab}/{cd}/<locator-hash>` — the single blob-path builder
654    /// behind [`Self::cache_blob_path`] (`folder` = `cache`) and
655    /// [`Self::pinned_blob_path`] (`folder` = `pinned`), which differ only by the
656    /// folder token. Composes the per-namespace dir
657    /// ([`Self::cache_folder_namespace_dir`]) with the locator-hash shard, so the layout lives
658    /// in one place. `namespace` and the locator hash are validated.
659    fn cache_folder_blob_path(
660        &self,
661        folder: &str,
662        namespace: &str,
663        id: &str,
664    ) -> Result<PathBuf, PathTokenError> {
665        Ok(self
666            .cache_folder_namespace_dir(folder, namespace)?
667            .join(Self::id_shard(id)?))
668    }
669
670    /// `storage/<folder>/<namespace>` for a cache folder (`cache` evictable / `pinned`
671    /// kept), `namespace` validated as a single path token. The per-namespace dir both
672    /// cache folders compose onto; [`Self::cache_namespace_dir`] is the evictable case
673    /// the budget sweep walks.
674    fn cache_folder_namespace_dir(
675        &self,
676        folder: &str,
677        namespace: &str,
678    ) -> Result<PathBuf, PathTokenError> {
679        validate_path_token(namespace)?;
680        Ok(self.storage_dir().join(folder).join(namespace))
681    }
682
683    /// coven's own copy of a **host-provided Local** blob:
684    /// `storage/local/<namespace>/<id>`. This is NOT a cache copy — it is the blob's
685    /// home while its release is Local (a host-provided blob has no user path). It
686    /// is never evicted: the budget sweep walks only [`Self::cache_dir`], never
687    /// `storage/local`. Both `namespace` and `id` are validated as single path
688    /// tokens (the blob columns come from a row any write-capable member authored),
689    /// so neither can escape the store. `Err` if either is unsafe.
690    pub fn local_blob_path(&self, namespace: &str, id: &str) -> Result<PathBuf, PathTokenError> {
691        validate_path_token(namespace)?;
692        validate_path_token(id)?;
693        Ok(self
694            .path
695            .join("storage")
696            .join("local")
697            .join(namespace)
698            .join(id))
699    }
700
701    pub async fn require_local_blob_path(
702        &self,
703        namespace: &str,
704        id: &str,
705    ) -> Result<PathBuf, RequiredLocalBlobPathError> {
706        let path = self
707            .local_blob_path(namespace, id)
708            .map_err(RequiredLocalBlobPathError::Path)?;
709        match file_exists(&path).await {
710            Ok(true) => Ok(path),
711            Ok(false) => Err(RequiredLocalBlobPathError::Missing {
712                namespace: namespace.to_string(),
713                id: id.to_string(),
714            }),
715            Err(error) => Err(RequiredLocalBlobPathError::File(error)),
716        }
717    }
718
719    pub async fn local_blob_path_if_present(
720        &self,
721        namespace: &str,
722        id: &str,
723        expected_size: u64,
724    ) -> Result<Option<PathBuf>, LocalBlobStoreError> {
725        let path = self.local_blob_path(namespace, id)?;
726        if !file_exists(&path)
727            .await
728            .map_err(LocalBlobStoreError::File)?
729        {
730            return Ok(None);
731        }
732        let actual_size = tokio::fs::metadata(&path)
733            .await
734            .map_err(|source| {
735                LocalBlobStoreError::File(FileError::at("stat local blob", &path, source))
736            })?
737            .len();
738        if actual_size != expected_size {
739            return Err(LocalBlobStoreError::SizeMismatch {
740                path,
741                expected_size,
742                actual_size,
743            });
744        }
745        Ok(Some(path))
746    }
747
748    pub async fn remove_local_blob(
749        &self,
750        namespace: &str,
751        id: &str,
752    ) -> Result<bool, LocalBlobRemovalError> {
753        let path = self
754            .local_blob_path(namespace, id)
755            .map_err(LocalBlobRemovalError::Path)?;
756        remove_file(&path)
757            .await
758            .map_err(LocalBlobRemovalError::File)
759    }
760
761    /// The evictable-cache root, `storage/cache`, holding every namespace's subtree.
762    /// The per-namespace budget sweep walks only one namespace's subtree under it,
763    /// `storage/cache/<namespace>`.
764    pub fn cache_dir(&self) -> PathBuf {
765        self.storage_dir().join("cache")
766    }
767
768    /// One namespace's evictable-cache subtree, `storage/cache/<namespace>`. The
769    /// cache budget enforcement walks only this tree, so
770    /// a namespace evicts against its own budget without touching another namespace's
771    /// files. `namespace` is validated as a single path token; `Err` if it is unsafe.
772    fn cache_namespace_dir(&self, namespace: &str) -> Result<PathBuf, PathTokenError> {
773        self.cache_folder_namespace_dir("cache", namespace)
774    }
775
776    pub async fn cached_blob_files(
777        &self,
778        namespace: &str,
779    ) -> Result<Vec<CachedBlobFile>, StoreBlobFileError> {
780        let directory = self
781            .cache_namespace_dir(namespace)
782            .map_err(StoreBlobFileError::Path)?;
783        walk_files(&directory)
784            .await
785            .map_err(StoreBlobFileError::File)
786            .map(|files| {
787                files
788                    .into_iter()
789                    .map(|(path, recency, size)| CachedBlobFile {
790                        path,
791                        recency,
792                        size,
793                    })
794                    .collect()
795            })
796    }
797
798    pub async fn remove_cached_blob_file(
799        &self,
800        file: &CachedBlobFile,
801    ) -> Result<bool, StoreBlobFileError> {
802        remove_file(file.path())
803            .await
804            .map_err(StoreBlobFileError::File)
805    }
806
807    /// Remove in-progress write temporaries left by an earlier process — blob
808    /// files and payload-spool files alike. Files created at or after
809    /// `process_start` belong to the current process and are left untouched.
810    pub fn remove_orphaned_write_temps(
811        &self,
812        process_start: std::time::SystemTime,
813    ) -> std::io::Result<()> {
814        let storage = self.storage_dir();
815        for directory in [
816            storage.join("local"),
817            storage.join("cache"),
818            storage.join("pinned"),
819            self.payload_spool_dir(),
820        ] {
821            self.remove_orphaned_temps_in_dir(&directory, process_start)?;
822        }
823        Ok(())
824    }
825
826    fn remove_orphaned_temps_in_dir(
827        &self,
828        dir: &Path,
829        process_start: std::time::SystemTime,
830    ) -> std::io::Result<()> {
831        let entries = match std::fs::read_dir(dir) {
832            Ok(entries) => entries,
833            Err(error) if error.kind() == std::io::ErrorKind::NotFound => {
834                debug!(
835                    path = %dir.display(),
836                    store_dir = %self.display(),
837                    "blob directory absent during orphaned temp cleanup"
838                );
839                return Ok(());
840            }
841            Err(error) => return Err(error),
842        };
843        for entry in entries {
844            let entry = entry?;
845            let path = entry.path();
846            let file_type = entry.file_type()?;
847            if file_type.is_dir() {
848                self.remove_orphaned_temps_in_dir(&path, process_start)?;
849            } else if file_type.is_file()
850                && crate::local_file::AtomicStagedFile::is_staging_path(&path)
851            {
852                let modified = entry.metadata()?.modified()?;
853                if modified >= process_start {
854                    debug!(
855                        path = %path.display(),
856                        "leaving fresh blob temp created at or after process start"
857                    );
858                    continue;
859                }
860                match std::fs::remove_file(&path) {
861                    Ok(()) => {}
862                    Err(error) if error.kind() == std::io::ErrorKind::NotFound => {
863                        debug!(
864                            path = %path.display(),
865                            "file already absent during local blob cleanup"
866                        );
867                    }
868                    Err(error) => return Err(error),
869                }
870            } else if file_type.is_file()
871                && path.file_name().and_then(|name| name.to_str()).is_none()
872            {
873                debug!(
874                    path = %path.display(),
875                    "skipping blob path with non-utf8 file name during orphaned temp cleanup"
876                );
877            }
878        }
879        Ok(())
880    }
881
882    /// Create the store directory tree if it is absent.
883    pub fn ensure_created(&self) -> std::io::Result<()> {
884        std::fs::create_dir_all(&self.path)
885    }
886
887    /// Remove the complete store directory tree. Absence is success: the tree
888    /// is already gone.
889    pub fn remove_tree(&self) -> std::io::Result<()> {
890        match std::fs::remove_dir_all(&self.path) {
891            Err(error) if error.kind() != std::io::ErrorKind::NotFound => Err(error),
892            _ => Ok(()),
893        }
894    }
895
896    #[cfg(any(test, feature = "test-utils"))]
897    pub async fn store_local_blob(
898        &self,
899        namespace: &str,
900        id: &str,
901        bytes: &[u8],
902    ) -> Result<(), LocalBlobStoreError> {
903        let destination = self.local_blob_path(namespace, id)?;
904        let mut staged = self
905            .stage_atomic_file(&destination)
906            .await
907            .map_err(LocalBlobStoreError::File)?;
908        staged
909            .write_bytes(bytes)
910            .await
911            .map_err(LocalBlobStoreError::File)?;
912        staged.commit().await.map_err(LocalBlobStoreError::File)
913    }
914
915    #[cfg(any(test, feature = "test-utils"))]
916    pub async fn read_local_blob(
917        &self,
918        namespace: &str,
919        id: &str,
920        expected_size: u64,
921    ) -> Result<Option<Vec<u8>>, LocalBlobStoreError> {
922        let Some(path) = self
923            .local_blob_path_if_present(namespace, id, expected_size)
924            .await?
925        else {
926            return Ok(None);
927        };
928        tokio::fs::read(&path).await.map(Some).map_err(|source| {
929            LocalBlobStoreError::File(FileError::at("read local blob", path, source))
930        })
931    }
932}
933
934async fn file_exists(path: &Path) -> Result<bool, FileError> {
935    match tokio::fs::metadata(path).await {
936        Ok(metadata) => Ok(metadata.is_file()),
937        Err(error) if error.kind() == std::io::ErrorKind::NotFound => Ok(false),
938        Err(source) => Err(FileError::at("stat store blob", path, source)),
939    }
940}
941
942async fn exact_file_facts(path: &Path) -> Result<(u64, crate::object_hash::ObjectHash), FileError> {
943    use sha2::{Digest, Sha256};
944    use tokio::io::AsyncReadExt;
945
946    let mut file = tokio::fs::File::open(path)
947        .await
948        .map_err(|source| FileError::at("open store blob", path, source))?;
949    let mut size = 0_u64;
950    let mut hasher = Sha256::new();
951    let mut buffer = vec![0_u8; 1 << 20];
952    loop {
953        let read = file
954            .read(&mut buffer)
955            .await
956            .map_err(|source| FileError::at("read store blob", path, source))?;
957        if read == 0 {
958            break;
959        }
960        size = size
961            .checked_add(read as u64)
962            .ok_or_else(|| FileError::SizeOverflow {
963                subject: "store blob",
964                path: path.to_path_buf(),
965            })?;
966        hasher.update(&buffer[..read]);
967    }
968    Ok((
969        size,
970        crate::object_hash::ObjectHash::from_digest(hasher.finalize().into()),
971    ))
972}
973
974async fn file_is_exact(
975    path: &Path,
976    expected_size: u64,
977    expected_hash: crate::object_hash::ObjectHash,
978) -> Result<bool, StoreBlobFileError> {
979    if !file_exists(path).await.map_err(StoreBlobFileError::File)? {
980        return Ok(false);
981    }
982    let (size, hash) = exact_file_facts(path)
983        .await
984        .map_err(StoreBlobFileError::File)?;
985    Ok(size == expected_size && hash == expected_hash)
986}
987
988async fn remove_file(path: &Path) -> Result<bool, FileError> {
989    match tokio::fs::remove_file(path).await {
990        Ok(()) => Ok(true),
991        Err(error) if error.kind() == std::io::ErrorKind::NotFound => Ok(false),
992        Err(source) => Err(FileError::at("remove store blob", path, source)),
993    }
994}
995
996async fn walk_files(path: &Path) -> Result<Vec<(PathBuf, u64, u64)>, FileError> {
997    let mut files = Vec::new();
998    let mut pending = vec![path.to_path_buf()];
999    while let Some(directory) = pending.pop() {
1000        let mut entries = match tokio::fs::read_dir(&directory).await {
1001            Ok(entries) => entries,
1002            Err(error) if error.kind() == std::io::ErrorKind::NotFound => continue,
1003            Err(source) => {
1004                return Err(FileError::at(
1005                    "read store blob directory",
1006                    directory,
1007                    source,
1008                ))
1009            }
1010        };
1011        while let Some(entry) = entries.next_entry().await.map_err(|source| {
1012            FileError::at("read store blob directory entry", &directory, source)
1013        })? {
1014            let entry_path = entry.path();
1015            let metadata = match entry.metadata().await {
1016                Ok(metadata) => metadata,
1017                Err(error) if error.kind() == std::io::ErrorKind::NotFound => continue,
1018                Err(source) => return Err(FileError::at("stat store blob", entry_path, source)),
1019            };
1020            if metadata.is_dir() {
1021                pending.push(entry_path);
1022            } else if !crate::local_file::AtomicStagedFile::is_staging_path(&entry_path) {
1023                let recency = metadata
1024                    .modified()
1025                    .map_err(|source| {
1026                        FileError::at("read store blob modification time", &entry_path, source)
1027                    })?
1028                    .duration_since(std::time::UNIX_EPOCH)
1029                    .map_err(|source| FileError::ModifiedBeforeUnixEpoch {
1030                        path: entry_path.clone(),
1031                        source,
1032                    })?
1033                    .as_millis() as u64;
1034                files.push((entry_path, recency, metadata.len()));
1035            }
1036        }
1037    }
1038    Ok(files)
1039}
1040
1041/// The single-writer store lock: an exclusive advisory lock on
1042/// `<store>/.coven-lock`, held for the life of a full open handle (and its
1043/// running sync loop). A second full open of the same store is refused with
1044/// [`StoreOpenGuardError::AlreadyOpen`] while the lock is held — the invariant
1045/// that keeps two writers from racing the same db and blob store.
1046///
1047/// # Read-only opens take no lock
1048///
1049/// A read-only open deliberately does **not** touch this lock. The lock is
1050/// exclusive, so a shared lock on the same file would block against a writer
1051/// that already holds it (and vice versa) — a reader could never coexist with
1052/// the writer it exists to read alongside. But a read-only open needs no lock
1053/// at all: the lock guards against a second *writer*, and a read-only handle
1054/// holds a `SQLITE_OPEN_READONLY` connection that cannot write. So a read-only
1055/// open skips the guard entirely. Cross-process safety comes from WAL mode (a
1056/// reader sees committed rows while the writer commits more), not from this
1057/// lock; the blob cache a reader may populate is per-device scratch written
1058/// atomically (temp + rename), so a reader and the writer touching the same
1059/// cache file never tear it. This lets one writer and any number of read-only
1060/// readers coexist on one store.
1061pub struct StoreOpenGuard {
1062    _file: std::fs::File,
1063}
1064
1065#[derive(Debug, thiserror::Error)]
1066pub enum StoreOpenGuardError {
1067    #[error("store is already open: {}", store_dir.display())]
1068    AlreadyOpen { store_dir: PathBuf },
1069    #[error("store database path has no parent: {}", path.display())]
1070    NoParent { path: PathBuf },
1071    #[error("store lock file: {0}")]
1072    File(#[from] FileError),
1073}
1074
1075impl StoreOpenGuard {
1076    pub fn acquire(store_dir: &StoreDir) -> Result<Self, StoreOpenGuardError> {
1077        let db_path = store_dir.db_path();
1078        let Some(dir) = db_path.parent() else {
1079            return Err(StoreOpenGuardError::NoParent { path: db_path });
1080        };
1081        std::fs::create_dir_all(dir).map_err(|source| {
1082            StoreOpenGuardError::File(FileError::at("create store directory", dir, source))
1083        })?;
1084        let lock_path = dir.join(".coven-lock");
1085        let file = std::fs::OpenOptions::new()
1086            .read(true)
1087            .write(true)
1088            .create(true)
1089            .truncate(false)
1090            .open(&lock_path)
1091            .map_err(|source| {
1092                StoreOpenGuardError::File(FileError::at("open store lock", &lock_path, source))
1093            })?;
1094        match Self::try_lock_exclusive(&file) {
1095            Ok(()) => Ok(Self { _file: file }),
1096            Err(std::fs::TryLockError::WouldBlock) => Err(StoreOpenGuardError::AlreadyOpen {
1097                store_dir: dir.to_path_buf(),
1098            }),
1099            Err(std::fs::TryLockError::Error(source)) => Err(StoreOpenGuardError::File(
1100                FileError::at("lock store", lock_path, source),
1101            )),
1102        }
1103    }
1104
1105    #[cfg(not(target_os = "android"))]
1106    fn try_lock_exclusive(file: &std::fs::File) -> Result<(), std::fs::TryLockError> {
1107        file.try_lock()
1108    }
1109
1110    /// std's `File::try_lock` is an `Unsupported` stub on Android — its cfg
1111    /// list carries `linux` but not `android` — so take the same
1112    /// `flock(LOCK_EX | LOCK_NB)` std takes on Linux, via rustix.
1113    #[cfg(target_os = "android")]
1114    fn try_lock_exclusive(file: &std::fs::File) -> Result<(), std::fs::TryLockError> {
1115        rustix::fs::flock(file, rustix::fs::FlockOperation::NonBlockingLockExclusive).map_err(
1116            |errno| {
1117                if errno == rustix::io::Errno::WOULDBLOCK {
1118                    std::fs::TryLockError::WouldBlock
1119                } else {
1120                    std::fs::TryLockError::Error(errno.into())
1121                }
1122            },
1123        )
1124    }
1125
1126    /// Acquire the guard for a test, panicking on refusal.
1127    #[cfg(any(test, feature = "test-utils"))]
1128    pub fn acquire_for_test(store_dir: &StoreDir) -> std::sync::Arc<Self> {
1129        std::sync::Arc::new(Self::acquire(store_dir).expect("acquire store open guard"))
1130    }
1131}
1132
1133impl Deref for StoreDir {
1134    type Target = Path;
1135
1136    fn deref(&self) -> &Path {
1137        &self.path
1138    }
1139}
1140
1141impl AsRef<Path> for StoreDir {
1142    fn as_ref(&self) -> &Path {
1143        &self.path
1144    }
1145}
1146
1147impl From<PathBuf> for StoreDir {
1148    fn from(path: PathBuf) -> Self {
1149        Self::new(path)
1150    }
1151}
1152
1153/// A temp dir plus a [`StoreDir`] rooted at it. The returned `TempDir` must be
1154/// held for the directory to outlive the test.
1155#[cfg(any(test, feature = "test-utils"))]
1156pub fn temp_store_dir() -> (tempfile::TempDir, StoreDir) {
1157    let tmp = tempfile::tempdir().expect("temp dir");
1158    let dir = StoreDir::new_ephemeral(tmp.path());
1159    (tmp, dir)
1160}
1161
1162#[cfg(test)]
1163#[path = "store_dir_tests.rs"]
1164mod tests;
coven_foundation/store_dir.rs

coven_foundation/
store_dir.rs
