feat(storage): 本地复制优先尝试写时复制(reflink)并支持自动回退 (#7082)

app/adapters/system/host.py

@@ -494,9 +494,20 @@ class SystemUtils:
     @staticmethod
     def copy(src: Path, dest: Path) -> Tuple[int, str]:
         """
-        复制
+        复制文件。
+
+        若源与目标在同一磁盘且支持写时复制 (reflink)，优先尝试克隆；
+        失败或不支持时自动回退为常规复制。
+
+        :param src: 源文件路径
+        :param dest: 目标文件路径
+        :return: (状态码, 错误信息)，0 表示成功
         """
         try:
+            if SystemUtils.is_same_disk(src, dest):
+                code, _ = SystemUtils.reflink(src, dest)
+                if code == 0:
+                    return 0, ""
             shutil.copy2(src, dest)
             return 0, ""
         except Exception as err:

app/modules/filemanager/storages/local.py

@@ -319,12 +319,47 @@ class LocalStorage(StorageBase):
         except Exception as err:
             logger.debug(f"【本地】清理临时文件失败：{directory} - {err}")
 
+    @staticmethod
+    def _try_reflink_copy(src: Path, dest: Path) -> bool:
+        """
+        若源与目标位于同一磁盘且支持写时复制，优先尝试通过引用链接克隆。
+
+        :param src: 源文件路径
+        :param dest: 目标文件路径
+        :return: 成功完成克隆并同步时间戳与权限策略返回 True，否则返回 False
+        """
+        if not SystemUtils.is_same_disk(src, dest):
+            return False
+        try:
+            code, _ = SystemUtils.reflink(src, dest)
+            if code != 0:
+                return False
+            src_stat = src.stat()
+            os.utime(dest, ns=(src_stat.st_atime_ns, src_stat.st_mtime_ns))
+            sentinel = dest.parent / f".mode-probe-{os.getpid()}"
+            try:
+                sentinel.write_bytes(b"")
+                dest_mode = sentinel.stat().st_mode & 0o777
+                dest.chmod(dest_mode)
+            finally:
+                sentinel.unlink(missing_ok=True)
+            return True
+        except Exception as err:
+            logger.debug(f"【本地】尝试写时复制克隆失败，回退常规复制：{err}")
+            if dest.exists():
+                try:
+                    dest.unlink()
+                except OSError:
+                    pass
+            return False
+
     def _write_atomically(self, src: Path, dest: Path) -> bool:
         """
         先复制到同目录中间文件，完成后原子替换目标名称。
 
         目标已有内容在替换前保留；此处保证名称替换的原子性，不额外执行断电持久化同步。
         结束时尽力删除中间文件，进度展示会收尾到 100，不能仅据此判断复制成功。
+        同一文件系统且支持写时复制时优先使用 reflink，失败或不支持时自动回退为常规复制。
 
         :param src: 源文件路径，成功后仍保留
         :param dest: 完整目标文件路径；父目录需已存在
@@ -332,6 +367,14 @@ class LocalStorage(StorageBase):
         """
         self._cleanup_stale_partials(dest.parent)
         partial = self._partial_path(dest)
+        if runtime_stop_state.consume_transfer_stop(src.as_posix()):
+            logger.info(f"【本地】{src} 复制已取消")
+            return False
+
+        # 同一文件系统优先尝试写时复制克隆，成功即完成原子替换
+        if self._try_reflink_copy(src, partial):
+            os.replace(partial, dest)
+            return True
         # 进度只在需要展示时才回调 UI，但代理内部始终按固定间隔上报——那是判定
         # 「传输是否还在推进」的心跳，不能因为不展示进度就关掉
         progress_callback = (
@@ -466,6 +509,8 @@ class LocalStorage(StorageBase):
         :param new_name: 目标文件名
         :return: 目标写入完成返回 True，取消或失败返回 False
         """
+        if not fileitem.path:
+            return False
         return self._write_atomically(Path(fileitem.path), path / new_name)
 
     def move(

tests/fixtures/architecture/mypy-baseline.json

@@ -640,7 +640,7 @@
     "type-arg": 1
   },
   "app/modules/filemanager/storages/local.py": {
-    "arg-type": 8,
+    "arg-type": 7,
     "assignment": 1,
     "truthy-function": 1
   },

tests/test_storage_reflink.py

@@ -252,3 +252,138 @@ def test_filemanager_check_reflink_unsupported_storage(monkeypatch: pytest.Monke
     success, message = module.test()
     assert success is False
     assert "测试远端目录 的存储不支持 reflink 整理方式" in message
+
+
+def test_local_storage_copy_prioritizes_reflink_on_same_disk(
+    monkeypatch: pytest.MonkeyPatch, tmp_path: Path,
+) -> None:
+    """
+    测试本地复制在同文件系统且支持写时复制时，优先尝试引用链接。
+
+    :param monkeypatch: pytest 猴子补丁夹具
+    :param tmp_path: pytest 临时测试目录
+    """
+    storage = _make_local_storage()
+    source = tmp_path / "src.bin"
+    source.write_bytes(b"reflink copy content")
+    target_dir = tmp_path / "dest"
+    target_dir.mkdir()
+
+    reflink_called = []
+
+    def _mock_reflink(src: Path, dest: Path) -> Tuple[int, str]:
+        reflink_called.append((src, dest))
+        dest.write_bytes(src.read_bytes())
+        return 0, ""
+
+    monkeypatch.setattr(SystemUtils, "reflink", _mock_reflink)
+    monkeypatch.setattr(SystemUtils, "is_same_disk", lambda _s, _d: True)
+
+    fileitem = FileItem(storage="local", path=source.as_posix(), name="src.bin")
+    success = storage.copy(fileitem, target_dir, "dest.bin")
+
+    assert success is True
+    assert len(reflink_called) == 1
+    assert (target_dir / "dest.bin").read_bytes() == b"reflink copy content"
+
+
+def test_local_storage_copy_falls_back_when_reflink_fails(
+    monkeypatch: pytest.MonkeyPatch, tmp_path: Path,
+) -> None:
+    """
+    测试本地复制在引用链接失败时，自动回退到常规文件复制。
+
+    :param monkeypatch: pytest 猴子补丁夹具
+    :param tmp_path: pytest 临时测试目录
+    """
+    storage = _make_local_storage()
+    source = tmp_path / "src.bin"
+    source.write_bytes(b"fallback copy content")
+    target_dir = tmp_path / "dest"
+    target_dir.mkdir()
+
+    reflink_called = []
+
+    def _failing_reflink(src: Path, dest: Path) -> Tuple[int, str]:
+        reflink_called.append((src, dest))
+        return -1, "Operation not supported"
+
+    monkeypatch.setattr(SystemUtils, "reflink", _failing_reflink)
+    monkeypatch.setattr(SystemUtils, "is_same_disk", lambda _s, _d: True)
+
+    fileitem = FileItem(storage="local", path=source.as_posix(), name="src.bin")
+    success = storage.copy(fileitem, target_dir, "dest.bin")
+
+    assert success is True
+    assert len(reflink_called) == 1
+    assert (target_dir / "dest.bin").read_bytes() == b"fallback copy content"
+
+
+def test_local_storage_copy_skips_reflink_on_different_disks(
+    monkeypatch: pytest.MonkeyPatch, tmp_path: Path,
+) -> None:
+    """
+    测试跨磁盘复制时不尝试引用链接，直接走常规复制。
+
+    :param monkeypatch: pytest 猴子补丁夹具
+    :param tmp_path: pytest 临时测试目录
+    """
+    storage = _make_local_storage()
+    source = tmp_path / "src.bin"
+    source.write_bytes(b"cross disk copy content")
+    target_dir = tmp_path / "dest"
+    target_dir.mkdir()
+
+    reflink_called = []
+
+    def _mock_reflink(src: Path, dest: Path) -> Tuple[int, str]:
+        reflink_called.append((src, dest))
+        return 0, ""
+
+    monkeypatch.setattr(SystemUtils, "reflink", _mock_reflink)
+    monkeypatch.setattr(SystemUtils, "is_same_disk", lambda _s, _d: False)
+
+    fileitem = FileItem(storage="local", path=source.as_posix(), name="src.bin")
+    success = storage.copy(fileitem, target_dir, "dest.bin")
+
+    assert success is True
+    assert len(reflink_called) == 0
+    assert (target_dir / "dest.bin").read_bytes() == b"cross disk copy content"
+
+
+def test_system_utils_copy_reflink_priority_and_fallback(
+    monkeypatch: pytest.MonkeyPatch, tmp_path: Path,
+) -> None:
+    """
+    测试 SystemUtils.copy 优先尝试引用链接并在失败时回退常规复制。
+
+    :param monkeypatch: pytest 猴子补丁夹具
+    :param tmp_path: pytest 临时测试目录
+    """
+    source = tmp_path / "src.bin"
+    source.write_bytes(b"system utils copy test")
+    dest1 = tmp_path / "dest1.bin"
+    dest2 = tmp_path / "dest2.bin"
+
+    calls = []
+
+    def _mock_reflink(src: Path, dest: Path) -> Tuple[int, str]:
+        calls.append((src, dest))
+        if dest == dest1:
+            dest.write_bytes(src.read_bytes())
+            return 0, ""
+        return -1, "CoW not supported"
+
+    monkeypatch.setattr(SystemUtils, "reflink", _mock_reflink)
+    monkeypatch.setattr(SystemUtils, "is_same_disk", lambda _s, _d: True)
+
+    code1, msg1 = SystemUtils.copy(source, dest1)
+    assert code1 == 0
+    assert msg1 == ""
+    assert dest1.read_bytes() == b"system utils copy test"
+
+    code2, msg2 = SystemUtils.copy(source, dest2)
+    assert code2 == 0
+    assert msg2 == ""
+    assert dest2.read_bytes() == b"system utils copy test"
+    assert len(calls) == 2