diff --git a/CubeShim/shim/src/container/mod.rs b/CubeShim/shim/src/container/mod.rs index f29b721dc..dc7d0db88 100644 --- a/CubeShim/shim/src/container/mod.rs +++ b/CubeShim/shim/src/container/mod.rs @@ -36,6 +36,98 @@ use crate::{infof, warnf}; pub const GUEST_DEV_SHM: &str = "/run/cube-containers/sandbox/shm"; pub const ANNO_APP_SNAPSHOT_CONTAINER_ID: &str = "cube.appsnapshot.container.id"; +/// Upper bound on the dedicated vsock connect in start_log_forward. It runs +/// while holding log_forward_lifecycle, so an unbounded connect would serialize +/// and stall all other log-forward lifecycle operations on this container. +const LOG_FORWARD_CONNECT_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(10); + +#[derive(Default)] +struct LogForward { + handle: Option>, + cancel: Option>, +} + +/// Shared, clone-safe owner of the init log-forwarding background task. +/// +/// `Container` is `#[derive(Clone)]` and clones are pulled out of the map in +/// paths like delete/wait, so the task must be owned through a single shared +/// slot rather than a per-clone `Arc`. `lifecycle` (permits = 1) +/// serializes start/stop across all clones so exactly one caller drains the +/// task to completion instead of falling back to `abort()`. +#[derive(Clone)] +struct LogForwardHandle { + slot: Arc>, + lifecycle: Arc, +} + +impl LogForwardHandle { + fn new() -> Self { + Self { + slot: Arc::new(Mutex::new(LogForward::default())), + lifecycle: Arc::new(tokio::sync::Semaphore::new(1)), + } + } + + /// Acquire the start/stop serialization permit. Held for the whole + /// duration of a start or stop so the two never interleave across clones. + async fn acquire(&self) -> tokio::sync::OwnedSemaphorePermit { + self.lifecycle + .clone() + .acquire_owned() + .await + .expect("log-forward lifecycle semaphore closed") + } + + /// Cancel and await the current task to completion. Caller must hold the + /// lifecycle permit. The slot mutex is released before the `.await` so it + /// is never held across task termination. + /// + /// The `handle.await` is intentionally unbounded: draining instead of + /// aborting is the whole point of the fix, and a bounded drain + `abort()` + /// fallback would reintroduce the skipped-IO-drain hazard this removes. In + /// the pathological case (the task stuck mid-`file.write_all` on a hung log + /// file rather than parked on `cancel.changed()`), this blocks while holding + /// the lifecycle permit. On the `disconnect_agent`/`resume`/`kill_container` + /// paths the drain additionally runs under the sandbox `containers` mutex + /// (the map is iterated in place via `unset_client`, or the entry is borrowed + /// via `get_mut` in `kill_container` -> `signal_container`), so a wedged task + /// stalls every other lifecycle op on the sandbox. On the clone path (`delete_container`, + /// which drains via `destroy_container` -> `signal_container`) the container + /// is cloned out of the map and the `containers` lock is released before the + /// drain — but the service layer still holds the sandbox-wide `Mutex` + /// across the whole RPC (task_srv serializes every op on it), so a wedged + /// drain freezes pause/resume/kill/delete for the whole sandbox on this path + /// too, and this path previously took the `abort()` fallback and returned + /// promptly, so draining is a deliberate *new* blocking point on it, not a + /// pre-existing one. (`wait_container` clones out of the map too but never + /// drains.) We accept it: the forwarding loops append to local log files + /// whose writes do not normally stall. + async fn drain(&self) { + let (cancel, handle) = { + let mut slot = self.slot.lock().await; + (slot.cancel.take(), slot.handle.take()) + }; + if let Some(tx) = cancel { + let _ = tx.send(true); + } + if let Some(handle) = handle { + let _ = handle.await; + } + } + + /// Install a freshly started task. Caller must hold the lifecycle permit + /// and must have already drained any previous task. + async fn store( + &self, + cancel: tokio::sync::watch::Sender, + handle: tokio::task::JoinHandle<()>, + ) { + let mut slot = self.slot.lock().await; + slot.cancel = Some(cancel); + slot.handle = Some(handle); + } +} + fn validate_log_path_component(id: &str) -> CResult<()> { if id.is_empty() || id.contains('/') || id.contains("..") || id.contains('\0') { return Err(format!("invalid container id for log path: {}", id)); @@ -61,13 +153,9 @@ pub struct Container { /// Background task forwarding container stdout/stderr to log files. /// Template creation: /data/log/template//stdout|stderr (755 dir). /// Normal sandbox: ./stdout and ./stderr relative to the bundle directory. - /// Aborted on pause/snapshot/disconnect/kill/destroy; restarted on resume via start_log_forward. - log_forward_handle: Option>>, - /// Cancel sender for the log-forwarding task. Sending true on this watch - /// channel wakes both forward_stdout and forward_stderr select! loops so - /// they exit immediately; paired with log_forward_handle so callers can - /// await clean termination before proceeding with pause / snapshot. - log_forward_cancel: Option>, + /// Clones share the task ownership so exactly one caller takes and awaits + /// it; start/stop are serialized across clones by an internal semaphore. + log_forward: LogForwardHandle, } impl Container { @@ -109,8 +197,7 @@ impl Container { execs: Arc::new(Mutex::new(HashMap::new())), tx_containerd, app_snapshot, - log_forward_handle: None, - log_forward_cancel: None, + log_forward: LogForwardHandle::new(), }; Ok(c) } @@ -161,19 +248,8 @@ impl Container { /// forward_init_log_stdout/stderr and awaits the background task so vsock /// reads are finished before pause, snapshot, or destroy proceeds. pub async fn stop_log_forward(&mut self) { - if let Some(tx) = self.log_forward_cancel.take() { - let _ = tx.send(true); - } - if let Some(handle) = self.log_forward_handle.take() { - match Arc::try_unwrap(handle) { - Ok(h) => { - let _ = h.await; - } - Err(h) => { - h.abort(); - } - } - } + let _lifecycle = self.log_forward.acquire().await; + self.log_forward.drain().await; } pub async fn unset_client(&mut self) { @@ -575,25 +651,26 @@ impl Container { /// read has stopped before proceeding. pub async fn start_log_forward(&mut self) -> CResult<()> { // Cancel and await any previous instance before starting a new one. - if let Some(tx) = self.log_forward_cancel.take() { - let _ = tx.send(true); - } - if let Some(handle) = self.log_forward_handle.take() { - match Arc::try_unwrap(handle) { - Ok(h) => { - let _ = h.await; - } - Err(h) => { - h.abort(); - } - } - } + let _lifecycle = self.log_forward.acquire().await; + self.log_forward.drain().await; // Open a dedicated vsock connection for streaming I/O so that the // main client connection used for control-plane RPCs is never blocked. - let log_conn = AsyncUtils::connect_agent(&self.sandbox_id) - .await - .map_err(|e| format!("connect agent for log forwarding failed:{}", e))?; + // Bound the connect: it runs while holding log_forward_lifecycle, so a + // hung agent would otherwise wedge every start/stop (pause, snapshot, + // kill, destroy) on this container and its clones indefinitely. + let log_conn = tokio::time::timeout( + LOG_FORWARD_CONNECT_TIMEOUT, + AsyncUtils::connect_agent(&self.sandbox_id), + ) + .await + .map_err(|_| { + format!( + "connect agent for log forwarding timed out after {}s", + LOG_FORWARD_CONNECT_TIMEOUT.as_secs() + ) + })? + .map_err(|e| format!("connect agent for log forwarding failed:{}", e))?; let log_client = agent_ttrpc::AgentServiceClient::new(log_conn); // Write log files: @@ -637,16 +714,18 @@ impl Container { stderr_path ); - // Create a watch cancel channel. unset_client() sends true on the tx - // side; the rx is cloned into each of forward_stdout and forward_stderr - // so both loops wake and exit immediately via tokio::select!. + // Create a watch cancel channel. The tx is stored in the log_forward + // slot; on stop (pause / snapshot / kill / destroy) `stop_log_forward` + // calls `LogForwardHandle::drain()`, which takes the tx and sends true. + // The rx is cloned into each of forward_init_log_stdout and + // forward_init_log_stderr so both loops wake and exit immediately via + // tokio::select!. let (cancel_tx, cancel_rx) = tokio::sync::watch::channel(false); let handle = log_exec .start_log_forward(log_client, self.log.clone(), cancel_rx) .await; - self.log_forward_handle = Some(Arc::new(handle)); - self.log_forward_cancel = Some(cancel_tx); + self.log_forward.store(cancel_tx, handle).await; Ok(()) } @@ -1089,3 +1168,131 @@ impl Container { self.id.clone() } } + +#[cfg(test)] +mod log_forward_tests { + use super::LogForwardHandle; + use std::sync::atomic::{AtomicUsize, Ordering}; + use std::sync::Arc; + use std::time::Duration; + + /// Spawn a task that mimics a log-forward loop: it runs until cancelled, + /// then records a clean exit. This is what the pre-fix `abort()` fallback + /// used to skip. + async fn install_task(handle: &LogForwardHandle, drained: Arc) { + let (cancel_tx, mut cancel_rx) = tokio::sync::watch::channel(false); + let task = tokio::spawn(async move { + loop { + if cancel_rx.changed().await.is_err() { + return; + } + if *cancel_rx.borrow() { + // Simulate the IO drain that pause/snapshot/destroy rely on. + tokio::time::sleep(Duration::from_millis(10)).await; + drained.fetch_add(1, Ordering::SeqCst); + return; + } + } + }); + let _permit = handle.acquire().await; + handle.store(cancel_tx, task).await; + } + + /// The core invariant of the fix: a task installed through the shared slot + /// is always drained to completion, even when a *clone* of the handle stops + /// it. The pre-fix code fell back to `abort()` here and skipped the drain. + #[tokio::test] + async fn drain_completes_when_stopped_through_clone() { + let handle = LogForwardHandle::new(); + let drained = Arc::new(AtomicUsize::new(0)); + + install_task(&handle, drained.clone()).await; + + let clone = handle.clone(); + let _permit = clone.acquire().await; + clone.drain().await; + + assert_eq!( + drained.load(Ordering::SeqCst), + 1, + "task must be awaited to a clean exit, not aborted" + ); + } + + /// Concurrent stops from two clones must serialize and both observe a clean + /// slot: exactly one drains the task, neither aborts. + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn concurrent_stops_serialize_and_drain() { + let handle = LogForwardHandle::new(); + let drained = Arc::new(AtomicUsize::new(0)); + + install_task(&handle, drained.clone()).await; + + let a = handle.clone(); + let b = handle.clone(); + let stop_a = tokio::spawn(async move { + let _permit = a.acquire().await; + a.drain().await; + }); + let stop_b = tokio::spawn(async move { + let _permit = b.acquire().await; + b.drain().await; + }); + let (ra, rb) = tokio::join!(stop_a, stop_b); + ra.unwrap(); + rb.unwrap(); + + assert_eq!( + drained.load(Ordering::SeqCst), + 1, + "exactly one caller must drain the task to completion" + ); + } + + /// A stop arriving while a start holds the permit mid-connect must wait for + /// the start to install its task, then drain that task to completion — never + /// racing the half-built slot. This is the interleaving the semaphore + /// serializes. + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn stop_waits_for_in_flight_start_then_drains() { + let handle = LogForwardHandle::new(); + let drained = Arc::new(AtomicUsize::new(0)); + + // Simulate start_log_forward holding the permit through its connect + // before it stores a task: acquire the permit and keep it. + let start_permit = handle.acquire().await; + + // A concurrent stop tries to acquire; it must block behind the start. + let stop_handle = handle.clone(); + let stop_drained = drained.clone(); + let stop = tokio::spawn(async move { + let _permit = stop_handle.acquire().await; + stop_handle.drain().await; + stop_drained.load(Ordering::SeqCst) + }); + + // Finish the start: install the task, then release the permit. + let (cancel_tx, mut cancel_rx) = tokio::sync::watch::channel(false); + let task_drained = drained.clone(); + let task = tokio::spawn(async move { + loop { + if cancel_rx.changed().await.is_err() { + return; + } + if *cancel_rx.borrow() { + tokio::time::sleep(Duration::from_millis(10)).await; + task_drained.fetch_add(1, Ordering::SeqCst); + return; + } + } + }); + handle.store(cancel_tx, task).await; + drop(start_permit); + + let observed = stop.await.unwrap(); + assert_eq!( + observed, 1, + "stop must drain the task the start installed, not abort it" + ); + } +} diff --git a/tests/e2e/sdk_compat/cases/lifecycle/test_log_forward_cycles.py b/tests/e2e/sdk_compat/cases/lifecycle/test_log_forward_cycles.py new file mode 100644 index 000000000..6a4d17fbd --- /dev/null +++ b/tests/e2e/sdk_compat/cases/lifecycle/test_log_forward_cycles.py @@ -0,0 +1,151 @@ +# Copyright (c) 2026 Tencent Inc. +# SPDX-License-Identifier: Apache-2.0 + +"""E2E lifecycle smoke test for repeated pause/resume. + +Companion to commit "shim: serialize log-forward start/stop across Container +clones" (CubeShim/shim/src/container/mod.rs). Pause tears down each container's +log-forward task (``disconnect_agent`` -> ``unset_client`` -> ``stop_log_forward``) +and resume restarts it (``set_client`` -> ``start_log_forward``); the fix shares +one mutexed slot and serializes start/stop with a semaphore so exactly one +caller drains the task to completion instead of falling back to ``abort()``. + +Scope note: these SDK-level checks do NOT directly observe the skipped-IO-drain +bug. ``run_command`` streams through envd's process API over HTTP and +``read_file``/``write_file`` hit the sandbox filesystem — neither path traverses +the shim's init log-forward vsock task that the fix changes, so the assertions +here also pass against the pre-fix code. The actual regression guards for the +drain invariant are the Rust unit tests in ``container/mod.rs`` +(``log_forward_tests``). What these tests add is a black-box guarantee that +repeated pause/resume cycles stay healthy end-to-end: a wedged drain/resume +would trip ``wait_until_running`` and a corrupted restart would diverge the +command output or marker file. +""" + +from __future__ import annotations + +import pytest + +from framework.assertions import assert_command_ok +from framework.capabilities import COMMANDS, PAUSE_RESUME +from framework.lifecycle import ( + wait_until_data_plane_ready, + wait_until_paused, + wait_until_running, +) + +pytestmark = [ + pytest.mark.e2e, + pytest.mark.sdk_compat, + pytest.mark.lifecycle, + pytest.mark.p1, + pytest.mark.requires_capability(PAUSE_RESUME), + pytest.mark.requires_capability(COMMANDS), +] + +_CYCLES = 4 +# 16KB comfortably exceeds the 4096-byte read frame used by the exec-output +# relay, so the payload spans several frames rather than a single read. +_OUTPUT_SIZE = 16384 + + +def test_repeated_pause_resume_keeps_commands_working(sdk_sandbox, sdk_e2e_config): + """Each pause/resume drives stop/start_log_forward; the cycle must stay healthy.""" + marker_path = "/tmp/sdk-compat-log-forward-cycles.txt" + sdk_sandbox.write_file(marker_path, "cycle-0") + + # Pause through the handle returned by the previous resume rather than reusing + # the original pre-pause handle across cycles: a resume may invalidate the old + # handle's connection, which would fail later cycles for reasons unrelated to + # the shim change. The fixture owns `sdk_sandbox`; only the intermediate + # resumed handles are closed here. + current = sdk_sandbox + try: + for cycle in range(_CYCLES): + current.pause(timeout=sdk_e2e_config.default_timeout) + assert wait_until_paused( + current, timeout=sdk_e2e_config.default_timeout + ) == "paused", f"cycle {cycle}: sandbox did not reach paused" + + resumed = current.resume_or_connect(timeout=sdk_e2e_config.default_timeout) + if current is not sdk_sandbox: + current.close() + current = resumed + + assert wait_until_running( + current, timeout=sdk_e2e_config.default_timeout + ) == "running", f"cycle {cycle}: sandbox did not resume to running" + + # Control-plane `running` can precede CubeProxy/envd readiness, so + # wait for the data plane before hitting read_file/run_command. + wait_until_data_plane_ready( + current, + timeout=sdk_e2e_config.default_timeout, + command_timeout=sdk_e2e_config.command_timeout, + ) + + # File state must survive every cycle (drain/restart must not corrupt). + assert current.read_file(marker_path) == f"cycle-{cycle}", ( + f"cycle {cycle}: marker file content diverged" + ) + + result = current.run_command( + f"printf 'resumed-{cycle}'", + timeout=sdk_e2e_config.command_timeout, + ) + assert_command_ok(result) + assert result.stdout == f"resumed-{cycle}", ( + f"cycle {cycle}: command output diverged: {result.stdout!r}" + ) + + current.write_file(marker_path, f"cycle-{cycle + 1}") + finally: + if current is not sdk_sandbox: + current.close() + + +def test_pause_resume_preserves_large_command_output( + sdk_sandbox, sdk_backend, sdk_e2e_config +): + """A large-output command after resume must arrive intact. + + Complements ``test_repeated_pause_resume_keeps_commands_working`` by checking + that a large exec-output payload survives a resume without truncation or + corruption. Like that test, this exercises the envd exec-output stream rather + than the shim's init log-forward vsock task directly; the drain invariant + itself is guarded by the Rust unit tests. Kept as a large-payload lifecycle + smoke check. + """ + # The 16KB payload spans several 4096-byte read frames (the len used by the + # envd exec-output relay). Another backend's SDK may cap or line-wrap + # run_command stdout below 16KB, which would fail the exact-equality assert + # for reasons unrelated to this change. + if sdk_backend != "cubesandbox": + pytest.skip( + f"large-output drain regression is cube-shim specific, got {sdk_backend!r}" + ) + sdk_sandbox.pause(timeout=sdk_e2e_config.default_timeout) + resumed = sdk_sandbox.resume_or_connect(timeout=sdk_e2e_config.default_timeout) + try: + assert wait_until_running( + resumed, timeout=sdk_e2e_config.default_timeout + ) == "running", "sandbox did not resume to running" + + # Control-plane `running` can precede CubeProxy/envd readiness, so wait + # for the data plane before running the command. + wait_until_data_plane_ready( + resumed, + timeout=sdk_e2e_config.default_timeout, + command_timeout=sdk_e2e_config.command_timeout, + ) + + payload = "z" * _OUTPUT_SIZE + result = resumed.run_command( + f"printf '%s' '{payload}'", + timeout=sdk_e2e_config.command_timeout, + ) + assert_command_ok(result) + assert len(result.stdout) == _OUTPUT_SIZE + assert result.stdout == payload + finally: + resumed.close()