[Backport release-1.19] fix(clients): give each sidecar health check attempt a timeout

dapr/aio/clients/health.py

@@ -18,6 +18,7 @@
 
 import aiohttp
 
+from dapr.clients.health import _attempt_timeout
 from dapr.clients.http.conf import DAPR_API_TOKEN_HEADER, DAPR_USER_AGENT, USER_AGENT_HEADER
 from dapr.clients.http.helpers import get_api_url
 from dapr.conf import settings
@@ -32,22 +33,31 @@ async def wait_for_sidecar():
             headers[DAPR_API_TOKEN_HEADER] = settings.DAPR_API_TOKEN
         timeout = float(settings.DAPR_HEALTH_TIMEOUT)
 
-        start = time.time()
+        start = time.monotonic()
         ssl_context = DaprHealth.get_ssl_context()
 
         connector = aiohttp.TCPConnector(ssl=ssl_context)
         async with aiohttp.ClientSession(connector=connector) as session:
             while True:
                 try:
-                    async with session.get(health_url, headers=headers) as response:
+                    # Cap each probe at the remaining budget instead of aiohttp's
+                    # 300 s default, so a sidecar that never answers cannot block
+                    # past DAPR_HEALTH_TIMEOUT.
+                    probe_timeout = aiohttp.ClientTimeout(total=_attempt_timeout(start + timeout))
+                    async with session.get(
+                        health_url, headers=headers, timeout=probe_timeout
+                    ) as response:
                         if 200 <= response.status < 300:
                             break
+                except asyncio.TimeoutError:
+                    # aiohttp's timeout error has an empty message; say what happened.
+                    print(f'Health check on {health_url} failed: timed out')
                 except aiohttp.ClientError as e:
                     print(f'Health check on {health_url} failed: {e}')
                 except Exception as e:
                     print(f'Unexpected error during health check: {e}')
 
-                remaining = (start + timeout) - time.time()
+                remaining = (start + timeout) - time.monotonic()
                 if remaining <= 0:
                     raise TimeoutError(f'Dapr health check timed out, after {timeout}.')
                 await asyncio.sleep(min(1, remaining))

dapr/clients/health.py

@@ -22,6 +22,16 @@
 from dapr.clients.http.helpers import get_api_url
 from dapr.conf import settings
 
+# Shortest time one health probe may wait for the sidecar. Each probe is otherwise capped
+# at what is left of DAPR_HEALTH_TIMEOUT; the floor lets the last probe still get a reply.
+_MIN_ATTEMPT_TIMEOUT_SECONDS = 0.5
+
+
+def _attempt_timeout(deadline: float) -> float:
+    """Seconds one health probe may take: what is left until ``deadline`` (a
+    ``time.monotonic()`` value), but never less than _MIN_ATTEMPT_TIMEOUT_SECONDS."""
+    return max(deadline - time.monotonic(), _MIN_ATTEMPT_TIMEOUT_SECONDS)
+
 
 class DaprHealth:
     @staticmethod
@@ -41,19 +51,32 @@ def wait_for_sidecar():
             headers[DAPR_API_TOKEN_HEADER] = settings.DAPR_API_TOKEN
         timeout = float(settings.DAPR_HEALTH_TIMEOUT)
 
-        start = time.time()
+        start = time.monotonic()
         while True:
             try:
                 req = urllib.request.Request(health_url, headers=headers)
-                with urllib.request.urlopen(req, context=DaprHealth.get_ssl_context()) as response:
+                # Without a timeout, a sidecar that accepts the connection but never
+                # answers would block here forever, past DAPR_HEALTH_TIMEOUT. urllib
+                # applies this per socket operation (connect, each read), not to the
+                # whole request, so a probe can overrun the budget somewhat, but it
+                # can no longer block forever.
+                with urllib.request.urlopen(
+                    req,
+                    timeout=_attempt_timeout(start + timeout),
+                    context=DaprHealth.get_ssl_context(),
+                ) as response:
                     if 200 <= response.status < 300:
                         break
             except urllib.error.URLError as e:
                 print(f'Health check on {health_url} failed: {e.reason}')
+            except TimeoutError:
+                # urllib wraps only connect/send errors in URLError; a read timeout in
+                # getresponse() (a sidecar that accepts but never answers) is bare.
+                print(f'Health check on {health_url} failed: timed out')
             except Exception as e:
                 print(f'Unexpected error during health check: {e}')
 
-            remaining = (start + timeout) - time.time()
+            remaining = (start + timeout) - time.monotonic()
             if remaining <= 0:
                 raise TimeoutError(f'Dapr health check timed out, after {timeout}.')
             time.sleep(min(1, remaining))

tests/clients/fake_servers.py

@@ -0,0 +1,27 @@
+# -*- coding: utf-8 -*-
+
+"""
+Copyright 2025 The Dapr Authors
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+    http://www.apache.org/licenses/LICENSE-2.0
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+"""
+
+import socket
+import unittest
+
+
+def listen_without_replying(test: unittest.TestCase) -> int:
+    """Open a local port that accepts connections (the OS completes the handshake from
+    the listen backlog) but never sends a reply, like a hung sidecar. Returns the port."""
+    listener = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+    test.addCleanup(listener.close)
+    listener.bind(('127.0.0.1', 0))
+    listener.listen(16)
+    return listener.getsockname()[1]

tests/clients/test_healthcheck.py

@@ -13,6 +13,7 @@
 limitations under the License.
 """
 
+import threading
 import time
 import unittest
 from unittest.mock import MagicMock, patch
@@ -21,6 +22,8 @@
 from dapr.conf import settings
 from dapr.version import __version__
 
+from .fake_servers import listen_without_replying
+
 
 class DaprHealthCheckTests(unittest.TestCase):
     @patch.object(settings, 'DAPR_HTTP_ENDPOINT', 'http://domain.com:3500')
@@ -75,3 +78,39 @@ def test_wait_for_sidecar_timeout(self, mock_urlopen):
 
         self.assertGreaterEqual(time.time() - start, 2.5)
         self.assertGreater(mock_urlopen.call_count, 1)
+
+
+class DaprHealthCheckUnresponsiveSidecarTests(unittest.TestCase):
+    # How long the test waits before calling wait_for_sidecar() hung. Well above
+    # DAPR_HEALTH_TIMEOUT so a slow CI machine does not fail it.
+    HANG_LIMIT_SECONDS = 15
+
+    @patch.object(settings, 'DAPR_HEALTH_TIMEOUT', '1')
+    def test_wait_for_sidecar_times_out_when_sidecar_never_replies(self):
+        port = listen_without_replying(self)
+        outcome = {}
+
+        def wait():
+            try:
+                DaprHealth.wait_for_sidecar()
+            except Exception as error:
+                outcome['error'] = error
+
+        with (
+            patch.object(settings, 'DAPR_HTTP_ENDPOINT', f'http://127.0.0.1:{port}'),
+            patch('builtins.print') as mock_print,
+        ):
+            # Run in a daemon thread so a regression fails this test instead of hanging
+            # the whole session.
+            waiter = threading.Thread(target=wait, daemon=True)
+            start = time.time()
+            waiter.start()
+            waiter.join(self.HANG_LIMIT_SECONDS)
+
+        self.assertFalse(waiter.is_alive(), 'wait_for_sidecar() blocked past DAPR_HEALTH_TIMEOUT')
+        self.assertIsInstance(outcome.get('error'), TimeoutError)
+        self.assertIn('Dapr health check timed out', str(outcome['error']))
+        self.assertLess(time.time() - start, self.HANG_LIMIT_SECONDS)
+        logged = [str(call.args[0]) for call in mock_print.call_args_list if call.args]
+        self.assertTrue(any(line.endswith('failed: timed out') for line in logged), logged)
+        self.assertFalse(any(line.startswith('Unexpected error') for line in logged), logged)

tests/clients/test_healthcheck_async.py

@@ -22,6 +22,8 @@
 from dapr.conf import settings
 from dapr.version import __version__
 
+from .fake_servers import listen_without_replying
+
 
 class DaprHealthCheckAsyncTests(unittest.IsolatedAsyncioTestCase):
     @patch.object(settings, 'DAPR_HTTP_ENDPOINT', 'http://domain.com:3500')
@@ -192,6 +194,27 @@ async def test_multiple_health_checks_concurrent(self, mock_get):
         # Verify multiple calls were made
         self.assertGreaterEqual(mock_get.call_count, 3)
 
+    @patch.object(settings, 'DAPR_HEALTH_TIMEOUT', '1')
+    async def test_wait_for_sidecar_times_out_when_sidecar_never_replies(self):
+        port = listen_without_replying(self)
+        hang_limit_seconds = 15  # well above DAPR_HEALTH_TIMEOUT, for slow CI machines
+
+        with (
+            patch.object(settings, 'DAPR_HTTP_ENDPOINT', f'http://127.0.0.1:{port}'),
+            patch('builtins.print') as mock_print,
+        ):
+            task = asyncio.ensure_future(DaprHealth.wait_for_sidecar())
+            done, _ = await asyncio.wait({task}, timeout=hang_limit_seconds)
+            if not done:
+                task.cancel()
+                self.fail('wait_for_sidecar() blocked past DAPR_HEALTH_TIMEOUT')
+
+        error = task.exception()
+        self.assertIsInstance(error, TimeoutError)
+        self.assertIn('Dapr health check timed out', str(error))
+        logged = [str(call.args[0]) for call in mock_print.call_args_list if call.args]
+        self.assertTrue(any(line.endswith('failed: timed out') for line in logged), logged)
+
 
 if __name__ == '__main__':
     unittest.main()