fix(bigtable): pack unpacked error details in _get_status and harden batcher error handling

packages/google-cloud-bigtable/google/cloud/bigtable/batcher.py

@@ -20,7 +20,10 @@
 
 from google.api_core import exceptions as core_exceptions
 
-from google.cloud.bigtable.data.exceptions import MutationsExceptionGroup
+from google.cloud.bigtable.data.exceptions import (
+    MutationsExceptionGroup,
+    RetryExceptionGroup,
+)
 from google.cloud.bigtable.data.mutations import RowMutationEntry
 
 FLUSH_COUNT = 100  # after this many elements, send out the batch
@@ -137,8 +140,12 @@ def _close_batcher(self):
                 # return that error to the user. In standard execution paths,
                 # FailedMutationEntryError always has an Exception cause;
                 # defensively wrap in Unknown if __cause__ is None.
-                if error.__cause__ is not None:
-                    self._exceptions.put(error.__cause__)
+                cause = error.__cause__
+                if isinstance(cause, RetryExceptionGroup) and cause.exceptions:
+                    # Use the terminal exception from the retry chain.
+                    cause = cause.exceptions[-1]
+                if cause is not None:
+                    self._exceptions.put(cause)
                 else:
                     unknown_err = core_exceptions.Unknown(str(error))
                     unknown_err.__cause__ = error

packages/google-cloud-bigtable/google/cloud/bigtable/data/_async/mutations_batcher.py

@@ -460,22 +460,23 @@ async def _execute_mutate_rows(
             # This is an internal callback used by the legacy synchronous shim,
             # though async callbacks are awaited when running in async mode.
             if self._user_batch_completed_callback:
+                result = None
                 try:
                     result = self._user_batch_completed_callback(statuses)
                     if CrossSync.is_async:
                         if inspect.isawaitable(result):
                             await result
-                    else:
-                        if inspect.isawaitable(result):
-                            if inspect.iscoroutine(result):
-                                result.close()
-                            raise TypeError(
-                                "_user_batch_completed_callback must be a synchronous callable"
-                            )
                 except Exception as exc:
                     _LOGGER.warning(
                         f"Exception raised in user batch completion callback: {exc}"
                     )
+                if not CrossSync.is_async:
+                    if inspect.isawaitable(result):
+                        if inspect.iscoroutine(result):
+                            result.close()
+                        raise TypeError(
+                            "_user_batch_completed_callback must be a synchronous callable"
+                        )
         return []
 
     def _add_exceptions(self, excs: list[FailedMutationEntryError]):
@@ -605,11 +606,11 @@ async def _wait_for_batch_results(
             return []
         exceptions: list[FailedMutationEntryError] = []
         for task, batch in tasks:
-            if CrossSync.is_async:
-                # futures don't need to be awaited in sync mode
-                await task
             try:
-                exc_list = task.result()
+                if CrossSync.is_async:
+                    exc_list = await task
+                else:
+                    exc_list = task.result()
                 if exc_list:
                     # expect a list of FailedMutationEntryError objects
                     for exc in exc_list:

packages/google-cloud-bigtable/google/cloud/bigtable/data/_helpers.py

@@ -33,6 +33,7 @@
 from google.api_core import exceptions as core_exceptions
 from google.api_core import retry as retries
 from google.api_core.retry import RetryFailureReason, exponential_sleep_generator
+from google.protobuf import any_pb2, message
 from google.rpc import code_pb2, status_pb2
 
 from google.cloud.bigtable.data.exceptions import (
@@ -287,22 +288,61 @@ def _get_status(exc: Optional[Exception]) -> status_pb2.Status:
     Returns:
         status_pb2.Status: A Status proto object.
     """
-    if (
-        isinstance(exc, core_exceptions.GoogleAPICallError)
-        and exc.grpc_status_code is not None
-    ):
-        return status_pb2.Status(  # type: ignore[unreachable]
-            code=exc.grpc_status_code.value[0],
-            message=exc.message,
-            details=exc.details,
-        )
+    if isinstance(exc, core_exceptions.GoogleAPICallError):
+        grpc_code = exc.grpc_status_code
+        if grpc_code is not None:
+            try:
+                details: list[any_pb2.Any] = []
+                for detail in exc.details or ():
+                    try:
+                        if isinstance(detail, any_pb2.Any):
+                            details.append(detail)
+                        elif isinstance(detail, message.Message):
+                            packed = any_pb2.Any()
+                            packed.Pack(detail)
+                            details.append(packed)
+                        elif isinstance(getattr(detail, "_pb", None), message.Message):
+                            packed = any_pb2.Any()
+                            packed.Pack(detail._pb)
+                            details.append(packed)
+                    except Exception:
+                        continue
+                return status_pb2.Status(
+                    code=grpc_code.value[0],
+                    message=exc.message or "",
+                    details=details,
+                )
+            except Exception:
+                pass
 
     return status_pb2.Status(
         code=code_pb2.UNKNOWN,
-        message=str(exc) if exc else "An unknown error has occurred",
+        message=_exc_message(exc),
     )
 
 
+def _exc_message(exc: Optional[Exception]) -> str:
+    """
+    Helper function that renders an exception as a string, without raising.
+
+    ``GoogleAPICallError.__str__`` reads ``exc.details``, which raises if the
+    exception was built with a malformed ``details`` value, so fall back to the
+    raw message when that happens.
+
+    Args:
+        exc: The exception to render.
+    Returns:
+        str: A description of the exception.
+    """
+    if exc is None:
+        return "An unknown error has occurred"
+    try:
+        return str(exc)
+    except Exception:
+        raw_message = getattr(exc, "message", None)
+        return raw_message if isinstance(raw_message, str) else type(exc).__name__
+
+
 def _validate_timeouts(
     operation_timeout: float, attempt_timeout: float | None, allow_none: bool = False
 ):
@@ -337,11 +377,18 @@ def _get_error_type(
 
     Args:
       - call_code: Exception type or gRPC status code.
+    Raises:
+      - ValueError: if call_code is not a valid retryable gRPC error status code
+        or exception type.
     """
     if isinstance(call_code, type):
         return call_code
-    else:
-        return type(core_exceptions.from_grpc_status(call_code, ""))
+    if isinstance(call_code, bool):
+        raise ValueError(f"Invalid gRPC error status code: {call_code!r}")
+    error_type = type(core_exceptions.from_grpc_status(call_code, ""))
+    if error_type is core_exceptions.GoogleAPICallError:
+        raise ValueError(f"Invalid gRPC error status code: {call_code!r}")
+    return error_type
 
 
 def _get_retryable_errors(

packages/google-cloud-bigtable/google/cloud/bigtable/data/_sync_autogen/mutations_batcher.py

@@ -393,18 +393,19 @@ def _execute_mutate_rows(
         finally:
             self._flow_control.remove_from_flow(batch)
             if self._user_batch_completed_callback:
+                result = None
                 try:
                     result = self._user_batch_completed_callback(statuses)
-                    if inspect.isawaitable(result):
-                        if inspect.iscoroutine(result):
-                            result.close()
-                        raise TypeError(
-                            "_user_batch_completed_callback must be a synchronous callable"
-                        )
                 except Exception as exc:
                     _LOGGER.warning(
                         f"Exception raised in user batch completion callback: {exc}"
                     )
+                if inspect.isawaitable(result):
+                    if inspect.iscoroutine(result):
+                        result.close()
+                    raise TypeError(
+                        "_user_batch_completed_callback must be a synchronous callable"
+                    )
         return []
 
     def _add_exceptions(self, excs: list[FailedMutationEntryError]):

packages/google-cloud-bigtable/tests/unit/data/_async/test_mutations_batcher.py

@@ -1099,13 +1099,67 @@ async def test__execute_mutate_rows_batch_completed_callback_coroutine(self):
             async with self._make_one(table) as instance:
                 instance._user_batch_completed_callback = async_callback
                 batch = [self._make_mutation()]
-                result = await instance._execute_mutate_rows(batch, mock.Mock())
-                assert result == []
                 if CrossSync.is_async:
+                    result = await instance._execute_mutate_rows(batch, mock.Mock())
+                    assert result == []
                     assert called_with == [[status_pb2.Status(code=code_pb2.OK)]]
                 else:
+                    with pytest.raises(
+                        TypeError,
+                        match="_user_batch_completed_callback must be a synchronous callable",
+                    ):
+                        instance._execute_mutate_rows(batch, mock.Mock())
                     assert called_with == []
 
+    @CrossSync.pytest
+    async def test__execute_mutate_rows_batch_completed_callback_unexpected_error(self):
+        from google.rpc import code_pb2, status_pb2
+
+        with mock.patch.object(CrossSync._MutateRowsOperation, "start") as mutate_rows:
+            mutate_rows.side_effect = RuntimeError("unexpected error")
+            table = mock.Mock()
+            table.default_mutate_rows_operation_timeout = 17
+            table.default_mutate_rows_attempt_timeout = 13
+            table.default_mutate_rows_retryable_errors = ()
+            callback = mock.Mock()
+            async with self._make_one(table) as instance:
+                instance._user_batch_completed_callback = callback
+                batch = [self._make_mutation(), self._make_mutation()]
+                with pytest.raises(RuntimeError, match="unexpected error"):
+                    await instance._execute_mutate_rows(batch, mock.Mock())
+                callback.assert_called_once_with(
+                    [
+                        status_pb2.Status(code=code_pb2.UNKNOWN),
+                        status_pb2.Status(code=code_pb2.UNKNOWN),
+                    ]
+                )
+
+    @CrossSync.pytest
+    async def test__wait_for_batch_results_unexpected_task_exception(self):
+        """Unexpected exceptions raised by batch tasks should be wrapped in FailedMutationEntryError."""
+        batch = [self._make_mutation(), self._make_mutation()]
+        expected_err = RuntimeError("task failed")
+
+        async with self._make_one() as instance:
+            with mock.patch.object(
+                instance, "_execute_mutate_rows", side_effect=expected_err
+            ):
+                with mock.patch.object(
+                    instance._flow_control, "add_to_flow"
+                ) as flow_mock:
+
+                    async def gen(x):
+                        yield x
+
+                    flow_mock.side_effect = lambda x: gen(x)
+                    await instance._flush_internal(batch)
+                    assert len(instance._oldest_exceptions) == 2
+                    for failed_entry in instance._oldest_exceptions:
+                        assert failed_entry.index is None
+                        assert failed_entry.__cause__ is expected_err
+                    instance._oldest_exceptions.clear()
+                    instance._newest_exceptions.clear()
+
     @CrossSync.pytest
     async def test__raise_exceptions(self):
         """Raise exceptions and reset error state"""

packages/google-cloud-bigtable/tests/unit/data/_sync_autogen/test_mutations_batcher.py

@@ -979,10 +979,61 @@ def test__execute_mutate_rows_batch_completed_callback_coroutine(self):
             with self._make_one(table) as instance:
                 instance._user_batch_completed_callback = async_callback
                 batch = [self._make_mutation()]
-                result = instance._execute_mutate_rows(batch, mock.Mock())
-                assert result == []
+                with pytest.raises(
+                    TypeError,
+                    match="_user_batch_completed_callback must be a synchronous callable",
+                ):
+                    instance._execute_mutate_rows(batch, mock.Mock())
                 assert called_with == []
 
+    def test__execute_mutate_rows_batch_completed_callback_unexpected_error(self):
+        from google.rpc import code_pb2, status_pb2
+
+        with mock.patch.object(
+            CrossSync._Sync_Impl._MutateRowsOperation, "start"
+        ) as mutate_rows:
+            mutate_rows.side_effect = RuntimeError("unexpected error")
+            table = mock.Mock()
+            table.default_mutate_rows_operation_timeout = 17
+            table.default_mutate_rows_attempt_timeout = 13
+            table.default_mutate_rows_retryable_errors = ()
+            callback = mock.Mock()
+            with self._make_one(table) as instance:
+                instance._user_batch_completed_callback = callback
+                batch = [self._make_mutation(), self._make_mutation()]
+                with pytest.raises(RuntimeError, match="unexpected error"):
+                    instance._execute_mutate_rows(batch, mock.Mock())
+                callback.assert_called_once_with(
+                    [
+                        status_pb2.Status(code=code_pb2.UNKNOWN),
+                        status_pb2.Status(code=code_pb2.UNKNOWN),
+                    ]
+                )
+
+    def test__wait_for_batch_results_unexpected_task_exception(self):
+        """Unexpected exceptions raised by batch tasks should be wrapped in FailedMutationEntryError."""
+        batch = [self._make_mutation(), self._make_mutation()]
+        expected_err = RuntimeError("task failed")
+        with self._make_one() as instance:
+            with mock.patch.object(
+                instance, "_execute_mutate_rows", side_effect=expected_err
+            ):
+                with mock.patch.object(
+                    instance._flow_control, "add_to_flow"
+                ) as flow_mock:
+
+                    def gen(x):
+                        yield x
+
+                    flow_mock.side_effect = lambda x: gen(x)
+                    instance._flush_internal(batch)
+                    assert len(instance._oldest_exceptions) == 2
+                    for failed_entry in instance._oldest_exceptions:
+                        assert failed_entry.index is None
+                        assert failed_entry.__cause__ is expected_err
+                    instance._oldest_exceptions.clear()
+                    instance._newest_exceptions.clear()
+
     def test__raise_exceptions(self):
         """Raise exceptions and reset error state"""
         from google.cloud.bigtable.data import exceptions