liminal-server 0.25.4

use super::{
    Arc, Atom, ConnectionIncarnation, ConnectionRuntime, Duration, Frame, Instant,
    PushSlotDisposition, PushSlotResolution, Receiver, RecvTimeoutError, ServerError, SocketAddr,
    SupervisorInner, TryRecvError, Weak, WorkerRefusal,
};

/// Handle for one supervised connection process.
#[derive(Clone, Debug)]
pub struct ConnectionHandle {
    pub(super) pid: u64,
    pub(super) peer_addr: Option<SocketAddr>,
    pub(super) connection_incarnation: Option<ConnectionIncarnation>,
    pub(super) supervisor: Arc<SupervisorInner>,
}

impl ConnectionHandle {
    /// Every later report of the frames this connection writes to its
    /// transport. Test-only (R3b F8, ruling 2a).
    #[cfg(test)]
    pub(crate) fn observe_frames_written(&self) -> Receiver<Vec<liminal::protocol::FrameType>> {
        self.supervisor.runtime.observe_frames_written(self.pid)
    }

    /// Returns the beamr process id for this connection.
    #[must_use]
    pub const fn pid(&self) -> u64 {
        self.pid
    }

    /// Returns the peer address if it was available from the accepted stream.
    #[must_use]
    pub const fn peer_addr(&self) -> Option<SocketAddr> {
        self.peer_addr
    }

    /// Returns the durable participant connection incarnation, when this
    /// supervisor has a complete participant service installed.
    ///
    /// `None` identifies a services adapter that does not advertise participant
    /// lifecycle semantics.
    #[must_use]
    pub const fn connection_incarnation(&self) -> Option<ConnectionIncarnation> {
        self.connection_incarnation
    }

    /// Returns whether the beamr process is still live.
    #[must_use]
    pub fn is_live(&self) -> bool {
        self.supervisor
            .scheduler
            .process_table()
            .get(self.pid)
            .is_some()
    }

    /// Requests an error exit for tests and supervisor control paths.
    ///
    /// # Errors
    /// Returns [`ServerError`] when the process is no longer live.
    pub fn request_crash(&self) -> Result<(), ServerError> {
        if self
            .supervisor
            .scheduler
            .enqueue_atom_message(self.pid, Atom::ERROR)
        {
            Ok(())
        } else {
            Err(ServerError::ListenerAccept {
                message: format!("connection process {} is not live", self.pid),
            })
        }
    }
}

/// Awaits the correlated reply to a single server-initiated push.
///
/// Returned by [`ConnectionSupervisor::push_to_connection`]. The reply slot is
/// resolved when the originating connection process receives a `PushReply` frame
/// carrying the same correlation id, so [`PushReplyAwaiter::receive`] blocks
/// (bounded) for that one correlated answer.
#[derive(Debug)]
pub struct PushReplyAwaiter {
    pub(super) correlation_id: u64,
    pub(super) receiver: Receiver<PushSlotResolution>,
    /// This push's absolute reply deadline, mirrored from its slot. `None` (the
    /// default push) selects the no-deadline receive path, which NEVER touches
    /// the runtime — byte-compatible with 0.2.3, no shared-lock exposure.
    /// `Some` lets `receive` wait `min(caller quantum, time until deadline)` and
    /// resolve expiry promptly, so the caller's quantum can never select a
    /// deadlined push's terminal outcome.
    pub(super) deadline: Option<Instant>,
    /// Weak handle to the owning runtime, used ONLY by the explicit-deadline
    /// path to resolve expiry host-side at [`receive`](Self::receive). A
    /// no-deadline push never upgrades it. `Weak` so the awaiter never keeps the
    /// runtime alive; if it is already gone, the slot (and its sender) is gone
    /// with it — the connection side is torn down.
    pub(super) runtime: Weak<ConnectionRuntime>,
}

impl PushReplyAwaiter {
    /// Returns the correlation id this awaiter is matched on.
    #[must_use]
    pub const fn correlation_id(&self) -> u64 {
        self.correlation_id
    }

    /// Blocks up to `timeout` for the client's correlated reply payload.
    ///
    /// `timeout` is a WAIT QUANTUM ONLY — a MAXIMUM wait, not a promise to
    /// block: an elapsed poll is a benign re-arm, never a failure; the reply's
    /// lifetime belongs to the push. A caller may re-invoke `receive`
    /// indefinitely after a [`ServerError::PushReplyTimeout`]: the reserved slot
    /// is untouched and a later reply is still delivered byte-exact. The poll
    /// quantum never changes the protocol outcome — for a deadlined push the
    /// call waits no longer than the EARLIER of the caller's quantum and the
    /// push's deadline, so the terminal expiry is returned promptly once due,
    /// never held until the quantum ends and never deferred past it.
    ///
    /// A push with no explicit deadline never touches shared supervisor state
    /// here: the elapsed quantum returns straight from the channel wait
    /// (behaviour-compatible with 0.2.3 — no registry lock, no contention, no
    /// poison exposure on the unchanged API).
    ///
    /// # Errors
    /// Returns [`ServerError::PushReplyTimeout`] when no reply arrived within this
    /// `timeout` quantum and the push's deadline (if any) is not yet due (a
    /// benign re-arm — call again to keep waiting);
    /// [`ServerError::PushReplyExpired`] when the push carried an explicit reply
    /// deadline (via
    /// [`push_to_connection_with_deadline`](ConnectionSupervisor::push_to_connection_with_deadline))
    /// and that deadline is due (terminal: the slot is removed and its §5 cap
    /// admission released; returned as soon as the deadline passes, even
    /// mid-quantum — but evaluated at observation points, not against the wall
    /// clock: a reply already delivered when this call observes the slot wins
    /// over expiry, even if it arrived after the deadline instant); or
    /// [`ServerError::PushReplyDisconnected`] when the connection process
    /// dropped the reply slot (the connection closed — the prompt worker-death
    /// signal). The variants are distinct so callers classify by type, not
    /// message.
    pub fn receive(&self, timeout: Duration) -> Result<Vec<u8>, ServerError> {
        self.deadline.map_or_else(
            || self.receive_no_deadline(timeout),
            |deadline| self.receive_deadlined(timeout, deadline),
        )
    }

    /// The default-push receive: exactly the 0.2.3 shape. One bounded channel
    /// wait; an elapsed quantum is a benign timeout straight from the channel —
    /// no runtime upgrade, no registry lock, EVER (unrelated registry work can
    /// never stretch this call past its quantum, and registry poison cannot
    /// reach it).
    fn receive_no_deadline(&self, timeout: Duration) -> Result<Vec<u8>, ServerError> {
        match self.receiver.recv_timeout(timeout) {
            Ok(resolution) => self.resolution_into_result(resolution),
            Err(RecvTimeoutError::Timeout) => Err(ServerError::PushReplyTimeout {
                correlation_id: self.correlation_id,
            }),
            Err(RecvTimeoutError::Disconnected) => Err(ServerError::PushReplyDisconnected {
                correlation_id: self.correlation_id,
            }),
        }
    }

    /// The deadlined receive: waits `min(caller quantum, time until deadline)`
    /// and re-evaluates reply-first-then-expiry on every wake, so the caller's
    /// quantum can never select the terminal outcome (S1). Order per iteration:
    ///
    /// 1. Deliver a reply already in hand — an answer that is here must never be
    ///    reported as a timeout OR an expiry (the observation-point rule).
    /// 2. If the deadline is due, resolve expiry atomically against the registry
    ///    (`expire_slot`) and return the terminal outcome promptly — even when
    ///    the caller's quantum has time left (the quantum is a max wait).
    /// 3. Otherwise wait for the earlier of quantum-remaining and deadline; a
    ///    wake re-runs 1-2, and an exhausted quantum before the deadline is the
    ///    benign `PushReplyTimeout` re-arm with the slot untouched.
    fn receive_deadlined(
        &self,
        timeout: Duration,
        deadline: Instant,
    ) -> Result<Vec<u8>, ServerError> {
        let started = Instant::now();
        loop {
            if let Some(result) = self.try_take_reply() {
                return result;
            }
            let now = Instant::now();
            if now >= deadline {
                return self.expire_slot();
            }
            let quantum_left = timeout.saturating_sub(now.duration_since(started));
            if quantum_left.is_zero() {
                return Err(ServerError::PushReplyTimeout {
                    correlation_id: self.correlation_id,
                });
            }
            match self
                .receiver
                .recv_timeout(quantum_left.min(deadline.duration_since(now)))
            {
                Ok(resolution) => return self.resolution_into_result(resolution),
                Err(RecvTimeoutError::Disconnected) => {
                    return Err(ServerError::PushReplyDisconnected {
                        correlation_id: self.correlation_id,
                    });
                }
                // Re-loop: deliver a reply that raced the wake, expire a
                // now-due deadline, or report the exhausted quantum benignly.
                Err(RecvTimeoutError::Timeout) => {}
            }
        }
    }

    /// Resolves a due deadline against the registry's atomic removal transition.
    fn expire_slot(&self) -> Result<Vec<u8>, ServerError> {
        let timeout_error = || ServerError::PushReplyTimeout {
            correlation_id: self.correlation_id,
        };
        let Some(runtime) = self.runtime.upgrade() else {
            // The runtime is gone, and the slot map (with every sender) with it:
            // the connection side is torn down. Re-check the channel so the
            // dropped sender reads as the established DISCONNECTED outcome — a
            // dead runtime must not be misreported as a benign healthy-but-slow
            // timeout (S4).
            return self
                .try_take_reply()
                .unwrap_or(Err(ServerError::PushReplyDisconnected {
                    correlation_id: self.correlation_id,
                }));
        };
        match runtime.expire_push_if_due(self.correlation_id) {
            PushSlotDisposition::Expired => Err(ServerError::PushReplyExpired {
                correlation_id: self.correlation_id,
            }),
            // Unreachable by construction (this is only called with the deadline
            // due, and the registry re-reads a monotonic clock); honest benign
            // fallback rather than a panic.
            PushSlotDisposition::Live => Err(timeout_error()),
            // Another path (a concurrent `resolve_push`, or connection close)
            // removed the slot under the registry lock while we waited on it. Its
            // send, if any, happens under that same lock, so re-check the channel:
            // a delivered reply is present now; a dropped sender is disconnected.
            PushSlotDisposition::Absent => self
                .try_take_reply()
                .unwrap_or_else(|| Err(timeout_error())),
        }
    }

    /// SRV-027 R2: the outcome if the slot has resolved, read without waiting
    /// by the requester's slice after its wake: `Some(Ok(payload))` for the
    /// worker's reply, `Some(Err(..))` for a refusal, a frame the worker's
    /// connection could not carry, or the worker gone (the slot swept at its
    /// close), `None` while nothing has arrived.
    pub(in crate::server::connection) fn try_receive(
        &self,
    ) -> Option<Result<Vec<u8>, PushReplyFailure>> {
        match self.receiver.try_recv() {
            Ok(PushSlotResolution::WorkerRefused(refusal)) => {
                Some(Err(PushReplyFailure::Worker(refusal)))
            }
            Ok(resolution) => Some(
                self.resolution_into_result(resolution)
                    .map_err(PushReplyFailure::Server),
            ),
            Err(TryRecvError::Empty) => None,
            Err(TryRecvError::Disconnected) => Some(Err(PushReplyFailure::Server(
                ServerError::PushReplyDisconnected {
                    correlation_id: self.correlation_id,
                },
            ))),
        }
    }

    /// Non-blocking check for a reply already sitting in the channel. `Some` with
    /// the payload or a disconnected error; `None` when the channel is still empty
    /// (no reply yet — the caller re-arms).
    pub(super) fn try_take_reply(&self) -> Option<Result<Vec<u8>, ServerError>> {
        match self.receiver.try_recv() {
            Ok(resolution) => Some(self.resolution_into_result(resolution)),
            Err(TryRecvError::Disconnected) => Some(Err(ServerError::PushReplyDisconnected {
                correlation_id: self.correlation_id,
            })),
            Err(TryRecvError::Empty) => None,
        }
    }

    /// Turns one delivered [`PushSlotResolution`] into this awaiter's result, so
    /// all three receive paths read a resolution the same way and none of them
    /// can grow its own interpretation of one.
    ///
    /// Note what is NOT here: a dropped `Sender` never reaches this function. It
    /// is a channel disconnect, and so stays the retryable
    /// [`ServerError::PushReplyDisconnected`] on every path that kills a slot
    /// without naming a reason — connection close included.
    fn resolution_into_result(
        &self,
        resolution: PushSlotResolution,
    ) -> Result<Vec<u8>, ServerError> {
        match resolution {
            PushSlotResolution::Reply(payload) => Ok(payload),
            PushSlotResolution::WorkerRefused(refusal) => Err(ServerError::PushReplyRefused {
                correlation_id: self.correlation_id,
                reason: refusal.code().caller_name().to_owned(),
            }),
            PushSlotResolution::ReplyRefused { reason } => Err(ServerError::PushReplyRefused {
                correlation_id: self.correlation_id,
                reason,
            }),
            PushSlotResolution::OutboundFrameTooLarge {
                needed,
                capacity,
                queued,
            } => Err(ServerError::PushFrameExceedsOutboundCapacity {
                correlation_id: self.correlation_id,
                needed,
                capacity,
                queued,
            }),
        }
    }
}

/// Routed calls retain words only inside the validated, redacted record. Host
/// push APIs predate worker refusals and return the stable code alone, keeping
/// worker data out of their generic error's derived debug representation.
#[derive(Debug)]
pub(in crate::server::connection) enum PushReplyFailure {
    Worker(WorkerRefusal),
    Server(ServerError),
}

impl PushReplyFailure {
    /// Settles the pending operation and its tap exactly once. Worker words
    /// are released only after the caller's current membership permits the
    /// read, and only the stable refusal code reaches the tap.
    pub(in crate::server::connection) fn into_frame(
        self,
        state: &mut super::state::ConnectionProcessState,
        op_id: u64,
        stream_id: u32,
        conversation_id: u64,
        subject: &str,
        route_words: impl FnOnce(&str, &ServerError) -> String,
    ) -> Frame {
        let (reason_code, name, words) = match self {
            Self::Worker(refusal) => match state
                .membership
                .release(subject, crate::auth_membership::MembershipAction::Read)
            {
                Ok(_) => (
                    refusal.code().reason_code(),
                    refusal.code().caller_name().to_owned(),
                    refusal.into_words(),
                ),
                Err(error) => (0xffff, error.name().to_owned(), error.to_string()),
            },
            Self::Server(error) => {
                let words = route_words(subject, &error);
                (0xffff, words.clone(), words)
            }
        };
        state.pending_replies.refuse(op_id, &name);
        Frame::ConversationError {
            flags: 0,
            stream_id,
            conversation_id,
            reason_code,
            message: Some(words),
        }
    }
}