Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 2 additions & 2 deletions ARCHITECTURE.md
Original file line number Diff line number Diff line change
Expand Up @@ -145,7 +145,7 @@ buzz-admin (operator CLI: relay membership + key generation)
buzz-test-client (integration test harness + manual CLI)
```

**Key architectural principle:** The relay is the single source of truth. `buzz-relay` orchestrates all subsystems by calling them directly — it imports `buzz-db`, `buzz-auth`, `buzz-pubsub`, `buzz-search`, `buzz-audit`, and `buzz-workflow`. However, those subsystems are isolated from each other: `buzz-workflow` never calls `buzz-pubsub`, `buzz-search` never calls `buzz-db`, etc. Cross-subsystem coordination happens only through the relay. In multi-community mode, the relay also owns propagation of `TenantContext`; service crates should receive community-scoped inputs rather than independently deriving tenancy from client-controlled event tags.
**Key architectural principle:** The relay is the single source of truth. `buzz-relay` orchestrates all subsystems by calling them directly — it imports `buzz-db`, `buzz-auth`, `buzz-pubsub`, `buzz-search`, `buzz-audit`, and `buzz-workflow`. However, those subsystems are isolated from each other: `buzz-workflow` never calls `buzz-pubsub`, `buzz-search` never calls `buzz-db`, etc. The one sanctioned exception is `buzz-audit`, which calls `buzz_db::deletion::guard_community_write` to admit each audit append into its community before taking the chain lock. Cross-subsystem coordination happens only through the relay. In multi-community mode, the relay also owns propagation of `TenantContext`; service crates should receive community-scoped inputs rather than independently deriving tenancy from client-controlled event tags.

---

Expand Down Expand Up @@ -596,7 +596,7 @@ Tamper-evident append-only log with SHA-256 hash chaining.

**Hash covers:** seq (big-endian bytes), timestamp (RFC3339), event_id, event_kind (big-endian), actor_pubkey, action string, channel_id (16 bytes or 16 zero bytes if None), canonical metadata JSON (BTreeMap for deterministic key ordering), prev_hash.

**Single-writer guarantee:** `pg_advisory_lock` before each transaction. Lock released in all branches including panic (`catch_unwind`).
**Single-writer guarantee:** each append runs in one transaction that first passes community write admission (the shared community deletion lock plus the active/undeleted check, via `buzz-db`'s `DeletionStore::guard_transaction`), then takes a per-community `pg_advisory_xact_lock`. Both locks release on commit or rollback, including a dropped future.

**10 audit actions:** `EventCreated`, `EventDeleted`, `ChannelCreated`, `ChannelUpdated`, `ChannelDeleted`, `MemberAdded`, `MemberRemoved`, `AuthSuccess`, `AuthFailure`, `RateLimitExceeded`.

Expand Down
2 changes: 1 addition & 1 deletion Cargo.lock

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

2 changes: 1 addition & 1 deletion crates/buzz-audit/Cargo.toml
Original file line number Diff line number Diff line change
Expand Up @@ -9,6 +9,7 @@ description = "Hash-chain audit log for Buzz"

[dependencies]
buzz-core = { workspace = true }
buzz-db = { workspace = true }
buzz-datastore-tracing = { workspace = true }
sqlx = { workspace = true }
tokio = { workspace = true }
Expand All @@ -21,4 +22,3 @@ metrics = { workspace = true }
thiserror = { workspace = true }
sha2 = { workspace = true }
hex = { workspace = true }
futures-util = { workspace = true }
7 changes: 7 additions & 0 deletions crates/buzz-audit/src/error.rs
Original file line number Diff line number Diff line change
Expand Up @@ -14,6 +14,10 @@ pub enum AuditError {
#[error("database error: {0}")]
Database(#[from] sqlx::Error),

/// Community write admission failed before the audit chain was locked.
#[error("audit write admission failed")]
Admission(#[from] buzz_db::DbError),

/// The `prev_hash` of an entry does not match the hash of the preceding
/// entry in the same community's chain.
#[error(
Expand Down Expand Up @@ -77,6 +81,9 @@ mod tests {
AuditError::HashMismatch { seq: 42 },
AuditError::UnknownAction,
AuditError::UnsupportedHashVersion { version: 99 },
AuditError::Admission(buzz_db::DbError::AccessDenied(format!(
"community {community} is write-fenced"
))),
];

for err in &domain_errors {
Expand Down
103 changes: 62 additions & 41 deletions crates/buzz-audit/src/service.rs
Original file line number Diff line number Diff line change
@@ -1,6 +1,5 @@
use chrono::{DateTime, Utc};
use futures_util::FutureExt as _;
use sqlx::{Acquire, PgPool, Row};
use sqlx::{PgPool, Row};
use tracing::{debug, warn};
use uuid::Uuid;

Expand All @@ -17,16 +16,14 @@ use crate::{
/// The `created_at` stamped on a new entry.
///
/// Reduced to the precision Postgres round-trips before it is hashed — see
/// [`to_storage_precision`]. Split out from [`AuditService::log_inner`] so the
/// [`to_storage_precision`]. Split out from [`AuditService::log`] so the
/// invariant is testable without a database.
fn log_timestamp() -> DateTime<Utc> {
to_storage_precision(Utc::now())
}

/// 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`].
/// Per-community audit-chain advisory lock key. Admission takes the community
/// deletion lock first; this key then serializes appends within that community.
const AUDIT_LOCK_NAMESPACE: &str = "buzz_audit:";

/// Append-only, per-community hash-chain audit log backed by Postgres.
Expand All @@ -47,50 +44,22 @@ 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).
/// Community admission precedes the transaction-scoped audit-chain lock.
/// Both locks are released on commit or rollback, including a dropped future.
#[datastore_span(
name = "audit_log",
system = "postgresql",
fields(action = %entry.action)
)]
pub async fn log(&self, entry: NewAuditEntry) -> Result<AuditEntry, AuditError> {
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 mut tx = self.pool.begin().await?;
buzz_db::deletion::guard_community_write(&mut tx, entry.community_id).await?;
let lock_key = format!("{AUDIT_LOCK_NAMESPACE}{}", entry.community_id);
sqlx::query("SELECT pg_advisory_lock(hashtextextended($1, 0))")
sqlx::query("SELECT pg_advisory_xact_lock(hashtextextended($1, 0))")
.bind(&lock_key)
.execute(&mut *conn)
.execute(&mut *tx)
.await?;

// 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))
.catch_unwind()
.await;

let _ = sqlx::query("SELECT pg_advisory_unlock(hashtextextended($1, 0))")
.bind(&lock_key)
.execute(&mut *conn)
.await;

match result {
Ok(inner_result) => inner_result,
Err(panic_payload) => std::panic::resume_unwind(panic_payload),
}
}

async fn log_inner(
&self,
conn: &mut sqlx::pool::PoolConnection<sqlx::Postgres>,
entry: NewAuditEntry,
) -> Result<AuditEntry, AuditError> {
let mut tx = conn.begin().await?;

// 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();
Expand Down Expand Up @@ -321,6 +290,58 @@ mod postgres_tests {
id
}

#[tokio::test]
#[ignore = "requires Postgres"]
async fn audit_append_requires_application_admission_with_triggers_bypassed() {
let _g = db_lock().lock().await;
let url = std::env::var("DATABASE_URL").expect("Postgres lane must provide DATABASE_URL");
let pool = sqlx::postgres::PgPoolOptions::new()
.max_connections(1)
.connect(&url)
.await
.expect("connect test database");
let active = make_community(&pool).await;
for state in ["quiescing", "fenced"] {
let id = make_community(&pool).await;
let mut tx = pool.begin().await.expect("begin lifecycle transition");
sqlx::query(
"SELECT set_config('buzz.deletion_executor_community', $1, true), \
set_config('buzz.deletion_fence_generation', \
(SELECT deletion_fence_generation::text FROM communities WHERE id = $2), true)",
)
.bind(id.to_string())
.bind(id)
.execute(&mut *tx)
.await
.expect("authorize lifecycle transition");
sqlx::query("UPDATE communities SET deletion_state = $2 WHERE id = $1")
.bind(id)
.bind(state)
.execute(&mut *tx)
.await
.expect("set lifecycle state");
tx.commit().await.expect("commit lifecycle state");

sqlx::query("SET session_replication_role = replica")
.execute(&pool)
.await
.expect("disable ordinary triggers");
let error = AuditService::new(pool.clone())
.log(new_entry(id, AuditAction::EventCreated))
.await
.expect_err("fenced audit append must reject before chain lock");
assert!(
matches!(error, AuditError::Admission(buzz_db::DbError::AccessDenied(message)) if message.contains("write-fenced")),
"expected application admission failure"
);
}
let entry = AuditService::new(pool.clone())
.log(new_entry(active, AuditAction::EventCreated))
.await
.expect("active audit append succeeds");
assert_eq!(entry.seq, 1);
}

fn new_entry(community_id: Uuid, action: AuditAction) -> NewAuditEntry {
NewAuditEntry {
community_id: CommunityId::from_uuid(community_id),
Expand Down
5 changes: 3 additions & 2 deletions crates/buzz-db/src/lib.rs
Original file line number Diff line number Diff line change
Expand Up @@ -60,8 +60,9 @@ pub const DB_CONNECTION_TERMINALS: [(DbPoolRole, DbConnectionStep, DbConnectionO
pub const DB_CONNECTION_RAW_SERIES_PER_POD: usize = runtime::CONNECTION_RAW_SERIES_PER_POD;
pub(crate) use runtime::{
begin_community_event_write_transaction,
begin_community_event_write_transaction_with_legacy_metrics, insert_mentions_in_transaction,
observability, route_proof, ReadSessionInner, RouteDecision, RoutePredicate,
begin_community_event_write_transaction_with_legacy_metrics, begin_community_write_transaction,
insert_mentions_in_transaction, observability, route_proof, ReadSessionInner, RouteDecision,
RoutePredicate,
};
pub use store::{
admin_moderation, allowlist, api_token, archived_identities, artifact, channel,
Expand Down
8 changes: 4 additions & 4 deletions crates/buzz-db/src/runtime/admitted_tx.rs
Original file line number Diff line number Diff line change
@@ -1,4 +1,4 @@
//! Community-admitted event-write transactions.
//! Community-admitted write transactions.

use std::collections::BTreeSet;

Expand All @@ -9,12 +9,12 @@ use uuid::Uuid;
use crate::deletion::{DeletionStore, ServingWriteLease};
use crate::Result;

/// An event-write transaction that has passed community admission.
/// A transaction that has passed community write admission.
///
/// Its only constructors take the shared community admission lock (or
/// validate a serving write lease) on the transaction they wrap, for the
/// community they record, so construction and admission are one step.
/// Event-write helpers take `&mut AdmittedTx` and read the community from it,
/// Write helpers take `&mut AdmittedTx` and read the community from it,
/// so the compiler rejects a raw [`sqlx::Transaction`] or a transaction
/// admitted for a different community.
///
Expand Down Expand Up @@ -154,7 +154,7 @@ impl AdmittedTx {
}

/// Commit the transaction. This is the only commit path for admitted
/// event writes.
/// writes.
///
/// Refreshes the TTL of every channel that received an event first, as
/// the last statement before COMMIT.
Expand Down
24 changes: 20 additions & 4 deletions crates/buzz-db/src/runtime/mod.rs
Original file line number Diff line number Diff line change
Expand Up @@ -132,9 +132,9 @@ pub(crate) async fn insert_mentions_in_transaction(
Ok(())
}

/// Start a tenant-local event-write transaction and take the shared community
/// Start a tenant-local write transaction and take the shared community
/// deletion lock before any serving mutation.
async fn begin_community_event_write_transaction_with_metric_population(
async fn begin_community_write_transaction_with_metric_population(
pool: &PgPool,
community: CommunityId,
operation: observability::WriterOperation,
Expand Down Expand Up @@ -163,7 +163,7 @@ pub(crate) async fn begin_community_event_write_transaction(
community: CommunityId,
operation: observability::WriterOperation,
) -> Result<AdmittedTx> {
begin_community_event_write_transaction_with_metric_population(
begin_community_write_transaction_with_metric_population(
pool,
community,
operation,
Expand All @@ -177,7 +177,7 @@ pub(crate) async fn begin_community_event_write_transaction_with_legacy_metrics(
community: CommunityId,
operation: observability::WriterOperation,
) -> Result<AdmittedTx> {
begin_community_event_write_transaction_with_metric_population(
begin_community_write_transaction_with_metric_population(
pool,
community,
operation,
Expand All @@ -186,6 +186,22 @@ pub(crate) async fn begin_community_event_write_transaction_with_legacy_metrics(
.await
}

/// Begin a community write before any tenant mutation or row lock. The
/// admission lock is held by the returned transaction until commit or rollback.
pub(crate) async fn begin_community_write_transaction(
pool: &PgPool,
community: CommunityId,
operation: observability::WriterOperation,
) -> Result<AdmittedTx> {
begin_community_write_transaction_with_metric_population(
pool,
community,
operation,
CommunityEventWriteMetricPopulation::TypedOnly,
)
.await
}

/// Database handle. Clone is cheap (Arc-backed pool).
#[derive(Clone, Debug)]
pub struct Db {
Expand Down
Loading
Loading