feat: add panic tracking and isolation with PanicTracker
- Implemented PanicTracker for logging panics using tracing. - Introduced PanicInfo struct to capture panic details. - Added tests for panic catching and hook restoration. feat: implement token-bucket rate limiter - Created RateLimiter for controlling token consumption with burst capacity. - Added methods for acquiring tokens and checking available tokens. - Included tests for rate limiting behavior and error handling. feat: add automatic retry with exponential backoff - Developed retry function to handle transient errors with configurable retry logic. - Introduced RetryConfig for customizing retry parameters. - Added tests for various retry scenarios and error handling. feat: implement graceful shutdown coordination - Created ShutdownManager to handle OS signals and manage task shutdown. - Introduced ShutdownSignal for tasks to observe shutdown requests. - Added tests for shutdown behavior and task management. feat: add timeout functionality for async operations - Implemented with_timeout function to enforce execution time limits. - Created Timeout struct for wrapping futures with deadlines. - Added tests for timeout behavior and error handling.
This commit is contained in:
asepharyana
@@ -50,6 +50,14 @@ jobs:
|
||||
flags: "--no-default-features --features compute"
|
||||
- name: bg only
|
||||
flags: "--no-default-features --features bg"
|
||||
- name: resiliency only
|
||||
flags: "--no-default-features --features resiliency"
|
||||
- name: traffic only
|
||||
flags: "--no-default-features --features traffic"
|
||||
- name: lifecycle only
|
||||
flags: "--no-default-features --features lifecycle"
|
||||
- name: observability only
|
||||
flags: "--no-default-features --features observability"
|
||||
steps:
|
||||
- uses: actions/checkout@v4
|
||||
- uses: dtolnay/rust-toolchain@stable
|
||||
|
||||
Generated
+71
@@ -61,6 +61,17 @@ dependencies = [
|
||||
"windows-sys",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "getrandom"
|
||||
version = "0.2.17"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "ff2abc00be7fca6ebc474524697ae276ad847ad0a6b3faa4bcb027e9a4614ad0"
|
||||
dependencies = [
|
||||
"cfg-if",
|
||||
"libc",
|
||||
"wasi",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "hermit-abi"
|
||||
version = "0.5.2"
|
||||
@@ -110,6 +121,7 @@ name = "mytheclipse"
|
||||
version = "1.0.0"
|
||||
dependencies = [
|
||||
"num_cpus",
|
||||
"rand",
|
||||
"rayon",
|
||||
"tokio",
|
||||
"tracing",
|
||||
@@ -170,6 +182,15 @@ version = "0.2.17"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "a89322df9ebe1c1578d689c92318e070967d1042b512afbe49518723f4e6d5cd"
|
||||
|
||||
[[package]]
|
||||
name = "ppv-lite86"
|
||||
version = "0.2.21"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "85eae3c4ed2f50dcfe72643da4befc30deadb458a9b590d720cde2f2b1e97da9"
|
||||
dependencies = [
|
||||
"zerocopy",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "proc-macro2"
|
||||
version = "1.0.107"
|
||||
@@ -188,6 +209,36 @@ dependencies = [
|
||||
"proc-macro2",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "rand"
|
||||
version = "0.8.8"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "e058c7de0b26af77780c769414d6257830bb240f3c38477dbc2c16e5f54d6d4c"
|
||||
dependencies = [
|
||||
"libc",
|
||||
"rand_chacha",
|
||||
"rand_core",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "rand_chacha"
|
||||
version = "0.3.1"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "e6c10a63a0fa32252be49d21e7709d4d4baf8d231c2dbce1eaa8141b9b127d88"
|
||||
dependencies = [
|
||||
"ppv-lite86",
|
||||
"rand_core",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "rand_core"
|
||||
version = "0.6.4"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "ec0be4795e2f6a28069bec0b5ff3e2ac9bafc99e6a9a7dc3547996c5c816922c"
|
||||
dependencies = [
|
||||
"getrandom",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "rayon"
|
||||
version = "1.12.0"
|
||||
@@ -406,3 +457,23 @@ checksum = "ae137229bcbd6cdf0f7b80a31df61766145077ddf49416a728b02cb3921ff3fc"
|
||||
dependencies = [
|
||||
"windows-link",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "zerocopy"
|
||||
version = "0.8.56"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "556764e583adb45a9f8d413c2a147fa7e8d821e48e12b14fd560b607998b75eb"
|
||||
dependencies = [
|
||||
"zerocopy-derive",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "zerocopy-derive"
|
||||
version = "0.8.56"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "f2ab42fc20575779bd240faa45f94a74256f755c0fa9e89f0ede20d91d0cdfc1"
|
||||
dependencies = [
|
||||
"proc-macro2",
|
||||
"quote",
|
||||
"syn 2.0.119",
|
||||
]
|
||||
|
||||
+8
-3
@@ -8,14 +8,15 @@ repository = "https://github.com/asepharyana/mytheclipse"
|
||||
homepage = "https://github.com/asepharyana/mytheclipse"
|
||||
documentation = "https://docs.rs/mytheclipse"
|
||||
authors = ["asepharyana <superaseph@gmail.com>"]
|
||||
description = "Resource-aware abstractions for async I/O, heavy compute, and background queue management."
|
||||
description = "Resource-aware abstractions for async I/O, heavy compute, background queue management, resiliency, traffic control, lifecycle, and observability."
|
||||
readme = "README.md"
|
||||
keywords = ["async", "concurrency", "rayon", "tokio", "resource-management"]
|
||||
keywords = ["async", "concurrency", "rayon", "tokio", "resource-management", "resiliency", "retry", "circuit-breaker", "rate-limit", "observability", "cron", "shutdown"]
|
||||
categories = ["asynchronous", "concurrency", "rust-patterns"]
|
||||
|
||||
[dependencies]
|
||||
tokio = { version = "1.53", features = ["full"], optional = true }
|
||||
rayon = { version = "1.12", optional = true }
|
||||
rand = { version = "0.8", optional = true }
|
||||
num_cpus = "1.17"
|
||||
tracing = "0.1"
|
||||
|
||||
@@ -28,7 +29,11 @@ default = []
|
||||
io = ["dep:tokio"]
|
||||
compute = ["dep:rayon"]
|
||||
bg = ["dep:tokio"]
|
||||
full = ["io", "compute", "bg"]
|
||||
resiliency = ["dep:tokio", "dep:rand"]
|
||||
traffic = ["dep:tokio"]
|
||||
lifecycle = ["dep:tokio"]
|
||||
observability = ["dep:tokio"]
|
||||
full = ["io", "compute", "bg", "resiliency", "traffic", "lifecycle", "observability"]
|
||||
|
||||
[[example]]
|
||||
name = "main"
|
||||
|
||||
@@ -4,7 +4,7 @@
|
||||
[](https://docs.rs/mytheclipse)
|
||||
[](LICENSE-MIT)
|
||||
|
||||
Resource-aware execution primitives for Rust: async I/O, heavy compute, and background queue management, sized automatically from the host's logical core count and exposed through a single, lazily-initialized engine context.
|
||||
Resource-aware execution primitives and reliability abstractions for Rust: async I/O, heavy compute, background queue management, resiliency, traffic control, lifecycle management, and observability. Sized automatically from the host's logical core count and exposed through a single, lazily-initialized engine context alongside self-contained, constructible utilities.
|
||||
|
||||
## Resource Sizing
|
||||
|
||||
@@ -16,12 +16,28 @@ Given $N$ logical cores (via `num_cpus::get()`):
|
||||
| **Compute** | $\max(1, N - 1)$ | 7 | Sized `rayon::ThreadPool` + `catch_unwind` |
|
||||
| **Background Queue** | $\max(2, \lfloor N / 2 \rfloor)$ | 4 | `tokio::sync::Semaphore` + `tokio::spawn` |
|
||||
|
||||
The three execution primitives (`io`, `compute`, `bg`) are sized from the host's CPU core count through `mytheclipse::context()`. Resiliency, traffic control, lifecycle, and observability utilities are constructible per-instance without global state.
|
||||
|
||||
## Features
|
||||
|
||||
- **`io`**: enables `mytheclipse::spawn_io`, instrumented async task spawning.
|
||||
- **`compute`**: enables `mytheclipse::compute`, panic-isolated execution on a sized Rayon pool.
|
||||
- **`bg`**: enables `mytheclipse::spawn_bg`, semaphore-bounded background tasks.
|
||||
- **`full`**: enables all three subsystems.
|
||||
- **`resiliency`**: fault tolerance abstractions:
|
||||
- `retry` — auto-retry with exponential backoff and full/equal jitter.
|
||||
- `CircuitBreaker` — failure threshold and cooldown state machine (Closed/Open/HalfOpen).
|
||||
- `with_timeout` / `timeout` / `Timeout` — hard execution deadlines.
|
||||
- **`traffic`**: traffic & load control:
|
||||
- `RateLimiter` — lazy token-bucket rate limiter with burst capacity.
|
||||
- `BackpressureQueue` — bounded queue with `DropOldest`, `Reject`, and `Block` overflow policies.
|
||||
- `ConcurrencyLimiter` — synchronous RAII semaphore wrapper for capping concurrent operations.
|
||||
- **`lifecycle`**: system lifecycle coordination:
|
||||
- `ShutdownManager` / `ShutdownSignal` — OS signal catching (SIGINT/SIGTERM/Ctrl-C) and graceful task draining.
|
||||
- `CronSchedule` / `schedule` — self-contained 5-field cron parser and async timer scheduler.
|
||||
- **`observability`**: runtime visibility:
|
||||
- `MetricsCollector` — thread-safe statistics collector with Prometheus text exposition format export.
|
||||
- `PanicTracker` — non-fatal panic logging with tracing context and boundary isolation.
|
||||
- **`full`**: enables all subsystems: `io`, `compute`, `bg`, `resiliency`, `traffic`, `lifecycle`, `observability`.
|
||||
|
||||
Zero features enabled by default (`default = []`), so you only pull in the dependencies your application actually uses.
|
||||
|
||||
@@ -31,40 +47,106 @@ Add to your `Cargo.toml`:
|
||||
|
||||
```toml
|
||||
[dependencies]
|
||||
mytheclipse = { version = "0.1", features = ["full"] }
|
||||
mytheclipse = { version = "0.2", features = ["full"] }
|
||||
```
|
||||
|
||||
Use the entry points directly:
|
||||
### 1. Execution Primitives
|
||||
|
||||
```rust
|
||||
#[tokio::main]
|
||||
async fn main() {
|
||||
// Optional explicit bootstrap: logs or validates resource sizing upfront.
|
||||
// Omit it and the first call to any primitive below will initialize it lazily.
|
||||
let ctx = mytheclipse::init();
|
||||
println!(
|
||||
"io_threads={} compute_threads={} bg_concurrency={}",
|
||||
ctx.io_threads, ctx.compute_threads, ctx.bg_concurrency
|
||||
);
|
||||
|
||||
// 1. Async I/O (instrumented with tracing)
|
||||
let io = mytheclipse::spawn_io(async {
|
||||
// ... network / disk work ...
|
||||
42
|
||||
});
|
||||
// Async I/O (instrumented with tracing)
|
||||
let io = mytheclipse::spawn_io(async { 42 });
|
||||
|
||||
// 2. Heavy Compute (isolated from worker panics)
|
||||
// Heavy Compute (isolated from worker panics)
|
||||
let sum = mytheclipse::compute(|| (1..=1_000_000u64).sum::<u64>())?;
|
||||
|
||||
// 3. Background Queue (concurrency-bounded)
|
||||
let bg = mytheclipse::spawn_bg(async {
|
||||
// ... deferred cleanup / telemetry ...
|
||||
}).await;
|
||||
// Background Queue (concurrency-bounded)
|
||||
let bg = mytheclipse::spawn_bg(async { /* task */ }).await;
|
||||
|
||||
let _ = (io.await, bg.await);
|
||||
}
|
||||
```
|
||||
|
||||
### 2. Resiliency & Fault Tolerance
|
||||
|
||||
```rust
|
||||
use std::time::Duration;
|
||||
|
||||
// Auto-Retry with Exponential Backoff + Jitter
|
||||
let result = mytheclipse::retry(
|
||||
mytheclipse::RetryConfig::default(),
|
||||
|err| err.is_transient(),
|
||||
|| async { make_network_request().await },
|
||||
).await?;
|
||||
|
||||
// Circuit Breaker
|
||||
let breaker = mytheclipse::CircuitBreaker::new(mytheclipse::CircuitBreakerConfig::default());
|
||||
let value = breaker.call(|| fetch_remote_resource())?;
|
||||
|
||||
// Timeout & Deadlines
|
||||
let value = mytheclipse::with_timeout(Duration::from_secs(5), async {
|
||||
long_running_task().await
|
||||
}).await?;
|
||||
```
|
||||
|
||||
### 3. Traffic & Resource Control
|
||||
|
||||
```rust
|
||||
// Rate Limiter (Token Bucket)
|
||||
let limiter = mytheclipse::RateLimiter::new(100.0, 10);
|
||||
limiter.acquire().await?;
|
||||
|
||||
// Concurrency Limiter (Sync Semaphore)
|
||||
let limiter = mytheclipse::ConcurrencyLimiter::new(10);
|
||||
let _permit = limiter.acquire(); // released on drop
|
||||
|
||||
// Backpressure Queue (Graceful Degradation)
|
||||
let queue = mytheclipse::BackpressureQueue::new(100, mytheclipse::OverflowPolicy::DropOldest);
|
||||
queue.push(job).await?;
|
||||
let next_job = queue.pop().await;
|
||||
```
|
||||
|
||||
### 4. Lifecycle & State Management
|
||||
|
||||
```rust
|
||||
// Graceful Shutdown Manager
|
||||
let shutdown = mytheclipse::ShutdownManager::new();
|
||||
let sig = shutdown.handle();
|
||||
tokio::spawn(async move {
|
||||
let mut sig = sig;
|
||||
tokio::select! {
|
||||
_ = sig.wait() => { /* clean up */ }
|
||||
_ = worker_loop() => {}
|
||||
}
|
||||
});
|
||||
shutdown.drain(Duration::from_secs(10)).await;
|
||||
|
||||
// Cron Periodic Job Scheduler (Self-contained, no external crates)
|
||||
let cron = mytheclipse::CronSchedule::parse("0 1 * * *")?; // 1 AM daily
|
||||
let job = mytheclipse::schedule("0 1 * * *", || async {
|
||||
clean_cache().await;
|
||||
})?;
|
||||
```
|
||||
|
||||
### 5. Telemetry & Observability
|
||||
|
||||
```rust
|
||||
// Centralized Metrics Collector (Prometheus text exposition format)
|
||||
let metrics = mytheclipse::MetricsCollector::new();
|
||||
metrics.record_task(Duration::from_millis(15));
|
||||
metrics.inc_counter("http_requests_total", 1);
|
||||
let prometheus_output = metrics.export_prometheus();
|
||||
|
||||
// Panic Isolation Tracker
|
||||
let guard = mytheclipse::PanicTracker::install(); // logs panics with tracing span
|
||||
let result = mytheclipse::PanicTracker::catch(|| {
|
||||
risky_operation()
|
||||
});
|
||||
```
|
||||
|
||||
## Running the Example
|
||||
|
||||
```bash
|
||||
@@ -78,4 +160,4 @@ Licensed under either of:
|
||||
- Apache License, Version 2.0 ([LICENSE-APACHE](LICENSE-APACHE) or <http://www.apache.org/licenses/LICENSE-2.0>)
|
||||
- MIT license ([LICENSE-MIT](LICENSE-MIT) or <http://opensource.org/licenses/MIT>)
|
||||
|
||||
at your option.
|
||||
at your option.
|
||||
|
||||
+149
-5
@@ -1,10 +1,14 @@
|
||||
//! Demonstrates `mytheclipse`'s three execution primitives end to end, including
|
||||
//! automatic recovery from a panicking compute closure.
|
||||
//! Demonstrates `mytheclipse`'s capabilities end to end:
|
||||
//! execution primitives, resiliency & fault tolerance, traffic control,
|
||||
//! lifecycle management, and observability.
|
||||
|
||||
use std::time::Duration;
|
||||
|
||||
#[tokio::main]
|
||||
async fn main() {
|
||||
tracing_subscriber::fmt::init();
|
||||
|
||||
println!("=== 1. Execution Primitives & Context ===");
|
||||
let ctx = mytheclipse::init();
|
||||
println!(
|
||||
"engine context: io_threads={} compute_threads={} bg_concurrency={}",
|
||||
@@ -12,20 +16,20 @@ async fn main() {
|
||||
);
|
||||
|
||||
let io_handle = mytheclipse::spawn_io(async {
|
||||
tokio::time::sleep(std::time::Duration::from_millis(50)).await;
|
||||
tokio::time::sleep(Duration::from_millis(10)).await;
|
||||
42u64
|
||||
});
|
||||
|
||||
let sum_result = mytheclipse::compute(|| (1..=1_000u64).sum::<u64>());
|
||||
|
||||
let panic_result: Result<u64, mytheclipse::MytheclipseError> = mytheclipse::compute(|| {
|
||||
panic!("intentional panic to demonstrate isolation");
|
||||
panic!("intentional panic to demonstrate compute isolation");
|
||||
});
|
||||
|
||||
let recovery_result = mytheclipse::compute(|| 2u64 + 2u64);
|
||||
|
||||
let bg_handle = mytheclipse::spawn_bg(async {
|
||||
tokio::time::sleep(std::time::Duration::from_millis(10)).await;
|
||||
tokio::time::sleep(Duration::from_millis(10)).await;
|
||||
"bg-task-done"
|
||||
})
|
||||
.await;
|
||||
@@ -38,4 +42,144 @@ async fn main() {
|
||||
println!("compute panic-isolation result: {panic_result:?}");
|
||||
println!("compute pool still usable after panic: {recovery_result:?}");
|
||||
println!("spawn_bg result: {bg_value}");
|
||||
|
||||
println!("\n=== 2. Resiliency & Fault Tolerance ===");
|
||||
// Auto-Retry with Exponential Backoff & Full Jitter
|
||||
let attempts = std::sync::atomic::AtomicU32::new(0);
|
||||
let retry_res = mytheclipse::retry(
|
||||
mytheclipse::RetryConfig {
|
||||
max_attempts: 4,
|
||||
base_delay: Duration::from_millis(10),
|
||||
..Default::default()
|
||||
},
|
||||
|_| true,
|
||||
|| async {
|
||||
let count = attempts.fetch_add(1, std::sync::atomic::Ordering::SeqCst) + 1;
|
||||
if count < 3 {
|
||||
Err("temporary network failure")
|
||||
} else {
|
||||
Ok("connected successfully!")
|
||||
}
|
||||
},
|
||||
)
|
||||
.await;
|
||||
println!(
|
||||
"retry result (succeeded on attempt {}): {retry_res:?}",
|
||||
attempts.load(std::sync::atomic::Ordering::SeqCst)
|
||||
);
|
||||
|
||||
// Circuit Breaker
|
||||
let breaker = mytheclipse::CircuitBreaker::new(mytheclipse::CircuitBreakerConfig {
|
||||
failure_threshold: 2,
|
||||
open_timeout: Duration::from_millis(50),
|
||||
..Default::default()
|
||||
});
|
||||
let _ = breaker.call(|| Err::<(), _>("service unavailable"));
|
||||
let _ = breaker.call(|| Err::<(), _>("service unavailable"));
|
||||
println!(
|
||||
"circuit breaker state after 2 failures: {:?}",
|
||||
breaker.state()
|
||||
);
|
||||
let call_when_open: Result<(), mytheclipse::CircuitError<&str>> =
|
||||
breaker.call(|| Ok::<(), &str>(()));
|
||||
println!("call blocked by circuit breaker: {call_when_open:?}");
|
||||
|
||||
// Timeout & Deadlines
|
||||
let timeout_res = mytheclipse::with_timeout(Duration::from_millis(20), async {
|
||||
tokio::time::sleep(Duration::from_millis(100)).await;
|
||||
"finished"
|
||||
})
|
||||
.await;
|
||||
println!("with_timeout result on slow task: {timeout_res:?}");
|
||||
|
||||
println!("\n=== 3. Traffic & Resource Control ===");
|
||||
// Rate Limiter (Token Bucket)
|
||||
let limiter = mytheclipse::RateLimiter::new(100.0, 2);
|
||||
println!(
|
||||
"rate limiter initial tokens: {}",
|
||||
limiter.available_tokens()
|
||||
);
|
||||
let _ = limiter.try_acquire();
|
||||
println!(
|
||||
"rate limiter tokens after 1 acquire: {}",
|
||||
limiter.available_tokens()
|
||||
);
|
||||
|
||||
// Concurrency Limiter (Sync Semaphore Wrapper)
|
||||
let conc = mytheclipse::ConcurrencyLimiter::new(2);
|
||||
let permit1 = conc.try_acquire().expect("permit 1");
|
||||
let permit2 = conc.try_acquire().expect("permit 2");
|
||||
println!(
|
||||
"concurrency limiter in use: {}/{}",
|
||||
conc.in_use(),
|
||||
conc.max()
|
||||
);
|
||||
assert!(conc.try_acquire().is_none());
|
||||
drop(permit1);
|
||||
drop(permit2);
|
||||
println!(
|
||||
"concurrency limiter in use after drop: {}/{}",
|
||||
conc.in_use(),
|
||||
conc.max()
|
||||
);
|
||||
|
||||
// Backpressure Queue (Graceful Degradation)
|
||||
let queue = mytheclipse::BackpressureQueue::new(2, mytheclipse::OverflowPolicy::DropOldest);
|
||||
queue.push("item-1").await.unwrap();
|
||||
queue.push("item-2").await.unwrap();
|
||||
queue.push("item-3").await.unwrap(); // item-1 will be dropped
|
||||
println!(
|
||||
"backpressure queue: len={}, dropped={}, next={:?}",
|
||||
queue.len(),
|
||||
queue.dropped_count(),
|
||||
queue.pop().await
|
||||
);
|
||||
|
||||
println!("\n=== 4. Lifecycle & State Management ===");
|
||||
// Cron Scheduler (Minimal 5-field parser without external crate)
|
||||
let cron = mytheclipse::CronSchedule::parse("0 1 * * *").expect("valid cron expression");
|
||||
let now = mytheclipse::cron::CronTime::now();
|
||||
let next_runs = cron.next_five(now);
|
||||
println!("cron '0 1 * * *' next 3 scheduled fire times:");
|
||||
for t in next_runs.iter().take(3) {
|
||||
println!(
|
||||
" -> {:04}-{:02}-{:02} {:02}:{:02}:00 UTC",
|
||||
t.year, t.month, t.day, t.hour, t.minute
|
||||
);
|
||||
}
|
||||
|
||||
// Graceful Shutdown Manager
|
||||
let shutdown = mytheclipse::ShutdownManager::new();
|
||||
let sig = shutdown.handle();
|
||||
println!(
|
||||
"shutdown signal status before request: is_shutdown={}",
|
||||
sig.is_shutdown()
|
||||
);
|
||||
shutdown.request();
|
||||
println!(
|
||||
"shutdown signal status after request: is_shutdown={}",
|
||||
sig.is_shutdown()
|
||||
);
|
||||
|
||||
println!("\n=== 5. Telemetry & Observability ===");
|
||||
// Centralized Metrics Collector (Prometheus text exposition format)
|
||||
let metrics = mytheclipse::MetricsCollector::new();
|
||||
metrics.record_task(Duration::from_millis(15));
|
||||
metrics.record_task(Duration::from_millis(25));
|
||||
metrics.set_active_threads(ctx.compute_threads);
|
||||
metrics.set_queue_capacity(100);
|
||||
metrics.set_queue_remaining(98);
|
||||
metrics.inc_counter("http_requests_total", 42);
|
||||
metrics.set_gauge("memory_usage_mb", 128.5);
|
||||
|
||||
println!("Prometheus Export Output:\n---");
|
||||
print!("{}", metrics.export_prometheus());
|
||||
println!("---");
|
||||
|
||||
// Panic & Span Isolation Tracker
|
||||
let isolated_panic = mytheclipse::PanicTracker::catch(|| {
|
||||
panic!("worker routine isolated error");
|
||||
});
|
||||
println!("PanicTracker::catch result: {isolated_panic:?}");
|
||||
println!("\nAll mytheclipse features executed successfully!");
|
||||
}
|
||||
|
||||
@@ -0,0 +1,310 @@
|
||||
//! A bounded queue with configurable overflow handling (feature `traffic`).
|
||||
//!
|
||||
//! [`BackpressureQueue`] buffers items up to a fixed capacity. When full, the
|
||||
//! configured [`OverflowPolicy`] decides the graceful-degradation behavior:
|
||||
//! drop the oldest item to make room, reject the new item and hand it back to
|
||||
//! the caller, or block the caller until a slot frees up.
|
||||
|
||||
use std::collections::VecDeque;
|
||||
use std::sync::atomic::{AtomicU64, Ordering};
|
||||
use std::sync::{Arc, Mutex};
|
||||
|
||||
use tokio::sync::Notify;
|
||||
use tracing::Instrument;
|
||||
|
||||
/// Behaviour applied when the queue is at capacity.
|
||||
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
|
||||
pub enum OverflowPolicy {
|
||||
/// Discard the oldest item to free a slot, then accept the new one.
|
||||
DropOldest,
|
||||
/// Return the new item to the caller as an error without enqueueing it.
|
||||
Reject,
|
||||
/// Block the calling coroutine until a slot frees up.
|
||||
Block,
|
||||
}
|
||||
|
||||
/// The error returned when an item cannot be enqueued.
|
||||
#[derive(Debug)]
|
||||
pub enum BackpressureError<T> {
|
||||
/// The queue was full and the [`OverflowPolicy::Reject`] policy declined
|
||||
/// the item. The rejected item is returned so the caller may handle it.
|
||||
QueueFull(T),
|
||||
}
|
||||
|
||||
impl<T> std::fmt::Display for BackpressureError<T> {
|
||||
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
|
||||
write!(f, "backpressure queue is full")
|
||||
}
|
||||
}
|
||||
|
||||
impl<T: std::fmt::Debug> std::error::Error for BackpressureError<T> {}
|
||||
|
||||
struct Inner<T> {
|
||||
queue: Mutex<VecDeque<T>>,
|
||||
not_full: Notify,
|
||||
not_empty: Notify,
|
||||
capacity: usize,
|
||||
policy: OverflowPolicy,
|
||||
accepted: AtomicU64,
|
||||
dropped: AtomicU64,
|
||||
rejected: AtomicU64,
|
||||
closed: AtomicU64,
|
||||
}
|
||||
|
||||
const CLOSED: u64 = 1;
|
||||
const OPEN: u64 = 0;
|
||||
|
||||
/// A thread-safe, bounded work queue with graceful-degradation overflow
|
||||
/// handling.
|
||||
///
|
||||
/// Construct with [`BackpressureQueue::new`]; await [`push`](Self::push) /
|
||||
/// [`pop`](Self::pop) in async contexts, or use
|
||||
/// [`try_push`](Self::try_push) to fail fast.
|
||||
#[derive(Clone)]
|
||||
pub struct BackpressureQueue<T> {
|
||||
inner: Arc<Inner<T>>,
|
||||
}
|
||||
|
||||
impl<T> BackpressureQueue<T> {
|
||||
/// Builds a queue with `capacity` slots and the given overflow policy.
|
||||
///
|
||||
/// # Panics
|
||||
///
|
||||
/// Panics if `capacity` is zero.
|
||||
pub fn new(capacity: usize, policy: OverflowPolicy) -> Self {
|
||||
assert!(capacity > 0, "backpressure queue capacity must be > 0");
|
||||
Self {
|
||||
inner: Arc::new(Inner {
|
||||
queue: Mutex::new(VecDeque::with_capacity(capacity)),
|
||||
not_full: Notify::new(),
|
||||
not_empty: Notify::new(),
|
||||
capacity,
|
||||
policy,
|
||||
accepted: AtomicU64::new(0),
|
||||
dropped: AtomicU64::new(0),
|
||||
rejected: AtomicU64::new(0),
|
||||
closed: AtomicU64::new(OPEN),
|
||||
}),
|
||||
}
|
||||
}
|
||||
|
||||
/// Enqueues `item`, applying the configured overflow policy when full.
|
||||
///
|
||||
/// For [`OverflowPolicy::Block`], awaits a free slot first (in a
|
||||
/// `mytheclipse_backpressure_task` span); for [`OverflowPolicy::Reject`],
|
||||
/// returns the item back as [`BackpressureError::QueueFull`] without
|
||||
/// blocking; for [`OverflowPolicy::DropOldest`], never fails and evicts
|
||||
/// the oldest item.
|
||||
pub async fn push(&self, item: T) -> Result<(), BackpressureError<T>> {
|
||||
if self.inner.closed.load(Ordering::Acquire) == CLOSED {
|
||||
return Err(BackpressureError::QueueFull(item));
|
||||
}
|
||||
match self.inner.policy {
|
||||
OverflowPolicy::Block => {
|
||||
let span = tracing::info_span!("mytheclipse_backpressure_task");
|
||||
loop {
|
||||
{
|
||||
let mut queue = self.inner.queue.lock().unwrap();
|
||||
if queue.len() < self.inner.capacity {
|
||||
queue.push_back(item);
|
||||
self.inner.accepted.fetch_add(1, Ordering::AcqRel);
|
||||
drop(queue);
|
||||
self.inner.not_empty.notify_one();
|
||||
return Ok(());
|
||||
}
|
||||
}
|
||||
let waiter = self.inner.not_full.notified();
|
||||
tokio::pin!(waiter);
|
||||
waiter.as_mut().instrument(span.clone()).await;
|
||||
}
|
||||
}
|
||||
OverflowPolicy::Reject => self.try_push(item),
|
||||
OverflowPolicy::DropOldest => {
|
||||
self.try_push_dropping(item);
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Non-blocking enqueue honoring [`OverflowPolicy::DropOldest`] and
|
||||
/// [`OverflowPolicy::Reject`]; for [`OverflowPolicy::Block`] it behaves
|
||||
/// like [`OverflowPolicy::Reject`] (never blocks).
|
||||
pub fn try_push(&self, item: T) -> Result<(), BackpressureError<T>> {
|
||||
let mut queue = self.inner.queue.lock().unwrap();
|
||||
if self.inner.closed.load(Ordering::Acquire) == CLOSED {
|
||||
return Err(BackpressureError::QueueFull(item));
|
||||
}
|
||||
if queue.len() >= self.inner.capacity {
|
||||
match self.inner.policy {
|
||||
OverflowPolicy::DropOldest => {
|
||||
let _dropped = queue.pop_front();
|
||||
self.inner.dropped.fetch_add(1, Ordering::AcqRel);
|
||||
}
|
||||
_ => {
|
||||
self.inner.rejected.fetch_add(1, Ordering::AcqRel);
|
||||
return Err(BackpressureError::QueueFull(item));
|
||||
}
|
||||
}
|
||||
}
|
||||
queue.push_back(item);
|
||||
self.inner.accepted.fetch_add(1, Ordering::AcqRel);
|
||||
drop(queue);
|
||||
self.inner.not_empty.notify_one();
|
||||
Ok(())
|
||||
}
|
||||
|
||||
fn try_push_dropping(&self, item: T) {
|
||||
let mut queue = self.inner.queue.lock().unwrap();
|
||||
if queue.len() >= self.inner.capacity {
|
||||
let _dropped = queue.pop_front();
|
||||
self.inner.dropped.fetch_add(1, Ordering::AcqRel);
|
||||
}
|
||||
queue.push_back(item);
|
||||
self.inner.accepted.fetch_add(1, Ordering::AcqRel);
|
||||
drop(queue);
|
||||
self.inner.not_empty.notify_one();
|
||||
}
|
||||
|
||||
/// Awaits the next item, blocking until one is available.
|
||||
pub async fn pop(&self) -> Option<T> {
|
||||
loop {
|
||||
{
|
||||
let mut queue = self.inner.queue.lock().unwrap();
|
||||
if let Some(item) = queue.pop_front() {
|
||||
self.inner.not_full.notify_one();
|
||||
return Some(item);
|
||||
}
|
||||
if self.inner.closed.load(Ordering::Acquire) == CLOSED && queue.is_empty() {
|
||||
return None;
|
||||
}
|
||||
}
|
||||
self.inner.not_empty.notified().await;
|
||||
}
|
||||
}
|
||||
|
||||
/// Attempts to pop an item without blocking.
|
||||
pub fn try_pop(&self) -> Option<T> {
|
||||
let mut queue = self.inner.queue.lock().unwrap();
|
||||
let item = queue.pop_front();
|
||||
if item.is_some() {
|
||||
self.inner.not_full.notify_one();
|
||||
}
|
||||
item
|
||||
}
|
||||
|
||||
/// The number of items currently buffered.
|
||||
pub fn len(&self) -> usize {
|
||||
self.inner.queue.lock().unwrap().len()
|
||||
}
|
||||
|
||||
/// Whether the queue is empty.
|
||||
pub fn is_empty(&self) -> bool {
|
||||
self.len() == 0
|
||||
}
|
||||
|
||||
/// The queue's capacity.
|
||||
pub fn capacity(&self) -> usize {
|
||||
self.inner.capacity
|
||||
}
|
||||
|
||||
/// How much room remains.
|
||||
pub fn remaining(&self) -> usize {
|
||||
self.inner.capacity - self.len()
|
||||
}
|
||||
|
||||
/// Number of items successfully enqueued.
|
||||
pub fn accepted_count(&self) -> u64 {
|
||||
self.inner.accepted.load(Ordering::Acquire)
|
||||
}
|
||||
|
||||
/// Number of items evicted by drop policies.
|
||||
pub fn dropped_count(&self) -> u64 {
|
||||
self.inner.dropped.load(Ordering::Acquire)
|
||||
}
|
||||
|
||||
/// Number of items rejected by the reject/closed paths.
|
||||
pub fn rejected_count(&self) -> u64 {
|
||||
self.inner.rejected.load(Ordering::Acquire)
|
||||
}
|
||||
|
||||
/// Closes the queue: no further items are accepted and `pop` drains the
|
||||
/// remaining items then returns `None`.
|
||||
pub fn close(&self) {
|
||||
self.inner.closed.store(CLOSED, Ordering::Release);
|
||||
self.inner.not_empty.notify_waiters();
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
use std::time::Duration;
|
||||
|
||||
#[tokio::test]
|
||||
async fn push_pop_roundtrip() {
|
||||
let q: BackpressureQueue<u32> = BackpressureQueue::new(4, OverflowPolicy::Reject);
|
||||
q.push(1).await.unwrap();
|
||||
q.push(2).await.unwrap();
|
||||
assert_eq!(q.pop().await, Some(1));
|
||||
assert_eq!(q.pop().await, Some(2));
|
||||
assert!(q.is_empty());
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn reject_returns_item_when_full() {
|
||||
let q: BackpressureQueue<u32> = BackpressureQueue::new(2, OverflowPolicy::Reject);
|
||||
q.push(1).await.unwrap();
|
||||
q.push(2).await.unwrap();
|
||||
let err = q.push(3).await.unwrap_err();
|
||||
match err {
|
||||
BackpressureError::QueueFull(v) => assert_eq!(v, 3),
|
||||
}
|
||||
assert_eq!(q.len(), 2);
|
||||
assert_eq!(q.rejected_count(), 1);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn drop_oldest_evicts_front() {
|
||||
let q: BackpressureQueue<u32> = BackpressureQueue::new(2, OverflowPolicy::DropOldest);
|
||||
q.push(1).await.unwrap();
|
||||
q.push(2).await.unwrap();
|
||||
q.push(3).await.unwrap();
|
||||
assert_eq!(q.len(), 2);
|
||||
assert_eq!(q.pop().await, Some(2)); // 1 was dropped
|
||||
assert_eq!(q.dropped_count(), 1);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn block_waits_for_a_slot() {
|
||||
let q: BackpressureQueue<u32> = BackpressureQueue::new(1, OverflowPolicy::Block);
|
||||
q.push(1).await.unwrap();
|
||||
// A second push must block until pop frees a slot.
|
||||
let q2 = q.clone();
|
||||
let pusher = tokio::spawn(async move {
|
||||
q2.push(2).await.unwrap();
|
||||
});
|
||||
tokio::time::sleep(Duration::from_millis(20)).await;
|
||||
assert_eq!(q.pop().await, Some(1));
|
||||
pusher.await.unwrap();
|
||||
assert_eq!(q.pop().await, Some(2));
|
||||
assert_eq!(q.accepted_count(), 2);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn close_drains_then_returns_none() {
|
||||
let q: BackpressureQueue<u32> = BackpressureQueue::new(2, OverflowPolicy::Reject);
|
||||
q.push(1).await.unwrap();
|
||||
q.push(2).await.unwrap();
|
||||
let q2 = q.clone();
|
||||
let drainer = tokio::spawn(async move {
|
||||
let mut seen = Vec::new();
|
||||
while let Some(v) = q2.pop().await {
|
||||
seen.push(v);
|
||||
}
|
||||
seen
|
||||
});
|
||||
q.close();
|
||||
assert_eq!(drainer.await.unwrap(), vec![1, 2]);
|
||||
assert!(q.push(3).await.is_err());
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,365 @@
|
||||
//! Circuit breaker for isolating failures in calls to external services
|
||||
//! (feature `resiliency`).
|
||||
//!
|
||||
//! A [`CircuitBreaker`] tracks consecutive failures on calls it guards. Once
|
||||
//! failures reach a threshold it *trips open*, refusing further calls for a
|
||||
//! cooldown window so the target gets time to recover. After the window it
|
||||
//! transitions to *half-open*, admitting a small number of probe calls; a
|
||||
//! successful probe closes the circuit, a failed probe re-opens it.
|
||||
|
||||
use std::sync::atomic::{AtomicU64, AtomicU8, AtomicUsize, Ordering};
|
||||
use std::sync::{Arc, Mutex};
|
||||
use std::time::{Duration, Instant};
|
||||
|
||||
/// The current state of a circuit.
|
||||
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
|
||||
pub enum CircuitState {
|
||||
/// Requests flow normally; failures accumulate toward the threshold.
|
||||
Closed,
|
||||
/// Requests are refused while the circuit waits to recover.
|
||||
Open,
|
||||
/// A limited number of probe requests are admitted to test recovery.
|
||||
HalfOpen,
|
||||
}
|
||||
|
||||
const CLOSED: u8 = 0;
|
||||
const OPEN: u8 = 1;
|
||||
const HALF_OPEN: u8 = 2;
|
||||
|
||||
/// Configuration for a [`CircuitBreaker`].
|
||||
#[derive(Debug, Clone)]
|
||||
pub struct CircuitBreakerConfig {
|
||||
/// Consecutive failures that trip `Closed` -> `Open`.
|
||||
pub failure_threshold: u64,
|
||||
/// How long to remain `Open` before moving to `HalfOpen`.
|
||||
pub open_timeout: Duration,
|
||||
/// Maximum concurrent probe calls admitted while `HalfOpen`.
|
||||
pub half_open_max_calls: usize,
|
||||
/// Consecutive successes that close `HalfOpen` -> `Closed`.
|
||||
pub close_successes: u64,
|
||||
}
|
||||
|
||||
impl Default for CircuitBreakerConfig {
|
||||
fn default() -> Self {
|
||||
Self {
|
||||
failure_threshold: 5,
|
||||
open_timeout: Duration::from_secs(30),
|
||||
half_open_max_calls: 1,
|
||||
close_successes: 1,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// The result of a guarded call.
|
||||
#[derive(Debug)]
|
||||
pub enum CircuitError<E> {
|
||||
/// The circuit is open and the call was refused without executing.
|
||||
Open,
|
||||
/// The guarded operation failed (and the failure was recorded).
|
||||
Inner(E),
|
||||
}
|
||||
|
||||
impl<E: std::fmt::Display> std::fmt::Display for CircuitError<E> {
|
||||
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
|
||||
match self {
|
||||
Self::Open => write!(f, "circuit breaker is open"),
|
||||
Self::Inner(err) => write!(f, "guarded call failed: {err}"),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl<E: std::fmt::Debug + std::fmt::Display> std::error::Error for CircuitError<E> {}
|
||||
|
||||
struct Inner {
|
||||
state: AtomicU8,
|
||||
failures: AtomicU64,
|
||||
successes: AtomicU64,
|
||||
half_open_in_flight: AtomicUsize,
|
||||
opened_at: Mutex<Option<Instant>>,
|
||||
config: CircuitBreakerConfig,
|
||||
}
|
||||
|
||||
/// A thread-safe circuit breaker.
|
||||
///
|
||||
/// Construct with [`CircuitBreaker::new`]; use [`CircuitBreaker::call`] to
|
||||
/// guard a synchronous operation, or [`CircuitBreaker::allow_request`] +
|
||||
/// [`CircuitBreaker::record_success`] / [`CircuitBreaker::record_failure`] to
|
||||
/// guard an async call that cannot hold a borrow across an `.await`.
|
||||
#[derive(Clone)]
|
||||
pub struct CircuitBreaker {
|
||||
inner: Arc<Inner>,
|
||||
}
|
||||
|
||||
impl CircuitBreaker {
|
||||
/// Builds a new breaker with the given configuration.
|
||||
pub fn new(config: CircuitBreakerConfig) -> Self {
|
||||
Self {
|
||||
inner: Arc::new(Inner {
|
||||
state: AtomicU8::new(CLOSED),
|
||||
failures: AtomicU64::new(0),
|
||||
successes: AtomicU64::new(0),
|
||||
half_open_in_flight: AtomicUsize::new(0),
|
||||
opened_at: Mutex::new(None),
|
||||
config,
|
||||
}),
|
||||
}
|
||||
}
|
||||
|
||||
/// Returns the current circuit state, applying the open->half-open
|
||||
/// transition if the cooldown has elapsed.
|
||||
pub fn state(&self) -> CircuitState {
|
||||
match self.inner.state.load(Ordering::Acquire) {
|
||||
OPEN => {
|
||||
let opened = self.inner.opened_at.lock().unwrap();
|
||||
if opened
|
||||
.map(|t| t.elapsed() >= self.inner.config.open_timeout)
|
||||
.unwrap_or(false)
|
||||
{
|
||||
self.transition(HALF_OPEN);
|
||||
CircuitState::HalfOpen
|
||||
} else {
|
||||
CircuitState::Open
|
||||
}
|
||||
}
|
||||
HALF_OPEN => CircuitState::HalfOpen,
|
||||
_ => CircuitState::Closed,
|
||||
}
|
||||
}
|
||||
|
||||
/// Whether requests should currently be refused.
|
||||
pub fn is_open(&self) -> bool {
|
||||
self.state() == CircuitState::Open
|
||||
}
|
||||
|
||||
/// Whether a request may proceed right now without tripping the breaker.
|
||||
///
|
||||
/// Useful as a gate for an async call: call this before `.await`ing, then
|
||||
/// report the outcome via [`CircuitBreaker::record_success`] /
|
||||
/// [`CircuitBreaker::record_failure`].
|
||||
pub fn allow_request(&self) -> bool {
|
||||
match self.state() {
|
||||
CircuitState::Open => false,
|
||||
CircuitState::HalfOpen => {
|
||||
let in_flight = self.inner.half_open_in_flight.load(Ordering::Acquire);
|
||||
if in_flight < self.inner.config.half_open_max_calls {
|
||||
self.inner
|
||||
.half_open_in_flight
|
||||
.fetch_add(1, Ordering::AcqRel);
|
||||
true
|
||||
} else {
|
||||
false
|
||||
}
|
||||
}
|
||||
CircuitState::Closed => true,
|
||||
}
|
||||
}
|
||||
|
||||
/// Runs `f`, recording its outcome and refusing the call if the circuit is
|
||||
/// open.
|
||||
pub fn call<T, E, F>(&self, f: F) -> Result<T, CircuitError<E>>
|
||||
where
|
||||
F: FnOnce() -> Result<T, E>,
|
||||
{
|
||||
if !self.allow_request() {
|
||||
return Err(CircuitError::Open);
|
||||
}
|
||||
let result = f();
|
||||
match result {
|
||||
Ok(value) => {
|
||||
self.record_result(true);
|
||||
Ok(value)
|
||||
}
|
||||
Err(err) => {
|
||||
self.record_result(false);
|
||||
Err(CircuitError::Inner(err))
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Records that a (previously-admitted) call succeeded.
|
||||
///
|
||||
/// Use with [`CircuitBreaker::allow_request`] when guarding an async call.
|
||||
pub fn record_success(&self) {
|
||||
self.record_result(true)
|
||||
}
|
||||
|
||||
/// Records that a (previously-admitted) call failed.
|
||||
///
|
||||
/// Use with [`CircuitBreaker::allow_request`] when guarding an async call.
|
||||
pub fn record_failure(&self) {
|
||||
self.record_result(false)
|
||||
}
|
||||
|
||||
/// Resets the breaker to its initial `Closed` state.
|
||||
pub fn reset(&self) {
|
||||
self.inner.state.store(CLOSED, Ordering::Release);
|
||||
self.inner.failures.store(0, Ordering::Release);
|
||||
self.inner.successes.store(0, Ordering::Release);
|
||||
self.inner.half_open_in_flight.store(0, Ordering::Release);
|
||||
*self.inner.opened_at.lock().unwrap() = None;
|
||||
}
|
||||
|
||||
fn record_result(&self, success: bool) {
|
||||
match self.inner.state.load(Ordering::Acquire) {
|
||||
HALF_OPEN => {
|
||||
// A probe finished; release its slot regardless of outcome.
|
||||
self.inner
|
||||
.half_open_in_flight
|
||||
.fetch_sub(1, Ordering::AcqRel);
|
||||
if success {
|
||||
let successes = self.inner.successes.fetch_add(1, Ordering::AcqRel) + 1;
|
||||
if successes >= self.inner.config.close_successes {
|
||||
self.close();
|
||||
}
|
||||
} else {
|
||||
self.inner.failures.fetch_add(1, Ordering::AcqRel);
|
||||
self.open();
|
||||
}
|
||||
}
|
||||
OPEN => {
|
||||
if success {
|
||||
self.close();
|
||||
}
|
||||
}
|
||||
_ => {
|
||||
if success {
|
||||
self.inner.failures.store(0, Ordering::Release);
|
||||
} else {
|
||||
let failures = self.inner.failures.fetch_add(1, Ordering::AcqRel) + 1;
|
||||
if failures >= self.inner.config.failure_threshold {
|
||||
self.open();
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
fn transition(&self, next: u8) {
|
||||
self.inner.state.store(next, Ordering::Release);
|
||||
}
|
||||
|
||||
fn close(&self) {
|
||||
self.inner.state.store(CLOSED, Ordering::Release);
|
||||
self.inner.failures.store(0, Ordering::Release);
|
||||
self.inner.successes.store(0, Ordering::Release);
|
||||
self.inner.half_open_in_flight.store(0, Ordering::Release);
|
||||
self.inner.opened_at.lock().unwrap().take();
|
||||
}
|
||||
|
||||
fn open(&self) {
|
||||
self.inner.state.store(OPEN, Ordering::Release);
|
||||
self.inner.failures.store(0, Ordering::Release);
|
||||
self.inner.successes.store(0, Ordering::Release);
|
||||
self.inner.half_open_in_flight.store(0, Ordering::Release);
|
||||
*self.inner.opened_at.lock().unwrap() = Some(Instant::now());
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
fn breaker() -> CircuitBreaker {
|
||||
CircuitBreaker::new(CircuitBreakerConfig {
|
||||
failure_threshold: 3,
|
||||
open_timeout: Duration::from_millis(50),
|
||||
half_open_max_calls: 1,
|
||||
close_successes: 1,
|
||||
})
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn closed_start_state() {
|
||||
let b = breaker();
|
||||
assert_eq!(b.state(), CircuitState::Closed);
|
||||
assert!(!b.is_open());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn trips_open_after_threshold() {
|
||||
let b = breaker();
|
||||
for _ in 0..3 {
|
||||
let r: Result<(), CircuitError<u8>> = b.call(|| Err(1u8));
|
||||
assert!(r.is_err());
|
||||
}
|
||||
assert_eq!(b.state(), CircuitState::Open);
|
||||
assert!(b.is_open());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn open_refuses_calls() {
|
||||
let b = breaker();
|
||||
for _ in 0..3 {
|
||||
let _: Result<(), CircuitError<u8>> = b.call(|| Err(1u8));
|
||||
}
|
||||
let refused: Result<(), CircuitError<u8>> = b.call(|| Ok(()));
|
||||
assert!(matches!(refused, Err(CircuitError::Open)));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn success_resets_failure_count_in_closed() {
|
||||
let b = breaker();
|
||||
let _: Result<(), CircuitError<u8>> = b.call(|| Err(1u8));
|
||||
let _: Result<(), CircuitError<u8>> = b.call(|| Err(1u8));
|
||||
assert_eq!(b.state(), CircuitState::Closed);
|
||||
let r: Result<(), CircuitError<u8>> = b.call(|| Ok(()));
|
||||
assert!(r.is_ok());
|
||||
// Failure count reset — two more failures should not trip (needs 3).
|
||||
let _: Result<(), CircuitError<u8>> = b.call(|| Err(1u8));
|
||||
let _: Result<(), CircuitError<u8>> = b.call(|| Err(1u8));
|
||||
assert_eq!(b.state(), CircuitState::Closed);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn half_open_closes_after_success() {
|
||||
let b = breaker();
|
||||
for _ in 0..3 {
|
||||
let _: Result<(), CircuitError<u8>> = b.call(|| Err(1u8));
|
||||
}
|
||||
assert_eq!(b.state(), CircuitState::Open);
|
||||
|
||||
std::thread::sleep(Duration::from_millis(60));
|
||||
// Once half-open, a success closes the circuit.
|
||||
let admitted = {
|
||||
// allow_request true implies half-open probe admitted
|
||||
b.allow_request()
|
||||
};
|
||||
if admitted {
|
||||
b.record_success();
|
||||
}
|
||||
assert_eq!(b.state(), CircuitState::Closed);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn half_open_failure_reopens() {
|
||||
let b = breaker();
|
||||
for _ in 0..3 {
|
||||
let _: Result<(), CircuitError<u8>> = b.call(|| Err(1u8));
|
||||
}
|
||||
std::thread::sleep(Duration::from_millis(60));
|
||||
if b.allow_request() {
|
||||
b.record_failure();
|
||||
}
|
||||
assert_eq!(b.state(), CircuitState::Open);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn reset_returns_to_closed() {
|
||||
let b = breaker();
|
||||
for _ in 0..3 {
|
||||
let _: Result<(), CircuitError<u8>> = b.call(|| Err(1u8));
|
||||
}
|
||||
assert_eq!(b.state(), CircuitState::Open);
|
||||
b.reset();
|
||||
assert_eq!(b.state(), CircuitState::Closed);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn tracks_inner_error_and_success() {
|
||||
let b = breaker();
|
||||
let ok: Result<u32, CircuitError<u8>> = b.call(|| Ok(5u32));
|
||||
assert_eq!(ok.unwrap(), 5);
|
||||
let err: Result<u32, CircuitError<u8>> = b.call(|| Err(9u8));
|
||||
assert!(matches!(err, Err(CircuitError::Inner(9))));
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,180 @@
|
||||
//! A synchronous concurrency limiter (feature `traffic`).
|
||||
//!
|
||||
//! [`ConcurrencyLimiter`] bounds how many threads may hold a permit
|
||||
//! simultaneously, providing a way to cap concurrent access to an expensive
|
||||
//! resource (e.g. at most 10 tasks may hold a database connection or compress
|
||||
//! an image at once) without requiring a runtime. It is built on std's
|
||||
//! `Mutex` + `Condvar` and therefore works in plain, non-async threads.
|
||||
|
||||
use std::sync::Arc;
|
||||
use std::sync::{Condvar, Mutex};
|
||||
|
||||
struct Inner {
|
||||
state: Mutex<usize>,
|
||||
available: Condvar,
|
||||
max: usize,
|
||||
}
|
||||
|
||||
/// A thread-safe cap on concurrent in-flight sections.
|
||||
#[derive(Clone)]
|
||||
pub struct ConcurrencyLimiter {
|
||||
inner: Arc<Inner>,
|
||||
}
|
||||
|
||||
/// An RAII guard holding a concurrency permit.
|
||||
///
|
||||
/// Released (the permit returning to the limiter) when this guard is dropped.
|
||||
#[must_use = "dropping the permit releases the slot; if the caller wants to hold it, keep it alive"]
|
||||
pub struct ConcurrencyPermit {
|
||||
inner: Arc<Inner>,
|
||||
}
|
||||
|
||||
impl ConcurrencyLimiter {
|
||||
/// Builds a limiter allowing at most `max_concurrent` held permits.
|
||||
///
|
||||
/// # Panics
|
||||
///
|
||||
/// Panics if `max_concurrent` is zero.
|
||||
pub fn new(max_concurrent: usize) -> Self {
|
||||
assert!(max_concurrent > 0, "concurrency limit must be > 0");
|
||||
Self {
|
||||
inner: Arc::new(Inner {
|
||||
state: Mutex::new(max_concurrent),
|
||||
available: Condvar::new(),
|
||||
max: max_concurrent,
|
||||
}),
|
||||
}
|
||||
}
|
||||
|
||||
/// Blocks the calling thread until a permit is available, then returns it.
|
||||
pub fn acquire(&self) -> ConcurrencyPermit {
|
||||
let mut available = self.inner.state.lock().unwrap();
|
||||
while *available == 0 {
|
||||
available = self
|
||||
.inner
|
||||
.available
|
||||
.wait(available)
|
||||
.expect("concurrency limiter condvar poisoned");
|
||||
}
|
||||
*available -= 1;
|
||||
ConcurrencyPermit {
|
||||
inner: Arc::clone(&self.inner),
|
||||
}
|
||||
}
|
||||
|
||||
/// Attempts to acquire a permit without blocking.
|
||||
///
|
||||
/// Returns `None` if the limiter is currently at its maximum.
|
||||
pub fn try_acquire(&self) -> Option<ConcurrencyPermit> {
|
||||
let mut available = self.inner.state.lock().unwrap();
|
||||
if *available == 0 {
|
||||
return None;
|
||||
}
|
||||
*available -= 1;
|
||||
Some(ConcurrencyPermit {
|
||||
inner: Arc::clone(&self.inner),
|
||||
})
|
||||
}
|
||||
|
||||
/// How many permits are currently held.
|
||||
pub fn in_use(&self) -> usize {
|
||||
let available = *self.inner.state.lock().unwrap();
|
||||
self.inner.max - available
|
||||
}
|
||||
|
||||
/// The maximum number of concurrently held permits.
|
||||
pub fn max(&self) -> usize {
|
||||
self.inner.max
|
||||
}
|
||||
}
|
||||
|
||||
impl Drop for ConcurrencyPermit {
|
||||
fn drop(&mut self) {
|
||||
let mut available = self.inner.state.lock().unwrap();
|
||||
*available += 1;
|
||||
self.inner.available.notify_one();
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
use std::sync::atomic::{AtomicUsize, Ordering};
|
||||
use std::time::Duration;
|
||||
|
||||
#[test]
|
||||
fn acquires_and_releases() {
|
||||
let limiter = ConcurrencyLimiter::new(1);
|
||||
{
|
||||
let _permit = limiter.try_acquire().expect("should acquire");
|
||||
assert_eq!(limiter.in_use(), 1);
|
||||
}
|
||||
assert_eq!(limiter.in_use(), 0);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn try_acquire_fails_at_cap() {
|
||||
let limiter = ConcurrencyLimiter::new(2);
|
||||
let _a = limiter.try_acquire().expect("first");
|
||||
let _b = limiter.try_acquire().expect("second");
|
||||
assert!(limiter.try_acquire().is_none());
|
||||
drop(_a);
|
||||
assert!(limiter.try_acquire().is_some());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn acquire_blocks_until_slot_frees() {
|
||||
let limiter = ConcurrencyLimiter::new(1);
|
||||
let permit = limiter.try_acquire().unwrap();
|
||||
|
||||
let in_flight = Arc::new(AtomicUsize::new(0));
|
||||
let max_observed = Arc::new(AtomicUsize::new(0));
|
||||
let limiter2 = limiter.clone();
|
||||
let in_flight2 = Arc::clone(&in_flight);
|
||||
let max2 = Arc::clone(&max_observed);
|
||||
|
||||
let thread = std::thread::spawn(move || {
|
||||
let _permit = limiter2.acquire();
|
||||
in_flight2.fetch_add(1, Ordering::SeqCst);
|
||||
max2.fetch_max(in_flight2.load(Ordering::SeqCst), Ordering::SeqCst);
|
||||
});
|
||||
|
||||
std::thread::sleep(Duration::from_millis(40));
|
||||
assert_eq!(in_flight.load(Ordering::SeqCst), 0);
|
||||
// Explicitly drop permit so waiting thread unblocks
|
||||
drop(permit);
|
||||
thread.join().unwrap();
|
||||
assert_eq!(max_observed.load(Ordering::SeqCst), 1);
|
||||
assert_eq!(in_flight.load(Ordering::SeqCst), 1);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn limits_concurrent_sections() {
|
||||
let limiter = ConcurrencyLimiter::new(2);
|
||||
let in_flight = Arc::new(AtomicUsize::new(0));
|
||||
let max_observed = Arc::new(AtomicUsize::new(0));
|
||||
let mut threads = Vec::new();
|
||||
for _ in 0..8 {
|
||||
let limiter = limiter.clone();
|
||||
let in_flight = Arc::clone(&in_flight);
|
||||
let max_observed = Arc::clone(&max_observed);
|
||||
threads.push(std::thread::spawn(move || {
|
||||
let _permit = limiter.acquire();
|
||||
in_flight.fetch_add(1, Ordering::SeqCst);
|
||||
max_observed.fetch_max(in_flight.load(Ordering::SeqCst), Ordering::SeqCst);
|
||||
std::thread::sleep(Duration::from_millis(10));
|
||||
in_flight.fetch_sub(1, Ordering::SeqCst);
|
||||
}));
|
||||
}
|
||||
for t in threads {
|
||||
t.join().unwrap();
|
||||
}
|
||||
assert!(max_observed.load(Ordering::SeqCst) <= 2);
|
||||
}
|
||||
|
||||
#[test]
|
||||
#[should_panic]
|
||||
fn zero_limit_panics() {
|
||||
let _ = ConcurrencyLimiter::new(0);
|
||||
}
|
||||
}
|
||||
+475
@@ -0,0 +1,475 @@
|
||||
//! Minimal cron scheduling (feature `lifecycle`).
|
||||
//!
|
||||
//! [`CronSchedule`] parses the standard five-field cron expression
|
||||
//! (`minute hour day-of-month month day-of-week`) and computes next fire
|
||||
//! times. [`schedule`] drives a [`CronSchedule`] on a Tokio runtime, running a
|
||||
//! job each time it fires. The implementation is self-contained — no external
|
||||
//! cron crate — and deliberately ignores time zones (dates are interpreted in
|
||||
//! UTC).
|
||||
|
||||
use std::time::Duration;
|
||||
|
||||
use tracing::Instrument;
|
||||
|
||||
/// Error parsing a cron expression.
|
||||
#[derive(Debug, Clone, PartialEq, Eq)]
|
||||
pub struct CronParseError(pub String);
|
||||
|
||||
impl std::fmt::Display for CronParseError {
|
||||
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
|
||||
write!(f, "invalid cron expression: {}", self.0)
|
||||
}
|
||||
}
|
||||
|
||||
impl std::error::Error for CronParseError {}
|
||||
|
||||
/// Error scheduling a cron job.
|
||||
#[derive(Debug, Clone, PartialEq, Eq)]
|
||||
pub enum CronError {
|
||||
/// The expression did not parse.
|
||||
Parse(CronParseError),
|
||||
}
|
||||
|
||||
impl std::fmt::Display for CronError {
|
||||
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
|
||||
match self {
|
||||
Self::Parse(err) => write!(f, "cron error: {err}"),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl std::error::Error for CronError {}
|
||||
|
||||
/// A simple wall-clock instant used by cron computations.
|
||||
///
|
||||
/// Fields follow the chronological order used in cron expressions, resolved to
|
||||
/// the local (UTC) civil calendar.
|
||||
#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord)]
|
||||
pub struct CronTime {
|
||||
/// Year, e.g. `2026`.
|
||||
pub year: i32,
|
||||
/// Month, 1-12.
|
||||
pub month: u32,
|
||||
/// Day of month, 1-31.
|
||||
pub day: u32,
|
||||
/// Hour, 0-23.
|
||||
pub hour: u32,
|
||||
/// Minute, 0-59.
|
||||
pub minute: u32,
|
||||
/// Second, 0-59 (always 0 for scheduled fires).
|
||||
pub second: u32,
|
||||
}
|
||||
|
||||
/// A parsed cron schedule.
|
||||
#[derive(Debug, Clone)]
|
||||
pub struct CronSchedule {
|
||||
minutes: Vec<u8>,
|
||||
hours: Vec<u8>,
|
||||
days_of_month: Vec<u8>,
|
||||
months: Vec<u8>,
|
||||
days_of_week: Vec<u8>,
|
||||
}
|
||||
|
||||
/// A handle to a running cron job; dropping it aborts the task.
|
||||
#[derive(Debug)]
|
||||
pub struct CronJob {
|
||||
handle: tokio::task::JoinHandle<()>,
|
||||
}
|
||||
|
||||
impl CronJob {
|
||||
/// Cancels the scheduled job.
|
||||
pub fn cancel(self) {
|
||||
self.handle.abort();
|
||||
}
|
||||
}
|
||||
|
||||
impl CronSchedule {
|
||||
/// Parses the standard five-field cron expression
|
||||
/// `minute hour day-of-month month day-of-week`.
|
||||
///
|
||||
/// Supported tokens per field: `*`, `*/step`, a literal, a list
|
||||
/// (`a,b,c`), an inclusive range (`a-b`), and a ranged step (`a-b/step`).
|
||||
/// Day-of-week also accepts `SUN`..`SAT` (Sun=0) names. Day-of-month and
|
||||
/// day-of-week are combined with an OR when both are restricted, matching
|
||||
/// standard cron semantics.
|
||||
pub fn parse(expr: &str) -> Result<Self, CronParseError> {
|
||||
let fields: Vec<&str> = expr.split_whitespace().collect();
|
||||
if fields.len() != 5 {
|
||||
return Err(CronParseError(format!(
|
||||
"expected exactly 5 fields, got {}",
|
||||
fields.len()
|
||||
)));
|
||||
}
|
||||
let minutes = parse_field(fields[0], 0, 59, false)?;
|
||||
let hours = parse_field(fields[1], 0, 23, false)?;
|
||||
let days_of_month = parse_field(fields[2], 1, 31, false)?;
|
||||
let months = parse_field(fields[3], 1, 12, false)?;
|
||||
let days_of_week = parse_field(fields[4], 0, 6, true)?;
|
||||
|
||||
Ok(Self {
|
||||
minutes,
|
||||
hours,
|
||||
days_of_month,
|
||||
months,
|
||||
days_of_week,
|
||||
})
|
||||
}
|
||||
|
||||
/// Returns the next fire time strictly after `from`, or `None` if the
|
||||
/// expression cannot fire within the scan horizon (e.g. Feb 30).
|
||||
///
|
||||
/// Scans forward minute-by-minute up to roughly five years.
|
||||
pub fn next_after(&self, from: CronTime) -> Option<CronTime> {
|
||||
// Scan whole minutes (second = 0), starting strictly after `from`.
|
||||
let mut act = from.advance_minute();
|
||||
let horizon = 5 * 366 * 24 * 60; // minutes in ~5 years
|
||||
for _ in 0..horizon {
|
||||
if self.matches(act) {
|
||||
return Some(act);
|
||||
}
|
||||
act = act.advance_minute();
|
||||
}
|
||||
None
|
||||
}
|
||||
|
||||
fn matches(&self, t: CronTime) -> bool {
|
||||
let month_ok = self.months.contains(&(t.month as u8));
|
||||
let dom_ok = self.days_of_month.contains(&(t.day as u8));
|
||||
let dow_ok = self.days_of_week.contains(&(weekday(t) as u8));
|
||||
// Standard cron: if both dom and dow are restricted, a match on EITHER
|
||||
// is sufficient; if only one is restricted, it must match; if neither,
|
||||
// always true.
|
||||
let dom_restricted = self.days_of_month.len() < 31;
|
||||
let dow_restricted = self.days_of_week.len() < 7;
|
||||
let day_ok = match (dom_restricted, dow_restricted) {
|
||||
(true, true) => dom_ok || dow_ok,
|
||||
(true, false) => dom_ok,
|
||||
(false, true) => dow_ok,
|
||||
(false, false) => true,
|
||||
};
|
||||
month_ok
|
||||
&& day_ok
|
||||
&& self.hours.contains(&(t.hour as u8))
|
||||
&& self.minutes.contains(&(t.minute as u8))
|
||||
}
|
||||
|
||||
/// The next several fire times, for display/tests.
|
||||
pub fn next_five(&self, from: CronTime) -> Vec<CronTime> {
|
||||
let mut out = Vec::new();
|
||||
let mut cur = from;
|
||||
for _ in 0..5 {
|
||||
match self.next_after(cur) {
|
||||
Some(next) => {
|
||||
out.push(next);
|
||||
cur = next;
|
||||
}
|
||||
None => break,
|
||||
}
|
||||
}
|
||||
out
|
||||
}
|
||||
}
|
||||
|
||||
/// Spawns a task that runs `job` on `expr`'s schedule until cancelled.
|
||||
///
|
||||
/// The job is a closure returning a future; each invocation runs in a
|
||||
/// `mytheclipse_cron_task` tracing span. A fire time that was missed while the
|
||||
/// job was running is not backlogged — the next scheduled fire is used.
|
||||
pub fn schedule<F, Fut>(expr: &str, mut job: F) -> Result<CronJob, CronError>
|
||||
where
|
||||
F: FnMut() -> Fut + Send + 'static,
|
||||
Fut: std::future::Future<Output = ()> + Send + 'static,
|
||||
{
|
||||
let schedule = CronSchedule::parse(expr).map_err(CronError::Parse)?;
|
||||
let expr_owned = expr.to_string();
|
||||
let handle = tokio::spawn(async move {
|
||||
loop {
|
||||
let now = CronTime::now();
|
||||
let next = schedule.next_after(now);
|
||||
let next = match next {
|
||||
Some(next) => next,
|
||||
None => {
|
||||
// Unreachable schedule; wait and re-check periodically.
|
||||
tokio::time::sleep(Duration::from_secs(60)).await;
|
||||
continue;
|
||||
}
|
||||
};
|
||||
let now_ts = CronTime::now().to_timestamp();
|
||||
let until = tokio::time::Instant::now()
|
||||
+ Duration::from_secs(next.to_timestamp().saturating_sub(now_ts).max(1) as u64);
|
||||
tokio::time::sleep_until(until).await;
|
||||
let span = tracing::info_span!(
|
||||
"mytheclipse_cron_task",
|
||||
expr = %expr_owned
|
||||
);
|
||||
job().instrument(span).await;
|
||||
}
|
||||
});
|
||||
Ok(CronJob { handle })
|
||||
}
|
||||
|
||||
impl CronTime {
|
||||
/// The current wall-clock time (UTC), truncated to minutes for firing.
|
||||
pub fn now() -> Self {
|
||||
// std has no direct civil-date conversion; derive from SystemTime.
|
||||
now_utc()
|
||||
}
|
||||
|
||||
fn to_timestamp(self) -> i64 {
|
||||
days_from_civil(self.year, self.month, self.day) * 86_400
|
||||
+ (self.hour as i64) * 3_600
|
||||
+ (self.minute as i64) * 60
|
||||
+ self.second as i64
|
||||
}
|
||||
|
||||
fn advance_minute(self) -> Self {
|
||||
let mut t = self.add_seconds(60);
|
||||
t.second = 0;
|
||||
t
|
||||
}
|
||||
|
||||
fn add_seconds(self, secs: i64) -> Self {
|
||||
let ts = self.to_timestamp() + secs;
|
||||
from_timestamp(ts)
|
||||
}
|
||||
}
|
||||
|
||||
/// Weekday: 0 = Sunday .. 6 = Saturday (matches cron).
|
||||
fn weekday(t: CronTime) -> u32 {
|
||||
// 1970-01-01 was a Thursday (weekday 4).
|
||||
let days = days_from_civil(t.year, t.month, t.day);
|
||||
// ((days + 4) mod 7) => 0=Sunday
|
||||
((days % 7 + 11) % 7) as u32
|
||||
}
|
||||
|
||||
/// Days since 1970-01-01 for a civil date (Howard Hinnant's algorithm).
|
||||
fn days_from_civil(y: i32, m: u32, d: u32) -> i64 {
|
||||
let y = y as i64 - (m as i64 <= 2) as i64;
|
||||
let era = (if y >= 0 { y } else { y - 399 }) / 400;
|
||||
let yoe = y - era * 400; // [0, 399]
|
||||
let mp = (m as i64 + 9) % 12; // March=0
|
||||
let doy = (153 * mp + 2) / 5 + d as i64 - 1;
|
||||
let doe = yoe * 365 + yoe / 4 - yoe / 100 + doy;
|
||||
era * 146_097 + doe - 719_468
|
||||
}
|
||||
|
||||
/// Converts a Unix timestamp (seconds) to a civil [`CronTime`].
|
||||
fn from_timestamp(ts: i64) -> CronTime {
|
||||
let days = ts.div_euclid(86_400);
|
||||
let z = days + 719_468; // days since civil epoch
|
||||
let era = if z >= 0 { z } else { z - 146_096 } / 146_097;
|
||||
let doe = z - era * 146_097; // [0, 146096]
|
||||
let yoe = (doe - doe / 1_460 + doe / 36_524 - doe / 146_096) / 365;
|
||||
let y = yoe + era * 400;
|
||||
let doy = doe - (365 * yoe + yoe / 4 - yoe / 100);
|
||||
let mp = (5 * doy + 2) / 153;
|
||||
let d = doy - (153 * mp + 2) / 5 + 1;
|
||||
let m = if mp < 10 { mp + 3 } else { mp - 9 } as u32;
|
||||
let y = if m <= 2 { y + 1 } else { y } as i32;
|
||||
|
||||
let (h, mi, s) = split_seconds(ts);
|
||||
CronTime {
|
||||
year: y,
|
||||
month: m,
|
||||
day: d as u32,
|
||||
hour: h,
|
||||
minute: mi,
|
||||
second: s,
|
||||
}
|
||||
}
|
||||
|
||||
fn split_seconds(ts: i64) -> (u32, u32, u32) {
|
||||
let rem = ts.rem_euclid(86_400);
|
||||
(
|
||||
(rem / 3_600) as u32,
|
||||
((rem % 3_600) / 60) as u32,
|
||||
(rem % 60) as u32,
|
||||
)
|
||||
}
|
||||
|
||||
/// Returns the current UTC civil time plus second accuracy from SystemTime.
|
||||
fn now_utc() -> CronTime {
|
||||
let now = std::time::SystemTime::now()
|
||||
.duration_since(std::time::UNIX_EPOCH)
|
||||
.unwrap_or_default();
|
||||
from_timestamp(now.as_secs() as i64)
|
||||
}
|
||||
|
||||
/// Parses a single cron field into a sorted list of allowed values.
|
||||
fn parse_field(
|
||||
field: &str,
|
||||
min: u8,
|
||||
max: u8,
|
||||
allow_names: bool,
|
||||
) -> Result<Vec<u8>, CronParseError> {
|
||||
let mut out = Vec::new();
|
||||
for part in field.split(',') {
|
||||
let part = part.trim();
|
||||
if part.is_empty() {
|
||||
return Err(CronParseError(format!("empty field element in `{field}`")));
|
||||
}
|
||||
// Parse `base` optionally followed by `/step`.
|
||||
let (range, step) = match part.split_once('/') {
|
||||
Some((r, s)) => (r, parse_num(s, "step")?),
|
||||
None => (part, 1),
|
||||
};
|
||||
// Parse `*`, `a-b`, or literal.
|
||||
match range {
|
||||
"*" => {
|
||||
let mut v = min;
|
||||
while v <= max {
|
||||
out.push(v);
|
||||
v += step;
|
||||
}
|
||||
}
|
||||
_ if range.contains('-') => {
|
||||
let (a, b) = range
|
||||
.split_once('-')
|
||||
.ok_or_else(|| CronParseError(format!("bad range `{range}`")))?;
|
||||
let a = parse_value(a, min, max, allow_names)?;
|
||||
let b = parse_value(b, min, max, allow_names)?;
|
||||
if a > b {
|
||||
return Err(CronParseError(format!("range start {a} > end {b}")));
|
||||
}
|
||||
let mut v = a;
|
||||
while v <= b {
|
||||
out.push(v);
|
||||
v += step;
|
||||
}
|
||||
}
|
||||
_ => {
|
||||
let v = parse_value(range, min, max, allow_names)?;
|
||||
out.push(v);
|
||||
}
|
||||
}
|
||||
}
|
||||
out.sort_unstable();
|
||||
out.dedup();
|
||||
Ok(out)
|
||||
}
|
||||
|
||||
fn parse_num(s: &str, what: &str) -> Result<u8, CronParseError> {
|
||||
s.parse::<u8>()
|
||||
.map_err(|_| CronParseError(format!("invalid {what} `{s}`")))
|
||||
}
|
||||
|
||||
fn parse_value(s: &str, min: u8, max: u8, allow_names: bool) -> Result<u8, CronParseError> {
|
||||
let upper = s.to_uppercase();
|
||||
// Day-of-week names.
|
||||
if allow_names {
|
||||
let name = match upper.as_str() {
|
||||
"SUN" => Some(0),
|
||||
"MON" => Some(1),
|
||||
"TUE" => Some(2),
|
||||
"WED" => Some(3),
|
||||
"THU" => Some(4),
|
||||
"FRI" => Some(5),
|
||||
"SAT" => Some(6),
|
||||
_ => None,
|
||||
};
|
||||
if let Some(v) = name {
|
||||
return Ok(v);
|
||||
}
|
||||
}
|
||||
let v = parse_num(s, "value")?;
|
||||
if v < min || v > max {
|
||||
return Err(CronParseError(format!(
|
||||
"value {v} out of range {min}..{max}"
|
||||
)));
|
||||
}
|
||||
Ok(v)
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
fn t(y: i32, m: u32, d: u32, h: u32, mi: u32) -> CronTime {
|
||||
CronTime {
|
||||
year: y,
|
||||
month: m,
|
||||
day: d,
|
||||
hour: h,
|
||||
minute: mi,
|
||||
second: 0,
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn parses_star() {
|
||||
let s = CronSchedule::parse("* * * * *").unwrap();
|
||||
assert_eq!(s.minutes.len(), 60);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn parses_list_and_range() {
|
||||
let s = CronSchedule::parse("1,15 9-17 * * 1-5").unwrap();
|
||||
assert_eq!(s.minutes, vec![1, 15]);
|
||||
assert_eq!(s.hours, (9..=17).collect::<Vec<_>>());
|
||||
assert_eq!(s.days_of_week, vec![1, 2, 3, 4, 5]);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn parses_step_and_names() {
|
||||
let s = CronSchedule::parse("*/15 * * * MON,WED").unwrap();
|
||||
assert_eq!(s.minutes, vec![0, 15, 30, 45]);
|
||||
assert_eq!(s.days_of_week, vec![1, 3]);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn rejects_too_few_fields() {
|
||||
assert!(CronSchedule::parse("* * * *").is_err());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn rejects_out_of_range() {
|
||||
assert!(CronSchedule::parse("60 * * * *").is_err());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn next_after_every_minute() {
|
||||
let s = CronSchedule::parse("* * * * *").unwrap();
|
||||
let next = s.next_after(t(2026, 8, 28, 10, 30)).unwrap();
|
||||
assert_eq!(next, t(2026, 8, 28, 10, 31));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn next_after_midnight() {
|
||||
let s = CronSchedule::parse("0 0 * * *").unwrap();
|
||||
let next = s.next_after(t(2026, 8, 28, 23, 59)).unwrap();
|
||||
assert_eq!(next, t(2026, 8, 29, 0, 0));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn next_after_weekday_only() {
|
||||
// 2026-08-28 is a Friday (5). "0 9 * * MON-FRI" fires next Monday.
|
||||
let s = CronSchedule::parse("0 9 * * MON-FRI").unwrap();
|
||||
let next = s.next_after(t(2026, 8, 28, 23, 0)).unwrap();
|
||||
// Friday 23:00 -> next is Monday 09:00 (Aug 31).
|
||||
assert_eq!(next, t(2026, 8, 31, 9, 0));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn next_after_leap_year_feb() {
|
||||
let s = CronSchedule::parse("0 12 29 2 *").unwrap();
|
||||
// Next Feb 29 after 2026 is 2028.
|
||||
let next = s.next_after(t(2026, 1, 1, 0, 0)).unwrap();
|
||||
assert_eq!(next, t(2028, 2, 29, 12, 0));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn impossible_expression_returns_none() {
|
||||
let s = CronSchedule::parse("0 0 30 2 *").unwrap(); // Feb 30 never exists
|
||||
assert!(s.next_after(t(2026, 1, 1, 0, 0)).is_none());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn next_five_every_hour() {
|
||||
let s = CronSchedule::parse("0 * * * *").unwrap();
|
||||
let five = s.next_five(t(2026, 8, 28, 9, 45));
|
||||
assert_eq!(five.len(), 5);
|
||||
assert_eq!(five[0], t(2026, 8, 28, 10, 0));
|
||||
assert_eq!(five[1], t(2026, 8, 28, 11, 0));
|
||||
assert_eq!(five[4], t(2026, 8, 28, 14, 0));
|
||||
}
|
||||
}
|
||||
+42
-2
@@ -1,18 +1,45 @@
|
||||
//! Shared error type for mytheclipse execution primitives.
|
||||
|
||||
/// Errors surfaced by mytheclipse's panic-isolated execution primitives.
|
||||
/// Errors surfaced by mytheclipse's execution primitives.
|
||||
///
|
||||
/// Marked `#[non_exhaustive]` so new variants can be added without a
|
||||
/// breaking change; downstream `match` expressions should include a
|
||||
/// wildcard arm.
|
||||
#[non_exhaustive]
|
||||
#[derive(Debug)]
|
||||
#[derive(Debug, PartialEq, Eq)]
|
||||
pub enum MytheclipseError {
|
||||
/// A closure submitted to [`crate::compute::compute`] panicked.
|
||||
///
|
||||
/// The contained string is a best-effort rendering of the panic
|
||||
/// payload; the compute thread pool itself remains usable afterward.
|
||||
ComputePanic(String),
|
||||
/// A deadline elapsed before the wrapped future completed
|
||||
/// ([`crate::timeout::with_timeout`]).
|
||||
Timeout,
|
||||
/// A request was refused because the circuit breaker is open
|
||||
/// ([`crate::circuit_breaker`]).
|
||||
CircuitOpen,
|
||||
/// A retry loop gave up after exhausting its maximum attempts
|
||||
/// ([`crate::retry()`]). The contained count is the number of attempts made.
|
||||
RetryExhausted { attempts: u32 },
|
||||
/// A request was refused because the rate limiter had no tokens left
|
||||
/// ([`crate::ratelimit`]).
|
||||
RateLimited,
|
||||
/// A queue was full and the configured overflow policy rejected the item
|
||||
/// ([`crate::backpressure`]).
|
||||
QueueFull(String),
|
||||
/// A concurrency limiter was already at its maximum and refused a request
|
||||
/// ([`crate::concurrency`]).
|
||||
ConcurrencyLimitExceeded,
|
||||
/// Shutdown has been requested for the process or manager
|
||||
/// ([`crate::shutdown`]).
|
||||
Shutdown,
|
||||
/// An observability operation failed, e.g. metric export
|
||||
/// ([`crate::metrics`]).
|
||||
Metrics(String),
|
||||
/// Invalid configuration or input, e.g. a malformed cron expression
|
||||
/// ([`crate::cron`]).
|
||||
Config(String),
|
||||
}
|
||||
|
||||
impl std::fmt::Display for MytheclipseError {
|
||||
@@ -21,6 +48,19 @@ impl std::fmt::Display for MytheclipseError {
|
||||
Self::ComputePanic(message) => {
|
||||
write!(f, "compute closure panicked: {message}")
|
||||
}
|
||||
Self::Timeout => write!(f, "operation timed out"),
|
||||
Self::CircuitOpen => write!(f, "circuit breaker is open"),
|
||||
Self::RetryExhausted { attempts } => {
|
||||
write!(f, "retry exhausted after {attempts} attempts")
|
||||
}
|
||||
Self::RateLimited => write!(f, "rate limit exceeded"),
|
||||
Self::QueueFull(detail) => write!(f, "queue is full: {detail}"),
|
||||
Self::ConcurrencyLimitExceeded => {
|
||||
write!(f, "concurrency limit exceeded")
|
||||
}
|
||||
Self::Shutdown => write!(f, "shutdown requested"),
|
||||
Self::Metrics(detail) => write!(f, "metrics error: {detail}"),
|
||||
Self::Config(detail) => write!(f, "configuration error: {detail}"),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
+73
-7
@@ -1,8 +1,9 @@
|
||||
//! # mytheclipse
|
||||
//!
|
||||
//! Resource-aware abstractions for async I/O, heavy compute, and
|
||||
//! background queue management, built on a single lazily-initialized
|
||||
//! global engine context.
|
||||
//! Resource-aware abstractions for async I/O, heavy compute, background queue
|
||||
//! management, resiliency, traffic control, lifecycle, and observability —
|
||||
//! built on a single lazily-initialized global engine context plus a set of
|
||||
//! self-contained, constructible utilities.
|
||||
//!
|
||||
//! Call [`init`] once at startup (or simply let the first call to any
|
||||
//! entry point below trigger it lazily) and then use whichever of the
|
||||
@@ -11,11 +12,18 @@
|
||||
//! - [`spawn_io`] (feature `io`) — spawn an async I/O task, tracing-instrumented.
|
||||
//! - [`compute()`] (feature `compute`) — run CPU-bound work on a sized Rayon pool, panic-isolated.
|
||||
//! - [`spawn_bg`] (feature `bg`) — spawn a background task under bounded concurrency.
|
||||
//! - [`retry()`] / [`circuit_breaker`] / [`timeout()`] (feature `resiliency`) — fault tolerance.
|
||||
//! - [`ratelimit`] / [`backpressure`] / [`concurrency`] (feature `traffic`) — load control.
|
||||
//! - [`shutdown`] / [`cron`] (feature `lifecycle`) — lifecycle management.
|
||||
//! - [`metrics`] / [`panic_tracker`] (feature `observability`) — runtime visibility.
|
||||
//!
|
||||
//! Enable the `full` feature to pull in all three at once.
|
||||
//! Enable the `full` feature to pull in all of the above at once. The three
|
||||
//! execution primitives are sized from the host's logical core count via the
|
||||
//! engine context; the resiliency/traffic/lifecycle/observability utilities
|
||||
//! are self-contained and constructed explicitly (e.g.
|
||||
//! `RateLimiter::new(...)`, `CircuitBreaker::new(...)`).
|
||||
|
||||
pub mod context;
|
||||
#[cfg(feature = "compute")]
|
||||
pub mod error;
|
||||
|
||||
#[cfg(feature = "io")]
|
||||
@@ -27,9 +35,37 @@ pub mod compute;
|
||||
#[cfg(feature = "bg")]
|
||||
pub mod bg;
|
||||
|
||||
pub use context::{context, EngineContext};
|
||||
#[cfg(feature = "resiliency")]
|
||||
pub mod retry;
|
||||
|
||||
#[cfg(feature = "compute")]
|
||||
#[cfg(feature = "resiliency")]
|
||||
pub mod circuit_breaker;
|
||||
|
||||
#[cfg(feature = "resiliency")]
|
||||
pub mod timeout;
|
||||
|
||||
#[cfg(feature = "traffic")]
|
||||
pub mod ratelimit;
|
||||
|
||||
#[cfg(feature = "traffic")]
|
||||
pub mod backpressure;
|
||||
|
||||
#[cfg(feature = "traffic")]
|
||||
pub mod concurrency;
|
||||
|
||||
#[cfg(feature = "lifecycle")]
|
||||
pub mod shutdown;
|
||||
|
||||
#[cfg(feature = "lifecycle")]
|
||||
pub mod cron;
|
||||
|
||||
#[cfg(feature = "observability")]
|
||||
pub mod metrics;
|
||||
|
||||
#[cfg(feature = "observability")]
|
||||
pub mod panic_tracker;
|
||||
|
||||
pub use context::{context, EngineContext};
|
||||
pub use error::MytheclipseError;
|
||||
|
||||
#[cfg(feature = "io")]
|
||||
@@ -41,6 +77,36 @@ pub use compute::compute;
|
||||
#[cfg(feature = "bg")]
|
||||
pub use bg::spawn_bg;
|
||||
|
||||
#[cfg(feature = "resiliency")]
|
||||
pub use retry::{retry, JitterKind, RetryConfig, RetryError};
|
||||
|
||||
#[cfg(feature = "resiliency")]
|
||||
pub use circuit_breaker::{CircuitBreaker, CircuitBreakerConfig, CircuitError, CircuitState};
|
||||
|
||||
#[cfg(feature = "resiliency")]
|
||||
pub use timeout::{timeout, with_timeout, Timeout, TimeoutError};
|
||||
|
||||
#[cfg(feature = "traffic")]
|
||||
pub use ratelimit::{RateLimitError, RateLimiter};
|
||||
|
||||
#[cfg(feature = "traffic")]
|
||||
pub use backpressure::{BackpressureError, BackpressureQueue, OverflowPolicy};
|
||||
|
||||
#[cfg(feature = "traffic")]
|
||||
pub use concurrency::{ConcurrencyLimiter, ConcurrencyPermit};
|
||||
|
||||
#[cfg(feature = "lifecycle")]
|
||||
pub use shutdown::{ShutdownManager, ShutdownSignal};
|
||||
|
||||
#[cfg(feature = "lifecycle")]
|
||||
pub use cron::{schedule, CronError, CronJob, CronParseError, CronSchedule};
|
||||
|
||||
#[cfg(feature = "observability")]
|
||||
pub use metrics::{MetricsCollector, MetricsSnapshot};
|
||||
|
||||
#[cfg(feature = "observability")]
|
||||
pub use panic_tracker::{PanicGuard, PanicInfo, PanicTracker};
|
||||
|
||||
/// Bootstraps the global [`EngineContext`].
|
||||
///
|
||||
/// See [`context::init`] for full semantics: this is safe to call any
|
||||
|
||||
+337
@@ -0,0 +1,337 @@
|
||||
//! A centralized, thread-safe metrics collector (feature `observability`).
|
||||
//!
|
||||
//! [`MetricsCollector`] records runtime health statistics — active threads,
|
||||
//! remaining queue capacity, average task duration, custom counters and gauges
|
||||
//! — and can render them in the [Prometheus text exposition format](https://prometheus.io/docs/instrumenting/exposition_formats/)
|
||||
//! for a scraper, or as a structured [`MetricsSnapshot`] that a downstream adapter
|
||||
//! can forward (e.g. to OpenTelemetry).
|
||||
|
||||
use std::collections::HashMap;
|
||||
use std::sync::atomic::{AtomicU64, AtomicUsize, Ordering};
|
||||
use std::sync::{Arc, Mutex};
|
||||
use std::time::Duration;
|
||||
|
||||
/// A single histogram bucket: accumulates `count` observations and their
|
||||
/// total duration for an average.
|
||||
#[derive(Debug, Clone, Default)]
|
||||
struct Histogram {
|
||||
count: u64,
|
||||
total_micros: u128,
|
||||
}
|
||||
|
||||
#[derive(Default)]
|
||||
struct Inner {
|
||||
active_threads: AtomicUsize,
|
||||
queue_capacity_total: AtomicUsize,
|
||||
queue_capacity_remaining: AtomicUsize,
|
||||
task_count: AtomicU64,
|
||||
total_task_micros: AtomicU64,
|
||||
counters: Mutex<HashMap<String, u64>>,
|
||||
gauges: Mutex<HashMap<String, f64>>,
|
||||
histograms: Mutex<HashMap<String, Histogram>>,
|
||||
}
|
||||
|
||||
/// A thread-safe handle to a metrics collector.
|
||||
///
|
||||
/// Cheap to clone (shared state). Collects data updated from any thread and
|
||||
/// exports it on demand.
|
||||
#[derive(Clone, Default)]
|
||||
pub struct MetricsCollector {
|
||||
inner: Arc<Inner>,
|
||||
}
|
||||
|
||||
impl MetricsCollector {
|
||||
/// Builds an empty collector.
|
||||
pub fn new() -> Self {
|
||||
Self::default()
|
||||
}
|
||||
|
||||
/// Records that a task ran for `duration`, updating the running task
|
||||
/// count and average.
|
||||
pub fn record_task(&self, duration: Duration) {
|
||||
self.inner.task_count.fetch_add(1, Ordering::AcqRel);
|
||||
self.inner
|
||||
.total_task_micros
|
||||
.fetch_add(duration.as_micros() as u64, Ordering::AcqRel);
|
||||
}
|
||||
|
||||
/// Sets the observed number of active worker threads.
|
||||
pub fn set_active_threads(&self, n: usize) {
|
||||
self.inner.active_threads.store(n, Ordering::Release);
|
||||
}
|
||||
|
||||
/// Sets the queue's total capacity.
|
||||
pub fn set_queue_capacity(&self, n: usize) {
|
||||
self.inner.queue_capacity_total.store(n, Ordering::Release);
|
||||
}
|
||||
|
||||
/// Sets the queue's currently remaining capacity.
|
||||
pub fn set_queue_remaining(&self, n: usize) {
|
||||
self.inner
|
||||
.queue_capacity_remaining
|
||||
.store(n, Ordering::Release);
|
||||
}
|
||||
|
||||
/// Increments a counter by `by`.
|
||||
pub fn inc_counter(&self, name: &str, by: u64) {
|
||||
*self
|
||||
.inner
|
||||
.counters
|
||||
.lock()
|
||||
.unwrap()
|
||||
.entry(name.to_string())
|
||||
.or_insert(0) += by;
|
||||
}
|
||||
|
||||
/// Sets a gauge to `value`.
|
||||
pub fn set_gauge(&self, name: &str, value: f64) {
|
||||
self.inner
|
||||
.gauges
|
||||
.lock()
|
||||
.unwrap()
|
||||
.insert(name.to_string(), value);
|
||||
}
|
||||
|
||||
/// Records `duration` into the histogram named `name`.
|
||||
pub fn observe(&self, name: &str, duration: Duration) {
|
||||
let mut histos = self.inner.histograms.lock().unwrap();
|
||||
let h = histos.entry(name.to_string()).or_default();
|
||||
h.count += 1;
|
||||
h.total_micros += duration.as_micros();
|
||||
}
|
||||
|
||||
/// Total number of recorded tasks.
|
||||
pub fn task_count(&self) -> u64 {
|
||||
self.inner.task_count.load(Ordering::Acquire)
|
||||
}
|
||||
|
||||
/// Average task duration, if any tasks have been recorded.
|
||||
pub fn avg_task_duration(&self) -> Option<Duration> {
|
||||
let count = self.task_count();
|
||||
if count == 0 {
|
||||
return None;
|
||||
}
|
||||
let total = self.inner.total_task_micros.load(Ordering::Acquire);
|
||||
Some(Duration::from_micros(total / count))
|
||||
}
|
||||
|
||||
/// Number of active worker threads last recorded.
|
||||
pub fn active_threads(&self) -> usize {
|
||||
self.inner.active_threads.load(Ordering::Acquire)
|
||||
}
|
||||
|
||||
/// Takes a structured snapshot of the collector's state.
|
||||
pub fn snapshot(&self) -> MetricsSnapshot {
|
||||
MetricsSnapshot {
|
||||
active_threads: self.active_threads(),
|
||||
queue_capacity_total: self.inner.queue_capacity_total.load(Ordering::Acquire),
|
||||
queue_capacity_remaining: self.inner.queue_capacity_remaining.load(Ordering::Acquire),
|
||||
task_count: self.task_count(),
|
||||
avg_task_duration_micros: self.avg_task_duration().map(|d| d.as_micros() as u64),
|
||||
counters: self
|
||||
.inner
|
||||
.counters
|
||||
.lock()
|
||||
.unwrap()
|
||||
.iter()
|
||||
.map(|(k, v)| (k.clone(), *v))
|
||||
.collect(),
|
||||
gauges: self
|
||||
.inner
|
||||
.gauges
|
||||
.lock()
|
||||
.unwrap()
|
||||
.iter()
|
||||
.map(|(k, v)| (k.clone(), *v))
|
||||
.collect(),
|
||||
histograms: self
|
||||
.inner
|
||||
.histograms
|
||||
.lock()
|
||||
.unwrap()
|
||||
.iter()
|
||||
.map(|(k, h)| {
|
||||
(
|
||||
k.clone(),
|
||||
HistogramSnapshot {
|
||||
count: h.count,
|
||||
avg_micros: if h.count > 0 {
|
||||
Some((h.total_micros / h.count as u128) as u64)
|
||||
} else {
|
||||
None
|
||||
},
|
||||
},
|
||||
)
|
||||
})
|
||||
.collect(),
|
||||
}
|
||||
}
|
||||
|
||||
/// Renders the collector in the Prometheus text exposition format.
|
||||
///
|
||||
/// The output is suitable to serve with
|
||||
/// `Content-Type: text/plain; version=0.0.4`.
|
||||
pub fn export_prometheus(&self) -> String {
|
||||
let mut out = String::new();
|
||||
let snap = self.snapshot();
|
||||
|
||||
out.push_str("# HELP mytheclipse_active_threads Number of active worker threads.\n");
|
||||
out.push_str("# TYPE mytheclipse_active_threads gauge\n");
|
||||
out.push_str(&format!(
|
||||
"mytheclipse_active_threads {}\n",
|
||||
snap.active_threads
|
||||
));
|
||||
out.push_str("# HELP mytheclipse_queue_capacity_total Total queue capacity.\n");
|
||||
out.push_str("# TYPE mytheclipse_queue_capacity_total gauge\n");
|
||||
out.push_str(&format!(
|
||||
"mytheclipse_queue_capacity_total {}\n",
|
||||
snap.queue_capacity_total
|
||||
));
|
||||
out.push_str("# HELP mytheclipse_queue_capacity_remaining Remaining queue capacity.\n");
|
||||
out.push_str("# TYPE mytheclipse_queue_capacity_remaining gauge\n");
|
||||
out.push_str(&format!(
|
||||
"mytheclipse_queue_capacity_remaining {}\n",
|
||||
snap.queue_capacity_remaining
|
||||
));
|
||||
out.push_str("# HELP mytheclipse_task_count Total tasks recorded.\n");
|
||||
out.push_str("# TYPE mytheclipse_task_count counter\n");
|
||||
out.push_str(&format!("mytheclipse_task_count {}\n", snap.task_count));
|
||||
if let Some(avg) = snap.avg_task_duration_micros {
|
||||
out.push_str(
|
||||
"# HELP mytheclipse_task_duration_avg Average task duration in microseconds.\n",
|
||||
);
|
||||
out.push_str("# TYPE mytheclipse_task_duration_avg gauge\n");
|
||||
out.push_str(&format!("mytheclipse_task_duration_avg {avg}\n"));
|
||||
}
|
||||
|
||||
let mut counters: Vec<_> = snap.counters.into_iter().collect();
|
||||
counters.sort_by(|a, b| a.0.cmp(&b.0));
|
||||
for (name, value) in counters {
|
||||
out.push_str(&format!("# TYPE {name} counter\n"));
|
||||
out.push_str(&format!("{name} {value}\n"));
|
||||
}
|
||||
|
||||
let mut gauges: Vec<_> = snap.gauges.into_iter().collect();
|
||||
gauges.sort_by(|a, b| a.0.cmp(&b.0));
|
||||
for (name, value) in gauges {
|
||||
out.push_str(&format!("# TYPE {name} gauge\n"));
|
||||
out.push_str(&format!("{name} {value}\n"));
|
||||
}
|
||||
|
||||
let mut histos: Vec<_> = snap.histograms.into_iter().collect();
|
||||
histos.sort_by(|a, b| a.0.cmp(&b.0));
|
||||
for (name, h) in histos {
|
||||
if let Some(avg) = h.avg_micros {
|
||||
out.push_str(&format!("# TYPE {name}_count counter\n"));
|
||||
out.push_str(&format!("{name}_count {}\n", h.count));
|
||||
out.push_str(&format!("# TYPE {name}_avg gauge\n"));
|
||||
out.push_str(&format!("{name}_avg {avg}\n"));
|
||||
}
|
||||
}
|
||||
|
||||
out
|
||||
}
|
||||
}
|
||||
|
||||
/// A structured, serializable view of a [`MetricsCollector`], suitable for
|
||||
/// forwarding to OpenTelemetry or another backend.
|
||||
#[derive(Debug, Clone)]
|
||||
pub struct MetricsSnapshot {
|
||||
/// Number of active worker threads last recorded.
|
||||
pub active_threads: usize,
|
||||
/// Total queue capacity last recorded.
|
||||
pub queue_capacity_total: usize,
|
||||
/// Remaining queue capacity last recorded.
|
||||
pub queue_capacity_remaining: usize,
|
||||
/// Total tasks recorded.
|
||||
pub task_count: u64,
|
||||
/// Average task duration in microseconds, if any.
|
||||
pub avg_task_duration_micros: Option<u64>,
|
||||
/// Named counters.
|
||||
pub counters: HashMap<String, u64>,
|
||||
/// Named gauges.
|
||||
pub gauges: HashMap<String, f64>,
|
||||
/// Named histogram aggregates.
|
||||
pub histograms: HashMap<String, HistogramSnapshot>,
|
||||
}
|
||||
|
||||
/// Aggregated view of one histogram.
|
||||
#[derive(Debug, Clone)]
|
||||
pub struct HistogramSnapshot {
|
||||
/// Number of observations.
|
||||
pub count: u64,
|
||||
/// Average observation value in microseconds, if any.
|
||||
pub avg_micros: Option<u64>,
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
#[test]
|
||||
fn records_task_and_average() {
|
||||
let m = MetricsCollector::new();
|
||||
assert_eq!(m.task_count(), 0);
|
||||
assert!(m.avg_task_duration().is_none());
|
||||
|
||||
m.record_task(Duration::from_millis(10));
|
||||
m.record_task(Duration::from_millis(20));
|
||||
assert_eq!(m.task_count(), 2);
|
||||
assert_eq!(m.avg_task_duration().unwrap(), Duration::from_millis(15));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn tracks_threads_and_queue() {
|
||||
let m = MetricsCollector::new();
|
||||
m.set_active_threads(8);
|
||||
m.set_queue_capacity(100);
|
||||
m.set_queue_remaining(42);
|
||||
let snap = m.snapshot();
|
||||
assert_eq!(snap.active_threads, 8);
|
||||
assert_eq!(snap.queue_capacity_total, 100);
|
||||
assert_eq!(snap.queue_capacity_remaining, 42);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn custom_counters_and_gauges() {
|
||||
let m = MetricsCollector::new();
|
||||
m.inc_counter("reqs", 3);
|
||||
m.inc_counter("reqs", 2);
|
||||
m.set_gauge("temp", 21.5);
|
||||
let snap = m.snapshot();
|
||||
assert_eq!(snap.counters["reqs"], 5);
|
||||
assert_eq!(snap.gauges["temp"], 21.5);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn observe_accumulates_histogram() {
|
||||
let m = MetricsCollector::new();
|
||||
m.observe("latency", Duration::from_millis(100));
|
||||
m.observe("latency", Duration::from_millis(300));
|
||||
let snap = m.snapshot();
|
||||
let h = &snap.histograms["latency"];
|
||||
assert_eq!(h.count, 2);
|
||||
assert_eq!(h.avg_micros.unwrap(), 200_000);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn prometheus_export_contains_lines() {
|
||||
let m = MetricsCollector::new();
|
||||
m.record_task(Duration::from_millis(5));
|
||||
m.set_active_threads(4);
|
||||
m.inc_counter("my_reqs", 7);
|
||||
m.set_gauge("my_temp", 1.5);
|
||||
m.observe("my_lat_ms", Duration::from_millis(12));
|
||||
|
||||
let out = m.export_prometheus();
|
||||
assert!(out.contains("# TYPE mytheclipse_task_count counter"));
|
||||
assert!(out.contains("mytheclipse_task_count 1"));
|
||||
assert!(out.contains("mytheclipse_active_threads 4"));
|
||||
assert!(out.contains("# TYPE my_reqs counter"));
|
||||
assert!(out.contains("my_reqs 7"));
|
||||
assert!(out.contains("# TYPE my_temp gauge"));
|
||||
assert!(out.contains("my_temp 1.5"));
|
||||
assert!(out.contains("my_lat_ms_count 1"));
|
||||
assert!(out.contains("my_lat_ms_avg 12000"));
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,152 @@
|
||||
//! Span and panic isolation tracking (feature `observability`).
|
||||
//!
|
||||
//! [`PanicTracker::install`] installs a global panic hook that logs every
|
||||
//! panic through [`tracing`] inside a `mytheclipse_panic_task` span — with the
|
||||
//! panic message and source location — without stopping the application.
|
||||
//! [`PanicTracker::catch`] wraps a closure so a panic is captured and reported
|
||||
//! as a [`PanicInfo`] instead of unwinding across a boundary.
|
||||
|
||||
use std::panic::{catch_unwind, AssertUnwindSafe};
|
||||
use std::sync::Arc;
|
||||
|
||||
/// Information about a captured panic.
|
||||
#[derive(Debug, Clone, PartialEq, Eq)]
|
||||
pub struct PanicInfo {
|
||||
/// Best-effort rendering of the panic payload.
|
||||
pub message: String,
|
||||
/// The source location of the panic, if available.
|
||||
pub location: Option<String>,
|
||||
}
|
||||
|
||||
#[allow(deprecated)]
|
||||
type PanicHookFn = dyn Fn(&std::panic::PanicInfo<'_>) + Send + Sync + 'static;
|
||||
|
||||
/// A RAII guard that, while held, keeps the mytheclipse panic hook installed,
|
||||
/// restoring the previously-installed hook on drop.
|
||||
///
|
||||
/// Returned by [`PanicTracker::install`].
|
||||
#[must_use = "dropping the guard restores the previous panic hook"]
|
||||
pub struct PanicGuard {
|
||||
previous: Arc<PanicHookFn>,
|
||||
}
|
||||
|
||||
/// Installs a panic hook that logs panics through [`tracing`].
|
||||
pub struct PanicTracker;
|
||||
|
||||
impl PanicTracker {
|
||||
/// Installs a global panic hook that records each panic via
|
||||
/// `tracing::error!` inside a `mytheclipse_panic_task` span, then re-invokes
|
||||
/// the previously-installed hook so default panic output (and any user
|
||||
/// hook) still runs.
|
||||
///
|
||||
/// The application does not stop; the panic still unwinds normally, but a
|
||||
/// trace is captured first. Returns a [`PanicGuard`] that restores the
|
||||
/// previous hook when dropped.
|
||||
#[allow(deprecated)]
|
||||
pub fn install() -> PanicGuard {
|
||||
let previous: Box<PanicHookFn> = std::panic::take_hook();
|
||||
let previous: Arc<PanicHookFn> = Arc::from(previous);
|
||||
let for_hook = Arc::clone(&previous);
|
||||
std::panic::set_hook(Box::new(move |info| {
|
||||
let span = tracing::error_span!("mytheclipse_panic_task");
|
||||
let message = payload_to_string(info.payload());
|
||||
let location = info
|
||||
.location()
|
||||
.map(|l| format!("{}:{}:{}", l.file(), l.line(), l.column()));
|
||||
let _guard = span.enter();
|
||||
match location {
|
||||
Some(loc) => tracing::error!("panic caught: {message} at {loc}"),
|
||||
None => tracing::error!("panic caught: {message}"),
|
||||
}
|
||||
for_hook(info);
|
||||
}));
|
||||
PanicGuard { previous }
|
||||
}
|
||||
|
||||
/// Runs `f`, catching any panic and returning it as an
|
||||
/// [`Err(PanicInfo)`](PanicInfo) rather than unwinding across the caller.
|
||||
///
|
||||
/// Useful to isolate a panic inside a worker thread so it cannot bring
|
||||
/// down the rest of the process. The caller is responsible for deciding
|
||||
/// whether to continue after a caught panic.
|
||||
pub fn catch<T>(f: impl FnOnce() -> T) -> Result<T, PanicInfo> {
|
||||
let wrapped = AssertUnwindSafe(f);
|
||||
catch_unwind(wrapped).map_err(|payload| PanicInfo {
|
||||
message: payload_to_string(&*payload),
|
||||
location: None,
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
impl Drop for PanicGuard {
|
||||
fn drop(&mut self) {
|
||||
// Take out the hook we installed so we don't call ourselves, then
|
||||
// restore the one that was active before it.
|
||||
let _current = std::panic::take_hook();
|
||||
let previous = Arc::clone(&self.previous);
|
||||
std::panic::set_hook(Box::new(move |info| previous(info)));
|
||||
}
|
||||
}
|
||||
|
||||
/// Renders a panic payload (a `&str`, `String`, or fallback) to a string.
|
||||
fn payload_to_string(payload: &(dyn std::any::Any + Send)) -> String {
|
||||
if let Some(message) = payload.downcast_ref::<&str>() {
|
||||
(*message).to_string()
|
||||
} else if let Some(message) = payload.downcast_ref::<String>() {
|
||||
message.clone()
|
||||
} else {
|
||||
"panic with a non-string payload".to_string()
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
use std::sync::atomic::{AtomicUsize, Ordering};
|
||||
|
||||
#[test]
|
||||
fn catch_returns_ok_on_success() {
|
||||
let result = PanicTracker::catch(|| 1 + 1);
|
||||
assert_eq!(result.unwrap(), 2);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn catch_captures_panic_as_err() {
|
||||
let result: Result<u32, PanicInfo> = PanicTracker::catch(|| panic!("boom"));
|
||||
assert!(result.is_err());
|
||||
assert_eq!(result.unwrap_err().message, "boom");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn catch_captures_string_payload() {
|
||||
let err = PanicTracker::catch(|| panic!("{}", String::from("stringy"))).unwrap_err();
|
||||
assert_eq!(err.message, "stringy");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn hook_is_restored_after_guard_drop() {
|
||||
// Install then drop — previous hook (default) restored.
|
||||
let _guard = PanicTracker::install();
|
||||
drop(_guard);
|
||||
let _ = PanicTracker::catch(|| panic!("irrelevant"));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn hook_runs_without_and_panics_are_still_catchable() {
|
||||
let calls = Arc::new(AtomicUsize::new(0));
|
||||
let calls_hook = Arc::clone(&calls);
|
||||
let prev = std::panic::take_hook();
|
||||
std::panic::set_hook(Box::new(move |info| {
|
||||
calls_hook.fetch_add(1, Ordering::SeqCst);
|
||||
let _ = info.location().is_some();
|
||||
}));
|
||||
let _guard = PanicTracker::install();
|
||||
// A panic in another thread triggers the hook but not the main thread.
|
||||
std::thread::spawn(|| panic!("worker boom"))
|
||||
.join()
|
||||
.unwrap_err();
|
||||
drop(_guard);
|
||||
std::panic::set_hook(prev);
|
||||
assert!(calls.load(Ordering::SeqCst) >= 1);
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,189 @@
|
||||
//! A token-bucket rate limiter (feature `traffic`).
|
||||
//!
|
||||
//! [`RateLimiter`] caps how many tokens can be consumed per unit time, so a
|
||||
//! caller/request stream is throttled (e.g. at most 100 jobs per second).
|
||||
//! Tokens are refilled lazily from elapsed time rather than by a background
|
||||
//! task, and a configurable burst capacity permits short bursts beyond the
|
||||
//! steady-state rate.
|
||||
|
||||
use std::sync::Mutex;
|
||||
use std::time::{Duration, Instant};
|
||||
|
||||
use tracing::Instrument;
|
||||
|
||||
use crate::error::MytheclipseError;
|
||||
|
||||
/// The error returned by rate-limit acquisition.
|
||||
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
|
||||
pub enum RateLimitError {
|
||||
/// The limiter's capacity was exceeded and no retry path was taken.
|
||||
///
|
||||
/// Returned by [`RateLimiter::try_acquire`] when no token is available, or
|
||||
/// wrapped into the shared [`MytheclipseError::RateLimited`] variant.
|
||||
Limited,
|
||||
}
|
||||
|
||||
impl std::fmt::Display for RateLimitError {
|
||||
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
|
||||
match self {
|
||||
Self::Limited => write!(f, "rate limit exceeded"),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl std::error::Error for RateLimitError {}
|
||||
|
||||
struct Inner {
|
||||
capacity: u64,
|
||||
refill_per_sec: f64,
|
||||
tokens: Mutex<f64>,
|
||||
last_refill: Mutex<Instant>,
|
||||
}
|
||||
|
||||
/// A thread-safe, async-compatible token-bucket rate limiter.
|
||||
///
|
||||
/// Construct with [`RateLimiter::new`], then [`RateLimiter::acquire`] to await
|
||||
/// a token or [`RateLimiter::try_acquire`] to fail fast when exhausted.
|
||||
#[derive(Clone)]
|
||||
pub struct RateLimiter {
|
||||
inner: std::sync::Arc<Inner>,
|
||||
}
|
||||
|
||||
impl RateLimiter {
|
||||
/// Builds a limiter refilling at `rate_per_sec` tokens/second with a burst
|
||||
/// capacity of `burst_capacity`.
|
||||
///
|
||||
/// # Panics
|
||||
///
|
||||
/// Panics if `rate_per_sec` is not finite and positive, or if
|
||||
/// `burst_capacity` is zero.
|
||||
pub fn new(rate_per_sec: f64, burst_capacity: u64) -> Self {
|
||||
assert!(
|
||||
rate_per_sec.is_finite() && rate_per_sec > 0.0,
|
||||
"rate must be positive"
|
||||
);
|
||||
assert!(burst_capacity > 0, "burst capacity must be > 0");
|
||||
Self {
|
||||
inner: std::sync::Arc::new(Inner {
|
||||
capacity: burst_capacity,
|
||||
refill_per_sec: rate_per_sec,
|
||||
tokens: Mutex::new(burst_capacity as f64),
|
||||
last_refill: Mutex::new(Instant::now()),
|
||||
}),
|
||||
}
|
||||
}
|
||||
|
||||
/// Awaits until a token is available and consumes one.
|
||||
///
|
||||
/// Returns [`RateLimitError::Limited`] (in practice, never for `acquire`)
|
||||
/// on closure; the suspended waiter is woken when the bucket refills. The
|
||||
/// wait is wrapped in a `mytheclipse_ratelimit_task` tracing span.
|
||||
pub async fn acquire(&self) -> Result<(), RateLimitError> {
|
||||
loop {
|
||||
if self.try_acquire().is_ok() {
|
||||
return Ok(());
|
||||
}
|
||||
let span = tracing::info_span!("mytheclipse_ratelimit_task");
|
||||
// Sleep for the time it takes to refill a single token.
|
||||
let sleep = Duration::from_secs_f64(1.0 / self.inner.refill_per_sec);
|
||||
tokio::time::sleep(sleep).instrument(span).await;
|
||||
}
|
||||
}
|
||||
|
||||
/// Consumes one token immediately if one is available.
|
||||
///
|
||||
/// Returns `Err(RateLimitError::Limited)` (or the caller may map it to
|
||||
/// [`MytheclipseError::RateLimited`]) when the bucket is empty.
|
||||
pub fn try_acquire(&self) -> Result<(), RateLimitError> {
|
||||
let mut tokens = self.inner.tokens.lock().unwrap();
|
||||
let mut last = self.inner.last_refill.lock().unwrap();
|
||||
self.refill(&mut tokens, *last);
|
||||
*last = Instant::now();
|
||||
|
||||
if *tokens >= 1.0 {
|
||||
*tokens -= 1.0;
|
||||
Ok(())
|
||||
} else {
|
||||
Err(RateLimitError::Limited)
|
||||
}
|
||||
}
|
||||
|
||||
/// The approximate number of tokens currently available (including burst
|
||||
/// headroom), for metrics/observability purposes.
|
||||
pub fn available_tokens(&self) -> u64 {
|
||||
let mut tokens = self.inner.tokens.lock().unwrap();
|
||||
let last = self.inner.last_refill.lock().unwrap();
|
||||
self.refill(&mut tokens, *last);
|
||||
tokens.floor() as u64
|
||||
}
|
||||
|
||||
fn refill(&self, tokens: &mut f64, last: Instant) {
|
||||
let elapsed_secs = last.elapsed().as_secs_f64();
|
||||
let added = elapsed_secs * self.inner.refill_per_sec;
|
||||
*tokens = (*tokens + added).min(self.inner.capacity as f64);
|
||||
}
|
||||
|
||||
/// `acquire` mapped onto the shared error type, e.g. for code returning
|
||||
/// [`MytheclipseError`].
|
||||
pub async fn acquire_err(&self) -> Result<(), MytheclipseError> {
|
||||
self.acquire()
|
||||
.await
|
||||
.map_err(|_| MytheclipseError::RateLimited)
|
||||
}
|
||||
|
||||
/// `try_acquire` mapped onto the shared error type.
|
||||
pub fn try_acquire_err(&self) -> Result<(), MytheclipseError> {
|
||||
self.try_acquire()
|
||||
.map_err(|_| MytheclipseError::RateLimited)
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
#[test]
|
||||
fn burst_capacity_consumed_immediately() {
|
||||
let limiter = RateLimiter::new(1000.0, 3);
|
||||
assert!(limiter.try_acquire().is_ok());
|
||||
assert!(limiter.try_acquire().is_ok());
|
||||
assert!(limiter.try_acquire().is_ok());
|
||||
assert_eq!(limiter.try_acquire(), Err(RateLimitError::Limited));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn available_tokens_bounded_by_capacity() {
|
||||
let limiter = RateLimiter::new(1000.0, 5);
|
||||
assert_eq!(limiter.available_tokens(), 5);
|
||||
let _ = limiter.try_acquire();
|
||||
assert_eq!(limiter.available_tokens(), 4);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn acquire_waits_for_refill() {
|
||||
let limiter = RateLimiter::new(1000.0, 1);
|
||||
assert!(limiter.try_acquire().is_ok());
|
||||
let start = Instant::now();
|
||||
limiter
|
||||
.acquire()
|
||||
.await
|
||||
.expect("acquire should eventually succeed");
|
||||
assert!(start.elapsed() >= Duration::from_millis(1));
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn try_acquire_err_maps_to_shared_error() {
|
||||
let limiter = RateLimiter::new(1000.0, 1);
|
||||
let _ = limiter.try_acquire();
|
||||
assert_eq!(
|
||||
limiter.try_acquire_err(),
|
||||
Err(MytheclipseError::RateLimited)
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
#[should_panic]
|
||||
fn zero_burst_panics() {
|
||||
let _ = RateLimiter::new(1.0, 0);
|
||||
}
|
||||
}
|
||||
+256
@@ -0,0 +1,256 @@
|
||||
//! Automatic retry with exponential backoff and jitter (feature `resiliency`).
|
||||
//!
|
||||
//! [`retry`] re-runs a fallible async operation according to a
|
||||
//! [`RetryConfig`], sleeping an exponentially-growing, optionally-jittered
|
||||
//! delay between attempts. A predicate decides which errors are retryable, so
|
||||
//! permanent failures (e.g. a 4xx response) short-circuit immediately while
|
||||
//! transient ones (network hiccups, connection refused) are retried.
|
||||
|
||||
use std::future::Future;
|
||||
use std::time::Duration;
|
||||
|
||||
use rand::Rng;
|
||||
use tracing::Instrument;
|
||||
|
||||
/// How much random jitter to apply to each backoff delay.
|
||||
///
|
||||
/// Jitter prevents the "thundering herd" of many retrying clients waking
|
||||
/// simultaneously; [`Full`](JitterKind::Full) is the most aggressive and is
|
||||
/// the recommended default for distributed systems.
|
||||
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
|
||||
pub enum JitterKind {
|
||||
/// No randomness: sleep exactly the computed backoff.
|
||||
None,
|
||||
/// Sleep in `[delay, 2 * delay)` — sometimes called "equal jitter".
|
||||
Equal,
|
||||
/// Sleep in `[0, delay)` — "full jitter".
|
||||
Full,
|
||||
}
|
||||
|
||||
/// Configuration driving [`retry`].
|
||||
#[derive(Debug, Clone)]
|
||||
pub struct RetryConfig {
|
||||
/// Total number of attempts (including the first). Must be `>= 1`.
|
||||
pub max_attempts: u32,
|
||||
/// Initial delay before the first retry.
|
||||
pub base_delay: Duration,
|
||||
/// Upper bound on the computed backoff delay.
|
||||
pub max_delay: Duration,
|
||||
/// Exponential growth factor applied after each failure.
|
||||
pub factor: f64,
|
||||
/// Jitter strategy applied to each delay.
|
||||
pub jitter: JitterKind,
|
||||
}
|
||||
|
||||
impl Default for RetryConfig {
|
||||
fn default() -> Self {
|
||||
Self {
|
||||
max_attempts: 3,
|
||||
base_delay: Duration::from_millis(500),
|
||||
max_delay: Duration::from_secs(10),
|
||||
factor: 2.0,
|
||||
jitter: JitterKind::Full,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// The error returned by [`retry`] once it gives up.
|
||||
#[derive(Debug)]
|
||||
pub enum RetryError<E> {
|
||||
/// All attempts were made and the last one failed.
|
||||
///
|
||||
/// `attempts` is the total number of attempts performed and `last` is the
|
||||
/// error produced by the final attempt.
|
||||
Exhausted { attempts: u32, last: E },
|
||||
}
|
||||
|
||||
impl<E> RetryError<E> {
|
||||
/// Returns a reference to the error produced by the final attempt.
|
||||
pub fn last(&self) -> &E {
|
||||
match self {
|
||||
Self::Exhausted { last, .. } => last,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl<E: std::fmt::Display> std::fmt::Display for RetryError<E> {
|
||||
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
|
||||
match self {
|
||||
Self::Exhausted { attempts, last } => {
|
||||
write!(f, "retry exhausted after {attempts} attempts: {last}")
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl<E: std::fmt::Debug + std::fmt::Display> std::error::Error for RetryError<E> {}
|
||||
|
||||
/// Retries `op` according to `config`, retrying only errors for which
|
||||
/// `filter` returns `true`.
|
||||
///
|
||||
/// On the first attempt and after each retryable failure, `op` is invoked
|
||||
/// again (rebuilt from the captured closure) until it succeeds, the retryable
|
||||
/// error becomes permanent, or `max_attempts` is reached. Between attempts the
|
||||
/// coroutine sleeps for a backoff delay computed as
|
||||
/// `min(max_delay, base_delay * factor^attempt)` with jitter applied per
|
||||
/// [`RetryConfig::jitter`].
|
||||
///
|
||||
/// Each attempt runs inside a `mytheclipse_retry_task` tracing span carrying
|
||||
/// the attempt index and total.
|
||||
pub async fn retry<T, E, F, Fut, P>(
|
||||
config: RetryConfig,
|
||||
filter: P,
|
||||
mut op: F,
|
||||
) -> Result<T, RetryError<E>>
|
||||
where
|
||||
F: FnMut() -> Fut,
|
||||
Fut: Future<Output = Result<T, E>>,
|
||||
P: Fn(&E) -> bool,
|
||||
{
|
||||
let mut attempt: u32 = 0;
|
||||
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) => return Ok(value),
|
||||
Err(err) => {
|
||||
let retryable = filter(&err);
|
||||
if !retryable || attempt >= config.max_attempts {
|
||||
return Err(RetryError::Exhausted {
|
||||
attempts: attempt,
|
||||
last: err,
|
||||
});
|
||||
}
|
||||
let delay = backoff_delay(&config, attempt, rand::thread_rng());
|
||||
tokio::time::sleep(delay).await;
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Computes the (jittered) delay to sleep before retry `attempt` (1-based).
|
||||
///
|
||||
/// Kept as a pure function for testability.
|
||||
fn backoff_delay<R: Rng>(config: &RetryConfig, attempt: u32, mut rng: R) -> Duration {
|
||||
let exponent = attempt.saturating_sub(1) as f64; // first retry uses base
|
||||
let computed = config.base_delay.as_millis() as f64 * config.factor.powf(exponent);
|
||||
let max_ms = config.max_delay.as_millis() as f64;
|
||||
let capped = computed.min(max_ms);
|
||||
|
||||
let millis = match config.jitter {
|
||||
JitterKind::None => capped,
|
||||
JitterKind::Equal => capped / 2.0 + rng.gen_range(0.0..capped / 2.0),
|
||||
JitterKind::Full => rng.gen_range(0.0..capped),
|
||||
};
|
||||
|
||||
Duration::from_millis(millis.clamp(0.0, max_ms) as u64)
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
use std::cell::Cell;
|
||||
|
||||
fn always<E>(_: &E) -> bool {
|
||||
true
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn succeeds_on_first_attempt() {
|
||||
let calls = Cell::new(0u32);
|
||||
let result = retry(RetryConfig::default(), always, || async {
|
||||
calls.set(calls.get() + 1);
|
||||
Ok::<_, ()>(42u32)
|
||||
})
|
||||
.await;
|
||||
assert_eq!(result.unwrap(), 42);
|
||||
assert_eq!(calls.get(), 1);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn succeeds_after_transient_failures() {
|
||||
let config = RetryConfig {
|
||||
max_attempts: 5,
|
||||
base_delay: Duration::from_millis(1),
|
||||
..RetryConfig::default()
|
||||
};
|
||||
let calls = Cell::new(0u32);
|
||||
let result = retry(config, always, || async {
|
||||
calls.set(calls.get() + 1);
|
||||
if calls.get() < 3 {
|
||||
Err::<u32, u8>(9)
|
||||
} else {
|
||||
Ok(7u32)
|
||||
}
|
||||
})
|
||||
.await;
|
||||
assert_eq!(result.unwrap(), 7);
|
||||
assert_eq!(calls.get(), 3);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn exhausts_after_max_attempts() {
|
||||
let config = RetryConfig {
|
||||
max_attempts: 3,
|
||||
base_delay: Duration::from_millis(1),
|
||||
..RetryConfig::default()
|
||||
};
|
||||
let calls = Cell::new(0u32);
|
||||
let result = retry(config, always, || async {
|
||||
calls.set(calls.get() + 1);
|
||||
Err::<u32, u8>(42)
|
||||
})
|
||||
.await;
|
||||
assert!(matches!(
|
||||
result,
|
||||
Err(RetryError::Exhausted { attempts: 3, .. })
|
||||
));
|
||||
assert_eq!(result.unwrap_err().last(), &42);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn non_retryable_error_short_circuits() {
|
||||
let config = RetryConfig {
|
||||
max_attempts: 10,
|
||||
base_delay: Duration::from_millis(1),
|
||||
..RetryConfig::default()
|
||||
};
|
||||
let calls = Cell::new(0u32);
|
||||
let result = retry(
|
||||
config,
|
||||
|e: &u16| *e != 403,
|
||||
|| async {
|
||||
calls.set(calls.get() + 1);
|
||||
Err::<u32, u16>(403)
|
||||
},
|
||||
)
|
||||
.await;
|
||||
assert!(matches!(
|
||||
result,
|
||||
Err(RetryError::Exhausted { attempts: 1, .. })
|
||||
));
|
||||
assert_eq!(calls.get(), 1);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn full_jitter_is_within_bounds_and_capped() {
|
||||
let config = RetryConfig {
|
||||
max_attempts: 3,
|
||||
base_delay: Duration::from_secs(2),
|
||||
max_delay: Duration::from_secs(4),
|
||||
factor: 10.0,
|
||||
jitter: JitterKind::Full,
|
||||
};
|
||||
let mut rng = rand::thread_rng();
|
||||
for _ in 0..1000 {
|
||||
let d = backoff_delay(&config, 2, &mut rng);
|
||||
assert!(d <= config.max_delay);
|
||||
}
|
||||
}
|
||||
}
|
||||
+214
@@ -0,0 +1,214 @@
|
||||
//! Graceful shutdown coordination (feature `lifecycle`).
|
||||
//!
|
||||
//! [`ShutdownManager`] watches for OS signals (SIGINT/SIGTERM on Unix, Ctrl-C
|
||||
//! everywhere) and broadcasts a shutdown signal to every registered task, then
|
||||
//! waits for those tasks to finish before the process exits. Tasks subscribe
|
||||
//! via a cloneable [`ShutdownSignal`] and cooperatively stop when it fires.
|
||||
|
||||
use std::sync::Arc;
|
||||
use std::time::Duration;
|
||||
|
||||
use tokio::sync::watch;
|
||||
use tracing::Instrument;
|
||||
|
||||
/// A cloneable token a task holds to observe whether shutdown has been
|
||||
/// requested.
|
||||
#[derive(Clone)]
|
||||
pub struct ShutdownSignal {
|
||||
rx: watch::Receiver<bool>,
|
||||
}
|
||||
|
||||
impl ShutdownSignal {
|
||||
/// Whether shutdown has been requested.
|
||||
pub fn is_shutdown(&self) -> bool {
|
||||
*self.rx.borrow()
|
||||
}
|
||||
|
||||
/// Awaits until shutdown is requested, then returns.
|
||||
pub async fn wait(&mut self) {
|
||||
if *self.rx.borrow() {
|
||||
return;
|
||||
}
|
||||
// Borrow avoids holding the receiver while yielding.
|
||||
let _ = self.rx.changed().await;
|
||||
}
|
||||
}
|
||||
|
||||
/// Coordinates graceful shutdown of tracked background tasks.
|
||||
///
|
||||
/// Construct with [`ShutdownManager::new`], hand each long-running task a
|
||||
/// [`ShutdownSignal`] via [`ShutdownManager::handle`], register its join
|
||||
/// handle with [`ShutdownManager::register`], and finally defer the process
|
||||
/// exit until [`ShutdownManager::drain`] completes.
|
||||
#[derive(Clone)]
|
||||
pub struct ShutdownManager {
|
||||
inner: Arc<Inner>,
|
||||
}
|
||||
|
||||
struct Inner {
|
||||
tx: watch::Sender<bool>,
|
||||
/// Kept alive so `tx.send` always has a receiver and therefore always
|
||||
/// updates the stored value — even if `request()` fires before any
|
||||
/// [`ShutdownSignal`] has been handed out.
|
||||
_keepalive: watch::Receiver<bool>,
|
||||
tasks: std::sync::Mutex<Vec<tokio::task::JoinHandle<()>>>,
|
||||
}
|
||||
|
||||
impl Default for ShutdownManager {
|
||||
fn default() -> Self {
|
||||
Self::new()
|
||||
}
|
||||
}
|
||||
|
||||
impl ShutdownManager {
|
||||
/// Builds a new shutdown manager with no tasks tracked.
|
||||
pub fn new() -> Self {
|
||||
let (tx, keepalive) = watch::channel(false);
|
||||
Self {
|
||||
inner: Arc::new(Inner {
|
||||
tx,
|
||||
_keepalive: keepalive,
|
||||
tasks: std::sync::Mutex::new(Vec::new()),
|
||||
}),
|
||||
}
|
||||
}
|
||||
|
||||
/// Returns a new [`ShutdownSignal`] this manager will fire on shutdown.
|
||||
pub fn handle(&self) -> ShutdownSignal {
|
||||
ShutdownSignal {
|
||||
rx: self.inner.tx.subscribe(),
|
||||
}
|
||||
}
|
||||
|
||||
/// Registers a background task so shutdown waits for it to complete.
|
||||
///
|
||||
/// The task is expected to observe its [`ShutdownSignal`] and stop
|
||||
/// promptly once it fires; `drain` gives tasks a grace window.
|
||||
pub fn register(&self, handle: tokio::task::JoinHandle<()>) {
|
||||
self.inner.tasks.lock().unwrap().push(handle);
|
||||
}
|
||||
|
||||
/// Whether shutdown has been requested.
|
||||
pub fn is_shutdown(&self) -> bool {
|
||||
*self.inner.tx.borrow()
|
||||
}
|
||||
|
||||
/// Awaits until shutdown is requested (by an OS signal or an explicit
|
||||
/// [`ShutdownManager::request`]).
|
||||
pub async fn wait_for_shutdown(&self) {
|
||||
let mut rx = self.inner.tx.subscribe();
|
||||
if *rx.borrow() {
|
||||
return;
|
||||
}
|
||||
let _ = rx.changed().await;
|
||||
}
|
||||
|
||||
/// Requests shutdown programmatically (also invoked by the signal
|
||||
/// handler). Safe to call more than once.
|
||||
pub fn request(&self) {
|
||||
let _ = self.inner.tx.send(true);
|
||||
}
|
||||
|
||||
/// Waits for shutdown to be requested, then awaits all registered tasks,
|
||||
/// allowing at most `grace` per task before giving up.
|
||||
///
|
||||
/// Runs in a `mytheclipse_shutdown_task` tracing span.
|
||||
pub async fn drain(&self, grace: Duration) {
|
||||
let span = tracing::info_span!("mytheclipse_shutdown_task");
|
||||
self.wait_for_shutdown().instrument(span.clone()).await;
|
||||
|
||||
let tasks = {
|
||||
let mut guard = self.inner.tasks.lock().unwrap();
|
||||
std::mem::take(&mut *guard)
|
||||
};
|
||||
for task in tasks {
|
||||
let _ = tokio::time::timeout(grace, task)
|
||||
.instrument(span.clone())
|
||||
.await;
|
||||
}
|
||||
}
|
||||
|
||||
/// Waits for the next OS termination signal (SIGINT/SIGTERM on Unix,
|
||||
/// Ctrl-C elsewhere).
|
||||
pub async fn wait_for_os_signal(&self) {
|
||||
let _ = os_signal().await;
|
||||
}
|
||||
}
|
||||
|
||||
/// Awaits SIGINT/SIGTERM on Unix, or Ctrl-C on other platforms.
|
||||
#[cfg(unix)]
|
||||
async fn os_signal() {
|
||||
use tokio::signal::unix::{signal, SignalKind};
|
||||
|
||||
let mut sigint = signal(SignalKind::interrupt()).expect("failed to install SIGINT handler");
|
||||
let mut sigterm = signal(SignalKind::terminate()).expect("failed to install SIGTERM handler");
|
||||
|
||||
tokio::select! {
|
||||
_ = sigint.recv() => {},
|
||||
_ = sigterm.recv() => {},
|
||||
}
|
||||
}
|
||||
|
||||
/// Awaits Ctrl-C on non-Unix platforms.
|
||||
#[cfg(not(unix))]
|
||||
async fn os_signal() {
|
||||
use tokio::signal::ctrl_c;
|
||||
let _ = ctrl_c().await;
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
#[tokio::test]
|
||||
async fn signal_fires_on_request() {
|
||||
let manager = ShutdownManager::new();
|
||||
let mut sig = manager.handle();
|
||||
assert!(!sig.is_shutdown());
|
||||
assert!(!manager.is_shutdown());
|
||||
|
||||
let observer = tokio::spawn(async move {
|
||||
sig.wait().await;
|
||||
sig.is_shutdown()
|
||||
});
|
||||
|
||||
manager.request();
|
||||
assert!(manager.is_shutdown());
|
||||
assert!(observer.await.unwrap());
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn request_is_idempotent() {
|
||||
let manager = ShutdownManager::new();
|
||||
manager.request();
|
||||
manager.request();
|
||||
let sig = manager.handle();
|
||||
assert!(sig.is_shutdown());
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn drain_waits_for_registered_tasks() {
|
||||
let manager = ShutdownManager::new();
|
||||
let sig = manager.handle();
|
||||
let handle = tokio::spawn(async move {
|
||||
let mut sig = sig;
|
||||
sig.wait().await;
|
||||
});
|
||||
manager.register(handle);
|
||||
manager.request();
|
||||
// drain completes promptly because the task stops on the signal.
|
||||
manager.drain(Duration::from_secs(5)).await;
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn drain_times_out_a_slow_task() {
|
||||
let manager = ShutdownManager::new();
|
||||
let _sig = manager.handle();
|
||||
// A task that never observes shutdown — it must be timed out.
|
||||
let slow = tokio::spawn(std::future::pending::<()>());
|
||||
manager.register(slow);
|
||||
manager.request();
|
||||
// drain should return (time out) rather than hang forever.
|
||||
manager.drain(Duration::from_millis(50)).await;
|
||||
}
|
||||
}
|
||||
+155
@@ -0,0 +1,155 @@
|
||||
//! Hard execution time bounds for async work (feature `resiliency`).
|
||||
//!
|
||||
//! Provides [`with_timeout`] (a convenience that resolves a future to a
|
||||
//! [`Result`] with an elapsed-vs-completed outcome) and [`Timeout`], a
|
||||
//! stand-alone [`Future`] wrapper you can build once and hand to
|
||||
//! [`crate::io::spawn_io`] / [`crate::bg::spawn_bg`]/`tokio::spawn` so the
|
||||
//! bound is enforced wherever the future actually runs.
|
||||
|
||||
use std::future::Future;
|
||||
use std::pin::Pin;
|
||||
use std::task::{Context, Poll};
|
||||
use std::time::Duration;
|
||||
|
||||
use tracing::Instrument;
|
||||
|
||||
use crate::error::MytheclipseError;
|
||||
|
||||
/// The outcome of timing out a single future.
|
||||
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
|
||||
pub enum TimeoutError {
|
||||
/// The deadline elapsed before the future completed.
|
||||
Elapsed,
|
||||
}
|
||||
|
||||
impl std::fmt::Display for TimeoutError {
|
||||
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
|
||||
match self {
|
||||
Self::Elapsed => write!(f, "deadline elapsed"),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl std::error::Error for TimeoutError {}
|
||||
|
||||
/// Runs `future` to completion, giving up after `dur` and returning
|
||||
/// [`TimeoutError::Elapsed`] if the bound is exceeded.
|
||||
///
|
||||
/// The future is cancelled on timeout: its task is aborted when the deadline
|
||||
/// fires, so no lingering resource is pinned. The call is wrapped in a
|
||||
/// `mytheclipse_timeout_task` tracing span.
|
||||
///
|
||||
/// # Panics
|
||||
///
|
||||
/// Panics if called outside the context of a running Tokio runtime.
|
||||
pub async fn with_timeout<T, F>(dur: Duration, future: F) -> Result<T, TimeoutError>
|
||||
where
|
||||
F: Future<Output = T>,
|
||||
{
|
||||
let span = tracing::info_span!("mytheclipse_timeout_task");
|
||||
match tokio::time::timeout(dur, future.instrument(span)).await {
|
||||
Ok(value) => Ok(value),
|
||||
Err(_) => Err(TimeoutError::Elapsed),
|
||||
}
|
||||
}
|
||||
|
||||
/// A future that enforces a [`Duration`] deadline on an inner future.
|
||||
///
|
||||
/// Unlike [`with_timeout`], this returns an invocable [`Future`] rather than
|
||||
/// polling to completion, so it can be constructed ahead of time and spawned
|
||||
/// through any executor:
|
||||
///
|
||||
/// ```no_run
|
||||
/// use mytheclipse::timeout::Timeout;
|
||||
/// # fn _r() {
|
||||
/// let bounded: Timeout<_> = Timeout::new(
|
||||
/// std::time::Duration::from_secs(1),
|
||||
/// async { "done" },
|
||||
/// );
|
||||
/// # }
|
||||
/// ```
|
||||
///
|
||||
/// If the deadline elapses first, the completed output is `Err(TimeoutError)`.
|
||||
pub struct Timeout<T> {
|
||||
inner: Pin<Box<dyn Future<Output = Result<T, TimeoutError>> + Send>>,
|
||||
}
|
||||
|
||||
impl<T: Send + 'static> Timeout<T> {
|
||||
/// Wraps `future` with a deadline of `dur`.
|
||||
pub fn new<F>(dur: Duration, future: F) -> Self
|
||||
where
|
||||
F: Future<Output = T> + Send + 'static,
|
||||
{
|
||||
let future = async move {
|
||||
match tokio::time::timeout(dur, future).await {
|
||||
Ok(value) => Ok(value),
|
||||
Err(_) => Err(TimeoutError::Elapsed),
|
||||
}
|
||||
};
|
||||
Self {
|
||||
inner: Box::pin(future),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl<T> Future for Timeout<T> {
|
||||
type Output = Result<T, TimeoutError>;
|
||||
|
||||
fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
|
||||
// `inner` is a `Pin<Box<...>>`. Because `Box` is `Unpin`, calling
|
||||
// `as_mut` on the boxed pin yields `Pin<&mut (dyn Future + Send)>`
|
||||
// which polls the inner future. Pinning is upheld by the box, so no
|
||||
// unsafe projection is needed.
|
||||
self.get_mut().inner.as_mut().poll(cx)
|
||||
}
|
||||
}
|
||||
|
||||
/// A hard deadline for a single future, returning `Result<T, MytheclipseError>`.
|
||||
///
|
||||
/// Convenience twin of [`with_timeout`] that maps the outcome onto the crate's
|
||||
/// shared error type, e.g. for use in code that already returns
|
||||
/// [`MytheclipseError`].
|
||||
pub async fn timeout<T, F>(dur: Duration, future: F) -> Result<T, MytheclipseError>
|
||||
where
|
||||
F: Future<Output = T>,
|
||||
{
|
||||
with_timeout(dur, future)
|
||||
.await
|
||||
.map_err(|_| MytheclipseError::Timeout)
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
#[tokio::test]
|
||||
async fn completes_within_bound_returns_value() {
|
||||
let value = with_timeout(Duration::from_secs(5), async { 42u32 }).await;
|
||||
assert_eq!(value.unwrap(), 42);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn exceeding_bound_yields_elapsed() {
|
||||
let outcome = with_timeout(Duration::from_millis(20), async {
|
||||
tokio::time::sleep(Duration::from_secs(5)).await;
|
||||
42u32
|
||||
})
|
||||
.await;
|
||||
assert_eq!(outcome, Err(TimeoutError::Elapsed));
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn timeout_wrapper_maps_to_shared_error() {
|
||||
let outcome = timeout(Duration::from_millis(10), async {
|
||||
tokio::time::sleep(Duration::from_secs(5)).await;
|
||||
})
|
||||
.await;
|
||||
assert_eq!(outcome, Err(MytheclipseError::Timeout));
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn timeout_future_is_spawnable() {
|
||||
let bounded = Timeout::new(Duration::from_secs(5), async { 7u32 });
|
||||
assert_eq!(tokio::spawn(bounded).await.unwrap().unwrap(), 7);
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user