From 36c9685488d3f13eff2b15a3a1d0f39cb2acbffb Mon Sep 17 00:00:00 2001 From: Leonid Ryzhyk Date: Fri, 14 Aug 2026 12:02:22 -0700 Subject: [PATCH 1/2] pipeline-manager: set the row lock timeout per transaction Every transaction now applies `lock_timeout` itself with `SET LOCAL`, rather than the connection carrying it as an `options` startup parameter. The startup parameter locks the manager out of any database reached through PgBouncer, which rejects startup parameters it does not know: FATAL: unsupported startup parameter: options Neither the pool nor the separate connection that listens on the pipeline table can be opened, so the manager cannot reach the database at all. `SET LOCAL` holds for a direct connection and for every pooling mode: a pooler keeps a transaction on one server connection from BEGIN to COMMIT, and Postgres reverts the setting at commit, so it cannot leak to the next client of that connection. It costs one round trip per transaction, and `db::transaction` documents the trade-off. `db::transaction::{begin, begin_read_only}` are now the only places that start a transaction, which keeps a call site from ending up without the timeout. The timeout itself is unchanged at 10 seconds. One intended consequence: the transactions refinery runs for the migrations no longer have a lock timeout. That is what a rolling upgrade wants, as an instance starting up then waits for the migration of another one to finish rather than failing to start. Signed-off-by: Leonid Ryzhyk --- .../src/api/support_data_collector.rs | 13 +- crates/pipeline-manager/src/db.rs | 1 + .../src/db/storage_postgres.rs | 241 ++++++++---------- crates/pipeline-manager/src/db/test.rs | 79 +++++- crates/pipeline-manager/src/db/transaction.rs | 68 +++++ 5 files changed, 244 insertions(+), 158 deletions(-) create mode 100644 crates/pipeline-manager/src/db/transaction.rs diff --git a/crates/pipeline-manager/src/api/support_data_collector.rs b/crates/pipeline-manager/src/api/support_data_collector.rs index 1c179c8f1f1..55317a85535 100644 --- a/crates/pipeline-manager/src/api/support_data_collector.rs +++ b/crates/pipeline-manager/src/api/support_data_collector.rs @@ -7,6 +7,7 @@ use crate::db::operations::pipeline::{ cleanup_old_support_data_collections, store_support_data_collection, }; use crate::db::storage::Storage; +use crate::db::transaction; use crate::db::types::combined_status::CombinedStatus; use crate::db::types::pipeline::PipelineId; use crate::db::types::tenant::TenantId; @@ -1324,7 +1325,7 @@ impl SupportDataCollector { ) -> Result<(), Box> { // Get a client from the pool and create a transaction let mut client = self.state.db.lock().await.pool.get().await?; - let txn = client.transaction().await?; + let txn = transaction::begin(&mut client).await?; store_support_data_collection(&txn, pipeline_id, tenant_id, support_bundle).await?; @@ -1338,7 +1339,7 @@ impl SupportDataCollector { pipeline_id: PipelineId, ) -> Result<(), Box> { let mut client = self.state.db.lock().await.pool.get().await?; - let txn = client.transaction().await?; + let txn = transaction::begin(&mut client).await?; let r = cleanup_old_support_data_collections(&txn, pipeline_id, self.retention_count as i64) .await?; @@ -1716,7 +1717,7 @@ mod tests { // Verify support bundle data was stored let mut client = db.lock().await.pool.get().await.unwrap(); - let txn = client.transaction().await.unwrap(); + let txn = transaction::begin(&mut client).await.unwrap(); let bundles = crate::db::operations::pipeline::get_support_bundle_data(&txn, pipeline_id, 10) .await @@ -1764,7 +1765,7 @@ mod tests { // Verify support bundle data was deleted (due to CASCADE) let mut client = db.lock().await.pool.get().await.unwrap(); - let txn = client.transaction().await.unwrap(); + let txn = transaction::begin(&mut client).await.unwrap(); let bundles = crate::db::operations::pipeline::get_support_bundle_data(&txn, pipeline_id, 10) .await @@ -1913,7 +1914,7 @@ mod tests { // Verify we have 5 collections initially let mut client = db.lock().await.pool.get().await.unwrap(); - let txn = client.transaction().await.unwrap(); + let txn = transaction::begin(&mut client).await.unwrap(); let bundles = crate::db::operations::pipeline::get_support_bundle_data(&txn, pipeline_id, 10) .await @@ -1929,7 +1930,7 @@ mod tests { // Verify only 3 collections remain (retention_count) let mut client = db.lock().await.pool.get().await.unwrap(); - let txn = client.transaction().await.unwrap(); + let txn = transaction::begin(&mut client).await.unwrap(); let bundles = crate::db::operations::pipeline::get_support_bundle_data( &txn, pipeline_id, diff --git a/crates/pipeline-manager/src/db.rs b/crates/pipeline-manager/src/db.rs index 44a0b858d2f..6a983f03bfa 100644 --- a/crates/pipeline-manager/src/db.rs +++ b/crates/pipeline-manager/src/db.rs @@ -8,4 +8,5 @@ pub(crate) mod storage; pub mod storage_postgres; #[cfg(test)] pub(crate) mod test; +pub(crate) mod transaction; pub mod types; diff --git a/crates/pipeline-manager/src/db/storage_postgres.rs b/crates/pipeline-manager/src/db/storage_postgres.rs index f621abbcecc..e7f266a5121 100644 --- a/crates/pipeline-manager/src/db/storage_postgres.rs +++ b/crates/pipeline-manager/src/db/storage_postgres.rs @@ -6,6 +6,7 @@ use crate::db::operations; #[cfg(feature = "postgresql_embedded")] use crate::db::pg_setup; use crate::db::storage::{ExtendedPipelineDescrRunner, Storage}; +use crate::db::transaction; use crate::db::types::api_key::ApiKeyDescr; use crate::db::types::monitor::{ ClusterMonitorEvent, ClusterMonitorEventId, ExtendedClusterMonitorEvent, @@ -36,7 +37,7 @@ use deadpool_postgres::{Manager, Pool, RecyclingMethod}; use feldera_types::config::{PipelineConfig, RuntimeConfig}; use feldera_types::error::ErrorResponse; use feldera_types::runtime_status::{BootstrapConfig, RuntimeDesiredStatus, RuntimeStatus}; -use tokio_postgres::{IsolationLevel, Row}; +use tokio_postgres::Row; use tracing::{debug, info}; use uuid::Uuid; @@ -92,7 +93,7 @@ pub struct StoragePostgres { impl Storage for StoragePostgres { async fn check_connection(&self) -> Result<(), DBError> { let mut client = self.pool.get().await?; - let txn = client.transaction().await?; + let txn = transaction::begin(&mut client).await?; operations::connectivity::check_connection(&txn).await?; txn.commit().await?; Ok(()) @@ -105,7 +106,7 @@ impl Storage for StoragePostgres { provider: String, ) -> Result { let mut client = self.pool.get().await?; - let txn = client.transaction().await?; + let txn = transaction::begin(&mut client).await?; let tenant_id = operations::tenant::get_or_create_tenant_id(&txn, new_id, name, provider).await?; txn.commit().await?; @@ -114,7 +115,7 @@ impl Storage for StoragePostgres { async fn get_tenant_name(&self, tenant_id: TenantId) -> Result { let mut client = self.pool.get().await?; - let txn = client.transaction().await?; + let txn = transaction::begin(&mut client).await?; let tenant_name = operations::tenant::get_tenant_name(&txn, tenant_id).await?; txn.commit().await?; Ok(tenant_name) @@ -122,7 +123,7 @@ impl Storage for StoragePostgres { async fn resolve_tenant_selector(&self, selector: &str) -> Result { let mut client = self.pool.get().await?; - let txn = client.transaction().await?; + let txn = transaction::begin(&mut client).await?; let result = operations::tenant::resolve_tenant_selector(&txn, selector).await?; txn.commit().await?; Ok(result) @@ -130,7 +131,7 @@ impl Storage for StoragePostgres { async fn get_tenant(&self, selector: &str) -> Result { let mut client = self.pool.get().await?; - let txn = client.transaction().await?; + let txn = transaction::begin(&mut client).await?; let result = operations::tenant::get_tenant(&txn, selector).await?; txn.commit().await?; Ok(result) @@ -138,7 +139,7 @@ impl Storage for StoragePostgres { async fn find_tenant_id_by_name(&self, name: &str) -> Result, DBError> { let mut client = self.pool.get().await?; - let txn = client.transaction().await?; + let txn = transaction::begin(&mut client).await?; let result = operations::tenant::find_tenant_id_by_name(&txn, name).await?; txn.commit().await?; Ok(result) @@ -151,7 +152,7 @@ impl Storage for StoragePostgres { provider: &str, ) -> Result<(TenantInfo, bool), DBError> { let mut client = self.pool.get().await?; - let txn = client.transaction().await?; + let txn = transaction::begin(&mut client).await?; let result = operations::tenant::get_or_create_tenant(&txn, new_id, name, provider).await?; txn.commit().await?; Ok(result) @@ -159,7 +160,7 @@ impl Storage for StoragePostgres { async fn delete_tenant(&self, tenant_id: TenantId) -> Result<(), DBError> { let mut client = self.pool.get().await?; - let txn = client.transaction().await?; + let txn = transaction::begin(&mut client).await?; operations::tenant::delete_tenant(&txn, tenant_id).await?; txn.commit().await?; Ok(()) @@ -172,7 +173,7 @@ impl Storage for StoragePostgres { displace_existing: bool, ) -> Result, DBError> { let mut client = self.pool.get().await?; - let txn = client.transaction().await?; + let txn = transaction::begin(&mut client).await?; let displaced = operations::tenant::rename_tenant(&txn, tenant_id, new_name, displace_existing).await?; txn.commit().await?; @@ -181,7 +182,7 @@ impl Storage for StoragePostgres { async fn list_tenants(&self) -> Result, DBError> { let mut client = self.pool.get().await?; - let txn = client.transaction().await?; + let txn = transaction::begin(&mut client).await?; let result = operations::tenant::list_tenants(&txn).await?; txn.commit().await?; Ok(result) @@ -201,7 +202,7 @@ impl Storage for StoragePostgres { origin: MembershipOrigin, ) -> Result<(TenantId, UserId, Role), DBError> { let mut client = self.pool.get().await?; - let txn = client.transaction().await?; + let txn = transaction::begin(&mut client).await?; let result = operations::user::resolve_login( &txn, new_tenant_id, @@ -225,7 +226,7 @@ impl Storage for StoragePostgres { subject: &str, ) -> Result, DBError> { let mut client = self.pool.get().await?; - let txn = client.transaction().await?; + let txn = transaction::begin(&mut client).await?; let result = operations::user::list_user_memberships(&txn, provider, subject).await?; txn.commit().await?; Ok(result) @@ -242,7 +243,7 @@ impl Storage for StoragePostgres { origin: MembershipOrigin, ) -> Result<(), DBError> { let mut client = self.pool.get().await?; - let txn = client.transaction().await?; + let txn = transaction::begin(&mut client).await?; operations::user::enroll_in_existing_tenants( &txn, new_user_id, @@ -266,7 +267,7 @@ impl Storage for StoragePostgres { email: Option<&str>, ) -> Result { let mut client = self.pool.get().await?; - let txn = client.transaction().await?; + let txn = transaction::begin(&mut client).await?; let result = operations::user::get_or_create_user(&txn, new_id, provider, subject, email).await?; txn.commit().await?; @@ -282,7 +283,7 @@ impl Storage for StoragePostgres { ttl_seconds: i64, ) -> Result { let mut client = self.pool.get().await?; - let txn = client.transaction().await?; + let txn = transaction::begin(&mut client).await?; let result = operations::user::claim_profile_refresh( &txn, new_id, @@ -303,7 +304,7 @@ impl Storage for StoragePostgres { profile: &UserProfile, ) -> Result<(), DBError> { let mut client = self.pool.get().await?; - let txn = client.transaction().await?; + let txn = transaction::begin(&mut client).await?; operations::user::store_user_profile(&txn, provider, subject, profile).await?; txn.commit().await?; Ok(()) @@ -311,7 +312,7 @@ impl Storage for StoragePostgres { async fn list_tenant_members(&self, tenant_id: TenantId) -> Result, DBError> { let mut client = self.pool.get().await?; - let txn = client.transaction().await?; + let txn = transaction::begin(&mut client).await?; let result = operations::user::list_tenant_members(&txn, tenant_id).await?; txn.commit().await?; Ok(result) @@ -325,7 +326,7 @@ impl Storage for StoragePostgres { origin: MembershipOrigin, ) -> Result<(), DBError> { let mut client = self.pool.get().await?; - let txn = client.transaction().await?; + let txn = transaction::begin(&mut client).await?; operations::user::upsert_member_role(&txn, tenant_id, user_id, role, origin).await?; txn.commit().await?; Ok(()) @@ -341,7 +342,7 @@ impl Storage for StoragePostgres { role: Role, ) -> Result { let mut client = self.pool.get().await?; - let txn = client.transaction().await?; + let txn = transaction::begin(&mut client).await?; let result = operations::user::preprovision_member( &txn, new_user_id, @@ -358,7 +359,7 @@ impl Storage for StoragePostgres { async fn remove_member(&self, tenant_id: TenantId, user_id: UserId) -> Result<(), DBError> { let mut client = self.pool.get().await?; - let txn = client.transaction().await?; + let txn = transaction::begin(&mut client).await?; operations::user::remove_member(&txn, tenant_id, user_id).await?; txn.commit().await?; Ok(()) @@ -366,7 +367,7 @@ impl Storage for StoragePostgres { async fn list_api_keys(&self, tenant_id: TenantId) -> Result, DBError> { let mut client = self.pool.get().await?; - let txn = client.transaction().await?; + let txn = transaction::begin(&mut client).await?; let api_keys = operations::api_key::list_api_keys(&txn, tenant_id).await?; txn.commit().await?; Ok(api_keys) @@ -374,7 +375,7 @@ impl Storage for StoragePostgres { async fn get_api_key(&self, tenant_id: TenantId, name: &str) -> Result { let mut client = self.pool.get().await?; - let txn = client.transaction().await?; + let txn = transaction::begin(&mut client).await?; let api_key = operations::api_key::get_api_key(&txn, tenant_id, name).await?; txn.commit().await?; Ok(api_key) @@ -382,7 +383,7 @@ impl Storage for StoragePostgres { async fn delete_api_key(&self, tenant_id: TenantId, name: &str) -> Result<(), DBError> { let mut client = self.pool.get().await?; - let txn = client.transaction().await?; + let txn = transaction::begin(&mut client).await?; let result = operations::api_key::delete_api_key(&txn, tenant_id, name).await?; txn.commit().await?; Ok(result) @@ -397,7 +398,7 @@ impl Storage for StoragePostgres { role: MintableKeyRole, ) -> Result<(), DBError> { let mut client = self.pool.get().await?; - let txn = client.transaction().await?; + let txn = transaction::begin(&mut client).await?; let result = operations::api_key::store_api_key_hash(&txn, tenant_id, id, name, key, role).await?; txn.commit().await?; @@ -406,7 +407,7 @@ impl Storage for StoragePostgres { async fn validate_api_key(&self, key: &str) -> Result<(TenantId, Role), DBError> { let mut client = self.pool.get().await?; - let txn = client.transaction().await?; + let txn = transaction::begin(&mut client).await?; let result = operations::api_key::validate_api_key(&txn, key).await?; txn.commit().await?; Ok(result) @@ -414,7 +415,7 @@ impl Storage for StoragePostgres { async fn list_oidc_trust(&self, tenant_id: TenantId) -> Result, DBError> { let mut client = self.pool.get().await?; - let txn = client.transaction().await?; + let txn = transaction::begin(&mut client).await?; let result = operations::oidc_trust::list_oidc_trust(&txn, tenant_id).await?; txn.commit().await?; Ok(result) @@ -426,7 +427,7 @@ impl Storage for StoragePostgres { name: &str, ) -> Result { let mut client = self.pool.get().await?; - let txn = client.transaction().await?; + let txn = transaction::begin(&mut client).await?; let result = operations::oidc_trust::get_oidc_trust(&txn, tenant_id, name).await?; txn.commit().await?; Ok(result) @@ -434,7 +435,7 @@ impl Storage for StoragePostgres { async fn delete_oidc_trust(&self, tenant_id: TenantId, name: &str) -> Result<(), DBError> { let mut client = self.pool.get().await?; - let txn = client.transaction().await?; + let txn = transaction::begin(&mut client).await?; let result = operations::oidc_trust::delete_oidc_trust(&txn, tenant_id, name).await?; txn.commit().await?; Ok(result) @@ -453,7 +454,7 @@ impl Storage for StoragePostgres { issuer_policy: TenantIssuerPolicy, ) -> Result<(), DBError> { let mut client = self.pool.get().await?; - let txn = client.transaction().await?; + let txn = transaction::begin(&mut client).await?; operations::oidc_trust::create_oidc_trust( &txn, tenant_id, @@ -473,7 +474,7 @@ impl Storage for StoragePostgres { async fn is_trusted_issuer(&self, issuer: &str) -> Result { let mut client = self.pool.get().await?; - let txn = client.transaction().await?; + let txn = transaction::begin(&mut client).await?; let result = operations::oidc_trust::is_trusted_issuer(&txn, issuer).await?; txn.commit().await?; Ok(result) @@ -486,7 +487,7 @@ impl Storage for StoragePostgres { audiences: &[String], ) -> Result, DBError> { let mut client = self.pool.get().await?; - let txn = client.transaction().await?; + let txn = transaction::begin(&mut client).await?; let result = operations::oidc_trust::match_oidc_trust(&txn, issuer, subject, audiences).await?; txn.commit().await?; @@ -498,7 +499,7 @@ impl Storage for StoragePostgres { tenant_id: TenantId, ) -> Result, DBError> { let mut client = self.pool.get().await?; - let txn = client.transaction().await?; + let txn = transaction::begin(&mut client).await?; let pipelines = operations::pipeline::list_pipelines(&txn, tenant_id).await?; txn.commit().await?; Ok(pipelines) @@ -509,7 +510,7 @@ impl Storage for StoragePostgres { tenant_id: TenantId, ) -> Result, DBError> { let mut client = self.pool.get().await?; - let txn = client.transaction().await?; + let txn = transaction::begin(&mut client).await?; let pipelines = operations::pipeline::list_pipelines_for_monitoring(&txn, tenant_id).await?; txn.commit().await?; @@ -522,7 +523,7 @@ impl Storage for StoragePostgres { name: &str, ) -> Result { let mut client = self.pool.get().await?; - let txn = client.transaction().await?; + let txn = transaction::begin(&mut client).await?; let pipeline = operations::pipeline::get_pipeline(&txn, tenant_id, name, false).await?; txn.commit().await?; Ok(pipeline) @@ -534,7 +535,7 @@ impl Storage for StoragePostgres { name: &str, ) -> Result { let mut client = self.pool.get().await?; - let txn = client.transaction().await?; + let txn = transaction::begin(&mut client).await?; let pipeline = operations::pipeline::get_pipeline_for_monitoring(&txn, tenant_id, name, false).await?; txn.commit().await?; @@ -547,7 +548,7 @@ impl Storage for StoragePostgres { pipeline_id: PipelineId, ) -> Result { let mut client = self.pool.get().await?; - let txn = client.transaction().await?; + let txn = transaction::begin(&mut client).await?; let pipeline = operations::pipeline::get_pipeline_by_id(&txn, tenant_id, pipeline_id, false).await?; txn.commit().await?; @@ -560,7 +561,7 @@ impl Storage for StoragePostgres { pipeline_id: PipelineId, ) -> Result { let mut client = self.pool.get().await?; - let txn = client.transaction().await?; + let txn = transaction::begin(&mut client).await?; let pipeline = operations::pipeline::get_pipeline_by_id_for_monitoring( &txn, tenant_id, @@ -580,12 +581,7 @@ impl Storage for StoragePostgres { provision_called: bool, ) -> Result { let mut client = self.pool.get().await?; - let txn = client - .build_transaction() - .isolation_level(IsolationLevel::RepeatableRead) - .read_only(true) - .start() - .await?; + let txn = transaction::begin_read_only(&mut client).await?; let pipeline_monitoring = operations::pipeline::get_pipeline_by_id_for_monitoring( &txn, tenant_id, @@ -633,7 +629,7 @@ impl Storage for StoragePostgres { pipeline: PipelineDescr, ) -> Result { let mut client = self.pool.get().await?; - let txn = client.transaction().await?; + let txn = transaction::begin(&mut client).await?; // Create new pipeline operations::pipeline::new_pipeline( @@ -663,7 +659,7 @@ impl Storage for StoragePostgres { pipeline: PipelineDescr, ) -> Result<(bool, ExtendedPipelineDescr), DBError> { let mut client = self.pool.get().await?; - let txn = client.transaction().await?; + let txn = transaction::begin(&mut client).await?; // Check if pipeline exists let current = @@ -732,7 +728,7 @@ impl Storage for StoragePostgres { platform_version: &str, ) -> Result<(), DBError> { let mut client = self.pool.get().await?; - let txn = client.transaction().await?; + let txn = transaction::begin(&mut client).await?; // Check if pipeline exists let current = @@ -789,7 +785,7 @@ impl Storage for StoragePostgres { program_config: &Option, ) -> Result { let mut client = self.pool.get().await?; - let txn = client.transaction().await?; + let txn = transaction::begin(&mut client).await?; // Update existing pipeline operations::pipeline::update_pipeline( @@ -826,7 +822,7 @@ impl Storage for StoragePostgres { pipeline_name: &str, ) -> Result { let mut client = self.pool.get().await?; - let txn = client.transaction().await?; + let txn = transaction::begin(&mut client).await?; let pipeline_id = operations::pipeline::delete_pipeline(&txn, tenant_id, pipeline_name).await?; txn.commit().await?; @@ -840,7 +836,7 @@ impl Storage for StoragePostgres { program_version_guard: Version, ) -> Result<(), DBError> { let mut client = self.pool.get().await?; - let txn = client.transaction().await?; + let txn = transaction::begin(&mut client).await?; operations::pipeline::set_program_status( &txn, tenant_id, @@ -867,7 +863,7 @@ impl Storage for StoragePostgres { program_version_guard: Version, ) -> Result<(), DBError> { let mut client = self.pool.get().await?; - let txn = client.transaction().await?; + let txn = transaction::begin(&mut client).await?; operations::pipeline::set_program_status( &txn, tenant_id, @@ -896,7 +892,7 @@ impl Storage for StoragePostgres { program_info: &serde_json::Value, ) -> Result<(), DBError> { let mut client = self.pool.get().await?; - let txn = client.transaction().await?; + let txn = transaction::begin(&mut client).await?; operations::pipeline::set_program_status( &txn, tenant_id, @@ -923,7 +919,7 @@ impl Storage for StoragePostgres { program_version_guard: Version, ) -> Result<(), DBError> { let mut client = self.pool.get().await?; - let txn = client.transaction().await?; + let txn = transaction::begin(&mut client).await?; operations::pipeline::set_program_status( &txn, tenant_id, @@ -954,7 +950,7 @@ impl Storage for StoragePostgres { program_info_integrity_checksum: &str, ) -> Result<(), DBError> { let mut client = self.pool.get().await?; - let txn = client.transaction().await?; + let txn = transaction::begin(&mut client).await?; operations::pipeline::set_program_status( &txn, tenant_id, @@ -982,7 +978,7 @@ impl Storage for StoragePostgres { sql_compilation: &SqlCompilationInfo, ) -> Result<(), DBError> { let mut client = self.pool.get().await?; - let txn = client.transaction().await?; + let txn = transaction::begin(&mut client).await?; operations::pipeline::set_program_status( &txn, tenant_id, @@ -1010,7 +1006,7 @@ impl Storage for StoragePostgres { rust_compilation: &RustCompilationInfo, ) -> Result<(), DBError> { let mut client = self.pool.get().await?; - let txn = client.transaction().await?; + let txn = transaction::begin(&mut client).await?; operations::pipeline::set_program_status( &txn, tenant_id, @@ -1038,7 +1034,7 @@ impl Storage for StoragePostgres { system_error: &str, ) -> Result<(), DBError> { let mut client = self.pool.get().await?; - let txn = client.transaction().await?; + let txn = transaction::begin(&mut client).await?; operations::pipeline::set_program_status( &txn, tenant_id, @@ -1064,7 +1060,7 @@ impl Storage for StoragePostgres { pipeline_name: &str, ) -> Result<(), DBError> { let mut client = self.pool.get().await?; - let txn = client.transaction().await?; + let txn = transaction::begin(&mut client).await?; operations::pipeline::dismiss_deployment_error(&txn, tenant_id, pipeline_name).await?; txn.commit().await?; Ok(()) @@ -1079,7 +1075,7 @@ impl Storage for StoragePostgres { dismiss_error: bool, ) -> Result { let mut client = self.pool.get().await?; - let txn = client.transaction().await?; + let txn = transaction::begin(&mut client).await?; let pipeline_id = operations::pipeline::set_deployment_resources_desired_status( &txn, tenant_id, @@ -1100,7 +1096,7 @@ impl Storage for StoragePostgres { pipeline_name: &str, ) -> Result { let mut client = self.pool.get().await?; - let txn = client.transaction().await?; + let txn = transaction::begin(&mut client).await?; let pipeline_id = operations::pipeline::set_deployment_resources_desired_status( &txn, tenant_id, @@ -1121,7 +1117,7 @@ impl Storage for StoragePostgres { pipeline_name: &str, ) -> Result<(bool, PipelineId), DBError> { let mut client = self.pool.get().await?; - let txn = client.transaction().await?; + let txn = transaction::begin(&mut client).await?; let pipeline = operations::pipeline::get_pipeline_for_monitoring(&txn, tenant_id, pipeline_name, true) .await?; @@ -1153,7 +1149,7 @@ impl Storage for StoragePostgres { deployment_config: serde_json::Value, ) -> Result<(), DBError> { let mut client = self.pool.get().await?; - let txn = client.transaction().await?; + let txn = transaction::begin(&mut client).await?; // If the pipeline currently is already Stopped and is desired to be Stopped, // then there is no need to transition to Provisioning let pipeline = operations::pipeline::get_pipeline_by_id_for_monitoring( @@ -1200,7 +1196,7 @@ impl Storage for StoragePostgres { deployment_resources_status_details: serde_json::Value, ) -> Result<(), DBError> { let mut client = self.pool.get().await?; - let txn = client.transaction().await?; + let txn = transaction::begin(&mut client).await?; operations::pipeline::remain_deployment_resources_status_provisioning( &txn, tenant_id, @@ -1225,7 +1221,7 @@ impl Storage for StoragePostgres { deployment_runtime_desired_status: RuntimeDesiredStatus, ) -> Result<(), DBError> { let mut client = self.pool.get().await?; - let txn = client.transaction().await?; + let txn = transaction::begin(&mut client).await?; operations::pipeline::set_deployment_resources_status_provisioned( &txn, tenant_id, @@ -1254,7 +1250,7 @@ impl Storage for StoragePostgres { storage_status_details: Option, ) -> Result<(), DBError> { let mut client = self.pool.get().await?; - let txn = client.transaction().await?; + let txn = transaction::begin(&mut client).await?; operations::pipeline::remain_deployment_resources_status_provisioned( &txn, tenant_id, @@ -1280,7 +1276,7 @@ impl Storage for StoragePostgres { storage_status_details: Option, ) -> Result<(), DBError> { let mut client = self.pool.get().await?; - let txn = client.transaction().await?; + let txn = transaction::begin(&mut client).await?; let pipeline = operations::pipeline::get_pipeline_by_id_for_monitoring( &txn, tenant_id, @@ -1331,7 +1327,7 @@ impl Storage for StoragePostgres { deployment_resources_status_details: serde_json::Value, ) -> Result<(), DBError> { let mut client = self.pool.get().await?; - let txn = client.transaction().await?; + let txn = transaction::begin(&mut client).await?; operations::pipeline::remain_deployment_resources_status_stopping( &txn, tenant_id, @@ -1352,7 +1348,7 @@ impl Storage for StoragePostgres { version_guard: Version, ) -> Result<(), DBError> { let mut client = self.pool.get().await?; - let txn = client.transaction().await?; + let txn = transaction::begin(&mut client).await?; operations::pipeline::set_deployment_resources_status_stopped( &txn, tenant_id, @@ -1372,7 +1368,7 @@ impl Storage for StoragePostgres { deployment_error: ErrorResponse, ) -> Result<(), DBError> { let mut client = self.pool.get().await?; - let txn = client.transaction().await?; + let txn = transaction::begin(&mut client).await?; let pipeline = operations::pipeline::get_pipeline_by_id_for_monitoring( &txn, tenant_id, @@ -1408,7 +1404,7 @@ impl Storage for StoragePostgres { pipeline_name: &str, ) -> Result { let mut client = self.pool.get().await?; - let txn = client.transaction().await?; + let txn = transaction::begin(&mut client).await?; let pipeline = operations::pipeline::get_pipeline_for_monitoring(&txn, tenant_id, pipeline_name, true) .await?; @@ -1432,7 +1428,7 @@ impl Storage for StoragePostgres { pipeline_id: PipelineId, ) -> Result<(), DBError> { let mut client = self.pool.get().await?; - let txn = client.transaction().await?; + let txn = transaction::begin(&mut client).await?; operations::pipeline::set_storage_status( &txn, tenant_id, @@ -1451,7 +1447,7 @@ impl Storage for StoragePostgres { pipeline_name: &str, ) -> Result<(), DBError> { let mut client = self.pool.get().await?; - let txn = client.transaction().await?; + let txn = transaction::begin(&mut client).await?; operations::pipeline::increment_notify_counter(&txn, tenant_id, pipeline_name).await?; txn.commit().await?; Ok(()) @@ -1461,7 +1457,7 @@ impl Storage for StoragePostgres { &self, ) -> Result, DBError> { let mut client = self.pool.get().await?; - let txn = client.transaction().await?; + let txn = transaction::begin(&mut client).await?; let pipeline_ids = operations::pipeline::list_pipeline_ids_across_all_tenants(&txn).await?; txn.commit().await?; Ok(pipeline_ids) @@ -1471,7 +1467,7 @@ impl Storage for StoragePostgres { &self, ) -> Result, DBError> { let mut client = self.pool.get().await?; - let txn = client.transaction().await?; + let txn = transaction::begin(&mut client).await?; let pipelines = operations::pipeline::list_pipelines_across_all_tenants_for_monitoring(&txn).await?; txn.commit().await?; @@ -1485,7 +1481,7 @@ impl Storage for StoragePostgres { total_workers: usize, ) -> Result<(), DBError> { let mut client = self.pool.get().await?; - let txn = client.transaction().await?; + let txn = transaction::begin(&mut client).await?; let pipelines = operations::pipeline::list_pipelines_across_all_tenants_needing_sql_compilation_clear( &txn, @@ -1543,7 +1539,7 @@ impl Storage for StoragePostgres { total_workers: usize, ) -> Result, DBError> { let mut client = self.pool.get().await?; - let txn = client.transaction().await?; + let txn = transaction::begin(&mut client).await?; let next_pipeline_program = operations::pipeline::get_next_sql_compilation( &txn, platform_version, @@ -1562,7 +1558,7 @@ impl Storage for StoragePostgres { total_workers: usize, ) -> Result<(), DBError> { let mut client = self.pool.get().await?; - let txn = client.transaction().await?; + let txn = transaction::begin(&mut client).await?; let pipelines = operations::pipeline::list_pipelines_across_all_tenants_needing_rust_compilation_clear( &txn, @@ -1626,7 +1622,7 @@ impl Storage for StoragePostgres { total_workers: usize, ) -> Result, DBError> { let mut client = self.pool.get().await?; - let txn = client.transaction().await?; + let txn = transaction::begin(&mut client).await?; let next_pipeline_program = operations::pipeline::get_next_rust_compilation( &txn, platform_version, @@ -1640,7 +1636,7 @@ impl Storage for StoragePostgres { async fn count_pipelines_needing_compilation(&self) -> Result { let mut client = self.pool.get().await?; - let txn = client.transaction().await?; + let txn = transaction::begin(&mut client).await?; let count = operations::pipeline::count_pipelines_needing_compilation(&txn).await?; txn.commit().await?; Ok(count) @@ -1659,7 +1655,7 @@ impl Storage for StoragePostgres { DBError, > { let mut client = self.pool.get().await?; - let txn = client.transaction().await?; + let txn = transaction::begin(&mut client).await?; let pipeline_programs = operations::pipeline::list_pipeline_programs_across_all_tenants(&txn).await?; txn.commit().await?; @@ -1673,12 +1669,7 @@ impl Storage for StoragePostgres { how_many: u64, ) -> Result<(ExtendedPipelineDescrMonitoring, Vec), DBError> { let mut client = self.pool.get().await?; - let txn = client - .build_transaction() - .isolation_level(IsolationLevel::RepeatableRead) - .read_only(true) - .start() - .await?; + let txn = transaction::begin_read_only(&mut client).await?; let pipeline = operations::pipeline::get_pipeline_for_monitoring( &txn, tenant_id, @@ -1694,7 +1685,7 @@ impl Storage for StoragePostgres { async fn list_cluster_monitor_events(&self) -> Result, DBError> { let mut client = self.pool.get().await?; - let txn = client.transaction().await?; + let txn = transaction::begin(&mut client).await?; let events = operations::cluster_monitor::list_cluster_monitor_events_short(&txn).await?; txn.commit().await?; Ok(events) @@ -1705,7 +1696,7 @@ impl Storage for StoragePostgres { event_id: ClusterMonitorEventId, ) -> Result { let mut client = self.pool.get().await?; - let txn = client.transaction().await?; + let txn = transaction::begin(&mut client).await?; let event = operations::cluster_monitor::get_cluster_monitor_event_short(&txn, event_id).await?; txn.commit().await?; @@ -1717,7 +1708,7 @@ impl Storage for StoragePostgres { event_id: ClusterMonitorEventId, ) -> Result { let mut client = self.pool.get().await?; - let txn = client.transaction().await?; + let txn = transaction::begin(&mut client).await?; let event = operations::cluster_monitor::get_cluster_monitor_event_extended(&txn, event_id).await?; txn.commit().await?; @@ -1726,7 +1717,7 @@ impl Storage for StoragePostgres { async fn get_latest_cluster_monitor_event_short(&self) -> Result { let mut client = self.pool.get().await?; - let txn = client.transaction().await?; + let txn = transaction::begin(&mut client).await?; let event = operations::cluster_monitor::get_latest_cluster_monitor_event_short(&txn).await?; txn.commit().await?; @@ -1737,7 +1728,7 @@ impl Storage for StoragePostgres { &self, ) -> Result { let mut client = self.pool.get().await?; - let txn = client.transaction().await?; + let txn = transaction::begin(&mut client).await?; let event = operations::cluster_monitor::get_latest_cluster_monitor_event_extended(&txn).await?; txn.commit().await?; @@ -1750,7 +1741,7 @@ impl Storage for StoragePostgres { event_descr: NewClusterMonitorEvent, ) -> Result<(), DBError> { let mut client = self.pool.get().await?; - let txn = client.transaction().await?; + let txn = transaction::begin(&mut client).await?; operations::cluster_monitor::new_cluster_monitor_event(&txn, new_id, event_descr).await?; txn.commit().await?; Ok(()) @@ -1762,7 +1753,7 @@ impl Storage for StoragePostgres { retention_num: u16, ) -> Result<(u64, u64), DBError> { let mut client = self.pool.get().await?; - let txn = client.transaction().await?; + let txn = transaction::begin(&mut client).await?; let (num_deleted_due_to_timestamp, num_deleted_due_to_limit) = operations::cluster_monitor::delete_cluster_monitor_events_beyond_retention( &txn, @@ -1780,12 +1771,7 @@ impl Storage for StoragePostgres { pipeline_name: String, ) -> Result, DBError> { let mut client = self.pool.get().await?; - let txn = client - .build_transaction() - .isolation_level(IsolationLevel::RepeatableRead) - .read_only(true) - .start() - .await?; + let txn = transaction::begin_read_only(&mut client).await?; let events = operations::pipeline_monitor::list_pipeline_monitor_events_short( &txn, tenant_id, @@ -1802,12 +1788,7 @@ impl Storage for StoragePostgres { pipeline_name: String, ) -> Result, DBError> { let mut client = self.pool.get().await?; - let txn = client - .build_transaction() - .isolation_level(IsolationLevel::RepeatableRead) - .read_only(true) - .start() - .await?; + let txn = transaction::begin_read_only(&mut client).await?; let events = operations::pipeline_monitor::list_pipeline_monitor_events_extended( &txn, tenant_id, @@ -1825,12 +1806,7 @@ impl Storage for StoragePostgres { event_id: PipelineMonitorEventId, ) -> Result { let mut client = self.pool.get().await?; - let txn = client - .build_transaction() - .isolation_level(IsolationLevel::RepeatableRead) - .read_only(true) - .start() - .await?; + let txn = transaction::begin_read_only(&mut client).await?; let event = operations::pipeline_monitor::get_pipeline_monitor_event_short( &txn, tenant_id, @@ -1849,12 +1825,7 @@ impl Storage for StoragePostgres { event_id: PipelineMonitorEventId, ) -> Result { let mut client = self.pool.get().await?; - let txn = client - .build_transaction() - .isolation_level(IsolationLevel::RepeatableRead) - .read_only(true) - .start() - .await?; + let txn = transaction::begin_read_only(&mut client).await?; let event = operations::pipeline_monitor::get_pipeline_monitor_event_extended( &txn, tenant_id, @@ -1872,12 +1843,7 @@ impl Storage for StoragePostgres { pipeline_name: String, ) -> Result { let mut client = self.pool.get().await?; - let txn = client - .build_transaction() - .isolation_level(IsolationLevel::RepeatableRead) - .read_only(true) - .start() - .await?; + let txn = transaction::begin_read_only(&mut client).await?; let event = operations::pipeline_monitor::get_latest_pipeline_monitor_event_short( &txn, tenant_id, @@ -1894,12 +1860,7 @@ impl Storage for StoragePostgres { pipeline_name: String, ) -> Result { let mut client = self.pool.get().await?; - let txn = client - .build_transaction() - .isolation_level(IsolationLevel::RepeatableRead) - .read_only(true) - .start() - .await?; + let txn = transaction::begin_read_only(&mut client).await?; let event = operations::pipeline_monitor::get_latest_pipeline_monitor_event_extended( &txn, tenant_id, @@ -1915,7 +1876,7 @@ impl Storage for StoragePostgres { retention_num: u32, ) -> Result { let mut client = self.pool.get().await?; - let txn = client.transaction().await?; + let txn = transaction::begin(&mut client).await?; let num_deleted = operations::pipeline_monitor::delete_pipeline_monitor_events_exceeding_retention( &txn, @@ -1965,13 +1926,15 @@ impl StoragePostgres { db_config: &DatabaseConfig, #[cfg(feature = "postgresql_embedded")] pg_inst: Option, ) -> Result { - let mut config = db_config.tokio_postgres_config()?; - let new_options = if let Some(options) = config.get_options() { - format!("{options} -c lock_timeout=10000") - } else { - "-c lock_timeout=10000".to_string() - }; - config.options(new_options); + // The connection deliberately carries no `options` startup parameter: + // a pooler in between rejects parameters it does not know. The lock + // timeout the manager needs is set per transaction instead, by + // `db::transaction` (which explains the trade-off in detail). Note that + // this leaves the transactions refinery runs for the migrations without + // a lock timeout, which is what a rolling upgrade wants: an instance + // starting up waits for the migration of another one to finish rather + // than failing to start. + let config = db_config.tokio_postgres_config()?; debug!( "Opening Postgres connection with configuration: {:?}", config @@ -2043,7 +2006,7 @@ impl StoragePostgres { pub(crate) async fn ensure_default_tenant(&self) -> Result<(), DBError> { let default_tenant = TenantRecord::default(); let mut client = self.pool.get().await?; - let txn = client.transaction().await?; + let txn = transaction::begin(&mut client).await?; operations::tenant::ensure_default_tenant( &txn, default_tenant.id.0, @@ -2058,7 +2021,7 @@ impl StoragePostgres { /// Performs the conversion of the pipeline table fields which are YAML to become JSON. async fn perform_yaml_to_json_migration(&self) -> Result<(), DBError> { let mut client = self.pool.get().await?; - let txn = client.transaction().await?; + let txn = transaction::begin(&mut client).await?; let stmt = txn .prepare_cached( "SELECT p.id, p.runtime_config, p.program_config, p.program_info, p.deployment_error, p.deployment_config diff --git a/crates/pipeline-manager/src/db/test.rs b/crates/pipeline-manager/src/db/test.rs index a5784613320..b5b5c2a261b 100644 --- a/crates/pipeline-manager/src/db/test.rs +++ b/crates/pipeline-manager/src/db/test.rs @@ -5,6 +5,8 @@ use crate::db::error::DBError::InvalidResourcesStatusNotRemain; use crate::db::operations::pipeline::get_pipeline_by_id_for_monitoring; use crate::db::storage::{ExtendedPipelineDescrRunner, Storage}; use crate::db::storage_postgres::{StoragePostgres, is_pipeline_assigned_to_worker}; +use crate::db::transaction; +use crate::db::transaction::LOCK_TIMEOUT; use crate::db::types::api_key::{ApiKeyDescr, ApiKeyId}; use crate::db::types::monitor::{ ClusterMonitorEvent, ClusterMonitorEventId, ExtendedClusterMonitorEvent, @@ -41,6 +43,7 @@ use crate::oidc::destination::{TenantIssuerPolicy, validate_tenant_oidc_url}; use crate::oidc::trust_name::validate_oidc_trust_name; use async_trait::async_trait; use chrono::{DateTime, TimeZone, Utc}; +use deadpool_postgres::GenericClient; use feldera_types::checkpoint::CheckpointMetadata; use feldera_types::config::{ DevTweaks, FtConfig, PipelineConfig, ProgramIr, ResourceConfig, RuntimeConfig, @@ -4365,6 +4368,55 @@ async fn pipeline_client_metadata_update_while_running() { assert_eq!(unchanged.refresh_version, before.refresh_version); } +/// Reads the lock timeout that applies to `client`, in milliseconds. Zero is +/// the Postgres default and means a statement waits for a lock indefinitely. +async fn lock_timeout_ms(client: &impl GenericClient) -> i64 { + client + .query_one( + "SELECT setting::bigint FROM pg_settings WHERE name = 'lock_timeout'", + &[], + ) + .await + .unwrap() + .get(0) +} + +/// The lock timeout has to be carried by each transaction rather than by the +/// connection: a pooler in between rejects the connection option that would +/// carry it, and a session-level setting does not follow the client to the +/// server connection its next transaction lands on. +#[tokio::test] +async fn transaction_lock_timeout_is_transaction_local() { + let handle = test_setup().await; + let mut client = handle.db.pool.get().await.unwrap(); + let expected_ms = LOCK_TIMEOUT.as_millis() as i64; + + // The session holds no timeout of its own: the test server leaves the + // parameter at the Postgres default of waiting indefinitely. + assert_eq!( + lock_timeout_ms(&client).await, + 0, + "the connection carries the lock timeout instead of leaving it to each transaction" + ); + + let txn = transaction::begin(&mut client).await.unwrap(); + assert_eq!(lock_timeout_ms(&txn).await, expected_ms); + txn.commit().await.unwrap(); + + assert_eq!( + lock_timeout_ms(&client).await, + 0, + "the lock timeout outlived its transaction, so it can leak to whichever \ + client uses this connection next" + ); + + // Read-only transactions are bounded as well: reading a row still waits for + // any conflicting lock a schema change holds. + let txn = transaction::begin_read_only(&mut client).await.unwrap(); + assert_eq!(lock_timeout_ms(&txn).await, expected_ms); + txn.commit().await.unwrap(); +} + #[tokio::test] async fn pipeline_concurrent_access_stall() { let handle = test_setup().await; @@ -4397,8 +4449,8 @@ async fn pipeline_concurrent_access_stall() { // Non-conflicting for (row_lock1, row_lock2) in [(false, false), (true, false), (false, true)] { - let txn1 = client1.transaction().await.unwrap(); - let txn2 = client2.transaction().await.unwrap(); + let txn1 = transaction::begin(&mut client1).await.unwrap(); + let txn2 = transaction::begin(&mut client2).await.unwrap(); get_pipeline_by_id_for_monitoring(&txn1, tenant_id, pipeline.id, row_lock1) .await .unwrap(); @@ -4409,17 +4461,15 @@ async fn pipeline_concurrent_access_stall() { txn2.commit().await.unwrap(); } - // Conflicting - let txn1 = client1.transaction().await.unwrap(); - let txn2 = client2.transaction().await.unwrap(); + // Conflicting. Both transactions are started the way the manager starts + // them, which is what bounds the wait for the row lock: the test would hang + // here if that bound were lost. It consequently runs for the duration of + // the timeout. + let txn1 = transaction::begin(&mut client1).await.unwrap(); + let txn2 = transaction::begin(&mut client2).await.unwrap(); get_pipeline_by_id_for_monitoring(&txn1, tenant_id, pipeline.id, true) .await .unwrap(); - // The lock timeout has been set globally via an option. As such, the below is not needed. - // This test does take some time as a consequence (the actual timeout used: 10 seconds). - // However, this is an important check to make sure that when the system is deployed, - // the lock timeout is enforced. - // txn2.execute("SET LOCAL lock_timeout = 10000", &[]).await.unwrap(); let ts_start = Instant::now(); let error = get_pipeline_by_id_for_monitoring(&txn2, tenant_id, pipeline.id, true) .await @@ -4430,7 +4480,10 @@ async fn pipeline_concurrent_access_stall() { DBError::LockTookTooLong ); let elapsed = ts_start.elapsed(); - assert!(elapsed >= Duration::from_millis(8000) && elapsed <= Duration::from_millis(30000)); + assert!( + elapsed >= LOCK_TIMEOUT.mul_f64(0.8) && elapsed <= LOCK_TIMEOUT * 3, + "waited {elapsed:?} for the row lock, which is not around the {LOCK_TIMEOUT:?} timeout" + ); txn1.commit().await.unwrap(); txn2.commit().await.unwrap(); } @@ -4489,13 +4542,13 @@ async fn pipeline_concurrent_access_deadlock() { // - T2 tries to lock pipeline 1 -> waits or deadlock let mut client1 = handle.db.pool.get().await.unwrap(); let mut client2 = handle.db.pool.get().await.unwrap(); - let txn2 = client2.transaction().await.unwrap(); + let txn2 = transaction::begin(&mut client2).await.unwrap(); get_pipeline_by_id_for_monitoring(&txn2, tenant_id, pipeline2.id, true) .await .unwrap(); let (tx, rx) = oneshot::channel::<()>(); let join_handle = spawn(async move { - let txn1 = client1.transaction().await.unwrap(); + let txn1 = transaction::begin(&mut client1).await.unwrap(); get_pipeline_by_id_for_monitoring(&txn1, tenant_id, pipeline1.id, true) .await .unwrap(); diff --git a/crates/pipeline-manager/src/db/transaction.rs b/crates/pipeline-manager/src/db/transaction.rs new file mode 100644 index 00000000000..35f8e473d3e --- /dev/null +++ b/crates/pipeline-manager/src/db/transaction.rs @@ -0,0 +1,68 @@ +//! Transaction creation for the management database. +//! +//! Every transaction bounds how long its statements wait for a lock. The API +//! server and the runner both lock pipeline rows with `SELECT ... FOR UPDATE` +//! before acting on them; without a bound, a caller that finds a row locked +//! waits indefinitely instead of being told to retry. +//! +//! The bound is applied with `SET LOCAL` inside each transaction rather than +//! through the connection's `options` startup parameter, because the management +//! database is often reached through a connection pooler: +//! +//! - PgBouncer rejects startup parameters it does not know +//! (`unsupported startup parameter: options`), which leaves the manager unable +//! to connect at all. Listing `options` in `ignore_startup_parameters` silences +//! the rejection but makes PgBouncer drop the setting, so the timeout would +//! silently not apply. +//! - A session-level `SET` fares no better: in transaction pooling mode the +//! server connection it lands on is not the one the next transaction gets, so +//! the setting both misses its target and leaks to an unrelated client. +//! +//! `SET LOCAL` travels with the transaction, which a pooler keeps pinned to one +//! server connection from `BEGIN` to `COMMIT`, and Postgres reverts at commit. +//! It therefore holds for a direct connection and for every pooling mode, at the +//! cost of one round trip per transaction. + +use crate::db::error::DBError; +use deadpool_postgres::{Client, Transaction}; +use std::sync::LazyLock; +use std::time::Duration; +use tokio_postgres::IsolationLevel; + +/// How long a statement waits for a lock before Postgres aborts it with +/// `lock_not_available`, which reaches the caller as [`DBError::LockTookTooLong`] +/// (HTTP code 503: retry). It has to stay at a millisecond or more, as Postgres +/// reads `lock_timeout = 0` as no timeout at all. +pub(crate) const LOCK_TIMEOUT: Duration = Duration::from_secs(10); + +/// Statement that applies [`LOCK_TIMEOUT`] to the current transaction. +static SET_LOCK_TIMEOUT: LazyLock = + LazyLock::new(|| format!("SET LOCAL lock_timeout = {}", LOCK_TIMEOUT.as_millis())); + +/// Starts a read-write transaction with a bounded lock wait. +pub(crate) async fn begin(client: &mut Client) -> Result, DBError> { + let txn = client.transaction().await?; + set_lock_timeout(&txn).await?; + Ok(txn) +} + +/// Starts a read-only `REPEATABLE READ` transaction with a bounded lock wait. +/// All its statements read the same snapshot, and none of them can write. +pub(crate) async fn begin_read_only(client: &mut Client) -> Result, DBError> { + let txn = client + .build_transaction() + .isolation_level(IsolationLevel::RepeatableRead) + .read_only(true) + .start() + .await?; + set_lock_timeout(&txn).await?; + Ok(txn) +} + +/// Bounds by [`LOCK_TIMEOUT`] how long each statement of `txn` waits for a lock. +async fn set_lock_timeout(txn: &Transaction<'_>) -> Result<(), DBError> { + // A simple query, not `execute`, to keep this off the statement cache: a + // pooler supports prepared statements only in specific configurations. + txn.batch_execute(&SET_LOCK_TIMEOUT).await?; + Ok(()) +} From 8ec0248999001880ea564f948d2308d25593e26e Mon Sep 17 00:00:00 2001 From: Leonid Ryzhyk Date: Fri, 14 Aug 2026 12:55:10 -0700 Subject: [PATCH 2/2] pipeline-manager: report what Postgres said when a call fails A failing database call reported the category its failure fell into and nothing else, so a connection that PgBouncer refuses read: Error: Postgres connection pool error: 'Error occurred while creating a new object: db error' `tokio_postgres::Error` displays that category and leaves the message the server sent to the error it wraps, which the report dropped. Appending the innermost cause turns the same failure into: Error: Postgres connection pool error: 'Error occurred while creating a new object: db error: FATAL: unsupported startup parameter in options: lock_timeout' The Postgres, pool and migration errors now all report that cause, both in `Display` and in the JSON the API returns. Only the two ends of the chain are needed, as every wrapper in between displays the error it got. `DBError` still exposes no `source`, so a caller that walks the chain itself cannot repeat what the message already carries. Signed-off-by: Leonid Ryzhyk --- crates/pipeline-manager/src/db/error.rs | 114 ++++++++++++++++++++++-- crates/pipeline-manager/src/db/test.rs | 18 ++++ 2 files changed, 125 insertions(+), 7 deletions(-) diff --git a/crates/pipeline-manager/src/db/error.rs b/crates/pipeline-manager/src/db/error.rs index ea3caa6820d..b3381abc783 100644 --- a/crates/pipeline-manager/src/db/error.rs +++ b/crates/pipeline-manager/src/db/error.rs @@ -9,6 +9,7 @@ use crate::db::types::tenant::TenantId; use crate::db::types::user::InvalidMembershipOrigin; use crate::db::types::utils::ValidationError; use crate::db::types::version::Version; +use crate::error::source_error; use actix_web::{HttpResponse, ResponseError, body::BoxBody, http::StatusCode}; use deadpool_postgres::PoolError; use feldera_types::error::DetailedError; @@ -343,6 +344,27 @@ impl DBError { } } +/// Renders `error` with the innermost error that caused it. +/// +/// The half of a Postgres failure that identifies it sits in that innermost +/// error: `tokio_postgres::Error` displays the category the failure falls into +/// ("db error", "error connecting to server") and leaves what the server said to +/// its source. Reporting the outer error alone therefore says that something went +/// wrong and nothing about what. +/// +/// Only those two ends of the chain are needed, as every wrapper in between +/// displays its own source: refinery names the migration and then quotes the +/// error it got, and the pool error quotes the connection error it got. +fn message_with_cause(error: &dyn StdError) -> String { + let message = error.to_string(); + let cause = source_error(error).to_string(); + if message.ends_with(&cause) { + message + } else { + format!("{message}: {cause}") + } +} + fn serialize_pg_error( error: &PgError, backtrace: &Backtrace, @@ -352,7 +374,7 @@ where S: Serializer, { let mut ser = serializer.serialize_struct("PgError", 2)?; - ser.serialize_field("error", &error.to_string())?; + ser.serialize_field("error", &message_with_cause(error))?; ser.serialize_field("backtrace", &backtrace.to_string())?; ser.end() } @@ -366,7 +388,7 @@ where S: Serializer, { let mut ser = serializer.serialize_struct("PgPoolError", 2)?; - ser.serialize_field("error", &error.to_string())?; + ser.serialize_field("error", &message_with_cause(error))?; ser.serialize_field("backtrace", &backtrace.to_string())?; ser.end() } @@ -380,7 +402,7 @@ where S: Serializer, { let mut ser = serializer.serialize_struct("RefineryError", 2)?; - ser.serialize_field("error", &error.to_string())?; + ser.serialize_field("error", &message_with_cause(error))?; ser.serialize_field("backtrace", &backtrace.to_string())?; ser.end() } @@ -513,13 +535,25 @@ impl Display for DBError { fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { match self { DBError::PostgresError { error, .. } => { - write!(f, "Unexpected Postgres error: '{error}'") + write!( + f, + "Unexpected Postgres error: '{}'", + message_with_cause(&**error) + ) } DBError::PostgresPoolError { error, .. } => { - write!(f, "Postgres connection pool error: '{error}'") + write!( + f, + "Postgres connection pool error: '{}'", + message_with_cause(&**error) + ) } DBError::PostgresMigrationError { error, .. } => { - write!(f, "DB schema migration error: '{error}'") + write!( + f, + "DB schema migration error: '{}'", + message_with_cause(&**error) + ) } #[cfg(feature = "postgresql_embedded")] DBError::PgEmbedError { error, .. } => { @@ -1296,8 +1330,74 @@ impl From for ErrorResponse { #[cfg(test)] mod test { - use crate::db::error::DBError; + use crate::db::error::{DBError, message_with_cause}; use crate::error::ManagerError; + use std::error::Error as StdError; + use std::fmt; + + /// Error with a `Display` and a source, both of which the test dictates. + #[derive(Debug)] + struct Layer { + message: String, + cause: Option>, + } + + impl Layer { + fn new(message: &str, cause: Option) -> Self { + Self { + message: message.to_string(), + cause: cause.map(Box::new), + } + } + } + + impl fmt::Display for Layer { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.write_str(&self.message) + } + } + + impl StdError for Layer { + fn source(&self) -> Option<&(dyn StdError + 'static)> { + self.cause.as_deref().map(|cause| cause as &dyn StdError) + } + } + + #[test] + fn the_innermost_cause_is_reported() { + // The shape a refused connection arrives in: each wrapper displays the + // error it got, except the innermost one, which nobody displays. + let innermost = Layer::new("FATAL: unsupported startup parameter", None); + let middle = Layer::new("db error", Some(innermost)); + let outer = Layer::new( + "Error occurred while creating a new object: db error", + Some(middle), + ); + assert_eq!( + message_with_cause(&outer), + "Error occurred while creating a new object: db error: \ + FATAL: unsupported startup parameter" + ); + } + + #[test] + fn a_cause_already_displayed_is_reported_once() { + // A wrapper whose innermost cause is the error it already displays. + let cause = Layer::new("db error", None); + let outer = Layer::new( + "Error occurred while creating a new object: db error", + Some(cause), + ); + assert_eq!( + message_with_cause(&outer), + "Error occurred while creating a new object: db error" + ); + } + + #[test] + fn an_error_without_causes_is_unchanged() { + assert_eq!(message_with_cause(&Layer::new("alone", None)), "alone"); + } #[test] fn invalid_bootstrap_error_serialization() { diff --git a/crates/pipeline-manager/src/db/test.rs b/crates/pipeline-manager/src/db/test.rs index b5b5c2a261b..dce52be5e30 100644 --- a/crates/pipeline-manager/src/db/test.rs +++ b/crates/pipeline-manager/src/db/test.rs @@ -4381,6 +4381,24 @@ async fn lock_timeout_ms(client: &impl GenericClient) -> i64 { .get(0) } +/// A failed statement has to report what the server said. `tokio_postgres` +/// displays the category the failure falls into and leaves the message itself to +/// the error it wraps, so a report of the outermost error alone is useless for +/// diagnosis. +#[tokio::test] +async fn postgres_error_reports_what_the_server_said() { + let handle = test_setup().await; + let mut client = handle.db.pool.get().await.unwrap(); + let txn = transaction::begin(&mut client).await.unwrap(); + + let error: DBError = txn.batch_execute("SELECT 1/0").await.unwrap_err().into(); + let reported = error.to_string(); + assert!( + reported.contains("division by zero"), + "the server's message is missing from {reported:?}" + ); +} + /// The lock timeout has to be carried by each transaction rather than by the /// connection: a pooler in between rejects the connection option that would /// carry it, and a session-level setting does not follow the client to the