OXYGENHUB

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

350 of 413 lines run, 79 of 133 functions · run #209 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()) {
18214×
            peak_memory_mib = Some(m);
18314×
        }
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())];
26722×
    let mut pass_on = |sink: &mut dyn Sink, last: bool| {
26844×
        for (i, pipe) in [&stdout, &stderr].into_iter().enumerate() {
26944×
            let all = pipe.contents();
27044×
            let (offset, partial) = &mut said[i];
27144×
            partial.extend_from_slice(&all[*offset..]);
27244×
            *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
            }
27844×
            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 });
28243×
            }
283
        }
28429×
        while let Ok(note) = notes.try_recv() {
2857×
            sink.record(Record { t: t(&started), stream: "flow".into(), text: note });
2867×
        }
28722×
    };
2889×
    let ran = std::thread::scope(|scope| {
2899×
        let worker = scope.spawn(|| run_modules(&engine, spec, &io));
29022×
        while !worker.is_finished() {
29113×
            std::thread::sleep(Duration::from_millis(100));
29213×
            pass_on(sink, false);
293
            // Once stopped, keep moving the epoch: a store made after the
294
            // first increment (its module still compiling when the time
295
            // limit or the cancel came) sets its deadline one epoch past
296
            // that, and a single increment would never reach it (#93).
29713×
            if timed_out || cancelled {
2982×
                engine.increment_epoch();
29911×
            }
30013×
            if !timed_out && !cancelled && sink.cancelled() {
3011×
                cancelled = true;
3021×
                engine.increment_epoch();
3031×
                sink.record(Record { t: t(&started), stream: "flow".into(), text: "stopped: the run was cancelled".into() });
30412×
            }
30513×
            if Instant::now() >= deadline && !timed_out && !cancelled {
3061×
                timed_out = true;
3071×
                engine.increment_epoch();
3081×
                sink.record(Record {
3091×
                    t: t(&started),
3101×
                    stream: "flow".into(),
3111×
                    text: format!("stopped at the time limit of {} s", timeout.as_secs()),
3121×
                });
31312×
            }
314
        }
3159×
        worker.join().unwrap_or_else(|_| Err(anyhow::anyhow!("the WASM worker panicked")))
3169×
    });
3179×
    pass_on(sink, true);
3189×
    let (exit_code, peak) = match ran {
3199×
        Ok((code, peak)) => (Some(code), Some(peak)),
3200×
        Err(e) => {
3210×
            sink.record(Record { t: t(&started), stream: "flow".into(), text: format!("the WASM tier could not run the job: {e:#}") });
3220×
            (Some(125), None)
323
        }
324
    };
325
    Ok(Outcome {
3269×
        exit_code: if timed_out || cancelled { None } else { exit_code },
3279×
        duration_ms: t(&started),
3289×
        environment,
3299×
        timed_out,
3309×
        cancelled,
3319×
        peak_memory_mib: peak.map(|b| (b as u64).div_ceil(1 << 20)),
3329×
        pruned: Vec::new(),
333
    })
3349×
}
335
336
/// `flow __wasm SPEC`: run the spec's modules in turn, each in a fresh
337
/// instance; the exit code is the first non-zero one, after all have run,
338
/// so one failing module does not hide another's results.
3390×
pub fn wasm_main(spec_path: &Path) -> i32 {
3400×
    let ran = (|| {
3410×
        let spec: Spec = serde_json::from_slice(&std::fs::read(spec_path)?).context("read the spec")?;
3420×
        let engine = ww(wasmtime::Engine::new(&wasmtime::Config::new()), || "start wasmtime".into())?;
3430×
        run_modules(&engine, &spec, &Io::Inherit)
344
    })();
3450×
    match ran {
3460×
        Ok((code, _)) => code,
3470×
        Err(e) => {
3480×
            eprintln!("{FAILED}{e:#}");
3490×
            125
350
        }
351
    }
3520×
}
353
354
/// wasmtime's own error, as this crate's: with its whole chain of causes.
35568×
fn ww<T>(r: wasmtime::Result<T>, what: impl FnOnce() -> String) -> Result<T> {
35668×
    r.map_err(|e| anyhow::anyhow!("{}: {e:#}", what()))
35768×
}
358
359
/// Where a module's output and the tier's notes go: inherited by the
360
/// helper process (notes on stderr, marked), or pipes the runner reads.
361
enum Io {
362
    Inherit,
363
    Pipes {
364
        stdout: wasmtime_wasi::p2::pipe::MemoryOutputPipe,
365
        stderr: wasmtime_wasi::p2::pipe::MemoryOutputPipe,
366
        notes: mpsc::Sender<String>,
367
    },
368
}
369
370
impl Io {
3717×
    fn note(&self, text: String) {
3727×
        match self {
3730×
            Io::Inherit => eprintln!("{NOTE}{text}"),
3747×
            Io::Pipes { notes, .. } => {
3757×
                let _ = notes.send(text);
3767×
            }
377
        }
3787×
    }
379
}
380
381
/// The sandbox's memory limit, and the most a module asked for within it.
382
struct Limits {
383
    memory: usize,
384
    peak: usize,
385
}
386
387
impl wasmtime::ResourceLimiter for Limits {
3889×
    fn memory_growing(&mut self, _current: usize, desired: usize, maximum: Option<usize>) -> wasmtime::Result<bool> {
3899×
        let ok = desired <= self.memory && maximum.is_none_or(|m| desired <= m);
3909×
        if ok {
3918×
            self.peak = self.peak.max(desired);
3928×
        }
3939×
        Ok(ok)
3949×
    }
395
3960×
    fn table_growing(&mut self, _current: usize, desired: usize, maximum: Option<usize>) -> wasmtime::Result<bool> {
3970×
        Ok(maximum.is_none_or(|m| desired <= m))
3980×
    }
399
}
400
401
struct State {
402
    wasi: wasmtime_wasi::p1::WasiP1Ctx,
403
    limits: Limits,
404
}
405
406
/// Run each module in turn: the first non-zero exit code after all ran,
407
/// and the most memory any of them held, in bytes.
4089×
fn run_modules(engine: &wasmtime::Engine, spec: &Spec, io: &Io) -> Result<(i32, usize)> {
4099×
    let mut linker: wasmtime::Linker<State> = wasmtime::Linker::new(engine);
4109×
    ww(wasmtime_wasi::p1::add_to_linker_sync(&mut linker, |s: &mut State| &mut s.wasi), || "link WASI".into())?;
4119×
    let (mut first_failure, mut peak) = (0, 0);
41210×
    for path in &spec.modules {
41310×
        if spec.modules.len() > 1 {
4142×
            io.note(format!("running {path}"));
4158×
        }
41610×
        let (code, held) = run_module(engine, &linker, spec, path, io)?;
41710×
        peak = peak.max(held);
41810×
        if spec.modules.len() > 1 {
4192×
            io.note(format!("{path}: exit {code}"));
4208×
        }
42110×
        if first_failure == 0 && code != 0 {
4224×
            first_failure = code;
4236×
        }
424
    }
4259×
    Ok((first_failure, peak))
4269×
}
427
42810×
fn run_module(engine: &wasmtime::Engine, linker: &wasmtime::Linker<State>, spec: &Spec, path: &str, io: &Io) -> Result<(i32, usize)> {
429
    use wasmtime_wasi::{FsPerms, WasiCtxBuilder};
43010×
    let module = ww(wasmtime::Module::from_file(engine, spec.workspace.join(path)), || format!("module {path} does not load"))?;
43110×
    let name = path.rsplit('/').next().unwrap_or(path);
43210×
    let mut argv = vec![name.to_string()];
43310×
    argv.extend(spec.args.iter().cloned());
43410×
    let mut wasi = WasiCtxBuilder::new();
43510×
    match io {
4360×
        Io::Inherit => {
4370×
            wasi.inherit_stdout().inherit_stderr();
4380×
        }
43910×
        Io::Pipes { stdout, stderr, .. } => {
44010×
            wasi.stdout(stdout.clone()).stderr(stderr.clone());
44110×
        }
442
    }
44310×
    wasi.args(&argv).envs(&spec.env).env("PWD", "/src").allow_blocking_current_thread(true);
44420×
    let open = |wasi: &mut WasiCtxBuilder, dir: &Path, at: &str| {
44520×
        ww(wasi.preopened_dir(dir, at, FsPerms::ReadWrite).map(|_| ()), || format!("open {} as {at}", dir.display()))
44620×
    };
44710×
    open(&mut wasi, &spec.workspace, "/src")?;
448
    // Relative paths resolve against the workspace too.
44910×
    open(&mut wasi, &spec.workspace, ".")?;
45010×
    for (dir, target) in &spec.preopens {
4510×
        open(&mut wasi, dir, target)?;
452
    }
45310×
    let limits = Limits { memory: spec.memory_mib as usize * 1024 * 1024, peak: 0 };
45410×
    let mut store = wasmtime::Store::new(engine, State { wasi: wasi.build_p1(), limits });
45510×
    store.limiter(|s| &mut s.limits);
456
    // With epoch interruption on (in process), one tick from now traps:
457
    // the runner ticks only to stop the module.
45810×
    store.set_epoch_deadline(1);
45910×
    let instance = ww(linker.instantiate(&mut store, &module), || format!("module {path} cannot start here"))?;
46010×
    let start =
46110×
        ww(instance.get_typed_func::<(), ()>(&mut store, "_start"), || format!("module {path} has no _start: is it a WASI command?"))?;
46210×
    let code = match start.call(&mut store, ()) {
4636×
        Ok(()) => 0,
4644×
        Err(e) => match e.downcast_ref::<wasmtime_wasi::I32Exit>() {
4651×
            Some(exit) => exit.0,
466
            None => {
4673×
                io.note(format!("{path} trapped: {e:#}"));
4683×
                TRAPPED
469
            }
470
        },
471
    };
47210×
    Ok((code, store.data().limits.peak))
47310×
}
474
475
#[cfg(test)]
476
mod tests {
477
    use super::*;
478
479
    /// A spec running `text` (WAT) as the only module, with 64 MiB.
4803×
    fn spec_of(dir: &Path, text: &str) -> Spec {
4813×
        std::fs::write(dir.join("m.wasm"), wat::parse_str(text).unwrap()).unwrap();
4823×
        Spec { workspace: dir.to_path_buf(), modules: vec!["m.wasm".into()], args: vec![], env: vec![], memory_mib: 64, preopens: vec![] }
4833×
    }
484
4853×
    fn env() -> Environment {
4863×
        Environment { runtime: "wasm".into(), parts: vec![], rootfs: String::new(), argv: vec![], env: vec![] }
4873×
    }
488
489
    /// Records, and whether to say the run was cancelled.
490
    struct Collect(Vec<Record>, bool);
491
    impl Sink for Collect {
4926×
        fn record(&mut self, r: Record) {
4936×
            self.0.push(r);
4946×
        }
4955×
        fn cancelled(&mut self) -> bool {
4965×
            self.1
4975×
        }
498
    }
499
500
    const PRINT: &str = r#"(module
501
      (import "wasi_snapshot_preview1" "fd_write" (func $w (param i32 i32 i32 i32) (result i32)))
502
      (memory (export "memory") 2)
503
      (data (i32.const 16) "line one\nline two")
504
      (func (export "_start")
505
        (i32.store (i32.const 0) (i32.const 16)) (i32.store (i32.const 4) (i32.const 17))
506
        (drop (call $w (i32.const 1) (i32.const 0) (i32.const 1) (i32.const 8)))))"#;
507
508
    const SPIN: &str = r#"(module (func (export "_start") (loop $l (br $l))))"#;
509
510
    #[test]
5111×
    fn in_process_output_streams_as_lines() {
5121×
        let dir = tempfile::tempdir().unwrap();
5131×
        let mut sink = Collect(vec![], false);
5141×
        let o = run_in_thread(&spec_of(dir.path(), PRINT), env(), Duration::from_secs(30), &mut sink).unwrap();
5151×
        assert_eq!(o.exit_code, Some(0));
5162×
        let out: Vec<&str> = sink.0.iter().filter(|r| r.stream == "out").map(|r| r.text.as_str()).collect();
5171×
        assert_eq!(out, ["line one", "line two"], "a last line without its newline still comes");
5181×
        assert_eq!(o.peak_memory_mib, Some(1), "two 64 KiB pages, in whole MiB");
5191×
    }
520
521
    #[test]
5221×
    fn in_process_stops_at_the_time_limit_and_when_cancelled() {
5231×
        let dir = tempfile::tempdir().unwrap();
5241×
        let mut sink = Collect(vec![], false);
5251×
        let started = Instant::now();
5261×
        let o = run_in_thread(&spec_of(dir.path(), SPIN), env(), Duration::from_millis(300), &mut sink).unwrap();
5271×
        assert!(o.timed_out && o.exit_code.is_none(), "{:?}", sink.0);
5281×
        assert!(started.elapsed() < Duration::from_secs(5), "the endless loop was interrupted");
5291×
        let mut sink = Collect(vec![], true);
5301×
        let o = run_in_thread(&spec_of(dir.path(), SPIN), env(), Duration::from_secs(30), &mut sink).unwrap();
5311×
        assert!(o.cancelled && o.exit_code.is_none());
5321×
        assert!(sink.0.iter().any(|r| r.text == "stopped: the run was cancelled"));
5331×
    }
534
535
    #[test]
5361×
    fn globs_match_names_with_stars() {
5371×
        assert!(wildcard("*.wasm", "oxy_core-1a2b.wasm"));
5381×
        assert!(wildcard("oxy_core-*.wasm", "oxy_core-1a2b.wasm"));
5391×
        assert!(!wildcard("oxy_core-*.wasm", "oxy_core-1a2b.d"));
5401×
        assert!(wildcard("a*b*c", "aXbYc") && !wildcard("a*b*c", "aXcYb"));
5411×
        assert!(wildcard("plain.wasm", "plain.wasm") && !wildcard("plain.wasm", "other.wasm"));
5421×
    }
543
544
    #[test]
5451×
    fn modules_are_files_of_the_job() {
5461×
        let dir = tempfile::tempdir().unwrap();
5471×
        let deps = dir.path().join("target/wasm32-wasip1/debug/deps");
5481×
        std::fs::create_dir_all(&deps).unwrap();
5493×
        for f in ["b-2.wasm", "a-1.wasm", "a-1.d"] {
5503×
            std::fs::write(deps.join(f), b"").unwrap();
5513×
        }
5521×
        let ws = dir.path();
5531×
        assert_eq!(
5541×
            modules(ws, "/src/target/wasm32-wasip1/debug/deps/*.wasm").unwrap(),
555
            ["target/wasm32-wasip1/debug/deps/a-1.wasm", "target/wasm32-wasip1/debug/deps/b-2.wasm"]
556
        );
5571×
        assert_eq!(modules(ws, "./target/wasm32-wasip1/debug/deps/b-2.wasm").unwrap(), ["target/wasm32-wasip1/debug/deps/b-2.wasm"]);
5581×
        assert!(modules(ws, "target/wasm32-wasip1/debug/deps/*.so").unwrap_err().to_string().contains("no module matches"));
5591×
        assert!(modules(ws, "../etc/passwd").is_err());
5601×
        assert!(modules(ws, "/etc/x.wasm").is_err());
5611×
        assert!(modules(ws, "target/*/debug/a.wasm").unwrap_err().to_string().contains("only the file name"));
5621×
    }
563
}