[cross-repo from server#435] Conformance finding: namespaces coverage remains below full parity (#156)

Author: durable-workflow-ops

src/durable_workflow/worker.py

@@ -580,6 +580,30 @@ def _server_supports_query_tasks(info: dict[str, Any]) -> bool:
     return isinstance(capabilities, dict) and capabilities.get("query_tasks") is True
 
 
+def _server_long_poll_timeout(info: dict[str, Any]) -> float | None:
+    worker_protocol = info.get("worker_protocol")
+    if not isinstance(worker_protocol, dict):
+        return None
+
+    capabilities = worker_protocol.get("server_capabilities")
+    if not isinstance(capabilities, dict):
+        return None
+
+    timeout = capabilities.get("long_poll_timeout")
+    if isinstance(timeout, bool):
+        return None
+    if isinstance(timeout, (int, float)):
+        return float(timeout) if timeout > 0 else None
+    if isinstance(timeout, str):
+        try:
+            parsed = float(timeout)
+        except ValueError:
+            return None
+        return parsed if parsed > 0 else None
+
+    return None
+
+
 def _contract_version_matches(value: Any, expected: int) -> bool:
     if isinstance(value, int):
         return value == expected
@@ -635,6 +659,7 @@ def __init__(
             raise ValueError("heartbeat_interval must be positive")
 
         self._poll_timeout = poll_timeout
+        self._poll_http_timeout = poll_timeout
         self.max_concurrent_workflow_tasks = max_concurrent_workflow_tasks
         self.max_concurrent_activity_tasks = max_concurrent_activity_tasks
         self._stop = asyncio.Event()
@@ -757,6 +782,9 @@ async def _register(self) -> None:
 
         _validate_server_compatibility(info)
         self._query_tasks_supported = _server_supports_query_tasks(info)
+        server_long_poll_timeout = _server_long_poll_timeout(info)
+        if server_long_poll_timeout is not None:
+            self._poll_http_timeout = max(self._poll_http_timeout, server_long_poll_timeout + 5.0)
         log.debug(
             "server compatibility accepted: app_version=%s control_plane=%s worker_protocol=%s",
             info.get("version", "unknown"),
@@ -1536,7 +1564,7 @@ async def _poll_workflow_tasks(self) -> None:
                 task = await self.client.poll_workflow_task(
                     worker_id=self.worker_id,
                     task_queue=self.task_queue,
-                    timeout=self._poll_timeout,
+                    timeout=self._poll_http_timeout,
                 )
             except Exception as e:
                 self._wf_semaphore.release()
@@ -1613,7 +1641,7 @@ async def _poll_activity_tasks(self) -> None:
                 task = await self.client.poll_activity_task(
                     worker_id=self.worker_id,
                     task_queue=self.task_queue,
-                    timeout=self._poll_timeout,
+                    timeout=self._poll_http_timeout,
                 )
             except Exception as e:
                 self._act_semaphore.release()
@@ -1650,7 +1678,7 @@ async def _poll_query_tasks(self, *, client: Client | None = None, track_tasks:
                 task = await client.poll_query_task(
                     worker_id=self.worker_id,
                     task_queue=self.task_queue,
-                    timeout=self._poll_timeout,
+                    timeout=self._poll_http_timeout,
                 )
             except Exception as e:
                 self._record_poll_metrics("query", "error", time.perf_counter() - poll_start)
@@ -1739,7 +1767,7 @@ async def _stop_query_task_thread(self) -> None:
 
         await asyncio.to_thread(
             thread.join,
-            min(self._shutdown_timeout, self._poll_timeout + 1),
+            min(self._shutdown_timeout, self._poll_http_timeout + 1),
         )
 
     async def run(self) -> None:
@@ -1908,7 +1936,7 @@ async def run_until(
                     task = await self.client.poll_workflow_task(
                         worker_id=self.worker_id,
                         task_queue=self.task_queue,
-                        timeout=self._poll_timeout,
+                        timeout=self._poll_http_timeout,
                     )
                     self._record_poll_metrics(
                         "workflow",
@@ -1938,7 +1966,7 @@ async def run_until(
                 task = await self.client.poll_activity_task(
                     worker_id=self.worker_id,
                     task_queue=self.task_queue,
-                    timeout=self._poll_timeout,
+                    timeout=self._poll_http_timeout,
                 )
                 self._record_poll_metrics(
                     "activity",

tests/test_worker.py

@@ -257,6 +257,7 @@ def compatible_cluster_info(**overrides: object) -> dict[str, object]:
             "version": PROTOCOL_VERSION,
             "server_capabilities": {
                 "query_tasks": True,
+                "long_poll_timeout": 30,
             },
         },
     }
@@ -314,6 +315,39 @@ async def test_register(self, mock_client: AsyncMock) -> None:
         assert process_metrics["process_id"] > 0
         assert "process_started_at" in process_metrics
 
+    @pytest.mark.asyncio
+    async def test_register_keeps_http_timeout_above_server_long_poll(self, mock_client: AsyncMock) -> None:
+        mock_client.get_cluster_info = AsyncMock(
+            return_value=compatible_cluster_info(
+                worker_protocol={
+                    "version": PROTOCOL_VERSION,
+                    "server_capabilities": {
+                        "query_tasks": True,
+                        "long_poll_timeout": 12,
+                    },
+                }
+            )
+        )
+        worker = Worker(
+            mock_client,
+            task_queue="q1",
+            workflows=[TestWorkflow],
+            activities=[echo_activity],
+            worker_id="w-short-poll",
+            poll_timeout=0.01,
+        )
+
+        async def poll_once(**_: object) -> None:
+            worker._stop.set()
+            return None
+
+        mock_client.poll_workflow_task.side_effect = poll_once
+
+        await worker._register()
+        await worker._poll_workflow_tasks()
+
+        assert mock_client.poll_workflow_task.call_args.kwargs["timeout"] == 17.0
+
     @pytest.mark.asyncio
     async def test_register_omits_query_task_capability_when_server_does_not_support_it(
         self, mock_client: AsyncMock