Skip to content

feat(Gax): add stall control for resumable uploads #9682

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
bshaffer wants to merge 5 commits into
base: main
Choose a base branch
from
Open

feat(Gax): add stall control for resumable uploads #9682

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
e516095
feat(Gax): add stall control for resumable uploads
bshaffer Sep 3, 2026
8d72117
fix(Gax): restore ?int typehint for timeoutMillis in sendRequest
bshaffer Sep 3, 2026
d0266a6
simplify stall control logic, add tests
bshaffer Sep 3, 2026
3cda3dc
chore: stall control GAPIC showcase tests
bshaffer Sep 11, 2026
6e9f907
fix cs, phpstan, and showcase tests
bshaffer Sep 15, 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
2 changes: 1 addition & 1 deletion .github/workflows/conformance-tests-gax-showcase.yaml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -11,7 +11,7 @@ jobs:
name: GAPIC Showcase Conformance Tests
runs-on: ubuntu-latest
env:
GAPIC_SHOWCASE_VERSION: 0.42.0
GAPIC_SHOWCASE_VERSION: 0.44.0
OS: linux
ARCH: amd64
steps:
Expand Down
9 changes: 8 additions & 1 deletion Gax/src/ResumableUpload/ResumableUpload.php
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -32,6 +32,7 @@

namespace Google\ApiCore\ResumableUpload;

use Google\ApiCore\ApiException;
use Google\ApiCore\Call;
use Google\Protobuf\Internal\Message;
use Psr\Http\Message\StreamInterface;
Expand Down Expand Up @@ -113,9 +114,15 @@ public function setUploadUrl(string $uploadUrl): void
* every chunk upload or query. The callback should accept two arguments:
* (int $bytesUploaded, ResumableUpload $upload).
* @type int $totalTimeoutMillis Optional. The total timeout in milliseconds for the
* entire resumable upload operation. Defaults to 600000 (10 minutes).
* entire resumable upload operation. Defaults to 600000 (10 minutes) when stall
* control is not enabled; when stall control is enabled, no default is set.
* @type int $transferStallMinimumRate Optional. The minimum data transfer speed in
* MiB/s for stall control. Must be set together with transferStallTimeout.
* @type int $transferStallTimeout Optional. The stall timeout interval in seconds
* for stall control. Must be set together with transferStallMinimumRate.
* }
* @return Message
* @throws ApiException
*/
public function startUpload(StreamInterface $dataStream, array $resumableUploadOptions = []): Message
{
Expand Down
176 changes: 156 additions & 20 deletions Gax/src/ResumableUpload/ResumableUploadClient.php
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -41,6 +41,7 @@
use Google\ApiCore\ValidationException;
use Google\Protobuf\Internal\Message;
use Google\Rpc\Code;
use GuzzleHttp\Exception\ConnectException;
use GuzzleHttp\Exception\RequestException;
use GuzzleHttp\Promise\Create;
use GuzzleHttp\Psr7\Request;
Expand Down Expand Up @@ -68,6 +69,8 @@ class ResumableUploadClient
private const MAX_RECOVERY_ATTEMPTS = 3;

private ?ResponseInterface $finalResponse = null;
/** @var callable|null */
private $clock = null;

/**
* @param ResumableUploadTransportInterface $transport Transport implementing buildRequest and sendRawRequest.
Expand All @@ -83,6 +86,22 @@ public function __construct(
) {
}

/**
* For testing purposes only. Sets a custom clock callable returning float seconds.
*
* @param ?callable $clock
* @internal
*/
public function setClock(?callable $clock): void
{
$this->clock = $clock;
}

private function getMicrotime(): float
{
return $this->clock ? ($this->clock)() : microtime(true);
}

/**
* Starts the resumable upload exchange using the provided data stream.
*
Expand Down Expand Up @@ -110,9 +129,14 @@ public function __construct(
* every chunk upload or query. The callback should accept two arguments:
* (int $bytesUploaded, ResumableUpload $upload).
* @type int $totalTimeoutMillis Optional. The total timeout in milliseconds for the
* entire resumable upload operation. Defaults to 600000 (10 minutes).
* entire resumable upload operation. Defaults to 600000 (10 minutes) when stall
* control is not enabled; when stall control is enabled, no default is set.
* @type string $uploadUrl Optional. An existing resumable upload session URL
* to resume an upload across process restarts or interruptions.
* @type int $transferStallMinimumRate Optional. The minimum data transfer speed in
* MiB/s for stall control. Must be set together with transferStallTimeout.
* @type int $transferStallTimeout Optional. The stall timeout interval in seconds
* for stall control. Must be set together with transferStallMinimumRate.
* }
* @return Message
* @throws ApiException
Expand All @@ -126,19 +150,35 @@ public function startUpload(
): Message {
$this->finalResponse = null;
$uploadUrl = $upload->getUploadUrl() ?? $resumableUploadOptions['uploadUrl'] ?? null;
$totalTimeoutMillis = (float) ($resumableUploadOptions['totalTimeoutMillis']
?? self::DEFAULT_TOTAL_TIMEOUT_MILLIS);
$deadlineMs = microtime(true) * 1000 + $totalTimeoutMillis;

$stallRate = isset($resumableUploadOptions['transferStallMinimumRate'])
? (int) $resumableUploadOptions['transferStallMinimumRate']
: null;
$stallTimeout = isset($resumableUploadOptions['transferStallTimeout'])
? (int) $resumableUploadOptions['transferStallTimeout']
: null;

$stallControlEnabled = $stallRate > 0 && $stallTimeout > 0;
$stallRate = $stallControlEnabled ? $stallRate : null;
$stallTimeout = $stallControlEnabled ? $stallTimeout : null;

$totalTimeoutMillis = $resumableUploadOptions['totalTimeoutMillis']
?? ($stallControlEnabled ? null : self::DEFAULT_TOTAL_TIMEOUT_MILLIS);
Comment on lines +165 to +166

@Hectorhammett Hectorhammett Sep 16, 2026

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

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

Nit:

Not sure I love the coalescing operator AND a tertiary operator together. It feels a bit loaded.

$globalDeadlineMs = $totalTimeoutMillis !== null
? $this->getMicrotime() * 1000 + (float) $totalTimeoutMillis
: null;

$state = new ResumableUploadState(
$resumableUploadOptions['chunkSize'] ?? self::DEFAULT_CHUNK_SIZE,
$resumableUploadOptions['progressCallback'] ?? null,
$uploadUrl,
$uploadUrl !== null ? self::PHASE_RECOVERY : self::PHASE_STARTING
$uploadUrl !== null ? self::PHASE_RECOVERY : self::PHASE_STARTING,
$stallRate,
$stallTimeout
);

while ($state->phase !== self::PHASE_DONE) {
$this->checkDeadline($deadlineMs);
$this->checkDeadline($state, $globalDeadlineMs);
try {
$state->phase = match ($state->phase) {
self::PHASE_STARTING => $call->getMessage() !== null
Expand All @@ -153,15 +193,25 @@ public function startUpload(
'A Call with request message is required when starting a new resumable upload.'
),
self::PHASE_TRANSMITTING,
self::PHASE_FINALIZING => $this->phaseUploading($state, $upload, $dataStream),
self::PHASE_RECOVERY => $this->phaseRecovery($state, $upload, $dataStream),
self::PHASE_FINALIZING => $this->phaseUploading(
$state,
$upload,
$dataStream,
$globalDeadlineMs
),
self::PHASE_RECOVERY => $this->phaseRecovery(
$state,
$upload,
$dataStream,
$globalDeadlineMs
),
default => throw new ApiException("Unexpected phase: {$state->phase}", 0, ApiStatus::INTERNAL),
};
} catch (Throwable $e) {
$state->phase = $this->handleException(
$e,
$state,
$deadlineMs
$globalDeadlineMs
);
}
}
Expand Down Expand Up @@ -251,32 +301,58 @@ private function phaseStarting(
private function phaseUploading(
ResumableUploadState $state,
ResumableUpload $upload,
StreamInterface $dataStream
StreamInterface $dataStream,
?float $globalDeadlineMs = null
): string {
$state->prepareBuffer($dataStream);

$headers = [];
$headers['X-Goog-Upload-Offset'] = (string) $state->committedOffset;
$body = (string) $state->buffer;
$chunkBytes = strlen($body);
$chunkSizeMiB = $chunkBytes / 1048576.0;

if ($state->isEof) {
$phase = self::PHASE_FINALIZING;
$headers['X-Goog-Upload-Command'] = strlen($body) > 0 ? 'upload, finalize' : 'finalize';
$headers['X-Goog-Upload-Command'] = $chunkBytes > 0 ? 'upload, finalize' : 'finalize';
} else {
$phase = self::PHASE_TRANSMITTING;
$headers['X-Goog-Upload-Command'] = 'upload';
}

$now = $this->getMicrotime();
$timeoutMillis = null;

if ($state->isStallControlEnabled()) {
$chunkTimeout = $state->calculateNextChunkTimeout($chunkSizeMiB);
if ($globalDeadlineMs !== null) {
$globalRemaining = ($globalDeadlineMs / 1000.0) - $now;

@Hectorhammett Hectorhammett Sep 16, 2026

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

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

This is an absolute minute nit, but I like the idea of moving this 1000.0 into a constant:

$globalRemaining = ($globalDeadlineMs / self::MILLS_IN_A_SECOND)

I just like having consistent naming and text that actually reminds immediately what are we doing for numbers. Of course this is extremely minute, so not blocking this review on this of course.

$chunkTimeout = min($chunkTimeout, $globalRemaining);
}
$timeoutMillis = max(1, (int) round($chunkTimeout * 1000));
}

$response = $this->sendRequest(
new Request('POST', (string) $state->uploadUrl, $headers, $body)
new Request('POST', (string) $state->uploadUrl, $headers, $body),
$timeoutMillis
);
if ($response->getStatusCode() !== 200) {
$this->handleErrorResponse($response);
}

if ($state->isStallControlEnabled()) {
$completionTime = $this->getMicrotime();
$elapsed = $completionTime - $now;
$state->recordChunkTransfer(
$chunkSizeMiB,
$elapsed,
$completionTime
);
}

if ($state->progressCallback && $headers['X-Goog-Upload-Command'] !== 'finalize') {
($state->progressCallback)(
$state->committedOffset + strlen($body),
$state->committedOffset + $chunkBytes,
$upload
);
}
Expand All @@ -300,14 +376,29 @@ private function phaseUploading(
private function phaseRecovery(
ResumableUploadState $state,
ResumableUpload $upload,
StreamInterface $dataStream
StreamInterface $dataStream,
?float $globalDeadlineMs = null
): string {
if (empty($state->uploadUrl)) {
throw new ValidationException('Cannot recover resumable upload: uploadUrl is not set.');
}

$now = $this->getMicrotime();
$timeoutMillis = null;

if ($state->isStallControlEnabled()) {
$stallTimeout = (float) $state->stallTimeout;
if ($globalDeadlineMs !== null) {
$remaining = ($globalDeadlineMs / 1000.0) - $now;
$stallTimeout = min($stallTimeout, $remaining);
}
$timeoutMillis = max(1, (int) round($stallTimeout * 1000));
}

$headers = ['X-Goog-Upload-Command' => 'query'];
$response = $this->sendRequest(
new Request('POST', (string) $state->uploadUrl, $headers, '')
new Request('POST', (string) $state->uploadUrl, $headers, ''),
$timeoutMillis
);
$statusCode = $response->getStatusCode();
if ($statusCode === 200) {
Expand Down Expand Up @@ -382,24 +473,47 @@ private function sendRequest(
return $response;
}

private function checkDeadline(float $deadlineMs, ?\Throwable $previous = null): void
{
if (microtime(true) * 1000 >= $deadlineMs) {
private function checkDeadline(
ResumableUploadState $state,
?float $globalDeadlineMs,
?\Throwable $previous = null
): void {
$now = $this->getMicrotime();

// 1. Check global deadline if set
if ($globalDeadlineMs !== null && $now * 1000 >= $globalDeadlineMs) {
throw new ApiException(
'Resumable upload total timeout exceeded.',
Code::DEADLINE_EXCEEDED,
ApiStatus::DEADLINE_EXCEEDED,
$previous ? ['previous' => $previous] : []
);
}

// 2. Check stall detection if stall control is active
if ($state->isStallControlEnabled()) {
$stallTimeout = $state->stallTimeout;

// Check if accumulated positive lag exceeded the stall timeout clock
if ($state->lag > 0 && $state->timeoutStarted !== null) {
if ($now > $state->timeoutStarted + $stallTimeout) {
throw new ApiException(
'Upload stalled.',
Code::DEADLINE_EXCEEDED,
ApiStatus::DEADLINE_EXCEEDED,
$previous ? ['previous' => $previous] : []
);
}
}
}
}

private function handleException(
\Throwable $e,
ResumableUploadState $state,
float $deadlineMs
?float $globalDeadlineMs
): string {
$this->checkDeadline($deadlineMs, $e);
$this->checkDeadline($state, $globalDeadlineMs, $e);

$code = (int) $e->getCode();
if ($e instanceof RequestException) {
Expand All @@ -424,6 +538,28 @@ private function handleException(
return self::PHASE_RECOVERY;
}

// If request timed out during stall control, raise Upload stalled
if ($state->isStallControlEnabled()
&& ($e instanceof RequestException || $e instanceof ConnectException)
&& in_array($code, [0, 408])
) {
throw new ApiException(
'Upload stalled.',
Code::DEADLINE_EXCEEDED,
ApiStatus::DEADLINE_EXCEEDED,
['previous' => $e]
);
}

// If request timed out or connection was dropped during an active upload session,
// transition to recovery to query server for committed bytes, provided uploadUrl is set
if ($state->uploadUrl !== null
&& ($e instanceof RequestException || $e instanceof ConnectException)
&& in_array($code, [0, 408])
) {
return self::PHASE_RECOVERY;
}

if ($e instanceof ApiException || $e instanceof ValidationException) {
throw $e;
}
Expand Down
Loading
Loading