diff --git a/src/durable_workflow/worker.py b/src/durable_workflow/worker.py index fd01521..f07e6d3 100644 --- a/src/durable_workflow/worker.py +++ b/src/durable_workflow/worker.py @@ -3956,11 +3956,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: @@ -4122,7 +4123,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: @@ -4174,7 +4175,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() @@ -4226,6 +4227,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" diff --git a/tests/fixtures/replay_regressions/workflow-only-worker-resumes-external-activity.json b/tests/fixtures/replay_regressions/workflow-only-worker-resumes-external-activity.json new file mode 100644 index 0000000..e6d23ee --- /dev/null +++ b/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"} +} diff --git a/tests/integration/test_worker_task_families.py b/tests/integration/test_worker_task_families.py new file mode 100644 index 0000000..93a826d --- /dev/null +++ b/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) diff --git a/tests/test_replay_regression_corpus.py b/tests/test_replay_regression_corpus.py index a8bb247..b7acd62 100644 --- a/tests/test_replay_regression_corpus.py +++ b/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" @@ -500,6 +501,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")) diff --git a/tests/test_worker.py b/tests/test_worker.py index ee75842..04b8208 100644 --- a/tests/test_worker.py +++ b/tests/test_worker.py @@ -4362,6 +4362,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, ) @@ -4556,7 +4558,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") ) @@ -4688,6 +4690,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( @@ -4949,6 +4976,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(