feat(tracing): add PostgreSQL tracing spans

crates/buzz-audit/src/service.rs

@@ -49,7 +49,7 @@ impl AuditService {
     /// 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).
-    #[instrument(skip(self, entry), fields(action = %entry.action))]
+    #[instrument(target = "buzz_datastore", name = "audit_log", skip_all, fields(otel.kind = "client", db.system.name = "postgresql"))]
     pub async fn log(&self, entry: NewAuditEntry) -> Result<AuditEntry, AuditError> {
         let mut conn = self.pool.acquire().await?;
 
@@ -125,7 +125,7 @@ impl AuditService {
 
         audit_entry.hash = compute_hash(&audit_entry)?.to_vec();
 
-        debug!(seq, "writing audit entry");
+        debug!("writing audit entry");
 
         sqlx::query(
             r#"
@@ -156,7 +156,7 @@ impl AuditService {
     /// Reads exactly that community's chain — it can never observe another
     /// community's entries or head. Returns `Ok(false)` if the range is empty,
     /// `Ok(true)` if the segment is internally consistent.
-    #[instrument(skip(self))]
+    #[instrument(target = "buzz_datastore", name = "audit_verify_chain", skip_all, fields(otel.kind = "client", db.system.name = "postgresql"))]
     pub async fn verify_chain(
         &self,
         community: CommunityId,
@@ -208,7 +208,7 @@ impl AuditService {
     /// Returns up to `limit` entries from one community's chain starting at
     /// `from_seq`, ordered by sequence number. Scoped to `community` — never
     /// returns another community's rows.
-    #[instrument(skip(self))]
+    #[instrument(target = "buzz_datastore", name = "audit_get_entries", skip_all, fields(otel.kind = "client", db.system.name = "postgresql"))]
     pub async fn get_entries(
         &self,
         community: CommunityId,
@@ -274,7 +274,7 @@ mod tests {
 
     async fn test_pool() -> Option<PgPool> {
         let url = 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 fixture
         PgPool::connect(&url).await.ok()
     }
 

crates/buzz-db/src/event.rs

@@ -589,8 +589,8 @@ pub(crate) fn row_to_stored_event(row: sqlx::postgres::PgRow) -> Result<Option<S
     // Avoid the Value → String → parse round-trip: deserialize directly from the Value.
     let event: nostr::Event = match serde_json::from_value(event_json) {
         Ok(e) => e,
-        Err(e) => {
-            tracing::warn!("failed to reconstruct event from DB row: {e}");
+        Err(_e) => {
+            tracing::warn!("failed to reconstruct event from DB row");
             return Ok(None);
         }
     };
@@ -1506,7 +1506,7 @@ mod tests {
     use super::*;
     use nostr::{EventBuilder, Keys, Kind, Tag};
 
-    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 fixture
 
     async fn setup_pool() -> PgPool {
         let database_url = std::env::var("BUZZ_TEST_DATABASE_URL")

crates/buzz-db/src/replica_fence.rs

@@ -315,6 +315,7 @@ impl ReplicaFence {
 ///
 /// This is a name-and-shape check only; it cannot detect a sabotaged
 /// function body. [`verify_floor_guard_behavior`] proves the semantics.
+#[tracing::instrument(target = "buzz_datastore", name = "replica_fence_verify_catalog", skip_all, fields(otel.kind = "client", db.system.name = "postgresql"))]
 pub async fn verify_floor_guard_catalog(pool: &PgPool) -> crate::Result<()> {
     // tgtype bits: 1 = ROW, 2 = BEFORE, 4 = INSERT, 16 = UPDATE, 64 = INSTEAD.
     // Required: ROW + INSERT + UPDATE set, BEFORE + INSTEAD clear.
@@ -369,6 +370,7 @@ pub async fn verify_floor_guard_catalog(pool: &PgPool) -> crate::Result<()> {
 /// `SET CONSTRAINTS ALL IMMEDIATE` makes the deferred trigger fire per
 /// statement so each adversary is observable under a savepoint; deferral to
 /// COMMIT is separately pinned by the held-transaction fixture.
+#[tracing::instrument(target = "buzz_datastore", name = "replica_fence_verify_behavior", skip_all, fields(otel.kind = "client", db.system.name = "postgresql"))]
 pub async fn verify_floor_guard_behavior(pool: &PgPool) -> crate::Result<()> {
     use crate::error::DbError;
 
@@ -542,6 +544,7 @@ pub enum ProbeError {
 /// The statements are separately awaited on a single pinned connection;
 /// a single SELECT would not guarantee evaluation order across the
 /// subexpressions, reopening the race this ordering exists to close.
+#[tracing::instrument(target = "buzz_datastore", name = "replica_fence_sample_writer", skip_all, fields(otel.kind = "client", db.system.name = "postgresql"))]
 async fn sample_writer(writer: &PgPool) -> Result<WriterSample, ProbeError> {
     let mut conn = writer.acquire().await?;
 
@@ -698,6 +701,7 @@ pub const AURORA_IDENTITY_FN: &str = "aurora_db_instance_identifier";
 /// (undefined_function, SQLSTATE 42883); transient errors surface as `Err`
 /// so the caller can retry the probe on a later request instead of caching
 /// a wrong answer.
+#[tracing::instrument(target = "buzz_datastore", name = "replica_fence_reader_supports_aurora_identity", skip_all, fields(otel.kind = "client", db.system.name = "postgresql"))]
 pub async fn reader_supports_aurora_identity(conn: &mut PgConnection) -> Result<bool, sqlx::Error> {
     match sqlx::query(sqlx::AssertSqlSafe(format!(
         "SELECT {AURORA_IDENTITY_FN}()"
@@ -719,6 +723,7 @@ pub async fn reader_supports_aurora_identity(conn: &mut PgConnection) -> Result<
 /// [`reader_supports_aurora_identity`] confirmed it — the function
 /// reference fails at parse time on plain Postgres). `None` when the row
 /// is missing there (migration not yet replayed): fail closed.
+#[tracing::instrument(target = "buzz_datastore", name = "replica_fence_observe_heartbeat", skip_all, fields(otel.kind = "client", db.system.name = "postgresql"))]
 pub async fn observe_heartbeat(
     conn: &mut PgConnection,
     aurora: bool,

crates/buzz-relay/src/handlers/command_executor.rs

@@ -97,6 +97,12 @@ enum PersistResult {
 /// persists without the event record. On retry, the event INSERT succeeds
 /// (no conflict), and the mutation re-executes — which is safe for idempotent
 /// operations (open_dm, hide_dm, update_approval, upsert_workflow).
+#[tracing::instrument(
+    target = "buzz_datastore",
+    name = "persist_command_event",
+    skip_all,
+    fields(otel.kind = "client", db.system.name = "postgresql")
+)]
 async fn persist_command_event(
     state: &Arc<AppState>,
     tenant: &TenantContext,

crates/buzz-search/Cargo.toml

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

crates/buzz-search/src/query.rs

@@ -10,6 +10,7 @@
 
 use buzz_core::CommunityId;
 use sqlx::{PgPool, QueryBuilder, Row};
+use tracing::Instrument;
 use uuid::Uuid;
 
 use crate::error::SearchError;
@@ -297,7 +298,16 @@ pub async fn search(pool: &PgPool, query: &SearchQuery) -> Result<SearchResult,
     qb.push(" OFFSET ");
     qb.push_bind(offset);
 
-    let rows = qb.build().fetch_all(pool).await?;
+    let rows = qb
+        .build()
+        .fetch_all(pool)
+        .instrument(tracing::info_span!(
+            target: "buzz_datastore",
+            "search",
+            otel.kind = "client",
+            db.system.name = "postgresql"
+        ))
+        .await?;
 
     let mut hits = Vec::with_capacity(rows.len());
     for row in rows {