From 2356e4074d61ce04eeb601838e863c7277580b15 Mon Sep 17 00:00:00 2001 From: Mattie Fu Date: Fri, 14 Aug 2026 21:50:48 +0000 Subject: [PATCH] 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(