From 4ffd9ad4317029afd3c6f0e872980b3b751997c4 Mon Sep 17 00:00:00 2001 From: ThearchyHelios Date: Mon, 19 Oct 2026 11:48:43 +0800 Subject: [PATCH] =?UTF-8?q?=E5=8F=98=E5=8C=96=E8=AE=B0=E5=BD=95=E7=9A=84?= =?UTF-8?q?=20affects=20=E5=85=88=E5=8D=A0=E4=BD=8F=E9=82=A3=E4=B8=80?= =?UTF-8?q?=E8=A1=8C=E5=86=8D=E8=AF=BB=E6=94=B9=E5=86=99=EF=BC=88=E5=A4=8D?= =?UTF-8?q?=E6=A0=B8=20L5=EF=BC=89=EF=BC=9A=E5=8F=A3=E5=BE=84=E8=AF=95?= =?UTF-8?q?=E7=AE=97=E8=AE=B0=E6=B2=A1=E8=AF=95=E7=AE=97=E7=9A=84=E9=82=A3?= =?UTF-8?q?=E5=87=A0=E7=89=88=EF=BC=88ledger/recall.=5Fprobe=5Fskipped?= =?UTF-8?q?=EF=BC=89=E5=92=8C=E6=8E=A2=E6=9F=A5=E5=88=B7=E6=96=B0=E5=BD=B1?= =?UTF-8?q?=E5=93=8D=E9=9D=A2=EF=BC=88data/source=5Fchanges.record?= =?UTF-8?q?=EF=BC=89=E4=BB=A5=E5=89=8D=E9=83=BD=E6=98=AF=E5=85=88=E8=AF=BB?= =?UTF-8?q?=E5=86=8D=E6=95=B4=E4=BB=BD=E5=86=99=EF=BC=8C=E4=BA=A4=E9=94=99?= =?UTF-8?q?=E6=97=B6=E4=BA=92=E7=9B=B8=E7=9B=96=E6=8E=89=E3=80=82=E7=8E=B0?= =?UTF-8?q?=E5=9C=A8=E4=B8=A4=E5=A4=84=E9=83=BD=E5=85=88=E5=8F=91=E4=B8=80?= =?UTF-8?q?=E5=8F=A5=E6=8C=89=E6=9D=A1=E4=BB=B6=E7=9A=84=20UPDATE=EF=BC=88?= =?UTF-8?q?record=20=E9=A1=BA=E6=89=8B=E8=AE=B0=E5=8F=88=E8=A7=81=E5=88=B0?= =?UTF-8?q?=E4=B8=80=E6=AC=A1=EF=BC=8CSQLite=20=E4=B8=8A=E6=8B=BF=E5=88=B0?= =?UTF-8?q?=E5=86=99=E9=94=81=E3=80=81PostgreSQL=20=E4=B8=8A=E9=94=81?= =?UTF-8?q?=E4=BD=8F=E8=BF=99=E4=B8=80=E8=A1=8C=EF=BC=89=EF=BC=8C=E5=86=8D?= =?UTF-8?q?=E8=AF=BB=E5=87=BA=E6=9D=A5=E6=94=B9=E3=80=82=E8=A1=A5=E5=B9=B6?= =?UTF-8?q?=E5=8F=91=E6=B5=8B=E8=AF=95=EF=BC=88=E5=85=AD=E7=89=88=E5=8F=A3?= =?UTF-8?q?=E5=BE=84=E5=90=8C=E6=97=B6=E8=AE=B0=E3=80=81=E6=8E=A2=E6=9F=A5?= =?UTF-8?q?=E5=90=8C=E6=97=B6=E5=88=B7=E6=96=B0=EF=BC=8C=E4=B8=80=E6=9D=A1?= =?UTF-8?q?=E4=B8=8D=E4=B8=A2=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- backend/app/data/source_changes.py | 13 ++++++---- backend/app/ledger/recall.py | 10 ++++++-- backend/tests/test_ledger_recall.py | 37 +++++++++++++++++++++++++++++ 3 files changed, 54 insertions(+), 6 deletions(-) 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, "探查刷新的那一份也在"