From 198154442c4297c94e8743caec81294b478c0d3a Mon Sep 17 00:00:00 2001 From: asepharyana Date: Sat, 29 Aug 2026 17:05:37 +0700 Subject: [PATCH] fix: clippy clean for round-2 (pipeline module export, lint cleanup) --- .hermes/plans/mytheclipse-round2-spec.md | 24 ++ .../src/server/axum_server.rs | 1 - crates/mytheclipse-queue/src/lib.rs | 13 +- crates/mytheclipse-queue/src/pipeline.rs | 137 +++++++++ crates/mytheclipse-queue/src/traits.rs | 2 +- crates/mytheclipse-queue/src/worker.rs | 1 - crates/mytheclipse-storage/src/multipart.rs | 2 - crates/mytheclipse/src/dlock.rs | 192 ++++++++++++ crates/mytheclipse/src/leader.rs | 2 - crates/mytheclipse/src/lib.rs | 11 + crates/mytheclipse/src/pool.rs | 1 - crates/mytheclipse/src/retry.rs | 2 +- crates/mytheclipse/src/service_builder.rs | 291 ++++++++++++++++++ 13 files changed, 666 insertions(+), 13 deletions(-) create mode 100644 .hermes/plans/mytheclipse-round2-spec.md create mode 100644 crates/mytheclipse-queue/src/pipeline.rs create mode 100644 crates/mytheclipse/src/dlock.rs create mode 100644 crates/mytheclipse/src/service_builder.rs diff --git a/.hermes/plans/mytheclipse-round2-spec.md b/.hermes/plans/mytheclipse-round2-spec.md new file mode 100644 index 0000000..fd6e23b --- /dev/null +++ b/.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` 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) diff --git a/crates/mytheclipse-http/src/server/axum_server.rs b/crates/mytheclipse-http/src/server/axum_server.rs index f077f1d..369dfdf 100644 --- a/crates/mytheclipse-http/src/server/axum_server.rs +++ b/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 { diff --git a/crates/mytheclipse-queue/src/lib.rs b/crates/mytheclipse-queue/src/lib.rs index a91a311..7856979 100644 --- a/crates/mytheclipse-queue/src/lib.rs +++ b/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}; diff --git a/crates/mytheclipse-queue/src/pipeline.rs b/crates/mytheclipse-queue/src/pipeline.rs new file mode 100644 index 0000000..54405a5 --- /dev/null +++ b/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: Send + Sync { + async fn process(&self, input: I) -> Result; +} + +/// 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 +where + S: Stage, + I: Send + 'static, + O: Send + 'static, +{ + stage: Arc, + _phantom: PhantomData<(I, O)>, +} + +impl StageRunner +where + S: Stage + '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, + output: mpsc::Sender, + ) -> JoinHandle> + 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 Default for StageRunner +where + S: Stage + 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 for DoubleStage { + async fn process(&self, input: u32) -> Result { + Ok(input * 2) + } + } + + #[tokio::test] + async fn stage_runner_doubles_values() { + let (tx, rx_in) = mpsc::channel::(16); + let (tx_out, mut rx_out) = mpsc::channel::(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()); + } +} diff --git a/crates/mytheclipse-queue/src/traits.rs b/crates/mytheclipse-queue/src/traits.rs index 4193aa6..0462ba9 100644 --- a/crates/mytheclipse-queue/src/traits.rs +++ b/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 diff --git a/crates/mytheclipse-queue/src/worker.rs b/crates/mytheclipse-queue/src/worker.rs index 15d8b96..18bc5ef 100644 --- a/crates/mytheclipse-queue/src/worker.rs +++ b/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; diff --git a/crates/mytheclipse-storage/src/multipart.rs b/crates/mytheclipse-storage/src/multipart.rs index b32ba47..2cd58a5 100644 --- a/crates/mytheclipse-storage/src/multipart.rs +++ b/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; diff --git a/crates/mytheclipse/src/dlock.rs b/crates/mytheclipse/src/dlock.rs new file mode 100644 index 0000000..0f31b54 --- /dev/null +++ b/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>>; + +/// 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; + + /// 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, 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 { + 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()); + } +} diff --git a/crates/mytheclipse/src/leader.rs b/crates/mytheclipse/src/leader.rs index 358b517..7d02f70 100644 --- a/crates/mytheclipse/src/leader.rs +++ b/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; diff --git a/crates/mytheclipse/src/lib.rs b/crates/mytheclipse/src/lib.rs index 0ce674e..35a3642 100644 --- a/crates/mytheclipse/src/lib.rs +++ b/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")] diff --git a/crates/mytheclipse/src/pool.rs b/crates/mytheclipse/src/pool.rs index b9bbc4a..99cd7ce 100644 --- a/crates/mytheclipse/src/pool.rs +++ b/crates/mytheclipse/src/pool.rs @@ -3,7 +3,6 @@ //! Provides a `Pool` trait and a built-in `SemaphorePool` implementation //! that distributes items drawn from a `Vec` under a counting semaphore. -use std::pin::Pin; use std::sync::Arc; use async_trait::async_trait; diff --git a/crates/mytheclipse/src/retry.rs b/crates/mytheclipse/src/retry.rs index 844dfec..5daff20 100644 --- a/crates/mytheclipse/src/retry.rs +++ b/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(config: &RetryConfig, attempt: u32, mut rng: R) -> Duration { +pub(crate) fn backoff_delay(config: &RetryConfig, attempt: u32, mut rng: R) -> Duration { let exponent = attempt.saturating_sub(1) as f64; // first retry uses base let computed = config.base_delay.as_millis() as f64 * config.factor.powf(exponent); let max_ms = config.max_delay.as_millis() as f64; diff --git a/crates/mytheclipse/src/service_builder.rs b/crates/mytheclipse/src/service_builder.rs new file mode 100644 index 0000000..3797e56 --- /dev/null +++ b/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 { + Inner(E), + Retry(RetryError), + CircuitOpen, + Timeout, + RateLimited, +} + +impl std::fmt::Display for RunError { + 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 std::error::Error for RunError {} +#[cfg(not(feature = "resiliency"))] +impl std::error::Error for RunError {} + +// 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, + #[cfg(feature = "resiliency")] + circuit: Option, + #[cfg(feature = "traffic")] + rate_limiter: Option, + 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(&self) -> Result<(), RunError> { + #[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(&self, f: F) -> Result> + where + F: FnMut() -> Pin> + 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> = 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::(()) } else { Ok::(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))); + } +}