lean_ctx/core/
archive_fts.rs

1use rusqlite::{Connection, params};
2use std::path::PathBuf;
3use std::sync::Mutex;
4
5use super::data_dir::lean_ctx_data_dir;
6
7static DB: std::sync::LazyLock<Mutex<Option<Connection>>> =
8    std::sync::LazyLock::new(|| Mutex::new(open_db()));
9
10/// Default maximum on-disk size for the archive FTS database. Overridable via
11/// `LEAN_CTX_ARCHIVE_DB_MAX_MB`. Without enforcement this DB grew unbounded
12/// (observed 576 MB in the field — see EPIC 6 / #2364).
13const DEFAULT_MAX_DB_MB: u64 = 500;
14
15/// Run cap enforcement roughly every N inserts to amortize the VACUUM cost.
16const ENFORCE_EVERY_N_INSERTS: usize = 200;
17
18/// If the `-wal` sidecar ever exceeds this size, force a TRUNCATE checkpoint on
19/// the next write regardless of insert count. This bounds the footprint when a
20/// concurrent reader in another lean-ctx process has been holding back
21/// autocheckpoint (observed 256 MB WAL caused by a stale/orphaned daemon).
22const WAL_TRUNCATE_THRESHOLD_BYTES: u64 = 32 * 1024 * 1024;
23
24fn max_db_bytes() -> u64 {
25    std::env::var("LEAN_CTX_ARCHIVE_DB_MAX_MB")
26        .ok()
27        .and_then(|v| v.trim().parse::<u64>().ok())
28        .filter(|m| *m > 0)
29        .unwrap_or(DEFAULT_MAX_DB_MB)
30        .saturating_mul(1024 * 1024)
31}
32
33fn db_path() -> PathBuf {
34    lean_ctx_data_dir()
35        .unwrap_or_else(|_| PathBuf::from(".lean-ctx"))
36        .join("archives")
37        .join("index.db")
38}
39
40/// Current on-disk size of the archive DB in bytes (including WAL). Used by
41/// `doctor` to surface the footprint budget.
42pub fn db_size_bytes() -> u64 {
43    let base = db_path();
44    let mut total = 0u64;
45    for suffix in ["", "-wal", "-shm"] {
46        let p = if suffix.is_empty() {
47            base.clone()
48        } else {
49            PathBuf::from(format!("{}{suffix}", base.display()))
50        };
51        if let Ok(meta) = std::fs::metadata(&p) {
52            total = total.saturating_add(meta.len());
53        }
54    }
55    total
56}
57
58/// Current size of just the `-wal` sidecar file in bytes.
59fn wal_bytes() -> u64 {
60    let wal = PathBuf::from(format!("{}-wal", db_path().display()));
61    std::fs::metadata(&wal).map_or(0, |m| m.len())
62}
63
64fn open_db() -> Option<Connection> {
65    let path = db_path();
66    if let Some(parent) = path.parent() {
67        let _ = std::fs::create_dir_all(parent);
68    }
69    let conn = Connection::open(&path).ok()?;
70    conn.execute_batch(
71        // `busy_timeout` lets a checkpoint wait for a concurrent reader instead of
72        // bailing immediately, and an explicit `wal_autocheckpoint` keeps the WAL
73        // bounded even when several lean-ctx processes (daemon + MCP + CLI) hold
74        // the same DB open. Without these, a stale reader (e.g. an orphaned
75        // daemon) blocked autocheckpoint and the WAL grew to 256 MB in the field.
76        "PRAGMA journal_mode=WAL;
77         PRAGMA synchronous=NORMAL;
78         PRAGMA busy_timeout=5000;
79         PRAGMA wal_autocheckpoint=1000;
80         CREATE TABLE IF NOT EXISTS archive_meta (
81             archive_id TEXT PRIMARY KEY,
82             tool TEXT NOT NULL,
83             command TEXT NOT NULL,
84             created_at TEXT NOT NULL
85         );
86         CREATE VIRTUAL TABLE IF NOT EXISTS archive_fts USING fts5(
87             tool,
88             command,
89             content,
90             archive_id UNINDEXED
91         );",
92    )
93    .ok()?;
94    Some(conn)
95}
96
97pub fn index_entry(archive_id: &str, tool: &str, command: &str, content: &str) {
98    let guard = DB.lock().ok();
99    let Some(conn) = guard.as_ref().and_then(|g| g.as_ref()) else {
100        return;
101    };
102
103    let exists: bool = conn
104        .query_row(
105            "SELECT 1 FROM archive_meta WHERE archive_id = ?1",
106            params![archive_id],
107            |_| Ok(true),
108        )
109        .unwrap_or(false);
110
111    if exists {
112        return;
113    }
114
115    let created_at = chrono::Utc::now().to_rfc3339();
116    let _ = conn.execute(
117        "INSERT OR IGNORE INTO archive_meta (archive_id, tool, command, created_at) VALUES (?1, ?2, ?3, ?4)",
118        params![archive_id, tool, command, created_at],
119    );
120    let _ = conn.execute(
121        "INSERT INTO archive_fts (archive_id, tool, command, content) VALUES (?1, ?2, ?3, ?4)",
122        params![archive_id, tool, command, content],
123    );
124
125    // Amortized cap enforcement: only check periodically, since size checks +
126    // VACUUM are not free.
127    let count: i64 = conn
128        .query_row("SELECT COUNT(*) FROM archive_meta", [], |row| row.get(0))
129        .unwrap_or(0);
130    if (count as usize).is_multiple_of(ENFORCE_EVERY_N_INSERTS) {
131        enforce_cap_locked(conn);
132    }
133
134    // Bound the WAL even between cap-enforcement passes: if a concurrent reader
135    // held back autocheckpoint and the sidecar ballooned, reclaim it now.
136    if wal_bytes() > WAL_TRUNCATE_THRESHOLD_BYTES {
137        let _ = conn.execute_batch("PRAGMA wal_checkpoint(TRUNCATE);");
138    }
139}
140
141/// Enforces the on-disk size cap by deleting the oldest archive entries (by
142/// `created_at`) in batches until the DB is back under budget, then reclaims
143/// space with VACUUM. Operates on an already-locked connection.
144fn enforce_cap_locked(conn: &Connection) {
145    let cap = max_db_bytes();
146    if db_size_bytes() <= cap {
147        return;
148    }
149    // Delete in batches of ~10% of current rows (min 50) until under cap or empty.
150    for _ in 0..50 {
151        let count: i64 = conn
152            .query_row("SELECT COUNT(*) FROM archive_meta", [], |row| row.get(0))
153            .unwrap_or(0);
154        if count == 0 {
155            break;
156        }
157        let batch = (count / 10).max(50);
158        let ids: Vec<String> = conn
159            .prepare("SELECT archive_id FROM archive_meta ORDER BY created_at ASC LIMIT ?1")
160            .and_then(|mut stmt| {
161                let rows = stmt.query_map(params![batch], |row| row.get::<_, String>(0))?;
162                Ok(rows.flatten().collect::<Vec<_>>())
163            })
164            .unwrap_or_default();
165        if ids.is_empty() {
166            break;
167        }
168        for id in &ids {
169            let _ = conn.execute(
170                "DELETE FROM archive_meta WHERE archive_id = ?1",
171                params![id],
172            );
173            let _ = conn.execute("DELETE FROM archive_fts WHERE archive_id = ?1", params![id]);
174            // Drop the backing `.txt`/`.meta.json` too — deleting only the DB row
175            // would orphan the (much larger) content file on disk (#417).
176            super::archive::remove_files(id);
177        }
178        let _ = conn.execute_batch("PRAGMA wal_checkpoint(TRUNCATE); VACUUM;");
179        if db_size_bytes() <= cap {
180            break;
181        }
182    }
183}
184
185/// Public entry point to enforce the archive DB size cap on demand (e.g. from
186/// idle maintenance or `doctor`). Returns the resulting size in bytes.
187pub fn enforce_cap() -> u64 {
188    if let Ok(guard) = DB.lock()
189        && let Some(conn) = guard.as_ref()
190    {
191        enforce_cap_locked(conn);
192    }
193    db_size_bytes()
194}
195
196pub fn remove_entry(archive_id: &str) {
197    let guard = DB.lock().ok();
198    let Some(conn) = guard.as_ref().and_then(|g| g.as_ref()) else {
199        return;
200    };
201    let _ = conn.execute(
202        "DELETE FROM archive_meta WHERE archive_id = ?1",
203        params![archive_id],
204    );
205    let _ = conn.execute(
206        "DELETE FROM archive_fts WHERE archive_id = ?1",
207        params![archive_id],
208    );
209}
210
211#[derive(Debug, Clone)]
212pub struct FtsResult {
213    pub archive_id: String,
214    pub tool: String,
215    pub command: String,
216    pub snippet: String,
217    pub rank: f64,
218}
219
220pub fn search(query: &str, limit: usize) -> Vec<FtsResult> {
221    let guard = DB.lock().ok();
222    let Some(conn) = guard.as_ref().and_then(|g| g.as_ref()) else {
223        return Vec::new();
224    };
225
226    let Ok(mut stmt) = conn.prepare(
227        "SELECT archive_id, tool, command, snippet(archive_fts, 2, '»', '«', '…', 40), rank
228         FROM archive_fts
229         WHERE archive_fts MATCH ?1
230         ORDER BY rank
231         LIMIT ?2",
232    ) else {
233        return Vec::new();
234    };
235
236    stmt.query_map(params![query, limit as i64], |row| {
237        Ok(FtsResult {
238            archive_id: row.get(0)?,
239            tool: row.get(1)?,
240            command: row.get(2)?,
241            snippet: row.get(3)?,
242            rank: row.get(4)?,
243        })
244    })
245    .ok()
246    .map(|rows| rows.flatten().collect::<Vec<_>>())
247    .unwrap_or_default()
248}
249
250pub fn entry_count() -> usize {
251    let guard = DB.lock().ok();
252    let Some(conn) = guard.as_ref().and_then(|g| g.as_ref()) else {
253        return 0;
254    };
255    conn.query_row("SELECT COUNT(*) FROM archive_meta", [], |row| {
256        row.get::<_, i64>(0)
257    })
258    .unwrap_or(0) as usize
259}
260
261#[cfg(test)]
262mod tests {
263    use super::*;
264
265    #[test]
266    fn fts_roundtrip() {
267        let _lock = crate::core::data_dir::test_env_lock();
268        let tmp = tempfile::tempdir().unwrap();
269        crate::test_env::set_var("LEAN_CTX_DATA_DIR", tmp.path());
270
271        // Force re-open by directly testing open_db
272        let conn = open_db().expect("should open");
273        conn.execute(
274            "INSERT INTO archive_meta (archive_id, tool, command, created_at) VALUES ('t1', 'shell', 'git log', '2026-01-01')",
275            [],
276        ).unwrap();
277        conn.execute(
278            "INSERT INTO archive_fts (archive_id, tool, command, content) VALUES ('t1', 'shell', 'git log', 'commit abc refactored the parser module')",
279            [],
280        ).unwrap();
281
282        let mut stmt = conn
283            .prepare("SELECT archive_id FROM archive_fts WHERE archive_fts MATCH 'parser'")
284            .unwrap();
285        let ids: Vec<String> = stmt
286            .query_map([], |row| row.get(0))
287            .unwrap()
288            .flatten()
289            .collect();
290        assert_eq!(ids, vec!["t1"]);
291    }
292
293    #[test]
294    fn open_db_bounds_the_wal() {
295        let _lock = crate::core::data_dir::test_env_lock();
296        let tmp = tempfile::tempdir().unwrap();
297        crate::test_env::set_var("LEAN_CTX_DATA_DIR", tmp.path());
298
299        let conn = open_db().expect("should open");
300
301        // WAL journal mode is required for the FTS write path.
302        let journal_mode: String = conn
303            .query_row("PRAGMA journal_mode;", [], |row| row.get(0))
304            .unwrap();
305        assert_eq!(journal_mode.to_lowercase(), "wal");
306
307        // A bounded (non-zero) autocheckpoint is what keeps the sidecar from
308        // growing unbounded when another process holds the DB open.
309        let autocheckpoint: i64 = conn
310            .query_row("PRAGMA wal_autocheckpoint;", [], |row| row.get(0))
311            .unwrap();
312        assert_eq!(autocheckpoint, 1000);
313    }
314}
lean_ctx/core/archive_fts.rs

lean_ctx/core/
archive_fts.rs
