OXYGENHUB

Coverage / crates/flow/src/wasm.rs 84.6%

352 of 416 lines run, 79 of 133 functions · run #226 on main · run missed

1
//! The WASM tier (spec §4.1, oxygenflow#75): a job's `exec` names a WASI
2
//! preview 1 module, or a glob of modules run one after another, which
3
//! wasmtime runs with the job's workspace at `/src` and nothing else: no
4
//! network, no other file of the runner. Every runner can offer it, and it
5
//! is the tier phones and browsers run (docs/handover.md).
6
//!
7
//! As in the syscall tier the module runs in a child of the runner
8
//! (`flow __wasm SPEC`), so the runner streams its output as it comes,
9
//! stops it at the time limit or when the run is cancelled, and measures
10
//! its memory the same way; wasmtime itself also holds it to the sandbox's
11
//! memory limit.
12
13
use crate::microvm::{Outcome, Sink};
14
use crate::syscall::{lines, self_hash, FAILED};
15
use anyhow::{bail, Context, Result};
16
use flow_core::api::Record;
17
use flow_core::{Environment, Job, MountSource, Sandbox};
18
use serde::{Deserialize, Serialize};
19
use std::path::{Path, PathBuf};
20
use std::sync::mpsc;
21
use std::time::{Duration, Instant};
22
23
/// The runtime, named in every receipt this tier writes.
24
pub const WASMTIME: &str = "49.0.4";
25
26
/// Whether this build has the tier: a build for a system wasmtime does not
27
/// run on (Haiku, oxygenflow#91) leaves it out with the `wasm` feature, and
28
/// its runner does not offer it.
29
pub const AVAILABLE: bool = cfg!(feature = "wasm");
30
31
/// What a build without the tier says when asked to run a job in it.
32
const WITHOUT: &str = "this flow was built without the WASM tier (the `wasm` feature): it runs on systems wasmtime supports";
33
34
/// A line the helper writes to say something about the job itself (which
35
/// module starts), shown as the runner's own note rather than the job's.
36
const NOTE: &str = "\u{1e}N ";
37
38
/// The exit code for a module that trapped (crashed) rather than exiting.
39
#[cfg(feature = "wasm")]
40
const TRAPPED: i32 = 134;
41
42
/// What the helper runs, written by the runner for `flow __wasm`.
43
#[derive(Serialize, Deserialize)]
44
struct Spec {
45
    /// The job's files, preopened as `/src` (and as the current directory).
46
    workspace: PathBuf,
47
    /// The modules to run in turn, relative to the workspace.
48
    modules: Vec<String>,
49
    /// The arguments after the module.
50
    args: Vec<String>,
51
    env: Vec<(String, String)>,
52
    memory_mib: u32,
53
    /// The job's mounts: a directory of the runner and where the module sees it.
54
    preopens: Vec<(PathBuf, String)>,
55
}
56
57
/// The receipt environment of a job in this tier: the runtime and its
58
/// version, this runner's binary, the memory allowed, and every module run
59
/// with its hash, so the receipt says exactly what code ran.
6012×
pub fn environment(sandbox: &Sandbox, job: &Job, modules: &[(String, String)]) -> Environment {
61
    Environment {
6212×
        runtime: "wasm".into(),
6312×
        parts: vec![
6412×
            ("wasmtime".into(), WASMTIME.into()),
6512×
            ("flow".into(), self_hash().to_string()),
6612×
            ("memory_mib".into(), sandbox.memory_mib.to_string()),
67
        ]
6812×
        .into_iter()
6914×
        .chain(modules.iter().map(|(path, hash)| (format!("module {path}"), hash.clone())))
7012×
        .collect(),
7112×
        rootfs: String::new(),
7212×
        argv: job.exec.clone(),
7324×
        env: job.env.iter().map(|(k, v)| (k.clone(), v.clone())).collect(),
74
    }
7512×
}
76
77
/// The modules `program` names in `workspace`: one file, or the files a
78
/// glob matches (`*` in its last part only), sorted. A path may start with
79
/// `/src/`, where the module sees the workspace.
8018×
pub fn modules(workspace: &Path, program: &str) -> Result<Vec<String>> {
8118×
    let rel = program.strip_prefix("/src/").unwrap_or(program).trim_start_matches("./");
8246×
    if rel.is_empty() || rel.starts_with('/') || rel.split('/').any(|p| p == "..") {
832×
        bail!("module {program:?}: give a path inside the job's files, such as target/wasm32-wasip1/debug/app.wasm");
8416×
    }
8516×
    let (dir, last) = rel.rsplit_once('/').unwrap_or(("", rel));
8616×
    if dir.contains('*') {
871×
        bail!("module {program:?}: only the file name may hold `*`");
8815×
    }
8915×
    if !last.contains('*') {
9011×
        if !workspace.join(rel).is_file() {
910×
            bail!("module {rel} is not among the job's files (its scope or an earlier job's artifact)");
9211×
        }
9311×
        return Ok(vec![rel.to_string()]);
944×
    }
954×
    let mut found: Vec<String> = std::fs::read_dir(workspace.join(dir))
964×
        .with_context(|| format!("module {program:?}: the directory {dir} is not among the job's files"))?
9710×
        .filter_map(|e| e.ok())
9810×
        .filter(|e| e.file_type().is_ok_and(|t| t.is_file()))
9910×
        .filter_map(|e| e.file_name().into_string().ok())
10010×
        .filter(|name| wildcard(last, name))
1016×
        .map(|name| if dir.is_empty() { name } else { format!("{dir}/{name}") })
1024×
        .collect();
1034×
    found.sort();
1044×
    if found.is_empty() {
1051×
        bail!("no module matches {program:?}");
1063×
    }
1073×
    Ok(found)
10818×
}
109
110
/// Whether `name` matches `pattern`, where `*` stands for any run of characters.
11117×
fn wildcard(pattern: &str, name: &str) -> bool {
11217×
    let parts: Vec<&str> = pattern.split('*').collect();
11317×
    let (first, last) = (parts[0], parts[parts.len() - 1]);
11417×
    if parts.len() == 1 {
1152×
        return pattern == name;
11615×
    }
11715×
    if !name.starts_with(first) || !name[first.len()..].ends_with(last) || name.len() < first.len() + last.len() {
1186×
        return false;
1199×
    }
1209×
    let mut rest = &name[first.len()..name.len() - last.len()];
1219×
    for middle in &parts[1..parts.len() - 1] {
1221×
        match rest.find(middle) {
1231×
            Some(i) => rest = &rest[i + middle.len()..],
1240×
            None => return false,
125
        }
126
    }
1279×
    true
12817×
}
129
130
/// Run `job` in this tier with its files in `workspace` (as the syscall
131
/// tier lays them out); `scratch` holds the helper's spec and the job's
132
/// mounts.
13312×
pub fn run(sandbox: &Sandbox, job: &Job, workspace: &Path, scratch: &Path, timeout: Duration, sink: &mut dyn Sink) -> Result<Outcome> {
13412×
    if !AVAILABLE {
1350×
        bail!("job {}: {WITHOUT}", job.name);
13612×
    }
13712×
    if !job.secrets.is_empty() || !job.identities.is_empty() {
1380×
        bail!("job {}: the WASM tier takes no secrets or identities yet; run it in a microvm or syscall sandbox", job.name);
13912×
    }
14012×
    let Some(program) = job.exec.first() else { bail!("job {}: exec names no module", job.name) };
14112×
    let modules = modules(workspace, program)?;
14212×
    let hashed: Vec<(String, String)> =
14314×
        modules.iter().map(|m| Ok((m.clone(), crate::home::hash_file(&workspace.join(m))?))).collect::<Result<_>>()?;
14412×
    let environment = environment(sandbox, job, &hashed);
14512×
    let mut preopens = Vec::new();
14612×
    let mut disks = 0;
14712×
    for (i, m) in job.mounts.iter().enumerate() {
1480×
        let dir = match m.source {
149
            MountSource::Disk => {
1500×
                disks += 1;
1510×
                crate::syscall::disk_dir(scratch, disks - 1)
152
            }
1530×
            MountSource::Tmpfs => scratch.join(format!("tmpfs-{i}")),
154
        };
1550×
        std::fs::create_dir_all(&dir)?;
1560×
        preopens.push((dir, m.target.clone()));
157
    }
15812×
    let spec = Spec {
15912×
        workspace: workspace.to_path_buf(),
16012×
        modules,
16112×
        args: job.exec[1..].to_vec(),
16224×
        env: job.env.iter().map(|(k, v)| (k.clone(), v.clone())).collect(),
16312×
        memory_mib: sandbox.memory_mib,
16412×
        preopens,
165
    };
16612×
    if in_process() {
1676×
        return run_in_thread(&spec, environment, timeout, sink);
1686×
    }
1696×
    let spec_path = scratch.join("wasm.json");
1706×
    std::fs::write(&spec_path, serde_json::to_vec(&spec)?)?;
171
1726×
    let started = Instant::now();
1736×
    let mut child = std::process::Command::new(crate::tier::self_exe())
1746×
        .arg("__wasm")
1756×
        .arg(&spec_path)
1766×
        .env_clear()
1776×
        .current_dir(scratch)
1786×
        .stdin(std::process::Stdio::null())
1796×
        .stdout(std::process::Stdio::piped())
1806×
        .stderr(std::process::Stdio::piped())
1816×
        .spawn()
1826×
        .context("start the WASM helper")?;
1836×
    let (tx, rx) = mpsc::channel::<(bool, Vec<u8>)>();
1846×
    let out = child.stdout.take().unwrap();
1856×
    let err = child.stderr.take().unwrap();
1866×
    let tx2 = tx.clone();
1876×
    let readers = [std::thread::spawn(move || lines(out, true, tx)), std::thread::spawn(move || lines(err, false, tx2))];
18815×
    let t = |s: &Instant| s.elapsed().as_millis() as u64;
1896×
    let (mut timed_out, mut cancelled, mut failed_to_start) = (false, false, false);
1906×
    let mut peak_memory_mib = None;
1916×
    let deadline = started + timeout;
192
    loop {
19315×
        if let Some(m) = crate::facts::peak_rss_mib(child.id()) {
19415×
            peak_memory_mib = Some(m);
19515×
        }
19615×
        match rx.recv_timeout(Duration::from_millis(200)) {
1979×
            Ok((stdout, line)) => {
1989×
                let text = String::from_utf8_lossy(&line).into_owned();
1999×
                let rec = |stream: &str, text: &str| Record { t: t(&started), stream: stream.into(), text: text.into() };
2009×
                if let Some(note) = text.strip_prefix(NOTE).filter(|_| !stdout) {
2015×
                    sink.record(rec("flow", note));
2025×
                } else if let Some(why) = text.strip_prefix(FAILED).filter(|_| !stdout) {
2030×
                    failed_to_start = true;
2040×
                    sink.record(rec("flow", &format!("the WASM tier could not run the job: {why}")));
2050×
                } else {
2064×
                    sink.record(rec(if stdout { "out" } else { "err" }, &text));
207
                }
208
            }
2090×
            Err(mpsc::RecvTimeoutError::Timeout) => {}
2106×
            Err(mpsc::RecvTimeoutError::Disconnected) => break,
211
        }
2129×
        if !timed_out && !cancelled && sink.cancelled() {
2130×
            cancelled = true;
2140×
            let _ = child.kill();
2150×
            sink.record(Record { t: t(&started), stream: "flow".into(), text: "stopped: the run was cancelled".into() });
2169×
        }
2179×
        if Instant::now() >= deadline && !timed_out && !cancelled {
2180×
            timed_out = true;
2190×
            let _ = child.kill();
2200×
            sink.record(Record {
2210×
                t: t(&started),
2220×
                stream: "flow".into(),
2230×
                text: format!("stopped at the time limit of {} s", timeout.as_secs()),
2240×
            });
2259×
        }
226
    }
2276×
    let status = child.wait()?;
22812×
    for r in readers {
22912×
        let _ = r.join();
23012×
    }
2316×
    let _ = std::fs::remove_file(&spec_path);
2326×
    let exit_code = status.code();
2336×
    if exit_code.is_none() && !timed_out && !cancelled && !failed_to_start {
2340×
        sink.record(Record {
2350×
            t: t(&started),
2360×
            stream: "flow".into(),
2370×
            text: format!("the WASM helper ended without an exit code ({status})"),
2380×
        });
2396×
    }
240
    Ok(Outcome {
2416×
        exit_code: if timed_out || cancelled { None } else { exit_code },
2426×
        duration_ms: t(&started),
2436×
        environment,
2446×
        timed_out,
2456×
        cancelled,
2466×
        peak_memory_mib,
2476×
        pruned: Vec::new(),
248
    })
24912×
}
250
251
/// Whether modules run in a thread of this process rather than in a child
252
/// `flow __wasm`: on Android, where the runner is a library inside an app
253
/// that cannot start itself as a command (oxygenhub-android#8), and
254
/// anywhere `FLOW_WASM_IN_PROCESS` is set (how the tests check this way).
25512×
pub fn in_process() -> bool {
25612×
    cfg!(target_os = "android") || std::env::var_os("FLOW_WASM_IN_PROCESS").is_some()
25712×
}
258
259
#[cfg(feature = "wasm")]
260
/// The most of a module's output kept before it is read, per stream.
261
const PIPE_BYTES: usize = 64 << 20;
262
263
#[cfg(feature = "wasm")]
264
/// Run the spec's modules in a thread of this process, streaming their
265
/// output as it comes: the pipes are read every 100 ms, and a time limit
266
/// or a cancelled run interrupts the module through wasmtime's epoch.
2679×
fn run_in_thread(spec: &Spec, environment: Environment, timeout: Duration, sink: &mut dyn Sink) -> Result<Outcome> {
268
    use wasmtime_wasi::p2::pipe::MemoryOutputPipe;
2699×
    let mut config = wasmtime::Config::new();
2709×
    config.epoch_interruption(true);
2719×
    let engine = ww(wasmtime::Engine::new(&config), || "start wasmtime".into())?;
2729×
    let (stdout, stderr) = (MemoryOutputPipe::new(PIPE_BYTES), MemoryOutputPipe::new(PIPE_BYTES));
2739×
    let (notes_tx, notes) = mpsc::channel::<String>();
2749×
    let io = Io::Pipes { stdout: stdout.clone(), stderr: stderr.clone(), notes: notes_tx };
2759×
    let started = Instant::now();
27623×
    let t = |s: &Instant| s.elapsed().as_millis() as u64;
2779×
    let deadline = started + timeout;
2789×
    let (mut timed_out, mut cancelled) = (false, false);
279
    // Bytes of each stream already passed on, and the part line not ended yet.
2809×
    let mut said = [(0usize, Vec::<u8>::new()), (0usize, Vec::<u8>::new())];
28122×
    let mut pass_on = |sink: &mut dyn Sink, last: bool| {
28244×
        for (i, pipe) in [&stdout, &stderr].into_iter().enumerate() {
28344×
            let all = pipe.contents();
28444×
            let (offset, partial) = &mut said[i];
28544×
            partial.extend_from_slice(&all[*offset..]);
28644×
            *offset = all.len();
28759×
            while let Some(n) = partial.iter().position(|b| *b == b'\n') {
2884×
                let line: Vec<u8> = partial.drain(..=n).collect();
2894×
                let text = String::from_utf8_lossy(&line[..line.len() - 1]).into_owned();
2904×
                sink.record(Record { t: t(&started), stream: if i == 0 { "out" } else { "err" }.into(), text });
291
            }
29244×
            if last && !partial.is_empty() {
2931×
                let text = String::from_utf8_lossy(partial).into_owned();
2941×
                partial.clear();
2951×
                sink.record(Record { t: t(&started), stream: if i == 0 { "out" } else { "err" }.into(), text });
29643×
            }
297
        }
29829×
        while let Ok(note) = notes.try_recv() {
2997×
            sink.record(Record { t: t(&started), stream: "flow".into(), text: note });
3007×
        }
30122×
    };
3029×
    let ran = std::thread::scope(|scope| {
3039×
        let worker = scope.spawn(|| run_modules(&engine, spec, &io));
30422×
        while !worker.is_finished() {
30513×
            std::thread::sleep(Duration::from_millis(100));
30613×
            pass_on(sink, false);
307
            // Once stopped, keep moving the epoch: a store made after the
308
            // first increment (its module still compiling when the time
309
            // limit or the cancel came) sets its deadline one epoch past
310
            // that, and a single increment would never reach it (#93).
31113×
            if timed_out || cancelled {
3122×
                engine.increment_epoch();
31311×
            }
31413×
            if !timed_out && !cancelled && sink.cancelled() {
3151×
                cancelled = true;
3161×
                engine.increment_epoch();
3171×
                sink.record(Record { t: t(&started), stream: "flow".into(), text: "stopped: the run was cancelled".into() });
31812×
            }
31913×
            if Instant::now() >= deadline && !timed_out && !cancelled {
3201×
                timed_out = true;
3211×
                engine.increment_epoch();
3221×
                sink.record(Record {
3231×
                    t: t(&started),
3241×
                    stream: "flow".into(),
3251×
                    text: format!("stopped at the time limit of {} s", timeout.as_secs()),
3261×
                });
32712×
            }
328
        }
3299×
        worker.join().unwrap_or_else(|_| Err(anyhow::anyhow!("the WASM worker panicked")))
3309×
    });
3319×
    pass_on(sink, true);
3329×
    let (exit_code, peak) = match ran {
3339×
        Ok((code, peak)) => (Some(code), Some(peak)),
3340×
        Err(e) => {
3350×
            sink.record(Record { t: t(&started), stream: "flow".into(), text: format!("the WASM tier could not run the job: {e:#}") });
3360×
            (Some(125), None)
337
        }
338
    };
339
    Ok(Outcome {
3409×
        exit_code: if timed_out || cancelled { None } else { exit_code },
3419×
        duration_ms: t(&started),
3429×
        environment,
3439×
        timed_out,
3449×
        cancelled,
3459×
        peak_memory_mib: peak.map(|b| (b as u64).div_ceil(1 << 20)),
3469×
        pruned: Vec::new(),
347
    })
3489×
}
349
350
#[cfg(feature = "wasm")]
351
/// `flow __wasm SPEC`: run the spec's modules in turn, each in a fresh
352
/// instance; the exit code is the first non-zero one, after all have run,
353
/// so one failing module does not hide another's results.
3540×
pub fn wasm_main(spec_path: &Path) -> i32 {
3550×
    let ran = (|| {
3560×
        let spec: Spec = serde_json::from_slice(&std::fs::read(spec_path)?).context("read the spec")?;
3570×
        let engine = ww(wasmtime::Engine::new(&wasmtime::Config::new()), || "start wasmtime".into())?;
3580×
        run_modules(&engine, &spec, &Io::Inherit)
359
    })();
3600×
    match ran {
3610×
        Ok((code, _)) => code,
3620×
        Err(e) => {
3630×
            eprintln!("{FAILED}{e:#}");
3640×
            125
365
        }
366
    }
3670×
}
368
369
/// `flow __wasm SPEC` in a build without the tier: says so.
370
#[cfg(not(feature = "wasm"))]
371
pub fn wasm_main(_spec_path: &Path) -> i32 {
372
    eprintln!("{FAILED}{WITHOUT}");
373
    125
374
}
375
376
#[cfg(not(feature = "wasm"))]
377
fn run_in_thread(_spec: &Spec, _environment: Environment, _timeout: Duration, _sink: &mut dyn Sink) -> Result<Outcome> {
378
    bail!(WITHOUT)
379
}
380
381
#[cfg(feature = "wasm")]
382
/// wasmtime's own error, as this crate's: with its whole chain of causes.
38368×
fn ww<T>(r: wasmtime::Result<T>, what: impl FnOnce() -> String) -> Result<T> {
38468×
    r.map_err(|e| anyhow::anyhow!("{}: {e:#}", what()))
38568×
}
386
387
#[cfg(feature = "wasm")]
388
/// Where a module's output and the tier's notes go: inherited by the
389
/// helper process (notes on stderr, marked), or pipes the runner reads.
390
enum Io {
391
    Inherit,
392
    Pipes {
393
        stdout: wasmtime_wasi::p2::pipe::MemoryOutputPipe,
394
        stderr: wasmtime_wasi::p2::pipe::MemoryOutputPipe,
395
        notes: mpsc::Sender<String>,
396
    },
397
}
398
399
#[cfg(feature = "wasm")]
400
impl Io {
4017×
    fn note(&self, text: String) {
4027×
        match self {
4030×
            Io::Inherit => eprintln!("{NOTE}{text}"),
4047×
            Io::Pipes { notes, .. } => {
4057×
                let _ = notes.send(text);
4067×
            }
407
        }
4087×
    }
409
}
410
411
#[cfg(feature = "wasm")]
412
/// The sandbox's memory limit, and the most a module asked for within it.
413
struct Limits {
414
    memory: usize,
415
    peak: usize,
416
}
417
418
#[cfg(feature = "wasm")]
419
impl wasmtime::ResourceLimiter for Limits {
4209×
    fn memory_growing(&mut self, _current: usize, desired: usize, maximum: Option<usize>) -> wasmtime::Result<bool> {
4219×
        let ok = desired <= self.memory && maximum.is_none_or(|m| desired <= m);
4229×
        if ok {
4238×
            self.peak = self.peak.max(desired);
4248×
        }
4259×
        Ok(ok)
4269×
    }
427
4280×
    fn table_growing(&mut self, _current: usize, desired: usize, maximum: Option<usize>) -> wasmtime::Result<bool> {
4290×
        Ok(maximum.is_none_or(|m| desired <= m))
4300×
    }
431
}
432
433
#[cfg(feature = "wasm")]
434
struct State {
435
    wasi: wasmtime_wasi::p1::WasiP1Ctx,
436
    limits: Limits,
437
}
438
439
#[cfg(feature = "wasm")]
440
/// Run each module in turn: the first non-zero exit code after all ran,
441
/// and the most memory any of them held, in bytes.
4429×
fn run_modules(engine: &wasmtime::Engine, spec: &Spec, io: &Io) -> Result<(i32, usize)> {
4439×
    let mut linker: wasmtime::Linker<State> = wasmtime::Linker::new(engine);
4449×
    ww(wasmtime_wasi::p1::add_to_linker_sync(&mut linker, |s: &mut State| &mut s.wasi), || "link WASI".into())?;
4459×
    let (mut first_failure, mut peak) = (0, 0);
44610×
    for path in &spec.modules {
44710×
        if spec.modules.len() > 1 {
4482×
            io.note(format!("running {path}"));
4498×
        }
45010×
        let (code, held) = run_module(engine, &linker, spec, path, io)?;
45110×
        peak = peak.max(held);
45210×
        if spec.modules.len() > 1 {
4532×
            io.note(format!("{path}: exit {code}"));
4548×
        }
45510×
        if first_failure == 0 && code != 0 {
4564×
            first_failure = code;
4576×
        }
458
    }
4599×
    Ok((first_failure, peak))
4609×
}
461
462
#[cfg(feature = "wasm")]
46310×
fn run_module(engine: &wasmtime::Engine, linker: &wasmtime::Linker<State>, spec: &Spec, path: &str, io: &Io) -> Result<(i32, usize)> {
464
    use wasmtime_wasi::{FsPerms, WasiCtxBuilder};
46510×
    let module = ww(wasmtime::Module::from_file(engine, spec.workspace.join(path)), || format!("module {path} does not load"))?;
46610×
    let name = path.rsplit('/').next().unwrap_or(path);
46710×
    let mut argv = vec![name.to_string()];
46810×
    argv.extend(spec.args.iter().cloned());
46910×
    let mut wasi = WasiCtxBuilder::new();
47010×
    match io {
4710×
        Io::Inherit => {
4720×
            wasi.inherit_stdout().inherit_stderr();
4730×
        }
47410×
        Io::Pipes { stdout, stderr, .. } => {
47510×
            wasi.stdout(stdout.clone()).stderr(stderr.clone());
47610×
        }
477
    }
47810×
    wasi.args(&argv).envs(&spec.env).env("PWD", "/src").allow_blocking_current_thread(true);
47920×
    let open = |wasi: &mut WasiCtxBuilder, dir: &Path, at: &str| {
48020×
        ww(wasi.preopened_dir(dir, at, FsPerms::ReadWrite).map(|_| ()), || format!("open {} as {at}", dir.display()))
48120×
    };
48210×
    open(&mut wasi, &spec.workspace, "/src")?;
483
    // Relative paths resolve against the workspace too.
48410×
    open(&mut wasi, &spec.workspace, ".")?;
48510×
    for (dir, target) in &spec.preopens {
4860×
        open(&mut wasi, dir, target)?;
487
    }
48810×
    let limits = Limits { memory: spec.memory_mib as usize * 1024 * 1024, peak: 0 };
48910×
    let mut store = wasmtime::Store::new(engine, State { wasi: wasi.build_p1(), limits });
49010×
    store.limiter(|s| &mut s.limits);
491
    // With epoch interruption on (in process), one tick from now traps:
492
    // the runner ticks only to stop the module.
49310×
    store.set_epoch_deadline(1);
49410×
    let instance = ww(linker.instantiate(&mut store, &module), || format!("module {path} cannot start here"))?;
49510×
    let start =
49610×
        ww(instance.get_typed_func::<(), ()>(&mut store, "_start"), || format!("module {path} has no _start: is it a WASI command?"))?;
49710×
    let code = match start.call(&mut store, ()) {
4986×
        Ok(()) => 0,
4994×
        Err(e) => match e.downcast_ref::<wasmtime_wasi::I32Exit>() {
5001×
            Some(exit) => exit.0,
501
            None => {
5023×
                io.note(format!("{path} trapped: {e:#}"));
5033×
                TRAPPED
504
            }
505
        },
506
    };
50710×
    Ok((code, store.data().limits.peak))
50810×
}
509
510
#[cfg(all(test, feature = "wasm"))]
511
mod tests {
512
    use super::*;
513
514
    /// A spec running `text` (WAT) as the only module, with 64 MiB.
5153×
    fn spec_of(dir: &Path, text: &str) -> Spec {
5163×
        std::fs::write(dir.join("m.wasm"), wat::parse_str(text).unwrap()).unwrap();
5173×
        Spec { workspace: dir.to_path_buf(), modules: vec!["m.wasm".into()], args: vec![], env: vec![], memory_mib: 64, preopens: vec![] }
5183×
    }
519
5203×
    fn env() -> Environment {
5213×
        Environment { runtime: "wasm".into(), parts: vec![], rootfs: String::new(), argv: vec![], env: vec![] }
5223×
    }
523
524
    /// Records, and whether to say the run was cancelled.
525
    struct Collect(Vec<Record>, bool);
526
    impl Sink for Collect {
5276×
        fn record(&mut self, r: Record) {
5286×
            self.0.push(r);
5296×
        }
5305×
        fn cancelled(&mut self) -> bool {
5315×
            self.1
5325×
        }
533
    }
534
535
    const PRINT: &str = r#"(module
536
      (import "wasi_snapshot_preview1" "fd_write" (func $w (param i32 i32 i32 i32) (result i32)))
537
      (memory (export "memory") 2)
538
      (data (i32.const 16) "line one\nline two")
539
      (func (export "_start")
540
        (i32.store (i32.const 0) (i32.const 16)) (i32.store (i32.const 4) (i32.const 17))
541
        (drop (call $w (i32.const 1) (i32.const 0) (i32.const 1) (i32.const 8)))))"#;
542
543
    const SPIN: &str = r#"(module (func (export "_start") (loop $l (br $l))))"#;
544
545
    #[test]
5461×
    fn in_process_output_streams_as_lines() {
5471×
        let dir = tempfile::tempdir().unwrap();
5481×
        let mut sink = Collect(vec![], false);
5491×
        let o = run_in_thread(&spec_of(dir.path(), PRINT), env(), Duration::from_secs(30), &mut sink).unwrap();
5501×
        assert_eq!(o.exit_code, Some(0));
5512×
        let out: Vec<&str> = sink.0.iter().filter(|r| r.stream == "out").map(|r| r.text.as_str()).collect();
5521×
        assert_eq!(out, ["line one", "line two"], "a last line without its newline still comes");
5531×
        assert_eq!(o.peak_memory_mib, Some(1), "two 64 KiB pages, in whole MiB");
5541×
    }
555
556
    #[test]
5571×
    fn in_process_stops_at_the_time_limit_and_when_cancelled() {
5581×
        let dir = tempfile::tempdir().unwrap();
5591×
        let mut sink = Collect(vec![], false);
5601×
        let started = Instant::now();
5611×
        let o = run_in_thread(&spec_of(dir.path(), SPIN), env(), Duration::from_millis(300), &mut sink).unwrap();
5621×
        assert!(o.timed_out && o.exit_code.is_none(), "{:?}", sink.0);
5631×
        assert!(started.elapsed() < Duration::from_secs(5), "the endless loop was interrupted");
5641×
        let mut sink = Collect(vec![], true);
5651×
        let o = run_in_thread(&spec_of(dir.path(), SPIN), env(), Duration::from_secs(30), &mut sink).unwrap();
5661×
        assert!(o.cancelled && o.exit_code.is_none());
5671×
        assert!(sink.0.iter().any(|r| r.text == "stopped: the run was cancelled"));
5681×
    }
569
570
    #[test]
5711×
    fn globs_match_names_with_stars() {
5721×
        assert!(wildcard("*.wasm", "oxy_core-1a2b.wasm"));
5731×
        assert!(wildcard("oxy_core-*.wasm", "oxy_core-1a2b.wasm"));
5741×
        assert!(!wildcard("oxy_core-*.wasm", "oxy_core-1a2b.d"));
5751×
        assert!(wildcard("a*b*c", "aXbYc") && !wildcard("a*b*c", "aXcYb"));
5761×
        assert!(wildcard("plain.wasm", "plain.wasm") && !wildcard("plain.wasm", "other.wasm"));
5771×
    }
578
579
    #[test]
5801×
    fn modules_are_files_of_the_job() {
5811×
        let dir = tempfile::tempdir().unwrap();
5821×
        let deps = dir.path().join("target/wasm32-wasip1/debug/deps");
5831×
        std::fs::create_dir_all(&deps).unwrap();
5843×
        for f in ["b-2.wasm", "a-1.wasm", "a-1.d"] {
5853×
            std::fs::write(deps.join(f), b"").unwrap();
5863×
        }
5871×
        let ws = dir.path();
5881×
        assert_eq!(
5891×
            modules(ws, "/src/target/wasm32-wasip1/debug/deps/*.wasm").unwrap(),
590
            ["target/wasm32-wasip1/debug/deps/a-1.wasm", "target/wasm32-wasip1/debug/deps/b-2.wasm"]
591
        );
5921×
        assert_eq!(modules(ws, "./target/wasm32-wasip1/debug/deps/b-2.wasm").unwrap(), ["target/wasm32-wasip1/debug/deps/b-2.wasm"]);
5931×
        assert!(modules(ws, "target/wasm32-wasip1/debug/deps/*.so").unwrap_err().to_string().contains("no module matches"));
5941×
        assert!(modules(ws, "../etc/passwd").is_err());
5951×
        assert!(modules(ws, "/etc/x.wasm").is_err());
5961×
        assert!(modules(ws, "target/*/debug/a.wasm").unwrap_err().to_string().contains("only the file name"));
5971×
    }
598
}