diff --git a/CHANGELOG.md b/CHANGELOG.md index e2ce1f736..1c5ab1a90 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -55,6 +55,7 @@ to include examples, links to docs, or any other relevant information. ### Fixed +- Standalone activity start requests now include a unique request ID so RPC retries are deduplicated. - OpenTelemetry trace and span IDs propagated by concurrent workers no longer interfere with each other, preserving the correct parent-child hierarchy. - The `google-adk` extra now depends on `mcp`, so fresh installs of diff --git a/temporalio/client/_impl.py b/temporalio/client/_impl.py index 9595351f2..2799deb27 100644 --- a/temporalio/client/_impl.py +++ b/temporalio/client/_impl.py @@ -627,6 +627,7 @@ async def _build_start_activity_execution_request( req = temporalio.api.workflowservice.v1.StartActivityExecutionRequest( namespace=self._client.namespace, identity=self._client.identity, + request_id=str(uuid.uuid4()), activity_id=input.id, activity_type=temporalio.api.common.v1.ActivityType( name=input.activity_type @@ -677,8 +678,8 @@ async def _build_start_activity_execution_request( # Set priority req.priority.CopyFrom(input.priority._to_proto()) - # Add request_id, links, and completion callbacks from the Nexus context - # if not in a Nexus context, this is a no-op + # Nexus starts use the inbound request ID so retries across the Nexus + # boundary resolve to the same activity execution. temporalio.nexus._operation_context._apply_nexus_context_to_start_activity_request( req ) diff --git a/tests/test_activity.py b/tests/test_activity.py index 874e2b620..3a63f51ac 100644 --- a/tests/test_activity.py +++ b/tests/test_activity.py @@ -2,9 +2,11 @@ import uuid from dataclasses import dataclass from datetime import timedelta +from unittest import mock import pytest +import temporalio.api.workflowservice.v1 from temporalio import activity, workflow from temporalio.client import ( ActivityExecutionCount, @@ -78,6 +80,31 @@ def sync_increment(self, x: int) -> int: return x + 1 +async def test_start_activity_generates_request_id() -> None: + start_activity_execution = mock.AsyncMock( + return_value=temporalio.api.workflowservice.v1.StartActivityExecutionResponse( + run_id="activity-run-id" + ) + ) + service_client = mock.MagicMock() + service_client.config.identity = "test-identity" + service_client.workflow_service.start_activity_execution = start_activity_execution + client = Client(service_client) + + for activity_id in ("activity-id-1", "activity-id-2"): + await client.start_activity( + increment, + 1, + id=activity_id, + task_queue="task-queue", + start_to_close_timeout=timedelta(seconds=1), + ) + + requests = [call.args[0] for call in start_activity_execution.call_args_list] + assert all(uuid.UUID(request.request_id).version == 4 for request in requests) + assert requests[0].request_id != requests[1].request_id + + class TestDescribe: @pytest.fixture async def activity_handle(self, client: Client, env: WorkflowEnvironment):