From 2356e4074d61ce04eeb601838e863c7277580b15 Mon Sep 17 00:00:00 2001 From: Mattie Fu Date: Fri, 14 Aug 2026 21:50:48 +0000 Subject: [PATCH 1/2] feat(bigtable): fail V3 mutate_rows entries the server never acknowledged Change-Id: I9a166818c5829b446553fc1427d1f1d1ec078d0d --- .../bigtable/data/_async/_mutate_rows.py | 23 ++++++++ .../data/_sync_autogen/_mutate_rows.py | 13 +++++ .../unit/data/_async/test__mutate_rows.py | 54 +++++++++++++++++-- .../tests/unit/data/_async/test_client.py | 5 +- .../data/_sync_autogen/test__mutate_rows.py | 39 ++++++++++++-- .../unit/data/_sync_autogen/test_client.py | 2 +- 6 files changed, 127 insertions(+), 9 deletions(-) diff --git a/packages/google-cloud-bigtable/google/cloud/bigtable/data/_async/_mutate_rows.py b/packages/google-cloud-bigtable/google/cloud/bigtable/data/_async/_mutate_rows.py index 974e450d232b..6edee8f137b4 100644 --- a/packages/google-cloud-bigtable/google/cloud/bigtable/data/_async/_mutate_rows.py +++ b/packages/google-cloud-bigtable/google/cloud/bigtable/data/_async/_mutate_rows.py @@ -117,6 +117,9 @@ def __init__( self.mutations = [_EntryWithProto(m, m._to_pb()) for m in mutation_entries] self.remaining_indices = list(range(len(self.mutations))) self.errors: dict[int, list[Exception]] = {} + # track which original entries received a response entry from the server, + # so we can detect entries the server never acknowledged (see start()) + self._acknowledged_indices: set[int] = set() # set up metrics self._operation_metric = metric @@ -138,6 +141,24 @@ async def start(self): for idx in incomplete_indices: self._handle_entry_error(idx, exc) finally: + # response completeness check: the server must return one + # response entry for every request entry over the life of the + # operation. If fewer distinct entries were acknowledged than + # were sent, the stream closed without reporting an outcome for + # some mutations; we cannot assume they were applied, so fail + # them instead of silently treating them as successful. + if len(self._acknowledged_indices) != len(self.mutations): + for idx in range(len(self.mutations)): + if idx not in self._acknowledged_indices and ( + idx not in self.errors + ): + self.errors[idx] = [ + core_exceptions.ClientError( + "No response entry received for mutation " + "entry; the server acknowledged fewer " + "entries than were sent" + ) + ] # raise exception detailing incomplete mutations all_errors: list[Exception] = [] for idx, exc_list in self.errors.items(): @@ -194,6 +215,8 @@ async def _run_attempt(self): for result in result_list.entries: # convert sub-request index to global index orig_idx = active_request_indices[result.index] + # record that the server returned a response for this entry + self._acknowledged_indices.add(orig_idx) entry_error = core_exceptions.from_grpc_status( result.status.code, result.status.message, diff --git a/packages/google-cloud-bigtable/google/cloud/bigtable/data/_sync_autogen/_mutate_rows.py b/packages/google-cloud-bigtable/google/cloud/bigtable/data/_sync_autogen/_mutate_rows.py index 40e19dd85847..69a7da60ea52 100644 --- a/packages/google-cloud-bigtable/google/cloud/bigtable/data/_sync_autogen/_mutate_rows.py +++ b/packages/google-cloud-bigtable/google/cloud/bigtable/data/_sync_autogen/_mutate_rows.py @@ -97,6 +97,7 @@ def __init__( self.mutations = [_EntryWithProto(m, m._to_pb()) for m in mutation_entries] self.remaining_indices = list(range(len(self.mutations))) self.errors: dict[int, list[Exception]] = {} + self._acknowledged_indices: set[int] = set() self._operation_metric = metric def start(self): @@ -112,6 +113,17 @@ def start(self): for idx in incomplete_indices: self._handle_entry_error(idx, exc) finally: + if len(self._acknowledged_indices) != len(self.mutations): + for idx in range(len(self.mutations)): + if ( + idx not in self._acknowledged_indices + and idx not in self.errors + ): + self.errors[idx] = [ + core_exceptions.ClientError( + "No response entry received for mutation entry; the server acknowledged fewer entries than were sent" + ) + ] all_errors: list[Exception] = [] for idx, exc_list in self.errors.items(): if len(exc_list) == 0: @@ -159,6 +171,7 @@ def _run_attempt(self): for result_list in result_generator: for result in result_list.entries: orig_idx = active_request_indices[result.index] + self._acknowledged_indices.add(orig_idx) entry_error = core_exceptions.from_grpc_status( result.status.code, result.status.message, diff --git a/packages/google-cloud-bigtable/tests/unit/data/_async/test__mutate_rows.py b/packages/google-cloud-bigtable/tests/unit/data/_async/test__mutate_rows.py index 8ff6e42532b4..fc3d0dee4e06 100644 --- a/packages/google-cloud-bigtable/tests/unit/data/_async/test__mutate_rows.py +++ b/packages/google-cloud-bigtable/tests/unit/data/_async/test__mutate_rows.py @@ -53,8 +53,13 @@ def _make_mutation(self, count=1, size=1): return mutation @CrossSync.convert - async def _mock_stream(self, mutation_list, error_dict): + async def _mock_stream(self, mutation_list, error_dict, omit_indices=None): + omit_indices = omit_indices or set() for idx, entry in enumerate(mutation_list): + if idx in omit_indices: + # simulate a server that closes the stream without returning a + # response entry for this mutation + continue code = error_dict.get(idx, 0) yield MutateRowsResponse( entries=[ @@ -64,12 +69,12 @@ async def _mock_stream(self, mutation_list, error_dict): ] ) - def _make_mock_gapic(self, mutation_list, error_dict=None): + def _make_mock_gapic(self, mutation_list, error_dict=None, omit_indices=None): mock_fn = CrossSync.Mock() if error_dict is None: error_dict = {} mock_fn.side_effect = lambda *args, **kwargs: self._mock_stream( - mutation_list, error_dict + mutation_list, error_dict, omit_indices ) return mock_fn @@ -163,6 +168,9 @@ async def test_mutate_rows_operation(self): instance = self._make_one( client, table, entries, operation_timeout, operation_timeout, metric ) + # _run_attempt is mocked out, so simulate it acknowledging every + # entry to satisfy the response-completeness check in start() + instance._acknowledged_indices = set(range(len(entries))) await instance.start() assert attempt_mock.call_count == 1 @@ -265,9 +273,49 @@ async def test_mutate_rows_exception_retryable_eventually_pass(self, exc_type): metric, retryable_exceptions=(exc_type,), ) + # _run_attempt is mocked out, so simulate it acknowledging every + # entry to satisfy the response-completeness check in start() + instance._acknowledged_indices = set(range(len(entries))) await instance.start() assert attempt_mock.call_count == num_retries + 1 + @CrossSync.pytest + async def test_mutate_rows_unacknowledged_entries_fail(self): + """ + If the server closes the stream without returning a response entry for + every request entry, the unacknowledged entries must be surfaced as + failures instead of being silently treated as successful. + """ + from google.cloud.bigtable.data.exceptions import ( + FailedMutationEntryError, + MutationsExceptionGroup, + ) + + mutations = [ + self._make_mutation(), + self._make_mutation(), + self._make_mutation(), + ] + # server returns responses for indices 0 and 2, but omits index 1 + mock_gapic_fn = self._make_mock_gapic(mutations, omit_indices={1}) + instance = self._make_one( + mutation_entries=mutations, + ) + with mock.patch.object(instance, "_gapic_fn", mock_gapic_fn): + with pytest.raises(MutationsExceptionGroup) as exc_info: + await instance.start() + # only the omitted entry should be reported as failed + assert len(exc_info.value.exceptions) == 1 + failed = exc_info.value.exceptions[0] + assert isinstance(failed, FailedMutationEntryError) + assert failed.index == 1 + assert "fewer" in str(failed.__cause__) + # a single attempt is enough; omitted entries are not retried + assert mock_gapic_fn.call_count == 1 + # acknowledged entries should not be reported as failures + assert 0 not in instance.errors + assert 2 not in instance.errors + @CrossSync.pytest async def test_mutate_rows_incomplete_ignored(self): """ diff --git a/packages/google-cloud-bigtable/tests/unit/data/_async/test_client.py b/packages/google-cloud-bigtable/tests/unit/data/_async/test_client.py index 62ab9b5f96f0..b935b9d8d358 100644 --- a/packages/google-cloud-bigtable/tests/unit/data/_async/test_client.py +++ b/packages/google-cloud-bigtable/tests/unit/data/_async/test_client.py @@ -3349,9 +3349,10 @@ async def test_bulk_mutate_error_recovery(self): async with self._make_client(project="project") as client: table = client.get_table("instance", "table") with mock.patch.object(client._gapic_client, "mutate_rows") as mock_gapic: - # fail with a retryable error, then a non-retryable one + # first entry fails with a retryable error, the others succeed; + # the retry then resolves the first entry mock_gapic.side_effect = [ - self._mock_response([DeadlineExceeded("mock")]), + self._mock_response([DeadlineExceeded("mock"), None, None]), self._mock_response([None]), ] mutation = mutations.SetCell( diff --git a/packages/google-cloud-bigtable/tests/unit/data/_sync_autogen/test__mutate_rows.py b/packages/google-cloud-bigtable/tests/unit/data/_sync_autogen/test__mutate_rows.py index 2fe86a41fef0..05f709132270 100644 --- a/packages/google-cloud-bigtable/tests/unit/data/_sync_autogen/test__mutate_rows.py +++ b/packages/google-cloud-bigtable/tests/unit/data/_sync_autogen/test__mutate_rows.py @@ -52,8 +52,11 @@ def _make_mutation(self, count=1, size=1): mutation.size = lambda: size return mutation - def _mock_stream(self, mutation_list, error_dict): + def _mock_stream(self, mutation_list, error_dict, omit_indices=None): + omit_indices = omit_indices or set() for idx, entry in enumerate(mutation_list): + if idx in omit_indices: + continue code = error_dict.get(idx, 0) yield MutateRowsResponse( entries=[ @@ -63,12 +66,12 @@ def _mock_stream(self, mutation_list, error_dict): ] ) - def _make_mock_gapic(self, mutation_list, error_dict=None): + def _make_mock_gapic(self, mutation_list, error_dict=None, omit_indices=None): mock_fn = CrossSync._Sync_Impl.Mock() if error_dict is None: error_dict = {} mock_fn.side_effect = lambda *args, **kwargs: self._mock_stream( - mutation_list, error_dict + mutation_list, error_dict, omit_indices ) return mock_fn @@ -145,6 +148,7 @@ def test_mutate_rows_operation(self): instance = self._make_one( client, table, entries, operation_timeout, operation_timeout, metric ) + instance._acknowledged_indices = set(range(len(entries))) instance.start() assert attempt_mock.call_count == 1 @@ -230,9 +234,38 @@ def test_mutate_rows_exception_retryable_eventually_pass(self, exc_type): metric, retryable_exceptions=(exc_type,), ) + instance._acknowledged_indices = set(range(len(entries))) instance.start() assert attempt_mock.call_count == num_retries + 1 + def test_mutate_rows_unacknowledged_entries_fail(self): + """If the server closes the stream without returning a response entry for + every request entry, the unacknowledged entries must be surfaced as + failures instead of being silently treated as successful.""" + from google.cloud.bigtable.data.exceptions import ( + FailedMutationEntryError, + MutationsExceptionGroup, + ) + + mutations = [ + self._make_mutation(), + self._make_mutation(), + self._make_mutation(), + ] + mock_gapic_fn = self._make_mock_gapic(mutations, omit_indices={1}) + instance = self._make_one(mutation_entries=mutations) + with mock.patch.object(instance, "_gapic_fn", mock_gapic_fn): + with pytest.raises(MutationsExceptionGroup) as exc_info: + instance.start() + assert len(exc_info.value.exceptions) == 1 + failed = exc_info.value.exceptions[0] + assert isinstance(failed, FailedMutationEntryError) + assert failed.index == 1 + assert "fewer" in str(failed.__cause__) + assert mock_gapic_fn.call_count == 1 + assert 0 not in instance.errors + assert 2 not in instance.errors + def test_mutate_rows_incomplete_ignored(self): """MutateRowsIncomplete exceptions should not be added to error list""" from google.api_core.exceptions import DeadlineExceeded diff --git a/packages/google-cloud-bigtable/tests/unit/data/_sync_autogen/test_client.py b/packages/google-cloud-bigtable/tests/unit/data/_sync_autogen/test_client.py index eb954beb5113..ad1e7881afb2 100644 --- a/packages/google-cloud-bigtable/tests/unit/data/_sync_autogen/test_client.py +++ b/packages/google-cloud-bigtable/tests/unit/data/_sync_autogen/test_client.py @@ -2817,7 +2817,7 @@ def test_bulk_mutate_error_recovery(self): table = client.get_table("instance", "table") with mock.patch.object(client._gapic_client, "mutate_rows") as mock_gapic: mock_gapic.side_effect = [ - self._mock_response([DeadlineExceeded("mock")]), + self._mock_response([DeadlineExceeded("mock"), None, None]), self._mock_response([None]), ] mutation = mutations.SetCell( From 905ae22f03058eec9ce2a424929cb845411a2ff7 Mon Sep 17 00:00:00 2001 From: Mattie Fu Date: Mon, 17 Aug 2026 13:26:46 +0000 Subject: [PATCH 2/2] fix(bigtable): retry unacknowledged MutateRows entries per-entry instead of a count check Change-Id: I4ce3e38536a6aa6270806868074eb51c7832653a --- .../bigtable/data/_async/_mutate_rows.py | 35 +++---- .../data/_sync_autogen/_mutate_rows.py | 20 ++-- .../unit/data/_async/test__mutate_rows.py | 91 ++++++++++--------- .../tests/unit/data/_async/test_client.py | 23 ++++- .../data/_sync_autogen/test__mutate_rows.py | 70 ++++++++------ .../unit/data/_sync_autogen/test_client.py | 15 +++ 6 files changed, 141 insertions(+), 113 deletions(-) diff --git a/packages/google-cloud-bigtable/google/cloud/bigtable/data/_async/_mutate_rows.py b/packages/google-cloud-bigtable/google/cloud/bigtable/data/_async/_mutate_rows.py index 6edee8f137b4..c54f3c97bfb3 100644 --- a/packages/google-cloud-bigtable/google/cloud/bigtable/data/_async/_mutate_rows.py +++ b/packages/google-cloud-bigtable/google/cloud/bigtable/data/_async/_mutate_rows.py @@ -117,9 +117,6 @@ def __init__( self.mutations = [_EntryWithProto(m, m._to_pb()) for m in mutation_entries] self.remaining_indices = list(range(len(self.mutations))) self.errors: dict[int, list[Exception]] = {} - # track which original entries received a response entry from the server, - # so we can detect entries the server never acknowledged (see start()) - self._acknowledged_indices: set[int] = set() # set up metrics self._operation_metric = metric @@ -141,24 +138,6 @@ async def start(self): for idx in incomplete_indices: self._handle_entry_error(idx, exc) finally: - # response completeness check: the server must return one - # response entry for every request entry over the life of the - # operation. If fewer distinct entries were acknowledged than - # were sent, the stream closed without reporting an outcome for - # some mutations; we cannot assume they were applied, so fail - # them instead of silently treating them as successful. - if len(self._acknowledged_indices) != len(self.mutations): - for idx in range(len(self.mutations)): - if idx not in self._acknowledged_indices and ( - idx not in self.errors - ): - self.errors[idx] = [ - core_exceptions.ClientError( - "No response entry received for mutation " - "entry; the server acknowledged fewer " - "entries than were sent" - ) - ] # raise exception detailing incomplete mutations all_errors: list[Exception] = [] for idx, exc_list in self.errors.items(): @@ -215,8 +194,6 @@ async def _run_attempt(self): for result in result_list.entries: # convert sub-request index to global index orig_idx = active_request_indices[result.index] - # record that the server returned a response for this entry - self._acknowledged_indices.add(orig_idx) entry_error = core_exceptions.from_grpc_status( result.status.code, result.status.message, @@ -237,6 +214,18 @@ async def _run_attempt(self): self._handle_entry_error(idx, exc) # bubble up exception to be handled by retry wrapper raise + # Any entries that were sent but never received a response entry (a + # successfully-closed but incomplete stream) must not be treated as + # successful. Record a retryable error so idempotent entries are retried + # and non-idempotent entries surface as failures instead of being + # silently dropped. + for idx in active_request_indices.values(): + self._handle_entry_error( + idx, + bt_exceptions._MutateRowsIncomplete( + "no response entry received for mutation" + ), + ) # check if attempt succeeded, or needs to be retried if self.remaining_indices: # unfinished work; raise exception to trigger retry diff --git a/packages/google-cloud-bigtable/google/cloud/bigtable/data/_sync_autogen/_mutate_rows.py b/packages/google-cloud-bigtable/google/cloud/bigtable/data/_sync_autogen/_mutate_rows.py index 69a7da60ea52..ac2c1d9edac3 100644 --- a/packages/google-cloud-bigtable/google/cloud/bigtable/data/_sync_autogen/_mutate_rows.py +++ b/packages/google-cloud-bigtable/google/cloud/bigtable/data/_sync_autogen/_mutate_rows.py @@ -97,7 +97,6 @@ def __init__( self.mutations = [_EntryWithProto(m, m._to_pb()) for m in mutation_entries] self.remaining_indices = list(range(len(self.mutations))) self.errors: dict[int, list[Exception]] = {} - self._acknowledged_indices: set[int] = set() self._operation_metric = metric def start(self): @@ -113,17 +112,6 @@ def start(self): for idx in incomplete_indices: self._handle_entry_error(idx, exc) finally: - if len(self._acknowledged_indices) != len(self.mutations): - for idx in range(len(self.mutations)): - if ( - idx not in self._acknowledged_indices - and idx not in self.errors - ): - self.errors[idx] = [ - core_exceptions.ClientError( - "No response entry received for mutation entry; the server acknowledged fewer entries than were sent" - ) - ] all_errors: list[Exception] = [] for idx, exc_list in self.errors.items(): if len(exc_list) == 0: @@ -171,7 +159,6 @@ def _run_attempt(self): for result_list in result_generator: for result in result_list.entries: orig_idx = active_request_indices[result.index] - self._acknowledged_indices.add(orig_idx) entry_error = core_exceptions.from_grpc_status( result.status.code, result.status.message, @@ -186,6 +173,13 @@ def _run_attempt(self): for idx in active_request_indices.values(): self._handle_entry_error(idx, exc) raise + for idx in active_request_indices.values(): + self._handle_entry_error( + idx, + bt_exceptions._MutateRowsIncomplete( + "no response entry received for mutation" + ), + ) if self.remaining_indices: raise bt_exceptions._MutateRowsIncomplete diff --git a/packages/google-cloud-bigtable/tests/unit/data/_async/test__mutate_rows.py b/packages/google-cloud-bigtable/tests/unit/data/_async/test__mutate_rows.py index fc3d0dee4e06..5a17b2582e00 100644 --- a/packages/google-cloud-bigtable/tests/unit/data/_async/test__mutate_rows.py +++ b/packages/google-cloud-bigtable/tests/unit/data/_async/test__mutate_rows.py @@ -57,8 +57,8 @@ async def _mock_stream(self, mutation_list, error_dict, omit_indices=None): omit_indices = omit_indices or set() for idx, entry in enumerate(mutation_list): if idx in omit_indices: - # simulate a server that closes the stream without returning a - # response entry for this mutation + # simulate a server that closes the stream OK without returning + # a response entry for this mutation continue code = error_dict.get(idx, 0) yield MutateRowsResponse( @@ -168,9 +168,6 @@ async def test_mutate_rows_operation(self): instance = self._make_one( client, table, entries, operation_timeout, operation_timeout, metric ) - # _run_attempt is mocked out, so simulate it acknowledging every - # entry to satisfy the response-completeness check in start() - instance._acknowledged_indices = set(range(len(entries))) await instance.start() assert attempt_mock.call_count == 1 @@ -273,49 +270,9 @@ async def test_mutate_rows_exception_retryable_eventually_pass(self, exc_type): metric, retryable_exceptions=(exc_type,), ) - # _run_attempt is mocked out, so simulate it acknowledging every - # entry to satisfy the response-completeness check in start() - instance._acknowledged_indices = set(range(len(entries))) await instance.start() assert attempt_mock.call_count == num_retries + 1 - @CrossSync.pytest - async def test_mutate_rows_unacknowledged_entries_fail(self): - """ - If the server closes the stream without returning a response entry for - every request entry, the unacknowledged entries must be surfaced as - failures instead of being silently treated as successful. - """ - from google.cloud.bigtable.data.exceptions import ( - FailedMutationEntryError, - MutationsExceptionGroup, - ) - - mutations = [ - self._make_mutation(), - self._make_mutation(), - self._make_mutation(), - ] - # server returns responses for indices 0 and 2, but omits index 1 - mock_gapic_fn = self._make_mock_gapic(mutations, omit_indices={1}) - instance = self._make_one( - mutation_entries=mutations, - ) - with mock.patch.object(instance, "_gapic_fn", mock_gapic_fn): - with pytest.raises(MutationsExceptionGroup) as exc_info: - await instance.start() - # only the omitted entry should be reported as failed - assert len(exc_info.value.exceptions) == 1 - failed = exc_info.value.exceptions[0] - assert isinstance(failed, FailedMutationEntryError) - assert failed.index == 1 - assert "fewer" in str(failed.__cause__) - # a single attempt is enough; omitted entries are not retried - assert mock_gapic_fn.call_count == 1 - # acknowledged entries should not be reported as failures - assert 0 not in instance.errors - assert 2 not in instance.errors - @CrossSync.pytest async def test_mutate_rows_incomplete_ignored(self): """ @@ -422,3 +379,47 @@ async def test_run_attempt_partial_success_non_retryable(self): assert len(instance.errors[1]) == 1 assert instance.errors[1][0].grpc_status_code == 300 assert 2 not in instance.errors + + @CrossSync.pytest + async def test_run_attempt_missing_entry_retryable(self): + """If the server closes the stream successfully but omits a response + entry, the unanswered mutation must not be treated as successful. It + should be recorded as a retryable _MutateRowsIncomplete error so + idempotent entries are retried.""" + from google.cloud.bigtable.data.exceptions import _MutateRowsIncomplete + + mutations = [ + self._make_mutation(), + self._make_mutation(), + self._make_mutation(), + ] + # server omits the response entry for index 1 + mock_gapic_fn = self._make_mock_gapic(mutations, omit_indices={1}) + instance = self._make_one(mutation_entries=mutations) + instance.is_retryable = lambda x: True + with mock.patch.object(instance, "_gapic_fn", mock_gapic_fn): + with pytest.raises(_MutateRowsIncomplete): + await instance._run_attempt() + assert instance.remaining_indices == [1] + assert 0 not in instance.errors + assert len(instance.errors[1]) == 1 + assert isinstance(instance.errors[1][0], _MutateRowsIncomplete) + assert 2 not in instance.errors + + @CrossSync.pytest + async def test_run_attempt_missing_entry_non_retryable(self): + """A missing response entry for a non-retryable mutation is surfaced as + a failure rather than being silently dropped.""" + from google.cloud.bigtable.data.exceptions import _MutateRowsIncomplete + + mutations = [self._make_mutation(), self._make_mutation()] + # server omits the response entry for index 0 + mock_gapic_fn = self._make_mock_gapic(mutations, omit_indices={0}) + instance = self._make_one(mutation_entries=mutations) + instance.is_retryable = lambda x: False + with mock.patch.object(instance, "_gapic_fn", mock_gapic_fn): + await instance._run_attempt() + assert instance.remaining_indices == [] + assert len(instance.errors[0]) == 1 + assert isinstance(instance.errors[0][0], _MutateRowsIncomplete) + assert 1 not in instance.errors diff --git a/packages/google-cloud-bigtable/tests/unit/data/_async/test_client.py b/packages/google-cloud-bigtable/tests/unit/data/_async/test_client.py index b935b9d8d358..df081761665a 100644 --- a/packages/google-cloud-bigtable/tests/unit/data/_async/test_client.py +++ b/packages/google-cloud-bigtable/tests/unit/data/_async/test_client.py @@ -1505,6 +1505,25 @@ async def test_call_metadata(self, include_app_profile, fn_name, fn_args, gapic_ transport_mock = mock.MagicMock() rpc_mock = CrossSync.Mock() transport_mock._wrapped_methods.__getitem__.return_value = rpc_mock + if gapic_fn == "mutate_rows": + # An unacknowledged MutateRows entry is now treated as a retryable + # incomplete mutation, so an empty mock stream would retry until the + # operation timeout. Return a success entry for the single mutation + # so the operation completes after a single attempt. + from google.cloud.bigtable_v2.types import MutateRowsResponse + from google.rpc import status_pb2 + + @CrossSync.convert + async def mutate_rows_stream(*args, **kwargs): + yield MutateRowsResponse( + entries=[ + MutateRowsResponse.Entry( + index=0, status=status_pb2.Status(code=0) + ) + ] + ) + + rpc_mock.side_effect = mutate_rows_stream gapic_client = client._gapic_client if CrossSync.is_async: # inner BigtableClient is held as ._client for BigtableAsyncClient @@ -3349,8 +3368,8 @@ async def test_bulk_mutate_error_recovery(self): async with self._make_client(project="project") as client: table = client.get_table("instance", "table") with mock.patch.object(client._gapic_client, "mutate_rows") as mock_gapic: - # first entry fails with a retryable error, the others succeed; - # the retry then resolves the first entry + # first entry fails with a retryable error (other two succeed), + # then the retried entry succeeds mock_gapic.side_effect = [ self._mock_response([DeadlineExceeded("mock"), None, None]), self._mock_response([None]), diff --git a/packages/google-cloud-bigtable/tests/unit/data/_sync_autogen/test__mutate_rows.py b/packages/google-cloud-bigtable/tests/unit/data/_sync_autogen/test__mutate_rows.py index 05f709132270..e3979f6152fa 100644 --- a/packages/google-cloud-bigtable/tests/unit/data/_sync_autogen/test__mutate_rows.py +++ b/packages/google-cloud-bigtable/tests/unit/data/_sync_autogen/test__mutate_rows.py @@ -148,7 +148,6 @@ def test_mutate_rows_operation(self): instance = self._make_one( client, table, entries, operation_timeout, operation_timeout, metric ) - instance._acknowledged_indices = set(range(len(entries))) instance.start() assert attempt_mock.call_count == 1 @@ -234,38 +233,9 @@ def test_mutate_rows_exception_retryable_eventually_pass(self, exc_type): metric, retryable_exceptions=(exc_type,), ) - instance._acknowledged_indices = set(range(len(entries))) instance.start() assert attempt_mock.call_count == num_retries + 1 - def test_mutate_rows_unacknowledged_entries_fail(self): - """If the server closes the stream without returning a response entry for - every request entry, the unacknowledged entries must be surfaced as - failures instead of being silently treated as successful.""" - from google.cloud.bigtable.data.exceptions import ( - FailedMutationEntryError, - MutationsExceptionGroup, - ) - - mutations = [ - self._make_mutation(), - self._make_mutation(), - self._make_mutation(), - ] - mock_gapic_fn = self._make_mock_gapic(mutations, omit_indices={1}) - instance = self._make_one(mutation_entries=mutations) - with mock.patch.object(instance, "_gapic_fn", mock_gapic_fn): - with pytest.raises(MutationsExceptionGroup) as exc_info: - instance.start() - assert len(exc_info.value.exceptions) == 1 - failed = exc_info.value.exceptions[0] - assert isinstance(failed, FailedMutationEntryError) - assert failed.index == 1 - assert "fewer" in str(failed.__cause__) - assert mock_gapic_fn.call_count == 1 - assert 0 not in instance.errors - assert 2 not in instance.errors - def test_mutate_rows_incomplete_ignored(self): """MutateRowsIncomplete exceptions should not be added to error list""" from google.api_core.exceptions import DeadlineExceeded @@ -356,3 +326,43 @@ def test_run_attempt_partial_success_non_retryable(self): assert len(instance.errors[1]) == 1 assert instance.errors[1][0].grpc_status_code == 300 assert 2 not in instance.errors + + def test_run_attempt_missing_entry_retryable(self): + """If the server closes the stream successfully but omits a response + entry, the unanswered mutation must not be treated as successful. It + should be recorded as a retryable _MutateRowsIncomplete error so + idempotent entries are retried.""" + from google.cloud.bigtable.data.exceptions import _MutateRowsIncomplete + + mutations = [ + self._make_mutation(), + self._make_mutation(), + self._make_mutation(), + ] + mock_gapic_fn = self._make_mock_gapic(mutations, omit_indices={1}) + instance = self._make_one(mutation_entries=mutations) + instance.is_retryable = lambda x: True + with mock.patch.object(instance, "_gapic_fn", mock_gapic_fn): + with pytest.raises(_MutateRowsIncomplete): + instance._run_attempt() + assert instance.remaining_indices == [1] + assert 0 not in instance.errors + assert len(instance.errors[1]) == 1 + assert isinstance(instance.errors[1][0], _MutateRowsIncomplete) + assert 2 not in instance.errors + + def test_run_attempt_missing_entry_non_retryable(self): + """A missing response entry for a non-retryable mutation is surfaced as + a failure rather than being silently dropped.""" + from google.cloud.bigtable.data.exceptions import _MutateRowsIncomplete + + mutations = [self._make_mutation(), self._make_mutation()] + mock_gapic_fn = self._make_mock_gapic(mutations, omit_indices={0}) + instance = self._make_one(mutation_entries=mutations) + instance.is_retryable = lambda x: False + with mock.patch.object(instance, "_gapic_fn", mock_gapic_fn): + instance._run_attempt() + assert instance.remaining_indices == [] + assert len(instance.errors[0]) == 1 + assert isinstance(instance.errors[0][0], _MutateRowsIncomplete) + assert 1 not in instance.errors diff --git a/packages/google-cloud-bigtable/tests/unit/data/_sync_autogen/test_client.py b/packages/google-cloud-bigtable/tests/unit/data/_sync_autogen/test_client.py index ad1e7881afb2..0c3c3b49b7b3 100644 --- a/packages/google-cloud-bigtable/tests/unit/data/_sync_autogen/test_client.py +++ b/packages/google-cloud-bigtable/tests/unit/data/_sync_autogen/test_client.py @@ -1222,6 +1222,21 @@ def test_call_metadata(self, include_app_profile, fn_name, fn_args, gapic_fn): transport_mock = mock.MagicMock() rpc_mock = CrossSync._Sync_Impl.Mock() transport_mock._wrapped_methods.__getitem__.return_value = rpc_mock + if gapic_fn == "mutate_rows": + from google.rpc import status_pb2 + + from google.cloud.bigtable_v2.types import MutateRowsResponse + + def mutate_rows_stream(*args, **kwargs): + yield MutateRowsResponse( + entries=[ + MutateRowsResponse.Entry( + index=0, status=status_pb2.Status(code=0) + ) + ] + ) + + rpc_mock.side_effect = mutate_rows_stream gapic_client = client._gapic_client gapic_client._transport = transport_mock gapic_client._is_universe_domain_valid = True