liminal-server 0.24.0

//! 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 by the operator's
//! [`max_connection_outbound_bytes`](crate::config::types::LimitsConfig::max_connection_outbound_bytes):
//! 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 that 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 liminal::protocol::{Frame, ProtocolError, encode, encoded_len};

/// 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 {
    /// A far call could not release its authenticated answer.
    LinkCalls(String),
    /// 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::LinkCalls(message) => write!(formatter, "far call release failed: {message}"),
            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 {}

/// Proof that one frame can NEVER be carried on a connection at its current
/// bound: the encoded frame is larger than the whole outbound buffer, so no
/// amount of draining could make room for it.
///
/// Deliberately NOT produced for a frame that would fit an empty buffer but not
/// the current free space — that is congestion, and waiting fixes it.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(super) struct FrameExceedsCapacity {
    /// Encoded size of the rejected frame; exceeds `capacity` on its own.
    pub(super) needed: usize,
    /// The outbound buffer's whole capacity.
    pub(super) capacity: usize,
    /// Bytes queued when the frame was rejected (context for the log only — the
    /// judgement is `needed` against `capacity` alone).
    pub(super) queued: usize,
}

impl OutboundError {
    /// `Some` when this failure proves the frame exceeds the WHOLE capacity, and
    /// so can never be carried; `None` for congestion, an encode failure, or a
    /// socket write error, each of which says nothing about the frame's size.
    ///
    /// This is the single place the certain case is distinguished from the
    /// recoverable one. Widening it would let a busy moment read as a permanent
    /// failure, so it is pinned by its own test.
    pub(super) const fn frame_exceeds_capacity(&self) -> Option<FrameExceedsCapacity> {
        match *self {
            Self::Overflow {
                queued,
                needed,
                capacity,
            } if needed > capacity => Some(FrameExceedsCapacity {
                needed,
                capacity,
                queued,
            }),
            _ => None,
        }
    }
}

/// 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,
    /// The kind of every frame enqueued since the last
    /// [`Self::take_enqueued`]. Test-only (R3b F8, ruling 2a).
    #[cfg(test)]
    enqueued: Vec<liminal::protocol::FrameType>,
}

impl OutboundWriter {
    /// Creates an outbound writer bounded by `capacity` bytes.
    ///
    /// There is deliberately NO default constructor. The bound is an operator
    /// setting
    /// ([`max_connection_outbound_bytes`](crate::config::types::LimitsConfig::max_connection_outbound_bytes)),
    /// and a writer that could mint its own 4 MiB silently is how this bound came
    /// to be invisible to operators in the first place: it was met only as a
    /// connection that had stopped carrying something. Every caller now says what
    /// the bound is, and the standing value comes from config.
    pub(super) const fn with_capacity(capacity: usize) -> Self {
        Self {
            buffer: VecDeque::new(),
            capacity,
            #[cfg(test)]
            enqueued: Vec::new(),
        }
    }

    /// 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);
        #[cfg(test)]
        self.enqueued.push(frame.frame_type());
        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 `sink`, reporting the [`DrainOutcome`] tri-state.
    ///
    /// The sink is any [`std::io::Write`]: the live path passes the connection's
    /// non-blocking `TcpStream`, and a transport with no socket passes its own
    /// write half. Nothing below reads a socket fact — the byte budget, the
    /// partial-write tracking and the error classification are transport-neutral —
    /// so one writer serves every transport instead of being copied per sink.
    ///
    /// 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,
        sink: &mut dyn Write,
        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 sink 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 sink.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()
    }

    /// The kinds of the frames enqueued since the last call. Taken once the
    /// buffer has drained, they are the frames written. Test-only (R3b F8,
    /// ruling 2a).
    #[cfg(test)]
    pub(super) fn take_enqueued(&mut self) -> Vec<liminal::protocol::FrameType> {
        std::mem::take(&mut self.enqueued)
    }

    /// 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;