From acad1124b30c862b52a8f2a025a24339f49b2980 Mon Sep 17 00:00:00 2001 From: Albert Callarisa Date: Mon, 27 Jul 2026 15:41:35 +0200 Subject: [PATCH] Implements detached workflows from a workflow contexts Signed-off-by: Albert Callarisa --- dapr/ext/workflow/AGENTS.md | 12 + .../workflow/_durabletask/internal/helpers.py | 42 +++ dapr/ext/workflow/_durabletask/task.py | 49 ++++ dapr/ext/workflow/_durabletask/worker.py | 79 +++++- dapr/ext/workflow/dapr_workflow_context.py | 32 +++ dapr/ext/workflow/workflow_context.py | 45 +++ examples/workflow/detached.py | 78 ++++++ .../durabletask/test_detached_workflow.py | 262 ++++++++++++++++++ .../workflow/test_dapr_workflow_context.py | 68 +++++ 9 files changed, 666 insertions(+), 1 deletion(-) create mode 100644 examples/workflow/detached.py create mode 100644 tests/ext/workflow/durabletask/test_detached_workflow.py diff --git a/dapr/ext/workflow/AGENTS.md b/dapr/ext/workflow/AGENTS.md index 4874a576c..7ead627f9 100644 --- a/dapr/ext/workflow/AGENTS.md +++ b/dapr/ext/workflow/AGENTS.md @@ -150,10 +150,22 @@ Passed to workflow functions as the first argument: - `instance_id`, `current_utc_datetime`, `is_replaying` — properties - `call_activity(activity, *, input, retry_policy, app_id)` → `Task` - `call_child_workflow(workflow, *, input, instance_id, retry_policy, app_id)` → `Task` +- `schedule_new_workflow(workflow, *, input, instance_id, start_at, app_id, app_namespace)` → `str` (instance ID; fire-and-forget) - `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)` +**Detached vs child workflows** — use `call_child_workflow` when the parent needs to `yield` on the result. Use `schedule_new_workflow` (detached) when the parent should spawn and move on: + +- Fire-and-forget: no awaitable Task, returns the spawned instance ID synchronously. +- No parent linkage on the spawned instance (no completion or failure flows back). +- Deterministic default instance ID: derived from the parent instance ID + sequence number so replay resolves to the same history record. +- Cross-app: pass `app_id=...` — the runtime evaluates `WorkflowAccessPolicy` for the target app. +- Purge/terminate are not recursive across the boundary — the spawned instance manages its own lifecycle. +- History propagation is intentionally not offered (detached spawns do not inherit the caller's propagated history). + +See `examples/workflow/detached.py` for a per-tenant fan-out. + Module-level functions: - `when_all(tasks)` → `WhenAllTask` — wait for all tasks to complete - `when_any(tasks)` → `WhenAnyTask` — wait for first task to complete diff --git a/dapr/ext/workflow/_durabletask/internal/helpers.py b/dapr/ext/workflow/_durabletask/internal/helpers.py index 387a9ee10..f8435d558 100644 --- a/dapr/ext/workflow/_durabletask/internal/helpers.py +++ b/dapr/ext/workflow/_durabletask/internal/helpers.py @@ -244,6 +244,48 @@ def new_create_child_workflow_action( ) +def new_create_detached_workflow_action( + id: int, + name: str, + instance_id: str, + encoded_input: Optional[str], + router: Optional[pb.TaskRouter] = None, + scheduled_start_timestamp: Optional[timestamp_pb2.Timestamp] = None, +) -> pb.WorkflowAction: + """Build a WorkflowAction that spawns a detached workflow. + + The detached workflow is fully decoupled from the caller: no parent + linkage is recorded on the new instance and no completion or failure + flows back. Detached spawns intentionally do not propagate the caller's + history, so no historyPropagationScope is exposed here. + """ + detached = pb.CreateDetachedWorkflowAction( + instanceId=instance_id, + name=name, + input=get_string_value(encoded_input), + router=router, + ) + if scheduled_start_timestamp is not None: + detached.scheduledStartTimestamp.CopyFrom(scheduled_start_timestamp) + return pb.WorkflowAction( + id=id, + createDetachedWorkflow=detached, + router=router, + ) + + +def new_detached_workflow_instance_created_event( + event_id: int, instance_id: str +) -> pb.HistoryEvent: + return pb.HistoryEvent( + eventId=event_id, + timestamp=timestamp_pb2.Timestamp(), + detachedWorkflowInstanceCreated=pb.DetachedWorkflowInstanceCreatedEvent( + instanceId=instance_id, + ), + ) + + def is_empty(v: wrappers_pb2.StringValue): return v is None or v.value == '' diff --git a/dapr/ext/workflow/_durabletask/task.py b/dapr/ext/workflow/_durabletask/task.py index e3a190c41..2a9f381a0 100644 --- a/dapr/ext/workflow/_durabletask/task.py +++ b/dapr/ext/workflow/_durabletask/task.py @@ -179,6 +179,55 @@ def call_sub_orchestrator( """ pass + @abstractmethod + def schedule_new_workflow( + self, + workflow: Union[Orchestrator[TInput, TOutput], str], + *, + input: Optional[TInput] = None, + instance_id: Optional[str] = None, + start_at: Optional[datetime] = None, + app_id: Optional[str] = None, + app_namespace: Optional[str] = None, + ) -> str: + """Spawn a detached workflow instance and return its ID synchronously. + + Unlike ``call_sub_orchestrator``, the spawned workflow is fully + decoupled from the caller: no parent linkage is recorded on the new + instance and no completion or failure flows back. There is no + awaitable task — the call resolves as soon as the runtime accepts the + action. + + Parameters + ---------- + workflow: Orchestrator[TInput, TOutput] | str + A reference to the workflow function to spawn, or its registered + name (string form is required for cross-app spawns). + input: TInput | None + Optional JSON-serializable input to pass to the spawned workflow. + instance_id: str | None + A unique instance ID for the spawned workflow. If not specified, + a deterministic ID derived from the caller's instance ID and a + dedicated detached counter is used so replay resolves to the + same record in history. + start_at: datetime | None + Wall-clock time at which the spawned workflow should begin + executing. If not specified or in the past, the workflow starts + immediately. + app_id: str | None + The app ID that will execute the spawned workflow. If not + specified, the same app as the caller is used. + app_namespace: str | None + The Dapr namespace of the target app. Only meaningful with + ``app_id``. + + Returns + ------- + str + The instance ID of the spawned workflow. + """ + pass + @abstractmethod def wait_for_external_event( self, name: str, *, timeout: Optional[Union[datetime, timedelta]] = None diff --git a/dapr/ext/workflow/_durabletask/worker.py b/dapr/ext/workflow/_durabletask/worker.py index f7d0c04cc..53342eec8 100644 --- a/dapr/ext/workflow/_durabletask/worker.py +++ b/dapr/ext/workflow/_durabletask/worker.py @@ -1170,6 +1170,7 @@ def __init__(self, instance_id: str): self._pending_actions: dict[int, pb.WorkflowAction] = {} self._pending_tasks: dict[int, task.CompletableTask] = {} self._sequence_number = 0 + self._detached_counter = 0 self._current_utc_datetime = datetime(1000, 1, 1) self._instance_id = instance_id self._app_id = None @@ -1241,7 +1242,14 @@ def set_complete( self._is_complete = True self._completion_status = status - self._pending_actions.clear() # Cancel any pending actions + # Cancel any pending actions except detached-workflow spawns, which are + # fire-and-forget: the action is effective the moment schedule_new_workflow + # returns, so it must survive the caller's completion. + self._pending_actions = { + id_: a + for id_, a in self._pending_actions.items() + if a.HasField('createDetachedWorkflow') + } self._result = result result_json: Optional[str] = None @@ -1465,6 +1473,49 @@ def call_sub_orchestrator( ) return self._pending_tasks.get(id, task.CompletableTask()) + def schedule_new_workflow( + self, + workflow: Union[task.Orchestrator[TInput, TOutput], str], + *, + input: Optional[TInput] = None, + instance_id: Optional[str] = None, + start_at: Optional[datetime] = None, + app_id: Optional[str] = None, + app_namespace: Optional[str] = None, + ) -> str: + id = self.next_sequence_number() + workflow_name = workflow if isinstance(workflow, str) else task.get_name(workflow) + if instance_id is None: + self._detached_counter += 1 + instance_id = f'{self.instance_id}-{self._detached_counter}' + + router: Optional[pb.TaskRouter] = None + if self._app_id is not None or app_id is not None or app_namespace is not None: + router = pb.TaskRouter() + if self._app_id is not None: + router.sourceAppID = self._app_id + if app_id is not None: + router.targetAppID = app_id + if app_namespace is not None: + router.targetAppNamespace = app_namespace + + scheduled_start: Optional[timestamp_pb2.Timestamp] = None + if start_at is not None: + scheduled_start = timestamp_pb2.Timestamp() + scheduled_start.FromDatetime(start_at) + + encoded_input = shared.to_json(input) if input is not None else None + action = ph.new_create_detached_workflow_action( + id, + workflow_name, + instance_id, + encoded_input, + router, + scheduled_start_timestamp=scheduled_start, + ) + self._pending_actions[id] = action + return instance_id + def call_activity_function_helper( self, id: Optional[int], @@ -1972,6 +2023,30 @@ def process_event(self, ctx: _RuntimeOrchestrationContext, event: pb.HistoryEven expected_task_name=event.childWorkflowInstanceCreated.name, actual_task_name=action.createChildWorkflow.name, ) + elif event.HasField('detachedWorkflowInstanceCreated'): + task_id = event.eventId + if task_id in ctx._pending_actions and ph.is_optional_timer_action( + ctx._pending_actions[task_id] + ): + ctx._drop_optional_pending_at(task_id) + action = ctx._pending_actions.pop(task_id, None) + if not action: + raise _get_non_determinism_error( + task_id, task.get_name(ctx.schedule_new_workflow) + ) + elif not action.HasField('createDetachedWorkflow'): + expected_method_name = task.get_name(ctx.schedule_new_workflow) + raise _get_wrong_action_type_error(task_id, expected_method_name, action) + elif ( + action.createDetachedWorkflow.instanceId + != event.detachedWorkflowInstanceCreated.instanceId + ): + raise _get_wrong_action_name_error( + task_id, + method_name=task.get_name(ctx.schedule_new_workflow), + expected_task_name=event.detachedWorkflowInstanceCreated.instanceId, + actual_task_name=action.createDetachedWorkflow.instanceId, + ) elif event.HasField('childWorkflowInstanceCompleted'): task_id = event.childWorkflowInstanceCompleted.taskScheduledId sub_orch_task = ctx._pending_tasks.pop(task_id, None) @@ -2239,6 +2314,8 @@ def _get_method_name_for_action(action: pb.WorkflowAction) -> str: return task.get_name(task.OrchestrationContext.create_timer) elif action_type == 'createChildWorkflow': return task.get_name(task.OrchestrationContext.call_sub_orchestrator) + elif action_type == 'createDetachedWorkflow': + return task.get_name(task.OrchestrationContext.schedule_new_workflow) # elif action_type == "sendEvent": # return task.get_name(task.OrchestrationContext.send_event) else: diff --git a/dapr/ext/workflow/dapr_workflow_context.py b/dapr/ext/workflow/dapr_workflow_context.py index 7ceee0cbf..1ba599116 100644 --- a/dapr/ext/workflow/dapr_workflow_context.py +++ b/dapr/ext/workflow/dapr_workflow_context.py @@ -153,6 +153,38 @@ def wf(ctx: task.OrchestrationContext, inp: TInput): def get_propagated_history(self) -> Optional[PropagatedHistory]: return self.__obj.get_propagated_history() + def schedule_new_workflow( + self, + workflow: Union[Workflow, str], + *, + input: Optional[TInput] = None, + instance_id: Optional[str] = None, + start_at: Optional[datetime] = None, + app_id: Optional[str] = None, + app_namespace: Optional[str] = None, + ) -> str: + if isinstance(workflow, str): + workflow_name = workflow + elif hasattr(workflow, '_dapr_alternate_name'): + workflow_name = workflow.__dict__['_dapr_alternate_name'] + else: + workflow_name = workflow.__name__ + + if app_id is not None: + self._logger.debug( + f'{self.instance_id}: Spawning detached workflow {workflow_name} on app {app_id}' + ) + else: + self._logger.debug(f'{self.instance_id}: Spawning detached workflow {workflow_name}') + return self.__obj.schedule_new_workflow( + workflow_name, + input=input, + instance_id=instance_id, + start_at=start_at, + app_id=app_id, + app_namespace=app_namespace, + ) + def wait_for_external_event( self, name: str, diff --git a/dapr/ext/workflow/workflow_context.py b/dapr/ext/workflow/workflow_context.py index e3e98fe9e..6da0b8baa 100644 --- a/dapr/ext/workflow/workflow_context.py +++ b/dapr/ext/workflow/workflow_context.py @@ -180,6 +180,51 @@ def get_propagated_history(self) -> Optional[PropagatedHistory]: no history was propagated.""" pass + @abstractmethod + def schedule_new_workflow( + self, + workflow: Union[Workflow[TOutput], str], + *, + input: Optional[TInput] = None, + instance_id: Optional[str] = None, + start_at: Optional[datetime] = None, + app_id: Optional[str] = None, + app_namespace: Optional[str] = None, + ) -> str: + """Spawn a detached workflow instance and return its ID synchronously. + + Unlike ``call_child_workflow``, the spawned workflow is fully + decoupled from the caller: no parent linkage is recorded, no + completion or failure flows back, and there is no awaitable task. + + Parameters + ---------- + workflow: Workflow[TOutput] | str + A reference to the workflow function, or its registered name + (string form is required for cross-app spawns). + input: TInput | None + Optional JSON-serializable input to pass to the spawned workflow. + instance_id: str | None + Instance ID for the spawned workflow. When omitted, the runtime + derives a deterministic ID from the caller's instance ID so + replay resolves to the same history record. + start_at: datetime | None + Wall-clock time at which the spawned workflow should begin + executing. If not specified or in the past, the workflow starts + immediately. + app_id: str | None + The app ID that will execute the spawned workflow. + app_namespace: str | None + The Dapr namespace of the target app. Only meaningful with + ``app_id``. + + Returns + ------- + str + The instance ID of the spawned workflow. + """ + pass + @abstractmethod def wait_for_external_event( self, diff --git a/examples/workflow/detached.py b/examples/workflow/detached.py new file mode 100644 index 000000000..0aa33f511 --- /dev/null +++ b/examples/workflow/detached.py @@ -0,0 +1,78 @@ +# -*- coding: utf-8 -*- +# 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. + +"""Detached workflow fan-out example. + +The parent workflow spawns one detached workflow per tenant. Detached spawns +are fire-and-forget: the parent receives the spawned instance ID +synchronously and completes without waiting for the detached workflows to +finish. Each detached instance runs independently, with no parent linkage +in its history. + +This differs from ``call_child_workflow`` (see child_workflow.py), where the +parent yields a Task and blocks until the child completes. +""" + +import dapr.ext.workflow as wf + +wfr = wf.WorkflowRuntime() + +TENANTS = ['acme', 'globex', 'initech'] + + +@wfr.workflow +def parent_workflow(ctx: wf.DaprWorkflowContext, tenants: list[str]): + spawned_ids: list[str] = [] + for tenant in tenants: + detached_id = f'tenant-{tenant}' + spawned = ctx.schedule_new_workflow(tenant_workflow, input=tenant, instance_id=detached_id) + spawned_ids.append(spawned) + if not ctx.is_replaying: + print(f'*** Spawned detached workflow {spawned}', flush=True) + return spawned_ids + + +@wfr.workflow +def tenant_workflow(ctx: wf.DaprWorkflowContext, tenant: str): + if not ctx.is_replaying: + print(f'*** Tenant workflow started for {tenant}', flush=True) + yield ctx.call_activity(process_tenant, input=tenant) + return f'{tenant}-done' + + +@wfr.activity +def process_tenant(ctx: wf.WorkflowActivityContext, tenant: str) -> str: + print(f'*** Processing tenant {tenant}', flush=True) + return f'processed:{tenant}' + + +if __name__ == '__main__': + wfr.start() + + wf_client = wf.DaprWorkflowClient() + parent_id = wf_client.schedule_new_workflow(workflow=parent_workflow, input=TENANTS) + + parent_state = wf_client.wait_for_workflow_completion(parent_id, timeout_in_seconds=30) + print(f'*** Parent workflow {parent_id} finished: {parent_state.runtime_status}', flush=True) + + # The detached workflows continue running independently of the parent. + # Poll each to confirm they eventually complete. + for tenant in TENANTS: + detached_id = f'tenant-{tenant}' + state = wf_client.wait_for_workflow_completion(detached_id, timeout_in_seconds=30) + print( + f'*** Detached workflow {detached_id} finished: ' + f'{state.runtime_status} output={state.serialized_output}', + flush=True, + ) + + wfr.shutdown() diff --git a/tests/ext/workflow/durabletask/test_detached_workflow.py b/tests/ext/workflow/durabletask/test_detached_workflow.py new file mode 100644 index 000000000..231f63b19 --- /dev/null +++ b/tests/ext/workflow/durabletask/test_detached_workflow.py @@ -0,0 +1,262 @@ +# 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. + +"""Tests for detached-workflow support in the vendored durabletask engine. + +Detached workflows are fire-and-forget: the caller receives an instance ID +synchronously, no awaitable task is produced, and no completion or failure +flows back. On replay, the caller's history contains a single +DetachedWorkflowInstanceCreatedEvent per spawn that must reconcile against +the CreateDetachedWorkflowAction produced during execution. +""" + +import logging +from datetime import datetime, timezone + +import dapr.ext.workflow._durabletask.internal.helpers as helpers +import dapr.ext.workflow._durabletask.internal.protos as pb +from dapr.ext.workflow._durabletask import task, worker + +logging.basicConfig(level=logging.DEBUG) +TEST_LOGGER = logging.getLogger('tests-detached') + +TEST_INSTANCE_ID = 'parent-abc' + + +def _run(registry: worker._Registry, entry_name: str, encoded_input=None): + new_events = [ + helpers.new_workflow_started_event(), + helpers.new_execution_started_event( + entry_name, TEST_INSTANCE_ID, encoded_input=encoded_input + ), + ] + executor = worker._OrchestrationExecutor(registry, TEST_LOGGER) + return executor.execute(TEST_INSTANCE_ID, [], new_events) + + +def test_schedule_new_workflow_produces_detached_action_with_default_id(): + def child(ctx: task.OrchestrationContext, _): + pass + + def parent(ctx: task.OrchestrationContext, _): + spawned_id = ctx.schedule_new_workflow(child, input={'x': 1}) + return spawned_id + + registry = worker._Registry() + child_name = registry.add_orchestrator(child) + parent_name = registry.add_orchestrator(parent) + + result = _run(registry, parent_name) + actions = result.actions + + detached_actions = [a for a in actions if a.HasField('createDetachedWorkflow')] + assert len(detached_actions) == 1 + action = detached_actions[0] + assert action.createDetachedWorkflow.name == child_name + assert action.createDetachedWorkflow.instanceId == f'{TEST_INSTANCE_ID}-1' + assert action.createDetachedWorkflow.input.value == '{"x": 1}' + + complete = [a for a in actions if a.HasField('completeWorkflow')] + assert len(complete) == 1 + assert complete[0].completeWorkflow.workflowStatus == pb.ORCHESTRATION_STATUS_COMPLETED + assert complete[0].completeWorkflow.result.value == f'"{TEST_INSTANCE_ID}-1"' + + +def test_schedule_new_workflow_with_explicit_id_and_app_id(): + def child(ctx: task.OrchestrationContext, _): + pass + + def parent(ctx: task.OrchestrationContext, _): + ctx.schedule_new_workflow( + 'child', input=None, instance_id='detached-xyz', app_id='other-app' + ) + + registry = worker._Registry() + parent_name = registry.add_orchestrator(parent) + + exec_evt = helpers.new_execution_started_event( + parent_name, TEST_INSTANCE_ID, encoded_input=None + ) + exec_evt.router.sourceAppID = 'source-app' + new_events = [helpers.new_workflow_started_event(), exec_evt] + + executor = worker._OrchestrationExecutor(registry, TEST_LOGGER) + result = executor.execute(TEST_INSTANCE_ID, [], new_events) + + detached_actions = [a for a in result.actions if a.HasField('createDetachedWorkflow')] + assert len(detached_actions) == 1 + action = detached_actions[0] + assert action.createDetachedWorkflow.instanceId == 'detached-xyz' + assert action.createDetachedWorkflow.name == 'child' + assert action.router.sourceAppID == 'source-app' + assert action.router.targetAppID == 'other-app' + assert not action.createDetachedWorkflow.HasField('input') + + +def test_schedule_new_workflow_does_not_block_completion(): + """The parent must complete in the same execution tick as the spawn.""" + + def child(ctx: task.OrchestrationContext, _): + pass + + def parent(ctx: task.OrchestrationContext, _): + ctx.schedule_new_workflow(child) + return 'parent-done' + + registry = worker._Registry() + registry.add_orchestrator(child) + parent_name = registry.add_orchestrator(parent) + + result = _run(registry, parent_name) + + action_types = [a.WhichOneof('workflowActionType') for a in result.actions] + assert 'createDetachedWorkflow' in action_types + assert 'completeWorkflow' in action_types + complete = [a for a in result.actions if a.HasField('completeWorkflow')][0] + assert complete.completeWorkflow.workflowStatus == pb.ORCHESTRATION_STATUS_COMPLETED + assert complete.completeWorkflow.result.value == '"parent-done"' + + +def test_detached_event_reconciles_on_replay(): + """A history with DetachedWorkflowInstanceCreatedEvent must replay cleanly. + + We drive the workflow across two executor passes: the first spawns the + detached workflow and waits on an external event; the second replays the + recorded detached-created event and unblocks on the external event. + """ + + def child(ctx: task.OrchestrationContext, _): + pass + + def parent(ctx: task.OrchestrationContext, _): + ctx.schedule_new_workflow(child) + yield ctx.wait_for_external_event('go') + return 'done' + + registry = worker._Registry() + registry.add_orchestrator(child) + parent_name = registry.add_orchestrator(parent) + + spawn_id = f'{TEST_INSTANCE_ID}-1' + old_events = [ + helpers.new_workflow_started_event(), + helpers.new_execution_started_event(parent_name, TEST_INSTANCE_ID, encoded_input=None), + helpers.new_detached_workflow_instance_created_event(1, spawn_id), + ] + new_events = [helpers.new_event_raised_event('go')] + + executor = worker._OrchestrationExecutor(registry, TEST_LOGGER) + result = executor.execute(TEST_INSTANCE_ID, old_events, new_events) + + complete = [a for a in result.actions if a.HasField('completeWorkflow')] + assert len(complete) == 1 + assert complete[0].completeWorkflow.workflowStatus == pb.ORCHESTRATION_STATUS_COMPLETED + assert complete[0].completeWorkflow.result.value == '"done"' + + +def test_replay_detects_wrong_instance_id_mismatch(): + """Replaying a history whose recorded instance ID disagrees with the current + execution should surface a non-determinism error.""" + + def child(ctx: task.OrchestrationContext, _): + pass + + def parent(ctx: task.OrchestrationContext, _): + ctx.schedule_new_workflow(child, instance_id='current-id') + yield ctx.wait_for_external_event('go') + return 'done' + + registry = worker._Registry() + registry.add_orchestrator(child) + parent_name = registry.add_orchestrator(parent) + + old_events = [ + helpers.new_workflow_started_event(), + helpers.new_execution_started_event(parent_name, TEST_INSTANCE_ID, encoded_input=None), + helpers.new_detached_workflow_instance_created_event(1, 'stale-history-id'), + ] + new_events = [helpers.new_event_raised_event('go')] + + executor = worker._OrchestrationExecutor(registry, TEST_LOGGER) + result = executor.execute(TEST_INSTANCE_ID, old_events, new_events) + + complete = [a for a in result.actions if a.HasField('completeWorkflow')] + assert len(complete) == 1 + assert complete[0].completeWorkflow.workflowStatus == pb.ORCHESTRATION_STATUS_FAILED + assert complete[0].completeWorkflow.failureDetails.errorType == 'NonDeterminismError' + + +def test_schedule_new_workflow_with_start_at_sets_scheduled_start_timestamp(): + start = datetime(2027, 1, 15, 12, 30, 0, tzinfo=timezone.utc) + + def child(ctx: task.OrchestrationContext, _): + pass + + def parent(ctx: task.OrchestrationContext, _): + ctx.schedule_new_workflow(child, start_at=start) + + registry = worker._Registry() + registry.add_orchestrator(child) + parent_name = registry.add_orchestrator(parent) + + result = _run(registry, parent_name) + + detached_actions = [a for a in result.actions if a.HasField('createDetachedWorkflow')] + assert len(detached_actions) == 1 + action = detached_actions[0] + assert action.createDetachedWorkflow.HasField('scheduledStartTimestamp') + assert ( + action.createDetachedWorkflow.scheduledStartTimestamp.ToDatetime(tzinfo=timezone.utc) + == start + ) + + +def test_schedule_new_workflow_with_app_namespace_sets_router(): + def child(ctx: task.OrchestrationContext, _): + pass + + def parent(ctx: task.OrchestrationContext, _): + ctx.schedule_new_workflow('child', app_id='other-app', app_namespace='prod') + + registry = worker._Registry() + parent_name = registry.add_orchestrator(parent) + + result = _run(registry, parent_name) + + detached_actions = [a for a in result.actions if a.HasField('createDetachedWorkflow')] + assert len(detached_actions) == 1 + action = detached_actions[0] + assert action.router.targetAppID == 'other-app' + assert action.router.targetAppNamespace == 'prod' + + +def test_multiple_detached_spawns_use_sequential_default_ids(): + def child(ctx: task.OrchestrationContext, _): + pass + + def parent(ctx: task.OrchestrationContext, _): + for _ in range(3): + ctx.schedule_new_workflow(child) + + registry = worker._Registry() + registry.add_orchestrator(child) + parent_name = registry.add_orchestrator(parent) + + result = _run(registry, parent_name) + + detached_actions = [a for a in result.actions if a.HasField('createDetachedWorkflow')] + assert len(detached_actions) == 3 + spawned_ids = [a.createDetachedWorkflow.instanceId for a in detached_actions] + assert spawned_ids == [ + f'{TEST_INSTANCE_ID}-1', + f'{TEST_INSTANCE_ID}-2', + f'{TEST_INSTANCE_ID}-3', + ] diff --git a/tests/ext/workflow/test_dapr_workflow_context.py b/tests/ext/workflow/test_dapr_workflow_context.py index 22350dda8..b762a5519 100644 --- a/tests/ext/workflow/test_dapr_workflow_context.py +++ b/tests/ext/workflow/test_dapr_workflow_context.py @@ -29,6 +29,7 @@ mock_call_activity = 'call_activity' mock_call_sub_orchestrator = 'call_sub_orchestrator' mock_custom_status = 'custom_status' +mock_detached_instance_id = 'detached-instance-001' class FakeOrchestrationContext: @@ -36,6 +37,7 @@ def __init__(self): self.instance_id = mock_instance_id self.custom_status = None self._propagated_history = None + self.last_schedule_call: dict = {} def create_timer(self, fire_at): return mock_create_timer @@ -48,6 +50,19 @@ def call_sub_orchestrator( ): return mock_call_sub_orchestrator + def schedule_new_workflow( + self, workflow, *, input, instance_id, start_at=None, app_id, app_namespace=None + ): + self.last_schedule_call = { + 'workflow': workflow, + 'input': input, + 'instance_id': instance_id, + 'start_at': start_at, + 'app_id': app_id, + 'app_namespace': app_namespace, + } + return instance_id if instance_id is not None else mock_detached_instance_id + def set_custom_status(self, custom_status): self.custom_status = custom_status @@ -120,3 +135,56 @@ 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_schedule_new_workflow_with_function_ref(self): + with mock.patch( + 'dapr.ext.workflow._durabletask.worker._RuntimeOrchestrationContext', + return_value=FakeOrchestrationContext(), + ): + fake_context = worker._RuntimeOrchestrationContext(mock_instance_id) + dapr_wf_ctx = DaprWorkflowContext(fake_context) + + spawned_id = dapr_wf_ctx.schedule_new_workflow( + self.mock_client_child_wf, input={'x': 1} + ) + + assert spawned_id == mock_detached_instance_id + assert fake_context.last_schedule_call['workflow'] == 'mock_client_child_wf' + assert fake_context.last_schedule_call['input'] == {'x': 1} + assert fake_context.last_schedule_call['instance_id'] is None + assert fake_context.last_schedule_call['app_id'] is None + + def test_schedule_new_workflow_with_string_name_and_app_id(self): + with mock.patch( + 'dapr.ext.workflow._durabletask.worker._RuntimeOrchestrationContext', + return_value=FakeOrchestrationContext(), + ): + fake_context = worker._RuntimeOrchestrationContext(mock_instance_id) + dapr_wf_ctx = DaprWorkflowContext(fake_context) + + spawned_id = dapr_wf_ctx.schedule_new_workflow( + 'remote_wf', input='payload', instance_id='detached-42', app_id='other-app' + ) + + assert spawned_id == 'detached-42' + assert fake_context.last_schedule_call['workflow'] == 'remote_wf' + assert fake_context.last_schedule_call['input'] == 'payload' + assert fake_context.last_schedule_call['instance_id'] == 'detached-42' + assert fake_context.last_schedule_call['app_id'] == 'other-app' + + def test_schedule_new_workflow_resolves_alternate_name(self): + def some_wf(ctx, inp): + return inp + + some_wf._dapr_alternate_name = 'renamed-wf' + + with mock.patch( + 'dapr.ext.workflow._durabletask.worker._RuntimeOrchestrationContext', + return_value=FakeOrchestrationContext(), + ): + fake_context = worker._RuntimeOrchestrationContext(mock_instance_id) + dapr_wf_ctx = DaprWorkflowContext(fake_context) + + dapr_wf_ctx.schedule_new_workflow(some_wf) + + assert fake_context.last_schedule_call['workflow'] == 'renamed-wf'