Failed multipart uploads are aborted while parts are in flight, and a failed write-mode open() leaves a half-built S3File #976

Problem

Two error paths of the multipart write and copy code leave work behind.

1. A failed multipart upload is aborted while other parts are still uploading. When a part fails, _finish_multipart_upload() cancels the remaining futures and calls AbortMultipartUpload at once (pyathena/filesystem/s3.py:1370-1384). Future.cancel() does not stop a part that is already running, so those UploadPart requests finish after the abort. S3File.discard() does the same (s3.py:2406-2422).
   • With block_size=5 MiB and max_workers=4, when part 1 fails, the abort is sent before parts 2 and 3 are stored.
   • The AbortMultipartUpload documentation states that parts still uploading during the abort might succeed, and that the upload may have to be aborted again to free all storage consumed by its parts. Nothing aborts it again here.
   • The helper serves S3File.commit() (s3.py:2389) and the multipart copy in cp_file() (s3.py:1274), so in-flight UploadPartCopy requests are affected the same way.
2. A write-mode open() that fails validation leaves a half-built S3File. S3File.__init__ calls AbstractBufferedFile.__init__ first (s3.py:2186-2195) and validates afterwards. The block size check (s3.py:2212-2215) and the missing-key check (s3.py:2198-2199) raise ValueError before append_block, multipart_upload and multipart_upload_parts are set (s3.py:2217, :2245-2246).
   • When the object is garbage collected, fsspec's __del__ calls close(), which flushes and raises AttributeError: 'S3File' object has no attribute 'append_block'. Python prints it as "Exception ignored in" after every failed open().
   • close() also never reaches self._executor.shutdown() (s3.py:2248-2251).

Expected:

• a failed multipart upload waits for the part requests that are already running and then aborts the upload, so no part outlives the abort;
• a write-mode open() that raises ValueError leaves nothing to close and sends no request.

Fix together: #926 (block size error messages), #944 and #952 (no upper bound on the write block size; duplicates of each other) and #953 (writes fail after 10,000 parts) change the same S3File write validation. PR #949 (#945) changes the same except blocks in _finish_multipart_upload() and S3File.commit().

Reproduction

import gc, sys, threading, time
from unittest.mock import MagicMock
from pyathena.filesystem.s3 import S3FileSystem

MIN = S3FileSystem.MULTIPART_UPLOAD_MIN_PART_SIZE
events, lock = [], threading.Lock()
def call(method, **request):
    name = method if isinstance(method, str) else method._extract_mock_name().split(".")[-1]
    if name == "upload_part":
        if request["PartNumber"] == 1:
            raise OSError("part 1 failed")
        time.sleep(0.5)
        name = f"part {request['PartNumber']} stored"
    with lock:
        events.append(name)
    return {"UploadId": "u", "ETag": '"e"'}

fs = S3FileSystem(connection=MagicMock(), skip_instance_cache=True)
fs._call = call
try:
    with fs.open("s3://bucket/key", "wb", block_size=MIN, max_workers=4) as f:
        f.write(b"x" * MIN * 3)
except OSError as e:
    print(repr(e))
time.sleep(1)
print(events[:2], sorted(events[2:]))
# OSError('part 1 failed')
# ['create_multipart_upload', 'abort_multipart_upload'] ['part 2 stored', 'part 3 stored']

sys.unraisablehook = lambda u: print("ignored in __del__:", repr(u.exc_value))
events.clear()
for path, block_size in [("s3://bucket/key", 1024), ("s3://bucket", MIN)]:
    try:
        fs.open(path, "wb", block_size=block_size)
    except ValueError as e:
        print(repr(e))
    gc.collect()
print(events)
# ValueError('Block size must be >= 5242880MB.')
# ignored in __del__: AttributeError("'S3File' object has no attribute 'append_block'")
# ValueError('The path does not contain a key.')
# ignored in __del__: AttributeError("'S3File' object has no attribute 'append_block'")
# []

Environment

• PyAthena master e0e85da, fsspec 2026.9.0, botocore from uv.lock, Python 3.13.1.
• Found in a full audit of pyathena/filesystem/. Unless noted, checked offline with mocked S3 responses (fs._call replaced) or botocore's Stubber with dummy credentials.

Proposed fix (optional)

• In _finish_multipart_upload() and S3File.discard(), cancel the pending futures, wait for the running ones with concurrent.futures.wait(), and then send AbortMultipartUpload.
• Validate the S3File arguments before AbstractBufferedFile.__init__, so a failed open() creates no file object to close. Defining the attributes earlier is not enough: with append_block, multipart_upload and multipart_upload_parts set, garbage collection of the failed file reached commit() and sent a PutObject that creates an empty object.
• Add offline tests for both paths.