drep/llm/
concurrency.rs

1//! Concurrency cap for LLM requests.
2//!
3//! One thing, on purpose: a bounded number of in-flight requests, nothing
4//! else. A local commit gate reviewing a handful of changed files does not need
5//! server-oriented rate-limit machinery:
6//!
7//! - **Per-repo semaphores** are a map with one entry. One invocation, one
8//!   repo, one bucket.
9//! - **A requests-per-minute window and a tokens-per-minute budget** are a
10//!   second, worse implementation of what `open-agent-sdk` 0.7.0 already
11//!   does: it classifies 429 as retryable and backs off, so a client-side
12//!   throttle duplicates the backoff without owning it.
13//! - **Two-step token accounting** exists to reconcile an estimate against
14//!   actuals across a queue. With no token budget, nothing to reconcile, so
15//!   the "never hold the lock across an await" hazard disappears with it.
16//! - **A circuit breaker** protects a shared service from a stampede. One
17//!   developer committing is not a stampede.
18//!
19//! If a real workload later shows `max_concurrent` is not enough, add the
20//! narrowest mechanism that fixes it.
21//!
22//! [`tokio::sync::Semaphore::acquire`] returns `Result` because the
23//! semaphore can be closed. This limiter never closes it, so the error
24//! case is unreachable in practice; it is mapped to an empty guard rather
25//! than propagated as an error the caller cannot act on.
26
27use std::sync::Arc;
28
29use tokio::sync::{Semaphore, SemaphorePermit};
30
31/// A bounded concurrency limiter.
32///
33/// Cheap to clone (the semaphore is wrapped in an `Arc`); typically built
34/// once per process and shared across the analyzer.
35#[derive(Clone, Debug)]
36pub struct Limiter {
37    semaphore: Arc<Semaphore>,
38}
39
40/// RAII guard holding one slot in the limiter.
41///
42/// Drop the guard to release the slot. In the (unreachable) error case
43/// where the semaphore was closed, the guard holds no permit and dropping
44/// it is a no-op - the limiter cannot make a phantom permit do useful
45/// work, but the type stays total so callers do not have to handle a
46/// second error variant.
47#[must_use = "dropping the guard immediately releases the slot, so a bare \
48              `limiter.acquire().await;` provides no backpressure at all"]
49pub struct LimiterGuard<'a> {
50    /// The held permit, if any. `Option` so the unreachable error case
51    /// can leave the guard empty; `Drop` below explicitly drops the
52    /// permit so the slot returns to the pool.
53    permit: Option<SemaphorePermit<'a>>,
54}
55
56impl Drop for LimiterGuard<'_> {
57    fn drop(&mut self) {
58        // Explicitly drop the held permit so the slot is released now
59        // rather than whenever the field happens to be reclaimed. The
60        // `take` also makes the field's role unambiguous to the
61        // dead-code lint: without it, holding a value that is "only
62        // dropped" can warn.
63        drop(self.permit.take());
64    }
65}
66
67impl Limiter {
68    /// Build a limiter that allows at most `max_concurrent` in-flight
69    /// requests. `max_concurrent = 0` is allowed by the type system but
70    /// makes every `acquire` await forever; callers are expected to set a
71    /// positive value from `LlmConfig::max_concurrent` (default 3).
72    pub fn new(max_concurrent: usize) -> Self {
73        Self {
74            semaphore: Arc::new(Semaphore::new(max_concurrent)),
75        }
76    }
77
78    /// Acquire a slot. The slot is held until the returned guard is
79    /// dropped.
80    ///
81    /// Backpressure is the entire job: callers await, and once `K` requests
82    /// are in flight, the next one queues here until one of the holders
83    /// drops its guard.
84    pub async fn acquire(&self) -> LimiterGuard<'_> {
85        // `Semaphore::acquire` only returns `Err(AcquireError)` when the
86        // semaphore has been closed. We never close it, so the error is
87        // unreachable; mapping it to an empty guard keeps the API total
88        // without resorting to `unwrap` outside tests.
89        let permit = self.semaphore.acquire().await.ok();
90        LimiterGuard { permit }
91    }
92
93    /// The number of slots currently available for acquisition.
94    ///
95    /// Reflects the configured maximum before any acquisition, decreases by
96    /// one per held guard, and returns to the maximum once every guard has
97    /// been dropped. Test-only - exposed via the API so tests can pin the
98    /// behaviour without reaching into the semaphore.
99    pub fn available(&self) -> usize {
100        self.semaphore.available_permits()
101    }
102}
103
104#[cfg(test)]
105mod tests;
drep/llm/concurrency.rs

drep/llm/
concurrency.rs
