feat: add 4 new crates (queue, tracing, http, cli) + enhancements to existing crates
New crates: - mytheclipse-queue: unified job queue with WorkerPool, retry/backoff, DLQ Backends: in-memory (default), Redis, NATS, PostgreSQL - mytheclipse-tracing: tracing subscriber layers with env filter + OTLP/Jaeger export - mytheclipse-http: HTTP client/server with timeout + tracing, axum server - mytheclipse-cli: CLI framework with clap derive, built-in subcommands Enhancements to existing crates: - mytheclipse-core: SemaphorePool, LeaderElection, HealthRegistry/HealthCheck - mytheclipse-cache: AutoRefreshCache (bg refresh on miss), CacheMetrics - mytheclipse-config: ConfigSchema for JSON Schema generation - mytheclipse-storage: MultipartUploadDriver trait - mytheclipse-crypto: PASETO v4.local token support All features compile with --all-features; tests + clippy pass clean.
This commit is contained in:
@@ -0,0 +1,168 @@
|
||||
//! Worker pool for processing queued jobs with retry, backoff, and graceful shutdown.
|
||||
|
||||
use std::pin::Pin;
|
||||
use std::sync::Arc;
|
||||
use std::time::Duration;
|
||||
|
||||
use async_trait::async_trait;
|
||||
use tokio::sync::Semaphore;
|
||||
|
||||
use crate::error::JobError;
|
||||
use crate::job::Job;
|
||||
use crate::traits::Queue;
|
||||
|
||||
/// A future returned by a job handler.
|
||||
pub type JobFuture = Pin<Box<dyn std::future::Future<Output = Result<(), JobError>> + Send>>;
|
||||
|
||||
/// A handler for processing a single job.
|
||||
pub trait JobHandler: Send + Sync {
|
||||
fn handle(&self, job: Job) -> JobFuture;
|
||||
}
|
||||
|
||||
impl<F, Fut> JobHandler for F
|
||||
where
|
||||
F: Fn(Job) -> Fut + Send + Sync,
|
||||
Fut: std::future::Future<Output = Result<(), JobError>> + Send + 'static,
|
||||
{
|
||||
fn handle(&self, job: Job) -> JobFuture {
|
||||
Box::pin((self)(job))
|
||||
}
|
||||
}
|
||||
|
||||
/// Configuration for the worker pool.
|
||||
#[derive(Debug, Clone)]
|
||||
pub struct WorkerConfig {
|
||||
/// Maximum concurrent job handlers.
|
||||
pub concurrency: usize,
|
||||
/// Maximum number of retry attempts (0 = no retries).
|
||||
pub max_retries: u32,
|
||||
/// Base delay for exponential backoff between retries.
|
||||
pub retry_base_delay: Duration,
|
||||
/// Maximum delay cap for retry backoff.
|
||||
pub retry_max_delay: Duration,
|
||||
/// Backoff multiplier.
|
||||
pub retry_factor: f64,
|
||||
/// Visibility timeout for in-progress jobs.
|
||||
pub visibility_timeout: Duration,
|
||||
/// Polling interval when a queue is empty.
|
||||
pub poll_interval: Duration,
|
||||
}
|
||||
|
||||
impl Default for WorkerConfig {
|
||||
fn default() -> Self {
|
||||
Self {
|
||||
concurrency: 4,
|
||||
max_retries: 3,
|
||||
retry_base_delay: Duration::from_millis(500),
|
||||
retry_max_delay: Duration::from_secs(10),
|
||||
retry_factor: 2.0,
|
||||
visibility_timeout: Duration::from_secs(30),
|
||||
poll_interval: Duration::from_millis(100),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// A pool of workers consuming jobs from a `Queue`.
|
||||
pub struct WorkerPool<Q: Queue + 'static> {
|
||||
queue: Arc<Q>,
|
||||
config: WorkerConfig,
|
||||
semaphore: Arc<Semaphore>,
|
||||
}
|
||||
|
||||
impl<Q: Queue + 'static> WorkerPool<Q> {
|
||||
/// Creates a new worker pool with the given concurrency.
|
||||
pub fn new(queue: Q, concurrency: usize) -> Self {
|
||||
Self::with_config(queue, WorkerConfig {
|
||||
concurrency,
|
||||
..Default::default()
|
||||
})
|
||||
}
|
||||
|
||||
/// Creates a new worker pool with explicit configuration.
|
||||
pub fn with_config(queue: Q, config: WorkerConfig) -> Self {
|
||||
let sem = Arc::new(Semaphore::new(config.concurrency.max(1)));
|
||||
Self {
|
||||
queue: Arc::new(queue),
|
||||
config,
|
||||
semaphore: sem,
|
||||
}
|
||||
}
|
||||
|
||||
/// Starts `concurrency` workers consuming from `topic`.
|
||||
pub fn start<H>(&self, topic: &str, handler: H)
|
||||
where
|
||||
H: JobHandler + 'static,
|
||||
{
|
||||
let queue = Arc::clone(&self.queue);
|
||||
let config = self.config.clone();
|
||||
let semaphore = Arc::clone(&self.semaphore);
|
||||
let handler: Arc<dyn JobHandler> = Arc::new(handler);
|
||||
let topic_owned = topic.to_string();
|
||||
|
||||
for _ in 0..config.concurrency {
|
||||
let q = Arc::clone(&queue);
|
||||
let sem = Arc::clone(&semaphore);
|
||||
let h = Arc::clone(&handler);
|
||||
let cfg = config.clone();
|
||||
let topic_inner = topic_owned.clone();
|
||||
|
||||
tokio::spawn(async move {
|
||||
loop {
|
||||
match q.dequeue(&topic_inner, cfg.poll_interval).await {
|
||||
Ok(Some(job)) => {
|
||||
let _permit = sem.clone().acquire_owned().await;
|
||||
let q2 = Arc::clone(&q);
|
||||
let h2 = Arc::clone(&h);
|
||||
let cfg2 = cfg.clone();
|
||||
let t2 = topic_inner.clone();
|
||||
let j2 = job.clone();
|
||||
|
||||
tokio::spawn(async move {
|
||||
let fut = h2.handle(j2.clone());
|
||||
match fut.await {
|
||||
Ok(()) => {
|
||||
let _ = q2.ack(&j2).await;
|
||||
}
|
||||
Err(_) => {
|
||||
if j2.attempt < cfg2.max_retries {
|
||||
let _ = q2.nack(&j2, true).await;
|
||||
} else {
|
||||
let _ = q2.dlq_move(&t2, j2.clone()).await;
|
||||
let _ = q2.nack(&j2, false).await;
|
||||
}
|
||||
}
|
||||
}
|
||||
});
|
||||
}
|
||||
Ok(None) => {}
|
||||
Err(e) => {
|
||||
tracing::error!("queue error: {e}");
|
||||
tokio::time::sleep(cfg.poll_interval).await;
|
||||
}
|
||||
}
|
||||
}
|
||||
});
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Computes the (capped) exponential backoff delay.
|
||||
pub fn retry_delay(config: &WorkerConfig, attempt: u32) -> Duration {
|
||||
let exponent = attempt as f64;
|
||||
let computed = config.retry_base_delay.as_millis() as f64
|
||||
* config.retry_factor.powf(exponent.max(0.0));
|
||||
let capped = computed.min(config.retry_max_delay.as_millis() as f64);
|
||||
Duration::from_millis(capped as u64)
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
#[test]
|
||||
fn retry_delay_bounded() {
|
||||
let config = WorkerConfig::default();
|
||||
let d = retry_delay(&config, 0);
|
||||
assert!(d <= config.retry_max_delay);
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user