diff --git a/dapr/ext/workflow/AGENTS.md b/dapr/ext/workflow/AGENTS.md index 56d2aa28c..e6fabc41e 100644 --- a/dapr/ext/workflow/AGENTS.md +++ b/dapr/ext/workflow/AGENTS.md @@ -168,6 +168,7 @@ Passed to workflow functions as the first argument: - `create_timer(fire_at)` → `Task` (accepts `datetime` or `timedelta`) - `wait_for_external_event(name)` → `Task` - `set_custom_status(status)` / `continue_as_new(new_input, *, save_events)` +- `new_guid()` → `uuid.UUID` — deterministic across workflow replays Module-level functions: - `when_all(tasks)` → `WhenAllTask` — wait for all tasks to complete diff --git a/dapr/ext/workflow/_durabletask/task.py b/dapr/ext/workflow/_durabletask/task.py index e3a190c41..6a3dc7451 100644 --- a/dapr/ext/workflow/_durabletask/task.py +++ b/dapr/ext/workflow/_durabletask/task.py @@ -13,6 +13,7 @@ from __future__ import annotations import math +import uuid from abc import ABC, abstractmethod from datetime import datetime, timedelta from typing import Any, Callable, Generator, Generic, Optional, TypeVar, Union @@ -89,6 +90,17 @@ def set_custom_status(self, custom_status: str) -> None: """ pass + @abstractmethod + def new_guid(self) -> uuid.UUID: + """Create a new GUID that is safe for replay within an orchestration. + + Returns + ------- + uuid.UUID + A new, deterministically generated GUID. + """ + pass + @abstractmethod def create_timer(self, fire_at: Union[datetime, timedelta]) -> Task: """Create a Timer Task to fire after at the specified deadline. diff --git a/dapr/ext/workflow/_durabletask/worker.py b/dapr/ext/workflow/_durabletask/worker.py index 703e8937e..e357fe88c 100644 --- a/dapr/ext/workflow/_durabletask/worker.py +++ b/dapr/ext/workflow/_durabletask/worker.py @@ -17,6 +17,7 @@ import random import threading import time +import uuid import warnings from concurrent.futures import ThreadPoolExecutor from datetime import datetime, timedelta, timezone @@ -1729,6 +1730,12 @@ def current_utc_datetime(self, value: datetime): def is_replaying(self) -> bool: return self._is_replaying + def new_guid(self) -> uuid.UUID: + # Call the mixin implementation directly because ``super().new_guid()`` + # first resolves to the abstract method ``OrchestrationContext.new_guid()``, + # which returns ``None``. + return deterministic.DeterministicContextMixin.new_guid(self) + def set_custom_status(self, custom_status: str) -> None: if custom_status is not None and not isinstance(custom_status, str): warnings.warn( diff --git a/dapr/ext/workflow/dapr_workflow_context.py b/dapr/ext/workflow/dapr_workflow_context.py index 7ceee0cbf..16e8050ce 100644 --- a/dapr/ext/workflow/dapr_workflow_context.py +++ b/dapr/ext/workflow/dapr_workflow_context.py @@ -15,6 +15,7 @@ from datetime import datetime, timedelta from typing import Any, Callable, List, Optional, TypeVar, Union +from uuid import UUID from dapr.ext.workflow._durabletask import task from dapr.ext.workflow.logger import Logger, LoggerOptions @@ -57,6 +58,10 @@ def set_custom_status(self, custom_status: str) -> None: self._logger.debug(f'{self.instance_id}: Setting custom status to {custom_status}') self.__obj.set_custom_status(custom_status) + def new_guid(self) -> UUID: + self._logger.debug(f'{self.instance_id}: Generating new GUID') + return self.__obj.new_guid() + def create_timer(self, fire_at: Union[datetime, timedelta]) -> task.Task: self._logger.debug(f'{self.instance_id}: Creating timer to fire at {fire_at} time') return self.__obj.create_timer(fire_at) diff --git a/dapr/ext/workflow/workflow_context.py b/dapr/ext/workflow/workflow_context.py index e3e98fe9e..956bac326 100644 --- a/dapr/ext/workflow/workflow_context.py +++ b/dapr/ext/workflow/workflow_context.py @@ -18,6 +18,7 @@ from abc import ABC, abstractmethod from datetime import datetime, timedelta from typing import Any, Callable, Generator, Optional, TypeVar, Union +from uuid import UUID from dapr.ext.workflow._durabletask import task from dapr.ext.workflow.propagation import PropagatedHistory, PropagationScope @@ -90,6 +91,22 @@ def set_custom_status(self, custom_status: str) -> None: """Set the custom status.""" pass + @abstractmethod + def new_guid(self) -> UUID: + """Create a new GUID that is safe for replay within a workflow. + + The GUID is deterministically derived from the workflow instance ID, + the current replay-safe time, and a counter that increments on each + call, so calling this repeatedly returns different values within a + single execution while remaining stable across replays. + + Returns + ------- + uuid.UUID + A replay-safe, deterministically generated GUID. + """ + pass + @abstractmethod def create_timer(self, fire_at: Union[datetime, timedelta]) -> task.Task: """Create a Timer Task to fire after at the specified deadline. diff --git a/tests/ext/workflow/test_dapr_workflow_context.py b/tests/ext/workflow/test_dapr_workflow_context.py index 22350dda8..0cf25cede 100644 --- a/tests/ext/workflow/test_dapr_workflow_context.py +++ b/tests/ext/workflow/test_dapr_workflow_context.py @@ -14,6 +14,7 @@ """ import unittest +import uuid from datetime import datetime from unittest import mock @@ -120,3 +121,15 @@ def test_get_propagated_history_returns_none_when_not_set(self): fake = worker._RuntimeOrchestrationContext(mock_instance_id) dapr_wf_ctx = DaprWorkflowContext(fake) assert dapr_wf_ctx.get_propagated_history() is None + + def test_new_guid_is_unique_within_execution_and_deterministic_across_replays(self): + instance_id = 'abc' + ctx1 = DaprWorkflowContext(worker._RuntimeOrchestrationContext(instance_id)) + ctx2 = DaprWorkflowContext(worker._RuntimeOrchestrationContext(instance_id)) + guid_count = 3 + g1 = [ctx1.new_guid() for _ in range(guid_count)] + g2 = [ctx2.new_guid() for _ in range(guid_count)] + + assert all(isinstance(g, uuid.UUID) for g in g1) + assert len(set(g1)) == guid_count # unique within an execution + assert g1 == g2 # deterministic across replays diff --git a/tests/integration/test_workflow_replay_safety.py b/tests/integration/test_workflow_replay_safety.py new file mode 100644 index 000000000..ab6e97fa4 --- /dev/null +++ b/tests/integration/test_workflow_replay_safety.py @@ -0,0 +1,127 @@ +# Copyright 2026 The Dapr Authors +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# http://www.apache.org/licenses/LICENSE-2.0 +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""End-to-end verification of replay-safe workflow context behavior. + +These tests run workflows through a real Dapr sidecar. After an activity +completes, the workflow is resumed by replaying its history, so the replay +assertions exercise ``DaprWorkflowContext`` rather than only the underlying +deterministic context. +""" + +from __future__ import annotations + +import json +from typing import Any +from uuid import UUID + +import pytest + +HOST = '127.0.0.1' +GRPC_PORT = '13501' +COMPLETION_TIMEOUT_S = 60 +GUID_COUNT = 3 + + +@pytest.fixture(scope='module') +def sidecar(dapr_env): + """Start the Dapr sidecar used by the workflow replay tests.""" + return dapr_env.start_sidecar(app_id='test-workflow-replay-safety') + + +def _run_workflow(workflow, *, activities: tuple[Any, ...] = ()) -> dict[str, Any] | list[str]: + """Run a workflow to completion through a real Dapr sidecar and decode its JSON output.""" + from dapr.ext.workflow import DaprWorkflowClient, WorkflowRuntime + + runtime = WorkflowRuntime(host=HOST, port=GRPC_PORT) + runtime.register_workflow(workflow) + for activity in activities: + runtime.register_activity(activity) + + client = DaprWorkflowClient(host=HOST, port=GRPC_PORT) + runtime.start() + try: + instance_id = client.schedule_new_workflow(workflow) + state = client.wait_for_workflow_completion( + instance_id, timeout_in_seconds=COMPLETION_TIMEOUT_S + ) + finally: + client.close() + runtime.shutdown() + + assert state is not None, 'workflow did not reach a terminal state' + assert state.runtime_status.name == 'COMPLETED' + return json.loads(state.serialized_output) + + +def test_new_guid_generates_unique_guids_within_execution(sidecar): + """Each call to ``new_guid()`` produces a unique GUID within one execution.""" + from dapr.ext.workflow import DaprWorkflowContext, WorkflowActivityContext + + def echo_guids(ctx: WorkflowActivityContext, values: list[str]) -> list[str]: + return values + + def generate_guids(ctx: DaprWorkflowContext, _input: None): + generated_guids = [str(ctx.new_guid()) for _ in range(GUID_COUNT)] + yield ctx.call_activity(echo_guids, input=generated_guids) + return generated_guids + + generated_guids = _run_workflow(generate_guids, activities=(echo_guids,)) + + assert isinstance(generated_guids, list) + + # Validate serialized GUIDs + for value in generated_guids: + UUID(value) + + assert len(set(generated_guids)) == len(generated_guids) + + +def test_new_guid_is_deterministic_across_replays(sidecar): + """GUID sequences remain unchanged when the workflow replays its history. + + The echo activity returns the GUIDS from the initial execution. + During replay, that result comes from history instead of executing the + activity again, so comparing the generated and echoed values detects if replay + produced different GUIDs than the initial execution. + """ + from dapr.ext.workflow import DaprWorkflowContext, WorkflowActivityContext + + def echo_guids(ctx: WorkflowActivityContext, values: list[str]) -> list[str]: + return values + + def generate_guids(ctx: DaprWorkflowContext, _input: None): + first_guids = [str(ctx.new_guid()) for _ in range(GUID_COUNT)] + first_echo = yield ctx.call_activity(echo_guids, input=first_guids) + + second_guids = [str(ctx.new_guid()) for _ in range(GUID_COUNT)] + second_echo = yield ctx.call_activity(echo_guids, input=second_guids) + + return { + 'first_guids': first_guids, + 'first_echo': first_echo, + 'second_guids': second_guids, + 'second_echo': second_echo, + } + + result = _run_workflow(generate_guids, activities=(echo_guids,)) + + assert isinstance(result, dict) + assert result['first_guids'] == result['first_echo'] + assert result['second_guids'] == result['second_echo'] + + # Validate serialized GUIDs + for values in (result['first_guids'], result['second_guids']): + for value in values: + UUID(value) + + all_guids = result['first_guids'] + result['second_guids'] + assert len(set(all_guids)) == len(all_guids)