From b9a5bcbeccf0ba8b1865d47c3ffc2a45d29909b1 Mon Sep 17 00:00:00 2001 From: aiongg Date: Sat, 14 Nov 2026 05:37:36 +0000 Subject: [PATCH] DES-10 T3: linked asset store and the engine's file loader AssetData::linked carries a LinkedSource (absolute and relative path, document, size, fingerprint, content hash). Linked bytes load through a process-wide single-flight LRU store (512 MiB, 1 GiB per file) and the Loader the engine installs in Session::new: absolute path first, then relative; a changed fingerprint is checked against the hash, and other bytes are Modified. Proxy.data is an AssetData so its decodes have a key of their own. --- Cargo.lock | 1 + crates/doc/src/asset_data.rs | 197 ++++++++++++++-- crates/doc/src/asset_data/store.rs | 348 +++++++++++++++++++++++++++++ crates/doc/src/lib.rs | 5 +- crates/engine/Cargo.toml | 2 + crates/engine/src/asset_io.rs | 253 +++++++++++++++++++++ crates/engine/src/lib.rs | 2 + 7 files changed, 786 insertions(+), 22 deletions(-) create mode 100644 crates/doc/src/asset_data/store.rs create mode 100644 crates/engine/src/asset_io.rs diff --git a/Cargo.lock b/Cargo.lock index 4d0f6a9d9..d3e98571f 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2086,6 +2086,7 @@ dependencies = [ "same-file", "serde", "serde_json", + "sha2 0.10.9", "tempfile", "thiserror 2.0.21", "unicode-segmentation", diff --git a/crates/doc/src/asset_data.rs b/crates/doc/src/asset_data.rs index 670625203..67ca3d0fa 100644 --- a/crates/doc/src/asset_data.rs +++ b/crates/doc/src/asset_data.rs @@ -1,10 +1,21 @@ //! The bytes of a placed file ([`crate::Asset::data`]): held in memory (embedded) or read from //! the linked file when needed. Caches of decoded images key by the value's [`SourceKey`], //! which outlives the bytes: a decoded image stays usable after the encoded bytes are dropped. +//! +//! Linked bytes are read by the [`Loader`] the engine installs ([`install_loader`]; this crate +//! does no file I/O) and kept in a process-wide store of at most [`STORE_BUDGET`] bytes, least +//! recently used first out. A file over [`MAX_LINKED_BYTES`] is never read. use std::fmt; +use std::path::{Path, PathBuf}; use std::sync::atomic::{AtomicU64, Ordering}; -use std::sync::{Arc, Weak}; +use std::sync::{Arc, Mutex, PoisonError, Weak}; + +use crate::datamerge::Fingerprint; + +mod store; + +pub use store::{Loaded, Loader, MAX_LINKED_BYTES, STORE_BUDGET, install_loader}; /// Names one version of an asset's bytes. Every [`AssetData`] made from new bytes gets a key no /// other value in this process has had; clones share it. Keys are never reused, so a cache @@ -14,7 +25,8 @@ use std::sync::{Arc, Weak}; pub struct SourceKey(pub u64); impl SourceKey { - fn next() -> Self { + /// A key no other value has had. + pub fn fresh() -> Self { static NEXT: AtomicU64 = AtomicU64::new(1); SourceKey(NEXT.fetch_add(1, Ordering::Relaxed)) } @@ -33,10 +45,64 @@ pub enum Unavailable { NoLoader, } +/// Where a linked asset's bytes are, and what they must be: the [`crate::Asset`] link record +/// ([`LinkedSource::of`]) with the document it is resolved against. +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct LinkedSource { + /// The file's absolute path ([`crate::Asset::link`]), tried first. + pub path: String, + /// The path relative to the document's folder ([`crate::Asset::link_relative`]), tried when + /// the absolute path doesn't hold the bytes. + pub relative: Option, + /// The document file `relative` is resolved against. + pub document: Option, + /// The size of the bytes. + pub len: u64, + /// Size and modification time of the file when it was known to hold the bytes: a file that + /// still has them is read without hashing. + pub fingerprint: Option, + /// The bytes' content hash ([`crate::Asset::link_hash`], `sha256:` and hex digits), checked + /// when the file's fingerprint differs. + pub hash: Option, +} + +impl LinkedSource { + /// The source `asset`'s link record names, resolved against the document file at + /// `document`. `None` without a link, or without the size its fingerprint records. + pub fn of(asset: &crate::Asset, document: Option<&Path>) -> Option { + let fingerprint = asset.link_fingerprint?; + Some(LinkedSource { + path: asset.link.clone()?, + relative: asset.link_relative.clone(), + document: document.map(Path::to_path_buf), + len: fingerprint.size, + fingerprint: Some(fingerprint), + hash: asset.link_hash.clone(), + }) + } +} + +/// A linked source, with the fingerprint a load last found the file at (hash-checked), so later +/// loads of a file whose time changed but whose bytes didn't skip the hash. +struct Linked { + source: LinkedSource, + trusted: Mutex>, +} + +impl Linked { + fn trusted(&self) -> Option { + *self.trusted.lock().unwrap_or_else(PoisonError::into_inner) + } + + fn trust(&self, fingerprint: Fingerprint) { + *self.trusted.lock().unwrap_or_else(PoisonError::into_inner) = Some(fingerprint); + } +} + #[derive(Clone)] enum Repr { Embedded(Arc>), - Linked { len: u64 }, + Linked(Arc), } /// An asset's bytes: embedded in the document, or linked (read from the file on demand). @@ -50,27 +116,71 @@ pub struct AssetData { impl AssetData { /// Embedded bytes, under a new key. pub fn embedded(bytes: impl Into>>) -> Self { - AssetData { key: SourceKey::next(), repr: Repr::Embedded(bytes.into()) } + AssetData { key: SourceKey::fresh(), repr: Repr::Embedded(bytes.into()) } } - /// The bytes of a linked file of `len` bytes, under a new key. - pub fn linked(len: u64) -> Self { - AssetData { key: SourceKey::next(), repr: Repr::Linked { len } } + /// Embedded bytes under `key`, which the caller made ([`SourceKey::fresh`]) for exactly these + /// bytes: bytes derived again from the same input (a layer variant of the same file) keep + /// the key their decodes are cached under. Different bytes under one key would draw stale + /// pixels. + pub fn embedded_under(key: SourceKey, bytes: impl Into>>) -> Self { + AssetData { key, repr: Repr::Embedded(bytes.into()) } + } + + /// The bytes of the linked file `source` names, under a new key. + pub fn linked(source: LinkedSource) -> Self { + AssetData { key: SourceKey::fresh(), repr: Repr::Linked(Arc::new(Linked { source, trusted: Mutex::new(None) })) } + } + + /// This value pointing at `source` (the file moved, the document was saved elsewhere, the + /// fingerprint was brought up to date). It keeps the key when both name the same bytes + /// (same size and content hash), so nothing decoded is decoded again. + pub fn with_source(&self, source: LinkedSource) -> Self { + let same = self.linked_source().is_some_and(|old| old.len == source.len && old.hash.is_some() && old.hash == source.hash); + let mut new = AssetData::linked(source); + if same { + new.key = self.key; + } + new } /// The bytes, when they are embedded. pub fn embedded_bytes(&self) -> Option<&Arc>> { match &self.repr { Repr::Embedded(b) => Some(b), - Repr::Linked { .. } => None, + Repr::Linked(_) => None, } } - /// The bytes: embedded ones, or the linked file's. + /// Where linked bytes are read from. + pub fn linked_source(&self) -> Option<&LinkedSource> { + match &self.repr { + Repr::Embedded(_) => None, + Repr::Linked(l) => Some(&l.source), + } + } + + /// The bytes: embedded ones, or the linked file's (read once by the installed [`Loader`], + /// and kept while the store's budget allows). Concurrent loads of one value read it once. pub fn load(&self) -> Result>, Unavailable> { match &self.repr { Repr::Embedded(b) => Ok(Arc::clone(b)), - Repr::Linked { .. } => Err(Unavailable::NoLoader), + Repr::Linked(l) => store::global().load(self.key, l, store::installed().as_deref()), + } + } + + /// The bytes without reading anything: embedded ones, or linked ones the store holds. + pub fn resident(&self) -> Option>> { + match &self.repr { + Repr::Embedded(b) => Some(Arc::clone(b)), + Repr::Linked(_) => store::global().resident(self.key), + } + } + + /// Drop the store's copy of linked bytes (they are read again when next loaded). + pub fn evict(&self) { + if self.is_linked() { + store::global().evict(self.key); } } @@ -82,7 +192,7 @@ impl AssetData { pub fn len(&self) -> u64 { match &self.repr { Repr::Embedded(b) => b.len() as u64, - Repr::Linked { len } => *len, + Repr::Linked(l) => l.source.len, } } @@ -91,7 +201,7 @@ impl AssetData { } pub fn is_linked(&self) -> bool { - matches!(self.repr, Repr::Linked { .. }) + matches!(self.repr, Repr::Linked(_)) } /// A handle that doesn't keep embedded bytes alive, and upgrades to this value (same key) @@ -99,16 +209,30 @@ impl AssetData { pub fn downgrade(&self) -> WeakAssetData { let repr = match &self.repr { Repr::Embedded(b) => WeakRepr::Embedded(Arc::downgrade(b)), - Repr::Linked { len } => WeakRepr::Linked { len: *len }, + Repr::Linked(l) => WeakRepr::Linked(Arc::clone(l)), }; WeakAssetData { key: self.key, repr } } } +/// Advances when linked files may have come back or changed ([`links_changed`]). +static LINK_GENERATION: AtomicU64 = AtomicU64::new(0); + +/// Say that linked files may have changed on disk (a relink, an update, a status check): caches +/// that remembered a linked file as unavailable ask for it again. +pub fn links_changed() { + LINK_GENERATION.fetch_add(1, Ordering::AcqRel); +} + +/// The count [`links_changed`] advances. +pub fn link_generation() -> u64 { + LINK_GENERATION.load(Ordering::Acquire) +} + #[derive(Clone)] enum WeakRepr { Embedded(Weak>), - Linked { len: u64 }, + Linked(Arc), } /// [`AssetData::downgrade`]. @@ -123,7 +247,7 @@ impl WeakAssetData { pub fn upgrade(&self) -> Option { let repr = match &self.repr { WeakRepr::Embedded(w) => Repr::Embedded(w.upgrade()?), - WeakRepr::Linked { len } => Repr::Linked { len: *len }, + WeakRepr::Linked(l) => Repr::Linked(Arc::clone(l)), }; Some(AssetData { key: self.key, repr }) } @@ -157,7 +281,7 @@ impl PartialEq for AssetData { fn eq(&self, other: &Self) -> bool { match (&self.repr, &other.repr) { (Repr::Embedded(a), Repr::Embedded(b)) => a == b, - (Repr::Linked { len: a }, Repr::Linked { len: b }) => self.key == other.key && a == b, + (Repr::Linked(a), Repr::Linked(b)) => self.key == other.key && a.source.len == b.source.len, _ => false, } } @@ -174,6 +298,10 @@ impl fmt::Debug for AssetData { mod tests { use super::*; + pub(super) fn source(path: &str, len: u64) -> LinkedSource { + LinkedSource { path: path.into(), relative: None, document: None, len, fingerprint: None, hash: None } + } + #[test] fn clones_keep_the_key_and_new_bytes_get_a_new_one() { let a = AssetData::embedded(vec![1, 2, 3]); @@ -195,12 +323,41 @@ mod tests { } #[test] - fn linked_bytes_need_a_loader() { - let l = AssetData::linked(10); + fn linked_assets_keep_their_key_only_for_the_same_bytes() { + let mut s = source("/nowhere/no-loader.png", 10); + let l = AssetData::linked(s.clone()); assert!(l.is_linked()); assert_eq!(l.len(), 10); - assert!(l.embedded_bytes().is_none()); - assert_eq!(l.load(), Err(Unavailable::NoLoader)); + assert!(l.embedded_bytes().is_none() && l.resident().is_none()); assert!(AssetData::default().is_empty()); + assert_ne!(l.with_source(s.clone()).key(), l.key(), "no hash: nothing says the bytes are the same"); + s.hash = Some("sha256:00".into()); + let hashed = AssetData::linked(s.clone()); + s.path = "/elsewhere/no-loader.png".into(); + let moved = hashed.with_source(s.clone()); + assert_eq!((moved.key(), moved.linked_source().map(|s| s.path.as_str())), (hashed.key(), Some("/elsewhere/no-loader.png"))); + s.hash = Some("sha256:01".into()); + assert_ne!(hashed.with_source(s).key(), hashed.key(), "other bytes, another key"); + let weak = hashed.downgrade(); + drop(hashed); + assert!(weak.upgrade().is_some(), "a linked value holds no bytes to let go of"); + } + + #[test] + fn a_link_record_names_its_source() { + let fingerprint = Fingerprint { size: 7, mtime: 9 }; + let asset = crate::Asset { + link: Some("/a/b.png".into()), + link_relative: Some("b.png".into()), + link_fingerprint: Some(fingerprint), + link_hash: Some("sha256:ab".into()), + ..Default::default() + }; + let s = LinkedSource::of(&asset, Some(Path::new("/a/doc.dcx"))).unwrap(); + assert_eq!( + (s.len, s.fingerprint, s.relative.as_deref(), s.document.as_deref()), + (7, Some(fingerprint), Some("b.png"), Some(Path::new("/a/doc.dcx"))) + ); + assert!(LinkedSource::of(&crate::Asset { link_fingerprint: None, ..asset }, None).is_none(), "the size comes from the fingerprint"); } } diff --git a/crates/doc/src/asset_data/store.rs b/crates/doc/src/asset_data/store.rs new file mode 100644 index 000000000..259bd8c66 --- /dev/null +++ b/crates/doc/src/asset_data/store.rs @@ -0,0 +1,348 @@ +//! The process-wide store of linked files' bytes ([`super::AssetData::load`]). + +use std::collections::HashMap; +use std::sync::{Arc, Condvar, LazyLock, Mutex, MutexGuard, PoisonError, RwLock}; + +use super::{Linked, LinkedSource, SourceKey, Unavailable}; +use crate::datamerge::Fingerprint; + +/// Linked bytes kept in memory at most, over all files. A file larger than this is read for the +/// one load that asks for it and not kept. +pub const STORE_BUDGET: u64 = 512 << 20; + +/// The largest linked file read: a larger one is [`Unavailable::TooLarge`]. +pub const MAX_LINKED_BYTES: u64 = 1 << 30; + +/// A linked file's bytes as a [`Loader`] read them. +pub struct Loaded { + pub bytes: Vec, + /// The fingerprint the file had, when its bytes were found to be the expected ones by their + /// hash: later loads trust that fingerprint as they trust [`LinkedSource::fingerprint`]. + pub fingerprint: Option, +} + +/// Reads linked files (the engine's, [`install_loader`]). +pub trait Loader: Send + Sync { + /// The bytes of the file `source` names, if it holds the expected ones: its size is + /// `source.len`, and its fingerprint is `source.fingerprint` or `trusted`, or its content + /// hash is `source.hash`. Other bytes are [`Unavailable::Modified`]; no file is + /// [`Unavailable::Missing`]; one over [`MAX_LINKED_BYTES`] is [`Unavailable::TooLarge`]. + fn load(&self, source: &LinkedSource, trusted: Option) -> Result; +} + +static LOADER: RwLock>> = RwLock::new(None); + +/// Read linked files with `loader` from now on (in the whole process). +pub fn install_loader(loader: Arc) { + *LOADER.write().unwrap_or_else(PoisonError::into_inner) = Some(loader); +} + +pub(super) fn installed() -> Option> { + LOADER.read().unwrap_or_else(PoisonError::into_inner).clone() +} + +pub(super) fn global() -> &'static Store { + static STORE: LazyLock = LazyLock::new(|| Store::new(STORE_BUDGET, MAX_LINKED_BYTES)); + &STORE +} + +type Outcome = Result>, Unavailable>; + +/// One load in progress, which the other loads of the same bytes wait for. +#[derive(Default)] +struct Flight { + outcome: Mutex>, + landed: Condvar, +} + +impl Flight { + fn wait(&self) -> Outcome { + let mut g = self.outcome.lock().unwrap_or_else(PoisonError::into_inner); + loop { + if let Some(outcome) = g.as_ref() { + return outcome.clone(); + } + g = self.landed.wait(g).unwrap_or_else(PoisonError::into_inner); + } + } +} + +struct Kept { + bytes: Arc>, + stamp: u64, +} + +#[derive(Default)] +struct Inner { + kept: HashMap, + flights: HashMap>, + bytes: u64, + clock: u64, +} + +impl Inner { + /// The bytes kept under `key`, marked used. + fn touch(&mut self, key: SourceKey) -> Option>> { + self.clock += 1; + let clock = self.clock; + self.kept.get_mut(&key).map(|k| { + k.stamp = clock; + Arc::clone(&k.bytes) + }) + } + + /// Keep `bytes` under `key`, dropping the least recently used others beyond `budget`. + fn keep(&mut self, key: SourceKey, bytes: Arc>, budget: u64) { + self.clock += 1; + let n = bytes.len() as u64; + if let Some(old) = self.kept.insert(key, Kept { bytes, stamp: self.clock }) { + self.bytes = self.bytes.saturating_sub(old.bytes.len() as u64); + } + self.bytes = self.bytes.saturating_add(n); + while self.bytes > budget { + let Some(oldest) = self.kept.iter().filter(|(k, _)| **k != key).min_by_key(|(_, k)| k.stamp).map(|(k, _)| *k) else { break }; + if let Some(old) = self.kept.remove(&oldest) { + self.bytes = self.bytes.saturating_sub(old.bytes.len() as u64); + } + } + } +} + +/// A single-flight, least-recently-used store of loaded bytes by [`SourceKey`]. +pub(super) struct Store { + inner: Mutex, + budget: u64, + cap: u64, +} + +/// Lands a flight when dropped: with its outcome, or as missing when the loader panicked, so +/// the loads waiting for it don't wait forever. +struct Landing<'a> { + store: &'a Store, + key: SourceKey, + flight: Arc, + outcome: Option, +} + +impl Drop for Landing<'_> { + fn drop(&mut self) { + let outcome = self.outcome.take().unwrap_or(Err(Unavailable::Missing)); + { + let mut g = self.store.lock(); + g.flights.remove(&self.key); + if let Ok(bytes) = &outcome + && bytes.len() as u64 <= self.store.budget + { + g.keep(self.key, Arc::clone(bytes), self.store.budget); + } + } + *self.flight.outcome.lock().unwrap_or_else(PoisonError::into_inner) = Some(outcome); + self.flight.landed.notify_all(); + } +} + +impl Store { + pub(super) fn new(budget: u64, cap: u64) -> Self { + Store { inner: Mutex::new(Inner::default()), budget, cap } + } + + fn lock(&self) -> MutexGuard<'_, Inner> { + self.inner.lock().unwrap_or_else(PoisonError::into_inner) + } + + /// The bytes of `linked`, kept under `key`: from the store, from a load of them in progress, + /// or read by `loader`. + pub(super) fn load(&self, key: SourceKey, linked: &Linked, loader: Option<&dyn Loader>) -> Outcome { + if linked.source.len > self.cap { + return Err(Unavailable::TooLarge); + } + let flight = { + let mut g = self.lock(); + if let Some(bytes) = g.touch(key) { + return Ok(bytes); + } + if let Some(f) = g.flights.get(&key).cloned() { + drop(g); + return f.wait(); + } + let Some(loader) = loader else { return Err(Unavailable::NoLoader) }; + let f = Arc::new(Flight::default()); + g.flights.insert(key, Arc::clone(&f)); + (f, loader) + }; + let (flight, loader) = flight; + let mut landing = Landing { store: self, key, flight, outcome: None }; + // Read outside the lock: other bytes load and are found meanwhile. + let outcome = match loader.load(&linked.source, linked.trusted()) { + Ok(l) if l.bytes.len() as u64 > self.cap => Err(Unavailable::TooLarge), + Ok(l) => { + if let Some(fingerprint) = l.fingerprint { + linked.trust(fingerprint); + } + Ok(Arc::new(l.bytes)) + } + Err(e) => Err(e), + }; + landing.outcome = Some(outcome.clone()); + drop(landing); + outcome + } + + /// The bytes kept under `key`, without loading them. + pub(super) fn resident(&self, key: SourceKey) -> Option>> { + self.lock().touch(key) + } + + pub(super) fn evict(&self, key: SourceKey) { + let mut g = self.lock(); + if let Some(old) = g.kept.remove(&key) { + g.bytes = g.bytes.saturating_sub(old.bytes.len() as u64); + } + } + + #[cfg(test)] + fn kept_bytes(&self) -> u64 { + self.lock().bytes + } +} + +#[cfg(test)] +mod tests { + use super::super::tests::source; + use super::super::{AssetData, Repr}; + use super::*; + use std::sync::atomic::{AtomicUsize, Ordering}; + + /// Serves `len` bytes of `fill` for every path, counting its reads; slow, so loads overlap. + #[derive(Default)] + struct Counting { + reads: AtomicUsize, + delay_ms: u64, + fail: Option, + trusted: Mutex>>, + } + + impl Loader for Counting { + fn load(&self, source: &LinkedSource, trusted: Option) -> Result { + self.reads.fetch_add(1, Ordering::SeqCst); + self.trusted.lock().unwrap().push(trusted); + std::thread::sleep(std::time::Duration::from_millis(self.delay_ms)); + if let Some(e) = self.fail { + return Err(e); + } + let fill = source.path.bytes().last().unwrap_or(0); + Ok(Loaded { bytes: vec![fill; source.len as usize], fingerprint: Some(Fingerprint { size: source.len, mtime: 5 }) }) + } + } + + fn linked(data: &AssetData) -> &Linked { + match &data.repr { + Repr::Linked(l) => l, + Repr::Embedded(_) => panic!("not linked"), + } + } + + fn load(store: &Store, data: &AssetData, loader: &dyn Loader) -> Outcome { + store.load(data.key(), linked(data), Some(loader)) + } + + #[test] + fn linked_bytes_are_read_once_and_kept() { + let store = Store::new(100, 1000); + let loader = Counting::default(); + let a = AssetData::linked(source("a", 10)); + assert_eq!(load(&store, &a, &loader).unwrap().as_slice(), [b'a'; 10]); + assert_eq!(load(&store, &a.clone(), &loader).unwrap().len(), 10); + assert_eq!(loader.reads.load(Ordering::SeqCst), 1); + assert_eq!(store.resident(a.key()).map(|b| b.len()), Some(10)); + assert_eq!(*loader.trusted.lock().unwrap(), [None]); + store.evict(a.key()); + load(&store, &a, &loader).unwrap(); + assert_eq!(loader.reads.load(Ordering::SeqCst), 2, "evicted bytes are read again"); + assert_eq!( + loader.trusted.lock().unwrap().last(), + Some(&Some(Fingerprint { size: 10, mtime: 5 })), + "a hash-checked fingerprint is trusted next time" + ); + assert_eq!(AssetData::linked(source("x", 1)).load(), Err(Unavailable::NoLoader), "no loader installed in these tests"); + } + + #[test] + fn the_least_recently_used_bytes_go_past_the_budget() { + let store = Store::new(100, 1000); + let loader = Counting::default(); + let [a, b, c] = ["a", "b", "c"].map(|p| AssetData::linked(source(p, 40))); + load(&store, &a, &loader).unwrap(); + load(&store, &b, &loader).unwrap(); + load(&store, &a, &loader).unwrap(); + load(&store, &c, &loader).unwrap(); + assert!(store.resident(b.key()).is_none(), "b was used least recently"); + assert!(store.resident(a.key()).is_some() && store.resident(c.key()).is_some()); + assert_eq!(store.kept_bytes(), 80); + } + + #[test] + fn a_file_over_the_budget_is_returned_but_not_kept_and_one_over_the_cap_is_not_read() { + let store = Store::new(100, 1000); + let loader = Counting::default(); + let small = AssetData::linked(source("s", 50)); + load(&store, &small, &loader).unwrap(); + let big = AssetData::linked(source("b", 200)); + assert_eq!(load(&store, &big, &loader).unwrap().len(), 200); + assert!(store.resident(big.key()).is_none()); + assert!(store.resident(small.key()).is_some(), "an unkept file evicts nothing"); + let huge = AssetData::linked(source("h", 1001)); + assert_eq!(load(&store, &huge, &loader), Err(Unavailable::TooLarge)); + assert_eq!(loader.reads.load(Ordering::SeqCst), 2); + } + + #[test] + fn concurrent_loads_of_the_same_bytes_read_once() { + let store = Store::new(1000, 1000); + let loader = Counting { delay_ms: 50, ..Default::default() }; + let a = AssetData::linked(source("a", 10)); + let got: Vec = std::thread::scope(|s| { + let handles: Vec<_> = (0..8).map(|_| s.spawn(|| load(&store, &a, &loader))).collect(); + handles.into_iter().map(|h| h.join().unwrap()).collect() + }); + assert!(got.iter().all(|g| g.as_ref().is_ok_and(|b| b.len() == 10))); + assert_eq!(loader.reads.load(Ordering::SeqCst), 1); + } + + #[test] + fn a_failed_load_is_shared_and_tried_again_later() { + let store = Store::new(1000, 1000); + let loader = Counting { delay_ms: 30, fail: Some(Unavailable::Modified), ..Default::default() }; + let a = AssetData::linked(source("a", 10)); + let got: Vec = std::thread::scope(|s| { + let handles: Vec<_> = (0..4).map(|_| s.spawn(|| load(&store, &a, &loader))).collect(); + handles.into_iter().map(|h| h.join().unwrap()).collect() + }); + assert!(got.iter().all(|g| *g == Err(Unavailable::Modified))); + assert!(store.resident(a.key()).is_none(), "nothing adopted"); + let reads = loader.reads.load(Ordering::SeqCst); + load(&store, &a, &loader).unwrap_err(); + assert_eq!(loader.reads.load(Ordering::SeqCst), reads + 1, "failures aren't remembered here"); + } + + /// A loader that panics lands its flight: the loads waiting for it return. + #[test] + fn a_panicking_loader_does_not_hang_the_waiters() { + struct Panics; + impl Loader for Panics { + fn load(&self, _: &LinkedSource, _: Option) -> Result { + std::thread::sleep(std::time::Duration::from_millis(30)); + panic!("loader fault"); + } + } + let store = Store::new(1000, 1000); + let a = AssetData::linked(source("a", 10)); + std::thread::scope(|s| { + let first = s.spawn(|| load(&store, &a, &Panics)); + std::thread::sleep(std::time::Duration::from_millis(5)); + let waiter = s.spawn(|| load(&store, &a, &Panics)); + assert!(first.join().is_err()); + assert!(waiter.join().is_ok_and(|r| r.is_err())); + }); + } +} diff --git a/crates/doc/src/lib.rs b/crates/doc/src/lib.rs index 1de2cc742..2288df5f3 100644 --- a/crates/doc/src/lib.rs +++ b/crates/doc/src/lib.rs @@ -45,7 +45,7 @@ use std::collections::BTreeMap; use std::sync::Arc; pub use anchored::{AnchorPosition, AnchoredObject, OBJECT_MARK}; -pub use asset_data::{AssetData, SourceKey, Unavailable, WeakAssetData}; +pub use asset_data::{AssetData, LinkedSource, SourceKey, Unavailable, WeakAssetData}; pub use assets::{AssetIndex, find_same_asset, insert_asset, remap_story_assets, story_asset_refs}; pub use attrs::*; pub use datamerge::{ @@ -697,8 +697,9 @@ impl Storage { pub struct Proxy { pub mime: String, pub pixels: (u32, u32), + /// The preview's encoded bytes, under a key of their own (decoded images are cached by it). #[serde(skip)] - pub data: Arc>, + pub data: AssetData, } impl std::fmt::Debug for Proxy { diff --git a/crates/engine/Cargo.toml b/crates/engine/Cargo.toml index 292025c33..954385e38 100644 --- a/crates/engine/Cargo.toml +++ b/crates/engine/Cargo.toml @@ -42,6 +42,8 @@ craft-xml-dtd = { workspace = true } craft-svg = { workspace = true } caseless = { workspace = true } unicode-segmentation = { workspace = true } +# Content hashes of linked files (`asset_io::content_hash`). +sha2 = "0.10" [target.'cfg(not(target_arch = "wasm32"))'.dependencies] craft-atomic = { workspace = true } diff --git a/crates/engine/src/asset_io.rs b/crates/engine/src/asset_io.rs new file mode 100644 index 000000000..bc870135c --- /dev/null +++ b/crates/engine/src/asset_io.rs @@ -0,0 +1,253 @@ +//! Reading linked files' bytes for the document model ([`designcraft_doc::asset_data`]): the +//! [`Loader`] every [`crate::Session`] installs. Files are read through [`crate::io`] (the +//! automation roots of the loading thread apply) and only ever yield the bytes the link record +//! names: other bytes at the path are [`Unavailable::Modified`] and are never handed out. +//! +//! The web build installs no loader: its linked assets are [`Unavailable::NoLoader`]. + +use std::path::{Path, PathBuf}; + +use designcraft_doc::asset_data::{Loaded, Loader, MAX_LINKED_BYTES}; +use designcraft_doc::{Fingerprint, LinkedSource, Unavailable}; + +/// Read linked files with [`FileLoader`] in this process (on the web: nothing to install). +pub fn install() { + #[cfg(not(target_arch = "wasm32"))] + designcraft_doc::asset_data::install_loader(std::sync::Arc::new(FileLoader)); +} + +/// The engine's [`Loader`]: [`read_linked`]. +pub struct FileLoader; + +impl Loader for FileLoader { + fn load(&self, source: &LinkedSource, trusted: Option) -> Result { + read_linked(source, trusted) + } +} + +/// `sha256:` and the hex digits of the SHA-256 of `bytes` ([`designcraft_doc::Asset::link_hash`]). +pub fn content_hash(bytes: &[u8]) -> String { + use sha2::Digest; + use std::fmt::Write; + let digest = sha2::Sha256::digest(bytes); + let mut out = String::with_capacity(7 + 64); + out.push_str("sha256:"); + for b in digest { + let _ = write!(out, "{b:02x}"); + } + out +} + +/// Size and modification time (whole seconds) from a file's metadata, as link records store them. +pub fn fingerprint_from(meta: &crate::io::Metadata) -> Fingerprint { + let mtime = meta + .modified() + .ok() + .and_then(|t| t.duration_since(std::time::UNIX_EPOCH).ok()) + .map(|d| i64::try_from(d.as_secs()).unwrap_or(i64::MAX)) + .unwrap_or(0); + Fingerprint { size: meta.len(), mtime } +} + +/// The files that may hold `source`'s bytes, in the order they are tried: the absolute path, +/// then the relative one resolved against the document. Paths [`crate::io::document_path`] +/// refuses are left out. +pub fn candidates(source: &LinkedSource) -> Vec { + let document = source.document.as_deref(); + let mut out: Vec = crate::io::document_path(&source.path, document).into_iter().collect(); + if let (Some(relative), Some(document)) = (source.relative.as_deref(), document) + && let Ok(p) = crate::io::document_path(relative, Some(document)) + && !out.contains(&p) + { + out.push(p); + } + out +} + +/// Why one candidate file doesn't serve. +enum Miss { + Absent, + Modified, + TooLarge, +} + +/// The bytes of the first candidate file ([`candidates`]) that holds `source`'s bytes: one of +/// `source.len` bytes whose fingerprint is `source.fingerprint` or `trusted`, or whose content +/// hash is `source.hash`. A source that records neither fingerprint nor hash is taken by its +/// size. The fingerprint of a file accepted by its hash is returned with the bytes. +pub fn read_linked(source: &LinkedSource, trusted: Option) -> Result { + if source.len > MAX_LINKED_BYTES { + return Err(Unavailable::TooLarge); + } + let mut worst = Miss::Absent; + for path in candidates(source) { + match read_candidate(&path, source, trusted) { + Ok(loaded) => return Ok(loaded), + Err(Miss::Absent) => {} + Err(miss @ Miss::TooLarge) => worst = miss, + Err(Miss::Modified) => { + if !matches!(worst, Miss::TooLarge) { + worst = Miss::Modified; + } + } + } + } + Err(match worst { + Miss::Absent => Unavailable::Missing, + Miss::Modified => Unavailable::Modified, + Miss::TooLarge => Unavailable::TooLarge, + }) +} + +fn read_candidate(path: &Path, source: &LinkedSource, trusted: Option) -> Result { + let meta = crate::io::metadata(path).map_err(|_| Miss::Absent)?; + if !meta.is_file() { + return Err(Miss::Absent); + } + let found = fingerprint_from(&meta); + if found.size > MAX_LINKED_BYTES { + return Err(Miss::TooLarge); + } + if found.size != source.len { + return Err(Miss::Modified); + } + let known = Some(found) == source.fingerprint || Some(found) == trusted; + if !known && source.hash.is_none() && source.fingerprint.is_some() { + // Nothing to tell the bytes by. + return Err(Miss::Modified); + } + // At most one byte past the size: a file that grew since is caught without reading it all. + let bytes = crate::io::read_capped(path, source.len).map_err(|_| Miss::Absent)?; + if bytes.len() as u64 != source.len { + return Err(Miss::Modified); + } + if known { + return Ok(Loaded { bytes, fingerprint: None }); + } + match &source.hash { + Some(hash) if *hash != content_hash(&bytes) => Err(Miss::Modified), + Some(_) => Ok(Loaded { bytes, fingerprint: Some(found) }), + None => Ok(Loaded { bytes, fingerprint: None }), + } +} + +#[cfg(test)] +mod tests { + use super::*; + use designcraft_doc::AssetData; + + struct Dir(PathBuf); + + impl Dir { + fn new(name: &str) -> Self { + let dir = std::env::temp_dir().join(format!("dc-asset-io-{name}-{}", std::process::id())); + let _ = std::fs::remove_dir_all(&dir); + std::fs::create_dir_all(&dir).unwrap(); + Dir(dir) + } + } + + impl Drop for Dir { + fn drop(&mut self) { + let _ = std::fs::remove_dir_all(&self.0); + } + } + + /// The link record of the file at `path` as it is now. + fn source_of(path: &Path) -> LinkedSource { + let bytes = std::fs::read(path).unwrap(); + LinkedSource { + path: path.to_string_lossy().into_owned(), + relative: None, + document: None, + len: bytes.len() as u64, + fingerprint: Some(fingerprint_from(&std::fs::metadata(path).unwrap())), + hash: Some(content_hash(&bytes)), + } + } + + /// Set the file's modification time `secs` seconds into the past. + fn age(path: &Path, secs: u64) { + let file = std::fs::OpenOptions::new().write(true).open(path).unwrap(); + file.set_modified(std::time::SystemTime::now() - std::time::Duration::from_secs(secs)).unwrap(); + } + + #[test] + fn content_hashes_are_sha256_hex() { + assert_eq!(content_hash(b"abc"), "sha256:ba7816bf8f01cfea414140de5dae2223b00361a396177a9cb410ff61f20015ad"); + } + + #[test] + fn a_linked_file_is_read_when_it_holds_the_recorded_bytes() { + let dir = Dir::new("read"); + let path = dir.0.join("a.png"); + std::fs::write(&path, b"original").unwrap(); + let source = source_of(&path); + assert_eq!(read_linked(&source, None).ok().map(|l| (l.bytes, l.fingerprint)), Some((b"original".to_vec(), None))); + // Same bytes, another time: accepted by the hash, and the new fingerprint returned. + age(&path, 1000); + let now = fingerprint_from(&std::fs::metadata(&path).unwrap()); + let loaded = read_linked(&source, None).ok().unwrap(); + assert_eq!((loaded.bytes.as_slice(), loaded.fingerprint), (b"original".as_slice(), Some(now))); + assert!(read_linked(&source, Some(now)).is_ok_and(|l| l.fingerprint.is_none()), "a trusted fingerprint needs no hash"); + let unverifiable = LinkedSource { hash: None, ..source.clone() }; + assert_eq!(read_linked(&unverifiable, None).err(), Some(Unavailable::Modified), "nothing to tell the bytes by"); + } + + #[test] + fn other_bytes_at_the_path_are_modified_and_a_missing_file_is_missing() { + let dir = Dir::new("modified"); + let path = dir.0.join("a.png"); + std::fs::write(&path, b"original").unwrap(); + let source = source_of(&path); + std::fs::write(&path, b"changed!").unwrap(); + age(&path, 1000); + assert_eq!(read_linked(&source, None).err(), Some(Unavailable::Modified), "same size, other bytes"); + std::fs::write(&path, b"longer than before").unwrap(); + assert_eq!(read_linked(&source, None).err(), Some(Unavailable::Modified), "another size"); + std::fs::remove_file(&path).unwrap(); + assert_eq!(read_linked(&source, None).err(), Some(Unavailable::Missing)); + let too_large = LinkedSource { len: MAX_LINKED_BYTES + 1, ..source }; + assert_eq!(read_linked(&too_large, None).err(), Some(Unavailable::TooLarge)); + } + + #[test] + fn the_relative_path_serves_when_the_absolute_one_does_not() { + let dir = Dir::new("relative"); + std::fs::create_dir_all(dir.0.join("moved/links")).unwrap(); + let away = dir.0.join("moved/links/a.png"); + std::fs::write(&away, b"original").unwrap(); + let mut source = source_of(&away); + source.path = dir.0.join("gone/links/a.png").to_string_lossy().into_owned(); + assert_eq!(read_linked(&source, None).err(), Some(Unavailable::Missing)); + source.relative = Some("links/a.png".into()); + source.document = Some(dir.0.join("moved/doc.dcx")); + assert_eq!(read_linked(&source, None).ok().map(|l| l.bytes), Some(b"original".to_vec())); + // A changed file at the absolute path doesn't hide the one that holds the bytes. + std::fs::create_dir_all(dir.0.join("gone/links")).unwrap(); + std::fs::write(dir.0.join("gone/links/a.png"), b"other!!!").unwrap(); + age(&dir.0.join("gone/links/a.png"), 1000); + assert_eq!(read_linked(&source, None).ok().map(|l| l.bytes), Some(b"original".to_vec())); + } + + /// Through the installed loader and the store: a file changed on disk is never adopted. + #[test] + fn asset_data_loads_through_the_installed_loader() { + install(); + let dir = Dir::new("installed"); + let path = dir.0.join("a.png"); + std::fs::write(&path, b"original").unwrap(); + let data = AssetData::linked(source_of(&path)); + let changed = AssetData::linked(source_of(&path)); + assert_eq!(data.load().ok().map(|b| b.to_vec()), Some(b"original".to_vec())); + assert!(data.resident().is_some()); + std::fs::write(&path, b"changed!").unwrap(); + age(&path, 1000); + assert_eq!(data.load().ok().map(|b| b.to_vec()), Some(b"original".to_vec()), "kept bytes are the recorded ones"); + assert_eq!(changed.load().err(), Some(Unavailable::Modified)); + assert!(changed.resident().is_none(), "nothing adopted"); + std::fs::remove_file(&path).unwrap(); + data.evict(); + assert_eq!(data.load().err(), Some(Unavailable::Missing)); + } +} diff --git a/crates/engine/src/lib.rs b/crates/engine/src/lib.rs index ac899412e..e18c78058 100644 --- a/crates/engine/src/lib.rs +++ b/crates/engine/src/lib.rs @@ -7,6 +7,7 @@ #![deny(clippy::unwrap_used, clippy::expect_used, clippy::panic, clippy::unimplemented, clippy::todo, clippy::unreachable)] #![forbid(unsafe_code)] +pub mod asset_io; #[cfg(not(target_arch = "wasm32"))] pub mod atomic; pub mod autosize; @@ -467,6 +468,7 @@ impl Session { pub fn new() -> Self { guard::install_panic_hook(); + asset_io::install(); Session { docs: vec![], active: None,