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
//! Background periodic TTL sweep worker.
//!
//! Mirrors [`Flusher`](crate::flusher::Flusher): wakes on a fixed interval, runs a
//! sweep closure with a per-pass budget, and backs off to 1 s when `more` work
//! remains. Wired to [`ShutdownSignal`](crate::shutdown::ShutdownSignal).
use std::time::Duration;
use crate::error::DbResult;
/// Aggregated result of one worker tick across all collections.
///
/// Public because it appears in [`Sweeper::start`]'s closure signature, the same
/// way [`Flusher`](crate::Flusher) exposes its own worker seam.
pub struct SweepPassResult {
/// Rows deleted this tick. Reported for callers that wrap the sweep closure;
/// the worker's own scheduling ignores it and keys only off `more`.
pub deleted: usize,
/// At least one collection still has due rows the budget did not reach.
/// The worker drops to a 1 s interval until this comes back `false`.
pub more: bool,
}
/// Background periodic TTL sweep handle. Stops and joins its thread on `stop()` or
/// `Drop`.
pub struct Sweeper {
stop: crate::shutdown::ShutdownSignal,
handle: Option<std::thread::JoinHandle<()>>,
}
impl Sweeper {
/// Start a background sweep thread with its own shutdown signal.
pub fn start(
sweep_fn: impl Fn(usize) -> DbResult<SweepPassResult> + Send + 'static,
interval: Duration,
budget: usize,
) -> Self {
Self::start_with_signal(
sweep_fn,
interval,
budget,
crate::shutdown::ShutdownSignal::new(),
)
}
/// Start a background sweep thread controlled by an external shutdown signal.
pub fn start_with_signal(
sweep_fn: impl Fn(usize) -> DbResult<SweepPassResult> + Send + 'static,
interval: Duration,
budget: usize,
signal: crate::shutdown::ShutdownSignal,
) -> Self {
let stop = signal.clone();
let configured = interval;
let handle = std::thread::spawn(move || {
let mut current_interval = configured;
while !stop.is_shutdown() {
if stop.wait_timeout(current_interval) {
break;
}
match sweep_fn(budget) {
Ok(r) if r.more => current_interval = Duration::from_secs(1),
Ok(_) => current_interval = configured,
Err(e) => {
tracing::error!(error = %e, "ttl sweep error");
current_interval = configured;
}
}
}
});
Self {
stop: signal,
handle: Some(handle),
}
}
/// Signal the worker to stop and join its thread.
pub fn stop(&mut self) {
self.stop.shutdown();
if let Some(h) = self.handle.take() {
let _ = h.join();
}
}
}
impl Drop for Sweeper {
fn drop(&mut self) {
self.stop();
}
}