`ROLLBACK TO SAVEPOINT` publishes phantom change notifications for writes that were never kept

[jjhafer]

Summary

Change observation buffers preupdate events and flushes them on SQLite's commit hook. ROLLBACK TO SAVEPOINT does not fire sqlite3_rollback_hook, so events captured between the SAVEPOINT and the ROLLBACK TO stay in the broker's buffer and are published when the surrounding transaction eventually commits.

Subscribers therefore receive notifications for rows that do not exist. This is reachable from JavaScript — execute() accepts SAVEPOINT / ROLLBACK TO statements — and does not require attached databases or any Rust-side API.

Reproduction

Confirmed against master (a5a4364). This test passes, and its 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;

#[tokio::test]
async fn savepoint_rollback_publishes_phantom_change() {
   let temp = TempDir::new().unwrap();
   let path = temp.path().join("sp.db");
   let db = DatabaseWrapper::connect(&path, None).await.unwrap();
   db.execute("CREATE TABLE t (id INTEGER PRIMARY KEY, v TEXT)".into(), vec![])
      .execute().await.unwrap();

   db.enable_observation(ObserverConfig::new().with_tables(["t"]));
   let mut rx = db.observable().unwrap().subscribe(["t"]);

   {
      let mut w = db.acquire_writer().await.unwrap();
      sqlx::query("BEGIN").execute(&mut *w).await.unwrap();
      sqlx::query("INSERT INTO t (v) VALUES ('kept')").execute(&mut *w).await.unwrap();
      sqlx::query("SAVEPOINT sp").execute(&mut *w).await.unwrap();
      sqlx::query("INSERT INTO t (v) VALUES ('rolled_back')").execute(&mut *w).await.unwrap();
      sqlx::query("ROLLBACK TO sp").execute(&mut *w).await.unwrap();
      sqlx::query("COMMIT").execute(&mut *w).await.unwrap();
   }

   let mut got = vec![];
   while let Ok(Ok(c)) = timeout(Duration::from_millis(150), rx.recv()).await {
      got.push(format!("{:?}", c.new_values));
   }
   let rows = db.fetch_all("SELECT v FROM t ORDER BY id".into(), vec![])
      .execute().await.unwrap();

   println!("NOTIFICATIONS: {}", got.len());
   for g in &got { println!("  {}", g); }
   println!("ROWS IN TABLE: {:?}", rows);
}

Observed output:

NOTIFICATIONS: 2
  Some([Integer(1), Text("kept")])
  Some([Integer(2), Text("rolled_back")])
ROWS IN TABLE: [{"v": String("kept")}]

Two notifications, one row. rolled_back was never committed, but every subscriber to t was told it was.

Expected

Only the kept insert should be published. Events captured after a SAVEPOINT that is later rolled back should be discarded, matching the behavior of a full ROLLBACK — which does fire sqlite3_rollback_hook and correctly clears the buffer.

Cause

ObservationBroker holds a single flat Vec<PreUpdateEvent> buffer with exactly two lifecycle transitions, both driven by native hooks in crates/sqlx-sqlite-observer/src/hooks.rs:

• commit_callback → on_commit() — drains the buffer and publishes
• rollback_callback → on_rollback() — discards the buffer

SQLite invokes the rollback hook only for a rollback of the outermost transaction. A statement-level or savepoint-level rollback does not reach it, so there is no signal that a subset of the buffer should be dropped. The buffer has no notion of nesting, so even with such a signal there is currently no way to express "discard everything after mark N."

Scope

This is independent of the observation-sharing work in #53 and its attached-database routing — it reproduces on a single database with nothing attached, and predates that changeset. It is filed separately because it lives on the same buffer-lifecycle surface, so a fix should be designed with that code in view.

Worth noting that #53 raises the impact: observation is now database-wide, so one caller's savepoint rollback publishes phantom events to every subscriber of that database rather than to a single handle's own subscribers.

Possible directions

Not a recommendation, just what the surface allows:

1. Track savepoint depth in the broker. Record a buffer high-water mark on SAVEPOINT and truncate to it on ROLLBACK TO. Requires the broker to observe savepoint statements, which no hook reports — it would mean intercepting SQL text at the writer boundary, which is fragile (comments, casing, nested and reused savepoint names, RELEASE).
2. Use sqlite3_preupdate_depth() to record each event's nesting depth and discard events above the target depth on a savepoint rollback. Still needs a signal that a savepoint rollback happened.
3. Document savepoints as unsupported under observation and reject or warn on SAVEPOINT / ROLLBACK TO when observation is enabled. Cheapest and honest; loses partial-rollback capability for observed databases.

Option 3 may be the right near-term answer given that no native hook reports savepoint boundaries at all.

Related

• Also unfixed and adjacent: TableInfo is cached per table and never invalidated after DDL, so ALTER TABLE that adds or reorders columns leaves pk_columns stale and extract_primary_key reports the wrong column. Separate issue.