mirror of
https://github.com/ThearchyHelios/AgentLab.git
synced 2026-11-15 00:21:58 +00:00
复核 R-b13(版本拦截与零碎):preflight 报错列出全部活着的、版本不同的进程,各标明心跳在走还是停着、等租约过期(以前只点名先看到心跳的那个,人停掉它再起还会被另一个拦下);源码指纹把依赖锁文件 requirements.lock 的内容也算进去,只升了依赖也算版本不同,锁文件取不到时换成几个关键包的版本,Docker 镜像里留一份锁文件;存量库 run_events 撞了号时把结果记一次(settings 表 cluster:run_events_seq_duplicates,记下撞号的那几处,最多 200 处),之后启动不再整表 GROUP BY、只按索引核对那几处还撞不撞,都处理掉了才再数一遍、建唯一索引、删掉记录。隔离后换身份那几条日志写出新旧完整身份已在 R-b9 那一笔里改了。README「升级」一段跟着补一句。补测试:一个心跳在走、一个心跳停着,报错两个都列出、各标状态;锁文件变了指纹跟着变、取不到时用包版本;撞号第一次整表数、第二次只核对不扫、处理掉之后建上索引并删掉记录
This commit is contained in:
@@ -76,6 +76,8 @@ SH
|
|||||||
WORKDIR /app
|
WORKDIR /app
|
||||||
COPY LICENSE COMMERCIAL-LICENSE.md /app/
|
COPY LICENSE COMMERCIAL-LICENSE.md /app/
|
||||||
COPY backend/app /app/backend/app
|
COPY backend/app /app/backend/app
|
||||||
|
# 锁文件留一份:进程登记的代码版本把它算进源码指纹(app/cluster/version.py:升了依赖的进程也算版本不同)
|
||||||
|
COPY backend/requirements.lock /app/backend/requirements.lock
|
||||||
COPY --from=web /src/frontend/dist /app/web
|
COPY --from=web /src/frontend/dist /app/web
|
||||||
|
|
||||||
ENV AGENTLAB_WEB_DIST=/app/web \
|
ENV AGENTLAB_WEB_DIST=/app/web \
|
||||||
|
|||||||
@@ -1194,8 +1194,8 @@ SQL 形状。各进程各自的并发闸门(口径比对、自动对比、起
|
|||||||
另一台读,可能读到不一致的页。所以多台机器时,记录表写得多的部署建议用一台机器多进程,或者把数据目录放在支持可靠文件锁的存储上。
|
另一台读,可能读到不一致的页。所以多台机器时,记录表写得多的部署建议用一台机器多进程,或者把数据目录放在支持可靠文件锁的存储上。
|
||||||
沙箱会话(microVM)在各进程自己手上,不跨进程共享。
|
沙箱会话(microVM)在各进程自己手上,不跨进程共享。
|
||||||
|
|
||||||
**升级。** 不支持新旧版本的进程同时连着一个库,代码里拦着:每个进程登记时带上代码版本(包版本、构建号或 git 提交、源码指纹)和
|
**升级。** 不支持新旧版本的进程同时连着一个库,代码里拦着:每个进程登记时带上代码版本(包版本、构建号或 git 提交、源码指纹——源码加上依赖锁文件,只升了依赖也算版本不同)和
|
||||||
库结构版本,库里还有活着的、版本不同的进程时,新版本的进程在补列之前就拒绝启动,报错里写明是哪几个进程、什么版本;同版本的照常
|
库结构版本,库里还有活着的、版本不同的进程时,新版本的进程在补列之前就拒绝启动,报错里列出全部这样的进程、各是什么版本、心跳还在不在走;同版本的照常
|
||||||
加入。升级步骤:
|
加入。升级步骤:
|
||||||
|
|
||||||
1. 停掉全部旧版本的进程(平稳关停的马上删掉自己的登记;强杀的等它的租约过期,新进程启动时会等);
|
1. 停掉全部旧版本的进程(平稳关停的马上删掉自己的登记;强杀的等它的租约过期,新进程启动时会等);
|
||||||
|
|||||||
@@ -4,7 +4,8 @@
|
|||||||
规矩(认领、租约、接手的说法),两边各按各的写同一批行。以前只在 README 里写「升级要先停掉全部进程」,现在代码里拦住。
|
规矩(认领、租约、接手的说法),两边各按各的写同一批行。以前只在 README 里写「升级要先停掉全部进程」,现在代码里拦住。
|
||||||
|
|
||||||
- 代码版本 code_version:包版本 + 构建号(AGENTLAB_BUILD,Docker 镜像构建时写进去)或 git 提交 + 源码指纹(app/ 下全部 .py
|
- 代码版本 code_version:包版本 + 构建号(AGENTLAB_BUILD,Docker 镜像构建时写进去)或 git 提交 + 源码指纹(app/ 下全部 .py
|
||||||
的内容哈希)。比的是源码指纹:同一份代码一定相同,改了一行就不同,取不到 git、没有构建号时也区分得开。
|
的内容,加上依赖锁文件 requirements.lock 的内容;Docker 镜像里也留着这个文件)。比的是源码指纹:同一份代码、同一套依赖一定相同,
|
||||||
|
改了一行、升了一个依赖就不同,取不到 git、没有构建号时也区分得开。锁文件取不到(不是从仓库或镜像装的)时换成几个关键包的版本。
|
||||||
- 库结构版本 schema_version:模型里各表各列(名字、类型、可空)加迁移清单的哈希。
|
- 库结构版本 schema_version:模型里各表各列(名字、类型、可空)加迁移清单的哈希。
|
||||||
|
|
||||||
怎么拦:PostgreSQL 时,补列之前先看一眼(preflight)——还有活着的、版本不同的进程就不启动;它的心跳停着(多半是强杀之后留下的、
|
怎么拦:PostgreSQL 时,补列之前先看一眼(preflight)——还有活着的、版本不同的进程就不启动;它的心跳停着(多半是强杀之后留下的、
|
||||||
@@ -33,6 +34,30 @@ class VersionMismatch(RuntimeError):
|
|||||||
"""库里还有活着的、版本不同的进程。"""
|
"""库里还有活着的、版本不同的进程。"""
|
||||||
|
|
||||||
|
|
||||||
|
#: 依赖锁文件(仓库里 backend/requirements.lock;Docker 镜像里 /app/backend/requirements.lock)
|
||||||
|
LOCK_FILE = _APP_DIR.parent / "requirements.lock"
|
||||||
|
#: 锁文件取不到时拿这几个包的版本代替(升级时最容易带出行为差别的那几个)
|
||||||
|
KEY_PACKAGES = ("fastapi", "pydantic", "sqlalchemy", "asyncpg", "psycopg", "langgraph", "langgraph-checkpoint",
|
||||||
|
"langgraph-checkpoint-postgres", "langchain-core", "starlette", "uvicorn")
|
||||||
|
|
||||||
|
|
||||||
|
def dependency_marker() -> bytes:
|
||||||
|
"""依赖这一份进指纹的样子:锁文件的内容;取不到就是几个关键包的版本。"""
|
||||||
|
try:
|
||||||
|
return b"lock:" + LOCK_FILE.read_bytes()
|
||||||
|
except OSError:
|
||||||
|
pass
|
||||||
|
from importlib.metadata import PackageNotFoundError, version
|
||||||
|
|
||||||
|
parts = []
|
||||||
|
for name in KEY_PACKAGES:
|
||||||
|
try:
|
||||||
|
parts.append(f"{name}=={version(name)}")
|
||||||
|
except PackageNotFoundError:
|
||||||
|
parts.append(f"{name}==")
|
||||||
|
return ("packages:" + ";".join(parts)).encode()
|
||||||
|
|
||||||
|
|
||||||
@lru_cache(maxsize=1)
|
@lru_cache(maxsize=1)
|
||||||
def source_fingerprint() -> str:
|
def source_fingerprint() -> str:
|
||||||
digest = hashlib.sha256()
|
digest = hashlib.sha256()
|
||||||
@@ -44,6 +69,8 @@ def source_fingerprint() -> str:
|
|||||||
digest.update(path.read_bytes())
|
digest.update(path.read_bytes())
|
||||||
except OSError:
|
except OSError:
|
||||||
continue
|
continue
|
||||||
|
# 依赖升级(锁文件变了)也算版本不同:代码一行没改、换了一版 SQLAlchemy / LangGraph 的进程同样不许混进来(复核 R-b13)
|
||||||
|
digest.update(b"\0deps\0" + dependency_marker())
|
||||||
return digest.hexdigest()[:12]
|
return digest.hexdigest()[:12]
|
||||||
|
|
||||||
|
|
||||||
@@ -115,10 +142,13 @@ async def _peers(conn: Any) -> list[dict[str, Any]] | None:
|
|||||||
return [dict(r) for r in rows]
|
return [dict(r) for r in rows]
|
||||||
|
|
||||||
|
|
||||||
def _describe(peer: dict[str, Any]) -> str:
|
def _describe(peer: dict[str, Any], beating: set[str] | None = None) -> str:
|
||||||
code = peer.get("code_version") or "(升级前的版本,没有记版本)"
|
code = peer.get("code_version") or "(升级前的版本,没有记版本)"
|
||||||
schema = peer.get("schema_version") or "未知"
|
schema = peer.get("schema_version") or "未知"
|
||||||
return f"{identity.describe(peer['id'])}(代码 {code},库结构 {schema})"
|
state = ""
|
||||||
|
if beating is not None:
|
||||||
|
state = ",心跳在走" if peer["id"] in beating else ",心跳停着、等租约过期"
|
||||||
|
return f"{identity.describe(peer['id'])}(代码 {code},库结构 {schema}{state})"
|
||||||
|
|
||||||
|
|
||||||
async def preflight(*, observe_s: float | None = None) -> None:
|
async def preflight(*, observe_s: float | None = None) -> None:
|
||||||
@@ -141,11 +171,12 @@ async def preflight(*, observe_s: float | None = None) -> None:
|
|||||||
return
|
return
|
||||||
for peer in mismatched:
|
for peer in mismatched:
|
||||||
first_beats.setdefault(peer["id"], peer.get("heartbeat_at"))
|
first_beats.setdefault(peer["id"], peer.get("heartbeat_at"))
|
||||||
# 心跳还在走:真活着,照实报错。心跳停着的(强杀、断电留下的登记)等它们的租约过期
|
# 心跳还在走:真活着,照实报错。心跳停着的(强杀、断电留下的登记)等它们的租约过期。报错时列出全部活着的、版本不同的
|
||||||
beating = [p for p in mismatched if p.get("heartbeat_at") != first_beats.get(p["id"])]
|
# 进程(复核 R-b13:只点名心跳在走的那个,人停掉它再起又会被另一个拦下)
|
||||||
|
beating = {p["id"] for p in mismatched if p.get("heartbeat_at") != first_beats.get(p["id"])}
|
||||||
waited = loop.time() - started
|
waited = loop.time() - started
|
||||||
if beating or waited >= lease_s() * 2 + observe:
|
if beating or waited >= lease_s() * 2 + observe:
|
||||||
raise VersionMismatch(_message(beating or mismatched))
|
raise VersionMismatch(_message(mismatched, beating=beating))
|
||||||
if waited >= observe and not warned:
|
if waited >= observe and not warned:
|
||||||
warned = True
|
warned = True
|
||||||
logger.warning("库里登记着 %d 个版本不同的进程,心跳停着(多半是强杀之后留下的):等它们的租约过期再启动",
|
logger.warning("库里登记着 %d 个版本不同的进程,心跳停着(多半是强杀之后留下的):等它们的租约过期再启动",
|
||||||
@@ -174,8 +205,8 @@ def _naive(moment: Any) -> Any:
|
|||||||
return moment.replace(tzinfo=None) if getattr(moment, "tzinfo", None) else moment
|
return moment.replace(tzinfo=None) if getattr(moment, "tzinfo", None) else moment
|
||||||
|
|
||||||
|
|
||||||
def _message(peers: list[dict[str, Any]]) -> str:
|
def _message(peers: list[dict[str, Any]], *, beating: set[str] | None = None) -> str:
|
||||||
listed = ";".join(_describe(p) for p in peers)
|
listed = ";".join(_describe(p, beating) for p in peers)
|
||||||
return (f"库里还有 {len(peers)} 个版本不同的 AgentLab 进程活着:{listed}。本进程是代码 {code_version()}、库结构 "
|
return (f"库里还有 {len(peers)} 个版本不同的 AgentLab 进程活着:{listed}。本进程是代码 {code_version()}、库结构 "
|
||||||
f"{schema_version()}。不支持新旧版本同时运行:先停掉这些进程再启动新版本(升级步骤见 README「多进程」的「升级」)")
|
f"{schema_version()}。不支持新旧版本同时运行:先停掉这些进程再启动新版本(升级步骤见 README「多进程」的「升级」)")
|
||||||
|
|
||||||
|
|||||||
+74
-5
@@ -600,10 +600,19 @@ def _migrate(conn) -> None: # pragma: no cover - 同步回调
|
|||||||
_backfill_not_null(conn, columns)
|
_backfill_not_null(conn, columns)
|
||||||
|
|
||||||
|
|
||||||
|
#: 存量库里 run_events 撞了号、这一次没建唯一索引时记下的结果(settings 表里;cluster: 开头的不是用户设置,api/settings 不列)
|
||||||
|
SEQ_DUPLICATES_KEY = "cluster:run_events_seq_duplicates"
|
||||||
|
#: 记下的撞号最多几处(之后启动只核对这几处还在不在)
|
||||||
|
SEQ_DUPLICATES_KEPT = 200
|
||||||
|
|
||||||
|
|
||||||
def _unique_run_event_seq(conn: Any, columns: dict[str, set[str]]) -> None: # pragma: no cover - 同步回调
|
def _unique_run_event_seq(conn: Any, columns: dict[str, set[str]]) -> None: # pragma: no cover - 同步回调
|
||||||
"""run_events 的 (run_id, seq) 补唯一索引(ux_run_events_run_seq,换掉原来不唯一的 ix_run_events_run_seq)。存量库里可能已经
|
"""run_events 的 (run_id, seq) 补唯一索引(ux_run_events_run_seq,换掉原来不唯一的 ix_run_events_run_seq)。存量库里可能已经
|
||||||
有撞了号的(多进程之前的进程内计数、按需裁判的老写法):先数一数,有就照实记日志、这一次不建(照旧用不唯一的索引),不让启动
|
有撞了号的(多进程之前的进程内计数、按需裁判的老写法):先数一数,有就照实记日志、这一次不建(照旧用不唯一的索引),不让启动
|
||||||
失败——那几次运行的封存核对本来就对不上,删哪一条都不对,要人看。"""
|
失败——那几次运行的封存核对本来就对不上,删哪一条都不对,要人看。
|
||||||
|
|
||||||
|
数一次就记下结果(settings 表的 SEQ_DUPLICATES_KEY:撞了号的那几处),之后启动不再整表 GROUP BY(复核 R-b13):只按索引核对
|
||||||
|
记下的那几处还撞不撞,还撞着就一句话带过;都清掉了才再整表数一遍、建索引、删掉这条记录。"""
|
||||||
from sqlalchemy import inspect, text
|
from sqlalchemy import inspect, text
|
||||||
|
|
||||||
if "run_events" not in columns:
|
if "run_events" not in columns:
|
||||||
@@ -611,16 +620,76 @@ def _unique_run_event_seq(conn: Any, columns: dict[str, set[str]]) -> None: # p
|
|||||||
names = {ix["name"] for ix in inspect(conn).get_indexes("run_events")}
|
names = {ix["name"] for ix in inspect(conn).get_indexes("run_events")}
|
||||||
if "ux_run_events_run_seq" in names:
|
if "ux_run_events_run_seq" in names:
|
||||||
return
|
return
|
||||||
dupes = conn.execute(text("SELECT run_id, seq, COUNT(*) FROM run_events GROUP BY run_id, seq HAVING COUNT(*) > 1 "
|
known = _known_seq_duplicates(conn, columns)
|
||||||
"LIMIT 20")).all()
|
if known:
|
||||||
|
still = _still_duplicated(conn, known["pairs"])
|
||||||
|
if still:
|
||||||
|
logger.info("run_events 里还有撞了号的事件(%s 记下的 %d 处里还剩 %d 处):(run_id, seq) 的唯一索引照旧不建,"
|
||||||
|
"处理掉之后下次启动会再核对一遍", known.get("found_at") or "之前", len(known["pairs"]), still)
|
||||||
|
return
|
||||||
|
dupes = conn.execute(text("SELECT run_id, seq FROM run_events GROUP BY run_id, seq HAVING COUNT(*) > 1 LIMIT :n"),
|
||||||
|
{"n": SEQ_DUPLICATES_KEPT + 1}).all()
|
||||||
if dupes:
|
if dupes:
|
||||||
runs = sorted({str(r[0]) for r in dupes})
|
runs = sorted({str(r[0]) for r in dupes})
|
||||||
logger.warning("run_events 里有撞了号的事件(%d 处以上,涉及运行 %s):这一次不建 (run_id, seq) 的唯一索引。"
|
logger.warning("run_events 里有撞了号的事件(%d 处%s,涉及运行 %s):这一次不建 (run_id, seq) 的唯一索引。"
|
||||||
"这几次运行的封存核对对不上,请人工核对后处理", len(dupes), "、".join(runs[:5]))
|
"这几次运行的封存核对对不上,请人工核对后处理(之后启动只核对这几处,不再整表查)",
|
||||||
|
min(len(dupes), SEQ_DUPLICATES_KEPT), "以上" if len(dupes) > SEQ_DUPLICATES_KEPT else "",
|
||||||
|
"、".join(runs[:5]))
|
||||||
|
_remember_seq_duplicates(conn, [[str(r[0]), int(r[1])] for r in dupes[:SEQ_DUPLICATES_KEPT]],
|
||||||
|
more=len(dupes) > SEQ_DUPLICATES_KEPT)
|
||||||
return
|
return
|
||||||
conn.execute(text("CREATE UNIQUE INDEX IF NOT EXISTS ux_run_events_run_seq ON run_events (run_id, seq)"))
|
conn.execute(text("CREATE UNIQUE INDEX IF NOT EXISTS ux_run_events_run_seq ON run_events (run_id, seq)"))
|
||||||
if "ix_run_events_run_seq" in names:
|
if "ix_run_events_run_seq" in names:
|
||||||
conn.execute(text("DROP INDEX IF EXISTS ix_run_events_run_seq"))
|
conn.execute(text("DROP INDEX IF EXISTS ix_run_events_run_seq"))
|
||||||
|
if known is not None:
|
||||||
|
_forget_seq_duplicates(conn)
|
||||||
|
logger.info("run_events 里的撞号都处理掉了:补上了 (run_id, seq) 的唯一索引")
|
||||||
|
|
||||||
|
|
||||||
|
def _known_seq_duplicates(conn: Any, columns: dict[str, set[str]]) -> dict[str, Any] | None: # pragma: no cover
|
||||||
|
if "settings" not in columns:
|
||||||
|
return None
|
||||||
|
from sqlalchemy import select
|
||||||
|
|
||||||
|
from app.db.models import Setting
|
||||||
|
|
||||||
|
row = conn.execute(select(Setting.value).where(Setting.key == SEQ_DUPLICATES_KEY)).first()
|
||||||
|
value = row[0] if row else None
|
||||||
|
if not isinstance(value, dict) or not isinstance(value.get("pairs"), list):
|
||||||
|
return None
|
||||||
|
return value
|
||||||
|
|
||||||
|
|
||||||
|
def _still_duplicated(conn: Any, pairs: list[Any]) -> int: # pragma: no cover
|
||||||
|
"""记下的那几处里还撞着的有几处(每处按 (run_id, seq) 索引查一次)。"""
|
||||||
|
from sqlalchemy import func, select
|
||||||
|
|
||||||
|
from app.db.models import RunEvent
|
||||||
|
|
||||||
|
left = 0
|
||||||
|
for pair in pairs:
|
||||||
|
if not (isinstance(pair, list) and len(pair) == 2):
|
||||||
|
continue
|
||||||
|
n = conn.execute(select(func.count()).select_from(RunEvent).where(
|
||||||
|
RunEvent.run_id == str(pair[0]), RunEvent.seq == int(pair[1]))).scalar()
|
||||||
|
left += int((n or 0) > 1)
|
||||||
|
return left
|
||||||
|
|
||||||
|
|
||||||
|
def _remember_seq_duplicates(conn: Any, pairs: list[list[Any]], *, more: bool) -> None: # pragma: no cover
|
||||||
|
from app.db.models import Setting
|
||||||
|
|
||||||
|
_forget_seq_duplicates(conn)
|
||||||
|
conn.execute(Setting.__table__.insert().values(
|
||||||
|
key=SEQ_DUPLICATES_KEY, value={"pairs": pairs, "more": more, "found_at": utcnow().isoformat(timespec="seconds")}))
|
||||||
|
|
||||||
|
|
||||||
|
def _forget_seq_duplicates(conn: Any) -> None: # pragma: no cover
|
||||||
|
from sqlalchemy import delete
|
||||||
|
|
||||||
|
from app.db.models import Setting
|
||||||
|
|
||||||
|
conn.execute(delete(Setting).where(Setting.key == SEQ_DUPLICATES_KEY))
|
||||||
|
|
||||||
|
|
||||||
#: init_db 在 PostgreSQL 上拿的会话级咨询锁的号:几个进程同时启动时,建表和迁移一个一个来(同一个号别处别用)
|
#: init_db 在 PostgreSQL 上拿的会话级咨询锁的号:几个进程同时启动时,建表和迁移一个一个来(同一个号别处别用)
|
||||||
|
|||||||
@@ -640,6 +640,109 @@ async def test_preflight_refuses_a_beating_peer_and_waits_out_a_stale_one():
|
|||||||
await workers.register()
|
await workers.register()
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.skipif(app_settings.db_dialect != "postgresql", reason="preflight 只在 PostgreSQL 上看")
|
||||||
|
async def test_preflight_names_every_live_process_of_another_version():
|
||||||
|
"""复核 R-b13:一个心跳在走、一个心跳停着——报错两个都列出来(只点名先看到心跳的那个,人停掉它再起还会被另一个拦下)。"""
|
||||||
|
from app.cluster import version
|
||||||
|
|
||||||
|
await _peer_version(PEER, "0.0.9+old+aaaaaaaaaaaa")
|
||||||
|
await _peer_version(GHOST, "0.0.8+older+bbbbbbbbbbbb")
|
||||||
|
|
||||||
|
async def beat() -> None:
|
||||||
|
for _ in range(15):
|
||||||
|
await asyncio.sleep(0.1)
|
||||||
|
async with SessionLocal() as session:
|
||||||
|
row = await session.get(ClusterWorker, PEER)
|
||||||
|
row.heartbeat_at = utcnow()
|
||||||
|
await session.commit()
|
||||||
|
|
||||||
|
beating = asyncio.create_task(beat())
|
||||||
|
with pytest.raises(version.VersionMismatch) as e:
|
||||||
|
await version.preflight(observe_s=0.5)
|
||||||
|
await beating
|
||||||
|
text = str(e.value)
|
||||||
|
assert "2 个版本不同" in text and "aaaaaaaaaaaa" in text and "bbbbbbbbbbbb" in text
|
||||||
|
assert "心跳在走" in text and "心跳停着" in text
|
||||||
|
|
||||||
|
|
||||||
|
def test_the_fingerprint_changes_with_the_dependency_lock(monkeypatch, tmp_path):
|
||||||
|
"""复核 R-b13:依赖锁文件变了(升了依赖)源码指纹跟着变;锁文件取不到时换成几个关键包的版本。"""
|
||||||
|
from app.cluster import version
|
||||||
|
|
||||||
|
lock = tmp_path / "requirements.lock"
|
||||||
|
monkeypatch.setattr(version, "LOCK_FILE", lock)
|
||||||
|
|
||||||
|
def fingerprint() -> str:
|
||||||
|
version.source_fingerprint.cache_clear()
|
||||||
|
return version.source_fingerprint()
|
||||||
|
|
||||||
|
try:
|
||||||
|
lock.write_text("sqlalchemy==2.0.40\n")
|
||||||
|
first = fingerprint()
|
||||||
|
assert fingerprint() == first
|
||||||
|
lock.write_text("sqlalchemy==2.0.41\n")
|
||||||
|
assert fingerprint() != first, "升了一个依赖:指纹不同"
|
||||||
|
lock.unlink()
|
||||||
|
assert version.dependency_marker().startswith(b"packages:") and b"sqlalchemy==" in version.dependency_marker()
|
||||||
|
assert fingerprint() not in (first,)
|
||||||
|
finally:
|
||||||
|
version.source_fingerprint.cache_clear()
|
||||||
|
|
||||||
|
|
||||||
|
async def test_a_known_seq_collision_is_not_rescanned_on_every_start(tmp_path, caplog):
|
||||||
|
"""复核 R-b13:存量库里有撞了号的事件——第一次启动整表数一遍、记下结果、不建唯一索引;之后启动只核对记下的那几处,不再整表
|
||||||
|
GROUP BY;撞号处理掉之后再数一遍、建索引、删掉记录。用一个单独的 SQLite 库(两种库的测试里都跑)。"""
|
||||||
|
import logging
|
||||||
|
|
||||||
|
from sqlalchemy import event, inspect, text
|
||||||
|
|
||||||
|
from app.db import base as db_base
|
||||||
|
from app.db.models import Setting
|
||||||
|
|
||||||
|
target = db_base.make_engine(f"sqlite+aiosqlite:///{tmp_path / 'seq.db'}")
|
||||||
|
scans: list[str] = []
|
||||||
|
event.listen(target.sync_engine, "before_cursor_execute",
|
||||||
|
lambda conn, cursor, statement, *a: scans.append(statement) if "GROUP BY run_id, seq" in statement else None)
|
||||||
|
try:
|
||||||
|
await db_base.create_schema(target)
|
||||||
|
from sqlalchemy.ext.asyncio import async_sessionmaker
|
||||||
|
|
||||||
|
async with target.begin() as conn:
|
||||||
|
await conn.execute(text("DROP INDEX ux_run_events_run_seq"))
|
||||||
|
await conn.execute(text("CREATE INDEX ix_run_events_run_seq ON run_events (run_id, seq)"))
|
||||||
|
async with async_sessionmaker(target, expire_on_commit=False)() as session:
|
||||||
|
session.add(Run(id="r1", status="succeeded", graph={}, input={}))
|
||||||
|
await session.flush()
|
||||||
|
session.add_all([RunEvent(id=eid, run_id="r1", seq=seq, type="log", ts=0, data={})
|
||||||
|
for eid, seq in (("e1", 1), ("e2", 1), ("e3", 2))])
|
||||||
|
await session.commit()
|
||||||
|
|
||||||
|
async def start() -> set[str]:
|
||||||
|
async with target.begin() as conn:
|
||||||
|
def run(sync_conn):
|
||||||
|
cols = {t: {c["name"] for c in inspect(sync_conn).get_columns(t)} for t in ("run_events", "settings")}
|
||||||
|
db_base._unique_run_event_seq(sync_conn, cols)
|
||||||
|
return {ix["name"] for ix in inspect(sync_conn).get_indexes("run_events")}
|
||||||
|
return await conn.run_sync(run)
|
||||||
|
|
||||||
|
caplog.set_level(logging.INFO, logger="app.db.base")
|
||||||
|
assert "ux_run_events_run_seq" not in await start() and len(scans) == 1
|
||||||
|
async with target.connect() as conn:
|
||||||
|
marker = (await conn.execute(text(f"SELECT value FROM settings WHERE key = '{db_base.SEQ_DUPLICATES_KEY}'"))).scalar()
|
||||||
|
assert '"r1"' in str(marker) and "1" in str(marker)
|
||||||
|
assert "ux_run_events_run_seq" not in await start() and len(scans) == 1, "第二次启动不再整表 GROUP BY"
|
||||||
|
assert any("还有撞了号的事件" in r.getMessage() for r in caplog.records)
|
||||||
|
async with target.begin() as conn:
|
||||||
|
await conn.execute(text("DELETE FROM run_events WHERE id = 'e2'"))
|
||||||
|
assert "ux_run_events_run_seq" in await start() and len(scans) == 2, "处理掉之后再数一遍、建上索引"
|
||||||
|
async with target.connect() as conn:
|
||||||
|
left = (await conn.execute(text(f"SELECT COUNT(*) FROM settings WHERE key = '{db_base.SEQ_DUPLICATES_KEY}'"))).scalar()
|
||||||
|
assert left == 0
|
||||||
|
assert Setting.__tablename__ == "settings"
|
||||||
|
finally:
|
||||||
|
await target.dispose()
|
||||||
|
|
||||||
|
|
||||||
@pytest.mark.sqlite_only
|
@pytest.mark.sqlite_only
|
||||||
async def test_an_idle_single_process_writes_nothing_for_coordination(monkeypatch):
|
async def test_an_idle_single_process_writes_nothing_for_coordination(monkeypatch):
|
||||||
"""复核 R-b3:SQLite(只有一个进程)空闲时协调不写库——不续租约、不写当班、不周期接手;要名额、放名额也不落库。用另一条连接
|
"""复核 R-b3:SQLite(只有一个进程)空闲时协调不写库——不续租约、不写当班、不周期接手;要名额、放名额也不落库。用另一条连接
|
||||||
|
|||||||
Reference in New Issue
Block a user