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
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
//! Auto-rewrite policy — WHEN the AOF should compact itself. Split from
//! `aof.rs` for the 500-LOC house rule; the mechanics (rewrite_from /
//! concurrent rewrite) stay there, this file owns only the decision.
use crate::aof::Aof;
/// When to auto-compact the AOF. Three independent triggers — ANY hit fires
/// a rewrite (each disabled by its zero): growth (`pct` percent past the
/// last-rewrite size, gated by `min_size` — Redis's classic pair), an
/// absolute cap (`bytes`), and staleness (`interval_secs` since the last
/// rewrite, provided the log actually grew). The growth rule alone is too
/// sluggish for long-lived instances: a 2.2 GB log must reach 4.4 GB before
/// 100% growth fires, and a real deployment rode that to 12-second replays
/// and an OOM loop — the absolute and time rules exist to cap exactly that.
///
/// The default has every rule off: the log is never compacted on its own.
///
/// ```
/// use kevy_persist::RewritePolicy;
///
/// let p = RewritePolicy::default().with_pct(100).with_min_size(64 << 20);
/// // a 1 MiB log reaches 64 MiB only by growing past any baseline
/// assert!(!p.baseline_matters(1 << 20));
/// assert!(p.baseline_matters(48 << 20));
/// assert!(p.with_interval_secs(3600).baseline_matters(0));
/// ```
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Default)]
#[non_exhaustive]
pub struct RewritePolicy {
/// Growth percentage past the last-rewrite baseline (0 = rule off).
///
/// ```
/// use kevy_persist::{Aof, Argv, Fsync, RewritePolicy};
///
/// let path = std::env::temp_dir().join(format!("policy-{}-doc-{}.aof", "pct", std::process::id()));
/// let mut aof = Aof::open(&path, Fsync::No)?;
/// let set = Argv::from(vec![b"SET".to_vec(), b"k".to_vec(), vec![b'v'; 100]]);
/// let doubled = RewritePolicy::default().with_pct(100);
/// assert!(!aof.rewrite_due(doubled), "a fresh log has not grown");
/// while aof.size_bytes() < 2 * aof.size_at_last_rewrite() {
/// aof.append(&set)?;
/// }
/// assert!(aof.rewrite_due(doubled));
/// # drop(aof);
/// # std::fs::remove_file(&path)?;
/// # Ok::<(), std::io::Error>(())
/// ```
pub pct: u32,
/// Minimum size before the growth rule may fire.
///
/// ```
/// use kevy_persist::{Aof, Argv, Fsync, RewritePolicy};
///
/// let path = std::env::temp_dir().join(format!("policy-{}-doc-{}.aof", "min", std::process::id()));
/// let mut aof = Aof::open(&path, Fsync::No)?;
/// let set = Argv::from(vec![b"SET".to_vec(), b"k".to_vec(), vec![b'v'; 100]]);
/// for _ in 0..10 {
/// aof.append(&set)?;
/// }
/// let growth = RewritePolicy::default().with_pct(100);
/// assert!(aof.rewrite_due(growth));
/// assert!(!aof.rewrite_due(growth.with_min_size(1 << 20)), "still too small to bother");
/// # drop(aof);
/// # std::fs::remove_file(&path)?;
/// # Ok::<(), std::io::Error>(())
/// ```
pub min_size: u64,
/// Absolute size cap (0 = rule off).
///
/// ```
/// use kevy_persist::{Aof, Argv, Fsync, RewritePolicy};
///
/// let path = std::env::temp_dir().join(format!("policy-{}-doc-{}.aof", "bytes", std::process::id()));
/// let mut aof = Aof::open(&path, Fsync::No)?;
/// let set = Argv::from(vec![b"SET".to_vec(), b"k".to_vec(), vec![b'v'; 100]]);
/// let cap = RewritePolicy::default().with_bytes(1024);
/// assert!(!aof.rewrite_due(cap));
/// for _ in 0..10 {
/// aof.append(&set)?;
/// }
/// assert!(aof.rewrite_due(cap), "past 1 KiB, whatever the growth");
/// # drop(aof);
/// # std::fs::remove_file(&path)?;
/// # Ok::<(), std::io::Error>(())
/// ```
pub bytes: u64,
/// Rewrite at least this often while the log grows (0 = rule off).
///
/// ```
/// use kevy_persist::{Aof, Argv, Fsync, RewritePolicy};
///
/// let path = std::env::temp_dir().join(format!("policy-{}-doc-{}.aof", "interval", std::process::id()));
/// let mut aof = Aof::open(&path, Fsync::No)?;
/// let set = Argv::from(vec![b"SET".to_vec(), b"k".to_vec(), vec![b'v'; 100]]);
/// let hourly = RewritePolicy::default().with_interval_secs(3600);
/// aof.append(&set)?;
/// assert!(!aof.rewrite_due(hourly), "grown, but not yet an hour old");
/// # drop(aof);
/// # std::fs::remove_file(&path)?;
/// # Ok::<(), std::io::Error>(())
/// ```
pub interval_secs: u64,
}
impl RewritePolicy {
/// Set [`RewritePolicy::pct`].
///
/// ```
/// assert_eq!(kevy_persist::RewritePolicy::default().with_pct(50).pct, 50);
/// ```
#[must_use]
pub fn with_pct(mut self, pct: u32) -> Self {
self.pct = pct;
self
}
/// Set [`RewritePolicy::min_size`].
///
/// ```
/// assert_eq!(kevy_persist::RewritePolicy::default().with_min_size(1 << 20).min_size, 1 << 20);
/// ```
#[must_use]
pub fn with_min_size(mut self, bytes: u64) -> Self {
self.min_size = bytes;
self
}
/// Set [`RewritePolicy::bytes`].
///
/// ```
/// assert_eq!(kevy_persist::RewritePolicy::default().with_bytes(1 << 30).bytes, 1 << 30);
/// ```
#[must_use]
pub fn with_bytes(mut self, bytes: u64) -> Self {
self.bytes = bytes;
self
}
/// Set [`RewritePolicy::interval_secs`].
///
/// ```
/// assert_eq!(kevy_persist::RewritePolicy::default().with_interval_secs(60).interval_secs, 60);
/// ```
#[must_use]
pub fn with_interval_secs(mut self, secs: u64) -> Self {
self.interval_secs = secs;
self
}
/// Whether the growth-rule baseline can change any decision for a log
/// that is `len` bytes long at open. It cannot when the staleness rule
/// is off and `len` is so far under `min_size` that growth past the
/// baseline is already implied by reaching `min_size`: the rule then
/// fires at `min_size` whatever the baseline, provided the baseline is
/// at most `len`. Open paths skip the O(keys) baseline estimate then.
pub fn baseline_matters(&self, len: u64) -> bool {
self.interval_secs > 0
|| (self.pct > 0
&& len.saturating_mul(100u64.saturating_add(u64::from(self.pct)))
> self.min_size.saturating_mul(100))
}
}
impl Aof {
/// Should this AOF be auto-compacted under `policy`? See
/// [`RewritePolicy`] for the three rules. Always false while a
/// concurrent rewrite is already in flight.
pub fn rewrite_due(&self, policy: RewritePolicy) -> bool {
if self.is_rewriting() {
return false;
}
let cur = self.size_bytes();
let baseline = self.size_at_last_rewrite().max(1);
// Growth: (cur - baseline)/baseline >= pct%, once past min_size.
if policy.pct > 0
&& cur >= policy.min_size
&& cur.saturating_mul(100)
>= baseline.saturating_mul(100u64.saturating_add(u64::from(policy.pct)))
{
return true;
}
// Absolute cap: the log is simply too big, growth ratio be damned.
if policy.bytes > 0 && cur >= policy.bytes {
return true;
}
// Staleness: it has been a while AND there is something to fold in.
policy.interval_secs > 0
&& cur > self.size_at_last_rewrite()
&& self.last_rewrite_at().elapsed().as_secs() >= policy.interval_secs
}
}