feat(Bigtable): expose totalTimeoutMillis in RetrySettings for Read and Mutate Rows (#8223)

bshaffer · web-flow · commit 1b4a1c937d3b · 2025-06-18T20:19:44.000Z
diff --git a/Bigtable/src/BigtableClient.php b/Bigtable/src/BigtableClient.php
@@ -21,6 +21,8 @@
 use Google\ApiCore\ClientOptionsTrait;
 use Google\ApiCore\CredentialsWrapper;
 use Google\ApiCore\Serializer;
+use Google\ApiCore\Transport\GrpcTransport;
+use Google\ApiCore\Transport\RestTransport;
 use Google\ApiCore\Transport\TransportInterface;
 use Google\ApiCore\ValidationException;
 use Google\Auth\FetchAuthTokenInterface;
@@ -94,14 +96,13 @@ class BigtableClient
      *           This option accepts either a path to a credentials file, or a
      *           decoded credentials file as a PHP array.
      *           *Advanced usage*: In addition, this option can also accept a
-     *           pre-constructed {@see Google\Auth\FetchAuthTokenInterface} object
-     *           or {@see Google\ApiCore\CredentialsWrapper} object. Note that when
-     *           one of these objects are provided, any settings in
-     *           $config['credentialsConfig'] will be ignored.
+     *           pre-constructed {@see FetchAuthTokenInterface} object
+     *           or {@see CredentialsWrapper} object. Note that when one of
+     *           these objects are provided, any settings in
+     *           `$config['credentialsConfig']` will be ignored.
      *     @type array $credentialsConfig Options used to configure credentials,
      *           including auth token caching, for the client. For a full list of
-     *           supporting configuration options, see
-     *           {@see Google\ApiCore\CredentialsWrapper}.
+     *           supporting configuration options, see {@see CredentialsWrapper}.
      *     @type bool $disableRetries Determines whether or not retries defined by
      *           the client configuration should be disabled. **Defaults to**
      *           `false`.
@@ -112,9 +113,8 @@ class BigtableClient
      *           executing network requests. May be either the string `rest` or
      *           `grpc`. **Defaults to** `grpc` if gRPC support is detected on
      *           the system. *Advanced usage*: Additionally, it is possible to
-     *           pass in an already instantiated
-     *           {@see Google\ApiCore\Transport\TransportInterface} object. Note
-     *           that when this object is provided, any settings in
+     *           pass in an already instantiated {@see TransportInterface}
+     *           object. Note that when this object is provided, any settings in
      *           $config['transportConfig'] and the $config['apiEndpoint']
      *           setting will be ignored.
      *     @type array $transportConfig Configuration options that will be used to
@@ -124,9 +124,8 @@ class BigtableClient
      *               'grpc' => [...],
      *               'rest' => [...]
      *           ];
-     *           See the `build` method on {@see Google\ApiCore\Transport\GrpcTransport}
-     *           and {@see Google\ApiCore\Transport\RestTransport} for the
-     *           supported options.
+     *           See the `build` method on {@see GrpcTransport} and
+     *           {@see RestTransport} for the supported options.
      *     @type string $quotaProject Specifies a user project to bill for
      *           access charges associated with the request.
      *     @type bool $pingAndWarm EXPERIMENTAL When true, calls the
diff --git a/Bigtable/src/ResumableStream.php b/Bigtable/src/ResumableStream.php
@@ -84,6 +84,8 @@ class ResumableStream implements \IteratorAggregate
      */
     private $delayFunction;
 
+    private int $totalTimeoutMillis;
+
     /**
      * Constructs a resumable stream.
      *
@@ -94,11 +96,16 @@ class ResumableStream implements \IteratorAggregate
      *        calling `$apiFunction`.
      * @param callable $retryFunction Function which determines whether to retry or not.
      * @param array $callOptions {
-     *        @option RetrySettings|array $retrySettings {
-     *                @option int $maxRetries Number of times to retry. **Defaults to** `3`.
-     *                Only maxRetries works for RetrySettings in this API.
-     *            }
-     *   }
+     *     Configuration options.
+     *
+     *     @type RetrySettings|array $retrySettings {
+     *         Only $maxRetries and $totalTimeoutMillis work for RetrySettings in this API.
+     *
+     *         @type int $maxRetries Number of times to retry. **Defaults to** `3`.
+     *         @type int $totalTimeoutMillis The max accumulative timeout in total for the
+     *               operation. `-1` signifies no total timeout. **Defaults to** `-1`.
+     *     }
+     * }
      */
     public function __construct(
         GapicClient $gapicClient,
@@ -112,6 +119,7 @@ public function __construct(
         $this->method = $method;
         $this->request = $request;
         $this->retries = $this->getMaxRetries($callOptions);
+        $this->totalTimeoutMillis = $this->getTotalTimeoutMillis($callOptions);
         $this->argumentFunction = $argumentFunction;
         $this->retryFunction = $retryFunction;
         $this->callOptions = $callOptions;
@@ -149,7 +157,9 @@ public function readAll()
     {
         // Reset $currentAttempts on successful row read, but keep total attempts for the header.
         $currentAttempt = $totalAttempt = 0;
+        $startTimeMillis = floor(microtime(true) * 1000);
         do {
+            $this->checkTotalTimeout($startTimeMillis);
             $ex = null;
             list($this->request, $this->callOptions) =
                 ($this->argumentFunction)($this->request, $this->callOptions);
@@ -220,4 +230,32 @@ private function getMaxRetries(array $options): int
 
         return $retrySettings['maxRetries'] ?? ResumableStream::DEFAULT_MAX_RETRIES;
     }
+
+    private function getTotalTimeoutMillis(array $options): ?int
+    {
+        $retrySettings = $options['retrySettings'] ?? [];
+
+        if ($retrySettings instanceof RetrySettings) {
+            return $retrySettings->getTotalTimeoutMillis();
+        }
+
+        return $retrySettings['totalTimeoutMillis'] ?? -1;
+    }
+
+    private function checkTotalTimeout(int $startTimeMillis): void
+    {
+        if ($this->totalTimeoutMillis >= 0) {
+            $elapsedTimeMillis = floor(microtime(true) * 1000) - $startTimeMillis;
+            if ($elapsedTimeMillis > $this->totalTimeoutMillis) {
+                throw new ApiException('Operation timeout exceeeded ', Code::DEADLINE_EXCEEDED);
+            }
+            $remainingTimeMillis = $this->totalTimeoutMillis - $elapsedTimeMillis;
+            // Set timeoutMillis if it's unset or less than the remaining time
+            // in order to preempt requests from exceeding total timeout
+            if (!isset($this->callOptions['timeoutMillis'])
+                || $this->callOptions['timeoutMillis'] > $remainingTimeMillis) {
+                $this->callOptions['timeoutMillis'] = $remainingTimeMillis;
+            }
+        }
+    }
 }
diff --git a/Bigtable/src/Table.php b/Bigtable/src/Table.php
@@ -120,7 +120,16 @@ public function __construct(
      * @param array $rowMutations An associative array with the key being the
      *        row key and the value being the
      *        {@see \Google\Cloud\Bigtable\Mutations} to perform.
-     * @param array $options [optional] Configuration options.
+     * @param array $options {
+     *     Configuration options.
+     *
+     *     @option RetrySettings|array $retrySettings {
+     *         Only $maxRetries and $totalTimeoutMillis work for RetrySettings in this API.
+     *
+     *         @option int $maxRetries Number of times to retry. **Defaults to** `3`.
+     *         @option int $totalTimeoutMillis The max accumulative timeout in total.
+     *     }
+     * }
      * @return void
      * @throws ApiException|BigtableDataOperationException If the remote call fails or operation fails.
      * @throws \InvalidArgumentException If rowMutations is a list instead of associative array indexed by row key.
@@ -155,13 +164,15 @@ public function mutateRows(array $rowMutations, array $options = [])
      *
      * @param string $rowKey The row key of the row to mutate.
      * @param Mutations $mutations Mutations to apply on row.
-     * @param array $options [optional] {
+     * @param array $options {
      *     Configuration options.
      *
-     *        @param RetrySettings|array $retrySettings {
-     *               @option int $maxRetries Number of times to retry. **Defaults to** `3`.
-     *                  Only maxRetries works for RetrySettings in this API.
-     *               }
+     *     @option RetrySettings|array $retrySettings {
+     *         Only $maxRetries and $totalTimeoutMillis work for RetrySettings in this API.
+     *
+     *         @option int $maxRetries Number of times to retry. **Defaults to** `3`.
+     *         @option int $totalTimeoutMillis The max accumulative timeout in total.
+     *     }
      * }
      * @return void
      * @throws ApiException If the remote call fails.
@@ -201,10 +212,12 @@ public function mutateRow($rowKey, Mutations $mutations, array $options = [])
      * @param array $options [optional] {
      *     Configuration options.
      *
-     *     @param RetrySettings|array $retrySettings {
-     *            @option int $maxRetries Number of times to retry. **Defaults to** `3`.
-     *                Only maxRetries works for RetrySettings in this API.
-     *            }
+     *     @type RetrySettings|array $retrySettings {
+     *         Only $maxRetries and $totalTimeoutMillis work for RetrySettings in this API.
+     *
+     *         @type int $maxRetries Number of times to retry. **Defaults to** `3`.
+     *         @type int $totalTimeoutMillis The max accumulative timeout in total for the operation.
+     *     }
      * }
      * @return void
      * @throws ApiException|BigtableDataOperationException If the remote call fails or operation fails
@@ -274,10 +287,12 @@ public function upsert(array $rows, array $options = [])
      *           To learn more please see {@see \Google\Cloud\Bigtable\Filter} which
      *           provides static factory methods for the various filter types.
      *     @type int $rowsLimit The number of rows to scan.
-     * @param RetrySettings|array $retrySettings {
-     *        @option int $maxRetries Number of times to retry. **Defaults to** `3`.
-     *                Only maxRetries works for RetrySettings in this API.
-     *        }
+     *     @type RetrySettings|array $retrySettings {
+     *         Only $maxRetries and $totalTimeoutMillis work for RetrySettings in this API.
+     *
+     *         @type int $maxRetries Number of times to retry. **Defaults to** `3`.
+     *         @type int $totalTimeoutMillis The max accumulative timeout in total for the operation.
+     *     }
      * }
      * @return ChunkFormatter
      */
diff --git a/Bigtable/tests/Conformance/proxy/known_failures.txt b/Bigtable/tests/Conformance/proxy/known_failures.txt
@@ -1,8 +1,5 @@
 _Retry_WithRoutingCookie
 _Retry_WithRetryInfo
-TestMutateRows_Generic_DeadlineExceeded
-TestReadRow_Generic_DeadlineExceeded
 TestReadRows_Retry_StreamReset
 TestReadRows_Retry_LastScannedRow_Reverse
 TestReadRows_ReverseScans_FeatureFlag_Enabled
-TestReadRows_Generic_DeadlineExceeded
diff --git a/Bigtable/tests/Conformance/proxy/src/ProxyService.php b/Bigtable/tests/Conformance/proxy/src/ProxyService.php
@@ -133,10 +133,23 @@ public function ReadRow(GRPC\ContextInterface $ctx, Testproxy\ReadRowRequest $in
             'appProfileId' => $config->getAppProfileId(),
         ]);
 
-        $rowData = $table->readRow($in->getRowKey(), [
-            'filter' => $in->getFilter(),
-            'timeoutMillis' => $this->getTimeoutMillis($config->getPerOperationTimeout()),
-        ]);
+        $timeoutMillis = $this->getTimeoutMillis($config->getPerOperationTimeout());
+        try {
+            $rowData = $table->readRow($in->getRowKey(), [
+                'filter' => $in->getFilter(),
+                'retrySettings' => [
+                    // set total timeout for the operation (including retries)
+                    'totalTimeoutMillis' => $timeoutMillis,
+                ],
+                // the conformance tests check for this, but I'm not sure why
+                'rowsLimit' => 1,
+            ]);
+        } catch (\Google\ApiCore\ApiException $e) {
+            return $out->setStatus(new Status([
+                'code' => $e->getCode(),
+                'message' => $e->getMessage(),
+            ]));
+        }
 
         if ($rowData) {
             $row = $this->arrayToRowProto($rowData);
@@ -195,13 +208,16 @@ public function ReadRows(GRPC\ContextInterface $ctx, Testproxy\ReadRowsRequest $
 
         $this->logger->debug(json_encode($rowKeys));
 
+        $timeoutMillis = $this->getTimeoutMillis($config->getPerOperationTimeout());
         $stream = $table->readRows([
             'rowKeys' => $rowKeys,
             'rowRanges' => $ranges,
             'filter' => $request->getFilter(),
             'rowsLimit' => $request->getRowsLimit(),
             'reversed' => $request->getReversed(),
-            'timeoutMillis' => $this->getTimeoutMillis($config->getPerOperationTimeout()),
+            'retrySettings' => [
+                'totalTimeoutMillis' => $timeoutMillis,
+            ],
         ]);
 
         $rows = [];
@@ -309,25 +325,31 @@ public function BulkMutateRows(GRPC\ContextInterface $ctx, Testproxy\MutateRowsR
             $mutations[$entry->getRowKey()] = $this->protoToMutations($entry->getMutations());
         }
 
+        $timeoutMillis = $this->getTimeoutMillis($config->getPerOperationTimeout());
         try {
             $table->mutateRows($mutations, [
-                'timeoutMillis' => $this->getTimeoutMillis($config->getPerOperationTimeout())
+                'retrySettings' => [
+                    'totalTimeoutMillis' => $timeoutMillis,
+                ]
             ]);
         } catch (BigtableDataOperationException $e) {
             $failedEntries = [];
             foreach ($e->getMetadata() as $metadata) {
-                $status = new Status([
-                    'code' => $metadata['statusCode'],
-                    'message' => $metadata['message'],
-                ]);
                 $failedEntries[] = new Entry([
-                    'index' => $metadata['index'],
-                    'status' => $status,
+                    'index' => $metadata['index'] ?? 0,
+                    'status' => new Status([
+                        'code' => $metadata['statusCode'],
+                        'message' => $metadata['message'],
+                    ]),
                 ]);
             }
             $out->setEntries($failedEntries);
+            $out->setStatus(new Status([
+                'code' => $e->getCode(),
+                'message' => $e->getMessage(),
+            ]));
         } catch (\Google\ApiCore\ApiException $e) {
-            $out = $out->setStatus(new Status([
+            $out->setStatus(new Status([
                 'code' => $e->getCode(),
                 'message' => $e->getMessage(),
             ]));
