feat(storage): Enable full object checksum PR 3/3: integrate full-object checksum in AsyncMultiRangeDownloader

[chandra-siri]

1. Overview of the Solution

This solution implements end-to-end full-object checksum validation in AsyncMultiRangeDownloader for the asynchronous Google Cloud Storage Python client library. As asynchronous multiplexed downloads of non-contiguous ranges are performed concurrently over a single bidirectional gRPC connection, this feature automatically and incrementally calculates a rolling checksum as bytes arrive and validates it against the server's authoritative object checksum once the download completes.

The technical approach consists of three coordinated layers:

• _AsyncReadObjectStream (Stream Ingestion): Safely extracts the authoritative server checksum (full_obj_server_crc32c) and finalization status (is_finalized) from the object metadata received in the first data payload response of the stream.
• _ReadResumptionStrategy & _DownloadState (Verification Logic): Computes an isolated, persistent rolling checksum in the individual _DownloadState object to ensure calculations do not bleed across concurrent multiplexed ranges. Crucially, the rolling hash updates only after buffer writes succeed to prevent state corruption during retry re-connects, raising a DataCorruption exception on completion if a mismatch occurs.
• AsyncMultiRangeDownloader (Orchestration & Cleanup): Detects candidate full-object ranges (e.g., (0, 0) or (0, persisted_size)), propagates checksum settings to the resumption strategy, and guarantees robust cleanup (closing the stream immediately and unregistering IDs) if data corruption or write errors occur.

2. What This PR Specifically Does

This PR implements Step 3: Downloader Orchestration & End-to-End Integration/System Tests of the solution:

• Relocates raise_if_no_fast_crc32c() validation to the execution phase (download_ranges()) instead of construction time.
• Propagates stream details (is_finalized, full_obj_server_crc32c) to the resumption state dictionary.
• Detects implicit full-object downloads ((0, 0)) or explicit full-object downloads ((0, persisted_size)) post-open(), and flags them for validation.
• Implements the robust cleanup guarantee in download_ranges(): wraps execution in a robust try...finally block to close the stream immediately and unregister multiplexer range IDs upon a DataCorruption exception.
• Adds integration tests in test_async_multi_range_downloader.py and extensive end-to-end system tests in test_zonal.py checking finalized, unfinalized (appendable), explicit, implicit, and bypassed range downloads against live GCS buckets.

[gemini-code-assist]

Code Review

This pull request introduces checksum validation for full object downloads in the AsyncMultiRangeDownloader by tracking rolling checksums and handling DataCorruption exceptions. It also defers the CRC32C C-extension check to the download phase. The review feedback points out an optimization opportunity in the is_full_object_read heuristic to prevent unnecessary CRC32C computations when checksums are disabled or when the object is unfinalized.

[kalragauri]

Do we have a system test that fails if the validation is skipped or applied incorrectly?
If not, consider adding a test that simulates a validation failure to verify that the corruption error is bubbled up the stack.