fix(files): детерминированное освобождение S3-потока при отдаче исходного файла (T9.1, T9.2, T9.4, T9.5)

- ClosingStreamingResponse закрывает body_iterator в finally (под shield), ChainedStream закрывает
  исходный поток и до начала чтения; заголовок X-Content-Type-Options: nosniff;
- ASCII-имя в Content-Disposition строится из file_type, а не из расширения исходного имени;
- ObjectStorage.stream типизирован как AsyncGenerator, убраны два type: ignore.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_012vJkmBGKKqMoEMzvsXG3y2

api/src/api/endpoints/files/read.py

@@ -16,6 +16,7 @@ from fastapi.responses import StreamingResponse
 
 from src.api.dependencies.auth import require
 from src.api.dependencies.files import PrimaryFilesServiceDep
+from src.common.streaming import ClosingStreamingResponse
 from src.schemas.common import ErrorDetail, Page, PageParams
 from src.schemas.files import FileOut
 
@@ -95,7 +96,9 @@ async def get_file_content(
 ) -> StreamingResponse:
     """Содержимое исходного файла из S3 потоком (в память целиком не читается); `inline` — для просмотра в браузере."""
     stream, media_type, header = await service.content(file_id, disposition)
-    return StreamingResponse(stream, media_type=media_type, headers={"Content-Disposition": header})
+    return ClosingStreamingResponse(
+        stream, media_type=media_type, headers={"Content-Disposition": header, "X-Content-Type-Options": "nosniff"}
+    )
 
 
 @router.delete(

api/src/common/streaming.py

@@ -0,0 +1,50 @@
+"""Потоковая отдача файла с детерминированным освобождением источника (T9.1, Review Focus 4 плана Среза 3b).
+
+Starlette `StreamingResponse.stream_response` не вызывает `aclose()` у `body_iterator`: при обрыве клиента или если тело
+не начали итерировать поток S3 освобождался бы только сборщиком мусора. `ClosingStreamingResponse` закрывает итератор
+в `finally`, а `ChainedStream` гарантирует, что `aclose` закрывает и исходный поток, даже если обёртку не начинали
+читать (у неначатого асинхронного генератора `finally` не выполняется, поэтому обёртка — объект, а не генератор).
+"""
+
+from collections.abc import AsyncGenerator
+
+import anyio
+from fastapi.responses import StreamingResponse
+
+
+class ChainedStream:
+    """Уже прочитанная первая порция и остальной поток `rest`; `aclose` всегда закрывает `rest` (там `close` + `release`
+    ответа S3), в том числе при завершении, ошибке и обрыве потребителя, начатом и не начатом чтении."""
+
+    def __init__(self, first: bytes, rest: AsyncGenerator[bytes, None]) -> None:
+        self._first = first
+        self._rest = rest
+
+    def __aiter__(self) -> "ChainedStream":
+        return self
+
+    async def __anext__(self) -> bytes:
+        if self._first:
+            chunk, self._first = self._first, b""
+            return chunk
+        return await anext(self._rest)
+
+    async def aclose(self) -> None:
+        """Закрыть исходный поток (идемпотентно: у завершённого генератора `aclose` — no-op)."""
+        await self._rest.aclose()
+
+
+class ClosingStreamingResponse(StreamingResponse):
+    """`StreamingResponse`, закрывающий `body_iterator` (`aclose`) после отдачи, ошибки или обрыва клиента.
+
+    Закрытие идёт под `CancelScope(shield=True)`: при обрыве клиента задача отдачи отменяется, а освобождение
+    соединения S3 требует `await`."""
+
+    async def stream_response(self, send) -> None:  # type: ignore[no-untyped-def]  # noqa: ANN001
+        try:
+            await super().stream_response(send)
+        finally:
+            aclose = getattr(self.body_iterator, "aclose", None)
+            if aclose is not None:
+                with anyio.CancelScope(shield=True):
+                    await aclose()

api/src/core/clients.py

@@ -2,7 +2,7 @@
 по docs/context/03-architecture.md (референс — reference-old-arm/api/src/core/clients.py).
 """
 
-from collections.abc import AsyncIterator
+from collections.abc import AsyncGenerator
 from io import BytesIO
 
 from miniopy_async import Minio
@@ -47,7 +47,7 @@ class S3Storage:
             response.close()
             await response.release()
 
-    async def stream(self, obj_name: str, chunk_size: int = 1 << 20) -> AsyncIterator[bytes]:
+    async def stream(self, obj_name: str, chunk_size: int = 1 << 20) -> AsyncGenerator[bytes, None]:
         """Объект порциями; `NoSuchKey` → `ObjectNotFoundError` до первой порции; ответ отпускается и при обрыве."""
         try:
             response = await self.client.get_object(self.bucket, obj_name)

api/src/core/storage.py

@@ -4,7 +4,7 @@
 `services/directory_import.py`: тот же интерфейс нужен загрузке файлов и маппингу результатов (Срез 2c, D-12).
 """
 
-from collections.abc import AsyncIterator
+from collections.abc import AsyncGenerator
 from typing import Protocol
 
 
@@ -23,7 +23,7 @@ class ObjectStorage(Protocol):
         """Содержимое объекта; отсутствие объекта — `ObjectNotFoundError`."""
         ...
 
-    def stream(self, obj_name: str, chunk_size: int = 1 << 20) -> AsyncIterator[bytes]:
+    def stream(self, obj_name: str, chunk_size: int = 1 << 20) -> AsyncGenerator[bytes, None]:
         """Объект порциями по `chunk_size` байт; отсутствие объекта — `ObjectNotFoundError` до первой порции.
 
         Закрытие итератора (`aclose`) освобождает соединение с S3."""

api/src/services/primary_files.py

@@ -18,7 +18,7 @@ audit-trail «События файлов и документов»; design D-4)
 
 import hashlib
 import uuid
-from collections.abc import AsyncIterator
+from collections.abc import AsyncGenerator
 from dataclasses import dataclass
 from typing import Literal
 
@@ -30,7 +30,8 @@ from starlette import status as http_status
 from src.api.dependencies.audit import RequestAudit
 from src.common.clock import utc_now
 from src.common.exceptions import ConflictError, DomainError, DomainValidationError, NotFoundError
-from src.common.filenames import CONTENT_TYPES, content_disposition, extension, safe_filename
+from src.common.filenames import CONTENT_TYPES, content_disposition, safe_filename
+from src.common.streaming import ChainedStream
 from src.contract import CONTRACT_VERSION, request_subject
 from src.core.config import Settings
 from src.core.storage import ObjectNotFoundError, ObjectStorage
@@ -116,16 +117,10 @@ class RejectedFile:
         )
 
 
-async def _chain(first: bytes, rest: AsyncIterator[bytes]) -> AsyncIterator[bytes]:
-    """Первая (уже прочитанная) порция и остальной поток; при завершении, ошибке и обрыве потребителя поток S3
-    закрывается (`aclose` — там `close` + `release` ответа), соединение не остаётся занятым."""
-    try:
-        if first:
-            yield first
-        async for chunk in rest:
-            yield chunk
-    finally:
-        await rest.aclose()  # type: ignore[attr-defined]
+def _chain(first: bytes, rest: AsyncGenerator[bytes, None]) -> ChainedStream:
+    """Первая (уже прочитанная) порция и остальной поток; `aclose` закрывает поток S3 (там `close` + `release`
+    ответа) — соединение не остаётся занятым при завершении, ошибке и обрыве потребителя (в т. ч. не начатом)."""
+    return ChainedStream(first, rest)
 
 
 def upload_key(file_id: uuid.UUID, job_id: uuid.UUID, name: str) -> str:
@@ -255,7 +250,7 @@ class PrimaryFilesService:
 
     async def content(
         self, file_id: uuid.UUID, disposition: Literal["attachment", "inline"] = "attachment"
-    ) -> tuple[AsyncIterator[bytes], str, str]:
+    ) -> tuple[ChainedStream, str, str]:
         """Исходный файл потоком: `(поток, Content-Type, Content-Disposition)` (spec documents-api, D-10).
 
         Файл любого статуса (в т. ч. `SUPERSEDED`) в области видимости; чужой и несуществующий — 404. Ключ и имя
@@ -275,11 +270,11 @@ class PrimaryFilesService:
         except ObjectNotFoundError:
             raise FileContentNotFoundError() from None
         except BaseException:
-            await iterator.aclose()  # type: ignore[attr-defined]
+            await iterator.aclose()
             raise
         name = file.original_name or ""
-        suffix = extension(name) or file.file_type.value.lower()
-        header = content_disposition(name or f"file.{suffix}", disposition=disposition, ascii_fallback=f"file.{suffix}")
+        ascii_name = f"file.{file.file_type.value.lower()}"
+        header = content_disposition(name or ascii_name, disposition=disposition, ascii_fallback=ascii_name)
         media_type = CONTENT_TYPES.get(file.file_type.value.lower(), "application/octet-stream")
         return _chain(first, iterator), media_type, header
 

api/tests/conftest.py

@@ -13,7 +13,7 @@ S3 — `FakeStorage` (объекты в словаре).
 """
 
 import uuid
-from collections.abc import AsyncIterator, Awaitable, Callable
+from collections.abc import AsyncGenerator, AsyncIterator, Awaitable, Callable
 from contextlib import AbstractAsyncContextManager, AsyncExitStack, asynccontextmanager
 from functools import partial
 from pathlib import Path
@@ -222,7 +222,7 @@ class FakeStorage:
         except KeyError:
             raise ObjectNotFoundError(obj_name) from None
 
-    async def stream(self, obj_name: str, chunk_size: int = 1 << 20) -> AsyncIterator[bytes]:
+    async def stream(self, obj_name: str, chunk_size: int = 1 << 20) -> AsyncGenerator[bytes, None]:
         try:
             data = self.objects[obj_name]
         except KeyError:

api/tests/test_api_files_content.py

@@ -1,12 +1,13 @@
 """Исходный файл потоком (spec documents-api «Исходный файл»; design D-10)."""
 
-from collections.abc import AsyncIterator
+from collections.abc import AsyncGenerator
 
 import pytest
 from httpx import AsyncClient
 from sqlalchemy import select
 from sqlalchemy.ext.asyncio import AsyncSession
 
+from src.common.streaming import ClosingStreamingResponse
 from src.core.storage import ObjectNotFoundError
 from src.models.enums import FileStatus, FileType, RecognitionStatus
 from src.models.files import S3Object, StoredFile
@@ -20,6 +21,7 @@ PDF = b"%PDF-1.4\n" + bytes(range(256)) * 4096  # ~1 МиБ: проверка п
 HTTP_OK = 200
 HTTP_NOT_FOUND = 404
 HTTP_FORBIDDEN = 403
+SECOND_BODY = 3
 
 
 async def _file(
@@ -46,6 +48,7 @@ async def test_inline_pdf(app, seeded: AsyncSession, auth_client: AuthClientFact
     assert response.headers["content-type"] == "application/pdf"
     assert response.headers["content-disposition"] == "inline; filename=\"file.pdf\"; filename*=UTF-8''aep_6.pdf"
     assert response.content == PDF
+    assert response.headers["x-content-type-options"] == "nosniff"  # T9.5
 
 
 async def test_cyrillic_name_and_missing(app, seeded: AsyncSession, auth_client: AuthClientFactory) -> None:  # noqa: ANN001
@@ -136,7 +139,7 @@ class _Source:
         self._fail_after = fail_after
         self.iterator = self._run()
 
-    async def _run(self) -> AsyncIterator[bytes]:
+    async def _run(self) -> AsyncGenerator[bytes, None]:
         try:
             for index in range(3):
                 if self._fail_after == index:
@@ -165,3 +168,54 @@ async def test_chain_closes_source_on_error_and_yields_all() -> None:
     complete = _Source()
     assert b"".join([chunk async for chunk in _chain(await anext(complete.iterator, b""), complete.iterator)]) == b"xxx"
     assert complete.closed
+
+
+async def test_chain_aclose_closes_source_when_never_iterated() -> None:
+    """T9.1: поток не начали отдавать (первая порция уже прочитана) — `aclose` всё равно закрывает S3-поток."""
+    source = _Source()
+    chain = _chain(await anext(source.iterator, b""), source.iterator)
+    await chain.aclose()
+    assert source.closed
+
+
+async def test_response_closes_stream_when_client_drops_after_first_chunk() -> None:
+    """T9.1: `send` падает на втором теле (обрыв клиента) — `body_iterator` закрыт, S3-поток освобождён."""
+    source = _Source()
+    chain = _chain(await anext(source.iterator, b""), source.iterator)
+    response = ClosingStreamingResponse(chain, media_type="application/pdf")
+    sent: list[str] = []
+
+    async def send(message: dict) -> None:  # type: ignore[type-arg]
+        sent.append(message["type"])
+        if len(sent) == SECOND_BODY:  # start, первая порция, вторая порция
+            raise OSError("клиент отключился")
+
+    with pytest.raises(OSError, match="клиент"):
+        await response.stream_response(send)
+    assert source.closed
+
+
+async def test_response_closes_stream_when_send_fails_at_start() -> None:
+    """T9.1: тело не начали итерировать (падает уже отправка заголовков) — поток тоже закрыт."""
+    source = _Source()
+    chain = _chain(await anext(source.iterator, b""), source.iterator)
+    response = ClosingStreamingResponse(chain)
+
+    async def send(message: dict) -> None:  # type: ignore[type-arg]
+        raise OSError("клиент отключился")
+
+    with pytest.raises(OSError, match="клиент"):
+        await response.stream_response(send)
+    assert source.closed
+
+
+async def test_ascii_extension_from_file_type(app, seeded: AsyncSession, auth_client: AuthClientFactory) -> None:  # noqa: ANN001
+    """T9.2: ASCII-имя берётся из `file_type`, а не из расширения исходного имени (нестандартное/кириллическое)."""
+    storage = FakeStorage()
+    app.state.storage = storage
+    file = await _file(seeded, storage, "отчёт.экс", FileType.XLSX)
+    client = await auth_client(KA_RF)
+    response = await client.get(f"/files/{file.id}/content")
+    disposition = response.headers["content-disposition"]
+    assert disposition.startswith("attachment; filename=\"file.xlsx\"; filename*=UTF-8''")
+    assert disposition.isascii()