conmonrs/streaming_server/

session.rs

//! What exec and attach sessions share: the connection to the client, its
//! input and the final status.

mod pending_stdin;

use pending_stdin::PendingStdin;

use super::{Session, StreamingServer};
use crate::{
    attach::SharedContainerAttach,
    container_io::SharedContainerIO,
    streaming_server::{
        idle::Activity,
        remotecommand::{
            ClientEvent, ClientInput, ExpectedStreams, Framing, MESSAGE_TOO_LARGE, Operation,
            Output, SpdyStreams, Status, Stdin, Version, WebSocketInput,
        },
        spdy::{self, Connection},
        web::{self, WebSocket},
    },
};
use anyhow::{Context, Result, bail};
use bytes::Bytes;
use futures::stream::StreamExt;
use serde::{
    Deserialize, Deserializer,
    de::{IgnoredAny, MapAccess, Visitor},
};
use std::{
    fmt,
    future::Future,
    ops::ControlFlow,
    sync::{
        Arc,
        atomic::{AtomicBool, Ordering},
    },
};
use tokio::{
    select,
    sync::{
        mpsc::{self, Receiver as MpscReceiver, Sender as MpscSender},
        watch,
    },
    task::{self, JoinHandle},
    time::{self, Duration},
};
use tokio_util::sync::CancellationToken;
use tracing::{Instrument, Span, debug, error, error_span, info, trace, warn};

/// How much stdin the read task holds while the container does not take it,
/// on top of the stdin channel. The socket is still read meanwhile, so that
/// resize events and the close do not wait behind it. The message which
/// fills it comes on top, also when it is larger than all of it.
pub(super) const PENDING_STDIN_MAX_BYTES: usize = 1024 * 1024;

/// How many entries of stdin wait for the session loop. Each is at most
/// [`STDIN_CHUNK_SIZE`](pending_stdin::STDIN_CHUNK_SIZE), more of them would
/// only hold more stdin which the container did not take.
const STDIN_QUEUE_SIZE: usize = 1;

/// How often a session pings the client, to notice a client which went away
/// also while its stdin waits for the container. A WebSocket session pings
/// only while stdin waits, a SPDY connection all the time.
pub(super) const PING_INTERVAL: Duration = if cfg!(test) {
    Duration::from_millis(100)
} else {
    spdy::PING_INTERVAL
};

/// How long the stdin which the client sent right before closing may wait for
/// the container to take a part of it, see [`ClientRx::stdin_taken`]. A slow
/// container gets all of it, only one which stopped reading its stdin holds
/// up the end of the session this long.
pub(super) const STDIN_FLUSH_TIMEOUT: Duration = if cfg!(test) {
    Duration::from_secs(2)
} else {
    Duration::from_secs(5)
};

/// The latest resize event of a client, older ones are replaced.
pub(super) type ResizeRx = watch::Receiver<Option<Bytes>>;

/// How the input of a client ended.
enum InputEnd<I> {
    /// The client closed the connection cleanly.
    Closed,

    /// The client went away.
    Gone,

    /// The client sent a message larger than the server takes, its input
    /// still has to drain the connection, see [`ClientInput::drain`].
    TooLarge(I),
}

/// How a single event of a client ended its input, see
/// [`StreamingServer::read_event`].
#[derive(Debug, PartialEq)]
enum EventEnd {
    Closed,
    TooLarge,
}

/// What the input of a client hands to the loop of its session.
pub(super) struct ClientRx {
    stdin: MpscReceiver<Stdin>,
    resize: ResizeRx,

    /// Cancelled once the client sent a message larger than the server
    /// takes, which fails the session.
    too_large: CancellationToken,

    /// Touched whenever the container took a part of the stdin, at the clean
    /// close of the client, and while the runtime starts an exec process
    /// with stdin.
    /// The stdin before the close gets forwarded until the container did not
    /// take any of it for [`STDIN_FLUSH_TIMEOUT`].
    stdin_taken: Activity,
}

#[cfg(test)]
impl ClientRx {
    /// The input of a client which sends nothing while the returned senders
    /// live.
    pub(super) fn idle() -> (Self, (MpscSender<Stdin>, watch::Sender<Option<Bytes>>)) {
        let (stdin_tx, stdin) = mpsc::channel(STDIN_QUEUE_SIZE);
        let (resize_tx, resize) = watch::channel(None);
        let client = Self {
            stdin,
            resize,
            too_large: CancellationToken::new(),
            stdin_taken: Activity::new(),
        };
        (client, (stdin_tx, resize_tx))
    }
}

/// Terminal resize event for exec and attach. Kubernetes decodes it with
/// Go, which matches the names of the fields regardless of their case, so
/// clients send `{"width":80}` as well as `{"Width":80}`, and leaves a
/// missing or null field at zero, and a null event all zero. For `Width` and
/// `Height`, ASCII case folding accepts the same names as the Unicode case
/// folding of Go, which only adds the Kelvin sign for `k` and the long s for
/// `s`.
#[derive(Debug, Default, PartialEq)]
struct ResizeEvent {
    width: u16,
    height: u16,
}

impl ResizeEvent {
    fn parse(data: &[u8]) -> Result<Self> {
        let event: Option<Self> = serde_json::from_slice(data).context("unmarshal resize event")?;
        Ok(event.unwrap_or_default())
    }
}

impl<'de> Deserialize<'de> for ResizeEvent {
    fn deserialize<D: Deserializer<'de>>(deserializer: D) -> Result<Self, D::Error> {
        /// The fields in the order of the document, like Go, where the last
        /// one of the names which match wins.
        struct Fields;

        impl<'de> Visitor<'de> for Fields {
            type Value = ResizeEvent;

            fn expecting(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
                f.write_str("a terminal size")
            }

            fn visit_map<A: MapAccess<'de>>(self, mut map: A) -> Result<ResizeEvent, A::Error> {
                let mut event = ResizeEvent::default();
                while let Some(name) = map.next_key::<String>()? {
                    let field = if name.eq_ignore_ascii_case("width") {
                        &mut event.width
                    } else if name.eq_ignore_ascii_case("height") {
                        &mut event.height
                    } else {
                        map.next_value::<IgnoredAny>()?;
                        continue;
                    };
                    if let Some(value) = map.next_value()? {
                        *field = value;
                    }
                }
                Ok(event)
            }
        }

        deserializer.deserialize_map(Fields)
    }
}

impl StreamingServer {
    /// Handle a single WebSocket connection, in the span of its session.
    pub(super) async fn handle_websocket(socket: WebSocket, session: Session, framing: Framing) {
        let span = Self::session_span(&session);
        Self::run_websocket(socket, session, framing)
            .instrument(span)
            .await;
    }

    /// Run the session of a WebSocket connection.
    async fn run_websocket(socket: WebSocket, session: Session, framing: Framing) {
        let discard = socket.get_ref().discarder();
        let written = socket.get_ref().written();
        let (sender, messages) = socket.split();
        let activity = Activity::new();
        let mut output = Output::websocket(
            sender,
            framing,
            activity.clone(),
            written.clone(),
            session.common().idle_timeout,
        );
        // A client which went away already ends the session in its loop,
        // which tears it down.
        if let Err(e) = output
            .announce(session.common().stdout, session.common().stderr)
            .await
        {
            debug!("Unable to announce the session: {e:#}");
        }
        let input = WebSocketInput {
            messages,
            framing,
            activity: activity.clone(),
            written,
            discard,
        };
        Self::run_session(output, input, session, activity).await;
        debug!("Closing WebSocket connection");
    }

    /// Handle a connection which got upgraded to SPDY, in the span of its
    /// session, which the tasks of the connection inherit: a protocol error
    /// of the client names the container as well.
    pub(super) async fn handle_spdy<T>(io: T, session: Session, version: Version)
    where
        T: tokio::io::AsyncRead + tokio::io::AsyncWrite + Send + 'static,
    {
        let span = Self::session_span(&session);
        Self::run_spdy(io, session, version).instrument(span).await;
    }

    /// Wait for the streams of the client of a SPDY connection, then run the
    /// session on them.
    async fn run_spdy<T>(io: T, session: Session, version: Version)
    where
        T: tokio::io::AsyncRead + tokio::io::AsyncWrite + Send + 'static,
    {
        let terminal = session.terminal();
        // A terminal merges stderr into stdout, Kubernetes clients do not
        // open a stream for it then.
        let expected = ExpectedStreams {
            stdin: session.common().stdin,
            stdout: session.common().stdout,
            stderr: session.common().stderr && !terminal,
        };
        // The session ends the connection once it was idle, which reports
        // the timeout to the client first.
        let mut connection = Connection::serve(io, None, Some(PING_INTERVAL), move |headers| {
            SpdyStreams::accept(headers, version)
        });
        let activity = connection.activity();
        let streams = match SpdyStreams::wait(&mut connection, expected).await {
            Ok(streams) => streams,
            Err(e) => {
                debug!("Unable to start the session: {e:#}");
                connection.handle().abort();
                return;
            }
        };
        let (input, output) = streams.into_parts(connection, version).await;
        let output = Output::spdy(output, session.common().idle_timeout);
        Self::run_session(output, input, session, activity).await;
        debug!("Closing SPDY connection");
    }

    /// The span of a session and of the tasks it starts, which shows the
    /// container ID at every log level, like the span of a request for a
    /// container does, see `enter_request` of the RPC interface.
    fn session_span(session: &Session) -> Span {
        error_span!("session", container_id = &*session.common().container_id)
    }

    /// Run a session until it ended or the client went away. `activity`
    /// tells when the connection was active the last time, for the idle
    /// timeout of the session.
    async fn run_session(
        output: Output,
        input: impl ClientInput,
        session: Session,
        activity: Activity,
    ) {
        let peer_closed = output.peer_closed();
        let (stdin_tx, stdin_rx) = mpsc::channel(STDIN_QUEUE_SIZE);
        // Resize events go through their own channel, so that they do not
        // queue behind stdin the container does not read. Only the latest
        // size matters.
        let (resize_tx, resize_rx) = watch::channel(None);
        let too_large = CancellationToken::new();
        let stdin_taken = Activity::new();
        let client = ClientRx {
            stdin: stdin_rx,
            resize: resize_rx,
            too_large: too_large.clone(),
            stdin_taken: stdin_taken.clone(),
        };

        let mut send_task = Self::write_task(output, client, session, activity.clone());
        let mut recv_task = Self::read_task(
            input,
            stdin_tx,
            resize_tx,
            peer_closed,
            too_large,
            stdin_taken.clone(),
        );

        tokio::select! {
            rv_a = (&mut send_task) => {
                let output = Self::sent(rv_a);
                // The client answers the close of a WebSocket, the read task
                // drops what it sends until then, see
                // [`web::await_close_answer`]. A close which did not go out
                // gets no answer. A SPDY connection ended already, which ends
                // the read task. A client which sent a message which was too
                // large, also one whose session failed for it, cannot answer
                // anymore, the connection gets drained instead.
                if let Some(output) = output {
                    match time::timeout(web::CLOSE_TIMEOUT, &mut recv_task).await {
                        Ok(Ok(InputEnd::TooLarge(input))) => input.drain(output).await,
                        Ok(_) => {}
                        Err(_) => debug!("Client did not answer the close in time"),
                    }
                }
                recv_task.abort();
            },
            rv_b = (&mut recv_task) => {
                match rv_b {
                    // After a clean close the session forwards the stdin which
                    // the client sent before, while the container takes it.
                    // The read task forwarded it since the close already, the
                    // wait goes on from the last part the container took.
                    Ok(InputEnd::Closed) => {
                        debug!("Client closed the connection cleanly, or after its stdin ended");
                        if stdin_taken
                            .until_stalled(stdin_taken.last(), STDIN_FLUSH_TIMEOUT, &mut send_task)
                            .await
                            .is_none()
                        {
                            debug!("Session did not end after the close, stopping it");
                        }
                    }
                    // The session fails, the client still gets its status and
                    // the close, which tells why, and what it still sends
                    // gets drained, so that they do not get lost. A client
                    // still busy with the rest of its message reads them only
                    // once it got rid of it, see [`web::Discard`]. The loop
                    // of the session ends right away, the status and the
                    // close end once the client stopped reading, see
                    // [`Output::close`].
                    Ok(InputEnd::TooLarge(input)) => {
                        info!("Client sent a message larger than the limit");
                        if let Ok(Some(output)) = (&mut send_task).await {
                            input.drain(output).await;
                        }
                    }
                    // The session ends a SPDY connection once it sent
                    // everything, which ends the input as well.
                    Ok(InputEnd::Gone) if send_task.is_finished() => {
                        Self::sent((&mut send_task).await);
                    }
                    Ok(InputEnd::Gone) => debug!("Client connection ended"),
                    Err(e) => error!("Unable to receive messages: {e}"),
                }
                send_task.abort();
            }
        }
    }

    /// Log how the write task ended, return its output once the close went
    /// out.
    fn sent(result: Result<Option<Output>, task::JoinError>) -> Option<Output> {
        match result {
            Ok(output) => {
                debug!("All messages sent");
                output
            }
            Err(e) => {
                error!("Unable to send messages: {e}");
                None
            }
        }
    }

    /// Build a common write task based on the session type. The loop of the
    /// session only decides its final status, which goes out after it, so
    /// that the idle timeout never cuts a status off: a client gets exactly
    /// one, or none if it went away. A session whose connection was idle for
    /// its timeout fails, like any other session which did not end, so that
    /// the client does not take it for a success, unless the process exited
    /// already. An exec process gets killed then, nothing is left to end it.
    /// The task returns the output once the close went out, see
    /// [`Output::close`].
    fn write_task(
        sender: Output,
        client: ClientRx,
        session: Session,
        activity: Activity,
    ) -> JoinHandle<Option<Output>> {
        let idle_timeout = session.common().idle_timeout;
        let too_large = client.too_large.clone();
        let task = async move {
            let mut sender = sender;
            let mut status = None;
            let (result, operation) = match session {
                Session::Exec(s) => (
                    async {
                        // The runtime runs in a task of its own. A client
                        // which goes away meanwhile must not kill it halfway,
                        // the process it started still gets registered, and
                        // torn down once the task finished. A client which
                        // sent too much fails the session meanwhile, it
                        // does not wait for its status. The terminal gets
                        // the size of the client meanwhile. The stdin which
                        // the client sent before a close waits for the
                        // process, the runtime does not count as a container
                        // which stopped taking it while it runs (it gets
                        // killed after its deadline). A clean close without
                        // pending stdin waits for the runtime as well then,
                        // the process gets torn down in order afterwards.
                        // Without stdin, nothing waits.
                        let stdin = s.common.stdin;
                        let start = Self::start_exec(*s, client.resize.clone());
                        let mut start = task::spawn(start.in_current_span());
                        let period = STDIN_FLUSH_TIMEOUT / 2;
                        let mut starting = time::interval_at(time::Instant::now() + period, period);
                        let started = loop {
                            select! {
                                started = &mut start => break started.context("join exec start")??,
                                () = too_large.cancelled() => bail!(MESSAGE_TOO_LARGE),
                                _ = starting.tick(), if stdin => client.stdin_taken.touch(),
                            }
                        };
                        // The client waited for the runtime, which is no
                        // idleness of its own, and the stdin gets its time
                        // from the start of the process.
                        activity.touch();
                        client.stdin_taken.touch();
                        let kill = started.teardown.kill.clone();
                        let exec = Self::exec_loop(started, &mut sender, client, &mut status);
                        let exec = Self::until_too_large(exec, &too_large);
                        Self::until_idle(exec, &activity, idle_timeout, || {
                            kill.store(true, Ordering::Release);
                        })
                        .await
                    }
                    .await,
                    Operation::Exec,
                ),
                Session::Attach(s) => (
                    Self::until_idle(
                        Self::until_too_large(
                            Self::attach_loop(s, &mut sender, client, &mut status),
                            &too_large,
                        ),
                        &activity,
                        idle_timeout,
                        || {},
                    )
                    .await,
                    Operation::Attach,
                ),
            };

            // The close tells clients why, also ones which do not read the
            // status. Their session failed, its end does not wait for the
            // idle timeout, see [`Output::close_too_large`].
            if too_large.is_cancelled() {
                sender.close_too_large();
            }
            if let Some(status) = Self::final_status(result, status, operation)
                && let Err(e) = sender.status(&status).await
            {
                debug!("Unable to send the status: {e:#}");
            }

            // Also for a message which came while the status went out.
            if too_large.is_cancelled() {
                sender.close_too_large();
            }
            sender.close().await.then_some(sender)
        };
        task::spawn(task.in_current_span())
    }

    /// The one status a client gets, out of how the loop of its session
    /// ended and the status the loop decided on, none for a client which
    /// went away.
    fn final_status(
        result: Result<()>,
        status: Option<Status>,
        operation: Operation,
    ) -> Option<Status> {
        match result {
            Ok(()) => status,
            // The process exited before, its status tells the client.
            Err(e) if status.is_some() => {
                debug!("Sending the exit status despite: {e:#}");
                status
            }
            // Otherwise the client sees a clean close and assumes success.
            // Mostly the client caused it: a command which the container
            // does not have, a container which is gone or a connection
            // which was idle. The span of the session names the container.
            Err(e) => {
                warn!("Unable to run {operation}: {e:#}");
                Some(Status::failed(operation, format_args!("{e:#}")))
            }
        }
    }

    /// Run the loop of a session until the client sent a message larger
    /// than the server takes, which fails it. The loop may wait for a client
    /// which does not read its output then, it gets dropped, so that the
    /// status and the close go out right away.
    async fn until_too_large(
        run: impl Future<Output = Result<()>>,
        too_large: &CancellationToken,
    ) -> Result<()> {
        select! {
            result = run => result,
            () = too_large.cancelled() => bail!(MESSAGE_TOO_LARGE),
        }
    }

    /// Run the loop of a session until the connection was idle for the
    /// timeout, which fails it. `on_idle` runs before the loop gets dropped.
    async fn until_idle(
        run: impl Future<Output = Result<()>>,
        activity: &Activity,
        timeout: Option<Duration>,
        on_idle: impl FnOnce(),
    ) -> Result<()> {
        if let Some(result) = activity.until_idle(timeout, run, on_idle).await {
            result
        } else {
            debug!("Closing idle connection");
            bail!(
                "no activity on the connection for the idle timeout of {:?}",
                timeout.unwrap_or_default()
            )
        }
    }

    /// Build a common read task. Every message of the client it reads counts
    /// as activity, also stdin. Once stdin backs up behind a process which
    /// does not take it, the client is not read anymore and the connection
    /// may go idle, like a spdystream connection whose reader waits. Its
    /// pings get no answer meanwhile either, a WebSocket client which fails
    /// without pongs, like the one of client-go after about a minute, ends
    /// the session once the process does not take stdin for that long. The
    /// pings sit behind the stdin in the socket, reading them would mean
    /// buffering the stdin without a bound.
    fn read_task<I>(
        mut input: I,
        stdin_tx: MpscSender<Stdin>,
        resize_tx: watch::Sender<Option<Bytes>>,
        peer_closed: Arc<AtomicBool>,
        too_large: CancellationToken,
        stdin_taken: Activity,
    ) -> JoinHandle<InputEnd<I>>
    where
        I: ClientInput,
    {
        let task = async move {
            let mut pending = PendingStdin::default();
            let end = loop {
                select! {
                    permit = stdin_tx.reserve(), if !pending.is_empty() => match permit {
                        Ok(permit) => permit.send(pending.pop()),
                        // The session ended, nothing takes stdin anymore.
                        Err(_) => pending.clear(),
                    },
                    // Once stdin got closed, nothing gets added anymore, so
                    // keep reading for the end of the connection.
                    event = input.next(), if !pending.is_full() || pending.closed => match event {
                        Some(event) => {
                            if let ControlFlow::Break(end) =
                                Self::read_event(event, &mut pending, &resize_tx)
                            {
                                break Some(end);
                            }
                        }
                        None => break None,
                    },
                }
            };
            match end {
                Some(EventEnd::Closed) => peer_closed.store(true, Ordering::Release),
                Some(EventEnd::TooLarge) => too_large.cancel(),
                None => {}
            }

            // Stdin sent right before a clean close, like the close of the
            // stream itself, still has to reach the container. Only while
            // the container takes it, a container which does not read it
            // must not delay the end of the session for long. A client which
            // went away or failed the session gets nothing forwarded
            // anymore. The client gets the answer to its close meanwhile,
            // also when the session gets stopped after.
            if end == Some(EventEnd::Closed) {
                stdin_taken.touch();
                let flush = async {
                    while !pending.is_empty() {
                        if stdin_tx.send(pending.pop()).await.is_err() {
                            break;
                        }
                    }
                };
                let flush =
                    stdin_taken.until_stalled(time::Instant::now(), STDIN_FLUSH_TIMEOUT, flush);
                if tokio::join!(flush, input.answer_close()).0.is_none() {
                    debug!("Dropping stdin the container did not take after the close");
                }
            }
            match end {
                Some(EventEnd::Closed) => InputEnd::Closed,
                Some(EventEnd::TooLarge) => InputEnd::TooLarge(input),
                None => InputEnd::Gone,
            }
        };
        task::spawn(task.in_current_span())
    }

    /// Handle a single event of the client, break once its input ended.
    fn read_event(
        event: ClientEvent,
        pending: &mut PendingStdin,
        resize_tx: &watch::Sender<Option<Bytes>>,
    ) -> ControlFlow<EventEnd> {
        match event {
            ClientEvent::Stdin(stdin) => pending.push(stdin),
            ClientEvent::Resize(data) => {
                resize_tx.send_replace(Some(data));
            }
            ClientEvent::Closed => return ControlFlow::Break(EventEnd::Closed),
            ClientEvent::TooLarge => return ControlFlow::Break(EventEnd::TooLarge),
        }
        ControlFlow::Continue(())
    }
}

/// Forwards the stdin and the resize events of a client to the process, for
/// the loops of exec and attach sessions, which forward the output
/// meanwhile.
pub(super) struct SessionInput {
    io: SharedContainerIO,
    attach: SharedContainerAttach,
    client: ClientRx,

    /// Stdin which waits for the container.
    pending: Option<Bytes>,

    /// Pings the client while stdin waits.
    ping: time::Interval,

    /// Whether stdin gets forwarded, until the client closed it.
    pub(super) stdin_open: bool,
}

/// What the loop of a session has to handle.
#[derive(Debug, PartialEq)]
pub(super) enum InputEvent {
    /// A resize event of the client, see [`SessionInput::resize`]. Resizing
    /// does not depend on stdin, a terminal without it still has a size.
    Resize(Bytes),

    /// The stdin of the client is not read while it waits for the container,
    /// so a client which goes away meanwhile is only noticed by a failing
    /// send. The session pings it.
    Ping,

    /// The client closed its stdin.
    StdinClosed,

    /// The client closed the connection.
    Ended,

    /// The client sent a message larger than the server takes, which fails
    /// the session.
    TooLarge,
}

impl SessionInput {
    /// Forward the input of `client` to the process of `io` and `attach`,
    /// stdin only if `stdin` is set.
    pub(super) fn new(
        io: SharedContainerIO,
        attach: SharedContainerAttach,
        client: ClientRx,
        stdin: bool,
    ) -> Self {
        let mut ping = time::interval_at(time::Instant::now() + PING_INTERVAL, PING_INTERVAL);
        ping.set_missed_tick_behavior(time::MissedTickBehavior::Delay);
        Self {
            io,
            attach,
            client,
            pending: None,
            ping,
            stdin_open: stdin,
        }
    }

    /// Forward stdin until the session has something to handle. The session
    /// may cancel it for its output, nothing gets lost then.
    pub(super) async fn next(&mut self) -> InputEvent {
        loop {
            select! {
                () = self.client.too_large.cancelled() => return InputEvent::TooLarge,
                Ok(()) = self.client.resize.changed() => {
                    if let Some(data) = self.client.resize.borrow_and_update().clone() {
                        return InputEvent::Resize(data);
                    }
                }
                // Wait for the container to take the stdin without blocking
                // the output meanwhile.
                () = self.attach.send_stdin(self.pending.clone().unwrap_or_default()), if self.pending.is_some() => {
                    self.pending = None;
                    self.client.stdin_taken.touch();
                }
                _ = self.ping.tick(), if self.pending.is_some() => return InputEvent::Ping,
                stdin = self.client.stdin.recv(), if self.pending.is_none() => match stdin {
                    // The input fails the session before it ends.
                    None if self.client.too_large.is_cancelled() => return InputEvent::TooLarge,
                    None => return InputEvent::Ended,
                    Some(_) if !self.stdin_open => {}
                    Some(Stdin::Data(data)) => self.pending = Some(data),
                    Some(Stdin::Close) => {
                        self.stdin_open = false;
                        return InputEvent::StdinClosed;
                    }
                },
            }
        }
    }

    /// Close the stdin of the container after the stdin before. A terminal
    /// cannot be closed without ending its output, it gets its end of file
    /// character instead, like from a user pressing Ctrl-D. A process
    /// reading lines sees the end of its input then, one after a line
    /// without its ending needs a second one, like a user would type it.
    pub(super) fn close_container_stdin(&mut self) {
        // Right after `next` returned `StdinClosed`, which it only does while
        // no stdin waits, so the end of file character replaces nothing.
        debug_assert!(self.pending.is_none(), "stdin waits for the container");
        match self.io.terminal_eof() {
            Some(eof) => self.pending = Some(Bytes::from(vec![eof])),
            None => {
                self.attach.close_container_stdin();
            }
        }
    }

    /// Handle a resize event of the client, its JSON.
    pub(super) fn resize(&self, data: &[u8]) {
        Self::resize_with(data, |width, height| self.io.try_resize(width, height));
    }

    /// Handle a resize event of the client, its JSON, with `resize`, which
    /// returns whether there is a terminal. Clients send them regardless of
    /// a terminal, a broken one does not end the session.
    pub(super) fn resize_with(data: &[u8], resize: impl FnOnce(u16, u16) -> Result<bool>) {
        if let Err(e) = Self::try_resize(data, resize) {
            debug!("Ignoring resize event: {e:#}");
        }
    }

    fn try_resize(data: &[u8], resize: impl FnOnce(u16, u16) -> Result<bool>) -> Result<()> {
        let e = ResizeEvent::parse(data)?;
        trace!("Got resize message: {e:?}");
        // Like CRI-O, a terminal never gets a size without rows or columns,
        // which clients send before they know their own.
        if e.width == 0 || e.height == 0 {
            debug!("Ignoring resize event without a size: {e:?}");
            return Ok(());
        }
        if !resize(e.width, e.height)? {
            debug!("Ignoring resize event, container has no terminal");
        }
        Ok(())
    }
}

/// The entry points of the fuzz targets.
#[cfg(any(test, feature = "fuzzing"))]
pub(super) mod fuzzing;

#[cfg(test)]
mod tests;