liminal-server 0.25.4

//! The WebSocket connection process's close and fate paths: a crash, a fate's
//! close, a normal close and the abnormal transport close.

use super::{SliceStep, WebSocketConnectionProcess};
use crate::ServerError;
use crate::server::participant::ConnectionFateClass;
use beamr::native::native_process::NativeOutcome;
use beamr::process::ExitReason;
use tungstenite::protocol::frame::CloseFrame;
use tungstenite::protocol::frame::coding::CloseCode;

impl WebSocketConnectionProcess {
    /// Terminal crash path: release resources, record the crash, stop.
    pub(super) fn stop_crashed(&mut self, pid: u64) -> NativeOutcome {
        self.close_transport_abnormal();
        self.release_conversations();
        self.runtime
            .mark_crashed(pid, ExitReason::Error, self.peer_addr);
        NativeOutcome::Stop(ExitReason::Error)
    }

    pub(super) fn fail_slice(&mut self, pid: u64, error: &ServerError) -> NativeOutcome {
        // SRV-043 R2: the server's stop closing a connection before its
        // readiness registered is no fault; the teardown is the same.
        if matches!(error, ServerError::ReadinessClosedByStop { .. }) {
            tracing::debug!(
                connection_pid = pid,
                "connection closed by the server's stop before its readiness registered"
            );
        } else {
            tracing::error!(connection_pid = pid, %error, "connection readiness contract failed");
        }
        self.stop_crashed(pid)
    }

    pub(super) fn complete_connection_fate(
        &self,
        class: ConnectionFateClass,
    ) -> Result<(), ServerError> {
        let conversations = self.state.participant_conversations.tracked_conversations();
        self.runtime.complete_connection_fate(
            self.state.connection_incarnation,
            class,
            &conversations,
        )
    }

    pub(super) fn fail_fate(&mut self, pid: u64, error: &ServerError) -> SliceStep {
        tracing::error!(connection_pid = pid, %error, "websocket connection fate fold failed");
        self.close_transport_abnormal();
        self.release_conversations();
        self.runtime
            .mark_crashed(pid, ExitReason::Error, self.peer_addr);
        SliceStep::Stop(ExitReason::Error)
    }

    pub(super) fn finish_fate_close(&mut self, pid: u64, class: ConnectionFateClass) -> SliceStep {
        // Unbudgeted best-effort drain; the outcome is deliberately dropped.
        // What is lost is the unsent tail of the terminal response, which the
        // client will not receive before the close frame. The close decision is
        // already taken here, and retrying or sleeping is forbidden on this
        // path; the loss window is the peer's own half-close.
        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(class) {
            return self.fail_fate(pid, &error);
        }
        self.finish_normal_close(pid)
    }

    /// Finishes a client-initiated or terminal-rejection close: one unbudgeted
    /// best-effort drain (so terminal responses are not truncated by the slice
    /// budget), a best-effort transport close frame, then the shared cleanup.
    pub(super) fn finish_normal_close(&mut self, pid: u64) -> SliceStep {
        // The drain outcome is only debug-logged: a `WouldBlock` here loses the residue
        // after the socket buffer fills, so the client can miss the tail of its
        // own close response. Accepted because the close is already decided and
        // this path may not sleep, poll or retry — the loss window is the peer's
        // half-close. The close frame and flush below are silent for the same
        // reason and are only debug-logged: what they lose is the courtesy
        // close-handshake, after which the dropped socket says the same thing.
        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 Some(socket) = self.socket.as_mut() {
            if let Err(error) = socket.close(Some(CloseFrame {
                code: CloseCode::Normal,
                reason: "".into(),
            })) {
                tracing::debug!(connection_pid = pid, %error, "websocket close frame not sent");
            }
            if let Err(error) = socket.flush() {
                tracing::debug!(connection_pid = pid, %error, "websocket close flush incomplete");
            }
        }
        self.release_conversations();
        self.runtime.finish(pid);
        SliceStep::Stop(ExitReason::Normal)
    }

    /// Best-effort abnormal-close notice (diagnostics for the peer only; the
    /// lifecycle decision was already taken). Never blocks, never retries.
    pub(super) fn close_transport_abnormal(&mut self) {
        if let Some(socket) = self.socket.as_mut() {
            if let Err(error) = socket.close(Some(CloseFrame {
                code: CloseCode::Protocol,
                reason: "liminal websocket transport contract violation".into(),
            })) {
                tracing::debug!(%error, "the websocket close frame was not sent; the dropped socket ends it");
            }
            if let Err(error) = socket.flush() {
                tracing::debug!(%error, "the websocket close flush did not finish; the dropped socket ends it");
            }
        }
    }

    /// The TCP process's conversation/pending-reply release, byte-for-byte.
    pub(super) fn release_conversations(&mut self) {
        self.release_participant_publication();
        self.state.pending_replies.cancel_all();
        for (_conversation_id, conversation) in std::mem::take(&mut self.state.conversations) {
            conversation.finalize();
        }
    }
}