ursula 0.5.1

Ursula deployment binary: Durable Streams server, gateway, and event-time indexer.
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
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149

//! Process-local Raft WAL disk-pressure state shared by admission, readiness,
//! metrics, and the leadership gate.

use std::sync::Arc;
use std::sync::atomic::AtomicBool;
use std::sync::atomic::AtomicU64;
use std::sync::atomic::Ordering;

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) enum WalDiskTransition {
    NoChange,
    EnterPressure,
    LeavePressure,
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) struct WalDiskSnapshot {
    pub available_bytes: u64,
    pub min_available_bytes: u64,
    pub resume_available_bytes: u64,
    pub pressure: bool,
    pub stat_errors: u64,
}

#[derive(Debug)]
struct WalDiskMonitorInner {
    available_bytes: AtomicU64,
    min_available_bytes: u64,
    resume_available_bytes: u64,
    pressure: AtomicBool,
    stat_errors: AtomicU64,
}

/// Hysteretic free-space guard for the configured Raft WAL filesystem.
#[derive(Debug, Clone)]
pub(crate) struct WalDiskMonitor {
    inner: Arc<WalDiskMonitorInner>,
}

impl Default for WalDiskMonitor {
    fn default() -> Self {
        Self::new(0, 0)
    }
}

impl WalDiskMonitor {
    pub(crate) fn new(min_available_bytes: u64, resume_available_bytes: u64) -> Self {
        Self {
            inner: Arc::new(WalDiskMonitorInner {
                available_bytes: AtomicU64::new(0),
                min_available_bytes,
                resume_available_bytes,
                pressure: AtomicBool::new(false),
                stat_errors: AtomicU64::new(0),
            }),
        }
    }

    pub(crate) fn enabled(&self) -> bool {
        self.inner.min_available_bytes > 0
    }

    pub(crate) fn is_pressured(&self) -> bool {
        self.inner.pressure.load(Ordering::Acquire)
    }

    pub(crate) fn observe_available(&self, available_bytes: u64) -> WalDiskTransition {
        self.inner
            .available_bytes
            .store(available_bytes, Ordering::Release);
        if !self.enabled() {
            return WalDiskTransition::NoChange;
        }

        let pressured = self.is_pressured();
        let next = if pressured {
            available_bytes < self.inner.resume_available_bytes
        } else {
            available_bytes < self.inner.min_available_bytes
        };
        if next == pressured {
            return WalDiskTransition::NoChange;
        }
        self.inner.pressure.store(next, Ordering::Release);
        if next {
            WalDiskTransition::EnterPressure
        } else {
            WalDiskTransition::LeavePressure
        }
    }

    /// Fail closed when the filesystem cannot be inspected. The error counter
    /// distinguishes a stat failure from a genuine zero-free-byte sample.
    pub(crate) fn observe_error(&self) -> WalDiskTransition {
        self.inner.stat_errors.fetch_add(1, Ordering::Relaxed);
        if !self.enabled() || self.inner.pressure.swap(true, Ordering::AcqRel) {
            WalDiskTransition::NoChange
        } else {
            WalDiskTransition::EnterPressure
        }
    }

    pub(crate) fn snapshot(&self) -> WalDiskSnapshot {
        WalDiskSnapshot {
            available_bytes: self.inner.available_bytes.load(Ordering::Acquire),
            min_available_bytes: self.inner.min_available_bytes,
            resume_available_bytes: self.inner.resume_available_bytes,
            pressure: self.is_pressured(),
            stat_errors: self.inner.stat_errors.load(Ordering::Relaxed),
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn disk_pressure_uses_hysteresis() {
        let monitor = WalDiskMonitor::new(100, 200);
        assert_eq!(monitor.observe_available(150), WalDiskTransition::NoChange);
        assert_eq!(
            monitor.observe_available(99),
            WalDiskTransition::EnterPressure
        );
        assert!(monitor.is_pressured());
        assert_eq!(monitor.observe_available(150), WalDiskTransition::NoChange);
        assert_eq!(
            monitor.observe_available(200),
            WalDiskTransition::LeavePressure
        );
        assert!(!monitor.is_pressured());
    }

    #[test]
    fn disk_stat_error_fails_closed_and_counts() {
        let monitor = WalDiskMonitor::new(100, 200);
        assert_eq!(monitor.observe_error(), WalDiskTransition::EnterPressure);
        assert_eq!(monitor.observe_error(), WalDiskTransition::NoChange);
        assert_eq!(monitor.snapshot().stat_errors, 2);
    }

    #[test]
    fn zero_minimum_disables_pressure() {
        let monitor = WalDiskMonitor::default();
        assert_eq!(monitor.observe_error(), WalDiskTransition::NoChange);
        assert!(!monitor.is_pressured());
    }
}