fix(Spanner): add retry for RST_STREAM

.github/workflows/spanner-emulator-system-tests.yaml

@@ -45,7 +45,8 @@ jobs:
 
       - name: Install dependencies
         run: |
-          composer config repositories.local  --json '{"type":"path", "url": "../Core", "options": {"versions": {"google/cloud-core": "1.45"}}}' -d Spanner
+          # ensure composer uses local Core instead of pulling from packagist
+          composer config repositories.local  --json '{"type":"path", "url": "../Core", "options": {"versions": {"google/cloud-core": "1.100"}}}' -d Spanner
           composer update --prefer-dist --no-interaction --no-suggest -d Spanner/
 
       - name: Run system tests

Spanner/composer.json

@@ -5,7 +5,7 @@
     "minimum-stability": "stable",
     "require": {
         "ext-grpc": "*",
-        "google/cloud-core": "^1.43",
+        "google/cloud-core": "^1.49",
         "google/gax": "^1.1"
     },
     "require-dev": {

Spanner/src/Database.php

@@ -20,6 +20,7 @@
 use Google\ApiCore\ValidationException;
 use Google\Cloud\Core\Exception\AbortedException;
 use Google\Cloud\Core\Exception\NotFoundException;
+use Google\Cloud\Core\Exception\ServiceException;
 use Google\Cloud\Core\Iam\Iam;
 use Google\Cloud\Core\Iterator\ItemIterator;
 use Google\Cloud\Core\LongRunning\LongRunningConnectionInterface;
@@ -37,6 +38,7 @@
 use Google\Cloud\Spanner\Transaction;
 use Google\Cloud\Spanner\V1\SpannerClient as GapicSpannerClient;
 use Google\Cloud\Spanner\V1\TypeCode;
+use Google\Rpc\Code;
 
 /**
  * Represents a Cloud Spanner Database.
@@ -871,11 +873,16 @@ public function runTransaction(callable $operation, array $options = [])
         };
 
         $delayFn = function (\Exception $e) {
-            if (!($e instanceof AbortedException)) {
-                throw $e;
+            if ($e instanceof AbortedException) {
+                return $e->getRetryDelay();
             }
-
-            return $e->getRetryDelay();
+            if ($e instanceof ServiceException &&
+                $e->getCode() === Code::INTERNAL &&
+                strpos($e->getMessage(), 'RST_STREAM') !== false
+            ) {
+                return $e->getRetryDelay();
+            }
+            throw $e;
         };
 
         $transactionFn = function ($operation, $session, $options) use ($startTransactionFn) {

Spanner/tests/Unit/DatabaseTest.php

@@ -19,6 +19,7 @@
 
 use Google\Cloud\Core\Exception\AbortedException;
 use Google\Cloud\Core\Exception\NotFoundException;
+use Google\Cloud\Core\Exception\ServerException;
 use Google\Cloud\Core\Iam\Iam;
 use Google\Cloud\Core\Iterator\ItemIterator;
 use Google\Cloud\Core\LongRunning\LongRunningConnectionInterface;
@@ -46,6 +47,7 @@
 use Google\Cloud\Spanner\Timestamp;
 use Google\Cloud\Spanner\Transaction;
 use Google\Cloud\Spanner\V1\SpannerClient;
+use Google\Rpc\Code;
 use Yoast\PHPUnitPolyfills\TestCases\TestCase;
 use Prophecy\Argument;
 use Yoast\PHPUnitPolyfills\Polyfills\ExpectException;
@@ -745,6 +747,31 @@ public function testRunTransactionNestedTransaction()
         });
     }
 
+    public function testRunTransactionShouldRetryOnRstStreamErrors()
+    {
+        $this->expectException('Google\Cloud\Core\Exception\ServerException');
+        $this->expectExceptionMessage('RST_STREAM');
+        $err = new ServerException('RST_STREAM', Code::INTERNAL);
+
+        $this->connection->beginTransaction(Argument::allOf(
+            Argument::withEntry('session', $this->session->name()),
+            Argument::withEntry(
+                'database',
+                DatabaseAdminClient::databaseName(
+                    self::PROJECT,
+                    self::INSTANCE,
+                    self::DATABASE
+                )
+            )
+        ))
+            ->shouldBeCalledTimes(3)
+            ->willThrow($err);
+
+        $this->database->runTransaction(function ($t) {
+            // do nothing
+        }, ['maxRetries' => 2]);
+    }
+
     public function testRunTransactionRetry()
     {
         $abort = new AbortedException('foo', 409, null, [