diff --git a/api/src/repositories/consolidations.py b/api/src/repositories/consolidations.py index 2ccfa13..65e44f7 100644 --- a/api/src/repositories/consolidations.py +++ b/api/src/repositories/consolidations.py @@ -10,7 +10,7 @@ import uuid from collections import defaultdict from collections.abc import Collection, Sequence -from sqlalchemy import Select, exists, func, select +from sqlalchemy import Select, exists, false, func, select from sqlalchemy.ext.asyncio import AsyncSession from src.domain.checklist import EntityFilesState @@ -132,14 +132,11 @@ async def lock_versions_for_entity( ).scalar_one_or_none() if group_id is not None: await lock_group_bindings(session, group_id) - entity_group = select(LegalEntity.company_group_id).where(LegalEntity.id == entity_id).scalar_subquery() + in_group = Consolidation.company_group_id == group_id if group_id is not None else false() superset = ( select(ConsolidationVersion.id) .join(Consolidation, Consolidation.id == ConsolidationVersion.consolidation_id) - .where( - ConsolidationVersion.status.in_(statuses), - is_member | (Consolidation.company_group_id == entity_group), - ) + .where(ConsolidationVersion.status.in_(statuses), is_member | in_group) .order_by(ConsolidationVersion.id) .with_for_update(of=ConsolidationVersion) ) diff --git a/api/tests/test_api_consolidation_members.py b/api/tests/test_api_consolidation_members.py index 8c887a2..24750fd 100644 --- a/api/tests/test_api_consolidation_members.py +++ b/api/tests/test_api_consolidation_members.py @@ -158,8 +158,8 @@ async def test_add_member_really_binds_confirmed_documents( auth_client: AuthClientFactory, seeded: AsyncSession ) -> None: """Добавленное ЮЛ с подтверждённой ОСВ за квартал версии: `consolidation_document` создан без шпионов; статус - версии подхват не меняет; подтверждённая ОСВ другого года не подхватывается.""" - version, inside, outside, _ = await _version(seeded) + версии подхват не меняет (`INITIATED` остаётся); подтверждённая ОСВ другого года не подхватывается.""" + version, inside, outside, _ = await _version(seeded, status=ConsolidationStatus.INITIATED) fitting = await make_confirmed_osv( seeded, entity=outside[0], @@ -184,7 +184,7 @@ async def test_add_member_really_binds_confirmed_documents( assert response.status_code == HTTP_OK, response.text assert await _bound_documents(seeded, version.id) == {fitting.id} - assert response.json()["status"] == ConsolidationStatus.DATA_CORRECTION.value + assert response.json()["status"] == ConsolidationStatus.INITIATED.value @pytest.mark.integration diff --git a/api/tests/test_consolidation_documents.py b/api/tests/test_consolidation_documents.py index 66964ff..e764b00 100644 --- a/api/tests/test_consolidation_documents.py +++ b/api/tests/test_consolidation_documents.py @@ -37,12 +37,13 @@ from src.models.enums import ( PeriodQuarter, RecognitionStatus, ) -from src.models.files import Document +from src.models.files import Document, StoredFile from src.repositories import consolidations as repo from src.repositories import files as files_repo from src.services import consolidation_documents from src.services.audit import AuditService from src.services.consolidation_documents import fits +from src.services.result_mapping import supersede_previous from tests.conftest import AuthClientFactory from tests.factories import ( make_company_group, @@ -403,15 +404,31 @@ async def test_create_picks_up_documents_confirmed_before(seeded: AsyncSession, assert len(await _history(seeded, version_id)) == 1 +async def _latest_with_smaller_id( + session: AsyncSession, entity: LegalEntity, *, account: str = ACCOUNT, equal: bool = False +) -> tuple[Document, Document]: + """Два `CONFIRMED` документа одного ключа (квартальная ОСВ за Q3): `(победитель, проигравший)`. + + Без `equal` более поздний `confirmed_at` получает документ с МЕНЬШИМ `id` — выбор «по наибольшему id» вместо + «по самому позднему `confirmed_at`» проиграет. С `equal` время одинаковое — побеждает больший `id` (решение №35). + """ + first = await _confirmed(session, entity, quarter=Q3, account=account) + second = await _confirmed(session, entity, quarter=Q3, account=account) + smaller, larger = sorted((first, second), key=lambda document: document.id) + now = utc_now() + smaller.confirmed_at = now + larger.confirmed_at = now if equal else now - datetime.timedelta(hours=1) + await session.flush() + return (larger, smaller) if equal else (smaller, larger) + + @pytest.mark.integration async def test_create_picks_latest_confirmed_of_same_key(seeded: AsyncSession, auth_client: AuthClientFactory) -> None: """«Две подтверждённые версии одного документа»: два незаменённых файла с `CONFIRMED` ОСВ 62 за Q3 — привязан документ с более поздним `confirmed_at`.""" group, entities = await _group(seeded) gsk = await _gsk(seeded, group, entities) - now = utc_now() - await _confirmed(seeded, entities[0], quarter=Q3, account="62", confirmed_at=now - datetime.timedelta(hours=1)) - later = await _confirmed(seeded, entities[0], quarter=Q3, account="62", confirmed_at=now) + later, _ = await _latest_with_smaller_id(seeded, entities[0], account="62") client = await auth_client(KA_RF) response = await client.post("/consolidations", json=_create_body(group, gsk)) @@ -458,9 +475,7 @@ async def test_bind_confirmed_replaces_older_binding_of_same_key(seeded: AsyncSe version = await _version(seeded, group, entities) consolidation = await seeded.get(Consolidation, version.consolidation_id) assert consolidation is not None - now = utc_now() - older = await _confirmed(seeded, entities[0], quarter=Q3, confirmed_at=now - datetime.timedelta(hours=1)) - newer = await _confirmed(seeded, entities[0], quarter=Q3, confirmed_at=now) + newer, older = await _latest_with_smaller_id(seeded, entities[0]) await _bind(seeded, version, older) count = await consolidation_documents.bind_confirmed( @@ -471,6 +486,23 @@ async def test_bind_confirmed_replaces_older_binding_of_same_key(seeded: AsyncSe assert await _bound(seeded, version.id) == {newer.id} +@pytest.mark.integration +async def test_bind_confirmed_tie_on_confirmed_at_takes_larger_id(seeded: AsyncSession) -> None: + """Решение №35: при равном `confirmed_at` привязывается документ с большим `id`.""" + group, entities = await _group(seeded) + version = await _version(seeded, group, entities) + consolidation = await seeded.get(Consolidation, version.consolidation_id) + assert consolidation is not None + winner, _ = await _latest_with_smaller_id(seeded, entities[0], equal=True) + + count = await consolidation_documents.bind_confirmed( + seeded, version=version, consolidation=consolidation, entity_ids=[entities[0].id] + ) + + assert count == 1 + assert await _bound(seeded, version.id) == {winner.id} + + # --- гонки (настоящие параллельные транзакции) ------------------------------------------------------------------------ @@ -679,3 +711,102 @@ async def test_confirm_waits_for_create_and_binds_new_version( assert await _committed_bound(maker, version_id) == {data["document_id"]} async with maker() as session: assert await _status(session, version_id) is ConsolidationStatus.DATA_CORRECTION + + +async def _committed_replacement(maker: async_sessionmaker[AsyncSession], data: dict[str, Any]) -> uuid.UUID: + """У ЮЛ документа `data` — прежний файл с `CONFIRMED` квартальной ОСВ 60 за Q2 (тот же ключ автозамены, что у + документа `data`, файл которого станет новым). Возвращает id прежнего документа.""" + async with maker() as session, session.begin(): + entity = await session.get(LegalEntity, data["owner_id"]) + assert entity is not None + return (await _confirmed(session, entity, quarter=Q2)).id + + +async def _supersede(maker: async_sessionmaker[AsyncSession], data: dict[str, Any]) -> list[uuid.UUID]: + """Автозамена в отдельной транзакции: новый файл — файл документа `data` (как в конце `map_file`).""" + async with maker() as session, session.begin(): + new_file = await session.get(StoredFile, data["file_id"]) + new_document = await session.get(Document, data["document_id"]) + assert new_file is not None + assert new_document is not None + replaced = await supersede_previous( + session, new_file=new_file, new_documents=[new_document], uploader_id=None, now=utc_now() + ) + return [file.id for file in replaced] + + +async def _replace_event(maker: async_sessionmaker[AsyncSession]) -> AuditEvent: + async with maker() as session: + stmt = select(AuditEvent).where(AuditEvent.action == AuditAction.FILE_REPLACE) + return (await session.execute(stmt)).scalar_one() + + +@pytest.mark.integration +async def test_supersede_waits_for_pick_up_and_unbinds_replaced_document( + committed_sessionmaker: async_sessionmaker[AsyncSession], +) -> None: + """R-2, автозамена ∥ подхват: правка периметра держит версию, добавила ЮЛ и подхватила его `CONFIRMED` документ + (ещё не зафиксировано). Автозамена файла этого документа блокирует надмножество версий ГК, ждёт правку и после её + commit снимает уже видимую привязку; `unbound_version_ids` — эта версия. Блокируй автозамена только версии, где + документ уже привязан, — она не увидела бы незафиксированную привязку и оставила бы её висеть.""" + maker = committed_sessionmaker + data = await _committed_setup(maker) + old_document_id = await _committed_replacement(maker, data) + + async with maker() as editor, editor.begin(): + version = await repo.lock_version(editor, data["version_id"]) + assert version is not None + consolidation = await editor.get(Consolidation, version.consolidation_id) + assert consolidation is not None + editor.add(ConsolidationMember(consolidation_version_id=version.id, legal_entity_id=data["owner_id"])) + await editor.flush() + assert ( + await consolidation_documents.bind_confirmed( + editor, version=version, consolidation=consolidation, entity_ids=[data["owner_id"]] + ) + == 1 + ) + replacement = asyncio.create_task(_supersede(maker, data)) + await _wait_for_lock_waiter(maker) + assert not replacement.done() + replaced = await asyncio.wait_for(replacement, LOCK_WAIT_SECONDS) + + assert len(replaced) == 1 + assert old_document_id not in await _committed_bound(maker, data["version_id"]) + event = await _replace_event(maker) + assert event.after_value is not None + assert event.after_value["unbound_version_ids"] == [str(data["version_id"])] + + +@pytest.mark.integration +async def test_supersede_waits_for_create_and_unbinds_replaced_document( + committed_sessionmaker: async_sessionmaker[AsyncSession], +) -> None: + """R-2, автозамена ∥ создание версии: создание держит advisory ГК и подхватило `CONFIRMED` документ в новую + (невидимую до commit) версию; автозамена ждёт advisory ГК и после commit снимает привязку новой версии.""" + maker = committed_sessionmaker + data = await _committed_setup(maker, spare=0, with_version=False) + old_document_id = await _committed_replacement(maker, data) + + async with maker() as creator, creator.begin(): + await repo.lock_group_bindings(creator, data["group_id"]) # как `ConsolidationService.create` + group = await creator.get(CompanyGroup, data["group_id"]) + assert group is not None + entities = [await creator.get(LegalEntity, entity_id) for entity_id in data["entity_ids"]] + version = await _version(creator, group, [entity for entity in entities if entity is not None]) + consolidation = await creator.get(Consolidation, version.consolidation_id) + assert consolidation is not None + bound = await consolidation_documents.bind_confirmed( + creator, version=version, consolidation=consolidation, entity_ids=data["entity_ids"] + ) + assert bound == 1 + version_id = version.id + replacement = asyncio.create_task(_supersede(maker, data)) + await _wait_for_lock_waiter(maker) + assert not replacement.done() + await asyncio.wait_for(replacement, LOCK_WAIT_SECONDS) + + assert old_document_id not in await _committed_bound(maker, version_id) + event = await _replace_event(maker) + assert event.after_value is not None + assert event.after_value["unbound_version_ids"] == [str(version_id)] diff --git a/openspec/changes/slice-4-consolidation/design.md b/openspec/changes/slice-4-consolidation/design.md index b9e6fd5..a1c0dd1 100644 --- a/openspec/changes/slice-4-consolidation/design.md +++ b/openspec/changes/slice-4-consolidation/design.md @@ -214,7 +214,7 @@ legal_entity_name, account_code, missing_quarters[]}}`. Фильтры `GET …/ | 27 | 2026-10-06 | Члены, остающиеся в периметре, не пересоздаются (сохраняются `id` и `is_head_entity`); новые — `is_head_entity = false`. Непустая смена периметра выставляет `consolidation_version.updated_by/at` | признак головной организации — решение аналитика, смена состава его не сбрасывает; `updated_*` показывают последнюю правку версии (как у `transition`) | нулевая | | 28 | 2026-10-06 | Общий пролог операций над версией вынесен в `services/consolidation_guard.py`: `lock_visible_version` (`FOR UPDATE` → видимость → 404) и `lock_editable_version` (+ активная блокировка и `ensure_version_editable`); `VERSION_NOT_FOUND` живёт там и реэкспортируется из `services.consolidations`. `take`/`release` берут только `lock_visible_version` (редактируемость им не нужна). Поведение сервисов блокировки и статусов не менялось | закрывает T4.2 (три копии пролога; четвёртая — Task 11); порядок «версия → видимость → редактируемость» одинаков везде | правка пролога меняет все операции сразу (задумано) | | 29 | 2026-10-06 | До Task 6 отвязка ушедших ЮЛ — `consolidation_documents.unbind_entity(session, *, version_id, entity_id) -> int`: прямой `DELETE` строк `consolidation_document` документов этого ЮЛ в этой версии; `bind_confirmed` для добавленных — заглушка (вызывается, возвращает 0), тест подхвата проверяет вызов шпионом | план Task 5: до Task 6 отвязка прямым удалением; имена и сигнатуры зафиксированы за Task 6 | до Task 6 добавленное ЮЛ не получает привязок документов (ячейка `CONFIRMED` появится в Task 6/8) | -| 30 | 2026-10-06 | R-2, гонка «создание версии ∥ подтверждение/автозамена» закрыта `pg_advisory_xact_lock(hashtextextended('consolidation-group:<ГК>', 0))` (`repositories.consolidations.lock_group_bindings`): `ConsolidationService.create` берёт его сразу после проверки видимости ГК (до чтения ГСК, проверки дубля периода и подхвата), `lock_versions_for_entity` — первым шагом, если у ЮЛ есть ГК (ЮЛ без ГК не может быть членом версии). Полный порядок: файл → документ → advisory ГК → версии; автозамена — advisory ЮЛ → старые файлы → advisory ГК → версии; `PUT …/members` advisory не берёт. Тесты на двух сессиях в обе стороны (`test_create_waits_for_confirm_…`, `test_confirm_waits_for_create_…`), мутанты без advisory в любой из сторон их роняют | новая версия невидима до commit, `FOR UPDATE` версий её не ловит; advisory по ГК — минимальная сериализация, не трогающая правку периметра | создание консолидаций одной ГК (любых периодов) сериализуется с подтверждениями и автозаменами документов ЮЛ этой ГК — ожидание на время короткой транзакции; параллельные создания одного периода теперь получают 409 `CONSOLIDATION_EXISTS` проверкой, а не UNIQUE | +| 30 | 2026-10-06 | R-2, гонка «создание версии ∥ подтверждение/автозамена» закрыта `pg_advisory_xact_lock(hashtextextended('consolidation-group:<ГК>', 0))` (`repositories.consolidations.lock_group_bindings`): `ConsolidationService.create` берёт его сразу после проверки видимости ГК (до чтения ГСК, проверки дубля периода и подхвата), `lock_versions_for_entity` — первым шагом, если у ЮЛ есть ГК (ЮЛ без ГК не может быть членом версии). Полный порядок: файл → документ → advisory ГК → версии; автозамена — advisory ЮЛ → старые файлы → advisory ГК → версии; `PUT …/members` advisory не берёт. Тесты на двух сессиях в обе стороны (`test_create_waits_for_confirm_…`, `test_confirm_waits_for_create_…`), мутанты без advisory в любой из сторон их роняют | новая версия невидима до commit, `FOR UPDATE` версий её не ловит; advisory по ГК — минимальная сериализация, не трогающая правку периметра | создание консолидаций одной ГК (любых периодов) сериализуется с подтверждениями и автозаменами документов ЮЛ этой ГК — ожидание на время короткой транзакции; параллельные создания одного периода теперь получают 409 `CONSOLIDATION_EXISTS` проверкой, а не UNIQUE. Все подтверждения (и автозамены) документов ЮЛ одной ГК теперь сериализуются на advisory-блокировке этой ГК, даже если у ГК нет ни одной версии «в работе» (lock берётся до поиска версий) | | 31 | 2026-10-06 | R-3: на `FOR NO KEY UPDATE` переведены только `lock_document` и `lock_file_documents`; `lock_file` (и `_lock_file` маппинга) остаётся `FOR UPDATE` | на `stored_file` нет FK из таблиц консолидации, в циклы привязки файл не входит; удаление файла (`DELETE stored_file`) под `FOR NO KEY UPDATE` повышало бы блокировку уже после захвата и могло бы встречно ждать `KEY SHARE` вставки задания | нулевая; если файл войдёт в цикл — та же правка `key_share=True` | | 32 | 2026-10-06 | R-3 в текущих потоках не воспроизводится через API: подхват привязывает только документы, `CONFIRMED` в зафиксированном снимке, а подтверждение держит документ, который ещё не `CONFIRMED` (иначе сразу 409). Тест `test_document_lock_does_not_block_binding_insert` проверяет сам цикл блокировок на примитивах (`lock_document` + ожидание версии ∥ `lock_version` + вставка привязки): с `FOR UPDATE` — `DeadlockDetectedError`, с `FOR NO KEY UPDATE` — проходит. API-тесты «подтверждение ∥ `PUT …/members`» в обе стороны проверяют R-2 (ожидание и непотерянную привязку) | диспатч требовал мутант-проверку R-3; на уровне API цикла нет, но страховка дешёвая и защищает будущие пути (Task 7/11) | нулевая | | 33 | 2026-10-06 | Автозамена снимает привязки ВСЕХ документов заменённого файла, а не только совпавших с новым по ключу автозамены | spec primary-files: «привязки документов заменённых файлов»; файл `SUPERSEDED` целиком, а подхват берёт документы лишь незаменённых файлов — оставшаяся привязка была бы «висячей» | документ заменённого файла, не повторённый в новом файле, выпадает из версии (пробел в чек-листе) — так и задумано спекой |