Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
3c64563a44 | ||
|
|
2f445d3b85 | ||
|
|
2985fa0a0e | ||
|
|
7a063fa75c |
@@ -0,0 +1,29 @@
|
|||||||
|
# Implementation Spec: Round 10 — COMPLETE
|
||||||
|
|
||||||
|
## Goal
|
||||||
|
Auto-integration + ergonomics: rate-limit workers, auto-metrics on service calls —
|
||||||
|
reduce manual wiring/boilerplate.
|
||||||
|
|
||||||
|
## New Features
|
||||||
|
|
||||||
|
### 1. AutoMetricsServiceBuilder (mytheclipse-core, observability)
|
||||||
|
File: `crates/mytheclipse/src/auto_metrics_service.rs`
|
||||||
|
- Composes ServiceBuilder + MetricsCollector (+ MetricsBridge when resiliency)
|
||||||
|
- `.run()` auto-records: calls_total counter (labelled by outcome ok/err/timeout/
|
||||||
|
circuit_open/rate_limited) + duration histogram; emits bridge when attached
|
||||||
|
- Chainable .with_collector/.with_bridge/.with_builders
|
||||||
|
- 1 test
|
||||||
|
|
||||||
|
### 2. RateLimitedWorkerPool (mytheclipse-queue, in-memory)
|
||||||
|
File: `crates/mytheclipse-queue/src/worker_rate_limited.rs`
|
||||||
|
- Wraps WorkerPool with RateLimitedQueue — token-bucket back-pressured dequeue,
|
||||||
|
prevents workers hammering upstream beyond rate limit
|
||||||
|
- new(queue, worker_cfg, rate_per_sec, burst) + start(topic, handler)
|
||||||
|
- 1 test (construction)
|
||||||
|
|
||||||
|
## Files
|
||||||
|
- new: core/src/auto_metrics_service.rs, queue/src/worker_rate_limited.rs
|
||||||
|
- core/lib.rs: +module+export AutoMetricsServiceBuilder
|
||||||
|
- queue/lib.rs: +module+export RateLimitedWorkerPool (rewrote export block)
|
||||||
|
|
||||||
|
Build: exit 0. Tests: 0 FAILED (86 core pass). Clippy: 0 new warnings.
|
||||||
@@ -1,18 +1,23 @@
|
|||||||
# Implementation Spec: Round 8 — COMPLETE
|
# Round 8 — COMPLETE
|
||||||
|
|
||||||
## New Feature
|
## New Feature
|
||||||
|
### ResilientHttpClient (mytheclipse-http, resilience feature)
|
||||||
### ResilientHttpClient (mytheclipse-http, resilience)
|
- File: `crates/mytheclipse-http/src/resilient_client.rs`
|
||||||
File: `crates/mytheclipse-http/src/resilient_client.rs`
|
- `ResilientClientConfig { timeout, max_attempts, rate_per_sec, rate_burst, circuit_breaker }`
|
||||||
- `ResilientHttpClient` — reqwest Client + ServiceBuilder pipeline (retry/circuit/timeout)
|
- `ResilientHttpClient::new(config)` builds `ServiceBuilder` pipeline
|
||||||
- `ResilientClientConfig` — timeout, max_attempts, rate, circuit_breaker
|
- `send(req)`, `get(url)`, `post(url, body)` — all run through `ServiceBuilder::run`
|
||||||
- `send(req)` / `get(url)` / `post(url, body)` — all run through ServiceBuilder::run
|
- Error type `RunError<Box<dyn std::error::Error + Send + Sync>>`
|
||||||
- Error type `RunError<Box<dyn Error>>` (HttpError alias)
|
- Feature: `resilience = ["dep:reqwest", "dep:tokio", "dep:mytheclipse"]`
|
||||||
- 2 tests
|
- mytheclipse dep now `features=["full"]` (was observability)
|
||||||
|
- 2 tests (config defaults + build)
|
||||||
|
|
||||||
## Modified
|
## Modified
|
||||||
- http/Cargo.toml: +resilience feature, mytheclipse dep features=full
|
- http/Cargo.toml — resilience feature + mytheclipse full features
|
||||||
- http/lib.rs: +module +export
|
- http/lib.rs — module + re-export
|
||||||
- core/lib.rs: pub use RunError, ServiceConfig from service_builder
|
- core/lib.rs — pub use RunError, ServiceConfig (needed by http crate)
|
||||||
|
- error.rs — RateLimit(String) variant (queue crate, round 6 carryover)
|
||||||
|
|
||||||
Build: exit 0. Tests: 0 FAILED. Clippy: 0 new warnings.
|
## Build: exit 0. Tests: 0 FAILED. Clippy: 0 new warnings.
|
||||||
|
|
||||||
|
## Skill created: rust-workspace-abstractions (software-development)
|
||||||
|
Captures feature-gating, cross-crate deps, trait/async patterns, ownership patterns, error types, testing conventions for workspace abstraction authoring.
|
||||||
|
|||||||
@@ -0,0 +1,16 @@
|
|||||||
|
# Implementation Spec: Round 9 — COMPLETE
|
||||||
|
|
||||||
|
## New Feature
|
||||||
|
|
||||||
|
### RetryExt (mytheclipse-core, resiliency)
|
||||||
|
File: `crates/mytheclipse/src/retry_ext.rs`
|
||||||
|
- `RetryExt` trait — `.retry(config, predicate, self_fn)` extension pada Future<Output=Result<T,E>>
|
||||||
|
- Delegasi ke `crate::retry::retry`
|
||||||
|
- Non-Send Pin<Box<...>> return (single-threaded test OK)
|
||||||
|
- 1 test (retries_then_succeeds)
|
||||||
|
|
||||||
|
## Files
|
||||||
|
- new: retry_ext.rs
|
||||||
|
- core/lib.rs: +module +pub use RetryExt
|
||||||
|
|
||||||
|
Build: exit 0. Tests: 0 FAILED. Clippy: 0 new warnings.
|
||||||
@@ -1,3 +1,17 @@
|
|||||||
|
# [1.13.0](https://github.com/asepharyana/mytheclipse/compare/v1.12.0...v1.13.0) (2026-08-29)
|
||||||
|
|
||||||
|
|
||||||
|
### Features
|
||||||
|
|
||||||
|
* round-10 abstractions — AutoMetricsServiceBuilder, RateLimitedWorkerPool ([2f445d3](https://github.com/asepharyana/mytheclipse/commit/2f445d3b8539c89f819224e421a555bc605aac91))
|
||||||
|
|
||||||
|
# [1.12.0](https://github.com/asepharyana/mytheclipse/compare/v1.11.0...v1.12.0) (2026-08-29)
|
||||||
|
|
||||||
|
|
||||||
|
### Features
|
||||||
|
|
||||||
|
* round-9 abstractions — RetryExt ergonomic retry, ResilientHttpClient ([7a063fa](https://github.com/asepharyana/mytheclipse/commit/7a063fa75c6ca243d1e76a485e117a3b334b25e9))
|
||||||
|
|
||||||
# [1.11.0](https://github.com/asepharyana/mytheclipse/compare/v1.10.0...v1.11.0) (2026-08-29)
|
# [1.11.0](https://github.com/asepharyana/mytheclipse/compare/v1.10.0...v1.11.0) (2026-08-29)
|
||||||
|
|
||||||
|
|
||||||
|
|||||||
Generated
+10
-10
@@ -2827,7 +2827,7 @@ dependencies = [
|
|||||||
|
|
||||||
[[package]]
|
[[package]]
|
||||||
name = "mytheclipse"
|
name = "mytheclipse"
|
||||||
version = "1.11.0"
|
version = "1.13.0"
|
||||||
dependencies = [
|
dependencies = [
|
||||||
"async-trait",
|
"async-trait",
|
||||||
"num_cpus",
|
"num_cpus",
|
||||||
@@ -2841,7 +2841,7 @@ dependencies = [
|
|||||||
|
|
||||||
[[package]]
|
[[package]]
|
||||||
name = "mytheclipse-cache"
|
name = "mytheclipse-cache"
|
||||||
version = "1.11.0"
|
version = "1.13.0"
|
||||||
dependencies = [
|
dependencies = [
|
||||||
"async-trait",
|
"async-trait",
|
||||||
"moka",
|
"moka",
|
||||||
@@ -2854,7 +2854,7 @@ dependencies = [
|
|||||||
|
|
||||||
[[package]]
|
[[package]]
|
||||||
name = "mytheclipse-cli"
|
name = "mytheclipse-cli"
|
||||||
version = "1.11.0"
|
version = "1.13.0"
|
||||||
dependencies = [
|
dependencies = [
|
||||||
"clap",
|
"clap",
|
||||||
"tokio",
|
"tokio",
|
||||||
@@ -2863,7 +2863,7 @@ dependencies = [
|
|||||||
|
|
||||||
[[package]]
|
[[package]]
|
||||||
name = "mytheclipse-config"
|
name = "mytheclipse-config"
|
||||||
version = "1.11.0"
|
version = "1.13.0"
|
||||||
dependencies = [
|
dependencies = [
|
||||||
"dotenvy",
|
"dotenvy",
|
||||||
"notify",
|
"notify",
|
||||||
@@ -2878,7 +2878,7 @@ dependencies = [
|
|||||||
|
|
||||||
[[package]]
|
[[package]]
|
||||||
name = "mytheclipse-crypto"
|
name = "mytheclipse-crypto"
|
||||||
version = "1.11.0"
|
version = "1.13.0"
|
||||||
dependencies = [
|
dependencies = [
|
||||||
"aead",
|
"aead",
|
||||||
"aes-gcm",
|
"aes-gcm",
|
||||||
@@ -2900,7 +2900,7 @@ dependencies = [
|
|||||||
|
|
||||||
[[package]]
|
[[package]]
|
||||||
name = "mytheclipse-event"
|
name = "mytheclipse-event"
|
||||||
version = "1.11.0"
|
version = "1.13.0"
|
||||||
dependencies = [
|
dependencies = [
|
||||||
"async-nats",
|
"async-nats",
|
||||||
"async-trait",
|
"async-trait",
|
||||||
@@ -2916,7 +2916,7 @@ dependencies = [
|
|||||||
|
|
||||||
[[package]]
|
[[package]]
|
||||||
name = "mytheclipse-http"
|
name = "mytheclipse-http"
|
||||||
version = "1.11.0"
|
version = "1.13.0"
|
||||||
dependencies = [
|
dependencies = [
|
||||||
"async-trait",
|
"async-trait",
|
||||||
"axum",
|
"axum",
|
||||||
@@ -2932,7 +2932,7 @@ dependencies = [
|
|||||||
|
|
||||||
[[package]]
|
[[package]]
|
||||||
name = "mytheclipse-queue"
|
name = "mytheclipse-queue"
|
||||||
version = "1.11.0"
|
version = "1.13.0"
|
||||||
dependencies = [
|
dependencies = [
|
||||||
"async-nats",
|
"async-nats",
|
||||||
"async-trait",
|
"async-trait",
|
||||||
@@ -2948,7 +2948,7 @@ dependencies = [
|
|||||||
|
|
||||||
[[package]]
|
[[package]]
|
||||||
name = "mytheclipse-storage"
|
name = "mytheclipse-storage"
|
||||||
version = "1.11.0"
|
version = "1.13.0"
|
||||||
dependencies = [
|
dependencies = [
|
||||||
"async-trait",
|
"async-trait",
|
||||||
"aws-config",
|
"aws-config",
|
||||||
@@ -2964,7 +2964,7 @@ dependencies = [
|
|||||||
|
|
||||||
[[package]]
|
[[package]]
|
||||||
name = "mytheclipse-tracing"
|
name = "mytheclipse-tracing"
|
||||||
version = "1.11.0"
|
version = "1.13.0"
|
||||||
dependencies = [
|
dependencies = [
|
||||||
"opentelemetry 0.25.0",
|
"opentelemetry 0.25.0",
|
||||||
"tokio",
|
"tokio",
|
||||||
|
|||||||
@@ -1,6 +1,6 @@
|
|||||||
[package]
|
[package]
|
||||||
name = "mytheclipse-cache"
|
name = "mytheclipse-cache"
|
||||||
version = "1.11.0"
|
version = "1.13.0"
|
||||||
edition = "2021"
|
edition = "2021"
|
||||||
rust-version = "1.75"
|
rust-version = "1.75"
|
||||||
license = "MIT OR Apache-2.0"
|
license = "MIT OR Apache-2.0"
|
||||||
|
|||||||
@@ -1,6 +1,6 @@
|
|||||||
[package]
|
[package]
|
||||||
name = "mytheclipse-cli"
|
name = "mytheclipse-cli"
|
||||||
version = "1.11.0"
|
version = "1.13.0"
|
||||||
edition = "2021"
|
edition = "2021"
|
||||||
rust-version = "1.75"
|
rust-version = "1.75"
|
||||||
license = "MIT OR Apache-2.0"
|
license = "MIT OR Apache-2.0"
|
||||||
|
|||||||
@@ -1,6 +1,6 @@
|
|||||||
[package]
|
[package]
|
||||||
name = "mytheclipse-config"
|
name = "mytheclipse-config"
|
||||||
version = "1.11.0"
|
version = "1.13.0"
|
||||||
edition = "2021"
|
edition = "2021"
|
||||||
rust-version = "1.75"
|
rust-version = "1.75"
|
||||||
license = "MIT OR Apache-2.0"
|
license = "MIT OR Apache-2.0"
|
||||||
|
|||||||
@@ -1,6 +1,6 @@
|
|||||||
[package]
|
[package]
|
||||||
name = "mytheclipse-crypto"
|
name = "mytheclipse-crypto"
|
||||||
version = "1.11.0"
|
version = "1.13.0"
|
||||||
edition = "2021"
|
edition = "2021"
|
||||||
rust-version = "1.75"
|
rust-version = "1.75"
|
||||||
license = "MIT OR Apache-2.0"
|
license = "MIT OR Apache-2.0"
|
||||||
|
|||||||
@@ -1,6 +1,6 @@
|
|||||||
[package]
|
[package]
|
||||||
name = "mytheclipse-event"
|
name = "mytheclipse-event"
|
||||||
version = "1.11.0"
|
version = "1.13.0"
|
||||||
edition = "2021"
|
edition = "2021"
|
||||||
rust-version = "1.75"
|
rust-version = "1.75"
|
||||||
license = "MIT OR Apache-2.0"
|
license = "MIT OR Apache-2.0"
|
||||||
|
|||||||
@@ -1,6 +1,6 @@
|
|||||||
[package]
|
[package]
|
||||||
name = "mytheclipse-http"
|
name = "mytheclipse-http"
|
||||||
version = "1.11.0"
|
version = "1.13.0"
|
||||||
edition = "2021"
|
edition = "2021"
|
||||||
rust-version = "1.75"
|
rust-version = "1.75"
|
||||||
license = "MIT OR Apache-2.0"
|
license = "MIT OR Apache-2.0"
|
||||||
|
|||||||
@@ -1,6 +1,6 @@
|
|||||||
[package]
|
[package]
|
||||||
name = "mytheclipse-queue"
|
name = "mytheclipse-queue"
|
||||||
version = "1.11.0"
|
version = "1.13.0"
|
||||||
edition = "2021"
|
edition = "2021"
|
||||||
rust-version = "1.75"
|
rust-version = "1.75"
|
||||||
license = "MIT OR Apache-2.0"
|
license = "MIT OR Apache-2.0"
|
||||||
|
|||||||
@@ -9,43 +9,6 @@
|
|||||||
//! - **Redis** (`redis`) — LIST-based queue with atomic moves.
|
//! - **Redis** (`redis`) — LIST-based queue with atomic moves.
|
||||||
//! - **NATS JetStream** (`nats`) — durable consumer with ACK/NACK.
|
//! - **NATS JetStream** (`nats`) — durable consumer with ACK/NACK.
|
||||||
//! - **PostgreSQL** (`postgres`) — `SKIP LOCKED` polling.
|
//! - **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 error;
|
||||||
pub mod job;
|
pub mod job;
|
||||||
@@ -61,22 +24,15 @@ pub mod backpressure_enqueue;
|
|||||||
#[cfg(feature = "in-memory")]
|
#[cfg(feature = "in-memory")]
|
||||||
pub mod rate_limited;
|
pub mod rate_limited;
|
||||||
#[cfg(feature = "in-memory")]
|
#[cfg(feature = "in-memory")]
|
||||||
|
pub mod worker_rate_limited;
|
||||||
|
#[cfg(feature = "in-memory")]
|
||||||
pub use backpressure_enqueue::{BackpressureEnforcer, BackpressureError, enqueue_with_backpressure};
|
pub use backpressure_enqueue::{BackpressureEnforcer, BackpressureError, enqueue_with_backpressure};
|
||||||
#[cfg(feature = "in-memory")]
|
#[cfg(feature = "in-memory")]
|
||||||
pub use rate_limited::{RateLimitedQueue, RateLimitQueueError};
|
pub use rate_limited::{RateLimitedQueue, RateLimitQueueError};
|
||||||
|
#[cfg(feature = "in-memory")]
|
||||||
|
pub use worker_rate_limited::RateLimitedWorkerPool;
|
||||||
|
|
||||||
#[cfg(feature = "in-memory")]
|
#[cfg(feature = "in-memory")]
|
||||||
pub mod pipeline;
|
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")]
|
#[cfg(feature = "in-memory")]
|
||||||
pub use pipeline::{StageRunner, Stage, StageError};
|
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
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -1,6 +1,6 @@
|
|||||||
[package]
|
[package]
|
||||||
name = "mytheclipse-storage"
|
name = "mytheclipse-storage"
|
||||||
version = "1.11.0"
|
version = "1.13.0"
|
||||||
edition = "2021"
|
edition = "2021"
|
||||||
rust-version = "1.75"
|
rust-version = "1.75"
|
||||||
license = "MIT OR Apache-2.0"
|
license = "MIT OR Apache-2.0"
|
||||||
|
|||||||
@@ -1,6 +1,6 @@
|
|||||||
[package]
|
[package]
|
||||||
name = "mytheclipse-tracing"
|
name = "mytheclipse-tracing"
|
||||||
version = "1.11.0"
|
version = "1.13.0"
|
||||||
edition = "2021"
|
edition = "2021"
|
||||||
rust-version = "1.75"
|
rust-version = "1.75"
|
||||||
license = "MIT OR Apache-2.0"
|
license = "MIT OR Apache-2.0"
|
||||||
|
|||||||
@@ -1,6 +1,6 @@
|
|||||||
[package]
|
[package]
|
||||||
name = "mytheclipse"
|
name = "mytheclipse"
|
||||||
version = "1.11.0"
|
version = "1.13.0"
|
||||||
edition = "2021"
|
edition = "2021"
|
||||||
rust-version = "1.75"
|
rust-version = "1.75"
|
||||||
license = "MIT OR Apache-2.0"
|
license = "MIT OR Apache-2.0"
|
||||||
|
|||||||
@@ -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);
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -34,6 +34,14 @@ pub mod bg;
|
|||||||
#[cfg(feature = "resiliency")]
|
#[cfg(feature = "resiliency")]
|
||||||
pub mod retry;
|
pub mod retry;
|
||||||
#[cfg(feature = "resiliency")]
|
#[cfg(feature = "resiliency")]
|
||||||
|
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;
|
pub mod circuit_breaker;
|
||||||
#[cfg(feature = "resiliency")]
|
#[cfg(feature = "resiliency")]
|
||||||
pub mod timeout;
|
pub mod timeout;
|
||||||
|
|||||||
@@ -0,0 +1,72 @@
|
|||||||
|
//! Ergonomic retry extension trait (feature `resiliency`).
|
||||||
|
//!
|
||||||
|
//! [`RetryExt`] extends any [`std::future::Future<Output = Result<T, E>>`]
|
||||||
|
//! with a `.retry()` method that delegates to [`crate::retry::retry`].
|
||||||
|
|
||||||
|
use std::future::Future;
|
||||||
|
use std::pin::Pin;
|
||||||
|
|
||||||
|
use crate::retry::{retry, RetryConfig, RetryError};
|
||||||
|
|
||||||
|
/// Extension trait adding ergonomic `.retry()` to any fallible future.
|
||||||
|
pub trait RetryExt<T, E>: Future<Output = Result<T, E>> + Sized + 'static
|
||||||
|
where
|
||||||
|
E: std::fmt::Debug + Send + 'static,
|
||||||
|
T: Send + 'static,
|
||||||
|
{
|
||||||
|
/// Retries the future's result via a reconstructive `self_fn` closure,
|
||||||
|
/// delegating to [`crate::retry::retry`]. The original future is consumed
|
||||||
|
/// on the first attempt; subsequent attempts use `self_fn()`.
|
||||||
|
fn retry<F, Fut, P>(
|
||||||
|
self,
|
||||||
|
config: RetryConfig,
|
||||||
|
predicate: P,
|
||||||
|
self_fn: F,
|
||||||
|
) -> Pin<Box<dyn Future<Output = Result<T, RetryError<E>>>>>
|
||||||
|
where
|
||||||
|
F: FnMut() -> Fut + 'static,
|
||||||
|
Fut: Future<Output = Result<T, E>> + 'static,
|
||||||
|
P: Fn(&E) -> bool + 'static,
|
||||||
|
{
|
||||||
|
Box::pin(async move {
|
||||||
|
let _ = self.await;
|
||||||
|
retry(config, predicate, self_fn).await
|
||||||
|
})
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
impl<Fut, T, E> RetryExt<T, E> for Fut
|
||||||
|
where
|
||||||
|
Fut: Future<Output = Result<T, E>> + Send + 'static,
|
||||||
|
E: std::fmt::Debug + Send + 'static,
|
||||||
|
T: Send + 'static,
|
||||||
|
{
|
||||||
|
}
|
||||||
|
|
||||||
|
#[cfg(test)]
|
||||||
|
mod tests {
|
||||||
|
use super::*;
|
||||||
|
use std::time::Duration;
|
||||||
|
use std::sync::Arc;
|
||||||
|
use std::sync::atomic::{AtomicU32, Ordering};
|
||||||
|
|
||||||
|
#[tokio::test]
|
||||||
|
async fn retry_ext_retries_then_succeeds() {
|
||||||
|
let attempts = Arc::new(AtomicU32::new(0));
|
||||||
|
let a = Arc::clone(&attempts);
|
||||||
|
let cfg = RetryConfig { max_attempts: 3, base_delay: Duration::from_millis(1), ..RetryConfig::default() };
|
||||||
|
|
||||||
|
let op = move || {
|
||||||
|
let a = Arc::clone(&a);
|
||||||
|
async move {
|
||||||
|
let n = a.fetch_add(1, Ordering::SeqCst);
|
||||||
|
if n < 2 { Err::<(), String>("transient".into()) } else { Ok(()) }
|
||||||
|
}
|
||||||
|
};
|
||||||
|
|
||||||
|
let fut = async { Err::<(), String>("first".into()) };
|
||||||
|
let result = fut.retry(cfg, |_: &String| true, op).await;
|
||||||
|
assert!(result.is_ok());
|
||||||
|
assert_eq!(attempts.load(Ordering::SeqCst), 3);
|
||||||
|
}
|
||||||
|
}
|
||||||
Reference in New Issue
Block a user