liminal-server 0.2.4

//! Per-connection outbound byte buffer with cooperative, partial-write draining.
//!
//! Every server-originated frame (acks, errors, `Push`, `Disconnect`, `Pong`,
//! `Deliver`, ...) is encoded into this bounded buffer and drained on the
//! connection's scheduler slice. This replaces the previous direct `write_all`
//! on the connection's NON-BLOCKING socket (ledger G4): `write_all` on a
//! non-blocking stream fails the instant the kernel send buffer fills, which a
//! frame larger than the socket buffer (~64 KiB) hits mid-write, corrupting or
//! dropping the stream. Here a [`std::io::Write::write`] loop tracks partial
//! progress and a `WouldBlock` mid-frame simply leaves the residue queued for the
//! next slice — a queued frame streams out across as many slices as the peer's read
//! rate requires, so the socket send buffer is no longer the size ceiling.
//!
//! The buffer is bounded (default [`DEFAULT_OUTBOUND_CAPACITY`]): an enqueue that
//! would exceed the cap, or an unrecoverable write error, is a fatal condition —
//! a desynced stream must never survive, so the caller tears the connection down.
//! The honest delivery bound is therefore the buffer capacity, not the socket send
//! buffer: any single frame up to [`DEFAULT_OUTBOUND_CAPACITY`] is delivered intact
//! across as many slices as needed, and the delivery pump uses [`OutboundWriter::
//! has_room`] to hold a burst's residue back for a later slice rather than overflow.
//! A single frame *larger than the whole buffer* can never be queued and remains a
//! fatal overflow that tears the connection down (the spec-inherent per-frame bound).

use std::collections::VecDeque;
use std::io::Write;
use std::net::TcpStream;

use liminal::protocol::{Frame, ProtocolError, encode, encoded_len};

/// Default cap on a single connection's outbound byte buffer (4 MiB).
pub(super) const DEFAULT_OUTBOUND_CAPACITY: usize = 4 * 1024 * 1024;

/// A fatal condition on the outbound path: either the bounded buffer overflowed
/// or the socket reported an unrecoverable write error. Both tear the connection
/// down; a partially written, desynced stream is never allowed to survive.
#[derive(Debug)]
pub(super) enum OutboundError {
    /// Enqueuing the frame would push the buffer past its capacity.
    Overflow {
        /// Bytes already queued and not yet drained.
        queued: usize,
        /// Bytes the rejected frame would have added.
        needed: usize,
        /// The buffer's fixed capacity.
        capacity: usize,
    },
    /// The frame could not be encoded into wire bytes.
    Encode(ProtocolError),
    /// The socket reported an unrecoverable write error while draining.
    Write(std::io::Error),
}

impl std::fmt::Display for OutboundError {
    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            Self::Overflow {
                queued,
                needed,
                capacity,
            } => write!(
                formatter,
                "outbound buffer overflow: {queued} queued + {needed} needed exceeds \
                 capacity {capacity}"
            ),
            Self::Encode(error) => write!(formatter, "outbound frame encode failed: {error}"),
            Self::Write(error) => write!(formatter, "outbound socket write failed: {error}"),
        }
    }
}

impl std::error::Error for OutboundError {}

/// The tri-state result of one [`OutboundWriter::drain`] (R2, §1.2(1)).
///
/// The distinction is dark under the current busy loop — every caller re-services
/// the connection on the next scheduler slice regardless — but it is the exact
/// signal the park-flip consumes: writable readiness interest is armed **iff**
/// [`Self::WouldBlockWithResidue`], and only then. `Drained` and `Progress` never
/// arm writable interest (there is nothing to wait for, or the socket last
/// accepted a write so it is presumed writable and the next slice re-services).
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(super) enum DrainOutcome {
    /// The buffer was fully emptied; no residue is queued.
    Drained,
    /// Partial progress: the per-drain byte budget was exhausted with residue
    /// still queued, and the last socket write SUCCEEDED — so the socket is, as
    /// far as this drain observed, still writable. Reachable only when a caller
    /// passes an explicit budget; the busy-loop live path passes `None` and so
    /// never yields this. The park-flip does NOT arm writable interest here — it
    /// re-services on the next slice — but the state is distinct so the park-flip
    /// can bound per-slice outbound work without conflating it with a full send
    /// buffer.
    Progress,
    /// A socket write returned `WouldBlock` with residue still queued: the kernel
    /// send buffer is full. This is the ONLY state on which the park-flip arms
    /// writable readiness interest, so the connection re-wakes when the send
    /// buffer drains rather than busy-polling.
    WouldBlockWithResidue,
}

/// A bounded, per-connection outbound byte queue drained with partial-write
/// tracking on the connection's scheduler slice.
#[derive(Debug)]
pub(super) struct OutboundWriter {
    buffer: VecDeque<u8>,
    capacity: usize,
}

impl OutboundWriter {
    /// Creates an outbound writer with the default 4 MiB capacity.
    pub(super) const fn new() -> Self {
        Self::with_capacity(DEFAULT_OUTBOUND_CAPACITY)
    }

    /// Creates an outbound writer with an explicit capacity (used by tests to
    /// force overflow with a small cap).
    pub(super) const fn with_capacity(capacity: usize) -> Self {
        Self {
            buffer: VecDeque::new(),
            capacity,
        }
    }

    /// Encodes `frame` into the buffer, rejecting it when the buffer would exceed
    /// its capacity.
    ///
    /// # Errors
    /// Returns [`OutboundError::Overflow`] when the frame does not fit within the
    /// remaining capacity and [`OutboundError::Encode`] when the frame cannot be
    /// encoded. Either is fatal for the connection.
    pub(super) fn enqueue_frame(&mut self, frame: &Frame) -> Result<(), OutboundError> {
        let needed = encoded_len(frame).map_err(OutboundError::Encode)?;
        let queued = self.buffer.len();
        let projected = queued.checked_add(needed).ok_or(OutboundError::Overflow {
            queued,
            needed,
            capacity: self.capacity,
        })?;
        if projected > self.capacity {
            return Err(OutboundError::Overflow {
                queued,
                needed,
                capacity: self.capacity,
            });
        }
        let mut bytes = vec![0_u8; needed];
        let written = encode(frame, &mut bytes).map_err(OutboundError::Encode)?;
        // `written` never exceeds `needed` (encode fills the sized buffer), so
        // truncating to the reported count is a safe no-op in the normal case and
        // a defensive guard against a short write otherwise.
        bytes.truncate(written);
        self.buffer.extend(bytes);
        Ok(())
    }

    /// The buffer's fixed capacity in bytes.
    pub(super) const fn capacity(&self) -> usize {
        self.capacity
    }

    /// Whether `needed` more bytes fit within the remaining capacity.
    ///
    /// The delivery pump peeks this before enqueuing the next `Deliver` frame: when
    /// a frame that *would* fit an empty buffer does not fit the current free space,
    /// the pump holds it back for a later slice (after the drain frees room) instead
    /// of overflowing — so a pipelined burst never tears down a healthy fast reader.
    pub(super) fn has_room(&self, needed: usize) -> bool {
        self.buffer
            .len()
            .checked_add(needed)
            .is_some_and(|projected| projected <= self.capacity)
    }

    /// Drains queued bytes to `stream`, reporting the [`DrainOutcome`] tri-state.
    ///
    /// Writes proceed from the front of the queue with a `write()` loop that
    /// tracks partial progress; a `WouldBlock` (the non-blocking socket's send
    /// buffer is full) leaves the residue queued for the next slice and returns
    /// [`DrainOutcome::WouldBlockWithResidue`]. `Interrupted` retries. Any other
    /// error — or a zero-length write, which means the peer is gone — is fatal.
    ///
    /// `budget` optionally bounds how many bytes this single drain writes: once
    /// that many bytes have been written and residue still remains, the drain
    /// stops and returns [`DrainOutcome::Progress`] (the last write succeeded, so
    /// the socket is presumed writable). The live busy-loop path passes `None` —
    /// byte-for-byte identical to the pre-tri-state behaviour, yielding only
    /// `Drained` or `WouldBlockWithResidue`. The budget seam exists so the
    /// park-flip can bound per-slice outbound work; it is NOT wired live here to
    /// avoid a silent change to per-slice throughput (only the return type is
    /// enriched now).
    ///
    /// # Errors
    /// Returns [`OutboundError::Write`] on an unrecoverable socket write error.
    pub(super) fn drain(
        &mut self,
        stream: &mut TcpStream,
        budget: Option<usize>,
    ) -> Result<DrainOutcome, OutboundError> {
        let mut written_total: usize = 0;
        while !self.buffer.is_empty() {
            // Stop on the byte budget BEFORE the next write: the budget bounds
            // bytes written this drain, and residue remaining means the caller
            // (park-flip) re-services next slice — the socket last accepted a
            // write, so it is presumed writable (`Progress`, not a WouldBlock).
            let remaining_budget = match budget {
                Some(limit) if written_total >= limit => return Ok(DrainOutcome::Progress),
                Some(limit) => Some(limit - written_total),
                None => None,
            };
            // `as_slices().0` is the front contiguous run; a wrapped queue drains in
            // two passes across loop iterations, so no reallocation is forced. When a
            // budget is set, the write is capped to the remaining budget so the drain
            // stops after exactly `budget` bytes rather than after the first
            // whole-buffer-accepting write (which would never observe the budget).
            let front = self.buffer.as_slices().0;
            let front = remaining_budget.map_or(front, |limit| {
                front.get(..limit.min(front.len())).unwrap_or(front)
            });
            match stream.write(front) {
                Ok(0) => {
                    return Err(OutboundError::Write(std::io::Error::new(
                        std::io::ErrorKind::WriteZero,
                        "connection peer accepted zero bytes",
                    )));
                }
                Ok(written) => {
                    self.buffer.drain(..written);
                    written_total = written_total.saturating_add(written);
                }
                Err(error) if error.kind() == std::io::ErrorKind::WouldBlock => {
                    return Ok(DrainOutcome::WouldBlockWithResidue);
                }
                Err(error) if error.kind() == std::io::ErrorKind::Interrupted => {}
                Err(error) => return Err(OutboundError::Write(error)),
            }
        }
        Ok(DrainOutcome::Drained)
    }

    /// Number of bytes currently queued and not yet drained.
    #[cfg(test)]
    pub(super) fn queued_len(&self) -> usize {
        self.buffer.len()
    }

    /// Removes and returns all queued bytes, so a test can decode what would have
    /// been written without going through a socket.
    #[cfg(test)]
    pub(super) fn take_bytes(&mut self) -> Vec<u8> {
        self.buffer.drain(..).collect()
    }
}

#[cfg(test)]
#[path = "outbound_tests.rs"]
mod tests;