refactor(transport)!: share fan-out execution across services

NEWS.md

@@ -1,3 +1,5 @@
+**08/06/2026:** Fan-out execution is now shared across services. Chunking is how a query is divided structurally (a Water Data/NGWMN URL over the byte limit); fan-out is how the pieces are distributed operationally. Only the first is protocol-specific, so the executor moved to `dataretrieval.transport.fanout` (`FanOut`, over a three-member `FanOutPlan` protocol) while chunk planning stays in `dataretrieval.ogc`. Water Use no longer re-implements the fan-out gather and inherits resume, progress reporting, and `API_USGS_CONCURRENT`: a multi-location pull interrupted by a rate limit now raises a resumable interruption whose `.call.resume()` re-issues only the locations that did not finish, instead of discarding every completed one. The interruption taxonomy moved to the `dataretrieval.interruptions` leaf and its base class is now `FanOutInterrupted`; **`ChunkInterrupted` is a permanent alias of the same class**, so `except ChunkInterrupted` keeps working. **Breaking change:** a Water Use fan-out interrupted by a 5xx or 429 now raises `ServiceInterrupted`/`QuotaExhausted` rather than `ServiceUnavailable`/`RateLimited` — both are still `DataRetrievalError`, so broad handlers are unaffected, but a narrow `except ServiceUnavailable` around a Water Use call must widen. **Breaking change:** `wateruse.MAX_CONCURRENT_REQUESTS` is removed; set `API_USGS_CONCURRENT` (which now outranks any service default) or read `wateruse.DEFAULT_CONCURRENT_REQUESTS`.
+
 **08/02/2026:** Added an internal API-neutral transport layer for guarded HTTP clients, host-scoped authentication, cursor pagination, bounded retry, response aggregation, progress, and sync-over-async dispatch. Water Use and the non-OGC Statistics API now consume transport directly instead of private OGC execution helpers; WQP, NLDI, and StreamStats opt into bounded transient retry while deprecated NWIS behavior remains unchanged. OGC retains CQL2, request construction, feature shaping, chunk planning, resumable calls, and interruption types, with compatibility imports at previous private paths. Failed pagination and fan-out still raise rather than returning partial data, and no public signatures or return shapes changed.
 
 **08/02/2026:** Phase 1 OGC boundary stabilization: the `dataretrieval.ogc` package now exposes a deliberate, small facade (`OgcDialect`, `prepare_request_args`, `get_ogc_data`, `fetch_ogc_request`) that service adapters (NGWMN, Water Data's generic wrapper) import. Internal request-construction helpers moved to a new `ogc.requests` module; the dialect type and endpoint constants live in the leaf `ogc.policy` module. `ogc.shaping` no longer depends on `ogc.engine` at all, and the complete runtime OGC import graph is now acyclic. `_default_headers` now accepts a target URL and adds `X-Api-Key` only for `api.waterdata.usgs.gov`; shared sync and async clients also strip the key before following any cross-host redirect, including external rating-asset downloads. `waterdata.utils` no longer bulk-re-exports private OGC symbols, and consumers import implementation helpers from their canonical modules. No public API, return-value, or deprecation changes.

dataretrieval/__init__.py

@@ -45,23 +45,24 @@
     URLTooLong,
 )
 
-# Parallel-chunks control (a context manager). Defined with the chunker in
-# ``dataretrieval.ogc.chunking``; surfaced here for a stable public path
-# ``from dataretrieval import parallel_chunks``.
-from dataretrieval.ogc.chunking import parallel_chunks
-
-# Resumable chunk-interruption exceptions. They are defined in
-# ``dataretrieval.ogc.interruptions`` rather than ``dataretrieval.exceptions``
-# because they carry pandas/httpx state and a resumable ``ChunkedCall`` handle,
+# Resumable fan-out interruption exceptions. They are defined in
+# ``dataretrieval.interruptions`` rather than ``dataretrieval.exceptions``
+# because they carry pandas/httpx state and a resumable ``FanOut`` handle,
 # which would pull heavy dependencies into the lightweight exceptions module.
-# Surfaced here so callers get a stable public path:
-# ``from dataretrieval import ChunkInterrupted``.
-from dataretrieval.ogc.interruptions import (
+# They are not under ``ogc`` because Water Use raises them too. Surfaced here so
+# callers get a stable public path: ``from dataretrieval import ChunkInterrupted``.
+from dataretrieval.interruptions import (
     ChunkInterrupted,
+    FanOutInterrupted,
     QuotaExhausted,
     ServiceInterrupted,
 )
 
+# Parallel-chunks control (a context manager). Defined with the chunker in
+# ``dataretrieval.ogc.chunking``; surfaced here for a stable public path
+# ``from dataretrieval import parallel_chunks``.
+from dataretrieval.ogc.chunking import parallel_chunks
+
 from . import (
     exceptions,
     ngwmn,
@@ -96,8 +97,9 @@
     "TransientError",
     "URLTooLong",
     "Unchunkable",
-    # resumable chunk-interruption exceptions (defined in ogc.interruptions)
+    # resumable fan-out interruption exceptions (defined in interruptions)
     "ChunkInterrupted",
+    "FanOutInterrupted",
     "QuotaExhausted",
     "ServiceInterrupted",
     # parallel-chunks control (defined in ogc.chunking)

dataretrieval/interruptions.py

@@ -0,0 +1,299 @@
+"""Resumable fan-out interruption exceptions — the public resume contract.
+
+When a fanned-out request fails mid-stream (a 429, a 5xx, or a bare transport
+error), the work already completed is preserved and the call is resumable: the
+raised exception carries a ``.call`` handle whose ``resume()`` re-issues only
+the still-pending sub-requests. These exception types are that contract,
+re-exported at the top level (``from dataretrieval import ChunkInterrupted``).
+The execution machinery that raises and resumes them is
+:class:`dataretrieval.transport.fanout.FanOut`.
+
+Vocabulary, consistently: a **fan-out** is one logical query the service forces
+into several requests; a **sub-request** is one unit of a fan-out; a **chunk**
+is specifically a *byte-driven* slice, which is OGC planning vocabulary and
+belongs to :class:`~dataretrieval.ogc.planning.ChunkPlan`. Water Use fans out
+without chunking anything — the NWDC simply accepts one location per request —
+so the base class is :class:`FanOutInterrupted`.
+
+``ChunkInterrupted`` is retained as an alias of that same class, not a
+deprecated shim to delete later: it is the name published in the user guide and
+caught in user code, and aliasing costs nothing to keep. ``except
+ChunkInterrupted`` and ``except FanOutInterrupted`` are the same handler.
+
+This is a top-level leaf rather than a member of ``ogc`` or ``transport``,
+for the reason ADR 0006 gives for ``combining``, ``progress``, and
+``credentials``: adapters need it whether or not they go through transport, and
+an exception taxonomy is not HTTP execution policy. It stays out of
+:mod:`dataretrieval.exceptions` because it carries pandas/httpx state, which
+would pull heavy dependencies into that lightweight leaf.
+"""
+
+from __future__ import annotations
+
+import socket
+from typing import TYPE_CHECKING, Any, ClassVar
+
+import httpx
+import pandas as pd
+
+from dataretrieval.exceptions import DataRetrievalError, RateLimited, TransientError
+
+if TYPE_CHECKING:
+    from dataretrieval.transport.fanout import FanOut
+
+
+class FanOutInterrupted(DataRetrievalError):
+    """
+    Base class for mid-stream sub-request failures whose completed work
+    is preserved and resumable.
+
+    A ``FanOutInterrupted`` subclass means: a sub-request failed, but
+    ``FanOut`` still owns whatever completed successfully before
+    the failure. Call ``self.call.resume()`` to pick up where the
+    failure stopped you — only still-pending sub-requests are
+    re-issued.
+
+    Subclasses describe *why* ``FanOut`` stopped so callers can
+    pick a retry policy: :class:`QuotaExhausted` for 429 (wait for the
+    rate-limit window), :class:`ServiceInterrupted` for 5xx (wait for
+    the upstream to recover). The ``.call`` handle is the same object
+    across every interruption of a single fanned-out call — frames
+    accumulate across retries.
+
+    Attributes
+    ----------
+    call : FanOut or None
+        Resumable handle into the ``FanOut`` that raised this
+        exception. ``None`` only on hand-constructed exceptions (test
+        fixtures), where ``.call``-derived accessors degrade to
+        empty/``None``.
+    retry_after : float or None
+        Seconds the server suggested waiting (``Retry-After`` header).
+        ``None`` when the server gave no hint.
+    completed_chunks : int
+        Number of sub-requests successfully completed before the failure.
+    total_chunks : int
+        Total sub-requests in the plan.
+    partial_frame : pandas.DataFrame
+        Combined frame of work completed by the moment this exception
+        was raised. Snapshot at raise time — does NOT advance on a
+        later ``call.resume()`` (use ``exc.call.partial_frame`` for
+        the live view).
+    partial_response : httpx.Response or None
+        Raw aggregate response covering the completed sub-requests at
+        raise time; ``None`` if nothing had completed yet. Same snapshot
+        semantics as ``partial_frame``. (Raw, not finalized — use
+        ``exc.call.resume()`` for the finalized ``(df, metadata)`` result.)
+
+    Examples
+    --------
+    Retry on any transient interruption, honoring the server's
+    ``Retry-After`` hint when present and falling back to a fixed wait
+    otherwise. Each new interruption keeps the already-completed work
+    intact — only the still-pending sub-requests are re-issued.
+
+    .. code-block:: python
+
+        import time
+        from dataretrieval import ChunkInterrupted
+
+        # ``getter`` is any chunked OGC getter — e.g.
+        # ``waterdata.get_daily`` or ``ngwmn.get_water_level``.
+        try:
+            df, md = getter(monitoring_location_id=long_list_of_sites)
+        except ChunkInterrupted as exc:
+            while True:
+                time.sleep(exc.retry_after or 5 * 60)
+                try:
+                    df, md = exc.call.resume()
+                    break
+                except ChunkInterrupted as next_exc:
+                    exc = next_exc
+    """
+
+    # Subclasses override with a ``str.format`` template; the format
+    # call sees ``completed_chunks`` and ``total_chunks`` as kwargs.
+    _MESSAGE_TEMPLATE: ClassVar[str] = (
+        "Chunked request interrupted after {completed_chunks}/"
+        "{total_chunks} sub-requests; call .call.resume() to continue."
+    )
+
+    def __init__(
+        self,
+        *,
+        completed_chunks: int,
+        total_chunks: int,
+        call: FanOut | None = None,
+        retry_after: float | None = None,
+        cause: BaseException | None = None,
+    ) -> None:
+        message = self._MESSAGE_TEMPLATE.format(
+            completed_chunks=completed_chunks, total_chunks=total_chunks
+        )
+        if cause is not None:
+            cause_msg = str(cause) or type(cause).__name__
+            message = f"{message} Cause: {type(cause).__name__}: {cause_msg}"
+        super().__init__(message)
+        self.completed_chunks = completed_chunks
+        self.total_chunks = total_chunks
+        self.call = call
+        self.retry_after = retry_after
+        # Snapshot partial state at raise time so the exception stays a stable
+        # record of the failure moment: ``exc.partial_frame`` /
+        # ``.partial_response`` do NOT advance on a later ``call.resume()``
+        # (that live view is on ``call.partial_frame`` / ``.partial_response``).
+        # This keeps each interruption in a resume loop a faithful record of
+        # what it saw, rather than every exception aliasing the shared call's
+        # advancing state. ``.copy()`` guards the single-chunk fast path, where
+        # the combined frame may be returned verbatim.
+        if call is None:
+            self.partial_frame: pd.DataFrame = pd.DataFrame()
+            self.partial_response: httpx.Response | None = None
+        else:
+            self.partial_frame = call.partial_frame.copy()
+            self.partial_response = call.partial_response
+
+    def __getstate__(self) -> dict[str, Any]:
+        # Drop the live FanOut before pickling: its ``.fetch`` is an
+        # undecorated module function pickle can't reference by name, so the
+        # interruption can't cross a process boundary with ``.call`` attached.
+        # The degraded ``call=None`` form keeps the counts, retry hint, and the
+        # snapshotted partial frame / response — plain instance attributes the
+        # base ``__getstate__`` already pickles; only ``.resume()`` is lost
+        # (cross-process resume was never possible anyway).
+        return {**super().__getstate__(), "call": None}
+
+
+class QuotaExhausted(FanOutInterrupted):
+    """
+    A sub-request returned HTTP 429 — the per-key rate-limit window
+    is exhausted. Subclass of :class:`FanOutInterrupted`.
+
+    The completed sub-requests are preserved on ``.call``; once the
+    rate-limit window resets, ``.call.resume()`` re-issues only the
+    still-pending work. ``partial_frame`` holds what completed
+    before the 429.
+    """
+
+    _MESSAGE_TEMPLATE = (
+        "HTTP 429 after {completed_chunks}/{total_chunks} sub-requests; "
+        "catch QuotaExhausted (or FanOutInterrupted) to access "
+        ".partial_frame or .call.resume() once the rate-limit "
+        "window has rolled over."
+    )
+
+
+class ServiceInterrupted(FanOutInterrupted):
+    """
+    A sub-request returned HTTP 5xx — the upstream service failed
+    transiently. Subclass of :class:`FanOutInterrupted`.
+
+    The completed sub-requests are preserved on ``.call``; once the
+    upstream recovers, ``.call.resume()`` resumes only the
+    still-pending work.
+    """
+
+    _MESSAGE_TEMPLATE = (
+        "Service error after {completed_chunks}/{total_chunks} "
+        "sub-requests; catch ServiceInterrupted (or FanOutInterrupted) "
+        "and call .call.resume() once the upstream service recovers."
+    )
+
+
+# Resolver failures that will not resolve differently on a later attempt. The
+# temporary ones (notably EAI_AGAIN -- "try again", raised while a resolver is
+# still coming up, on VPN reconnect, or after a laptop wakes) are deliberately
+# absent: those are worth another try. Looked up defensively because the EAI_*
+# constants are platform-dependent; an unrecognized code stays retryable, since
+# spending a few seconds on a retry is cheaper than dropping a recoverable call.
+_PERMANENT_DNS_ERRORS = frozenset(
+    code
+    for code in (
+        getattr(socket, name, None) for name in ("EAI_NONAME", "EAI_FAIL", "EAI_NODATA")
+    )
+    if code is not None
+)
+
+
+def _deterministic_failure(exc: BaseException) -> bool:
+    """Whether a transport failure would fail identically on every retry.
+
+    An unsupported scheme or a request we built wrong is settled before a byte
+    goes out, and a hostname the resolver rejects outright won't be accepted on
+    the next attempt either -- so retrying only delays the error the caller
+    needs. A *temporary* resolver failure is not in that class and stays
+    retryable (see :data:`_PERMANENT_DNS_ERRORS`).
+
+    The original failure is several layers down and not always an explicit
+    ``raise ... from``: a DNS failure reaches us as ``NetworkError`` ->
+    ``httpx.ConnectError`` -> ``httpcore.ConnectError`` -> ``socket.gaierror``,
+    linked by ``__context__`` (implicit chaining) rather than ``__cause__``.
+
+    Both links of every frame are visited, not just the first one present. A
+    frame can carry an explicit ``__cause__`` *and* an unrelated ``__context__``
+    (any ``raise X from Y`` inside an ``except`` block produces exactly that), so
+    following only the cause would walk off down the explicit branch and miss a
+    ``gaierror`` sitting on the implicit one -- spending the whole retry budget
+    on a hostname that will never resolve. The ``seen`` set keeps a chain that
+    rejoins itself, or points back at an ancestor, from looping.
+    """
+    seen: set[int] = set()
+    pending: list[BaseException | None] = [exc]
+    while pending:
+        current = pending.pop()
+        if current is None or id(current) in seen:
+            continue
+        seen.add(id(current))
+        if isinstance(current, (httpx.UnsupportedProtocol, httpx.LocalProtocolError)):
+            return True
+        if isinstance(current, socket.gaierror):
+            # Return, not continue: the first resolver code found settles the chain.
+            return current.errno in _PERMANENT_DNS_ERRORS
+        pending += [current.__cause__, current.__context__]
+    return False
+
+
+def _classify_transient(
+    exc: BaseException,
+) -> tuple[type[FanOutInterrupted], float | None] | None:
+    """Classify one failure as a resumable interruption."""
+    if isinstance(exc, RateLimited):
+        return QuotaExhausted, exc.retry_after
+    if isinstance(exc, TransientError):
+        return ServiceInterrupted, exc.retry_after
+    if isinstance(exc, (httpx.HTTPError, httpx.InvalidURL)):
+        # Some failures will fail the same way every time -- a bad scheme, a
+        # hostname that doesn't resolve. Offering to resume one would just
+        # hide the real error behind a retry that can never work.
+        if _deterministic_failure(exc):
+            return None
+        return ServiceInterrupted, None
+    return None
+
+
+def _classify_chunk_error(
+    exc: BaseException,
+) -> tuple[type[FanOutInterrupted], float | None] | None:
+    """Walk a wrapped pagination failure for a resumable transport cause."""
+    current: BaseException | None = exc
+    while current is not None:
+        result = _classify_transient(current)
+        if result is not None:
+            return result
+        current = current.__cause__
+    return None
+
+
+#: The name this taxonomy was published under, kept as a permanent alias so
+#: ``except ChunkInterrupted`` keeps working. Same class object, not a subclass.
+ChunkInterrupted = FanOutInterrupted
+
+__all__ = [
+    "ChunkInterrupted",
+    "_deterministic_failure",
+    "FanOutInterrupted",
+    "QuotaExhausted",
+    "ServiceInterrupted",
+    "_classify_chunk_error",
+    "_classify_transient",
+]