diff --git a/backend/app/data/source_changes.py b/backend/app/data/source_changes.py index cb1cb35e..fd42a545 100644 --- a/backend/app/data/source_changes.py +++ b/backend/app/data/source_changes.py @@ -29,7 +29,7 @@ from dataclasses import dataclass from datetime import datetime from typing import Any -from sqlalchemy import func, or_, select +from sqlalchemy import func, or_, select, update from sqlalchemy.exc import IntegrityError from sqlalchemy.ext.asyncio import AsyncSession @@ -150,11 +150,16 @@ async def record(session: AsyncSession, source: Any, change: NewChange, detected raise ValueError(f"不认识的严重程度:{change.severity}") fp = fingerprint(change.kind, change.table, change.column, change.key) now = utcnow() + # 已经有了的:先按条件 UPDATE 记下又见到一次(同时占住那一行:SQLite 上拿到写锁、PostgreSQL 上锁住这一行),再读出来改 + # 影响面——口径试算同时往 affects 里记没试算的那几版(ledger/recall._probe_skipped,同样先占住再读改写),谁也不盖掉谁 + seen = await session.execute(update(SourceChange).where( + SourceChange.source_id == source.id, SourceChange.fingerprint == fp) + .values(last_seen_at=now, seen_count=func.coalesce(SourceChange.seen_count, 0) + 1) + .execution_options(synchronize_session=False)) found = (await session.execute(select(SourceChange).where( - SourceChange.source_id == source.id, SourceChange.fingerprint == fp))).scalar_one_or_none() + SourceChange.source_id == source.id, SourceChange.fingerprint == fp) + .execution_options(populate_existing=True))).scalar_one_or_none() if seen.rowcount else None if found is not None: - found.last_seen_at = now - found.seen_count = int(found.seen_count or 0) + 1 if found.status == "open": # 影响按这一次刷新;表结构变了之后口径试算的情况(ledger/recall 记的 probes)不是探查算的,留着 probes = (found.affects or {}).get("probes") if isinstance(found.affects, dict) else None diff --git a/backend/app/ledger/recall.py b/backend/app/ledger/recall.py index b03425d8..491a05ad 100644 --- a/backend/app/ledger/recall.py +++ b/backend/app/ledger/recall.py @@ -66,7 +66,7 @@ from dataclasses import dataclass, field from datetime import datetime, timedelta, timezone from typing import Any -from sqlalchemy import func, select +from sqlalchemy import func, select, update from app.core.events import EventType from app.db.base import SessionLocal, utcnow @@ -1585,8 +1585,14 @@ async def _probe_skipped(change: Mapping[str, Any], caliber: Any, pinned: Any, g note = {"workflow_id": caliber.workflow_id, "version": caliber.version, "name": str(name), "outcome": "skipped", "missing": missing, "reason": reason, "inputs_from": inputs_from} async with SessionLocal() as session: - row = (await session.execute(select(SourceChange).where(SourceChange.id == change.get("id")))).scalar_one_or_none() + # 先占住那一行再读改写(SQLite 上拿到写锁、PostgreSQL 上锁住这一行):几版口径同时记、探查同时刷新影响 + # (source_changes.record)时一个接一个来,谁也不盖掉谁 + await session.execute(update(SourceChange).where(SourceChange.id == change.get("id")) + .values(seen_count=SourceChange.seen_count).execution_options(synchronize_session=False)) + row = (await session.execute(select(SourceChange).where(SourceChange.id == change.get("id")) + .execution_options(populate_existing=True))).scalar_one_or_none() if row is None: + await session.rollback() return affects = dict(row.affects or {}) kept = [p for p in affects.get(PROBES_KEY) or [] if isinstance(p, dict) diff --git a/backend/tests/test_ledger_recall.py b/backend/tests/test_ledger_recall.py index 751a7ed7..043587df 100644 --- a/backend/tests/test_ledger_recall.py +++ b/backend/tests/test_ledger_recall.py @@ -954,3 +954,40 @@ async def test_a_probe_that_cannot_assemble_its_inputs_says_which_are_missing(en await session.commit() again = (await session.execute(select(SourceChange).where(SourceChange.id == out["id"]))).scalar_one() assert again.affects["probes"] == [note] + + +async def test_probe_notes_and_change_refreshes_do_not_overwrite_each_other(engine_up): + """同一条变化上几处同时写 affects(几版口径各记一条没试算、探查又刷新了一次影响):先占住那一行再读改写,谁也不盖掉谁。""" + import asyncio + from types import SimpleNamespace + + from app.data import source_changes as sc + from app.db.models import DataSource + from app.db.models_sources import SourceChange + + async with SessionLocal() as session: + source = (await session.execute(select(DataSource).where(DataSource.name == S.CURRENT["source"]))).scalar_one() + change, _ = await sc.record(session, source, sc.NewChange(kind="column_removed", table="orders", column="amount", + severity="action", key="v4", + affects={"catalog": [], "templates": [], + "rules_touched": False}), + sc.Detected(job_kind="introspect")) + await session.commit() + out = sc.change_out(change) + source_id = source.id + + async def refresh() -> None: + async with SessionLocal() as session: + src = await session.get(DataSource, source_id) + await sc.record(session, src, sc.NewChange(kind="column_removed", table="orders", column="amount", + severity="action", key="v4", + affects={"catalog": [], "templates": [], "rules_touched": True}), + sc.Detected(job_kind="introspect")) + await session.commit() + + calibers = [SimpleNamespace(workflow_id=f"wf-{i}", version=1) for i in range(6)] + await asyncio.gather(*(recall._probe_skipped(out, c, None, {}, "defaults", "缺参数") for c in calibers), refresh()) + async with SessionLocal() as session: + row = (await session.execute(select(SourceChange).where(SourceChange.id == out["id"]))).scalar_one() + assert sorted(p["workflow_id"] for p in row.affects["probes"]) == [f"wf-{i}" for i in range(6)], row.affects + assert row.affects["rules_touched"] is True, "探查刷新的那一份也在"