fix(aliases): удаление алиаса под advisory-блокировкой (T11.3, M-4)
aliases.remove берёт lock_alias_key по normalized_alias и перечитывает алиас под блокировкой: при гонке с move_alias ручного сопоставления отвечает 404 без фантомного события об удалении. Тесты: удаление ∥ перенос алиаса; ручное сопоставление ждёт блокировку алиаса (design №66). Refs #21 Co-Authored-By: Claude Sonnet 4.5 <noreply@anthropic.com> Claude-Session: cedd7915-2efe-41e5-988c-a9117033598b
This commit is contained in:
parent
830c02ab7c
commit
74be7d357a
@ -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()
|
||||||
|
|
||||||
|
|
||||||
|
|||||||
@ -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)
|
||||||
|
|||||||
@ -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
|
||||||
|
|||||||
Loading…
x
Reference in New Issue
Block a user