diff --git a/docs/source/changelog.rst b/docs/source/changelog.rst index f6f67d66e..2d88b7c14 100644 --- a/docs/source/changelog.rst +++ b/docs/source/changelog.rst @@ -29,6 +29,12 @@ 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``, + ``retry_wait`` and ``retry_statuses`` options; subclasses can override + ``HTTPFileSystem._is_retryable`` to change the decision (#2124) + 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..154dbda8a 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,78 @@ 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, statuses=_RETRYABLE_STATUSES): + """Whether a failed HTTP read is worth repeating. + + 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 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, 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. ``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. + """ + 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 +135,9 @@ def __init__( client_kwargs=None, get_client=get_client, encoded=False, + retries=3, + retry_wait=1.0, + retry_statuses=None, **storage_options, ): """ @@ -87,6 +163,20 @@ 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. + 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() @@ -100,6 +190,26 @@ 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 + 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 @@ -232,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) @@ -245,10 +365,16 @@ 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, self._is_retryable + ) async def _get_file( self, rpath, lpath, chunk_size=5 * 2**20, callback=DEFAULT_CALLBACK, **kwargs @@ -379,6 +505,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 +524,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 +720,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 +741,8 @@ def __init__( size=None, loop=None, asynchronous=False, + retries=None, + retry_wait=None, **kwargs, ): if mode != "rb": @@ -614,6 +751,11 @@ 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._is_retryable = getattr(fs, "_is_retryable", _is_retryable) self.details = {"name": url, "size": size, "type": "file"} super().__init__( fs=fs, @@ -694,51 +836,80 @@ 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']})", + 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 e27ce359e..e5c575936 100644 --- a/fsspec/implementations/tests/test_http.py +++ b/fsspec/implementations/tests/test_http.py @@ -10,8 +10,19 @@ 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.implementations.http import ( + _RETRYABLE_STATUSES, + HTTPFileSystem, + HTTPStreamFile, +) +from fsspec.tests.conftest import ( # noqa: F401 + HTTPTestHandler, + data, + reset_faults, + reset_files, + server, + win, +) def test_list(server): @@ -650,3 +661,229 @@ 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 _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: + 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 + + +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 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)