# CI/CD、备份与回滚：让发布和恢复都可重复
## 用途、学习程度与前置
本地成功不等于每个人都能构建，备份文件存在也不等于灾难时能恢复。本章面向已经准备部署、需要持续发版的 AI 全栈应用。L2 要求是会把质量检查放进 CI，使用可追溯制品部署，并能在隔离环境验证一次备份恢复。
前置是 Git、退出码、Docker 基础与数据库表的概念。示例采用 Python 3.12 及以上的标准库 SQLite，运行过程在临时目录中完成，不接触业务数据库。它验证恢复机制，不声称已执行任何云发布、真实数据库备份或灾备演练。
## 把发布拆成不同阶段
持续集成 CI 负责在固定环境检查代码：安装锁定依赖、类型检查、必要测试、构建、AI 回归。持续交付负责产出可部署制品，并给出经过验证的候选版本；是否自动进入生产取决于团队策略。把这些概念分开，可以避免把“CI 绿了”直接等同于“生产已经更新”。
理想制品带提交哈希、构建时间、依赖锁与评测报告。测试环境和生产使用同一个镜像摘要，只替换受控运行配置。若生产重新构建一次，依赖、镜像标签或远程资源变化可能导致它不再是被测试的东西。前端文件带内容哈希时还要考虑旧页面引用旧 chunk：发布后短期保留旧资产，避免正在使用的页面突然加载失败。
流水线和应用同样有权限边界。PR 的检查流程通常只需读仓库，不应拿生产密钥。外部贡献或不可信脚本不能因为进入 CI 就获得部署权限。第三方 Action 使用完整提交 SHA 可固定执行内容；支持的环境优先使用短期身份授权，并限制仓库、分支和环境条件，而不是长期共享万能密钥。[GitHub Actions 安全实践](https://docs.github.com/en/actions/reference/security/secure-use)
## 回滚应用与恢复数据不是同一动作
回滚应用是把运行制品指回上一版本。数据库迁移可能已经改变结构和数据，旧应用未必还能工作。因此表结构尽量采用先扩展、后切换、再收缩的方式：先新增可兼容字段，发布兼容新旧字段的应用，迁移数据，再在后续版本删除旧字段。不要在同一次发布里删除旧列并默认旧镜像随时可回滚。
AI 应用还多出提示词、检索索引、embedding 模型、重排器与工具定义。只回滚服务代码，却留下新维度的向量或新提示格式，可能造成新故障。版本清单应记录这些依赖关系；有些变化适合版本化双写或双索引，有些需要明确迁移窗口。
数据恢复通常意味着回到较早的时间点，会影响之后写入的数据。恢复前要冻结或引导写流量，并确定如何对账、补偿与重放事件。不能因为新版本有 bug 就直接把数据库还原到昨天，把正常业务记录一起丢掉。发布失败时先判断能否回滚应用、前向修复或关闭功能，再决定是否需要恢复数据。
## 备份策略要回答四个问题
第一，备份什么：业务数据库、用户上传、配置、加密密钥相关恢复材料，以及重建检索索引需要的原始文档。向量索引可重建时不一定需要同等备份频率，但原始文档、元数据和权限关系必须保留。
第二，能丢多少：RPO 是可接受的数据损失时间窗口。第三，多久恢复：RTO 是恢复服务所需时间目标。第四，谁能恢复：备份的加密、访问权限、异地副本、保留规则与密钥恢复能力要一起设计。备份和密钥丢在同一个失效环境里，往往无法真正恢复。
逻辑备份导出数据库对象与数据，物理备份保存数据库文件级状态，各有适用范围。PostgreSQL 的 pg_dump 可导出一致性快照，但单数据库 dump 不包含全部集群级对象；角色等需要额外方案。恢复能力还受版本、扩展和权限配置影响，不能只检查文件大小。[PostgreSQL 逻辑备份](https://www.postgresql.org/docs/current/backup-dump.html)
不要在数据库运行时随意复制某一个文件就当完整备份，尤其要考虑 WAL 和并发写入。SQLite 提供连接级 backup API，可以把数据库内容备份到另一个连接；本章用它演示“备份、恢复、比对”这条完整路径。[SQLite backup API](https://docs.python.org/3/library/sqlite3.html#sqlite3.Connection.backup)
## 完整示例：备份到新数据库，再恢复验证
保存为 backup_and_restore.py，只需已安装 Python，无第三方依赖。
```python
# backup_and_restore.py
from pathlib import Path
from tempfile import TemporaryDirectory
from contextlib import closing
from unittest.mock import patch
import os
import sqlite3
def rows(path: Path) -> list[tuple[int, str]]:
# with 管事务，显式 close 管连接生命周期
con = sqlite3.connect(path)
try:
return con.execute("SELECT id, body FROM notes ORDER BY id").fetchall()
finally:
con.close()
def copy_database(source: Path, target: Path) -> None:
if not source.is_file():
raise FileNotFoundError("备份源不存在")
# mode=ro 防止源文件消失时 connect 悄悄创建一个空库，也禁止向源库写入。
with closing(sqlite3.connect(source.resolve().as_uri() + "?mode=ro",
uri=True)) as src:
# 不先 exists 再打开：创建与“必须不存在”由一次系统调用共同完成。
# 竞争方先创建时，这一步直接抛 FileExistsError，未取得所有权就不清理。
fd = os.open(target, os.O_CREAT | os.O_EXCL | os.O_RDWR, 0o600)
owned = os.fstat(fd)
completed = False
try:
# 目标必须是刚预留的文件；mode=rw 不允许丢失后重新创建。
with closing(sqlite3.connect(target.resolve().as_uri() + "?mode=rw",
uri=True)) as dst:
src.backup(dst) # SQLite 提供的一致性备份接口
completed = True
finally:
os.close(fd) # Windows 下也要先关闭本函数的句柄再尝试删除。
if not completed:
try:
current = target.lstat()
if (current.st_dev, current.st_ino) == (owned.st_dev, owned.st_ino):
target.unlink() # 只清理本次创建且身份仍匹配的失败目标。
except FileNotFoundError:
pass
def check_target_protection(source: Path, root: Path) -> None:
original_source = source.read_bytes()
existing = root / "existing.sqlite"
existing.write_bytes(b"keep-existing-target")
try:
copy_database(source, existing)
raise AssertionError("已有目标必须拒绝覆盖")
except FileExistsError:
pass
assert existing.read_bytes() == b"keep-existing-target"
# 确定性竞争：在本函数即将申请排他创建的瞬间，让另一个创建者先占位。
# 这不依赖线程时序，重复运行也会覆盖相同的竞争窗口。
raced = root / "raced.sqlite"
real_open = os.open
def competitor_wins(name, flags, mode=0o777):
if Path(name) == raced:
with raced.open("xb") as competing_file:
competing_file.write(b"owned-by-another-creator")
return real_open(name, flags, mode)
with patch("os.open", side_effect=competitor_wins):
try:
copy_database(source, raced)
raise AssertionError("竞争创建的目标必须拒绝覆盖")
except FileExistsError:
pass
assert raced.read_bytes() == b"owned-by-another-creator"
# 已取得目标所有权后才发生备份错误：只删除本次的残留，损坏源仍保留。
invalid = root / "invalid.sqlite"
invalid.write_bytes(b"this is not a SQLite database")
failed = root / "failed.sqlite"
try:
copy_database(invalid, failed)
raise AssertionError("损坏源应导致备份失败")
except sqlite3.DatabaseError:
pass
assert not failed.exists()
assert invalid.read_bytes() == b"this is not a SQLite database"
assert source.read_bytes() == original_source
print("保护检查通过：已有目标、竞争创建、失败清理、源库保留")
def main() -> None:
# 临时目录只用于教学；退出时会删除，不能当持久备份仓库
with TemporaryDirectory(prefix="restore-demo-") as directory:
root = Path(directory)
source = root / "source.sqlite"
backup = root / "backup.sqlite"
restored = root / "restored.sqlite"
con = sqlite3.connect(source)
try:
with con:
con.execute("CREATE TABLE notes(id INTEGER PRIMARY KEY, body TEXT NOT NULL)")
con.executemany("INSERT INTO notes VALUES(?, ?)", [
(1, "第一条"), (2, "第二条")
])
finally:
con.close()
expected = rows(source)
copy_database(source, backup)
copy_database(backup, restored)
actual = rows(restored)
if actual != expected:
raise RuntimeError("恢复数据与源快照不一致")
check = sqlite3.connect(restored)
try:
integrity = check.execute("PRAGMA integrity_check").fetchone()[0]
finally:
check.close()
if integrity != "ok":
raise RuntimeError("完整性校验失败：" + integrity)
print("恢复成功：2 条记录，完整性检查 ok")
check_target_protection(source, root)
if __name__ == "__main__":
main()
```
```bash
python backup_and_restore.py
```
Linux 如只提供 python3，则使用 python3 backup_and_restore.py。预期先输出“恢复成功：2 条记录，完整性检查 ok”，再输出“保护检查通过：已有目标、竞争创建、失败清理、源库保留”，退出码为 0。任意比较失败会抛异常并返回非零码，CI 会识别失败。
逐段解析：rows 按稳定顺序读取，避免仅因数据库无序返回造成误判；copy_database 通过排他创建取得一个新目标，防止把恢复演练写进现有数据库；backup 通过数据库 API 复制，不假设底层文件可以直接拷贝；最后既比对业务记录，也做完整性检查。完整性检查只证明数据库结构一致，不证明业务语义正确，所以两种检查都保留。
“先检查不存在，再打开文件”有一个时间窗口：检查后可能被另一个进程创建，而普通 sqlite3.connect 会打开那个已有数据库。这里用 os.open 的 O_CREAT 与 O_EXCL 一起完成检查和创建；目标此前存在或者竞争方先创建，都会抛 FileExistsError。创建失败发生在清理逻辑之前，因此不会把别人的目标误删。成功创建后保存文件身份；备份失败时关闭连接与预留句柄，再确认路径仍指向自己的文件才删除。[Python 文件打开标志](https://docs.python.org/3/library/os.html#os.open)
这段代码的目录合同是：目标位于应用控制的私有恢复目录，其他任务可以竞争创建同名文件，但不能随意删除、替换已经预留的文件。SQLite 仍按路径打开数据库，这不是在恶意用户可任意改写目录的环境中防御路径替换的沙箱。零六〇〇在 POSIX 上请求仅所有者读写，Windows 目录访问仍需由 ACL 控制。源连接使用只读 URI，目标连接使用必须已存在的读写 URI；恢复失败不会把源当作写入目标。
保护实验分别覆盖两种失败阶段：已有目标与竞争占位都发生在取得目标所有权之前，原字节必须保留；损坏源导致 backup 报错发生在预留之后，本次残留必须删除，而损坏源和有效源的字节均不变。竞争用标准库 patch 在排他创建前插入另一个创建者，固定了发生顺序，不把偶然通过的一次线程调度当作竞争验证。这里实际验证的是临时目录中的 SQLite 教学数据与确定性竞争窗口，没有运行真实业务库恢复或跨主机灾备。
## 放入 CI 的最小配置
下面是完整的 .github/workflows/verify.yml 示例，前提是仓库根目录已有上述 Python 文件。Action 的固定提交在采用时应核对来源和更新策略；示例展示固定版本的方式，不能把教材永久当依赖更新来源。
```yaml
name: Verify restore drill
on:
push:
pull_request:
permissions:
contents: read
jobs:
verify:
runs-on: ubuntu-24.04
timeout-minutes: 5
steps:
- name: Checkout source
uses: actions/checkout@11bd71901bbe5b1630ceea73d27597364c9af683
- name: Verify backup and restoration
run: python3 backup_and_restore.py
```
on 决定触发事件；permissions 把默认仓库权限缩到读取；timeout-minutes 防止卡住占用执行器；run 的非零退出码阻止通过。这个流程只有离线验证，没有发布动作。真实项目再加入对应包管理器的冻结安装、类型检查、构建与评测；需要密钥的发布阶段应独立设置环境权限与审核策略。
## 常见错误与排障
“备份能生成但恢复失败”要检查数据库版本、扩展、角色权限、字符编码和备份是否完整，不要先覆盖原环境反复试。恢复演练始终指向新实例、新目录或隔离数据库。先验证 schema、关键行数与业务抽样，再测试应用能否读写。
回滚后仍故障，检查配置、迁移、缓存、队列消息格式和 AI 索引版本，而不仅是镜像。部署脚本如果忽略退出码，可能出现前一步失败、后一步仍切流的假成功；发布要有明确成功条件与停止点。监控恢复后也要观察一段代表性流量，不能只看一次健康接口。
## 从一次提交到一个可回退版本的完整路径
前端项目的构建产物往往是静态文件，后端项目还包含运行时依赖和数据库行为，AI 应用又增加模型、提示和检索版本。因此发布单位不应只是一条 Git 提交，而是一份明确的版本清单。清单让你知道服务镜像、前端资源、数据库迁移、提示配置和索引分别属于哪个可兼容组合。
一次典型流程是：拉取确定提交，安装锁定依赖，执行必要检查，构建制品，保存制品摘要与报告，在测试环境验证同一制品，再按策略切换生产流量。每个阶段失败都应停止后续依赖动作。不能在测试失败后继续构建并发布，也不能因为构建成功就忽略评测或数据库兼容性。
同一份制品在不同环境只应替换允许的运行配置，而不是重新下载一套可能变化的依赖。前端某些配置在构建期注入，修改后会改变制品；后端环境变量多在运行时读取。设计发布清单时应区分这两类配置，避免把“只是换个地址”误当成没有重新构建风险。
制品摘要证明对应内容一致，不证明内容正确或安全。你仍需知道它由哪个流程构建、依赖来自哪里、哪些检查通过。构建日志不能包含密钥，制品也不应携带本地凭据。将来源和检查结果与摘要关联，比只给镜像打一个 latest 标签更能支持审计与回滚。
## 流水线的参数、权限和并发规则
CI 运行器是能执行仓库代码的计算环境，因此它的凭据暴露面要谨慎设计。外部 PR 的代码、依赖安装脚本和构建脚本都可能运行任意命令。只读校验任务不需要生产写权限；发布任务才申请最小的部署身份，并绑定允许的分支、环境和工作流条件。
依赖锁保证解析结果稳定，但还要使用对应包管理器的冻结安装模式，避免 CI 自动改锁或解析另一组版本。缓存可以减少下载时间，但缓存 key 应包含运行时和锁文件信息。缓存命中不应绕过完整性检查；缓存错了可以重建，不能把它作为不可替代的制品来源。
并发发布需要规则。两个提交同时部署时，较旧任务可能更晚完成并覆盖较新版本。可以让同一环境串行部署，或者在切换前比较预期版本。取消旧流水线也需要理解动作阶段：构建阶段取消通常简单，数据库迁移或流量切换中途取消可能留下部分状态，必须依靠部署状态记录恢复。
超时要覆盖整项任务，并给出可诊断的结束原因。把日志最后一行当成成功标志不可靠，程序退出码和检查结果才是流程接口。长耗时评测可以拆成快速回归与发布评测，但拆分后要确保真正发布依赖完整结果，而不是只依赖快检绿色图标。
## 应用回滚的真正前提是数据仍兼容
数据库结构更新最危险的地方在于它会影响同时运行的新旧实例。滚动更新期间，新版本可能已写入新格式，旧版本还在读。即使只部署单机，也可能有旧队列任务或用户长连接仍在处理旧格式。迁移设计需要考虑这段重叠窗口，而不是假设所有程序瞬间一起切换。
扩展阶段只增加兼容结构，例如新增可空字段或新表，让旧程序继续工作。切换阶段让新程序逐步读写新结构，必要时双写并核对。收缩阶段确认旧程序和旧任务不再依赖旧字段，才删除历史结构。三个阶段可以分多个发布完成，换来明确回滚窗口。
双写也不是免费保险。两次写入如果不在同一事务中，可能一边成功一边失败；跨数据库时更需要对账和修复。回填数据时应分批、可重试、记录进度，并避免长事务阻塞正常业务。回填完成不能只看任务退出，还要检查空值数量、约束和抽样一致性。
AI 索引迁移与此相似。换 embedding 模型可能改变维度或语义空间，不能把新向量混进旧索引后仅回滚应用。可以建立新索引、离线构建、对照评测、切换引用并保留旧索引一段时间。原始文档、切块规则和权限元数据需要一起版本化，否则无法可靠重建。
## 第二个完整示例：一次可兼容的数据库扩展
保存为 migration_demo.py，Python 3.12，仅标准库和临时 SQLite 数据库。它不连接生产系统。例子演示新增字段后，旧读法仍然可用，新读法在回填前后都能得到正确结果。
```python
# migration_demo.py
import sqlite3
from pathlib import Path
from tempfile import TemporaryDirectory
def old_reader(connection):
return connection.execute("SELECT id, body FROM notes ORDER BY id").fetchall()
def new_reader(connection):
return connection.execute(
"SELECT id, COALESCE(search_text, body) FROM notes ORDER BY id"
).fetchall()
with TemporaryDirectory(prefix="migration-demo-") as folder:
connection = sqlite3.connect(Path(folder) / "demo.sqlite")
try:
with connection:
connection.execute("CREATE TABLE notes(id INTEGER PRIMARY KEY, body TEXT NOT NULL)")
connection.executemany("INSERT INTO notes VALUES(?, ?)", [
(1, " 退款进度 "), (2, "物流查询")
])
before = old_reader(connection)
with connection:
connection.execute("ALTER TABLE notes ADD COLUMN search_text TEXT")
assert old_reader(connection) == before
assert new_reader(connection) == before
with connection:
connection.execute("UPDATE notes SET search_text = trim(body) WHERE search_text IS NULL")
assert old_reader(connection) == before
assert new_reader(connection) == [(1, "退款进度"), (2, "物流查询")]
missing = connection.execute(
"SELECT count(*) FROM notes WHERE search_text IS NULL"
).fetchone()[0]
print("旧程序仍兼容，待回填记录数", missing)
finally:
connection.close()
```
执行 python migration_demo.py，预期输出待回填记录数为零。旧读法保留原始正文，新读法使用清洗后的检索字段。COALESCE 让新代码在回填尚未完成时使用旧字段，因此扩展和回填可以分开进行。这只是兼容性机制演示，不代表所有数据库的 DDL 都具有相同锁与事务行为。
接着将新查询改成直接返回 search_text，不使用回退，再把回填步骤移到读操作之后，观察新程序会拿到空值。这就是第三个故障实验：结构已经存在不代表数据已经准备好。程序应该对迁移中状态有明确语义，而不是把空值一律当作业务数据缺失。
## 备份不只是文件，还包含恢复所需的依赖
恢复数据库需要相容的数据库版本、扩展、角色、权限与配置。恢复应用还可能需要对象存储中的附件、加密密钥、外部回调配置和索引来源。备份清单应围绕“能恢复用户任务”制定，不只是围绕“哪个目录容易复制”。
恢复环境应默认隔离，避免恢复出来的旧任务立即向真实用户发送邮件、回调或付款请求。可以禁用外部写工具、替换出口配置，并使用专门的测试身份验证读写。恢复数据后先看基本完整性，再跑关键业务检查，最后才考虑切换真实流量。
RPO 与 RTO 需要可测量。若要求最多丢失十五分钟数据，每天一次全量备份显然不够，可能需要日志归档或增量机制。若要求一小时恢复，而实际下载备份需要两小时，备份频率再高也无法满足恢复时长。容量增长会改变恢复时间，因此演练要用接近实际规模的数据。
保留策略需要考虑错误发现延迟。某次逻辑错误可能几天后才被发现，如果所有备份都已被新状态覆盖，就没有干净恢复点。多时间层级保留与独立副本有助于应对这类问题，但也增加敏感数据存储范围，应配合加密、访问控制和删除策略。
备份验证可以分层进行：每天检查任务状态、文件完整性和可读取性，定期恢复到隔离环境并运行业务校验，再定期演练人员、权限和流程是否可用。自动化能减少重复工作，但不能替代“谁能在事故时取得密钥和恢复权限”的组织安排。
## 为什么恢复前要先保护事故现场
事故发生时，第一步不一定是还原。应先限制继续错误写入，保存必要日志、当前数据库状态和版本信息，再判断影响范围。直接覆盖现场可能丢失对账线索，也可能把恢复后需要补回的有效业务数据一起删掉。
如果新版本写错某个派生字段，但原始记录仍完整，前向修复往往比整库恢复更合适。如果误删关键数据，则需要从备份中提取缺失部分或进行时间点恢复。选择取决于错误类型、业务一致性和可接受损失，不能把“回滚”作为所有事故的同一个按钮。
恢复完成后，需要确认外部系统和本地状态一致。例如支付服务已成功处理某笔交易，但数据库恢复到了处理前，本地若重新发起会有重复风险。应使用外部业务标识查询并对账，保留原操作编号。数据库恢复的是本地状态，不会让现实世界的邮件、付款和用户操作倒流。
## 综合练习：给恢复演练增加业务校验
基于原来的 backup_and_restore.py，增加一个业务规则：notes 的正文不能为空，而且 id 唯一。数据库完整性检查只能说明内部结构可读，不能识别所有业务错误。练习需要在结构正常但正文为空时判失败。
完整参考实现
将下面函数加入原备份脚本，并在恢复后调用 validate_business(restored)。它依赖原脚本已经导入的 sqlite3 与 Path；也可以连同这两个 import 单独保存后对明确的演练数据库调用。
```python
def validate_business(path: Path) -> None:
connection = sqlite3.connect(path)
try:
total, distinct_ids = connection.execute(
"SELECT count(*), count(DISTINCT id) FROM notes"
).fetchone()
empty = connection.execute(
"SELECT count(*) FROM notes WHERE body IS NULL OR trim(body) = ''"
).fetchone()[0]
if total != distinct_ids:
raise RuntimeError("业务主键不唯一")
if empty:
raise RuntimeError(f"存在 {empty} 条空正文")
finally:
connection.close()
```
在源库插入一条只有空格的正文后再备份恢复，数据库 PRAGMA integrity_check 仍可能为 ok，但 validate_business 必须失败。恢复验证因此需要同时保留结构检查和业务合同。正式系统还可以检查订单状态转移、余额对账与附件引用，具体规则来自业务模型，不应由通用备份工具随意猜测。
参考答案
在 copy_database(source, backup) 后另开源库连接插入第三条，保持 expected 是之前的两条。恢复后 actual 等于 expected，证明恢复的是备份时状态；第三条不在这份备份中，若没有日志或增量备份就无法靠它恢复。预创建 restored 会触发 FileExistsError，体现恢复目标必须明确而且受保护。生产中把时间、快照位置和演练报告持久记录。