Cancelled AioS3FileSystem operations keep retrying in executor threads #1203

Part of #1205 (step 1).

Problem

AioS3FileSystem runs the sync S3FileSystem operations in its executor threads (S3AioExecutor), and each S3 request is retried with retry_api_call() inside S3Core. If the awaiting task is cancelled, the thread keeps retrying a throttled request. It sends the remaining attempts and waits through the backoff, which with the default RetryConfig is 8 attempts and about 127 seconds plus jitter. Nobody awaits the result.

Expected: after cancellation, no further S3 attempt is made and the backoff wait ends.

Reproduction

No AWS calls. HeadObject and ListObjectsV2 are stubbed to raise ThrottlingException, and RetryConfig(multiplier=0) removes the waits.

import asyncio
import time
from unittest.mock import MagicMock

from botocore.exceptions import ClientError

from pyathena.filesystem.s3_async import AioS3FileSystem
from pyathena.util import RetryConfig

THROTTLE = ClientError({"Error": {"Code": "ThrottlingException", "Message": "x"}}, "HeadObject")
calls = []


def throttled(**kwargs):
    calls.append(1)
    time.sleep(0.02)
    raise THROTTLE


async def main():
    connection = MagicMock()
    connection.s3_client.head_object.side_effect = throttled
    connection.s3_client.list_objects_v2.side_effect = throttled
    connection.retry_config = RetryConfig(multiplier=0)
    fs = AioS3FileSystem(connection=connection, skip_instance_cache=True)
    task = asyncio.create_task(fs._info("s3://bucket/key"))
    await asyncio.sleep(0.01)
    task.cancel()
    await asyncio.gather(task, return_exceptions=True)
    print(len(calls))  # 1 at cancellation
    await asyncio.sleep(1)
    print(len(calls))  # 8: the executor thread kept retrying


asyncio.run(main())

Environment

• PyAthena master at b2d7aa2
• Python 3.15

Proposed fix (optional)

Once retry_api_call() can take a stop condition (#1202), S3AioExecutor can set that condition for the work of a cancelled task. The condition would then reach S3Core requests running in its threads, for example through a context variable copied into the executor thread. Validate with unit tests that cancel AioS3FileSystem operations against stubbed throttling responses.

Scope extended by #1205

This issue is step 1 of #1205 for the S3 filesystem. It uses the stop signal from #1202. Besides the case above, it covers the following:

• S3ThreadPoolExecutor.submit() does not copy the submitter's context (pyathena/filesystem/s3_executor.py:127-128). Part uploads and range reads therefore would not see the stop. S3AioThreadPool and S3AioExecutor already propagate it.

• S3File.discard(), the abort paths, and AioS3StreamedFile._discard() wait for running parts to finish their retry chain (pyathena/filesystem/s3.py:3348, 1871, 1722; pyathena/filesystem/s3_async.py:1604). The abort went out 4.5 s (sync) and 3.7 s (async) after the first failure with attempt=3, multiplier=1. That is about 127 s with the defaults.

• Sibling work is not cancelled after a failure:

  • S3File range reads (pyathena/filesystem/s3.py:3435-3442, 3038; pyathena/filesystem/s3_async.py:1400)
  • recursive chmod (pyathena/filesystem/s3.py:2298-2304)
  • a cancelled async multipart copy when its max_workers exceeds the shared pool (pyathena/filesystem/s3_async.py:616-728).

  ReadAheadStream cancels queued reads (pyathena/filesystem/reader.py:147).

• Protected cleanup (aborts and discards) masks the stop, so that cleanup always completes.