From 395374ad877797df3ab921105b2b835f49e7c5bd Mon Sep 17 00:00:00 2001 From: Brent Shaffer Date: Wed, 2 Sep 2026 08:01:09 +0000 Subject: [PATCH] feat(Gax): support chunkSize option on resume and surface actual chunk size --- Gax/src/GapicClientTrait.php | 3 +- Gax/src/ResumableUpload/ResumableUpload.php | 35 ++++++++++++- .../ResumableUpload/ResumableUploadClient.php | 17 ++++++- .../ResumableUpload/ResumableUploadState.php | 4 +- .../ResumableUpload/ResumableUploadTrait.php | 5 ++ .../ResumableUploadClientTest.php | 51 +++++++++++++++++++ .../ResumableUpload/ResumableUploadTest.php | 45 ++++++++++++++++ .../ResumableUploadTraitTest.php | 11 +++- 8 files changed, 165 insertions(+), 6 deletions(-) diff --git a/Gax/src/GapicClientTrait.php b/Gax/src/GapicClientTrait.php index fd9d570353e..dfd25f6e889 100644 --- a/Gax/src/GapicClientTrait.php +++ b/Gax/src/GapicClientTrait.php @@ -919,7 +919,8 @@ private function startResumableUploadCall( $this->resumableUploadClient, $call, $optionalArgs, - $optionalArgs['uploadUrl'] ?? null + $optionalArgs['uploadUrl'] ?? null, + $optionalArgs['chunkSize'] ?? null ); } diff --git a/Gax/src/ResumableUpload/ResumableUpload.php b/Gax/src/ResumableUpload/ResumableUpload.php index f08feffd608..1030dea798e 100644 --- a/Gax/src/ResumableUpload/ResumableUpload.php +++ b/Gax/src/ResumableUpload/ResumableUpload.php @@ -63,12 +63,14 @@ class ResumableUpload * } * @param ?string $uploadUrl An existing resumable upload session URL to resume an upload * across process restarts or interruptions. + * @param ?int $chunkSize Optional. The chunk size in bytes for the upload. */ public function __construct( private ResumableUploadClient $resumableUploadClient, private Call $call, private array $callOptions = [], - private ?string $uploadUrl = null + private ?string $uploadUrl = null, + private ?int $chunkSize = null ) { } @@ -95,12 +97,39 @@ public function setUploadUrl(string $uploadUrl): void $this->uploadUrl = $uploadUrl; } + /** + * Returns the actual chunk size in bytes used for the upload, if determined. + * This may be the user-specified chunk size, or the chunk size adjusted to + * match the server-specified chunk granularity. + * + * @return ?int + */ + public function getChunkSize(): ?int + { + return $this->chunkSize; + } + + /** + * Sets the chunk size in bytes for the upload. + * + * @param int $chunkSize + * @return void + */ + public function setChunkSize(int $chunkSize): void + { + $this->chunkSize = $chunkSize; + } + /** * Starts or resumes the resumable upload exchange using the provided data stream. * If this instance already has an `uploadUrl` (e.g. created via `$client->resumeUpload($methodName, $uploadUrl)` * or after a previous start/interruption), calling `startUpload($dataStream, $resumableUploadOptions)` queries * the server for the current byte offset and resumes transmitting remaining chunks. * + * When resuming an upload, it is recommended that the data stream is rewound to offset 0. + * If the stream is not rewound and not seekable, an error may occur if unconfirmed + * chunks must be re-read. + * * @param StreamInterface $dataStream * @param array $resumableUploadOptions { * Optional. @@ -119,6 +148,10 @@ public function setUploadUrl(string $uploadUrl): void */ public function startUpload(StreamInterface $dataStream, array $resumableUploadOptions = []): Message { + if ($this->chunkSize !== null && !isset($resumableUploadOptions['chunkSize'])) { + $resumableUploadOptions['chunkSize'] = $this->chunkSize; + } + return $this->resumableUploadClient->startUpload( $this, $dataStream, diff --git a/Gax/src/ResumableUpload/ResumableUploadClient.php b/Gax/src/ResumableUpload/ResumableUploadClient.php index 639b7c67e58..6fa80489183 100644 --- a/Gax/src/ResumableUpload/ResumableUploadClient.php +++ b/Gax/src/ResumableUpload/ResumableUploadClient.php @@ -130,12 +130,17 @@ public function startUpload( ?? self::DEFAULT_TOTAL_TIMEOUT_MILLIS); $deadlineMs = microtime(true) * 1000 + $totalTimeoutMillis; + $chunkSize = $resumableUploadOptions['chunkSize'] + ?? $upload->getChunkSize() + ?? self::DEFAULT_CHUNK_SIZE; + $state = new ResumableUploadState( - $resumableUploadOptions['chunkSize'] ?? self::DEFAULT_CHUNK_SIZE, + $chunkSize, $resumableUploadOptions['progressCallback'] ?? null, $uploadUrl, $uploadUrl !== null ? self::PHASE_RECOVERY : self::PHASE_STARTING ); + $upload->setChunkSize($chunkSize); while ($state->phase !== self::PHASE_DONE) { $this->checkDeadline($deadlineMs); @@ -232,6 +237,16 @@ private function phaseStarting( } $granularityHeader = $response->getHeaderLine('X-Goog-Upload-Chunk-Granularity'); $state->chunkGranularity = !empty($granularityHeader) ? (int) $granularityHeader : 1; + if ($state->chunkGranularity > 0 && ($state->chunkSize % $state->chunkGranularity !== 0)) { + $state->chunkSize = (int) ( + floor($state->chunkSize / $state->chunkGranularity) * $state->chunkGranularity + ); + if ($state->chunkSize === 0) { + $state->chunkSize = $state->chunkGranularity; + } + } + $upload->setChunkSize($state->chunkSize); + $statusHeader = $response->getHeaderLine('X-Goog-Upload-Status'); if (empty($statusHeader)) { // Missing X-Goog-Upload-Status header on Start is a Category 1 transient error diff --git a/Gax/src/ResumableUpload/ResumableUploadState.php b/Gax/src/ResumableUpload/ResumableUploadState.php index 0522ce1d393..a0ce3dcc5f7 100644 --- a/Gax/src/ResumableUpload/ResumableUploadState.php +++ b/Gax/src/ResumableUpload/ResumableUploadState.php @@ -82,6 +82,7 @@ public function prepareBuffer(StreamInterface $dataStream): void if ($effectiveChunkSize === 0) { $effectiveChunkSize = $this->chunkGranularity; } + $this->chunkSize = $effectiveChunkSize; } if ($this->committedOffset > 0 && $dataStream->tell() !== $this->committedOffset) { @@ -111,7 +112,8 @@ public function prepareBuffer(StreamInterface $dataStream): void $e ); } - $this->isEof = $dataStream->eof(); + $this->isEof = $dataStream->eof() + || ($dataStream->getSize() !== null && $dataStream->tell() >= $dataStream->getSize()); } public function commitBuffer(): void diff --git a/Gax/src/ResumableUpload/ResumableUploadTrait.php b/Gax/src/ResumableUpload/ResumableUploadTrait.php index 8bd06106a1f..5750ab0b876 100644 --- a/Gax/src/ResumableUpload/ResumableUploadTrait.php +++ b/Gax/src/ResumableUpload/ResumableUploadTrait.php @@ -44,11 +44,16 @@ trait ResumableUploadTrait /** * Resume an existing resumable upload session. * + * When resuming an upload, it is recommended that the data stream is rewound to offset 0. + * If the stream is not rewound and not seekable, an error may occur if unconfirmed + * chunks must be re-read. + * * @param string $uploadUrl The resumable upload session URL. * @param string $methodName The API method name. * @param array $optionalArgs { * Optional. * + * @type int $chunkSize Optional. The chunk size in bytes to use for resuming the upload. * @type array $headers Optional. Key-value array of custom HTTP headers to * include with upload requests. * @type int $timeoutMillis Optional. The timeout in milliseconds for the diff --git a/Gax/tests/Unit/ResumableUpload/ResumableUploadClientTest.php b/Gax/tests/Unit/ResumableUpload/ResumableUploadClientTest.php index 605ced35668..c5d46ba5a0b 100644 --- a/Gax/tests/Unit/ResumableUpload/ResumableUploadClientTest.php +++ b/Gax/tests/Unit/ResumableUpload/ResumableUploadClientTest.php @@ -750,4 +750,55 @@ public function testRecoveryThrowsExceptionWhenMissingUploadStatusHeaderOnQuery( $client->startUpload($upload, Utils::streamFor('hello'), $call); } + + public function testStartUploadSurfacesActualChunkSizeAndAdjustsForGranularity() + { + $requests = []; + $httpHandler = function ($request, $options = []) use (&$requests) { + $requests[] = $request; + if (count($requests) === 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' => '256' + ])); + } + if (count($requests) === 2) { + return \GuzzleHttp\Promise\Create::promiseFor(new \GuzzleHttp\Psr7\Response(200, [ + 'X-Goog-Upload-Status' => 'active' + ])); + } + 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())->will(function ($args) { + return new \GuzzleHttp\Psr7\Request('POST', 'https://test.googleapis.com/' . $args[0], $args[2] ?? []); + }); + + $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 1000 bytes with requested chunkSize 600. + // Server specifies granularity 256. 600 adjusted down to closest multiple is 512. + $payload = str_repeat('a', 1000); + $client->startUpload($upload, Utils::streamFor($payload), $call, [], [ + 'chunkSize' => 600 + ]); + + $this->assertEquals(512, $upload->getChunkSize()); + $this->assertCount(3, $requests); + $this->assertEquals('start', $requests[0]->getHeaderLine('X-Goog-Upload-Command')); + $this->assertEquals('upload', $requests[1]->getHeaderLine('X-Goog-Upload-Command')); + $this->assertSame(512, strlen((string) $requests[1]->getBody())); + $this->assertEquals('upload, finalize', $requests[2]->getHeaderLine('X-Goog-Upload-Command')); + $this->assertSame(488, strlen((string) $requests[2]->getBody())); + } } diff --git a/Gax/tests/Unit/ResumableUpload/ResumableUploadTest.php b/Gax/tests/Unit/ResumableUpload/ResumableUploadTest.php index 5153cf8fc8d..8bc057d4ef9 100644 --- a/Gax/tests/Unit/ResumableUpload/ResumableUploadTest.php +++ b/Gax/tests/Unit/ResumableUpload/ResumableUploadTest.php @@ -251,6 +251,51 @@ public function testInvalidInitializationThrowsException() new ResumableUpload($client, null); } + public function testChunkSizeGetterAndSetter() + { + $requestBuilder = $this->prophesize(\Google\ApiCore\RequestBuilder::class)->reveal(); + $client = new ResumableUploadClient( + $this->createStubTransport($requestBuilder, function () { + }), + $this->prophesize(CredentialsWrapper::class)->reveal() + ); + $call = new Call('v1/test:create', Timestamp::class, new Timestamp(), [], Call::RESUMABLE_UPLOAD_CALL); + $upload = new ResumableUpload($client, $call, [], null, 524288); + + $this->assertEquals(524288, $upload->getChunkSize()); + $upload->setChunkSize(1048576); + $this->assertEquals(1048576, $upload->getChunkSize()); + } + + public function testResumeWithChunkSize() + { + $requests = []; + $httpHandler = $this->createMockHttpHandler([ + new Response(200, ['X-Goog-Upload-Status' => 'active', 'X-Goog-Upload-Size-Received' => '0']), + new Response(200, ['X-Goog-Upload-Status' => 'active']), + new Response(200, ['X-Goog-Upload-Status' => 'final'], '"1970-01-01T00:00:00Z"') + ], $requests); + + $requestBuilder = $this->prophesize(\Google\ApiCore\RequestBuilder::class)->reveal(); + $client = new ResumableUploadClient( + $this->createStubTransport($requestBuilder, $httpHandler), + $this->prophesize(CredentialsWrapper::class)->reveal() + ); + $call = new Call('test.method', Timestamp::class, null, [], Call::RESUMABLE_UPLOAD_CALL); + $upload = new ResumableUpload($client, $call, [], 'https://upload.url/session123', 6); + + $stream = Utils::streamFor('hello world'); + $upload->startUpload($stream); + + $this->assertEquals(6, $upload->getChunkSize()); + $this->assertCount(3, $requests); + $this->assertEquals('query', $requests[0]->getHeaderLine('X-Goog-Upload-Command')); + $this->assertEquals('upload', $requests[1]->getHeaderLine('X-Goog-Upload-Command')); + $this->assertEquals('hello ', (string) $requests[1]->getBody()); + $this->assertEquals('upload, finalize', $requests[2]->getHeaderLine('X-Goog-Upload-Command')); + $this->assertEquals('world', (string) $requests[2]->getBody()); + } + private function createMockHttpHandler(array $responses, ?array &$requests = []): callable { return function ($request, $options = []) use (&$responses, &$requests) { diff --git a/Gax/tests/Unit/ResumableUpload/ResumableUploadTraitTest.php b/Gax/tests/Unit/ResumableUpload/ResumableUploadTraitTest.php index 8b46027275c..0f146344eba 100644 --- a/Gax/tests/Unit/ResumableUpload/ResumableUploadTraitTest.php +++ b/Gax/tests/Unit/ResumableUpload/ResumableUploadTraitTest.php @@ -94,16 +94,23 @@ public function getResumableUploadClient(): ResumableUploadClient $resumed = $client->resumeUpload( 'https://upload.url/session123', 'createYouTubeVideoUpload', - ['timeoutMillis' => 5000] + ['timeoutMillis' => 5000, 'chunkSize' => 1048576] ); $this->assertInstanceOf(ResumableUpload::class, $resumed); + $this->assertEquals(1048576, $resumed->getChunkSize()); + $this->assertEquals('https://upload.url/session123', $resumed->getUploadUrl()); $ref = new \ReflectionClass($resumed); $clientProp = $ref->getProperty('resumableUploadClient'); $this->assertSame($client->getResumableUploadClient(), $clientProp->getValue($resumed)); $optionsProp = $ref->getProperty('callOptions'); $this->assertSame( - ['timeoutMillis' => 5000, 'uploadUrl' => 'https://upload.url/session123', 'headers' => []], + [ + 'timeoutMillis' => 5000, + 'chunkSize' => 1048576, + 'uploadUrl' => 'https://upload.url/session123', + 'headers' => [] + ], $optionsProp->getValue($resumed) ); $callProp = $ref->getProperty('call');