diff --git a/dockerfiles/dynamicconfig/docker.yaml b/dockerfiles/dynamicconfig/docker.yaml index 4cc5e363..2c411e7b 100644 --- a/dockerfiles/dynamicconfig/docker.yaml +++ b/dockerfiles/dynamicconfig/docker.yaml @@ -22,3 +22,9 @@ frontend.enableUpdateWorkflowExecutionAsyncAccepted: frontend.workerVersioningWorkflowAPIs: - value: true constraints: {} +history.enableSignalWithStartFromWorkflow: + - value: true + constraints: {} +history.enableChasm: + - value: true + constraints: {} diff --git a/dockerfiles/go.Dockerfile b/dockerfiles/go.Dockerfile index ac697207..4e40bd9e 100644 --- a/dockerfiles/go.Dockerfile +++ b/dockerfiles/go.Dockerfile @@ -1,5 +1,5 @@ # Build in a full featured container -FROM golang:1.24 as build +FROM golang:1.25 AS build WORKDIR /app diff --git a/features/signalwithstart/both_workflows_visible/README.md b/features/signalwithstart/both_workflows_visible/README.md new file mode 100644 index 00000000..a1b4cecd --- /dev/null +++ b/features/signalwithstart/both_workflows_visible/README.md @@ -0,0 +1,13 @@ +# Signal-with-start from a workflow: payloads, memo, and visibility + +A workflow calls `workflow.signal_with_start_workflow(...)` passing a workflow +input argument, a signal argument, and a memo. This exercises the full +payload-serialization path (covering what the proto-binary server test verifies, +since Python encodes these payloads automatically). + +Verifies: both the caller and the started target complete; the target returns the +signal value; and the memo passed in the request is visible on the target. + +## Server requirements +Namespace dynamic config `history.enableSignalWithStartFromWorkflow=true` and +`history.enableChasm=true`. diff --git a/features/signalwithstart/both_workflows_visible/feature.py b/features/signalwithstart/both_workflows_visible/feature.py new file mode 100644 index 00000000..a513166b --- /dev/null +++ b/features/signalwithstart/both_workflows_visible/feature.py @@ -0,0 +1,90 @@ +import uuid +from dataclasses import dataclass +from datetime import timedelta +from typing import Optional + +from temporalio import workflow +from temporalio.client import WorkflowHandle + +from harness.python.feature import Runner, register_feature + +WORKFLOW_INPUT = "workflow-input" +SIGNAL_VALUE = "signal-input" +MEMO_KEY = "memo-key" +MEMO_VALUE = "memo-value" + + +@dataclass +class SwsResult: + workflow_id: str + run_id: str + + +@workflow.defn +class TargetWorkflow: + def __init__(self) -> None: + self._signal_value: Optional[str] = None + + @workflow.run + async def run(self, _input: str) -> str: + await workflow.wait_condition(lambda: self._signal_value is not None) + assert self._signal_value is not None + return self._signal_value + + @workflow.signal + def my_signal(self, value: str) -> None: + self._signal_value = value + + +@workflow.defn +class CallerWorkflow: + @workflow.run + async def run(self, target_id: str) -> SwsResult: + handle = await workflow.signal_with_start_workflow( + TargetWorkflow.run, + WORKFLOW_INPUT, + id=target_id, + task_queue=workflow.info().task_queue, + signal=TargetWorkflow.my_signal, + signal_args=SIGNAL_VALUE, + memo={MEMO_KEY: MEMO_VALUE}, + ) + return SwsResult(workflow_id=handle.id, run_id=handle.run_id or "") + + +async def start(runner: Runner) -> WorkflowHandle: + target_id = f"signalwithstart-both-visible-target-{uuid.uuid4()}" + return await runner.client.start_workflow( + CallerWorkflow.run, + target_id, + id=f"signalwithstart-both-visible-caller-{uuid.uuid4()}", + task_queue=runner.task_queue, + execution_timeout=timedelta(minutes=1), + ) + + +async def check_result(runner: Runner, handle: WorkflowHandle) -> None: + # Caller completes and returns the started target's identity. + result: SwsResult = await handle.result() + assert result.run_id, "expected a non-empty target run id" + + # Target completes after receiving the signal and returns the signal value. + target = runner.client.get_workflow_handle(result.workflow_id, run_id=result.run_id) + target_result = await target.result() + assert target_result == SIGNAL_VALUE, ( + f"expected target to return {SIGNAL_VALUE!r}, got {target_result!r}" + ) + + # The memo passed in the signal-with-start request is visible on the target. + desc = await target.describe() + memo = await desc.memo() + assert memo.get(MEMO_KEY) == MEMO_VALUE, ( + f"expected memo {MEMO_KEY}={MEMO_VALUE}, got {memo}" + ) + + +register_feature( + workflows=[CallerWorkflow, TargetWorkflow], + start=start, + check_result=check_result, +) diff --git a/features/signalwithstart/id_conflict_fail/README.md b/features/signalwithstart/id_conflict_fail/README.md new file mode 100644 index 00000000..e2a960a5 --- /dev/null +++ b/features/signalwithstart/id_conflict_fail/README.md @@ -0,0 +1,14 @@ +# Signal-with-start from a workflow: FAIL conflict policy is rejected + +A workflow calls `workflow.signal_with_start_workflow(...)` with +`id_conflict_policy=FAIL`. This policy is not supported for signal-with-start, so +the server rejects the scheduled Nexus operation. The rejection surfaces as a +workflow task failure (not a catchable error), so the caller workflow never +completes. + +Verifies: the caller's history contains a workflow task failure whose message +indicates the conflict policy is not supported. + +## Server requirements +Namespace dynamic config `history.enableSignalWithStartFromWorkflow=true` and +`history.enableChasm=true`. diff --git a/features/signalwithstart/id_conflict_fail/feature.py b/features/signalwithstart/id_conflict_fail/feature.py new file mode 100644 index 00000000..6bb0ecf3 --- /dev/null +++ b/features/signalwithstart/id_conflict_fail/feature.py @@ -0,0 +1,105 @@ +import uuid +from datetime import timedelta +from typing import Optional + +from temporalio import workflow +from temporalio.api.enums.v1 import EventType +from temporalio.client import WorkflowHandle +from temporalio.common import WorkflowIDConflictPolicy + +from harness.python.feature import Runner, register_feature + +SIGNAL_VALUE = "test-signal-value" + +_setup: dict = {} + + +@workflow.defn +class TargetWorkflow: + def __init__(self) -> None: + self._signal_value: Optional[str] = None + + @workflow.run + async def run(self) -> str: + await workflow.wait_condition(lambda: self._signal_value is not None) + assert self._signal_value is not None + return self._signal_value + + @workflow.signal + def my_signal(self, value: str) -> None: + self._signal_value = value + + +@workflow.defn +class CallerWorkflow: + @workflow.run + async def run(self, target_id: str) -> str: + # WORKFLOW_ID_CONFLICT_POLICY_FAIL is rejected by the server when + # scheduling the Nexus operation (signal-with-required-start is not a + # supported operation). This surfaces as a workflow task failure rather + # than a catchable error, so this run never completes; the client asserts + # on the workflow task failure in history. + handle = await workflow.signal_with_start_workflow( + TargetWorkflow.run, + id=target_id, + task_queue=workflow.info().task_queue, + signal=TargetWorkflow.my_signal, + signal_args=SIGNAL_VALUE, + id_conflict_policy=WorkflowIDConflictPolicy.FAIL, + ) + return handle.run_id or "" + + +async def start(runner: Runner) -> WorkflowHandle: + target_id = f"signalwithstart-id-conflict-fail-target-{uuid.uuid4()}" + _setup["target_id"] = target_id + + # Start a running target to mirror the server test setup. + target = await runner.client.start_workflow( + TargetWorkflow.run, + id=target_id, + task_queue=runner.task_queue, + execution_timeout=timedelta(minutes=1), + ) + _setup["target_run_id"] = target.result_run_id + + return await runner.client.start_workflow( + CallerWorkflow.run, + target_id, + id=f"signalwithstart-id-conflict-fail-caller-{uuid.uuid4()}", + task_queue=runner.task_queue, + execution_timeout=timedelta(minutes=1), + ) + + +async def check_result(runner: Runner, handle: WorkflowHandle) -> None: + # The caller never completes; wait for the workflow task failure that rejects + # the unsupported conflict policy. + event = await runner.wait_for_event( + handle, + lambda e: e.event_type == EventType.EVENT_TYPE_WORKFLOW_TASK_FAILED, + timeout=30.0, + ) + message = event.workflow_task_failed_event_attributes.failure.message + assert "not supported" in message.lower(), ( + f"expected 'not supported' rejection, got: {message}" + ) + + # Cleanup the caller (stuck failing its workflow task) and the target. + for wf_id, run_id in ( + (handle.id, None), + (_setup["target_id"], _setup["target_run_id"]), + ): + try: + await runner.client.get_workflow_handle(wf_id, run_id=run_id).terminate( + "cleanup" + ) + except Exception: + pass + + +register_feature( + workflows=[CallerWorkflow, TargetWorkflow], + start=start, + check_result=check_result, +) diff --git a/features/signalwithstart/id_conflict_terminate_existing/README.md b/features/signalwithstart/id_conflict_terminate_existing/README.md new file mode 100644 index 00000000..b6f41cc4 --- /dev/null +++ b/features/signalwithstart/id_conflict_terminate_existing/README.md @@ -0,0 +1,13 @@ +# Signal-with-start from a workflow: TERMINATE_EXISTING conflict policy + +A target workflow is already running. A workflow calls +`workflow.signal_with_start_workflow(...)` with +`id_conflict_policy=TERMINATE_EXISTING`. The running run is terminated and a new +run is started. + +Verifies: the returned run id differs from the original, and the original run's +status is TERMINATED. + +## Server requirements +Namespace dynamic config `history.enableSignalWithStartFromWorkflow=true` and +`history.enableChasm=true`. diff --git a/features/signalwithstart/id_conflict_terminate_existing/feature.py b/features/signalwithstart/id_conflict_terminate_existing/feature.py new file mode 100644 index 00000000..5c132b4d --- /dev/null +++ b/features/signalwithstart/id_conflict_terminate_existing/feature.py @@ -0,0 +1,105 @@ +import uuid +from dataclasses import dataclass +from datetime import timedelta +from typing import Optional + +from temporalio import workflow +from temporalio.client import WorkflowExecutionStatus, WorkflowHandle +from temporalio.common import WorkflowIDConflictPolicy + +from harness.python.feature import Runner, register_feature + +SIGNAL_VALUE = "test-signal-value" + +_setup: dict = {} + + +@dataclass +class SwsResult: + workflow_id: str + run_id: str + + +@workflow.defn +class TargetWorkflow: + def __init__(self) -> None: + self._signal_value: Optional[str] = None + + @workflow.run + async def run(self) -> str: + await workflow.wait_condition(lambda: self._signal_value is not None) + assert self._signal_value is not None + return self._signal_value + + @workflow.signal + def my_signal(self, value: str) -> None: + self._signal_value = value + + +@workflow.defn +class CallerWorkflow: + @workflow.run + async def run(self, target_id: str) -> SwsResult: + handle = await workflow.signal_with_start_workflow( + TargetWorkflow.run, + id=target_id, + task_queue=workflow.info().task_queue, + signal=TargetWorkflow.my_signal, + signal_args=SIGNAL_VALUE, + id_conflict_policy=WorkflowIDConflictPolicy.TERMINATE_EXISTING, + ) + return SwsResult(workflow_id=handle.id, run_id=handle.run_id or "") + + +async def start(runner: Runner) -> WorkflowHandle: + target_id = f"signalwithstart-id-conflict-terminate-target-{uuid.uuid4()}" + + # Start the target and leave it running. TERMINATE_EXISTING should terminate + # this run and start a new one. + target = await runner.client.start_workflow( + TargetWorkflow.run, + id=target_id, + task_queue=runner.task_queue, + execution_timeout=timedelta(minutes=1), + ) + _setup["original_run_id"] = target.result_run_id + + return await runner.client.start_workflow( + CallerWorkflow.run, + target_id, + id=f"signalwithstart-id-conflict-terminate-caller-{uuid.uuid4()}", + task_queue=runner.task_queue, + execution_timeout=timedelta(minutes=1), + ) + + +async def check_result(runner: Runner, handle: WorkflowHandle) -> None: + result: SwsResult = await handle.result() + assert result.run_id, "expected a non-empty run id" + assert result.run_id != _setup["original_run_id"], ( + "expected a new run id with TERMINATE_EXISTING" + ) + + # The original run must have been terminated. + original = runner.client.get_workflow_handle( + result.workflow_id, run_id=_setup["original_run_id"] + ) + desc = await original.describe() + assert desc.status == WorkflowExecutionStatus.TERMINATED, ( + f"expected original run TERMINATED, got {desc.status}" + ) + + # Cleanup the freshly-started run. + try: + await runner.client.get_workflow_handle( + result.workflow_id, run_id=result.run_id + ).terminate("cleanup") + except Exception: + pass + + +register_feature( + workflows=[CallerWorkflow, TargetWorkflow], + start=start, + check_result=check_result, +) diff --git a/features/signalwithstart/id_conflict_use_existing/README.md b/features/signalwithstart/id_conflict_use_existing/README.md new file mode 100644 index 00000000..9e3c5994 --- /dev/null +++ b/features/signalwithstart/id_conflict_use_existing/README.md @@ -0,0 +1,12 @@ +# Signal-with-start from a workflow: USE_EXISTING conflict policy + +A target workflow is already running. A workflow calls +`workflow.signal_with_start_workflow(...)` with +`id_conflict_policy=USE_EXISTING`. The existing run is signaled and its run id is +returned; no new run is started. + +Verifies: the returned run id equals the already-running target's run id. + +## Server requirements +Namespace dynamic config `history.enableSignalWithStartFromWorkflow=true` and +`history.enableChasm=true`. diff --git a/features/signalwithstart/id_conflict_use_existing/feature.py b/features/signalwithstart/id_conflict_use_existing/feature.py new file mode 100644 index 00000000..2d73e0e2 --- /dev/null +++ b/features/signalwithstart/id_conflict_use_existing/feature.py @@ -0,0 +1,87 @@ +import uuid +from dataclasses import dataclass +from datetime import timedelta +from typing import Optional + +from temporalio import workflow +from temporalio.client import WorkflowHandle +from temporalio.common import WorkflowIDConflictPolicy + +from harness.python.feature import Runner, register_feature + +SIGNAL_VALUE = "test-signal-value" + +_setup: dict = {} + + +@dataclass +class SwsResult: + workflow_id: str + run_id: str + + +@workflow.defn +class TargetWorkflow: + def __init__(self) -> None: + self._signal_value: Optional[str] = None + + @workflow.run + async def run(self) -> str: + await workflow.wait_condition(lambda: self._signal_value is not None) + assert self._signal_value is not None + return self._signal_value + + @workflow.signal + def my_signal(self, value: str) -> None: + self._signal_value = value + + +@workflow.defn +class CallerWorkflow: + @workflow.run + async def run(self, target_id: str) -> SwsResult: + handle = await workflow.signal_with_start_workflow( + TargetWorkflow.run, + id=target_id, + task_queue=workflow.info().task_queue, + signal=TargetWorkflow.my_signal, + signal_args=SIGNAL_VALUE, + id_conflict_policy=WorkflowIDConflictPolicy.USE_EXISTING, + ) + return SwsResult(workflow_id=handle.id, run_id=handle.run_id or "") + + +async def start(runner: Runner) -> WorkflowHandle: + target_id = f"signalwithstart-id-conflict-use-existing-target-{uuid.uuid4()}" + + # Start the target and leave it running. USE_EXISTING should signal this run + # and return its run id without starting a new one. + target = await runner.client.start_workflow( + TargetWorkflow.run, + id=target_id, + task_queue=runner.task_queue, + execution_timeout=timedelta(minutes=1), + ) + _setup["original_run_id"] = target.result_run_id + + return await runner.client.start_workflow( + CallerWorkflow.run, + target_id, + id=f"signalwithstart-id-conflict-use-existing-caller-{uuid.uuid4()}", + task_queue=runner.task_queue, + execution_timeout=timedelta(minutes=1), + ) + + +async def check_result(runner: Runner, handle: WorkflowHandle) -> None: + result: SwsResult = await handle.result() + assert result.run_id == _setup["original_run_id"], ( + f"expected existing run id {_setup['original_run_id']}, got {result.run_id}" + ) + + +register_feature( + workflows=[CallerWorkflow, TargetWorkflow], + start=start, + check_result=check_result, +) diff --git a/features/signalwithstart/id_reuse_allow_duplicate/README.md b/features/signalwithstart/id_reuse_allow_duplicate/README.md new file mode 100644 index 00000000..dc112f0f --- /dev/null +++ b/features/signalwithstart/id_reuse_allow_duplicate/README.md @@ -0,0 +1,11 @@ +# Signal-with-start from a workflow: ALLOW_DUPLICATE reuse policy + +A target workflow id has a previously-completed run. A workflow calls +`workflow.signal_with_start_workflow(...)` with `id_reuse_policy=ALLOW_DUPLICATE`. +Because duplicates are allowed for closed runs, a new execution is started. + +Verifies: the returned run id differs from the prior completed run's run id. + +## Server requirements +Namespace dynamic config `history.enableSignalWithStartFromWorkflow=true` and +`history.enableChasm=true`. diff --git a/features/signalwithstart/id_reuse_allow_duplicate/feature.py b/features/signalwithstart/id_reuse_allow_duplicate/feature.py new file mode 100644 index 00000000..73b447b9 --- /dev/null +++ b/features/signalwithstart/id_reuse_allow_duplicate/feature.py @@ -0,0 +1,89 @@ +import uuid +from dataclasses import dataclass +from datetime import timedelta +from typing import Optional + +from temporalio import workflow +from temporalio.client import WorkflowHandle +from temporalio.common import WorkflowIDReusePolicy + +from harness.python.feature import Runner, register_feature + +SIGNAL_VALUE = "test-signal-value" + +_setup: dict = {} + + +@dataclass +class SwsResult: + workflow_id: str + run_id: str + + +@workflow.defn +class TargetWorkflow: + def __init__(self) -> None: + self._signal_value: Optional[str] = None + + @workflow.run + async def run(self) -> str: + await workflow.wait_condition(lambda: self._signal_value is not None) + assert self._signal_value is not None + return self._signal_value + + @workflow.signal + def my_signal(self, value: str) -> None: + self._signal_value = value + + +@workflow.defn +class CallerWorkflow: + @workflow.run + async def run(self, target_id: str) -> SwsResult: + handle = await workflow.signal_with_start_workflow( + TargetWorkflow.run, + id=target_id, + task_queue=workflow.info().task_queue, + signal=TargetWorkflow.my_signal, + signal_args=SIGNAL_VALUE, + id_reuse_policy=WorkflowIDReusePolicy.ALLOW_DUPLICATE, + ) + return SwsResult(workflow_id=handle.id, run_id=handle.run_id or "") + + +async def start(runner: Runner) -> WorkflowHandle: + target_id = f"signalwithstart-id-reuse-allow-dup-target-{uuid.uuid4()}" + + # Start and complete the target so a closed run exists for target_id. + target = await runner.client.start_workflow( + TargetWorkflow.run, + id=target_id, + task_queue=runner.task_queue, + execution_timeout=timedelta(minutes=1), + ) + _setup["original_run_id"] = target.result_run_id + await target.signal(TargetWorkflow.my_signal, "setup") + await target.result() + + return await runner.client.start_workflow( + CallerWorkflow.run, + target_id, + id=f"signalwithstart-id-reuse-allow-dup-caller-{uuid.uuid4()}", + task_queue=runner.task_queue, + execution_timeout=timedelta(minutes=1), + ) + + +async def check_result(runner: Runner, handle: WorkflowHandle) -> None: + result: SwsResult = await handle.result() + assert result.run_id, "expected a non-empty run id" + assert result.run_id != _setup["original_run_id"], ( + "expected a new run id with ALLOW_DUPLICATE after the prior run completed" + ) + + +register_feature( + workflows=[CallerWorkflow, TargetWorkflow], + start=start, + check_result=check_result, +) diff --git a/features/signalwithstart/id_reuse_allow_duplicate_failed_only/README.md b/features/signalwithstart/id_reuse_allow_duplicate_failed_only/README.md new file mode 100644 index 00000000..462fecea --- /dev/null +++ b/features/signalwithstart/id_reuse_allow_duplicate_failed_only/README.md @@ -0,0 +1,14 @@ +# Signal-with-start from a workflow: ALLOW_DUPLICATE_FAILED_ONLY reuse policy + +Covers both sub-cases of `id_reuse_policy=ALLOW_DUPLICATE_FAILED_ONLY` when called +from a workflow via `workflow.signal_with_start_workflow(...)`: + +1. Target completed successfully -> the operation is rejected. +2. Target was terminated (a failed/closed outcome) -> a new run is started. + +Verifies: case 1 fails; case 2 starts a new run with a different run id. + +## Server requirements +Namespace dynamic config `history.enableSignalWithStartFromWorkflow=true` and +`history.enableChasm=true`. Also set +`system.workflowIdReuseMinimalInterval=0` to avoid reuse-interval throttling. diff --git a/features/signalwithstart/id_reuse_allow_duplicate_failed_only/feature.py b/features/signalwithstart/id_reuse_allow_duplicate_failed_only/feature.py new file mode 100644 index 00000000..c524b33d --- /dev/null +++ b/features/signalwithstart/id_reuse_allow_duplicate_failed_only/feature.py @@ -0,0 +1,143 @@ +import uuid +from dataclasses import dataclass +from datetime import timedelta +from typing import Optional + +from temporalio import workflow +from temporalio.client import WorkflowHandle +from temporalio.common import WorkflowIDReusePolicy + +from harness.python.feature import Runner, register_feature + +SIGNAL_VALUE = "test-signal-value" + + +@dataclass +class SwsResult: + workflow_id: str + run_id: str + error: Optional[str] = None + + +def _error_chain(err: BaseException) -> str: + """Flatten an exception and its causes; the useful detail lives on the cause + of the NexusOperationError, not its top-level message.""" + parts = [] + cur: Optional[BaseException] = err + for _ in range(8): + if cur is None: + break + parts.append(f"{type(cur).__name__}: {cur}") + cur = getattr(cur, "cause", None) or cur.__cause__ + return " | ".join(parts) + + +@workflow.defn +class TargetWorkflow: + def __init__(self) -> None: + self._signal_value: Optional[str] = None + + @workflow.run + async def run(self) -> str: + await workflow.wait_condition(lambda: self._signal_value is not None) + assert self._signal_value is not None + return self._signal_value + + @workflow.signal + def my_signal(self, value: str) -> None: + self._signal_value = value + + +@workflow.defn +class CallerWorkflow: + @workflow.run + async def run(self, target_id: str) -> SwsResult: + try: + handle = await workflow.signal_with_start_workflow( + TargetWorkflow.run, + id=target_id, + task_queue=workflow.info().task_queue, + signal=TargetWorkflow.my_signal, + signal_args=SIGNAL_VALUE, + id_reuse_policy=WorkflowIDReusePolicy.ALLOW_DUPLICATE_FAILED_ONLY, + ) + return SwsResult(workflow_id=handle.id, run_id=handle.run_id or "") + except Exception as err: + return SwsResult(workflow_id=target_id, run_id="", error=_error_chain(err)) + + +async def _run_caller(runner: Runner, target_id: str) -> SwsResult: + handle = await runner.client.start_workflow( + CallerWorkflow.run, + target_id, + id=f"signalwithstart-allow-dup-failed-only-caller-{uuid.uuid4()}", + task_queue=runner.task_queue, + execution_timeout=timedelta(minutes=1), + ) + return await handle.result() + + +async def start(runner: Runner) -> WorkflowHandle: + # Sub-case 1: target completed successfully -> ALLOW_DUPLICATE_FAILED_ONLY + # should reject the operation. start() drives this and returns its caller + # handle; the remaining sub-case runs in check_result. + target_id = f"signalwithstart-allow-dup-failed-only-completed-{uuid.uuid4()}" + target = await runner.client.start_workflow( + TargetWorkflow.run, + id=target_id, + task_queue=runner.task_queue, + execution_timeout=timedelta(minutes=1), + ) + await target.signal(TargetWorkflow.my_signal, "setup") + await target.result() + + return await runner.client.start_workflow( + CallerWorkflow.run, + target_id, + id=f"signalwithstart-allow-dup-failed-only-caller-{uuid.uuid4()}", + task_queue=runner.task_queue, + execution_timeout=timedelta(minutes=1), + ) + + +async def check_result(runner: Runner, handle: WorkflowHandle) -> None: + # Sub-case 1: completed target -> operation fails. + completed_result: SwsResult = await handle.result() + assert completed_result.error, ( + "expected failure for ALLOW_DUPLICATE_FAILED_ONLY against a completed run" + ) + + # Sub-case 2: terminated target -> a new run should start. + target_id = f"signalwithstart-allow-dup-failed-only-terminated-{uuid.uuid4()}" + target = await runner.client.start_workflow( + TargetWorkflow.run, + id=target_id, + task_queue=runner.task_queue, + execution_timeout=timedelta(minutes=1), + ) + original_run_id = target.result_run_id + await target.terminate("setup") + + terminated_result = await _run_caller(runner, target_id) + assert not terminated_result.error, ( + f"expected success after terminated run, got error: {terminated_result.error}" + ) + assert terminated_result.run_id, "expected a non-empty new run id" + assert terminated_result.run_id != original_run_id, ( + "expected a new run id after the previous run was terminated" + ) + + # Cleanup the freshly-started run. + try: + await runner.client.get_workflow_handle( + target_id, run_id=terminated_result.run_id + ).terminate("cleanup") + except Exception: + pass + + +register_feature( + workflows=[CallerWorkflow, TargetWorkflow], + start=start, + check_result=check_result, +) diff --git a/features/signalwithstart/id_reuse_reject_duplicate/README.md b/features/signalwithstart/id_reuse_reject_duplicate/README.md new file mode 100644 index 00000000..4c93d49e --- /dev/null +++ b/features/signalwithstart/id_reuse_reject_duplicate/README.md @@ -0,0 +1,13 @@ +# Signal-with-start from a workflow: REJECT_DUPLICATE reuse policy + +A target workflow id has a previously-completed run. A workflow calls +`workflow.signal_with_start_workflow(...)` with `id_reuse_policy=REJECT_DUPLICATE`. +The operation is rejected because a closed run with that id already exists. + +Verifies: the signal-with-start operation fails (the caller captures the failure, +whose cause message indicates a duplicate/already-started workflow). + +## Server requirements +Namespace dynamic config `history.enableSignalWithStartFromWorkflow=true` and +`history.enableChasm=true`. Also set +`system.workflowIdReuseMinimalInterval=0` to avoid reuse-interval throttling. diff --git a/features/signalwithstart/id_reuse_reject_duplicate/feature.py b/features/signalwithstart/id_reuse_reject_duplicate/feature.py new file mode 100644 index 00000000..98372b6c --- /dev/null +++ b/features/signalwithstart/id_reuse_reject_duplicate/feature.py @@ -0,0 +1,106 @@ +import uuid +from dataclasses import dataclass +from datetime import timedelta +from typing import Optional + +from temporalio import workflow +from temporalio.client import WorkflowHandle +from temporalio.common import WorkflowIDReusePolicy + +from harness.python.feature import Runner, register_feature + +SIGNAL_VALUE = "test-signal-value" + + +@dataclass +class SwsResult: + workflow_id: str + run_id: str + error: Optional[str] = None + + +def _error_chain(err: BaseException) -> str: + """Flatten an exception and its causes into a single message. The useful + detail (e.g. "already started") lives on the cause of the NexusOperationError, + not its top-level message.""" + parts = [] + cur: Optional[BaseException] = err + for _ in range(8): + if cur is None: + break + parts.append(f"{type(cur).__name__}: {cur}") + cur = getattr(cur, "cause", None) or cur.__cause__ + return " | ".join(parts) + + +@workflow.defn +class TargetWorkflow: + def __init__(self) -> None: + self._signal_value: Optional[str] = None + + @workflow.run + async def run(self) -> str: + await workflow.wait_condition(lambda: self._signal_value is not None) + assert self._signal_value is not None + return self._signal_value + + @workflow.signal + def my_signal(self, value: str) -> None: + self._signal_value = value + + +@workflow.defn +class CallerWorkflow: + @workflow.run + async def run(self, target_id: str) -> SwsResult: + # With REJECT_DUPLICATE and a previously-completed run, the operation is + # expected to fail. Capture the failure so the client can assert on it. + try: + handle = await workflow.signal_with_start_workflow( + TargetWorkflow.run, + id=target_id, + task_queue=workflow.info().task_queue, + signal=TargetWorkflow.my_signal, + signal_args=SIGNAL_VALUE, + id_reuse_policy=WorkflowIDReusePolicy.REJECT_DUPLICATE, + ) + return SwsResult(workflow_id=handle.id, run_id=handle.run_id or "") + except Exception as err: + return SwsResult(workflow_id=target_id, run_id="", error=_error_chain(err)) + + +async def start(runner: Runner) -> WorkflowHandle: + target_id = f"signalwithstart-id-reuse-reject-dup-target-{uuid.uuid4()}" + + # Start and complete the target so a closed run exists for target_id. + target = await runner.client.start_workflow( + TargetWorkflow.run, + id=target_id, + task_queue=runner.task_queue, + execution_timeout=timedelta(minutes=1), + ) + await target.signal(TargetWorkflow.my_signal, "setup") + await target.result() + + return await runner.client.start_workflow( + CallerWorkflow.run, + target_id, + id=f"signalwithstart-id-reuse-reject-dup-caller-{uuid.uuid4()}", + task_queue=runner.task_queue, + execution_timeout=timedelta(minutes=1), + ) + + +async def check_result(runner: Runner, handle: WorkflowHandle) -> None: + result: SwsResult = await handle.result() + assert result.error, "expected signal-with-start to fail with REJECT_DUPLICATE" + assert "duplicate" in result.error.lower() or "already" in result.error.lower(), ( + f"unexpected failure message: {result.error}" + ) + + +register_feature( + workflows=[CallerWorkflow, TargetWorkflow], + start=start, + check_result=check_result, +) diff --git a/features/signalwithstart/signal_terminated_workflow/README.md b/features/signalwithstart/signal_terminated_workflow/README.md new file mode 100644 index 00000000..de4514d0 --- /dev/null +++ b/features/signalwithstart/signal_terminated_workflow/README.md @@ -0,0 +1,12 @@ +# Signal-with-start from a workflow: terminated target + +A target workflow is started and then terminated (closed). A workflow then calls +`workflow.signal_with_start_workflow(...)` for the same workflow id. Because the +previous run is closed, a fresh run is started. + +Verifies: the returned run id differs from the terminated run's run id (a new +execution was started). + +## Server requirements +Namespace dynamic config `history.enableSignalWithStartFromWorkflow=true` and +`history.enableChasm=true`. diff --git a/features/signalwithstart/signal_terminated_workflow/feature.py b/features/signalwithstart/signal_terminated_workflow/feature.py new file mode 100644 index 00000000..5db81d7e --- /dev/null +++ b/features/signalwithstart/signal_terminated_workflow/feature.py @@ -0,0 +1,95 @@ +import uuid +from dataclasses import dataclass +from datetime import timedelta +from typing import Optional + +from temporalio import workflow +from temporalio.client import WorkflowHandle + +from harness.python.feature import Runner, register_feature + +SIGNAL_VALUE = "test-signal-value" + +_setup: dict = {} + + +@dataclass +class SwsResult: + workflow_id: str + run_id: str + + +@workflow.defn +class TargetWorkflow: + def __init__(self) -> None: + self._signal_value: Optional[str] = None + + @workflow.run + async def run(self) -> str: + await workflow.wait_condition(lambda: self._signal_value is not None) + assert self._signal_value is not None + return self._signal_value + + @workflow.signal + def my_signal(self, value: str) -> None: + self._signal_value = value + + +@workflow.defn +class CallerWorkflow: + @workflow.run + async def run(self, target_id: str) -> SwsResult: + handle = await workflow.signal_with_start_workflow( + TargetWorkflow.run, + id=target_id, + task_queue=workflow.info().task_queue, + signal=TargetWorkflow.my_signal, + signal_args=SIGNAL_VALUE, + ) + return SwsResult(workflow_id=handle.id, run_id=handle.run_id or "") + + +async def start(runner: Runner) -> WorkflowHandle: + target_id = f"signalwithstart-signal-terminated-target-{uuid.uuid4()}" + + # Start the target, then terminate it. Signal-with-start should start a fresh + # run because the previous one is closed. + target = await runner.client.start_workflow( + TargetWorkflow.run, + id=target_id, + task_queue=runner.task_queue, + execution_timeout=timedelta(minutes=1), + ) + _setup["original_run_id"] = target.result_run_id + await target.terminate("setup") + + return await runner.client.start_workflow( + CallerWorkflow.run, + target_id, + id=f"signalwithstart-signal-terminated-caller-{uuid.uuid4()}", + task_queue=runner.task_queue, + execution_timeout=timedelta(minutes=1), + ) + + +async def check_result(runner: Runner, handle: WorkflowHandle) -> None: + result: SwsResult = await handle.result() + assert result.run_id, "expected a non-empty run id" + assert result.run_id != _setup["original_run_id"], ( + "expected a new run id after the previous run was terminated" + ) + + # Cleanup the freshly-started run. + try: + await runner.client.get_workflow_handle( + result.workflow_id, run_id=result.run_id + ).terminate("cleanup") + except Exception: + pass + + +register_feature( + workflows=[CallerWorkflow, TargetWorkflow], + start=start, + check_result=check_result, +) diff --git a/features/signalwithstart/start_delay/README.md b/features/signalwithstart/start_delay/README.md new file mode 100644 index 00000000..cfa7e293 --- /dev/null +++ b/features/signalwithstart/start_delay/README.md @@ -0,0 +1,12 @@ +# Signal-with-start from a workflow: start delay + +A workflow calls `workflow.signal_with_start_workflow(...)` with a `start_delay`. +The target is scheduled to start after the delay; the signal is buffered and +delivered when the run begins. + +Verifies: a run id is returned, and after the delay the target starts, receives +the buffered signal, and completes returning the signal value. + +## Server requirements +Namespace dynamic config `history.enableSignalWithStartFromWorkflow=true` and +`history.enableChasm=true`. diff --git a/features/signalwithstart/start_delay/feature.py b/features/signalwithstart/start_delay/feature.py new file mode 100644 index 00000000..61610a3c --- /dev/null +++ b/features/signalwithstart/start_delay/feature.py @@ -0,0 +1,80 @@ +import uuid +from dataclasses import dataclass +from datetime import timedelta +from typing import Optional + +from temporalio import workflow +from temporalio.client import WorkflowHandle + +from harness.python.feature import Runner, register_feature + +SIGNAL_VALUE = "test-signal-value" +START_DELAY = timedelta(seconds=2) + + +@dataclass +class SwsResult: + workflow_id: str + run_id: str + + +@workflow.defn +class TargetWorkflow: + def __init__(self) -> None: + self._signal_value: Optional[str] = None + + @workflow.run + async def run(self) -> str: + await workflow.wait_condition(lambda: self._signal_value is not None) + assert self._signal_value is not None + return self._signal_value + + @workflow.signal + def my_signal(self, value: str) -> None: + self._signal_value = value + + +@workflow.defn +class CallerWorkflow: + @workflow.run + async def run(self, target_id: str) -> SwsResult: + handle = await workflow.signal_with_start_workflow( + TargetWorkflow.run, + id=target_id, + task_queue=workflow.info().task_queue, + signal=TargetWorkflow.my_signal, + signal_args=SIGNAL_VALUE, + start_delay=START_DELAY, + ) + return SwsResult(workflow_id=handle.id, run_id=handle.run_id or "") + + +async def start(runner: Runner) -> WorkflowHandle: + target_id = f"signalwithstart-start-delay-target-{uuid.uuid4()}" + return await runner.client.start_workflow( + CallerWorkflow.run, + target_id, + id=f"signalwithstart-start-delay-caller-{uuid.uuid4()}", + task_queue=runner.task_queue, + execution_timeout=timedelta(minutes=1), + ) + + +async def check_result(runner: Runner, handle: WorkflowHandle) -> None: + result: SwsResult = await handle.result() + assert result.run_id, "expected a non-empty target run id with start delay" + + # After the start delay elapses the target begins, receives the buffered + # signal, and completes returning the signal value. + target = runner.client.get_workflow_handle(result.workflow_id, run_id=result.run_id) + target_result = await target.result() + assert target_result == SIGNAL_VALUE, ( + f"expected target to return {SIGNAL_VALUE!r}, got {target_result!r}" + ) + + +register_feature( + workflows=[CallerWorkflow, TargetWorkflow], + start=start, + check_result=check_result, +)