Fix AsyncFileSystem._cat_ranges error handling with on_error parameter

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

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
 # ---------------------------------------------------------------------------