fix(audit): make delivery durable and tenant-fair

crates/buzz-audit/src/entry.rs

@@ -43,11 +43,11 @@ pub struct AuditEntry {
 /// `TenantContext`), never a client-supplied value — the same provenance rule
 /// the whole multi-tenant model rests on.
 ///
-/// Not `Serialize`/`Deserialize`: this is an in-process input struct (consumed
-/// by `AuditService::log`, threaded through the in-memory audit sink), never
-/// crossing a wire or DB boundary as a whole. Keeping it non-deserializable
-/// reinforces the fence — there is no path by which a client-supplied blob
-/// becomes a `NewAuditEntry` (and thus a `CommunityId`).
+/// Not `Serialize`/`Deserialize`: this is an in-process input struct consumed by
+/// `AuditService`; durable outbox columns are bound individually, never decoded
+/// from a client-supplied blob. Keeping it non-deserializable reinforces the
+/// fence — there is no path by which client JSON becomes a `NewAuditEntry` (and
+/// thus a `CommunityId`).
 #[derive(Debug, Clone, PartialEq, Eq)]
 pub struct NewAuditEntry {
     /// Server-resolved community this entry belongs to. Typed as [`CommunityId`]

crates/buzz-audit/src/lib.rs

@@ -32,4 +32,4 @@ pub use action::AuditAction;
 pub use entry::{AuditEntry, NewAuditEntry};
 pub use error::AuditError;
 pub use hash::{compute_hash, GENESIS_HASH};
-pub use service::AuditService;
+pub use service::{AuditService, PendingAuditResult};

crates/buzz-audit/src/service.rs

@@ -1,6 +1,6 @@
 use chrono::{DateTime, Utc};
 use futures_util::FutureExt as _;
-use sqlx::{Acquire, PgPool, Row};
+use sqlx::{Acquire, PgPool, Postgres, Row, Transaction};
 use tracing::{debug, warn};
 use uuid::Uuid;
 
@@ -26,9 +26,35 @@ fn log_timestamp() -> DateTime<Utc> {
 /// Per-community advisory lock key. Derived in Postgres from the community UUID
 /// so two communities never serialize each other's audit writes (which would be
 /// both a throughput bottleneck and a cross-tenant timing oracle). The lock is
-/// taken with `pg_advisory_lock(hashtextextended(...))` — see [`AuditService::log`].
+/// taken with `pg_advisory_lock` for direct appends and
+/// `pg_advisory_xact_lock` for durable outbox delivery.
 const AUDIT_LOCK_NAMESPACE: &str = "buzz_audit:";
 
+/// Outcome of one durable audit-outbox delivery attempt.
+#[derive(Debug)]
+pub enum PendingAuditResult {
+    /// No community currently has a due head entry.
+    Idle,
+    /// One pending entry was appended and removed from the outbox atomically.
+    Appended,
+    /// A timeout deferred the entry without blocking other communities.
+    Deferred {
+        /// Number of delivery attempts made for this entry.
+        attempt_count: i32,
+        /// Delay before the entry becomes eligible again.
+        retry_delay_ms: u64,
+        /// PostgreSQL SQLSTATE that made the attempt retryable.
+        sqlstate: String,
+    },
+}
+
+struct ClaimedAuditEntry {
+    id: Uuid,
+    entry: NewAuditEntry,
+    enqueued_at: DateTime<Utc>,
+    attempt_count: i32,
+}
+
 /// Append-only, per-community hash-chain audit log backed by Postgres.
 ///
 /// Each community has an independent chain keyed `(community_id, seq)`. Writes
@@ -45,6 +71,172 @@ impl AuditService {
         Self { pool }
     }
 
+    /// Persist an audit intent before the producer reports completion.
+    ///
+    /// Delivery is asynchronous, but the intent survives relay shutdown and is
+    /// ordered by `enqueue_seq` within its community. A logical `dedupe_key`
+    /// suppresses retries after both pending and completed delivery.
+    pub async fn enqueue(
+        &self,
+        entry: NewAuditEntry,
+        dedupe_key: Option<&str>,
+    ) -> Result<(), AuditError> {
+        let mut tx = self.pool.begin().await?;
+        if let Some(dedupe_key) = dedupe_key {
+            let inserted = sqlx::query(
+                "INSERT INTO audit_delivery_keys (community_id, dedupe_key) \
+                 VALUES ($1, $2) \
+                 ON CONFLICT (community_id, dedupe_key) DO NOTHING",
+            )
+            .bind(entry.community_id.as_uuid())
+            .bind(dedupe_key)
+            .execute(&mut *tx)
+            .await?;
+            if inserted.rows_affected() == 0 {
+                tx.commit().await?;
+                return Ok(());
+            }
+        }
+        sqlx::query(
+            "INSERT INTO audit_outbox \
+                 (community_id, action, actor_pubkey, object_id, detail) \
+             VALUES ($1, $2, $3, $4, $5)",
+        )
+        .bind(entry.community_id.as_uuid())
+        .bind(entry.action.as_str())
+        .bind(entry.actor_pubkey.as_deref())
+        .bind(entry.object_id.as_deref())
+        .bind(&entry.detail)
+        .execute(&mut *tx)
+        .await?;
+        tx.commit().await?;
+        Ok(())
+    }
+
+    /// Deliver one due per-community head from the durable outbox.
+    ///
+    /// A short claim lease lets multiple relay processes share the outbox.
+    /// Timeout failures are rescheduled with capped exponential backoff, so a
+    /// contended community cannot monopolize a worker. Appending the hash-chain
+    /// row and deleting its outbox row occur in the same transaction.
+    pub async fn deliver_next_pending(&self) -> Result<PendingAuditResult, AuditError> {
+        let Some(claimed) = self.claim_next_pending().await? else {
+            return Ok(PendingAuditResult::Idle);
+        };
+
+        match self.append_claimed(&claimed).await {
+            Ok(()) => Ok(PendingAuditResult::Appended),
+            Err(error) => {
+                let Some(sqlstate) = retryable_sqlstate(&error) else {
+                    return Err(error);
+                };
+                let retry_delay_ms = retry_delay_ms(claimed.attempt_count);
+                self.defer_claimed(&claimed, retry_delay_ms).await?;
+                Ok(PendingAuditResult::Deferred {
+                    attempt_count: claimed.attempt_count,
+                    retry_delay_ms,
+                    sqlstate,
+                })
+            }
+        }
+    }
+
+    async fn claim_next_pending(&self) -> Result<Option<ClaimedAuditEntry>, AuditError> {
+        let row = sqlx::query(
+            r#"
+            WITH candidate AS (
+                SELECT pending.community_id, pending.id
+                FROM audit_outbox AS pending
+                WHERE pending.next_attempt_at <= clock_timestamp()
+                  AND NOT EXISTS (
+                      SELECT 1
+                      FROM audit_outbox AS older
+                      WHERE older.community_id = pending.community_id
+                        AND older.enqueue_seq < pending.enqueue_seq
+                  )
+                ORDER BY pending.next_attempt_at, pending.enqueue_seq
+                FOR UPDATE OF pending SKIP LOCKED
+                LIMIT 1
+            )
+            UPDATE audit_outbox AS pending
+            SET attempt_count = pending.attempt_count + 1,
+                next_attempt_at = clock_timestamp() + INTERVAL '5 seconds'
+            FROM candidate
+            WHERE pending.community_id = candidate.community_id
+              AND pending.id = candidate.id
+            RETURNING pending.id, pending.community_id, pending.action,
+                      pending.actor_pubkey, pending.object_id, pending.detail,
+                      pending.enqueued_at, pending.attempt_count
+            "#,
+        )
+        .fetch_optional(&self.pool)
+        .await?;
+
+        let Some(row) = row else {
+            return Ok(None);
+        };
+        let action_string: String = row.get("action");
+        let action = action_string
+            .parse()
+            .map_err(|_| AuditError::UnknownAction)?;
+        Ok(Some(ClaimedAuditEntry {
+            id: row.get("id"),
+            entry: NewAuditEntry {
+                community_id: CommunityId::from_uuid(row.get("community_id")),
+                action,
+                actor_pubkey: row.get("actor_pubkey"),
+                object_id: row.get("object_id"),
+                detail: row.get("detail"),
+            },
+            enqueued_at: row.get("enqueued_at"),
+            attempt_count: row.get("attempt_count"),
+        }))
+    }
+
+    async fn append_claimed(&self, claimed: &ClaimedAuditEntry) -> Result<(), AuditError> {
+        let mut tx = self.pool.begin().await?;
+        let lock_key = format!("{AUDIT_LOCK_NAMESPACE}{}", claimed.entry.community_id);
+        sqlx::query("SELECT pg_advisory_xact_lock(hashtextextended($1, 0))")
+            .bind(&lock_key)
+            .execute(&mut *tx)
+            .await?;
+        Self::append_in_transaction(
+            &mut tx,
+            claimed.entry.clone(),
+            to_storage_precision(claimed.enqueued_at),
+        )
+        .await?;
+        let deleted = sqlx::query("DELETE FROM audit_outbox WHERE community_id = $1 AND id = $2")
+            .bind(claimed.entry.community_id.as_uuid())
+            .bind(claimed.id)
+            .execute(&mut *tx)
+            .await?;
+        if deleted.rows_affected() != 1 {
+            return Err(AuditError::Database(sqlx::Error::RowNotFound));
+        }
+        tx.commit().await?;
+        Ok(())
+    }
+
+    async fn defer_claimed(
+        &self,
+        claimed: &ClaimedAuditEntry,
+        retry_delay_ms: u64,
+    ) -> Result<(), AuditError> {
+        sqlx::query(
+            "UPDATE audit_outbox \
+             SET next_attempt_at = clock_timestamp() \
+                 + make_interval(secs => $3::double precision / 1000.0) \
+             WHERE community_id = $1 AND id = $2",
+        )
+        .bind(claimed.entry.community_id.as_uuid())
+        .bind(claimed.id)
+        .bind(retry_delay_ms as f64)
+        .execute(&self.pool)
+        .await?;
+        Ok(())
+    }
+
     /// Append a new entry to the calling community's chain.
     ///
     /// Serialized per-community via `pg_advisory_lock`. Postgres advisory locks
@@ -91,6 +283,17 @@ impl AuditService {
     ) -> Result<AuditEntry, AuditError> {
         let mut tx = conn.begin().await?;
 
+        let audit_entry = Self::append_in_transaction(&mut tx, entry, log_timestamp()).await?;
+        tx.commit().await?;
+
+        Ok(audit_entry)
+    }
+
+    async fn append_in_transaction(
+        tx: &mut Transaction<'_, Postgres>,
+        entry: NewAuditEntry,
+        created_at: DateTime<Utc>,
+    ) -> Result<AuditEntry, AuditError> {
         // The stored row keys on the raw UUID; the typed `CommunityId` on the
         // input is the provenance fence, dereferenced here at the DB boundary.
         let community_id = *entry.community_id.as_uuid();
@@ -102,7 +305,7 @@ impl AuditService {
              ORDER BY seq DESC LIMIT 1",
         )
         .bind(community_id)
-        .fetch_optional(&mut *tx)
+        .fetch_optional(&mut **tx)
         .await?;
 
         let (prev_seq, prev_hash): (i64, Option<Vec<u8>>) = match head {
@@ -114,8 +317,6 @@ impl AuditService {
         };
         let seq = prev_seq + 1;
 
-        let created_at: DateTime<Utc> = log_timestamp();
-
         let mut audit_entry = AuditEntry {
             community_id,
             seq,
@@ -148,11 +349,9 @@ impl AuditService {
         .bind(audit_entry.object_id.as_deref())
         .bind(&audit_entry.detail)
         .bind(audit_entry.created_at)
-        .execute(&mut *tx)
+        .execute(&mut **tx)
         .await?;
 
-        tx.commit().await?;
-
         Ok(audit_entry)
     }
 
@@ -248,6 +447,19 @@ impl AuditService {
     }
 }
 
+fn retryable_sqlstate(error: &AuditError) -> Option<String> {
+    let AuditError::Database(sqlx::Error::Database(database_error)) = error else {
+        return None;
+    };
+    let code = database_error.code()?;
+    matches!(code.as_ref(), "55P03" | "57014").then(|| code.into_owned())
+}
+
+fn retry_delay_ms(attempt_count: i32) -> u64 {
+    let exponent = attempt_count.saturating_sub(1).clamp(0, 5) as u32;
+    (50u64.saturating_mul(1u64 << exponent)).min(1_000)
+}
+
 fn row_to_audit_entry(row: &sqlx::postgres::PgRow) -> Result<AuditEntry, AuditError> {
     let action_str: String = row.get("action");
     let action: AuditAction = action_str.parse().map_err(|_| {

crates/buzz-db/src/runtime/migration.rs

@@ -699,7 +699,7 @@ mod tests {
         let mut migrations: Vec<_> = MIGRATOR.iter().collect();
         migrations.sort_by_key(|migration| migration.version);
 
-        assert_eq!(migrations.len(), 40);
+        assert_eq!(migrations.len(), 41);
         assert_eq!(migrations[0].version, 1);
         assert_eq!(&*migrations[0].description, "initial schema");
         assert!(migrations[0]
@@ -1338,6 +1338,26 @@ mod tests {
         assert!(!desired_schema.contains("NEW.kind IN (7, 9, 1059, 40007, 46010)"));
     }
 
+    #[test]
+    fn durable_audit_outbox_is_additive_deduplicated_and_fenced() {
+        let migration = MIGRATOR
+            .iter()
+            .find(|migration| migration.version == 41)
+            .expect("durable audit outbox migration");
+        let sql = migration.sql.as_str();
+        assert!(sql.contains("CREATE TABLE audit_outbox"));
+        assert!(sql.contains("CREATE TABLE audit_delivery_keys"));
+        assert!(sql.contains("GENERATED ALWAYS AS IDENTITY"));
+        assert!(sql.contains("attach_community_write_fence('audit_outbox')"));
+        assert!(sql.contains("attach_community_write_fence('audit_delivery_keys')"));
+
+        let desired_schema = include_str!("../../../../schema/schema.sql");
+        assert!(desired_schema.contains("CREATE TABLE audit_outbox"));
+        assert!(desired_schema.contains("CREATE TABLE audit_delivery_keys"));
+        assert!(desired_schema.contains("attach_community_write_fence('audit_outbox')"));
+        assert!(desired_schema.contains("attach_community_write_fence('audit_delivery_keys')"));
+    }
+
     #[test]
     fn migration_lint_detects_tables_missing_community_id_by_default() {
         let sql = r#"
@@ -1776,6 +1796,9 @@ mod tests {
         let mut expected_fences = migration.fence_attachments.clone();
         expected_fences.remove("product_feedback");
         expected_fences.remove("rate_limit_violations");
+        for additive in MIGRATOR.iter().filter(|migration| migration.version > 29) {
+            expected_fences.extend(surface(additive.sql.as_ref()).fence_attachments);
+        }
         assert_eq!(
             expected_fences, schema.fence_attachments,
             "write-fence attachment targets differ after recovery policy"
@@ -2425,6 +2448,8 @@ mod tests {
             "channels",
             "scheduled_workflow_fires",
             "audit_log",
+            "audit_outbox",
+            "audit_delivery_keys",
         ] {
             let exists = sqlx::query_scalar::<_, bool>(
                 "SELECT EXISTS (SELECT 1 FROM information_schema.tables WHERE table_schema = 'public' AND table_name = $1)",