From f835c75385b50d34e1ebf02ba7bb6d14bdc3b764 Mon Sep 17 00:00:00 2001 From: Durable Workflow Date: Thu, 10 Sep 2026 00:36:23 +0000 Subject: [PATCH 1/2] fix: use published Server with precise schedule dispatch --- polyglot/qualified-artifact-tuple.json | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/polyglot/qualified-artifact-tuple.json b/polyglot/qualified-artifact-tuple.json index 2eb5545..2254fe2 100644 --- a/polyglot/qualified-artifact-tuple.json +++ b/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" } From 234cc92c9f41c2ede8450ee782447240cf65d0cc Mon Sep 17 00:00:00 2001 From: Durable Workflow Date: Thu, 10 Sep 2026 00:52:58 +0000 Subject: [PATCH 2/2] fix: keep polyglot worker alive during poll backpressure --- .github/workflows/polyglot-validation.yml | 6 +- polyglot/python_worker/activities.py | 26 +++++-- .../tests/test_native_binary_workers.py | 67 +++++++++++++++++++ scripts/polyglot.sh | 6 +- 4 files changed, 94 insertions(+), 11 deletions(-) diff --git a/.github/workflows/polyglot-validation.yml b/.github/workflows/polyglot-validation.yml index 56d28f3..7f844ca 100644 --- a/.github/workflows/polyglot-validation.yml +++ b/.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 diff --git a/polyglot/python_worker/activities.py b/polyglot/python_worker/activities.py index 149e9a2..ed72ff3 100644 --- a/polyglot/python_worker/activities.py +++ b/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 diff --git a/polyglot/python_worker/tests/test_native_binary_workers.py b/polyglot/python_worker/tests/test_native_binary_workers.py index bebf44d..35664f4 100644 --- a/polyglot/python_worker/tests/test_native_binary_workers.py +++ b/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]] = [ diff --git a/scripts/polyglot.sh b/scripts/polyglot.sh index 265f02d..6a4cebb 100755 --- a/scripts/polyglot.sh +++ b/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