fix(kv): commit a write and its trigger fan-out in one transaction

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) <noreply@anthropic.com>

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<dyn KvRepo>,
     authz: Arc<dyn AuthzRepo>,
-    events: Arc<dyn ServiceEventEmitter>,
+    writer: Arc<dyn KvWriter>,
     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<bool, KvError> {
         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<bool, KvError> {
         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<bool, KvError> {