liminal-server 0.25.4

//! The WebSocket connection process's parking: readiness armed once and
//! rearmed, deadline timers kept in step, and the final probe before it parks.

use super::WebSocketConnectionProcess;
use crate::ServerError;
use crate::server::connection::participant_delivery::has_held_participant_head;
use crate::server::connection::readiness_refusal::{ReadinessCall, readiness_refusal};
use beamr::native::native_process::NativeContext;
use beamr::scheduler::Interest;
use beamr::term::Term;
use std::os::fd::AsRawFd;
use std::time::Instant;

impl WebSocketConnectionProcess {
    pub(super) fn arm_readiness(
        &mut self,
        pid: u64,
        ctx: &NativeContext<'_>,
        interest: Interest,
    ) -> Result<(), ServerError> {
        let facility = ctx
            .readiness_facility()
            .ok_or_else(|| ServerError::ListenerAccept {
                message: "connection scheduler started without its required readiness service"
                    .to_owned(),
            })?;
        if let Some(token) = self.readiness_token {
            return facility
                .rearm(&token, interest)
                .map_err(|error| readiness_refusal(pid, ReadinessCall::Rearm, &error));
        }
        let socket = self
            .socket
            .as_ref()
            .ok_or_else(|| ServerError::ListenerAccept {
                message: "cannot register readiness for a missing connection socket".to_owned(),
            })?;
        let fd = socket.get_ref().as_raw_fd();
        let token = facility
            .register(fd, interest, pid, self.runtime.ready_atom())
            .map_err(|error| readiness_refusal(pid, ReadinessCall::Register, &error))?;
        if let Err(error) = self.runtime.set_readiness_token_once(pid, token, fd) {
            self.runtime.deregister_unpublished_readiness(token);
            return Err(error);
        }
        self.readiness_token = Some(token);
        Ok(())
    }

    /// The TCP process's deadline-timer sync plus the keepalive timer: retired
    /// pending-reply timers are cancelled, due deadlines are armed, and one
    /// timer wake is kept armed for the next keepalive due instant so a parked
    /// connection still pings on schedule.
    pub(super) fn sync_deadline_timers(
        &mut self,
        pid: u64,
        ctx: &mut NativeContext<'_>,
    ) -> Result<(), ServerError> {
        for timer in self.state.pending_replies.take_retired_timers() {
            ctx.cancel_timer(timer);
        }
        for (op_id, delay) in self.state.pending_replies.timers_to_arm(Instant::now()) {
            let timer = ctx
                .send_after(delay, pid, Term::atom(self.runtime.ready_atom()))
                .ok_or_else(|| ServerError::ListenerAccept {
                    message: "connection scheduler has no timer facility for reply deadlines"
                        .to_owned(),
                })?;
            if !self.state.pending_replies.install_timer(op_id, timer) {
                ctx.cancel_timer(timer);
            }
        }
        if let Some(keepalive) = self.keepalive.as_mut() {
            let now = Instant::now();
            if let Some((timer, armed_for)) = keepalive.armed.take() {
                if armed_for == keepalive.next_due && armed_for > now {
                    // Still armed for the current deadline; keep it.
                    keepalive.armed = Some((timer, armed_for));
                } else {
                    // Fired or stale (the schedule advanced): retire it.
                    ctx.cancel_timer(timer);
                }
            }
            if keepalive.armed.is_none() {
                let delay = keepalive.next_due.saturating_duration_since(now);
                let timer = ctx
                    .send_after(delay, pid, Term::atom(self.runtime.ready_atom()))
                    .ok_or_else(|| ServerError::ListenerAccept {
                        message: "connection scheduler has no timer facility for keepalive pings"
                            .to_owned(),
                    })?;
                keepalive.armed = Some((timer, keepalive.next_due));
            }
        }
        Ok(())
    }

    /// Post-arm barrier probe: socket bytes, budget-stranded inbound, pumped
    /// sources, queued controls, and native mail — the TCP probe plus the
    /// inbound-budget flag that covers tungstenite's internal reassembly buffer.
    pub(super) fn final_probe(
        &self,
        pid: u64,
        ctx: &NativeContext<'_>,
    ) -> Result<bool, ServerError> {
        let socket_ready = if let Some(socket) = self.socket.as_ref() {
            let mut byte = [0_u8; 1];
            match socket.get_ref().peek(&mut byte) {
                Ok(_) => true,
                Err(error) if error.kind() == std::io::ErrorKind::WouldBlock => false,
                Err(error) if error.kind() == std::io::ErrorKind::Interrupted => true,
                Err(error) => {
                    return Err(ServerError::ListenerAccept {
                        message: format!("connection readiness probe failed: {error}"),
                    });
                }
            }
        } else {
            true
        };
        let subscription_ready = !self.state.held_deliveries.is_empty()
            || self
                .state
                .subscriptions
                .values()
                .any(|subscription| subscription.is_overflowed() || subscription.has_pending());
        let participant_ready = if self.state.held_pushes.capacity_refused()
            && has_held_participant_head(&self.state)
        {
            false
        } else {
            match self.state.participant_publication.as_ref() {
                Some(inbox) => {
                    inbox
                        .has_pending()
                        .map_err(|error| ServerError::ListenerAccept {
                            message: format!("participant publication final probe failed: {error}"),
                        })?
                }
                None => false,
            }
        };
        let reply_ready = self.state.pending_replies.has_due(Instant::now())
            || self
                .state
                .pending_replies
                .conversations_with_entries()
                .into_iter()
                .filter_map(|id| self.state.conversations.get(&id))
                .any(super::super::super::conversation::ConnectionConversation::has_pending_reply);
        Ok(socket_ready
            || self.inbound_budget_exhausted
            || subscription_ready
            || participant_ready
            || reply_ready
            || self.runtime.has_control(pid)
            || self.runtime.ready_pending(pid)
            || ctx.has_messages())
    }
}