fix(cache): 缓存损坏或临时云存储错误时降级重建

src/quack/cache.py

@@ -9,12 +9,13 @@
 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
 
 from quack.config import Config
 from quack.consts import CACHE_METADATA_FILENAME
-from quack.exceptions import ChecksumError
+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
@@ -68,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)
@@ -192,10 +196,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):
@@ -204,13 +211,22 @@ def load(self, target: Target, update_access_time: bool = True) -> None:
                 if update_access_time:
                     self.update_access_time(target)
                 return
-            except ChecksumError:
+            except CacheCorruptionError:
                 logger.warning("本地缓存已损坏，从云存储重新下载")
-
-        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)
+                shutil.rmtree(self.local_backend.get_cache_path(target), ignore_errors=True)
+
+        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 CacheCorruptionError:
+            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)
 

src/quack/cache_test.py

@@ -1,8 +1,28 @@
 import os
 from unittest import mock
 
-from quack.cache import TargetCacheBackendTypeCloud
+import pytest
+
+from quack.cache import TargetCacheBackendTypeCloud, TargetCacheBackendTypeLocal
 from quack.config import Config
+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:
@@ -29,6 +49,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(
@@ -53,6 +122,58 @@ 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
+        mock_local_backend.return_value.load.side_effect = [CacheCorruptionError("缓存归档解压失败"), 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 = CacheCorruptionError("缓存归档解压失败")
+
+        with pytest.raises(CacheCorruptionError):
+            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")

src/quack/exceptions.py

@@ -1,4 +1,8 @@
-class ChecksumError(Exception):
+class CacheCorruptionError(Exception):
+    pass
+
+
+class ChecksumError(CacheCorruptionError):
     pass
 
 
@@ -7,7 +11,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

src/quack/models/target.py

@@ -13,7 +13,7 @@
 from pydantic import Field
 
 from quack.config import Config
-from quack.exceptions import CloudStorageError
+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
@@ -91,36 +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)
-        cache_exists = cache.hit()
 
         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("找到缓存，直接从缓存加载...")
-                cache.load()
-            else:
-                logger.info(f"正在存入缓存，路径：{self.cache_path}")
-                try:
-                    cache.save()
-                except CloudStorageError as e:
-                    logger.error(f"存入缓存失败：{e}")
+            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()
+                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:
+                        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} 执行完毕！")