//! Metadata-keyed component loads and weak compiled versions; candidate copies never enter the path cache. #[cfg(test)] mod tests { use super::*; /// PROTOCOL 10.3: a pending component load does not occupy admission. #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn loading_does_not_hold_admission() { let folder = std::env::temp_dir().join(format!("sausage-admission-{}", uuid::Uuid::now_v7())); let world = crate::world::World::create(&folder).unwrap(); let path = world.plugin_dir("probe").join("plugin.wasm"); std::fs::create_dir_all(path.parent().unwrap()).unwrap(); std::fs::write(&path, []).unwrap(); let host = crate::Host::start_with_cache(&folder, &folder.join("cache")) .await .unwrap(); let entry = host .0 .components .paths .lock() .unwrap() .entry(path) .or_default() .clone(); let blocked = entry.lock().await; let caller = host.clone(); let pending = tokio::spawn(async move { caller.call("probe.echo", serde_json::json!({})).await }); tokio::time::timeout(std::time::Duration::from_secs(5), async { while Arc::strong_count(&entry) < 3 { tokio::task::yield_now().await; } }) .await .unwrap(); let admitted = host.0.admission.try_write().is_ok(); drop(blocked); assert!(pending.await.unwrap().is_err()); host.stop().await; std::fs::remove_dir_all(folder).unwrap(); assert!(admitted, "a component load held the admission gate"); } /// PROTOCOL 10.3: temporary discovery and activation copies never accumulate cached paths. #[tokio::test] async fn candidate_copies_do_not_enter_path_cache() { let root = Path::new(env!("CARGO_MANIFEST_DIR")).parent().unwrap(); let target = root.join("target/core/guests"); assert!( std::process::Command::new("cargo") .args([ "build", "--offline", "--target", "wasm32-wasip2", "--manifest-path" ]) .arg(root.join("cli/tests/plugins/probe/Cargo.toml")) .arg("--target-dir") .arg(&target) .status() .unwrap() .success() ); let folder = std::env::temp_dir().join(format!("sausage-component-cache-{}", uuid::Uuid::now_v7())); std::fs::create_dir_all(&folder).unwrap(); let runtime = Runtime::new(&root.join("target/core/cache-test")).unwrap(); let cache = Components::default(); for name in ["discover-test.wasm", "activate-test.wasm"] { let path = folder.join(name); std::fs::copy(target.join("wasm32-wasip2/debug/probe.wasm"), &path).unwrap(); let loaded = cache.load(path.clone(), runtime.clone()).await.unwrap(); assert!(cache.paths.lock().unwrap().is_empty()); let weak = Arc::downgrade(&loaded.prepared); drop(loaded); std::fs::remove_file(path).unwrap(); assert!(weak.upgrade().is_none()); } std::fs::remove_dir(folder).unwrap(); } } use crate::{ error, runtime::{Prepared, Runtime}, }; use sha2::{Digest, Sha256}; use std::{ collections::HashMap, path::{Path, PathBuf}, sync::{Arc, Mutex, Weak}, time::SystemTime, }; use tokio::sync::{Mutex as AsyncMutex, OnceCell}; #[derive(Clone, PartialEq, Eq)] pub(crate) struct Fingerprint { size: u64, modified: SystemTime, #[cfg(unix)] identity: (u64, u64, i64, i64), } impl Fingerprint { pub(crate) async fn read(path: &Path) -> Result { let metadata = tokio::fs::metadata(path) .await .map_err(|_| error(-32001, "no such plugin"))?; Ok(Self { size: metadata.len(), modified: metadata .modified() .map_err(|e| error(-32603, &e.to_string()))?, #[cfg(unix)] identity: { use std::os::unix::fs::MetadataExt; ( metadata.dev(), metadata.ino(), metadata.ctime(), metadata.ctime_nsec(), ) }, }) } } #[derive(Clone)] pub(crate) struct Loaded { pub(crate) prepared: Arc, pub(crate) hash: String, pub(crate) fingerprint: Fingerprint, // Retain the per-hash cell only while a cached path refers to this version. _compiled: Arc>>, } type PathEntry = Arc>>; #[derive(Default)] pub(crate) struct Components { paths: Mutex>, compiled: Mutex>>>>, } impl Components { pub(crate) async fn load( &self, path: PathBuf, runtime: Arc, ) -> Result { let candidate = path .file_name() .and_then(|n| n.to_str()) .is_some_and(|n| n.starts_with("activate-") || n.starts_with("discover-")); let entry = if candidate { Arc::default() } else { self.paths .lock() .unwrap() .entry(path.clone()) .or_default() .clone() }; let mut cached = entry.lock().await; let fingerprint = match Fingerprint::read(&path).await { Ok(metadata) => metadata, Err(error) => { *cached = None; return Err(error); } }; if let Some(version) = cached.as_ref() && version.fingerprint == fingerprint { return Ok(version.clone()); } let (bytes, hash) = tokio::task::spawn_blocking(move || -> std::io::Result<_> { let bytes = std::fs::read(path)?; let hash = format!("{:x}", Sha256::digest(&bytes)); Ok((bytes, hash)) }) .await .map_err(|e| error(-32603, &e.to_string()))? .map_err(|e| error(-32603, &e.to_string()))?; let compiled = { let mut versions = self.compiled.lock().unwrap(); versions.retain(|_, v| v.strong_count() > 0); if let Some(cell) = versions.get(&hash).and_then(Weak::upgrade) { cell } else { let cell = Arc::new(OnceCell::new()); versions.insert(hash.clone(), Arc::downgrade(&cell)); cell } }; let prepared = compiled .get_or_try_init(|| async move { tokio::task::spawn_blocking(move || -> anyhow::Result<_> { let component = runtime.compile(&bytes)?; Ok(Arc::new(runtime.prepare(&component))) }) .await .map_err(|e| error(-32603, &e.to_string()))? .map_err(|e| error(-32603, &e.to_string())) }) .await? .clone(); let loaded = Loaded { prepared, hash, fingerprint, _compiled: compiled, }; if !candidate { *cached = Some(loaded.clone()); } Ok(loaded) } }