From f7635dec70c9620ea0988439eca7eed0c1c0d559 Mon Sep 17 00:00:00 2001 From: Durable Workflow Date: Mon, 19 Oct 2026 21:24:45 +0000 Subject: [PATCH 1/2] Correct duplicate-start policies and current conflict classification --- CHANGELOG.md | 11 ++ docs/sdk-reference.md | 44 +++++-- pyproject.toml | 6 +- src/durable_workflow/client.py | 15 ++- src/durable_workflow/errors.py | 8 +- .../test_duplicate_start_policy.py | 123 ++++++++++++++++++ tests/test_errors.py | 8 +- 7 files changed, 195 insertions(+), 20 deletions(-) create mode 100644 tests/integration/test_duplicate_start_policy.py diff --git a/CHANGELOG.md b/CHANGELOG.md index 95e43e77..3219ea2a 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -14,6 +14,17 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 `deprecate_patch` for the same change ID, while validating each call's range and preserving retained duplicate-marker sequences. +## [2.6.1] - 2026-10-19 + +### Fixed + +- Raise `WorkflowAlreadyStarted` for the current Server's + `instance_already_started` duplicate-start reason, retaining support for the + historical `duplicate_not_allowed` reason. +- Correct start-policy documentation to `fail` and `use-existing`, including + active versus closed instances and preservation of original input/deadlines. + Start admission and workflow lifecycle behavior are unchanged. + ## [2.6.0] - 2026-10-18 ### Added diff --git a/docs/sdk-reference.md b/docs/sdk-reference.md index 7bda9607..cc4b0fa7 100644 --- a/docs/sdk-reference.md +++ b/docs/sdk-reference.md @@ -84,14 +84,42 @@ own `/api` routes. For example, use `http://server:8080`, not `http://server:8080/api`. Managed-runtime paths such as `https://cloud.example/api/runtime/v1/namespaces/acme` are valid as written. -A workflow ID is the durable identity of an instance, not a per-attempt request -ID. Starting the same ID again with the default `reject` duplicate policy raises -the typed `WorkflowAlreadyStarted` exception. Generate a unique ID for each new -instance, as the quickstart does. For an intentionally idempotent start, catch -`WorkflowAlreadyStarted` and reconnect with -`client.get_workflow_handle(workflow_id)`; choose `allow` or -`terminate_existing` only when creating another run or replacing the current -instance is the intended behavior. +### Duplicate workflow IDs + +A workflow ID is the durable identity of an instance. The PHP Server start +contract accepts these policies: + +| `duplicate_policy` | ID not yet used | Existing pending, running or waiting run | Closed instance | +| --- | --- | --- | --- | +| Omitted, `None`, or `"fail"` | Start a new instance | Raise `WorkflowAlreadyStarted` | Raise `WorkflowAlreadyStarted` | +| `"use-existing"` | Start a new instance | Return a handle to the original run | Raise `WorkflowAlreadyStarted` | + +An active duplicate keeps its original input, run and deadlines. It does not +apply the new request's arguments. Duplicate requests can append start-accepted +or start-rejected audit events without creating another business execution. + +```python +handle = await client.start_workflow( + workflow_type="greeter", + workflow_id="greet-customer-42", + task_queue="python-workers", + input=["world"], + duplicate_policy="use-existing", +) +``` + +Use a unique workflow ID for each independent execution, as the quickstart does. +To inspect an existing instance, including a closed one, use +`client.get_workflow_handle(workflow_id)` and its describe/result methods instead +of starting it again. `"use-existing"` does not restart or replace a closed run. +The SDK forwards the policy unchanged. `"reject"`, `"allow"` and +`"terminate_existing"` are invalid values and the PHP Server rejects them with +HTTP 422 (`InvalidArgument`). + +A resource-keyed ID can coordinate admission to one durable instance. It is not +a physical global mutex: cancellation or termination cannot revoke a request +already sent to an external system. Protect those effects with downstream +idempotency and, where required, resource-enforced fencing. For a fuller deployable example, see [`examples/order_processing`](https://github.com/durable-workflow/sdk-python/tree/main/examples/order_processing), which runs a diff --git a/pyproject.toml b/pyproject.toml index e6896db4..9551dbd3 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta" [project] name = "durable-workflow" -version = "2.6.0" +version = "2.6.1" description = "Python client and worker SDK for Durable Workflow Cloud and self-hosted Server" readme = "README.md" requires-python = ">=3.10" @@ -71,8 +71,8 @@ durable-workflow-replay-conformance = "durable_workflow.replay_conformance:main" durable-workflow-workflow-updates-conformance = "durable_workflow.workflow_updates_conformance:main" [tool.durable-workflow] -product-train = "2.6.0" -registry-version = "2.6.0" +product-train = "2.6.1" +registry-version = "2.6.1" supported-server-versions = "2.8.0" worker-protocol-version = "1.20" control-plane-version = "2" diff --git a/src/durable_workflow/client.py b/src/durable_workflow/client.py index f61b63f6..db47a0d1 100644 --- a/src/durable_workflow/client.py +++ b/src/durable_workflow/client.py @@ -3527,9 +3527,18 @@ async def start_workflow( ``workflow_type`` is the language-neutral type key registered via :func:`durable_workflow.workflow.defn`. ``task_queue`` selects which worker pool picks up the workflow. ``workflow_id`` is the - caller-supplied instance id — if it collides with an existing - workflow, behavior depends on ``duplicate_policy`` - (``reject`` | ``allow`` | ``terminate_existing``). + caller-supplied instance ID. The PHP Server start contract accepts + ``duplicate_policy="fail"`` (also the default when omitted) or + ``"use-existing"``. ``fail`` raises :class:`WorkflowAlreadyStarted` + if the ID already exists. ``use-existing`` returns the original run + only while it is pending, running or waiting. Both policies reject + a closed instance. Neither replaces a run or changes its input or + deadlines. Use a new workflow ID for a new independent execution. + + The SDK forwards the policy unchanged. ``reject``, ``allow`` and + ``terminate_existing`` are not supported wire values. A resource-keyed + workflow ID coordinates start admission; it does not revoke work + already running in an external system. ``input`` is a list of positional arguments passed to the workflow's ``run`` method; the SDK encodes the list with the default payload diff --git a/src/durable_workflow/errors.py b/src/durable_workflow/errors.py index c6c3830c..85ce0503 100644 --- a/src/durable_workflow/errors.py +++ b/src/durable_workflow/errors.py @@ -312,8 +312,10 @@ def __init__(self, workflow_id: str) -> None: class WorkflowAlreadyStarted(DurableWorkflowError): """A start request collided with an existing instance id. - Raised when duplicate-start policy is ``reject`` (the default) and the - caller-supplied ``workflow_id`` is already in use. + The PHP Server raises this for an existing ``workflow_id`` when the + duplicate policy is ``fail`` (the default), or when ``use-existing`` + addresses a closed instance. ``use-existing`` returns a handle only + while the original run is pending, running or waiting. """ def __init__(self, workflow_id: str) -> None: @@ -874,7 +876,7 @@ def update_validation_failed(default: str) -> UpdateValidationFailed: if status == 409: if reason == "schedule_already_exists": raise ScheduleAlreadyExists(context) - if reason == "duplicate_not_allowed": + if reason in ("duplicate_not_allowed", "instance_already_started"): raise WorkflowAlreadyStarted(context) if reason == "run_not_active": if operation == "signal": diff --git a/tests/integration/test_duplicate_start_policy.py b/tests/integration/test_duplicate_start_policy.py new file mode 100644 index 00000000..5551d6d5 --- /dev/null +++ b/tests/integration/test_duplicate_start_policy.py @@ -0,0 +1,123 @@ +"""Qualify start policies against real admission, audit history and completion.""" + +from __future__ import annotations + +import json +from typing import Any +from uuid import uuid4 + +import httpx +import pytest + +from durable_workflow import Client, Worker, workflow +from durable_workflow.errors import InvalidArgument, WorkflowAlreadyStarted, WorkflowNotFound + + +@workflow.defn(name="tests.duplicate-policy.echo") +class PolicyWorkflow: + def run(self, ctx: Any, value: Any) -> Any: + yield from () + return value + + +@pytest.mark.asyncio +@pytest.mark.parametrize("initial_policy", [None, "fail", "use-existing"]) +async def test_duplicate_policies_preserve_active_and_completed_original( + server_url: str, server_token: str, initial_policy: str | None, +) -> None: + suffix = uuid4().hex + queue = f"policy-{suffix}" + workflow_id = f"policy-{suffix}" + value = {"integer": 9223372036854775807, "null": None, "text": "original 雪"} + async with Client(server_url, token=server_token, namespace="default") as client: + starts: list[tuple[dict[str, Any], int]] = [] + + async def observe(response: httpx.Response) -> None: + request = response.request + if request.method == "POST" and request.url.path.endswith("/workflows"): + starts.append((json.loads(request.content), response.status_code)) + + client._http.event_hooks["response"].append(observe) + + async def start(policy: str | None, *, original: bool = False) -> Any: + return await client.start_workflow( + workflow_type="tests.duplicate-policy.echo", task_queue=queue, + workflow_id=workflow_id, input=[value if original else "replacement"], + execution_timeout_seconds=120 if original else 600, + run_timeout_seconds=60 if original else 300, duplicate_policy=policy, + ) + + handle = await start(initial_policy, original=True) + assert starts[-1][1] == 201 + assert starts[-1][0].get("duplicate_policy") == initial_policy + assert ("duplicate_policy" in starts[-1][0]) is (initial_policy is not None) + + for completed in (False, True): + if completed: + worker = Worker( + client, task_queue=queue, workflows=[PolicyWorkflow], activities=[], + poll_timeout=0.5, shutdown_timeout=10, + ) + await worker.run_until(workflow_id=workflow_id, timeout=30, poll_interval=0.1) + assert await handle.result(timeout=30) == value + + before = await handle.describe_run() + assert (before.status.lower() == "completed") is completed + assert before.input is not None + assert before.execution_deadline_at and before.run_deadline_at + history = (await handle.get_history())["events"] + + for policy in (None, "fail", "use-existing"): + returned = policy == "use-existing" and not completed + if returned: + existing = await start(policy) + assert (existing.workflow_id, existing.run_id) == (handle.workflow_id, handle.run_id) + else: + with pytest.raises(WorkflowAlreadyStarted) as duplicate: + await start(policy) + assert duplicate.value.workflow_id == workflow_id + request, status = starts[-1] + assert request.get("duplicate_policy") == policy + assert ("duplicate_policy" in request) is (policy is not None) + assert status == (200 if returned else 409) + after_history = (await handle.get_history())["events"] + assert after_history[:len(history)] == history + # Canonical duplicate starts append an admission audit event, + # while retaining the original business execution. + assert [event["event_type"] for event in after_history[len(history):]] == [ + "StartAccepted" if returned else "StartRejected", + ] + history = after_history + + for obsolete in ("reject", "allow", "terminate_existing"): + with pytest.raises(InvalidArgument): + await start(obsolete) + assert starts[-1][0]["duplicate_policy"] == obsolete + assert starts[-1][1] == 422 + assert (await handle.get_history())["events"] == history + + after = await handle.describe_run() + for field in ("run_id", "input", "status", "execution_deadline_at", "run_deadline_at", "output"): + assert getattr(after, field) == getattr(before, field), field + runs = await handle.list_runs() + assert len(runs.runs) == 1 + assert runs.runs[0].run_id == handle.run_id + assert sum(event["event_type"] == "WorkflowCompleted" for event in history) == int(completed) + print(json.dumps({"initial_policy": initial_policy, "workflow_id": workflow_id, + "run_id": handle.run_id, "start_statuses": [status for _, status in starts]})) + + +@pytest.mark.asyncio +@pytest.mark.parametrize("obsolete", ["reject", "allow", "terminate_existing"]) +async def test_invalid_policy_does_not_admit_a_fresh_workflow( + server_url: str, server_token: str, obsolete: str, +) -> None: + workflow_id = f"invalid-policy-{uuid4().hex}" + async with Client(server_url, token=server_token, namespace="default") as client: + with pytest.raises(InvalidArgument): + await client.start_workflow( + workflow_type="tests.duplicate-policy.echo", task_queue="invalid-policy", + workflow_id=workflow_id, input=[], duplicate_policy=obsolete, + ) + with pytest.raises(WorkflowNotFound): + await client.describe_workflow(workflow_id) diff --git a/tests/test_errors.py b/tests/test_errors.py index 6cedf8bf..2cbc0801 100644 --- a/tests/test_errors.py +++ b/tests/test_errors.py @@ -57,9 +57,11 @@ def test_404_generic(self) -> None: with pytest.raises(ServerError): _raise_for_status(404, {"reason": "other"}) - def test_409_duplicate(self) -> None: - with pytest.raises(WorkflowAlreadyStarted): - _raise_for_status(409, {"reason": "duplicate_not_allowed"}, context="wf-1") + @pytest.mark.parametrize("reason", ["duplicate_not_allowed", "instance_already_started"]) + def test_409_duplicate(self, reason: str) -> None: + with pytest.raises(WorkflowAlreadyStarted) as duplicate: + _raise_for_status(409, {"reason": reason}, context="resource-017") + assert duplicate.value.workflow_id == "resource-017" def test_409_query_rejected(self) -> None: with pytest.raises(QueryFailed): From e1eb1d931dda0bcec41de922834f6e09cab86756 Mon Sep 17 00:00:00 2001 From: Durable Workflow Date: Mon, 19 Oct 2026 21:27:01 +0000 Subject: [PATCH 2/2] Make original deadline assertions independently diagnostic --- tests/integration/test_duplicate_start_policy.py | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/tests/integration/test_duplicate_start_policy.py b/tests/integration/test_duplicate_start_policy.py index 5551d6d5..0efccb67 100644 --- a/tests/integration/test_duplicate_start_policy.py +++ b/tests/integration/test_duplicate_start_policy.py @@ -64,7 +64,8 @@ async def start(policy: str | None, *, original: bool = False) -> Any: before = await handle.describe_run() assert (before.status.lower() == "completed") is completed assert before.input is not None - assert before.execution_deadline_at and before.run_deadline_at + assert before.execution_deadline_at + assert before.run_deadline_at history = (await handle.get_history())["events"] for policy in (None, "fail", "use-existing"):