From ef491eaadca5e54f54a8f27a46e0e11a2771bd2b Mon Sep 17 00:00:00 2001 From: Durable Workflow Date: Mon, 12 Oct 2026 02:45:09 +0000 Subject: [PATCH 1/2] Default Python workers to protocol 1.20 --- .github/workflows/ci.yml | 1 - README.md | 18 +++++----- docker-compose.test.yml | 4 +-- docs/cooperative-cancellation-design.md | 4 +-- pyproject.toml | 2 +- src/durable_workflow/client.py | 3 +- src/durable_workflow/worker.py | 17 ++++++++-- tests/test_client.py | 7 +++- tests/test_cooperative_cancellation_worker.py | 2 +- tests/test_worker.py | 33 +++++++++++++++++++ 10 files changed, 71 insertions(+), 20 deletions(-) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 134f7264..6381a072 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -159,7 +159,6 @@ jobs: timeout-minutes: 25 needs: [lint, test] env: - DURABLE_WORKFLOW_WORKER_PROTOCOL_VERSION: ${{ inputs.cooperative_qualification && '1.20' || '1.19' }} DURABLE_WORKFLOW_COOPERATIVE_QUALIFICATION: ${{ inputs.cooperative_qualification && '1' || '0' }} DURABLE_WORKFLOW_NATIVE_SOURCE: ${{ github.workspace }}/native COMPOSE_FILE: docker-compose.test.yml diff --git a/README.md b/README.md index 4bf1f431..1d6baebe 100644 --- a/README.md +++ b/README.md @@ -115,12 +115,15 @@ capabilities with Server at startup. Use stable `2.x` SDK and Server channels for new applications. The [compatibility guide](https://durable-workflow.com/docs/2.0/compatibility/) documents protocol and upgrade guarantees. -## Cooperative cancellation release candidate +## Worker protocol and cooperative cancellation + +Workers use protocol 1.20 by default. For a Server that supports only 1.19, set +`DURABLE_WORKFLOW_WORKER_PROTOCOL_VERSION=1.19`. This changes the request protocol +without enabling optional worker features. Use `request_cancellation()` for bounded, replayable workflow cleanup. Opt in against a Server that advertises protocol 1.20 and the required capabilities: -set `DURABLE_WORKFLOW_WORKER_PROTOCOL_VERSION=1.20` and include -`cooperative_cancellation` in the Worker's `capabilities`. Durable local callback +include `cooperative_cancellation` in the Worker's `capabilities`. Durable local callback admission also needs `prepared_local_activities`, with `prepared_local_activity_cancellation_policies` for explicit local policies. Independently cancellable scopes remain disabled. @@ -156,11 +159,10 @@ to true. CI verifies that checkout, builds the candidate Server, enables protoco authority and exact source provenance. An optional exact `native_commit` mounts that public Native checkout read-only into the test stack. The image's published Composer authority stays intact and the evidence identifies the source overlay. -These are source qualification runs. For a local -candidate, set `DURABLE_WORKFLOW_WORKER_PROTOCOL_VERSION=1.20` before starting -Compose and `DURABLE_WORKFLOW_COOPERATIVE_QUALIFICATION=1` for pytest. These cases -fail if the runtime does not discover the required capability. Ordinary CI -keeps protocol 1.19 and skips this unpublished feature's connected cases. +These are source qualification runs. For a local candidate, set +`DURABLE_WORKFLOW_COOPERATIVE_QUALIFICATION=1` for pytest. These cases fail if the +runtime does not discover the required capability. Ordinary CI uses the SDK's +default protocol and skips the optional connected cooperative cases. ## License diff --git a/docker-compose.test.yml b/docker-compose.test.yml index c82ed761..33254b75 100644 --- a/docker-compose.test.yml +++ b/docker-compose.test.yml @@ -30,8 +30,8 @@ services: WORKFLOW_SERVER_AUTH_DRIVER: token WORKFLOW_SERVER_AUTH_TOKEN: "test-token" DW_WORKFLOW_TASK_TIMEOUT: 10 - # Candidate cooperative qualification is explicit. Published defaults remain 1.19. - DW_WORKER_PROTOCOL_VERSION: "${DURABLE_WORKFLOW_WORKER_PROTOCOL_VERSION:-1.19}" + # Optional cooperative qualification remains explicit. + DW_WORKER_PROTOCOL_VERSION: "${DURABLE_WORKFLOW_WORKER_PROTOCOL_VERSION:-1.20}" depends_on: mysql: condition: service_healthy diff --git a/docs/cooperative-cancellation-design.md b/docs/cooperative-cancellation-design.md index bbe92604..cffec18f 100644 --- a/docs/cooperative-cancellation-design.md +++ b/docs/cooperative-cancellation-design.md @@ -1,8 +1,8 @@ # Cooperative cancellation source design This describes the unfinished source candidate for shared cancellation issue 136. -The default worker protocol remains 1.19. Cooperation requires explicit protocol -1.20 opt-in and a compatible Server and Native backend. The shared specification +The default worker protocol is 1.20. Cooperation requires an explicit worker +capability opt-in and a compatible Server and Native backend. The shared specification and published mixed-language qualification remain pending. ## Recorded context diff --git a/pyproject.toml b/pyproject.toml index 57bf33f2..7d730290 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -74,7 +74,7 @@ durable-workflow-workflow-updates-conformance = "durable_workflow.workflow_updat product-train = "2.6.0-rc.2" registry-version = "2.6.0rc2" supported-server-versions = "2.5.16" -worker-protocol-version = "1.19" +worker-protocol-version = "1.20" control-plane-version = "2" durable-selection = true durable-selection-minimum-worker-protocol-version = "1.19" diff --git a/src/durable_workflow/client.py b/src/durable_workflow/client.py index 4b078396..07035b31 100644 --- a/src/durable_workflow/client.py +++ b/src/durable_workflow/client.py @@ -72,7 +72,8 @@ from .retry_policy import TransportRetryPolicy from .worker_session import WorkerSessionOptions -PROTOCOL_VERSION = "1.19" +PROTOCOL_VERSION = "1.20" +MINIMUM_WORKER_PROTOCOL_VERSION = "1.19" CONTROL_PLANE_VERSION = "2" PORTABLE_WORKER_AFFINITY_CAPABILITY_MANIFEST: dict[str, dict[str, str | bool]] = { "local_activities": { diff --git a/src/durable_workflow/worker.py b/src/durable_workflow/worker.py index f07e6d3c..dfee7271 100644 --- a/src/durable_workflow/worker.py +++ b/src/durable_workflow/worker.py @@ -55,6 +55,7 @@ CONTROL_PLANE_REQUEST_CONTRACT_VERSION, CONTROL_PLANE_VERSION, DEFAULT_SDK_VERSION, + MINIMUM_WORKER_PROTOCOL_VERSION, PORTABLE_WORKER_AFFINITY_CAPABILITY_MANIFEST, PROTOCOL_VERSION, Client, @@ -874,6 +875,16 @@ def _server_protocol_compatible(server_version: str, sdk_version: str) -> bool: def _validate_server_compatibility(info: dict[str, Any]) -> None: + requested_protocol = _protocol_version_from_env( + "DURABLE_WORKFLOW_WORKER_PROTOCOL_VERSION", PROTOCOL_VERSION, + ) + if requested_protocol not in (MINIMUM_WORKER_PROTOCOL_VERSION, PROTOCOL_VERSION): + raise RuntimeError( + "Server compatibility error: sdk-python supports worker protocol " + f"{MINIMUM_WORKER_PROTOCOL_VERSION} or protocol {PROTOCOL_VERSION}; " + f"requested {requested_protocol!r}." + ) + control_plane = info.get("control_plane") if not isinstance(control_plane, dict): raise RuntimeError( @@ -911,15 +922,15 @@ def _validate_server_compatibility(info: dict[str, Any]) -> None: if not isinstance(worker_protocol, dict): raise RuntimeError( "Server compatibility error: missing worker_protocol manifest; " - f"sdk-python requires worker_protocol.version {PROTOCOL_VERSION}." + f"sdk-python requires worker_protocol.version {requested_protocol}." ) worker_protocol_version = _manifest_version(worker_protocol) - if not _server_protocol_compatible(worker_protocol_version, PROTOCOL_VERSION): + if not _server_protocol_compatible(worker_protocol_version, requested_protocol): raise RuntimeError( "Server compatibility error: incompatible worker_protocol.version " f"{worker_protocol_version!r}; sdk-python requires major-equal and " - f"minor>={PROTOCOL_VERSION!r}." + f"minor>={requested_protocol!r}." ) auth_composition = info.get("auth_composition_contract") diff --git a/tests/test_client.py b/tests/test_client.py index 2a77c366..0d2ba510 100644 --- a/tests/test_client.py +++ b/tests/test_client.py @@ -164,10 +164,15 @@ def test_control_plane_headers(self, client: Client) -> None: def test_worker_headers(self, client: Client) -> None: h = client._headers(worker=True) - assert PROTOCOL_VERSION == "1.19" + assert PROTOCOL_VERSION == "1.20" assert h["X-Durable-Workflow-Protocol-Version"] == PROTOCOL_VERSION assert "X-Durable-Workflow-Control-Plane-Version" not in h + def test_explicit_previous_worker_protocol(self, client: Client, monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setenv("DURABLE_WORKFLOW_WORKER_PROTOCOL_VERSION", "1.19") + assert client._headers(worker=True)["X-Durable-Workflow-Protocol-Version"] == "1.19" + assert client._headers(worker=False)["X-Durable-Workflow-Control-Plane-Version"] == "2" + def test_conformance_protocol_header_overrides(self, client: Client, monkeypatch: pytest.MonkeyPatch) -> None: monkeypatch.setenv("DURABLE_WORKFLOW_CONTROL_PLANE_VERSION", "3") monkeypatch.setenv("DURABLE_WORKFLOW_WORKER_PROTOCOL_VERSION", "2.0") diff --git a/tests/test_cooperative_cancellation_worker.py b/tests/test_cooperative_cancellation_worker.py index 692a337a..d1f2f7e7 100644 --- a/tests/test_cooperative_cancellation_worker.py +++ b/tests/test_cooperative_cancellation_worker.py @@ -134,7 +134,7 @@ async def complete(self, **kwargs: Any) -> dict[str, Any]: return {"outcome": "completed"} async def worker(self, monkeypatch: pytest.MonkeyPatch, **kwargs: Any) -> Worker: - monkeypatch.setenv("DURABLE_WORKFLOW_WORKER_PROTOCOL_VERSION", "1.20") + monkeypatch.delenv("DURABLE_WORKFLOW_WORKER_PROTOCOL_VERSION", raising=False) # These transport/replay fixtures use local callbacks that share mock # state with this process. They never poll or advertise remote handlers. local_callbacks = kwargs.pop("activities", []) diff --git a/tests/test_worker.py b/tests/test_worker.py index 04b8208f..43b6cbc5 100644 --- a/tests/test_worker.py +++ b/tests/test_worker.py @@ -1113,6 +1113,39 @@ async def test_register_rejects_worker_protocol_major_mismatch(self, mock_client await worker._register() mock_client.register_worker.assert_not_called() + @pytest.mark.asyncio + async def test_register_accepts_explicit_previous_protocol( + self, mock_client: AsyncMock, monkeypatch: pytest.MonkeyPatch, + ) -> None: + monkeypatch.setenv("DURABLE_WORKFLOW_WORKER_PROTOCOL_VERSION", "1.19") + mock_client.get_cluster_info = AsyncMock( + return_value=compatible_cluster_info(worker_protocol={"version": "1.19"}) + ) + worker = Worker(mock_client, task_queue="q1", workflows=[TestWorkflow], activities=[]) + await worker._register() + mock_client.register_worker.assert_called_once() + + @pytest.mark.asyncio + async def test_default_protocol_refuses_previous_server_before_registration(self, mock_client: AsyncMock) -> None: + mock_client.get_cluster_info = AsyncMock( + return_value=compatible_cluster_info(worker_protocol={"version": "1.19"}) + ) + worker = Worker(mock_client, task_queue="q1", workflows=[TestWorkflow], activities=[]) + with pytest.raises(RuntimeError, match="minor>='1.20'"): + await worker._register() + mock_client.register_worker.assert_not_called() + + @pytest.mark.asyncio + @pytest.mark.parametrize("version", ["1.18", "1.21", "2.0"]) + async def test_register_rejects_unsupported_protocol_override( + self, mock_client: AsyncMock, monkeypatch: pytest.MonkeyPatch, version: str, + ) -> None: + monkeypatch.setenv("DURABLE_WORKFLOW_WORKER_PROTOCOL_VERSION", version) + worker = Worker(mock_client, task_queue="q1", workflows=[TestWorkflow], activities=[]) + with pytest.raises(RuntimeError, match="supports worker protocol 1.19 or protocol 1.20"): + await worker._register() + mock_client.register_worker.assert_not_called() + @pytest.mark.asyncio async def test_register_accepts_higher_compatible_minor_protocol(self, mock_client: AsyncMock) -> None: # Server is one minor ahead of the SDK. MINOR bumps in workflow:v2's From 89d78650f838b1a9af8153d4f76a21753e6538b0 Mon Sep 17 00:00:00 2001 From: Durable Workflow Date: Mon, 12 Oct 2026 02:46:16 +0000 Subject: [PATCH 2/2] Exercise default cooperation and explicit legacy refusals --- tests/test_cancellation_scope_opening.py | 4 ++-- tests/test_cooperative_cancellation_client.py | 11 ++++------- tests/test_cooperative_remote_worker.py | 6 +++--- tests/test_prepared_local_activity_client.py | 4 ++-- tests/test_release_metadata.py | 4 ++-- 5 files changed, 13 insertions(+), 16 deletions(-) diff --git a/tests/test_cancellation_scope_opening.py b/tests/test_cancellation_scope_opening.py index 9188b23c..8bec86f7 100644 --- a/tests/test_cancellation_scope_opening.py +++ b/tests/test_cancellation_scope_opening.py @@ -152,8 +152,8 @@ async def test_invalid_authority_fails_before_io(change: dict[str, Any]) -> None send.assert_not_awaited() -async def test_published_protocol_refuses_opening_before_io(monkeypatch: pytest.MonkeyPatch) -> None: - monkeypatch.delenv("DURABLE_WORKFLOW_WORKER_PROTOCOL_VERSION") +async def test_previous_protocol_refuses_opening_before_io(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setenv("DURABLE_WORKFLOW_WORKER_PROTOCOL_VERSION", "1.19") async with Client("http://server", namespace="tenant") as client: with patch.object(client, "_request", new_callable=AsyncMock) as send: with pytest.raises(ValueError, match="1.20"): diff --git a/tests/test_cooperative_cancellation_client.py b/tests/test_cooperative_cancellation_client.py index 11a3f81f..e99d56c6 100644 --- a/tests/test_cooperative_cancellation_client.py +++ b/tests/test_cooperative_cancellation_client.py @@ -406,14 +406,11 @@ async def test_activity_stop_receipt_rejects_invalid_identity_before_io( @pytest.mark.asyncio -@pytest.mark.parametrize("version", [None, "1.19", "2.20", "invalid"]) -async def test_activity_stop_receipt_requires_explicit_worker_opt_in( - client: Client, monkeypatch: pytest.MonkeyPatch, version: str | None, +@pytest.mark.parametrize("version", ["1.19", "2.20", "invalid"]) +async def test_activity_stop_receipt_requires_compatible_protocol( + client: Client, monkeypatch: pytest.MonkeyPatch, version: str, ) -> None: - if version is None: - monkeypatch.delenv("DURABLE_WORKFLOW_WORKER_PROTOCOL_VERSION", raising=False) - else: - monkeypatch.setenv("DURABLE_WORKFLOW_WORKER_PROTOCOL_VERSION", version) + monkeypatch.setenv("DURABLE_WORKFLOW_WORKER_PROTOCOL_VERSION", version) with patch.object(client._http, "request", new_callable=AsyncMock) as send, pytest.raises(ValueError): await client.acknowledge_activity_cancellation( task_id="task/1", activity_attempt_id="attempt-1", lease_owner="worker-1", request_id="original-request", diff --git a/tests/test_cooperative_remote_worker.py b/tests/test_cooperative_remote_worker.py index e75fbc1a..6d6610da 100644 --- a/tests/test_cooperative_remote_worker.py +++ b/tests/test_cooperative_remote_worker.py @@ -94,7 +94,7 @@ async def application_failure(kind: str) -> None: @pytest_asyncio.fixture async def owner(monkeypatch: pytest.MonkeyPatch): # type: ignore[no-untyped-def] - monkeypatch.setenv("DURABLE_WORKFLOW_WORKER_PROTOCOL_VERSION", "1.20") + monkeypatch.delenv("DURABLE_WORKFLOW_WORKER_PROTOCOL_VERSION", raising=False) client = AsyncMock(spec=Client) client.get_cluster_info.return_value = compatible_cluster_info(worker_protocol={ "version": "1.20", "server_capabilities": {"cooperative_cancellation": True}, @@ -317,8 +317,8 @@ async def test_client_observation_keeps_worker_credentials_namespace_and_fence(m assert send.await_args.kwargs["json"] == {"activity_attempt_id": "attempt", "lease_owner": "owner"} -async def test_client_observation_requires_explicit_protocol_before_io(monkeypatch: pytest.MonkeyPatch) -> None: - monkeypatch.delenv("DURABLE_WORKFLOW_WORKER_PROTOCOL_VERSION", raising=False) +async def test_client_observation_refuses_previous_protocol_before_io(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setenv("DURABLE_WORKFLOW_WORKER_PROTOCOL_VERSION", "1.19") async with Client("http://runtime.test") as client: with (patch.object(client._http, "request", new_callable=AsyncMock) as send, pytest.raises(ValueError, match="1.20")): diff --git a/tests/test_prepared_local_activity_client.py b/tests/test_prepared_local_activity_client.py index b499da28..10164750 100644 --- a/tests/test_prepared_local_activity_client.py +++ b/tests/test_prepared_local_activity_client.py @@ -66,8 +66,8 @@ async def test_invalid_claim_or_authority_never_starts_transport(change: dict[st send.assert_not_awaited() -async def test_default_protocol_refuses_before_io(monkeypatch: pytest.MonkeyPatch) -> None: - monkeypatch.delenv("DURABLE_WORKFLOW_WORKER_PROTOCOL_VERSION") +async def test_previous_protocol_refuses_before_io(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setenv("DURABLE_WORKFLOW_WORKER_PROTOCOL_VERSION", "1.19") async with Client("http://server") as client: with patch.object(client, "_request", new_callable=AsyncMock) as send: with pytest.raises(ValueError, match="1.20"): diff --git a/tests/test_release_metadata.py b/tests/test_release_metadata.py index 846722b6..252c319d 100644 --- a/tests/test_release_metadata.py +++ b/tests/test_release_metadata.py @@ -45,13 +45,13 @@ def test_worker_release_identity_matches_supported_server_and_protocol() -> None assert release["product-train"] == project["version"] assert release["registry-version"] == project["version"].replace("-rc.", "rc") assert check_release_metadata.VERSION_PATTERN.fullmatch(release["supported-server-versions"]) is not None - assert release["worker-protocol-version"] == PROTOCOL_VERSION == "1.19" + assert release["worker-protocol-version"] == PROTOCOL_VERSION == "1.20" assert release["durable-selection"] is True assert release["durable-selection-minimum-worker-protocol-version"] == "1.19" assert declared_runtime_protocol_version( (REPO_ROOT / "src" / "durable_workflow" / "client.py").read_text(encoding="utf-8") - ) == "1.19" + ) == "1.20" def pypi_json(source: SourceMetadata, **overrides: object) -> bytes: