feat(tracing): add PostgreSQL tracing spans (#3678)

## Why
Expose PostgreSQL datastore latency within existing request traces so
slow logical database operations can be identified without recording
tenant data or query arguments.

## What
- Add client spans around logical PostgreSQL operations across the
database facade, search, audit, replica fencing, and command persistence
- Use a dedicated `buzz_datastore` target and `db.system.name =
"postgresql"` for filtering and backend classification
- Exclude health-check database calls and scrub raw identifiers and
errors from newly traced paths

## Risk Assessment
Medium — this instruments frequently used datastore paths and increases
trace volume when enabled, but does not change SQL execution or
datastore behavior. Existing OpenTelemetry filtering controls export.

## References
- Pre-push clippy and fast unit-test hooks passed

Generated with Amp

---------

Signed-off-by: David Grochowski <dgrochowski@squareup.com>
Co-authored-by: Amp <amp@ampcode.com>

Cargo.lock

@@ -914,6 +914,7 @@ name = "buzz-audit"
 version = "0.1.0"
 dependencies = [
  "buzz-core",
+ "buzz-datastore-tracing",
  "chrono",
  "futures-util",
  "hex",
@@ -1028,11 +1029,27 @@ dependencies = [
  "zeroize",
 ]
 
+[[package]]
+name = "buzz-datastore-tracing"
+version = "0.1.0"
+dependencies = [
+ "opentelemetry 0.32.0",
+ "opentelemetry_sdk 0.32.1",
+ "proc-macro2",
+ "quote",
+ "syn 2.0.117",
+ "tokio",
+ "tracing",
+ "tracing-opentelemetry",
+ "tracing-subscriber",
+]
+
 [[package]]
 name = "buzz-db"
 version = "0.1.0"
 dependencies = [
  "buzz-core",
+ "buzz-datastore-tracing",
  "chrono",
  "hex",
  "metrics",
@@ -1220,6 +1237,7 @@ dependencies = [
  "buzz-auth",
  "buzz-conformance",
  "buzz-core",
+ "buzz-datastore-tracing",
  "buzz-db",
  "buzz-media",
  "buzz-pubsub",
@@ -1316,9 +1334,11 @@ name = "buzz-search"
 version = "0.1.0"
 dependencies = [
  "buzz-core",
+ "buzz-datastore-tracing",
  "sqlx",
  "thiserror 2.0.18",
  "tokio",
+ "tracing",
  "uuid",
 ]
 

Cargo.toml

@@ -28,6 +28,7 @@ members = [
     "crates/buzz-dev-mcp",
     "crates/buzz-voice",
     "crates/buzz-backend-kubernetes",
+    "crates/buzz-datastore-tracing",
     "examples/countdown-bot",
 ]
 exclude = ["desktop/src-tauri"]
@@ -143,6 +144,7 @@ buzz-media = { path = "crates/buzz-media" }
 buzz-sdk = { path = "crates/buzz-sdk" }
 buzz-ws-client = { path = "crates/buzz-ws-client" }
 buzz-relay-mesh = { path = "crates/buzz-relay-mesh" }
+buzz-datastore-tracing = { path = "crates/buzz-datastore-tracing" }
 
 # CI profile — builds the relay for desktop e2e. Dependencies keep full
 # release optimization (warm from main's cache; they carry the runtime hot

crates/buzz-audit/Cargo.toml

@@ -9,6 +9,7 @@ description = "Hash-chain audit log for Buzz"
 
 [dependencies]
 buzz-core = { workspace = true }
+buzz-datastore-tracing = { workspace = true }
 sqlx        = { workspace = true }
 tokio       = { workspace = true }
 serde       = { workspace = true }

crates/buzz-audit/src/service.rs

@@ -1,10 +1,11 @@
 use chrono::{DateTime, Utc};
 use futures_util::FutureExt as _;
 use sqlx::{Acquire, PgPool, Row};
-use tracing::{debug, instrument, warn};
+use tracing::{debug, warn};
 use uuid::Uuid;
 
 use buzz_core::CommunityId;
+use buzz_datastore_tracing::datastore_span;
 
 use crate::{
     action::AuditAction,
@@ -49,7 +50,11 @@ 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))]
+    #[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?;
 
@@ -156,7 +161,11 @@ 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))]
+    #[datastore_span(
+        name = "audit_verify_chain",
+        system = "postgresql",
+        fields(from_seq = from_seq, to_seq = to_seq)
+    )]
     pub async fn verify_chain(
         &self,
         community: CommunityId,
@@ -208,7 +217,11 @@ 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))]
+    #[datastore_span(
+        name = "audit_get_entries",
+        system = "postgresql",
+        fields(from_seq = from_seq, limit = limit)
+    )]
     pub async fn get_entries(
         &self,
         community: CommunityId,
@@ -274,7 +287,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-datastore-tracing/Cargo.toml

@@ -0,0 +1,24 @@
+[package]
+name = "buzz-datastore-tracing"
+version.workspace = true
+edition.workspace = true
+rust-version.workspace = true
+license.workspace = true
+repository.workspace = true
+description = "Privacy-preserving datastore tracing policy macros for Buzz"
+
+[lib]
+proc-macro = true
+
+[dependencies]
+proc-macro2 = "1"
+quote = "1"
+syn = { version = "2", features = ["full"] }
+
+[dev-dependencies]
+opentelemetry = { workspace = true }
+opentelemetry_sdk = { workspace = true, features = ["testing"] }
+tokio = { workspace = true }
+tracing = { workspace = true }
+tracing-opentelemetry = { workspace = true }
+tracing-subscriber = { workspace = true }

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

@@ -0,0 +1,168 @@
+//! Policy-enforcing instrumentation for logical datastore operations.
+
+use proc_macro::TokenStream;
+use proc_macro2::TokenStream as TokenStream2;
+use quote::{format_ident, quote};
+use syn::parse::{Parse, ParseStream};
+use syn::{
+    parenthesized, parse_macro_input, Error, Ident, ItemFn, LitStr, Result, ReturnType, Token, Type,
+};
+
+struct DatastoreArgs {
+    name: LitStr,
+    system: LitStr,
+    fields: Option<TokenStream2>,
+}
+
+impl Parse for DatastoreArgs {
+    fn parse(input: ParseStream<'_>) -> Result<Self> {
+        let mut name = None;
+        let mut system = None;
+        let mut fields = None;
+
+        while !input.is_empty() {
+            let key: Ident = input.parse()?;
+            match key.to_string().as_str() {
+                "name" => {
+                    if name.is_some() {
+                        return Err(Error::new(key.span(), "duplicate `name` argument"));
+                    }
+                    input.parse::<Token![=]>()?;
+                    name = Some(input.parse()?);
+                }
+                "system" => {
+                    if system.is_some() {
+                        return Err(Error::new(key.span(), "duplicate `system` argument"));
+                    }
+                    input.parse::<Token![=]>()?;
+                    system = Some(input.parse()?);
+                }
+                "fields" => {
+                    if fields.is_some() {
+                        return Err(Error::new(key.span(), "duplicate `fields` argument"));
+                    }
+                    let content;
+                    parenthesized!(content in input);
+                    fields = Some(content.parse()?);
+                }
+                _ => {
+                    return Err(Error::new(
+                        key.span(),
+                        "expected `name`, `system`, or `fields`",
+                    ))
+                }
+            }
+            if !input.is_empty() {
+                input.parse::<Token![,]>()?;
+            }
+        }
+
+        Ok(Self {
+            name: name.ok_or_else(|| Error::new(input.span(), "missing `name`"))?,
+            system: system.ok_or_else(|| Error::new(input.span(), "missing `system`"))?,
+            fields,
+        })
+    }
+}
+
+/// Instruments an async logical datastore operation according to Buzz policy.
+///
+/// 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.
+#[proc_macro_attribute]
+pub fn datastore_span(args: TokenStream, item: TokenStream) -> TokenStream {
+    let args = parse_macro_input!(args as DatastoreArgs);
+    let mut function = parse_macro_input!(item as ItemFn);
+
+    if function.sig.asyncness.is_none() {
+        return Error::new_spanned(
+            function.sig.fn_token,
+            "`datastore_span` requires an async function",
+        )
+        .into_compile_error()
+        .into();
+    }
+    if args.system.value() != "postgresql" {
+        return Error::new_spanned(
+            args.system,
+            "unsupported datastore system; only `postgresql` is currently supported",
+        )
+        .into_compile_error()
+        .into();
+    }
+
+    let name = args.name;
+    let extra_fields = args.fields.map(|tokens| quote!(, #tokens));
+    function.attrs.push(syn::parse_quote!(
+        #[::tracing::instrument(
+            target = "buzz_datastore",
+            name = #name,
+            skip_all,
+            fields(
+                otel.kind = "client",
+                db.system.name = "postgresql",
+                otel.status_code = ::tracing::field::Empty
+                #extra_fields
+            )
+        )]
+    ));
+
+    let return_type: Type = match &function.sig.output {
+        ReturnType::Type(_, ty) => *ty.clone(),
+        ReturnType::Default => syn::parse_quote!(()),
+    };
+    let returns_result = match &return_type {
+        Type::Path(path) => path
+            .path
+            .segments
+            .last()
+            .is_some_and(|segment| segment.ident == "Result"),
+        _ => false,
+    };
+    let original_body = function.block;
+    let result = format_ident!("__buzz_datastore_result_7f3a9c");
+    let record_error = returns_result.then(|| {
+        quote! {
+            if #result.is_err() {
+                ::tracing::Span::current().record("otel.status_code", "ERROR");
+            }
+        }
+    });
+    function.block = Box::new(syn::parse_quote!({
+        let #result: #return_type = (async #original_body).await;
+        #record_error
+        #result
+    }));
+
+    quote!(#function).into()
+}
+
+#[cfg(test)]
+mod tests {
+    use super::DatastoreArgs;
+
+    #[test]
+    fn parses_safe_tracing_fields() {
+        let args: DatastoreArgs = syn::parse_str(
+            r#"name = "audit", system = "postgresql", fields(action = %entry.action, from_seq = from_seq, limit = limit)"#,
+        )
+        .expect("valid arguments");
+        assert!(args.fields.is_some());
+    }
+
+    #[test]
+    fn rejects_duplicate_arguments() {
+        for args in [
+            r#"name = "one", name = "two", system = "postgresql""#,
+            r#"name = "one", system = "postgresql", system = "postgresql""#,
+            r#"name = "one", system = "postgresql", fields(a = 1), fields(b = 2)"#,
+        ] {
+            let error = match syn::parse_str::<DatastoreArgs>(args) {
+                Ok(_) => panic!("duplicate accepted"),
+                Err(error) => error,
+            };
+            assert!(error.to_string().starts_with("duplicate `"));
+        }
+    }
+}

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

@@ -0,0 +1,80 @@
+use buzz_datastore_tracing::datastore_span;
+use opentelemetry::trace::{SpanKind, Status, TracerProvider as _};
+use opentelemetry_sdk::trace::{InMemorySpanExporter, SdkTracerProvider};
+use tracing_subscriber::prelude::*;
+
+const DIRECT_ERROR: &str = "raw-secret-direct-error";
+const QUESTION_ERROR: &str = "raw-secret-question-error";
+
+fn question_path(fail: bool) -> Result<(), &'static str> {
+    if fail {
+        Err(QUESTION_ERROR)
+    } else {
+        Ok(())
+    }
+}
+
+#[datastore_span(name = "test_operation", system = "postgresql", fields(limit = limit))]
+async fn operation(
+    limit: usize,
+    direct_error: bool,
+    question_error: bool,
+) -> Result<usize, &'static str> {
+    if direct_error {
+        return Err(DIRECT_ERROR);
+    }
+    question_path(question_error)?;
+    Ok(limit)
+}
+
+#[tokio::test(flavor = "current_thread")]
+async fn exports_policy_fields_without_error_or_argument_data() {
+    let exporter = InMemorySpanExporter::default();
+    let provider = SdkTracerProvider::builder()
+        .with_simple_exporter(exporter.clone())
+        .build();
+    let subscriber = tracing_subscriber::registry()
+        .with(tracing_opentelemetry::layer().with_tracer(provider.tracer("datastore-macro-test")));
+    let _subscriber_guard = tracing::subscriber::set_default(subscriber);
+
+    assert_eq!(operation(7, false, false).await, Ok(7));
+    assert_eq!(operation(8, true, false).await, Err(DIRECT_ERROR));
+    assert_eq!(operation(9, false, true).await, Err(QUESTION_ERROR));
+
+    provider.force_flush().expect("spans flush");
+    let spans = exporter.get_finished_spans().expect("exported spans");
+    assert_eq!(spans.len(), 3);
+
+    for (span, (expected_limit, expected_status)) in spans.iter().zip([
+        (7_i64, Status::Unset),
+        (8_i64, Status::error("")),
+        (9_i64, Status::error("")),
+    ]) {
+        assert_eq!(span.name, "test_operation");
+        assert_eq!(span.span_kind, SpanKind::Client);
+        assert_eq!(span.status, expected_status);
+
+        let attributes = span
+            .attributes
+            .iter()
+            .map(|attribute| (attribute.key.as_str(), attribute.value.to_string()))
+            .collect::<Vec<_>>();
+        assert!(attributes.contains(&("target", "buzz_datastore".to_owned())));
+        assert!(attributes.contains(&("db.system.name", "postgresql".to_owned())));
+        assert!(attributes.contains(&("limit", expected_limit.to_string())));
+        assert!(!attributes
+            .iter()
+            .any(|(key, _)| { matches!(*key, "direct_error" | "question_error") }));
+
+        let exported = format!("{span:?}");
+        assert!(!exported.contains(DIRECT_ERROR));
+        assert!(!exported.contains(QUESTION_ERROR));
+        assert!(span.events.iter().all(|event| {
+            !format!("{event:?}").contains(DIRECT_ERROR)
+                && !format!("{event:?}").contains(QUESTION_ERROR)
+        }));
+        if let Status::Error { description } = &span.status {
+            assert!(description.is_empty());
+        }
+    }
+}

crates/buzz-db/Cargo.toml

@@ -9,6 +9,7 @@ description = "Postgres event store and data access layer for Buzz"
 
 [dependencies]
 buzz-core = { workspace = true }
+buzz-datastore-tracing = { workspace = true }
 sqlx        = { workspace = true }
 tokio       = { workspace = true }
 serde       = { workspace = true }

crates/buzz-db/src/replica_fence.rs

@@ -63,6 +63,8 @@ use chrono::{DateTime, Utc};
 use sqlx::{PgConnection, PgPool, Row};
 use uuid::Uuid;
 
+use buzz_datastore_tracing::datastore_span;
+
 /// Seconds of `created_at` history the commit-time floor guard tolerates.
 ///
 /// Must exceed the relay's ingest envelope (±900 s) by enough slack that a
@@ -315,6 +317,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.
+#[datastore_span(name = "replica_fence_verify_catalog", system = "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 +372,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.
+#[datastore_span(name = "replica_fence_verify_behavior", system = "postgresql")]
 pub async fn verify_floor_guard_behavior(pool: &PgPool) -> crate::Result<()> {
     use crate::error::DbError;
 
@@ -698,6 +702,10 @@ 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.
+#[datastore_span(
+    name = "replica_fence_reader_supports_aurora_identity",
+    system = "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}()"

crates/buzz-relay/Cargo.toml

@@ -19,6 +19,7 @@ path = "src/main.rs"
 buzz-core = { workspace = true }
 buzz-conformance = { workspace = true }
 buzz-db = { workspace = true }
+buzz-datastore-tracing = { workspace = true }
 buzz-auth = { workspace = true }
 buzz-pubsub = { workspace = true }
 buzz-audit = { workspace = true }

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

@@ -19,6 +19,7 @@ use uuid::Uuid;
 
 use buzz_core::kind::*;
 use buzz_core::tenant::{CommunityId, TenantContext};
+use buzz_datastore_tracing::datastore_span;
 use buzz_db::workflow::{ApprovalStatus, RunStatus};
 use buzz_db::DbError;
 use buzz_workflow::executor::TriggerContext;
@@ -97,6 +98,7 @@ 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).
+#[datastore_span(name = "persist_command_event", system = "postgresql")]
 async fn persist_command_event(
     state: &Arc<AppState>,
     tenant: &TenantContext,

crates/buzz-search/Cargo.toml

@@ -9,9 +9,11 @@ description = "Postgres full-text search for Buzz, scoped by community"
 
 [dependencies]
 buzz-core   = { workspace = true }
+buzz-datastore-tracing = { 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

@@ -8,10 +8,12 @@
 //!
 //! See conformance row 50.
 
-use buzz_core::CommunityId;
 use sqlx::{PgPool, QueryBuilder, Row};
 use uuid::Uuid;
 
+use buzz_core::CommunityId;
+use buzz_datastore_tracing::datastore_span;
+
 use crate::error::SearchError;
 
 /// Channel-scope filter for a community-scoped FTS query.
@@ -213,6 +215,7 @@ fn normalized_search_text(q: &str) -> Option<String> {
 ///
 /// `community_id = $ctx` is the first predicate and is non-negotiable. There
 /// is no code path through this function that omits it.
+#[datastore_span(name = "search", system = "postgresql")]
 pub async fn search(pool: &PgPool, query: &SearchQuery) -> Result<SearchResult, SearchError> {
     let Some(search_text) = normalized_search_text(&query.q) else {
         return Ok(SearchResult {