Fix aget_chunk against real async filesystems

Pointed at a live S3 endpoint for the first time (moto server + s3fs), the
async path failed on every chunk:

    HTTPClientError: ... got Future <...> attached to a different loop

fsspec drives an async filesystem's coroutines on a private event loop of
its own, in a background thread.  Awaiting fs._cat_file() from the caller's
loop therefore uses a client built on one loop from another, which
aiobotocore rejects.  Its blocking API is the supported way in, so
aget_chunk now hands get_chunk to a worker thread: that call dispatches to
fsspec's own loop, so the thread parks on a queue rather than on a socket,
and afetch() keeps overlapping fetches as before.

memory:// cannot catch this -- it is not an async backend, so aget_chunk
took the fallback branch there and everything passed.  The test now asserts
the mechanism instead of only the result: afetch must reach get_chunk, which
fails if anyone awaits the filesystem coroutine again.

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

Author: FrancescAlted

bench/ndarray/fsspec-concurrency.py

@@ -15,9 +15,9 @@
 hide) but the gain never was, because nothing that runs offline has a round trip
 to hide.  This script answers it against a real endpoint.
 
-It also runs ``afetch()``, whose async path (``aget_chunk`` -> ``fs._cat_file``)
-has never executed at all: ``memory://`` is not an async backend, so only its
-blocking fallback is covered by the test suite.
+It also runs ``afetch()``, which is worth keeping in the sweep: the first time
+this script was pointed at a real S3 endpoint, that path failed outright, and
+``memory://`` cannot reproduce it because it is not an async backend.
 
 Usage
 -----
@@ -159,8 +159,7 @@ def main():
     )
 
     if not args.skip_afetch:
-        # The async path, which no test has ever run: aget_chunk reaches
-        # fs._cat_file directly rather than falling back to the blocking read
+        # The async path, which only a real async backend exercises
         report(
             "slice, afetch (async path)",
             {c: timed(args.urlpath, item, c, use_afetch=True) for c in levels},

src/blosc2/proxy.py

@@ -793,21 +793,25 @@ def get_chunk(self, nchunk: int) -> bytes:
         return data[: struct.unpack("<i", data[12:16])[0]]
 
     async def aget_chunk(self, nchunk: int) -> bytes:
-        """Same as :meth:`get_chunk`, but letting several fetches overlap.
+        """Same as :meth:`get_chunk`, but without blocking the caller's event loop.
 
         This is what makes :meth:`Proxy.afetch` worth using against an object
         store, where a slice spanning many chunks is nearly all round-trip
-        latency.  Backends without an async implementation fall back to the
-        blocking path, which costs nothing but gains nothing either.
+        latency.
+
+        The fetch goes to a worker thread rather than being awaited directly.
+        Awaiting an async filesystem's coroutine looks like the obvious thing to
+        do and does not work: fsspec drives those on a private event loop of its
+        own, so a client created there and awaited here raises "got Future
+        attached to a different loop" (seen with s3fs).  Its blocking API is the
+        supported way in, and it hands off to that same private loop, so the
+        thread parks on a queue rather than on a socket.
         """
         offset = int(self._offsets[nchunk])
         if offset < 0:
             return self._special_chunk(offset)
-        if not getattr(self._fs, "async_impl", False):
-            return self.get_chunk(nchunk)
-        end = offset + int(self._extents[nchunk])
-        data = await self._fs._cat_file(self._path, start=offset, end=end)
-        return data[: struct.unpack("<i", data[12:16])[0]]
+        loop = asyncio.get_running_loop()
+        return await loop.run_in_executor(None, self.get_chunk, nchunk)
 
     def _special_chunk(self, offset: int) -> bytes:
         """Rebuild a run-length chunk, which lives in its offset instead of the file."""

tests/test_fsspec.py

@@ -310,13 +310,27 @@ def test_lazy_fetches_only_touched_chunks(monkeypatch):
     assert fetched == [1, 2]
 
 
-def test_lazy_afetch():
+def test_lazy_afetch(monkeypatch):
     import asyncio
 
     a = blosc2.arange(0, 1000, dtype="i4", chunks=(100,))
     p = blosc2.open(_put("afetch.b2nd", a), lazy=True)
+
+    # aget_chunk must go through the blocking get_chunk in a worker thread.
+    # Awaiting an async filesystem's own coroutine instead raises "got Future
+    # attached to a different loop" on s3fs, which memory:// cannot reproduce
+    # because it is not an async backend at all
+    fetched = []
+    orig = blosc2.FsspecNDSource.get_chunk
+    monkeypatch.setattr(
+        blosc2.FsspecNDSource,
+        "get_chunk",
+        lambda self, nchunk: (fetched.append(nchunk), orig(self, nchunk))[1],
+    )
+
     cache = asyncio.run(p.afetch(slice(150, 250)))
     assert np.array_equal(cache[150:250], a[150:250])
+    assert fetched == [1, 2]
 
 
 def test_lazy_fetch_is_serial_when_asked(monkeypatch):