diff --git a/api/src/common/streaming.py b/api/src/common/streaming.py index 119f222..93e0edc 100644 --- a/api/src/common/streaming.py +++ b/api/src/common/streaming.py @@ -10,6 +10,7 @@ from collections.abc import AsyncGenerator import anyio from fastapi.responses import StreamingResponse +from starlette.types import Send class ChainedStream: @@ -30,7 +31,9 @@ class ChainedStream: return await anext(self._rest) async def aclose(self) -> None: - """Закрыть исходный поток (идемпотентно: у завершённого генератора `aclose` — no-op).""" + """Закрыть исходный поток и сбросить префетченную порцию: после закрытия `__anext__` её не отдаёт + (идемпотентно: у завершённого генератора `aclose` — no-op).""" + self._first = b"" await self._rest.aclose() @@ -40,7 +43,7 @@ class ClosingStreamingResponse(StreamingResponse): Закрытие идёт под `CancelScope(shield=True)`: при обрыве клиента задача отдачи отменяется, а освобождение соединения S3 требует `await`.""" - async def stream_response(self, send) -> None: # type: ignore[no-untyped-def] # noqa: ANN001 + async def stream_response(self, send: Send) -> None: try: await super().stream_response(send) finally: diff --git a/api/tests/test_api_files_content.py b/api/tests/test_api_files_content.py index 06b05e6..40b6eea 100644 --- a/api/tests/test_api_files_content.py +++ b/api/tests/test_api_files_content.py @@ -178,6 +178,19 @@ async def test_chain_aclose_closes_source_when_never_iterated() -> None: assert source.closed +async def test_chain_does_not_yield_prefetched_chunk_after_aclose() -> None: + """PB.2: закрытая обёртка не отдаёт префетченную порцию (иначе потребитель читал бы данные закрытого потока).""" + source = _Source() + first = await anext(source.iterator, b"") + assert first == b"x" + chain = _chain(first, source.iterator) + await chain.aclose() + assert source.closed + with pytest.raises(StopAsyncIteration): + await chain.__anext__() + await chain.aclose() # повторное закрытие безопасно + + async def test_response_closes_stream_when_client_drops_after_first_chunk() -> None: """T9.1: `send` падает на втором теле (обрыв клиента) — `body_iterator` закрыт, S3-поток освобождён.""" source = _Source()