refactor: 使用 zstd 替代 gzip 压缩

CLAUDE.md

@@ -93,5 +93,5 @@ src/quack/
 ```
 .quack-cache/<app-name>/<target-name>/<checksum>/
 ├── _metadata.json     # 元数据
-└── _archive.tar.gz    # 产物归档
+└── _archive.tar.zst   # 产物归档
 ```

pyproject.toml

@@ -16,6 +16,7 @@ dependencies = [
     "pytest>=8.3.5",
     "pyyaml>=6.0.2",
     "xdg-base-dirs>=6.0.2",
+    "zstandard>=0.23.0",
 ]
 
 [project.scripts]

src/quack/models/target.py

@@ -64,7 +64,7 @@ def cache_path(self) -> str:
 
     @property
     def cache_archive_filename(self) -> str:
-        return f"{self.name}.tar.gz"
+        return f"{self.name}.tar.zst"
 
     def compute_checksum(self) -> str:
         hash_tuple = [dep.checksum_value for dep in self.dependencies]

src/quack/models/target_test.py

@@ -37,7 +37,7 @@ def test_cache_path(self, mock_test_spec: mock.Mock):
         assert mock_test_spec.targets["quack:test"].cache_path.startswith("quack:test/")
 
     def test_cache_archive_filename(self, mock_test_spec: mock.Mock):
-        assert mock_test_spec.targets["quack:test"].cache_archive_filename == "quack:test.tar.gz"
+        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):

src/quack/utils/archiver.py

@@ -4,27 +4,56 @@
 
 import os
 import subprocess
+import tarfile
 import tempfile
 from collections.abc import Iterable
 
+import zstandard as zstd
+
 
 class Archiver:
     @staticmethod
     def archive(paths: Iterable[str], archive_path: str) -> None:
-        paths_str = " ".join(paths) if paths else "-T /dev/null"
-        cmd = f"tar czf {archive_path} {paths_str}"
-        env = os.environ.copy()
-        # 防止 macOS tar 包含 ._ 资源分支文件
-        env["COPYFILE_DISABLE"] = "1"
-        _ = subprocess.run(cmd, shell=True, check=True, env=env)
+        with tempfile.NamedTemporaryFile(suffix=".tar", delete=False) as tmp_tar:
+            tmp_tar_path = tmp_tar.name
+
+        try:
+            with tarfile.open(tmp_tar_path, "w") as tar:
+                for path in paths:
+                    tar.add(path, arcname=path)
+
+            with open(tmp_tar_path, "rb") as f_in:
+                tar_data = f_in.read()
+
+            cctx = zstd.ZstdCompressor()
+            compressed_data = cctx.compress(tar_data)
+
+            if dirname := os.path.dirname(archive_path):
+                os.makedirs(dirname, exist_ok=True)
+            with open(archive_path, "wb") as f_out:
+                f_out.write(compressed_data)
+        finally:
+            if os.path.exists(tmp_tar_path):
+                os.unlink(tmp_tar_path)
 
     @staticmethod
     def extract(archive_path: str, dest_path: str = ".") -> None:
-        # Create a temporary directory for extraction
         with tempfile.TemporaryDirectory() as temp_dir:
-            # 先解压到临时目录
-            cmd_extract = f"tar xf {archive_path} -C {temp_dir}"
-            _ = subprocess.run(cmd_extract, shell=True, check=True)
+            with open(archive_path, "rb") as f_in:
+                compressed_data = f_in.read()
+
+            dctx = zstd.ZstdDecompressor()
+            tar_data = dctx.decompress(compressed_data)
+
+            with tempfile.NamedTemporaryFile(suffix=".tar", delete=False) as tmp_tar:
+                tmp_tar.write(tar_data)
+                tmp_tar_path = tmp_tar.name
+
+            try:
+                with tarfile.open(tmp_tar_path, "r") as tar:
+                    tar.extractall(temp_dir, filter="data")
+            finally:
+                os.unlink(tmp_tar_path)
 
             # 使用 rsync 同步到目标目录，相同内容的文件不会被覆盖
             cmd_rsync = f"rsync --recursive --links --checksum {temp_dir}/ {dest_path}/"

src/quack/utils/archiver_test.py

@@ -9,7 +9,7 @@ def test_archiver(self, tmp_path):
         tmp_file = tmp_path / "test.txt"
         tmp_file.write_text("test")
 
-        tmp_archive = tmp_path / "test.tar.gz"
+        tmp_archive = tmp_path / "test.tar.zst"
         Archiver.archive([str(tmp_file)], str(tmp_archive))
         assert tmp_archive.exists()
 
@@ -23,7 +23,7 @@ def test_extract_behavior(self, tmp_path):
         tmp_file.write_text("original content")
         origin_timestamp = os.path.getmtime(tmp_file)
 
-        tmp_archive = tmp_path / "test.tar.gz"
+        tmp_archive = tmp_path / "test.tar.zst"
         Archiver.archive([str(tmp_file)], str(tmp_archive))
 
         # 测试1: 内容相同时保留时间戳