diff --git a/src/boxlite/src/litebox/box_impl.rs b/src/boxlite/src/litebox/box_impl.rs index d16ed7e11..eea120bc4 100644 --- a/src/boxlite/src/litebox/box_impl.rs +++ b/src/boxlite/src/litebox/box_impl.rs @@ -631,7 +631,7 @@ impl BoxImpl { use super::BoxBuilder; use std::sync::Arc; - let state = self.state.read().clone(); + let mut state = self.state.read().clone(); let is_first_start = state.status == BoxStatus::Configured; // Retrieve the lock (allocated in create()) @@ -653,6 +653,15 @@ impl BoxImpl { // LockGuard acquires lock on creation and releases on drop. let _guard = LockGuard::new(&*locker); + if state.status != BoxStatus::Running { + state.advance_lifecycle_generation()?; + let mut current_state = self.state.write(); + current_state.lifecycle_generation = state.lifecycle_generation; + self.runtime + .box_manager + .save_box(&self.config.id, ¤t_state)?; + } + // Build the box (lock is held) // The returned cleanup_guard stays armed until we disarm it after all // operations succeed. If any operation fails, the guard's Drop will diff --git a/src/boxlite/src/litebox/init/mod.rs b/src/boxlite/src/litebox/init/mod.rs index 61cceb8a9..41de9c37b 100644 --- a/src/boxlite/src/litebox/init/mod.rs +++ b/src/boxlite/src/litebox/init/mod.rs @@ -200,8 +200,15 @@ impl BoxBuilder { let status = state.status; let reuse_rootfs = status == BoxStatus::Stopped; let skip_guest_wait = status == BoxStatus::Running; + let lifecycle_generation = state.lifecycle_generation; - let ctx = InitPipelineContext::new(config, runtime.clone(), reuse_rootfs, skip_guest_wait); + let ctx = InitPipelineContext::new( + config, + runtime.clone(), + reuse_rootfs, + skip_guest_wait, + lifecycle_generation, + ); let ctx = Arc::new(Mutex::new(ctx)); let ctx_for_cleanup = Arc::clone(&ctx); diff --git a/src/boxlite/src/litebox/init/tasks/guest_init.rs b/src/boxlite/src/litebox/init/tasks/guest_init.rs index c0a505928..b1e1412ea 100644 --- a/src/boxlite/src/litebox/init/tasks/guest_init.rs +++ b/src/boxlite/src/litebox/init/tasks/guest_init.rs @@ -32,6 +32,7 @@ impl PipelineTask for GuestInitTask { container_mounts, network_spec, ca_cert_pem, + lifecycle_generation, ) = { let mut ctx = ctx.lock().await; @@ -63,6 +64,7 @@ impl PipelineTask for GuestInitTask { container_mounts, network_spec, ca_cert_pem, + ctx.lifecycle_generation, ) }; @@ -75,6 +77,7 @@ impl PipelineTask for GuestInitTask { &container_mounts, &network_spec, ca_cert_pem.as_deref(), + lifecycle_generation, ) .await .inspect_err(|e| log_task_error(&box_id, task_name, e))?; @@ -104,6 +107,7 @@ async fn run_guest_init( container_mounts: &[ContainerMount], network_spec: &NetworkSpec, ca_cert_pem: Option<&str>, + lifecycle_generation: u64, ) -> BoxliteResult<()> { let container_id_str = container_id.as_str(); @@ -141,6 +145,7 @@ async fn run_guest_init( rootfs_init.clone(), container_mounts.to_vec(), ca_certs, + lifecycle_generation, ) .await?; tracing::info!(container_id = %returned_id, "Container initialized"); diff --git a/src/boxlite/src/litebox/init/types.rs b/src/boxlite/src/litebox/init/types.rs index 2e82943a7..407874f69 100644 --- a/src/boxlite/src/litebox/init/types.rs +++ b/src/boxlite/src/litebox/init/types.rs @@ -287,6 +287,7 @@ pub struct InitPipelineContext { pub config: BoxConfig, pub runtime: SharedRuntimeImpl, pub guard: CleanupGuard, + pub lifecycle_generation: u64, pub reuse_rootfs: bool, /// Skip waiting for guest ready signal (for reattach to running box). pub skip_guest_wait: bool, @@ -312,12 +313,14 @@ impl InitPipelineContext { runtime: SharedRuntimeImpl, reuse_rootfs: bool, skip_guest_wait: bool, + lifecycle_generation: u64, ) -> Self { let guard = CleanupGuard::new(runtime.clone(), config.id.clone()); Self { config, runtime, guard, + lifecycle_generation, reuse_rootfs, skip_guest_wait, layout: None, diff --git a/src/boxlite/src/litebox/state.rs b/src/boxlite/src/litebox/state.rs index 2f7b87faf..add189633 100644 --- a/src/boxlite/src/litebox/state.rs +++ b/src/boxlite/src/litebox/state.rs @@ -217,6 +217,9 @@ pub struct BoxState { /// Serde default keeps existing DB rows readable without migration. #[serde(default)] pub error_reason: Option, + /// Monotonic identity for each VM run. + #[serde(default)] + pub lifecycle_generation: u64, } /// Health status of a box. @@ -315,9 +318,18 @@ impl BoxState { lock_id: None, health_status: HealthStatus::new(), error_reason: None, + lifecycle_generation: 0, } } + pub fn advance_lifecycle_generation(&mut self) -> BoxliteResult { + self.lifecycle_generation = self + .lifecycle_generation + .checked_add(1) + .ok_or_else(|| BoxliteError::Internal("lifecycle generation overflow".to_string()))?; + Ok(self.lifecycle_generation) + } + /// Set lock ID and update timestamp. pub fn set_lock_id(&mut self, lock_id: LockId) { self.lock_id = Some(lock_id); @@ -990,5 +1002,22 @@ mod tests { assert_eq!(state.health_status.state, HealthState::None); assert_eq!(state.health_status.failures, 0); assert!(state.health_status.last_check.is_none()); + assert_eq!(state.lifecycle_generation, 0); + } + + #[test] + fn lifecycle_generation_advances_monotonically() { + let mut state = BoxState::new(); + + assert_eq!(state.advance_lifecycle_generation().unwrap(), 1); + assert_eq!(state.advance_lifecycle_generation().unwrap(), 2); + } + + #[test] + fn lifecycle_generation_rejects_overflow() { + let mut state = BoxState::new(); + state.lifecycle_generation = u64::MAX; + + assert!(state.advance_lifecycle_generation().is_err()); } } diff --git a/src/boxlite/src/portal/interfaces/container.rs b/src/boxlite/src/portal/interfaces/container.rs index b4618c294..2c9c93722 100644 --- a/src/boxlite/src/portal/interfaces/container.rs +++ b/src/boxlite/src/portal/interfaces/container.rs @@ -99,6 +99,7 @@ impl ContainerInterface { rootfs: ContainerRootfsInitConfig, mounts: Vec, ca_certs: Vec, + lifecycle_generation: u64, ) -> BoxliteResult { let proto_config = ProtoContainerConfig { entrypoint: image_config.final_cmd(), @@ -140,6 +141,7 @@ impl ContainerInterface { rootfs: Some(rootfs.into_proto()), mounts: proto_mounts, ca_certs: ca_certs.into_iter().map(|pem| CaCert { pem }).collect(), + lifecycle_generation, }; let response = self.client.init(request).await?.into_inner(); diff --git a/src/boxlite/tests/init_stdio.rs b/src/boxlite/tests/init_stdio.rs new file mode 100644 index 000000000..3f7e3ff68 --- /dev/null +++ b/src/boxlite/tests/init_stdio.rs @@ -0,0 +1,58 @@ +mod common; + +use boxlite::runtime::options::{BoxOptions, BoxliteOptions}; +use boxlite::{BoxCommand, BoxliteRuntime}; +use std::time::Duration; + +#[tokio::test] +async fn init_output_larger_than_pipes_does_not_block_entrypoint() { + let home = boxlite_test_utils::home::PerTestBoxHome::new(); + let runtime = BoxliteRuntime::new(BoxliteOptions { + home_dir: home.path.clone(), + image_registries: common::test_registries(), + }) + .expect("create runtime"); + let handle = runtime + .create( + BoxOptions { + entrypoint: Some(vec![ + "sh".to_string(), + "-c".to_string(), + "dd if=/dev/zero bs=1024 count=1024; dd if=/dev/zero bs=1024 count=1024 >&2; touch /tmp/init-output-drained; exec sleep 300" + .to_string(), + ]), + ..common::alpine_opts_auto() + }, + None, + ) + .await + .expect("create box"); + let box_id = handle.id().to_string(); + + handle.start().await.expect("start box"); + + let ready = tokio::time::timeout(Duration::from_secs(15), async { + loop { + let execution = handle + .exec( + BoxCommand::new("test") + .arg("-f") + .arg("/tmp/init-output-drained"), + ) + .await + .expect("start readiness check"); + let result = execution.wait().await.expect("wait for readiness check"); + if result.exit_code == 0 { + return; + } + tokio::time::sleep(Duration::from_millis(100)).await; + } + }) + .await; + + let _ = handle.stop().await; + let _ = runtime.remove(&box_id, false).await; + let _ = runtime.shutdown(Some(common::TEST_SHUTDOWN_TIMEOUT)).await; + + ready.expect("init stdout or stderr filled its pipe before the entrypoint could continue"); +} diff --git a/src/guest/src/container/lifecycle.rs b/src/guest/src/container/lifecycle.rs index 7c7129fb3..6a62bfe5c 100644 --- a/src/guest/src/container/lifecycle.rs +++ b/src/guest/src/container/lifecycle.rs @@ -5,7 +5,8 @@ use super::command::ContainerCommand; use super::spec::UserMount; -use super::stdio::ContainerStdio; +use super::stdio::{ContainerStdio, InitOutputCompletion, InitStdioFds}; +use super::zygote::{self, InitSpec}; use super::{kill, spec, start}; use crate::layout::GuestLayout; use crate::service::exec::InitHealthCheck; @@ -13,6 +14,7 @@ use boxlite_shared::errors::{BoxliteError, BoxliteResult}; use libcontainer::container::Container as LibContainer; use libcontainer::signal::Signal; use std::collections::HashMap; +use std::os::fd::AsRawFd; use std::path::{Path, PathBuf}; /// OCI container @@ -55,6 +57,12 @@ pub struct Container { is_shutdown: std::sync::atomic::AtomicBool, } +pub(crate) struct StartedContainer { + pub container: Container, + pub init_pid: nix::unistd::Pid, + pub output_completion: InitOutputCompletion, +} + impl Container { /// Create and start an OCI container /// @@ -80,8 +88,7 @@ impl Container { /// - Empty rootfs or entrypoint /// - Failed to create container directory /// - Failed to create or start container - /// - Init process exited immediately - pub fn start( + pub async fn start( container_id: &str, rootfs: impl AsRef, entrypoint: Vec, @@ -89,7 +96,7 @@ impl Container { workdir: impl AsRef, user: &str, user_mounts: Vec, - ) -> BoxliteResult { + ) -> BoxliteResult { let rootfs = rootfs.as_ref(); let workdir = workdir.as_ref(); @@ -167,20 +174,29 @@ impl Container { // Create stdio pipes before container creation. // These keep the init process alive by holding stdin open. - let (stdio, init_fds) = ContainerStdio::new()?; - - // Create and start container with custom stdio - start::create_container_with_stdio(container_id, &state_root, &bundle_path, init_fds)?; - start::start_container(container_id, &state_root)?; - - Ok(Self { - id: container_id.to_string(), - state_root, - bundle_path, - env: env_map, - user: (uid, gid), - stdio, - is_shutdown: std::sync::atomic::AtomicBool::new(false), + let (stdio, init_fds, output_completion) = ContainerStdio::new()?; + + create_init_via_zygote( + container_id, + state_root.clone(), + bundle_path.clone(), + init_fds, + ) + .await?; + let init_pid = start::start_container(container_id, &state_root)?; + + Ok(StartedContainer { + container: Self { + id: container_id.to_string(), + state_root, + bundle_path, + env: env_map, + user: (uid, gid), + stdio, + is_shutdown: std::sync::atomic::AtomicBool::new(false), + }, + init_pid, + output_completion, }) } @@ -269,15 +285,12 @@ impl Container { ) } - /// Drain init process stdout and stderr. - /// - /// Reads all available data from the init process pipes using non-blocking I/O. - /// Can only be called once — subsequent calls return empty strings. + /// Return the output tail captured from the init process. /// /// # Returns /// /// `(stdout, stderr)` — captured output from the init process. - pub fn drain_init_output(&mut self) -> (String, String) { + pub fn drain_init_output(&self) -> (String, String) { self.stdio.drain_output() } @@ -304,7 +317,6 @@ impl Container { pub fn diagnose_exit(&mut self) -> String { let container_state_path = self.container_state_path(); - // Drain init process output before building diagnostics let (init_stdout, init_stderr) = self.drain_init_output(); // Try to load container state from libcontainer @@ -429,6 +441,35 @@ impl Container { } } +async fn create_init_via_zygote( + container_id: &str, + state_root: PathBuf, + bundle_path: PathBuf, + init_fds: InitStdioFds, +) -> BoxliteResult<()> { + let spec = InitSpec { + container_id: container_id.to_string(), + state_root, + bundle_path, + }; + + tokio::task::spawn_blocking(move || { + let raw_fds = [ + init_fds.stdin.as_raw_fd(), + init_fds.stdout.as_raw_fd(), + init_fds.stderr.as_raw_fd(), + ]; + let result = zygote::ZYGOTE + .get() + .expect("zygote not started") + .build_init(spec, raw_fds); + drop(init_fds); + result + }) + .await + .map_err(|error| BoxliteError::Internal(format!("init build join error: {error}")))? +} + // ==================== // Init Health Check // ==================== diff --git a/src/guest/src/container/mod.rs b/src/guest/src/container/mod.rs index 9695fab96..457a3c3a5 100644 --- a/src/guest/src/container/mod.rs +++ b/src/guest/src/container/mod.rs @@ -74,6 +74,8 @@ mod start; #[cfg(target_os = "linux")] mod stdio; #[cfg(target_os = "linux")] +mod supervisor; +#[cfg(target_os = "linux")] pub(crate) mod zygote; #[cfg(target_os = "linux")] @@ -82,3 +84,5 @@ pub(crate) use command::SpawnResult; pub use lifecycle::Container; #[cfg(target_os = "linux")] pub use spec::UserMount; +#[cfg(target_os = "linux")] +pub(crate) use supervisor::{InitSupervisor, InitTerminal}; diff --git a/src/guest/src/container/start.rs b/src/guest/src/container/start.rs index df7d2d6da..74d94e452 100644 --- a/src/guest/src/container/start.rs +++ b/src/guest/src/container/start.rs @@ -5,9 +5,8 @@ use super::spec; use boxlite_shared::errors::{BoxliteError, BoxliteResult}; -use libcontainer::container::builder::ContainerBuilder; use libcontainer::container::Container as LibContainer; -use libcontainer::syscall::syscall::SyscallType; +use nix::unistd::Pid; use std::fs; use std::path::{Path, PathBuf}; @@ -158,84 +157,8 @@ pub(crate) fn create_oci_bundle( // Execution Functions (Execute Phase) // ==================== -/// Create container using libcontainer (does not start it) -/// -/// Uses default stdio (inherited from parent process). -/// For custom stdio, use `create_container_with_stdio`. -#[allow(dead_code)] -pub(crate) fn create_container( - container_id: &str, - state_root: &Path, - bundle_path: &Path, -) -> BoxliteResult<()> { - ContainerBuilder::new(container_id.to_string(), SyscallType::default()) - .with_root_path(state_root) - .map_err(|e| BoxliteError::Internal(format!("Failed to set container root path: {}", e)))? - .validate_id() - .map_err(|e| BoxliteError::Internal(format!("Invalid container ID: {}", e)))? - .as_init(bundle_path) - .with_systemd(false) - .with_detach(true) - .build() - .map_err(|e| { - BoxliteError::Internal(format!( - "Failed to create container {} at bundle {}: {}", - container_id, - bundle_path.display(), - e - )) - })?; - - tracing::info!(container_id, "Created OCI container"); - Ok(()) -} - -/// Create container with custom stdio file descriptors. -/// -/// This allows the init process to use pipes controlled by boxlite-guest, -/// keeping interactive entrypoints (like /bin/sh) alive by holding stdin open. -/// -/// # Arguments -/// -/// * `container_id` - Unique container identifier -/// * `state_root` - Directory for libcontainer state -/// * `bundle_path` - OCI bundle directory with config.json -/// * `stdio_fds` - Custom stdio file descriptors for init process -pub(crate) fn create_container_with_stdio( - container_id: &str, - state_root: &Path, - bundle_path: &Path, - stdio_fds: super::stdio::InitStdioFds, -) -> BoxliteResult<()> { - // Note: with_stdin/stdout/stderr must be called before as_init() - // because they're methods on ContainerBuilder, not InitContainerBuilder - ContainerBuilder::new(container_id.to_string(), SyscallType::default()) - .with_root_path(state_root) - .map_err(|e| BoxliteError::Internal(format!("Failed to set container root path: {}", e)))? - .validate_id() - .map_err(|e| BoxliteError::Internal(format!("Invalid container ID: {}", e)))? - .with_stdin(stdio_fds.stdin) - .with_stdout(stdio_fds.stdout) - .with_stderr(stdio_fds.stderr) - .as_init(bundle_path) - .with_systemd(false) - .with_detach(true) - .build() - .map_err(|e| { - BoxliteError::Internal(format!( - "Failed to create container {} at bundle {}: {}", - container_id, - bundle_path.display(), - e - )) - })?; - - tracing::info!(container_id, "Created OCI container with custom stdio"); - Ok(()) -} - /// Start the container (executes entrypoint) -pub(crate) fn start_container(container_id: &str, state_root: &Path) -> BoxliteResult<()> { +pub(crate) fn start_container(container_id: &str, state_root: &Path) -> BoxliteResult { let container_state_path = state_root.join(container_id); let mut container = LibContainer::load(container_state_path.clone()).map_err(|e| { @@ -251,8 +174,15 @@ pub(crate) fn start_container(container_id: &str, state_root: &Path) -> BoxliteR BoxliteError::Internal(format!("Failed to start container {}: {}", container_id, e)) })?; + let pid = container.pid().ok_or_else(|| { + BoxliteError::Internal(format!( + "Container {} started without an init PID", + container_id + )) + })?; + tracing::info!(container_id, "Started OCI container"); - Ok(()) + Ok(pid) } // ==================== diff --git a/src/guest/src/container/stdio.rs b/src/guest/src/container/stdio.rs index 1b88c41db..35d002045 100644 --- a/src/guest/src/container/stdio.rs +++ b/src/guest/src/container/stdio.rs @@ -35,13 +35,18 @@ use boxlite_shared::errors::{BoxliteError, BoxliteResult}; use nix::unistd::pipe; use std::io::Read; -use std::os::unix::io::{AsRawFd, OwnedFd}; +use std::os::unix::io::OwnedFd; +use std::sync::{Arc, Mutex}; +use tokio::sync::oneshot; + +const MAX_CAPTURE: usize = 4096; +const DRAIN_BUFFER_SIZE: usize = 8192; /// Stdio configuration for container init process. /// /// Holds pipe file descriptors: /// - stdin_tx: write-end held open (blocks init's read forever) -/// - stdout_rx/stderr_rx: read-ends for optional log capture +/// - stdout/stderr tails: bounded diagnostic output retained by background drains /// /// # Lifecycle /// @@ -56,11 +61,9 @@ pub struct ContainerStdio { #[allow(dead_code)] stdin_tx: OwnedFd, - /// Read-end of stdout pipe (taken by drain_output for log capture) - stdout_rx: Option, + stdout_tail: Arc>>, - /// Read-end of stderr pipe (taken by drain_output for log capture) - stderr_rx: Option, + stderr_tail: Arc>>, } /// File descriptors to pass to container init process. @@ -83,17 +86,34 @@ pub struct InitStdioFds { pub stderr: OwnedFd, } +pub(crate) struct InitOutputCompletion { + stdout: oneshot::Receiver>, + stderr: oneshot::Receiver>, +} + +impl InitOutputCompletion { + pub(crate) async fn wait(self) -> Result<(), String> { + let (stdout, stderr) = tokio::join!(self.stdout, self.stderr); + stdout + .map_err(|_| "init stdout drain stopped before reporting completion".to_string())??; + stderr + .map_err(|_| "init stderr drain stopped before reporting completion".to_string())??; + Ok(()) + } +} + impl ContainerStdio { /// Create new stdio pipes for container init. /// - /// Returns `(ContainerStdio, InitStdioFds)` where: + /// Returns `(ContainerStdio, InitStdioFds, InitOutputCompletion)` where: /// - `ContainerStdio`: held by boxlite-guest to keep init alive /// - `InitStdioFds`: passed to libcontainer for init process + /// - `InitOutputCompletion`: resolves when stdout and stderr reach EOF /// /// # Errors /// /// Returns error if pipe creation fails. - pub fn new() -> BoxliteResult<(Self, InitStdioFds)> { + pub fn new() -> BoxliteResult<(Self, InitStdioFds, InitOutputCompletion)> { // Create stdin pipe: init reads from rx, we hold tx open let (stdin_rx, stdin_tx) = pipe() .map_err(|e| BoxliteError::Internal(format!("Failed to create stdin pipe: {}", e)))?; @@ -106,11 +126,18 @@ impl ContainerStdio { let (stderr_rx, stderr_tx) = pipe() .map_err(|e| BoxliteError::Internal(format!("Failed to create stderr pipe: {}", e)))?; - // nix::unistd::pipe() returns OwnedFd directly + let stdout_tail = Arc::new(Mutex::new(Vec::with_capacity(MAX_CAPTURE))); + let stderr_tail = Arc::new(Mutex::new(Vec::with_capacity(MAX_CAPTURE))); + + let stdout_complete = + spawn_output_drain("boxlite-init-stdout", stdout_rx, stdout_tail.clone())?; + let stderr_complete = + spawn_output_drain("boxlite-init-stderr", stderr_rx, stderr_tail.clone())?; + let container_stdio = Self { stdin_tx, - stdout_rx: Some(stdout_rx), - stderr_rx: Some(stderr_rx), + stdout_tail, + stderr_tail, }; let init_fds = InitStdioFds { @@ -121,62 +148,86 @@ impl ContainerStdio { tracing::debug!("Created container stdio pipes"); - Ok((container_stdio, init_fds)) + Ok(( + container_stdio, + init_fds, + InitOutputCompletion { + stdout: stdout_complete, + stderr: stderr_complete, + }, + )) } - /// Drain all available output from init process stdout and stderr. - /// - /// Takes ownership of the pipe read-ends and reads with non-blocking I/O. - /// Can only be called once — subsequent calls return empty strings. + /// Return the bounded output tail retained by the background drains. /// /// # Returns /// /// `(stdout, stderr)` — captured output, truncated to 4 KiB each. - pub fn drain_output(&mut self) -> (String, String) { - let stdout = drain_fd(self.stdout_rx.take()); - let stderr = drain_fd(self.stderr_rx.take()); - (stdout, stderr) + pub fn drain_output(&self) -> (String, String) { + ( + output_tail(&self.stdout_tail), + output_tail(&self.stderr_tail), + ) } } -/// Read all available data from an fd using non-blocking I/O. -fn drain_fd(fd: Option) -> String { - const MAX_CAPTURE: usize = 4096; - - let Some(fd) = fd else { - return String::new(); - }; - - // Set non-blocking so read returns immediately when no more data - let raw_fd = fd.as_raw_fd(); - let flags = nix::fcntl::fcntl(raw_fd, nix::fcntl::FcntlArg::F_GETFL); - if let Ok(flags) = flags { - let mut new_flags = nix::fcntl::OFlag::from_bits_truncate(flags); - new_flags.insert(nix::fcntl::OFlag::O_NONBLOCK); - let _ = nix::fcntl::fcntl(raw_fd, nix::fcntl::FcntlArg::F_SETFL(new_flags)); - } +fn spawn_output_drain( + name: &str, + fd: OwnedFd, + tail: Arc>>, +) -> BoxliteResult>> { + let (complete_tx, complete_rx) = oneshot::channel(); + std::thread::Builder::new() + .name(name.to_string()) + .spawn(move || { + let _ = complete_tx.send(drain_fd(fd, tail)); + }) + .map(|_| complete_rx) + .map_err(|error| { + BoxliteError::Internal(format!("Failed to start init output drain: {error}")) + }) +} +fn drain_fd(fd: OwnedFd, tail: Arc>>) -> Result<(), String> { let mut file = std::fs::File::from(fd); - let mut buf = vec![0u8; MAX_CAPTURE]; - let mut total = 0; + let mut buffer = [0; DRAIN_BUFFER_SIZE]; - // Read in a loop to drain the pipe buffer loop { - match file.read(&mut buf[total..]) { - Ok(0) => break, // EOF - Ok(n) => { - total += n; - if total >= MAX_CAPTURE { - break; - } + match file.read(&mut buffer) { + Ok(0) => return Ok(()), + Ok(bytes_read) => { + let mut captured = tail.lock().unwrap_or_else(|poisoned| poisoned.into_inner()); + append_tail(&mut captured, &buffer[..bytes_read]); + } + Err(error) if error.kind() == std::io::ErrorKind::Interrupted => continue, + Err(error) => { + tracing::warn!(%error, "Init output drain stopped"); + return Err(error.to_string()); } - Err(e) if e.kind() == std::io::ErrorKind::WouldBlock => break, - Err(_) => break, } } +} + +fn append_tail(tail: &mut Vec, bytes: &[u8]) { + if bytes.len() >= MAX_CAPTURE { + tail.clear(); + tail.extend_from_slice(&bytes[bytes.len() - MAX_CAPTURE..]); + return; + } + + let overflow = tail + .len() + .saturating_add(bytes.len()) + .saturating_sub(MAX_CAPTURE); + if overflow > 0 { + tail.drain(..overflow); + } + tail.extend_from_slice(bytes); +} - buf.truncate(total); - String::from_utf8_lossy(&buf).into_owned() +fn output_tail(tail: &Mutex>) -> String { + let captured = tail.lock().unwrap_or_else(|poisoned| poisoned.into_inner()); + String::from_utf8_lossy(&captured).into_owned() } #[cfg(test)] @@ -184,27 +235,43 @@ mod tests { use super::*; use std::io::Write; use std::os::unix::io::AsRawFd; + use std::sync::mpsc; + use std::thread; + use std::time::{Duration, Instant}; + + fn wait_for_output( + stdio: &ContainerStdio, + expected_stdout: &str, + expected_stderr: &str, + ) -> (String, String) { + let deadline = Instant::now() + Duration::from_secs(1); + loop { + let output = stdio.drain_output(); + if output.0 == expected_stdout && output.1 == expected_stderr { + return output; + } + assert!( + Instant::now() < deadline, + "timed out waiting for init output" + ); + thread::sleep(Duration::from_millis(5)); + } + } #[test] fn test_stdio_creation() { let result = ContainerStdio::new(); assert!(result.is_ok()); - let (stdio, init_fds) = result.unwrap(); + let (stdio, init_fds, _) = result.unwrap(); - // Verify all FDs are valid (positive integers) assert!(stdio.stdin_tx.as_raw_fd() >= 0); - assert!(stdio.stdout_rx.as_ref().unwrap().as_raw_fd() >= 0); - assert!(stdio.stderr_rx.as_ref().unwrap().as_raw_fd() >= 0); assert!(init_fds.stdin.as_raw_fd() >= 0); assert!(init_fds.stdout.as_raw_fd() >= 0); assert!(init_fds.stderr.as_raw_fd() >= 0); - // Verify all FDs are unique let fds = [ stdio.stdin_tx.as_raw_fd(), - stdio.stdout_rx.as_ref().unwrap().as_raw_fd(), - stdio.stderr_rx.as_ref().unwrap().as_raw_fd(), init_fds.stdin.as_raw_fd(), init_fds.stdout.as_raw_fd(), init_fds.stderr.as_raw_fd(), @@ -218,9 +285,8 @@ mod tests { #[test] fn test_drain_output_captures_data() { - let (mut stdio, init_fds) = ContainerStdio::new().unwrap(); + let (stdio, init_fds, _) = ContainerStdio::new().unwrap(); - // Write to the init side of pipes (simulating init process output) let mut stdout_writer = std::fs::File::from(init_fds.stdout); let mut stderr_writer = std::fs::File::from(init_fds.stderr); stdout_writer.write_all(b"hello stdout").unwrap(); @@ -228,26 +294,99 @@ mod tests { drop(stdout_writer); drop(stderr_writer); - let (stdout, stderr) = stdio.drain_output(); + let (stdout, stderr) = wait_for_output(&stdio, "hello stdout", "hello stderr"); assert_eq!(stdout, "hello stdout"); assert_eq!(stderr, "hello stderr"); } #[test] - fn test_drain_output_returns_empty_on_second_call() { - let (mut stdio, init_fds) = ContainerStdio::new().unwrap(); + fn test_drain_output_returns_current_tail() { + let (stdio, init_fds, _) = ContainerStdio::new().unwrap(); let mut stdout_writer = std::fs::File::from(init_fds.stdout); stdout_writer.write_all(b"data").unwrap(); drop(stdout_writer); drop(init_fds.stderr); - let (stdout, _) = stdio.drain_output(); + let (stdout, stderr) = wait_for_output(&stdio, "data", ""); assert_eq!(stdout, "data"); - - // Second call returns empty (fds already taken) let (stdout2, stderr2) = stdio.drain_output(); - assert_eq!(stdout2, ""); - assert_eq!(stderr2, ""); + assert_eq!(stdout2, stdout); + assert_eq!(stderr2, stderr); + } + + #[test] + fn test_drain_output_does_not_wait_for_open_writer() { + let (stdio, init_fds, _) = ContainerStdio::new().unwrap(); + let stdout_writer = std::fs::File::from(init_fds.stdout); + drop(init_fds.stderr); + + let (snapshot_tx, snapshot_rx) = mpsc::channel(); + let snapshotter = thread::spawn(move || { + let _ = snapshot_tx.send(stdio.drain_output()); + }); + + let snapshot = snapshot_rx.recv_timeout(Duration::from_secs(1)); + drop(stdout_writer); + snapshotter.join().unwrap(); + + assert!(snapshot.is_ok(), "draining output waited for pipe EOF"); + } + + #[test] + fn test_drain_output_keeps_large_writers_unblocked() { + let (stdio, init_fds, _) = ContainerStdio::new().unwrap(); + let mut output = vec![b'x'; 1024 * 1024]; + output.extend_from_slice(b"tail-marker"); + drop(init_fds.stderr); + + let (completed_tx, completed_rx) = mpsc::channel(); + let writer = thread::spawn(move || { + let mut stdout_writer = std::fs::File::from(init_fds.stdout); + let result = stdout_writer.write_all(&output); + let _ = completed_tx.send(result); + }); + + let completed = completed_rx.recv_timeout(Duration::from_secs(5)); + if completed.is_err() { + drop(stdio); + let _ = writer.join(); + panic!("init stdout writer blocked while the reader was open"); + } + completed.unwrap().unwrap(); + writer.join().unwrap(); + + let deadline = Instant::now() + Duration::from_secs(1); + loop { + let (stdout, _) = stdio.drain_output(); + if stdout.ends_with("tail-marker") { + break; + } + assert!( + Instant::now() < deadline, + "timed out waiting for output tail" + ); + thread::sleep(Duration::from_millis(5)); + } + } + + #[test] + fn test_output_completion_waits_for_both_pipes_to_reach_eof() { + let (_stdio, init_fds, completion) = ContainerStdio::new().unwrap(); + let stdout_writer = std::fs::File::from(init_fds.stdout); + drop(init_fds.stderr); + + let (result_tx, result_rx) = mpsc::channel(); + thread::spawn(move || { + let runtime = tokio::runtime::Runtime::new().unwrap(); + let _ = result_tx.send(runtime.block_on(completion.wait())); + }); + + assert!(result_rx.recv_timeout(Duration::from_millis(50)).is_err()); + drop(stdout_writer); + assert_eq!( + result_rx.recv_timeout(Duration::from_secs(1)).unwrap(), + Ok(()) + ); } } diff --git a/src/guest/src/container/supervisor.rs b/src/guest/src/container/supervisor.rs new file mode 100644 index 000000000..3b6dff161 --- /dev/null +++ b/src/guest/src/container/supervisor.rs @@ -0,0 +1,117 @@ +use super::stdio::InitOutputCompletion; +use super::zygote::{self, WaitResult}; +use nix::unistd::Pid; +use tokio::sync::watch; + +#[derive(Clone, Debug, PartialEq, Eq)] +pub(crate) enum InitTerminal { + Exited { code: i32 }, + Signaled { signal: i32 }, + Failed { reason: String }, +} + +#[derive(Clone)] +pub(crate) struct InitSupervisor { + generation: u64, + terminal: watch::Sender>, +} + +impl InitSupervisor { + pub(crate) fn new(generation: u64) -> Self { + let (terminal, _) = watch::channel(None); + Self { + generation, + terminal, + } + } + + pub(crate) fn matches_generation(&self, generation: u64) -> bool { + self.generation == generation + } + + pub(crate) fn fail(&self, reason: impl Into) { + self.terminal.send_replace(Some(InitTerminal::Failed { + reason: reason.into(), + })); + } + + pub(crate) fn supervise(&self, pid: Pid, output_completion: InitOutputCompletion) { + let terminal = self.terminal.clone(); + tokio::spawn(async move { + let (process_result, output_result) = + tokio::join!(wait_for_process(pid), output_completion.wait()); + + let outcome = match (process_result, output_result) { + (Ok(InitTerminal::Exited { code }), Ok(())) => InitTerminal::Exited { code }, + (Ok(InitTerminal::Signaled { signal }), Ok(())) => { + InitTerminal::Signaled { signal } + } + (Ok(InitTerminal::Failed { reason }), _) => InitTerminal::Failed { reason }, + (Err(reason), Ok(())) => InitTerminal::Failed { reason }, + (_, Err(reason)) => InitTerminal::Failed { reason }, + }; + terminal.send_replace(Some(outcome)); + }); + } + + /// Reconnects must observe an already-published terminal result immediately. + pub(crate) async fn wait(&self) -> InitTerminal { + let mut terminal = self.terminal.subscribe(); + loop { + if let Some(outcome) = terminal.borrow().clone() { + return outcome; + } + if terminal.changed().await.is_err() { + return InitTerminal::Failed { + reason: "init supervisor stopped".to_string(), + }; + } + } + } +} + +async fn wait_for_process(pid: Pid) -> Result { + loop { + let result = tokio::task::spawn_blocking(move || { + zygote::ZYGOTE.get().expect("zygote not started").wait(pid) + }) + .await + .map_err(|error| format!("init wait task failed: {error}"))? + .map_err(|error| format!("zygote init wait failed: {error}"))?; + + match result { + WaitResult::StillAlive => { + tokio::time::sleep(std::time::Duration::from_millis(10)).await + } + WaitResult::Exited { code } => return Ok(InitTerminal::Exited { code }), + WaitResult::Signaled { signal } => return Ok(InitTerminal::Signaled { signal }), + WaitResult::Failed { error } => return Ok(InitTerminal::Failed { reason: error }), + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[tokio::test] + async fn wait_returns_an_already_published_terminal_value() { + let supervisor = InitSupervisor::new(7); + supervisor.fail("create failed"); + + assert_eq!( + supervisor.wait().await, + InitTerminal::Failed { + reason: "create failed".to_string() + } + ); + } + + #[test] + fn generation_must_match_the_supervised_init() { + let supervisor = InitSupervisor::new(7); + + assert!(supervisor.matches_generation(7)); + assert!(!supervisor.matches_generation(8)); + } +} diff --git a/src/guest/src/container/zygote.rs b/src/guest/src/container/zygote.rs index d19900824..790b418f0 100644 --- a/src/guest/src/container/zygote.rs +++ b/src/guest/src/container/zygote.rs @@ -60,6 +60,13 @@ pub(crate) struct BuildSpec { pub gid: u32, } +#[derive(Serialize, Deserialize, Debug, Clone, PartialEq)] +pub(crate) struct InitSpec { + pub container_id: String, + pub state_root: PathBuf, + pub bundle_path: PathBuf, +} + /// Build outcome. Invalid states are unrepresentable. #[derive(Serialize, Deserialize, Debug, Clone, PartialEq)] pub(crate) enum BuildResult { @@ -67,6 +74,12 @@ pub(crate) enum BuildResult { Failed { error: String }, } +#[derive(Serialize, Deserialize, Debug, Clone, PartialEq)] +enum InitBuildResult { + Built, + Failed { error: String }, +} + /// Process exit outcome from waitpid, serialized over IPC. /// /// The zygote is the only process that can call waitpid on container @@ -93,10 +106,13 @@ pub(crate) enum WaitResult { enum ZygoteRequest { /// Build a new container process. May include SCM_RIGHTS fds for stdio pipes. Build(BuildSpec), + BuildInit(InitSpec), /// Wait for a container process to exit and return its exit status. /// The zygote must handle this because it's the parent of all container /// processes (they were created by clone3() inside the zygote). - Wait { pid: i32 }, + Wait { + pid: i32, + }, } /// Tagged IPC response from zygote to parent, matched 1:1 with requests. @@ -106,6 +122,7 @@ enum ZygoteRequest { #[derive(Serialize, Deserialize, Debug, Clone)] enum ZygoteResponse { Build(BuildResult), + BuildInit(InitBuildResult), Wait(WaitResult), } @@ -163,6 +180,21 @@ impl Zygote { } } + pub fn build_init(&self, spec: InitSpec, fds: [RawFd; 3]) -> BoxliteResult<()> { + let sock = self.sock.lock().unwrap(); + let fd = sock.as_raw_fd(); + send_request(fd, &ZygoteRequest::BuildInit(spec), Some(fds))?; + match recv_response(fd)? { + ZygoteResponse::BuildInit(InitBuildResult::Built) => Ok(()), + ZygoteResponse::BuildInit(InitBuildResult::Failed { error }) => { + Err(BoxliteError::Internal(error)) + } + other => Err(BoxliteError::Internal(format!( + "expected BuildInit response, got: {other:?}" + ))), + } + } + /// Wait for a container process to exit. Returns exit status. /// /// Container processes are direct children of the zygote (created by @@ -213,6 +245,13 @@ fn serve(sock: OwnedFd) -> ! { std::process::exit(1); } } + Ok((ZygoteRequest::BuildInit(spec), fds)) => { + let result = do_build_init(spec, fds); + if let Err(e) = send_response(fd, &ZygoteResponse::BuildInit(result)) { + eprintln!("[zygote] send_response failed: {e}"); + std::process::exit(1); + } + } Ok((ZygoteRequest::Wait { pid }, _)) => { let result = do_wait(pid); if let Err(e) = send_response(fd, &ZygoteResponse::Wait(result)) { @@ -230,6 +269,41 @@ fn serve(sock: OwnedFd) -> ! { } } +fn do_build_init(spec: InitSpec, fds: Option<[RawFd; 3]>) -> InitBuildResult { + let Some([stdin, stdout, stderr]) = fds else { + return InitBuildResult::Failed { + error: "init build missing stdio file descriptors".to_string(), + }; + }; + + let result = (|| -> Result<(), String> { + // SAFETY: SCM_RIGHTS transferred exclusive ownership to the zygote. + let stdin = unsafe { OwnedFd::from_raw_fd(stdin) }; + let stdout = unsafe { OwnedFd::from_raw_fd(stdout) }; + let stderr = unsafe { OwnedFd::from_raw_fd(stderr) }; + + ContainerBuilder::new(spec.container_id.clone(), SyscallType::default()) + .with_root_path(spec.state_root) + .map_err(|e| format!("Failed to set container root path: {e}"))? + .validate_id() + .map_err(|e| format!("Invalid container ID: {e}"))? + .with_stdin(stdin) + .with_stdout(stdout) + .with_stderr(stderr) + .as_init(&spec.bundle_path) + .with_systemd(false) + .with_detach(true) + .build() + .map_err(|e| format!("init build failed: {e}"))?; + Ok(()) + })(); + + match result { + Ok(()) => InitBuildResult::Built, + Err(error) => InitBuildResult::Failed { error }, + } +} + /// Execute a container tenant build. Called inside the zygote (single-threaded). /// /// This is the same ContainerBuilder chain that was in `command.rs build_and_spawn()`, @@ -540,6 +614,24 @@ mod tests { assert_eq!(result, decoded); } + #[test] + fn init_build_result_serde_roundtrip() { + let result = InitBuildResult::Built; + let json = serde_json::to_vec(&result).unwrap(); + let decoded: InitBuildResult = serde_json::from_slice(&json).unwrap(); + assert_eq!(result, decoded); + } + + #[test] + fn init_build_result_failed_serde_roundtrip() { + let result = InitBuildResult::Failed { + error: "init build failed: container not found".to_string(), + }; + let json = serde_json::to_vec(&result).unwrap(); + let decoded: InitBuildResult = serde_json::from_slice(&json).unwrap(); + assert_eq!(result, decoded); + } + // --- WaitResult serde tests --- // WaitResult crosses the IPC boundary; verify it survives JSON serialization. diff --git a/src/guest/src/service/container.rs b/src/guest/src/service/container.rs index a07869771..15319d703 100644 --- a/src/guest/src/service/container.rs +++ b/src/guest/src/service/container.rs @@ -7,14 +7,16 @@ use std::path::Path; use crate::service::server::GuestServer; use boxlite_shared::{ - container_init_response, rootfs_init, Container as ContainerService, ContainerInitError, - ContainerInitRequest, ContainerInitResponse, ContainerInitSuccess, Filesystem, RootfsInit, + container_exited, container_init_response, container_wait_response, rootfs_init, + Container as ContainerService, ContainerExited, ContainerInitError, ContainerInitRequest, + ContainerInitResponse, ContainerInitSuccess, ContainerWaitFailed, ContainerWaitRequest, + ContainerWaitResponse, Filesystem, RootfsInit, }; use nix::mount::{mount, MsFlags}; use tonic::{Request, Response, Status}; use tracing::{debug, error, info, warn}; -use crate::container::{Container, UserMount}; +use crate::container::{Container, InitSupervisor, InitTerminal, UserMount}; use crate::layout::GuestLayout; use crate::storage::block_device::BlockDeviceMount; @@ -119,21 +121,62 @@ impl ContainerService for GuestServer { } } - // Extract container config - let config = init_req - .container_config - .ok_or_else(|| Status::invalid_argument("Missing container_config in Init request"))?; + let lifecycle_generation = init_req.lifecycle_generation; + let supervisor = { + let mut supervisors = self.init_supervisors.lock().await; + match supervisors.get(&container_id) { + Some(existing) if !existing.matches_generation(lifecycle_generation) => { + return Ok(Response::new(ContainerInitResponse { + result: Some(container_init_response::Result::Error(ContainerInitError { + reason: + "Container init already exists for another lifecycle generation" + .to_string(), + })), + })); + } + Some(_) => { + return Ok(Response::new(ContainerInitResponse { + result: Some(container_init_response::Result::Error(ContainerInitError { + reason: "Container init is already in progress".to_string(), + })), + })); + } + None => { + let supervisor = InitSupervisor::new(lifecycle_generation); + supervisors.insert(container_id.clone(), supervisor.clone()); + supervisor + } + } + }; + + let config = match init_req.container_config { + Some(config) => config, + None => { + return Ok(Response::new(init_error( + &supervisor, + "Missing container_config in Init request".to_string(), + ))); + } + }; - // Validate configuration if config.entrypoint.is_empty() { error!("Invalid container config: entrypoint cannot be empty"); - return Ok(Response::new(ContainerInitResponse { - result: Some(container_init_response::Result::Error(ContainerInitError { - reason: "Invalid container config: entrypoint cannot be empty".to_string(), - })), - })); + return Ok(Response::new(init_error( + &supervisor, + "Invalid container config: entrypoint cannot be empty".to_string(), + ))); } + let rootfs_init = match init_req.rootfs { + Some(rootfs_init) => rootfs_init, + None => { + return Ok(Response::new(init_error( + &supervisor, + "Missing rootfs in Container.Init request".to_string(), + ))); + } + }; + info!("🚀 Starting OCI container with received configuration"); // Compute rootfs paths from container_id @@ -148,27 +191,18 @@ impl ContainerService for GuestServer { // Create bundle rootfs directory if let Err(e) = std::fs::create_dir_all(&bundle_rootfs) { error!("Failed to create bundle rootfs directory: {}", e); - return Ok(Response::new(ContainerInitResponse { - result: Some(container_init_response::Result::Error(ContainerInitError { - reason: format!("Failed to create bundle rootfs directory: {}", e), - })), - })); + return Ok(Response::new(init_error( + &supervisor, + format!("Failed to create bundle rootfs directory: {}", e), + ))); } // Handle rootfs initialization based on strategy - let rootfs_init = init_req - .rootfs - .ok_or_else(|| Status::invalid_argument("Missing rootfs in Container.Init request"))?; - if let Err(reason) = prepare_rootfs(&rootfs_init, &container_id, &shared_rootfs, &self.layout) { error!("{}", reason); - return Ok(Response::new(ContainerInitResponse { - result: Some(container_init_response::Result::Error(ContainerInitError { - reason, - })), - })); + return Ok(Response::new(init_error(&supervisor, reason))); } // Bind mount shared rootfs to bundle rootfs @@ -180,11 +214,10 @@ impl ContainerService for GuestServer { None::<&str>, ) { error!("Failed to bind mount rootfs: {}", e); - return Ok(Response::new(ContainerInitResponse { - result: Some(container_init_response::Result::Error(ContainerInitError { - reason: format!("Failed to bind mount rootfs: {}", e), - })), - })); + return Ok(Response::new(init_error( + &supervisor, + format!("Failed to bind mount rootfs: {}", e), + ))); } // Install CA certs into container trust store (from gRPC CACert field). @@ -261,39 +294,20 @@ impl ContainerService for GuestServer { &config.workdir, &config.user, user_mounts, - ) { - Ok(mut container) => { - debug!(container_id = %container_id, "Container started, checking if init process is running"); - // Verify container init process is running - if !container.is_running() { - // Gather diagnostic information (includes init stdout/stderr) - let diagnostics = container.diagnose_exit(); - - error!( - "Container init process exited immediately after start. Diagnostics: {}", - diagnostics - ); - - return Ok(Response::new(ContainerInitResponse { - result: Some(container_init_response::Result::Error(ContainerInitError { - reason: format!( - "Container init process exited immediately. {}", - diagnostics - ), - })), - })); - } - + ) + .await + { + Ok(started) => { info!( container_id = %container_id, - "✅ Container started successfully and ready for exec" + "Container init started" ); - // Store container in registry self.containers.lock().await.insert( container_id.clone(), - std::sync::Arc::new(tokio::sync::Mutex::new(container)), + std::sync::Arc::new(tokio::sync::Mutex::new(started.container)), ); + supervisor.supervise(started.init_pid, started.output_completion); Ok(Response::new(ContainerInitResponse { result: Some(container_init_response::Result::Success( @@ -303,12 +317,64 @@ impl ContainerService for GuestServer { } Err(e) => { error!("Failed to start container: {}", e); - Ok(Response::new(ContainerInitResponse { - result: Some(container_init_response::Result::Error(ContainerInitError { - reason: format!("Failed to start container: {}", e), - })), - })) + Ok(Response::new(init_error( + &supervisor, + format!("Failed to start container: {}", e), + ))) } } } + + async fn wait( + &self, + request: Request, + ) -> Result, Status> { + let request = request.into_inner(); + if request.container_id.is_empty() { + return Err(Status::invalid_argument("missing container_id")); + } + + let supervisor = self + .init_supervisors + .lock() + .await + .get(&request.container_id) + .cloned() + .ok_or_else(|| Status::unavailable("container init is not ready"))?; + + if !supervisor.matches_generation(request.lifecycle_generation) { + return Err(Status::failed_precondition( + "container lifecycle generation does not match", + )); + } + + let result = match supervisor.wait().await { + InitTerminal::Exited { code } => { + container_wait_response::Result::Exited(ContainerExited { + cause: Some(container_exited::Cause::ExitCode(code)), + }) + } + InitTerminal::Signaled { signal } => { + container_wait_response::Result::Exited(ContainerExited { + cause: Some(container_exited::Cause::Signal(signal)), + }) + } + InitTerminal::Failed { reason } => { + container_wait_response::Result::Failed(ContainerWaitFailed { reason }) + } + }; + + Ok(Response::new(ContainerWaitResponse { + result: Some(result), + })) + } +} + +fn init_error(supervisor: &InitSupervisor, reason: String) -> ContainerInitResponse { + supervisor.fail(reason.clone()); + ContainerInitResponse { + result: Some(container_init_response::Result::Error(ContainerInitError { + reason, + })), + } } diff --git a/src/guest/src/service/exec/mod.rs b/src/guest/src/service/exec/mod.rs index 12a3a6bc3..54ba44843 100644 --- a/src/guest/src/service/exec/mod.rs +++ b/src/guest/src/service/exec/mod.rs @@ -428,7 +428,7 @@ async fn spawn_with_executor( Ok(h) => h, Err(e) => { // Check if container init died — provide actionable diagnostics - let mut container = container_ref.lock().await; + let container = container_ref.lock().await; if !container.is_running() { let (init_stdout, init_stderr) = container.drain_init_output(); let mut msg = format!( diff --git a/src/guest/src/service/server.rs b/src/guest/src/service/server.rs index 31cceafbb..e469bcfd8 100644 --- a/src/guest/src/service/server.rs +++ b/src/guest/src/service/server.rs @@ -1,4 +1,4 @@ -use crate::container::Container; +use crate::container::{Container, InitSupervisor}; use crate::layout::GuestLayout; use crate::service::exec::registry::ExecutionRegistry; use boxlite_shared::{BoxliteResult, Transport}; @@ -34,6 +34,8 @@ pub(crate) struct GuestServer { /// Container registry: container_id -> Container pub containers: Arc>>>>, + pub init_supervisors: Arc>>, + /// Execution registry for tracking running executions pub registry: ExecutionRegistry, @@ -51,6 +53,7 @@ impl GuestServer { layout, init_state: Arc::new(Mutex::new(GuestInitState::default())), containers: Arc::new(Mutex::new(HashMap::new())), + init_supervisors: Arc::new(Mutex::new(HashMap::new())), registry: ExecutionRegistry::new(), frozen_mounts: Mutex::new(Vec::new()), } diff --git a/src/shared/proto/boxlite/v1/service.proto b/src/shared/proto/boxlite/v1/service.proto index d3b1b0dae..c5e03164f 100644 --- a/src/shared/proto/boxlite/v1/service.proto +++ b/src/shared/proto/boxlite/v1/service.proto @@ -11,6 +11,9 @@ service Container { // Initialize OCI container (called after GuestInit) // Prepares rootfs, then starts the container with the provided configuration rpc Init(ContainerInitRequest) returns (ContainerInitResponse); + + // Wait for the init process of one lifecycle generation to terminate. + rpc Wait(ContainerWaitRequest) returns (ContainerWaitResponse); } // Guest agent management @@ -208,6 +211,9 @@ message ContainerInitRequest { // Additional CA certificates to install in the container trust store. // Used for MITM secret substitution — the container trusts the proxy CA. repeated CACert ca_certs = 5; + // Generated by the host for this VM run. A stale monitor must not observe + // the terminal result of a later run with the same container ID. + uint64 lifecycle_generation = 6; } // A CA certificate to add to the container's trust store. @@ -251,6 +257,29 @@ message ContainerInitError { string reason = 1; } +message ContainerWaitRequest { + string container_id = 1; + uint64 lifecycle_generation = 2; +} + +message ContainerWaitResponse { + oneof result { + ContainerExited exited = 1; + ContainerWaitFailed failed = 2; + } +} + +message ContainerExited { + oneof cause { + int32 exit_code = 1; + int32 signal = 2; + } +} + +message ContainerWaitFailed { + string reason = 1; +} + // Container configuration (OCI-derived, from image) message ContainerConfig { // Entrypoint command (e.g., ["/bin/sh", "-c", "echo hello"])