hiqlite 0.14.0

Hiqlite - highly-available, embeddable, raft-based SQLite + cache
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
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126

use crate::config::RateLimitConfig;
use crate::{Client, Error};
use std::cmp::max;
use std::sync::atomic::Ordering;
use std::time::Duration;
use tokio::sync::oneshot;
use tokio::task;

impl Client {
    pub(crate) fn spawn_rate_limit_ticker(
        &self,
        config_cache: Option<RateLimitConfig>,
        config_db: Option<RateLimitConfig>,
        cache_await: crossbeam::channel::Receiver<oneshot::Sender<Result<(), Error>>>,
        db_await: crossbeam::channel::Receiver<oneshot::Sender<Result<(), Error>>>,
    ) {
        task::spawn(Self::run_ticker(
            self.clone(),
            config_cache,
            config_db,
            cache_await,
            db_await,
        ));
    }

    #[allow(unused_variables)]
    async fn run_ticker(
        slf: Client,
        config_cache: Option<RateLimitConfig>,
        config_db: Option<RateLimitConfig>,
        cache_await: crossbeam::channel::Receiver<oneshot::Sender<Result<(), Error>>>,
        db_await: crossbeam::channel::Receiver<oneshot::Sender<Result<(), Error>>>,
    ) {
        let mut interval = tokio::time::interval(Duration::from_secs(1));

        loop {
            interval.tick().await;

            #[cfg(feature = "cache")]
            if let Some(config) = &config_cache {
                let lim = slf.inner.rate_limit_cache.as_ref().unwrap();

                lim.try_update(Ordering::SeqCst, Ordering::SeqCst, |current| {
                    Some(max(current + config.rps, config.burst))
                })
                .ok();

                while let Ok(tx) = cache_await.try_recv() {
                    if tx.send(slf.try_rate_limit_cache()).is_err() {
                        lim.fetch_add(1, Ordering::Relaxed);
                    }
                }
            }

            #[cfg(feature = "sqlite")]
            if let Some(config) = &config_db {
                let lim = slf.inner.rate_limit_db.as_ref().unwrap();

                lim.try_update(Ordering::SeqCst, Ordering::SeqCst, |current| {
                    Some(max(current + config.rps, config.burst))
                })
                .ok();

                while let Ok(tx) = db_await.try_recv() {
                    if tx.send(slf.try_rate_limit_db()).is_err() {
                        lim.fetch_add(1, Ordering::Relaxed);
                    }
                }
            }
        }
    }

    #[cfg(feature = "cache")]
    pub(crate) fn try_rate_limit_cache(&self) -> Result<(), Error> {
        if let Some(v) = &self.inner.rate_limit_cache {
            v.try_update(Ordering::SeqCst, Ordering::SeqCst, |current| {
                if current > 0 { Some(current - 1) } else { None }
            })
            .map_err(|_| Error::RateLimit("Raft rate-limit hit for the Cache layer".into()))?;
        }

        Ok(())
    }

    #[cfg(feature = "cache")]
    pub(crate) async fn rate_limit_cache(&self) -> Result<(), Error> {
        match self.try_rate_limit_cache() {
            Ok(_) => Ok(()),
            Err(err) => {
                let (tx, rx) = oneshot::channel();
                if self.inner.rate_limit_cache_await.try_send(tx).is_ok() {
                    rx.await.unwrap()
                } else {
                    Err(err)
                }
            }
        }
    }

    #[cfg(feature = "sqlite")]
    pub(crate) fn try_rate_limit_db(&self) -> Result<(), Error> {
        if let Some(v) = &self.inner.rate_limit_db {
            v.try_update(Ordering::SeqCst, Ordering::SeqCst, |current| {
                if current > 0 { Some(current - 1) } else { None }
            })
            .map_err(|_| Error::RateLimit("Raft rate-limit hit for the DB layer".into()))?;
        }

        Ok(())
    }

    #[cfg(feature = "sqlite")]
    pub(crate) async fn rate_limit_db(&self) -> Result<(), Error> {
        match self.try_rate_limit_db() {
            Ok(_) => Ok(()),
            Err(err) => {
                let (tx, rx) = oneshot::channel();
                if self.inner.rate_limit_db_await.try_send(tx).is_ok() {
                    rx.await.unwrap()
                } else {
                    Err(err)
                }
            }
        }
    }
}