nspasojevic commited on
Commit
f23b8ab
·
verified ·
1 Parent(s): 2e148a3

Package sizes computed on the Space (package-download background job: pages the local index, one HF tree listing per (repo, revision), cached; nginx /package_sizes.json -> 8020, complete:false while pending). Restart loads media adds 13-25.

Browse files
docker/hf-space/nginx.conf.template CHANGED
@@ -155,6 +155,21 @@ server {
155
  proxy_read_timeout 300s;
156
  }
157
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
158
  location /images {
159
  if ($request_method = OPTIONS) { add_header Content-Length 0 always; add_header Content-Type "text/plain" always; return 204; }
160
  rewrite ^/images(.*)$ /v3/images$1 break;
 
155
  proxy_read_timeout 300s;
156
  }
157
 
158
+ # Every indexed asset's download size (the Explorer's Size column): package-download keeps the table current
159
+ # (pipeline/package_download/sizes.py), so the dist's static package_sizes.json is never served. While
160
+ # package-download is not up, or fails, a 200 with complete:false: the Explorer asks again in 60 s, where
161
+ # a 5xx would end its polling for the session. A 404 (sizes switched off) passes through: no sizes.
162
+ location = /package_sizes.json {
163
+ if ($request_method = OPTIONS) { add_header Content-Length 0 always; add_header Content-Type "text/plain" always; return 204; }
164
+ proxy_pass http://127.0.0.1:8020;
165
+ proxy_intercept_errors on;
166
+ error_page 500 502 503 504 = @package_sizes_pending;
167
+ }
168
+ location @package_sizes_pending {
169
+ default_type application/json;
170
+ return 200 '{"index":"usdsearch-dev","complete":false,"generated_at":null,"basis":"package sizes are not available yet","total":0,"sizes":{}}';
171
+ }
172
+
173
  location /images {
174
  if ($request_method = OPTIONS) { add_header Content-Length 0 always; add_header Content-Type "text/plain" always; return 204; }
175
  rewrite ^/images(.*)$ /v3/images$1 break;
pipeline/package_download/__init__.py CHANGED
@@ -10,10 +10,14 @@ laid out exactly as on Hugging Face, contents at the archive root.
10
 
11
  Bytes come through the in-container hf-storage-api gRPC server (FileFolderService.ListStat + FileObjectService.
12
  ReadFromAddress), the same server deepsearch-api's own download already reads through: one HF token, one cache,
13
- one access boundary (only catalogued folders resolve). This process holds no HF token.
14
 
15
  ``scope=dependencies``, and any URL that is not inside a catalogued package, is forwarded unchanged to
16
  deepsearch-api's ``/v3/download/asset`` — the old asset + dependencies bundle.
17
 
 
 
 
 
18
  Design and decision log: docs/plans/hf-space-package-download.md.
19
  """
 
10
 
11
  Bytes come through the in-container hf-storage-api gRPC server (FileFolderService.ListStat + FileObjectService.
12
  ReadFromAddress), the same server deepsearch-api's own download already reads through: one HF token, one cache,
13
+ one access boundary (only catalogued folders resolve). Downloads use no HF token of this process.
14
 
15
  ``scope=dependencies``, and any URL that is not inside a catalogued package, is forwarded unchanged to
16
  deepsearch-api's ``/v3/download/asset`` — the old asset + dependencies bundle.
17
 
18
+ ``GET /package_sizes.json`` is every indexed asset's download size for the Explorer's Size column: a background
19
+ task pages this Space's index and sums each package folder from one HF tree listing per (repo, revision), with the
20
+ read-only corpus token (``sizes.py``).
21
+
22
  Design and decision log: docs/plans/hf-space-package-download.md.
23
  """
pipeline/package_download/__main__.py CHANGED
@@ -10,6 +10,7 @@ import uvicorn
10
  from .app import create_app
11
  from .catalog import Catalog
12
  from .settings import Settings
 
13
 
14
 
15
  def main() -> int:
@@ -27,7 +28,12 @@ def main() -> int:
27
  "no hf-storage-api catalog at %s: forwarding every download to deepsearch-api", settings.catalog_path
28
  )
29
  catalog = Catalog({})
30
- uvicorn.run(create_app(settings, catalog=catalog), host=settings.host, port=settings.port, log_config=None)
 
 
 
 
 
31
  return 0
32
 
33
 
 
10
  from .app import create_app
11
  from .catalog import Catalog
12
  from .settings import Settings
13
+ from .sizes import PackageSizes, SizesSettings
14
 
15
 
16
  def main() -> int:
 
28
  "no hf-storage-api catalog at %s: forwarding every download to deepsearch-api", settings.catalog_path
29
  )
30
  catalog = Catalog({})
31
+ # /package_sizes.json: every indexed asset's download size, kept current in the background (sizes.py).
32
+ sizes_settings = SizesSettings.from_env()
33
+ sizes = PackageSizes(sizes_settings) if sizes_settings.enabled else None
34
+ uvicorn.run(
35
+ create_app(settings, catalog=catalog, sizes=sizes), host=settings.host, port=settings.port, log_config=None
36
+ )
37
  return 0
38
 
39
 
pipeline/package_download/app.py CHANGED
@@ -1,4 +1,5 @@
1
- """HTTP surface: ``GET /download/asset`` (also ``/v3/download/asset``), ``/health``, ``/metrics``.
 
2
 
3
  Drop-in for usdsearch's download route (see the package docstring); nginx proxies ``/download`` here.
4
  """
@@ -24,6 +25,7 @@ from starlette.background import BackgroundTask
24
  from . import bundle, telemetry
25
  from .catalog import Catalog
26
  from .settings import Settings
 
27
 
28
  logger = logging.getLogger("package_download")
29
 
@@ -181,6 +183,7 @@ def create_app(
181
  source: Any = None,
182
  url_mapping: Optional[UrlMapping] = None,
183
  fallback_transport: Optional[httpx.AsyncBaseTransport] = None,
 
184
  ) -> FastAPI:
185
  settings = settings or Settings.from_env()
186
  to_internal, to_public, to_public_url = url_mapping or _default_url_mapping()
@@ -215,9 +218,13 @@ def create_app(
215
  "package-download: %d catalogued package(s), storage API %s, fallback %s",
216
  len(catalog), settings.storage_api_endpoint, settings.fallback_url,
217
  )
 
 
218
  try:
219
  yield
220
  finally:
 
 
221
  await app.state.fallback.aclose()
222
  if owned_source:
223
  await source.close()
@@ -248,6 +255,15 @@ def create_app(
248
  async def prometheus():
249
  return metrics.render()
250
 
 
 
 
 
 
 
 
 
 
251
  async def forward(request: Request) -> Response:
252
  """Hand the request to deepsearch-api's own /v3/download/asset, unchanged."""
253
  params = [(k, v) for k, v in request.query_params.multi_items() if k != "scope"]
 
1
+ """HTTP surface: ``GET /download/asset`` (also ``/v3/download/asset``), ``/package_sizes.json``, ``/health``,
2
+ ``/metrics``.
3
 
4
  Drop-in for usdsearch's download route (see the package docstring); nginx proxies ``/download`` here.
5
  """
 
25
  from . import bundle, telemetry
26
  from .catalog import Catalog
27
  from .settings import Settings
28
+ from .sizes import PackageSizes
29
 
30
  logger = logging.getLogger("package_download")
31
 
 
183
  source: Any = None,
184
  url_mapping: Optional[UrlMapping] = None,
185
  fallback_transport: Optional[httpx.AsyncBaseTransport] = None,
186
+ sizes: Optional[PackageSizes] = None,
187
  ) -> FastAPI:
188
  settings = settings or Settings.from_env()
189
  to_internal, to_public, to_public_url = url_mapping or _default_url_mapping()
 
218
  "package-download: %d catalogued package(s), storage API %s, fallback %s",
219
  len(catalog), settings.storage_api_endpoint, settings.fallback_url,
220
  )
221
+ if sizes is not None:
222
+ sizes.start()
223
  try:
224
  yield
225
  finally:
226
+ if sizes is not None:
227
+ await sizes.stop()
228
  await app.state.fallback.aclose()
229
  if owned_source:
230
  await source.close()
 
255
  async def prometheus():
256
  return metrics.render()
257
 
258
+ if sizes is not None:
259
+ @app.get("/package_sizes.json")
260
+ async def package_sizes(request: Request):
261
+ body, encoding = sizes.body(request.headers.get("accept-encoding", ""))
262
+ headers = {"Cache-Control": "no-cache", "Vary": "Accept-Encoding"}
263
+ if encoding:
264
+ headers["Content-Encoding"] = encoding
265
+ return Response(body, media_type="application/json", headers=headers)
266
+
267
  async def forward(request: Request) -> Response:
268
  """Hand the request to deepsearch-api's own /v3/download/asset, unchanged."""
269
  params = [(k, v) for k, v in request.query_params.multi_items() if k != "scope"]
pipeline/package_download/sizes.py ADDED
@@ -0,0 +1,458 @@
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
+ """``GET /package_sizes.json``: every indexed asset's download size, for the Explorer's Size column.
2
+
3
+ The Explorer (SimReady Central's ``utils/packageSizes.js``) reads ONE table and matches each hit's
4
+ ``source.base_key`` exactly (``https://huggingface.co/datasets/<repo>/blob/<revision>/<path>``). An asset's size is
5
+ its package folder at its pinned revision: every file under the asset's top-level folder (the whole repo for a
6
+ file at the repo root), summed from one recursive HF tree listing per (repo, revision) — the definition of the
7
+ one-off table of 2026-10-16 (hf-space-work/package-sizes-1016/full_sizes.py), reproduced byte for byte.
8
+
9
+ A background task keeps the table current. At start, and every ``interval`` seconds, it pages this Space's own
10
+ index (deepsearch-api's query-less search, ``page_size`` hits a page, by cursor) for every base_key, lists each
11
+ (repo, revision) it has no sums for, and caches those sums on disk: a new asset or a re-pin costs one listing for
12
+ its new revision only. A pinned revision (a sha) never changes; a branch name is listed again after
13
+ ``relist_after``. HF requests are paced (at most one per ``hf_interval``) and back off on 429 / 5xx. A listing
14
+ that still fails keeps the last table: its group stays pending (``complete: false``, which the Explorer polls
15
+ again every 60 s) until a later pass lists it. A repo HF refuses outright (401/403/404: gone, gated, private)
16
+ settles unsized and is asked again after ``relist_after``.
17
+ """
18
+
19
+ from __future__ import annotations
20
+
21
+ import asyncio
22
+ import gzip
23
+ import hashlib
24
+ import json
25
+ import logging
26
+ import os
27
+ import time
28
+ from collections import Counter, defaultdict
29
+ from dataclasses import dataclass, fields
30
+ from datetime import datetime, timezone
31
+ from email.utils import parsedate_to_datetime
32
+ from pathlib import Path
33
+ from typing import Awaitable, Callable, Dict, Mapping, Optional, Tuple
34
+ from urllib.parse import quote, unquote
35
+
36
+ import httpx
37
+
38
+ from .. import hf_addressing
39
+
40
+ logger = logging.getLogger("package_download.sizes")
41
+
42
+ PREFIX = "PACKAGE_SIZES_"
43
+ BASIS = (
44
+ "each asset's package folder (every file under its top-level folder) at its pinned revision, "
45
+ "from one HF tree listing per (repo, revision)"
46
+ )
47
+
48
+
49
+ @dataclass(frozen=True)
50
+ class SizesSettings:
51
+ enabled: bool = True
52
+ # The Explorer's index key: /config.js's only server ("usdsearch-dev", nginx.conf.template).
53
+ index: str = "usdsearch-dev"
54
+ # deepsearch-api's v3 search (nginx's /search_hybrid), loopback.
55
+ search_url: str = "http://127.0.0.1:8000/v3/deepsearch/search"
56
+ page_size: int = 500
57
+ # The service-account Basic tag of the Explorer's own background probes (SimReady Central's utils/svcAuth.js):
58
+ # not access control, it keeps these walks out of the usage telemetry's user searches. Empty user: no tag.
59
+ search_user: str = "simready-central-svc-user"
60
+ search_password: str = "simready-central-svc-pass"
61
+ # Seconds between passes; after a pass that left groups pending; after a failed index walk.
62
+ interval: float = 900.0
63
+ pending_interval: float = 300.0
64
+ retry_interval: float = 30.0
65
+ hf_endpoint: str = "https://huggingface.co"
66
+ # At most one HF request per this many seconds, across every listing.
67
+ hf_interval: float = 1.0
68
+ # Tries per HF request (429, 5xx, a dropped connection); waits follow Retry-After, else 30 s doubling,
69
+ # at most hf_max_backoff. A request that still fails ends the pass's listings.
70
+ hf_attempts: int = 4
71
+ hf_max_backoff: float = 600.0
72
+ # Empty: HF_STORAGE_API_TOKEN (the Space's read-only corpus token), else HF_TOKEN, else anonymous.
73
+ hf_token: str = ""
74
+ timeout: float = 60.0
75
+ relist_after: float = 6 * 3600.0
76
+ # Outside .build/restore/, which the index restore empties at every start.
77
+ cache_dir: Path = Path("/app/hf-space-pipeline/.build/package_sizes")
78
+
79
+ @classmethod
80
+ def from_env(cls, env: Optional[Mapping[str, str]] = None) -> "SizesSettings":
81
+ env = os.environ if env is None else env
82
+ values = {}
83
+ for field in fields(cls):
84
+ raw = env.get(PREFIX + field.name.upper())
85
+ if raw is None or raw == "":
86
+ continue
87
+ default = field.default
88
+ if isinstance(default, bool):
89
+ values[field.name] = raw.strip().lower() in ("1", "true", "yes", "on")
90
+ elif isinstance(default, int):
91
+ values[field.name] = int(raw)
92
+ elif isinstance(default, float):
93
+ values[field.name] = float(raw)
94
+ elif isinstance(default, Path):
95
+ values[field.name] = Path(raw)
96
+ else:
97
+ values[field.name] = raw
98
+ if not values.get("hf_token"):
99
+ values["hf_token"] = (env.get("HF_STORAGE_API_TOKEN") or env.get("HF_TOKEN") or "").strip()
100
+ return cls(**values)
101
+
102
+
103
+ @dataclass(frozen=True, order=True)
104
+ class Group:
105
+ """One (repo, revision): everything one recursive tree listing answers."""
106
+
107
+ kind: str # the API's path segment: datasets | models | spaces
108
+ repo_id: str
109
+ revision: str
110
+
111
+ def __str__(self) -> str:
112
+ return f"{self.kind}/{self.repo_id}@{self.revision[:8]}"
113
+
114
+ @property
115
+ def cache_name(self) -> str:
116
+ return hashlib.sha256(f"{self.kind}/{self.repo_id}@{self.revision}".encode()).hexdigest()[:32] + ".json"
117
+
118
+
119
+ @dataclass
120
+ class Sums:
121
+ whole: int
122
+ folders: Dict[str, int]
123
+ files: int
124
+ listed_at: float
125
+
126
+ def size_of(self, path: str) -> int:
127
+ # full_sizes.py: the top-level folder's files, or the whole repo for a file at the root.
128
+ return self.folders.get(path.split("/", 1)[0], 0) if "/" in path else self.whole
129
+
130
+
131
+ def parse_key(key: str, endpoint: str = "https://huggingface.co") -> Optional[Tuple[Group, str]]:
132
+ """``<endpoint>/[datasets/|spaces/]<org>/<name>/blob/<revision>/<path>`` → (group, repo path), else None."""
133
+ prefix = endpoint.rstrip("/") + "/"
134
+ if not isinstance(key, str) or not key.startswith(prefix):
135
+ return None
136
+ parts = key[len(prefix):].split("?", 1)[0].split("/")
137
+ kind = parts.pop(0) if parts and parts[0] in ("datasets", "spaces") else "models"
138
+ if len(parts) < 5 or parts[2] != "blob" or not all(parts[:2]):
139
+ return None
140
+ revision, path = unquote(parts[3]), unquote("/".join(parts[4:]))
141
+ if not revision or not path:
142
+ return None
143
+ return Group(kind, f"{parts[0]}/{parts[1]}", revision), path
144
+
145
+
146
+ class IndexUnavailable(Exception):
147
+ pass
148
+
149
+
150
+ class HFRefused(Exception):
151
+ """HF answered 401/403/404: the repo or revision is gone, gated or private to this token."""
152
+
153
+
154
+ class HFUnavailable(Exception):
155
+ """HF kept answering 429 / 5xx (or the connection kept failing) after every retry."""
156
+
157
+
158
+ def _retry_after(response: httpx.Response) -> Optional[float]:
159
+ raw = response.headers.get("retry-after")
160
+ if not raw:
161
+ return None
162
+ try:
163
+ return max(0.0, float(raw))
164
+ except ValueError:
165
+ pass
166
+ try:
167
+ return max(0.0, (parsedate_to_datetime(raw) - datetime.now(timezone.utc)).total_seconds())
168
+ except (TypeError, ValueError):
169
+ return None
170
+
171
+
172
+ class PackageSizes:
173
+ """The table and the task that keeps it current. ``body()`` is what the route sends."""
174
+
175
+ def __init__(
176
+ self,
177
+ settings: Optional[SizesSettings] = None,
178
+ *,
179
+ search_transport: Optional[httpx.AsyncBaseTransport] = None,
180
+ hf_transport: Optional[httpx.AsyncBaseTransport] = None,
181
+ clock: Callable[[], float] = time.monotonic,
182
+ wall: Callable[[], float] = time.time,
183
+ sleep: Callable[[float], Awaitable[None]] = asyncio.sleep,
184
+ ):
185
+ self.settings = settings or SizesSettings.from_env()
186
+ self._search_transport, self._hf_transport = search_transport, hf_transport
187
+ self._clock, self._wall, self._sleep = clock, wall, sleep
188
+ self._keys: Dict[str, Optional[Tuple[Group, str]]] = {}
189
+ self._walked = False
190
+ self._sums: Dict[Group, Sums] = {}
191
+ self._refused: Dict[Group, float] = {} # group -> wall time it may be asked again
192
+ self._next_hf = 0.0
193
+ self._task: Optional[asyncio.Task] = None
194
+ self._search: Optional[httpx.AsyncClient] = None
195
+ self._hf: Optional[httpx.AsyncClient] = None
196
+ self._body = self._gz = b""
197
+ self.sized = 0
198
+ self._rebuild()
199
+
200
+ # --- serving ---------------------------------------------------------------------------------------------
201
+
202
+ def body(self, accept_encoding: str = "") -> Tuple[bytes, Optional[str]]:
203
+ """The table's JSON, gzipped for a client that takes gzip (nginx doesn't compress proxied responses)."""
204
+ if "gzip" in (accept_encoding or "").lower():
205
+ return self._gz, "gzip"
206
+ return self._body, None
207
+
208
+ @property
209
+ def complete(self) -> bool:
210
+ return self._walked and not self._pending()
211
+
212
+ def _pending(self) -> set:
213
+ return {v[0] for v in self._keys.values() if v and v[0] not in self._sums and v[0] not in self._refused}
214
+
215
+ def _rebuild(self) -> None:
216
+ sizes = {}
217
+ for key, parsed in self._keys.items():
218
+ if parsed and parsed[0] in self._sums:
219
+ n = self._sums[parsed[0]].size_of(parsed[1])
220
+ if n > 0:
221
+ sizes[key] = n
222
+ doc = {
223
+ "index": self.settings.index,
224
+ "complete": self.complete,
225
+ "generated_at": datetime.fromtimestamp(self._wall(), timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ"),
226
+ "basis": BASIS,
227
+ "total": len(self._keys),
228
+ "sizes": dict(sorted(sizes.items())),
229
+ }
230
+ self.sized = len(sizes)
231
+ self._body = json.dumps(doc, separators=(",", ":")).encode()
232
+ self._gz = gzip.compress(self._body, compresslevel=6, mtime=0)
233
+
234
+ # --- the task --------------------------------------------------------------------------------------------
235
+
236
+ def start(self) -> None:
237
+ self._task = asyncio.get_running_loop().create_task(self.run(), name="package-sizes")
238
+
239
+ async def stop(self) -> None:
240
+ if self._task is not None:
241
+ self._task.cancel()
242
+ await asyncio.gather(self._task, return_exceptions=True)
243
+ self._task = None
244
+ await self.close()
245
+
246
+ async def close(self) -> None:
247
+ for client in (self._search, self._hf):
248
+ if client is not None:
249
+ await client.aclose()
250
+ self._search = self._hf = None
251
+
252
+ async def run(self) -> None:
253
+ self.load_cache()
254
+ while True:
255
+ try:
256
+ delay = await self.refresh()
257
+ except asyncio.CancelledError:
258
+ raise
259
+ except Exception: # noqa: BLE001 — the table stays as it was; try again later
260
+ logger.exception("package sizes: pass failed")
261
+ delay = self.settings.retry_interval
262
+ await self._sleep(delay)
263
+
264
+ def _clients(self) -> Tuple[httpx.AsyncClient, httpx.AsyncClient]:
265
+ s = self.settings
266
+ if self._search is None:
267
+ auth = (s.search_user, s.search_password) if s.search_user else None
268
+ self._search = httpx.AsyncClient(
269
+ timeout=s.timeout, transport=self._search_transport, trust_env=False, auth=auth
270
+ )
271
+ if self._hf is None:
272
+ headers = {"Authorization": f"Bearer {s.hf_token}"} if s.hf_token else {}
273
+ self._hf = httpx.AsyncClient(
274
+ timeout=s.timeout, transport=self._hf_transport, headers=headers, follow_redirects=True
275
+ )
276
+ return self._search, self._hf
277
+
278
+ async def refresh(self) -> float:
279
+ """One pass: walk the index, list what is missing (biggest group first). Returns the delay to the next."""
280
+ s = self.settings
281
+ try:
282
+ keys = await self.walk_index()
283
+ except IndexUnavailable as exc:
284
+ logger.info("package sizes: index not readable yet (%s); next try in %.0f s", exc, s.retry_interval)
285
+ return s.retry_interval
286
+ self._keys, self._walked = keys, True
287
+ now = self._wall()
288
+ self._refused = {g: t for g, t in self._refused.items() if t > now}
289
+ counts = Counter(v[0] for v in keys.values() if v)
290
+ stale = {
291
+ g for g, sums in self._sums.items()
292
+ if not hf_addressing.looks_like_sha(g.revision) and now - sums.listed_at >= s.relist_after
293
+ }
294
+ todo = sorted(
295
+ (g for g in counts if (g not in self._sums or g in stale) and g not in self._refused),
296
+ key=lambda g: (-counts[g], g),
297
+ )
298
+ self._rebuild()
299
+ for group in todo:
300
+ try:
301
+ sums, pages = await self.list_group(group)
302
+ except HFRefused as exc:
303
+ logger.warning("package sizes: %s refused (%s); %d asset(s) unsized", group, exc, counts[group])
304
+ self._refused[group] = now + s.relist_after
305
+ except HFUnavailable as exc:
306
+ logger.warning("package sizes: listing %s failed (%s); the rest waits for the next pass", group, exc)
307
+ break
308
+ else:
309
+ self._sums[group] = sums
310
+ self._write_cache(group, sums)
311
+ logger.info(
312
+ "package sizes: listed %s: %d asset(s), %d file(s) in %d page(s)", group, counts[group], sums.files,
313
+ pages,
314
+ )
315
+ self._rebuild()
316
+ self._rebuild()
317
+ pending = self._pending()
318
+ logger.info(
319
+ "package sizes: %d asset(s), %d sized, %d group(s), %d pending%s", len(keys), self.sized, len(counts),
320
+ len(pending), "" if pending else " (complete)",
321
+ )
322
+ return s.pending_interval if pending else s.interval
323
+
324
+ async def walk_index(self) -> Dict[str, Optional[Tuple[Group, str]]]:
325
+ """Every base_key in the index (deepsearch-api's query-less search, by cursor)."""
326
+ search, _ = self._clients()
327
+ s = self.settings
328
+ keys: Dict[str, Optional[Tuple[Group, str]]] = {}
329
+ cursor = None
330
+ for _ in range(10_000):
331
+ body = {"limit": s.page_size}
332
+ if cursor:
333
+ body["cursor"] = cursor
334
+ try:
335
+ r = await search.post(s.search_url, json=body)
336
+ r.raise_for_status()
337
+ doc = r.json()
338
+ except (httpx.HTTPError, ValueError) as exc:
339
+ raise IndexUnavailable(f"{type(exc).__name__}: {exc}") from exc
340
+ hits = doc.get("hits") or []
341
+ for hit in hits:
342
+ key = (hit.get("source") or {}).get("base_key")
343
+ if isinstance(key, str) and key:
344
+ keys[key] = parse_key(key, s.hf_endpoint)
345
+ if not doc.get("has_more"):
346
+ return keys
347
+ cursor = doc.get("next_cursor")
348
+ if not hits or not cursor: # more to come but no way to it: never a complete table from a part
349
+ raise IndexUnavailable("has_more without hits or a cursor")
350
+ raise IndexUnavailable("the walk did not end")
351
+
352
+ async def list_group(self, group: Group) -> Tuple[Sums, int]:
353
+ """One recursive tree listing (every page): the whole repo's bytes and each top-level folder's."""
354
+ url = (
355
+ f"{self.settings.hf_endpoint.rstrip('/')}/api/{group.kind}/{group.repo_id}/tree/"
356
+ f"{quote(group.revision, safe='')}"
357
+ )
358
+ params: Optional[dict] = {"recursive": "true", "expand": "false"}
359
+ whole, files, pages = 0, 0, 0
360
+ folders: Dict[str, int] = defaultdict(int)
361
+ while url:
362
+ r = await self._hf_get(url, params)
363
+ pages += 1
364
+ try:
365
+ items = r.json()
366
+ except ValueError as exc:
367
+ raise HFUnavailable(f"page {pages} is not JSON") from exc
368
+ if not isinstance(items, list):
369
+ raise HFUnavailable(f"page {pages} is not a list")
370
+ for item in items:
371
+ if not isinstance(item, dict) or item.get("type") != "file":
372
+ continue
373
+ size, path = item.get("size"), item.get("path")
374
+ if isinstance(size, bool) or not isinstance(size, int) or not isinstance(path, str):
375
+ raise HFUnavailable(f"page {pages} has a file without a size or path")
376
+ whole += size
377
+ files += 1
378
+ if "/" in path:
379
+ folders[path.split("/", 1)[0]] += size
380
+ url = r.links.get("next", {}).get("url")
381
+ params = None # the next page's URL carries its own query
382
+ return Sums(whole, dict(folders), files, self._wall()), pages
383
+
384
+ async def _hf_get(self, url: str, params: Optional[dict]) -> httpx.Response:
385
+ _, hf = self._clients()
386
+ s = self.settings
387
+ problem = ""
388
+ for attempt in range(max(1, s.hf_attempts)):
389
+ await self._pace()
390
+ wait = None
391
+ try:
392
+ r = await hf.get(url, params=params)
393
+ except httpx.HTTPError as exc:
394
+ problem = type(exc).__name__
395
+ else:
396
+ if r.status_code == 200:
397
+ return r
398
+ if r.status_code in (401, 403, 404):
399
+ raise HFRefused(f"HTTP {r.status_code}")
400
+ problem = f"HTTP {r.status_code}"
401
+ if r.status_code != 429 and r.status_code < 500:
402
+ raise HFUnavailable(problem)
403
+ wait = _retry_after(r)
404
+ if attempt + 1 >= s.hf_attempts:
405
+ break
406
+ if wait is None:
407
+ wait = 30.0 * 2 ** attempt
408
+ if wait > s.hf_max_backoff:
409
+ raise HFUnavailable(f"{problem}, Retry-After {wait:.0f} s")
410
+ logger.info("package sizes: HF %s; waiting %.0f s", problem, wait)
411
+ # The pace holds every later request back as well.
412
+ self._next_hf = max(self._next_hf, self._clock() + wait)
413
+ raise HFUnavailable(problem or "no answer")
414
+
415
+ async def _pace(self) -> None:
416
+ wait = self._next_hf - self._clock()
417
+ if wait > 0:
418
+ await self._sleep(wait)
419
+ self._next_hf = max(self._next_hf, self._clock()) + self.settings.hf_interval
420
+
421
+ # --- the on-disk cache: one file per group -----------------------------------------------------------------
422
+
423
+ def _groups_dir(self) -> Path:
424
+ return Path(self.settings.cache_dir) / "groups"
425
+
426
+ def load_cache(self) -> int:
427
+ loaded = 0
428
+ folder = self._groups_dir()
429
+ if not folder.is_dir():
430
+ return 0
431
+ for path in sorted(folder.glob("*.json")):
432
+ try:
433
+ doc = json.loads(path.read_text())
434
+ group = Group(doc["kind"], doc["repo_id"], doc["revision"])
435
+ if path.name != group.cache_name:
436
+ continue
437
+ self._sums[group] = Sums(
438
+ int(doc["whole"]), {str(k): int(v) for k, v in doc["folders"].items()}, int(doc["files"]),
439
+ float(doc["listed_at"]),
440
+ )
441
+ loaded += 1
442
+ except (OSError, ValueError, KeyError, TypeError, AttributeError) as exc:
443
+ logger.warning("package sizes: ignoring cache file %s (%s)", path.name, exc)
444
+ if loaded:
445
+ logger.info("package sizes: %d group(s) from %s", loaded, folder)
446
+ return loaded
447
+
448
+ def _write_cache(self, group: Group, sums: Sums) -> None:
449
+ folder = self._groups_dir()
450
+ try:
451
+ folder.mkdir(parents=True, exist_ok=True)
452
+ doc = {"kind": group.kind, "repo_id": group.repo_id, "revision": group.revision, "whole": sums.whole,
453
+ "folders": sums.folders, "files": sums.files, "listed_at": sums.listed_at}
454
+ tmp = folder / f".{group.cache_name}.tmp"
455
+ tmp.write_text(json.dumps(doc, separators=(",", ":"), sort_keys=True))
456
+ os.replace(tmp, folder / group.cache_name)
457
+ except OSError as exc: # the table still has it; only a restart lists it again
458
+ logger.warning("package sizes: could not cache %s (%s)", group, exc)