liminal-server 0.23.1

//! The queued request names the worker the route actually selected.

use super::{ConnectionControl, ConnectionRuntime, PushReplyAwaiter, ServerError, lock};
use std::sync::{Arc, Weak};

pub(in crate::server::connection) struct RoutedPush {
    pub(in crate::server::connection) awaiter: PushReplyAwaiter,
    pub(in crate::server::connection) station: Option<Arc<str>>,
}

impl ConnectionRuntime {
    /// SRV-027 R2: pushes a routed request's `payload` to the worker
    /// connection `pid` on `subject`, the slot carrying the requester's
    /// `wake`, and hands back the awaiter the requester's slice reads
    /// non-blocking ([`PushReplyAwaiter::try_receive`]). The same
    /// INSERT -> CONFIRM -> PUBLISH order and publication invariant as the
    /// supervisor's pushes: an `Err` means no `Push` was published. The slot
    /// carries no deadline of its own; the requester's pending-reply deadline
    /// governs, and its expiry cancels the slot by correlation id.
    ///
    /// # Errors
    /// The worker's `max_pending_pushes_per_connection` cap, a worker whose
    /// record closed during registration, a scheduler that is gone, or a
    /// control the queue refused: each by name, with nothing published.
    pub(in crate::server::connection) fn route_push(
        &self,
        pid: u64,
        subject: &str,
        payload: Vec<u8>,
        wake: Option<crate::server::connection::wake::ReadyWaker>,
    ) -> Result<RoutedPush, ServerError> {
        let station = self.route_station(pid);
        let correlation_id = self.next_push_correlation_id();
        let receiver =
            self.register_push(pid, correlation_id, None, Some(subject.to_owned()), wake)?;
        if !self.confirm_push_registration(pid, correlation_id) {
            return Err(ServerError::ListenerAccept {
                message: format!(
                    "cannot route to connection process {pid}: the connection closed during push registration"
                ),
            });
        }
        let Some(scheduler) = self.scheduler.upgrade() else {
            self.cancel_push(correlation_id);
            return Err(ServerError::ListenerAccept {
                message: format!(
                    "cannot route to connection process {pid}: the connection scheduler is gone"
                ),
            });
        };
        let control = ConnectionControl::Push {
            correlation_id,
            payload,
        };
        if self.publish_control(&scheduler, pid, control) {
            Ok(RoutedPush {
                station,
                awaiter: PushReplyAwaiter {
                    correlation_id,
                    receiver,
                    deadline: None,
                    runtime: Weak::new(),
                },
            })
        } else {
            self.cancel_push(correlation_id);
            Err(ServerError::ListenerAccept {
                message: format!(
                    "cannot route to connection process {pid}: the control was not published"
                ),
            })
        }
    }

    fn route_station(&self, pid: u64) -> Option<Arc<str>> {
        let console = self.signals.get()?.console()?;
        let recorded = lock(&self.records, "connection registry").map(|records| {
            records
                .connections
                .get(&pid)
                .and_then(|record| record.registration.as_ref())
                .filter(|registration| console.tap().reserved(&registration.identity).is_none())
                .map(|registration| Arc::from(registration.identity.as_str()))
        });
        match recorded {
            Ok(station) => station,
            Err(error) => {
                console.observe_failed();
                tracing::warn!(%error, "SERVICE_TAP_TARGET_FAILED");
                None
            }
        }
    }
}