fix(storage): respect deadline in transfer_manager.upload_many

[rameshvarun]

Fixes #17959

Right now transfer_manager.upload_many's deadline parameter doesn't actually cause a TimeoutError - it's essentially a no-op and the deadline is not respected. This MR fixes that.

• In process mode, the worker processes are terminated. In thread mode this is not possible.
• transfer_manager has other instances where deadline handling is broken. I only fixed upload_many for easier review. I can investigate the others as a follow-up.
• test_upload_many_terminates_process_workers_on_deadline is a somewhat tautological test. I did have Claude mock up a deeper test involving actual sub-processes, but it seemed illegible, so I opted not to include it.

[google-cla]

Thanks for your pull request! It looks like this may be your first contribution to a Google open source project. Before we can look at your pull request, you'll need to sign a Contributor License Agreement (CLA).

View this failed invocation of the CLA check for more information.

For the most up to date status, view the checks section at the bottom of the pull request.

[gemini-code-assist]

Code Review

This pull request refactors 'upload_many' to explicitly manage the executor instead of using a context manager, allowing worker processes to be terminated and a 'TimeoutError' to be raised when a deadline is exceeded. The review feedback identifies critical issues: first, 'ProcessPoolExecutor._processes' stores process objects as keys rather than values, so iterating over '.values()' will cause an 'AttributeError' in production; second, the 'cancel_futures' argument in 'executor.shutdown()' is unsupported in Python 3.7 and 3.8, leading to compatibility failures. Additionally, the reviewer suggests optimizing the unit tests by mocking the timeout wait instead of using real sleeps, and updating the process mock to match CPython's actual structure.

[gemini-code-assist]

Instead of spawning a real background thread that sleeps for 5 seconds (which can slow down the test suite and potentially cause flakiness), we can mock concurrent.futures.wait to return a non-empty not_done set, just like we do in the process-mode test. This makes the test fast, deterministic, and clean.

def test_upload_many_raises_timeout_error_when_deadline_exceeded():
    # Thread-mode: A stuck upload must not make upload_many hang past the deadline.
    with mock.patch("concurrent.futures.wait") as wait_patch:
        # A non-empty not_done set signals the deadline was exceeded.
        wait_patch.return_value = (set(), {concurrent.futures.Future()})
        with pytest.raises(concurrent.futures.TimeoutError):
            transfer_manager.upload_many(
                [(io.BytesIO(b"data"), mock.Mock(spec=Blob))],
                worker_type=transfer_manager.THREAD,
                deadline=0.1,
            )

[rameshvarun]

My opinion:

• The current test only takes 0.1s, not 5s. So it shouldn't meaningfully slow down CI. I'm sure we can even reduce the deadline to 0.01.
• I don't see why it would be flaky.
• Mocking concurrent.futures.wait in my opinion isn't testing deeply enough.

I will let the reviewer decide.

[rameshvarun]

Thanks for your pull request! It looks like this may be your first contribution to a Google open source project. Before we can look at your pull request, you'll need to sign a Contributor License Agreement (CLA).

View this failed invocation of the CLA check for more information.

For the most up to date status, view the checks section at the bottom of the pull request.

I signed the CLA.

[parthea]

/gcbrun