diff --git a/.github/workflows/conformance-tests-gax-showcase.yaml b/.github/workflows/conformance-tests-gax-showcase.yaml index a5a98544400..14ab3f12c93 100644 --- a/.github/workflows/conformance-tests-gax-showcase.yaml +++ b/.github/workflows/conformance-tests-gax-showcase.yaml @@ -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: diff --git a/Gax/src/ResumableUpload/ResumableUpload.php b/Gax/src/ResumableUpload/ResumableUpload.php index f08feffd608..3a0e52d7b61 100644 --- a/Gax/src/ResumableUpload/ResumableUpload.php +++ b/Gax/src/ResumableUpload/ResumableUpload.php @@ -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; @@ -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 { diff --git a/Gax/src/ResumableUpload/ResumableUploadClient.php b/Gax/src/ResumableUpload/ResumableUploadClient.php index 639b7c67e58..ef86354ec0b 100644 --- a/Gax/src/ResumableUpload/ResumableUploadClient.php +++ b/Gax/src/ResumableUpload/ResumableUploadClient.php @@ -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; @@ -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. @@ -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. * @@ -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 @@ -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); + $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 @@ -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 ); } } @@ -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; + $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 ); } @@ -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) { @@ -382,9 +473,15 @@ 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, @@ -392,14 +489,31 @@ private function checkDeadline(float $deadlineMs, ?\Throwable $previous = null): $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) { @@ -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; } diff --git a/Gax/src/ResumableUpload/ResumableUploadState.php b/Gax/src/ResumableUpload/ResumableUploadState.php index 0522ce1d393..60e129b6d04 100644 --- a/Gax/src/ResumableUpload/ResumableUploadState.php +++ b/Gax/src/ResumableUpload/ResumableUploadState.php @@ -35,6 +35,7 @@ use Google\ApiCore\ApiException; use Google\ApiCore\ApiStatus; use Google\ApiCore\ValidationException; +use Google\Rpc\Code; use Psr\Http\Message\StreamInterface; /** @@ -52,19 +53,25 @@ class ResumableUploadState public ?string $previousBuffer = null; public int $previousOffset = 0; public bool $isEof = false; + public float $lag = 0.0; + public ?float $timeoutStarted = null; /** * @param int $chunkSize * @param callable|null $progressCallback * @param ?string $uploadUrl * @param string $phase + * @param ?int $stallMinimumRate + * @param ?int $stallTimeout */ public function __construct( public int $chunkSize, /** @var callable|null $progressCallback */ public $progressCallback, public ?string $uploadUrl, - public string $phase + public string $phase, + public ?int $stallMinimumRate = null, + public ?int $stallTimeout = null ) { } @@ -175,5 +182,69 @@ public function reconcileRecoveryOffset( $this->committedOffset = $serverOffset; $this->buffer = null; } + + if ($this->buffer === '') { + $this->buffer = null; + } + } + + /** + * Whether stall control is active for this upload session. + */ + public function isStallControlEnabled(): bool + { + return $this->stallMinimumRate !== null + && $this->stallMinimumRate > 0 + && $this->stallTimeout !== null + && $this->stallTimeout > 0; + } + + /** + * Calculates the timeout for the next chunk transfer in seconds. + * + * @param float $chunkSizeMiB + * @return float Timeout in seconds. + */ + public function calculateNextChunkTimeout(float $chunkSizeMiB): float + { + if (!$this->isStallControlEnabled()) { + return 0.0; + } + $expectedTime = $chunkSizeMiB / $this->stallMinimumRate; + return $expectedTime - $this->lag + $this->stallTimeout; + } + + /** + * Records chunk transfer completion and updates aggregate lag and stall timeout clock. + * + * @param float $chunkSizeMiB + * @param float $elapsed Seconds taken to transfer the chunk. + * @param float $currentTime Current timestamp in seconds. + * @throws ApiException + */ + public function recordChunkTransfer(float $chunkSizeMiB, float $elapsed, float $currentTime): void + { + if (!$this->isStallControlEnabled()) { + return; + } + $expectedTime = $chunkSizeMiB / $this->stallMinimumRate; + $currentLag = $elapsed - $expectedTime; + $this->lag = max(0.0, $this->lag + $currentLag); + + if ($this->lag > 0) { + if ($this->timeoutStarted === null) { + $this->timeoutStarted = $currentTime; + } + if ($currentTime > $this->timeoutStarted + $this->stallTimeout) { + throw new ApiException( + 'Upload stalled.', + Code::DEADLINE_EXCEEDED, + ApiStatus::DEADLINE_EXCEEDED + ); + } + } else { + // Recovered from stall: lag is 0, exit stall detection + $this->timeoutStarted = null; + } } } diff --git a/Gax/tests/Conformance/ResumableUploadTest.php b/Gax/tests/Conformance/ResumableUploadTest.php index 4ea01801176..7b7b92de057 100644 --- a/Gax/tests/Conformance/ResumableUploadTest.php +++ b/Gax/tests/Conformance/ResumableUploadTest.php @@ -37,7 +37,9 @@ class ResumableUploadTest extends TestCase private function createClientAndUpload( string $data, ?callable $progressCallback = null, - array $headers = [] + array $headers = [], + array $additionalResumableUploadOptions = [], + array $additionalCallOptions = [] ): UploadMediaResponse { $options = [ 'apiEndpoint' => self::SHOWCASE_HOST, @@ -54,16 +56,17 @@ private function createClientAndUpload( $options['credentials'] = new \Google\ApiCore\InsecureCredentialsWrapper(); } else { $options['hasEmulator'] = true; + $options['credentials'] = new \Google\ApiCore\InsecureCredentialsWrapper(); } $client = new ResumableUploadServiceClient($options); - $callOptions = [ + $callOptions = array_merge([ 'headers' => $headers - ]; - $resumableUploadOptions = [ + ], $additionalCallOptions); + $resumableUploadOptions = array_merge([ 'chunkSize' => 1024 - ]; + ], $additionalResumableUploadOptions); if ($progressCallback !== null) { $resumableUploadOptions['progressCallback'] = $progressCallback; } @@ -174,4 +177,108 @@ public function testFatalErrorOnStartThrowsException() $this->createClientAndUpload($payload, null, $headers); } + + public function testStallControlHappyPath() + { + $payload = 'data uploaded with stall control enabled and no delays'; + $resumableUploadOptions = [ + 'transferStallMinimumRate' => 10, + 'transferStallTimeout' => 5, + ]; + + $result = $this->createClientAndUpload($payload, null, [], $resumableUploadOptions); + $this->assertInstanceOf(UploadMediaResponse::class, $result); + } + + public function testStallControlUploadStalledOnDelay() + { + $payload = 'data exceeding chunk deadline stall timeout'; + $headers = [ + 'X-Goog-Test-Scenario-Config' => json_encode([ + 'delay_ms' => 1500, + ]) + ]; + $resumableUploadOptions = [ + 'transferStallMinimumRate' => 10, + 'transferStallTimeout' => 1, + ]; + + $this->expectException(ApiException::class); + $this->expectExceptionMessage('Upload stalled.'); + + $this->createClientAndUpload($payload, null, $headers, $resumableUploadOptions); + } + + public function testStallControlUploadStalledAfterOffset() + { + $chunkSize = 256 * 1024; + $payload = str_repeat('c', $chunkSize * 2); + $headers = [ + 'X-Goog-Test-Scenario-Config' => json_encode([ + 'delay_ms' => 1500, + 'after_offset' => $chunkSize, + ]) + ]; + $resumableUploadOptions = [ + 'chunkSize' => $chunkSize, + 'transferStallMinimumRate' => 10, + 'transferStallTimeout' => 1, + ]; + + $this->expectException(ApiException::class); + $this->expectExceptionMessage('Upload stalled.'); + + $this->createClientAndUpload($payload, null, $headers, $resumableUploadOptions); + } + + public function testStallControlChunkDeadlineExceededConvertsToUploadStalledWhenGlobalDeadlineNotExceeded() + { + $payload = 'data stalling with global deadline not exceeded'; + $headers = [ + 'X-Goog-Test-Scenario-Config' => json_encode([ + 'delay_ms' => 1500, + ]) + ]; + $resumableUploadOptions = [ + 'transferStallMinimumRate' => 10, + 'transferStallTimeout' => 1, + 'totalTimeoutMillis' => 30000, + ]; + + $this->expectException(ApiException::class); + $this->expectExceptionMessage('Upload stalled.'); + + $this->createClientAndUpload($payload, null, $headers, $resumableUploadOptions); + } + + public function testStallControlThrowsDeadlineExceededWhenGlobalDeadlineExceeded() + { + $payload = 'data stalling with global deadline exceeded'; + $headers = [ + 'X-Goog-Test-Scenario-Config' => json_encode([ + 'delay_ms' => 1500, + ]) + ]; + $resumableUploadOptions = [ + 'transferStallMinimumRate' => 10, + 'transferStallTimeout' => 5, + 'totalTimeoutMillis' => 500, + ]; + + $this->expectException(ApiException::class); + $this->expectExceptionMessage('Resumable upload total timeout exceeded.'); + + $this->createClientAndUpload($payload, null, $headers, $resumableUploadOptions); + } + + public function testStallControlDisabledWhenOnlyOneOptionProvided() + { + $payload = 'data with partial stall control options'; + $resumableUploadOptions = [ + 'transferStallMinimumRate' => 10, + ]; + + $result = $this->createClientAndUpload($payload, null, [], $resumableUploadOptions); + $this->assertInstanceOf(UploadMediaResponse::class, $result); + } } diff --git a/Gax/tests/Unit/ResumableUpload/ResumableUploadClientTest.php b/Gax/tests/Unit/ResumableUpload/ResumableUploadClientTest.php index 605ced35668..af992a7fd29 100644 --- a/Gax/tests/Unit/ResumableUpload/ResumableUploadClientTest.php +++ b/Gax/tests/Unit/ResumableUpload/ResumableUploadClientTest.php @@ -750,4 +750,689 @@ public function testRecoveryThrowsExceptionWhenMissingUploadStatusHeaderOnQuery( $client->startUpload($upload, Utils::streamFor('hello'), $call); } + public function testStallControlDisabledWhenOnlyOneOptionProvided() + { + $this->expectException(ApiException::class); + $this->expectExceptionMessage('Resumable upload total timeout exceeded.'); + + $currentTime = 0.0; + $requestBuilder = $this->prophesize(\Google\ApiCore\RequestBuilder::class); + $requestBuilder->build(Argument::any(), Argument::any(), Argument::any())->willReturn( + new \GuzzleHttp\Psr7\Request('POST', 'https://test.googleapis.com/test') + ); + + $httpHandler = function ($request, $options = []) use (&$currentTime) { + $currentTime = 700.0; // Advance time past 600s default global deadline + return \GuzzleHttp\Promise\Create::promiseFor(new \GuzzleHttp\Psr7\Response(200, [ + 'X-Goog-Upload-Status' => 'active', + 'X-Goog-Upload-URL' => 'https://upload.url/123' + ])); + }; + + $client = new ResumableUploadClient( + $this->createStubTransport($requestBuilder->reveal(), $httpHandler), + $this->prophesize(CredentialsWrapper::class)->reveal() + ); + $client->setClock(function () use (&$currentTime) { + return $currentTime; + }); + + $call = new Call('test.method', Timestamp::class, new Timestamp(), [], Call::RESUMABLE_UPLOAD_CALL); + $upload = new ResumableUpload($client, $call); + + // Only transferStallMinimumRate is set -> stall control disabled, default 10m global deadline applies + $upload->startUpload(Utils::streamFor('hello'), [ + 'transferStallMinimumRate' => 10.0, + ]); + } + + public function testStallControlDoesNotProvideDefaultGlobalDeadline() + { + $capturedOptions = []; + $httpHandler = function ($request, $options = []) use (&$capturedOptions) { + $capturedOptions[] = $options; + if (count($capturedOptions) === 1) { + return \GuzzleHttp\Promise\Create::promiseFor(new \GuzzleHttp\Psr7\Response(200, [ + 'X-Goog-Upload-Status' => 'active', + 'X-Goog-Upload-URL' => 'https://upload.url/123' + ])); + } + return \GuzzleHttp\Promise\Create::promiseFor(new \GuzzleHttp\Psr7\Response(200, [ + 'X-Goog-Upload-Status' => 'final' + ], '"1970-01-01T00:00:00Z"')); + }; + + $requestBuilder = $this->prophesize(\Google\ApiCore\RequestBuilder::class); + $requestBuilder->build(Argument::any(), Argument::any(), Argument::any())->willReturn( + new \GuzzleHttp\Psr7\Request('POST', 'https://test.googleapis.com/test') + ); + + $client = new ResumableUploadClient( + $this->createStubTransport($requestBuilder->reveal(), $httpHandler), + $this->prophesize(CredentialsWrapper::class)->reveal() + ); + + // Advance simulated time past default 10-minute (600s) timeout + $currentTime = 1000.0; + $client->setClock(function () use (&$currentTime) { + return $currentTime; + }); + + $call = new Call('test.method', Timestamp::class, new Timestamp(), [], Call::RESUMABLE_UPLOAD_CALL); + $upload = new ResumableUpload($client, $call); + + // Stall control enabled: minimum rate 10 MiB/s, stall timeout 60s + // totalTimeoutMillis is NOT set, so no default global deadline applies + $result = $upload->startUpload(Utils::streamFor('hello'), [ + 'transferStallMinimumRate' => 10.0, + 'transferStallTimeout' => 60.0, + ]); + + $this->assertInstanceOf(Timestamp::class, $result); + } + + public function testStallControlAppendix1WalkthroughWithClient() + { + // 160 MiB in bytes = 160 * 1048576 = 167772160 bytes + $chunkBytes = 167772160; + $currentTime = 0.0; + $capturedTimeouts = []; + + $requests = []; + $httpHandler = function ($request, $options = []) use (&$requests, &$currentTime, &$capturedTimeouts) { + $requests[] = $request; + if (isset($options['timeout'])) { + $capturedTimeouts[] = $options['timeout']; + } + + $count = count($requests); + if ($count === 1) { + // Initial start request at t = 0 + return \GuzzleHttp\Promise\Create::promiseFor(new \GuzzleHttp\Psr7\Response(200, [ + 'X-Goog-Upload-Status' => 'active', + 'X-Goog-Upload-URL' => 'https://upload.url/123' + ])); + } + if ($count === 2) { + // Chunk 1: transferred in 8s (t becomes 8) + $currentTime = 8.0; + return \GuzzleHttp\Promise\Create::promiseFor(new \GuzzleHttp\Psr7\Response(200, [ + 'X-Goog-Upload-Status' => 'active' + ])); + } + if ($count === 3) { + // Chunk 2: transferred in 26s (t becomes 34) + $currentTime = 34.0; + return \GuzzleHttp\Promise\Create::promiseFor(new \GuzzleHttp\Psr7\Response(200, [ + 'X-Goog-Upload-Status' => 'active' + ])); + } + if ($count === 4) { + // Chunk 3: transferred in 14s (t becomes 48) + $currentTime = 48.0; + return \GuzzleHttp\Promise\Create::promiseFor(new \GuzzleHttp\Psr7\Response(200, [ + 'X-Goog-Upload-Status' => 'active' + ])); + } + if ($count === 5) { + // Chunk 4: transferred in 10s (t becomes 58) + $currentTime = 58.0; + return \GuzzleHttp\Promise\Create::promiseFor(new \GuzzleHttp\Psr7\Response(200, [ + 'X-Goog-Upload-Status' => 'active' + ])); + } + if ($count === 6) { + // Chunk 5: transferred in 8s (t becomes 66) + $currentTime = 66.0; + return \GuzzleHttp\Promise\Create::promiseFor(new \GuzzleHttp\Psr7\Response(200, [ + 'X-Goog-Upload-Status' => 'final' + ], '"1970-01-01T00:00:00Z"')); + } + }; + + $requestBuilder = $this->prophesize(\Google\ApiCore\RequestBuilder::class); + $requestBuilder->build(Argument::any(), Argument::any(), Argument::any())->willReturn( + new \GuzzleHttp\Psr7\Request('POST', 'https://test.googleapis.com/test') + ); + + $client = new ResumableUploadClient( + $this->createStubTransport($requestBuilder->reveal(), $httpHandler), + $this->prophesize(CredentialsWrapper::class)->reveal() + ); + $client->setClock(function () use (&$currentTime) { + return $currentTime; + }); + + $call = new Call('test.method', Timestamp::class, new Timestamp(), [], Call::RESUMABLE_UPLOAD_CALL); + $upload = new ResumableUpload($client, $call); + + $result = $upload->startUpload($this->createSizedStream($chunkBytes * 5), [ + 'chunkSize' => $chunkBytes, + 'transferStallMinimumRate' => 10.0, + 'transferStallTimeout' => 60.0, + ]); + + $this->assertInstanceOf(Timestamp::class, $result); + $this->assertCount(6, $requests); + + // Next chunk timeouts from Appendix 1: [76, 76, 66, 68, 74] + $this->assertEquals([76, 76, 66, 68, 74], $capturedTimeouts); + } + + public function testStallControlUploadStalledRaisesApiException() + { + $this->expectException(ApiException::class); + $this->expectExceptionMessage('Upload stalled.'); + + $chunkBytes = 167772160; // 160 MiB + $currentTime = 0.0; + + $requests = []; + $httpHandler = function ($request, $options = []) use (&$requests, &$currentTime) { + $requests[] = $request; + $count = count($requests); + if ($count === 1) { + return \GuzzleHttp\Promise\Create::promiseFor(new \GuzzleHttp\Psr7\Response(200, [ + 'X-Goog-Upload-Status' => 'active', + 'X-Goog-Upload-URL' => 'https://upload.url/123' + ])); + } + if ($count === 2) { + // Chunk 1: transferred in 26s -> enters stall detection at t = 26 + $currentTime = 26.0; + return \GuzzleHttp\Promise\Create::promiseFor(new \GuzzleHttp\Psr7\Response(200, [ + 'X-Goog-Upload-Status' => 'active' + ])); + } + if ($count === 3) { + // Chunk 2: transfer takes 65s (from t = 26 to t = 91; 65 > stallTimeout 60) + $currentTime = 91.0; + return \GuzzleHttp\Promise\Create::promiseFor(new \GuzzleHttp\Psr7\Response(200, [ + 'X-Goog-Upload-Status' => 'final' + ], '"1970-01-01T00:00:00Z"')); + } + }; + + $requestBuilder = $this->prophesize(\Google\ApiCore\RequestBuilder::class); + $requestBuilder->build(Argument::any(), Argument::any(), Argument::any())->willReturn( + new \GuzzleHttp\Psr7\Request('POST', 'https://test.googleapis.com/test') + ); + + $client = new ResumableUploadClient( + $this->createStubTransport($requestBuilder->reveal(), $httpHandler), + $this->prophesize(CredentialsWrapper::class)->reveal() + ); + $client->setClock(function () use (&$currentTime) { + return $currentTime; + }); + + $call = new Call('test.method', Timestamp::class, new Timestamp(), [], Call::RESUMABLE_UPLOAD_CALL); + $upload = new ResumableUpload($client, $call); + + $upload->startUpload($this->createSizedStream($chunkBytes * 2), [ + 'chunkSize' => $chunkBytes, + 'transferStallMinimumRate' => 10.0, + 'transferStallTimeout' => 60.0, + ]); + } + + public function testStallControlChunkDeadlineExceededConvertsToUploadStalledWhenGlobalDeadlineNotSet() + { + $this->expectException(ApiException::class); + $this->expectExceptionMessage('Upload stalled.'); + + $chunkBytes = 167772160; // 160 MiB + $currentTime = 0.0; + + $requests = []; + $httpHandler = function ($request, $options = []) use (&$requests, &$currentTime) { + $requests[] = $request; + $count = count($requests); + if ($count === 1) { + return \GuzzleHttp\Promise\Create::promiseFor(new \GuzzleHttp\Psr7\Response(200, [ + 'X-Goog-Upload-Status' => 'active', + 'X-Goog-Upload-URL' => 'https://upload.url/123' + ])); + } + // Chunk transfer hangs past chunk deadline (chunk deadline is 76s, time reaches 80s) + $currentTime = 80.0; + return \GuzzleHttp\Promise\Create::rejectionFor( + new \GuzzleHttp\Exception\RequestException('cURL error 28: Operation timed out', $request) + ); + }; + + $requestBuilder = $this->prophesize(\Google\ApiCore\RequestBuilder::class); + $requestBuilder->build(Argument::any(), Argument::any(), Argument::any())->willReturn( + new \GuzzleHttp\Psr7\Request('POST', 'https://test.googleapis.com/test') + ); + + $client = new ResumableUploadClient( + $this->createStubTransport($requestBuilder->reveal(), $httpHandler), + $this->prophesize(CredentialsWrapper::class)->reveal() + ); + $client->setClock(function () use (&$currentTime) { + return $currentTime; + }); + + $call = new Call('test.method', Timestamp::class, new Timestamp(), [], Call::RESUMABLE_UPLOAD_CALL); + $upload = new ResumableUpload($client, $call); + + $upload->startUpload($this->createSizedStream($chunkBytes), [ + 'chunkSize' => $chunkBytes, + 'transferStallMinimumRate' => 10.0, + 'transferStallTimeout' => 60.0, + ]); + } + + public function testStallControlChunkDeadlineExceededConvertsToUploadStalledWhenGlobalDeadlineNotExceeded() + { + $this->expectException(ApiException::class); + $this->expectExceptionMessage('Upload stalled.'); + + $chunkBytes = 167772160; // 160 MiB + $currentTime = 0.0; + + $requests = []; + $httpHandler = function ($request, $options = []) use (&$requests, &$currentTime) { + $requests[] = $request; + $count = count($requests); + if ($count === 1) { + return \GuzzleHttp\Promise\Create::promiseFor(new \GuzzleHttp\Psr7\Response(200, [ + 'X-Goog-Upload-Status' => 'active', + 'X-Goog-Upload-URL' => 'https://upload.url/123' + ])); + } + // Chunk deadline is 76s. Time reaches 80s. + // Global deadline is 1000s (not exceeded). + $currentTime = 80.0; + return \GuzzleHttp\Promise\Create::rejectionFor( + new \GuzzleHttp\Exception\RequestException('cURL error 28: Operation timed out', $request) + ); + }; + + $requestBuilder = $this->prophesize(\Google\ApiCore\RequestBuilder::class); + $requestBuilder->build(Argument::any(), Argument::any(), Argument::any())->willReturn( + new \GuzzleHttp\Psr7\Request('POST', 'https://test.googleapis.com/test') + ); + + $client = new ResumableUploadClient( + $this->createStubTransport($requestBuilder->reveal(), $httpHandler), + $this->prophesize(CredentialsWrapper::class)->reveal() + ); + $client->setClock(function () use (&$currentTime) { + return $currentTime; + }); + + $call = new Call('test.method', Timestamp::class, new Timestamp(), [], Call::RESUMABLE_UPLOAD_CALL); + $upload = new ResumableUpload($client, $call); + + $upload->startUpload($this->createSizedStream($chunkBytes), [ + 'chunkSize' => $chunkBytes, + 'transferStallMinimumRate' => 10.0, + 'transferStallTimeout' => 60.0, + 'totalTimeoutMillis' => 1000000, // 1000 seconds + ]); + } + + public function testStallControlChunkDeadlineExceededThrowsDeadlineExceededWhenGlobalDeadlineExceeded() + { + $this->expectException(ApiException::class); + $this->expectExceptionMessage('Resumable upload total timeout exceeded.'); + + $chunkBytes = 167772160; // 160 MiB + $currentTime = 0.0; + + $requests = []; + $httpHandler = function ($request, $options = []) use (&$requests, &$currentTime) { + $requests[] = $request; + $count = count($requests); + if ($count === 1) { + return \GuzzleHttp\Promise\Create::promiseFor(new \GuzzleHttp\Psr7\Response(200, [ + 'X-Goog-Upload-Status' => 'active', + 'X-Goog-Upload-URL' => 'https://upload.url/123' + ])); + } + // Global deadline is 50s. Chunk timeout is 76s, trimmed to 50s. + // Time reaches 55s -> global deadline exceeded. + $currentTime = 55.0; + return \GuzzleHttp\Promise\Create::rejectionFor( + new \GuzzleHttp\Exception\RequestException('cURL error 28: Operation timed out', $request) + ); + }; + + $requestBuilder = $this->prophesize(\Google\ApiCore\RequestBuilder::class); + $requestBuilder->build(Argument::any(), Argument::any(), Argument::any())->willReturn( + new \GuzzleHttp\Psr7\Request('POST', 'https://test.googleapis.com/test') + ); + + $client = new ResumableUploadClient( + $this->createStubTransport($requestBuilder->reveal(), $httpHandler), + $this->prophesize(CredentialsWrapper::class)->reveal() + ); + $client->setClock(function () use (&$currentTime) { + return $currentTime; + }); + + $call = new Call('test.method', Timestamp::class, new Timestamp(), [], Call::RESUMABLE_UPLOAD_CALL); + $upload = new ResumableUpload($client, $call); + + try { + $upload->startUpload($this->createSizedStream($chunkBytes), [ + 'chunkSize' => $chunkBytes, + 'transferStallMinimumRate' => 10.0, + 'transferStallTimeout' => 60.0, + 'totalTimeoutMillis' => 50000, // 50 seconds + ]); + } catch (ApiException $e) { + $this->assertSame(\Google\Rpc\Code::DEADLINE_EXCEEDED, $e->getCode()); + $this->assertSame('Resumable upload total timeout exceeded.', $e->getMessage()); + throw $e; + } + } + + public function testStallControlRecoveryAndRetryTimeouts() + { + $chunkBytes = 167772160; // 160 MiB + $currentTime = 0.0; + $capturedRecoveryTimeout = null; + $capturedRetryTimeout = null; + + $requests = []; + $httpHandler = function ( + $request, + $options = [] + ) use ( + &$requests, + &$currentTime, + &$capturedRecoveryTimeout, + &$capturedRetryTimeout + ) { + $requests[] = $request; + $count = count($requests); + if ($count === 1) { + return \GuzzleHttp\Promise\Create::promiseFor(new \GuzzleHttp\Psr7\Response(200, [ + 'X-Goog-Upload-Status' => 'active', + 'X-Goog-Upload-URL' => 'https://upload.url/123', + 'X-Goog-Upload-Chunk-Granularity' => '1' + ])); + } + if ($count === 2) { + // Chunk 1 attempt 1 takes 10s then returns 308 (recovery needed) + $currentTime = 10.0; + return \GuzzleHttp\Promise\Create::promiseFor(new \GuzzleHttp\Psr7\Response(308)); + } + if ($count === 3) { + // Query recovery request uses stallTimeout (60s) + $capturedRecoveryTimeout = $options['timeout'] ?? null; + $currentTime = 11.0; + return \GuzzleHttp\Promise\Create::promiseFor(new \GuzzleHttp\Psr7\Response(200, [ + 'X-Goog-Upload-Status' => 'active', + 'X-Goog-Upload-Size-Received' => '0' + ])); + } + if ($count === 4) { + // Retried chunk upload request uses calculated next chunk timeout (76s) + $capturedRetryTimeout = $options['timeout'] ?? null; + $currentTime = 20.0; + return \GuzzleHttp\Promise\Create::promiseFor(new \GuzzleHttp\Psr7\Response(200, [ + 'X-Goog-Upload-Status' => 'final' + ], '"1970-01-01T00:00:00Z"')); + } + }; + + $requestBuilder = $this->prophesize(\Google\ApiCore\RequestBuilder::class); + $requestBuilder->build(Argument::any(), Argument::any(), Argument::any())->willReturn( + new \GuzzleHttp\Psr7\Request('POST', 'https://test.googleapis.com/test') + ); + + $client = new ResumableUploadClient( + $this->createStubTransport($requestBuilder->reveal(), $httpHandler), + $this->prophesize(CredentialsWrapper::class)->reveal() + ); + $client->setClock(function () use (&$currentTime) { + return $currentTime; + }); + + $call = new Call('test.method', Timestamp::class, new Timestamp(), [], Call::RESUMABLE_UPLOAD_CALL); + $upload = new ResumableUpload($client, $call); + + $result = $upload->startUpload($this->createSizedStream($chunkBytes), [ + 'chunkSize' => $chunkBytes, + 'transferStallMinimumRate' => 10.0, + 'transferStallTimeout' => 60.0, + ]); + + $this->assertInstanceOf(Timestamp::class, $result); + $this->assertEquals(60, $capturedRecoveryTimeout); + $this->assertEquals(76, $capturedRetryTimeout); + } + + public function testStallControlDoesNotApplyAttemptTimeoutSlicing() + { + $chunkBytes = 167772160; // 160 MiB + $capturedTimeout = null; + + $httpHandler = function ($request, $options = []) use (&$capturedTimeout) { + static $count = 0; + $count++; + if ($count === 1) { + return \GuzzleHttp\Promise\Create::promiseFor(new \GuzzleHttp\Psr7\Response(200, [ + 'X-Goog-Upload-Status' => 'active', + 'X-Goog-Upload-URL' => 'https://upload.url/123' + ])); + } + $capturedTimeout = $options['timeout'] ?? null; + return \GuzzleHttp\Promise\Create::promiseFor(new \GuzzleHttp\Psr7\Response(200, [ + 'X-Goog-Upload-Status' => 'final' + ], '"1970-01-01T00:00:00Z"')); + }; + + $requestBuilder = $this->prophesize(\Google\ApiCore\RequestBuilder::class); + $requestBuilder->build(Argument::any(), Argument::any(), Argument::any())->willReturn( + new \GuzzleHttp\Psr7\Request('POST', 'https://test.googleapis.com/test') + ); + + $client = new ResumableUploadClient( + $this->createStubTransport($requestBuilder->reveal(), $httpHandler), + $this->prophesize(CredentialsWrapper::class)->reveal() + ); + + $call = new Call('test.method', Timestamp::class, new Timestamp(), [], Call::RESUMABLE_UPLOAD_CALL); + $upload = new ResumableUpload($client, $call); + + $upload->startUpload($this->createSizedStream($chunkBytes), [ + 'chunkSize' => $chunkBytes, + 'transferStallMinimumRate' => 10.0, + 'transferStallTimeout' => 60.0, + ]); + + // Expected full chunk timeout: 160 / 10 + 60 = 76s. + // If attempt slicing (2.0 * 160 / 10 = 32s) was applied, it would have been 32. + $this->assertEquals(76, $capturedTimeout); + } + + public function testStallControlChunkTimeoutTrimsToGlobalDeadline() + { + $chunkBytes = 167772160; // 160 MiB + $capturedTimeout = null; + + $httpHandler = function ($request, $options = []) use (&$capturedTimeout) { + static $count = 0; + $count++; + if ($count === 1) { + return \GuzzleHttp\Promise\Create::promiseFor(new \GuzzleHttp\Psr7\Response(200, [ + 'X-Goog-Upload-Status' => 'active', + 'X-Goog-Upload-URL' => 'https://upload.url/123' + ])); + } + $capturedTimeout = $options['timeout'] ?? null; + return \GuzzleHttp\Promise\Create::promiseFor(new \GuzzleHttp\Psr7\Response(200, [ + 'X-Goog-Upload-Status' => 'final' + ], '"1970-01-01T00:00:00Z"')); + }; + + $requestBuilder = $this->prophesize(\Google\ApiCore\RequestBuilder::class); + $requestBuilder->build(Argument::any(), Argument::any(), Argument::any())->willReturn( + new \GuzzleHttp\Psr7\Request('POST', 'https://test.googleapis.com/test') + ); + + $currentTime = 10.0; + $client = new ResumableUploadClient( + $this->createStubTransport($requestBuilder->reveal(), $httpHandler), + $this->prophesize(CredentialsWrapper::class)->reveal() + ); + $client->setClock(function () use (&$currentTime) { + return $currentTime; + }); + + $call = new Call('test.method', Timestamp::class, new Timestamp(), [], Call::RESUMABLE_UPLOAD_CALL); + $upload = new ResumableUpload($client, $call); + + // Global deadline started at t = 10s with totalTimeoutMillis = 40000ms => deadline at t = 50s. + // Chunk timeout would be 76s, but remaining global deadline is 50 - 10 = 40s. + $upload->startUpload($this->createSizedStream($chunkBytes), [ + 'chunkSize' => $chunkBytes, + 'transferStallMinimumRate' => 10.0, + 'transferStallTimeout' => 60.0, + 'totalTimeoutMillis' => 40000, + ]); + + $this->assertEquals(40, $capturedTimeout); + } + + public function testStallControlRecoveryTrimsToGlobalDeadline() + { + $chunkBytes = 167772160; // 160 MiB + $currentTime = 0.0; + $capturedRecoveryTimeout = null; + + $requests = []; + $httpHandler = function ($request, $options = []) use (&$requests, &$currentTime, &$capturedRecoveryTimeout) { + $requests[] = $request; + $count = count($requests); + if ($count === 1) { + return \GuzzleHttp\Promise\Create::promiseFor(new \GuzzleHttp\Psr7\Response(200, [ + 'X-Goog-Upload-Status' => 'active', + 'X-Goog-Upload-URL' => 'https://upload.url/123', + 'X-Goog-Upload-Chunk-Granularity' => '1' + ])); + } + if ($count === 2) { + // Chunk 1 fails at t = 10s + $currentTime = 10.0; + return \GuzzleHttp\Promise\Create::promiseFor(new \GuzzleHttp\Psr7\Response(308)); + } + if ($count === 3) { + // Recovery request at t = 10s. + // Global deadline started at t = 0 with 25s totalTimeout => expires at t = 25s. + // Remaining global deadline is 25 - 10 = 15s (< 60s stallTimeout). + $capturedRecoveryTimeout = $options['timeout'] ?? null; + $currentTime = 11.0; + return \GuzzleHttp\Promise\Create::promiseFor(new \GuzzleHttp\Psr7\Response(200, [ + 'X-Goog-Upload-Status' => 'active', + 'X-Goog-Upload-Size-Received' => '0' + ])); + } + return \GuzzleHttp\Promise\Create::promiseFor(new \GuzzleHttp\Psr7\Response(200, [ + 'X-Goog-Upload-Status' => 'final' + ], '"1970-01-01T00:00:00Z"')); + }; + + $requestBuilder = $this->prophesize(\Google\ApiCore\RequestBuilder::class); + $requestBuilder->build(Argument::any(), Argument::any(), Argument::any())->willReturn( + new \GuzzleHttp\Psr7\Request('POST', 'https://test.googleapis.com/test') + ); + + $client = new ResumableUploadClient( + $this->createStubTransport($requestBuilder->reveal(), $httpHandler), + $this->prophesize(CredentialsWrapper::class)->reveal() + ); + $client->setClock(function () use (&$currentTime) { + return $currentTime; + }); + + $call = new Call('test.method', Timestamp::class, new Timestamp(), [], Call::RESUMABLE_UPLOAD_CALL); + $upload = new ResumableUpload($client, $call); + + $upload->startUpload($this->createSizedStream($chunkBytes), [ + 'chunkSize' => $chunkBytes, + 'transferStallMinimumRate' => 10.0, + 'transferStallTimeout' => 60.0, + 'totalTimeoutMillis' => 25000, // 25s global deadline + ]); + + // Trims 60s stallTimeout to 15s remaining global deadline + $this->assertEquals(15, $capturedRecoveryTimeout); + } + + private function createSizedStream(int $totalBytes): StreamInterface + { + return new class($totalBytes) implements StreamInterface { + private int $pos = 0; + public function __construct(private int $totalSize) + { + } + public function __toString(): string + { + return ''; + } + public function close(): void + { + } + public function detach() + { + return null; + } + public function getSize(): ?int + { + return $this->totalSize; + } + public function tell(): int + { + return $this->pos; + } + public function eof(): bool + { + return $this->pos >= $this->totalSize; + } + public function isSeekable(): bool + { + return true; + } + public function seek(int $offset, int $whence = SEEK_SET): void + { + $this->pos = $offset; + } + public function rewind(): void + { + $this->pos = 0; + } + public function isWritable(): bool + { + return false; + } + public function write(string $string): int + { + return 0; + } + public function isReadable(): bool + { + return true; + } + public function read(int $length): string + { + $len = min($length, $this->totalSize - $this->pos); + $this->pos += $len; + return str_repeat('a', $len); + } + public function getContents(): string + { + return ''; + } + public function getMetadata(?string $key = null) + { + return null; + } + }; + } } diff --git a/Gax/tests/Unit/ResumableUpload/ResumableUploadStateTest.php b/Gax/tests/Unit/ResumableUpload/ResumableUploadStateTest.php index 33cf829006b..574a174712d 100644 --- a/Gax/tests/Unit/ResumableUpload/ResumableUploadStateTest.php +++ b/Gax/tests/Unit/ResumableUpload/ResumableUploadStateTest.php @@ -240,4 +240,83 @@ public function testReconcileRecoveryOffsetExhaustsAttemptsWhenOffsetUnchanged() $this->expectExceptionMessage('Exhausted recovery attempts with unchanged offset'); $state->reconcileRecoveryOffset(10, $stream, 3); } + + public function testCalculateNextChunkTimeoutDisabledPolicy() + { + $state = new ResumableUploadState(10, null, null, 'starting'); + $this->assertSame(0.0, $state->calculateNextChunkTimeout(160.0)); + } + + public function testCalculateNextChunkTimeoutEnabledPolicy() + { + $state = new ResumableUploadState(10, null, null, 'starting', 10, 60); + + // Expected time = 160 / 10 = 16, lag = 0, timeout = 60 => 16 - 0 + 60 = 76 + $this->assertSame(76.0, $state->calculateNextChunkTimeout(160.0)); + + // When lag is 10: 16 - 10 + 60 = 66 + $state->lag = 10.0; + $this->assertSame(66.0, $state->calculateNextChunkTimeout(160.0)); + } + + public function testRecordChunkTransferAppendix1Walkthrough() + { + $state = new ResumableUploadState(10, null, null, 'transmitting', 10, 60); + + // Initial values: lag: 0 (s), Not in stall detection + $this->assertSame(0.0, $state->lag); + $this->assertNull($state->timeoutStarted); + $this->assertSame(76.0, $state->calculateNextChunkTimeout(160.0)); + + // Chunk 1: 160 MiB in 8 seconds at t = 8. + $state->recordChunkTransfer(160.0, 8.0, 8.0); + // lag cannot go negative => lag = 0 + $this->assertSame(0.0, $state->lag); + $this->assertNull($state->timeoutStarted); + $this->assertSame(76.0, $state->calculateNextChunkTimeout(160.0)); + + // Chunk 2: 160 MiB in 26 seconds at t = 34. + $state->recordChunkTransfer(160.0, 26.0, 34.0); + // lag is 10, timeout clock starts at 34 + $this->assertSame(10.0, $state->lag); + $this->assertSame(34.0, $state->timeoutStarted); + $this->assertSame(66.0, $state->calculateNextChunkTimeout(160.0)); + + // Chunk 3: 160 MiB in 14 seconds at t = 48. + $state->recordChunkTransfer(160.0, 14.0, 48.0); + // lag is 8, timeoutStarted remains 34, 14s elapsed < 60s + $this->assertSame(8.0, $state->lag); + $this->assertSame(34.0, $state->timeoutStarted); + $this->assertSame(68.0, $state->calculateNextChunkTimeout(160.0)); + + // Chunk 4: 160 MiB in 10 seconds at t = 58. + $state->recordChunkTransfer(160.0, 10.0, 58.0); + // lag is 2, timeoutStarted remains 34, 24s elapsed < 60s + $this->assertSame(2.0, $state->lag); + $this->assertSame(34.0, $state->timeoutStarted); + $this->assertSame(74.0, $state->calculateNextChunkTimeout(160.0)); + + // Chunk 5: 160 MiB in 8 seconds at t = 66. + $state->recordChunkTransfer(160.0, 8.0, 66.0); + // lag is 0, exit stall detection + $this->assertSame(0.0, $state->lag); + $this->assertNull($state->timeoutStarted); + $this->assertSame(76.0, $state->calculateNextChunkTimeout(160.0)); + } + + public function testRecordChunkTransferThrowsStalledException() + { + $this->expectException(ApiException::class); + $this->expectExceptionMessage('Upload stalled.'); + + $state = new ResumableUploadState(10, null, null, 'transmitting', 10, 60); + + // Chunk 1: slow chunk puts it into stall detection at t = 26 + $state->recordChunkTransfer(160.0, 26.0, 26.0); + $this->assertSame(10.0, $state->lag); + $this->assertSame(26.0, $state->timeoutStarted); + + // Chunk 2 finishes at t = 95 (elapsed 69s since stall began at 26; 69 > 60) + $state->recordChunkTransfer(160.0, 69.0, 95.0); + } }