Publish Python 2.5.6 task-family polling fix

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "durable-workflow"
-version = "2.5.5"
+version = "2.5.6"
 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.5.5"
-registry-version = "2.5.5"
+product-train = "2.5.6"
+registry-version = "2.5.6"
 supported-server-versions = "2.5.16"
 worker-protocol-version = "1.19"
 control-plane-version = "2"

src/durable_workflow/worker.py

@@ -4077,11 +4077,12 @@ async def run(self) -> None:
         try:
             if self._stop.is_set():
                 return
-            wf_loop = asyncio.create_task(self._poll_workflow_tasks())
-            act_loop = asyncio.create_task(self._poll_activity_tasks())
-            hb_loop = asyncio.create_task(self._heartbeat_loop())
-            loops = [wf_loop, act_loop, hb_loop]
-            if self._query_tasks_supported:
+            loops = [asyncio.create_task(self._heartbeat_loop())]
+            if self.workflows:
+                loops.append(asyncio.create_task(self._poll_workflow_tasks()))
+            if self.activities:
+                loops.append(asyncio.create_task(self._poll_activity_tasks()))
+            if self.workflows and self._query_tasks_supported:
                 if self._can_clone_client_for_query_tasks():
                     self._start_query_task_thread()
                 else:
@@ -4243,7 +4244,7 @@ async def run_until(
             if self._stop.is_set():
                 raise asyncio.CancelledError
             background_tasks.append(asyncio.create_task(self._heartbeat_loop()))
-            if self._query_tasks_supported:
+            if self.workflows and self._query_tasks_supported:
                 if self._can_clone_client_for_query_tasks():
                     self._start_query_task_thread()
                 else:
@@ -4295,7 +4296,7 @@ async def _run_until_loop(
                     f"workflow {workflow_id} not terminal after {timeout}s (status={desc.status})"
                 )
 
-            if next_task_kind == "workflow":
+            if self.workflows and (next_task_kind == "workflow" or not self.activities):
                 await self._reserve_workflow_capacity()
                 try:
                     poll_start = time.perf_counter()
@@ -4347,6 +4348,9 @@ async def _run_until_loop(
                     next_task_kind = "workflow"
                 continue
 
+            if not self.activities:
+                await asyncio.sleep(poll_interval)
+                continue
             self._validate_cooperative_activity_handlers()
             if len(self._remote_activity_processes) >= self.max_concurrent_activity_tasks:
                 next_task_kind = "workflow"

tests/fixtures/replay_regressions/workflow-only-worker-resumes-external-activity.json

@@ -0,0 +1,22 @@
+{
+  "$schema": "https://raw.githubusercontent.com/durable-workflow/.github/main/regression-corpus/evidence-schema.json",
+  "fixture_schema": "durable-workflow.replay-regression/v1",
+  "id": "python-workflow-only-worker-resumes-external-activity",
+  "protocol_version": "1.19",
+  "bindings": ["python"],
+  "workflow": {
+    "type": "golden.single-activity",
+    "input": ["external-worker"],
+    "payload_codec": "avro"
+  },
+  "history": [
+    {"event_type": "WorkflowStarted", "payload": {"payload_codec": "avro"}},
+    {"event_type": "ActivityScheduled", "payload": {"sequence": 1, "activity_type": "golden.greet"}},
+    {"event_type": "ActivityCompleted", "payload": {
+      "sequence": 1,
+      "payload_codec": "avro",
+      "result": "wwHioz3/VYAiNwooZWNobzpleHRlcm5hbC13b3JrZXI="
+    }}
+  ],
+  "expected": {"type": "complete_workflow", "result": "echo:external-worker"}
+}

tests/integration/test_worker_task_families.py

@@ -0,0 +1,107 @@
+"""Exercise ordinary workers and count their real HTTP poll requests."""
+
+from __future__ import annotations
+
+import asyncio
+import importlib.metadata
+import json
+import uuid
+from collections import Counter
+from typing import Any
+
+import httpx
+import pytest
+
+from durable_workflow import Client, Worker, activity, workflow
+
+
+@activity.defn(name="tests.task-family.echo")
+async def echo(value: str) -> str:
+    return f"echo:{value}"
+
+
+@workflow.defn(name="tests.task-family.activity")
+class ActivityWorkflow:
+    def run(self, ctx: Any, value: str) -> Any:
+        return (yield ctx.schedule_activity("tests.task-family.echo", [value]))
+
+
+@workflow.defn(name="tests.task-family.timer")
+class TimerWorkflow:
+    def run(self, ctx: Any, value: str) -> Any:
+        yield ctx.sleep(2)
+        return f"timer:{value}"
+
+
+def record_polls(client: Client) -> Counter[str]:
+    counts: Counter[str] = Counter()
+
+    async def record(request: httpx.Request) -> None:
+        for family in ("workflow", "activity", "query"):
+            if request.url.path.endswith(f"/worker/{family}-tasks/poll"):
+                counts[family] += 1
+
+    client._http.event_hooks["request"].append(record)
+    return counts
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("mode", ["split", "mixed", "timer"])
+@pytest.mark.parametrize("entry_point", ["run", "run_until"])
+async def test_workers_execute_without_polling_empty_task_families(
+    server_url: str, server_token: str, mode: str, entry_point: str,
+) -> None:
+    suffix = uuid.uuid4().hex[:12]
+    queue = f"task-family-{suffix}"
+    async with (
+        Client(server_url, token=server_token, namespace="default") as control,
+        Client(server_url, token=server_token, namespace="default") as workflow_client,
+        Client(server_url, token=server_token, namespace="default") as activity_client,
+    ):
+        workflow_polls = record_polls(workflow_client)
+        activity_polls = record_polls(activity_client)
+        workflow_definition = TimerWorkflow if mode == "timer" else ActivityWorkflow
+        worker = Worker(
+            workflow_client, task_queue=queue,
+            workflows=[workflow_definition],
+            activities=[echo] if mode == "mixed" else [],
+            poll_timeout=0.5, shutdown_timeout=10,
+        )
+        workers = [worker]
+        if mode == "split":
+            workers.append(Worker(
+                activity_client, task_queue=queue, workflows=[], activities=[echo],
+                poll_timeout=0.5, shutdown_timeout=10,
+            ))
+
+        handle = await control.start_workflow(
+            workflow_type="tests.task-family.timer" if mode == "timer" else "tests.task-family.activity",
+            task_queue=queue, workflow_id=f"task-family-{suffix}", input=[suffix],
+        )
+        runners = [asyncio.create_task(
+            selected.run() if entry_point == "run" else selected.run_until(
+                workflow_id=handle.workflow_id, timeout=30, poll_interval=0.05,
+            )
+        ) for selected in workers]
+        try:
+            expected = f"timer:{suffix}" if mode == "timer" else f"echo:{suffix}"
+            assert await handle.result(timeout=30, poll_interval=0.1) == expected
+            if entry_point == "run_until":
+                terminal = await asyncio.wait_for(asyncio.gather(*runners), timeout=10)
+                assert all((result.status or "").lower() == "completed" for result in terminal)
+        finally:
+            await asyncio.gather(*(selected.stop() for selected in workers))
+            await asyncio.wait_for(asyncio.gather(*runners), timeout=10)
+
+        assert workflow_polls["workflow"] > 0
+        assert (workflow_polls["activity"] > 0) is (mode == "mixed")
+        if mode == "split":
+            assert activity_polls["activity"] > 0
+            assert activity_polls["workflow"] == 0
+            assert activity_polls["query"] == 0
+        print(json.dumps({
+            "sdk_version": importlib.metadata.version("durable-workflow"),
+            "mode": mode, "entry_point": entry_point, "workflow_id": handle.workflow_id,
+            "workflow_worker_polls": dict(workflow_polls),
+            "activity_worker_polls": dict(activity_polls), "outcome": "pass",
+        }), flush=True)

tests/test_replay_regression_corpus.py

@@ -9,7 +9,7 @@
 import pytest
 
 from durable_workflow import Replayer, Worker, serializer, workflow
-from durable_workflow.client import Client, WorkflowStreamAppendItem
+from durable_workflow.client import PROTOCOL_VERSION, Client, WorkflowExecution, WorkflowStreamAppendItem
 from durable_workflow.errors import NonDeterministicReplayError, WorkflowCancelled, WorkflowPayloadDecodeError
 from durable_workflow.workflow import (
     LocalActivityExecutionAborted,
@@ -31,6 +31,7 @@
     PostConditionReceiversWorkflow,
     UpdateSignalConditionTimerWorkflow,
 )
+from tests.test_worker import compatible_cluster_info
 
 FIXTURE_SCHEMA = "durable-workflow.replay-regression/v1"
 FIXTURE_DIR = Path(__file__).parent / "fixtures" / "replay_regressions"
@@ -489,6 +490,46 @@ def test_checked_in_replay_regression_corpus_uses_official_replayer(
     _execute_fixture(fixture)
 
 
+@pytest.mark.asyncio
+async def test_workflow_only_driver_resumes_external_activity_without_activity_polling() -> None:
+    fixture = json.loads(
+        (FIXTURE_DIR / "workflow-only-worker-resumes-external-activity.json").read_text(encoding="utf-8")
+    )
+    client = AsyncMock(spec=Client)
+    client.get_cluster_info.return_value = compatible_cluster_info(worker_protocol={"version": PROTOCOL_VERSION})
+    client.register_worker.return_value = {"registered": True}
+    client.deregister_worker_registration.return_value = {"outcome": "deregistered"}
+    client.poll_activity_task.side_effect = AssertionError("workflow-only worker must not poll activities")
+    client.poll_workflow_task.side_effect = [None, {
+        "task_id": "external-activity-resume-task", "workflow_id": "external-activity-caller",
+        "run_id": "original-run", "workflow_type": fixture["workflow"]["type"],
+        "workflow_task_attempt": 1, "history_events": fixture["history"],
+        "arguments": serializer.encode(fixture["workflow"]["input"], codec="avro"),
+        "payload_codec": "avro",
+    }]
+    client.complete_workflow_task.return_value = {"outcome": "completed"}
+
+    async def describe(workflow_id: str) -> WorkflowExecution:
+        return WorkflowExecution(
+            workflow_id=workflow_id, run_id="original-run", workflow_type=fixture["workflow"]["type"],
+            status="completed" if client.complete_workflow_task.await_count else "running",
+        )
+
+    client.describe_workflow.side_effect = describe
+    worker = Worker(client, task_queue="external-activity", workflows=[GoldenSingleActivityWorkflow], activities=[])
+
+    result = await worker.run_until(workflow_id="external-activity-caller", timeout=1, poll_interval=0.01)
+
+    assert result.status == "completed"
+    assert result.run_id == "original-run"
+    assert client.poll_workflow_task.await_count == 2
+    client.poll_activity_task.assert_not_awaited()
+    client.complete_workflow_task.assert_awaited_once()
+    commands = client.complete_workflow_task.call_args.kwargs["commands"]
+    assert [command["type"] for command in commands] == [fixture["expected"]["type"]]
+    assert serializer.decode_envelope(commands[0]["result"]) == fixture["expected"]["result"]
+
+
 @pytest.mark.asyncio
 async def test_worker_replays_recorded_side_effects_across_history_pages() -> None:
     fixture = json.loads((FIXTURE_DIR / "paged-recorded-side-effects.json").read_text(encoding="utf-8"))

tests/test_worker.py

@@ -4403,6 +4403,8 @@ async def request(method: str, path: str, **kwargs: object) -> httpx.Response:
                 client,
                 task_queue="orders",
                 worker_id="worker-only",
+                workflows=[TestWorkflow],
+                activities=[echo_activity],
                 poll_timeout=0.01,
                 heartbeat_interval=0.01,
             )
@@ -4597,7 +4599,7 @@ async def test_worker_loop_error_remains_primary_when_deregistration_fails(self,
             },
         )
         mock_client.deregister_worker_registration.side_effect = cleanup_error
-        worker = Worker(mock_client, task_queue="q1")
+        worker = Worker(mock_client, task_queue="q1", workflows=[TestWorkflow])
         worker._poll_workflow_tasks = AsyncMock(  # type: ignore[method-assign]
             side_effect=RuntimeError("worker loop failed")
         )
@@ -4729,6 +4731,31 @@ async def poll_query_task(self, **_: object) -> None:
 
 
 class TestPollLoops:
+    @pytest.mark.asyncio
+    @pytest.mark.parametrize("has_workflows,has_activities", [(True, False), (False, True), (False, False)])
+    async def test_run_polls_only_registered_task_families(
+        self, mock_client: AsyncMock, has_workflows: bool, has_activities: bool,
+    ) -> None:
+        worker = Worker(
+            mock_client,
+            task_queue="q1",
+            workflows=[TestWorkflow] if has_workflows else [],
+            activities=[echo_activity] if has_activities else [],
+            poll_timeout=0.01,
+        )
+        run_task = asyncio.create_task(worker.run())
+        try:
+            await asyncio.sleep(0.05)
+        finally:
+            await worker.stop()
+            await run_task
+
+        assert bool(mock_client.poll_workflow_task.await_count) is has_workflows
+        assert bool(mock_client.poll_activity_task.await_count) is has_activities
+        assert bool(mock_client.poll_query_task.await_count) is has_workflows
+        mock_client.register_worker.assert_awaited_once()
+        mock_client.deregister_worker_registration.assert_awaited_once_with(worker.worker_id)
+
     @pytest.mark.asyncio
     async def test_run_starts_both_loops(self, mock_client: AsyncMock) -> None:
         worker = Worker(
@@ -4990,6 +5017,33 @@ def test_process_metrics_memory_bytes_is_current_resident_set(
 
 
 class TestRunUntil:
+    @pytest.mark.asyncio
+    @pytest.mark.parametrize("has_workflows,has_activities", [(True, False), (False, True), (False, False)])
+    async def test_run_until_polls_only_registered_task_families(
+        self, mock_client: AsyncMock, has_workflows: bool, has_activities: bool,
+    ) -> None:
+        mock_client.describe_workflow = AsyncMock(side_effect=[
+            WorkflowExecution(workflow_id="wf-1", run_id="run-1", workflow_type="test-wf", status="running"),
+            WorkflowExecution(workflow_id="wf-1", run_id="run-1", workflow_type="test-wf", status="running"),
+            WorkflowExecution(workflow_id="wf-1", run_id="run-1", workflow_type="test-wf", status="completed"),
+        ])
+        worker = Worker(
+            mock_client,
+            task_queue="q1",
+            workflows=[TestWorkflow] if has_workflows else [],
+            activities=[echo_activity] if has_activities else [],
+            poll_timeout=0.01,
+        )
+
+        result = await worker.run_until(workflow_id="wf-1", timeout=1.0, poll_interval=0.01)
+
+        assert result.status == "completed"
+        assert bool(mock_client.poll_workflow_task.await_count) is has_workflows
+        assert bool(mock_client.poll_activity_task.await_count) is has_activities
+        if not has_workflows:
+            mock_client.poll_query_task.assert_not_awaited()
+        mock_client.deregister_worker_registration.assert_awaited_once_with(worker.worker_id)
+
     @pytest.mark.asyncio
     async def test_run_until_returns_terminal_description(self, mock_client: AsyncMock) -> None:
         mock_client.describe_workflow = AsyncMock(