fix(consolidation): ревью Task 6 — гонки автозамены с подхватом, выбор по confirmed_at, одна выборка ГК
- T6.0: тесты на двух сессиях «автозамена ∥ подхват при правке периметра» и «автозамена ∥ создание версии»: автозамена ждёт, снимает незафиксированную ранее привязку, unbound_version_ids — эта версия. - T6.2: тесты выбора самого позднего confirmed_at строятся так, чтобы «наибольший id» проигрывал; тест равенства confirmed_at (больший id, решение №35). - T6.3: реальный подхват через PUT …/members — версия в INITIATED, статус не меняется. - T6.4: lock_versions_for_entity не читает ГК ЮЛ второй раз (подзапрос заменён значением). - T6.5: design №30 — цена: подтверждения документов ЮЛ одной ГК сериализуются на advisory даже без версий в работе. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_012vJkmBGKKqMoEMzvsXG3y2
This commit is contained in:
Andrey Epifancev
parent
36bc966bf6
commit
0e1fbd8036
@ -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)
|
||||
)
|
||||
|
||||
@ -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
|
||||
|
||||
@ -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)]
|
||||
|
||||
@ -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` целиком, а подхват берёт документы лишь незаменённых файлов — оставшаяся привязка была бы «висячей» | документ заменённого файла, не повторённый в новом файле, выпадает из версии (пробел в чек-листе) — так и задумано спекой |
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user