add error handling for open method

Author: Pulkit0110

google/cloud/storage/_experimental/asyncio/async_multi_range_downloader.py

@@ -102,11 +102,13 @@ async def open(self, metadata: Optional[List[Tuple[str, str]]] = None) -> None:
         if self._is_stream_open:
             raise ValueError("Stream is already open")
 
+        read_handle = self.read_handle if self.read_handle else None
+
         read_object_spec = _storage_v2.BidiReadObjectSpec(
             bucket=self._full_bucket_name,
             object=self.object_name,
             generation=self.generation_number if self.generation_number else None,
-            read_handle=self.read_handle if self.read_handle else None,
+            read_handle=read_handle,
         )
         self.first_bidi_read_req = _storage_v2.BidiReadObjectRequest(
             read_object_spec=read_object_spec

google/cloud/storage/_experimental/asyncio/async_read_object_stream.py

@@ -14,6 +14,9 @@
 
 from typing import Any, Dict, List, IO
 
+import grpc
+from google.api_core import exceptions
+from google.rpc import status_pb2
 from google_crc32c import Checksum
 from google.cloud import _storage_v2 as storage_v2
 from google.cloud.storage.exceptions import DataCorruption
@@ -23,6 +26,11 @@
 from google.cloud._storage_v2.types.storage import BidiReadObjectRedirectedError
 
 
+_BIDI_READ_REDIRECTED_TYPE_URL = (
+    "type.googleapis.com/google.storage.v2.BidiReadObjectRedirectedError"
+)
+
+
 class _DownloadState:
     """A helper class to track the state of a single range download."""
 
@@ -74,8 +82,8 @@ def update_state_from_response(
         """Processes a server response, performs integrity checks, and updates state."""
 
         # Capture read_handle if provided.
-        if response.read_handle and response.read_handle.handle:
-            state["read_handle"] = response.read_handle.handle
+        if response.read_handle:
+            state["read_handle"] = response.read_handle
 
         download_states = state["download_states"]
 
@@ -97,7 +105,7 @@ def update_state_from_response(
                 raise DataCorruption(
                     response,
                     f"Offset mismatch for read_id {read_id}. "
-                    f"Expected {read_state.next_expected_offset}, got {chunk_offset}"
+                    f"Expected {read_state.next_expected_offset}, got {chunk_offset}",
                 )
 
             # Checksum Verification
@@ -111,7 +119,7 @@ def update_state_from_response(
                     raise DataCorruption(
                         response,
                         f"Checksum mismatch for read_id {read_id}. "
-                        f"Server sent {server_checksum}, client calculated {client_checksum}."
+                        f"Server sent {server_checksum}, client calculated {client_checksum}.",
                     )
 
             # Update State & Write Data
@@ -130,16 +138,52 @@ def update_state_from_response(
                     raise DataCorruption(
                         response,
                         f"Byte count mismatch for read_id {read_id}. "
-                        f"Expected {read_state.initial_length}, got {read_state.bytes_written}"
+                        f"Expected {read_state.initial_length}, got {read_state.bytes_written}",
                     )
 
     async def recover_state_on_failure(self, error: Exception, state: Any) -> None:
         """Handles BidiReadObjectRedirectedError for reads."""
         # This would parse the gRPC error details, extract the routing_token,
         # and store it on the shared state object.
-        cause = getattr(error, "cause", error)
-        if isinstance(cause, BidiReadObjectRedirectedError):
-            state["routing_token"] = cause.routing_token
-            if cause.read_handle and cause.read_handle.handle:
-                state["read_handle"] = cause.read_handle.handle
-                print(f"Recover state: Updated read_handle from redirect: {state['read_handle']}")
+        grpc_error = None
+        if isinstance(error, exceptions.Aborted) and error.errors:
+            grpc_error = error.errors[0]
+
+        if grpc_error:
+            if isinstance(grpc_error, BidiReadObjectRedirectedError):
+                if grpc_error.routing_token:
+                    state["routing_token"] = grpc_error.routing_token
+                if grpc_error.read_handle:
+                    state["read_handle"] = grpc_error.read_handle
+                return
+
+            if hasattr(grpc_error, "trailing_metadata"):
+                trailers = grpc_error.trailing_metadata()
+                if not trailers:
+                    return
+                status_details_bin = None
+                for key, value in trailers:
+                    if key == "grpc-status-details-bin":
+                        status_details_bin = value
+                        break
+
+                if status_details_bin:
+                    status_proto = status_pb2.Status()
+                    try:
+                        status_proto.ParseFromString(status_details_bin)
+                        for detail in status_proto.details:
+                            if detail.type_url == _BIDI_READ_REDIRECTED_TYPE_URL:
+                                redirect_proto = (
+                                    BidiReadObjectRedirectedError.deserialize(
+                                        detail.value
+                                    )
+                                )
+                                if redirect_proto.routing_token:
+                                    state[
+                                        "routing_token"
+                                    ] = redirect_proto.routing_token
+                                if redirect_proto.read_handle:
+                                    state["read_handle"] = redirect_proto.read_handle
+                                break
+                    except Exception as e:
+                        print(f"--- Error unpacking redirect in _on_open_error: {e}")

google/cloud/storage/_experimental/asyncio/retry/reads_resumption_strategy.py

@@ -774,6 +774,5 @@ def flush(self):
         def has_unconsumed_tail(self) -> bool:
             return self._decoder.has_unconsumed_tail
 
-
 else:  # pragma: NO COVER
     _BrotliDecoder = None  # type: ignore # pragma: NO COVER