diff --git a/core/Cargo.toml b/core/Cargo.toml index f32bf45eb..c1c3ab1f0 100644 --- a/core/Cargo.toml +++ b/core/Cargo.toml @@ -130,9 +130,6 @@ sd-supervisor = { path = "../crates/supervisor" } sd-bake = { path = "../crates/bake" } sd-pvcache = { path = "../crates/pvcache" } -# Vector database for memory files (optional for now) -# lancedb = "0.15" # Embedded vector database (conflicts with gpui) - # Media processing dependencies blurhash = "0.2" image = "0.25" diff --git a/core/src/domain/mod.rs b/core/src/domain/mod.rs index 9955bb471..7428a547c 100644 --- a/core/src/domain/mod.rs +++ b/core/src/domain/mod.rs @@ -12,7 +12,6 @@ pub mod file; pub mod library; pub mod location; pub mod media_data; -pub mod memory; pub mod resource; pub mod resource_manager; pub mod resource_registry; @@ -29,7 +28,6 @@ pub use file::{EntryKind, File, Sidecar}; pub use library::Library; pub use location::{IndexMode, Location, ScanState}; pub use media_data::{AudioMediaData, ImageMediaData, VideoMediaData}; -pub use memory::{MemoryFile, MemoryMetadata, MemoryScope}; pub use resource::{EventEmitter, Identifiable}; pub use resource_manager::ResourceManager; pub use space::{ diff --git a/core/src/infra/source_dirs.rs b/core/src/infra/source_dirs.rs index 9478f5c8c..37328dfd4 100644 --- a/core/src/infra/source_dirs.rs +++ b/core/src/infra/source_dirs.rs @@ -1,12 +1,20 @@ -//! Canonical per-source directory layout. +//! Canonical on-disk layout for what a machine keeps about its drives. //! -//! Everything a source owns on this machine lives in one directory: -//! `sources//` under the daemon data dir holds its session-restore -//! snapshot, its thumbnail cache, and its durable store (whose file name -//! belongs to `sd_store::SourceManager`). These are machine-local artifacts; -//! the registration itself is library metadata and lives in the `sources` -//! table. Clients never assume this layout — per-source paths surface through -//! the `core.ephemeral_status` query. +//! Two directories, because two different things are being stored. A *medium* +//! is a drive: `mediums//` holds the arena's session-restore snapshot and +//! its thumbnail cache, both of which describe the whole drive whatever is +//! registered over it. A *source* is a scope over one: `sources//` holds +//! its durable store (whose file name belongs to `sd_store::SourceManager`) +//! and its streamed block cache. +//! +//! Keeping them apart is what lets a source nest inside another. Both would +//! otherwise want the same directory for the same drive's map, and the inner +//! one would end up with a second copy of it. +//! +//! These are machine-local artifacts; the registration itself is library +//! metadata and lives in the `sources` table. Clients never assume this +//! layout — per-source paths surface through the `core.ephemeral_status` +//! query. use anyhow::{Context, Result}; use std::{ @@ -16,6 +24,7 @@ use std::{ use uuid::Uuid; const SOURCES_DIR: &str = "sources"; +const MEDIUMS_DIR: &str = "mediums"; const SNAPSHOT_FILE: &str = "ephemeral.snapshot"; const THUMBS_FILE: &str = "thumbs.pvcache"; const BLOCKS_DIR: &str = "blocks"; @@ -64,14 +73,33 @@ impl SourceDirs { Ok(dir) } - /// A source's session-restore snapshot. - pub fn snapshot_file(&self, id: Uuid) -> PathBuf { - self.source_dir(id).join(SNAPSHOT_FILE) + /// A medium's directory, created on demand. + pub fn create_medium_dir(&self, id: Uuid) -> Result { + let dir = self.medium_dir(id); + fs::create_dir_all(&dir) + .with_context(|| format!("create medium directory {}", dir.display()))?; + Ok(dir) } - /// A source's thumbnail hot-tier cache. + /// A medium's directory: the drive's map and everything derived from it. + pub fn medium_dir(&self, id: Uuid) -> PathBuf { + self.root + .parent() + .map(|parent| parent.join(MEDIUMS_DIR)) + .unwrap_or_else(|| self.root.join(MEDIUMS_DIR)) + .join(id.simple().to_string()) + } + + /// A medium's session-restore snapshot: the arena's durable copy. + pub fn snapshot_file(&self, id: Uuid) -> PathBuf { + self.medium_dir(id).join(SNAPSHOT_FILE) + } + + /// A medium's thumbnail hot-tier cache. Keyed here rather than per source + /// because a tile is a rendering of the drive's contents, and two sources + /// over one drive should not each decode it. pub fn thumbs_file(&self, id: Uuid) -> PathBuf { - self.source_dir(id).join(THUMBS_FILE) + self.medium_dir(id).join(THUMBS_FILE) } /// A source's streamed block cache. Inside the source's directory so @@ -118,8 +146,8 @@ impl SourceDirs { let Ok(id) = Uuid::try_parse(stem) else { continue; }; - if let Err(err) = self.create_source_dir(id) { - tracing::warn!("Could not prepare directory for source {id}: {err:#}"); + if let Err(err) = self.create_medium_dir(id) { + tracing::warn!("Could not prepare directory for medium {id}: {err:#}"); continue; } self.snapshot_file(id) @@ -162,16 +190,27 @@ mod tests { assert_eq!(dirs.root(), data.path().join("sources")); let source_dir = dirs.source_dir(id); assert_eq!(source_dir, dirs.root().join(id.simple().to_string())); + + // The map of a drive and what is kept off it are different directories, + // so a source nested in another does not want the same one. + let medium_dir = dirs.medium_dir(id); + assert_eq!( + medium_dir, + data.path().join("mediums").join(id.simple().to_string()) + ); assert_eq!( dirs.snapshot_file(id), - source_dir.join("ephemeral.snapshot") + medium_dir.join("ephemeral.snapshot") ); - assert_eq!(dirs.thumbs_file(id), source_dir.join("thumbs.pvcache")); + assert_eq!(dirs.thumbs_file(id), medium_dir.join("thumbs.pvcache")); // Resolution creates nothing; creation is explicit. assert!(!source_dir.exists()); + assert!(!medium_dir.exists()); dirs.create_source_dir(id).unwrap(); + dirs.create_medium_dir(id).unwrap(); assert!(source_dir.exists()); + assert!(medium_dir.exists()); } #[test] diff --git a/core/src/ops/indexing/ephemeral/cache.rs b/core/src/ops/indexing/ephemeral/cache.rs index 51171eca8..fcf9d2bcc 100644 --- a/core/src/ops/indexing/ephemeral/cache.rs +++ b/core/src/ops/indexing/ephemeral/cache.rs @@ -12,7 +12,7 @@ //! //! Path → partition resolution is longest-root-prefix over registered sources. -use super::sources::{SourceRecord, SourceRegistry, VolumeAnchor}; +use super::sources::{MediumKey, SourceRecord, SourceRegistry, VolumeAnchor}; use super::store::SourceStore; use super::EphemeralIndex; use crate::infra::db::entities::source; @@ -33,12 +33,15 @@ use tokio::sync::mpsc; use tokio::sync::RwLock as TokioRwLock; use uuid::Uuid; -/// One source's partition: its index plus the per-source path state the old -/// global cache used to track globally. -pub struct SourceSlot { - /// Source identity; `None` for the scratch partition. - pub id: Option, - /// Root at last attach. Scratch has no root. +/// One medium's partition: the arena for a drive, plus the path state that +/// belongs to the map rather than to any registration. +/// +/// Keyed by medium and not by source, so a source nested inside another shares +/// this rather than forking it. See [`MediumKey`]. +pub struct MediumSlot { + /// Which drive this maps. `Scratch` for paths under no registered source. + pub medium: MediumKey, + /// Where the medium begins. Scratch has no root. root: RwLock>, index: Arc>, indexed_paths: RwLock>, @@ -52,21 +55,17 @@ pub struct SourceSlot { /// Single restore attempt per session: concurrent callers await the same /// load instead of racing three copies of a 100 MB deserialization. restore_once: tokio::sync::OnceCell, - /// Serializes snapshot saves for this source. + /// Serializes snapshot saves for this medium. save_lock: tokio::sync::Mutex<()>, /// Entry count at the last completed save; identical partitions skip the /// rewrite (a burst of browse jobs otherwise re-saves 100 MB per job). last_saved_entries: std::sync::atomic::AtomicU64, - /// The source's durable store, opened once per session on first write. - /// `None` once opening has failed or the partition has nowhere to write: - /// scratch has no identity, and a cache with no directory has no disk. - store: tokio::sync::OnceCell>>, } -impl SourceSlot { - fn new(id: Option, root: Option) -> std::io::Result> { +impl MediumSlot { + fn new(medium: MediumKey, root: Option) -> std::io::Result> { Ok(Arc::new(Self { - id, + medium, root: RwLock::new(root), index: Arc::new(TokioRwLock::new(EphemeralIndex::new()?)), indexed_paths: RwLock::new(HashSet::new()), @@ -77,10 +76,17 @@ impl SourceSlot { restore_once: tokio::sync::OnceCell::new(), save_lock: tokio::sync::Mutex::new(()), last_saved_entries: std::sync::atomic::AtomicU64::new(u64::MAX), - store: tokio::sync::OnceCell::new(), })) } + /// The medium's stable id, which keys its directory on disk. + pub fn id(&self) -> Option { + match self.medium { + MediumKey::Scratch => None, + _ => Some(self.medium.id()), + } + } + pub fn root(&self) -> Option { self.root.read().clone() } @@ -156,9 +162,14 @@ pub struct EphemeralIndexCache { /// Per-source directory layout; `None` means no persistence. dirs: Option, /// Live partitions by source id. - slots: RwLock>>, + /// Live partitions by medium. One drive, one arena, however many sources + /// are registered over it. + slots: RwLock>>, + /// Durable stores by source id. A medium can host several, since a source + /// nested inside another persists its own subtree. + stores: RwLock>>, /// Fallback partition for paths under no registered source. - scratch: Arc, + scratch: Arc, /// Roots that have just become browsable, announced so whoever owns /// filesystem watching can arm them. /// @@ -195,7 +206,8 @@ impl EphemeralIndexCache { db: RwLock::new(None), dirs, slots: RwLock::new(HashMap::new()), - scratch: SourceSlot::new(None, None)?, + scratch: MediumSlot::new(MediumKey::Scratch, None)?, + stores: RwLock::new(HashMap::new()), restored_roots: RwLock::new(None), created_at: Instant::now(), }) @@ -321,19 +333,29 @@ impl EphemeralIndexCache { /// All registered sources with their live state. pub fn sources(&self) -> Vec { - let records: Vec = self.registry.lock().all().to_vec(); + let (records, mediums): (Vec, Vec) = { + let registry = self.registry.lock(); + let records = registry.all().to_vec(); + let mediums = records + .iter() + .map(|record| registry.medium_of(record).0) + .collect(); + (records, mediums) + }; let slots = self.slots.read(); records .into_iter() - .map(|record| { - let slot = slots.get(&record.id); + .zip(mediums) + .map(|(record, medium)| { + let slot = slots.get(&medium); SourceStatus { attached: record.root.exists(), restored: slot .map(|s| s.restored.load(Ordering::Acquire)) .unwrap_or(false), directory: self.dirs.as_ref().map(|d| d.source_dir(record.id)), - thumbs_path: self.dirs.as_ref().map(|d| d.thumbs_file(record.id)), + // The hot tier belongs to the map, so it is the medium's. + thumbs_path: self.dirs.as_ref().map(|d| d.thumbs_file(medium.id())), id: record.id, root: record.root, volume_uuid: record.volume_uuid, @@ -345,26 +367,28 @@ impl EphemeralIndexCache { .collect() } - /// Get (or lazily create) the live slot for a registry record. - fn slot_for_record(&self, record: &SourceRecord) -> Arc { - if let Some(slot) = self.slots.read().get(&record.id) { + /// Get (or lazily create) the live partition for the medium a record sits + /// on. Two sources on one drive get the same one. + fn slot_for_record(&self, record: &SourceRecord) -> Arc { + let (key, root) = self.registry.lock().medium_of(record); + if let Some(slot) = self.slots.read().get(&key) { return slot.clone(); } let mut slots = self.slots.write(); slots - .entry(record.id) + .entry(key.clone()) .or_insert_with(|| { - let slot = SourceSlot::new(Some(record.id), Some(record.root.clone())) - .expect("create ephemeral index for source"); - slot.set_detached(!record.root.exists()); + let slot = MediumSlot::new(key, Some(root.clone())) + .expect("create ephemeral index for medium"); + slot.set_detached(!root.exists()); slot }) .clone() } - /// Resolve the partition owning `path`: the registered source whose root - /// is the longest prefix, else scratch. - pub fn resolve(&self, path: &Path) -> Arc { + /// Resolve the partition owning `path`: the medium under the registered + /// source whose root is the longest prefix, else scratch. + pub fn resolve(&self, path: &Path) -> Arc { let record = self.registry.lock().resolve(path).cloned(); match record { Some(record) => self.slot_for_record(&record), @@ -379,30 +403,39 @@ impl EphemeralIndexCache { slot.is_detached() } - /// The durable store owning `path`, opened on first use. + /// The durable store that should hold `path`, opened on first use. /// - /// `None` for the scratch partition and for a cache with no sources - /// directory: neither has an identity to key a store on. Browsing keeps - /// working in both cases, which is what makes the arena the read path and - /// the store an addition to it. + /// The innermost registered source wins, which is what makes a source + /// inside another persist its own subtree without the outer one writing it + /// twice. Keyed by source rather than by medium for the same reason: one + /// drive, one map, any number of things kept off it. + /// + /// `None` for a path under no source and for a cache with no sources + /// directory. Browsing keeps working in both cases, which is what makes the + /// arena the read path and the store an addition to it. pub async fn store_for(&self, path: &Path) -> Option> { - let slot = self.resolve(path); - let (Some(id), Some(dirs), Some(root)) = (slot.id, self.dirs.as_ref(), slot.root()) else { - return None; + let record = self.registry.lock().resolve(path).cloned()?; + let dirs = self.dirs.as_ref()?; + + if let Some(store) = self.stores.read().get(&record.id) { + return Some(store.clone()); + } + + let store = match SourceStore::open(dirs, record.id, record.root.clone()).await { + Ok(store) => store, + Err(error) => { + tracing::error!(source = %record.id, %error, "source store unavailable"); + return None; + } }; - slot.store - .get_or_init(|| async { - match SourceStore::open(dirs, id, root).await { - Ok(store) => Some(store), - Err(error) => { - tracing::error!(source = %id, %error, "source store unavailable"); - None - } - } - }) - .await - .clone() + Some( + self.stores + .write() + .entry(record.id) + .or_insert(store) + .clone(), + ) } /// The index owning `path`, unconditionally (scratch fallback). @@ -576,12 +609,16 @@ impl EphemeralIndexCache { async fn attempt_restore( dirs: Option, record: SourceRecord, - slot: Arc, + slot: Arc, ) -> bool { let Some(dirs) = dirs else { return false; }; - let snapshot_path = dirs.snapshot_file(record.id); + // The snapshot is the arena's durable copy, so it belongs to the medium + // rather than to whatever is registered over it. + let medium_id = slot.medium.id(); + let medium_root = slot.root().unwrap_or_else(|| record.root.clone()); + let snapshot_path = dirs.snapshot_file(medium_id); let loaded = match EphemeralIndex::load_snapshot(&snapshot_path) { Ok(Some((index, meta))) => Some((index, meta)), Ok(None) => None, @@ -597,16 +634,16 @@ impl EphemeralIndexCache { return false; }; - // The snapshot names the source it was taken for. The file is keyed by - // source id in its path, so a mismatch means the cache directory was + // The snapshot names the medium it was taken for. The file is keyed by + // medium id in its path, so a mismatch means the cache directory was // copied or edited from outside; adopting it would bind one drive's // contents to another's identity. - if meta.source_id != record.id { + if meta.source_id != medium_id { tracing::warn!( - "Snapshot at {} belongs to source {}, not {}; ignoring", + "Snapshot at {} belongs to medium {}, not {}; ignoring", snapshot_path.display(), meta.source_id, - record.id + medium_id ); return false; } @@ -615,10 +652,10 @@ impl EphemeralIndexCache { // paths from the old mount baked into the snapshot. Reindexing the // present drive is cheaper than being subtly wrong; the stale // snapshot goes so the next save is clean. - if record.root.exists() && meta.root_path != record.root { + if medium_root.exists() && meta.root_path != medium_root { tracing::info!( "Snapshot for {} was taken at {}; discarding for reindex", - record.root.display(), + medium_root.display(), meta.root_path.display() ); let _ = std::fs::remove_file(&snapshot_path); @@ -645,8 +682,8 @@ impl EphemeralIndexCache { slot.set_detached(!meta.root_path.exists()); tracing::info!( - "Restored source {} from snapshot ({}, {})", - record.id, + "Restored medium {} from snapshot ({}, {})", + medium_id, meta.root_path.display(), if slot.is_detached() { "detached" @@ -683,7 +720,9 @@ impl EphemeralIndexCache { return Ok(()); }; let slot = self.slot_for_record(&record); - let snapshot_path = dirs.snapshot_file(record.id); + let medium_id = slot.medium.id(); + let medium_root = slot.root().unwrap_or_else(|| record.root.clone()); + let snapshot_path = dirs.snapshot_file(medium_id); // Funnel through the restore gate so a save can never precede the // session's restore attempt. @@ -755,16 +794,35 @@ impl EphemeralIndexCache { ); return Ok(()); } - index.save_snapshot(&snapshot_path, record.id, &record.root)?; + index.save_snapshot(&snapshot_path, medium_id, &medium_root)?; index.clear_dirty(); slot.last_saved_entries .store(entry_count, Ordering::Release); (entry_count, stats.total_file_bytes) }; - // Persist counts on the registry row so listings can show a drive's - // size without loading its snapshot. The snapshot itself is already on - // disk, so a failure here costs a stale count in listings rather than - // the index: report it and keep the save successful. + // Persist counts on the registry row so listings can show a size without + // loading a snapshot. What a source reports is what its store holds, + // not what the partition around it does: the arena maps the whole + // drive, and a source is a scope over part of it. The arena's figures + // stand in only when a source spans its whole medium, which is the case + // that has no store to ask yet. + // + // A store holding nothing has not been written yet, and a source over an + // already-mapped medium is in exactly that state until its adoption walk + // runs. Reporting zero there would empty a listing that has a full drive + // behind it, so the partition's figure stands until the store has one of + // its own. + let (entry_count, total_bytes) = match self.store_for(&record.root).await { + Some(store) => match store.counts().await { + Some((records, bytes)) if records > 0 => (records, bytes), + _ => (entry_count, total_bytes), + }, + None => (entry_count, total_bytes), + }; + + // The snapshot itself is already on disk, so a failure here costs a + // stale count in listings rather than the index: report it and keep the + // save successful. let updated = self .registry .lock() @@ -833,7 +891,7 @@ impl EphemeralIndexCache { self.resolve(path).indexed_paths.write().remove(path); } - fn fold_slots(&self, mut f: impl FnMut(&SourceSlot) -> T) -> Vec { + fn fold_slots(&self, mut f: impl FnMut(&MediumSlot) -> T) -> Vec { let mut out: Vec = self.slots.read().values().map(|s| f(s)).collect(); out.push(f(&self.scratch)); out @@ -930,7 +988,7 @@ impl EphemeralIndexCache { /// restore gate: the next touch re-restores from its snapshot instead of /// carrying a spent gate over an empty arena. pub async fn clear_all(&self) -> usize { - let old_slots: Vec> = { + let old_slots: Vec> = { let mut slots = self.slots.write(); let old: Vec<_> = slots.values().cloned().collect(); slots.clear(); @@ -1139,8 +1197,22 @@ mod tests { .await .unwrap(); - let slot = cache.resolve(Path::new("/mnt/drive/file.txt")); - assert_eq!(slot.id, Some(nested)); + // Both sources sit on one medium, because nesting narrows what is + // persisted rather than which map a file belongs to. + let outer = cache.resolve(Path::new("/mnt/other.txt")); + let inner = cache.resolve(Path::new("/mnt/drive/file.txt")); + assert!(Arc::ptr_eq(&outer, &inner)); + assert_eq!(inner.medium, MediumKey::Root(PathBuf::from("/mnt"))); + + // The innermost source is still the one that persists it. + assert_eq!( + cache + .registry + .lock() + .resolve(Path::new("/mnt/drive/file.txt")) + .map(|s| s.id), + Some(nested) + ); } #[test] @@ -1350,6 +1422,64 @@ mod tests { ); } + /// A source inside another narrows what is kept, not which map a file + /// belongs to. + /// + /// Keyed by source id, a nested registration forked the index: paths + /// under the inner root resolved to a second partition, the outer one + /// kept an unreachable copy of them, and each ledger minted its own + /// uuid for the same file, so a tag applied through one root was + /// invisible through the other. + #[tokio::test] + async fn a_nested_source_shares_the_drive_it_sits_on() { + let data = tempfile::tempdir().unwrap(); + let library = test_library(data.path()).await; + let root_dir = tempfile::tempdir().unwrap(); + let root = root_dir.path().to_path_buf(); + let inner = root.join("Photos"); + std::fs::create_dir_all(&inner).unwrap(); + + let cache = EphemeralIndexCache::with_sources_dir(Some(data.path().to_path_buf())) + .expect("cache"); + cache.attach_library(library.clone()).await.expect("attach"); + + let anchor = tracked_volume(&library, &root).await; + let drive = cache + .register_source(&root, Some(anchor.clone())) + .await + .expect("drive"); + let photos = cache + .register_source(&inner, Some(anchor)) + .await + .expect("photos"); + assert_ne!(drive, photos, "two registrations, two sources"); + + // One drive, one arena. + let outside = cache.resolve(&root.join("notes.txt")); + let inside = cache.resolve(&inner.join("a.jpg")); + assert!( + Arc::ptr_eq(&outside, &inside), + "a nested source forked the map of its own drive" + ); + + // And one snapshot, so a browse of either cannot overwrite the + // other's copy of the same drive. + let dirs = SourceDirs::new(data.path().join("sources")).expect("layout"); + assert_eq!( + dirs.snapshot_file(inside.medium.id()), + dirs.snapshot_file(outside.medium.id()) + ); + + // What differs is what each one keeps. + let outer_store = cache.store_for(&root.join("notes.txt")).await; + let inner_store = cache.store_for(&inner.join("a.jpg")).await; + assert!(outer_store.is_some() && inner_store.is_some()); + assert!( + !Arc::ptr_eq(outer_store.as_ref().unwrap(), inner_store.as_ref().unwrap()), + "the innermost source persists its own subtree" + ); + } + /// A change deep inside a watched source has to route to that source. /// /// A source rooted at the whole drive registers exactly one watched diff --git a/core/src/ops/indexing/ephemeral/sources.rs b/core/src/ops/indexing/ephemeral/sources.rs index adcb8f09a..47336d5b9 100644 --- a/core/src/ops/indexing/ephemeral/sources.rs +++ b/core/src/ops/indexing/ephemeral/sources.rs @@ -61,6 +61,40 @@ pub struct SourceRecord { pub total_bytes: Option, } +/// What an arena partition belongs to. +/// +/// A partition is a property of the medium rather than of a registration: one +/// drive, one map, whatever is persisted off it. Keying it by source id instead +/// is what made a source nested inside another fork the index rather than +/// narrow it, down to minting a second uuid for the same file. +/// +/// A root with no tracked volume under it is its own medium, which is the +/// network share and the fingerprint-less mount. +#[derive(Debug, Clone, PartialEq, Eq, Hash)] +pub enum MediumKey { + Volume(Uuid), + Root(PathBuf), + /// Paths under no registered source at all. + Scratch, +} + +/// Namespace for medium ids. Changing it renames every medium's directory, so +/// it is pinned by a test. +const MEDIUM_NAMESPACE: Uuid = Uuid::from_u128(0x9f2c_4e77_1b3a_4d58_9c21_6f0e_7a84_bd39); + +impl MediumKey { + /// A stable id, so a medium can own a directory without a registry row of + /// its own. Derived rather than minted: the same drive is the same medium + /// on every run. + pub fn id(&self) -> Uuid { + match self { + Self::Volume(uuid) => Uuid::new_v5(&MEDIUM_NAMESPACE, uuid.as_bytes()), + Self::Root(path) => Uuid::new_v5(&MEDIUM_NAMESPACE, path.to_string_lossy().as_bytes()), + Self::Scratch => Uuid::nil(), + } + } +} + impl SourceRecord { /// Rebuild from a stored row, resolving the absolute root against wherever /// the anchoring volume is mounted now. @@ -74,7 +108,6 @@ impl SourceRecord { (Some(_), Some(mount)) => join_relative(mount, &relative_root), _ => PathBuf::from(&relative_root), }; - Self { id: row.uuid, name: row.name, @@ -201,6 +234,41 @@ impl SourceRegistry { record } + /// The medium a source sits on, and where that medium begins. + /// + /// A tracked volume is the medium wherever one is anchored. Without one the + /// outermost registered root standing over this source is, which is what + /// makes a source inside a network share share the share's map rather than + /// starting a second one. Registration order does not matter: adding an + /// outer source later moves the inner one onto its medium, because this is + /// asked of the registry rather than remembered on the record. + pub fn medium_of(&self, record: &SourceRecord) -> (MediumKey, PathBuf) { + if let Some(uuid) = record.volume_uuid { + let mount = self + .sources + .iter() + .filter(|source| source.volume_uuid == Some(uuid) && source.is_locatable()) + .map(|source| source.root.clone()) + .min_by_key(|root| root.as_os_str().len()) + .unwrap_or_else(|| record.root.clone()); + return (MediumKey::Volume(uuid), mount); + } + + let outermost = self + .sources + .iter() + .filter(|source| { + source.volume_uuid.is_none() + && source.is_locatable() + && record.root.starts_with(&source.root) + }) + .map(|source| source.root.clone()) + .min_by_key(|root| root.as_os_str().len()) + .unwrap_or_else(|| record.root.clone()); + + (MediumKey::Root(outermost.clone()), outermost) + } + /// All registered sources. pub fn all(&self) -> &[SourceRecord] { &self.sources diff --git a/core/src/ops/indexing/ephemeral/store.rs b/core/src/ops/indexing/ephemeral/store.rs index d37bfa7aa..c35291968 100644 --- a/core/src/ops/indexing/ephemeral/store.rs +++ b/core/src/ops/indexing/ephemeral/store.rs @@ -226,6 +226,22 @@ impl SourceStore { self.send(Ingest::FinishSweep { unreachable }).await; } + /// What this source actually persists: records, and the bytes behind them. + /// + /// Read from the store rather than counted off the arena, because the arena + /// maps the whole drive and a source is a scope over part of it. Asking the + /// partition would report a nested source as owning everything around it. + pub async fn counts(&self) -> Option<(u64, u64)> { + self.flush().await; + let row: (i64, Option) = sqlx::query_as( + "SELECT (SELECT COUNT(*) FROM record), (SELECT SUM(size) FROM facet_file)", + ) + .fetch_one(self.db.pool()) + .await + .ok()?; + Some((row.0.max(0) as u64, row.1.unwrap_or(0).max(0) as u64)) + } + /// Commit everything queued so far and wait for it to land. pub async fn flush(&self) { let (done, wait) = oneshot::channel(); diff --git a/core/src/service/thumbs/service.rs b/core/src/service/thumbs/service.rs index fc4e3e422..36e7cc41d 100644 --- a/core/src/service/thumbs/service.rs +++ b/core/src/service/thumbs/service.rs @@ -152,7 +152,9 @@ impl ThumbService { async fn request_one(&self, path: &PathBuf, priority: u32) -> Option { let slot = self.ephemeral.resolve(path); - let source_id = slot.id?; + // The hot tier follows the arena, so it is keyed by the medium rather + // than by whatever is persisted off it. + let source_id = slot.id()?; let dirs = self.dirs.as_ref()?; // The uuid has to come from the index rather than be derived here, so