liminal-server 0.25.4

//! The WebSocket connection process's keepalive: the RFC 6455 Ping schedule
//! and its service each slice.

use super::WebSocketConnectionProcess;
use crate::server::connection::outbound::OutboundError;
use beamr::timer::TimerRef;
use std::time::{Duration, Instant};
use tungstenite::Message;

/// Q-A keepalive schedule: one transport-level Ping per interval per
/// connection. Pings are transport liveness ONLY — they never mint application
/// events, never re-arm application state, and never serve as a source of
/// truth; failure detection remains the socket's typed terminal events. Wakes
/// are timer-driven (`send_after` of the connection's `READY` atom), never
/// polled.
#[derive(Debug)]
pub(super) struct KeepaliveSchedule {
    pub(super) interval: Duration,
    pub(super) next_due: Instant,
    /// The armed one-shot wake timer and the deadline it was armed for.
    pub(super) armed: Option<(TimerRef, Instant)>,
}

impl KeepaliveSchedule {
    /// Builds the schedule with its first due instant. `None` when the
    /// interval overflows the monotonic clock — an extreme configured value
    /// must surface as a typed refusal, never an `Instant` addition panic
    /// (the S5 precedent).
    pub(super) fn new(interval: Duration) -> Option<Self> {
        Instant::now().checked_add(interval).map(|next_due| Self {
            interval,
            next_due,
            armed: None,
        })
    }
}

impl WebSocketConnectionProcess {
    /// Q-A: writes one transport-level Ping when the interval is due. Bounded
    /// to one ping per interval per connection; a transport-wedged buffer skips
    /// the ping (the next interval retries) rather than growing anything.
    pub(super) fn service_keepalive(&mut self, pid: u64) -> Result<(), OutboundError> {
        let Some(interval) = self.keepalive_interval else {
            return Ok(());
        };
        if self.keepalive.is_none() {
            let Some(schedule) = KeepaliveSchedule::new(interval) else {
                return Err(OutboundError::Write(std::io::Error::other(
                    "websocket.ping_interval_ms overflows the monotonic clock",
                )));
            };
            self.keepalive = Some(schedule);
        }
        let Some(keepalive) = self.keepalive.as_mut() else {
            return Ok(());
        };
        let now = Instant::now();
        if now < keepalive.next_due {
            return Ok(());
        }
        let Some(next_due) = now.checked_add(keepalive.interval) else {
            return Err(OutboundError::Write(std::io::Error::other(
                "websocket.ping_interval_ms overflows the monotonic clock",
            )));
        };
        keepalive.next_due = next_due;
        let Some(socket) = self.socket.as_mut() else {
            return Ok(());
        };
        match socket.write(Message::Ping(tungstenite::Bytes::new())) {
            Ok(()) => {
                self.outbound.note_transport_write_pending();
                Ok(())
            }
            Err(tungstenite::Error::Io(error))
                if error.kind() == std::io::ErrorKind::WouldBlock
                    || error.kind() == std::io::ErrorKind::Interrupted =>
            {
                self.outbound.note_transport_write_pending();
                Ok(())
            }
            Err(tungstenite::Error::WriteBufferFull(_)) => {
                tracing::debug!(
                    connection_pid = pid,
                    "keepalive ping skipped: transport write buffer is full"
                );
                Ok(())
            }
            Err(tungstenite::Error::Io(error)) => Err(OutboundError::Write(error)),
            Err(other) => Err(OutboundError::Write(std::io::Error::other(
                other.to_string(),
            ))),
        }
    }
}