photon_backend/delivery/
worker_pool.rs

1//! Bounded handler worker pool per subscription partition.
2
3use std::sync::Arc;
4
5use tokio::sync::Semaphore;
6
7use crate::error::{PhotonError, Result};
8
9/// Limits concurrent handler tasks for a subscription.
10pub struct WorkerPool {
11    semaphore: Arc<Semaphore>,
12}
13
14impl WorkerPool {
15    /// Create a pool with at most `max_concurrent` in-flight handlers.
16    #[must_use]
17    pub fn new(max_concurrent: usize) -> Self {
18        Self {
19            semaphore: Arc::new(Semaphore::new(max_concurrent.max(1))),
20        }
21    }
22
23    /// Pool size from `PHOTON_HANDLER_POOL_SIZE` (default 64).
24    #[must_use]
25    pub fn from_env() -> Self {
26        let max = std::env::var("PHOTON_HANDLER_POOL_SIZE")
27            .ok()
28            .and_then(|s| s.parse().ok())
29            .unwrap_or(64);
30        Self::new(max)
31    }
32
33    /// Acquire a permit before running a handler task.
34    ///
35    /// # Errors
36    ///
37    /// Returns [`PhotonError::Internal`] if the semaphore has been closed.
38    pub async fn acquire(&self) -> Result<tokio::sync::OwnedSemaphorePermit> {
39        self.semaphore
40            .clone()
41            .acquire_owned()
42            .await
43            .map_err(|_| PhotonError::Internal("handler worker pool semaphore closed".into()))
44    }
45}
photon_backend/delivery/worker_pool.rs

photon_backend/delivery/
worker_pool.rs
