polyc-runtime 2026.8.2

//! Shared admission control for edge webhook handlers (#795).
//!
//! Exactly one edge handler used to bound how many turns it dials
//! concurrently (a `turn_limit` semaphore, acquired before spawning a
//! turn and shed with a `503` when full). Every other edge — Slack,
//! Telegram, Discord, email, A2A — accepted unbounded concurrent load,
//! dialing the agent for every inbound webhook no matter how many were
//! already in flight.
//!
//! [`AdmissionGate`] is that same bounded-concurrency pattern, extracted so
//! every edge can front its agent dial with it. A caller that fails to
//! [`AdmissionGate::try_admit`] must shed the request — respond with
//! `polyc_proto::admission_shed_text` (or the edge's own transport-level
//! equivalent, e.g. an HTTP `503`) — rather than dial the agent.

use std::sync::Arc;

use tokio::sync::{OwnedSemaphorePermit, Semaphore};

/// Bounds how many turns an edge dials concurrently.
///
/// Cheap to clone — every handle shares the same underlying limit, the same
/// shape as `AgentDialer`/`Client` elsewhere in an edge's `AppState`.
#[derive(Clone)]
pub struct AdmissionGate {
    limit: Arc<Semaphore>,
}

impl AdmissionGate {
    /// Builds a gate that admits at most `max_concurrent_turns` turns at
    /// once. A caller beyond that bound must shed instead of dialing.
    #[must_use]
    pub fn new(max_concurrent_turns: usize) -> Self {
        Self {
            limit: Arc::new(Semaphore::new(max_concurrent_turns)),
        }
    }

    /// Tries to admit one more turn. `None` means the gate is already at
    /// capacity — the caller must shed rather than dial the agent.
    ///
    /// The returned [`AdmissionPermit`] holds the slot until dropped; hold
    /// it for the lifetime of the spawned turn.
    #[must_use]
    pub fn try_admit(&self) -> Option<AdmissionPermit> {
        Arc::clone(&self.limit)
            .try_acquire_owned()
            .ok()
            .map(AdmissionPermit)
    }
}

/// Held for the lifetime of one admitted turn; dropping it frees the slot
/// back to the gate. Never read directly — its whole purpose is the `Drop`
/// impl inherited from the wrapped [`OwnedSemaphorePermit`].
#[must_use = "dropping this immediately releases the admission slot"]
pub struct AdmissionPermit(
    #[allow(dead_code, reason = "held only for its Drop")] OwnedSemaphorePermit,
);

#[cfg(test)]
mod tests {
    use super::AdmissionGate;

    #[test]
    fn admits_up_to_the_limit_then_sheds() {
        let gate = AdmissionGate::new(2);
        let first = gate.try_admit();
        assert!(first.is_some());
        let second = gate.try_admit();
        assert!(second.is_some());
        // The gate is now full — a third caller must shed, not dial.
        assert!(gate.try_admit().is_none());
        drop(first);
        drop(second);
    }

    #[test]
    fn dropping_a_permit_frees_the_slot() {
        let gate = AdmissionGate::new(1);
        let permit = gate.try_admit();
        assert!(permit.is_some());
        assert!(gate.try_admit().is_none(), "already at capacity");
        drop(permit);
        assert!(
            gate.try_admit().is_some(),
            "freed slot must be admittable again"
        );
    }

    #[test]
    fn clones_share_the_same_underlying_limit() {
        let gate = AdmissionGate::new(1);
        let clone = gate.clone();
        let _held = gate.try_admit().expect("first admit");
        // The clone shares the same semaphore, so it observes the same
        // exhausted capacity — not an independent limit of its own.
        assert!(clone.try_admit().is_none());
    }
}