[1fanwang]

Tracking issue

No tracking issue.

Why are the changes needed?

A concurrent S3 download can report success after a ranged request fails, leaving the caller with a partial file and a background Task exception was never retrieved warning.

What changes were proposed in this pull request?

Propagate the failed request to the caller, then cancel and await the remaining download tasks.

How was this patch tested?

Testing Done

The reproducer uses the real s3fs and aiobotocore HTTP path against a local aiohttp server. One range returns HTTP 403 while another remains open. The failed download raises PermissionError, and the other connection closes. This does not use an S3 service.

Scenario	Command	Result
Failed range with a concurrent request	python async_download_repro.py	PermissionError reaches the caller; the other connection closes

Reproducer source: async_download_repro.py

import asyncio
from http import HTTPStatus
from pathlib import Path
from tempfile import TemporaryDirectory

from aiohttp import hdrs, web
from aiohttp.test_utils import TestServer

from flytekitplugins.async_fsspec import AsyncS3FileSystem


async def main() -> None:
    ready = asyncio.Event()
    peer_closed: asyncio.Future[bool] = asyncio.get_running_loop().create_future()

    async def serve(request: web.Request) -> web.StreamResponse:
        if request.method == hdrs.METH_HEAD:
            return web.Response(headers={hdrs.CONTENT_LENGTH: "8", hdrs.ETAG: '"x"'})
        print(f"{request.method} range={request.headers[hdrs.RANGE]}")
        if request.headers[hdrs.RANGE] == "bytes=0-3":
            await ready.wait()
            print("response_status=403")
            return web.Response(
                status=HTTPStatus.FORBIDDEN,
                body=b"<Error><Code>AccessDenied</Code><Message>injected ranged failure</Message></Error>",
                content_type="application/xml",
            )
        response = web.StreamResponse(
            status=HTTPStatus.PARTIAL_CONTENT,
            headers={hdrs.CONTENT_RANGE: "bytes 4-7/8"},
        )
        response.enable_chunked_encoding()
        await response.prepare(request)
        await response.write(b"EFGH")
        ready.set()
        transport = request.transport
        try:
            await asyncio.sleep(1)
        finally:
            peer_closed.set_result(transport is None or transport.is_closing())
        try:
            await response.write_eof()
        except (ConnectionError, RuntimeError):
            pass
        return response

    app = web.Application()
    app.router.add_route(hdrs.METH_ANY, "/{path:.*}", serve)
    async with TestServer(app) as server:
        fs = AsyncS3FileSystem(
            anon=True,
            asynchronous=True,
            client_kwargs={"endpoint_url": str(server.make_url("/"))},
            config_kwargs={"s3": {"addressing_style": "path"}},
        )
        with TemporaryDirectory() as directory:
            output = Path(directory) / "partial.bin"
            try:
                await fs._get_file(
                    rpath="s3://b/k",
                    lpath=str(output),
                    chunksize=4,
                    concurrent_download=2,
                )
            except PermissionError as error:
                print(f"{type(error).__name__}: {error}")
            else:
                raise AssertionError("the partial download returned success")
            print(f"partial_hex={output.read_bytes().hex()}")
            closed: bool = await asyncio.wait_for(fut=peer_closed, timeout=5)
            print(f"sibling_connection_closed={closed}")
            assert closed
        await fs._s3creator.__aexit__(None, None, None)


asyncio.run(main())

python async_download_repro.py

GET range=bytes=0-3
GET range=bytes=4-7
response_status=403
PermissionError: injected ranged failure
partial_hex=0000000045464748
sibling_connection_closed=True
exit_code=0

The earlier comparison returned a partial file without an error on the base:

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

Setup process

Python 3.12 with the SDK development dependencies and async-fsspec plugin source on PYTHONPATH.

Screenshots

Not applicable.

Check all the applicable boxes

• I updated the documentation accordingly.
• All new and existing tests passed.
• All commits are signed-off.

Related PRs

None.

Docs link

No separate documentation change.