Merge pull request 'fix(aliases,consolidations): блокировка удаления алиаса, границы входов (T11.3, M-4, M-6)' (#26) from fix/21-aliases-bounds into main

Reviewed-on: #26
This commit is contained in:
aep 2026-10-06 14:01:19 +03:00
commit 4879fdca88
11 changed files with 205 additions and 11 deletions

View File

@ -715,8 +715,8 @@ Q2–Q4 не подходят (R-14a). Привязываются только `
(`legal_entity_id`, `related_group_id`) и без значений ключей. Особенности: повторное добавление в ту же группу (`legal_entity_id`, `related_group_id`) и без значений ключей. Особенности: повторное добавление в ту же группу
отдаёт `MEMBER_OF_OTHER_GROUP` с нейтральным текстом «ЮЛ уже входит в группу этого вида» (его раньше срабатывает индекс отдаёт `MEMBER_OF_OTHER_GROUP` с нейтральным текстом «ЮЛ уже входит в группу этого вида» (его раньше срабатывает индекс
«ЮЛ + вид»); гонку за один алиас у разных ЮЛ (`ALIAS_TAKEN`) БД не ловит (уникален только `(ЮЛ, алиас)`) — её «ЮЛ + вид»); гонку за один алиас у разных ЮЛ (`ALIAS_TAKEN`) БД не ловит (уникален только `(ЮЛ, алиас)`) — её
сериализует advisory-блокировка по нормализованному алиасу (`lock_alias_key`: добавление алиаса и ручное сопоставление сериализует advisory-блокировка по нормализованному алиасу (`lock_alias_key`: добавление и удаление алиаса, ручное
контрагента); сопоставление контрагента; удаление перечитывает алиас под блокировкой и отвечает 404, если его уже убрал перенос);
email без учёта регистра проверяет только сервис, поэтому гонка с другим регистром остаётся открытым ограничением. email без учёта регистра проверяет только сервис, поэтому гонка с другим регистром остаётся открытым ограничением.
Любое иное нарушение целостности (FK, check, неизвестное unique) — прежняя строка `нарушение целостности данных в БД`. Любое иное нарушение целостности (FK, check, неизвестное unique) — прежняя строка `нарушение целостности данных в БД`.

View File

@ -381,11 +381,15 @@ async def lock_alias_key(session: AsyncSession, normalized: str) -> None:
await session.execute(select(func.pg_advisory_xact_lock(key))) await session.execute(select(func.pg_advisory_xact_lock(key)))
async def get_alias(session: AsyncSession, entity_id: uuid.UUID, alias_id: uuid.UUID) -> LegalEntityAlias | None: async def get_alias(
"""Алиас по id, только если он принадлежит ЮЛ `entity_id`.""" session: AsyncSession, entity_id: uuid.UUID, alias_id: uuid.UUID, *, refresh: bool = False
) -> LegalEntityAlias | None:
"""Алиас по id, только если он принадлежит ЮЛ `entity_id`; `refresh` — состояние строки заменяется свежим из БД."""
stmt = select(LegalEntityAlias).where( stmt = select(LegalEntityAlias).where(
LegalEntityAlias.id == alias_id, LegalEntityAlias.legal_entity_id == entity_id LegalEntityAlias.id == alias_id, LegalEntityAlias.legal_entity_id == entity_id
) )
if refresh:
stmt = stmt.execution_options(populate_existing=True)
return (await session.execute(stmt)).scalar_one_or_none() return (await session.execute(stmt)).scalar_one_or_none()

View File

@ -28,6 +28,9 @@ YEAR_MAX = 2100
DEFAULT_OSV_LIMIT = 500 DEFAULT_OSV_LIMIT = 500
MAX_OSV_LIMIT = 2000 MAX_OSV_LIMIT = 2000
MAX_TAKEOVER_REASON = 1000 MAX_TAKEOVER_REASON = 1000
MAX_MEMBERS = 500 # M-6: периметр версии (ЮЛ одной ГК), с запасом на порядок выше реальных групп
MAX_STATUS_COMMENT = 1000 # как причина перехвата
MAX_OSV_QUERY = 500 # как NAME_MAX_LENGTH справочников
CENT = Decimal("0.01") CENT = Decimal("0.01")
@ -137,14 +140,14 @@ class VersionCard(VersionBrief):
class MembersIn(BaseModel): class MembersIn(BaseModel):
"""Тело `PUT …/members`: новый периметр версии (дубли схлопываются сервисом).""" """Тело `PUT …/members`: новый периметр версии (дубли схлопываются сервисом)."""
legal_entity_ids: list[uuid.UUID] legal_entity_ids: list[uuid.UUID] = Field(max_length=MAX_MEMBERS)
class StatusIn(BaseModel): class StatusIn(BaseModel):
"""Тело `POST …/status`: ручной переход (`CANCELLED` — комментарий обязателен; возврат в `INITIATED`).""" """Тело `POST …/status`: ручной переход (`CANCELLED` — комментарий обязателен; возврат в `INITIATED`)."""
to_status: ConsolidationStatus to_status: ConsolidationStatus
comment: str | None = None comment: str | None = Field(default=None, max_length=MAX_STATUS_COMMENT)
class ChecklistRuleOut(BaseModel): class ChecklistRuleOut(BaseModel):
@ -240,7 +243,7 @@ class OsvFilters(BaseModel):
counterparty_type: CounterpartyType | None = None counterparty_type: CounterpartyType | None = None
unmapped: bool | None = None unmapped: bool | None = None
match_kind: list[CounterpartyMatchKind] | None = None match_kind: list[CounterpartyMatchKind] | None = None
q: str | None = None q: str | None = Field(default=None, max_length=MAX_OSV_QUERY)
class OsvPageParams(BaseModel): class OsvPageParams(BaseModel):

View File

@ -77,6 +77,12 @@ class AliasService:
"""Удаляет алиас этого ЮЛ (иначе 404); `before_value` аудита — удалённые значения.""" """Удаляет алиас этого ЮЛ (иначе 404); `before_value` аудита — удалённые значения."""
entity = await self._get_entity(entity_id) entity = await self._get_entity(entity_id)
alias = await repo.get_alias(self._session, entity.id, alias_id) alias = await repo.get_alias(self._session, entity.id, alias_id)
if alias is None:
raise NotFoundError(ALIAS_NOT_FOUND)
# T11.3: та же advisory-блокировка, что у `add` и ручного сопоставления (design №66); без неё удаление
# гонялось бы с `move_alias`, и обе стороны писали бы событие об удалении одного алиаса
await repo.lock_alias_key(self._session, alias.normalized_alias)
alias = await repo.get_alias(self._session, entity.id, alias_id, refresh=True) # перенос мог удалить его
if alias is None: if alias is None:
raise NotFoundError(ALIAS_NOT_FOUND) raise NotFoundError(ALIAS_NOT_FOUND)
before = alias_payload(alias) before = alias_payload(alias)

View File

@ -23,10 +23,16 @@ VERSION_NOT_FOUND = "версия консолидации не найдена"
async def lock_visible_version( async def lock_visible_version(
session: AsyncSession, scope: VisibilityScope, version_id: uuid.UUID session: AsyncSession, scope: VisibilityScope, version_id: uuid.UUID
) -> tuple[ConsolidationVersion, Consolidation]: ) -> tuple[ConsolidationVersion, Consolidation]:
"""Версия под `FOR UPDATE` и её консолидация, если ГК видна пользователю; иначе 404 `VERSION_NOT_FOUND`.""" """Версия под `FOR UPDATE` и её консолидация, если ГК видна пользователю; иначе 404 `VERSION_NOT_FOUND`.
Видимость проверяется ДО блокировки (M-6): невидимая версия отвечает 404 сразу, не ожидая чужую транзакцию, которая
держит её строку. Затем `FOR UPDATE` с обновлением состояния; исчезнувшая за это время версия — тоже 404.
Повторной проверки видимости нет: ГК и филиал версия не меняет."""
found = await repo.get_version(session, scope, version_id)
if found is None:
raise NotFoundError(VERSION_NOT_FOUND)
version = await repo.lock_version(session, version_id) version = await repo.lock_version(session, version_id)
found = None if version is None else await repo.get_version(session, scope, version_id) if version is None:
if version is None or found is None:
raise NotFoundError(VERSION_NOT_FOUND) raise NotFoundError(VERSION_NOT_FOUND)
return version, found[1] return version, found[1]

View File

@ -2,7 +2,7 @@
audit-trail `OSV_COUNTERPARTY_MAP`, `LEGAL_ENTITY_CHANGE`; design D-10; §3 «Ручное сопоставление»). audit-trail `OSV_COUNTERPARTY_MAP`, `LEGAL_ENTITY_CHANGE`; design D-10; §3 «Ручное сопоставление»).
Одна транзакция, порядок проверок и блокировок: Одна транзакция, порядок проверок и блокировок:
1. `lock_editable_version` — версия `FOR UPDATE` → видимость (404) → статус «в работе» и своя блокировка (409). 1. `lock_editable_version` — видимость (404) → версия `FOR UPDATE` → статус «в работе» и своя блокировка (409).
2. Строка — строка свода версии (`entry_in_version`, иначе 404 «строка не найдена»); ЮЛ существует (без policy: 2. Строка — строка свода версии (`entry_in_version`, иначе 404 «строка не найдена»); ЮЛ существует (без policy:
контрагент может быть из другого филиала; иначе 404 `LEGAL_ENTITY_NOT_FOUND`). контрагент может быть из другого филиала; иначе 404 `LEGAL_ENTITY_NOT_FOUND`).
3. `normalized = normalize_alias(имя строки)`; непустой — advisory-блокировка алиаса (`lock_alias_key`, общая со 3. `normalized = normalize_alias(имя строки)`; непустой — advisory-блокировка алиаса (`lock_alias_key`, общая со

View File

@ -468,3 +468,20 @@ async def test_takeover_during_cancellation_sees_cancelled_version(
assert response.json()["detail"]["code"] == "VERSION_NOT_EDITABLE" assert response.json()["detail"]["code"] == "VERSION_NOT_EDITABLE"
async with committed_sessionmaker() as session: async with committed_sessionmaker() as session:
assert all(lock.released_at is not None for lock in await _locks(session, version_id)) assert all(lock.released_at is not None for lock in await _locks(session, version_id))
@pytest.mark.integration
async def test_foreign_version_is_hidden_without_waiting_for_row_lock(
committed_sessionmaker: async_sessionmaker[AsyncSession], committed_client: AuthClientFactory
) -> None:
"""M-6: видимость проверяется до `FOR UPDATE` — чужая версия отвечает 404 сразу, даже когда её строку держит
правка владельца (иначе невидимый пользователь ждал бы чужую транзакцию и мог влиять на её длительность)."""
async with committed_sessionmaker() as setup, setup.begin():
version_id = (await _version(setup, office_name=KRASNODAR)).id
client = await committed_client(KA_RF)
async with committed_sessionmaker() as holder, holder.begin():
assert await repo.lock_version(holder, version_id) is not None
response = await asyncio.wait_for(client.post(_lock_url(version_id)), LOCK_WAIT_SECONDS)
assert response.status_code == HTTP_NOT_FOUND

View File

@ -17,6 +17,8 @@ from sqlalchemy import func, select, text
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
from src.api.dependencies.audit import RequestAudit from src.api.dependencies.audit import RequestAudit
from src.common.clock import utc_now
from src.domain.aliases import normalize_alias
from src.domain.policy import VisibilityScope from src.domain.policy import VisibilityScope
from src.models.access import AppUser from src.models.access import AppUser
from src.models.audit_ex import AuditEvent from src.models.audit_ex import AuditEvent
@ -33,6 +35,7 @@ from src.models.enums import (
) )
from src.models.files import Document from src.models.files import Document
from src.models.osv import OsvEntry from src.models.osv import OsvEntry
from src.repositories import directories as directories_repo
from src.schemas.consolidations import MapCounterpartyIn from src.schemas.consolidations import MapCounterpartyIn
from src.services.audit import AuditService from src.services.audit import AuditService
from src.services.counterparty_mapping import CounterpartyMappingService from src.services.counterparty_mapping import CounterpartyMappingService
@ -65,6 +68,8 @@ NAME = "Пекарня Счастливочная ООО"
ALIEN_INN = "7700000000" ALIEN_INN = "7700000000"
MAX_CANDIDATES = 100 MAX_CANDIDATES = 100
LOCK_WAIT_SECONDS = 10.0 LOCK_WAIT_SECONDS = 10.0
ADMIN = "admin@armka.local"
ENTITIES = "/directories/legal-entities"
@dataclass @dataclass
@ -820,3 +825,73 @@ async def test_concurrent_mappings_in_two_versions_serialize_on_shared_rows(
events = await _events(session, AuditAction.OSV_COUNTERPARTY_MAP) events = await _events(session, AuditAction.OSV_COUNTERPARTY_MAP)
assert [event.consolidation_version_id for event in events] == [q2_id, q3_id] assert [event.consolidation_version_id for event in events] == [q2_id, q3_id]
assert events[1].before_value == {"counterparty_id": str(chkf_id), "match_kind": "PROPAGATED"} assert events[1].before_value == {"counterparty_id": str(chkf_id), "match_kind": "PROPAGATED"}
@pytest.mark.integration
async def test_manual_mapping_waits_for_alias_advisory_lock(
committed_sessionmaker: async_sessionmaker[AsyncSession], committed_client: AuthClientFactory
) -> None:
"""M-4 (design №66): ручное сопоставление берёт `lock_alias_key` по нормализованному имени строки — пока другая
транзакция держит блокировку этого алиаса (справочник, перенос), запрос ждёт, а после её commit завершается."""
async with committed_sessionmaker() as setup, setup.begin():
world = await _world(setup)
entry_id = world.entries[("60", 1)]
client = await committed_client(KA_RF)
async with committed_sessionmaker() as holder, holder.begin():
await directories_repo.lock_alias_key(holder, normalize_alias(NAME))
racing = asyncio.create_task(
client.post(_map_url(world, entry_id), json={"legal_entity_id": str(world.schastl.id)})
)
await _wait_for_lock_waiter(committed_sessionmaker)
assert not racing.done()
response = await asyncio.wait_for(racing, LOCK_WAIT_SECONDS)
assert response.status_code == HTTP_OK, response.text
assert response.json()["alias_created"] is True
@pytest.mark.integration
async def test_alias_delete_racing_with_move_alias_is_404_without_phantom_event(
committed_sessionmaker: async_sessionmaker[AsyncSession], committed_client: AuthClientFactory
) -> None:
"""T11.3: `DELETE …/aliases/{id}` берёт ту же advisory-блокировку, что ручное сопоставление: пока перенос
(`move_alias`) не зафиксирован, удаление ждёт, после commit находит алиас уже удалённым — 404, а фантомного
события `LEGAL_ENTITY_CHANGE` об удалении (вторая запись про тот же алиас) нет."""
async with committed_sessionmaker() as setup, setup.begin():
world = await _world(setup)
owner = await _user(setup, KA_RF)
alias = LegalEntityAlias(
legal_entity_id=world.schastl.id,
alias=NAME,
normalized_alias=normalize_alias(NAME),
created_by=owner.id,
created_at=utc_now(),
)
setup.add(alias)
await setup.flush()
alias_id, owner_id = alias.id, owner.id
entry_id = world.entries[("60", 1)]
admin = await committed_client(ADMIN)
async with committed_sessionmaker() as holder, holder.begin():
audit = RequestAudit(service=AuditService(holder), user_id=owner_id, ip_address=None, user_agent=None)
service = CounterpartyMappingService(holder, audit, VisibilityScope(office_id=None))
moved = await service.map(
world.version.id, entry_id, MapCounterpartyIn(legal_entity_id=world.rb.id, move_alias=True)
)
assert moved.alias_moved_from == world.schastl.id
racing = asyncio.create_task(admin.delete(f"{ENTITIES}/{world.schastl.id}/aliases/{alias_id}"))
await _wait_for_lock_waiter(committed_sessionmaker)
assert not racing.done()
response = await asyncio.wait_for(racing, LOCK_WAIT_SECONDS)
assert response.status_code == HTTP_NOT_FOUND, response.text
async with committed_sessionmaker() as session:
removals = [
event
for event in await _events(session, AuditAction.LEGAL_ENTITY_CHANGE)
if event.object_id == alias_id and event.before_value is not None
]
assert len(removals) == 1 # только удаление при переносе
assert await session.get(LegalEntityAlias, alias_id) is None

View File

@ -0,0 +1,47 @@
"""Верхние границы входов консолидаций (spec consolidations-api «Границы входных данных», M-6 из #21): превышение —
422 со списком ошибок валидации (формат приложения: `detail[].loc/msg`) до обращения к БД."""
import uuid
import pytest
from sqlalchemy.ext.asyncio import AsyncSession
from src.schemas.consolidations import MAX_MEMBERS, MAX_OSV_QUERY, MAX_STATUS_COMMENT
from tests.conftest import AuthClientFactory
HTTP_UNPROCESSABLE = 422
KA_RF = "ka_rf@armka.local"
VERSIONS = "/consolidations/versions"
@pytest.mark.integration
async def test_members_over_limit_is_422(auth_client: AuthClientFactory, seeded: AsyncSession) -> None: # noqa: ARG001
client = await auth_client(KA_RF)
ids = [str(uuid.uuid4()) for _ in range(MAX_MEMBERS + 1)]
response = await client.put(f"{VERSIONS}/{uuid.uuid4()}/members", json={"legal_entity_ids": ids})
assert response.status_code == HTTP_UNPROCESSABLE
assert response.json()["detail"][0]["loc"] == ["body", "legal_entity_ids"]
@pytest.mark.integration
async def test_status_comment_over_limit_is_422(auth_client: AuthClientFactory, seeded: AsyncSession) -> None: # noqa: ARG001
client = await auth_client(KA_RF)
response = await client.post(
f"{VERSIONS}/{uuid.uuid4()}/status", json={"to_status": "CANCELLED", "comment": "а" * (MAX_STATUS_COMMENT + 1)}
)
assert response.status_code == HTTP_UNPROCESSABLE
assert response.json()["detail"][0]["loc"] == ["body", "comment"]
@pytest.mark.integration
async def test_osv_query_over_limit_is_422(auth_client: AuthClientFactory, seeded: AsyncSession) -> None: # noqa: ARG001
client = await auth_client(KA_RF)
response = await client.get(f"{VERSIONS}/{uuid.uuid4()}/osv", params={"q": "а" * (MAX_OSV_QUERY + 1)})
assert response.status_code == HTTP_UNPROCESSABLE
assert response.json()["detail"][0]["loc"] == ["query", "q"]

View File

@ -13,10 +13,14 @@ from src.domain.consolidated_osv import CounterpartyType as DomainCounterpartyTy
from src.models.enums import CounterpartyMatchKind, FileStatus, PeriodQuarter from src.models.enums import CounterpartyMatchKind, FileStatus, PeriodQuarter
from src.schemas import consolidations as schemas from src.schemas import consolidations as schemas
from src.schemas.consolidations import ( from src.schemas.consolidations import (
MAX_MEMBERS,
MAX_OSV_QUERY,
MAX_STATUS_COMMENT,
ChecklistCellOut, ChecklistCellOut,
ConsolidatedRowOut, ConsolidatedRowOut,
ConsolidationCreate, ConsolidationCreate,
MapCounterpartyIn, MapCounterpartyIn,
MembersIn,
OsvFilters, OsvFilters,
OsvPageParams, OsvPageParams,
StatusIn, StatusIn,
@ -121,3 +125,20 @@ def test_counterparty_type_is_the_domain_enum() -> None:
assert OsvFilters(counterparty_type="GK").counterparty_type is DomainCounterpartyType.GK assert OsvFilters(counterparty_type="GK").counterparty_type is DomainCounterpartyType.GK
with pytest.raises(ValidationError): with pytest.raises(ValidationError):
OsvFilters(counterparty_type="OTHER") # type: ignore[arg-type] OsvFilters(counterparty_type="OTHER") # type: ignore[arg-type]
def test_members_in_upper_bound() -> None:
"""M-6: периметр версии — не больше `MAX_MEMBERS` идентификаторов (до разбора дублей сервисом)."""
ids = [uuid.uuid4() for _ in range(MAX_MEMBERS)]
assert len(MembersIn(legal_entity_ids=ids).legal_entity_ids) == MAX_MEMBERS
with pytest.raises(ValidationError):
MembersIn(legal_entity_ids=[*ids, uuid.uuid4()])
def test_status_comment_and_osv_query_upper_bounds() -> None:
assert StatusIn(to_status="CANCELLED", comment="а" * MAX_STATUS_COMMENT).comment
with pytest.raises(ValidationError):
StatusIn(to_status="CANCELLED", comment="а" * (MAX_STATUS_COMMENT + 1))
assert OsvFilters(q="а" * MAX_OSV_QUERY).q
with pytest.raises(ValidationError):
OsvFilters(q="а" * (MAX_OSV_QUERY + 1))

View File

@ -149,3 +149,18 @@ SHALL закрываться только документом с `osv_account_c
#### Scenario: Файл в обработке #### Scenario: Файл в обработке
- **WHEN** у ЮЛ загружен файл, задание ещё не завершено, других документов нет - **WHEN** у ЮЛ загружен файл, задание ещё не завершено, других документов нет
- **THEN** все ячейки ЮЛ — `PROCESSING` - **THEN** все ячейки ЮЛ — `PROCESSING`
### Requirement: Границы входных данных
Входы консолидаций SHALL иметь верхние границы: `legal_entity_ids` тела `PUT …/members` — не больше 500 идентификаторов,
`comment` тела `POST …/status` — не больше 1000 символов, `q` фильтров `GET …/osv` и экспорта — не больше 500 символов.
Превышение SHALL давать 422 со списком ошибок валидации (`detail[].loc` указывает поле) до обращения к БД. Версия
консолидации SHALL проверяться на видимость до блокировки `FOR UPDATE`: невидимая версия отвечает 404 сразу, не ожидая
транзакцию, которая держит её строку.
#### Scenario: Периметр длиннее предела
- **WHEN** в `PUT …/members` передано 501 ЮЛ
- **THEN** 422, `detail[0].loc = ["body", "legal_entity_ids"]`, периметр не изменился
#### Scenario: Невидимая версия при занятой строке
- **WHEN** версию чужого филиала держит `FOR UPDATE` другая транзакция, пользователь без видимости вызывает `POST …/lock`
- **THEN** 404 `VERSION_NOT_FOUND` без ожидания