perf(cron): run store schema DDL once per process, not per query

src/openhuman/cron/store.rs

@@ -6,6 +6,9 @@ use crate::openhuman::cron::{
 use anyhow::{Context, Result};
 use chrono::{DateTime, Utc};
 use rusqlite::{params, Connection};
+use std::collections::HashSet;
+use std::path::{Path, PathBuf};
+use std::sync::{Mutex, OnceLock};
 use uuid::Uuid;
 
 const MAX_CRON_OUTPUT_BYTES: usize = 16 * 1024;
@@ -766,19 +769,121 @@ fn add_column_if_missing(conn: &Connection, name: &str, sql_type: &str) -> Resul
     }
 }
 
-fn with_connection<T>(config: &Config, f: impl FnOnce(&Connection) -> Result<T>) -> Result<T> {
-    let db_path = config.workspace_dir.join("cron").join("jobs.db");
-    if let Some(parent) = db_path.parent() {
-        std::fs::create_dir_all(parent)
-            .with_context(|| format!("Failed to create cron directory: {}", parent.display()))?;
+/// Tracks which cron database files have already had their schema DDL (the
+/// `CREATE TABLE`/`CREATE INDEX` batch plus the `add_column_if_missing`
+/// migration probes) run against them in this process. `with_connection`
+/// deliberately keeps opening a fresh `rusqlite::Connection` per call —
+/// `Connection` is `!Sync`, so caching a single shared one would need a
+/// process-wide mutex that serializes every caller. What repeats needlessly on
+/// every open is the DDL batch itself: 8 statements plus 11
+/// `PRAGMA table_info(cron_jobs)` scans, paid before every one of the 16 store
+/// ops (`due_jobs` every scheduler tick, `record_run` /
+/// `reschedule_after_run` / `record_last_run` per executed job, every
+/// RPC-driven `list_jobs` / `get_job`, …). Gating just that batch behind a
+/// per-path "already initialized" set keeps it to one execution per process per
+/// database file while every call still gets its own connection.
+///
+/// This mirrors `flows::store`'s `INITIALIZED_SCHEMAS` (R-m8), which was itself
+/// derived from this store's `with_connection` idiom — bringing the accepted
+/// gating back to the store it was copied from.
+///
+/// Keyed by path rather than a single flag: tests each open an independent
+/// per-`TempDir` workspace within the same test binary, and a bare
+/// `OnceLock<()>` would silently skip schema creation for every database path
+/// after the first test to run in the process.
+static INITIALIZED_SCHEMAS: OnceLock<Mutex<HashSet<PathBuf>>> = OnceLock::new();
+
+/// On-disk schema version stamped into `PRAGMA user_version` by [`init_schema`]
+/// and checked by [`ensure_schema_initialized`] on a cache hit. **Bump this
+/// whenever a table or `add_column_if_missing` migration is added to
+/// [`init_schema`]** so a database replaced at runtime with an older/partial
+/// schema is re-migrated rather than trusted.
+const CRON_SCHEMA_VERSION: i64 = 1;
+
+/// Ensures the schema on `conn` (a connection to `db_path`) is present and fully
+/// migrated, running the DDL + migrations at most once per process per path.
+///
+/// **The on-disk `PRAGMA user_version` is the authority, and the fast path is
+/// lock-free.** [`init_schema`] stamps [`CRON_SCHEMA_VERSION`] only after a full,
+/// successful migration, so a connection whose `user_version` already matches is
+/// known-good and returns without touching any process-global state — the common
+/// case (an already-initialized store) never acquires the initialization mutex.
+/// Reading `user_version` is a single database-header read, far cheaper than the
+/// 8-statement DDL batch + 11 metadata scans it replaces.
+///
+/// **Initialization is serialized and atomic per process.** Only a version
+/// *mismatch* takes the [`INITIALIZED_SCHEMAS`] lock, and `user_version` is
+/// re-read under it, so two first callers racing on the same fresh path run the
+/// DDL exactly once — the loser observes the winner's stamp on the recheck and
+/// returns (CodeRabbit: initialization must be atomic per path). Independent db
+/// paths contend only during that rare init window, never on the hot path.
+///
+/// **Trust, but verify.** Before this gating existed the DDL ran on every
+/// `with_connection` call, so a database deleted or replaced at runtime (a
+/// workspace reset, a manual deletion, a disk-recovery restore) self-healed on
+/// the next call. The version gate restores that: a stale/zero `user_version` —
+/// a fresh file, or an older/partial schema swapped in under a live process
+/// (`cron_jobs` present but missing a migrated column such as
+/// `agent_id`/`profile_id`, or the `cron_runs` table) — falls through to the
+/// idempotent [`init_schema`] and is re-migrated rather than trusted and later
+/// failing with `no such column` (CodeRabbit / Codex review on #5708).
+///
+/// [`INITIALIZED_SCHEMAS`] no longer gates the DDL — the on-disk version does —
+/// and is kept purely as a **diagnostic marker**: a path present in the set
+/// whose on-disk version no longer matches was initialized by *this* process and
+/// has since been deleted or replaced, which is worth a warning; a path absent
+/// from the set is an ordinary first-ever init and stays silent.
+fn ensure_schema_initialized(conn: &Connection, db_path: &Path) -> Result<()> {
+    let is_current = || -> bool {
+        conn.query_row("PRAGMA user_version", [], |row| row.get::<_, i64>(0))
+            .unwrap_or(0)
+            == CRON_SCHEMA_VERSION
+    };
+
+    // Lock-free fast path: an already-migrated database carries
+    // CRON_SCHEMA_VERSION in its header, so the common case never acquires the
+    // process-global initialization mutex.
+    if is_current() {
+        return Ok(());
     }
 
-    let conn = Connection::open(&db_path)
-        .with_context(|| format!("Failed to open cron DB: {}", db_path.display()))?;
+    // Mismatch ⇒ (re-)initialization is required. Serialize it so two first
+    // callers for the same path cannot both run the DDL, and re-read the version
+    // under the lock — another thread may have migrated between the lock-free
+    // read above and acquiring the guard.
+    let initialized = INITIALIZED_SCHEMAS.get_or_init(|| Mutex::new(HashSet::new()));
+    let mut guard = initialized
+        .lock()
+        .unwrap_or_else(|poisoned| poisoned.into_inner());
+    if is_current() {
+        return Ok(());
+    }
+
+    // Diagnostic only (see the doc comment): a cached path whose on-disk schema
+    // no longer matches was deleted or replaced under a live process — a
+    // first-ever init leaves the path absent and stays silent.
+    if guard.contains(db_path) {
+        tracing::warn!(
+            target: "cron",
+            db = %db_path.display(),
+            expected_version = CRON_SCHEMA_VERSION,
+            "[cron] a database this process already initialized no longer matches the expected schema version (deleted or replaced at runtime?) — re-running schema init"
+        );
+    }
 
+    init_schema(conn)?;
+    guard.insert(db_path.to_path_buf());
+    Ok(())
+}
+
+/// The actual schema DDL + post-hoc column migrations, split out of
+/// `with_connection` so [`ensure_schema_initialized`] can gate it. Does **not**
+/// carry `PRAGMA foreign_keys = ON`: that is a per-connection setting (it is not
+/// persisted in the database file) and must be reapplied on every open, so it
+/// lives in `with_connection` outside this gate.
+fn init_schema(conn: &Connection) -> Result<()> {
     conn.execute_batch(
-        "PRAGMA foreign_keys = ON;
-         CREATE TABLE IF NOT EXISTS cron_jobs (
+        "CREATE TABLE IF NOT EXISTS cron_jobs (
             id               TEXT PRIMARY KEY,
             expression       TEXT NOT NULL,
             command          TEXT NOT NULL,
@@ -826,17 +931,46 @@ fn with_connection<T>(config: &Config, f: impl FnOnce(&Connection) -> Result<T>)
     )
     .context("Failed to initialize cron schema")?;
 
-    add_column_if_missing(&conn, "schedule", "TEXT")?;
-    add_column_if_missing(&conn, "job_type", "TEXT NOT NULL DEFAULT 'shell'")?;
-    add_column_if_missing(&conn, "prompt", "TEXT")?;
-    add_column_if_missing(&conn, "name", "TEXT")?;
-    add_column_if_missing(&conn, "session_target", "TEXT NOT NULL DEFAULT 'isolated'")?;
-    add_column_if_missing(&conn, "model", "TEXT")?;
-    add_column_if_missing(&conn, "enabled", "INTEGER NOT NULL DEFAULT 1")?;
-    add_column_if_missing(&conn, "delivery", "TEXT")?;
-    add_column_if_missing(&conn, "delete_after_run", "INTEGER NOT NULL DEFAULT 0")?;
-    add_column_if_missing(&conn, "agent_id", "TEXT")?;
-    add_column_if_missing(&conn, "profile_id", "TEXT")?;
+    add_column_if_missing(conn, "schedule", "TEXT")?;
+    add_column_if_missing(conn, "job_type", "TEXT NOT NULL DEFAULT 'shell'")?;
+    add_column_if_missing(conn, "prompt", "TEXT")?;
+    add_column_if_missing(conn, "name", "TEXT")?;
+    add_column_if_missing(conn, "session_target", "TEXT NOT NULL DEFAULT 'isolated'")?;
+    add_column_if_missing(conn, "model", "TEXT")?;
+    add_column_if_missing(conn, "enabled", "INTEGER NOT NULL DEFAULT 1")?;
+    add_column_if_missing(conn, "delivery", "TEXT")?;
+    add_column_if_missing(conn, "delete_after_run", "INTEGER NOT NULL DEFAULT 0")?;
+    add_column_if_missing(conn, "agent_id", "TEXT")?;
+    add_column_if_missing(conn, "profile_id", "TEXT")?;
+
+    // Stamp the schema version last, so [`ensure_schema_initialized`] only
+    // trusts a cache hit whose on-disk schema is fully migrated. Bump
+    // CRON_SCHEMA_VERSION whenever a table or `add_column_if_missing` migration
+    // is added above.
+    conn.pragma_update(None, "user_version", CRON_SCHEMA_VERSION)
+        .context("Failed to stamp cron schema version")?;
+
+    Ok(())
+}
+
+fn with_connection<T>(config: &Config, f: impl FnOnce(&Connection) -> Result<T>) -> Result<T> {
+    let db_path = config.workspace_dir.join("cron").join("jobs.db");
+    if let Some(parent) = db_path.parent() {
+        std::fs::create_dir_all(parent)
+            .with_context(|| format!("Failed to create cron directory: {}", parent.display()))?;
+    }
+
+    let conn = Connection::open(&db_path)
+        .with_context(|| format!("Failed to open cron DB: {}", db_path.display()))?;
+
+    // Per-connection pragma: `foreign_keys` is NOT persisted in the database
+    // file, so it must be reapplied on every open regardless of the schema-init
+    // cache below — it is required on every connection for the `cron_runs`
+    // `ON DELETE CASCADE` FK to be enforced.
+    conn.execute_batch("PRAGMA foreign_keys = ON;")
+        .context("Failed to set cron DB connection pragmas")?;
+
+    ensure_schema_initialized(&conn, &db_path)?;
 
     f(&conn)
 }

src/openhuman/cron/store_tests.rs

@@ -612,3 +612,104 @@ fn dedup_named_jobs_ignores_unnamed_jobs() {
     assert_eq!(removed, 0);
     assert_eq!(list_jobs(&config).unwrap().len(), 2);
 }
+
+/// Regression: gating the DDL behind a per-path "already initialized" set
+/// (see [`INITIALIZED_SCHEMAS`]) must not cost the store its self-healing.
+///
+/// Before the gate existed, the DDL ran on every `with_connection` call, so a
+/// database deleted or replaced at runtime (a workspace reset, a manual
+/// deletion, a disk-recovery restore) recovered on the very next call —
+/// `Connection::open` creates a fresh empty file and `CREATE TABLE IF NOT
+/// EXISTS` repopulates it. With a naive cache the set still reports
+/// "initialized" while the file behind it is empty, and every query afterwards
+/// fails `no such table: cron_jobs` until the process restarts. This pins the
+/// verify-on-hit in `ensure_schema_initialized` that restores it.
+#[test]
+fn schema_reinitializes_when_the_database_file_is_deleted_at_runtime() {
+    let tmp = TempDir::new().unwrap();
+    let config = test_config(&tmp);
+
+    // First use populates the per-path cache and creates the schema.
+    add_job(&config, "*/5 * * * *", "echo before-deletion").unwrap();
+    assert_eq!(
+        list_jobs(&config).unwrap().len(),
+        1,
+        "sanity: the job was persisted"
+    );
+
+    // Simulate a workspace reset / manual deletion while the process lives on.
+    let db_path = config.workspace_dir.join("cron").join("jobs.db");
+    assert!(
+        db_path.exists(),
+        "sanity: the cron db exists before deletion"
+    );
+    std::fs::remove_file(&db_path).unwrap();
+    // Defensive: drop any journal sidecars too, so SQLite cannot resurrect
+    // pages from them (cron uses the default rollback journal, not WAL, so
+    // these normally do not exist between calls — removing them is a no-op).
+    let _ = std::fs::remove_file(db_path.with_extension("db-wal"));
+    let _ = std::fs::remove_file(db_path.with_extension("db-shm"));
+    let _ = std::fs::remove_file(db_path.with_extension("db-journal"));
+
+    // The cache still says this path is initialized. Without the verify-on-hit
+    // this errors with `no such table: cron_jobs`.
+    let after = list_jobs(&config)
+        .expect("a deleted database must be re-initialized, not left wedged at 'no such table'");
+    assert!(
+        after.is_empty(),
+        "the recreated database starts empty — the prior job is genuinely gone"
+    );
+
+    // And the store is fully usable again, not merely readable.
+    let recreated = add_job(&config, "*/5 * * * *", "echo after-deletion").unwrap();
+    assert_eq!(
+        get_job(&config, &recreated.id).unwrap().command,
+        "echo after-deletion"
+    );
+    assert_eq!(list_jobs(&config).unwrap().len(), 1);
+}
+
+/// Regression (CodeRabbit / Codex on #5708): a cache hit must be validated
+/// against the *whole* schema, not just the presence of one table. A database
+/// replaced at runtime with an older/partial schema — `cron_jobs` present but a
+/// migrated column dropped — must be re-migrated, not trusted and then failed on
+/// the incomplete schema. The `PRAGMA user_version` check is what catches it.
+#[test]
+fn older_on_disk_schema_under_a_cached_path_is_remigrated() {
+    let tmp = TempDir::new().unwrap();
+    let config = test_config(&tmp);
+
+    // First use creates the full (versioned) schema and caches the path.
+    let original = add_job(&config, "*/5 * * * *", "echo v1").unwrap();
+
+    // Simulate a workspace restore of an OLDER database swapped in under the
+    // same (already-cached) path: drop a migrated column and clear the version
+    // stamp, exactly as a pre-migration database would look on disk.
+    let db_path = config.workspace_dir.join("cron").join("jobs.db");
+    {
+        let raw = rusqlite::Connection::open(&db_path).unwrap();
+        raw.execute_batch(
+            "ALTER TABLE cron_jobs DROP COLUMN profile_id;
+             PRAGMA user_version = 0;",
+        )
+        .unwrap();
+    }
+
+    // The path is still cached. With a single-table `sqlite_master` probe this
+    // would be trusted and `list_jobs` (which selects `profile_id`) would fail
+    // with `no such column`. The version check detects the drift and re-migrates.
+    let listed = list_jobs(&config)
+        .expect("an older on-disk schema under a cached path must be re-migrated, not trusted");
+    assert_eq!(
+        listed.len(),
+        1,
+        "the pre-existing row survives DROP COLUMN and the schema is repaired"
+    );
+    assert_eq!(listed[0].id, original.id);
+    // The migrated column is back (reads as None for the pre-existing row).
+    assert_eq!(get_job(&config, &original.id).unwrap().profile_id, None);
+
+    // And the store is fully usable again.
+    let recreated = add_job(&config, "*/5 * * * *", "echo v2").unwrap();
+    assert_eq!(get_job(&config, &recreated.id).unwrap().command, "echo v2");
+}