From fc8e337d0f9fc343ee67fc5a761b3e55bde47030 Mon Sep 17 00:00:00 2001 From: openqj Date: Tue, 27 Oct 2026 00:05:18 +0800 Subject: [PATCH] =?UTF-8?q?HA=20=E9=98=B6=E6=AE=B5=201=EF=BC=9Aworktree=5F?= =?UTF-8?q?host=20+=20=E8=B7=A8=E4=B8=BB=E6=9C=BA=20/pr=20=E9=98=9F?= =?UTF-8?q?=E5=88=97=E5=8C=96=20(#50)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * 部署自动化：唯一发布链路 deploy.py + host_deploy.sh，sha/commit 双重验真，失败回滚 旧链路三次栽过的坑：通用 tar 路径导致旧包误部署、传输不验 sha、 主机脚本多份漂移、health 通过不等于新代码上线。本单把四点变硬门禁： - 产物唯一命名 codex-app-server-.tar.gz + .sha256，远端唯一路径 - git archive 产物 + DEPLOYED_COMMIT 清单，默认拒绝脏工作区 - 传输链 scp -> scp_run -> chunk 分片，每路传完远端验 sha，不符删掉重来 - 主机侧唯一脚本：先验 sha/清单，staging 解包 + 编译检查，换目录上线， work/ 原样保留，health + 上线 commit 不符即回滚并退出非零 - deploy.py 成功后独立 ssh 再读一次 live 清单二次核对 - CI：host-verify 加 deploy/tests；release.yml 只在手动/tag + production environment 审批时部署，主干推送绝不自动部署生产 - deploy.sh 标记为旧链路弃用 - health 的 commit 字段已在父提交 HA 阶段 0 里随 server.py 落地， 本分支整体包含它；commit 为空时只靠文件核对，门禁同样成立 测试：host 32、cloud 487（4 跳过）、deploy 9，全部通过； host_deploy.sh 仿真覆盖成功/坏 sha/清单不符/health commit 不符回滚。 生产实机部署未做，等确认后再用新链路推。 * HA stage 1: worktree_host + queued cross-host PR jobs --------- Co-authored-by: host --- .github/workflows/host-verify.yml | 1 + .github/workflows/release.yml | 95 +++++++++ .gitignore | 2 + cloud/autoscaler.py | 8 + cloud/db.py | 3 +- cloud/pr_exec.py | 233 +++++++++++++++++++++ cloud/server.py | 91 ++++----- cloud/tests/test_pr_jobs.py | 317 +++++++++++++++++++++++++++++ cloud/worker.py | 114 ++++++++++- deploy/README.md | 63 +++++- deploy/deploy.py | 325 ++++++++++++++++++++++++++++++ deploy/host_deploy.sh | 214 ++++++++++++++++++++ deploy/tests/test_deploy.py | 93 +++++++++ deploy/tests/test_host_deploy.py | 147 ++++++++++++++ 14 files changed, 1639 insertions(+), 67 deletions(-) create mode 100644 .github/workflows/release.yml create mode 100644 cloud/pr_exec.py create mode 100644 cloud/tests/test_pr_jobs.py create mode 100644 deploy/deploy.py create mode 100644 deploy/host_deploy.sh create mode 100644 deploy/tests/test_deploy.py create mode 100644 deploy/tests/test_host_deploy.py diff --git a/.github/workflows/host-verify.yml b/.github/workflows/host-verify.yml index a5d8e2e..199dd89 100644 --- a/.github/workflows/host-verify.yml +++ b/.github/workflows/host-verify.yml @@ -13,3 +13,4 @@ jobs: python-version: "3.12" - run: python3 -m unittest discover -s tests -v - run: python3 -m unittest discover -s cloud/tests -v + - run: python3 -m unittest discover -s deploy/tests -v diff --git a/.github/workflows/release.yml b/.github/workflows/release.yml new file mode 100644 index 0000000..1c2928f --- /dev/null +++ b/.github/workflows/release.yml @@ -0,0 +1,95 @@ +name: release +# 发布链路：主干推送绝不自动部署生产。 +# 触发只有两种，两种都要经过 host-verify 同款测试： +# 1. 手动 workflow_dispatch（在 Actions 里点 Run，选 environment: production， +# GitHub 的 environment 审批人要点批准才会继续，见仓库 Settings > Environments）。 +# 2. 打 tag v*（例如 v0.2.0），同样进 production environment 审批。 +# 部署本身不由 Actions 直接拼 ssh 命令，而是复用 deploy/deploy.py： +# - 构建产物来自 git archive，含 DEPLOYED_COMMIT + sha256； +# - 传输、远端验 sha、主机 staging 换目录、失败回滚、RELEASE_OK 核对 +# 全在 deploy.py / host_deploy.sh 里，Actions 只负责“何时触发 + 审批”。 +# SSH 凭据只走 secrets，不写进日志：DEPLOY_HOST、DEPLOY_USER、 +# DEPLOY_SSH_KEY（私钥内容，Actions 里写 ~/.ssh/id_ed25519，0600）。 +# 没有配这些 secrets 时，build 作业仍产出可下载的发布包 artifact， +# deploy 作业会明确失败并提示“未配部署凭据，改走 VM 侧 deploy.py”， +# 绝不静默跳过部署又报成功。 +on: + workflow_dispatch: + inputs: + ref: + description: "要部署的 ref（默认 main 的 HEAD）" + required: false + default: "main" + push: + tags: + - "v*" +permissions: + contents: read +jobs: + build: + runs-on: ubuntu-latest + outputs: + commit: ${{ steps.build.outputs.commit }} + steps: + - uses: actions/checkout@v4 + with: + ref: ${{ inputs.ref || github.ref }} + - uses: actions/setup-python@v5 + with: + python-version: "3.12" + - name: 测试（与 host-verify 同款，失败则不产发布包） + run: | + python3 -m unittest discover -s tests -v + python3 -m unittest discover -s cloud/tests -v + python3 -m unittest discover -s deploy/tests -v + - name: 构建发布包（git archive + DEPLOYED_COMMIT + sha256） + id: build + run: | + python3 deploy/deploy.py --build-only --skip-tests \ + --ref "${{ inputs.ref || github.ref }}" \ + --out-dir "$GITHUB_WORKSPACE/release-artifact" + echo "commit=$(git rev-parse HEAD)" >> "$GITHUB_OUTPUT" + ls -lh release-artifact/ + - name: 上传发布包 artifact + uses: actions/upload-artifact@v4 + with: + name: codex-app-server-release + path: release-artifact/ + retention-days: 30 + deploy: + needs: build + runs-on: ubuntu-latest + environment: production + steps: + - uses: actions/checkout@v4 + with: + ref: ${{ inputs.ref || github.ref }} + - uses: actions/setup-python@v5 + with: + python-version: "3.12" + - name: 下载发布包 + uses: actions/download-artifact@v4 + with: + name: codex-app-server-release + path: release-artifact + - name: 检查部署凭据是否齐全 + run: | + if [ -z "${{ secrets.DEPLOY_HOST }}" ] || [ -z "${{ secrets.DEPLOY_SSH_KEY }}" ]; then + echo "未配 DEPLOY_HOST / DEPLOY_SSH_KEY，无法自动部署生产。" >&2 + echo "发布包已在 artifact 里，改走 VM 侧：python3 deploy/deploy.py --tarball <包> --skip-tests" >&2 + exit 1 + fi + - name: 写 SSH key（0600，不打印） + run: | + mkdir -p ~/.ssh + printf '%s\n' "${{ secrets.DEPLOY_SSH_KEY }}" > ~/.ssh/id_ed25519 + chmod 600 ~/.ssh/id_ed25519 + - name: 部署（deploy.py：传输验 sha + 主机换目录 + 回滚 + 双重核对） + env: + SSH_PROXY_CMD: "" + run: | + TARBALL=$(ls release-artifact/*.tar.gz | head -n 1) + DEPLOY_USER="${{ secrets.DEPLOY_USER }}" + python3 deploy/deploy.py "${{ secrets.DEPLOY_HOST }}" \ + "${DEPLOY_USER:-root}" \ + --tarball "$TARBALL" --skip-tests diff --git a/.gitignore b/.gitignore index 9d6d74e..e801aac 100644 --- a/.gitignore +++ b/.gitignore @@ -13,3 +13,5 @@ host-config-real.json host-state-real.json.bak* worker-heartbeat.json worker-heartbeat.json.tmp +DEPLOYED_COMMIT +RELEASE.json diff --git a/cloud/autoscaler.py b/cloud/autoscaler.py index f8e9de9..5a3a843 100644 --- a/cloud/autoscaler.py +++ b/cloud/autoscaler.py @@ -34,6 +34,7 @@ import json import math import os import signal +import socket import subprocess import sys import time @@ -160,6 +161,13 @@ class Autoscaler: cmd.extend(["--db", self.db_path]) cmd.extend(["--worker-id", worker_id, "--worker-capacity", str(self.per_worker_capacity)]) + # HA 阶段 1：宿主名强制追加（末尾胜出）。scaler 跑在宿主上， + # socket.gethostname() 就是宿主名；容器内 worker 若用自己的 + # hostname 会拿到容器名，worktree_host 就错了，所以这里必须 + # 显式传，不能让 worker 自己猜。 + if not any(a == "--host-name" or a.startswith("--host-name=") + for a in self.worker_args): + cmd.extend(["--host-name", socket.gethostname()]) return cmd def _start_worker(self, worker_id): diff --git a/cloud/db.py b/cloud/db.py index 1b75d74..ac8c3ed 100644 --- a/cloud/db.py +++ b/cloud/db.py @@ -911,7 +911,8 @@ def init_postgres_schema(conn): tcols = {row[1] for row in conn.execute( "PRAGMA table_info(tasks)").fetchall()} for _name, _type in (("executor", "TEXT"), ("pr_url", "TEXT"), - ("pr_number", "INTEGER")): + ("pr_number", "INTEGER"), + ("worktree_host", "TEXT")): if _name not in tcols: conn.execute( "ALTER TABLE tasks ADD COLUMN %s %s" % (_name, _type)) diff --git a/cloud/pr_exec.py b/cloud/pr_exec.py new file mode 100644 index 0000000..9715cf7 --- /dev/null +++ b/cloud/pr_exec.py @@ -0,0 +1,233 @@ +"""跨主机 PR 执行（HA 阶段 1）。 + +背景：`/pr` 原来只在 server 进程内同步跑，worktree 从 server 本机 +文件系统解析——server 与 worker 同主机是隐含前提。多机下任务的 +worktree 可能在另一台主机上，server 本机根本没有那个目录。 + +本模块提供两条路（与 server 的 `/pr` 端点、worker 的 pr 任务循环 +共用，逻辑只写一次）： + +- 同步路（worktree 在本机）：`execute_task_pr` 直接跑，与旧端点 + 行为一致（写回 pr_url/pr_number、记审计 pr_created、成功发 + review-requested、落盘失败发 merge-failed、跳过不发）。 +- 队列路（worktree 在别机）：`enqueue_pr_job` 把任务推进 `pr_jobs` + 表；持有 worktree 的主机上的 local worker 每轮用 CAS 认领并跑 + 同一个 `execute_task_pr`，结果同样写回任务行。server 与 worker + 之间不走新网络通道（不引 SSH 密钥分发），只复用已有的 DB 队列。 + +只有 local worker 跑 pr 任务：host-worktrees/ 与 tenants/ 在主机上 +都可见（server 今天就是这么给 Docker 任务做 /pr 的），而容器内 +worker 看不到 host-worktrees/，也不一定有 gh。Docker worker 跳过 +pr 任务循环。 +""" + +import logging +import time +import uuid + +from cloud.auth import AuthStore +from cloud.db import connect as _db_connect +from cloud.github_app import resolve_app_token +from cloud.pr import (PrSkipped, PrStepError, create_task_pr, + parse_git_remote_owner_repo) +from cloud.push import dispatch_event as _push_dispatch + +_log = logging.getLogger("cloud.pr_exec") + + +def ensure_pr_jobs_table(conn): + """幂等建 pr_jobs 表（workers/webhook_deliveries 同款懒建表模式）。 + + PG 适配层把 `?` 译成 `%s`、把 AUTOINCREMENT 译成 IDENTITY， + 因此同一条 DDL 在 SQLite 与 Postgres 下都能跑；调用方在 + enqueue/claim 前各调一次即可。 + """ + conn.execute( + "CREATE TABLE IF NOT EXISTS pr_jobs(" + "id TEXT PRIMARY KEY, task_id TEXT NOT NULL, " + "project_id TEXT NOT NULL, host TEXT NOT NULL, " + "status TEXT NOT NULL, pr_url TEXT, pr_number INTEGER, " + "error TEXT, created_at REAL, updated_at REAL, " + "started_at REAL, finished_at REAL)" + ) + conn.commit() + + +def find_open_pr_job(conn, task_id): + """找该任务未完成的 pr 任务（去重：同一任务同时只排一个）。""" + ensure_pr_jobs_table(conn) + row = conn.execute( + "SELECT * FROM pr_jobs WHERE task_id = ?" + " AND status IN ('queued', 'running')" + " ORDER BY created_at ASC LIMIT 1", + (task_id,)).fetchone() + if row is None: + return None + try: + return dict(row) + except TypeError: + return {"id": row[0], "task_id": row[1], "project_id": row[2], + "host": row[3], "status": row[4]} + + +def enqueue_pr_job(conn, task_id, project_id, host): + """排一个 pr 任务（调用方已做去重）。返回 job id。""" + ensure_pr_jobs_table(conn) + job_id = uuid.uuid4().hex + now = time.time() + conn.execute( + "INSERT INTO pr_jobs(id, task_id, project_id, host, status," + " created_at, updated_at) VALUES(?, ?, ?, ?, 'queued', ?, ?)", + (job_id, task_id, project_id, host, now, now)) + conn.commit() + return job_id + + +def claim_pr_job(conn, host): + """CAS 认领本主机的一个 queued pr 任务（最老优先）。 + + 两个同主机 worker 抢同一任务时只有一个能拿到 rowcount == 1。 + 返回任务 dict 或 None。 + """ + ensure_pr_jobs_table(conn) + row = conn.execute( + "SELECT * FROM pr_jobs WHERE host = ? AND status = 'queued'" + " ORDER BY created_at ASC LIMIT 1", + (host,)).fetchone() + if row is None: + return None + job_id = row["id"] if hasattr(row, "__getitem__") else row[0] + now = time.time() + cur = conn.execute( + "UPDATE pr_jobs SET status = 'running', started_at = ?," + " updated_at = ? WHERE id = ? AND status = 'queued'", + (now, now, job_id)) + conn.commit() + if cur.rowcount != 1: + return None + try: + return dict(row) + except TypeError: + return {"id": row[0], "task_id": row[1], "project_id": row[2], + "host": row[3], "status": "running"} + + +def finish_pr_job(conn, job_id, status, pr_url=None, pr_number=None, + error=None): + """写回 pr 任务终态（completed/failed）。""" + ensure_pr_jobs_table(conn) + now = time.time() + conn.execute( + "UPDATE pr_jobs SET status = ?, pr_url = ?, pr_number = ?," + " error = ?, finished_at = ?, updated_at = ? WHERE id = ?", + (status, pr_url, pr_number, error, now, now, job_id)) + conn.commit() + + +def _notify(project_id, event_type, payload, db_path): + """PR 事件推送（best-effort，与 server._notify_pr_event 同语义）。""" + try: + result = _push_dispatch(db_path, event_type, project_id, payload) + except Exception as exc: # noqa: BLE001 -- 通知失败绝不炸 PR + _log.warning("pr notify %s failed: %s", event_type, exc) + return + if result.get("errors"): + _log.warning("pr notify %s errors: %s", event_type, result["errors"]) + + +def execute_task_pr(db_path, task_id, user_id, worktrees_root=None, + git_runner=None, gh_runner=None, gh_which=None, + app_token_client=None, github_api=None): + """给已 merged 的任务建 PR（同步执行体，server 端点与 worker 共用）。 + + 步骤与旧 `_task_pr` 端点内联的逻辑一致：查项目 auto_pr/repo_url、 + 有 App 安装记录时现换 installation token 走 App 路径、否则走 + gh + PAT；成功写回任务 pr_url/pr_number、记审计 pr_created、 + 发 review-requested；PrStepError 时发 merge-failed 后原样抛出； + PrSkipped 时不发任何事件、原样抛出。调用方按旧端点语义映射 + 响应（PrSkipped -> 200 pr_skipped，PrStepError -> 502）。 + + user_id：审计 pr_created 的操作人；server 端点传调用方， + worker 的 pr 任务循环传 "system"（与 sweep 同款约定）。 + token 只在本次调用内存里出现，不进库、不进日志。 + """ + conn = _db_connect(str(db_path)) + try: + row = conn.execute( + "SELECT * FROM tasks WHERE id = ?", (task_id,)).fetchone() + if row is None: + raise PrSkipped("task_not_found") + task = dict(row) + prow = conn.execute( + "SELECT auto_pr, repo_url FROM projects WHERE id = ?", + (task["project_id"],)).fetchone() + finally: + conn.close() + if task["status"] != "merged": + raise PrSkipped("not_merged") + auto_pr = 0 + repo_url = None + if prow is not None: + try: + auto_pr = prow["auto_pr"] or 0 + except (KeyError, IndexError, TypeError): + auto_pr = 0 + try: + repo_url = prow["repo_url"] + except (KeyError, IndexError, TypeError): + repo_url = None + app_token = None + auth_via = "pat" + repo_full_name = None + if repo_url: + parsed = parse_git_remote_owner_repo(repo_url) + if parsed: + repo_full_name = "%s/%s" % parsed + if repo_full_name: + conn = _db_connect(str(db_path)) + try: + app_token = resolve_app_token( + conn, repo_full_name, http_client=app_token_client) + except Exception: + app_token = None + finally: + conn.close() + if app_token: + auth_via = "app" + try: + result = create_task_pr( + task_id, auto_pr, + worktrees_root=worktrees_root, + git_runner=git_runner, gh_runner=gh_runner, gh_which=gh_which, + app_token=app_token, github_api=github_api) + except PrSkipped: + raise + except PrStepError as exc: + _notify(task["project_id"], "merge-failed", + {"task_id": task_id, "step": exc.step, + "error": exc.message}, db_path) + raise + now = time.time() + conn = _db_connect(str(db_path)) + try: + conn.execute( + "UPDATE tasks SET pr_url = ?, pr_number = ?," + " updated_at = ? WHERE id = ?", + (result["pr_url"], result["pr_number"], now, task_id)) + conn.commit() + finally: + conn.close() + store = AuthStore(db_path) + try: + store.audit(user_id, task["project_id"], "pr_created", + result["pr_url"]) + finally: + try: + store.conn.close() + except Exception: + pass + _notify(task["project_id"], "review-requested", + {"task_id": task_id, "pr_url": result["pr_url"], + "pr_number": result["pr_number"]}, db_path) + return {"pr_url": result["pr_url"], "pr_number": result["pr_number"], + "branch": result["branch"], "auth_via": auth_via} diff --git a/cloud/server.py b/cloud/server.py index 29973d6..fab43d3 100644 --- a/cloud/server.py +++ b/cloud/server.py @@ -42,6 +42,7 @@ import html import json import logging import os +import socket import sqlite3 import threading import time @@ -61,8 +62,9 @@ from cloud.db import PostgresIntegrityError as _PgIntegrityError from cloud.db import PostgresOperationalError as _PgOperationalError from cloud.kb import KBStore from cloud.memory import MemoryStore -from cloud.pr import (PrSkipped, PrStepError, create_task_pr, - parse_git_remote_owner_repo) +from cloud.pr import PrSkipped, PrStepError +from cloud.pr_exec import (enqueue_pr_job, execute_task_pr, + find_open_pr_job) from cloud.workers import ( heartbeat_worker as _registry_heartbeat, list_workers as _registry_list, @@ -103,7 +105,6 @@ from cloud.github_app import ( record_installation_repos, remove_installation, remove_installation_repos, - resolve_app_token, save_app_config, set_fix_review_task_id, ) @@ -322,6 +323,9 @@ def migrate_tasks_table(conn): "goal": "TEXT", "verify_command": "TEXT", "goal_max_rounds": "INTEGER", + # HA 阶段 1：认领任务的 worker 所在主机名；/pr 据此判断 + # worktree 在本机（同步跑）还是别机（排 pr 任务给该主机）。 + "worktree_host": "TEXT", } for name, coltype in new_cols.items(): if name not in cols: @@ -456,6 +460,7 @@ def _task_to_dict(row): "goal": _row_get(row, "goal"), "verify_command": _row_get(row, "verify_command"), "goal_max_rounds": _row_get(row, "goal_max_rounds"), + "worktree_host": _row_get(row, "worktree_host"), } @@ -1638,6 +1643,10 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler): 成功时把 pr_url/pr_number 写回任务并记审计 pr_created。 推送：成功时发 review-requested，落盘失败时发 merge-failed （均为 best-effort，不影响 PR 响应；跳过时不发）。 + HA 阶段 1：任务的 worktree_host 与本机不同时，不直接跑， + 而是排 pr 任务给该主机（返回 200 + pr_queued，同一任务 + 未完成的任务去重），由该主机上的 local worker 执行； + worktree 在本机（或未记录）时同步跑，与旧行为一致。 App-token：项目 repo_url 能解析出 github.com 仓库名且 库里有该仓库的 App 安装记录时，现换 installation token 走 App 路径（响应里 auth_via=app），否则走原来的 gh + @@ -1660,51 +1669,36 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler): "status": task["status"], }) return - conn = self._db() - try: - prow = conn.execute( - "SELECT auto_pr, repo_url FROM projects WHERE id = ?", - (task["project_id"],)).fetchone() - finally: - conn.close() - auto_pr = 0 - repo_url = None - if prow is not None: - try: - auto_pr = prow["auto_pr"] or 0 - except (KeyError, IndexError, TypeError): - auto_pr = 0 - try: - repo_url = prow["repo_url"] - except (KeyError, IndexError, TypeError): - repo_url = None - app_token = None - auth_via = "pat" - repo_full_name = None - if repo_url: - parsed = parse_git_remote_owner_repo(repo_url) - if parsed: - repo_full_name = "%s/%s" % parsed - if repo_full_name: + # HA 阶段 1：worktree 在别机时不直接跑，排 pr 任务给该主机； + # 本机（或 worktree_host 未记录的老任务）走同步路。 + worktree_host = task.get("worktree_host") + if worktree_host and worktree_host != socket.gethostname(): conn = self._db() try: - app_token = resolve_app_token( - conn, repo_full_name, - http_client=getattr( - self.server, "pr_app_token_client", None)) + existing = find_open_pr_job(conn, task_id) + if existing is not None: + job_id = existing["id"] + else: + job_id = enqueue_pr_job( + conn, task_id, task["project_id"], worktree_host) finally: conn.close() - if app_token: - auth_via = "app" + self._send_json(200, { + "task_id": task_id, "status": task["status"], + "pr_queued": True, "pr_job_id": job_id, + "worktree_host": worktree_host, + }) + return try: - result = create_task_pr( - task_id, auto_pr, + result = execute_task_pr( + self.server.db_path, task_id, user_id, worktrees_root=getattr( self.server, "host_worktrees_dir", None), git_runner=getattr(self.server, "pr_git_runner", None), gh_runner=getattr(self.server, "pr_gh_runner", None), gh_which=getattr(self.server, "pr_gh_which", None), - app_token=app_token, + app_token_client=getattr( + self.server, "pr_app_token_client", None), github_api=getattr(self.server, "pr_github_api", None)) except PrSkipped as exc: self._send_json(200, { @@ -1713,41 +1707,24 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler): }) return except PrStepError as exc: - self._notify_pr_event( - task["project_id"], "merge-failed", - {"task_id": task_id, "step": exc.step, - "error": exc.message}) + # merge-failed 事件已在 execute_task_pr 内发出，这里只映射响应。 self._send_json(502, { "error": exc.message, "step": exc.step, "task_id": task_id, "status": task["status"], }) return - now = time.time() conn = self._db() try: - conn.execute( - "UPDATE tasks SET pr_url = ?, pr_number = ?," - " updated_at = ? WHERE id = ?", - (result["pr_url"], result["pr_number"], now, task_id), - ) - conn.commit() task = _task_to_dict( conn.execute( "SELECT * FROM tasks WHERE id = ?", (task_id,)).fetchone()) finally: conn.close() - with _auth_store(self.server.db_path) as store: - store.audit(user_id, task["project_id"], "pr_created", - result["pr_url"]) - self._notify_pr_event( - task["project_id"], "review-requested", - {"task_id": task_id, "pr_url": result["pr_url"], - "pr_number": result["pr_number"]}) self._send_json(200, dict(task, pr_url=result["pr_url"], pr_number=result["pr_number"], branch=result["branch"], - auth_via=auth_via)) + auth_via=result["auth_via"])) def _heartbeat_task(self, user_id, task_id): row = self._fetch_task(task_id) diff --git a/cloud/tests/test_pr_jobs.py b/cloud/tests/test_pr_jobs.py new file mode 100644 index 0000000..93274a6 --- /dev/null +++ b/cloud/tests/test_pr_jobs.py @@ -0,0 +1,317 @@ +"""HA 阶段 1 单测：pr 任务队列 + worktree_host + /pr 跨主机分支。 + +覆盖：pr_jobs 的 enqueue/claim(CAS)/finish 与去重、tasks 的 +worktree_host 迁移（server 与 worker 两支）、认领时写主机名、 +/pr 在 worktree_host 为别机时排队（不直接跑、不重复排）、 +local worker 的 pr 任务循环（CAS 认领 -> 执行 -> 写回终态）、 +autoscaler 给 worker 强制追加 --host-name。 + +git/GitHub 一律 fake 或 monkeypatch，不跑真实命令、不联网。 +""" + +import http.client +import json +import os +import socket +import sqlite3 +import sys +import tempfile +import threading +import unittest +from pathlib import Path +from types import SimpleNamespace + +try: + from cloud.auth import AuthStore + from cloud.autoscaler import Autoscaler + from cloud.db import connect as _db_connect + from cloud.pr_exec import (claim_pr_job, enqueue_pr_job, + find_open_pr_job, finish_pr_job) + from cloud.server import create_server, migrate_tasks_table + from cloud.worker import Worker +except ImportError: + sys.path.insert(0, str(Path(__file__).resolve().parents[2])) + from cloud.auth import AuthStore + from cloud.autoscaler import Autoscaler + from cloud.db import connect as _db_connect + from cloud.pr_exec import (claim_pr_job, enqueue_pr_job, + ensure_pr_jobs_table, find_open_pr_job, + finish_pr_job) + from cloud.server import create_server, migrate_tasks_table + from cloud.worker import Worker + + +def _old_tasks_table(db_path): + conn = sqlite3.connect(db_path) + try: + conn.execute( + "CREATE TABLE tasks(id TEXT PRIMARY KEY, project_id TEXT," + " prompt TEXT, status TEXT, depends_on TEXT, priority INTEGER," + " created_by TEXT, created_at REAL, updated_at REAL," + " last_heartbeat REAL)") + conn.commit() + finally: + conn.close() + + +def _merged_task(db_path, task_id, project_id, host=None): + conn = sqlite3.connect(db_path) + try: + conn.execute( + "INSERT INTO tasks(id, project_id, prompt, status," + " depends_on, priority, created_by, created_at," + " updated_at, last_heartbeat) VALUES(?, ?, ?, 'merged'," + " '[]', 0, 'system', 0, 0, NULL)", + (task_id, project_id, "do it")) + if host is not None: + conn.execute("UPDATE tasks SET worktree_host = ? WHERE id = ?", + (host, task_id)) + conn.commit() + finally: + conn.close() + + +class PrJobsTableTestCase(unittest.TestCase): + def setUp(self): + self.tmp = tempfile.TemporaryDirectory() + self.db_path = os.path.join(self.tmp.name, "control.db") + + def tearDown(self): + self.tmp.cleanup() + + def test_enqueue_claim_finish_and_cas(self): + conn = _db_connect(self.db_path) + try: + job1 = enqueue_pr_job(conn, "t1", "p1", "host-a") + job2 = enqueue_pr_job(conn, "t2", "p1", "host-a") + self.assertIsNotNone(find_open_pr_job(conn, "t1")) + claimed = claim_pr_job(conn, "host-a") + self.assertEqual(claimed["task_id"], "t1") + # 第二次认领拿不到同一任务（CAS），只能拿到下一个。 + claimed2 = claim_pr_job(conn, "host-a") + self.assertEqual(claimed2["task_id"], "t2") + self.assertIsNone(claim_pr_job(conn, "host-a")) + # 别的主机认领不到 host-a 的任务。 + self.assertIsNone(claim_pr_job(conn, "host-b")) + finish_pr_job(conn, job1, "completed", + "https://github.com/o/a/pull/1", 1) + self.assertIsNone(find_open_pr_job(conn, "t1")) + row = conn.execute( + "SELECT status, pr_url, pr_number FROM pr_jobs" + " WHERE id = ?", (job1,)).fetchone() + self.assertEqual(row[0], "completed") + self.assertEqual(row[1], "https://github.com/o/a/pull/1") + self.assertEqual(row[2], 1) + finally: + conn.close() + + +class WorktreeHostMigrationTestCase(unittest.TestCase): + def setUp(self): + self.tmp = tempfile.TemporaryDirectory() + self.db_path = os.path.join(self.tmp.name, "control.db") + _old_tasks_table(self.db_path) + + def tearDown(self): + self.tmp.cleanup() + + def test_server_migrate_adds_worktree_host(self): + conn = _db_connect(self.db_path) + try: + migrate_tasks_table(conn) + cols = {row[1] for row in conn.execute( + "PRAGMA table_info(tasks)").fetchall()} + self.assertIn("worktree_host", cols) + # 再跑一次是 no-op。 + migrate_tasks_table(conn) + finally: + conn.close() + + def test_worker_ensure_adds_worktree_host(self): + worker = Worker(self.db_path, "/nonexistent/host.py", + env_key="TEST_KEY") + conn = _db_connect(self.db_path) + try: + worker._ensure_result_columns(conn) + cols = {row[1] for row in conn.execute( + "PRAGMA table_info(tasks)").fetchall()} + self.assertIn("worktree_host", cols) + finally: + conn.close() + + def test_claim_writes_host_name(self): + worker = Worker(self.db_path, "/nonexistent/host.py", + env_key="TEST_KEY", host_name="worker-host-9") + conn = _db_connect(self.db_path) + try: + conn.execute( + "INSERT INTO tasks(id, project_id, prompt, status," + " depends_on, priority, created_by, created_at," + " updated_at, last_heartbeat) VALUES('t1', 'p1', 'x'," + " 'queued', '[]', 0, 'system', 0, 0, NULL)") + conn.commit() + finally: + conn.close() + self.assertTrue(worker._claim("t1", executor="local")) + conn = _db_connect(self.db_path) + try: + row = conn.execute( + "SELECT status, worktree_host FROM tasks WHERE id = 't1'" + ).fetchone() + self.assertEqual(row[0], "running") + self.assertEqual(row[1], "worker-host-9") + finally: + conn.close() + + +class PrEndpointRemoteTestCase(unittest.TestCase): + def setUp(self): + self.tmp = tempfile.TemporaryDirectory() + self.db_path = os.path.join(self.tmp.name, "control.db") + self.auth = AuthStore(self.db_path) + self.owner_id, self.owner_key = self.auth.create_user("owner") + self.server = create_server(self.db_path, "127.0.0.1", 0) + self.port = self.server.server_address[1] + self.thread = threading.Thread( + target=self.server.serve_forever, daemon=True) + self.thread.start() + + def tearDown(self): + self.server.shutdown() + self.server.server_close() + self.thread.join(timeout=5) + try: + self.auth.conn.close() + except Exception: + pass + self.tmp.cleanup() + + def _req(self, method, path, body=None, key=None): + conn = http.client.HTTPConnection( + "127.0.0.1", self.port, timeout=10) + hdrs = {} + payload = None + if body is not None: + payload = json.dumps(body).encode("utf-8") + hdrs["Content-Type"] = "application/json" + if key is not None: + hdrs["X-API-Key"] = key + try: + conn.request(method, path, body=payload, headers=hdrs) + resp = conn.getresponse() + data = resp.read() + finally: + conn.close() + try: + parsed = json.loads(data.decode("utf-8")) if data else None + except ValueError: + parsed = None + return resp.status, parsed + + def test_remote_host_queues_and_dedupes(self): + status, project = self._req( + "POST", "/projects", {"name": "p", "auto_pr": 1}, + self.owner_key) + self.assertEqual(status, 201, project) + # 项目成员默认 owner 即 maintainer（create 时建成员行）。 + _merged_task(self.db_path, "t-remote", project["id"], + host="elsewhere-host-1") + status, body = self._req( + "POST", "/tasks/t-remote/pr", {}, self.owner_key) + self.assertEqual(status, 200, body) + self.assertTrue(body.get("pr_queued")) + self.assertEqual(body.get("worktree_host"), "elsewhere-host-1") + job_id = body.get("pr_job_id") + self.assertIsNotNone(job_id) + # 再调一次：去重，返回同一个 job，不重复排队。 + status, body2 = self._req( + "POST", "/tasks/t-remote/pr", {}, self.owner_key) + self.assertEqual(status, 200, body2) + self.assertEqual(body2.get("pr_job_id"), job_id) + conn = _db_connect(self.db_path) + try: + rows = conn.execute( + "SELECT COUNT(*) FROM pr_jobs WHERE task_id = 't-remote'" + ).fetchone() + self.assertEqual(rows[0], 1) + finally: + conn.close() + + +class WorkerPrJobsTestCase(unittest.TestCase): + def setUp(self): + self.tmp = tempfile.TemporaryDirectory() + self.db_path = os.path.join(self.tmp.name, "control.db") + _old_tasks_table(self.db_path) + + def tearDown(self): + self.tmp.cleanup() + + def test_local_worker_runs_pr_job(self): + import cloud.worker as worker_mod + calls = [] + + def _fake_execute(db_path, task_id, user_id, **kwargs): + calls.append((task_id, user_id)) + return {"pr_url": "https://github.com/o/a/pull/9", + "pr_number": 9, "branch": "host/t1", + "auth_via": "pat"} + + worker = Worker(self.db_path, "/nonexistent/host.py", + env_key="TEST_KEY", host_name="elsewhere-host-1") + conn = _db_connect(self.db_path) + try: + worker._ensure_result_columns(conn) + conn.execute( + "INSERT INTO tasks(id, project_id, prompt, status," + " depends_on, priority, created_by, created_at," + " updated_at, last_heartbeat) VALUES('t1', 'p1', 'x'," + " 'merged', '[]', 0, 'system', 0, 0, NULL)") + conn.commit() + enqueue_pr_job(conn, "t1", "p1", "elsewhere-host-1") + enqueue_pr_job(conn, "t2", "p1", "other-host") + finally: + conn.close() + orig = worker_mod.execute_task_pr + worker_mod.execute_task_pr = _fake_execute + try: + results = worker._run_pr_jobs() + finally: + worker_mod.execute_task_pr = orig + self.assertEqual(len(results), 1) + self.assertEqual(results[0]["task_id"], "t1") + self.assertEqual(results[0]["status"], "completed") + self.assertEqual(calls, [("t1", "system")]) + conn = _db_connect(self.db_path) + try: + row = conn.execute( + "SELECT status, pr_url FROM pr_jobs WHERE task_id = 't1'" + ).fetchone() + self.assertEqual(row[0], "completed") + self.assertEqual(row[1], "https://github.com/o/a/pull/9") + # 别的主机的任务没被碰。 + row2 = conn.execute( + "SELECT status FROM pr_jobs WHERE task_id = 't2'" + ).fetchone() + self.assertEqual(row2[0], "queued") + finally: + conn.close() + + def test_docker_worker_skips_pr_jobs(self): + from cloud.sandbox import DockerExecutor + worker = Worker(self.db_path, "/nonexistent/host.py", + env_key="TEST_KEY", + executor=DockerExecutor("/fake/host.py")) + conn = _db_connect(self.db_path) + try: + enqueue_pr_job(conn, "t1", "p1", worker.host_name) + finally: + conn.close() + self.assertEqual(worker._run_pr_jobs(), []) + + def test_autoscaler_forces_host_name(self): + scaler = Autoscaler(self.db_path, []) + cmd = scaler._worker_command("w1") + self.assertIn("--host-name", cmd) + self.assertEqual(cmd[cmd.index("--host-name") + 1], + socket.gethostname()) diff --git a/cloud/worker.py b/cloud/worker.py index e18a0fd..fdab7d9 100644 --- a/cloud/worker.py +++ b/cloud/worker.py @@ -129,6 +129,8 @@ from cloud.cost import CostStore, estimate_tokens, price_usd_micros, sum_usage from cloud.db import connect as _db_connect from cloud.logging_config import get_logger, log_event, setup_logging from cloud.models import ModelRouter, is_fallbackable, record_fallback +from cloud.pr import PrSkipped, PrStepError +from cloud.pr_exec import (claim_pr_job, execute_task_pr, finish_pr_job) from cloud.queue import budget_exceeded, ready_tasks from cloud.repos import RepoError, resolve_repo from cloud.sandbox import CloudExecutor, DockerExecutor, LocalExecutor @@ -245,7 +247,8 @@ class Worker: worker_id=None, capacity=1, fallback_models=None, tenant_isolation=False, tenants_root=None, tenant_template_home=None, tenant_template_repo=None, - egress_proxy_port=0, claim_executor_only=False): + egress_proxy_port=0, claim_executor_only=False, + host_name=None, pr_jobs_only=False): self.db_path = str(db_path) self.host_py_path = host_py_path self.config_path = config_path @@ -338,6 +341,15 @@ class Worker: # 容器的 NULL 任务（那些任务在宿主上跑就丢了 Docker # 外层，违背“按任务二选一”的路由意图）。 self.claim_executor_only = bool(claim_executor_only) + # HA 阶段 1：本 worker 所在主机名。认领任务时写进 + # tasks.worktree_host，/pr 据此判断 worktree 在哪台主机。 + # 容器内 worker 的 socket.gethostname() 是容器名、不是宿主名， + # 所以 autoscaler 必须显式传 --host-name（宿主名），见 + # autoscaler._worker_command；单机/测试默认本机名即可。 + self.host_name = host_name or socket.gethostname() + # 只跑 pr 任务（HA 阶段 1 的跨主机 /pr 演练与专用 pr-runner + # 用）：不认领普通任务，只处理本主机的 pr_jobs。 + self.pr_jobs_only = bool(pr_jobs_only) # 运维心跳：路径优先显式参数，其次环境变量，默认当前目录下 # 的 worker-heartbeat.json；主机路径不在代码里硬编码，主机 # 部署时用 --heartbeat-file 或环境变量传入（见 deploy/README）。 @@ -467,25 +479,28 @@ class Worker: if executor is None: cur = conn.execute( "UPDATE tasks SET status = 'running'," - " last_heartbeat = ?, updated_at = ?" + " last_heartbeat = ?, updated_at = ?," + " worktree_host = ?" " WHERE id = ? AND status = 'queued'", - (now, now, task_id), + (now, now, self.host_name, task_id), ) elif self.claim_executor_only: cur = conn.execute( "UPDATE tasks SET status = 'running'," - " last_heartbeat = ?, updated_at = ?" + " last_heartbeat = ?, updated_at = ?," + " worktree_host = ?" " WHERE id = ? AND status = 'queued'" " AND executor = ?", - (now, now, task_id, executor), + (now, now, self.host_name, task_id, executor), ) else: cur = conn.execute( "UPDATE tasks SET status = 'running'," - " last_heartbeat = ?, updated_at = ?" + " last_heartbeat = ?, updated_at = ?," + " worktree_host = ?" " WHERE id = ? AND status = 'queued'" " AND (executor IS NULL OR executor = ?)", - (now, now, task_id, executor), + (now, now, self.host_name, task_id, executor), ) conn.commit() return cur.rowcount == 1 @@ -523,7 +538,8 @@ class Worker: "executor": "TEXT", "pr_url": "TEXT", "pr_number": "INTEGER", "goal": "TEXT", "verify_command": "TEXT", - "goal_max_rounds": "INTEGER"}.items(): + "goal_max_rounds": "INTEGER", + "worktree_host": "TEXT"}.items(): if name not in cols: try: conn.execute( @@ -1651,6 +1667,73 @@ class Worker: "actual_model": actual_model, "cost_record_id": cost_id} + def _run_pr_jobs(self): + """跑本主机的 pr 任务（HA 阶段 1），返回处理结果列表。 + + 只有 local worker 跑：host-worktrees/ 与 tenants/ 在主机上 + 都可见，容器内 worker 看不到 host-worktrees/。每轮把本主机 + 的 queued 任务按 CAS 认领跑完（pr 任务稀少，串行即可）； + execute_task_pr 内部已写回任务 pr_url/pr_number、记审计、 + 发 review-requested/merge-failed，这里只写回任务终态。 + 跳过（PrSkipped，如任务已不是 merged）与落盘失败都记 + failed + error，任务本身保持 merged、可重调 /pr 再排队。 + """ + if self._executor_name() != "local": + return [] + results = [] + while True: + conn = _db_connect(self.db_path) + try: + job = claim_pr_job(conn, self.host_name) + finally: + conn.close() + if job is None: + break + job_id = job["id"] + task_id = job["task_id"] + try: + result = execute_task_pr( + self.db_path, task_id, "system") + conn = _db_connect(self.db_path) + try: + finish_pr_job(conn, job_id, "completed", + result["pr_url"], result["pr_number"]) + finally: + conn.close() + results.append({"pr_job_id": job_id, "task_id": task_id, + "status": "completed", + "pr_url": result["pr_url"]}) + log_event(self.logger, "pr_job_completed", + task_id=task_id, status="completed") + except PrSkipped as exc: + err = "skipped: %s" % exc.reason + conn = _db_connect(self.db_path) + try: + finish_pr_job(conn, job_id, "failed", error=err) + finally: + conn.close() + results.append({"pr_job_id": job_id, "task_id": task_id, + "status": "failed", "error": err}) + except PrStepError as exc: + err = "%s: %s" % (exc.step, exc.message) + conn = _db_connect(self.db_path) + try: + finish_pr_job(conn, job_id, "failed", error=err) + finally: + conn.close() + results.append({"pr_job_id": job_id, "task_id": task_id, + "status": "failed", "error": err}) + except Exception as exc: # noqa: BLE001 -- 任务失败不能炸循环 + err = str(exc) + conn = _db_connect(self.db_path) + try: + finish_pr_job(conn, job_id, "failed", error=err) + finally: + conn.close() + results.append({"pr_job_id": job_id, "task_id": task_id, + "status": "failed", "error": err}) + return results + def run_all(self, max_tasks=None): """循环认领执行，直到无就绪任务、预算超限或达到 max_tasks。 @@ -1668,6 +1751,11 @@ class Worker: 停机信号（_loop_stop/stop_event）只阻止新认领，不会 丢弃已提交的任务；排空等待由 run_loop 的 finally 负责。 """ + # pr 任务先跑（HA 阶段 1）：每轮先把本主机的 queued pr 任务 + # 跑完再认领普通任务；pr_jobs_only 时只跑 pr 任务。 + self._run_pr_jobs() + if self.pr_jobs_only: + return [] if self.capacity <= 1: results = [] while max_tasks is None or len(results) < max_tasks: @@ -1960,6 +2048,14 @@ def main(argv=None): help="worker 并发上限：同一 worker 内最多同时跑" "几个任务（默认 1，即旧串行行为；>1 时用" "线程池并行，认领仍是 CAS，成本写库持锁）") + parser.add_argument("--host-name", default=None, + help="本 worker 所在主机名（HA 阶段 1）：认领任务" + "时写进 tasks.worktree_host，/pr 据此判断" + "worktree 在哪台主机；容器内 worker 必须由" + "autoscaler 显式传宿主名，不能用容器名") + parser.add_argument("--pr-jobs-only", action="store_true", + help="只跑 pr 任务（HA 阶段 1 的跨主机 /pr 演练" + "与专用 pr-runner 用）：不认领普通任务") parser.add_argument("--docker-image", default="codex-app-server:latest") parser.add_argument("--docker-cpus", default="2") parser.add_argument("--docker-memory", default="2g") @@ -2040,6 +2136,8 @@ def main(argv=None): heartbeat_file=args.heartbeat_file, worker_id=args.worker_id, capacity=args.worker_capacity, + host_name=args.host_name, + pr_jobs_only=args.pr_jobs_only, tenant_isolation=args.tenant_isolation, tenants_root=args.tenants_root, tenant_template_home=args.tenant_template_home, diff --git a/deploy/README.md b/deploy/README.md index 709a8f0..01ca454 100644 --- a/deploy/README.md +++ b/deploy/README.md @@ -4,7 +4,68 @@ 传代码、备份、重启、验证，不改任务逻辑；运维脚本只负责检查与演练， 不跑模型、不推代码。 -## 部署：deploy.sh +## 发布链路（新，默认用这个）：deploy.py + host_deploy.sh + +VM 侧运行，一条命令走完构建、传输、验真、部署、回滚核对： + +```sh +python3 deploy/deploy.py [user] +# 干跑（只构建、不传输、不动主机）： +python3 deploy/deploy.py --dry-run +# 只构建发布包（给 CI artifact / 手动检查用）： +python3 deploy/deploy.py --build-only --out-dir work/external-install/releases +# 部署已有发布包（CI release 作业就是这么调的）： +python3 deploy/deploy.py --tarball <包> --skip-tests +``` + +它修掉的正是旧手工链路三次栽过的坑： + +- **产物唯一命名，旧包无法冒充新包。** 发布包是 + `codex-app-server-.tar.gz` + `.sha256`，远端路径是 + `/tmp/codex-app-server-releases/.tar.gz`，不再用 + `/tmp/codex-app-server-code.tar.gz` 这种通用路径。scp 断了、 + 旧包残留，都不可能被下一步误当成新包。 +- **传输、主机、二次核对三道 sha/commit 门。** 传输链自动 + fallback：scp（SSH_PROXY_CMD）→ ssh-helpers/scp_run.py（此前 + 实测最稳）→ ssh-helpers/chunk_send.py（base64 分片）；每路 + 传完都在远端跑 `sha256sum` 比对，不符就删掉该次产物再试 + 下一路，绝不把错包交给主机脚本。主机侧只用 + `deploy/host_deploy.sh` 这一份（VM 侧随包一起传过去，不依赖 + 主机旧代码里的版本），它先验 tar sha、再验包内 + `DEPLOYED_COMMIT`、再 staging 解包 + 编译检查，任何一步 + 不符都直接退出、旧代码原地不动。 +- **换目录代替原地 rm + 解包。** 主机侧先停服务，再把 live + 换成 old、staging 换成 live，`work/` 与 + `host-config-real.json`（如有）原样搬回；中途失败不会留 + 半解包的 live。备份名含旧 commit + 时间戳，留最近 5 份。 +- **health 通过不再等于上线。** 主机脚本要求 health + `ok=true` 且 `backend=postgres` 且 `db_ok=true`，且 live 的 + `DEPLOYED_COMMIT` 等于期望 commit；deploy.py 成功后还要 + 独立 ssh 再读一次 live 清单核对，两处都对上才打印成功。 + 旧链路“health 通过但跑的是旧代码”在新链路里会触发回滚 + 并退出非零。health 的 `commit` 字段已在父提交 HA 阶段 0 + 里随 server.py 一起落地（`_deployed_commit`，无文件时为 + null）；本单不依赖它也能成立，commit 为空时只靠文件核对。 +- **产物来自 git archive，默认拒绝脏工作区。** 部署的永远是 + 已提交的 HEAD，不是工作区里没提交的临时改动；`--allow-dirty` + 也只告警，部署的仍是 HEAD，不会把脏文件偷偷带上去。 +- **CI 发布与手工发布同一套脚本。** `.github/workflows/release.yml` + 只在手动 workflow_dispatch 或打 `v*` tag 时触发，走 + `environment: production` 审批；主干推送绝不自动部署生产。 + Actions 里没配 `DEPLOY_HOST` / `DEPLOY_SSH_KEY` 时，build + 作业仍产出发布包 artifact，deploy 作业明确失败并提示改走 + VM 侧 deploy.py，绝不静默跳过又报成功。 + +多机滚动（HA 阶段 2 的前置）：当前 deploy.py 一次只推一台 +主机，多机时按主机逐台跑、每台 health + commit 都通过再推 +下一台，第一台失败就停，不要并行推两台。`--hosts` 批量滚动 +是后续一单，不在本轮做。 + +密钥规则与旧脚本一致：deploy.py 绝不接受密码参数、不读 +env 文件、不打印密钥；SSH 走 key，或调用方已有的 +ssh-helpers pty 通道（密码只走 `SSH_PASSWORD` 环境变量）。 + +## 部署：deploy.sh（旧链路，已弃用，新发布不要再用） VM 侧运行： diff --git a/deploy/deploy.py b/deploy/deploy.py new file mode 100644 index 0000000..0e2bca1 --- /dev/null +++ b/deploy/deploy.py @@ -0,0 +1,325 @@ +#!/usr/bin/env python3 +"""统一发布入口：取代手工的 deploy.sh + scp_run + chunk_send + grep 验真。 + +为什么需要它（此前三次栽过的坑，见 deploy/README.md）： + 1. 通用 tar 路径 /tmp/codex-app-server-code.tar.gz：一次 scp 失败， + 旧包还躺在原路径，下一步手工跑主机脚本就把旧代码“部署成功”了。 + 2. 传输不验 sha：scp 断在半路、chunk 首片写反，主机都可能拿到错包。 + 3. 主机脚本多份：deploy.sh 内嵌版、deploy_*_host.sh 各自演进，行为漂移。 + 4. health 通过不等于新代码上线：旧代码的 health 照样 ok=true。 + +本脚本把以上四点变成硬门禁，任何一步不符都退出非零、绝不打印成功： + - 产物唯一命名：codex-app-server-.tar.gz + .sha256， + 远端路径 /tmp/codex-app-server-releases/.tar.gz，旧包无法冒充新包。 + - 产物来自 git archive（已提交的 HEAD），默认拒绝工作区脏文件； + tar 根目录带 DEPLOYED_COMMIT 清单，主机侧先验清单再动代码。 + - 传输链自动 fallback：scp（SSH_PROXY_CMD）-> ssh-helpers/scp_run.py + -> ssh-helpers/chunk_send.py（base64 分片），每种方式传完都先在远端 + 跑 sha256sum 比对，不符就删掉该次产物再试下一种，绝不把错包交给主机脚本。 + - 主机侧只用 deploy/host_deploy.sh 这一份（VM 侧随包一起传过去， + 不依赖主机旧代码里的版本），它做 staging 换目录 + 回滚。 + - 上线证据双重核对：主机脚本打印的 RELEASE_OK 里的 commit， + 再加 VM 侧独立 ssh cat live 的 DEPLOYED_COMMIT，两者都必须等于期望 commit。 + +用法： + python3 deploy/deploy.py [user] [--ref HEAD] [--dry-run] + python3 deploy/deploy.py --build-only [--out-dir DIR] + python3 deploy/deploy.py --tarball --commit + python3 deploy/deploy.py --skip-tests # CI 里测试已跑过时用 + +密钥规则：本脚本绝不接受密码参数、不读任何 env 文件、不打印任何密钥。 +SSH 密码只由调用方已有的通道处理（key，或 ~/workspace/ssh-helpers 的 +pty helper，密码只走 SSH_PASSWORD 环境变量）；GitHub Actions 里走 key。 +""" +import argparse +import hashlib +import json +import os +import shutil +import subprocess +import sys +import tarfile +import tempfile +from datetime import datetime, timezone + +REPO_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) +HOST_SCRIPT = os.path.join(REPO_ROOT, "deploy", "host_deploy.sh") +HELPER_DIR = os.path.expanduser("~/workspace/ssh-helpers") +REMOTE_RELEASES_DIR = "/tmp/codex-app-server-releases" + + +def run(cmd, cwd=None, check=True): + """跑命令并返回 stdout；失败时把 stderr 透传后按 check 决定是否退出。""" + r = subprocess.run(cmd, cwd=cwd, capture_output=True, text=True) + if check and r.returncode != 0: + print("命令失败：%s" % " ".join(cmd), file=sys.stderr) + print((r.stdout + r.stderr)[-2000:], file=sys.stderr) + sys.exit(1) + return r + + +def git_commit(ref): + return run(["git", "rev-parse", ref], cwd=REPO_ROOT).stdout.strip() + + +def check_clean_tree(allow_dirty): + """已提交内容与工作区不一致时，默认拒绝；--allow-dirty 只告警， + 且部署的仍是 HEAD 的 archive，脏文件不会被偷偷带上去。""" + diff = run(["git", "diff", "--quiet"], cwd=REPO_ROOT, check=False) + cached = run(["git", "diff", "--cached", "--quiet"], cwd=REPO_ROOT, check=False) + if diff.returncode != 0 or cached.returncode != 0: + if not allow_dirty: + print("工作区有未提交改动，先提交再部署（或显式 --allow-dirty，" + "但部署的仍是 HEAD，不含脏文件）。", file=sys.stderr) + sys.exit(2) + print("警告：工作区有脏文件，部署的仍是 HEAD 的 archive，" + "脏文件不会上线。") + + +def sha256_of(path): + h = hashlib.sha256() + with open(path, "rb") as f: + for chunk in iter(lambda: f.read(1024 * 1024), b""): + h.update(chunk) + return h.hexdigest() + + +def release_tarball_name(commit): + return "codex-app-server-%s.tar.gz" % commit + + +def build_release(ref, out_dir): + """从 git archive 建发布包，返回 (tarball_path, sha256, commit)。""" + commit = git_commit(ref) + os.makedirs(out_dir, exist_ok=True) + tarball = os.path.join(out_dir, release_tarball_name(commit)) + with tempfile.TemporaryDirectory() as tmp: + # git archive 输出是二进制 tar 流，直接写文件，不经过文本解码。 + with open(os.path.join(tmp, "archive.tar"), "wb") as f: + r = subprocess.run(["git", "archive", ref], cwd=REPO_ROOT, stdout=f) + if r.returncode != 0: + print("git archive 失败：%s" % ref, file=sys.stderr) + sys.exit(1) + with tarfile.open(os.path.join(tmp, "archive.tar")) as tf: + tf.extractall(tmp, filter="data") + with open(os.path.join(tmp, "DEPLOYED_COMMIT"), "w") as f: + f.write(commit + "\n") + with open(os.path.join(tmp, "RELEASE.json"), "w") as f: + json.dump({ + "commit": commit, "ref": ref, + "built_at": datetime.now(timezone.utc).isoformat(), + }, f, indent=2, sort_keys=True) + f.write("\n") + with tarfile.open(tarball, "w:gz", format=tarfile.PAX_FORMAT) as tf: + for root, dirs, files in os.walk(tmp): + dirs.sort() + for name in sorted(files): + full = os.path.join(root, name) + arc = os.path.relpath(full, tmp) + if arc in ("archive.tar",): + continue + tf.add(full, arcname=arc) + sha = sha256_of(tarball) + with open(tarball + ".sha256", "w") as f: + f.write("%s %s\n" % (sha, os.path.basename(tarball))) + return tarball, sha, commit + + +def tarball_commit(tarball): + with tarfile.open(tarball, "r:gz") as tf: + try: + member = tf.getmember("DEPLOYED_COMMIT") + except KeyError: + print("发布包里没有 DEPLOYED_COMMIT，拒绝部署：%s" % tarball, + file=sys.stderr) + sys.exit(2) + data = tf.extractfile(member).read().decode().strip() + return data + + +def parse_release_ok(output, expected_commit): + """主机脚本成功输出里必须有 RELEASE_OK 且 commit 等于期望值。""" + for line in output.splitlines(): + if line.startswith("RELEASE_OK"): + fields = dict( + part.split("=", 1) for part in line.split()[1:] + if "=" in part) + return fields.get("commit") == expected_commit + return False + + +# ---------------- 传输：ssh / scp 抽象 ---------------- + +def helper_path(name): + path = os.path.join(HELPER_DIR, name) + return path if os.path.isfile(path) else None + + +def ssh_cmd(host, user, remote_cmd): + """返回一条 ssh 命令。优先用已验证稳定的 helper，否则用 + SSH_PROXY_CMD 环境变量，否则直连。绝不把密码拼进命令。""" + helper = helper_path("ssh_run.py") + if helper: + return ["python3", helper, "ssh", host, user, remote_cmd] + proxy = os.environ.get("SSH_PROXY_CMD") + base = ["ssh", "-o", "StrictHostKeyChecking=no", + "-o", "UserKnownHostsFile=/dev/null"] + if proxy: + base += ["-o", "ProxyCommand %s" % proxy] + return base + ["%s@%s" % (user, host), remote_cmd] + + +def scp_cmd(host, user, local, remote): + helper = helper_path("scp_run.py") + if helper: + # helper 自带 egress proxy + 密码 pty，是此前实测最稳的一路。 + return ["python3", helper, "scp", host, user, local, remote] + proxy = os.environ.get("SSH_PROXY_CMD") + base = ["scp", "-o", "StrictHostKeyChecking=no", + "-o", "UserKnownHostsFile=/dev/null"] + if proxy: + base += ["-o", "ProxyCommand %s" % proxy] + return base + [local, "%s@%s:%s" % (user, host, remote)] + + +def remote_sha(host, user, remote_path): + r = subprocess.run(ssh_cmd(host, user, "sha256sum %s" % remote_path), + capture_output=True, text=True) + if r.returncode != 0: + return None + return r.stdout.split()[0] if r.stdout.split() else None + + +def remote_rm(host, user, remote_path): + subprocess.run(ssh_cmd(host, user, "rm -f %s" % remote_path), + capture_output=True, text=True) + + +def transfer(host, user, local, remote, expected_sha): + """传输链：scp -> helper scp -> chunk 分片。每路传完都验远端 sha， + 不符删掉重来；全部失败才退出，绝不把错包交给主机脚本。""" + attempts = [] + # 1. scp（含 helper 路由，由 scp_cmd 内部决定） + attempts.append(("scp", scp_cmd(host, user, local, remote))) + chunk = helper_path("chunk_send.py") + if chunk and os.environ.get("SSH_PASSWORD"): + remote_b64 = remote + ".b64" + attempts.append(("chunk", None)) # 特殊处理，见下 + for name, cmd in attempts: + print("传输尝试：%s -> %s" % (name, remote)) + if name == "chunk": + r = subprocess.run( + ["python3", chunk, local, remote_b64, host, user], + capture_output=True, text=True) + if r.returncode != 0: + print("chunk 传输失败，换下一路。", file=sys.stderr) + continue + r = subprocess.run( + ssh_cmd(host, user, + "base64 -d %s > %s && sha256sum %s" + % (remote_b64, remote, remote)), + capture_output=True, text=True) + if r.returncode != 0: + print("chunk 解码失败，换下一路。", file=sys.stderr) + continue + else: + r = subprocess.run(cmd, capture_output=True, text=True) + if r.returncode != 0: + print("scp 失败（%s），换下一路。" % (r.stderr or "").strip()[-200:], + file=sys.stderr) + continue + actual = remote_sha(host, user, remote) + if actual == expected_sha: + print("传输验真通过：%s sha=%s" % (name, actual)) + return + print("远端 sha 不符（expected=%s actual=%s），删掉该次产物，换下一路。" + % (expected_sha, actual), file=sys.stderr) + remote_rm(host, user, remote) + print("所有传输方式都失败或验真不通过，部署未开始（旧代码不动）。", + file=sys.stderr) + sys.exit(1) + + +def run_tests(): + for suite in ("tests", "cloud/tests"): + run(["python3", "-m", "unittest", "discover", "-s", suite], + cwd=REPO_ROOT) + + +def main(argv=None): + parser = argparse.ArgumentParser(description="codex-app-server 统一发布入口") + parser.add_argument("host", nargs="?", help="部署目标主机") + parser.add_argument("user", nargs="?", default="root", help="SSH 用户") + parser.add_argument("--ref", default="HEAD") + parser.add_argument("--tarball", help="直接部署已有发布包（CI 产物）") + parser.add_argument("--commit", help="配合 --tarball 指定期望 commit") + parser.add_argument("--out-dir", default=os.path.join( + REPO_ROOT, "work", "external-install", "releases")) + parser.add_argument("--allow-dirty", action="store_true") + parser.add_argument("--skip-tests", action="store_true") + parser.add_argument("--build-only", action="store_true") + parser.add_argument("--dry-run", action="store_true") + args = parser.parse_args(argv) + + if args.tarball: + commit = args.commit or tarball_commit(args.tarball) + tarball, sha = args.tarball, sha256_of(args.tarball) + if tarball_commit(tarball) != commit: + print("发布包内 DEPLOYED_COMMIT 与 --commit 不符，拒绝部署。", + file=sys.stderr) + sys.exit(2) + else: + if not args.build_only and not args.dry_run and not args.host: + parser.error("需要 host，或用 --build-only / --dry-run") + check_clean_tree(args.allow_dirty) + if not args.skip_tests: + run_tests() + tarball, sha, commit = build_release(args.ref, args.out_dir) + print("发布包：%s" % tarball) + print("commit=%s sha256=%s" % (commit, sha)) + + if args.build_only: + print("仅构建，不传输、不部署。") + return 0 + if args.dry_run: + print("[dry-run] 将传输 %s 到 %s@%s:%s/%s（唯一路径，传完验 sha）" + % (tarball, args.user, args.host, REMOTE_RELEASES_DIR, + os.path.basename(tarball))) + print("[dry-run] 将传主机脚本 %s 并执行：" + "sh host_deploy.sh ，" + "成功须打印 RELEASE_OK 且 commit 一致，否则回滚。" % HOST_SCRIPT) + return 0 + + host, user = args.host, args.user + remote_tar = "%s/%s" % (REMOTE_RELEASES_DIR, os.path.basename(tarball)) + remote_script = "%s/host_deploy-%s.sh" % (REMOTE_RELEASES_DIR, commit) + run(ssh_cmd(host, user, "mkdir -p %s" % REMOTE_RELEASES_DIR), check=False) + transfer(host, user, tarball, remote_tar, sha) + transfer(host, user, HOST_SCRIPT, remote_script, sha256_of(HOST_SCRIPT)) + print("执行主机部署脚本（唯一脚本，staging 换目录 + 失败回滚）…") + r = subprocess.run( + ssh_cmd(host, user, + "sh %s %s %s %s" % (remote_script, remote_tar, sha, commit)), + capture_output=True, text=True) + print(r.stdout) + if r.stderr: + print(r.stderr, file=sys.stderr) + if r.returncode != 0 or not parse_release_ok(r.stdout, commit): + print("主机部署未返回有效的 RELEASE_OK，部署失败（主机侧已回滚或未动）。", + file=sys.stderr) + sys.exit(1) + # 独立二次核对：不只信主机脚本的打印，再 ssh 读一次 live 清单。 + live = subprocess.run( + ssh_cmd(host, user, "cat /opt/codex-app-server/DEPLOYED_COMMIT"), + capture_output=True, text=True) + if live.stdout.strip() != commit: + print("二次核对失败：live DEPLOYED_COMMIT=%s，期望 %s。" + "请立即用 deploy/rollback.sh 检查，不要再推新版。" + % (live.stdout.strip(), commit), file=sys.stderr) + sys.exit(1) + print("部署成功且二次核对通过：commit=%s sha256=%s" % (commit, sha)) + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/deploy/host_deploy.sh b/deploy/host_deploy.sh new file mode 100644 index 0000000..d5d1dae --- /dev/null +++ b/deploy/host_deploy.sh @@ -0,0 +1,214 @@ +#!/bin/sh +# 主机侧唯一部署脚本（取代 deploy.sh 内嵌 HOST_SCRIPT 与 +# ~/workspace/ssh-helpers/deploy_*_host.sh 等多份手工脚本）。 +# +# 设计目标：旧 tar 绝不能被误部署，health 通过绝不能代替“新代码上线”。 +# +# 调用（VM 侧 deploy.py 传过来，绝不由人工在主机上拼路径）： +# sh host_deploy.sh +# tarball：远端唯一路径，例如 /tmp/codex-app-server-releases/.tar.gz +# 绝不再用 /tmp/codex-app-server-code.tar.gz 这种通用路径。 +# expected_sha256：VM 侧算好的 tar sha256，主机侧先验再动任何东西。 +# expected_commit：git 完整 commit，tar 根目录的 DEPLOYED_COMMIT +# 必须与它一致，否则直接退出，旧代码原地不动。 +# +# 流程（任何一步失败都走回滚，退出非零）： +# 1. 验 tar sha256，不符直接退出，不备份、不解包、不重启。 +# 2. 验 tar 内 DEPLOYED_COMMIT，不符直接退出。 +# 3. 解包到 staging 目录（与 APP_DIR 同一文件系统，便于 mv 换目录）， +# 并对 staging 做 python 编译检查，语法坏的包绝不上线。 +# 4. 备份当前 live 代码（备份名含旧 commit + 时间戳，留最近 5 份； +# 备份 tar 遇运行中文件变化返回 1 时不直接 abort，只要备份文件 +# 非空就继续，见 PR #46 教训）。 +# 5. 停 server + worker/autoscaler，再换目录：live -> old， +# staging -> live，work/ 与 host-config-real.json（如有）原样搬回。 +# 换目录代替“原地 rm + 解包”，中途失败不会留半解包的 live。 +# 6. 启动并验 health 60 秒：ok=true 且 backend=postgres 且 db_ok=true， +# 且 live 的 DEPLOYED_COMMIT 等于 expected_commit；health 里 +# 若有 commit 字段且非空，也必须等于 expected_commit。 +# 7. 成功打印 RELEASE_OK commit=... sha=... backup=...，删掉 old 目录 +# （备份 tar 已留档）；失败则把 old 换回、重启、再验 health， +# 回滚后 health 恢复也算部署失败（新代码未上线），退出非零。 +# +# 可覆盖的环境变量（仅用于本地仿真测试，生产用默认值）： +# CODEX_DEPLOY_APP_DIR、CODEX_DEPLOY_BACKUP_DIR、 +# CODEX_DEPLOY_HEALTH_INTERVAL、CODEX_DEPLOY_HEALTH_ATTEMPTS +# 回滚边界：只回代码，不回数据库（迁移只加列、幂等，见 deploy.sh 注释）。 +# 密钥：本脚本不读 env 文件、不打印任何密钥。 +set -eu + +TARBALL=${1:-} +EXPECTED_SHA=${2:-} +EXPECTED_COMMIT=${3:-} +if [ -z "$TARBALL" ] || [ -z "$EXPECTED_SHA" ] || [ -z "$EXPECTED_COMMIT" ]; then + echo "用法：sh host_deploy.sh " >&2 + exit 2 +fi +if [ ! -f "$TARBALL" ]; then + echo "部署包不存在，拒绝部署（旧代码不动）：$TARBALL" >&2 + exit 2 +fi + +APP_DIR=${CODEX_DEPLOY_APP_DIR:-/opt/codex-app-server} +APP_BASE=$(basename "$APP_DIR") +BACKUP_DIR=${CODEX_DEPLOY_BACKUP_DIR:-/opt/codex-app-server-backups} +STAGING_DIR=${CODEX_DEPLOY_STAGING_DIR:-/opt/codex-app-server-staging} +OLD_DIR=${CODEX_DEPLOY_OLD_DIR:-/opt/codex-app-server-old} +FAILED_DIR=${CODEX_DEPLOY_FAILED_DIR:-/opt/codex-app-server-failed} +PRESERVE_DIR=${CODEX_DEPLOY_PRESERVE_DIR:-/opt/codex-app-server-preserve} +HEALTH_INTERVAL=${CODEX_DEPLOY_HEALTH_INTERVAL:-5} +HEALTH_ATTEMPTS=${CODEX_DEPLOY_HEALTH_ATTEMPTS:-12} + +# --- 1. 验 sha：不符直接退出，什么都不动 ------------------------------- +ACTUAL_SHA=$(sha256sum "$TARBALL" | awk '{print $1}') +if [ "$ACTUAL_SHA" != "$EXPECTED_SHA" ]; then + echo "部署包 sha256 不符，拒绝部署（旧代码不动）：" >&2 + echo " expected=$EXPECTED_SHA" >&2 + echo " actual=$ACTUAL_SHA tarball=$TARBALL" >&2 + exit 2 +fi + +# --- 2. 验 tar 内清单：DEPLOYED_COMMIT 必须等于 expected ------------- +TARBALL_COMMIT=$(tar -xzf "$TARBALL" -O DEPLOYED_COMMIT 2>/dev/null | tr -d ' \n\r' || true) +if [ "$TARBALL_COMMIT" != "$EXPECTED_COMMIT" ]; then + echo "部署包内 DEPLOYED_COMMIT 与期望 commit 不符，拒绝部署（旧代码不动）：" >&2 + echo " expected=$EXPECTED_COMMIT tarball_commit=$TARBALL_COMMIT" >&2 + exit 2 +fi + +# --- 3. staging 解包 + 编译检查 ---------------------------------------- +rm -rf "$STAGING_DIR" +mkdir -p "$STAGING_DIR" +tar -xzf "$TARBALL" -C "$STAGING_DIR" +python3 -m compileall -q "$STAGING_DIR" || { + echo "staging 编译检查失败，拒绝部署（旧代码不动）" >&2 + rm -rf "$STAGING_DIR" + exit 2 +} + +check_health() { + OUT=$(curl -s --max-time 5 http://127.0.0.1:8765/health || true) + [ -n "$OUT" ] && printf '%s' "$OUT" | EXPECTED_COMMIT="$EXPECTED_COMMIT" python3 -c ' +import json, os, sys +try: + d = json.load(sys.stdin) +except Exception: + sys.exit(1) +if not (d.get("ok") is True and d.get("backend") == "postgres" + and d.get("db_ok") is True): + sys.exit(1) +commit = d.get("commit") +if commit is not None and commit != os.environ.get("EXPECTED_COMMIT"): + sys.exit(1) +sys.exit(0) +' +} + +wait_health() { + i=0 + while [ "$i" -lt "$HEALTH_ATTEMPTS" ]; do + if check_health; then return 0; fi + sleep "$HEALTH_INTERVAL" + i=$((i + 1)) + done + return 1 +} + +worker_unit() { + if systemctl is-active -q codex-autoscaler 2>/dev/null; then + echo codex-autoscaler + else + echo codex-cloud-worker + fi +} + +# --- 4. 备份 live（名含旧 commit，留 5 份） ------------------------------ +mkdir -p "$BACKUP_DIR" +PREV_COMMIT=unknown +if [ -f "$APP_DIR/DEPLOYED_COMMIT" ]; then + PREV_COMMIT=$(tr -d ' \n\r' < "$APP_DIR/DEPLOYED_COMMIT" || echo unknown) + [ -n "$PREV_COMMIT" ] || PREV_COMMIT=unknown +fi +STAMP=$(date +%Y%m%d%H%M%S) +BACKUP_FILE="$BACKUP_DIR/codex-app-server-${PREV_COMMIT}-${STAMP}.tar.gz" +# PR #46 教训：备份 tar 遇运行中文件变化返回 1 时不 abort，文件非空即继续。 +tar --exclude="$APP_BASE/work" --exclude="$APP_BASE/worker-heartbeat.json" \ + -czf "$BACKUP_FILE" -C "$(dirname "$APP_DIR")" "$APP_BASE" || true +if [ ! -s "$BACKUP_FILE" ]; then + echo "备份失败（备份文件为空），拒绝部署（旧代码不动）" >&2 + rm -rf "$STAGING_DIR" + exit 2 +fi +echo "已备份当前代码到：$BACKUP_FILE" +ls -t "$BACKUP_DIR"/codex-app-server-*.tar.gz 2>/dev/null \ + | tail -n +6 | xargs -r rm -f + +WORKER_UNIT=$(worker_unit) +echo "停服务并换目录：worker_unit=$WORKER_UNIT" +systemctl stop "$WORKER_UNIT" codex-cloud-server || true + +# --- 5. 换目录：live -> old，staging -> live，保留文件搬回 -------------- +rm -rf "$OLD_DIR" "$PRESERVE_DIR" +mkdir -p "$PRESERVE_DIR" +for name in work host-config-real.json; do + if [ -e "$APP_DIR/$name" ]; then + mv "$APP_DIR/$name" "$PRESERVE_DIR/$name" + fi +done +mv "$APP_DIR" "$OLD_DIR" +mv "$STAGING_DIR" "$APP_DIR" +for name in work host-config-real.json; do + if [ -e "$PRESERVE_DIR/$name" ]; then + mv "$PRESERVE_DIR/$name" "$APP_DIR/$name" + fi +done +rmdir "$PRESERVE_DIR" 2>/dev/null || true +if id codexsvc >/dev/null 2>&1; then + chown -R codexsvc:codexsvc "$APP_DIR" +fi + +rollback() { + echo "新代码 health 未通过，开始回滚到备份：$BACKUP_FILE" >&2 + systemctl stop "$WORKER_UNIT" codex-cloud-server || true + rm -rf "$FAILED_DIR" "$PRESERVE_DIR" + mkdir -p "$PRESERVE_DIR" + for name in work host-config-real.json; do + if [ -e "$APP_DIR/$name" ]; then + mv "$APP_DIR/$name" "$PRESERVE_DIR/$name" + fi + done + mv "$APP_DIR" "$FAILED_DIR" + mv "$OLD_DIR" "$APP_DIR" + for name in work host-config-real.json; do + if [ -e "$PRESERVE_DIR/$name" ]; then + mv "$PRESERVE_DIR/$name" "$APP_DIR/$name" + fi + done + rmdir "$PRESERVE_DIR" 2>/dev/null || true + if id codexsvc >/dev/null 2>&1; then + chown -R codexsvc:codexsvc "$APP_DIR" + fi + systemctl start codex-cloud-server "$WORKER_UNIT" || true + if wait_health; then + echo "部署失败，已回滚到备份，health 已恢复（新代码未上线）" >&2 + else + echo "已回滚但 health 仍失败，需人工介入" >&2 + fi + exit 1 +} + +# --- 6. 启动 + 验 health + 验上线 commit ------------------------------- +systemctl start codex-cloud-server "$WORKER_UNIT" || true +if ! wait_health; then + rollback +fi +LIVE_COMMIT=$(tr -d ' \n\r' < "$APP_DIR/DEPLOYED_COMMIT" 2>/dev/null || true) +if [ "$LIVE_COMMIT" != "$EXPECTED_COMMIT" ]; then + echo "上线 commit 不符（health 通过但代码不是新版），开始回滚：" >&2 + echo " expected=$EXPECTED_COMMIT live=$LIVE_COMMIT" >&2 + rollback +fi + +rm -rf "$OLD_DIR" +echo "RELEASE_OK commit=$EXPECTED_COMMIT sha=$EXPECTED_SHA backup=$BACKUP_FILE" +exit 0 diff --git a/deploy/tests/test_deploy.py b/deploy/tests/test_deploy.py new file mode 100644 index 0000000..611ec94 --- /dev/null +++ b/deploy/tests/test_deploy.py @@ -0,0 +1,93 @@ +"""deploy.py 的单测：只测纯逻辑与发布包结构，不碰真实主机。 + +主机侧 host_deploy.sh 的仿真见 test_host_deploy.py（假 systemctl/curl）。 +""" +import hashlib +import os +import subprocess +import sys +import tarfile +import tempfile +import unittest + +REPO_ROOT = os.path.dirname(os.path.dirname(os.path.dirname( + os.path.abspath(__file__)))) +sys.path.insert(0, os.path.join(REPO_ROOT, "deploy")) +import deploy as deploy_mod # noqa: E402 + + +class TestDeployPure(unittest.TestCase): + def test_release_tarball_name_is_commit_scoped(self): + name = deploy_mod.release_tarball_name("abc123") + self.assertEqual(name, "codex-app-server-abc123.tar.gz") + # 绝不再用通用名：不同 commit 绝不可能撞到同一个远端路径。 + self.assertNotEqual( + deploy_mod.release_tarball_name("abc123"), + deploy_mod.release_tarball_name("def456")) + + def test_parse_release_ok_requires_matching_commit(self): + out = "部署成功\nRELEASE_OK commit=abc sha=dead backup=/tmp/x.tar.gz\n" + self.assertTrue(deploy_mod.parse_release_ok(out, "abc")) + self.assertFalse(deploy_mod.parse_release_ok(out, "other")) + self.assertFalse(deploy_mod.parse_release_ok("部署成功，health 通过\n", "abc")) + + def test_sha256_of(self): + with tempfile.NamedTemporaryFile(delete=False) as f: + f.write(b"hello") + path = f.name + try: + self.assertEqual(deploy_mod.sha256_of(path), + hashlib.sha256(b"hello").hexdigest()) + finally: + os.unlink(path) + + def test_tarball_commit_missing_manifest_refuses(self): + with tempfile.NamedTemporaryFile(suffix=".tar.gz", delete=False) as f: + path = f.name + try: + with tarfile.open(path, "w:gz") as tf: + pass + with self.assertRaises(SystemExit): + deploy_mod.tarball_commit(path) + finally: + os.unlink(path) + + +class TestBuildRelease(unittest.TestCase): + def test_build_from_temp_git_repo(self): + with tempfile.TemporaryDirectory() as tmp: + repo = os.path.join(tmp, "repo") + os.makedirs(repo) + subprocess.run(["git", "init", "-q"], cwd=repo, check=True) + subprocess.run(["git", "config", "user.email", "t@t"], + cwd=repo, check=True) + subprocess.run(["git", "config", "user.name", "t"], + cwd=repo, check=True) + with open(os.path.join(repo, "hello.py"), "w") as f: + f.write("x = 1\n") + subprocess.run(["git", "add", "."], cwd=repo, check=True) + subprocess.run(["git", "commit", "-qm", "init"], + cwd=repo, check=True) + commit = subprocess.run( + ["git", "rev-parse", "HEAD"], cwd=repo, capture_output=True, + text=True, check=True).stdout.strip() + # 把 deploy 模块的 REPO_ROOT 临时指到这个小仓库。 + old = deploy_mod.REPO_ROOT + deploy_mod.REPO_ROOT = repo + try: + out_dir = os.path.join(tmp, "out") + tarball, sha, got = deploy_mod.build_release("HEAD", out_dir) + finally: + deploy_mod.REPO_ROOT = old + self.assertEqual(got, commit) + self.assertEqual(deploy_mod.sha256_of(tarball), sha) + self.assertEqual(deploy_mod.tarball_commit(tarball), commit) + with tarfile.open(tarball, "r:gz") as tf: + names = tf.getnames() + self.assertIn("DEPLOYED_COMMIT", names) + self.assertIn("RELEASE.json", names) + self.assertIn("hello.py", names) + + +if __name__ == "__main__": + unittest.main() diff --git a/deploy/tests/test_host_deploy.py b/deploy/tests/test_host_deploy.py new file mode 100644 index 0000000..b87b1d0 --- /dev/null +++ b/deploy/tests/test_host_deploy.py @@ -0,0 +1,147 @@ +"""host_deploy.sh 的仿真测试：假 systemctl / 假 curl，不碰真实主机。 + +每个用例都在临时目录里建一个假 APP_DIR，host_deploy.sh 的 +CODEX_DEPLOY_* 环境变量把它指过去。生产默认值不受影响。 +""" +import hashlib +import json +import os +import subprocess +import tarfile +import tempfile +import unittest + +REPO_ROOT = os.path.dirname(os.path.dirname(os.path.dirname( + os.path.abspath(__file__)))) +HOST_SCRIPT = os.path.join(REPO_ROOT, "deploy", "host_deploy.sh") + + +def make_tarball(source_dir, tarball): + with tarfile.open(tarball, "w:gz", format=tarfile.PAX_FORMAT) as tf: + for root, dirs, files in os.walk(source_dir): + dirs.sort() + for name in sorted(files): + full = os.path.join(root, name) + tf.add(full, arcname=os.path.relpath(full, source_dir)) + with open(tarball, "rb") as f: + return hashlib.sha256(f.read()).hexdigest() + + +class HostDeploySim(unittest.TestCase): + def setUp(self): + self.tmp = tempfile.TemporaryDirectory() + base = self.tmp.name + self.app = os.path.join(base, "codex-app-server") + os.makedirs(os.path.join(self.app, "work")) + with open(os.path.join(self.app, "DEPLOYED_COMMIT"), "w") as f: + f.write("oldcommit\n") + with open(os.path.join(self.app, "work", "keep.txt"), "w") as f: + f.write("keep\n") + with open(os.path.join(self.app, "old.py"), "w") as f: + f.write("x = 0\n") + # 假 systemctl：is-active 一律失败（走 cloud-worker 分支），其余成功。 + fake_bin = os.path.join(base, "fakebin") + os.makedirs(fake_bin) + with open(os.path.join(fake_bin, "systemctl"), "w") as f: + f.write("#!/bin/sh\n" + "echo \"$@\" >> \"$FAKE_SYSTEMCTL_LOG\"\n" + "if [ \"$1\" = \"is-active\" ]; then exit 1; fi\n" + "exit 0\n") + with open(os.path.join(fake_bin, "curl"), "w") as f: + f.write("#!/bin/sh\ncat \"$FAKE_HEALTH_JSON\"\n") + os.chmod(os.path.join(fake_bin, "systemctl"), 0o755) + os.chmod(os.path.join(fake_bin, "curl"), 0o755) + self.fake_bin = fake_bin + self.health_path = os.path.join(base, "health.json") + self.log_path = os.path.join(base, "systemctl.log") + open(self.log_path, "w").close() + self.env = dict(os.environ) + self.env["PATH"] = fake_bin + os.pathsep + os.environ["PATH"] + self.env["CODEX_DEPLOY_APP_DIR"] = self.app + self.env["CODEX_DEPLOY_BACKUP_DIR"] = os.path.join(base, "backups") + self.env["CODEX_DEPLOY_STAGING_DIR"] = os.path.join(base, "staging") + self.env["CODEX_DEPLOY_OLD_DIR"] = os.path.join(base, "old") + self.env["CODEX_DEPLOY_FAILED_DIR"] = os.path.join(base, "failed") + self.env["CODEX_DEPLOY_PRESERVE_DIR"] = os.path.join(base, "preserve") + self.env["CODEX_DEPLOY_HEALTH_INTERVAL"] = "0.1" + self.env["CODEX_DEPLOY_HEALTH_ATTEMPTS"] = "3" + self.env["FAKE_HEALTH_JSON"] = self.health_path + self.env["FAKE_SYSTEMCTL_LOG"] = self.log_path + + def tearDown(self): + self.tmp.cleanup() + + def set_health(self, payload): + with open(self.health_path, "w") as f: + json.dump(payload, f) + + def build_new_tarball(self, commit="newcommit"): + src = os.path.join(self.tmp.name, "src") + os.makedirs(src, exist_ok=True) + with open(os.path.join(src, "DEPLOYED_COMMIT"), "w") as f: + f.write(commit + "\n") + with open(os.path.join(src, "new.py"), "w") as f: + f.write("x = 1\n") + tarball = os.path.join(self.tmp.name, "release.tar.gz") + sha = make_tarball(src, tarball) + return tarball, sha + + def run_host(self, tarball, sha, commit): + return subprocess.run( + ["sh", HOST_SCRIPT, tarball, sha, commit], + capture_output=True, text=True, env=self.env) + + def live_commit(self): + with open(os.path.join(self.app, "DEPLOYED_COMMIT")) as f: + return f.read().strip() + + def test_success_swaps_and_preserves_work(self): + self.set_health({"ok": True, "backend": "postgres", "db_ok": True, + "commit": "newcommit"}) + tarball, sha = self.build_new_tarball() + r = self.run_host(tarball, sha, "newcommit") + self.assertEqual(r.returncode, 0, r.stdout + r.stderr) + self.assertIn("RELEASE_OK", r.stdout) + self.assertIn("commit=newcommit", r.stdout) + self.assertEqual(self.live_commit(), "newcommit") + with open(os.path.join(self.app, "work", "keep.txt")) as f: + self.assertEqual(f.read().strip(), "keep") + self.assertTrue(os.path.exists(os.path.join(self.app, "new.py"))) + self.assertFalse(os.path.exists(os.path.join(self.app, "old.py"))) + backups = os.listdir(self.env["CODEX_DEPLOY_BACKUP_DIR"]) + self.assertEqual(len(backups), 1) + self.assertIn("oldcommit", backups[0]) + + def test_bad_sha_never_touches_live(self): + self.set_health({"ok": True, "backend": "postgres", "db_ok": True, + "commit": "newcommit"}) + tarball, _sha = self.build_new_tarball() + r = self.run_host(tarball, "0" * 64, "newcommit") + self.assertNotEqual(r.returncode, 0) + self.assertEqual(self.live_commit(), "oldcommit") + self.assertFalse(os.path.exists(self.env["CODEX_DEPLOY_BACKUP_DIR"]) + and os.listdir(self.env["CODEX_DEPLOY_BACKUP_DIR"])) + + def test_manifest_mismatch_never_touches_live(self): + self.set_health({"ok": True, "backend": "postgres", "db_ok": True, + "commit": "newcommit"}) + tarball, sha = self.build_new_tarball(commit="othercommit") + r = self.run_host(tarball, sha, "newcommit") + self.assertNotEqual(r.returncode, 0) + self.assertEqual(self.live_commit(), "oldcommit") + + def test_health_failure_rolls_back(self): + # health 通过但 commit 是旧的：这是旧链路会误报成功的情况， + # 新链路必须回滚，live 回到 oldcommit。 + self.set_health({"ok": True, "backend": "postgres", "db_ok": True, + "commit": "oldcommit"}) + tarball, sha = self.build_new_tarball() + r = self.run_host(tarball, sha, "newcommit") + self.assertNotEqual(r.returncode, 0) + self.assertEqual(self.live_commit(), "oldcommit") + with open(os.path.join(self.app, "work", "keep.txt")) as f: + self.assertEqual(f.read().strip(), "keep") + + +if __name__ == "__main__": + unittest.main()