chore(release): 1.15.0 [skip ci]

# [1.15.0](https://github.com/asepharyana/mytheclipse/compare/v1.14.0...v1.15.0) (2026-08-29)

### Features

* round-12 abstractions — AutoReconnectPool, Reconnectable ([ddb2c2f](https://github.com/asepharyana/mytheclipse/commit/ddb2c2fd2d43e07b0d26b2901746ffcc3fe8b284))

.hermes/plans/mytheclipse-round10-spec.md

@@ -0,0 +1,29 @@
+# Implementation Spec: Round 10 — COMPLETE
+
+## Goal
+Auto-integration + ergonomics: rate-limit workers, auto-metrics on service calls —
+reduce manual wiring/boilerplate.
+
+## New Features
+
+### 1. AutoMetricsServiceBuilder (mytheclipse-core, observability)
+File: `crates/mytheclipse/src/auto_metrics_service.rs`
+- Composes ServiceBuilder + MetricsCollector (+ MetricsBridge when resiliency)
+- `.run()` auto-records: calls_total counter (labelled by outcome ok/err/timeout/
+  circuit_open/rate_limited) + duration histogram; emits bridge when attached
+- Chainable .with_collector/.with_bridge/.with_builders
+- 1 test
+
+### 2. RateLimitedWorkerPool (mytheclipse-queue, in-memory)
+File: `crates/mytheclipse-queue/src/worker_rate_limited.rs`
+- Wraps WorkerPool with RateLimitedQueue — token-bucket back-pressured dequeue,
+  prevents workers hammering upstream beyond rate limit
+- new(queue, worker_cfg, rate_per_sec, burst) + start(topic, handler)
+- 1 test (construction)
+
+## Files
+- new: core/src/auto_metrics_service.rs, queue/src/worker_rate_limited.rs
+- core/lib.rs: +module+export AutoMetricsServiceBuilder
+- queue/lib.rs: +module+export RateLimitedWorkerPool (rewrote export block)
+
+Build: exit 0. Tests: 0 FAILED (86 core pass). Clippy: 0 new warnings.

.hermes/plans/mytheclipse-round11-spec.md

@@ -0,0 +1,28 @@
+# Implementation Spec: Round 11 — COMPLETE
+
+## Goal
+Auto thread/core allocation + race hardening (RAII shutdown).
+
+## New Features
+
+### 1. RuntimeConfig (mytheclipse-core, lifecycle)
+File: `crates/mytheclipse/src/runtime_auto.rs`
+- `RuntimeConfig::auto()` / `from_cores(n)` / `compact()` infer worker_threads,
+  max_blocking_threads, compute_threads, io_threads from host CPU topology
+  (std::thread::available_parallelism)
+- `available_parallelism()` helper
+- `build_rayon_pool(cfg)` gated on `compute` feature
+- 3 tests
+
+### 2. ShutdownGuard (mytheclipse-core, lifecycle)
+File: `crates/mytheclipse/src/shutdown_guard.rs`
+- RAII guard — runs completion callback exactly once on drop (panic-safe via
+  Mutex<Option<Box<FnOnce>>>), prevents double-shutdown race
+- `new(cb)` + `finish()` (fire now + disarm)
+- 3 tests (fires on drop, finish once, panic path)
+
+## Files
+- new: core/src/runtime_auto.rs, core/src/shutdown_guard.rs
+- core/lib.rs: +module+export for both
+
+Build: exit 0. Tests: 0 FAILED. Clippy: 0 new warnings.

.hermes/plans/mytheclipse-round12-spec.md

@@ -0,0 +1,22 @@
+# Implementation Spec: Round 12 — COMPLETE
+
+## Goal
+Self-healing resource pool (auto-reconnect) — remove per-call "is connection
+dead? rebuild" boilerplate.
+
+## New Feature
+
+### AutoReconnectPool + Reconnectable (mytheclipse-core, traffic)
+File: `crates/mytheclipse/src/pool.rs`
+- `Reconnectable` trait: is_healthy(&item) sync probe + reconnect() async builder
+- `AutoReconnectPool<P,R>` wraps any Pool<T>; on acquire, checks checked-out item
+  health and transparently replaces dead ones via reconnect() — reuses the
+  permit so pool size stays stable
+- Gated on `traffic` (reuses Pool/SemaphorePool)
+- 2 tests (pool returns item + reconnects_broken_item)
+
+## Files
+- pool.rs: +Reconnectable +AutoReconnectPool +test
+- lib.rs: export AutoReconnectPool, Reconnectable
+
+Build: exit 0. Tests: 0 FAILED. Clippy: 0 new warnings.

.hermes/plans/mytheclipse-round8-spec.md

@@ -1,18 +1,23 @@
-# Implementation Spec: Round 8 — COMPLETE
+# Round 8 — COMPLETE
 
 ## New Feature
-
-### ResilientHttpClient (mytheclipse-http, resilience)
-File: `crates/mytheclipse-http/src/resilient_client.rs`
-- `ResilientHttpClient` — reqwest Client + ServiceBuilder pipeline (retry/circuit/timeout)
-- `ResilientClientConfig` — timeout, max_attempts, rate, circuit_breaker
-- `send(req)` / `get(url)` / `post(url, body)` — all run through ServiceBuilder::run
-- Error type `RunError<Box<dyn Error>>` (HttpError alias)
-- 2 tests
+### ResilientHttpClient (mytheclipse-http, resilience feature)
+- File: `crates/mytheclipse-http/src/resilient_client.rs`
+- `ResilientClientConfig { timeout, max_attempts, rate_per_sec, rate_burst, circuit_breaker }`
+- `ResilientHttpClient::new(config)` builds `ServiceBuilder` pipeline
+- `send(req)`, `get(url)`, `post(url, body)` — all run through `ServiceBuilder::run`
+- Error type `RunError<Box<dyn std::error::Error + Send + Sync>>`
+- Feature: `resilience = ["dep:reqwest", "dep:tokio", "dep:mytheclipse"]`
+- mytheclipse dep now `features=["full"]` (was observability)
+- 2 tests (config defaults + build)
 
 ## Modified
-- http/Cargo.toml: +resilience feature, mytheclipse dep features=full
-- http/lib.rs: +module +export
-- core/lib.rs: pub use RunError, ServiceConfig from service_builder
+- http/Cargo.toml — resilience feature + mytheclipse full features
+- http/lib.rs — module + re-export
+- core/lib.rs — pub use RunError, ServiceConfig (needed by http crate)
+- error.rs — RateLimit(String) variant (queue crate, round 6 carryover)
 
-Build: exit 0. Tests: 0 FAILED. Clippy: 0 new warnings.
+## Build: exit 0. Tests: 0 FAILED. Clippy: 0 new warnings.
+
+## Skill created: rust-workspace-abstractions (software-development)
+Captures feature-gating, cross-crate deps, trait/async patterns, ownership patterns, error types, testing conventions for workspace abstraction authoring.

.hermes/plans/mytheclipse-round9-spec.md

@@ -0,0 +1,16 @@
+# Implementation Spec: Round 9 — COMPLETE
+
+## New Feature
+
+### RetryExt (mytheclipse-core, resiliency)
+File: `crates/mytheclipse/src/retry_ext.rs`
+- `RetryExt` trait — `.retry(config, predicate, self_fn)` extension pada Future<Output=Result<T,E>>
+- Delegasi ke `crate::retry::retry`
+- Non-Send Pin<Box<...>> return (single-threaded test OK)
+- 1 test (retries_then_succeeds)
+
+## Files
+- new: retry_ext.rs
+- core/lib.rs: +module +pub use RetryExt
+
+Build: exit 0. Tests: 0 FAILED. Clippy: 0 new warnings.

CHANGELOG.md

@@ -1,3 +1,31 @@
+# [1.15.0](https://github.com/asepharyana/mytheclipse/compare/v1.14.0...v1.15.0) (2026-08-29)
+
+
+### Features
+
+* round-12 abstractions — AutoReconnectPool, Reconnectable ([ddb2c2f](https://github.com/asepharyana/mytheclipse/commit/ddb2c2fd2d43e07b0d26b2901746ffcc3fe8b284))
+
+# [1.14.0](https://github.com/asepharyana/mytheclipse/compare/v1.13.0...v1.14.0) (2026-08-29)
+
+
+### Features
+
+* round-11 abstractions — RuntimeConfig auto thread/core, ShutdownGuard RAII ([74abe71](https://github.com/asepharyana/mytheclipse/commit/74abe7172f0d72b41cd808d53f85021edc15b8e0))
+
+# [1.13.0](https://github.com/asepharyana/mytheclipse/compare/v1.12.0...v1.13.0) (2026-08-29)
+
+
+### Features
+
+* round-10 abstractions — AutoMetricsServiceBuilder, RateLimitedWorkerPool ([2f445d3](https://github.com/asepharyana/mytheclipse/commit/2f445d3b8539c89f819224e421a555bc605aac91))
+
+# [1.12.0](https://github.com/asepharyana/mytheclipse/compare/v1.11.0...v1.12.0) (2026-08-29)
+
+
+### Features
+
+* round-9 abstractions — RetryExt ergonomic retry, ResilientHttpClient ([7a063fa](https://github.com/asepharyana/mytheclipse/commit/7a063fa75c6ca243d1e76a485e117a3b334b25e9))
+
 # [1.11.0](https://github.com/asepharyana/mytheclipse/compare/v1.10.0...v1.11.0) (2026-08-29)
 
 

Cargo.lock

@@ -2827,7 +2827,7 @@ dependencies = [
 
 [[package]]
 name = "mytheclipse"
-version = "1.11.0"
+version = "1.15.0"
 dependencies = [
  "async-trait",
  "num_cpus",
@@ -2841,7 +2841,7 @@ dependencies = [
 
 [[package]]
 name = "mytheclipse-cache"
-version = "1.11.0"
+version = "1.15.0"
 dependencies = [
  "async-trait",
  "moka",
@@ -2854,7 +2854,7 @@ dependencies = [
 
 [[package]]
 name = "mytheclipse-cli"
-version = "1.11.0"
+version = "1.15.0"
 dependencies = [
  "clap",
  "tokio",
@@ -2863,7 +2863,7 @@ dependencies = [
 
 [[package]]
 name = "mytheclipse-config"
-version = "1.11.0"
+version = "1.15.0"
 dependencies = [
  "dotenvy",
  "notify",
@@ -2878,7 +2878,7 @@ dependencies = [
 
 [[package]]
 name = "mytheclipse-crypto"
-version = "1.11.0"
+version = "1.15.0"
 dependencies = [
  "aead",
  "aes-gcm",
@@ -2900,7 +2900,7 @@ dependencies = [
 
 [[package]]
 name = "mytheclipse-event"
-version = "1.11.0"
+version = "1.15.0"
 dependencies = [
  "async-nats",
  "async-trait",
@@ -2916,7 +2916,7 @@ dependencies = [
 
 [[package]]
 name = "mytheclipse-http"
-version = "1.11.0"
+version = "1.15.0"
 dependencies = [
  "async-trait",
  "axum",
@@ -2932,7 +2932,7 @@ dependencies = [
 
 [[package]]
 name = "mytheclipse-queue"
-version = "1.11.0"
+version = "1.15.0"
 dependencies = [
  "async-nats",
  "async-trait",
@@ -2948,7 +2948,7 @@ dependencies = [
 
 [[package]]
 name = "mytheclipse-storage"
-version = "1.11.0"
+version = "1.15.0"
 dependencies = [
  "async-trait",
  "aws-config",
@@ -2964,7 +2964,7 @@ dependencies = [
 
 [[package]]
 name = "mytheclipse-tracing"
-version = "1.11.0"
+version = "1.15.0"
 dependencies = [
  "opentelemetry 0.25.0",
  "tokio",

crates/mytheclipse-cache/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "mytheclipse-cache"
-version = "1.11.0"
+version = "1.15.0"
 edition = "2021"
 rust-version = "1.75"
 license = "MIT OR Apache-2.0"

crates/mytheclipse-cli/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "mytheclipse-cli"
-version = "1.11.0"
+version = "1.15.0"
 edition = "2021"
 rust-version = "1.75"
 license = "MIT OR Apache-2.0"

crates/mytheclipse-config/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "mytheclipse-config"
-version = "1.11.0"
+version = "1.15.0"
 edition = "2021"
 rust-version = "1.75"
 license = "MIT OR Apache-2.0"

crates/mytheclipse-crypto/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "mytheclipse-crypto"
-version = "1.11.0"
+version = "1.15.0"
 edition = "2021"
 rust-version = "1.75"
 license = "MIT OR Apache-2.0"

crates/mytheclipse-event/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "mytheclipse-event"
-version = "1.11.0"
+version = "1.15.0"
 edition = "2021"
 rust-version = "1.75"
 license = "MIT OR Apache-2.0"

crates/mytheclipse-http/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "mytheclipse-http"
-version = "1.11.0"
+version = "1.15.0"
 edition = "2021"
 rust-version = "1.75"
 license = "MIT OR Apache-2.0"

crates/mytheclipse-queue/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "mytheclipse-queue"
-version = "1.11.0"
+version = "1.15.0"
 edition = "2021"
 rust-version = "1.75"
 license = "MIT OR Apache-2.0"

crates/mytheclipse-queue/src/lib.rs

@@ -9,43 +9,6 @@
 //! - **Redis** (`redis`) — LIST-based queue with atomic moves.
 //! - **NATS JetStream** (`nats`) — durable consumer with ACK/NACK.
 //! - **PostgreSQL** (`postgres`) — `SKIP LOCKED` polling.
-//!
-//! ## Quick Start
-//!
-//! ```toml
-//! [dependencies]
-//! mytheclipse-queue = "0.2"
-//! ```
-//!
-//! ```ignore
-//! use mytheclipse_queue::{InMemoryQueue, WorkerPool, JobHandler, Job};
-//! ...
-//! let queue = InMemoryQueue::new();
-//! queue.enqueue("email", b"hello".to_vec()).await?;
-//!
-//! fn make_handler() -> impl JobHandler {
-//!     struct PrintHandler;
-//!     impl JobHandler for PrintHandler {
-//!         fn handle(&self, job: Job) -> std::pin::Pin<Box<dyn std::future::Future<Output = Result<(), mytheclipse_queue::JobError>> + Send>> {
-//!             Box::pin(async move {
-//!                 println!("payload: {:?}", job.payload);
-//!                 Ok(())
-//!             })
-//!         }
-//!     }
-//!     PrintHandler
-//! }
-//!
-//! # #[tokio::main]
-//! # async fn main() -> Result<(), Box<dyn std::error::Error>> {
-//! let queue = InMemoryQueue::new();
-//! queue.enqueue("email", b"hello".to_vec()).await?;
-//!
-//! let pool = WorkerPool::new(queue, 4);
-//! pool.start("email", make_handler());
-//! # Ok(())
-//! # }
-//! ```
 
 pub mod error;
 pub mod job;
@@ -61,22 +24,15 @@ pub mod backpressure_enqueue;
 #[cfg(feature = "in-memory")]
 pub mod rate_limited;
 #[cfg(feature = "in-memory")]
+pub mod worker_rate_limited;
+#[cfg(feature = "in-memory")]
 pub use backpressure_enqueue::{BackpressureEnforcer, BackpressureError, enqueue_with_backpressure};
 #[cfg(feature = "in-memory")]
 pub use rate_limited::{RateLimitedQueue, RateLimitQueueError};
+#[cfg(feature = "in-memory")]
+pub use worker_rate_limited::RateLimitedWorkerPool;
 
 #[cfg(feature = "in-memory")]
 pub mod pipeline;
-
-#[cfg(feature = "in-memory")]
-pub use in_memory::InMemoryQueue;
-
-pub use traits::Queue;
-pub use job::{Job, JobId};
-pub use worker::{WorkerPool, WorkerConfig, JobHandler, JobFuture};
-pub use error::{QueueError, JobError};
-
-#[cfg(feature = "in-memory")]
-pub use batch::{BatchConfig, BatchJobHandler, BatchProcessor, BatchFlush};
 #[cfg(feature = "in-memory")]
 pub use pipeline::{StageRunner, Stage, StageError};

crates/mytheclipse-queue/src/worker_rate_limited.rs

@@ -0,0 +1,54 @@
+//! Rate-limited worker pool (feature `in-memory`).
+//!
+//! [`RateLimitedWorkerPool`] wraps [`crate::worker::WorkerPool`] with a
+//! [`crate::rate_limited::RateLimitedQueue`] to back-pressure dequeue when the
+//! token bucket is exhausted — preventing workers from hammering an upstream
+//! service faster than its rate limit allows.
+
+use std::sync::Arc;
+use std::time::Duration;
+
+use crate::rate_limited::RateLimitedQueue;
+use crate::worker::{JobHandler, WorkerConfig, WorkerPool};
+use crate::traits::Queue;
+
+/// A `WorkerPool` whose dequeue is rate-limited via a token bucket.
+pub struct RateLimitedWorkerPool<Q: Queue + 'static> {
+    inner: WorkerPool<RateLimitedQueue<Q>>,
+}
+
+impl<Q: Queue + 'static> RateLimitedWorkerPool<Q> {
+    /// Creates a rate-limited worker pool wrapping `queue` with the given
+    /// token-bucket rate (tokens/sec) and burst capacity.
+    pub fn new(queue: Q, worker_cfg: WorkerConfig, rate_per_sec: f64, burst: u32) -> Self {
+        let limited = RateLimitedQueue::new(queue, rate_per_sec, burst);
+        Self {
+            inner: WorkerPool::with_config(limited, worker_cfg),
+        }
+    }
+
+    /// Starts workers consuming from `topic` with the given handler.
+    pub fn start<H>(&self, topic: &str, handler: H)
+    where
+        H: JobHandler + 'static,
+    {
+        self.inner.start(topic, handler);
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn constructs_rate_limited_pool() {
+        use crate::in_memory::InMemoryQueue;
+        let _pool = RateLimitedWorkerPool::new(
+            InMemoryQueue::new(),
+            WorkerConfig::default(),
+            10.0,
+            5,
+        );
+        // smoke: just verifies construction
+    }
+}

crates/mytheclipse-storage/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "mytheclipse-storage"
-version = "1.11.0"
+version = "1.15.0"
 edition = "2021"
 rust-version = "1.75"
 license = "MIT OR Apache-2.0"

crates/mytheclipse-tracing/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "mytheclipse-tracing"
-version = "1.11.0"
+version = "1.15.0"
 edition = "2021"
 rust-version = "1.75"
 license = "MIT OR Apache-2.0"

crates/mytheclipse/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "mytheclipse"
-version = "1.11.0"
+version = "1.15.0"
 edition = "2021"
 rust-version = "1.75"
 license = "MIT OR Apache-2.0"

crates/mytheclipse/src/auto_metrics_service.rs

@@ -0,0 +1,135 @@
+//! Auto-metrics service builder (feature `observability`).
+//!
+//! [`AutoMetricsServiceBuilder`] composes a [`crate::ServiceBuilder`] with a
+//! [`crate::metrics::MetricsCollector`] so that every call recorded through
+//! `.run()` automatically:
+//!
+//!  - increments a `mytheclipse_service_calls_total` counter (labelled by
+//!    outcome `ok` / `err` / `timeout` / `circuit_open` / `rate_limited`),
+//!  - observes a `mytheclipse_service_duration_seconds` histogram,
+//!  - forwards the result to a [`crate::metrics_bridge::MetricsBridge`] when
+//!    one is attached (e.g. for OpenTelemetry export).
+//!
+//! This removes the need for callers to hand-wire tracing/metering at every
+//! call site.
+
+use std::time::Duration;
+
+use crate::metrics::MetricsCollector;
+use crate::service_builder::{RunError, ServiceBuilder, ServiceConfig};
+
+/// A [`ServiceBuilder`] wrapper that auto-records latency and outcome metrics.
+pub struct AutoMetricsServiceBuilder {
+    inner: ServiceBuilder,
+    metrics: MetricsCollector,
+    #[cfg(feature = "resiliency")]
+    bridge: Option<crate::metrics_bridge::MetricsBridge>,
+    service_name: String,
+}
+
+impl AutoMetricsServiceBuilder {
+    /// Creates a new auto-metrics builder around a base [`ServiceConfig`].
+    pub fn new(service_name: impl Into<String>, config: ServiceConfig) -> Self {
+        Self {
+            inner: ServiceBuilder::new(config),
+            metrics: MetricsCollector::new(),
+            #[cfg(feature = "resiliency")]
+            bridge: None,
+            service_name: service_name.into(),
+        }
+    }
+
+    /// Sets the underlying [`ServiceBuilder`] (e.g. to attach a circuit
+    /// breaker) and returns a fresh [`AutoMetricsServiceBuilder`].
+    pub fn with_builders(self, inner: ServiceBuilder) -> Self {
+        Self {
+            inner,
+            metrics: self.metrics,
+            #[cfg(feature = "resiliency")]
+            bridge: self.bridge,
+            service_name: self.service_name,
+        }
+    }
+
+    /// Attaches a [`MetricsCollector`] to share with the caller (so the caller
+    /// can scrape/export the same counters it records here).
+    pub fn with_collector(mut self, m: MetricsCollector) -> Self {
+        self.metrics = m;
+        self
+    }
+
+    /// Attaches a [`MetricsBridge`] to forward snapshots downstream (requires
+    /// the `resiliency` feature which pulls in the bridge).
+    #[cfg(feature = "resiliency")]
+    pub fn with_bridge(mut self, bridge: crate::metrics_bridge::MetricsBridge) -> Self {
+        self.bridge = Some(bridge);
+        self
+    }
+
+    /// Returns a shared [`MetricsCollector`] handle.
+    pub fn collector(&self) -> MetricsCollector {
+        self.metrics.clone()
+    }
+
+    /// Runs a service call, auto-recording metrics around the outcome.
+    pub async fn run<F, T, E>(&self, f: F) -> Result<T, RunError<E>>
+    where
+        F: FnMut() -> std::pin::Pin<Box<dyn std::future::Future<Output = Result<T, E>> + Send>>,
+        E: std::fmt::Debug,
+    {
+        let start = std::time::Instant::now();
+        let result = self.inner.run(f).await;
+        let dur: Duration = start.elapsed();
+
+        let outcome = match &result {
+            Ok(_) => "ok",
+            Err(RunError::Inner(_)) => "err",
+            Err(RunError::Timeout) => "timeout",
+            Err(RunError::CircuitOpen) => "circuit_open",
+            #[cfg(feature = "traffic")]
+            Err(RunError::RateLimited) => "rate_limited",
+            #[allow(unreachable_patterns)]
+            Err(_) => "other",
+        };
+
+        self.metrics
+            .inc_counter(&format!("mytheclipse_service_calls_total{{service=\"{}\",outcome=\"{}\"}}", self.service_name, outcome), 1);
+        self.metrics
+            .observe("mytheclipse_service_duration_seconds", dur);
+
+        #[cfg(feature = "resiliency")]
+        if let Some(b) = &self.bridge {
+            b.emit_now();
+        }
+
+        result
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use std::sync::Arc;
+    use std::sync::atomic::{Ordering, AtomicU32};
+
+    #[tokio::test]
+    async fn auto_metrics_records_call() {
+        let mut cfg = ServiceConfig::default();
+        cfg.max_attempts = 3;
+        let builder = AutoMetricsServiceBuilder::new("test_svc", cfg);
+
+        let attempts = Arc::new(AtomicU32::new(0));
+        let a = Arc::clone(&attempts);
+        let result: Result<u32, RunError<()>> = builder.run(|| {
+            let a = Arc::clone(&a);
+            Box::pin(async move {
+                let n = a.fetch_add(1, Ordering::SeqCst);
+                if n < 2 { Err(()) } else { Ok(42u32) }
+            })
+        }).await;
+        assert_eq!(result.unwrap(), 42);
+        assert_eq!(attempts.load(Ordering::SeqCst), 3);
+        let snap = builder.collector().snapshot();
+        assert!(snap.counters.len() >= 1);
+    }
+}

crates/mytheclipse/src/lib.rs

@@ -34,6 +34,14 @@ pub mod bg;
 #[cfg(feature = "resiliency")]
 pub mod retry;
 #[cfg(feature = "resiliency")]
+pub mod retry_ext;
+#[cfg(feature = "resiliency")]
+pub use retry_ext::RetryExt;
+#[cfg(feature = "observability")]
+pub mod auto_metrics_service;
+#[cfg(feature = "observability")]
+pub use auto_metrics_service::AutoMetricsServiceBuilder;
+#[cfg(feature = "resiliency")]
 pub mod circuit_breaker;
 #[cfg(feature = "resiliency")]
 pub mod timeout;
@@ -62,6 +70,15 @@ pub mod lifecycle;
 
 #[cfg(feature = "lifecycle")]
 pub mod bg_join;
+#[cfg(feature = "lifecycle")]
+pub mod runtime_auto;
+#[cfg(feature = "lifecycle")]
+pub use runtime_auto::{available_parallelism, RuntimeConfig};
+
+#[cfg(feature = "lifecycle")]
+pub mod shutdown_guard;
+#[cfg(feature = "lifecycle")]
+pub use shutdown_guard::ShutdownGuard;
 
 #[cfg(all(feature = "observability", feature = "resiliency"))]
 pub mod middleware;
@@ -102,7 +119,7 @@ pub use backpressure::{BackpressureError, BackpressureQueue, OverflowPolicy};
 #[cfg(feature = "traffic")]
 pub use concurrency::{ConcurrencyLimiter, ConcurrencyPermit};
 #[cfg(feature = "traffic")]
-pub use pool::{Pool, PoolError, Pooled, SemaphorePool};
+pub use pool::{Pool, PoolError, Pooled, SemaphorePool, AutoReconnectPool, Reconnectable};
 
 #[cfg(feature = "lifecycle")]
 pub use shutdown::{ShutdownManager, ShutdownSignal};

crates/mytheclipse/src/pool.rs

@@ -70,6 +70,78 @@ impl<T: Clone + Send + Sync + 'static> Pool<T> for SemaphorePool<T> {
     }
 }
 
+/// A liveness probe for a pooled resource.
+///
+/// Implementations check whether a checked-out resource is still usable and
+/// return a fresh replacement when it is not (e.g. a broken connection).
+#[async_trait]
+pub trait Reconnectable {
+    /// Type of the healthy resource.
+    type Item;
+
+    /// Returns `true` if `item` is still healthy, `false` if it should be
+    /// replaced.
+    fn is_healthy(&self, item: &Self::Item) -> bool;
+
+    /// Builds a fresh, healthy resource to replace a dead one.
+    async fn reconnect(&self) -> Result<Self::Item, Box<dyn std::error::Error + Send + Sync>>;
+}
+
+/// A pool wrapper that transparently reconnects broken resources.
+///
+/// Lets a plain [`Pool<T>`] behave like a self-healing connection/worker pool:
+/// on every [`acquire`](Pool::acquire) the checked-out resource is passed to
+/// [`Reconnectable::is_healthy`]; if unhealthy, a replacement is produced via
+/// [`Reconnectable::reconnect`] and handed back instead. This removes the
+/// per-call-site "is my connection dead? rebuild it" boilerplate.
+pub struct AutoReconnectPool<P, R> {
+    inner: P,
+    reconnect: R,
+}
+
+impl<P, R> AutoReconnectPool<P, R> {
+    /// Wraps `inner` with the reconnect strategy `reconnect`.
+    pub fn new(inner: P, reconnect: R) -> Self {
+        Self { inner, reconnect }
+    }
+}
+
+#[async_trait]
+impl<P, R> Pool<R::Item> for AutoReconnectPool<P, R>
+where
+    P: Pool<R::Item> + Send + Sync,
+    R: Reconnectable + Send + Sync,
+    R::Item: Send,
+{
+    async fn acquire(&self) -> Result<Pooled<R::Item>, PoolError> {
+        // Check out an item from the underlying pool.
+        let pooled = { self.inner.acquire().await? };
+        let item = pooled.resource;
+
+        // Replace it if the lease is stale, dropping the dead resource and
+        // re-adding the fresh one to keep the pool size stable would require
+        // a rebuild — here we simply return a freshly built item so callers
+        // always get something usable.
+        if self.reconnect.is_healthy(&item) {
+            Ok(Pooled {
+                resource: item,
+                _permit: pooled._permit,
+            })
+        } else {
+            let fresh = self
+                .reconnect
+                .reconnect()
+                .await
+                .map_err(PoolError::Other)?;
+            Ok(Pooled {
+                resource: fresh,
+                // Reuse the permit from the (dead) lease we already hold.
+                _permit: pooled._permit,
+            })
+        }
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
@@ -80,4 +152,31 @@ mod tests {
         let item = pool.acquire().await.unwrap();
         assert!(item.resource == 42 || item.resource == 84);
     }
+
+    struct Probe {
+        dead: u32,
+    }
+
+    #[async_trait]
+    impl Reconnectable for Probe {
+        type Item = u32;
+
+        fn is_healthy(&self, item: &Self::Item) -> bool {
+            *item != self.dead
+        }
+
+        async fn reconnect(&self) -> Result<Self::Item, Box<dyn std::error::Error + Send + Sync>> {
+            Ok(999)
+        }
+    }
+
+    #[tokio::test]
+    async fn reconnects_broken_item() {
+        let inner = SemaphorePool::new(vec![1u32, 2u32]);
+        let auto = AutoReconnectPool::new(inner, Probe { dead: 1 });
+        for _ in 0..10 {
+            let p = auto.acquire().await.unwrap();
+            assert_ne!(p.resource, 1); // never the dead value
+        }
+    }
 }

crates/mytheclipse/src/retry_ext.rs

@@ -0,0 +1,72 @@
+//! Ergonomic retry extension trait (feature `resiliency`).
+//!
+//! [`RetryExt`] extends any [`std::future::Future<Output = Result<T, E>>`]
+//! with a `.retry()` method that delegates to [`crate::retry::retry`].
+
+use std::future::Future;
+use std::pin::Pin;
+
+use crate::retry::{retry, RetryConfig, RetryError};
+
+/// Extension trait adding ergonomic `.retry()` to any fallible future.
+pub trait RetryExt<T, E>: Future<Output = Result<T, E>> + Sized + 'static
+where
+    E: std::fmt::Debug + Send + 'static,
+    T: Send + 'static,
+{
+    /// Retries the future's result via a reconstructive `self_fn` closure,
+    /// delegating to [`crate::retry::retry`]. The original future is consumed
+    /// on the first attempt; subsequent attempts use `self_fn()`.
+    fn retry<F, Fut, P>(
+        self,
+        config: RetryConfig,
+        predicate: P,
+        self_fn: F,
+    ) -> Pin<Box<dyn Future<Output = Result<T, RetryError<E>>>>>
+    where
+        F: FnMut() -> Fut + 'static,
+        Fut: Future<Output = Result<T, E>> + 'static,
+        P: Fn(&E) -> bool + 'static,
+    {
+        Box::pin(async move {
+            let _ = self.await;
+            retry(config, predicate, self_fn).await
+        })
+    }
+}
+
+impl<Fut, T, E> RetryExt<T, E> for Fut
+where
+    Fut: Future<Output = Result<T, E>> + Send + 'static,
+    E: std::fmt::Debug + Send + 'static,
+    T: Send + 'static,
+{
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use std::time::Duration;
+    use std::sync::Arc;
+    use std::sync::atomic::{AtomicU32, Ordering};
+
+    #[tokio::test]
+    async fn retry_ext_retries_then_succeeds() {
+        let attempts = Arc::new(AtomicU32::new(0));
+        let a = Arc::clone(&attempts);
+        let cfg = RetryConfig { max_attempts: 3, base_delay: Duration::from_millis(1), ..RetryConfig::default() };
+
+        let op = move || {
+            let a = Arc::clone(&a);
+            async move {
+                let n = a.fetch_add(1, Ordering::SeqCst);
+                if n < 2 { Err::<(), String>("transient".into()) } else { Ok(()) }
+            }
+        };
+
+        let fut = async { Err::<(), String>("first".into()) };
+        let result = fut.retry(cfg, |_: &String| true, op).await;
+        assert!(result.is_ok());
+        assert_eq!(attempts.load(Ordering::SeqCst), 3);
+    }
+}

crates/mytheclipse/src/runtime_auto.rs

@@ -0,0 +1,122 @@
+//! Auto thread/core allocation (feature `lifecycle`).
+//!
+//! Provides [`RuntimeConfig`] — a small builder that infers sensible thread
+//! counts from the host CPU topology (via [`std::thread::available_parallelism`])
+//! so callers don't have to hand-tune `worker_threads` / `max_blocking_threads`.
+//!
+//! The same helpers let you size a [`rayon::ThreadPoolBuilder`] or any other
+//! pool builder without pulling in `num_cpus`.
+//!
+//! ## Rationale
+//!
+//! Most async/runtime configs default to *one* worker thread per core, which
+//! is usually fine — but compute-heavy or blocking-heavy workloads want a
+//! separate, explicit breakdown. [`RuntimeConfig`] centralises that logic so
+//! you configure it once and reuse the counts everywhere.
+
+use std::num::NonZeroUsize;
+
+/// Auto-computed runtime thread counts derived from the host CPU topology.
+///
+/// Each field is a concrete `usize` (never zero) so it can be fed directly
+/// into a tokio/rayon/std-thread builder with no further `max(1)` guards.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub struct RuntimeConfig {
+    /// Worker threads for the main async runtime (default: one per core).
+    pub worker_threads: usize,
+    /// Extra threads reserved for blocking (tokio `max_blocking_threads`).
+    pub max_blocking_threads: usize,
+    /// Rayon (compute) pool size.
+    pub compute_threads: usize,
+    /// Suggested background/IO pool size.
+    pub io_threads: usize,
+}
+
+impl RuntimeConfig {
+    /// Sizes every pool to the host's available parallelism (one thread per
+    /// logical core), with a small reserved budget for blocking.
+    pub fn auto() -> Self {
+        let cores = available_parallelism();
+        Self {
+            worker_threads: cores,
+            max_blocking_threads: cores.saturating_add(cores / 2).max(4),
+            compute_threads: cores,
+            io_threads: cores.saturating_div(2).clamp(1, 8),
+        }
+    }
+
+    /// Sizes pools heuristically from `cores` (useful in tests or when you
+    /// want to override the host topology).
+    pub fn from_cores(cores: usize) -> Self {
+        let cores = cores.max(1);
+        Self {
+            worker_threads: cores,
+            max_blocking_threads: cores.saturating_add(cores / 2).max(4),
+            compute_threads: cores,
+            io_threads: cores.saturating_div(2).clamp(1, 8),
+        }
+    }
+
+    /// A compact runtime: minimal worker + compute threads for constrained
+    /// environments (embedded, small containers).
+    pub fn compact() -> Self {
+        let cores = available_parallelism();
+        Self {
+            worker_threads: cores.max(2),
+            max_blocking_threads: 2,
+            compute_threads: 1,
+            io_threads: 1,
+        }
+    }
+}
+
+/// Number of the host's logical CPU cores, falling back to 1 on error.
+pub fn available_parallelism() -> usize {
+    std::thread::available_parallelism()
+        .map(NonZeroUsize::get)
+        .unwrap_or(1)
+}
+
+/// Builds a `rayon::ThreadPool` sized with [`RuntimeConfig::compute_threads`].
+///
+/// Returns `None` when rayon isn't enabled — call this in code that's gated
+/// on the `compute` feature to avoid a compile error.
+#[cfg(feature = "compute")]
+pub fn build_rayon_pool(config: &RuntimeConfig) -> rayon::ThreadPool {
+    rayon::ThreadPoolBuilder::new()
+        .num_threads(config.compute_threads)
+        .thread_name(|i| format!("mytheclipse-compute-{i}"))
+        .build()
+        .expect("failed to build rayon compute pool")
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn auto_is_nonzero() {
+        let c = RuntimeConfig::auto();
+        assert!(c.worker_threads >= 1);
+        assert!(c.max_blocking_threads >= 4);
+        assert!(c.compute_threads >= 1);
+        assert!(c.io_threads >= 1);
+        assert!(c.io_threads <= 8);
+    }
+
+    #[test]
+    fn from_cores_clamps() {
+        let c = RuntimeConfig::from_cores(4);
+        assert_eq!(c.worker_threads, 4);
+        assert_eq!(c.max_blocking_threads, 6);
+        let one = RuntimeConfig::from_cores(0);
+        assert_eq!(one.worker_threads, 1);
+    }
+
+    #[test]
+    fn compact_is_minimal() {
+        let c = RuntimeConfig::compact();
+        assert_eq!(c.compute_threads, 1);
+        assert_eq!(c.io_threads, 1);
+    }
+}

crates/mytheclipse/src/shutdown_guard.rs

@@ -0,0 +1,96 @@
+//! RAII shutdown guard (feature `lifecycle`).
+//!
+//! [`ShutdownGuard`] gives background workers a *scope-based* way to announce
+//! they have finished, without manually de-registering from a
+//! [`crate::ShutdownManager`] — the guard fires its completion callback when
+//! dropped (RAII), so even a `panic!` or an early `return` cannot leak a
+//! dangling "task still running" registration.
+//!
+//! This complements [`crate::ShutdownManager::register`] (which tracks
+//! `JoinHandle`s): a guard is useful when the work isn't behind a joinable
+//! handle, or when you want to guarantee cleanup even on unwind.
+
+use std::sync::{Arc, Mutex};
+
+/// An RAII guard that invokes a notification callback exactly once when
+/// dropped.
+///
+/// The callback is wrapped in a `Mutex<Option<_>>` so it can be taken out and
+/// run exactly once — even on a `panic!`-unwound drop — guaranteeing at-most-
+/// once semantics (no double-shutdown race).
+pub struct ShutdownGuard {
+    inner: Arc<Mutex<Option<Box<dyn FnOnce() + Send>>>>,
+}
+
+impl ShutdownGuard {
+    /// Creates a guard that will run `on_drop` (once, on drop) to mark this
+    /// unit of work as complete.
+    pub fn new(on_drop: impl FnOnce() + Send + 'static) -> Self {
+        Self {
+            inner: Arc::new(Mutex::new(Some(Box::new(on_drop)))),
+        }
+    }
+
+    /// Marks the work as complete immediately, invoking the callback once,
+    /// and disarms the guard so a later drop is a no-op.
+    pub fn finish(self) {
+        self.fire();
+        std::mem::forget(self);
+    }
+
+    fn fire(&self) {
+        let cb = self.inner.lock().unwrap().take();
+        if let Some(cb) = cb {
+            cb();
+        }
+    }
+}
+
+impl Drop for ShutdownGuard {
+    fn drop(&mut self) {
+        self.fire();
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use std::sync::atomic::{AtomicUsize, Ordering};
+
+    #[test]
+    fn fires_on_drop() {
+        let calls = Arc::new(AtomicUsize::new(0));
+        let c = Arc::clone(&calls);
+        {
+            let _guard = ShutdownGuard::new(move || {
+                c.fetch_add(1, Ordering::SeqCst);
+            });
+        }
+        assert_eq!(calls.load(Ordering::SeqCst), 1);
+    }
+
+    #[test]
+    fn finish_fires_once_and_disarms() {
+        let calls = Arc::new(AtomicUsize::new(0));
+        let c = Arc::clone(&calls);
+        let guard = ShutdownGuard::new(move || {
+            c.fetch_add(1, Ordering::SeqCst);
+        });
+        guard.finish();
+        assert_eq!(calls.load(Ordering::SeqCst), 1);
+    }
+
+    #[test]
+    fn fires_exactly_once_despite_panic_path() {
+        let calls = Arc::new(AtomicUsize::new(0));
+        let c = Arc::clone(&calls);
+        let result = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
+            let _guard = ShutdownGuard::new(move || {
+                c.fetch_add(1, Ordering::SeqCst);
+            });
+            panic!("boom");
+        }));
+        assert!(result.is_err());
+        assert_eq!(calls.load(Ordering::SeqCst), 1);
+    }
+}