fix: Handle Redis connection errors gracefully instead of panicking (#111)

* fix: Handle Redis connection errors gracefully instead of panicking

When Redis goes down (e.g. master failover), a broken pipe error would
cause `get_queued_query_count` to panic due to an `.unwrap()`, which
then poisoned the RwLock used in metrics callbacks, cascading into
further panics and leaving pods unresponsive until the liveness probe
killed them.

- Replace `.unwrap()` with proper error propagation in
  `get_queued_query_count`
- Handle poisoned locks, closed channels, and panicked threads
  gracefully in both metrics callbacks

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

* Rename e to error

* changelog

---------

Co-authored-by: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

Author: sbernauer

CHANGELOG.md

@@ -4,6 +4,14 @@ All notable changes to this project will be documented in this file.
 
 ## [Unreleased]
 
+### Fixed
+
+- Handle Redis connection errors (e.g. broken pipe during master failover) gracefully instead of panicking.
+  Previously, `get_queued_query_count` would `.unwrap()` on the Redis result, causing a panic that poisoned
+  the metrics `RwLock`, cascading into further panics and leaving pods unresponsive ([#111]).
+
+[#111]: https://github.com/stackabletech/trino-lb/pull/111
+
 ## [0.6.0] - 2026-02-17
 
 ### Added

trino-lb-persistence/src/redis/mod.rs

@@ -116,6 +116,12 @@ pub enum Error {
     #[snafu(display("Invalid response from compare and set lua script. Expected either 0 or 1"))]
     InvalidCASScriptResponse { response: u64 },
 
+    #[snafu(display("Failed to get queued query count for cluster group {cluster_group:?}"))]
+    GetQueuedQueryCount {
+        source: RedisError,
+        cluster_group: String,
+    },
+
     #[snafu(display("Failed to list queued queries for cluster group {cluster_group:?}"))]
     ListQueuedQueries {
         source: RedisError,
@@ -428,7 +434,7 @@ where
             .connection()
             .scard::<_, Option<u64>>(queued_query_set_name(cluster_group))
             .await
-            .unwrap()
+            .context(GetQueuedQueryCountSnafu { cluster_group })?
             // The set might not be there yet, as no queries have been queued for this cluster group so far.
             .unwrap_or_default())
     }

trino-lb/src/metrics.rs

@@ -142,19 +142,42 @@ impl Metrics {
 
         meter
             .register_callback(&[queued_queries_metric.as_any()], move |observer| {
-                ping_sender.send(()).unwrap();
+                if let Err(err) = ping_sender.send(()) {
+                    error!(error = ?err, "Failed to send ping for queued_queries metric");
+                    return;
+                }
                 let queued_queries = std::thread::scope(|s| {
-                    s.spawn(|| metrics_receiver.write().unwrap().blocking_recv().unwrap())
-                        .join()
-                        .unwrap()
+                    s.spawn(|| {
+                        let mut receiver = match metrics_receiver.write() {
+                            Ok(r) => r,
+                            Err(err) => {
+                                error!(error = %err, "Failed to acquire write lock for queued_queries metric");
+                                return None;
+                            }
+                        };
+                        match receiver.blocking_recv() {
+                            Some(v) => Some(v),
+                            None => {
+                                error!("queued_queries metrics channel closed");
+                                None
+                            }
+                        }
+                    })
+                    .join()
+                    .unwrap_or_else(|err| {
+                        error!(error = ?err, "queued_queries metrics thread panicked");
+                        None
+                    })
                 });
 
-                for (cluster_group, queued) in queued_queries {
-                    observer.observe_u64(
-                        &queued_queries_metric,
-                        queued,
-                        [KeyValue::new("cluster-group", cluster_group)].as_ref(),
-                    );
+                if let Some(queued_queries) = queued_queries {
+                    for (cluster_group, queued) in queued_queries {
+                        observer.observe_u64(
+                            &queued_queries_metric,
+                            queued,
+                            [KeyValue::new("cluster-group", cluster_group)].as_ref(),
+                        );
+                    }
                 }
             })
             .context(RegisterMetricsCallbackSnafu)?;
@@ -183,24 +206,47 @@ impl Metrics {
             .register_callback(
                 &[cluster_counts_per_state_metric.as_any()],
                 move |observer| {
-                    ping_sender.send(()).unwrap();
+                    if let Err(err) = ping_sender.send(()) {
+                        error!(error = ?err, "Failed to send ping for cluster_counts_per_state metric");
+                        return;
+                    }
                     let cluster_counts = std::thread::scope(|s| {
-                        s.spawn(|| metrics_receiver.write().unwrap().blocking_recv().unwrap())
-                            .join()
-                            .unwrap()
+                        s.spawn(|| {
+                            let mut receiver = match metrics_receiver.write() {
+                                Ok(r) => r,
+                                Err(err) => {
+                                    error!(error = %err, "Failed to acquire write lock for cluster_counts_per_state metric");
+                                    return None;
+                                }
+                            };
+                            match receiver.blocking_recv() {
+                                Some(v) => Some(v),
+                                None => {
+                                    error!("cluster_counts_per_state metrics channel closed");
+                                    None
+                                }
+                            }
+                        })
+                        .join()
+                        .unwrap_or_else(|err| {
+                            error!(error = ?err, "cluster_counts_per_state metrics thread panicked");
+                            None
+                        })
                     });
 
-                    for (cluster_group, counts) in cluster_counts {
-                        for (state, count) in counts {
-                            observer.observe_u64(
-                                &cluster_counts_per_state_metric,
-                                count,
-                                [
-                                    KeyValue::new("cluster-group", cluster_group.clone()),
-                                    KeyValue::new::<_, &str>("state", state.into()),
-                                ]
-                                .as_ref(),
-                            );
+                    if let Some(cluster_counts) = cluster_counts {
+                        for (cluster_group, counts) in cluster_counts {
+                            for (state, count) in counts {
+                                observer.observe_u64(
+                                    &cluster_counts_per_state_metric,
+                                    count,
+                                    [
+                                        KeyValue::new("cluster-group", cluster_group.clone()),
+                                        KeyValue::new::<_, &str>("state", state.into()),
+                                    ]
+                                    .as_ref(),
+                                );
+                            }
                         }
                     }
                 },