fix(spanner): release transaction lock if inline begin fails (#18409)

When execute_update() or batch_update() starts an inline begin
(transaction_id is None), it acquires self._lock while waiting for the
transaction ID from the server. If that request failed with an error,
the lock was never released.

This was a latent bug in _async/transaction.py that became an active
deadlock when async/sync parity was recently restored in snapshot.py:
reintroducing _wait_for_transaction_begin() meant subsequent queries in
the same transaction now try to acquire self._lock and hang.

Wrap the request execution in try...finally in _async/transaction.py and
regenerate sync transaction.py so self._lock is always released.

Also:
- Add unit tests for failed inline begin in both async and sync clients.
- Add a GitHub Actions workflow to run Spanner mock server tests in CI.
- Include the mockserver session in default local Nox runs.

Author: olavloite

.github/workflows/spanner-mockserver-tests.yml

@@ -0,0 +1,59 @@
+permissions:
+  contents: read
+
+concurrency:
+  group: ${{ github.workflow }}-${{ github.head_ref || github.run_id }}
+  cancel-in-progress: true
+
+on:
+  pull_request:
+    paths:
+      - 'packages/google-cloud-spanner/**'
+      - '.github/workflows/spanner-mockserver-tests.yml'
+  push:
+    branches:
+      - main
+    paths:
+      - 'packages/google-cloud-spanner/**'
+      - '.github/workflows/spanner-mockserver-tests.yml'
+
+defaults:
+  run:
+    working-directory: packages/google-cloud-spanner
+
+name: Run Spanner mockserver tests
+jobs:
+  check_changes:
+    runs-on: ubuntu-latest
+    outputs:
+      run_spanner: ${{ steps.filter.outputs.spanner }}
+    steps:
+      - uses: actions/checkout@df4cb1c069e1874edd31b4311f1884172cec0e10 # v6.0.3
+        with:
+          persist-credentials: false
+      - uses: dorny/paths-filter@fbd0ab8f3e69293af611ebaee6363fc25e6d187d # v4.0.1
+        id: filter
+        with:
+          filters: |
+            spanner:
+              - 'packages/google-cloud-spanner/**'
+              - '.github/workflows/spanner-mockserver-tests.yml'
+
+  mockserver-tests:
+    needs: check_changes
+    if: ${{ needs.check_changes.outputs.run_spanner == 'true' }}
+    runs-on: ubuntu-latest
+
+    steps:
+      - name: Checkout code
+        uses: actions/checkout@df4cb1c069e1874edd31b4311f1884172cec0e10 # v6.0.3
+        with:
+          persist-credentials: false
+      - name: Set up Python 3.12
+        uses: actions/setup-python@ece7cb06caefa5fff74198d8649806c4678c61a1 # v6
+        with:
+          python-version: "3.12"
+      - name: Install nox
+        run: python -m pip install nox
+      - name: Run nox
+        run: nox -s mockserver

packages/google-cloud-spanner/google/cloud/spanner_v1/_async/transaction.py

@@ -543,49 +543,50 @@ async def execute_update(
             is_inline_begin = True
             await self._lock.acquire()
 
-        execute_sql_request = ExecuteSqlRequest(
-            session=session.name,
-            transaction=self._build_transaction_selector_pb(),
-            sql=dml,
-            params=params_pb,
-            param_types=param_types,
-            query_mode=query_mode,
-            query_options=query_options,
-            seqno=seqno,
-            request_options=request_options,
-            last_statement=last_statement,
-        )
+        try:
+            execute_sql_request = ExecuteSqlRequest(
+                session=session.name,
+                transaction=self._build_transaction_selector_pb(),
+                sql=dml,
+                params=params_pb,
+                param_types=param_types,
+                query_mode=query_mode,
+                query_options=query_options,
+                seqno=seqno,
+                request_options=request_options,
+                last_statement=last_statement,
+            )
 
-        nth_request = database._next_nth_request
-        attempt = AtomicCounter(0)
+            nth_request = database._next_nth_request
+            attempt = AtomicCounter(0)
 
-        async def wrapped_method(*args, **kwargs):
-            attempt.increment()
-            call_metadata, error_augmenter = database.with_error_augmentation(
-                nth_request, attempt.value, metadata
-            )
-            execute_sql_method = functools.partial(
-                api.execute_sql,
-                request=execute_sql_request,
-                metadata=call_metadata,
-                retry=retry,
-                timeout=timeout,
-            )
-            with error_augmenter:
-                return await execute_sql_method(*args, **kwargs)
-
-        result_set_pb: ResultSet = await self._execute_request(
-            wrapped_method,
-            execute_sql_request,
-            metadata,
-            f"CloudSpanner.{type(self).__name__}.execute_update",
-            trace_attributes,
-        )
+            async def wrapped_method(*args, **kwargs):
+                attempt.increment()
+                call_metadata, error_augmenter = database.with_error_augmentation(
+                    nth_request, attempt.value, metadata
+                )
+                execute_sql_method = functools.partial(
+                    api.execute_sql,
+                    request=execute_sql_request,
+                    metadata=call_metadata,
+                    retry=retry,
+                    timeout=timeout,
+                )
+                with error_augmenter:
+                    return await execute_sql_method(*args, **kwargs)
 
-        self._update_for_result_set_pb(result_set_pb)
+            result_set_pb: ResultSet = await self._execute_request(
+                wrapped_method,
+                execute_sql_request,
+                metadata,
+                f"CloudSpanner.{type(self).__name__}.execute_update",
+                trace_attributes,
+            )
 
-        if is_inline_begin:
-            self._lock.release()
+            self._update_for_result_set_pb(result_set_pb)
+        finally:
+            if is_inline_begin:
+                self._lock.release()
 
         if result_set_pb._pb.HasField("precommit_token"):
             await self._update_for_precommit_token_pb(result_set_pb.precommit_token)
@@ -703,45 +704,46 @@ async def batch_update(
             is_inline_begin = True
             await self._lock.acquire()
 
-        execute_batch_dml_request = ExecuteBatchDmlRequest(
-            session=session.name,
-            transaction=self._build_transaction_selector_pb(),
-            statements=parsed,
-            seqno=seqno,
-            request_options=request_options,
-            last_statements=last_statement,
-        )
+        try:
+            execute_batch_dml_request = ExecuteBatchDmlRequest(
+                session=session.name,
+                transaction=self._build_transaction_selector_pb(),
+                statements=parsed,
+                seqno=seqno,
+                request_options=request_options,
+                last_statements=last_statement,
+            )
 
-        nth_request = database._next_nth_request
-        attempt = AtomicCounter(0)
+            nth_request = database._next_nth_request
+            attempt = AtomicCounter(0)
 
-        async def wrapped_method(*args, **kwargs):
-            attempt.increment()
-            call_metadata, error_augmenter = database.with_error_augmentation(
-                nth_request, attempt.value, metadata
-            )
-            execute_batch_dml_method = functools.partial(
-                api.execute_batch_dml,
-                request=execute_batch_dml_request,
-                metadata=call_metadata,
-                retry=retry,
-                timeout=timeout,
-            )
-            with error_augmenter:
-                return await execute_batch_dml_method(*args, **kwargs)
-
-        response_pb: ExecuteBatchDmlResponse = await self._execute_request(
-            wrapped_method,
-            execute_batch_dml_request,
-            metadata,
-            "CloudSpanner.DMLTransaction",
-            trace_attributes,
-        )
+            async def wrapped_method(*args, **kwargs):
+                attempt.increment()
+                call_metadata, error_augmenter = database.with_error_augmentation(
+                    nth_request, attempt.value, metadata
+                )
+                execute_batch_dml_method = functools.partial(
+                    api.execute_batch_dml,
+                    request=execute_batch_dml_request,
+                    metadata=call_metadata,
+                    retry=retry,
+                    timeout=timeout,
+                )
+                with error_augmenter:
+                    return await execute_batch_dml_method(*args, **kwargs)
 
-        self._update_for_execute_batch_dml_response_pb(response_pb)
+            response_pb: ExecuteBatchDmlResponse = await self._execute_request(
+                wrapped_method,
+                execute_batch_dml_request,
+                metadata,
+                "CloudSpanner.DMLTransaction",
+                trace_attributes,
+            )
 
-        if is_inline_begin:
-            self._lock.release()
+            self._update_for_execute_batch_dml_response_pb(response_pb)
+        finally:
+            if is_inline_begin:
+                self._lock.release()
 
         if (
             len(response_pb.result_sets) > 0

packages/google-cloud-spanner/google/cloud/spanner_v1/transaction.py

@@ -446,46 +446,48 @@ def execute_update(
         if self._transaction_id is None:
             is_inline_begin = True
             self._lock.acquire()
-        execute_sql_request = ExecuteSqlRequest(
-            session=session.name,
-            transaction=self._build_transaction_selector_pb(),
-            sql=dml,
-            params=params_pb,
-            param_types=param_types,
-            query_mode=query_mode,
-            query_options=query_options,
-            seqno=seqno,
-            request_options=request_options,
-            last_statement=last_statement,
-        )
-        nth_request = database._next_nth_request
-        attempt = AtomicCounter(0)
-
-        def wrapped_method(*args, **kwargs):
-            attempt.increment()
-            call_metadata, error_augmenter = database.with_error_augmentation(
-                nth_request, attempt.value, metadata
+        try:
+            execute_sql_request = ExecuteSqlRequest(
+                session=session.name,
+                transaction=self._build_transaction_selector_pb(),
+                sql=dml,
+                params=params_pb,
+                param_types=param_types,
+                query_mode=query_mode,
+                query_options=query_options,
+                seqno=seqno,
+                request_options=request_options,
+                last_statement=last_statement,
             )
-            execute_sql_method = functools.partial(
-                api.execute_sql,
-                request=execute_sql_request,
-                metadata=call_metadata,
-                retry=retry,
-                timeout=timeout,
+            nth_request = database._next_nth_request
+            attempt = AtomicCounter(0)
+
+            def wrapped_method(*args, **kwargs):
+                attempt.increment()
+                call_metadata, error_augmenter = database.with_error_augmentation(
+                    nth_request, attempt.value, metadata
+                )
+                execute_sql_method = functools.partial(
+                    api.execute_sql,
+                    request=execute_sql_request,
+                    metadata=call_metadata,
+                    retry=retry,
+                    timeout=timeout,
+                )
+                with error_augmenter:
+                    return execute_sql_method(*args, **kwargs)
+
+            result_set_pb: ResultSet = self._execute_request(
+                wrapped_method,
+                execute_sql_request,
+                metadata,
+                f"CloudSpanner.{type(self).__name__}.execute_update",
+                trace_attributes,
             )
-            with error_augmenter:
-                return execute_sql_method(*args, **kwargs)
-
-        result_set_pb: ResultSet = self._execute_request(
-            wrapped_method,
-            execute_sql_request,
-            metadata,
-            f"CloudSpanner.{type(self).__name__}.execute_update",
-            trace_attributes,
-        )
-        self._update_for_result_set_pb(result_set_pb)
-        if is_inline_begin:
-            self._lock.release()
+            self._update_for_result_set_pb(result_set_pb)
+        finally:
+            if is_inline_begin:
+                self._lock.release()
         if result_set_pb._pb.HasField("precommit_token"):
             self._update_for_precommit_token_pb(result_set_pb.precommit_token)
         return result_set_pb.stats.row_count_exact
@@ -586,42 +588,44 @@ def batch_update(
         if self._transaction_id is None:
             is_inline_begin = True
             self._lock.acquire()
-        execute_batch_dml_request = ExecuteBatchDmlRequest(
-            session=session.name,
-            transaction=self._build_transaction_selector_pb(),
-            statements=parsed,
-            seqno=seqno,
-            request_options=request_options,
-            last_statements=last_statement,
-        )
-        nth_request = database._next_nth_request
-        attempt = AtomicCounter(0)
-
-        def wrapped_method(*args, **kwargs):
-            attempt.increment()
-            call_metadata, error_augmenter = database.with_error_augmentation(
-                nth_request, attempt.value, metadata
+        try:
+            execute_batch_dml_request = ExecuteBatchDmlRequest(
+                session=session.name,
+                transaction=self._build_transaction_selector_pb(),
+                statements=parsed,
+                seqno=seqno,
+                request_options=request_options,
+                last_statements=last_statement,
             )
-            execute_batch_dml_method = functools.partial(
-                api.execute_batch_dml,
-                request=execute_batch_dml_request,
-                metadata=call_metadata,
-                retry=retry,
-                timeout=timeout,
+            nth_request = database._next_nth_request
+            attempt = AtomicCounter(0)
+
+            def wrapped_method(*args, **kwargs):
+                attempt.increment()
+                call_metadata, error_augmenter = database.with_error_augmentation(
+                    nth_request, attempt.value, metadata
+                )
+                execute_batch_dml_method = functools.partial(
+                    api.execute_batch_dml,
+                    request=execute_batch_dml_request,
+                    metadata=call_metadata,
+                    retry=retry,
+                    timeout=timeout,
+                )
+                with error_augmenter:
+                    return execute_batch_dml_method(*args, **kwargs)
+
+            response_pb: ExecuteBatchDmlResponse = self._execute_request(
+                wrapped_method,
+                execute_batch_dml_request,
+                metadata,
+                "CloudSpanner.DMLTransaction",
+                trace_attributes,
             )
-            with error_augmenter:
-                return execute_batch_dml_method(*args, **kwargs)
-
-        response_pb: ExecuteBatchDmlResponse = self._execute_request(
-            wrapped_method,
-            execute_batch_dml_request,
-            metadata,
-            "CloudSpanner.DMLTransaction",
-            trace_attributes,
-        )
-        self._update_for_execute_batch_dml_response_pb(response_pb)
-        if is_inline_begin:
-            self._lock.release()
+            self._update_for_execute_batch_dml_response_pb(response_pb)
+        finally:
+            if is_inline_begin:
+                self._lock.release()
         if (
             len(response_pb.result_sets) > 0
             and response_pb.result_sets[0].precommit_token