fix(async-fsspec): propagate concurrent download failures

[1fanwang]

Tracking issue

None.

Why are the changes needed?

Concurrent S3 downloads could return success after a ranged request failed. The caller received a partial file while asyncio logged Task exception was never retrieved.

What changes were proposed in this pull request?

Read completed task results so download errors reach the caller. On failure or cancellation, cancel and await the remaining chunk tasks.

How was this patch tested?

This command uses the real s3fs and aiobotocore HTTP path:

Setup process

PYTHONPATH=plugins/flytekit-async-fsspec python - <<'PY'
import asyncio
from pathlib import Path
from aiohttp import web
from flytekitplugins.async_fsspec import AsyncS3FileSystem
w=asyncio.Event()
async def h(r):
 if r.method=="HEAD": return web.Response(headers={"Content-Length":"8","ETag":'"x"'})
 if r.headers["Range"]=="bytes=0-3":
  await w.wait(); return web.Response(status=403,body=b"<Error><Code>AccessDenied</Code><Message>injected ranged failure</Message></Error>",content_type="application/xml")
 x=web.StreamResponse(status=206,headers={"Content-Range":"bytes 4-7/8"});x.enable_chunked_encoding();await x.prepare(r);await x.write(b"EFGH");w.set();t=r.transport;await asyncio.sleep(1);print("closed",t is None or t.is_closing())
 try: await x.write_eof()
 except (ConnectionError,RuntimeError): pass
 return x
async def main():
 a=web.Application();a.router.add_route("*","/{p:.*}",h);z=web.AppRunner(a);await z.setup();s=web.TCPSite(z,"127.0.0.1",0);await s.start();port=s._server.sockets[0].getsockname()[1]
 f=AsyncS3FileSystem(anon=True,asynchronous=True,client_kwargs={"endpoint_url":f"http://127.0.0.1:{port}"},config_kwargs={"s3":{"addressing_style":"path"}});p=Path("partial.bin");p.unlink(missing_ok=True)
 try: await f._get_file("s3://b/k",str(p),chunksize=4,concurrent_download=2);print("returned")
 except Exception as e: print(type(e).__name__,e)
 print(p.read_bytes().hex());await asyncio.sleep(1);await f._s3creator.__aexit__(None,None,None);await z.cleanup()
asyncio.run(main())
PY

master: returned; 0000000045464748; closed False
branch: PermissionError injected ranged failure; 0000000045464748; closed True

The expected file was 4142434445464748. Plugin tests pass.

Screenshots

Not applicable.

Check all the applicable boxes

• Documentation is unchanged.
• All async fsspec plugin tests passed.
• All commits are signed off.

Related PRs

None.

Docs link

Not applicable.