From e76c06ba3853548096533fcb569374414e07a86b Mon Sep 17 00:00:00 2001
From: Uwe Klotz <uwe.klotz@slowtec.de>
Date: Sun, 24 Sep 2023 17:26:35 +0200
Subject: [PATCH] sync: prevent lock poisoning in `watch::Receiver::wait_for`
 (#6021)

---
 tokio/src/sync/watch.rs | 21 ++++++++++++++++++---
 1 file changed, 18 insertions(+), 3 deletions(-)

diff --git a/tokio/src/sync/watch.rs b/tokio/src/sync/watch.rs
index 7287dd9fa..c5c82f90f 100644
--- a/tokio/src/sync/watch.rs
+++ b/tokio/src/sync/watch.rs
@@ -699,7 +699,7 @@ impl<T> Receiver<T> {
         changed_impl(&self.shared, &mut self.version).await
     }
 
-    /// Waits for a value that satisifes the provided condition.
+    /// Waits for a value that satisfies the provided condition.
     ///
     /// This method will call the provided closure whenever something is sent on
     /// the channel. Once the closure returns `true`, this method will return a
@@ -772,8 +772,23 @@ impl<T> Receiver<T> {
                 let has_changed = self.version != new_version;
                 self.version = new_version;
 
-                if (!closed || has_changed) && f(&inner) {
-                    return Ok(Ref { inner, has_changed });
+                if !closed || has_changed {
+                    let result = panic::catch_unwind(panic::AssertUnwindSafe(|| f(&inner)));
+                    match result {
+                        Ok(true) => {
+                            return Ok(Ref { inner, has_changed });
+                        }
+                        Ok(false) => {
+                            // Skip the value.
+                        }
+                        Err(panicked) => {
+                            // Drop the read-lock to avoid poisoning it.
+                            drop(inner);
+                            // Forward the panic to the caller.
+                            panic::resume_unwind(panicked);
+                            // Unreachable
+                        }
+                    };
                 }
             }