Close the async streaming response on the event loop thread (#941)

StreamingResponseSource.close() ran the producer task cancel and
response.close() from the parser executor thread. Under TLS that raced
the event loop handling a server connection abort and could raise
AttributeError from asyncio's SSL shutdown instead of the real
StreamFailureError. Schedule the cleanup with call_soon_threadsafe.

Co-authored-by: Claude Fable 5 <noreply@anthropic.com>

Author: joe-clickhouse

CHANGELOG.md

@@ -22,6 +22,7 @@
 - The native streaming response buffer again detects mid-stream server exceptions proactively. Its in-band exception scan built the markers as `__exception__<tag>` and `<tag>__exception__`, but the server separates `__exception__` from the tag with a CRLF on both markers (`__exception__\r\n<tag>` ... `<tag>\r\n__exception__`), so the scan never matched and the exception block was only recovered by the last-chunk fallback in `NativeTransform.parse_response`. When the block spanned a transport-chunk boundary that fallback saw just a fragment and surfaced a truncated or garbled error instead of the real ClickHouse exception. Both the pure Python and compiled Cython buffers are corrected. Closes [#915](https://github.com/ClickHouse/clickhouse-connect/issues/915).
 - DB-API `Cursor.description` now reports the result type's top-level nullability instead of hardcoding `null_ok=True`, including the implicit null values supported by `Variant`, `Dynamic`, and `SimpleAggregateFunction` over a nullable element type. Existing `type_code` values are unchanged, and types whose nullability is unknown report `None`. The empty-result metadata probe also recognizes leading ClickHouse comments, including nested block comments, and is best effort, so a failed probe leaves `description` empty instead of raising after the original query succeeded. Closes [#902](https://github.com/ClickHouse/clickhouse-connect/issues/902), [#907](https://github.com/ClickHouse/clickhouse-connect/issues/907), and [#909](https://github.com/ClickHouse/clickhouse-connect/issues/909).
 - Compound values stored in JSON shared data, such as arrays of objects, heterogeneous arrays, and nested arrays, are now decoded to Python objects instead of being returned as raw bytes. `Date`, `DateTime`, and `DateTime64` values in shared data, both as scalars and inside arrays, now decode as well. Closes [#897](https://github.com/ClickHouse/clickhouse-connect/issues/897).
+- `AsyncClient` no longer tears down the aiohttp response from the parser's executor thread when a query fails mid-stream. The synchronous cleanup cancelled the producer task and closed the response directly, which raced with the event loop handling the server's connection abort and could surface an `AttributeError` from asyncio's SSL shutdown on TLS connections instead of the real `StreamFailureError`. Cleanup is now scheduled onto the event loop with `call_soon_threadsafe`.
 
 ## 1.6.0, 2026-07-23
 

clickhouse_connect/driver/streaming.py

@@ -46,6 +46,7 @@ def __init__(self, response, encoding: str | None = None, exception_tag: str | N
         # Multiple accesses to .gen must return the same generator, not create new ones
         self._gen_cache: Iterator[bytes] | None = None
 
+        self._loop: asyncio.AbstractEventLoop | None = None
         self._producer_task: asyncio.Task | None = None
         self._producer_started = threading.Event()
         self._producer_error: Exception | None = None
@@ -90,6 +91,7 @@ async def producer():
                 self.queue.shutdown()
                 self._release_lease()
 
+        self._loop = loop
         self._producer_task = loop.create_task(producer())
         self._producer_started.set()
 
@@ -196,12 +198,22 @@ def close(self):
         """Synchronous cleanup resources"""
         self.queue.shutdown()
 
-        if self._producer_task and not self._producer_task.done():
-            self._producer_task.cancel()
+        def cleanup():
+            if self._producer_task and not self._producer_task.done():
+                self._producer_task.cancel()
+            if self.response and not self.response.closed:
+                if not self._producer_completed:
+                    self.response.close()
 
-        if self.response and not self.response.closed:
-            if not self._producer_completed:
-                self.response.close()
+        # Task cancellation and aiohttp response teardown must run on the event
+        # loop thread. close() is normally called from an executor thread.
+        if self._loop is not None and not self._loop.is_closed():
+            try:
+                self._loop.call_soon_threadsafe(cleanup)
+            except RuntimeError:
+                cleanup()
+        else:
+            cleanup()
         self._release_lease()
 
 

tests/unit_tests/test_streaming_source.py

@@ -1,5 +1,6 @@
 import asyncio
 import gzip
+import threading
 import time
 import zlib
 from unittest.mock import Mock
@@ -352,6 +353,51 @@ def slow_consume():
     assert result == chunks
 
 
+@pytest.mark.asyncio
+async def test_sync_close_runs_on_event_loop_thread():
+    """close() from an executor thread runs response teardown on the loop thread."""
+    loop_thread = threading.current_thread()
+    close_thread = None
+
+    class StalledContent:
+        @staticmethod
+        async def read(n=-1):
+            await asyncio.sleep(3600)
+
+    class StalledResponse:
+        def __init__(self):
+            self.content = StalledContent()
+            self.headers = {}
+            self.closed = False
+
+        def close(self):
+            nonlocal close_thread
+            close_thread = threading.current_thread()
+            self.closed = True
+
+    response = StalledResponse()
+
+    source = StreamingResponseSource(response, encoding=None)
+    loop = asyncio.get_running_loop()
+    await source.start_producer(loop)
+
+    await loop.run_in_executor(None, source.close)
+
+    for _ in range(100):
+        if response.closed:
+            break
+        await asyncio.sleep(0.01)
+
+    assert response.closed
+    assert close_thread is loop_thread
+
+    if source._producer_task is not None:
+        try:
+            await source._producer_task
+        except asyncio.CancelledError:
+            pass
+
+
 class MockTransform:
     """Mock NativeTransform."""