sausage/cli/src/components.rs

//! 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<Self, serde_json::Value> {
        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<Prepared>,
    pub(crate) hash: String,
    pub(crate) fingerprint: Fingerprint,
    // Retain the per-hash cell only while a cached path refers to this version.
    _compiled: Arc<OnceCell<Arc<Prepared>>>,
}
type PathEntry = Arc<AsyncMutex<Option<Loaded>>>;
#[derive(Default)]
pub(crate) struct Components {
    paths: Mutex<HashMap<PathBuf, PathEntry>>,
    compiled: Mutex<HashMap<String, Weak<OnceCell<Arc<Prepared>>>>>,
}
impl Components {
    pub(crate) async fn load(
        &self,
        path: PathBuf,
        runtime: Arc<Runtime>,
    ) -> Result<Loaded, serde_json::Value> {
        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)
    }
}