lazaret/crates/lazaret-host-coordinator/tests/runtime-composition/handoff.rs

use super::{
    AttachedRunscSignal, CoordinatorId, Duration, Fixture, GenerationId, HOST_COORDINATOR_DRIVER,
    HostProcessOwnershipSlot, MatrixError, ParticipantState, Path, PathBuf, PidfsProcessIdentityV1,
    PlatformObservation, ProcessOwnershipCommitAck, ProcessOwnershipContext, RetainedProcess,
    SandboxCoordinator, Signal, TargetCommand, UnixListener, check_platform, child_broker,
    current_process_identity, failed_probe_blocks_sandbox_start, fs, host_path,
    networked_sandbox_view_isolated, read_start_time, sandbox_participant, thread,
};

pub(super) fn run_handoff_matrix() -> Result<(), MatrixError> {
    require_matrix_prerequisites()?;
    replaced_handoff_authority_refuses_target_work()?;
    cross_bundle_handoff_refuses_target_work()
}

pub(super) fn run_network_cases() -> Result<(), MatrixError> {
    failed_probe_blocks_sandbox_start()?;
    networked_sandbox_view_isolated()
}

pub(super) fn require_matrix_prerequisites() -> Result<(), MatrixError> {
    let release = fs::read_to_string("/proc/sys/kernel/osrelease")?;
    check_platform(&PlatformObservation {
        operating_system: std::env::consts::OS,
        architecture: std::env::consts::ARCH,
        kernel_release: release.trim(),
    })
    .map_err(|error| MatrixError::Skip(error.to_string()))?;
    require_runtime_prerequisites()
}

fn require_runtime_prerequisites() -> Result<(), MatrixError> {
    let runsc = std::env::var_os("PATH")
        .into_iter()
        .flat_map(|path| std::env::split_paths(&path).collect::<Vec<_>>())
        .map(|directory| directory.join("runsc"))
        .find(|candidate| candidate.is_file())
        .ok_or_else(|| MatrixError::Skip("runsc is not installed on PATH".into()))?;
    for path in [
        runsc,
        PathBuf::from("/bin/sh"),
        PathBuf::from("/bin/sleep"),
        PathBuf::from("/lib64/libc.so.6"),
        PathBuf::from("/lib64/libgcc_s.so.1"),
        PathBuf::from("/lib64/libm.so.6"),
        PathBuf::from("/lib64/libtinfo.so.6"),
        PathBuf::from("/lib64/ld-linux-x86-64.so.2"),
    ] {
        if !path.is_file() {
            return Err(MatrixError::Skip(format!(
                "required runtime fixture is absent: {}",
                path.display()
            )));
        }
    }
    let socket_directory = tempfile::tempdir()?;
    let socket_path = socket_directory.path().join("runsc-control.sock");
    match UnixListener::bind(&socket_path) {
        Ok(listener) => drop(listener),
        Err(error) if error.kind() == std::io::ErrorKind::PermissionDenied => {
            return Err(MatrixError::Skip(format!(
                "filesystem Unix sockets required by runsc are denied: {error}"
            )));
        }
        Err(error) => return Err(error.into()),
    }
    Ok(())
}

pub(super) fn lost_acknowledgment_refuses_target_work() -> Result<(), MatrixError> {
    let fixture = Fixture::new()?;
    let mut pending = fixture.launch_pending()?;
    let staged = pending
        .receive_staged(&SandboxCoordinator::new(&fixture.store))
        .map_err(failure)?;
    let participant = sandbox_participant(&fixture.store);
    assert_eq!(participant.state(), ParticipantState::Intent);
    let lazaret_core::records::ParticipantOwnership::Sandbox(ownership) =
        participant.ownership().unwrap()
    else {
        unreachable!();
    };
    assert_eq!(
        ownership.attached_runsc.authoritative_pidfs().unwrap(),
        &staged.identity()
    );
    pending.abort().map_err(failure)?;
    if fixture.activity.exists() {
        return Err(failure("lost acknowledgment released sandbox work"));
    }
    Ok(())
}

pub(super) fn replaced_handoff_authority_refuses_target_work() -> Result<(), MatrixError> {
    let fixture = Fixture::new()?;
    let handoff = fixture
        .bundle
        .capture_handoff_authority(&fixture.configuration)
        .map_err(failure)?;
    let mut pending = fixture.launch_pending()?;
    let staged = pending
        .receive_staged(&SandboxCoordinator::new(&fixture.store))
        .map_err(failure)?;
    let workspace = host_path(fixture.configuration.workspace().canonical_label());
    let selected = workspace.with_extension("selected");
    fs::rename(&workspace, &selected)?;
    fs::create_dir(&workspace)?;
    let marker = workspace.join("replacement-marker");
    fs::write(&marker, b"unchanged")?;
    if pending.acknowledge(staged, &handoff).is_ok() {
        return Err(failure("replaced handoff authority released sandbox work"));
    }
    if fs::read(&marker)? != b"unchanged"
        || workspace.join("activity").exists()
        || selected.join("activity").exists()
    {
        return Err(failure("refused handoff mutated a workspace tree"));
    }
    Ok(())
}

fn cross_bundle_handoff_refuses_target_work() -> Result<(), MatrixError> {
    let fixture = Fixture::new()?;
    let handoff = fixture
        .bundle
        .capture_handoff_authority(&fixture.configuration)
        .map_err(failure)?;
    let alternate = fixture.alternate_bundle()?;
    if fixture.bundle.command().arguments() != alternate.command().arguments() {
        return Err(failure("cross-bundle fixture changed runsc argv"));
    }
    let mut pending = fixture.launch_pending_with_command(alternate.command())?;
    let staged = pending
        .receive_staged(&SandboxCoordinator::new(&fixture.store))
        .map_err(failure)?;
    if pending.acknowledge(staged, &handoff).is_ok() || fixture.activity.exists() {
        return Err(failure("cross-bundle handoff released sandbox work"));
    }
    Ok(())
}

pub(super) fn normal_stop_and_stop_resume() -> Result<(), MatrixError> {
    let fixture = Fixture::new()?;
    let mut runtime = fixture.launch_committed()?;
    wait_for_activity(&fixture.activity)?;
    runtime
        .signal_attached_runsc(AttachedRunscSignal::Stop)
        .map_err(failure)?;
    thread::sleep(Duration::from_millis(50));
    let stopped_runsc_size = fs::metadata(&fixture.activity)?.len();
    runtime
        .signal_attached_runsc(AttachedRunscSignal::Continue)
        .map_err(failure)?;
    wait_for_activity_growth(&fixture.activity, stopped_runsc_size)?;
    let supervisor = runtime
        .runtime_supervisor_identity()
        .ok_or_else(|| failure("Runtime Supervisor identity is absent"))?;
    runtime
        .signal_runtime_supervisor(Signal::STOP)
        .map_err(failure)?;
    wait_for_process_stop(supervisor)?;
    let stopped_size = fs::metadata(&fixture.activity)?.len();
    wait_for_activity_growth(&fixture.activity, stopped_size)
        .map_err(|_| failure("stopped Runtime Supervisor incorrectly stopped a present sandbox"))?;
    runtime
        .signal_runtime_supervisor(Signal::CONT)
        .map_err(failure)?;
    runtime
        .stop(&SandboxCoordinator::new(&fixture.store))
        .map_err(failure)?;
    assert_eq!(
        sandbox_participant(&fixture.store).state(),
        ParticipantState::Clean
    );
    assert_activity_stable(&fixture.activity)
}

pub(super) fn attached_runsc_loss_stops_guest() -> Result<(), MatrixError> {
    let fixture = Fixture::new()?;
    let mut runtime = fixture.launch_committed()?;
    wait_for_activity(&fixture.activity)?;
    runtime
        .signal_attached_runsc(AttachedRunscSignal::Kill)
        .map_err(failure)?;
    runtime.wait_for_attached_runsc_exit().map_err(failure)?;
    assert_eq!(
        sandbox_participant(&fixture.store).state(),
        ParticipantState::Committed
    );
    assert_activity_stable(&fixture.activity)
}

pub(super) fn runtime_supervisor_loss_stops_guest() -> Result<(), MatrixError> {
    let fixture = Fixture::new()?;
    let runtime = fixture.launch_committed()?;
    wait_for_activity(&fixture.activity)?;
    runtime
        .signal_runtime_supervisor(Signal::KILL)
        .map_err(failure)?;
    runtime
        .wait_for_runtime_supervisor_exit()
        .map_err(failure)?;
    assert_eq!(
        sandbox_participant(&fixture.store).state(),
        ParticipantState::Committed
    );
    assert_activity_stable(&fixture.activity)
}

pub(super) fn host_coordinator_loss_stops_guest() -> Result<(), MatrixError> {
    let tree = tempfile::tempdir()?;
    let ready = tree.path().join("host-coordinator-ready");
    let activity = tree.path().join("workspace/activity");
    let parent = current_process_identity()?;
    let target = TargetCommand::new(std::env::current_exe()?)
        .argument(HOST_COORDINATOR_DRIVER)
        .argument(tree.path())
        .argument(&ready);
    let mut driver = ExactChild::new(child_broker().acquire(
        parent,
        ProcessOwnershipContext {
            domain_id: lazaret_core::identity::DomainId::new([1; 32]),
            generation_id: GenerationId::new(7),
            coordinator_id: CoordinatorId::new([3; 32]),
            slot: HostProcessOwnershipSlot::SandboxHostCoordinator,
        },
        &target,
        |_| Ok(()),
        |staged| Ok(ProcessOwnershipCommitAck::v1(staged)),
    )?);
    for _ in 0..400 {
        if ready.is_file() {
            break;
        }
        if driver.process()?.has_exited()? {
            driver.reap()?;
            return Err(failure("Host Coordinator driver exited before readiness"));
        }
        thread::sleep(Duration::from_millis(5));
    }
    if !ready.is_file() {
        driver.kill_and_reap()?;
        return Err(failure("Host Coordinator driver readiness timed out"));
    }
    wait_for_activity(&activity)?;
    driver.kill_and_reap()?;
    thread::sleep(Duration::from_millis(100));
    assert_activity_stable(&activity)
}

struct ExactChild(Option<RetainedProcess>);

impl ExactChild {
    const fn new(process: RetainedProcess) -> Self {
        Self(Some(process))
    }

    fn process(&self) -> std::io::Result<&RetainedProcess> {
        self.0
            .as_ref()
            .ok_or_else(|| std::io::Error::other("exact test child handle is absent"))
    }

    fn reap(&mut self) -> std::io::Result<()> {
        let process = self
            .0
            .take()
            .ok_or_else(|| std::io::Error::other("exact test child handle is absent"))?;
        child_broker().reap(process)
    }

    fn kill_and_reap(&mut self) -> std::io::Result<()> {
        self.process()?.send_signal(Signal::KILL)?;
        self.reap()
    }
}

impl Drop for ExactChild {
    fn drop(&mut self) {
        if let Some(process) = self.0.take() {
            let _ = process.send_signal(Signal::KILL);
            let _ = child_broker().reap(process);
        }
    }
}

pub(super) fn run_host_coordinator_driver(arguments: &[std::ffi::OsString]) -> std::io::Result<()> {
    let tree = arguments
        .first()
        .map(PathBuf::from)
        .ok_or_else(|| std::io::Error::other("Host Coordinator fixture path is missing"))?;
    let ready = arguments
        .get(1)
        .map(PathBuf::from)
        .ok_or_else(|| std::io::Error::other("Host Coordinator ready path is missing"))?;
    if arguments.len() != 2 {
        return Err(std::io::Error::other(
            "Host Coordinator driver arguments are malformed",
        ));
    }
    let fixture = Fixture::in_directory(&tree).map_err(matrix_io)?;
    let _runtime = fixture.launch_committed().map_err(matrix_io)?;
    wait_for_activity(&fixture.activity).map_err(matrix_io)?;
    fs::write(ready, b"ready")?;
    loop {
        thread::park();
    }
}

pub(super) fn matrix_io(error: MatrixError) -> std::io::Error {
    match error {
        MatrixError::Skip(detail) | MatrixError::Failure(detail) => std::io::Error::other(detail),
    }
}

pub(super) fn failure(error: impl std::fmt::Display) -> MatrixError {
    MatrixError::Failure(error.to_string())
}

fn wait_for_activity(path: &Path) -> Result<(), MatrixError> {
    for _ in 0..400 {
        if path.metadata().is_ok_and(|metadata| metadata.len() >= 2) {
            return Ok(());
        }
        thread::sleep(Duration::from_millis(5));
    }
    Err(failure("guest activity did not start"))
}

fn wait_for_activity_growth(path: &Path, previous_size: u64) -> Result<(), MatrixError> {
    for _ in 0..400 {
        if path
            .metadata()
            .is_ok_and(|metadata| metadata.len() > previous_size)
        {
            return Ok(());
        }
        thread::sleep(Duration::from_millis(5));
    }
    Err(failure("guest activity did not advance"))
}

fn wait_for_process_stop(identity: PidfsProcessIdentityV1) -> Result<(), MatrixError> {
    let host_pid = identity.host_pid().get();
    for _ in 0..400 {
        if read_start_time(host_pid)? != identity.start_time_ticks().get() {
            return Err(failure(
                "Runtime Supervisor identity changed while observing stop",
            ));
        }
        let stat = fs::read_to_string(format!("/proc/{host_pid}/stat"))?;
        let state = stat
            .rsplit_once(") ")
            .and_then(|(_, suffix)| suffix.as_bytes().first())
            .copied();
        if state == Some(b'T') {
            return Ok(());
        }
        thread::sleep(Duration::from_millis(5));
    }
    Err(failure("Runtime Supervisor did not reach stopped state"))
}

fn assert_activity_stable(path: &Path) -> Result<(), MatrixError> {
    let before = fs::metadata(path)?.len();
    thread::sleep(Duration::from_millis(100));
    let after = fs::metadata(path)?.len();
    if before == after {
        Ok(())
    } else {
        Err(failure(
            "guest activity continued after causal authority loss",
        ))
    }
}