put_file() publishes a truncated or empty object when the write fails

[laughingman7743]

Problem

S3FileSystem.put_file() writes the local file in a with block that holds both the remote file and the local file:

with (
    self.open(rpath, "xb" if mode == "create" else "wb", ...) as remote,
    open(lpath, "rb") as local,
):
    while data := local.read(remote.blocksize):
        remote.write(data)
        callback.relative_update(len(data))

AbstractBufferedFile.__exit__() closes the remote file whatever the exception, and close() commits it.
So a put_file() that fails publishes what was written so far:

1. A failure after the first block publishes a truncated object. If reading the local file, the progress callback, or anything else inside the loop raises after some blocks were written (including KeyboardInterrupt), the remote file completes its multipart upload with the parts uploaded so far. The existing object is replaced with a prefix of the local file, and the caller sees the original exception.
2. An unreadable local file replaces the object with an empty one. The remote file is opened before the local one. When open(lpath, "rb") raises (e.g. PermissionError), nothing was written, so S3File.commit() calls touch(), and an empty PutObject overwrites the existing object.

AioS3FileSystem._put_file_in_transaction() writes through the same with pattern (code reading). Inside a transaction, the commit is deferred, so the partial or empty object is written when the transaction commits, if the caller catches the error.

This is the same defect as #997 for pipe_file(), which #1003 fixed by closing a file whose write fails with S3File._close_without_commit() instead of committing it.

Expected: a put_file() that fails leaves the existing object unchanged and aborts its multipart upload, if one was started.

Related (pre-existing, every write path): _finish_multipart_upload() aborts the upload only on Exception. A KeyboardInterrupt while close() waits for the parts in the final completion leaves the multipart upload behind, without aborting it. The data is not corrupted; an incomplete upload remains until a lifecycle rule or clear_multipart_uploads() removes it.

Reproduction

Run offline: the filesystem is built as in TestS3FileSystem._make_fs() in tests/pyathena/filesystem/test_s3.py, with the request helpers mocked.

import os
import tempfile
from types import SimpleNamespace
from unittest import mock

from fsspec.callbacks import Callback

from tests.pyathena.filesystem.test_s3 import TestS3FileSystem


class FailingCallback(Callback):
    def relative_update(self, inc=1):
        raise RuntimeError("callback failed")


def make_fs():
    fs = TestS3FileSystem._make_fs()
    fs.default_cache_type = "bytes"
    fs._put_object = mock.MagicMock()
    fs._create_multipart_upload = mock.MagicMock(return_value=SimpleNamespace(upload_id="u"))
    fs._upload_part = mock.MagicMock(
        side_effect=lambda **kw: SimpleNamespace(etag='"e"', part_number=kw["part_number"])
    )
    fs._finish_multipart_upload = mock.MagicMock()
    return fs


# 1. A failure after the first block completes the upload with that block only.
fs = make_fs()
with tempfile.NamedTemporaryFile(delete=False) as t:
    t.write(b"x" * (11 * 2**20))
try:
    fs.put_file(t.name, "s3://bucket/key", callback=FailingCallback())
except RuntimeError as e:
    print("1. raised:", e)
print(
    "1. uploaded parts:",
    sum(len(c.kwargs["body"]) for c in fs._upload_part.call_args_list),
    "of",
    11 * 2**20,
    "bytes; completed:",
    fs._finish_multipart_upload.called,
)

# 2. An unreadable local file replaces the object with an empty one.
fs = make_fs()
with tempfile.NamedTemporaryFile(delete=False) as t:
    t.write(b"x" * 10)
os.chmod(t.name, 0)
try:
    fs.put_file(t.name, "s3://bucket/key")
except PermissionError as e:
    print("2. raised:", type(e).__name__)
print("2. PutObject calls:", [c.kwargs for c in fs._put_object.call_args_list])

Output on master 16aef64:

1. raised: callback failed
1. uploaded parts: 5242880 of 11534336 bytes; completed: True
2. raised: PermissionError
2. PutObject calls: [{'bucket': 'bucket', 'key': 'key', 'body': None}]

Environment

• PyAthena master 16aef64, fsspec 2026.9.0, Python 3.13.1.
• Checked offline with mocked request helpers; not reproduced against real S3.
• v3.37.0 has the same with block in put_file() (code reading; not run on v3.37.0).

Proposed fix (optional)

• Open the local file before the remote file, so an unreadable local file fails before anything is opened on S3.
• Write without fsspec's with block, as pipe_file() does since Write pipe_file() data without committing a failed write #1003: on any BaseException inside the loop, close the remote file with S3File._close_without_commit() and re-raise; close (commit) it only after the loop completes. Apply the same to AioS3FileSystem._put_file_in_transaction().
• Optionally, make _finish_multipart_upload() abort the upload on BaseException as well, for the related case.
• Add offline tests for a failure inside the loop (single block and multipart, in and outside a transaction) and for an unreadable local file, checking that no PutObject or completion is sent.

Found while reviewing #1003.