diff --git a/Dockerfile b/Dockerfile index 9137ed6c..53bf9fbc 100644 --- a/Dockerfile +++ b/Dockerfile @@ -93,6 +93,8 @@ HEALTHCHECK --interval=30s --timeout=5s --start-period=60s --start-interval=2s - # 版本号放在最后：它每次打包都不同，放前面会让装依赖那几层的缓存全部失效 ARG VERSION=dev +# 进程登记时带上构建号（app/cluster/version.py：库里还有活着的、版本不同的进程时新进程不许加入） +ENV AGENTLAB_BUILD=${VERSION} LABEL org.opencontainers.image.title="AgentLab" \ org.opencontainers.image.description="可视化 Agent 编排实验台" \ org.opencontainers.image.version="${VERSION}" \ diff --git a/README.md b/README.md index 8bec1a43..2b015bcc 100644 --- a/README.md +++ b/README.md @@ -1194,7 +1194,15 @@ SQL 形状。各进程各自的并发闸门（口径比对、自动对比、起 另一台读，可能读到不一致的页。所以多台机器时，记录表写得多的部署建议用一台机器多进程，或者把数据目录放在支持可靠文件锁的存储上。 沙箱会话（microVM）在各进程自己手上，不跨进程共享。 -**升级。** 升级要先停掉全部进程再起新版本：不支持新旧版本的进程同时连着一个库。 +**升级。** 不支持新旧版本的进程同时连着一个库，代码里拦着：每个进程登记时带上代码版本（包版本、构建号或 git 提交、源码指纹）和 +库结构版本，库里还有活着的、版本不同的进程时，新版本的进程在补列之前就拒绝启动，报错里写明是哪几个进程、什么版本；同版本的照常 +加入。升级步骤： + +1. 停掉全部旧版本的进程（平稳关停的马上删掉自己的登记；强杀的等它的租约过期，新进程启动时会等）； +2. 备份（`pg_dump` 加数据目录，见上一节）； +3. 起新版本：第一个进程补列补索引，之后的照常加入。 + +只停了一部分就起新版本，新版本起不来，旧版本照常跑，不会两边各按各的写同一批行。 **测试。** `scripts/test-multiprocess.sh` 用本机 docker 的 `postgres:16` 起临时库，在上面起两个工作进程，用演示模型跑端到端 （`backend/tests/test_multiprocess_e2e.py`，缺省跳过，`AGENTLAB_MULTIPROCESS_E2E=1` 时跑）：运行由任一进程认领、另一个进程上的 diff --git a/backend/app/cluster/version.py b/backend/app/cluster/version.py new file mode 100644 index 00000000..6e7a7d21 --- /dev/null +++ b/backend/app/cluster/version.py @@ -0,0 +1,189 @@ +"""这个进程是哪一版代码、要的是哪一版库结构；库里还有活着的、版本不同的进程时不许加入（不支持新旧版本混跑）。 + +**为什么要拦。** 升级时只停了一部分进程、或者新旧两批容器同时连着一个库：新版的进程启动时补列、补索引，旧版的进程不认识新的列和 +规矩（认领、租约、接手的说法），两边各按各的写同一批行。以前只在 README 里写「升级要先停掉全部进程」，现在代码里拦住。 + +- 代码版本 code_version：包版本 + 构建号（AGENTLAB_BUILD，Docker 镜像构建时写进去）或 git 提交 + 源码指纹（app/ 下全部 .py + 的内容哈希）。比的是源码指纹：同一份代码一定相同，改了一行就不同，取不到 git、没有构建号时也区分得开。 +- 库结构版本 schema_version：模型里各表各列（名字、类型、可空）加迁移清单的哈希。 + +怎么拦：PostgreSQL 时，补列之前先看一眼（preflight）——还有活着的、版本不同的进程就不启动；它的心跳停着（多半是强杀之后留下的、 +租约还没到期）就等它过期再起，心跳还在走就照实报错：哪几个进程、什么版本、要先停掉再升级。登记那一下再在事务级咨询锁里查一遍 +（cluster/workers.register），两个不同版本同时启动也只有先登记上的那一个起得来。SQLite 只有一个进程，不用看。 +""" +from __future__ import annotations + +import asyncio +import hashlib +import logging +import os +import subprocess +from functools import lru_cache +from pathlib import Path +from typing import Any + +from app.cluster import identity + +logger = logging.getLogger(__name__) + +_APP_DIR = Path(__file__).resolve().parents[1] + + +class VersionMismatch(RuntimeError): + """库里还有活着的、版本不同的进程。""" + + +@lru_cache(maxsize=1) +def source_fingerprint() -> str: + digest = hashlib.sha256() + for path in sorted(_APP_DIR.rglob("*.py")): + if "__pycache__" in path.parts: + continue + digest.update(str(path.relative_to(_APP_DIR)).encode()) + try: + digest.update(path.read_bytes()) + except OSError: + continue + return digest.hexdigest()[:12] + + +def _package() -> str: + try: + from importlib.metadata import version + + return version("agentlab-backend") + except Exception: # noqa: BLE001 + return "0" + + +def _build() -> str: + if build := (os.environ.get("AGENTLAB_BUILD") or "").strip(): + return build[:40] + try: + out = subprocess.run(["git", "rev-parse", "--short=10", "HEAD"], cwd=str(_APP_DIR), capture_output=True, + text=True, timeout=3) + return out.stdout.strip() if out.returncode == 0 else "" + except (OSError, subprocess.SubprocessError): + return "" + + +@lru_cache(maxsize=1) +def code_version() -> str: + """包版本+构建号或 git 提交+源码指纹，例如 0.1.0+e6a3ee8f12+3f9a…（最后一段是比较用的）。""" + return "+".join(p for p in (_package(), _build(), source_fingerprint()) if p) + + +def fingerprint_of(version: str | None) -> str: + return (version or "").rsplit("+", 1)[-1] + + +@lru_cache(maxsize=1) +def schema_version() -> str: + from app.db import models # noqa: F401 模型都登记进 metadata + from app.db.base import _COLUMN_MIGRATIONS, _INDEX_MIGRATIONS, Base + + digest = hashlib.sha256() + for name in sorted(Base.metadata.tables): + table = Base.metadata.tables[name] + digest.update(name.encode()) + for col in table.columns: + digest.update(f"{col.name}:{col.type!r}:{col.nullable}".encode()) + digest.update(repr(_COLUMN_MIGRATIONS).encode() + repr(_INDEX_MIGRATIONS).encode()) + return digest.hexdigest()[:12] + + +def same(code: str | None, schema: str | None) -> bool: + return fingerprint_of(code) == fingerprint_of(code_version()) and (schema or "") == schema_version() + + +async def _peers(conn: Any) -> list[dict[str, Any]] | None: + """库里登记着的别的进程（id、版本、租约、心跳）；表还不在（新库）交回 None。在一条连接上查（preflight 在建表之前）。""" + from sqlalchemy import inspect, text + + def has(sync_conn: Any) -> tuple[bool, set[str]]: + insp = inspect(sync_conn) + if not insp.has_table("cluster_workers"): + return False, set() + return True, {c["name"] for c in insp.get_columns("cluster_workers")} + + exists, cols = await conn.run_sync(has) + if not exists: + return None + extra = ", ".join(c for c in ("code_version", "schema_version") if c in cols) + rows = (await conn.execute(text(f"SELECT id, lease_until, heartbeat_at{', ' + extra if extra else ''} " + "FROM cluster_workers WHERE id <> :me"), {"me": identity.owner()})).mappings().all() + return [dict(r) for r in rows] + + +def _describe(peer: dict[str, Any]) -> str: + code = peer.get("code_version") or "（升级前的版本，没有记版本）" + schema = peer.get("schema_version") or "未知" + return f"{identity.describe(peer['id'])}（代码 {code}，库结构 {schema}）" + + +async def preflight(*, observe_s: float | None = None) -> None: + """PostgreSQL 时、建表补列之前：库里还有活着的、版本不同的进程就不启动（见模块说明）。observe_s：看它心跳还在不在走的时长 + （缺省两拍心跳）。""" + from app.core.config import settings + + if settings.db_dialect != "postgresql": + return + from app.cluster.workers import heartbeat_s, lease_s + from app.db.base import utcnow + + observe = 2 * heartbeat_s() + 1.0 if observe_s is None else observe_s + loop = asyncio.get_running_loop() + started = loop.time() + first_beats: dict[str, Any] = {} + warned = False + while True: + mismatched = await _mismatched_alive(utcnow()) + if not mismatched: + return + for peer in mismatched: + 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"])] + waited = loop.time() - started + if beating or waited >= lease_s() * 2 + observe: + raise VersionMismatch(_message(beating or mismatched)) + if waited >= observe and not warned: + warned = True + logger.warning("库里登记着 %d 个版本不同的进程，心跳停着（多半是强杀之后留下的）：等它们的租约过期再启动", + len(mismatched)) + await asyncio.sleep(min(0.5, max(0.1, heartbeat_s() / 2))) + + +async def _mismatched_alive(now: Any) -> list[dict[str, Any]]: + from app.db.base import engine + + async with engine.connect() as conn: + peers = await _peers(conn) + if not peers: + return [] + naive = now.replace(tzinfo=None) + alive = [p for p in peers if p.get("lease_until") is not None and _naive(p["lease_until"]) > naive] + return [p for p in alive if not same(p.get("code_version"), p.get("schema_version"))] + + +def _naive(moment: Any) -> Any: + return moment.replace(tzinfo=None) if getattr(moment, "tzinfo", None) else moment + + +def _message(peers: list[dict[str, Any]]) -> str: + listed = "；".join(_describe(p) for p in peers) + return (f"库里还有 {len(peers)} 个版本不同的 AgentLab 进程活着：{listed}。本进程是代码 {code_version()}、库结构 " + f"{schema_version()}。不支持新旧版本混跑：先停掉这些进程再起新版本（升级步骤见 README「多进程」的「升级」）") + + +def check_rows(rows: list[Any], now: Any) -> None: + """登记那一下（cluster/workers.register，持着事务级咨询锁）再查一遍：别的活着的进程里有版本不同的就抛 VersionMismatch。""" + naive = now.replace(tzinfo=None) + bad = [{"id": r.id, "code_version": r.code_version, "schema_version": r.schema_version} for r in rows + if r.lease_until is not None and _naive(r.lease_until) > naive and r.id != identity.owner() + and not same(r.code_version, r.schema_version)] + if bad: + raise VersionMismatch(_message(bad)) + + +__all__ = ["VersionMismatch", "check_rows", "code_version", "preflight", "same", "schema_version", "source_fingerprint"] diff --git a/backend/app/cluster/workers.py b/backend/app/cluster/workers.py index db9dd624..ca7edf45 100644 --- a/backend/app/cluster/workers.py +++ b/backend/app/cluster/workers.py @@ -25,6 +25,10 @@ from app.db.base import SessionLocal, utcnow logger = logging.getLogger(__name__) +#: 登记时全站串行用的事务级咨询锁的号 +_REGISTER_KEY = 0x4167576F726B6572 # "AgWorker" + + def lease_s() -> float: """租约多长（秒）：至少是心跳间隔的三倍——库一时慢了、一拍没续上，不算进程没了。""" return max(float(settings.worker_lease_s), 3.0 * float(settings.worker_heartbeat_s)) @@ -83,15 +87,25 @@ async def register(*, exclusive: bool = False) -> None: 守卫已经确认没有别的进程在用这个数据目录，见 cluster/guard.py）。""" from app.db.models_cluster import ClusterWorker + from app.cluster import version + me = identity.owner() now = utcnow() async with SessionLocal() as session: if exclusive: await session.execute(delete(ClusterWorker).where(ClusterWorker.id != me)) + elif session.get_bind().dialect.name == "postgresql": + # 查「还有没有活着的、版本不同的进程」和登记在同一个事务里、全站串行：两个不同版本同时启动只有先登记上的起得来 + from sqlalchemy import text + + await session.execute(text("SELECT pg_advisory_xact_lock(:k)"), {"k": _REGISTER_KEY}) + others = (await session.execute(select(ClusterWorker).where(ClusterWorker.id != me))).scalars().all() + version.check_rows(list(others), now) row = await session.get(ClusterWorker, me) if row is None: session.add(ClusterWorker(id=me, host=identity.host(), pid=os.getpid(), started_at=now, heartbeat_at=now, - lease_until=now + timedelta(seconds=lease_s()), info=_info())) + lease_until=now + timedelta(seconds=lease_s()), info=_info(), + code_version=version.code_version(), schema_version=version.schema_version())) else: row.heartbeat_at, row.lease_until = now, now + timedelta(seconds=lease_s()) await session.commit() @@ -137,6 +151,7 @@ async def listing() -> list[dict[str, Any]]: async with SessionLocal() as session: rows = (await session.execute(select(ClusterWorker).order_by(ClusterWorker.started_at, ClusterWorker.id))).scalars().all() return [{"id": r.id, "host": r.host, "pid": r.pid, "alive": r.lease_until > now, "me": r.id == identity.owner(), + "code_version": r.code_version, "schema_version": r.schema_version, "started_at": r.started_at.isoformat() if r.started_at else None, "heartbeat_at": r.heartbeat_at.isoformat() if r.heartbeat_at else None} for r in rows] diff --git a/backend/app/db/base.py b/backend/app/db/base.py index 83d0b117..1747e6e1 100644 --- a/backend/app/db/base.py +++ b/backend/app/db/base.py @@ -454,6 +454,9 @@ _COLUMN_MIGRATIONS: list[tuple[str, str, str]] = [ ("publish_plans", "owner", "VARCHAR(120)"), # 问数据：等不及的那次复核还在后台跑（实时流在它回来之前不关；多进程时别的进程上的流也看得到） ("chat_turn_jobs", "late_review_at", "DATETIME"), + # 工作进程登记的代码版本、库结构版本（app/cluster/version.py：不许新旧版本混跑） + ("cluster_workers", "code_version", "VARCHAR(200)"), + ("cluster_workers", "schema_version", "VARCHAR(64)"), ] #: 迁移补上的列要建的索引：(索引名, 表, 列或多列)。create_all 不碰已有的表，表上新列的索引也就不会建。 diff --git a/backend/app/db/models_cluster.py b/backend/app/db/models_cluster.py index 9555be21..6e2fe63f 100644 --- a/backend/app/db/models_cluster.py +++ b/backend/app/db/models_cluster.py @@ -38,6 +38,9 @@ class ClusterWorker(Base): lease_until: Mapped[datetime] = mapped_column(UTCDateTime, default=utcnow, index=True) #: 版本、数据目录之类给人看的信息 info: Mapped[dict[str, Any]] = mapped_column(default=dict) + #: 代码版本、库结构版本（app/cluster/version.py）：库里有活着的、版本不同的进程时新进程不许加入 + code_version: Mapped[str | None] = mapped_column(String(200), default=None) + schema_version: Mapped[str | None] = mapped_column(String(64), default=None) class ClusterLease(Base): diff --git a/backend/app/main.py b/backend/app/main.py index c558ae52..5aa95a8e 100644 --- a/backend/app/main.py +++ b/backend/app/main.py @@ -62,6 +62,10 @@ async def lifespan(_: FastAPI) -> AsyncIterator[None]: from app.db.base import check_pgbouncer check_pgbouncer() + # 不许新旧版本混跑（app/cluster/version.py）：库里还有活着的、版本不同的进程就不启动——在补列之前看 + from app.cluster import version + + await version.preflight() await init_db() # 多进程协调（app/cluster）：登记这个工作进程、开始心跳和进程间通知、争调度当班。要在下面的启动整理之前：整理只碰属主不在了的 # 行，要先知道别的进程谁还活着 diff --git a/backend/tests/test_cluster.py b/backend/tests/test_cluster.py index 557faf08..a819a16f 100644 --- a/backend/tests/test_cluster.py +++ b/backend/tests/test_cluster.py @@ -14,6 +14,7 @@ import pytest from sqlalchemy import delete, select from app.cluster import identity, leader as leader_mod, slots, state, workers +from app.core.config import settings as app_settings from app.db.base import SessionLocal, new_id, utcnow from app.db.models import Run, RunEvent from app.db.models_cluster import ClusterLease, ClusterRequest, ClusterWorker, LaneTicket, RunSlot @@ -375,3 +376,78 @@ def test_a_daily_quota_holds_across_two_processes(): outs = [p.communicate(timeout=120) for p in procs] assert all(p.returncode == 0 for p in procs), [o[1][-2000:] for o in outs] assert sum(int(o[0].strip().splitlines()[-1]) for o in outs) == 10 + + +# -------------------------------------------------------------------------- +# 不许新旧版本混跑（cluster/version.py） +# -------------------------------------------------------------------------- + +async def _peer_version(owner: str, code: str | None, *, lease_s: float = 60.0) -> None: + now = utcnow() + async with SessionLocal() as session: + session.add(ClusterWorker(id=owner, host="peer-host", pid=4242, started_at=now, heartbeat_at=now, + lease_until=now + timedelta(seconds=lease_s), code_version=code, + schema_version=None if code is None else _schema())) + await session.commit() + + +def _schema() -> str: + from app.cluster import version + + return version.schema_version() + + +async def _rows(): + async with SessionLocal() as session: + return list((await session.execute(select(ClusterWorker))).scalars()) + + +async def test_a_live_process_of_another_version_is_named_and_refused(): + from app.cluster import version + + await _peer_version(PEER, "0.0.9+old+aaaaaaaaaaaa") + with pytest.raises(version.VersionMismatch) as e: + version.check_rows(await _rows(), utcnow()) + assert "peer-host 上的进程 4242" in str(e.value) and "0.0.9+old+aaaaaaaaaaaa" in str(e.value) and "先停掉" in str(e.value) + + +async def test_the_same_version_joins_and_expired_or_unversioned_dead_ones_do_not_count(): + from app.cluster import version + + await _peer_version(PEER, version.code_version()) + await _peer_version(GHOST, "0.0.9+old+aaaaaaaaaaaa", lease_s=-5) + version.check_rows(await _rows(), utcnow()) + assert version.same("0.1.0+other-build+" + version.source_fingerprint(), version.schema_version()), \ + "比的是源码指纹：构建号写法不同、代码相同算同一版" + + +@pytest.mark.skipif(app_settings.db_dialect != "postgresql", reason="只在 PostgreSQL 上拦（SQLite 只有一个进程）") +async def test_preflight_refuses_a_beating_peer_and_waits_out_a_stale_one(): + from app.cluster import version + + await _peer_version(PEER, "0.0.9+old+aaaaaaaaaaaa") + + async def beat() -> None: + for _ in range(20): + 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): + await version.preflight(observe_s=0.5) + await beating + async with SessionLocal() as session: + await session.execute(delete(ClusterWorker).where(ClusterWorker.id == PEER)) + await session.commit() + # 强杀之后留下的旧版登记（心跳停着、租约还有 1 秒）：等它过期就起 + await _peer_version(GHOST, "0.0.9+old+aaaaaaaaaaaa", lease_s=1.0) + loop = asyncio.get_running_loop() + t0 = loop.time() + await version.preflight(observe_s=0.3) + assert 0.5 < loop.time() - t0 < 5 + with pytest.raises(version.VersionMismatch): + await _peer_version(PEER, "0.0.9+old+aaaaaaaaaaaa") + await workers.register()