fix: bug: topic reader close method freezes indefinitely

CHANGELOG.md

@@ -1,6 +1,7 @@
+* Fix topic reader `close()` hanging forever when the stream is broken with uncommitted offsets pending: the commit flush now stops as soon as the stream reports an error instead of waiting for acks that can never arrive
+
 ## 3.33.2 ##
 * Fix sync `QuerySessionPool` and `QuerySession.create()` hanging for 600 seconds and then failing with a non-retriable `TimeoutError` when a new session's attach stream fails before its first message (e.g. during a node restart); the attach error is now raised immediately, so retriable errors such as `Unavailable` are retried
-
 ## 3.33.1 ##
 * Add `ydb.convert_floats_to_embedding_bytes` to encode numeric vectors for YDB KNN queries
 

ydb/_topic_reader/topic_reader_asyncio.py

@@ -1062,8 +1062,14 @@ async def flush(self):
         for session in self._partition_sessions.values():
             futures.extend(w.future for w in session._ack_waiters)
 
-        if futures:
-            await asyncio.wait(futures)
+        if not futures:
+            return
+
+        all_acks_received = asyncio.ensure_future(asyncio.wait(futures))
+        try:
+            await asyncio.wait((all_acks_received, self._first_error), return_when=asyncio.FIRST_COMPLETED)
+        finally:
+            all_acks_received.cancel()
 
     async def close(self, flush: bool):
         if self._closed:

ydb/_topic_reader/topic_reader_asyncio_test.py

@@ -423,6 +423,33 @@ async def test_flush(self, stream, stream_reader_started: ReaderStream, partitio
 
         await wait_for_fast(stream_reader_started.close(False))
 
+    async def test_flush_stops_on_stream_error(self, stream_reader_started: ReaderStream, partition_session):
+        waiter = partition_session.add_waiter(self.partition_session_committed_offset + 1)
+
+        with pytest.raises(WaitConditionError):
+            await wait_for_fast(stream_reader_started.flush(), timeout=0.1)
+
+        stream_reader_started._set_first_error(issues.Unavailable("connection lost"))
+
+        await wait_for_fast(stream_reader_started.flush())
+        assert not waiter.future.done()
+
+        await wait_for_fast(stream_reader_started.close(False))
+
+        with pytest.raises(topic_reader_asyncio.PublicTopicReaderPartitionExpiredError):
+            waiter.future.result()
+
+    async def test_close_with_flush_on_broken_stream_does_not_hang(
+        self, stream_reader_started: ReaderStream, partition_session
+    ):
+        waiter = partition_session.add_waiter(self.partition_session_committed_offset + 1)
+        stream_reader_started._set_first_error(issues.Unavailable("connection lost"))
+
+        await wait_for_fast(stream_reader_started.close(True))
+
+        with pytest.raises(topic_reader_asyncio.PublicTopicReaderPartitionExpiredError):
+            waiter.future.result()
+
     async def test_commit_ranges_for_received_messages(
         self, stream, stream_reader_started: ReaderStream, partition_session
     ):