diff --git a/CHANGELOG.md b/CHANGELOG.md index f64b2f33..10af6ab6 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/nc_py_api/ex_app/integration_fastapi.py b/nc_py_api/ex_app/integration_fastapi.py index 431b00e8..c2533741 100644 --- a/nc_py_api/ex_app/integration_fastapi.py +++ b/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: diff --git a/tests_unit/test_fetch_model_file.py b/tests_unit/test_fetch_model_file.py index 2692e1cd..40a242ec 100644 --- a/tests_unit/test_fetch_model_file.py +++ b/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]