From b73c980f093973aae67422b12371c3fc04c1629e Mon Sep 17 00:00:00 2001 From: "moxt-ai[bot]" Date: Thu, 9 Jul 2026 14:37:48 +0800 Subject: [PATCH 1/5] =?UTF-8?q?fix:=20cache=20corruption=20=E6=97=B6?= =?UTF-8?q?=E8=87=AA=E5=8A=A8=20fallback=20=E9=87=8D=E6=96=B0=E7=94=9F?= =?UTF-8?q?=E6=88=90=EF=BC=8C=E4=B8=8A=E4=BC=A0=E5=A4=B1=E8=B4=A5=E4=B8=8D?= =?UTF-8?q?=E9=98=BB=E6=96=AD=20job?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 代理链路不稳定导致 S3 body 截断，出现两类故障： 1. ZstdError：从 S3 下载的 .tar.zst 残缺，本地解压失败 2. IncompleteBody：上传阶段 body 与 Content-Length 不一致 修复： - cache.py load()：本地缓存 ZstdError 时删除本地并从云端重新下载； 云端缓存也损坏时清理本地并 re-raise，让 target.py 走重建路径 - target.py execute()：NORMAL 模式下 load() 抛出 ZstdError 时 fallback 到重新构建并保存新缓存，不再 crash job - 上传失败（CloudStorageError）从 sys.exit(1) 改为 warning， 缓存上传失败不影响构建结果 - 新增对应测试：corrupt local cache fallback、corrupt cloud cache re-raise Co-Authored-By: via Moxt --- src/quack/cache.py | 11 ++++++-- src/quack/cache_test.py | 56 ++++++++++++++++++++++++++++++++++++++ src/quack/models/target.py | 15 +++++++--- 3 files changed, 76 insertions(+), 6 deletions(-) diff --git a/src/quack/cache.py b/src/quack/cache.py index 447d715..ff1fde3 100644 --- a/src/quack/cache.py +++ b/src/quack/cache.py @@ -9,6 +9,7 @@ from pathlib import Path from typing import final, override +import zstandard as zstd from loguru import logger from xdg_base_dirs import xdg_cache_home @@ -204,13 +205,19 @@ def load(self, target: Target, update_access_time: bool = True) -> None: if update_access_time: self.update_access_time(target) return - except ChecksumError: + except (ChecksumError, zstd.ZstdError): logger.warning("本地缓存已损坏，从云存储重新下载") + shutil.rmtree(self.local_backend.get_cache_path(target), ignore_errors=True) logger.info(f"正在从云存储加载 Target {target.name} 的缓存...") self.cloud_client.download(self.get_archive_path(target), self.local_backend.get_archive_path(target)) self.cloud_client.download(self.get_metadata_path(target), self.local_backend.get_metadata_path(target)) - self.local_backend.load(target) + try: + self.local_backend.load(target) + except zstd.ZstdError: + logger.warning(f"云存储中 Target {target.name} 的缓存已损坏，将重新生成") + shutil.rmtree(self.local_backend.get_cache_path(target), ignore_errors=True) + raise if update_access_time: self.update_access_time(target) diff --git a/src/quack/cache_test.py b/src/quack/cache_test.py index 1bb1b10..f408fed 100644 --- a/src/quack/cache_test.py +++ b/src/quack/cache_test.py @@ -1,6 +1,9 @@ import os from unittest import mock +import pytest +import zstandard as zstd + from quack.cache import TargetCacheBackendTypeCloud from quack.config import Config @@ -53,6 +56,59 @@ def test_load_not_exists( # 验证本地加载被调用 assert mock_local_backend.return_value.load.call_count == 1 + @mock.patch("quack.cache.shutil.rmtree") + @mock.patch("quack.cache.CloudClient") + @mock.patch("quack.cache.TargetCacheBackendTypeLocal") + def test_load_corrupt_local_cache_falls_back_to_cloud( + self, + mock_local_backend: mock.Mock, + mock_cloud_client_class: mock.Mock, + mock_rmtree: mock.Mock, + mock_test_spec: mock.Mock, + ): + mock_cloud_client = mock.Mock() + mock_cloud_client_class.return_value = mock_cloud_client + + config = Config.model_construct() + target = mock_test_spec.targets["quack:test"] + target._checksum_value = "" + backend = TargetCacheBackendTypeCloud(config, mock_test_spec.app_name) + + mock_local_backend.return_value.exists.return_value = True + # First call (local) raises; second call (after re-download) succeeds + mock_local_backend.return_value.load.side_effect = [zstd.ZstdError("did not decompress full frame"), None] + + backend.load(target) + + assert mock_cloud_client.download.call_count == 2 + assert mock_rmtree.called + + @mock.patch("quack.cache.shutil.rmtree") + @mock.patch("quack.cache.CloudClient") + @mock.patch("quack.cache.TargetCacheBackendTypeLocal") + def test_load_corrupt_cloud_cache_raises( + self, + mock_local_backend: mock.Mock, + mock_cloud_client_class: mock.Mock, + mock_rmtree: mock.Mock, + mock_test_spec: mock.Mock, + ): + mock_cloud_client = mock.Mock() + mock_cloud_client_class.return_value = mock_cloud_client + + config = Config.model_construct() + target = mock_test_spec.targets["quack:test"] + target._checksum_value = "" + backend = TargetCacheBackendTypeCloud(config, mock_test_spec.app_name) + + mock_local_backend.return_value.exists.return_value = False + mock_local_backend.return_value.load.side_effect = zstd.ZstdError("did not decompress full frame") + + with pytest.raises(zstd.ZstdError): + backend.load(target) + + assert mock_rmtree.called + @mock.patch.dict(os.environ, {"PATH": "/usr/bin:/bin"}, clear=True) @mock.patch("quack.cache.CloudClient") @mock.patch("quack.cache.TargetCacheBackendTypeLocal") diff --git a/src/quack/models/target.py b/src/quack/models/target.py index 37e54b5..4c30b73 100644 --- a/src/quack/models/target.py +++ b/src/quack/models/target.py @@ -9,6 +9,7 @@ from pathlib import Path from typing import TYPE_CHECKING +import zstandard as zstd from loguru import logger from pydantic import Field @@ -113,14 +114,20 @@ def execute( if cache_exists: logger.info("找到缓存，直接从缓存加载...") - cache.load() - else: + try: + cache.load() + except zstd.ZstdError: + logger.warning(f"缓存已损坏，将重新生成 Target {self.name}...") + self.prepare_deps(config, app_name, cache_backend) + self.operations.build.execute() + cache_exists = False + + if not cache_exists: logger.info(f"正在存入缓存，路径：{self.cache_path}") try: cache.save() except CloudStorageError as e: - logger.error(f"存入缓存失败：{e}") - sys.exit(1) + logger.warning(f"上传缓存失败，将跳过云端缓存：{e}") elapsed = time.time() - start_time logger.success(f"Target {self.name} 执行完毕！") From f9950281a5c99fe4b63318548b3fe70125a7971b Mon Sep 17 00:00:00 2001 From: tigerBeA Date: Thu, 9 Jul 2026 15:12:01 +0800 Subject: [PATCH 2/5] fix: handle transient cloud cache failures --- src/quack/cache.py | 22 +++++---- src/quack/cache_test.py | 50 +++++++++++++++++++++ src/quack/exceptions.py | 7 ++- src/quack/models/target.py | 8 ++-- src/quack/models/target_test.py | 34 ++++++++++++++ src/quack/utils/cloud.py | 80 ++++++++++++++++++++++++++++++--- src/quack/utils/cloud_test.py | 47 +++++++++++++++++++ 7 files changed, 228 insertions(+), 20 deletions(-) create mode 100644 src/quack/utils/cloud_test.py diff --git a/src/quack/cache.py b/src/quack/cache.py index ff1fde3..43ab379 100644 --- a/src/quack/cache.py +++ b/src/quack/cache.py @@ -15,7 +15,7 @@ from quack.config import Config from quack.consts import CACHE_METADATA_FILENAME -from quack.exceptions import ChecksumError +from quack.exceptions import ChecksumError, CloudStorageTransientError from quack.models.target import Target from quack.utils.archiver import Archiver from quack.utils.ci_environment import CIEnvironment @@ -193,10 +193,13 @@ def exists(self, target: Target) -> bool: def update_access_time(self, target: Target) -> None: """重新上传一次 metadata 文件，来标识其被访问过""" - self.cloud_client.upload( - self.local_backend.get_metadata_path(target), - self.get_metadata_path(target), - ) + try: + self.cloud_client.upload( + self.local_backend.get_metadata_path(target), + self.get_metadata_path(target), + ) + except CloudStorageTransientError as e: + logger.warning(f"更新缓存访问时间失败，将跳过：{e}") def load(self, target: Target, update_access_time: bool = True) -> None: if self.local_backend.exists(target): @@ -209,15 +212,18 @@ def load(self, target: Target, update_access_time: bool = True) -> None: logger.warning("本地缓存已损坏，从云存储重新下载") shutil.rmtree(self.local_backend.get_cache_path(target), ignore_errors=True) - logger.info(f"正在从云存储加载 Target {target.name} 的缓存...") - self.cloud_client.download(self.get_archive_path(target), self.local_backend.get_archive_path(target)) - self.cloud_client.download(self.get_metadata_path(target), self.local_backend.get_metadata_path(target)) try: + logger.info(f"正在从云存储加载 Target {target.name} 的缓存...") + self.cloud_client.download(self.get_archive_path(target), self.local_backend.get_archive_path(target)) + self.cloud_client.download(self.get_metadata_path(target), self.local_backend.get_metadata_path(target)) self.local_backend.load(target) except zstd.ZstdError: logger.warning(f"云存储中 Target {target.name} 的缓存已损坏，将重新生成") shutil.rmtree(self.local_backend.get_cache_path(target), ignore_errors=True) raise + except CloudStorageTransientError: + shutil.rmtree(self.local_backend.get_cache_path(target), ignore_errors=True) + raise if update_access_time: self.update_access_time(target) diff --git a/src/quack/cache_test.py b/src/quack/cache_test.py index f408fed..8ae0465 100644 --- a/src/quack/cache_test.py +++ b/src/quack/cache_test.py @@ -6,6 +6,7 @@ from quack.cache import TargetCacheBackendTypeCloud from quack.config import Config +from quack.exceptions import CloudStorageError, CloudStorageTransientError class TestTargetCacheBackendTypeCloud: @@ -32,6 +33,55 @@ def test_load_exists( # 验证 update_access_time 被调用（上传 metadata） assert mock_cloud_client.upload.called + @mock.patch("quack.cache.CloudClient") + @mock.patch("quack.cache.TargetCacheBackendTypeLocal") + def test_load_exists_ignores_transient_access_time_update_failure( + self, + mock_local_backend: mock.Mock, + mock_cloud_client_class: mock.Mock, + mock_test_spec: mock.Mock, + ): + mock_cloud_client = mock.Mock() + mock_cloud_client.upload.side_effect = CloudStorageTransientError( + "上传文件失败", + "IncompleteBody", + code="IncompleteBody", + ) + mock_cloud_client_class.return_value = mock_cloud_client + + config = Config.model_construct() + target = mock_test_spec.targets["quack:test"] + target._checksum_value = "" + backend = TargetCacheBackendTypeCloud(config, mock_test_spec.app_name) + + mock_local_backend.return_value.exists.return_value = True + + backend.load(target) + + mock_local_backend.return_value.load.assert_called_once() + + @mock.patch("quack.cache.CloudClient") + @mock.patch("quack.cache.TargetCacheBackendTypeLocal") + def test_load_exists_raises_non_transient_access_time_update_failure( + self, + mock_local_backend: mock.Mock, + mock_cloud_client_class: mock.Mock, + mock_test_spec: mock.Mock, + ): + mock_cloud_client = mock.Mock() + mock_cloud_client.upload.side_effect = CloudStorageError("上传文件失败", "AccessDenied", code="AccessDenied") + mock_cloud_client_class.return_value = mock_cloud_client + + config = Config.model_construct() + target = mock_test_spec.targets["quack:test"] + target._checksum_value = "" + backend = TargetCacheBackendTypeCloud(config, mock_test_spec.app_name) + + mock_local_backend.return_value.exists.return_value = True + + with pytest.raises(CloudStorageError): + backend.load(target) + @mock.patch("quack.cache.CloudClient") @mock.patch("quack.cache.TargetCacheBackendTypeLocal") def test_load_not_exists( diff --git a/src/quack/exceptions.py b/src/quack/exceptions.py index 5c5d916..17423b8 100644 --- a/src/quack/exceptions.py +++ b/src/quack/exceptions.py @@ -7,7 +7,12 @@ class ConfigError(Exception): class CloudStorageError(Exception): - def __init__(self, message: str, details: str = ""): + def __init__(self, message: str, details: str = "", code: str | None = None): self.message = message self.details = details + self.code = code super().__init__(f"{message}: {details}" if details else message) + + +class CloudStorageTransientError(CloudStorageError): + pass diff --git a/src/quack/models/target.py b/src/quack/models/target.py index 4c30b73..681837e 100644 --- a/src/quack/models/target.py +++ b/src/quack/models/target.py @@ -14,7 +14,7 @@ from pydantic import Field from quack.config import Config -from quack.exceptions import CloudStorageError +from quack.exceptions import CloudStorageTransientError from quack.models.base import BaseModel from quack.models.command import Command from quack.models.dependency import Dependency, DependencyTypeTarget @@ -116,8 +116,8 @@ def execute( logger.info("找到缓存，直接从缓存加载...") try: cache.load() - except zstd.ZstdError: - logger.warning(f"缓存已损坏，将重新生成 Target {self.name}...") + except (CloudStorageTransientError, zstd.ZstdError) as e: + logger.warning(f"缓存加载失败，将重新生成 Target {self.name}：{e}") self.prepare_deps(config, app_name, cache_backend) self.operations.build.execute() cache_exists = False @@ -126,7 +126,7 @@ def execute( logger.info(f"正在存入缓存，路径：{self.cache_path}") try: cache.save() - except CloudStorageError as e: + except CloudStorageTransientError as e: logger.warning(f"上传缓存失败，将跳过云端缓存：{e}") elapsed = time.time() - start_time diff --git a/src/quack/models/target_test.py b/src/quack/models/target_test.py index affb76e..d3fd6e8 100644 --- a/src/quack/models/target_test.py +++ b/src/quack/models/target_test.py @@ -4,6 +4,7 @@ from pydantic import ValidationError from quack.config import Config +from quack.exceptions import CloudStorageError, CloudStorageTransientError from quack.models.target import Target, TargetExecutionMode @@ -65,6 +66,39 @@ def test_execute_cache_hit(self, mock_target_cache, mock_test_spec: mock.Mock): target.execute(config, mock_test_spec.app_name, mock.Mock) mock_target_cache.return_value.load.assert_called_once() + @mock.patch("quack.cache.TargetCache") + def test_execute_ignores_transient_cache_save_failure(self, mock_target_cache, mock_test_spec: mock.Mock): + config = Config.model_construct() + target = mock_test_spec.targets["quack:test"] + target._checksum_value = "" + + mock_target_cache.return_value.hit.return_value = False + mock_target_cache.return_value.save.side_effect = CloudStorageTransientError( + "上传文件失败", + "IncompleteBody", + code="IncompleteBody", + ) + with mock.patch("quack.models.command.Command.execute") as mock_build: + target.execute(config, mock_test_spec.app_name, mock.Mock) + + mock_build.assert_called_once() + mock_target_cache.return_value.save.assert_called_once() + + @mock.patch("quack.cache.TargetCache") + def test_execute_raises_non_transient_cache_save_failure(self, mock_target_cache, mock_test_spec: mock.Mock): + config = Config.model_construct() + target = mock_test_spec.targets["quack:test"] + target._checksum_value = "" + + mock_target_cache.return_value.hit.return_value = False + mock_target_cache.return_value.save.side_effect = CloudStorageError( + "上传文件失败", + "AccessDenied", + code="AccessDenied", + ) + with mock.patch("quack.models.command.Command.execute"), pytest.raises(CloudStorageError): + target.execute(config, mock_test_spec.app_name, mock.Mock) + @mock.patch("quack.cache.TargetCache") def test_execute_load_only(self, mock_target_cache, mock_test_spec: mock.Mock): config = Config.model_construct() diff --git a/src/quack/utils/cloud.py b/src/quack/utils/cloud.py index 53fa551..62013f0 100644 --- a/src/quack/utils/cloud.py +++ b/src/quack/utils/cloud.py @@ -2,15 +2,49 @@ import fnmatch import os +import re from collections.abc import Iterator from dataclasses import dataclass from datetime import datetime import boto3 +from boto3.exceptions import S3TransferFailedError, S3UploadFailedError from botocore.config import Config as BotocoreConfig -from botocore.exceptions import ClientError, NoCredentialsError - -from quack.exceptions import CloudStorageError +from botocore.exceptions import ( + ClientError, + ConnectionClosedError, + ConnectTimeoutError, + EndpointConnectionError, + HTTPClientError, + NoCredentialsError, + ProxyConnectionError, + ReadTimeoutError, +) + +from quack.exceptions import CloudStorageError, CloudStorageTransientError + +TRANSIENT_ERROR_CODES = { + "500", + "503", + "IncompleteBody", + "InternalError", + "RequestTimeout", + "RequestTimeoutException", + "ServiceUnavailable", + "SlowDown", + "Throttling", + "ThrottlingException", + "TooManyRequestsException", +} + +TRANSIENT_NETWORK_ERRORS = ( + ConnectionClosedError, + ConnectTimeoutError, + EndpointConnectionError, + HTTPClientError, + ProxyConnectionError, + ReadTimeoutError, +) @dataclass @@ -20,6 +54,38 @@ class CloudFileMetadata: size: int +def _get_client_error_code(error: ClientError) -> str: + code = error.response.get("Error", {}).get("Code", "") + return str(code) + + +def _extract_error_code(error: Exception) -> str: + if isinstance(error, ClientError): + return _get_client_error_code(error) + + if isinstance(error.__context__, ClientError): + return _get_client_error_code(error.__context__) + + match = re.search(r"An error occurred \(([^)]+)\)", str(error)) + if match: + return match.group(1) + + match = re.search(r"\(([^)]+)\)", str(error)) + return match.group(1) if match else "" + + +def _is_transient_error(error: Exception) -> bool: + if isinstance(error, TRANSIENT_NETWORK_ERRORS) or isinstance(error.__context__, TRANSIENT_NETWORK_ERRORS): + return True + return _extract_error_code(error) in TRANSIENT_ERROR_CODES + + +def _raise_cloud_storage_error(message: str, path: str, error: Exception) -> None: + code = _extract_error_code(error) + exception_class = CloudStorageTransientError if _is_transient_error(error) else CloudStorageError + raise exception_class(f"{message}：{path}", str(error), code=code or None) from error + + class CloudClient: """统一的云存储客户端（支持 OSS 和 S3，使用 boto3）""" @@ -137,8 +203,8 @@ def upload(self, path: str, dest: str) -> None: self._client.upload_file(local_file, self._bucket_name, object_key) else: raise CloudStorageError(f"路径不存在或不是文件/目录：{path}") - except ClientError as e: - raise CloudStorageError(f"上传文件失败：{path}", str(e)) from e + except (ClientError, S3TransferFailedError, S3UploadFailedError, *TRANSIENT_NETWORK_ERRORS) as e: + _raise_cloud_storage_error("上传文件失败", path, e) def download(self, path: str, dest: str) -> None: """下载文件或目录""" @@ -175,8 +241,8 @@ def download(self, path: str, dest: str) -> None: if dir_path: os.makedirs(dir_path, exist_ok=True) self._client.download_file(self._bucket_name, key, dest) - except ClientError as e: - raise CloudStorageError(f"下载文件失败：{path}", str(e)) from e + except (ClientError, S3TransferFailedError, *TRANSIENT_NETWORK_ERRORS) as e: + _raise_cloud_storage_error("下载文件失败", path, e) def read(self, path: str) -> str | None: """读取文件内容""" diff --git a/src/quack/utils/cloud_test.py b/src/quack/utils/cloud_test.py new file mode 100644 index 0000000..200dfef --- /dev/null +++ b/src/quack/utils/cloud_test.py @@ -0,0 +1,47 @@ +from pathlib import Path +from unittest import mock + +import pytest +from boto3.exceptions import S3UploadFailedError +from botocore.exceptions import ClientError + +from quack.exceptions import CloudStorageError, CloudStorageTransientError +from quack.utils.cloud import CloudClient + + +def test_upload_wraps_incomplete_body_as_transient_error(tmp_path: Path): + local_file = tmp_path / "cache.tar.zst" + local_file.write_bytes(b"cache") + + client = CloudClient.__new__(CloudClient) + client._base_path = "" + client._bucket_name = "bucket" + client._client = mock.Mock() + client._client.upload_file.side_effect = S3UploadFailedError( + "Failed to upload cache.tar.zst: An error occurred (IncompleteBody) when calling the PutObject operation" + ) + + with pytest.raises(CloudStorageTransientError) as exc_info: + client.upload(str(local_file), "dest/cache.tar.zst") + + assert exc_info.value.code == "IncompleteBody" + + +def test_upload_keeps_access_denied_as_non_transient_error(tmp_path: Path): + local_file = tmp_path / "cache.tar.zst" + local_file.write_bytes(b"cache") + + client = CloudClient.__new__(CloudClient) + client._base_path = "" + client._bucket_name = "bucket" + client._client = mock.Mock() + client._client.upload_file.side_effect = ClientError( + {"Error": {"Code": "AccessDenied", "Message": "Access denied"}}, + "PutObject", + ) + + with pytest.raises(CloudStorageError) as exc_info: + client.upload(str(local_file), "dest/cache.tar.zst") + + assert not isinstance(exc_info.value, CloudStorageTransientError) + assert exc_info.value.code == "AccessDenied" From 65f40aa1e7cd92be598d02251f1e4c93aef22b9a Mon Sep 17 00:00:00 2001 From: tigerBeA Date: Thu, 9 Jul 2026 15:35:08 +0800 Subject: [PATCH 3/5] test: cover transient cache failure paths --- src/quack/models/target_test.py | 37 +++++++++++++++++++++++++++++++++ src/quack/utils/cloud.py | 10 ++++++--- src/quack/utils/cloud_test.py | 35 ++++++++++++++++++------------- 3 files changed, 65 insertions(+), 17 deletions(-) diff --git a/src/quack/models/target_test.py b/src/quack/models/target_test.py index d3fd6e8..30f1fe9 100644 --- a/src/quack/models/target_test.py +++ b/src/quack/models/target_test.py @@ -66,6 +66,43 @@ def test_execute_cache_hit(self, mock_target_cache, mock_test_spec: mock.Mock): target.execute(config, mock_test_spec.app_name, mock.Mock) mock_target_cache.return_value.load.assert_called_once() + @mock.patch("quack.cache.TargetCache") + def test_execute_rebuilds_after_transient_cache_load_failure(self, mock_target_cache, mock_test_spec: mock.Mock): + config = Config.model_construct() + target = mock_test_spec.targets["quack:test"] + target._checksum_value = "" + + mock_target_cache.return_value.hit.return_value = True + mock_target_cache.return_value.load.side_effect = CloudStorageTransientError( + "下载文件失败", + "IncompleteBody", + code="IncompleteBody", + ) + with mock.patch("quack.models.command.Command.execute") as mock_build: + target.execute(config, mock_test_spec.app_name, mock.Mock) + + mock_target_cache.return_value.load.assert_called_once() + mock_build.assert_called_once() + mock_target_cache.return_value.save.assert_called_once() + + @mock.patch("quack.cache.TargetCache") + def test_execute_raises_non_transient_cache_load_failure(self, mock_target_cache, mock_test_spec: mock.Mock): + config = Config.model_construct() + target = mock_test_spec.targets["quack:test"] + target._checksum_value = "" + + mock_target_cache.return_value.hit.return_value = True + mock_target_cache.return_value.load.side_effect = CloudStorageError( + "下载文件失败", + "AccessDenied", + code="AccessDenied", + ) + with mock.patch("quack.models.command.Command.execute") as mock_build, pytest.raises(CloudStorageError): + target.execute(config, mock_test_spec.app_name, mock.Mock) + + mock_build.assert_not_called() + mock_target_cache.return_value.save.assert_not_called() + @mock.patch("quack.cache.TargetCache") def test_execute_ignores_transient_cache_save_failure(self, mock_target_cache, mock_test_spec: mock.Mock): config = Config.model_construct() diff --git a/src/quack/utils/cloud.py b/src/quack/utils/cloud.py index 62013f0..cf549f5 100644 --- a/src/quack/utils/cloud.py +++ b/src/quack/utils/cloud.py @@ -63,8 +63,9 @@ def _extract_error_code(error: Exception) -> str: if isinstance(error, ClientError): return _get_client_error_code(error) - if isinstance(error.__context__, ClientError): - return _get_client_error_code(error.__context__) + for wrapped_error in [error.__context__, error.__cause__]: + if isinstance(wrapped_error, ClientError): + return _get_client_error_code(wrapped_error) match = re.search(r"An error occurred \(([^)]+)\)", str(error)) if match: @@ -75,7 +76,10 @@ def _extract_error_code(error: Exception) -> str: def _is_transient_error(error: Exception) -> bool: - if isinstance(error, TRANSIENT_NETWORK_ERRORS) or isinstance(error.__context__, TRANSIENT_NETWORK_ERRORS): + wrapped_errors = [error.__context__, error.__cause__] + if isinstance(error, TRANSIENT_NETWORK_ERRORS) or any( + isinstance(wrapped_error, TRANSIENT_NETWORK_ERRORS) for wrapped_error in wrapped_errors + ): return True return _extract_error_code(error) in TRANSIENT_ERROR_CODES diff --git a/src/quack/utils/cloud_test.py b/src/quack/utils/cloud_test.py index 200dfef..efebbbb 100644 --- a/src/quack/utils/cloud_test.py +++ b/src/quack/utils/cloud_test.py @@ -9,17 +9,31 @@ from quack.utils.cloud import CloudClient -def test_upload_wraps_incomplete_body_as_transient_error(tmp_path: Path): - local_file = tmp_path / "cache.tar.zst" - local_file.write_bytes(b"cache") - +def _cloud_client_with_upload_error(error: Exception) -> CloudClient: client = CloudClient.__new__(CloudClient) client._base_path = "" client._bucket_name = "bucket" client._client = mock.Mock() - client._client.upload_file.side_effect = S3UploadFailedError( - "Failed to upload cache.tar.zst: An error occurred (IncompleteBody) when calling the PutObject operation" + client._client.upload_file.side_effect = error + return client + + +def _s3_upload_error_from_client_error(code: str) -> S3UploadFailedError: + client_error = ClientError( + {"Error": {"Code": code, "Message": code}}, + "PutObject", ) + try: + raise S3UploadFailedError("Failed to upload cache.tar.zst") from client_error + except S3UploadFailedError as e: + return e + + +def test_upload_wraps_incomplete_body_as_transient_error(tmp_path: Path): + local_file = tmp_path / "cache.tar.zst" + local_file.write_bytes(b"cache") + + client = _cloud_client_with_upload_error(_s3_upload_error_from_client_error("IncompleteBody")) with pytest.raises(CloudStorageTransientError) as exc_info: client.upload(str(local_file), "dest/cache.tar.zst") @@ -31,14 +45,7 @@ def test_upload_keeps_access_denied_as_non_transient_error(tmp_path: Path): local_file = tmp_path / "cache.tar.zst" local_file.write_bytes(b"cache") - client = CloudClient.__new__(CloudClient) - client._base_path = "" - client._bucket_name = "bucket" - client._client = mock.Mock() - client._client.upload_file.side_effect = ClientError( - {"Error": {"Code": "AccessDenied", "Message": "Access denied"}}, - "PutObject", - ) + client = _cloud_client_with_upload_error(_s3_upload_error_from_client_error("AccessDenied")) with pytest.raises(CloudStorageError) as exc_info: client.upload(str(local_file), "dest/cache.tar.zst") From 4cade923aecd7a5382776063a0da2facdaf94d29 Mon Sep 17 00:00:00 2001 From: tigerBeA Date: Thu, 9 Jul 2026 16:08:14 +0800 Subject: [PATCH 4/5] fix: cover transient cache lookup failures --- src/quack/models/target.py | 8 ++++- src/quack/models/target_test.py | 17 +++++++++ src/quack/utils/cloud.py | 61 +++++++++++++++++++++++---------- src/quack/utils/cloud_test.py | 38 ++++++++++++++++++-- 4 files changed, 102 insertions(+), 22 deletions(-) diff --git a/src/quack/models/target.py b/src/quack/models/target.py index 681837e..b7bf442 100644 --- a/src/quack/models/target.py +++ b/src/quack/models/target.py @@ -95,7 +95,13 @@ def execute( logger.info(f"正在查找 Target {self.name} 的缓存...") cache = TargetCache(config, app_name, self, cache_backend) - cache_exists = cache.hit() + try: + cache_exists = cache.hit() + except CloudStorageTransientError as e: + if mode == TargetExecutionMode.LOAD_ONLY: + raise + logger.warning(f"缓存命中检查失败，将重新生成 Target {self.name}：{e}") + cache_exists = False if mode == TargetExecutionMode.DEPS_ONLY: self.prepare_deps(config, app_name, cache_backend) diff --git a/src/quack/models/target_test.py b/src/quack/models/target_test.py index 30f1fe9..14bdadc 100644 --- a/src/quack/models/target_test.py +++ b/src/quack/models/target_test.py @@ -66,6 +66,23 @@ def test_execute_cache_hit(self, mock_target_cache, mock_test_spec: mock.Mock): target.execute(config, mock_test_spec.app_name, mock.Mock) mock_target_cache.return_value.load.assert_called_once() + @mock.patch("quack.cache.TargetCache") + def test_execute_rebuilds_after_transient_cache_hit_failure(self, mock_target_cache, mock_test_spec: mock.Mock): + config = Config.model_construct() + target = mock_test_spec.targets["quack:test"] + target._checksum_value = "" + + mock_target_cache.return_value.hit.side_effect = CloudStorageTransientError( + "检查文件是否存在失败", + "Could not connect to the endpoint URL", + ) + with mock.patch("quack.models.command.Command.execute") as mock_build: + target.execute(config, mock_test_spec.app_name, mock.Mock) + + mock_build.assert_called_once() + mock_target_cache.return_value.load.assert_not_called() + mock_target_cache.return_value.save.assert_called_once() + @mock.patch("quack.cache.TargetCache") def test_execute_rebuilds_after_transient_cache_load_failure(self, mock_target_cache, mock_test_spec: mock.Mock): config = Config.model_construct() diff --git a/src/quack/utils/cloud.py b/src/quack/utils/cloud.py index cf549f5..3a5a099 100644 --- a/src/quack/utils/cloud.py +++ b/src/quack/utils/cloud.py @@ -6,11 +6,13 @@ from collections.abc import Iterator from dataclasses import dataclass from datetime import datetime +from typing import NoReturn import boto3 -from boto3.exceptions import S3TransferFailedError, S3UploadFailedError +from boto3.exceptions import Boto3Error from botocore.config import Config as BotocoreConfig from botocore.exceptions import ( + BotoCoreError, ClientError, ConnectionClosedError, ConnectTimeoutError, @@ -59,32 +61,51 @@ def _get_client_error_code(error: ClientError) -> str: return str(code) +def _iter_error_chain(error: Exception) -> Iterator[Exception]: + seen: set[int] = set() + pending = [error] + while pending: + current_error = pending.pop() + if id(current_error) in seen: + continue + + seen.add(id(current_error)) + yield current_error + + pending.extend( + wrapped_error + for wrapped_error in [ + current_error.__context__, + current_error.__cause__, + getattr(current_error, "last_exception", None), + ] + if isinstance(wrapped_error, Exception) + ) + + def _extract_error_code(error: Exception) -> str: - if isinstance(error, ClientError): - return _get_client_error_code(error) + for current_error in _iter_error_chain(error): + if isinstance(current_error, ClientError): + return _get_client_error_code(current_error) - for wrapped_error in [error.__context__, error.__cause__]: - if isinstance(wrapped_error, ClientError): - return _get_client_error_code(wrapped_error) + match = re.search(r"An error occurred \(([^)]+)\)", str(current_error)) + if match: + return match.group(1) - match = re.search(r"An error occurred \(([^)]+)\)", str(error)) - if match: - return match.group(1) + match = re.search(r"\(([^)]+)\)", str(current_error)) + if match: + return match.group(1) - match = re.search(r"\(([^)]+)\)", str(error)) - return match.group(1) if match else "" + return "" def _is_transient_error(error: Exception) -> bool: - wrapped_errors = [error.__context__, error.__cause__] - if isinstance(error, TRANSIENT_NETWORK_ERRORS) or any( - isinstance(wrapped_error, TRANSIENT_NETWORK_ERRORS) for wrapped_error in wrapped_errors - ): + if any(isinstance(current_error, TRANSIENT_NETWORK_ERRORS) for current_error in _iter_error_chain(error)): return True return _extract_error_code(error) in TRANSIENT_ERROR_CODES -def _raise_cloud_storage_error(message: str, path: str, error: Exception) -> None: +def _raise_cloud_storage_error(message: str, path: str, error: Exception) -> NoReturn: code = _extract_error_code(error) exception_class = CloudStorageTransientError if _is_transient_error(error) else CloudStorageError raise exception_class(f"{message}：{path}", str(error), code=code or None) from error @@ -187,7 +208,9 @@ def exists(self, path: str) -> bool: except ClientError as e: if e.response["Error"]["Code"] == "404": return False - raise CloudStorageError(f"检查文件是否存在失败：{path}", str(e)) from e + _raise_cloud_storage_error("检查文件是否存在失败", path, e) + except (Boto3Error, BotoCoreError) as e: + _raise_cloud_storage_error("检查文件是否存在失败", path, e) def upload(self, path: str, dest: str) -> None: """上传文件或目录""" @@ -207,7 +230,7 @@ def upload(self, path: str, dest: str) -> None: self._client.upload_file(local_file, self._bucket_name, object_key) else: raise CloudStorageError(f"路径不存在或不是文件/目录：{path}") - except (ClientError, S3TransferFailedError, S3UploadFailedError, *TRANSIENT_NETWORK_ERRORS) as e: + except (ClientError, Boto3Error, BotoCoreError) as e: _raise_cloud_storage_error("上传文件失败", path, e) def download(self, path: str, dest: str) -> None: @@ -245,7 +268,7 @@ def download(self, path: str, dest: str) -> None: if dir_path: os.makedirs(dir_path, exist_ok=True) self._client.download_file(self._bucket_name, key, dest) - except (ClientError, S3TransferFailedError, *TRANSIENT_NETWORK_ERRORS) as e: + except (ClientError, Boto3Error, BotoCoreError) as e: _raise_cloud_storage_error("下载文件失败", path, e) def read(self, path: str) -> str | None: diff --git a/src/quack/utils/cloud_test.py b/src/quack/utils/cloud_test.py index efebbbb..9660e04 100644 --- a/src/quack/utils/cloud_test.py +++ b/src/quack/utils/cloud_test.py @@ -2,8 +2,8 @@ from unittest import mock import pytest -from boto3.exceptions import S3UploadFailedError -from botocore.exceptions import ClientError +from boto3.exceptions import RetriesExceededError, S3UploadFailedError +from botocore.exceptions import ClientError, EndpointConnectionError, ReadTimeoutError from quack.exceptions import CloudStorageError, CloudStorageTransientError from quack.utils.cloud import CloudClient @@ -18,6 +18,16 @@ def _cloud_client_with_upload_error(error: Exception) -> CloudClient: return client +def _cloud_client_with_download_error(error: Exception) -> CloudClient: + client = CloudClient.__new__(CloudClient) + client._base_path = "" + client._bucket_name = "bucket" + client._client = mock.Mock() + client._client.get_paginator.return_value.paginate.return_value = [{"Contents": [{"Key": "cache.tar.zst"}]}] + client._client.download_file.side_effect = error + return client + + def _s3_upload_error_from_client_error(code: str) -> S3UploadFailedError: client_error = ClientError( {"Error": {"Code": code, "Message": code}}, @@ -52,3 +62,27 @@ def test_upload_keeps_access_denied_as_non_transient_error(tmp_path: Path): assert not isinstance(exc_info.value, CloudStorageTransientError) assert exc_info.value.code == "AccessDenied" + + +def test_download_wraps_retries_exceeded_timeout_as_transient_error(tmp_path: Path): + local_file = tmp_path / "cache.tar.zst" + retry_error = RetriesExceededError(ReadTimeoutError(endpoint_url="https://s3.example")) + client = _cloud_client_with_download_error(retry_error) + + with pytest.raises(CloudStorageTransientError) as exc_info: + client.download("cache.tar.zst", str(local_file)) + + assert exc_info.value.code is None + + +def test_exists_wraps_endpoint_connection_error_as_transient_error(): + client = CloudClient.__new__(CloudClient) + client._base_path = "" + client._bucket_name = "bucket" + client._client = mock.Mock() + client._client.head_object.side_effect = EndpointConnectionError(endpoint_url="https://s3.example") + + with pytest.raises(CloudStorageTransientError) as exc_info: + client.exists("cache-metadata.json") + + assert exc_info.value.code is None From eb4376b2d773fecd44074d09cc4c00e0f358f7d1 Mon Sep 17 00:00:00 2001 From: tigerBeA Date: Thu, 9 Jul 2026 16:37:56 +0800 Subject: [PATCH 5/5] refactor(cache): tighten cache recovery boundaries --- src/quack/cache.py | 11 +- src/quack/cache_test.py | 29 +++-- src/quack/exceptions.py | 6 +- src/quack/models/target.py | 74 ++++++------ src/quack/models/target_test.py | 198 ++++++++++++++++++++++---------- src/quack/utils/cloud.py | 8 +- src/quack/utils/cloud_test.py | 18 +++ 7 files changed, 228 insertions(+), 116 deletions(-) diff --git a/src/quack/cache.py b/src/quack/cache.py index 43ab379..4ed608d 100644 --- a/src/quack/cache.py +++ b/src/quack/cache.py @@ -15,7 +15,7 @@ from quack.config import Config from quack.consts import CACHE_METADATA_FILENAME -from quack.exceptions import ChecksumError, CloudStorageTransientError +from quack.exceptions import CacheCorruptionError, CloudStorageTransientError from quack.models.target import Target from quack.utils.archiver import Archiver from quack.utils.ci_environment import CIEnvironment @@ -69,7 +69,10 @@ def load(self, target: Target) -> None: archive_path = self.get_archive_path(target) size = os.path.getsize(archive_path) logger.info(f"正在从本地加载 Target {target.name} 的缓存（大小：{format_size(size)}）...") - Archiver.extract(archive_path) + try: + Archiver.extract(archive_path) + except zstd.ZstdError as e: + raise CacheCorruptionError(f"缓存归档解压失败：{archive_path}") from e metadata_path = self.get_metadata_path(target) if os.path.exists(metadata_path): os.utime(metadata_path, None) @@ -208,7 +211,7 @@ def load(self, target: Target, update_access_time: bool = True) -> None: if update_access_time: self.update_access_time(target) return - except (ChecksumError, zstd.ZstdError): + except CacheCorruptionError: logger.warning("本地缓存已损坏，从云存储重新下载") shutil.rmtree(self.local_backend.get_cache_path(target), ignore_errors=True) @@ -217,7 +220,7 @@ def load(self, target: Target, update_access_time: bool = True) -> None: self.cloud_client.download(self.get_archive_path(target), self.local_backend.get_archive_path(target)) self.cloud_client.download(self.get_metadata_path(target), self.local_backend.get_metadata_path(target)) self.local_backend.load(target) - except zstd.ZstdError: + except CacheCorruptionError: logger.warning(f"云存储中 Target {target.name} 的缓存已损坏，将重新生成") shutil.rmtree(self.local_backend.get_cache_path(target), ignore_errors=True) raise diff --git a/src/quack/cache_test.py b/src/quack/cache_test.py index 8ae0465..85a47e6 100644 --- a/src/quack/cache_test.py +++ b/src/quack/cache_test.py @@ -2,11 +2,27 @@ from unittest import mock import pytest -import zstandard as zstd -from quack.cache import TargetCacheBackendTypeCloud +from quack.cache import TargetCacheBackendTypeCloud, TargetCacheBackendTypeLocal from quack.config import Config -from quack.exceptions import CloudStorageError, CloudStorageTransientError +from quack.exceptions import CacheCorruptionError, CloudStorageError, CloudStorageTransientError + + +class TestTargetCacheBackendTypeLocal: + def test_load_wraps_corrupt_archive_as_cache_corruption(self, tmp_path, monkeypatch, mock_test_spec: mock.Mock): + monkeypatch.setattr("quack.cache.xdg_cache_home", lambda: tmp_path) + + config = Config.model_construct() + target = mock_test_spec.targets["quack:test"] + target._checksum_value = "" + backend = TargetCacheBackendTypeLocal(config, mock_test_spec.app_name) + + os.makedirs(backend.get_cache_path(target), exist_ok=True) + with open(backend.get_archive_path(target), "wb") as f: + _ = f.write(b"not a zstd archive") + + with pytest.raises(CacheCorruptionError): + backend.load(target) class TestTargetCacheBackendTypeCloud: @@ -125,8 +141,7 @@ def test_load_corrupt_local_cache_falls_back_to_cloud( backend = TargetCacheBackendTypeCloud(config, mock_test_spec.app_name) mock_local_backend.return_value.exists.return_value = True - # First call (local) raises; second call (after re-download) succeeds - mock_local_backend.return_value.load.side_effect = [zstd.ZstdError("did not decompress full frame"), None] + mock_local_backend.return_value.load.side_effect = [CacheCorruptionError("缓存归档解压失败"), None] backend.load(target) @@ -152,9 +167,9 @@ def test_load_corrupt_cloud_cache_raises( backend = TargetCacheBackendTypeCloud(config, mock_test_spec.app_name) mock_local_backend.return_value.exists.return_value = False - mock_local_backend.return_value.load.side_effect = zstd.ZstdError("did not decompress full frame") + mock_local_backend.return_value.load.side_effect = CacheCorruptionError("缓存归档解压失败") - with pytest.raises(zstd.ZstdError): + with pytest.raises(CacheCorruptionError): backend.load(target) assert mock_rmtree.called diff --git a/src/quack/exceptions.py b/src/quack/exceptions.py index 17423b8..f8e12bd 100644 --- a/src/quack/exceptions.py +++ b/src/quack/exceptions.py @@ -1,4 +1,8 @@ -class ChecksumError(Exception): +class CacheCorruptionError(Exception): + pass + + +class ChecksumError(CacheCorruptionError): pass diff --git a/src/quack/models/target.py b/src/quack/models/target.py index b7bf442..1898df9 100644 --- a/src/quack/models/target.py +++ b/src/quack/models/target.py @@ -9,12 +9,11 @@ from pathlib import Path from typing import TYPE_CHECKING -import zstandard as zstd from loguru import logger from pydantic import Field from quack.config import Config -from quack.exceptions import CloudStorageTransientError +from quack.exceptions import CacheCorruptionError, CloudStorageTransientError from quack.models.base import BaseModel from quack.models.command import Command from quack.models.dependency import Dependency, DependencyTypeTarget @@ -92,48 +91,49 @@ def execute( logger.info(f"正在执行 Target {self.name}...") logger.info(f"Target {self.name} Checksum 值：{self.checksum_value}") - logger.info(f"正在查找 Target {self.name} 的缓存...") - - cache = TargetCache(config, app_name, self, cache_backend) - try: - cache_exists = cache.hit() - except CloudStorageTransientError as e: - if mode == TargetExecutionMode.LOAD_ONLY: - raise - logger.warning(f"缓存命中检查失败，将重新生成 Target {self.name}：{e}") - cache_exists = False if mode == TargetExecutionMode.DEPS_ONLY: self.prepare_deps(config, app_name, cache_backend) - elif mode == TargetExecutionMode.LOAD_ONLY: - if cache_exists: - logger.info("找到缓存，直接从缓存加载...") - cache.load() - else: - logger.error("未找到缓存，无法进行加载") - sys.exit(1) else: - if not cache_exists: - logger.info(f"未找到对应的缓存，开始重新生成缓存：{self.operations.build.command}") - self.prepare_deps(config, app_name, cache_backend) - self.operations.build.execute() - - if cache_exists: - logger.info("找到缓存，直接从缓存加载...") - try: + logger.info(f"正在查找 Target {self.name} 的缓存...") + cache = TargetCache(config, app_name, self, cache_backend) + try: + cache_exists = cache.hit() + except CloudStorageTransientError as e: + if mode == TargetExecutionMode.LOAD_ONLY: + raise + logger.warning(f"缓存命中检查失败，将重新生成 Target {self.name}：{e}") + cache_exists = False + + if mode == TargetExecutionMode.LOAD_ONLY: + if cache_exists: + logger.info("找到缓存，直接从缓存加载...") cache.load() - except (CloudStorageTransientError, zstd.ZstdError) as e: - logger.warning(f"缓存加载失败，将重新生成 Target {self.name}：{e}") + else: + logger.error("未找到缓存，无法进行加载") + sys.exit(1) + else: + if not cache_exists: + logger.info(f"未找到对应的缓存，开始重新生成缓存：{self.operations.build.command}") self.prepare_deps(config, app_name, cache_backend) self.operations.build.execute() - cache_exists = False - - if not cache_exists: - logger.info(f"正在存入缓存，路径：{self.cache_path}") - try: - cache.save() - except CloudStorageTransientError as e: - logger.warning(f"上传缓存失败，将跳过云端缓存：{e}") + + if cache_exists: + logger.info("找到缓存，直接从缓存加载...") + try: + cache.load() + except (CloudStorageTransientError, CacheCorruptionError) as e: + logger.warning(f"缓存加载失败，将重新生成 Target {self.name}：{e}") + self.prepare_deps(config, app_name, cache_backend) + self.operations.build.execute() + cache_exists = False + + if not cache_exists: + logger.info(f"正在存入缓存，路径：{self.cache_path}") + try: + cache.save() + except CloudStorageTransientError as e: + logger.warning(f"上传缓存失败，将跳过云端缓存：{e}") elapsed = time.time() - start_time logger.success(f"Target {self.name} 执行完毕！") diff --git a/src/quack/models/target_test.py b/src/quack/models/target_test.py index 14bdadc..258f916 100644 --- a/src/quack/models/target_test.py +++ b/src/quack/models/target_test.py @@ -1,13 +1,65 @@ +from typing import ClassVar, cast from unittest import mock import pytest from pydantic import ValidationError +from quack.cache import TargetCacheBackendType from quack.config import Config -from quack.exceptions import CloudStorageError, CloudStorageTransientError +from quack.exceptions import CacheCorruptionError, CloudStorageError, CloudStorageTransientError from quack.models.target import Target, TargetExecutionMode +class RecordingCacheBackend: + NAME: ClassVar[str] = "recording" + exists_result: ClassVar[bool] = False + exists_error: ClassVar[Exception | None] = None + load_error: ClassVar[Exception | None] = None + save_error: ClassVar[Exception | None] = None + exists_calls: ClassVar[int] = 0 + load_calls: ClassVar[int] = 0 + save_calls: ClassVar[int] = 0 + + def __init__(self, _config: Config, _app_name: str) -> None: + pass + + def exists(self, _target: Target) -> bool: + type(self).exists_calls += 1 + error = type(self).exists_error + if error is not None: + raise error + return type(self).exists_result + + def load(self, _target: Target) -> None: + type(self).load_calls += 1 + error = type(self).load_error + if error is not None: + raise error + + def save(self, _target: Target) -> None: + type(self).save_calls += 1 + error = type(self).save_error + if error is not None: + raise error + + +def reset_recording_cache_backend( + *, + exists_result: bool = False, + exists_error: Exception | None = None, + load_error: Exception | None = None, + save_error: Exception | None = None, +) -> type[TargetCacheBackendType]: + RecordingCacheBackend.exists_result = exists_result + RecordingCacheBackend.exists_error = exists_error + RecordingCacheBackend.load_error = load_error + RecordingCacheBackend.save_error = save_error + RecordingCacheBackend.exists_calls = 0 + RecordingCacheBackend.load_calls = 0 + RecordingCacheBackend.save_calls = 0 + return cast(type[TargetCacheBackendType], RecordingCacheBackend) + + class TestTarget: def test_init(self): with pytest.raises(ValidationError) as exc_info: @@ -40,132 +92,154 @@ def test_cache_path(self, mock_test_spec: mock.Mock): def test_cache_archive_filename(self, mock_test_spec: mock.Mock): assert mock_test_spec.targets["quack:test"].cache_archive_filename == "quack:test.tar.zst" - @mock.patch("quack.cache.TargetCache") - def test_execute_deps_only(self, mock_target_cache, mock_test_spec: mock.Mock): + def test_execute_deps_only(self, mock_test_spec: mock.Mock): config = Config.model_construct() target = mock_test_spec.targets["quack:test"] target._checksum_value = "" + cache_backend = reset_recording_cache_backend() # 当 mode=TargetExecutionMode.DEPS_ONLY 仅构建依赖项 - mock_target_cache.return_value.hit.return_value = False target.execute( config, mock_test_spec.app_name, - mock.Mock, + cache_backend, mode=TargetExecutionMode.DEPS_ONLY, ) - @mock.patch("quack.cache.TargetCache") - def test_execute_cache_hit(self, mock_target_cache, mock_test_spec: mock.Mock): + assert RecordingCacheBackend.exists_calls == 0 + assert RecordingCacheBackend.load_calls == 0 + assert RecordingCacheBackend.save_calls == 0 + + def test_execute_cache_hit(self, mock_test_spec: mock.Mock): config = Config.model_construct() target = mock_test_spec.targets["quack:test"] target._checksum_value = "" + cache_backend = reset_recording_cache_backend(exists_result=True) # 当缓存命中时，直接加载缓存 - mock_target_cache.return_value.hit.return_value = True - target.execute(config, mock_test_spec.app_name, mock.Mock) - mock_target_cache.return_value.load.assert_called_once() + target.execute(config, mock_test_spec.app_name, cache_backend) + assert RecordingCacheBackend.load_calls == 1 - @mock.patch("quack.cache.TargetCache") - def test_execute_rebuilds_after_transient_cache_hit_failure(self, mock_target_cache, mock_test_spec: mock.Mock): + def test_execute_rebuilds_after_transient_cache_hit_failure(self, mock_test_spec: mock.Mock): config = Config.model_construct() target = mock_test_spec.targets["quack:test"] target._checksum_value = "" - - mock_target_cache.return_value.hit.side_effect = CloudStorageTransientError( - "检查文件是否存在失败", - "Could not connect to the endpoint URL", + cache_backend = reset_recording_cache_backend( + exists_error=CloudStorageTransientError( + "检查文件是否存在失败", + "Could not connect to the endpoint URL", + ) ) + with mock.patch("quack.models.command.Command.execute") as mock_build: - target.execute(config, mock_test_spec.app_name, mock.Mock) + target.execute(config, mock_test_spec.app_name, cache_backend) mock_build.assert_called_once() - mock_target_cache.return_value.load.assert_not_called() - mock_target_cache.return_value.save.assert_called_once() + assert RecordingCacheBackend.load_calls == 0 + assert RecordingCacheBackend.save_calls == 1 - @mock.patch("quack.cache.TargetCache") - def test_execute_rebuilds_after_transient_cache_load_failure(self, mock_target_cache, mock_test_spec: mock.Mock): + def test_execute_rebuilds_after_transient_cache_load_failure(self, mock_test_spec: mock.Mock): config = Config.model_construct() target = mock_test_spec.targets["quack:test"] target._checksum_value = "" - - mock_target_cache.return_value.hit.return_value = True - mock_target_cache.return_value.load.side_effect = CloudStorageTransientError( - "下载文件失败", - "IncompleteBody", - code="IncompleteBody", + cache_backend = reset_recording_cache_backend( + exists_result=True, + load_error=CloudStorageTransientError( + "下载文件失败", + "IncompleteBody", + code="IncompleteBody", + ), ) + with mock.patch("quack.models.command.Command.execute") as mock_build: - target.execute(config, mock_test_spec.app_name, mock.Mock) + target.execute(config, mock_test_spec.app_name, cache_backend) - mock_target_cache.return_value.load.assert_called_once() + assert RecordingCacheBackend.load_calls == 1 mock_build.assert_called_once() - mock_target_cache.return_value.save.assert_called_once() + assert RecordingCacheBackend.save_calls == 1 - @mock.patch("quack.cache.TargetCache") - def test_execute_raises_non_transient_cache_load_failure(self, mock_target_cache, mock_test_spec: mock.Mock): + def test_execute_rebuilds_after_corrupt_cache_load(self, mock_test_spec: mock.Mock): config = Config.model_construct() target = mock_test_spec.targets["quack:test"] target._checksum_value = "" + cache_backend = reset_recording_cache_backend( + exists_result=True, + load_error=CacheCorruptionError("缓存归档解压失败"), + ) + + with mock.patch("quack.models.command.Command.execute") as mock_build: + target.execute(config, mock_test_spec.app_name, cache_backend) - mock_target_cache.return_value.hit.return_value = True - mock_target_cache.return_value.load.side_effect = CloudStorageError( - "下载文件失败", - "AccessDenied", - code="AccessDenied", + assert RecordingCacheBackend.load_calls == 1 + mock_build.assert_called_once() + assert RecordingCacheBackend.save_calls == 1 + + def test_execute_raises_non_transient_cache_load_failure(self, mock_test_spec: mock.Mock): + config = Config.model_construct() + target = mock_test_spec.targets["quack:test"] + target._checksum_value = "" + cache_backend = reset_recording_cache_backend( + exists_result=True, + load_error=CloudStorageError( + "下载文件失败", + "AccessDenied", + code="AccessDenied", + ), ) + with mock.patch("quack.models.command.Command.execute") as mock_build, pytest.raises(CloudStorageError): - target.execute(config, mock_test_spec.app_name, mock.Mock) + target.execute(config, mock_test_spec.app_name, cache_backend) mock_build.assert_not_called() - mock_target_cache.return_value.save.assert_not_called() + assert RecordingCacheBackend.save_calls == 0 - @mock.patch("quack.cache.TargetCache") - def test_execute_ignores_transient_cache_save_failure(self, mock_target_cache, mock_test_spec: mock.Mock): + def test_execute_ignores_transient_cache_save_failure(self, mock_test_spec: mock.Mock): config = Config.model_construct() target = mock_test_spec.targets["quack:test"] target._checksum_value = "" - - mock_target_cache.return_value.hit.return_value = False - mock_target_cache.return_value.save.side_effect = CloudStorageTransientError( - "上传文件失败", - "IncompleteBody", - code="IncompleteBody", + cache_backend = reset_recording_cache_backend( + exists_result=False, + save_error=CloudStorageTransientError( + "上传文件失败", + "IncompleteBody", + code="IncompleteBody", + ), ) + with mock.patch("quack.models.command.Command.execute") as mock_build: - target.execute(config, mock_test_spec.app_name, mock.Mock) + target.execute(config, mock_test_spec.app_name, cache_backend) mock_build.assert_called_once() - mock_target_cache.return_value.save.assert_called_once() + assert RecordingCacheBackend.save_calls == 1 - @mock.patch("quack.cache.TargetCache") - def test_execute_raises_non_transient_cache_save_failure(self, mock_target_cache, mock_test_spec: mock.Mock): + def test_execute_raises_non_transient_cache_save_failure(self, mock_test_spec: mock.Mock): config = Config.model_construct() target = mock_test_spec.targets["quack:test"] target._checksum_value = "" - - mock_target_cache.return_value.hit.return_value = False - mock_target_cache.return_value.save.side_effect = CloudStorageError( - "上传文件失败", - "AccessDenied", - code="AccessDenied", + cache_backend = reset_recording_cache_backend( + exists_result=False, + save_error=CloudStorageError( + "上传文件失败", + "AccessDenied", + code="AccessDenied", + ), ) + with mock.patch("quack.models.command.Command.execute"), pytest.raises(CloudStorageError): - target.execute(config, mock_test_spec.app_name, mock.Mock) + target.execute(config, mock_test_spec.app_name, cache_backend) - @mock.patch("quack.cache.TargetCache") - def test_execute_load_only(self, mock_target_cache, mock_test_spec: mock.Mock): + def test_execute_load_only(self, mock_test_spec: mock.Mock): config = Config.model_construct() target = mock_test_spec.targets["quack:test"] target._checksum_value = "" + cache_backend = reset_recording_cache_backend(exists_result=False) # 当 mode=TargetExecutionMode.LOAD_ONLY 且缓存未命中时，应该退出 - mock_target_cache.return_value.hit.return_value = False with pytest.raises(SystemExit): target.execute( config, mock_test_spec.app_name, - mock.Mock, + cache_backend, mode=TargetExecutionMode.LOAD_ONLY, ) diff --git a/src/quack/utils/cloud.py b/src/quack/utils/cloud.py index 3a5a099..077c6af 100644 --- a/src/quack/utils/cloud.py +++ b/src/quack/utils/cloud.py @@ -48,6 +48,8 @@ ReadTimeoutError, ) +CLIENT_ERROR_MESSAGE_RE = re.compile(r"An error occurred \(([^)]+)\) when calling the \w+ operation") + @dataclass class CloudFileMetadata: @@ -88,11 +90,7 @@ def _extract_error_code(error: Exception) -> str: if isinstance(current_error, ClientError): return _get_client_error_code(current_error) - match = re.search(r"An error occurred \(([^)]+)\)", str(current_error)) - if match: - return match.group(1) - - match = re.search(r"\(([^)]+)\)", str(current_error)) + match = CLIENT_ERROR_MESSAGE_RE.search(str(current_error)) if match: return match.group(1) diff --git a/src/quack/utils/cloud_test.py b/src/quack/utils/cloud_test.py index 9660e04..83cc029 100644 --- a/src/quack/utils/cloud_test.py +++ b/src/quack/utils/cloud_test.py @@ -39,6 +39,12 @@ def _s3_upload_error_from_client_error(code: str) -> S3UploadFailedError: return e +def _s3_upload_error_from_message(code: str) -> S3UploadFailedError: + return S3UploadFailedError( + f"Failed to upload cache.tar.zst: An error occurred ({code}) when calling the PutObject operation: {code}" + ) + + def test_upload_wraps_incomplete_body_as_transient_error(tmp_path: Path): local_file = tmp_path / "cache.tar.zst" local_file.write_bytes(b"cache") @@ -51,6 +57,18 @@ def test_upload_wraps_incomplete_body_as_transient_error(tmp_path: Path): assert exc_info.value.code == "IncompleteBody" +def test_upload_extracts_error_code_from_s3_transfer_message(tmp_path: Path): + local_file = tmp_path / "cache.tar.zst" + local_file.write_bytes(b"cache") + + client = _cloud_client_with_upload_error(_s3_upload_error_from_message("IncompleteBody")) + + with pytest.raises(CloudStorageTransientError) as exc_info: + client.upload(str(local_file), "dest/cache.tar.zst") + + assert exc_info.value.code == "IncompleteBody" + + def test_upload_keeps_access_denied_as_non_transient_error(tmp_path: Path): local_file = tmp_path / "cache.tar.zst" local_file.write_bytes(b"cache")