From 01e4e5a8f51fac0c1198e790e81cd3dfc6e62105 Mon Sep 17 00:00:00 2001 From: MechaCat02 Date: Tue, 14 Jul 2026 19:28:17 +0200 Subject: [PATCH] fix(kv): commit a write and its trigger fan-out in one transaction MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Audit #6. `KvServiceImpl` wrote the row, waited for it to commit, then asked the emitter to resolve matching triggers and insert outbox rows — a second transaction on a second connection. If that second step failed, the row was permanently in the store with its trigger having never fired: invisible to the caller (the write "succeeded"), unrecoverable by any retry, and only ever logged. The code said as much: // Audit finding (Medium): this is non-transactional with the data // write — the row above has already committed by the time emit() // runs, so an emit failure means triggers silently don't fire. Introduce `atomic_write::KvWriter` — the mutating half of the service (the write AND the fan-out it produces), so the two can share a transaction: * `PostgresKvWriter` opens a tx, writes via `kv_repo::*_on(&mut *tx, …)`, runs the fan-out on the SAME connection via `emit_on(&mut *tx, …)`, and commits. An emit failure now rolls the write back and surfaces to the script, which is the honest outcome — the caller learns the write did not happen instead of silently getting a store that disagrees with its triggers. * `BestEffortKvWriter` keeps the old write-then-log-on-emit-failure semantics for the in-memory unit tests (no Postgres). Both sit behind one trait, so the service body has a single code path and the choice is a constructor detail (`with_atomic_writes(pool)` in the host). Pool-deadlock rule, documented on the module: everything inside the tx runs on the tx's connection. A writer that held a tx and then reached for a second pooled connection could starve — the pool is sized to the execution concurrency cap, so N executions each wanting 2 connections deadlock. The fan-out takes `&mut *tx` and never touches a repo. `tests/atomic_write.rs` pins it by injecting an outbox failure (a Postgres BEFORE-INSERT trigger scoped to the test's own app_id, so parallel tests are unaffected): the set errors and the key is NOT in the store; likewise a delete rolls back rather than dropping a key nothing downstream hears about. Co-Authored-By: Claude Opus 4.8 (1M context) --- crates/manager-core/src/atomic_write.rs | 297 ++++++++++++++++++++++ crates/manager-core/src/kv_repo.rs | 179 ++++++++----- crates/manager-core/src/kv_service.rs | 157 +++--------- crates/manager-core/src/lib.rs | 1 + crates/manager-core/tests/atomic_write.rs | 271 ++++++++++++++++++++ crates/picloud/src/lib.rs | 18 +- 6 files changed, 736 insertions(+), 187 deletions(-) create mode 100644 crates/manager-core/src/atomic_write.rs create mode 100644 crates/manager-core/tests/atomic_write.rs diff --git a/crates/manager-core/src/atomic_write.rs b/crates/manager-core/src/atomic_write.rs new file mode 100644 index 0000000..17612cd --- /dev/null +++ b/crates/manager-core/src/atomic_write.rs @@ -0,0 +1,297 @@ +//! The atomic data-plane write: a row mutation **and** the trigger fan-out it +//! produces, committed together or not at all. +//! +//! ## Why +//! +//! Every stateful service used to write the row, wait for it to commit, and +//! *then* ask the emitter to resolve matching triggers and insert outbox rows. +//! Those are separate transactions on separate connections, so the window +//! between them is a durability hole: if the outbox insert failed — or the +//! process died — the row was permanently in the store with its trigger having +//! never fired, and no retry could ever notice. The services could only log the +//! failure (`event_emit_failure = true`) and carry on. +//! +//! A `Writer` closes that hole. It takes a transaction, does the write, runs the +//! fan-out on the **same connection** (`outbox_event_emitter::emit_on(&mut *tx, +//! …)`), and commits. An emit failure now rolls the write back and surfaces as +//! an error to the script, which is the honest outcome: the caller learns the +//! write did not happen, rather than silently getting a store that disagrees +//! with its triggers. +//! +//! ## The pool-deadlock rule +//! +//! Everything inside a transaction MUST run on that transaction's connection. +//! A writer that held a `tx` and then reached for a *second* pooled connection +//! (e.g. by calling a repo method) could deadlock: the pool is sized to the +//! execution concurrency cap, so N executions each wanting 2 connections can +//! starve each other forever. That is why the fan-out and the quota reads below +//! all take `&mut *tx` and never touch a repo. +//! +//! ## Two implementations +//! +//! * `Postgres*Writer` — transactional, used by the host. +//! * `BestEffort*Writer` — writes through the repo trait, then emits +//! best-effort. It is what the in-memory unit tests run against (no Postgres), +//! and it preserves the pre-existing semantics exactly. +//! +//! Both sit behind the same trait, so the service body has a single code path; +//! which one is in play is a constructor choice, not a branch in the hot path. + +use std::sync::Arc; + +use async_trait::async_trait; +use picloud_shared::{KvError, SdkCallCx, ServiceEvent, ServiceEventEmitter}; +use serde_json::Value; +use sqlx::PgPool; + +use crate::kv_repo::{self, KvRepo}; +use crate::outbox_event_emitter::emit_on; + +/// Map a `sqlx` failure from the transaction plumbing itself (begin/commit). +fn tx_err(e: &sqlx::Error) -> KvError { + KvError::Backend(format!("database error: {e}")) +} + +/// A KV mutation's `ServiceEvent`. Shared by both writers so the event shape has +/// one home. +fn kv_event( + op: &'static str, + collection: &str, + key: &str, + payload: Option, + old_payload: Option, +) -> ServiceEvent { + ServiceEvent { + source: "kv", + op, + collection: Some(collection.to_string()), + key: Some(key.to_string()), + payload, + old_payload, + } +} + +/// The mutating half of the KV service: the row write **and** the trigger +/// fan-out that follows from it. Reads stay on `KvRepo`. +#[async_trait] +pub trait KvWriter: Send + Sync { + /// Upsert + fan out. Returns the previous value — `None` means this was an + /// insert rather than an update. + async fn set( + &self, + cx: &SdkCallCx, + collection: &str, + key: &str, + value: Value, + ) -> Result, KvError>; + + /// Compare-and-swap + fan out (only on a successful swap). Returns whether + /// the swap happened. + async fn set_if( + &self, + cx: &SdkCallCx, + collection: &str, + key: &str, + expected: Option, + new: Value, + ) -> Result; + + /// Delete + fan out (only when the key was present). Returns the removed + /// value. + async fn delete( + &self, + cx: &SdkCallCx, + collection: &str, + key: &str, + ) -> Result, KvError>; +} + +// ---------------------------------------------------------------------------- +// Transactional (Postgres) +// ---------------------------------------------------------------------------- + +pub struct PostgresKvWriter { + pool: PgPool, +} + +impl PostgresKvWriter { + #[must_use] + pub fn new(pool: PgPool) -> Self { + Self { pool } + } +} + +#[async_trait] +impl KvWriter for PostgresKvWriter { + async fn set( + &self, + cx: &SdkCallCx, + collection: &str, + key: &str, + value: Value, + ) -> Result, KvError> { + let mut tx = self.pool.begin().await.map_err(|e| tx_err(&e))?; + let previous = kv_repo::set_on(&mut *tx, cx.app_id, collection, key, value.clone()).await?; + let op = if previous.is_some() { + "update" + } else { + "insert" + }; + let event = kv_event(op, collection, key, Some(value), previous.clone()); + // Same connection as the write above — the outbox rows commit with it. + emit_on(&mut tx, cx, &event) + .await + .map_err(|e| KvError::Backend(format!("event emit: {e}")))?; + tx.commit().await.map_err(|e| tx_err(&e))?; + Ok(previous) + } + + async fn set_if( + &self, + cx: &SdkCallCx, + collection: &str, + key: &str, + expected: Option, + new: Value, + ) -> Result { + // `insert` when the precondition was "absent", else `update`. + let op = if expected.is_some() { + "update" + } else { + "insert" + }; + let mut tx = self.pool.begin().await.map_err(|e| tx_err(&e))?; + let swapped = + kv_repo::set_if_on(&mut *tx, cx.app_id, collection, key, expected, new.clone()).await?; + if swapped { + let event = kv_event(op, collection, key, Some(new), None); + emit_on(&mut tx, cx, &event) + .await + .map_err(|e| KvError::Backend(format!("event emit: {e}")))?; + } + tx.commit().await.map_err(|e| tx_err(&e))?; + Ok(swapped) + } + + async fn delete( + &self, + cx: &SdkCallCx, + collection: &str, + key: &str, + ) -> Result, KvError> { + let mut tx = self.pool.begin().await.map_err(|e| tx_err(&e))?; + let previous = kv_repo::delete_on(&mut *tx, cx.app_id, collection, key).await?; + if previous.is_some() { + let event = kv_event("delete", collection, key, None, previous.clone()); + emit_on(&mut tx, cx, &event) + .await + .map_err(|e| KvError::Backend(format!("event emit: {e}")))?; + } + tx.commit().await.map_err(|e| tx_err(&e))?; + Ok(previous) + } +} + +// ---------------------------------------------------------------------------- +// Best-effort (repo trait + emitter trait) +// ---------------------------------------------------------------------------- + +/// Emit, downgrading a failure to an error-level log. The write has already +/// committed by this point, so there is nothing useful to return to the caller — +/// which is exactly the durability hole `PostgresKvWriter` exists to close. +/// `event_emit_failure = true` is there to be grepped/alerted on. +async fn best_effort_emit(events: &dyn ServiceEventEmitter, cx: &SdkCallCx, event: ServiceEvent) { + let (source, op) = (event.source, event.op); + if let Err(e) = events.emit(cx, event).await { + tracing::error!( + error = %e, source, op, event_emit_failure = true, + "event emit failed — the write committed but its triggers will not fire" + ); + } +} + +pub struct BestEffortKvWriter { + repo: Arc, + events: Arc, +} + +impl BestEffortKvWriter { + #[must_use] + pub fn new(repo: Arc, events: Arc) -> Self { + Self { repo, events } + } +} + +#[async_trait] +impl KvWriter for BestEffortKvWriter { + async fn set( + &self, + cx: &SdkCallCx, + collection: &str, + key: &str, + value: Value, + ) -> Result, KvError> { + let previous = self + .repo + .set(cx.app_id, collection, key, value.clone()) + .await?; + let op = if previous.is_some() { + "update" + } else { + "insert" + }; + best_effort_emit( + &*self.events, + cx, + kv_event(op, collection, key, Some(value), previous.clone()), + ) + .await; + Ok(previous) + } + + async fn set_if( + &self, + cx: &SdkCallCx, + collection: &str, + key: &str, + expected: Option, + new: Value, + ) -> Result { + let op = if expected.is_some() { + "update" + } else { + "insert" + }; + let swapped = self + .repo + .set_if(cx.app_id, collection, key, expected, new.clone()) + .await?; + if swapped { + best_effort_emit( + &*self.events, + cx, + kv_event(op, collection, key, Some(new), None), + ) + .await; + } + Ok(swapped) + } + + async fn delete( + &self, + cx: &SdkCallCx, + collection: &str, + key: &str, + ) -> Result, KvError> { + let previous = self.repo.delete(cx.app_id, collection, key).await?; + if previous.is_some() { + best_effort_emit( + &*self.events, + cx, + kv_event("delete", collection, key, None, previous.clone()), + ) + .await; + } + Ok(previous) + } +} diff --git a/crates/manager-core/src/kv_repo.rs b/crates/manager-core/src/kv_repo.rs index c7a1a20..a4f0f64 100644 --- a/crates/manager-core/src/kv_repo.rs +++ b/crates/manager-core/src/kv_repo.rs @@ -91,6 +91,118 @@ impl PostgresKvRepo { const KV_LIST_MAX_LIMIT: u32 = 1_000; const KV_LIST_DEFAULT_LIMIT: u32 = 100; +// ---------------------------------------------------------------------------- +// Connection-scoped mutations +// +// Generic over the executor so the same SQL serves the pooled `KvRepo` methods +// above and `crate::atomic_write`, which runs the write and the resulting +// trigger fan-out on ONE connection inside ONE transaction. +// ---------------------------------------------------------------------------- + +/// Upsert. Returns the previous value, so the caller knows whether this was an +/// `insert` or an `update` for the emitted `ServiceEvent`. +pub(crate) async fn set_on<'c, E>( + exec: E, + app_id: AppId, + collection: &str, + key: &str, + value: serde_json::Value, +) -> Result, KvRepoError> +where + E: sqlx::PgExecutor<'c>, +{ + // `RETURNING` after `ON CONFLICT DO UPDATE` can't see the old value, so + // capture the prior value in a CTE alongside the upsert. + let row: Option<(Option,)> = sqlx::query_as( + "WITH prev AS (\ + SELECT value FROM kv_entries \ + WHERE app_id = $1 AND collection = $2 AND key = $3\ + ), \ + upserted AS (\ + INSERT INTO kv_entries (app_id, collection, key, value) \ + VALUES ($1, $2, $3, $4) \ + ON CONFLICT (app_id, collection, key) DO UPDATE \ + SET value = EXCLUDED.value, updated_at = NOW() \ + RETURNING 1\ + ) \ + SELECT (SELECT value FROM prev) FROM upserted", + ) + .bind(app_id.into_inner()) + .bind(collection) + .bind(key) + .bind(value) + .fetch_optional(exec) + .await?; + Ok(row.and_then(|(v,)| v)) +} + +/// Compare-and-swap. Writes `new` iff the current value equals `expected` +/// (`expected = None` → iff the key is ABSENT). Returns whether it wrote. +pub(crate) async fn set_if_on<'c, E>( + exec: E, + app_id: AppId, + collection: &str, + key: &str, + expected: Option, + new: serde_json::Value, +) -> Result +where + E: sqlx::PgExecutor<'c>, +{ + let affected = match expected { + // Swap iff the current value matches (JSONB `=` is semantic equality). + Some(exp) => sqlx::query( + "UPDATE kv_entries SET value = $4, updated_at = NOW() \ + WHERE app_id = $1 AND collection = $2 AND key = $3 AND value = $5", + ) + .bind(app_id.into_inner()) + .bind(collection) + .bind(key) + .bind(new) + .bind(exp) + .execute(exec) + .await? + .rows_affected(), + // Insert iff absent. + None => sqlx::query( + "INSERT INTO kv_entries (app_id, collection, key, value) \ + VALUES ($1, $2, $3, $4) \ + ON CONFLICT (app_id, collection, key) DO NOTHING", + ) + .bind(app_id.into_inner()) + .bind(collection) + .bind(key) + .bind(new) + .execute(exec) + .await? + .rows_affected(), + }; + Ok(affected == 1) +} + +/// Returns the deleted value if present, `None` if the row didn't exist. +pub(crate) async fn delete_on<'c, E>( + exec: E, + app_id: AppId, + collection: &str, + key: &str, +) -> Result, KvRepoError> +where + E: sqlx::PgExecutor<'c>, +{ + let row: Option<(serde_json::Value,)> = sqlx::query_as( + "DELETE FROM kv_entries \ + WHERE app_id = $1 AND collection = $2 AND key = $3 \ + RETURNING value", + ) + .bind(app_id.into_inner()) + .bind(collection) + .bind(key) + .fetch_optional(exec) + .await?; + Ok(row.map(|(v,)| v)) +} + #[async_trait] impl KvRepo for PostgresKvRepo { async fn get( @@ -118,30 +230,7 @@ impl KvRepo for PostgresKvRepo { key: &str, value: serde_json::Value, ) -> Result, KvRepoError> { - // `RETURNING` after `ON CONFLICT DO UPDATE` exposes the old - // value via the `xmax`/old-row trick: capture the prior value - // with a CTE so callers know whether this was insert vs update. - let row: Option<(Option,)> = sqlx::query_as( - "WITH prev AS (\ - SELECT value FROM kv_entries \ - WHERE app_id = $1 AND collection = $2 AND key = $3\ - ), \ - upserted AS (\ - INSERT INTO kv_entries (app_id, collection, key, value) \ - VALUES ($1, $2, $3, $4) \ - ON CONFLICT (app_id, collection, key) DO UPDATE \ - SET value = EXCLUDED.value, updated_at = NOW() \ - RETURNING 1\ - ) \ - SELECT (SELECT value FROM prev) FROM upserted", - ) - .bind(app_id.into_inner()) - .bind(collection) - .bind(key) - .bind(value) - .fetch_optional(&self.pool) - .await?; - Ok(row.and_then(|(v,)| v)) + set_on(&self.pool, app_id, collection, key, value).await } async fn set_if( @@ -152,35 +241,7 @@ impl KvRepo for PostgresKvRepo { expected: Option, new: serde_json::Value, ) -> Result { - let affected = match expected { - // Swap iff the current value matches (JSONB `=` is semantic equality). - Some(exp) => sqlx::query( - "UPDATE kv_entries SET value = $4, updated_at = NOW() \ - WHERE app_id = $1 AND collection = $2 AND key = $3 AND value = $5", - ) - .bind(app_id.into_inner()) - .bind(collection) - .bind(key) - .bind(new) - .bind(exp) - .execute(&self.pool) - .await? - .rows_affected(), - // Insert iff absent. - None => sqlx::query( - "INSERT INTO kv_entries (app_id, collection, key, value) \ - VALUES ($1, $2, $3, $4) \ - ON CONFLICT (app_id, collection, key) DO NOTHING", - ) - .bind(app_id.into_inner()) - .bind(collection) - .bind(key) - .bind(new) - .execute(&self.pool) - .await? - .rows_affected(), - }; - Ok(affected == 1) + set_if_on(&self.pool, app_id, collection, key, expected, new).await } async fn delete( @@ -189,17 +250,7 @@ impl KvRepo for PostgresKvRepo { collection: &str, key: &str, ) -> Result, KvRepoError> { - let row: Option<(serde_json::Value,)> = sqlx::query_as( - "DELETE FROM kv_entries \ - WHERE app_id = $1 AND collection = $2 AND key = $3 \ - RETURNING value", - ) - .bind(app_id.into_inner()) - .bind(collection) - .bind(key) - .fetch_optional(&self.pool) - .await?; - Ok(row.map(|(v,)| v)) + delete_on(&self.pool, app_id, collection, key).await } async fn has(&self, app_id: AppId, collection: &str, key: &str) -> Result { diff --git a/crates/manager-core/src/kv_service.rs b/crates/manager-core/src/kv_service.rs index 00a2f4f..61caab2 100644 --- a/crates/manager-core/src/kv_service.rs +++ b/crates/manager-core/src/kv_service.rs @@ -12,16 +12,17 @@ //! Cross-app isolation isn't affected — every query is keyed by //! `cx.app_id`, never an argument. //! 3. `ServiceEvent` emission after each mutation (`insert` / `update` -//! / `delete`). v1.1.0 ships a `NoopEventEmitter` so this is a -//! no-op until the outbox emitter lands later in v1.1.1. +//! / `delete`), via the injected `KvWriter` — which for the host is +//! transactional, so a mutation and the trigger fan-out it produces +//! commit together. See `crate::atomic_write`. use std::sync::Arc; use async_trait::async_trait; -use picloud_shared::{ - KvError, KvListPage, KvService, SdkCallCx, ServiceEvent, ServiceEventEmitter, -}; +use picloud_shared::{KvError, KvListPage, KvService, SdkCallCx, ServiceEventEmitter}; +use sqlx::PgPool; +use crate::atomic_write::{BestEffortKvWriter, KvWriter, PostgresKvWriter}; use crate::authz::{self, AuthzRepo, Capability}; use crate::kv_repo::{KvRepo, KvRepoError}; @@ -47,9 +48,11 @@ pub fn kv_max_value_bytes_from_env() -> usize { } pub struct KvServiceImpl { + /// Reads only. Mutations go through `writer`, which owns the write AND the + /// trigger fan-out so the two can share a transaction. repo: Arc, authz: Arc, - events: Arc, + writer: Arc, max_value_bytes: usize, } @@ -71,13 +74,38 @@ impl KvServiceImpl { max_value_bytes: usize, ) -> Self { Self { + writer: Arc::new(BestEffortKvWriter::new(repo.clone(), events)), repo, authz, - events, max_value_bytes, } } + /// Swap the best-effort writer for the transactional one: the row write and + /// its trigger fan-out then commit together, so an outbox failure rolls the + /// write back instead of silently losing the event. The host always calls + /// this; the in-memory unit tests do not (they have no Postgres). + #[must_use] + pub fn with_atomic_writes(mut self, pool: PgPool) -> Self { + self.writer = Arc::new(PostgresKvWriter::new(pool)); + self + } + + /// Encode `value` and enforce the per-key byte cap. Runs before authz so an + /// anonymous public script can't push an oversized payload at Postgres. + fn check_value_size(&self, value: &serde_json::Value) -> Result<(), KvError> { + let encoded_len = serde_json::to_vec(value) + .map(|v| v.len()) + .map_err(|e| KvError::Backend(format!("encode value: {e}")))?; + if encoded_len > self.max_value_bytes { + return Err(KvError::ValueTooLarge { + limit: self.max_value_bytes, + actual: encoded_len, + }); + } + Ok(()) + } + async fn check_read(&self, cx: &SdkCallCx) -> Result<(), KvError> { authz::script_gate( &*self.authz, @@ -135,54 +163,9 @@ impl KvService for KvServiceImpl { value: serde_json::Value, ) -> Result<(), KvError> { validate_collection(collection)?; - let encoded_len = serde_json::to_vec(&value) - .map(|v| v.len()) - .map_err(|e| KvError::Backend(format!("encode value: {e}")))?; - if encoded_len > self.max_value_bytes { - return Err(KvError::ValueTooLarge { - limit: self.max_value_bytes, - actual: encoded_len, - }); - } + self.check_value_size(&value)?; self.check_write(cx).await?; - let previous = self - .repo - .set(cx.app_id, collection, key, value.clone()) - .await?; - let op = if previous.is_some() { - "update" - } else { - "insert" - }; - // Emit unconditionally; the noop emitter drops it, the outbox - // emitter persists it. - // - // Audit finding (Medium): this is non-transactional with the - // data write — the row above has already committed by the time - // emit() runs, so an emit failure means triggers silently - // don't fire. Logged at `error` level (with - // `event_emit_failure = true` for grepability); operators see - // the gap. The full transactional refactor — extending the - // ServiceEventEmitter trait with `emit_in_tx` and the KvRepo - // surface with `set_with_tx` — is deferred to a v1.2 design - // pass (pubsub_repo::fan_out_publish is the reference shape). - if let Err(e) = self - .events - .emit( - cx, - ServiceEvent { - source: "kv", - op, - collection: Some(collection.to_string()), - key: Some(key.to_string()), - payload: Some(value), - old_payload: previous, - }, - ) - .await - { - tracing::error!(error = %e, source = "kv", op, event_emit_failure = true, "event emit failed"); - } + self.writer.set(cx, collection, key, value).await?; Ok(()) } @@ -195,75 +178,15 @@ impl KvService for KvServiceImpl { new: serde_json::Value, ) -> Result { validate_collection(collection)?; - let encoded_len = serde_json::to_vec(&new) - .map(|v| v.len()) - .map_err(|e| KvError::Backend(format!("encode value: {e}")))?; - if encoded_len > self.max_value_bytes { - return Err(KvError::ValueTooLarge { - limit: self.max_value_bytes, - actual: encoded_len, - }); - } + self.check_value_size(&new)?; self.check_write(cx).await?; - // `insert` when the precondition was "absent", else `update`. The op is - // only emitted on a successful swap. - let op = if expected.is_some() { - "update" - } else { - "insert" - }; - let swapped = self - .repo - .set_if(cx.app_id, collection, key, expected, new.clone()) - .await?; - if swapped { - // Non-transactional with the write (same caveat as `set`). - if let Err(e) = self - .events - .emit( - cx, - ServiceEvent { - source: "kv", - op, - collection: Some(collection.to_string()), - key: Some(key.to_string()), - payload: Some(new), - old_payload: None, - }, - ) - .await - { - tracing::error!(error = %e, source = "kv", op, event_emit_failure = true, "event emit failed"); - } - } - Ok(swapped) + self.writer.set_if(cx, collection, key, expected, new).await } async fn delete(&self, cx: &SdkCallCx, collection: &str, key: &str) -> Result { validate_collection(collection)?; self.check_write(cx).await?; - let previous = self.repo.delete(cx.app_id, collection, key).await?; - let was_present = previous.is_some(); - if was_present { - if let Err(e) = self - .events - .emit( - cx, - ServiceEvent { - source: "kv", - op: "delete", - collection: Some(collection.to_string()), - key: Some(key.to_string()), - payload: None, - old_payload: previous, - }, - ) - .await - { - tracing::error!(error = %e, source = "kv", op = "delete", event_emit_failure = true, "event emit failed"); - } - } - Ok(was_present) + Ok(self.writer.delete(cx, collection, key).await?.is_some()) } async fn has(&self, cx: &SdkCallCx, collection: &str, key: &str) -> Result { diff --git a/crates/manager-core/src/lib.rs b/crates/manager-core/src/lib.rs index be52121..c870ddf 100644 --- a/crates/manager-core/src/lib.rs +++ b/crates/manager-core/src/lib.rs @@ -26,6 +26,7 @@ pub mod app_user_verification_repo; pub mod apply_api; pub mod apply_service; pub mod apps_api; +pub mod atomic_write; pub mod auth; pub mod auth_api; pub mod auth_bootstrap; diff --git a/crates/manager-core/tests/atomic_write.rs b/crates/manager-core/tests/atomic_write.rs new file mode 100644 index 0000000..7b34655 --- /dev/null +++ b/crates/manager-core/tests/atomic_write.rs @@ -0,0 +1,271 @@ +//! Audit fix #6: a data write and the trigger fan-out it produces must commit +//! ATOMICALLY. +//! +//! Before this, a service wrote the row (commit #1), then asked the emitter to +//! resolve matching triggers and insert outbox rows (commit #2). If commit #2 +//! failed, the row was permanently in the store with its trigger having never +//! fired — unobservable to the caller, unrecoverable by any retry, and only +//! logged. `PostgresKvWriter` runs both on ONE connection in ONE transaction, so +//! an emit failure rolls the write back and surfaces as an error. +//! +//! Fault injection: a Postgres BEFORE-INSERT trigger on `outbox` that raises, +//! scoped to THIS test's freshly-minted `app_id`, so it cannot affect any test +//! running in parallel against the same database. Skips when `DATABASE_URL` is +//! unset. + +use picloud_manager_core::atomic_write::{KvWriter, PostgresKvWriter}; +use picloud_shared::{AppId, ExecutionId, RequestId, ScriptId, SdkCallCx}; +use sqlx::postgres::PgPoolOptions; +use sqlx::PgPool; +use uuid::Uuid; + +async fn pool_or_skip() -> Option { + let Ok(url) = std::env::var("DATABASE_URL") else { + eprintln!("atomic_write: DATABASE_URL unset — skipping"); + return None; + }; + let pool = PgPoolOptions::new() + .max_connections(3) + .connect(&url) + .await + .expect("connect"); + sqlx::migrate!("./migrations") + .run(&pool) + .await + .expect("migrate"); + Some(pool) +} + +fn cx(app_id: Uuid) -> SdkCallCx { + SdkCallCx { + app_id: AppId::from(app_id), + script_id: ScriptId::new(), + principal: None, + execution_id: ExecutionId::new(), + request_id: RequestId::new(), + trigger_depth: 0, + root_execution_id: ExecutionId::new(), + is_dead_letter_handler: false, + event: None, + } +} + +/// An app with one script and one enabled KV trigger watching `collection`. +struct Fixture { + app: Uuid, +} + +async fn setup(pool: &PgPool, collection: &str) -> Fixture { + let uniq = Uuid::new_v4().simple().to_string(); + let (group,): (Uuid,) = + sqlx::query_as("INSERT INTO groups (slug, name) VALUES ($1, $1) RETURNING id") + .bind(format!("aw-grp-{uniq}")) + .fetch_one(pool) + .await + .expect("group"); + let (app,): (Uuid,) = + sqlx::query_as("INSERT INTO apps (slug, name, group_id) VALUES ($1, $1, $2) RETURNING id") + .bind(format!("aw-app-{uniq}")) + .bind(group) + .fetch_one(pool) + .await + .expect("app"); + let (admin,): (Uuid,) = sqlx::query_as( + "INSERT INTO admin_users (username, password_hash) VALUES ($1, 'x') RETURNING id", + ) + .bind(format!("aw-admin-{uniq}")) + .fetch_one(pool) + .await + .expect("admin"); + let (script,): (Uuid,) = sqlx::query_as( + "INSERT INTO scripts (app_id, name, source) VALUES ($1, 'handler', '') RETURNING id", + ) + .bind(app) + .fetch_one(pool) + .await + .expect("script"); + let (trigger,): (Uuid,) = sqlx::query_as( + "INSERT INTO triggers ( \ + app_id, script_id, kind, retry_max_attempts, retry_backoff, \ + retry_base_ms, registered_by_principal \ + ) VALUES ($1, $2, 'kv', 3, 'exponential', 1000, $3) RETURNING id", + ) + .bind(app) + .bind(script) + .bind(admin) + .fetch_one(pool) + .await + .expect("trigger"); + sqlx::query( + "INSERT INTO kv_trigger_details (trigger_id, collection_glob, ops) \ + VALUES ($1, $2, '{}')", + ) + .bind(trigger) + .bind(collection) + .execute(pool) + .await + .expect("kv_trigger_details"); + Fixture { app } +} + +/// Make every outbox insert for `app` fail. Scoped by app_id so parallel tests +/// (which use their own fresh apps) are untouched. +async fn break_outbox_for(pool: &PgPool, app: Uuid, tag: &str) { + sqlx::query(&format!( + "CREATE FUNCTION break_outbox_{tag}() RETURNS TRIGGER AS $$ \ + BEGIN \ + IF NEW.app_id = '{app}'::uuid THEN \ + RAISE EXCEPTION 'injected outbox failure'; \ + END IF; \ + RETURN NEW; \ + END; $$ LANGUAGE plpgsql" + )) + .execute(pool) + .await + .expect("create fn"); + sqlx::query(&format!( + "CREATE TRIGGER break_outbox_{tag} BEFORE INSERT ON outbox \ + FOR EACH ROW EXECUTE FUNCTION break_outbox_{tag}()" + )) + .execute(pool) + .await + .expect("create trigger"); +} + +async fn unbreak_outbox(pool: &PgPool, tag: &str) { + sqlx::query(&format!( + "DROP TRIGGER IF EXISTS break_outbox_{tag} ON outbox" + )) + .execute(pool) + .await + .expect("drop trigger"); + sqlx::query(&format!("DROP FUNCTION IF EXISTS break_outbox_{tag}()")) + .execute(pool) + .await + .expect("drop fn"); +} + +async fn kv_count(pool: &PgPool, app: Uuid) -> i64 { + let (n,): (i64,) = sqlx::query_as("SELECT COUNT(*) FROM kv_entries WHERE app_id = $1") + .bind(app) + .fetch_one(pool) + .await + .expect("count kv"); + n +} + +async fn outbox_count(pool: &PgPool, app: Uuid) -> i64 { + let (n,): (i64,) = sqlx::query_as("SELECT COUNT(*) FROM outbox WHERE app_id = $1") + .bind(app) + .fetch_one(pool) + .await + .expect("count outbox"); + n +} + +/// `scripts` is RESTRICT on app delete (code is not data), so drop it first. +async fn cleanup(pool: &PgPool, app: Uuid) { + sqlx::query("DELETE FROM scripts WHERE app_id = $1") + .bind(app) + .execute(pool) + .await + .expect("cleanup scripts"); + sqlx::query("DELETE FROM apps WHERE id = $1") + .bind(app) + .execute(pool) + .await + .expect("cleanup app"); +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn a_failed_fan_out_rolls_the_kv_write_back() { + let Some(pool) = pool_or_skip().await else { + return; + }; + let f = setup(&pool, "watched").await; + let writer = PostgresKvWriter::new(pool.clone()); + let cx = cx(f.app); + // A distinct, valid SQL identifier for this run's trigger/function names. + let tag = Uuid::new_v4().simple().to_string(); + + // --- Control: with the outbox healthy, the write lands AND fans out. ----- + writer + .set(&cx, "watched", "k", serde_json::json!(1)) + .await + .expect("healthy set succeeds"); + assert_eq!(kv_count(&pool, f.app).await, 1, "the row is written"); + assert_eq!( + outbox_count(&pool, f.app).await, + 1, + "the matching trigger fanned out" + ); + + // --- The fix: a fan-out failure must roll the write back. --------------- + break_outbox_for(&pool, f.app, &tag).await; + let err = writer + .set(&cx, "watched", "k2", serde_json::json!(2)) + .await + .expect_err("an outbox failure must surface as an error, not be swallowed"); + unbreak_outbox(&pool, &tag).await; + + assert!( + format!("{err}").contains("event emit"), + "the error should name the emit failure, got: {err}" + ); + assert_eq!( + kv_count(&pool, f.app).await, + 1, + "k2 must NOT be in the store — the write is rolled back with its fan-out. \ + A committed row whose trigger never fired is exactly the durability hole \ + the transactional outbox closes." + ); + assert_eq!( + outbox_count(&pool, f.app).await, + 1, + "and no outbox row for the rolled-back write" + ); + + // --- An unwatched collection still writes (no trigger → no fan-out). ---- + writer + .set(&cx, "unwatched", "k", serde_json::json!(3)) + .await + .expect("a write with no matching trigger needs no outbox row"); + assert_eq!(kv_count(&pool, f.app).await, 2); + assert_eq!(outbox_count(&pool, f.app).await, 1, "still no new fan-out"); + + cleanup(&pool, f.app).await; +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn a_failed_fan_out_rolls_a_delete_back() { + let Some(pool) = pool_or_skip().await else { + return; + }; + let f = setup(&pool, "watched").await; + let writer = PostgresKvWriter::new(pool.clone()); + let cx = cx(f.app); + let tag = Uuid::new_v4().simple().to_string(); + + writer + .set(&cx, "watched", "doomed", serde_json::json!(1)) + .await + .expect("seed"); + assert_eq!(kv_count(&pool, f.app).await, 1); + + // A delete that can't fan out must leave the row in place — otherwise the + // key is gone and nothing downstream ever learns it was removed. + break_outbox_for(&pool, f.app, &tag).await; + let err = writer + .delete(&cx, "watched", "doomed") + .await + .expect_err("outbox failure surfaces"); + unbreak_outbox(&pool, &tag).await; + assert!(format!("{err}").contains("event emit")); + assert_eq!( + kv_count(&pool, f.app).await, + 1, + "the delete rolled back — the key is still there" + ); + + cleanup(&pool, f.app).await; +} diff --git a/crates/picloud/src/lib.rs b/crates/picloud/src/lib.rs index ff041b0..d85ef8d 100644 --- a/crates/picloud/src/lib.rs +++ b/crates/picloud/src/lib.rs @@ -170,12 +170,18 @@ pub async fn build_app( let kv_repo = Arc::new(PostgresKvRepo::new(pool.clone())); let docs_repo = Arc::new(PostgresDocsRepo::new(pool.clone())); let events: Arc = Arc::new(OutboxEventEmitter::new(pool.clone())); - let kv: Arc = Arc::new(KvServiceImpl::with_max_value_bytes( - kv_repo.clone(), - authz.clone(), - events.clone(), - picloud_manager_core::kv_service::kv_max_value_bytes_from_env(), - )); + // `with_atomic_writes` makes each KV mutation and the trigger fan-out it + // produces commit in ONE transaction — an outbox failure rolls the write + // back rather than leaving a committed row whose trigger never fires. + let kv: Arc = Arc::new( + KvServiceImpl::with_max_value_bytes( + kv_repo.clone(), + authz.clone(), + events.clone(), + picloud_manager_core::kv_service::kv_max_value_bytes_from_env(), + ) + .with_atomic_writes(pool.clone()), + ); // §11.6 shared group collections (KV): a separate store keyed by the owning // group, resolved from cx.app_id's chain. Reuses the KV value-size cap. // §11.6 shared-collection repos hoisted so both the SDK services AND the M4