saddle-runtime 0.2.0-rc.5

//! Isolated 0.2 physical-thread Runtime candidate.
//!
//! This module is not wired into `Application` and does not freeze an API. It
//! models the approved two-worker profile and exposes an internal probe used
//! by the audit binary.

#![allow(dead_code)]

use std::{
    collections::BTreeMap,
    fs,
    sync::{
        Arc, Barrier,
        atomic::{AtomicBool, Ordering},
    },
    time::{Duration, Instant},
};

const PROFILE_ID: &str = "saddle-0.2-linux-x86_64-2w";
const PROFILE_SHA256: &str = "7f6b8c2b3a600114959d97edd7b79ae409847e01cec2f760b9a7ce6d26123de4";
const CONTROL_NAME: &str = "saddle-control";
const WORKER_NAME: &str = "saddle-worker";
const WORKER_THREADS: usize = 2;
const STACK_BYTES: usize = 2_097_152;
const APPROVED_CPUSET: &[usize] = &[0, 1];
const INVENTORY_DEADLINE: Duration = Duration::from_secs(5);

static RUNTIME_LEASED: AtomicBool = AtomicBool::new(false);

#[derive(Debug, Eq, PartialEq)]
pub(crate) enum ThreadControlError {
    HostProfileMismatch,
    RuntimeAlreadyExists,
    RuntimeBuildFailed,
    InventoryTimeout,
    InventoryMismatch,
    Io,
}

pub(crate) struct ApprovedRuntime {
    runtime: Option<tokio::runtime::Runtime>,
}

impl ApprovedRuntime {
    pub(crate) fn build(cpu_quota: usize, cpuset: &[usize]) -> Result<Self, ThreadControlError> {
        if cpu_quota != WORKER_THREADS || cpuset != APPROVED_CPUSET {
            return Err(ThreadControlError::HostProfileMismatch);
        }
        if RUNTIME_LEASED
            .compare_exchange(false, true, Ordering::AcqRel, Ordering::Acquire)
            .is_err()
        {
            return Err(ThreadControlError::RuntimeAlreadyExists);
        }

        let runtime = tokio::runtime::Builder::new_multi_thread()
            .worker_threads(WORKER_THREADS)
            .thread_name(WORKER_NAME)
            .thread_stack_size(STACK_BYTES)
            // Tokio 1.53.1 requires this value to be non-zero. One is the
            // smallest dynamic blocking-pool bound, not a disable switch.
            .max_blocking_threads(1)
            .enable_all()
            .build()
            .map_err(|_| {
                RUNTIME_LEASED.store(false, Ordering::Release);
                ThreadControlError::RuntimeBuildFailed
            })?;
        Ok(Self {
            runtime: Some(runtime),
        })
    }

    pub(crate) fn runtime(&self) -> &tokio::runtime::Runtime {
        self.runtime
            .as_ref()
            .expect("approved runtime remains owned")
    }
}

impl Drop for ApprovedRuntime {
    fn drop(&mut self) {
        drop(self.runtime.take());
        RUNTIME_LEASED.store(false, Ordering::Release);
    }
}

#[derive(Clone, Debug, Eq, PartialEq)]
pub(crate) struct ThreadInventory {
    names: BTreeMap<String, usize>,
}

impl ThreadInventory {
    fn snapshot() -> Result<Self, ThreadControlError> {
        let mut names = BTreeMap::new();
        for entry in fs::read_dir("/proc/self/task").map_err(|_| ThreadControlError::Io)? {
            let entry = entry.map_err(|_| ThreadControlError::Io)?;
            let comm = fs::read_to_string(entry.path().join("comm"))
                .map_err(|_| ThreadControlError::Io)?;
            *names.entry(comm.trim().to_owned()).or_insert(0) += 1;
        }
        Ok(Self { names })
    }

    fn count(&self, name: &str) -> usize {
        self.names.get(name).copied().unwrap_or(0)
    }

    fn total(&self) -> usize {
        self.names.values().sum()
    }

    fn canonical(&self) -> String {
        self.names
            .iter()
            .map(|(name, count)| format!("{name}:{count}"))
            .collect::<Vec<_>>()
            .join(",")
    }
}

fn wait_for_inventory(
    predicate: impl Fn(&ThreadInventory) -> bool,
) -> Result<ThreadInventory, ThreadControlError> {
    let start = Instant::now();
    loop {
        let inventory = ThreadInventory::snapshot()?;
        if predicate(&inventory) {
            return Ok(inventory);
        }
        if start.elapsed() >= INVENTORY_DEADLINE {
            return Err(ThreadControlError::InventoryTimeout);
        }
        std::thread::yield_now();
    }
}

fn set_control_thread_name() -> Result<(), ThreadControlError> {
    fs::write("/proc/thread-self/comm", format!("{CONTROL_NAME}\n"))
        .map_err(|_| ThreadControlError::Io)
}

fn allowed_cpuset() -> Result<String, ThreadControlError> {
    let status = fs::read_to_string("/proc/self/status").map_err(|_| ThreadControlError::Io)?;
    status
        .lines()
        .find_map(|line| line.strip_prefix("Cpus_allowed_list:\t"))
        .map(str::to_owned)
        .ok_or(ThreadControlError::Io)
}

fn assert_inventory(
    inventory: &ThreadInventory,
    control: usize,
    workers: usize,
) -> Result<(), ThreadControlError> {
    if inventory.total() == control + workers
        && inventory.count(CONTROL_NAME) == control
        && inventory.count(WORKER_NAME) == workers
    {
        Ok(())
    } else {
        Err(ThreadControlError::InventoryMismatch)
    }
}

/// Runs the standalone Linux inventory and reachability evidence.
pub(crate) fn run_probe() -> Result<(), ThreadControlError> {
    set_control_thread_name()?;
    let before = ThreadInventory::snapshot()?;
    assert_inventory(&before, 1, 0)?;
    println!("profile_id={PROFILE_ID}");
    println!("profile_sha256={PROFILE_SHA256}");
    println!("profile_workers={WORKER_THREADS}");
    println!("profile_stack_bytes={STACK_BYTES}");
    println!("host_cpuset={}", allowed_cpuset()?);
    println!("before={}", before.canonical());

    let approved = ApprovedRuntime::build(WORKER_THREADS, APPROVED_CPUSET)?;
    let workers = wait_for_inventory(|inventory| {
        inventory.total() == 1 + WORKER_THREADS
            && inventory.count(CONTROL_NAME) == 1
            && inventory.count(WORKER_NAME) == WORKER_THREADS
    })?;
    assert_inventory(&workers, 1, WORKER_THREADS)?;
    println!("workers={}", workers.canonical());

    assert!(matches!(
        ApprovedRuntime::build(WORKER_THREADS, APPROVED_CPUSET),
        Err(ThreadControlError::RuntimeAlreadyExists)
    ));
    println!("second_approved_runtime=denied");

    let entered = Arc::new(Barrier::new(2));
    let release = Arc::new(Barrier::new(2));
    let task_entered = Arc::clone(&entered);
    let task_release = Arc::clone(&release);
    let blocking = approved.runtime().spawn_blocking(move || {
        task_entered.wait();
        task_release.wait();
    });
    entered.wait();
    let with_blocking = wait_for_inventory(|inventory| {
        inventory.total() == 2 + WORKER_THREADS
            && inventory.count(CONTROL_NAME) == 1
            && inventory.count(WORKER_NAME) == WORKER_THREADS + 1
    })?;
    println!("blocking_reachable={}", with_blocking.canonical());
    release.wait();
    approved.runtime().block_on(blocking).unwrap();

    drop(approved);
    let after = wait_for_inventory(|inventory| {
        inventory.total() == 1 && inventory.count(CONTROL_NAME) == 1
    })?;
    assert_inventory(&after, 1, 0)?;
    println!("after={}", after.canonical());
    Ok(())
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn profile_mismatch_fails_without_default_or_degrade() {
        assert!(matches!(
            ApprovedRuntime::build(1, &[0]),
            Err(ThreadControlError::HostProfileMismatch)
        ));
        assert!(matches!(
            ApprovedRuntime::build(2, &[0, 2]),
            Err(ThreadControlError::HostProfileMismatch)
        ));
        assert!(!RUNTIME_LEASED.load(Ordering::Acquire));
    }

    #[test]
    fn profile_identity_matches_gate_policy() {
        assert_eq!(PROFILE_ID, "saddle-0.2-linux-x86_64-2w");
        assert_eq!(
            PROFILE_SHA256,
            "7f6b8c2b3a600114959d97edd7b79ae409847e01cec2f760b9a7ce6d26123de4"
        );
        assert_eq!(WORKER_THREADS, 2);
        assert_eq!(STACK_BYTES, 2_097_152);
        assert_eq!(APPROVED_CPUSET, [0, 1]);
    }
}