Add database pressure observability

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#"