salamander/
commit.rs

1//! docs/phase-1.5.md §2 (WP-4) / OQ-1 — the group-commit policy.
2//!
3//! `commit()` fsyncs and is always available. A `CommitPolicy` lets the DB
4//! call it *for* the caller once a threshold is crossed, so a write storm
5//! can amortize fsyncs without the caller hand-rolling a commit cadence.
6//! Additive: the default is `Manual` (Phase-1 behavior — no auto-commit).
7
8use std::time::Duration;
9
10/// When [`crate::Salamander::append`] should trigger an automatic
11/// `commit()` on the caller's behalf.
12///
13/// Thresholds are **combinable** — set any subset with the `and_*` builders;
14/// a commit fires when *any* active threshold is crossed. Byte and count
15/// thresholds are **exact** (checked after every append, so a crossing
16/// commits immediately). The time threshold is **best-effort**: with no
17/// background thread in Phase 1.5 (that's a Phase 3 concern), it can only be
18/// noticed on the next append after the interval has elapsed — an idle DB
19/// does not self-commit.
20///
21/// ```
22/// use salamander::CommitPolicy;
23/// // fsync every 64 KiB *or* every 200 ms, whichever comes first:
24/// let policy = CommitPolicy::every_bytes(64 * 1024).and_millis(200);
25/// ```
26#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
27pub struct CommitPolicy {
28    every_bytes: Option<u64>,
29    every_count: Option<u64>,
30    every_millis: Option<u64>,
31}
32
33impl CommitPolicy {
34    /// No auto-commit — the caller drives durability by calling `commit()`
35    /// itself. This is the default (`CommitPolicy::default()`).
36    pub const fn manual() -> Self {
37        Self {
38            every_bytes: None,
39            every_count: None,
40            every_millis: None,
41        }
42    }
43
44    /// Commit once this many uncommitted **payload bytes** have accumulated.
45    pub const fn every_bytes(bytes: u64) -> Self {
46        Self {
47            every_bytes: Some(bytes),
48            every_count: None,
49            every_millis: None,
50        }
51    }
52
53    /// Commit once this many events have been appended since the last commit.
54    pub const fn every_count(count: u64) -> Self {
55        Self {
56            every_bytes: None,
57            every_count: Some(count),
58            every_millis: None,
59        }
60    }
61
62    /// Commit on the first append at least this many milliseconds after the
63    /// last commit (best-effort — see the type docs).
64    pub const fn every_millis(millis: u64) -> Self {
65        Self {
66            every_bytes: None,
67            every_count: None,
68            every_millis: Some(millis),
69        }
70    }
71
72    /// Add a byte threshold to an existing policy (combinable).
73    pub const fn and_bytes(mut self, bytes: u64) -> Self {
74        self.every_bytes = Some(bytes);
75        self
76    }
77
78    /// Add a count threshold to an existing policy (combinable).
79    pub const fn and_count(mut self, count: u64) -> Self {
80        self.every_count = Some(count);
81        self
82    }
83
84    /// Add a time threshold to an existing policy (combinable).
85    pub const fn and_millis(mut self, millis: u64) -> Self {
86        self.every_millis = Some(millis);
87        self
88    }
89
90    /// Whether an append that left `bytes`/`count` uncommitted, `elapsed`
91    /// since the last commit, should trigger a commit now. Only ever called
92    /// with `count >= 1` (right after an append), so a bare time threshold
93    /// never fsyncs an empty log.
94    pub(crate) fn should_commit(&self, bytes: u64, count: u64, elapsed: Duration) -> bool {
95        crossed(self.every_bytes, bytes)
96            || crossed(self.every_count, count)
97            || crossed(self.every_millis, elapsed.as_millis() as u64)
98    }
99}
100
101fn crossed(threshold: Option<u64>, value: u64) -> bool {
102    threshold.is_some_and(|t| value >= t)
103}
104
105#[cfg(test)]
106mod tests {
107    use super::*;
108
109    const SECOND: Duration = Duration::from_secs(1);
110    const ZERO: Duration = Duration::ZERO;
111
112    #[test]
113    fn manual_never_commits() {
114        let p = CommitPolicy::manual();
115        assert!(!p.should_commit(u64::MAX, u64::MAX, SECOND));
116        assert_eq!(p, CommitPolicy::default());
117    }
118
119    #[test]
120    fn byte_threshold_is_inclusive() {
121        let p = CommitPolicy::every_bytes(100);
122        assert!(!p.should_commit(99, 1, ZERO));
123        assert!(p.should_commit(100, 1, ZERO));
124        assert!(p.should_commit(101, 1, ZERO));
125    }
126
127    #[test]
128    fn count_threshold_is_inclusive() {
129        let p = CommitPolicy::every_count(3);
130        assert!(!p.should_commit(0, 2, ZERO));
131        assert!(p.should_commit(0, 3, ZERO));
132    }
133
134    #[test]
135    fn time_threshold_fires_only_past_the_interval() {
136        let p = CommitPolicy::every_millis(200);
137        assert!(!p.should_commit(0, 1, Duration::from_millis(199)));
138        assert!(p.should_commit(0, 1, Duration::from_millis(200)));
139    }
140
141    #[test]
142    fn combined_fires_on_whichever_crosses_first() {
143        // Big count budget, small byte budget: bytes should trigger.
144        let p = CommitPolicy::every_count(1_000).and_bytes(50);
145        assert!(p.should_commit(50, 1, ZERO));
146        // Neither crossed yet.
147        assert!(!p.should_commit(49, 2, ZERO));
148    }
149}
salamander/commit.rs

salamander/
commit.rs
