fix(bigtable): trip accelerator breaker and fall back on InvalidChunk

packages/google-cloud-bigtable/google/cloud/bigtable/data/_accelerator/_fallback.py

@@ -43,6 +43,8 @@
 from google.api_core import exceptions as core_exceptions
 from grpc import RpcError, StatusCode
 
+from google.cloud.bigtable.data.exceptions import InvalidChunk
+
 if TYPE_CHECKING:
     from google.cloud.bigtable.data._accelerator._daemon import AcceleratorDaemon
 
@@ -148,6 +150,7 @@ def handle_accelerator_error(
     exception for the caller to propagate:
 
     * daemon subprocess dead -> trip the breaker, fall back (it will not recover)
+    * :class:`InvalidChunk` -> trip the breaker, fall back immediately
     * ``UNIMPLEMENTED`` -> trip the breaker, fall back immediately
     * any other gRPC error -> translate and raise
     * a non-gRPC exception -> re-raise unchanged (never masked as a fallback)
@@ -166,6 +169,14 @@ def handle_accelerator_error(
         )
         breaker.trip()
         raise _AcceleratorFallback() from exc
+    if isinstance(exc, InvalidChunk):
+        _LOGGER.warning(
+            "Accelerator daemon returned an invalid ReadRows chunk; permanently "
+            "falling back to the native Bigtable client for this table.",
+            exc_info=exc,
+        )
+        breaker.trip()
+        raise _AcceleratorFallback() from exc
     if not isinstance(exc, RpcError):
         # A bug in our own merge machinery, not a daemon result. Do not mask it
         # as a fallback; let it propagate unchanged.

packages/google-cloud-bigtable/tests/unit/data/test_accelerator_dispatch.py

@@ -312,6 +312,39 @@ def test_non_grpc_exception_reraises(self):
             self._run(t._read_row_via_accelerator(query, 30.0, 20.0))
         assert not t._accelerator_breaker.bypass()
 
+    def test_invalid_chunk_trips_breaker_and_raises_fallback(self):
+        from google.protobuf import wrappers_pb2
+
+        from google.cloud.bigtable.data.read_rows_query import ReadRowsQuery
+        from google.cloud.bigtable_v2.types.bigtable import ReadRowsResponse
+
+        async def _malformed_stream():
+            yield ReadRowsResponse(
+                chunks=[
+                    ReadRowsResponse.CellChunk(
+                        row_key=b"",
+                        family_name=wrappers_pb2.StringValue(value="cf1"),
+                        qualifier=wrappers_pb2.BytesValue(value=b"q1"),
+                        timestamp_micros=1000,
+                        value=b"v1",
+                        commit_row=True,
+                    )
+                ]
+            )
+
+        async def _awaitable_stream():
+            return _malformed_stream()
+
+        client_mock = mock.AsyncMock()
+        client_mock.read_rows.return_value = _awaitable_stream()
+        daemon = mock.Mock(is_running=True)
+        t = _bare_target(accel_client=client_mock, accel_daemon=daemon)
+
+        query = ReadRowsQuery(row_keys=b"", limit=1)
+        with pytest.raises(_AcceleratorFallback):
+            self._run(t._read_row_via_accelerator(query, 30.0, 20.0))
+        assert t._accelerator_breaker.bypass()
+
 
 # ---------------------------------------------------------------------------
 # read_row: accelerator dispatch and fallback to native

packages/google-cloud-bigtable/tests/unit/data/test_accelerator_fallback.py

@@ -27,6 +27,7 @@
     _grpc_code,
     handle_accelerator_error,
 )
+from google.cloud.bigtable.data.exceptions import InvalidChunk
 
 # ---------------------------------------------------------------------------
 # AcceleratorBreaker
@@ -216,6 +217,18 @@ def test_internal_error_translated_and_raised(self):
             handle_accelerator_error(exc, daemon=daemon, breaker=breaker)
         assert not breaker.bypass()
 
+    # -- InvalidChunk --------------------------------------------------------
+
+    def test_invalid_chunk_trips_breaker_and_falls_back(self):
+        daemon = mock.Mock()
+        daemon.is_running = True
+        breaker = AcceleratorBreaker()
+
+        exc = InvalidChunk("first row chunk is missing key")
+        with pytest.raises(_AcceleratorFallback):
+            handle_accelerator_error(exc, daemon=daemon, breaker=breaker)
+        assert breaker.bypass()
+
     # -- Non-gRPC exceptions are re-raised unchanged -------------------------
 
     def test_non_grpc_exception_reraises_unchanged(self):