rightkit_db/
checkpoint_job.rs

1//! Caller-driven WAL checkpoints: PASSIVE when the log grows, RESTART when it is large and the
2//! database has been quiet.
3//!
4//! The crate starts no thread. The owner calls [`WalCheckpointJob::tick`] from the loop or timer
5//! it already runs, and [`WalCheckpointJob::note_activity`] whenever the app uses the database.
6//! A tick is single-flight: while one checkpoint runs on a job, a concurrent tick returns
7//! `Ok(None)` instead of starting a second. Thresholds are in bytes of the `-wal` file.
8//!
9//! A RESTART waits (under the connection's busy timeout) for readers still using the log, so call
10//! it from the thread that owns the writer connection, not from a request path.
11
12use crate::{checkpoint, checkpoint_restart, wal_path, CheckpointStatus, Result};
13use rusqlite::Connection;
14use std::path::{Path, PathBuf};
15use std::sync::atomic::{AtomicBool, Ordering};
16use std::sync::{Mutex, MutexGuard};
17use std::time::{Duration, Instant};
18
19const MIB: u64 = 1024 * 1024;
20
21/// When a [`WalCheckpointJob`] checkpoints.
22#[derive(Clone, Copy, Debug, PartialEq, Eq)]
23pub struct WalCheckpointPolicy {
24    /// PASSIVE once the log is at least this large. A PASSIVE checkpoint never waits for readers
25    /// or writers. Default 16 MiB.
26    pub passive_above_bytes: u64,
27    /// RESTART once the log is at least this large and the database has been idle for
28    /// `quiet_for`. Default 256 MiB.
29    pub restart_above_bytes: u64,
30    /// How long the database must have been idle before a RESTART. Default 30 seconds.
31    pub quiet_for: Duration,
32}
33
34impl Default for WalCheckpointPolicy {
35    fn default() -> Self {
36        Self {
37            passive_above_bytes: 16 * MIB,
38            restart_above_bytes: 256 * MIB,
39            quiet_for: Duration::from_secs(30),
40        }
41    }
42}
43
44/// The checkpoint a tick ran.
45#[derive(Clone, Copy, Debug, PartialEq, Eq)]
46pub enum WalCheckpointMode {
47    Passive,
48    Restart,
49}
50
51/// What one [`WalCheckpointJob::tick`] did.
52#[derive(Clone, Copy, Debug, PartialEq, Eq)]
53pub struct WalCheckpointRun {
54    pub mode: WalCheckpointMode,
55    /// Size of the `-wal` file when the mode was chosen.
56    pub wal_bytes: u64,
57    /// SQLite's report: busy flag, frames in the log, frames copied.
58    pub status: CheckpointStatus,
59}
60
61/// A checkpoint job for one database. Share it by reference; it holds no connection.
62#[derive(Debug)]
63pub struct WalCheckpointJob {
64    wal: PathBuf,
65    policy: WalCheckpointPolicy,
66    last_activity: Mutex<Instant>,
67    running: AtomicBool,
68}
69
70impl WalCheckpointJob {
71    /// A job for the database at `db`, idle from now.
72    pub fn new(db: impl AsRef<Path>, policy: WalCheckpointPolicy) -> Result<Self> {
73        Ok(Self {
74            wal: wal_path(db.as_ref())?,
75            policy,
76            last_activity: Mutex::new(Instant::now()),
77            running: AtomicBool::new(false),
78        })
79    }
80
81    /// The policy this job was built with.
82    pub fn policy(&self) -> WalCheckpointPolicy {
83        self.policy
84    }
85
86    /// Records that the database was used now. Call it on every use; a RESTART needs the database
87    /// to have been idle for `quiet_for`.
88    pub fn note_activity(&self) {
89        self.note_activity_at(Instant::now());
90    }
91
92    /// [`Self::note_activity`] at `now`, for callers that keep their own clock.
93    pub fn note_activity_at(&self, now: Instant) {
94        let mut last = lock_recover(&self.last_activity);
95        if now > *last {
96            *last = now;
97        }
98    }
99
100    /// Runs the checkpoint due now on `conn`, the writer for this database, if the policy calls
101    /// for one. `Ok(None)` when nothing is due or another tick is still running.
102    pub fn tick(&self, conn: &Connection) -> Result<Option<WalCheckpointRun>> {
103        self.tick_at(conn, Instant::now())
104    }
105
106    /// [`Self::tick`] at `now`, for callers that keep their own clock.
107    pub fn tick_at(&self, conn: &Connection, now: Instant) -> Result<Option<WalCheckpointRun>> {
108        if self
109            .running
110            .compare_exchange(false, true, Ordering::AcqRel, Ordering::Acquire)
111            .is_err()
112        {
113            return Ok(None);
114        }
115        let _flight = Flight(&self.running);
116        let wal_bytes = std::fs::metadata(&self.wal).map_or(0, |m| m.len());
117        if wal_bytes == 0 {
118            return Ok(None);
119        }
120        let idle = now.saturating_duration_since(*lock_recover(&self.last_activity));
121        let restart_due =
122            wal_bytes >= self.policy.restart_above_bytes && idle >= self.policy.quiet_for;
123        let mode = if restart_due {
124            WalCheckpointMode::Restart
125        } else if wal_bytes >= self.policy.passive_above_bytes {
126            WalCheckpointMode::Passive
127        } else {
128            return Ok(None);
129        };
130        let status = match mode {
131            WalCheckpointMode::Restart => checkpoint_restart(conn)?,
132            WalCheckpointMode::Passive => checkpoint(conn, "PRAGMA wal_checkpoint(PASSIVE)")?,
133        };
134        Ok(Some(WalCheckpointRun {
135            mode,
136            wal_bytes,
137            status,
138        }))
139    }
140}
141
142/// Clears the single-flight flag when a tick ends, including on an error.
143struct Flight<'a>(&'a AtomicBool);
144
145impl Drop for Flight<'_> {
146    fn drop(&mut self) {
147        self.0.store(false, Ordering::Release);
148    }
149}
150
151fn lock_recover<T>(mutex: &Mutex<T>) -> MutexGuard<'_, T> {
152    mutex
153        .lock()
154        .unwrap_or_else(|poisoned| poisoned.into_inner())
155}
rightkit_db/checkpoint_job.rs

rightkit_db/
checkpoint_job.rs
