rs-zero 0.2.6

Rust-first microservice framework inspired by go-zero engineering practices
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

use redis::aio::MultiplexedConnection;

use crate::{
    cache_redis::{
        RedisCacheError, RedisCacheResult, RedisCommandOutcome, classify_redis_error,
        redis_breaker_rejection,
    },
    lock::LockResult,
    resil::BreakerGuard,
};

use super::RedisDistributedLock;

#[derive(Clone)]
pub(crate) enum RedisLockBackend {
    Single(redis::Client),
    Cluster(redis::cluster::ClusterClient),
}

impl std::fmt::Debug for RedisLockBackend {
    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            Self::Single(_) => formatter.write_str("RedisLockBackend::Single"),
            Self::Cluster(_) => formatter.write_str("RedisLockBackend::Cluster"),
        }
    }
}

impl RedisDistributedLock {
    pub(crate) async fn single_connection(
        &self,
        client: &redis::Client,
    ) -> LockResult<MultiplexedConnection> {
        let guard = self.allow_breaker("LOCK_CONNECT").await?;
        let result = tokio::time::timeout(
            self.config.redis.connect_timeout,
            client.get_multiplexed_async_connection(),
        )
        .await
        .map_err(|_| RedisCacheError::Timeout("lock connect".to_string()))?
        .map_err(|error| RedisCacheError::Connection(error.to_string()));
        self.record_breaker_outcome(guard, &result).await;
        result.map_err(Into::into)
    }

    pub(crate) async fn cluster_connection(
        &self,
        client: &redis::cluster::ClusterClient,
    ) -> LockResult<redis::cluster_async::ClusterConnection> {
        let guard = self.allow_breaker("LOCK_CLUSTER_CONNECT").await?;
        let result = tokio::time::timeout(
            self.config.redis.connect_timeout,
            client.get_async_connection(),
        )
        .await
        .map_err(|_| RedisCacheError::Timeout("lock cluster connect".to_string()))?
        .map_err(|error| RedisCacheError::Connection(error.to_string()));
        self.record_breaker_outcome(guard, &result).await;
        result.map_err(Into::into)
    }

    async fn allow_breaker(
        &self,
        operation: &'static str,
    ) -> RedisCacheResult<Option<BreakerGuard>> {
        let Some(breaker) = &self.breaker else {
            return Ok(None);
        };

        match breaker.allow().await {
            Ok(guard) => Ok(Some(guard)),
            Err(error) => {
                self.recorder
                    .record_breaker(RedisCommandOutcome::BreakerRejected);
                Err(redis_breaker_rejection(operation, error))
            }
        }
    }

    async fn record_breaker_outcome<T>(
        &self,
        guard: Option<BreakerGuard>,
        result: &RedisCacheResult<T>,
    ) {
        let Some(guard) = guard else {
            return;
        };

        match result {
            Ok(_) => {
                guard.record_success().await;
                self.recorder.record_breaker(RedisCommandOutcome::Success);
            }
            Err(error) if crate::cache_redis::store_breaker::is_breaker_acceptable_error(error) => {
                guard.record_success().await;
                self.recorder.record_breaker(classify_redis_error(error));
            }
            Err(error) => {
                guard.record_failure().await;
                self.recorder.record_breaker(classify_redis_error(error));
            }
        }
    }
}