feat(storage): add option to disable checksums and improve robustness of full_object_checksum validation (#17665)

This PR introduces performance options and robustness improvements to
the asynchronous append operations in `AsyncAppendableObjectWriter`.

### Key Changes:
1. **Disable Checksums for Performance**:
- Added the `enable_checksum` optional parameter (default: `True`) to
`AsyncAppendableObjectWriter.append()`.
- When set to `False`, the writer skips calculation of chunk-level
CRC32C checksums in `_WriteResumptionStrategy`, improving append
throughput.
- Updated unit tests to verify that chunk-level checksums are omitted
when disabled.

2. **Robustness of Finalization Checksum Validation**:
- Implemented type checking (`isinstance(int)`) and range checking (`[0,
2**32-1]`) for `full_object_checksum` in `finalize()`.
- Wrapped stream receiving and response processing inside a
`try...finally` block. This guarantees that the underlying stream is
closed and local writer state variables are reset on mismatch
exceptions/errors, preventing stream leaks.
- Simplified the finalization request building code into a clean
`if-elif-else` format.
- Added corresponding unit tests verifying validation error exceptions
and proper stream cleanup.

Author: chandra-siri

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

@@ -386,6 +386,7 @@ async def append(
         data: bytes,
         retry_policy: Optional[AsyncRetry] = None,
         metadata: Optional[List[Tuple[str, str]]] = None,
+        enable_checksum: bool = True,
     ) -> None:
         """Appends data to the Appendable object with automatic retries.
 
@@ -406,6 +407,9 @@ async def append(
         :type metadata: List[Tuple[str, str]]
         :param metadata: (Optional) The metadata to be sent with the request.
 
+        :type enable_checksum: bool
+        :param enable_checksum: (Optional) If True, calculates and checks checksums for each chunk. Defaults to True.
+
         :raises ValueError: If the stream is not open.
         """
         if not self._is_stream_open:
@@ -487,7 +491,12 @@ async def generator():
             return generator()
 
         # State initialization
-        write_state = _WriteState(_MAX_CHUNK_SIZE_BYTES, buffer, self.flush_interval)
+        write_state = _WriteState(
+            _MAX_CHUNK_SIZE_BYTES,
+            buffer,
+            self.flush_interval,
+            enable_checksum=enable_checksum,
+        )
         write_state.write_handle = self.write_handle
         write_state.persisted_size = self.persisted_size
         # offset is set during `open()` call.
@@ -641,22 +650,32 @@ async def finalize(
         if not self._is_stream_open:
             raise ValueError("Stream is not open. Call open() before finalize().")
 
-        finalize_req = _storage_v2.BidiWriteObjectRequest(finish_write=True)
-
-        if full_object_checksum is not None:
-            finalize_req.object_checksums = _storage_v2.ObjectChecksums(
-                crc32c=full_object_checksum
+        if full_object_checksum is None:
+            finalize_req = _storage_v2.BidiWriteObjectRequest(finish_write=True)
+        elif isinstance(full_object_checksum, bool) or not isinstance(
+            full_object_checksum, int
+        ):
+            raise TypeError("full_object_checksum must be an integer.")
+        elif not (0 <= full_object_checksum <= 0xFFFFFFFF):
+            raise ValueError("full_object_checksum must be a 32-bit unsigned integer.")
+        else:
+            finalize_req = _storage_v2.BidiWriteObjectRequest(
+                finish_write=True,
+                object_checksums=_storage_v2.ObjectChecksums(
+                    crc32c=full_object_checksum
+                ),
             )
 
-        await self.write_obj_stream.send(finalize_req)
-        response = await self.write_obj_stream.recv()
-        self.object_resource = response.resource
-        self.persisted_size = self.object_resource.size
-        await self.write_obj_stream.close()
-
-        self._is_stream_open = False
-        self.offset = None
-        return self.object_resource
+        try:
+            await self.write_obj_stream.send(finalize_req)
+            response = await self.write_obj_stream.recv()
+            self.object_resource = response.resource
+            self.persisted_size = self.object_resource.size
+            return self.object_resource
+        finally:
+            await self.write_obj_stream.close()
+            self._is_stream_open = False
+            self.offset = None
 
     @property
     def is_stream_open(self) -> bool:

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

@@ -407,6 +407,12 @@ async def download_ranges(
         :type retry_policy: :class:`~google.api_core.retry_async.AsyncRetry`
         :param retry_policy: (Optional) The retry policy to use for the operation.
 
+        :type metadata: List[Tuple[str, str]]
+        :param metadata: (Optional) The metadata to be sent with the request.
+
+        :type enable_checksum: bool
+        :param enable_checksum: (Optional) If True, checksums are verified for downloaded data. Defaults to True.
+
         :raises ValueError: if the underlying bidi-GRPC stream is not open.
         :raises ValueError: if the length of read_ranges is more than 1000.
         :raises DataCorruption: if a checksum mismatch is detected while reading data.

packages/google-cloud-storage/google/cloud/storage/asyncio/retry/writes_resumption_strategy.py

@@ -44,6 +44,7 @@ def __init__(
         chunk_size: int,
         user_buffer: IO[bytes],
         flush_interval: int,
+        enable_checksum: bool = True,
     ):
         self.chunk_size = chunk_size
         self.user_buffer = user_buffer
@@ -59,6 +60,7 @@ def __init__(
         self.write_handle: Union[bytes, storage_type.BidiWriteHandle, None] = None
         self.routing_token: Optional[str] = None
         self.is_finalized: bool = False
+        self.enable_checksum: bool = enable_checksum
 
 
 class _WriteResumptionStrategy(_BaseResumptionStrategy):
@@ -84,7 +86,8 @@ def generate_requests(
                 break
 
             checksummed_data = storage_type.ChecksummedData(content=chunk)
-            checksummed_data.crc32c = google_crc32c.value(chunk)
+            if write_state.enable_checksum:
+                checksummed_data.crc32c = google_crc32c.value(chunk)
 
             request = storage_type.BidiWriteObjectRequest(
                 write_offset=write_state.bytes_sent,

packages/google-cloud-storage/tests/unit/asyncio/retry/test_writes_resumption_strategy.py

@@ -128,6 +128,22 @@ def test_generate_requests_checksum_verification(self, strategy):
         expected_int = google_crc32c.value(chunk_data)
         assert requests[0].checksummed_data.crc32c == expected_int
 
+    def test_generate_requests_checksum_disabled(self, strategy):
+        """Verify CRC32C is not calculated if enable_checksum is False."""
+        chunk_data = b"test_data"
+        mock_buffer = io.BytesIO(chunk_data)
+        write_state = _WriteState(
+            chunk_size=10,
+            user_buffer=mock_buffer,
+            flush_interval=10,
+            enable_checksum=False,
+        )
+        state = {"write_state": write_state}
+
+        requests = strategy.generate_requests(state)
+
+        assert not requests[0].checksummed_data.crc32c
+
     def test_generate_requests_flush_logic_exact_interval(self, strategy):
         """Verify the flush bit is set exactly when the interval is reached."""
         mock_buffer = io.BytesIO(b"A" * 12)

packages/google-cloud-storage/tests/unit/asyncio/test_async_appendable_object_writer.py

@@ -299,6 +299,29 @@ async def test_append_data_less_than_flush_interval(self, mock_appendable_writer
         assert writer.offset == data_len
         assert writer.bytes_appended_since_last_flush == data_len
 
+    @pytest.mark.asyncio
+    async def test_append_with_checksum_disabled(self, mock_appendable_writer):
+        """Verify append propagates enable_checksum=False to the _WriteState."""
+        writer = self._make_one(mock_appendable_writer["mock_client"])
+        writer._is_stream_open = True
+        writer.persisted_size = 0
+        writer.write_obj_stream = mock_appendable_writer["mock_stream"]
+        writer.write_obj_stream.send = AsyncMock()
+
+        with mock.patch(
+            "google.cloud.storage.asyncio.async_appendable_object_writer._BidiStreamRetryManager"
+        ) as MockManager:
+            mock_execute = AsyncMock()
+            MockManager.return_value.execute = mock_execute
+
+            await writer.append(DATA_LESS_THAN_FLUSH_INTERVAL, enable_checksum=False)
+
+            # Check that execute was called with a state dictionary containing write_state having enable_checksum=False
+            mock_execute.assert_called_once()
+            state_arg = mock_execute.call_args[0][0]
+            assert "write_state" in state_arg
+            assert state_arg["write_state"].enable_checksum is False
+
     @pytest.mark.parametrize(
         "data_len",
         [
@@ -552,3 +575,53 @@ async def test_close_with_checksum_without_finalize_raises(
             match="full_object_checksum can only be provided when finalize_on_close is True",
         ):
             await writer.close(finalize_on_close=False, full_object_checksum=checksum)
+
+    @pytest.mark.asyncio
+    async def test_finalize_invalid_checksum_type(self, mock_appendable_writer):
+        writer = self._make_one(mock_appendable_writer["mock_client"])
+        writer._is_stream_open = True
+        writer.write_obj_stream = mock_appendable_writer["mock_stream"]
+
+        with pytest.raises(TypeError, match="full_object_checksum must be an integer"):
+            await writer.finalize(full_object_checksum="not-an-int")
+
+        with pytest.raises(TypeError, match="full_object_checksum must be an integer"):
+            await writer.finalize(full_object_checksum=True)
+
+    @pytest.mark.asyncio
+    async def test_finalize_invalid_checksum_range(self, mock_appendable_writer):
+        writer = self._make_one(mock_appendable_writer["mock_client"])
+        writer._is_stream_open = True
+        writer.write_obj_stream = mock_appendable_writer["mock_stream"]
+
+        # negative
+        with pytest.raises(
+            ValueError, match="full_object_checksum must be a 32-bit unsigned integer"
+        ):
+            await writer.finalize(full_object_checksum=-1)
+
+        # overflow
+        with pytest.raises(
+            ValueError, match="full_object_checksum must be a 32-bit unsigned integer"
+        ):
+            await writer.finalize(full_object_checksum=0x100000000)
+
+    @pytest.mark.asyncio
+    async def test_finalize_mismatch_closes_stream(self, mock_appendable_writer):
+        writer = self._make_one(mock_appendable_writer["mock_client"])
+        writer._is_stream_open = True
+        writer.write_obj_stream = mock_appendable_writer["mock_stream"]
+
+        # Mock recv to raise an exception (like server rejecting checksum mismatch)
+        from google.api_core.exceptions import InvalidArgument
+
+        mock_appendable_writer["mock_stream"].recv.side_effect = InvalidArgument(
+            "checksum mismatch"
+        )
+
+        with pytest.raises(InvalidArgument):
+            await writer.finalize(full_object_checksum=12345)
+
+        # Assert stream was closed and local state reset despite exception
+        mock_appendable_writer["mock_stream"].close.assert_awaited()
+        assert not writer._is_stream_open