From ef3d4d2ac59ca8d1e4574d61853a1e705b60923f Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E7=8F=AD=E6=89=AC?= Date: Thu, 8 Oct 2026 11:26:59 +0800 Subject: [PATCH 1/3] bump version --- src/modelscope_hub/version.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/modelscope_hub/version.py b/src/modelscope_hub/version.py index 62ee45e..a59471b 100644 --- a/src/modelscope_hub/version.py +++ b/src/modelscope_hub/version.py @@ -1,3 +1,3 @@ """Version information for modelscope_hub.""" -__version__ = "0.4.4+main" +__version__ = "0.4.6+main" From 8413a1496ddca8c0e0a59d49f2b9d96dd8ea80e6 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E7=8F=AD=E6=89=AC?= Date: Thu, 8 Oct 2026 15:53:14 +0800 Subject: [PATCH 2/3] fix cross-region download --- src/modelscope_hub/_download.py | 158 +++++++++++------ tests/test_download_inter_region.py | 251 ++++++++++++++++++++++++++++ 2 files changed, 357 insertions(+), 52 deletions(-) create mode 100644 tests/test_download_inter_region.py diff --git a/src/modelscope_hub/_download.py b/src/modelscope_hub/_download.py index 069bab1..bb5fab5 100644 --- a/src/modelscope_hub/_download.py +++ b/src/modelscope_hub/_download.py @@ -335,10 +335,15 @@ def __init__(self, legacy_client: LegacyClient, config: HubConfig) -> None: self._client = legacy_client self._config = config self._cached_region: str | None = None - # Cache resolved inter-region probe results to avoid redundant HEAD probes - # when downloading multiple files from the same repo in parallel. - self._inter_region_cache: dict[tuple[str, str], tuple[str | None, str]] = {} + # Cache only confirmed OSS-internal resolutions. A repository can contain + # Git metadata and LFS objects behind different download routes, so cache + # entries are scoped to repo/type/revision plus the redirect endpoint. + self._inter_region_cache: dict[tuple[str, str, str, str], tuple[str | None, str]] = {} + # Guard the cache and the per-route in-flight probes. The probe itself is + # deliberately performed outside this lock so distinct storage routes can + # resolve in parallel. self._inter_region_lock = threading.Lock() + self._inter_region_inflight: dict[tuple[str, str, str, str], threading.Event] = {} # ------------------------------------------------------------------ # User-agent & headers @@ -422,6 +427,24 @@ def _probe_redirect_url( pass return "" + @staticmethod + def _storage_route_key(redirect_url: str) -> str | None: + """Return a stable identity for the storage endpoint behind a redirect. + + Presigned query parameters and object paths are intentionally excluded: + they differ per file and request, whereas the endpoint hostname is the + stable route boundary that must isolate inter-region cache entries. + """ + try: + parsed = urlparse(redirect_url) + host = parsed.hostname + if not parsed.scheme or not host: + return None + port = f":{parsed.port}" if parsed.port is not None else "" + return f"{parsed.scheme.lower()}://{host.lower()}{port}" + except (TypeError, ValueError): + return None + def _get_inter_cloud_regions(self) -> list[str]: """Read the inter-cloud peer region list from the environment.""" from .constants import _env @@ -435,6 +458,7 @@ def _resolve_inter_region_headers( headers: dict[str, str], cookies: dict | None = None, peer_regions: list[str] | None = None, + initial_redirect_url: str | None = None, ) -> tuple[dict[str, str], str]: """Probe peer regions and return headers with the best region for OSS internal download. @@ -453,8 +477,12 @@ def _resolve_inter_region_headers( current_region = headers.get("x-aliyun-region-id", "").lower() - # Step 1: Probe with current (local) region - redirect_url = self._probe_redirect_url(url, headers, cookies) + # Step 1: Probe with current (local) region. Callers that already + # established the storage route pass the initial redirect to avoid an + # identical second HEAD request. + redirect_url = ( + self._probe_redirect_url(url, headers, cookies) if initial_redirect_url is None else initial_redirect_url + ) if self._is_oss_internal_url(redirect_url): logger.debug("Inter-region: local region already yields OSS internal URL, skipping.") return headers, "local" @@ -1127,57 +1155,83 @@ def _download_with_resume( download_headers = self._build_download_headers(user_agent) - # Inter-region acceleration: probe peer regions for OSS internal URL. - # Cache per (repo_id, repo_type) — all files in the same repo share one - # OSS bucket so one probe is sufficient for the entire download session. + # Inter-region acceleration: first discover the file's storage route, + # then cache only a confirmed internal resolution for that exact route. + # Git metadata may return 200 without a redirect while LFS objects use + # OSS; a metadata miss must therefore never suppress an LFS probe. # Progress bar prefix: "⚡ " = local OSS, "⇄ " = peer OSS, " " = CDN, "" = not configured source_prefix = "" peer_regions = self._get_inter_cloud_regions() if peer_regions: - cache_key = (repo_id, repo_type) - with self._inter_region_lock: - cached = self._inter_region_cache.get(cache_key) - if cached is None: - # First thread for this repo — do the probe under lock so - # other threads wait instead of issuing redundant HEAD reqs. - try: - probe_url = self._client.get_download_url( - repo_id, - repo_type, - file_path, - revision, - ) - cookies = None - if self._client.token: - cookies = {"m_session_id": self._client.token} - download_headers, source = self._resolve_inter_region_headers( - probe_url, - download_headers, - cookies, - peer_regions=peer_regions, - ) - resolved_region = download_headers.get("x-aliyun-region-id") - self._inter_region_cache[cache_key] = (resolved_region, source) - except Exception as exc: - logger.warning( - "Failed to resolve inter-region acceleration: %s. Falling back to default.", - exc, - ) - self._inter_region_cache[cache_key] = (None, "default") - cached = (None, "default") - - if cached is None: - cached = self._inter_region_cache.get(cache_key) - - if cached is not None: - resolved_region, source = cached - if resolved_region is not None: - download_headers["x-aliyun-region-id"] = resolved_region - source_prefix = { - "local": "\u26a1 ", - "peer": "\u21c4 ", - "default": " ", - }[source] + source = "default" + cached: tuple[str | None, str] | None = None + probe_event: threading.Event | None = None + try: + probe_url = self._client.get_download_url(repo_id, repo_type, file_path, revision) + cookies = {"m_session_id": self._client.token} if self._client.token else None + initial_redirect_url = self._probe_redirect_url(probe_url, download_headers, cookies) + route_key = self._storage_route_key(initial_redirect_url) + + if route_key is not None: + cache_key = (repo_id, repo_type, revision, route_key) + while cached is None: + is_probe_owner = False + with self._inter_region_lock: + cached = self._inter_region_cache.get(cache_key) + if cached is not None: + break + probe_event = self._inter_region_inflight.get(cache_key) + if probe_event is None: + probe_event = threading.Event() + self._inter_region_inflight[cache_key] = probe_event + is_probe_owner = True + + if not is_probe_owner: + assert probe_event is not None + probe_event.wait() + continue + + try: + download_headers, source = self._resolve_inter_region_headers( + probe_url, + download_headers, + cookies, + peer_regions=peer_regions, + initial_redirect_url=initial_redirect_url, + ) + resolved_region = download_headers.get("x-aliyun-region-id") + if source != "default": + with self._inter_region_lock: + self._inter_region_cache[cache_key] = (resolved_region, source) + cached = (resolved_region, source) + finally: + with self._inter_region_lock: + self._inter_region_inflight.pop(cache_key, None) + assert probe_event is not None + probe_event.set() + + # A non-internal result is intentionally not cached. The + # current file falls back now; a waiting file retries as + # the next single-flight owner instead of inheriting it. + if cached is None: + break + + if cached is not None: + resolved_region, source = cached + if resolved_region is not None: + download_headers["x-aliyun-region-id"] = resolved_region + except Exception as exc: + logger.warning( + "Failed to resolve inter-region acceleration: %s. Falling back to default.", + exc, + ) + source = "default" + + source_prefix = { + "local": "\u26a1 ", + "peer": "\u21c4 ", + "default": " ", + }[source] if use_parallel: assert file_size is not None # guaranteed by use_parallel above diff --git a/tests/test_download_inter_region.py b/tests/test_download_inter_region.py new file mode 100644 index 0000000..66879a1 --- /dev/null +++ b/tests/test_download_inter_region.py @@ -0,0 +1,251 @@ +"""Regression tests for inter-region download route isolation.""" + +from __future__ import annotations + +import threading +from concurrent.futures import ThreadPoolExecutor +from pathlib import Path +from urllib.parse import parse_qs, urlparse + +from modelscope_hub._download import DownloadManager +from modelscope_hub.config import HubConfig + +_REPO_ID = "wangxingjun778/xxx" + + +class _Response: + status_code = 200 + headers = {"Content-Length": "2"} + + @staticmethod + def iter_content(chunk_size: int): + del chunk_size + yield b"ok" + + +class _Client: + endpoint = "https://modelscope.cn" + token = None + + def __init__(self) -> None: + self.download_headers: dict[str, dict[str, str]] = {} + + @staticmethod + def get_download_url(repo_id: str, repo_type: str, file_path: str, revision: str) -> str: + return f"https://modelscope.cn/api/v1/{repo_type}s/{repo_id}/repo?revision={revision}&FilePath={file_path}" + + def download_stream( + self, + *, + repo_id: str, + repo_type: str, + file_path: str, + revision: str, + headers: dict[str, str], + ) -> _Response: + del repo_id, repo_type, revision + self.download_headers[file_path] = headers + return _Response() + + +def _manager(tmp_path: Path) -> tuple[DownloadManager, _Client]: + client = _Client() + return DownloadManager(client, HubConfig(cache_dir=tmp_path / "cache", token="")), client + + +def _file_path(url: str) -> str: + return parse_qs(urlparse(url).query)["FilePath"][0] + + +def _download(manager: DownloadManager, tmp_path: Path, file_path: str, revision: str = "master") -> None: + target = tmp_path / "downloads" / revision / file_path + target.parent.mkdir(parents=True, exist_ok=True) + manager._download_with_resume(_REPO_ID, "dataset", file_path, revision, target) + + +def _route_probe(peer_by_path: dict[str, str], calls: list[tuple[str, str]]): + def probe(url: str, headers: dict[str, str], cookies=None, timeout: float = 5.0) -> str: + del cookies, timeout + path = _file_path(url) + region = headers.get("x-aliyun-region-id", "") + calls.append((path, region)) + if path == ".gitattributes": + return "" + bucket, peer = peer_by_path[path].split("@") + if region == peer: + return f"https://{bucket}.oss-{peer}-internal.aliyuncs.com/object?signature=peer" + return f"https://{bucket}.oss-cn-hangzhou.aliyuncs.com/object?signature=public" + + return probe + + +def _configure_regions(monkeypatch) -> None: + monkeypatch.setenv("MODELSCOPE_DOWNLOAD_INTRA_CLOUD", "false") + monkeypatch.setenv("MODELSCOPE_DOWNLOAD_INTER_CLOUD_REGIONS", "peer-a,peer-b") + + +def test_metadata_default_route_does_not_block_lfs_peer_probe(tmp_path, monkeypatch): + """REG-01: serial metadata 200 must not populate an LFS default cache.""" + _configure_regions(monkeypatch) + manager, client = _manager(tmp_path) + calls: list[tuple[str, str]] = [] + monkeypatch.setattr(manager, "_probe_redirect_url", _route_probe({"large.tar": "bucket-a@peer-a"}, calls)) + + _download(manager, tmp_path, ".gitattributes") + _download(manager, tmp_path, "large.tar") + + assert client.download_headers["large.tar"]["x-aliyun-region-id"] == "peer-a" + assert ("large.tar", "peer-a") in calls + assert len(manager._inter_region_cache) == 1 + + +def test_reversed_file_order_keeps_lfs_internal_route(tmp_path, monkeypatch): + """REG-02: an earlier LFS resolution must be independent of later metadata.""" + _configure_regions(monkeypatch) + manager, client = _manager(tmp_path) + calls: list[tuple[str, str]] = [] + monkeypatch.setattr(manager, "_probe_redirect_url", _route_probe({"large.tar": "bucket-a@peer-a"}, calls)) + + _download(manager, tmp_path, "large.tar") + _download(manager, tmp_path, ".gitattributes") + + assert client.download_headers["large.tar"]["x-aliyun-region-id"] == "peer-a" + assert ".gitattributes" in client.download_headers + assert len(manager._inter_region_cache) == 1 + + +def test_same_storage_route_reuses_confirmed_internal_result(tmp_path, monkeypatch): + """REG-04: signed URLs differ per object but the same endpoint shares one probe.""" + _configure_regions(monkeypatch) + manager, client = _manager(tmp_path) + calls: list[tuple[str, str]] = [] + monkeypatch.setattr( + manager, + "_probe_redirect_url", + _route_probe({"one.tar": "bucket-a@peer-a", "two.tar": "bucket-a@peer-a"}, calls), + ) + + _download(manager, tmp_path, "one.tar") + _download(manager, tmp_path, "two.tar") + + assert client.download_headers["one.tar"]["x-aliyun-region-id"] == "peer-a" + assert client.download_headers["two.tar"]["x-aliyun-region-id"] == "peer-a" + assert calls.count(("one.tar", "peer-a")) == 1 + assert ("two.tar", "peer-a") not in calls + + +def test_different_storage_routes_do_not_share_peer_region(tmp_path, monkeypatch): + """REG-05: route/bucket isolation prevents a peer result crossing objects.""" + _configure_regions(monkeypatch) + manager, client = _manager(tmp_path) + calls: list[tuple[str, str]] = [] + monkeypatch.setattr( + manager, + "_probe_redirect_url", + _route_probe({"one.tar": "bucket-a@peer-a", "two.tar": "bucket-b@peer-b"}, calls), + ) + + _download(manager, tmp_path, "one.tar") + _download(manager, tmp_path, "two.tar") + + assert client.download_headers["one.tar"]["x-aliyun-region-id"] == "peer-a" + assert client.download_headers["two.tar"]["x-aliyun-region-id"] == "peer-b" + assert ("two.tar", "peer-b") in calls + assert len(manager._inter_region_cache) == 2 + + +def test_revision_is_part_of_storage_route_cache_scope(tmp_path, monkeypatch): + """REG-06: a new revision must resolve its own route before reuse.""" + _configure_regions(monkeypatch) + manager, client = _manager(tmp_path) + calls: list[tuple[str, str]] = [] + monkeypatch.setattr(manager, "_probe_redirect_url", _route_probe({"large.tar": "bucket-a@peer-a"}, calls)) + + _download(manager, tmp_path, "large.tar", revision="v1") + _download(manager, tmp_path, "large.tar", revision="v2") + + assert client.download_headers["large.tar"]["x-aliyun-region-id"] == "peer-a" + assert calls.count(("large.tar", "peer-a")) == 2 + assert len(manager._inter_region_cache) == 2 + + +def test_probe_failure_does_not_cache_default_result(tmp_path, monkeypatch): + """REG-07: a transient initial failure must not poison a later LFS request.""" + _configure_regions(monkeypatch) + manager, client = _manager(tmp_path) + attempts = 0 + + def probe(url: str, headers: dict[str, str], cookies=None, timeout: float = 5.0) -> str: + nonlocal attempts + del cookies, timeout + attempts += 1 + if attempts == 1: + raise RuntimeError("transient probe failure") + if headers.get("x-aliyun-region-id") == "peer-a": + return "https://bucket-a.oss-peer-a-internal.aliyuncs.com/object" + return "https://bucket-a.oss-cn-hangzhou.aliyuncs.com/object" + + monkeypatch.setattr(manager, "_probe_redirect_url", probe) + + _download(manager, tmp_path, "large.tar") + _download(manager, tmp_path, "large.tar", revision="retry") + + assert client.download_headers["large.tar"]["x-aliyun-region-id"] == "peer-a" + assert len(manager._inter_region_cache) == 1 + + +def test_peer_miss_is_not_negative_cached(tmp_path, monkeypatch): + """REG-08: after an all-peer miss, a later request must probe again.""" + _configure_regions(monkeypatch) + manager, client = _manager(tmp_path) + resolve_internal = False + + def probe(url: str, headers: dict[str, str], cookies=None, timeout: float = 5.0) -> str: + del url, cookies, timeout + if resolve_internal and headers.get("x-aliyun-region-id") == "peer-a": + return "https://bucket-a.oss-peer-a-internal.aliyuncs.com/object" + return "https://bucket-a.oss-cn-hangzhou.aliyuncs.com/object" + + monkeypatch.setattr(manager, "_probe_redirect_url", probe) + + _download(manager, tmp_path, "large.tar") + assert "x-aliyun-region-id" not in client.download_headers["large.tar"] + assert not manager._inter_region_cache + + resolve_internal = True + _download(manager, tmp_path, "large.tar", revision="retry") + + assert client.download_headers["large.tar"]["x-aliyun-region-id"] == "peer-a" + assert len(manager._inter_region_cache) == 1 + + +def test_parallel_same_route_uses_one_peer_probe(tmp_path, monkeypatch): + """REG-03/04: concurrent files share a route-specific in-flight probe.""" + _configure_regions(monkeypatch) + manager, client = _manager(tmp_path) + initial_barrier = threading.Barrier(2) + calls: list[tuple[str, str]] = [] + calls_lock = threading.Lock() + + def probe(url: str, headers: dict[str, str], cookies=None, timeout: float = 5.0) -> str: + del cookies, timeout + path = _file_path(url) + region = headers.get("x-aliyun-region-id", "") + if not region: + initial_barrier.wait(timeout=3) + with calls_lock: + calls.append((path, region)) + if region == "peer-a": + return "https://bucket-a.oss-peer-a-internal.aliyuncs.com/object" + return "https://bucket-a.oss-cn-hangzhou.aliyuncs.com/object" + + monkeypatch.setattr(manager, "_probe_redirect_url", probe) + + with ThreadPoolExecutor(max_workers=2) as executor: + futures = [executor.submit(_download, manager, tmp_path, name) for name in ("one.tar", "two.tar")] + for future in futures: + future.result() + + assert client.download_headers["one.tar"]["x-aliyun-region-id"] == "peer-a" + assert client.download_headers["two.tar"]["x-aliyun-region-id"] == "peer-a" + assert sum(region == "peer-a" for _, region in calls) == 1 From 3270b5f80a4563349aaff108b9588a28352bdfc7 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E7=8F=AD=E6=89=AC?= Date: Thu, 8 Oct 2026 15:54:42 +0800 Subject: [PATCH 3/3] remove ut --- tests/test_download_inter_region.py | 251 ---------------------------- 1 file changed, 251 deletions(-) delete mode 100644 tests/test_download_inter_region.py diff --git a/tests/test_download_inter_region.py b/tests/test_download_inter_region.py deleted file mode 100644 index 66879a1..0000000 --- a/tests/test_download_inter_region.py +++ /dev/null @@ -1,251 +0,0 @@ -"""Regression tests for inter-region download route isolation.""" - -from __future__ import annotations - -import threading -from concurrent.futures import ThreadPoolExecutor -from pathlib import Path -from urllib.parse import parse_qs, urlparse - -from modelscope_hub._download import DownloadManager -from modelscope_hub.config import HubConfig - -_REPO_ID = "wangxingjun778/xxx" - - -class _Response: - status_code = 200 - headers = {"Content-Length": "2"} - - @staticmethod - def iter_content(chunk_size: int): - del chunk_size - yield b"ok" - - -class _Client: - endpoint = "https://modelscope.cn" - token = None - - def __init__(self) -> None: - self.download_headers: dict[str, dict[str, str]] = {} - - @staticmethod - def get_download_url(repo_id: str, repo_type: str, file_path: str, revision: str) -> str: - return f"https://modelscope.cn/api/v1/{repo_type}s/{repo_id}/repo?revision={revision}&FilePath={file_path}" - - def download_stream( - self, - *, - repo_id: str, - repo_type: str, - file_path: str, - revision: str, - headers: dict[str, str], - ) -> _Response: - del repo_id, repo_type, revision - self.download_headers[file_path] = headers - return _Response() - - -def _manager(tmp_path: Path) -> tuple[DownloadManager, _Client]: - client = _Client() - return DownloadManager(client, HubConfig(cache_dir=tmp_path / "cache", token="")), client - - -def _file_path(url: str) -> str: - return parse_qs(urlparse(url).query)["FilePath"][0] - - -def _download(manager: DownloadManager, tmp_path: Path, file_path: str, revision: str = "master") -> None: - target = tmp_path / "downloads" / revision / file_path - target.parent.mkdir(parents=True, exist_ok=True) - manager._download_with_resume(_REPO_ID, "dataset", file_path, revision, target) - - -def _route_probe(peer_by_path: dict[str, str], calls: list[tuple[str, str]]): - def probe(url: str, headers: dict[str, str], cookies=None, timeout: float = 5.0) -> str: - del cookies, timeout - path = _file_path(url) - region = headers.get("x-aliyun-region-id", "") - calls.append((path, region)) - if path == ".gitattributes": - return "" - bucket, peer = peer_by_path[path].split("@") - if region == peer: - return f"https://{bucket}.oss-{peer}-internal.aliyuncs.com/object?signature=peer" - return f"https://{bucket}.oss-cn-hangzhou.aliyuncs.com/object?signature=public" - - return probe - - -def _configure_regions(monkeypatch) -> None: - monkeypatch.setenv("MODELSCOPE_DOWNLOAD_INTRA_CLOUD", "false") - monkeypatch.setenv("MODELSCOPE_DOWNLOAD_INTER_CLOUD_REGIONS", "peer-a,peer-b") - - -def test_metadata_default_route_does_not_block_lfs_peer_probe(tmp_path, monkeypatch): - """REG-01: serial metadata 200 must not populate an LFS default cache.""" - _configure_regions(monkeypatch) - manager, client = _manager(tmp_path) - calls: list[tuple[str, str]] = [] - monkeypatch.setattr(manager, "_probe_redirect_url", _route_probe({"large.tar": "bucket-a@peer-a"}, calls)) - - _download(manager, tmp_path, ".gitattributes") - _download(manager, tmp_path, "large.tar") - - assert client.download_headers["large.tar"]["x-aliyun-region-id"] == "peer-a" - assert ("large.tar", "peer-a") in calls - assert len(manager._inter_region_cache) == 1 - - -def test_reversed_file_order_keeps_lfs_internal_route(tmp_path, monkeypatch): - """REG-02: an earlier LFS resolution must be independent of later metadata.""" - _configure_regions(monkeypatch) - manager, client = _manager(tmp_path) - calls: list[tuple[str, str]] = [] - monkeypatch.setattr(manager, "_probe_redirect_url", _route_probe({"large.tar": "bucket-a@peer-a"}, calls)) - - _download(manager, tmp_path, "large.tar") - _download(manager, tmp_path, ".gitattributes") - - assert client.download_headers["large.tar"]["x-aliyun-region-id"] == "peer-a" - assert ".gitattributes" in client.download_headers - assert len(manager._inter_region_cache) == 1 - - -def test_same_storage_route_reuses_confirmed_internal_result(tmp_path, monkeypatch): - """REG-04: signed URLs differ per object but the same endpoint shares one probe.""" - _configure_regions(monkeypatch) - manager, client = _manager(tmp_path) - calls: list[tuple[str, str]] = [] - monkeypatch.setattr( - manager, - "_probe_redirect_url", - _route_probe({"one.tar": "bucket-a@peer-a", "two.tar": "bucket-a@peer-a"}, calls), - ) - - _download(manager, tmp_path, "one.tar") - _download(manager, tmp_path, "two.tar") - - assert client.download_headers["one.tar"]["x-aliyun-region-id"] == "peer-a" - assert client.download_headers["two.tar"]["x-aliyun-region-id"] == "peer-a" - assert calls.count(("one.tar", "peer-a")) == 1 - assert ("two.tar", "peer-a") not in calls - - -def test_different_storage_routes_do_not_share_peer_region(tmp_path, monkeypatch): - """REG-05: route/bucket isolation prevents a peer result crossing objects.""" - _configure_regions(monkeypatch) - manager, client = _manager(tmp_path) - calls: list[tuple[str, str]] = [] - monkeypatch.setattr( - manager, - "_probe_redirect_url", - _route_probe({"one.tar": "bucket-a@peer-a", "two.tar": "bucket-b@peer-b"}, calls), - ) - - _download(manager, tmp_path, "one.tar") - _download(manager, tmp_path, "two.tar") - - assert client.download_headers["one.tar"]["x-aliyun-region-id"] == "peer-a" - assert client.download_headers["two.tar"]["x-aliyun-region-id"] == "peer-b" - assert ("two.tar", "peer-b") in calls - assert len(manager._inter_region_cache) == 2 - - -def test_revision_is_part_of_storage_route_cache_scope(tmp_path, monkeypatch): - """REG-06: a new revision must resolve its own route before reuse.""" - _configure_regions(monkeypatch) - manager, client = _manager(tmp_path) - calls: list[tuple[str, str]] = [] - monkeypatch.setattr(manager, "_probe_redirect_url", _route_probe({"large.tar": "bucket-a@peer-a"}, calls)) - - _download(manager, tmp_path, "large.tar", revision="v1") - _download(manager, tmp_path, "large.tar", revision="v2") - - assert client.download_headers["large.tar"]["x-aliyun-region-id"] == "peer-a" - assert calls.count(("large.tar", "peer-a")) == 2 - assert len(manager._inter_region_cache) == 2 - - -def test_probe_failure_does_not_cache_default_result(tmp_path, monkeypatch): - """REG-07: a transient initial failure must not poison a later LFS request.""" - _configure_regions(monkeypatch) - manager, client = _manager(tmp_path) - attempts = 0 - - def probe(url: str, headers: dict[str, str], cookies=None, timeout: float = 5.0) -> str: - nonlocal attempts - del cookies, timeout - attempts += 1 - if attempts == 1: - raise RuntimeError("transient probe failure") - if headers.get("x-aliyun-region-id") == "peer-a": - return "https://bucket-a.oss-peer-a-internal.aliyuncs.com/object" - return "https://bucket-a.oss-cn-hangzhou.aliyuncs.com/object" - - monkeypatch.setattr(manager, "_probe_redirect_url", probe) - - _download(manager, tmp_path, "large.tar") - _download(manager, tmp_path, "large.tar", revision="retry") - - assert client.download_headers["large.tar"]["x-aliyun-region-id"] == "peer-a" - assert len(manager._inter_region_cache) == 1 - - -def test_peer_miss_is_not_negative_cached(tmp_path, monkeypatch): - """REG-08: after an all-peer miss, a later request must probe again.""" - _configure_regions(monkeypatch) - manager, client = _manager(tmp_path) - resolve_internal = False - - def probe(url: str, headers: dict[str, str], cookies=None, timeout: float = 5.0) -> str: - del url, cookies, timeout - if resolve_internal and headers.get("x-aliyun-region-id") == "peer-a": - return "https://bucket-a.oss-peer-a-internal.aliyuncs.com/object" - return "https://bucket-a.oss-cn-hangzhou.aliyuncs.com/object" - - monkeypatch.setattr(manager, "_probe_redirect_url", probe) - - _download(manager, tmp_path, "large.tar") - assert "x-aliyun-region-id" not in client.download_headers["large.tar"] - assert not manager._inter_region_cache - - resolve_internal = True - _download(manager, tmp_path, "large.tar", revision="retry") - - assert client.download_headers["large.tar"]["x-aliyun-region-id"] == "peer-a" - assert len(manager._inter_region_cache) == 1 - - -def test_parallel_same_route_uses_one_peer_probe(tmp_path, monkeypatch): - """REG-03/04: concurrent files share a route-specific in-flight probe.""" - _configure_regions(monkeypatch) - manager, client = _manager(tmp_path) - initial_barrier = threading.Barrier(2) - calls: list[tuple[str, str]] = [] - calls_lock = threading.Lock() - - def probe(url: str, headers: dict[str, str], cookies=None, timeout: float = 5.0) -> str: - del cookies, timeout - path = _file_path(url) - region = headers.get("x-aliyun-region-id", "") - if not region: - initial_barrier.wait(timeout=3) - with calls_lock: - calls.append((path, region)) - if region == "peer-a": - return "https://bucket-a.oss-peer-a-internal.aliyuncs.com/object" - return "https://bucket-a.oss-cn-hangzhou.aliyuncs.com/object" - - monkeypatch.setattr(manager, "_probe_redirect_url", probe) - - with ThreadPoolExecutor(max_workers=2) as executor: - futures = [executor.submit(_download, manager, tmp_path, name) for name in ("one.tar", "two.tar")] - for future in futures: - future.result() - - assert client.download_headers["one.tar"]["x-aliyun-region-id"] == "peer-a" - assert client.download_headers["two.tar"]["x-aliyun-region-id"] == "peer-a" - assert sum(region == "peer-a" for _, region in calls) == 1