From 8806a717a84524bd418e9e019d065fc9bd6c9034 Mon Sep 17 00:00:00 2001
From: crusaderky <crusaderky@gmail.com>
Date: Tue, 8 Aug 2023 13:25:33 +0100
Subject: [PATCH 1/2] Better logging for anomalous task termination

---
 distributed/worker_state_machine.py | 4 +++-
 1 file changed, 3 insertions(+), 1 deletion(-)

diff --git a/distributed/worker_state_machine.py b/distributed/worker_state_machine.py
index 985db35d24b..56bba0f6d18 100644
--- a/distributed/worker_state_machine.py
+++ b/distributed/worker_state_machine.py
@@ -3656,9 +3656,11 @@ def _finish_async_instruction(
             stim = task.result()
         except asyncio.CancelledError:
             # This should exclusively happen in Worker.close()
+            logger.warning(f"Async instruction for {task} ended with CancelledError")
             return
         except BaseException:  # pragma: nocover
-            logger.exception("async instruction handlers should never raise!")
+            # This should never happen
+            logger.exception(f"Unhandled exception in async instruction for {task}")
             raise
 
         # Capture metric events in _transition_to_memory()

From 640c59202844170ba3d5af775ec7e8e925b0f5bd Mon Sep 17 00:00:00 2001
From: crusaderky <crusaderky@gmail.com>
Date: Tue, 8 Aug 2023 16:56:06 +0100
Subject: [PATCH 2/2] fix test

---
 distributed/tests/test_client.py | 24 +++++++++++++-----------
 1 file changed, 13 insertions(+), 11 deletions(-)

diff --git a/distributed/tests/test_client.py b/distributed/tests/test_client.py
index d393869cec9..2bbe15c86e6 100644
--- a/distributed/tests/test_client.py
+++ b/distributed/tests/test_client.py
@@ -5165,17 +5165,19 @@ def test_quiet_client_close(loop):
             sleep(0.200)  # stop part-way
         sleep(0.1)  # let things settle
 
-        out = logger.getvalue()
-        lines = out.strip().split("\n")
-        assert len(lines) <= 2
-        for line in lines:
-            assert (
-                not line
-                or "heartbeat from unregistered worker" in line
-                or "unaware of this worker" in line
-                or "garbage" in line
-                or set(line) == {"-"}
-            ), line
+    out = logger.getvalue()
+    lines = out.strip().split("\n")
+    unexpected_lines = [
+        line
+        for line in lines
+        if line
+        and "heartbeat from unregistered worker" not in line
+        and "unaware of this worker" not in line
+        and "garbage" not in line
+        and "ended with CancelledError" not in line
+        and set(line) != {"-"}
+    ]
+    assert not unexpected_lines, lines
 
 
 @pytest.mark.slow