perf(storage): verify CRC32C of large read chunks off the event-loop thread

.librarian/generator-input/client-post-processing/storage-integration.yaml

@@ -590,7 +590,7 @@ replacements:
           # this dependency.
           "google-resumable-media >= 2.7.2, < 3.0.0",
           "requests >= 2.22.0, < 3.0.0",
-          "google-crc32c >= 1.6.0, < 2.0.0",
+          "google-crc32c >= 1.9.0, < 2.0.0",
       ]
       extras = {
           # TODO: Make these extra dependencies as mandatory once gRPC out of
@@ -659,7 +659,7 @@ replacements:
       grpc-google-iam-v1==0.14.2
       opentelemetry-api==1.1.0
       requests==2.22.0
-      google-crc32c==1.6.0
+      google-crc32c==1.9.0
     count: 1
   - paths: [
       packages/google-cloud-storage/testing/constraints-3.11.txt,

packages/google-cloud-storage/google/cloud/storage/asyncio/async_multi_range_downloader.py

@@ -40,6 +40,7 @@
     _BidiStreamRetryManager,
 )
 from google.cloud.storage.asyncio.retry.reads_resumption_strategy import (
+    _CRC32C_MAX_PENDING,
     _DownloadState,
     _ReadResumptionStrategy,
 )
@@ -90,7 +91,7 @@ class AsyncMultiRangeDownloader:
         mrd = await AsyncMultiRangeDownloader.create_mrd(
             client, bucket_name="chandrasiri-rs", object_name="test_open9"
         )
-        my_buff1 = open('my_fav_file.txt', 'wb')
+        my_buff1 = open("my_fav_file.txt", "wb")
         my_buff2 = BytesIO()
         my_buff3 = BytesIO()
         my_buff4 = any_object_which_provides_BytesIO_like_interface()
@@ -394,6 +395,13 @@ async def download_ranges(
 
         :type enable_checksum: bool
         :param enable_checksum: (Optional) If True, checksums are verified for downloaded data. Defaults to True.
+            Chunks of ``GCS_PYTHON_CLIENT_CRC32C_OFFLOAD_MIN_BYTES`` bytes or
+            more (default 1 MiB) are verified on a worker thread and written to
+            their buffer once the checksum matches; chunks still waiting are
+            discarded if the call is cancelled or a mismatch is found, and a
+            retry requests them again. Setting
+            ``GCS_PYTHON_CLIENT_CRC32C_OFFLOAD_MIN_BYTES=0`` verifies every
+            chunk inline.
 
         :raises ValueError: if the underlying bidi-GRPC stream is not open.
         :raises ValueError: if the length of read_ranges is more than 1000.
@@ -447,6 +455,7 @@ async def download_ranges(
 
         read_ids = set(download_states.keys())
         queue = self._multiplexer.register(read_ids)
+        strategy = _ReadResumptionStrategy()
 
         try:
             attempt_count = 0
@@ -522,20 +531,45 @@ async def generator():
                                         data_range.read_range.read_id
                                     )
                         yield item
+                        # The strategy has consumed `item` by now: write the
+                        # chunks whose checksums are done, and keep the
+                        # backlog bounded.
+                        await strategy.verify_pending_checksums(
+                            state, _CRC32C_MAX_PENDING
+                        )
+
+                    # Write (or reject) every held chunk before the attempt
+                    # ends, so a mismatch is raised inside the retry loop,
+                    # as an inline one is.
+                    await strategy.verify_pending_checksums(state)
 
                 return generator()
 
-            strategy = _ReadResumptionStrategy()
             retry_manager = _BidiStreamRetryManager(
                 strategy, send_and_recv_via_multiplexer
             )
 
             try:
-                await retry_manager.execute(initial_state, retry_policy)
+                try:
+                    await retry_manager.execute(initial_state, retry_policy)
+                except DataCorruption:
+                    raise
+                except Exception:
+                    # Chunks received before the stream error still have
+                    # in-flight checksum futures in initial_state. Verify them
+                    # before raising the stream error so an earlier corrupt
+                    # chunk raises DataCorruption (matching inline verification)
+                    # and valid chunks already received are written.
+                    await strategy.verify_pending_checksums(initial_state)
+                    raise
+                await strategy.verify_pending_checksums(initial_state)
             except DataCorruption:
                 if self.is_stream_open:
                     await self.close()
                 raise
+            finally:
+                # Drop chunks still held, e.g. after cancellation.
+                strategy.discard_pending_checksums(initial_state)
 
             if initial_state.get("read_handle"):
                 self.read_handle = initial_state["read_handle"]