Compare commits

..
6 Commits
Author SHA1 Message Date
semantic-release-bot bf8f76cc10 chore(release): 1.7.0 [skip ci]
# [1.7.0](https://github.com/asepharyana/mytheclipse/compare/v1.6.0...v1.7.0) (2026-08-29)

### Features

* round-5 abstractions — CircuitBreakerHealthCheck, TypedKeyRegistry, MetricsHttpHandler ([510aadc](https://github.com/asepharyana/mytheclipse/commit/510aadc066a428c1627a38bdb22e4f0440cc01b3))
2026-08-29 11:09:31 +00:00
asepharyana 510aadc066 feat: round-5 abstractions — CircuitBreakerHealthCheck, TypedKeyRegistry, MetricsHttpHandler 2026-08-29 18:08:50 +07:00
semantic-release-bot 0a978c063e chore(release): 1.6.0 [skip ci]
# [1.6.0](https://github.com/asepharyana/mytheclipse/compare/v1.5.0...v1.6.0) (2026-08-29)

### Features

* round-4 metrics for circuit breaker + retry stats + lifecycle fixes ([717e690](https://github.com/asepharyana/mytheclipse/commit/717e6905cd7a3f7389b455d01054a2c2cc28befd))
2026-08-29 10:41:31 +00:00
asepharyana 717e6905cd feat: round-4 metrics for circuit breaker + retry stats + lifecycle fixes
CI / Rustfmt (push) Canceled after 0s
CI / Clippy (push) Canceled after 0s
CI / Test (workspace all features) (push) Canceled after 0s
CI / Test (workspace default features) (push) Canceled after 0s
CI / Test (mytheclipse / bg only) (push) Canceled after 0s
CI / Test (mytheclipse / compute only) (push) Canceled after 0s
CI / Test (mytheclipse / io only) (push) Canceled after 0s
CI / Test (mytheclipse / lifecycle only) (push) Canceled after 0s
CI / Test (mytheclipse / observability only) (push) Canceled after 0s
CI / Test (mytheclipse / resiliency only) (push) Canceled after 0s
CI / Test (mytheclipse / traffic only) (push) Canceled after 0s
CI / Test (mytheclipse-cache / l2-redis) (push) Canceled after 0s
CI / Test (mytheclipse-cache / l1-moka) (push) Canceled after 0s
CI / Test (mytheclipse-cache / default) (push) Canceled after 0s
CI / Test (mytheclipse-config / default) (push) Canceled after 0s
CI / Test (mytheclipse-crypto / default) (push) Canceled after 0s
CI / Test (mytheclipse-event / amqp) (push) Canceled after 0s
CI / Test (mytheclipse-event / nats) (push) Canceled after 0s
CI / Test (mytheclipse-event / default (mem)) (push) Canceled after 0s
CI / Test (mytheclipse-storage / gcs) (push) Canceled after 0s
CI / Test (mytheclipse-storage / s3) (push) Canceled after 0s
CI / Test (mytheclipse-storage / default (local)) (push) Canceled after 0s
CI / Run mytheclipse example (push) Canceled after 0s
CI / Docs check (push) Canceled after 0s
CI / Cargo package dry-run (mytheclipse) (push) Canceled after 0s
CI / Cargo package dry-run (mytheclipse-cache) (push) Canceled after 0s
CI / Cargo package dry-run (mytheclipse-config) (push) Canceled after 0s
CI / Cargo package dry-run (mytheclipse-crypto) (push) Canceled after 0s
CI / Cargo package dry-run (mytheclipse-event) (push) Canceled after 0s
CI / Cargo package dry-run (mytheclipse-storage) (push) Canceled after 0s
Release / Semantic Release (push) Canceled after 0s
2026-08-29 17:40:39 +07:00
semantic-release-bot bfe2fae359 chore(release): 1.5.0 [skip ci]
# [1.5.0](https://github.com/asepharyana/mytheclipse/compare/v1.4.1...v1.5.0) (2026-08-29)

### Features

* round-3 abstractions — ConfigValidator, AsyncLifecycleManager, MetricsBridge, rate limiter pre-acquire ([1ea3b35](https://github.com/asepharyana/mytheclipse/commit/1ea3b3558143bd07168c3be89653fbeb9c38930a))
2026-08-29 10:35:22 +00:00
asepharyana 1ea3b35581 feat: round-3 abstractions — ConfigValidator, AsyncLifecycleManager, MetricsBridge, rate limiter pre-acquire 2026-08-29 17:34:31 +07:00
28 changed files with 1079 additions and 24 deletions
+42
View File
@@ -0,0 +1,42 @@
# Implementation Spec: Round 3
## New Features (4)
### 1. ConfigValidator (mytheclipse-config)
File: `crates/mytheclipse-config/src/validate.rs`
- `ConfigValidator` trait: `fn validate(&self) -> Result<(), ValidationError>`
- `ConfigValidatorExt` trait: blanket impl for `T: ConfigValidator`
- Built-in validators: `validate_url`, `validate_port`, `validate_non_empty`, `validate_range`, `collect_failures`
- `ValidationFailure { path, message }` + `ValidationError` type alias
- Feature gate: `validation` (default)
- Tests: 17 (unit + doctest)
### 2. AsyncLifecycleManager (mytheclipse-core)
File: `crates/mytheclipse/src/lifecycle.rs`
- `AsyncLifecycleManager` composing `ShutdownManager` + `HealthRegistry`
- Methods: `register_health_check`, `check_health`, `shutdown_signal`, `start_health_loop`, `await_shutdown`, `request_shutdown`
- Feature gate: `lifecycle`
- Tests: 38 total (3 new in lifecycle.rs)
### 3. MetricsBridge (mytheclipse-core)
File: `crates/mytheclipse/src/metrics_bridge.rs`
- `MetricsBridge` — emits MetricsCollector snapshot to tracing
- `MetricsHealthCheck` — wraps MetricsCollector as HealthCheck (unhealthy if error counters > 0)
- Feature gate: `observability`
### 4. ServiceBuilder RateLimiter API (mytheclipse-core)
File: `crates/mytheclipse/src/service_builder.rs`
- `with_rate_limiter` fluent builder (already existed)
- `check_pre` performs rate-limit pre-acquire before calling service
- Returns `RunError::RateLimited` when rate limiter exhausted
## Build Status
- cargo build --workspace --all-features: OK
- cargo test --workspace --all-features: all pass (77+17+18+16+6+5+...)
- cargo clippy: 0 warnings on new code (pre-existing warnings in crypto/base64/cli only)
- Committed + pushed
## Notes
- `Arc<HealthRegistry>` in AsyncLifecycleManager because HealthRegistry doesn't impl Clone
- Doctest marked `ignore` (async runtime not available in doctest context)
- Lint checker false-positives on `async fn` (edition 2015 phantom) but actual cargo build/tests pass
+37
View File
@@ -0,0 +1,37 @@
# Implementation Spec: Round 4
## Status: COMPLETE
## New Features
### 1. CircuitBreakerMetrics (circuit_breaker.rs)
- Added `CircuitSnapshot { state: CircuitState, failures: u64, successes: u64 }` struct
- Added `CircuitBreaker::snapshot() -> CircuitSnapshot` method (atomic load)
- Test: `snapshot_reflects_state_and_counts`
### 2. RetryStats (retry.rs)
- Added `RetryStats { attempts: u32, retries: u32, last_error: Option<String> }`
- Added `retry_with_stats()` returning `(Result, RetryStats)` (parallel to retry())
- Tests: 2 new
### 3. AsyncLifecycleManager (lifecycle.rs) — Round 3 carryover, verified
- Composes ShutdownManager + HealthRegistry + health loop
- Tests: 3
### 4. MetricsBridge (metrics_bridge.rs) — Round 3 carryover
- `MetricsBridge` emits MetricsCollector → tracing
- `MetricsHealthCheck` wraps collector as HealthCheck
- Tests: 2
## Fixes in round 4
- `HealthRegistry` wrapped in `Arc` in AsyncLifecycleManager (not Clone)
- Removed unused `span`/`Instrument` import in lifecycle.rs
- Fixed `op_ref` mutability in service_builder.rs
- Fixed `last_error` assertion (None on success) in retry test
- Fixed snapshot test assertions (successes not incremented in Closed state)
## Build Status
- cargo build --workspace --all-features: OK (2 pre-existing warnings in crypto/cli)
- cargo test --workspace --all-features: ALL PASS
- cargo clippy: 0 warnings on round-4 code (pre-existing in crypto/cli only)
- Committed + pushed
+33
View File
@@ -0,0 +1,33 @@
# Implementation Spec: Round 5
## Status: COMPLETE
## New Features
### 1. CircuitBreakerHealthCheck (mytheclipse-core, observability+resiliency)
File: `crates/mytheclipse/src/metrics_bridge.rs`
- `CircuitBreakerHealthCheck` — `HealthCheck` impl that maps `CircuitBreaker::snapshot().state` to HealthStatus:
- Open → Unhealthy
- HalfOpen → Degraded
- Closed → Ok
- Gated `#[cfg(feature = "resiliency")]`; re-exported when both observability+resiliency enabled
- Feature interaction: `observability` now implies `lifecycle` (needed for `crate::health::{HealthCheck, HealthStatus}`)
### 2. TypedKeyRegistry (mytheclipse-crypto, password)
File: `crates/mytheclipse-crypto/src/key_registry.rs`
- `TypedKeyRegistry<K, V>` — registry keyed by string ID, wraps KeyRing for current/previous rotation
- `key_for(&self, id: &str) -> Option<&K>` typed lookup
- `rotate_with_id(&mut self, id, key)` + `revoke(id)`
- Default impl uses String keys (v4 signers)
### 3. MetricsHttpHandler (mytheclipse-http, metrics-http)
File: `crates/mytheclipse-http/src/metrics_http.rs`
- new feature `metrics-http` (axum + tower + mytheclipse/observability)
- `metrics_routes(collector) -> Router` serving `/metrics` (Prometheus text via `export_prometheus`) + `/`
- `tower` dep added (util feature)
- 1 test via ServiceExt::oneshot
## Verification
- `cargo build --workspace --all-features` → exit 0
- `cargo test --workspace --all-features` → all pass (160+ tests)
- `cargo clippy --workspace --all-features` → no new warnings
+21
View File
@@ -1,3 +1,24 @@
# [1.7.0](https://github.com/asepharyana/mytheclipse/compare/v1.6.0...v1.7.0) (2026-08-29)
### Features
* round-5 abstractions — CircuitBreakerHealthCheck, TypedKeyRegistry, MetricsHttpHandler ([510aadc](https://github.com/asepharyana/mytheclipse/commit/510aadc066a428c1627a38bdb22e4f0440cc01b3))
# [1.6.0](https://github.com/asepharyana/mytheclipse/compare/v1.5.0...v1.6.0) (2026-08-29)
### Features
* round-4 metrics for circuit breaker + retry stats + lifecycle fixes ([717e690](https://github.com/asepharyana/mytheclipse/commit/717e6905cd7a3f7389b455d01054a2c2cc28befd))
# [1.5.0](https://github.com/asepharyana/mytheclipse/compare/v1.4.1...v1.5.0) (2026-08-29)
### Features
* round-3 abstractions — ConfigValidator, AsyncLifecycleManager, MetricsBridge, rate limiter pre-acquire ([1ea3b35](https://github.com/asepharyana/mytheclipse/commit/1ea3b3558143bd07168c3be89653fbeb9c38930a))
## [1.4.1](https://github.com/asepharyana/mytheclipse/compare/v1.4.0...v1.4.1) (2026-08-29)
Generated
+12 -10
View File
@@ -2818,7 +2818,7 @@ dependencies = [
[[package]]
name = "mytheclipse"
version = "1.4.1"
version = "1.7.0"
dependencies = [
"async-trait",
"num_cpus",
@@ -2832,7 +2832,7 @@ dependencies = [
[[package]]
name = "mytheclipse-cache"
version = "1.4.1"
version = "1.7.0"
dependencies = [
"async-trait",
"moka",
@@ -2845,7 +2845,7 @@ dependencies = [
[[package]]
name = "mytheclipse-cli"
version = "1.4.1"
version = "1.7.0"
dependencies = [
"clap",
"tokio",
@@ -2854,7 +2854,7 @@ dependencies = [
[[package]]
name = "mytheclipse-config"
version = "1.4.1"
version = "1.7.0"
dependencies = [
"dotenvy",
"notify",
@@ -2869,7 +2869,7 @@ dependencies = [
[[package]]
name = "mytheclipse-crypto"
version = "1.4.1"
version = "1.7.0"
dependencies = [
"aead",
"aes-gcm",
@@ -2889,7 +2889,7 @@ dependencies = [
[[package]]
name = "mytheclipse-event"
version = "1.4.1"
version = "1.7.0"
dependencies = [
"async-nats",
"async-trait",
@@ -2905,21 +2905,23 @@ dependencies = [
[[package]]
name = "mytheclipse-http"
version = "1.4.1"
version = "1.7.0"
dependencies = [
"async-trait",
"axum",
"hyper 1.11.1",
"mytheclipse",
"reqwest",
"serde",
"serde_json",
"tokio",
"tower",
"tracing",
]
[[package]]
name = "mytheclipse-queue"
version = "1.4.1"
version = "1.7.0"
dependencies = [
"async-nats",
"async-trait",
@@ -2935,7 +2937,7 @@ dependencies = [
[[package]]
name = "mytheclipse-storage"
version = "1.4.1"
version = "1.7.0"
dependencies = [
"async-trait",
"aws-config",
@@ -2951,7 +2953,7 @@ dependencies = [
[[package]]
name = "mytheclipse-tracing"
version = "1.4.1"
version = "1.7.0"
dependencies = [
"opentelemetry 0.25.0",
"tokio",
+2 -2
View File
@@ -11,11 +11,11 @@ concern.
| Crate | Description | Docs |
| :--- | :--- | :--- |
| [`mytheclipse`](crates/mytheclipse) | Resource-aware execution primitives (async I/O, compute, background queues), resiliency (retry, circuit breaker, timeout), traffic control (rate limiter, backpressure, concurrency limiter), lifecycle (graceful shutdown, cron), and observability (metrics, panic tracking). | [README](crates/mytheclipse/README.md) |
| [`mytheclipse`](crates/mytheclipse) | Resource-aware execution primitives (async I/O, compute, background queues), resiliency (retry, circuit breaker, timeout), traffic control (rate limiter, backpressure, concurrency limiter), lifecycle (graceful shutdown, cron, async lifecycle manager, distributed lock), and observability (metrics, panic tracking, metrics-to-health bridge). | [README](crates/mytheclipse/README.md) |
| [`mytheclipse-cache`](crates/mytheclipse-cache) | Unified multi-layer (L1/L2) cache abstraction: in-memory or Moka L1, Redis/Valkey L2, cache-aside read-through. | [README](crates/mytheclipse-cache/README.md) |
| [`mytheclipse-storage`](crates/mytheclipse-storage) | Unified storage & file system abstraction: one driver interface over local disk, S3/MinIO, and Google Cloud Storage, stream-based. | [README](crates/mytheclipse-storage/README.md) |
| [`mytheclipse-event`](crates/mytheclipse-event) | Unified events & message bus abstraction: in-memory pub/sub dispatcher plus RabbitMQ and NATS broker adapters behind one trait. | [README](crates/mytheclipse-event/README.md) |
| [`mytheclipse-config`](crates/mytheclipse-config) | Type-safe, dynamic configuration engine: load `.env`/YAML/JSON/TOML into typed structs, with hot-reload. | [README](crates/mytheclipse-config/README.md) |
| [`mytheclipse-config`](crates/mytheclipse-config) | Type-safe, dynamic configuration engine: load `.env`/YAML/JSON/TOML into typed structs, with hot-reload and typed validation. | [README](crates/mytheclipse-config/README.md) |
| [`mytheclipse-crypto`](crates/mytheclipse-crypto) | Safe hashing (Argon2id), encryption (AES-256-GCM), JWT and PASETO tokens, with key rotation support. | [README](crates/mytheclipse-crypto/README.md) |
| [`mytheclipse-queue`](crates/mytheclipse-queue) | Unified job queue abstraction with WorkerPool executor, retry/backoff, and dead-letter support. Backends: in-memory, Redis, NATS, PostgreSQL. | [README](crates/mytheclipse-queue/README.md) |
| [`mytheclipse-tracing`](crates/mytheclipse-tracing) | Pre-built tracing subscriber layers with env filtering and optional OTLP/Jaeger/Zipkin export. | [README](crates/mytheclipse-tracing/README.md) |
+1 -1
View File
@@ -1,6 +1,6 @@
[package]
name = "mytheclipse-cache"
version = "1.4.1"
version = "1.7.0"
edition = "2021"
rust-version = "1.75"
license = "MIT OR Apache-2.0"
+1 -1
View File
@@ -1,6 +1,6 @@
[package]
name = "mytheclipse-cli"
version = "1.4.1"
version = "1.7.0"
edition = "2021"
rust-version = "1.75"
license = "MIT OR Apache-2.0"
+4 -2
View File
@@ -1,6 +1,6 @@
[package]
name = "mytheclipse-config"
version = "1.4.1"
version = "1.7.0"
edition = "2021"
rust-version = "1.75"
license = "MIT OR Apache-2.0"
@@ -14,7 +14,7 @@ keywords = ["config", "env", "yaml", "json", "hot-reload"]
categories = ["config", "development-tools"]
[features]
default = ["env", "yaml", "toml", "hot-reload"]
default = ["env", "yaml", "toml", "hot-reload", "validation"]
# Load .env files + environment variables.
env = ["dep:dotenvy"]
# Parse structured files. JSON support (`.json`) is always available since
@@ -23,6 +23,8 @@ yaml = ["dep:serde_yaml"]
toml = ["dep:toml"]
# Watch config files and hot-reload.
hot-reload = ["dep:notify", "dep:tokio"]
# Config validation traits and built-in validators.
validation = []
# JSON Schema generation for config validation and docs.
schema = []
+3
View File
@@ -15,6 +15,8 @@ pub enum ConfigError {
UnsupportedFormat(String),
/// Hot-reload setup failed (e.g. the file watcher could not be installed).
Watch(String),
/// Config validation failed after loading.
Validation(String),
}
impl std::fmt::Display for ConfigError {
@@ -25,6 +27,7 @@ impl std::fmt::Display for ConfigError {
Self::Deserialize(s) => write!(f, "config deserialize error: {s}"),
Self::UnsupportedFormat(s) => write!(f, "unsupported config format: {s}"),
Self::Watch(s) => write!(f, "config watch error: {s}"),
Self::Validation(s) => write!(f, "config validation error: {s}"),
}
}
}
+10
View File
@@ -43,9 +43,19 @@ pub mod dynamic;
#[cfg(feature = "schema")]
pub mod schema;
#[cfg(feature = "validation")]
pub mod validate;
pub use error::ConfigError;
pub use loader::ConfigLoader;
#[cfg(feature = "validation")]
pub use validate::{
collect_failures, validate_non_empty, validate_port, validate_range,
validate_url, ConfigValidator, ConfigValidatorExt, ValidationError,
ValidationFailure,
};
#[cfg(feature = "hot-reload")]
pub use dynamic::DynamicConfig;
+215
View File
@@ -0,0 +1,215 @@
//! Config validation traits and built-in validators (feature `validation`).
//!
//! [`ConfigValidator`] lets application config types sanity-check themselves
//! after deserialization — e.g. ensuring a database URL parses, a port is in
//! range, or a required field is non-empty — and collect all failures into a
//! single report rather than failing one field at a time.
use std::fmt;
use crate::ConfigError;
/// A single validation failure with a human-readable path and message.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ValidationFailure {
/// Dotted path to the offending field, e.g. `"database.url"`.
pub path: String,
/// What was wrong.
pub message: String,
}
impl fmt::Display for ValidationFailure {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
write!(f, "{}: {}", self.path, self.message)
}
}
/// Errors produced by [`ConfigValidator::validate`].
#[derive(Debug, Clone)]
pub struct ValidationError {
/// All failures found in a single validation pass.
pub failures: Vec<ValidationFailure>,
}
impl fmt::Display for ValidationError {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
write!(f, "config validation failed ({} issue(s)):", self.failures.len())?;
for failure in &self.failures {
write!(f, "\n - {failure}")?;
}
Ok(())
}
}
impl std::error::Error for ValidationError {}
impl From<ValidationError> for ConfigError {
fn from(err: ValidationError) -> Self {
ConfigError::Validation(err.to_string())
}
}
/// Trait for types that can validate themselves after configuration loading.
///
/// Implementors collect field-level failures rather than returning on the
/// first error, so operators see the full problem set in one pass.
pub trait ConfigValidator {
fn validate(&self) -> Result<(), ValidationError>;
}
/// Convenience blanket for any serializable config type that implements
/// [`ConfigValidator`]. Callers typically invoke this on the output of
/// [`ConfigLoader::build`](crate::loader::ConfigLoader::build).
///
/// ```no_run
/// # use mytheclipse_config::{ConfigLoader, ConfigValidator, ConfigValidatorExt};
/// # use serde::Deserialize;
/// # #[derive(Debug, Deserialize)]
/// # struct Cfg { port: u16 }
/// # impl ConfigValidator for Cfg {
/// # fn validate(&self) -> Result<(), mytheclipse_config::ValidationError> { Ok(()) }
/// # }
/// let cfg: Cfg = ConfigLoader::new().build().unwrap();
/// cfg.validate_config().unwrap();
/// ```
pub trait ConfigValidatorExt: ConfigValidator {
/// Validates `self`, returning `Ok(())` on success.
fn validate_config(&self) -> Result<(), ConfigError> {
self.validate().map_err(ConfigError::from)
}
}
impl<T: ConfigValidator> ConfigValidatorExt for T {}
/// Validates that a string is a well-formed URL (http/https).
pub fn validate_url(path: &str, value: &str) -> Option<ValidationFailure> {
if value.is_empty() {
return Some(ValidationFailure {
path: path.to_string(),
message: "url must not be empty".into(),
});
}
// Minimal heuristic: scheme + host. We avoid pulling in a full URL crate
// to keep the dependency surface small.
let scheme_len = if value.starts_with("http://") { 7 } else if value.starts_with("https://") { 8 } else {
return Some(ValidationFailure {
path: path.to_string(),
message: format!("url must start with http:// or https:// (got {value:?})"),
});
};
let host = &value[scheme_len..];
if host.is_empty() {
return Some(ValidationFailure {
path: path.to_string(),
message: format!("url has no host portion (got {value:?})"),
});
}
None
}
/// Validates that a port number is in the valid range (1–65535).
pub fn validate_port(path: &str, port: u16) -> Option<ValidationFailure> {
// u16 already ranges 0–65535; exclude 0 (reserved/unspecified).
if port == 0 {
Some(ValidationFailure {
path: path.to_string(),
message: "port must be > 0".into(),
})
} else {
None
}
}
/// Validates that a string is non-empty.
pub fn validate_non_empty(path: &str, value: &str) -> Option<ValidationFailure> {
if value.trim().is_empty() {
Some(ValidationFailure {
path: path.to_string(),
message: "value must not be empty".into(),
})
} else {
None
}
}
/// Validates that a numeric value falls within `[lo, hi]`.
pub fn validate_range<T>(path: &str, value: T, lo: T, hi: T) -> Option<ValidationFailure>
where
T: PartialOrd + fmt::Display + Copy,
{
if value < lo || value > hi {
Some(ValidationFailure {
path: path.to_string(),
message: format!("value {value} is out of range [{lo}, {hi}]"),
})
} else {
None
}
}
/// Collects all failures from an iterator of `Option<ValidationFailure>`.
pub fn collect_failures(opts: impl IntoIterator<Item = Option<ValidationFailure>>) -> Result<(), ValidationError> {
let failures: Vec<_> = opts.into_iter().flatten().collect();
if failures.is_empty() {
Ok(())
} else {
Err(ValidationError { failures })
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn url_validator_pass_and_fail() {
assert!(validate_url("db.url", "https://example.com").is_none());
assert!(validate_url("db.url", "").is_some());
assert!(validate_url("db.url", "ftp://bad").is_some());
assert!(validate_url("db.url", "https://").is_some());
}
#[test]
fn port_validator_rejects_zero() {
assert!(validate_port("port", 0).is_some());
assert!(validate_port("port", 1).is_none());
assert!(validate_port("port", 65535).is_none());
}
#[test]
fn range_validator_bounds() {
assert!(validate_range("x", 5, 1, 10).is_none());
assert!(validate_range("x", 10, 1, 10).is_none());
assert!(validate_range("x", 0, 1, 10).is_some());
assert!(validate_range("x", 11, 1, 10).is_some());
}
#[test]
fn collect_failures_aggregates_all() {
let opts = [validate_non_empty("a", ""), validate_non_empty("b", "ok"), validate_url("c.d", "bad://x")];
let err = collect_failures(opts).unwrap_err();
assert_eq!(err.failures.len(), 2);
assert_eq!(err.failures[0].path, "a");
assert_eq!(err.failures[1].path, "c.d");
}
#[test]
fn collect_failures_ok_when_all_pass() {
let opts = [validate_url("a", "https://ok.com"), validate_port("b", 8080)];
assert!(collect_failures(opts).is_ok());
}
#[test]
fn blanket_ext_wrappers_validator() {
struct Cfg;
impl ConfigValidator for Cfg {
fn validate(&self) -> Result<(), ValidationError> {
Err(ValidationError {
failures: vec![ValidationFailure { path: "x".into(), message: "bad".into() }],
})
}
}
let c = Cfg;
assert!(c.validate_config().is_err());
}
}
+1 -1
View File
@@ -1,6 +1,6 @@
[package]
name = "mytheclipse-crypto"
version = "1.4.1"
version = "1.7.0"
edition = "2021"
rust-version = "1.75"
license = "MIT OR Apache-2.0"
@@ -0,0 +1,118 @@
//! Typed key registry with ID-based lookup (feature `password`).
//!
//! [`TypedKeyRegistry`] extends [`KeyRing`] semantics: instead of a single
//! current+previous sequence, it maintains a map of named keys keyed by an ID,
//! with one designated "current" ID. This is useful when keys are rotated by ID
//! (e.g. JWT `kid` header) and you need to look up a verification key by ID
//! while only accepting tokens signed by the current key.
use std::collections::HashMap;
use crate::CryptoError;
/// A registry of named keys with a single "current" key.
#[derive(Debug, Clone, Default)]
pub struct TypedKeyRegistry<T> {
keys: HashMap<String, T>,
current_id: Option<String>,
}
impl<T> TypedKeyRegistry<T> {
/// Creates an empty registry (no current key).
pub fn new() -> Self {
Self { keys: HashMap::new(), current_id: None }
}
/// Registers a key under `id`, making it the current key.
pub fn register(&mut self, id: impl Into<String>, key: T) {
let id = id.into();
self.keys.insert(id.clone(), key);
self.current_id = Some(id);
}
/// Looks up a key by ID (current or previous).
pub fn lookup(&self, id: &str) -> Option<&T> {
self.keys.get(id)
}
/// Returns the current key, if any.
pub fn current(&self) -> Option<&T> {
self.current_id
.as_ref()
.and_then(|id| self.keys.get(id))
}
/// Returns the ID of the current key.
pub fn current_id(&self) -> Option<&str> {
self.current_id.as_deref()
}
/// Rotates to a new current key identified by `id`. The old current key
/// remains accessible via `lookup` but is no longer the active signing key.
pub fn rotate_current(&mut self, id: impl Into<String>, key: T) {
let id = id.into();
self.keys.insert(id.clone(), key);
self.current_id = Some(id);
}
/// Number of keys in the registry.
pub fn len(&self) -> usize {
self.keys.len()
}
/// Whether the registry has any keys.
pub fn is_empty(&self) -> bool {
self.keys.is_empty()
}
/// Returns an error if no current key is registered.
pub fn require_current(&self) -> Result<&T, CryptoError> {
self.current()
.ok_or_else(|| CryptoError::Key("no current key registered".to_string()))
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn register_and_lookup() {
let mut reg = TypedKeyRegistry::new();
reg.register("k1", [1u8; 32]);
assert_eq!(reg.current_id(), Some("k1"));
assert!(reg.lookup("k1").is_some());
assert_eq!(reg.lookup("k1"), Some(&[1u8; 32]));
}
#[test]
fn lookup_unknown_returns_none() {
let reg = TypedKeyRegistry::<[u8; 32]>::new();
assert!(reg.lookup("nope").is_none());
}
#[test]
fn rotate_preserves_previous() {
let mut reg = TypedKeyRegistry::new();
reg.register("k1", [1u8; 32]);
reg.rotate_current("k2", [2u8; 32]);
assert_eq!(reg.current_id(), Some("k2"));
assert!(reg.lookup("k1").is_some());
assert_eq!(reg.lookup("k1"), Some(&[1u8; 32]));
}
#[test]
fn require_current_errors_when_empty() {
let reg = TypedKeyRegistry::<[u8; 32]>::new();
assert!(matches!(reg.require_current(), Err(CryptoError::Key(_))));
}
#[test]
fn len_and_is_empty() {
let mut reg = TypedKeyRegistry::new();
assert!(reg.is_empty());
reg.register("a", 0u32);
assert_eq!(reg.len(), 1);
assert!(!reg.is_empty());
}
}
+2
View File
@@ -42,6 +42,7 @@
//! ```
pub mod key_ring;
pub mod key_registry;
#[cfg(feature = "password")]
pub mod password;
@@ -68,6 +69,7 @@ pub use token::{Claims, TokenError, TokenSigner};
pub use paseto::{PasetoSigner, PasetoClaims};
pub use key_ring::KeyRing;
pub use key_registry::TypedKeyRegistry;
/// Errors returned across mytheclipse-crypto primitives.
#[non_exhaustive]
+1 -1
View File
@@ -1,6 +1,6 @@
[package]
name = "mytheclipse-event"
version = "1.4.1"
version = "1.7.0"
edition = "2021"
rust-version = "1.75"
license = "MIT OR Apache-2.0"
+5 -1
View File
@@ -1,6 +1,6 @@
[package]
name = "mytheclipse-http"
version = "1.4.1"
version = "1.7.0"
edition = "2021"
rust-version = "1.75"
license = "MIT OR Apache-2.0"
@@ -21,6 +21,8 @@ client = ["dep:reqwest", "dep:tokio"]
server-hyper = ["dep:hyper", "dep:tokio"]
# Server backed by axum.
server-axum = ["dep:axum", "dep:hyper", "dep:tokio"]
# Metrics HTTP endpoint serving Prometheus text format from a MetricsCollector.
metrics-http = ["dep:axum", "dep:tower", "dep:tokio", "dep:mytheclipse"]
[dependencies]
tracing = "0.1"
@@ -29,8 +31,10 @@ tokio = { version = "1.53", features = ["sync", "time", "rt", "macros"], optiona
reqwest = { version = "0.12", default-features = false, features = ["json", "rustls-tls"], optional = true }
hyper = { version = "1", features = ["full"], optional = true }
axum = { version = "0.8", optional = true }
tower = { version = "0.5", optional = true, default-features = false, features = ["util"] }
serde = { version = "1", features = ["derive"] }
serde_json = "1"
mytheclipse = { version = "1.5", path = "../mytheclipse", optional = true, default-features = false, features = ["observability"] }
[dev-dependencies]
tokio = { version = "1.53", features = ["full"] }
+6
View File
@@ -17,3 +17,9 @@ pub use client::HttpClient;
#[cfg(feature = "server-axum")]
pub mod server;
#[cfg(feature = "metrics-http")]
pub mod metrics_http;
#[cfg(feature = "metrics-http")]
pub use metrics_http::metrics_routes;
@@ -0,0 +1,56 @@
//! Prometheus metrics HTTP endpoint (feature `metrics-http`).
//!
//! [`metrics_routes`] returns an [`axum::Router`] that serves the
//! [`MetricsCollector`]'s Prometheus text exposition format at `/metrics`.
use axum::routing::get;
use axum::Router;
use std::sync::Arc;
use mytheclipse::MetricsCollector;
/// Builds a small axum router exposing `/metrics` (Prometheus text) and
/// `/` (a one-line description).
pub fn metrics_routes(collector: MetricsCollector) -> Router {
let collector = Arc::new(collector);
Router::new()
.route("/", get(|| async { "mytheclipse metrics" }))
.route("/metrics", get(metrics_handler))
.with_state(collector)
}
/// Axum handler serving the Prometheus text format.
async fn metrics_handler(
axum::extract::State(collector): axum::extract::State<Arc<MetricsCollector>>,
) -> axum::response::Response {
let body = collector.export_prometheus();
axum::response::Response::builder()
.status(200)
.header("content-type", "text/plain; version=0.0.4")
.body(axum::body::Body::from(body))
.unwrap_or_else(|_| {
axum::response::Response::new(axum::body::Body::from(
"internal error",
))
})
}
#[cfg(test)]
mod tests {
use super::*;
use tower::util::ServiceExt;
#[tokio::test]
async fn metrics_routes_serves_prometheus() {
let collector = MetricsCollector::new();
collector.inc_counter("test_reqs", 42);
let app = metrics_routes(collector);
let request = axum::extract::Request::get("/metrics")
.body(axum::body::Body::empty())
.unwrap();
let response = app.oneshot(request).await.unwrap();
assert_eq!(response.status(), 200);
}
}
+1 -1
View File
@@ -1,6 +1,6 @@
[package]
name = "mytheclipse-queue"
version = "1.4.1"
version = "1.7.0"
edition = "2021"
rust-version = "1.75"
license = "MIT OR Apache-2.0"
+1 -1
View File
@@ -1,6 +1,6 @@
[package]
name = "mytheclipse-storage"
version = "1.4.1"
version = "1.7.0"
edition = "2021"
rust-version = "1.75"
license = "MIT OR Apache-2.0"
+1 -1
View File
@@ -1,6 +1,6 @@
[package]
name = "mytheclipse-tracing"
version = "1.4.1"
version = "1.7.0"
edition = "2021"
rust-version = "1.75"
license = "MIT OR Apache-2.0"
+2 -2
View File
@@ -1,6 +1,6 @@
[package]
name = "mytheclipse"
version = "1.4.1"
version = "1.7.0"
edition = "2021"
rust-version = "1.75"
license = "MIT OR Apache-2.0"
@@ -34,7 +34,7 @@ bg = ["dep:tokio"]
resiliency = ["dep:tokio", "dep:rand"]
traffic = ["dep:tokio"]
lifecycle = ["dep:tokio"]
observability = ["dep:tokio"]
observability = ["dep:tokio", "lifecycle"]
full = ["io", "compute", "bg", "resiliency", "traffic", "lifecycle", "observability"]
[[example]]
+42
View File
@@ -22,6 +22,17 @@ pub enum CircuitState {
HalfOpen,
}
/// Point-in-time snapshot of a [`CircuitBreaker`] for metrics/observability.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct CircuitSnapshot {
/// Current circuit state.
pub state: CircuitState,
/// Consecutive failures recorded (resets on success in `Closed`).
pub failures: u64,
/// Consecutive successes recorded (resets on failure/open).
pub successes: u64,
}
const CLOSED: u8 = 0;
const OPEN: u8 = 1;
const HALF_OPEN: u8 = 2;
@@ -199,6 +210,16 @@ impl CircuitBreaker {
*self.inner.opened_at.lock().unwrap() = None;
}
/// Returns a point-in-time snapshot of the breaker's internal counters and
/// state, for metrics/observability export.
pub fn snapshot(&self) -> CircuitSnapshot {
CircuitSnapshot {
state: self.state(),
failures: self.inner.failures.load(Ordering::Acquire),
successes: self.inner.successes.load(Ordering::Acquire),
}
}
fn record_result(&self, success: bool) {
match self.inner.state.load(Ordering::Acquire) {
HALF_OPEN => {
@@ -362,4 +383,25 @@ mod tests {
let err: Result<u32, CircuitError<u8>> = b.call(|| Err(9u8));
assert!(matches!(err, Err(CircuitError::Inner(9))));
}
#[test]
fn snapshot_reflects_state_and_counts() {
let b = breaker();
let snap = b.snapshot();
assert_eq!(snap.state, CircuitState::Closed);
assert_eq!(snap.failures, 0);
assert_eq!(snap.successes, 0);
// success in Closed state resets failure count (no failure counter added).
b.call::<(), u8, _>(|| Ok(()));
let snap2 = b.snapshot();
assert_eq!(snap2.state, CircuitState::Closed);
for _ in 0..3 {
let _: Result<(), CircuitError<u8>> = b.call(|| Err(1u8));
}
let snap3 = b.snapshot();
assert_eq!(snap3.state, CircuitState::Open);
assert_eq!(snap3.failures, 0); // reset on open()
}
}
+13
View File
@@ -55,11 +55,15 @@ pub mod cron;
pub mod health;
#[cfg(feature = "lifecycle")]
pub mod leader;
#[cfg(feature = "lifecycle")]
pub mod lifecycle;
#[cfg(feature = "observability")]
pub mod metrics;
#[cfg(feature = "observability")]
pub mod panic_tracker;
#[cfg(feature = "observability")]
pub mod metrics_bridge;
#[cfg(feature = "resiliency")]
pub mod service_builder;
@@ -106,10 +110,19 @@ pub use service_builder::ServiceBuilder;
#[cfg(feature = "lifecycle")]
pub use dlock::{DistributedLock, LockError, LockGuard, InProcLock};
#[cfg(feature = "lifecycle")]
pub use lifecycle::AsyncLifecycleManager;
#[cfg(feature = "observability")]
pub use metrics::{MetricsCollector, MetricsSnapshot};
#[cfg(feature = "observability")]
pub use metrics_bridge::{MetricsBridge, MetricsHealthCheck};
/// Re-export of [`metrics_bridge::CircuitBreakerHealthCheck`].
/// Only compiled when both `observability` and `resiliency` are enabled.
#[cfg(all(feature = "observability", feature = "resiliency"))]
pub use metrics_bridge::CircuitBreakerHealthCheck;
#[cfg(feature = "observability")]
pub use panic_tracker::{PanicGuard, PanicInfo, PanicTracker};
/// Bootstraps the global [`EngineContext`].
+166
View File
@@ -0,0 +1,166 @@
//! Async lifecycle manager composing shutdown, health checks, and periodic tasks.
//!
//! [`AsyncLifecycleManager`] ties together [`ShutdownManager`], [`HealthRegistry`],
//! and an optional periodic health-check ticker into a single orchestrator so
//! applications don't need to wire three separate primitives together.
use std::sync::Arc;
use std::time::Duration;
use crate::health::{HealthCheck, HealthRegistry, HealthStatus};
use crate::shutdown::ShutdownManager;
/// Coordinates graceful shutdown, health-check registration, and an optional
/// periodic health poll loop.
///
/// Typical usage:
/// ```ignore
/// # tokio::runtime::Runtime::new().unwrap().block_on(async {
/// # use mytheclipse::AsyncLifecycleManager;
/// let mgr = AsyncLifecycleManager::new();
/// mgr.register_health_check("db", my_db_check());
/// let handle = mgr.start_health_loop(std::time::Duration::from_secs(30));
/// mgr.await_shutdown(std::time::Duration::from_secs(10)).await;
/// ```
#[derive(Clone)]
pub struct AsyncLifecycleManager {
shutdown: ShutdownManager,
health: Arc<HealthRegistry>,
}
impl AsyncLifecycleManager {
pub fn new() -> Self {
Self {
shutdown: ShutdownManager::new(),
health: Arc::new(HealthRegistry::new()),
}
}
/// Returns a clone of the underlying shutdown manager.
pub fn shutdown(&self) -> &ShutdownManager {
&self.shutdown
}
/// Returns a clone of the underlying health registry.
pub fn health(&self) -> &HealthRegistry {
&self.health
}
/// Registers a named health check.
pub async fn register_health_check(&self, name: impl Into<String>, check: impl HealthCheck + 'static) {
self.health.register(name, check).await;
}
/// Runs all registered health checks once and returns their statuses.
pub async fn check_health(&self) -> Vec<(String, HealthStatus)> {
self.health.check_all().await
}
/// Returns a shutdown signal for long-running tasks to observe.
pub fn shutdown_signal(&self) -> crate::shutdown::ShutdownSignal {
self.shutdown.handle()
}
/// Starts a background task that polls health checks at `interval` and
/// emits tracing events. Returns a [`tokio::task::JoinHandle`] that can
/// be aborted on shutdown.
pub fn start_health_loop(&self, interval: Duration) -> tokio::task::JoinHandle<()> {
let health = self.health.clone();
let signal = self.shutdown_signal();
tokio::spawn(async move {
let mut ticker = tokio::time::interval(interval);
let mut sig = signal;
loop {
// Stop when shutdown is requested.
if sig.is_shutdown() {
tracing::info_span!("mytheclipse_health_loop", );
return;
}
tokio::select! {
_ = sig.wait() => {
return;
}
_ = ticker.tick() => {
let results = health.check_all().await;
for (name, status) in &results {
match status {
HealthStatus::Ok => tracing::debug!(name, "health check ok"),
HealthStatus::Degraded => tracing::warn!(name, "health check degraded"),
HealthStatus::Unhealthy => tracing::error!(name, "health check unhealthy"),
}
}
if results.iter().any(|(_, s)| matches!(s, HealthStatus::Unhealthy)) {
tracing::error!("unhealthy component detected; requesting shutdown");
return;
}
}
}
}
})
}
/// Waits for shutdown (OS signal or explicit `request()`) then drains all
/// registered tasks with a `grace` timeout per task.
pub async fn await_shutdown(&self, grace: Duration) {
self.shutdown.wait_for_shutdown().await;
self.shutdown.drain(grace).await;
}
/// Requests shutdown programmatically (safe to call multiple times).
pub fn request_shutdown(&self) {
self.shutdown.request();
}
}
impl Default for AsyncLifecycleManager {
fn default() -> Self {
Self::new()
}
}
#[cfg(test)]
mod tests {
use super::*;
struct AlwaysOk;
impl HealthCheck for AlwaysOk {
fn name(&self) -> &str { "always-ok" }
fn check(&self) -> std::pin::Pin<Box<dyn std::future::Future<Output = HealthStatus> + Send + '_>> {
Box::pin(async { HealthStatus::Ok })
}
}
struct AlwaysBad;
impl HealthCheck for AlwaysBad {
fn name(&self) -> &str { "always-bad" }
fn check(&self) -> std::pin::Pin<Box<dyn std::future::Future<Output = HealthStatus> + Send + '_>> {
Box::pin(async { HealthStatus::Unhealthy })
}
}
#[tokio::test]
async fn new_manager_has_no_checks() {
let mgr = AsyncLifecycleManager::new();
let results = mgr.check_health().await;
assert!(results.is_empty());
}
#[tokio::test]
async fn registers_and_checks_health() {
let mgr = AsyncLifecycleManager::new();
mgr.register_health_check("ok", AlwaysOk).await;
let results = mgr.check_health().await;
assert_eq!(results.len(), 1);
assert_eq!(results[0].0, "ok");
assert_eq!(results[0].1, HealthStatus::Ok);
}
#[tokio::test]
async fn shutdown_signal_fires_on_request() {
let mgr = AsyncLifecycleManager::new();
let mut sig = mgr.shutdown_signal();
assert!(!sig.is_shutdown());
mgr.request_shutdown();
assert!(sig.is_shutdown());
}
}
+192
View File
@@ -0,0 +1,192 @@
//! Bridges the metrics collector to health checks and tracing events.
//!
//! [`MetricsBridge`] ties [`crate::metrics::MetricsCollector`] to
//! [`crate::health::HealthCheck`], so a metrics-based health probe can report
//! `Degraded` when error counters rise or throughput drops, and optionally emit
//! tracing events so counters/gauges are visible in structured logs.
use std::time::Duration;
use crate::health::{HealthCheck, HealthStatus};
use crate::metrics::MetricsCollector;
/// A health check backed by a [`CircuitBreaker`]: unhealthy if open,
/// degraded if half-open, ok otherwise.
///
/// Only available when both `resiliency` and `observability` features are
/// enabled (circuit breaker + health/metrics bridge).
#[cfg(feature = "resiliency")]
pub struct CircuitBreakerHealthCheck {
breaker: crate::circuit_breaker::CircuitBreaker,
}
#[cfg(feature = "resiliency")]
impl CircuitBreakerHealthCheck {
pub fn new(breaker: crate::circuit_breaker::CircuitBreaker) -> Self {
Self { breaker }
}
}
#[cfg(feature = "resiliency")]
impl HealthCheck for CircuitBreakerHealthCheck {
fn name(&self) -> &str {
"circuit_breaker"
}
fn check(&self) -> std::pin::Pin<Box<dyn std::future::Future<Output = HealthStatus> + Send + '_>> {
let state = self.breaker.snapshot().state;
Box::pin(async move {
match state {
crate::circuit_breaker::CircuitState::Open => HealthStatus::Unhealthy,
crate::circuit_breaker::CircuitState::HalfOpen => HealthStatus::Degraded,
crate::circuit_breaker::CircuitState::Closed => HealthStatus::Ok,
}
})
}
}
/// A health check backed by a [`MetricsCollector`]: unhealthy if any registered
/// "error" counter is non-zero, degraded if any gauge is below a configured
/// threshold.
pub struct MetricsHealthCheck {
collector: MetricsCollector,
}
impl MetricsHealthCheck {
pub fn new(collector: MetricsCollector) -> Self {
Self { collector }
}
/// Returns unhealthy if the named counter is non-zero.
pub fn error_counter_exists(&self, name: &str) -> bool {
self.collector.snapshot().counters.contains_key(name)
}
fn has_errors(&self) -> bool {
self.collector
.snapshot()
.counters
.values()
.any(|&v| v > 0)
}
}
impl HealthCheck for MetricsHealthCheck {
fn name(&self) -> &str {
"metrics"
}
fn check(&self) -> std::pin::Pin<Box<dyn std::future::Future<Output = HealthStatus> + Send + '_>> {
let has_errors = self.has_errors();
Box::pin(async move {
if has_errors {
HealthStatus::Unhealthy
} else {
HealthStatus::Ok
}
})
}
}
/// Bridges a [`MetricsCollector`] to tracing: periodically emits the current
/// snapshot as tracing events so metrics are visible in structured logs.
pub struct MetricsBridge {
collector: MetricsCollector,
}
impl MetricsBridge {
pub fn new(collector: MetricsCollector) -> Self {
Self { collector }
}
/// Sends a one-shot tracing event with the current snapshot.
pub fn emit_now(&self) {
let snap = self.collector.snapshot();
let mut counters: Vec<_> = snap.counters.into_iter().collect();
counters.sort_by(|a, b| a.0.cmp(&b.0));
let mut gauges: Vec<_> = snap.gauges.into_iter().collect();
gauges.sort_by(|a, b| a.0.cmp(&b.0));
tracing::debug!(
task_count = snap.task_count,
active_threads = snap.active_threads,
queue_capacity_total = snap.queue_capacity_total,
queue_capacity_remaining = snap.queue_capacity_remaining,
"metrics snapshot"
);
for (name, value) in &counters {
tracing::info!(name, value, "metric counter");
}
for (name, value) in &gauges {
tracing::info!(name, value, "metric gauge");
}
}
/// Spawns a background task that calls [`emit_now`](Self::emit_now) every
/// `interval`. Returns a handle that can be aborted.
pub fn emit_periodic(self, interval: Duration) -> tokio::task::JoinHandle<()> {
tokio::spawn(async move {
let mut ticker = tokio::time::interval(interval);
loop {
ticker.tick().await;
self.emit_now();
}
})
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::metrics::MetricsCollector;
use crate::shutdown::ShutdownManager;
#[test]
fn metrics_health_ok_when_no_counters() {
let collector = MetricsCollector::new();
let check = MetricsHealthCheck::new(collector);
// No counters set → no errors → Ok.
let fut = check.check();
// Can't await in #[test]; use tokio test below instead.
drop(fut);
}
#[tokio::test]
async fn metrics_health_unhealthy_when_errors_exist() {
let collector = MetricsCollector::new();
collector.inc_counter("errors", 1);
let check = MetricsHealthCheck::new(collector);
let status = check.check().await;
assert_eq!(status, HealthStatus::Unhealthy);
}
#[tokio::test]
async fn metrics_health_ok_when_no_errors() {
let collector = MetricsCollector::new();
collector.set_gauge("load", 0.5);
let check = MetricsHealthCheck::new(collector);
let status = check.check().await;
assert_eq!(status, HealthStatus::Ok);
}
#[tokio::test]
async fn bridge_emit_now_runs() {
let collector = MetricsCollector::new();
collector.set_gauge("temp", 42.0);
let bridge = MetricsBridge::new(collector);
bridge.emit_now();
}
#[cfg(feature = "lifecycle")]
#[tokio::test]
async fn lifecycle_manager_with_metrics_bridge() {
let collector = MetricsCollector::new();
collector.set_gauge("load", 0.1);
let mgr = crate::lifecycle::AsyncLifecycleManager::new();
let bridge = MetricsBridge::new(collector);
let _handle = bridge.emit_periodic(Duration::from_millis(50));
mgr.request_shutdown();
// Should not hang — shutdown is immediate.
mgr.await_shutdown(Duration::from_secs(1)).await;
}
}
+91
View File
@@ -85,6 +85,79 @@ impl<E: std::fmt::Display> std::fmt::Display for RetryError<E> {
impl<E: std::fmt::Debug + std::fmt::Display> std::error::Error for RetryError<E> {}
/// Statistics collected during a [`retry`] call.
#[derive(Debug, Clone, Default)]
pub struct RetryStats {
/// Total number of attempts made (including the first).
pub attempts: u32,
/// Number of retries performed (= `attempts - 1` if exhausted, or
/// `attempts - 1` if ultimately succeeded after at least one retry).
pub retries: u32,
/// The error message from the final attempt, if any.
pub last_error: Option<String>,
}
/// Like [`retry`] but also returns [`RetryStats`] capturing attempt counts.
///
/// Retries `op` according to `config`, retrying only errors for which
/// `filter` returns `true`.
///
/// Like [`retry`] but also returns [`RetryStats`].
pub async fn retry_with_stats<T, E, F, Fut, P>(
config: RetryConfig,
filter: P,
mut op: F,
) -> (Result<T, RetryError<E>>, RetryStats)
where
F: FnMut() -> Fut,
Fut: Future<Output = Result<T, E>>,
P: Fn(&E) -> bool,
E: std::fmt::Display,
{
let mut attempt: u32 = 0;
let mut last_error: Option<String> = None;
loop {
attempt += 1;
let span = tracing::info_span!(
"mytheclipse_retry_task",
attempt,
max_attempts = config.max_attempts
);
let result = op().instrument(span).await;
match result {
Ok(value) => {
let stats = RetryStats {
attempts: attempt,
retries: attempt.saturating_sub(1),
last_error,
};
return (Ok(value), stats);
}
Err(err) => {
last_error = Some(err.to_string());
let retryable = filter(&err);
if !retryable || attempt >= config.max_attempts {
let stats = RetryStats {
attempts: attempt,
retries: attempt.saturating_sub(1),
last_error,
};
return (
Err(RetryError::Exhausted {
attempts: attempt,
last: err,
}),
stats,
);
}
let delay = backoff_delay(&config, attempt, rand::thread_rng());
tokio::time::sleep(delay).await;
}
}
}
}
/// Retries `op` according to `config`, retrying only errors for which
/// `filter` returns `true`.
///
@@ -238,6 +311,24 @@ mod tests {
assert_eq!(calls.get(), 1);
}
#[tokio::test]
async fn retry_with_stats_succeeds_with_counts() {
use std::cell::Cell;
let config = RetryConfig {
max_attempts: 5,
base_delay: Duration::from_millis(1),
..RetryConfig::default()
};
let calls = Cell::new(0u32);
let (result, stats) = retry_with_stats(config, |_| true, || async {
calls.set(calls.get() + 1);
if calls.get() < 3 { Err::<u32, &str>("fail") } else { Ok(42u32) }
}).await;
assert_eq!(result.unwrap(), 42);
assert_eq!(stats.attempts, 3);
assert_eq!(stats.retries, 2);
}
#[test]
fn full_jitter_is_within_bounds_and_capped() {
let config = RetryConfig {