diff --git a/crates/buzz-audit/src/error.rs b/crates/buzz-audit/src/error.rs index b4ffd24d83..737661518c 100644 --- a/crates/buzz-audit/src/error.rs +++ b/crates/buzz-audit/src/error.rs @@ -35,11 +35,47 @@ pub enum AuditError { #[error("unknown audit action in database")] UnknownAction, + /// A batch append was given entries that do not all belong to the same + /// chain. Batches are single-community by contract (the caller partitions + /// before appending); this is a caller bug, never a data-dependent state. + #[error("audit batch entries do not share a single chain")] + MixedBatch, + /// A JSON serialization error occurred (e.g. while canonicalising `detail`). #[error("serialization error: {0}")] Serialization(#[from] serde_json::Error), } +impl AuditError { + /// True when retrying the **same input** cannot succeed — the failure is a + /// deterministic function of the entry (or batch), not of infrastructure + /// state. Callers use this to separate poison isolation (bisection) from + /// retry/backpressure: an outage must never classify healthy entries as + /// poison, and a poisoned entry must never be retried forever. + /// + /// Database errors are deterministic only for SQLSTATE classes `22` (data + /// exception) and `23` (integrity constraint violation) — entry-dependent + /// by construction. Everything else (pool, io, connection, unknown) is + /// treated as infrastructure: retry may succeed, so fail toward retrying. + pub fn is_deterministic(&self) -> bool { + match self { + AuditError::Serialization(_) | AuditError::MixedBatch => true, + // Read-side verification outcomes; not raised by appends, but if + // ever surfaced to a retry loop they cannot be fixed by retrying. + AuditError::ChainViolation { .. } + | AuditError::HashMismatch { .. } + | AuditError::UnknownAction => true, + AuditError::Database(e) => match e { + sqlx::Error::Database(db) => db + .code() + .map(|c| c.starts_with("22") || c.starts_with("23")) + .unwrap_or(false), + _ => false, + }, + } + } +} + #[cfg(test)] mod tests { use super::*; @@ -69,6 +105,7 @@ mod tests { AuditError::ChainViolation { seq: 7 }, AuditError::HashMismatch { seq: 42 }, AuditError::UnknownAction, + AuditError::MixedBatch, ]; for err in &domain_errors { diff --git a/crates/buzz-audit/src/service.rs b/crates/buzz-audit/src/service.rs index 913131a591..5d529ed4ce 100644 --- a/crates/buzz-audit/src/service.rs +++ b/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, QueryBuilder, Row}; use tracing::{debug, instrument, warn}; use uuid::Uuid; @@ -37,16 +37,52 @@ impl AuditService { /// Append a new entry to the calling community's chain. /// - /// Serialized per-community via `pg_advisory_lock`. Postgres advisory locks - /// are session-scoped, so we acquire before the transaction and release - /// after commit (or on any error path). + /// Delegates to [`Self::log_batch`] with a batch of one so there is exactly + /// one chain-append implementation (lock protocol, head read, hash linkage). #[instrument(skip(self, entry), fields(action = %entry.action))] pub async fn log(&self, entry: NewAuditEntry) -> Result { + let mut logged = self.log_batch(vec![entry]).await?; + Ok(logged + .pop() + .expect("log_batch of one entry returns one entry")) + } + + /// Append a batch of entries — all for the **same community** — to that + /// community's chain in one transaction. + /// + /// Contract: + /// - Every entry must share one `community_id`; a mixed batch is rejected + /// with [`AuditError::MixedBatch`] before any lock is taken (callers + /// partition per community — batches never span tenants, so one + /// community's failure can never roll back another's rows). + /// - Input order is preserved: entries are chained and inserted in the + /// order given. + /// - Atomic: either every entry in the batch commits or none does. + /// + /// Serialized per-community via `pg_advisory_lock` (the same key domain as + /// [`Self::log`]). Postgres advisory locks are session-scoped, so we + /// acquire before the transaction and release after commit (or on any + /// error path). The chain head is re-read from the DB inside the + /// transaction, under the lock — there is no in-memory head authority, so + /// this is safe across relay replicas. + #[instrument(skip(self, entries), fields(batch_len = entries.len()))] + pub async fn log_batch( + &self, + entries: Vec, + ) -> Result, AuditError> { + let Some(first) = entries.first() else { + return Ok(Vec::new()); + }; + let community = first.community_id; + if entries.iter().any(|e| e.community_id != community) { + return Err(AuditError::MixedBatch); + } + let mut conn = self.pool.acquire().await?; // Per-community advisory lock: hash the namespaced community id to an // i64 lock key inside Postgres. Communities lock independently. - let lock_key = format!("{AUDIT_LOCK_NAMESPACE}{}", entry.community_id); + let lock_key = format!("{AUDIT_LOCK_NAMESPACE}{community}"); sqlx::query("SELECT pg_advisory_lock(hashtextextended($1, 0))") .bind(&lock_key) .execute(&mut *conn) @@ -55,7 +91,7 @@ impl AuditService { // Run the chain append and release the lock regardless of outcome. // catch_unwind so a panic still releases the lock before the connection // returns to the pool. - let result = std::panic::AssertUnwindSafe(self.log_inner(&mut conn, entry)) + let result = std::panic::AssertUnwindSafe(self.log_batch_inner(&mut conn, entries)) .catch_unwind() .await; @@ -70,16 +106,17 @@ impl AuditService { } } - async fn log_inner( + async fn log_batch_inner( &self, conn: &mut sqlx::pool::PoolConnection, - entry: NewAuditEntry, - ) -> Result { + entries: Vec, + ) -> Result, AuditError> { let mut tx = conn.begin().await?; - // The stored row keys on the raw UUID; the typed `CommunityId` on the + // The stored rows key 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(); + // `log_batch` verified all entries share this community. + let community_id = *entries[0].community_id.as_uuid(); // Head of THIS community's chain — scoped by community_id. let head = sqlx::query( @@ -91,55 +128,67 @@ impl AuditService { .fetch_optional(&mut *tx) .await?; - let (prev_seq, prev_hash): (i64, Option>) = match head { + let (mut prev_seq, mut prev_hash): (i64, Option>) = match head { Some(row) => ( row.get::("seq"), Some(row.get::, _>("hash")), ), None => (0, None), // community's first entry }; - let seq = prev_seq + 1; - - let created_at: DateTime = Utc::now(); - - let mut audit_entry = AuditEntry { - community_id, - seq, - hash: Vec::new(), - prev_hash, - action: entry.action, - actor_pubkey: entry.actor_pubkey, - object_id: entry.object_id, - detail: entry.detail, - created_at, - }; - audit_entry.hash = compute_hash(&audit_entry)?.to_vec(); + // Chain the batch in input order: each entry links to its predecessor + // (the last committed row for the first entry, the previous batch + // member thereafter). + let mut logged = Vec::with_capacity(entries.len()); + for entry in entries { + let seq = prev_seq + 1; + let created_at: DateTime = Utc::now(); + + let mut audit_entry = AuditEntry { + community_id, + seq, + hash: Vec::new(), + prev_hash, + action: entry.action, + actor_pubkey: entry.actor_pubkey, + object_id: entry.object_id, + detail: entry.detail, + created_at, + }; + audit_entry.hash = compute_hash(&audit_entry)?.to_vec(); + + prev_seq = seq; + prev_hash = Some(audit_entry.hash.clone()); + logged.push(audit_entry); + } - debug!(seq, "writing audit entry"); + debug!( + first_seq = logged.first().map(|e| e.seq), + last_seq = logged.last().map(|e| e.seq), + "writing audit entries" + ); - sqlx::query( - r#" - INSERT INTO audit_log - (community_id, seq, hash, prev_hash, action, actor_pubkey, object_id, detail, created_at) - VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9) - "#, - ) - .bind(audit_entry.community_id) - .bind(audit_entry.seq) - .bind(&audit_entry.hash) - .bind(audit_entry.prev_hash.as_deref()) - .bind(audit_entry.action.as_str()) - .bind(audit_entry.actor_pubkey.as_deref()) - .bind(audit_entry.object_id.as_deref()) - .bind(&audit_entry.detail) - .bind(audit_entry.created_at) - .execute(&mut *tx) - .await?; + // One multi-row INSERT for the whole batch. + let mut qb: QueryBuilder = QueryBuilder::new( + "INSERT INTO audit_log + (community_id, seq, hash, prev_hash, action, actor_pubkey, object_id, detail, created_at) ", + ); + qb.push_values(logged.iter(), |mut b, e| { + b.push_bind(e.community_id) + .push_bind(e.seq) + .push_bind(&e.hash) + .push_bind(e.prev_hash.as_deref()) + .push_bind(e.action.as_str()) + .push_bind(e.actor_pubkey.as_deref()) + .push_bind(e.object_id.as_deref()) + .push_bind(&e.detail) + .push_bind(e.created_at); + }); + qb.build().execute(&mut *tx).await?; tx.commit().await?; - Ok(audit_entry) + Ok(logged) } /// Verify the hash chain for one community over `[from_seq, to_seq]`. @@ -486,6 +535,149 @@ mod tests { assert!(matches!(r, Err(AuditError::HashMismatch { seq: 1 }))); } + /// Batched appends chain exactly like sequential singles: within the + /// batch each entry links to its predecessor, the first links to the + /// pre-batch head, and the whole chain verifies. + #[tokio::test] + #[ignore = "requires Postgres"] + async fn batch_chains_in_order_and_links_to_prior_head() { + let _g = db_lock().lock().await; + let Some(pool) = test_pool().await else { + return; + }; + let svc = AuditService::new(pool.clone()); + let c = make_community(&pool).await; + + // Pre-batch head via the single-entry path. + let e1 = svc + .log(new_entry(c, AuditAction::EventCreated)) + .await + .unwrap(); + + let logged = svc + .log_batch(vec![ + new_entry(c, AuditAction::ChannelCreated), + new_entry(c, AuditAction::MemberAdded), + new_entry(c, AuditAction::EventDeleted), + ]) + .await + .unwrap(); + + assert_eq!(logged.len(), 3); + assert_eq!(logged[0].seq, 2); + assert_eq!(logged[1].seq, 3); + assert_eq!(logged[2].seq, 4); + assert_eq!(logged[0].prev_hash.as_deref(), Some(e1.hash.as_slice())); + assert_eq!( + logged[1].prev_hash.as_deref(), + Some(logged[0].hash.as_slice()) + ); + assert_eq!( + logged[2].prev_hash.as_deref(), + Some(logged[1].hash.as_slice()) + ); + assert!(svc + .verify_chain(CommunityId::from_uuid(c), 1, 4) + .await + .unwrap()); + + // And a single append after the batch links to the batch's tail. + let e5 = svc + .log(new_entry(c, AuditAction::ChannelDeleted)) + .await + .unwrap(); + assert_eq!(e5.seq, 5); + assert_eq!(e5.prev_hash.as_deref(), Some(logged[2].hash.as_slice())); + } + + /// A mixed-community batch is rejected before any write; the error is + /// classified deterministic (retrying identical input cannot succeed). + #[tokio::test] + #[ignore = "requires Postgres"] + async fn mixed_batch_is_rejected_without_writing() { + let _g = db_lock().lock().await; + let Some(pool) = test_pool().await else { + return; + }; + let svc = AuditService::new(pool.clone()); + let a = make_community(&pool).await; + let b = make_community(&pool).await; + + let err = svc + .log_batch(vec![ + new_entry(a, AuditAction::EventCreated), + new_entry(b, AuditAction::EventCreated), + ]) + .await + .unwrap_err(); + assert!(matches!(err, AuditError::MixedBatch)); + assert!(err.is_deterministic()); + + for community in [a, b] { + let rows = svc + .get_entries(CommunityId::from_uuid(community), 1, 10) + .await + .unwrap(); + assert!(rows.is_empty(), "mixed batch must write nothing"); + } + } + + #[tokio::test] + async fn empty_batch_is_a_noop() { + let Some(pool) = test_pool().await else { + return; + }; + let svc = AuditService::new(pool); + assert!(svc.log_batch(Vec::new()).await.unwrap().is_empty()); + } + + /// Atomicity: a batch containing one entry Postgres rejects (jsonb cannot + /// store \u0000 — SQLSTATE 22P05) commits nothing, and the error is + /// classified deterministic so the caller bisects instead of retrying. + #[tokio::test] + #[ignore = "requires Postgres"] + async fn poisoned_batch_is_atomic_and_deterministic() { + let _g = db_lock().lock().await; + let Some(pool) = test_pool().await else { + return; + }; + let svc = AuditService::new(pool.clone()); + let c = make_community(&pool).await; + + let mut poison = new_entry(c, AuditAction::MemberAdded); + poison.detail = serde_json::json!({"bad": "nul\u{0}byte"}); + + let err = svc + .log_batch(vec![ + new_entry(c, AuditAction::EventCreated), + poison, + new_entry(c, AuditAction::ChannelCreated), + ]) + .await + .unwrap_err(); + assert!( + err.is_deterministic(), + "jsonb NUL rejection must classify deterministic, got: {err}" + ); + + let rows = svc + .get_entries(CommunityId::from_uuid(c), 1, 10) + .await + .unwrap(); + assert!(rows.is_empty(), "failed batch must roll back atomically"); + } + + /// Classification unit checks that need no Postgres: infra-shaped sqlx + /// errors must NOT be deterministic (outages are retried, not bisected). + #[test] + fn infra_errors_are_not_deterministic() { + assert!(!AuditError::Database(sqlx::Error::PoolTimedOut).is_deterministic()); + assert!(!AuditError::Database(sqlx::Error::WorkerCrashed).is_deterministic()); + let ser = serde_json::from_str::("not json").unwrap_err(); + assert!(AuditError::Serialization(ser).is_deterministic()); + assert!(AuditError::MixedBatch.is_deterministic()); + } + #[tokio::test] #[ignore = "requires Postgres"] async fn verify_empty_range_is_false() { diff --git a/crates/buzz-relay/src/config.rs b/crates/buzz-relay/src/config.rs index dca34ec8e3..8bbc634efb 100644 --- a/crates/buzz-relay/src/config.rs +++ b/crates/buzz-relay/src/config.rs @@ -190,6 +190,15 @@ pub struct Config { /// serving media GET/HEAD. Default off for staged client rollout. pub require_media_get_auth: bool, + /// Maximum entries drained into one audit flush batch. + /// Set via `BUZZ_AUDIT_BATCH_MAX` (default 100, must be positive). + pub audit_batch_max: usize, + /// How long the audit worker waits to fill a batch after its first entry + /// arrives. Set via `BUZZ_AUDIT_BATCH_INTERVAL_MS` (default 50, must be + /// positive). A batch flushes when it reaches `audit_batch_max` entries or + /// this window elapses, whichever is first. + pub audit_batch_interval: Duration, + /// Optional override for ephemeral channel TTL (in seconds). /// When set, any channel created with a TTL tag will use this value instead /// of the client-provided one. Useful for testing ephemeral expiry quickly. @@ -652,6 +661,10 @@ impl Config { }) .unwrap_or(false); + let audit_batch_max = positive_u64_from_env("BUZZ_AUDIT_BATCH_MAX", 100)? as usize; + let audit_batch_interval = + Duration::from_millis(positive_u64_from_env("BUZZ_AUDIT_BATCH_INTERVAL_MS", 50)?); + let ephemeral_ttl_override = std::env::var("BUZZ_EPHEMERAL_TTL_OVERRIDE") .ok() .and_then(|v| v.parse::().ok()) @@ -849,6 +862,8 @@ impl Config { media_max_concurrent_uploads_per_pubkey, media_uploads_per_minute, require_media_get_auth, + audit_batch_max, + audit_batch_interval, ephemeral_ttl_override, git_repo_path, git_max_pack_bytes, diff --git a/crates/buzz-relay/src/state.rs b/crates/buzz-relay/src/state.rs index bc41cdcc97..8a495df60d 100644 --- a/crates/buzz-relay/src/state.rs +++ b/crates/buzz-relay/src/state.rs @@ -588,29 +588,65 @@ impl AppState { let audit_for_worker = Arc::clone(&audit_arc); let audit_cancel = CancellationToken::new(); let audit_cancel_worker = audit_cancel.clone(); + let audit_batch_max = config.audit_batch_max.max(1); + let audit_batch_interval = config.audit_batch_interval; let audit_worker_handle = tokio::spawn(async move { - // Normal operation: process entries as they arrive. - loop { + // Normal operation: batch entries as they arrive. A batch opens + // when the first entry is received and flushes when it reaches + // `audit_batch_max` entries or `audit_batch_interval` elapses, + // whichever comes first — so a lone entry still lands within one + // interval and a burst amortizes to one txn per community per batch. + let mut batch: Vec = Vec::with_capacity(audit_batch_max); + 'run: loop { + // Wait (unbounded) for the batch-opening entry. tokio::select! { entry = audit_rx.recv() => { match entry { - Some(entry) => log_audit_entry(&audit_for_worker, entry).await, - None => break, // channel closed + Some(entry) => batch.push(entry), + None => break 'run, // channel closed } } _ = audit_cancel_worker.cancelled() => { // Close the receiver: rejects future sends and lets us // drain everything already buffered without a race. audit_rx.close(); - break; + break 'run; } } + // Fill window: bounded by size and elapsed time. + let deadline = tokio::time::Instant::now() + audit_batch_interval; + while batch.len() < audit_batch_max { + tokio::select! { + entry = audit_rx.recv() => { + match entry { + Some(entry) => batch.push(entry), + None => break, // closed: flush, then exit via outer recv + } + } + _ = tokio::time::sleep_until(deadline) => break, + _ = audit_cancel_worker.cancelled() => { + audit_rx.close(); + break; + } + } + } + flush_audit_batch(&audit_for_worker, std::mem::take(&mut batch)).await; } // Drain: recv() returns buffered entries, then None once empty. + // Batched like the normal path so shutdown flushes are fast. let mut drained = 0u32; - while let Some(entry) = audit_rx.recv().await { - log_audit_entry(&audit_for_worker, entry).await; - drained += 1; + loop { + while batch.len() < audit_batch_max { + match audit_rx.recv().await { + Some(entry) => batch.push(entry), + None => break, + } + } + if batch.is_empty() { + break; + } + drained += batch.len() as u32; + flush_audit_batch(&audit_for_worker, std::mem::take(&mut batch)).await; } if drained > 0 { tracing::info!(drained, "audit worker flushed remaining entries"); @@ -1104,16 +1140,111 @@ impl AuditShutdownHandle { } } -/// Log a single audit entry with metrics. Extracted so the normal loop -/// and the post-cancel drain share the same logic. -async fn log_audit_entry(audit: &buzz_audit::AuditService, entry: buzz_audit::NewAuditEntry) { - let t = std::time::Instant::now(); - if let Err(e) = audit.log(entry).await { - metrics::counter!("buzz_audit_log_errors_total").increment(1); - tracing::error!("Audit log failed: {e}"); - } else { - metrics::histogram!("buzz_audit_log_seconds").record(t.elapsed().as_secs_f64()); +/// Flush a drained batch: partition FIFO-per-community, then append each +/// community's run atomically via `AuditService::log_batch`. Extracted so the +/// normal loop and the post-cancel drain share the same logic. +/// +/// Partitioning preserves arrival order **within** each community (the only +/// order the hash chain observes); cross-community order carries no chain +/// meaning and communities already interleave freely under the per-community +/// advisory lock. +async fn flush_audit_batch( + audit: &buzz_audit::AuditService, + batch: Vec, +) { + if batch.is_empty() { + return; + } + metrics::histogram!("buzz_audit_batch_entries").record(batch.len() as f64); + // Linear-scan partition: batches hold at most a handful of distinct + // communities per drain window, so a Vec beats a HashMap here and keeps + // first-seen community order deterministic. + let mut per_community: Vec<( + buzz_core::tenant::CommunityId, + Vec, + )> = Vec::new(); + for entry in batch { + match per_community + .iter_mut() + .find(|(community, _)| *community == entry.community_id) + { + Some((_, entries)) => entries.push(entry), + None => per_community.push((entry.community_id, vec![entry])), + } + } + for (_, entries) in per_community { + append_isolating_poison(audit, entries).await; + } +} + +/// How many times an infrastructure failure (pool exhaustion, connection +/// drop, statement timeout) is retried before the slice is dropped with an +/// error, matching the drop-on-error contract the unbatched worker had. +const AUDIT_INFRA_ATTEMPTS: u32 = 3; +const AUDIT_INFRA_BACKOFF: std::time::Duration = std::time::Duration::from_millis(200); + +/// Append one community's entries, isolating poison to single entries. +/// +/// - Infrastructure errors are retried `AUDIT_INFRA_ATTEMPTS` times with +/// backoff: an outage must never classify healthy entries as poison. +/// - Deterministic errors (`AuditError::is_deterministic`) bisect the slice: +/// halves re-append independently (each `log_batch` re-reads the chain head +/// under the community lock, so linkage survives splitting), converging on +/// the poisoned entry, which alone is dropped. A batch of N with one bad +/// entry loses exactly 1 — never the other N-1. +async fn append_isolating_poison( + audit: &buzz_audit::AuditService, + entries: Vec, +) { + // Recursive async fn needs boxing; depth is log2(batch_max), trivial. + fn go<'a>( + audit: &'a buzz_audit::AuditService, + entries: Vec, + ) -> std::pin::Pin + Send + 'a>> { + Box::pin(async move { + if entries.is_empty() { + return; + } + let mut attempt = 0u32; + loop { + attempt += 1; + let t = std::time::Instant::now(); + match audit.log_batch(entries.clone()).await { + Ok(_) => { + metrics::histogram!("buzz_audit_log_seconds") + .record(t.elapsed().as_secs_f64()); + return; + } + Err(e) if !e.is_deterministic() && attempt < AUDIT_INFRA_ATTEMPTS => { + tracing::warn!( + attempt, + len = entries.len(), + "audit batch append hit infra error, retrying: {e}" + ); + tokio::time::sleep(AUDIT_INFRA_BACKOFF * attempt).await; + } + Err(e) if e.is_deterministic() && entries.len() > 1 => { + tracing::warn!( + len = entries.len(), + "audit batch append failed deterministically, bisecting: {e}" + ); + let mut left = entries; + let right = left.split_off(left.len() / 2); + go(audit, left).await; + go(audit, right).await; + return; + } + Err(e) => { + metrics::counter!("buzz_audit_log_errors_total") + .increment(entries.len() as u64); + tracing::error!(dropped = entries.len(), "Audit log failed: {e}"); + return; + } + } + } + }) } + go(audit, entries).await } impl std::fmt::Debug for AppState { @@ -1701,4 +1832,113 @@ mod tests { "community-B session stays live — ban does not cross the tenant fence" ); } + + async fn audit_test_pool() -> Option { + let url = std::env::var("DATABASE_URL") + .unwrap_or_else(|_| "postgres://buzz:buzz_dev@localhost:5432/buzz".into()); + sqlx::PgPool::connect(&url).await.ok() + } + + async fn make_audit_community(pool: &sqlx::PgPool) -> Uuid { + let id = Uuid::new_v4(); + sqlx::query("INSERT INTO communities (id, host) VALUES ($1, $2)") + .bind(id) + .bind(format!("audit-batch-{id}.example")) + .execute(pool) + .await + .expect("insert test community"); + id + } + + fn audit_entry(community: Uuid, marker: &str) -> buzz_audit::NewAuditEntry { + buzz_audit::NewAuditEntry { + community_id: buzz_core::tenant::CommunityId::from_uuid(community), + action: buzz_audit::AuditAction::EventCreated, + actor_pubkey: None, + object_id: Some(marker.to_string()), + detail: serde_json::json!({ "marker": marker }), + } + } + + /// THE poison-isolation property: a batch of N with one Postgres-rejected + /// entry (jsonb NUL, SQLSTATE 22P05 — deterministic) commits exactly N-1 + /// entries, in order, with an intact verifiable chain. Bisection converges + /// on the poison; healthy neighbors are never dropped. + #[tokio::test] + #[ignore = "requires Postgres"] + async fn poison_batch_drops_exactly_the_poisoned_entry() { + let Some(pool) = audit_test_pool().await else { + return; + }; + let svc = buzz_audit::AuditService::new(pool.clone()); + let c = make_audit_community(&pool).await; + + let mut batch: Vec = + (0..7).map(|i| audit_entry(c, &format!("ok-{i}"))).collect(); + let mut poison = audit_entry(c, "poison"); + poison.detail = serde_json::json!({ "bad": "nul\u{0}byte" }); + batch.insert(3, poison); // 8 entries, poison mid-batch + + append_isolating_poison(&svc, batch).await; + + let rows = svc + .get_entries(buzz_core::tenant::CommunityId::from_uuid(c), 1, 100) + .await + .unwrap(); + let markers: Vec<_> = rows.iter().filter_map(|e| e.object_id.clone()).collect(); + assert_eq!( + markers, + (0..7).map(|i| format!("ok-{i}")).collect::>(), + "exactly the poisoned entry is dropped, order preserved" + ); + assert!( + svc.verify_chain(buzz_core::tenant::CommunityId::from_uuid(c), 1, 7) + .await + .unwrap(), + "chain must verify after bisection re-linked around the poison" + ); + } + + /// Cross-tenant fence for the flush path: one community's poison drops + /// nothing from another community sharing the same drained batch. + #[tokio::test] + #[ignore = "requires Postgres"] + async fn flush_partitions_poison_to_its_own_community() { + let Some(pool) = audit_test_pool().await else { + return; + }; + let svc = buzz_audit::AuditService::new(pool.clone()); + let a = make_audit_community(&pool).await; + let b = make_audit_community(&pool).await; + + let mut a_poison = audit_entry(a, "a-poison"); + a_poison.detail = serde_json::json!({ "bad": "nul\u{0}byte" }); + // Interleaved arrival order across tenants. + let batch = vec![ + audit_entry(a, "a-0"), + audit_entry(b, "b-0"), + a_poison, + audit_entry(b, "b-1"), + audit_entry(a, "a-1"), + ]; + + flush_audit_batch(&svc, batch).await; + + let a_rows = svc + .get_entries(buzz_core::tenant::CommunityId::from_uuid(a), 1, 10) + .await + .unwrap(); + let b_rows = svc + .get_entries(buzz_core::tenant::CommunityId::from_uuid(b), 1, 10) + .await + .unwrap(); + let a_markers: Vec<_> = a_rows.iter().filter_map(|e| e.object_id.clone()).collect(); + let b_markers: Vec<_> = b_rows.iter().filter_map(|e| e.object_id.clone()).collect(); + assert_eq!(a_markers, vec!["a-0", "a-1"], "A loses only its poison"); + assert_eq!( + b_markers, + vec!["b-0", "b-1"], + "B is untouched by A's poison" + ); + } }