jerry/buzz
1
0
Fork 0
mirror of https://github.com/block/buzz.git synced 2026-10-30 22:11:36 +00:00
Code Issues Packages Projects Releases Wiki Activity

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

Browse Source 
Signed-off-by: Luke Tornquist <tornquist@squareup.com>
This commit is contained in:
Luke Tornquist
2026-08-31 16:33:04 -04:00
parent cb3144999b
commit 8c1b636d44
16 changed files with 1231 additions and 221 deletions
Download Patch File Download Diff File Expand all files Collapse all files

10
crates/buzz-audit/src/entry.rs
View File

@@ -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`]

2
crates/buzz-audit/src/lib.rs
View File

@@ -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};

228
crates/buzz-audit/src/service.rs
View File

@@ -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(|_| {

27
crates/buzz-db/src/runtime/migration.rs
View File

@@ -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)",

58
crates/buzz-db/src/runtime/tests.rs
View File

@@ -2389,6 +2389,64 @@ async fn session_timeouts_install_through_db_new_and_bound_lock_waits() {
drop_scratch_db(&admin, db.pool.clone(), &name).await;
}
/// The idle-transaction timeout must terminate the wedged backend, and SQLx
/// must replace it with a freshly configured writer connection.
#[tokio::test]
#[ignore = "requires Postgres"]
async fn idle_transaction_timeout_reaps_backend_and_pool_replaces_connection() {
let admin = PgPool::connect(&admin_url().await)
.await
.expect("connect admin");
let (seed_pool, name) = create_scratch_db(&admin, "idle_txn_reap").await;
seed_pool.close().await;
let base = admin_url().await;
let idx = base.rfind('/').expect("db url has a path segment");
let scratch_url = format!("{}/{}", &base[..idx], name);
let db = Db::new(&DbConfig {
database_url: scratch_url,
max_connections: 1,
min_connections: 0,
idle_txn_timeout_ms: 100,
..DbConfig::default()
})
.await
.expect("connect Db with idle transaction timeout");
let mut wedged = db.pool.acquire().await.expect("acquire writer connection");
let old_pid: i32 = sqlx::query_scalar("SELECT pg_backend_pid()")
.fetch_one(&mut *wedged)
.await
.expect("read original backend pid");
sqlx::query("BEGIN")
.execute(&mut *wedged)
.await
.expect("begin transaction");
tokio::time::sleep(std::time::Duration::from_millis(500)).await;
sqlx::query("SELECT 1")
.execute(&mut *wedged)
.await
.expect_err("idle transaction backend must be terminated");
drop(wedged);
let mut replacement =
tokio::time::timeout(std::time::Duration::from_secs(3), db.pool.acquire())
.await
.expect("pool did not replace terminated backend")
.expect("acquire replacement writer connection");
let (new_pid, timeout): (i32, String) = sqlx::query_as(
"SELECT pg_backend_pid(), current_setting('idle_in_transaction_session_timeout')",
)
.fetch_one(&mut *replacement)
.await
.expect("inspect replacement backend");
assert_ne!(new_pid, old_pid, "pool must replace the terminated backend");
assert_eq!(timeout, "100ms", "replacement must reinstall the timeout");
drop(replacement);
drop_scratch_db(&admin, db.pool.clone(), &name).await;
}
/// The armed writer pool (`Db::new`) must enforce the floor end-to-end
/// through the public insert APIs, and the session GUC must be verifiably
/// set on pooled connections.

4
crates/buzz-db/src/store/deletion.rs
View File

@@ -58,6 +58,8 @@ pub const EXPECTED_SCOPED_TABLES: &[&str] = &[
"api_tokens",
"archived_identities",
"audit_log",
"audit_outbox",
"audit_delivery_keys",
"channel_members",
"channels",
"community_bans",
@@ -109,6 +111,8 @@ pub const PURGE_SCOPED_TABLES: &[&str] = &[
"parameterized_event_watermarks",
"git_repo_names",
"archived_identities",
"audit_outbox",
"audit_delivery_keys",
"audit_log",
"community_bans",
"pubkey_allowlist",

66
crates/buzz-relay/src/api/media.rs
View File

@@ -456,32 +456,41 @@ pub async fn upload_blob(
)
.increment(1);
// Audit via bounded channel — same pattern as event audit.
if let Some(audit_tx) = &state.audit_tx {
// Persist the audit intent before returning — same pattern as event audit.
if let Some(audit_queue) = &state.audit_queue {
let desc = descriptor.clone();
if let Err(e) = audit_tx
.send(NewAuditEntry {
community_id: auth.tenant.community(),
action: AuditAction::MediaUploaded,
actor_pubkey: Some(auth.auth_event.pubkey.to_bytes().to_vec()),
object_id: Some(desc.sha256.clone()),
detail: serde_json::json!({
"sha256": desc.sha256,
"size": desc.size,
"mime": desc.mime_type,
}),
})
let dedupe_key = media_audit_dedupe_key(&auth.auth_event);
audit_queue
.send(
NewAuditEntry {
community_id: auth.tenant.community(),
action: AuditAction::MediaUploaded,
actor_pubkey: Some(auth.auth_event.pubkey.to_bytes().to_vec()),
object_id: Some(desc.sha256.clone()),
detail: serde_json::json!({
"sha256": desc.sha256,
"size": desc.size,
"mime": desc.mime_type,
}),
},
Some(&dedupe_key),
)
.await
{
tracing::error!("Media audit channel closed — entry lost: {e}");
metrics::counter!("buzz_audit_send_errors_total").increment(1);
}
.map_err(|e| {
tracing::error!("Media audit intent could not be persisted: {e}");
metrics::counter!("buzz_audit_send_errors_total").increment(1);
MediaError::Internal
})?;
}
serving_write.finish().await.map_err(serving_lease_lost)?;
Ok(Json(descriptor))
}
fn media_audit_dedupe_key(auth_event: &nostr::Event) -> String {
format!("media_uploaded:{}", auth_event.id)
}
pub(crate) fn media_base_url_for_tenant(config_relay_url: &str, tenant_host: &str) -> String {
let scheme = if config_relay_url.trim_start().starts_with("wss://")
|| config_relay_url.trim_start().starts_with("https://")
@@ -1023,6 +1032,27 @@ mod tests {
const VALID_HASH: &str = "abcdef0123456789abcdef0123456789abcdef0123456789abcdef0123456789";
#[test]
fn media_audit_dedupe_key_tracks_the_signed_upload_action() {
let first = EventBuilder::new(Kind::from(24242), "Upload media")
.sign_with_keys(&Keys::generate())
.expect("sign first upload authorization");
let second = EventBuilder::new(Kind::from(24242), "Upload media")
.sign_with_keys(&Keys::generate())
.expect("sign second upload authorization");
assert_eq!(
media_audit_dedupe_key(&first),
media_audit_dedupe_key(&first),
"a retry with the same signed authorization must dedupe",
);
assert_ne!(
media_audit_dedupe_key(&first),
media_audit_dedupe_key(&second),
"distinct signed uploads of identical bytes must remain distinct",
);
}
#[test]
fn serving_write_error_taxonomy_separates_fence_from_backend_failure() {
let fenced = anyhow::Error::from(buzz_db::DbError::AccessDenied("fenced".to_string()));

76
crates/buzz-relay/src/handlers/event.rs
View File

@@ -339,11 +339,11 @@ pub async fn fan_out_pubsub_event(state: &Arc<AppState>, channel_event: buzz_pub
/// Schedule post-commit delivery/side effects for a stored event.
///
/// This intentionally returns after only the bounded audit enqueue has completed:
/// This intentionally returns after only the durable audit enqueue has completed:
/// NIP-01 `OK` means the event was durably accepted, not that Redis publish,
/// local fan-out, or workflow triggering have completed. Keeping audit enqueue on
/// the awaited path preserves the bounded-channel backpressure posture when the
/// audit DB is overloaded; the spawned task still runs the same guarded fan-out
/// the awaited path preserves audit intent before acceptance when the audit DB
/// is overloaded; the spawned task still runs the same guarded fan-out
/// path, Redis publish, `mark_local_event` echo dedupe, and delivery metrics as
/// the former inline path.
pub(crate) async fn dispatch_persistent_event(
@@ -353,17 +353,10 @@ pub(crate) async fn dispatch_persistent_event(
kind_u32: u32,
actor_pubkey_hex: &str,
threaded_visibility: Option<crate::state::ThreadedChannelVisibility>,
) -> usize {
) -> Result<usize, buzz_audit::AuditError> {
enqueue_persistent_event_audit(tenant, state, stored_event, kind_u32, actor_pubkey_hex).await?;
let event_id_hex = stored_event.event.id.to_hex();
enqueue_event_created_audit(
tenant,
state,
stored_event,
kind_u32,
actor_pubkey_hex,
&event_id_hex,
)
.await;
let tenant = tenant.clone();
let state = Arc::clone(state);
@@ -389,7 +382,31 @@ pub(crate) async fn dispatch_persistent_event(
);
});
0
Ok(0)
}
/// Repair only the durable audit intent for an already-persisted event.
///
/// Internal producers use this when an earlier attempt committed the event but
/// failed to enqueue its audit intent. Their duplicate retry must not repeat
/// fan-out or workflows, but it must repair the missing audit handoff.
pub(crate) async fn enqueue_persistent_event_audit(
tenant: &TenantContext,
state: &Arc<AppState>,
stored_event: &StoredEvent,
kind_u32: u32,
actor_pubkey_hex: &str,
) -> Result<(), buzz_audit::AuditError> {
let event_id_hex = stored_event.event.id.to_hex();
enqueue_event_created_audit(
tenant,
state,
stored_event,
kind_u32,
actor_pubkey_hex,
&event_id_hex,
)
.await
}
/// Run post-commit delivery/side effects for a stored event.
@@ -506,7 +523,7 @@ async fn dispatch_persistent_event_inner(
// `search_index_tx` mpsc are gone with the Typesense backend.
if enqueue_audit {
enqueue_event_created_audit(
if let Err(error) = enqueue_event_created_audit(
tenant,
state,
stored_event,
@@ -514,7 +531,10 @@ async fn dispatch_persistent_event_inner(
actor_pubkey_hex,
&event_id_hex,
)
.await;
.await
{
error!(event_id = %event_id_hex, %error, "Pubsub audit intent could not be persisted");
}
}
// Skip workflow triggering for workflow-execution kinds and relay-signed workflow messages.
@@ -567,17 +587,14 @@ async fn enqueue_event_created_audit(
kind_u32: u32,
actor_pubkey_hex: &str,
event_id_hex: &str,
) {
let Some(audit_tx) = &state.audit_tx else {
return;
) -> Result<(), buzz_audit::AuditError> {
let Some(audit_queue) = &state.audit_queue else {
return Ok(());
};
// Audit via bounded channel (capacity 1000). Uses .send().await so entries
// are never silently dropped — backpressure propagates to the event handler
// if the queue is full. This is intentional: the audit advisory lock already
// serializes writes (at most 1 in-flight), so a full queue means the audit
// DB is genuinely overloaded and the relay should slow down rather than
// accumulate unbounded in-memory state. DB write failures in the worker are
// logged but not retried (same as the previous per-event tokio::spawn).
// Persist the audit intent before returning the NIP-01 acceptance. Delivery
// is asynchronous and retries both lock and statement timeouts from the
// durable outbox; a contended community cannot block another community's
// chain, and restart resumes any pending entry.
let audit_entry = buzz_audit::NewAuditEntry {
community_id: tenant.community(),
action: buzz_audit::AuditAction::EventCreated,
@@ -594,10 +611,13 @@ async fn enqueue_event_created_audit(
"channel_id": stored_event.channel_id,
}),
};
if let Err(e) = audit_tx.send(audit_entry).await {
error!(event_id = %event_id_hex, "Audit channel closed — entry lost: {e}");
let dedupe_key = format!("event_created:{event_id_hex}");
if let Err(e) = audit_queue.send(audit_entry, Some(&dedupe_key)).await {
error!(event_id = %event_id_hex, "Audit intent could not be persisted: {e}");
metrics::counter!("buzz_audit_send_errors_total").increment(1);
return Err(e);
}
Ok(())
}
/// Handle an EVENT message from a WebSocket connection.

6
crates/buzz-relay/src/handlers/ingest.rs
View File

@@ -3120,7 +3120,8 @@ async fn ingest_event_inner(
&pubkey_hex,
threaded_visibility.clone(),
)
.await;
.await
.map_err(|e| IngestError::Internal(format!("audit persistence failed: {e}")))?;
info!(event_id = %event_id_hex, kind = kind_u32, "Event ingested via pipeline");
return Ok(IngestResult {
@@ -3263,7 +3264,8 @@ async fn ingest_event_inner(
&pubkey_hex,
threaded_visibility.clone(),
)
.await;
.await
.map_err(|e| IngestError::Internal(format!("audit persistence failed: {e}")))?;
info!(event_id = %event_id_hex, kind = kind_u32, "Event ingested via pipeline");

9
crates/buzz-relay/src/handlers/moderation_notices.rs
View File

@@ -28,7 +28,7 @@ use uuid::Uuid;
use buzz_core::kind::{event_kind_u32, KIND_STREAM_MESSAGE};
use buzz_core::tenant::TenantContext;
use super::event::dispatch_persistent_event;
use super::event::{dispatch_persistent_event, enqueue_persistent_event_audit};
use super::side_effects::emit_group_discovery_events;
use crate::state::AppState;
@@ -180,7 +180,7 @@ pub async fn send_moderation_notice(
.await?;
let kind_u32 = event_kind_u32(&stored.event);
dispatch_persistent_event(tenant, state, &stored, kind_u32, &relay_pubkey_hex, None).await;
dispatch_persistent_event(tenant, state, &stored, kind_u32, &relay_pubkey_hex, None).await?;
Ok(())
}
@@ -212,7 +212,10 @@ async fn publish_moderation_profile(
.await?;
if was_inserted {
let kind_u32 = event_kind_u32(&stored.event);
dispatch_persistent_event(tenant, state, &stored, kind_u32, relay_pubkey_hex, None).await;
dispatch_persistent_event(tenant, state, &stored, kind_u32, relay_pubkey_hex, None).await?;
} else {
let kind_u32 = event_kind_u32(&stored.event);
enqueue_persistent_event_audit(tenant, state, &stored, kind_u32, relay_pubkey_hex).await?;
}
Ok(())
}

69
crates/buzz-relay/src/handlers/side_effects.rs
View File

@@ -16,7 +16,7 @@ use buzz_core::kind::{
use buzz_core::StoredEvent;
use buzz_db::channel::{MemberRecord, MemberRole};
use super::event::dispatch_persistent_event;
use super::event::{dispatch_persistent_event, enqueue_persistent_event_audit};
use crate::protocol::RelayMessage;
use crate::state::AppState;
use buzz_core::tenant::TenantContext;
@@ -1042,7 +1042,10 @@ async fn emit_addressable_discovery_event(
.await?;
if was_inserted {
let kind_u32 = event_kind_u32(&stored.event);
dispatch_persistent_event(tenant, state, &stored, kind_u32, relay_pubkey_hex, None).await;
dispatch_persistent_event(tenant, state, &stored, kind_u32, relay_pubkey_hex, None).await?;
} else {
let kind_u32 = event_kind_u32(&stored.event);
enqueue_persistent_event_audit(tenant, state, &stored, kind_u32, relay_pubkey_hex).await?;
}
Ok(())
}
@@ -1064,7 +1067,7 @@ async fn store_group_members_event(
state: &Arc<AppState>,
channel_id: Uuid,
member_snapshot: &mut buzz_db::channel::LockedMemberSnapshot,
) -> anyhow::Result<Option<buzz_core::StoredEvent>> {
) -> anyhow::Result<(buzz_core::StoredEvent, bool)> {
let group_id = channel_id.to_string();
let tags = group_members_tags(&group_id, &member_snapshot.members)?;
let relay_pubkey = state.relay_keypair.public_key().to_bytes();
@@ -1094,16 +1097,17 @@ async fn store_group_members_event(
let (stored, inserted) = member_snapshot
.replace_member_event(tenant.community(), channel_id, &event)
.await?;
Ok(inserted.then_some(stored))
Ok((stored, inserted))
}
async fn dispatch_group_members_event(
tenant: &TenantContext,
state: &Arc<AppState>,
stored: Option<buzz_core::StoredEvent>,
stored: (buzz_core::StoredEvent, bool),
relay_pubkey_hex: &str,
) {
if let Some(stored) = stored {
) -> anyhow::Result<()> {
let (stored, inserted) = stored;
if inserted {
dispatch_persistent_event(
tenant,
state,
@@ -1112,8 +1116,18 @@ async fn dispatch_group_members_event(
relay_pubkey_hex,
None,
)
.await;
.await?;
} else {
enqueue_persistent_event_audit(
tenant,
state,
&stored,
KIND_NIP29_GROUP_MEMBERS,
relay_pubkey_hex,
)
.await?;
}
Ok(())
}
/// Emit NIP-29 group discovery events (39000, 39001, 39002) signed by the relay keypair.
@@ -1229,7 +1243,7 @@ pub async fn emit_group_discovery_events(
let stored_members =
store_group_members_event(tenant, state, channel_id, &mut member_snapshot).await?;
member_snapshot.release().await?;
dispatch_group_members_event(tenant, state, stored_members, &relay_pubkey_hex).await;
dispatch_group_members_event(tenant, state, stored_members, &relay_pubkey_hex).await?;
Ok(())
}
@@ -3124,7 +3138,16 @@ async fn publish_nip43_membership_list_inner(
&relay_pubkey_hex,
None,
)
.await;
.await?;
} else {
enqueue_persistent_event_audit(
tenant,
state,
&stored,
KIND_NIP43_MEMBERSHIP_LIST,
&relay_pubkey_hex,
)
.await?;
}
info!(member_count, "NIP-43 membership list published");
@@ -3231,7 +3254,7 @@ pub async fn reconcile_large_channel_member_snapshots(
let stored_members =
store_group_members_event(&tenant, state, channel_id, &mut member_snapshot).await?;
member_snapshot.release().await?;
dispatch_group_members_event(&tenant, state, stored_members, &relay_pubkey_hex).await;
dispatch_group_members_event(&tenant, state, stored_members, &relay_pubkey_hex).await?;
Ok::<bool, anyhow::Error>(true)
}
.await;
@@ -3438,6 +3461,14 @@ pub async fn publish_nipia_archival_list(
.replace_addressable_event(tenant.community(), &event, None)
.await?;
if !was_inserted {
enqueue_persistent_event_audit(
tenant,
state,
&stored,
KIND_IA_ARCHIVED_LIST,
&relay_pubkey_hex,
)
.await?;
continue;
}
@@ -3461,7 +3492,7 @@ pub async fn publish_nipia_archival_list(
&relay_pubkey_hex,
None,
)
.await;
.await?;
info!(
archived_count = archived.len(),
"NIP-IA archived identities list published"
@@ -3551,7 +3582,16 @@ pub async fn publish_dm_visibility_snapshot(
&relay_pubkey_hex,
None,
)
.await;
.await?;
} else {
enqueue_persistent_event_audit(
tenant,
state,
&stored,
KIND_DM_VISIBILITY,
&relay_pubkey_hex,
)
.await?;
}
info!(
@@ -3610,10 +3650,11 @@ async fn publish_nipia_delta(
.insert_event(tenant.community(), &event, None)
.await?;
if !was_inserted {
enqueue_persistent_event_audit(tenant, state, &stored, kind, &relay_pubkey_hex).await?;
return Ok(());
}
dispatch_persistent_event(tenant, state, &stored, kind, &relay_pubkey_hex, None).await;
dispatch_persistent_event(tenant, state, &stored, kind, &relay_pubkey_hex, None).await?;
info!(
target = %target_pubkey_hex,

7
crates/buzz-relay/src/main.rs
View File

@@ -1152,9 +1152,10 @@ async fn main() -> anyhow::Result<()> {
serve(router, health_router, Arc::clone(&state)).await?;
state.community_revalidator_cancel.cancel();
// Signal the audit worker to stop accepting, flush buffered entries, and
// exit. Uses a CancellationToken so it works regardless of how many
// Arc<AppState> clones are still alive in background tasks.
// Signal the audit worker to deliver ready entries and exit. Deferred
// entries remain in the durable outbox for another worker or restart. Uses
// a CancellationToken so it works regardless of how many Arc<AppState>
// clones are still alive in background tasks.
audit_shutdown
.drain(std::time::Duration::from_secs(5))
.await;

806
crates/buzz-relay/src/state.rs
View File

@@ -10,7 +10,7 @@ use axum::body::Bytes;
use axum::extract::ws::{Message as WsMessage, Utf8Bytes as WsUtf8Bytes};
use dashmap::DashMap;
use futures_util::future::join_all;
use tokio::sync::{mpsc, watch, Semaphore};
use tokio::sync::{mpsc, watch, Notify, Semaphore};
use tokio::task::JoinHandle;
use tokio_util::sync::CancellationToken;
use uuid::Uuid;
@@ -693,8 +693,8 @@ pub struct AppState {
/// access check so open channels stay zero-cost. Invalidated on a flip.
pub channel_visibility_cache: Arc<moka::sync::Cache<(CommunityId, Uuid), String>>,
/// Bounded channel for audit logging, absent when audit logging is disabled.
pub audit_tx: Option<mpsc::Sender<buzz_audit::NewAuditEntry>>,
/// Durable audit outbox producer, absent when audit logging is disabled.
pub audit_queue: Option<AuditQueue>,
/// Media storage client (S3/MinIO).
pub media_storage: Arc<MediaStorage>,
/// Single-flight + cache state for the hourly S3 storage sweep. See
@@ -776,8 +776,9 @@ impl AppState {
/// Constructs `AppState` from its component services.
///
/// Returns `(state, audit_shutdown)`. The caller should call
/// `audit_shutdown.drain().await` during graceful shutdown so queued
/// audit entries are flushed before the process exits.
/// `audit_shutdown.drain().await` during graceful shutdown so ready audit
/// entries are delivered; deferred entries already live durably in the
/// outbox and resume on another worker or after restart.
#[allow(clippy::too_many_arguments)]
pub fn new(
config: Config,
@@ -796,43 +797,20 @@ impl AppState {
let search_arc = Arc::new(search);
let audit_arc = audit.into().map(Arc::new);
let (audit_tx, mut audit_rx) = mpsc::channel::<buzz_audit::NewAuditEntry>(1000);
let audit_for_worker = audit_arc.clone();
let audit_cancel = CancellationToken::new();
let audit_cancel_worker = audit_cancel.clone();
let audit_worker_handle = tokio::spawn(async move {
let Some(audit_for_worker) = audit_for_worker else {
audit_cancel_worker.cancelled().await;
return;
};
// Normal operation: process entries as they arrive.
loop {
tokio::select! {
entry = audit_rx.recv() => {
match entry {
Some(entry) => log_audit_entry(&audit_for_worker, entry).await,
None => break, // 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;
}
}
let (audit_queue, audit_shutdown) = match audit_arc.clone() {
Some(audit) => {
let (queue, shutdown) = start_audit_worker(audit);
(Some(queue), shutdown)
}
// Drain: recv() returns buffered entries, then None once empty.
let mut drained = 0u32;
while let Some(entry) = audit_rx.recv().await {
log_audit_entry(&audit_for_worker, entry).await;
drained += 1;
None => {
let cancel = CancellationToken::new();
let cancel_worker = cancel.clone();
let handle = tokio::spawn(async move {
cancel_worker.cancelled().await;
});
(None, AuditShutdownHandle { cancel, handle })
}
if drained > 0 {
tracing::info!(drained, "audit worker flushed remaining entries");
}
tracing::warn!("audit log worker exited (expected on shutdown)");
});
};
let git_max_concurrent_ops = config.git_max_concurrent_ops;
let media_max_concurrent_uploads = config.media_max_concurrent_uploads;
@@ -857,7 +835,6 @@ impl AppState {
Arc::new(RedisNip98ReplayGuard::new(redis_pool.clone()));
let gif_http_client = crate::api::gifs::build_gif_http_client();
let admission_rate_limiter = Arc::new(RedisRateLimiter::new(redis_pool.clone()));
let audit_enabled = audit_arc.is_some();
let state = Self {
config: Arc::new(config),
db,
@@ -905,7 +882,7 @@ impl AppState {
.support_invalidation_closures()
.build(),
),
audit_tx: audit_enabled.then_some(audit_tx),
audit_queue,
media_storage: Arc::new(media_storage),
storage_sweep: Arc::new(tokio::sync::Mutex::new(
crate::storage_sweep::StorageSweepState::default(),
@@ -946,13 +923,7 @@ impl AppState {
tracer: Arc::new(crate::conformance::NoopTracer),
mesh: Arc::new(std::sync::OnceLock::new()),
};
(
state,
AuditShutdownHandle {
cancel: audit_cancel,
handle: audit_worker_handle,
},
)
(state, audit_shutdown)
}
/// Inter-relay mesh handle. `None` ⇒ mesh-off / single-instance: callers
@@ -1319,61 +1290,133 @@ pub struct ThreadedChannelVisibility {
/// Handle for graceful audit worker shutdown.
///
/// Signals the worker to stop accepting new entries, drain its buffer,
/// and exit. Independent of `Arc<AppState>` lifetime — works even when
/// background tasks (reaper, pubsub, health) still hold state clones.
/// Signals the worker to finish its current delivery pass and exit. Pending
/// entries remain durable in PostgreSQL. Independent of `Arc<AppState>`
/// lifetime — works even when background tasks (reaper, pubsub, health) still
/// hold state clones.
pub struct AuditShutdownHandle {
cancel: CancellationToken,
handle: JoinHandle<()>,
}
impl AuditShutdownHandle {
/// Signal the audit worker to drain and wait up to `timeout` for it to finish.
/// Stop the audit worker and wait up to `timeout` for it to finish.
///
/// Producers persist entries before waking the worker, so shutdown does not
/// wait indefinitely on a contended community and pending rows resume after
/// restart.
pub async fn drain(self, timeout: std::time::Duration) {
self.cancel.cancel();
match tokio::time::timeout(timeout, self.handle).await {
let mut handle = self.handle;
match tokio::time::timeout(timeout, &mut handle).await {
Ok(Ok(())) => tracing::info!("Audit worker drained cleanly"),
Ok(Err(e)) => tracing::error!("Audit worker panicked: {e}"),
Err(_) => tracing::error!(
?timeout,
"Audit worker did not drain in time — exiting anyway"
),
Err(_) => {
tracing::error!(?timeout, "Audit worker did not drain in time — aborting it");
handle.abort();
if let Err(error) = handle.await {
if !error.is_cancelled() {
tracing::error!(%error, "Audit worker failed while being aborted");
}
}
}
}
}
}
/// 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();
let mut retry_delay_ms = 50u64;
let mut retries = 0u64;
loop {
match audit.log(entry.clone()).await {
Ok(_) => {
metrics::histogram!("buzz_audit_log_seconds").record(t.elapsed().as_secs_f64());
return;
/// Durable audit producer backed by the shared PostgreSQL outbox.
#[derive(Clone)]
pub struct AuditQueue {
audit: Arc<AuditService>,
wake: Arc<Notify>,
}
impl AuditQueue {
/// Persist one entry and wake a worker after the transaction commits.
pub async fn send(
&self,
entry: buzz_audit::NewAuditEntry,
dedupe_key: Option<&str>,
) -> Result<(), buzz_audit::AuditError> {
self.audit.enqueue(entry, dedupe_key).await?;
self.wake.notify_one();
Ok(())
}
}
fn start_audit_worker(audit: Arc<AuditService>) -> (AuditQueue, AuditShutdownHandle) {
let wake = Arc::new(Notify::new());
let queue = AuditQueue {
audit: Arc::clone(&audit),
wake: Arc::clone(&wake),
};
let cancel = CancellationToken::new();
let cancel_worker = cancel.clone();
let handle = tokio::spawn(async move {
let mut final_delivery_attempted = false;
loop {
// Allow one final delivery attempt after cancellation so a producer
// that just committed an intent can still observe graceful delivery.
// The second check bounds teardown even while the outbox stays busy.
if cancel_worker.is_cancelled() {
if final_delivery_attempted {
break;
}
final_delivery_attempted = true;
}
Err(buzz_audit::AuditError::Database(sqlx::Error::Database(database_error)))
if database_error.code().as_deref() == Some("55P03") =>
{
retries += 1;
metrics::counter!("buzz_audit_log_lock_retries_total").increment(1);
tracing::warn!(
retries,
let started = std::time::Instant::now();
match audit.deliver_next_pending().await {
Ok(buzz_audit::PendingAuditResult::Appended) => {
metrics::histogram!("buzz_audit_log_seconds")
.record(started.elapsed().as_secs_f64());
}
Ok(buzz_audit::PendingAuditResult::Deferred {
attempt_count,
retry_delay_ms,
"Audit advisory lock timed out; preserving entry for retry"
);
tokio::time::sleep(std::time::Duration::from_millis(retry_delay_ms)).await;
retry_delay_ms = (retry_delay_ms * 2).min(1_000);
}
Err(error) => {
metrics::counter!("buzz_audit_log_errors_total").increment(1);
tracing::error!("Audit log failed: {error}");
return;
sqlstate,
}) => {
metrics::counter!(
"buzz_audit_log_retries_total",
"sqlstate" => sqlstate.clone()
)
.increment(1);
if sqlstate == "55P03" {
metrics::counter!("buzz_audit_log_lock_retries_total").increment(1);
}
tracing::warn!(
attempt_count,
retry_delay_ms,
%sqlstate,
"Audit delivery timed out; durable entry rescheduled"
);
}
Ok(buzz_audit::PendingAuditResult::Idle) => {
if cancel_worker.is_cancelled() {
break;
}
tokio::select! {
_ = cancel_worker.cancelled() => {},
_ = wake.notified() => {},
_ = tokio::time::sleep(std::time::Duration::from_millis(100)) => {},
}
}
Err(error) => {
metrics::counter!("buzz_audit_log_errors_total").increment(1);
tracing::error!(%error, "Audit outbox delivery failed; entry remains durable");
if cancel_worker.is_cancelled() {
break;
}
tokio::select! {
_ = cancel_worker.cancelled() => break,
_ = wake.notified() => {},
_ = tokio::time::sleep(std::time::Duration::from_secs(1)) => {},
}
}
}
}
}
tracing::warn!("audit log worker exited (expected on shutdown)");
});
(queue, AuditShutdownHandle { cancel, handle })
}
impl std::fmt::Debug for AppState {
@@ -1472,26 +1515,16 @@ pub(crate) mod tests {
let observer = sqlx::PgPool::connect(&database_url)
.await
.expect("connect observer pool");
let application_name = format!("audit-retry-test-{}", Uuid::new_v4());
let hook_application_name = application_name.clone();
let audit_pool = sqlx::postgres::PgPoolOptions::new()
.max_connections(1)
.after_connect(move |conn, _meta| {
let application_name = hook_application_name.clone();
Box::pin(async move {
sqlx::query(
"SELECT set_config('application_name', $1, false), \
set_config('lock_timeout', '100', false)",
)
.bind(application_name)
.execute(&mut *conn)
.await?;
Ok(())
})
})
.connect(&database_url)
.await
.expect("connect audit pool");
let audit_pool = buzz_db::Db::connect_writer_pool(&buzz_db::DbConfig {
database_url: database_url.clone(),
max_connections: 1,
min_connections: 0,
lock_timeout_ms: 100,
statement_timeout_ms: 0,
..buzz_db::DbConfig::default()
})
.await
.expect("connect audit pool");
let community_id = Uuid::new_v4();
sqlx::query("INSERT INTO communities (id, host) VALUES ($1, $2)")
@@ -1519,37 +1552,28 @@ pub(crate) mod tests {
.expect("hold community audit lock");
let audit = Arc::new(AuditService::new(audit_pool));
let worker = tokio::spawn({
let audit = Arc::clone(&audit);
async move { log_audit_entry(&audit, entry).await }
});
let (queue, shutdown) = start_audit_worker(Arc::clone(&audit));
queue
.send(entry, None)
.await
.expect("durably enqueue audit entry");
// Observe one timed-out advisory-lock attempt and then a second wait.
// Releasing during the first wait would not prove that the worker
// preserved and retried the original queue entry.
// Observe the durable row reach a second attempt. Releasing during the
// first wait would not prove that the worker preserved and retried it.
tokio::time::timeout(std::time::Duration::from_secs(3), async {
let mut saw_first_wait = false;
let mut saw_retry_gap = false;
loop {
let waiting: bool = sqlx::query_scalar(
"SELECT EXISTS (\
SELECT 1 FROM pg_stat_activity \
WHERE application_name = $1 \
AND query LIKE 'SELECT pg_advisory_lock%' \
AND wait_event = 'advisory'\
)",
let attempts: i32 = sqlx::query_scalar(
"SELECT COALESCE(MAX(attempt_count), 0)::integer \
FROM audit_outbox \
WHERE community_id = $1 AND object_id = $2",
)
.bind(&application_name)
.bind(community_id)
.bind(&object_id)
.fetch_one(&observer)
.await
.expect("inspect audit lock waiter");
if waiting {
if saw_retry_gap {
break;
}
saw_first_wait = true;
} else if saw_first_wait {
saw_retry_gap = true;
.expect("inspect durable retry count");
if attempts >= 2 {
break;
}
tokio::time::sleep(std::time::Duration::from_millis(5)).await;
}
@@ -1562,10 +1586,25 @@ pub(crate) mod tests {
.execute(&mut *holder)
.await
.expect("release community audit lock");
tokio::time::timeout(std::time::Duration::from_secs(3), worker)
.await
.expect("audit worker did not finish after lock release")
.expect("audit worker task panicked");
tokio::time::timeout(std::time::Duration::from_secs(3), async {
loop {
let rows: i64 = sqlx::query_scalar(
"SELECT count(*) FROM audit_log WHERE community_id = $1 AND object_id = $2",
)
.bind(community_id)
.bind(&object_id)
.fetch_one(&observer)
.await
.expect("count retried audit rows");
if rows == 1 {
break;
}
tokio::time::sleep(std::time::Duration::from_millis(5)).await;
}
})
.await
.expect("audit worker did not finish after lock release");
shutdown.drain(std::time::Duration::from_secs(1)).await;
let rows: i64 = sqlx::query_scalar(
"SELECT count(*) FROM audit_log WHERE community_id = $1 AND object_id = $2",
@@ -1590,6 +1629,503 @@ pub(crate) mod tests {
.expect("remove test community");
}
#[tokio::test]
#[ignore = "requires Postgres"]
async fn audit_worker_retries_statement_timeout_until_original_entry_is_appended_once() {
let database_url = std::env::var("DATABASE_URL").expect("DATABASE_URL");
let observer = sqlx::PgPool::connect(&database_url)
.await
.expect("connect observer pool");
let audit_pool = buzz_db::Db::connect_writer_pool(&buzz_db::DbConfig {
database_url: database_url.clone(),
max_connections: 1,
min_connections: 0,
lock_timeout_ms: 1_000,
statement_timeout_ms: 100,
..buzz_db::DbConfig::default()
})
.await
.expect("connect audit pool");
let community_id = Uuid::new_v4();
sqlx::query("INSERT INTO communities (id, host) VALUES ($1, $2)")
.bind(community_id)
.bind(format!("audit-statement-retry-{community_id}.example"))
.execute(&observer)
.await
.expect("insert test community");
let object_id = format!("audit-statement-retry-object-{}", Uuid::new_v4());
let entry = buzz_audit::NewAuditEntry {
community_id: CommunityId::from_uuid(community_id),
action: buzz_audit::AuditAction::EventCreated,
actor_pubkey: Some(vec![0xcd; 32]),
object_id: Some(object_id.clone()),
detail: serde_json::json!({"test": "statement-timeout-retry"}),
};
let lock_key = format!("buzz_audit:{community_id}");
let mut holder = observer.acquire().await.expect("acquire lock holder");
sqlx::query("SELECT pg_advisory_lock(hashtextextended($1, 0))")
.bind(&lock_key)
.execute(&mut *holder)
.await
.expect("hold community audit lock");
let audit = Arc::new(AuditService::new(audit_pool));
let (queue, shutdown) = start_audit_worker(Arc::clone(&audit));
queue
.send(entry, None)
.await
.expect("durably enqueue audit entry");
tokio::time::timeout(std::time::Duration::from_secs(3), async {
loop {
let attempts: i32 = sqlx::query_scalar(
"SELECT COALESCE(MAX(attempt_count), 0)::integer \
FROM audit_outbox \
WHERE community_id = $1 AND object_id = $2",
)
.bind(community_id)
.bind(&object_id)
.fetch_one(&observer)
.await
.expect("inspect durable statement-timeout retry count");
if attempts >= 2 {
break;
}
tokio::time::sleep(std::time::Duration::from_millis(5)).await;
}
})
.await
.expect("audit worker did not exercise statement-timeout retry");
sqlx::query("SELECT pg_advisory_unlock(hashtextextended($1, 0))")
.bind(&lock_key)
.execute(&mut *holder)
.await
.expect("release community audit lock");
tokio::time::timeout(std::time::Duration::from_secs(3), async {
loop {
let rows: i64 = sqlx::query_scalar(
"SELECT count(*) FROM audit_log WHERE community_id = $1 AND object_id = $2",
)
.bind(community_id)
.bind(&object_id)
.fetch_one(&observer)
.await
.expect("count retried audit rows");
if rows == 1 {
break;
}
tokio::time::sleep(std::time::Duration::from_millis(5)).await;
}
})
.await
.expect("audit worker did not finish after statement timeout");
shutdown.drain(std::time::Duration::from_secs(1)).await;
let rows: i64 = sqlx::query_scalar(
"SELECT count(*) FROM audit_log WHERE community_id = $1 AND object_id = $2",
)
.bind(community_id)
.bind(&object_id)
.fetch_one(&observer)
.await
.expect("count retried audit rows");
assert_eq!(rows, 1, "the preserved entry must be appended exactly once");
sqlx::query("DELETE FROM audit_log WHERE community_id = $1 AND object_id = $2")
.bind(community_id)
.bind(&object_id)
.execute(&observer)
.await
.expect("remove test audit row");
sqlx::query("DELETE FROM communities WHERE id = $1")
.bind(community_id)
.execute(&observer)
.await
.expect("remove test community");
}
#[tokio::test]
#[ignore = "requires Postgres"]
async fn audit_queue_dedupes_logical_event_while_pending_and_after_delivery() {
let database_url = std::env::var("DATABASE_URL").expect("DATABASE_URL");
let observer = sqlx::PgPool::connect(&database_url)
.await
.expect("connect observer pool");
let audit_pool = buzz_db::Db::connect_writer_pool(&buzz_db::DbConfig {
database_url,
max_connections: 2,
min_connections: 0,
lock_timeout_ms: 100,
statement_timeout_ms: 0,
..buzz_db::DbConfig::default()
})
.await
.expect("connect audit pool");
let community_id = Uuid::new_v4();
sqlx::query("INSERT INTO communities (id, host) VALUES ($1, $2)")
.bind(community_id)
.bind(format!("audit-dedupe-{community_id}.example"))
.execute(&observer)
.await
.expect("insert test community");
let object_id = format!("audit-dedupe-object-{}", Uuid::new_v4());
let dedupe_key = format!("event_created:{object_id}");
let entry = buzz_audit::NewAuditEntry {
community_id: CommunityId::from_uuid(community_id),
action: buzz_audit::AuditAction::EventCreated,
actor_pubkey: Some(vec![0xde; 32]),
object_id: Some(object_id.clone()),
detail: serde_json::json!({"test": "logical-dedupe"}),
};
let audit = Arc::new(AuditService::new(audit_pool));
let (queue, shutdown) = start_audit_worker(audit);
queue
.send(entry.clone(), Some(&dedupe_key))
.await
.expect("enqueue original audit intent");
queue
.send(entry.clone(), Some(&dedupe_key))
.await
.expect("dedupe pending audit retry");
tokio::time::timeout(std::time::Duration::from_secs(3), async {
loop {
let rows: i64 = sqlx::query_scalar(
"SELECT count(*) FROM audit_log WHERE community_id = $1 AND object_id = $2",
)
.bind(community_id)
.bind(&object_id)
.fetch_one(&observer)
.await
.expect("count delivered audit rows");
if rows == 1 {
break;
}
tokio::time::sleep(std::time::Duration::from_millis(5)).await;
}
})
.await
.expect("original audit intent was not delivered");
queue
.send(entry, Some(&dedupe_key))
.await
.expect("dedupe delivered audit retry");
tokio::time::sleep(std::time::Duration::from_millis(100)).await;
let (audit_rows, pending_rows, key_rows): (i64, i64, i64) = sqlx::query_as(
"SELECT \
(SELECT count(*) FROM audit_log WHERE community_id = $1 AND object_id = $2), \
(SELECT count(*) FROM audit_outbox WHERE community_id = $1), \
(SELECT count(*) FROM audit_delivery_keys \
WHERE community_id = $1 AND dedupe_key = $3)",
)
.bind(community_id)
.bind(&object_id)
.bind(&dedupe_key)
.fetch_one(&observer)
.await
.expect("inspect logical audit dedupe state");
assert_eq!(audit_rows, 1, "logical event must be audited exactly once");
assert_eq!(
pending_rows, 0,
"duplicate retry must not recreate outbox row"
);
assert_eq!(key_rows, 1, "delivery key must survive completed delivery");
shutdown.drain(std::time::Duration::from_secs(1)).await;
sqlx::query("DELETE FROM audit_log WHERE community_id = $1")
.bind(community_id)
.execute(&observer)
.await
.expect("remove audit rows");
sqlx::query("DELETE FROM audit_delivery_keys WHERE community_id = $1")
.bind(community_id)
.execute(&observer)
.await
.expect("remove audit delivery key");
sqlx::query("DELETE FROM communities WHERE id = $1")
.bind(community_id)
.execute(&observer)
.await
.expect("remove test community");
}
#[tokio::test]
#[ignore = "requires Postgres"]
async fn durable_audit_queue_survives_saturation_and_contended_tenant_teardown() {
let database_url = std::env::var("DATABASE_URL").expect("DATABASE_URL");
let observer = sqlx::PgPool::connect(&database_url)
.await
.expect("connect observer pool");
let audit_pool = buzz_db::Db::connect_writer_pool(&buzz_db::DbConfig {
database_url: database_url.clone(),
max_connections: 4,
min_connections: 0,
lock_timeout_ms: 100,
statement_timeout_ms: 0,
..buzz_db::DbConfig::default()
})
.await
.expect("connect audit pool");
let community_a = Uuid::new_v4();
let community_b = Uuid::new_v4();
for (id, label) in [(community_a, "a"), (community_b, "b")] {
sqlx::query("INSERT INTO communities (id, host) VALUES ($1, $2)")
.bind(id)
.bind(format!("audit-fairness-{label}-{id}.example"))
.execute(&observer)
.await
.expect("insert test community");
}
let lock_key_a = format!("buzz_audit:{community_a}");
let mut holder = observer.acquire().await.expect("acquire lock holder");
sqlx::query("SELECT pg_advisory_lock(hashtextextended($1, 0))")
.bind(&lock_key_a)
.execute(&mut *holder)
.await
.expect("hold community A audit lock");
let audit = Arc::new(AuditService::new(audit_pool));
let (queue, shutdown) = start_audit_worker(Arc::clone(&audit));
for index in 0..=1000 {
queue
.send(
buzz_audit::NewAuditEntry {
community_id: CommunityId::from_uuid(community_a),
action: buzz_audit::AuditAction::EventCreated,
actor_pubkey: Some(vec![0xaa; 32]),
object_id: Some(format!("audit-saturated-a-{index}")),
detail: serde_json::json!({"index": index}),
},
None,
)
.await
.expect("durably enqueue community A audit entry");
}
let object_b = format!("audit-progress-b-{}", Uuid::new_v4());
queue
.send(
buzz_audit::NewAuditEntry {
community_id: CommunityId::from_uuid(community_b),
action: buzz_audit::AuditAction::EventCreated,
actor_pubkey: Some(vec![0xbb; 32]),
object_id: Some(object_b.clone()),
detail: serde_json::json!({"test": "cross-community-progress"}),
},
None,
)
.await
.expect("durably enqueue community B audit entry");
tokio::time::timeout(std::time::Duration::from_secs(3), async {
loop {
let rows: i64 = sqlx::query_scalar(
"SELECT count(*) FROM audit_log WHERE community_id = $1 AND object_id = $2",
)
.bind(community_b)
.bind(&object_b)
.fetch_one(&observer)
.await
.expect("count community B audit rows");
if rows == 1 {
break;
}
tokio::time::sleep(std::time::Duration::from_millis(10)).await;
}
})
.await
.expect("contended community A blocked community B progress");
let pending_a: i64 =
sqlx::query_scalar("SELECT count(*) FROM audit_outbox WHERE community_id = $1")
.bind(community_a)
.fetch_one(&observer)
.await
.expect("count durable community A audit intents");
assert_eq!(pending_a, 1001, "saturated entries must remain durable");
let shutdown_started = std::time::Instant::now();
shutdown.drain(std::time::Duration::from_secs(1)).await;
assert!(
shutdown_started.elapsed() < std::time::Duration::from_secs(1),
"durable queue teardown must not wait on a contended tenant"
);
sqlx::query("SELECT pg_advisory_unlock(hashtextextended($1, 0))")
.bind(&lock_key_a)
.execute(&mut *holder)
.await
.expect("release community A audit lock");
let (_queue_after_restart, shutdown_after_restart) = start_audit_worker(Arc::clone(&audit));
tokio::time::timeout(std::time::Duration::from_secs(3), async {
loop {
let rows: i64 = sqlx::query_scalar(
"SELECT count(*) FROM audit_log \
WHERE community_id = $1 AND object_id = 'audit-saturated-a-0'",
)
.bind(community_a)
.fetch_one(&observer)
.await
.expect("count recovered community A audit row");
if rows == 1 {
break;
}
tokio::time::sleep(std::time::Duration::from_millis(10)).await;
}
})
.await
.expect("restart did not resume the durable community A intent");
shutdown_after_restart
.drain(std::time::Duration::from_secs(1))
.await;
sqlx::query("DELETE FROM audit_outbox WHERE community_id IN ($1, $2)")
.bind(community_a)
.bind(community_b)
.execute(&observer)
.await
.expect("remove pending audit test rows");
sqlx::query("DELETE FROM audit_log WHERE community_id IN ($1, $2)")
.bind(community_a)
.bind(community_b)
.execute(&observer)
.await
.expect("remove audit test rows");
sqlx::query("DELETE FROM communities WHERE id IN ($1, $2)")
.bind(community_a)
.bind(community_b)
.execute(&observer)
.await
.expect("remove test communities");
}
#[tokio::test]
#[ignore = "requires Postgres"]
async fn audit_worker_cancellation_stops_a_continuously_ready_backlog() {
let database_url = std::env::var("DATABASE_URL").expect("DATABASE_URL");
let observer = sqlx::PgPool::connect(&database_url)
.await
.expect("connect observer pool");
let audit_pool = buzz_db::Db::connect_writer_pool(&buzz_db::DbConfig {
database_url,
max_connections: 4,
min_connections: 0,
lock_timeout_ms: 100,
statement_timeout_ms: 0,
..buzz_db::DbConfig::default()
})
.await
.expect("connect audit pool");
let community_id = Uuid::new_v4();
sqlx::query("INSERT INTO communities (id, host) VALUES ($1, $2)")
.bind(community_id)
.bind(format!("audit-ready-backlog-{community_id}.example"))
.execute(&observer)
.await
.expect("insert test community");
sqlx::query(
"INSERT INTO audit_outbox (community_id, action, object_id, detail) \
SELECT $1, 'event_created', 'audit-ready-' || item::text, \
jsonb_build_object('item', item) \
FROM generate_series(1, 10000) AS item",
)
.bind(community_id)
.execute(&observer)
.await
.expect("seed continuously ready audit backlog");
let audit = Arc::new(AuditService::new(audit_pool));
let (_queue, shutdown) = start_audit_worker(audit);
tokio::time::timeout(std::time::Duration::from_secs(3), async {
loop {
let appended: i64 =
sqlx::query_scalar("SELECT count(*) FROM audit_log WHERE community_id = $1")
.bind(community_id)
.fetch_one(&observer)
.await
.expect("count appended audit rows");
if appended > 0 {
break;
}
tokio::time::sleep(std::time::Duration::from_millis(5)).await;
}
})
.await
.expect("audit worker did not begin draining ready backlog");
shutdown.drain(std::time::Duration::from_secs(1)).await;
let appended_after_shutdown: i64 =
sqlx::query_scalar("SELECT count(*) FROM audit_log WHERE community_id = $1")
.bind(community_id)
.fetch_one(&observer)
.await
.expect("count rows after shutdown");
let pending_after_shutdown: i64 =
sqlx::query_scalar("SELECT count(*) FROM audit_outbox WHERE community_id = $1")
.bind(community_id)
.fetch_one(&observer)
.await
.expect("count durable rows after shutdown");
assert!(
pending_after_shutdown > 0,
"shutdown must not drain the entire continuously ready backlog"
);
tokio::time::sleep(std::time::Duration::from_millis(100)).await;
let appended_later: i64 =
sqlx::query_scalar("SELECT count(*) FROM audit_log WHERE community_id = $1")
.bind(community_id)
.fetch_one(&observer)
.await
.expect("count rows after worker should have stopped");
assert_eq!(
appended_later, appended_after_shutdown,
"audit worker must not remain detached after shutdown"
);
sqlx::query("DELETE FROM audit_outbox WHERE community_id = $1")
.bind(community_id)
.execute(&observer)
.await
.expect("remove pending audit rows");
sqlx::query("DELETE FROM audit_log WHERE community_id = $1")
.bind(community_id)
.execute(&observer)
.await
.expect("remove appended audit rows");
sqlx::query("DELETE FROM communities WHERE id = $1")
.bind(community_id)
.execute(&observer)
.await
.expect("remove test community");
}
#[tokio::test]
async fn audit_shutdown_aborts_a_worker_that_exceeds_its_timeout() {
let completed = Arc::new(AtomicBool::new(false));
let completed_worker = Arc::clone(&completed);
let handle = tokio::spawn(async move {
tokio::time::sleep(std::time::Duration::from_millis(100)).await;
completed_worker.store(true, Ordering::SeqCst);
});
let shutdown = AuditShutdownHandle {
cancel: CancellationToken::new(),
handle,
};
shutdown.drain(std::time::Duration::from_millis(5)).await;
tokio::time::sleep(std::time::Duration::from_millis(125)).await;
assert!(
!completed.load(Ordering::SeqCst),
"timed-out worker must be aborted instead of detached"
);
}
#[test]
fn send_to_resets_grace_counter_on_success() {
let (mgr, id, _rx, _ctrl_rx, _cancel, bp) = setup_conn(16);

19
crates/buzz-relay/src/workflow_sink.rs
View File

@@ -16,7 +16,7 @@ use nostr::{EventBuilder, Kind, Tag};
use tracing::info;
use uuid::Uuid;
use crate::handlers::event::dispatch_persistent_event;
use crate::handlers::event::{dispatch_persistent_event, enqueue_persistent_event_audit};
use crate::state::AppState;
/// Resolves `@Name` mentions in workflow message text to the pubkeys of the
@@ -445,7 +445,7 @@ impl ActionSink for RelayActionSink {
// 5. Post-persist side effects (fan-out, search, audit)
// Only if actually inserted (idempotency guard).
if was_inserted {
let _ = dispatch_persistent_event(
dispatch_persistent_event(
&tenant,
&state,
&stored_event,
@@ -453,8 +453,21 @@ impl ActionSink for RelayActionSink {
&author_pubkey_hex,
None,
)
.await;
.await
.map_err(|e| ActionSinkError::Database(e.to_string()))?;
} else {
enqueue_persistent_event_audit(
&tenant,
&state,
&stored_event,
kind_u32,
&author_pubkey_hex,
)
.await
.map_err(|e| ActionSinkError::Database(e.to_string()))?;
}
if was_inserted {
// A threaded reply changed its thread's counters — push a fresh
// relay-signed kind:39005 so subscribed clients update badge
// counts without refetching the head window, exactly as the

37
migrations/0041_durable_audit_outbox.sql Normal file
View File

@@ -0,0 +1,37 @@
-- Durable, fair handoff between accepted relay operations and the per-community
-- audit hash chain. Producers commit an intent here before replying; workers
-- claim only the oldest due row in each community, so one contended chain does
-- not block unrelated tenants. The worker appends audit_log and deletes this
-- row in one transaction.
-- Logical producer keys survive outbox delivery. This makes retry repair
-- idempotent both while an intent is pending and after it has been appended.
CREATE TABLE audit_delivery_keys (
community_id UUID NOT NULL REFERENCES communities(id),
dedupe_key TEXT NOT NULL,
created_at TIMESTAMPTZ NOT NULL DEFAULT clock_timestamp(),
PRIMARY KEY (community_id, dedupe_key)
);
CREATE TABLE audit_outbox (
community_id UUID NOT NULL REFERENCES communities(id),
id UUID NOT NULL DEFAULT gen_random_uuid(),
enqueue_seq BIGINT GENERATED ALWAYS AS IDENTITY,
action VARCHAR(64) NOT NULL,
actor_pubkey BYTEA,
object_id TEXT,
detail JSONB NOT NULL,
enqueued_at TIMESTAMPTZ NOT NULL DEFAULT clock_timestamp(),
next_attempt_at TIMESTAMPTZ NOT NULL DEFAULT clock_timestamp(),
attempt_count INTEGER NOT NULL DEFAULT 0 CHECK (attempt_count >= 0),
PRIMARY KEY (community_id, id)
);
CREATE INDEX idx_audit_outbox_due
ON audit_outbox (next_attempt_at, enqueue_seq);
CREATE INDEX idx_audit_outbox_community_order
ON audit_outbox (community_id, enqueue_seq);
SELECT attach_community_write_fence('audit_outbox');
SELECT attach_community_write_fence('audit_delivery_keys');

28
schema/schema.sql
View File

@@ -657,6 +657,32 @@ CREATE TABLE audit_log (
CREATE UNIQUE INDEX idx_audit_log_hash ON audit_log (community_id, hash);
CREATE TABLE audit_delivery_keys (
community_id UUID NOT NULL REFERENCES communities(id),
dedupe_key TEXT NOT NULL,
created_at TIMESTAMPTZ NOT NULL DEFAULT clock_timestamp(),
PRIMARY KEY (community_id, dedupe_key)
);
CREATE TABLE audit_outbox (
community_id UUID NOT NULL REFERENCES communities(id),
id UUID NOT NULL DEFAULT gen_random_uuid(),
enqueue_seq BIGINT GENERATED ALWAYS AS IDENTITY,
action VARCHAR(64) NOT NULL,
actor_pubkey BYTEA,
object_id TEXT,
detail JSONB NOT NULL,
enqueued_at TIMESTAMPTZ NOT NULL DEFAULT clock_timestamp(),
next_attempt_at TIMESTAMPTZ NOT NULL DEFAULT clock_timestamp(),
attempt_count INTEGER NOT NULL DEFAULT 0 CHECK (attempt_count >= 0),
PRIMARY KEY (community_id, id)
);
CREATE INDEX idx_audit_outbox_due
ON audit_outbox (next_attempt_at, enqueue_seq);
CREATE INDEX idx_audit_outbox_community_order
ON audit_outbox (community_id, enqueue_seq);
-- ── NIP-56 reports (kind:1984 ingest) ─────────────────────────────────────────
-- One row per accepted report event. Reports are signals, never triggers:
-- nothing auto-actions on them (NIP-56). Reporter identity is visible to
@@ -1731,6 +1757,8 @@ $$;
SELECT attach_community_write_fence('api_tokens');
SELECT attach_community_write_fence('archived_identities');
SELECT attach_community_write_fence('audit_log');
SELECT attach_community_write_fence('audit_outbox');
SELECT attach_community_write_fence('audit_delivery_keys');
SELECT attach_community_write_fence('channel_members');
SELECT attach_community_write_fence('channels');
SELECT attach_community_write_fence('community_bans');