diff --git a/docs/developers-guide.md b/docs/developers-guide.md index eaa5d48f..fbe7bfc3 100644 --- a/docs/developers-guide.md +++ b/docs/developers-guide.md @@ -45,6 +45,12 @@ src/engine/connection/exec/ +-- protocol.rs # Protocol proxy loops, stdin forwarding, | # Stdout Purity Contract, STDIN_BUFFER_CAPACITY, | # STDIN_SETTLE_TIMEOUT, ProtocolProxyIo ++-- acp_helpers.rs # ACP initialize-frame rewriting; forwards the +| # first newline-delimited frame from host stdin, +| # masks terminal/* and fs/* clientCapabilities, +| # and re-serialises the JSON before forwarding +| # to the container; bounded by +| # MAX_FIRST_FRAME_BYTES +-- attached.rs # Attached-mode session, terminal resize, | # SIGWINCH handling, stdin echo forwarding +-- terminal.rs # Terminal size detection (stty), resize helpers, @@ -60,10 +66,11 @@ src/engine/connection/exec/ | # PODBOT_DISABLE_STDIN_FORWARDING_FOR_TESTS=1 +-- session.rs # ExecSessionOptions struct and | # protocol_session_options helper; controls -| # per-call session knobs (e.g. stdin forwarding -| # disable seam for tests; -| # with_protocol_stdin_forwarding_disabled(bool) -| # is compiled only for #[cfg(test)] builds +| # per-call session knobs (stdin forwarding +| # disable seam for tests via +| # with_protocol_stdin_forwarding_disabled(bool); +| # ACP initialize rewriting opt-in via +| # with_acp_initialize_rewrite_enabled(bool)) +-- runtime_helpers.rs # Blocking runtime helpers for synchronous exec | # wrappers; block_on_runtime detects nested | # Tokio contexts and routes to block_in_place @@ -84,6 +91,18 @@ src/engine/connection/exec/ +-- detached_helpers.rs # Detached mode test helpers +-- lifecycle_helpers.rs # Shared lifecycle fixtures +-- protocol_helpers.rs # Protocol-specific test helpers + +-- protocol_acp_tests.rs # Unit and shared BDD fixture helpers + | # for ACP capability masking; + | # RecordingInputWriter, frame + | # builders, run_forwarding harness, + | # and pub(super) BDD fixtures + +-- protocol_acp_bdd_tests.rs # rstest-bdd BDD scenarios for ACP + | # capability masking (masked + | # initialize, malformed pass-through, + | # unblocked pass-through) + +-- protocol_acp_forwarding_tests.rs # Forwarding path tests: rewrite + # enabled/disabled flag, oversized + # frame timeout safety ``` ### 3.1. Execution modes diff --git a/src/engine/connection/exec/acp_helpers.rs b/src/engine/connection/exec/acp_helpers.rs index 234e0826..aaa95180 100644 --- a/src/engine/connection/exec/acp_helpers.rs +++ b/src/engine/connection/exec/acp_helpers.rs @@ -33,15 +33,22 @@ pub(super) const ACP_FILE_SYSTEM_CAPABILITY: &str = "fs"; /// which are masked before the frame is forwarded to the container. pub(super) const ACP_TERMINAL_CAPABILITY: &str = "terminal"; +/// Decision produced by each iteration of the first-frame scanning loop. enum InitialFrameAction { + /// Keep reading; the complete first frame has not yet been buffered. Continue, + /// A complete frame was buffered; rewrite it before forwarding. ForwardMasked, + /// Forward the buffered bytes unchanged for the given reason. ForwardUnchanged(ForwardUnchangedReason), } +/// Reason why the first ACP frame is forwarded without capability rewriting. #[derive(Clone, Copy)] enum ForwardUnchangedReason { + /// The frame exceeded [`MAX_FIRST_FRAME_BYTES`] before a newline was found. ExceededMaximumSize, + /// Host stdin reached EOF before a newline-delimited frame was complete. EofBeforeNewline, } @@ -71,6 +78,7 @@ where } } +/// Read the next chunk into `first_frame` and decide how to proceed. async fn next_initial_frame_action( buffered_stdin: &mut tokio::io::BufReader, first_frame: &mut Vec, @@ -100,6 +108,7 @@ where }) } +/// Emit a debug trace for a frame forwarded without rewriting. fn log_unmodified_forwarding(reason: ForwardUnchangedReason, bytes: usize) { match reason { ForwardUnchangedReason::ExceededMaximumSize => log_exceeded_maximum_size(bytes), @@ -107,6 +116,7 @@ fn log_unmodified_forwarding(reason: ForwardUnchangedReason, bytes: usize) { } } +/// Emit a debug trace when the first frame exceeded the size limit. fn log_exceeded_maximum_size(bytes: usize) { tracing::debug!( bytes, @@ -114,6 +124,7 @@ fn log_exceeded_maximum_size(bytes: usize) { ); } +/// Emit a debug trace when EOF was reached before a newline was found. fn log_eof_before_newline(bytes: usize) { tracing::debug!( bytes, @@ -121,6 +132,8 @@ fn log_eof_before_newline(bytes: usize) { ); } +/// Write `first_frame` to `input` unchanged after logging the forwarding +/// reason. async fn forward_unmodified_initial_frame( input: &mut Pin>, first_frame: &[u8], @@ -130,6 +143,7 @@ async fn forward_unmodified_initial_frame( input.write_all(first_frame).await } +/// Mask `first_frame` and write the result to `input`. async fn forward_masked_initial_frame( input: &mut Pin>, first_frame: &[u8], @@ -140,12 +154,15 @@ async fn forward_masked_initial_frame( Ok(()) } +/// Emit a debug trace when the first frame was masked before forwarding. fn log_masked_frame_forwarded(masked_frame: &[u8], first_frame: &[u8]) { if masked_frame != first_frame { tracing::debug!("ACP initialize frame masked and forwarded"); } } +/// Fill the internal buffer and copy up to the size limit into `first_frame`. +/// Returns `(bytes_consumed, has_complete_frame)`. async fn read_next_bounded_frame_chunk( buffered_stdin: &mut tokio::io::BufReader, first_frame: &mut Vec, @@ -229,6 +246,9 @@ pub(super) fn split_frame_line_ending(frame: &[u8]) -> (&[u8], &[u8]) { (frame, b"") } +/// Remove `terminal` and `fs` from `params.clientCapabilities` in an ACP +/// `initialize` message. Returns `true` when at least one capability was +/// removed. fn remove_masked_acp_capabilities(message: &mut Value) -> bool { if message.get("method").and_then(Value::as_str) != Some(ACP_INITIALIZE_METHOD) { return false; diff --git a/src/engine/connection/exec/protocol.rs b/src/engine/connection/exec/protocol.rs index bdf7e462..602599e2 100644 --- a/src/engine/connection/exec/protocol.rs +++ b/src/engine/connection/exec/protocol.rs @@ -19,6 +19,43 @@ //! `handle_log_output_chunk` for `LogOutput::StdOut` and `LogOutput::Console` //! messages from the container. All other code paths route to stderr or return //! errors without touching stdout. +//! +//! ## Concurrency Model +//! +//! `run_protocol_session_with_io_async` runs two concurrent paths: +//! +//! 1. **Stdin forwarding task** — spawned via `spawn_stdin_forwarding_task`, +//! which takes ownership of the host-stdin reader and the container-input +//! writer. The task runs on the Tokio thread pool and is represented by a +//! `JoinHandle>`. Ownership of the handle is retained by the +//! caller function for the lifetime of the session. +//! +//! 2. **Output loop** — driven directly on the caller task via +//! `run_output_loop_async`. It polls the container output stream to +//! completion before any stdin shutdown logic runs. +//! +//! ### Task coordination +//! +//! The output loop is awaited to completion first. Once it returns (success or +//! error), `settle_stdin_forwarding_task` awaits the stdin handle with a +//! [`STDIN_SETTLE_TIMEOUT`] deadline of 50 ms. This grace period covers the +//! common case where container EOF propagates to the forwarding task within +//! normal pipe-flush timings. +//! +//! ### Timeout and cancellation +//! +//! If the grace period expires, `abort_stdin_forwarding_task` calls +//! `JoinHandle::abort` and immediately drops the handle **without awaiting +//! it**. Awaiting after abort would block indefinitely if host stdin is stalled +//! in a non-cancellable kernel read; dropping the handle mirrors the +//! attached-session teardown path and keeps shutdown bounded. +//! +//! A `JoinError` carrying `is_cancelled()` is mapped to `Ok(())` because an +//! explicit abort is an intentional shutdown signal, not a failure. +//! +//! When `disable_stdin_forwarding` is set, the forwarding task reads from a +//! `HeldOpenStdin` duplex adapter that never yields bytes and never closes. +//! In that mode a timeout is expected and is silently treated as success. use std::io; use std::pin::Pin; @@ -42,9 +79,13 @@ use crate::error::PodbotError; /// Host-side stdio handles used by the protocol byte proxy. pub(super) struct ProtocolProxyIo { + /// Host stdin reader supplied to the forwarding task. stdin: HostStdin, + /// Host stdout writer used for container stdout and console output. stdout: HostStdout, + /// Host stderr writer used for container stderr and error diagnostics. stderr: HostStderr, + /// Per-session configuration knobs applied to this proxy run. options: ProtocolSessionOptions, } @@ -63,13 +104,19 @@ const STDIN_SETTLE_TIMEOUT: Duration = Duration::from_millis(50); /// low while preventing unbounded accumulation during high-throughput scenarios. const STDIN_BUFFER_CAPACITY: usize = 65_536; +/// Per-session configuration knobs for protocol-mode exec proxy loops. #[derive(Debug, Clone, Copy, Default, PartialEq, Eq)] pub(super) struct ProtocolSessionOptions { + /// When `true`, host stdin is replaced by a held-open no-op reader so that + /// the process's inherited stdin is never forwarded to the container. disable_stdin_forwarding: bool, + /// When `true`, the first ACP `initialize` frame is rewritten to remove + /// `terminal` and `fs` capabilities before being forwarded to the container. rewrite_acp_initialize: bool, } impl ProtocolSessionOptions { + /// Create default protocol-session options with production behaviour. pub(super) const fn new() -> Self { Self { disable_stdin_forwarding: false, @@ -77,6 +124,7 @@ impl ProtocolSessionOptions { } } + /// Disable or enable protocol stdin forwarding for this session. pub(super) const fn with_stdin_forwarding_disabled(mut self, disable: bool) -> Self { self.disable_stdin_forwarding = disable; self @@ -89,16 +137,18 @@ impl ProtocolSessionOptions { } } +/// An `AsyncRead` adapter that stays open indefinitely without producing +/// bytes, used to suppress inherited process stdin while the forwarding task +/// remains alive. struct HeldOpenStdin { - // `tokio::io::duplex(1)` gives us the smallest possible in-memory pipe to - // keep `reader` and `_writer_guard` alive without producing an immediate - // EOF. No bytes are ever written through this seam; the 1-byte buffer just - // satisfies Tokio's duplex constructor while avoiding wasted memory. + /// The read half of the in-memory duplex pipe; never yields bytes. reader: tokio::io::DuplexStream, + /// Keeps the write half alive so `reader` never sees EOF. _writer_guard: tokio::io::DuplexStream, } impl tokio::io::AsyncRead for HeldOpenStdin { + /// Delegates to the inner duplex reader, which blocks indefinitely without producing bytes. fn poll_read( mut self: Pin<&mut Self>, cx: &mut Context<'_>, @@ -123,12 +173,15 @@ impl ProtocolProxyIo Self { self.options = options; self } } +/// Run a protocol exec session using real host stdio and the given session +/// options. pub(super) async fn run_protocol_session_async_with_options( request: &ExecRequest, output: Pin> + Send>>, @@ -144,6 +197,8 @@ pub(super) async fn run_protocol_session_async_with_options( run_protocol_session_with_io_async(request, output, input, stdio).await } +/// Return the appropriate stdin reader for a protocol session. Yields a +/// held-open no-op reader when stdin forwarding is disabled. fn protocol_host_stdin(options: ProtocolSessionOptions) -> Pin> { if options.disable_stdin_forwarding || stdin_forwarding_disabled_for_tests() { let (writer_guard, reader) = tokio::io::duplex(1); @@ -192,6 +247,8 @@ where stdin_result } +/// Wait for the stdin forwarding task to complete within a short grace +/// period. async fn settle_stdin_forwarding_task( container_id: &str, mut stdin_task: JoinHandle>, @@ -216,6 +273,7 @@ async fn settle_stdin_forwarding_task( classify_stdin_forwarding_task_result(container_id, join_result) } +/// Map a join result from the stdin forwarding task to a `PodbotError`. fn classify_stdin_forwarding_task_result( container_id: &str, join_result: Result, tokio::task::JoinError>, @@ -234,6 +292,7 @@ fn classify_stdin_forwarding_task_result( } } +/// Abort and drop the stdin forwarding task without awaiting it. fn abort_stdin_forwarding_task(stdin_task: JoinHandle>) { if !stdin_task.is_finished() { stdin_task.abort(); @@ -244,6 +303,8 @@ fn abort_stdin_forwarding_task(stdin_task: JoinHandle>) { } } +/// Copy host stdin to the container exec input, optionally rewriting the +/// first ACP `initialize` frame before the raw copy begins. async fn forward_host_stdin_to_exec_async( host_stdin: HostStdin, mut input: Pin>, @@ -267,6 +328,8 @@ where #[path = "protocol_acp_tests.rs"] mod acp_tests; +/// Drain the container output stream, routing each chunk to host stdout or +/// stderr. async fn run_output_loop_async( container_id: &str, output: &mut Pin> + Send>>, @@ -286,6 +349,7 @@ where Ok(()) } +/// Route a single container log-output chunk to the appropriate host stream. async fn handle_log_output_chunk( container_id: &str, chunk: LogOutput, @@ -307,6 +371,8 @@ where } } +/// Write and flush a byte slice to `writer`, mapping I/O failures to +/// `PodbotError`. async fn write_output_chunk( container_id: &str, writer: &mut Writer, diff --git a/src/engine/connection/exec/session.rs b/src/engine/connection/exec/session.rs index 71e5fa01..0ef1e966 100644 --- a/src/engine/connection/exec/session.rs +++ b/src/engine/connection/exec/session.rs @@ -6,7 +6,12 @@ use super::protocol::ProtocolSessionOptions; /// stream behaviour. #[derive(Debug, Clone, Copy, Default)] pub(crate) struct ExecSessionOptions { + /// When `true`, protocol-mode sessions replace host stdin with a held-open + /// no-op reader so that the process's inherited stdin is not forwarded. disable_protocol_stdin_forwarding: bool, + /// When `true`, the first ACP `initialize` frame sent from host stdin is + /// rewritten to remove `terminal` and `fs` capabilities before being + /// forwarded to the container. rewrite_acp_initialize: bool, } @@ -44,6 +49,8 @@ impl ExecSessionOptions { } } +/// Convert exec-session options into the lower-level [`ProtocolSessionOptions`] +/// consumed by the protocol proxy loop. pub(super) const fn protocol_session_options( options: ExecSessionOptions, ) -> ProtocolSessionOptions {