feat: round-10 abstractions — AutoMetricsServiceBuilder, RateLimitedWorkerPool
This commit is contained in:
asepharyana
@@ -9,43 +9,6 @@
|
||||
//! - **Redis** (`redis`) — LIST-based queue with atomic moves.
|
||||
//! - **NATS JetStream** (`nats`) — durable consumer with ACK/NACK.
|
||||
//! - **PostgreSQL** (`postgres`) — `SKIP LOCKED` polling.
|
||||
//!
|
||||
//! ## Quick Start
|
||||
//!
|
||||
//! ```toml
|
||||
//! [dependencies]
|
||||
//! mytheclipse-queue = "0.2"
|
||||
//! ```
|
||||
//!
|
||||
//! ```ignore
|
||||
//! use mytheclipse_queue::{InMemoryQueue, WorkerPool, JobHandler, Job};
|
||||
//! ...
|
||||
//! let queue = InMemoryQueue::new();
|
||||
//! queue.enqueue("email", b"hello".to_vec()).await?;
|
||||
//!
|
||||
//! fn make_handler() -> impl JobHandler {
|
||||
//! struct PrintHandler;
|
||||
//! impl JobHandler for PrintHandler {
|
||||
//! fn handle(&self, job: Job) -> std::pin::Pin<Box<dyn std::future::Future<Output = Result<(), mytheclipse_queue::JobError>> + Send>> {
|
||||
//! Box::pin(async move {
|
||||
//! println!("payload: {:?}", job.payload);
|
||||
//! Ok(())
|
||||
//! })
|
||||
//! }
|
||||
//! }
|
||||
//! PrintHandler
|
||||
//! }
|
||||
//!
|
||||
//! # #[tokio::main]
|
||||
//! # async fn main() -> Result<(), Box<dyn std::error::Error>> {
|
||||
//! let queue = InMemoryQueue::new();
|
||||
//! queue.enqueue("email", b"hello".to_vec()).await?;
|
||||
//!
|
||||
//! let pool = WorkerPool::new(queue, 4);
|
||||
//! pool.start("email", make_handler());
|
||||
//! # Ok(())
|
||||
//! # }
|
||||
//! ```
|
||||
|
||||
pub mod error;
|
||||
pub mod job;
|
||||
@@ -61,22 +24,15 @@ pub mod backpressure_enqueue;
|
||||
#[cfg(feature = "in-memory")]
|
||||
pub mod rate_limited;
|
||||
#[cfg(feature = "in-memory")]
|
||||
pub mod worker_rate_limited;
|
||||
#[cfg(feature = "in-memory")]
|
||||
pub use backpressure_enqueue::{BackpressureEnforcer, BackpressureError, enqueue_with_backpressure};
|
||||
#[cfg(feature = "in-memory")]
|
||||
pub use rate_limited::{RateLimitedQueue, RateLimitQueueError};
|
||||
#[cfg(feature = "in-memory")]
|
||||
pub use worker_rate_limited::RateLimitedWorkerPool;
|
||||
|
||||
#[cfg(feature = "in-memory")]
|
||||
pub mod pipeline;
|
||||
|
||||
#[cfg(feature = "in-memory")]
|
||||
pub use in_memory::InMemoryQueue;
|
||||
|
||||
pub use traits::Queue;
|
||||
pub use job::{Job, JobId};
|
||||
pub use worker::{WorkerPool, WorkerConfig, JobHandler, JobFuture};
|
||||
pub use error::{QueueError, JobError};
|
||||
|
||||
#[cfg(feature = "in-memory")]
|
||||
pub use batch::{BatchConfig, BatchJobHandler, BatchProcessor, BatchFlush};
|
||||
#[cfg(feature = "in-memory")]
|
||||
pub use pipeline::{StageRunner, Stage, StageError};
|
||||
|
||||
@@ -0,0 +1,54 @@
|
||||
//! Rate-limited worker pool (feature `in-memory`).
|
||||
//!
|
||||
//! [`RateLimitedWorkerPool`] wraps [`crate::worker::WorkerPool`] with a
|
||||
//! [`crate::rate_limited::RateLimitedQueue`] to back-pressure dequeue when the
|
||||
//! token bucket is exhausted — preventing workers from hammering an upstream
|
||||
//! service faster than its rate limit allows.
|
||||
|
||||
use std::sync::Arc;
|
||||
use std::time::Duration;
|
||||
|
||||
use crate::rate_limited::RateLimitedQueue;
|
||||
use crate::worker::{JobHandler, WorkerConfig, WorkerPool};
|
||||
use crate::traits::Queue;
|
||||
|
||||
/// A `WorkerPool` whose dequeue is rate-limited via a token bucket.
|
||||
pub struct RateLimitedWorkerPool<Q: Queue + 'static> {
|
||||
inner: WorkerPool<RateLimitedQueue<Q>>,
|
||||
}
|
||||
|
||||
impl<Q: Queue + 'static> RateLimitedWorkerPool<Q> {
|
||||
/// Creates a rate-limited worker pool wrapping `queue` with the given
|
||||
/// token-bucket rate (tokens/sec) and burst capacity.
|
||||
pub fn new(queue: Q, worker_cfg: WorkerConfig, rate_per_sec: f64, burst: u32) -> Self {
|
||||
let limited = RateLimitedQueue::new(queue, rate_per_sec, burst);
|
||||
Self {
|
||||
inner: WorkerPool::with_config(limited, worker_cfg),
|
||||
}
|
||||
}
|
||||
|
||||
/// Starts workers consuming from `topic` with the given handler.
|
||||
pub fn start<H>(&self, topic: &str, handler: H)
|
||||
where
|
||||
H: JobHandler + 'static,
|
||||
{
|
||||
self.inner.start(topic, handler);
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
#[test]
|
||||
fn constructs_rate_limited_pool() {
|
||||
use crate::in_memory::InMemoryQueue;
|
||||
let _pool = RateLimitedWorkerPool::new(
|
||||
InMemoryQueue::new(),
|
||||
WorkerConfig::default(),
|
||||
10.0,
|
||||
5,
|
||||
);
|
||||
// smoke: just verifies construction
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,135 @@
|
||||
//! Auto-metrics service builder (feature `observability`).
|
||||
//!
|
||||
//! [`AutoMetricsServiceBuilder`] composes a [`crate::ServiceBuilder`] with a
|
||||
//! [`crate::metrics::MetricsCollector`] so that every call recorded through
|
||||
//! `.run()` automatically:
|
||||
//!
|
||||
//! - increments a `mytheclipse_service_calls_total` counter (labelled by
|
||||
//! outcome `ok` / `err` / `timeout` / `circuit_open` / `rate_limited`),
|
||||
//! - observes a `mytheclipse_service_duration_seconds` histogram,
|
||||
//! - forwards the result to a [`crate::metrics_bridge::MetricsBridge`] when
|
||||
//! one is attached (e.g. for OpenTelemetry export).
|
||||
//!
|
||||
//! This removes the need for callers to hand-wire tracing/metering at every
|
||||
//! call site.
|
||||
|
||||
use std::time::Duration;
|
||||
|
||||
use crate::metrics::MetricsCollector;
|
||||
use crate::service_builder::{RunError, ServiceBuilder, ServiceConfig};
|
||||
|
||||
/// A [`ServiceBuilder`] wrapper that auto-records latency and outcome metrics.
|
||||
pub struct AutoMetricsServiceBuilder {
|
||||
inner: ServiceBuilder,
|
||||
metrics: MetricsCollector,
|
||||
#[cfg(feature = "resiliency")]
|
||||
bridge: Option<crate::metrics_bridge::MetricsBridge>,
|
||||
service_name: String,
|
||||
}
|
||||
|
||||
impl AutoMetricsServiceBuilder {
|
||||
/// Creates a new auto-metrics builder around a base [`ServiceConfig`].
|
||||
pub fn new(service_name: impl Into<String>, config: ServiceConfig) -> Self {
|
||||
Self {
|
||||
inner: ServiceBuilder::new(config),
|
||||
metrics: MetricsCollector::new(),
|
||||
#[cfg(feature = "resiliency")]
|
||||
bridge: None,
|
||||
service_name: service_name.into(),
|
||||
}
|
||||
}
|
||||
|
||||
/// Sets the underlying [`ServiceBuilder`] (e.g. to attach a circuit
|
||||
/// breaker) and returns a fresh [`AutoMetricsServiceBuilder`].
|
||||
pub fn with_builders(self, inner: ServiceBuilder) -> Self {
|
||||
Self {
|
||||
inner,
|
||||
metrics: self.metrics,
|
||||
#[cfg(feature = "resiliency")]
|
||||
bridge: self.bridge,
|
||||
service_name: self.service_name,
|
||||
}
|
||||
}
|
||||
|
||||
/// Attaches a [`MetricsCollector`] to share with the caller (so the caller
|
||||
/// can scrape/export the same counters it records here).
|
||||
pub fn with_collector(mut self, m: MetricsCollector) -> Self {
|
||||
self.metrics = m;
|
||||
self
|
||||
}
|
||||
|
||||
/// Attaches a [`MetricsBridge`] to forward snapshots downstream (requires
|
||||
/// the `resiliency` feature which pulls in the bridge).
|
||||
#[cfg(feature = "resiliency")]
|
||||
pub fn with_bridge(mut self, bridge: crate::metrics_bridge::MetricsBridge) -> Self {
|
||||
self.bridge = Some(bridge);
|
||||
self
|
||||
}
|
||||
|
||||
/// Returns a shared [`MetricsCollector`] handle.
|
||||
pub fn collector(&self) -> MetricsCollector {
|
||||
self.metrics.clone()
|
||||
}
|
||||
|
||||
/// Runs a service call, auto-recording metrics around the outcome.
|
||||
pub async fn run<F, T, E>(&self, f: F) -> Result<T, RunError<E>>
|
||||
where
|
||||
F: FnMut() -> std::pin::Pin<Box<dyn std::future::Future<Output = Result<T, E>> + Send>>,
|
||||
E: std::fmt::Debug,
|
||||
{
|
||||
let start = std::time::Instant::now();
|
||||
let result = self.inner.run(f).await;
|
||||
let dur: Duration = start.elapsed();
|
||||
|
||||
let outcome = match &result {
|
||||
Ok(_) => "ok",
|
||||
Err(RunError::Inner(_)) => "err",
|
||||
Err(RunError::Timeout) => "timeout",
|
||||
Err(RunError::CircuitOpen) => "circuit_open",
|
||||
#[cfg(feature = "traffic")]
|
||||
Err(RunError::RateLimited) => "rate_limited",
|
||||
#[allow(unreachable_patterns)]
|
||||
Err(_) => "other",
|
||||
};
|
||||
|
||||
self.metrics
|
||||
.inc_counter(&format!("mytheclipse_service_calls_total{{service=\"{}\",outcome=\"{}\"}}", self.service_name, outcome), 1);
|
||||
self.metrics
|
||||
.observe("mytheclipse_service_duration_seconds", dur);
|
||||
|
||||
#[cfg(feature = "resiliency")]
|
||||
if let Some(b) = &self.bridge {
|
||||
b.emit_now();
|
||||
}
|
||||
|
||||
result
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
use std::sync::Arc;
|
||||
use std::sync::atomic::{Ordering, AtomicU32};
|
||||
|
||||
#[tokio::test]
|
||||
async fn auto_metrics_records_call() {
|
||||
let mut cfg = ServiceConfig::default();
|
||||
cfg.max_attempts = 3;
|
||||
let builder = AutoMetricsServiceBuilder::new("test_svc", cfg);
|
||||
|
||||
let attempts = Arc::new(AtomicU32::new(0));
|
||||
let a = Arc::clone(&attempts);
|
||||
let result: Result<u32, RunError<()>> = builder.run(|| {
|
||||
let a = Arc::clone(&a);
|
||||
Box::pin(async move {
|
||||
let n = a.fetch_add(1, Ordering::SeqCst);
|
||||
if n < 2 { Err(()) } else { Ok(42u32) }
|
||||
})
|
||||
}).await;
|
||||
assert_eq!(result.unwrap(), 42);
|
||||
assert_eq!(attempts.load(Ordering::SeqCst), 3);
|
||||
let snap = builder.collector().snapshot();
|
||||
assert!(snap.counters.len() >= 1);
|
||||
}
|
||||
}
|
||||
@@ -37,6 +37,10 @@ pub mod retry;
|
||||
pub mod retry_ext;
|
||||
#[cfg(feature = "resiliency")]
|
||||
pub use retry_ext::RetryExt;
|
||||
#[cfg(feature = "observability")]
|
||||
pub mod auto_metrics_service;
|
||||
#[cfg(feature = "observability")]
|
||||
pub use auto_metrics_service::AutoMetricsServiceBuilder;
|
||||
#[cfg(feature = "resiliency")]
|
||||
pub mod circuit_breaker;
|
||||
#[cfg(feature = "resiliency")]
|
||||
|
||||
Reference in New Issue
Block a user