fix(chunking): Propagate CancelledError ahead of transient siblings

`_fan_out_rest` previously walked `failures` looking for the first
recognized transient and wrapped it as `ChunkInterrupted`. If a
sibling sub-request happened to raise `asyncio.CancelledError`
alongside (or after) a transient HTTPError, the transient was
classified first and the cancellation was silently consumed — the
user's stop signal got rewritten as a "retry me" handle.

Adds a pre-walk that propagates any non-`Exception` BaseException
(`CancelledError`, `KeyboardInterrupt`, `SystemExit`) unmodified
before the transient-classification walk runs. asyncio's
cancellation contract takes precedence over the chunker's retry
semantics.

Regression test: `test_async_fan_out_cancellation_wins_over_transient_sibling`
deterministically sequences a 3-call fan-out (probe + transient +
CancelledError) by relying on gather's submission-order scheduling
when the inner fetch_async has no awaits. Pre-fix the test got
`ServiceInterrupted`; post-fix it gets `CancelledError`.

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>

Author: thodson-usgs

dataretrieval/waterdata/chunking.py

@@ -1462,18 +1462,29 @@ async def _fan_out_rest(
 
     Completed pairs survive a sibling's transient failure via
     ``return_exceptions=True``, so the partial result stays
-    recoverable through :meth:`ChunkedCall.resume`. On any failure,
-    prefer raising the first *recognized transient* — so the user
-    still gets a resumable :class:`ChunkInterrupted` even when a
-    non-transient bug landed earlier in submission order. Fall back
-    to the first failure (preserving its type) when nothing is
-    transient.
+    recoverable through :meth:`ChunkedCall.resume`. Failure
+    precedence, in order:
+
+    1. **Cancellation / interrupt signals** (``asyncio.CancelledError``,
+       ``KeyboardInterrupt``, ``SystemExit`` — anything that isn't an
+       :class:`Exception` subclass) propagate unmodified. Wrapping
+       them as a transient :class:`ChunkInterrupted` would silently
+       swallow the user's stop signal in favor of a resumable handle.
+    2. **Recognized transients** (``RateLimited``, ``ServiceUnavailable``,
+       bare ``httpx.HTTPError``) wrap as :class:`ChunkInterrupted`, so
+       the user still gets a resumable handle even when a non-transient
+       failure landed earlier in submission order.
+    3. **Otherwise** raise the first failure (submission order),
+       preserving its type.
     """
     results = await asyncio.gather(
         *(track(i, args) for i, args in enumerate(sub_args_rest, start=1)),
         return_exceptions=True,
     )
     failures = [r for r in results if isinstance(r, BaseException)]
+    for exc in failures:
+        if not isinstance(exc, Exception):
+            raise exc
     for exc in failures:
         if (interrupted := call.wrap_failure(exc)) is not None:
             raise interrupted from exc

tests/waterdata_chunking_test.py

@@ -1464,6 +1464,43 @@ def fetch(args):
     assert len(df) == len(sync_calls)
 
 
+def test_async_fan_out_cancellation_wins_over_transient_sibling(monkeypatch):
+    """``asyncio.CancelledError`` raised by any sub-request must
+    propagate unmodified, even when a sibling raises a recognized
+    transient (which would otherwise wrap as a resumable
+    :class:`ChunkInterrupted`). Cancellation is asyncio's abort
+    signal — letting a transient-classification path consume it
+    would silently swallow the user's stop request.
+
+    fetch_async has no ``await`` inside its body, so gather schedules
+    the tasks in submission order and each runs synchronously to its
+    raise — making ``call_count`` deterministic for this test:
+    1 = probe, 2 = first fan-out task (transient), 3 = second
+    fan-out task (cancellation).
+    """
+    call_count = {"async": 0}
+
+    async def fetch_async(args):
+        call_count["async"] += 1
+        if call_count["async"] == 1:
+            return pd.DataFrame({"id": [_atom_id(args)]}), _ok_response(remaining=99)
+        if call_count["async"] == 2:
+            raise ServiceUnavailable("503: transient sibling")
+        if call_count["async"] == 3:
+            raise asyncio.CancelledError("user cancel")
+        return pd.DataFrame({"id": [_atom_id(args)]}), _ok_response(remaining=99)
+
+    fetch = _async_chunked_fetch(monkeypatch, fetch_async)
+
+    # 8 × 20-byte sites force the planner to >=3 sub-args under
+    # url_limit=240, so the fan-out gather sees at least the
+    # transient (call 2) AND the cancellation (call 3).
+    sites = [f"S{i}" * 10 for i in range(1, 9)]
+
+    with pytest.raises(asyncio.CancelledError):
+        fetch({"sites": sites})
+
+
 def test_combine_chunk_responses_does_not_mutate_input_urls():
     """Regression for the _set_response_url aliasing bug.