Document separate workflow and activity worker processes

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

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.

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

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()))

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/