arcbox-daemon 0.6.3

//! Service startup: DNS, Docker API, gRPC servers.
//!
//! gRPC is started once with all services. Machine/Sandbox/Snapshot services
//! use `SharedRuntime` (backed by `OnceLock`) and return `UNAVAILABLE` until
//! `init_runtime()` fills it. This avoids restarting the gRPC server mid-boot.

use std::net::{IpAddr, Ipv4Addr};
use std::sync::Arc;

use anyhow::{Context, Result};
use arcbox_api::{SharedRuntime, SystemServiceImpl};
use arcbox_core::{Runtime, VmLifecycleState};
use arcbox_docker::{DockerApiServer, DockerContextManager, ServerConfig};
use tokio::sync::watch;
use tracing::{info, warn};

use crate::context::{DaemonContext, ServiceHandles};
use crate::dns_service::DnsService;

#[cfg(target_os = "macos")]
const ROUTE_POLL_INTERVAL: std::time::Duration = std::time::Duration::from_secs(30);
#[cfg(target_os = "macos")]
const ROUTE_EVENT_DEBOUNCE: std::time::Duration = std::time::Duration::from_millis(250);

/// Starts the gRPC server with all services.
///
/// Called before `init_runtime()` — Machine/Sandbox/Snapshot services will
/// return `UNAVAILABLE` until the runtime is set. SystemService works
/// immediately so clients can observe setup progress.
pub async fn start_grpc(
    ctx: &DaemonContext,
    shared_runtime: SharedRuntime,
) -> Result<tokio::task::JoinHandle<()>> {
    let socket_path = &ctx.layout.grpc_socket;
    let listener = crate::control_plane::bind(socket_path)?;

    info!(socket = %socket_path.display(), "control plane listening (Connect + gRPC + gRPC-Web)");

    let system_service = SystemServiceImpl::new(
        Arc::clone(&ctx.setup_state),
        Arc::clone(&shared_runtime),
        Arc::clone(&ctx.early_runtime),
    );
    // Every service is served over Connect (CORE-53, CORE-68). The sandbox
    // control/data-plane split (CORE-57) is preserved inside it, so a cloud
    // deployment can still host those halves in different processes.
    // Reflection rides along and answers over all three wire formats.
    let app = crate::control_plane::into_app(crate::control_plane::connect_router(
        Arc::clone(&shared_runtime),
        system_service,
    )?);

    let shutdown = ctx.shutdown.clone();
    let handle = tokio::spawn(crate::control_plane::serve(listener, app, shutdown));

    Ok(handle)
}

/// Starts DNS, Docker API, and Docker CLI integration.
///
/// Called after `init_runtime()` — the runtime must be available.
pub async fn start_services(
    ctx: &DaemonContext,
    runtime: &Arc<Runtime>,
    grpc: tokio::task::JoinHandle<()>,
) -> Result<ServiceHandles> {
    // DNS service.
    let dns_service = DnsService::bind(Arc::clone(runtime.network_manager()), ctx.dns_port)
        .await
        .context("Failed to start DNS service")?;

    register_host_dns(runtime).await;

    let dns_shutdown = ctx.shutdown.clone();
    let dns = tokio::spawn(async move {
        if let Err(e) = dns_service.run(dns_shutdown).await {
            tracing::error!("DNS service error: {}", e);
        }
    });

    // VM-host-only mode: the Docker API, Docker CLI integration, and the
    // Kubernetes proxy all depend on the Linux VM, so they are skipped when it
    // is not booted.
    let linux_vm = runtime.config().vm.autostart;

    // Docker API server. Bound here rather than inside the spawned task so a
    // bind failure fails startup: the Docker socket is the daemon's primary
    // API, and a task that only logs the error would leave the pipeline
    // publishing READY for a daemon no client can reach (CORE-71). DNS above
    // already works this way.
    let docker = if linux_vm {
        let docker_server = DockerApiServer::new(
            ServerConfig {
                socket_path: ctx.layout.docker_socket.clone(),
            },
            Arc::clone(runtime),
        );
        let listener = docker_server.bind().with_context(|| {
            format!(
                "Failed to bind the Docker API socket: {}",
                ctx.layout.docker_socket.display()
            )
        })?;
        let docker_shutdown = ctx.shutdown.clone();
        Some(tokio::spawn(async move {
            if let Err(e) = docker_server.serve(listener, docker_shutdown).await {
                tracing::error!("Docker API server error: {}", e);
            }
        }))
    } else {
        None
    };

    // Docker CLI integration (optional).
    if linux_vm && ctx.docker_integration {
        match DockerContextManager::new_with_context_name(
            ctx.layout.docker_socket.clone(),
            ctx.profile.docker_context_name(),
        ) {
            Ok(ctx_manager) => {
                if let Err(e) = ctx_manager.enable() {
                    warn!("Failed to enable Docker integration: {}", e);
                } else {
                    info!("Docker CLI integration enabled");
                }
            }
            Err(e) => {
                warn!("Failed to create Docker context manager: {}", e);
            }
        }
    }

    // Kubernetes API proxy (TCP 127.0.0.1:16443 → guest vsock).
    let kubernetes_proxy = if linux_vm {
        crate::kubernetes_proxy::start(Arc::clone(runtime)).await
    } else {
        None
    };

    // Mirror route-install events into SetupStatus. VM (re)starts install
    // the container route from vm_lifecycle, outside the cold-start
    // recovery path that sets the flag directly — without this bridge,
    // route_installed would stay stale until the next daemon restart.
    let route_events = runtime.event_bus().subscribe();
    let route_state = Arc::clone(&ctx.setup_state);
    let route_shutdown = ctx.shutdown.clone();
    drop(tokio::spawn(async move {
        route_status_loop(route_events, route_state, route_shutdown).await;
    }));

    #[cfg(target_os = "macos")]
    let route_guard = linux_vm.then(|| {
        let runtime = Arc::clone(runtime);
        let setup_state = Arc::clone(&ctx.setup_state);
        let shutdown = ctx.shutdown.clone();
        tokio::spawn(async move {
            container_route_guard(runtime, setup_state, shutdown).await;
        })
    });
    #[cfg(not(target_os = "macos"))]
    let route_guard = None;

    Ok(ServiceHandles {
        dns,
        docker,
        grpc,
        kubernetes_proxy,
        route_guard,
    })
}

// =============================================================================
// Helpers
// =============================================================================

/// Starts the `vm_running` mirror for the System VM.
///
/// Spawned from `init_runtime`, not from [`start_services`]: the VM reaches
/// `VmLifecycleState::Running` partway through `Runtime::init` — that is what
/// publishes `VM_READY` — and `init` then waits for the guest container
/// runtime. A mirror started once `boot_runtime` has returned would report
/// `vm_running = false` across that whole window while the agent is already
/// answering, which is exactly what the field promises it is not.
pub fn spawn_vm_running_mirror(ctx: &DaemonContext, runtime: &Arc<Runtime>) {
    let state = runtime.subscribe_system_vm_state();
    let setup_state = Arc::clone(&ctx.setup_state);
    let shutdown = ctx.shutdown.clone();
    drop(tokio::spawn(async move {
        vm_running_loop(state, setup_state, shutdown).await;
    }));
}

/// Mirrors the System VM's lifecycle state into `SetupState.vm_running`.
///
/// The flag reports readiness level 2 — `VmLifecycleState::is_ready`, the
/// agent has answered a ping — which is the level that matches what a client
/// reads the field to mean. Level 1 (`MachineState::Running`) counts a VM
/// whose agent is not up, so RPCs against it fail; level 3 (guest dockerd)
/// is narrower than "the VM is running" and already has its own signal.
///
/// Driven from the lifecycle watch rather than set once on a successful guest
/// query, so it tracks both edges: idle stop, backend switch, and crash
/// recovery all move it without anything else having to remember to.
async fn vm_running_loop(
    mut state: watch::Receiver<VmLifecycleState>,
    setup_state: Arc<arcbox_api::SetupState>,
    shutdown: tokio_util::sync::CancellationToken,
) {
    loop {
        setup_state.set_vm_running(state.borrow_and_update().is_ready());
        tokio::select! {
            () = shutdown.cancelled() => break,
            // The sender lives as long as the runtime; an error means it is
            // gone, so there is nothing left to mirror.
            changed = state.changed() => if changed.is_err() { break },
        }
    }
}

/// Mirrors VM lifecycle events into `SetupState.route_installed`.
///
/// `ContainerRouteInstalled` sets the flag; `MachineStopped` clears it
/// (the bridge interface — and with it the host route — dies with the VM).
async fn route_status_loop(
    mut events: tokio::sync::broadcast::Receiver<arcbox_core::event::Event>,
    setup_state: Arc<arcbox_api::SetupState>,
    shutdown: tokio_util::sync::CancellationToken,
) {
    use arcbox_core::event::Event;
    use tokio::sync::broadcast::error::RecvError;

    loop {
        tokio::select! {
            () = shutdown.cancelled() => break,
            event = events.recv() => match event {
                Ok(Event::ContainerRouteInstalled { .. }) => {
                    setup_state.set_route_installed(true);
                }
                Ok(Event::MachineStopped { .. }) => {
                    setup_state.set_route_installed(false);
                }
                Ok(_) => {}
                // Missed events under load; state converges on the next
                // route-install or stop event.
                Err(RecvError::Lagged(_)) => {}
                Err(RecvError::Closed) => break,
            },
        }
    }
}

#[cfg(target_os = "macos")]
async fn container_route_guard(
    runtime: Arc<Runtime>,
    setup_state: Arc<arcbox_api::SetupState>,
    shutdown: tokio_util::sync::CancellationToken,
) {
    use arcbox_core::route_reconciler::RouteMode;

    let mut ticker = tokio::time::interval(ROUTE_POLL_INTERVAL);
    ticker.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
    let mut watcher = open_route_watcher();
    let mut active: Option<(String, RouteMode)> = None;
    let mut consecutive_failures = 0u32;

    loop {
        let route_event = tokio::select! {
            () = shutdown.cancelled() => break,
            _ = ticker.tick() => false,
            event = next_managed_route_event(watcher.as_ref()) => {
                match event {
                    Ok(()) => true,
                    Err(error) if error.raw_os_error() == Some(libc::ENOBUFS) => {
                        tracing::debug!(
                            "route event queue overflowed; reconciling current state"
                        );
                        true
                    }
                    Err(error) => {
                        tracing::warn!(%error, "route event watcher failed; polling remains active");
                        watcher = None;
                        false
                    }
                }
            }
        };

        if route_event {
            tokio::select! {
                () = shutdown.cancelled() => break,
                () = tokio::time::sleep(ROUTE_EVENT_DEBOUNCE) => {}
            }
            if let Some(watcher) = watcher.as_ref() {
                drain_route_events(watcher);
            }
        } else if watcher.is_none() {
            watcher = open_route_watcher();
        }

        let runtime_for_bridge = Arc::clone(&runtime);
        let bridge = match tokio::task::spawn_blocking(move || {
            resolve_container_bridge(&runtime_for_bridge)
        })
        .await
        {
            Ok(bridge) => bridge,
            Err(error) => {
                tracing::warn!(%error, "container bridge resolution task failed");
                None
            }
        };
        let Some(bridge) = bridge else {
            active = None;
            setup_state.set_route_installed(false);
            consecutive_failures = 0;
            continue;
        };

        let current_mode = active
            .as_ref()
            .filter(|(active_bridge, _)| active_bridge == &bridge)
            .map(|(_, mode)| *mode);
        let result = match current_mode {
            Some(mode) => {
                arcbox_core::route_reconciler::reconcile_route_for_bridge(&bridge, mode).await
            }
            None => arcbox_core::route_reconciler::initialize_route_for_bridge(&bridge).await,
        };

        match result {
            Ok(mode) => {
                let was_installed = setup_state.current().route_installed;
                active = Some((bridge, mode));
                setup_state.set_route_installed(true);
                consecutive_failures = 0;
                if !was_installed {
                    runtime.event_bus().publish(
                        arcbox_core::event::Event::ContainerRouteInstalled {
                            name: arcbox_core::DEFAULT_MACHINE_NAME.to_string(),
                        },
                    );
                }
            }
            Err(error) => {
                setup_state.set_route_installed(false);
                consecutive_failures = consecutive_failures.saturating_add(1);
                if should_log_route_failure(consecutive_failures) {
                    tracing::warn!(
                        %error,
                        %bridge,
                        consecutive_failures,
                        "container route reconciliation failed"
                    );
                }
            }
        }
    }
}

#[cfg(target_os = "macos")]
fn open_route_watcher() -> Option<tokio::io::unix::AsyncFd<arcbox_route::RouteWatcher>> {
    match arcbox_route::RouteWatcher::open().and_then(tokio::io::unix::AsyncFd::new) {
        Ok(watcher) => Some(watcher),
        Err(error) => {
            tracing::warn!(%error, "route event watcher unavailable; using polling");
            None
        }
    }
}

#[cfg(target_os = "macos")]
async fn next_managed_route_event(
    watcher: Option<&tokio::io::unix::AsyncFd<arcbox_route::RouteWatcher>>,
) -> std::io::Result<()> {
    let Some(watcher) = watcher else {
        return std::future::pending().await;
    };

    loop {
        let mut ready = watcher.readable().await?;
        match ready.try_io(|inner| inner.get_ref().read_event()) {
            Ok(Ok(Some(event))) if event.network.is_some_and(is_managed_route) => return Ok(()),
            Ok(Ok(_)) => {}
            Ok(Err(error)) if error.kind() == std::io::ErrorKind::InvalidData => {
                tracing::debug!(%error, "ignored malformed route event");
            }
            Ok(Err(error)) => return Err(error),
            Err(_would_block) => {}
        }
    }
}

#[cfg(target_os = "macos")]
fn drain_route_events(watcher: &tokio::io::unix::AsyncFd<arcbox_route::RouteWatcher>) {
    for _ in 0..256 {
        match watcher.get_ref().read_event() {
            Ok(_) => {}
            Err(error) if error.kind() == std::io::ErrorKind::WouldBlock => break,
            Err(_) => break,
        }
    }
}

#[cfg(target_os = "macos")]
fn is_managed_route(network: arcbox_route::Ipv4Net) -> bool {
    let network = network.to_string();
    network == arcbox_core::route_reconciler::CONTAINER_SUBNET
        || arcbox_core::route_reconciler::CONTAINER_SPLIT_SUBNETS
            .iter()
            .any(|candidate| network == *candidate)
}

#[cfg(all(target_os = "macos", feature = "vmnet"))]
fn resolve_container_bridge(runtime: &Runtime) -> Option<String> {
    let machine = runtime
        .machine_manager()
        .get(arcbox_core::DEFAULT_MACHINE_NAME)?;
    if !matches!(
        machine.state,
        arcbox_core::machine::MachineState::Starting | arcbox_core::machine::MachineState::Running
    ) {
        return None;
    }
    runtime
        .machine_manager()
        .vmnet_bridge_name(arcbox_core::DEFAULT_MACHINE_NAME)
}

#[cfg(all(target_os = "macos", not(feature = "vmnet")))]
fn resolve_container_bridge(runtime: &Runtime) -> Option<String> {
    let machine = runtime
        .machine_manager()
        .get(arcbox_core::DEFAULT_MACHINE_NAME)?;
    if !matches!(
        machine.state,
        arcbox_core::machine::MachineState::Starting | arcbox_core::machine::MachineState::Running
    ) {
        return None;
    }
    let mac = runtime
        .machine_manager()
        .bridge_mac(arcbox_core::DEFAULT_MACHINE_NAME)?;
    arcbox_core::bridge_discovery::resolve_bridge_by_mac(&mac).map(|bridge| bridge.name)
}

#[cfg(target_os = "macos")]
fn should_log_route_failure(consecutive_failures: u32) -> bool {
    consecutive_failures == 1 || consecutive_failures.is_multiple_of(30)
}

async fn register_host_dns(runtime: &Arc<Runtime>) {
    let network_cfg = &runtime.config().network;
    let gateway_ip = network_cfg
        .gateway
        .as_ref()
        .and_then(|s| s.parse::<Ipv4Addr>().ok())
        .or_else(|| first_address_in_subnet(&network_cfg.subnet))
        .unwrap_or(Ipv4Addr::new(10, 0, 2, 1));
    let ip = IpAddr::V4(gateway_ip);
    runtime
        .register_host_dns(
            &[
                "host".into(),
                "host.docker.internal".into(),
                "gateway.docker.internal".into(),
            ],
            ip,
        )
        .await;
}

fn first_address_in_subnet(subnet: &str) -> Option<Ipv4Addr> {
    let (ip_str, prefix_str) = subnet.split_once('/')?;
    let base: Ipv4Addr = ip_str.parse().ok()?;
    let prefix: u8 = prefix_str.parse().ok()?;
    if prefix == 0 || prefix >= 32 {
        return None;
    }
    let mask: u32 = (!0u32) << (32 - prefix);
    let network = u32::from(base) & mask;
    let first = network.checked_add(1)?;
    let broadcast = network | !mask;
    if first > broadcast {
        None
    } else {
        Some(Ipv4Addr::from(first))
    }
}

#[cfg(test)]
mod tests {
    use std::time::Duration;

    use arcbox_api::SetupState;
    use arcbox_core::event::{Event, EventBus};

    use super::*;

    /// Polls until `route_installed` matches `want` or times out.
    async fn wait_for_route_installed(state: &SetupState, want: bool) -> bool {
        for _ in 0..200 {
            if state.current().route_installed == want {
                return true;
            }
            tokio::time::sleep(Duration::from_millis(5)).await;
        }
        false
    }

    /// Polls until `vm_running` matches `want` or times out.
    async fn wait_for_vm_running(state: &SetupState, want: bool) -> bool {
        for _ in 0..200 {
            if state.current().vm_running == want {
                return true;
            }
            tokio::time::sleep(Duration::from_millis(5)).await;
        }
        false
    }

    /// The flag has to fall as well as rise. Its predecessor was set once by
    /// cold-start recovery and never cleared, so a client saw `vm_running`
    /// stay true for the rest of the daemon's life after any VM stop.
    #[tokio::test]
    async fn vm_running_loop_follows_the_lifecycle_both_ways() {
        let (lifecycle, rx) = watch::channel(VmLifecycleState::Stopped);
        let setup_state = Arc::new(SetupState::new());
        let shutdown = tokio_util::sync::CancellationToken::new();

        let task = tokio::spawn(vm_running_loop(
            rx,
            Arc::clone(&setup_state),
            shutdown.clone(),
        ));

        assert!(wait_for_vm_running(&setup_state, false).await);

        lifecycle.send(VmLifecycleState::Running).expect("receiver");
        assert!(wait_for_vm_running(&setup_state, true).await);

        // Idle still counts as running: the VM is up, just quiet.
        lifecycle.send(VmLifecycleState::Idle).expect("receiver");
        tokio::time::sleep(Duration::from_millis(20)).await;
        assert!(setup_state.current().vm_running);

        // The regression this test exists for.
        lifecycle.send(VmLifecycleState::Stopped).expect("receiver");
        assert!(wait_for_vm_running(&setup_state, false).await);

        shutdown.cancel();
        tokio::time::timeout(Duration::from_secs(1), task)
            .await
            .expect("loop exits on shutdown")
            .expect("loop task panicked");
    }

    #[tokio::test]
    async fn route_status_loop_mirrors_events_into_setup_state() {
        let bus = EventBus::new();
        let setup_state = Arc::new(SetupState::new());
        let shutdown = tokio_util::sync::CancellationToken::new();

        let task = tokio::spawn(route_status_loop(
            bus.subscribe(),
            Arc::clone(&setup_state),
            shutdown.clone(),
        ));

        assert!(!setup_state.current().route_installed);

        bus.publish(Event::ContainerRouteInstalled {
            name: "default".into(),
        });
        assert!(wait_for_route_installed(&setup_state, true).await);

        // Unrelated events leave the flag untouched.
        bus.publish(Event::MachineIdle {
            name: "default".into(),
        });
        tokio::time::sleep(Duration::from_millis(20)).await;
        assert!(setup_state.current().route_installed);

        // The bridge dies with the VM, so MachineStopped clears the flag.
        bus.publish(Event::MachineStopped {
            name: "default".into(),
        });
        assert!(wait_for_route_installed(&setup_state, false).await);

        shutdown.cancel();
        tokio::time::timeout(Duration::from_secs(1), task)
            .await
            .expect("loop exits on shutdown")
            .expect("loop task panicked");
    }
}