agentmux_srv/backend/storage/filestore/
checkpointer.rs

1// Copyright 2025-2026, AgentMux Corp.
2// SPDX-License-Identifier: Apache-2.0
3
4//! Background WAL checkpoints for a file-backed [`FileStore`] (Phase 5a-3,
5//! SPEC_AGENT_PANE_BOUNDED_LIVE_WINDOW_MIGRATION_2026_09_23.md §6.3.7 item 5).
6//!
7//! With SQLite's automatic checkpoint, whichever write crosses the WAL
8//! threshold (1000 pages) runs the checkpoint itself — and every transcript
9//! append rewrites a whole 64 KB part, so that happens about every 60 lines,
10//! holding that line's event (which since 5a-3 waits for its write) for the
11//! checkpoint's duration: the p99 tail. Store connections therefore run with
12//! `wal_autocheckpoint=0`, and this thread checkpoints instead, on its own
13//! connection: in WAL mode a PASSIVE checkpoint runs alongside writers and
14//! readers, taking no lock they wait on.
15//!
16//! PASSIVE copies what it can and never blocks; pages a reader still needs
17//! stay in the WAL until a later pass. The srv's 30-minute TRUNCATE loop
18//! (`bootstrap::spawn_wal_checkpoint_loop`) still resets the file.
19
20use std::path::{Path, PathBuf};
21use std::sync::atomic::{AtomicBool, Ordering};
22use std::sync::Arc;
23use std::time::Duration;
24
25/// Checkpoint once the WAL file is at least this large (SQLite's default
26/// automatic threshold is 1000 pages of 4 KB).
27const WAL_THRESHOLD_BYTES: u64 = 4 * 1024 * 1024;
28/// How often the WAL size is looked at.
29const POLL: Duration = Duration::from_millis(250);
30
31pub(super) struct Checkpointer {
32    stop: Arc<AtomicBool>,
33    thread: Option<std::thread::JoinHandle<()>>,
34}
35
36impl Checkpointer {
37    /// Start checkpointing `db`. `None` if the thread's connection can't be
38    /// opened — the caller then keeps SQLite's automatic checkpoints.
39    pub(super) fn start(db: &Path) -> Option<Self> {
40        let conn = rusqlite::Connection::open(db).ok()?;
41        conn.execute_batch("PRAGMA busy_timeout=5000;").ok()?;
42        let wal = wal_path(db);
43        let stop = Arc::new(AtomicBool::new(false));
44        let thread_stop = Arc::clone(&stop);
45        let thread = std::thread::Builder::new()
46            .name("filestore-checkpoint".into())
47            .spawn(move || {
48                while !thread_stop.load(Ordering::Relaxed) {
49                    std::thread::park_timeout(POLL);
50                    if thread_stop.load(Ordering::Relaxed) {
51                        break;
52                    }
53                    let big = std::fs::metadata(&wal).map(|m| m.len() >= WAL_THRESHOLD_BYTES).unwrap_or(false);
54                    if big {
55                        if let Err(e) = conn.execute_batch("PRAGMA wal_checkpoint(PASSIVE);") {
56                            tracing::debug!(error = %e, "filestore: background checkpoint failed; retrying next poll");
57                        }
58                    }
59                }
60            })
61            .ok()?;
62        Some(Self { stop, thread: Some(thread) })
63    }
64}
65
66impl Drop for Checkpointer {
67    fn drop(&mut self) {
68        self.stop.store(true, Ordering::Relaxed);
69        if let Some(thread) = self.thread.take() {
70            thread.thread().unpark();
71            let _ = thread.join();
72        }
73    }
74}
75
76fn wal_path(db: &Path) -> PathBuf {
77    let mut name = db.as_os_str().to_os_string();
78    name.push("-wal");
79    PathBuf::from(name)
80}
agentmux_srv/backend/storage/filestore/checkpointer.rs

agentmux_srv/backend/storage/filestore/
checkpointer.rs
