From e2a4228769747debd763c6c14ce1890a99165f1a Mon Sep 17 00:00:00 2001 From: Di Wu Date: Fri, 16 Oct 2026 14:04:26 -0700 Subject: [PATCH 1/6] Add async GCS external storage support --- README.md | 1 + changelog/added/bouncing-cloudberry.md | 1 + pyproject.toml | 1 + temporalio/contrib/gcp/gcsdriver/README.md | 101 +++++ temporalio/contrib/gcp/gcsdriver/__init__.py | 13 + temporalio/contrib/gcp/gcsdriver/_client.py | 42 ++ temporalio/contrib/gcp/gcsdriver/_driver.py | 251 ++++++++++++ .../contrib/gcp/gcsdriver/gcloud_aio.py | 62 +++ tests/contrib/gcp/gcsdriver/__init__.py | 0 tests/contrib/gcp/gcsdriver/conftest.py | 32 ++ tests/contrib/gcp/gcsdriver/test_gcsdriver.py | 361 ++++++++++++++++++ .../gcsdriver/test_gcsdriver_integration.py | 83 ++++ tests/contrib/gcp/gcsdriver/workflows.py | 20 + uv.lock | 89 ++++- 14 files changed, 1056 insertions(+), 1 deletion(-) create mode 100644 changelog/added/bouncing-cloudberry.md create mode 100644 temporalio/contrib/gcp/gcsdriver/README.md create mode 100644 temporalio/contrib/gcp/gcsdriver/__init__.py create mode 100644 temporalio/contrib/gcp/gcsdriver/_client.py create mode 100644 temporalio/contrib/gcp/gcsdriver/_driver.py create mode 100644 temporalio/contrib/gcp/gcsdriver/gcloud_aio.py create mode 100644 tests/contrib/gcp/gcsdriver/__init__.py create mode 100644 tests/contrib/gcp/gcsdriver/conftest.py create mode 100644 tests/contrib/gcp/gcsdriver/test_gcsdriver.py create mode 100644 tests/contrib/gcp/gcsdriver/test_gcsdriver_integration.py create mode 100644 tests/contrib/gcp/gcsdriver/workflows.py diff --git a/README.md b/README.md index ef90aa9146..7cfbc94ba8 100644 --- a/README.md +++ b/README.md @@ -557,6 +557,7 @@ Some things to note about driver selection: ###### Built-in Drivers - **[S3 Storage Driver](temporalio/contrib/aws/s3driver/)**: ⚠️ **Experimental** ⚠️ Amazon S3 driver. Ships with an aioboto3 client, or bring your own by subclassing `S3StorageDriverClient`. +- **[GCS Storage Driver](temporalio/contrib/gcp/gcsdriver/)**: ⚠️ **Experimental** ⚠️ Google Cloud Storage driver. Ships with a gcloud-aio-storage client adapter, or bring your own by subclassing `GCSStorageDriverClient`. ###### Custom Drivers diff --git a/changelog/added/bouncing-cloudberry.md b/changelog/added/bouncing-cloudberry.md new file mode 100644 index 0000000000..a0502f3cf5 --- /dev/null +++ b/changelog/added/bouncing-cloudberry.md @@ -0,0 +1 @@ +Add an experimental Google Cloud Storage external payload storage driver with an optional async `gcloud-aio-storage` adapter. diff --git a/pyproject.toml b/pyproject.toml index f803404db6..2de46a242f 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -51,6 +51,7 @@ cloud-run-worker-otel = [ "protobuf<7", ] aioboto3 = ["aioboto3>=10.4.0", "types-aioboto3[s3]>=10.4.0"] +gcloud-aio-storage = ["gcloud-aio-storage>=9.6.4,<10"] google-genai = ["temporalio-google-genai[mcp]>=0.1.0"] strands-agents = ["temporalio-strands-agents>=0.1.0"] diff --git a/temporalio/contrib/gcp/gcsdriver/README.md b/temporalio/contrib/gcp/gcsdriver/README.md new file mode 100644 index 0000000000..b61c10f063 --- /dev/null +++ b/temporalio/contrib/gcp/gcsdriver/README.md @@ -0,0 +1,101 @@ +# Google Cloud Storage Driver + +> ⚠️ **This package is experimental.** ⚠️ + +`GCSStorageDriver` stores and retrieves Temporal payloads in Google Cloud Storage, +following the Python SDK's [S3 driver](../../aws/s3driver/) conventions. + +Install the optional async client dependency: + +```bash +python -m pip install "temporalio[gcloud-aio-storage]" +``` + +```python +import dataclasses + +from gcloud.aio.storage import Storage + +from temporalio.client import Client +from temporalio.contrib.gcp.gcsdriver import GCSStorageDriver +from temporalio.contrib.gcp.gcsdriver.gcloud_aio import new_gcloud_aio_client +from temporalio.converter import DataConverter, ExternalStorage + +async with Storage() as storage: + driver = GCSStorageDriver( + client=new_gcloud_aio_client(storage), + bucket="my-temporal-payloads", + ) + client = await Client.connect( + "localhost:7233", + data_converter=dataclasses.replace( + DataConverter.default, + external_storage=ExternalStorage(drivers=[driver]), + ), + ) + # Run clients and workers inside this context so their storage client stays open. +``` + +The application owns the `Storage` client and configures credentials, request +timeouts, and retry behavior. See the [gcloud-aio documentation](https://talkiq.github.io/gcloud-aio/) +for authentication and customization. The library has no general retry policy; +applications can supply a `Storage` subclass with async retries and timeouts. +The adapter preserves the client's resumable upload handling. + +## Configuration + +- Configure this driver on every Temporal client and worker that stores or retrieves + its payloads, using the same `driver_name` (default: `gcp.gcsdriver`). +- `bucket` accepts a bucket name or a callable taking `(StorageDriverStoreContext, + Payload)` and returning a bucket name for each upload. Reads use the bucket in + the stored claim. +- The bucket must already exist. Use a flat namespace bucket and credentials with + `storage.objects.get` and `storage.objects.create` permissions. +- `ExternalStorage.payload_size_threshold` defaults to 256 KiB. The driver accepts + serialized payloads up to `max_payload_size` (default: 50 MiB). +- Retain objects for as long as workflow histories may need them, including replay. + The driver leaves deletion and lifecycle policies to the application. + +To use another async GCS client, subclass `GCSStorageDriverClient` and implement +`object_exists`, `put_object`, and `get_object`. `put_object` must leave an existing +object unchanged. Importing the driver and client ABC requires no GCS dependencies. + +## Object names and integrity + +Objects contain the complete serialized Temporal `Payload` protobuf, including +metadata. Names use the same layout and percent-encoding as the Python S3 driver: + +```text +v0/ns/{namespace}/wt/{workflow-type}/wi/{workflow-id}/ri/{run-id}/d/sha256/{hex-digest} +v0/ns/{namespace}/at/{activity-type}/ai/{activity-id}/ri/{run-id}/d/sha256/{hex-digest} +v0/d/sha256/{hex-digest} +``` + +The last form applies when target context is unavailable. Missing identity fields +use `null`. Dynamic segments escape UTF-8 bytes outside `a-z`, `A-Z`, `0-9`, and +`! - _ . * ' ( )` as uppercase `%XX`. Object names over 1,024 UTF-8 bytes are rejected. + +Identical serialized bytes in the same context share an object. The driver checks +existence before uploading; the built-in adapter also uses conditional creation to +prevent concurrent writers from overwriting objects. Claims contain `bucket`, `key`, +`hash_algorithm`, and `hash_value`. Downloads verify SHA-256 over the exact stored +bytes before parsing the protobuf. Equivalent values serialized by different SDKs +may produce different bytes and keys. + +## Testing + +Unit tests run without cloud credentials. To also run the emulator and Temporal +worker integration tests, start a local emulator: + +```bash +docker run --rm -p 127.0.0.1:4443:4443 fsouza/fake-gcs-server:1.56.1 \ + -scheme http -external-url http://127.0.0.1:4443 +``` + +In another terminal, run: + +```bash +STORAGE_EMULATOR_HOST=http://127.0.0.1:4443 poe test -s tests/contrib/gcp/gcsdriver +``` + +Emulator integration tests are skipped when `STORAGE_EMULATOR_HOST` is unset. diff --git a/temporalio/contrib/gcp/gcsdriver/__init__.py b/temporalio/contrib/gcp/gcsdriver/__init__.py new file mode 100644 index 0000000000..8032618e8a --- /dev/null +++ b/temporalio/contrib/gcp/gcsdriver/__init__.py @@ -0,0 +1,13 @@ +"""Google Cloud Storage driver for Temporal external storage. + +.. warning:: + This API is experimental. +""" + +from temporalio.contrib.gcp.gcsdriver._client import GCSStorageDriverClient +from temporalio.contrib.gcp.gcsdriver._driver import GCSStorageDriver + +__all__ = [ + "GCSStorageDriverClient", + "GCSStorageDriver", +] diff --git a/temporalio/contrib/gcp/gcsdriver/_client.py b/temporalio/contrib/gcp/gcsdriver/_client.py new file mode 100644 index 0000000000..fbcde1cce7 --- /dev/null +++ b/temporalio/contrib/gcp/gcsdriver/_client.py @@ -0,0 +1,42 @@ +"""GCS storage driver client abstraction for the GCS storage driver. + +.. warning:: + This API is experimental. +""" + +from __future__ import annotations + +from abc import ABC, abstractmethod +from collections.abc import Mapping + + +class GCSStorageDriverClient(ABC): + """Abstract base class for GCS object operations. + + Implementations must support ``object_exists``, ``put_object``, and + ``get_object``. Resumable uploads (if needed) are an internal concern of + each implementation. Uploads must leave existing objects unchanged. + + .. warning:: + This API is experimental. + """ + + @abstractmethod + async def put_object(self, *, bucket: str, key: str, data: bytes) -> None: + """Upload *data* if absent at the given GCS *bucket* and *key*.""" + + @abstractmethod + async def object_exists(self, *, bucket: str, key: str) -> bool: + """Return ``True`` if an object exists at the given *bucket* and *key*.""" + + @abstractmethod + async def get_object(self, *, bucket: str, key: str) -> bytes: + """Download and return the bytes stored at the given GCS *bucket* and *key*.""" + + def describe(self) -> Mapping[str, str]: + """Return client-specific diagnostic metadata (e.g. region, credentials + source) that the driver appends to error messages. Implementations may + override this to surface configuration that is useful for debugging + common misconfigurations. Returns an empty mapping by default. + """ + return {} diff --git a/temporalio/contrib/gcp/gcsdriver/_driver.py b/temporalio/contrib/gcp/gcsdriver/_driver.py new file mode 100644 index 0000000000..e699ed46aa --- /dev/null +++ b/temporalio/contrib/gcp/gcsdriver/_driver.py @@ -0,0 +1,251 @@ +"""Google Cloud Storage driver for Temporal external storage. + +.. warning:: + This API is experimental. +""" + +from __future__ import annotations + +import asyncio +import hashlib +import string +from collections.abc import Callable, Coroutine, Sequence +from typing import Any, TypeVar + +from temporalio.api.common.v1 import Payload +from temporalio.contrib.gcp.gcsdriver._client import GCSStorageDriverClient +from temporalio.converter import ( + StorageDriver, + StorageDriverActivityInfo, + StorageDriverClaim, + StorageDriverRetrieveContext, + StorageDriverStoreContext, + StorageDriverWorkflowInfo, +) + +_T = TypeVar("_T") + +# S3's safe character set for object key names. See +# https://docs.aws.amazon.com/AmazonS3/latest/userguide/object-keys.html +_S3_SAFE_CHARS = frozenset(string.ascii_letters + string.digits + "!-_.*'()") + + +def _percent_encode(val: str | None) -> str | None: + """Percent-encode a path segment per the S3 key spec. + + Encodes every byte outside :data:`_S3_SAFE_CHARS` as ``%XX`` with + upper-case hex digits, operating on the UTF-8 bytes so non-ASCII + characters are escaped byte by byte. Returns ``None`` for empty or + missing values so callers can substitute ``"null"``. + """ + if not val: + return None + return "".join( + chr(byte) if chr(byte) in _S3_SAFE_CHARS else f"%{byte:02X}" + for byte in val.encode("utf-8") + ) + + +def _format_client_context(client: GCSStorageDriverClient) -> str: + """Format the client's ``describe()`` output as ", k=v, k=v" for error + messages. Returns an empty string when the client reports no metadata or + describe itself raises (diagnostic output must never mask the real error). + """ + try: + info = client.describe() + except Exception: + return "" + if not info: + return "" + return "".join(f", {k}={v}" for k, v in info.items()) + + +async def _gather_with_cancellation( + coros: Sequence[Coroutine[Any, Any, _T]], +) -> list[_T]: + """Run coroutines concurrently, cancelling all remaining tasks if one fails.""" + if not coros: + return [] + tasks = [asyncio.ensure_future(c) for c in coros] + try: + return list(await asyncio.gather(*tasks)) + except BaseException: + for t in tasks: + t.cancel() + await asyncio.gather(*tasks, return_exceptions=True) + raise + + +class GCSStorageDriver(StorageDriver): + """Driver for storing and retrieving Temporal payloads in Google Cloud Storage. + + Requires an :class:`GCSStorageDriverClient` and a ``bucket``. Payloads are keyed by + a SHA-256 hash of their serialized bytes, segmented by namespace and + workflow/activity identifiers derived from the serialization context. + + .. warning:: + This API is experimental. + """ + + def __init__( + self, + client: GCSStorageDriverClient, + bucket: str | Callable[[StorageDriverStoreContext, Payload], str], + driver_name: str = "gcp.gcsdriver", + max_payload_size: int = 50 * 1024 * 1024, + ): + """Constructs the GCS driver. + + Args: + client: An :class:`GCSStorageDriverClient` implementation. Use + :func:`temporalio.contrib.gcp.gcsdriver.gcloud_aio.new_gcloud_aio_client` to + wrap a gcloud-aio-storage client. + bucket: GCS bucket name or a callable that + accepts ``(StorageDriverStoreContext, Payload)`` and returns + a bucket name. A callable allows dynamic per-payload bucket + selection. + driver_name: Name of this driver instance. Defaults to + ``"gcp.gcsdriver"``. Override when registering + multiple GCSStorageDriver instances with distinct configurations + under the same ``temporalio.converter.ExternalStorage.drivers`` list. + max_payload_size: Maximum serialized payload size in bytes that the + driver will accept. Defaults to 52428800 (50 MiB). Raise this + value if your workload requires larger payloads; lower it to + enforce stricter limits. + """ + if max_payload_size <= 0: + raise ValueError("max_payload_size must be greater than zero") + self._client = client + self._bucket = bucket + self._driver_name = driver_name or "gcp.gcsdriver" + self._max_payload_size = max_payload_size + + def name(self) -> str: + """Return the driver instance name.""" + return self._driver_name + + def type(self) -> str: + """Return the driver type identifier.""" + return "gcp.gcsdriver" + + def _get_bucket(self, context: StorageDriverStoreContext, payload: Payload) -> str: + """Resolve bucket using the configured strategy.""" + if callable(self._bucket): + return self._bucket(context, payload) + return self._bucket + + async def store( + self, + context: StorageDriverStoreContext, + payloads: Sequence[Payload], + ) -> list[StorageDriverClaim]: + """Stores payloads in GCS and returns a ``StorageDriverClaim`` for each one. + + Payloads are keyed by their SHA-256 hash, so identical serialized bytes + share the same GCS object. Deduplication is best-effort because the same + Python value may serialize differently across payload converter versions + (e.g. proto binary). The returned list is the same length as + ``payloads``. + """ + context_segments = "" + target = context.target + namespace = _percent_encode(target.namespace) if target is not None else None + namespace_segment = f"/ns/{namespace}" if namespace else "" + if isinstance(target, StorageDriverWorkflowInfo): + wf_type = _percent_encode(target.type) or "null" + wf_id = _percent_encode(target.id) or "null" + wf_run_id = _percent_encode(target.run_id) or "null" + context_segments = f"/wt/{wf_type}/wi/{wf_id}/ri/{wf_run_id}" + elif isinstance(target, StorageDriverActivityInfo): + act_type = _percent_encode(target.type) or "null" + act_id = _percent_encode(target.id) or "null" + act_run_id = _percent_encode(target.run_id) or "null" + context_segments = f"/at/{act_type}/ai/{act_id}/ri/{act_run_id}" + + async def _upload(payload: Payload) -> StorageDriverClaim: + bucket = self._get_bucket(context, payload) + + payload_bytes = payload.SerializeToString() + if len(payload_bytes) > self._max_payload_size: + raise ValueError( + f"Payload size {len(payload_bytes)} bytes exceeds the configured " + f"max_payload_size of {self._max_payload_size} bytes" + ) + + hash_digest = hashlib.sha256(payload_bytes).hexdigest() + + digest_segments = f"/d/sha256/{hash_digest}" + + key = f"v0{namespace_segment}{context_segments}{digest_segments}" + if len(key.encode("utf-8")) > 1024: + raise ValueError("GCS object name exceeds maximum 1024 UTF-8 bytes") + + try: + if not await self._client.object_exists(bucket=bucket, key=key): + await self._client.put_object( + bucket=bucket, key=key, data=payload_bytes + ) + except Exception as e: + raise RuntimeError( + f"GCSStorageDriver store failed [bucket={bucket}, key={key}" + f"{_format_client_context(self._client)}]" + ) from e + + return StorageDriverClaim( + claim_data={ + "bucket": bucket, + "key": key, + "hash_algorithm": "sha256", + "hash_value": hash_digest, + }, + ) + + return await _gather_with_cancellation([_upload(p) for p in payloads]) + + async def retrieve( + self, + context: StorageDriverRetrieveContext, # noqa: ARG002 + claims: Sequence[StorageDriverClaim], + ) -> list[Payload]: + """Retrieves payloads from GCS for the given ``StorageDriverClaim`` list.""" + + async def _download(claim: StorageDriverClaim) -> Payload: + bucket = claim.claim_data["bucket"] + key = claim.claim_data["key"] + + try: + payload_bytes = await self._client.get_object(bucket=bucket, key=key) + except Exception as e: + raise RuntimeError( + f"GCSStorageDriver retrieve failed [bucket={bucket}, key={key}" + f"{_format_client_context(self._client)}]" + ) from e + + hash_algorithm = claim.claim_data.get("hash_algorithm") + expected_hash = claim.claim_data.get("hash_value") + if not hash_algorithm or not expected_hash: + raise ValueError( + f"GCSStorageDriver claim is missing required content hash information " + f"[bucket={bucket}, key={key}]: " + f"claim_data must contain 'hash_algorithm' and 'hash_value'" + ) + if hash_algorithm != "sha256": + raise ValueError( + f"GCSStorageDriver unsupported hash algorithm " + f"[bucket={bucket}, key={key}]: " + f"expected sha256, got {hash_algorithm}" + ) + actual_hash = hashlib.sha256(payload_bytes).hexdigest() + if actual_hash != expected_hash: + raise ValueError( + f"GCSStorageDriver integrity check failed " + f"[bucket={bucket}, key={key}]: " + f"expected {hash_algorithm}:{expected_hash}, " + f"got {hash_algorithm}:{actual_hash}" + ) + + payload = Payload() + payload.ParseFromString(payload_bytes) + return payload + + return await _gather_with_cancellation([_download(c) for c in claims]) diff --git a/temporalio/contrib/gcp/gcsdriver/gcloud_aio.py b/temporalio/contrib/gcp/gcsdriver/gcloud_aio.py new file mode 100644 index 0000000000..2ca4ac7842 --- /dev/null +++ b/temporalio/contrib/gcp/gcsdriver/gcloud_aio.py @@ -0,0 +1,62 @@ +"""Gcloud-aio-storage adapter for the GCS storage driver client. + +.. warning:: + This API is experimental. +""" + +from __future__ import annotations + +from aiohttp import ClientResponseError +from gcloud.aio.storage import Storage + +from temporalio.contrib.gcp.gcsdriver._client import GCSStorageDriverClient + + +class _GcloudAioStorageDriverClient(GCSStorageDriverClient): + """Adapter for an application-owned gcloud-aio-storage client.""" + + def __init__(self, client: Storage) -> None: + """Wrap the async storage client.""" + self._client = client + + async def object_exists(self, *, bucket: str, key: str) -> bool: + """Check existence via object metadata.""" + try: + await self._client.download_metadata(bucket, key) + return True + except ClientResponseError as err: + if err.status == 404: + return False + raise + + async def put_object(self, *, bucket: str, key: str, data: bytes) -> None: + """Upload bytes using conditional object creation.""" + try: + await self._client.upload( + bucket, + key, + data, + content_type="application/octet-stream", + parameters={"ifGenerationMatch": "0"}, + ) + except ClientResponseError as err: + # Another writer may create the content-addressed object after our check. + if err.status != 412: + raise + + async def get_object(self, *, bucket: str, key: str) -> bytes: + """Download the object's bytes.""" + return await self._client.download(bucket, key) + + +def new_gcloud_aio_client(client: Storage) -> GCSStorageDriverClient: + """Create a GCS storage driver client from a gcloud-aio-storage client. + + Args: + client: An application-owned ``gcloud.aio.storage.Storage`` client. Keep + it open until all Temporal clients and workers using it have stopped. + + .. warning:: + This API is experimental. + """ + return _GcloudAioStorageDriverClient(client) diff --git a/tests/contrib/gcp/gcsdriver/__init__.py b/tests/contrib/gcp/gcsdriver/__init__.py new file mode 100644 index 0000000000..e69de29bb2 diff --git a/tests/contrib/gcp/gcsdriver/conftest.py b/tests/contrib/gcp/gcsdriver/conftest.py new file mode 100644 index 0000000000..2f8c03e542 --- /dev/null +++ b/tests/contrib/gcp/gcsdriver/conftest.py @@ -0,0 +1,32 @@ +"""Fixtures for GCS tests against an explicitly configured local emulator.""" + +from __future__ import annotations + +import os +import uuid +from collections.abc import AsyncIterator + +import aiohttp +import pytest +from gcloud.aio.storage import Storage + + +@pytest.fixture +def gcs_bucket() -> str: + return f"temporal-test-{uuid.uuid4()}" + + +@pytest.fixture +async def gcs_storage(gcs_bucket: str) -> AsyncIterator[Storage]: + endpoint = os.environ.get("STORAGE_EMULATOR_HOST") + if not endpoint: + pytest.skip("Set STORAGE_EMULATOR_HOST to run GCS emulator integration tests") + async with aiohttp.ClientSession() as session: + async with session.post( + f"{endpoint.rstrip('/')}/storage/v1/b", + params={"project": "temporal-test"}, + json={"name": gcs_bucket}, + ) as response: + response.raise_for_status() + async with Storage(api_root=endpoint) as storage: + yield storage diff --git a/tests/contrib/gcp/gcsdriver/test_gcsdriver.py b/tests/contrib/gcp/gcsdriver/test_gcsdriver.py new file mode 100644 index 0000000000..8d05499f80 --- /dev/null +++ b/tests/contrib/gcp/gcsdriver/test_gcsdriver.py @@ -0,0 +1,361 @@ +"""Tests for the GCS storage driver and its async client adapter.""" + +from __future__ import annotations + +import asyncio +import hashlib +from unittest.mock import AsyncMock, MagicMock + +import pytest +from aiohttp import ClientResponseError +from gcloud.aio.storage import Storage + +from temporalio.api.common.v1 import Payload +from temporalio.contrib.gcp.gcsdriver import GCSStorageDriver, GCSStorageDriverClient +from temporalio.contrib.gcp.gcsdriver.gcloud_aio import new_gcloud_aio_client +from temporalio.converter import ( + ExternalStorage, + StorageDriverActivityInfo, + StorageDriverClaim, + StorageDriverRetrieveContext, + StorageDriverStoreContext, + StorageDriverWorkflowInfo, +) + + +class MemoryClient(GCSStorageDriverClient): + def __init__(self) -> None: + self.objects: dict[tuple[str, str], bytes] = {} + self.uploads = 0 + + async def object_exists(self, *, bucket: str, key: str) -> bool: + return (bucket, key) in self.objects + + async def put_object(self, *, bucket: str, key: str, data: bytes) -> None: + self.uploads += 1 + self.objects.setdefault((bucket, key), data) + + async def get_object(self, *, bucket: str, key: str) -> bytes: + return self.objects[bucket, key] + + +@pytest.fixture +def client() -> MemoryClient: + return MemoryClient() + + +def payload(value: bytes = b"hello") -> Payload: + return Payload( + metadata={"encoding": b"binary/plain", "custom": b"\x00\xff"}, data=value + ) + + +def test_configuration(client: MemoryClient) -> None: + driver = GCSStorageDriver(client, "bucket") + assert driver.name() == driver.type() == "gcp.gcsdriver" + assert GCSStorageDriver(client, "bucket", driver_name="archive").name() == "archive" + for limit in (0, -1): + with pytest.raises(ValueError, match="max_payload_size"): + GCSStorageDriver(client, "bucket", max_payload_size=limit) + + +@pytest.mark.parametrize( + "target,prefix", + [ + (None, "v0"), + ( + StorageDriverWorkflowInfo( + namespace="ns", type="Workflow", id="wf", run_id="run" + ), + "v0/ns/ns/wt/Workflow/wi/wf/ri/run", + ), + ( + StorageDriverActivityInfo(namespace="ns", type="Activity", id="act"), + "v0/ns/ns/at/Activity/ai/act/ri/null", + ), + ( + StorageDriverWorkflowInfo(namespace="ns"), + "v0/ns/ns/wt/null/wi/null/ri/null", + ), + ( + StorageDriverWorkflowInfo(namespace="ns /%", id="日本語 😀", type="A+B=C"), + "v0/ns/ns%20%2F%25/wt/A%2BB%3DC/wi/%E6%97%A5%E6%9C%AC%E8%AA%9E%20%F0%9F%98%80/ri/null", + ), + ], +) +async def test_keys( + client: MemoryClient, + target: StorageDriverWorkflowInfo | StorageDriverActivityInfo | None, + prefix: str, +) -> None: + original = payload() + [claim] = await GCSStorageDriver(client, "bucket").store( + StorageDriverStoreContext(target=target), [original] + ) + digest = hashlib.sha256(original.SerializeToString()).hexdigest() + assert claim.claim_data == { + "bucket": "bucket", + "key": f"{prefix}/d/sha256/{digest}", + "hash_algorithm": "sha256", + "hash_value": digest, + } + + +async def test_round_trip_and_deduplication(client: MemoryClient) -> None: + driver = GCSStorageDriver(client, "bucket") + originals = [payload(), payload(b"world")] + claims = await driver.store(StorageDriverStoreContext(), originals) + assert await driver.store(StorageDriverStoreContext(), originals) == claims + assert client.uploads == 2 + assert ( + await GCSStorageDriver(client, "new-bucket").retrieve( + StorageDriverRetrieveContext(), claims + ) + == originals + ) + assert await driver.store(StorageDriverStoreContext(), []) == [] + assert await driver.retrieve(StorageDriverRetrieveContext(), []) == [] + + +async def test_callable_bucket(client: MemoryClient) -> None: + context = StorageDriverStoreContext( + target=StorageDriverWorkflowInfo(namespace="ns") + ) + + def choose_bucket(ctx: StorageDriverStoreContext, value: Payload) -> str: + assert ctx is context + return value.data.decode() + + claims = await GCSStorageDriver(client, choose_bucket).store( + context, [payload(b"first"), payload(b"second")] + ) + assert [c.claim_data["bucket"] for c in claims] == ["first", "second"] + + +async def test_size_includes_metadata(client: MemoryClient) -> None: + original = payload() + driver = GCSStorageDriver(client, "bucket", max_payload_size=original.ByteSize()) + await driver.store(StorageDriverStoreContext(), [original]) + original.metadata["extra"] = b"too large" + with pytest.raises(ValueError, match="max_payload_size"): + await driver.store(StorageDriverStoreContext(), [original]) + assert client.uploads == 1 + + +async def test_object_name_limit(client: MemoryClient) -> None: + context = StorageDriverStoreContext( + target=StorageDriverWorkflowInfo(namespace="ns", id="日" * 200) + ) + with pytest.raises(ValueError, match="1024 UTF-8 bytes"): + await GCSStorageDriver(client, "bucket").store(context, [payload()]) + assert not client.objects + + +@pytest.mark.parametrize( + "field,value", + [("hash_algorithm", "md5"), ("hash_value", ""), ("hash_algorithm", "")], +) +async def test_invalid_hash_claim(client: MemoryClient, field: str, value: str) -> None: + driver = GCSStorageDriver(client, "bucket") + [claim] = await driver.store(StorageDriverStoreContext(), [payload()]) + data = dict(claim.claim_data) + data[field] = value + with pytest.raises(ValueError, match="hash"): + await driver.retrieve( + StorageDriverRetrieveContext(), [StorageDriverClaim(data)] + ) + + +async def test_corruption(client: MemoryClient) -> None: + driver = GCSStorageDriver(client, "bucket") + [claim] = await driver.store(StorageDriverStoreContext(), [payload()]) + client.objects["bucket", claim.claim_data["key"]] = b"corrupted" + with pytest.raises(ValueError, match="integrity check failed"): + await driver.retrieve(StorageDriverRetrieveContext(), [claim]) + + +async def test_external_storage_reference_and_threshold(client: MemoryClient) -> None: + driver = GCSStorageDriver(client, "bucket") + storage = ExternalStorage(drivers=[driver], payload_size_threshold=100) + original = payload(b"large" * 100) + assert await storage._store_payload(payload()) == payload() + reference = await storage._store_payload(original) + assert reference.metadata["encoding"] == b"json/protobuf" + assert ( + reference.metadata["messageType"] + == b"temporal.api.sdk.v1.ExternalStorageReference" + ) + assert reference.external_payloads[0].size_bytes == original.ByteSize() + assert await storage._retrieve_payload(reference) == original + + +@pytest.mark.parametrize("operation", ["object_exists", "put_object", "get_object"]) +async def test_provider_error_context(operation: str) -> None: + client = AsyncMock(spec=GCSStorageDriverClient) + client.object_exists.return_value = False + cause = ConnectionError("connection lost") + getattr(client, operation).side_effect = cause + driver = GCSStorageDriver(client, "bucket") + with pytest.raises(RuntimeError, match="bucket=bucket, key=") as error: + if operation == "get_object": + await driver.retrieve( + StorageDriverRetrieveContext(), + [StorageDriverClaim({"bucket": "bucket", "key": "key"})], + ) + else: + await driver.store(StorageDriverStoreContext(), [payload()]) + assert error.value.__cause__ is cause + + +@pytest.mark.parametrize("retrieving", [False, True]) +@pytest.mark.parametrize("cancel_parent", [False, True]) +async def test_batch_cancellation(retrieving: bool, cancel_parent: bool) -> None: + client = AsyncMock(spec=GCSStorageDriverClient) + client.object_exists.return_value = False + both_started = asyncio.Event() + release = asyncio.Event() + started = 0 + cancelled = 0 + + async def transfer(**_kwargs: object) -> bytes: + nonlocal started, cancelled + started += 1 + index = started + if started == 2: + both_started.set() + try: + await release.wait() + if index == 1: + raise ConnectionError("failed") + await asyncio.Event().wait() + except asyncio.CancelledError: + cancelled += 1 + raise + return b"" + + client.get_object.side_effect = transfer + client.put_object.side_effect = transfer + driver = GCSStorageDriver(client, "bucket") + task: asyncio.Task[list[Payload]] | asyncio.Task[list[StorageDriverClaim]] + if retrieving: + claim = StorageDriverClaim({"bucket": "bucket", "key": "key"}) + task = asyncio.create_task( + driver.retrieve(StorageDriverRetrieveContext(), [claim, claim]) + ) + else: + task = asyncio.create_task( + driver.store(StorageDriverStoreContext(), [payload(), payload(b"other")]) + ) + await both_started.wait() + if cancel_parent: + task.cancel() + with pytest.raises(asyncio.CancelledError): + await task + assert cancelled == 2 + else: + release.set() + with pytest.raises(RuntimeError): + await task + assert cancelled == 1 + + +@pytest.mark.parametrize("retrieving", [False, True]) +async def test_concurrent_transfers_preserve_order(retrieving: bool) -> None: + client = AsyncMock(spec=GCSStorageDriverClient) + client.object_exists.return_value = False + originals = [payload(str(i).encode()) for i in range(3)] + futures: list[asyncio.Future[bytes]] = [ + asyncio.get_running_loop().create_future() for _ in originals + ] + started = 0 + all_started = asyncio.Event() + + async def transfer(**_kwargs: object) -> bytes: + nonlocal started + future = futures[started] + started += 1 + if started == len(originals): + all_started.set() + return await future + + client.get_object.side_effect = transfer + client.put_object.side_effect = transfer + driver = GCSStorageDriver(client, "bucket") + claims = [ + StorageDriverClaim( + { + "bucket": "bucket", + "key": str(i), + "hash_algorithm": "sha256", + "hash_value": hashlib.sha256(p.SerializeToString()).hexdigest(), + } + ) + for i, p in enumerate(originals) + ] + task: asyncio.Task[list[Payload]] | asyncio.Task[list[StorageDriverClaim]] + if retrieving: + task = asyncio.create_task( + driver.retrieve(StorageDriverRetrieveContext(), claims) + ) + else: + task = asyncio.create_task(driver.store(StorageDriverStoreContext(), originals)) + await all_started.wait() + for i in reversed(range(len(originals))): + futures[i].set_result(originals[i].SerializeToString()) + await asyncio.sleep(0) + if retrieving: + assert await task == originals + else: + for actual, expected in zip(await task, claims, strict=True): + assert isinstance(actual, StorageDriverClaim) + assert actual.claim_data["hash_value"] == expected.claim_data["hash_value"] + + +def response_error(status: int) -> ClientResponseError: + return ClientResponseError(MagicMock(), (), status=status) + + +@pytest.mark.parametrize("status", [None, 404, 403, 500]) +async def test_adapter_object_exists(status: int | None) -> None: + storage = AsyncMock(spec=Storage) + if status: + storage.download_metadata.side_effect = response_error(status) + client = new_gcloud_aio_client(storage) + if status in (403, 500): + with pytest.raises(ClientResponseError): + await client.object_exists(bucket="bucket", key="key") + else: + assert await client.object_exists(bucket="bucket", key="key") is ( + status is None + ) + storage.download_metadata.assert_awaited_once_with("bucket", "key") + + +@pytest.mark.parametrize("status", [None, 412, 403, 500]) +async def test_adapter_conditional_upload(status: int | None) -> None: + storage = AsyncMock(spec=Storage) + if status: + storage.upload.side_effect = response_error(status) + client = new_gcloud_aio_client(storage) + if status in (403, 500): + with pytest.raises(ClientResponseError): + await client.put_object(bucket="bucket", key="key", data=b"data") + else: + await client.put_object(bucket="bucket", key="key", data=b"data") + storage.upload.assert_awaited_once_with( + "bucket", + "key", + b"data", + content_type="application/octet-stream", + parameters={"ifGenerationMatch": "0"}, + ) + + +async def test_adapter_download() -> None: + storage = AsyncMock(spec=Storage) + storage.download.return_value = b"\x00\xff" + assert ( + await new_gcloud_aio_client(storage).get_object(bucket="bucket", key="key") + == b"\x00\xff" + ) + storage.download.assert_awaited_once_with("bucket", "key") diff --git a/tests/contrib/gcp/gcsdriver/test_gcsdriver_integration.py b/tests/contrib/gcp/gcsdriver/test_gcsdriver_integration.py new file mode 100644 index 0000000000..e01c5a0eb9 --- /dev/null +++ b/tests/contrib/gcp/gcsdriver/test_gcsdriver_integration.py @@ -0,0 +1,83 @@ +"""Driver and worker integration tests using a local GCS emulator.""" + +import asyncio +import dataclasses +import uuid +from datetime import timedelta + +import pytest +from gcloud.aio.storage import Storage + +from temporalio.api.common.v1 import Payload +from temporalio.contrib.gcp.gcsdriver import GCSStorageDriver +from temporalio.contrib.gcp.gcsdriver.gcloud_aio import new_gcloud_aio_client +from temporalio.converter import ( + DataConverter, + ExternalStorage, + StorageDriverRetrieveContext, + StorageDriverStoreContext, + StorageDriverWorkflowInfo, +) +from temporalio.testing import WorkflowEnvironment +from tests.contrib.gcp.gcsdriver.workflows import EchoWorkflow, echo_activity +from tests.helpers import new_worker + + +@pytest.mark.parametrize("size", [0, 6 * 1024 * 1024]) +async def test_gcs_emulator_round_trip( + gcs_storage: Storage, gcs_bucket: str, size: int +) -> None: + client = new_gcloud_aio_client(gcs_storage) + driver = GCSStorageDriver(client, gcs_bucket) + original = Payload( + metadata={"encoding": b"binary/plain", "custom": b"\x00\xff"}, data=b"x" * size + ) + context = StorageDriverStoreContext( + target=StorageDriverWorkflowInfo( + namespace="ns", id="日本語 /%+", type="Workflow" + ) + ) + claims = await driver.store(context, [original]) + assert await driver.store(context, [original]) == claims + assert await driver.retrieve(StorageDriverRetrieveContext(), claims) == [original] + key = claims[0].claim_data["key"] + await asyncio.gather( + client.put_object(bucket=gcs_bucket, key=key, data=b"must not overwrite"), + client.put_object(bucket=gcs_bucket, key=key, data=b"must not overwrite"), + ) + assert ( + await client.get_object(bucket=gcs_bucket, key=key) + == original.SerializeToString() + ) + + +async def test_gcs_workflow_round_trip( + gcs_storage: Storage, gcs_bucket: str, env: WorkflowEnvironment +) -> None: + driver = GCSStorageDriver(new_gcloud_aio_client(gcs_storage), gcs_bucket) + client = await env.connect_client( + data_converter=dataclasses.replace( + DataConverter.default, + external_storage=ExternalStorage( + drivers=[driver], payload_size_threshold=256 + ), + ) + ) + value = "large workflow input 日本語" * 100 + workflow_id = str(uuid.uuid4()) + async with new_worker(client, EchoWorkflow, activities=[echo_activity]) as worker: + assert ( + await client.execute_workflow( + EchoWorkflow.run, + value, + id=workflow_id, + task_queue=worker.task_queue, + execution_timeout=timedelta(seconds=30), + ) + == value + ) + objects = await gcs_storage.list_objects(gcs_bucket) + keys = [item["name"] for item in objects["items"]] + assert len(keys) == 2 + assert all(f"/wt/EchoWorkflow/wi/{workflow_id}/ri/" in key for key in keys) + assert sum("/ri/null/" in key for key in keys) == 1 diff --git a/tests/contrib/gcp/gcsdriver/workflows.py b/tests/contrib/gcp/gcsdriver/workflows.py new file mode 100644 index 0000000000..7db8ef2afc --- /dev/null +++ b/tests/contrib/gcp/gcsdriver/workflows.py @@ -0,0 +1,20 @@ +"""Workflow definitions kept separate from optional GCS client imports.""" + +from datetime import timedelta + +import temporalio.activity +import temporalio.workflow + + +@temporalio.activity.defn +async def echo_activity(value: str) -> str: + return value + + +@temporalio.workflow.defn +class EchoWorkflow: + @temporalio.workflow.run + async def run(self, value: str) -> str: + return await temporalio.workflow.execute_activity( + echo_activity, value, start_to_close_timeout=timedelta(seconds=10) + ) diff --git a/uv.lock b/uv.lock index 6e93742687..53bc33c99a 100644 --- a/uv.lock +++ b/uv.lock @@ -614,6 +614,59 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/86/eb/f9e2ce6888b5e77072169d884afcd957e69d3f293b03b4eb4dfad48fe358/cfn_lint-1.53.0-py3-none-any.whl", hash = "sha256:8a91d1f0d8d18614410a0c1676213df8cd0ac55fa462f3f6369fb9fdaa7ae28d", size = 5049598, upload-time = "2026-07-09T18:09:14.108Z" }, ] +[[package]] +name = "chardet" +version = "7.6.0" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/b1/51/cd61c567092a6cec796144510a68aff158ebfc1df82950a45bae65f28413/chardet-7.6.0.tar.gz", hash = "sha256:93d9df6089ded42ed1fe9f57e272c0b74bd0464d45c0c7d50f09f26f31105c3c", size = 914462, upload-time = "2026-08-14T20:36:59.305Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/8c/cd/dc9df6b57c043037920a3896ccb08a21e0595fe592b7acb9fc85acaf0698/chardet-7.6.0-cp310-cp310-macosx_10_9_x86_64.whl", hash = "sha256:cbaca8f563a9de07ab1a53157dba93802e54c26afe3339892afcc7c59ea4ef1b", size = 1093508, upload-time = "2026-08-14T20:36:07.451Z" }, + { url = "https://files.pythonhosted.org/packages/b9/66/acab13c5bbe55410273530a41b83fdcde0b653f6bded34702504c0ab95cb/chardet-7.6.0-cp310-cp310-macosx_11_0_arm64.whl", hash = "sha256:55a4c31adc7c7e83ad412f2f66b6b7358d0d4fe67505e7f58e18f68f75d341bb", size = 1074662, upload-time = "2026-08-14T20:36:09.209Z" }, + { url = "https://files.pythonhosted.org/packages/20/38/b32daa70f8bacd47e515ec9f85d571c87f31ff1c43c95aea7780eaefe7fc/chardet-7.6.0-cp310-cp310-manylinux2014_aarch64.manylinux_2_17_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:7b586cab9e9072dddd89bc2bd27ee72808d0c84ec73695fe6ec0f3c46b057c65", size = 1465648, upload-time = "2026-08-14T20:36:10.774Z" }, + { url = "https://files.pythonhosted.org/packages/f2/34/0fd9d566df9647d0820a35b4d119db971da3ae9be20899d2b4af11d7ab58/chardet-7.6.0-cp310-cp310-manylinux2014_x86_64.manylinux_2_17_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:4b81d3f7d7914442d5f7d515b8c6d79cee6b794bc208971fb6902f176671166a", size = 1484628, upload-time = "2026-08-14T20:36:12.456Z" }, + { url = "https://files.pythonhosted.org/packages/ad/be/1c57863e21c1d6cb6b0fcdde313f8a24039eca01cc90ef0f0986c0964969/chardet-7.6.0-cp310-cp310-manylinux_2_31_riscv64.manylinux_2_39_riscv64.whl", hash = "sha256:dde4080fb6bb8db96e8c44893771bcc0d235f4c22cdddb194a765a65e3a72ba7", size = 1449178, upload-time = "2026-08-14T20:36:14.233Z" }, + { url = "https://files.pythonhosted.org/packages/70/22/98d824fe8bce808168dd364e843ccee858bb709e7e72c9b8a60f8d5201e4/chardet-7.6.0-cp310-cp310-win_amd64.whl", hash = "sha256:d6030886e7da2740bf299b6a8cc75b4dcc2c90db0ca8fe0a6e4fd0bfd071dabd", size = 1158522, upload-time = "2026-08-14T20:36:15.858Z" }, + { url = "https://files.pythonhosted.org/packages/11/2e/d8634bee23a07bf512512ddf6218a68e47f045ae061c0cc80657ed79dcc0/chardet-7.6.0-cp311-cp311-macosx_10_9_x86_64.whl", hash = "sha256:6424512f576fa7e88b7431d38a42d57552c8f717465a975fc42e497cd280d833", size = 1088243, upload-time = "2026-08-14T20:36:17.135Z" }, + { url = "https://files.pythonhosted.org/packages/55/95/bd6d59026638cec47dace85858171fbecadd2f9e58cb2b973dc515aa790c/chardet-7.6.0-cp311-cp311-macosx_11_0_arm64.whl", hash = "sha256:284136186ff90735f901ed0a1c6d41e7af67c666841cc0eceb58482a21b7056c", size = 1068688, upload-time = "2026-08-14T20:36:19.183Z" }, + { url = "https://files.pythonhosted.org/packages/7f/4a/60ed03656b28c1f4d378bc3cfe8a6cdda62c7c289398f925610fbbabfd00/chardet-7.6.0-cp311-cp311-manylinux2014_aarch64.manylinux_2_17_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:5e9b31b9ae93872d66439b046a1e08c2ea99791f3c254dce1e2633e395c5587c", size = 1487945, upload-time = "2026-08-14T20:36:20.64Z" }, + { url = "https://files.pythonhosted.org/packages/03/25/9c8db4f951e974a4db5558d9eea62e1fd5b5889c9d0ad0315eed67c5cdb3/chardet-7.6.0-cp311-cp311-manylinux2014_x86_64.manylinux_2_17_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:aa03322e07ac08d520ec50bb50c73143d0892d1adc067d4c5e58f4ef4b2363a8", size = 1510112, upload-time = "2026-08-14T20:36:22.248Z" }, + { url = "https://files.pythonhosted.org/packages/76/1b/59eb88a78d8f5855c27c25788088df82834ad067f3dddaf4d86ef03cf2d3/chardet-7.6.0-cp311-cp311-manylinux_2_31_riscv64.manylinux_2_39_riscv64.whl", hash = "sha256:0ad9bc6dab4f338673353fa3f0dc96122f559aaf746087408106e2fcbf132fe8", size = 1462470, upload-time = "2026-08-14T20:36:23.557Z" }, + { url = "https://files.pythonhosted.org/packages/02/af/46c80c317f9b4dd61f15c33b1e073fbdd64d13c70fe19e34943071dc9e50/chardet-7.6.0-cp311-cp311-win_amd64.whl", hash = "sha256:360260d074d8712ac1e9048fcafb0fdde246f9d0b12555748ad0017c5ecee43d", size = 1157092, upload-time = "2026-08-14T20:36:25.082Z" }, + { url = "https://files.pythonhosted.org/packages/6f/62/64da80dad0c804e743b4156f379183578f1e33918856ae928dc9248a6002/chardet-7.6.0-cp312-cp312-macosx_10_13_x86_64.whl", hash = "sha256:19fea52164e6e00f2a21ed418f42e4b0162a09199274c86d07ad3efd661317c4", size = 1094073, upload-time = "2026-08-14T20:36:26.53Z" }, + { url = "https://files.pythonhosted.org/packages/44/99/934fb862d102c8756008597f4398323f32cef329f16e87fbb3bf76d4f4be/chardet-7.6.0-cp312-cp312-macosx_11_0_arm64.whl", hash = "sha256:a12023d48d0e207791c01161d03cb3c0d85c6a15f345eb9d3d56063a63d1e40f", size = 1071612, upload-time = "2026-08-14T20:36:28.067Z" }, + { url = "https://files.pythonhosted.org/packages/71/e9/b04e0ec576a77e79fe37279a9a5d5b1ae752d365e43df2eca0d0eee4cea5/chardet-7.6.0-cp312-cp312-manylinux2014_aarch64.manylinux_2_17_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:249993b88ac7a58cad2781acea8f379152a28a719c9b401d614898c63a8c83da", size = 1489219, upload-time = "2026-08-14T20:36:29.357Z" }, + { url = "https://files.pythonhosted.org/packages/7d/a2/c4d99299e9ce7fad561f8bb56babbbbdd3bb6b4fbd7c0ec674c1dbdd2cc5/chardet-7.6.0-cp312-cp312-manylinux2014_x86_64.manylinux_2_17_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:2cf0adaca8b1c4bacfade9d0a1e4f8f70b1bb122833d6f07ab90e3adc84eb13a", size = 1518293, upload-time = "2026-08-14T20:36:30.879Z" }, + { url = "https://files.pythonhosted.org/packages/56/1d/49f13052b74303bab2789d098063cbd19758217949ea54ffa216b6098cb3/chardet-7.6.0-cp312-cp312-manylinux_2_31_riscv64.manylinux_2_39_riscv64.whl", hash = "sha256:cf6d08c2373b7772a558d141f9e8cee53fe1d222341bac612e4d558b04995f73", size = 1459077, upload-time = "2026-08-14T20:36:32.149Z" }, + { url = "https://files.pythonhosted.org/packages/0d/53/8da1f4758286efd8faf71356facddb382788ecf1bbd7c70d63e2e18a4898/chardet-7.6.0-cp312-cp312-win_amd64.whl", hash = "sha256:406936df1328a3284fef366eaa2bfd1cccd0ef1b10cb99781dd5b022ea644b84", size = 1160778, upload-time = "2026-08-14T20:36:33.436Z" }, + { url = "https://files.pythonhosted.org/packages/a3/29/16a7419edfbd60e901e6a797cbc3e038cb2a81903bc16c029db755f0156f/chardet-7.6.0-cp313-cp313-macosx_10_13_x86_64.whl", hash = "sha256:57e6846cc13ce1ff59979f4ec9da770c57e12aa99046073f632de5a51d9a6f20", size = 1088449, upload-time = "2026-08-14T20:36:34.723Z" }, + { url = "https://files.pythonhosted.org/packages/1d/36/3a14b0f8ddeb302f157281ca656a3ce6874b78e2d6af03682f520b487245/chardet-7.6.0-cp313-cp313-macosx_11_0_arm64.whl", hash = "sha256:089e3bb81a0a07e94f15461ded9f9ee66d349615b1a9fd557d4de1003e2fc12e", size = 1065126, upload-time = "2026-08-14T20:36:36.21Z" }, + { url = "https://files.pythonhosted.org/packages/d2/4c/f59a39c2bfe4ac99baba8da842e8d2ea0b84dff7ba53a96e7ad8c71602d7/chardet-7.6.0-cp313-cp313-manylinux2014_aarch64.manylinux_2_17_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:43ea433e43a23c55e8e17f3fad1e07f5cfe5450c73124b95b0d849c21ad379ee", size = 1482492, upload-time = "2026-08-14T20:36:37.649Z" }, + { url = "https://files.pythonhosted.org/packages/bd/eb/93e8036681157f2217a18769927a035984526e6dbd5e91f28a375ca41c14/chardet-7.6.0-cp313-cp313-manylinux2014_x86_64.manylinux_2_17_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:2b5d31f9b7f793e15e81cca877e7ccd72bffffa2a3443a9d47be9dfee84fad69", size = 1514185, upload-time = "2026-08-14T20:36:39.119Z" }, + { url = "https://files.pythonhosted.org/packages/10/04/0066d7ab2c135e404a6fa166bb7fa49d1c7bf7af07b86ed95b1c48a348c7/chardet-7.6.0-cp313-cp313-manylinux_2_31_riscv64.manylinux_2_39_riscv64.whl", hash = "sha256:c54b6a8d3b219560fa5cf4c28df932c37471afe047afdc152067104e741f38c1", size = 1454420, upload-time = "2026-08-14T20:36:40.549Z" }, + { url = "https://files.pythonhosted.org/packages/6c/9f/e965f1d9eddfb86cf118f980d329cbee43c4ac4b562448b369a6b6ef36af/chardet-7.6.0-cp313-cp313-win_amd64.whl", hash = "sha256:b3b4c96c4df93899b3c8b9e8159e06b1f55c66d7ca384d91481108e251a06eb0", size = 1159067, upload-time = "2026-08-14T20:36:41.816Z" }, + { url = "https://files.pythonhosted.org/packages/cf/78/e14991c9487277ed7d006fff58f3084ac792ed94cced081788abb95df70c/chardet-7.6.0-cp314-cp314-macosx_10_15_x86_64.whl", hash = "sha256:c6061adf247ab5dda173b67010e13904c6071717660c7c8077fb50aca362b264", size = 1087678, upload-time = "2026-08-14T20:36:43.2Z" }, + { url = "https://files.pythonhosted.org/packages/7a/40/0f95e04cb1820e0a582cd6d86bbf26be8302a94ccf330f8ba5f69735389d/chardet-7.6.0-cp314-cp314-macosx_11_0_arm64.whl", hash = "sha256:fc1e1571321baf8927582fe34363ad7f02279f11c8c2839c14b4c76894148db6", size = 1065310, upload-time = "2026-08-14T20:36:44.489Z" }, + { url = "https://files.pythonhosted.org/packages/3b/a1/04404dcc9d6e1253b02583e562d2c7ddca50a7e91f460d62eff7e1d07c92/chardet-7.6.0-cp314-cp314-manylinux2014_aarch64.manylinux_2_17_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:8900f6c7cf6b015b17a51767cc6144689059ba1cdceaa383d29eb037ac28579e", size = 1485028, upload-time = "2026-08-14T20:36:45.797Z" }, + { url = "https://files.pythonhosted.org/packages/f4/88/360064c4c7d9d0664561dae03b74c871d2f5332b329f5c99f1c997fb869a/chardet-7.6.0-cp314-cp314-manylinux2014_x86_64.manylinux_2_17_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:cedbc584789eb2edfde20fd03669972a833ce6019e60014ae613f9bfc440e8e3", size = 1510242, upload-time = "2026-08-14T20:36:47.206Z" }, + { url = "https://files.pythonhosted.org/packages/43/4c/302869fa1c69a5a41e4e78782680b12552ffd4286c3ae3c839b7b8df53d4/chardet-7.6.0-cp314-cp314-manylinux_2_31_riscv64.manylinux_2_39_riscv64.whl", hash = "sha256:d5dc835e40e0e09c2c3eab43731a8b5127834f42786dda09ba2f4b699ccd527a", size = 1456456, upload-time = "2026-08-14T20:36:48.668Z" }, + { url = "https://files.pythonhosted.org/packages/7c/a9/ff4fef15ed25fc3f945a3b981ae0f43c8559b3fbedb40267e59e583d105b/chardet-7.6.0-cp314-cp314-win_amd64.whl", hash = "sha256:0f304de7041afaec0195ad6464937cd112392002e9d72ed15d55f20a9abd3a13", size = 1159103, upload-time = "2026-08-14T20:36:50.107Z" }, + { url = "https://files.pythonhosted.org/packages/31/44/94e6f89485ce6c630e8fec6d388e3fa747e58b7246b0cc8c5c532ba98650/chardet-7.6.0-cp314-cp314t-macosx_10_15_x86_64.whl", hash = "sha256:a4f0a368ad04d5def08bdfaa17c7e15e71552f93923dc2aa9b2f7d9dee02fbb6", size = 1120302, upload-time = "2026-08-14T20:36:51.484Z" }, + { url = "https://files.pythonhosted.org/packages/e0/cd/8e68ba12f13aaf4ee82d54ef8a1f83d62942e7a6bc1e579dd2d530a3e26e/chardet-7.6.0-cp314-cp314t-macosx_11_0_arm64.whl", hash = "sha256:75d6c3a4d2046d49e83d2d2206eb073a1f390743e856d90c1bbc19949b26acf4", size = 1093157, upload-time = "2026-08-14T20:36:52.987Z" }, + { url = "https://files.pythonhosted.org/packages/f3/f6/6a35342b9efa69dcceab0ea8966571c6442a59c336bf460f0a2af95ed234/chardet-7.6.0-cp314-cp314t-manylinux2014_aarch64.manylinux_2_17_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:459e2b1c98f9a86a4698112aa42dffa802bbbff883c1ff144071f87224125862", size = 1521213, upload-time = "2026-08-14T20:36:54.812Z" }, + { url = "https://files.pythonhosted.org/packages/df/8c/8f09bfabbaedae45caa72b996e96d5e3f6436dfddc55c1311ffa2f6bd7d2/chardet-7.6.0-cp314-cp314t-manylinux2014_x86_64.manylinux_2_17_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:0bdb6f03107b7ace3f44e0edd91aa24456ee558787df265cc19daf45785b31c7", size = 1528427, upload-time = "2026-08-14T20:36:56.224Z" }, + { url = "https://files.pythonhosted.org/packages/0c/3d/2540627b193112e08b8045f3cad9295570866208555aa6625b63cc69c6cd/chardet-7.6.0-cp315-cp315-macosx_10_15_x86_64.whl", hash = "sha256:0c44a32da32cc8b23d6b20d98ace15ec7600950e4955d1bf5ab1f849b0187fdb", size = 1087374, upload-time = "2026-08-14T21:04:14.994Z" }, + { url = "https://files.pythonhosted.org/packages/fe/de/dac4f550cde73c4732b4916e72ec489ae36933fbbd1697e4122d3ae2e9dd/chardet-7.6.0-cp315-cp315-macosx_11_0_arm64.whl", hash = "sha256:7bbc8a9652c7f859c593847f220c1d264f25749369abb1a267b404ee8cceb209", size = 1064688, upload-time = "2026-08-14T21:04:16.806Z" }, + { url = "https://files.pythonhosted.org/packages/06/45/f4f3f288496797d2cf1d1e9036f920ed2b4c79787b21c42a3314f4a6d532/chardet-7.6.0-cp315-cp315-manylinux2014_aarch64.manylinux_2_17_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:83512a475a2f3886166aa0bca1bbb39343a4eb3186dd5532127d6f2591d09118", size = 1486320, upload-time = "2026-08-14T21:04:18.535Z" }, + { url = "https://files.pythonhosted.org/packages/c0/81/3ca30c16e6c6015b737fca22d9342957cc617d8a65329d3e963ad754a31b/chardet-7.6.0-cp315-cp315-manylinux2014_x86_64.manylinux_2_17_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:271ab71ec1be61dbbce0436de0848895c03eae051c379e937a39573d9ce403d9", size = 1515374, upload-time = "2026-08-14T21:04:20.724Z" }, + { url = "https://files.pythonhosted.org/packages/6e/98/163a75b8b3372a6b6503f5f5a4154a222ac135c7a706b5d176f8e4b237a2/chardet-7.6.0-cp315-cp315-manylinux_2_31_riscv64.manylinux_2_39_riscv64.whl", hash = "sha256:da86fc1b40ff5996fbb5e4c2d2dca770eac2c893cef157dacc050b8b4d929846", size = 1469504, upload-time = "2026-08-14T21:04:22.352Z" }, + { url = "https://files.pythonhosted.org/packages/d7/4f/3ad1b1bd27ae7f0d3d13cf711366525cd781a9e067950d8a09aa280916cb/chardet-7.6.0-cp315-cp315-win_amd64.whl", hash = "sha256:b73f277c1ac09c4f8076c4214b816c7aa78a0a2f0cb7156742f4303f856bedc3", size = 1158551, upload-time = "2026-08-14T21:04:24.259Z" }, + { url = "https://files.pythonhosted.org/packages/8d/92/22a609c68c5123ebdccd8fe1a80e423609012ce20166de19a2ed3955b2f7/chardet-7.6.0-cp315-cp315t-macosx_10_15_x86_64.whl", hash = "sha256:61238d5945b36af9a2ad13494f8969b7deb3c3b4abe223e54670c064e73f5328", size = 1119186, upload-time = "2026-08-14T21:04:25.761Z" }, + { url = "https://files.pythonhosted.org/packages/29/35/75a90143e4200e197f4f6cf5895d379e22bc785d0c7711120df18fa5347c/chardet-7.6.0-cp315-cp315t-macosx_11_0_arm64.whl", hash = "sha256:f2ec3c78cc6b54bf8e091ec4ee885473078b5d7ef18ab1b01c86ae1e98bf88f7", size = 1092596, upload-time = "2026-08-14T21:04:27.884Z" }, + { url = "https://files.pythonhosted.org/packages/d1/2c/b6d5f47d878c46e04ae6fcb58e0969467925bc0819b333c682e0c41bbc5c/chardet-7.6.0-cp315-cp315t-manylinux2014_aarch64.manylinux_2_17_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:167d7ba3ee08b654e36d7b43ebd9a36606c9a12e2fabdb361757a095ca3b7e3d", size = 1516644, upload-time = "2026-08-14T21:04:29.624Z" }, + { url = "https://files.pythonhosted.org/packages/25/d2/2bc3f1066c6f27bc3fa3a98dfd8a2a9c1e969a9d6bdc1edcd35278791fc4/chardet-7.6.0-cp315-cp315t-manylinux2014_x86_64.manylinux_2_17_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:f14f46ef1977e41ce1f4814ca6984cea7f8b6baf8cbc6626ef7bf3d13cf7ea13", size = 1533221, upload-time = "2026-08-14T21:04:31.39Z" }, + { url = "https://files.pythonhosted.org/packages/cf/6e/5a0b348fa4cd7847567a28c6e697ccf58391960bfd13a6e7473ee23ca2f2/chardet-7.6.0-py3-none-any.whl", hash = "sha256:4076d795897ce45239825956a1334e134322ecc4bfe84dbb12acd5390de0fbc1", size = 680279, upload-time = "2026-08-14T20:36:57.763Z" }, +] + [[package]] name = "charset-normalizer" version = "3.4.9" @@ -1199,6 +1252,36 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/9a/9a/e35b4a917281c0b8419d4207f4334c8e8c5dbf4f3f5f9ada73958d937dcc/frozenlist-1.8.0-py3-none-any.whl", hash = "sha256:0c18a16eab41e82c295618a77502e17b195883241c563b00f0aa5106fc4eaa0d", size = 13409, upload-time = "2025-10-06T05:38:16.721Z" }, ] +[[package]] +name = "gcloud-aio-auth" +version = "5.5.1" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "aiohttp" }, + { name = "chardet" }, + { name = "cryptography" }, + { name = "pyjwt" }, + { name = "tenacity" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/66/c9/cf630a761d4bbe4de88c52362b55dc29ca24735b1d893a6d6c5990879c47/gcloud_aio_auth-5.5.1.tar.gz", hash = "sha256:8039b9423445e6ad702649e2b73c1352dd380c884cf6401963c5065f583ed30c", size = 15765, upload-time = "2026-10-02T12:44:07.558Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/5d/a3/2766b5e0691743d1865aafe2323c8fd343df080597f287e52292c795c2e3/gcloud_aio_auth-5.5.1-py3-none-any.whl", hash = "sha256:3f3ca5a79c176586c142a017346b88535a5ac6ec492329f9e17334216d829894", size = 18359, upload-time = "2026-10-02T12:44:08.325Z" }, +] + +[[package]] +name = "gcloud-aio-storage" +version = "9.6.5" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "aiofiles" }, + { name = "cryptography" }, + { name = "gcloud-aio-auth" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/15/be/528ac1d1506eaf4526c2c0039d68dec99b243d50a556cd688968d1498e1a/gcloud_aio_storage-9.6.5.tar.gz", hash = "sha256:44a9529e429d7460f1a55b49a0f77906d988ee68483513abb3c0d66dedf43abd", size = 14309, upload-time = "2026-10-02T12:59:14.278Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/c3/7d/88c82ec77bbf867734011fa4292c69b11e27c823db3dfb55879e5dbd139f/gcloud_aio_storage-9.6.5-py3-none-any.whl", hash = "sha256:07bf233a065539bebbd1a34ce61505d6dba2b6674f23f8e0319606d2f01111ab", size = 17005, upload-time = "2026-10-02T12:59:15.309Z" }, +] + [[package]] name = "google-adk" version = "2.8.0" @@ -4461,6 +4544,9 @@ cloud-run-worker-otel = [ deepagents = [ { name = "temporalio-deepagents", marker = "python_full_version >= '3.11'" }, ] +gcloud-aio-storage = [ + { name = "gcloud-aio-storage" }, +] google-adk = [ { name = "temporalio-google-adk", extra = ["mcp"] }, ] @@ -4536,6 +4622,7 @@ dev = [ [package.metadata] requires-dist = [ { name = "aioboto3", marker = "extra == 'aioboto3'", specifier = ">=10.4.0" }, + { name = "gcloud-aio-storage", marker = "extra == 'gcloud-aio-storage'", specifier = ">=9.6.4,<10" }, { name = "grpcio", marker = "extra == 'grpc'", specifier = ">=1.48.2,<2" }, { name = "nexus-rpc", specifier = "==1.4.0" }, { name = "opentelemetry-api", marker = "extra == 'cloud-run-worker-otel'", specifier = ">=1.26,<2" }, @@ -4563,7 +4650,7 @@ requires-dist = [ { name = "types-protobuf", specifier = ">=3.20,<8.0.0" }, { name = "typing-extensions", specifier = ">=4.2.0,<5" }, ] -provides-extras = ["grpc", "opentelemetry", "pydantic", "openai-agents", "google-adk", "langgraph", "langsmith", "deepagents", "lambda-worker-otel", "cloud-run-worker-otel", "aioboto3", "google-genai", "strands-agents"] +provides-extras = ["grpc", "opentelemetry", "pydantic", "openai-agents", "google-adk", "langgraph", "langsmith", "deepagents", "lambda-worker-otel", "cloud-run-worker-otel", "aioboto3", "gcloud-aio-storage", "google-genai", "strands-agents"] [package.metadata.requires-dev] dev = [ From 96d214e4025aeb133d86753372d5335f75e49153 Mon Sep 17 00:00:00 2001 From: Di Wu Date: Fri, 16 Oct 2026 14:11:26 -0700 Subject: [PATCH 2/6] Clarify GCS object key encoding --- temporalio/contrib/gcp/gcsdriver/_driver.py | 11 +++++------ 1 file changed, 5 insertions(+), 6 deletions(-) diff --git a/temporalio/contrib/gcp/gcsdriver/_driver.py b/temporalio/contrib/gcp/gcsdriver/_driver.py index e699ed46aa..bbcdbb6001 100644 --- a/temporalio/contrib/gcp/gcsdriver/_driver.py +++ b/temporalio/contrib/gcp/gcsdriver/_driver.py @@ -25,15 +25,14 @@ _T = TypeVar("_T") -# S3's safe character set for object key names. See -# https://docs.aws.amazon.com/AmazonS3/latest/userguide/object-keys.html -_S3_SAFE_CHARS = frozenset(string.ascii_letters + string.digits + "!-_.*'()") +# Match the Python S3 driver's segment encoding for consistent storage keys. +_KEY_SAFE_CHARS = frozenset(string.ascii_letters + string.digits + "!-_.*'()") def _percent_encode(val: str | None) -> str | None: - """Percent-encode a path segment per the S3 key spec. + """Percent-encode a segment of a GCS object name. - Encodes every byte outside :data:`_S3_SAFE_CHARS` as ``%XX`` with + Encodes every byte outside :data:`_KEY_SAFE_CHARS` as ``%XX`` with upper-case hex digits, operating on the UTF-8 bytes so non-ASCII characters are escaped byte by byte. Returns ``None`` for empty or missing values so callers can substitute ``"null"``. @@ -41,7 +40,7 @@ def _percent_encode(val: str | None) -> str | None: if not val: return None return "".join( - chr(byte) if chr(byte) in _S3_SAFE_CHARS else f"%{byte:02X}" + chr(byte) if chr(byte) in _KEY_SAFE_CHARS else f"%{byte:02X}" for byte in val.encode("utf-8") ) From c9c127de4ad4a698154c2955d5cee7df0c4dfa73 Mon Sep 17 00:00:00 2001 From: Di Wu Date: Fri, 16 Oct 2026 14:12:15 -0700 Subject: [PATCH 3/6] Remove S3 comparison from GCS encoding --- temporalio/contrib/gcp/gcsdriver/_driver.py | 1 - 1 file changed, 1 deletion(-) diff --git a/temporalio/contrib/gcp/gcsdriver/_driver.py b/temporalio/contrib/gcp/gcsdriver/_driver.py index bbcdbb6001..c5f39fa459 100644 --- a/temporalio/contrib/gcp/gcsdriver/_driver.py +++ b/temporalio/contrib/gcp/gcsdriver/_driver.py @@ -25,7 +25,6 @@ _T = TypeVar("_T") -# Match the Python S3 driver's segment encoding for consistent storage keys. _KEY_SAFE_CHARS = frozenset(string.ascii_letters + string.digits + "!-_.*'()") From b8b136644df900415b9085052ce53352357ef488 Mon Sep 17 00:00:00 2001 From: Di Wu Date: Fri, 16 Oct 2026 14:19:55 -0700 Subject: [PATCH 4/6] Handle expired GCS resumable upload sessions --- temporalio/contrib/gcp/gcsdriver/gcloud_aio.py | 8 ++++++-- tests/contrib/gcp/gcsdriver/test_gcsdriver.py | 15 +++++++++++++-- .../gcp/gcsdriver/test_gcsdriver_integration.py | 5 +++-- 3 files changed, 22 insertions(+), 6 deletions(-) diff --git a/temporalio/contrib/gcp/gcsdriver/gcloud_aio.py b/temporalio/contrib/gcp/gcsdriver/gcloud_aio.py index 2ca4ac7842..78eef82ee6 100644 --- a/temporalio/contrib/gcp/gcsdriver/gcloud_aio.py +++ b/temporalio/contrib/gcp/gcsdriver/gcloud_aio.py @@ -41,8 +41,12 @@ async def put_object(self, *, bucket: str, key: str, data: bytes) -> None: ) except ClientResponseError as err: # Another writer may create the content-addressed object after our check. - if err.status != 412: - raise + if err.status == 412: + return + # Retrying a resumable upload after a 412 can report a missing session. + if err.status == 404 and await self.object_exists(bucket=bucket, key=key): + return + raise async def get_object(self, *, bucket: str, key: str) -> bytes: """Download the object's bytes.""" diff --git a/tests/contrib/gcp/gcsdriver/test_gcsdriver.py b/tests/contrib/gcp/gcsdriver/test_gcsdriver.py index 8d05499f80..02299009f6 100644 --- a/tests/contrib/gcp/gcsdriver/test_gcsdriver.py +++ b/tests/contrib/gcp/gcsdriver/test_gcsdriver.py @@ -331,13 +331,14 @@ async def test_adapter_object_exists(status: int | None) -> None: storage.download_metadata.assert_awaited_once_with("bucket", "key") -@pytest.mark.parametrize("status", [None, 412, 403, 500]) +@pytest.mark.parametrize("status", [None, 412, 403, 404, 500]) async def test_adapter_conditional_upload(status: int | None) -> None: storage = AsyncMock(spec=Storage) if status: storage.upload.side_effect = response_error(status) + storage.download_metadata.side_effect = response_error(404) client = new_gcloud_aio_client(storage) - if status in (403, 500): + if status in (403, 404, 500): with pytest.raises(ClientResponseError): await client.put_object(bucket="bucket", key="key", data=b"data") else: @@ -351,6 +352,16 @@ async def test_adapter_conditional_upload(status: int | None) -> None: ) +async def test_adapter_resumable_conflict_with_missing_session() -> None: + storage = AsyncMock(spec=Storage) + storage.upload.side_effect = response_error(404) + storage.download_metadata.return_value = {"name": "key"} + await new_gcloud_aio_client(storage).put_object( + bucket="bucket", key="key", data=b"data" + ) + storage.download_metadata.assert_awaited_once_with("bucket", "key") + + async def test_adapter_download() -> None: storage = AsyncMock(spec=Storage) storage.download.return_value = b"\x00\xff" diff --git a/tests/contrib/gcp/gcsdriver/test_gcsdriver_integration.py b/tests/contrib/gcp/gcsdriver/test_gcsdriver_integration.py index e01c5a0eb9..839d623cff 100644 --- a/tests/contrib/gcp/gcsdriver/test_gcsdriver_integration.py +++ b/tests/contrib/gcp/gcsdriver/test_gcsdriver_integration.py @@ -41,9 +41,10 @@ async def test_gcs_emulator_round_trip( assert await driver.store(context, [original]) == claims assert await driver.retrieve(StorageDriverRetrieveContext(), claims) == [original] key = claims[0].claim_data["key"] + replacement = b"must not overwrite" + original.data await asyncio.gather( - client.put_object(bucket=gcs_bucket, key=key, data=b"must not overwrite"), - client.put_object(bucket=gcs_bucket, key=key, data=b"must not overwrite"), + client.put_object(bucket=gcs_bucket, key=key, data=replacement), + client.put_object(bucket=gcs_bucket, key=key, data=replacement), ) assert ( await client.get_object(bucket=gcs_bucket, key=key) From 7436c065e9e4a870c8dfe883b62369c8855b4bdc Mon Sep 17 00:00:00 2001 From: Di Wu Date: Fri, 16 Oct 2026 14:32:03 -0700 Subject: [PATCH 5/6] Use single-request GCS uploads --- temporalio/contrib/gcp/gcsdriver/README.md | 4 +++- temporalio/contrib/gcp/gcsdriver/gcloud_aio.py | 4 +--- tests/contrib/gcp/gcsdriver/test_gcsdriver.py | 12 +++++++----- 3 files changed, 11 insertions(+), 9 deletions(-) diff --git a/temporalio/contrib/gcp/gcsdriver/README.md b/temporalio/contrib/gcp/gcsdriver/README.md index b61c10f063..0cf6cba620 100644 --- a/temporalio/contrib/gcp/gcsdriver/README.md +++ b/temporalio/contrib/gcp/gcsdriver/README.md @@ -40,7 +40,9 @@ The application owns the `Storage` client and configures credentials, request timeouts, and retry behavior. See the [gcloud-aio documentation](https://talkiq.github.io/gcloud-aio/) for authentication and customization. The library has no general retry policy; applications can supply a `Storage` subclass with async retries and timeouts. -The adapter preserves the client's resumable upload handling. +The adapter uses a single request for each upload, including payloads over 5 MiB. +Interrupted uploads must restart from the beginning. Conditional creation prevents +retries from overwriting existing objects. ## Configuration diff --git a/temporalio/contrib/gcp/gcsdriver/gcloud_aio.py b/temporalio/contrib/gcp/gcsdriver/gcloud_aio.py index 78eef82ee6..8364703d07 100644 --- a/temporalio/contrib/gcp/gcsdriver/gcloud_aio.py +++ b/temporalio/contrib/gcp/gcsdriver/gcloud_aio.py @@ -38,14 +38,12 @@ async def put_object(self, *, bucket: str, key: str, data: bytes) -> None: data, content_type="application/octet-stream", parameters={"ifGenerationMatch": "0"}, + force_resumable_upload=False, ) except ClientResponseError as err: # Another writer may create the content-addressed object after our check. if err.status == 412: return - # Retrying a resumable upload after a 412 can report a missing session. - if err.status == 404 and await self.object_exists(bucket=bucket, key=key): - return raise async def get_object(self, *, bucket: str, key: str) -> bytes: diff --git a/tests/contrib/gcp/gcsdriver/test_gcsdriver.py b/tests/contrib/gcp/gcsdriver/test_gcsdriver.py index 02299009f6..23db018505 100644 --- a/tests/contrib/gcp/gcsdriver/test_gcsdriver.py +++ b/tests/contrib/gcp/gcsdriver/test_gcsdriver.py @@ -349,17 +349,19 @@ async def test_adapter_conditional_upload(status: int | None) -> None: b"data", content_type="application/octet-stream", parameters={"ifGenerationMatch": "0"}, + force_resumable_upload=False, ) -async def test_adapter_resumable_conflict_with_missing_session() -> None: +async def test_adapter_upload_404_propagates_when_object_exists() -> None: storage = AsyncMock(spec=Storage) storage.upload.side_effect = response_error(404) storage.download_metadata.return_value = {"name": "key"} - await new_gcloud_aio_client(storage).put_object( - bucket="bucket", key="key", data=b"data" - ) - storage.download_metadata.assert_awaited_once_with("bucket", "key") + with pytest.raises(ClientResponseError): + await new_gcloud_aio_client(storage).put_object( + bucket="bucket", key="key", data=b"data" + ) + storage.download_metadata.assert_not_awaited() async def test_adapter_download() -> None: From 17e860ed8e65b08913d3e124baffb84f8761e0dd Mon Sep 17 00:00:00 2001 From: Di Wu Date: Fri, 16 Oct 2026 15:48:39 -0700 Subject: [PATCH 6/6] Simplify GCS adapter tests --- tests/contrib/gcp/gcsdriver/test_gcsdriver.py | 15 ++------------- 1 file changed, 2 insertions(+), 13 deletions(-) diff --git a/tests/contrib/gcp/gcsdriver/test_gcsdriver.py b/tests/contrib/gcp/gcsdriver/test_gcsdriver.py index 23db018505..94271c7123 100644 --- a/tests/contrib/gcp/gcsdriver/test_gcsdriver.py +++ b/tests/contrib/gcp/gcsdriver/test_gcsdriver.py @@ -217,7 +217,7 @@ async def test_batch_cancellation(retrieving: bool, cancel_parent: bool) -> None started = 0 cancelled = 0 - async def transfer(**_kwargs: object) -> bytes: + async def transfer(**_kwargs: object) -> None: nonlocal started, cancelled started += 1 index = started @@ -231,7 +231,6 @@ async def transfer(**_kwargs: object) -> bytes: except asyncio.CancelledError: cancelled += 1 raise - return b"" client.get_object.side_effect = transfer client.put_object.side_effect = transfer @@ -336,7 +335,7 @@ async def test_adapter_conditional_upload(status: int | None) -> None: storage = AsyncMock(spec=Storage) if status: storage.upload.side_effect = response_error(status) - storage.download_metadata.side_effect = response_error(404) + storage.download_metadata.return_value = {"name": "key"} client = new_gcloud_aio_client(storage) if status in (403, 404, 500): with pytest.raises(ClientResponseError): @@ -351,16 +350,6 @@ async def test_adapter_conditional_upload(status: int | None) -> None: parameters={"ifGenerationMatch": "0"}, force_resumable_upload=False, ) - - -async def test_adapter_upload_404_propagates_when_object_exists() -> None: - storage = AsyncMock(spec=Storage) - storage.upload.side_effect = response_error(404) - storage.download_metadata.return_value = {"name": "key"} - with pytest.raises(ClientResponseError): - await new_gcloud_aio_client(storage).put_object( - bucket="bucket", key="key", data=b"data" - ) storage.download_metadata.assert_not_awaited()