rskit-resilience 0.1.0-alpha.2

Fault-tolerance: retry, circuit breaker, bulkhead, rate limiter + tower layers
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110

use std::future::Future;
use std::pin::Pin;
use std::task::{Context, Poll};

use rskit_errors::AppError;

use crate::CircuitBreaker;

/// Tower layer that wraps a service with a [`CircuitBreaker`].
#[derive(Clone)]
pub struct CircuitBreakerLayer {
    breaker: CircuitBreaker,
}

impl CircuitBreakerLayer {
    /// Create a new [`CircuitBreakerLayer`] from the given breaker.
    #[must_use]
    pub fn new(breaker: CircuitBreaker) -> Self {
        Self { breaker }
    }

    /// Borrow the configured circuit breaker.
    #[must_use]
    pub const fn breaker(&self) -> &CircuitBreaker {
        &self.breaker
    }
}

impl<S> tower::Layer<S> for CircuitBreakerLayer {
    type Service = CircuitBreakerService<S>;

    fn layer(&self, inner: S) -> Self::Service {
        CircuitBreakerService {
            inner,
            breaker: self.breaker.clone(),
        }
    }
}

/// Tower service that gates calls through a [`CircuitBreaker`].
#[derive(Clone)]
pub struct CircuitBreakerService<S> {
    inner: S,
    breaker: CircuitBreaker,
}

impl<S, Req> tower::Service<Req> for CircuitBreakerService<S>
where
    S: tower::Service<Req, Error = AppError> + Clone + Send + 'static,
    S::Future: Send + 'static,
    S::Response: Send + 'static,
    Req: Send + 'static,
{
    type Response = S::Response;
    type Error = AppError;
    type Future = Pin<Box<dyn Future<Output = Result<S::Response, AppError>> + Send>>;

    fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
        self.inner.poll_ready(cx)
    }

    fn call(&mut self, req: Req) -> Self::Future {
        let clone = self.inner.clone();
        let mut service = std::mem::replace(&mut self.inner, clone);
        let breaker = self.breaker.clone();
        Box::pin(async move { breaker.execute(|| service.call(req)).await })
    }
}

#[cfg(test)]
mod tests {
    use rskit_errors::{AppError, ErrorCode};
    use tower::{Service, ServiceBuilder, ServiceExt};

    use super::*;
    use crate::{CbConfig, CbState};

    #[tokio::test]
    async fn passes_through_success() {
        let breaker =
            CircuitBreaker::new(CbConfig::new("test-layer-cb").with_max_failures(3)).unwrap();
        let layer = CircuitBreakerLayer::new(breaker.clone());
        assert_eq!(layer.breaker().state(), CbState::Closed);
        let service = tower::service_fn(|req: i32| async move { Ok::<i32, AppError>(req + 1) });
        let mut service = ServiceBuilder::new().layer(layer).service(service);

        let result = service.ready().await.unwrap().call(9).await;
        assert_eq!(result.unwrap(), 10);
    }

    #[tokio::test]
    async fn opens_and_rejects() {
        let breaker =
            CircuitBreaker::new(CbConfig::new("test-layer-cb").with_max_failures(2)).unwrap();
        let service = tower::service_fn(|_req: i32| async {
            Err::<i32, AppError>(AppError::new(ErrorCode::Internal, "fail"))
        });
        let mut service = ServiceBuilder::new()
            .layer(CircuitBreakerLayer::new(breaker.clone()))
            .service(service);

        for _ in 0..2 {
            let _ = service.ready().await.unwrap().call(0).await;
        }

        let result = service.ready().await.unwrap().call(0).await;
        assert!(result.is_err());
        assert_eq!(breaker.state(), CbState::Open);
    }
}