Track Rust sandbox exec bugs: async output collection bypasses timeouts

The DLL is deliberately a native ABI test double, not a VM implementation. It supplies capability/host metadata and a valid sequence of 512 stderr events, each 64 KiB, then an exit event. The defect and both controls use the unchanged production NVX Rust code. Tokio is pinned to the version in the examined repository's lockfile.

From an NVX checkout containing the examined revision, export source outside the worktree:

$root = Join-Path $env:TEMP "nvx-async-timeout-repro-9a2721a7"
New-Item -ItemType Directory -Path $root -Force | Out-Null
git archive --format=zip "--output=$root\nvx.zip" 9a2721a713a1bb78f0c94c189900cf89a728a26e
Expand-Archive -LiteralPath "$root\nvx.zip" -DestinationPath "$root\nvx-src"
New-Item -ItemType Directory -Path "$root\repro" | Out-Null

Save the three files below inside $root\repro, then run:

$env:CARGO_TARGET_DIR = "$root\target"
cargo +1.95.0 build --manifest-path "$root\repro\Cargo.toml" --lib --bin nvx-async-timeout-repro
& "$root\target\debug\nvx-async-timeout-repro.exe" "$root\target\debug\timeout_agent_fixture.dll"

The executable prints all three results and asserts that the two controls return TimedOut, while async collection stays pending after one second and only completes after the retained stderr is released. All handles and executions are released before the process exits.

Cargo.toml

[package]
name = "nvx-async-timeout-repro"
version = "0.0.0"
edition = "2024"

[lib]
name = "timeout_agent_fixture"
path = "shim.rs"
crate-type = ["cdylib"]

[[bin]]
name = "nvx-async-timeout-repro"
path = "main.rs"

[dependencies]
aci_edge_sandboxes = { path = '..\nvx-src\aci_edge_sandboxes', default-features = false, features = ["agent", "async"] }
aci_edge_sandboxes_model = { path = '..\nvx-src\aci_edge_sandboxes\model' }
serde_json = "1.0"
sha2 = "0.10"
tokio = { version = "=1.53.1", features = ["rt-multi-thread", "time", "io-util"] }

shim.rs

use std::collections::HashMap;
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::{Mutex, OnceLock};

use aci_edge_sandboxes::Capabilities;
use aci_edge_sandboxes_model::setup::RuntimeDigests;
use aci_edge_sandboxes_model::wire::{ExecEnd, HostInfo, SANDBOX_ABI_VERSION};
use aci_edge_sandboxes_model::ExecOutcome;

const CHUNK_BYTES: usize = 64 * 1024;
const OUTPUT_CHUNKS: usize = 512;
static NEXT: AtomicU64 = AtomicU64::new(1);
static EXECS: OnceLock<Mutex<HashMap<u64, usize>>> = OnceLock::new();

fn execs() -> &'static Mutex<HashMap<u64, usize>> {
    EXECS.get_or_init(|| Mutex::new(HashMap::new()))
}

#[repr(C)]
pub struct Event {
    kind: u32,
    data: *const u8,
    length: usize,
}

unsafe fn reply(bytes: &[u8], out: *mut *mut u8, length: *mut usize) -> i32 {
    let buffer = bytes.to_vec().into_boxed_slice();
    unsafe {
        *length = buffer.len();
        *out = Box::into_raw(buffer).cast();
    }
    0
}

#[unsafe(no_mangle)]
pub extern "C" fn aci_edge_sandbox_api_version() -> u32 {
    SANDBOX_ABI_VERSION
}

#[unsafe(no_mangle)]
pub unsafe extern "C" fn aci_edge_sandbox_host_open(
    _: *const u8, _: usize, host: *mut u64, _: *mut *mut u8, _: *mut usize,
) -> i32 {
    unsafe { *host = 1 };
    0
}

#[unsafe(no_mangle)]
pub extern "C" fn aci_edge_sandbox_host_close(_: u64) -> i32 { 0 }

#[unsafe(no_mangle)]
pub unsafe extern "C" fn aci_edge_sandbox_capabilities(
    _: u64, out: *mut *mut u8, length: *mut usize,
) -> i32 {
    let mut capabilities = Capabilities::new("native-ABI-fixture");
    capabilities.exec.command_line = true;
    capabilities.exec.cancel = true;
    let bytes = serde_json::to_vec(&capabilities).unwrap();
    unsafe { reply(&bytes, out, length) }
}

#[unsafe(no_mangle)]
pub unsafe extern "C" fn aci_edge_sandbox_host_info(
    _: u64, out: *mut *mut u8, length: *mut usize,
) -> i32 {
    let info = HostInfo::new(None, RuntimeDigests {
        openvmm: [0; 32], kernel: [0; 32], initrd: [0; 32],
    });
    let bytes = serde_json::to_vec(&info).unwrap();
    unsafe { reply(&bytes, out, length) }
}

macro_rules! query {
    ($($name:ident),+ $(,)?) => {$(
        #[unsafe(no_mangle)]
        pub unsafe extern "C" fn $name(
            _: u64, out: *mut *mut u8, length: *mut usize,
        ) -> i32 {
            unsafe { reply(b"{}", out, length) }
        }
    )+};
}
query!(
    aci_edge_sandbox_probe, aci_edge_exec_close_input, aci_edge_exec_cancel,
    aci_edge_image_list,
);

macro_rules! call {
    ($($name:ident),+ $(,)?) => {$(
        #[unsafe(no_mangle)]
        pub unsafe extern "C" fn $name(
            _: u64, _: *const u8, _: usize, out: *mut *mut u8, length: *mut usize,
        ) -> i32 {
            unsafe { reply(b"{}", out, length) }
        }
    )+};
}
call!(
    aci_edge_sandbox_validate_provision, aci_edge_sandbox_validate_exec,
    aci_edge_sandbox_start, aci_edge_sandbox_stop, aci_edge_sandbox_deprovision,
    aci_edge_sandbox_diagnostics, aci_edge_sandbox_guest_logs,
    aci_edge_image_register, aci_edge_image_verify, aci_edge_image_unregister,
);

#[unsafe(no_mangle)]
pub unsafe extern "C" fn aci_edge_sandbox_provision(
    _: u64, _: *const u8, _: usize, out: *mut *mut u8, length: *mut usize,
) -> i32 {
    unsafe { reply(br#"{"token":"00000000000000000000000000000001"}"#, out, length) }
}

#[unsafe(no_mangle)]
pub unsafe extern "C" fn aci_edge_sandbox_exec(
    _: u64, _: *const u8, _: usize, _: *const u8, _: usize,
    exec: *mut u64, _: *mut *mut u8, _: *mut usize,
) -> i32 {
    let handle = NEXT.fetch_add(1, Ordering::Relaxed);
    execs().lock().unwrap().insert(handle, OUTPUT_CHUNKS);
    unsafe { *exec = handle };
    0
}

#[unsafe(no_mangle)]
pub unsafe extern "C" fn aci_edge_exec_next(
    exec: u64, _: i64, event: *mut Event,
) -> i32 {
    static CHUNK: [u8; CHUNK_BYTES] = [b'x'; CHUNK_BYTES];
    static END: OnceLock<Vec<u8>> = OnceLock::new();
    let mut executions = execs().lock().unwrap();
    let remaining = executions.get_mut(&exec).unwrap();
    let (kind, data) = if *remaining > 0 {
        *remaining -= 1;
        (2, &CHUNK[..])
    } else {
        (3, &END.get_or_init(|| {
            serde_json::to_vec(&ExecEnd::Outcome(ExecOutcome::Exited(0))).unwrap()
        })[..])
    };
    unsafe { *event = Event { kind, data: data.as_ptr(), length: data.len() } };
    0
}

#[unsafe(no_mangle)]
pub unsafe extern "C" fn aci_edge_exec_write(
    _: u64, _: *const u8, length: usize, _: i64,
    written: *mut usize, _: *mut *mut u8, _: *mut usize,
) -> i32 {
    unsafe { *written = length };
    0
}

#[unsafe(no_mangle)]
pub extern "C" fn aci_edge_exec_release(exec: u64) -> i32 {
    execs().lock().unwrap().remove(&exec);
    0
}

#[unsafe(no_mangle)]
pub unsafe extern "C" fn aci_edge_buffer_free(data: *mut u8, length: usize) {
    unsafe { drop(Box::from_raw(std::ptr::slice_from_raw_parts_mut(data, length))) };
}

main.rs

use std::path::PathBuf;
use std::time::{Duration, Instant};

use aci_edge_sandboxes::agent::{
    AgentConfig, RuntimeFiles, RuntimeSource, SetupConfig,
};
use aci_edge_sandboxes::{
    AciEdgeSandbox, AsyncAciEdgeSandbox, ExecOutcome, ExecRequest, ProvisionRequest,
};
use sha2::{Digest, Sha256};

const WORKLOAD_TIMEOUT: Duration = Duration::from_millis(100);
const OUTER_TIMEOUT: Duration = Duration::from_millis(1000);

fn main() {
    let library = PathBuf::from(std::env::args_os().nth(1).expect("supply the fixture DLL path"));
    let library = std::path::absolute(library).unwrap();
    let approved: [u8; 32] = Sha256::digest(std::fs::read(&library).unwrap()).into();
    let state = library.parent().unwrap().join("fixture-state");
    let runtime_files = RuntimeFiles {
        openvmm: library.clone(),
        kernel: library.clone(),
        initrd: library.clone(),
        approved_sha256: None,
    };
    let setup = SetupConfig::new(&state, RuntimeSource::Files(runtime_files));
    let client = AciEdgeSandbox::agent(AgentConfig::new(setup, &library, approved)).unwrap();
    let sandbox = client.provision(&ProvisionRequest::new()).unwrap().sandbox_id;
    client.start(&sandbox).unwrap();
    let request = || ExecRequest::command_line("fixture-stderr-32m").with_timeout(WORKLOAD_TIMEOUT);

    let started = Instant::now();
    let mut synchronous = client.exec(&sandbox, &request()).unwrap();
    let retained = synchronous.take_stderr().unwrap();
    let output = synchronous.wait_with_output().unwrap();
    println!(
        "sync wait_with_output: outcome={:?}, elapsed={}ms",
        output.outcome, started.elapsed().as_millis(),
    );
    assert_eq!(output.outcome, ExecOutcome::TimedOut);
    drop(retained);

    let rt = tokio::runtime::Builder::new_multi_thread().enable_all().build().unwrap();
    rt.block_on(async {
        let asynchronous = AsyncAciEdgeSandbox::new(client.clone());
        let started = Instant::now();
        let mut execution = asynchronous.exec(sandbox.clone(), request()).await.unwrap();
        let retained = execution.take_stderr().unwrap();
        let outcome = tokio::time::timeout(OUTER_TIMEOUT, execution.wait()).await.unwrap().unwrap();
        println!(
            "async wait: outcome={outcome:?}, elapsed={}ms",
            started.elapsed().as_millis(),
        );
        assert_eq!(outcome, ExecOutcome::TimedOut);
        drop(retained);

        let started = Instant::now();
        let mut execution = asynchronous.exec(sandbox.clone(), request()).await.unwrap();
        let retained = execution.take_stderr().unwrap();
        let mut waiting = Box::pin(execution.wait_with_output());
        let timed_out = tokio::time::timeout(OUTER_TIMEOUT, &mut waiting).await.is_err();
        println!(
            "async wait_with_output: pending_after_{}ms={timed_out}, workload_timeout={}ms",
            started.elapsed().as_millis(), WORKLOAD_TIMEOUT.as_millis(),
        );
        drop(retained);
        let output = tokio::time::timeout(OUTER_TIMEOUT, waiting).await.unwrap().unwrap();
        println!("after releasing retained stderr: outcome={:?}", output.outcome);
        assert!(timed_out, "expected to reproduce the collector ordering bug");
        assert_eq!(output.outcome, ExecOutcome::Exited(0));
    });
    client.stop(&sandbox).unwrap();
    client.deprovision(&sandbox).unwrap();
}