lazaret/crates/lazaret-runtime-supervisor/src/supervision/runtime.rs

fn runsc_target(arguments: &RuntimeArguments, pid_file: i32) -> io::Result<TargetCommand> {
    let run = arguments
        .runsc_arguments
        .iter()
        .position(|argument| argument == "run")
        .ok_or_else(|| io::Error::other("attached runsc command has no run operation"))?;
    let mut target = TargetCommand::new(arguments.runsc_program.clone());
    for argument in &arguments.runsc_arguments[..=run] {
        target = target.argument(argument);
    }
    target = target.argument(format!("-pid-file=/proc/self/fd/{pid_file}"));
    for argument in &arguments.runsc_arguments[run + 1..] {
        target = target.argument(argument);
    }
    Ok(target.with_parent_death_kill_edge())
}

fn runsc_delete_target(arguments: &RuntimeArguments) -> io::Result<TargetCommand> {
    let run = arguments
        .runsc_arguments
        .iter()
        .position(|argument| argument == "run")
        .ok_or_else(|| io::Error::other("attached runsc command has no run operation"))?;
    let container = arguments
        .runsc_arguments
        .last()
        .filter(|container| {
            !container.is_empty() && !container.as_encoded_bytes().starts_with(b"-")
        })
        .ok_or_else(|| io::Error::other("attached runsc command has no container ID"))?;
    let mut target = TargetCommand::new(arguments.runsc_program.clone());
    for argument in &arguments.runsc_arguments[..run] {
        target = target.argument(argument);
    }
    Ok(target
        .argument("delete")
        .argument("--force")
        .argument(container))
}

fn attached_runsc_standard_io(target: &TargetCommand) -> io::Result<ChildStandardIo> {
    if target.parent_edge() != ParentEdge::ParentDeathKill {
        return Err(io::Error::other(
            "attached runsc requires the parent-death KILL edge",
        ));
    }
    Ok(ChildStandardIo::silenced())
}

fn wait_for_runsc_ready(pid_file: &File, process: &RetainedProcess) -> io::Result<u32> {
    let deadline = Instant::now() + READINESS_TIMEOUT;
    let mut bytes = [0_u8; 32];
    loop {
        require_lifetime_held()?;
        if process.has_exited()? {
            return Err(io::Error::other("attached runsc exited before readiness"));
        }
        let length = pid_file.read_at(&mut bytes, 0)?;
        if length > 0 {
            return std::str::from_utf8(&bytes[..length])
                .ok()
                .map(str::trim)
                .and_then(|value| value.parse().ok())
                .filter(|pid| *pid > 0)
                .ok_or_else(|| io::Error::other("runsc pid-file readiness is malformed"));
        }
        if Instant::now() >= deadline {
            return Err(io::Error::other("attached runsc readiness timed out"));
        }
        std::thread::sleep(Duration::from_millis(5));
    }
}

#[allow(clippy::too_many_arguments, clippy::too_many_lines)]
fn supervise(
    control: &std::io::Stdout,
    process: RetainedProcess,
    mut git_routing: Option<RuntimeGitRouting>,
    mut network: Option<PreparedNetwork>,
    namespaces: (NamespaceIdentity, NamespaceIdentity),
    state_anchor: &mut Option<RuntimeStateAnchorMount>,
    state_anchor_absence: &mut Option<RuntimeStateAnchorAbsenceAuthority>,
    delete: &TargetCommand,
) -> io::Result<()> {
    let mut process = Some(process);
    let mut state_anchor = StateAnchorSupervision {
        namespaces,
        mount: state_anchor,
        absence: state_anchor_absence,
    };
    loop {
        let exited = match process
            .as_ref()
            .map_or(Ok(false), RetainedProcess::has_exited)
        {
            Ok(exited) => exited,
            Err(error) => return fail_active_runtime(process, git_routing, network, error),
        };
        if exited {
            let runsc = process
                .take()
                .ok_or_else(|| io::Error::other("attached runsc handle is absent"))?;
            cleanup_stopped_runtime(runsc, git_routing, network, delete, false)?;
            send_message(control, &SupervisorEvent::AttachedRunscExited)?;
            return Ok(());
        }
        if git_routing
            .as_ref()
            .is_some_and(|routing| routing.process.has_exited().unwrap_or(true))
        {
            return fail_active_runtime(
                process,
                git_routing,
                network,
                io::Error::other("Git router exited while the sandbox was active"),
            );
        }
        if network
            .as_ref()
            .is_some_and(|network| network.helper_exited().unwrap_or(true))
        {
            network
                .take()
                .ok_or_else(|| io::Error::other("connectivity helper handle is absent"))?
                .reap_exited()?;
            send_message(control, &SupervisorEvent::HelperExited)?;
        }
        let lease_lost = match lifetime_lost() {
            Ok(lost) => lost,
            Err(error) => return fail_active_runtime(process, git_routing, network, error),
        };
        if lease_lost {
            if let Some(process) = process.take() {
                cleanup_stopped_runtime(process, git_routing, network, delete, true)?;
            } else {
                cleanup_runtime_components(git_routing, network)?;
            }
            return Ok(());
        }
        let mut descriptors = [PollFd::new(control, PollFlags::IN)];
        if let Err(error) = poll(
            &mut descriptors,
            Some(&Timespec {
                tv_sec: 0,
                tv_nsec: 10_000_000,
            }),
        ) {
            return fail_active_runtime(process, git_routing, network, error.into());
        }
        let events = descriptors[0].revents();
        if events.intersects(PollFlags::HUP | PollFlags::ERR) {
            return fail_active_runtime(
                process,
                git_routing,
                network,
                io::Error::new(
                    io::ErrorKind::UnexpectedEof,
                    "runtime control channel closed",
                ),
            );
        }
        if events.intersects(PollFlags::IN) {
            let message = match receive_message(control) {
                Ok(message) => message,
                Err(error) => {
                    return fail_active_runtime(process, git_routing, network, error);
                }
            };
            match handle_supervisor_message(
                control,
                &message,
                &mut process,
                &mut git_routing,
                &mut network,
                &mut state_anchor,
                delete,
            ) {
                Ok(true) => return Ok(()),
                Ok(false) => {}
                Err(error) => {
                    return fail_active_runtime(process, git_routing, network, error);
                }
            }
        }
    }
}

struct StateAnchorSupervision<'a> {
    namespaces: (NamespaceIdentity, NamespaceIdentity),
    mount: &'a mut Option<RuntimeStateAnchorMount>,
    absence: &'a mut Option<RuntimeStateAnchorAbsenceAuthority>,
}

fn handle_supervisor_message(
    control: &std::io::Stdout,
    message: &CoordinatorMessage,
    process: &mut Option<RetainedProcess>,
    git_routing: &mut Option<RuntimeGitRouting>,
    network: &mut Option<PreparedNetwork>,
    state_anchor: &mut StateAnchorSupervision<'_>,
    delete: &TargetCommand,
) -> io::Result<bool> {
    match message {
        CoordinatorMessage::SignalAttachedRunsc(signal) => {
            process
                .as_ref()
                .ok_or_else(|| io::Error::other("attached runsc is already stopped"))?
                .send_signal(signal.signal())?;
            send_message(control, &SupervisorEvent::SignalApplied(*signal))?;
        }
        CoordinatorMessage::StopSandbox => {
            terminate_and_reap(
                process
                    .take()
                    .ok_or_else(|| io::Error::other("attached runsc is already stopped"))?,
            )?;
            child_broker().run_transient(delete)?;
            send_message(control, &SupervisorEvent::SandboxStopped)?;
        }
        CoordinatorMessage::StopGitRouting => {
            if process.is_some() {
                return Err(io::Error::other("Git-routing stop preceded sandbox stop"));
            }
            let routing = git_routing
                .take()
                .ok_or_else(|| io::Error::other("Git routing is already stopped"))?;
            let mount = routing.mount;
            stop_git_routing(Some(routing))?;
            send_message(control, &SupervisorEvent::GitRoutingStopped { mount })?;
        }
        CoordinatorMessage::StopSidecar => {
            if process.is_some() {
                return Err(io::Error::other("sidecar stop preceded sandbox stop"));
            }
            stop_network(network.take())?;
            send_message(control, &SupervisorEvent::SidecarStopped)?;
        }
        CoordinatorMessage::UnmountStateAnchor
        | CoordinatorMessage::ObserveStateAnchorAbsence { .. } => {
            handle_state_anchor_message(
                control,
                message,
                process.is_some() || git_routing.is_some() || network.is_some(),
                state_anchor,
            )?;
        }
        CoordinatorMessage::Exit => {
            if process.is_some()
                || git_routing.is_some()
                || network.is_some()
                || state_anchor.mount.is_some()
            {
                return Err(io::Error::other(
                    "Runtime Supervisor exit preceded resource cleanup",
                ));
            }
            send_message(control, &SupervisorEvent::Exited)?;
            return Ok(true);
        }
        CoordinatorMessage::Stop => {
            if let Some(process) = process.take() {
                terminate_and_reap(process)?;
            }
            stop_network(network.take())?;
            stop_git_routing(git_routing.take())?;
            send_message(control, &SupervisorEvent::Stopped)?;
            return Ok(true);
        }
        CoordinatorMessage::HelperStagedAck(_)
        | CoordinatorMessage::MountStateAnchor
        | CoordinatorMessage::StateAnchorStagedAck(_)
        | CoordinatorMessage::StartGitRouting
        | CoordinatorMessage::GitRoutingStagedAck(_)
        | CoordinatorMessage::StartSidecar
        | CoordinatorMessage::NetworkAttestationAccepted(_)
        | CoordinatorMessage::NetworkReadinessCommitted
        | CoordinatorMessage::StagedAck(_)
        | CoordinatorMessage::SandboxCommitted => {
            return Err(io::Error::other("unexpected duplicate coordinator message"));
        }
    }
    Ok(false)
}

fn handle_state_anchor_message(
    control: &std::io::Stdout,
    message: &CoordinatorMessage,
    runtime_resources_present: bool,
    state: &mut StateAnchorSupervision<'_>,
) -> io::Result<()> {
    match message {
        CoordinatorMessage::UnmountStateAnchor => {
            if runtime_resources_present {
                return Err(io::Error::other(
                    "State Anchor unmount preceded runtime resource stop",
                ));
            }
            let mount = state
                .mount
                .as_ref()
                .ok_or_else(|| io::Error::other("State Anchor is absent"))?
                .mount_identity();
            let absence = state
                .mount
                .take()
                .ok_or_else(|| io::Error::other("State Anchor is absent"))?
                .unmount()
                .map_err(io::Error::other)?;
            *state.absence = Some(absence);
            send_message(control, &SupervisorEvent::StateAnchorUnmounted { mount })
        }
        CoordinatorMessage::ObserveStateAnchorAbsence {
            user_namespace,
            mount_namespace,
            mount,
        } => {
            if (*user_namespace, *mount_namespace) != state.namespaces {
                return send_message(
                    control,
                    &SupervisorEvent::Refused(
                        "State Anchor absence request namespace authority mismatched".into(),
                    ),
                );
            }
            match state.absence.as_ref().map_or_else(
                || {
                    Err(lazaret_state_anchor::StateAnchorError::InvalidLayout(
                        "State Anchor absence authority is absent",
                    ))
                },
                |authority| authority.observe_absence(*mount),
            ) {
                Ok(observed) => send_message(
                    control,
                    &SupervisorEvent::StateAnchorAbsenceObserved {
                        user_namespace: state.namespaces.0,
                        mount_namespace: state.namespaces.1,
                        mount: *mount,
                        observed,
                    },
                ),
                Err(error) => send_message(control, &SupervisorEvent::Refused(error.to_string())),
            }
        }
        _ => Err(io::Error::other("message is not a State Anchor command")),
    }
}

fn fail_active_runtime(
    process: Option<RetainedProcess>,
    git_routing: Option<RuntimeGitRouting>,
    network: Option<PreparedNetwork>,
    error: io::Error,
) -> io::Result<()> {
    if let Some(process) = process {
        fail_with_runtime_cleanup(process, git_routing, network, error)
    } else {
        fail_runtime_components(git_routing, network, error)
    }
}

fn terminate_and_reap(process: RetainedProcess) -> io::Result<()> {
    let term_error = process.send_signal(Signal::TERM).err();
    let deadline = Instant::now() + TERMINATION_GRACE;
    while !process.has_exited()? && Instant::now() < deadline {
        std::thread::sleep(Duration::from_millis(5));
    }
    if !process.has_exited()? {
        process.send_signal(Signal::KILL)?;
    }
    child_broker().reap(process)?;
    if let Some(error) = term_error {
        Err(error)
    } else {
        Ok(())
    }
}

fn fail_with_runtime_cleanup(
    process: RetainedProcess,
    git_routing: Option<RuntimeGitRouting>,
    network: Option<PreparedNetwork>,
    error: io::Error,
) -> io::Result<()> {
    let runsc_cleanup = terminate_and_reap(process);
    let network_cleanup = stop_network(network);
    let routing_cleanup = stop_git_routing(git_routing);
    match (runsc_cleanup, routing_cleanup, network_cleanup) {
        (Ok(()), Ok(()), Ok(())) => Err(error),
        (runsc, routing, network) => Err(io::Error::new(
            error.kind(),
            format!(
                "{error}; exact runtime cleanup failed: runsc={runsc:?}, Git-routing={routing:?}, network={network:?}"
            ),
        )),
    }
}

fn cleanup_stopped_runtime(
    process: RetainedProcess,
    git_routing: Option<RuntimeGitRouting>,
    network: Option<PreparedNetwork>,
    delete: &TargetCommand,
    terminate: bool,
) -> io::Result<()> {
    let runsc_cleanup = if terminate {
        terminate_and_reap(process)
    } else {
        child_broker().reap(process)
    };
    let delete_cleanup = child_broker().run_transient(delete).map(drop);
    let network_cleanup = stop_network(network);
    let routing_cleanup = stop_git_routing(git_routing);
    let failures = [
        ("attached runsc", runsc_cleanup),
        ("runsc container", delete_cleanup),
        ("network", network_cleanup),
        ("Git routing", routing_cleanup),
    ]
    .into_iter()
    .filter_map(|(component, result)| result.err().map(|error| format!("{component}: {error}")))
    .collect::<Vec<_>>();
    if failures.is_empty() {
        Ok(())
    } else {
        Err(io::Error::other(failures.join("; ")))
    }
}

fn fail_runtime_components<T>(
    git_routing: Option<RuntimeGitRouting>,
    network: Option<PreparedNetwork>,
    error: io::Error,
) -> io::Result<T> {
    match cleanup_runtime_components(git_routing, network) {
        Ok(()) => Err(error),
        Err(cleanup) => Err(io::Error::new(
            error.kind(),
            format!("{error}; exact component cleanup failed: {cleanup}"),
        )),
    }
}

fn cleanup_runtime_components(
    git_routing: Option<RuntimeGitRouting>,
    network: Option<PreparedNetwork>,
) -> io::Result<()> {
    let routing_cleanup = stop_git_routing(git_routing);
    let network_cleanup = stop_network(network);
    match (routing_cleanup, network_cleanup) {
        (Ok(()), Ok(())) => Ok(()),
        (routing, network) => Err(io::Error::other(format!(
            "Git-routing={routing:?}, network={network:?}"
        ))),
    }
}

fn stop_git_routing(routing: Option<RuntimeGitRouting>) -> io::Result<()> {
    let Some(routing) = routing else {
        return Ok(());
    };
    let RuntimeGitRouting {
        process,
        shutdown,
        sentry,
        mountpoint,
        mount,
    } = routing;
    let sentry_cleanup = sentry.unmount();
    drop(shutdown);
    let process_cleanup = wait_and_reap(process);
    let mount_cleanup = inherited_directory_at(&mountpoint)
        .and_then(mount_identity)
        .and_then(|observed| {
            if observed == mount {
                Err(io::Error::other(
                    "Git-routing mount remains after router shutdown",
                ))
            } else {
                Ok(())
            }
        });
    let failures = [
        ("Git Sentry", sentry_cleanup),
        ("router process", process_cleanup),
        ("router mount", mount_cleanup),
    ]
    .into_iter()
    .filter_map(|(component, result)| result.err().map(|error| format!("{component}: {error}")))
    .collect::<Vec<_>>();
    if !failures.is_empty() {
        return Err(io::Error::other(failures.join("; ")));
    }
    Ok(())
}

fn wait_and_reap(process: RetainedProcess) -> io::Result<()> {
    let deadline = Instant::now() + TERMINATION_GRACE;
    while !process.has_exited()? && Instant::now() < deadline {
        std::thread::sleep(Duration::from_millis(5));
    }
    if !process.has_exited()? {
        return terminate_and_reap(process).and_then(|()| {
            Err(io::Error::other(
                "Git router did not exit after its shutdown channel closed",
            ))
        });
    }
    child_broker().reap(process)
}

fn fail_network<T>(network: Option<PreparedNetwork>, error: io::Error) -> io::Result<T> {
    match stop_network(network) {
        Ok(()) => Err(error),
        Err(cleanup) => Err(io::Error::new(
            error.kind(),
            format!("{error}; exact connectivity-helper cleanup failed: {cleanup}"),
        )),
    }
}

fn stop_network(network: Option<PreparedNetwork>) -> io::Result<()> {
    match network {
        Some(network) => network.stop(),
        None => Ok(()),
    }
}

fn require_lifetime_held() -> io::Result<()> {
    if lifetime_lost()? {
        Err(io::Error::other("Host Coordinator lifetime lease is lost"))
    } else {
        Ok(())
    }
}

fn lifetime_lost() -> io::Result<bool> {
    let stderr = std::io::stderr();
    let mut descriptors = [PollFd::new(&stderr, PollFlags::IN)];
    poll(&mut descriptors, Some(&Timespec::default()))?;
    Ok(descriptors[0]
        .revents()
        .intersects(PollFlags::IN | PollFlags::HUP | PollFlags::ERR))
}

fn send_message<T: lazaret_core::records::SchemaType + Serialize>(
    descriptor: impl AsFd,
    message: &T,
) -> io::Result<()> {
    let encoded = encode(message).map_err(|error| io::Error::other(error.to_string()))?;
    if encoded.len() > MAX_CONTROL_MESSAGE {
        return Err(io::Error::other("runtime control message is oversized"));
    }
    let length = u32::try_from(encoded.len())
        .map_err(|_| io::Error::other("runtime control message is oversized"))?
        .to_be_bytes();
    send_all(descriptor.as_fd(), &length)?;
    send_all(descriptor.as_fd(), &encoded)
}

fn receive_message<T>(descriptor: impl AsFd) -> io::Result<T>
where
    T: lazaret_core::records::SchemaType + serde::de::DeserializeOwned + Serialize,
{
    let mut length = [0_u8; 4];
    receive_exact(descriptor.as_fd(), &mut length)?;
    let length = usize::try_from(u32::from_be_bytes(length))
        .map_err(|_| io::Error::other("runtime control length is invalid"))?;
    if length > MAX_CONTROL_MESSAGE {
        return Err(io::Error::other("runtime control message is oversized"));
    }
    let mut encoded = vec![0_u8; length];
    receive_exact(descriptor.as_fd(), &mut encoded)?;
    decode(&encoded).map_err(|error| io::Error::other(error.to_string()))
}

fn send_all(descriptor: impl AsFd + Copy, mut bytes: &[u8]) -> io::Result<()> {
    while !bytes.is_empty() {
        let sent = write(descriptor, bytes)?;
        if sent == 0 {
            return Err(io::Error::new(
                io::ErrorKind::WriteZero,
                "runtime control channel closed",
            ));
        }
        bytes = &bytes[sent..];
    }
    Ok(())
}

fn receive_exact(descriptor: impl AsFd + Copy, mut bytes: &mut [u8]) -> io::Result<()> {
    while !bytes.is_empty() {
        let received = read(descriptor, &mut *bytes)?;
        if received == 0 {
            return Err(io::Error::new(
                io::ErrorKind::UnexpectedEof,
                "runtime control channel closed",
            ));
        }
        bytes = &mut bytes[received..];
    }
    Ok(())
}