`TableInfo` is cached permanently and never invalidated after DDL, producing wrong or silently missing change notifications

[jjhafer]

Summary

ObservationBroker caches a TableInfo (primary-key column indices and a WITHOUT ROWID flag) per table, and never refreshes it. ensure_table_info only queries tables that have no cached entry:

// crates/sqlx-sqlite-observer/src/conn_mgr.rs
let tables_to_query: Vec<String> = observed
   .into_iter()
   .filter(|table| self.broker.get_table_info(table).is_none())
   .collect();

set_table_info is called from nowhere else, so once a table's info is cached it is fixed for the lifetime of the broker. Any DDL that changes column positions leaves those cached indices pointing at the wrong columns.

There are two distinct failure modes, and neither is visible to a subscriber:

1. The index still lands in bounds → a different column's value is reported as the primary key.
2. The index falls out of bounds → extract_primary_key returns SchemaMismatch, on_commit logs it and drops the change entirely, so the subscriber is never told about a write that committed.

Mode 2's only trace is an error! log, and every crate in this workspace pins tracing with release_max_level_off (keyed on debug_assertions), so in a release build the notification vanishes with no diagnostic at all.

Reproduction

Confirmed against master (a5a4364). Both scenarios below are from a single test run; the printed output is the bug.

#![cfg(feature = "observer")]
use std::time::Duration;
use sqlx_sqlite_observer::ObserverConfig;
use sqlx_sqlite_toolkit::DatabaseWrapper;
use tempfile::TempDir;
use tokio::time::timeout;

async fn drain(rx: &mut tokio::sync::broadcast::Receiver<sqlx_sqlite_observer::TableChange>)
   -> Vec<String> {
   let mut out = vec![];
   while let Ok(Ok(c)) = timeout(Duration::from_millis(150), rx.recv()).await {
      out.push(format!("table={} pk={:?} rowid={:?}", c.table, c.primary_key, c.rowid));
   }
   out
}

#[tokio::test]
async fn tableinfo_goes_stale_after_ddl() {
   let temp = TempDir::new().unwrap();
   let db = DatabaseWrapper::connect(&temp.path().join("ddl.db"), None).await.unwrap();

   // --- mode 1: table recreated with the PK in a different position ---
   db.execute("CREATE TABLE t (a TEXT, id INTEGER PRIMARY KEY)".into(), vec![])
      .execute().await.unwrap();
   db.enable_observation(ObserverConfig::new().with_tables(["t"]));
   let mut rx = db.observable().unwrap().subscribe(["t"]);

   db.execute("INSERT INTO t (a) VALUES ('warm')".into(), vec![]).execute().await.unwrap();
   println!("[1] baseline (pk really at index 1): {:?}", drain(&mut rx).await);

   db.execute("DROP TABLE t".into(), vec![]).execute().await.unwrap();
   db.execute("CREATE TABLE t (id INTEGER PRIMARY KEY, a TEXT)".into(), vec![])
      .execute().await.unwrap();
   let _ = drain(&mut rx).await;

   db.execute("INSERT INTO t (id, a) VALUES (99, 'NOT_THE_PK')".into(), vec![])
      .execute().await.unwrap();
   println!("[1] after recreate: {:?}", drain(&mut rx).await);

   // --- mode 2: ALTER TABLE DROP COLUMN shifts the cached index out of bounds ---
   db.execute("CREATE TABLE u (x TEXT, y TEXT, id INTEGER PRIMARY KEY)".into(), vec![])
      .execute().await.unwrap();
   db.enable_observation(ObserverConfig::new().with_tables(["u"]));
   let mut rx2 = db.observable().unwrap().subscribe(["u"]);
   db.execute("INSERT INTO u (x,y) VALUES ('a','b')".into(), vec![]).execute().await.unwrap();
   println!("[2] baseline (pk really at index 2): {:?}", drain(&mut rx2).await);

   db.execute("ALTER TABLE u DROP COLUMN x".into(), vec![]).execute().await.unwrap();
   let _ = drain(&mut rx2).await;
   db.execute("INSERT INTO u (y) VALUES ('c')".into(), vec![]).execute().await.unwrap();
   println!("[2] after DROP COLUMN: {:?}", drain(&mut rx2).await);
}

Observed output:

[1] baseline (pk really at index 1): ["table=t pk=[Integer(1)] rowid=Some(1)"]
[1] after recreate: ["table=t pk=[Text(\"NOT_THE_PK\")] rowid=Some(99)"]
[2] baseline (pk really at index 2): ["table=u pk=[Integer(1)] rowid=Some(1)"]
[2] after DROP COLUMN: []

Reading those two results:

• [1] — the real primary key of the inserted row is 99. The observer reports the text "NOT_THE_PK" as the primary key, because the cached index 1 now points at column a instead of id. A consumer keying off primary_key gets a value that isn't a key at all and doesn't identify the row. (rowid happens to be correct here, so a consumer using rowid instead is unaffected in this particular case.)
• [2] — the insert committed, and the subscriber received nothing. The cached index 2 is out of bounds for the now-2-column table, so the change was dropped.

Expected

Cached TableInfo should be invalidated when the schema changes, so notifications after DDL either carry the correct primary key or are re-derived from the current schema. A committed write to an observed table should never produce a silently dropped notification.

Affected beyond primary_key

TableInfo::without_rowid is cached the same way and feeds the decision about whether rowid is meaningful. A table recreated from a rowid table to WITHOUT ROWID (or the reverse) keeps the stale flag, so rowid is reported as meaningful when it isn't, or suppressed when it is.

Notes on scope

• This is not introduced by the database-wide observation work — it reproduces on a single database with nothing attached, and predates it.
• That work does raise the impact, for the same reason it raises ROLLBACK TO SAVEPOINT publishes phantom change notifications for writes that were never kept #60's: observation is now a property of the database rather than of one handle, so one caller's migration corrupts the primary keys seen by every subscriber to that database.
• Realistic trigger: run_migrations() is the ordinary way DDL reaches a database in this plugin, and nothing about it invalidates the cache. A long-lived app that observes a table and later applies a migration recreating it hits mode 1 or 2 with no warning.

Possible directions

Not a recommendation — just what the surface allows:

1. Gate the cache on PRAGMA schema_version. SQLite increments it on every schema change, so ensure_table_info could read it per acquisition and clear the table_info map when it differs from the value the cache was built under. This is the most complete option; note each schema (main and each attached alias) has its own schema_version, which needs care now that hook routing is per-schema.
2. Store the column count in TableInfo and treat a mismatch against sqlite3_preupdate_count as "stale, re-query." Cheap and self-correcting, but only catches mode 2 — mode 1 above keeps the same column count, so this would not detect it.
3. Invalidate explicitly after migrations. Having run_migrations() clear the cache is a small, targeted mitigation that covers the most common path, but it misses ad-hoc DDL issued through execute(), as both reproductions above do.

Option 1 looks like the real fix; option 3 is worth considering regardless as cheap insurance.

Related

• ROLLBACK TO SAVEPOINT publishes phantom change notifications for writes that were never kept #60 — ROLLBACK TO SAVEPOINT publishes phantom notifications. Different mechanism (buffer lifecycle rather than schema cache), but the same theme: a notification that does not correspond to committed state.