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 1 commit 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
1 commit
Select commit
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
10 changes: 6 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 @@ -1198,7 +1198,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 +1228,7 @@ def upload_chunks_concurrently(
checksum=checksum,
headers=headers.copy(),
retry=retry,
timeout=timeout,
)
)

Expand All @@ -1241,9 +1242,9 @@ 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))
container.cancel(blob._get_transport(client), timeout=timeout)
raise
Comment on lines 1246 to 1248

@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.



Expand All @@ -1258,6 +1259,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 +1282,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
124 changes: 113 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 @@ -21,6 +21,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 +1032,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 +1084,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 +1142,100 @@ 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_with_metadata_and_encryption():
import datetime

Expand Down Expand Up @@ -1239,12 +1336,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 +1488,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