From 4380d453d7babf3a4329724cb9afd254d9532bda Mon Sep 17 00:00:00 2001 From: Linghong Date: Tue, 22 Sep 2026 14:40:55 +0000 Subject: [PATCH] =?UTF-8?q?=E4=BF=AE=E5=A4=8D=20CI=20=E7=BC=93=E5=AD=98?= =?UTF-8?q?=E5=AF=B9=E8=B1=A1=E8=AF=AF=E5=88=A4=E4=B8=8E=E4=BA=A7=E7=89=A9?= =?UTF-8?q?=E4=B8=8B=E8=BD=BD=E5=AE=8C=E6=95=B4=E6=80=A7=E6=A3=80=E6=9F=A5?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 按 ZIP 成员内容和元数据识别等价缓存对象，保留真实冲突拒绝规则。 核对产物大小与 ZIP 完整性，为截断和临时传输失败增加有界重试。 补充回归测试并同步部署文档与共享记忆。 通过 80 项缓存测试、17 项工作流测试及真实产物和 HTTP 截断验证。 --- deploy/container/README.md | 4 + .../shared-memory/development-workflow.md | 2 + scripts/gitea_cache_snapshot.py | 100 +++++++++++++- scripts/maintain-gitea-rust-cache.py | 120 +++++++++++++---- scripts/test_gitea_cache_maintenance.py | 122 +++++++++++++++++- scripts/test_gitea_cache_snapshot.py | 46 +++++++ 6 files changed, 361 insertions(+), 33 deletions(-) diff --git a/deploy/container/README.md b/deploy/container/README.md index 0f992bafa..801cbe671 100644 --- a/deploy/container/README.md +++ b/deploy/container/README.md @@ -112,6 +112,10 @@ runner 配置保留原 `ubuntu-latest` 映射，`genarrative-ci` 继续映射到 ### Rust 测试组编译对象快照 +宿主下载每份 artifact 时核对 Gitea `size_in_bytes`、响应 `Content-Length`（若提供）和实际字节数，并在下载完成后立即检查 ZIP 格式与 CRC。截断、损坏归档或临时传输故障最多尝试 3 次，每次重新获取签名下载地址；不向签名地址转发 API Token。重试仍失败则删除临时下载并保留现役镜像，不能把 EOF 当作完整下载成功。 + +同一 sccache key 的完整对象 SHA 不同时，不直接视为编译结果不同：sccache 对象 ZIP 的成员写入顺序可能不同。仅对不同 job 新增对象之间的冲突，按成员名核对内容 SHA-256、权限及 ZIP 元数据，全部一致才保留一份原始对象并更新使用时间；真实内容差异、异常 ZIP 和继承对象冲突仍拒绝。此比较不改写缓存 key 或对象，不依赖 CRC 代替内容校验。 + 自动维护由宿主 systemd timer 调用 `scripts/maintain-gitea-rust-cache.py`，只管理 Gitea CI 测试镜像，不修改 Jenkins、生产发布、本地开发或客户端发行构建。六个 Rust job 仅在 master push 中导出本次 CI 新增的 sccache 对象；已命中的继承对象只上传新近使用时间，通过 Gitea 原生 V4 artifact 接口上传；PR 不发布。维护器选择已结束且六组产物完整的最新 master run，校验提交、任务尝试、工具链与来源镜像，与六组实际使用的同一镜像快照合并去重，并按新近使用时间限制快照总容量为 4 GiB，然后从无对象缓存基础镜像组装新镜像，**不重复执行 Cargo 预热编译，也不要求源 run 事先全绿**。缺组、取消或校验失败时保留现役版，不混合不同 run 的对象来假装完整快照。 维护 journal 分阶段记录来源 run、基础镜像重建或复用、artifact 下载、对象合并、镜像组装校验、导出、载入及空闲等待；长操作记录开始和结束耗时，失败输出对应私有 `artifacts//build.log` 路径。构建的详细下载与 Docker 输出仍只写该日志，不回显 Token、命令环境或认证配置。 diff --git a/docs/project-memory/shared-memory/development-workflow.md b/docs/project-memory/shared-memory/development-workflow.md index bb3a362a7..cabe36649 100644 --- a/docs/project-memory/shared-memory/development-workflow.md +++ b/docs/project-memory/shared-memory/development-workflow.md @@ -96,6 +96,8 @@ SpacetimeDB 任务统一先读取 `.codex/skills/genarrative-spacetimedb/SKILL.m ## Gitea CI 依赖闭合 +缓存维护下载必须核对 artifact 元数据大小与实际响应，并立即检查 ZIP 完整性；有长度上限不等于能检测短读。网络或归档损坏最多重试 3 次且重新取签名链接。sccache 同 key 的不同 ZIP 成员排列会改变整个对象 SHA；新增对象去重仅在成员内容 SHA、权限和 ZIP 元数据均一致时接受排列差异，真实内容及继承对象冲突仍拒绝。禁止通过任取一份冲突对象绕过完整性契约。 + Buildx 0.30.1 的 `inspect` 不支持 `--format`，builder 驱动校验读取普通输出的 `Driver:` 字段。相关命令须在宿主真实插件上验证；测试替身应拒绝不支持的参数，避免把模拟命令成功误当兼容性证据。 Gitea 基础镜像通过专用 `genarrative-ci-images` Buildx builder 持久复用 Cargo/npm 下载缓存；稳定 cache mount 与 commit、lock 哈希无关，以 `sharing=locked` 隔离并发写入，仅供可信宿主构建、不开放给 PR。最终镜像显式物化当前依赖下载快照，仍不包含 node_modules/target 或上一版 sccache 层。首次可用 `seed-downloads` 从可信完整 Image ID 提取包缓存，操作账号须与维护服务一致；部署要求及 builder GC 空间目标见 `deploy/container/README.md`。构建上下文必须覆盖 AGC vendor 与编辑器 bridge 的全部本地 path manifest，普通源码变化不应使依赖层失效。维护 journal 提供阶段耗时和失败 build.log 定位。 diff --git a/scripts/gitea_cache_snapshot.py b/scripts/gitea_cache_snapshot.py index 46e117e15..f270ea1b8 100644 --- a/scripts/gitea_cache_snapshot.py +++ b/scripts/gitea_cache_snapshot.py @@ -167,6 +167,14 @@ def _snapshot_member(archive: Path) -> tuple[zipfile.ZipFile, zipfile.ZipInfo]: raise +def validate_artifact_zip(archive: Path) -> None: + """Validate the bounded stored ZIP envelope before consuming an artifact.""" + bundle, _ = _snapshot_member(archive) + with bundle: + if bundle.testzip() is not None: + raise SnapshotError("artifact ZIP CRC check failed") + + def _tar_stream(archive: Path): bundle, member = _snapshot_member(archive) try: @@ -440,6 +448,94 @@ def _file_sha256(path: Path) -> str: digest.update(chunk) +def _object_zip_signature(stream: BinaryIO, expected: _Object) -> tuple: + """只允许 sccache 对象的 ZIP 成员排列不同，不放宽内容或元数据校验。""" + with tempfile.TemporaryFile() as temporary: + digest = hashlib.sha256() + total = 0 + while chunk := stream.read(1024 * 1024): + total += len(chunk) + if total > expected.size: + raise SnapshotError("cache object grew while comparing ZIP contents") + digest.update(chunk) + temporary.write(chunk) + if (total, digest.hexdigest()) != (expected.size, expected.sha256): + raise SnapshotError("cache object checksum changed while comparing ZIP contents") + temporary.seek(0) + with zipfile.ZipFile(temporary) as bundle: + infos = bundle.infolist() + if (not infos or len(infos) > MAX_OBJECTS + or len({info.filename for info in infos}) != len(infos) + or sum(info.file_size for info in infos) > expected.size): + raise SnapshotError("invalid sccache ZIP member set") + members = [] + for info in sorted(infos, key=lambda entry: entry.filename): + _safe_zip_path(info.filename) + mode = info.external_attr >> 16 + if (info.is_dir() or info.flag_bits & 1 or stat.S_ISLNK(mode) + or stat.S_IFMT(mode) not in (0, stat.S_IFREG) + or info.compress_type != zipfile.ZIP_STORED): + raise SnapshotError("unsupported sccache ZIP member") + with bundle.open(info) as contents: + size, checksum = _hash_stream(contents, info.file_size) + members.append(( + info.filename, size, checksum, info.CRC, info.compress_size, + info.compress_type, info.date_time, info.flag_bits, + info.external_attr, info.internal_attr, info.create_system, + info.create_version, info.extract_version, info.reserved, + info.extra, info.comment, + )) + return bundle.comment, tuple(members) + + +def _equivalent_delta_paths(archives: Sequence[_Archive]) -> set[str]: + variants: dict[str, set[tuple[int, str | None]]] = {} + for archive in archives: + for obj in archive.objects: + variants.setdefault(obj.path, set()).add((obj.size, obj.sha256)) + conflicts = {path for path, versions in variants.items() if len(versions) > 1} + if not conflicts: + return conflicts + for path in conflicts: + if len({size for size, _ in variants[path]}) != 1: + raise SnapshotError(f"conflicting content for duplicate object: {path}") + signatures = {} + # 每份归档最多额外顺序读取一次，仅将冲突对象暂存到磁盘供 ZIP 随机读取。 + for archive in archives: + wanted = {obj.path: obj for obj in archive.objects if obj.path in conflicts} + if not wanted: + continue + if _archive_signature(archive.input.archive) != archive.signature: + raise SnapshotError("artifact archive changed while comparing cache objects") + bundle, raw, tar = _tar_stream(archive.input.archive) + try: + for member in tar: + expected = wanted.get(member.name) + if expected is None: + continue + if not member.isreg() or member.size != expected.size: + raise SnapshotError("cache object changed while comparing ZIP contents") + stream = tar.extractfile(member) + if stream is None: + raise SnapshotError("cache object is missing while comparing ZIP contents") + try: + with stream: + signature = _object_zip_signature(stream, expected) + except (SnapshotError, zipfile.BadZipFile, NotImplementedError) as error: + raise SnapshotError(f"conflicting content for duplicate object: {member.name}") from error + if member.name in signatures and signatures[member.name] != signature: + raise SnapshotError(f"conflicting content for duplicate object: {member.name}") + signatures[member.name] = signature + del wanted[member.name] + if wanted: + raise SnapshotError("cache objects disappeared while comparing ZIP contents") + finally: + tar.close() + raw.close() + bundle.close() + return conflicts + + def _select_objects( archives: Sequence[_Archive], base_root: Path, @@ -447,6 +543,7 @@ def _select_objects( maximum: int, ) -> tuple[_Object, ...]: merged = dict(base) + equivalent_paths = _equivalent_delta_paths(archives) delta_paths = {obj.path for archive in archives for obj in archive.objects} touched_paths = {touch.path for archive in archives for touch in archive.touched} overlap = delta_paths & touched_paths @@ -490,7 +587,8 @@ def _select_objects( mtime_ns=max(current.mtime_ns, candidate.mtime_ns), source_index=None, ) - elif (current.size, current.sha256) != (candidate.size, candidate.sha256): + elif ((current.size, current.sha256) != (candidate.size, candidate.sha256) + and candidate.path not in equivalent_paths): raise SnapshotError(f"conflicting content for duplicate object: {candidate.path}") elif candidate.mtime_ns > current.mtime_ns: merged[candidate.path] = _Object( diff --git a/scripts/maintain-gitea-rust-cache.py b/scripts/maintain-gitea-rust-cache.py index 1219fc3a1..fc948355d 100644 --- a/scripts/maintain-gitea-rust-cache.py +++ b/scripts/maintain-gitea-rust-cache.py @@ -5,6 +5,7 @@ import argparse import contextlib import datetime import hashlib +import http.client import json import os from pathlib import Path @@ -18,8 +19,11 @@ import time import urllib.error import urllib.parse import urllib.request +import zipfile -from gitea_cache_snapshot import ArtifactIdentity, ArtifactInput, merge_snapshots +from gitea_cache_snapshot import ( + ArtifactIdentity, ArtifactInput, SnapshotError, merge_snapshots, validate_artifact_zip, +) from gitea_cache_upload_cleanup import cleanup_upload_chunks @@ -46,6 +50,7 @@ RUST_JOBS = set(RUST_JOB_IDS) ARTIFACT_PREFIX = "rust-cache-v1-" MAX_DOWNLOAD = 4 * 1024 ** 3 + 129 * 1024 ** 2 EXPORT_STEP = "Publish master Rust cache artifact" +DOWNLOAD_ATTEMPTS = 3 def now(): @@ -183,35 +188,90 @@ class Api: raise RuntimeError(f"Gitea API HTTP {error.code}") from None return content if raw else (json.loads(content) if content else None) - def download(self, path, destination): - """REST V4 archive redirects to a signed URL; never forward the API token.""" + def download(self, path, destination, expected_size): + """Fetch one artifact archive, retrying incomplete signed-URL transfers.""" + if (isinstance(expected_size, bool) or not isinstance(expected_size, int) + or expected_size <= 0): + raise RuntimeError("artifact has invalid size metadata") + if expected_size > MAX_DOWNLOAD: + raise RuntimeError("artifact exceeds per-job size limit") + + for attempt in range(DOWNLOAD_ATTEMPTS): + try: + self._download_once(path, destination, expected_size) + return + except RetryableArtifactDownload as error: + destination.unlink(missing_ok=True) + log(f"artifact download attempt {attempt + 1}/{DOWNLOAD_ATTEMPTS} failed: {error}") + if attempt + 1 == DOWNLOAD_ATTEMPTS: + raise RuntimeError("artifact download remained incomplete after retries") from error + time.sleep(1 << attempt) + except Exception: + destination.unlink(missing_ok=True) + raise + + def _download_once(self, path, destination, expected_size): + """Obtain a fresh signed URL and validate its complete ZIP response.""" token = self.token_file.read_text().strip() url = self.url + "/" + path.lstrip("/") opener = urllib.request.build_opener(NoRedirect()) request = urllib.request.Request(url, headers={"Authorization": "token " + token}) + target = None try: - response = opener.open(request, timeout=60) - except urllib.error.HTTPError as error: - error.close() - if error.code not in (301, 302, 303, 307, 308): - raise RuntimeError(f"artifact download HTTP {error.code}") from None - target = urllib.parse.urljoin(url, error.headers.get("Location", "")) - parsed, origin = urllib.parse.urlsplit(target), urllib.parse.urlsplit(self.url) - if (parsed.scheme != "https" or parsed.netloc != origin.netloc - or parsed.username or parsed.password or target == url): - raise RuntimeError("artifact redirect must stay on configured HTTPS Gitea origin") from None - response = opener.open(target, timeout=60) - try: - with response, destination.open("wb") as out: - total = 0 - while chunk := response.read(1024 * 1024): - total += len(chunk) - if total > MAX_DOWNLOAD: + try: + response = opener.open(request, timeout=60) + except urllib.error.HTTPError as error: + error.close() + if error.code not in (301, 302, 303, 307, 308): + if error.code >= 500: + raise RetryableArtifactDownload(f"artifact download HTTP {error.code}") from None + raise RuntimeError(f"artifact download HTTP {error.code}") from None + target = urllib.parse.urljoin(url, error.headers.get("Location", "")) + parsed, origin = urllib.parse.urlsplit(target), urllib.parse.urlsplit(self.url) + if (parsed.scheme != "https" or parsed.netloc != origin.netloc + or parsed.username or parsed.password or target == url): + raise RuntimeError("artifact redirect must stay on configured HTTPS Gitea origin") from None + if target is not None: + try: + response = opener.open(target, timeout=60) + except urllib.error.HTTPError as error: + error.close() + if error.code >= 500: + raise RetryableArtifactDownload(f"artifact download HTTP {error.code}") from None + raise RuntimeError(f"artifact download HTTP {error.code}") from None + with response: + content_length = getattr(response, "headers", {}).get("Content-Length") + if content_length is not None: + try: + content_length = int(content_length) + except (TypeError, ValueError): + raise RetryableArtifactDownload("artifact response has invalid Content-Length") from None + if content_length > MAX_DOWNLOAD: raise RuntimeError("artifact exceeds per-job size limit") - out.write(chunk) - except Exception: - destination.unlink(missing_ok=True) - raise + if content_length != expected_size: + raise RetryableArtifactDownload( + f"artifact response size differs: expected {expected_size} bytes, " + f"Content-Length is {content_length}") + with destination.open("wb") as out: + total = 0 + while chunk := response.read(1024 * 1024): + total += len(chunk) + if total > MAX_DOWNLOAD: + raise RuntimeError("artifact exceeds per-job size limit") + if total > expected_size: + raise RetryableArtifactDownload( + f"artifact response exceeds metadata: expected {expected_size} bytes, " + f"received at least {total}") + out.write(chunk) + if total != expected_size: + raise RetryableArtifactDownload( + f"artifact response is truncated: expected {expected_size} bytes, received {total}") + try: + validate_artifact_zip(destination) + except (SnapshotError, zipfile.BadZipFile): + raise RetryableArtifactDownload("artifact response is not a valid ZIP") from None + except (http.client.HTTPException, urllib.error.URLError, OSError) as error: + raise RetryableArtifactDownload("artifact transfer failed") from error def pages(self, path, key): separator = "&" if "?" in path else "?" @@ -230,6 +290,10 @@ class NoRedirect(urllib.request.HTTPRedirectHandler): return None +class RetryableArtifactDownload(RuntimeError): + """A signed archive transfer may be retried with a newly issued URL.""" + + class Maintenance: def __init__(self, config): self.config = config @@ -371,7 +435,10 @@ class Maintenance: if len(used) != 1: break images.update(used) - selected.append({"id": matches[0]["id"], "name": name, + size = matches[0].get("size_in_bytes") + if isinstance(size, bool) or not isinstance(size, int) or size <= 0 or size > MAX_DOWNLOAD: + break + selected.append({"id": matches[0]["id"], "name": name, "size_in_bytes": size, "job": RUST_JOB_IDS[job["name"]], "attempt": job["run_attempt"]}) if len(selected) != len(RUST_JOB_IDS) or len(images) != 1: continue @@ -431,7 +498,8 @@ class Maintenance: archive = work / (str(export["id"]) + ".zip") with operation(f"download cache artifact job={export['job']} attempt={export['attempt']}", build_log=build_log): - self.api.download(self.repo_api + f'/actions/artifacts/{export["id"]}/zip', archive) + self.api.download(self.repo_api + f'/actions/artifacts/{export["id"]}/zip', archive, + export["size_in_bytes"]) inputs.append(ArtifactInput(archive, ArtifactIdentity( self.config["repository"], source["run_id"], export["attempt"], export["job"], sha))) snapshot = work / "snapshot" diff --git a/scripts/test_gitea_cache_maintenance.py b/scripts/test_gitea_cache_maintenance.py index 730e44049..ed7c48b3e 100644 --- a/scripts/test_gitea_cache_maintenance.py +++ b/scripts/test_gitea_cache_maintenance.py @@ -11,6 +11,7 @@ import re import tempfile import unittest from unittest.mock import patch +import zipfile SCRIPT = Path(__file__).with_name("maintain-gitea-rust-cache.py") @@ -29,6 +30,26 @@ def runner_config(image: str) -> str: return f'runners:\n - "genarrative-ci:docker://{image}"\n' +def zip_bytes(*, size=None) -> bytes: + """Make a valid stored ZIP, optionally padded to an exact download size.""" + content = b"x" * 40000 if size else b"cache artifact" + output = io.BytesIO() + with zipfile.ZipFile(output, "w", zipfile.ZIP_STORED) as archive: + archive.writestr("snapshot.tar", content) + value = output.getvalue() + if size is None: + return value + if not len(value) < size <= len(value) + 65535: + raise AssertionError("requested ZIP size cannot be represented by its comment") + output = io.BytesIO(value) + with zipfile.ZipFile(output, "a") as archive: + archive.comment = b"p" * (size - len(value)) + value = output.getvalue() + if len(value) != size: + raise AssertionError("ZIP comment did not produce the requested size") + return value + + class FakeApi: def __init__(self): self.requests: list[str] = [] @@ -299,7 +320,7 @@ class GiteaCacheMaintenanceTest(unittest.TestCase): "steps": [{"name": maintenance_module.EXPORT_STEP, "conclusion": "success"}]} for i, name in enumerate(maintenance_module.RUST_JOB_IDS, start=1)] artifacts = [{"id": job["id"], "name": "rust-cache-v1-" + maintenance_module.RUST_JOB_IDS[job["name"]] - + "-attempt-1", "expired": False, "workflow_run": run} for job in jobs] + + "-attempt-1", "size_in_bytes": 10, "expired": False, "workflow_run": run} for job in jobs] class Api: def pages(self, path, key): return iter({"workflow_runs": [run], "jobs": jobs, "artifacts": artifacts}[key]) @@ -320,6 +341,7 @@ class GiteaCacheMaintenanceTest(unittest.TestCase): result = self.select_source(instance) self.assertEqual(result["run_id"], 44) self.assertEqual(len(result["exports"]), 6) + self.assertEqual({item["size_in_bytes"] for item in result["exports"]}, {10}) run["event"] = "pull_request" self.assertIsNone(self.select_source(instance)) run["event"] = "push" @@ -375,7 +397,7 @@ class GiteaCacheMaintenanceTest(unittest.TestCase): return "rustc test\n" return "" instance.docker = docker - instance.api.download = lambda path, destination: destination.write_bytes(b"download") + instance.api.download = lambda path, destination, expected_size: destination.write_bytes(b"download") def merge(inputs, output, **kwargs): self.assertEqual(len(inputs), 6) self.assertEqual(kwargs["expected_inherited_source_sha"], "b" * 40) @@ -419,23 +441,111 @@ class GiteaCacheMaintenanceTest(unittest.TestCase): def test_signed_download_drops_token_and_rejects_other_origins(self): api = maintenance_module.Api(self.config["api_url"], self.token) destination = self.root / "artifact.zip" + archive = zip_bytes() seen = [] target = ["https://gitea.example.test/api/v1/repos/team/project/actions/artifacts/1/zip/raw?sig=test"] + + class Response(io.BytesIO): + def __init__(self, value): + super().__init__(value) + self.headers = {"Content-Length": str(len(value))} + class Opener: def open(self, request, timeout): seen.append(request) if not isinstance(request, str): raise maintenance_module.urllib.error.HTTPError(request.full_url, 302, "Found", {"Location": target[0]}, None) - return io.BytesIO(b"archive") + return Response(archive) + with patch.object(maintenance_module.urllib.request, "build_opener", return_value=Opener()): - api.download("repos/team/project/actions/artifacts/1/zip", destination) - self.assertEqual(destination.read_bytes(), b"archive") + api.download("repos/team/project/actions/artifacts/1/zip", destination, len(archive)) + self.assertEqual(destination.read_bytes(), archive) self.assertEqual(seen[0].get_header("Authorization"), "token test-token") self.assertIsInstance(seen[1], str) # signed URL request has no Authorization header target[0] = "https://other.example.test/download" with self.assertRaisesRegex(RuntimeError, "configured HTTPS Gitea origin"): - api.download("repos/team/project/actions/artifacts/1/zip", self.root / "rejected.zip") + api.download("repos/team/project/actions/artifacts/1/zip", self.root / "rejected.zip", len(archive)) self.assertFalse((self.root / "rejected.zip").exists()) + self.assertEqual(len(seen), 3) # invalid redirect is deterministic and is not retried + + def test_download_retries_transport_and_truncated_response_with_fresh_signed_urls(self): + api = maintenance_module.Api(self.config["api_url"], self.token) + destination = self.root / "artifact.zip" + archive = zip_bytes(size=100000) + seen, signed_responses = [], [] + origin_requests = 0 + target = "https://gitea.example.test/api/v1/repos/team/project/actions/artifacts/1/zip/raw?sig=test" + + class Response(io.BytesIO): + def __init__(self, value): + super().__init__(value) + self.headers = {"Content-Length": "100000"} + + class Opener: + def open(self, request, timeout): + nonlocal origin_requests + seen.append(request) + if not isinstance(request, str): + origin_requests += 1 + if origin_requests == 1: + raise maintenance_module.urllib.error.URLError("connection reset") + raise maintenance_module.urllib.error.HTTPError(request.full_url, 302, "Found", {"Location": target}, None) + signed_responses.append(request) + return Response(b"x" * 512 if len(signed_responses) == 1 else archive) + + with patch.object(maintenance_module.urllib.request, "build_opener", return_value=Opener()), \ + patch.object(maintenance_module.time, "sleep") as sleep: + api.download("repos/team/project/actions/artifacts/1/zip", destination, 100000) + self.assertEqual(destination.read_bytes(), archive) + self.assertEqual(len(signed_responses), 2) + self.assertEqual(origin_requests, 3) + self.assertEqual([item.args for item in sleep.call_args_list], [(1,), (2,)]) + + def test_download_rejects_malformed_zip_after_retries(self): + api = maintenance_module.Api(self.config["api_url"], self.token) + destination = self.root / "artifact.zip" + seen = [] + target = "https://gitea.example.test/api/v1/repos/team/project/actions/artifacts/1/zip/raw?sig=test" + + class Response(io.BytesIO): + headers = {"Content-Length": "9"} + + class Opener: + def open(self, request, timeout): + seen.append(request) + if not isinstance(request, str): + raise maintenance_module.urllib.error.HTTPError(request.full_url, 302, "Found", {"Location": target}, None) + return Response(b"not a zip") + + with patch.object(maintenance_module.urllib.request, "build_opener", return_value=Opener()), \ + patch.object(maintenance_module.time, "sleep") as sleep: + with self.assertRaisesRegex(RuntimeError, "incomplete after retries"): + api.download("repos/team/project/actions/artifacts/1/zip", destination, 9) + self.assertFalse(destination.exists()) + self.assertEqual(sum(not isinstance(item, str) for item in seen), 3) + self.assertEqual([item.args for item in sleep.call_args_list], [(1,), (2,)]) + + def test_download_rejects_valid_zip_when_its_size_differs_from_metadata(self): + api = maintenance_module.Api(self.config["api_url"], self.token) + destination = self.root / "artifact.zip" + archive = zip_bytes() + seen = [] + target = "https://gitea.example.test/api/v1/repos/team/project/actions/artifacts/1/zip/raw?sig=test" + + class Opener: + def open(self, request, timeout): + seen.append(request) + if not isinstance(request, str): + raise maintenance_module.urllib.error.HTTPError(request.full_url, 302, "Found", {"Location": target}, None) + return io.BytesIO(archive) + + with patch.object(maintenance_module.urllib.request, "build_opener", return_value=Opener()), \ + patch.object(maintenance_module.time, "sleep") as sleep: + with self.assertRaisesRegex(RuntimeError, "incomplete after retries"): + api.download("repos/team/project/actions/artifacts/1/zip", destination, len(archive) + 1) + self.assertFalse(destination.exists()) + self.assertEqual(sum(not isinstance(item, str) for item in seen), 3) + self.assertEqual([item.args for item in sleep.call_args_list], [(1,), (2,)]) def test_pending_chunk_cleanup_requires_old_terminal_master_run(self): instance = self.maintenance({"versions": [], "current": None, "attempts": [{"run_id": 4}]}) diff --git a/scripts/test_gitea_cache_snapshot.py b/scripts/test_gitea_cache_snapshot.py index 135cd903b..e94bf2742 100644 --- a/scripts/test_gitea_cache_snapshot.py +++ b/scripts/test_gitea_cache_snapshot.py @@ -37,6 +37,16 @@ def object_path(character: str) -> str: return f"objects/{key[0]}/{key[1]}/{key}" +def cache_object(entries, *, mode=0o100644) -> bytes: + output = io.BytesIO() + with zipfile.ZipFile(output, "w", compression=zipfile.ZIP_STORED) as bundle: + for name, contents in entries: + info = zipfile.ZipInfo(name) + info.external_attr = mode << 16 + bundle.writestr(info, contents) + return output.getvalue() + + class SnapshotMergeTest(unittest.TestCase): def setUp(self) -> None: self.temporary_directory = tempfile.TemporaryDirectory(ignore_cleanup_errors=True) @@ -207,6 +217,42 @@ class SnapshotMergeTest(unittest.TestCase): with self.assertRaisesRegex(snapshot.SnapshotError, "conflicting content"): self.merge(inputs) + def test_deduplicates_cache_zip_member_order_preserving_payload_and_newest_touch(self) -> None: + path = object_path("a") + entries = [("lib.rlib", b"compiled"), ("lib.rmeta", b"metadata"), ("stderr", b"warning")] + first = cache_object(entries) + second = cache_object(list(reversed(entries))) + self.assertEqual(len(first), len(second)) + self.assertNotEqual(hashlib.sha256(first).digest(), hashlib.sha256(second).digest()) + inputs = [ + self.archive("smoke", "smoke", 45, [(path, first, 100)]), + self.archive("lane-1", "lane-1", 45, [(path, second, 300)]), + self.archive("lane-2", "lane-2", 45, [(path, second, 200)]), + ] + result = self.merge(inputs) + output = self.root / "merged" / path + self.assertEqual((result.object_count, result.total_bytes), (1, len(first))) + self.assertEqual(output.read_bytes(), first) + self.assertEqual(output.stat().st_mtime_ns, 300) + + def test_rejects_cache_zip_payload_or_permissions_conflicts(self) -> None: + path = object_path("a") + first = cache_object([("lib.rlib", b"one"), ("stderr", b"err")]) + variants = { + "payload": cache_object([("stderr", b"err"), ("lib.rlib", b"two")]), + "permissions": cache_object([("stderr", b"err"), ("lib.rlib", b"one")], mode=0o100755), + } + for name, second in variants.items(): + with self.subTest(name=name): + self.assertEqual(len(first), len(second)) + inputs = [ + self.archive("first", "smoke", 45, [(path, first, 100)]), + self.archive("second", "lane-1", 45, [(path, second, 200)]), + ] + with self.assertRaisesRegex(snapshot.SnapshotError, "conflicting content"): + self.merge(inputs) + self.assertFalse((self.root / "merged").exists()) + def test_rejects_delta_that_conflicts_with_inherited_key(self) -> None: path = object_path("f") self.base_object(path, b"base", 1)