From 66527f2942e1d81f22456e14cb204d3f2cfda05a Mon Sep 17 00:00:00 2001 From: Anderson Leal Date: Tue, 28 Jul 2026 15:53:07 -0300 Subject: [PATCH] (MOT-4235) feat(database): add row-change events and error hints --- .github/workflows/database-e2e.yml | 5 +- database/Cargo.lock | 2 - database/Cargo.toml | 2 - database/README.md | 46 +- database/skills/SKILL.md | 57 +- database/src/config.rs | 4 +- database/src/driver/mod.rs | 1 + database/src/driver/mysql.rs | 10 + database/src/driver/postgres.rs | 60 ++ database/src/driver/sqlite.rs | 472 ++++++++++++++- database/src/error.rs | 8 - database/src/handlers/begin_transaction.rs | 1 + database/src/handlers/commit_transaction.rs | 6 + database/src/handlers/execute.rs | 5 + database/src/handlers/list_databases.rs | 1 + database/src/handlers/mod.rs | 46 ++ database/src/handlers/prepare.rs | 1 + database/src/handlers/query.rs | 1 + database/src/handlers/rollback_transaction.rs | 101 ++++ database/src/handlers/run_statement.rs | 2 + database/src/handlers/transaction.rs | 52 +- database/src/handlers/transaction_execute.rs | 19 +- database/src/main.rs | 33 +- database/src/transaction.rs | 70 ++- database/src/triggers/bus.rs | 568 ++++++++++++++++++ database/src/triggers/handler.rs | 115 +++- database/src/triggers/mod.rs | 22 +- database/src/triggers/row_change.rs | 411 ------------- database/src/triggers/sql.rs | 425 +++++++++++++ database/tests/e2e/README.md | 14 +- database/tests/e2e/docker-compose.yml | 8 - database/tests/e2e/run-tests.sh | 9 +- .../harness/src/cases-interactive-tx.ts | 50 ++ .../workers/harness/src/cases-row-change.ts | 173 ------ .../workers/harness/src/cases-row-changed.ts | 211 +++++++ .../workers/harness/src/database-config.ts | 4 +- .../tests/e2e/workers/harness/src/runner.ts | 10 +- database/tests/integration.rs | 2 + 38 files changed, 2275 insertions(+), 752 deletions(-) create mode 100644 database/src/triggers/bus.rs delete mode 100644 database/src/triggers/row_change.rs create mode 100644 database/src/triggers/sql.rs delete mode 100644 database/tests/e2e/workers/harness/src/cases-row-change.ts create mode 100644 database/tests/e2e/workers/harness/src/cases-row-changed.ts diff --git a/.github/workflows/database-e2e.yml b/.github/workflows/database-e2e.yml index d3e427045..68a5d6bff 100644 --- a/.github/workflows/database-e2e.yml +++ b/.github/workflows/database-e2e.yml @@ -49,9 +49,8 @@ jobs: cache: 'npm' cache-dependency-path: database/tests/e2e/workers/harness/package-lock.json - # GHA `services:` blocks can't pass `-c wal_level=logical` to postgres, - # which the row-change tests require. Reuse the same docker-compose - # stack the harness uses locally for dev/CI parity. + # The harness brings up its own postgres + mysql via docker-compose + # (not GHA `services:`) so local and CI runs share one stack definition. - name: Install iii engine (next) run: | curl -fsSL --retry 3 --retry-connrefused --retry-delay 5 \ diff --git a/database/Cargo.lock b/database/Cargo.lock index 8b426486e..508d36d31 100644 --- a/database/Cargo.lock +++ b/database/Cargo.lock @@ -593,12 +593,10 @@ dependencies = [ "chrono", "clap", "deadpool-postgres", - "futures-util", "iii-console-ui", "iii-helpers", "iii-sdk", "mysql_async", - "postgres-protocol", "postgres-types", "r2d2", "r2d2_sqlite", diff --git a/database/Cargo.toml b/database/Cargo.toml index 3e9d1bb01..48bdeb1df 100644 --- a/database/Cargo.toml +++ b/database/Cargo.toml @@ -33,13 +33,11 @@ url = "2" chrono = { version = "0.4", features = ["serde"] } base64 = "0.22" bytes = "1" -futures-util = "0.3" # Postgres tokio-postgres = { version = "0.7", features = ["with-chrono-0_4", "with-serde_json-1", "with-uuid-1"] } deadpool-postgres = "0.14" postgres-types = { version = "0.2", features = ["with-chrono-0_4", "with-serde_json-1"] } -postgres-protocol = "0.6" tokio-postgres-rustls = "0.13" # `db-tokio-postgres` registers FromSql/ToSql for NUMERIC. postgres-types # itself ships no NUMERIC FromSql impl — String accepts only TEXT-family OIDs, diff --git a/database/README.md b/database/README.md index 531c69de5..71f64628f 100644 --- a/database/README.md +++ b/database/README.md @@ -195,21 +195,47 @@ const { rows } = await iii.trigger({ ## Triggers -### `database::row-change` -Postgres only. Streams row-level changes via logical replication (`pgoutput`). +### `database::row-changed` -> **NOTE (v1.0.0):** Event dispatch is not yet functional. The publication and replication slot are created at startup, but the streaming decode loop is stubbed pending an upstream `tokio-postgres` replication API release. Operators can pre-provision slots and publications now; events will start flowing in a later release. +Fires after this worker commits a row change. Driver-agnostic — no logical +replication, no per-database setup, identical on SQLite, Postgres and MySQL. ```yaml triggers: - - type: database::row-change + - type: database::row-changed config: - db: primary - schema: public - tables: [orders, payments] + db: primary # required + table: orders # optional; case- and schema-insensitive + ops: [insert] # optional; insert / update / delete / other ``` -The worker derives slot/publication names from `trigger_id`: `iii_slot__<8hex>` and `iii_pub__<8hex>`, where the 8-hex-char suffix is an FNV-1a-32 hash of the original `trigger_id`. The hash guarantees that two distinct trigger_ids (e.g. `orders-v1` vs `orders.v1`) produce distinct names even though both sanitize to `orders_v1`. The sanitized prefix is truncated at 40 chars so the final name fits in Postgres' 63-byte slot-name limit. Operators can override slot/publication names explicitly with `slot_name`/`publication_name`. Drop them with `pg_drop_replication_slot('')` and `DROP PUBLICATION ` if the worker is decommissioned without graceful shutdown. +Event: `{ db, table, op, affected_rows, returning?, at }`, where `op` is +`insert` / `update` / `delete` / `other`. + +**This is not change data capture.** It reports mutations made *through this +worker* — `execute`, `executeBatch`, `transaction`, and the interactive +transaction surface. A write applied by psql, another worker, or a +database-side trigger is invisible to it. That covers the case it exists for +(the worker is the only writer, and something needs to know when rows land) +and nothing more. + +Four things worth knowing: + +- **Announced on commit, never before.** Statements inside an interactive + transaction are buffered until `commitTransaction`; a rollback — including + the timeout watcher's — drops the buffer. Atomic batches announce their + statements in order only after the whole batch commits. +- **Delivery is best-effort.** Dispatch happens after commit and is not durable + or atomic with the database write. There is no replay, retry, or exactly-once + guarantee; a crash between commit and dispatch can lose an event. Subscriber + failures are logged and never fail the write. +- **`table` can be null.** The table is read off the SQL. A CTE-wrapped write + (`WITH … INSERT`) still fires, with `table: null`, rather than being dropped; + a binding that named a table simply does not match it. Omit `table` to match + every write, including these. +- **`runStatement` does not fire.** The prepared-run path returns rows, not an + affected-row count, and an event that invented one would be lying. Use + `execute` when you need the change announced. ## Errors @@ -224,8 +250,6 @@ Returned `IIIError::Handler` bodies carry a stable `code` field: | `UNKNOWN_DB` | `db` parameter doesn't match any configured database. | | `INVALID_PARAM` | JSON value couldn't be coerced for the target driver, transaction-control SQL was sent to `transactionExecute` (use `commitTransaction` / `rollbackTransaction`), or a `transaction`/`executeBatch` batch contained transaction-control SQL or an empty statement. | | `DRIVER_ERROR` | Wraps underlying driver error with `driver` and `inner_code` (nullable). `inner_code` format is per-driver: Postgres = SQLSTATE 5-char string (e.g. `42P01`), MySQL = server error number as string, SQLite = `rusqlite::ErrorCode` debug name. Pool-acquire failures use the message form `pool connection failed ()` where `` is one of `tls`, `auth`, `network`, `server-policy`, or `unknown` — a redacted hint so untrusted callers can self-triage without seeing host/userinfo/db fragments. The full driver error is in the worker's stderr via `tracing::warn!`. | -| `REPLICATION_SLOT_EXISTS` | Startup-only: another instance owns the slot. | -| `UNSUPPORTED` | Operation not supported on the chosen driver. | | `CONFIG_ERROR` | Config parse or pool init failure. | ## Driver compatibility @@ -237,7 +261,6 @@ A few operations are no-ops on certain drivers. They emit a `tracing::warn!` rat | `execute` with `returning: [...]` | ✓ | ✓ | warn-once + ignore | | `transaction` `isolation: read_committed` / `repeatable_read` | warn + use serializable | ✓ | ✓ | | `transaction` `isolation: serializable` | ✓ (`BEGIN IMMEDIATE`) | ✓ | ✓ | -| `database::row-change` trigger | — | setup-only in v1.0.0 (see above) | — | ## Troubleshooting @@ -249,7 +272,6 @@ A few operations are no-ops on certain drivers. They emit a `tracing::warn!` rat - `(auth)` — credential or pg_hba/SCRAM rejection. Includes Neon's `?channel_binding=require` failing through the pooler endpoint (drop the URL param, use `tls.mode` in YAML). - `(network)` — TCP refuse, DNS, route, or peer reset. Check host/port reachability and any firewalls. - `(server-policy)` — server reachable and TLS+auth OK, but the server actively refused (e.g. `max_connections` exceeded, admin shutdown). Look at the worker stderr for the underlying driver message. -- **Replication slot already exists**: another instance is consuming the slot. Either reuse the slot name or run `SELECT pg_drop_replication_slot('')`. ## License diff --git a/database/skills/SKILL.md b/database/skills/SKILL.md index f5ce66ea7..65dce3ce9 100644 --- a/database/skills/SKILL.md +++ b/database/skills/SKILL.md @@ -29,14 +29,13 @@ point. Placeholder syntax: `?` for SQLite and MySQL, `$1`/`$2`/… for Postgres. `database::runStatement`). - You need read-your-writes across round-trips with logic between steps (`database::beginTransaction` … `commitTransaction` / `rollbackTransaction`). -- You want to react to Postgres row-level changes once logical replication - streaming ships (`database::row-change` trigger — see below). ## Boundaries - Not a migration tool, ORM, or schema designer — pass raw SQL only. -- Not a general pub/sub bus — use `database::row-change` only for Postgres - table change feeds, not for application events. +- Not a general pub/sub bus. `database::row-changed` reports only what THIS + worker wrote, on commit — not change data capture; a write from psql or + another worker is invisible to it. - `database::query` is read-oriented; use `database::execute` for writes. Running a SELECT through `execute` discards rows. - Prepared handles pin a pool connection until TTL expiry — not transactions. @@ -82,45 +81,19 @@ Interactive transactions auto-roll back when `timeout_ms` elapses (default 30 s, max 5 min). Prepared handles default to a 1 h TTL (max 24 h) with no explicit release call — let them expire or stop using them when done. -## Reactive triggers +## Reacting to writes -Register a `database::row-change` trigger when a function should run -automatically on Postgres INSERT/UPDATE/DELETE for specific tables — without -polling with `database::query`. +Register a `database::row-changed` trigger to be told when this worker commits +a change, instead of polling: -Reach for it when: - -- A downstream worker or workflow must react to row mutations in near real - time on Postgres. -- You need decoded row payloads (old/new values) from logical replication - rather than polling an outbox table. - -Do not bind when: - -- The writer already has the new row in its `execute` or `transactionExecute` - return payload. -- You are on SQLite or MySQL — this trigger type is Postgres-only. -- You need events today — v1.0.0 returns `UNSUPPORTED` on `registerTrigger` - pending an upstream `tokio-postgres` replication API release. - -### How to bind - -1. Register a handler: `registerFunction('stream::on-row-change', handler)`. -2. Register the trigger: - -```typescript -iii.registerTrigger({ - type: 'database::row-change', - function_id: 'stream::on-row-change', - config: { - db: 'primary', - schema: 'public', - tables: ['orders', 'payments'], - // optional: slot_name, publication_name — see get function info - }, -}) +```json +{ "trigger_type": "database::row-changed", "config": { "db": "primary", "table": "orders", "ops": ["insert"] } } ``` -Config: `db`, `schema` (default `public`), `tables`. Slot/publication names -derive from `trigger_id` unless overridden. For event payload shape, call -`get function info` on the trigger type or handler function id. +The event is `{ db, table, op, affected_rows, returning?, at }`. It fires on +commit — an interactive transaction's writes are announced by +`commitTransaction`, and a rollback announces nothing. `table` is null when the +statement's table cannot be read off the SQL (a CTE-wrapped write), and +`runStatement` does not fire because it has no affected-row count to report. +Delivery is best-effort: it is not durable with the commit and has no replay or +exactly-once guarantee. diff --git a/database/src/config.rs b/database/src/config.rs index b0b969a41..951f98383 100644 --- a/database/src/config.rs +++ b/database/src/config.rs @@ -298,8 +298,8 @@ pub fn redact_url(input: &str) -> String { /// Max 63 chars (Postgres NAMEDATALEN - 1). /// /// This is the chokepoint for any operator-supplied identifier that gets -/// interpolated into a SQL string via `format!()` (replication slots, -/// publication names, schema/table names, cursor table). Validation is +/// interpolated into a SQL string via `format!()` (schema/table names, +/// cursor table). Validation is /// strict ASCII because the alternative — quoting and escaping per-driver — /// is fragile and the v1.0 surface does not need unicode identifiers. pub fn validate_sql_identifier(s: &str) -> Result<(), String> { diff --git a/database/src/driver/mod.rs b/database/src/driver/mod.rs index a40b903e3..4ce326cd4 100644 --- a/database/src/driver/mod.rs +++ b/database/src/driver/mod.rs @@ -51,4 +51,5 @@ pub struct TxStatement { pub struct TxStepResult { pub affected_rows: u64, pub rows: Vec, + pub columns: Vec, } diff --git a/database/src/driver/mysql.rs b/database/src/driver/mysql.rs index bae7a202e..0d3c581ce 100644 --- a/database/src/driver/mysql.rs +++ b/database/src/driver/mysql.rs @@ -205,6 +205,14 @@ pub async fn transaction( let step_result: Result = if returns_rows { match conn.exec_iter(stmt.sql.as_str(), bound).await { Ok(mut iter) => { + let columns = iter + .columns_ref() + .iter() + .map(|col| ColumnMeta { + name: col.name_str().to_string(), + ty: format!("{:?}", col.column_type()), + }) + .collect(); let raw: Result, _> = iter.collect().await; match raw { Ok(raw_rows) => { @@ -213,6 +221,7 @@ pub async fn transaction( Ok(TxStepResult { affected_rows: cells_rows.len() as u64, rows: cells_rows, + columns, }) } Err(e) => Err(step_err(idx, e)), @@ -225,6 +234,7 @@ pub async fn transaction( Ok(_) => Ok(TxStepResult { affected_rows: conn.affected_rows(), rows: vec![], + columns: vec![], }), Err(e) => Err(step_err(idx, e)), } diff --git a/database/src/driver/postgres.rs b/database/src/driver/postgres.rs index 66e9ab18b..ac0dd6e8c 100644 --- a/database/src/driver/postgres.rs +++ b/database/src/driver/postgres.rs @@ -384,6 +384,18 @@ pub async fn transaction( let step = if returns_rows { match tx_client.query(&stmt.sql, bound_refs.as_slice()).await { Ok(rows) => { + let columns = rows + .first() + .map(|row| { + row.columns() + .iter() + .map(|col| ColumnMeta { + name: col.name().to_string(), + ty: col.type_().name().to_string(), + }) + .collect() + }) + .unwrap_or_default(); let mut cells_rows: Vec = Vec::with_capacity(rows.len()); for row in &rows { let mut cells = Vec::with_capacity(row.columns().len()); @@ -395,6 +407,7 @@ pub async fn transaction( TxStepResult { affected_rows: cells_rows.len() as u64, rows: cells_rows, + columns, } } Err(e) => { @@ -407,6 +420,7 @@ pub async fn transaction( Ok(n) => TxStepResult { affected_rows: n, rows: vec![], + columns: vec![], }, Err(e) => { let _ = tx_client.batch_execute("ROLLBACK").await; @@ -456,6 +470,10 @@ pub async fn tx_begin( /// `COMMIT` the in-progress transaction on a pinned client. pub async fn tx_commit(client: &mut crate::pool::postgres::PgClient) -> Result<(), DbError> { + // PostgreSQL accepts COMMIT in an aborted transaction but reports the + // command tag ROLLBACK; batch_execute discards that tag. Probe while the + // transaction is still open so 25P02 takes the ordinary commit-error path. + client.simple_query("SELECT 1").await.map_err(map_err)?; client.batch_execute("COMMIT").await.map_err(map_err) } @@ -1093,6 +1111,48 @@ mod tests { assert!(matches!(&r.rows[0].0[0], RowValue::BigInt(0))); } + #[tokio::test(flavor = "multi_thread")] + async fn pg_interactive_commit_rejects_an_aborted_transaction() { + let Some(p) = fresh_pool().await else { return }; + let _ = execute(&p, "DROP TABLE IF EXISTS db_w_aborted_tx", &[], &[]).await; + execute( + &p, + "CREATE TABLE db_w_aborted_tx (n INT NOT NULL)", + &[], + &[], + ) + .await + .unwrap(); + + let mut client = p.acquire().await.unwrap(); + tx_begin(&mut client, None).await.unwrap(); + tx_execute( + &mut client, + "INSERT INTO db_w_aborted_tx VALUES (1)", + &[], + &[], + ) + .await + .unwrap(); + tx_execute( + &mut client, + "INSERT INTO db_w_aborted_tx VALUES (NULL)", + &[], + &[], + ) + .await + .unwrap_err(); + + assert!(tx_commit(&mut client).await.is_err()); + tx_rollback(&mut client).await.unwrap(); + drop(client); + + let rows = query(&p, "SELECT COUNT(*) FROM db_w_aborted_tx", &[], 30_000) + .await + .unwrap(); + assert!(matches!(&rows.rows[0].0[0], RowValue::BigInt(0))); + } + #[tokio::test(flavor = "multi_thread")] async fn pg_run_prepared_executes_with_params() { let Some(p) = fresh_pool().await else { return }; diff --git a/database/src/driver/sqlite.rs b/database/src/driver/sqlite.rs index f45f9085c..669368887 100644 --- a/database/src/driver/sqlite.rs +++ b/database/src/driver/sqlite.rs @@ -21,7 +21,9 @@ pub async fn query( tokio::task::spawn_blocking(move || -> Result { conn.with(|c| { - let mut stmt = c.prepare(&sql).map_err(map_err)?; + let mut stmt = c + .prepare(&sql) + .map_err(|e| enrich_schema_err(c, &sql, map_err(e)))?; // `database::query` is the READ surface a narrowed agent policy // grants; enforcement must live here, not in the docs — live // testing showed agents running raw INSERTs through it. @@ -111,6 +113,149 @@ pub(crate) fn map_err(e: rusqlite::Error) -> DbError { } } +/// A schema-mismatch error names what is MISSING but never what EXISTS — and +/// an agent that guessed a column name once will guess it again. Append the +/// real schema so the first failure carries its own correction. Discovery run +/// 2: fifteen delivered events, ONE surviving ledger row — the inspector's +/// INSERTs disagreed with the coordinator's CREATE TABLE on a column name, +/// and the bare "has no column named value" left it guessing for 13 turns. +fn enrich_schema_err(c: &rusqlite::Connection, sql: &str, e: DbError) -> DbError { + let DbError::DriverError { + driver, + code, + message, + failed_index, + } = e + else { + return e; + }; + let message = match schema_hint(c, sql, &message) { + Some(hint) => format!("{message}; {hint}"), + None => message, + }; + DbError::DriverError { + driver, + code, + message, + failed_index, + } +} + +fn schema_hint(c: &rusqlite::Connection, sql: &str, message: &str) -> Option { + // `INSERT INTO t (bad) …` → "table t has no column named bad": the + // message itself names the table. + if let Some(rest) = message.strip_prefix("table ") { + if let Some(table) = rest.split(" has no column named ").next() { + if rest.contains(" has no column named ") { + return columns_hint(c, table); + } + } + } + // `UPDATE t SET bad = …` / `SELECT bad FROM t` → "no such column: bad": + // use SQLite's qualifier when present; otherwise hint only when the + // statement has one unambiguous source. + if message.contains("no such column") { + let qualified = message + .split_once("no such column: ") + .and_then(|(_, missing)| missing.split_whitespace().next()) + .and_then(|missing| missing.rsplit('.').nth(1)); + let table = qualified + .map(str::to_string) + .or_else(|| crate::triggers::sql::classify(sql).and_then(|m| m.table)) + .or_else(|| crate::triggers::sql::table_after_from(sql))?; + return columns_hint(c, &table); + } + // "no such table: t" → say what tables DO exist. + if let Some(rest) = message.strip_prefix("no such table: ") { + let missing = rest.split(" in ").next().unwrap_or(rest).trim(); + let names = existing_tables(c)?; + if names.is_empty() { + return Some(format!("`{missing}` not found and no tables exist yet")); + } + return Some(format!("existing tables: ({})", names.join(", "))); + } + // A syntax error on SQL that is valid PostgreSQL: name the dialect gap. + // The bare `near "INSERT": syntax error` reads as a typo and gets retried + // verbatim — discovery run 6 lost its whole ledger that way. + if message.contains("syntax error") { + return dialect_hint(sql); + } + None +} + +/// PostgreSQL constructs SQLite rejects, answered with the SQLite way. +fn dialect_hint(sql: &str) -> Option { + is_data_modifying_cte(sql).then(|| { + "SQLite does not support data-modifying CTEs (INSERT/UPDATE/DELETE inside `WITH`) — that \ + is PostgreSQL syntax. Use one statement per call (a plain `INSERT … ON CONFLICT … \ + RETURNING` reports what it wrote), or database::transaction for a multi-step atomic \ + sequence" + .to_string() + }) +} + +/// Whether the statement opens a `WITH` whose first parenthesised body is a +/// write — `WITH x AS (INSERT …)`. Keyword-level, like the mutation +/// classifier: a false negative just leaves the bare error in place. +fn is_data_modifying_cte(sql: &str) -> bool { + let upper = sql.trim_start().to_ascii_uppercase(); + if !upper.starts_with("WITH") { + return false; + } + let Some(open) = upper.find('(') else { + return false; + }; + let body = upper[open + 1..].trim_start(); + ["INSERT", "UPDATE", "DELETE", "REPLACE", "MERGE"] + .iter() + .any(|verb| body.starts_with(verb)) +} + +/// `name declared-type` for every column of `table`, via the parameterized +/// pragma function — no identifier interpolation. +fn columns_hint(c: &rusqlite::Connection, table: &str) -> Option { + let mut stmt = c + .prepare("SELECT name, type FROM pragma_table_info(?1)") + .ok()?; + let cols: Vec = stmt + .query_map([table], |row| { + let name: String = row.get(0)?; + let ty: String = row.get(1)?; + Ok(if ty.is_empty() { + name + } else { + format!("{name} {ty}") + }) + }) + .ok()? + .filter_map(Result::ok) + .collect(); + if cols.is_empty() { + return None; + } + Some(format!("table {table} columns: ({})", cols.join(", "))) +} + +fn existing_tables(c: &rusqlite::Connection) -> Option> { + const MAX: usize = 20; + let mut stmt = c + .prepare( + "SELECT name FROM sqlite_master WHERE type = 'table' \ + AND name NOT LIKE 'sqlite_%' ORDER BY name", + ) + .ok()?; + let mut names: Vec = stmt + .query_map([], |row| row.get::<_, String>(0)) + .ok()? + .filter_map(Result::ok) + .collect(); + if names.len() > MAX { + names.truncate(MAX); + names.push("…".into()); + } + Some(names) +} + /// Stamp a transaction-step index onto an existing `DbError`. Used inside /// `run_tx_steps` to preserve the failed-step index when an error bubbles up /// from a helper (e.g. `row_value_at`) that has no notion of "which step is @@ -156,16 +301,26 @@ fn is_insert(sql: &str) -> bool { sql.trim_start().to_ascii_uppercase().starts_with("INSERT") } -/// Returns true if the prepared statement will surface result rows, or the -/// caller explicitly requested row capture via `returning`. SQLite's -/// `Statement::column_count` is the planner's source-of-truth: it returns -/// `> 0` for any statement shape that produces rows — `SELECT`, -/// `WITH cte AS (...) SELECT ...`, `VALUES (...)`, `PRAGMA foreign_keys`, -/// `EXPLAIN QUERY PLAN`, and any DML with a `RETURNING` clause regardless -/// of casing or whitespace. Replaces brittle text-prefix matches that -/// false-negatived CTE-prefixed SELECTs and aborted transactions for them. -fn statement_returns_rows(stmt: &rusqlite::Statement<'_>, returning: &[String]) -> bool { - !returning.is_empty() || stmt.column_count() > 0 +/// SQLite's planner is the source of truth for whether a prepared statement +/// produces rows, including CTEs, PRAGMAs, and DML with `RETURNING`. +fn statement_returns_rows(stmt: &rusqlite::Statement<'_>) -> bool { + stmt.column_count() > 0 +} + +fn validate_returning(stmt: &rusqlite::Statement<'_>, returning: &[String]) -> Result<(), DbError> { + if !returning.is_empty() && stmt.column_count() == 0 { + return Err(DbError::DriverError { + driver: "sqlite".into(), + code: Some("RETURNING_MISMATCH".into()), + message: format!( + "`returning` was requested but the statement returns no rows — \ + write the clause into the SQL itself: ... RETURNING {}", + returning.join(", ") + ), + failed_index: None, + }); + } + Ok(()) } pub async fn execute( @@ -203,8 +358,20 @@ pub async fn execute( // and DML-with-RETURNING split across lines, falling through to // `c.execute(...)` which errored with ExecuteReturnedResults. let (affected_rows, returned_rows, returned_columns) = { - let mut stmt = c.prepare(&sql).map_err(map_err)?; - if statement_returns_rows(&stmt, &returning) { + let mut stmt = c + .prepare(&sql) + .map_err(|e| enrich_schema_err(c, &sql, map_err(e)))?; + // A `returning` OPTION against a statement that produces no + // rows is a contradiction the caller needs to hear about: the + // option does not inject a RETURNING clause, so running the + // statement query-style would insert the row and then report + // affected_rows: 0 with no rows — silent garbage that + // downstream consumers (the row-changed event's identity, a + // caller reading its ids back) build on. Live run rctest9: + // fifteen identity-less events, an aggregator that rightly + // refused them, and a barrier that starved. + validate_returning(&stmt, &returning)?; + if statement_returns_rows(&stmt) { let columns: Vec = stmt .columns() .into_iter() @@ -367,9 +534,19 @@ fn run_tx_steps( // `is_select || is_returning` heuristic and fell through to // `c.execute(...)`, which errors with ExecuteReturnedResults and // aborts the entire transaction. - let mut prepared = c.prepare(&stmt.sql).map_err(|e| step_err(idx, e))?; - if statement_returns_rows(&prepared, &[]) { - let n = prepared.columns().len(); + let mut prepared = c + .prepare(&stmt.sql) + .map_err(|e| enrich_schema_err(c, &stmt.sql, step_err(idx, e)))?; + if statement_returns_rows(&prepared) { + let columns: Vec = prepared + .columns() + .into_iter() + .map(|col| ColumnMeta { + name: col.name().to_string(), + ty: col.decl_type().unwrap_or("").to_string(), + }) + .collect(); + let n = columns.len(); let mut rows_out: Vec = Vec::new(); let mut rows = prepared .query(bound_refs.as_slice()) @@ -388,6 +565,7 @@ fn run_tx_steps( results.push(TxStepResult { affected_rows: rows_out.len() as u64, rows: rows_out, + columns, }); } else { let affected = prepared @@ -396,6 +574,7 @@ fn run_tx_steps( results.push(TxStepResult { affected_rows: affected as u64, rows: vec![], + columns: vec![], }); } } @@ -512,8 +691,11 @@ pub async fn tx_execute( bound.iter().map(|v| v as &dyn rusqlite::ToSql).collect(); let (affected_rows, returned_rows, returned_columns) = { - let mut stmt = c.prepare(&sql).map_err(map_err)?; - if statement_returns_rows(&stmt, &returning) { + let mut stmt = c + .prepare(&sql) + .map_err(|e| enrich_schema_err(c, &sql, map_err(e)))?; + validate_returning(&stmt, &returning)?; + if statement_returns_rows(&stmt) { let columns: Vec = stmt .columns() .into_iter() @@ -604,7 +786,9 @@ pub async fn run_prepared( let bound: Vec = params.iter().map(json_param_to_sql).collect(); let bound_refs: Vec<&dyn rusqlite::ToSql> = bound.iter().map(|v| v as &dyn rusqlite::ToSql).collect(); - let mut stmt = c.prepare(&sql).map_err(map_err)?; + let mut stmt = c + .prepare(&sql) + .map_err(|e| enrich_schema_err(c, &sql, map_err(e)))?; let columns: Vec = stmt .columns() .into_iter() @@ -1194,4 +1378,254 @@ mod tests { other => panic!("expected DriverError, got {other:?}"), } } + + #[tokio::test(flavor = "multi_thread")] + async fn returning_option_without_a_returning_clause_is_refused() { + // The rctest9 failure shape: the option forces the query path, a plain + // INSERT yields no rows, and the caller got affected_rows: 0 with no + // rows while the insert silently succeeded — identity-less events all + // the way down. Refusing loudly turns a starved run into a one-call fix. + let pool = + SqlitePool::new("sqlite::memory:", &crate::config::PoolConfig::default()).unwrap(); + execute( + &pool, + "CREATE TABLE t (id INTEGER PRIMARY KEY, n INT)", + &[], + &[], + ) + .await + .unwrap(); + + let err = execute( + &pool, + "INSERT INTO t (n) VALUES (1)", + &[], + &["id".into(), "n".into()], + ) + .await + .unwrap_err(); + let msg = err.to_string(); + assert!(msg.contains("RETURNING id, n"), "must name the fix: {msg}"); + // Nothing was inserted by the refused call. + let q = query(&pool, "SELECT COUNT(*) AS c FROM t", &[], 5_000) + .await + .unwrap(); + assert_eq!(q.rows[0].0[0].clone().into_json(), serde_json::json!(0)); + + // The same statement WITH the clause works and reports real rows. + let ok = execute( + &pool, + "INSERT INTO t (n) VALUES (1) RETURNING id, n", + &[], + &["id".into(), "n".into()], + ) + .await + .unwrap(); + assert_eq!(ok.affected_rows, 1); + assert_eq!(ok.returned_rows.len(), 1); + } + + #[tokio::test(flavor = "multi_thread")] + async fn interactive_returning_mismatch_is_refused_before_insert() { + let p = pool().await; + let mut slot = Some(p.acquire().await.unwrap()); + tx_begin(&mut slot, None).await.unwrap(); + tx_execute( + &mut slot, + "CREATE TABLE t (id INTEGER PRIMARY KEY, n INT)", + &[], + &[], + ) + .await + .unwrap(); + + let err = tx_execute( + &mut slot, + "INSERT INTO t (n) VALUES (1)", + &[], + &["id".into()], + ) + .await + .unwrap_err(); + assert!(err.to_string().contains("RETURNING id"), "{err}"); + + let count = run_prepared(&mut slot, "SELECT COUNT(*) FROM t", &[]) + .await + .unwrap(); + assert_eq!(count.rows[0].0[0].clone().into_json(), serde_json::json!(0)); + tx_rollback(&mut slot).await.unwrap(); + } + + /// The schema-drift fix: a mismatch error carries the table's REAL + /// columns (or the real table names), so the first failure is + /// self-correcting instead of the start of a guess loop. + #[tokio::test(flavor = "multi_thread")] + async fn schema_errors_carry_the_actual_schema() { + let p = pool().await; + execute( + &p, + "CREATE TABLE receiving (shipment_id TEXT PRIMARY KEY, shipment_value NUMERIC)", + &[], + &[], + ) + .await + .unwrap(); + + // INSERT against a wrong column: sqlite names the table itself. + let err = execute( + &p, + "INSERT INTO receiving (shipment_id, value) VALUES ('a', 1)", + &[], + &[], + ) + .await + .unwrap_err(); + let msg = format!("{err:?}"); + assert!(msg.contains("has no column named value"), "{msg}"); + assert!( + msg.contains("table receiving columns: (shipment_id TEXT, shipment_value NUMERIC)"), + "the fix is the columns list: {msg}" + ); + + // SELECT against a wrong column: the table comes off the FROM clause. + let err = query(&p, "SELECT value FROM receiving", &[], 1_000) + .await + .unwrap_err(); + let msg = format!("{err:?}"); + assert!(msg.contains("no such column"), "{msg}"); + assert!(msg.contains("shipment_value NUMERIC"), "{msg}"); + + // UPDATE against a wrong column: the table comes off the classifier. + let err = execute(&p, "UPDATE receiving SET value = 2", &[], &[]) + .await + .unwrap_err(); + let msg = format!("{err:?}"); + assert!(msg.contains("shipment_value NUMERIC"), "{msg}"); + + execute(&p, "CREATE TABLE a (id INT, a_value TEXT)", &[], &[]) + .await + .unwrap(); + execute(&p, "CREATE TABLE b (id INT, b_value TEXT)", &[], &[]) + .await + .unwrap(); + + // A qualified joined-column error names the qualified table, not the + // first FROM source. + let err = query( + &p, + "SELECT b.missing FROM a JOIN b ON a.id = b.id", + &[], + 1_000, + ) + .await + .unwrap_err(); + let msg = format!("{err:?}"); + assert!( + msg.contains("table b columns: (id INT, b_value TEXT)"), + "{msg}" + ); + assert!(!msg.contains("a_value"), "{msg}"); + + // An unqualified missing column in a multi-table query is ambiguous, + // so no table schema is safer than a wrong one. + let err = query( + &p, + "SELECT missing FROM a JOIN b ON a.id = b.id", + &[], + 1_000, + ) + .await + .unwrap_err(); + let msg = format!("{err:?}"); + assert!(!msg.contains("columns: ("), "{msg}"); + + // Missing table: the existing tables are named. + let err = query(&p, "SELECT * FROM receiving_shipments", &[], 1_000) + .await + .unwrap_err(); + let msg = format!("{err:?}"); + assert!(msg.contains("no such table"), "{msg}"); + assert!(msg.contains("existing tables: (a, b, receiving)"), "{msg}"); + + // A non-schema error stays untouched (no hint appended). + let err = execute( + &p, + "INSERT INTO receiving (shipment_id) VALUES ('a', 'b')", + &[], + &[], + ) + .await + .unwrap_err(); + let msg = format!("{err:?}"); + assert!(!msg.contains("columns: ("), "{msg}"); + } + + /// Discovery run 6: the inspector wrote a PostgreSQL data-modifying CTE, + /// got `near \"INSERT\": syntax error`, read it as a typo, and retried the + /// same dialect — 0 rows written. The error now names the dialect gap. + #[tokio::test(flavor = "multi_thread")] + async fn a_data_modifying_cte_is_named_as_postgres_syntax() { + let p = pool().await; + execute( + &p, + "CREATE TABLE t (id TEXT PRIMARY KEY, n INTEGER)", + &[], + &[], + ) + .await + .unwrap(); + + let err = execute( + &p, + "WITH inserted AS (INSERT INTO t (id, n) VALUES ('a', 1) RETURNING id) \ + UPDATE t SET n = 2 WHERE id IN (SELECT id FROM inserted)", + &[], + &[], + ) + .await + .unwrap_err(); + let msg = format!("{err:?}"); + assert!(msg.contains("syntax error"), "{msg}"); + assert!( + msg.contains("data-modifying CTEs") && msg.contains("PostgreSQL"), + "the dialect gap must be named: {msg}" + ); + assert!( + msg.contains("database::transaction"), + "the fix is named: {msg}" + ); + + // A read-only CTE is valid SQLite and must keep working. + query( + &p, + "WITH ids AS (SELECT id FROM t) SELECT COUNT(*) c FROM ids", + &[], + 1_000, + ) + .await + .expect("a read-only CTE is ordinary SQLite"); + + // An ordinary typo gets no dialect hint. + let err = execute(&p, "INSERTT INTO t (id) VALUES ('x')", &[], &[]) + .await + .unwrap_err(); + assert!(!format!("{err:?}").contains("PostgreSQL")); + } + + #[test] + fn data_modifying_cte_detection_is_keyword_level() { + assert!(is_data_modifying_cte( + "WITH x AS (INSERT INTO t VALUES (1)) SELECT 1" + )); + assert!(is_data_modifying_cte( + " with x as ( update t set a = 1 ) select 1" + )); + assert!(is_data_modifying_cte("WITH x AS (DELETE FROM t) SELECT 1")); + // Read-only CTEs and plain statements are not flagged. + assert!(!is_data_modifying_cte( + "WITH x AS (SELECT 1) SELECT * FROM x" + )); + assert!(!is_data_modifying_cte("INSERT INTO t VALUES (1)")); + assert!(!is_data_modifying_cte("SELECT 1")); + } } diff --git a/database/src/error.rs b/database/src/error.rs index 42aefd88a..0c6d2834f 100644 --- a/database/src/error.rs +++ b/database/src/error.rs @@ -50,14 +50,6 @@ pub enum DbError { failed_index: Option, }, - #[serde(rename = "REPLICATION_SLOT_EXISTS")] - #[error("replication slot {slot} already in use")] - ReplicationSlotExists { slot: String }, - - #[serde(rename = "UNSUPPORTED")] - #[error("operation {op} not supported on driver {driver}")] - Unsupported { op: String, driver: String }, - #[serde(rename = "CONFIG_ERROR")] #[error("config error: {message}")] ConfigError { message: String }, diff --git a/database/src/handlers/begin_transaction.rs b/database/src/handlers/begin_transaction.rs index 7dc414f2a..67b165425 100644 --- a/database/src/handlers/begin_transaction.rs +++ b/database/src/handlers/begin_transaction.rs @@ -142,6 +142,7 @@ pub(crate) mod tests { handles: Arc::new(HandleRegistry::new()), transactions: TxRegistry::new(), log: Logger::new(), + row_changes: None, } } diff --git a/database/src/handlers/commit_transaction.rs b/database/src/handlers/commit_transaction.rs index 7fdad3371..7c140d37d 100644 --- a/database/src/handlers/commit_transaction.rs +++ b/database/src/handlers/commit_transaction.rs @@ -67,9 +67,14 @@ pub async fn handle(state: &AppState, req: CommitTxReq) -> Result { + state.drop_row_changes(&req.transaction_id); // COMMIT failed — issue best-effort ROLLBACK to leave the // connection in a known state before the pool's recycler // (Postgres deadpool uses Fast recycle, which does NOT issue @@ -163,6 +168,7 @@ mod tests { handles: std::sync::Arc::new(crate::handle::HandleRegistry::new()), transactions: crate::transaction::TxRegistry::new(), log: iii_helpers::observability::Logger::new(), + row_changes: None, }; crate::handlers::execute::handle( diff --git a/database/src/handlers/execute.rs b/database/src/handlers/execute.rs index 5434c8eeb..a7c47a627 100644 --- a/database/src/handlers/execute.rs +++ b/database/src/handlers/execute.rs @@ -57,6 +57,10 @@ pub async fn handle(state: &AppState, req: ExecuteReq) -> Result, pub transactions: TxRegistry, pub log: Logger, + /// `database::row-changed` bindings and their fan-out. Absent when the + /// worker runs without an engine connection (tests). + pub row_changes: Option>, +} + +impl AppState { + /// Announce a committed change, if anything is listening. Every mutating + /// handler ends with this; the bus decides whether the statement changed + /// rows and who cares. + pub async fn emit_row_change( + &self, + db: &str, + sql: &str, + affected_rows: u64, + returning: Option<&[serde_json::Map]>, + ) { + if let Some(bus) = &self.row_changes { + bus.emit(db, sql, affected_rows, returning).await; + } + } + + /// Buffer a change made inside an interactive transaction until its commit. + pub fn stage_row_change( + &self, + transaction_id: &str, + db: &str, + sql: &str, + affected_rows: u64, + returning: Option<&[serde_json::Map]>, + ) { + if let Some(bus) = &self.row_changes { + bus.stage(transaction_id, db, sql, affected_rows, returning); + } + } + + pub async fn commit_row_changes(&self, transaction_id: &str) { + if let Some(bus) = &self.row_changes { + bus.commit(transaction_id).await; + } + } + + pub fn drop_row_changes(&self, transaction_id: &str) { + if let Some(bus) = &self.row_changes { + bus.rollback(transaction_id); + } + } } impl AppState { diff --git a/database/src/handlers/prepare.rs b/database/src/handlers/prepare.rs index 2498b58f0..6adae3a1a 100644 --- a/database/src/handlers/prepare.rs +++ b/database/src/handlers/prepare.rs @@ -97,6 +97,7 @@ mod tests { handles: Arc::new(HandleRegistry::new()), transactions: crate::transaction::TxRegistry::new(), log: iii_helpers::observability::Logger::new(), + row_changes: None, } } diff --git a/database/src/handlers/query.rs b/database/src/handlers/query.rs index 9158a1be9..3dde87fd9 100644 --- a/database/src/handlers/query.rs +++ b/database/src/handlers/query.rs @@ -124,6 +124,7 @@ mod tests { handles: Arc::new(HandleRegistry::new()), transactions: crate::transaction::TxRegistry::new(), log: iii_helpers::observability::Logger::new(), + row_changes: None, } } diff --git a/database/src/handlers/rollback_transaction.rs b/database/src/handlers/rollback_transaction.rs index 12aba9eed..7e0a230ea 100644 --- a/database/src/handlers/rollback_transaction.rs +++ b/database/src/handlers/rollback_transaction.rs @@ -41,6 +41,9 @@ pub async fn handle(state: &AppState, req: RollbackTxReq) -> Result driver::sqlite::tx_rollback(slot).await, @@ -87,11 +90,33 @@ mod tests { use super::*; use crate::handlers::begin_transaction::tests::state; use serde_json::{json, Value}; + use std::sync::Arc; + use std::time::Duration; fn req(v: Value) -> RollbackTxReq { serde_json::from_value(v).unwrap() } + fn state_with_bus() -> (AppState, Arc) { + let mut st = state(); + let bus = Arc::new(crate::triggers::RowChangeBus::new( + Arc::new(iii_sdk::IIIClient::new("ws://127.0.0.1:9")), + 100, + )); + st.row_changes = Some(bus.clone()); + (st, bus) + } + + async fn wait_until_taken(state: &AppState) { + tokio::time::timeout(Duration::from_secs(1), async { + while !state.transactions.is_empty().await { + tokio::task::yield_now().await; + } + }) + .await + .unwrap(); + } + #[tokio::test(flavor = "multi_thread")] async fn rollback_returns_rolled_back_true_and_removes_from_registry() { let st = state(); @@ -120,6 +145,81 @@ mod tests { assert!(err.contains("TRANSACTION_NOT_FOUND"), "got: {err}"); } + #[tokio::test(flavor = "multi_thread")] + async fn rollback_drops_changes_staged_by_an_in_flight_execute() { + let (st, bus) = state_with_bus(); + let begin = crate::handlers::begin_transaction::handle( + &st, + serde_json::from_value(json!({ "db": "primary" })).unwrap(), + ) + .await + .unwrap(); + let id = begin.transaction.id; + let lock = st.transactions.lock(&id).await.unwrap(); + + let task_state = st.clone(); + let task_id = id.clone(); + let rollback = tokio::spawn(async move { + handle( + &task_state, + RollbackTxReq { + transaction_id: task_id, + }, + ) + .await + }); + wait_until_taken(&st).await; + + st.stage_row_change(&id, "primary", "INSERT INTO t VALUES (1)", 1, None); + assert_eq!(bus.pending_count(&id), 1); + drop(lock); + + rollback.await.unwrap().unwrap(); + assert_eq!(bus.pending_count(&id), 0); + } + + #[tokio::test(flavor = "multi_thread")] + async fn rollback_losing_to_commit_does_not_clear_pending_changes() { + let (st, bus) = state_with_bus(); + let begin = crate::handlers::begin_transaction::handle( + &st, + serde_json::from_value(json!({ "db": "primary" })).unwrap(), + ) + .await + .unwrap(); + let id = begin.transaction.id; + st.stage_row_change(&id, "primary", "INSERT INTO t VALUES (1)", 1, None); + let lock = st.transactions.lock(&id).await.unwrap(); + + let task_state = st.clone(); + let task_id = id.clone(); + let commit = tokio::spawn(async move { + crate::handlers::commit_transaction::handle( + &task_state, + crate::handlers::commit_transaction::CommitTxReq { + transaction_id: task_id, + }, + ) + .await + }); + wait_until_taken(&st).await; + + let err = handle( + &st, + RollbackTxReq { + transaction_id: id.clone(), + }, + ) + .await + .unwrap_err(); + assert!(err.contains("TRANSACTION_NOT_FOUND"), "{err}"); + assert_eq!(bus.pending_count(&id), 1); + + drop(lock); + commit.await.unwrap().unwrap(); + assert_eq!(bus.pending_count(&id), 0); + } + #[tokio::test(flavor = "multi_thread")] async fn writes_inside_rolled_back_tx_are_not_visible() { let tmp = tempfile::NamedTempFile::new().unwrap(); @@ -136,6 +236,7 @@ mod tests { handles: std::sync::Arc::new(crate::handle::HandleRegistry::new()), transactions: crate::transaction::TxRegistry::new(), log: iii_helpers::observability::Logger::new(), + row_changes: None, }; crate::handlers::execute::handle( diff --git a/database/src/handlers/run_statement.rs b/database/src/handlers/run_statement.rs index e4dd3b8eb..c24264155 100644 --- a/database/src/handlers/run_statement.rs +++ b/database/src/handlers/run_statement.rs @@ -63,6 +63,7 @@ mod tests { handles: Arc::new(HandleRegistry::new()), transactions: crate::transaction::TxRegistry::new(), log: iii_helpers::observability::Logger::new(), + row_changes: None, } } @@ -80,6 +81,7 @@ mod tests { handles: Arc::new(HandleRegistry::new()), transactions: crate::transaction::TxRegistry::new(), log: iii_helpers::observability::Logger::new(), + row_changes: None, }; (st, tmp) } diff --git a/database/src/handlers/transaction.rs b/database/src/handlers/transaction.rs index 2f1f65a55..2bbb15f1a 100644 --- a/database/src/handlers/transaction.rs +++ b/database/src/handlers/transaction.rs @@ -102,6 +102,9 @@ pub async fn handle(state: &AppState, req: TxReq) -> Result { stmts.push(TxStatement { sql: s.sql, params }); } + // Kept for the row-changed announcement: `stmts` is moved into the driver. + let sql_texts: Vec = stmts.iter().map(|s| s.sql.clone()).collect(); + let result = match &pool { Pool::Sqlite(p) => driver::sqlite::transaction(p, stmts, isolation).await, Pool::Postgres(p) => driver::postgres::transaction(p, stmts, isolation).await, @@ -109,24 +112,36 @@ pub async fn handle(state: &AppState, req: TxReq) -> Result { }; match result { - Ok(steps) => Ok(TxResp { - committed: true, - results: Some( - steps - .into_iter() - .map(|s| TxStepResp { - affected_rows: s.affected_rows, - rows: s - .rows - .into_iter() - .map(|r| r.0.into_iter().map(|v| v.into_json()).collect::>()) - .collect::>(), - }) - .collect(), - ), - failed_index: None, - error: None, - }), + Ok(steps) => { + // Committed as a unit — announce each statement that changed rows, + // in statement order. Nothing is announced for a batch that rolled + // back: those rows do not exist. + for (stmt, step) in sql_texts.iter().zip(steps.iter()) { + let returned_rows = + crate::handlers::query_rows_to_objects(&step.columns, step.rows.clone()); + state + .emit_row_change(&db, stmt, step.affected_rows, Some(&returned_rows)) + .await; + } + Ok(TxResp { + committed: true, + results: Some( + steps + .into_iter() + .map(|s| TxStepResp { + affected_rows: s.affected_rows, + rows: s + .rows + .into_iter() + .map(|r| r.0.into_iter().map(|v| v.into_json()).collect::>()) + .collect::>(), + }) + .collect(), + ), + failed_index: None, + error: None, + }) + } Err(e) => { // Preserve None for non-step failures (pool acquire, BEGIN, etc.) // — those errors don't have a specific statement index, and @@ -166,6 +181,7 @@ mod tests { handles: Arc::new(HandleRegistry::new()), transactions: crate::transaction::TxRegistry::new(), log: iii_helpers::observability::Logger::new(), + row_changes: None, } } diff --git a/database/src/handlers/transaction_execute.rs b/database/src/handlers/transaction_execute.rs index 34b98ae20..ef0724fdb 100644 --- a/database/src/handlers/transaction_execute.rs +++ b/database/src/handlers/transaction_execute.rs @@ -102,7 +102,7 @@ pub async fn handle(state: &AppState, req: TxExecuteReq) -> Result { state.log.debug( "db_tx_statement", @@ -117,6 +117,16 @@ pub async fn handle(state: &AppState, req: TxExecuteReq) -> Result Result Result<()> { let handles = Arc::new(HandleRegistry::new()); let transactions = TxRegistry::new(); let log = Logger::new(); + let row_changes = Arc::new(database::triggers::RowChangeBus::new( + iii.clone(), + ROW_CHANGE_DISPATCH_TIMEOUT_MS, + )); let state = AppState { pools: Arc::new(RwLock::new(pools)), config: Arc::new(RwLock::new(cfg)), handles: handles.clone(), transactions: transactions.clone(), log: log.clone(), + row_changes: Some(row_changes.clone()), }; let _evictor = handles.spawn_evictor(); - let _tx_watcher = transactions.spawn_timeout_watcher(log.clone()); + let _tx_watcher = transactions.spawn_timeout_watcher(log.clone(), Some(row_changes.clone())); { let st = state.clone(); @@ -308,11 +316,22 @@ async fn main() -> Result<()> { ); } - let _row_change = iii.register_trigger_type(RegisterTriggerType::new( - "database::row-change", - "Postgres logical replication. Stubbed in v1.0 pending tokio-postgres replication API.", - RowChangeTrigger, - )); + // The worker announces its own writes. Registered AFTER the functions so + // the console can attribute the type, and gated on the databases that + // actually exist — a binding on a typo'd handle would listen to nothing. + let _row_changed = iii.register_trigger_type( + RegisterTriggerType::new( + database::triggers::ROW_CHANGED_TYPE, + "Fires after this worker commits a row change, filtered by `db`, optional `table`, and optional `ops`. \ + Reports only mutations made THROUGH this worker — not change data capture.", + database::triggers::RowChangedHandler { + bus: row_changes.clone(), + config: state.config.clone(), + }, + ) + .trigger_request_format::() + .call_request_format::(), + ); configuration::register_config_trigger(&iii, state.clone()) .context("registering configuration change trigger")?; diff --git a/database/src/transaction.rs b/database/src/transaction.rs index 4ab1a56fd..8818e83d0 100644 --- a/database/src/transaction.rs +++ b/database/src/transaction.rs @@ -161,7 +161,15 @@ impl TxRegistry { /// transaction whose deadline has passed, and removes the entry. The /// returned `JoinHandle` is owned by `main.rs` for the lifetime of the /// worker process; it loops forever until the runtime is dropped. - pub fn spawn_timeout_watcher(&self, log: Logger) -> tokio::task::JoinHandle<()> { + /// `row_changes` is the `database::row-changed` buffer, when one exists. + /// A transaction the watcher rolls back must have its staged changes + /// dropped with it: those rows never existed, and leaving the buffer + /// behind both leaks and risks announcing them later. + pub fn spawn_timeout_watcher( + &self, + log: Logger, + row_changes: Option>, + ) -> tokio::task::JoinHandle<()> { let me = self.clone(); tokio::spawn(async move { let mut interval = tokio::time::interval(Duration::from_secs(1)); @@ -171,12 +179,16 @@ impl TxRegistry { interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip); loop { interval.tick().await; - me.run_one_sweep(&log).await; + me.run_one_sweep(&log, row_changes.as_deref()).await; } }) } - async fn run_one_sweep(&self, log: &Logger) { + async fn run_one_sweep( + &self, + log: &Logger, + row_changes: Option<&crate::triggers::RowChangeBus>, + ) { let now = Utc::now(); // Take all expired entries in a single write-lock to keep the // critical section short; process them outside the lock. @@ -201,6 +213,9 @@ impl TxRegistry { // its ref), so dropping the guard at the end of the scope will // also drop the PinnedConn → conn returns to its pool. let mut guard = entry.conn.lock_owned().await; + if let Some(bus) = row_changes { + bus.rollback(&id); + } let result = rollback_inline(&mut guard).await; let now2 = Utc::now(); let duration_ms = (now2 - started_at).num_milliseconds().max(0); @@ -378,7 +393,7 @@ mod tests { assert_eq!(reg.len().await, 1, "entry still present before sweep"); let log = Logger::new(); - reg.run_one_sweep(&log).await; + reg.run_one_sweep(&log, None).await; assert_eq!(reg.len().await, 0, "expired entry must be removed"); // Subsequent lock fails with TRANSACTION_NOT_FOUND. @@ -389,6 +404,51 @@ mod tests { } } + #[tokio::test(flavor = "multi_thread")] + async fn timeout_drops_changes_staged_by_an_in_flight_execute() { + let reg = TxRegistry::new(); + let mut conn = sqlite_pinned().await; + if let PinnedConn::Sqlite(slot) = &mut conn { + driver::sqlite::tx_begin(slot, None).await.unwrap(); + } + let h = reg + .insert( + "primary".into(), + DriverKind::Sqlite, + conn, + Duration::from_millis(1), + ) + .await; + tokio::time::sleep(Duration::from_millis(20)).await; + let lock = reg.lock(&h.id).await.unwrap(); + let bus = Arc::new(crate::triggers::RowChangeBus::new( + Arc::new(iii_sdk::IIIClient::new("ws://127.0.0.1:9")), + 100, + )); + + let sweep_reg = reg.clone(); + let sweep_bus = bus.clone(); + let sweep = tokio::spawn(async move { + sweep_reg + .run_one_sweep(&Logger::new(), Some(sweep_bus.as_ref())) + .await; + }); + tokio::time::timeout(Duration::from_secs(1), async { + while !reg.is_empty().await { + tokio::task::yield_now().await; + } + }) + .await + .unwrap(); + + bus.stage(&h.id, "primary", "INSERT INTO t VALUES (1)", 1, None); + assert_eq!(bus.pending_count(&h.id), 1); + drop(lock); + + sweep.await.unwrap(); + assert_eq!(bus.pending_count(&h.id), 0); + } + /// Live entries (deadline still in the future) must not be touched. #[tokio::test(flavor = "multi_thread")] async fn timeout_sweep_leaves_live_entries_alone() { @@ -404,7 +464,7 @@ mod tests { ) .await; let log = Logger::new(); - reg.run_one_sweep(&log).await; + reg.run_one_sweep(&log, None).await; assert_eq!(reg.len().await, 1, "live entry must survive sweep"); } } diff --git a/database/src/triggers/bus.rs b/database/src/triggers/bus.rs new file mode 100644 index 000000000..54bd54820 --- /dev/null +++ b/database/src/triggers/bus.rs @@ -0,0 +1,568 @@ +//! The `database::row-changed` binding table and fan-out. +//! +//! The worker owns this trigger type, so it owns the bindings: the engine hands +//! each `register_trigger` to the handler, which files it here, and every +//! successful mutation looks up the matching bindings and calls their functions +//! directly (`harness/src/events.rs` does the same for turn events). +//! +//! Two rules shape it: +//! +//! * **Emit after the write is durable, never before.** A statement inside a +//! transaction has not happened until the commit does, so interactive +//! transactions buffer here and flush on commit; a rollback drops the buffer. +//! Announcing a row that then rolls back is worse than announcing nothing. +//! * **Only what this worker wrote.** This is not change data capture. A +//! mutation applied by psql, another worker, or a trigger inside the database +//! is invisible here, by construction. + +use std::collections::HashMap; +use std::sync::{Arc, Mutex}; + +use iii_sdk::protocol::TriggerRequest; +use iii_sdk::IIIClient; +use serde::{Deserialize, Serialize}; +use serde_json::Value; + +use super::sql::{self, Op}; + +/// The trigger type this worker registers. +pub const ROW_CHANGED_TYPE: &str = "database::row-changed"; + +/// Per-binding config: which database, and optionally which table. +#[derive(Debug, Clone, Deserialize, Serialize, schemars::JsonSchema)] +#[serde(deny_unknown_fields)] +pub struct RowChangedConfig { + /// Database handle, as named in the worker's config. Required — a binding + /// that watched every database would fire for traffic its owner never + /// asked about. + pub db: String, + /// Table filter. Matched case-insensitively and ignoring a schema + /// qualifier. Omit to hear every table in the database. + #[serde(default)] + pub table: Option, + /// Operation filter. Omit to hear every operation. + #[serde(default)] + pub ops: Option>, +} + +/// What a subscriber receives. +#[derive(Debug, Clone, Serialize, schemars::JsonSchema)] +pub struct RowChangedEvent { + pub db: String, + /// `null` when the statement was recognisably a write but its table could + /// not be read out of the SQL (a CTE-wrapped write, for example). + pub table: Option, + pub op: Op, + pub affected_rows: u64, + /// The `RETURNING` rows, when the caller asked for them. Absent otherwise — + /// this trigger reports that a change happened, not the new row. + #[serde(skip_serializing_if = "Option::is_none")] + pub returning: Option>>, + /// Epoch millis at emit time. + pub at: i64, +} + +#[derive(Clone, Debug)] +struct Subscriber { + instance_id: String, + function_id: String, + /// The metadata the engine stored on this binding, handed back verbatim at + /// fire time. Load-bearing, not decoration: a harness binding carries its + /// `__binding` pointer here, and a dispatch without it reaches the harness + /// with nothing to resolve — the event is simply dropped, silently. + metadata: Option, + config: SubscriberFilter, +} + +#[derive(Clone, Debug)] +struct SubscriberFilter { + db: String, + table: Option, + ops: Option>, +} + +impl SubscriberFilter { + fn matches(&self, db: &str, table: Option<&str>, op: Op) -> bool { + if self.db != db { + return false; + } + let table_matches = match (&self.table, table) { + (None, _) => true, + // A binding that named a table never matches an event whose table + // could not be determined: silence beats a wrong match. + (Some(_), None) => false, + (Some(want), Some(got)) => sql::same_table(want, got), + }; + table_matches && self.ops.as_ref().is_none_or(|ops| ops.contains(&op)) + } +} + +/// One buffered change, waiting for its transaction to commit. +#[derive(Clone, Debug)] +struct Pending { + db: String, + table: Option, + op: Op, + affected_rows: u64, + returning: Option>>, +} + +#[derive(Default)] +struct Inner { + subscribers: Vec, + /// transaction id → changes not yet committed. + pending: HashMap>, +} + +/// Everything the bus does that does not need the engine: bookkeeping and +/// matching. Kept separate so the buffer semantics — the part with real +/// consequences if wrong — are testable without a live client. +impl Inner { + fn register( + &mut self, + instance_id: String, + function_id: String, + metadata: Option, + cfg: RowChangedConfig, + ) { + // Idempotent on the engine's instance id: a re-registration replaces + // rather than doubles. + self.subscribers.retain(|s| s.instance_id != instance_id); + self.subscribers.push(Subscriber { + instance_id, + function_id, + metadata, + config: SubscriberFilter { + db: cfg.db, + table: cfg.table, + ops: cfg.ops, + }, + }); + } + + fn unregister(&mut self, instance_id: &str) { + self.subscribers.retain(|s| s.instance_id != instance_id); + } + + fn stage( + &mut self, + transaction_id: &str, + db: &str, + sql: &str, + affected_rows: u64, + returning: Option<&[serde_json::Map]>, + ) { + if affected_rows == 0 { + return; + } + let Some(mutation) = sql::classify(sql) else { + return; + }; + self.pending + .entry(transaction_id.to_string()) + .or_default() + .push(Pending { + db: db.to_string(), + table: mutation.table, + op: mutation.op, + affected_rows, + returning: returning.map(<[_]>::to_vec).filter(|r| !r.is_empty()), + }); + } + + fn take_pending(&mut self, transaction_id: &str) -> Vec { + self.pending.remove(transaction_id).unwrap_or_default() + } + + /// Matching subscribers as (function id, stored metadata) pairs. + fn targets_for(&self, db: &str, table: Option<&str>, op: Op) -> Vec<(String, Option)> { + self.subscribers + .iter() + .filter(|s| s.config.matches(db, table, op)) + .map(|s| (s.function_id.clone(), s.metadata.clone())) + .collect() + } +} + +pub struct RowChangeBus { + iii: Arc, + dispatch_timeout_ms: u64, + inner: Mutex, +} + +impl RowChangeBus { + pub fn new(iii: Arc, dispatch_timeout_ms: u64) -> Self { + Self { + iii, + dispatch_timeout_ms, + inner: Mutex::new(Inner::default()), + } + } + + fn lock(&self) -> std::sync::MutexGuard<'_, Inner> { + self.inner.lock().unwrap_or_else(|e| e.into_inner()) + } + + pub fn register( + &self, + instance_id: String, + function_id: String, + metadata: Option, + cfg: RowChangedConfig, + ) { + self.lock() + .register(instance_id, function_id, metadata, cfg); + } + + pub fn unregister(&self, instance_id: &str) { + self.lock().unregister(instance_id); + } + + pub fn subscriber_count(&self) -> usize { + self.lock().subscribers.len() + } + + #[cfg(test)] + pub(crate) fn pending_count(&self, transaction_id: &str) -> usize { + self.lock().pending.get(transaction_id).map_or(0, Vec::len) + } + + /// Announce a committed change. `sql` is classified here so callers stay + /// one line; a statement that changes no rows fires nothing. + pub async fn emit( + &self, + db: &str, + sql: &str, + affected_rows: u64, + returning: Option<&[serde_json::Map]>, + ) { + if affected_rows == 0 { + return; + } + let Some(mutation) = sql::classify(sql) else { + return; + }; + let event = RowChangedEvent { + db: db.to_string(), + table: mutation.table.clone(), + op: mutation.op, + affected_rows, + returning: returning.map(|r| r.to_vec()).filter(|r| !r.is_empty()), + at: now_ms(), + }; + self.fan_out(event).await; + } + + /// Buffer a change made inside an interactive transaction. Nothing is + /// announced until [`commit`] — the row does not exist for anyone else yet. + pub fn stage( + &self, + transaction_id: &str, + db: &str, + sql: &str, + affected_rows: u64, + returning: Option<&[serde_json::Map]>, + ) { + self.lock() + .stage(transaction_id, db, sql, affected_rows, returning); + } + + /// Flush a committed transaction's buffered changes, in statement order. + pub async fn commit(&self, transaction_id: &str) { + let staged = self.lock().take_pending(transaction_id); + for p in staged { + self.fan_out(RowChangedEvent { + db: p.db, + table: p.table, + op: p.op, + affected_rows: p.affected_rows, + returning: p.returning, + at: now_ms(), + }) + .await; + } + } + + /// Drop a rolled-back transaction's buffer. Also the timeout path: a + /// transaction the watcher auto-rolls back must not announce anything. + pub fn rollback(&self, transaction_id: &str) { + self.lock().pending.remove(transaction_id); + } + + async fn fan_out(&self, event: RowChangedEvent) { + let targets = self + .lock() + .targets_for(&event.db, event.table.as_deref(), event.op); + if targets.is_empty() { + return; + } + let payload = match serde_json::to_value(&event) { + Ok(v) => v, + Err(e) => { + tracing::warn!(error = %e, "row-changed event serialize failed"); + return; + } + }; + for (function_id, metadata) in targets { + let request = TriggerRequest { + function_id: function_id.clone(), + payload: payload.clone(), + action: Some(iii_sdk::TriggerAction::Void), + timeout_ms: Some(self.dispatch_timeout_ms), + }; + // `Void` enqueues and returns without awaiting subscriber + // execution. The stored metadata rides along — the target may + // need it to know which binding this is. + let res = match metadata { + Some(m) => self.iii.trigger(request.metadata(m)).await, + None => self.iii.trigger(request).await, + }; + if let Err(e) = res { + tracing::warn!( + function_id = %function_id, + error = %e, + "row-changed dispatch failed" + ); + } + } + } +} + +fn now_ms() -> i64 { + use std::time::{SystemTime, UNIX_EPOCH}; + SystemTime::now() + .duration_since(UNIX_EPOCH) + .map(|d| d.as_millis() as i64) + .unwrap_or(0) +} + +#[cfg(test)] +mod tests { + use super::*; + + fn filter(db: &str, table: Option<&str>) -> SubscriberFilter { + SubscriberFilter { + db: db.into(), + table: table.map(str::to_string), + ops: None, + } + } + + #[test] + fn a_table_filter_matches_case_and_schema_insensitively() { + let f = filter("primary", Some("orders")); + assert!(f.matches("primary", Some("orders"), Op::Insert)); + assert!(f.matches("primary", Some("ORDERS"), Op::Insert)); + assert!(f.matches("primary", Some("public.orders"), Op::Insert)); + assert!(!f.matches("primary", Some("order_items"), Op::Insert)); + assert!(!f.matches("analytics", Some("orders"), Op::Insert)); + } + + #[test] + fn a_db_only_filter_hears_every_table() { + let f = filter("primary", None); + assert!(f.matches("primary", Some("orders"), Op::Insert)); + assert!(f.matches("primary", Some("payments"), Op::Update)); + // Including the ones whose table could not be classified. + assert!(f.matches("primary", None, Op::Other)); + assert!(!f.matches("other", Some("orders"), Op::Delete)); + } + + #[test] + fn a_named_table_never_matches_an_unknown_one() { + // Silence beats a wrong match: a CTE-wrapped write reports no table, + // and a binding watching `orders` must not claim it. + let f = filter("primary", Some("orders")); + assert!(!f.matches("primary", None, Op::Other)); + } + + #[test] + fn an_op_filter_matches_only_selected_operations() { + let f = SubscriberFilter { + db: "primary".into(), + table: Some("orders".into()), + ops: Some(vec![Op::Insert, Op::Delete]), + }; + assert!(f.matches("primary", Some("orders"), Op::Insert)); + assert!(f.matches("primary", Some("orders"), Op::Delete)); + assert!(!f.matches("primary", Some("orders"), Op::Update)); + assert!(!f.matches("primary", Some("orders"), Op::Other)); + } + + #[test] + fn staged_changes_survive_until_commit_and_die_on_rollback() { + let mut inner = Inner::default(); + inner.stage("tx1", "primary", "INSERT INTO a (n) VALUES (1)", 1, None); + inner.stage("tx1", "primary", "UPDATE a SET n = 2", 1, None); + inner.stage("tx2", "primary", "DELETE FROM b", 3, None); + assert_eq!(inner.pending.get("tx1").map(Vec::len), Some(2)); + + // A rollback drops only its own buffer — announcing a row that rolled + // back is the failure this buffering exists to prevent. + inner.pending.remove("tx1"); + assert!(!inner.pending.contains_key("tx1")); + assert_eq!(inner.pending.get("tx2").map(Vec::len), Some(1)); + + // Taking a committed buffer drains it rather than leaking the entry. + let drained = inner.take_pending("tx2"); + assert_eq!(drained.len(), 1); + assert_eq!(drained[0].op, Op::Delete); + assert!(!inner.pending.contains_key("tx2")); + // Committing an unknown transaction yields nothing, not a panic. + assert!(inner.take_pending("never-existed").is_empty()); + } + + #[test] + fn staging_ignores_statements_that_change_no_rows() { + let mut inner = Inner::default(); + inner.stage("tx1", "primary", "SELECT * FROM a", 0, None); + inner.stage("tx1", "primary", "CREATE TABLE b (id INT)", 0, None); + inner.stage("tx1", "primary", "UPDATE a SET n = 1", 0, None); + assert!(!inner.pending.contains_key("tx1")); + } + + #[test] + fn staged_changes_keep_returning_rows() { + let mut inner = Inner::default(); + let rows = vec![serde_json::json!({ "id": 7 }).as_object().unwrap().clone()]; + inner.stage( + "tx1", + "primary", + "INSERT INTO a (id) VALUES (7) RETURNING id", + 1, + Some(&rows), + ); + assert_eq!( + inner.pending["tx1"][0].returning.as_ref().unwrap()[0]["id"], + 7 + ); + } + + #[test] + fn config_rejects_unknown_filter_keys() { + let err = serde_json::from_value::(serde_json::json!({ + "db": "primary", + "tabl": "orders" + })) + .unwrap_err(); + assert!(err.to_string().contains("unknown field"), "{err}"); + } + + #[test] + fn unknown_table_serializes_as_null() { + let value = serde_json::to_value(RowChangedEvent { + db: "primary".into(), + table: None, + op: Op::Other, + affected_rows: 1, + returning: None, + at: 0, + }) + .unwrap(); + assert_eq!(value["table"], serde_json::Value::Null); + } + + #[test] + fn registration_is_idempotent_on_the_instance_id() { + let mut inner = Inner::default(); + let cfg = || RowChangedConfig { + db: "primary".into(), + table: None, + ops: None, + }; + inner.register("i1".into(), "app::on-change".into(), None, cfg()); + inner.register("i1".into(), "app::on-change".into(), None, cfg()); + assert_eq!(inner.subscribers.len(), 1); + inner.register("i2".into(), "app::other".into(), None, cfg()); + assert_eq!(inner.subscribers.len(), 2); + inner.unregister("i1"); + assert_eq!(inner.subscribers.len(), 1); + // Unregistering something unknown is a no-op, not a panic. + inner.unregister("nope"); + assert_eq!(inner.subscribers.len(), 1); + } + + #[test] + fn fan_out_targets_only_matching_subscribers() { + let mut inner = Inner::default(); + inner.register( + "i1".into(), + "app::orders".into(), + None, + RowChangedConfig { + db: "primary".into(), + table: Some("orders".into()), + ops: None, + }, + ); + inner.register( + "i2".into(), + "app::everything".into(), + None, + RowChangedConfig { + db: "primary".into(), + table: None, + ops: None, + }, + ); + inner.register( + "i3".into(), + "app::other-db".into(), + None, + RowChangedConfig { + db: "analytics".into(), + table: None, + ops: None, + }, + ); + + let names = |db: &str, t: Option<&str>, op: Op| -> Vec { + inner + .targets_for(db, t, op) + .into_iter() + .map(|(f, _)| f) + .collect() + }; + assert_eq!( + names("primary", Some("orders"), Op::Insert), + vec!["app::orders".to_string(), "app::everything".to_string()] + ); + assert_eq!( + names("primary", Some("payments"), Op::Update), + vec!["app::everything".to_string()] + ); + // An unclassifiable table reaches only the db-wide watcher. + assert_eq!( + names("primary", None, Op::Other), + vec!["app::everything".to_string()] + ); + assert_eq!( + names("analytics", Some("orders"), Op::Delete), + vec!["app::other-db".to_string()] + ); + } + + #[test] + fn stored_metadata_rides_along_to_the_target() { + // The bug this pins: dispatching without the binding's metadata + // reaches the harness with no `__binding` to resolve, and the event is + // dropped silently — the write happened, nobody heard it. + let mut inner = Inner::default(); + inner.register( + "i1".into(), + "harness::trigger::deliver".into(), + Some(serde_json::json!({ "__binding": "sub_abc" })), + RowChangedConfig { + db: "primary".into(), + table: None, + ops: None, + }, + ); + let targets = inner.targets_for("primary", Some("orders"), Op::Insert); + assert_eq!(targets.len(), 1); + assert_eq!( + targets[0].1.as_ref().and_then(|m| m.get("__binding")), + Some(&serde_json::json!("sub_abc")) + ); + } +} diff --git a/database/src/triggers/handler.rs b/database/src/triggers/handler.rs index 27f1dce56..8f962f89e 100644 --- a/database/src/triggers/handler.rs +++ b/database/src/triggers/handler.rs @@ -1,29 +1,112 @@ -//! TriggerHandler implementations for `database::row-change`. Wired into -//! the worker via `iii.register_trigger_type` from main.rs. +//! The `database::row-changed` TriggerHandler. +//! +//! Thin by design: the engine hands a registration here, this validates the +//! config and files it in the [`RowChangeBus`]; the mutating handlers do the +//! emitting. Registration fails loudly for a database that is not configured — +//! a binding on a typo'd handle would otherwise sit there listening to nothing. + +use std::sync::Arc; use async_trait::async_trait; use iii_sdk::errors::Error; use iii_sdk::trigger::{TriggerConfig, TriggerHandler}; -fn iii_err(err: T) -> Error { - Error::Handler(serde_json::to_string(&err).unwrap_or_else(|_| "{}".into())) +use super::bus::{RowChangeBus, RowChangedConfig}; +use crate::config::WorkerConfig; + +pub struct RowChangedHandler { + pub bus: Arc, + /// Live configuration, swapped together with the pools on hot reload. + pub config: Arc>, } -/// `database::row-change` trigger handler. v1.0 stubs the streaming decoder -/// pending an upstream tokio-postgres replication API release. `register_trigger` -/// returns Unsupported so callers see a clear error instead of silently never -/// receiving events. -pub struct RowChangeTrigger; +fn config_error(message: String) -> Error { + Error::Handler(serde_json::json!({ "code": "CONFIG_ERROR", "message": message }).to_string()) +} #[async_trait] -impl TriggerHandler for RowChangeTrigger { - async fn register_trigger(&self, _config: TriggerConfig) -> Result<(), Error> { - Err(iii_err(crate::error::DbError::Unsupported { - op: "row-change".into(), - driver: "postgres (pending tokio-postgres replication API release)".into(), - })) +impl TriggerHandler for RowChangedHandler { + async fn register_trigger(&self, config: TriggerConfig) -> Result<(), Error> { + let cfg: RowChangedConfig = serde_json::from_value(config.config.clone()) + .map_err(|e| config_error(format!("row-changed config: {e}")))?; + + let live = self.config.read().await; + if !live.databases.contains_key(&cfg.db) { + let mut known = live.databases.keys().cloned().collect::>(); + known.sort(); + return Err(config_error(format!( + "unknown db `{}`; available: [{}]", + cfg.db, + known.join(", ") + ))); + } + drop(live); + + let table = cfg.table.clone(); + self.bus.register( + config.id.clone(), + config.function_id.clone(), + config.metadata.clone(), + cfg, + ); + tracing::info!( + instance = %config.id, + function = %config.function_id, + table = ?table, + "row-changed trigger registered" + ); + Ok(()) } - async fn unregister_trigger(&self, _config: TriggerConfig) -> Result<(), Error> { + + async fn unregister_trigger(&self, config: TriggerConfig) -> Result<(), Error> { + self.bus.unregister(&config.id); + tracing::info!(instance = %config.id, "row-changed trigger unregistered"); Ok(()) } } + +#[cfg(test)] +mod tests { + use super::*; + + fn trigger(id: &str, db: &str) -> TriggerConfig { + TriggerConfig { + id: id.into(), + function_id: "app::on-change".into(), + config: serde_json::json!({ "db": db }), + metadata: None, + } + } + + #[tokio::test] + async fn registration_uses_the_live_database_config() { + let config = Arc::new(tokio::sync::RwLock::new(WorkerConfig::default())); + let bus = Arc::new(RowChangeBus::new( + Arc::new(iii_sdk::IIIClient::new("ws://127.0.0.1:9")), + 100, + )); + let handler = RowChangedHandler { + bus, + config: config.clone(), + }; + + handler + .register_trigger(trigger("initial", "primary")) + .await + .unwrap(); + + let mut live = config.write().await; + let db = live.databases.remove("primary").unwrap(); + live.databases.insert("analytics".into(), db); + drop(live); + + assert!(handler + .register_trigger(trigger("removed", "primary")) + .await + .is_err()); + handler + .register_trigger(trigger("added", "analytics")) + .await + .unwrap(); + } +} diff --git a/database/src/triggers/mod.rs b/database/src/triggers/mod.rs index 1e7e0df59..7a105d931 100644 --- a/database/src/triggers/mod.rs +++ b/database/src/triggers/mod.rs @@ -1,8 +1,20 @@ -//! Trigger background tasks. Each trigger runs as its own tokio task spawned -//! at worker startup. +//! `database::row-changed` — the worker announcing its own writes. //! -//! Only `handler` is part of the public crate surface (consumed by main.rs). -//! `row_change` is an implementation module. +//! The database is often the state machine of a run: workers write rows and +//! something else needs to know. Until now nothing emitted database change +//! events, so a coordinator had to be told out-of-band (a state key written +//! alongside the row) or poll. +//! +//! This is deliberately NOT change data capture. It reports mutations THIS +//! worker performed, on commit, by classifying the SQL it was given. A write +//! applied by psql, another worker, or a database-side trigger is invisible +//! here. That covers the case it exists for — agents whose only writer is this +//! worker — and nothing more, which is why it needs no logical replication, no +//! driver-specific setup, and works identically on SQLite, Postgres and MySQL. +pub mod bus; pub mod handler; -pub(crate) mod row_change; +pub mod sql; + +pub use bus::{RowChangeBus, RowChangedConfig, RowChangedEvent, ROW_CHANGED_TYPE}; +pub use handler::RowChangedHandler; diff --git a/database/src/triggers/row_change.rs b/database/src/triggers/row_change.rs deleted file mode 100644 index aab6faf49..000000000 --- a/database/src/triggers/row_change.rs +++ /dev/null @@ -1,411 +0,0 @@ -//! row-change trigger — Postgres logical replication via pgoutput. -//! -//! v1.0 scope: -//! - Create a publication for the configured tables (idempotent, real impl). -//! - Create a logical replication slot with output plugin `pgoutput` (idempotent, real impl). -//! - Stream events; decode INSERT/UPDATE/DELETE for the configured tables. -//! - Advance LSN only on caller `ack: true`. -//! -//! IMPLEMENTATION STATUS: setup is complete (publication + slot creation are -//! tested). The streaming decode loop is a STUB — see `run_loop` and -//! `connect_replication`. The decoder belongs in this same module and consumes -//! `postgres_protocol::message::backend::LogicalReplicationMessage` from a -//! `client.copy_both_simple()` stream over a *replication-mode* connection. -//! -//! The currently-pinned `tokio-postgres = "0.7.17"` does not expose the -//! replication API (the unreleased master branch on github does). When that -//! API ships, replace `connect_replication`'s stub with a real implementation -//! and fill in `run_loop`. Reference: -//! https://github.com/sfackler/rust-postgres/blob/master/tokio-postgres/tests/test/replication.rs - -// Pre-staged setup code (`connect_for_setup`, `ensure_publication_and_slot`, -// `RowChangeConfig::validate`) is exercised by gated integration tests but -// has no production caller until the streaming decode loop ships. Allow -// dead code at the module level so the lib build is clean; the items will -// become live when `run_loop` is wired up. -#![allow(dead_code)] - -use crate::error::DbError; -use serde::{Deserialize, Serialize}; -use tokio_postgres::{Client, Config, NoTls}; - -#[derive(Debug, Clone, Deserialize)] -pub struct RowChangeConfig { - pub trigger_id: String, - #[serde(rename = "db")] - pub db_name: String, - #[serde(default = "default_schema")] - pub schema: String, - pub tables: Vec, - #[serde(default)] - pub slot_name: Option, - #[serde(default)] - pub publication_name: Option, -} - -fn default_schema() -> String { - "public".into() -} - -#[derive(Debug, Clone, Serialize)] -pub struct RowChangeEvent { - pub db: String, - pub schema: String, - pub table: String, - pub op: String, // "INSERT" | "UPDATE" | "DELETE" - pub new: Option, - pub old: Option, - pub committed_at: chrono::DateTime, - pub lsn: String, -} - -pub fn derive_names(cfg: &RowChangeConfig) -> (String, String) { - // Sanitize trigger_id for use in a Postgres identifier (slot/publication - // names accept `[A-Za-z0-9_]` only). Distinct trigger_ids can sanitize to - // the same form (`Orders.v1` and `orders-v1` both become `orders_v1`); if - // we used the sanitized form alone, two registrations would silently - // share one replication slot and consume each other's events. Append an - // FNV-1a-32 hash of the *original* trigger_id so distinct inputs always - // produce distinct outputs while collision-free identifiers stay readable. - // - // Truncate the sanitized prefix at 40 chars so the final name fits inside - // Postgres' 63-byte slot_name limit: `iii_slot_` (9) + sanitized (≤40) - // + `_` + 8 hex chars = 58. - let sanitized: String = cfg - .trigger_id - .chars() - .map(|c| { - if c.is_ascii_alphanumeric() { - c.to_ascii_lowercase() - } else { - '_' - } - }) - .take(40) - .collect(); - let h = fnv1a_32(cfg.trigger_id.as_bytes()); - let slot = cfg - .slot_name - .clone() - .unwrap_or_else(|| format!("iii_slot_{sanitized}_{h:08x}")); - let pubname = cfg - .publication_name - .clone() - .unwrap_or_else(|| format!("iii_pub_{sanitized}_{h:08x}")); - (slot, pubname) -} - -fn fnv1a_32(bytes: &[u8]) -> u32 { - let mut hash: u32 = 0x811c_9dc5; - for &b in bytes { - hash ^= b as u32; - hash = hash.wrapping_mul(0x0100_0193); - } - hash -} - -/// Open a normal (non-replication-mode) connection. Suitable for setup -/// (publication + slot creation). NOT suitable for the streaming decode loop. -pub async fn connect_for_setup( - url: &str, - tls_cfg: &crate::config::TlsConfig, -) -> Result { - // Don't echo the underlying parse error — tokio_postgres's error message - // can include the offending URL, which would leak any embedded password - // into logs. Surface a generic message instead. - let cfg: Config = url - .parse() - .map_err(|_: tokio_postgres::Error| DbError::ConfigError { - message: "postgres url parse failed; check the configured url".into(), - })?; - // Same connector as `pool::postgres`. `disable` falls back to NoTls. - let client_and_conn = match crate::pool::tls::make_pg_connector(tls_cfg)? { - Some(connector) => cfg - .connect(connector) - .await - .map(|(c, conn)| (c, futures_util::future::Either::Left(conn))), - None => cfg - .connect(NoTls) - .await - .map(|(c, conn)| (c, futures_util::future::Either::Right(conn))), - } - .map_err(crate::driver::postgres::map_err)?; - let (client, conn) = client_and_conn; - tokio::spawn(async move { - if let Err(e) = conn.await { - tracing::error!(error = ?e, "row-change setup connection terminated"); - } - }); - Ok(client) -} - -/// STUB: open a replication-mode connection. The crates.io `tokio-postgres -/// = 0.7.17` doesn't expose the replication API. When upstream cuts a -/// release with `Config::replication_mode`, replace this stub. -#[allow(dead_code)] -pub async fn connect_replication(_url: &str) -> Result { - Err(DbError::Unsupported { - op: "connect_replication".into(), - driver: "postgres (pending tokio-postgres replication API release)".into(), - }) -} - -impl RowChangeConfig { - /// Validate operator-supplied identifiers that flow into `format!()` - /// SQL strings: `slot_name`, `publication_name`, `schema`, and each - /// element of `tables` (split on `.` for qualified names). Validation - /// uses the strict ASCII identifier rule from `crate::config`. - pub fn validate(&self) -> Result<(), DbError> { - let cfg_err = |e: String| DbError::ConfigError { message: e }; - crate::config::validate_sql_identifier(&self.schema) - .map_err(|e| cfg_err(format!("row-change schema: {e}")))?; - if let Some(slot) = &self.slot_name { - crate::config::validate_sql_identifier(slot) - .map_err(|e| cfg_err(format!("row-change slot_name: {e}")))?; - } - if let Some(pubname) = &self.publication_name { - crate::config::validate_sql_identifier(pubname) - .map_err(|e| cfg_err(format!("row-change publication_name: {e}")))?; - } - for t in &self.tables { - // Qualified names allowed (`schema.table`); validate each part. - for part in t.split('.') { - crate::config::validate_sql_identifier(part) - .map_err(|e| cfg_err(format!("row-change tables entry `{t}`: {e}")))?; - } - } - Ok(()) - } -} - -pub async fn ensure_publication_and_slot( - client: &mut Client, - cfg: &RowChangeConfig, -) -> Result<(), DbError> { - cfg.validate()?; - let (slot, pubname) = derive_names(cfg); - let pub_exists = client - .query_one( - "SELECT EXISTS(SELECT 1 FROM pg_publication WHERE pubname = $1) AS ex", - &[&pubname], - ) - .await - .map_err(crate::driver::postgres::map_err)? - .get::<_, bool>("ex"); - - if !pub_exists { - let qualified: Vec = cfg - .tables - .iter() - .map(|t| { - if t.contains('.') { - t.clone() - } else { - format!("{}.{t}", cfg.schema) - } - }) - .collect(); - let stmt = format!( - "CREATE PUBLICATION {pubname} FOR TABLE {}", - qualified.join(", ") - ); - client - .simple_query(&stmt) - .await - .map_err(crate::driver::postgres::map_err)?; - } - - let slot_exists = client - .query_one( - "SELECT EXISTS(SELECT 1 FROM pg_replication_slots WHERE slot_name = $1) AS ex", - &[&slot], - ) - .await - .map_err(crate::driver::postgres::map_err)? - .get::<_, bool>("ex"); - - if !slot_exists { - let stmt = - format!("SELECT * FROM pg_create_logical_replication_slot('{slot}', 'pgoutput')"); - match client.simple_query(&stmt).await { - Ok(_) => { - tracing::info!(slot = %slot, publication = %pubname, "created replication artifacts"); - } - Err(e) => { - if e.to_string().contains("already exists") { - return Err(DbError::ReplicationSlotExists { slot }); - } else { - return Err(crate::driver::postgres::map_err(e)); - } - } - } - } - Ok(()) -} - -#[async_trait::async_trait] -pub trait QueryPollLikeDispatcher: Send + Sync { - async fn dispatch(&self, ev: RowChangeEvent) -> Result; -} - -/// STUB: streaming decoder loop. Requires a replication-mode `Client` from -/// `connect_replication` (also currently a stub). When the upstream replication -/// API ships, fill this in using `client.copy_both_simple` and -/// `postgres_protocol::message::backend::LogicalReplicationMessage::parse`. -pub async fn run_loop( - _client: Client, - _cfg: RowChangeConfig, - _dispatch: std::sync::Arc, -) -> Result<(), DbError> { - tracing::warn!( - "row-change run_loop is a stub — pgoutput decode requires the unreleased \ - tokio-postgres replication API. See module-level docstring for status." - ); - Ok(()) -} - -#[cfg(test)] -mod tests { - use super::*; - - fn url() -> Option { - std::env::var("TEST_POSTGRES_URL").ok() - } - - #[tokio::test(flavor = "multi_thread")] - async fn slot_and_publication_names_use_sanitized_trigger_id() { - let cfg = RowChangeConfig { - trigger_id: "my:trigger.id-with/funky chars".into(), - db_name: "primary".into(), - schema: "public".into(), - tables: vec!["orders".into()], - slot_name: None, - publication_name: None, - }; - let (slot, pubname) = derive_names(&cfg); - assert!(slot.starts_with("iii_slot_")); - assert!(pubname.starts_with("iii_pub_")); - // No characters that aren't [a-z0-9_]. - assert!(slot - .chars() - .all(|c| c.is_ascii_lowercase() || c.is_ascii_digit() || c == '_')); - assert!(pubname - .chars() - .all(|c| c.is_ascii_lowercase() || c.is_ascii_digit() || c == '_')); - // 63-byte slot_name limit on postgres. - assert!(slot.len() <= 63, "slot name too long: {} bytes", slot.len()); - assert!( - pubname.len() <= 63, - "publication name too long: {} bytes", - pubname.len() - ); - } - - #[tokio::test(flavor = "multi_thread")] - async fn distinct_trigger_ids_produce_distinct_slot_names() { - // Regression: prior versions sanitized trigger_id to lower-alnum-with- - // underscores; `Orders.v1`, `orders-v1`, `orders_v1`, `orders v1` all - // collapsed to `orders_v1` and silently shared one replication slot, - // letting one trigger consume another's events. Distinct trigger_ids - // must produce distinct slot/publication names. - let mk = |id: &str| RowChangeConfig { - trigger_id: id.into(), - db_name: "primary".into(), - schema: "public".into(), - tables: vec!["orders".into()], - slot_name: None, - publication_name: None, - }; - let ids = [ - "Orders.v1", - "orders-v1", - "orders_v1", - "orders v1", - "ORDERS_V1", - ]; - let mut slots = std::collections::HashSet::new(); - let mut pubs = std::collections::HashSet::new(); - for id in ids { - let (s, p) = derive_names(&mk(id)); - assert!(slots.insert(s.clone()), "slot collision on `{id}`: {s}"); - assert!(pubs.insert(p.clone()), "pub collision on `{id}`: {p}"); - } - } - - #[tokio::test(flavor = "multi_thread")] - async fn long_trigger_id_truncates_to_postgres_limit() { - // Even a pathological trigger_id must produce a valid postgres slot - // name (≤ 63 bytes). Hash suffix preserves uniqueness across the - // truncation boundary. - let cfg = RowChangeConfig { - trigger_id: "a".repeat(100), - db_name: "primary".into(), - schema: "public".into(), - tables: vec!["orders".into()], - slot_name: None, - publication_name: None, - }; - let (slot, pubname) = derive_names(&cfg); - assert!(slot.len() <= 63, "slot too long: {}", slot.len()); - assert!( - pubname.len() <= 63, - "publication too long: {}", - pubname.len() - ); - } - - #[tokio::test(flavor = "multi_thread")] - async fn explicit_slot_and_publication_names_bypass_derivation() { - let cfg = RowChangeConfig { - trigger_id: "anything".into(), - db_name: "primary".into(), - schema: "public".into(), - tables: vec!["orders".into()], - slot_name: Some("custom_slot".into()), - publication_name: Some("custom_pub".into()), - }; - let (slot, pubname) = derive_names(&cfg); - assert_eq!(slot, "custom_slot"); - assert_eq!(pubname, "custom_pub"); - } - - #[tokio::test(flavor = "multi_thread")] - async fn create_slot_and_publication_idempotent() { - let Some(u) = url() else { return }; - let cfg = RowChangeConfig { - trigger_id: "test_idem".into(), - db_name: "primary".into(), - schema: "public".into(), - tables: vec!["public.test_idem_t".into()], - slot_name: Some("iii_slot_test_idem".into()), - publication_name: Some("iii_pub_test_idem".into()), - }; - let tls = crate::config::TlsConfig { - mode: crate::config::TlsMode::Disable, - ..Default::default() - }; - let mut client = connect_for_setup(&u, &tls).await.unwrap(); - // Cleanup from prior run - let _ = client - .simple_query("SELECT pg_drop_replication_slot('iii_slot_test_idem')") - .await; - let _ = client - .simple_query("DROP PUBLICATION IF EXISTS iii_pub_test_idem") - .await; - let _ = client - .simple_query("DROP TABLE IF EXISTS public.test_idem_t") - .await; - client - .simple_query("CREATE TABLE public.test_idem_t (id SERIAL PRIMARY KEY, n INT)") - .await - .unwrap(); - - ensure_publication_and_slot(&mut client, &cfg) - .await - .unwrap(); - // Running again is idempotent. - ensure_publication_and_slot(&mut client, &cfg) - .await - .unwrap(); - } -} diff --git a/database/src/triggers/sql.rs b/database/src/triggers/sql.rs new file mode 100644 index 000000000..6368377bc --- /dev/null +++ b/database/src/triggers/sql.rs @@ -0,0 +1,425 @@ +//! Which table a statement mutates, read off the SQL text. +//! +//! This is a classifier, not a parser, and the distinction matters: it decides +//! whether an event fires and what `table` it carries, so being wrong in the +//! quiet direction — dropping an event — is the failure to avoid. Anything +//! recognisably DML produces an event; when the table cannot be read out with +//! confidence the event still fires with `table: null` and the subscriber that +//! filtered on a table simply does not match it. +//! +//! What it deliberately does not do: resolve CTEs, follow `INSERT … SELECT` +//! sources, or expand views. A statement it cannot pin down is reported as +//! unknown rather than guessed at. + +use serde::{Deserialize, Serialize}; + +/// The kind of row change a statement makes. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, schemars::JsonSchema)] +#[serde(rename_all = "lowercase")] +pub enum Op { + Insert, + Update, + Delete, + /// Recognisably a write, but not one of the three above (a CTE-wrapped + /// statement, `MERGE`, a driver-specific form). Subscribers still hear + /// about it. + Other, +} + +/// A statement's effect: what it does, and to what. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct Mutation { + pub op: Op, + /// `None` when the statement mutates something this classifier will not + /// guess at. A binding filtered on a table never matches these. + pub table: Option, +} + +/// Classify one statement. `None` means "not a row change" — a read, DDL, +/// transaction control, or anything else that cannot change rows. +pub fn classify(sql: &str) -> Option { + let stripped = strip_leading_noise(sql); + let mut words = stripped.split_whitespace(); + let first = words.next()?.to_ascii_uppercase(); + + match first.as_str() { + "INSERT" | "REPLACE" => { + // MySQL permits `INSERT t` / `REPLACE t`; INTO is optional. + let rest = strip_insert_modifiers(stripped[first.len()..].trim_start()); + Some(Mutation { + op: Op::Insert, + table: first_identifier(rest), + }) + } + "UPDATE" => { + // `UPDATE t SET …`, `UPDATE OR IGNORE t SET …`. + let after = stripped[first.len()..].trim_start(); + let after = strip_conflict_clause(after); + Some(Mutation { + op: Op::Update, + table: first_identifier(after), + }) + } + "DELETE" => { + let rest = skip_until_keyword(&stripped, "FROM")?; + Some(Mutation { + op: Op::Delete, + table: first_identifier(rest), + }) + } + // A CTE can wrap any of the above. Rather than parse the CTE list, say + // "a write happened, table unknown" — a subscriber watching the whole + // db still hears it, and one filtered on a table is not told a lie. + "WITH" if mentions_dml(&stripped) => Some(Mutation { + op: Op::Other, + table: None, + }), + "MERGE" | "UPSERT" => Some(Mutation { + op: Op::Other, + table: skip_until_keyword(&stripped, "INTO").and_then(first_identifier), + }), + _ => None, + } +} + +/// Drop leading whitespace and SQL comments so the first keyword is really the +/// first keyword. +fn strip_leading_noise(sql: &str) -> String { + let mut s = sql.trim_start(); + loop { + if let Some(rest) = s.strip_prefix("--") { + s = match rest.find('\n') { + Some(i) => rest[i + 1..].trim_start(), + None => "", + }; + } else if let Some(rest) = s.strip_prefix("/*") { + s = match rest.find("*/") { + Some(i) => rest[i + 2..].trim_start(), + None => "", + }; + } else { + break; + } + } + s.to_string() +} + +/// The text following the first standalone occurrence of `keyword`. +fn skip_until_keyword<'a>(sql: &'a str, keyword: &str) -> Option<&'a str> { + let upper = sql.to_ascii_uppercase(); + let mut from = 0usize; + while let Some(idx) = upper[from..].find(keyword) { + let start = from + idx; + let end = start + keyword.len(); + let before_ok = start == 0 + || !upper[..start] + .chars() + .next_back() + .is_some_and(|c| c.is_alphanumeric() || c == '_'); + let after_ok = upper[end..] + .chars() + .next() + .is_none_or(|c| !(c.is_alphanumeric() || c == '_')); + if before_ok && after_ok { + return Some(&sql[end..]); + } + from = end; + } + None +} + +/// The SELECT source when it is a single table. Ambiguous joins and +/// comma-separated sources deliberately produce no schema hint. +pub(crate) fn table_after_from(sql: &str) -> Option { + let stripped = strip_leading_noise(sql); + let rest = skip_until_keyword(&stripped, "FROM")?; + let end = ["WHERE", "GROUP", "ORDER", "HAVING", "LIMIT", "UNION"] + .iter() + .filter_map(|keyword| { + skip_until_keyword(rest, keyword).map(|after| rest.len() - after.len() - keyword.len()) + }) + .min() + .unwrap_or(rest.len()); + let sources = &rest[..end]; + if sources.contains(',') || skip_until_keyword(sources, "JOIN").is_some() { + return None; + } + first_identifier(sources) +} + +/// `OR REPLACE` / `OR IGNORE` / … between `UPDATE` and the table name. +fn strip_conflict_clause(s: &str) -> &str { + let Some(prefix) = s.get(..2) else { + return s; + }; + let after_or = &s[2..]; + if !prefix.eq_ignore_ascii_case("OR") + || !after_or.chars().next().is_some_and(char::is_whitespace) + { + return s; + } + let after_or = after_or.trim_start(); + let action_end = after_or.find(char::is_whitespace).unwrap_or(after_or.len()); + after_or[action_end..].trim_start() +} + +/// MySQL modifiers and optional `INTO` between INSERT/REPLACE and the table. +/// SQLite's `OR ` form is stripped first by the existing helper. +fn strip_insert_modifiers(s: &str) -> &str { + let mut rest = strip_conflict_clause(s); + loop { + let trimmed = rest.trim_start(); + let end = trimmed.find(char::is_whitespace).unwrap_or(trimmed.len()); + let word = &trimmed[..end]; + if ["LOW_PRIORITY", "DELAYED", "HIGH_PRIORITY", "IGNORE"] + .iter() + .any(|modifier| word.eq_ignore_ascii_case(modifier)) + { + rest = &trimmed[end..]; + continue; + } + if word.eq_ignore_ascii_case("INTO") { + return &trimmed[end..]; + } + return trimmed; + } +} + +/// The first identifier in `s`, unquoted. Stops at whitespace, `(`, or a +/// comma — enough for `t`, `"t"`, `` `t` ``, `[t]`, `schema.t`, `t(col,…)`. +fn first_identifier(s: &str) -> Option { + let s = s.trim_start(); + let mut out = String::new(); + let mut quote: Option = None; + for c in s.chars() { + match (quote, c) { + (Some(q), c) if c == closing(q) => quote = None, + (Some(_), c) => out.push(c), + (None, '"') | (None, '`') | (None, '[') => quote = Some(c), + (None, c) if c.is_whitespace() || c == '(' || c == ',' || c == ';' => break, + (None, c) => out.push(c), + } + } + let out = out.trim().to_string(); + (!out.is_empty()).then_some(out) +} + +fn closing(open: char) -> char { + match open { + '[' => ']', + c => c, + } +} + +fn mentions_dml(sql: &str) -> bool { + let upper = sql.to_ascii_uppercase(); + ["INSERT", "UPDATE", "DELETE", "MERGE"] + .iter() + .any(|k| skip_until_keyword(&upper, k).is_some()) +} + +/// Whether two table references name the same table, ignoring case and any +/// schema qualifier. `public.orders`, `"Orders"` and `orders` all match — a +/// subscriber should not have to guess how the writer spelled it. +pub fn same_table(a: &str, b: &str) -> bool { + fn bare(t: &str) -> String { + let t = t.rsplit('.').next().unwrap_or(t).trim(); + let t = [('"', '"'), ('`', '`'), ('[', ']')] + .into_iter() + .find_map(|(open, close)| t.strip_prefix(open)?.strip_suffix(close)) + .unwrap_or(t); + t.to_ascii_lowercase() + } + bare(a) == bare(b) +} + +#[cfg(test)] +mod tests { + use super::*; + + fn m(sql: &str) -> Option { + classify(sql) + } + + #[test] + fn table_after_from_reads_the_select_source() { + assert_eq!( + table_after_from("SELECT a, b FROM receiving_shipments WHERE x = 1"), + Some("receiving_shipments".into()) + ); + assert_eq!( + table_after_from("select * from \"Orders\" join x on 1"), + None + ); + assert_eq!(table_after_from("PRAGMA journal_mode"), None); + assert_eq!( + table_after_from("SELECT missing FROM a JOIN b ON a.id = b.id"), + None + ); + assert_eq!( + table_after_from("SELECT missing FROM a, b WHERE a.id = b.id"), + None + ); + } + + #[test] + fn the_three_ordinary_forms() { + assert_eq!( + m("INSERT INTO orders (id) VALUES (1)"), + Some(Mutation { + op: Op::Insert, + table: Some("orders".into()) + }) + ); + assert_eq!( + m("UPDATE orders SET n = 1 WHERE id = 2"), + Some(Mutation { + op: Op::Update, + table: Some("orders".into()) + }) + ); + assert_eq!( + m("DELETE FROM orders WHERE id = 2"), + Some(Mutation { + op: Op::Delete, + table: Some("orders".into()) + }) + ); + } + + #[test] + fn reads_and_ddl_are_not_row_changes() { + for sql in [ + "SELECT * FROM orders", + "CREATE TABLE orders (id INT)", + "DROP TABLE orders", + "ALTER TABLE orders ADD COLUMN n INT", + "BEGIN", + "COMMIT", + "PRAGMA foreign_keys = ON", + ] { + assert_eq!(m(sql), None, "{sql} must not fire a row change"); + } + } + + #[test] + fn conflict_clauses_do_not_become_the_table_name() { + // The bug this pins: `INSERT OR REPLACE INTO t` naively reading the + // word after INSERT would report a table called `OR`. + assert_eq!( + m("INSERT OR REPLACE INTO beats (n) VALUES (1)") + .unwrap() + .table, + Some("beats".into()) + ); + assert_eq!( + m("INSERT OR IGNORE INTO beats (n) VALUES (1)") + .unwrap() + .table, + Some("beats".into()) + ); + assert_eq!( + m("UPDATE OR ROLLBACK beats SET n = 1").unwrap().table, + Some("beats".into()) + ); + assert_eq!( + m("UPDATE OR ROLLBACK beats SET n = 1").unwrap().table, + Some("beats".into()) + ); + assert_eq!( + m("INSERT OR\nREPLACE INTO beats (n) VALUES (1)") + .unwrap() + .table, + Some("beats".into()) + ); + assert_eq!( + m("REPLACE INTO beats (n) VALUES (1)").unwrap(), + Mutation { + op: Op::Insert, + table: Some("beats".into()) + } + ); + } + + #[test] + fn mysql_insert_and_replace_allow_optional_into() { + for sql in [ + "INSERT orders (id) VALUES (1)", + "INSERT LOW_PRIORITY IGNORE orders (id) VALUES (1)", + "REPLACE orders (id) VALUES (1)", + "REPLACE DELAYED INTO orders (id) VALUES (1)", + ] { + assert_eq!(m(sql).unwrap().table.as_deref(), Some("orders"), "{sql}"); + } + } + + #[test] + fn quoted_and_qualified_identifiers_unwrap() { + for (sql, want) in [ + (r#"INSERT INTO "Orders" (id) VALUES (1)"#, "Orders"), + ("INSERT INTO `orders` (id) VALUES (1)", "orders"), + ("INSERT INTO [orders] (id) VALUES (1)", "orders"), + ("INSERT INTO public.orders (id) VALUES (1)", "public.orders"), + ("INSERT INTO orders(id) VALUES (1)", "orders"), + ("DELETE FROM orders;", "orders"), + ] { + assert_eq!(m(sql).unwrap().table.as_deref(), Some(want), "{sql}"); + } + } + + #[test] + fn leading_comments_do_not_hide_the_verb() { + assert_eq!( + m("-- seed the table\nINSERT INTO orders (id) VALUES (1)") + .unwrap() + .table, + Some("orders".into()) + ); + assert_eq!( + m("/* batch 2 */ DELETE FROM orders").unwrap().op, + Op::Delete + ); + } + + #[test] + fn lowercase_and_ragged_whitespace_still_classify() { + assert_eq!( + m(" insert\n into\n orders (id) values (1)").unwrap(), + Mutation { + op: Op::Insert, + table: Some("orders".into()) + } + ); + } + + #[test] + fn a_cte_wrapped_write_fires_with_an_unknown_table() { + // Not parsed — but a write did happen, and dropping the event is worse + // than reporting it without a table. + let got = m("WITH moved AS (DELETE FROM a RETURNING *) INSERT INTO b SELECT * FROM moved") + .unwrap(); + assert_eq!(got.op, Op::Other); + assert_eq!(got.table, None); + // A read-only CTE is still not a row change. + assert_eq!(m("WITH x AS (SELECT 1) SELECT * FROM x"), None); + } + + #[test] + fn table_matching_ignores_case_and_schema() { + assert!(same_table("orders", "ORDERS")); + assert!(same_table("public.orders", "orders")); + assert!(same_table("orders", "public.orders")); + assert!(same_table("\"Orders\"", "orders")); + assert!(same_table("`orders`", "orders")); + assert!(same_table("[orders]", "orders")); + assert!(!same_table("orders", "order_items")); + } + + #[test] + fn a_statement_naming_no_table_is_reported_not_dropped() { + // `INSERT INTO` with nothing after it is malformed SQL the driver will + // reject, but the classifier must not panic or invent a name. + assert_eq!(m("INSERT INTO").unwrap().table, None); + assert_eq!(m("DELETE FROM ").unwrap().table, None); + } +} diff --git a/database/tests/e2e/README.md b/database/tests/e2e/README.md index a6bdb4823..c3e8534e0 100644 --- a/database/tests/e2e/README.md +++ b/database/tests/e2e/README.md @@ -3,9 +3,10 @@ Self-asserting smoke harness for the `database` worker. Validates the function surface (query / execute / prepareStatement / runStatement / transaction), the **interactive-transaction** surface (begin / -transactionQuery / transactionExecute / commit / rollback), the -`row-change` slot/publication derivation contract, and the side-channel- -finalization repros from the `/review` of branch `feat/database-and-skills` +transactionQuery / transactionExecute / commit / rollback), +`database::row-changed` delivery, and the side-channel-finalization repros +from the `/review` of branch +`feat/database-and-skills` against real **SQLite**, **PostgreSQL 16**, and **MySQL 8.4** with one command. @@ -34,7 +35,7 @@ Runs locally and in CI (`.github/workflows/database-e2e.yml`). ``` Builds the worker (`cargo build --release --bin database`), brings up -the docker stack with `wal_level=logical`, starts the engine, seeds the +the docker stack, starts the engine, seeds the `database` configuration entry, starts the database worker, and runs the selected case groups across all 3 drivers. Exits 0 on PASS, 1 on any FAIL. @@ -71,6 +72,8 @@ overridden: | `III_BIN` | `$(command -v iii)` then `$HOME/.local/bin/iii` | Engine binary | | `WORKER_BIN_TARGET` | `$WORKER_SRC/target/release/database` | Built worker | | `WORKER_BIN_LINK` | `$HOME/.iii/workers/database` | Symlink the engine reads | +| `TEST_POSTGRES_URL` | `postgres://iii:iii@127.0.0.1:55432/iii_test` | Override the PostgreSQL E2E database | +| `TEST_MYSQL_URL` | `mysql://iii:iii@127.0.0.1:53306/iii_test` | Override the MySQL E2E database | | `COMPOSE` | `docker compose` | Compose command. Set to `podman-compose` for rootless podman; the script auto-switches its healthcheck strategy to `podman inspect` (since podman-compose 1.x doesn't implement compose v2's `--wait`). | | `HARNESS_MODE` | `full` | `full` / `no-bypass` / `bypass-only`. Set by the flags above; you usually don't need to set this directly. | | `HARNESS_TIMEOUT` | `180` | Seconds to wait for the test sentinel | @@ -104,13 +107,14 @@ accepted; outside-tx COUNT=1`). | File | Role | |---|---| | `run-tests.sh` | Orchestrator | -| `docker-compose.yml` | Postgres (wal_level=logical) + MySQL with healthchecks | +| `docker-compose.yml` | Postgres + MySQL with healthchecks | | `config.yaml` | Engine infra only (queue, observability) | | `workers/harness/src/seed-configuration.ts` | Bootstrap: `configuration::register` for id `database` | | `workers/harness/src/database-config.ts` | E2e `databases` value (sqlite + pg + mysql) | | `workers/harness/fixtures/database.schema.json` | JSON Schema fixture (sync with Rust via export test) | | `workers/harness/` | TypeScript smoke-test worker (runs as a host process) | | `workers/harness/src/cases-interactive-tx.ts` | Interactive-transaction lifecycle cases | +| `workers/harness/src/cases-row-changed.ts` | Row-change trigger delivery cases | | `workers/harness/src/cases-tx-control-bypass.ts` | Side-channel-finalization repros | | `reports/report.json` | Per-case results (latest run) | diff --git a/database/tests/e2e/docker-compose.yml b/database/tests/e2e/docker-compose.yml index 0ac998b27..c8aff852d 100644 --- a/database/tests/e2e/docker-compose.yml +++ b/database/tests/e2e/docker-compose.yml @@ -5,14 +5,6 @@ services: POSTGRES_USER: iii POSTGRES_PASSWORD: iii POSTGRES_DB: iii_test - command: - - postgres - - -c - - wal_level=logical - - -c - - max_wal_senders=4 - - -c - - max_replication_slots=4 ports: - "55432:5432" volumes: diff --git a/database/tests/e2e/run-tests.sh b/database/tests/e2e/run-tests.sh index 2d4361fbf..07e100fc0 100755 --- a/database/tests/e2e/run-tests.sh +++ b/database/tests/e2e/run-tests.sh @@ -180,11 +180,10 @@ fi echo "[run-tests] both services healthy" # 5. (Optional) Run cargo unit + integration tests against the live DBs. -# The 5 gated tests in src/{driver,pool}/{postgres,mysql}.rs and -# src/triggers/row_change.rs early-return when TEST_*_URL is unset; we set -# both here so they exercise real connections, binary param encoding, -# replication-slot creation, etc — finer-grained than the e2e harness -# alone. CI passes --with-cargo-test; local runs skip this by default. +# The gated tests in src/{driver,pool}/{postgres,mysql}.rs early-return when +# TEST_*_URL is unset; we set both here so they exercise real connections, +# binary param encoding, etc — finer-grained than the e2e harness alone. +# CI passes --with-cargo-test; local runs skip this by default. if [[ "$WITH_CARGO_TEST" -eq 1 ]]; then echo "[run-tests] cargo test --all-features (with TEST_POSTGRES_URL + TEST_MYSQL_URL)" ( diff --git a/database/tests/e2e/workers/harness/src/cases-interactive-tx.ts b/database/tests/e2e/workers/harness/src/cases-interactive-tx.ts index 16bc53ab7..780ba31dc 100644 --- a/database/tests/e2e/workers/harness/src/cases-interactive-tx.ts +++ b/database/tests/e2e/workers/harness/src/cases-interactive-tx.ts @@ -83,6 +83,56 @@ export const INTERACTIVE_TX_CASES: TestCase[] = [ } }, }, + { + // PostgreSQL reports COMMIT on an aborted transaction with the command + // tag ROLLBACK. The worker must surface an error instead of treating that + // as a commit and flushing bookkeeping for writes that never became durable. + name: 'postgres aborted interactive tx rejects commit', + applies: ['pg_db'], + async run({ driver, call, expectError }) { + await call('database::execute', { db: driver, sql: 'DROP TABLE IF EXISTS itx_aborted' }) + await call('database::execute', { + db: driver, + sql: 'CREATE TABLE itx_aborted (n INT NOT NULL)', + }) + let id: string | undefined + try { + const begin = await call('database::beginTransaction', { db: driver }) + id = begin.transaction.id + await call('database::transactionExecute', { + transaction_id: id, + sql: 'INSERT INTO itx_aborted VALUES (1)', + }) + await expectError( + () => + call('database::transactionExecute', { + transaction_id: id, + sql: 'INSERT INTO itx_aborted VALUES (NULL)', + }), + 'DRIVER_ERROR', + ) + await expectError( + () => call('database::commitTransaction', { transaction_id: id }), + 'DRIVER_ERROR', + ) + + const verify = await call('database::query', { + db: driver, + sql: 'SELECT COUNT(*) AS c FROM itx_aborted', + }) + expectEqual(Number(verify.rows[0].c), 0, 'aborted transaction persisted no rows') + } finally { + if (id) { + try { + await call('database::rollbackTransaction', { transaction_id: id }) + } catch { + /* commit failure already finalized it */ + } + } + await call('database::execute', { db: driver, sql: 'DROP TABLE itx_aborted' }) + } + }, + }, { name: 'interactive tx commit then query returns TRANSACTION_NOT_FOUND', async run({ driver, call, expectError }) { diff --git a/database/tests/e2e/workers/harness/src/cases-row-change.ts b/database/tests/e2e/workers/harness/src/cases-row-change.ts deleted file mode 100644 index ecc691079..000000000 --- a/database/tests/e2e/workers/harness/src/cases-row-change.ts +++ /dev/null @@ -1,173 +0,0 @@ -import type { TestCase } from './cases.ts' -import { expect, expectEqual } from './cases.ts' - -/** - * row-change trigger validation. The streaming decoder is stubbed in v1.0 - * (worker rejects `database::row-change` registration with `UNSUPPORTED`), - * so we can't exercise the dispatch path end-to-end yet. What we CAN validate - * is the slot/publication name derivation contract that the worker pins in - * its README — distinct caller-supplied `trigger_id`s must produce distinct - * Postgres replication-slot names so two registrations don't silently share - * one slot once the streaming runtime ships. - * - * Pre-fix: `derive_names` lowercased and replaced non-alnum with `_`, so - * `orders-v1` and `orders.v1` both became `orders_v1` and the second - * registration would silently reuse the first slot. Post-fix: an FNV-1a-32 - * hash of the original trigger_id is appended, guaranteeing uniqueness. - * - * This file mirrors the Rust `derive_names` algorithm in TS so we can - * compute the same names the worker would and assert Postgres treats them - * as distinct identifiers via `pg_create_logical_replication_slot`. - */ - -/** FNV-1a-32 over UTF-8 bytes. Mirrors `triggers/row_change.rs::fnv1a_32`. */ -function fnv1a32(s: string): string { - let hash = 0x811c9dc5 >>> 0 - const bytes = Buffer.from(s, 'utf8') - for (const b of bytes) { - hash = (hash ^ b) >>> 0 - hash = Math.imul(hash, 0x01000193) >>> 0 - } - return hash.toString(16).padStart(8, '0') -} - -/** - * TS port of `triggers/row_change.rs::derive_names`. Lowercases ASCII - * alphanumerics, replaces every other char with `_`, truncates the - * sanitized prefix at 40 chars to fit Postgres' 63-byte slot_name limit, - * and appends an 8-hex-char FNV-1a-32 hash of the *original* trigger_id. - */ -function deriveSlotName(triggerId: string): string { - const sanitized = Array.from(triggerId) - .map((c) => (/[a-zA-Z0-9]/.test(c) ? c.toLowerCase() : '_')) - .slice(0, 40) - .join('') - return `iii_slot_${sanitized}_${fnv1a32(triggerId)}` -} - -async function dropSlotIfExists( - call: (id: string, payload: unknown) => Promise, - driver: string, - slot: string, -): Promise { - // pg_drop_replication_slot errors if the slot is missing; pre-check then drop. - // Quote-escape the slot name as a SQL literal: replace any `'` with `''` - // (slot names from derive_names are `[a-z0-9_]` only, so this is defensive). - const lit = slot.replace(/'/g, "''") - const exists = await call('database::query', { - db: driver, - sql: `SELECT 1 FROM pg_replication_slots WHERE slot_name = '${lit}'`, - }) - if (exists.row_count > 0) { - await call('database::execute', { - db: driver, - sql: `SELECT pg_drop_replication_slot('${lit}')`, - }) - } -} - -export const ROW_CHANGE_CASES: TestCase[] = [ - { - name: 'row-change derive_names: collision-prone trigger_ids produce distinct postgres slots', - applies: ['pg_db'], - async run({ driver, call }) { - // These three inputs all sanitized to `orders_v1` in the pre-fix code - // (lowercase + replace non-alnum with `_`). Post-fix, the appended hash - // makes them distinct. We use 3 (not the full 5) because the docker - // postgres image is configured with `max_replication_slots=4`, leaving - // headroom for the long-trigger-id test that runs immediately after. - const ids = ['Orders.v1', 'orders-v1', 'orders v1'] - const slots = ids.map(deriveSlotName) - - // Sanity: TS-derived names must all be distinct. - const unique = new Set(slots) - expectEqual(unique.size, ids.length, 'TS-derived slot names must be unique across collision-prone inputs') - - // Each slot must respect Postgres' 63-byte limit. - for (const s of slots) { - expect(s.length <= 63, `slot name too long (${s.length} bytes): ${s}`) - } - - // Pre-clean any leftovers from a previous run. - for (const slot of slots) { - await dropSlotIfExists(call, driver, slot) - } - - try { - // Create all five slots. If two collided, the second create call would - // fail with `replication slot ... already exists`. - for (const slot of slots) { - await call('database::execute', { - db: driver, - sql: `SELECT * FROM pg_create_logical_replication_slot('${slot}', 'pgoutput')`, - }) - } - - // Verify Postgres now lists all five as distinct slots. - const inList = slots.map((s) => `'${s}'`).join(', ') - const q = await call('database::query', { - db: driver, - sql: `SELECT slot_name FROM pg_replication_slots WHERE slot_name IN (${inList}) ORDER BY slot_name`, - }) - expectEqual(q.row_count, ids.length, 'all collision-prone inputs produced distinct slots in postgres') - } finally { - // Cleanup so re-running the harness against the same docker volume is idempotent. - for (const slot of slots) { - try { - await dropSlotIfExists(call, driver, slot) - } catch { - /* best-effort cleanup */ - } - } - } - }, - }, - { - name: 'row-change derive_names: long trigger_id stays within postgres slot-name limit', - applies: ['pg_db'], - async run({ driver, call }) { - // Pathological trigger_id: 200 chars. Without truncation the derived - // name would exceed Postgres' 63-byte slot_name cap and slot creation - // would fail; the hash suffix preserves uniqueness across the truncation. - const a = 'a'.repeat(200) - const b = 'a'.repeat(200) + 'b' // distinct trigger_id, same first-40 sanitized prefix - const slotA = deriveSlotName(a) - const slotB = deriveSlotName(b) - - expect(slotA !== slotB, `long trigger_ids collided: ${slotA}`) - expect(slotA.length <= 63, `slotA too long (${slotA.length}): ${slotA}`) - expect(slotB.length <= 63, `slotB too long (${slotB.length}): ${slotB}`) - - // Pre-clean. - await dropSlotIfExists(call, driver, slotA) - await dropSlotIfExists(call, driver, slotB) - - try { - await call('database::execute', { - db: driver, - sql: `SELECT * FROM pg_create_logical_replication_slot('${slotA}', 'pgoutput')`, - }) - await call('database::execute', { - db: driver, - sql: `SELECT * FROM pg_create_logical_replication_slot('${slotB}', 'pgoutput')`, - }) - const q = await call('database::query', { - db: driver, - sql: `SELECT slot_name FROM pg_replication_slots WHERE slot_name IN ('${slotA}', '${slotB}')`, - }) - expectEqual(q.row_count, 2, 'long-trigger-id slots created and distinct') - } finally { - try { - await dropSlotIfExists(call, driver, slotA) - } catch { - /* best-effort */ - } - try { - await dropSlotIfExists(call, driver, slotB) - } catch { - /* best-effort */ - } - } - }, - }, -] diff --git a/database/tests/e2e/workers/harness/src/cases-row-changed.ts b/database/tests/e2e/workers/harness/src/cases-row-changed.ts new file mode 100644 index 000000000..736e870c0 --- /dev/null +++ b/database/tests/e2e/workers/harness/src/cases-row-changed.ts @@ -0,0 +1,211 @@ +import type { TestCase } from './cases.ts' +import { expect, expectEqual } from './cases.ts' + +const EVENT_TIMEOUT_MS = 5_000 +const SILENCE_WINDOW_MS = 500 + +interface RowChangedEvent { + db: string + table: string | null + op: 'insert' | 'update' | 'delete' | 'other' + affected_rows: number + returning?: Record[] + at: number +} + +const sleep = (ms: number) => new Promise((resolve) => setTimeout(resolve, ms)) + +export const ROW_CHANGED_CASES: TestCase[] = [ + { + name: 'row-changed filters ops and emits committed mutations only', + async run({ driver, dialect, call, iii }) { + const table = 'e2e_row_changed' + const functionId = `harness::row_changed_${driver}` + const insertFunctionId = `${functionId}_inserts` + const events: RowChangedEvent[] = [] + const insertEvents: RowChangedEvent[] = [] + let cursor = 0 + let insertCursor = 0 + let activeTransaction: string | undefined + + await call('database::execute', { db: driver, sql: `DROP TABLE IF EXISTS ${table}` }) + await call('database::execute', { + db: driver, + sql: `CREATE TABLE ${table} (id ${dialect.idColumnDDL()}, n INT NOT NULL)`, + }) + + const functionRef = iii.registerFunction( + functionId, + async (payload: RowChangedEvent) => { + events.push(payload) + return null + }, + { description: 'Database row-changed E2E event sink.' }, + ) + const triggerRef = iii.registerTrigger({ + type: 'database::row-changed', + function_id: functionId, + config: { db: driver, table }, + }) + const insertFunctionRef = iii.registerFunction( + insertFunctionId, + async (payload: RowChangedEvent) => { + insertEvents.push(payload) + return null + }, + { description: 'Database row-changed insert-only E2E event sink.' }, + ) + const insertTriggerRef = iii.registerTrigger({ + type: 'database::row-changed', + function_id: insertFunctionId, + config: { db: driver, table, ops: ['insert'] }, + }) + + const nextEvent = async (): Promise => { + const deadline = Date.now() + EVENT_TIMEOUT_MS + while (events.length <= cursor && Date.now() < deadline) await sleep(20) + if (events.length <= cursor) throw new Error(`row-changed event ${cursor + 1} was not delivered`) + return events[cursor++] + } + const nextInsertEvent = async (): Promise => { + const deadline = Date.now() + EVENT_TIMEOUT_MS + while (insertEvents.length <= insertCursor && Date.now() < deadline) await sleep(20) + if (insertEvents.length <= insertCursor) { + throw new Error(`insert-filtered row-changed event ${insertCursor + 1} was not delivered`) + } + return insertEvents[insertCursor++] + } + const expectSilence = async (): Promise => { + await sleep(SILENCE_WINDOW_MS) + expectEqual(events.length, cursor, 'row-changed emitted an unexpected event') + expectEqual( + insertEvents.length, + insertCursor, + 'insert-filtered row-changed emitted an unexpected event', + ) + } + const expectEvent = (event: RowChangedEvent, op: RowChangedEvent['op'], affectedRows = 1): void => { + expectEqual(event.db, driver, 'row-changed db') + expectEqual(event.table, table, 'row-changed table') + expectEqual(event.op, op, 'row-changed op') + expectEqual(event.affected_rows, affectedRows, 'row-changed affected_rows') + expect(Number.isFinite(event.at) && event.at > 0, 'row-changed at is an epoch timestamp') + } + + try { + const registered = await iii.trigger< + Record, + { registered_triggers: Array<{ trigger_type: string; function_id: string }> } + >({ function_id: 'engine::registered-triggers::list', payload: {} }) + for (const expectedFunction of [functionId, insertFunctionId]) { + expect( + registered.registered_triggers.some( + (trigger) => + trigger.trigger_type === 'database::row-changed' && + trigger.function_id === expectedFunction, + ), + `row-changed trigger registration for ${expectedFunction} is visible to the engine`, + ) + } + + const p1 = dialect.placeholder(1) + const p2 = dialect.placeholder(2) + const returning = driver === 'mysql_db' ? [] : ['id', 'n'] + const returningSql = returning.length > 0 ? ' RETURNING id, n' : '' + // MySQL permits INSERT without INTO; using that form here also pins + // the classifier regression while SQLite/PostgreSQL use standard SQL. + const insertPrefix = driver === 'mysql_db' ? 'INSERT' : 'INSERT INTO' + + await call('database::execute', { + db: driver, + sql: `${insertPrefix} ${table} (n) VALUES (${p1})${returningSql}`, + params: [10], + returning, + }) + const inserted = await nextEvent() + expectEvent(inserted, 'insert') + expectEvent(await nextInsertEvent(), 'insert') + if (returning.length > 0) { + expectEqual(Number(inserted.returning?.[0]?.n), 10, 'row-changed direct RETURNING value') + } + + await call('database::transaction', { + db: driver, + statements: [ + { + sql: `INSERT INTO ${table} (n) VALUES (${p1})${returningSql}`, + params: [15], + }, + ], + }) + const atomic = await nextEvent() + expectEvent(atomic, 'insert') + expectEvent(await nextInsertEvent(), 'insert') + if (returning.length > 0) { + expectEqual(Number(atomic.returning?.[0]?.n), 15, 'row-changed atomic RETURNING value') + } + + await call('database::execute', { + db: driver, + sql: `UPDATE ${table} SET n = ${p1} WHERE n = ${p2}`, + params: [11, 10], + }) + expectEvent(await nextEvent(), 'update') + + await call('database::execute', { + db: driver, + sql: `DELETE FROM ${table} WHERE n = ${p1}`, + params: [11], + }) + expectEvent(await nextEvent(), 'delete') + + await call('database::execute', { + db: driver, + sql: `UPDATE ${table} SET n = ${p1} WHERE n = ${p2}`, + params: [99, -1], + }) + await expectSilence() + + activeTransaction = (await call('database::beginTransaction', { db: driver })).transaction.id + await call('database::transactionExecute', { + transaction_id: activeTransaction, + sql: `INSERT INTO ${table} (n) VALUES (${p1})${returningSql}`, + params: [20], + returning, + }) + await expectSilence() + await call('database::commitTransaction', { transaction_id: activeTransaction }) + activeTransaction = undefined + const committed = await nextEvent() + expectEvent(committed, 'insert') + expectEvent(await nextInsertEvent(), 'insert') + if (returning.length > 0) { + expectEqual(Number(committed.returning?.[0]?.n), 20, 'row-changed committed RETURNING value') + } + + activeTransaction = (await call('database::beginTransaction', { db: driver })).transaction.id + await call('database::transactionExecute', { + transaction_id: activeTransaction, + sql: `INSERT INTO ${table} (n) VALUES (${p1})`, + params: [30], + }) + await call('database::rollbackTransaction', { transaction_id: activeTransaction }) + activeTransaction = undefined + await expectSilence() + } finally { + if (activeTransaction) { + try { + await call('database::rollbackTransaction', { transaction_id: activeTransaction }) + } catch { + /* transaction may already be finalized */ + } + } + insertTriggerRef.unregister() + insertFunctionRef.unregister() + triggerRef.unregister() + functionRef.unregister() + await call('database::execute', { db: driver, sql: `DROP TABLE IF EXISTS ${table}` }) + } + }, + }, +] diff --git a/database/tests/e2e/workers/harness/src/database-config.ts b/database/tests/e2e/workers/harness/src/database-config.ts index 510321a45..e889a56c9 100644 --- a/database/tests/e2e/workers/harness/src/database-config.ts +++ b/database/tests/e2e/workers/harness/src/database-config.ts @@ -18,14 +18,14 @@ export const DATABASE_CONFIG_VALUE = { pool: { ...DEFAULT_POOL }, }, pg_db: { - url: 'postgres://iii:iii@127.0.0.1:55432/iii_test', + url: process.env.TEST_POSTGRES_URL ?? 'postgres://iii:iii@127.0.0.1:55432/iii_test', pool: { ...DEFAULT_POOL }, // Local docker postgres uses a self-signed cert that doesn't chain to // any system CA. The worker's tls.mode defaults to require. tls: { mode: 'disable' as const }, }, mysql_db: { - url: 'mysql://iii:iii@127.0.0.1:53306/iii_test', + url: process.env.TEST_MYSQL_URL ?? 'mysql://iii:iii@127.0.0.1:53306/iii_test', pool: { ...DEFAULT_POOL }, // Local docker mysql:8.4 ships an auto-generated self-signed cert. tls: { mode: 'disable' as const }, diff --git a/database/tests/e2e/workers/harness/src/runner.ts b/database/tests/e2e/workers/harness/src/runner.ts index 52fcc1390..937ebaa0f 100644 --- a/database/tests/e2e/workers/harness/src/runner.ts +++ b/database/tests/e2e/workers/harness/src/runner.ts @@ -8,8 +8,8 @@ import { PROTOCOL_CASES } from './cases-protocol.ts' import { TRANSACTION_EDGE_CASES } from './cases-transaction.ts' import { INTERACTIVE_TX_CASES } from './cases-interactive-tx.ts' import { CONCURRENCY_CASES } from './cases-concurrency.ts' -import { ROW_CHANGE_CASES } from './cases-row-change.ts' import { TX_CONTROL_BYPASS_CASES } from './cases-tx-control-bypass.ts' +import { ROW_CHANGED_CASES } from './cases-row-changed.ts' interface CaseResult { driver: DriverKey @@ -144,16 +144,16 @@ export class Runner { record(await this.runCase(driver, c)) } - // Boundary, protocol, transaction-edge, interactive-tx, concurrency, - // row-change cases. Each test is self-contained (creates and drops - // its own scratch tables / replication slots) so order doesn't matter. + // Boundary, protocol, transaction-edge, interactive-tx, and + // concurrency cases. Each test is self-contained (creates and drops + // its own scratch tables) so order doesn't matter. for (const c of [ ...BOUNDARY_CASES, ...PROTOCOL_CASES, ...TRANSACTION_EDGE_CASES, ...INTERACTIVE_TX_CASES, ...CONCURRENCY_CASES, - ...ROW_CHANGE_CASES, + ...ROW_CHANGED_CASES, ]) { if (!matchesDriver(driver, c)) continue record(await this.runCase(driver, c)) diff --git a/database/tests/integration.rs b/database/tests/integration.rs index 5cd7b968f..40438b5c6 100644 --- a/database/tests/integration.rs +++ b/database/tests/integration.rs @@ -33,6 +33,7 @@ async fn build_state() -> AppState { handles: Arc::new(HandleRegistry::new()), transactions: TxRegistry::new(), log: Logger::new(), + row_changes: None, } } @@ -219,6 +220,7 @@ async fn build_pool_creates_missing_sqlite_parent_dir() { handles: Arc::new(HandleRegistry::new()), transactions: TxRegistry::new(), log: Logger::new(), + row_changes: None, }; // The freshly-created on-disk db is usable end-to-end.