From 4cc08cc3de7bf2e88c986a0bca7afb088ad6f3f5 Mon Sep 17 00:00:00 2001 From: Andrey Epifancev Date: Tue, 6 Oct 2026 05:37:39 +0400 Subject: [PATCH] =?UTF-8?q?fix(consolidation):=20=D1=80=D0=B5=D0=B2=D1=8C?= =?UTF-8?q?=D1=8E=20Task=207=20=E2=80=94=20=D1=82=D0=B5=D1=81=D1=82=20?= =?UTF-8?q?=D0=BD=D0=B0=D0=B4=D0=BC=D0=BD=D0=BE=D0=B6=D0=B5=D1=81=D1=82?= =?UTF-8?q?=D0=B2=D0=B0=20=D0=B1=D0=BB=D0=BE=D0=BA=D0=B8=D1=80=D0=BE=D0=B2?= =?UTF-8?q?=D0=BE=D0=BA=20=D1=85=D1=83=D0=BA=D0=B0=20=D0=BC=D0=B0=D0=BF?= =?UTF-8?q?=D0=BF=D0=B8=D0=BD=D0=B3=D0=B0,=20=D1=83=D1=81=D1=82=D0=BE?= =?UTF-8?q?=D0=B9=D1=87=D0=B8=D0=B2=D0=BE=D1=81=D1=82=D1=8C=20=D0=B7=D0=B0?= =?UTF-8?q?=D0=B3=D1=80=D1=83=D0=B7=D0=BA=D0=B8=20=D0=BA=20=D0=BE=D1=88?= =?UTF-8?q?=D0=B8=D0=B1=D0=BA=D0=B5=20=D0=BF=D0=B5=D1=80=D0=B5=D1=85=D0=BE?= =?UTF-8?q?=D0=B4=D0=B0?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit T7.0: двухсессионный тест маппинг ∥ перевод версии в DOCUMENTS_PROCESSING (ловит мутант statuses={DOCUMENTS_PROCESSING}). T7.1 (R-10, design №43): ошибка перехода после загрузки логируется, принятые файлы и 201 сохраняются. T7.3: лог пропуска без uploaded_by только при наличии NEEDS_CHECK. T7.4: TransactionFactory импортируется из session_audit. T7.2/T7.5: дополнена строка design №39. Co-Authored-By: Claude Sonnet 5.5 Claude-Session: https://claude.ai/code/session_012vJkmBGKKqMoEMzvsXG3y2 --- api/src/services/consolidation_documents.py | 8 +-- api/src/services/primary_files.py | 47 +++++++++------- api/tests/test_api_files_upload.py | 29 ++++++++++ api/tests/test_consolidation_documents.py | 56 +++++++++++++++---- .../changes/slice-4-consolidation/design.md | 3 +- 5 files changed, 107 insertions(+), 36 deletions(-) diff --git a/api/src/services/consolidation_documents.py b/api/src/services/consolidation_documents.py index 4fdbd5f..dd0bbce 100644 --- a/api/src/services/consolidation_documents.py +++ b/api/src/services/consolidation_documents.py @@ -279,15 +279,15 @@ async def on_documents_mapped( заблокированы, автозамена прежних версий уже взяла advisory ЮЛ → старые файлы → advisory ГК → версии): порядок «задание → файл → документы → версии» не нарушается, версии блокируются последними. """ - if file.uploaded_by is None: + needs_check = [document for document in documents if document.status is DocumentStatus.NEEDS_CHECK] + if not needs_check: + return [] + if file.uploaded_by is None: # лог — только когда переход мог состояться (есть NEEDS_CHECK) logger.info( "Переход версий по маппингу пропущен: у файла нет uploaded_by (changed_by обязателен). file_id={file_id}", file_id=file.id, ) return [] - needs_check = [document for document in documents if document.status is DocumentStatus.NEEDS_CHECK] - if not needs_check: - return [] # цикл импортов: consolidation_status → consolidations → consolidation_documents (design №36) from src.services.consolidation_status import transition # noqa: PLC0415 diff --git a/api/src/services/primary_files.py b/api/src/services/primary_files.py index 85e42b6..cb285dd 100644 --- a/api/src/services/primary_files.py +++ b/api/src/services/primary_files.py @@ -18,8 +18,6 @@ audit-trail «События файлов и документов»; design D-4) import hashlib import uuid -from collections.abc import Callable -from contextlib import AbstractAsyncContextManager from dataclasses import dataclass from loguru import logger @@ -58,6 +56,7 @@ from src.services.consolidation_guard import VERSION_NOT_FOUND from src.services.documents import file_out from src.services.legal_entities import LEGAL_ENTITY_NOT_FOUND from src.services.result_mapping import RESULTS_PREFIX +from src.services.session_audit import TransactionFactory OBJECT_TYPE = "stored_file" UPLOADS_PREFIX = "uploads" @@ -69,9 +68,6 @@ FILE_NOT_FOUND = "файл не найден" # одинаково для чу FILE_HAS_CONFIRMED_DOCUMENTS = "FILE_HAS_CONFIRMED_DOCUMENTS" FILE_SUPERSEDES_OTHERS = "FILE_SUPERSEDES_OTHERS" -#: фабрика отдельной короткой транзакции (в проде — `src.db.db.session_transaction`): commit на выходе -TransactionFactory = Callable[[], AbstractAsyncContextManager[AsyncSession]] - class StorageUnavailableError(DomainError): """503 `STORAGE_UNAVAILABLE`: S3 не подключено (dev без S3) или не приняло объект.""" @@ -168,7 +164,11 @@ class PrimaryFilesService: В ОТДЕЛЬНОЙ транзакции (`session_transaction`, как аудит входа): `lock_version` (`FOR UPDATE`) → статус перечитывается — если уже не `INITIATED` (параллельный переход, аннулирование), ничего не меняется. Версия блокируется на короткое время и вне транзакции запроса, чтобы загрузка не держала её на время приёма файлов - (D-9). Транзакция запроса к этому моменту версию не трогала, взаимных ожиданий нет.""" + (D-9). Транзакция запроса к этому моменту версию не трогала, взаимных ожиданий нет. + + Приём файлов важнее побочного статуса (R-10, design №43): любая ошибка перехода (`Exception`) логируется и + НЕ меняет ответ — принятые файлы остаются, версия остаётся `INITIATED` до следующей загрузки или + подтверждения. `BaseException` (`asyncio.CancelledError`) не перехватывается.""" # импорт внутри функции — как в consolidation_documents (design №36): модули консолидации импортируют друг друга from src.services.consolidation_status import transition # noqa: PLC0415 @@ -177,20 +177,27 @@ class PrimaryFilesService: from src.db.db import session_transaction # noqa: PLC0415 factory = session_transaction - async with factory() as session: - version = await consolidations_repo.lock_version(session, version_id) - if version is None or version.status is not ConsolidationStatus.INITIATED: - return - await transition( - session, - AuditService(session), - version=version, - to=ConsolidationStatus.DOCUMENTS_PROCESSING, - changed_by=self._audit.user_id, - comment=None, - ip_address=self._audit.ip_address, - user_agent=self._audit.user_agent, - now=utc_now(), + try: + async with factory() as session: + version = await consolidations_repo.lock_version(session, version_id) + if version is None or version.status is not ConsolidationStatus.INITIATED: + return + await transition( + session, + AuditService(session), + version=version, + to=ConsolidationStatus.DOCUMENTS_PROCESSING, + changed_by=self._audit.user_id, + comment=None, + ip_address=self._audit.ip_address, + user_agent=self._audit.user_agent, + now=utc_now(), + ) + except Exception: + logger.exception( + "Не удалось перевести версию в DOCUMENTS_PROCESSING после загрузки, файлы приняты. " + "error_category=version_transition_error, version_id={version_id}", + version_id=version_id, ) def require_storage(self) -> ObjectStorage: diff --git a/api/tests/test_api_files_upload.py b/api/tests/test_api_files_upload.py index 5a68fbe..9d9d053 100644 --- a/api/tests/test_api_files_upload.py +++ b/api/tests/test_api_files_upload.py @@ -17,6 +17,7 @@ from typing import Any import httpx import pytest from fastapi import FastAPI +from loguru import logger from sqlalchemy import func, select, update from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker @@ -736,3 +737,31 @@ async def test_upload_without_version_does_not_touch_versions( assert response.status_code == HTTP_CREATED, response.text assert await _version_state(seeded, version.id) is ConsolidationStatus.INITIATED assert len(storage.objects) == 1 + + +@pytest.mark.integration +async def test_upload_keeps_accepted_files_when_version_transition_fails( + auth_client: AuthClientFactory, seeded: AsyncSession, storage: FakeStorage, monkeypatch: pytest.MonkeyPatch +) -> None: + """R-10 (design №43): приём файлов важнее побочного статуса. Переход версии падает исключением — ответ 201, принятый + файл сохранён, версия остаётся `INITIATED`, ошибка — в логе.""" + + async def broken_transition(*args: Any, **kwargs: Any) -> None: + raise RuntimeError("переход недоступен") + + monkeypatch.setattr("src.services.consolidation_status.transition", broken_transition) + version, (member, *_), _ = await _version(seeded) + client = await auth_client(KA_RF) + messages: list[str] = [] + sink = logger.add(lambda message: messages.append(str(message)), level="ERROR") + try: + response = await _post_in_version(client, member.id, version.id, ("osv.xls", _xls("transition-fails"))) + finally: + logger.remove(sink) + + assert response.status_code == HTTP_CREATED, response.text + [accepted] = response.json()["accepted"] + assert (await seeded.get(StoredFile, uuid.UUID(accepted["file_id"]))) is not None + assert len(storage.objects) == 1 + assert await _version_state(seeded, version.id) is ConsolidationStatus.INITIATED + assert any(str(version.id) in message and "version_transition_error" in message for message in messages) diff --git a/api/tests/test_consolidation_documents.py b/api/tests/test_consolidation_documents.py index 700dd62..faa7ee5 100644 --- a/api/tests/test_consolidation_documents.py +++ b/api/tests/test_consolidation_documents.py @@ -936,6 +936,18 @@ async def test_on_documents_mapped_without_uploader_logs_and_skips(seeded: Async assert any(str(file.id) in message and "uploaded_by" in message for message in messages) +async def _mapped_in_own_transaction(maker: async_sessionmaker[AsyncSession], data: dict[str, Any]) -> list[uuid.UUID]: + """Хук маппинга файла `data["file_id"]` в своей транзакции (как цикл маппинга: файл и документы уже есть).""" + async with maker() as session, session.begin(): + file = await session.get(StoredFile, data["file_id"]) + document = await session.get(Document, data["document_id"]) + assert file is not None + assert document is not None + return await consolidation_documents.on_documents_mapped( + session, file=file, documents=[document], now=utc_now() + ) + + @pytest.mark.integration async def test_on_documents_mapped_waits_for_member_edit_and_sees_new_member( committed_sessionmaker: async_sessionmaker[AsyncSession], @@ -952,21 +964,43 @@ async def test_on_documents_mapped_waits_for_member_edit_and_sees_new_member( assert file is not None file.uploaded_by = data["user_id"] - async def mapped() -> list[uuid.UUID]: - async with maker() as session, session.begin(): - file = await session.get(StoredFile, data["file_id"]) - document = await session.get(Document, data["document_id"]) - assert file is not None - assert document is not None - return await consolidation_documents.on_documents_mapped( - session, file=file, documents=[document], now=utc_now() - ) - async with maker() as editor, editor.begin(): assert await repo.lock_version(editor, data["version_id"]) is not None editor.add(ConsolidationMember(consolidation_version_id=data["version_id"], legal_entity_id=data["owner_id"])) await editor.flush() - hook = asyncio.create_task(mapped()) + hook = asyncio.create_task(_mapped_in_own_transaction(maker, data)) + await _wait_for_lock_waiter(maker) + assert not hook.done() + moved = await asyncio.wait_for(hook, LOCK_WAIT_SECONDS) + + assert moved == [data["version_id"]] + async with maker() as session: + assert await _status(session, data["version_id"]) is ConsolidationStatus.DATA_CORRECTION + + +@pytest.mark.integration +async def test_on_documents_mapped_waits_for_transition_to_documents_processing( + committed_sessionmaker: async_sessionmaker[AsyncSession], +) -> None: + """R-2/№40, маппинг ∥ загрузка (`after_upload_to_version`): версия `INITIATED` держится транзакцией, которая + переводит её в `DOCUMENTS_PROCESSING` (не зафиксировано). Хук блокирует надмножество версий «в работе» (а не только + `DOCUMENTS_PROCESSING`), поэтому встаёт в ожидание строки версии и после commit видит `DOCUMENTS_PROCESSING` — + переход в `DATA_CORRECTION` не теряется. Хук, ищущий версии только в `DOCUMENTS_PROCESSING`, снимок `INITIATED` + пропустил бы без ожидания и вернул бы `[]`.""" + maker = committed_sessionmaker + data = await _committed_setup(maker, members=2, spare=0) # владелец документа — член периметра, версия INITIATED + async with maker() as session, session.begin(): + file = await session.get(StoredFile, data["file_id"]) + assert file is not None + file.uploaded_by = data["user_id"] + + async with maker() as uploader, uploader.begin(): + version = await repo.lock_version(uploader, data["version_id"]) + assert version is not None + assert version.status is ConsolidationStatus.INITIATED + version.status = ConsolidationStatus.DOCUMENTS_PROCESSING + await uploader.flush() + hook = asyncio.create_task(_mapped_in_own_transaction(maker, data)) await _wait_for_lock_waiter(maker) assert not hook.done() moved = await asyncio.wait_for(hook, LOCK_WAIT_SECONDS) diff --git a/openspec/changes/slice-4-consolidation/design.md b/openspec/changes/slice-4-consolidation/design.md index c338e24..44e8358 100644 --- a/openspec/changes/slice-4-consolidation/design.md +++ b/openspec/changes/slice-4-consolidation/design.md @@ -223,7 +223,8 @@ legal_entity_name, account_code, missing_quarters[]}}`. Фильтры `GET …/ | 36 | 2026-10-06 | Переход версии при подтверждении пишется `transition(...)` с `user_id` подтвердившего, но без IP/user-agent (сигнатура хука `on_document_confirmed(session, audit: AuditService, *, document, user_id, now)` фиксирована планом); `transition` импортируется внутри хука (цикл `consolidation_status → consolidations → consolidation_documents`) | audit-trail требует для автоматического перехода только `user_id` инициатора | в `CONSOLIDATION_STATUS_CHANGE` автоматического перехода нет IP/UA — они есть в соседнем `DOCUMENT_CONFIRM` той же транзакции | | 37 | 2026-10-06 | Отдельная транзакция перехода после загрузки (`after_upload_to_version`) берётся из `app.state.session_audit.transaction` (та же фабрика `session_transaction`, что у аудита входа; без неё — `src.db.db.session_transaction`); `get_primary_files_service` передаёт её в `PrimaryFilesService(…, transaction)` | отдельной зависимости «фабрика транзакций» в приложении нет, а фикстуры тестов уже подставляют фабрику в `SessionAudit`; не меняем `app.py` и `conftest` | если `SessionAudit` когда-либо уйдёт из `app.state`, сервис молча возьмёт боевой `session_transaction` (в тестах это была бы чужая БД) | | 38 | 2026-10-06 | Переход после загрузки пишет `CONSOLIDATION_STATUS_CHANGE` с IP и user-agent запроса (в отличие от хука подтверждения, №36); `transition` вызывается из сервиса загрузки с `RequestAudit`-контекстом, но в своей транзакции и своим `AuditService(session)` | сервис знает IP/UA запроса, а `transition` их принимает; потери информации нет | нет | -| 39 | 2026-10-06 | Версия в статусе `DOCUMENTS_PROCESSING`/`DATA_CORRECTION` при загрузке не меняется, `after_upload_to_version` — no-op; если между проверкой и переходом версию аннулировали/перевели, переход не выполняется (статус перечитывается под `FOR UPDATE`), загрузка при этом остаётся принятой (201) | проверка до приёма и переход после — две разные транзакции, окно между ними неустранимо без блокировки версии, которую спека запрещает | версия, аннулированная в окне, остаётся аннулированной; файл принят и виден в ЮЛ, но к версии не привязан (привязка — только подтверждением) | +| 39 | 2026-10-06 | Версия в статусе `DOCUMENTS_PROCESSING`/`DATA_CORRECTION` при загрузке не меняется, `after_upload_to_version` — no-op; если между проверкой и переходом версию аннулировали/перевели, переход не выполняется (статус перечитывается под `FOR UPDATE`), загрузка при этом остаётся принятой (201) | проверка до приёма и переход после — две разные транзакции, окно между ними неустранимо без блокировки версии, которую спека запрещает | версия, аннулированная в окне, остаётся аннулированной; файл принят и виден в ЮЛ, но к версии не привязан (привязка — только подтверждением). Окно покрывает и удаление ЮЛ из периметра между проверкой и переходом: версия всё равно перейдёт в `DOCUMENTS_PROCESSING` — допустимо. Отдельная транзакция перехода открывает второе подключение к БД, пока запрос держит своё: при множестве параллельных загрузок в контексте версии это давит на пул (по умолчанию 5+10) — пока теоретически | | 40 | 2026-10-06 | `on_documents_mapped` блокирует через `lock_versions_for_entity` надмножество по `IN_WORK_STATUSES` и фильтрует `DOCUMENTS_PROCESSING` ПОСЛЕ блокировки, а не передаёт `statuses={DOCUMENTS_PROCESSING}`; вызывается в `_map_candidate` ПОСЛЕ `map_file`, то есть после автозамены | при фильтре статуса в блокирующем запросе версия, которую параллельно переводят `INITIATED → DOCUMENTS_PROCESSING`, не ждалась бы; порядок «файл → документы → advisory ГК → версии» сохраняется, автозамена внутри `map_file` берёт advisory ЮЛ → старые файлы → advisory ГК → версии раньше хука; вызов до автозамены держал бы версии и затем ждал старые файлы (встречная блокировка с подтверждением) | блокируется чуть больше версий ГК (все «в работе»), на время короткой транзакции маппинга | | 41 | 2026-10-06 | Хук получает документы файла отдельным `SELECT` по `stored_file_id` после `map_file` (возвращаемое значение `map_file` — предупреждения — не менялось) | сигнатура `map_file` и её тесты не трогаются; документы только что записаны в этой же транзакции | один лишний запрос на файл | | 42 | 2026-10-06 | Переход по маппингу — только если среди документов файла есть `NEEDS_CHECK`, подходящий версии по `fits` (`fits` — D-5: период, год, член); статус файла/документа в других статусах (`RECOGNIZED`) переход не вызывает, привязки не создаются | спека osv-result-mapping | версия с неподходящим по периоду `NEEDS_CHECK` документом остаётся в `DOCUMENTS_PROCESSING` до подтверждения подходящего | +| 43 | 2026-10-06 | R-10: ошибка (`Exception`) отдельной транзакции перехода после загрузки (`after_upload_to_version`) логируется (`logger.exception`, `error_category=version_transition_error`) и не меняет ответ: 201 с принятыми файлами; `BaseException` (`asyncio.CancelledError`) не перехватывается | приём файлов важнее побочного статуса; иначе 500 откатил бы транзакцию запроса, а принятые объекты S3 остались бы сиротами (как в `SessionAudit`, ошибка записи побочного события ответ не меняет) | версия может остаться `INITIATED` после загрузки до следующей загрузки или подтверждения документа; сбой виден только в логе |