From f764b05467507be94c1894eb0a94cee0c0ece64a Mon Sep 17 00:00:00 2001
From: zhixiangli <zhixiangli@google.com>
Date: Tue, 9 Jun 2026 21:47:45 +0800
Subject: [PATCH] Fix async cat ranges on error

---
 fsspec/asyn.py             | 12 ++++++++++--
 fsspec/tests/test_async.py | 27 +++++++++++++++++++++++++++
 2 files changed, 37 insertions(+), 2 deletions(-)

diff --git a/fsspec/asyn.py b/fsspec/asyn.py
index dc19ceb12..f6a58fc2a 100644
--- a/fsspec/asyn.py
+++ b/fsspec/asyn.py
@@ -528,8 +528,11 @@ async def _cat_ranges(
         starts, ends: int or list
             Bytes limits of the read. If using a single int, the same value will be
             used to read all the specified files.
+        on_error: "return" or "raise"
+            If "return" (default), any per-range exception is placed in the output
+            list at the corresponding position. Otherwise the first such exception
+            is raised. Matches ``AbstractFileSystem.cat_ranges``.
         """
-        # TODO: on_error
         if max_gap is not None:
             # use utils.merge_offset_ranges
             raise NotImplementedError
@@ -546,9 +549,14 @@ async def _cat_ranges(
             for p, s, e in zip(paths, starts, ends)
         ]
         batch_size = batch_size or self.batch_size
-        return await _run_coros_in_chunks(
+        out = await _run_coros_in_chunks(
             coros, batch_size=batch_size, nofiles=True, return_exceptions=True
         )
+        if on_error != "return":
+            ex = next(filter(is_exception, out), None)
+            if ex is not None:
+                raise ex
+        return out
 
     async def _put_file(self, lpath, rpath, mode="overwrite", **kwargs):
         raise NotImplementedError
diff --git a/fsspec/tests/test_async.py b/fsspec/tests/test_async.py
index e7363bc8b..bd0593347 100644
--- a/fsspec/tests/test_async.py
+++ b/fsspec/tests/test_async.py
@@ -254,6 +254,33 @@ def test_rm_file_without_implementation():
         fs.rm_file("test/file.txt")
 
 
+class _CatRangesFS(fsspec.asyn.AsyncFileSystem):
+    # Mirrors the gcsfs/s3fs pattern: overrides _cat_file, inherits _cat_ranges.
+    cachable = False
+    store = {"ok": b"0123456789"}
+
+    async def _cat_file(self, path, start=None, end=None, **kwargs):
+        if path not in self.store:
+            raise FileNotFoundError(path)
+        return self.store[path][start:end]
+
+
+def test_cat_ranges_on_error_return_keeps_exceptions_inline():
+    # Default on_error="return": a failed range comes back as an exception value,
+    # positionally aligned with the inputs (matches AbstractFileSystem.cat_ranges).
+    fs = _CatRangesFS()
+    out = fs.cat_ranges(["ok", "missing"], [0, 0], [4, 4])
+    assert out[0] == b"0123"
+    assert isinstance(out[1], FileNotFoundError)
+
+
+def test_cat_ranges_on_error_raise_propagates():
+    # on_error="raise" must raise the first error instead of returning it inline.
+    fs = _CatRangesFS()
+    with pytest.raises(FileNotFoundError):
+        fs.cat_ranges(["ok", "missing"], [0, 0], [4, 4], on_error="raise")
+
+
 # ---------------------------------------------------------------------------
 # Tests for the prefix= hint that _glob passes to _find
 # ---------------------------------------------------------------------------