lazaret/crates/lazaret-host-coordinator/src/runtime.rs

use std::fmt;
use std::io;
use std::os::unix::ffi::OsStrExt;
use std::path::Path;
use std::time::{Duration, Instant};

use crate::state_anchor::StateAnchorCoordinator;
use lazaret_core::identity::{
    BundleDigest, ContainerId, GenerationId, HostPath, MountIdentity, NamespaceIdentity,
    PidfsProcessIdentityV1, PreparedArtifactId,
};
use lazaret_core::network::CanonicalDerivedNetworkPolicy;
use lazaret_core::records::{
    HostProcessOwnershipSlot, NetworkReadinessAttestation, NetworkReadinessEvidence, Participant,
    ParticipantOwnership, ParticipantRole, ParticipantState, PersistedHostProcessOwnership,
    ProcessOwnershipCommitAck, SandboxOwnership, StagedProcessOwnership,
};
use lazaret_linux::ipc::private_seqpacket_pair;
use lazaret_linux::process::{
    AcknowledgedChildAcquisition, RetainedProcess, child_broker, current_process_identity,
};
use lazaret_runtime_supervisor::sandbox::SandboxHandoffAuthority;
use lazaret_runtime_supervisor::supervision::{
    AttachedRunscSignal, CoordinatorChannel, RuntimeSupervisorLaunch, SupervisorEvent,
};
use rustix::pipe::{PipeFlags, pipe_with};
use rustix::process::Signal;

use crate::domain_store::{DomainStoreError, DomainStoreMutation, HostCoordinatorDomainStore};
use crate::git_routing::GitRoutingCoordinator;
use crate::network::{NetworkCoordinator, NetworkCoordinatorError};

const SUPERVISOR_EXIT_TIMEOUT: Duration = Duration::from_secs(5);

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub struct StateAnchorAbsenceObservation {
    pub user_namespace: NamespaceIdentity,
    pub mount_namespace: NamespaceIdentity,
    pub mount: MountIdentity,
    pub observed: MountIdentity,
}

#[derive(Debug)]
pub enum SandboxCoordinatorError {
    Store(DomainStoreError),
    Network(NetworkCoordinatorError),
    Io(io::Error),
    InvalidState(&'static str),
}

impl fmt::Display for SandboxCoordinatorError {
    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self {
            Self::Store(error) => {
                write!(formatter, "Sandbox Domain Store operation failed: {error}")
            }
            Self::Network(error) => write!(formatter, "Sandbox network operation failed: {error}"),
            Self::Io(error) => write!(formatter, "Sandbox process operation failed: {error}"),
            Self::InvalidState(detail) => write!(formatter, "Sandbox state refused: {detail}"),
        }
    }
}

impl std::error::Error for SandboxCoordinatorError {
    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
        match self {
            Self::Store(error) => Some(error),
            Self::Network(error) => Some(error),
            Self::Io(error) => Some(error),
            Self::InvalidState(_) => None,
        }
    }
}

impl From<DomainStoreError> for SandboxCoordinatorError {
    fn from(error: DomainStoreError) -> Self {
        Self::Store(error)
    }
}

impl From<io::Error> for SandboxCoordinatorError {
    fn from(error: io::Error) -> Self {
        Self::Io(error)
    }
}

impl From<NetworkCoordinatorError> for SandboxCoordinatorError {
    fn from(error: NetworkCoordinatorError) -> Self {
        Self::Network(error)
    }
}

pub struct SandboxCoordinator<'store> {
    domain_store: &'store HostCoordinatorDomainStore,
}

impl<'store> SandboxCoordinator<'store> {
    #[must_use]
    pub const fn new(domain_store: &'store HostCoordinatorDomainStore) -> Self {
        Self { domain_store }
    }

    /// Persists bare Sandbox intent and the immutable physical launch snapshot.
    ///
    /// # Errors
    /// Refuses a missing generation, changed physical identity, or existing nonmatching state.
    pub fn begin_intent(
        &self,
        generation_id: GenerationId,
        container_id: ContainerId,
        runtime_root: &Path,
        runtime_root_identity: MountIdentity,
        bundle_digest: BundleDigest,
        host_coordinator: PidfsProcessIdentityV1,
    ) -> Result<(), SandboxCoordinatorError> {
        let snapshot = self.domain_store.snapshot()?;
        if !snapshot
            .generations
            .iter()
            .any(|generation| generation.generation_id == generation_id)
        {
            return Err(SandboxCoordinatorError::InvalidState(
                "Sandbox generation is absent",
            ));
        }
        let ownership = SandboxOwnership {
            container_id: Some(container_id),
            runtime_root: Some(HostPath::new(runtime_root.as_os_str().as_bytes().to_vec())),
            runtime_root_identity: Some(runtime_root_identity),
            bundle_digest: Some(bundle_digest),
            host_coordinator: PersistedHostProcessOwnership::committed(host_coordinator),
            runtime_supervisor: PersistedHostProcessOwnership::unassigned(),
            attached_runsc: PersistedHostProcessOwnership::unassigned(),
        };
        let intent = sandbox_participant(generation_id, ParticipantState::Intent, ownership)?;
        match sandbox_role(&snapshot.participants, generation_id) {
            None => self.domain_store.apply(&[
                DomainStoreMutation::PutParticipant(Participant::bare_intent(
                    generation_id,
                    ParticipantRole::Sandbox,
                )),
                DomainStoreMutation::PutParticipant(intent),
            ])?,
            Some(current) if current == &intent => {}
            Some(_) => {
                return Err(SandboxCoordinatorError::InvalidState(
                    "Sandbox intent already differs",
                ));
            }
        }
        Ok(())
    }

    /// # Errors
    /// Refuses a mismatched RS tuple or any skipped ownership phase.
    pub fn stage_runtime_supervisor(
        &self,
        staged: StagedProcessOwnership,
    ) -> Result<(), SandboxCoordinatorError> {
        self.require_staged(staged, HostProcessOwnershipSlot::SandboxRuntimeSupervisor)?;
        let (current, mut ownership) = self.intent_ownership(staged.generation_id())?;
        let expected = PersistedHostProcessOwnership::staged(staged.identity());
        if ownership.runtime_supervisor == expected {
            return Ok(());
        }
        if !ownership.runtime_supervisor.is_unassigned() {
            return Err(SandboxCoordinatorError::InvalidState(
                "Runtime Supervisor ownership changed",
            ));
        }
        ownership.runtime_supervisor = expected;
        self.put_intent(current.generation_id(), ownership)
    }

    /// # Errors
    /// Refuses an RS staged identity that is not durably exact.
    pub fn acknowledge_runtime_supervisor(
        &self,
        staged: StagedProcessOwnership,
    ) -> Result<ProcessOwnershipCommitAck, SandboxCoordinatorError> {
        self.require_staged(staged, HostProcessOwnershipSlot::SandboxRuntimeSupervisor)?;
        let (_, ownership) = self.intent_ownership(staged.generation_id())?;
        if ownership.runtime_supervisor != PersistedHostProcessOwnership::staged(staged.identity())
        {
            return Err(SandboxCoordinatorError::InvalidState(
                "staged Runtime Supervisor ownership is not durable",
            ));
        }
        Ok(ProcessOwnershipCommitAck::v1(staged))
    }

    /// Commits the executing RS and durably stages its exact attached-runsc child.
    ///
    /// # Errors
    /// Refuses any tuple, phase, or immutable-evidence mismatch.
    pub fn stage_attached_runsc(
        &self,
        staged: StagedProcessOwnership,
    ) -> Result<ProcessOwnershipCommitAck, SandboxCoordinatorError> {
        self.require_staged(staged, HostProcessOwnershipSlot::SandboxAttachedRunsc)?;
        let (current, mut ownership) = self.intent_ownership(staged.generation_id())?;
        let rs = *ownership
            .runtime_supervisor
            .authoritative_pidfs()
            .map_err(|_| {
                SandboxCoordinatorError::InvalidState("Runtime Supervisor is not staged")
            })?;
        if !ownership.runtime_supervisor.is_staged() {
            return Err(SandboxCoordinatorError::InvalidState(
                "Runtime Supervisor readiness cannot skip staging",
            ));
        }
        if !ownership.attached_runsc.is_unassigned() {
            return Err(SandboxCoordinatorError::InvalidState(
                "attached runsc ownership changed",
            ));
        }
        ownership.runtime_supervisor = PersistedHostProcessOwnership::committed(rs);
        ownership.attached_runsc = PersistedHostProcessOwnership::staged(staged.identity());
        self.put_intent(current.generation_id(), ownership)?;
        Ok(ProcessOwnershipCommitAck::v1(staged))
    }

    /// Promotes the exact attached runsc only after the RS reports target readiness.
    ///
    /// # Errors
    /// Refuses any incomplete physical snapshot or changed staged identity.
    pub fn commit_ready(
        &self,
        staged: StagedProcessOwnership,
    ) -> Result<(), SandboxCoordinatorError> {
        self.require_staged(staged, HostProcessOwnershipSlot::SandboxAttachedRunsc)?;
        let (_, mut ownership) = self.intent_ownership(staged.generation_id())?;
        if ownership.attached_runsc != PersistedHostProcessOwnership::staged(staged.identity())
            || !ownership.runtime_supervisor.is_current_committed()
            || !ownership.host_coordinator.is_current_committed()
        {
            return Err(SandboxCoordinatorError::InvalidState(
                "Sandbox readiness does not match staged ownership",
            ));
        }
        ownership.attached_runsc = PersistedHostProcessOwnership::committed(staged.identity());
        self.domain_store
            .apply(&[DomainStoreMutation::PutParticipant(sandbox_participant(
                staged.generation_id(),
                ParticipantState::Committed,
                ownership,
            )?)])?;
        Ok(())
    }

    /// # Errors
    /// Refuses any Sandbox participant that is not exactly Committed.
    pub fn begin_release(
        &self,
        generation_id: GenerationId,
    ) -> Result<(), SandboxCoordinatorError> {
        let current = self.participant(generation_id)?;
        if current.state() != ParticipantState::Committed {
            return Err(SandboxCoordinatorError::InvalidState(
                "Sandbox is not committed",
            ));
        }
        let ownership =
            current
                .ownership()
                .cloned()
                .ok_or(SandboxCoordinatorError::InvalidState(
                    "Sandbox ownership is absent",
                ))?;
        self.domain_store
            .apply(&[DomainStoreMutation::PutParticipant(
                Participant::snapshot(
                    generation_id,
                    ParticipantRole::Sandbox,
                    ParticipantState::ReleaseIntent,
                    Some(ownership),
                )
                .map_err(|_| SandboxCoordinatorError::InvalidState("Sandbox release is invalid"))?,
            )])?;
        Ok(())
    }

    /// # Errors
    /// Refuses cleanup without exact `ReleaseIntent` and observed runtime absence.
    pub fn complete_clean_after_absence(
        &self,
        generation_id: GenerationId,
    ) -> Result<(), SandboxCoordinatorError> {
        let current = self.participant(generation_id)?;
        if current.state() != ParticipantState::ReleaseIntent {
            return Err(SandboxCoordinatorError::InvalidState(
                "Sandbox release intent is absent",
            ));
        }
        let clean = Participant::snapshot(
            generation_id,
            ParticipantRole::Sandbox,
            ParticipantState::Clean,
            Some(ParticipantOwnership::empty(ParticipantRole::Sandbox)),
        )
        .map_err(|_| SandboxCoordinatorError::InvalidState("Sandbox Clean is invalid"))?;
        self.domain_store
            .apply(&[DomainStoreMutation::CleanParticipantAfterAbsence(clean)])?;
        Ok(())
    }

    fn require_staged(
        &self,
        staged: StagedProcessOwnership,
        slot: HostProcessOwnershipSlot,
    ) -> Result<(), SandboxCoordinatorError> {
        let snapshot = self.domain_store.snapshot()?;
        let generation = snapshot
            .generations
            .iter()
            .find(|generation| generation.generation_id == staged.generation_id())
            .ok_or(SandboxCoordinatorError::InvalidState(
                "staged Sandbox generation is absent",
            ))?;
        if staged.domain_id() != snapshot.domain.domain_id
            || staged.coordinator_id() != generation.coordinator_id
            || staged.slot() != slot
        {
            return Err(SandboxCoordinatorError::InvalidState(
                "staged Sandbox ownership tuple mismatches generation",
            ));
        }
        Ok(())
    }

    fn intent_ownership(
        &self,
        generation_id: GenerationId,
    ) -> Result<(Participant, SandboxOwnership), SandboxCoordinatorError> {
        let current = self.participant(generation_id)?;
        if current.state() != ParticipantState::Intent {
            return Err(SandboxCoordinatorError::InvalidState(
                "Sandbox acquisition is not in Intent",
            ));
        }
        let Some(ParticipantOwnership::Sandbox(ownership)) = current.ownership() else {
            return Err(SandboxCoordinatorError::InvalidState(
                "Sandbox ownership shape is invalid",
            ));
        };
        Ok((current.clone(), ownership.clone()))
    }

    fn put_intent(
        &self,
        generation_id: GenerationId,
        ownership: SandboxOwnership,
    ) -> Result<(), SandboxCoordinatorError> {
        self.domain_store
            .apply(&[DomainStoreMutation::PutParticipant(sandbox_participant(
                generation_id,
                ParticipantState::Intent,
                ownership,
            )?)])?;
        Ok(())
    }

    fn participant(
        &self,
        generation_id: GenerationId,
    ) -> Result<Participant, SandboxCoordinatorError> {
        sandbox_role(&self.domain_store.snapshot()?.participants, generation_id)
            .cloned()
            .ok_or(SandboxCoordinatorError::InvalidState(
                "Sandbox participant is absent",
            ))
    }
}

mod acquisition;
mod committed;

pub use acquisition::{AwaitingAttachedRunsc, AwaitingSandboxReadiness};
pub use committed::CommittedSandboxRuntime;
use committed::{cleanup_supervisor, sandbox_participant, sandbox_role};
#[cfg(test)]
mod tests;