diff --git a/.decapod/managed/specs/INTERFACES.md b/.decapod/managed/specs/INTERFACES.md index f0946c3..27ed831 100644 --- a/.decapod/managed/specs/INTERFACES.md +++ b/.decapod/managed/specs/INTERFACES.md @@ -103,7 +103,7 @@ pub enum ApiError { - Parameters are always adapter-bound, never interpolated into SQL. `Parameter` enumerates `Null` / `Bool` / `Integer` / `Real` / `Text`. - `execute` is the caller-owned schema surface: dactyl never silently creates tables. - `Row` provides strict `get` / `try_get`, lenient `get_bool` / `get_int` / `get_real` / `get_str` / `get_json`, borrowed `get_str_ref` / `get_json_ref`, and `is_null`, with explicit `ColumnNotFound` / `Conversion` errors. Named lookup is left-to-right first-match for duplicate aliases. SQL NULL maps to `Option` or a `Conversion` that mentions NULL for non-Option targets. Rows own their cells; borrowed accessors are tied to `&Row` only (dactyl #25 / #2; DecapodLabs/decapod#1111). -- `transaction` is atomic: any per-statement failure rolls back the whole unit on SQLite and is rejected by the Neon `/batch` endpoint. +- `transaction` is atomic: any per-statement failure rolls back the whole unit on SQLite and is rejected by the Neon `/batch` endpoint (dactyl #24). Nesting is not supported (no SAVEPOINT). dactyl does not retry and exposes no deadline parameter; callers own retry/idempotency after ambiguous transport failures. Empty batch → `Ok([])`. Conformance proves failure-injection on SQLite and Neon mock plus an event-plus-state fixture. - `query!("sql")` lexically analyzes the literal at compile time and returns the rewritten SQL as a `String` for the caller to pass to `query`. ### Multi-backend vision diff --git a/CHANGELOG.md b/CHANGELOG.md index 659df1d..5e7763f 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -10,10 +10,12 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Added - *(dactyl)* complete typed/NULL-safe named row projections for [#25](https://github.com/DecapodLabs/dactyl/issues/25): `try_get`, `is_null`, borrowed `get_str_ref` / `get_json_ref`, explicit first-match duplicate-alias semantics, NULL conversion messages, unit + SQLite/Neon matrix conformance (also DecapodLabs/decapod#1111, dactyl #2). +- *(dactyl)* complete atomic batch contract for [#24](https://github.com/DecapodLabs/dactyl/issues/24): Neon-mock failure-injection, event-plus-state fixture on both adapters, nesting/retry/timeout/idempotency docs; neon adapter surfaces non-2xx batch bodies without requiring a success-shaped decode. ### Documentation - *(dactyl)* document the full `Row` projection contract (scalars, NULL, missing columns, aliases, ownership/lifetime) in the README and crate docs. +- *(dactyl)* document `transaction` atomicity, nesting, retry, timeout, and idempotency semantics in README and crate docs. ## [0.2.3](https://github.com/DecapodLabs/dactyl/compare/dactyl-db-macros-v0.2.1...dactyl-db-macros-v0.2.3) - 2026-08-01 diff --git a/README.md b/README.md index 6907778..628699e 100644 --- a/README.md +++ b/README.md @@ -9,7 +9,7 @@ The same SQL string produces the same logical rows regardless of the active back - **One import, many backends** — SQLite and Neon ship today; Redis, MySQL, and Cassandra are planned behind the same `query` surface. - **Ambient selection** — `DATASTORE` env var picks the active backend at runtime. No `init()`, no per-call datastore argument, no global connection cache. - **Safe parameter binding** — `query(sql, &[params])` binds typed values; SQL injection via parameter values is structurally impossible. -- **Atomic batches** — `transaction(&[Statement])` commits all-or-nothing on every backend. +- **Atomic batches** — `transaction(&[Statement])` commits all-or-nothing on every backend (no nesting; caller-owned retry/idempotency; see contract below). - **Caller-owned schema** — dactyl never silently creates tables. `execute("create table ...")` is the only way dactyl touches schema. ## Quick Start @@ -107,6 +107,19 @@ The active backend is chosen by ambient environment variables — no `init()` ca `Row` provides `get` / `try_get`, lenient scalar getters, `is_null`, and borrowed `get_str_ref` / `get_json_ref` under the projection contract above. +## Atomic batches (`transaction`) + +Stable contract for multi-statement units of work ([#24](https://github.com/DecapodLabs/dactyl/issues/24); prerequisite for DecapodLabs/decapod#1111 / #1120): + +| Concern | Semantics | +|---|---| +| Atomicity | Any per-statement failure aborts the whole unit. SQLite uses a real transaction; Neon uses one `POST /batch` that the server accepts or rejects as a unit. Empty slice → `Ok([])`. | +| Nesting | **Not supported.** No SAVEPOINTs. Each call uses a fresh adapter; put every statement in one slice. | +| Retry | **dactyl does not retry.** Callers own retry policy. | +| Timeout | **No public deadline.** Neon uses reqwest defaults; SQLite is local. | +| Idempotency | **Not idempotent.** Replays may conflict or double-write. Design deterministic keys / upserts if retrying after ambiguous transport failures. | +| Proof | Conformance covers SQLite + Neon-mock failure injection and an event-plus-state fixture (state row + event row in one batch; mid-batch failure leaves neither). | + ## License This project is licensed under the MIT License - see the [LICENSE](LICENSE) file for details. diff --git a/src/adapter/neon/mod.rs b/src/adapter/neon/mod.rs index e9f8c9d..74f9afc 100644 --- a/src/adapter/neon/mod.rs +++ b/src/adapter/neon/mod.rs @@ -19,6 +19,11 @@ //! } //! ``` //! +//! `/batch` is the Neon half of [`crate::transaction`]: the server must apply +//! the statement list as one atomic unit (all commit or all reject). Non-2xx +//! responses are surfaced as [`DactylError::Adapter`] with the response body; +//! dactyl does not partially apply a failed batch client-side. +//! //! Propodus owns auth; dactyl only forwards the opaque `bearer` token. use serde::{Deserialize, Serialize}; @@ -151,15 +156,19 @@ impl Adapter for NeonAdapter { .send() .map_err(|e| DactylError::Adapter(format!("neon batch send: {e}")))?; let status = resp.status(); - let body: BatchResponse = resp - .json() - .map_err(|e| DactylError::Adapter(format!("neon batch decode: {e}")))?; + // Read bytes first so non-2xx error bodies (often not BatchResponse) + // still surface as Adapter errors with the server payload. + let bytes = resp + .bytes() + .map_err(|e| DactylError::Adapter(format!("neon batch body: {e}")))?; if !status.is_success() { return Err(DactylError::Adapter(format!( "neon batch status {status}: {}", - serde_json::to_string(&body).unwrap_or_default() + String::from_utf8_lossy(&bytes) ))); } + let body: BatchResponse = serde_json::from_slice(&bytes) + .map_err(|e| DactylError::Adapter(format!("neon batch decode: {e}")))?; let mut results = Vec::with_capacity(body.results.len()); for res in body.results { results.push(rows_from_response(res)?); diff --git a/src/lib.rs b/src/lib.rs index 4ba707a..025a14d 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -121,9 +121,55 @@ pub fn execute(sql: &str, params: &[Parameter]) -> Result { /// Execute an atomic batch of parameterized statements. /// +/// # Atomicity (dactyl #24) +/// /// On any per-statement error the whole unit rolls back and the function -/// returns the error; no partial state is committed. Equivalent semantics -/// are provided for SQLite (transaction) and Neon (`/batch` endpoint). +/// returns [`DactylError`]; **no partial state is committed**. Semantics: +/// +/// | Backend | Mechanism | +/// |---|---| +/// | SQLite | Single rusqlite transaction: begin → statements → commit, or drop = rollback | +/// | Neon | One `POST {endpoint}/batch` request; the server must accept/reject the batch as a unit | +/// +/// An empty `statements` slice is a successful no-op and returns `Ok(vec![])`. +/// +/// # Nesting +/// +/// **Not supported.** Each call builds a fresh short-lived adapter. There is no +/// SAVEPOINT API and no nesting of `transaction` inside another open unit. +/// Independent concurrent `transaction` calls are separate atomic units, not +/// nested subtransactions. Callers that need multi-step atomicity must put +/// every statement in a **single** `transaction(&[...])` slice. +/// +/// # Retry +/// +/// **dactyl does not retry.** A failed batch leaves no committed partial state +/// on either adapter (when the Neon transport returns a definitive error). +/// Callers own retry policy. After a **transport timeout or dropped connection**, +/// the client cannot distinguish “never applied” from “applied but response +/// lost”; retries must use **idempotent** statement design (deterministic keys, +/// upserts) if re-execution is possible. +/// +/// # Timeout +/// +/// **No public deadline parameter.** SQLite is process-local. Neon uses the +/// reqwest client’s default timeouts. Callers that need tighter bounds should +/// enforce them outside dactyl (process supervisor, HTTP proxy, or a future +/// env-based client config — not part of this surface). +/// +/// # Idempotency +/// +/// `transaction` itself is **not** idempotent. Replaying a previously +/// successful batch may insert duplicates or hit primary-key conflicts. +/// Design statements for safe replay when the caller’s retry policy may +/// re-submit after ambiguous failures. +/// +/// # Returns +/// +/// On success, one [`Rows`] per input statement (writes often yield empty +/// row sets; `SELECT` statements yield projections). On failure, an +/// [`DactylError::Adapter`] (or conversion error while decoding Neon rows) +/// and no committed partial state. pub fn transaction(statements: &[Statement]) -> Result, DactylError> { if statements.is_empty() { return Ok(Vec::new()); diff --git a/tests/conformance.rs b/tests/conformance.rs index f189a96..1635085 100644 --- a/tests/conformance.rs +++ b/tests/conformance.rs @@ -8,7 +8,8 @@ //! Covers dactyl issues: //! - #2 every store × every adapter × parameterized reads/writes //! - #23 parameter binding, NULL/bool/int/real/text + injection attempt -//! - #24 atomic transaction + rollback-on-failure +//! - #24 atomic transaction + rollback-on-failure (SQLite + Neon mock), +//! event-plus-state fixture, nesting/retry/timeout/idempotency contract //! - #25 typed named extraction, NULL, missing column, conversion error, //! duplicate aliases, full scalar matrix through both adapters, and //! borrowed-vs-owned accessors (also DecapodLabs/decapod#1111) @@ -21,6 +22,8 @@ use std::collections::HashMap; use std::net::SocketAddr; use std::sync::Arc; +use axum::http::StatusCode; +use axum::response::{IntoResponse, Response}; use axum::{extract::State, routing::post, Json, Router}; use serde::{Deserialize, Serialize}; use tempfile::TempDir; @@ -80,53 +83,256 @@ struct MockTable { } /// Mock neon-server state, shared across all tests. +/// +/// `/batch` is **atomic**: statements apply against a snapshot; any error +/// restores the snapshot so no partial writes remain (mirrors Propodus +/// all-or-nothing batch contract for dactyl #24). #[derive(Default)] struct MockState { tables: Mutex>, } impl MockState { - async fn handle( - State(state): State>, - Json(req): Json, - ) -> Json { - let tables = state.tables.lock().await; - let table = table_of(&req.sql); - let data = tables.get(&table).cloned().unwrap_or_default(); - let columns = if data.columns.is_empty() { - vec!["id".into(), "title".into(), "status".into()] - } else { - data.columns - }; - Json(MockResponse { - columns, - rows: data.rows, - }) + async fn handle(State(state): State>, Json(req): Json) -> Response { + let mut tables = state.tables.lock().await; + match apply_statement(&mut tables, &req.sql, req.params.as_ref()) { + Ok(resp) => Json(resp).into_response(), + Err(msg) => ( + StatusCode::BAD_REQUEST, + Json(serde_json::json!({ "error": msg })), + ) + .into_response(), + } } async fn handle_batch( State(state): State>, Json(req): Json, - ) -> Json { - let tables = state.tables.lock().await; + ) -> Response { + let mut tables = state.tables.lock().await; + // Snapshot for full rollback on any per-statement failure. + let snapshot = tables.clone(); let mut results = Vec::new(); for stmt in &req.statements { - let table = table_of(&stmt.sql); - let data = tables.get(&table).cloned().unwrap_or_default(); - let columns = if data.columns.is_empty() { - vec!["id".into(), "title".into(), "status".into()] + match apply_statement(&mut tables, &stmt.sql, stmt.params.as_ref()) { + Ok(resp) => results.push(resp), + Err(msg) => { + *tables = snapshot; + return ( + StatusCode::CONFLICT, + Json(serde_json::json!({ + "error": format!("batch aborted: {msg}"), + })), + ) + .into_response(); + } + } + } + (StatusCode::OK, Json(MockBatchResponse { results })).into_response() + } +} + +/// Apply one SQL statement against the mock table map. +/// +/// Supports a small dialect used by conformance tests: `CREATE TABLE`, +/// `INSERT INTO … VALUES ($n,…)`, and `SELECT` (including `count(*)`). +fn apply_statement( + tables: &mut HashMap, + sql: &str, + params: Option<&serde_json::Value>, +) -> Result { + let lower = sql.to_ascii_lowercase(); + let trimmed = lower.trim(); + if trimmed.starts_with("create table") { + return apply_create(tables, &lower); + } + if trimmed.starts_with("insert into") { + return apply_insert(tables, sql, params); + } + if trimmed.starts_with("select") { + return apply_select(tables, &lower); + } + // Fallback: treat as a read of `FROM ` if present. + apply_select(tables, &lower) +} + +fn apply_create( + tables: &mut HashMap, + lower_sql: &str, +) -> Result { + // create table [if not exists] name ( + let after = lower_sql + .split_once("table") + .map(|(_, rest)| rest.trim()) + .ok_or_else(|| "create table: missing name".to_string())?; + let after = after + .strip_prefix("if not exists") + .map(str::trim) + .unwrap_or(after); + let name: String = after + .chars() + .take_while(|c| c.is_ascii_alphanumeric() || *c == '_') + .collect(); + if name.is_empty() { + return Err("create table: empty name".into()); + } + let cols = parse_create_columns(lower_sql); + tables.entry(name).or_insert_with(|| MockTable { + columns: cols, + rows: Vec::new(), + }); + Ok(MockResponse { + columns: vec![], + rows: vec![], + }) +} + +fn parse_create_columns(lower_sql: &str) -> Vec { + let Some(start) = lower_sql.find('(') else { + return vec![]; + }; + let Some(end) = lower_sql.rfind(')') else { + return vec![]; + }; + if end <= start { + return vec![]; + } + lower_sql[start + 1..end] + .split(',') + .filter_map(|part| { + let col = part + .split_whitespace() + .next() + .unwrap_or("") + .trim_matches('"'); + if col.is_empty() || col == "primary" { + None } else { - data.columns - }; - results.push(MockResponse { - columns, - rows: data.rows, - }); + Some(col.to_string()) + } + }) + .collect() +} + +fn apply_insert( + tables: &mut HashMap, + sql: &str, + params: Option<&serde_json::Value>, +) -> Result { + let lower = sql.to_ascii_lowercase(); + let after = lower + .split_once("insert into") + .map(|(_, r)| r.trim()) + .ok_or_else(|| "insert: parse failed".to_string())?; + let table: String = after + .chars() + .take_while(|c| c.is_ascii_alphanumeric() || *c == '_') + .collect(); + if table.is_empty() { + return Err("insert: empty table".into()); + } + let cols = parse_insert_columns(sql).unwrap_or_else(|| { + tables + .get(&table) + .map(|t| t.columns.clone()) + .unwrap_or_default() + }); + let p = params_list(params); + if cols.is_empty() { + return Err(format!("insert into {table}: no columns")); + } + if p.len() < cols.len() { + return Err(format!( + "insert into {table}: expected {} params, got {}", + cols.len(), + p.len() + )); + } + let mut obj = serde_json::Map::new(); + for (i, c) in cols.iter().enumerate() { + obj.insert(c.clone(), p[i].clone()); + } + let entry = tables.entry(table.clone()).or_insert_with(|| MockTable { + columns: cols.clone(), + rows: Vec::new(), + }); + if entry.columns.is_empty() { + entry.columns = cols.clone(); + } + // Primary-key style uniqueness on `id` when present. + if let Some(id) = obj.get("id") { + if entry.rows.iter().any(|r| r.get("id") == Some(id)) { + return Err(format!("duplicate key id={id} in {table}")); } - Json(MockBatchResponse { results }) + } + entry.rows.push(serde_json::Value::Object(obj)); + Ok(MockResponse { + columns: entry.columns.clone(), + rows: vec![], + }) +} + +fn parse_insert_columns(sql: &str) -> Option> { + let lower = sql.to_ascii_lowercase(); + let after_table = lower.split_once("insert into")?.1.trim(); + let rest = after_table.find('(').map(|i| &after_table[i..])?; + let end = rest.find(')')?; + let inner = &rest[1..end]; + // Only the column list before VALUES. + if !lower.contains("values") { + return None; + } + let cols: Vec = inner + .split(',') + .map(|c| c.trim().trim_matches('"').to_string()) + .filter(|c| !c.is_empty()) + .collect(); + if cols.is_empty() { + None + } else { + Some(cols) + } +} + +fn params_list(params: Option<&serde_json::Value>) -> Vec { + match params { + Some(serde_json::Value::Array(a)) => a.clone(), + Some(other) => vec![other.clone()], + None => vec![], } } +fn apply_select( + tables: &mut HashMap, + lower_sql: &str, +) -> Result { + let table = table_of(lower_sql); + let data = tables.get(&table).cloned().unwrap_or_default(); + let columns = if data.columns.is_empty() { + // Legacy default for seeded store fixtures. + vec!["id".into(), "title".into(), "status".into()] + } else { + data.columns.clone() + }; + + if lower_sql.contains("count(*)") { + let alias = if lower_sql.contains(" as cnt") { + "cnt" + } else { + "count(*)" + }; + return Ok(MockResponse { + columns: vec![alias.into()], + rows: vec![serde_json::json!({ alias: data.rows.len() as i64 })], + }); + } + + Ok(MockResponse { + columns, + rows: data.rows, + }) +} + fn table_of(sql: &str) -> String { sql.to_ascii_lowercase() .split_whitespace() @@ -144,7 +350,6 @@ fn table_of(sql: &str) -> String { struct MockRequest { sql: String, #[serde(default)] - #[allow(dead_code)] params: Option, } @@ -163,7 +368,6 @@ struct MockBatchRequest { struct MockStatement { sql: String, #[serde(default)] - #[allow(dead_code)] params: Option, } @@ -188,6 +392,36 @@ async fn spawn_mock(state: Arc) -> SocketAddr { addr } +/// Helper: run a closure against a live mock Neon endpoint, then shut it down. +fn with_neon_mock(seed: HashMap, f: F) +where + F: FnOnce(&str), +{ + let state = Arc::new(MockState { + tables: Mutex::new(seed), + }); + let (done_tx, done_rx) = tokio::sync::oneshot::channel::<()>(); + let (ready_tx, ready_rx) = std::sync::mpsc::channel::(); + let mock_thread = std::thread::spawn({ + let state = state.clone(); + move || { + let rt = tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .expect("mock rt"); + rt.block_on(async { + let addr = spawn_mock(state).await; + let _ = ready_tx.send(format!("http://{addr}")); + let _ = done_rx.await; + }); + } + }); + let endpoint = ready_rx.recv().expect("ready"); + f(&endpoint); + let _ = done_tx.send(()); + let _ = mock_thread.join(); +} + // --------------------------------------------------------------------------- // Fixtures // --------------------------------------------------------------------------- @@ -807,7 +1041,7 @@ fn typed_projection_cross_adapter_parity() { } /// #24: atomic transaction batch commits on success and rolls back fully on -/// any per-statement failure. No partial state remains. +/// any per-statement failure (SQLite). No partial state remains. #[test] fn atomic_transaction_rollback_on_failure() { let _guard = lock_env(); @@ -855,6 +1089,254 @@ fn atomic_transaction_rollback_on_failure() { "row id=3 must NOT have been committed: full rollback" ); + // Empty batch is a successful no-op. + let empty = dactyl_db::transaction(&[]).expect("empty batch"); + assert!(empty.is_empty()); + + clear_env(); +} + +/// #24: Neon/mock failure-injection — a multi-statement `/batch` that fails +/// mid-way leaves no partial state (mirrors SQLite rollback proof). +#[test] +fn atomic_transaction_rollback_neon_mock() { + let _guard = lock_env(); + + let mut seed = HashMap::new(); + seed.insert( + "tx".into(), + MockTable { + columns: vec!["id".into(), "value".into()], + rows: vec![ + serde_json::json!({"id": 1, "value": "val1"}), + serde_json::json!({"id": 2, "value": "val2"}), + ], + }, + ); + + with_neon_mock(seed, |endpoint| { + select_neon(endpoint); + + // Successful batch appends id=3. + dactyl_db::transaction(&[Statement::new( + "insert into tx (id, value) values ($1, $2)", + vec![Parameter::Integer(3), Parameter::Text("val3".into())], + )]) + .expect("neon batch success"); + + let rows = dactyl_db::query("select count(*) as cnt from tx", &[]).expect("count"); + let cnt: i64 = rows.as_slice()[0].get("cnt").expect("cnt"); + assert_eq!(cnt, 3, "successful batch must persist"); + + // Failing batch: first insert would add id=4, second hits duplicate id=1. + // The whole unit must abort — id=4 must not remain. + let res = dactyl_db::transaction(&[ + Statement::new( + "insert into tx (id, value) values ($1, $2)", + vec![Parameter::Integer(4), Parameter::Text("val4".into())], + ), + Statement::new( + "insert into tx (id, value) values ($1, $2)", + vec![Parameter::Integer(1), Parameter::Text("duplicate".into())], + ), + ]); + assert!( + matches!(res, Err(DactylError::Adapter(_))), + "neon batch failure must be Adapter error: {res:?}" + ); + + let after = dactyl_db::query("select count(*) as cnt from tx", &[]).expect("count after"); + let cnt_after: i64 = after.as_slice()[0].get("cnt").expect("cnt"); + assert_eq!( + cnt_after, 3, + "row id=4 must NOT have been committed on neon mock rollback" + ); + }); + + clear_env(); +} + +/// #24: event-plus-state atomicity fixture (SQLite). +/// +/// A domain-shaped unit of work updates durable state and appends an event +/// in one `transaction`. Mid-batch failure leaves neither side committed. +#[test] +fn atomic_event_plus_state_sqlite() { + let _guard = lock_env(); + let tmp = TempDir::new().expect("tempdir"); + let path = tmp.path().join("event_state.db"); + select_sqlite(path.to_str().unwrap()); + + dactyl_db::execute( + "create table state (id integer primary key, value text not null)", + &[], + ) + .expect("create state"); + dactyl_db::execute( + "create table events ( + id integer primary key, + kind text not null, + payload text not null + )", + &[], + ) + .expect("create events"); + + // Success path: state row + event row commit together. + dactyl_db::transaction(&[ + Statement::new( + "insert into state (id, value) values ($1, $2)", + vec![Parameter::Integer(1), Parameter::Text("ready".into())], + ), + Statement::new( + "insert into events (id, kind, payload) values ($1, $2, $3)", + vec![ + Parameter::Integer(1), + Parameter::Text("state.changed".into()), + Parameter::Text(r#"{"id":1,"value":"ready"}"#.into()), + ], + ), + ]) + .expect("event+state success"); + + let state_cnt: i64 = dactyl_db::query("select count(*) as cnt from state", &[]) + .unwrap() + .as_slice()[0] + .get("cnt") + .unwrap(); + let event_cnt: i64 = dactyl_db::query("select count(*) as cnt from events", &[]) + .unwrap() + .as_slice()[0] + .get("cnt") + .unwrap(); + assert_eq!(state_cnt, 1); + assert_eq!(event_cnt, 1); + + // Failure path: state insert would succeed, event hits duplicate PK → both roll back. + let res = dactyl_db::transaction(&[ + Statement::new( + "insert into state (id, value) values ($1, $2)", + vec![Parameter::Integer(2), Parameter::Text("partial".into())], + ), + Statement::new( + "insert into events (id, kind, payload) values ($1, $2, $3)", + vec![ + Parameter::Integer(1), // duplicate event id + Parameter::Text("state.changed".into()), + Parameter::Text(r#"{"id":2}"#.into()), + ], + ), + ]); + assert!(res.is_err(), "duplicate event key must fail batch: {res:?}"); + + let state_after: i64 = dactyl_db::query("select count(*) as cnt from state", &[]) + .unwrap() + .as_slice()[0] + .get("cnt") + .unwrap(); + let event_after: i64 = dactyl_db::query("select count(*) as cnt from events", &[]) + .unwrap() + .as_slice()[0] + .get("cnt") + .unwrap(); + assert_eq!( + state_after, 1, + "state id=2 must not commit when event side fails" + ); + assert_eq!( + event_after, 1, + "events must stay at the successful batch only" + ); + + clear_env(); +} + +/// #24: event-plus-state atomicity fixture (Neon mock). +#[test] +fn atomic_event_plus_state_neon_mock() { + let _guard = lock_env(); + + let mut seed = HashMap::new(); + seed.insert( + "state".into(), + MockTable { + columns: vec!["id".into(), "value".into()], + rows: vec![], + }, + ); + seed.insert( + "events".into(), + MockTable { + columns: vec!["id".into(), "kind".into(), "payload".into()], + rows: vec![], + }, + ); + + with_neon_mock(seed, |endpoint| { + select_neon(endpoint); + + dactyl_db::transaction(&[ + Statement::new( + "insert into state (id, value) values ($1, $2)", + vec![Parameter::Integer(1), Parameter::Text("ready".into())], + ), + Statement::new( + "insert into events (id, kind, payload) values ($1, $2, $3)", + vec![ + Parameter::Integer(1), + Parameter::Text("state.changed".into()), + Parameter::Text(r#"{"id":1}"#.into()), + ], + ), + ]) + .expect("neon event+state success"); + + let state_cnt: i64 = dactyl_db::query("select count(*) as cnt from state", &[]) + .unwrap() + .as_slice()[0] + .get("cnt") + .unwrap(); + let event_cnt: i64 = dactyl_db::query("select count(*) as cnt from events", &[]) + .unwrap() + .as_slice()[0] + .get("cnt") + .unwrap(); + assert_eq!(state_cnt, 1); + assert_eq!(event_cnt, 1); + + let res = dactyl_db::transaction(&[ + Statement::new( + "insert into state (id, value) values ($1, $2)", + vec![Parameter::Integer(2), Parameter::Text("partial".into())], + ), + Statement::new( + "insert into events (id, kind, payload) values ($1, $2, $3)", + vec![ + Parameter::Integer(1), + Parameter::Text("state.changed".into()), + Parameter::Text(r#"{"id":2}"#.into()), + ], + ), + ]); + assert!( + matches!(res, Err(DactylError::Adapter(_))), + "neon event+state failure: {res:?}" + ); + + let state_after: i64 = dactyl_db::query("select count(*) as cnt from state", &[]) + .unwrap() + .as_slice()[0] + .get("cnt") + .unwrap(); + let event_after: i64 = dactyl_db::query("select count(*) as cnt from events", &[]) + .unwrap() + .as_slice()[0] + .get("cnt") + .unwrap(); + assert_eq!(state_after, 1, "no partial state on neon mock"); + assert_eq!(event_after, 1, "no partial events on neon mock"); + }); + clear_env(); }