fix(bigtable): data client should acknowledge all mutations in batch

[mutianf]

fail V3 mutate_rows entries the server never acknowledged

[gemini-code-assist]

Code Review

This pull request introduces a response completeness check to both the async and sync implementations of _mutate_rows.py, ensuring that any mutation entries not acknowledged by the server are explicitly failed with a ClientError rather than being silently treated as successful. While the added test coverage is thorough, the reviewer correctly identified a critical bug in both implementations: when mutations are retried across multiple attempts, obsolete acknowledgments and errors from previous attempts are not cleared, which prevents the completeness check from correctly identifying silently dropped entries in subsequent attempts. To resolve this, the state of retried entries should be cleared at the start of each attempt.

[gemini-code-assist]

Bug: Unacknowledged entries in retried attempts will not be correctly failed with ClientError

There is a subtle bug here when mutations are retried across multiple attempts:

1. Scenario:

   • A mutation entry at index i fails with a retryable error (e.g., UNAVAILABLE) in Attempt 1.
   • This adds i to self._acknowledged_indices and the retryable error to self.errors[i].
   • Since it's retryable, it is retried in Attempt 2.
   • In Attempt 2, the stream finishes successfully but silently drops/ignores index i (i.e., no response entry is returned for it, and no stream-level exception is raised).

2. The Issue:

   • In the finally block, len(self._acknowledged_indices) != len(self.mutations) is checked. Since i was acknowledged in Attempt 1, it is already in self._acknowledged_indices. If all other entries were acknowledged, this check will evaluate to False and the completeness check won't run.
   • Even if it runs, idx not in self._acknowledged_indices will be False and idx not in self.errors will be False (due to the error from Attempt 1).
   • Thus, the entry will not be marked with ClientError. Instead, the operation will fail with the obsolete retryable error from Attempt 1, which might cause the client to incorrectly retry the entire batch.

3. Recommended Solution:
   At the start of each attempt (e.g., at the beginning of _run_attempt), we should clear the state of the entries being retried so that we only consider acknowledgments and errors from the current attempt:

   self._acknowledged_indices.difference_update(self.remaining_indices)
   for idx in self.remaining_indices:
       self.errors.pop(idx, None)

[gemini-code-assist]

Bug: Unacknowledged entries in retried attempts will not be correctly failed with ClientError

This is the same issue as identified in the async version (_async/_mutate_rows.py). When mutations are retried across multiple attempts, obsolete acknowledgments and errors from previous attempts are not cleared, preventing the completeness check from correctly identifying and failing silently dropped entries in the final attempt.

Recommended Solution:
At the start of each attempt (e.g., at the beginning of _run_attempt), clear the state of the entries being retried:

self._acknowledged_indices.difference_update(self.remaining_indices)
for idx in self.remaining_indices:
    self.errors.pop(idx, None)