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<dyn std::error::Error + Send + Sync>> {
         // 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<dyn std::error::Error + Send + Sync>> {
         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/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<S>(
     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<DBError> 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<Box<Layer>>,
+    }
+
+    impl Layer {
+        fn new(message: &str, cause: Option<Layer>) -> 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/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<TenantId, DBError> {
         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<String, DBError> {
         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<TenantId, DBError> {
         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<TenantInfo, 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_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<Option<TenantId>, 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<Option<TenantInfo>, 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<Vec<TenantInfo>, 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<Vec<UserMembership>, 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<UserId, DBError> {
         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<bool, DBError> {
         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<Vec<TenantMember>, 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<UserId, DBError> {
         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<Vec<ApiKeyDescr>, 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<ApiKeyDescr, DBError> {
         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<Vec<OidcTrustDescr>, 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<OidcTrustDescr, 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::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<bool, 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::is_trusted_issuer(&txn, issuer).await?;
         txn.commit().await?;
         Ok(result)
@@ -486,7 +487,7 @@ impl Storage for StoragePostgres {
         audiences: &[String],
     ) -> Result<Vec<(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::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<Vec<ExtendedPipelineDescr>, 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<Vec<ExtendedPipelineDescrMonitoring>, 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<ExtendedPipelineDescr, 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(&txn, tenant_id, name, false).await?;
         txn.commit().await?;
         Ok(pipeline)
@@ -534,7 +535,7 @@ impl Storage for StoragePostgres {
         name: &str,
     ) -> Result<ExtendedPipelineDescrMonitoring, 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, name, false).await?;
         txn.commit().await?;
@@ -547,7 +548,7 @@ impl Storage for StoragePostgres {
         pipeline_id: PipelineId,
     ) -> Result<ExtendedPipelineDescr, 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(&txn, tenant_id, pipeline_id, false).await?;
         txn.commit().await?;
@@ -560,7 +561,7 @@ impl Storage for StoragePostgres {
         pipeline_id: PipelineId,
     ) -> Result<ExtendedPipelineDescrMonitoring, 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,
@@ -580,12 +581,7 @@ impl Storage for StoragePostgres {
         provision_called: bool,
     ) -> Result<ExtendedPipelineDescrRunner, 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_monitoring = operations::pipeline::get_pipeline_by_id_for_monitoring(
             &txn,
             tenant_id,
@@ -633,7 +629,7 @@ impl Storage for StoragePostgres {
         pipeline: PipelineDescr,
     ) -> Result<ExtendedPipelineDescr, DBError> {
         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<serde_json::Value>,
     ) -> Result<ExtendedPipelineDescr, DBError> {
         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<PipelineId, DBError> {
         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<PipelineId, DBError> {
         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<PipelineId, DBError> {
         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<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_provisioned(
             &txn,
             tenant_id,
@@ -1280,7 +1276,7 @@ impl Storage for StoragePostgres {
         storage_status_details: Option<serde_json::Value>,
     ) -> 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<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?;
@@ -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<Vec<(TenantId, PipelineId)>, 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<Vec<(TenantId, ExtendedPipelineDescrMonitoring)>, 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<Option<(TenantId, ExtendedPipelineDescr)>, 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<Option<(TenantId, ExtendedPipelineDescr)>, 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<u64, DBError> {
         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<SupportBundleData>), 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<Vec<ClusterMonitorEvent>, 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<ClusterMonitorEvent, DBError> {
         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<ExtendedClusterMonitorEvent, DBError> {
         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<ClusterMonitorEvent, DBError> {
         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<ExtendedClusterMonitorEvent, DBError> {
         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<Vec<PipelineMonitorEvent>, 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<Vec<ExtendedPipelineMonitorEvent>, 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<PipelineMonitorEvent, 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 event = operations::pipeline_monitor::get_pipeline_monitor_event_short(
             &txn,
             tenant_id,
@@ -1849,12 +1825,7 @@ impl Storage for StoragePostgres {
         event_id: PipelineMonitorEventId,
     ) -> Result<ExtendedPipelineMonitorEvent, 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 event = operations::pipeline_monitor::get_pipeline_monitor_event_extended(
             &txn,
             tenant_id,
@@ -1872,12 +1843,7 @@ impl Storage for StoragePostgres {
         pipeline_name: String,
     ) -> Result<PipelineMonitorEvent, 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 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<ExtendedPipelineMonitorEvent, 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 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<u64, DBError> {
         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<postgresql_embedded::PostgreSQL>,
     ) -> Result<Self, DBError> {
-        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..dce52be5e30 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,73 @@ 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)
+}
+
+/// 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
+/// 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 +4467,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 +4479,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 +4498,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 +4560,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<String> =
+    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<Transaction<'_>, 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<Transaction<'_>, 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(())
+}