From fc04a590cc01ef22ca09e91ee38cf9be6d800303 Mon Sep 17 00:00:00 2001 From: Durable Workflow Date: Mon, 19 Oct 2026 20:29:24 +0000 Subject: [PATCH 1/2] Document separate workflow and activity worker processes --- README.md | 1 + docs/guides/separate-workers.md | 142 +++++++++++++++++++++++++++++++ docs/reference/worker.md | 4 + examples/separate_workers.py | 143 ++++++++++++++++++++++++++++++++ mkdocs.yml | 1 + 5 files changed, 291 insertions(+) create mode 100644 docs/guides/separate-workers.md create mode 100644 examples/separate_workers.py diff --git a/README.md b/README.md index 96aeb7af..2cee7530 100644 --- a/README.md +++ b/README.md @@ -99,6 +99,7 @@ for the complete cross-SDK contract. - [Python SDK guide](https://durable-workflow.com/docs/2.0/polyglot/python/) - [Complete SDK reference](docs/sdk-reference.md) - [Runnable examples](examples/) +- [Separate workflow and activity worker processes](docs/guides/separate-workers.md) - [Symmetric SDK playground](https://github.com/durable-workflow/sample-app#symmetric-sdk-playground) ## Runtime choices diff --git a/docs/guides/separate-workers.md b/docs/guides/separate-workers.md new file mode 100644 index 00000000..0bce972f --- /dev/null +++ b/docs/guides/separate-workers.md @@ -0,0 +1,142 @@ +# Separate workflow and activity processes + +Run workflow coordination separately from remote activity callbacks when they +need different CPU, memory or concurrency limits. A `Worker` polls workflow +tasks only when it has workflow handlers, and activity tasks only when it has +activity handlers. Select the handler lists to separate roles. Both concurrency +settings must remain positive; setting one to zero is not a way to disable it. + +The [complete example](https://github.com/durable-workflow/sdk-python/blob/main/examples/separate_workers.py) +has workflow-only, activity-only, start and inspection commands. It uses the +normal installed SDK, without importing the repository's `src` directory. + +## Install and connect + +Use a POSIX shell on Linux, macOS or WSL with Python 3.10 or newer. From a checkout +of the SDK repository: + +```bash +python -m venv .venv +. .venv/bin/activate +python -m pip install 'durable-workflow==2.6.0' +export SERVER_URL=http://localhost:8080 +export WORKFLOW_NAMESPACE=default +export WORKFLOW_TOKEN=your-local-development-token +``` + +The qualification baseline is published Server **2.9.5** and Python SDK +**2.6.0**. Start Server using its [self-hosted installation guide](https://github.com/durable-workflow/server#run-server). +Its API, queue worker and scheduler must be running and `/api/ready` must be +ready. The Python processes talk to Server over HTTP and do not connect to its +database. `SERVER_URL` is the origin without `/api`. For Cloud, use the complete +provisioned namespace runtime URL and credentials instead. + +The local example uses one development token. In a deployed application, set +`WORKFLOW_TOKEN` separately for each process: a worker credential for worker +commands, a client credential for `start`, and an operator credential for +`inspect`. Do not copy an operator token into workers to make inspection work. + +## Start two workers + +Activate the same environment and set the connection variables in each terminal. +Run the workflow role in the first terminal: + +```bash +python examples/separate_workers.py workflow --worker-id python-workflows-1 --slots 4 +``` + +Run the activity role in the second: + +```bash +python examples/separate_workers.py activity --worker-id python-activities-1 --slots 2 +``` + +Every live process needs its own worker ID. The two registrations use the same +`python-split` task queue and namespace, with different supported types: + +| Worker | Workflow types | Activity types | +| --- | --- | --- | +| `python-workflows-1` | `examples.split-workers.greeter` | empty | +| `python-activities-1` | empty | `examples.split-workers.greet` | + +In a third terminal, inspect the actual roster, then start a workflow: + +```bash +python examples/separate_workers.py inspect +python examples/separate_workers.py start --name world +python examples/separate_workers.py start --name retry --retry-once +``` + +The final results are `hello, world` and `hello, retry`. The activity process +prints its worker ID, process ID and attempt number. The retry example fails +attempt 1 deliberately; Server schedules attempt 2 under the activity's recorded +two-attempt policy. The start command only submits and observes. It does not run +either handler itself, and its 30-second wait is not a cancellation request. + +The roster reports both numeric slot settings even though each process has only +one handler family. Supported type lists determine which family it serves. +Do not count the inactive family's positive setting as usable handler capacity. + +## Route activities to a distinct queue + +The shared-queue example separates processes by supported types. It does not +create a separate activity queue. For an explicit queue split, start workers +with matching queue names: + +```bash +# Workflow terminal +python examples/separate_workers.py workflow --worker-id python-workflows-orders --queue orders +# Activity terminal +python examples/separate_workers.py activity --worker-id python-activities-heavy --queue heavy +# Client terminal +python examples/separate_workers.py start --queue orders --activity-queue heavy --retry-once +``` + +The workflow passes `queue="heavy"` to `ctx.schedule_activity(...)`. With no +override, the activity uses the workflow's queue. An activity worker on `heavy` +will not serve work scheduled to `orders`. Namespace, queue and supported type +must all match. Build compatibility adds another routing constraint during +[worker rollouts](https://durable-workflow.com/docs/2.0/polyglot/worker-build-id-rollout/). + +## Size and stop each process + +`--slots` changes only that role's `max_concurrent_*_tasks`. Set container or +process CPU and memory limits independently. Activity slots must also fit the +database, HTTP connection pool, downstream rate limits and callback memory +budget. More processes multiply local capacity; use +[server-side admission budgets](https://durable-workflow.com/docs/2.0/polyglot/task-queue-admission/) +when the whole namespace or queue needs a ceiling. + +An `async def` activity must yield while waiting. Blocking CPU or synchronous +I/O inside it blocks that process's event loop, including polling and heartbeat +work. Use suitable nonblocking libraries or bounded executors/processes for the +workload; a separate activity process does not remove downstream contention. + +Press Ctrl-C in each worker terminal, or send SIGTERM to each worker process. +The example awaits `Worker.stop()` and its run task. It stops polling, drains +accepted work within its 10-second shutdown budget, and deregisters. Give the +process supervisor a termination grace period longer than this budget plus +HTTP cleanup time. Errors during deregistration or drain cause a nonzero exit +and must be investigated. Stopping the client does not stop workers or cancel +the workflow. A later `inspect` should no longer list successfully stopped +worker IDs; a missing process alone is not proof of successful deregistration. + +## Local resources and recovery + +- Ordinary `ctx.schedule_activity(...)` calls are remote activity tasks. They + can be picked up by a compatible replacement activity worker after normal + lease/retry handling. Application side effects still need idempotency. +- Opted-in local activities execute in the **workflow worker**. Register their + callbacks on that worker; an activity-only process cannot supply them. They + consume workflow-process resources and need their documented durable local + activity and cancellation capabilities. +- Worker sessions intentionally bind work to the session owner while its lease + is active. Process separation does not make an in-memory client, file handle + or session durable or migrate it to any available process. +- Sticky history caches belong to workflow workers and are optional. Eviction + or process replacement uses cold replay from Server history. Keep correctness + independent of cache contents. + +See [Worker lifecycle and shutdown](../reference/worker.md), +[local activities and sessions](../sdk-reference.md#local-activities), and +[sticky execution](sticky-execution.md) for the respective contracts. diff --git a/docs/reference/worker.md b/docs/reference/worker.md index 964cea39..5578f9c9 100644 --- a/docs/reference/worker.md +++ b/docs/reference/worker.md @@ -1,5 +1,9 @@ # Worker +See [Separate workflow and activity processes](../guides/separate-workers.md) +for a runnable deployment example with distinct registrations, independent +slots, remote activity retries and signal-driven shutdown. + ## Lifecycle and shutdown `Worker.run()` and `Worker.run_until()` register through the worker protocol diff --git a/examples/separate_workers.py b/examples/separate_workers.py new file mode 100644 index 00000000..f21df468 --- /dev/null +++ b/examples/separate_workers.py @@ -0,0 +1,143 @@ +"""Run workflow and remote activity handlers in separate processes. + +See docs/guides/separate-workers.md for installation and three-terminal usage. +""" +from __future__ import annotations + +import argparse +import asyncio +import json +import logging +import os +import signal +from uuid import uuid4 + +from durable_workflow import ActivityRetryPolicy, Client, Worker, activity, workflow + +WORKFLOW_TYPE = "examples.split-workers.greeter" +ACTIVITY_TYPE = "examples.split-workers.greet" + + +@activity.defn(name=ACTIVITY_TYPE) +async def greet(name: str, retry_once: bool) -> str: + info = activity.context().info + print(json.dumps({ + "event": "activity_attempt", "worker_id": info.worker_id, + "attempt": info.attempt_number, "process_id": os.getpid(), + }), flush=True) + if retry_once and info.attempt_number == 1: + raise RuntimeError("intentional first-attempt failure for the example") + return f"hello, {name}" + + +@workflow.defn(name=WORKFLOW_TYPE) +class Greeter: + def run(self, ctx, name: str, retry_once: bool, activity_queue: str | None): + return (yield ctx.schedule_activity( + ACTIVITY_TYPE, + [name, retry_once], + queue=activity_queue, + retry_policy=ActivityRetryPolicy(max_attempts=2, initial_interval_seconds=1), + start_to_close_timeout=5, + schedule_to_close_timeout=20, + )) + + +def positive_slots(value: str) -> int: + slots = int(value) + if slots < 1: + raise argparse.ArgumentTypeError("slots must be at least 1") + return slots + + +def parse_args() -> argparse.Namespace: + parser = argparse.ArgumentParser(description=__doc__) + roles = parser.add_subparsers(dest="role", required=True) + for role in ("workflow", "activity"): + worker = roles.add_parser(role, help=f"run only {role} handlers") + worker.add_argument("--worker-id", required=True, help="unique for each live worker process") + worker.add_argument("--queue", default="python-split") + worker.add_argument("--slots", type=positive_slots, default=4) + start = roles.add_parser("start", help="start a workflow and wait up to 30 seconds") + start.add_argument("--queue", default="python-split") + start.add_argument("--name", default="world") + start.add_argument("--retry-once", action="store_true") + start.add_argument("--activity-queue", help="optional explicit queue for the remote activity") + inspect = roles.add_parser("inspect", help="read the registered worker roster (operator credential)") + inspect.add_argument("--queue", default="python-split") + return parser.parse_args() + + +async def run_worker(client: Client, args: argparse.Namespace) -> None: + workflow_role = args.role == "workflow" + worker = Worker( + client, + task_queue=args.queue, + worker_id=args.worker_id, + workflows=[Greeter] if workflow_role else [], + activities=[] if workflow_role else [greet], + max_concurrent_workflow_tasks=args.slots if workflow_role else 1, + max_concurrent_activity_tasks=1 if workflow_role else args.slots, + poll_timeout=5, + shutdown_timeout=10, + ) + shutdown = asyncio.Event() + loop = asyncio.get_running_loop() + for requested_signal in (signal.SIGINT, signal.SIGTERM): + loop.add_signal_handler(requested_signal, shutdown.set) + run_task = asyncio.create_task(worker.run()) + shutdown_task = asyncio.create_task(shutdown.wait()) + try: + done, _ = await asyncio.wait({run_task, shutdown_task}, return_when=asyncio.FIRST_COMPLETED) + if run_task in done: + await run_task + finally: + shutdown_task.cancel() + await asyncio.gather(shutdown_task, return_exceptions=True) + try: + await worker.stop() + finally: + await run_task + for requested_signal in (signal.SIGINT, signal.SIGTERM): + loop.remove_signal_handler(requested_signal) + print(json.dumps({"event": "worker_stopped", "worker_id": worker.worker_id}), flush=True) + + +async def main(args: argparse.Namespace) -> None: + server_url = os.environ.get("SERVER_URL") + token = os.environ.get("WORKFLOW_TOKEN") + if not server_url or not token: + raise SystemExit("set SERVER_URL and WORKFLOW_TOKEN; see the separate-workers guide") + async with Client( + server_url, token=token, namespace=os.environ.get("WORKFLOW_NAMESPACE", "default"), + ) as client: + if args.role in ("workflow", "activity"): + await run_worker(client, args) + elif args.role == "inspect": + roster = await client.list_workers(task_queue=args.queue) + print(json.dumps([{ + "worker_id": worker.worker_id, + "task_queue": worker.task_queue, + "status": worker.status, + "workflow_types": worker.supported_workflow_types, + "activity_types": worker.supported_activity_types, + "workflow_slots": worker.max_concurrent_workflow_tasks, + "activity_slots": worker.max_concurrent_activity_tasks, + } for worker in roster.workers], indent=2)) + else: + handle = await client.start_workflow( + workflow_type=WORKFLOW_TYPE, + workflow_id=f"python-split-{uuid4().hex}", + task_queue=args.queue, + input=[args.name, args.retry_once, args.activity_queue], + execution_timeout_seconds=60, + run_timeout_seconds=60, + ) + print(json.dumps({"event": "started", "workflow_id": handle.workflow_id, "run_id": handle.run_id})) + result = await handle.result(timeout=30, poll_interval=0.2) + print(json.dumps({"event": "completed", "result": result}), flush=True) + + +if __name__ == "__main__": + logging.basicConfig(level=logging.INFO) + asyncio.run(main(parse_args())) diff --git a/mkdocs.yml b/mkdocs.yml index 7e890dc6..5bb07e0e 100644 --- a/mkdocs.yml +++ b/mkdocs.yml @@ -77,6 +77,7 @@ nav: - Start here: index.md - SDK reference: sdk-reference.md - Sticky execution: guides/sticky-execution.md + - Separate worker processes: guides/separate-workers.md - Query deadlines: query-task-authority.md - SDK guide: https://durable-workflow.com/docs/2.0/polyglot/python/ - PyPI: https://pypi.org/project/durable-workflow/ From 2310f1dbfc5d8bbb046a55d6a25d0005ea1f7558 Mon Sep 17 00:00:00 2001 From: Durable Workflow Date: Mon, 19 Oct 2026 20:33:44 +0000 Subject: [PATCH 2/2] Exercise split worker example with real subprocesses --- .../test_separate_worker_example.py | 104 ++++++++++++++++++ 1 file changed, 104 insertions(+) create mode 100644 tests/integration/test_separate_worker_example.py diff --git a/tests/integration/test_separate_worker_example.py b/tests/integration/test_separate_worker_example.py new file mode 100644 index 00000000..ff2b4bfc --- /dev/null +++ b/tests/integration/test_separate_worker_example.py @@ -0,0 +1,104 @@ +"""Run the documented worker roles as independent processes against Server.""" +from __future__ import annotations + +import asyncio +import json +import os +import signal +import sys +import time +from pathlib import Path +from uuid import uuid4 + +import pytest + +from durable_workflow import Client + +EXAMPLE = Path(__file__).resolve().parents[2] / "examples" / "separate_workers.py" + + +@pytest.mark.asyncio +@pytest.mark.parametrize(("separate_queues", "retry_once"), [(False, False), (False, True), (True, True)]) +async def test_separate_workers_retry_and_deregister( + server_url: str, server_token: str, separate_queues: bool, retry_once: bool, +) -> None: + suffix = uuid4().hex[:12] + workflow_queue = f"split-workflow-{suffix}" + activity_queue = f"split-activity-{suffix}" if separate_queues else workflow_queue + worker_ids = [f"split-workflow-{suffix}", f"split-activity-{suffix}"] + environment = { + **os.environ, "SERVER_URL": server_url, "WORKFLOW_TOKEN": server_token, + "WORKFLOW_NAMESPACE": "default", "PYTHONUNBUFFERED": "1", + } + processes: list[asyncio.subprocess.Process] = [] + output: list[str] = [] + try: + for role, worker_id, queue in zip( + ("workflow", "activity"), worker_ids, (workflow_queue, activity_queue), strict=True, + ): + processes.append(await asyncio.create_subprocess_exec( + sys.executable, str(EXAMPLE), role, "--worker-id", worker_id, + "--queue", queue, "--slots", "1", env=environment, + stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.STDOUT, + )) + async with Client(server_url, token=server_token, namespace="default") as client: + deadline = time.monotonic() + 10 + while True: + roster = await client.list_workers() + workers = {worker.worker_id: worker for worker in roster.workers} + if set(worker_ids) <= workers.keys(): + break + assert all(process.returncode is None for process in processes), "worker exited during registration" + assert time.monotonic() < deadline, "separate worker registrations did not arrive" + await asyncio.sleep(0.1) + workflow_worker, activity_worker = (workers[worker_id] for worker_id in worker_ids) + assert workflow_worker.task_queue == workflow_queue + assert activity_worker.task_queue == activity_queue + assert workflow_worker.supported_workflow_types == ["examples.split-workers.greeter"] + assert workflow_worker.supported_activity_types == [] + assert activity_worker.supported_workflow_types == [] + assert activity_worker.supported_activity_types == ["examples.split-workers.greet"] + assert processes[0].pid != processes[1].pid + + arguments = ["--queue", workflow_queue, "--name", suffix] + if retry_once: + arguments.append("--retry-once") + if separate_queues: + arguments.extend(["--activity-queue", activity_queue]) + caller = await asyncio.create_subprocess_exec( + sys.executable, str(EXAMPLE), "start", *arguments, env=environment, + stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.STDOUT, + ) + try: + result, _ = await asyncio.wait_for(caller.communicate(), timeout=35) + finally: + if caller.returncode is None: + caller.kill() + await caller.wait() + assert caller.returncode == 0, result.decode()[-8000:] + events = [json.loads(line) for line in result.decode().splitlines() if line.startswith("{")] + assert events[-1] == {"event": "completed", "result": f"hello, {suffix}"} + finally: + for process in processes: + if process.returncode is None: + process.send_signal(signal.SIGTERM) + for process in processes: + try: + stdout, _ = await asyncio.wait_for(process.communicate(), timeout=15) + except asyncio.TimeoutError: + process.kill() + stdout, _ = await process.communicate() + output.append(stdout.decode()) + print(json.dumps({"separate_queues": separate_queues, "retry_once": retry_once, "worker_output": output}), + flush=True) + + assert all(process.returncode == 0 for process in processes), output + assert '"event": "activity_attempt"' not in output[0], output + activity_events = [json.loads(line) for line in output[1].splitlines() if line.startswith("{")] + attempts = [event for event in activity_events if event["event"] == "activity_attempt"] + assert [event["attempt"] for event in attempts] == ([1, 2] if retry_once else [1]), output + assert all(event["worker_id"] == worker_ids[1] for event in attempts), output + assert all(event["process_id"] == processes[1].pid for event in attempts), output + async with Client(server_url, token=server_token, namespace="default") as client: + remaining = await client.list_workers() + assert not (set(worker_ids) & {worker.worker_id for worker in remaining.workers})