From f1bd549d1acb53430b1ba7e5f4f6fed860a7d9ce Mon Sep 17 00:00:00 2001 From: Maciej Myszkiewicz Date: Wed, 2 Sep 2026 09:55:23 +0200 Subject: [PATCH] Gate flows schema-init on user_version, not a single-table probe MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `ensure_schema_initialized` confirmed a cached "already initialized" path with a single-table `sqlite_master` presence probe for `flow_definitions`. That honours the cache whenever the table merely exists, so a `flows.db` replaced at runtime with an older/partial schema — the table present but missing a migrated column (`require_approval` / `graph_hash`) or one of the other tables — passed the probe, and the next query then failed with `no such column` / `no such table` until the process restarted. The probe only restored self-healing for a deleted (empty) database, not a drifted one. The gate also dropped the `INITIALIZED_SCHEMAS` guard before `init_schema`, so two first callers for the same path could both run the DDL. Replace the presence probe with a `PRAGMA user_version` gate and make the fast path lock-free: - `init_schema` stamps `FLOWS_DB_SCHEMA_VERSION` into `user_version` only after a full, successful migration. - `ensure_schema_initialized` reads `user_version` lock-free first and returns on a match, so the common already-initialized case — which runs once per node per live run via `upsert_flow_run_step`, not just at store open — never acquires the process mutex. - Only a version mismatch takes the `INITIALIZED_SCHEMAS` lock, re-reads `user_version` under it (double-check), then runs the idempotent `init_schema`, so init is atomic per path and a stale/partial on-disk schema (including a drifted column) is re-migrated instead of trusted. - `INITIALIZED_SCHEMAS` no longer gates the DDL; it is kept as a diagnostic marker so the "deleted/replaced at runtime" warning fires only for a path this process already initialized, not on every fresh-boot init (version 0). Add `flows::older_on_disk_schema_under_a_cached_path_is_remigrated`, which drops the `require_approval` migrated column and resets `user_version` under a cached path, then asserts the next op re-migrates rather than failing `no such column`. These two findings were raised (CodeRabbit/Codex) on the equivalent in-repo fix in the host before this store was extracted here; the lock-free variant was verified on tinyhumansai/openhuman#5708. --- crates/tinyflows-sqlite/src/flows/mod.rs | 154 ++++++++++++------ .../src/flows/schema_tests.rs | 63 +++++++ 2 files changed, 165 insertions(+), 52 deletions(-) diff --git a/crates/tinyflows-sqlite/src/flows/mod.rs b/crates/tinyflows-sqlite/src/flows/mod.rs index 04698406..ab78556e 100644 --- a/crates/tinyflows-sqlite/src/flows/mod.rs +++ b/crates/tinyflows-sqlite/src/flows/mod.rs @@ -35,76 +35,116 @@ pub use run_steps::*; pub use runs::*; pub use suggestions::*; -/// Tracks which flows database files have already had their schema DDL (the -/// `CREATE TABLE`/`CREATE INDEX` batch, `PRAGMA journal_mode = WAL`, and the -/// `add_column_if_missing` migration probe) run against them in this process -/// (R-m8). `with_connection` deliberately keeps opening a fresh, lightweight +/// Diagnostic marker recording which flows database files this process has +/// initialized (R-m8). It no longer *gates* the DDL — the on-disk +/// `PRAGMA user_version` does that (see [`ensure_schema_initialized`]) — it only +/// distinguishes an ordinary first-ever init (path absent ⇒ silent) from a +/// database this process already initialized whose schema has since drifted on +/// disk (path present, version stale ⇒ worth a warning). +/// +/// `with_connection` deliberately keeps opening a fresh, lightweight /// `rusqlite::Connection` per call — `Connection` is `!Sync`, so caching a /// single shared one would need a process-wide mutex that serializes every /// caller, including the concurrent-writer scenario [`upsert_flow_run_step`]'s /// `BEGIN IMMEDIATE` fix (R-m1) depends on being able to run from independent -/// connections. What actually repeats needlessly on every open is the DDL +/// connections. What actually repeated needlessly on every open was the DDL /// batch itself — including once per node per live run via -/// `upsert_flow_run_step`. 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. +/// `upsert_flow_run_step`; the version gate now keeps it to one execution per +/// process per database file while every call still gets its own connection. /// /// 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. +/// `OnceLock<()>` would report the wrong marker for every database path after +/// the first opened in the process. static INITIALIZED_SCHEMAS: OnceLock>> = OnceLock::new(); -/// Runs the one-time schema DDL + migrations against `conn` unless `db_path` -/// has already been initialized in this process (see [`INITIALIZED_SCHEMAS`]). -/// Only marks `db_path` as initialized *after* [`init_schema`] succeeds, so a -/// transient failure (e.g. disk I/O) is retried on the next call rather than -/// permanently wedging the store into believing a schema exists that was -/// never created. +/// On-disk schema version stamped into `flows.db`'s `PRAGMA user_version` by +/// [`init_schema`] and checked by [`ensure_schema_initialized`]. **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. +/// +/// Distinct from `tinyflows::model::CURRENT_SCHEMA_VERSION`, which versions the +/// graph JSON payload — this versions the SQLite file's own schema. +const FLOWS_DB_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 [`FLOWS_DB_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. This matters here specifically: `with_connection` (and +/// so this check) runs once per node per live run via `upsert_flow_run_step`, +/// not just at store open, so keeping the hot path lock-free is load-bearing. +/// Reading `user_version` is a single database-header read, far cheaper than the +/// ~11-statement DDL batch 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. Independent database paths contend only during that rare init +/// window, never on the hot path. /// -/// **Trust, but verify.** A cache hit is confirmed against the file actually on -/// disk before it is honoured. Before this gating existed, the DDL ran on every +/// **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 very next call: `Connection::open` silently creates a fresh empty file, -/// and `CREATE TABLE IF NOT EXISTS` immediately repopulated it. Caching removes -/// that safety net: the set still says "initialized" while the file behind it is -/// empty, so every subsequent query fails with `no such table` until the process -/// restarts. One indexed `sqlite_master` lookup is far cheaper than the ~11 -/// statement DDL batch and restores the self-healing, so it is paid on each hit -/// rather than trusting a cache entry that the filesystem may have invalidated. +/// the next call: `Connection::open` creates a fresh empty file and +/// `CREATE TABLE IF NOT EXISTS` repopulates it. The version gate restores that +/// for the *whole* schema, not just one table's presence: a stale/zero +/// `user_version` — a fresh file, or an older/partial schema swapped in under a +/// live process (`flow_definitions` present but missing a migrated column such +/// as `require_approval` / `graph_hash`, or one of the other tables) — falls +/// through to the idempotent [`init_schema`] and is re-migrated rather than +/// trusted and later failing with `no such table` / `no such column`. A +/// single-table `sqlite_master` probe could not catch column drift. +/// +/// [`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<()> { - use rusqlite::OptionalExtension; + let is_current = || -> bool { + conn.query_row("PRAGMA user_version", [], |row| row.get::<_, i64>(0)) + .unwrap_or(0) + == FLOWS_DB_SCHEMA_VERSION + }; - let initialized = INITIALIZED_SCHEMAS.get_or_init(|| Mutex::new(HashSet::new())); - { - let guard = initialized - .lock() - .unwrap_or_else(|poisoned| poisoned.into_inner()); - if guard.contains(db_path) { - let schema_present: bool = conn - .query_row( - "SELECT 1 FROM sqlite_master WHERE type = 'table' AND name = 'flow_definitions'", - [], - |_| Ok(true), - ) - .optional() - .context("Failed to probe flows schema presence")? - .unwrap_or(false); - if schema_present { - return Ok(()); - } - tracing::warn!( - target: "flows", - db = %db_path.display(), - "[flows] schema cached as initialized but the database has no tables (deleted or replaced at runtime?) — re-running schema init" - ); - } + // Lock-free fast path: an already-migrated database carries + // FLOWS_DB_SCHEMA_VERSION in its header, so the common case never acquires + // the process-global initialization mutex. + if is_current() { + return Ok(()); } - init_schema(conn)?; + + // 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: "flows", + db = %db_path.display(), + "[flows] 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(()) } @@ -112,8 +152,11 @@ fn ensure_schema_initialized(conn: &Connection, db_path: &Path) -> Result<()> { /// The actual schema DDL: 5 `CREATE TABLE IF NOT EXISTS` + 6 `CREATE INDEX IF /// NOT EXISTS` + `PRAGMA journal_mode = WAL` (a persistent db-file setting, /// not per-connection — safe, and now guaranteed, to run only once) plus the -/// `require_approval` post-hoc column migration. Split out of +/// `require_approval` / `graph_hash` post-hoc column migrations. Split out of /// `with_connection` so [`ensure_schema_initialized`] can gate it (R-m8). +/// Stamps [`FLOWS_DB_SCHEMA_VERSION`] into `user_version` last, so the gate can +/// distinguish a fully-migrated database from an older/partial one on a +/// cache hit. fn init_schema(conn: &Connection) -> Result<()> { conn.execute_batch( "PRAGMA journal_mode = WAL; @@ -201,6 +244,13 @@ fn init_schema(conn: &Connection) -> Result<()> { // approval. add_column_if_missing(conn, "flow_runs", "graph_hash", "TEXT")?; + // Stamp the schema version last, so [`ensure_schema_initialized`] only + // trusts a cache hit whose on-disk schema is fully migrated. Bump + // FLOWS_DB_SCHEMA_VERSION whenever a table or `add_column_if_missing` + // migration is added above. + conn.pragma_update(None, "user_version", FLOWS_DB_SCHEMA_VERSION) + .context("Failed to stamp flows schema version")?; + Ok(()) } diff --git a/crates/tinyflows-sqlite/src/flows/schema_tests.rs b/crates/tinyflows-sqlite/src/flows/schema_tests.rs index 3909bc51..47f1479e 100644 --- a/crates/tinyflows-sqlite/src/flows/schema_tests.rs +++ b/crates/tinyflows-sqlite/src/flows/schema_tests.rs @@ -123,3 +123,66 @@ fn schema_reinitializes_when_the_database_file_is_deleted_at_runtime() { let (flows_final, _) = list_flows(&dir).unwrap(); assert_eq!(flows_final.len(), 1); } + +/// Companion to the deletion test: a database *replaced* at runtime with an +/// older/partial schema (rather than deleted) must also be re-migrated, not +/// trusted. This is the case a single-table `sqlite_master` presence probe +/// could not catch — `flow_definitions` is still there, so the probe would +/// honour the cache and the next read of a migrated column would fail with +/// `no such column`. The `PRAGMA user_version` gate detects the drift. +#[test] +fn older_on_disk_schema_under_a_cached_path_is_remigrated() { + let tmp = TempDir::new().unwrap(); + let dir = test_dir(&tmp); + + // First use creates the full (versioned) schema and caches the path. + let original = create_flow( + &dir, + "v1".to_string(), + trigger_graph(), + true, // require_approval — the migrated column we'll drop below + true, + ) + .unwrap(); + assert!(original.require_approval); + + // 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 = dir.join("flows.db"); + { + let raw = rusqlite::Connection::open(&db_path).unwrap(); + raw.execute_batch( + "ALTER TABLE flow_definitions DROP COLUMN require_approval; + PRAGMA user_version = 0;", + ) + .unwrap(); + } + + // The path is still cached. With the old single-table `sqlite_master` probe + // this database would be trusted and `list_flows` (which selects + // `require_approval`) would fail with `no such column`. The version check + // detects the drift and re-migrates via the idempotent `init_schema`. + let (listed, skipped) = list_flows(&dir) + .expect("an older on-disk schema under a cached path must be re-migrated, not trusted"); + assert_eq!(skipped, 0); + 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, reading its default for the pre-existing row. + let reloaded = get_flow(&dir, &original.id).unwrap().unwrap(); + assert!( + !reloaded.require_approval, + "the re-added column defaults to 0 for the pre-existing row" + ); + + // And the store is fully usable again, not merely readable. + let recreated = create_flow(&dir, "v2".to_string(), trigger_graph(), false, true) + .expect("writes must work against the re-migrated schema"); + assert_ne!(recreated.id, original.id); + let (flows_final, _) = list_flows(&dir).unwrap(); + assert_eq!(flows_final.len(), 2); +}