diff --git a/CHANGELOG.md b/CHANGELOG.md index 08feb3dc..b23590c6 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,6 +5,7 @@ ### Changes - `docs/releasing.md`: how a compose-only change ships (as a release), and how to read the dense promote watch. Details: TC-950. +- A SQL request that writes is one transaction: an `execute` (with its `schema` statements), a `batch`, a named statement, and a `query` that writes (`WITH … INSERT`) or carries a `PRAGMA` (admin only). If any statement fails, or a deferred constraint fails at the commit, none of them is kept: the local file is unchanged and matches the durable copy in Postgres, and the request costs no Postgres round trip. This holds for a single statement too, including `INSERT OR FAIL` and a column constraint with `ON CONFLICT FAIL`. Telemetry: such a request counts two more native SQL calls (the `SAVEPOINT` and its `RELEASE` or `ROLLBACK`). No migration. Details: TC-1491. ## [1.21.7] - 2026-10-21 diff --git a/docs/request-telemetry.md b/docs/request-telemetry.md index 6495e540..7768584b 100644 --- a/docs/request-telemetry.md +++ b/docs/request-telemetry.md @@ -90,7 +90,7 @@ SQLx query timing follows the execution/logger lifetime, including network I/O a | Quota/admin | In-memory overrides and size mirrors do not invent queries; usage/hosted-space enumeration and any actual pool work are counted | | SQL/DuckDB identity/fence/alias resolution and durable artifact load/save/CAS/delta | Shared pool, including PostgreSQL bulk scopes; cache hits perform only their actual work | | Request-awaited read audit | Separate measured batch context, fanned out before durable receipts; details below | -| User SQL rusqlite actor | Context on every work message; actual statement/schema/batch/growth-control SQL, prepare/constraint errors, cold-open PRAGMAs, size/promotion/checkpoint SQL | +| User SQL rusqlite actor | Context on every work message; actual statement/schema/batch/growth-control SQL (a multi-statement request also counts the `SAVEPOINT` and `RELEASE`/`ROLLBACK` of its request transaction, TC-1491), prepare/constraint errors, cold-open PRAGMAs, size/promotion/checkpoint SQL | | SQL publication | The current withdrawn publication handler executes no SQL; it is not an uninstrumented database path | | Optional DuckDB actor | Context on every message and the initiating cold-open worker; JSON/Arrow queries, schema/batch/transaction/growth/catalog SQL, locked security settings and promotion/copy/describe SQL | | DuckDB import/cold export blocking workers | Explicit worker context for guarded admission/validation/checkpoint/copy SQL; artifact repository work is measured separately by SQLx | diff --git a/tinycloud-core/src/sql/database.rs b/tinycloud-core/src/sql/database.rs index 1efdc553..0d6bc364 100644 --- a/tinycloud-core/src/sql/database.rs +++ b/tinycloud-core/src/sql/database.rs @@ -269,26 +269,36 @@ pub fn spawn_actor( ), }; conn.authorizer(None::) -> Authorization>); + inject_open_transaction(&conn, &space_id, &db_name); + // A request ends its transaction, by commit or rollback. + // One still open here (a rollback that itself failed) + // would swallow the next request's writes, so serve no + // further request on this connection. + // // TC-1586: the connection is shared by every request on // this database. After any request in which the // authorizer allowed a PRAGMA outside the plain schema // reports, reset it; if that fails, serve no further - // request on this connection (the next one rebuilds it). - if pragmas.load(Ordering::Relaxed) + // request on this connection either (the next one + // rebuilds it). + let retire = if !conn.is_autocommit() { + Some("SQL connection transaction could not be ended") + } else if pragmas.load(Ordering::Relaxed) && connection_reset_failed(&conn, &mode, &space_id, &db_name) { - tracing::error!( - "SQL connection state could not be reset; retiring the database actor" - ); + Some("SQL connection state could not be reset") + } else { + None + }; + if let Some(reason) = retire { + tracing::error!("{reason}; retiring the database actor"); // Deregister first, so the request that follows the // error builds a new actor instead of finding this one. databases.remove_if(&(space_id.clone(), db_name.clone()), |_, handle| { handle.id == id }); - let _ = response_tx.send(Err(SqlError::Internal( - "SQL connection state could not be reset".into(), - ))); + let _ = response_tx.send(Err(SqlError::Internal(reason.into()))); break None; } @@ -490,6 +500,50 @@ fn page_count(conn: &rusqlite::Connection) -> Result { .map_err(|e| SqlError::Internal(e.to_string())) } +const REQUEST_BEGIN: &str = "SAVEPOINT tinycloud_request"; +const REQUEST_KEEP: &str = "RELEASE tinycloud_request"; + +/// Whether `request` runs in a transaction of its own (TC-1491). +/// +/// Every request that can write does, so it is all-or-nothing whatever its +/// conflict clauses say: a statement that stops with `OR FAIL` or +/// `ON CONFLICT FAIL` keeps the rows it changed first when it runs on its +/// own, and a request with several statements keeps its earlier ones. +/// A request that carries a `PRAGMA` (admin only) is wrapped like any other. +/// Every pragma the authorizer admits is transactional (`user_version` and +/// `application_id` are undone by the rollback; the rest are reports), and +/// the authorizer's allow-list is what keeps connection-level settings out of +/// the transaction. The actor's TC-1586 connection reset runs after the +/// transaction has ended. +/// +/// Whether a query writes is `parser::classify_sql` (TC-1585), which sees +/// writes nested in a query tree: a `WITH … INSERT` is wrapped like any other +/// write. It does not depend on the caller's abilities (a read-tier grant is +/// only a shortcut: its writes are refused anyway), and a statement that does +/// not parse is left to `handle_request_authorized` to report. This is the one +/// place that decides it; `Execute`, `Batch` and `ExecuteStatement` are wrapped +/// whatever they contain, a read in a batch included. +fn needs_transaction( + request: &SqlRequest, + caveats: &Option, + authority: &SqlAuthority, +) -> bool { + match request { + SqlRequest::Query { sql, .. } => { + !authority.is_read_tier_only() + && (parser::is_pragma_sql(sql) + || parser::classify_sql(sql) + .map(|class| !class.is_read_only) + .unwrap_or(false)) + } + SqlRequest::Execute { .. } | SqlRequest::Batch { .. } => true, + SqlRequest::ExecuteStatement { name, .. } => caveats + .as_ref() + .is_some_and(|c| c.find_statement(name).is_some()), + SqlRequest::Export => false, + } +} + /// `DdlPolicy::Log` bookkeeping for one request: count every statement that /// only `write` let through, and warn once. The message is constant and the /// counter is labelled by statement kind alone, so neither carries SQL text, @@ -528,12 +582,76 @@ fn handle_message( ) } +/// Run `request` as one transaction when it needs one: every statement is +/// kept or none is. +/// +/// The transaction is a `SAVEPOINT`, which opens one when none is active; +/// users cannot issue transaction control themselves, the authorizer denies +/// it. A failure at any statement, or at the commit (a deferred constraint), +/// is a `ROLLBACK`: the local file is unchanged and matches the durable copy, +/// which is only written after a fully successful request. See +/// [`needs_transaction`] for what is wrapped. The wrap sits inside the request +/// and outside the actor's TC-1586 connection reset, which runs after the +/// transaction has ended. +/// +/// When a transaction is already open, [`handle_message_without_growth_authorized`]'s +/// guard has it and already undoes the whole request on any error (reporting +/// growth ahead of the SQL error), so this adds nothing there. +/// +/// If the transaction is still open after the final `ROLLBACK`, the actor +/// retires the connection instead of serving another request on it. fn handle_message_authorized( conn: &rusqlite::Connection, request: &SqlRequest, caveats: &Option, authority: &SqlAuthority, pragmas: &PragmaTracker, +) -> Result { + if !conn.is_autocommit() || !needs_transaction(request, caveats, authority) { + return handle_request_authorized(conn, request, caveats, authority, pragmas); + } + { + let _timer = QueryTimer::new(); + conn.execute_batch(REQUEST_BEGIN) + .map_err(|e| SqlError::Internal(e.to_string()))?; + } + let result = handle_request_authorized(conn, request, caveats, authority, pragmas); + // An early error return can leave the request's authorizer installed. + conn.authorizer(None::) -> Authorization>); + // SQLite can already have undone the transaction (ROLLBACK conflict + // handling); there is then nothing left to undo. + if result.is_err() && conn.is_autocommit() { + return result; + } + // A failure rolls the whole transaction back. `ROLLBACK TO` + `RELEASE` + // would instead commit an empty write transaction and rewrite the header. + let end = if result.is_ok() { + REQUEST_KEEP + } else { + "ROLLBACK" + }; + let end_result = { + let _timer = QueryTimer::new(); + conn.execute_batch(end) + }; + if let Err(error) = end_result { + // RELEASE of the outermost savepoint is the COMMIT. A deferred + // constraint can fail it and leave the writes and the savepoint open. + if !conn.is_autocommit() { + let _timer = QueryTimer::new(); + let _ = conn.execute_batch("ROLLBACK"); + } + return result.and(Err(SqlError::Sqlite(error.to_string()))); + } + result +} + +fn handle_request_authorized( + conn: &rusqlite::Connection, + request: &SqlRequest, + caveats: &Option, + authority: &SqlAuthority, + pragmas: &PragmaTracker, ) -> Result { let mut ddl_log = DdlLog::default(); let authorizer = |parsed: &parser::ParsedQuery| { @@ -762,6 +880,27 @@ fn injected_reset_failure(_space_id: &str, _db_name: &str) -> bool { false } +/// Test seam: databases whose connection is left inside a transaction after +/// their next request, as a failed final `ROLLBACK` would leave it. +#[cfg(test)] +pub(crate) static OPEN_TRANSACTIONS: std::sync::Mutex> = + std::sync::Mutex::new(Vec::new()); + +#[cfg(test)] +fn inject_open_transaction(conn: &rusqlite::Connection, space_id: &str, db_name: &str) { + let injected = OPEN_TRANSACTIONS + .lock() + .unwrap() + .iter() + .any(|(space, db)| space == space_id && db == db_name); + if injected { + conn.execute_batch("BEGIN").unwrap(); + } +} + +#[cfg(not(test))] +fn inject_open_transaction(_conn: &rusqlite::Connection, _space_id: &str, _db_name: &str) {} + fn is_insert_statement(parsed: &parser::ParsedQuery) -> bool { matches!( parsed.statements.as_slice(), @@ -1380,6 +1519,424 @@ mod publication_rollback_tests { "Original" ); } + + // TC-1491: a request with several statements is all-or-nothing. + + fn exec(sql: &str) -> SqlRequest { + SqlRequest::Execute { + schema: None, + sql: sql.into(), + params: vec![], + } + } + + fn batch(statements: &[&str]) -> SqlRequest { + SqlRequest::Batch { + statements: statements + .iter() + .map(|sql| SqlStatement { + sql: (*sql).into(), + params: vec![], + }) + .collect(), + } + } + + async fn run( + actor: &DatabaseHandle, + request: SqlRequest, + growth: GrowthPolicy, + ) -> Result { + actor + .execute(request, None, "tinycloud.sql/admin".into(), growth) + .await + } + + async fn scalar(actor: &DatabaseHandle, sql: &str) -> String { + let result = run( + actor, + SqlRequest::Query { + sql: sql.into(), + params: vec![], + max_rows: None, + max_bytes: None, + }, + GrowthPolicy::Allow, + ) + .await + .unwrap(); + match result.response { + SqlResponse::Query(query) => match &query.rows[0][0] { + SqlValue::Integer(value) => value.to_string(), + SqlValue::Text(value) => value.clone(), + other => format!("{other:?}"), + }, + other => panic!("expected a query response, got {other:?}"), + } + } + + fn spawn_test_actor(dir: &tempfile::TempDir, memory_threshold: u64) -> DatabaseHandle { + spawn_actor( + "space".into(), + "atomic".into(), + dir.path().to_str().unwrap().into(), + memory_threshold, + Arc::new(DashMap::new()), + ) + } + + #[tokio::test] + async fn tc1491_failed_batch_leaves_the_local_file_unchanged() { + for memory_threshold in [u64::MAX, 0] { + for growth in [GrowthPolicy::Allow, GrowthPolicy::Forbid] { + let dir = tempfile::tempdir().unwrap(); + let actor = spawn_test_actor(&dir, memory_threshold); + run( + &actor, + exec("CREATE TABLE t (id INTEGER PRIMARY KEY, v TEXT NOT NULL)"), + GrowthPolicy::Allow, + ) + .await + .unwrap(); + run( + &actor, + exec("INSERT INTO t (v) VALUES ('keep')"), + GrowthPolicy::Allow, + ) + .await + .unwrap(); + let before = actor.checkpoint().await.unwrap(); + + // Two statements apply, the third violates NOT NULL. + let failed = run( + &actor, + batch(&[ + "UPDATE t SET v = 'changed'", + "INSERT INTO t (v) VALUES ('prefix')", + "INSERT INTO t (v) VALUES (NULL)", + ]), + growth, + ) + .await; + assert!(matches!(failed, Err(SqlError::Sqlite(_))), "{failed:?}"); + + assert_eq!(scalar(&actor, "SELECT count(*) FROM t").await, "1"); + assert_eq!(scalar(&actor, "SELECT v FROM t").await, "keep"); + // The `Forbid` guard predates this change and ends a rejected + // request with ROLLBACK TO + RELEASE, which commits an empty + // transaction and bumps the header's change counter; its + // content is checked above, its bytes only for `Allow`. + if growth == GrowthPolicy::Allow { + assert!( + actor.checkpoint().await.unwrap() == before, + "a rejected batch must not change a single byte of the checkpoint" + ); + } + } + } + } + + #[tokio::test] + async fn tc1491_failed_schema_request_leaves_no_schema_behind() { + for memory_threshold in [u64::MAX, 0] { + let dir = tempfile::tempdir().unwrap(); + let actor = spawn_test_actor(&dir, memory_threshold); + run( + &actor, + exec("CREATE TABLE base (id INTEGER)"), + GrowthPolicy::Allow, + ) + .await + .unwrap(); + let failed = run( + &actor, + SqlRequest::Execute { + schema: Some(vec![ + "CREATE TABLE extra (id INTEGER)".into(), + "CREATE INDEX extra_id ON extra (id)".into(), + ]), + sql: "INSERT INTO missing_table VALUES (1)".into(), + params: vec![], + }, + GrowthPolicy::Allow, + ) + .await; + assert!(failed.is_err()); + assert_eq!( + scalar( + &actor, + "SELECT count(*) FROM sqlite_master WHERE name IN ('extra', 'extra_id')" + ) + .await, + "0" + ); + } + } + + #[tokio::test] + async fn tc1491_successful_batch_and_ddl_batch_apply_in_full() { + let dir = tempfile::tempdir().unwrap(); + let actor = spawn_test_actor(&dir, 0); + let result = run( + &actor, + batch(&[ + "CREATE TABLE a (id INTEGER PRIMARY KEY, v TEXT)", + "CREATE INDEX a_v ON a (v)", + "INSERT INTO a (v) VALUES ('x')", + "INSERT INTO a (v) VALUES ('y')", + ]), + GrowthPolicy::Allow, + ) + .await + .unwrap(); + let SqlResponse::Batch(response) = result.response else { + panic!("expected a batch response"); + }; + assert_eq!(response.results.len(), 4); + assert_eq!(response.results[3].last_insert_row_id, Some(2)); + assert_eq!(scalar(&actor, "SELECT count(*) FROM a").await, "2"); + + // A failing DDL batch undoes its DDL too. + let failed = run( + &actor, + batch(&["CREATE TABLE b (id INTEGER)", "CREATE TABLE a (id INTEGER)"]), + GrowthPolicy::Allow, + ) + .await; + assert!(failed.is_err()); + assert_eq!( + scalar( + &actor, + "SELECT count(*) FROM sqlite_master WHERE name = 'b'" + ) + .await, + "0" + ); + } + + #[tokio::test] + async fn tc1491_user_transaction_control_is_still_denied() { + let dir = tempfile::tempdir().unwrap(); + let actor = spawn_test_actor(&dir, 0); + run(&actor, exec("CREATE TABLE t (v TEXT)"), GrowthPolicy::Allow) + .await + .unwrap(); + for control in [ + "BEGIN", + "COMMIT", + "ROLLBACK", + "END", + "SAVEPOINT mine", + "RELEASE tinycloud_request", + "ROLLBACK TO tinycloud_request", + ] { + for request in [ + batch(&["INSERT INTO t VALUES ('a')", control]), + batch(&[control, "INSERT INTO t VALUES ('a')"]), + SqlRequest::Execute { + schema: Some(vec![control.into()]), + sql: "INSERT INTO t VALUES ('a')".into(), + params: vec![], + }, + ] { + let result = run(&actor, request, GrowthPolicy::Allow).await; + assert!(result.is_err(), "{control} must be refused: {result:?}"); + assert_eq!( + scalar(&actor, "SELECT count(*) FROM t").await, + "0", + "{control} must not let a prefix commit" + ); + } + } + } + + /// A request that carries a `PRAGMA` is one transaction like any other: + /// a failure leaves neither its statements nor the header value it set, + /// and the connection is still served afterwards. + #[tokio::test] + async fn tc1491_pragma_requests_are_one_transaction() { + for memory_threshold in [u64::MAX, 0] { + let dir = tempfile::tempdir().unwrap(); + let actor = spawn_test_actor(&dir, memory_threshold); + run( + &actor, + exec("CREATE TABLE t (v TEXT NOT NULL)"), + GrowthPolicy::Allow, + ) + .await + .unwrap(); + let before = actor.checkpoint().await.unwrap(); + + // `user_version` is on the admin PRAGMA allow-list (TC-1586); + // `foreign_keys = OFF` is not. + for failing in [ + batch(&[ + "PRAGMA user_version = 7", + "INSERT INTO t VALUES ('prefix')", + "INSERT INTO t VALUES (NULL)", + ]), + batch(&[ + "INSERT INTO t VALUES ('prefix')", + "PRAGMA application_id = 9", + "INSERT INTO t VALUES (NULL)", + ]), + SqlRequest::Execute { + schema: Some(vec![ + "PRAGMA user_version = 7".into(), + "INSERT INTO t VALUES ('prefix')".into(), + ]), + sql: "INSERT INTO t VALUES (NULL)".into(), + params: vec![], + }, + ] { + let failed = run(&actor, failing, GrowthPolicy::Allow).await; + assert!(failed.is_err()); + assert_eq!(scalar(&actor, "SELECT count(*) FROM t").await, "0"); + assert_eq!(scalar(&actor, "PRAGMA user_version").await, "0"); + assert_eq!(scalar(&actor, "PRAGMA application_id").await, "0"); + assert!( + actor.checkpoint().await.unwrap() == before, + "a rejected request with a PRAGMA must not change a single byte" + ); + } + + // The reports an admin may run, in a request that writes: they + // return normally inside the transaction and the write is kept. + run( + &actor, + SqlRequest::Execute { + schema: Some( + [ + "PRAGMA journal_mode", + "PRAGMA foreign_keys", + "PRAGMA page_count", + "PRAGMA schema_version", + "PRAGMA freelist_count", + "PRAGMA data_version", + "PRAGMA table_info(t)", + "PRAGMA integrity_check", + "PRAGMA quick_check", + "PRAGMA foreign_key_check", + "PRAGMA user_version = 3", + ] + .map(String::from) + .to_vec(), + ), + sql: "INSERT INTO t VALUES ('kept')".into(), + params: vec![], + }, + GrowthPolicy::Allow, + ) + .await + .unwrap(); + assert_eq!(scalar(&actor, "SELECT count(*) FROM t").await, "1"); + assert_eq!(scalar(&actor, "PRAGMA user_version").await, "3"); + // A single PRAGMA request runs, and the connection keeps the + // state it opened with (the TC-1586 reset ran after the request). + assert_eq!(scalar(&actor, "PRAGMA foreign_keys").await, "1"); + } + } + + /// A write that stops with `OR FAIL` or `ON CONFLICT FAIL` keeps the rows + /// it changed first when it runs on its own; the request is one + /// transaction, so it keeps none of them, in memory and on disk. + #[test] + fn tc1491_a_single_write_with_fail_conflict_resolution_leaves_nothing() { + let dir = tempfile::tempdir().unwrap(); + let file = dir.path().join("fail.db"); + for mode in [StorageMode::InMemory, StorageMode::File(file.clone())] { + let conn = storage::open_connection(&mode).unwrap(); + // `ON CONFLICT FAIL` on a column is outside what the SQL parser + // accepts, so the schema is created on the connection directly. + conn.execute_batch( + "CREATE TABLE u (id INTEGER PRIMARY KEY, v TEXT UNIQUE ON CONFLICT FAIL); + CREATE TABLE n (id INTEGER PRIMARY KEY, v TEXT NOT NULL ON CONFLICT FAIL); + INSERT INTO u VALUES (10, 'x'), (11, 'y'), (12, 'z');", + ) + .unwrap(); + let bytes = || -> Vec { + match &mode { + StorageMode::File(path) => { + conn.query_row("PRAGMA wal_checkpoint(TRUNCATE)", [], |_| Ok(())) + .unwrap(); + std::fs::read(path).unwrap() + } + // In memory there is no file; the rows are the check. + StorageMode::InMemory => Vec::new(), + } + }; + let before = bytes(); + for sql in [ + "INSERT OR FAIL INTO u VALUES (1, 'a'), (2, 'b'), (1, 'dup')", + "UPDATE u SET v = 'same'", + "UPDATE OR FAIL u SET v = 'same'", + "INSERT INTO n VALUES (1, 'a'), (2, 'b'), (3, NULL)", + ] { + for request in [ + exec(sql), + SqlRequest::Query { + sql: sql.into(), + params: vec![], + max_rows: None, + max_bytes: None, + }, + batch(&[sql]), + ] { + let failed = handle_message(&conn, &request, &None, "tinycloud.sql/write"); + // `UPDATE OR FAIL` is outside what the SQL parser accepts: + // refused, and nothing changes either. + assert!(failed.is_err(), "{sql}: {failed:?}"); + let rows = |sql: &str| -> String { + conn.query_row(sql, [], |row| row.get(0)).unwrap() + }; + assert_eq!( + rows("SELECT count(*) || ':' || group_concat(v) FROM u"), + "3:x,y,z" + ); + assert_eq!(rows("SELECT count(*) || '' FROM n"), "0"); + assert!(conn.is_autocommit()); + assert!(bytes() == before, "{sql}: the file must not change"); + } + } + } + } + + /// A connection still inside a transaction after its request is retired: + /// the next request would otherwise run inside it. + #[tokio::test] + async fn tc1491_a_connection_left_in_a_transaction_retires_the_actor() { + let dir = tempfile::tempdir().unwrap(); + let key = ("space".to_string(), "leaks".to_string()); + for growth in [GrowthPolicy::Allow, GrowthPolicy::Forbid] { + let databases = Arc::new(DashMap::new()); + let actor = spawn_actor( + key.0.clone(), + key.1.clone(), + dir.path().to_str().unwrap().into(), + u64::MAX, + databases.clone(), + ); + databases.insert(key.clone(), actor.clone()); + run(&actor, exec("CREATE TABLE t (v TEXT)"), GrowthPolicy::Allow) + .await + .unwrap(); + + OPEN_TRANSACTIONS.lock().unwrap().push(key.clone()); + let result = run(&actor, exec("INSERT INTO t VALUES ('x')"), growth).await; + OPEN_TRANSACTIONS.lock().unwrap().retain(|k| *k != key); + assert!(matches!(result, Err(SqlError::Internal(_))), "{result:?}"); + assert!( + !databases.contains_key(&key), + "the retired actor must leave the registry" + ); + assert!( + run(&actor, exec("INSERT INTO t VALUES ('y')"), growth) + .await + .is_err(), + "a retired actor serves no further request" + ); + } + } } /// TC-1581: schema-changing statements need `schema`; `write` is data only. diff --git a/tinycloud-core/src/sql/service.rs b/tinycloud-core/src/sql/service.rs index f9c634e4..0f826b5c 100644 --- a/tinycloud-core/src/sql/service.rs +++ b/tinycloud-core/src/sql/service.rs @@ -1289,24 +1289,24 @@ mod tests { .unwrap() .unwrap(); - // Unguarded requests preserve partial-batch semantics. The actor can - // therefore contain more pages than its last durable checkpoint. - let failed = service + // A failing request no longer leaves a committed prefix (TC-1491), + // so put the actor ahead of its last durable checkpoint directly: + // the durable-size comparison must still hold against live state + // that was never persisted. + let actor = service.handle(&space, "main").await.unwrap(); + actor .execute( - &space, - "main", SqlRequest::Execute { - schema: Some(vec![ - "INSERT INTO data VALUES (2, hex(zeroblob(65536)))".into() - ]), - sql: "INSERT INTO missing_table VALUES (1)".into(), + schema: None, + sql: "INSERT INTO data VALUES (2, hex(zeroblob(65536)))".into(), params: vec![], }, None, "tinycloud.sql/admin".into(), + GrowthPolicy::Allow, ) - .await; - assert!(matches!(failed, Err(SqlError::Sqlite(_)))); + .await + .unwrap(); let rejected = service .execute_without_growth( &space, @@ -4126,7 +4126,7 @@ mod tests { params: vec![SqlValue::Text("x".repeat(256 * 1024))], }, None, - "tinycloud.sql/write".into(), + "tinycloud.sql/admin".into(), ) .await; assert!(matches!(rejected, Err(SqlError::StorageWouldGrow))); @@ -4161,7 +4161,7 @@ mod tests { params: vec![], }, None, - "tinycloud.sql/write".into(), + "tinycloud.sql/admin".into(), ) .await .unwrap(); @@ -4254,4 +4254,452 @@ mod tests { } found } + + /// Counts every repository call, by kind. + struct CallCountingRepository { + inner: Arc, + calls: Arc>>, + } + + impl CallCountingRepository { + fn note(&self, call: &'static str) { + self.calls.lock().unwrap().push(call); + } + } + + #[async_trait] + impl DatabaseArtifactRepository for CallCountingRepository { + async fn load_meta( + &self, + service: &str, + space: &str, + name: &str, + ) -> Result, DatabaseArtifactError> { + self.note("load_meta"); + self.inner.load_meta(service, space, name).await + } + + async fn read_chunks( + &self, + service: &str, + space: &str, + name: &str, + meta: &ArtifactMeta, + sink: &mut dyn ArtifactSink, + ) -> Result<(), DatabaseArtifactError> { + self.note("read_chunks"); + self.inner + .read_chunks(service, space, name, meta, sink) + .await + } + + async fn save( + &self, + service: &str, + space: &str, + name: &str, + payload: Vec, + expected: ArtifactExpectation, + ) -> Result { + self.note("save"); + self.inner + .save(service, space, name, payload, expected) + .await + } + + async fn save_delta( + &self, + service: &str, + space: &str, + name: &str, + payload: Vec, + expected: ArtifactExpectation, + ) -> Result { + self.note("save_delta"); + self.inner + .save_delta(service, space, name, payload, expected) + .await + } + } + + fn atomic_batch(statements: &[&str]) -> SqlRequest { + SqlRequest::Batch { + statements: statements + .iter() + .map(|sql| SqlStatement { + sql: (*sql).into(), + params: Vec::new(), + }) + .collect(), + } + } + + async fn rows_of(service: &SqlService, space: &SpaceId, sql: &str) -> Vec> { + let result = service + .execute( + space, + "main", + SqlRequest::Query { + sql: sql.into(), + params: Vec::new(), + max_rows: None, + max_bytes: None, + }, + None, + "tinycloud.sql/read".into(), + ) + .await + .unwrap(); + match result.response { + SqlResponse::Query(query) => query.rows, + other => panic!("expected a query response, got {other:?}"), + } + } + + /// TC-1491: a failing batch leaves the local file unchanged, writes + /// nothing to Postgres, and the local file still equals the durable + /// artifact after the next successful write. + #[tokio::test] + async fn tc1491_failed_batch_matches_what_the_durable_artifact_holds() { + for memory_threshold in [u64::MAX, 0] { + let repo = artifact_repository().await; + let calls = Arc::new(std::sync::Mutex::new(Vec::new())); + let cache = TempDir::new().unwrap(); + let space = test_space_id("sql-atomic-batch"); + let service = SqlService::new( + cache.path().to_string_lossy().into_owned(), + memory_threshold, + Arc::new(CallCountingRepository { + inner: repo.clone(), + calls: calls.clone(), + }), + ); + let write = |sql: &'static str| { + service.execute( + &space, + "main", + SqlRequest::Execute { + schema: None, + sql: sql.into(), + params: Vec::new(), + }, + None, + "tinycloud.sql/admin".into(), + ) + }; + write("CREATE TABLE t (id INTEGER PRIMARY KEY, v TEXT NOT NULL)") + .await + .unwrap(); + write("INSERT INTO t (v) VALUES ('durable')").await.unwrap(); + let handle = service.handle(&space, "main").await.unwrap(); + let before = handle.checkpoint().await.unwrap(); + + // The failing request: no repository call at all, no local change. + calls.lock().unwrap().clear(); + let failed = service + .execute( + &space, + "main", + atomic_batch(&[ + "INSERT INTO t (v) VALUES ('prefix-1')", + "UPDATE t SET v = 'prefix-2'", + "INSERT INTO t (v) VALUES (NULL)", + ]), + None, + "tinycloud.sql/write".into(), + ) + .await; + assert!(matches!(failed, Err(SqlError::Sqlite(_))), "{failed:?}"); + assert!( + calls.lock().unwrap().is_empty(), + "a rejected batch costs no repository (Postgres) round trip: {:?}", + calls.lock().unwrap() + ); + assert_eq!(handle.checkpoint().await.unwrap(), before); + + // A successful 3-statement batch costs exactly the one persist + // call a single statement costs. + calls.lock().unwrap().clear(); + service + .execute( + &space, + "main", + atomic_batch(&[ + "INSERT INTO t (v) VALUES ('next-1')", + "INSERT INTO t (v) VALUES ('next-2')", + "INSERT INTO t (v) VALUES ('next-3')", + ]), + None, + "tinycloud.sql/write".into(), + ) + .await + .unwrap(); + let persisted = calls.lock().unwrap().clone(); + assert_eq!(persisted.len(), 1, "{persisted:?}"); + assert!( + matches!(persisted[0], "save" | "save_delta"), + "{persisted:?}" + ); + + // What a cold node rebuilds from the repository is the local state. + let local = rows_of(&service, &space, "SELECT id, v FROM t ORDER BY id").await; + assert_eq!(local.len(), 4); + assert!(!format!("{local:?}").contains("prefix")); + let cold_cache = TempDir::new().unwrap(); + let cold = SqlService::new( + cold_cache.path().to_string_lossy().into_owned(), + memory_threshold, + repo, + ); + assert_eq!( + rows_of(&cold, &space, "SELECT id, v FROM t ORDER BY id").await, + local + ); + } + } + /// TC-1491 + TC-1585: a `WITH … INSERT` is a write, so in a batch it is + /// one of the statements that must be kept or undone with the rest. A + /// later failure leaves nothing of it locally, writes nothing to Postgres, + /// and a cold node rebuilt from the durable artifact agrees. + #[tokio::test] + async fn tc1491_nested_write_cte_in_a_failed_batch_leaves_nothing() { + for memory_threshold in [u64::MAX, 0] { + let repo = artifact_repository().await; + let calls = Arc::new(std::sync::Mutex::new(Vec::new())); + let cache = TempDir::new().unwrap(); + let space = test_space_id("sql-atomic-cte"); + let service = SqlService::new( + cache.path().to_string_lossy().into_owned(), + memory_threshold, + Arc::new(CallCountingRepository { + inner: repo.clone(), + calls: calls.clone(), + }), + ); + let run = |request: SqlRequest| { + service.execute(&space, "main", request, None, "tinycloud.sql/admin".into()) + }; + let batch = |statements: &[&str]| SqlRequest::Batch { + statements: statements + .iter() + .map(|sql| SqlStatement { + sql: (*sql).into(), + params: Vec::new(), + }) + .collect(), + }; + run(SqlRequest::Execute { + schema: None, + sql: "CREATE TABLE t (id INTEGER PRIMARY KEY, v TEXT NOT NULL)".into(), + params: Vec::new(), + }) + .await + .unwrap(); + let handle = service.handle(&space, "main").await.unwrap(); + let before = handle.checkpoint().await.unwrap(); + + calls.lock().unwrap().clear(); + let failed = run(batch(&[ + "WITH c AS (SELECT 'cte-insert' AS v) INSERT INTO t (v) SELECT v FROM c", + "INSERT INTO t (v) VALUES (NULL)", + ])) + .await; + assert!(matches!(failed, Err(SqlError::Sqlite(_))), "{failed:?}"); + assert!( + calls.lock().unwrap().is_empty(), + "a rejected batch costs no repository (Postgres) round trip: {:?}", + calls.lock().unwrap() + ); + assert_eq!(handle.checkpoint().await.unwrap(), before); + assert!(rows_of(&service, &space, "SELECT v FROM t") + .await + .is_empty()); + + // The same CTE write kept when the batch succeeds, and durable. + run(batch(&[ + "WITH c AS (SELECT 'cte-insert' AS v) INSERT INTO t (v) SELECT v FROM c", + "INSERT INTO t (v) VALUES ('plain')", + ])) + .await + .unwrap(); + let local = rows_of(&service, &space, "SELECT id, v FROM t ORDER BY id").await; + assert_eq!(local.len(), 2); + let cold_cache = TempDir::new().unwrap(); + let cold = SqlService::new( + cold_cache.path().to_string_lossy().into_owned(), + memory_threshold, + repo, + ); + assert_eq!( + rows_of(&cold, &space, "SELECT id, v FROM t ORDER BY id").await, + local + ); + } + } + + /// TC-1491: a single write is all-or-nothing whatever its conflict clause + /// says. A statement that stops with `OR FAIL` or `ON CONFLICT FAIL` + /// leaves the rows it changed first when it runs on its own; the request + /// is one transaction, so it leaves none of them: the local file is + /// unchanged and no repository (Postgres) call is made. + #[tokio::test] + async fn tc1491_a_single_write_with_fail_conflict_resolution_leaves_nothing() { + for memory_threshold in [u64::MAX, 0] { + let repo = artifact_repository().await; + let calls = Arc::new(std::sync::Mutex::new(Vec::new())); + let cache = TempDir::new().unwrap(); + let space = test_space_id("sql-atomic-or-fail"); + let service = SqlService::new( + cache.path().to_string_lossy().into_owned(), + memory_threshold, + Arc::new(CallCountingRepository { + inner: repo, + calls: calls.clone(), + }), + ); + let run = |sql: &str, ability: &str| { + service.execute( + &space, + "main", + SqlRequest::Execute { + schema: None, + sql: sql.into(), + params: Vec::new(), + }, + None, + ability.into(), + ) + }; + run( + "CREATE TABLE u (id INTEGER PRIMARY KEY, v TEXT UNIQUE)", + "tinycloud.sql/admin", + ) + .await + .unwrap(); + run( + "INSERT INTO u VALUES (10,'x'),(11,'y'),(12,'z')", + "tinycloud.sql/write", + ) + .await + .unwrap(); + let handle = service.handle(&space, "main").await.unwrap(); + let before = handle.checkpoint().await.unwrap(); + + for sql in ["INSERT OR FAIL INTO u VALUES (1,'a'),(2,'b'),(1,'dup')"] { + calls.lock().unwrap().clear(); + let failed = run(sql, "tinycloud.sql/write").await; + assert!( + matches!(failed, Err(SqlError::Sqlite(_))), + "{sql}: {failed:?}" + ); + assert!( + calls.lock().unwrap().is_empty(), + "{sql}: a rejected write costs no repository call" + ); + assert_eq!( + rows_of(&service, &space, "SELECT count(*) FROM u").await, + vec![vec![SqlValue::Integer(3)]], + "{sql}" + ); + assert_eq!( + rows_of(&service, &space, "SELECT group_concat(v) FROM u").await, + vec![vec![SqlValue::Text("x,y,z".into())]], + "{sql}" + ); + assert!( + handle.checkpoint().await.unwrap() == before, + "{sql}: the checkpoint must not change" + ); + } + + // The same write as a `query` request, which the classifier + // treats as a write too. + let failed = service + .execute( + &space, + "main", + SqlRequest::Query { + sql: "INSERT OR FAIL INTO u VALUES (1,'a'),(2,'b'),(1,'dup')".into(), + params: Vec::new(), + max_rows: None, + max_bytes: None, + }, + None, + "tinycloud.sql/write".into(), + ) + .await; + assert!(failed.is_err(), "{failed:?}"); + assert!(handle.checkpoint().await.unwrap() == before); + } + } + + /// TC-1491: a deferred foreign key is checked when the request's + /// transaction commits. A violation there rolls the request back: nothing + /// is kept, the error is reported, and no repository call is made. + #[tokio::test] + async fn tc1491_a_deferred_constraint_failing_at_commit_leaves_nothing() { + for memory_threshold in [u64::MAX, 0] { + let repo = artifact_repository().await; + let calls = Arc::new(std::sync::Mutex::new(Vec::new())); + let cache = TempDir::new().unwrap(); + let space = test_space_id("sql-atomic-deferred-fk"); + let service = SqlService::new( + cache.path().to_string_lossy().into_owned(), + memory_threshold, + Arc::new(CallCountingRepository { + inner: repo, + calls: calls.clone(), + }), + ); + let run = |request: SqlRequest| { + service.execute(&space, "main", request, None, "tinycloud.sql/admin".into()) + }; + let execute = |sql: &str| SqlRequest::Execute { + schema: None, + sql: sql.into(), + params: Vec::new(), + }; + for ddl in [ + "CREATE TABLE parents (id INTEGER PRIMARY KEY)", + "CREATE TABLE children (parent_id INTEGER REFERENCES parents(id) DEFERRABLE INITIALLY DEFERRED)", + "INSERT INTO parents VALUES (1)", + "INSERT INTO children VALUES (1)", + ] { + run(execute(ddl)).await.unwrap(); + } + let handle = service.handle(&space, "main").await.unwrap(); + let before = handle.checkpoint().await.unwrap(); + + for failing in [ + execute("DELETE FROM parents"), + atomic_batch(&[ + "INSERT INTO parents VALUES (2)", + "DELETE FROM parents WHERE id = 1", + ]), + ] { + calls.lock().unwrap().clear(); + let failed = run(failing).await; + assert!(matches!(failed, Err(SqlError::Sqlite(_))), "{failed:?}"); + assert!( + calls.lock().unwrap().is_empty(), + "a request rejected at commit costs no repository call: {:?}", + calls.lock().unwrap() + ); + assert_eq!( + rows_of(&service, &space, "SELECT id FROM parents ORDER BY id").await, + vec![vec![SqlValue::Integer(1)]] + ); + assert!(handle.checkpoint().await.unwrap() == before); + } + + // The connection is not left inside a transaction: the next + // request commits and is persisted. + calls.lock().unwrap().clear(); + run(execute("DELETE FROM children")).await.unwrap(); + assert_eq!(calls.lock().unwrap().len(), 1); + assert!(rows_of(&service, &space, "SELECT * FROM children") + .await + .is_empty()); + } + } } diff --git a/tinycloud-core/tests/embedded_request_telemetry.rs b/tinycloud-core/tests/embedded_request_telemetry.rs index 244404eb..082a76fe 100644 --- a/tinycloud-core/tests/embedded_request_telemetry.rs +++ b/tinycloud-core/tests/embedded_request_telemetry.rs @@ -181,7 +181,9 @@ async fn sqlite_counts_schema_execution_errors_and_guard_controls() { .unwrap(); }) .await; - assert_eq!(context.snapshot().query_count, 4); + // Schema statement, statement, and the request transaction's SAVEPOINT and + // RELEASE (TC-1491), plus the two size PRAGMAs. + assert_eq!(context.snapshot().query_count, 6); let failed = request("aaaaaaaa-aaaa-4aaa-8aaa-aaaaaaaaaaaa"); scope(Arc::clone(&failed), async { actor @@ -199,7 +201,8 @@ async fn sqlite_counts_schema_execution_errors_and_guard_controls() { .expect_err("a constraint failure is an execution attempt"); }) .await; - assert_eq!(failed.snapshot().query_count, 1); + // The statement, and the request transaction's SAVEPOINT and ROLLBACK. + assert_eq!(failed.snapshot().query_count, 3); assert!(failed.snapshot().query_time_ns > 0); let guarded = request("bbbbbbbb-bbbb-4bbb-8bbb-bbbbbbbbbbbb"); scope(