Wrap potential error in the _cat_file_concurrent

[clacrow]

I'm using zarr with s3 and s3fs store backend. I have a encountered problems with _cat_file_concurrent method, with large load it can throw an AioTimeoutError not respecting s3fs retries. Adding a missing _error_wrapper in the place of failure.

[martindurant]

Am I understanding, that in your situation the initial _call_s3 succeeds, but then the subsequent real stalls? I would not have thought that was retriable - are you finding (via logging) that you do sometimes succeed on second/third calls?

[martindurant]

Perhaps this could be tested via a mock on the read method, to produce a timeout ont he first call.

[clacrow]

Yes, something like that. it happens only on a large workload to 5-10 out of 60 jobs. We had different issues and we inserted custom error handler for reties of timeouts, but it did not catch that. After that change the problem disappears.

I haven't run the fix with a debug log level yet because the logs are GBs in size, but I will and let you know :)

Ending of the trace:

  File "<redacted>/.pixi/envs/default/lib/python3.14/site-packages/zarr/storage/_fsspec.py", line 289, in get
    value = prototype.buffer.from_bytes(await self.fs._cat_file(path))
                                        ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "<redacted>/.pixi/envs/default/lib/python3.14/site-packages/s3fs/core.py", line 1327, in _cat_file
    return await self._cat_file_concurrent(
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
    ...<6 lines>...
    )
    ^
  File "<redacted>/.pixi/envs/default/lib/python3.14/site-packages/s3fs/core.py", line 1368, in _cat_file_concurrent
    results = await asyncio.gather(
              ^^^^^^^^^^^^^^^^^^^^^
    ...<4 lines>...
    )
    ^
  File "<redacted>/.pixi/envs/default/lib/python3.14/site-packages/s3fs/core.py", line 1359, in _read_chunk
    data = await resp["Body"].read()
           ^^^^^^^^^^^^^^^^^^^^^^^^^
  File "<redacted>/.pixi/envs/default/lib/python3.14/site-packages/aiobotocore/response.py", line 63, in read
    raise AioReadTimeoutError(
        endpoint_url=self.__wrapped__.url, error=e
    )
aiobotocore.response.AioReadTimeoutError: Read timeout on endpoint URL: "https://<redacted>

[clacrow]

Extensive tests showed that the error in that place is only sometimes recoverable (and other times retry gives same error again and again). Due to this I am currently pivoting to retry on a higher level that will be more reliable than this solution.