Skip to content

fix(storage): honor timeout in upload_chunks_concurrently #18765

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
abhishekacharya200 wants to merge 2 commits into
base: main
Choose a base branch
from
Open

fix(storage): honor timeout in upload_chunks_concurrently #18765

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
ccc12b7
fix(storage): honor timeout in upload_chunks_concurrently
abhishekacharya200 Oct 19, 2026
1520e8b
fix(storage): keep the original error when cancelling an upload fails
abhishekacharya200 Oct 31, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
17 changes: 13 additions & 4 deletions packages/google-cloud-storage/google/cloud/storage/transfer_manager.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,6 +20,7 @@
import functools
import inspect
import io
import logging
import os
import pickle
import struct
Expand All @@ -36,6 +37,8 @@
from google.cloud.storage.exceptions import DataCorruption, InvalidPathError
from google.cloud.storage.retry import DEFAULT_RETRY

_logger = logging.getLogger(__name__)

TM_DEFAULT_CHUNK_SIZE = 32 * 1024 * 1024
DEFAULT_MAX_WORKERS = 8
MAX_CRC32C_ZERO_ARRAY_SIZE = 4 * 1024 * 1024
Expand Down Expand Up @@ -1198,7 +1201,7 @@ def upload_chunks_concurrently(

container = XMLMPUContainer(url, filename, headers=headers, retry=retry)

container.initiate(transport=transport, content_type=content_type)
container.initiate(transport=transport, content_type=content_type, timeout=timeout)
upload_id = container.upload_id

size = os.path.getsize(filename)
Expand Down Expand Up @@ -1228,6 +1231,7 @@ def upload_chunks_concurrently(
checksum=checksum,
headers=headers.copy(),
retry=retry,
timeout=timeout,
)
)

Expand All @@ -1241,9 +1245,13 @@ def upload_chunks_concurrently(
part_number, etag = future.result()
container.register_part(part_number, etag)

container.finalize(blob._get_transport(client))
container.finalize(blob._get_transport(client), timeout=timeout)
except Exception:
container.cancel(blob._get_transport(client))
# A failed cleanup must not hide the error that caused it.
try:
container.cancel(blob._get_transport(client), timeout=timeout)
except Exception as exc:
_logger.warning("Failed to cancel multipart upload %s: %s", upload_id, exc)
raise
Comment on lines 1249 to 1255

@gemini-code-assist gemini-code-assist Bot Oct 19, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

medium

If container.cancel raises an exception (for example, due to a timeout or network issue during the cleanup attempt), it will mask the original exception that triggered the except block. To ensure the original error is preserved and propagated to the caller for proper debugging, wrap the container.cancel call in a nested try...except block. Avoid silently ignoring the exception with a broad except Exception: pass block; instead, log the exception using logger.warning to aid in debugging and prevent masking underlying issues.

    except Exception:
        try:
            container.cancel(blob._get_transport(client), timeout=timeout)
        except Exception as e:
            logger.warning("Failed to cancel container: %s", e)
        raise
References
  1. Avoid broad except Exception: blocks that silently return None. Instead, log the exception (e.g., using logger.warning) to aid in debugging and prevent masking underlying issues.

@abhishekacharya200 abhishekacharya200 Oct 22, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Good point. This was already the case before this change, but passing the caller's timeout to cancel makes it a little more likely. I kept the PR focused on the timeout, and I'm happy to wrap the cancel call so the original error is always the one raised, or to leave cancel on the default timeout, whichever the maintainers prefer.

@shradhakatyal shradhakatyal Oct 26, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Thanks for checking! Let's keep passing timeout=timeout to container.cancel (so fail-fast or extended caller timeouts are respected consistently across all requests) and wrap container.cancel in a try...except Exception block so a cleanup failure never masks the original upload exception.

Two small notes when adding that:

  1. transfer_manager.py doesn't define a logger yet — please use _logger = logging.getLogger(name) (matching blob.py) and include upload_id in the _logger.warning(...) message.

  2. Since nox -s cover enforces 100% branch coverage, please also add a unit test where container.cancel raises during cleanup to verify the original exception is preserved and the warning is logged.

@abhishekacharya200 abhishekacharya200 Oct 31, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Thanks @shradhakatyal, done in 1520e8b. cancel keeps the caller's timeout and is wrapped so a failed cleanup never masks the upload error. The warning includes the upload id, and a new test covers a failing cancel. Ready for another look.



Expand All @@ -1258,6 +1266,7 @@ def _upload_part(
checksum,
headers,
retry,
timeout,
):
"""Helper function that runs inside a thread or subprocess to upload a part.

Expand All @@ -1280,7 +1289,7 @@ def _upload_part(
headers=headers,
retry=retry,
)
part.upload(client._http)
part.upload(client._http, timeout=timeout)
return (part_number, part.etag)


Expand Down
166 changes: 155 additions & 11 deletions packages/google-cloud-storage/tests/unit/test_transfer_manager.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -12,6 +12,7 @@
# See the License for the specific language governing permissions and
# limitations under the License.

import logging
import os
import pickle
import tempfile
Expand All @@ -21,6 +22,7 @@
from google.api_core import exceptions

from google.cloud.storage import Blob, Client, transfer_manager
from google.cloud.storage.constants import _DEFAULT_TIMEOUT
from google.cloud.storage.exceptions import DataCorruption
from google.cloud.storage.retry import DEFAULT_RETRY

Expand Down Expand Up @@ -1031,13 +1033,17 @@ def test_upload_chunks_concurrently():
worker_type=transfer_manager.THREAD,
)
container_mock.initiate.assert_called_once_with(
transport=transport, content_type=blob.content_type
transport=transport,
content_type=blob.content_type,
timeout=_DEFAULT_TIMEOUT,
)
container_mock.register_part.assert_any_call(1, ETAG)
container_mock.register_part.assert_any_call(2, ETAG)
container_mock.finalize.assert_called_once_with(bucket.client._http)
container_mock.finalize.assert_called_once_with(
bucket.client._http, timeout=_DEFAULT_TIMEOUT
)

part_mock.upload.assert_called_with(transport)
part_mock.upload.assert_called_with(transport, timeout=_DEFAULT_TIMEOUT)


def test_upload_chunks_concurrently_quotes_urls():
Expand Down Expand Up @@ -1079,17 +1085,21 @@ def test_upload_chunks_concurrently_quotes_urls():
)

container_mock.initiate.assert_called_once_with(
transport=transport, content_type=blob.content_type
transport=transport,
content_type=blob.content_type,
timeout=_DEFAULT_TIMEOUT,
)
container_mock.register_part.assert_any_call(1, ETAG)
container_mock.register_part.assert_any_call(2, ETAG)
container_mock.finalize.assert_called_once_with(bucket.client._http)
container_mock.finalize.assert_called_once_with(
bucket.client._http, timeout=_DEFAULT_TIMEOUT
)

container_cls_mock.assert_called_once_with(
quoted_url, FILENAME, headers=mock.ANY, retry=DEFAULT_RETRY
)

part_mock.upload.assert_called_with(transport)
part_mock.upload.assert_called_with(transport, timeout=_DEFAULT_TIMEOUT)


def test_upload_chunks_concurrently_passes_concurrency_options():
Expand Down Expand Up @@ -1133,12 +1143,141 @@ def test_upload_chunks_concurrently_passes_concurrency_options():
pass # The futures don't actually work, so we expect this to abort.
# Conveniently, that gives us a chance to test the auto-delete
# exception handling feature.
container_mock.cancel.assert_called_once_with(transport)
container_mock.cancel.assert_called_once_with(
transport, timeout=_DEFAULT_TIMEOUT
)

pool_patch.assert_called_with(max_workers=MAX_WORKERS)
wait_patch.assert_called_with(mock.ANY, timeout=DEADLINE, return_when=mock.ANY)


def test_upload_chunks_concurrently_passes_timeout():
bucket = mock.Mock()
bucket.name = "bucket"
bucket.client = _PickleableMockClient(identify_as_client=True)
transport = bucket.client._http
bucket.user_project = None

blob = Blob("blob", bucket)
blob.content_type = FAKE_CONTENT_TYPE

FILENAME = "file_a.txt"
SIZE = 2048
TIMEOUT = (3, 7)

container_mock = mock.Mock()
container_mock.upload_id = "abcd"
part_mock = mock.Mock()
part_mock.etag = "efgh"

with (
mock.patch("os.path.getsize", return_value=SIZE),
mock.patch(
"google.cloud.storage.transfer_manager.XMLMPUContainer",
return_value=container_mock,
),
mock.patch(
"google.cloud.storage.transfer_manager.XMLMPUPart", return_value=part_mock
),
):
transfer_manager.upload_chunks_concurrently(
FILENAME,
blob,
chunk_size=SIZE // 2,
worker_type=transfer_manager.THREAD,
timeout=TIMEOUT,
)
container_mock.initiate.assert_called_once_with(
transport=transport, content_type=blob.content_type, timeout=TIMEOUT
)
assert part_mock.upload.call_args_list == [
mock.call(transport, timeout=TIMEOUT),
mock.call(transport, timeout=TIMEOUT),
]
container_mock.finalize.assert_called_once_with(transport, timeout=TIMEOUT)


def test_upload_chunks_concurrently_passes_timeout_to_cancel():
bucket = mock.Mock()
bucket.name = "bucket"
bucket.client = _PickleableMockClient(identify_as_client=True)
transport = bucket.client._http
bucket.user_project = None

blob = Blob("blob", bucket)

FILENAME = "file_a.txt"
SIZE = 2048
TIMEOUT = 15

container_mock = mock.Mock()
container_mock.upload_id = "abcd"
part_mock = mock.Mock()
part_mock.upload.side_effect = exceptions.ServiceUnavailable("part failed")

with (
mock.patch("os.path.getsize", return_value=SIZE),
mock.patch(
"google.cloud.storage.transfer_manager.XMLMPUContainer",
return_value=container_mock,
),
mock.patch(
"google.cloud.storage.transfer_manager.XMLMPUPart", return_value=part_mock
),
):
with pytest.raises(exceptions.ServiceUnavailable):
transfer_manager.upload_chunks_concurrently(
FILENAME,
blob,
chunk_size=SIZE // 2,
worker_type=transfer_manager.THREAD,
timeout=TIMEOUT,
)
container_mock.cancel.assert_called_once_with(transport, timeout=TIMEOUT)
container_mock.finalize.assert_not_called()


def test_upload_chunks_concurrently_keeps_original_error_when_cancel_fails(caplog):
bucket = mock.Mock()
bucket.name = "bucket"
bucket.client = _PickleableMockClient(identify_as_client=True)
bucket.user_project = None

blob = Blob("blob", bucket)

FILENAME = "file_a.txt"
SIZE = 2048

container_mock = mock.Mock()
container_mock.upload_id = "abcd"
container_mock.cancel.side_effect = exceptions.DeadlineExceeded("cancel failed")
part_mock = mock.Mock()
part_mock.upload.side_effect = exceptions.ServiceUnavailable("part failed")

with (
mock.patch("os.path.getsize", return_value=SIZE),
mock.patch(
"google.cloud.storage.transfer_manager.XMLMPUContainer",
return_value=container_mock,
),
mock.patch(
"google.cloud.storage.transfer_manager.XMLMPUPart", return_value=part_mock
),
):
with caplog.at_level(logging.WARNING, logger=transfer_manager.__name__):
with pytest.raises(exceptions.ServiceUnavailable):
transfer_manager.upload_chunks_concurrently(
FILENAME,
blob,
chunk_size=SIZE // 2,
worker_type=transfer_manager.THREAD,
)

container_mock.cancel.assert_called_once()
assert "abcd" in caplog.text
assert "cancel failed" in caplog.text


def test_upload_chunks_concurrently_with_metadata_and_encryption():
import datetime

Expand Down Expand Up @@ -1239,12 +1378,16 @@ def test_upload_chunks_concurrently_with_metadata_and_encryption():
URL, FILENAME, headers=expected_headers, retry=DEFAULT_RETRY
)
container_mock.initiate.assert_called_once_with(
transport=transport, content_type=blob.content_type
transport=transport,
content_type=blob.content_type,
timeout=_DEFAULT_TIMEOUT,
)
container_mock.register_part.assert_any_call(1, ETAG)
container_mock.register_part.assert_any_call(2, ETAG)
container_mock.finalize.assert_called_once_with(transport)
part_mock.upload.assert_called_with(blob.client._http)
container_mock.finalize.assert_called_once_with(
transport, timeout=_DEFAULT_TIMEOUT
)
part_mock.upload.assert_called_with(blob.client._http, timeout=_DEFAULT_TIMEOUT)


class _PickleableMockBlob:
Expand Down Expand Up @@ -1387,8 +1530,9 @@ def test__upload_part():
None,
{"key", "value"},
retry=DEFAULT_RETRY,
timeout=(3, 7),
)
part.upload.assert_called_once()
part.upload.assert_called_once_with(mock.ANY, timeout=(3, 7))

assert result == (1, ETAG)

Expand Down
Loading