fix(ci): restore full CI green — test-matrix, clippy, rustfmt, and rustdoc gates

Root cause of the failing CI run was that examples/tests referencing
feature-gated items were auto-detected (no required-features), so
`--all-targets` compiled them under feature combinations where those
modules didn't exist. Fixes:

- mytheclipse Cargo.toml: declare the `high_level` example and
  `race_stress` integration test with required-features = ["full"];
  `cargo build --all-targets` now skips them when full is off. This
  clears the whole test-matrix (workspace default, all-features, and
  every single-feature config) which all failed on E0432/E0433.
- lib.rs: auto_metrics_service depends on service_builder, so regate it
  behind all(observability, resiliency) instead of observability alone
  (observability-only build compiled the module without resiliency).
- mytheclipse-tracing: gate `pub mod fmt` behind any
  tracing-subscriber-providing feature so --no-default-features compiles.
- mytheclipse-queue: gate `pub mod worker` behind in-memory (worker.rs
  requires tokio, only provided by in-memory).
- clippy -D warnings fixes: deprecated base64 0.22 free fns -> Engine
  (paseto), unused key field, needless mut (service_builder), unused
  import/dead var/missing is_empty (bg_join), dead is_expired (dlock),
  while-let-iterator->for (parallel_map), type_complexity (shutdown_guard),
  MutexGuard held across await (middleware, now clones Arc'd layers),
  if-let-Err->is_err (queue), unused CliBuilder fields now wired into clap.
- rustdoc -D warnings: resolve retry/MetricsBridge/CircuitBreaker/KeyRing
  intra-doc links and fix the unparseable lifecycle.rs code fence.
- cargo fmt --all to satisfy the Rustfmt gate.

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/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/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/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-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/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/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/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

@@ -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

@@ -121,9 +121,7 @@ where
     F: Fn(I::Item) -> Result<T, String> + Send + Sync,
 {
     let wrapped = AssertUnwindSafe(f);
-    let collected: Vec<Result<T, String>> = context()
+    let collected: Vec<Result<T, String>> = context().compute_pool.install(move || {
-        .compute_pool
-        .install(move || {
         let f = wrapped;
         items
             .into_par_iter()
@@ -165,10 +163,7 @@ where
 /// ).unwrap();
 /// assert_eq!(a + b, (0..2_000_000u64).sum::<u64>());
 /// ```
-pub fn compute_join<A, RA, B, RB>(
+pub fn compute_join<A, RA, B, RB>(a: A, b: B) -> Result<(RA, RB), MytheclipseError>
-    a: A,
-    b: B,
-) -> Result<(RA, RB), MytheclipseError>
 where
     A: FnOnce() -> RA + Send,
     RA: Send,
@@ -177,12 +172,16 @@ where
 {
     let a = AssertUnwindSafe(a);
     let b = AssertUnwindSafe(b);
-    context()
+    context().compute_pool.install(|| {
-        .compute_pool
-        .install(|| {
         let (ra, rb) = rayon::join(
-                move || catch_unwind(a).map_err(|p| MytheclipseError::ComputePanic(panic_payload_to_string(p))),
+            move || {
-                move || catch_unwind(b).map_err(|p| MytheclipseError::ComputePanic(panic_payload_to_string(p))),
+                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?))
     })
@@ -211,9 +210,7 @@ where
     F: Fn(I::Item) -> Result<(), String> + Send + Sync,
 {
     let wrapped = AssertUnwindSafe(f);
-    let collected: Vec<Result<(), String>> = context()
+    let collected: Vec<Result<(), String>> = context().compute_pool.install(move || {
-        .compute_pool
-        .install(move || {
         let f = wrapped;
         items
             .into_par_iter()
@@ -226,10 +223,7 @@ where
 
     if collected.iter().any(|r| r.is_err()) {
         Err(ComputeErrors {
-            errors: collected
+            errors: collected.into_iter().filter_map(|r| r.err()).collect(),
-                .into_iter()
-                .filter_map(|r| r.err())
-                .collect(),
         })
     } else {
         Ok(())

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, compute_join, compute_map, compute_par_for_each, ComputeErrors};
 #[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};
 

crates/mytheclipse/src/lifecycle.rs

@@ -15,9 +15,7 @@ use crate::shutdown::ShutdownManager;
 ///
 /// Typical usage:
 /// ```ignore
-/// # tokio::runtime::Runtime::new().unwrap().block_on(async {
+/// let mgr = mytheclipse::AsyncLifecycleManager::new();
-/// # use mytheclipse::AsyncLifecycleManager;
-/// let mgr = AsyncLifecycleManager::new();
 /// mgr.register_health_check("db", my_db_check());
 /// let handle = mgr.start_health_loop(std::time::Duration::from_secs(30));
 /// mgr.await_shutdown(std::time::Duration::from_secs(10)).await;
@@ -47,7 +45,11 @@ impl AsyncLifecycleManager {
     }
 
     /// Registers a named health check.
-    pub async fn register_health_check(&self, name: impl Into<String>, check: impl HealthCheck + 'static) {
+    pub async fn register_health_check(
+        &self,
+        name: impl Into<String>,
+        check: impl HealthCheck + 'static,
+    ) {
         self.health.register(name, check).await;
     }
 
@@ -124,16 +126,26 @@ mod tests {
 
     struct AlwaysOk;
     impl HealthCheck for AlwaysOk {
-        fn name(&self) -> &str { "always-ok" }
+        fn name(&self) -> &str {
-        fn check(&self) -> std::pin::Pin<Box<dyn std::future::Future<Output = HealthStatus> + Send + '_>> {
+            "always-ok"
+        }
+        fn check(
+            &self,
+        ) -> std::pin::Pin<Box<dyn std::future::Future<Output = HealthStatus> + Send + '_>>
+        {
             Box::pin(async { HealthStatus::Ok })
         }
     }
 
     struct AlwaysBad;
     impl HealthCheck for AlwaysBad {
-        fn name(&self) -> &str { "always-bad" }
+        fn name(&self) -> &str {
-        fn check(&self) -> std::pin::Pin<Box<dyn std::future::Future<Output = HealthStatus> + Send + '_>> {
+            "always-bad"
+        }
+        fn check(
+            &self,
+        ) -> std::pin::Pin<Box<dyn std::future::Future<Output = HealthStatus> + Send + '_>>
+        {
             Box::pin(async { HealthStatus::Unhealthy })
         }
     }

crates/mytheclipse/src/metrics_bridge.rs

@@ -10,8 +10,8 @@ use std::time::Duration;
 use crate::health::{HealthCheck, HealthStatus};
 use crate::metrics::MetricsCollector;
 
-/// A health check backed by a [`CircuitBreaker`]: unhealthy if open,
+/// A health check backed by a [`crate::circuit_breaker::CircuitBreaker`]:
-/// degraded if half-open, ok otherwise.
+/// unhealthy if open, degraded if half-open, ok otherwise.
 ///
 /// Only available when both `resiliency` and `observability` features are
 /// enabled (circuit breaker + health/metrics bridge).
@@ -33,7 +33,9 @@ impl HealthCheck for CircuitBreakerHealthCheck {
         "circuit_breaker"
     }
 
-    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 + '_>> {
         let state = self.breaker.snapshot().state;
         Box::pin(async move {
             match state {
@@ -63,11 +65,7 @@ impl MetricsHealthCheck {
     }
 
     fn has_errors(&self) -> bool {
-        self.collector
+        self.collector.snapshot().counters.values().any(|&v| v > 0)
-            .snapshot()
-            .counters
-            .values()
-            .any(|&v| v > 0)
     }
 }
 
@@ -76,7 +74,9 @@ impl HealthCheck for MetricsHealthCheck {
         "metrics"
     }
 
-    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 + '_>> {
         let has_errors = self.has_errors();
         Box::pin(async move {
             if has_errors {

crates/mytheclipse/src/middleware.rs

@@ -24,11 +24,8 @@ impl std::fmt::Display for PipelineError {
 impl std::error::Error for PipelineError {}
 
 /// A single async middleware stage.
-pub type BoxMiddleware<S> = Arc<
+pub type BoxMiddleware<S> =
-    dyn Fn(S) -> Pin<Box<dyn Future<Output = Result<S, PipelineError>> + Send>>
+    Arc<dyn Fn(S) -> Pin<Box<dyn Future<Output = Result<S, PipelineError>> + Send>> + Send + Sync>;
-        + Send
-        + Sync,
->;
 
 /// Helper to box any `async fn` middleware.
 pub fn mw<S, F, Fut>(f: F) -> BoxMiddleware<S>
@@ -48,7 +45,9 @@ pub struct MiddlewarePipeline<S> {
 
 impl<S: Send + 'static> MiddlewarePipeline<S> {
     pub fn new() -> Self {
-        Self { layers: Arc::new(Mutex::new(Vec::new())) }
+        Self {
+            layers: Arc::new(Mutex::new(Vec::new())),
+        }
     }
 
     /// Appends a middleware stage.
@@ -58,7 +57,10 @@ impl<S: Send + 'static> MiddlewarePipeline<S> {
 
     /// Applies every layer in order, short-circuiting on the first error.
     pub async fn apply(&self, state: S) -> Result<S, PipelineError> {
-        let layers = self.layers.lock().unwrap();
+        // Clone the Arc'd layers out so the MutexGuard is dropped before any
+        // await point (holding it across `layer(current).await` is unsound —
+        // a re-entrant layer could deadlock).
+        let layers: Vec<BoxMiddleware<S>> = self.layers.lock().unwrap().clone();
         let mut current = state;
         for layer in layers.iter() {
             current = layer(current).await?;
@@ -100,7 +102,9 @@ mod tests {
     async fn short_circuits_on_error() {
         let p: MiddlewarePipeline<String> = MiddlewarePipeline::new();
         let reject = mw(|_s: String| async {
-            Err::<_, PipelineError>(PipelineError { msg: "rejected".into() })
+            Err::<_, PipelineError>(PipelineError {
+                msg: "rejected".into(),
+            })
         });
         p.add(reject);
         assert!(matches!(p.apply("x".to_string()).await, Err(_)));

crates/mytheclipse/src/parallel_map.rs

@@ -234,8 +234,7 @@ where
     // Producer: feed items into the bounded channel (backpressures when all
     // workers are busy — no full materialization).
     tokio::spawn(async move {
-        let mut it = items.into_iter();
+        for item in items {
-        while let Some(item) = it.next() {
             if tx.send(item).await.is_err() {
                 break; // all workers dropped
             }
@@ -286,11 +285,9 @@ mod tests {
 
     #[tokio::test]
     async fn maps_in_order_with_bounded_concurrency() {
-        let out = parallel_map(
+        let out = parallel_map(vec![1, 2, 3, 4], 2, |x: i32| async move {
-            vec![1, 2, 3, 4],
+            Ok::<_, std::io::Error>(x * 2)
-            2,
+        })
-            |x: i32| async move { Ok::<_, std::io::Error>(x * 2) },
-        )
         .await
         .unwrap();
         assert_eq!(out, vec![2, 4, 6, 8]);
@@ -298,17 +295,13 @@ mod tests {
 
     #[tokio::test]
     async fn aggregates_errors_from_failing_tasks() {
-        let out = parallel_map(
+        let out = parallel_map(vec![1, 2, 3], 4, |x: i32| async move {
-            vec![1, 2, 3],
-            4,
-            |x: i32| async move {
             if x == 2 {
                 Err(std::io::Error::new(std::io::ErrorKind::Other, "boom"))
             } else {
                 Ok::<_, std::io::Error>(x)
             }
-            },
+        })
-        )
         .await;
         assert!(out.is_err());
         assert_eq!(out.unwrap_err().len(), 1);
@@ -316,11 +309,10 @@ mod tests {
 
     #[tokio::test]
     async fn empty_input_returns_empty() {
-        let out: Result<Vec<i32>, AggregateError> = parallel_map(
+        let out: Result<Vec<i32>, AggregateError> =
-            Vec::<i32>::new(),
+            parallel_map(Vec::<i32>::new(), 4, |x: i32| async move {
-            4,
+                Ok::<_, std::io::Error>(x)
-            |x: i32| async move { Ok::<_, std::io::Error>(x) },
+            })
-        )
             .await;
         assert_eq!(out.unwrap(), vec![]);
     }
@@ -330,17 +322,13 @@ mod tests {
         use std::sync::atomic::{AtomicUsize, Ordering};
         let count = Arc::new(AtomicUsize::new(0));
         let c = Arc::clone(&count);
-        let out = parallel_for_each(
+        let out = parallel_for_each(vec![1_i32, 2, 3, 4, 5], 2, move |_: i32| {
-            vec![1_i32, 2, 3, 4, 5],
-            2,
-            move |_: i32| {
             let c = Arc::clone(&c);
             async move {
                 c.fetch_add(1, Ordering::SeqCst);
                 Ok::<_, std::io::Error>(())
             }
-            },
+        })
-        )
         .await;
         assert!(out.is_ok());
         assert_eq!(count.load(Ordering::SeqCst), 5);

crates/mytheclipse/src/pool.rs

@@ -60,7 +60,11 @@ impl<T: Clone> SemaphorePool<T> {
 #[async_trait]
 impl<T: Clone + Send + Sync + 'static> Pool<T> for SemaphorePool<T> {
     async fn acquire(&self) -> Result<Pooled<T>, PoolError> {
-        let permit = self.semaphore.clone().acquire_owned().await
+        let permit = self
+            .semaphore
+            .clone()
+            .acquire_owned()
+            .await
             .map_err(|_| PoolError::Exhausted)?;
         let idx = ACQUIRE_COUNT.fetch_add(1, Ordering::Relaxed) % self.items.len();
         Ok(Pooled {
@@ -158,11 +162,7 @@ where
                 _permit: pooled._permit,
             })
         } else {
-            let fresh = self
+            let fresh = self.reconnect.reconnect().await.map_err(PoolError::Other)?;
-                .reconnect
-                .reconnect()
-                .await
-                .map_err(PoolError::Other)?;
             Ok(Pooled {
                 resource: fresh,
                 // Reuse the permit from the (dead) lease we already hold.

crates/mytheclipse/src/pool_health.rs

@@ -8,7 +8,7 @@ use std::sync::Arc;
 use std::time::Duration;
 
 use crate::health::{HealthCheck, HealthRegistry, HealthStatus};
-use crate::pool::{Pool, Pooled, PoolError, SemaphorePool};
+use crate::pool::{Pool, PoolError, Pooled, SemaphorePool};
 
 /// A health check backed by a closure.
 struct ClosureCheck {
@@ -21,7 +21,9 @@ impl HealthCheck for ClosureCheck {
         &self.name
     }
 
-    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 + '_>> {
         let status = (self.check)();
         Box::pin(async move { status })
     }

crates/mytheclipse/src/retry.rs

@@ -320,10 +320,19 @@ mod tests {
             ..RetryConfig::default()
         };
         let calls = Cell::new(0u32);
-        let (result, stats) = retry_with_stats(config, |_| true, || async {
+        let (result, stats) = retry_with_stats(
+            config,
+            |_| true,
+            || async {
                 calls.set(calls.get() + 1);
-            if calls.get() < 3 { Err::<u32, &str>("fail") } else { Ok(42u32) }
+                if calls.get() < 3 {
-        }).await;
+                    Err::<u32, &str>("fail")
+                } else {
+                    Ok(42u32)
+                }
+            },
+        )
+        .await;
         assert_eq!(result.unwrap(), 42);
         assert_eq!(stats.attempts, 3);
         assert_eq!(stats.retries, 2);

crates/mytheclipse/src/retry_ext.rs

@@ -76,21 +76,29 @@ where
 #[cfg(test)]
 mod tests {
     use super::*;
-    use std::time::Duration;
-    use std::sync::Arc;
     use std::sync::atomic::{AtomicU32, Ordering};
+    use std::sync::Arc;
+    use std::time::Duration;
 
     #[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 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(()) }
+                if n < 2 {
+                    Err::<(), String>("transient".into())
+                } else {
+                    Ok(())
+                }
             }
         };
 

crates/mytheclipse/src/service_builder.rs

@@ -11,10 +11,10 @@ use tracing::Instrument;
 
 #[cfg(feature = "resiliency")]
 use crate::circuit_breaker::CircuitBreaker;
-#[cfg(feature = "resiliency")]
-use crate::retry::{retry, RetryConfig, RetryError};
 #[cfg(feature = "traffic")]
 use crate::ratelimit::RateLimiter;
+#[cfg(feature = "resiliency")]
+use crate::retry::{retry, RetryConfig, RetryError};
 
 /// Error returned by [`ServiceBuilder::run`].
 #[derive(Debug)]
@@ -55,7 +55,10 @@ pub struct ServiceConfig {
 #[cfg(not(feature = "traffic"))]
 impl Default for ServiceConfig {
     fn default() -> Self {
-        Self { max_attempts: 0, timeout: Duration::ZERO }
+        Self {
+            max_attempts: 0,
+            timeout: Duration::ZERO,
+        }
     }
 }
 
@@ -71,7 +74,12 @@ pub struct ServiceConfig {
 #[cfg(feature = "traffic")]
 impl Default for ServiceConfig {
     fn default() -> Self {
-        Self { max_attempts: 0, timeout: Duration::ZERO, rate_per_sec: 0.0, rate_burst: 0 }
+        Self {
+            max_attempts: 0,
+            timeout: Duration::ZERO,
+            rate_per_sec: 0.0,
+            rate_burst: 0,
+        }
     }
 }
 
@@ -146,7 +154,11 @@ impl ServiceBuilder {
     #[cfg(feature = "resiliency")]
     fn record(&self, ok: bool) {
         if let Some(cb) = &self.circuit {
-            if ok { cb.record_success(); } else { cb.record_failure(); }
+            if ok {
+                cb.record_success();
+            } else {
+                cb.record_failure();
+            }
         }
     }
 
@@ -162,7 +174,7 @@ impl ServiceBuilder {
         #[cfg(feature = "resiliency")]
         {
             if let Some(retry_cfg) = &self.retry_cfg {
-                let mut op = f;
+                let op = f;
                 let result: Result<T, RunError<E>> = if dur > Duration::ZERO {
                     // We can't easily combine retry + timeout with FnMut due to
                     // closure capture rules, so use a manual retry loop instead:
@@ -171,7 +183,8 @@ impl ServiceBuilder {
                     let mut op_ref = op;
                     loop {
                         attempt_no += 1;
-                        let span = tracing::info_span!("mytheclipse_service_call", attempt = attempt_no);
+                        let span =
+                            tracing::info_span!("mytheclipse_service_call", attempt = attempt_no);
                         let fut = op_ref();
                         let attempt_result = tokio::time::timeout(dur, fut.instrument(span)).await;
                         match attempt_result {
@@ -185,7 +198,11 @@ impl ServiceBuilder {
                                     return Err(RunError::Inner(e));
                                 }
                                 // retryable — backoff and retry
-                                let delay = crate::retry::backoff_delay(&cfg, attempt_no, rand::thread_rng());
+                                let delay = crate::retry::backoff_delay(
+                                    &cfg,
+                                    attempt_no,
+                                    rand::thread_rng(),
+                                );
                                 tokio::time::sleep(delay).await;
                             }
                             Err(_) => {
@@ -194,7 +211,11 @@ impl ServiceBuilder {
                                     return Err(RunError::Timeout);
                                 }
                                 // retryable timeout — backoff and retry
-                                let delay = crate::retry::backoff_delay(&cfg, attempt_no, rand::thread_rng());
+                                let delay = crate::retry::backoff_delay(
+                                    &cfg,
+                                    attempt_no,
+                                    rand::thread_rng(),
+                                );
                                 tokio::time::sleep(delay).await;
                             }
                         }
@@ -202,13 +223,16 @@ impl ServiceBuilder {
                 } else {
                     // retry() expects FnMut() -> Fut (not boxed), so adapt.
                     let mut inner_op = op;
-                    retry(retry_cfg.clone(), |_: &E| true, || {
+                    retry(
+                        retry_cfg.clone(),
+                        |_: &E| true,
+                        || {
                             let span = tracing::info_span!("mytheclipse_service_call");
                             let fut = inner_op();
-                        async move {
+                            async move { fut.instrument(span).await }
-                            fut.instrument(span).await
+                        },
-                        }
+                    )
-                    }).await
+                    .await
                     .map_err(|e| {
                         self.record(false);
                         RunError::Retry(e)
@@ -224,7 +248,8 @@ impl ServiceBuilder {
                 let mut op = f;
                 let span = tracing::info_span!("mytheclipse_service_call");
                 let result = if dur > Duration::ZERO {
-                    tokio::time::timeout(dur, op().instrument(span)).await
+                    tokio::time::timeout(dur, op().instrument(span))
+                        .await
                         .map_err(|_| RunError::Timeout)?
                         .map_err(RunError::Inner)
                 } else {
@@ -267,13 +292,19 @@ mod tests {
         cfg.max_attempts = 3;
         let builder = ServiceBuilder::new(cfg);
         let attempts = Arc::new(std::sync::atomic::AtomicU32::new(0));
-        let result = builder.run(|| {
+        let result = builder
+            .run(|| {
                 let a = Arc::clone(&attempts);
                 Box::pin(async move {
                     let n = a.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
-                if n < 2 { Err::<u32, _>(()) } else { Ok::<u32, _>(42) }
+                    if n < 2 {
+                        Err::<u32, _>(())
+                    } else {
+                        Ok::<u32, _>(42)
+                    }
                 })
-        }).await;
+            })
+            .await;
         assert_eq!(result.unwrap(), 42);
     }
 
@@ -282,10 +313,14 @@ mod tests {
         let mut cfg = ServiceConfig::default();
         cfg.timeout = Duration::from_millis(5);
         let builder = ServiceBuilder::new(cfg);
-        let result = builder.run(|| Box::pin(async {
+        let result = builder
+            .run(|| {
+                Box::pin(async {
                     tokio::time::sleep(Duration::from_secs(1)).await;
                     Ok::<_, ()>(42u32)
-        })).await;
+                })
+            })
+            .await;
         assert!(matches!(result, Err(RunError::Timeout)));
     }
 }

crates/mytheclipse/src/shutdown_guard.rs

@@ -37,8 +37,12 @@ use std::sync::{Arc, Mutex};
 /// ShutdownGuard::new(move || { d2.fetch_add(1, Ordering::SeqCst); }).finish();
 /// assert_eq!(done.load(Ordering::SeqCst), 2);
 /// ```
+// Type alias for the stored once-only callback, so the nested Arc<Mutex<…>> field type
+// stays within clippy's `type_complexity` threshold.
+type GuardFn = Box<dyn FnOnce() + Send>;
+
 pub struct ShutdownGuard {
-    inner: Arc<Mutex<Option<Box<dyn FnOnce() + Send>>>>,
+    inner: Arc<Mutex<Option<GuardFn>>>,
 }
 
 impl ShutdownGuard {

crates/mytheclipse/tests/race_stress.rs

@@ -113,7 +113,9 @@ async fn aggregate_error_collects_all_errors_under_contention() {
         })
         .collect();
 
-    let err = AggregateError::from_results(results).err().expect("expected error");
+    let err = AggregateError::from_results(results)
+        .err()
+        .expect("expected error");
     assert_eq!(err.len(), 34, "expected 34 errors (0..99 every 3rd)");
 }