synd-api 0.4.0

syndicationd backend api
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

use std::time::Duration;

use tokio::{
    task::JoinHandle,
    time::{self, MissedTickBehavior},
};
use tokio_util::sync::CancellationToken;
use tracing::debug;

use super::DaemonSessions;

const MIN_SWEEP_INTERVAL: Duration = Duration::from_millis(100);

/// Daemon-wide controller that expires sessions whose leases are no longer renewed.
pub(crate) struct DaemonSessionSweeper {
    sessions: DaemonSessions,
    schedule: DaemonSessionSweepSchedule,
    shutdown: CancellationToken,
}

impl DaemonSessionSweeper {
    pub(crate) fn new(sessions: DaemonSessions, shutdown: CancellationToken) -> Self {
        let schedule = DaemonSessionSweepSchedule::from_interval(sessions.sweep_interval());

        Self {
            sessions,
            schedule,
            shutdown,
        }
    }

    pub(crate) fn spawn(self) -> DaemonSessionSweeperHandle {
        debug!(
            sweep_interval_ms = self.schedule.interval().as_millis(),
            "Started daemon session sweeper"
        );

        DaemonSessionSweeperHandle {
            task: tokio::spawn(self.run()),
        }
    }

    async fn run(self) {
        let mut interval = time::interval(self.schedule.interval());
        interval.set_missed_tick_behavior(MissedTickBehavior::Skip);

        loop {
            tokio::select! {
                () = self.shutdown.cancelled() => {
                    debug!("Stopped daemon session sweeper");
                    break;
                }
                _ = interval.tick() => {
                    let outcome = self.sessions.sweep_expired();
                    debug!(
                        expired_sessions = outcome.expired_session_count(),
                        active_sessions = outcome.active_sessions(),
                        "Daemon session sweeper tick"
                    );
                }
            }
        }
    }
}

/// Handle that keeps the daemon session sweeper task alive.
pub(crate) struct DaemonSessionSweeperHandle {
    task: JoinHandle<()>,
}

impl Drop for DaemonSessionSweeperHandle {
    fn drop(&mut self) {
        self.task.abort();
    }
}

/// Sweep cadence used by the daemon session sweeper.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
struct DaemonSessionSweepSchedule {
    interval: Duration,
}

impl DaemonSessionSweepSchedule {
    fn from_interval(interval: Duration) -> Self {
        Self {
            interval: interval.max(MIN_SWEEP_INTERVAL),
        }
    }

    fn interval(self) -> Duration {
        self.interval
    }
}

#[cfg(test)]
mod tests {
    use std::time::Duration;

    use super::{DaemonSessionSweepSchedule, MIN_SWEEP_INTERVAL};

    #[test]
    fn preserves_configured_sweep_interval() {
        let schedule = DaemonSessionSweepSchedule::from_interval(Duration::from_secs(5));

        assert_eq!(schedule.interval(), Duration::from_secs(5));
    }

    #[test]
    fn sweep_interval_has_minimum() {
        let schedule = DaemonSessionSweepSchedule::from_interval(Duration::ZERO);

        assert_eq!(schedule.interval(), MIN_SWEEP_INTERVAL);
    }
}