From 85d54ca21fa1b7f43d19493649071f892da0e290 Mon Sep 17 00:00:00 2001 From: Guilherme Souza Date: Thu, 3 Sep 2026 06:59:07 -0300 Subject: [PATCH] fix(storage): close the upload handle we opened, and only that one `_request` closed the uploaded file handle after the `except HTTPStatusError` block, so the close only ran when the request succeeded. Any non-2xx raised `StorageApiError` straight past it and the handle stayed open. Since the handle only ever existed inside the local `files` mapping, the caller had no way to close it either. Python reports it as `ResourceWarning: unclosed file`. The close also lived in the wrong place. `_request` sees only a `BufferedReader` and cannot tell one that storage3 opened from one the caller passed in, so it closed both. A caller who supplies their own stream had it closed under them on the success path. Move the close to the two functions that open the file, `_upload_or_update` and `upload_to_signed_url`, which know whether they own the handle. A handle storage3 opened is closed on both paths; a caller-supplied stream is left alone on both paths. Supersedes #1575 and #1586, which both fixed the leak by wrapping `_request` in `try/finally`. That closes the leak but keeps the close in the frame that cannot tell the two cases apart, and extends the closing of caller-owned streams to the error path as well. Co-authored-by: Tushar Pagar <240662211+tushardev-365@users.noreply.github.com> Co-authored-by: chengwudi1 --- src/storage/src/storage3/_async/file_api.py | 68 ++++++++------- src/storage/src/storage3/_sync/file_api.py | 68 ++++++++------- src/storage/tests/_async/test_file_api.py | 95 +++++++++++++++++++++ src/storage/tests/_sync/test_file_api.py | 95 +++++++++++++++++++++ 4 files changed, 260 insertions(+), 66 deletions(-) create mode 100644 src/storage/tests/_async/test_file_api.py create mode 100644 src/storage/tests/_sync/test_file_api.py diff --git a/src/storage/src/storage3/_async/file_api.py b/src/storage/src/storage3/_async/file_api.py index 54abfe11..6c0744a8 100644 --- a/src/storage/src/storage3/_async/file_api.py +++ b/src/storage/src/storage3/_async/file_api.py @@ -89,10 +89,6 @@ async def _request( message, "InternalError", exc.response.status_code ) from err - # close the resource before returning the response - if files and "file" in files and isinstance(files["file"][1], BufferedReader): - files["file"][1].close() - return response async def create_signed_upload_url( @@ -185,25 +181,27 @@ async def upload_to_signed_url( or isinstance(file, bytes) or isinstance(file, FileIO) ): - # bytes or byte-stream-like object received + # bytes or byte-stream-like object received -- the caller owns it + opened = None _file = {"file": (filename, file, content_type)} else: - # str or pathlib.path received - _file = { - "file": ( - filename, - open(file, "rb"), - content_type, - ) - } - response = await self._request( - "PUT", - final_url, - files=_file, - headers=headers, - data=_data, - query_params=query_params, - ) + # str or pathlib.path received -- we own the handle, so we close it + opened = open(file, "rb") + _file = {"file": (filename, opened, content_type)} + + try: + response = await self._request( + "PUT", + final_url, + files=_file, + headers=headers, + data=_data, + query_params=query_params, + ) + finally: + if opened is not None: + opened.close() + data: UploadData = response.json() return UploadResponse(path=path, Key=data["Key"]) @@ -562,21 +560,25 @@ async def _upload_or_update( or isinstance(file, bytes) or isinstance(file, FileIO) ): - # bytes or byte-stream-like object received + # bytes or byte-stream-like object received -- the caller owns it + opened = None files = {"file": (filename, file, content_type)} else: - # str or pathlib.path received - files = { - "file": ( - filename, - open(file, "rb"), - content_type, - ) - } + # str or pathlib.path received -- we own the handle, so we close it + opened = open(file, "rb") + files = {"file": (filename, opened, content_type)} - response = await self._request( - method, ["object", self.id, *path], files=files, headers=headers, data=_data - ) + try: + response = await self._request( + method, + ["object", self.id, *path], + files=files, + headers=headers, + data=_data, + ) + finally: + if opened is not None: + opened.close() data: UploadData = response.json() diff --git a/src/storage/src/storage3/_sync/file_api.py b/src/storage/src/storage3/_sync/file_api.py index 08b8c96d..ac5cb0f0 100644 --- a/src/storage/src/storage3/_sync/file_api.py +++ b/src/storage/src/storage3/_sync/file_api.py @@ -89,10 +89,6 @@ def _request( message, "InternalError", exc.response.status_code ) from err - # close the resource before returning the response - if files and "file" in files and isinstance(files["file"][1], BufferedReader): - files["file"][1].close() - return response def create_signed_upload_url( @@ -185,25 +181,27 @@ def upload_to_signed_url( or isinstance(file, bytes) or isinstance(file, FileIO) ): - # bytes or byte-stream-like object received + # bytes or byte-stream-like object received -- the caller owns it + opened = None _file = {"file": (filename, file, content_type)} else: - # str or pathlib.path received - _file = { - "file": ( - filename, - open(file, "rb"), - content_type, - ) - } - response = self._request( - "PUT", - final_url, - files=_file, - headers=headers, - data=_data, - query_params=query_params, - ) + # str or pathlib.path received -- we own the handle, so we close it + opened = open(file, "rb") + _file = {"file": (filename, opened, content_type)} + + try: + response = self._request( + "PUT", + final_url, + files=_file, + headers=headers, + data=_data, + query_params=query_params, + ) + finally: + if opened is not None: + opened.close() + data: UploadData = response.json() return UploadResponse(path=path, Key=data["Key"]) @@ -560,21 +558,25 @@ def _upload_or_update( or isinstance(file, bytes) or isinstance(file, FileIO) ): - # bytes or byte-stream-like object received + # bytes or byte-stream-like object received -- the caller owns it + opened = None files = {"file": (filename, file, content_type)} else: - # str or pathlib.path received - files = { - "file": ( - filename, - open(file, "rb"), - content_type, - ) - } + # str or pathlib.path received -- we own the handle, so we close it + opened = open(file, "rb") + files = {"file": (filename, opened, content_type)} - response = self._request( - method, ["object", self.id, *path], files=files, headers=headers, data=_data - ) + try: + response = self._request( + method, + ["object", self.id, *path], + files=files, + headers=headers, + data=_data, + ) + finally: + if opened is not None: + opened.close() data: UploadData = response.json() diff --git a/src/storage/tests/_async/test_file_api.py b/src/storage/tests/_async/test_file_api.py new file mode 100644 index 00000000..3411fb6f --- /dev/null +++ b/src/storage/tests/_async/test_file_api.py @@ -0,0 +1,95 @@ +from pathlib import Path +from typing import Any +from unittest.mock import Mock + +import pytest +from httpx import Headers, HTTPStatusError, Request, Response +from storage3.exceptions import StorageApiError +from yarl import URL + +from .. import AsyncBucketProxy + + +def _error_response() -> Mock: + response = Mock(spec=Response) + response.status_code = 409 + response.json.return_value = { + "message": "The resource already exists", + "error": "Duplicate", + "statusCode": 409, + } + response.raise_for_status = Mock( + side_effect=HTTPStatusError( + "Conflict", request=Mock(spec=Request), response=response + ) + ) + return response + + +def _success_response() -> Mock: + response = Mock(spec=Response) + response.status_code = 200 + response.json.return_value = {"Key": "bucket/upload.txt"} + response.raise_for_status = Mock() + return response + + +def _proxy(response: Mock, captured: dict) -> AsyncBucketProxy: + """A proxy whose transport records the `files` mapping storage3 built for the request.""" + + async def request(*args: Any, **kwargs: Any) -> Mock: + captured["files"] = kwargs["files"] + return response + + client = Mock(headers=Headers()) + client.request = request + return AsyncBucketProxy("bucket", URL("http://example.com"), Headers(), client) + + +@pytest.fixture +def source(tmp_path: Path) -> str: + path = tmp_path / "upload.txt" + path.write_bytes(b"payload") + return str(path) + + +async def test_upload_closes_file_handle_on_error(source: str) -> None: + """A handle storage3 opened itself must be closed when the upload fails.""" + captured: dict = {} + + with pytest.raises(StorageApiError): + await _proxy(_error_response(), captured).upload("upload.txt", source) + + assert captured["files"]["file"][1].closed + + +async def test_upload_closes_file_handle_on_success(source: str) -> None: + """Regression: the success path must keep closing the handle it opened.""" + captured: dict = {} + + await _proxy(_success_response(), captured).upload("upload.txt", source) + + assert captured["files"]["file"][1].closed + + +async def test_update_closes_file_handle_on_error(source: str) -> None: + """update() shares _upload_or_update, so it leaks the same way.""" + captured: dict = {} + + with pytest.raises(StorageApiError): + await _proxy(_error_response(), captured).update("upload.txt", source) + + assert captured["files"]["file"][1].closed + + +async def test_caller_supplied_handle_is_left_open(source: str) -> None: + """storage3 must not close a stream the caller owns, on either path.""" + captured: dict = {} + + with open(source, "rb") as handle: + with pytest.raises(StorageApiError): + await _proxy(_error_response(), captured).upload("upload.txt", handle) + assert not handle.closed + + await _proxy(_success_response(), captured).upload("upload.txt", handle) + assert not handle.closed diff --git a/src/storage/tests/_sync/test_file_api.py b/src/storage/tests/_sync/test_file_api.py new file mode 100644 index 00000000..eff10690 --- /dev/null +++ b/src/storage/tests/_sync/test_file_api.py @@ -0,0 +1,95 @@ +from pathlib import Path +from typing import Any +from unittest.mock import Mock + +import pytest +from httpx import Headers, HTTPStatusError, Request, Response +from storage3.exceptions import StorageApiError +from yarl import URL + +from .. import SyncBucketProxy + + +def _error_response() -> Mock: + response = Mock(spec=Response) + response.status_code = 409 + response.json.return_value = { + "message": "The resource already exists", + "error": "Duplicate", + "statusCode": 409, + } + response.raise_for_status = Mock( + side_effect=HTTPStatusError( + "Conflict", request=Mock(spec=Request), response=response + ) + ) + return response + + +def _success_response() -> Mock: + response = Mock(spec=Response) + response.status_code = 200 + response.json.return_value = {"Key": "bucket/upload.txt"} + response.raise_for_status = Mock() + return response + + +def _proxy(response: Mock, captured: dict) -> SyncBucketProxy: + """A proxy whose transport records the `files` mapping storage3 built for the request.""" + + def request(*args: Any, **kwargs: Any) -> Mock: + captured["files"] = kwargs["files"] + return response + + client = Mock(headers=Headers()) + client.request = request + return SyncBucketProxy("bucket", URL("http://example.com"), Headers(), client) + + +@pytest.fixture +def source(tmp_path: Path) -> str: + path = tmp_path / "upload.txt" + path.write_bytes(b"payload") + return str(path) + + +def test_upload_closes_file_handle_on_error(source: str) -> None: + """A handle storage3 opened itself must be closed when the upload fails.""" + captured: dict = {} + + with pytest.raises(StorageApiError): + _proxy(_error_response(), captured).upload("upload.txt", source) + + assert captured["files"]["file"][1].closed + + +def test_upload_closes_file_handle_on_success(source: str) -> None: + """Regression: the success path must keep closing the handle it opened.""" + captured: dict = {} + + _proxy(_success_response(), captured).upload("upload.txt", source) + + assert captured["files"]["file"][1].closed + + +def test_update_closes_file_handle_on_error(source: str) -> None: + """update() shares _upload_or_update, so it leaks the same way.""" + captured: dict = {} + + with pytest.raises(StorageApiError): + _proxy(_error_response(), captured).update("upload.txt", source) + + assert captured["files"]["file"][1].closed + + +def test_caller_supplied_handle_is_left_open(source: str) -> None: + """storage3 must not close a stream the caller owns, on either path.""" + captured: dict = {} + + with open(source, "rb") as handle: + with pytest.raises(StorageApiError): + _proxy(_error_response(), captured).upload("upload.txt", handle) + assert not handle.closed + + _proxy(_success_response(), captured).upload("upload.txt", handle) + assert not handle.closed