Merge branch 'main' into chore/update-uv-lock-deps

Author: tconley1428

temporalio/client.py

@@ -23,8 +23,9 @@ from temporalio.contrib.aws.s3driver.aioboto3 import new_aioboto3_client
 from temporalio.converter import DataConverter, ExternalStorage
 
 session = aioboto3.Session()
-# Credentials and region are resolved automatically from the standard AWS credential
-# chain e.g. environment variables, ~/.aws/config, IAM instance profile, and so on.
+# To see how to set credentials and region via environment, config objects, or configuration files, 
+# see:
+# https://docs.aws.amazon.com/boto3/latest/guide/configuration.html
 async with session.client("s3") as s3_client:
     driver = S3StorageDriver(
         client=new_aioboto3_client(s3_client),

temporalio/contrib/aws/s3driver/README.md

@@ -7,6 +7,7 @@
 from __future__ import annotations
 
 from abc import ABC, abstractmethod
+from collections.abc import Mapping
 
 
 class S3StorageDriverClient(ABC):
@@ -30,3 +31,11 @@ async def object_exists(self, *, bucket: str, key: str) -> bool:
     @abstractmethod
     async def get_object(self, *, bucket: str, key: str) -> bytes:
         """Download and return the bytes stored at the given S3 *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 {}

temporalio/contrib/aws/s3driver/_client.py

@@ -26,6 +26,20 @@
 _T = TypeVar("_T")
 
 
+def _format_client_context(client: S3StorageDriverClient) -> 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]:
@@ -156,7 +170,8 @@ async def _upload(payload: Payload) -> StorageDriverClaim:
                     )
             except Exception as e:
                 raise RuntimeError(
-                    f"S3StorageDriver store failed [bucket={bucket}, key={key}]"
+                    f"S3StorageDriver store failed [bucket={bucket}, key={key}"
+                    f"{_format_client_context(self._client)}]"
                 ) from e
 
             return StorageDriverClaim(
@@ -185,7 +200,8 @@ async def _download(claim: StorageDriverClaim) -> Payload:
                 payload_bytes = await self._client.get_object(bucket=bucket, key=key)
             except Exception as e:
                 raise RuntimeError(
-                    f"S3StorageDriver retrieve failed [bucket={bucket}, key={key}]"
+                    f"S3StorageDriver retrieve failed [bucket={bucket}, key={key}"
+                    f"{_format_client_context(self._client)}]"
                 ) from e
 
             hash_algorithm = claim.claim_data.get("hash_algorithm")

temporalio/contrib/aws/s3driver/_driver.py

@@ -7,6 +7,7 @@
 from __future__ import annotations
 
 import io
+from collections.abc import Mapping
 
 from botocore.exceptions import ClientError
 from types_aiobotocore_s3.client import S3Client
@@ -34,6 +35,13 @@ def __init__(self, client: S3Client) -> None:
         """
         self._client = client
 
+    def describe(self) -> Mapping[str, str]:
+        """Region of the wrapped aioboto3 client, surfaced in driver error
+        messages to short-circuit the most common silent 403 misconfiguration.
+        """
+        region = self._client.meta.region_name
+        return {"region": region} if region else {}
+
     async def object_exists(self, *, bucket: str, key: str) -> bool:
         """Check existence via aioboto3's ``head_object``."""
         try:

temporalio/contrib/aws/s3driver/aioboto3.py

@@ -26,6 +26,8 @@
     S3StorageDriver,
     S3StorageDriverClient,
 )
+from temporalio.contrib.aws.s3driver._driver import _format_client_context
+from temporalio.contrib.aws.s3driver.aioboto3 import _Aioboto3StorageDriverClient
 from temporalio.converter import (
     JSONPlainPayloadConverter,
     StorageDriverActivityInfo,
@@ -34,7 +36,7 @@
     StorageDriverStoreContext,
     StorageDriverWorkflowInfo,
 )
-from tests.contrib.aws.s3driver.conftest import BUCKET
+from tests.contrib.aws.s3driver.conftest import BUCKET, REGION
 
 _CONVERTER = JSONPlainPayloadConverter()
 
@@ -618,7 +620,7 @@ async def test_store_nonexistent_bucket_raises(
             await driver.store(make_store_context(), [payload])
         assert (
             str(exc_info.value)
-            == f"S3StorageDriver store failed [bucket={bucket}, key={expected_key}]"
+            == f"S3StorageDriver store failed [bucket={bucket}, key={expected_key}, region={REGION}]"
         )
         assert isinstance(exc_info.value.__cause__, ClientError)
         assert (
@@ -636,7 +638,7 @@ async def test_retrieve_nonexistent_key_raises(
             await driver.retrieve(StorageDriverRetrieveContext(), [claim])
         assert (
             str(exc_info.value)
-            == f"S3StorageDriver retrieve failed [bucket={BUCKET}, key={key}]"
+            == f"S3StorageDriver retrieve failed [bucket={BUCKET}, key={key}, region={REGION}]"
         )
         assert isinstance(exc_info.value.__cause__, ClientError)
         assert (
@@ -655,7 +657,7 @@ async def test_retrieve_nonexistent_bucket_raises(
             await driver.retrieve(StorageDriverRetrieveContext(), [claim])
         assert (
             str(exc_info.value)
-            == f"S3StorageDriver retrieve failed [bucket={bucket}, key={key}]"
+            == f"S3StorageDriver retrieve failed [bucket={bucket}, key={key}, region={REGION}]"
         )
         assert isinstance(exc_info.value.__cause__, ClientError)
         assert (
@@ -839,3 +841,49 @@ async def test_retrieve_cancels_remaining_on_failure(
         assert (
             len(faulty_client.cancelled) == 2
         ), "Expected 2 remaining tasks to be cancelled"
+
+
+# ---------------------------------------------------------------------------
+# TestAioboto3StorageDriverClientDescribe
+# ---------------------------------------------------------------------------
+
+
+class TestAioboto3StorageDriverClientDescribe:
+    def _make_client(self, region: str | None) -> _Aioboto3StorageDriverClient:
+        mock_s3 = MagicMock()
+        mock_s3.meta.region_name = region
+        return _Aioboto3StorageDriverClient(mock_s3)
+
+    def test_returns_region(self) -> None:
+        client = self._make_client(region="ap-southeast-1")
+        assert client.describe() == {"region": "ap-southeast-1"}
+
+    def test_omits_region_when_none(self) -> None:
+        client = self._make_client(region=None)
+        assert client.describe() == {}
+
+    def test_omits_region_when_empty_string(self) -> None:
+        client = self._make_client(region="")
+        assert client.describe() == {}
+
+
+# ---------------------------------------------------------------------------
+# TestFormatClientContext
+# ---------------------------------------------------------------------------
+
+
+class TestFormatClientContext:
+    def test_formats_entry(self) -> None:
+        client = MagicMock(spec=S3StorageDriverClient)
+        client.describe.return_value = {"region": "us-east-1"}
+        assert _format_client_context(client) == ", region=us-east-1"
+
+    def test_returns_empty_string_for_empty_describe(self) -> None:
+        client = MagicMock(spec=S3StorageDriverClient)
+        client.describe.return_value = {}
+        assert _format_client_context(client) == ""
+
+    def test_returns_empty_string_when_describe_raises(self) -> None:
+        client = MagicMock(spec=S3StorageDriverClient)
+        client.describe.side_effect = RuntimeError("oops")
+        assert _format_client_context(client) == ""

tests/contrib/aws/s3driver/test_s3driver.py

@@ -506,4 +506,4 @@ async def test_s3_store_failure_surfaces_in_workflow_history(
     msg = app_error.message
     assert f"S3StorageDriver store failed [bucket={bad_bucket}, key=" in msg
     assert f"/wt/LargeOutputNoRetryWorkflow/wi/{workflow_id}/ri/" in msg
-    assert f"/d/sha256/{expected_hash}]" in msg
+    assert f"/d/sha256/{expected_hash}, region={REGION}]" in msg

tests/contrib/aws/s3driver/test_s3driver_worker.py

@@ -220,20 +220,23 @@ async def test_start_activity_calls_interceptor(
 
     activity_id = str(uuid.uuid4())
     task_queue = str(uuid.uuid4())
+    start_delay = timedelta(seconds=3)
 
     await intercepted_client.start_activity(
         increment,
         args=(1,),
         id=activity_id,
         task_queue=task_queue,
         start_to_close_timeout=timedelta(seconds=5),
+        start_delay=start_delay,
     )
 
     assert len(interceptor.start_activity_calls) == 1
     call = interceptor.start_activity_calls[0]
     assert call.id == activity_id
     assert call.task_queue == task_queue
     assert call.activity_type == "increment"
+    assert call.start_delay == start_delay
 
 
 async def test_describe_activity_calls_interceptor(
@@ -413,6 +416,18 @@ async def test_count_activities_calls_interceptor(
     assert call.query == query
 
 
+async def test_start_activity_rejects_negative_start_delay(client: Client):
+    with pytest.raises(ValueError, match="start_delay must be non-negative"):
+        await client.start_activity(
+            increment,
+            args=(1,),
+            id=str(uuid.uuid4()),
+            task_queue=str(uuid.uuid4()),
+            start_to_close_timeout=timedelta(seconds=5),
+            start_delay=timedelta(seconds=-1),
+        )
+
+
 async def test_get_result(client: Client, env: WorkflowEnvironment):
     if env.supports_time_skipping:
         pytest.skip(