feat(Gax): support chunkSize on resume and surface actual chunk size

Gax/src/GapicClientTrait.php

@@ -919,7 +919,8 @@ private function startResumableUploadCall(
             $this->resumableUploadClient,
             $call,
             $optionalArgs,
-            $optionalArgs['uploadUrl'] ?? null
+            $optionalArgs['uploadUrl'] ?? null,
+            $optionalArgs['chunkSize'] ?? null
         );
     }
 

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,

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

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

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

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()));
+    }
 }

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) {

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');