liminal-server 0.25.4

//! Pending calls reserve before send and release on exact terminal events.

use super::{CallError, CallReply};
use std::collections::BTreeMap;
use std::num::NonZeroUsize;
use std::sync::{Arc, Mutex, MutexGuard, mpsc};

type Answer = Result<CallReply, CallError>;
struct Entry {
    answer: mpsc::SyncSender<Answer>,
    wake: Arc<dyn Fn() + Send + Sync>,
}
#[derive(Default)]
struct State {
    next: u64,
    entries: BTreeMap<u64, Entry>,
    closed: Option<CallError>,
}

/// One bounded table per keyed transport, with no timers or background polling.
pub struct PendingCalls {
    bound: NonZeroUsize,
    state: Mutex<State>,
}
impl std::fmt::Debug for PendingCalls {
    fn fmt(&self, out: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        out.debug_struct("PendingCalls")
            .field("bound", &self.bound)
            .finish_non_exhaustive()
    }
}
impl PendingCalls {
    /// The configured positive cap bounds pending senders and retained reply slots.
    #[must_use]
    pub fn new(bound: NonZeroUsize) -> Arc<Self> {
        Arc::new(Self {
            bound,
            state: Mutex::default(),
        })
    }
    fn lock(&self) -> Result<MutexGuard<'_, State>, CallError> {
        self.state.lock().map_err(|_| CallError::Unavailable)
    }
    /// Reserves before dispatch, so a refused call never reaches the far worker.
    ///
    /// # Errors
    /// Capacity, exhausted ids, closed transport or poisoned state.
    pub fn reserve(
        self: &Arc<Self>,
        wake: Arc<dyn Fn() + Send + Sync>,
    ) -> Result<PendingCall, CallError> {
        let (answer, receiver) = mpsc::sync_channel(1);
        let mut state = self.lock()?;
        if let Some(reason) = state.closed {
            return Err(reason);
        }
        if state.entries.len() >= self.bound.get() {
            return Err(CallError::Capacity);
        }
        state.next = state.next.checked_add(1).ok_or(CallError::Unavailable)?;
        let id = state.next;
        state.entries.insert(id, Entry { answer, wake });
        drop(state);
        Ok(PendingCall {
            id,
            receiver,
            owner: Arc::clone(self),
        })
    }
    /// Settles only the exact reservation; a late or duplicated reply cannot match a younger call.
    ///
    /// # Errors
    /// Poisoned state or impossible full response slot.
    pub fn reply(&self, reply: CallReply) -> Result<bool, CallError> {
        self.finish(reply.id(), Ok(reply))
    }
    /// Names deadline, cancellation, dispatch refusal or peer loss at its original operation.
    ///
    /// # Errors
    /// Poisoned state or impossible full response slot.
    pub fn refuse(&self, id: u64, reason: CallError) -> Result<bool, CallError> {
        self.finish(id, Err(reason))
    }
    fn finish(&self, id: u64, answer: Answer) -> Result<bool, CallError> {
        let entry = self.lock()?.entries.remove(&id);
        let Some(entry) = entry else {
            return Ok(false);
        };
        deliver(entry, answer)?;
        Ok(true)
    }
    /// A link end drains every outstanding operation exactly once and prevents new admissions.
    ///
    /// # Errors
    /// Poisoned state or impossible full response slot.
    pub fn close(&self, reason: CallError) -> Result<(), CallError> {
        let mut state = self.lock()?;
        if state.closed.is_some() {
            return Ok(());
        }
        state.closed = Some(reason);
        let entries = std::mem::take(&mut state.entries);
        drop(state);
        let mut failure = None;
        for entry in entries.into_values() {
            if let Err(error) = deliver(entry, Err(reason)) {
                failure = Some(error);
            }
        }
        failure.map_or(Ok(()), Err)
    }
    /// Counts actual retained reservations for leak assertions.
    ///
    /// # Errors
    /// The table was poisoned.
    pub fn count(&self) -> Result<usize, CallError> {
        Ok(self.lock()?.entries.len())
    }
}

fn deliver(entry: Entry, answer: Answer) -> Result<(), CallError> {
    let Entry {
        answer: sender,
        wake,
    } = entry;
    match sender.try_send(answer) {
        Ok(()) | Err(mpsc::TrySendError::Disconnected(_)) => {
            wake();
            Ok(())
        }
        Err(mpsc::TrySendError::Full(_)) => Err(CallError::Unavailable),
    }
}

/// A caller owns exactly one reserved operation and cancels it when it leaves.
pub struct PendingCall {
    id: u64,
    receiver: mpsc::Receiver<Answer>,
    owner: Arc<PendingCalls>,
}
impl std::fmt::Debug for PendingCall {
    fn fmt(&self, out: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        out.debug_struct("PendingCall")
            .field("id", &self.id)
            .finish_non_exhaustive()
    }
}
impl PendingCall {
    /// The never-reused id to seal into the outgoing call.
    #[must_use]
    pub const fn id(&self) -> u64 {
        self.id
    }
    /// Blocks on a delivered outcome, never a clock. Used by synchronous wire clients.
    ///
    /// # Errors
    /// The named terminal outcome or the response owner unexpectedly departed.
    pub fn receive(&self) -> Answer {
        self.receiver.recv().map_err(|_| CallError::Unavailable)?
    }
    /// The connection slice reads once after its registered wake.
    ///
    /// # Errors
    /// The response owner unexpectedly departed.
    pub fn ready(&self) -> Result<Option<Answer>, CallError> {
        match self.receiver.try_recv() {
            Ok(answer) => Ok(Some(answer)),
            Err(mpsc::TryRecvError::Empty) => Ok(None),
            Err(mpsc::TryRecvError::Disconnected) => Err(CallError::Unavailable),
        }
    }
}
impl Drop for PendingCall {
    fn drop(&mut self) {
        if let Err(error) = self.owner.refuse(self.id, CallError::Cancelled) {
            tracing::error!(%error, "link call reservation retirement failed");
        }
    }
}