148 lines
4.7 KiB
Rust
148 lines
4.7 KiB
Rust
//! Backpressure-aware enqueuer (in-memory backend).
|
|
//!
|
|
//! [`BackpressureEnforcer`] tracks in-flight jobs and caps the number of
|
|
//! pending enqueues per topic, returning [`BackpressureError`] instead of
|
|
//! blocking when the cap is exceeded.
|
|
|
|
use std::sync::atomic::{AtomicU64, Ordering};
|
|
use std::sync::Arc;
|
|
|
|
use tokio::sync::Semaphore;
|
|
|
|
use crate::traits::Queue;
|
|
|
|
/// Errors returned by [`BackpressureEnforcer`].
|
|
#[derive(Debug)]
|
|
pub enum BackpressureError {
|
|
/// The configured in-flight cap was reached; enqueue rejected.
|
|
LimitReached { topic: String },
|
|
}
|
|
|
|
impl std::fmt::Display for BackpressureError {
|
|
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
|
|
match self {
|
|
Self::LimitReached { topic } => {
|
|
write!(f, "backpressure: in-flight limit reached for topic {topic}")
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
impl std::error::Error for BackpressureError {}
|
|
|
|
/// Enforces a maximum number of in-flight jobs per topic.
|
|
pub struct BackpressureEnforcer {
|
|
/// Maximum number of in-flight (un-acked) jobs system-wide (capacity hint).
|
|
#[allow(dead_code)]
|
|
max_inflight: usize,
|
|
/// Per-topic in-flight counter.
|
|
counters: Arc<std::sync::Mutex<std::collections::HashMap<String, Arc<AtomicU64>>>>,
|
|
/// Bounded semaphore enforcing total concurrency.
|
|
#[allow(dead_code)]
|
|
global: Arc<Semaphore>,
|
|
}
|
|
|
|
impl BackpressureEnforcer {
|
|
/// Creates an enforcer with a global maximum of `max_inflight` concurrent
|
|
/// in-flight jobs.
|
|
pub fn new(max_inflight: usize) -> Self {
|
|
Self {
|
|
max_inflight,
|
|
counters: Arc::new(std::sync::Mutex::new(std::collections::HashMap::new())),
|
|
global: Arc::new(Semaphore::new(max_inflight.max(1))),
|
|
}
|
|
}
|
|
|
|
/// Returns the per-topic in-flight count, creating zero if absent.
|
|
fn get_counter(&self, topic: &str) -> Arc<AtomicU64> {
|
|
let mut map = self.counters.lock().unwrap();
|
|
map.entry(topic.to_string())
|
|
.or_insert_with(|| Arc::new(AtomicU64::new(0)));
|
|
Arc::clone(map.get(topic).unwrap())
|
|
}
|
|
|
|
/// Attempts to acquire a backpressure slot non-blockingly.
|
|
/// Returns Err if at capacity.
|
|
pub async fn try_enqueue<Q: Queue + ?Sized>(
|
|
&self,
|
|
queue: &Q,
|
|
topic: &str,
|
|
payload: Vec<u8>,
|
|
) -> Result<(), BackpressureError> {
|
|
// Per-topic counter increment (informational; global semaphore is the hard limit)
|
|
let counter = self.get_counter(topic);
|
|
counter.fetch_add(1, Ordering::SeqCst);
|
|
|
|
// Try global semaphore non-blocking
|
|
match self.global.clone().try_acquire_owned() {
|
|
Ok(_permit) => {
|
|
let _ = queue.enqueue(topic, payload).await;
|
|
Ok(())
|
|
}
|
|
Err(_) => {
|
|
counter.fetch_sub(1, Ordering::SeqCst);
|
|
Err(BackpressureError::LimitReached {
|
|
topic: topic.to_string(),
|
|
})
|
|
}
|
|
}
|
|
}
|
|
|
|
/// Increments the in-flight counter when a job is delivered.
|
|
pub fn inc_delivered(&self, topic: &str) {
|
|
self.get_counter(topic).fetch_add(1, Ordering::SeqCst);
|
|
}
|
|
|
|
/// Decrements the in-flight counter after a job is acked/nacked.
|
|
pub fn dec_finished(&self, topic: &str) {
|
|
self.get_counter(topic).fetch_sub(1, Ordering::SeqCst);
|
|
}
|
|
|
|
/// Current in-flight count for a topic.
|
|
pub fn inflight(&self, topic: &str) -> u64 {
|
|
self.get_counter(topic).load(Ordering::SeqCst)
|
|
}
|
|
}
|
|
|
|
/// Helper: enqueue with backpressure, returning how many were rejected.
|
|
pub async fn enqueue_with_backpressure<Q: Queue + ?Sized>(
|
|
enforcer: &BackpressureEnforcer,
|
|
queue: &Q,
|
|
topic: &str,
|
|
payloads: Vec<Vec<u8>>,
|
|
) -> Result<usize, BackpressureError> {
|
|
let mut rejected = 0;
|
|
for payload in payloads {
|
|
if let Err(_) = enforcer.try_enqueue(queue, topic, payload).await {
|
|
rejected += 1;
|
|
}
|
|
}
|
|
Ok(rejected)
|
|
}
|
|
|
|
#[cfg(test)]
|
|
mod tests {
|
|
use super::*;
|
|
use crate::in_memory::InMemoryQueue;
|
|
|
|
#[tokio::test]
|
|
async fn try_enqueue_within_limit_succeeds() {
|
|
let reg = BackpressureEnforcer::new(2);
|
|
let queue = InMemoryQueue::new();
|
|
let result = reg.try_enqueue(&queue, "t", b"x".to_vec()).await;
|
|
assert!(result.is_ok());
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn try_enqueue_rejects_when_full() {
|
|
let reg = BackpressureEnforcer::new(1);
|
|
let queue = InMemoryQueue::new();
|
|
|
|
// acquire the single global permit without releasing
|
|
let _first = reg.global.clone().acquire_owned().await.unwrap();
|
|
|
|
let result = reg.try_enqueue(&queue, "t", b"x".to_vec()).await;
|
|
assert!(matches!(result, Err(BackpressureError::LimitReached { .. })));
|
|
}
|
|
}
|