diff --git a/deploy/container/README.md b/deploy/container/README.md index ee7cbeaa0..5bf1fecb8 100644 --- a/deploy/container/README.md +++ b/deploy/container/README.md @@ -186,6 +186,8 @@ timer 在上次执行结束后约 5 分钟再次检查，文件锁防止人工 入口遇到“已发送 FetchTask，但上游响应未完整结束”会持久化 `uncertain` 并拒绝继续领取/自动切换；不会因为客户端的 5 秒超时就认定服务端事务已回滚。其它 RPC 和任务上报仍继续转发。维护者需先核实 Gitea 在途领取事务与该 runner 的任务全部收敛，在维护窗口停止入口，核实并修复它的 `tasks.json` 任务账本及 `uncertain` / `inflight` 标记后再启动并恢复领取。网关按 Gitea 1.26.4 / Runner 2.0.0 的 Connect Protobuf 协议，在最终日志确认及 Runner 执行结束后的最终任务上报确认后才清账；取消响应不等于进程停止，任务 ID 不按超时自动删除。禁止自动删这些标记绕过屏障。停用自动维护先 `systemctl disable --now genarrative-ci-cache.timer`；不要为了停 timer 停止运行中的 CI 容器。已接入的领取入口继续运行，不影响普通 CI。 +Gitea 1.26.4 的 `FetchTask` 若完整返回 HTTP 500、未压缩的 Connect JSON `code=unknown`，且 message 精确为 `rpc error: code = Internal desc = pick task: CreateTaskForRunner: update run <正整数>: run has changed`，网关原样返回错误并允许 runner 下次轮询，不写入 `uncertain`。已核对上游 [CreateTaskForRunner](https://github.com/go-gitea/gitea/blob/v1.26.4/models/actions/task.go) 和 [PickTask](https://github.com/go-gitea/gitea/blob/v1.26.4/services/actions/task.go)：此并发更新错误发生在分配事务提交前并回滚。该例外不覆盖其它 5xx、未知格式、响应截断或网络断连，也不会清除既有保护锁、维护暂停和活跃任务账本。网关将已回滚冲突和进入保护状态的原因分类写入容器日志，不输出 RPC 正文或凭据；旧版本留下的保护锁仍须核实后人工恢复。 + 人工 bootstrap 预热容器上限为 4 核、12 GiB，移除 capabilities，不挂宿主目录/socket，也不注入 Git/OSS/Jenkins 凭据。源码通过 `git archive` 复制，当前工作区、ignored 文件和 `.git` 不进入容器。最终从原镜像重新组装，仅复制 `/opt/genarrative-ci/rust-cache` 的 sccache、对象和来源元数据，不提交含源码/target 的预热容器；镜像本身的下载缓存与工具链校验保持原样。 快照由固定 Image ID 分发，每个 job 仅修改容器自己的写时复制层，缓存上限 4 GiB，只有 master push 在结束前回传新增对象；PR 不扫描基线、不打包、不回传。`ci-rust-cache.sh prepare` 清空继承的 `SCCACHE_*` 远程配置，使用独立配置和 Unix socket；旧镜像、工具链不匹配或限时 wrapper 探测失败时使用直接 rustc，正式编译启用 sccache 的 server IO 错误回退。真实编译/测试失败保留非零退出码。`report` 输出命中统计并停止本 job daemon，分片日志输出独立编译耗时。sccache 0.18.0 的只读模式在 miss 后仍打包再拒绝写入，不能用它宣称零 miss 开销；普通 CI 继续只向容器层写入。 diff --git a/docs/project-memory/shared-memory/development-workflow.md b/docs/project-memory/shared-memory/development-workflow.md index c4e72194e..bd74d29c4 100644 --- a/docs/project-memory/shared-memory/development-workflow.md +++ b/docs/project-memory/shared-memory/development-workflow.md @@ -112,6 +112,8 @@ Gitea 基础镜像通过专用 `genarrative-ci-images` Buildx builder 持久复 Gitea Rust 缓存自动维护由宿主 `genarrative-ci-cache.timer` 收集同一 master push run 六个 Rust job 的原生 V4 缓存产物，不重复执行 Cargo 预热。只传本轮新 key，命中对象只传使用时间；宿主与真实来源镜像对象合并、去重、按新近使用时间裁剪到 4 GiB，从无对象缓存基础镜像重新组装。源 run 不要求全绿，但取消、缺组、旧 attempt、未完成上传或混用来源镜像不得采用。网关暂停新 FetchTask、在途领取结束、持久化任务账本清空且内层活动容器为空才切换，不打断运行中的 CI。首次接入/升级网关需空闲窗口；Token 只需普通仓库 `write:repository`，不查管理员 API。候选装载后清理已收集 artifact，遗留项保留 7 天；真实 master CI 验证后才清理旧镜像，保留当前、一个回滚版、基础镜像及容器引用。部署入口见 `deploy/container/README.md`，合并代码不等于服务启用。 +领取网关只对已核对 Gitea 1.26.4 事务回滚的精确 `CreateTaskForRunner: update run : run has changed` Connect 错误允许 runner 重试，避免普通并发更新冲突永久阻断 CI；完整响应格式约束见 `deploy/container/README.md`。其它未知领取结果继续持久化保护，不自动清除 `uncertain`；恢复既有锁前需核实任务与在途请求收敛。容器日志只记录固定原因分类，不打印请求、响应或认证信息。 + 修改 Gitea workflow 的 job 显示名称、ID 或缓存导出组时，必须同步维护器的 `JOBS` / `RUST_JOB_IDS`；`test_gitea_cache_maintenance.py` 直接对照实际 workflow 检查全集和导出映射，避免自动刷新或镜像验收因名单漂移长期等待。维护器 `Api.request` 的 `method` 是必填关键字参数，GET 也必须显式指定，不根据 body 推断请求方法。 Gitea 缓存部署必须区分网络：runner 的 RPC 走 `gitea-runner-fetch-gate:8080`；内层 job 的 checkout/上传走映射到 `172.30.0.3` 的 `http://genarrative-station/git`；宿主专用 clone 走 `http://127.0.0.1:3003`。不要把 runner 可达的 `gitea:3000` 配给 job。内层 Docker 使用 `10.240.0.0/16`、每 job `/24` 的默认地址池，避开外层 `172.30/172.31` 网段；恢复领取前必须在真实 job 网络里验证 checkout 与 Gitea API，不能只验证 FetchTask。具体配置与遗留空网络处理见 `deploy/container/README.md`。 diff --git a/scripts/gitea-runner-fetch-gate.py b/scripts/gitea-runner-fetch-gate.py index 7f8346e65..49de66af8 100644 --- a/scripts/gitea-runner-fetch-gate.py +++ b/scripts/gitea-runner-fetch-gate.py @@ -8,7 +8,9 @@ import io import json import os from pathlib import Path +import re import socketserver +import sys import threading import time import urllib.parse @@ -22,6 +24,30 @@ HOP_HEADERS = { RPC_PREFIX = "/api/actions/runner.v1.RunnerService/" +def fetch_conflict_rolled_back(status, body, headers): + """仅识别 Gitea 1.26.4 在任务分配事务提交前返回的 run 更新冲突。""" + # PickTask/WithTx -> CreateTaskForRunner -> UpdateRunJob -> UpdateRun。 + # 该错误会回滚任务分配；其它 5xx、截断响应和未知协议仍须阻断领取。 + if (status != 500 or len(body) > 4096 + or headers.get("Content-Encoding", "identity").lower() != "identity" + or headers.get("Content-Type", "").split(";", 1)[0] != "application/json"): + return False + try: + error = json.loads(body) + except (ValueError, UnicodeError): + return False + return (isinstance(error, dict) and error.get("code") == "unknown" + and isinstance(error.get("message"), str) + and re.fullmatch( + r"rpc error: code = Internal desc = pick task: CreateTaskForRunner: " + r"update run [1-9][0-9]*: run has changed", error["message"]) is not None) + + +def diagnostic(message): + # 仅传固定分类，不输出上游正文、URL 或 RPC 凭据。 + print(f"[runner-fetch-gate] {message}", file=sys.stderr, flush=True) + + def protobuf_fields(data): """只解码已核对的 actions-proto-go v0.4.1 字段，不引入 protobuf 运行时。""" fields = {} @@ -153,8 +179,10 @@ class Gate: os.replace(temporary, self.directory / "tasks.json") self.sync_directory() - def fail_closed(self): + def fail_closed(self, reason): with self.lock: + if not self.uncertain: + diagnostic(f"uncertain: {reason}; operator recovery required") self.uncertain = self.paused = True self.mark("uncertain") @@ -238,6 +266,8 @@ class Gate: with self.lock: self.inflight -= 1 if not completed: + if not self.uncertain: + diagnostic("uncertain: incomplete FetchTask response; operator recovery required") self.uncertain = self.paused = True self.mark("uncertain") if self.inflight == 0 and not self.uncertain: @@ -369,11 +399,14 @@ class Proxy(http.server.BaseHTTPRequestHandler): completed = True if is_fetch: try: - if oversized or response.status != 200: + if not oversized and fetch_conflict_rolled_back(response.status, result, response.headers): + diagnostic("FetchTask transaction rolled back: run update conflict; runner may retry") + elif oversized or response.status != 200: raise ValueError("unknown FetchTask result") - self.server.gate.assigned(rpc_fields(bytes(result), response.headers)) + else: + self.server.gate.assigned(rpc_fields(bytes(result), response.headers)) except (ValueError, TypeError, OSError, EOFError, zlib.error): - self.server.gate.fail_closed() + self.server.gate.fail_closed("unrecognized FetchTask response or assignment tracking failure") self.server.gate.leave(True) is_fetch = False if oversized: @@ -398,7 +431,7 @@ class Proxy(http.server.BaseHTTPRequestHandler): self.server.gate.reported(method, rpc_fields(request, self.headers), rpc_fields(response, headers)) except (ValueError, TypeError, OSError, EOFError, zlib.error): - self.server.gate.fail_closed() + self.server.gate.fail_closed("task report tracking failure") class Control(socketserver.StreamRequestHandler): diff --git a/scripts/test_gitea_cache_gate.py b/scripts/test_gitea_cache_gate.py index ddc3a4a2c..45a572acf 100644 --- a/scripts/test_gitea_cache_gate.py +++ b/scripts/test_gitea_cache_gate.py @@ -326,6 +326,53 @@ class GateTests(unittest.TestCase): self.assertTrue(self.gate.control("status")["uncertain"]) self.assertTrue(MODULE.Gate(self.temp.name).control("status")["uncertain"]) + def test_rolled_back_run_conflict_allows_next_fetch_and_preserves_tasks(self): + self.fetch_task() + body = json.dumps({"code": "unknown", "message": + "rpc error: code = Internal desc = pick task: CreateTaskForRunner: " + "update run 3673: run has changed"}).encode() + self.upstream.responses[FETCH] = body + self.upstream.statuses[FETCH] = 500 + self.upstream.content_type = "application/json" + self.assertEqual(self.request(FETCH), (500, body)) + self.assertEqual(core_state(self.gate.control("status")), + {"paused": False, "inflight": 0, "uncertain": False}) + self.assertEqual(self.gate.control("status")["task_ids"], ["42"]) + self.assertFalse(MODULE.Gate(self.temp.name).control("status")["uncertain"]) + self.assertFalse((Path(self.temp.name) / "inflight").exists()) + self.upstream.statuses[FETCH] = 200 + self.upstream.content_type = "application/proto" + self.upstream.responses[FETCH] = field(1, field(1, 43)) + self.assertEqual(self.request(FETCH)[0], 200) + self.assertEqual(self.gate.control("status")["task_ids"], ["42", "43"]) + + def test_unrecognized_fetch_error_still_latches_uncertainty(self): + self.upstream.release.set() + self.upstream.content_type = "application/json" + self.upstream.statuses[FETCH] = 500 + self.upstream.responses[FETCH] = json.dumps({"code": "unknown", "message": + "rpc error: code = Internal desc = pick task: commit failed"}).encode() + self.assertEqual(self.request(FETCH)[0], 500) + self.assertTrue(self.gate.control("status")["uncertain"]) + self.assertEqual(self.request(FETCH)[0], 503) + + def test_conflict_exception_requires_exact_error_envelope(self): + error = {"code": "unknown", "message": + "rpc error: code = Internal desc = pick task: CreateTaskForRunner: " + "update run 3673: run has changed"} + headers = {"Content-Type": "application/json"} + body = json.dumps(error).encode() + for status, payload, response_headers in [ + (502, body, headers), (200, body, headers), + (500, body, {"Content-Type": "text/html"}), + (500, body, {**headers, "Content-Encoding": "gzip"}), + (500, body[:-1], headers), (500, b'[]', headers), + (500, json.dumps({**error, "code": "internal"}).encode(), headers), + (500, json.dumps({**error, "message": error["message"] + " extra"}).encode(), headers), + ]: + with self.subTest(status=status, payload=payload, headers=response_headers): + self.assertFalse(MODULE.fetch_conflict_rolled_back(status, payload, response_headers)) + def test_task_report_without_observed_assignment_requires_idle_bootstrap(self): self.assertEqual(self.request(UPDATE, state(42))[0], 200) self.wait_for(lambda: self.gate.control("status")["uncertain"])