Merge pull request #94 from durable-workflow/fix/configured-http-timeout

Honor configured Client HTTP timeouts

Author: rmcdaniel

CHANGELOG.md

@@ -6,6 +6,12 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+## [2.3.9] - 2026-10-04
+
+### Fixed
+- Honor the configured Client HTTP timeout for ordinary control, discovery and
+  worker requests. Explicit long-poll timeouts retain their existing grace.
+
 ## [2.3.8] - 2026-10-03
 
 ### Fixed

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "durable-workflow"
-version = "2.3.8"
+version = "2.3.9"
 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.3.8"
-registry-version = "2.3.8"
+product-train = "2.3.9"
+registry-version = "2.3.9"
 supported-server-versions = "2.4.8"
 worker-protocol-version = "1.19"
 control-plane-version = "2"

src/durable_workflow/client.py

@@ -1794,7 +1794,7 @@ async def _do_request() -> httpx.Response:
                 f"/api{path}",
                 headers=self._discovery_headers() if discovery else self._headers(worker=worker),
                 json=request_json,
-                timeout=timeout,
+                timeout=self.timeout if timeout is None else timeout,
             )
             # Raise HTTPStatusError for 4xx/5xx so retry policy can catch it
             resp.raise_for_status()

tests/test_client_timeouts.py

@@ -0,0 +1,93 @@
+from __future__ import annotations
+
+import asyncio
+import json
+from collections.abc import AsyncIterator
+from contextlib import asynccontextmanager
+
+import httpx
+import pytest
+
+from durable_workflow.client import Client
+from durable_workflow.retry_policy import TransportRetryPolicy
+
+
+@asynccontextmanager
+async def response_server(*, delay: float | None) -> AsyncIterator[str]:
+    """A real HTTP peer, with response progress controlled by the test."""
+    writers: list[asyncio.StreamWriter] = []
+    handlers: set[asyncio.Task[None]] = set()
+
+    async def respond(reader: asyncio.StreamReader, writer: asyncio.StreamWriter) -> None:
+        task = asyncio.current_task()
+        assert task is not None
+        handlers.add(task)
+        writers.append(writer)
+        try:
+            headers = await reader.readuntil(b"\r\n\r\n")
+            for header in headers.split(b"\r\n"):
+                if header.lower().startswith(b"content-length:"):
+                    await reader.readexactly(int(header.split(b":", 1)[1]))
+            if delay is None:
+                await reader.read()
+                return
+            await asyncio.sleep(delay)
+            body = json.dumps({"task": None, "poll_status": "empty"}).encode()
+            writer.write(
+                b"HTTP/1.1 200 OK\r\nContent-Type: application/json\r\n"
+                + f"Content-Length: {len(body)}\r\nConnection: close\r\n\r\n".encode()
+                + body
+            )
+            await writer.drain()
+        finally:
+            writer.close()
+            await writer.wait_closed()
+            handlers.discard(task)
+
+    server = await asyncio.start_server(respond, "127.0.0.1", 0)
+    try:
+        assert server.sockets
+        yield f"http://127.0.0.1:{server.sockets[0].getsockname()[1]}"
+    finally:
+        server.close()
+        await server.wait_closed()
+        for writer in writers:
+            writer.close()
+        if handlers:
+            await asyncio.gather(*handlers)
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("operation", ["control", "discovery", "worker"])
+async def test_configured_timeout_bounds_a_stalled_response(operation: str) -> None:
+    async with (
+        response_server(delay=None) as endpoint,
+        Client(endpoint, timeout=0.05, retry_policy=TransportRetryPolicy(max_attempts=1)) as client,
+    ):
+        if operation == "control":
+            request = client.health()
+        elif operation == "discovery":
+            request = client.get_cluster_info()
+        else:
+            request = client.complete_workflow_task(
+                task_id="synthetic-timeout-task",
+                lease_owner="synthetic-timeout-worker",
+                workflow_task_attempt=1,
+                commands=[],
+            )
+        with pytest.raises(httpx.ReadTimeout):
+            await asyncio.wait_for(request, timeout=2)
+
+
+@pytest.mark.asyncio
+async def test_explicit_poll_timeout_overrides_a_shorter_client_default() -> None:
+    async with (
+        response_server(delay=0.1) as endpoint,
+        Client(endpoint, timeout=0.01, retry_policy=TransportRetryPolicy(max_attempts=1)) as client,
+    ):
+        task = await client.poll_workflow_task(
+            worker_id="synthetic-timeout-worker",
+            task_queue="synthetic-timeout-queue",
+            timeout=0,
+        )
+        assert task is None