fix: adopt Server 2.3.9 in the service-mode artifact tuple

.github/workflows/polyglot-validation.yml

@@ -25,11 +25,7 @@ jobs:
         run: scripts/polyglot.sh
       - name: Dump logs
         if: failure()
-        run: >-
-          docker compose --project-directory polyglot
-          -f polyglot/docker-compose.yml
-          -p "$POLYGLOT_COMPOSE_PROJECT_NAME"
-          logs --no-color --timestamps
+        run: scripts/polyglot.sh logs
       - name: Tear down stack
         if: always()
         run: scripts/polyglot.sh down

polyglot/python_worker/activities.py

@@ -18,6 +18,7 @@
 from typing import Any
 
 from durable_workflow import Client, TransportRetryPolicy, Worker, activity, serializer
+from durable_workflow.errors import ServerError
 
 TASK_QUEUE = os.environ.get("POLYGLOT_PHP2PY_TASK_QUEUE", "polyglot-php-to-python")
 POLL_TIMEOUT_SECONDS = float(os.environ.get("DURABLE_WORKFLOW_POLL_TIMEOUT_SECONDS", "90"))
@@ -166,11 +167,26 @@ async def run_typed_error_worker(client: Client, worker_id: str) -> None:
 
     try:
         while True:
-            task = await client.poll_activity_task(
-                worker_id=worker_id,
-                task_queue=TASK_QUEUE,
-                timeout=POLL_TIMEOUT_SECONDS,
-            )
+            try:
+                task = await client.poll_activity_task(
+                    worker_id=worker_id,
+                    task_queue=TASK_QUEUE,
+                    timeout=POLL_TIMEOUT_SECONDS,
+                )
+            except ServerError as exc:
+                body = exc.body if isinstance(exc.body, dict) else {}
+                delay = body.get("retry_after_seconds")
+                if (
+                    exc.status != 429
+                    or body.get("reason") != "long_poll_capacity_exhausted"
+                    or body.get("retryable") is not True
+                    or type(delay) is not int
+                    or delay <= 0
+                ):
+                    raise
+                LOG.info("typed-error poll wait capacity exhausted; retrying in %ss", delay)
+                await asyncio.sleep(delay)
+                continue
             if task is None:
                 continue
 

polyglot/python_worker/tests/test_native_binary_workers.py

@@ -8,6 +8,7 @@
 import unittest
 from pathlib import Path
 from typing import Any
+from unittest.mock import AsyncMock, patch
 
 
 class _Definitions:
@@ -33,6 +34,15 @@ def query(_name):  # type: ignore[no-untyped-def]
 durable_workflow.workflow = _Definitions
 durable_workflow_errors = types.ModuleType("durable_workflow.errors")
 durable_workflow_errors.ActivityFailed = Exception
+
+
+class ServerError(Exception):
+    def __init__(self, status: int, body: object) -> None:
+        self.status = status
+        self.body = body
+
+
+durable_workflow_errors.ServerError = ServerError
 sys.modules["durable_workflow"] = durable_workflow
 sys.modules["durable_workflow.errors"] = durable_workflow_errors
 
@@ -197,6 +207,63 @@ def test_native_binary_echo_rejects_a_partial_binary_fixture(self) -> None:
             )
 
 
+class TypedErrorPollCapacityTest(unittest.IsolatedAsyncioTestCase):
+    def refusal(self, **overrides: Any) -> dict[str, Any]:
+        return {
+            "reason": "long_poll_capacity_exhausted",
+            "retryable": True,
+            "retry_after_seconds": 2,
+            **overrides,
+        }
+
+    async def test_capacity_refusal_waits_then_handles_the_next_task(self) -> None:
+        task = {"task_id": "accepted-after-wait"}
+        client = types.SimpleNamespace(
+            register_worker=AsyncMock(return_value={}),
+            poll_activity_task=AsyncMock(side_effect=[
+                ServerError(429, self.refusal()), task, asyncio.CancelledError(),
+            ]),
+        )
+        with (
+            patch.object(activities, "heartbeat_typed_error_worker", AsyncMock()),
+            patch.object(activities, "handle_typed_error_task", AsyncMock()) as handler,
+            patch.object(activities.asyncio, "sleep", AsyncMock()) as sleep,
+        ):
+            with self.assertRaises(asyncio.CancelledError):
+                await activities.run_typed_error_worker(client, "typed-error-worker")
+
+        sleep.assert_awaited_once_with(2)
+        handler.assert_awaited_once_with(client, "typed-error-worker", task)
+        self.assertEqual(3, client.poll_activity_task.await_count)
+
+    async def test_unrelated_or_malformed_refusals_are_not_hidden(self) -> None:
+        cases = [
+            ServerError(401, self.refusal()),
+            ServerError(500, self.refusal()),
+            ServerError(429, self.refusal(reason="namespace_quota_exhausted")),
+            ServerError(429, self.refusal(retryable=False)),
+            ServerError(429, self.refusal(retry_after_seconds=None)),
+            ServerError(429, self.refusal(retry_after_seconds=True)),
+            ServerError(429, self.refusal(retry_after_seconds=-1)),
+            ServerError(429, self.refusal(retry_after_seconds=0)),
+            ServerError(429, "invalid response"),
+        ]
+        for error in cases:
+            with self.subTest(status=error.status, body=error.body):
+                client = types.SimpleNamespace(
+                    register_worker=AsyncMock(return_value={}),
+                    poll_activity_task=AsyncMock(side_effect=error),
+                )
+                with (
+                    patch.object(activities, "heartbeat_typed_error_worker", AsyncMock()),
+                    patch.object(activities.asyncio, "sleep", AsyncMock()) as sleep,
+                ):
+                    with self.assertRaises(ServerError) as raised:
+                        await activities.run_typed_error_worker(client, "typed-error-worker")
+                self.assertIs(error, raised.exception)
+                sleep.assert_not_awaited()
+
+
 class TypedErrorTaskCodecBoundaryTest(unittest.IsolatedAsyncioTestCase):
     async def test_rejects_every_non_avro_root_tag_before_decode_or_handler_work(self) -> None:
         codec_cases: list[tuple[str, bool, Any]] = [

polyglot/qualified-artifact-tuple.json

@@ -7,7 +7,7 @@
     "sdk-php": "2.0.0",
     "sdk-python": "2.0.0",
     "sdk-rust": "2.0.1",
-    "server": "2.0.0",
+    "server": "2.3.9",
     "waterline": "2.0.0",
     "workflow": "2.0.12"
   }

scripts/polyglot.sh

@@ -28,13 +28,17 @@ compose=(docker compose --project-directory "$repo_root/polyglot" -f "$compose_f
 
 case "${1:-}" in
   '') ;;
+  logs)
+    "${compose[@]}" logs --no-color --timestamps
+    exit 0
+    ;;
   down)
     printf '==> PolyglotWorkflow: removing Compose project %s\n' "$COMPOSE_PROJECT_NAME"
     "${compose[@]}" down --volumes --remove-orphans
     exit 0
     ;;
   *)
-    printf 'Usage: %s [down]\n' "${0##*/}" >&2
+    printf 'Usage: %s [logs|down]\n' "${0##*/}" >&2
     exit 2
     ;;
 esac