Add database pressure observability (#6700)

## Why

Database pressure currently collapses several distinct delays into one
symptom. This adds the evidence layer needed to distinguish pool
acquisition wait, logical database operation time, advisory-lock wait,
and selected transaction duration before changing timeout or retry
policy.

This is the phase 2 Lane A observability bundle for
[#26](https://github.com/TheSentinel454/buzz/issues/26),
[#28](https://github.com/TheSentinel454/buzz/issues/28), and
[#33](https://github.com/TheSentinel454/buzz/issues/33). It is stacked
on #6668.

## What

- Record explicit reader/writer checkout wait and acquisition outcomes
with `buzz_db_pool_acquire_wait_seconds` and
`buzz_db_pool_acquisitions_total`.
- Extend the compile-time `#[datastore_span(name = "...")]` seam with
`buzz_db_operation_duration_seconds`, so operation labels remain static
source literals instead of request data.
- Route correctness-critical replacement, membership, push-gate,
deletion, and migration/schema-safety advisory locks through one
observer without changing their SQL, order, scope, or blocking behavior.
- Measure six internally owned transaction lifetimes with
`buzz_db_transaction_duration_seconds`, starting after `BEGIN` succeeds
and ending after explicit commit/rollback or scope exit.
- Emit root slow-operation warnings at 500 ms, logging the first slow
completion and then 1/100 per call site with only `operation`,
`outcome`, and `elapsed_ms`.
- Document names, units, fixed label vocabularies, measurement
boundaries, and blind spots in this PR description.

Fixed labels are deliberately small:

- `pool_role`: `writer`, `reader`
- `lock_type`: `replacement`, `membership`, `push_gate`, `deletion`,
`migration_schema_safety`
- `outcome`: `success`, `error`, `timeout` where SQLx/PostgreSQL can
distinguish it accurately
- `operation`: compile-time datastore names plus the six closed
transaction operation names documented in the runbook

No metric or slow warning contains community IDs, event IDs, event
kinds, coordinates, d-tags, SQL/query text, query IDs, returned errors,
or event content.

## Coverage boundaries

- Operation duration is the complete annotated logical function body,
not pure SQL execution; it may include implicit checkout, lock wait,
nested operations, and application work. Cancelled futures do not reach
its completion hook.
- Pool timing covers explicit helper checkouts, including proved-reader
routing and selected writer-owned transactions. Implicit SQLx checkout
through `&PgPool` remains folded into operation duration.
- Lock timing covers application-side blocking locks in the five named
families. Trigger/stored-procedure locks, channel-TTL locking, the usage
try-lock, and the audit service session lock remain outside this slice.
- Transaction timing covers only the six wholly owned boundaries
documented in the runbook. It excludes pool wait, `BEGIN`, asynchronous
rollback cleanup after an early return, and caller-owned
`Db::begin_transaction` lifetime.

## Relationship to #6229

#6229 is the incident-driven timeout precursor. This PR does not add or
change `statement_timeout`, `lock_timeout`,
`idle_in_transaction_session_timeout`, retries, audit durability, or
client-visible conflicts. It provides the missing distributions needed
to evaluate those policies later and intentionally leaves #6229's open
audit retry/durability finding untouched.

The branches overlap in `crates/buzz-db/src/lib.rs` and
`crates/buzz-db/src/migration.rs`, so a later rebase may need textual
conflict resolution, but the behavior is complementary rather than
duplicated.

## Risk assessment

Moderate-low. The primary risk is instrumentation overhead and added
static series. Cardinality is source-bounded, slow logs are
sampled/redacted root events, and the lock/transaction changes wrap
existing awaits without changing policy or ordering.

## Verification

Author workstation: `buzz-tornquist-db-pressure-observability`
(`2010927`), exact head `d7cf833e26c528adfcde3917ded80daf6f4ddac9`,
parent `6f50e6b2b2a996349149af61d35bdd6a355f77fd`.

- `cargo fmt --all --check` — passed
- `cargo clippy -p buzz-datastore-tracing -p buzz-db -p buzz-audit -p
buzz-search -p buzz-relay --all-targets -- -D warnings` — passed
- `cargo test -p buzz-datastore-tracing --quiet` — 4 passed
- `cargo test -p buzz-db --quiet` — 109 passed, 200 ignored
- `cargo test -p buzz-audit -p buzz-search --quiet` — 16 passed, 25
ignored
- `cargo test -p buzz-relay --lib --quiet -- --test-threads=1` — 906
passed, 48 ignored
- Native PostgreSQL focused tests for pool success/timeout/error, lock
success/contention/timeout/error, replacement, membership serialization,
push ordering, deletion fencing, migration/schema exclusion, and reader
fallback — 8 passed

The default-parallel relay run passed once; subsequent runs exposed the
existing load-sensitive
`api::mesh_demo::tests::demo_join_forwarded_arm_round_trips_echo` 504 at
the end of the suite. That test passes in isolation and the full relay
suite passes serially.

Independent exact-head review workstation:
`buzz-tornquist-db-pressure-observability-review` (`2013067`).
Formatting, the same all-target clippy command, datastore
instrumentation tests, DB unit tests, source privacy guards, and
diff/non-goal audits passed; no review findings.

Generated with Codex

---------

Signed-off-by: tornquist <tornquist@squareup.com>

crates/buzz-audit/Cargo.toml

@@ -17,6 +17,7 @@ serde_json  = { workspace = true }
 uuid        = { workspace = true }
 chrono      = { workspace = true }
 tracing     = { workspace = true }
+metrics     = { workspace = true }
 thiserror   = { workspace = true }
 sha2         = { workspace = true }
 hex          = { workspace = true }

crates/buzz-datastore-tracing/Cargo.toml

@@ -16,6 +16,8 @@ quote = "1"
 syn = { version = "2", features = ["full"] }
 
 [dev-dependencies]
+metrics = { workspace = true }
+metrics-util = { workspace = true }
 opentelemetry = { workspace = true }
 opentelemetry_sdk = { workspace = true, features = ["testing"] }
 tokio = { workspace = true }

crates/buzz-datastore-tracing/src/lib.rs

@@ -70,6 +70,9 @@ impl Parse for DatastoreArgs {
 /// PostgreSQL spans always omit function arguments, use the `buzz_datastore`
 /// target, and expose only canonical semantic fields plus explicitly supplied
 /// safe fields. An `Err` sets `otel.status_code` without inspecting the error.
+/// The literal `name` also labels a logical-operation duration histogram. Slow
+/// completions are sampled and logged with only that name, outcome, and elapsed
+/// time; arguments, error values, and return values are never formatted.
 #[proc_macro_attribute]
 pub fn datastore_span(args: TokenStream, item: TokenStream) -> TokenStream {
     let args = parse_macro_input!(args as DatastoreArgs);
@@ -129,9 +132,46 @@ pub fn datastore_span(args: TokenStream, item: TokenStream) -> TokenStream {
             }
         }
     });
+    let outcome = if returns_result {
+        quote! {
+            if #result.is_err() { "error" } else { "success" }
+        }
+    } else {
+        quote!("success")
+    };
     function.block = Box::new(syn::parse_quote!({
+        let __buzz_datastore_started_7f3a9c = ::std::time::Instant::now();
         let #result: #return_type = (async #original_body).await;
         #record_error
+        let __buzz_datastore_outcome_7f3a9c = #outcome;
+        let __buzz_datastore_elapsed_7f3a9c = __buzz_datastore_started_7f3a9c.elapsed();
+        ::metrics::histogram!(
+            "buzz_db_operation_duration_seconds",
+            "operation" => #name,
+            "outcome" => __buzz_datastore_outcome_7f3a9c,
+        )
+        .record(__buzz_datastore_elapsed_7f3a9c.as_secs_f64());
+        if __buzz_datastore_elapsed_7f3a9c >= ::std::time::Duration::from_millis(500) {
+            static __BUZZ_DATASTORE_SLOW_SAMPLE_7F3A9C:
+                ::std::sync::atomic::AtomicU64 = ::std::sync::atomic::AtomicU64::new(0);
+            if __BUZZ_DATASTORE_SLOW_SAMPLE_7F3A9C.fetch_add(
+                1,
+                ::std::sync::atomic::Ordering::Relaxed,
+            ) % 100 == 0 {
+                let __buzz_datastore_elapsed_ms_7f3a9c =
+                    __buzz_datastore_elapsed_7f3a9c
+                        .as_millis()
+                        .min(::std::primitive::u64::MAX as u128) as u64;
+                ::tracing::warn!(
+                    target: "buzz_datastore",
+                    parent: None,
+                    operation = #name,
+                    outcome = __buzz_datastore_outcome_7f3a9c,
+                    elapsed_ms = __buzz_datastore_elapsed_ms_7f3a9c,
+                    "slow datastore operation"
+                );
+            }
+        }
         #result
     }));
 

crates/buzz-datastore-tracing/tests/runtime.rs

@@ -1,6 +1,12 @@
 use buzz_datastore_tracing::datastore_span;
+use metrics_util::debugging::{DebugValue, DebuggingRecorder};
 use opentelemetry::trace::{SpanKind, Status, TracerProvider as _};
 use opentelemetry_sdk::trace::{InMemorySpanExporter, SdkTracerProvider};
+use std::collections::BTreeMap;
+use std::sync::{Arc, Mutex};
+use tracing::field::{Field, Visit};
+use tracing::{Event, Subscriber};
+use tracing_subscriber::layer::{Context, Layer};
 use tracing_subscriber::prelude::*;
 
 const DIRECT_ERROR: &str = "raw-secret-direct-error";
@@ -27,8 +33,48 @@ async fn operation(
     Ok(limit)
 }
 
+#[datastore_span(name = "slow_test_operation", system = "postgresql")]
+async fn slow_operation(delay: std::time::Duration) -> Result<(), &'static str> {
+    tokio::time::sleep(delay).await;
+    Err(DIRECT_ERROR)
+}
+
+#[derive(Default)]
+struct EventFields(BTreeMap<String, String>);
+
+impl Visit for EventFields {
+    fn record_debug(&mut self, field: &Field, value: &dyn std::fmt::Debug) {
+        self.0.insert(field.name().to_owned(), format!("{value:?}"));
+    }
+
+    fn record_str(&mut self, field: &Field, value: &str) {
+        self.0.insert(field.name().to_owned(), value.to_owned());
+    }
+
+    fn record_u64(&mut self, field: &Field, value: u64) {
+        self.0.insert(field.name().to_owned(), value.to_string());
+    }
+}
+
+#[derive(Clone, Default)]
+struct EventCapture(Arc<Mutex<Vec<EventFields>>>);
+
+impl<S> Layer<S> for EventCapture
+where
+    S: Subscriber,
+{
+    fn on_event(&self, event: &Event<'_>, _context: Context<'_, S>) {
+        let mut fields = EventFields::default();
+        event.record(&mut fields);
+        self.0.lock().expect("capture lock").push(fields);
+    }
+}
+
 #[tokio::test(flavor = "current_thread")]
 async fn exports_policy_fields_without_error_or_argument_data() {
+    let recorder = DebuggingRecorder::new();
+    let snapshotter = recorder.snapshotter();
+    let _metrics_guard = metrics::set_default_local_recorder(&recorder);
     let exporter = InMemorySpanExporter::default();
     let provider = SdkTracerProvider::builder()
         .with_simple_exporter(exporter.clone())
@@ -41,6 +87,37 @@ async fn exports_policy_fields_without_error_or_argument_data() {
     assert_eq!(operation(8, true, false).await, Err(DIRECT_ERROR));
     assert_eq!(operation(9, false, true).await, Err(QUESTION_ERROR));
 
+    let operation_samples = snapshotter
+        .snapshot()
+        .into_vec()
+        .into_iter()
+        .filter(|(key, ..)| key.key().name() == "buzz_db_operation_duration_seconds")
+        .map(|(key, _, _, value)| {
+            let DebugValue::Histogram(samples) = value else {
+                panic!("operation duration must be a histogram");
+            };
+            let labels = key
+                .key()
+                .labels()
+                .map(|label| (label.key().to_owned(), label.value().to_owned()))
+                .collect::<std::collections::BTreeMap<_, _>>();
+            (labels, samples)
+        })
+        .collect::<Vec<_>>();
+    assert_eq!(operation_samples.len(), 2);
+    for (labels, samples) in operation_samples {
+        assert_eq!(
+            labels.get("operation").map(String::as_str),
+            Some("test_operation")
+        );
+        assert!(matches!(
+            labels.get("outcome").map(String::as_str),
+            Some("success" | "error")
+        ));
+        assert!(!samples.is_empty());
+        assert!(samples.iter().all(|sample| sample.into_inner() >= 0.0));
+    }
+
     provider.force_flush().expect("spans flush");
     let spans = exporter.get_finished_spans().expect("exported spans");
     assert_eq!(spans.len(), 3);
@@ -78,3 +155,55 @@ async fn exports_policy_fields_without_error_or_argument_data() {
         }
     }
 }
+
+#[tokio::test(flavor = "current_thread")]
+async fn slow_operation_logging_is_guarded_sampled_and_redacted() {
+    let capture = EventCapture::default();
+    let subscriber = tracing_subscriber::registry().with(capture.clone());
+    let _subscriber_guard = tracing::subscriber::set_default(subscriber);
+
+    assert_eq!(
+        slow_operation(std::time::Duration::from_millis(1)).await,
+        Err(DIRECT_ERROR)
+    );
+    assert_eq!(
+        slow_operation(std::time::Duration::from_millis(510)).await,
+        Err(DIRECT_ERROR)
+    );
+    assert_eq!(
+        slow_operation(std::time::Duration::from_millis(510)).await,
+        Err(DIRECT_ERROR)
+    );
+
+    let events = capture.0.lock().expect("capture lock");
+    let slow = events
+        .iter()
+        .filter(|event| {
+            event
+                .0
+                .get("message")
+                .is_some_and(|message| message.contains("slow datastore operation"))
+        })
+        .collect::<Vec<_>>();
+    assert_eq!(
+        slow.len(),
+        1,
+        "first slow call is logged, next 99 are sampled out"
+    );
+    let fields = &slow[0].0;
+    assert_eq!(
+        fields.get("operation").map(String::as_str),
+        Some("slow_test_operation")
+    );
+    assert_eq!(fields.get("outcome").map(String::as_str), Some("error"));
+    assert!(fields
+        .get("elapsed_ms")
+        .and_then(|value| value.parse::<u64>().ok())
+        .is_some_and(|elapsed| elapsed >= 500));
+    assert_eq!(
+        fields.len(),
+        4,
+        "only message and fixed safe fields are logged"
+    );
+    assert!(!format!("{fields:?}").contains(DIRECT_ERROR));
+}

crates/buzz-db/src/channel.rs

@@ -478,14 +478,17 @@ async fn acquire_channel_membership_lock(
     community_id: CommunityId,
     channel_id: Uuid,
 ) -> Result<()> {
-    sqlx::query("SELECT pg_advisory_xact_lock(hashtextextended($1, 0))")
-        .bind(format!(
-            "{CHANNEL_MEMBERSHIP_LOCK_NAMESPACE}{}:{}",
-            community_id.as_uuid(),
-            channel_id
-        ))
-        .execute(&mut **tx)
-        .await?;
+    crate::observability::observe_advisory_lock(
+        crate::observability::LockType::Membership,
+        sqlx::query("SELECT pg_advisory_xact_lock(hashtextextended($1, 0))")
+            .bind(format!(
+                "{CHANNEL_MEMBERSHIP_LOCK_NAMESPACE}{}:{}",
+                community_id.as_uuid(),
+                channel_id
+            ))
+            .execute(&mut **tx),
+    )
+    .await?;
     Ok(())
 }
 
@@ -631,10 +634,13 @@ pub async fn lock_member_snapshot(
         relay_pubkey,
         Some(channel_id.as_bytes()),
     );
-    sqlx::query("SELECT pg_advisory_xact_lock($1)")
-        .bind(replacement_lock)
-        .execute(&mut *tx)
-        .await?;
+    crate::observability::observe_advisory_lock(
+        crate::observability::LockType::Replacement,
+        sqlx::query("SELECT pg_advisory_xact_lock($1)")
+            .bind(replacement_lock)
+            .execute(&mut *tx),
+    )
+    .await?;
     acquire_channel_membership_lock(&mut tx, community_id, channel_id).await?;
     let rows = sqlx::query(
         r#"

crates/buzz-db/src/community.rs

@@ -325,10 +325,13 @@ impl Db {
 
         // Serialize on the owner pubkey so concurrent creates to the same
         // owner cannot both pass the ownership count check.
-        sqlx::query("SELECT pg_advisory_xact_lock($1)")
-            .bind(relay_members::owner_count_advisory_lock_key(&owner_pubkey))
-            .execute(&mut *tx)
-            .await?;
+        crate::observability::observe_advisory_lock(
+            crate::observability::LockType::Membership,
+            sqlx::query("SELECT pg_advisory_xact_lock($1)")
+                .bind(relay_members::owner_count_advisory_lock_key(&owner_pubkey))
+                .execute(&mut *tx),
+        )
+        .await?;
 
         let row = sqlx::query(
             r#"

crates/buzz-db/src/deletion.rs

@@ -1210,12 +1210,15 @@ impl DeletionStore {
     /// Already-acquired leases remain renewable, verifiable, and releasable so
     /// admitted remote effects retain their exclusion proof until completion.
     pub async fn begin_quiescing(&self, token: &LeaseToken) -> Result<()> {
-        let mut tx = self.pool.begin().await?;
+        let (mut tx, transaction_timer) = crate::observability::begin_transaction(
+            &self.pool,
+            crate::observability::TransactionOperation::BeginCommunityDeletionQuiescing,
+        )
+        .await?;
+        transaction_timer
+            .observe(async {
         verify_lease(&mut tx, token, DeletionStage::Approved).await?;
-        sqlx::query("SELECT pg_advisory_xact_lock(community_deletion_lock_key($1))")
-            .bind(token.community_id.as_uuid())
-            .execute(&mut *tx)
-            .await?;
+        lock_community_deletion(&mut tx, token.community_id).await?;
         verify_lease(&mut tx, token, DeletionStage::Approved).await?;
         let (generation, archived_at): (i64, Option<DateTime<Utc>>) = sqlx::query_as(
             "SELECT deletion_fence_generation, archived_at FROM communities WHERE id = $1 FOR UPDATE",
@@ -1259,16 +1262,21 @@ impl DeletionStore {
         .await?;
         tx.commit().await?;
         Ok(())
+            })
+            .await
     }
 
     /// Acquire the universal durable fence after all pre-quiesce serving leases drain.
     pub async fn fence(&self, token: &LeaseToken) -> Result<i64> {
-        let mut tx = self.pool.begin().await?;
+        let (mut tx, transaction_timer) = crate::observability::begin_transaction(
+            &self.pool,
+            crate::observability::TransactionOperation::FenceCommunityDeletion,
+        )
+        .await?;
+        transaction_timer
+            .observe(async {
         verify_lease(&mut tx, token, DeletionStage::Approved).await?;
-        sqlx::query("SELECT pg_advisory_xact_lock(community_deletion_lock_key($1))")
-            .bind(token.community_id.as_uuid())
-            .execute(&mut *tx)
-            .await?;
+        lock_community_deletion(&mut tx, token.community_id).await?;
         verify_lease(&mut tx, token, DeletionStage::Approved).await?;
         let active_serving_writes = sqlx::query(
             "SELECT count(*)::BIGINT AS active_count, \
@@ -1331,6 +1339,8 @@ impl DeletionStore {
         .await?;
         tx.commit().await?;
         Ok(generation)
+            })
+            .await
     }
 
     /// Freeze the exact post-fence storage binding manifest.
@@ -1967,10 +1977,7 @@ impl DeletionStore {
         .ok_or_else(|| DbError::NotFound(format!("community deletion {request_id}")))?;
         // Every lifecycle transition takes the community lock before any row lock.
         // Inverting this order lets abort and the executor deadlock each other.
-        sqlx::query("SELECT pg_advisory_xact_lock(community_deletion_lock_key($1))")
-            .bind(community_id.as_uuid())
-            .execute(&mut *tx)
-            .await?;
+        lock_community_deletion(&mut tx, community_id).await?;
         let row = sqlx::query("SELECT * FROM community_deletion_requests WHERE id = $1 FOR UPDATE")
             .bind(request_id)
             .fetch_optional(&mut *tx)
@@ -2254,10 +2261,7 @@ impl DeletionStore {
         tx: &mut Transaction<'_, Postgres>,
         community: CommunityId,
     ) -> Result<()> {
-        sqlx::query("SELECT pg_advisory_xact_lock_shared(community_deletion_lock_key($1))")
-            .bind(community.as_uuid())
-            .execute(&mut **tx)
-            .await?;
+        lock_community_deletion_shared(tx, community).await?;
         let state: Option<String> = sqlx::query_scalar(
             "SELECT deletion_state FROM communities WHERE id = $1 AND deleted_at IS NULL",
         )
@@ -2286,10 +2290,7 @@ impl DeletionStore {
         tx: &mut Transaction<'_, Postgres>,
         lease: &ServingWriteLease,
     ) -> Result<()> {
-        sqlx::query("SELECT pg_advisory_xact_lock_shared(community_deletion_lock_key($1))")
-            .bind(lease.community_id.as_uuid())
-            .execute(&mut **tx)
-            .await?;
+        lock_community_deletion_shared(tx, lease.community_id).await?;
         let valid: bool = sqlx::query_scalar(
             "SELECT EXISTS(SELECT 1 FROM community_serving_write_leases lease \
              JOIN communities community ON community.id = lease.community_id \
@@ -2407,10 +2408,7 @@ impl DeletionStore {
     ) -> Result<()> {
         let lease_seconds = i64::try_from(lease_duration.as_secs()).unwrap_or(i64::MAX);
         let mut tx = self.pool.begin().await?;
-        sqlx::query("SELECT pg_advisory_xact_lock_shared(community_deletion_lock_key($1))")
-            .bind(lease.community_id.as_uuid())
-            .execute(&mut *tx)
-            .await?;
+        lock_community_deletion_shared(&mut tx, lease.community_id).await?;
         let lease_until: Option<DateTime<Utc>> = sqlx::query_scalar(
             "UPDATE community_serving_write_leases lease \
              SET lease_until = now() + make_interval(secs => $6), heartbeat_at = now() \
@@ -2465,10 +2463,7 @@ impl DeletionStore {
     /// admitted remote effect.
     pub async fn verify_serving_write_lease(&self, lease: &ServingWriteLease) -> Result<()> {
         let mut tx = self.pool.begin().await?;
-        sqlx::query("SELECT pg_advisory_xact_lock_shared(community_deletion_lock_key($1))")
-            .bind(lease.community_id.as_uuid())
-            .execute(&mut *tx)
-            .await?;
+        lock_community_deletion_shared(&mut tx, lease.community_id).await?;
         let valid: bool = sqlx::query_scalar(
             "SELECT EXISTS(SELECT 1 FROM community_serving_write_leases lease \
              JOIN communities community ON community.id = lease.community_id \
@@ -2572,6 +2567,34 @@ impl DeletionStore {
     }
 }
 
+async fn lock_community_deletion(
+    tx: &mut Transaction<'_, Postgres>,
+    community: CommunityId,
+) -> Result<()> {
+    crate::observability::observe_advisory_lock(
+        crate::observability::LockType::Deletion,
+        sqlx::query("SELECT pg_advisory_xact_lock(community_deletion_lock_key($1))")
+            .bind(community.as_uuid())
+            .execute(&mut **tx),
+    )
+    .await?;
+    Ok(())
+}
+
+async fn lock_community_deletion_shared(
+    tx: &mut Transaction<'_, Postgres>,
+    community: CommunityId,
+) -> Result<()> {
+    crate::observability::observe_advisory_lock(
+        crate::observability::LockType::Deletion,
+        sqlx::query("SELECT pg_advisory_xact_lock_shared(community_deletion_lock_key($1))")
+            .bind(community.as_uuid())
+            .execute(&mut **tx),
+    )
+    .await?;
+    Ok(())
+}
+
 /// Take the shared schema/destruction advisory lock for the current
 /// transaction.
 ///
@@ -2580,10 +2603,13 @@ impl DeletionStore {
 /// whole run (see [`crate::migration::run_migrations`]); shared holders do
 /// not block each other, so concurrent deletion executors are unaffected.
 async fn lock_schema_destruction_shared(conn: &mut PgConnection) -> Result<()> {
-    sqlx::query("SELECT pg_advisory_xact_lock_shared($1)")
-        .bind(SCHEMA_DESTRUCTION_LOCK_KEY)
-        .execute(conn)
-        .await?;
+    crate::observability::observe_advisory_lock(
+        crate::observability::LockType::MigrationSchemaSafety,
+        sqlx::query("SELECT pg_advisory_xact_lock_shared($1)")
+            .bind(SCHEMA_DESTRUCTION_LOCK_KEY)
+            .execute(conn),
+    )
+    .await?;
     Ok(())
 }
 
@@ -3302,7 +3328,7 @@ mod postgres_tests {
     async fn store() -> (Db, DeletionStore) {
         let database_url = std::env::var("BUZZ_TEST_DATABASE_URL")
             .or_else(|_| std::env::var("DATABASE_URL"))
-            .unwrap_or_else(|_| "postgres://buzz:buzz_dev@localhost:5432/buzz".to_string());
+            .unwrap_or_else(|_| "postgres://buzz:buzz_dev@localhost:5432/buzz".to_string()); // sadscan:disable np.postgres.1 -- local test-only credentials
         let db = Db::new(&DbConfig {
             database_url,
             max_connections: 5,

crates/buzz-db/src/lib.rs

@@ -50,6 +50,7 @@ pub mod git_repo;
 pub mod migration;
 /// Community moderation: reports, bans/timeouts, audit actions.
 pub mod moderation;
+mod observability;
 /// Monthly table partition management.
 pub mod partition;
 /// Buzz product-feedback sidecar persistence.
@@ -712,7 +713,7 @@ impl Db {
         };
         let aurora_identity = self.reader_aurora_identity.clone();
         tokio::spawn(async move {
-            match read_pool.acquire().await {
+            match observability::acquire(&read_pool, observability::PoolRole::Reader).await {
                 Ok(mut conn) => {
                     tracing::info!("read replica reachable at boot");
                     match replica_fence::reader_supports_aurora_identity(&mut conn).await {
@@ -854,7 +855,7 @@ impl Db {
         // `read_pool` separately would spend a second budget whenever the
         // capability is uncached — i.e. after a failed boot ping, which is
         // precisely the reader-unavailable case the bound must hold for.
-        let conn = match read_pool.acquire().await {
+        let conn = match observability::acquire(read_pool, observability::PoolRole::Reader).await {
             Ok(conn) => conn,
             Err(sqlx::Error::PoolTimedOut) => {
                 tracing::warn!("reader pool acquire timed out; routing to writer");
@@ -1030,7 +1031,8 @@ impl Db {
         &self,
         lock_key: i64,
     ) -> Result<Option<UsageMetricsLeader>> {
-        let mut connection = self.pool.acquire().await?;
+        let mut connection =
+            observability::acquire(&self.pool, observability::PoolRole::Writer).await?;
         let acquired = sqlx::query_scalar::<_, bool>("SELECT pg_try_advisory_lock($1)")
             .bind(lock_key)
             .fetch_one(&mut *connection)
@@ -1177,7 +1179,11 @@ impl Db {
     /// Returns a `'static` transaction because `PgPool` is `Arc`-backed internally.
     /// The transaction holds an owned pool handle, not a borrow.
     pub async fn begin_transaction(&self) -> Result<sqlx::Transaction<'static, sqlx::Postgres>> {
-        self.pool.begin().await.map_err(Into::into)
+        let connection =
+            observability::acquire(&self.pool, observability::PoolRole::Writer).await?;
+        sqlx::Transaction::begin(connection, None)
+            .await
+            .map_err(Into::into)
     }
 
     /// Inserts an event. Returns `(StoredEvent, was_inserted)` — `false` on duplicate.
@@ -4349,14 +4355,23 @@ impl Db {
             channel_id.as_ref().map(|id| id.as_bytes().as_slice()),
         );
 
-        let mut tx = self.pool.begin().await?;
+        let (mut tx, transaction_timer) = observability::begin_transaction(
+            &self.pool,
+            observability::TransactionOperation::ReplaceAddressableEvent,
+        )
+        .await?;
+        transaction_timer
+            .observe(async {
 
         // Serialize all writers for the same (kind, pubkey, channel_id) tuple.
         // Advisory lock is transaction-scoped — released on commit/rollback.
-        sqlx::query("SELECT pg_advisory_xact_lock($1)")
-            .bind(lock_key)
-            .execute(&mut *tx)
-            .await?;
+        observability::observe_advisory_lock(
+            observability::LockType::Replacement,
+            sqlx::query("SELECT pg_advisory_xact_lock($1)")
+                .bind(lock_key)
+                .execute(&mut *tx),
+        )
+        .await?;
 
         // Check for the newest existing event. ORDER BY + LIMIT 1 is defensive against
         // historical data where prior bugs may have left multiple live rows.
@@ -4452,6 +4467,8 @@ impl Db {
             StoredEvent::with_received_at(event.clone(), received_at, channel_id, true),
             true,
         ))
+            })
+            .await
     }
 
     /// Returns whether the relay-authored NIP-43 snapshot is absent or differs
@@ -4532,16 +4549,25 @@ impl Db {
             None,
         );
 
-        let mut tx = self.pool.begin().await?;
+        let (mut tx, transaction_timer) = observability::begin_transaction(
+            &self.pool,
+            observability::TransactionOperation::PublishNip43MembershipLocked,
+        )
+        .await?;
+        let (event, received_at, was_inserted, member_count) = transaction_timer
+            .observe(async {
 
         // Acquire the per-community snapshot lock BEFORE reading members.
         // This serializes the entire read-build-write cycle: a concurrent
         // publication will block here until our transaction commits, then
         // read the updated membership state.
-        sqlx::query("SELECT pg_advisory_xact_lock($1)")
-            .bind(lock_key)
-            .execute(&mut *tx)
-            .await?;
+        observability::observe_advisory_lock(
+            observability::LockType::Membership,
+            sqlx::query("SELECT pg_advisory_xact_lock($1)")
+                .bind(lock_key)
+                .execute(&mut *tx),
+        )
+        .await?;
 
         // Read current members inside the locked transaction.
         let rows = sqlx::query(
@@ -4616,24 +4642,24 @@ impl Db {
         .await?;
 
         let was_inserted = insert_result.rows_affected() > 0;
-        if !was_inserted {
+        if was_inserted {
+            tx.commit().await?;
+        } else {
             tx.rollback().await?;
-            return Ok((
-                StoredEvent::with_received_at(event, received_at, None, false),
-                false,
-                member_count,
-            ));
         }
+        Ok::<_, DbError>((event, received_at, was_inserted, member_count))
+            })
+            .await?;
 
-        tx.commit().await?;
-
-        if let Err(e) = crate::insert_mentions(&self.pool, community_id, &event, None).await {
-            tracing::warn!(event_id = %event.id, "Failed to insert mentions: {e}");
+        if was_inserted {
+            if let Err(e) = crate::insert_mentions(&self.pool, community_id, &event, None).await {
+                tracing::warn!(event_id = %event.id, "Failed to insert mentions: {e}");
+            }
         }
 
         Ok((
-            StoredEvent::with_received_at(event, received_at, None, true),
-            true,
+            StoredEvent::with_received_at(event, received_at, None, was_inserted),
+            was_inserted,
             member_count,
         ))
     }

crates/buzz-db/src/migration.rs

@@ -81,11 +81,16 @@ where
     F: FnOnce(PgConnection) -> Fut,
     Fut: Future<Output = (PgConnection, Result<T>)>,
 {
-    let mut lock_conn = pool.acquire().await?.detach();
-    sqlx::query("SELECT pg_advisory_lock($1)")
-        .bind(SCHEMA_DESTRUCTION_LOCK_KEY)
-        .execute(&mut lock_conn)
-        .await?;
+    let mut lock_conn = crate::observability::acquire(pool, crate::observability::PoolRole::Writer)
+        .await?
+        .detach();
+    crate::observability::observe_advisory_lock(
+        crate::observability::LockType::MigrationSchemaSafety,
+        sqlx::query("SELECT pg_advisory_lock($1)")
+            .bind(SCHEMA_DESTRUCTION_LOCK_KEY)
+            .execute(&mut lock_conn),
+    )
+    .await?;
     let (mut lock_conn, outcome) = op(lock_conn).await;
     let unlock = sqlx::query("SELECT pg_advisory_unlock($1)")
         .bind(SCHEMA_DESTRUCTION_LOCK_KEY)

crates/buzz-db/src/observability.rs

@@ -0,0 +1,636 @@
+//! Bounded-cardinality database pressure instrumentation primitives.
+//!
+//! Label values come only from the closed enums in this module. Callers must
+//! never derive labels from tenant data, events, SQL text, or query identifiers.
+
+use std::future::Future;
+use std::time::{Duration, Instant};
+
+#[derive(Clone, Copy, Debug, Eq, PartialEq)]
+pub(crate) enum PoolRole {
+    Writer,
+    Reader,
+}
+
+impl PoolRole {
+    #[cfg(test)]
+    pub(crate) const ALL: [Self; 2] = [Self::Writer, Self::Reader];
+
+    pub(crate) const fn as_str(self) -> &'static str {
+        match self {
+            Self::Writer => "writer",
+            Self::Reader => "reader",
+        }
+    }
+}
+
+#[derive(Clone, Copy, Debug, Eq, PartialEq)]
+pub(crate) enum LockType {
+    Replacement,
+    Membership,
+    PushGate,
+    Deletion,
+    MigrationSchemaSafety,
+}
+
+impl LockType {
+    #[cfg(test)]
+    pub(crate) const ALL: [Self; 5] = [
+        Self::Replacement,
+        Self::Membership,
+        Self::PushGate,
+        Self::Deletion,
+        Self::MigrationSchemaSafety,
+    ];
+
+    pub(crate) const fn as_str(self) -> &'static str {
+        match self {
+            Self::Replacement => "replacement",
+            Self::Membership => "membership",
+            Self::PushGate => "push_gate",
+            Self::Deletion => "deletion",
+            Self::MigrationSchemaSafety => "migration_schema_safety",
+        }
+    }
+}
+
+#[derive(Clone, Copy, Debug, Eq, PartialEq)]
+pub(crate) enum Outcome {
+    Success,
+    Error,
+    Timeout,
+}
+
+impl Outcome {
+    #[cfg(test)]
+    pub(crate) const ALL: [Self; 3] = [Self::Success, Self::Error, Self::Timeout];
+
+    pub(crate) const fn as_str(self) -> &'static str {
+        match self {
+            Self::Success => "success",
+            Self::Error => "error",
+            Self::Timeout => "timeout",
+        }
+    }
+
+    fn from_sqlx_error(error: &sqlx::Error) -> Self {
+        match error {
+            sqlx::Error::PoolTimedOut => Self::Timeout,
+            sqlx::Error::Database(database) if database.code().as_deref() == Some("55P03") => {
+                Self::Timeout
+            }
+            _ => Self::Error,
+        }
+    }
+}
+
+#[derive(Clone, Copy, Debug, Eq, PartialEq)]
+pub(crate) enum TransactionOperation {
+    ReplaceParameterizedEvent,
+    ReplaceAddressableEvent,
+    PublishNip43MembershipLocked,
+    AcceptPushLeaseEvent,
+    BeginCommunityDeletionQuiescing,
+    FenceCommunityDeletion,
+}
+
+impl TransactionOperation {
+    #[cfg(test)]
+    pub(crate) const ALL: [Self; 6] = [
+        Self::ReplaceParameterizedEvent,
+        Self::ReplaceAddressableEvent,
+        Self::PublishNip43MembershipLocked,
+        Self::AcceptPushLeaseEvent,
+        Self::BeginCommunityDeletionQuiescing,
+        Self::FenceCommunityDeletion,
+    ];
+
+    pub(crate) const fn as_str(self) -> &'static str {
+        match self {
+            Self::ReplaceParameterizedEvent => "replace_parameterized_event",
+            Self::ReplaceAddressableEvent => "replace_addressable_event",
+            Self::PublishNip43MembershipLocked => "publish_nip43_membership_locked",
+            Self::AcceptPushLeaseEvent => "accept_push_lease_event",
+            Self::BeginCommunityDeletionQuiescing => "begin_community_deletion_quiescing",
+            Self::FenceCommunityDeletion => "fence_community_deletion",
+        }
+    }
+}
+
+pub(crate) fn record_pool_acquire(role: PoolRole, outcome: Outcome, elapsed: Duration) {
+    metrics::histogram!(
+        "buzz_db_pool_acquire_wait_seconds",
+        "pool_role" => role.as_str(),
+        "outcome" => outcome.as_str(),
+    )
+    .record(elapsed.as_secs_f64());
+    metrics::counter!(
+        "buzz_db_pool_acquisitions_total",
+        "pool_role" => role.as_str(),
+        "outcome" => outcome.as_str(),
+    )
+    .increment(1);
+}
+
+pub(crate) async fn acquire(
+    pool: &sqlx::PgPool,
+    role: PoolRole,
+) -> sqlx::Result<sqlx::pool::PoolConnection<sqlx::Postgres>> {
+    let started = Instant::now();
+    let result = pool.acquire().await;
+    let outcome = result
+        .as_ref()
+        .map(|_| Outcome::Success)
+        .unwrap_or_else(Outcome::from_sqlx_error);
+    record_pool_acquire(role, outcome, started.elapsed());
+    result
+}
+
+pub(crate) async fn begin_transaction(
+    pool: &sqlx::PgPool,
+    operation: TransactionOperation,
+) -> sqlx::Result<(sqlx::Transaction<'static, sqlx::Postgres>, TransactionTimer)> {
+    let connection = acquire(pool, PoolRole::Writer).await?;
+    let transaction = sqlx::Transaction::begin(connection, None).await?;
+    Ok((transaction, TransactionTimer::start(operation)))
+}
+
+pub(crate) async fn observe_advisory_lock<T, F>(lock_type: LockType, future: F) -> sqlx::Result<T>
+where
+    F: Future<Output = sqlx::Result<T>>,
+{
+    let started = Instant::now();
+    let result = future.await;
+    let outcome = result
+        .as_ref()
+        .map(|_| Outcome::Success)
+        .unwrap_or_else(Outcome::from_sqlx_error);
+    metrics::histogram!(
+        "buzz_db_advisory_lock_wait_seconds",
+        "lock_type" => lock_type.as_str(),
+        "outcome" => outcome.as_str(),
+    )
+    .record(started.elapsed().as_secs_f64());
+    metrics::counter!(
+        "buzz_db_advisory_lock_acquisitions_total",
+        "lock_type" => lock_type.as_str(),
+        "outcome" => outcome.as_str(),
+    )
+    .increment(1);
+    result
+}
+
+pub(crate) struct TransactionTimer {
+    operation: TransactionOperation,
+    started: Instant,
+    outcome: Outcome,
+}
+
+impl TransactionTimer {
+    pub(crate) fn start(operation: TransactionOperation) -> Self {
+        Self {
+            operation,
+            started: Instant::now(),
+            outcome: Outcome::Error,
+        }
+    }
+
+    pub(crate) async fn observe<T, E, F>(mut self, future: F) -> Result<T, E>
+    where
+        F: Future<Output = Result<T, E>>,
+    {
+        let result = future.await;
+        if result.is_ok() {
+            self.outcome = Outcome::Success;
+        }
+        result
+    }
+}
+
+impl Drop for TransactionTimer {
+    fn drop(&mut self) {
+        metrics::histogram!(
+            "buzz_db_transaction_duration_seconds",
+            "operation" => self.operation.as_str(),
+            "outcome" => self.outcome.as_str(),
+        )
+        .record(self.started.elapsed().as_secs_f64());
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::{
+        acquire, observe_advisory_lock, record_pool_acquire, LockType, Outcome, PoolRole,
+        TransactionOperation, TransactionTimer,
+    };
+    use metrics_util::debugging::{DebugValue, DebuggingRecorder};
+    use std::collections::{BTreeMap, BTreeSet};
+    use std::time::Duration;
+
+    #[test]
+    fn label_vocabularies_are_closed_and_documented() {
+        assert_eq!(PoolRole::ALL.map(PoolRole::as_str), ["writer", "reader"]);
+        assert_eq!(
+            LockType::ALL.map(LockType::as_str),
+            [
+                "replacement",
+                "membership",
+                "push_gate",
+                "deletion",
+                "migration_schema_safety",
+            ]
+        );
+        assert_eq!(
+            Outcome::ALL.map(Outcome::as_str),
+            ["success", "error", "timeout"]
+        );
+        assert_eq!(
+            TransactionOperation::ALL.map(TransactionOperation::as_str),
+            [
+                "replace_parameterized_event",
+                "replace_addressable_event",
+                "publish_nip43_membership_locked",
+                "accept_push_lease_event",
+                "begin_community_deletion_quiescing",
+                "fence_community_deletion",
+            ]
+        );
+    }
+
+    #[tokio::test(flavor = "current_thread")]
+    async fn transaction_timer_observe_classifies_result_outcomes() {
+        let recorder = DebuggingRecorder::new();
+        let snapshotter = recorder.snapshotter();
+        let _guard = metrics::set_default_local_recorder(&recorder);
+
+        let success = TransactionTimer::start(TransactionOperation::ReplaceParameterizedEvent)
+            .observe(async { Ok::<_, &str>("committed") })
+            .await;
+        assert_eq!(success, Ok("committed"));
+
+        let error = TransactionTimer::start(TransactionOperation::AcceptPushLeaseEvent)
+            .observe(async { Err::<(), _>("rollback") })
+            .await;
+        assert_eq!(error, Err("rollback"));
+
+        let keys = snapshotter
+            .snapshot()
+            .into_vec()
+            .into_iter()
+            .map(|(key, ..)| {
+                let labels = key
+                    .key()
+                    .labels()
+                    .map(|label| (label.key().to_owned(), label.value().to_owned()))
+                    .collect::<BTreeMap<_, _>>();
+                (key.key().name().to_owned(), labels)
+            })
+            .collect::<BTreeSet<_>>();
+
+        for (operation, outcome) in [
+            ("replace_parameterized_event", "success"),
+            ("accept_push_lease_event", "error"),
+        ] {
+            assert!(keys.contains(&(
+                "buzz_db_transaction_duration_seconds".to_owned(),
+                [
+                    ("operation".to_owned(), operation.to_owned()),
+                    ("outcome".to_owned(), outcome.to_owned()),
+                ]
+                .into_iter()
+                .collect(),
+            )));
+        }
+    }
+
+    #[tokio::test(flavor = "current_thread")]
+    async fn primitives_record_fixed_success_error_and_timeout_labels() {
+        let recorder = DebuggingRecorder::new();
+        let snapshotter = recorder.snapshotter();
+        let _guard = metrics::set_default_local_recorder(&recorder);
+
+        record_pool_acquire(
+            PoolRole::Writer,
+            Outcome::Success,
+            Duration::from_millis(12),
+        );
+        record_pool_acquire(
+            PoolRole::Reader,
+            Outcome::Timeout,
+            Duration::from_millis(34),
+        );
+        let lock_ok: sqlx::Result<()> =
+            observe_advisory_lock(LockType::Replacement, async { Ok(()) }).await;
+        assert!(lock_ok.is_ok());
+        let lock_error: sqlx::Result<()> =
+            observe_advisory_lock(LockType::Membership, async { Err(sqlx::Error::PoolClosed) })
+                .await;
+        assert!(lock_error.is_err());
+
+        let committed: Result<(), ()> =
+            TransactionTimer::start(TransactionOperation::ReplaceParameterizedEvent)
+                .observe(async { Ok(()) })
+                .await;
+        assert!(committed.is_ok());
+        let rolled_back: Result<(), ()> =
+            TransactionTimer::start(TransactionOperation::AcceptPushLeaseEvent)
+                .observe(async { Err(()) })
+                .await;
+        assert!(rolled_back.is_err());
+
+        let snapshot = snapshotter.snapshot().into_vec();
+        let keys = snapshot
+            .iter()
+            .map(|(key, ..)| {
+                let labels = key
+                    .key()
+                    .labels()
+                    .map(|label| (label.key().to_owned(), label.value().to_owned()))
+                    .collect::<BTreeMap<_, _>>();
+                (key.key().name().to_owned(), labels)
+            })
+            .collect::<BTreeSet<_>>();
+
+        for expected in [
+            (
+                "buzz_db_pool_acquire_wait_seconds",
+                [("outcome", "success"), ("pool_role", "writer")],
+            ),
+            (
+                "buzz_db_pool_acquire_wait_seconds",
+                [("outcome", "timeout"), ("pool_role", "reader")],
+            ),
+            (
+                "buzz_db_pool_acquisitions_total",
+                [("outcome", "success"), ("pool_role", "writer")],
+            ),
+            (
+                "buzz_db_pool_acquisitions_total",
+                [("outcome", "timeout"), ("pool_role", "reader")],
+            ),
+            (
+                "buzz_db_advisory_lock_wait_seconds",
+                [("lock_type", "replacement"), ("outcome", "success")],
+            ),
+            (
+                "buzz_db_advisory_lock_wait_seconds",
+                [("lock_type", "membership"), ("outcome", "error")],
+            ),
+            (
+                "buzz_db_advisory_lock_acquisitions_total",
+                [("lock_type", "replacement"), ("outcome", "success")],
+            ),
+            (
+                "buzz_db_advisory_lock_acquisitions_total",
+                [("lock_type", "membership"), ("outcome", "error")],
+            ),
+            (
+                "buzz_db_transaction_duration_seconds",
+                [
+                    ("operation", "replace_parameterized_event"),
+                    ("outcome", "success"),
+                ],
+            ),
+            (
+                "buzz_db_transaction_duration_seconds",
+                [
+                    ("operation", "accept_push_lease_event"),
+                    ("outcome", "error"),
+                ],
+            ),
+        ] {
+            let expected_labels = expected
+                .1
+                .into_iter()
+                .map(|(key, value)| (key.to_owned(), value.to_owned()))
+                .collect::<BTreeMap<_, _>>();
+            assert!(
+                keys.contains(&(expected.0.to_owned(), expected_labels)),
+                "missing metric series {expected:?}; got {keys:?}"
+            );
+        }
+
+        for (key, _, _, value) in snapshot {
+            if key.key().name().ends_with("_seconds") {
+                let DebugValue::Histogram(samples) = value else {
+                    panic!("seconds metrics must be histograms");
+                };
+                assert!(samples.iter().all(|sample| sample.into_inner() >= 0.0));
+            } else if key.key().name().ends_with("_total") {
+                let DebugValue::Counter(value) = value else {
+                    panic!("total metrics must be counters");
+                };
+                assert_eq!(value, 1);
+            }
+        }
+    }
+
+    #[tokio::test(flavor = "current_thread")]
+    #[ignore = "requires Postgres"]
+    async fn pool_acquire_records_success_timeout_and_error_with_wait_time() {
+        let database_url = std::env::var("TEST_DATABASE_URL")
+            .unwrap_or_else(|_| "postgres://buzz:buzz_dev@localhost:5432/buzz".to_owned()); // sadscan:disable np.postgres.1 -- local test-only credentials
+        let pool = sqlx::postgres::PgPoolOptions::new()
+            .max_connections(1)
+            .acquire_timeout(Duration::from_millis(75))
+            .connect(&database_url)
+            .await
+            .expect("connect size-one test pool");
+        let recorder = DebuggingRecorder::new();
+        let snapshotter = recorder.snapshotter();
+        let _guard = metrics::set_default_local_recorder(&recorder);
+
+        let held = acquire(&pool, PoolRole::Writer)
+            .await
+            .expect("writer acquire succeeds");
+        let timeout = acquire(&pool, PoolRole::Reader)
+            .await
+            .expect_err("reader-labeled checkout times out while pool is saturated");
+        assert!(matches!(timeout, sqlx::Error::PoolTimedOut));
+        drop(held);
+        pool.close().await;
+        let closed = acquire(&pool, PoolRole::Writer)
+            .await
+            .expect_err("closed pool acquire errors");
+        assert!(matches!(closed, sqlx::Error::PoolClosed));
+
+        let mut outcomes = BTreeMap::<(String, String), Vec<f64>>::new();
+        for (key, _, _, value) in snapshotter.snapshot().into_vec() {
+            if key.key().name() != "buzz_db_pool_acquire_wait_seconds" {
+                continue;
+            }
+            let DebugValue::Histogram(samples) = value else {
+                panic!("pool wait must be a histogram");
+            };
+            let labels = key.key().labels().collect::<Vec<_>>();
+            let label = |name: &str| {
+                labels
+                    .iter()
+                    .find(|label| label.key() == name)
+                    .map(|label| label.value().to_owned())
+                    .unwrap_or_default()
+            };
+            outcomes.insert(
+                (label("pool_role"), label("outcome")),
+                samples
+                    .into_iter()
+                    .map(|sample| sample.into_inner())
+                    .collect(),
+            );
+        }
+        assert!(outcomes.contains_key(&("writer".to_owned(), "success".to_owned())));
+        assert!(outcomes.contains_key(&("writer".to_owned(), "error".to_owned())));
+        let timeout_samples = outcomes
+            .get(&("reader".to_owned(), "timeout".to_owned()))
+            .expect("reader timeout series");
+        assert!(
+            timeout_samples.iter().any(|sample| *sample >= 0.05),
+            "timeout wait must include the saturated checkout delay: {timeout_samples:?}"
+        );
+    }
+
+    #[tokio::test(flavor = "current_thread")]
+    #[ignore = "requires Postgres"]
+    async fn advisory_lock_records_success_contention_timeout_and_error() {
+        let database_url = std::env::var("TEST_DATABASE_URL")
+            .unwrap_or_else(|_| "postgres://buzz:buzz_dev@localhost:5432/buzz".to_owned()); // sadscan:disable np.postgres.1 -- local test-only credentials
+        let pool = sqlx::postgres::PgPoolOptions::new()
+            .max_connections(4)
+            .connect(&database_url)
+            .await
+            .expect("connect advisory-lock test pool");
+        let recorder = DebuggingRecorder::new();
+        let snapshotter = recorder.snapshotter();
+        let _guard = metrics::set_default_local_recorder(&recorder);
+
+        let mut success_tx = pool.begin().await.expect("begin success transaction");
+        observe_advisory_lock(
+            LockType::Replacement,
+            sqlx::query("SELECT pg_advisory_xact_lock($1)")
+                .bind(0x62757a7a6f627331_i64)
+                .execute(&mut *success_tx),
+        )
+        .await
+        .expect("uncontended lock succeeds");
+        success_tx
+            .rollback()
+            .await
+            .expect("rollback success transaction");
+
+        let contention_key = 0x62757a7a6f627332_i64;
+        let mut holder = pool.begin().await.expect("begin lock holder");
+        sqlx::query("SELECT pg_advisory_xact_lock($1)")
+            .bind(contention_key)
+            .execute(&mut *holder)
+            .await
+            .expect("holder acquires contention key");
+        let mut waiter = pool.begin().await.expect("begin lock waiter");
+        let waiter_task = tokio::spawn(async move {
+            let result = observe_advisory_lock(
+                LockType::Deletion,
+                sqlx::query("SELECT pg_advisory_xact_lock($1)")
+                    .bind(contention_key)
+                    .execute(&mut *waiter),
+            )
+            .await;
+            (waiter, result)
+        });
+        tokio::time::sleep(Duration::from_millis(60)).await;
+        assert!(
+            !waiter_task.is_finished(),
+            "waiter must be blocked by holder"
+        );
+        holder.commit().await.expect("release contention key");
+        let (waiter, waited) = waiter_task.await.expect("join lock waiter");
+        waited.expect("contended lock succeeds after release");
+        waiter.rollback().await.expect("rollback waiter");
+
+        let timeout_key = 0x62757a7a6f627333_i64;
+        let mut timeout_holder = pool.begin().await.expect("begin timeout holder");
+        sqlx::query("SELECT pg_advisory_xact_lock($1)")
+            .bind(timeout_key)
+            .execute(&mut *timeout_holder)
+            .await
+            .expect("holder acquires timeout key");
+        let mut timeout_waiter = pool.begin().await.expect("begin timeout waiter");
+        sqlx::query("SET LOCAL lock_timeout = '30ms'")
+            .execute(&mut *timeout_waiter)
+            .await
+            .expect("set test-only lock timeout");
+        let timed_out = observe_advisory_lock(
+            LockType::MigrationSchemaSafety,
+            sqlx::query("SELECT pg_advisory_xact_lock($1)")
+                .bind(timeout_key)
+                .execute(&mut *timeout_waiter),
+        )
+        .await
+        .expect_err("lock wait times out");
+        assert_eq!(
+            timed_out
+                .as_database_error()
+                .and_then(|error| error.code())
+                .as_deref(),
+            Some("55P03")
+        );
+        timeout_holder
+            .rollback()
+            .await
+            .expect("release timeout key");
+
+        let mut aborted = pool.begin().await.expect("begin error transaction");
+        sqlx::query("SELECT 1 / 0")
+            .execute(&mut *aborted)
+            .await
+            .expect_err("abort transaction before lock");
+        observe_advisory_lock(
+            LockType::Membership,
+            sqlx::query("SELECT pg_advisory_xact_lock($1)")
+                .bind(0x62757a7a6f627334_i64)
+                .execute(&mut *aborted),
+        )
+        .await
+        .expect_err("lock statement fails in aborted transaction");
+        aborted
+            .rollback()
+            .await
+            .expect("rollback aborted transaction");
+
+        let mut outcomes = BTreeMap::<(String, String), Vec<f64>>::new();
+        for (key, _, _, value) in snapshotter.snapshot().into_vec() {
+            if key.key().name() != "buzz_db_advisory_lock_wait_seconds" {
+                continue;
+            }
+            let DebugValue::Histogram(samples) = value else {
+                panic!("lock wait must be a histogram");
+            };
+            let labels = key.key().labels().collect::<Vec<_>>();
+            let label = |name: &str| {
+                labels
+                    .iter()
+                    .find(|label| label.key() == name)
+                    .map(|label| label.value().to_owned())
+                    .unwrap_or_default()
+            };
+            outcomes.insert(
+                (label("lock_type"), label("outcome")),
+                samples
+                    .into_iter()
+                    .map(|sample| sample.into_inner())
+                    .collect(),
+            );
+        }
+        assert!(outcomes.contains_key(&("replacement".to_owned(), "success".to_owned())));
+        assert!(outcomes.contains_key(&("membership".to_owned(), "error".to_owned())));
+        assert!(
+            outcomes.contains_key(&("migration_schema_safety".to_owned(), "timeout".to_owned()))
+        );
+        let contention = outcomes
+            .get(&("deletion".to_owned(), "success".to_owned()))
+            .expect("deletion contention series");
+        assert!(
+            contention.iter().any(|sample| *sample >= 0.04),
+            "lock timer must include the holder wait: {contention:?}"
+        );
+    }
+}

crates/buzz-db/src/push.rs

@@ -25,10 +25,13 @@ async fn acquire_push_gate_lock(
     tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
     community: CommunityId,
 ) -> Result<()> {
-    sqlx::query("SELECT pg_advisory_xact_lock(hashtextextended($1, 0))")
-        .bind(format!("{PUSH_GATE_LOCK_NAMESPACE}{}", community.as_uuid()))
-        .execute(&mut **tx)
-        .await?;
+    crate::observability::observe_advisory_lock(
+        crate::observability::LockType::PushGate,
+        sqlx::query("SELECT pg_advisory_xact_lock(hashtextextended($1, 0))")
+            .bind(format!("{PUSH_GATE_LOCK_NAMESPACE}{}", community.as_uuid()))
+            .execute(&mut **tx),
+    )
+    .await?;
     Ok(())
 }
 
@@ -220,7 +223,13 @@ pub async fn accept_lease_event(
     max_active_leases: i64,
 ) -> Result<AcceptLeaseOutcome> {
     let author = event.pubkey.as_bytes();
-    let mut tx = pool.begin().await?;
+    let (mut tx, transaction_timer) = crate::observability::begin_transaction(
+        pool,
+        crate::observability::TransactionOperation::AcceptPushLeaseEvent,
+    )
+    .await?;
+    transaction_timer
+        .observe(async {
     let mut address_lock = Vec::with_capacity(16 + author.len() + installation_id.len());
     address_lock.extend_from_slice(community.as_uuid().as_bytes());
     address_lock.extend_from_slice(author);
@@ -230,14 +239,20 @@ pub async fn accept_lease_event(
     author_lock.extend_from_slice(community.as_uuid().as_bytes());
     author_lock.extend_from_slice(author);
     let author_lock = i64::from_le_bytes(Sha256::digest(&author_lock)[..8].try_into().unwrap());
-    sqlx::query("SELECT pg_advisory_xact_lock($1)")
-        .bind(address_lock)
-        .execute(&mut *tx)
-        .await?;
-    sqlx::query("SELECT pg_advisory_xact_lock($1)")
-        .bind(author_lock)
-        .execute(&mut *tx)
-        .await?;
+    crate::observability::observe_advisory_lock(
+        crate::observability::LockType::PushGate,
+        sqlx::query("SELECT pg_advisory_xact_lock($1)")
+            .bind(address_lock)
+            .execute(&mut *tx),
+    )
+    .await?;
+    crate::observability::observe_advisory_lock(
+        crate::observability::LockType::PushGate,
+        sqlx::query("SELECT pg_advisory_xact_lock($1)")
+            .bind(author_lock)
+            .execute(&mut *tx),
+    )
+    .await?;
     // T1b: an activation can flip the community from "no eligible lease" to
     // "eligible", so it must serialize against the trigger's shared gate lock.
     // Acquired after the address/author locks to keep one global lock order.
@@ -392,6 +407,8 @@ pub async fn accept_lease_event(
     }
     tx.commit().await?;
     Ok(AcceptLeaseOutcome::Accepted)
+        })
+        .await
 }
 
 fn constraint_acceptance_outcome(error: &sqlx::Error) -> Option<AcceptLeaseOutcome> {
@@ -1271,7 +1288,7 @@ mod tests {
     async fn setup_pool() -> PgPool {
         let database_url = std::env::var("BUZZ_TEST_DATABASE_URL")
             .or_else(|_| std::env::var("DATABASE_URL"))
-            .unwrap_or_else(|_| "postgres://buzz:buzz_dev@localhost:5432/buzz".into());
+            .unwrap_or_else(|_| "postgres://buzz:buzz_dev@localhost:5432/buzz".into()); // sadscan:disable np.postgres.1 -- local test-only credentials
         let pool = PgPool::connect(&database_url)
             .await
             .expect("connect to test DB");

crates/buzz-db/src/relay_members.rs

@@ -473,10 +473,13 @@ pub async fn transfer_ownership(
 
     // 1. Serialize on the transferee so concurrent transfers to the same
     //    recipient cannot both pass the ownership count check.
-    sqlx::query("SELECT pg_advisory_xact_lock($1)")
-        .bind(owner_count_advisory_lock_key(&pubkey))
-        .execute(&mut *tx)
-        .await?;
+    crate::observability::observe_advisory_lock(
+        crate::observability::LockType::Membership,
+        sqlx::query("SELECT pg_advisory_xact_lock($1)")
+            .bind(owner_count_advisory_lock_key(&pubkey))
+            .execute(&mut *tx),
+    )
+    .await?;
 
     // 2. Lock the current owner row FOR UPDATE and verify the expected owner.
     //    FOR UPDATE prevents the stale-owner race: a concurrent transfer that
@@ -637,7 +640,7 @@ mod tests {
     use super::*;
     use uuid::Uuid;
 
-    const TEST_DB_URL: &str = "postgres://buzz:buzz_dev@localhost:5432/buzz";
+    const TEST_DB_URL: &str = "postgres://buzz:buzz_dev@localhost:5432/buzz"; // sadscan:disable np.postgres.1 -- local test-only credentials
 
     async fn setup_pool() -> PgPool {
         let database_url = std::env::var("BUZZ_TEST_DATABASE_URL")

crates/buzz-db/src/replaceable.rs

@@ -6,6 +6,7 @@ use chrono::{DateTime, Utc};
 use sqlx::{Acquire, Postgres, Transaction};
 use uuid::Uuid;
 
+use crate::observability::{self, LockType, TransactionOperation};
 use crate::{Db, DbError, Result};
 
 /// Result category for a parameterized-replaceable event write.
@@ -122,10 +123,13 @@ async fn replace_parameterized_event_in_transaction_impl(
         pubkey_bytes.as_slice(),
         Some(d_tag.as_bytes()),
     );
-    sqlx::query("SELECT pg_advisory_xact_lock($1)")
-        .bind(lock_key)
-        .execute(&mut **tx)
-        .await?;
+    observability::observe_advisory_lock(
+        LockType::Replacement,
+        sqlx::query("SELECT pg_advisory_xact_lock($1)")
+            .bind(lock_key)
+            .execute(&mut **tx),
+    )
+    .await?;
 
     let d_tag_count = event
         .tags
@@ -395,23 +399,31 @@ impl Db {
         d_tag: &str,
         channel_id: Option<Uuid>,
     ) -> Result<(StoredEvent, bool)> {
-        let mut tx = self.pool.begin().await?;
-        let result = self
-            .replace_parameterized_event_in_transaction(
-                &mut tx,
-                community_id,
-                event,
-                d_tag,
-                channel_id,
-                ParameterizedReplacePrecondition::Unconditional,
-            )
-            .await?;
-        let was_inserted = result.status == ParameterizedReplaceStatus::Inserted;
-        if was_inserted {
-            tx.commit().await?;
-        } else {
-            tx.rollback().await?;
-        }
-        Ok((result.event, was_inserted))
+        let (mut tx, transaction_timer) = observability::begin_transaction(
+            &self.pool,
+            TransactionOperation::ReplaceParameterizedEvent,
+        )
+        .await?;
+        transaction_timer
+            .observe(async {
+                let result = self
+                    .replace_parameterized_event_in_transaction(
+                        &mut tx,
+                        community_id,
+                        event,
+                        d_tag,
+                        channel_id,
+                        ParameterizedReplacePrecondition::Unconditional,
+                    )
+                    .await?;
+                let was_inserted = result.status == ParameterizedReplaceStatus::Inserted;
+                if was_inserted {
+                    tx.commit().await?;
+                } else {
+                    tx.rollback().await?;
+                }
+                Ok((result.event, was_inserted))
+            })
+            .await
     }
 }

crates/buzz-db/tests/observability_source.rs

@@ -0,0 +1,40 @@
+#[test]
+fn database_metrics_and_slow_logs_exclude_sensitive_or_unbounded_fields() {
+    let implementation = include_str!("../src/observability.rs");
+    let datastore_macro = include_str!("../../buzz-datastore-tracing/src/lib.rs");
+    let instrumentation = format!("{implementation}\n{datastore_macro}");
+
+    for forbidden in [
+        "\"community\" =>",
+        "\"event_id\" =>",
+        "\"event_kind\" =>",
+        "\"kind\" =>",
+        "\"sql\" =>",
+        "\"query\" =>",
+        "\"query_id\" =>",
+        "\"d_tag\" =>",
+        "\"coordinate\" =>",
+        "community =",
+        "event_id =",
+        "event_kind =",
+        "sql =",
+        "query_id =",
+        "d_tag =",
+        "coordinate =",
+    ] {
+        assert!(
+            !instrumentation.contains(forbidden),
+            "database instrumentation must not expose {forbidden}"
+        );
+    }
+
+    assert!(datastore_macro.contains("name: LitStr"));
+    assert!(datastore_macro.contains("\"operation\" => #name"));
+    assert!(datastore_macro.contains("elapsed_ms ="));
+    assert!(
+        datastore_macro.contains("parent: None"),
+        "slow warnings must not inherit dynamic datastore span fields"
+    );
+    // The runtime tracing-layer assertion covers field names because a source
+    // search would also match ordinary local variables such as `record_error`.
+}

crates/buzz-search/Cargo.toml

@@ -14,6 +14,7 @@ sqlx        = { workspace = true }
 uuid        = { workspace = true }
 thiserror   = { workspace = true }
 tracing     = { workspace = true }
+metrics     = { workspace = true }
 
 [dev-dependencies]
 tokio       = { workspace = true }