Stop the S3 retries of cancelled calls and of failed requests' siblings

[laughingman7743]

WHAT

Step 1 of #1205 for the S3 filesystem. It uses the stop scopes of #1223 (stop_retries_on(), unstoppable_retries()).

1. Cancelled AioS3FileSystem calls (pyathena/filesystem/s3_executor.py)

• S3AioThreadPool.run(), through which AioS3FileSystem runs its blocking calls, opens a stop scope whose event a cancellation of the waiting task sets. The S3 retries of the running call make no further attempt, and their backoff wait ends. A cancellation that arrives after the call has returned sets nothing, so work that the call left running, such as the parts of a write in a transaction, keeps its retries. The call records its end under a lock that the cancellation also takes. run_shielded() keeps its retries.
• S3ThreadPoolExecutor.submit() runs each callable in a copy of the submitter's context, as S3AioThreadPool.submit() already did. As a result, the scope reaches the part uploads and range reads that a call runs in parallel, for example the multipart path of a cancelled _pipe_file().
• S3AioExecutor already ran its functions in the submitter's context: run_coroutine_threadsafe() schedules its task with call_soon_threadsafe(), which uses the current context of the calling thread. A comment and a test now record this. It dispatches through its own _dispatch() coroutine instead of S3AioThreadPool.run(), so that the shutdown of the event loop, which cancels the dispatch tasks, does not stop the retries of running parts, as before.
• The S3Executor.submit() docstring states the requirement for custom executors: run the callable in a copy of the submitter's context.

2. Running siblings of a failed request

• Multipart writes (S3File, AioS3StreamedFile): the CreateMultipartUpload request, the part uploads, and the part copies run in a stop scope of the file. discard(), _discard(), S3File._close_without_commit(), and AioS3StreamedFile._close_without_commit() set it before waiting for the creation and the running parts. The last one sets it on the event loop before it queues the cleanup, which a creation or parts holding every thread of the file would otherwise delay. As a result, a failed part, a failed completion, an interrupted or cancelled write, or a discard no longer waits for a throttled request to finish its retry chain before the abort. A stopped creation raises the error of its last attempt, as one whose retries ran out.
• Multipart copies (S3FileSystem and AioS3FileSystem): a failed part copy, an interrupt, or a cancellation stops the retries of the creation and of the running part copies before the abort. Free the data of failed S3 requests and clean up async copies at loop shutdown #1192 already cancels the queued ones. In an AioS3FileSystem copy, a failed annotation copy also stops the running ones. The annotation copies that wait for a thread at that point are not sent. The error of the copy that failed first is raised, not that of an annotation that its failure stopped: the failing thread records it together with the stop, under a lock.
• Parallel range reads (S3File._fetch_range(), sync): a failure stops the running reads. In AioS3StreamedFile._fetch_range(), Free the data of failed S3 requests and clean up async copies at loop shutdown #1192 cancels the read tasks, which now stops their retries through S3AioThreadPool.run().
• Recursive chmod: once an update fails, the updates that have not started are cancelled and the running ones are stopped. Before, ThreadPoolExecutor.shutdown(wait=True) ran all the queued ones. The paths are listed before the updates are submitted.
• Interrupts during submission: a recursive chmod and a sync parallel range read schedule their requests inside the handler that stops and cancels them. Each future is created before it is scheduled with submit_to(), so that a call queued when an interrupt stops the scheduling is still cancelled instead of sending its first request.

Order of cleanup: each cleanup cancels the requests that have not started before it sets the stop. The stop wakes a backing-off request, whose thread could otherwise take a queued request before it is cancelled, and that request would send its first attempt.

3. Protected aborts: S3Core.abort_multipart_upload(), through which every abort goes (S3MultipartWriter.abort(), S3FileSystem._abort_multipart_upload(), clear_multipart_uploads()), runs its request in unstoppable_retries(). A stop of the work that an abort cleans up, or of its caller, therefore does not stop the abort's retries. That includes the aborts of a cancelled AioS3FileSystem.clear_multipart_uploads() call.

The requests that keep their retries after a cancellation are documented in docs/aio.md: the commit of a file, the CompleteMultipartUpload request of a multipart copy, and the aborts.

Not changed:

• Pools that read ahead (ReadAheadStream, Arrow's Parquet prefetch, fsspec's background cache) do not copy the submitter's context, so a stop does not reach them. Cancelled AioS3FileSystem operations keep retrying in executor threads #1203 suggested clearing the scope for their workers on builds where new threads inherit their creator's context. That is not needed now: PyAthena opens scopes only around single API calls, constructions in create_in_thread() (whose object is closed on cancellation), and single AioS3FileSystem calls, which leave no read-ahead running after them.

Behavior changes for the release notes:

• Cancelling an AioS3FileSystem coroutine stops the retries of its S3 requests in the filesystem's threads. A request already sent still runs to its end.
• A failed part, completion, or part copy no longer waits for running throttled parts to exhaust their retries before the abort.
• A recursive chmod stops after the first failed update: the keys that it has not updated keep their ACL. Before, the queued updates still ran.

WHY

Part of #1205 (step 1). Closes #1203.

Split from #1202 (PR #1223) because #1192 rewrote the same executor, range-read, and copy code. This PR is based on master after both.

TEST

Tested on Python 3.15 (macOS).

• just lint, just docs lint: passed.
• uv run --env-file .env pytest -n 4 tests/pyathena/filesystem/ (including the S3 integration tests) at the head: 1312 passed, 9 skipped. Not run locally: Python versions other than 3.15, which CI runs.
• New regression tests. They use stubs that send a throttled operation with retry_api_call() in the calling thread, as S3Core.call() does, with a backoff of at least EVENT_TIMEOUT (10) seconds that only the stop can shorten. With the source changes reverted to master, all fail except those that confirm behavior that already held: S3AioThreadPool and S3AioExecutor context propagation, run_shielded(), test_run_cancelled_after_return (which guards against a regression that an earlier revision of this PR introduced; the lock that closes its race window is not deterministically testable), and the [a1] order of the annotation test.
  • TestS3Executor.test_submit_context[S3ThreadPoolExecutor|S3AioThreadPool], TestS3AioExecutor.test_submit_context, TestS3AioThreadPool.test_run_cancelled_while_retrying[False|True], test_run_cancelled_after_return
  • TestS3FileSystem.test_failed_read_stops_running_reads, test_read_interrupted_submission_stops_running_reads, test_chmod_recursive_failure_stops_running_requests, test_chmod_recursive_interrupted_submission_stops_running_requests, test_copy_object_with_multipart_upload_failed_part_stops_running_parts, test_abort_multipart_upload_in_stopped_scope
  • TestS3File.test_discard_stops_running_parts, test_interrupted_creation_stops_retries, test_discard_in_stopped_scope
  • TestAioS3FileSystem.test_cancelled_call_stops_retries, test_copy_object_with_multipart_upload_failed_part_stops_running_parts, ..._cancelled_creation_stops_retries, ..._failed_annotation_stops_running_copies[a1|a2], ..._failed_annotation_skips_queued_copies
  • TestAioS3StreamedFile.test_failed_read_stops_running_reads, test_failed_part_stops_running_parts, test_cancelled_creation_stops_retries, test_discard_in_stopped_scope

The AWS jobs run when the PR is marked Ready.

🤖 Generated with Claude Code

[laughingman7743]

Self-review round 1 (behavior and implementation): FINDINGS, repaired

Range: git diff 18c1eea17dcceb4395f95841d3a45b5b28c0e42d..41a772764690d61d61ad7183a904196b185226e0. The first pass was at 7a4e1671; the repairs are in 41a77276.
Covered: s3_executor.py (context copy in S3ThreadPoolExecutor.submit(), the S3AioThreadPool.run() stop, S3AioExecutor dispatch), s3.py (parts, creation, _close_without_commit, discard, sync copy, range reads, chmod), s3_async.py (copy parts, creation and annotations, AioS3StreamedFile parts, creation and _discard), s3_core.py (abort), docs, and tests. The rebase onto #1192's _S3WorkItem, wait_shielded(), and futures-based copy was checked hunk by hunk. Passes: a manual pass, /code-review (high), and /simplify (reuse, simplification, efficiency, altitude).

Findings and repairs:

1. CreateMultipartUpload ran outside the stop scopes, so cleanup waited for a throttled creation's full retries. Repaired: creations of writes and copies run in the upload's scope, and S3File._close_without_commit() / AioS3StreamedFile._discard() set it before waiting for the creation. Tests: test_interrupted_creation_stops_retries, test_cancelled_creation_stops_retries, ..._cancelled_creation_stops_retries.
2. After context.run was dropped (run_coroutine_threadsafe() already uses the submitter's context through call_soon_threadsafe()), S3AioExecutor dispatched through S3AioThreadPool.run(). Loop shutdown would then stop running sync parts, and each part waited on two events (0.1 s polling). Repaired: S3AioExecutor._dispatch() without a stop scope.
3. S3AioThreadPool.run() set the stop on a cancellation that arrived after the call returned, which stopped work left running, such as transaction parts. Repaired (this line); test_run_cancelled_after_return fails without the guard.
4. The running annotation copies of an async copy kept retrying after one failed. Repaired, with ..._failed_annotation_stops_running_copies.
5. Docs overclaimed which requests a cancellation stops. Repaired: docs/aio.md names the commit, the copy's completion, and the aborts. The S3Executor.submit() contract now requires the submitter's context.
6. Altitude: the aborts are protected once, in S3Core.abort_multipart_upload(), instead of at three callers.

Deferred with reasons:

• One scope per loop, a shared "stop and cancel the batch" helper, and test-helper consolidation: the call sites wait differently, and explicit per-submit scopes read more clearly.
• Merging run_in_thread() with S3AioThreadPool.run(): they submit to different pools, and sharing the code would make filesystem import aio.
• chmod leaving the unupdated keys with their ACL: the Design: a lifecycle model for in-flight work under close, cancellation, and interrupts #1205 design. The release note in the PR body states it.

[laughingman7743]

Self-review round 2 (claims, callers, operations): FINDINGS, repaired

Range: git diff 18c1eea17dcceb4395f95841d3a45b5b28c0e42d..41a772764690d61d61ad7183a904196b185226e0.
Claims checked against code:

• Every abort goes through S3Core.abort_multipart_upload() (this line): S3MultipartWriter.abort(), _abort_multipart_upload() (sync and async copies), and clear_multipart_uploads().
• The requests that keep their retries after a cancellation are exactly the run_shielded() / wait_shielded() ones minus the now-stoppable copy creation: the AioS3StreamedFile commit, the copy's completion, and the aborts. This matches docs/aio.md.
• S3AioExecutor context propagation follows from the documented call_soon_threadsafe() behavior. test_submit_context passes on master too and is kept as a contract test.
• Each stop event is set only on failure, interrupt, cancellation, or discard. The parts of a successful write are not affected.

Callers: S3File and AioS3StreamedFile gain a _parts_stop attribute, set in __init__. The test helper _make_write_file, which bypasses __init__, was updated. A custom S3Executor that does not copy the submitter's context still runs, but scopes do not reach its work; the contract is documented.
Operator view: stops only remove attempts and shorten waits. Every abort keeps its retries, including user-requested ones in clear_multipart_uploads() under a cancelled call, as the PR body states.

Correction: "a stopped creation fails before creating anything" was too strong. A creation attempt that failed with a retryable error after S3 created the upload would leave it behind, as a retry would leave the first one. The body now says it raises the error of its last attempt, like one whose retries ran out.

Evidence: tests/pyathena/filesystem/ at 41a77276: 1308 passed, 9 skipped (Python 3.15, local, including the S3 integration tests). With the source reverted to master, all new tests fail except the 4 contract tests listed in the PR body.

[laughingman7743]

Independent review (relayed Codex results)

Reviewer: OpenAI Codex CLI 0.162.1 (model gpt-6-astra, reasoning effort max), codex exec -s read-only, in a detached snapshot of each reviewed commit, given only the literal diff and the requirements (no PR text, commit messages, or author findings). It made no edits, builds, tests, GitHub calls, or memory reads. Static review.

Round	Range	Session	Result
Initial	18c1eea1..41a77276	01a147e1-02c6-7b22-84ce-c3af59db9a04	FINDINGS (5)
Follow-up 1	41a77276..1bc9eb68	01a147f1-1bd3-79a1-9c7a-e1f6b5397f28	FINDINGS (3)
Follow-up 2	1bc9eb68..d05b0e82	01a147fc-ec50-7573-a631-bef6a801fc78	FINDINGS (2)
Follow-up 3	d05b0e82..f8ffed12	01a14807-486f-70e2-9647-f5c647ecb3df	CLEAN

Each follow-up was given the range-diff and checked that the earlier commits were unchanged. All findings were verified against the code before repair. Each is recorded inline on its repair.

Limits noted by the reviewer:

• The lock that orders a call's return with a cancellation in S3AioThreadPool.run() is not deterministically testable.
• The release of a read's data after an interrupt during submission is guarded by code, not by a test. The existing memory test interrupts wait().

Self-review of the repairs (both perspectives, narrow, by range-diff per round): CLEAN after the last round.

• Behavior: every cleanup still waits for the running requests before the abort, and a creation that cannot be cancelled is still waited for and aborted. Free the data of failed S3 requests and clean up async copies at loop shutdown #1192's frame clearing in _fetch_range is kept.
• Claims: the PR body describes the first-failure rule, the cleanup order, and the interrupt handling.
• Evidence: tests/pyathena/filesystem/ at f8ffed12: 1312 passed, 9 skipped. The new tests for these repairs fail on the commit before each repair, except the two noted above. The targeted tests passed 5 times in a row. The review snapshot and PR worktree were verified unchanged.

[laughingman7743]

Codex initial 1 (introduced): stopped annotation siblings could replace the initiating error; follow-up 1, item 2: the event loop recorded the first failure after the stop, so a stopped sibling could still win. Verified. Repaired: the thread of the copy that fails records its index together with the stop, under a lock (here). Any other exception result, such as an independently cancelled child (follow-up 1, item 1), is still raised. Test: ..._failed_annotation_stops_running_copies[a1|a2]. The [a2] case fails on the earlier code; the race of follow-up 1, item 2 is not deterministically testable.

Codex initial 4 (pre-existing): queued annotation copies still sent their first request. Repaired: a copy that starts after the stop is skipped. Test: ..._failed_annotation_skips_queued_copies, which fails without the repair.

[laughingman7743]

Codex initial 2 (introduced): future.done() did not delimit the end of the call. Verified. Repaired: the call records its return under a lock that the cancellation also takes (here). A cancellation that arrives after the return sets nothing. test_run_cancelled_after_return covers the after-return case; the exact race window is not deterministically testable.

[laughingman7743]

Codex initial 3 (P1, pre-existing): the stop of a streamed file waited for a thread when the creation or parts held every thread. Verified. Repaired: _close_without_commit() sets it on the event loop before it queues _discard() (here). Follow-up 2, item 2: the queued requests are cancelled before the stop. Test: TestAioS3StreamedFile.test_cancelled_creation_stops_retries, now with max_workers=1, which fails without the repair.

[laughingman7743]

Codex initial 5 (pre-existing): an interrupt during submission bypassed the stop-and-cancel handler. Follow-up 1, item 3: an interrupt inside submit() could leave the queued item untracked. Follow-up 2, item 2: setting the stop before the cancellation let a freed thread take a queued request. Verified. Repaired:

• Submission happens inside the handler.
• Each future is created first and scheduled with submit_to().
• Every cleanup cancels the requests that have not started before it sets the stop. This covers chmod (here), range reads, both copies, and both files' discard and close-without-commit paths.

Tests: test_chmod_recursive_interrupted_submission_stops_running_requests and test_read_interrupted_submission_stops_running_reads. They interrupt after the next call is queued, assert that it is never sent, and fail on the earlier code.

[laughingman7743]

Codex follow-up 2, item 1 (introduced): an interrupted range-read submission kept the last read in a frame variable. Verified. Repaired: the futures are tracked without a variable (here). Follow-up 3: CLEAN.