diff --git a/README.md b/README.md index 2ffb7d3..ca7533c 100644 --- a/README.md +++ b/README.md @@ -55,10 +55,11 @@ PostgreSQL 16, core NATS без JetStream), `web/` — фронтенд (React + OCR, `error` сервиса и ошибки маппинга не повторяются. Ошибки: 409 `FILE_NOT_RETRYABLE`, `MAX_ATTEMPTS_REACHED`, `VERSION_NOT_EDITABLE`. -Пакет включает файлы с `stored_file.consolidation_id` выбранной консолидации и ЮЛ из периметра выбранной +Пакет включает только файлы `ERROR` с `stored_file.consolidation_id` выбранной консолидации и ЮЛ из периметра выбранной версии; файлы другой консолидации и загрузки без контекста не включаются. Версия должна быть «в работе» (`INITIATED`, `DOCUMENTS_PROCESSING`, `DATA_CORRECTION`), пользовательская блокировка не нужна. -Версии одной консолидации могут видеть общий файл. Неподходящие файлы пропускаются: +Версии одной консолидации могут видеть общий файл. Другие статусы файлов не блокируются и не попадают +в ответ. Неподходящие `ERROR`-файлы пропускаются: `accepted: [{file_id, job_id, attempt}]`, `skipped: [{file_id, code}]`. Пустой пакет также отвечает 202. Для одиночного файла консолидации требуется хотя бы одна рабочая версия с ЮЛ файла в периметре. Повтор не меняет статусы версий. Два конкурентных запроса не создают две новые попытки. diff --git a/api/src/api/dependencies/files.py b/api/src/api/dependencies/files.py index 01d5969..150256a 100644 --- a/api/src/api/dependencies/files.py +++ b/api/src/api/dependencies/files.py @@ -66,10 +66,11 @@ StatementServiceDep = Annotated[StatementService, Depends(get_statement_service) def get_file_retry_service( session: SessionDep, + app_settings: SettingsDep, audit: Annotated[RequestAudit, Depends(get_request_audit)], scope: Annotated[VisibilityScope, Depends(get_scope)], ) -> FileRetryService: - return FileRetryService(session, audit, scope) + return FileRetryService(session, audit, scope, app_settings) FileRetryServiceDep = Annotated[FileRetryService, Depends(get_file_retry_service)] diff --git a/api/src/api/endpoints/files/retry.py b/api/src/api/endpoints/files/retry.py index 71081cf..b547232 100644 --- a/api/src/api/endpoints/files/retry.py +++ b/api/src/api/endpoints/files/retry.py @@ -11,13 +11,19 @@ from src.schemas.common import ErrorDetail from src.schemas.files import FileOut, RetryBatchOut router = APIRouter(dependencies=[Depends(require("DOCUMENT_EDIT"))]) -RESPONSES: dict[int | str, dict[str, Any]] = { +FILE_RESPONSES: dict[int | str, dict[str, Any]] = { 404: {"model": ErrorDetail, "description": "Нет или вне видимости"}, 409: {"model": ErrorDetail, "description": "FILE_NOT_RETRYABLE, MAX_ATTEMPTS_REACHED, VERSION_NOT_EDITABLE"}, } -@router.post("/files/{file_id}/retry", response_model=FileOut, status_code=202, responses=RESPONSES) +BATCH_RESPONSES: dict[int | str, dict[str, Any]] = { + 404: {"model": ErrorDetail, "description": "Нет или вне видимости"}, + 409: {"model": ErrorDetail, "description": "VERSION_NOT_EDITABLE"}, +} + + +@router.post("/files/{file_id}/retry", response_model=FileOut, status_code=202, responses=FILE_RESPONSES) async def retry_file(file_id: uuid.UUID, service: FileRetryServiceDep) -> FileOut: """Новая попытка tabular после TIMEOUT или NO_ACK, до max_attempts.""" return await service.retry(file_id) @@ -27,7 +33,7 @@ async def retry_file(file_id: uuid.UUID, service: FileRetryServiceDep) -> FileOu "/consolidations/versions/{version_id}/files/retry", response_model=RetryBatchOut, status_code=202, - responses=RESPONSES, + responses=BATCH_RESPONSES, ) async def retry_version_files(version_id: uuid.UUID, service: FileRetryServiceDep) -> RetryBatchOut: """Повтор подходящих файлов консолидации по периметру выбранной версии.""" diff --git a/api/src/repositories/files.py b/api/src/repositories/files.py index e51cd35..b839304 100644 --- a/api/src/repositories/files.py +++ b/api/src/repositories/files.py @@ -18,7 +18,7 @@ from sqlalchemy.ext.asyncio import AsyncSession from src.domain.policy import VisibilityScope, legal_entities_visible from src.models.access import AppUser from src.models.audit_ex import Comment -from src.models.consolidation import ConsolidationDocument, ConsolidationVersion +from src.models.consolidation import ConsolidationDocument, ConsolidationMember, ConsolidationVersion from src.models.directories import LegalEntity from src.models.enums import DocumentStatus, FileStatus, S3BucketRole from src.models.files import Document, RecognitionJob, S3Object, StoredFile @@ -298,3 +298,22 @@ async def list_desk_files( # noqa: PLR0913 - фильтр и пагинация ) files = list((await session.execute(page_stmt)).scalars().all()) return files, total + + +async def retry_candidates( + session: AsyncSession, scope: VisibilityScope, consolidation_id: uuid.UUID, version_id: uuid.UUID +) -> list[StoredFile]: + """ERROR-файлы консолидации по видимому периметру версии; без блокировки нормальных файлов.""" + members = select(ConsolidationMember.legal_entity_id).where( + ConsolidationMember.consolidation_version_id == version_id + ) + stmt = ( + _visible_files(select(StoredFile), scope) + .where( + StoredFile.consolidation_id == consolidation_id, + StoredFile.legal_entity_id.in_(members), + StoredFile.status == FileStatus.ERROR, + ) + .order_by(StoredFile.id) + ) + return list((await session.scalars(stmt)).all()) diff --git a/api/src/services/file_retry.py b/api/src/services/file_retry.py index 3d57d4a..8c25a00 100644 --- a/api/src/services/file_retry.py +++ b/api/src/services/file_retry.py @@ -9,6 +9,8 @@ from sqlalchemy.ext.asyncio import AsyncSession from src.api.dependencies.audit import RequestAudit from src.common.clock import utc_now from src.common.exceptions import ConflictError, NotFoundError +from src.contract import CONTRACT_VERSION, request_subject +from src.core.config import Settings, get_settings from src.domain.policy import IN_WORK_STATUSES, VisibilityScope from src.models.consolidation import ConsolidationMember, ConsolidationVersion from src.models.enums import AuditAction, AuditSection, FileStatus, RecognitionPipeline, RecognitionStatus @@ -20,8 +22,11 @@ from src.services.documents import file_out class FileRetryService: - def __init__(self, session: AsyncSession, audit: RequestAudit, scope: VisibilityScope) -> None: + def __init__( + self, session: AsyncSession, audit: RequestAudit, scope: VisibilityScope, settings: Settings | None = None + ) -> None: self.session, self.audit, self.scope = session, audit, scope + self.settings = settings if settings is not None else get_settings() async def _lock_files(self, files: list[StoredFile]) -> list[tuple[StoredFile, RecognitionJob | None]]: expected = {file.id: file.current_recognition_job_id for file in files} @@ -63,7 +68,6 @@ class FileRetryService: job is None or file.status is not FileStatus.ERROR or job.pipeline is not RecognitionPipeline.tabular - # Мб достаточно только по RecognitionStatus? or (job.status, job.error_code) not in {(RecognitionStatus.timeout, "TIMEOUT"), (RecognitionStatus.dlq, "NO_ACK")} or await self.session.scalar(select(exists().where(Document.stored_file_id == file.id))) @@ -78,8 +82,8 @@ class FileRetryService: new = RecognitionJob( file_id=file.id, pipeline=job.pipeline, - request_subject=job.request_subject, - schema_version=job.schema_version, + request_subject=request_subject(job.pipeline.value, self.settings.OPENBAO.NATS.request_subjects()), + schema_version=CONTRACT_VERSION, status=RecognitionStatus.queued, attempt=job.attempt + 1, max_attempts=job.max_attempts, @@ -125,9 +129,10 @@ class FileRetryService: select(ConsolidationVersion) .where( ConsolidationVersion.consolidation_id == file.consolidation_id, + ConsolidationVersion.status.in_(IN_WORK_STATUSES), ) .order_by(ConsolidationVersion.id) - .with_for_update() + .with_for_update(read=True) .execution_options(populate_existing=True) ) ).all() @@ -156,15 +161,7 @@ class FileRetryService: members = select(ConsolidationMember.legal_entity_id).where( ConsolidationMember.consolidation_version_id == version_id ) - candidates = list( - ( - await self.session.scalars( - files_repo._visible_files(select(StoredFile), self.scope) - .where(StoredFile.consolidation_id == consolidation.id, StoredFile.legal_entity_id.in_(members)) - .order_by(StoredFile.id) - ) - ).all() - ) + candidates = await files_repo.retry_candidates(self.session, self.scope, consolidation.id, version_id) locked = await self._lock_files(candidates) version = await versions_repo.lock_version(self.session, version_id) if version is None: @@ -174,7 +171,11 @@ class FileRetryService: current_members = set((await self.session.scalars(members)).all()) result = RetryBatchOut(accepted=[], skipped=[]) for file, job in locked: - if file.legal_entity_id not in current_members or file.consolidation_id != consolidation.id: + if ( + file.legal_entity_id not in current_members + or file.consolidation_id != consolidation.id + or file.status is not FileStatus.ERROR + ): continue reason = await self._reason(file, job) if reason: diff --git a/api/tests/test_api_file_retry.py b/api/tests/test_api_file_retry.py index dcecf1b..6c6c4d6 100644 --- a/api/tests/test_api_file_retry.py +++ b/api/tests/test_api_file_retry.py @@ -9,17 +9,19 @@ from sqlalchemy import delete, select from sqlalchemy.ext.asyncio import AsyncSession from src.api.dependencies.audit import RequestAudit +from src.api.dependencies.directories import get_app_settings from src.common.exceptions import ConflictError +from src.contract import CONTRACT_VERSION from src.domain.policy import VisibilityScope from src.models.audit_ex import AuditEvent -from src.models.consolidation import ConsolidationMember, ConsolidationVersion +from src.models.consolidation import Consolidation, ConsolidationMember, ConsolidationVersion from src.models.enums import ConsolidationStatus, FileStatus, RecognitionPipeline, RecognitionStatus -from src.models.files import RecognitionJob +from src.models.files import RecognitionJob, StoredFile from src.services.audit import AuditService from src.services.file_retry import FileRetryService from src.services.recognition_exchange import handle_message, publish_pending from tests.conftest import AuthClientFactory -from tests.factories import make_document, make_stored_file +from tests.factories import make_document, make_stored_file, make_version from tests.test_api_consolidation_members import _user, _version from tests.test_api_files_read import KA_RF, _rostov_entity from tests.test_recognition_exchange import FakeNats @@ -109,7 +111,7 @@ async def test_batch_filters_perimeter_and_skips_ineligible( assert response.status_code == HTTPStatus.ACCEPTED, response.text body = response.json() assert [row["file_id"] for row in body["accepted"]] == [str(good.id)] - assert body["skipped"] == [{"file_id": str(bad.id), "code": "FILE_NOT_RETRYABLE"}] + assert body["skipped"] == [] event = ( await seeded.execute( select(AuditEvent).where(AuditEvent.object_id == good.id, AuditEvent.action == "FILE_REPROCESS") @@ -236,7 +238,7 @@ async def test_concurrent_batch_retry(committed_sessionmaker) -> None: results = await asyncio.wait_for(asyncio.gather(retry(), retry()), timeout=10) assert sorted(len(result.accepted) for result in results) == [0, 1] - assert sorted(len(result.skipped) for result in results) == [0, 1] + assert all(result.skipped == [] for result in results) @pytest.mark.parametrize("change", ["status", "perimeter"]) @@ -343,3 +345,77 @@ async def test_batch_no_ack_at_attempt_limit_is_max_attempts_reached( assert response.json() == {"accepted": [], "skipped": [{"file_id": str(file.id), "code": "MAX_ATTEMPTS_REACHED"}]} jobs = await seeded.scalars(select(RecognitionJob).where(RecognitionJob.file_id == file.id)) assert [item.id for item in jobs.all()] == [job.id] + + +async def test_retry_uses_current_contract_and_subject( + seeded: AsyncSession, auth_client: AuthClientFactory, monkeypatch: pytest.MonkeyPatch +) -> None: + entity = await _rostov_entity(seeded) + file, old = await make_stored_file( + seeded, entity=entity, status=FileStatus.ERROR, job_status=RecognitionStatus.timeout + ) + old.error_code, old.max_attempts = "TIMEOUT", 3 + old.schema_version, old.request_subject = "0.9", "old.tabular.subject" + monkeypatch.setattr(get_app_settings().OPENBAO.NATS, "NATS_TABULAR_REQUEST_SUBJECT", "current.tabular.subject") + await seeded.flush() + client = await auth_client(KA_RF) + response = await client.post(f"/files/{file.id}/retry") + assert response.status_code == HTTPStatus.ACCEPTED, response.text + await seeded.refresh(file) + new = await seeded.get(RecognitionJob, file.current_recognition_job_id) + assert new is not None + assert new.schema_version == CONTRACT_VERSION and new.request_subject == "current.tabular.subject" + await seeded.refresh(old) + assert old.schema_version == "0.9" and old.request_subject == "old.tabular.subject" + + +@pytest.mark.parametrize( + "normal_status", [FileStatus.UPLOADED, FileStatus.PROCESSING, FileStatus.PROCESSED, FileStatus.SUPERSEDED] +) +async def test_batch_does_not_lock_normal_files(committed_sessionmaker, normal_status: FileStatus) -> None: + async with committed_sessionmaker() as session, session.begin(): + version, members, _, _ = await _version(session, members=1, locked_by=None) + normal, job = await make_stored_file( + session, entity=members[0], status=normal_status, job_status=RecognitionStatus.processing + ) + normal.consolidation_id = version.consolidation_id + user = await _user(session, KA_RF) + version_id, file_id, job_id, user_id = version.id, normal.id, job.id, user.id + + async def retry(): + async with committed_sessionmaker() as session, session.begin(): + audit = RequestAudit(AuditService(session), user_id, None, None) + return await FileRetryService(session, audit, VisibilityScope(None)).retry_version(version_id) + + async with committed_sessionmaker() as holder, holder.begin(): + await holder.execute(select(RecognitionJob).where(RecognitionJob.id == job_id).with_for_update()) + await holder.execute(select(StoredFile).where(StoredFile.id == file_id).with_for_update()) + result = await asyncio.wait_for(retry(), timeout=5) + assert result.accepted == [] and result.skipped == [] + + +@pytest.mark.parametrize("closed_status", [ConsolidationStatus.VERIFIED, ConsolidationStatus.CANCELLED]) +async def test_single_retry_does_not_lock_closed_versions(committed_sessionmaker, closed_status) -> None: + async with committed_sessionmaker() as session, session.begin(): + working, members, _, _ = await _version(session, members=1, locked_by=None) + consolidation = await session.get(Consolidation, working.consolidation_id) + user = await _user(session, KA_RF) + closed = await make_version( + session, consolidation=consolidation, version_no=2, members=members, initiator=user, status=closed_status + ) + file, job = await make_stored_file( + session, entity=members[0], status=FileStatus.ERROR, job_status=RecognitionStatus.timeout + ) + file.consolidation_id = working.consolidation_id + job.error_code, job.max_attempts = "TIMEOUT", 3 + file_id, user_id, closed_id = file.id, user.id, closed.id + + async def retry(): + async with committed_sessionmaker() as session, session.begin(): + audit = RequestAudit(AuditService(session), user_id, None, None) + return await FileRetryService(session, audit, VisibilityScope(None)).retry(file_id) + + async with committed_sessionmaker() as holder, holder.begin(): + await holder.execute(select(ConsolidationVersion).where(ConsolidationVersion.id == closed_id).with_for_update()) + card = await asyncio.wait_for(retry(), timeout=5) + assert card.status == FileStatus.UPLOADED diff --git a/api/tests/test_openapi_routes.py b/api/tests/test_openapi_routes.py index 8b8ec96..474e3c3 100644 --- a/api/tests/test_openapi_routes.py +++ b/api/tests/test_openapi_routes.py @@ -275,3 +275,12 @@ def test_version_member_declares_type_and_osv_accounts(schema: dict[str, Any]) - """front-api-gaps D-3: участник периметра — тип ЮЛ и счета ОСВ к периоду версии.""" member = schema["components"]["schemas"]["MemberOut"] assert {"entity_type", "osv_accounts", "checklist"} <= set(member["required"]) + + +def test_retry_routes_declare_different_conflicts(schema: dict[str, Any]) -> None: + single = schema["paths"]["/api/v1/files/{file_id}/retry"]["post"]["responses"]["409"]["description"] + batch = schema["paths"]["/api/v1/consolidations/versions/{version_id}/files/retry"]["post"]["responses"]["409"][ + "description" + ] + assert "FILE_NOT_RETRYABLE" in single and "MAX_ATTEMPTS_REACHED" in single + assert batch == "VERSION_NOT_EDITABLE" diff --git a/openspec/changes/retry-recognition-and-edit-osv-header/design.md b/openspec/changes/retry-recognition-and-edit-osv-header/design.md index f24fa77..48ee033 100644 --- a/openspec/changes/retry-recognition-and-edit-osv-header/design.md +++ b/openspec/changes/retry-recognition-and-edit-osv-header/design.md @@ -38,8 +38,9 @@ DocumentService умеет confirm; привязки централизован ### D2. Новая попытка вместо обновления прежнего job -Создать новый UUID job, queued, attempt + 1; перенести только file_id, pipeline, max_attempts и настройки -контракта/маршрута, необходимые публикатору. Ошибки, метрики, result_s3_object_id, service metadata, message ids +Создать новый UUID job, queued, attempt + 1; перенести file_id, pipeline, max_attempts. +schema_version брать из CONTRACT_VERSION, request_subject — из текущих настроек NATS, как при загрузке; +версию контракта и subject прежнего задания не копировать. Ошибки, метрики, result_s3_object_id, service metadata, message ids и времена старой попытки не копируются. requested_at — сейчас, reply_deadline формируется публикатором. Переключить current_recognition_job_id и вернуть файл из ERROR в UPLOADED. Прежний job неизменен. Это соответствует уточнённой задаче и не требует изменения дедупа парсера по старому job_id (OQ-73). @@ -47,7 +48,9 @@ DocumentService умеет confirm; привязки централизован ### D3. Пакет, частичные отказы и версии -Выбрать файлы выбранной консолидации по stored_file.consolidation_id, с ЮЛ из периметра версии и policy ЮЛ. +Отбор repositories/files.py::retry_candidates выбирает только ERROR-файлы выбранной консолидации +по stored_file.consolidation_id, с ЮЛ из периметра версии и policy ЮЛ, до любых блокировок. +Остальные статусы не блокируются и не попадают в skipped; после блокировки статус ERROR перепроверяется. Файл без документов не отбирается через consolidation_document. Дедуп по file_id. Согласованный пользователем общий набор у нескольких версий сохраняется; повтор не создаёт версионной копии файла. @@ -63,11 +66,15 @@ DocumentService умеет confirm; привязки централизован ### D4. Блокировки и гонки Retry следует порядку работающего обмена: текущее задание FOR UPDATE → файл FOR UPDATE → версии по id. +В одиночном повторе только рабочие версии берутся FOR SHARE: статус и периметр защищены от изменений, +обычное чтение не блокируется, VERIFIED/CANCELLED вообще не блокируются. После блокировки файла перечитать current_recognition_job_id; если изменился, одиночный запрос получает FILE_NOT_RETRYABLE, пакет пропускает файл. Не захватывать новое текущее задание, уже удерживая файл: это инвертирует порядок обмена. Лимит и отсутствие документов проверять заново под блокировкой. Для пакета сначала блокировать все отобранные прежние jobs по id, затем файлы по id, затем версии по id; после захвата версий перечитать статус и состав периметра. Не начинать пакет с FOR UPDATE версии. +OpenAPI одиночного повтора описывает FILE_NOT_RETRYABLE/MAX_ATTEMPTS_REACHED/VERSION_NOT_EDITABLE; +для пакетного 409 только VERSION_NOT_EDITABLE, остальные причины возвращаются в skipped. Для PATCH — файл → документ FOR NO KEY UPDATE → существующая advisory-блокировка привязок ГК → версии по id, как при confirm/delete. Проверить связанные статусы под блокировками и не использовать diff --git a/openspec/changes/retry-recognition-and-edit-osv-header/specs/primary-files/spec.md b/openspec/changes/retry-recognition-and-edit-osv-header/specs/primary-files/spec.md index e227367..6d5b908 100644 --- a/openspec/changes/retry-recognition-and-edit-osv-header/specs/primary-files/spec.md +++ b/openspec/changes/retry-recognition-and-edit-osv-header/specs/primary-files/spec.md @@ -18,11 +18,11 @@ - **THEN** 409 MAX_ATTEMPTS_REACHED, состояние не меняется ### Requirement: Пакет повторной обработки версии -`POST /api/v1/consolidations/versions/{version_id}/files/retry` SHALL рассматривать видимые файлы в статусе `ERROR` с -consolidation_id консолидации версии и ЮЛ из периметра версии; файлы других статусов в пакет не входят и не блокируются. -Для каждого допустимого файла SHALL создаваться новая попытка. Успех — 202 с accepted[{file_id, job_id, attempt}] и -skipped[{file_id, code}], где skipped — рассмотренные файлы, которые повторить нельзя (`FILE_NOT_RETRYABLE`, -`MAX_ATTEMPTS_REACHED`). Файлы без контекста консолидации MUST NOT включаться по одному лишь совпадению ЮЛ. +`POST /api/v1/consolidations/versions/{version_id}/files/retry` SHALL выбирать видимые ERROR-файлы +консолидации по периметру версии. Другие статусы не входят в пакет и не блокируются. +Ответ — 202 с accepted[{file_id, job_id, attempt}] и skipped[{file_id, code}]. В skipped SHALL входить +только ERROR-файлы, которые нельзя повторить (FILE_NOT_RETRYABLE, MAX_ATTEMPTS_REACHED). +Файлы без контекста консолидации MUST NOT включаться по совпадению ЮЛ. #### Scenario: Смешанный пакет - **WHEN** у версии есть файл timeout/TIMEOUT, OCR, успешный файл и файл с исчерпанным лимитом diff --git a/openspec/changes/retry-recognition-and-edit-osv-header/tasks.md b/openspec/changes/retry-recognition-and-edit-osv-header/tasks.md index 4600db3..4549712 100644 --- a/openspec/changes/retry-recognition-and-edit-osv-header/tasks.md +++ b/openspec/changes/retry-recognition-and-edit-osv-header/tasks.md @@ -79,4 +79,11 @@ - [x] 7.1 Ревьюер: `api/.env.test` — порт PostgreSQL 5432 по compose (свой порт — переменной `OPENBAO__DATABASE__DATABASE_URL`). - [x] 7.2 Ревьюер: дельта data-model переименована на v5.21, `openspec/config.yaml` — DBML v5.21, README/CLAUDE.md — миграции 0008/0009. - [x] 7.3 Ревьюер: синтетические тесты PATCH шапки ОСВ без `fixtures`, смешанный пакет и NO_ACK на пределе. -- [ ] 7.4 Автор: замечания комментария ревью (пакет — только файлы в `ERROR`, блокировки версий, `CONTRACT_VERSION`, репозиторий, OpenAPI). +- [x] 7.4 Автор: замечания комментария ревью (пакет — только файлы в `ERROR`, блокировки версий, `CONTRACT_VERSION`, репозиторий, OpenAPI). + +Раунд 1: все шесть замечаний к коду исправлены. Пакет отбирается репозиторием только по ERROR до блокировок, +статус перепроверяется под блокировкой; одиночный повтор берёт FOR SHARE только рабочих версий. +Новая попытка использует CONTRACT_VERSION и subject из SettingsDep. OpenAPI-конфликты разделены; +проверка пары status/error_code сохранена, вопросительный комментарий удалён. +Проверки: 101 passed (file_retry, синтетическая шапка ОСВ, OpenAPI), make lint, +openspec validate --strict, git diff --check — успешно. .env.test не менялся.