From 9ebeb6ce8d26272ddaf65232f0499828f4655698 Mon Sep 17 00:00:00 2001 From: Luca Foppiano Date: Wed, 9 Sep 2026 09:36:31 +0100 Subject: [PATCH 1/2] HTTP: retry transient failures of cat_file and HTTPFile block reads HTTPFileSystem never retried anything: every cat_file and every HTTPFile block read was one session.get plus one body read, so a single 503, a throttling 429 or a connection dropped mid-body killed the whole read. Add a small loop around those two paths that retries 408/425/429/5xx, dropped or reset connections, timeouts and truncated bodies, honouring Retry-After and otherwise waiting retry_wait * 2**n (capped, jittered). 404, 401/403 and other 4xx are never retried, so subclasses that map them in _raise_not_found_for_status keep working. The loop wraps request and body read together on purpose: a mid-body ClientPayloadError surfaces from r.read(), which a request-level retry client never sees. Two silent truncations become (retried) errors: a 206 body shorter than the requested range while the file size says more exists, and a 416 for a range inside the file, the CloudFront behaviour reported in #1895. New HTTPFileSystem options retries=3 and retry_wait=1.0, overridable per open(); retries=0 restores the previous behaviour. The shared test server gains header-driven fault injection (fail_status, fail_times, retry_after, truncate_body, short_body) and a reset_faults fixture. Refs #550, #1895. --- docs/source/changelog.rst | 5 + fsspec/implementations/http.py | 219 +++++++++++++++++----- fsspec/implementations/tests/test_http.py | 175 ++++++++++++++++- fsspec/tests/conftest.py | 56 ++++++ 4 files changed, 410 insertions(+), 45 deletions(-) diff --git a/docs/source/changelog.rst b/docs/source/changelog.rst index f6f67d66e..256c621c0 100644 --- a/docs/source/changelog.rst +++ b/docs/source/changelog.rst @@ -29,6 +29,11 @@ Enhancements - Allow ``simplecache`` files to be removed by age using their modification times (#2118) +- HTTP: retry transient failures (408/425/429/5xx, dropped connections, + timeouts, truncated range bodies, a 416 inside the file) of ``cat_file`` + and ``HTTPFile`` block reads, controlled by the new ``retries`` and + ``retry_wait`` options (#NNNN) + Fixes - Make ``merge_offset_ranges`` ``O(n log n)`` and keep merged blocks within diff --git a/fsspec/implementations/http.py b/fsspec/implementations/http.py index aff1a955c..3100603c7 100644 --- a/fsspec/implementations/http.py +++ b/fsspec/implementations/http.py @@ -1,6 +1,7 @@ import asyncio import io import logging +import random import re import weakref from copy import copy @@ -28,6 +29,75 @@ ex2 = re.compile(r"""(?Phttp[s]?://[-a-zA-Z0-9@:%_+.~#?&/=]+)""") logger = logging.getLogger("fsspec.http") +_RETRYABLE_STATUSES = frozenset({408, 425, 429, 500, 502, 503, 504}) +_RETRY_MAX_WAIT = 60.0 +_retry_sleep = asyncio.sleep # module attribute so tests can substitute a fake + + +def _is_retryable(exc): + """Whether a failed HTTP read is worth repeating. + + Throttling and server-side errors, dropped or reset connections, timeouts + and truncated bodies are transient; anything else (including the + ``FileNotFoundError``/``PermissionError`` a subclass may map 4xx codes to) + is deterministic and propagates at once. + """ + if isinstance(exc, aiohttp.ClientResponseError): + return exc.status in _RETRYABLE_STATUSES + if isinstance(exc, aiohttp.ClientSSLError): + return False + return isinstance( + exc, + ( + aiohttp.ClientConnectionError, + aiohttp.ClientPayloadError, + asyncio.TimeoutError, + ), + ) + + +def _retry_after(exc): + """Delta-seconds ``Retry-After`` carried by a response error, else None.""" + headers = getattr(exc, "headers", None) + if not headers: + return None + try: + return max(0.0, float(headers.get("Retry-After"))) + except (TypeError, ValueError): + return None + + +async def _with_retries(attempt, retries, retry_wait, label): + """Await ``attempt()``, repeating it on transient failures. + + ``attempt`` is a zero-argument coroutine function covering the whole + request *and* body read, so a connection dropped mid-body counts as a + failure too. A ``Retry-After`` header is honoured when present; otherwise + the wait doubles from ``retry_wait`` on each retry, capped at + ``_RETRY_MAX_WAIT`` seconds and jittered. Non-retryable errors and the + final failure propagate unchanged. + """ + for n in range(retries + 1): + try: + return await attempt() + except Exception as exc: + if n >= retries or not _is_retryable(exc): + raise + delay = _retry_after(exc) + if delay is None: + delay = min(_RETRY_MAX_WAIT, retry_wait * 2**n) * ( + 0.5 + random.random() + ) + logger.warning( + "%s: attempt %d/%d failed (%r); retrying in %.1fs", + label, + n + 1, + retries + 1, + exc, + delay, + ) + await _retry_sleep(delay) + async def get_client(**kwargs): return aiohttp.ClientSession(**kwargs) @@ -62,6 +132,8 @@ def __init__( client_kwargs=None, get_client=get_client, encoded=False, + retries=3, + retry_wait=1.0, **storage_options, ): """ @@ -87,6 +159,15 @@ def __init__( A callable, which takes keyword arguments and constructs an aiohttp.ClientSession. Its state will be managed by the HTTPFileSystem class. + retries: int + How many times a transient failure of a read (HTTP 408/425/429/5xx, + a dropped or reset connection, a timeout, a truncated range body) + is retried in ``cat_file`` and ``HTTPFile`` block reads; 0 disables + retries. + retry_wait: float + Seconds to wait before the first retry; doubled on each further + retry (capped at 60 s, with jitter). A ``Retry-After`` header sent + by the server takes precedence. storage_options: key-value Any other parameters passed on to requests cache_type, cache_options: defaults used in open() @@ -100,6 +181,12 @@ def __init__( self.client_kwargs = client_kwargs or {} self.get_client = get_client self.encoded = encoded + if retries < 0: + raise ValueError(f"retries must be >= 0, got {retries!r}") + if retry_wait < 0: + raise ValueError(f"retry_wait must be >= 0, got {retry_wait!r}") + self.retries = retries + self.retry_wait = retry_wait self.kwargs = storage_options self._session = None @@ -245,10 +332,14 @@ async def _cat_file(self, url, start=None, end=None, **kwargs): headers["Range"] = await self._process_limits(url, start, end) kw["headers"] = headers session = await self.set_session() - async with session.get(self.encode_url(url), **kw) as r: - out = await r.read() - self._raise_not_found_for_status(r, url) - return out + + async def _once(): + async with session.get(self.encode_url(url), **kw) as r: + out = await r.read() + self._raise_not_found_for_status(r, url) + return out + + return await _with_retries(_once, self.retries, self.retry_wait, url) async def _get_file( self, rpath, lpath, chunk_size=5 * 2**20, callback=DEFAULT_CALLBACK, **kwargs @@ -379,6 +470,8 @@ def _open( if mode != "rb": raise NotImplementedError block_size = block_size if block_size is not None else self.block_size + # per-open retry overrides are for HTTPFile only, never request options + retry_kw = {k: kwargs.pop(k) for k in ("retries", "retry_wait") if k in kwargs} kw = self.kwargs.copy() kw["asynchronous"] = self.asynchronous kw.update(kwargs) @@ -396,6 +489,7 @@ def _open( cache_type=cache_type or self.cache_type, cache_options=cache_options or self.cache_options, loop=self.loop, + **retry_kw, **kw, ) else: @@ -591,6 +685,12 @@ class HTTPFile(AbstractBufferedFile): size: None or int If given, this is the size of the file in bytes, and we don't attempt to call the server to find the value. + retries: int or None + Retries for a transient failure of a block read; None (default) uses + the value configured on the filesystem. + retry_wait: float or None + Wait before the first retry, doubled on each further retry; None + (default) uses the value configured on the filesystem. kwargs: all other key-values are passed to requests calls. """ @@ -606,6 +706,8 @@ def __init__( size=None, loop=None, asynchronous=False, + retries=None, + retry_wait=None, **kwargs, ): if mode != "rb": @@ -614,6 +716,10 @@ def __init__( self.loop = loop self.url = url self.session = session + self.retries = getattr(fs, "retries", 0) if retries is None else retries + self.retry_wait = ( + getattr(fs, "retry_wait", 1.0) if retry_wait is None else retry_wait + ) self.details = {"name": url, "size": size, "type": "file"} super().__init__( fs=fs, @@ -694,51 +800,76 @@ async def async_fetch_range(self, start, end): headers = kwargs.pop("headers", {}).copy() headers["Range"] = f"bytes={start}-{end - 1}" logger.debug(f"{self.url} : {headers['Range']}") - r = await self.session.get( - self.fs.encode_url(self.url), headers=headers, **kwargs - ) - async with r: - if r.status == 416: - # range request outside file - return b"" - r.raise_for_status() - # If the server has handled the range request, it should reply - # with status 206 (partial content). But we'll guess that a suitable - # Content-Range header or a Content-Length no more than the - # requested range also mean we have got the desired range. - response_is_range = ( - r.status == 206 - or self._parse_content_range(r.headers)[0] == start - or int(r.headers.get("Content-Length", end + 1)) <= end - start + async def _once(): + r = await self.session.get( + self.fs.encode_url(self.url), headers=headers, **kwargs ) + async with r: + if r.status == 416: + if self.size is not None and start < self.size: + # Some servers (CloudFront under load, see #1895) + # answer 416 to a range that lies inside the file; + # treating it as EOF would silently truncate the read. + raise aiohttp.ClientPayloadError( + f"{headers['Range']} of {self.url} reported " + f"unsatisfiable, but the file has {self.size} bytes" + ) + # range request outside file + return b"" + r.raise_for_status() - if response_is_range: - # partial content, as expected - out = await r.read() - elif start > 0: - raise ValueError( - "The HTTP server doesn't appear to support range requests. " - "Only reading this file from the beginning is supported. " - "Open with block_size=0 for a streaming file interface." + # If the server has handled the range request, it should reply + # with status 206 (partial content). But we'll guess that a + # suitable Content-Range header or a Content-Length no more + # than the requested range also mean we have got the desired + # range. + response_is_range = ( + r.status == 206 + or self._parse_content_range(r.headers)[0] == start + or int(r.headers.get("Content-Length", end + 1)) <= end - start ) - else: - # Response is not a range, but we want the start of the file, - # so we can read the required amount anyway. - cl = 0 - out = [] - while True: - chunk = await r.content.read(2**20) - # data size unknown, let's read until we have enough - if chunk: - out.append(chunk) - cl += len(chunk) - if cl > end - start: + + if response_is_range: + # partial content, as expected + out = await r.read() + if self.size is not None: + expected = min(end, self.size) - start + if len(out) < expected: + # short body with consistent headers: a truncated + # block would otherwise be served from the cache + raise aiohttp.ClientPayloadError( + f"{headers['Range']} of {self.url} returned " + f"{len(out)} bytes, expected {expected}" + ) + elif start > 0: + raise ValueError( + "The HTTP server doesn't appear to support range " + "requests. Only reading this file from the beginning " + "is supported. Open with block_size=0 for a streaming " + "file interface." + ) + else: + # Response is not a range, but we want the start of the + # file, so we can read the required amount anyway. + cl = 0 + out = [] + while True: + chunk = await r.content.read(2**20) + # data size unknown, let's read until we have enough + if chunk: + out.append(chunk) + cl += len(chunk) + if cl > end - start: + break + else: break - else: - break - out = b"".join(out)[: end - start] - return out + out = b"".join(out)[: end - start] + return out + + return await _with_retries( + _once, self.retries, self.retry_wait, f"{self.url} ({headers['Range']})" + ) _fetch_range = sync_wrapper(async_fetch_range) diff --git a/fsspec/implementations/tests/test_http.py b/fsspec/implementations/tests/test_http.py index e27ce359e..8e7b7b030 100644 --- a/fsspec/implementations/tests/test_http.py +++ b/fsspec/implementations/tests/test_http.py @@ -11,7 +11,14 @@ import fsspec.asyn import fsspec.utils from fsspec.implementations.http import HTTPStreamFile -from fsspec.tests.conftest import data, reset_files, server, win # noqa: F401 +from fsspec.tests.conftest import ( # noqa: F401 + HTTPTestHandler, + data, + reset_faults, + reset_files, + server, + win, +) def test_list(server): @@ -650,3 +657,169 @@ def test_protocol_independent_of_first_used_protocol(protocol): fs1 = filesystem("https") p1 = fs1.protocol[0] if isinstance(fs1.protocol, tuple) else fs1.protocol assert p0 == p1 == "http" + + +# --- retries ----------------------------------------------------------------- +# The server faults are driven by request headers, see HTTPTestHandler._serve_fault. + +_RETRY_HEADERS = {"give_length": "true", "head_ok": "true", "use_206": "true"} +_REALFILE_PATH = "/index/realfile" + + +def _retry_fs(fault_headers=None, **kwargs): + headers = dict(_RETRY_HEADERS, **(fault_headers or {})) + kwargs.setdefault("retry_wait", 0) + return fsspec.filesystem( + "http", headers=headers, skip_instance_cache=True, **kwargs + ) + + +def _gets(path=_REALFILE_PATH): + return HTTPTestHandler.get_counts.get(path, 0) + + +def _faults(path=_REALFILE_PATH): + return HTTPTestHandler.fault_counts.get(path, 0) + + +def test_retry_503_then_succeeds(server, reset_faults): + fs = _retry_fs({"fail_status": "503"}) + with fs.open(server.realfile) as f: + assert f.read(len(data)) == data + assert _faults() == 1 + assert _gets() == 2 + + +def test_retry_cat_file(server, reset_faults): + fs = _retry_fs({"fail_status": "503", "fail_times": "2"}) + assert fs.cat_file(server.realfile, start=10, end=200) == data[10:200] + assert _gets() == 3 + HTTPTestHandler.fault_counts.clear() + HTTPTestHandler.get_counts.clear() + assert fs.cat_file(server.realfile) == data + assert _gets() == 3 + + +@pytest.mark.parametrize("status", [400, 403, 416]) +def test_retry_not_for_4xx(server, reset_faults, status): + # via cat_file, which has no notion of file size: a 416 there is final + fs = _retry_fs({"fail_status": str(status), "fail_times": "5"}) + with pytest.raises(aiohttp.ClientResponseError) as e: + fs.cat_file(server.realfile) + assert e.value.status == status + assert _gets() == 1 + + +def test_retry_not_for_404(server, reset_faults): + fs = _retry_fs() + url = server.address + "/index/missing" + with pytest.raises(FileNotFoundError): + fs.cat_file(url) + assert _gets("/index/missing") == 1 + + +def test_retry_exhausted_raises_original(server, reset_faults): + fs = _retry_fs({"fail_status": "503", "fail_times": "10"}, retries=2) + with pytest.raises(aiohttp.ClientResponseError) as e: + fs.cat_file(server.realfile) + assert e.value.status == 503 + assert _gets() == 3 + + +def test_retry_disabled_keeps_old_behaviour(server, reset_faults): + fs = _retry_fs({"fail_status": "503"}, retries=0) + with pytest.raises(aiohttp.ClientResponseError) as e: + fs.cat_file(server.realfile) + assert e.value.status == 503 + assert _gets() == 1 + + +def test_retry_truncated_body(server, reset_faults): + fs = _retry_fs({"truncate_body": "true"}) + with fs.open(server.realfile) as f: + assert f.read(len(data)) == data + assert _faults() == 1 + assert _gets() == 2 + + HTTPTestHandler.fault_counts.clear() + fs = _retry_fs({"truncate_body": "true"}, retries=0) + with fs.open(server.realfile) as f: + with pytest.raises(aiohttp.ClientPayloadError): + f.read(len(data)) + + +def test_retry_short_206_body(server, reset_faults): + # headers are self-consistent, so only the length check can notice + fs = _retry_fs({"short_body": "true"}) + with fs.open(server.realfile) as f: + assert f.read(len(data)) == data + assert _faults() == 1 + assert _gets() == 2 + + HTTPTestHandler.fault_counts.clear() + fs = _retry_fs({"short_body": "true"}, retries=0) + with fs.open(server.realfile) as f: + with pytest.raises(aiohttp.ClientPayloadError, match="expected"): + f.read(len(data)) + + +def test_retry_416_inside_file(server, reset_faults): + # gh-1895: a 416 for a range that lies inside the file is not EOF + fs = _retry_fs({"fail_status": "416"}) + with fs.open(server.realfile) as f: + assert f.read(len(data)) == data + assert _faults() == 1 + assert _gets() == 2 + + HTTPTestHandler.fault_counts.clear() + fs = _retry_fs({"fail_status": "416"}, retries=0) + with fs.open(server.realfile) as f: + with pytest.raises(aiohttp.ClientPayloadError, match="unsatisfiable"): + f.read(len(data)) + + +def test_retry_honours_retry_after(server, reset_faults, monkeypatch): + sleeps = [] + + async def fake_sleep(delay): + sleeps.append(delay) + + monkeypatch.setattr(fsspec.implementations.http, "_retry_sleep", fake_sleep) + fs = _retry_fs({"fail_status": "429", "retry_after": "7"}, retry_wait=1) + assert fs.cat_file(server.realfile) == data + assert sleeps == [7.0] + + +def test_retry_backoff_sequence(server, reset_faults, monkeypatch): + sleeps = [] + + async def fake_sleep(delay): + sleeps.append(delay) + + monkeypatch.setattr(fsspec.implementations.http, "_retry_sleep", fake_sleep) + monkeypatch.setattr(fsspec.implementations.http.random, "random", lambda: 0.5) + fs = _retry_fs({"fail_status": "503", "fail_times": "3"}, retry_wait=1) + assert fs.cat_file(server.realfile) == data + assert sleeps == [1, 2, 4] + assert _gets() == 4 + + +def test_retry_option_validation(): + with pytest.raises(ValueError): + fsspec.filesystem("http", retries=-1, skip_instance_cache=True) + with pytest.raises(ValueError): + fsspec.filesystem("http", retry_wait=-1, skip_instance_cache=True) + + +def test_retry_per_open_override(server, reset_faults): + fs = _retry_fs({"fail_status": "503"}, retries=3) + with fs.open(server.realfile, retries=0) as f: + assert f.retries == 0 + with pytest.raises(aiohttp.ClientResponseError): + f.read(len(data)) + assert _gets() == 1 + + # the override must not leak into request kwargs on the streaming branch + with fs.open(server.realfile, block_size=0, retries=1) as f: + assert isinstance(f, HTTPStreamFile) + assert f.read() == data diff --git a/fsspec/tests/conftest.py b/fsspec/tests/conftest.py index 0a56521ed..6b286461d 100644 --- a/fsspec/tests/conftest.py +++ b/fsspec/tests/conftest.py @@ -44,6 +44,17 @@ def reset_files(): HTTPTestHandler.dynamic_files.clear() +@pytest.fixture +def reset_faults(): + # Per-path counters behind the fault-injection request headers + # (fail_status / truncate_body / short_body, see HTTPTestHandler._serve_fault) + HTTPTestHandler.fault_counts.clear() + HTTPTestHandler.get_counts.clear() + yield + HTTPTestHandler.fault_counts.clear() + HTTPTestHandler.get_counts.clear() + + class HTTPTestHandler(BaseHTTPRequestHandler): static_files = { "/index/realfile": data, @@ -57,6 +68,9 @@ class HTTPTestHandler(BaseHTTPRequestHandler): "/unauthorized": AssertionError("shouldn't access"), } dynamic_files = {} + # fault injection, keyed by request path; cleared by the reset_faults fixture + fault_counts = {} + get_counts = {} files = ChainMap(dynamic_files, static_files) @@ -73,9 +87,49 @@ def _respond(self, code=200, headers=None, data=b""): if data: self.wfile.write(data) + def _serve_fault(self, status, content_range, file_data): + """Serve the fault the client asked for, while its budget lasts. + + Request headers: ``fail_status: `` answers with that status (and + ``Retry-After: `` when given); ``truncate_body: true`` + announces the full Content-Length but sends only half the body before + the connection closes; ``short_body: true`` sends half the body with a + matching (short) Content-Length. ``fail_times: `` (default 1) is + how many GETs of this path fault before it is served normally. + Returns True when a fault was served. + """ + modes = [ + k + for k in ("fail_status", "truncate_body", "short_body") + if k in self.headers + ] + if not modes: + return False + budget = int(self.headers.get("fail_times", 1)) + if self.fault_counts.get(self.path, 0) >= budget: + return False + self.fault_counts[self.path] = self.fault_counts.get(self.path, 0) + 1 + if "fail_status" in self.headers: + headers = {"Content-Length": 0} + if "retry_after" in self.headers: + headers["Retry-After"] = self.headers["retry_after"] + self._respond(int(self.headers["fail_status"]), headers) + elif "truncate_body" in self.headers: + half = file_data[: len(file_data) // 2] + headers = {"Content-Length": len(file_data), "Content-Range": content_range} + self._respond(status, headers, half) + self.wfile.flush() + self.close_connection = True + else: + half = file_data[: len(file_data) // 2] + headers = {"Content-Length": len(half), "Content-Range": content_range} + self._respond(status, headers, half) + return True + def do_GET(self): baseurl = f"http://127.0.0.1:{self.server.server_port}" file_path = self.path + self.get_counts[self.path] = self.get_counts.get(self.path, 0) + 1 if file_path.endswith("/") and file_path.rstrip("/") in self.files: file_path = file_path.rstrip("/") file_data = self.files.get(file_path) @@ -107,6 +161,8 @@ def do_GET(self): file_data = file_data[-int(end) :] if "use_206" in self.headers: status = 206 + if self._serve_fault(status, content_range, file_data): + return if "give_length" in self.headers: if "gzip_encoding" in self.headers: file_data = gzip.compress(file_data) From 34de45b3d1c5d5562c9e7aa8efe9e1d56980c33c Mon Sep 17 00:00:00 2001 From: Luca Foppiano Date: Wed, 9 Sep 2026 15:11:15 +0100 Subject: [PATCH 2/2] HTTP: make the retried status codes configurable Add a ``retry_statuses`` option to HTTPFileSystem (default unchanged: 408/425/429/500/502/503/504) and an overridable ``_is_retryable(exc)`` method, so a caller can opt extra codes in (a CDN answering 403 under load) or opt default codes out without patching the module. Dropped connections, timeouts and truncated bodies are still always retried. HTTPFile picks the predicate up from its filesystem. --- docs/source/changelog.rst | 5 +- fsspec/implementations/http.py | 62 +++++++++++++++++---- fsspec/implementations/tests/test_http.py | 66 ++++++++++++++++++++++- 3 files changed, 119 insertions(+), 14 deletions(-) diff --git a/docs/source/changelog.rst b/docs/source/changelog.rst index 256c621c0..2d88b7c14 100644 --- a/docs/source/changelog.rst +++ b/docs/source/changelog.rst @@ -31,8 +31,9 @@ Enhancements - HTTP: retry transient failures (408/425/429/5xx, dropped connections, timeouts, truncated range bodies, a 416 inside the file) of ``cat_file`` - and ``HTTPFile`` block reads, controlled by the new ``retries`` and - ``retry_wait`` options (#NNNN) + and ``HTTPFile`` block reads, controlled by the new ``retries``, + ``retry_wait`` and ``retry_statuses`` options; subclasses can override + ``HTTPFileSystem._is_retryable`` to change the decision (#2124) Fixes diff --git a/fsspec/implementations/http.py b/fsspec/implementations/http.py index 3100603c7..154dbda8a 100644 --- a/fsspec/implementations/http.py +++ b/fsspec/implementations/http.py @@ -34,16 +34,16 @@ _retry_sleep = asyncio.sleep # module attribute so tests can substitute a fake -def _is_retryable(exc): +def _is_retryable(exc, statuses=_RETRYABLE_STATUSES): """Whether a failed HTTP read is worth repeating. - Throttling and server-side errors, dropped or reset connections, timeouts - and truncated bodies are transient; anything else (including the - ``FileNotFoundError``/``PermissionError`` a subclass may map 4xx codes to) - is deterministic and propagates at once. + A response error is retried when its status is in ``statuses``; dropped + or reset connections, timeouts and truncated bodies are always transient; + anything else (including the ``FileNotFoundError``/``PermissionError`` a + subclass may map 4xx codes to) is deterministic and propagates at once. """ if isinstance(exc, aiohttp.ClientResponseError): - return exc.status in _RETRYABLE_STATUSES + return exc.status in statuses if isinstance(exc, aiohttp.ClientSSLError): return False return isinstance( @@ -67,12 +67,15 @@ def _retry_after(exc): return None -async def _with_retries(attempt, retries, retry_wait, label): +async def _with_retries( + attempt, retries, retry_wait, label, is_retryable=_is_retryable +): """Await ``attempt()``, repeating it on transient failures. ``attempt`` is a zero-argument coroutine function covering the whole request *and* body read, so a connection dropped mid-body counts as a - failure too. A ``Retry-After`` header is honoured when present; otherwise + failure too. ``is_retryable(exc)`` decides whether a failure is worth + repeating. A ``Retry-After`` header is honoured when present; otherwise the wait doubles from ``retry_wait`` on each retry, capped at ``_RETRY_MAX_WAIT`` seconds and jittered. Non-retryable errors and the final failure propagate unchanged. @@ -81,7 +84,7 @@ async def _with_retries(attempt, retries, retry_wait, label): try: return await attempt() except Exception as exc: - if n >= retries or not _is_retryable(exc): + if n >= retries or not is_retryable(exc): raise delay = _retry_after(exc) if delay is None: @@ -134,6 +137,7 @@ def __init__( encoded=False, retries=3, retry_wait=1.0, + retry_statuses=None, **storage_options, ): """ @@ -168,6 +172,11 @@ def __init__( Seconds to wait before the first retry; doubled on each further retry (capped at 60 s, with jitter). A ``Retry-After`` header sent by the server takes precedence. + retry_statuses: iterable of int or None + HTTP status codes of a failed read that are retried; default + (None) is 408, 425, 429, 500, 502, 503 and 504. Dropped + connections, timeouts and truncated bodies are retried regardless. + Override ``_is_retryable`` in a subclass for finer control. storage_options: key-value Any other parameters passed on to requests cache_type, cache_options: defaults used in open() @@ -187,6 +196,20 @@ def __init__( raise ValueError(f"retry_wait must be >= 0, got {retry_wait!r}") self.retries = retries self.retry_wait = retry_wait + if retry_statuses is None: + self.retry_statuses = _RETRYABLE_STATUSES + else: + if isinstance(retry_statuses, str): + # "503" would otherwise iterate to {5, 0, 3} + raise ValueError( + f"retry_statuses must be an iterable of ints, got {retry_statuses!r}" + ) + try: + self.retry_statuses = frozenset(int(s) for s in retry_statuses) + except (TypeError, ValueError): + raise ValueError( + f"retry_statuses must be an iterable of ints, got {retry_statuses!r}" + ) from None self.kwargs = storage_options self._session = None @@ -319,6 +342,16 @@ def _raise_not_found_for_status(self, response, url): raise FileNotFoundError(url) response.raise_for_status() + def _is_retryable(self, exc): + """Whether a failed read should be tried again. + + Consulted by ``cat_file`` and ``HTTPFile`` block reads after each + failure. Response errors are retried when their status is in + ``retry_statuses``; connection errors, timeouts and truncated bodies + always are. Override to change the decision for a particular server. + """ + return _is_retryable(exc, self.retry_statuses) + async def _cat_file(self, url, start=None, end=None, **kwargs): kw = self.kwargs.copy() kw.update(kwargs) @@ -339,7 +372,9 @@ async def _once(): self._raise_not_found_for_status(r, url) return out - return await _with_retries(_once, self.retries, self.retry_wait, url) + return await _with_retries( + _once, self.retries, self.retry_wait, url, self._is_retryable + ) async def _get_file( self, rpath, lpath, chunk_size=5 * 2**20, callback=DEFAULT_CALLBACK, **kwargs @@ -720,6 +755,7 @@ def __init__( self.retry_wait = ( getattr(fs, "retry_wait", 1.0) if retry_wait is None else retry_wait ) + self._is_retryable = getattr(fs, "_is_retryable", _is_retryable) self.details = {"name": url, "size": size, "type": "file"} super().__init__( fs=fs, @@ -868,7 +904,11 @@ async def _once(): return out return await _with_retries( - _once, self.retries, self.retry_wait, f"{self.url} ({headers['Range']})" + _once, + self.retries, + self.retry_wait, + f"{self.url} ({headers['Range']})", + self._is_retryable, ) _fetch_range = sync_wrapper(async_fetch_range) diff --git a/fsspec/implementations/tests/test_http.py b/fsspec/implementations/tests/test_http.py index 8e7b7b030..e5c575936 100644 --- a/fsspec/implementations/tests/test_http.py +++ b/fsspec/implementations/tests/test_http.py @@ -10,7 +10,11 @@ import fsspec.asyn import fsspec.utils -from fsspec.implementations.http import HTTPStreamFile +from fsspec.implementations.http import ( + _RETRYABLE_STATUSES, + HTTPFileSystem, + HTTPStreamFile, +) from fsspec.tests.conftest import ( # noqa: F401 HTTPTestHandler, data, @@ -682,6 +686,12 @@ def _faults(path=_REALFILE_PATH): return HTTPTestHandler.fault_counts.get(path, 0) +def _rearm(): + """Restore the fault budget and GET counter for a second read in a test.""" + HTTPTestHandler.fault_counts.clear() + HTTPTestHandler.get_counts.clear() + + def test_retry_503_then_succeeds(server, reset_faults): fs = _retry_fs({"fail_status": "503"}) with fs.open(server.realfile) as f: @@ -823,3 +833,57 @@ def test_retry_per_open_override(server, reset_faults): with fs.open(server.realfile, block_size=0, retries=1) as f: assert isinstance(f, HTTPStreamFile) assert f.read() == data + + +def test_retry_custom_statuses(server, reset_faults): + # 403 is not retried by default, but can be opted in + fs = _retry_fs({"fail_status": "403"}, retry_statuses=[403]) + with fs.open(server.realfile) as f: + assert f.read(len(data)) == data + assert _gets() == 2 + _rearm() + assert fs.cat_file(server.realfile, start=10, end=200) == data[10:200] + assert _gets() == 2 + + # ... and the default codes can be opted out + fs = _retry_fs({"fail_status": "503"}, retry_statuses=[403]) + _rearm() + with pytest.raises(aiohttp.ClientResponseError): + fs.cat_file(server.realfile) + assert _gets() == 1 + + +def test_retry_statuses_validation(): + fs = fsspec.filesystem("http", skip_instance_cache=True) + assert fs.retry_statuses == _RETRYABLE_STATUSES + fs = fsspec.filesystem( + "http", retry_statuses=(500, "503"), skip_instance_cache=True + ) + assert fs.retry_statuses == frozenset({500, 503}) + with pytest.raises(ValueError): + fsspec.filesystem("http", retry_statuses=["abc"], skip_instance_cache=True) + with pytest.raises(ValueError): + fsspec.filesystem("http", retry_statuses=503, skip_instance_cache=True) + with pytest.raises(ValueError): + fsspec.filesystem("http", retry_statuses="503", skip_instance_cache=True) + + +def test_retry_is_retryable_override(server, reset_faults): + class BusyCDNFileSystem(HTTPFileSystem): + def _is_retryable(self, exc): + # a CDN that answers 403 under load + if isinstance(exc, aiohttp.ClientResponseError) and exc.status == 403: + return True + return super()._is_retryable(exc) + + fs = BusyCDNFileSystem( + headers=dict(_RETRY_HEADERS, fail_status="403"), + retry_wait=0, + skip_instance_cache=True, + ) + assert fs.cat_file(server.realfile, start=10, end=200) == data[10:200] + assert _gets() == 2 + _rearm() + with fs.open(server.realfile) as f: + assert f.read(len(data)) == data + assert _gets() == 2