OXYGENHUB

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

347 of 410 lines run, 79 of 133 functions · run #197 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
/// A line the helper writes to say something about the job itself (which
27
/// module starts), shown as the runner's own note rather than the job's.
28
const NOTE: &str = "\u{1e}N ";
29
30
/// The exit code for a module that trapped (crashed) rather than exiting.
31
const TRAPPED: i32 = 134;
32
33
/// What the helper runs, written by the runner for `flow __wasm`.
34
#[derive(Serialize, Deserialize)]
35
struct Spec {
36
    /// The job's files, preopened as `/src` (and as the current directory).
37
    workspace: PathBuf,
38
    /// The modules to run in turn, relative to the workspace.
39
    modules: Vec<String>,
40
    /// The arguments after the module.
41
    args: Vec<String>,
42
    env: Vec<(String, String)>,
43
    memory_mib: u32,
44
    /// The job's mounts: a directory of the runner and where the module sees it.
45
    preopens: Vec<(PathBuf, String)>,
46
}
47
48
/// The receipt environment of a job in this tier: the runtime and its
49
/// version, this runner's binary, the memory allowed, and every module run
50
/// with its hash, so the receipt says exactly what code ran.
5112×
pub fn environment(sandbox: &Sandbox, job: &Job, modules: &[(String, String)]) -> Environment {
52
    Environment {
5312×
        runtime: "wasm".into(),
5412×
        parts: vec![
5512×
            ("wasmtime".into(), WASMTIME.into()),
5612×
            ("flow".into(), self_hash().to_string()),
5712×
            ("memory_mib".into(), sandbox.memory_mib.to_string()),
58
        ]
5912×
        .into_iter()
6014×
        .chain(modules.iter().map(|(path, hash)| (format!("module {path}"), hash.clone())))
6112×
        .collect(),
6212×
        rootfs: String::new(),
6312×
        argv: job.exec.clone(),
6424×
        env: job.env.iter().map(|(k, v)| (k.clone(), v.clone())).collect(),
65
    }
6612×
}
67
68
/// The modules `program` names in `workspace`: one file, or the files a
69
/// glob matches (`*` in its last part only), sorted. A path may start with
70
/// `/src/`, where the module sees the workspace.
7118×
pub fn modules(workspace: &Path, program: &str) -> Result<Vec<String>> {
7218×
    let rel = program.strip_prefix("/src/").unwrap_or(program).trim_start_matches("./");
7346×
    if rel.is_empty() || rel.starts_with('/') || rel.split('/').any(|p| p == "..") {
742×
        bail!("module {program:?}: give a path inside the job's files, such as target/wasm32-wasip1/debug/app.wasm");
7516×
    }
7616×
    let (dir, last) = rel.rsplit_once('/').unwrap_or(("", rel));
7716×
    if dir.contains('*') {
781×
        bail!("module {program:?}: only the file name may hold `*`");
7915×
    }
8015×
    if !last.contains('*') {
8111×
        if !workspace.join(rel).is_file() {
820×
            bail!("module {rel} is not among the job's files (its scope or an earlier job's artifact)");
8311×
        }
8411×
        return Ok(vec![rel.to_string()]);
854×
    }
864×
    let mut found: Vec<String> = std::fs::read_dir(workspace.join(dir))
874×
        .with_context(|| format!("module {program:?}: the directory {dir} is not among the job's files"))?
8810×
        .filter_map(|e| e.ok())
8910×
        .filter(|e| e.file_type().is_ok_and(|t| t.is_file()))
9010×
        .filter_map(|e| e.file_name().into_string().ok())
9110×
        .filter(|name| wildcard(last, name))
926×
        .map(|name| if dir.is_empty() { name } else { format!("{dir}/{name}") })
934×
        .collect();
944×
    found.sort();
954×
    if found.is_empty() {
961×
        bail!("no module matches {program:?}");
973×
    }
983×
    Ok(found)
9918×
}
100
101
/// Whether `name` matches `pattern`, where `*` stands for any run of characters.
10217×
fn wildcard(pattern: &str, name: &str) -> bool {
10317×
    let parts: Vec<&str> = pattern.split('*').collect();
10417×
    let (first, last) = (parts[0], parts[parts.len() - 1]);
10517×
    if parts.len() == 1 {
1062×
        return pattern == name;
10715×
    }
10815×
    if !name.starts_with(first) || !name[first.len()..].ends_with(last) || name.len() < first.len() + last.len() {
1096×
        return false;
1109×
    }
1119×
    let mut rest = &name[first.len()..name.len() - last.len()];
1129×
    for middle in &parts[1..parts.len() - 1] {
1131×
        match rest.find(middle) {
1141×
            Some(i) => rest = &rest[i + middle.len()..],
1150×
            None => return false,
116
        }
117
    }
1189×
    true
11917×
}
120
121
/// Run `job` in this tier with its files in `workspace` (as the syscall
122
/// tier lays them out); `scratch` holds the helper's spec and the job's
123
/// mounts.
12412×
pub fn run(sandbox: &Sandbox, job: &Job, workspace: &Path, scratch: &Path, timeout: Duration, sink: &mut dyn Sink) -> Result<Outcome> {
12512×
    if !job.secrets.is_empty() || !job.identities.is_empty() {
1260×
        bail!("job {}: the WASM tier takes no secrets or identities yet; run it in a microvm or syscall sandbox", job.name);
12712×
    }
12812×
    let Some(program) = job.exec.first() else { bail!("job {}: exec names no module", job.name) };
12912×
    let modules = modules(workspace, program)?;
13012×
    let hashed: Vec<(String, String)> =
13114×
        modules.iter().map(|m| Ok((m.clone(), crate::home::hash_file(&workspace.join(m))?))).collect::<Result<_>>()?;
13212×
    let environment = environment(sandbox, job, &hashed);
13312×
    let mut preopens = Vec::new();
13412×
    let mut disks = 0;
13512×
    for (i, m) in job.mounts.iter().enumerate() {
1360×
        let dir = match m.source {
137
            MountSource::Disk => {
1380×
                disks += 1;
1390×
                crate::syscall::disk_dir(scratch, disks - 1)
140
            }
1410×
            MountSource::Tmpfs => scratch.join(format!("tmpfs-{i}")),
142
        };
1430×
        std::fs::create_dir_all(&dir)?;
1440×
        preopens.push((dir, m.target.clone()));
145
    }
14612×
    let spec = Spec {
14712×
        workspace: workspace.to_path_buf(),
14812×
        modules,
14912×
        args: job.exec[1..].to_vec(),
15024×
        env: job.env.iter().map(|(k, v)| (k.clone(), v.clone())).collect(),
15112×
        memory_mib: sandbox.memory_mib,
15212×
        preopens,
153
    };
15412×
    if in_process() {
1556×
        return run_in_thread(&spec, environment, timeout, sink);
1566×
    }
1576×
    let spec_path = scratch.join("wasm.json");
1586×
    std::fs::write(&spec_path, serde_json::to_vec(&spec)?)?;
159
1606×
    let started = Instant::now();
1616×
    let mut child = std::process::Command::new(crate::tier::self_exe())
1626×
        .arg("__wasm")
1636×
        .arg(&spec_path)
1646×
        .env_clear()
1656×
        .current_dir(scratch)
1666×
        .stdin(std::process::Stdio::null())
1676×
        .stdout(std::process::Stdio::piped())
1686×
        .stderr(std::process::Stdio::piped())
1696×
        .spawn()
1706×
        .context("start the WASM helper")?;
1716×
    let (tx, rx) = mpsc::channel::<(bool, Vec<u8>)>();
1726×
    let out = child.stdout.take().unwrap();
1736×
    let err = child.stderr.take().unwrap();
1746×
    let tx2 = tx.clone();
1756×
    let readers = [std::thread::spawn(move || lines(out, true, tx)), std::thread::spawn(move || lines(err, false, tx2))];
17615×
    let t = |s: &Instant| s.elapsed().as_millis() as u64;
1776×
    let (mut timed_out, mut cancelled, mut failed_to_start) = (false, false, false);
1786×
    let mut peak_memory_mib = None;
1796×
    let deadline = started + timeout;
180
    loop {
18115×
        if let Some(m) = crate::facts::peak_rss_mib(child.id()) {
18213×
            peak_memory_mib = Some(m);
18313×
        }
18415×
        match rx.recv_timeout(Duration::from_millis(200)) {
1859×
            Ok((stdout, line)) => {
1869×
                let text = String::from_utf8_lossy(&line).into_owned();
1879×
                let rec = |stream: &str, text: &str| Record { t: t(&started), stream: stream.into(), text: text.into() };
1889×
                if let Some(note) = text.strip_prefix(NOTE).filter(|_| !stdout) {
1895×
                    sink.record(rec("flow", note));
1905×
                } else if let Some(why) = text.strip_prefix(FAILED).filter(|_| !stdout) {
1910×
                    failed_to_start = true;
1920×
                    sink.record(rec("flow", &format!("the WASM tier could not run the job: {why}")));
1930×
                } else {
1944×
                    sink.record(rec(if stdout { "out" } else { "err" }, &text));
195
                }
196
            }
1970×
            Err(mpsc::RecvTimeoutError::Timeout) => {}
1986×
            Err(mpsc::RecvTimeoutError::Disconnected) => break,
199
        }
2009×
        if !timed_out && !cancelled && sink.cancelled() {
2010×
            cancelled = true;
2020×
            let _ = child.kill();
2030×
            sink.record(Record { t: t(&started), stream: "flow".into(), text: "stopped: the run was cancelled".into() });
2049×
        }
2059×
        if Instant::now() >= deadline && !timed_out && !cancelled {
2060×
            timed_out = true;
2070×
            let _ = child.kill();
2080×
            sink.record(Record {
2090×
                t: t(&started),
2100×
                stream: "flow".into(),
2110×
                text: format!("stopped at the time limit of {} s", timeout.as_secs()),
2120×
            });
2139×
        }
214
    }
2156×
    let status = child.wait()?;
21612×
    for r in readers {
21712×
        let _ = r.join();
21812×
    }
2196×
    let _ = std::fs::remove_file(&spec_path);
2206×
    let exit_code = status.code();
2216×
    if exit_code.is_none() && !timed_out && !cancelled && !failed_to_start {
2220×
        sink.record(Record {
2230×
            t: t(&started),
2240×
            stream: "flow".into(),
2250×
            text: format!("the WASM helper ended without an exit code ({status})"),
2260×
        });
2276×
    }
228
    Ok(Outcome {
2296×
        exit_code: if timed_out || cancelled { None } else { exit_code },
2306×
        duration_ms: t(&started),
2316×
        environment,
2326×
        timed_out,
2336×
        cancelled,
2346×
        peak_memory_mib,
2356×
        pruned: Vec::new(),
236
    })
23712×
}
238
239
/// Whether modules run in a thread of this process rather than in a child
240
/// `flow __wasm`: on Android, where the runner is a library inside an app
241
/// that cannot start itself as a command (oxygenhub-android#8), and
242
/// anywhere `FLOW_WASM_IN_PROCESS` is set (how the tests check this way).
24312×
pub fn in_process() -> bool {
24412×
    cfg!(target_os = "android") || std::env::var_os("FLOW_WASM_IN_PROCESS").is_some()
24512×
}
246
247
/// The most of a module's output kept before it is read, per stream.
248
const PIPE_BYTES: usize = 64 << 20;
249
250
/// Run the spec's modules in a thread of this process, streaming their
251
/// output as it comes: the pipes are read every 100 ms, and a time limit
252
/// or a cancelled run interrupts the module through wasmtime's epoch.
2539×
fn run_in_thread(spec: &Spec, environment: Environment, timeout: Duration, sink: &mut dyn Sink) -> Result<Outcome> {
254
    use wasmtime_wasi::p2::pipe::MemoryOutputPipe;
2559×
    let mut config = wasmtime::Config::new();
2569×
    config.epoch_interruption(true);
2579×
    let engine = ww(wasmtime::Engine::new(&config), || "start wasmtime".into())?;
2589×
    let (stdout, stderr) = (MemoryOutputPipe::new(PIPE_BYTES), MemoryOutputPipe::new(PIPE_BYTES));
2599×
    let (notes_tx, notes) = mpsc::channel::<String>();
2609×
    let io = Io::Pipes { stdout: stdout.clone(), stderr: stderr.clone(), notes: notes_tx };
2619×
    let started = Instant::now();
26223×
    let t = |s: &Instant| s.elapsed().as_millis() as u64;
2639×
    let deadline = started + timeout;
2649×
    let (mut timed_out, mut cancelled) = (false, false);
265
    // Bytes of each stream already passed on, and the part line not ended yet.
2669×
    let mut said = [(0usize, Vec::<u8>::new()), (0usize, Vec::<u8>::new())];
26724×
    let mut pass_on = |sink: &mut dyn Sink, last: bool| {
26848×
        for (i, pipe) in [&stdout, &stderr].into_iter().enumerate() {
26948×
            let all = pipe.contents();
27048×
            let (offset, partial) = &mut said[i];
27148×
            partial.extend_from_slice(&all[*offset..]);
27248×
            *offset = all.len();
27359×
            while let Some(n) = partial.iter().position(|b| *b == b'\n') {
2744×
                let line: Vec<u8> = partial.drain(..=n).collect();
2754×
                let text = String::from_utf8_lossy(&line[..line.len() - 1]).into_owned();
2764×
                sink.record(Record { t: t(&started), stream: if i == 0 { "out" } else { "err" }.into(), text });
277
            }
27848×
            if last && !partial.is_empty() {
2791×
                let text = String::from_utf8_lossy(partial).into_owned();
2801×
                partial.clear();
2811×
                sink.record(Record { t: t(&started), stream: if i == 0 { "out" } else { "err" }.into(), text });
28247×
            }
283
        }
28431×
        while let Ok(note) = notes.try_recv() {
2857×
            sink.record(Record { t: t(&started), stream: "flow".into(), text: note });
2867×
        }
28724×
    };
2889×
    let ran = std::thread::scope(|scope| {
2899×
        let worker = scope.spawn(|| run_modules(&engine, spec, &io));
29024×
        while !worker.is_finished() {
29115×
            std::thread::sleep(Duration::from_millis(100));
29215×
            pass_on(sink, false);
29315×
            if !timed_out && !cancelled && sink.cancelled() {
2941×
                cancelled = true;
2951×
                engine.increment_epoch();
2961×
                sink.record(Record { t: t(&started), stream: "flow".into(), text: "stopped: the run was cancelled".into() });
29714×
            }
29815×
            if Instant::now() >= deadline && !timed_out && !cancelled {
2991×
                timed_out = true;
3001×
                engine.increment_epoch();
3011×
                sink.record(Record {
3021×
                    t: t(&started),
3031×
                    stream: "flow".into(),
3041×
                    text: format!("stopped at the time limit of {} s", timeout.as_secs()),
3051×
                });
30614×
            }
307
        }
3089×
        worker.join().unwrap_or_else(|_| Err(anyhow::anyhow!("the WASM worker panicked")))
3099×
    });
3109×
    pass_on(sink, true);
3119×
    let (exit_code, peak) = match ran {
3129×
        Ok((code, peak)) => (Some(code), Some(peak)),
3130×
        Err(e) => {
3140×
            sink.record(Record { t: t(&started), stream: "flow".into(), text: format!("the WASM tier could not run the job: {e:#}") });
3150×
            (Some(125), None)
316
        }
317
    };
318
    Ok(Outcome {
3199×
        exit_code: if timed_out || cancelled { None } else { exit_code },
3209×
        duration_ms: t(&started),
3219×
        environment,
3229×
        timed_out,
3239×
        cancelled,
3249×
        peak_memory_mib: peak.map(|b| (b as u64).div_ceil(1 << 20)),
3259×
        pruned: Vec::new(),
326
    })
3279×
}
328
329
/// `flow __wasm SPEC`: run the spec's modules in turn, each in a fresh
330
/// instance; the exit code is the first non-zero one, after all have run,
331
/// so one failing module does not hide another's results.
3320×
pub fn wasm_main(spec_path: &Path) -> i32 {
3330×
    let ran = (|| {
3340×
        let spec: Spec = serde_json::from_slice(&std::fs::read(spec_path)?).context("read the spec")?;
3350×
        let engine = ww(wasmtime::Engine::new(&wasmtime::Config::new()), || "start wasmtime".into())?;
3360×
        run_modules(&engine, &spec, &Io::Inherit)
337
    })();
3380×
    match ran {
3390×
        Ok((code, _)) => code,
3400×
        Err(e) => {
3410×
            eprintln!("{FAILED}{e:#}");
3420×
            125
343
        }
344
    }
3450×
}
346
347
/// wasmtime's own error, as this crate's: with its whole chain of causes.
34868×
fn ww<T>(r: wasmtime::Result<T>, what: impl FnOnce() -> String) -> Result<T> {
34968×
    r.map_err(|e| anyhow::anyhow!("{}: {e:#}", what()))
35068×
}
351
352
/// Where a module's output and the tier's notes go: inherited by the
353
/// helper process (notes on stderr, marked), or pipes the runner reads.
354
enum Io {
355
    Inherit,
356
    Pipes {
357
        stdout: wasmtime_wasi::p2::pipe::MemoryOutputPipe,
358
        stderr: wasmtime_wasi::p2::pipe::MemoryOutputPipe,
359
        notes: mpsc::Sender<String>,
360
    },
361
}
362
363
impl Io {
3647×
    fn note(&self, text: String) {
3657×
        match self {
3660×
            Io::Inherit => eprintln!("{NOTE}{text}"),
3677×
            Io::Pipes { notes, .. } => {
3687×
                let _ = notes.send(text);
3697×
            }
370
        }
3717×
    }
372
}
373
374
/// The sandbox's memory limit, and the most a module asked for within it.
375
struct Limits {
376
    memory: usize,
377
    peak: usize,
378
}
379
380
impl wasmtime::ResourceLimiter for Limits {
3819×
    fn memory_growing(&mut self, _current: usize, desired: usize, maximum: Option<usize>) -> wasmtime::Result<bool> {
3829×
        let ok = desired <= self.memory && maximum.is_none_or(|m| desired <= m);
3839×
        if ok {
3848×
            self.peak = self.peak.max(desired);
3858×
        }
3869×
        Ok(ok)
3879×
    }
388
3890×
    fn table_growing(&mut self, _current: usize, desired: usize, maximum: Option<usize>) -> wasmtime::Result<bool> {
3900×
        Ok(maximum.is_none_or(|m| desired <= m))
3910×
    }
392
}
393
394
struct State {
395
    wasi: wasmtime_wasi::p1::WasiP1Ctx,
396
    limits: Limits,
397
}
398
399
/// Run each module in turn: the first non-zero exit code after all ran,
400
/// and the most memory any of them held, in bytes.
4019×
fn run_modules(engine: &wasmtime::Engine, spec: &Spec, io: &Io) -> Result<(i32, usize)> {
4029×
    let mut linker: wasmtime::Linker<State> = wasmtime::Linker::new(engine);
4039×
    ww(wasmtime_wasi::p1::add_to_linker_sync(&mut linker, |s: &mut State| &mut s.wasi), || "link WASI".into())?;
4049×
    let (mut first_failure, mut peak) = (0, 0);
40510×
    for path in &spec.modules {
40610×
        if spec.modules.len() > 1 {
4072×
            io.note(format!("running {path}"));
4088×
        }
40910×
        let (code, held) = run_module(engine, &linker, spec, path, io)?;
41010×
        peak = peak.max(held);
41110×
        if spec.modules.len() > 1 {
4122×
            io.note(format!("{path}: exit {code}"));
4138×
        }
41410×
        if first_failure == 0 && code != 0 {
4154×
            first_failure = code;
4166×
        }
417
    }
4189×
    Ok((first_failure, peak))
4199×
}
420
42110×
fn run_module(engine: &wasmtime::Engine, linker: &wasmtime::Linker<State>, spec: &Spec, path: &str, io: &Io) -> Result<(i32, usize)> {
422
    use wasmtime_wasi::{FsPerms, WasiCtxBuilder};
42310×
    let module = ww(wasmtime::Module::from_file(engine, spec.workspace.join(path)), || format!("module {path} does not load"))?;
42410×
    let name = path.rsplit('/').next().unwrap_or(path);
42510×
    let mut argv = vec![name.to_string()];
42610×
    argv.extend(spec.args.iter().cloned());
42710×
    let mut wasi = WasiCtxBuilder::new();
42810×
    match io {
4290×
        Io::Inherit => {
4300×
            wasi.inherit_stdout().inherit_stderr();
4310×
        }
43210×
        Io::Pipes { stdout, stderr, .. } => {
43310×
            wasi.stdout(stdout.clone()).stderr(stderr.clone());
43410×
        }
435
    }
43610×
    wasi.args(&argv).envs(&spec.env).env("PWD", "/src").allow_blocking_current_thread(true);
43720×
    let open = |wasi: &mut WasiCtxBuilder, dir: &Path, at: &str| {
43820×
        ww(wasi.preopened_dir(dir, at, FsPerms::ReadWrite).map(|_| ()), || format!("open {} as {at}", dir.display()))
43920×
    };
44010×
    open(&mut wasi, &spec.workspace, "/src")?;
441
    // Relative paths resolve against the workspace too.
44210×
    open(&mut wasi, &spec.workspace, ".")?;
44310×
    for (dir, target) in &spec.preopens {
4440×
        open(&mut wasi, dir, target)?;
445
    }
44610×
    let limits = Limits { memory: spec.memory_mib as usize * 1024 * 1024, peak: 0 };
44710×
    let mut store = wasmtime::Store::new(engine, State { wasi: wasi.build_p1(), limits });
44810×
    store.limiter(|s| &mut s.limits);
449
    // With epoch interruption on (in process), one tick from now traps:
450
    // the runner ticks only to stop the module.
45110×
    store.set_epoch_deadline(1);
45210×
    let instance = ww(linker.instantiate(&mut store, &module), || format!("module {path} cannot start here"))?;
45310×
    let start =
45410×
        ww(instance.get_typed_func::<(), ()>(&mut store, "_start"), || format!("module {path} has no _start: is it a WASI command?"))?;
45510×
    let code = match start.call(&mut store, ()) {
4566×
        Ok(()) => 0,
4574×
        Err(e) => match e.downcast_ref::<wasmtime_wasi::I32Exit>() {
4581×
            Some(exit) => exit.0,
459
            None => {
4603×
                io.note(format!("{path} trapped: {e:#}"));
4613×
                TRAPPED
462
            }
463
        },
464
    };
46510×
    Ok((code, store.data().limits.peak))
46610×
}
467
468
#[cfg(test)]
469
mod tests {
470
    use super::*;
471
472
    /// A spec running `text` (WAT) as the only module, with 64 MiB.
4733×
    fn spec_of(dir: &Path, text: &str) -> Spec {
4743×
        std::fs::write(dir.join("m.wasm"), wat::parse_str(text).unwrap()).unwrap();
4753×
        Spec { workspace: dir.to_path_buf(), modules: vec!["m.wasm".into()], args: vec![], env: vec![], memory_mib: 64, preopens: vec![] }
4763×
    }
477
4783×
    fn env() -> Environment {
4793×
        Environment { runtime: "wasm".into(), parts: vec![], rootfs: String::new(), argv: vec![], env: vec![] }
4803×
    }
481
482
    /// Records, and whether to say the run was cancelled.
483
    struct Collect(Vec<Record>, bool);
484
    impl Sink for Collect {
4856×
        fn record(&mut self, r: Record) {
4866×
            self.0.push(r);
4876×
        }
4885×
        fn cancelled(&mut self) -> bool {
4895×
            self.1
4905×
        }
491
    }
492
493
    const PRINT: &str = r#"(module
494
      (import "wasi_snapshot_preview1" "fd_write" (func $w (param i32 i32 i32 i32) (result i32)))
495
      (memory (export "memory") 2)
496
      (data (i32.const 16) "line one\nline two")
497
      (func (export "_start")
498
        (i32.store (i32.const 0) (i32.const 16)) (i32.store (i32.const 4) (i32.const 17))
499
        (drop (call $w (i32.const 1) (i32.const 0) (i32.const 1) (i32.const 8)))))"#;
500
501
    const SPIN: &str = r#"(module (func (export "_start") (loop $l (br $l))))"#;
502
503
    #[test]
5041×
    fn in_process_output_streams_as_lines() {
5051×
        let dir = tempfile::tempdir().unwrap();
5061×
        let mut sink = Collect(vec![], false);
5071×
        let o = run_in_thread(&spec_of(dir.path(), PRINT), env(), Duration::from_secs(30), &mut sink).unwrap();
5081×
        assert_eq!(o.exit_code, Some(0));
5092×
        let out: Vec<&str> = sink.0.iter().filter(|r| r.stream == "out").map(|r| r.text.as_str()).collect();
5101×
        assert_eq!(out, ["line one", "line two"], "a last line without its newline still comes");
5111×
        assert_eq!(o.peak_memory_mib, Some(1), "two 64 KiB pages, in whole MiB");
5121×
    }
513
514
    #[test]
5151×
    fn in_process_stops_at_the_time_limit_and_when_cancelled() {
5161×
        let dir = tempfile::tempdir().unwrap();
5171×
        let mut sink = Collect(vec![], false);
5181×
        let started = Instant::now();
5191×
        let o = run_in_thread(&spec_of(dir.path(), SPIN), env(), Duration::from_millis(300), &mut sink).unwrap();
5201×
        assert!(o.timed_out && o.exit_code.is_none(), "{:?}", sink.0);
5211×
        assert!(started.elapsed() < Duration::from_secs(5), "the endless loop was interrupted");
5221×
        let mut sink = Collect(vec![], true);
5231×
        let o = run_in_thread(&spec_of(dir.path(), SPIN), env(), Duration::from_secs(30), &mut sink).unwrap();
5241×
        assert!(o.cancelled && o.exit_code.is_none());
5251×
        assert!(sink.0.iter().any(|r| r.text == "stopped: the run was cancelled"));
5261×
    }
527
528
    #[test]
5291×
    fn globs_match_names_with_stars() {
5301×
        assert!(wildcard("*.wasm", "oxy_core-1a2b.wasm"));
5311×
        assert!(wildcard("oxy_core-*.wasm", "oxy_core-1a2b.wasm"));
5321×
        assert!(!wildcard("oxy_core-*.wasm", "oxy_core-1a2b.d"));
5331×
        assert!(wildcard("a*b*c", "aXbYc") && !wildcard("a*b*c", "aXcYb"));
5341×
        assert!(wildcard("plain.wasm", "plain.wasm") && !wildcard("plain.wasm", "other.wasm"));
5351×
    }
536
537
    #[test]
5381×
    fn modules_are_files_of_the_job() {
5391×
        let dir = tempfile::tempdir().unwrap();
5401×
        let deps = dir.path().join("target/wasm32-wasip1/debug/deps");
5411×
        std::fs::create_dir_all(&deps).unwrap();
5423×
        for f in ["b-2.wasm", "a-1.wasm", "a-1.d"] {
5433×
            std::fs::write(deps.join(f), b"").unwrap();
5443×
        }
5451×
        let ws = dir.path();
5461×
        assert_eq!(
5471×
            modules(ws, "/src/target/wasm32-wasip1/debug/deps/*.wasm").unwrap(),
548
            ["target/wasm32-wasip1/debug/deps/a-1.wasm", "target/wasm32-wasip1/debug/deps/b-2.wasm"]
549
        );
5501×
        assert_eq!(modules(ws, "./target/wasm32-wasip1/debug/deps/b-2.wasm").unwrap(), ["target/wasm32-wasip1/debug/deps/b-2.wasm"]);
5511×
        assert!(modules(ws, "target/wasm32-wasip1/debug/deps/*.so").unwrap_err().to_string().contains("no module matches"));
5521×
        assert!(modules(ws, "../etc/passwd").is_err());
5531×
        assert!(modules(ws, "/etc/x.wasm").is_err());
5541×
        assert!(modules(ws, "target/*/debug/a.wasm").unwrap_err().to_string().contains("only the file name"));
5551×
    }
556
}