supermachine 0.7.108

//! Prefix-tree capture-under-load probe.
//!
//! Restores a known-good Playwright/Node snapshot, starts an intentionally
//! non-quiescent in-guest workload (continuous append loop, Node timer state,
//! and repeated in-flight HTTP transfers), captures dirty differential nodes
//! mid-flight, restores every node, and verifies runner-facing invariants.
//!
//! Env:
//!   SUPERMACHINE_PROBE_SNAPSHOT   full snapshot dir to boot (restore.snap inside)
//!   SUPERMACHINE_PROBE_WORKDIR    scratch dir for captured nodes
//!   SUPERMACHINE_PROBE_OUT        CSV path (default: <workdir>/capture-under-load.csv)
//!   SUPERMACHINE_PROBE_REPS       captures to take (default 10; must be >=10)
//!   SUPERMACHINE_PROBE_MEM_MIB    guest RAM (default 8192)

#[cfg(all(target_os = "linux", target_arch = "x86_64"))]
fn main() {
    use std::fs::{self, File};
    use std::io::Write;
    use std::path::Path;
    use std::time::{Duration, Instant};
    use supermachine::{Image, VmConfig};

    let snap_src = std::env::var("SUPERMACHINE_PROBE_SNAPSHOT")
        .expect("set SUPERMACHINE_PROBE_SNAPSHOT to a full snapshot dir");
    let workdir = std::env::var("SUPERMACHINE_PROBE_WORKDIR")
        .expect("set SUPERMACHINE_PROBE_WORKDIR to a scratch dir");
    let out_path = std::env::var("SUPERMACHINE_PROBE_OUT")
        .unwrap_or_else(|_| format!("{workdir}/capture-under-load.csv"));
    let reps: usize = parse_env("SUPERMACHINE_PROBE_REPS", 10);
    let mem_mib: u32 = parse_env("SUPERMACHINE_PROBE_MEM_MIB", 8192);
    assert!(reps >= 10, "SUPERMACHINE_PROBE_REPS must be >= 10");

    fs::create_dir_all(&workdir).expect("workdir");
    let mut csv = File::create(&out_path).expect("create csv");
    writeln!(csv, "rep,capture_ms,diff_bytes,restore_exec,append_readable,dmesg_clean,timer_coherent,http_progress,summary").unwrap();

    let cfg = VmConfig::new().with_memory_mib(mem_mib);
    let image = Image::from_snapshot(&snap_src).expect("from_snapshot");
    let restored_base = Path::new(&snap_src).join("restore.snap");
    let vm = image.start(&cfg).expect("start");
    wait_for_exec(&vm);
    install_workload(&vm);

    for rep in 1..=reps {
        // Fire a large HTTP transfer and capture immediately while guest Node is
        // serving chunks and the append/timer loops continue mutating state.
        let trigger = format!(
            "node -e \"const http=require('http');const req=http.get('http://127.0.0.1:31386/stream?rep={rep}',res=>{{res.on('data',()=>{{}});res.on('end',()=>process.exit(0));}});req.on('error',e=>{{console.error(e);process.exit(1)}});setTimeout(()=>process.exit(0),25);\" >/tmp/supermachine-load-client-{rep}.log 2>&1 &"
        );
        checked_exec(&vm, &trigger, "start in-flight http client");
        std::thread::sleep(Duration::from_millis(5));

        let snap_dir = format!("{workdir}/node-{rep:02}");
        let _ = fs::remove_dir_all(&snap_dir);
        let t0 = Instant::now();
        let node = vm
            .snapshot_diff_live_dirty(&snap_dir, &restored_base)
            .expect("dirty capture under load");
        let capture_ms = t0.elapsed().as_secs_f64() * 1000.0;
        let diff_bytes = fs::metadata(Path::new(&snap_dir).join("restore.snap"))
            .expect("diff metadata")
            .len();

        let restored = node.start(&cfg).expect("restore captured node");
        wait_for_exec(&restored);
        let verdict = verify_restored(&restored);
        restored.stop().ok();

        writeln!(
            csv,
            "{rep},{capture_ms:.3},{diff_bytes},{},{},{},{},{},{}",
            verdict.restore_exec,
            verdict.append_readable,
            verdict.dmesg_clean,
            verdict.timer_coherent,
            verdict.http_progress,
            verdict.summary
        )
        .unwrap();
        csv.flush().unwrap();
        eprintln!(
            "rep={rep} capture_ms={capture_ms:.3} diff_bytes={diff_bytes} restore_exec={} append_readable={} dmesg_clean={} timer_coherent={} http_progress={} summary={}",
            verdict.restore_exec,
            verdict.append_readable,
            verdict.dmesg_clean,
            verdict.timer_coherent,
            verdict.http_progress,
            verdict.summary
        );
        if !verdict.pass() {
            std::process::exit(1);
        }
    }

    vm.stop().ok();
    eprintln!("wrote {out_path}");
}

#[cfg(all(target_os = "linux", target_arch = "x86_64"))]
#[derive(Debug)]
struct Verdict {
    restore_exec: bool,
    append_readable: bool,
    dmesg_clean: bool,
    timer_coherent: bool,
    http_progress: bool,
    summary: String,
}

#[cfg(all(target_os = "linux", target_arch = "x86_64"))]
impl Verdict {
    fn pass(&self) -> bool {
        self.restore_exec
            && self.append_readable
            && self.dmesg_clean
            && self.timer_coherent
            && self.http_progress
    }
}

#[cfg(all(target_os = "linux", target_arch = "x86_64"))]
fn parse_env<T>(name: &str, default: T) -> T
where
    T: std::str::FromStr,
{
    std::env::var(name)
        .ok()
        .and_then(|v| v.parse().ok())
        .unwrap_or(default)
}

#[cfg(all(target_os = "linux", target_arch = "x86_64"))]
fn wait_for_exec(vm: &supermachine::Vm) {
    for _ in 0..120 {
        if let Ok(o) = vm
            .exec_builder()
            .argv(["/bin/sh", "-lc", "echo up"])
            .output()
        {
            if o.success() {
                return;
            }
        }
        std::thread::sleep(std::time::Duration::from_millis(1000));
    }
    panic!("guest exec did not become ready");
}

#[cfg(all(target_os = "linux", target_arch = "x86_64"))]
fn checked_exec(vm: &supermachine::Vm, cmd: &str, label: &str) {
    let output = vm
        .exec_builder()
        .argv(["/bin/sh", "-lc", cmd])
        .output()
        .expect(label);
    if !output.success() {
        eprintln!(
            "{label} failed: {}",
            String::from_utf8_lossy(&output.stderr)
        );
        std::process::exit(1);
    }
}

#[cfg(all(target_os = "linux", target_arch = "x86_64"))]
fn install_workload(vm: &supermachine::Vm) {
    let script = r#"
set -eu
mkdir -p /root/supermachine-capture-under-load
cat >/root/supermachine-capture-under-load/server.js <<'JS'
const fs = require('fs');
const http = require('http');
const dir = '/root/supermachine-capture-under-load';
let ticks = 0;
let heap = [];
setInterval(() => {
  ticks += 1;
  heap.push({ ticks, payload: 't'.repeat(4096), at: Date.now() });
  if (heap.length > 1024) heap = heap.slice(-512);
  fs.writeFileSync(`${dir}/timer-state.json`, JSON.stringify({ ticks, heap: heap.length }));
}, 10);
setInterval(() => {
  fs.appendFileSync(`${dir}/append.log`, `line ${Date.now()} ${ticks} ${'a'.repeat(1024)}\n`);
}, 5);
http.createServer((req, res) => {
  if (req.url.startsWith('/stream')) {
    let sent = 0;
    const timer = setInterval(() => {
      sent += 1;
      fs.writeFileSync(`${dir}/http-progress.txt`, String(sent));
      res.write(Buffer.alloc(64 * 1024, sent % 251));
      if (sent >= 200) {
        clearInterval(timer);
        res.end('done');
      }
    }, 2);
    return;
  }
  res.end(JSON.stringify({ ok: true, ticks }));
}).listen(31386, '127.0.0.1');
JS
nohup node /root/supermachine-capture-under-load/server.js >/root/supermachine-capture-under-load/server.log 2>&1 &
for i in $(seq 1 100); do node -e "http=require('http');http.get('http://127.0.0.1:31386/ready',r=>process.exit(r.statusCode===200?0:1)).on('error',()=>process.exit(1))" && test -s /root/supermachine-capture-under-load/append.log && test -s /root/supermachine-capture-under-load/timer-state.json && exit 0; sleep .1; done
exit 1
"#;
    checked_exec(vm, script, "install workload");
}

#[cfg(all(target_os = "linux", target_arch = "x86_64"))]
fn verify_restored(vm: &supermachine::Vm) -> Verdict {
    let output = vm
        .exec_builder()
        .argv([
            "/bin/sh",
            "-lc",
            r#"
set -eu
cd /root/supermachine-capture-under-load
restore_exec=1
append_readable=0
dmesg_clean=0
timer_coherent=0
http_progress=0
[ -s append.log ] && tail -n 3 append.log >/tmp/append-tail && append_readable=1
node -e "const fs=require('fs');const s=JSON.parse(fs.readFileSync('timer-state.json','utf8')); if (s.ticks > 0 && s.heap > 0) process.exit(0); process.exit(1)" && timer_coherent=1
[ -s http-progress.txt ] && [ "$(cat http-progress.txt)" -gt 0 ] && http_progress=1
if dmesg | tail -n 120 | grep -Eiq 'I/O error|EXT4-fs error|Buffer I/O error|blk_update_request|kernel panic|Oops'; then dmesg_clean=0; else dmesg_clean=1; fi
printf 'restore_exec=%s append_readable=%s dmesg_clean=%s timer_coherent=%s http_progress=%s\n' "$restore_exec" "$append_readable" "$dmesg_clean" "$timer_coherent" "$http_progress"
"#,
        ])
        .output();
    match output {
        Ok(out) if out.success() => parse_verdict(&String::from_utf8_lossy(&out.stdout)),
        Ok(out) => Verdict {
            restore_exec: false,
            append_readable: false,
            dmesg_clean: false,
            timer_coherent: false,
            http_progress: false,
            summary: format!(
                "verify-command-failed-{}",
                String::from_utf8_lossy(&out.stderr).replace(',', ";")
            ),
        },
        Err(err) => Verdict {
            restore_exec: false,
            append_readable: false,
            dmesg_clean: false,
            timer_coherent: false,
            http_progress: false,
            summary: format!("exec-error-{err}"),
        },
    }
}

#[cfg(all(target_os = "linux", target_arch = "x86_64"))]
fn parse_verdict(stdout: &str) -> Verdict {
    let mut v = Verdict {
        restore_exec: false,
        append_readable: false,
        dmesg_clean: false,
        timer_coherent: false,
        http_progress: false,
        summary: stdout.trim().replace(',', ";"),
    };
    for part in stdout.split_whitespace() {
        if let Some((k, val)) = part.split_once('=') {
            let b = val == "1";
            match k {
                "restore_exec" => v.restore_exec = b,
                "append_readable" => v.append_readable = b,
                "dmesg_clean" => v.dmesg_clean = b,
                "timer_coherent" => v.timer_coherent = b,
                "http_progress" => v.http_progress = b,
                _ => {}
            }
        }
    }
    v
}

#[cfg(not(all(target_os = "linux", target_arch = "x86_64")))]
fn main() {
    eprintln!("linux/x86_64 only");
}