From 1ce6f71dc20badce04a8363cb29dc08877bf4970 Mon Sep 17 00:00:00 2001 From: "posthog[bot]" <206114724+posthog[bot]@users.noreply.github.com> Date: Sun, 16 Aug 2026 22:27:03 +0000 Subject: [PATCH] Recover from poisoned org DB client mutex A single op that panicked mid-query while holding the cached Postgres client guard poisoned the mutex for the life of the daemon process. Every later org DB write then panicked on the same lock, so metrics, notes, CAS, file change counts, and commit authorship summaries all stopped syncing with no signal beyond a stderr panic line. org_db.rs was the only place that used `.lock().expect(...)`; the rest of the repo already recovers with `unwrap_or_else(|p| p.into_inner())`. Apply the same recovery to all five lock sites. For the client mutex, treat a poisoned lock like the stale-connection case `run()` already handles: drop the entry from CONNECTIONS and redial, because a mid-query panic can leave the Postgres protocol state desynchronized. Also wrap the telemetry flush closure in catch_unwind so a panic reports its own message instead of being hidden behind the join handle's generic "task panicked". Co-Authored-By: Claude Opus 4.8 Generated-By: PostHog Desktop Task-Id: a64999a3-c9e3-4fa2-82bf-cebd47dcc615 --- src/api/org_db.rs | 43 ++++++++++++++++++++++------------ src/daemon/telemetry_worker.rs | 33 +++++++++++++++++++++++++- 2 files changed, 60 insertions(+), 16 deletions(-) diff --git a/src/api/org_db.rs b/src/api/org_db.rs index 9d0ef8e..b7cfbf6 100644 --- a/src/api/org_db.rs +++ b/src/api/org_db.rs @@ -215,13 +215,17 @@ fn connect(org_db_url: &str) -> Result { /// Get the cached client for `org_db_url`, connecting (and provisioning) lazily. fn get_or_connect(org_db_url: &str) -> Result>, 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>, 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( org_db_url: &str, op: impl FnOnce(&mut Client) -> Result, ) -> Result { 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 { diff --git a/src/daemon/telemetry_worker.rs b/src/daemon/telemetry_worker.rs index 2e4399e..b48cf85 100644 --- a/src/daemon/telemetry_worker.rs +++ b/src/daemon/telemetry_worker.rs @@ -268,8 +268,14 @@ async fn telemetry_flush_loop(buffer: Arc>) { }; // 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>) { } } +/// 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::() { + 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"); + } +}