fix(system): add opt-in Btrfs FSID space deduplication (#6141)

app/api/endpoints/dashboard.py

@@ -7,6 +7,7 @@ from sqlalchemy.orm import Session
 from app import schemas
 from app.chain.dashboard import DashboardChain
 from app.chain.storage import StorageChain
+from app.core.config import settings
 from app.core.security import verify_apitoken
 from app.db import get_db
 from app.db.models.transferhistory import TransferHistory
@@ -73,7 +74,8 @@ def _build_downloader(name: Optional[str] = None) -> schemas.DownloaderInfo:
     # 下载目录空间
     download_dirs = DirectoryHelper().get_local_download_dirs()
     _, free_space = SystemUtils.space_usage(
-        [Path(d.download_path) for d in download_dirs]
+        [Path(d.download_path) for d in download_dirs],
+        btrfs_fsid_dedup=settings.BTRFS_FSID_DEDUP,
     )
     # 下载器信息
     downloader_info = schemas.DownloaderInfo()

app/core/config.py

@@ -78,6 +78,8 @@ class ConfigModel(BaseModel):
     CONFIG_DIR: Optional[str] = None
     # 安全模式，仅保留核心 API，跳过插件、调度器、监控、命令和工作流等扩展启动项
     MOVIEPILOT_SAFE_MODE: bool = False
+    # 是否启用 Btrfs FSID 子卷容量去重（仅 Linux amd64/arm64）
+    BTRFS_FSID_DEDUP: bool = False
     # 是否调试模式
     DEBUG: bool = False
     # 是否开发模式

app/modules/filemanager/storages/local.py

@@ -3,7 +3,7 @@ from pathlib import Path
 from typing import Optional, List
 
 from app import schemas
-from app.core.config import global_vars
+from app.core.config import global_vars, settings
 from app.helper.directory import DirectoryHelper
 from app.log import logger
 from app.modules.filemanager.storages import StorageBase, transfer_process
@@ -341,7 +341,8 @@ class LocalStorage(StorageBase):
         directory_helper = DirectoryHelper()
         total_storage, free_storage = SystemUtils.space_usage(
             [Path(d.download_path) for d in directory_helper.get_local_download_dirs() if d.download_path] +
-            [Path(d.library_path) for d in directory_helper.get_local_library_dirs() if d.library_path]
+            [Path(d.library_path) for d in directory_helper.get_local_library_dirs() if d.library_path],
+            btrfs_fsid_dedup=settings.BTRFS_FSID_DEDUP,
         )
         return schemas.StorageUsage(
             total=total_storage,

app/utils/system.py

@@ -5,6 +5,7 @@ import platform
 import re
 import shutil
 import socket
+import struct
 import subprocess
 import sys
 import time
@@ -13,12 +14,24 @@ import uuid
 from pathlib import Path
 from typing import List, Optional, Tuple, Union
 
+try:
+    import fcntl
+except ImportError:
+    fcntl = None
+
 import psutil
 
 from app import schemas
 from version import APP_VERSION
 
 
+# Linux amd64/arm64 UAPI: _IOR(BTRFS_IOCTL_MAGIC, 31, struct btrfs_ioctl_fs_info_args)
+_BTRFS_IOC_FS_INFO = 0x8400941F
+_BTRFS_FS_INFO_SIZE = 1024
+_BTRFS_FSID_OFFSET = 16
+_BTRFS_FSID_SIZE = 16
+
+
 class SystemUtils:
     """
     系统工具类，提供系统相关的操作和信息获取方法。
@@ -550,31 +563,92 @@ class SystemUtils:
         return _calc_dir_size(path) if path.is_dir() else path.stat().st_size
 
     @staticmethod
-    def space_usage(dir_list: Union[Path, List[Path]]) -> Tuple[float, float]:
+    def _get_btrfs_fsid(dir_path: Path) -> Optional[bytes]:
+        """读取目录所属 Btrfs 文件系统的 FSID，无法确认时返回 None。"""
+        if not sys.platform.startswith("linux") or fcntl is None \
+                or not (SystemUtils.is_x86_64() or SystemUtils.is_aarch64()):
+            return None
+
+        fd = None
+        try:
+            flags = os.O_RDONLY | getattr(os, "O_DIRECTORY", 0) | getattr(os, "O_CLOEXEC", 0)
+            fd = os.open(dir_path, flags)
+            fs_info = bytearray(_BTRFS_FS_INFO_SIZE)
+            fcntl.ioctl(fd, _BTRFS_IOC_FS_INFO, fs_info, True)
+            if len(fs_info) != _BTRFS_FS_INFO_SIZE:
+                return None
+            num_devices = struct.unpack_from("=Q", fs_info, 8)[0]
+            fsid = bytes(fs_info[_BTRFS_FSID_OFFSET:_BTRFS_FSID_OFFSET + _BTRFS_FSID_SIZE])
+            if num_devices == 0 or fsid == bytes(_BTRFS_FSID_SIZE):
+                return None
+            return fsid
+        except (OSError, OverflowError, ValueError, struct.error):
+            return None
+        finally:
+            if fd is not None:
+                try:
+                    os.close(fd)
+                except OSError:
+                    pass
+
+    @staticmethod
+    def space_usage(dir_list: Union[Path, List[Path]], btrfs_fsid_dedup: bool = False) -> Tuple[float, float]:
         """
         计算多个目录的总可用空间/剩余空间（单位：Byte），并去除重复磁盘
+
+        :param dir_list: 待统计的目录或目录列表
+        :param btrfs_fsid_dedup: 是否在 Linux amd64/arm64 下以 Btrfs FSID 辅助去重
+
+        默认保持原有的驱动器号/st_dev 去重。显式启用后，st_dev 仍作为基础依据，
+        FSID 仅用于合并同一 Btrfs 文件系统中的不同子卷；证据缺失或冲突时
+        保守回退 st_dev，允许多计而不猜测合并独立存储。
         """
         if not dir_list:
             return 0.0, 0.0
         if not isinstance(dir_list, list):
             dir_list = [dir_list]
-        # 存储不重复的磁盘
+
-        disk_set = set()
+        use_btrfs_fsid = btrfs_fsid_dedup and sys.platform.startswith("linux") \
-        # 存储总剩余空间
+            and (SystemUtils.is_x86_64() or SystemUtils.is_aarch64())
+        if not use_btrfs_fsid:
+            # 默认分支保留原有行为，不打开目录或调用 ioctl。
+            disk_set = set()
+            total_free_space = 0.0
+            total_space = 0.0
+            for dir_path in dir_list:
+                if not dir_path:
+                    continue
+                if not dir_path.exists():
+                    continue
+                if os.name == "nt":
+                    disk = dir_path.drive
+                else:
+                    disk = os.stat(dir_path).st_dev
+                if disk not in disk_set:
+                    disk_set.add(disk)
+                    total_space += SystemUtils.total_space(dir_path)
+                    total_free_space += SystemUtils.free_space(dir_path)
+            return total_space, total_free_space
+
         total_free_space = 0.0
-        # 存储总空间
         total_space = 0.0
+        # 先按 st_dev 恢复原有去重语义，再用组内唯一可信的 FSID 合并 Btrfs 子卷。
+        disk_groups = {}
         for dir_path in dir_list:
-            if not dir_path:
+            if not dir_path or not dir_path.exists():
                 continue
-            if not dir_path.exists():
+            st_dev = os.stat(dir_path).st_dev
-                continue
+            if st_dev not in disk_groups:
-            # 获取目录所在磁盘
+                disk_groups[st_dev] = (dir_path, set())
-            if os.name == "nt":
+            btrfs_fsid = SystemUtils._get_btrfs_fsid(dir_path)
-                disk = dir_path.drive
+            if btrfs_fsid:
-            else:
+                disk_groups[st_dev][1].add(btrfs_fsid)
-                disk = os.stat(dir_path).st_dev
+
-            # 如果磁盘未出现过，则计算其剩余空间并加入总剩余空间中
+        disk_set = set()
+        for st_dev, (dir_path, fsids) in disk_groups.items():
+            # 同一 st_dev 出现冲突 FSID 可能是挂载变化导致的不一致快照，
+            # 此时禁止跨 st_dev 合并，避免少计独立存储。
+            disk = ("btrfs_fsid", next(iter(fsids))) if len(fsids) == 1 else ("st_dev", st_dev)
             if disk not in disk_set:
                 disk_set.add(disk)
                 total_space += SystemUtils.total_space(dir_path)

tests/test_system_utils.py

@@ -1,10 +1,17 @@
+import errno
+import itertools
+import os
+import struct
 import subprocess
 import tempfile
+from pathlib import Path
 from unittest import TestCase
-from unittest.mock import patch
+from unittest.mock import MagicMock, call, patch
+
+import pytest
 
 from app.helper.system import SystemHelper
-from app.core.config import settings
+from app.core.config import ConfigModel, settings
 from app.utils.system import SystemUtils
 
 
@@ -172,3 +179,395 @@ def test_execute_with_subprocess_redacts_unknown_error_userinfo_and_invalid_port
     assert not success
     assert "https://example.com:notaport/simple" in message
     assert "user:pass" not in message
+
+
+def _fake_stat_with_devs(dev_by_path):
+    """构造按路径返回指定 st_dev 的 os.stat 桩。"""
+    def _fake_stat(path, *args, **kwargs):
+        result = MagicMock()
+        result.st_dev = dev_by_path[str(path)]
+        return result
+    return _fake_stat
+
+
+def _fill_fs_info(fsid, num_devices=1, truncate=False):
+    """构造填充 BTRFS_IOC_FS_INFO 缓冲区的 ioctl 桩。"""
+    def _ioctl(fd, request, buffer, mutate_flag):
+        assert fd == 42
+        assert request == 0x8400941F
+        assert mutate_flag is True
+        if truncate:
+            del buffer[64:]
+            return 0
+        struct.pack_into("=Q", buffer, 8, num_devices)
+        buffer[16:32] = fsid
+        return 0
+    return _ioctl
+
+
+@pytest.fixture
+def linux_platform(monkeypatch):
+    """将当前用例的平台标识切换为 Linux amd64。"""
+    monkeypatch.setattr("app.utils.system.sys.platform", "linux")
+    monkeypatch.setattr(SystemUtils, "is_x86_64", lambda: True)
+    monkeypatch.setattr(SystemUtils, "is_aarch64", lambda: False)
+
+
+@pytest.mark.parametrize("num_devices", [1, 2])
+def test_get_btrfs_fsid_reads_kernel_result_and_closes_fd(num_devices, linux_platform):
+    fsid = bytes.fromhex("88e3aff5fa2946d591a55977be984655")
+    with patch("app.utils.system.os.open", return_value=42), \
+            patch("app.utils.system.fcntl.ioctl", side_effect=_fill_fs_info(fsid, num_devices)), \
+            patch("app.utils.system.os.close") as close_mock:
+        result = SystemUtils._get_btrfs_fsid(Path("/data"))
+
+    assert result == fsid
+    close_mock.assert_called_once_with(42)
+
+
+@pytest.mark.parametrize("error_number", [errno.ENOTTY, errno.EACCES, errno.EPERM, errno.EINVAL])
+def test_get_btrfs_fsid_falls_back_on_expected_ioctl_errors(error_number, linux_platform):
+    with patch("app.utils.system.os.open", return_value=42), \
+            patch("app.utils.system.fcntl.ioctl", side_effect=OSError(error_number, os.strerror(error_number))), \
+            patch("app.utils.system.os.close") as close_mock:
+        result = SystemUtils._get_btrfs_fsid(Path("/data"))
+
+    assert result is None
+    close_mock.assert_called_once_with(42)
+
+
+def test_get_btrfs_fsid_falls_back_when_directory_cannot_be_opened(linux_platform):
+    with patch("app.utils.system.os.open", side_effect=OSError(errno.EACCES, "denied")), \
+            patch("app.utils.system.os.close") as close_mock:
+        result = SystemUtils._get_btrfs_fsid(Path("/data"))
+
+    assert result is None
+    close_mock.assert_not_called()
+
+
+@pytest.mark.parametrize(
+    ("fsid", "num_devices", "truncate"),
+    [
+        (bytes(16), 1, False),
+        (b"valid-fsid-value", 0, False),
+        (b"valid-fsid-value", 1, True),
+    ],
+)
+def test_get_btrfs_fsid_rejects_invalid_kernel_results(fsid, num_devices, truncate, linux_platform):
+    with patch("app.utils.system.os.open", return_value=42), \
+            patch("app.utils.system.fcntl.ioctl", side_effect=_fill_fs_info(fsid, num_devices, truncate)), \
+            patch("app.utils.system.os.close") as close_mock:
+        result = SystemUtils._get_btrfs_fsid(Path("/data"))
+
+    assert result is None
+    close_mock.assert_called_once_with(42)
+
+
+def test_get_btrfs_fsid_is_disabled_outside_linux():
+    with patch("app.utils.system.sys.platform", "darwin"), \
+            patch("app.utils.system.os.open") as open_mock:
+        result = SystemUtils._get_btrfs_fsid(Path("/data"))
+
+    assert result is None
+    open_mock.assert_not_called()
+
+
+def test_get_btrfs_fsid_is_disabled_on_unsupported_linux_architecture():
+    with patch("app.utils.system.sys.platform", "linux"), \
+            patch.object(SystemUtils, "is_x86_64", return_value=False), \
+            patch.object(SystemUtils, "is_aarch64", return_value=False), \
+            patch("app.utils.system.os.open") as open_mock:
+        result = SystemUtils._get_btrfs_fsid(Path("/data"))
+
+    assert result is None
+    open_mock.assert_not_called()
+
+
+def test_btrfs_fsid_dedup_setting_is_opt_in():
+    assert ConfigModel().BTRFS_FSID_DEDUP is False
+    assert ConfigModel(BTRFS_FSID_DEDUP="true").BTRFS_FSID_DEDUP is True
+
+
+def test_space_usage_default_path_does_not_read_fsid():
+    with tempfile.TemporaryDirectory() as tmp1, tempfile.TemporaryDirectory() as tmp2:
+        paths = [Path(tmp1), Path(tmp2)]
+        dev_by_path = {str(paths[0]): 38, str(paths[1]): 32}
+        with patch("app.utils.system.os.stat", side_effect=_fake_stat_with_devs(dev_by_path)), \
+                patch.object(SystemUtils, "_get_btrfs_fsid") as fsid_mock, \
+                patch.object(SystemUtils, "total_space", return_value=2.0), \
+                patch.object(SystemUtils, "free_space", return_value=1.0):
+            total, free = SystemUtils.space_usage(paths)
+
+    assert total == 4.0
+    assert free == 2.0
+    fsid_mock.assert_not_called()
+
+
+@pytest.mark.parametrize("system_platform", ["darwin", "win32"])
+def test_space_usage_opt_in_does_not_read_fsid_outside_linux(system_platform):
+    path = MagicMock()
+    path.exists.return_value = True
+    path.drive = "D:"
+    with patch("app.utils.system.sys.platform", system_platform), \
+            patch.object(SystemUtils, "is_x86_64") as x86_mock, \
+            patch.object(SystemUtils, "is_aarch64") as arm_mock, \
+            patch.object(SystemUtils, "_get_btrfs_fsid") as fsid_mock, \
+            patch.object(SystemUtils, "total_space", return_value=2.0), \
+            patch.object(SystemUtils, "free_space", return_value=1.0), \
+            patch("app.utils.system.os.stat", return_value=MagicMock(st_dev=38)):
+        total, free = SystemUtils.space_usage([path], btrfs_fsid_dedup=True)
+
+    assert total == 2.0
+    assert free == 1.0
+    x86_mock.assert_not_called()
+    arm_mock.assert_not_called()
+    fsid_mock.assert_not_called()
+
+
+def test_space_usage_opt_in_uses_original_behavior_on_unsupported_linux_architecture():
+    with tempfile.TemporaryDirectory() as tmp1, tempfile.TemporaryDirectory() as tmp2:
+        paths = [Path(tmp1), Path(tmp2)]
+        dev_by_path = {str(path): 38 for path in paths}
+        with patch("app.utils.system.sys.platform", "linux"), \
+                patch.object(SystemUtils, "is_x86_64", return_value=False), \
+                patch.object(SystemUtils, "is_aarch64", return_value=False), \
+                patch("app.utils.system.os.stat", side_effect=_fake_stat_with_devs(dev_by_path)), \
+                patch.object(SystemUtils, "_get_btrfs_fsid") as fsid_mock, \
+                patch.object(SystemUtils, "total_space", return_value=2.0), \
+                patch.object(SystemUtils, "free_space", return_value=1.0):
+            total, free = SystemUtils.space_usage(paths, btrfs_fsid_dedup=True)
+
+    assert total == 2.0
+    assert free == 1.0
+    fsid_mock.assert_not_called()
+
+
+@pytest.mark.parametrize(("is_x86_64", "is_aarch64"), [(True, False), (False, True)])
+def test_space_usage_merges_btrfs_subvolumes_with_same_fsid(is_x86_64, is_aarch64, monkeypatch):
+    monkeypatch.setattr("app.utils.system.sys.platform", "linux")
+    monkeypatch.setattr(SystemUtils, "is_x86_64", lambda: is_x86_64)
+    monkeypatch.setattr(SystemUtils, "is_aarch64", lambda: is_aarch64)
+    fsid = bytes.fromhex("88e3aff5fa2946d591a55977be984655")
+    with tempfile.TemporaryDirectory() as tmp1, tempfile.TemporaryDirectory() as tmp2:
+        paths = [Path(tmp1), Path(tmp2)]
+        dev_by_path = {str(paths[0]): 38, str(paths[1]): 32}
+        with patch("app.utils.system.os.stat", side_effect=_fake_stat_with_devs(dev_by_path)), \
+                patch.object(SystemUtils, "_get_btrfs_fsid", side_effect=[fsid, fsid]), \
+                patch.object(SystemUtils, "total_space", return_value=3.49 * 1024 ** 4), \
+                patch.object(SystemUtils, "free_space", return_value=1.94 * 1024 ** 4):
+            total, free = SystemUtils.space_usage(paths, btrfs_fsid_dedup=True)
+
+    assert total == 3.49 * 1024 ** 4
+    assert free == 1.94 * 1024 ** 4
+
+
+def test_space_usage_counts_different_btrfs_fsids_separately(linux_platform):
+    with tempfile.TemporaryDirectory() as tmp1, tempfile.TemporaryDirectory() as tmp2:
+        paths = [Path(tmp1), Path(tmp2)]
+        dev_by_path = {str(paths[0]): 38, str(paths[1]): 32}
+        with patch("app.utils.system.os.stat", side_effect=_fake_stat_with_devs(dev_by_path)), \
+                patch.object(SystemUtils, "_get_btrfs_fsid", side_effect=[b"a" * 16, b"b" * 16]), \
+                patch.object(SystemUtils, "total_space", return_value=2.0), \
+                patch.object(SystemUtils, "free_space", return_value=1.0):
+            total, free = SystemUtils.space_usage(paths, btrfs_fsid_dedup=True)
+
+    assert total == 4.0
+    assert free == 2.0
+
+
+def test_space_usage_falls_back_to_st_dev_when_fsid_is_unavailable(linux_platform):
+    with tempfile.TemporaryDirectory() as tmp1, tempfile.TemporaryDirectory() as tmp2:
+        paths = [Path(tmp1), Path(tmp2)]
+        dev_by_path = {str(paths[0]): 60, str(paths[1]): 60}
+        with patch("app.utils.system.os.stat", side_effect=_fake_stat_with_devs(dev_by_path)), \
+                patch.object(SystemUtils, "_get_btrfs_fsid", return_value=None), \
+                patch.object(SystemUtils, "total_space", return_value=2.0), \
+                patch.object(SystemUtils, "free_space", return_value=1.0):
+            total, free = SystemUtils.space_usage(paths, btrfs_fsid_dedup=True)
+
+    assert total == 2.0
+    assert free == 1.0
+
+
+@pytest.mark.parametrize("fsids", [(b"a" * 16, None), (None, b"a" * 16)])
+def test_space_usage_keeps_st_dev_dedup_when_fsid_availability_differs(fsids, linux_platform):
+    with tempfile.TemporaryDirectory() as tmp1, tempfile.TemporaryDirectory() as tmp2:
+        paths = [Path(tmp1), Path(tmp2)]
+        dev_by_path = {str(path): 38 for path in paths}
+        with patch("app.utils.system.os.stat", side_effect=_fake_stat_with_devs(dev_by_path)), \
+                patch.object(SystemUtils, "_get_btrfs_fsid", side_effect=fsids), \
+                patch.object(SystemUtils, "total_space", return_value=2.0), \
+                patch.object(SystemUtils, "free_space", return_value=1.0):
+            total, free = SystemUtils.space_usage(paths, btrfs_fsid_dedup=True)
+
+    assert total == 2.0
+    assert free == 1.0
+
+
+def test_space_usage_keeps_st_dev_dedup_when_fsid_is_consistent(linux_platform):
+    with tempfile.TemporaryDirectory() as tmp1, tempfile.TemporaryDirectory() as tmp2:
+        paths = [Path(tmp1), Path(tmp2)]
+        dev_by_path = {str(path): 38 for path in paths}
+        with patch("app.utils.system.os.stat", side_effect=_fake_stat_with_devs(dev_by_path)), \
+                patch.object(SystemUtils, "_get_btrfs_fsid", side_effect=[b"a" * 16, b"a" * 16]), \
+                patch.object(SystemUtils, "total_space", return_value=2.0), \
+                patch.object(SystemUtils, "free_space", return_value=1.0):
+            total, free = SystemUtils.space_usage(paths, btrfs_fsid_dedup=True)
+
+    assert total == 2.0
+    assert free == 1.0
+
+
+def test_space_usage_does_not_merge_different_st_devs_when_one_fsid_is_unavailable(linux_platform):
+    with tempfile.TemporaryDirectory() as tmp1, tempfile.TemporaryDirectory() as tmp2:
+        paths = [Path(tmp1), Path(tmp2)]
+        dev_by_path = {str(paths[0]): 38, str(paths[1]): 32}
+        with patch("app.utils.system.os.stat", side_effect=_fake_stat_with_devs(dev_by_path)), \
+                patch.object(SystemUtils, "_get_btrfs_fsid", side_effect=[b"a" * 16, None]), \
+                patch.object(SystemUtils, "total_space", return_value=2.0), \
+                patch.object(SystemUtils, "free_space", return_value=1.0):
+            total, free = SystemUtils.space_usage(paths, btrfs_fsid_dedup=True)
+
+    assert total == 4.0
+    assert free == 2.0
+
+
+def test_space_usage_counts_a_repeated_path_once():
+    with tempfile.TemporaryDirectory() as tmp:
+        path = Path(tmp)
+        with patch.object(SystemUtils, "_get_btrfs_fsid", return_value=None), \
+                patch.object(SystemUtils, "total_space", return_value=2.0), \
+                patch.object(SystemUtils, "free_space", return_value=1.0):
+            total, free = SystemUtils.space_usage([path, path])
+
+    assert total == 2.0
+    assert free == 1.0
+
+
+def test_space_usage_merges_consistent_fsid_observed_within_same_st_dev(linux_platform):
+    with tempfile.TemporaryDirectory() as tmp1, tempfile.TemporaryDirectory() as tmp2, \
+            tempfile.TemporaryDirectory() as tmp3:
+        paths = [Path(tmp1), Path(tmp2), Path(tmp3)]
+        dev_by_path = {str(paths[0]): 38, str(paths[1]): 38, str(paths[2]): 32}
+        with patch("app.utils.system.os.stat", side_effect=_fake_stat_with_devs(dev_by_path)), \
+                patch.object(SystemUtils, "_get_btrfs_fsid", side_effect=[None, b"a" * 16, b"a" * 16]), \
+                patch.object(SystemUtils, "total_space", return_value=2.0), \
+                patch.object(SystemUtils, "free_space", return_value=1.0):
+            total, free = SystemUtils.space_usage(paths, btrfs_fsid_dedup=True)
+
+    assert total == 2.0
+    assert free == 1.0
+
+
+def test_space_usage_transitive_merge_is_independent_of_path_order(linux_platform):
+    fsid = b"a" * 16
+    records = [("first", 38, None), ("bridge", 38, fsid), ("other", 32, fsid)]
+
+    for permutation in itertools.permutations(records):
+        paths = [MagicMock(name=name) for name, _, _ in permutation]
+        for path in paths:
+            path.exists.return_value = True
+        dev_by_path = {str(path): record[1] for path, record in zip(paths, permutation)}
+        fsid_by_path = {str(path): record[2] for path, record in zip(paths, permutation)}
+        with patch("app.utils.system.os.stat", side_effect=_fake_stat_with_devs(dev_by_path)), \
+                patch.object(SystemUtils, "_get_btrfs_fsid", side_effect=lambda path: fsid_by_path[str(path)]), \
+                patch.object(SystemUtils, "total_space", return_value=2.0), \
+                patch.object(SystemUtils, "free_space", return_value=1.0):
+            total, free = SystemUtils.space_usage(paths, btrfs_fsid_dedup=True)
+
+        assert total == 2.0
+        assert free == 1.0
+
+
+def test_space_usage_conflicting_fsids_do_not_bridge_independent_groups(linux_platform):
+    with tempfile.TemporaryDirectory() as tmp1, tempfile.TemporaryDirectory() as tmp2, \
+            tempfile.TemporaryDirectory() as tmp3, tempfile.TemporaryDirectory() as tmp4:
+        paths = [Path(tmp1), Path(tmp2), Path(tmp3), Path(tmp4)]
+        dev_by_path = {
+            str(paths[0]): 38,
+            str(paths[1]): 38,
+            str(paths[2]): 32,
+            str(paths[3]): 101,
+        }
+        with patch("app.utils.system.os.stat", side_effect=_fake_stat_with_devs(dev_by_path)), \
+                patch.object(SystemUtils, "_get_btrfs_fsid",
+                             side_effect=[b"a" * 16, b"b" * 16, b"a" * 16, b"b" * 16]), \
+                patch.object(SystemUtils, "total_space", return_value=2.0), \
+                patch.object(SystemUtils, "free_space", return_value=1.0):
+            total, free = SystemUtils.space_usage(paths, btrfs_fsid_dedup=True)
+
+    assert total == 6.0
+    assert free == 3.0
+
+
+def test_space_usage_uses_earliest_path_for_each_final_group(linux_platform):
+    fsid = b"a" * 16
+    paths = [MagicMock(name=name) for name in ("first", "same-dev", "same-fsid", "independent")]
+    for path in paths:
+        path.exists.return_value = True
+    dev_by_path = {str(paths[0]): 38, str(paths[1]): 38, str(paths[2]): 32, str(paths[3]): 101}
+    fsid_by_path = {str(paths[0]): None, str(paths[1]): fsid, str(paths[2]): fsid, str(paths[3]): None}
+
+    with patch("app.utils.system.os.stat", side_effect=_fake_stat_with_devs(dev_by_path)), \
+            patch.object(SystemUtils, "_get_btrfs_fsid", side_effect=lambda path: fsid_by_path[str(path)]), \
+            patch.object(SystemUtils, "total_space", side_effect=[2.0, 3.0]) as total_mock, \
+            patch.object(SystemUtils, "free_space", side_effect=[1.0, 1.5]) as free_mock:
+        total, free = SystemUtils.space_usage(paths, btrfs_fsid_dedup=True)
+
+    assert total == 5.0
+    assert free == 2.5
+    assert total_mock.call_args_list == [call(paths[0]), call(paths[3])]
+    assert free_mock.call_args_list == [call(paths[0]), call(paths[3])]
+
+
+def test_space_usage_keeps_windows_drive_behavior_without_fsid_lookup():
+    path = MagicMock()
+    path.exists.return_value = True
+    path.drive = "D:"
+    with patch("app.utils.system.os.name", "nt"), \
+            patch.object(SystemUtils, "_get_btrfs_fsid") as fsid_mock, \
+            patch.object(SystemUtils, "total_space", return_value=2.0), \
+            patch.object(SystemUtils, "free_space", return_value=1.0):
+        total, free = SystemUtils.space_usage([path])
+
+    assert total == 2.0
+    assert free == 1.0
+    fsid_mock.assert_not_called()
+
+
+def test_local_storage_usage_forwards_btrfs_fsid_setting():
+    from app.modules.filemanager.storages import local as local_storage_module
+
+    download_dir = MagicMock(download_path="/downloads")
+    library_dir = MagicMock(library_path="/library")
+    with patch.object(local_storage_module.settings, "BTRFS_FSID_DEDUP", True), \
+            patch.object(local_storage_module.DirectoryHelper, "get_local_download_dirs",
+                         return_value=[download_dir]), \
+            patch.object(local_storage_module.DirectoryHelper, "get_local_library_dirs",
+                         return_value=[library_dir]), \
+            patch.object(SystemUtils, "space_usage", return_value=(4.0, 2.0)) as usage_mock:
+        usage = object.__new__(local_storage_module.LocalStorage).usage()
+
+    assert usage.total == 4.0
+    assert usage.available == 2.0
+    usage_mock.assert_called_once_with(
+        [Path("/downloads"), Path("/library")],
+        btrfs_fsid_dedup=True,
+    )
+
+
+def test_dashboard_downloader_forwards_btrfs_fsid_setting():
+    from app.api.endpoints import dashboard as dashboard_module
+
+    download_dir = MagicMock(download_path="/downloads")
+    with patch.object(dashboard_module.settings, "BTRFS_FSID_DEDUP", True), \
+            patch.object(dashboard_module.DirectoryHelper, "get_local_download_dirs",
+                         return_value=[download_dir]), \
+            patch.object(SystemUtils, "space_usage", return_value=(4.0, 2.0)) as usage_mock, \
+            patch.object(dashboard_module.DashboardChain, "downloader_info", return_value=[]):
+        dashboard_module._build_downloader()
+
+    usage_mock.assert_called_once_with(
+        [Path("/downloads")],
+        btrfs_fsid_dedup=True,
+    )