liminal-server 0.25.4

//! The WebSocket connection process's control vocabulary, revocation, server
//! pushes and mailbox messages.

use super::{PUSH_STREAM_ID, SliceStep, WebSocketConnectionProcess};
use crate::server::connection::outbound::DrainOutcome;
use crate::server::connection::supervisor::runtime::ConnectionControl;
use crate::server::participant::ConnectionFateClass;
use beamr::atom::Atom;
use beamr::native::native_process::NativeOutcome;
use beamr::process::ExitReason;
use beamr::term::Term;
use liminal::protocol::Frame;
use tungstenite::protocol::frame::CloseFrame;
use tungstenite::protocol::frame::coding::CloseCode;

impl WebSocketConnectionProcess {
    /// The TCP process's control handling against the WebSocket transport.
    pub(super) fn handle_control(
        &mut self,
        pid: u64,
        control: ConnectionControl,
    ) -> Option<NativeOutcome> {
        match control {
            ConnectionControl::NotifyShutdown => {
                self.notify_shutdown(pid, true);
                None
            }
            ConnectionControl::ForceClose => {
                self.notify_shutdown(pid, false);
                // A failed drain loses the enqueued `Disconnect`, and the client
                // learns of the shutdown from the transport instead of the frame.
                // Accepted: the force-close decision is taken, and a retry or
                // sleep here would hold the whole shutdown drain hostage to one
                // unresponsive peer.
                if let Err(error) = self.drain_outbound(None) {
                    tracing::debug!(connection_pid = pid, %error, "the close path's best-effort drain did not finish; the peer's half-close ends it");
                }
                if let Err(error) =
                    self.complete_connection_fate(ConnectionFateClass::ServerShutdown)
                {
                    return Some(match self.fail_fate(pid, &error) {
                        SliceStep::Continue => NativeOutcome::Continue,
                        SliceStep::Stop(reason) => NativeOutcome::Stop(reason),
                    });
                }
                if let Some(socket) = self.socket.as_mut() {
                    // Both errors are only debug-logged, as on the normal-close
                    // path: at server shutdown the only thing
                    // lost is the courtesy `Away` close handshake, and the peer
                    // reads the same shutdown from the dropped socket a moment
                    // later. Nothing on this path may sleep or retry to complete
                    // the handshake.
                    if let Err(error) = socket.close(Some(CloseFrame {
                        code: CloseCode::Away,
                        reason: "server shutdown".into(),
                    })) {
                        tracing::debug!(connection_pid = pid, %error, "the websocket close frame was not sent; the dropped socket ends it");
                    }
                    if let Err(error) = socket.flush() {
                        tracing::debug!(connection_pid = pid, %error, "the websocket close flush did not finish; the dropped socket ends it");
                    }
                }
                self.release_conversations();
                self.runtime.finish(pid);
                self.socket.take();
                Some(NativeOutcome::Stop(ExitReason::Normal))
            }
            ConnectionControl::Push {
                correlation_id,
                payload,
            } => {
                self.write_push(pid, correlation_id, payload);
                None
            }
            ConnectionControl::Revocation => self.revocation_sweep(pid),
        }
    }

    /// The TCP process's revocation sweep against the WebSocket transport
    /// (ACCESS-005 R4/R6).
    pub(super) fn revocation_sweep(&mut self, pid: u64) -> Option<NativeOutcome> {
        let close = self
            .state
            .revocation
            .as_ref()
            .and_then(crate::auth_revocation::ConnectionGrants::close_pending);
        if let Some(reason) = close {
            return Some(self.revocation_close(pid, reason));
        }
        for frame in
            super::super::super::apply::retire_withdrawn_rights(&self.runtime, &mut self.state)
        {
            if let Err(error) = self.outbound.enqueue_frame(&frame) {
                tracing::warn!(
                    connection_pid = pid,
                    %error,
                    "a withdrawn right's refusal could not be queued"
                );
            }
        }
        None
    }

    /// The TCP process's revocation close against the WebSocket transport
    /// (ACCESS-005 R5, R6): one attempt at the reason-bearing close, recorded
    /// as attempted, then the forced-close fate, cleanup and removal funnel.
    pub(super) fn revocation_close(
        &mut self,
        pid: u64,
        reason: crate::auth_revocation::RevocationRefusal,
    ) -> NativeOutcome {
        use crate::auth_revocation::CloseWrite;
        let write = match reason.close_reason() {
            Some(close) if self.socket.is_some() => {
                match self.outbound.enqueue_frame(&Frame::DisconnectReason {
                    flags: 0,
                    reason: close,
                }) {
                    Ok(()) => match self.drain_outbound(None) {
                        Ok(DrainOutcome::Drained) => CloseWrite::Flushed,
                        Ok(DrainOutcome::Progress | DrainOutcome::WouldBlockWithResidue) => {
                            CloseWrite::Residue
                        }
                        Err(_) => CloseWrite::TransportFailed,
                    },
                    Err(_) => CloseWrite::NotQueued,
                }
            }
            Some(_) => CloseWrite::NoStream,
            None => CloseWrite::NotQueued,
        };
        if let Some(authority) = self.runtime.revocation() {
            authority.record_close_write(pid, write);
        }
        tracing::info!(
            connection_pid = pid,
            reason = reason.name(),
            write = ?write,
            "closing websocket connection: a grant revocation withdrew the right that admitted it"
        );
        if let Err(error) = self.complete_connection_fate(ConnectionFateClass::ServerShutdown) {
            return match self.fail_fate(pid, &error) {
                SliceStep::Continue => NativeOutcome::Continue,
                SliceStep::Stop(reason) => NativeOutcome::Stop(reason),
            };
        }
        if let Some(socket) = self.socket.as_mut() {
            // The reason travelled in the liminal frame above; the WebSocket
            // close handshake is a courtesy whose loss changes nothing, and
            // is named when it happens.
            if let Err(error) = socket.close(Some(CloseFrame {
                code: CloseCode::Policy,
                reason: reason.name().into(),
            })) {
                tracing::debug!(%error, "websocket close frame not sent");
            }
            if let Err(error) = socket.flush() {
                tracing::debug!(%error, "websocket close flush failed");
            }
        }
        self.release_conversations();
        self.runtime.finish(pid);
        self.socket.take();
        NativeOutcome::Stop(ExitReason::Normal)
    }

    /// The TCP process's push writer against the message-framed queue: a missing
    /// socket or an encode failure cancels the push slot so the awaiter resolves
    /// promptly instead of hanging. An enqueue failure is settled by
    /// [`ConnectionRuntime::settle_push_enqueue_failure`](crate::server::connection::supervisor::runtime::ConnectionRuntime::settle_push_enqueue_failure), which names the one
    /// certain case — a frame larger than the whole outbound buffer — and
    /// cancels every recoverable one exactly as before.
    pub(super) fn write_push(&mut self, pid: u64, correlation_id: u64, payload: Vec<u8>) {
        if self.socket.is_none() {
            tracing::warn!(
                connection_pid = pid,
                correlation_id,
                "server push skipped because connection socket is unavailable"
            );
            self.runtime.cancel_push(correlation_id);
            return;
        }
        let frame = match Frame::new_push(PUSH_STREAM_ID, correlation_id, payload) {
            Ok(frame) => frame,
            Err(error) => {
                tracing::warn!(
                    connection_pid = pid,
                    correlation_id,
                    %error,
                    "server push frame could not be constructed"
                );
                self.runtime.cancel_push(correlation_id);
                return;
            }
        };
        match self.outbound.enqueue_frame(&frame) {
            Ok(()) => self.runtime.record_push_queued(),
            Err(error) => self
                .runtime
                .settle_push_enqueue_failure(pid, correlation_id, &error),
        }
    }

    /// The TCP process's shutdown notification, byte-for-byte: the liminal
    /// `Disconnect` frame rides as an ordinary canonical binary message.
    pub(super) fn notify_shutdown(&mut self, pid: u64, subscribers_only: bool) {
        if self.state.shutdown_notification_attempted {
            return;
        }
        if subscribers_only && self.state.subscriptions.is_empty() {
            return;
        }
        self.state.shutdown_notification_attempted = true;
        if self.socket.is_none() {
            tracing::warn!(
                connection_pid = pid,
                peer_addr = ?self.peer_addr,
                "shutdown notification skipped because connection socket is unavailable"
            );
            return;
        }
        match self.outbound.enqueue_frame(&Frame::Disconnect { flags: 0 }) {
            Ok(()) => {
                tracing::debug!(
                    connection_pid = pid,
                    peer_addr = ?self.peer_addr,
                    subscriber_count = self.state.subscriptions.len(),
                    "enqueued shutdown notification to connection"
                );
            }
            Err(error) => {
                tracing::warn!(
                    connection_pid = pid,
                    peer_addr = ?self.peer_addr,
                    %error,
                    "shutdown notification could not be enqueued; connection will not be retried"
                );
            }
        }
    }

    /// The TCP process's mailbox handling: crash requests, control drains, and
    /// the bare `READY` wake vocabulary, unchanged.
    pub(super) fn handle_message(&mut self, pid: u64, message: Term) -> Option<NativeOutcome> {
        if message == Term::atom(Atom::ERROR) {
            self.close_transport_abnormal();
            self.release_conversations();
            self.runtime
                .mark_crashed(pid, ExitReason::Error, self.peer_addr);
            return Some(NativeOutcome::Stop(ExitReason::Error));
        }
        if message.as_atom() == Some(self.runtime.control_atom()) {
            while let Some(control) = self.runtime.pop_control(pid) {
                if let Some(outcome) = self.handle_control(pid, control) {
                    return Some(outcome);
                }
            }
        }
        None
    }
}