修复 CI 缓存对象误判与产物下载完整性检查 #483
@@ -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/<source-sha>/build.log` 路径。构建的详细下载与 Docker 输出仍只写该日志,不回显 Token、命令环境或认证配置。
|
||||
|
||||
@@ -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 定位。
|
||||
|
||||
@@ -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(
|
||||
|
||||
@@ -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"
|
||||
|
||||
@@ -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}]})
|
||||
|
||||
@@ -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)
|
||||
|
||||
Reference in New Issue
Block a user