feldera · ryzhyk · Aug 15, 2026 · Aug 14, 2026 · Aug 14, 2026
diff --git 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
@@ -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
@@ -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() {