GenarrativeAI/Genarrative
10
0
Fork 0
Code Issues 20 Pull Requests 8 Actions Packages Projects 1 Releases Wiki Activity

实现 Gitea CI 编译缓存自动刷新与定向清理

Browse Source 
基于固定 master 提交构建全新缓存镜像，并在成功验证后保留当前版和回滚版。
增加任务领取网关，在不打断运行中 CI 的前提下切换镜像。
补充宿主维护服务、部署说明、共享记忆和定向行为测试。
修复构建测试在 Windows 与 WSL 下并发写入日志的竞态。
This commit is contained in:
孔令弘
2026-09-22 07:37:45 +00:00
parent 292e42cf5e
commit 32caa6914b
17 changed files with 1915 additions and 43 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.gitea/workflows/project-ci.yml
+3
View File
@@ -524,6 +524,9 @@ jobs:
- name: Install npm dependencies
run: bash scripts/ci-npm-ci-with-retry.sh
- name: Validate CI cache maintenance behavior
run: python3 -m unittest discover -s scripts -p 'test_gitea_cache_*.py'
- name: Run repository checks
run: npm run check:repository-ci
deploy/container/README.md
+46 -2
View File
@@ -89,7 +89,7 @@ runner 配置保留原 `ubuntu-latest` 映射，`genarrative-ci` 继续映射到
2. 执行 `load-runner`，确认该 ID 已进入 runner 内层 Docker。
3. 确认没有活跃 job，将当前 runner config 备份到仓库外的受控位置；备份不得进入 Git，也不得在文档或日志中回显注册信息。
4. 增加或替换 `genarrative-ci` 的精确 `docker://<Image ID>` 映射，然后执行 `docker restart --timeout 660 gitea-runner`。
5. 重跑真实 PR 的四个 CI job；全部通过且隔离边界复核完成后，才能清理旧镜像。
5. 用真实 CI 的九个 job 验证新镜像；全部通过且隔离边界复核完成后，才能清理旧镜像。
`docker restart --timeout 660` 只提供容器停止宽限，不是 Runner drain API；rootless DinD 的 supervisor 可能与 runner 同时停止内层 dockerd。重启前必须同时确认 Gitea 没有 `in_progress` run 且内层 `docker ps` 为空，不能依赖该 timeout 等待活跃 job。
@@ -97,9 +97,15 @@ runner 配置保留原 `ubuntu-latest` 映射，`genarrative-ci` 继续映射到
### Rust 测试组编译对象快照
自动维护由宿主 systemd timer 调用 `scripts/maintain-gitea-rust-cache.py`，只管理 Gitea CI 测试镜像，不修改 Jenkins、生产发布、本地开发或客户端发行构建。维护器固定最新 master SHA，**不要求该提交事先全绿**；编译/镜像校验失败则继续使用旧版。任务串行执行，编译输入未变化时跳过，CI 繁忙时延后预热；每次只取当时最新 master，不为中间提交建立队列。构建过程中固定 SHA，不混入后续提交。
切换先通过专属入口阻断新的 FetchTask，确认已转发的领取请求全部收到完整上游响应，再确认全局 runner 不 busy、Gitea 没有 in_progress run、内层 Docker 没有容器。有任务即恢复领取并延后，不停止任务。切换后等待使用该 Image ID 的完整真实 master push CI 通过，才允许下一次升级及定向清理；不会自动重跑失败用例或为了验收额外触发整轮 CI。首次接管的历史镜像默认不归自动清理管理。
维护状态、凭据、归档和配置备份保存在仓库外。当前版、回滚版、待验证候选、它们的基础镜像及容器引用的镜像均受保护。清理只针对维护器登记的专属 tag、完整 Image ID 和专用目录中的归档；禁止全局 prune。API、构建、验证或空闲检查失败时保留现役镜像与回滚资料，不以失败重跑制造全绿结果。
`scripts/build-gitea-rust-cache.sh` 在已验证的 job 镜像上生成候选镜像，覆盖 AGC Rust 两条 lane、crates、agent-run smoke、Backend 和 Native shell 的桌面壳测试。Native shell 的 release build smoke 显式清空两个 wrapper，不消费测试快照。固定 sccache `0.18.0` 的 Linux x64 musl 归档并校验 SHA-256，维护者从 origin/master 的确定提交预热编译对象，PR job 没有生成/发布公共快照的权限。
基础镜像必须不含 `/opt/genarrative-ci/rust-cache`；脚本在拉取源码、下载工具和预热前执行只读、断网检查，发现已有对象快照就拒绝构建。不能在旧缓存镜像上删除目录再叠加新快照，删除操作不会释放旧镜像层。切换且真实 CI 验证通过后，人工定向清理更旧的缓存镜像与导出归档，保留当前版、一个回滚版及运行中 CI 使用的版本；对象缓存上限不覆盖这些宿主文件，不使用全局 prune。
基础镜像必须不含 `/opt/genarrative-ci/rust-cache`；脚本在拉取源码、下载工具和预热前执行只读、断网检查，发现已有对象快照就拒绝构建。不能在旧缓存镜像上删除目录再叠加新快照，删除操作不会释放旧镜像层。切换且真实 CI 验证通过后，维护器定向清理自己登记的更旧缓存镜像与导出归档，保留当前版、一个回滚版、所需基础镜像及容器引用的版本；接管前的试验镜像/归档仍由维护者确认后人工清理。对象缓存上限不覆盖这些宿主文件，不使用全局 prune，也不清理其它构建的 Docker build cache。
```bash
bash scripts/build-gitea-rust-cache.sh genarrative/gitea-project-ci:20260920.2 genarrative/gitea-project-ci:rust-cache-candidate
@@ -108,6 +114,44 @@ bash scripts/gitea-ci-job-image.sh export /仓库外受控路径/ci-rust-cache.t
bash scripts/gitea-ci-job-image.sh load-runner genarrative/gitea-project-ci:rust-cache-candidate
```
自动维护调用构建脚本时增加第三个参数 `<完整 master SHA>`。脚本确认该 SHA 是抓取到的 master 祖先，源码和 `ci-rust-cache.sh` 均来自该提交；两参数人工调用仍默认使用最新 master。基础镜像的 `revision` 由同一份 build context 清单计算；输入变化时维护器先生成新的无对象缓存基础镜像。Rust 输入指纹保守包含代码、配置、资源、脚本和内嵌 skill，仅排除一般 `docs/` 及几个根说明文档，`docs/openapi/` 始终参与。
### 自动维护首次部署与恢复
首次部署在本变更合入 master 后进行。维护脚本安装到 `/opt/genarrative-ci-cache/scripts/`；运行状态、源码专用 clone、日志和归档放在 `/var/lib/genarrative-ci-cache/`，token 放在 `/etc/genarrative-ci-cache/api-token`（root 所有、0600），均不进入 Git。复制 `gitea-ci-cache.config.example.json` 为该目录的 `config.json` 并核实 runner ID；当前全局 runner 的只读 API 需要管理员账号的 `read:admin`，仓库 Actions/日志读取需要对应的只读权限，不能复用只有普通仓库权限的 token。维护器没有写 Gitea API 的调用。
领取入口由 `gitea-runner-fetch-gate.compose.yml` 启动，使用已验证且含 Python 3 的无对象缓存 CI Image ID（`GITEA_FETCH_GATE_IMAGE`）。它只连接现有 `gitea-actions` 内部网络，不发布宿主端口、不挂 Docker socket；仅转发 `/api/actions/` RPC，控制 socket 位于独立私有目录 `/var/lib/genarrative-ci-cache-gate/`。该目录必须由 root 持有且权限为 0700。普通 job 无法访问控制 socket。部署前先运行下述测试，不直接启用 timer。
```bash
python3 -m unittest discover -s scripts -p 'test_gitea_cache_*.py'
install -d -m 700 /etc/genarrative-ci-cache /var/lib/genarrative-ci-cache /var/lib/genarrative-ci-cache-gate
install -d /opt/genarrative-ci-cache/scripts
install -m 755 scripts/maintain-gitea-rust-cache.py scripts/gitea-runner-fetch-gate.py /opt/genarrative-ci-cache/scripts/
install -m 600 deploy/container/gitea-ci-cache.config.example.json /etc/genarrative-ci-cache/config.json
# 由维护者写入只读 API token，并核实 config.json；不要在终端回显 token。
# 设置 GITEA_FETCH_GATE_IMAGE 为已验证基础镜像的完整 Image ID 后：
docker compose -f deploy/container/gitea-runner-fetch-gate.compose.yml up -d
```
**首次接入需要空闲维护窗口**：确认无活跃 CI 且暂停新 CI 触发，再备份现有 runner 配置和注册文件，在 runner 部署的 `GITEA_INSTANCE_URL` 及 `/data/.runner` 的 `address` 中改用 `http://gitea-runner-fetch-gate:8080`，保留其余注册字段。同时在 `/data/config.yaml` 的 `runner.envs` 中设置 `GENARRATIVE_GITEA_REPOSITORY_URL: "http://gitea:3000/GenarrativeAI/Genarrative.git"`，与维护器的 `repository_url` 一致，沿用现有 job 已可达的内部 Git 通道。此专用变量也覆盖旧 PR 的 checkout；不要用同名 GITHUB_SERVER_URL 环境变量代替，Runner 会再次覆盖它。按原流程重启并验证 runner 注册。
不能仅改磁盘文件却不让进程加载；维护器还会检查独立 checkout URL，并确认入口实际见到了当前容器本次启动后的 FetchTask 来源 IP。此一次接入不由维护器冒险猜测空闲，也不对运行中 CI 动手。实例 API 根地址仍使用配置中的 HTTPS Gitea 地址，不能指向只支持 runner RPC 的入口。当前 Project CI 没有 `uses:` 或直接依赖 `github.server_url` 的 API 请求；后续新增这类调用时需沿用独立真实 Git/API 地址，不能假定只支持 RPC 的入口也是通用 Gitea 地址。
```bash
# --apply 缺省时只检查 API、runner、master 和入口路径，不修改配置或镜像。
python3 /opt/genarrative-ci-cache/scripts/maintain-gitea-rust-cache.py --config /etc/genarrative-ci-cache/config.json
install -m 644 deploy/systemd/genarrative-ci-cache.service deploy/systemd/genarrative-ci-cache.timer /etc/systemd/system/
systemctl daemon-reload
systemctl enable --now genarrative-ci-cache.timer
journalctl -u genarrative-ci-cache.service -n 50
```
timer 在上次执行结束后约 5 分钟再次检查，文件锁防止人工与定时执行重叠。构建日志位于状态目录 `artifacts/<SHA>/build.log`。同 SHA 构建失败后不每 5 分钟重复消耗资源；新 master 到来后自动重试，也可修复环境后显式运行 `--apply --retry`。切换事务及暂停归属先落状态文件；`ExecStopPost --resume` 恢复本维护器暂停的领取，下次执行再收敛中断的切换。其它人暂停的入口或 disabled runner 不由维护器擅自恢复。
候选切换后的验收读取真实 **master push** 的完整九个 job，要求全部 success、每个 job 均使用目标 Image ID，六个 Rust job 有启用缓存、正命中数和零缓存错误。测试失败、旧 PR 缺 prepare、混用镜像或缺日志均不清理旧版，也不伪造“缓存已验收”。仍保留源代码失败需要修复的原始结果。
入口遇到“已发送 FetchTask，但上游响应未完整结束”会持久化 `uncertain` 并拒绝继续领取/自动切换；不会因为客户端的 5 秒超时就认定服务端事务已回滚。其它 RPC 和任务上报仍继续转发。维护者需先核实 Gitea 在途领取事务与该 runner 的任务全部收敛，在维护窗口停止入口，清理它的 `uncertain` / `inflight` 标记后再启动并恢复领取。禁止自动删这些标记绕过屏障。停用自动维护先 `systemctl disable --now genarrative-ci-cache.timer`；不要为了停 timer 停止运行中的 CI 容器。已接入的领取入口继续运行，不影响普通 CI。
预热容器上限为 4 核、12 GiB，移除 capabilities，不挂宿主目录/socket，也不注入 Git/OSS/Jenkins 凭据。源码通过 `git archive` 复制，当前工作区、ignored 文件和 `.git` 不进入容器。最终从原镜像重新组装，仅复制 `/opt/genarrative-ci/rust-cache` 的 sccache、对象和来源元数据，不提交含源码/target 的预热容器；镜像本身的下载缓存与工具链校验保持原样。
快照由固定 Image ID 分发，每个 job 仅修改容器自己的写时复制层，缓存上限 4 GiB，构建末尾输出实际对象体积，结束后不回传。`ci-rust-cache.sh prepare` 清空继承的 `SCCACHE_*` 远程配置，使用独立配置和 Unix socket；旧镜像、工具链不匹配或限时 wrapper 探测失败时使用直接 rustc，正式编译启用 sccache 的 server IO 错误回退。真实编译/测试失败保留非零退出码。`report` 输出命中统计并停止本 job daemon，分片日志输出独立编译耗时。sccache 0.18.0 的只读模式在 miss 后仍打包再拒绝写入，不能用它宣称零 miss 开销；普通 CI 继续只向容器层写入。
deploy/container/gitea-ci-cache.config.example.json
+12
View File
@@ -0,0 +1,12 @@
{
"api_url": "https://git.genarrative.world/git/api/v1",
"repository": "GenarrativeAI/Genarrative",
"repository_url": "http://gitea:3000/GenarrativeAI/Genarrative.git",
"clone_url": "https://git.genarrative.world/git/GenarrativeAI/Genarrative.git",
"runner_container": "gitea-runner",
"runner_api_path": "admin/actions/runners/2",
"token_file": "/etc/genarrative-ci-cache/api-token",
"state_dir": "/var/lib/genarrative-ci-cache",
"gate_socket": "/var/lib/genarrative-ci-cache-gate/gate.sock",
"gate_url": "http://gitea-runner-fetch-gate:8080"
}
deploy/container/gitea-runner-fetch-gate.compose.yml
+23
View File
@@ -0,0 +1,23 @@
# 只加入 runner 与 Gitea 共用的内部网络，不发布端口，不挂 Docker socket。
services:
gitea-runner-fetch-gate:
image: ${GITEA_FETCH_GATE_IMAGE:?Set a verified no-object-cache CI image ID with Python 3}
container_name: gitea-runner-fetch-gate
restart: unless-stopped
entrypoint: ['python3', '/app/gitea-runner-fetch-gate.py']
environment:
GITEA_RUNNER_UPSTREAM: http://gitea:3000
read_only: true
cap_drop: [ALL]
security_opt: [no-new-privileges:true]
cpus: 0.5
mem_limit: 256m
pids_limit: 128
volumes:
- /opt/genarrative-ci-cache/scripts/gitea-runner-fetch-gate.py:/app/gitea-runner-fetch-gate.py:ro
- /var/lib/genarrative-ci-cache-gate:/control
networks: [gitea-actions]
networks:
gitea-actions:
external: true
deploy/systemd/genarrative-ci-cache.service
+22
View File
@@ -0,0 +1,22 @@
[Unit]
Description=Refresh and retain trusted Gitea CI Rust cache images
After=network-online.target docker.service
Wants=network-online.target
Requires=docker.service
ConditionPathExists=/etc/genarrative-ci-cache/config.json
[Service]
Type=oneshot
User=root
WorkingDirectory=/opt/genarrative-ci-cache
UMask=0077
ExecStart=/usr/bin/python3 /opt/genarrative-ci-cache/scripts/maintain-gitea-rust-cache.py --config /etc/genarrative-ci-cache/config.json --apply
ExecStopPost=/usr/bin/python3 /opt/genarrative-ci-cache/scripts/maintain-gitea-rust-cache.py --config /etc/genarrative-ci-cache/config.json --resume
TimeoutStartSec=3h
TimeoutStopSec=60s
Nice=10
IOSchedulingClass=best-effort
IOSchedulingPriority=7
[Install]
WantedBy=multi-user.target
deploy/systemd/genarrative-ci-cache.timer
+11
View File
@@ -0,0 +1,11 @@
[Unit]
Description=Check for new Gitea CI compilation inputs
[Timer]
OnBootSec=5min
OnUnitInactiveSec=5min
RandomizedDelaySec=30s
Unit=genarrative-ci-cache.service
[Install]
WantedBy=timers.target
docs/project-memory/shared-memory/development-workflow.md
+2
View File
@@ -96,6 +96,8 @@ SpacetimeDB 任务统一先读取 `.codex/skills/genarrative-spacetimedb/SKILL.m
## Gitea CI 依赖闭合
Gitea Rust 缓存自动维护由宿主 `genarrative-ci-cache.timer` 调用 `scripts/maintain-gitea-rust-cache.py`：最新 master 固定 SHA 构建，不要求源提交先全绿；输入未变跳过，失败保留现役版。只有专属 FetchTask 入口已暂停且在途请求完整结束、全局 runner/API/内层 Docker 均空闲时才切换；随后真实 master push 的完整九个 job 与六组缓存统计通过，才清理自己登记的过期镜像/归档。保留当前、一个回滚版、所需基础镜像和容器引用。首次部署需要空闲窗口和只读管理员 API 凭据，操作入口见 `deploy/container/README.md`；合并代码不会自动启用宿主服务。范围仅 Gitea CI 测试镜像，不改 Jenkins、生产、本地或发行构建。
AGC Rust 两条 lane、crates、smoke、Backend 和桌面壳测试使用镜像内可信 sccache 对象快照；Native shell release step 显式清空双 wrapper，前端/repository checks 不启用。维护者通过 `scripts/build-gitea-rust-cache.sh` 从远端 master 在限额、无宿主挂载的临时容器中按实际 cwd/profile/目标预热全部测试组，仅编译、不执行测试/应用；后端 workspace 与 spacetime-module 保持独立，AGC 的三个 cwd 入口之间清理预热 target，防止 fresh 判断漏产缓存键。最终镜像只追加 sccache、对象和来源元数据，不包含源码或 target。容量上限 4 GiB，不替代宿主旧镜像/归档清理。PR 只写当前容器层、不回传，不开放 Docker API/发布权限；继续禁用 incremental。`ci-rust-cache.sh` 在快照缺失、工具链不符或 wrapper 探测失败时直接编译，并隔离远程缓存配置和 daemon。分片日志记录编译耗时，收尾输出命中统计；两个 lane 的测试和前置检查不同，耗时差不是严格 A/B。线上存在活跃 CI 时不得重启 runner 或切换标签；全组启用前须刷新完整快照并逐组验证，详见开发运维文档。
`.gitea/workflows/project-ci.yml` 的客户端门禁拆成 lane 与功能 job，每个 job 只预热自己会构建的那几份依赖：`AI game creator shell Rust lane 1/2`、`lane 2/2` 各自预取一次 AGC 壳 manifest，并顺序运行两片 Rust bin 单测；`AI game creator shell Rust smoke` 同样只预取 AGC 壳 manifest（`agent-run` smoke 会用 `src-tauri/Cargo.toml` spawn `cargo run`），`AI game creator shell Rust crates` 预取 `server-rs/Cargo.toml` 与独立 crate，`Native shell tests` 预取桌面壳与 AGC 壳 manifest，`AI game creator shell web tests` 不触碰 Cargo，不预热。两条 Rust lane、smoke job 与 crates job 只用 cargo 与 node 内建模块，因此不执行 `npm ci`。两个被 `server-rs/Cargo.toml` 排除、且没有提交 `Cargo.lock` 的独立 crate（`agent-runtime-core`、`agent-runtime-orchestration`）只能在 `AI game creator shell Rust crates` 里用不带锁标志的 fetch。AGC 壳的 bin target 单测（约 2466 条）由 `apps/ai-game-creator-shell/scripts/run-rust-shell-test-shards.mjs` 编译后按 `--list` 名单分 4 片：每次分片调用用 `--shard-index=<i>` 只跑自己那片，片内保持 `--test-threads=1` 并使用独立 `TMPDIR`；两条 lane 之间并发，lane 内顺序运行两片，避免重复依赖预热和同一容器内多进程争抢。不要改回「一个 job 内多进程并行这几片」——同一容器里它们会争抢共享 `HOME`、target 目录与固定临时路径，实测比整套串行还慢。每个分片调用都会自校验「片并集等于全集且互斥」，因此改分片规则不会静默漏跑。Backend host workspace tests 使用 `cargo test --locked --workspace --exclude spacetime-module --no-fail-fast`，避免 `spacetime-module` 的 `spacetime-types` feature 统一污染普通领域 crate 的 host 测试；随后单独执行 `cargo test --locked -p spacetime-module --no-fail-fast`，由 `spacetime-module/src/active.rs` 在 host 测试构建期间提供仅测试期的 SpacetimeDB ABI 链接支持，使该 crate 的纯单元测试也纳入 Backend 门禁。`spacetime-module` 的 reducer / procedure 运行时行为仍必须通过真实 SpacetimeDB runtime/integration harness 验证，host 链接支持不得被当作运行时替身。Backend 另外执行 `cargo check --locked -p spacetime-module` 验证模块源码。AGC 壳检查还会运行 `platform-llm` 与 `shared-contracts` 的 server-rs workspace 测试，这些命令以及 AGC 壳测试必须带 `--locked`，避免在测试阶段重新解析 registry index；锁文件发生变化时应先更新受信任 CI 镜像缓存，再重跑门禁。
docs/project-memory/shared-memory/pitfalls.md
+2 -1
View File
@@ -5950,5 +5950,6 @@ Cocos Creator 根目录由 `package.json.creator.version` 与普通 `assets/`
- 处理：预热使用已核实的 Gitea 路径 `/workspace/GenarrativeAI/Genarrative`，并与分片运行器一样从 AGC `src-tauri` 启动 Cargo；保存 `workspace.txt`，路径不符时回退直接编译。wrapper 放在容器内固定路径，daemon 状态与 Unix socket 仍使用随机私有目录；预热环境与 workflow 的 Cargo 环境由定向契约测试核对。不要为命中率随意增加 `RUSTFLAGS`、改写源码路径或恢复共享可写 target。
- 验证：相同源码、资源上限和独立干净 target 下分别记录无缓存、冷缓存、热缓存的编译耗时和 hit/miss；只有真实热命中有净收益才切换候选镜像。PR 的写入始终留在 job 容器层，公共快照仍由可信维护流程生成。
- 统计：job 私有 daemon 设置 `SCCACHE_IDLE_TIMEOUT=0`，由 `report` 显式停止；最终测试 bin 的不可缓存编译或测试可能超过一分钟，短 idle timeout 会让 daemon 提前退出，结尾查询启动新 daemon 后误报零次请求。容器销毁仍会回收该 job 的全部进程。
- 磁盘：快照构建拒绝含 `/opt/genarrative-ci/rust-cache` 的基础镜像，始终从无对象缓存的镜像重建；容器内删除旧对象不能释放 Docker 底层。对象缓存容量上限不涵盖宿主旧镜像及导出归档，切换验证后按运维文档人工保留当前版和一个回滚版，同时保护运行中 CI 使用的镜像。
- 磁盘：快照构建拒绝含 `/opt/genarrative-ci/rust-cache` 的基础镜像，始终从无对象缓存的镜像重建；容器内删除旧对象不能释放 Docker 底层。对象缓存容量上限不涵盖宿主旧镜像及导出归档。自动维护只回收自己预先登记的 Image ID/tag 和专属归档，保留当前、一个回滚版、基础镜像及所有容器引用；内层按 ID 导入的镜像可能没有 tag，不能只按 tag 判断已清理。接管前历史试验版本仍需人工确认。
- 自动切换：Gitea 1.26.4 的 disabled 检查与 FetchTask 创建任务事务之间没有共同锁，Runner 2.0.0 的 fetch_timeout 也不能保证服务端事务回滚。不能以“禁用后睡几秒”或反复 docker ps 冒充领取屏障。专属入口阻断新 FetchTask，并完整读取在途响应；未知完成状态持久化 uncertain 并停止自动切换。Runner 的 .runner 文件还会因 label 更新回写，mtime 不等于地址变更时间；要结合已配置地址、当前容器启动时间与入口实际见到的 FetchTask 来源确认路由已生效。
- 扩展：预热所有 Rust 测试组时保留各自 cwd、profile、features 和锁策略；同一临时 target 的 Cargo fresh 不代表不同 cwd 都已生成缓存键，AGC 提示词契约、分片和 smoke 切换入口前清理预热 target。不要把 workspace 与 spacetime-module 合并成一次编译；Native shell release step 清空双 wrapper，避免将测试缓存扩展成发布缓存。当前 sccache 0.18.0 的 READ_ONLY 在 miss 后仍打包产物并产生 cache write error，不适合用来承诺“未命中无开销”。
docs/【开发运维】本地开发验证与生产运维-2026-05-15.md
+4 -2
View File
@@ -334,16 +334,18 @@ master 日常交付必须禁止直接 push，只允许经 PR 在最近一次 Pro
#### Rust 测试组的隔离编译缓存
每次生成快照必须使用不含对象缓存的原始 CI 基础镜像；构建脚本在拉取源码和预热前拒绝已存在 `/opt/genarrative-ci/rust-cache` 的基础镜像，避免重复叠加不可释放的旧对象层。切换并完成真实 CI 验证后，按 `deploy/container/README.md` 的保留规则人工定向清理旧缓存镜像及归档，保护运行中 CI 使用的版本。
每次生成快照必须使用不含对象缓存的原始 CI 基础镜像；构建脚本在拉取源码和预热前拒绝已存在 `/opt/genarrative-ci/rust-cache` 的基础镜像，避免重复叠加不可释放的旧对象层。宿主自动维护器 `scripts/maintain-gitea-rust-cache.py` 固定最新 master SHA 构建，不要求该提交事先全绿；源码或镜像校验失败保留旧版。切换后通过真实 master CI 才按 `deploy/container/README.md` 自动定向清理自己登记的旧镜像及归档，保护当前版、一个回滚版、所需基础镜像及全部容器引用；首次接管前的手工试验版本不自动认领。
AGC Rust 两条 lane、crates、agent-run smoke、Backend 和 Native shell 的桌面壳测试均启用 sccache。Native shell 的 release build smoke 显式清空两个 wrapper，保持发布构建原有 profile/features 与资源 staging；前端和 repository checks 不启用对象缓存。继续设置 `CARGO_INCREMENTAL=0`，不共享 target、不恢复 Actions 可写缓存。可信快照通过已有固定 Image ID 分发：镜像只增加固定版本的 sccache、编译对象和来源元数据，不包含源码、target 或凭据；容器写时复制层承接本 job 的新增对象，job 删除后丢弃，PR 没有 Docker API 或快照发布权限。该权限边界由 runner 基础设施保证，不能仅用 workflow 的分支条件替代。
维护者在受信任 checkout 中运行 `bash scripts/build-gitea-rust-cache.sh <已验证基础镜像> <候选镜像tag>`。脚本从 origin 获取 master 的确定提交，在无宿主目录挂载、无凭据且有 CPU/内存上限的临时容器中预热各组实际 Cargo 目标，只编译、不执行测试或应用。后端 workspace 仍排除 spacetime-module，再单独预热该模块，禁止为缓存合并 features。AGC 独立 crate 与插件沿用各自 manifest/锁文件策略；未提交锁文件的 crate 仍可能因新版本解析产生 miss。最终镜像从原基础镜像重新组装，只导出 sccache 对象，不提交预热容器。公共快照不接受 PR 上传，也不复用 Jenkins 发布缓存。工具链或系统依赖变化时重新生成快照，缓存命中仍由 sccache 的编译输入校验决定，不能省略 Cargo 构建。
维护者在受信任 checkout 中运行 `bash scripts/build-gitea-rust-cache.sh <已验证基础镜像> <候选镜像tag> [完整master-SHA]`。自动维护显式传入固定 SHA，脚本验证它为抓取到的 master 祖先，源码及缓存辅助脚本来自同一归档；人工省略 SHA 时默认最新 master。在无宿主目录挂载、无凭据且有 CPU/内存上限的临时容器中预热各组实际 Cargo 目标，只编译、不执行测试或应用。后端 workspace 仍排除 spacetime-module，再单独预热该模块，禁止为缓存合并 features。AGC 独立 crate 与插件沿用各自 manifest/锁文件策略；未提交锁文件的 crate 仍可能因新版本解析产生 miss。最终镜像从原基础镜像重新组装，只导出 sccache 对象，不提交预热容器。公共快照不接受 PR 上传，也不复用 Jenkins 发布缓存。工具链或系统依赖变化时重新生成快照，缓存命中仍由 sccache 的编译输入校验决定，不能省略 Cargo 构建。
各 Rust job 在编译前执行 `scripts/ci-rust-cache.sh prepare`：检查快照与 rustc 身份，隔离 sccache 配置和 daemon，限时探测真实 wrapper。旧镜像没有快照或探测失败时保留空 wrapper，输出 fallback 原因；缓存故障不得把真实编译/测试失败改成成功，也不允许重跑整个测试组掩盖失败。结束时 `report` 输出命中统计；分片日志单独记录 Cargo 编译耗时。对象缓存上限为 4 GiB，快照构建完成后输出实际体积；最终测试 bin 的链接仍须执行。全组开启时必须生成覆盖全部目标的新快照，不能把旧 AGC 单目标快照当作后端/桌面壳的预热验收。
候选镜像沿用 `gitea-ci-job-image.sh verify/export/load-runner` 验证和装载。现役 CI 正在运行时只允许准备候选镜像，不停止 job、不重启 runner、不切换标签；待确认无运行中的 job 后再按上述镜像更新顺序切换。验收分别记录独立 target 的无缓存、冷缓存和热缓存构建耗时及命中率，并检查失效/故障回退、两容器写入互不影响。未完成真实 CI 验证前不能宣称提速或推广到其它 job。
自动维护首次部署及故障恢复统一见 `deploy/container/README.md` 的“自动维护首次部署与恢复”。systemd timer 约每 5 分钟检查一次，持有文件锁，使用独立 clone/状态/归档目录；只管理 Gitea CI 测试镜像，不改变 Jenkins、生产、本地或客户端发行构建，Native shell release smoke 继续清空双 wrapper。自动切换使用专属 FetchTask 入口的暂停与在途计数屏障，不能将 Gitea disabled 或客户端超时当成事务完成证据；屏障收敛后还检查全局 runner busy、Gitea in_progress 和内层容器。首次接入入口必须安排空闲维护窗口，不在运行中的 CI 上试切；脚本合并不等于服务已在线启用。
Rust 对象 key 包含编译工作目录。预热必须使用已核实的 Gitea checkout 路径 `/workspace/GenarrativeAI/Genarrative`：后端、独立 crate、插件、桌面壳及提示词契约从仓库根目录启动 Cargo，AGC 分片从 `apps/ai-game-creator-shell/src-tauri` 启动，agent-run smoke 从 `apps/ai-game-creator-shell` 启动。切换 AGC 编译入口前清理该临时容器的 AGC target，防止 Cargo 的 fresh 判断跳过新 cwd 所需对象；CI 各 job 的 target 本来就独立。快照保存 `workspace.txt`，job 根路径不符时直接编译。不要仅设置 `SCCACHE_BASEDIRS` 就假定 Rust 可以跨 cwd 命中，也不为缓存改写 `RUSTFLAGS` 或源码路径语义。固定 sccache `0.18.0` 的基础设施错误码 `2` 会由 wrapper 回退执行本次 rustc；其它退出码原样返回，升级 sccache/Rust 时须复核此约定。
sccache `0.18.0` 还会 hash 大部分 `CARGO_*` 环境变量。因此 wrapper 固定在每个容器自己的 `/opt/genarrative-ci/rust-cache/rustc-wrapper`，prepare 重写 launcher 后才启用；daemon 状态与 socket 仍各自随机隔离。预热同步 workflow 的 `CARGO_INCREMENTAL`、`CARGO_HTTP_MULTIPLEXING`、`CARGO_NET_RETRY` 和 `CARGO_TERM_COLOR`，由现有 CI 契约测试验证一致；新增 Cargo 环境配置时须同步评估 cache key，不能只看快照文件是否存在。
scripts/build-gitea-rust-cache.sh
+21 -2
View File
@@ -5,6 +5,15 @@ set -euo pipefail
repo_root="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd -P)"
base_ref="${1:?usage: build-gitea-rust-cache.sh <verified-base-image> <candidate-tag>}"
candidate_tag="${2:?candidate image tag is required}"
requested_source_commit="${3:-}"
if [[ "$#" -gt 3 ]]; then
echo 'usage: build-gitea-rust-cache.sh <verified-base-image> <candidate-tag> [master-commit-sha]' >&2
exit 2
fi
if [[ -n "${requested_source_commit}" && ! "${requested_source_commit}" =~ ^[0-9a-fA-F]{40}$ ]]; then
echo 'master-commit-sha must be a complete 40-character SHA.' >&2
exit 2
fi
# 与实际 Gitea checkout 路径一致；Rust 对象 key 包含编译 cwd，不能随意换临时根。
workspace=/workspace/GenarrativeAI/Genarrative
[[ "${CI:-}" != true ]] || { echo 'Run on the trusted image builder, outside CI jobs.' >&2; exit 1; }
@@ -22,7 +31,17 @@ bash "${repo_root}/scripts/gitea-ci-job-image.sh" verify "${base_id}"
# 只归档远端 master 的确定提交；不复制当前工作区或本地凭据。
git -C "${repo_root}" fetch --no-tags origin refs/heads/master
source_commit="$(git -C "${repo_root}" rev-parse FETCH_HEAD^{commit})"
master_commit="$(git -C "${repo_root}" rev-parse FETCH_HEAD^{commit})"
if [[ -n "${requested_source_commit}" ]]; then
git -C "${repo_root}" cat-file -e "${requested_source_commit}^{commit}"
if ! git -C "${repo_root}" merge-base --is-ancestor "${requested_source_commit}" "${master_commit}"; then
echo "master-commit-sha is not contained in fetched master: ${requested_source_commit}" >&2
exit 1
fi
source_commit="$(git -C "${repo_root}" rev-parse "${requested_source_commit}^{commit}")"
else
source_commit="${master_commit}"
fi
work_dir="$(mktemp -d "${TMPDIR:-/tmp}/gitea-rust-cache.XXXXXX")"
container_id=''
cleanup() {
@@ -50,7 +69,7 @@ container_id="$(docker run --detach --cpus=4 --memory=12g --pids-limit=1024 \
docker exec "${container_id}" mkdir -p "${workspace}" /opt/genarrative-ci/rust-cache/objects
git -C "${repo_root}" archive "${source_commit}" | docker cp - "${container_id}:${workspace}"
docker cp "${work_dir}/snapshot/." "${container_id}:/opt/genarrative-ci/rust-cache/"
docker cp "${repo_root}/scripts/ci-rust-cache.sh" "${container_id}:/tmp/ci-rust-cache.sh"
docker exec "${container_id}" cp "${workspace}/scripts/ci-rust-cache.sh" /tmp/ci-rust-cache.sh
docker exec --interactive --workdir "${workspace}" "${container_id}" bash -s <<'WARM'
set -euo pipefail
rustc -vV > /opt/genarrative-ci/rust-cache/rustc.txt
scripts/gitea-ci-job-image.sh
+9 -33
View File
@@ -39,6 +39,7 @@ usage() {
cat <<'EOF'
用法：
bash scripts/gitea-ci-job-image.sh build
bash scripts/gitea-ci-job-image.sh revision
bash scripts/gitea-ci-job-image.sh verify [镜像引用]
bash scripts/gitea-ci-job-image.sh load-runner [镜像引用]
bash scripts/gitea-ci-job-image.sh export <归档路径> [镜像引用]
@@ -65,40 +66,15 @@ verify_image() {
command_name="${1:-}"
case "${command_name}" in
build)
image_revision="$(
revision)
# 与 build 的 IMAGE_REVISION 使用同一份输入顺序，用于维护器判断基础镜像是否过期。
(
cd "${repo_root}"
{
sha256sum \
deploy/container/gitea-ci-job.Dockerfile \
deploy/container/gitea-ci-job.Dockerfile.dockerignore \
deploy/container/gitea-ci-checkout.sh \
package.json \
package-lock.json \
apps/admin-web/package.json \
apps/ai-game-creator-shell/package.json \
apps/desktop-shell/package.json \
apps/mobile-shell/package.json \
apps/preview-deployer-web/package.json \
packages/image-canvas-core/package.json \
packages/image-canvas-react/package.json \
packages/shared/package.json \
tools/spine-json-export-validator/package.json \
apps/ai-game-creator-shell/src-tauri/Cargo.toml \
apps/ai-game-creator-shell/src-tauri/Cargo.lock \
server-rs/Cargo.toml \
server-rs/Cargo.lock \
apps/desktop-shell/src-tauri/Cargo.toml \
apps/desktop-shell/src-tauri/Cargo.lock
find server-rs/crates plugins/agc-*-editor/native/*-editor-bridge \
\( -name Cargo.toml -o -path 'plugins/agc-*-editor/native/*-editor-bridge/*' \) \
-type f -print0 \
| sort -z \
| xargs -0 -r sha256sum
} \
| sha256sum \
| awk '{ print $1 }'
)"
write_build_context_file_list | xargs -0 -r sha256sum | sha256sum | awk '{ print $1 }'
)
;;
build)
image_revision="$(bash "${BASH_SOURCE[0]}" revision)"
npm_lock_sha256="$(sha256sum "${repo_root}/package-lock.json")"
npm_lock_sha256="${npm_lock_sha256%% *}"
server_rust_lock_sha256="$(sha256sum "${repo_root}/server-rs/Cargo.lock")"
scripts/gitea-runner-fetch-gate.py
+270
View File
@@ -0,0 +1,270 @@
#!/usr/bin/env python3
"""Gitea Runner 领取屏障；控制 socket 仅供宿主缓存维护任务使用。"""
import http.client
import http.server
import json
import os
from pathlib import Path
import socketserver
import threading
import time
import urllib.parse
MAX_BODY = 32 * 1024 * 1024
HOP_HEADERS = {
"connection", "keep-alive", "proxy-authenticate", "proxy-authorization",
"te", "trailer", "transfer-encoding", "upgrade", "host", "content-length",
}
class Gate:
def __init__(self, directory):
self.directory = Path(directory)
self.directory.mkdir(parents=True, exist_ok=True)
self.lock = threading.Lock()
self.inflight = 0
self.last_fetch_peer = None
self.last_fetch_at = None
self.uncertain = (self.directory / "uncertain").exists()
if (self.directory / "inflight").exists():
self.uncertain = True
self.mark("uncertain")
self.paused = (self.directory / "paused").exists() or self.uncertain
def mark(self, name):
with (self.directory / name).open("w", encoding="ascii") as stream:
stream.write("1\n")
stream.flush()
os.fsync(stream.fileno())
self.sync_directory()
def sync_directory(self):
descriptor = os.open(self.directory, os.O_RDONLY | os.O_DIRECTORY)
try:
os.fsync(descriptor)
finally:
os.close(descriptor)
def snapshot(self):
return {"paused": self.paused, "inflight": self.inflight,
"uncertain": self.uncertain, "last_fetch_peer": self.last_fetch_peer,
"last_fetch_at": self.last_fetch_at}
def record_fetch(self, peer):
with self.lock:
self.last_fetch_peer = peer
self.last_fetch_at = time.time()
def control(self, action):
with self.lock:
if action == "pause":
self.mark("paused")
self.paused = True
elif action == "resume":
if self.uncertain:
return {**self.snapshot(), "error": "upstream completion uncertain; operator recovery required"}
(self.directory / "paused").unlink(missing_ok=True)
self.sync_directory()
self.paused = False
elif action != "status":
return {**self.snapshot(), "error": "unknown action"}
return self.snapshot()
def enter(self):
with self.lock:
if self.paused or self.uncertain:
return False
# 必须先落盘再转发；崩溃后不能把遗留的领取请求误认为已完成。
self.mark("inflight")
self.inflight += 1
return True
def leave(self, completed):
with self.lock:
self.inflight -= 1
if not completed:
self.uncertain = self.paused = True
self.mark("uncertain")
if self.inflight == 0 and not self.uncertain:
(self.directory / "inflight").unlink(missing_ok=True)
self.sync_directory()
def read_body(stream, headers):
"""解码 HTTP 请求；拒绝含糊 framing，不依赖下游连接关闭。"""
encodings = headers.get_all("Transfer-Encoding", [])
lengths = headers.get_all("Content-Length", [])
if encodings and lengths:
raise ValueError("ambiguous request framing")
if len(lengths) > 1 or len(encodings) > 1:
raise ValueError("duplicate request framing")
def exact(length):
data = stream.read(length)
if len(data) != length:
raise ValueError("incomplete request body")
return data
if not encodings:
length = int(lengths[0]) if lengths else 0
if not 0 <= length <= MAX_BODY:
raise ValueError("request too large")
return exact(length)
if encodings[0].strip().lower() != "chunked":
raise ValueError("unsupported transfer encoding")
body = bytearray()
while True:
line = stream.readline(8193)
if len(line) > 8192 or not line.endswith(b"\r\n"):
raise ValueError("invalid chunk header")
length = int(line.split(b";", 1)[0].strip(), 16)
if length < 0 or len(body) + length > MAX_BODY:
raise ValueError("request too large")
if length == 0:
trailer_size = 0
while True:
line = stream.readline(8193)
trailer_size += len(line)
if trailer_size > 8192 or not line.endswith(b"\r\n"):
raise ValueError("invalid request trailers")
if line == b"\r\n":
return bytes(body)
body.extend(exact(length))
if exact(2) != b"\r\n":
raise ValueError("invalid chunk terminator")
class Proxy(http.server.BaseHTTPRequestHandler):
protocol_version = "HTTP/1.1"
def log_message(self, *_args):
pass # 不输出 RPC 认证头、请求内容或带认证信息的 URL。
def reply(self, status, body, headers=()):
try:
self.send_response(status)
excluded = HOP_HEADERS | {
item.strip().lower() for key, value in headers if key.lower() == "connection"
for item in value.split(",")
}
for key, value in headers:
if key.lower() not in excluded:
self.send_header(key, value)
self.send_header("Content-Length", str(len(body)))
self.send_header("Connection", "close")
self.end_headers()
self.wfile.write(body)
except (OSError, ValueError):
pass
self.close_connection = True
def do_POST(self):
path = urllib.parse.urlsplit(self.path)
if (path.scheme or path.netloc or not path.path.startswith("/api/actions/")
or "%" in path.path or any(p in (".", "..") for p in path.path.split("/"))):
self.reply(404, b"runner RPC only\n")
return
try:
body = read_body(self.rfile, self.headers)
except (ValueError, OSError):
self.reply(400, b"invalid request body\n")
return
is_fetch = path.path.endswith("/FetchTask")
if is_fetch:
# 包括暂停时被拒绝的请求；只记录网络来源与时间以证明 daemon 路由。
self.server.gate.record_fetch(self.client_address[0])
if not self.server.gate.enter():
self.reply(503, b"runner maintenance\n")
return
completed = False
connection = None
try:
upstream = self.server.upstream
cls = http.client.HTTPSConnection if upstream.scheme == "https" else http.client.HTTPConnection
connection = cls(upstream.hostname, upstream.port, timeout=None)
try:
connection.connect()
except OSError:
# TCP/TLS 连接阶段尚未发送 RPC，不存在服务端分配事务。
completed = True
raise
excluded = HOP_HEADERS | {
name.strip().lower() for name in self.headers.get("Connection", "").split(",")
}
headers = {key: value for key, value in self.headers.items() if key.lower() not in excluded}
headers["Content-Length"] = str(len(body))
connection.request("POST", self.path, body=body, headers=headers)
response = connection.getresponse()
result = bytearray()
oversized = False
# 客户端即使断开，也继续读取上游完整响应，再释放领取屏障。
while block := response.read(65536):
if len(result) + len(block) <= MAX_BODY and not oversized:
result.extend(block)
else:
oversized = True
result.clear()
if response.length not in (None, 0):
raise http.client.IncompleteRead(bytes(result), response.length)
completed = True
if is_fetch:
self.server.gate.leave(True)
is_fetch = False
if oversized:
self.reply(502, b"upstream response too large\n")
else:
self.reply(response.status, bytes(result), response.getheaders())
except (OSError, http.client.HTTPException, ValueError):
self.reply(502, b"runner upstream unavailable\n")
finally:
if connection is not None:
connection.close()
if is_fetch:
self.server.gate.leave(completed)
class Control(socketserver.StreamRequestHandler):
def handle(self):
try:
line = self.rfile.readline(4097)
if len(line) > 4096 or not line.endswith(b"\n"):
raise ValueError("invalid control request")
request = json.loads(line)
result = self.server.gate.control(request["action"])
except (ValueError, KeyError, TypeError, OSError):
result = {"error": "invalid control request"}
self.wfile.write(json.dumps(result).encode("utf-8") + b"\n")
class ControlServer(socketserver.ThreadingUnixStreamServer):
daemon_threads = True
def create_proxy(address, upstream, gate):
parsed = urllib.parse.urlsplit(upstream)
if (parsed.scheme not in ("http", "https") or not parsed.hostname
or parsed.username or parsed.password or parsed.path not in ("", "/")
or parsed.query or parsed.fragment):
raise ValueError("upstream must be an HTTP(S) origin")
server = http.server.ThreadingHTTPServer(address, Proxy)
server.upstream = parsed
server.gate = gate
return server
def main():
gate = Gate(os.environ.get("GITEA_GATE_CONTROL_DIR", "/control"))
socket_path = gate.directory / "gate.sock"
socket_path.unlink(missing_ok=True)
control = ControlServer(str(socket_path), Control)
control.gate = gate
os.chmod(socket_path, 0o600)
proxy = create_proxy(("0.0.0.0", 8080), os.environ.get(
"GITEA_RUNNER_UPSTREAM", "http://gitea:3000"), gate)
threading.Thread(target=control.serve_forever, daemon=True).start()
proxy.serve_forever()
if __name__ == "__main__":
main()
scripts/maintain-gitea-rust-cache.py
+650
View File
File diff suppressed because it is too large Load Diff
scripts/project-ci-workflow.test.ts
+4 -3
View File
@@ -331,7 +331,8 @@ describe('project CI workflow', () => {
const rustLock = 'apps/ai-game-creator-shell/src-tauri/Cargo.lock';
for (const workspaceManifest of workspaceManifests) {
expect(imageBuildScript.split(workspaceManifest)).toHaveLength(3);
// 构建上下文与 revision 共用一份清单，不重复枚举 manifest。
expect(imageBuildScript.split(workspaceManifest)).toHaveLength(2);
expect(imageDockerignore).toContain(`!${workspaceManifest}`);
expect(imageDockerfile).toContain(
`COPY ${workspaceManifest} /usr/local/share/genarrative-ci/npm/${workspaceManifest}`,
@@ -339,8 +340,8 @@ describe('project CI workflow', () => {
}
for (const [path, expectedCount] of [
[rustManifest, 2],
[rustLock, 3],
[rustManifest, 1],
[rustLock, 2],
] as const) {
expect(imageBuildScript.split(path)).toHaveLength(expectedCount + 1);
expect(imageDockerignore).toContain(`!${path}`);
scripts/test_gitea_cache_builder.py
+201
View File
@@ -0,0 +1,201 @@
#!/usr/bin/env python3
"""Targeted regression tests for the trusted Gitea Rust-cache image builder."""
from __future__ import annotations
import os
from pathlib import Path
import shutil
import shlex
import stat
import subprocess
import tempfile
import textwrap
import unittest
REPOSITORY_ROOT = Path(__file__).resolve().parent.parent
BUILDER_NAME = "build-gitea-rust-cache.sh"
FIXED_SHA = "a" * 40
MASTER_SHA = "b" * 40
class GiteaRustCacheBuilderTest(unittest.TestCase):
def setUp(self) -> None:
self.temporary_directory = tempfile.TemporaryDirectory(ignore_cleanup_errors=True)
self.root = Path(self.temporary_directory.name)
self.repo = self.root / "repo"
scripts = self.repo / "scripts"
scripts.mkdir(parents=True)
shutil.copy2(REPOSITORY_ROOT / "scripts" / BUILDER_NAME, scripts / BUILDER_NAME)
(scripts / "gitea-ci-job-image.sh").write_text("#!/usr/bin/env bash\nexit 0\n")
(scripts / "ci-rust-cache.sh").write_text("#!/usr/bin/env bash\nexit 0\n")
self.trace = self.root / "trace.log"
self.bin = self.root / "bin"
self.bin.mkdir()
self.write_fake_tools()
self.execution_bin = self.wsl_path(self.bin)
self.wsl_fake_root: str | None = None
if os.name == "nt":
self.wsl_fake_root = f"/tmp/gitea-cache-builder-{self.root.name}"
subprocess.run(
[
"bash",
"-c",
"rm -rf {root}; mkdir -p {root}/bin; cp {source}/* {root}/bin/; chmod +x {root}/bin/*".format(
root=shlex.quote(self.wsl_fake_root),
source=shlex.quote(self.wsl_path(self.bin)),
),
],
check=True,
)
self.execution_bin = f"{self.wsl_fake_root}/bin"
def tearDown(self) -> None:
if self.wsl_fake_root is not None:
subprocess.run(
["bash", "-c", f"rm -rf {shlex.quote(self.wsl_fake_root)}"],
check=False,
)
self.temporary_directory.cleanup()
def write_tool(self, name: str, source: str) -> None:
target = self.bin / name
target.write_bytes(textwrap.dedent(source).encode("utf-8"))
target.chmod(target.stat().st_mode | stat.S_IXUSR)
def write_fake_tools(self) -> None:
self.write_tool(
"git",
"""#!/usr/bin/env bash
set -eu
printf 'git:%s\\n' "$*" >> "$TRACE"
if [[ "$1" == '-C' ]]; then shift 2; fi
case "$1" in
fetch|cat-file) exit 0 ;;
merge-base)
[[ "${FAKE_NON_ANCESTOR:-}" != 1 ]]
exit
;;
rev-parse)
if [[ "$2" == 'FETCH_HEAD^{commit}' ]]; then
printf '%s\\n' "$FAKE_MASTER_SHA"
else
printf '%s\\n' "$FAKE_FIXED_SHA"
fi
;;
archive) printf 'archived %s\\n' "$2" ;;
*) echo "unexpected git command: $*" >&2; exit 9 ;;
esac
""",
)
self.write_tool(
"docker",
"""#!/usr/bin/env bash
set -eu
# Drain the archive pipeline before logging to avoid concurrent
# appends to the shared trace on Windows/WSL filesystems.
if [[ "$1" == cp && "$2" == '-' ]]; then
cat >/dev/null
fi
printf 'docker:%s\\n' "$*" >> "$TRACE"
if [[ "$1" == image && "$2" == inspect ]]; then
printf 'sha256:%064d\\n' 0
elif [[ "$1" == run && " $* " == *' --detach '* ]]; then
printf 'fake-container\\n'
elif [[ "$1" == exec && " $* " == *' bash -s '* ]]; then
cat >/dev/null
fi
""",
)
self.write_tool(
"curl",
"""#!/usr/bin/env bash
set -eu
while [[ "$#" -gt 0 ]]; do
if [[ "$1" == --output ]]; then touch "$2"; exit 0; fi
shift
done
exit 9
""",
)
self.write_tool("sha256sum", "#!/usr/bin/env bash\ncat >/dev/null\n")
self.write_tool(
"tar",
"""#!/usr/bin/env bash
set -eu
while [[ "$#" -gt 0 ]]; do
if [[ "$1" == --directory ]]; then
mkdir -p "$2/sccache-v0.18.0-x86_64-unknown-linux-musl"
printf '#!/usr/bin/env bash\\nexit 0\\n' > "$2/sccache-v0.18.0-x86_64-unknown-linux-musl/sccache"
chmod +x "$2/sccache-v0.18.0-x86_64-unknown-linux-musl/sccache"
exit 0
fi
shift
done
exit 9
""",
)
@staticmethod
def wsl_path(path: Path) -> str:
value = path.resolve().as_posix()
if len(value) >= 3 and value[1] == ":":
return f"/mnt/{value[0].lower()}{value[2:]}"
return value
def run_builder(self, *arguments: str, non_ancestor: bool = False) -> subprocess.CompletedProcess[str]:
script_path = self.wsl_path(self.repo / "scripts" / BUILDER_NAME)
assignments = {
"TRACE": self.wsl_path(self.trace),
"FAKE_FIXED_SHA": FIXED_SHA,
"FAKE_MASTER_SHA": MASTER_SHA,
"FAKE_NON_ANCESTOR": "1" if non_ancestor else "",
}
exports = "; ".join(
f"export {name}={shlex.quote(value)}" for name, value in assignments.items()
)
command = (
f"unset CI; {exports}; export PATH={shlex.quote(self.execution_bin)}:\"$PATH\"; cd /; "
f"exec bash {shlex.quote(script_path)} "
+ " ".join(shlex.quote(argument) for argument in arguments)
)
return subprocess.run(
["bash", "-c", command],
env=os.environ,
text=True,
capture_output=True,
check=False,
)
def trace_text(self) -> str:
return self.trace.read_bytes().decode("utf-8", errors="replace")
def test_archives_requested_master_ancestor_and_uses_its_helper(self) -> None:
result = self.run_builder("trusted-base", "candidate", FIXED_SHA)
self.assertEqual(result.returncode, 0, result.stderr)
self.assertIn(f"snapshot_source={FIXED_SHA}", result.stdout)
trace = self.trace_text()
self.assertIn(f"archive {FIXED_SHA}", trace)
self.assertIn(
"docker:exec fake-container cp "
"/workspace/GenarrativeAI/Genarrative/scripts/ci-rust-cache.sh "
"/tmp/ci-rust-cache.sh",
trace,
)
def test_rejects_commit_outside_fetched_master(self) -> None:
result = self.run_builder("trusted-base", "candidate", FIXED_SHA, non_ancestor=True)
self.assertNotEqual(result.returncode, 0)
self.assertIn("not contained in fetched master", result.stderr)
self.assertNotIn(" archive ", self.trace_text())
def test_rejects_non_full_sha_before_calling_external_tools(self) -> None:
result = self.run_builder("trusted-base", "candidate", "abc123")
self.assertEqual(result.returncode, 2)
self.assertIn("complete 40-character SHA", result.stderr)
self.assertFalse(self.trace.exists())
if __name__ == "__main__":
unittest.main()
scripts/test_gitea_cache_gate.py
+211
View File
@@ -0,0 +1,211 @@
"""使用真实 HTTP/socket 验证领取屏障，不连接线上 Gitea。"""
import http.client
import http.server
import importlib.util
import json
import os
from pathlib import Path
import socket
import tempfile
import threading
import time
import unittest
if os.name != "posix":
raise unittest.SkipTest("领取屏障使用 Linux Unix socket 和目录 fsync；在 Linux/WSL 运行")
SPEC = importlib.util.spec_from_file_location(
"gitea_cache_gate", Path(__file__).with_name("gitea-runner-fetch-gate.py"))
MODULE = importlib.util.module_from_spec(SPEC)
SPEC.loader.exec_module(MODULE)
FETCH = "/api/actions/runner.v1.RunnerService/FetchTask"
UPDATE = "/api/actions/runner.v1.RunnerService/UpdateTask"
def core_state(state):
return {key: state[key] for key in ("paused", "inflight", "uncertain")}
class Upstream(http.server.BaseHTTPRequestHandler):
def log_message(self, *_args):
pass
def do_POST(self):
data = self.rfile.read(int(self.headers.get("Content-Length", 0)))
self.server.requests.append((self.path, data))
if self.path.endswith("/Redirect"):
self.send_response(302)
self.send_header("Location", "/api/actions/runner.v1.RunnerService/FetchTask")
self.send_header("Content-Length", "0")
self.end_headers()
return
if self.path == FETCH:
self.server.entered.set()
self.server.release.wait(10)
if self.server.truncated and self.path == FETCH:
self.send_response(200)
self.send_header("Content-Length", "100")
self.end_headers()
self.wfile.write(b"incomplete")
self.close_connection = True
return
body = b'{"task":null}'
self.send_response(200)
self.send_header("Content-Length", str(len(body)))
self.end_headers()
self.wfile.write(body)
class GateTests(unittest.TestCase):
def setUp(self):
self.temp = tempfile.TemporaryDirectory()
self.gate = MODULE.Gate(self.temp.name)
self.upstream = http.server.ThreadingHTTPServer(("127.0.0.1", 0), Upstream)
self.upstream.requests = []
self.upstream.entered = threading.Event()
self.upstream.release = threading.Event()
self.upstream.truncated = False
self.proxy = MODULE.create_proxy(("127.0.0.1", 0),
f"http://127.0.0.1:{self.upstream.server_port}", self.gate)
self.servers = [self.upstream, self.proxy]
for server in self.servers:
threading.Thread(target=server.serve_forever, daemon=True).start()
def tearDown(self):
self.upstream.release.set()
for server in reversed(self.servers):
server.shutdown()
server.server_close()
self.temp.cleanup()
def request(self, path, body=b"{}", chunked=False):
connection = http.client.HTTPConnection("127.0.0.1", self.proxy.server_port, timeout=5)
try:
if chunked:
connection.request("POST", path, body=[body], encode_chunked=True)
else:
connection.request("POST", path, body=body)
response = connection.getresponse()
result = response.status, response.read()
return result
finally:
connection.close()
def wait_for(self, predicate):
deadline = time.monotonic() + 3
while time.monotonic() < deadline:
if predicate():
return
time.sleep(0.01)
self.fail("condition did not become true")
def start_fetch(self):
self.result = []
thread = threading.Thread(target=lambda: self.result.append(self.request(FETCH)))
thread.start()
self.assertTrue(self.upstream.entered.wait(3))
return thread
def test_pause_holds_inflight_and_allows_reporting(self):
thread = self.start_fetch()
self.assertEqual(core_state(self.gate.control("pause")),
{"paused": True, "inflight": 1, "uncertain": False})
self.assertEqual(self.request(FETCH)[0], 503)
self.assertEqual(self.request(UPDATE)[0], 200)
self.assertEqual(sum(path == FETCH for path, _ in self.upstream.requests), 1)
self.upstream.release.set()
thread.join(3)
self.assertFalse(thread.is_alive())
self.assertEqual(self.result[0][0], 200)
self.assertEqual(self.gate.control("status")["inflight"], 0)
self.assertEqual(self.gate.control("resume")["paused"], False)
self.assertEqual(self.request(FETCH)[0], 200)
def test_disconnected_client_does_not_release_upstream_request(self):
client = socket.create_connection(self.proxy.server_address, timeout=3)
client.sendall(f"POST {FETCH} HTTP/1.1\r\nHost: localhost\r\nContent-Length: 2\r\n\r\n{{}}".encode())
self.assertTrue(self.upstream.entered.wait(3))
client.close()
state = self.gate.control("pause")
self.assertEqual(state["inflight"], 1)
self.upstream.release.set()
self.wait_for(lambda: self.gate.control("status")["inflight"] == 0)
self.assertFalse(self.gate.control("status")["uncertain"])
def test_truncated_upstream_latches_uncertainty_across_restart(self):
self.upstream.truncated = True
self.upstream.release.set()
self.assertEqual(self.request(FETCH)[0], 502)
self.wait_for(lambda: self.gate.control("status")["uncertain"])
self.assertTrue(self.gate.control("resume")["paused"])
restored = MODULE.Gate(self.temp.name)
self.assertTrue(restored.control("status")["uncertain"])
self.assertFalse(restored.enter())
def test_crashed_inflight_is_not_treated_as_successful_drain(self):
self.assertTrue(self.gate.enter())
restored = MODULE.Gate(self.temp.name)
self.assertTrue(restored.control("status")["uncertain"])
self.assertIn("error", restored.control("resume"))
self.gate.leave(True)
def test_pause_marker_survives_restart(self):
self.gate.control("pause")
restored = MODULE.Gate(self.temp.name)
self.assertEqual(core_state(restored.control("status")),
{"paused": True, "inflight": 0, "uncertain": False})
restored.control("resume")
self.assertFalse(MODULE.Gate(self.temp.name).control("status")["paused"])
def test_chunked_body_is_decoded_and_non_rpc_path_rejected(self):
self.assertEqual(self.request(UPDATE, b'{"state":"running"}', chunked=True)[0], 200)
self.assertEqual(self.upstream.requests[-1], (UPDATE, b'{"state":"running"}'))
self.assertEqual(self.request("/api/v1/repos")[0], 404)
self.assertEqual(self.request("/api/actions/../v1/repos")[0], 404)
def test_upstream_redirect_is_not_followed(self):
self.assertEqual(self.request("/api/actions/runner.v1.RunnerService/Redirect")[0], 302)
self.assertFalse(self.upstream.entered.is_set())
self.assertEqual(len(self.upstream.requests), 1)
def test_connection_refused_before_rpc_does_not_latch_uncertainty(self):
with socket.socket() as unused:
unused.bind(("127.0.0.1", 0))
port = unused.getsockname()[1]
self.proxy.upstream = MODULE.urllib.parse.urlsplit(f"http://127.0.0.1:{port}")
self.assertEqual(self.request(FETCH)[0], 502)
self.assertEqual(core_state(self.gate.control("status")),
{"paused": False, "inflight": 0, "uncertain": False})
@unittest.skipUnless(hasattr(socket, "AF_UNIX"), "Unix control socket required")
def test_control_socket_uses_newline_json(self):
path = str(Path(self.temp.name) / "gate.sock")
control = MODULE.ControlServer(path, MODULE.Control)
control.gate = self.gate
self.servers.append(control)
threading.Thread(target=control.serve_forever, daemon=True).start()
with socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) as client:
client.connect(path)
client.sendall(b'{"action":"pause"}\n')
payload = client.makefile("rb").readline()
self.assertEqual(core_state(json.loads(payload)),
{"paused": True, "inflight": 0, "uncertain": False})
def test_paused_fetch_records_live_route_and_restart_discards_proof(self):
self.gate.control("pause")
before = time.time()
self.assertEqual(self.request(FETCH)[0], 503)
state = self.gate.control("status")
self.assertEqual(state["last_fetch_peer"], "127.0.0.1")
self.assertGreaterEqual(state["last_fetch_at"], before)
self.assertEqual(self.upstream.requests, [])
self.assertEqual(self.request(UPDATE)[0], 200)
self.assertEqual(self.gate.control("status")["last_fetch_at"], state["last_fetch_at"])
restarted = MODULE.Gate(self.temp.name).control("status")
self.assertIsNone(restarted["last_fetch_peer"])
self.assertIsNone(restarted["last_fetch_at"])
if __name__ == "__main__":
unittest.main()
scripts/test_gitea_cache_maintenance.py
+424
View File
@@ -0,0 +1,424 @@
#!/usr/bin/env python3
"""High-risk state-transition tests for the host-only Rust cache maintainer."""
from __future__ import annotations
import importlib.util
import json
from pathlib import Path
import tempfile
import unittest
from unittest.mock import patch
SCRIPT = Path(__file__).with_name("maintain-gitea-rust-cache.py")
SPEC = importlib.util.spec_from_file_location("gitea_cache_maintenance", SCRIPT)
assert SPEC and SPEC.loader
maintenance_module = importlib.util.module_from_spec(SPEC)
SPEC.loader.exec_module(maintenance_module)
IMAGE = "sha256:" + "1" * 64
OLD_IMAGE = "sha256:" + "2" * 64
BASE_IMAGE = "sha256:" + "3" * 64
SHA = "a" * 40
def runner_config(image: str) -> str:
return f'runners:\n - "genarrative-ci:docker://{image}"\n'
class FakeApi:
def __init__(self, disabled: bool = False):
self.disabled = disabled
self.requests: list[str] = []
def request(self, path, body=None, raw=False):
self.requests.append(path)
if path.startswith("admin/actions/runners/"):
return {"disabled": self.disabled, "busy": False}
raise AssertionError(f"unexpected API request: {path}")
class GiteaCacheMaintenanceTest(unittest.TestCase):
def setUp(self):
self.temporary_directory = tempfile.TemporaryDirectory()
self.root = Path(self.temporary_directory.name)
self.token = self.root / "token"
self.token.write_text("test-token\n", encoding="utf-8")
self.config = {
"state_dir": str((self.root / "state").resolve()),
"api_url": "https://gitea.example.test/api/v1",
"token_file": str(self.token),
"repository": "team/project",
"repository_url": "http://gitea:3000/team/project.git",
"runner_api_path": "admin/actions/runners/1",
"runner_container": "gitea-runner",
"gate_socket": str(self.root / "gate.sock"),
"gate_url": "https://gitea.example.test",
"clone_url": "https://gitea.example.test/team/project.git",
}
def tearDown(self):
self.temporary_directory.cleanup()
def record(self, image, *, source=SHA, owned=True, verified_run=None, base=BASE_IMAGE):
return {
"image": image,
"tag": (
"genarrative/gitea-project-ci:rust-cache-auto-" + source
if source is not None else None
),
"source": source,
"inputs": "inputs",
"base": base,
"owned": owned,
"activated": "2026-01-01T00:00:00+00:00",
"verified_run": verified_run,
"artifact": str(self.root / "state" / "artifacts" / (source or "manual")),
}
def maintenance(self, state):
state_root = Path(self.config["state_dir"])
state_root.mkdir(parents=True, exist_ok=True)
(state_root / "state.json").write_text(json.dumps(state), encoding="utf-8")
instance = maintenance_module.Maintenance(self.config)
instance.api = FakeApi()
return instance
def activation_maintenance(self, *, idle_values, switch=None):
old = self.record(OLD_IMAGE, owned=False, verified_run=1)
candidate = self.record(IMAGE)
state = {
"versions": [old, candidate],
"current": OLD_IMAGE,
"rollback": None,
"candidate": IMAGE,
}
if switch is not None:
state["switch"] = switch
instance = self.maintenance(state)
config = {"value": runner_config(OLD_IMAGE)}
writes: list[str] = []
restarts: list[tuple] = []
statuses = 0
instance.check_gate_route = lambda: None
instance.read_config = lambda: config["value"]
instance.write_config = lambda value: (writes.append(value), config.__setitem__("value", value))
instance.image_info = lambda image, inner=False: {"Id": image}
def docker(*args, inner=False, **kwargs):
if args[:1] == ("restart",):
restarts.append(args)
return ""
if args[:2] == ("inspect", "--format"):
if args[2] == "{{.State.StartedAt}}":
return "2026-01-01T00:00:00+00:00"
if args[2] == "{{.State.Status}}":
return "running"
if args[:1] == ("logs",):
return "declare successfully"
raise AssertionError(f"unexpected docker call: {args}, inner={inner}")
instance.docker = docker
iterator = iter(idle_values)
instance.idle = lambda: next(iterator)
def gate(action):
nonlocal statuses
if action == "pause":
return {"paused": True}
if action == "resume":
return {"paused": False}
if action == "status":
statuses += 1
return {"paused": False} if statuses == 1 else {"paused": True, "inflight": 0}
raise AssertionError(action)
instance.gate = gate
return instance, writes, restarts
def test_cache_inputs_keeps_openapi_and_embedded_skill_text(self):
tree = "\n".join(
[
"100644 blob a\tdocs/guide.md",
"100644 blob b\tdocs/openapi/external.json",
"100644 blob c\t.codex/skills/example/SKILL.md",
"100644 blob d\tREADME.md",
]
)
digest = maintenance_module.cache_inputs(tree)
self.assertEqual(digest, maintenance_module.cache_inputs(
tree.replace("guide.md", "【说明】中文文档.md").replace("\n", "\0") + "\0"))
self.assertEqual(digest, maintenance_module.cache_inputs(tree.replace("guide.md", "other.md")))
self.assertNotEqual(digest, maintenance_module.cache_inputs(tree.replace("external.json", "changed.json")))
self.assertNotEqual(digest, maintenance_module.cache_inputs(tree.replace("SKILL.md", "PROMPT.md")))
def test_gate_route_requires_independent_checkout_and_live_runner_proof(self):
instance = self.maintenance({"versions": [], "current": None, "rollback": None, "candidate": None})
instance.read_config = lambda: runner_config(OLD_IMAGE)
with self.assertRaisesRegex(RuntimeError, "GENARRATIVE_GITEA_REPOSITORY_URL"):
instance.check_gate_route()
instance.read_config = lambda: runner_config(OLD_IMAGE) + (
' envs:\n GENARRATIVE_GITEA_REPOSITORY_URL: "http://gitea:3000/team/project.git"\n')
started = "2026-01-01T00:00:00Z"
def docker(*args, **kwargs):
if args[0] == "exec":
return json.dumps({"address": self.config["gate_url"]})
return json.dumps([{"State": {"StartedAt": started}, "NetworkSettings": {
"Networks": {"internal": {"IPAddress": "10.0.0.2"}}}}])
instance.docker = docker
proof = {"last_fetch_peer": "10.0.0.2", "last_fetch_at": maintenance_module.timestamp(started).timestamp() + 1}
instance.gate = lambda action: proof
instance.check_gate_route()
proof["last_fetch_at"] -= 2
with self.assertRaisesRegex(RuntimeError, "this runner startup"):
instance.check_gate_route()
proof.update(last_fetch_peer="10.0.0.3", last_fetch_at=9999999999)
with self.assertRaisesRegex(RuntimeError, "this runner startup"):
instance.check_gate_route()
def test_complete_nine_job_run_requires_real_rust_cache_statistics(self):
run = {"status": "completed", "conclusion": "success"}
jobs = [{"name": name, "conclusion": "success"} for name in maintenance_module.JOBS]
self.assertTrue(maintenance_module.fully_passed(run, jobs))
self.assertFalse(maintenance_module.fully_passed(run, jobs[:-1]))
good = (
f"worker image: {IMAGE}\n[rust-cache] mode=sccache\n"
"Cache hits 7\nCache errors 0\nCache read errors 0\nCache write errors 0\n"
)
self.assertTrue(maintenance_module.verifies_image({"name": "Backend tests"}, good, IMAGE))
self.assertFalse(maintenance_module.verifies_image({"name": "Backend tests"}, good.replace("Cache hits 7", "Cache hits 0"), IMAGE))
self.assertFalse(maintenance_module.verifies_image({"name": "Backend tests"}, good.replace("Cache write errors 0", "Cache write errors 1"), IMAGE))
def test_fetches_latest_master_without_ci_green_gate(self):
instance = self.maintenance({"versions": [], "current": None, "rollback": None, "candidate": None})
instance.repo.mkdir(parents=True)
(instance.repo / ".git").mkdir()
(instance.repo / ".git" / "genarrative-cache-maintenance").write_text("owned source checkout\n")
calls = []
def fake_command(*args, **kwargs):
calls.append(args)
if args[1:4] == ("remote", "get-url", "origin"):
return instance.config["clone_url"] + "\n"
if args[1:3] == ("rev-parse", "FETCH_HEAD^{commit}"):
return SHA + "\n"
if args[1:3] == ("ls-tree", "-rz"):
return "100644 blob x\tdocs/openapi/current.json\n"
return ""
with patch.object(maintenance_module, "command", fake_command):
self.assertEqual(instance.fetch_source()[0], SHA)
self.assertIn(("git", "remote", "get-url", "origin"), calls)
self.assertIn(("git", "fetch", "--no-tags", "origin", "refs/heads/master"), calls)
self.assertIn(("git", "clean", "-ffdx"), calls)
self.assertFalse(any("runs" in " ".join(call) for call in calls))
def test_idle_race_after_pause_restores_without_restart(self):
instance, writes, restarts = self.activation_maintenance(idle_values=[True, False])
with patch.object(maintenance_module.time, "sleep", lambda _: None):
self.assertFalse(instance.activate())
self.assertEqual(restarts, [])
self.assertEqual(writes, [])
self.assertEqual(instance.state["current"], OLD_IMAGE)
self.assertEqual(instance.state["candidate"], IMAGE)
self.assertNotIn("pause_owned", instance.state)
def test_uncertain_or_inflight_gate_never_restarts(self):
instance, _, restarts = self.activation_maintenance(idle_values=[True])
statuses = 0
def uncertain_gate(action):
nonlocal statuses
if action == "pause":
return {"paused": True}
if action == "resume":
return {"paused": False}
statuses += 1
return {"paused": False} if statuses == 1 else {"paused": True, "uncertain": True, "inflight": 1}
instance.gate = uncertain_gate
with patch.object(maintenance_module.time, "sleep", lambda _: None):
with self.assertRaisesRegex(RuntimeError, "uncertain"):
instance.activate()
self.assertEqual(restarts, [])
self.assertEqual(instance.state["candidate"], IMAGE)
self.assertEqual(instance.state["current"], OLD_IMAGE)
self.assertNotIn("pause_owned", instance.state)
def test_inflight_gate_defers_without_restart_and_resumes(self):
instance, _, restarts = self.activation_maintenance(idle_values=[True])
statuses = 0
def inflight_gate(action):
nonlocal statuses
if action == "pause":
return {"paused": True}
if action == "resume":
return {"paused": False}
statuses += 1
return {"paused": False} if statuses == 1 else {"paused": True, "inflight": 1}
instance.gate = inflight_gate
with patch.object(maintenance_module.time, "sleep", lambda _: None):
self.assertFalse(instance.activate())
self.assertEqual(restarts, [])
self.assertEqual(instance.state["candidate"], IMAGE)
self.assertEqual(instance.state["current"], OLD_IMAGE)
self.assertNotIn("pause_owned", instance.state)
def test_interrupted_switch_recovers_current_candidate_and_rollback(self):
backup = self.root / "state" / "backups" / "switch.yaml"
backup.parent.mkdir(parents=True)
backup.write_text(runner_config(OLD_IMAGE), encoding="utf-8")
instance, _, restarts = self.activation_maintenance(
idle_values=[True, True],
switch={"old": OLD_IMAGE, "new": IMAGE, "backup": str(backup)},
)
with patch.object(maintenance_module.time, "sleep", lambda _: None):
self.assertTrue(instance.recover_switch())
self.assertEqual(len(restarts), 1)
self.assertEqual(instance.state["current"], IMAGE)
self.assertIsNone(instance.state["candidate"])
self.assertEqual(instance.state["rollback"], OLD_IMAGE)
self.assertNotIn("switch", instance.state)
def test_protected_images_include_state_bases_config_and_container_references(self):
current = self.record(OLD_IMAGE, owned=False, verified_run=1, base=BASE_IMAGE)
candidate = self.record(IMAGE, base=BASE_IMAGE)
container_image = "sha256:" + "4" * 64
instance = self.maintenance({"versions": [current, candidate], "current": OLD_IMAGE, "rollback": OLD_IMAGE, "candidate": IMAGE})
instance.read_config = lambda: runner_config(OLD_IMAGE)
def docker(*args, inner=False, **kwargs):
if args[:2] == ("ps", "-aq"):
return "container-1\n"
if args[:2] == ("inspect", "--format"):
return container_image + "\n"
raise AssertionError(args)
instance.docker = docker
protected = instance.protected_images()
self.assertTrue({OLD_IMAGE, IMAGE, BASE_IMAGE, container_image}.issubset(protected))
def test_cleanup_keeps_manual_images_and_rejects_outside_artifacts(self):
current = self.record(OLD_IMAGE, owned=False, verified_run=1)
manual = self.record(IMAGE, owned=False)
stale = self.record("sha256:" + "5" * 64, source="b" * 40)
stale["artifact"] = str(self.root / "outside")
instance = self.maintenance({"versions": [current, manual, stale], "current": OLD_IMAGE, "rollback": None, "candidate": None})
instance.protected_images = lambda: set()
instance.remove_owned_image = lambda *_: True
with self.assertRaisesRegex(RuntimeError, "outside"):
instance.cleanup()
self.assertIn(manual, instance.state["versions"])
self.assertIn(stale, instance.state["versions"])
def test_additional_tags_are_never_deleted(self):
instance = self.maintenance({"versions": [], "current": None, "rollback": None, "candidate": None})
removals = []
tag = "genarrative/gitea-project-ci:rust-cache-auto-" + SHA
instance.image_info = lambda image, inner=False: {"RepoTags": [tag, "operator:keep"]}
instance.docker = lambda *args, inner=False, **kwargs: (
IMAGE + "\n" if args[:2] == ("image", "ls") else removals.append(args) or ""
)
self.assertFalse(instance.remove_owned_image(IMAGE, tag, set()))
self.assertEqual(removals, [])
def test_owned_image_removal_handles_dangling_and_already_absent_images(self):
instance = self.maintenance({"versions": [], "current": None, "rollback": None, "candidate": None})
tag = "genarrative/gitea-project-ci:rust-cache-auto-" + SHA
present = {True: True, False: True}
removals = []
def docker(*args, inner=False, **kwargs):
if args[:2] == ("image", "ls"):
return IMAGE + "\n" if present[inner] else ""
if args[:3] == ("image", "rm", IMAGE):
removals.append((args, inner))
present[inner] = False
return IMAGE + "\n"
raise AssertionError(args)
instance.docker = docker
instance.image_info = lambda image, inner=False: {"RepoTags": []}
self.assertTrue(instance.remove_owned_image(IMAGE, tag, set()))
self.assertEqual({inner for _, inner in removals}, {True, False})
calls = []
instance.docker = lambda *args, inner=False, **kwargs: calls.append((args, inner)) or ""
self.assertTrue(instance.remove_owned_image(
IMAGE, "genarrative/gitea-project-ci:rust-cache-auto-" + SHA, set()))
self.assertFalse(any(args[:3] == ("image", "rm", IMAGE) for args, _ in calls))
def test_manual_current_without_cache_stats_can_be_verified(self):
current = self.record(OLD_IMAGE, source=None, owned=False)
instance = self.maintenance({"versions": [current], "current": OLD_IMAGE, "rollback": None, "candidate": None})
jobs = [
{"id": index, "name": name, "conclusion": "success"}
for index, name in enumerate(sorted(maintenance_module.JOBS), start=1)
]
run = {
"id": 44,
"status": "completed",
"conclusion": "success",
"event": "push",
"path": "project-ci.yml@refs/heads/master",
"head_sha": SHA,
"started_at": "2026-01-02T00:00:00Z",
}
class VerificationApi:
def pages(self, path, key):
if path.endswith("/actions/runs?status=success&branch=master&event=push"):
return iter([run])
if path.endswith("/actions/runs/44/jobs"):
return iter(jobs)
raise AssertionError(path)
def request(self, path, body=None, raw=False):
if not raw or "/actions/jobs/" not in path:
raise AssertionError(path)
return f"runner image: {OLD_IMAGE}\n"
instance.api = VerificationApi()
with patch.object(maintenance_module.subprocess, "run", return_value=type("Result", (), {"returncode": 0})()):
self.assertTrue(instance.verify_current())
self.assertEqual(instance.version(OLD_IMAGE)["verified_run"], 44)
self.assertEqual(instance.version(OLD_IMAGE)["verified_sha"], SHA)
def test_recover_builds_adopts_only_pre_registered_attempt_tags(self):
tag = "genarrative/gitea-project-ci:rust-cache-auto-" + SHA
base_tag = "genarrative/gitea-project-ci:base-auto-" + SHA
artifact = self.root / "state" / "artifacts" / SHA
attempt = {"source": SHA, "inputs": "inputs", "tag": tag, "base_tag": base_tag, "artifact": str(artifact)}
instance = self.maintenance({
"versions": [], "current": None, "rollback": None, "candidate": None,
"attempts": [attempt],
})
queries = []
def docker(*args, inner=False, **kwargs):
if args[:3] == ("image", "ls", "--no-trunc"):
queries.append(args[-1])
return BASE_IMAGE + "\n" if args[-1] == base_tag else IMAGE + "\n"
raise AssertionError(args)
instance.docker = docker
instance.image_info = lambda image, inner=False: {
"Config": {"Labels": {
"world.genarrative.ci.rust-cache-source": SHA,
"world.genarrative.ci.rust-cache-base": BASE_IMAGE,
}}
}
instance.recover_builds()
self.assertEqual(queries, [base_tag, tag])
self.assertEqual(instance.state["attempts"], [])
self.assertEqual(instance.state["bases"], {BASE_IMAGE: base_tag})
self.assertEqual(instance.state["versions"][0]["image"], IMAGE)
self.assertTrue(instance.state["versions"][0]["owned"])
self.assertEqual(instance.state["candidate"], IMAGE)
if __name__ == "__main__":
unittest.main()