diff --git a/CHANGELOG.md b/CHANGELOG.md index a4b55bf..a346fc6 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,12 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +## [2.5.12] - 2026-10-13 + +### Fixed +- Surface the original canonical workflow query timeout instead of automatically + admitting another query through the HTTP transport retry policy. + ## [2.5.11] - 2026-10-13 ### Fixed diff --git a/pyproject.toml b/pyproject.toml index a4568b3..85f6afa 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta" [project] name = "durable-workflow" -version = "2.5.11" +version = "2.5.12" description = "Python client and worker SDK for Durable Workflow Cloud and self-hosted Server" readme = "README.md" requires-python = ">=3.10" @@ -71,8 +71,8 @@ durable-workflow-replay-conformance = "durable_workflow.replay_conformance:main" durable-workflow-workflow-updates-conformance = "durable_workflow.workflow_updates_conformance:main" [tool.durable-workflow] -product-train = "2.5.11" -registry-version = "2.5.11" +product-train = "2.5.12" +registry-version = "2.5.12" supported-server-versions = "2.5.16" worker-protocol-version = "1.20" control-plane-version = "2" diff --git a/src/durable_workflow/client.py b/src/durable_workflow/client.py index 2adc367..126cd3f 100644 --- a/src/durable_workflow/client.py +++ b/src/durable_workflow/client.py @@ -4323,7 +4323,9 @@ async def query_workflow( Queries are synchronous and non-mutating. The server runs the named query handler inside the workflow process and returns the decoded result. Raises :class:`~durable_workflow.errors.QueryFailed` if the - query was rejected or the handler errored. + query was rejected, the handler errored, or the original query timed + out. A canonical query timeout is surfaced without automatically + admitting another query. """ await self._require_query_support() body: dict[str, Any] = {} diff --git a/src/durable_workflow/retry_policy.py b/src/durable_workflow/retry_policy.py index 78bbe71..dd3616a 100644 --- a/src/durable_workflow/retry_policy.py +++ b/src/durable_workflow/retry_policy.py @@ -17,6 +17,7 @@ import json import logging import random +import re from collections.abc import Awaitable, Callable from dataclasses import dataclass from typing import TypeVar @@ -34,6 +35,24 @@ ) +def _query_timeout_refusal(exc: httpx.HTTPStatusError) -> bool: + request = exc.request + if (exc.response.status_code != 504 or request.method != "POST" + or request.headers.get("X-Durable-Workflow-Control-Plane-Version") != "2" + or re.search(r"/api/workflows/[^/]+/query/[^/]+$", request.url.path) is None): + return False + try: + body = exc.response.json() + except ValueError: + return False + # This response ends an admitted query. Another POST creates a new task, + # rather than retrying the original task or its remaining caller budget. + reason = body.get("reason") if isinstance(body, dict) else None + return isinstance(reason, str) and reason in { + "query_worker_timeout", "query_worker_execution_timeout", "query_task_not_claimed", + } + + def _storage_refusal(exc: Exception) -> tuple[ServerError, str | None] | None: if not isinstance(exc, httpx.HTTPStatusError): return None @@ -194,6 +213,8 @@ class TransportRetryPolicy: :class:`durable_workflow.workflow.ChildWorkflowRetryPolicy`. Uses exponential backoff with jitter to avoid thundering herd. + Canonical query timeouts end the original query and are surfaced without + another query admission; unrelated transient HTTP failures still retry. """ max_attempts: int = 3 @@ -213,7 +234,7 @@ def should_retry(self, exc: Exception, attempt: int) -> bool: # Retry 5xx server errors and 429 rate limit if isinstance(exc, httpx.HTTPStatusError): - if _poll_capacity_refusal(exc): + if _query_timeout_refusal(exc) or _poll_capacity_refusal(exc): return False return exc.response.status_code >= 500 or exc.response.status_code == 429 diff --git a/tests/test_query_timeout_retry.py b/tests/test_query_timeout_retry.py new file mode 100644 index 0000000..81aa15d --- /dev/null +++ b/tests/test_query_timeout_retry.py @@ -0,0 +1,73 @@ +from __future__ import annotations + +import httpx +import pytest + +from durable_workflow import Client +from durable_workflow.errors import QueryFailed +from durable_workflow.retry_policy import TransportRetryPolicy + + +@pytest.mark.parametrize("protocol", ["1.20", "1.19"]) +@pytest.mark.parametrize("reason", [ + "query_worker_execution_timeout", "query_task_not_claimed", "query_worker_timeout", +]) +@pytest.mark.parametrize("prefix", ["", "/api/runtime/v1/namespaces/example"]) +async def test_query_exposes_original_canonical_timeout_without_another_admission( + monkeypatch: pytest.MonkeyPatch, protocol: str, reason: str, prefix: str, +) -> None: + monkeypatch.setenv("DURABLE_WORKFLOW_WORKER_PROTOCOL_VERSION", protocol) + original = {"reason": reason, "message": "Original query ended.", + "control_plane": {"operation": "query", "reason": reason}} + admissions = 0 + discovery_protocols = [] + + def handle(request: httpx.Request) -> httpx.Response: + nonlocal admissions + if request.url.path.endswith("/api/cluster/info"): + discovery_protocols.append(request.headers["X-Durable-Workflow-Protocol-Version"]) + return httpx.Response(200, json={ + "worker_protocol": {"server_capabilities": {"query_tasks": True}}, + }) + assert request.url.path == prefix + "/api/workflows/original/query/total" + assert request.headers["X-Durable-Workflow-Control-Plane-Version"] == "2" + admissions += 1 + if admissions == 1: + return httpx.Response(504, json=original) + # The old policy hides the original failure behind a distinct admission. + return httpx.Response(200, json={"result": "replacement-query"}) + + async with Client("https://runtime.example" + prefix, worker_token="worker", control_token="control", + retry_policy=TransportRetryPolicy(initial_backoff_seconds=0, jitter=False)) as client: + await client._http.aclose() + client._http = httpx.AsyncClient(base_url=client.base_url, transport=httpx.MockTransport(handle)) + with pytest.raises(QueryFailed) as failure: + await client.query_workflow("original", "total") + assert admissions == 1 + assert discovery_protocols == [protocol] + assert failure.value.status == 504 and failure.value.reason == reason + assert failure.value.body == original + + +@pytest.mark.parametrize(("method", "path", "status", "body", "headers"), [ + ("POST", "/api/workflows/w/query/q", 504, {"reason": "gateway_timeout"}, + {"X-Durable-Workflow-Control-Plane-Version": "2"}), + ("POST", "/api/workflows/w/query/q", 503, {"reason": "query_worker_timeout"}, + {"X-Durable-Workflow-Control-Plane-Version": "2"}), + ("GET", "/api/workflows/w/query/q", 504, {"reason": "query_worker_timeout"}, + {"X-Durable-Workflow-Control-Plane-Version": "2"}), + ("POST", "/api/workflows/w/signals/q", 504, {"reason": "query_worker_timeout"}, + {"X-Durable-Workflow-Control-Plane-Version": "2"}), + ("POST", "/api/workflows/w/query/q", 504, {"reason": "query_worker_timeout"}, {}), + ("POST", "/api/workflows/w/query/q", 504, [], + {"X-Durable-Workflow-Control-Plane-Version": "2"}), + ("POST", "/api/workflows/w/query/q", 504, {"reason": []}, + {"X-Durable-Workflow-Control-Plane-Version": "2"}), +]) +def test_other_transport_failures_keep_existing_retry_policy( + method: str, path: str, status: int, body: object, headers: dict[str, str], +) -> None: + request = httpx.Request(method, "https://runtime.example" + path, headers=headers) + response = httpx.Response(status, request=request, json=body) + error = httpx.HTTPStatusError("transport failure", request=request, response=response) + assert TransportRetryPolicy().should_retry(error, 0) is True