sdd.md 16.2 KB

Edit Raw Blame History


SDD: supervisor/deploy_stage.py 并发部署冲突诊断化处理
对应 PRD:《supervisor/deploy_stage.stage_deployment() 并发部署冲突的诊断化处理》(FR-1~FR-5, AC-1~AC-6)。
1. 设计概述
本次改动不涉及 schema 变更（ux_deployment_active 唯一索引已存在，见 src/schema.py:332-333），
不引入新模块、不改变 stage_deployment() 的对外签名与返回值结构。改动局限于
supervisor/deploy_stage.py 一个模块内：


新增一个具名异常 ConcurrentDeploymentError（放置与文档风格对齐既有的
CheckoutFailedError/deploy_cutover.CutoverFailedError/deploy_rollback.RollbackFailedError）。
在 stage_deployment() 触发 INSERT INTO deployment ... 的语句外包一层 try/except
sqlite3.IntegrityError，把"因 ux_deployment_active 冲突导致的插入失败"精确识别出来并转译为
ConcurrentDeploymentError；非该冲突导致的 IntegrityError 原样重新抛出，不吞掉。
一个新增只读诊断辅助函数，用于在转译时尽力查出当前占用中的部署行（deployment_id/status），
供异常携带；查不到不阻塞异常抛出。
测试：改造现有并发测试的断言粒度，并新增两个终态放行回归测试（rolled_back、blocked_pending_human）。


涉及文件：


supervisor/deploy_stage.py（修改）
supervisor/test_deploy_stage.py（修改 + 新增测试）
src/schema.py：不改动（ux_deployment_active 的终态集合定义本次不变，PRD 非目标已明确）。

2. 接口设计
2.1 新增异常类 ConcurrentDeploymentError
放置位置：supervisor/deploy_stage.py，紧接在既有 CheckoutFailedError（第 38-42 行）之后，模块顶部，
文档风格与其保持一致（触发场景 + 是否重试 + 已产生的副作用状态，均在 docstring 内说明）。
class ConcurrentDeploymentError(Exception):
    """CONCURRENT_DEPLOYMENT -- ux_deployment_active 唯一索引在插入新 deployment
    行时检测到已存在一条未到终态（finalized/rolled_back/blocked_pending_human
    之外）的部署记录。与 CheckoutFailedError 不同：这次 INSERT 从未落地提交，
    没有半途而废的 deployment 行需要标记为 blocked_pending_human -- 调用方（Deploy
    Supervisor 轮询循环）应把这次调用视为"本轮跳过，下次轮询再试"，而不是
    需要人工介入的失败。诊断字段可通过属性访问，不只是拼进消息字符串：
      - git_commit_hash: 本次被拒绝、尝试发起的提交哈希。
      - blocking_deployment_id / blocking_status: 当前占用中的部署记录（若能
        通过只读查询定位到）；查不到时二者均为 None，异常仍正常抛出。
    """

    def __init__(self, git_commit_hash, blocking_deployment_id=None, blocking_status=None):
        self.git_commit_hash = git_commit_hash
        self.blocking_deployment_id = blocking_deployment_id
        self.blocking_status = blocking_status
        message = "cannot stage commit {}: a deployment is already in progress".format(git_commit_hash)
        if blocking_deployment_id is not None:
            message += " (blocking deployment_id={}, status={})".format(
                blocking_deployment_id, blocking_status
            )
        super().__init__(message)

2.2 stage_deployment() 对外签名与返回值
不变：stage_deployment(conn, git_commit_hash, repo_root=None, prod_db_path=None,
checkout_parent_dir=None, port=None, python_executable=None)，成功时返回值结构不变。
新增的唯一可观察差异是：冲突场景下抛出的异常类型从未声明的裸 sqlite3.IntegrityError
变为具名的 deploy_stage.ConcurrentDeploymentError。
2.3 新增内部辅助函数
def _is_concurrent_deployment_conflict(error):
    """区分「ux_deployment_active 唯一约束冲突」与其他 IntegrityError（例如未来
    schema 变化引入的其他约束、连接层面的完整性错误）。判别手段：sqlite 自
    3.8.2 起，对命名的 UNIQUE INDEX（区别于表列级 UNIQUE 约束）报错时会在
    异常消息中包含该索引名，形如 "UNIQUE constraint failed: index
    'ux_deployment_active'"。只匹配这一具体索引名字符串，不笼统捕获所有
    IntegrityError，避免把无关的完整性错误误判为并发冲突而掩盖真实问题。
    """
    return "ux_deployment_active" in str(error)


def _find_blocking_deployment(conn):
    """只读诊断查询：定位当前占用 ux_deployment_active 的那一条非终态部署行。
    仅在检测到并发冲突之后调用，纯粹用于丰富异常的诊断信息。查询本身失败
    （理论上不应发生，例如连接已损坏）不应掩盖原始的并发冲突 -- 这里吞掉查询期间
    的 sqlite3.Error 并返回 None，调用方在拿不到诊断信息时仍然正常抛出
    ConcurrentDeploymentError，只是不携带 blocking_deployment_id/blocking_status。
    """
    try:
        row = conn.execute(
            "SELECT deployment_id, status FROM deployment "
            "WHERE status NOT IN ('finalized', 'rolled_back', 'blocked_pending_human') "
            "ORDER BY rowid DESC LIMIT 1"
        ).fetchone()
        return tuple(row) if row else None
    except sqlite3.Error:
        return None


import sqlite3 需要新增到 supervisor/deploy_stage.py 顶部的标准库导入区（当前只导入了
os, shutil, socket, subprocess, sys, uuid）。
3. 控制流改造（stage_deployment() 内部）
现状（第 170-176 行）：
conn.execute(
    "INSERT INTO deployment "
    "(deployment_id, git_commit_hash, previous_commit_hash, status, previous_instance_dir, started_at) "
    "VALUES (?, ?, ?, 'staged', ?, CURRENT_TIMESTAMP)",
    (deployment_id, git_commit_hash, previous_commit_hash, previous_instance_dir),
)
conn.commit()


改造后（原地包一层 try/except，插入语句本身、参数、之后的 _checkout_candidate 起始点完全不变，
满足 PRD FR-2 "不改变顺序，只改变冲突时的异常转译"约束）：
try:
    conn.execute(
        "INSERT INTO deployment "
        "(deployment_id, git_commit_hash, previous_commit_hash, status, previous_instance_dir, started_at) "
        "VALUES (?, ?, ?, 'staged', ?, CURRENT_TIMESTAMP)",
        (deployment_id, git_commit_hash, previous_commit_hash, previous_instance_dir),
    )
    conn.commit()
except sqlite3.IntegrityError as e:
    conn.rollback()
    if not _is_concurrent_deployment_conflict(e):
        raise
    blocking = _find_blocking_deployment(conn)
    blocking_deployment_id, blocking_status = blocking if blocking else (None, None)
    raise ConcurrentDeploymentError(
        git_commit_hash=git_commit_hash,
        blocking_deployment_id=blocking_deployment_id,
        blocking_status=blocking_status,
    ) from e


conn.rollback() 的作用：sqlite3 默认 isolation_level 下，第一条 DML 语句会隐式开启一个事务；
INSERT 抛出 IntegrityError 时该事务从未 commit()，rollback() 在这里是防御性的连接状态清理
（确保后续 _find_blocking_deployment 的只读查询、以及调用方复用同一个 conn 做后续操作时，
连接处于干净状态），不是撤销任何已经落地的副作用——因为根本没有副作用落地。

之后的分支（_checkout_candidate 抛 subprocess.CalledProcessError -> CheckoutFailedError，
第 178-206 行）完全不变，不受本次改动影响（对应 PRD AC-5）。
4. 异常传播路径总览


场景
触发点
抛出的异常
deployment 表状态


已有非终态部署记录
INSERT 语句触发 ux_deployment_active 冲突
ConcurrentDeploymentError（携带 git_commit_hash/blocking_deployment_id/blocking_status）
本次尝试的行从未写入（INSERT 失败即回滚），不新增任何行


其他 IntegrityError（非本索引）
同上 INSERT 语句，但错误消息不含 ux_deployment_active
原始 sqlite3.IntegrityError 原样重新抛出
同上，不新增行


INSERT 成功，之后 checkout 失败
_checkout_candidate 内 git worktree add 失败
CheckoutFailedError（既有行为，不变）
已插入的行被 UPDATE ... SET status = 'blocked_pending_human'


INSERT 成功，checkout/db copy/launch 全部成功
无
无
行被 UPDATE ... SET status = 'candidate_starting'


5. 测试设计（对应 PRD FR-3/FR-4/FR-5，supervisor/test_deploy_stage.py）
5.1 改造：冲突路径断言收紧（对应 AC-1，替换现有 test_only_one_non_terminal_deployment_allowed）
def test_only_one_non_terminal_deployment_allowed(self):
    with patch.object(deploy_stage, "_launch_candidate") as mock_launch:
        mock_launch.return_value.pid = 1
        first = deploy_stage.stage_deployment(
            self.conn, self.commit_v2, repo_root=self.repo_dir, prod_db_path=self.prod_db_path, port=9001,
        )
        with self.assertRaises(deploy_stage.ConcurrentDeploymentError) as ctx:
            deploy_stage.stage_deployment(
                self.conn, self.commit_v1, repo_root=self.repo_dir, prod_db_path=self.prod_db_path, port=9002,
            )
    err = ctx.exception
    self.assertEqual(err.git_commit_hash, self.commit_v1)
    self.assertEqual(err.blocking_deployment_id, first["deployment_id"])
    self.assertEqual(err.blocking_status, "candidate_starting")
    # 冲突时不应新增第二条部署记录
    count = self.conn.execute("SELECT COUNT(*) FROM deployment").fetchone()[0]
    self.assertEqual(count, 1)


注：第一次调用的 mock_launch 让 stage_deployment() 把行状态更新为 candidate_starting
（第 195-200 行的既有逻辑），因此 blocking_status 断言为 "candidate_starting" 而非 "staged"。
5.2 新增：终态放行回归测试（对应 AC-3/AC-4，加入 StageDeploymentHappyPathTest 或独立测试类）
class TerminalStateAllowsNewDeploymentTest(DeployStageTestBase):
    def _seed_deployment(self, status):
        self.conn.execute(
            "INSERT INTO deployment (deployment_id, git_commit_hash, previous_commit_hash, status, "
            "candidate_instance_dir, previous_instance_dir) "
            "VALUES ('dep-prior', 'commit-prior', 'commit-minus-1', ?, '/old/candidate/dir', '/older/dir')",
            (status,),
        )
        self.conn.commit()

    def test_rolled_back_prior_deployment_allows_new_staging(self):
        self._seed_deployment("rolled_back")
        with patch.object(deploy_stage, "_launch_candidate") as mock_launch:
            mock_launch.return_value.pid = 111
            result = deploy_stage.stage_deployment(
                self.conn, self.commit_v2, repo_root=self.repo_dir, prod_db_path=self.prod_db_path, port=9101,
            )
        self.assertEqual(result["status"], "candidate_starting")
        row = self.conn.execute(
            "SELECT status FROM deployment WHERE deployment_id = ?", (result["deployment_id"],)
        ).fetchone()
        self.assertEqual(row[0], "candidate_starting")

    def test_blocked_pending_human_prior_deployment_allows_new_staging(self):
        self._seed_deployment("blocked_pending_human")
        with patch.object(deploy_stage, "_launch_candidate") as mock_launch:
            mock_launch.return_value.pid = 222
            result = deploy_stage.stage_deployment(
                self.conn, self.commit_v2, repo_root=self.repo_dir, prod_db_path=self.prod_db_path, port=9102,
            )
        self.assertEqual(result["status"], "candidate_starting")
        row = self.conn.execute(
            "SELECT status FROM deployment WHERE deployment_id = ?", (result["deployment_id"],)
        ).fetchone()
        self.assertEqual(row[0], "candidate_starting")

5.3 覆盖矩阵


PRD 验收标准
测试


AC-1（并发冲突，具名异常 + 诊断字段 + 不新增行）
test_only_one_non_terminal_deployment_allowed（改造后）


AC-2（finalized 放行，既有行为不回归）
BootstrapPreviousTest.test_prior_finalized_deployment_supplies_previous_fields（已存在，不改）


AC-3（rolled_back 放行）
test_rolled_back_prior_deployment_allows_new_staging（新增）


AC-4（blocked_pending_human 放行）
test_blocked_pending_human_prior_deployment_allows_new_staging（新增）


AC-5（CheckoutFailedError 路径不回归）
CheckoutFailedTest（已存在，不改）


AC-6（全量测试全绿）
python3 -m unittest discover -s src + python3 -m unittest discover -s supervisor -p "test_*.py"


6. Development Unit 分解
本次改动规模小、边界清楚、无跨模块依赖，符合 PRD 明确排除拆分的判断——分解为一个 Development Unit。
U-DS01：deploy_stage 并发冲突诊断化改造

类型：Bug 修复（纵向、单模块，具备独立可验证产物）。
目标：stage_deployment() 在遇到 ux_deployment_active 冲突时抛出携带诊断字段的具名异常
ConcurrentDeploymentError，并补齐终态放行的回归测试覆盖。
范围：


supervisor/deploy_stage.py：新增 import sqlite3；新增 ConcurrentDeploymentError 类
（§2.1）；新增 _is_concurrent_deployment_conflict/_find_blocking_deployment 辅助函数（§2.3）；
改造 stage_deployment() 的 INSERT 语句为 try/except（§3）。
supervisor/test_deploy_stage.py：改造 test_only_one_non_terminal_deployment_allowed
断言粒度（§5.1）；新增 TerminalStateAllowsNewDeploymentTest 两个用例（§5.2）。

不在范围内：src/schema.py（不改）、deploy_cutover.py/deploy_rollback.py（不改，
仅作为命名风格参照）、deploy_supervisor.py 轮询循环对新异常的具体处理策略（PRD 未要求，
调用方目前对 stage_deployment() 的调用点若有裸 except Exception 兜底，ConcurrentDeployment
Error 会被同样兜住，不需要额外改动；若需要针对性处理属于后续任务）。
接口契约：ConcurrentDeploymentError(git_commit_hash, blocking_deployment_id=None,
blocking_status=None)，属性只读、通过实例访问。stage_deployment() 签名与成功返回值结构不变。
依赖：无跨 Unit 依赖，无外部依赖新增。
Done 标准（可观察证据）：


supervisor/deploy_stage.py 中存在 ConcurrentDeploymentError 类定义，且
stage_deployment() 的并发路径抛出该类型而非裸 sqlite3.IntegrityError。
§5.3 覆盖矩阵中列出的 6 个测试全部存在且通过。
python3 -m unittest discover -s src 退出码 0。
python3 -m unittest discover -s supervisor -p "test_*.py" 退出码 0。

风险：低。唯一需要留意的实现细节是 §2.3 中 _is_concurrent_deployment_conflict 的字符串
判别方式依赖当前 sqlite3 版本的错误消息格式（"UNIQUE constraint failed: index '<name>'"）——
如果本地/CI 的 sqlite3 版本明显老旧导致消息格式不同，判别函数会退化为把该冲突当作"未知
IntegrityError"原样抛出（不会误判为其他类型的错误吞掉，只是不会转译成
ConcurrentDeploymentError）；test_only_one_non_terminal_deployment_allowed 断言的具体异常
类型会在这种环境下失败并如实暴露该差异，而不是静默通过，因此不需要额外的运行期兜底代码。

7. 与既有约定的一致性检查

异常命名/放置/docstring 风格：对齐 CheckoutFailedError（同模块）、CutoverFailedError
（deploy_cutover.py）、RollbackFailedError（deploy_rollback.py）——均为"具名、模块顶部、
docstring 说明触发场景与是否重试"。
不引入应用层锁：继续依赖 ux_deployment_active 数据库唯一约束作为并发正确性的唯一事实来源
（PRD 非目标已明确），本次只是把约束冲突的"表现形式"从裸异常变为诊断化异常，不改变并发正确性
机制本身。
测试风格：延续 supervisor/test_deploy_stage.py 现有的"真实 sqlite + 真实 git worktree，只 mock
_launch_candidate"约定，新增测试不引入额外 mock 面。