sausage/host/tests/events.rs

//! Durable event transactions, crash recovery, and world extension reactors.
use sausage::{harness::run_module, world::World};
use std::{
    path::Path,
    process::Command,
    time::{Duration, Instant},
};
const BUNDLE: &str = include_str!("../js/events-test-bundle.js");
struct Author(std::sync::Arc<dyn sausage::exec::Exec>);
#[async_trait::async_trait]
impl sausage::exec::Exec for Author {
    async fn exec(
        &self,
        command: sausage::exec::Command,
        mut options: sausage::exec::Options,
        output: &mut (dyn for<'a> FnMut(sausage::exec::Stream, &'a [u8]) + Send),
    ) -> anyhow::Result<i32> {
        for (key, value) in [
            ("GIT_AUTHOR_NAME", "Test"),
            ("GIT_AUTHOR_EMAIL", "test@example.com"),
            ("GIT_COMMITTER_NAME", "Test"),
            ("GIT_COMMITTER_EMAIL", "test@example.com"),
        ] {
            options.env.push((key.into(), value.into()));
        }
        self.0.exec(command, options, output).await
    }
}
async fn run(dir: &Path, mode: &str) -> anyhow::Result<()> {
    let world = World::open(dir)?;
    run_module(
        world,
        sausage::process::exec(World::open(dir)?)?,
        &format!("globalThis.eventsMode={mode:?};\n{BUNDLE}"),
    )
    .await
}
#[tokio::test(flavor = "current_thread")]
async fn transactions_and_feed() -> anyhow::Result<()> {
    let dir = tempfile::tempdir()?;
    run(dir.path(), "normal").await
}
#[tokio::test(flavor = "current_thread")]
async fn restart_child() -> anyhow::Result<()> {
    if let Ok(dir) = std::env::var("EVENTS_RESTART_DIR") {
        run(Path::new(&dir), &std::env::var("EVENTS_MODE")?).await?;
    }
    Ok(())
}
#[test]
fn sigkill_between_emit_and_reaction() -> anyhow::Result<()> {
    let dir = tempfile::tempdir()?;
    let mut child = Command::new(std::env::current_exe()?)
        .args(["--exact", "restart_child", "--nocapture"])
        .env("EVENTS_RESTART_DIR", dir.path())
        .env("EVENTS_MODE", "emit")
        .spawn()?;
    let deadline = Instant::now() + Duration::from_secs(30);
    while !dir.path().join("world/emitted").is_file() {
        if Instant::now() > deadline {
            child.kill()?;
            child.wait()?;
            anyhow::bail!("Event did not commit");
        }
        std::thread::yield_now();
    }
    child.kill()?;
    child.wait()?;
    for _ in 0..2 {
        let mut child = Command::new(std::env::current_exe()?)
            .args(["--exact", "restart_child", "--nocapture"])
            .env("EVENTS_RESTART_DIR", dir.path())
            .env("EVENTS_MODE", "consume")
            .spawn()?;
        let deadline = Instant::now() + Duration::from_secs(30);
        loop {
            if let Some(status) = child.try_wait()? {
                assert!(status.success());
                break;
            }
            if Instant::now() > deadline {
                child.kill()?;
                child.wait()?;
                anyhow::bail!("Reaction did not finish");
            }
            std::thread::yield_now();
        }
    }
    Ok(())
}
#[tokio::test(flavor = "current_thread")]
async fn world_extensions() -> anyhow::Result<()> {
    let userland = Path::new(env!("CARGO_MANIFEST_DIR")).join("../userland");
    let target = Path::new(env!("CARGO_TARGET_TMPDIR")).parent().unwrap();
    if !userland.join("cwd-init/init.o").is_file() {
        assert!(
            Command::new("nu")
                .args(["--no-config-file", "cwd-init/build.nu"])
                .current_dir(&userland)
                .status()?
                .success()
        );
    }
    assert!(
        Command::new("cargo")
            .args(["build", "--release", "-p", "git", "--manifest-path"])
            .arg(userland.join("Cargo.toml"))
            .args(["--target", "wasm32-wasip2", "--target-dir"])
            .arg(target)
            .args(["-j", "2", "--locked"])
            .current_dir(&userland)
            .status()?
            .success()
    );
    let dir = tempfile::tempdir()?;
    let world = World::open(dir.path())?;
    let disk = dir.path().join("world");
    std::fs::create_dir_all(disk.join("bin"))?;
    let git = target.join("wasm32-wasip2/release/git.wasm");
    std::fs::copy(git, disk.join("bin/git.wasm"))?;
    let repo = disk.join("src/sausage");
    std::fs::create_dir_all(&repo)?;
    assert!(
        Command::new("git")
            .args(["init", "-b", "main"])
            .arg(&repo)
            .status()?
            .success()
    );
    assert!(
        Command::new("git")
            .args(["commit", "--no-gpg-sign", "--allow-empty", "-m", "empty"])
            .env("GIT_AUTHOR_NAME", "Test")
            .env("GIT_AUTHOR_EMAIL", "test@example.com")
            .env("GIT_COMMITTER_NAME", "Test")
            .env("GIT_COMMITTER_EMAIL", "test@example.com")
            .current_dir(&repo)
            .status()?
            .success()
    );
    let exec = std::sync::Arc::new(Author(sausage::process::exec(World {
        root: world.root.clone(),
        host_dir: world.host_dir.clone(),
        processes: world.processes.clone(),
    })?));
    run_module(
        world,
        exec,
        &format!("globalThis.eventsMode='world';\n{BUNDLE}"),
    )
    .await
}