fix(ex_app): retry model downloads when the host throttles

CHANGELOG.md

@@ -2,6 +2,12 @@
 
 All notable changes to this project will be documented in this file.
 
+## [0.30.4 - 2026-08-11]
+
+### Fixed
+
+- ExApps: downloading a model from a direct link is retried when the host is throttling or temporarily failing (`408`, `425`, `429`, `500`, `502`, `503`, `504`) instead of failing the whole `init`. `Retry-After` is honoured up to a minute, otherwise the wait backs off exponentially. The number of extra attempts defaults to 5 and can be set per model with the new `max_retries` download option.
+
 ## [0.30.3 - 2026-08-11]
 
 ### Added

nc_py_api/ex_app/integration_fastapi.py

@@ -2,12 +2,16 @@
 
 import asyncio
 import builtins
+import contextlib
 import fnmatch
 import hashlib
 import json
 import os
+import time
 import typing
 import warnings
+from datetime import datetime, timezone
+from email.utils import parsedate_to_datetime
 from traceback import format_exc
 from urllib.parse import urlparse
 
@@ -30,8 +34,18 @@
 from .._misc import get_username_secret_from_headers
 from ..nextcloud import AsyncNextcloudApp, NextcloudApp
 from ..talk_bot import TalkBotMessage
+from .defs import LogLvl
 from .misc import persistent_storage
 
+MODEL_FETCH_RETRY_STATUSES = frozenset({408, 425, 429, 500, 502, 503, 504})
+"""Statuses for which downloading a model file is retried; model hosters answer ``429`` when rate-limiting."""
+
+MODEL_FETCH_RETRIES = 5
+"""How many additional attempts a model file download gets, see ``max_retries`` in ``fetch_models_task``."""
+
+MODEL_FETCH_MAX_DELAY = 60.0
+"""Upper bound in seconds for a single wait between attempts, also when the server asks for longer."""
+
 
 def _nc_app_internal(request: HTTPConnection) -> NextcloudApp:
     """Internal sync NextcloudApp factory (no deprecation warning)."""
@@ -99,6 +113,7 @@ def set_handlers(
             {
                 "model_url_1": {
                     "save_path": "path_or_filename_to_save_the_model_to",
+                    "max_retries": 5,
                 },
                 "huggingface_model_name_1": {
                     "max_workers": 4,
@@ -112,6 +127,8 @@ def set_handlers(
 
         .. note:: ``huggingface_hub`` package should be present for automatic models fetching.
                   All model options are optional and can be left empty.
+                  ``max_retries`` applies to direct links only and covers the statuses hosters use
+                  for throttling, e.g. ``429``; set it to ``0`` to fail on the first one.
 
     :param map_app_static: Should be folders ``js``, ``css``, ``l10n``, ``img`` automatically mounted in FastAPI or not.
 
@@ -199,6 +216,7 @@ def fetch_models_task(nc: NextcloudApp, models: dict[str, dict], progress_init_s
             {
                 "model_url_1": {
                     "save_path": "path_or_filename_to_save_the_model_to",
+                    "max_retries": 5,
                 },
                 "huggingface_model_name_1": {
                     "max_workers": 4,
@@ -211,6 +229,8 @@ def fetch_models_task(nc: NextcloudApp, models: dict[str, dict], progress_init_s
 
         .. note:: ``huggingface_hub`` package should be present for automatic models fetching.
                   All model options are optional and can be left empty.
+                  ``max_retries`` applies to direct links only and covers the statuses hosters use
+                  for throttling, e.g. ``429``; set it to ``0`` to fail on the first one.
 
     :param progress_init_start_value: Integer value defining from which percent the progress should start.
 
@@ -237,13 +257,69 @@ def fetch_models_task(nc: NextcloudApp, models: dict[str, dict], progress_init_s
     nc.set_init_status(100)
 
 
+def __retry_delay(response, attempt: int) -> float:
+    """Seconds to wait before the next attempt, following ``Retry-After`` when the server sends a usable one."""
+    retry_after = str(response.headers.get("Retry-After", "")).strip()
+    if retry_after:
+        try:
+            delay = float(retry_after)
+        except ValueError:
+            try:
+                delay = (parsedate_to_datetime(retry_after) - datetime.now(timezone.utc)).total_seconds()
+            except (TypeError, ValueError):
+                delay = -1.0
+        if delay >= 0:
+            return min(delay, MODEL_FETCH_MAX_DELAY)
+    return min(2.0**attempt, MODEL_FETCH_MAX_DELAY)
+
+
+def __request_model_file(model_path: str, nc: NextcloudApp, max_retries: int):
+    """Requests the model, retrying the statuses hosters use to throttle. Returns the last response either way."""
+    response = niquests.get(model_path, stream=True)
+    for attempt in range(max_retries):
+        if response.ok or response.status_code not in MODEL_FETCH_RETRY_STATUSES:
+            return response
+        delay = __retry_delay(response, attempt)
+        # the body of a throttled answer is never read, so the connection has to be released by hand;
+        # only transport errors are ignored here, a wrong call should still surface
+        with contextlib.suppress(OSError):
+            response.close()
+        nc.log(
+            LogLvl.WARNING,
+            f"Downloading of '{model_path}' returned {response.status_code},"
+            f" retrying in {delay:.0f}s ({attempt + 1}/{max_retries}).",
+        )
+        time.sleep(delay)
+        response = niquests.get(model_path, stream=True)
+    return response
+
+
+def __already_downloaded(result_path: str, linked_etag: str, total_size: int) -> bool:
+    """Whether the file on disk is the one the server offers, so downloading it again can be skipped."""
+    try:
+        existing_size = os.path.getsize(result_path)
+    except OSError:
+        return False
+    if not linked_etag or not total_size or total_size != existing_size:
+        return False
+    sha256_hash = hashlib.sha256()
+    with builtins.open(result_path, "rb") as file:
+        for byte_block in iter(lambda: file.read(4096), b""):
+            sha256_hash.update(byte_block)
+    return f'"{sha256_hash.hexdigest()}"' == linked_etag
+
+
 def __fetch_model_as_file(
     current_progress: int, progress_for_task: int, nc: NextcloudApp, model_path: str, download_options: dict
 ) -> str:
     result_path = download_options.pop("save_path", urlparse(model_path).path.split("/")[-1])
+    max_retries = int(download_options.pop("max_retries", MODEL_FETCH_RETRIES))
     tmp_path = result_path + ".tmp"
     try:
-        with FileLock(result_path + ".lock", timeout=7200), niquests.get(model_path, stream=True) as response:
+        with (
+            FileLock(result_path + ".lock", timeout=7200),
+            __request_model_file(model_path, nc, max_retries) as response,
+        ):
             if not response.ok:
                 raise ModelFetchError(
                     f"Downloading of '{model_path}' failed, returned ({response.status_code}) {response.text}"
@@ -257,18 +333,9 @@ def __fetch_model_as_file(
             if not linked_etag:
                 linked_etag = response.headers.get("X-Linked-ETag", response.headers.get("ETag", ""))
             total_size = int(response.headers.get("Content-Length", 0))
-            try:
-                existing_size = os.path.getsize(result_path)
-            except OSError:
-                existing_size = 0
-            if linked_etag and total_size and total_size == existing_size:
-                with builtins.open(result_path, "rb") as file:
-                    sha256_hash = hashlib.sha256()
-                    for byte_block in iter(lambda: file.read(4096), b""):
-                        sha256_hash.update(byte_block)
-                    if f'"{sha256_hash.hexdigest()}"' == linked_etag:
-                        nc.set_init_status(min(current_progress + progress_for_task, 99))
-                        return result_path
+            if __already_downloaded(result_path, linked_etag, total_size):
+                nc.set_init_status(min(current_progress + progress_for_task, 99))
+                return result_path
 
             try:
                 with builtins.open(tmp_path, "wb") as file:

tests_unit/test_fetch_model_file.py

@@ -2,6 +2,8 @@
 
 import hashlib
 import os
+from datetime import datetime, timedelta, timezone
+from email.utils import format_datetime
 from threading import Thread
 from unittest import mock
 
@@ -10,7 +12,10 @@
 from filelock import Timeout as FileLockTimeout
 
 from nc_py_api._exceptions import ModelFetchError
-from nc_py_api.ex_app.integration_fastapi import fetch_models_task
+from nc_py_api.ex_app.integration_fastapi import (
+    MODEL_FETCH_MAX_DELAY,
+    fetch_models_task,
+)
 
 
 class FakeResponse:
@@ -27,10 +32,14 @@ def __init__(self, content: bytes, etag: str = "", status_code: int = 200, ok: b
             "Content-Length": str(len(content)),
             "ETag": etag or f'"{sha}"',
         }
+        self.closed = 0
 
     def iter_raw(self, _chunk_size):
         yield self.content
 
+    def close(self):
+        self.closed += 1
+
     def __enter__(self):
         return self
 
@@ -212,3 +221,127 @@ def test_progress_updates_sent(self, tmp_path):
         assert nc.set_init_status.called
         # Last call should be 100 (completion)
         assert nc.set_init_status.call_args_list[-1] == mock.call(100)
+
+
+def _throttled(status_code: int = 429, retry_after: str = "") -> FakeResponse:
+    response = FakeResponse(b"", status_code=status_code, ok=False)
+    if retry_after:
+        response.headers["Retry-After"] = retry_after
+    return response
+
+
+def _serving(*responses):
+    """Serves the given responses to `niquests.get`, one per call."""
+    it = iter(responses)
+    return mock.patch("nc_py_api.ex_app.integration_fastapi.niquests.get", side_effect=lambda *_a, **_kw: next(it))
+
+
+class TestFetchModelRetries:
+    """Model hosters answer 429 when they throttle; the download has to survive that."""
+
+    def test_retries_until_the_download_succeeds(self, tmp_path):
+        save_path = str(tmp_path / "model.bin")
+        content = b"model-data"
+        throttled = (_throttled(429), _throttled(503))
+
+        with (
+            _serving(*throttled, FakeResponse(content)) as mocked,
+            mock.patch("nc_py_api.ex_app.integration_fastapi.time.sleep") as sleep,
+        ):
+            fetch_models_task(_mock_nc(), {"https://example.com/m.bin": {"save_path": save_path}}, 0)
+
+        assert mocked.call_count == 3
+        assert sleep.call_count == 2
+        # the body of a throttled answer is never read, so its connection has to be released explicitly
+        assert [response.closed for response in throttled] == [1, 1]
+        with open(save_path, "rb") as f:
+            assert f.read() == content
+
+    def test_gives_up_after_max_retries(self, tmp_path):
+        save_path = str(tmp_path / "model.bin")
+
+        with (
+            _serving(*[_throttled() for _ in range(10)]) as mocked,
+            mock.patch("nc_py_api.ex_app.integration_fastapi.time.sleep"),
+            pytest.raises(ModelFetchError),
+        ):
+            fetch_models_task(_mock_nc(), {"https://example.com/m.bin": {"save_path": save_path, "max_retries": 3}}, 0)
+
+        assert mocked.call_count == 4  # the first attempt plus `max_retries`
+
+    def test_the_default_is_five_extra_attempts(self, tmp_path):
+        """`fetch_models_task` documents five extra attempts, so a change of the default has to fail here.
+
+        The default is exhausted rather than satisfied: serving a success instead would also pass with a
+        larger default, because the loop returns as soon as it gets one.
+        """
+        save_path = str(tmp_path / "model.bin")
+
+        with (
+            _serving(*[_throttled() for _ in range(6)]) as mocked,
+            mock.patch("nc_py_api.ex_app.integration_fastapi.time.sleep") as sleep,
+            pytest.raises(ModelFetchError),
+        ):
+            fetch_models_task(_mock_nc(), {"https://example.com/m.bin": {"save_path": save_path}}, 0)
+
+        assert mocked.call_count == 6  # the first attempt plus five retries
+        assert sleep.call_count == 5
+
+    def test_max_retries_zero_fails_on_the_first_answer(self, tmp_path):
+        save_path = str(tmp_path / "model.bin")
+
+        with (
+            _serving(_throttled()) as mocked,
+            mock.patch("nc_py_api.ex_app.integration_fastapi.time.sleep") as sleep,
+            pytest.raises(ModelFetchError),
+        ):
+            fetch_models_task(_mock_nc(), {"https://example.com/m.bin": {"save_path": save_path, "max_retries": 0}}, 0)
+
+        assert mocked.call_count == 1
+        assert not sleep.called
+
+    def test_does_not_retry_statuses_that_will_not_change(self, tmp_path):
+        save_path = str(tmp_path / "model.bin")
+
+        with (
+            _serving(_throttled(404)) as mocked,
+            mock.patch("nc_py_api.ex_app.integration_fastapi.time.sleep") as sleep,
+            pytest.raises(ModelFetchError),
+        ):
+            fetch_models_task(_mock_nc(), {"https://example.com/m.bin": {"save_path": save_path}}, 0)
+
+        assert mocked.call_count == 1
+        assert not sleep.called
+
+
+class TestRetryDelays:
+    """`Retry-After` is followed when usable, but never long enough to hang an ExApp init."""
+
+    @staticmethod
+    def _waited(tmp_path, *responses) -> list[float]:
+        with (
+            _serving(*responses, FakeResponse(b"data")),
+            mock.patch("nc_py_api.ex_app.integration_fastapi.time.sleep") as sleep,
+        ):
+            fetch_models_task(_mock_nc(), {"https://example.com/m.bin": {"save_path": str(tmp_path / "m.bin")}}, 0)
+        return [call[0][0] for call in sleep.call_args_list]
+
+    def test_follows_retry_after_seconds(self, tmp_path):
+        assert self._waited(tmp_path, _throttled(retry_after="7")) == [7]
+
+    def test_caps_an_unreasonable_retry_after(self, tmp_path):
+        assert self._waited(tmp_path, _throttled(retry_after="3600")) == [MODEL_FETCH_MAX_DELAY]
+
+    def test_follows_retry_after_as_http_date(self, tmp_path):
+        soon = format_datetime(datetime.now(timezone.utc) + timedelta(seconds=5), usegmt=True)
+        assert 0 < self._waited(tmp_path, _throttled(retry_after=soon))[0] <= 10
+
+    def test_ignores_a_retry_after_in_the_past(self, tmp_path):
+        past = format_datetime(datetime.now(timezone.utc) - timedelta(seconds=30), usegmt=True)
+        assert self._waited(tmp_path, _throttled(retry_after=past)) == [1]
+
+    def test_ignores_a_malformed_retry_after(self, tmp_path):
+        assert self._waited(tmp_path, _throttled(retry_after="soon")) == [1]
+
+    def test_falls_back_to_exponential_backoff(self, tmp_path):
+        assert self._waited(tmp_path, _throttled(), _throttled(), _throttled()) == [1, 2, 4]