docs(storage): add zonal bucket pre-warmed writer pool sample

[NickGoog]

Adds storage_optimize_write_latency_pool sample (region tag: storage_optimize_write_latency_pool) demonstrating a pre-warmed pool of AsyncAppendableObjectWriter instances with finalize_on_close=False to avoid object creation and finalization metadata overhead on the critical write path.

Verified with both mock unit tests and live integration testing against a Rapid (zonal) bucket in us-central1-a:

Running live Python test against bucket=, prefix=live_py_pool_1790087453494
Python 1. Init pool (3 writers): 862.92 ms
Python 2. Write+flush: 82.28 ms
Python 4. Read back: b'0123456789', pool size after refill: 3
Ran 1 test in 1.612s
OK

Description

Fixes #

• Ask me for bug ID

Checklist

Testing

• I have tested this change on a live environment and verified it works as intended.

Compliance & Style

• I have followed the Sample Guidelines.
• The README is updated with all relevant information (if applicable).
• If this PR adds a new sample directory: I updated the CODEOWNERS file.

• N/A

---

Post-Approval Actions

• Please merge this PR for me once it is approved

[snippet-bot]

Here is the summary of changes.

You are about to add 1 region tag.

• storage/samples/snippets/zonal_buckets/storage_optimize_write_latency_pool.py:31, tag storage_optimize_write_latency_pool

---

This comment is generated by snippet-bot.
If you find problems with this result, please file an issue at:
https://github.com/googleapis/repo-automation-bots/issues.
To update this comment, add snippet-bot:force-run label or use the checkbox below:

• Refresh this comment

[gemini-code-assist]

Code Review

This pull request introduces a new code snippet and documentation demonstrating how to optimize write latency in zonal buckets using a pre-warmed writer pool, along with corresponding integration tests. The review feedback suggests two key improvements: concurrently initializing the writer pool using asyncio.gather to reduce startup latency, and wrapping the operations in a try...finally block to prevent resource leaks of the writers and the AsyncGrpcClient in case of exceptions.

[gemini-code-assist]

There are two main areas of improvement in this implementation:

1. Performance/Efficiency: The pre-warmed pool is currently initialized sequentially using a list comprehension with await. Since opening and flushing writers are I/O-bound operations, they can be executed concurrently using asyncio.gather. This will significantly reduce the pool startup latency (e.g., from ~860ms down to ~300ms for 3 writers).
2. Resource Leak / Exception Safety: If an exception occurs during the read operation (step 4) or if maintenance_task fails, the remaining writers in the pool are never closed, and the locally created AsyncGrpcClient is leaked. Wrapping the execution in a try...finally block ensures that all resources are reliably cleaned up.

    grpc_client_created = False
    if grpc_client is None:
        grpc_client = AsyncGrpcClient()
        grpc_client_created = True

    next_object_name = f"{key_prefix}_{pool_size}"

    async def new_prewarmed_writer(name: str) -> AsyncAppendableObjectWriter:
        w = AsyncAppendableObjectWriter(
            client=grpc_client,
            bucket_name=bucket_name,
            object_name=name,
            generation=0,
        )
        await w.open()
        await w.flush()  # Forces 0-byte object creation in the background.
        return w

    # 1. Init pool concurrently: Flushing incurs operation charges, so size the pool
    # carefully. Concurrently initializing the pool avoids sequential latency overhead.
    pool = await asyncio.gather(
        *(new_prewarmed_writer(f"{key_prefix}_{i}") for i in range(pool_size))
    )

    try:
        # 2. Write: Pop a pre-warmed writer; append() writes and flushes data
        # (~1-2 ms).
        writer = pool.pop(0)
        await writer.append(b"0123456789")

        # 3. Pool maintenance (run asynchronously off the critical write path):
        # Close the used writer without finalizing, refill the pool, and discard
        # stale writers.
        async def maintain_pool(used: AsyncAppendableObjectWriter, next_name: str):
            await used.close(finalize_on_close=False)
            pool.append(await new_prewarmed_writer(next_name))

        maintenance_task = asyncio.create_task(
            maintain_pool(writer, next_object_name)
        )

        # 4. Read: Unfinalized objects are readable after flush().
        mrd = AsyncMultiRangeDownloader(
            grpc_client, bucket_name, f"{key_prefix}_0"
        )
        await mrd.open()
        buf = BytesIO()
        await mrd.download_ranges([(0, 0, buf)])
        await mrd.close()

        await maintenance_task
    finally:
        # Ensure all remaining writers in the pool are closed even if an exception occurs.
        for rem in pool:
            await rem.close(finalize_on_close=False)
        # Close the gRPC client if it was created locally to prevent resource leaks.
        if grpc_client_created:
            await grpc_client.close()

[chandra-siri]

/gcbrun(8dd1f99)