mirror of
https://github.com/block/buzz.git
synced 2026-11-13 12:51:34 +00:00
relay: batch audit-log appends per community (T2a write-amp)
Amortize the per-message audit cost: the worker now drains up to BUZZ_AUDIT_BATCH_MAX entries (default 100) per BUZZ_AUDIT_BATCH_INTERVAL_MS window (default 50ms) instead of one txn + advisory lock round trip per entry. - AuditService::log_batch: single-community atomic multi-row append. Head re-read inside the txn under the same per-community advisory lock; entries chain in input order; log() delegates to a batch of one so there is exactly one chain-append implementation. Mixed-community batches are rejected (MixedBatch) before any lock is taken. - Worker flush partitions FIFO-per-community, appends each community's run atomically. Cross-community order carries no chain meaning. - Failure classification (AuditError::is_deterministic): SQLSTATE 22/23 are entry-dependent; everything else is infrastructure and retried with backoff (3 attempts) — an outage never marks healthy entries as poison. Deterministic failures bisect the slice so a poisoned batch of N drops exactly 1 entry, never the other N-1. - Shutdown drain reuses the same batched flush path. Postgres-gated tests: batch chain linkage + verify, mixed-batch rejection without writes, poisoned-batch atomicity + deterministic classification, bisection drops exactly the poison with chain intact, cross-tenant poison isolation in one drained batch. Co-authored-by: Tyler Longwell <tlongwell@block.xyz> Signed-off-by: Tyler Longwell <tlongwell@block.xyz>
This commit is contained in:
npub1qyvc0c5kl4gqv2fd97fsk46tu378sqgy35vc83rvgfwne90sel7s0ed67d
parent
8908bd6b71
commit
d1e632a94f
@@ -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 {
|
||||
|
||||
@@ -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<AuditEntry, AuditError> {
|
||||
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<NewAuditEntry>,
|
||||
) -> Result<Vec<AuditEntry>, 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<sqlx::Postgres>,
|
||||
entry: NewAuditEntry,
|
||||
) -> Result<AuditEntry, AuditError> {
|
||||
entries: Vec<NewAuditEntry>,
|
||||
) -> Result<Vec<AuditEntry>, 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<Vec<u8>>) = match head {
|
||||
let (mut prev_seq, mut prev_hash): (i64, Option<Vec<u8>>) = match head {
|
||||
Some(row) => (
|
||||
row.get::<i64, _>("seq"),
|
||||
Some(row.get::<Vec<u8>, _>("hash")),
|
||||
),
|
||||
None => (0, None), // community's first entry
|
||||
};
|
||||
let seq = prev_seq + 1;
|
||||
|
||||
let created_at: DateTime<Utc> = Utc::now();
|
||||
// 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> = 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,
|
||||
};
|
||||
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();
|
||||
|
||||
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<sqlx::Postgres> = 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::<serde_json::Value>("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() {
|
||||
|
||||
@@ -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::<i32>().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,
|
||||
|
||||
@@ -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<buzz_audit::NewAuditEntry> = 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<buzz_audit::NewAuditEntry>,
|
||||
) {
|
||||
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<buzz_audit::NewAuditEntry>,
|
||||
)> = 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<buzz_audit::NewAuditEntry>,
|
||||
) {
|
||||
// Recursive async fn needs boxing; depth is log2(batch_max), trivial.
|
||||
fn go<'a>(
|
||||
audit: &'a buzz_audit::AuditService,
|
||||
entries: Vec<buzz_audit::NewAuditEntry>,
|
||||
) -> std::pin::Pin<Box<dyn std::future::Future<Output = ()> + 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<sqlx::PgPool> {
|
||||
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<buzz_audit::NewAuditEntry> =
|
||||
(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::<Vec<_>>(),
|
||||
"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"
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user