Skip to content

fix: address race condition #1500

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
aseering wants to merge 3 commits into
base: main
Choose a base branch
from
Open

fix: address race condition #1500

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
3c0df35
Address race condition
aseering Feb 24, 2026
a1476b3
Merge branch 'main' into bug/race-condition
olavloite Feb 25, 2026
2421472
test: do not reset request count for concurrent tests
olavloite Feb 25, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 4 additions & 0 deletions google/cloud/spanner_v1/snapshot.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -616,6 +616,10 @@ def _get_streamed_result_set(
if self._transaction_id is None:
is_inline_begin = True
self._lock.acquire()
if self._transaction_id is not None:
is_inline_begin = False
self._lock.release()
request.transaction = TransactionSelector(id=self._transaction_id)

iterator = _restart_on_unavailable(
method=method,
Expand Down
6 changes: 6 additions & 0 deletions google/cloud/spanner_v1/transaction.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -497,6 +497,9 @@ def execute_update(
if self._transaction_id is None:
is_inline_begin = True
self._lock.acquire()
if self._transaction_id is not None:
is_inline_begin = False
self._lock.release()

execute_sql_request = ExecuteSqlRequest(
session=session.name,
Expand Down Expand Up @@ -651,6 +654,9 @@ def batch_update(
if self._transaction_id is None:
is_inline_begin = True
self._lock.acquire()
if self._transaction_id is not None:
is_inline_begin = False
self._lock.release()

execute_batch_dml_request = ExecuteBatchDmlRequest(
session=session.name,
Expand Down
11 changes: 7 additions & 4 deletions tests/unit/test_spanner.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -333,6 +333,7 @@ def _read_helper(
count=0,
partition=None,
directed_read_options=None,
concurrent=False,
):
VALUES = [["bharney", 31], ["phred", 32]]
VALUE_PBS = [[_make_value_pb(item) for item in row] for row in VALUES]
Expand All @@ -359,7 +360,8 @@ def _read_helper(
result_sets[i].values.extend(VALUE_PBS[i])

api.streaming_read.return_value = _MockIterator(*result_sets)
transaction._read_request_count = count
if not concurrent:
transaction._read_request_count = count

if partition is not None: # 'limit' and 'partition' incompatible
result_set = transaction.read(
Expand All @@ -386,7 +388,8 @@ def _read_helper(
directed_read_options=directed_read_options,
)

self.assertEqual(transaction._read_request_count, count + 1)
if not concurrent:
self.assertEqual(transaction._read_request_count, count + 1)

self.assertEqual(list(result_set), VALUES)
self.assertEqual(result_set.metadata, metadata_pb)
Expand Down Expand Up @@ -1105,13 +1108,13 @@ def test_transaction_for_concurrent_statement_should_begin_one_transaction_with_
threads.append(
threading.Thread(
target=self._read_helper,
kwargs={"transaction": transaction, "api": api},
kwargs={"transaction": transaction, "api": api, "concurrent": True},
)
)
threads.append(
threading.Thread(
target=self._read_helper,
kwargs={"transaction": transaction, "api": api},
kwargs={"transaction": transaction, "api": api, "concurrent": True},
)
)
for thread in threads:
Expand Down
Loading