Conversation
![gemini-code-assist[bot]](https://avatars.githubusercontent.com/in/956858?s=60&v=4){kind=small}
There was a problem hiding this comment.
Code Review
This pull request introduces a new resumable transfer library for Google APIs, implementing both synchronous (using requests) and asynchronous (using aiohttp) resumable upload sessions, supported by a sans-I/O protocol state machine and comprehensive tests. The feedback highlights several key areas for improvement: ensuring backward compatibility with Python 3.7/3.8 by replacing asyncio.to_thread with loop.run_in_executor, handling byte-type header keys in the state machine, rejecting unsupported str and dict stream types early, retrying timeouts globally in the synchronous session, and removing redundant deadline checks.
parthea
changed the title
…load_state.py Co-authored-by: gemini-code-assist[bot] <176961590+gemini-code-assist[bot]@users.noreply.github.com>
…load_async.py Co-authored-by: gemini-code-assist[bot] <176961590+gemini-code-assist[bot]@users.noreply.github.com>
…load.py Co-authored-by: gemini-code-assist[bot] <176961590+gemini-code-assist[bot]@users.noreply.github.com>
parthea
changed the title
daniel-sanche
left a comment
daniel-sanche
left a comment
There was a problem hiding this comment.
I'm still digesting this, but giving a quick first round of comments
| progress_queue.put_nowait(exc) | ||
| raise | ||
|
|
||
| task = asyncio.create_task(_run()) |
There was a problem hiding this comment.
This seems surprising to me. upload is a sync method, that starts a background task?
I'd expect upload to return the awaitable, and then the user can choose to await it directly, or create a background task
There was a problem hiding this comment.
upload() returns an awaitable AsyncUploadOperation so callers can either await session.upload(stream) directly or iterate async for p in op.progress(): before awaiting the final response.
Deferring task creation so execution starts lazily on await or .progress() is a forward-compatible change since the public API and calling patterns stay identical. Happy to follow up on that in a separate PR but I can come back to this if you have a strong opinion about it.
There was a problem hiding this comment.
It's a forward-compatible change if we assume the user is calling await session.upload(). But if they call session.upload() and write code that assumes it is running in the background, that could lead to problems if changed later
It makes me a bit nervous because it is non standard, and could lead to unexpected consequences. And managing background thread lifecycles is always difficult. I'd prefer a simpler approach if possible. But I think this can work if needed
This reverts commit f40173b.
daniel-sanche
left a comment
daniel-sanche
left a comment
There was a problem hiding this comment.
It's definitely looking closer, but I'm still seeing some issues. Try to make sure the test cases I shared are incorporated (or something like them)
I'm mostly concerned about the public API, and data loss issues. But I left comments on other things as well
| content_type: Optional[str] = None, | ||
| response_type: Optional[Any] = None, | ||
| start_retry: Optional[google.api_core.retry.Retry] = None, | ||
| start_timeout: Optional[float] = None, |
There was a problem hiding this comment.
It seems surprising to me that we expose start_retry and start_timeout, here, but then also allow overriding it on as an argument to initate
Is that needed? Could we just keep this on initiate?
There was a problem hiding this comment.
Good catch! initiate isn't called directly by users. We can remove the arguments to initiate which are considered a duplicate. We need to keep start_retry and start_timeout here otherwise callers using session.upload() would have no way to configure retry or timeout for the initial start request.
Done in 8322a1e
There is a section under Setting Scotty upload configuration in the requirements doc that mentions the need for configuring a retry policy for the initial start command, as well as the chunk upload.
| exceptions.UploadCancelledError, | ||
| exceptions.UnseekableStreamError, | ||
| ), | ||
| ): |
There was a problem hiding this comment.
nit: Maybe these should be in a constant list called TERMINAL_ERRORS? It's not clear from reading the code why some errors are handled different from others
| and exc.code in common.RECOVERABLE_STATUS_CODES | ||
| ) | ||
| ): | ||
| return True |
There was a problem hiding this comment.
Are you sure this should be above the user-supplied predicate? Would they want a say in whether to retry these? What about the ConnectionErrors below?
(Personally, I'm not convinved it's a good idea to allow a user-supplied predicate. Users may try to use this as both a whitelist or a blacklist, and it's not always clear how to compose their predicate with the system one. If it's not part of the requirements, I'd say we should just warn that custom predicates aren't supported for now)
There was a problem hiding this comment.
The requirements specify maintaining a similar UX to existing generated methods. To avoid adding technical debt, and causing confusion for users who would like to set custom predicates, custom predicates could remain supported for Category 1 transient errors only. I'll update the doc string to capture the limitation.
| errors. Terminal errors (``DeadlineExceeded``, ``TransferStalledError``, | ||
| ``UploadCancelledError``, and ``UnseekableStreamError``) are never | ||
| retried. | ||
| start_timeout: Optional timeout in seconds for the start request. |
There was a problem hiding this comment.
For some of these arguments, it looks like None means "use default". Can we calrify that in the docstrings?
(I also wonder if it would be cleaner to have default values/sentinels here in the signature, instead of replacing None values with defaults later. _DEFAULT_START_TIMEOUT, for instance)
There was a problem hiding this comment.
Fixed in c17a7d5. I hardcoded the timeout value in the docstring to avoid exposing the internal constant _DEFAULT_START_TIMEOUT.
| request_body=request_body, | ||
| size=computed_size, | ||
| progress_queue=progress_queue, | ||
| ) |
There was a problem hiding this comment.
I'm a bit confused about initiate. Do we expect users to call it before upload, or should it be internal?
Will calling initiate on an already-initiated sstream create a new url?
Would keeping these two stages separate be simpler here?
There was a problem hiding this comment.
Do we expect users to call it before upload, or should it be internal?
Good point. I'll move it to internal since we don't expect users to call it
Will calling initiate on an already-initiated sstream create a new url?
Yes, calling initiate() again sends another X-Goog-Upload-Command: start request and overwrites self._state._resumable_url with a new session URL. We won't call it twice
Would keeping these two stages separate be simpler here?
Generated client methods return an unstarted ResumableUploadSession so callers can either start a new transfer via session.upload() or resume an interrupted transfer via session.resume() on that same session object without issuing an extra start request. Renaming initiate() to _initiate() will make it clear that we don't expect users to call it.
|
|
||
| _, _, body_bytes = final_resp_tuple | ||
| self._response = self._format_response(body_bytes) | ||
| return self._response |
There was a problem hiding this comment.
The type annotations say this will be a ResponseProto, but _format_response will actually return bytes by default here (which is different from the sync implementation)
There was a problem hiding this comment.
Good catch! Fixed in 61853c9
I updated _format_response_payload so both sync and async return bytes when response_type is None
| >= self._config.stall_timeout | ||
| ): | ||
| self._get_deadline_remaining() | ||
| raise exceptions.TransferStalledError( |
There was a problem hiding this comment.
Gemini pointed out that requests.exceptions.Timeout is raised as soon as per_attempt_timeout is hit, which is defined as max(5.0, min(next_chunk_timeout, 2.0 * expected_sec)). That can be as low as 5 seconds, so it seems like it can exit much sooner than stall_timeout (120 seconds).
| progress_queue.put_nowait(exc) | ||
| raise | ||
|
|
||
| task = asyncio.create_task(_run()) |
There was a problem hiding this comment.
It's a forward-compatible change if we assume the user is calling await session.upload(). But if they call session.upload() and write code that assumes it is running in the background, that could lead to problems if changed later
It makes me a bit nervous because it is non standard, and could lead to unexpected consequences. And managing background thread lifecycles is always difficult. I'd prefer a simpler approach if possible. But I think this can work if needed
| """Awaits completion of the upload task and returns the server response.""" | ||
| return self._task.__await__() | ||
|
|
||
| async def progress(self) -> AsyncIterator[common.UploadProgress]: |
There was a problem hiding this comment.
I see both the sync and async classes have multiple ways to track progress, which leads to some degree of complexity. Personally, I think we should stick with one or the other, and keep things simpler
It looks like we could simplify a lot of code if we dropped this progress method, since we wouldn't need to manage the background task/progress queue. Or if you did want to keep a progress generator, it seems like th ecode would be simpler to expose it as iter_progress like the sync version.
I'm fine with keeping both methods if you want, it just adds more complexity than we probably need here
| A callable accepting an exception and returning a boolean. | ||
| """ | ||
|
|
||
| def should_retry(exc: Exception) -> bool: |
There was a problem hiding this comment.
Should asyncio.TimeoutError be included here?
2 participants
Towards b/457416314, b/556259599