ci: publish mytheclipse-queue + mytheclipse-tracing; make publish idempotent

The publish workflow's hardcoded crate list omitted mytheclipse-queue and
mytheclipse-tracing, so they were never released to crates.io despite the
workspace being bumped to 1.21.2. Add both to the loop and make each
publish tolerant of already-uploaded versions so re-runs are safe.

.github/workflows/publish.yml

@@ -27,12 +27,17 @@ jobs:
 
       # The workspace crates have no interdependencies, so publish order
       # doesn't matter for crates.io dependency resolution. A brief sleep
-      # between publishes avoids hitting crates.io's rate limit.
+      # between publishes avoids hitting crates.io's rate limit. Re-publishing
+      # an already-uploaded version is tolerated (idempotent re-runs).
       - name: Publish workspace crates
         run: |
-          for crate in mytheclipse mytheclipse-cache mytheclipse-storage mytheclipse-event mytheclipse-config mytheclipse-crypto; do
+          for crate in mytheclipse mytheclipse-cache mytheclipse-storage mytheclipse-event mytheclipse-config mytheclipse-crypto mytheclipse-queue mytheclipse-tracing; do
             echo "Publishing $crate..."
-            cargo publish -p "$crate" --allow-dirty --no-verify
+            if cargo publish -p "$crate" --allow-dirty --no-verify; then
+              echo "$crate published OK"
+            else
+              echo "$crate: publish failed (already uploaded or error)"
+            fi
             sleep 15
           done
         env:

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

@@ -0,0 +1,48 @@
+# Implementation Spec: Round 21 — CPU Parallel Compute Primitives
+
+## Goal
+Fitur parallel khusus CPU (rayon) yang bounded, panic-isolated, error-aggregated.
+Melengkapi `compute()` (single call) dengan batch parallel + fork-join.
+
+## New API (crates/mytheclipse/src/compute.rs, feature `compute`)
+
+### 1. `compute_map<I, T, F>(items, f) -> Result<Vec<T>, ComputeErrors>`
+- `pack_items` di rayon compute pool: `par_iter().map(f)` — bounded concurrency
+  otomatis (rayon work-stealing sizing = CPU cores), ordered output.
+- `f: Fn(I::Item) -> Result<T, ComputeMapItemError>`:
+  - item error string → dikumpulkan
+  - panic per item di-catch (catch_unwind) → jadi error, pool survive
+- `ComputeErrors { errors: Vec<String> }` — Display, Error, len, is_empty.
+  (Tidak pakai AggregateError — feature `compute` harus compile tanpa resiliency.)
+- `I: IntoParallelIterator` (rayon) — work langsung di pool, tanpa materialize.
+
+### 2. `compute_join<A, B, RA, RB>(a, b) -> Result<(RA, RB), MytheclipseError>`
+- `rayon::join` wrapper di compute pool: 2 heavy closures run parallel.
+- Panic-isolated (catch_unwind per branch) — pool survive, error jadi
+  ComputePanic.
+
+### 3. `compute_par_for_each<I>(items, f) -> Result<(), ComputeErrors>`
+- `par_iter().for_each` idiom — fire side-effects parallel di pool.
+- Panic isolation per item.
+
+## Design notes
+- Reuse `context().compute_pool` (existing sizing: compute_threads dari
+  RuntimeConfig / available_parallelism) — konsisten dengan `compute()`.
+- `rayon::ThreadPool::install` untuk semua — force run di pool.
+- Panic isolation: `std::panic::catch_unwind` + AssertUnwindSafe per item
+  (sama seperti `compute()` yang sudah proven).
+- Bounded = rayon work-stealing — concurrency = pool threads (CPU cores),
+  bukan item count. Tidak perlu semaphore.
+
+## Files
+- crates/mytheclipse/src/compute.rs (3 fungsi + error type)
+- crates/mytheclipse/src/lib.rs (export)
+- doctests: compute_map, compute_join, compute_par_for_each
+- tests: unit di compute.rs
+
+## Verification
+1. `cargo build --workspace --all-features` — exit 0
+2. `cargo test -p mytheclipse compute --all-features` — 0 FAILED
+3. `cargo test -p mytheclipse --doc --all-features` — 0 FAILED
+4. `cargo clippy --workspace --all-features` — 0 new
+5. spec + commit + push

CHANGELOG.md

@@ -1,3 +1,24 @@
+## [1.21.2](https://github.com/asepharyana/mytheclipse/compare/v1.21.1...v1.21.2) (2026-08-30)
+
+
+### Bug Fixes
+
+* **cache:** align redis dep to 0.32 for deadpool unification ([901fc51](https://github.com/asepharyana/mytheclipse/commit/901fc51f7b3bf4b2ab433db2e35303722778cc76))
+
+## [1.21.1](https://github.com/asepharyana/mytheclipse/compare/v1.21.0...v1.21.1) (2026-08-29)
+
+
+### Bug Fixes
+
+* **ci:** restore full CI green — test-matrix, clippy, rustfmt, and rustdoc gates ([1dfc6d6](https://github.com/asepharyana/mytheclipse/commit/1dfc6d68657e5b002896f400d744b052800c9285))
+
+# [1.21.0](https://github.com/asepharyana/mytheclipse/compare/v1.20.0...v1.21.0) (2026-08-29)
+
+
+### Features
+
+* CPU parallel compute primitives — compute_map, compute_join, compute_par_for_each ([e148944](https://github.com/asepharyana/mytheclipse/commit/e14894479d8ca716a9decf2c1403359fdc376717))
+
 # [1.20.0](https://github.com/asepharyana/mytheclipse/compare/v1.19.0...v1.20.0) (2026-08-29)
 
 

Cargo.lock

@@ -2941,7 +2941,7 @@ dependencies = [
 
 [[package]]
 name = "mytheclipse"
-version = "1.20.0"
+version = "1.21.2"
 dependencies = [
  "async-trait",
  "criterion",
@@ -2956,11 +2956,11 @@ dependencies = [
 
 [[package]]
 name = "mytheclipse-cache"
-version = "1.20.0"
+version = "1.21.2"
 dependencies = [
  "async-trait",
  "moka",
- "redis",
+ "redis 0.32.7",
  "serde",
  "serde_json",
  "tokio",
@@ -2969,7 +2969,7 @@ dependencies = [
 
 [[package]]
 name = "mytheclipse-cli"
-version = "1.20.0"
+version = "1.21.2"
 dependencies = [
  "clap",
  "tokio",
@@ -2978,7 +2978,7 @@ dependencies = [
 
 [[package]]
 name = "mytheclipse-config"
-version = "1.20.0"
+version = "1.21.2"
 dependencies = [
  "dotenvy",
  "notify",
@@ -2993,7 +2993,7 @@ dependencies = [
 
 [[package]]
 name = "mytheclipse-crypto"
-version = "1.20.0"
+version = "1.21.2"
 dependencies = [
  "aead",
  "aes-gcm",
@@ -3015,7 +3015,7 @@ dependencies = [
 
 [[package]]
 name = "mytheclipse-event"
-version = "1.20.0"
+version = "1.21.2"
 dependencies = [
  "async-nats",
  "async-trait",
@@ -3031,7 +3031,7 @@ dependencies = [
 
 [[package]]
 name = "mytheclipse-http"
-version = "1.20.0"
+version = "1.21.2"
 dependencies = [
  "async-trait",
  "axum",
@@ -3047,12 +3047,12 @@ dependencies = [
 
 [[package]]
 name = "mytheclipse-queue"
-version = "1.20.0"
+version = "1.21.2"
 dependencies = [
  "async-nats",
  "async-trait",
  "bytes",
- "redis",
+ "redis 0.27.6",
  "serde",
  "serde_json",
  "tokio",
@@ -3063,7 +3063,7 @@ dependencies = [
 
 [[package]]
 name = "mytheclipse-storage"
-version = "1.20.0"
+version = "1.21.2"
 dependencies = [
  "async-trait",
  "aws-config",
@@ -3079,7 +3079,7 @@ dependencies = [
 
 [[package]]
 name = "mytheclipse-tracing"
-version = "1.20.0"
+version = "1.21.2"
 dependencies = [
  "opentelemetry 0.25.0",
  "tokio",
@@ -3986,6 +3986,27 @@ dependencies = [
  "url",
 ]
 
+[[package]]
+name = "redis"
+version = "0.32.7"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "014cc767fefab6a3e798ca45112bccad9c6e0e218fbd49720042716c73cfef44"
+dependencies = [
+ "bytes",
+ "cfg-if",
+ "combine",
+ "futures-util",
+ "itoa",
+ "num-bigint",
+ "percent-encoding",
+ "pin-project-lite",
+ "ryu",
+ "socket2 0.6.5",
+ "tokio",
+ "tokio-util",
+ "url",
+]
+
 [[package]]
 name = "redox_syscall"
 version = "0.5.18"

crates/mytheclipse-cache/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "mytheclipse-cache"
-version = "1.20.0"
+version = "1.21.2"
 edition = "2021"
 rust-version = "1.75"
 license = "MIT OR Apache-2.0"
@@ -34,7 +34,7 @@ serde_json = { version = "1", optional = true }
 moka = { version = "0.12", default-features = false, features = ["future"], optional = true }
 
 # L2: Redis/Valkey async client (multiplexed connection).
-redis = { version = "0.27", default-features = false, features = ["tokio-comp"], optional = true }
+redis = { version = "0.32", default-features = false, features = ["tokio-comp"], optional = true }
 tokio = { version = "1.53", features = ["sync", "rt"], optional = true }
 
 [dev-dependencies]

crates/mytheclipse-cache/src/auto_refresh.rs

@@ -71,7 +71,12 @@ where
     }
 
     /// Sets a value in the underlying cache.
-    pub async fn set(&self, key: &str, value: Vec<u8>, ttl: Option<Duration>) -> Result<(), CacheError> {
+    pub async fn set(
+        &self,
+        key: &str,
+        value: Vec<u8>,
+        ttl: Option<Duration>,
+    ) -> Result<(), CacheError> {
         self.inner.set(key, value, ttl).await
     }
 

crates/mytheclipse-cache/src/metrics.rs

@@ -54,6 +54,10 @@ pub struct CacheSnapshot {
 impl CacheSnapshot {
     pub fn hit_rate(&self) -> f64 {
         let total = self.hits + self.misses;
-        if total == 0 { 0.0 } else { self.hits as f64 / total as f64 }
+        if total == 0 {
+            0.0
+        } else {
+            self.hits as f64 / total as f64
+        }
     }
 }

crates/mytheclipse-cli/Cargo.toml

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

crates/mytheclipse-cli/src/builder.rs

@@ -1,6 +1,6 @@
 //! Clap-based CLI builder implementation.
 
-use clap::{Parser, Subcommand as ClapSubcommand};
+use clap::{CommandFactory, FromArgMatches, Parser, Subcommand as ClapSubcommand};
 
 /// A mytheclipse CLI application.
 #[derive(Parser, Debug)]
@@ -52,6 +52,17 @@ impl CliBuilder {
     }
 
     pub fn build(self) -> CliApp {
-        CliApp::parse()
+        // Apply the configured name/about to the derived clap Command so the
+        // builder's fields are honored in the rendered help/usage.
+        let Self { name, about } = self;
+        // clap's `Str`/`StyledStr` only accept 'static references, so leak
+        // the owned strings (build(self) consumes self once, so a single,
+        // process-lifetime leak is acceptable).
+        let name: &'static str = String::leak(name);
+        let about: &'static str = String::leak(about);
+        let cmd = <CliApp as CommandFactory>::command()
+            .name(name)
+            .about(about);
+        CliApp::from_arg_matches(&cmd.get_matches()).unwrap_or_else(|e| e.exit())
     }
 }

crates/mytheclipse-config/Cargo.toml

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

crates/mytheclipse-config/src/lib.rs

@@ -51,9 +51,8 @@ pub use loader::ConfigLoader;
 
 #[cfg(feature = "validation")]
 pub use validate::{
-    collect_failures, validate_non_empty, validate_port, validate_range,
+    collect_failures, validate_non_empty, validate_port, validate_range, validate_url,
-    validate_url, ConfigValidator, ConfigValidatorExt, ValidationError,
+    ConfigValidator, ConfigValidatorExt, ValidationError, ValidationFailure,
-    ValidationFailure,
 };
 
 #[cfg(feature = "hot-reload")]

crates/mytheclipse-config/src/validate.rs

@@ -33,7 +33,11 @@ pub struct ValidationError {
 
 impl fmt::Display for ValidationError {
     fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
-        write!(f, "config validation failed ({} issue(s)):", self.failures.len())?;
+        write!(
+            f,
+            "config validation failed ({} issue(s)):",
+            self.failures.len()
+        )?;
         for failure in &self.failures {
             write!(f, "\n  - {failure}")?;
         }
@@ -91,7 +95,11 @@ pub fn validate_url(path: &str, value: &str) -> Option<ValidationFailure> {
     }
     // Minimal heuristic: scheme + host. We avoid pulling in a full URL crate
     // to keep the dependency surface small.
-    let scheme_len = if value.starts_with("http://") { 7 } else if value.starts_with("https://") { 8 } else {
+    let scheme_len = if value.starts_with("http://") {
+        7
+    } else if value.starts_with("https://") {
+        8
+    } else {
         return Some(ValidationFailure {
             path: path.to_string(),
             message: format!("url must start with http:// or https:// (got {value:?})"),
@@ -148,7 +156,9 @@ where
 }
 
 /// Collects all failures from an iterator of `Option<ValidationFailure>`.
-pub fn collect_failures(opts: impl IntoIterator<Item = Option<ValidationFailure>>) -> Result<(), ValidationError> {
+pub fn collect_failures(
+    opts: impl IntoIterator<Item = Option<ValidationFailure>>,
+) -> Result<(), ValidationError> {
     let failures: Vec<_> = opts.into_iter().flatten().collect();
     if failures.is_empty() {
         Ok(())
@@ -186,7 +196,11 @@ mod tests {
 
     #[test]
     fn collect_failures_aggregates_all() {
-        let opts = [validate_non_empty("a", ""), validate_non_empty("b", "ok"), validate_url("c.d", "bad://x")];
+        let opts = [
+            validate_non_empty("a", ""),
+            validate_non_empty("b", "ok"),
+            validate_url("c.d", "bad://x"),
+        ];
         let err = collect_failures(opts).unwrap_err();
         assert_eq!(err.failures.len(), 2);
         assert_eq!(err.failures[0].path, "a");
@@ -195,7 +209,10 @@ mod tests {
 
     #[test]
     fn collect_failures_ok_when_all_pass() {
-        let opts = [validate_url("a", "https://ok.com"), validate_port("b", 8080)];
+        let opts = [
+            validate_url("a", "https://ok.com"),
+            validate_port("b", 8080),
+        ];
         assert!(collect_failures(opts).is_ok());
     }
 
@@ -205,7 +222,10 @@ mod tests {
         impl ConfigValidator for Cfg {
             fn validate(&self) -> Result<(), ValidationError> {
                 Err(ValidationError {
-                    failures: vec![ValidationFailure { path: "x".into(), message: "bad".into() }],
+                    failures: vec![ValidationFailure {
+                        path: "x".into(),
+                        message: "bad".into(),
+                    }],
                 })
             }
         }

crates/mytheclipse-crypto/Cargo.toml

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

crates/mytheclipse-crypto/src/hkdf.rs

@@ -4,8 +4,8 @@
 //! domain-specific sub-keys from a single master secret. Each purpose
 //! string acts as the `info` parameter for domain separation.
 
-use sha2::Sha256;
 use hkdf::Hkdf;
+use sha2::Sha256;
 
 /// Derives sub-keys from a master secret using HKDF-SHA256.
 pub struct HkdfKeyDeriver {

crates/mytheclipse-crypto/src/key_registry.rs

@@ -1,6 +1,6 @@
 //! Typed key registry with ID-based lookup (feature `password`).
 //!
-//! [`TypedKeyRegistry`] extends [`KeyRing`] semantics: instead of a single
+//! [`TypedKeyRegistry`] extends `KeyRing` semantics: instead of a single
 //! current+previous sequence, it maintains a map of named keys keyed by an ID,
 //! with one designated "current" ID. This is useful when keys are rotated by ID
 //! (e.g. JWT `kid` header) and you need to look up a verification key by ID
@@ -20,7 +20,10 @@ pub struct TypedKeyRegistry<T> {
 impl<T> TypedKeyRegistry<T> {
     /// Creates an empty registry (no current key).
     pub fn new() -> Self {
-        Self { keys: HashMap::new(), current_id: None }
+        Self {
+            keys: HashMap::new(),
+            current_id: None,
+        }
     }
 
     /// Registers a key under `id`, making it the current key.
@@ -37,9 +40,7 @@ impl<T> TypedKeyRegistry<T> {
 
     /// Returns the current key, if any.
     pub fn current(&self) -> Option<&T> {
-        self.current_id
+        self.current_id.as_ref().and_then(|id| self.keys.get(id))
-            .as_ref()
-            .and_then(|id| self.keys.get(id))
     }
 
     /// Returns the ID of the current key.

crates/mytheclipse-crypto/src/lib.rs

@@ -41,8 +41,8 @@
 //! assert_eq!(claims["sub"], "u1");
 //! ```
 
-pub mod key_ring;
 pub mod key_registry;
+pub mod key_ring;
 
 #[cfg(feature = "password")]
 pub mod password;
@@ -53,10 +53,10 @@ pub mod encryption;
 #[cfg(feature = "tokens")]
 pub mod token;
 
-#[cfg(feature = "paseto")]
-pub mod paseto;
 #[cfg(feature = "derivation")]
 pub mod hkdf;
+#[cfg(feature = "paseto")]
+pub mod paseto;
 
 #[cfg(feature = "password")]
 pub use password::PasswordHasher;
@@ -68,10 +68,10 @@ pub use encryption::{AeadError, Encryptor};
 pub use token::{Claims, TokenError, TokenSigner};
 
 #[cfg(feature = "paseto")]
-pub use paseto::{PasetoSigner, PasetoClaims};
+pub use paseto::{PasetoClaims, PasetoSigner};
 
-pub use key_ring::KeyRing;
 pub use key_registry::TypedKeyRegistry;
+pub use key_ring::KeyRing;
 
 #[cfg(feature = "derivation")]
 pub use hkdf::HkdfKeyDeriver;

crates/mytheclipse-crypto/src/paseto.rs

@@ -5,6 +5,8 @@
 
 use std::time::{Duration, SystemTime};
 
+use base64::engine::general_purpose::STANDARD;
+use base64::Engine;
 use serde::{Deserialize, Serialize};
 
 /// Errors returned by PASETO operations.
@@ -58,26 +60,26 @@ impl PasetoClaims {
 ///
 /// This is a stub implementation. For production use with `pasetors` 0.6,
 /// the token format follows the PASETO v4.local specification.
-pub struct PasetoSigner {
+pub struct PasetoSigner {}
-    key: Vec<u8>,
-}
 
 impl PasetoSigner {
     /// Creates a new signer with the given 32-byte key.
     pub fn new(key: &[u8]) -> Result<Self, PasetoError> {
         if key.len() != 32 {
-            return Err(PasetoError::Sign("key must be 32 bytes for v4-local".to_string()));
+            return Err(PasetoError::Sign(
+                "key must be 32 bytes for v4-local".to_string(),
+            ));
         }
-        Ok(Self { key: key.to_vec() })
+        Ok(Self {})
     }
 
     /// Signs claims into a PASETO v4.local token string.
     pub fn sign(&self, claims: &PasetoClaims) -> Result<String, PasetoError> {
-        let payload = serde_json::to_string(claims)
+        let payload =
-            .map_err(|e| PasetoError::Sign(e.to_string()))?;
+            serde_json::to_string(claims).map_err(|e| PasetoError::Sign(e.to_string()))?;
         let nonce = rand::random::<[u8; 24]>();
-        let nonce_b64 = base64::encode(&nonce);
+        let nonce_b64 = STANDARD.encode(nonce);
-        let payload_b64 = base64::encode(payload);
+        let payload_b64 = STANDARD.encode(payload.as_bytes());
         Ok(format!("v4.local.{nonce_b64}.{payload_b64}"))
     }
 
@@ -88,10 +90,11 @@ impl PasetoSigner {
             return Err(PasetoError::InvalidToken);
         }
 
-        let payload_bytes = base64::decode(parts[3])
+        let payload_bytes = STANDARD
-            .map_err(|_| PasetoError::InvalidToken)?;
+            .decode(parts[3])
-        let claims: PasetoClaims = serde_json::from_slice(&payload_bytes)
             .map_err(|_| PasetoError::InvalidToken)?;
+        let claims: PasetoClaims =
+            serde_json::from_slice(&payload_bytes).map_err(|_| PasetoError::InvalidToken)?;
 
         let now = SystemTime::now()
             .duration_since(SystemTime::UNIX_EPOCH)

crates/mytheclipse-event/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "mytheclipse-event"
-version = "1.20.0"
+version = "1.21.2"
 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.20.0"
+version = "1.21.2"
 edition = "2021"
 rust-version = "1.75"
 license = "MIT OR Apache-2.0"

crates/mytheclipse-http/src/lib.rs

@@ -12,7 +12,7 @@
 #[cfg(feature = "resilience")]
 pub mod resilient_client;
 #[cfg(feature = "resilience")]
-pub use resilient_client::{ResilientHttpClient, ResilientClientConfig};
+pub use resilient_client::{ResilientClientConfig, ResilientHttpClient};
 
 #[cfg(feature = "client")]
 pub mod client;

crates/mytheclipse-http/src/metrics_http.rs

@@ -28,11 +28,7 @@ async fn metrics_handler(
         .status(200)
         .header("content-type", "text/plain; version=0.0.4")
         .body(axum::body::Body::from(body))
-        .unwrap_or_else(|_| {
+        .unwrap_or_else(|_| axum::response::Response::new(axum::body::Body::from("internal error")))
-            axum::response::Response::new(axum::body::Body::from(
-                "internal error",
-            ))
-        })
 }
 
 #[cfg(test)]

crates/mytheclipse-http/src/resilient_client.rs

@@ -77,25 +77,22 @@ impl ResilientHttpClient {
 
     /// Sends a pre-built `RequestBuilder` through the resiliency pipeline.
     /// Returns the response bytes on success.
-    pub async fn send(
+    pub async fn send(&self, req: RequestBuilder) -> Result<Vec<u8>, RunError<HttpError>> {
-        &self,
-        req: RequestBuilder,
-    ) -> Result<Vec<u8>, RunError<HttpError>> {
         let span = tracing::info_span!("resilient_http_send");
         let op = move || {
             let req = req.try_clone().unwrap();
             let fut: Pin<Box<dyn std::future::Future<Output = Result<Vec<u8>, HttpError>> + Send>> =
                 Box::pin(async move {
-                    let resp = req.send().instrument(tracing::trace_span!("http_send")).await?;
+                    let resp = req
+                        .send()
+                        .instrument(tracing::trace_span!("http_send"))
+                        .await?;
                     let bytes = resp.bytes().await?;
                     Ok::<Vec<u8>, HttpError>(bytes.to_vec())
                 });
             fut
         };
-        self.builder
+        self.builder.run(op).instrument(span).await
-            .run(op)
-            .instrument(span)
-            .await
     }
 
     /// Convenience: GET `url`, returning response bytes.

crates/mytheclipse-http/src/server/axum_server.rs

@@ -1,9 +1,6 @@
 //! Axum-based HTTP server with health endpoint.
 
-use axum::{
+use axum::{routing::get, Router};
-    routing::get,
-    Router,
-};
 use std::net::SocketAddr;
 
 /// A pre-configured HTTP server with health check and metrics endpoints.
@@ -19,10 +16,7 @@ impl HttpServer {
             .route("/health", get(|| async { "OK" }))
             .route("/", get(|| async { "mytheclipse-http" }));
 
-        Self {
+        Self { app: router, addr }
-            app: router,
-            addr,
-        }
     }
 
     /// Adds a custom route with a GET handler.

crates/mytheclipse-queue/Cargo.toml

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

crates/mytheclipse-queue/src/backpressure_enqueue.rs

@@ -113,7 +113,7 @@ pub async fn enqueue_with_backpressure<Q: Queue + ?Sized>(
 ) -> Result<usize, BackpressureError> {
     let mut rejected = 0;
     for payload in payloads {
-        if let Err(_) = enforcer.try_enqueue(queue, topic, payload).await {
+        if enforcer.try_enqueue(queue, topic, payload).await.is_err() {
             rejected += 1;
         }
     }
@@ -142,6 +142,9 @@ mod tests {
         let _first = reg.global.clone().acquire_owned().await.unwrap();
 
         let result = reg.try_enqueue(&queue, "t", b"x".to_vec()).await;
-        assert!(matches!(result, Err(BackpressureError::LimitReached { .. })));
+        assert!(matches!(
+            result,
+            Err(BackpressureError::LimitReached { .. })
+        ));
     }
 }

crates/mytheclipse-queue/src/batch.rs

@@ -17,7 +17,10 @@ use crate::traits::Queue;
 
 /// A handler that processes a batch of jobs atomically.
 pub trait BatchJobHandler: Send + Sync {
-    fn handle_batch(&self, jobs: Vec<Job>) -> Pin<Box<dyn std::future::Future<Output = Result<(), JobError>> + Send>>;
+    fn handle_batch(
+        &self,
+        jobs: Vec<Job>,
+    ) -> Pin<Box<dyn std::future::Future<Output = Result<(), JobError>> + Send>>;
 }
 
 impl<F, Fut> BatchJobHandler for F
@@ -25,7 +28,10 @@ where
     F: Fn(Vec<Job>) -> Fut + Send + Sync,
     Fut: std::future::Future<Output = Result<(), JobError>> + Send + 'static,
 {
-    fn handle_batch(&self, jobs: Vec<Job>) -> Pin<Box<dyn std::future::Future<Output = Result<(), JobError>> + Send>> {
+    fn handle_batch(
+        &self,
+        jobs: Vec<Job>,
+    ) -> Pin<Box<dyn std::future::Future<Output = Result<(), JobError>> + Send>> {
         Box::pin((self)(jobs))
     }
 }
@@ -85,7 +91,8 @@ impl<Q: Queue + 'static> BatchProcessor<Q> {
         let handler: Arc<dyn BatchJobHandler> = Arc::new(handler);
         let topic_owned = topic.to_string();
 
-        let (tx, mut rx): (mpsc::Sender<Job>, mpsc::Receiver<Job>) = mpsc::channel(config.batch_size);
+        let (tx, mut rx): (mpsc::Sender<Job>, mpsc::Receiver<Job>) =
+            mpsc::channel(config.batch_size);
 
         // Dequeue loop → forward to channel
         {
@@ -147,7 +154,9 @@ impl<Q: Queue + 'static> BatchProcessor<Q> {
                 if !batch.is_empty() {
                     Self::flush(&h, &semaphore, batch).await;
                 }
-                deadline.as_mut().reset(tokio::time::Instant::now() + config.batch_timeout);
+                deadline
+                    .as_mut()
+                    .reset(tokio::time::Instant::now() + config.batch_timeout);
             }
         });
 
@@ -210,7 +219,10 @@ mod tests {
         });
 
         for i in 0..3 {
-            bp.queue.enqueue("t", format!("job{}", i).into_bytes()).await.unwrap();
+            bp.queue
+                .enqueue("t", format!("job{}", i).into_bytes())
+                .await
+                .unwrap();
         }
 
         tokio::time::sleep(Duration::from_millis(300)).await;

crates/mytheclipse-queue/src/in_memory.rs

@@ -18,7 +18,10 @@ struct TopicQueue {
 impl std::fmt::Debug for TopicQueue {
     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
         f.debug_struct("TopicQueue")
-            .field("jobs_len", &self.jobs.try_lock().map(|j| j.len()).unwrap_or(0))
+            .field(
+                "jobs_len",
+                &self.jobs.try_lock().map(|j| j.len()).unwrap_or(0),
+            )
             .finish()
     }
 }
@@ -80,7 +83,10 @@ impl InMemoryQueue {
 impl Queue for InMemoryQueue {
     async fn enqueue(&self, topic: &str, payload: Vec<u8>) -> Result<(), QueueError> {
         let tq = self.get_topic(topic).await;
-        tq.jobs.lock().await.push(Job::new(JobId::generate(), topic, payload));
+        tq.jobs
+            .lock()
+            .await
+            .push(Job::new(JobId::generate(), topic, payload));
         tq.notify.notify_one();
         Ok(())
     }
@@ -140,7 +146,11 @@ mod tests {
     async fn enqueue_dequeue_roundtrip() {
         let q = InMemoryQueue::new();
         q.enqueue("test", b"hello".to_vec()).await.unwrap();
-        let job = q.dequeue("test", Duration::from_millis(500)).await.unwrap().unwrap();
+        let job = q
+            .dequeue("test", Duration::from_millis(500))
+            .await
+            .unwrap()
+            .unwrap();
         assert_eq!(job.payload, b"hello");
         assert_eq!(job.topic, "test");
     }

crates/mytheclipse-queue/src/lib.rs

@@ -11,28 +11,31 @@
 //! - **PostgreSQL** (`postgres`) — `SKIP LOCKED` polling.
 
 pub mod error;
-pub mod job;
 #[cfg(feature = "in-memory")]
 pub mod in_memory;
+pub mod job;
 pub mod traits;
+#[cfg(feature = "in-memory")]
 pub mod worker;
 
-#[cfg(feature = "in-memory")]
-pub mod batch;
 #[cfg(feature = "in-memory")]
 pub mod backpressure_enqueue;
 #[cfg(feature = "in-memory")]
+pub mod batch;
+#[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};
+pub use backpressure_enqueue::{
+    enqueue_with_backpressure, BackpressureEnforcer, BackpressureError,
+};
 #[cfg(feature = "in-memory")]
-pub use rate_limited::{RateLimitedQueue, RateLimitQueueError};
+pub use rate_limited::{RateLimitQueueError, RateLimitedQueue};
 #[cfg(feature = "in-memory")]
 pub use worker_rate_limited::RateLimitedWorkerPool;
 
 #[cfg(feature = "in-memory")]
 pub mod pipeline;
 #[cfg(feature = "in-memory")]
-pub use pipeline::{StageRunner, Stage, StageError};
+pub use pipeline::{Stage, StageError, StageRunner};

crates/mytheclipse-queue/src/pipeline.rs

@@ -75,16 +75,14 @@ where
             let mut input = input;
             loop {
                 match input.recv().await {
-                    Some(item) => {
+                    Some(item) => match stage.process(item).await {
-                        match stage.process(item).await {
                         Ok(out) => {
                             if output.send(out).await.is_err() {
                                 return Err(StageError::ChannelClosed);
                             }
                         }
                         Err(e) => return Err(e),
-                        }
+                    },
-                    }
                     None => return Ok(()),
                 }
             }

crates/mytheclipse-queue/src/rate_limited.rs

@@ -111,7 +111,11 @@ impl<Q: Queue + ?Sized> Queue for RateLimitedQueue<Q> {
         self.inner.enqueue(topic, payload).await
     }
 
-    async fn dequeue(&self, topic: &str, timeout: Duration) -> Result<Option<crate::job::Job>, QueueError> {
+    async fn dequeue(
+        &self,
+        topic: &str,
+        timeout: Duration,
+    ) -> Result<Option<crate::job::Job>, QueueError> {
         self.inner.dequeue(topic, timeout).await
     }
 
@@ -119,7 +123,11 @@ impl<Q: Queue + ?Sized> Queue for RateLimitedQueue<Q> {
         self.inner.ack(job).await
     }
 
-    async fn nack(&self, job: &crate::job::Job, requeue: bool) -> Result<(), crate::error::JobError> {
+    async fn nack(
+        &self,
+        job: &crate::job::Job,
+        requeue: bool,
+    ) -> Result<(), crate::error::JobError> {
         self.inner.nack(job, requeue).await
     }
 

crates/mytheclipse-queue/src/traits.rs

@@ -3,8 +3,8 @@
 use async_trait::async_trait;
 use std::time::Duration;
 
+use crate::error::{JobError, QueueError};
 use crate::job::Job;
-use crate::error::{QueueError, JobError};
 
 /// A handle to a single unit of queued work.
 ///

crates/mytheclipse-queue/src/worker.rs

@@ -71,10 +71,13 @@ pub struct WorkerPool<Q: Queue + 'static> {
 impl<Q: Queue + 'static> WorkerPool<Q> {
     /// Creates a new worker pool with the given concurrency.
     pub fn new(queue: Q, concurrency: usize) -> Self {
-        Self::with_config(queue, WorkerConfig {
+        Self::with_config(
+            queue,
+            WorkerConfig {
                 concurrency,
                 ..Default::default()
-        })
+            },
+        )
     }
 
     /// Creates a new worker pool with explicit configuration.
@@ -148,8 +151,8 @@ impl<Q: Queue + 'static> WorkerPool<Q> {
 /// Computes the (capped) exponential backoff delay.
 pub fn retry_delay(config: &WorkerConfig, attempt: u32) -> Duration {
     let exponent = attempt as f64;
-    let computed = config.retry_base_delay.as_millis() as f64
+    let computed =
-        * config.retry_factor.powf(exponent.max(0.0));
+        config.retry_base_delay.as_millis() as f64 * config.retry_factor.powf(exponent.max(0.0));
     let capped = computed.min(config.retry_max_delay.as_millis() as f64);
     Duration::from_millis(capped as u64)
 }

crates/mytheclipse-queue/src/worker_rate_limited.rs

@@ -6,8 +6,8 @@
 //! service faster than its rate limit allows.
 
 use crate::rate_limited::RateLimitedQueue;
-use crate::worker::{JobHandler, WorkerConfig, WorkerPool};
 use crate::traits::Queue;
+use crate::worker::{JobHandler, WorkerConfig, WorkerPool};
 
 /// A `WorkerPool` whose dequeue is rate-limited via a token bucket.
 pub struct RateLimitedWorkerPool<Q: Queue + 'static> {
@@ -40,12 +40,8 @@ mod tests {
     #[test]
     fn constructs_rate_limited_pool() {
         use crate::in_memory::InMemoryQueue;
-        let _pool = RateLimitedWorkerPool::new(
+        let _pool =
-            InMemoryQueue::new(),
+            RateLimitedWorkerPool::new(InMemoryQueue::new(), WorkerConfig::default(), 10.0, 5);
-            WorkerConfig::default(),
-            10.0,
-            5,
-        );
         // smoke: just verifies construction
     }
 }

crates/mytheclipse-queue/tests/rate_limited_stress.rs

@@ -36,7 +36,11 @@ async fn rate_limited_queue_no_item_loss_under_contention() {
         let s = Arc::clone(&seen);
         workers.push(tokio::spawn(async move {
             loop {
-                match q.dequeue("stress", Duration::from_millis(50)).await.unwrap() {
+                match q
+                    .dequeue("stress", Duration::from_millis(50))
+                    .await
+                    .unwrap()
+                {
                     Some(job) => {
                         let _ = String::from_utf8(job.payload).unwrap();
                         s.fetch_add(1, Ordering::SeqCst);

crates/mytheclipse-storage/Cargo.toml

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

crates/mytheclipse-storage/src/local.rs

@@ -67,9 +67,7 @@ impl StorageDriver for LocalFileStorage {
         let mut file = tokio::fs::File::create(&tmp)
             .await
             .map_err(|e| StorageError::Io(e.to_string()))?;
-        let written = tokio::io::copy(&mut data, &mut file)
+        let written = tokio::io::copy(&mut data, &mut file).await.map_err(|e| {
-            .await
-            .map_err(|e| {
             let _ = std::fs::remove_file(&tmp);
             StorageError::Io(e.to_string())
         })?;

crates/mytheclipse-tracing/Cargo.toml

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

crates/mytheclipse-tracing/src/fmt.rs

@@ -14,9 +14,7 @@ impl TracingLayer {
     pub fn install() {
         let filter = EnvFilter::try_from_default_env()
             .unwrap_or_else(|_| EnvFilter::new("mytheclipse=info"));
-        let _ = tracing_subscriber::fmt()
+        let _ = tracing_subscriber::fmt().with_env_filter(filter).try_init();
-            .with_env_filter(filter)
-            .try_init();
     }
 
     /// Returns a formatted layer for manual composition.

crates/mytheclipse-tracing/src/lib.rs

@@ -3,9 +3,21 @@
 //! Pre-built tracing layers combining all mytheclipse primitives with
 //! optional export backends (OTLP, Jaeger, Zipkin).
 
+#[cfg(any(
+    feature = "env",
+    feature = "otel",
+    feature = "jaeger",
+    feature = "zipkin"
+))]
 pub mod fmt;
 pub mod otel;
 
+#[cfg(any(
+    feature = "env",
+    feature = "otel",
+    feature = "jaeger",
+    feature = "zipkin"
+))]
 pub use fmt::TracingLayer;
 #[cfg(any(feature = "otel", feature = "jaeger", feature = "full"))]
 pub use otel::OtelLayer;

crates/mytheclipse/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "mytheclipse"
-version = "1.20.0"
+version = "1.21.2"
 edition = "2021"
 rust-version = "1.75"
 license = "MIT OR Apache-2.0"
@@ -43,11 +43,21 @@ name = "main"
 path = "examples/main.rs"
 required-features = ["full"]
 
+[[example]]
+name = "high_level"
+path = "examples/high_level.rs"
+required-features = ["full"]
+
 [[example]]
 name = "scaling_demo"
 path = "examples/scaling_demo.rs"
 required-features = ["full"]
 
+[[test]]
+name = "race_stress"
+path = "tests/race_stress.rs"
+required-features = ["full"]
+
 [[bench]]
 name = "primitives"
 path = "benches/primitives.rs"

crates/mytheclipse/benches/primitives.rs

@@ -18,8 +18,8 @@ use mytheclipse::aggregate_error::AggregateError;
 use mytheclipse::parallel_map::{parallel_for_each, parallel_map};
 use mytheclipse::pool::{Pool, SemaphorePool};
 use mytheclipse::ratelimit::RateLimiter;
-use mytheclipse::retry_ext::RetryExt;
 use mytheclipse::retry::RetryConfig;
+use mytheclipse::retry_ext::RetryExt;
 use mytheclipse::shutdown_guard::ShutdownGuard;
 
 fn rt() -> Runtime {
@@ -52,7 +52,11 @@ fn bench_parallel_for_each(c: &mut Criterion) {
     let rt = rt();
     c.bench_function("parallel_for_each/1000x8", |b| {
         b.to_async(&rt).iter(|| async {
-            parallel_for_each(0u32..1000, 8, |_| async move { Ok::<_, std::io::Error>(()) })
+            parallel_for_each(
+                0u32..1000,
+                8,
+                |_| async move { Ok::<_, std::io::Error>(()) },
+            )
             .await
             .unwrap();
             black_box(());

crates/mytheclipse/examples/high_level.rs

@@ -21,8 +21,8 @@ use mytheclipse::{
     pool::{AutoReconnectPool, Pool, Reconnectable, SemaphorePool},
     retry_ext::RetryExt,
     runtime_auto::RuntimeConfig,
-    shutdown_guard::ShutdownGuard,
     service_builder::ServiceConfig,
+    shutdown_guard::ShutdownGuard,
 };
 
 #[tokio::main]
@@ -66,7 +66,10 @@ async fn main() {
         }
     };
     let value = fut.retry(cfg, |_: &String| true, op).await.unwrap();
-    println!("3. RetryExt with {} attempts -> {value}", attempts.load(Ordering::SeqCst));
+    println!(
+        "3. RetryExt with {} attempts -> {value}",
+        attempts.load(Ordering::SeqCst)
+    );
 
     // 4. RAII ShutdownGuard — callback runs exactly once on drop, panic-safe.
     let fired = Arc::new(AtomicU32::new(0));
@@ -93,7 +96,8 @@ async fn main() {
     let svc = AutoMetricsServiceBuilder::new("demo_op", svc_cfg);
     let n = Arc::new(AtomicU32::new(0));
     let n2 = Arc::clone(&n);
-    let _: Result<u32, mytheclipse::service_builder::RunError<()>> = svc.run(|| {
+    let _: Result<u32, mytheclipse::service_builder::RunError<()>> = svc
+        .run(|| {
             let n2 = Arc::clone(&n2);
             Box::pin(async move {
                 tokio::time::sleep(Duration::from_millis(5)).await;
@@ -104,7 +108,8 @@ async fn main() {
                     Ok(7u32)
                 }
             })
-    }).await;
+        })
+        .await;
     let snap = svc.collector().snapshot();
     println!(
         "6. AutoMetrics -> {} counters, {} histograms",

crates/mytheclipse/examples/scaling_demo.rs

@@ -16,7 +16,10 @@ use mytheclipse::parallel_map::{parallel_for_each, ParallelConcurrency};
 #[tokio::main]
 async fn main() {
     let total = 100u32;
-    println!("host available_parallelism = {}", <() as ParallelConcurrency>::resolve(()));
+    println!(
+        "host available_parallelism = {}",
+        <() as ParallelConcurrency>::resolve(())
+    );
     println!("total items = {total}");
     println!();
 
@@ -49,6 +52,12 @@ async fn run(label: &str, concurrency: impl ParallelConcurrency + Copy, total: u
     let elapsed = start.elapsed();
 
     println!("{label}  resolved={resolved}");
-    println!("  peak in-flight = {} (bounded, never {total})", peak.load(Ordering::SeqCst));
+    println!(
-    println!("  elapsed        = {elapsed:?}  (sequential ~{}ms)", total * 1);
+        "  peak in-flight = {} (bounded, never {total})",
+        peak.load(Ordering::SeqCst)
+    );
+    println!(
+        "  elapsed        = {elapsed:?}  (sequential ~{}ms)",
+        total * 1
+    );
 }

crates/mytheclipse/src/aggregate_error.rs

@@ -117,12 +117,18 @@ impl std::error::Error for AggregateError {}
 
 impl From<Vec<Box<dyn std::error::Error + Send + Sync>>> for AggregateError {
     fn from(errors: Vec<Box<dyn std::error::Error + Send + Sync>>) -> Self {
-        Self { errors, context: None }
+        Self {
+            errors,
+            context: None,
+        }
     }
 }
 
 impl Extend<Box<dyn std::error::Error + Send + Sync>> for AggregateError {
-    fn extend<T: IntoIterator<Item = Box<dyn std::error::Error + Send + Sync>>>(&mut self, iter: T) {
+    fn extend<T: IntoIterator<Item = Box<dyn std::error::Error + Send + Sync>>>(
+        &mut self,
+        iter: T,
+    ) {
         self.errors.extend(iter);
     }
 }
@@ -160,8 +166,7 @@ mod tests {
 
     #[test]
     fn collects_values_when_all_ok() {
-        let results: Vec<Result<u32, std::io::Error>> =
+        let results: Vec<Result<u32, std::io::Error>> = vec![Ok(1), Ok(2), Ok(3)];
-            vec![Ok(1), Ok(2), Ok(3)];
         let out = AggregateError::from_results(results).unwrap();
         assert_eq!(out, vec![1, 2, 3]);
     }

crates/mytheclipse/src/auto_metrics_service.rs

@@ -74,8 +74,8 @@ impl AutoMetricsServiceBuilder {
         self
     }
 
-    /// Attaches a [`MetricsBridge`] to forward snapshots downstream (requires
+    /// Attaches a [`crate::metrics_bridge::MetricsBridge`] to forward snapshots
-    /// the `resiliency` feature which pulls in the bridge).
+    /// 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);
@@ -108,8 +108,13 @@ impl AutoMetricsServiceBuilder {
             Err(_) => "other",
         };
 
-        self.metrics
+        self.metrics.inc_counter(
-            .inc_counter(&format!("mytheclipse_service_calls_total{{service=\"{}\",outcome=\"{}\"}}", self.service_name, outcome), 1);
+            &format!(
+                "mytheclipse_service_calls_total{{service=\"{}\",outcome=\"{}\"}}",
+                self.service_name, outcome
+            ),
+            1,
+        );
         self.metrics
             .observe("mytheclipse_service_duration_seconds", dur);
 
@@ -125,8 +130,8 @@ impl AutoMetricsServiceBuilder {
 #[cfg(test)]
 mod tests {
     use super::*;
+    use std::sync::atomic::{AtomicU32, Ordering};
     use std::sync::Arc;
-    use std::sync::atomic::{Ordering, AtomicU32};
 
     #[tokio::test]
     async fn auto_metrics_records_call() {
@@ -136,13 +141,19 @@ mod tests {
 
         let attempts = Arc::new(AtomicU32::new(0));
         let a = Arc::clone(&attempts);
-        let result: Result<u32, RunError<()>> = builder.run(|| {
+        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) }
+                    if n < 2 {
+                        Err(())
+                    } else {
+                        Ok(42u32)
+                    }
                 })
-        }).await;
+            })
+            .await;
         assert_eq!(result.unwrap(), 42);
         assert_eq!(attempts.load(Ordering::SeqCst), 3);
         let snap = builder.collector().snapshot();

crates/mytheclipse/src/bg_join.rs

@@ -8,7 +8,6 @@
 //! *concurrency*, `BgJoiner` adds structured *lifetimes* so a service can wait
 //! for all in-flight work to settle before terminating.
 
-use std::future::Future;
 use std::sync::Arc;
 use std::time::Duration;
 
@@ -36,7 +35,9 @@ impl BgJoiner {
         F: std::future::Future + Send + 'static,
         F::Output: Send + 'static,
     {
-        let handle: JoinHandle<()> = tokio::spawn(async move { let _ = future.await; });
+        let handle: JoinHandle<()> = tokio::spawn(async move {
+            let _ = future.await;
+        });
         self.track(handle);
     }
 
@@ -55,6 +56,11 @@ impl BgJoiner {
         self.inner.lock().await.len()
     }
 
+    /// Returns `true` if there are no currently-tracked tasks.
+    pub async fn is_empty(&self) -> bool {
+        self.inner.lock().await.is_empty()
+    }
+
     /// Await every tracked task, dropping any that are still pending once
     /// `deadline` elapses. Returns the count of tasks that had not completed
     /// within the timeout.
@@ -66,7 +72,6 @@ impl BgJoiner {
         };
 
         let mut pending: Vec<JoinHandle<()>> = handles;
-        let mut dropped = 0usize;
 
         loop {
             if pending.is_empty() {
@@ -74,11 +79,11 @@ impl BgJoiner {
             }
 
             if now.elapsed() >= deadline {
-                dropped = pending.len();
+                let count = pending.len();
                 for h in pending.drain(..) {
                     h.abort();
                 }
-                return dropped;
+                return count;
             }
 
             let remaining = deadline.saturating_sub(now.elapsed());

crates/mytheclipse/src/compute.rs

@@ -2,6 +2,8 @@
 
 use std::panic::{catch_unwind, AssertUnwindSafe};
 
+use rayon::iter::{IntoParallelIterator, ParallelIterator};
+
 use crate::context::context;
 use crate::error::MytheclipseError;
 
@@ -45,6 +47,189 @@ fn panic_payload_to_string(payload: Box<dyn std::any::Any + Send>) -> String {
     }
 }
 
+/// Errors collected while mapping over items on the CPU compute pool.
+///
+/// Each entry is one item's error message (from an `Err` return or a panic).
+/// Unlike [`crate::aggregate_error::AggregateError`], this lives under the
+/// `compute` feature only and carries plain strings, so the compute pool
+/// primitives compile without the `resiliency` feature.
+#[derive(Debug, Default)]
+pub struct ComputeErrors {
+    errors: Vec<String>,
+}
+
+impl ComputeErrors {
+    /// Number of items that failed (returned `Err` or panicked).
+    pub fn len(&self) -> usize {
+        self.errors.len()
+    }
+
+    /// `true` when every item succeeded.
+    pub fn is_empty(&self) -> bool {
+        self.errors.is_empty()
+    }
+
+    /// Iterate over the error messages.
+    pub fn iter(&self) -> impl Iterator<Item = &str> {
+        self.errors.iter().map(String::as_str)
+    }
+}
+
+impl std::fmt::Display for ComputeErrors {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        write!(f, "{} compute item(s) failed:", self.errors.len())?;
+        for e in &self.errors {
+            write!(f, "\n  - {e}")?;
+        }
+        Ok(())
+    }
+}
+
+impl std::error::Error for ComputeErrors {}
+
+/// Runs `f` over every item **on the rayon compute pool**, in parallel,
+/// returning the results **in input order**.
+///
+/// This is the CPU-specific counterpart to [`crate::parallel_map::parallel_map`]:
+/// work is distributed across the compute pool's worker threads (sized from
+/// [`crate::runtime_auto::RuntimeConfig::auto`] — one per logical core) by
+/// rayon's work-stealing scheduler, so concurrency is bounded by the pool and
+/// **auto-scales to the host CPU** without a manual `concurrency` parameter.
+///
+/// Each item's closure runs inside [`std::panic::catch_unwind`]: a panic is
+/// caught and recorded as an error instead of unwinding across the pool. All
+/// errors (returned or panicked) are collected into a single
+/// [`ComputeErrors`].
+///
+/// ```
+/// use mytheclipse::compute::compute_map;
+///
+/// let squares = compute_map(vec![1u32, 2, 3, 4], |x| Ok::<_, String>(x * x)).unwrap();
+/// assert_eq!(squares, vec![1, 4, 9, 16]);
+///
+/// // Failures are aggregated, order is preserved:
+/// let out = compute_map(vec![1, 2, 3], |x| {
+///     if x == 2 { Err("boom".to_string()) } else { Ok(x * 10) }
+/// }).unwrap_err();
+/// assert_eq!(out.len(), 1);
+/// ```
+pub fn compute_map<I, T, F>(items: I, f: F) -> Result<Vec<T>, ComputeErrors>
+where
+    I: IntoParallelIterator + Send,
+    I::Item: Send,
+    T: Send,
+    F: Fn(I::Item) -> Result<T, String> + Send + Sync,
+{
+    let wrapped = AssertUnwindSafe(f);
+    let collected: Vec<Result<T, String>> = context().compute_pool.install(move || {
+        let f = wrapped;
+        items
+            .into_par_iter()
+            .map(|item| {
+                catch_unwind(AssertUnwindSafe(|| f(item)))
+                    .unwrap_or_else(|payload| Err(panic_payload_to_string(payload)))
+            })
+            .collect()
+    });
+
+    let mut values = Vec::with_capacity(collected.len());
+    let mut errors = Vec::new();
+    for r in collected {
+        match r {
+            Ok(v) => values.push(v),
+            Err(e) => errors.push(e),
+        }
+    }
+    if errors.is_empty() {
+        Ok(values)
+    } else {
+        Err(ComputeErrors { errors })
+    }
+}
+
+/// Runs two heavy closures in parallel on the compute pool, returning both
+/// results.
+///
+/// This wraps [`rayon::join`] with panic isolation: if either branch panics,
+/// its panic is converted into a [`MytheclipseError::ComputePanic`] and the
+/// other branch still completes. The pool remains usable afterward.
+///
+/// ```
+/// use mytheclipse::compute::compute_join;
+///
+/// let (a, b) = compute_join(
+///     || (0..1_000_000u64).sum::<u64>(),
+///     || (1_000_000..2_000_000u64).sum::<u64>(),
+/// ).unwrap();
+/// assert_eq!(a + b, (0..2_000_000u64).sum::<u64>());
+/// ```
+pub fn compute_join<A, RA, B, RB>(a: A, b: B) -> Result<(RA, RB), MytheclipseError>
+where
+    A: FnOnce() -> RA + Send,
+    RA: Send,
+    B: FnOnce() -> RB + Send,
+    RB: Send,
+{
+    let a = AssertUnwindSafe(a);
+    let b = AssertUnwindSafe(b);
+    context().compute_pool.install(|| {
+        let (ra, rb) = rayon::join(
+            move || {
+                catch_unwind(a)
+                    .map_err(|p| MytheclipseError::ComputePanic(panic_payload_to_string(p)))
+            },
+            move || {
+                catch_unwind(b)
+                    .map_err(|p| MytheclipseError::ComputePanic(panic_payload_to_string(p)))
+            },
+        );
+        Ok((ra?, rb?))
+    })
+}
+
+/// Runs `f` over every item on the compute pool in parallel, discarding
+/// return values (side effects only), collecting errors and panics.
+///
+/// ```
+/// use std::sync::atomic::{AtomicUsize, Ordering};
+/// use std::sync::Arc;
+/// use mytheclipse::compute::compute_par_for_each;
+///
+/// let count = Arc::new(AtomicUsize::new(0));
+/// let c = Arc::clone(&count);
+/// compute_par_for_each(0..100, move |x| {
+///     c.fetch_add(x as usize, Ordering::SeqCst);
+///     Ok::<_, String>(())
+/// }).unwrap();
+/// assert_eq!(count.load(Ordering::SeqCst), 4950);
+/// ```
+pub fn compute_par_for_each<I, F>(items: I, f: F) -> Result<(), ComputeErrors>
+where
+    I: IntoParallelIterator + Send,
+    I::Item: Send,
+    F: Fn(I::Item) -> Result<(), String> + Send + Sync,
+{
+    let wrapped = AssertUnwindSafe(f);
+    let collected: Vec<Result<(), String>> = context().compute_pool.install(move || {
+        let f = wrapped;
+        items
+            .into_par_iter()
+            .map(|item| {
+                catch_unwind(AssertUnwindSafe(|| f(item)))
+                    .unwrap_or_else(|payload| Err(panic_payload_to_string(payload)))
+            })
+            .collect()
+    });
+
+    if collected.iter().any(|r| r.is_err()) {
+        Err(ComputeErrors {
+            errors: collected.into_iter().filter_map(|r| r.err()).collect(),
+        })
+    } else {
+        Ok(())
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
@@ -57,4 +242,74 @@ mod tests {
         let recovered = compute(|| 1 + 1);
         assert_eq!(recovered.unwrap(), 2);
     }
+
+    #[test]
+    fn compute_map_ordered_and_aggregates_errors() {
+        let squares = compute_map(vec![1u32, 2, 3, 4], |x| Ok::<_, String>(x * x)).unwrap();
+        assert_eq!(squares, vec![1, 4, 9, 16]);
+
+        let err = compute_map(vec![1u32, 2, 3], |x| {
+            if x == 2 {
+                Err("boom".to_string())
+            } else {
+                Ok(x * 10)
+            }
+        })
+        .unwrap_err();
+        assert_eq!(err.len(), 1);
+        assert_eq!(err.iter().next().unwrap(), "boom");
+    }
+
+    #[test]
+    fn compute_map_panic_is_isolated_and_collected() {
+        let err = compute_map(vec![1u32, 2, 3], |x| {
+            if x == 2 {
+                panic!("item panic")
+            } else {
+                Ok::<_, String>(x)
+            }
+        })
+        .unwrap_err();
+        assert_eq!(err.len(), 1);
+        assert!(err.iter().next().unwrap().contains("item panic"));
+
+        // pool still usable
+        let recovered = compute_map(vec![1u32], |x| Ok::<_, String>(x + 1)).unwrap();
+        assert_eq!(recovered, vec![2]);
+    }
+
+    #[test]
+    fn compute_join_runs_both_branches() {
+        let (a, b) = compute_join(
+            || (0..100_000u64).sum::<u64>(),
+            || (100_000..200_000u64).sum::<u64>(),
+        )
+        .unwrap();
+        assert_eq!(a + b, (0..200_000u64).sum::<u64>());
+    }
+
+    #[test]
+    fn compute_join_panic_is_isolated() {
+        let a = compute_join(|| panic!("branch a"), || 42u32);
+        assert!(matches!(a, Err(MytheclipseError::ComputePanic(_))));
+
+        // pool still usable
+        let ok = compute_join(|| 1u32, || 2u32).unwrap();
+        assert_eq!(ok, (1, 2));
+    }
+
+    #[test]
+    fn compute_par_for_each_runs_all_side_effects() {
+        use std::sync::atomic::{AtomicUsize, Ordering};
+        use std::sync::Arc;
+
+        let count = Arc::new(AtomicUsize::new(0));
+        let c = Arc::clone(&count);
+        compute_par_for_each(0..100, move |x| {
+            c.fetch_add(x as usize, Ordering::SeqCst);
+            Ok::<_, String>(())
+        })
+        .unwrap();
+        assert_eq!(count.load(Ordering::SeqCst), 4950);
+    }
 }

crates/mytheclipse/src/dlock.rs

@@ -69,7 +69,12 @@ impl Drop for LockGuard {
 #[async_trait]
 pub trait DistributedLock: Send + Sync {
     /// Attempts to acquire the lock with the given lease duration.
-    async fn acquire(&self, key: &str, lease: Duration, timeout: Duration) -> Result<LockGuard, LockError>;
+    async fn acquire(
+        &self,
+        key: &str,
+        lease: Duration,
+        timeout: Duration,
+    ) -> Result<LockGuard, LockError>;
 
     /// Releases the lock.
     async fn release(&self, key: &str) -> Result<(), LockError>;
@@ -90,14 +95,6 @@ impl InProcLock {
             held: Arc::new(Mutex::new(std::collections::HashMap::new())),
         }
     }
-
-    fn is_expired(map: &std::collections::HashMap<String, Instant>, key: &str) -> bool {
-        if let Some(expiry) = map.get(key) {
-            *expiry <= Instant::now()
-        } else {
-            false
-        }
-    }
 }
 
 impl Default for InProcLock {
@@ -108,7 +105,12 @@ impl Default for InProcLock {
 
 #[async_trait]
 impl DistributedLock for InProcLock {
-    async fn acquire(&self, key: &str, lease: Duration, timeout: Duration) -> Result<LockGuard, LockError> {
+    async fn acquire(
+        &self,
+        key: &str,
+        lease: Duration,
+        timeout: Duration,
+    ) -> Result<LockGuard, LockError> {
         let deadline = Instant::now() + timeout;
         loop {
             {
@@ -154,7 +156,10 @@ mod tests {
     #[tokio::test]
     async fn lock_acquire_release() {
         let lock = InProcLock::new();
-        let guard = lock.acquire("key", Duration::from_secs(10), Duration::from_secs(1)).await.unwrap();
+        let guard = lock
+            .acquire("key", Duration::from_secs(10), Duration::from_secs(1))
+            .await
+            .unwrap();
         assert!(lock.release("key").await.is_ok());
         drop(guard);
     }
@@ -162,9 +167,14 @@ mod tests {
     #[tokio::test]
     async fn lock_rejects_second_acquire() {
         let lock = InProcLock::new();
-        let _guard1 = lock.acquire("key", Duration::from_secs(10), Duration::from_secs(1)).await.unwrap();
+        let _guard1 = lock
+            .acquire("key", Duration::from_secs(10), Duration::from_secs(1))
+            .await
+            .unwrap();
         // While guard1 is alive, a second acquire with short timeout should fail.
-        let result = lock.acquire("key", Duration::from_secs(10), Duration::from_millis(50)).await;
+        let result = lock
+            .acquire("key", Duration::from_secs(10), Duration::from_millis(50))
+            .await;
         assert!(result.is_err());
         drop(_guard1);
     }
@@ -172,21 +182,31 @@ mod tests {
     #[tokio::test]
     async fn lock_auto_releases_on_drop() {
         let lock = InProcLock::new();
-        let guard = lock.acquire("k", Duration::from_secs(10), Duration::from_secs(1)).await.unwrap();
+        let guard = lock
+            .acquire("k", Duration::from_secs(10), Duration::from_secs(1))
+            .await
+            .unwrap();
         drop(guard);
         // After drop, the lock should be releasable / re-acquirable.
-        let result = lock.acquire("k", Duration::from_secs(10), Duration::from_millis(50)).await;
+        let result = lock
+            .acquire("k", Duration::from_secs(10), Duration::from_millis(50))
+            .await;
         assert!(result.is_ok(), "lock should be free after guard drop");
     }
 
     #[tokio::test]
     async fn lock_expires_after_lease() {
         let lock = InProcLock::new();
-        let _guard = lock.acquire("key", Duration::from_millis(20), Duration::from_millis(5)).await.unwrap();
+        let _guard = lock
+            .acquire("key", Duration::from_millis(20), Duration::from_millis(5))
+            .await
+            .unwrap();
         drop(_guard);
         tokio::time::sleep(Duration::from_millis(30)).await;
         // Should be acquirable now.
-        let result = lock.acquire("key", Duration::from_millis(20), Duration::from_millis(5)).await;
+        let result = lock
+            .acquire("key", Duration::from_millis(20), Duration::from_millis(5))
+            .await;
         assert!(result.is_ok());
     }
 }

crates/mytheclipse/src/health.rs

@@ -26,7 +26,9 @@ impl fmt::Display for HealthStatus {
 /// A single health check.
 pub trait HealthCheck: Send + Sync {
     fn name(&self) -> &str;
-    fn check(&self) -> std::pin::Pin<Box<dyn std::future::Future<Output = HealthStatus> + Send + '_>>;
+    fn check(
+        &self,
+    ) -> std::pin::Pin<Box<dyn std::future::Future<Output = HealthStatus> + Send + '_>>;
 }
 
 /// A registered health check with its name and trait object.

crates/mytheclipse/src/lib.rs

@@ -12,7 +12,7 @@
 //! - [`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`] / [`CircuitBreaker`] / [`timeout()`] (feature `resiliency`) — fault tolerance.
+//! - [`retry()`] / [`CircuitBreaker`] / [`timeout()`] (feature `resiliency`) — fault tolerance.
 //! - [`RateLimiter`] / [`BackpressureQueue`] / [`ConcurrencyLimiter`] (feature `traffic`) — load control.
 //! - [`SemaphorePool`] (feature `traffic`) — shared bounded resource pool.
 //! - [`ShutdownManager`] / [`CronSchedule`] (feature `lifecycle`) — lifecycle + scheduling.
@@ -24,57 +24,59 @@
 pub mod context;
 pub mod error;
 
-#[cfg(feature = "io")]
-pub mod io;
-#[cfg(feature = "compute")]
-pub mod compute;
 #[cfg(feature = "bg")]
 pub mod bg;
+#[cfg(feature = "compute")]
+pub mod compute;
+#[cfg(feature = "io")]
+pub mod io;
 
-#[cfg(feature = "resiliency")]
-pub mod retry;
-#[cfg(feature = "resiliency")]
-pub mod retry_ext;
 #[cfg(feature = "resiliency")]
 pub mod aggregate_error;
 #[cfg(feature = "resiliency")]
 pub mod parallel_map;
 #[cfg(feature = "resiliency")]
-pub use retry_ext::RetryExt;
+pub mod retry;
+#[cfg(feature = "resiliency")]
+pub mod retry_ext;
 #[cfg(feature = "resiliency")]
 pub use aggregate_error::AggregateError;
 #[cfg(feature = "resiliency")]
-pub use parallel_map::{parallel_map, parallel_map_unordered, parallel_for_each, ParallelConcurrency};
+pub use parallel_map::{
-#[cfg(feature = "observability")]
+    parallel_for_each, parallel_map, parallel_map_unordered, ParallelConcurrency,
+};
+#[cfg(feature = "resiliency")]
+pub use retry_ext::RetryExt;
+#[cfg(all(feature = "observability", feature = "resiliency"))]
 pub mod auto_metrics_service;
-#[cfg(feature = "observability")]
+#[cfg(all(feature = "observability", feature = "resiliency"))]
 pub use auto_metrics_service::AutoMetricsServiceBuilder;
 #[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 = "traffic")]
 pub mod pool;
+#[cfg(feature = "traffic")]
+pub mod ratelimit;
 
-#[cfg(feature = "lifecycle")]
-pub mod shutdown;
 #[cfg(feature = "lifecycle")]
 pub mod cron;
 #[cfg(feature = "lifecycle")]
 pub mod health;
-#[cfg(all(feature = "observability", feature = "traffic"))]
-pub mod pool_health;
 #[cfg(feature = "lifecycle")]
 pub mod leader;
 #[cfg(feature = "lifecycle")]
 pub mod lifecycle;
+#[cfg(all(feature = "observability", feature = "traffic"))]
+pub mod pool_health;
+#[cfg(feature = "lifecycle")]
+pub mod shutdown;
 
 #[cfg(feature = "lifecycle")]
 pub mod bg_join;
@@ -94,59 +96,59 @@ pub mod middleware;
 #[cfg(feature = "observability")]
 pub mod metrics;
 #[cfg(feature = "observability")]
-pub mod panic_tracker;
-#[cfg(feature = "observability")]
 pub mod metrics_bridge;
+#[cfg(feature = "observability")]
+pub mod panic_tracker;
 
-#[cfg(feature = "resiliency")]
-pub mod service_builder;
 #[cfg(feature = "lifecycle")]
 pub mod dlock;
+#[cfg(feature = "resiliency")]
+pub mod service_builder;
 
 pub use context::{context, EngineContext};
 pub use error::MytheclipseError;
 
-#[cfg(feature = "io")]
-pub use io::spawn_io;
-#[cfg(feature = "compute")]
-pub use compute::compute;
 #[cfg(feature = "bg")]
 pub use bg::spawn_bg;
+#[cfg(feature = "compute")]
+pub use compute::{compute, compute_join, compute_map, compute_par_for_each, ComputeErrors};
+#[cfg(feature = "io")]
+pub use io::spawn_io;
 
-#[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 retry::{retry, JitterKind, RetryConfig, RetryError};
+#[cfg(feature = "resiliency")]
 pub use timeout::{timeout, with_timeout, Timeout, TimeoutError};
 
-#[cfg(feature = "traffic")]
+/// Re-export of `async-trait` so implementing [`pool::Reconnectable`] (and
-pub use ratelimit::{RateLimitError, RateLimiter};
+/// other async traits) doesn't require users to add their own `async-trait`
+/// dependency.
+pub use async_trait::async_trait;
 #[cfg(feature = "traffic")]
 pub use backpressure::{BackpressureError, BackpressureQueue, OverflowPolicy};
 #[cfg(feature = "traffic")]
 pub use concurrency::{ConcurrencyLimiter, ConcurrencyPermit};
 #[cfg(feature = "traffic")]
-pub use pool::{Pool, PoolError, Pooled, SemaphorePool, AutoReconnectPool, Reconnectable};
+pub use pool::{AutoReconnectPool, Pool, PoolError, Pooled, Reconnectable, SemaphorePool};
-/// Re-export of `async-trait` so implementing [`pool::Reconnectable`] (and
+#[cfg(feature = "traffic")]
-/// other async traits) doesn't require users to add their own `async-trait`
+pub use ratelimit::{RateLimitError, RateLimiter};
-/// dependency.
-pub use async_trait::async_trait;
 
-#[cfg(feature = "lifecycle")]
-pub use shutdown::{ShutdownManager, ShutdownSignal};
 #[cfg(feature = "lifecycle")]
 pub use cron::{schedule, CronError, CronJob, CronParseError, CronSchedule};
 #[cfg(feature = "lifecycle")]
 pub use health::{HealthCheck, HealthRegistry, HealthStatus};
 #[cfg(feature = "lifecycle")]
 pub use leader::{InProcLeaderElection, LeaderElection};
+#[cfg(feature = "lifecycle")]
+pub use shutdown::{ShutdownManager, ShutdownSignal};
 
 #[cfg(feature = "resiliency")]
 pub use service_builder::{RunError, ServiceBuilder, ServiceConfig};
 
 #[cfg(feature = "lifecycle")]
-pub use dlock::{DistributedLock, LockError, LockGuard, InProcLock};
+pub use dlock::{DistributedLock, InProcLock, LockError, LockGuard};
 #[cfg(feature = "lifecycle")]
 pub use lifecycle::AsyncLifecycleManager;
 
@@ -169,7 +171,7 @@ pub use pool_health::HealthCheckedPool;
 pub use bg_join::BgJoiner;
 
 #[cfg(all(feature = "observability", feature = "resiliency"))]
-pub use middleware::{MiddlewarePipeline, PipelineError, BoxMiddleware, mw};
+pub use middleware::{mw, BoxMiddleware, MiddlewarePipeline, PipelineError};
 #[cfg(feature = "observability")]
 pub use panic_tracker::{PanicGuard, PanicInfo, PanicTracker};