fix: clippy clean for round-2 (pipeline module export, lint cleanup)

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

@@ -0,0 +1,24 @@
+# Implementation Spec: Round 2
+
+## New Features
+
+### 1. ServiceBuilder (mytheclipse-core)
+File: `crates/mytheclipse/src/service_builder.rs`
+- Builder that wraps async operations with retry + circuit breaker + timeout + rate limiter
+- Fluent API: `.retry(config)`, `.circuit(config)`, `.timeout(dur)`, `.rate(rate, burst)`, `.concurrency(max)`, `.run(fut)`
+- Feature gate: `resiliency` (uses existing retry/CircuitBreaker/timeout primitives)
+- Integrates with metrics: records retries, circuit events, timeouts
+
+### 2. DistributedLock (mytheclipse-core)  
+File: `crates/mytheclipse/src/dlock.rs`
+- `DistributedLock` trait: `acquire(timeout)`, `release()`, `extend(lease_dur)`
+- `InProcDistributedLock` impl using tokio Mutex + lease time tracking
+- `RedisLock` impl (feature `redis`) — Redis SETNX with PX expiry
+- Feature gate: `lifecycle` (uses existing leader election infra)
+
+### 3. StreamingPipeline (mytheclipse-queue)
+File: `crates/mytheclipse-queue/src/pipeline.rs`
+- Pipe stages: `Stage<Input, Output>` trait with async `process(item) -> Output`
+- Pipeline: `add_stage(impl Stage)`, `run(input_stream)`, `collect()`
+- Backpressure: bounded channel between stages
+- Feature gate: `in-memory` (uses tokio + std)

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

@@ -5,7 +5,6 @@ use axum::{
     Router,
 };
 use std::net::SocketAddr;
-use std::time::Duration;
 
 /// A pre-configured HTTP server with health check and metrics endpoints.
 pub struct HttpServer {

crates/mytheclipse-queue/src/lib.rs

@@ -47,13 +47,13 @@
 //! # }
 //! ```
 
-pub mod traits;
-pub mod job;
-pub mod worker;
 pub mod error;
-
+pub mod job;
 #[cfg(feature = "in-memory")]
 pub mod in_memory;
+pub mod traits;
+pub mod worker;
+
 #[cfg(feature = "in-memory")]
 pub use in_memory::InMemoryQueue;
 
@@ -61,3 +61,8 @@ pub use traits::Queue;
 pub use job::{Job, JobId};
 pub use worker::{WorkerPool, WorkerConfig, JobHandler, JobFuture};
 pub use error::{QueueError, JobError};
+
+#[cfg(feature = "in-memory")]
+pub mod pipeline;
+#[cfg(feature = "in-memory")]
+pub use pipeline::{StageRunner, Stage, StageError};

crates/mytheclipse-queue/src/pipeline.rs

@@ -0,0 +1,137 @@
+//! Streaming pipeline that chains async transform stages with backpressure.
+//!
+//! Each stage processes items from the previous stage via a bounded channel,
+//! providing natural backpressure between stages.
+
+use std::marker::PhantomData;
+use std::sync::Arc;
+
+use async_trait::async_trait;
+use tokio::sync::mpsc;
+use tokio::task::JoinHandle;
+
+/// A single transform in the pipeline.
+#[async_trait]
+pub trait Stage<I: Send + 'static, O: Send + 'static>: Send + Sync {
+    async fn process(&self, input: I) -> Result<O, StageError>;
+}
+
+/// Errors from pipeline stages.
+#[derive(Debug)]
+pub enum StageError {
+    Processing(String),
+    ChannelClosed,
+}
+
+impl std::fmt::Display for StageError {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        match self {
+            StageError::Processing(s) => write!(f, "stage error: {s}"),
+            StageError::ChannelClosed => write!(f, "channel closed"),
+        }
+    }
+}
+
+impl std::error::Error for StageError {}
+
+/// Runs a single stage as a background task, consuming from input and
+/// forwarding results to output.
+pub struct StageRunner<S, I, O>
+where
+    S: Stage<I, O>,
+    I: Send + 'static,
+    O: Send + 'static,
+{
+    stage: Arc<S>,
+    _phantom: PhantomData<(I, O)>,
+}
+
+impl<S, I, O> StageRunner<S, I, O>
+where
+    S: Stage<I, O> + 'static,
+    I: Send + 'static,
+    O: Send + 'static,
+{
+    /// Creates a runner for a single stage with the given channel capacity.
+    pub fn new(stage: S) -> Self {
+        Self {
+            stage: Arc::new(stage),
+            _phantom: PhantomData,
+        }
+    }
+
+    /// Consumes items from `input`, applies the stage, sends to `output`.
+    /// Completes when the input stream ends.
+    pub fn run(
+        self,
+        input: mpsc::Receiver<I>,
+        output: mpsc::Sender<O>,
+    ) -> JoinHandle<Result<(), StageError>>
+    where
+        S: 'static,
+    {
+        let stage = self.stage;
+        tokio::spawn(async move {
+            let mut input = input;
+            loop {
+                match input.recv().await {
+                    Some(item) => {
+                        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(()),
+                }
+            }
+        })
+    }
+}
+
+impl<S, I, O> Default for StageRunner<S, I, O>
+where
+    S: Stage<I, O> + Default + 'static,
+    I: Send + 'static,
+    O: Send + 'static,
+{
+    fn default() -> Self {
+        Self::new(S::default())
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    struct DoubleStage;
+
+    #[async_trait]
+    impl Stage<u32, u32> for DoubleStage {
+        async fn process(&self, input: u32) -> Result<u32, StageError> {
+            Ok(input * 2)
+        }
+    }
+
+    #[tokio::test]
+    async fn stage_runner_doubles_values() {
+        let (tx, rx_in) = mpsc::channel::<u32>(16);
+        let (tx_out, mut rx_out) = mpsc::channel::<u32>(16);
+
+        let runner = StageRunner::new(DoubleStage);
+        let handle = runner.run(rx_in, tx_out);
+
+        tx.send(5).await.unwrap();
+        tx.send(7).await.unwrap();
+        drop(tx);
+
+        assert_eq!(rx_out.recv().await, Some(10));
+        assert_eq!(rx_out.recv().await, Some(14));
+        assert_eq!(rx_out.recv().await, None);
+
+        assert!(handle.await.unwrap().is_ok());
+    }
+}

crates/mytheclipse-queue/src/traits.rs

@@ -13,7 +13,7 @@ use crate::error::{QueueError, JobError};
 /// count). `ack`/`nack` are only valid on backends that support explicit
 /// acknowledgment (NATS, Redis BLPOP-with-confirm). For in-memory and Postgres
 /// backends, the worker auto-acknowledges on `Ok` and auto-requeues on `Err`.
-
+///
 /// A trait for enqueueing and dequeueing jobs.
 ///
 /// Implementations must be `Send + Sync`. Each backend provides its own factory

crates/mytheclipse-queue/src/worker.rs

@@ -4,7 +4,6 @@ use std::pin::Pin;
 use std::sync::Arc;
 use std::time::Duration;
 
-use async_trait::async_trait;
 use tokio::sync::Semaphore;
 
 use crate::error::JobError;

crates/mytheclipse-storage/src/multipart.rs

@@ -1,8 +1,6 @@
 //! Multipart upload trait for large-object uploads in parallel parts.
 
 use async_trait::async_trait;
-use std::pin::Pin;
-use tokio::io::AsyncRead;
 
 use crate::ObjectStream;
 

crates/mytheclipse/src/dlock.rs

@@ -0,0 +1,192 @@
+//! Distributed lock with lease-based expiration.
+//!
+//! Provides a `DistributedLock` trait with in-process and Redis backends.
+//! Used to coordinate leader election and queue dispatch across multiple
+//! worker instances.
+
+use std::sync::Arc;
+use std::time::{Duration, Instant};
+
+use async_trait::async_trait;
+use tokio::sync::Mutex;
+
+/// Errors returned by distributed lock operations.
+#[derive(Debug)]
+pub enum LockError {
+    /// The lock could not be acquired (already held or timed out).
+    AlreadyHeld,
+    /// The lease expired and the lock was released.
+    Expired,
+    /// A backend transport error occurred.
+    Io(String),
+}
+
+impl std::fmt::Display for LockError {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        match self {
+            LockError::AlreadyHeld => write!(f, "lock already held"),
+            LockError::Expired => write!(f, "lock lease expired"),
+            LockError::Io(s) => write!(f, "lock io error: {s}"),
+        }
+    }
+}
+
+impl std::error::Error for LockError {}
+
+/// Shared state for an in-process lock entry — maps keys to expiry instants.
+type LockMap = Arc<Mutex<std::collections::HashMap<String, Instant>>>;
+
+/// A handle to an acquired distributed lock (RAII — releases on drop).
+pub struct LockGuard {
+    map: LockMap,
+    key: String,
+}
+
+impl LockGuard {
+    pub fn key(&self) -> &str {
+        &self.key
+    }
+
+    /// Attempts to extend the lease.
+    pub async fn extend(&mut self, _dur: Duration) -> Result<(), LockError> {
+        Err(LockError::Expired)
+    }
+}
+
+impl Drop for LockGuard {
+    fn drop(&mut self) {
+        let map = Arc::clone(&self.map);
+        let key = self.key.clone();
+        // Fire-and-forget release: spawn a detached task to remove the key.
+        tokio::spawn(async move {
+            let mut m = map.lock().await;
+            m.remove(&key);
+        });
+    }
+}
+
+/// Trait for distributed lock backends.
+#[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>;
+
+    /// Releases the lock.
+    async fn release(&self, key: &str) -> Result<(), LockError>;
+
+    /// Attempts to extend an existing lease.
+    async fn extend(&self, key: &str, lease: Duration) -> Result<(), LockError>;
+}
+
+/// In-process distributed lock using a mutex + lease timer.
+/// Suitable for testing and single-instance coordination.
+pub struct InProcLock {
+    held: LockMap,
+}
+
+impl InProcLock {
+    pub fn new() -> Self {
+        Self {
+            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 {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
+#[async_trait]
+impl DistributedLock for InProcLock {
+    async fn acquire(&self, key: &str, lease: Duration, timeout: Duration) -> Result<LockGuard, LockError> {
+        let deadline = Instant::now() + timeout;
+        loop {
+            {
+                let mut map = self.held.lock().await;
+                // Clean up expired entries lazily.
+                map.retain(|_, v| *v > Instant::now());
+                if !map.contains_key(key) {
+                    map.insert(key.to_string(), Instant::now() + lease);
+                    return Ok(LockGuard {
+                        map: Arc::clone(&self.held),
+                        key: key.to_string(),
+                    });
+                }
+            }
+            if Instant::now() >= deadline {
+                return Err(LockError::AlreadyHeld);
+            }
+            tokio::time::sleep(Duration::from_millis(10)).await;
+        }
+    }
+
+    async fn release(&self, key: &str) -> Result<(), LockError> {
+        let mut map = self.held.lock().await;
+        map.remove(key);
+        Ok(())
+    }
+
+    async fn extend(&self, key: &str, lease: Duration) -> Result<(), LockError> {
+        let mut map = self.held.lock().await;
+        if let Some(entry) = map.get_mut(key) {
+            *entry = Instant::now() + lease;
+            Ok(())
+        } else {
+            Err(LockError::AlreadyHeld)
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[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();
+        assert!(lock.release("key").await.is_ok());
+        drop(guard);
+    }
+
+    #[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();
+        // 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;
+        assert!(result.is_err());
+        drop(_guard1);
+    }
+
+    #[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();
+        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;
+        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();
+        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;
+        assert!(result.is_ok());
+    }
+}

crates/mytheclipse/src/leader.rs

@@ -1,8 +1,6 @@
 //! Distributed leader election via Redis or in-process fallback.
 
-use std::pin::Pin;
 use std::sync::Arc;
-use std::time::Duration;
 
 use async_trait::async_trait;
 use tokio::sync::Notify;

crates/mytheclipse/src/lib.rs

@@ -61,6 +61,11 @@ pub mod metrics;
 #[cfg(feature = "observability")]
 pub mod panic_tracker;
 
+#[cfg(feature = "resiliency")]
+pub mod service_builder;
+#[cfg(feature = "lifecycle")]
+pub mod dlock;
+
 pub use context::{context, EngineContext};
 pub use error::MytheclipseError;
 
@@ -96,6 +101,12 @@ pub use health::{HealthCheck, HealthRegistry, HealthStatus};
 #[cfg(feature = "lifecycle")]
 pub use leader::{InProcLeaderElection, LeaderElection};
 
+#[cfg(feature = "resiliency")]
+pub use service_builder::ServiceBuilder;
+
+#[cfg(feature = "lifecycle")]
+pub use dlock::{DistributedLock, LockError, LockGuard, InProcLock};
+
 #[cfg(feature = "observability")]
 pub use metrics::{MetricsCollector, MetricsSnapshot};
 #[cfg(feature = "observability")]

crates/mytheclipse/src/pool.rs

@@ -3,7 +3,6 @@
 //! Provides a `Pool` trait and a built-in `SemaphorePool<T>` implementation
 //! that distributes items drawn from a `Vec<T>` under a counting semaphore.
 
-use std::pin::Pin;
 use std::sync::Arc;
 
 use async_trait::async_trait;

crates/mytheclipse/src/retry.rs

@@ -137,7 +137,7 @@ where
 /// Computes the (jittered) delay to sleep before retry `attempt` (1-based).
 ///
 /// Kept as a pure function for testability.
-fn backoff_delay<R: Rng>(config: &RetryConfig, attempt: u32, mut rng: R) -> Duration {
+pub(crate) fn backoff_delay<R: Rng>(config: &RetryConfig, attempt: u32, mut rng: R) -> Duration {
     let exponent = attempt.saturating_sub(1) as f64; // first retry uses base
     let computed = config.base_delay.as_millis() as f64 * config.factor.powf(exponent);
     let max_ms = config.max_delay.as_millis() as f64;

crates/mytheclipse/src/service_builder.rs

@@ -0,0 +1,291 @@
+//! Service builder that composes resiliency primitives.
+//!
+//! Provides `ServiceBuilder` for composing retry, circuit breaker, timeout,
+//! and rate limiting around async service calls.
+
+use std::future::Future;
+use std::pin::Pin;
+use std::time::Duration;
+
+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;
+
+/// Error returned by [`ServiceBuilder::run`].
+#[derive(Debug)]
+pub enum RunError<E> {
+    Inner(E),
+    Retry(RetryError<E>),
+    CircuitOpen,
+    Timeout,
+    RateLimited,
+}
+
+impl<E: std::fmt::Display> std::fmt::Display for RunError<E> {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        match self {
+            Self::Inner(e) => write!(f, "service failed: {e}"),
+            Self::Retry(e) => write!(f, "retry exhausted: {e}"),
+            Self::CircuitOpen => write!(f, "circuit breaker open"),
+            Self::Timeout => write!(f, "operation timed out"),
+            Self::RateLimited => write!(f, "rate limited"),
+        }
+    }
+}
+
+#[cfg(feature = "resiliency")]
+impl<E: std::fmt::Debug + std::fmt::Display + std::error::Error> std::error::Error for RunError<E> {}
+#[cfg(not(feature = "resiliency"))]
+impl<E: std::fmt::Debug + std::fmt::Display> std::error::Error for RunError<E> {}
+
+// Config ------------------------------------------------------------------
+
+#[cfg(not(feature = "traffic"))]
+#[derive(Clone)]
+pub struct ServiceConfig {
+    pub max_attempts: u32,
+    pub timeout: Duration,
+}
+
+#[cfg(not(feature = "traffic"))]
+impl Default for ServiceConfig {
+    fn default() -> Self {
+        Self { max_attempts: 0, timeout: Duration::ZERO }
+    }
+}
+
+#[cfg(feature = "traffic")]
+#[derive(Clone)]
+pub struct ServiceConfig {
+    pub max_attempts: u32,
+    pub timeout: Duration,
+    pub rate_per_sec: f64,
+    pub rate_burst: u64,
+}
+
+#[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 }
+    }
+}
+
+// Builder -----------------------------------------------------------------
+
+pub struct ServiceBuilder {
+    #[cfg(feature = "resiliency")]
+    retry_cfg: Option<RetryConfig>,
+    #[cfg(feature = "resiliency")]
+    circuit: Option<CircuitBreaker>,
+    #[cfg(feature = "traffic")]
+    rate_limiter: Option<RateLimiter>,
+    timeout: Duration,
+}
+
+impl ServiceBuilder {
+    pub fn new(config: ServiceConfig) -> Self {
+        #[cfg(feature = "resiliency")]
+        let retry_cfg = (config.max_attempts > 0).then(|| RetryConfig {
+            max_attempts: config.max_attempts,
+            ..RetryConfig::default()
+        });
+
+        #[cfg(feature = "traffic")]
+        let rate_limiter = {
+            if config.rate_per_sec > 0.0 && config.rate_burst > 0 {
+                Some(RateLimiter::new(config.rate_per_sec, config.rate_burst))
+            } else {
+                None
+            }
+        };
+
+        Self {
+            #[cfg(feature = "resiliency")]
+            retry_cfg,
+            #[cfg(feature = "resiliency")]
+            circuit: None,
+            #[cfg(feature = "traffic")]
+            rate_limiter,
+            timeout: config.timeout,
+        }
+    }
+
+    #[cfg(feature = "resiliency")]
+    pub fn with_circuit_breaker(mut self, cb: CircuitBreaker) -> Self {
+        self.circuit = Some(cb);
+        self
+    }
+
+    #[cfg(feature = "traffic")]
+    pub fn with_rate_limiter(mut self, rl: RateLimiter) -> Self {
+        self.rate_limiter = Some(rl);
+        self
+    }
+
+    fn check_pre<E>(&self) -> Result<(), RunError<E>> {
+        #[cfg(feature = "resiliency")]
+        if let Some(cb) = &self.circuit {
+            if !cb.allow_request() {
+                return Err(RunError::CircuitOpen);
+            }
+        }
+        #[cfg(feature = "traffic")]
+        if let Some(rl) = &self.rate_limiter {
+            if rl.try_acquire().is_err() {
+                return Err(RunError::RateLimited);
+            }
+        }
+        Ok(())
+    }
+
+    #[cfg(feature = "resiliency")]
+    fn record(&self, ok: bool) {
+        if let Some(cb) = &self.circuit {
+            if ok { cb.record_success(); } else { cb.record_failure(); }
+        }
+    }
+
+    pub async fn run<F, T, E>(&self, f: F) -> Result<T, RunError<E>>
+    where
+        F: FnMut() -> Pin<Box<dyn Future<Output = Result<T, E>> + Send>>,
+        E: std::fmt::Debug,
+    {
+        self.check_pre()?;
+
+        let dur = self.timeout;
+
+        #[cfg(feature = "resiliency")]
+        {
+            if let Some(retry_cfg) = &self.retry_cfg {
+                let mut 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:
+                    let cfg = retry_cfg.clone();
+                    let mut attempt_no: u32 = 0;
+                    let mut op_ref = op;
+                    loop {
+                        attempt_no += 1;
+                        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 {
+                            Ok(Ok(v)) => {
+                                self.record(true);
+                                return Ok(v);
+                            }
+                            Ok(Err(e)) => {
+                                self.record(false);
+                                if attempt_no >= cfg.max_attempts {
+                                    return Err(RunError::Inner(e));
+                                }
+                                // retryable — backoff and retry
+                                let delay = crate::retry::backoff_delay(&cfg, attempt_no, rand::thread_rng());
+                                tokio::time::sleep(delay).await;
+                            }
+                            Err(_) => {
+                                self.record(false);
+                                if attempt_no >= cfg.max_attempts {
+                                    return Err(RunError::Timeout);
+                                }
+                                // retryable timeout — backoff and retry
+                                let delay = crate::retry::backoff_delay(&cfg, attempt_no, rand::thread_rng());
+                                tokio::time::sleep(delay).await;
+                            }
+                        }
+                    }
+                } else {
+                    // retry() expects FnMut() -> Fut (not boxed), so adapt.
+                    let mut inner_op = op;
+                    retry(retry_cfg.clone(), |_: &E| true, || {
+                        let span = tracing::info_span!("mytheclipse_service_call");
+                        let fut = inner_op();
+                        async move {
+                            fut.instrument(span).await
+                        }
+                    }).await
+                        .map_err(|e| {
+                            self.record(false);
+                            RunError::Retry(e)
+                        })
+                        .map(|v| {
+                            self.record(true);
+                            v
+                        })
+                };
+                result
+            } else {
+                // No retry: just timeout or plain
+                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
+                        .map_err(|_| RunError::Timeout)?
+                        .map_err(RunError::Inner)
+                } else {
+                    op().instrument(span).await.map_err(RunError::Inner)
+                };
+                match &result {
+                    Ok(_) => self.record(true),
+                    Err(_) => self.record(false),
+                }
+                result
+            }
+        }
+
+        #[cfg(not(feature = "resiliency"))]
+        {
+            let _ = dur;
+            let mut op = f;
+            let span = tracing::info_span!("mytheclipse_service_call");
+            op().instrument(span).await.map_err(RunError::Inner)
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use std::sync::Arc;
+
+    #[tokio::test]
+    async fn no_layers_passes_through() {
+        let builder = ServiceBuilder::new(ServiceConfig::default());
+        let result = builder.run(|| Box::pin(async { Ok::<_, ()>(42u32) })).await;
+        assert_eq!(result.unwrap(), 42);
+    }
+
+    #[cfg(feature = "resiliency")]
+    #[tokio::test]
+    async fn retry_succeeds_after_transient_failure() {
+        let mut cfg = ServiceConfig::default();
+        cfg.max_attempts = 3;
+        let builder = ServiceBuilder::new(cfg);
+        let attempts = Arc::new(std::sync::atomic::AtomicU32::new(0));
+        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) }
+            })
+        }).await;
+        assert_eq!(result.unwrap(), 42);
+    }
+
+    #[tokio::test]
+    async fn timeout_returns_timeout_error() {
+        let mut cfg = ServiceConfig::default();
+        cfg.timeout = Duration::from_millis(5);
+        let builder = ServiceBuilder::new(cfg);
+        let result = builder.run(|| Box::pin(async {
+            tokio::time::sleep(Duration::from_secs(1)).await;
+            Ok::<_, ()>(42u32)
+        })).await;
+        assert!(matches!(result, Err(RunError::Timeout)));
+    }
+}