Merge pull request 'AURORA-1831: гонка technical_number при параллельном POST /projects' (#229) from feat/207-aurora-1831-project-tn-race into test
This commit is contained in:
commit
37bb563f5d
15
api/alembic/versions/0050_project_technical_number_lock.py
Normal file
15
api/alembic/versions/0050_project_technical_number_lock.py
Normal file
@ -0,0 +1,15 @@
|
|||||||
|
from sql_migration import run_sql_file_migration
|
||||||
|
|
||||||
|
|
||||||
|
revision = "0050"
|
||||||
|
down_revision = "0049"
|
||||||
|
branch_labels = None
|
||||||
|
depends_on = None
|
||||||
|
|
||||||
|
|
||||||
|
def upgrade() -> None:
|
||||||
|
run_sql_file_migration("0050_project_technical_number_lock.sql")
|
||||||
|
|
||||||
|
|
||||||
|
def downgrade() -> None:
|
||||||
|
pass
|
||||||
@ -0,0 +1,96 @@
|
|||||||
|
-- AURORA-1831: гонка technical_number при параллельном создании проектов.
|
||||||
|
-- v3.add_project: копия версии из 0018 + advisory lock перед вычислением счётчика
|
||||||
|
-- и понятная ошибка при переполнении счётчика (больше 99 проектов на оргединицу).
|
||||||
|
|
||||||
|
CREATE OR REPLACE FUNCTION v3.add_project(p_name character varying, p_year integer, p_org_unit_id integer, p_level character varying DEFAULT 'project'::character varying, p_parent_id integer DEFAULT NULL::integer, p_project_type character varying DEFAULT NULL::character varying, p_vsp_format character varying DEFAULT NULL::character varying, p_placement_type character varying DEFAULT NULL::character varying, p_object_address character varying DEFAULT NULL::character varying, p_staff_count integer DEFAULT NULL::integer, p_total_area numeric DEFAULT NULL::numeric, p_system_code character varying DEFAULT NULL::character varying)
|
||||||
|
RETURNS TABLE(project_id integer, limit_report_id integer, current_expenses_report_id integer)
|
||||||
|
LANGUAGE plpgsql
|
||||||
|
AS $function$
|
||||||
|
DECLARE
|
||||||
|
v_pid INT;
|
||||||
|
v_lim INT;
|
||||||
|
v_cur INT;
|
||||||
|
v_seq INT;
|
||||||
|
v_technical_number VARCHAR;
|
||||||
|
BEGIN
|
||||||
|
IF p_name IS NULL OR length(trim(p_name)) = 0 THEN
|
||||||
|
RAISE EXCEPTION 'p_name обязателен';
|
||||||
|
END IF;
|
||||||
|
IF p_year IS NULL THEN
|
||||||
|
RAISE EXCEPTION 'p_year обязателен';
|
||||||
|
END IF;
|
||||||
|
IF p_org_unit_id IS NULL THEN
|
||||||
|
RAISE EXCEPTION 'p_org_unit_id обязателен';
|
||||||
|
END IF;
|
||||||
|
IF NOT EXISTS (SELECT 1 FROM v3.org_unit ou WHERE id = p_org_unit_id AND ou.is_active) THEN
|
||||||
|
RAISE EXCEPTION 'org_unit id=% не существует', p_org_unit_id;
|
||||||
|
END IF;
|
||||||
|
IF p_parent_id IS NOT NULL AND
|
||||||
|
NOT EXISTS (SELECT 1 FROM v3.project WHERE id = p_parent_id AND level = 'program') THEN
|
||||||
|
RAISE EXCEPTION 'parent project id=% не существует или не имеет level=program', p_parent_id;
|
||||||
|
END IF;
|
||||||
|
|
||||||
|
-- AURORA-1831: сериализуем выдачу номера внутри одной оргединицы. Без блокировки
|
||||||
|
-- две параллельные транзакции читают один и тот же MAX и падают на уникальном индексе.
|
||||||
|
-- Блокировка снимается на COMMIT/ROLLBACK. Решение опирается на READ COMMITTED
|
||||||
|
-- (приложение не меняет изоляцию): после получения блокировки SELECT MAX берёт новый
|
||||||
|
-- снимок и видит строку, закоммиченную предыдущим владельцем. При REPEATABLE READ
|
||||||
|
-- и выше этого недостаточно.
|
||||||
|
PERFORM pg_advisory_xact_lock(hashtext('v3.project.technical_number'), p_org_unit_id);
|
||||||
|
|
||||||
|
-- Порядковый счётчик проекта для данного РФ/ВСП (org_unit)
|
||||||
|
SELECT COALESCE(MAX(right(technical_number, 2)::int), 0) + 1
|
||||||
|
INTO v_seq
|
||||||
|
FROM v3.project
|
||||||
|
WHERE org_unit_id = p_org_unit_id
|
||||||
|
AND technical_number IS NOT NULL;
|
||||||
|
|
||||||
|
IF v_seq > 99 THEN
|
||||||
|
RAISE EXCEPTION 'Исчерпан счётчик технических номеров проектов для org_unit id=% (максимум 99)', p_org_unit_id;
|
||||||
|
END IF;
|
||||||
|
|
||||||
|
v_technical_number := to_char(now(), 'YYMM') || '_'
|
||||||
|
|| lpad(p_org_unit_id::text, 4, '0')
|
||||||
|
|| lpad(v_seq::text, 2, '0');
|
||||||
|
|
||||||
|
INSERT INTO v3.project (
|
||||||
|
name, level, parent_id, org_unit_id,
|
||||||
|
project_type, vsp_format, placement_type, object_address, staff_count, total_area,
|
||||||
|
system_code, technical_number
|
||||||
|
) VALUES (
|
||||||
|
p_name, p_level, p_parent_id, p_org_unit_id,
|
||||||
|
p_project_type, p_vsp_format, p_placement_type, p_object_address, p_staff_count, p_total_area,
|
||||||
|
p_system_code, v_technical_number
|
||||||
|
) RETURNING id INTO v_pid;
|
||||||
|
|
||||||
|
-- Первый год: 2 отчёта (LIMIT + CURRENT_EXPENSES) + snapshot фаз + сетка статей.
|
||||||
|
SELECT y.limit_report_id, y.current_expenses_report_id
|
||||||
|
INTO v_lim, v_cur
|
||||||
|
FROM v3.add_project_year(v_pid, p_year, false) y;
|
||||||
|
|
||||||
|
-- Аудит: PROJECT_CREATE — все заполненные поля + id отчётов
|
||||||
|
PERFORM v3.log_event(
|
||||||
|
'PROJECT_CREATE', 'PROJECT',
|
||||||
|
jsonb_strip_nulls(jsonb_build_object(
|
||||||
|
'project_id', v_pid,
|
||||||
|
'name', p_name,
|
||||||
|
'year', p_year,
|
||||||
|
'level', p_level,
|
||||||
|
'parent_id', p_parent_id,
|
||||||
|
'org_unit_id', p_org_unit_id,
|
||||||
|
'project_type', p_project_type,
|
||||||
|
'vsp_format', p_vsp_format,
|
||||||
|
'placement_type', p_placement_type,
|
||||||
|
'object_address', p_object_address,
|
||||||
|
'system_code', p_system_code,
|
||||||
|
'technical_number', v_technical_number,
|
||||||
|
'staff_count', p_staff_count,
|
||||||
|
'total_area', p_total_area,
|
||||||
|
'limit_report_id', v_lim,
|
||||||
|
'current_expenses_report_id', v_cur
|
||||||
|
))
|
||||||
|
);
|
||||||
|
|
||||||
|
RETURN QUERY SELECT v_pid, v_lim, v_cur;
|
||||||
|
END;
|
||||||
|
$function$;
|
||||||
@ -0,0 +1,131 @@
|
|||||||
|
"""AURORA-1831: гонка technical_number в v3.add_project и переполнение счётчика.
|
||||||
|
|
||||||
|
Тест гонки коммитит в общую тестовую БД (два независимых соединения, фикстура
|
||||||
|
client с общей сессией не подходит), поэтому создаваемые проекты удаляются в finally.
|
||||||
|
Тест переполнения ничего не коммитит: всё откатывается.
|
||||||
|
"""
|
||||||
|
import asyncio
|
||||||
|
import re
|
||||||
|
|
||||||
|
import pytest
|
||||||
|
from sqlalchemy import text
|
||||||
|
from sqlalchemy.exc import DBAPIError
|
||||||
|
from sqlalchemy.ext.asyncio import create_async_engine
|
||||||
|
|
||||||
|
from src.core.config import settings
|
||||||
|
|
||||||
|
ORG_UNIT_ID = 2 # в фикстуре у этой оргединицы нет проектов
|
||||||
|
ADMIN_ID = 1
|
||||||
|
|
||||||
|
CALL_ADD_PROJECT = text(
|
||||||
|
"SELECT project_id FROM v3.add_project(:name, 2026, :org, 'project', NULL, "
|
||||||
|
"'Открытие ВСП', 'Типовой', 'own', 'Тестовая улица 1')"
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
async def _begin_as_admin(conn):
|
||||||
|
await conn.execute(
|
||||||
|
text("SELECT set_config('app.user_id', :uid, true)"), {"uid": str(ADMIN_ID)}
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
async def _wait_for_lock_waiter(monitor_conn, timeout: float = 10.0):
|
||||||
|
"""Ждёт, пока какой-либо бэкенд встанет в ожидание блокировки (pg_stat_activity)."""
|
||||||
|
deadline = asyncio.get_running_loop().time() + timeout
|
||||||
|
while asyncio.get_running_loop().time() < deadline:
|
||||||
|
row = await monitor_conn.execute(
|
||||||
|
text(
|
||||||
|
"SELECT count(*) FROM pg_stat_activity "
|
||||||
|
"WHERE datname = current_database() AND wait_event_type = 'Lock' "
|
||||||
|
"AND pid <> pg_backend_pid()"
|
||||||
|
)
|
||||||
|
)
|
||||||
|
if row.scalar_one() > 0:
|
||||||
|
return
|
||||||
|
await asyncio.sleep(0.02)
|
||||||
|
raise AssertionError("вторая транзакция не встала в ожидание блокировки")
|
||||||
|
|
||||||
|
|
||||||
|
async def _technical_number(conn, project_id: int) -> str:
|
||||||
|
row = await conn.execute(
|
||||||
|
text("SELECT technical_number FROM v3.project WHERE id = :id"), {"id": project_id}
|
||||||
|
)
|
||||||
|
return row.scalar_one()
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.asyncio
|
||||||
|
async def test_parallel_add_project_gets_distinct_sequential_numbers():
|
||||||
|
engine = create_async_engine(settings.DATABASE_URL, echo=False)
|
||||||
|
created: list[int] = []
|
||||||
|
conn_a = await engine.connect()
|
||||||
|
conn_b = await engine.connect()
|
||||||
|
monitor = await engine.connect()
|
||||||
|
try:
|
||||||
|
trans_a = await conn_a.begin()
|
||||||
|
await _begin_as_admin(conn_a)
|
||||||
|
pid_a = (
|
||||||
|
await conn_a.execute(CALL_ADD_PROJECT, {"name": "PT_conc_a", "org": ORG_UNIT_ID})
|
||||||
|
).scalar_one()
|
||||||
|
created.append(pid_a)
|
||||||
|
|
||||||
|
async def run_b():
|
||||||
|
async with conn_b.begin():
|
||||||
|
await _begin_as_admin(conn_b)
|
||||||
|
pid = (
|
||||||
|
await conn_b.execute(
|
||||||
|
CALL_ADD_PROJECT, {"name": "PT_conc_b", "org": ORG_UNIT_ID}
|
||||||
|
)
|
||||||
|
).scalar_one()
|
||||||
|
created.append(pid)
|
||||||
|
return pid
|
||||||
|
|
||||||
|
task_b = asyncio.create_task(run_b())
|
||||||
|
# До правки B упирается в unique index, после правки - в advisory lock.
|
||||||
|
# В обоих случаях он встаёт в ожидание блокировки до COMMIT транзакции A.
|
||||||
|
await _wait_for_lock_waiter(monitor)
|
||||||
|
await trans_a.commit()
|
||||||
|
pid_b = await asyncio.wait_for(task_b, timeout=10)
|
||||||
|
|
||||||
|
async with engine.connect() as check:
|
||||||
|
tn_a = await _technical_number(check, pid_a)
|
||||||
|
tn_b = await _technical_number(check, pid_b)
|
||||||
|
assert tn_a != tn_b
|
||||||
|
assert re.fullmatch(r"\d{4}_0002\d\d", tn_a)
|
||||||
|
assert int(tn_b[-2:]) == int(tn_a[-2:]) + 1
|
||||||
|
finally:
|
||||||
|
await conn_a.close()
|
||||||
|
await conn_b.close()
|
||||||
|
await monitor.close()
|
||||||
|
async with engine.begin() as cleanup:
|
||||||
|
await _begin_as_admin(cleanup)
|
||||||
|
for pid in created:
|
||||||
|
await cleanup.execute(text("SELECT v3.del_project(:id)"), {"id": pid})
|
||||||
|
await engine.dispose()
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.asyncio
|
||||||
|
async def test_add_project_counter_overflow_raises_clear_error():
|
||||||
|
engine = create_async_engine(settings.DATABASE_URL, echo=False)
|
||||||
|
try:
|
||||||
|
async with engine.connect() as conn:
|
||||||
|
trans = await conn.begin()
|
||||||
|
try:
|
||||||
|
await _begin_as_admin(conn)
|
||||||
|
await conn.execute(
|
||||||
|
text(
|
||||||
|
"INSERT INTO v3.project (name, level, org_unit_id, technical_number, "
|
||||||
|
"project_type, vsp_format, placement_type, object_address) "
|
||||||
|
"VALUES ('PT_overflow', 'project', :org, '2601_000299', "
|
||||||
|
"'Открытие ВСП', 'Типовой', 'own', 'Тестовая улица 1')"
|
||||||
|
),
|
||||||
|
{"org": ORG_UNIT_ID},
|
||||||
|
)
|
||||||
|
with pytest.raises(DBAPIError) as exc_info:
|
||||||
|
await conn.execute(
|
||||||
|
CALL_ADD_PROJECT, {"name": "PT_overflow_next", "org": ORG_UNIT_ID}
|
||||||
|
)
|
||||||
|
assert "счётчик" in str(exc_info.value.orig)
|
||||||
|
finally:
|
||||||
|
await trans.rollback()
|
||||||
|
finally:
|
||||||
|
await engine.dispose()
|
||||||
Loading…
x
Reference in New Issue
Block a user