agg/cmds/codex/
compact.rs

1use std::io::ErrorKind;
2use std::path::Path;
3use std::path::PathBuf;
4use std::time::Duration;
5
6use owo_colors::OwoColorize;
7use rootcause::prelude::ResultExt;
8use rootcause::report;
9use rusqlite::Connection;
10use rusqlite::OpenFlags;
11
12const DATABASE_FILE_NAME: &str = "logs_2.sqlite";
13const SQLITE_BUSY_TIMEOUT: Duration = Duration::from_secs(3);
14
15/// Compact the local Codex log database after validating its integrity and WAL checkpoint state.
16///
17/// # Errors
18/// - The configured Codex database does not exist or is not a regular file.
19/// - SQLite cannot obtain a write lock within three seconds.
20/// - An integrity check fails, WAL checkpoint is busy or incomplete, or vacuuming fails.
21pub fn run() -> rootcause::Result<()> {
22    let database_path = database_path()?;
23    let connection = open_database(&database_path)?;
24
25    verify_integrity(&connection, "pre-vacuum")?;
26
27    println!("{}", "Checkpointing the Codex log database".blue().bold());
28    let checkpoint = checkpoint_wal(&connection)?;
29    println!(
30        "{} busy={} log_frames={} checkpointed_frames={}",
31        "Checkpoint complete:".green().bold(),
32        checkpoint.busy,
33        checkpoint.log_frames,
34        checkpoint.checkpointed_frames
35    );
36
37    println!("{}", "Compacting the active Codex log database".blue().bold());
38    connection.execute_batch("VACUUM;")?;
39
40    verify_integrity(&connection, "post-vacuum")?;
41    let stats = database_stats(&connection)?;
42    println!(
43        "{} page_count={} freelist_count={}",
44        "Compaction complete:".green().bold(),
45        stats.page_count,
46        stats.freelist_count
47    );
48
49    Ok(())
50}
51
52fn database_path() -> rootcause::Result<PathBuf> {
53    let codex_home = std::env::var_os("CODEX_HOME")
54        .filter(|value| !value.is_empty())
55        .map_or_else(
56            || ytil_sys::dir::build_home_path(&[".codex"]),
57            |value| Ok(PathBuf::from(value)),
58        )?;
59    let database_path = codex_home.join(DATABASE_FILE_NAME);
60    let metadata = database_path.metadata().map_err(|error| {
61        if error.kind() == ErrorKind::NotFound {
62            report!("Codex log database not found").attach(format!("path={}", database_path.display()))
63        } else {
64            report!("cannot read Codex log database metadata")
65                .attach(format!("path={} error={error}", database_path.display()))
66        }
67    })?;
68
69    if !metadata.is_file() {
70        return Err(
71            report!("Codex log database is not a regular file").attach(format!("path={}", database_path.display()))
72        );
73    }
74
75    Ok(database_path)
76}
77
78fn open_database(database_path: &Path) -> rootcause::Result<Connection> {
79    let connection = Connection::open_with_flags(database_path, OpenFlags::SQLITE_OPEN_READ_WRITE)
80        .attach_with(|| format!("cannot open Codex log database | path={}", database_path.display()))?;
81    connection.busy_timeout(SQLITE_BUSY_TIMEOUT)?;
82    Ok(connection)
83}
84
85fn verify_integrity(connection: &Connection, phase: &str) -> rootcause::Result<()> {
86    let mut statement = connection.prepare("PRAGMA quick_check;")?;
87    let results = statement
88        .query_map([], |row| row.get::<_, String>(0))?
89        .collect::<rusqlite::Result<Vec<_>>>()?;
90
91    if results.as_slice() != ["ok"] {
92        return Err(
93            report!("Codex log database integrity check failed").attach(format!("phase={phase} results={results:#?}"))
94        );
95    }
96
97    Ok(())
98}
99
100fn checkpoint_wal(connection: &Connection) -> rootcause::Result<CheckpointState> {
101    let checkpoint = connection.query_row("PRAGMA wal_checkpoint(TRUNCATE);", [], |row| {
102        Ok(CheckpointState {
103            busy: row.get(0)?,
104            log_frames: row.get(1)?,
105            checkpointed_frames: row.get(2)?,
106        })
107    })?;
108
109    if checkpoint.busy != 0 {
110        return Err(report!("Codex log database WAL checkpoint is busy").attach(format!("checkpoint={checkpoint:#?}")));
111    }
112    if checkpoint.log_frames != checkpoint.checkpointed_frames {
113        return Err(report!("Codex log database WAL truncate checkpoint is incomplete")
114            .attach(format!("checkpoint={checkpoint:#?}")));
115    }
116
117    Ok(checkpoint)
118}
119
120fn database_stats(connection: &Connection) -> rootcause::Result<DatabaseStats> {
121    connection
122        .query_row(
123            "SELECT page_count, freelist_count FROM pragma_page_count(), pragma_freelist_count();",
124            [],
125            |row| {
126                Ok(DatabaseStats {
127                    page_count: row.get(0)?,
128                    freelist_count: row.get(1)?,
129                })
130            },
131        )
132        .map_err(Into::into)
133}
134
135#[derive(Debug)]
136struct CheckpointState {
137    busy: i64,
138    log_frames: i64,
139    checkpointed_frames: i64,
140}
141
142struct DatabaseStats {
143    page_count: i64,
144    freelist_count: i64,
145}
agg/cmds/codex/compact.rs

agg/cmds/codex/
compact.rs
