diff --git a/backend/app/core/config.py b/backend/app/core/config.py index bdb311f7..890db795 100644 --- a/backend/app/core/config.py +++ b/backend/app/core/config.py @@ -62,6 +62,9 @@ class Settings(BaseSettings): db_max_overflow: int = 20 db_pool_timeout: int = 30 db_pool_recycle: int = 1800 + # 启动时库还连不上(和 PostgreSQL 容器一起起、库还在初始化)最多等多少秒,按退避重试、照实打日志;过了再报错退出。 + # 密码不对、库不存在这种等也没用的不等 + db_startup_timeout: int = 90 # --- 多进程(app/cluster,docs/design/orchestration-round §5 W2-b)--- # 几个工作进程(scripts/prod.sh、Docker 的入口按它给 uvicorn --workers)。只有应用库是 PostgreSQL 时才能大于 1: diff --git a/backend/app/db/base.py b/backend/app/db/base.py index ee3edaca..b8224fb7 100644 --- a/backend/app/db/base.py +++ b/backend/app/db/base.py @@ -542,6 +542,68 @@ async def create_schema(target: AsyncEngine) -> None: await conn.run_sync(_migrate) +#: 连不上时等也没用的错误(SQLSTATE 的类):28 认证失败、3D 库不存在、42 语法或权限。别的(拒绝连接、库正在启动 57P03、 +#: 超时、网络断了)按「还没就绪」退避重试 +_NOT_TRANSIENT_SQLSTATES = ("28", "3D", "42") + + +def _not_transient(error: BaseException) -> bool: + from app.db.portable import _driver_error + + seen: Any = error + for _ in range(4): + state = str(getattr(seen, "sqlstate", "") or "") + if state and state != "57P03": + return state.startswith(_NOT_TRANSIENT_SQLSTATES) + seen = getattr(seen, "__cause__", None) or getattr(seen, "orig", None) + if seen is None: + break + cause = _driver_error(error) + return bool(cause is not None and str(cause.sqlstate).startswith(_NOT_TRANSIENT_SQLSTATES)) + + +async def wait_for_database(target: AsyncEngine | None = None, *, timeout: float | None = None, + sleep: Any = None, clock: Any = None) -> None: + """PostgreSQL 还连不上时按退避(1、2、4、5、5… 秒)重试,每次照实打一条日志,等够 timeout 秒(缺省 + AGENTLAB_DB_STARTUP_TIMEOUT)还不行就把最后一次的错抛出去。和数据库容器一起起、库还在初始化时,应用不至于一上来就崩。 + 密码不对、库不存在这种等也没用的立刻抛出。SQLite 不等。""" + import asyncio + import time + + from sqlalchemy import text + + from app.core.errors import first_line + + target = target or engine + if target.dialect.name != "postgresql": + return + budget = settings.db_startup_timeout if timeout is None else timeout + pause = sleep or asyncio.sleep + now = clock or time.monotonic + started = now() + delay = 1.0 + attempt = 0 + while True: + attempt += 1 + try: + async with target.connect() as conn: + await conn.execute(text("SELECT 1")) + if attempt > 1: + logger.info("应用库连上了(第 %d 次,等了 %.0f 秒)", attempt, now() - started) + return + except Exception as e: # noqa: BLE001 - 判断完是不是该等再决定抛不抛 + waited = now() - started + if _not_transient(e) or waited + delay > budget: + logger.error("应用库连不上,不再等了(等了 %.0f 秒):%s", waited, first_line(e)) + raise + logger.warning("应用库还没就绪(%s),%.0f 秒后重试(已等 %.0f 秒,最多 %d 秒)", + first_line(e), delay, waited, budget) + await pause(delay) + delay = min(delay * 2, 5.0) + + async def init_db() -> None: logger.info("应用库:%s", describe_database()) + # 库还没起来(和 PostgreSQL 容器一起起)时先等它;建表、补列、补索引在库里的咨询锁里做,几个工作进程同时启动只做一次 + await wait_for_database() await create_schema(engine) diff --git a/backend/app/db/to_postgres.py b/backend/app/db/to_postgres.py index bb6093cc..b861871d 100644 --- a/backend/app/db/to_postgres.py +++ b/backend/app/db/to_postgres.py @@ -584,6 +584,13 @@ def main(argv: list[str] | None = None) -> int: except (CopyRefused, ValueError) as e: print(f"没有导入:{e}", file=sys.stderr) return 2 + except Exception as e: # noqa: BLE001 - 连不上、库报错:照实说一句,不甩一屏调用栈 + from app.core.errors import first_line + + what = "核对" if args.dry_run else "导入" + tail = "什么都没写" if args.dry_run else "已经写进目标库的行留着,处理好之后重新运行只补缺的" + print(f"没有{what}完:{type(e).__name__}:{first_line(e) or '(没有说明)'}。{tail}", file=sys.stderr) + return 1 for line in render(report): print(line) return 0 if report.ok else 1 diff --git a/backend/app/ops.py b/backend/app/ops.py new file mode 100644 index 00000000..5694a489 --- /dev/null +++ b/backend/app/ops.py @@ -0,0 +1,163 @@ +"""部署脚本(deploy/deploy.sh、scripts/dev.sh)在容器里调的几件小事:`python -m app.ops <命令>`。 + +放在镜像里用 Python 做,不在宿主机上用 shell 做:数据目录归容器里的用户(uid 10001),宿主机上跑 deploy.sh 的人不一定写得 +进去;生成密码、改权限、拷 SQLite、删旧备份在容器里以同一个用户做,属主和权限都对。只用标准库,不连应用库、不读设置。 + + postgres-secrets 内置 PostgreSQL 的密码:<数据目录>/.postgres/password(给 PostgreSQL 容器建库时读)和 pgpass(给应用 + 连库时读,PGPASSFILE 指过来)。已有就不动;密码不打印、不进任何参数和日志 + state 数据目录用的是哪种库:没切过是 sqlite;--set builtin 记成「已切到内置 PostgreSQL」(deploy.sh 只在导入、 + 核对都通过之后才记,记下之前出错就还是 sqlite);migrating 是「开始往内置库里导了、还没切成」—— + 用的仍是 SQLite,只是让下一次 migrate-to-postgres 认得出数据卷里那半份是自己上次留下的 + backup-sqlite 拷一份 agentlab.db、checkpoints.db 到 <数据目录>/backups/sqlite/<时间>/(SQLite 的备份接口,一致的一份), + 只留最近 --keep 份 + prune 目录里按名字排序只留最近 --keep 个(pg_dump 的备份用) +""" + +from __future__ import annotations + +import argparse +import secrets +import sqlite3 +import sys +from datetime import datetime, timezone +from pathlib import Path + +from app.core import atomic + +#: 内置 PostgreSQL 的东西放在数据目录的这个子目录里 +POSTGRES_DIR = ".postgres" +STATE_FILE = "state" +MODES = ("sqlite", "builtin", "migrating") + + +def _stamp() -> str: + return datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ") + + +def postgres_secrets(data: Path, *, folder: str, host: str, port: int, database: str, user: str) -> bool: + """建密码文件和 pgpass(都是 0600)。已有密码就沿用(库是按它初始化的),pgpass 按现在的主机名重写。交回是不是新建的密码。""" + root = data / folder + root.mkdir(parents=True, exist_ok=True) + # 0711:别人列不出目录、读不到密码(两个文件 0600),但宿主机上跑 deploy.sh 的人能按名字读到 state(0644) + root.chmod(0o711) + password_file = root / "password" + made = False + if not password_file.exists(): + made = atomic.create_text(password_file, secrets.token_urlsafe(32) + "\n", mode=0o600) + password = password_file.read_text(encoding="utf-8").strip() + if not password: + raise SystemExit(f"{password_file} 是空的:删掉它再来会生成新的(库已经按旧密码初始化过的话,要连库一起重建)") + # pgpass 的格式:主机:端口:库:用户:密码,冒号和反斜杠要转义 + escaped = password.replace("\\", "\\\\").replace(":", "\\:") + atomic.write_text(root / "pgpass", f"{host}:{port}:{database}:{user}:{escaped}\n", mode=0o600) + return made + + +def read_state(data: Path) -> str: + """数据目录用的是哪种库:sqlite(没切过)、builtin(已切到内置 PostgreSQL)或 migrating(切到一半,用的仍是 SQLite)。""" + path = data / POSTGRES_DIR / STATE_FILE + try: + for line in path.read_text(encoding="utf-8").splitlines(): + if line.startswith("mode="): + mode = line.split("=", 1)[1].strip() + return mode if mode in MODES else "sqlite" + except FileNotFoundError: + pass + return "sqlite" + + +def write_state(data: Path, mode: str) -> None: + if mode not in MODES: + raise SystemExit(f"认不出的模式:{mode}({' / '.join(MODES)})") + root = data / POSTGRES_DIR + if not root.exists(): + root.mkdir(parents=True) + root.chmod(0o711) + # 不是密钥:宿主机上的 deploy.sh 直接读它,0644 + atomic.write_text(root / STATE_FILE, f"mode={mode}\nsince={datetime.now(timezone.utc).isoformat()}\n", mode=0o644) + + +def _copy_sqlite(src: Path, dst: Path) -> None: + # 读写打开、只许读:WAL 模式的库用 mode=ro 打开,关掉时会留下它自己建的 -wal(同 db/to_postgres.snapshot) + conn = sqlite3.connect(src) + conn.execute("PRAGMA query_only = 1") + try: + out = sqlite3.connect(dst) + try: + conn.backup(out) + finally: + out.close() + finally: + conn.close() + + +def prune(folder: Path, keep: int) -> list[Path]: + """folder 里按名字(名字里带时间,排序就是先后)只留最近 keep 个,交回删掉的。""" + if keep < 1 or not folder.is_dir(): + return [] + items = sorted((p for p in folder.iterdir() if not p.name.startswith(".")), key=lambda p: p.name) + gone = items[:-keep] if len(items) > keep else [] + for path in gone: + if path.is_dir(): + for child in sorted(path.rglob("*"), reverse=True): + child.unlink() if child.is_file() or child.is_symlink() else child.rmdir() + path.rmdir() + else: + path.unlink() + return gone + + +def backup_sqlite(data: Path, *, keep: int) -> Path | None: + """agentlab.db、checkpoints.db 拷到 backups/sqlite/<时间>/。两个都没有(库在 PostgreSQL)交回 None。""" + sources = [p for p in (data / "agentlab.db", data / "checkpoints.db") if p.is_file()] + if not sources: + return None + root = data / "backups" / "sqlite" + target = root / _stamp() + target.mkdir(parents=True, exist_ok=True) + for src in sources: + _copy_sqlite(src, target / src.name) + prune(root, keep) + return target + + +def main(argv: list[str] | None = None) -> int: + parser = argparse.ArgumentParser(prog="python -m app.ops", description="部署脚本在容器里调的几件小事") + sub = parser.add_subparsers(dest="command", required=True) + p = sub.add_parser("postgres-secrets") + p.add_argument("--data", default="/data") + p.add_argument("--dir", default=POSTGRES_DIR) + p.add_argument("--host", required=True) + p.add_argument("--port", type=int, default=5432) + p.add_argument("--db", default="agentlab") + p.add_argument("--user", default="agentlab") + s = sub.add_parser("state") + s.add_argument("--data", default="/data") + s.add_argument("--set", choices=MODES) + b = sub.add_parser("backup-sqlite") + b.add_argument("--data", default="/data") + b.add_argument("--keep", type=int, default=5) + r = sub.add_parser("prune") + r.add_argument("--dir", required=True) + r.add_argument("--keep", type=int, default=5) + args = parser.parse_args(argv) + + if args.command == "postgres-secrets": + made = postgres_secrets(Path(args.data), folder=args.dir, host=args.host, port=args.port, database=args.db, + user=args.user) + print("生成了新密码" if made else "沿用已有的密码") + elif args.command == "state": + if args.set: + write_state(Path(args.data), args.set) + print(read_state(Path(args.data))) + elif args.command == "backup-sqlite": + target = backup_sqlite(Path(args.data), keep=args.keep) + print(target if target is not None else "没有 SQLite 库,不用备份") + elif args.command == "prune": + for gone in prune(Path(args.dir), args.keep): + print(f"删掉旧备份 {gone.name}") + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/backend/tests/test_copy_to_postgres.py b/backend/tests/test_copy_to_postgres.py index 23a41350..1030f46c 100644 --- a/backend/tests/test_copy_to_postgres.py +++ b/backend/tests/test_copy_to_postgres.py @@ -182,6 +182,17 @@ def test_the_command_needs_a_target(tmp_path, monkeypatch, capsys): assert "只认 PostgreSQL" in capsys.readouterr().err +def test_an_unreachable_target_is_reported_in_one_line(tmp_path, capsys): + """目标库连不上(库的容器没起来、被停了):一句话说清楚、退出码 1,不甩一屏调用栈(deploy.sh migrate-to-postgres 照它回退)。""" + import sqlite3 + + sqlite3.connect(tmp_path / "agentlab.db").execute("CREATE TABLE t (x)").connection.close() + for extra, what in (([], "没有导入完"), (["--dry-run"], "没有核对完")): + assert to_postgres.main([str(tmp_path), "postgresql+asyncpg://u:p@127.0.0.1:1/agentlab", *extra]) == 1 + err = capsys.readouterr().err + assert err.startswith(what) and "Traceback" not in err and len(err.strip().splitlines()) == 1 + + def test_the_dispatcher_knows_its_commands(capsys): from app.db.__main__ import main diff --git a/backend/tests/test_db_startup.py b/backend/tests/test_db_startup.py new file mode 100644 index 00000000..c18f0bb1 --- /dev/null +++ b/backend/tests/test_db_startup.py @@ -0,0 +1,137 @@ +"""启动时等库(db/base.wait_for_database):和 PostgreSQL 容器一起起、库还没就绪时按退避重试、照实打日志,等够了再报错; +密码不对、库不存在这种等也没用的立刻报错;SQLite 不等。""" + +from __future__ import annotations + +import logging + +import pytest + +from app.db.base import make_engine, wait_for_database + + +class _Dialect: + def __init__(self, name: str) -> None: + self.name = name + + +class _Conn: + async def __aenter__(self): + return self + + async def __aexit__(self, *exc): + return False + + async def execute(self, _statement): + return None + + +class _Engine: + """connect() 依次抛 errors 里的错,用完了就连上。""" + + def __init__(self, errors: list[BaseException], dialect: str = "postgresql") -> None: + self.errors = list(errors) + self.dialect = _Dialect(dialect) + self.attempts = 0 + + def connect(self): + self.attempts += 1 + if self.errors: + raise self.errors.pop(0) + return _Conn() + + +class _Clock: + def __init__(self) -> None: + self.now = 0.0 + self.slept: list[float] = [] + + def __call__(self) -> float: + return self.now + + async def sleep(self, seconds: float) -> None: + self.slept.append(seconds) + self.now += seconds + + +class _SqlState(Exception): + def __init__(self, sqlstate: str) -> None: + super().__init__(f"sqlstate {sqlstate}") + self.sqlstate = sqlstate + + +async def test_retries_with_backoff_until_the_database_answers(caplog): + clock = _Clock() + engine = _Engine([ConnectionRefusedError("拒绝连接"), OSError("名字解析不了"), _SqlState("57P03")]) + with caplog.at_level(logging.INFO, logger="app.db.base"): + await wait_for_database(engine, timeout=60, sleep=clock.sleep, clock=clock) + assert engine.attempts == 4 and clock.slept == [1.0, 2.0, 4.0] + warnings = [r.getMessage() for r in caplog.records if r.levelno == logging.WARNING] + assert len(warnings) == 3 and all("应用库还没就绪" in w for w in warnings) + assert "拒绝连接" in warnings[0] and "最多 60 秒" in warnings[0] + assert any("应用库连上了(第 4 次" in r.getMessage() for r in caplog.records) + + +async def test_gives_up_after_the_budget_and_raises_the_last_error(caplog): + clock = _Clock() + engine = _Engine([ConnectionRefusedError(f"第 {i} 次") for i in range(100)]) + with caplog.at_level(logging.INFO, logger="app.db.base"), pytest.raises(ConnectionRefusedError, match="第 5 次"): + await wait_for_database(engine, timeout=20, sleep=clock.sleep, clock=clock) + # 退避 1、2、4、5、5:等到第 17 秒,再等 5 秒就过了 20 秒,不等了,抛第 6 次(下标 5)的错 + assert clock.slept == [1.0, 2.0, 4.0, 5.0, 5.0] + assert any("不再等了" in r.getMessage() for r in caplog.records if r.levelno == logging.ERROR) + + +@pytest.mark.parametrize("sqlstate", ["28P01", "28000", "3D000", "42501"]) +async def test_errors_waiting_cannot_fix_are_raised_at_once(sqlstate): + clock = _Clock() + engine = _Engine([_SqlState(sqlstate)]) + with pytest.raises(_SqlState): + await wait_for_database(engine, timeout=60, sleep=clock.sleep, clock=clock) + assert engine.attempts == 1 and clock.slept == [] + + +async def test_sqlstate_on_the_wrapped_cause_counts(): + clock = _Clock() + wrapper = RuntimeError("SQLAlchemy 包了一层") + wrapper.__cause__ = _SqlState("28P01") + with pytest.raises(RuntimeError): + await wait_for_database(_Engine([wrapper]), timeout=60, sleep=clock.sleep, clock=clock) + assert clock.slept == [] + + +async def test_sqlite_does_not_wait(): + engine = _Engine([ConnectionRefusedError("不该连")], dialect="sqlite") + await wait_for_database(engine, timeout=60) + assert engine.attempts == 0 + + +async def test_a_real_refused_connection_is_treated_as_not_ready_yet(): + """真的 asyncpg:端口上没有服务时(库容器还没起来)算「还没就绪」,按退避重试,到时间才抛。""" + pytest.importorskip("asyncpg") + clock = _Clock() + engine = make_engine("postgresql+asyncpg://agentlab:x@127.0.0.1:1/agentlab") + try: + with pytest.raises(OSError): + await wait_for_database(engine, timeout=4, sleep=clock.sleep, clock=clock) + finally: + await engine.dispose() + assert clock.slept == [1.0, 2.0] + + +async def test_a_missing_database_on_a_real_server_is_raised_at_once(): + """真的 PostgreSQL(AGENTLAB_TEST_DATABASE_URL):库不存在(3D000)立刻报错,不等。""" + from app.core.config import settings + + if settings.db_dialect != "postgresql": + pytest.skip("要 PostgreSQL") + base, _, _ = settings.db_url.rpartition("/") + clock = _Clock() + engine = make_engine(f"{base}/agentlab_no_such_database") + try: + with pytest.raises(Exception) as caught: + await wait_for_database(engine, timeout=60, sleep=clock.sleep, clock=clock) + finally: + await engine.dispose() + assert clock.slept == [], caught.value + assert "agentlab_no_such_database" in str(caught.value) diff --git a/backend/tests/test_ops.py b/backend/tests/test_ops.py new file mode 100644 index 00000000..2d905c85 --- /dev/null +++ b/backend/tests/test_ops.py @@ -0,0 +1,108 @@ +"""app/ops.py:deploy.sh、dev.sh 在容器里调的几件小事(密码文件、状态、SQLite 备份、删旧备份)。""" + +from __future__ import annotations + +import sqlite3 +import stat + +import pytest + +from app import ops + + +def _mode(path) -> int: + return stat.S_IMODE(path.stat().st_mode) + + +def test_secrets_are_generated_once_with_tight_permissions(tmp_path, capsys): + assert ops.main(["postgres-secrets", "--data", str(tmp_path), "--host", "agl-postgres"]) == 0 + assert capsys.readouterr().out.strip() == "生成了新密码" + folder = tmp_path / ".postgres" + password = (folder / "password").read_text().strip() + assert len(password) >= 32 + assert _mode(folder / "password") == 0o600 and _mode(folder / "pgpass") == 0o600 + assert _mode(folder) == 0o711, "别人列不出目录,但按名字读得到 state" + assert (folder / "pgpass").read_text() == f"agl-postgres:5432:agentlab:agentlab:{password}\n" + + assert ops.main(["postgres-secrets", "--data", str(tmp_path), "--host", "other", "--port", "55432"]) == 0 + out = capsys.readouterr().out + assert out.strip() == "沿用已有的密码" and password not in out, "密码不打印" + assert (folder / "password").read_text().strip() == password + assert (folder / "pgpass").read_text().startswith("other:55432:agentlab:agentlab:"), "pgpass 按现在的主机名重写" + + +def test_pgpass_escapes_colons_and_backslashes(tmp_path): + folder = tmp_path / ".postgres-dev" + folder.mkdir() + (folder / "password").write_text("a:b\\c\n") + assert ops.postgres_secrets(tmp_path, folder=".postgres-dev", host="127.0.0.1", port=55432, database="agentlab", + user="agentlab") is False + assert (folder / "pgpass").read_text() == "127.0.0.1:55432:agentlab:agentlab:a\\:b\\\\c\n" + + +def test_an_empty_password_file_is_refused(tmp_path): + (tmp_path / ".postgres").mkdir() + (tmp_path / ".postgres" / "password").write_text("\n") + with pytest.raises(SystemExit, match="是空的"): + ops.postgres_secrets(tmp_path, folder=".postgres", host="h", port=5432, database="d", user="u") + + +def test_state_defaults_to_sqlite_and_records_switches(tmp_path, capsys): + assert ops.read_state(tmp_path) == "sqlite" + for mode in ("migrating", "builtin", "sqlite"): + assert ops.main(["state", "--data", str(tmp_path), "--set", mode]) == 0 + assert capsys.readouterr().out.strip() == mode + assert ops.read_state(tmp_path) == mode + path = tmp_path / ".postgres" / "state" + assert _mode(path) == 0o644, "宿主机上的 deploy.sh 直接读它" + assert "since=" in path.read_text() + path.write_text("mode=whatever\n") + assert ops.read_state(tmp_path) == "sqlite", "认不出的当没切过" + with pytest.raises(SystemExit): + ops.main(["state", "--data", str(tmp_path), "--set", "external"]) + + +def test_prune_keeps_the_newest_by_name(tmp_path): + for name in ("agentlab-20260101T000000Z.dump", "agentlab-20260103T000000Z.dump", "agentlab-20260102T000000Z.dump"): + (tmp_path / name).write_bytes(b"x") + (tmp_path / "20260101T000000Z").mkdir() + (tmp_path / "20260101T000000Z" / "agentlab.db").write_bytes(b"x") + (tmp_path / ".hidden").write_bytes(b"x") + gone = ops.prune(tmp_path, 2) + assert sorted(p.name for p in gone) == ["20260101T000000Z", "agentlab-20260101T000000Z.dump"] + assert sorted(p.name for p in tmp_path.iterdir()) == [".hidden", "agentlab-20260102T000000Z.dump", + "agentlab-20260103T000000Z.dump"] + assert ops.prune(tmp_path / "missing", 2) == [] + + +def test_backup_sqlite_copies_a_consistent_snapshot_including_the_wal(tmp_path, capsys): + db = sqlite3.connect(tmp_path / "agentlab.db") + db.execute("PRAGMA journal_mode=WAL") + db.execute("CREATE TABLE t (v TEXT)") + db.execute("INSERT INTO t VALUES ('还在 WAL 里')") + db.commit() + # 不关连接、不做检查点:新写的那行还只在 -wal 里 + assert (tmp_path / "agentlab.db-wal").exists() + sqlite3.connect(tmp_path / "checkpoints.db").close() + try: + assert ops.main(["backup-sqlite", "--data", str(tmp_path), "--keep", "2"]) == 0 + finally: + db.close() + target = tmp_path / "backups" / "sqlite" + (stamp,) = list(target.iterdir()) + assert capsys.readouterr().out.strip() == str(stamp) + copy = sqlite3.connect(stamp / "agentlab.db") + assert copy.execute("SELECT v FROM t").fetchall() == [("还在 WAL 里",)] + copy.close() + assert (stamp / "checkpoints.db").exists() + # 只留最近 keep 份 + for name in ("20200101T000000Z", "20200102T000000Z"): + (target / name).mkdir() + ops.backup_sqlite(tmp_path, keep=2) + assert len(list(target.iterdir())) == 2 and not (target / "20200101T000000Z").exists() + + +def test_backup_sqlite_without_databases_does_nothing(tmp_path, capsys): + assert ops.main(["backup-sqlite", "--data", str(tmp_path)]) == 0 + assert "没有 SQLite 库" in capsys.readouterr().out + assert not (tmp_path / "backups").exists()