liminal-server 0.24.9

//! R1(vi) (§1.2(3b)) pending-reply state machine.
//!
//! Removing the in-slice blocking reply drain (the LIM-005 request-reply path)
//! removes the serialization it accidentally provided, so pending replies need
//! real bookkeeping: ONE bounded table per connection, keyed by a monotonic
//! internal operation id, each entry carrying the conversation id, stream id,
//! deadline, and terminal state. Matching is FIFO within a conversation.
//!
//! # Tombstone lifecycle (pair ruling, 2026-07-11)
//!
//! On timeout an entry becomes a TOMBSTONE, and the timeout error frame is written
//! in the woken slice. A tombstone is reclaimed by exactly TWO events — its late
//! reply arriving (consume: the reply is discarded, the tombstone freed) or its
//! conversation closing (close sweep). There is NO time-based expiry: any
//! time-based reclamation eventually races an actor that is slow-not-dead — exactly
//! what produced the timeout — and would let a very-late reply FIFO-match a younger
//! entry admitted after reclamation, delivering the wrong reply as the right one
//! (semantic corruption, which no signature attaches to at any probability).
//!
//! Bounding is by scope, not time: tombstones count against
//! `max_pending_replies_per_conversation` (§5, the sub-cap) ONLY; pending entries
//! count against both the sub-cap AND `max_pending_conversation_replies_per_connection`
//! (§5, the connection table). A conversation accumulating tombstones therefore
//! wedges ITSELF — new reply-requested admissions on it draw the typed cap refusal
//! until it closes — while sibling conversations and the connection table are
//! untouched. The self-wedge is the honest semantic: a conversation holding a
//! tombstone whose reply may still arrive IS ambiguous, and the refusal confines
//! that ambiguity to the party that created it.
//!
//! # Deadline mechanics
//!
//! Under the current busy loop, [`PendingReplyTable::expire_due`] is called each
//! slice. The DEADLINE-CHECK SEAM is named so the park-flip commit can add the
//! seventh wake source — a timer-driven `READY` wake at each entry's deadline
//! (contract R1(vi) as amended, Waffles' finding: nothing else guarantees a wake,
//! so without it a parked connection whose reply expires under zero other traffic
//! never wakes and the client waits forever for a timeout the connection believes
//! it handled). beamr 0.13 exposes a timer wheel (`Scheduler::timers`) but wiring a
//! per-entry timer to a `READY` enqueue is the park-flip's job; this table builds
//! the seam and leaves the wake source to it (flagged, not faked).

use std::time::{Duration, Instant};

use beamr::timer::TimerRef;
use liminal::protocol::{CONVERSATION_REPLY_REQUESTED_FLAG, Frame, MessageEnvelope};

use super::service_tap::CallTrace;
use crate::ServerError;
use crate::auth_membership::ReleaseCheck;
#[path = "pending_reply/tap.rs"]
mod tap;
#[path = "pending_reply/timers.rs"]
mod timers;

/// Server-error reason code carried on the reply timeout / cap-refusal frames,
/// matching the code the rest of the connection paths use.
const SERVER_ERROR_CODE: u16 = 0xFFFF;

/// Default bound on how long the server holds a reply-requested operation before
/// it times out. Replaces the old in-slice 5 s BLOCKING drain: the wait is no
/// longer in the slice path — the entry sits in the table and the deadline is
/// checked each slice — but the 5 s bound to a client's correlated reply is
/// unchanged, so client-visible timeout behaviour is preserved.
pub(super) const DEFAULT_REPLY_TIMEOUT: Duration =
    Duration::from_millis(crate::config::types::LimitsConfig::DEFAULT_REPLY_TIMEOUT_MS);

/// Terminal disposition of a table entry.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum EntryState {
    /// Awaiting the participant's reply; counts against BOTH caps.
    Pending,
    /// Timed out; the late reply may still arrive. Counts against the
    /// per-conversation sub-cap ONLY. Reclaimed only by late-reply consume or
    /// conversation-close sweep.
    Tombstone,
}

/// One in-flight (or timed-out) reply-requested operation.
#[derive(Debug)]
struct PendingReplyEntry {
    /// Monotonic internal operation id — the FIFO ordering key within a
    /// conversation. Younger entries have strictly larger ids.
    op_id: u64,
    conversation_id: u64,
    /// Client-chosen application stream the correlated reply / timeout rides.
    stream_id: u32,
    /// When a still-`Pending` entry becomes a tombstone.
    deadline: Instant,
    state: EntryState,
    /// Per-active-entry deadline wake. `None` only between admission and the
    /// admitting slice's timer synchronization, or after expiry.
    timer: Option<TimerRef>,
    /// The membership read this reply's caller was admitted under on a
    /// protected channel (ACCESS-006 R3), decided again at the reply's final
    /// release. `None` on a channel membership does not protect.
    release: Option<crate::auth_membership::ReleaseCheck>,
    tap: Option<CallTrace>,
    dispatched_at: Option<Instant>,
}

/// Per-connection pending-reply table (§1.2(3b)).
#[derive(Debug)]
pub(super) struct PendingReplyTable {
    /// All entries (pending and tombstone), ordered by ascending `op_id`, so the
    /// FRONT-most entry for a conversation is always its oldest — the only one a
    /// reply may consume (never a younger one).
    entries: Vec<PendingReplyEntry>,
    next_op_id: u64,
    /// §5 `max_pending_replies_per_conversation` sub-cap (default 8): counts
    /// pending + tombstone entries for one conversation.
    max_per_conversation: usize,
    /// §5 `max_pending_conversation_replies_per_connection` (default 32): counts
    /// PENDING entries across the whole connection.
    max_per_connection: usize,
    reply_timeout: Duration,
    /// Timer handles detached by completion/close and awaiting host cancellation.
    /// Bounded by the same typed pending-entry cap as `entries`.
    retired_timers: Vec<TimerRef>,
}

impl Default for PendingReplyTable {
    /// A table carrying the §5 default caps and the default reply timeout. Used by
    /// [`ConnectionProcessState::default`]; the connection replaces it with a table
    /// built from the runtime's configured limits at construction. The non-config
    /// runtimes use the same §5 defaults, so the values match.
    fn default() -> Self {
        Self::new(
            crate::config::types::LimitsConfig::DEFAULT_MAX_PENDING_REPLIES_PER_CONVERSATION,
            crate::config::types::LimitsConfig::DEFAULT_MAX_PENDING_CONVERSATION_REPLIES_PER_CONNECTION,
            DEFAULT_REPLY_TIMEOUT,
        )
    }
}

impl PendingReplyTable {
    /// Builds a table with the §5 caps and the reply timeout.
    pub(super) const fn new(
        max_per_conversation: usize,
        max_per_connection: usize,
        reply_timeout: Duration,
    ) -> Self {
        Self {
            entries: Vec::new(),
            next_op_id: 1,
            max_per_conversation,
            max_per_connection,
            reply_timeout,
            retired_timers: Vec::new(),
        }
    }

    /// Admits a reply-requested operation on `conversation_id`/`stream_id`,
    /// deadlined at `now + reply_timeout`, returning the reserved operation id.
    ///
    /// Admission is a RESERVATION made BEFORE the message is forwarded to the
    /// participant (cap-before-mutation): a cap-refused operation must never
    /// reach the services adapter — a refused-but-forwarded request would
    /// produce an orphan reply with no table entry, which could later
    /// FIFO-match a younger admitted operation (wrong-reply delivery). A caller
    /// whose forward fails AFTER admission rolls back the exact reservation via
    /// [`Self::cancel`].
    ///
    /// # Errors
    /// Returns [`ServerError::ConnectionCapReached`] when the per-conversation
    /// sub-cap (pending + tombstone) or the per-connection pending cap is full. A
    /// conversation whose sub-cap is full of tombstones self-wedges here while
    /// siblings proceed.
    pub(super) fn admit(
        &mut self,
        conversation_id: u64,
        stream_id: u32,
        now: Instant,
    ) -> Result<u64, ServerError> {
        // Sub-cap: pending + tombstones for THIS conversation.
        let per_conversation = self
            .entries
            .iter()
            .filter(|entry| entry.conversation_id == conversation_id)
            .count();
        if per_conversation >= self.max_per_conversation {
            return Err(ServerError::ConnectionCapReached {
                operation: "conversation reply".to_owned(),
                cap: "max_pending_replies_per_conversation",
                limit: self.max_per_conversation,
            });
        }
        // Connection table: PENDING entries only (tombstones do not count here).
        let per_connection_pending = self
            .entries
            .iter()
            .filter(|entry| entry.state == EntryState::Pending)
            .count();
        if per_connection_pending >= self.max_per_connection {
            return Err(ServerError::ConnectionCapReached {
                operation: "conversation reply".to_owned(),
                cap: "max_pending_conversation_replies_per_connection",
                limit: self.max_per_connection,
            });
        }
        let op_id = self.next_op_id;
        self.next_op_id += 1;
        self.entries.push(PendingReplyEntry {
            op_id,
            conversation_id,
            stream_id,
            deadline: now + self.reply_timeout,
            state: EntryState::Pending,
            timer: None,
            release: None,
            tap: None,
            dispatched_at: None,
        });
        Ok(op_id)
    }

    /// Carries `release` to the reply of the operation admitted as `op_id`
    /// (ACCESS-006 R3): [`Self::match_reply`] decides it again before the
    /// reply's bytes are released. An unknown `op_id` binds nothing.
    pub(super) fn bind_release(
        &mut self,
        op_id: u64,
        release: crate::auth_membership::ReleaseCheck,
    ) {
        if let Some(entry) = self.entries.iter_mut().find(|entry| entry.op_id == op_id) {
            entry.release = Some(release);
        }
    }

    /// Whether the operation admitted as `op_id` is still pending: not yet
    /// answered, cancelled, expired into a tombstone or swept with its
    /// conversation (SRV-027 R2 reads it for a routed call's slot).
    pub(super) fn is_pending(&self, op_id: u64) -> bool {
        self.entries
            .iter()
            .any(|entry| entry.op_id == op_id && entry.state == EntryState::Pending)
    }

    /// Rolls back exactly one reservation made by [`Self::admit`]: the entry with
    /// `op_id` is removed regardless of state. Used when the forward that followed
    /// a successful admission fails — the operation never reached the participant,
    /// so no reply can ever arrive for it and the reservation must not linger to
    /// time out into a spurious tombstone.
    pub(super) fn cancel(&mut self, op_id: u64) {
        if let Some(index) = self.entries.iter().position(|entry| entry.op_id == op_id) {
            let mut entry = self.entries.remove(index);
            self.retire_timer(entry.timer);
            if let Some(tap) = entry.tap.take() {
                tap.finish(
                    crate::tap::TapOutcome::Refused,
                    Some("call_cancelled"),
                    Instant::now(),
                );
            }
        }
    }

    /// Correlates a reply by its exact connection-local operation id.
    ///
    /// - Exact entry is `Pending`: it is removed and the correlated reply frame
    ///   returned to write on the connection's slice.
    /// - Exact entry is a `Tombstone`: this is the timed-out request's late reply —
    ///   the tombstone is consumed (freed) and the reply DISCARDED (`None`).
    /// - Unknown id or wrong conversation: the reply is discarded (`None`).
    pub(super) fn match_reply(
        &mut self,
        conversation_id: u64,
        op_id: u64,
        reply: MessageEnvelope,
    ) -> Option<Frame> {
        let index = self
            .entries
            .iter()
            .position(|entry| entry.op_id == op_id && entry.conversation_id == conversation_id)?;
        let mut entry = self.entries.remove(index);
        self.retire_timer(entry.timer);
        let frame = match entry.state {
            // ACCESS-006 R3: the final release seam, the same on every
            // transport. A membership denied since the request was admitted
            // releases none of the reply's bytes; the caller is told by the
            // refusal's name on its own stream, and the reply is dropped.
            EntryState::Pending => match entry.release.as_ref().map(ReleaseCheck::check) {
                Some(Err(refusal)) => Some(Frame::ConversationError {
                    flags: 0,
                    stream_id: entry.stream_id,
                    conversation_id: entry.conversation_id,
                    // No new reason code is allocated (R5): the conversation
                    // paths' server-error code, the refusal named in the words.
                    reason_code: SERVER_ERROR_CODE,
                    message: Some(refusal.to_string()),
                }),
                Some(Ok(_)) | None => Some(Frame::ConversationMessage {
                    flags: CONVERSATION_REPLY_REQUESTED_FLAG,
                    stream_id: entry.stream_id,
                    conversation_id: entry.conversation_id,
                    envelope: reply,
                }),
            },
            // The late reply for a timed-out request: consume the tombstone, drop
            // the reply. Never delivered late, never mis-correlated to a younger
            // request.
            EntryState::Tombstone => None,
        };
        if let Some(tap) = entry.tap.take()
            && let Some(frame) = frame.as_ref()
        {
            tap.frame(frame);
        }
        frame
    }

    /// DEADLINE-CHECK SEAM (R1(vi) expiry half). Converts every `Pending` entry
    /// whose deadline has passed into a `Tombstone` and returns the timeout error
    /// frame to write for each, in FIFO order.
    ///
    /// Called each slice under the busy loop. PARK-FLIP: the park-flip adds a
    /// timer-driven `READY` wake at each entry's deadline (contract R1(vi) as
    /// amended) so a parked connection with zero other traffic still wakes to
    /// write the timeout — this seam is where that wake feeds in; the tombstone
    /// transition and frame generation stay exactly here.
    pub(super) fn expire_due(&mut self, now: Instant) -> Vec<Frame> {
        let mut frames = Vec::new();
        for entry in &mut self.entries {
            if entry.state == EntryState::Pending && entry.deadline <= now {
                entry.state = EntryState::Tombstone;
                if let Some(tap) = entry.tap.take() {
                    tap.finish(crate::tap::TapOutcome::Unanswered, None, now);
                }
                // Dropping the TimerRef WITHOUT cancel is deliberate, not a
                // leak: the deadline timer either already fired (that wake is
                // why we are here) or fires once more as an R6-harmless
                // spurious wake; the wheel entry is consumed at fire and the
                // exposure is bounded by the reply-deadline horizon.
                entry.timer = None;
                frames.push(reply_timeout_frame(entry.stream_id, entry.conversation_id));
            }
        }
        frames
    }

    /// Conversations that currently hold at least one `Pending` entry — the set
    /// used for active-deadline bookkeeping. Bounded by the caps, so
    /// this is a small scan, not an unbounded one.
    #[cfg(test)]
    pub(super) fn conversations_awaiting_reply(&self) -> Vec<u64> {
        let mut ids: Vec<u64> = self
            .entries
            .iter()
            .filter(|entry| entry.state == EntryState::Pending)
            .map(|entry| entry.conversation_id)
            .collect();
        ids.sort_unstable();
        ids.dedup();
        ids
    }

    /// Conversations holding any ownership entry, pending or tombstoned. Reply
    /// drains and post-arm probes use this set so a late reply can consume a
    /// tombstone even when no younger operation is pending on the conversation.
    pub(super) fn conversations_with_entries(&self) -> Vec<u64> {
        let mut ids: Vec<u64> = self
            .entries
            .iter()
            .map(|entry| entry.conversation_id)
            .collect();
        ids.sort_unstable();
        ids.dedup();
        ids
    }

    /// Conversation-close sweep: removes ALL entries (pending and tombstone) for a
    /// closing conversation — one of the two tombstone-reclamation triggers. A
    /// pending entry swept here had its reply never arrive; the client already saw
    /// (or will not need) the reply because the conversation is gone.
    pub(super) fn remove_conversation(&mut self, conversation_id: u64) {
        let mut retained = Vec::with_capacity(self.entries.len());
        let entries = std::mem::take(&mut self.entries);
        for mut entry in entries {
            if entry.conversation_id == conversation_id {
                self.retire_timer(entry.timer);
                if let Some(tap) = entry.tap.take() {
                    tap.finish(
                        crate::tap::TapOutcome::Refused,
                        Some("conversation_closed"),
                        Instant::now(),
                    );
                }
            } else {
                retained.push(entry);
            }
        }
        self.entries = retained;
    }

    /// Connection-finalization cancel (§1.2(5)): drops every entry. Called before
    /// conversation actors are torn down, so no entry outlives its connection.
    pub(super) fn cancel_all(&mut self) {
        let entries = std::mem::take(&mut self.entries);
        for mut entry in entries {
            self.retire_timer(entry.timer);
            if let Some(tap) = entry.tap.take() {
                tap.finish(
                    crate::tap::TapOutcome::Refused,
                    Some("connection_closed"),
                    Instant::now(),
                );
            }
        }
    }

    /// Total entries (pending + tombstone) — test/observability.
    #[cfg(test)]
    pub(super) fn len(&self) -> usize {
        self.entries.len()
    }

    /// Pending entries for `conversation_id` (test/observability).
    #[cfg(test)]
    pub(super) fn pending_for(&self, conversation_id: u64) -> usize {
        self.entries
            .iter()
            .filter(|entry| {
                entry.conversation_id == conversation_id && entry.state == EntryState::Pending
            })
            .count()
    }

    /// Tombstones for `conversation_id` (test/observability).
    #[cfg(test)]
    pub(super) fn tombstones_for(&self, conversation_id: u64) -> usize {
        self.entries
            .iter()
            .filter(|entry| {
                entry.conversation_id == conversation_id && entry.state == EntryState::Tombstone
            })
            .count()
    }
}

/// Builds the reply-timeout error frame for a conversation stream — the same
/// `ConversationError` vocabulary the old in-slice drain produced on timeout, so
/// the client sees an unchanged error shape.
fn reply_timeout_frame(stream_id: u32, conversation_id: u64) -> Frame {
    Frame::ConversationError {
        flags: 0,
        stream_id,
        conversation_id,
        reason_code: SERVER_ERROR_CODE,
        message: Some(
            "conversation reply timed out: no participant reply arrived within the deadline"
                .to_owned(),
        ),
    }
}

/// Renders a cap-refusal into a `ConversationError` frame for the reply-requested
/// frame that was refused, so the client learns its request was not admitted.
pub(super) fn cap_refusal_frame(
    stream_id: u32,
    conversation_id: u64,
    error: &ServerError,
) -> Frame {
    Frame::ConversationError {
        flags: 0,
        stream_id,
        conversation_id,
        reason_code: SERVER_ERROR_CODE,
        message: Some(error.to_string()),
    }
}

/// Builds a protocol reply envelope carrying `payload` under the wildcard schema,
/// mirroring how the old drain framed a participant reply (schema/causal metadata
/// are not bridged in v1).
#[cfg(test)]
pub(super) fn test_reply_envelope(payload: &[u8]) -> MessageEnvelope {
    use liminal::protocol::{CausalContext, SchemaId as ProtocolSchemaId};
    MessageEnvelope::new(
        ProtocolSchemaId::new([0; ProtocolSchemaId::WIRE_LEN]),
        CausalContext::independent(),
        payload.to_vec(),
    )
}

#[cfg(test)]
#[path = "pending_reply_tests.rs"]
mod tests;