hive-agent: run the event store in WAL and give turn-stats writes a busy timeout
This commit is contained in:
parent
161dd8c44b
commit
62ab45297c
6 changed files with 110 additions and 0 deletions
|
|
@ -150,6 +150,7 @@ impl EventStore {
|
|||
let _ = std::fs::create_dir_all(parent);
|
||||
}
|
||||
let conn = Connection::open(path)?;
|
||||
crate::sqlite_writer::configure_wal(&conn)?;
|
||||
conn.execute_batch(SCHEMA)?;
|
||||
Ok(Self {
|
||||
conn: Mutex::new(conn),
|
||||
|
|
|
|||
|
|
@ -30,6 +30,7 @@ mod reminder_timer;
|
|||
mod reminders;
|
||||
mod schedule_timer;
|
||||
mod serve_common;
|
||||
mod sqlite_writer;
|
||||
mod state_entry_watch;
|
||||
mod stats;
|
||||
mod subagent_role_watch;
|
||||
|
|
|
|||
99
hive-agent/src/sqlite_writer.rs
Normal file
99
hive-agent/src/sqlite_writer.rs
Normal file
|
|
@ -0,0 +1,99 @@
|
|||
//! Shared connection setup for the harness's own sqlite writers.
|
||||
//!
|
||||
//! The default rollback journal costs an fsync per commit and takes an
|
||||
//! exclusive lock, so a write that overlaps a read on another connection (the
|
||||
//! stats page, the hourly vacuum) fails at once with `SQLITE_BUSY`. WAL with
|
||||
//! `synchronous=NORMAL` removes the per-row fsync and lets readers and the
|
||||
//! writer overlap, and the busy timeout makes the remaining writer-vs-writer
|
||||
//! overlap wait instead of dropping the row.
|
||||
//!
|
||||
//! WAL is only for databases that nothing outside the agent reads: a
|
||||
//! read-only connection to a WAL database needs the `-wal` and `-shm`
|
||||
//! sidecars to exist or be creatable, and hive-c0re reads as another uid
|
||||
//! across the agent-owned harness dir. Such a database takes just
|
||||
//! [`BUSY_TIMEOUT`].
|
||||
|
||||
use std::time::Duration;
|
||||
|
||||
use rusqlite::Connection;
|
||||
|
||||
/// How long a writer waits on a locked database before giving up.
|
||||
pub const BUSY_TIMEOUT: Duration = Duration::from_secs(5);
|
||||
|
||||
/// Put `conn` into WAL mode with `synchronous=NORMAL` and a busy timeout.
|
||||
///
|
||||
/// The timeout is set first so the journal-mode switch, which needs a lock,
|
||||
/// waits too. `journal_mode` persists in the database file; the other two are
|
||||
/// per connection, so every writer connection calls this.
|
||||
///
|
||||
/// When sqlite answers the switch with the old mode (it does that, not an
|
||||
/// error, on a filesystem that cannot do WAL), this logs a warning and leaves
|
||||
/// the database in that mode: a working rollback-journal database beats no
|
||||
/// database.
|
||||
///
|
||||
/// # Errors
|
||||
///
|
||||
/// Propagates sqlite pragma failures.
|
||||
pub fn configure_wal(conn: &Connection) -> rusqlite::Result<()> {
|
||||
conn.busy_timeout(BUSY_TIMEOUT)?;
|
||||
let mode =
|
||||
conn.pragma_update_and_check(None, "journal_mode", "WAL", |row| row.get::<_, String>(0))?;
|
||||
if !mode.eq_ignore_ascii_case("wal") {
|
||||
tracing::warn!(%mode, "sqlite journal_mode stayed put instead of switching to wal");
|
||||
}
|
||||
conn.pragma_update(None, "synchronous", "NORMAL")
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
#[test]
|
||||
fn configure_wal_sets_wal_normal_sync_and_busy_timeout() {
|
||||
let dir = tempfile::tempdir().unwrap();
|
||||
let conn = Connection::open(dir.path().join("w.sqlite")).unwrap();
|
||||
configure_wal(&conn).unwrap();
|
||||
|
||||
let mode: String = conn
|
||||
.query_row("PRAGMA journal_mode", [], |r| r.get(0))
|
||||
.unwrap();
|
||||
let sync: i64 = conn
|
||||
.query_row("PRAGMA synchronous", [], |r| r.get(0))
|
||||
.unwrap();
|
||||
let busy: i64 = conn
|
||||
.query_row("PRAGMA busy_timeout", [], |r| r.get(0))
|
||||
.unwrap();
|
||||
assert_eq!(mode, "wal");
|
||||
assert_eq!(sync, 1, "NORMAL");
|
||||
assert_eq!(busy, 5000);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn configure_wal_keeps_a_database_that_cannot_switch_usable() {
|
||||
// An in-memory database cannot switch to WAL; sqlite answers "memory".
|
||||
let conn = Connection::open_in_memory().unwrap();
|
||||
configure_wal(&conn).unwrap();
|
||||
let mode: String = conn
|
||||
.query_row("PRAGMA journal_mode", [], |r| r.get(0))
|
||||
.unwrap();
|
||||
assert_eq!(mode, "memory");
|
||||
conn.execute_batch("CREATE TABLE t(x); INSERT INTO t VALUES (1)")
|
||||
.unwrap();
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn a_second_connection_still_sees_committed_rows() {
|
||||
let dir = tempfile::tempdir().unwrap();
|
||||
let path = dir.path().join("w.sqlite");
|
||||
let writer = Connection::open(&path).unwrap();
|
||||
configure_wal(&writer).unwrap();
|
||||
writer.execute_batch("CREATE TABLE t(x)").unwrap();
|
||||
let reader = Connection::open(&path).unwrap();
|
||||
configure_wal(&reader).unwrap();
|
||||
writer.execute("INSERT INTO t VALUES (1)", []).unwrap();
|
||||
let n: i64 = reader
|
||||
.query_row("SELECT COUNT(*) FROM t", [], |r| r.get(0))
|
||||
.unwrap();
|
||||
assert_eq!(n, 1);
|
||||
}
|
||||
}
|
||||
|
|
@ -162,6 +162,10 @@ impl TurnStats {
|
|||
}
|
||||
let conn = Connection::open(path)
|
||||
.with_context(|| format!("open turn_stats db {}", path.display()))?;
|
||||
// Not WAL: hive-c0re reads this db read-only as another uid, which
|
||||
// fails on a WAL database whose sidecars it cannot create.
|
||||
conn.busy_timeout(crate::sqlite_writer::BUSY_TIMEOUT)
|
||||
.context("set turn_stats busy_timeout")?;
|
||||
conn.execute_batch(SCHEMA)
|
||||
.context("apply turn_stats schema")?;
|
||||
for stmt in MIGRATIONS {
|
||||
|
|
|
|||
|
|
@ -169,6 +169,7 @@ fn delete_trio(dir: &Path, stem: &str) {
|
|||
/// agent's `events.sqlite`. Returns the number of rows deleted.
|
||||
fn vacuum_events(path: &Path) -> Result<u64> {
|
||||
let conn = Connection::open(path)?;
|
||||
crate::sqlite_writer::configure_wal(&conn)?;
|
||||
let cutoff = chrono::Utc::now().timestamp() - STREAM_KEEP_SECS;
|
||||
let removed = conn.execute(
|
||||
"DELETE FROM events WHERE kind = 'stream' AND ts < ?1",
|
||||
|
|
|
|||
Loading…
Reference in a new issue