feat(storage): sync async with main branch (#1754)

googlyrahman · chandra-siri · gemini-code-assist[bot] · web-flow · commit a8940f877aac · 2026-02-19T18:46:55.000+05:30
Sync async with main branch

---------

Co-authored-by: Chandra Shekhar Sirimala &lt;chandrasiri@google.com&gt;
Co-authored-by: gemini-code-assist[bot] &lt;176961590+gemini-code-assist[bot]@users.noreply.github.com&gt;
Co-authored-by: release-please[bot] &lt;55107282+release-please[bot]@users.noreply.github.com&gt;
Co-authored-by: Chalmer Lowe &lt;chalmerlowe@google.com&gt;
Co-authored-by: Owl Bot &lt;gcf-owl-bot[bot]@users.noreply.github.com&gt;
Co-authored-by: Anthonios Partheniou &lt;partheniou@google.com&gt;
Co-authored-by: Pulkit Aggarwal &lt;54775856+Pulkit0110@users.noreply.github.com&gt;
Co-authored-by: Victor Chudnovsky &lt;vchudnov@google.com&gt;
Co-authored-by: gurusai-voleti &lt;gvoleti@google.com&gt;
diff --git a/.github/sync-repo-settings.yaml b/.github/sync-repo-settings.yaml
@@ -10,5 +10,5 @@ branchProtectionRules:
     - 'Kokoro'
     - 'cla/google'
     - 'Kokoro system-3.14'
-    - 'Kokoro system-3.9'
+    - 'Kokoro system-3.10'
     - 'OwlBot Post Processor'
diff --git a/.kokoro/presubmit/system-3.10.cfg b/.kokoro/presubmit/system-3.10.cfg
@@ -0,0 +1,13 @@
+# Format: //devtools/kokoro/config/proto/build.proto
+
+# Only run this nox session.
+env_vars: {
+    key: "NOX_SESSION"
+    value: "system-3.10"
+}
+
+# Credentials needed to test universe domain.
+env_vars: {
+  key: "SECRET_MANAGER_KEYS"
+  value: "client-library-test-universe-domain-credential"
+}
diff --git a/.librarian/generator-input/noxfile.py b/.librarian/generator-input/noxfile.py
@@ -27,8 +27,8 @@
 BLACK_PATHS = ["docs", "google", "tests", "noxfile.py", "setup.py"]
 
 DEFAULT_PYTHON_VERSION = "3.14"
-SYSTEM_TEST_PYTHON_VERSIONS = ["3.9", "3.14"]
-UNIT_TEST_PYTHON_VERSIONS = ["3.7", "3.8", "3.9", "3.10", "3.11", "3.12", "3.13", "3.14"]
+SYSTEM_TEST_PYTHON_VERSIONS = ["3.10", "3.14"]
+UNIT_TEST_PYTHON_VERSIONS = ["3.10", "3.11", "3.12", "3.13", "3.14"]
 CONFORMANCE_TEST_PYTHON_VERSIONS = ["3.12"]
 
 CURRENT_DIRECTORY = pathlib.Path(__file__).parent.absolute()
@@ -44,9 +44,6 @@
     "lint",
     "lint_setup_py",
     "system",
-    # TODO(https://github.com/googleapis/python-storage/issues/1499):
-    # Remove or restore testing for Python 3.7/3.8
-    "unit-3.9",
     "unit-3.10",
     "unit-3.11",
     "unit-3.12",
diff --git a/.librarian/generator-input/setup.py b/.librarian/generator-input/setup.py
@@ -87,9 +87,6 @@
         "License :: OSI Approved :: Apache Software License",
         "Programming Language :: Python",
         "Programming Language :: Python :: 3",
-        "Programming Language :: Python :: 3.7",
-        "Programming Language :: Python :: 3.8",
-        "Programming Language :: Python :: 3.9",
         "Programming Language :: Python :: 3.10",
         "Programming Language :: Python :: 3.11",
         "Programming Language :: Python :: 3.12",
diff --git a/google/cloud/storage/asyncio/async_appendable_object_writer.py b/google/cloud/storage/asyncio/async_appendable_object_writer.py
@@ -211,6 +211,7 @@ def __init__(
         self.bytes_appended_since_last_flush = 0
         self._routing_token: Optional[str] = None
         self.object_resource: Optional[_storage_v2.Object] = None
+        self._flush_count = 0
 
     async def state_lookup(self) -> int:
         """Returns the persisted_size
@@ -318,6 +319,8 @@ async def _do_open():
                 self.write_handle = self.write_obj_stream.write_handle
             if self.write_obj_stream.persisted_size is not None:
                 self.persisted_size = self.write_obj_stream.persisted_size
+                # set offset while opening
+                self.offset = self.persisted_size
 
             self._is_stream_open = True
             self._routing_token = None
@@ -403,36 +406,40 @@ async def generator():
                     write_state.user_buffer.seek(write_state.persisted_size)
                     write_state.bytes_sent = write_state.persisted_size
                     write_state.bytes_since_last_flush = 0
+                    self.bytes_appended_since_last_flush = 0
 
                     requests = strategy.generate_requests(state)
 
-                num_requests = len(requests)
-                for i, chunk_req in enumerate(requests):
-                    if i == num_requests - 1:
-                        chunk_req.state_lookup = True
-                        chunk_req.flush = True
+                for chunk_req in requests:
                     await self.write_obj_stream.send(chunk_req)
-
-                resp = await self.write_obj_stream.recv()
-                if resp:
-                    if resp.persisted_size is not None:
-                        self.persisted_size = resp.persisted_size
-                        state["write_state"].persisted_size = resp.persisted_size
-                        self.offset = self.persisted_size
-                    if resp.write_handle:
-                        self.write_handle = resp.write_handle
-                        state["write_state"].write_handle = resp.write_handle
-                    self.bytes_appended_since_last_flush = 0
-
-                yield resp
+                    if chunk_req.flush:
+                        self._flush_count += 1
+
+                    resp = None
+                    if chunk_req.state_lookup:
+                        # TODO: if there's error, it'll raise error
+                        # and will be handled by `recover_state_on_failure`
+                        resp = await self.write_obj_stream.recv()
+
+                    if resp:
+                        if resp.persisted_size is not None:
+                            self.persisted_size = resp.persisted_size
+                            state["write_state"].persisted_size = resp.persisted_size
+                            self.offset = self.persisted_size
+                        if resp.write_handle:
+                            self.write_handle = resp.write_handle
+                            state["write_state"].write_handle = resp.write_handle
+
+                    yield resp
 
             return generator()
 
         # State initialization
         write_state = _WriteState(_MAX_CHUNK_SIZE_BYTES, buffer, self.flush_interval)
         write_state.write_handle = self.write_handle
         write_state.persisted_size = self.persisted_size
-        write_state.bytes_sent = self.persisted_size
+        # offset is set during `open()` call.
+        write_state.bytes_sent = self.offset or 0
         write_state.bytes_since_last_flush = self.bytes_appended_since_last_flush
 
         retry_manager = _BidiStreamRetryManager(
@@ -442,11 +449,8 @@ async def generator():
         await retry_manager.execute({"write_state": write_state}, retry_policy)
 
         # Sync local markers
-        self.write_obj_stream.persisted_size = write_state.persisted_size
-        self.write_obj_stream.write_handle = write_state.write_handle
         self.bytes_appended_since_last_flush = write_state.bytes_since_last_flush
-        self.persisted_size = write_state.persisted_size
-        self.offset = write_state.persisted_size
+        self.offset = write_state.bytes_sent
 
     async def simple_flush(self) -> None:
         """Flushes the data to the server.
@@ -516,7 +520,6 @@ async def close(self, finalize_on_close=False) -> Union[int, _storage_v2.Object]
         await self.write_obj_stream.close()
 
         self._is_stream_open = False
-        self.offset = None
         return self.persisted_size
 
     async def finalize(self) -> _storage_v2.Object:
diff --git a/google/cloud/storage/asyncio/async_write_object_stream.py b/google/cloud/storage/asyncio/async_write_object_stream.py
@@ -190,7 +190,10 @@ async def requests_done(self):
         _utils.update_write_handle_if_exists(self, first_resp)
 
         if first_resp != grpc.aio.EOF:
-            self.persisted_size = first_resp.persisted_size
+            # this persisted_size will not be upto date., also what if response
+            # doesn't have persisted_size? , it'll throw error.
+            if hasattr(first_resp, "persisted_size"):
+                self.persisted_size = first_resp.persisted_size
             second_resp = await self.socket_like_rpc.recv()
             assert second_resp == grpc.aio.EOF
 
diff --git a/google/cloud/storage/asyncio/retry/writes_resumption_strategy.py b/google/cloud/storage/asyncio/retry/writes_resumption_strategy.py
@@ -47,6 +47,11 @@ def __init__(
         self.chunk_size = chunk_size
         self.user_buffer = user_buffer
         self.persisted_size: int = 0
+        # Bytes sent to the server (it may be unpersisted),
+        # i.e. latest object size = persisted_size + some more bytes.
+        # Please note: these bytes are sent from client to server, server might have also received it.
+        # but might not have persisted it yet (may be in memory buffer on server side).
+        # This variable is same as `offset variable` in the instance of `AppendableObjectWriter`.
         self.bytes_sent: int = 0
         self.bytes_since_last_flush: int = 0
         self.flush_interval: int = flush_interval
@@ -91,7 +96,7 @@ def generate_requests(
 
             if write_state.bytes_since_last_flush >= write_state.flush_interval:
                 request.flush = True
-                # reset counter after marking flush
+                request.state_lookup = True
                 write_state.bytes_since_last_flush = 0
 
             requests.append(request)
diff --git a/noxfile.py b/noxfile.py
@@ -17,7 +17,6 @@
 from __future__ import absolute_import
 import os
 import pathlib
-import re
 import shutil
 
 import nox
@@ -27,9 +26,8 @@
 BLACK_PATHS = ["docs", "google", "tests", "noxfile.py", "setup.py"]
 
 DEFAULT_PYTHON_VERSION = "3.14"
-SYSTEM_TEST_PYTHON_VERSIONS = ["3.9", "3.14"]
+SYSTEM_TEST_PYTHON_VERSIONS = ["3.10", "3.14"]
 UNIT_TEST_PYTHON_VERSIONS = [
-    "3.9",
     "3.10",
     "3.11",
     "3.12",
@@ -51,9 +49,6 @@
     "lint",
     "lint_setup_py",
     "system",
-    # TODO(https://github.com/googleapis/python-storage/issues/1499):
-    # Remove or restore testing for Python 3.7/3.8
-    "unit-3.9",
     "unit-3.10",
     "unit-3.11",
     "unit-3.12",
diff --git a/samples/snippets/zonal_buckets/storage_read_appendable_object_tail.py b/samples/snippets/zonal_buckets/storage_read_appendable_object_tail.py
@@ -28,7 +28,7 @@
     AsyncMultiRangeDownloader,
 )
 
-BYTES_TO_APPEND = b"fav_bytes."
+BYTES_TO_APPEND = b"fav_bytes." * 100 * 1024 * 1024
 NUM_BYTES_TO_APPEND_EVERY_SECOND = len(BYTES_TO_APPEND)
 
 
@@ -37,14 +37,16 @@ async def appender(writer: AsyncAppendableObjectWriter, duration: int):
     """Appends 10 bytes to the object every second for a given duration."""
     print("Appender started.")
     bytes_appended = 0
-    for i in range(duration):
+    start_time = time.monotonic()
+    # Run the appender for the specified duration.
+    while time.monotonic() - start_time < duration:
         await writer.append(BYTES_TO_APPEND)
         now = datetime.now().strftime("%Y-%m-%d %H:%M:%S.%f")[:-3]
         bytes_appended += NUM_BYTES_TO_APPEND_EVERY_SECOND
         print(
             f"[{now}] Appended {NUM_BYTES_TO_APPEND_EVERY_SECOND} new bytes. Total appended: {bytes_appended} bytes."
         )
-        await asyncio.sleep(1)
+        await asyncio.sleep(0.1)
     print("Appender finished.")
 
 
@@ -67,9 +69,7 @@ async def tailer(
             bytes_downloaded = output_buffer.getbuffer().nbytes
             if bytes_downloaded > 0:
                 now = datetime.now().strftime("%Y-%m-%d %H:%M:%S.%f")[:-3]
-                print(
-                    f"[{now}] Tailer read {bytes_downloaded} new bytes: {output_buffer.getvalue()}"
-                )
+                print(f"[{now}] Tailer read {bytes_downloaded} new bytes: ")
                 start_byte += bytes_downloaded
 
             await asyncio.sleep(0.1)  # Poll for new data every 0.1 seconds.
diff --git a/setup.py b/setup.py
@@ -123,9 +123,6 @@
         "License :: OSI Approved :: Apache Software License",
         "Programming Language :: Python",
         "Programming Language :: Python :: 3",
-        "Programming Language :: Python :: 3.7",
-        "Programming Language :: Python :: 3.8",
-        "Programming Language :: Python :: 3.9",
         "Programming Language :: Python :: 3.10",
         "Programming Language :: Python :: 3.11",
         "Programming Language :: Python :: 3.12",
@@ -138,7 +135,7 @@
     packages=packages,
     install_requires=dependencies,
     extras_require=extras,
-    python_requires=">=3.7",
+    python_requires=">=3.10",
     include_package_data=True,
     zip_safe=False,
 )
diff --git a/tests/perf/microbenchmarks/writes/config.py b/tests/perf/microbenchmarks/writes/config.py
@@ -86,7 +86,7 @@ def get_write_params() -> Dict[str, List[WriteParameters]]:
             num_files = num_processes * num_coros
 
             # Create a descriptive name for the parameter set
-            name = f"{workload_name}_{bucket_type}_{num_processes}p_{num_coros}c"
+            name = f"{workload_name}_{bucket_type}_{num_processes}p_{num_coros}c_{chunk_size_mib}csize"
 
             params[workload_name].append(
                 WriteParameters(
diff --git a/tests/perf/microbenchmarks/writes/test_writes.py b/tests/perf/microbenchmarks/writes/test_writes.py
@@ -82,7 +82,9 @@ async def upload_chunks_using_grpc_async(client, filename, other_params):
         uploaded_bytes += bytes_to_upload
     await writer.close()
 
-    assert uploaded_bytes == upload_size
+    # print('writer flush count', writer._flush_count)
+
+    assert writer.offset == upload_size
 
     end_time = time.monotonic_ns()
     elapsed_time = end_time - start_time
diff --git a/tests/system/test_zonal.py b/tests/system/test_zonal.py
@@ -1,8 +1,8 @@
 # py standard imports
-import asyncio
 import os
 import uuid
 from io import BytesIO
+import asyncio
 
 # python additional imports
 import google_crc32c
@@ -437,63 +437,6 @@ async def _read_and_verify(expected_content, generation=None):
     event_loop.run_until_complete(_run())
 
 
-def test_append_flushes_and_state_lookup(
-    storage_client, blobs_to_delete, event_loop, grpc_client
-):
-    """
-    System test for AsyncAppendableObjectWriter, verifying flushing behavior
-    for both small and large appends.
-    """
-    object_name = f"test-append-flush-varied-size-{uuid.uuid4()}"
-
-    async def _run():
-        writer = AsyncAppendableObjectWriter(grpc_client, _ZONAL_BUCKET, object_name)
-
-        # Schedule for cleanup
-        blobs_to_delete.append(storage_client.bucket(_ZONAL_BUCKET).blob(object_name))
-
-        # --- Part 1: Test with small data ---
-        small_data = b"small data"
-
-        await writer.open()
-        assert writer._is_stream_open
-
-        await writer.append(small_data)
-        persisted_size = await writer.state_lookup()
-        assert persisted_size == len(small_data)
-
-        # --- Part 2: Test with large data ---
-        large_data = os.urandom(38 * 1024 * 1024)
-
-        # Append data larger than the default flush interval (16 MiB).
-        # This should trigger the interval-based flushing logic.
-        await writer.append(large_data)
-
-        # Verify the total data has been persisted.
-        total_size = len(small_data) + len(large_data)
-        persisted_size = await writer.state_lookup()
-        assert persisted_size == total_size
-
-        # --- Part 3: Finalize and verify ---
-        final_object = await writer.close(finalize_on_close=True)
-
-        assert not writer._is_stream_open
-        assert final_object.size == total_size
-
-        # Verify the full content of the object.
-        full_data = small_data + large_data
-        mrd = AsyncMultiRangeDownloader(grpc_client, _ZONAL_BUCKET, object_name)
-        buffer = BytesIO()
-        await mrd.open()
-        # (0, 0) means read the whole object
-        await mrd.download_ranges([(0, 0, buffer)])
-        await mrd.close()
-        content = buffer.getvalue()
-        assert content == full_data
-
-    event_loop.run_until_complete(_run())
-
-
 def test_open_with_generation_zero(
     storage_client, blobs_to_delete, event_loop, grpc_client
 ):
diff --git a/tests/unit/asyncio/test_async_appendable_object_writer.py b/tests/unit/asyncio/test_async_appendable_object_writer.py
