From 6f8ad98b2c126cdf16527c75567bd160a1203c06 Mon Sep 17 00:00:00 2001 From: tornquist Date: Mon, 24 Aug 2026 16:47:23 +0000 Subject: [PATCH 1/3] Add database pressure observability Signed-off-by: tornquist --- Cargo.lock | 4 + crates/buzz-audit/Cargo.toml | 1 + crates/buzz-datastore-tracing/Cargo.toml | 2 + crates/buzz-datastore-tracing/src/lib.rs | 40 ++ .../buzz-datastore-tracing/tests/runtime.rs | 129 ++++ crates/buzz-db/src/channel.rs | 30 +- crates/buzz-db/src/community.rs | 11 +- crates/buzz-db/src/deletion.rs | 90 +-- crates/buzz-db/src/lib.rs | 54 +- crates/buzz-db/src/migration.rs | 15 +- crates/buzz-db/src/observability.rs | 580 ++++++++++++++++++ crates/buzz-db/src/push.rs | 50 +- crates/buzz-db/src/relay_members.rs | 13 +- crates/buzz-db/src/replaceable.rs | 19 +- crates/buzz-db/tests/observability_source.rs | 40 ++ crates/buzz-search/Cargo.toml | 1 + docs/database-observability.md | 40 ++ 17 files changed, 1025 insertions(+), 94 deletions(-) create mode 100644 crates/buzz-db/src/observability.rs create mode 100644 crates/buzz-db/tests/observability_source.rs create mode 100644 docs/database-observability.md diff --git a/Cargo.lock b/Cargo.lock index 18c53c18ca0..f89d845fd9c 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -920,6 +920,7 @@ dependencies = [ "chrono", "futures-util", "hex", + "metrics", "serde", "serde_json", "sha2 0.11.0", @@ -1035,6 +1036,8 @@ dependencies = [ name = "buzz-datastore-tracing" version = "0.1.0" dependencies = [ + "metrics", + "metrics-util", "opentelemetry 0.32.0", "opentelemetry_sdk 0.32.1", "proc-macro2", @@ -1360,6 +1363,7 @@ version = "0.1.0" dependencies = [ "buzz-core", "buzz-datastore-tracing", + "metrics", "sqlx", "thiserror 2.0.18", "tokio", diff --git a/crates/buzz-audit/Cargo.toml b/crates/buzz-audit/Cargo.toml index dfa73353ded..766ade65050 100644 --- a/crates/buzz-audit/Cargo.toml +++ b/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 } diff --git a/crates/buzz-datastore-tracing/Cargo.toml b/crates/buzz-datastore-tracing/Cargo.toml index e93900c54ce..fb7ba6f37d8 100644 --- a/crates/buzz-datastore-tracing/Cargo.toml +++ b/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 } diff --git a/crates/buzz-datastore-tracing/src/lib.rs b/crates/buzz-datastore-tracing/src/lib.rs index f2645cb8f37..217f2335dee 100644 --- a/crates/buzz-datastore-tracing/src/lib.rs +++ b/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 })); diff --git a/crates/buzz-datastore-tracing/tests/runtime.rs b/crates/buzz-datastore-tracing/tests/runtime.rs index b58dca8715f..3355190956f 100644 --- a/crates/buzz-datastore-tracing/tests/runtime.rs +++ b/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); + +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>>); + +impl Layer 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::>(); + (labels, samples) + }) + .collect::>(); + 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::>(); + 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::().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)); +} diff --git a/crates/buzz-db/src/channel.rs b/crates/buzz-db/src/channel.rs index a1890adb56e..98790e3d623 100644 --- a/crates/buzz-db/src/channel.rs +++ b/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#" diff --git a/crates/buzz-db/src/community.rs b/crates/buzz-db/src/community.rs index 64c116a5200..5df896a3500 100644 --- a/crates/buzz-db/src/community.rs +++ b/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#" diff --git a/crates/buzz-db/src/deletion.rs b/crates/buzz-db/src/deletion.rs index fbe69f22a68..0751cd55ffb 100644 --- a/crates/buzz-db/src/deletion.rs +++ b/crates/buzz-db/src/deletion.rs @@ -1121,12 +1121,13 @@ 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, mut transaction_timer) = crate::observability::begin_transaction( + &self.pool, + crate::observability::TransactionOperation::BeginCommunityDeletionQuiescing, + ) + .await?; 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>) = sqlx::query_as( "SELECT deletion_fence_generation, archived_at FROM communities WHERE id = $1 FOR UPDATE", @@ -1169,17 +1170,19 @@ impl DeletionStore { ) .await?; tx.commit().await?; + transaction_timer.mark_success(); Ok(()) } /// Acquire the universal durable fence after all pre-quiesce serving leases drain. pub async fn fence(&self, token: &LeaseToken) -> Result { - let mut tx = self.pool.begin().await?; + let (mut tx, mut transaction_timer) = crate::observability::begin_transaction( + &self.pool, + crate::observability::TransactionOperation::FenceCommunityDeletion, + ) + .await?; 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, \ @@ -1241,6 +1244,7 @@ impl DeletionStore { ) .await?; tx.commit().await?; + transaction_timer.mark_success(); Ok(generation) } @@ -1878,10 +1882,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) @@ -2165,10 +2166,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 = sqlx::query_scalar( "SELECT deletion_state FROM communities WHERE id = $1 AND deleted_at IS NULL", ) @@ -2197,10 +2195,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 \ @@ -2318,10 +2313,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> = sqlx::query_scalar( "UPDATE community_serving_write_leases lease \ SET lease_until = now() + make_interval(secs => $6), heartbeat_at = now() \ @@ -2376,10 +2368,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 \ @@ -2483,6 +2472,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. /// @@ -2491,10 +2508,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(()) } @@ -3131,7 +3151,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, diff --git a/crates/buzz-db/src/lib.rs b/crates/buzz-db/src/lib.rs index a5c03a256ed..04e6f67b334 100644 --- a/crates/buzz-db/src/lib.rs +++ b/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> { - 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> { - 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,21 @@ impl Db { channel_id.as_ref().map(|id| id.as_bytes().as_slice()), ); - let mut tx = self.pool.begin().await?; + let (mut tx, mut transaction_timer) = observability::begin_transaction( + &self.pool, + observability::TransactionOperation::ReplaceAddressableEvent, + ) + .await?; // 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. @@ -4384,6 +4397,7 @@ impl Db { if dominated { tx.rollback().await?; let received_at = chrono::Utc::now(); + transaction_timer.mark_success(); return Ok(( StoredEvent::with_received_at(event.clone(), received_at, channel_id, false), false, @@ -4435,6 +4449,7 @@ impl Db { // ON CONFLICT fired — the event ID already exists. Rollback the // soft-delete so we don't lose the previous replaceable event. tx.rollback().await?; + transaction_timer.mark_success(); return Ok(( StoredEvent::with_received_at(event.clone(), received_at, channel_id, false), false, @@ -4447,6 +4462,7 @@ impl Db { crate::insert_mentions_in_transaction(&mut tx, community_id, event, channel_id).await?; tx.commit().await?; + transaction_timer.mark_success(); Ok(( StoredEvent::with_received_at(event.clone(), received_at, channel_id, true), @@ -4532,16 +4548,23 @@ impl Db { None, ); - let mut tx = self.pool.begin().await?; + let (mut tx, mut transaction_timer) = observability::begin_transaction( + &self.pool, + observability::TransactionOperation::PublishNip43MembershipLocked, + ) + .await?; // 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( @@ -4618,6 +4641,7 @@ impl Db { let was_inserted = insert_result.rows_affected() > 0; if !was_inserted { tx.rollback().await?; + transaction_timer.mark_success(); return Ok(( StoredEvent::with_received_at(event, received_at, None, false), false, @@ -4626,6 +4650,8 @@ impl Db { } tx.commit().await?; + transaction_timer.mark_success(); + drop(transaction_timer); 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}"); diff --git a/crates/buzz-db/src/migration.rs b/crates/buzz-db/src/migration.rs index 94c7aea2faf..9df02c8abf9 100644 --- a/crates/buzz-db/src/migration.rs +++ b/crates/buzz-db/src/migration.rs @@ -81,11 +81,16 @@ where F: FnOnce(PgConnection) -> Fut, Fut: Future)>, { - 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) diff --git a/crates/buzz-db/src/observability.rs b/crates/buzz-db/src/observability.rs new file mode 100644 index 00000000000..4898571c287 --- /dev/null +++ b/crates/buzz-db/src/observability.rs @@ -0,0 +1,580 @@ +//! 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> { + 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(lock_type: LockType, future: F) -> sqlx::Result +where + F: Future>, +{ + 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) fn mark_success(&mut self) { + self.outcome = Outcome::Success; + } +} + +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 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 mut committed = + TransactionTimer::start(TransactionOperation::ReplaceParameterizedEvent); + committed.mark_success(); + drop(committed); + drop(TransactionTimer::start( + TransactionOperation::AcceptPushLeaseEvent, + )); + + 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::>(); + (key.key().name().to_owned(), labels) + }) + .collect::>(); + + 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::>(); + 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>::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::>(); + 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>::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::>(); + 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:?}" + ); + } +} diff --git a/crates/buzz-db/src/push.rs b/crates/buzz-db/src/push.rs index 0b3245ffcc2..623ac1951bb 100644 --- a/crates/buzz-db/src/push.rs +++ b/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,11 @@ pub async fn accept_lease_event( max_active_leases: i64, ) -> Result { let author = event.pubkey.as_bytes(); - let mut tx = pool.begin().await?; + let (mut tx, mut transaction_timer) = crate::observability::begin_transaction( + pool, + crate::observability::TransactionOperation::AcceptPushLeaseEvent, + ) + .await?; 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 +237,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. @@ -256,8 +269,10 @@ pub async fn accept_lease_event( let existing_author: Vec = row.try_get("author")?; let existing_installation: String = row.try_get("installation_id")?; if existing_author.as_slice() != author || existing_installation != installation_id { + transaction_timer.mark_success(); return Ok(AcceptLeaseOutcome::SourceEventCollision); } + transaction_timer.mark_success(); return Ok(AcceptLeaseOutcome::StaleEvent); } @@ -277,9 +292,11 @@ pub async fn accept_lease_event( || (version.source_created_at == current_created_at && version.source_event_id < current_event_id.as_slice()); if !wins_event { + transaction_timer.mark_success(); return Ok(AcceptLeaseOutcome::StaleEvent); } if version.generation <= current_generation { + transaction_timer.mark_success(); return Ok(AcceptLeaseOutcome::StaleGeneration); } } @@ -307,6 +324,7 @@ pub async fn accept_lease_event( .fetch_one(&mut *tx) .await?; if active_count >= max_active_leases { + transaction_timer.mark_success(); return Ok(AcceptLeaseOutcome::LeaseQuotaExceeded); } let duplicate: bool = sqlx::query_scalar( @@ -320,6 +338,7 @@ pub async fn accept_lease_event( .fetch_one(&mut *tx) .await?; if duplicate { + transaction_timer.mark_success(); return Ok(AcceptLeaseOutcome::EndpointAlreadyLeased); } } @@ -349,6 +368,7 @@ pub async fn accept_lease_event( .await { if let Some(outcome) = constraint_acceptance_outcome(&error) { + transaction_timer.mark_success(); return Ok(outcome); } return Err(error.into()); @@ -383,6 +403,7 @@ pub async fn accept_lease_event( .execute(&mut *tx).await { if let Some(outcome) = constraint_acceptance_outcome(&error) { + transaction_timer.mark_success(); return Ok(outcome); } return Err(error.into()); @@ -391,6 +412,7 @@ pub async fn accept_lease_event( backfill_push_match_jobs(&mut tx, community).await?; } tx.commit().await?; + transaction_timer.mark_success(); Ok(AcceptLeaseOutcome::Accepted) } @@ -1271,7 +1293,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"); diff --git a/crates/buzz-db/src/relay_members.rs b/crates/buzz-db/src/relay_members.rs index 402229cdec5..3cb86e8a437 100644 --- a/crates/buzz-db/src/relay_members.rs +++ b/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") diff --git a/crates/buzz-db/src/replaceable.rs b/crates/buzz-db/src/replaceable.rs index aa57a51666d..cb641f0522b 100644 --- a/crates/buzz-db/src/replaceable.rs +++ b/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,7 +399,11 @@ impl Db { d_tag: &str, channel_id: Option, ) -> Result<(StoredEvent, bool)> { - let mut tx = self.pool.begin().await?; + let (mut tx, mut transaction_timer) = observability::begin_transaction( + &self.pool, + TransactionOperation::ReplaceParameterizedEvent, + ) + .await?; let result = self .replace_parameterized_event_in_transaction( &mut tx, @@ -412,6 +420,7 @@ impl Db { } else { tx.rollback().await?; } + transaction_timer.mark_success(); Ok((result.event, was_inserted)) } } diff --git a/crates/buzz-db/tests/observability_source.rs b/crates/buzz-db/tests/observability_source.rs new file mode 100644 index 00000000000..724a9b47ed8 --- /dev/null +++ b/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`. +} diff --git a/crates/buzz-search/Cargo.toml b/crates/buzz-search/Cargo.toml index e28c5b68409..6c6b9ada221 100644 --- a/crates/buzz-search/Cargo.toml +++ b/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 } diff --git a/docs/database-observability.md b/docs/database-observability.md new file mode 100644 index 00000000000..c691f22049f --- /dev/null +++ b/docs/database-observability.md @@ -0,0 +1,40 @@ +# Database pressure observability + +Buzz exposes bounded-cardinality metrics that separate four application-visible components of database pressure. These are measurements only: they do not set PostgreSQL timeouts, retry failed work, resize pools, or change client-visible conflict behavior. + +## Metrics and fixed labels + +All duration values are recorded in seconds. All counter values are monotonically increasing counts. + +| Metric | Unit | Fixed labels | Measurement | +|---|---|---|---| +| `buzz_db_operation_duration_seconds` | seconds | `operation`, `outcome=success|error` | Complete body of a `#[datastore_span]` logical operation. | +| `buzz_db_pool_acquire_wait_seconds` | seconds | `pool_role=writer|reader`, `outcome=success|error|timeout` | Explicit `PgPool::acquire()` wait. | +| `buzz_db_pool_acquisitions_total` | count | `pool_role=writer|reader`, `outcome=success|error|timeout` | Explicit checkout attempts through the same helper. | +| `buzz_db_advisory_lock_wait_seconds` | seconds | `lock_type`, `outcome=success|error|timeout` | Await time for one blocking advisory-lock statement. | +| `buzz_db_advisory_lock_acquisitions_total` | count | `lock_type`, `outcome=success|error|timeout` | Blocking advisory-lock attempts through the same helper. | +| `buzz_db_transaction_duration_seconds` | seconds | `operation`, `outcome=success|error` | Selected transaction lifetimes wholly owned by `buzz-db`. | + +`lock_type` is one of `replacement`, `membership`, `push_gate`, `deletion`, or `migration_schema_safety`. + +Transaction `operation` is one of `replace_parameterized_event`, `replace_addressable_event`, `publish_nip43_membership_locked`, `accept_push_lease_event`, `begin_community_deletion_quiescing`, or `fence_community_deletion`. + +Logical-operation names come from the compile-time string literal required by `#[datastore_span(name = "...", system = "postgresql")]`. The current vocabulary is therefore the finite set of names in those reviewed annotations; adding or changing a series requires a source change and compilation. Names cannot be supplied from request data at runtime. No metric uses community IDs, event IDs, event kinds, coordinates, d-tags, SQL/query text, or query identifiers. + +`timeout` is emitted only when SQLx reports `PoolTimedOut` or a blocking advisory-lock statement returns PostgreSQL SQLSTATE `55P03`. Other failures are `error`. + +## Exact boundaries + +Logical operation duration starts immediately before the annotated function body and stops after it resolves. It can include an implicit SQLx checkout, advisory-lock wait, nested datastore calls, and application-side processing. It is not pure PostgreSQL statement execution time, so nested annotated calls may intentionally produce overlapping samples. A future cancelled before the body resolves does not reach the completion hook and therefore emits neither this duration nor a slow warning. + +Pool acquisition duration covers explicit checkouts routed through the instrumentation helper. Writer coverage includes caller-owned `Db::begin_transaction`, the usage-metrics leader checkout, migration lock checkout, and the selected internally owned transactions listed above. Reader coverage includes the boot reachability checkout and the proved-reader checkout used by replica routing. Passing `&PgPool` directly to SQLx performs an implicit checkout that SQLx does not expose separately at the current API seam; those waits remain folded into logical operation duration. Initial minimum-pool connection establishment is also outside this metric. Reader timeout fallback and writer routing are unchanged. + +Advisory-lock duration wraps only the existing lock statement. The SQL, key, blocking behavior, lock ordering, and transaction/session scope are unchanged. Coverage includes application-side replacement, membership/ownership, push lease/gate, community deletion, and migration/schema-safety locks. Advisory locks taken inside PostgreSQL triggers, stored functions, or migration SQL cannot be timed independently by the application. The channel-TTL transition lock, usage-leadership try-lock, and the separate audit service session lock remain outside the fixed families in this slice. + +Transaction duration starts after `BEGIN` succeeds and stops after explicit commit/rollback completes or the Rust scope exits on error. It excludes pool wait and `BEGIN`, which are represented by the acquisition and logical-operation metrics. On an early return that drops a transaction, the timer stops at scope exit and does not include SQLx's asynchronous rollback cleanup. Caller-owned transactions returned by `Db::begin_transaction` are deliberately not wrapped in a new public transaction type, so their complete lifetime is not measured. + +## Slow-operation warnings + +An annotated logical operation that takes at least 500 ms is eligible for a warning. Sampling is deterministic per call site: the first slow occurrence is logged, followed by one of every 100 slow occurrences. The warning is emitted as a root event so logging formatters cannot attach fields from the surrounding datastore span. It contains only the static `operation`, fixed `outcome`, and integer `elapsed_ms`; it never formats function arguments, returned errors, SQL, or event content. + +These measurements provide the evidence layer that was unavailable when PR #6229 selected timeout policy. They neither duplicate that PR's session settings nor address its separate audit durability/retry review finding. Use measured distributions by workload before changing timeout, retry, pool-budget, or routing policy in later phases. From 6363c3aedf3930e11f88d71c96248e03e6fca272 Mon Sep 17 00:00:00 2001 From: tornquist Date: Mon, 24 Aug 2026 18:34:12 +0000 Subject: [PATCH 2/3] docs: remove database observability guide Signed-off-by: tornquist --- docs/database-observability.md | 40 ---------------------------------- 1 file changed, 40 deletions(-) delete mode 100644 docs/database-observability.md diff --git a/docs/database-observability.md b/docs/database-observability.md deleted file mode 100644 index c691f22049f..00000000000 --- a/docs/database-observability.md +++ /dev/null @@ -1,40 +0,0 @@ -# Database pressure observability - -Buzz exposes bounded-cardinality metrics that separate four application-visible components of database pressure. These are measurements only: they do not set PostgreSQL timeouts, retry failed work, resize pools, or change client-visible conflict behavior. - -## Metrics and fixed labels - -All duration values are recorded in seconds. All counter values are monotonically increasing counts. - -| Metric | Unit | Fixed labels | Measurement | -|---|---|---|---| -| `buzz_db_operation_duration_seconds` | seconds | `operation`, `outcome=success|error` | Complete body of a `#[datastore_span]` logical operation. | -| `buzz_db_pool_acquire_wait_seconds` | seconds | `pool_role=writer|reader`, `outcome=success|error|timeout` | Explicit `PgPool::acquire()` wait. | -| `buzz_db_pool_acquisitions_total` | count | `pool_role=writer|reader`, `outcome=success|error|timeout` | Explicit checkout attempts through the same helper. | -| `buzz_db_advisory_lock_wait_seconds` | seconds | `lock_type`, `outcome=success|error|timeout` | Await time for one blocking advisory-lock statement. | -| `buzz_db_advisory_lock_acquisitions_total` | count | `lock_type`, `outcome=success|error|timeout` | Blocking advisory-lock attempts through the same helper. | -| `buzz_db_transaction_duration_seconds` | seconds | `operation`, `outcome=success|error` | Selected transaction lifetimes wholly owned by `buzz-db`. | - -`lock_type` is one of `replacement`, `membership`, `push_gate`, `deletion`, or `migration_schema_safety`. - -Transaction `operation` is one of `replace_parameterized_event`, `replace_addressable_event`, `publish_nip43_membership_locked`, `accept_push_lease_event`, `begin_community_deletion_quiescing`, or `fence_community_deletion`. - -Logical-operation names come from the compile-time string literal required by `#[datastore_span(name = "...", system = "postgresql")]`. The current vocabulary is therefore the finite set of names in those reviewed annotations; adding or changing a series requires a source change and compilation. Names cannot be supplied from request data at runtime. No metric uses community IDs, event IDs, event kinds, coordinates, d-tags, SQL/query text, or query identifiers. - -`timeout` is emitted only when SQLx reports `PoolTimedOut` or a blocking advisory-lock statement returns PostgreSQL SQLSTATE `55P03`. Other failures are `error`. - -## Exact boundaries - -Logical operation duration starts immediately before the annotated function body and stops after it resolves. It can include an implicit SQLx checkout, advisory-lock wait, nested datastore calls, and application-side processing. It is not pure PostgreSQL statement execution time, so nested annotated calls may intentionally produce overlapping samples. A future cancelled before the body resolves does not reach the completion hook and therefore emits neither this duration nor a slow warning. - -Pool acquisition duration covers explicit checkouts routed through the instrumentation helper. Writer coverage includes caller-owned `Db::begin_transaction`, the usage-metrics leader checkout, migration lock checkout, and the selected internally owned transactions listed above. Reader coverage includes the boot reachability checkout and the proved-reader checkout used by replica routing. Passing `&PgPool` directly to SQLx performs an implicit checkout that SQLx does not expose separately at the current API seam; those waits remain folded into logical operation duration. Initial minimum-pool connection establishment is also outside this metric. Reader timeout fallback and writer routing are unchanged. - -Advisory-lock duration wraps only the existing lock statement. The SQL, key, blocking behavior, lock ordering, and transaction/session scope are unchanged. Coverage includes application-side replacement, membership/ownership, push lease/gate, community deletion, and migration/schema-safety locks. Advisory locks taken inside PostgreSQL triggers, stored functions, or migration SQL cannot be timed independently by the application. The channel-TTL transition lock, usage-leadership try-lock, and the separate audit service session lock remain outside the fixed families in this slice. - -Transaction duration starts after `BEGIN` succeeds and stops after explicit commit/rollback completes or the Rust scope exits on error. It excludes pool wait and `BEGIN`, which are represented by the acquisition and logical-operation metrics. On an early return that drops a transaction, the timer stops at scope exit and does not include SQLx's asynchronous rollback cleanup. Caller-owned transactions returned by `Db::begin_transaction` are deliberately not wrapped in a new public transaction type, so their complete lifetime is not measured. - -## Slow-operation warnings - -An annotated logical operation that takes at least 500 ms is eligible for a warning. Sampling is deterministic per call site: the first slow occurrence is logged, followed by one of every 100 slow occurrences. The warning is emitted as a root event so logging formatters cannot attach fields from the surrounding datastore span. It contains only the static `operation`, fixed `outcome`, and integer `elapsed_ms`; it never formats function arguments, returned errors, SQL, or event content. - -These measurements provide the evidence layer that was unavailable when PR #6229 selected timeout policy. They neither duplicate that PR's session settings nor address its separate audit durability/retry review finding. Use measured distributions by workload before changing timeout, retry, pool-budget, or routing policy in later phases. From d75c4e5733b4bf22292703304a8bad26ea62789c Mon Sep 17 00:00:00 2001 From: tornquist Date: Tue, 25 Aug 2026 15:00:02 +0000 Subject: [PATCH 3/3] Harden transaction observability coverage Signed-off-by: tornquist --- .github/workflows/ci.yml | 12 +++++ crates/buzz-db/src/deletion.rs | 14 ++++-- crates/buzz-db/src/lib.rs | 40 ++++++++-------- crates/buzz-db/src/observability.rs | 74 +++++++++++++++++++++++++---- crates/buzz-db/src/push.rs | 15 ++---- crates/buzz-db/src/replaceable.rs | 41 ++++++++-------- 6 files changed, 134 insertions(+), 62 deletions(-) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 80167668121..1fa6fb94ad6 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -692,6 +692,18 @@ jobs: env: DATABASE_URL: postgres://buzz:${{ env.BUZZ_TEST_POSTGRES_PASSWORD }}@localhost:5432/buzz TEST_DATABASE_URL: postgres://buzz:${{ env.BUZZ_TEST_POSTGRES_PASSWORD }}@localhost:5432/buzz + - name: Database pressure observability PostgreSQL tests + # Explicit pool acquisition and advisory-lock metrics require real + # Postgres and are ignored by the infrastructure-free unit-test job. + run: | + filter='package(buzz-db) and test(/observability::tests::(pool_acquire_records_success_timeout_and_error_with_wait_time|advisory_lock_records_success_contention_timeout_and_error)/)' + cargo nextest run \ + --archive-file target/ci/backend-integration-tests.tar.zst \ + -E "${filter}" \ + --run-ignored ignored-only + env: + DATABASE_URL: postgres://buzz:${{ env.BUZZ_TEST_POSTGRES_PASSWORD }}@localhost:5432/buzz + TEST_DATABASE_URL: postgres://buzz:${{ env.BUZZ_TEST_POSTGRES_PASSWORD }}@localhost:5432/buzz - name: Start relay run: | chmod +x ./target/ci/buzz-relay diff --git a/crates/buzz-db/src/deletion.rs b/crates/buzz-db/src/deletion.rs index 0751cd55ffb..b0efcd67438 100644 --- a/crates/buzz-db/src/deletion.rs +++ b/crates/buzz-db/src/deletion.rs @@ -1121,11 +1121,13 @@ 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, mut transaction_timer) = crate::observability::begin_transaction( + 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?; lock_community_deletion(&mut tx, token.community_id).await?; verify_lease(&mut tx, token, DeletionStage::Approved).await?; @@ -1170,17 +1172,20 @@ impl DeletionStore { ) .await?; tx.commit().await?; - transaction_timer.mark_success(); Ok(()) + }) + .await } /// Acquire the universal durable fence after all pre-quiesce serving leases drain. pub async fn fence(&self, token: &LeaseToken) -> Result { - let (mut tx, mut transaction_timer) = crate::observability::begin_transaction( + 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?; lock_community_deletion(&mut tx, token.community_id).await?; verify_lease(&mut tx, token, DeletionStage::Approved).await?; @@ -1244,8 +1249,9 @@ impl DeletionStore { ) .await?; tx.commit().await?; - transaction_timer.mark_success(); Ok(generation) + }) + .await } /// Freeze the exact post-fence storage binding manifest. diff --git a/crates/buzz-db/src/lib.rs b/crates/buzz-db/src/lib.rs index 04e6f67b334..a6cd77c9796 100644 --- a/crates/buzz-db/src/lib.rs +++ b/crates/buzz-db/src/lib.rs @@ -4355,11 +4355,13 @@ impl Db { channel_id.as_ref().map(|id| id.as_bytes().as_slice()), ); - let (mut tx, mut transaction_timer) = observability::begin_transaction( + 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. @@ -4397,7 +4399,6 @@ impl Db { if dominated { tx.rollback().await?; let received_at = chrono::Utc::now(); - transaction_timer.mark_success(); return Ok(( StoredEvent::with_received_at(event.clone(), received_at, channel_id, false), false, @@ -4449,7 +4450,6 @@ impl Db { // ON CONFLICT fired — the event ID already exists. Rollback the // soft-delete so we don't lose the previous replaceable event. tx.rollback().await?; - transaction_timer.mark_success(); return Ok(( StoredEvent::with_received_at(event.clone(), received_at, channel_id, false), false, @@ -4462,12 +4462,13 @@ impl Db { crate::insert_mentions_in_transaction(&mut tx, community_id, event, channel_id).await?; tx.commit().await?; - transaction_timer.mark_success(); Ok(( 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 @@ -4548,11 +4549,13 @@ impl Db { None, ); - let (mut tx, mut transaction_timer) = observability::begin_transaction( + 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 @@ -4639,27 +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?; - transaction_timer.mark_success(); - 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?; - transaction_timer.mark_success(); - drop(transaction_timer); - - 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, )) } diff --git a/crates/buzz-db/src/observability.rs b/crates/buzz-db/src/observability.rs index 4898571c287..afe1d20b305 100644 --- a/crates/buzz-db/src/observability.rs +++ b/crates/buzz-db/src/observability.rs @@ -195,8 +195,15 @@ impl TransactionTimer { } } - pub(crate) fn mark_success(&mut self) { - self.outcome = Outcome::Success; + pub(crate) async fn observe(mut self, future: F) -> Result + where + F: Future>, + { + let result = future.await; + if result.is_ok() { + self.outcome = Outcome::Success; + } + result } } @@ -251,6 +258,52 @@ mod tests { ); } + #[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::>(); + (key.key().name().to_owned(), labels) + }) + .collect::>(); + + 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(); @@ -275,13 +328,16 @@ mod tests { .await; assert!(lock_error.is_err()); - let mut committed = - TransactionTimer::start(TransactionOperation::ReplaceParameterizedEvent); - committed.mark_success(); - drop(committed); - drop(TransactionTimer::start( - TransactionOperation::AcceptPushLeaseEvent, - )); + 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 diff --git a/crates/buzz-db/src/push.rs b/crates/buzz-db/src/push.rs index 623ac1951bb..3aa6cd9b3fe 100644 --- a/crates/buzz-db/src/push.rs +++ b/crates/buzz-db/src/push.rs @@ -223,11 +223,13 @@ pub async fn accept_lease_event( max_active_leases: i64, ) -> Result { let author = event.pubkey.as_bytes(); - let (mut tx, mut transaction_timer) = crate::observability::begin_transaction( + 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); @@ -269,10 +271,8 @@ pub async fn accept_lease_event( let existing_author: Vec = row.try_get("author")?; let existing_installation: String = row.try_get("installation_id")?; if existing_author.as_slice() != author || existing_installation != installation_id { - transaction_timer.mark_success(); return Ok(AcceptLeaseOutcome::SourceEventCollision); } - transaction_timer.mark_success(); return Ok(AcceptLeaseOutcome::StaleEvent); } @@ -292,11 +292,9 @@ pub async fn accept_lease_event( || (version.source_created_at == current_created_at && version.source_event_id < current_event_id.as_slice()); if !wins_event { - transaction_timer.mark_success(); return Ok(AcceptLeaseOutcome::StaleEvent); } if version.generation <= current_generation { - transaction_timer.mark_success(); return Ok(AcceptLeaseOutcome::StaleGeneration); } } @@ -324,7 +322,6 @@ pub async fn accept_lease_event( .fetch_one(&mut *tx) .await?; if active_count >= max_active_leases { - transaction_timer.mark_success(); return Ok(AcceptLeaseOutcome::LeaseQuotaExceeded); } let duplicate: bool = sqlx::query_scalar( @@ -338,7 +335,6 @@ pub async fn accept_lease_event( .fetch_one(&mut *tx) .await?; if duplicate { - transaction_timer.mark_success(); return Ok(AcceptLeaseOutcome::EndpointAlreadyLeased); } } @@ -368,7 +364,6 @@ pub async fn accept_lease_event( .await { if let Some(outcome) = constraint_acceptance_outcome(&error) { - transaction_timer.mark_success(); return Ok(outcome); } return Err(error.into()); @@ -403,7 +398,6 @@ pub async fn accept_lease_event( .execute(&mut *tx).await { if let Some(outcome) = constraint_acceptance_outcome(&error) { - transaction_timer.mark_success(); return Ok(outcome); } return Err(error.into()); @@ -412,8 +406,9 @@ pub async fn accept_lease_event( backfill_push_match_jobs(&mut tx, community).await?; } tx.commit().await?; - transaction_timer.mark_success(); Ok(AcceptLeaseOutcome::Accepted) + }) + .await } fn constraint_acceptance_outcome(error: &sqlx::Error) -> Option { diff --git a/crates/buzz-db/src/replaceable.rs b/crates/buzz-db/src/replaceable.rs index cb641f0522b..4904dfe209d 100644 --- a/crates/buzz-db/src/replaceable.rs +++ b/crates/buzz-db/src/replaceable.rs @@ -399,28 +399,31 @@ impl Db { d_tag: &str, channel_id: Option, ) -> Result<(StoredEvent, bool)> { - let (mut tx, mut transaction_timer) = observability::begin_transaction( + let (mut tx, transaction_timer) = observability::begin_transaction( &self.pool, TransactionOperation::ReplaceParameterizedEvent, ) .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?; - } - transaction_timer.mark_success(); - Ok((result.event, was_inserted)) + 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 } }