[cross-repo from server#138] Task Queue Priority + Fairness: dispatch by priority and fair across workload classes (Temporal-parity) (#44)

Author: durable-workflow-ops

src/durable_workflow/client.py

@@ -1877,6 +1877,9 @@ async def start_workflow(
         duplicate_policy: str | None = None,
         memo: dict[str, Any] | None = None,
         search_attributes: dict[str, Any] | None = None,
+        priority: int | None = None,
+        fairness_key: str | None = None,
+        fairness_weight: int | None = None,
     ) -> WorkflowHandle:
         """Start a new workflow instance and return a handle bound to it.
 
@@ -1895,6 +1898,16 @@ async def start_workflow(
 
         ``memo`` and ``search_attributes`` attach operator-facing metadata to
         the instance; see the main docs site for the key/value rules.
+
+        ``priority`` is an integer in the range ``0..9`` (lower numbers run
+        first when workers on a shared task queue are saturated; default
+        ``5``). ``fairness_key`` tags the workload class — typically a
+        tenant id, team name, or workflow type — so dispatch on a shared
+        task queue can be rebalanced across declared classes under
+        contention; tasks without a key share one class. ``fairness_weight``
+        (``1..1000``, default ``1``) lets a class take a proportionally
+        larger share of dispatch slots versus other classes on the same
+        queue.
         """
         body: dict[str, Any] = {
             "workflow_id": workflow_id,
@@ -1921,6 +1934,12 @@ async def start_workflow(
                 task_queue=task_queue,
             )
             body["search_attributes"] = search_attributes
+        if priority is not None:
+            body["priority"] = priority
+        if fairness_key is not None:
+            body["fairness_key"] = fairness_key
+        if fairness_weight is not None:
+            body["fairness_weight"] = fairness_weight
         data = await self._request("POST", "/workflows", json=body, context=workflow_id)
         return WorkflowHandle(
             self,

tests/test_client.py

@@ -230,6 +230,49 @@ async def test_duplicate_raises(self, client: Client) -> None:
                 workflow_type="greeter", task_queue="q1", workflow_id="wf-1"
             )
 
+    @pytest.mark.asyncio
+    async def test_priority_and_fairness_are_forwarded_on_start(self, client: Client) -> None:
+        resp = _mock_response(201, {
+            "workflow_id": "wf-prio",
+            "run_id": "run-prio",
+            "workflow_type": "greeter",
+        })
+        with patch.object(client._http, "request", new_callable=AsyncMock, return_value=resp) as mock:
+            await client.start_workflow(
+                workflow_type="greeter",
+                task_queue="shared",
+                workflow_id="wf-prio",
+                priority=1,
+                fairness_key="tenant-a",
+                fairness_weight=3,
+            )
+
+        body = mock.call_args.kwargs.get("json") or mock.call_args[1].get("json")
+        assert body["priority"] == 1
+        assert body["fairness_key"] == "tenant-a"
+        assert body["fairness_weight"] == 3
+
+    @pytest.mark.asyncio
+    async def test_priority_and_fairness_are_omitted_when_unset(self, client: Client) -> None:
+        resp = _mock_response(201, {
+            "workflow_id": "wf-nopri",
+            "run_id": "run-nopri",
+            "workflow_type": "greeter",
+        })
+        with patch.object(client._http, "request", new_callable=AsyncMock, return_value=resp) as mock:
+            await client.start_workflow(
+                workflow_type="greeter",
+                task_queue="shared",
+                workflow_id="wf-nopri",
+            )
+
+        body = mock.call_args.kwargs.get("json") or mock.call_args[1].get("json")
+        # Server-version skew: callers that don't opt in must not send the fields
+        # so the server defaults (priority=5, no fairness key) take effect.
+        assert "priority" not in body
+        assert "fairness_key" not in body
+        assert "fairness_weight" not in body
+
 
 class TestDescribeWorkflow:
     @pytest.mark.asyncio