From 8dec4130057420320500d7a9265ccb3cbfd0e030 Mon Sep 17 00:00:00 2001 From: asepharyana Date: Fri, 28 Aug 2026 18:44:54 +0700 Subject: [PATCH] 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. --- .github/workflows/ci.yml | 8 + Cargo.lock | 71 ++++++ Cargo.toml | 11 +- README.md | 124 ++++++++-- examples/main.rs | 154 ++++++++++++- src/backpressure.rs | 310 +++++++++++++++++++++++++ src/circuit_breaker.rs | 365 ++++++++++++++++++++++++++++++ src/concurrency.rs | 180 +++++++++++++++ src/cron.rs | 475 +++++++++++++++++++++++++++++++++++++++ src/error.rs | 44 +++- src/lib.rs | 80 ++++++- src/metrics.rs | 337 +++++++++++++++++++++++++++ src/panic_tracker.rs | 152 +++++++++++++ src/ratelimit.rs | 189 ++++++++++++++++ src/retry.rs | 256 +++++++++++++++++++++ src/shutdown.rs | 214 ++++++++++++++++++ src/timeout.rs | 155 +++++++++++++ 17 files changed, 3087 insertions(+), 38 deletions(-) create mode 100644 src/backpressure.rs create mode 100644 src/circuit_breaker.rs create mode 100644 src/concurrency.rs create mode 100644 src/cron.rs create mode 100644 src/metrics.rs create mode 100644 src/panic_tracker.rs create mode 100644 src/ratelimit.rs create mode 100644 src/retry.rs create mode 100644 src/shutdown.rs create mode 100644 src/timeout.rs diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 6d144f0..c95d23a 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -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 diff --git a/Cargo.lock b/Cargo.lock index 8d77291..6cf068d 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -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", +] diff --git a/Cargo.toml b/Cargo.toml index f56825e..c4b5913 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -8,14 +8,15 @@ repository = "https://github.com/asepharyana/mytheclipse" homepage = "https://github.com/asepharyana/mytheclipse" documentation = "https://docs.rs/mytheclipse" authors = ["asepharyana "] -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" diff --git a/README.md b/README.md index 6f56555..912a0da 100644 --- a/README.md +++ b/README.md @@ -4,7 +4,7 @@ [![Documentation](https://docs.rs/mytheclipse/badge.svg)](https://docs.rs/mytheclipse) [![License](https://img.shields.io/badge/license-MIT%20OR%20Apache--2.0-blue.svg)](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::())?; - // 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 ) - MIT license ([LICENSE-MIT](LICENSE-MIT) or ) -at your option. \ No newline at end of file +at your option. diff --git a/examples/main.rs b/examples/main.rs index ed4668e..c118a2b 100644 --- a/examples/main.rs +++ b/examples/main.rs @@ -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::()); let panic_result: Result = 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!"); } diff --git a/src/backpressure.rs b/src/backpressure.rs new file mode 100644 index 0000000..5e8fcd3 --- /dev/null +++ b/src/backpressure.rs @@ -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 { + /// 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 std::fmt::Display for BackpressureError { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + write!(f, "backpressure queue is full") + } +} + +impl std::error::Error for BackpressureError {} + +struct Inner { + queue: Mutex>, + 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 { + inner: Arc>, +} + +impl BackpressureQueue { + /// 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> { + 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> { + 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 { + 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 { + 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 = 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 = 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 = 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 = 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 = 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()); + } +} diff --git a/src/circuit_breaker.rs b/src/circuit_breaker.rs new file mode 100644 index 0000000..9ba1956 --- /dev/null +++ b/src/circuit_breaker.rs @@ -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 { + /// The circuit is open and the call was refused without executing. + Open, + /// The guarded operation failed (and the failure was recorded). + Inner(E), +} + +impl std::fmt::Display for CircuitError { + 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 std::error::Error for CircuitError {} + +struct Inner { + state: AtomicU8, + failures: AtomicU64, + successes: AtomicU64, + half_open_in_flight: AtomicUsize, + opened_at: Mutex>, + 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, +} + +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(&self, f: F) -> Result> + where + F: FnOnce() -> Result, + { + 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> = 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> = b.call(|| Err(1u8)); + } + let refused: Result<(), CircuitError> = b.call(|| Ok(())); + assert!(matches!(refused, Err(CircuitError::Open))); + } + + #[test] + fn success_resets_failure_count_in_closed() { + let b = breaker(); + let _: Result<(), CircuitError> = b.call(|| Err(1u8)); + let _: Result<(), CircuitError> = b.call(|| Err(1u8)); + assert_eq!(b.state(), CircuitState::Closed); + let r: Result<(), CircuitError> = b.call(|| Ok(())); + assert!(r.is_ok()); + // Failure count reset — two more failures should not trip (needs 3). + let _: Result<(), CircuitError> = b.call(|| Err(1u8)); + let _: Result<(), CircuitError> = 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> = 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> = 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> = 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> = b.call(|| Ok(5u32)); + assert_eq!(ok.unwrap(), 5); + let err: Result> = b.call(|| Err(9u8)); + assert!(matches!(err, Err(CircuitError::Inner(9)))); + } +} diff --git a/src/concurrency.rs b/src/concurrency.rs new file mode 100644 index 0000000..b5a69d2 --- /dev/null +++ b/src/concurrency.rs @@ -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, + available: Condvar, + max: usize, +} + +/// A thread-safe cap on concurrent in-flight sections. +#[derive(Clone)] +pub struct ConcurrencyLimiter { + inner: Arc, +} + +/// 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, +} + +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 { + 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); + } +} diff --git a/src/cron.rs b/src/cron.rs new file mode 100644 index 0000000..e5617cf --- /dev/null +++ b/src/cron.rs @@ -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, + hours: Vec, + days_of_month: Vec, + months: Vec, + days_of_week: Vec, +} + +/// 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 { + 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 { + // 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 { + 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(expr: &str, mut job: F) -> Result +where + F: FnMut() -> Fut + Send + 'static, + Fut: std::future::Future + 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, 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 { + s.parse::() + .map_err(|_| CronParseError(format!("invalid {what} `{s}`"))) +} + +fn parse_value(s: &str, min: u8, max: u8, allow_names: bool) -> Result { + 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::>()); + 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)); + } +} diff --git a/src/error.rs b/src/error.rs index e4015e6..6b4b9d2 100644 --- a/src/error.rs +++ b/src/error.rs @@ -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}"), } } } diff --git a/src/lib.rs b/src/lib.rs index 13eef40..47d3256 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -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 diff --git a/src/metrics.rs b/src/metrics.rs new file mode 100644 index 0000000..a985b64 --- /dev/null +++ b/src/metrics.rs @@ -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>, + gauges: Mutex>, + histograms: Mutex>, +} + +/// 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, +} + +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 { + 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, + /// Named counters. + pub counters: HashMap, + /// Named gauges. + pub gauges: HashMap, + /// Named histogram aggregates. + pub histograms: HashMap, +} + +/// 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, +} + +#[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")); + } +} diff --git a/src/panic_tracker.rs b/src/panic_tracker.rs new file mode 100644 index 0000000..96ff7b4 --- /dev/null +++ b/src/panic_tracker.rs @@ -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, +} + +#[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, +} + +/// 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 = std::panic::take_hook(); + let previous: Arc = 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(f: impl FnOnce() -> T) -> Result { + 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::() { + 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 = 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); + } +} diff --git a/src/ratelimit.rs b/src/ratelimit.rs new file mode 100644 index 0000000..eed6d15 --- /dev/null +++ b/src/ratelimit.rs @@ -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, + last_refill: Mutex, +} + +/// 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, +} + +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); + } +} diff --git a/src/retry.rs b/src/retry.rs new file mode 100644 index 0000000..844dfec --- /dev/null +++ b/src/retry.rs @@ -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 { + /// 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 RetryError { + /// Returns a reference to the error produced by the final attempt. + pub fn last(&self) -> &E { + match self { + Self::Exhausted { last, .. } => last, + } + } +} + +impl std::fmt::Display for RetryError { + 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 std::error::Error for RetryError {} + +/// 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( + config: RetryConfig, + filter: P, + mut op: F, +) -> Result> +where + F: FnMut() -> Fut, + Fut: Future>, + 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(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) -> 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::(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::(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::(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); + } + } +} diff --git a/src/shutdown.rs b/src/shutdown.rs new file mode 100644 index 0000000..2e8143b --- /dev/null +++ b/src/shutdown.rs @@ -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, +} + +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, +} + +struct Inner { + tx: watch::Sender, + /// 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, + tasks: std::sync::Mutex>>, +} + +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; + } +} diff --git a/src/timeout.rs b/src/timeout.rs new file mode 100644 index 0000000..c7d636f --- /dev/null +++ b/src/timeout.rs @@ -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(dur: Duration, future: F) -> Result +where + F: Future, +{ + 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 { + inner: Pin> + Send>>, +} + +impl Timeout { + /// Wraps `future` with a deadline of `dur`. + pub fn new(dur: Duration, future: F) -> Self + where + F: Future + 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 Future for Timeout { + type Output = Result; + + fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll { + // `inner` is a `Pin>`. 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`. +/// +/// 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(dur: Duration, future: F) -> Result +where + F: Future, +{ + 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); + } +}