Recover from poisoned org DB client mutex

src/api/org_db.rs

@@ -215,13 +215,17 @@ fn connect(org_db_url: &str) -> Result<Client, AutterError> {
 /// Get the cached client for `org_db_url`, connecting (and provisioning) lazily.
 fn get_or_connect(org_db_url: &str) -> Result<Arc<Mutex<Client>>, AutterError> {
     {
-        let map = CONNECTIONS.lock().expect("connection cache poisoned");
+        let map = CONNECTIONS
+            .lock()
+            .unwrap_or_else(|poisoned| poisoned.into_inner());
         if let Some(client) = map.get(org_db_url) {
             return Ok(client.clone());
         }
     }
     let client = Arc::new(Mutex::new(connect(org_db_url)?));
-    let mut map = CONNECTIONS.lock().expect("connection cache poisoned");
+    let mut map = CONNECTIONS
+        .lock()
+        .unwrap_or_else(|poisoned| poisoned.into_inner());
     // Another thread may have connected while we were dialing — keep theirs.
     Ok(map.entry(org_db_url.to_string()).or_insert(client).clone())
 }
@@ -234,26 +238,35 @@ fn get_or_connect(org_db_url: &str) -> Result<Arc<Mutex<Client>>, AutterError> {
 /// failure: the notes/CAS closures count per-row errors internally rather than
 /// propagating them, so a dead socket would otherwise look like an all-rows
 /// failure instead of triggering a reconnect.
+///
+/// A poisoned client mutex is handled the same way. The daemon caches the client
+/// for its whole life, so a single op that panicked mid-query would otherwise
+/// poison the mutex forever and turn every later upload into a panic. We recover
+/// instead: a mid-query panic can leave the Postgres protocol state out of sync,
+/// so we discard the connection and redial rather than reuse the same socket.
 fn run<T>(
     org_db_url: &str,
     op: impl FnOnce(&mut Client) -> Result<T, postgres::Error>,
 ) -> Result<T, AutterError> {
     let arc = get_or_connect(org_db_url)?;
+
+    // Reuse the cached connection only if we can lock it and it still round-trips.
+    if let Ok(mut guard) = arc.lock()
+        && guard.is_valid(Duration::from_secs(5)).is_ok()
     {
-        let mut guard = arc.lock().expect("org client mutex poisoned");
-        if guard.is_valid(Duration::from_secs(5)).is_err() {
-            // Cached connection is stale — drop it so the next get reconnects.
-            drop(guard);
-            CONNECTIONS
-                .lock()
-                .expect("connection cache poisoned")
-                .remove(org_db_url);
-            let fresh = get_or_connect(org_db_url)?;
-            let mut guard = fresh.lock().expect("org client mutex poisoned");
-            return op(&mut guard).map_err(map_db_err);
-        }
-        op(&mut guard).map_err(map_db_err)
+        return op(&mut guard).map_err(map_db_err);
     }
+
+    // Cached connection is stale or poisoned — drop it so we dial a fresh one.
+    CONNECTIONS
+        .lock()
+        .unwrap_or_else(|poisoned| poisoned.into_inner())
+        .remove(org_db_url);
+    let fresh = get_or_connect(org_db_url)?;
+    let mut guard = fresh
+        .lock()
+        .unwrap_or_else(|poisoned| poisoned.into_inner());
+    op(&mut guard).map_err(map_db_err)
 }
 
 fn map_db_err(e: postgres::Error) -> AutterError {

src/daemon/telemetry_worker.rs

@@ -268,8 +268,14 @@ async fn telemetry_flush_loop(buffer: Arc<Mutex<TelemetryBuffer>>) {
         };
 
         // Flush in a blocking task since the underlying HTTP clients are synchronous.
+        // Catch a panic inside the flush so its message is reported: the join handle
+        // only surfaces "task panicked", which hides the real cause.
         tokio::task::spawn_blocking(move || {
-            flush_telemetry_batch(snapshot);
+            if let Err(panic) = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
+                flush_telemetry_batch(snapshot);
+            })) {
+                tracing::error!("telemetry flush panicked: {}", panic_message(&panic));
+            }
         })
         .await
         .unwrap_or_else(|e| {
@@ -278,6 +284,17 @@ async fn telemetry_flush_loop(buffer: Arc<Mutex<TelemetryBuffer>>) {
     }
 }
 
+/// Extract a human-readable message from a caught panic payload.
+fn panic_message(panic: &(dyn std::any::Any + Send)) -> String {
+    if let Some(s) = panic.downcast_ref::<&str>() {
+        (*s).to_string()
+    } else if let Some(s) = panic.downcast_ref::<String>() {
+        s.clone()
+    } else {
+        "unknown panic".to_string()
+    }
+}
+
 fn flush_telemetry_batch(batch: TelemetryBuffer) {
     let config = Config::get();
 
@@ -971,3 +988,17 @@ impl SentryClient {
         }
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::panic_message;
+
+    #[test]
+    fn panic_message_reads_str_and_string_payloads() {
+        let str_panic = std::panic::catch_unwind(|| panic!("boom")).unwrap_err();
+        assert_eq!(panic_message(&str_panic), "boom");
+
+        let string_panic = std::panic::catch_unwind(|| panic!("count is {}", 3)).unwrap_err();
+        assert_eq!(panic_message(&string_panic), "count is 3");
+    }
+}