From 1ea8447e5b7dd6f77f1c9cf590f572fb63b1c45c Mon Sep 17 00:00:00 2001 From: Casper Nielsen Date: Mon, 28 Sep 2026 13:04:15 +0200 Subject: [PATCH 1/6] fix(clients): reconnect configuration subscriptions after the stream breaks ConfigurationWatcher read the SubscribeConfigurationAlpha1 stream once and stopped on the first error, so a daprd restart ended the subscription for good. The watcher now re-subscribes with exponential backoff (0.5s doubling to 8s, with jitter) when the stream fails with a retryable error or is closed by the sidecar, and stops on INVALID_ARGUMENT, NOT_FOUND, PERMISSION_DENIED, UNAUTHENTICATED and UNIMPLEMENTED. The clients keep the watchers so unsubscribe_configuration sends the id of the current stream and stops the reconnect loop; close() stops them too. The async client handed its grpc.aio stub to the thread-based watcher, which could never read the stream, so subscribe_configuration always returned None after blocking the event loop for 5s. It now uses an asyncio-based watcher with the same reconnect behaviour. Closes #1232 Co-Authored-By: Claude Opus 5.5 Signed-off-by: Casper Nielsen --- dapr/aio/clients/grpc/_response.py | 197 +++++++++++- dapr/aio/clients/grpc/client.py | 37 ++- dapr/clients/grpc/_response.py | 221 ++++++++++++-- dapr/clients/grpc/client.py | 39 ++- tests/clients/fake_dapr_server.py | 40 +++ tests/clients/test_configuration_watcher.py | 285 ++++++++++++++++++ .../test_configuration_watcher_async.py | 214 +++++++++++++ tests/clients/test_dapr_grpc_client_async.py | 6 +- 8 files changed, 1002 insertions(+), 37 deletions(-) create mode 100644 tests/clients/test_configuration_watcher.py create mode 100644 tests/clients/test_configuration_watcher_async.py diff --git a/dapr/aio/clients/grpc/_response.py b/dapr/aio/clients/grpc/_response.py index 5380ede6d..3f277e370 100644 --- a/dapr/aio/clients/grpc/_response.py +++ b/dapr/aio/clients/grpc/_response.py @@ -13,10 +13,22 @@ limitations under the License. """ -from typing import AsyncGenerator, Generic +import asyncio +import contextlib +import logging +from typing import AsyncGenerator, Callable, Dict, Generic, List, Optional, Text -from dapr.clients.grpc._response import DaprResponse, TCryptoResponse -from dapr.proto import api_v1 +from dapr.clients.grpc._response import ( + CONFIG_SUBSCRIBE_TIMEOUT_SECONDS, + ConfigurationResponse, + DaprResponse, + TCryptoResponse, + config_reconnect_delay, + is_retryable_config_error, +) +from dapr.proto import api_service_v1, api_v1 + +logger = logging.getLogger(__name__) class CryptoResponse(DaprResponse, Generic[TCryptoResponse]): @@ -87,3 +99,182 @@ class EncryptResponse(CryptoResponse[api_v1.EncryptResponse]): ... class DecryptResponse(CryptoResponse[api_v1.DecryptResponse]): ... + + +class AsyncConfigurationWatcher: + """Reads a SubscribeConfigurationAlpha1 stream in an asyncio task and calls the handler + for every update. + + This is the asyncio counterpart of ConfigurationWatcher: it re-subscribes with exponential + backoff when the stream fails with a retryable error or is closed by the sidecar, until + stop() is called or a non-retryable error is returned. ``subscription_id`` is the id of the + first stream; ``id`` is the id of the current stream. + """ + + def __init__(self) -> None: + self.store_name: Optional[str] = None + self.keys: Optional[List[str]] = None + self.id: str = '' + self.subscription_id: str = '' + self._ready = asyncio.Event() + self._stopping = False + self._call = None + self._task: Optional[asyncio.Task] = None + self._stream_established = False + + async def watch_configuration( + self, + stub: api_service_v1.DaprStub, + store_name: str, + keys: List[str], + handler: Callable[[Text, ConfigurationResponse], None], + config_metadata: Optional[Dict[str, str]] = None, + ): + """Starts the watcher task and returns the subscription id, or None if the sidecar did + not return one within CONFIG_SUBSCRIBE_TIMEOUT_SECONDS or rejected the subscription.""" + req = api_v1.SubscribeConfigurationRequest( + store_name=store_name, keys=keys, metadata=config_metadata or {} + ) + self.keys = keys + self.store_name = store_name + self._task = asyncio.create_task( + self._read_subscribe_config(stub, req, handler), + name=f'dapr-configuration-watcher-{store_name}', + ) + try: + await asyncio.wait_for(self._ready.wait(), timeout=CONFIG_SUBSCRIBE_TIMEOUT_SECONDS) + except asyncio.TimeoutError: + logger.warning( + 'Unable to get configuration subscription id for keys %s on store %s', + self.keys, + self.store_name, + ) + if not self.subscription_id: + await self.stop() + return None + return self.subscription_id + + @property + def stopped(self) -> bool: + return self._stopping + + def request_stop(self) -> None: + """Marks the watcher as stopping so it does not reconnect when the stream ends.""" + self._stopping = True + + async def stop(self) -> None: + """Stops reconnecting, cancels the current stream and waits for the task to exit.""" + self._stopping = True + call = self._call + cancel = getattr(call, 'cancel', None) + if callable(cancel): + cancel() + task = self._task + if task is None or task.done() or task is asyncio.current_task(): + return + task.cancel() + with contextlib.suppress(asyncio.CancelledError): + await task + + async def _read_subscribe_config( + self, + stub: api_service_v1.DaprStub, + req: api_v1.SubscribeConfigurationRequest, + handler: Callable[[Text, ConfigurationResponse], None], + ) -> None: + attempt = 0 + try: + while not self._stopping: + self._stream_established = False + try: + await self._consume_stream(stub, req, handler) + if not self._stopping: + logger.info( + 'Configuration subscription stream for keys %s on store %s was ' + 'closed by the sidecar, reconnecting', + self.keys, + self.store_name, + ) + except asyncio.CancelledError: + raise + except Exception as error: + if self._stopping: + break + if not is_retryable_config_error(error): + logger.error( + 'Configuration subscription for keys %s on store %s stopped: %s', + self.keys, + self.store_name, + error, + ) + break + logger.warning( + 'Configuration subscription stream for keys %s on store %s failed, ' + 'reconnecting: %s', + self.keys, + self.store_name, + error, + ) + finally: + self._call = None + if self._stopping: + break + if self._stream_established: + attempt = 0 + delay = config_reconnect_delay(attempt) + attempt += 1 + await self._wait_before_retry(delay) + finally: + # Unblock watch_configuration if the first stream never delivered an id. + self._ready.set() + logger.debug( + 'Configuration watcher for keys %s on store %s exited', self.keys, self.store_name + ) + + async def _consume_stream( + self, + stub: api_service_v1.DaprStub, + req: api_v1.SubscribeConfigurationRequest, + handler: Callable[[Text, ConfigurationResponse], None], + ) -> None: + call = stub.SubscribeConfigurationAlpha1(req) + self._call = call + async for response in call: + if not self._stream_established: + self._on_stream_established(response.id) + if self._stopping: + return + if len(response.items) > 0: + self._deliver(handler, ConfigurationResponse(response.items)) + + def _on_stream_established(self, server_id: str) -> None: + self.id = server_id + if not self.subscription_id: + self.subscription_id = server_id + else: + logger.info( + 'Configuration subscription %s for keys %s on store %s reconnected with new id %s', + self.subscription_id, + self.keys, + self.store_name, + server_id, + ) + self._stream_established = True + self._ready.set() + + def _deliver( + self, + handler: Callable[[Text, ConfigurationResponse], None], + response: ConfigurationResponse, + ) -> None: + try: + handler(self.subscription_id, response) + except Exception: + logger.exception( + 'Configuration handler for keys %s on store %s raised an exception', + self.keys, + self.store_name, + ) + + async def _wait_before_retry(self, delay: float) -> None: + await asyncio.sleep(delay) diff --git a/dapr/aio/clients/grpc/client.py b/dapr/aio/clients/grpc/client.py index fae40a3e9..7d21a839b 100644 --- a/dapr/aio/clients/grpc/client.py +++ b/dapr/aio/clients/grpc/client.py @@ -16,7 +16,7 @@ import asyncio import socket import time -from typing import Any, Awaitable, Callable, Dict, List, Optional, Sequence, Text, Union +from typing import Any, Awaitable, Callable, Dict, List, Optional, Sequence, Text, Tuple, Union from urllib.parse import urlencode from warnings import warn @@ -41,6 +41,7 @@ EncryptRequestIterator, ) from dapr.aio.clients.grpc._response import ( + AsyncConfigurationWatcher, DecryptResponse, EncryptResponse, ) @@ -72,7 +73,6 @@ BulkStateItem, BulkStatesResponse, ConfigurationResponse, - ConfigurationWatcher, DaprResponse, GetBulkSecretResponse, GetMetadataResponse, @@ -144,6 +144,8 @@ def __init__( ``DAPR_GRPC_MAX_INBOUND_MESSAGE_SIZE_BYTES`` is consulted to set the receive limit (matches the Java SDK property of the same name). """ + # Active configuration subscriptions, keyed by (store_name, id returned to the caller). + self._config_watchers: Dict[Tuple[str, str], AsyncConfigurationWatcher] = {} DaprHealth.wait_for_sidecar() self.retry_policy = retry_policy or RetryPolicy() @@ -164,6 +166,10 @@ def get_credentials(): async def close(self): """Closes Dapr runtime gRPC channel.""" + watchers = list(getattr(self, '_config_watchers', {}).values()) + self._config_watchers = {} + for watcher in watchers: + await watcher.stop() if hasattr(self, '_channel') and self._channel: await self._channel.close() @@ -1239,15 +1245,21 @@ async def subscribe_configuration( if not store_name or len(store_name) == 0 or len(store_name.strip()) == 0: raise ValueError('Config store name cannot be empty to get the configuration') - configWatcher = ConfigurationWatcher() - id = configWatcher.watch_configuration( + configWatcher = AsyncConfigurationWatcher() + id = await configWatcher.watch_configuration( self._stub, store_name, keys, handler, config_metadata ) + if id: + self._config_watchers[(store_name, id)] = configWatcher return id async def unsubscribe_configuration(self, store_name: str, id: str) -> bool: """Unsubscribes from configuration changes. + If the subscription was re-established after the stream broke (for example after a + sidecar restart), the sidecar knows it under a new id; that id is used here, so the id + returned by subscribe_configuration stays valid. + Args: store_name (str): the state store name to unsubscribe from id (str): the subscription id to unsubscribe @@ -1255,10 +1267,19 @@ async def unsubscribe_configuration(self, store_name: str, id: str) -> bool: Returns: bool: True if unsubscribed successfully, False otherwise """ - req = api_v1.UnsubscribeConfigurationRequest(store_name=store_name, id=id) - response: api_v1.UnsubscribeConfigurationResponse = ( - await self._stub.UnsubscribeConfiguration(req) - ) + watcher = self._config_watchers.pop((store_name, id), None) + server_id = id + if watcher is not None: + watcher.request_stop() + server_id = watcher.id or id + try: + req = api_v1.UnsubscribeConfigurationRequest(store_name=store_name, id=server_id) + response: api_v1.UnsubscribeConfigurationResponse = ( + await self._stub.UnsubscribeConfiguration(req) + ) + finally: + if watcher is not None: + await watcher.stop() return response.ok async def try_lock( diff --git a/dapr/clients/grpc/_response.py b/dapr/clients/grpc/_response.py index c18794bfc..259533ae1 100644 --- a/dapr/clients/grpc/_response.py +++ b/dapr/clients/grpc/_response.py @@ -17,6 +17,8 @@ import contextlib import json +import logging +import random import threading from enum import Enum from typing import ( @@ -37,6 +39,7 @@ from google.protobuf.any_pb2 import Any as GrpcAny from google.protobuf.message import Message as GrpcMessage +from grpc import RpcError, StatusCode # type: ignore from dapr.clients._constants import DEFAULT_JSON_CONTENT_TYPE from dapr.clients.grpc._helpers import ( @@ -54,6 +57,7 @@ if TYPE_CHECKING: from dapr.clients.grpc.client import DaprGrpcClient +logger = logging.getLogger(__name__) TCryptoResponse = TypeVar( 'TCryptoResponse', bound=Union[api_v1.EncryptResponse, api_v1.DecryptResponse] @@ -669,12 +673,69 @@ def items(self) -> Dict[Text, ConfigurationItem]: return self._items +# How long subscribe_configuration waits for the first response (which carries the id). +CONFIG_SUBSCRIBE_TIMEOUT_SECONDS = 5.0 +# Reconnect backoff for configuration subscription streams: starts at the initial delay and +# doubles on each consecutive failure, up to the max, plus up to 20% random jitter. +CONFIG_RECONNECT_INITIAL_BACKOFF_SECONDS = 0.5 +CONFIG_RECONNECT_MAX_BACKOFF_SECONDS = 8.0 +_CONFIG_RECONNECT_JITTER_RATIO = 0.2 +# How long stop() waits for the watcher thread to exit after cancelling the stream. +_CONFIG_WATCHER_JOIN_TIMEOUT_SECONDS = 2.0 +# Errors that will not go away by resubscribing (bad request, missing store, auth). +_CONFIG_NON_RETRYABLE_CODES = frozenset( + { + StatusCode.INVALID_ARGUMENT, + StatusCode.NOT_FOUND, + StatusCode.PERMISSION_DENIED, + StatusCode.UNAUTHENTICATED, + StatusCode.UNIMPLEMENTED, + } +) + + +def config_reconnect_delay(attempt: int) -> float: + """Returns the backoff delay in seconds before reconnect attempt ``attempt`` (0-based).""" + base = min( + CONFIG_RECONNECT_INITIAL_BACKOFF_SECONDS * (2**attempt), + CONFIG_RECONNECT_MAX_BACKOFF_SECONDS, + ) + return base + random.uniform(0, base * _CONFIG_RECONNECT_JITTER_RATIO) + + +def is_retryable_config_error(error: BaseException) -> bool: + """Returns True if a configuration subscription stream should be re-established after + ``error``. Only gRPC errors are retried, except the ones in _CONFIG_NON_RETRYABLE_CODES.""" + if not isinstance(error, RpcError): + return False + code = error.code() if callable(getattr(error, 'code', None)) else None + return code not in _CONFIG_NON_RETRYABLE_CODES + + class ConfigurationWatcher: + """Reads a SubscribeConfigurationAlpha1 stream on a daemon thread and calls the handler + for every update. + + If the stream fails with a retryable error or is closed by the sidecar (for example when + daprd restarts), the watcher subscribes again with the same request, with exponential + backoff, until stop() is called or a non-retryable error is returned. + + Each new stream gets a new id from the sidecar. ``subscription_id`` is the id of the first + stream and stays the same; ``id`` is the id of the current stream and is the one to pass to + UnsubscribeConfiguration. + """ + def __init__(self): self.store_name = None self.keys = None self.event: threading.Event = threading.Event() self.id: str = '' + self.subscription_id: str = '' + self._stop_event = threading.Event() + self._lock = threading.Lock() + self._call = None + self._thread: Optional[threading.Thread] = None + self._stream_established = False def watch_configuration( self, @@ -687,38 +748,158 @@ def watch_configuration( req = api_v1.SubscribeConfigurationRequest( store_name=store_name, keys=keys, metadata=config_metadata ) - thread = threading.Thread(target=self._read_subscribe_config, args=(stub, req, handler)) - thread.daemon = True - thread.start() self.keys = keys self.store_name = store_name - check = self.event.wait(timeout=5) - if not check: - print(f'Unable to get configuration id for keys {self.keys}') + self._thread = threading.Thread( + target=self._read_subscribe_config, + args=(stub, req, handler), + name=f'dapr-configuration-watcher-{store_name}', + daemon=True, + ) + self._thread.start() + check = self.event.wait(timeout=CONFIG_SUBSCRIBE_TIMEOUT_SECONDS) + if not check or not self.subscription_id: + if not check: + logger.warning( + 'Unable to get configuration subscription id for keys %s on store %s', + self.keys, + self.store_name, + ) + self.stop() return None - return self.id + return self.subscription_id + + @property + def stopped(self) -> bool: + return self._stop_event.is_set() + + def request_stop(self) -> None: + """Marks the watcher as stopping so it does not reconnect when the stream ends.""" + self._stop_event.set() + + def stop(self) -> None: + """Stops reconnecting, cancels the current stream and waits briefly for the thread.""" + self._stop_event.set() + with self._lock: + call = self._call + cancel = getattr(call, 'cancel', None) + if callable(cancel): + cancel() + thread = self._thread + if thread is not None and thread is not threading.current_thread(): + thread.join(timeout=_CONFIG_WATCHER_JOIN_TIMEOUT_SECONDS) def _read_subscribe_config( self, stub: api_service_v1.DaprStub, req: api_v1.SubscribeConfigurationRequest, handler: Callable[[Text, ConfigurationResponse], None], - ): + ) -> None: + attempt = 0 try: - responses: List[api_v1.SubscribeConfigurationResponse] = ( - stub.SubscribeConfigurationAlpha1(req) + while not self._stop_event.is_set(): + self._stream_established = False + try: + self._consume_stream(stub, req, handler) + if not self._stop_event.is_set(): + logger.info( + 'Configuration subscription stream for keys %s on store %s was ' + 'closed by the sidecar, reconnecting', + self.keys, + self.store_name, + ) + except Exception as error: + if self._stop_event.is_set(): + break + if not is_retryable_config_error(error): + logger.error( + 'Configuration subscription for keys %s on store %s stopped: %s', + self.keys, + self.store_name, + error, + ) + break + logger.warning( + 'Configuration subscription stream for keys %s on store %s failed, ' + 'reconnecting: %s', + self.keys, + self.store_name, + error, + ) + finally: + with self._lock: + self._call = None + if self._stop_event.is_set(): + break + if self._stream_established: + attempt = 0 + delay = config_reconnect_delay(attempt) + attempt += 1 + if self._wait_before_retry(delay): + break + finally: + # Unblock watch_configuration if the first stream never delivered an id. + self.event.set() + logger.debug( + 'Configuration watcher for keys %s on store %s exited', self.keys, self.store_name ) - isFirst = True - for response in responses: - if isFirst: - self.id = response.id - self.event.set() - isFirst = False - if len(response.items) > 0: - handler(response.id, ConfigurationResponse(response.items)) + + def _consume_stream( + self, + stub: api_service_v1.DaprStub, + req: api_v1.SubscribeConfigurationRequest, + handler: Callable[[Text, ConfigurationResponse], None], + ) -> None: + call = stub.SubscribeConfigurationAlpha1(req) + with self._lock: + if self._stop_event.is_set(): + cancel = getattr(call, 'cancel', None) + if callable(cancel): + cancel() + return + self._call = call + for response in call: + if not self._stream_established: + self._on_stream_established(response.id) + if self._stop_event.is_set(): + return + if len(response.items) > 0: + self._deliver(handler, ConfigurationResponse(response.items)) + + def _on_stream_established(self, server_id: str) -> None: + with self._lock: + self.id = server_id + if not self.subscription_id: + self.subscription_id = server_id + else: + logger.info( + 'Configuration subscription %s for keys %s on store %s reconnected ' + 'with new id %s', + self.subscription_id, + self.keys, + self.store_name, + server_id, + ) + self._stream_established = True + self.event.set() + + def _deliver( + self, + handler: Callable[[Text, ConfigurationResponse], None], + response: ConfigurationResponse, + ) -> None: + try: + handler(self.subscription_id, response) except Exception: - print(f'{self.store_name} configuration watcher for keys {self.keys} stopped.') - pass + logger.exception( + 'Configuration handler for keys %s on store %s raised an exception', + self.keys, + self.store_name, + ) + + def _wait_before_retry(self, delay: float) -> bool: + """Sleeps for ``delay`` seconds. Returns True if the watcher was stopped meanwhile.""" + return self._stop_event.wait(delay) class BulkPublishResponseFailedEntry: diff --git a/dapr/clients/grpc/client.py b/dapr/clients/grpc/client.py index b372bf396..d43750b92 100644 --- a/dapr/clients/grpc/client.py +++ b/dapr/clients/grpc/client.py @@ -16,7 +16,7 @@ import socket import threading import time -from typing import Any, Callable, Dict, List, Optional, Sequence, Text, Union +from typing import Any, Callable, Dict, List, Optional, Sequence, Text, Tuple, Union from urllib.parse import urlencode from warnings import warn @@ -145,6 +145,9 @@ def __init__( receive limit (matches the Java SDK property of the same name). retry_policy (RetryPolicy optional): Specifies retry behaviour """ + # Active configuration subscriptions, keyed by (store_name, id returned to the caller). + self._config_watchers: Dict[Tuple[str, str], ConfigurationWatcher] = {} + self._config_watchers_lock = threading.Lock() DaprHealth.wait_for_sidecar() self.retry_policy = retry_policy or RetryPolicy() @@ -211,9 +214,20 @@ def get_credentials(): def close(self): """Closes Dapr runtime gRPC channel.""" + self._stop_config_watchers() if hasattr(self, '_channel') and self._channel: self._channel.close() + def _stop_config_watchers(self) -> None: + lock = getattr(self, '_config_watchers_lock', None) + if lock is None: + return + with lock: + watchers = list(self._config_watchers.values()) + self._config_watchers = {} + for watcher in watchers: + watcher.stop() + def __del__(self): self.close() @@ -1304,11 +1318,18 @@ def subscribe_configuration( id = configWatcher.watch_configuration( self._stub, store_name, keys, handler, config_metadata ) + if id: + with self._config_watchers_lock: + self._config_watchers[(store_name, id)] = configWatcher return id def unsubscribe_configuration(self, store_name: str, id: str) -> bool: """Unsubscribes from configuration changes. + If the subscription was re-established after the stream broke (for example after a + sidecar restart), the sidecar knows it under a new id; that id is used here, so the id + returned by subscribe_configuration stays valid. + Args: store_name (str): the state store name to unsubscribe from id (str): the subscription id to unsubscribe @@ -1316,8 +1337,20 @@ def unsubscribe_configuration(self, store_name: str, id: str) -> bool: Returns: bool: True if unsubscribed successfully, False otherwise """ - req = api_v1.UnsubscribeConfigurationRequest(store_name=store_name, id=id) - response: api_v1.UnsubscribeConfigurationResponse = self._stub.UnsubscribeConfiguration(req) + with self._config_watchers_lock: + watcher = self._config_watchers.pop((store_name, id), None) + server_id = id + if watcher is not None: + watcher.request_stop() + server_id = watcher.id or id + try: + req = api_v1.UnsubscribeConfigurationRequest(store_name=store_name, id=server_id) + response: api_v1.UnsubscribeConfigurationResponse = self._stub.UnsubscribeConfiguration( + req + ) + finally: + if watcher is not None: + watcher.stop() return response.ok def try_lock( diff --git a/tests/clients/fake_dapr_server.py b/tests/clients/fake_dapr_server.py index 27479e9a5..45da06846 100644 --- a/tests/clients/fake_dapr_server.py +++ b/tests/clients/fake_dapr_server.py @@ -14,6 +14,8 @@ """ import json +import threading +import time from concurrent import futures from typing import Any, Dict, List, Optional, Tuple @@ -67,6 +69,17 @@ def __init__(self, grpc_port: int = 50001, http_port: int = 8080): self.actor_stream_plans: List[Dict[str, Any]] = [] self.actor_stream_initials: List[api_v1.SubscribeActorEventsRequestInitialAlpha1] = [] self.actor_stream_replies: List[api_v1.SubscribeActorEventsRequestAlpha1] = [] + # SubscribeConfigurationAlpha1: one plan is consumed per stream connection. Each plan is a + # dict with 'id' (subscription id sent in the first response), 'updates' (list of + # {key: value} dicts pushed after the first response), and 'end' ('abort' fails the + # stream with 'code' (default UNAVAILABLE), 'eof' closes cleanly, 'hold' keeps the stream + # open until the id is unsubscribed or the client goes away). 'reject' fails the call with + # that status code before any response is sent. + self.config_stream_plans: List[Dict[str, Any]] = [] + self.config_subscribe_requests: List[api_v1.SubscribeConfigurationRequest] = [] + self.config_unsubscribe_requests: List[api_v1.UnsubscribeConfigurationRequest] = [] + self._config_unsubscribed: set = set() + self._config_lock = threading.Lock() def set_bulk_publish_unimplemented_on_stable_next(self) -> None: """Make the next BulkPublishEvent (stable) call return UNIMPLEMENTED. @@ -498,7 +511,34 @@ def SubscribeConfiguration(self, request, context): responses.append(response) return api_v1.SubscribeConfigurationResponse(responses=responses) + def SubscribeConfigurationAlpha1(self, request, context): + with self._config_lock: + self.config_subscribe_requests.append(request) + plan = self.config_stream_plans.pop(0) if self.config_stream_plans else {} + if plan.get('reject') is not None: + context.abort(plan['reject'], 'rejected by test plan') + sub_id = plan.get('id', 'sub-id') + yield api_v1.SubscribeConfigurationResponse(id=sub_id) + for update in plan.get('updates', []): + items = { + key: ConfigurationItem(value=value, version='1') for key, value in update.items() + } + yield api_v1.SubscribeConfigurationResponse(id=sub_id, items=items) + + end_behavior = plan.get('end', 'hold') + if end_behavior == 'abort': + context.abort(plan.get('code', grpc.StatusCode.UNAVAILABLE), 'simulated disconnection') + elif end_behavior == 'hold': + while context.is_active(): + with self._config_lock: + if sub_id in self._config_unsubscribed: + return + time.sleep(0.01) + def UnsubscribeConfiguration(self, request, context): + with self._config_lock: + self.config_unsubscribe_requests.append(request) + self._config_unsubscribed.add(request.id) return api_v1.UnsubscribeConfigurationResponse(ok=True) def QueryStateAlpha1(self, request, context): diff --git a/tests/clients/test_configuration_watcher.py b/tests/clients/test_configuration_watcher.py new file mode 100644 index 000000000..442a47bb9 --- /dev/null +++ b/tests/clients/test_configuration_watcher.py @@ -0,0 +1,285 @@ +# -*- coding: utf-8 -*- + +""" +Copyright 2026 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 threading +import time +import unittest +from typing import Callable, List, Tuple +from unittest.mock import patch + +import grpc + +from dapr.clients.grpc._response import ( + CONFIG_RECONNECT_INITIAL_BACKOFF_SECONDS, + CONFIG_RECONNECT_MAX_BACKOFF_SECONDS, + ConfigurationResponse, + ConfigurationWatcher, + config_reconnect_delay, + is_retryable_config_error, +) +from dapr.clients.grpc.client import DaprGrpcClient +from dapr.conf import settings + +from .fake_dapr_server import FakeDaprSidecar + +STORE = 'configstore' +WAIT_TIMEOUT_SECONDS = 5.0 + + +def wait_until(condition: Callable[[], bool], timeout: float = WAIT_TIMEOUT_SECONDS) -> bool: + deadline = time.monotonic() + timeout + while time.monotonic() < deadline: + if condition(): + return True + time.sleep(0.01) + return condition() + + +class _FakeRpcError(grpc.RpcError): + def __init__(self, code: grpc.StatusCode): + self._code = code + + def code(self) -> grpc.StatusCode: + return self._code + + +class ConfigurationRetryHelpersTests(unittest.TestCase): + def test_delay_doubles_and_is_capped(self): + for attempt, base in [(0, 0.5), (1, 1.0), (2, 2.0), (3, 4.0), (4, 8.0), (10, 8.0)]: + delay = config_reconnect_delay(attempt) + self.assertGreaterEqual(delay, base) + self.assertLessEqual(delay, base * 1.2) + self.assertEqual(CONFIG_RECONNECT_INITIAL_BACKOFF_SECONDS, 0.5) + self.assertEqual(CONFIG_RECONNECT_MAX_BACKOFF_SECONDS, 8.0) + + def test_retryable_classification(self): + for code in ( + grpc.StatusCode.UNAVAILABLE, + grpc.StatusCode.UNKNOWN, + grpc.StatusCode.INTERNAL, + grpc.StatusCode.CANCELLED, + grpc.StatusCode.DEADLINE_EXCEEDED, + ): + self.assertTrue(is_retryable_config_error(_FakeRpcError(code)), code) + for code in ( + grpc.StatusCode.INVALID_ARGUMENT, + grpc.StatusCode.NOT_FOUND, + grpc.StatusCode.PERMISSION_DENIED, + grpc.StatusCode.UNAUTHENTICATED, + grpc.StatusCode.UNIMPLEMENTED, + ): + self.assertFalse(is_retryable_config_error(_FakeRpcError(code)), code) + self.assertFalse(is_retryable_config_error(ValueError('closed channel'))) + + +class ConfigurationWatcherReconnectTests(unittest.TestCase): + grpc_port = 50021 + http_port = 3521 + + @classmethod + def setUpClass(cls): + cls._fake_dapr_server = FakeDaprSidecar(grpc_port=cls.grpc_port, http_port=cls.http_port) + cls._fake_dapr_server.start() + settings.DAPR_HTTP_PORT = cls.http_port + settings.DAPR_HTTP_ENDPOINT = 'http://127.0.0.1:{}'.format(cls.http_port) + + @classmethod + def tearDownClass(cls): + cls._fake_dapr_server.stop() + + def setUp(self): + server = self._fake_dapr_server + server.config_stream_plans.clear() + server.config_subscribe_requests.clear() + server.config_unsubscribe_requests.clear() + self.delays: List[float] = [] + self.updates: List[Tuple[str, str]] = [] + + def record_wait(watcher: ConfigurationWatcher, delay: float) -> bool: + self.delays.append(delay) + return watcher._stop_event.is_set() + + patcher = patch.object(ConfigurationWatcher, '_wait_before_retry', record_wait) + patcher.start() + self.addCleanup(patcher.stop) + self.client = DaprGrpcClient(f'localhost:{self.grpc_port}') + self.addCleanup(self.client.close) + + def handler(self, subscription_id: str, response: ConfigurationResponse) -> None: + self.updates.append((subscription_id, response.items['k'].value)) + + def subscribe(self) -> str: + return self.client.subscribe_configuration( + store_name=STORE, keys=['k'], handler=self.handler + ) + + def watcher(self, subscription_id: str) -> ConfigurationWatcher: + return self.client._config_watchers[(STORE, subscription_id)] + + def test_reconnects_after_stream_error_and_keeps_delivering(self): + self._fake_dapr_server.config_stream_plans.extend( + [ + {'id': 'first', 'updates': [{'k': 'v1'}], 'end': 'abort'}, + {'id': 'second', 'updates': [{'k': 'v2'}], 'end': 'hold'}, + ] + ) + + subscription_id = self.subscribe() + + self.assertEqual(subscription_id, 'first') + self.assertTrue(wait_until(lambda: len(self.updates) == 2), self.updates) + self.assertEqual(self.updates, [('first', 'v1'), ('first', 'v2')]) + requests = self._fake_dapr_server.config_subscribe_requests + self.assertEqual(len(requests), 2) + self.assertEqual(requests[0], requests[1]) + self.assertEqual(self.watcher('first').id, 'second') + + def test_reconnects_after_clean_stream_end(self): + self._fake_dapr_server.config_stream_plans.extend( + [ + {'id': 'first', 'updates': [{'k': 'v1'}], 'end': 'eof'}, + {'id': 'second', 'updates': [{'k': 'v2'}], 'end': 'hold'}, + ] + ) + + self.subscribe() + + self.assertTrue(wait_until(lambda: len(self.updates) == 2), self.updates) + self.assertEqual(self.updates, [('first', 'v1'), ('first', 'v2')]) + + def test_unsubscribe_after_reconnect_uses_new_id_and_stops(self): + server = self._fake_dapr_server + server.config_stream_plans.extend( + [ + {'id': 'first', 'end': 'abort'}, + {'id': 'second', 'updates': [{'k': 'v2'}], 'end': 'hold'}, + ] + ) + subscription_id = self.subscribe() + self.assertTrue(wait_until(lambda: len(self.updates) == 1), self.updates) + watcher = self.watcher(subscription_id) + thread = watcher._thread + assert thread is not None + + self.assertTrue(self.client.unsubscribe_configuration(STORE, subscription_id)) + + self.assertEqual([r.id for r in server.config_unsubscribe_requests], ['second']) + self.assertEqual([r.store_name for r in server.config_unsubscribe_requests], [STORE]) + self.assertFalse(thread.is_alive()) + self.assertTrue(watcher.stopped) + self.assertNotIn((STORE, subscription_id), self.client._config_watchers) + time.sleep(0.2) + self.assertEqual(len(server.config_subscribe_requests), 2) + + def test_unsubscribe_unknown_id_is_sent_as_is(self): + self.assertTrue(self.client.unsubscribe_configuration(STORE, 'not-tracked')) + self.assertEqual( + [r.id for r in self._fake_dapr_server.config_unsubscribe_requests], ['not-tracked'] + ) + + def test_non_retryable_error_stops_without_reconnecting(self): + server = self._fake_dapr_server + server.config_stream_plans.extend( + [ + { + 'id': 'first', + 'updates': [{'k': 'v1'}], + 'end': 'abort', + 'code': grpc.StatusCode.PERMISSION_DENIED, + }, + {'id': 'second', 'updates': [{'k': 'v2'}], 'end': 'hold'}, + ] + ) + subscription_id = self.subscribe() + thread = self.watcher(subscription_id)._thread + assert thread is not None + + thread.join(timeout=WAIT_TIMEOUT_SECONDS) + + self.assertFalse(thread.is_alive()) + self.assertEqual(self.updates, [('first', 'v1')]) + self.assertEqual(len(server.config_subscribe_requests), 1) + self.assertEqual(self.delays, []) + + def test_non_retryable_error_on_first_connection_returns_none_quickly(self): + server = self._fake_dapr_server + server.config_stream_plans.extend( + [{'reject': grpc.StatusCode.INVALID_ARGUMENT}, {'id': 'unused', 'end': 'hold'}] + ) + threads_before = threading.active_count() + + started = time.monotonic() + subscription_id = self.subscribe() + + self.assertIsNone(subscription_id) + self.assertLess(time.monotonic() - started, 2.0) + self.assertEqual(len(server.config_subscribe_requests), 1) + self.assertEqual(self.client._config_watchers, {}) + self.assertTrue(wait_until(lambda: threading.active_count() <= threads_before)) + + def test_backoff_grows_on_consecutive_failures_and_resets_after_success(self): + server = self._fake_dapr_server + server.config_stream_plans.extend( + [ + {'reject': grpc.StatusCode.UNAVAILABLE}, + {'reject': grpc.StatusCode.UNAVAILABLE}, + {'reject': grpc.StatusCode.UNAVAILABLE}, + {'id': 'first', 'end': 'abort'}, + {'id': 'second', 'updates': [{'k': 'v1'}], 'end': 'hold'}, + ] + ) + + subscription_id = self.subscribe() + + self.assertEqual(subscription_id, 'first') + self.assertTrue(wait_until(lambda: len(self.updates) == 1), self.updates) + self.assertEqual(len(self.delays), 4, self.delays) + for delay, base in zip(self.delays, [0.5, 1.0, 2.0, 0.5]): + self.assertGreaterEqual(delay, base) + self.assertLessEqual(delay, base * 1.2) + + def test_handler_exception_does_not_stop_subscription(self): + self._fake_dapr_server.config_stream_plans.append( + {'id': 'first', 'updates': [{'k': 'boom'}, {'k': 'v2'}], 'end': 'hold'} + ) + + def handler(subscription_id: str, response: ConfigurationResponse) -> None: + value = response.items['k'].value + if value == 'boom': + raise RuntimeError('handler failure') + self.updates.append((subscription_id, value)) + + self.client.subscribe_configuration(store_name=STORE, keys=['k'], handler=handler) + + self.assertTrue(wait_until(lambda: len(self.updates) == 1), self.updates) + self.assertEqual(self.updates, [('first', 'v2')]) + self.assertEqual(len(self._fake_dapr_server.config_subscribe_requests), 1) + + def test_close_stops_watchers(self): + self._fake_dapr_server.config_stream_plans.append({'id': 'first', 'end': 'hold'}) + subscription_id = self.subscribe() + watcher = self.watcher(subscription_id) + thread = watcher._thread + assert thread is not None + + self.client.close() + + self.assertFalse(thread.is_alive()) + self.assertTrue(watcher.stopped) + self.assertEqual(self.client._config_watchers, {}) + + +if __name__ == '__main__': + unittest.main() diff --git a/tests/clients/test_configuration_watcher_async.py b/tests/clients/test_configuration_watcher_async.py new file mode 100644 index 000000000..072c2b90e --- /dev/null +++ b/tests/clients/test_configuration_watcher_async.py @@ -0,0 +1,214 @@ +# -*- coding: utf-8 -*- + +""" +Copyright 2026 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 asyncio +import time +import unittest +from typing import Callable, List, Tuple +from unittest.mock import patch + +import grpc + +from dapr.aio.clients.grpc._response import AsyncConfigurationWatcher +from dapr.aio.clients.grpc.client import DaprGrpcClientAsync +from dapr.clients.grpc._response import ConfigurationResponse +from dapr.conf import settings + +from .fake_dapr_server import FakeDaprSidecar + +STORE = 'configstore' +WAIT_TIMEOUT_SECONDS = 5.0 + + +async def wait_until(condition: Callable[[], bool], timeout: float = WAIT_TIMEOUT_SECONDS) -> bool: + deadline = time.monotonic() + timeout + while time.monotonic() < deadline: + if condition(): + return True + await asyncio.sleep(0.01) + return condition() + + +class AsyncConfigurationWatcherReconnectTests(unittest.IsolatedAsyncioTestCase): + grpc_port = 50022 + http_port = 3522 + + @classmethod + def setUpClass(cls): + cls._fake_dapr_server = FakeDaprSidecar(grpc_port=cls.grpc_port, http_port=cls.http_port) + cls._fake_dapr_server.start() + settings.DAPR_HTTP_PORT = cls.http_port + settings.DAPR_HTTP_ENDPOINT = 'http://127.0.0.1:{}'.format(cls.http_port) + + @classmethod + def tearDownClass(cls): + cls._fake_dapr_server.stop() + + async def asyncSetUp(self): + server = self._fake_dapr_server + server.config_stream_plans.clear() + server.config_subscribe_requests.clear() + server.config_unsubscribe_requests.clear() + self.delays: List[float] = [] + self.updates: List[Tuple[str, str]] = [] + + async def record_wait(watcher: AsyncConfigurationWatcher, delay: float) -> None: + self.delays.append(delay) + + patcher = patch.object(AsyncConfigurationWatcher, '_wait_before_retry', record_wait) + patcher.start() + self.addCleanup(patcher.stop) + self.client = DaprGrpcClientAsync(f'localhost:{self.grpc_port}') + self.addAsyncCleanup(self.client.close) + + def handler(self, subscription_id: str, response: ConfigurationResponse) -> None: + self.updates.append((subscription_id, response.items['k'].value)) + + async def subscribe(self) -> str: + return await self.client.subscribe_configuration( + store_name=STORE, keys=['k'], handler=self.handler + ) + + def watcher(self, subscription_id: str) -> AsyncConfigurationWatcher: + return self.client._config_watchers[(STORE, subscription_id)] + + async def test_subscribe_returns_id_and_delivers_updates(self): + self._fake_dapr_server.config_stream_plans.append( + {'id': 'first', 'updates': [{'k': 'v1'}], 'end': 'hold'} + ) + + subscription_id = await self.subscribe() + + self.assertEqual(subscription_id, 'first') + self.assertTrue(await wait_until(lambda: len(self.updates) == 1), self.updates) + self.assertEqual(self.updates, [('first', 'v1')]) + + async def test_reconnects_after_stream_error_and_keeps_delivering(self): + self._fake_dapr_server.config_stream_plans.extend( + [ + {'id': 'first', 'updates': [{'k': 'v1'}], 'end': 'abort'}, + {'id': 'second', 'updates': [{'k': 'v2'}], 'end': 'hold'}, + ] + ) + + await self.subscribe() + + self.assertTrue(await wait_until(lambda: len(self.updates) == 2), self.updates) + self.assertEqual(self.updates, [('first', 'v1'), ('first', 'v2')]) + self.assertEqual(len(self._fake_dapr_server.config_subscribe_requests), 2) + self.assertEqual(self.watcher('first').id, 'second') + + async def test_reconnects_after_clean_stream_end(self): + self._fake_dapr_server.config_stream_plans.extend( + [ + {'id': 'first', 'updates': [{'k': 'v1'}], 'end': 'eof'}, + {'id': 'second', 'updates': [{'k': 'v2'}], 'end': 'hold'}, + ] + ) + + await self.subscribe() + + self.assertTrue(await wait_until(lambda: len(self.updates) == 2), self.updates) + + async def test_unsubscribe_after_reconnect_uses_new_id_and_stops(self): + server = self._fake_dapr_server + server.config_stream_plans.extend( + [ + {'id': 'first', 'end': 'abort'}, + {'id': 'second', 'updates': [{'k': 'v2'}], 'end': 'hold'}, + ] + ) + subscription_id = await self.subscribe() + self.assertTrue(await wait_until(lambda: len(self.updates) == 1), self.updates) + watcher = self.watcher(subscription_id) + task = watcher._task + assert task is not None + + self.assertTrue(await self.client.unsubscribe_configuration(STORE, subscription_id)) + + self.assertEqual([r.id for r in server.config_unsubscribe_requests], ['second']) + self.assertTrue(task.done()) + self.assertNotIn((STORE, subscription_id), self.client._config_watchers) + await asyncio.sleep(0.2) + self.assertEqual(len(server.config_subscribe_requests), 2) + + async def test_non_retryable_error_stops_without_reconnecting(self): + server = self._fake_dapr_server + server.config_stream_plans.extend( + [ + { + 'id': 'first', + 'updates': [{'k': 'v1'}], + 'end': 'abort', + 'code': grpc.StatusCode.NOT_FOUND, + }, + {'id': 'second', 'updates': [{'k': 'v2'}], 'end': 'hold'}, + ] + ) + subscription_id = await self.subscribe() + task = self.watcher(subscription_id)._task + assert task is not None + + await asyncio.wait_for(asyncio.shield(task), timeout=WAIT_TIMEOUT_SECONDS) + + self.assertEqual(self.updates, [('first', 'v1')]) + self.assertEqual(len(server.config_subscribe_requests), 1) + self.assertEqual(self.delays, []) + + async def test_non_retryable_error_on_first_connection_returns_none_quickly(self): + self._fake_dapr_server.config_stream_plans.extend( + [{'reject': grpc.StatusCode.UNAUTHENTICATED}, {'id': 'unused', 'end': 'hold'}] + ) + + started = time.monotonic() + subscription_id = await self.subscribe() + + self.assertIsNone(subscription_id) + self.assertLess(time.monotonic() - started, 2.0) + self.assertEqual(len(self._fake_dapr_server.config_subscribe_requests), 1) + self.assertEqual(self.client._config_watchers, {}) + + async def test_backoff_grows_on_consecutive_failures_and_resets_after_success(self): + self._fake_dapr_server.config_stream_plans.extend( + [ + {'reject': grpc.StatusCode.UNAVAILABLE}, + {'reject': grpc.StatusCode.UNAVAILABLE}, + {'id': 'first', 'end': 'abort'}, + {'id': 'second', 'updates': [{'k': 'v1'}], 'end': 'hold'}, + ] + ) + + self.assertEqual(await self.subscribe(), 'first') + + self.assertTrue(await wait_until(lambda: len(self.updates) == 1), self.updates) + self.assertEqual(len(self.delays), 3, self.delays) + for delay, base in zip(self.delays, [0.5, 1.0, 0.5]): + self.assertGreaterEqual(delay, base) + self.assertLessEqual(delay, base * 1.2) + + async def test_close_stops_watchers(self): + self._fake_dapr_server.config_stream_plans.append({'id': 'first', 'end': 'hold'}) + subscription_id = await self.subscribe() + task = self.watcher(subscription_id)._task + assert task is not None + + await self.client.close() + + self.assertTrue(task.done()) + self.assertEqual(self.client._config_watchers, {}) + + +if __name__ == '__main__': + unittest.main() diff --git a/tests/clients/test_dapr_grpc_client_async.py b/tests/clients/test_dapr_grpc_client_async.py index fe863b7c2..42638d31e 100644 --- a/tests/clients/test_dapr_grpc_client_async.py +++ b/tests/clients/test_dapr_grpc_client_async.py @@ -24,6 +24,7 @@ from google.rpc import code_pb2, status_pb2 from dapr.aio.clients import DaprClient +from dapr.aio.clients.grpc._response import AsyncConfigurationWatcher from dapr.aio.clients.grpc.client import DaprGrpcClientAsync from dapr.clients.exceptions import DaprGrpcError from dapr.clients.grpc import conversation @@ -34,7 +35,6 @@ from dapr.clients.grpc._response import ( ConfigurationItem, ConfigurationResponse, - ConfigurationWatcher, DaprResponse, UnlockResponseStatus, ) @@ -836,7 +836,7 @@ async def test_get_configuration(self): async def test_subscribe_configuration(self): dapr = DaprGrpcClientAsync(f'{self.scheme}localhost:{self.grpc_port}') - def mock_watch(self, stub, store_name, keys, handler, config_metadata): + async def mock_watch(self, stub, store_name, keys, handler, config_metadata): handler( 'id', ConfigurationResponse( @@ -849,7 +849,7 @@ def handler(id: str, resp: ConfigurationResponse): self.assertEqual(resp.items['k'].value, 'test') self.assertEqual(resp.items['k'].version, '1.7.0') - with patch.object(ConfigurationWatcher, 'watch_configuration', mock_watch): + with patch.object(AsyncConfigurationWatcher, 'watch_configuration', mock_watch): await dapr.subscribe_configuration( store_name='configurationstore', keys=['k'], handler=handler ) From 89ce67fcfb7a78854b326d8b68b1f745dca4cd47 Mon Sep 17 00:00:00 2001 From: Casper Nielsen Date: Mon, 28 Sep 2026 22:39:10 +0200 Subject: [PATCH 2/6] fix(clients): await async config handlers, back off on short-lived streams, close robustly - The async client now awaits handlers that return a coroutine, so `async def` handlers run instead of being silently dropped. - The reconnect backoff only starts over after a stream has stayed up for CONFIG_STABLE_STREAM_SECONDS; a sidecar that closes the stream right after sending the id no longer gets reconnected every 0.5 s. - DaprGrpcClient.close() marks watchers stopped, closes the channel, then waits for the threads. gRPC client instrumentation (such as OpenTelemetry, which the workflow client enables on import) wraps server streams in an iterator without cancel(), so only closing the channel ends them. - The first-connection test checks the watcher's own thread instead of the global thread count, which other tests change. Co-Authored-By: Claude Opus 5.5 Signed-off-by: Casper Nielsen --- dapr/aio/clients/grpc/_response.py | 39 ++++++++++--- dapr/aio/clients/grpc/client.py | 3 +- dapr/clients/grpc/_response.py | 33 ++++++++++- dapr/clients/grpc/client.py | 15 +++-- tests/clients/test_configuration_watcher.py | 55 ++++++++++++++++++- .../test_configuration_watcher_async.py | 36 ++++++++++++ 6 files changed, 160 insertions(+), 21 deletions(-) diff --git a/dapr/aio/clients/grpc/_response.py b/dapr/aio/clients/grpc/_response.py index 3f277e370..0fac27a7f 100644 --- a/dapr/aio/clients/grpc/_response.py +++ b/dapr/aio/clients/grpc/_response.py @@ -15,8 +15,20 @@ import asyncio import contextlib +import inspect import logging -from typing import AsyncGenerator, Callable, Dict, Generic, List, Optional, Text +import time +from typing import ( + AsyncGenerator, + Awaitable, + Callable, + Dict, + Generic, + List, + Optional, + Text, + Union, +) from dapr.clients.grpc._response import ( CONFIG_SUBSCRIBE_TIMEOUT_SECONDS, @@ -24,12 +36,16 @@ DaprResponse, TCryptoResponse, config_reconnect_delay, + config_stream_was_stable, is_retryable_config_error, ) from dapr.proto import api_service_v1, api_v1 logger = logging.getLogger(__name__) +# A configuration handler for the async client: a plain function or an ``async def``. +AsyncConfigurationHandler = Callable[[Text, ConfigurationResponse], Union[None, Awaitable[None]]] + class CryptoResponse(DaprResponse, Generic[TCryptoResponse]): """An asynchronous iterable of cryptography API responses.""" @@ -121,13 +137,14 @@ def __init__(self) -> None: self._call = None self._task: Optional[asyncio.Task] = None self._stream_established = False + self._established_at = 0.0 async def watch_configuration( self, stub: api_service_v1.DaprStub, store_name: str, keys: List[str], - handler: Callable[[Text, ConfigurationResponse], None], + handler: AsyncConfigurationHandler, config_metadata: Optional[Dict[str, str]] = None, ): """Starts the watcher task and returns the subscription id, or None if the sidecar did @@ -180,7 +197,7 @@ async def _read_subscribe_config( self, stub: api_service_v1.DaprStub, req: api_v1.SubscribeConfigurationRequest, - handler: Callable[[Text, ConfigurationResponse], None], + handler: AsyncConfigurationHandler, ) -> None: attempt = 0 try: @@ -219,7 +236,7 @@ async def _read_subscribe_config( self._call = None if self._stopping: break - if self._stream_established: + if self._stream_established and config_stream_was_stable(self._established_at): attempt = 0 delay = config_reconnect_delay(attempt) attempt += 1 @@ -235,7 +252,7 @@ async def _consume_stream( self, stub: api_service_v1.DaprStub, req: api_v1.SubscribeConfigurationRequest, - handler: Callable[[Text, ConfigurationResponse], None], + handler: AsyncConfigurationHandler, ) -> None: call = stub.SubscribeConfigurationAlpha1(req) self._call = call @@ -245,7 +262,7 @@ async def _consume_stream( if self._stopping: return if len(response.items) > 0: - self._deliver(handler, ConfigurationResponse(response.items)) + await self._deliver(handler, ConfigurationResponse(response.items)) def _on_stream_established(self, server_id: str) -> None: self.id = server_id @@ -259,16 +276,20 @@ def _on_stream_established(self, server_id: str) -> None: self.store_name, server_id, ) + self._established_at = time.monotonic() self._stream_established = True self._ready.set() - def _deliver( + async def _deliver( self, - handler: Callable[[Text, ConfigurationResponse], None], + handler: AsyncConfigurationHandler, response: ConfigurationResponse, ) -> None: + # Accept both plain functions and ``async def`` handlers. try: - handler(self.subscription_id, response) + result = handler(self.subscription_id, response) + if inspect.isawaitable(result): + await result except Exception: logger.exception( 'Configuration handler for keys %s on store %s raised an exception', diff --git a/dapr/aio/clients/grpc/client.py b/dapr/aio/clients/grpc/client.py index 7d21a839b..b54c07e35 100644 --- a/dapr/aio/clients/grpc/client.py +++ b/dapr/aio/clients/grpc/client.py @@ -41,6 +41,7 @@ EncryptRequestIterator, ) from dapr.aio.clients.grpc._response import ( + AsyncConfigurationHandler, AsyncConfigurationWatcher, DecryptResponse, EncryptResponse, @@ -1218,7 +1219,7 @@ async def subscribe_configuration( self, store_name: str, keys: List[str], - handler: Callable[[Text, ConfigurationResponse], None], + handler: AsyncConfigurationHandler, config_metadata: Optional[Dict[str, str]] = dict(), ) -> Text: """Gets changed value from a config store with a key diff --git a/dapr/clients/grpc/_response.py b/dapr/clients/grpc/_response.py index 259533ae1..ab3b5080c 100644 --- a/dapr/clients/grpc/_response.py +++ b/dapr/clients/grpc/_response.py @@ -20,6 +20,7 @@ import logging import random import threading +import time from enum import Enum from typing import ( TYPE_CHECKING, @@ -680,6 +681,9 @@ def items(self) -> Dict[Text, ConfigurationItem]: CONFIG_RECONNECT_INITIAL_BACKOFF_SECONDS = 0.5 CONFIG_RECONNECT_MAX_BACKOFF_SECONDS = 8.0 _CONFIG_RECONNECT_JITTER_RATIO = 0.2 +# A stream must stay up this long before the reconnect backoff starts over; a sidecar that +# accepts the subscription and closes it straight away keeps backing off instead of looping. +CONFIG_STABLE_STREAM_SECONDS = 10.0 # How long stop() waits for the watcher thread to exit after cancelling the stream. _CONFIG_WATCHER_JOIN_TIMEOUT_SECONDS = 2.0 # Errors that will not go away by resubscribing (bad request, missing store, auth). @@ -703,6 +707,12 @@ def config_reconnect_delay(attempt: int) -> float: return base + random.uniform(0, base * _CONFIG_RECONNECT_JITTER_RATIO) +def config_stream_was_stable(established_at: float) -> bool: + """Returns True if a stream established at ``established_at`` (time.monotonic()) stayed up + long enough for the reconnect backoff to start over.""" + return time.monotonic() - established_at >= CONFIG_STABLE_STREAM_SECONDS + + def is_retryable_config_error(error: BaseException) -> bool: """Returns True if a configuration subscription stream should be re-established after ``error``. Only gRPC errors are retried, except the ones in _CONFIG_NON_RETRYABLE_CODES.""" @@ -736,6 +746,7 @@ def __init__(self): self._call = None self._thread: Optional[threading.Thread] = None self._stream_established = False + self._established_at = 0.0 def watch_configuration( self, @@ -779,15 +790,30 @@ def request_stop(self) -> None: def stop(self) -> None: """Stops reconnecting, cancels the current stream and waits briefly for the thread.""" - self._stop_event.set() + self.request_stop() + self.cancel_stream() + self.join() + + def cancel_stream(self) -> None: + """Cancels the current stream, if the call object supports it. + + Some wrappers (for example gRPC client instrumentation) hand back a plain iterator + without cancel(); closing the channel ends those streams instead. + """ with self._lock: call = self._call cancel = getattr(call, 'cancel', None) if callable(cancel): cancel() + + def join(self, timeout: Optional[float] = None) -> None: + """Waits up to ``timeout`` seconds (default _CONFIG_WATCHER_JOIN_TIMEOUT_SECONDS) for the + watcher thread to exit.""" thread = self._thread if thread is not None and thread is not threading.current_thread(): - thread.join(timeout=_CONFIG_WATCHER_JOIN_TIMEOUT_SECONDS) + thread.join( + timeout=_CONFIG_WATCHER_JOIN_TIMEOUT_SECONDS if timeout is None else timeout + ) def _read_subscribe_config( self, @@ -831,7 +857,7 @@ def _read_subscribe_config( self._call = None if self._stop_event.is_set(): break - if self._stream_established: + if self._stream_established and config_stream_was_stable(self._established_at): attempt = 0 delay = config_reconnect_delay(attempt) attempt += 1 @@ -880,6 +906,7 @@ def _on_stream_established(self, server_id: str) -> None: self.store_name, server_id, ) + self._established_at = time.monotonic() self._stream_established = True self.event.set() diff --git a/dapr/clients/grpc/client.py b/dapr/clients/grpc/client.py index d43750b92..671a20aae 100644 --- a/dapr/clients/grpc/client.py +++ b/dapr/clients/grpc/client.py @@ -214,19 +214,24 @@ def get_credentials(): def close(self): """Closes Dapr runtime gRPC channel.""" - self._stop_config_watchers() + watchers = self._take_config_watchers() + for watcher in watchers: + watcher.request_stop() + watcher.cancel_stream() + # Closing the channel ends any stream the watchers could not cancel themselves. if hasattr(self, '_channel') and self._channel: self._channel.close() + for watcher in watchers: + watcher.join() - def _stop_config_watchers(self) -> None: + def _take_config_watchers(self) -> List[ConfigurationWatcher]: lock = getattr(self, '_config_watchers_lock', None) if lock is None: - return + return [] with lock: watchers = list(self._config_watchers.values()) self._config_watchers = {} - for watcher in watchers: - watcher.stop() + return watchers def __del__(self): self.close() diff --git a/tests/clients/test_configuration_watcher.py b/tests/clients/test_configuration_watcher.py index 442a47bb9..0ce02f9d3 100644 --- a/tests/clients/test_configuration_watcher.py +++ b/tests/clients/test_configuration_watcher.py @@ -218,8 +218,6 @@ def test_non_retryable_error_on_first_connection_returns_none_quickly(self): server.config_stream_plans.extend( [{'reject': grpc.StatusCode.INVALID_ARGUMENT}, {'id': 'unused', 'end': 'hold'}] ) - threads_before = threading.active_count() - started = time.monotonic() subscription_id = self.subscribe() @@ -227,8 +225,19 @@ def test_non_retryable_error_on_first_connection_returns_none_quickly(self): self.assertLess(time.monotonic() - started, 2.0) self.assertEqual(len(server.config_subscribe_requests), 1) self.assertEqual(self.client._config_watchers, {}) - self.assertTrue(wait_until(lambda: threading.active_count() <= threads_before)) + # The watcher's own thread must exit; other tests' threads don't matter here. + self.assertTrue( + wait_until( + lambda: ( + not any( + t.name == f'dapr-configuration-watcher-{STORE}' and t.is_alive() + for t in threading.enumerate() + ) + ) + ) + ) + @patch('dapr.clients.grpc._response.CONFIG_STABLE_STREAM_SECONDS', 0) def test_backoff_grows_on_consecutive_failures_and_resets_after_success(self): server = self._fake_dapr_server server.config_stream_plans.extend( @@ -280,6 +289,46 @@ def test_close_stops_watchers(self): self.assertTrue(watcher.stopped) self.assertEqual(self.client._config_watchers, {}) + @patch('dapr.clients.grpc._response._CONFIG_WATCHER_JOIN_TIMEOUT_SECONDS', 30.0) + def test_close_stops_watchers_whose_stream_cannot_be_cancelled(self): + # gRPC client instrumentation (e.g. OpenTelemetry) wraps server streams in a plain + # generator without cancel(); close() must still end the watcher via the channel. + self._fake_dapr_server.config_stream_plans.append({'id': 'first', 'end': 'hold'}) + subscribe = self.client._stub.SubscribeConfigurationAlpha1 + self.client._stub.SubscribeConfigurationAlpha1 = lambda req: (r for r in subscribe(req)) + subscription_id = self.subscribe() + watcher = self.watcher(subscription_id) + self.assertFalse(hasattr(watcher._call, 'cancel')) + thread = watcher._thread + assert thread is not None + + started = time.monotonic() + self.client.close() + + # The channel is closed before waiting on the thread, so close() doesn't sit out the + # join timeout. + self.assertLess(time.monotonic() - started, 10.0) + self.assertFalse(thread.is_alive()) + + def test_backoff_keeps_growing_when_streams_close_right_after_the_id(self): + # A sidecar that accepts the subscription and closes it at once must not be + # reconnected every CONFIG_RECONNECT_INITIAL_BACKOFF_SECONDS. + self._fake_dapr_server.config_stream_plans.extend( + [ + {'id': 'first', 'end': 'abort'}, + {'id': 'second', 'end': 'abort'}, + {'id': 'third', 'updates': [{'k': 'v1'}], 'end': 'hold'}, + ] + ) + + self.assertEqual(self.subscribe(), 'first') + + self.assertTrue(wait_until(lambda: len(self.updates) == 1), self.updates) + self.assertEqual(len(self.delays), 2, self.delays) + for delay, base in zip(self.delays, [0.5, 1.0]): + self.assertGreaterEqual(delay, base) + self.assertLessEqual(delay, base * 1.2) + if __name__ == '__main__': unittest.main() diff --git a/tests/clients/test_configuration_watcher_async.py b/tests/clients/test_configuration_watcher_async.py index 072c2b90e..49f5bd038 100644 --- a/tests/clients/test_configuration_watcher_async.py +++ b/tests/clients/test_configuration_watcher_async.py @@ -180,6 +180,7 @@ async def test_non_retryable_error_on_first_connection_returns_none_quickly(self self.assertEqual(len(self._fake_dapr_server.config_subscribe_requests), 1) self.assertEqual(self.client._config_watchers, {}) + @patch('dapr.clients.grpc._response.CONFIG_STABLE_STREAM_SECONDS', 0) async def test_backoff_grows_on_consecutive_failures_and_resets_after_success(self): self._fake_dapr_server.config_stream_plans.extend( [ @@ -209,6 +210,41 @@ async def test_close_stops_watchers(self): self.assertTrue(task.done()) self.assertEqual(self.client._config_watchers, {}) + async def test_backoff_keeps_growing_when_streams_close_right_after_the_id(self): + self._fake_dapr_server.config_stream_plans.extend( + [ + {'id': 'first', 'end': 'abort'}, + {'id': 'second', 'end': 'abort'}, + {'id': 'third', 'updates': [{'k': 'v1'}], 'end': 'hold'}, + ] + ) + + self.assertEqual(await self.subscribe(), 'first') + + self.assertTrue(await wait_until(lambda: len(self.updates) == 1), self.updates) + self.assertEqual(len(self.delays), 2, self.delays) + for delay, base in zip(self.delays, [0.5, 1.0]): + self.assertGreaterEqual(delay, base) + self.assertLessEqual(delay, base * 1.2) + + async def test_async_handler_is_awaited(self): + self._fake_dapr_server.config_stream_plans.append( + {'id': 'first', 'updates': [{'k': 'v1'}], 'end': 'hold'} + ) + received: List[Tuple[str, str]] = [] + + async def async_handler(subscription_id: str, response: ConfigurationResponse) -> None: + await asyncio.sleep(0) + received.append((subscription_id, response.items['k'].value)) + + subscription_id = await self.client.subscribe_configuration( + store_name=STORE, keys=['k'], handler=async_handler + ) + + self.assertEqual(subscription_id, 'first') + self.assertTrue(await wait_until(lambda: len(received) == 1), received) + self.assertEqual(received, [('first', 'v1')]) + if __name__ == '__main__': unittest.main() From 44ed6a32e6947ed1e180e3572a9d6d55634ac9ed Mon Sep 17 00:00:00 2001 From: Casper Nielsen Date: Mon, 28 Sep 2026 23:09:48 +0200 Subject: [PATCH 3/6] fix(clients): keep configuration subscriptions alive through store errors and catch up after reconnects - Retry every gRPC error except UNIMPLEMENTED, PERMISSION_DENIED and UNAUTHENTICATED once a subscription has been established. daprd reports a failed store Subscribe (including a backend that is briefly unreachable) and a store that is not loaded as INVALID_ARGUMENT, so a reconnect during such an outage used to end the subscription. The first subscribe still fails fast on INVALID_ARGUMENT and NOT_FOUND. - After each reconnect, read the current values with GetConfiguration and pass them to the handler, so changes made while the stream was down are delivered. A failed read is logged and the subscription carries on. - unsubscribe_configuration stops a watcher that is between streams locally and returns True instead of sending the dead stream's id. - Remove a watcher from the client when it gives up on its own. - Cancel a stream opened after stop() was requested (async too). - Log reconnect failures as "CODE: details", at WARNING for the first failure of an outage and DEBUG for the retries after it. - The fake sidecar answers ok=False for ids without an open stream, like daprd, and can return configurable values from GetConfiguration. Co-Authored-By: Claude Opus 5.5 Signed-off-by: Casper Nielsen --- dapr/aio/clients/grpc/_response.py | 107 ++++++-- dapr/aio/clients/grpc/client.py | 41 ++- dapr/clients/grpc/_response.py | 172 +++++++++--- dapr/clients/grpc/client.py | 53 +++- tests/clients/fake_dapr_server.py | 73 +++-- tests/clients/test_configuration_watcher.py | 250 +++++++++++++++++- .../test_configuration_watcher_async.py | 192 +++++++++++++- tests/clients/test_dapr_grpc_client.py | 3 +- tests/clients/test_dapr_grpc_client_async.py | 3 +- 9 files changed, 794 insertions(+), 100 deletions(-) diff --git a/dapr/aio/clients/grpc/_response.py b/dapr/aio/clients/grpc/_response.py index 0fac27a7f..ae19c46bd 100644 --- a/dapr/aio/clients/grpc/_response.py +++ b/dapr/aio/clients/grpc/_response.py @@ -35,8 +35,10 @@ ConfigurationResponse, DaprResponse, TCryptoResponse, + cancel_config_call, config_reconnect_delay, config_stream_was_stable, + describe_config_error, is_retryable_config_error, ) from dapr.proto import api_service_v1, api_v1 @@ -121,13 +123,24 @@ class AsyncConfigurationWatcher: """Reads a SubscribeConfigurationAlpha1 stream in an asyncio task and calls the handler for every update. - This is the asyncio counterpart of ConfigurationWatcher: it re-subscribes with exponential - backoff when the stream fails with a retryable error or is closed by the sidecar, until - stop() is called or a non-retryable error is returned. ``subscription_id`` is the id of the - first stream; ``id`` is the id of the current stream. + This is the asyncio counterpart of ConfigurationWatcher and follows the same rules: it + re-subscribes with exponential backoff when the stream fails or is closed by the sidecar, + until stop() is called or a non-retryable error is returned (see + is_retryable_config_error). ``subscription_id`` is the id of the first stream and is what + the handler receives; ``id`` is the id of the current stream. + + After each reconnect (not after the first subscribe) the watcher reads the current values + of the keys with GetConfiguration and passes them to the handler, if there are any, so + changes made while the stream was down are delivered. This may repeat values that did not + change. If the read fails, a warning is logged and the subscription carries on. + + ``on_exit`` is called with the watcher when its task exits, whether it was stopped or gave + up on its own. """ - def __init__(self) -> None: + def __init__( + self, on_exit: Optional[Callable[['AsyncConfigurationWatcher'], None]] = None + ) -> None: self.store_name: Optional[str] = None self.keys: Optional[List[str]] = None self.id: str = '' @@ -138,6 +151,8 @@ def __init__(self) -> None: self._task: Optional[asyncio.Task] = None self._stream_established = False self._established_at = 0.0 + self._on_exit = on_exit + self._exited = False async def watch_configuration( self, @@ -175,6 +190,18 @@ async def watch_configuration( def stopped(self) -> bool: return self._stopping + @property + def exited(self) -> bool: + """True once the watcher task has finished (stopped or gave up).""" + return self._exited + + def live_stream_id(self) -> Optional[str]: + """Returns the id of the current stream if one is open and the sidecar has sent its id, + or None while the watcher is between streams (for example waiting to reconnect).""" + if self._call is not None and self._stream_established: + return self.id + return None + def request_stop(self) -> None: """Marks the watcher as stopping so it does not reconnect when the stream ends.""" self._stopping = True @@ -182,10 +209,7 @@ def request_stop(self) -> None: async def stop(self) -> None: """Stops reconnecting, cancels the current stream and waits for the task to exit.""" self._stopping = True - call = self._call - cancel = getattr(call, 'cancel', None) - if callable(cancel): - cancel() + cancel_config_call(self._call) task = self._task if task is None or task.done() or task is asyncio.current_task(): return @@ -200,6 +224,7 @@ async def _read_subscribe_config( handler: AsyncConfigurationHandler, ) -> None: attempt = 0 + outage_reported = False try: while not self._stopping: self._stream_established = False @@ -217,21 +242,26 @@ async def _read_subscribe_config( except Exception as error: if self._stopping: break - if not is_retryable_config_error(error): + if not is_retryable_config_error(error, established=bool(self.subscription_id)): logger.error( 'Configuration subscription for keys %s on store %s stopped: %s', self.keys, self.store_name, - error, + describe_config_error(error), ) break - logger.warning( + # Warn once per outage; the retries after that go to DEBUG until a new + # stream is up, which is logged at INFO. + first_failure = self._stream_established or not outage_reported + logger.log( + logging.WARNING if first_failure else logging.DEBUG, 'Configuration subscription stream for keys %s on store %s failed, ' 'reconnecting: %s', self.keys, self.store_name, - error, + describe_config_error(error), ) + outage_reported = True finally: self._call = None if self._stopping: @@ -242,11 +272,17 @@ async def _read_subscribe_config( attempt += 1 await self._wait_before_retry(delay) finally: + self._exited = True # Unblock watch_configuration if the first stream never delivered an id. self._ready.set() logger.debug( 'Configuration watcher for keys %s on store %s exited', self.keys, self.store_name ) + if self._on_exit is not None: + try: + self._on_exit(self) + except Exception: + logger.exception('Configuration watcher exit callback raised an exception') async def _consume_stream( self, @@ -255,18 +291,26 @@ async def _consume_stream( handler: AsyncConfigurationHandler, ) -> None: call = stub.SubscribeConfigurationAlpha1(req) + if self._stopping: + # stop() was requested while the call was being created. + cancel_config_call(call) + return self._call = call async for response in call: if not self._stream_established: - self._on_stream_established(response.id) + reconnected = self._on_stream_established(response.id) + if reconnected and not self._stopping: + await self._deliver_current_values(stub, req, handler) if self._stopping: return if len(response.items) > 0: await self._deliver(handler, ConfigurationResponse(response.items)) - def _on_stream_established(self, server_id: str) -> None: + def _on_stream_established(self, server_id: str) -> bool: + """Records the id of a new stream. Returns True if it replaces an earlier stream.""" self.id = server_id - if not self.subscription_id: + reconnected = bool(self.subscription_id) + if not reconnected: self.subscription_id = server_id else: logger.info( @@ -279,6 +323,37 @@ def _on_stream_established(self, server_id: str) -> None: self._established_at = time.monotonic() self._stream_established = True self._ready.set() + return reconnected + + async def _deliver_current_values( + self, + stub: api_service_v1.DaprStub, + req: api_v1.SubscribeConfigurationRequest, + handler: AsyncConfigurationHandler, + ) -> None: + """Reads the current values of the subscribed keys and passes them to the handler, so + changes made while the stream was down are not lost.""" + get_req = api_v1.GetConfigurationRequest( + store_name=req.store_name, keys=req.keys, metadata=req.metadata + ) + try: + response = await stub.GetConfiguration( + get_req, timeout=CONFIG_SUBSCRIBE_TIMEOUT_SECONDS + ) + except asyncio.CancelledError: + raise + except Exception as error: + if not self._stopping: + logger.warning( + 'Could not read the current configuration for keys %s on store %s after ' + 'reconnecting; changes made while the stream was down may be missed: %s', + self.keys, + self.store_name, + describe_config_error(error), + ) + return + if len(response.items) > 0 and not self._stopping: + await self._deliver(handler, ConfigurationResponse(response.items)) async def _deliver( self, diff --git a/dapr/aio/clients/grpc/client.py b/dapr/aio/clients/grpc/client.py index b54c07e35..f429f90ca 100644 --- a/dapr/aio/clients/grpc/client.py +++ b/dapr/aio/clients/grpc/client.py @@ -1246,20 +1246,29 @@ async def subscribe_configuration( if not store_name or len(store_name) == 0 or len(store_name.strip()) == 0: raise ValueError('Config store name cannot be empty to get the configuration') - configWatcher = AsyncConfigurationWatcher() + configWatcher = AsyncConfigurationWatcher(on_exit=self._forget_config_watcher) id = await configWatcher.watch_configuration( self._stub, store_name, keys, handler, config_metadata ) - if id: + # A watcher that already gave up has nothing left to unsubscribe. + if id and not configWatcher.exited: self._config_watchers[(store_name, id)] = configWatcher return id + def _forget_config_watcher(self, watcher: AsyncConfigurationWatcher) -> None: + """Drops a watcher from _config_watchers once its task has exited.""" + key = (watcher.store_name or '', watcher.subscription_id) + if self._config_watchers.get(key) is watcher: + del self._config_watchers[key] + async def unsubscribe_configuration(self, store_name: str, id: str) -> bool: """Unsubscribes from configuration changes. If the subscription was re-established after the stream broke (for example after a sidecar restart), the sidecar knows it under a new id; that id is used here, so the id - returned by subscribe_configuration stays valid. + returned by subscribe_configuration stays valid. If the watcher is between streams + (waiting to reconnect), there is nothing to remove on the sidecar: the watcher is + stopped locally and True is returned without calling the sidecar. Args: store_name (str): the state store name to unsubscribe from @@ -1269,18 +1278,24 @@ async def unsubscribe_configuration(self, store_name: str, id: str) -> bool: bool: True if unsubscribed successfully, False otherwise """ watcher = self._config_watchers.pop((store_name, id), None) - server_id = id - if watcher is not None: - watcher.request_stop() - server_id = watcher.id or id + if watcher is None: + return await self._send_unsubscribe_configuration(store_name, id) + watcher.request_stop() + server_id = watcher.live_stream_id() try: - req = api_v1.UnsubscribeConfigurationRequest(store_name=store_name, id=server_id) - response: api_v1.UnsubscribeConfigurationResponse = ( - await self._stub.UnsubscribeConfiguration(req) - ) + if server_id is None: + # Between streams (waiting to reconnect) the sidecar has no subscription to + # remove, so stopping the watcher is all there is to do. + return True + return await self._send_unsubscribe_configuration(store_name, server_id) finally: - if watcher is not None: - await watcher.stop() + await watcher.stop() + + async def _send_unsubscribe_configuration(self, store_name: str, id: str) -> bool: + req = api_v1.UnsubscribeConfigurationRequest(store_name=store_name, id=id) + response: api_v1.UnsubscribeConfigurationResponse = ( + await self._stub.UnsubscribeConfiguration(req) + ) return response.ok async def try_lock( diff --git a/dapr/clients/grpc/_response.py b/dapr/clients/grpc/_response.py index ab3b5080c..2348eccb9 100644 --- a/dapr/clients/grpc/_response.py +++ b/dapr/clients/grpc/_response.py @@ -24,6 +24,7 @@ from enum import Enum from typing import ( TYPE_CHECKING, + Any, Callable, Dict, Generator, @@ -686,8 +687,9 @@ def items(self) -> Dict[Text, ConfigurationItem]: CONFIG_STABLE_STREAM_SECONDS = 10.0 # How long stop() waits for the watcher thread to exit after cancelling the stream. _CONFIG_WATCHER_JOIN_TIMEOUT_SECONDS = 2.0 -# Errors that will not go away by resubscribing (bad request, missing store, auth). -_CONFIG_NON_RETRYABLE_CODES = frozenset( +# Errors that stop a subscription that was never established: subscribe_configuration then +# returns None straight away instead of retrying a request the sidecar rejected. +_CONFIG_FIRST_SUBSCRIBE_NON_RETRYABLE_CODES = frozenset( { StatusCode.INVALID_ARGUMENT, StatusCode.NOT_FOUND, @@ -696,6 +698,18 @@ def items(self) -> Dict[Text, ConfigurationItem]: StatusCode.UNIMPLEMENTED, } ) +# Errors that stop a subscription that was established before. daprd reports any failure of +# the store's Subscribe call, including a backend that is briefly unreachable, as +# INVALID_ARGUMENT, and a store that is not loaded yet after a restart as INVALID_ARGUMENT +# too, so those are retried. Only errors a retry with the same request and credentials cannot +# fix stop the subscription: the API is missing, or the sidecar rejects the caller. +_CONFIG_RESUBSCRIBE_NON_RETRYABLE_CODES = frozenset( + { + StatusCode.PERMISSION_DENIED, + StatusCode.UNAUTHENTICATED, + StatusCode.UNIMPLEMENTED, + } +) def config_reconnect_delay(attempt: int) -> float: @@ -713,31 +727,58 @@ def config_stream_was_stable(established_at: float) -> bool: return time.monotonic() - established_at >= CONFIG_STABLE_STREAM_SECONDS -def is_retryable_config_error(error: BaseException) -> bool: +def is_retryable_config_error(error: BaseException, established: bool = False) -> bool: """Returns True if a configuration subscription stream should be re-established after - ``error``. Only gRPC errors are retried, except the ones in _CONFIG_NON_RETRYABLE_CODES.""" + ``error``. + + Only gRPC errors are retried. ``established`` tells whether the subscription has had a + stream before: until then the codes in _CONFIG_FIRST_SUBSCRIBE_NON_RETRYABLE_CODES stop it, + afterwards only the codes in _CONFIG_RESUBSCRIBE_NON_RETRYABLE_CODES do. + """ if not isinstance(error, RpcError): return False code = error.code() if callable(getattr(error, 'code', None)) else None - return code not in _CONFIG_NON_RETRYABLE_CODES + if established: + return code not in _CONFIG_RESUBSCRIBE_NON_RETRYABLE_CODES + return code not in _CONFIG_FIRST_SUBSCRIBE_NON_RETRYABLE_CODES + + +def describe_config_error(error: BaseException) -> str: + """Returns a one-line description of ``error`` for log messages: the status code and + details for gRPC errors, instead of the multi-line repr of the call object.""" + if isinstance(error, RpcError): + code = error.code() if callable(getattr(error, 'code', None)) else None + details = error.details() if callable(getattr(error, 'details', None)) else None + name = getattr(code, 'name', None) or 'UNKNOWN' + return f'{name}: {details}' if details else name + return f'{type(error).__name__}: {error}' class ConfigurationWatcher: """Reads a SubscribeConfigurationAlpha1 stream on a daemon thread and calls the handler for every update. - If the stream fails with a retryable error or is closed by the sidecar (for example when - daprd restarts), the watcher subscribes again with the same request, with exponential - backoff, until stop() is called or a non-retryable error is returned. + If the stream fails or is closed by the sidecar (for example when daprd restarts), the + watcher subscribes again with the same request, with exponential backoff, until stop() is + called or a non-retryable error is returned (see is_retryable_config_error). Each new stream gets a new id from the sidecar. ``subscription_id`` is the id of the first stream and stays the same; ``id`` is the id of the current stream and is the one to pass to - UnsubscribeConfiguration. + UnsubscribeConfiguration. The handler always receives ``subscription_id``. + + A new stream only sends items for changes made after it opened. So after each reconnect + (not after the first subscribe) the watcher reads the current values of the keys with + GetConfiguration and passes them to the handler, if there are any. That delivers values + that changed while the stream was down, and may repeat values that did not change. If the + read fails, a warning is logged and the subscription carries on. + + ``on_exit`` is called with the watcher when its thread exits, whether it was stopped or + gave up on its own. """ - def __init__(self): - self.store_name = None - self.keys = None + def __init__(self, on_exit: Optional[Callable[['ConfigurationWatcher'], None]] = None): + self.store_name: Optional[str] = None + self.keys: Optional[List[str]] = None self.event: threading.Event = threading.Event() self.id: str = '' self.subscription_id: str = '' @@ -747,6 +788,8 @@ def __init__(self): self._thread: Optional[threading.Thread] = None self._stream_established = False self._established_at = 0.0 + self._on_exit = on_exit + self._exited = False def watch_configuration( self, @@ -784,6 +827,19 @@ def watch_configuration( def stopped(self) -> bool: return self._stop_event.is_set() + @property + def exited(self) -> bool: + """True once the watcher thread has finished (stopped or gave up).""" + return self._exited + + def live_stream_id(self) -> Optional[str]: + """Returns the id of the current stream if one is open and the sidecar has sent its id, + or None while the watcher is between streams (for example waiting to reconnect).""" + with self._lock: + if self._call is not None and self._stream_established: + return self.id + return None + def request_stop(self) -> None: """Marks the watcher as stopping so it does not reconnect when the stream ends.""" self._stop_event.set() @@ -802,9 +858,7 @@ def cancel_stream(self) -> None: """ with self._lock: call = self._call - cancel = getattr(call, 'cancel', None) - if callable(cancel): - cancel() + cancel_config_call(call) def join(self, timeout: Optional[float] = None) -> None: """Waits up to ``timeout`` seconds (default _CONFIG_WATCHER_JOIN_TIMEOUT_SECONDS) for the @@ -822,9 +876,11 @@ def _read_subscribe_config( handler: Callable[[Text, ConfigurationResponse], None], ) -> None: attempt = 0 + outage_reported = False try: while not self._stop_event.is_set(): - self._stream_established = False + with self._lock: + self._stream_established = False try: self._consume_stream(stub, req, handler) if not self._stop_event.is_set(): @@ -837,21 +893,26 @@ def _read_subscribe_config( except Exception as error: if self._stop_event.is_set(): break - if not is_retryable_config_error(error): + if not is_retryable_config_error(error, established=bool(self.subscription_id)): logger.error( 'Configuration subscription for keys %s on store %s stopped: %s', self.keys, self.store_name, - error, + describe_config_error(error), ) break - logger.warning( + # Warn once per outage; the retries after that go to DEBUG until a new + # stream is up, which is logged at INFO. + first_failure = self._stream_established or not outage_reported + logger.log( + logging.WARNING if first_failure else logging.DEBUG, 'Configuration subscription stream for keys %s on store %s failed, ' 'reconnecting: %s', self.keys, self.store_name, - error, + describe_config_error(error), ) + outage_reported = True finally: with self._lock: self._call = None @@ -864,11 +925,17 @@ def _read_subscribe_config( if self._wait_before_retry(delay): break finally: + self._exited = True # Unblock watch_configuration if the first stream never delivered an id. self.event.set() logger.debug( 'Configuration watcher for keys %s on store %s exited', self.keys, self.store_name ) + if self._on_exit is not None: + try: + self._on_exit(self) + except Exception: + logger.exception('Configuration watcher exit callback raised an exception') def _consume_stream( self, @@ -879,36 +946,64 @@ def _consume_stream( call = stub.SubscribeConfigurationAlpha1(req) with self._lock: if self._stop_event.is_set(): - cancel = getattr(call, 'cancel', None) - if callable(cancel): - cancel() + cancel_config_call(call) return self._call = call for response in call: if not self._stream_established: - self._on_stream_established(response.id) + reconnected = self._on_stream_established(response.id) + if reconnected and not self._stop_event.is_set(): + self._deliver_current_values(stub, req, handler) if self._stop_event.is_set(): return if len(response.items) > 0: self._deliver(handler, ConfigurationResponse(response.items)) - def _on_stream_established(self, server_id: str) -> None: + def _on_stream_established(self, server_id: str) -> bool: + """Records the id of a new stream. Returns True if it replaces an earlier stream.""" with self._lock: self.id = server_id - if not self.subscription_id: + reconnected = bool(self.subscription_id) + if not reconnected: self.subscription_id = server_id - else: - logger.info( - 'Configuration subscription %s for keys %s on store %s reconnected ' - 'with new id %s', - self.subscription_id, + self._established_at = time.monotonic() + self._stream_established = True + if reconnected: + logger.info( + 'Configuration subscription %s for keys %s on store %s reconnected with new id %s', + self.subscription_id, + self.keys, + self.store_name, + server_id, + ) + self.event.set() + return reconnected + + def _deliver_current_values( + self, + stub: api_service_v1.DaprStub, + req: api_v1.SubscribeConfigurationRequest, + handler: Callable[[Text, ConfigurationResponse], None], + ) -> None: + """Reads the current values of the subscribed keys and passes them to the handler, so + changes made while the stream was down are not lost.""" + get_req = api_v1.GetConfigurationRequest( + store_name=req.store_name, keys=req.keys, metadata=req.metadata + ) + try: + response = stub.GetConfiguration(get_req, timeout=CONFIG_SUBSCRIBE_TIMEOUT_SECONDS) + except Exception as error: + if not self._stop_event.is_set(): + logger.warning( + 'Could not read the current configuration for keys %s on store %s after ' + 'reconnecting; changes made while the stream was down may be missed: %s', self.keys, self.store_name, - server_id, + describe_config_error(error), ) - self._established_at = time.monotonic() - self._stream_established = True - self.event.set() + return + if len(response.items) > 0 and not self._stop_event.is_set(): + self._deliver(handler, ConfigurationResponse(response.items)) def _deliver( self, @@ -929,6 +1024,13 @@ def _wait_before_retry(self, delay: float) -> bool: return self._stop_event.wait(delay) +def cancel_config_call(call: Any) -> None: + """Cancels a gRPC call if the object supports it (instrumentation wrappers may not).""" + cancel = getattr(call, 'cancel', None) + if callable(cancel): + cancel() + + class BulkPublishResponseFailedEntry: """A failed entry from the bulk publish response. diff --git a/dapr/clients/grpc/client.py b/dapr/clients/grpc/client.py index 671a20aae..3897dbbdc 100644 --- a/dapr/clients/grpc/client.py +++ b/dapr/clients/grpc/client.py @@ -16,6 +16,7 @@ import socket import threading import time +import weakref from typing import Any, Callable, Dict, List, Optional, Sequence, Text, Tuple, Union from urllib.parse import urlencode from warnings import warn @@ -1319,21 +1320,43 @@ def subscribe_configuration( if not store_name or len(store_name) == 0 or len(store_name.strip()) == 0: raise ValueError('Config store name cannot be empty to get the configuration') - configWatcher = ConfigurationWatcher() + configWatcher = ConfigurationWatcher(on_exit=self._config_watcher_exit_callback()) id = configWatcher.watch_configuration( self._stub, store_name, keys, handler, config_metadata ) if id: with self._config_watchers_lock: - self._config_watchers[(store_name, id)] = configWatcher + # A watcher that already gave up has nothing left to unsubscribe. + if not configWatcher.exited: + self._config_watchers[(store_name, id)] = configWatcher return id + def _config_watcher_exit_callback(self) -> Callable[[ConfigurationWatcher], None]: + # Hold the client weakly so a running watcher does not keep an unclosed client alive. + client_ref = weakref.ref(self) + + def on_exit(watcher: ConfigurationWatcher) -> None: + client = client_ref() + if client is not None: + client._forget_config_watcher(watcher) + + return on_exit + + def _forget_config_watcher(self, watcher: ConfigurationWatcher) -> None: + """Drops a watcher from _config_watchers once its thread has exited.""" + key = (watcher.store_name or '', watcher.subscription_id) + with self._config_watchers_lock: + if self._config_watchers.get(key) is watcher: + del self._config_watchers[key] + def unsubscribe_configuration(self, store_name: str, id: str) -> bool: """Unsubscribes from configuration changes. If the subscription was re-established after the stream broke (for example after a sidecar restart), the sidecar knows it under a new id; that id is used here, so the id - returned by subscribe_configuration stays valid. + returned by subscribe_configuration stays valid. If the watcher is between streams + (waiting to reconnect), there is nothing to remove on the sidecar: the watcher is + stopped locally and True is returned without calling the sidecar. Args: store_name (str): the state store name to unsubscribe from @@ -1344,18 +1367,22 @@ def unsubscribe_configuration(self, store_name: str, id: str) -> bool: """ with self._config_watchers_lock: watcher = self._config_watchers.pop((store_name, id), None) - server_id = id - if watcher is not None: - watcher.request_stop() - server_id = watcher.id or id + if watcher is None: + return self._send_unsubscribe_configuration(store_name, id) + watcher.request_stop() + server_id = watcher.live_stream_id() try: - req = api_v1.UnsubscribeConfigurationRequest(store_name=store_name, id=server_id) - response: api_v1.UnsubscribeConfigurationResponse = self._stub.UnsubscribeConfiguration( - req - ) + if server_id is None: + # Between streams (waiting to reconnect) the sidecar has no subscription to + # remove, so stopping the watcher is all there is to do. + return True + return self._send_unsubscribe_configuration(store_name, server_id) finally: - if watcher is not None: - watcher.stop() + watcher.stop() + + def _send_unsubscribe_configuration(self, store_name: str, id: str) -> bool: + req = api_v1.UnsubscribeConfigurationRequest(store_name=store_name, id=id) + response: api_v1.UnsubscribeConfigurationResponse = self._stub.UnsubscribeConfiguration(req) return response.ok def try_lock( diff --git a/tests/clients/fake_dapr_server.py b/tests/clients/fake_dapr_server.py index 45da06846..291a93df3 100644 --- a/tests/clients/fake_dapr_server.py +++ b/tests/clients/fake_dapr_server.py @@ -74,12 +74,22 @@ def __init__(self, grpc_port: int = 50001, http_port: int = 8080): # {key: value} dicts pushed after the first response), and 'end' ('abort' fails the # stream with 'code' (default UNAVAILABLE), 'eof' closes cleanly, 'hold' keeps the stream # open until the id is unsubscribed or the client goes away). 'reject' fails the call with - # that status code before any response is sent. + # that status code before any response is sent. 'wait' (a threading.Event) delays the + # 'end' behaviour until the event is set. self.config_stream_plans: List[Dict[str, Any]] = [] self.config_subscribe_requests: List[api_v1.SubscribeConfigurationRequest] = [] self.config_unsubscribe_requests: List[api_v1.UnsubscribeConfigurationRequest] = [] self._config_unsubscribed: set = set() + # Ids of SubscribeConfigurationAlpha1 streams that are open; like daprd, + # UnsubscribeConfiguration returns ok=False for any other id. + self._config_active_ids: set = set() self._config_lock = threading.Lock() + # GetConfiguration: when config_values is set, it returns those values (for the + # requested keys, or all of them if no keys are given) instead of the default + # 'value' for every key; config_get_error makes it fail with that status code. + self.config_values: Optional[Dict[str, str]] = None + self.config_get_error: Optional[grpc.StatusCode] = None + self.config_get_requests: List[api_v1.GetConfigurationRequest] = [] def set_bulk_publish_unimplemented_on_stable_next(self) -> None: """Make the next BulkPublishEvent (stable) call return UNIMPLEMENTED. @@ -496,7 +506,18 @@ def GetBulkSecret(self, request, context) -> api_v1.GetBulkSecretResponse: return api_v1.GetBulkSecretResponse(data=resp) def GetConfiguration(self, request, context): + with self._config_lock: + self.config_get_requests.append(request) + values = None if self.config_values is None else dict(self.config_values) + get_error = self.config_get_error + if get_error is not None: + context.abort(get_error, 'GetConfiguration failed by test setting') items = dict() + if values is not None: + for key, value in values.items(): + if not request.keys or key in request.keys: + items[key] = ConfigurationItem(value=value, version='2') + return api_v1.GetConfigurationResponse(items=items) for key in request.keys: items[str(key)] = ConfigurationItem(value='value', version='1.5.0') return api_v1.GetConfigurationResponse(items=items) @@ -518,26 +539,44 @@ def SubscribeConfigurationAlpha1(self, request, context): if plan.get('reject') is not None: context.abort(plan['reject'], 'rejected by test plan') sub_id = plan.get('id', 'sub-id') - yield api_v1.SubscribeConfigurationResponse(id=sub_id) - for update in plan.get('updates', []): - items = { - key: ConfigurationItem(value=value, version='1') for key, value in update.items() - } - yield api_v1.SubscribeConfigurationResponse(id=sub_id, items=items) - - end_behavior = plan.get('end', 'hold') - if end_behavior == 'abort': - context.abort(plan.get('code', grpc.StatusCode.UNAVAILABLE), 'simulated disconnection') - elif end_behavior == 'hold': - while context.is_active(): - with self._config_lock: - if sub_id in self._config_unsubscribed: - return - time.sleep(0.01) + with self._config_lock: + self._config_active_ids.add(sub_id) + try: + yield api_v1.SubscribeConfigurationResponse(id=sub_id) + for update in plan.get('updates', []): + items = { + key: ConfigurationItem(value=value, version='1') + for key, value in update.items() + } + yield api_v1.SubscribeConfigurationResponse(id=sub_id, items=items) + + gate = plan.get('wait') + if gate is not None: + # Hold the stream open until the test sets this threading.Event. + gate.wait(timeout=10) + end_behavior = plan.get('end', 'hold') + if end_behavior == 'abort': + context.abort( + plan.get('code', grpc.StatusCode.UNAVAILABLE), 'simulated disconnection' + ) + elif end_behavior == 'hold': + while context.is_active(): + with self._config_lock: + if sub_id in self._config_unsubscribed: + return + time.sleep(0.01) + finally: + with self._config_lock: + self._config_active_ids.discard(sub_id) def UnsubscribeConfiguration(self, request, context): with self._config_lock: self.config_unsubscribe_requests.append(request) + if request.id not in self._config_active_ids: + return api_v1.UnsubscribeConfigurationResponse( + ok=False, message=f'subscription {request.id} does not exist' + ) + self._config_active_ids.discard(request.id) self._config_unsubscribed.add(request.id) return api_v1.UnsubscribeConfigurationResponse(ok=True) diff --git a/tests/clients/test_configuration_watcher.py b/tests/clients/test_configuration_watcher.py index 0ce02f9d3..b960c5432 100644 --- a/tests/clients/test_configuration_watcher.py +++ b/tests/clients/test_configuration_watcher.py @@ -16,7 +16,7 @@ import threading import time import unittest -from typing import Callable, List, Tuple +from typing import Callable, Iterator, List, Tuple from unittest.mock import patch import grpc @@ -27,6 +27,7 @@ ConfigurationResponse, ConfigurationWatcher, config_reconnect_delay, + describe_config_error, is_retryable_config_error, ) from dapr.clients.grpc.client import DaprGrpcClient @@ -48,12 +49,34 @@ def wait_until(condition: Callable[[], bool], timeout: float = WAIT_TIMEOUT_SECO class _FakeRpcError(grpc.RpcError): - def __init__(self, code: grpc.StatusCode): + def __init__(self, code: grpc.StatusCode, details: str = ''): self._code = code + self._details = details def code(self) -> grpc.StatusCode: return self._code + def details(self) -> str: + return self._details + + +class _CancelRecordingCall: + """Wraps a stream call and records cancel(). The call may be a plain generator when + OpenTelemetry grpc instrumentation is active, so it is not asked whether it was cancelled.""" + + def __init__(self, call: Iterator): + self._call = call + self.cancel_called = False + + def cancel(self) -> None: + self.cancel_called = True + cancel = getattr(self._call, 'cancel', None) + if callable(cancel): + cancel() + + def __iter__(self) -> Iterator: + return iter(self._call) + class ConfigurationRetryHelpersTests(unittest.TestCase): def test_delay_doubles_and_is_capped(self): @@ -83,6 +106,33 @@ def test_retryable_classification(self): self.assertFalse(is_retryable_config_error(_FakeRpcError(code)), code) self.assertFalse(is_retryable_config_error(ValueError('closed channel'))) + def test_retryable_classification_after_the_subscription_was_established(self): + # daprd reports a failed store Subscribe (e.g. backend unreachable) and a store that + # is not loaded as INVALID_ARGUMENT; once a subscription existed those are retried. + for code in ( + grpc.StatusCode.INVALID_ARGUMENT, + grpc.StatusCode.NOT_FOUND, + grpc.StatusCode.FAILED_PRECONDITION, + grpc.StatusCode.UNAVAILABLE, + grpc.StatusCode.INTERNAL, + ): + self.assertTrue(is_retryable_config_error(_FakeRpcError(code), established=True), code) + for code in ( + grpc.StatusCode.UNIMPLEMENTED, + grpc.StatusCode.PERMISSION_DENIED, + grpc.StatusCode.UNAUTHENTICATED, + ): + self.assertFalse(is_retryable_config_error(_FakeRpcError(code), established=True), code) + self.assertFalse(is_retryable_config_error(ValueError('closed'), established=True)) + + def test_describe_config_error_is_one_line(self): + self.assertEqual( + describe_config_error(_FakeRpcError(grpc.StatusCode.UNAVAILABLE, 'sidecar down')), + 'UNAVAILABLE: sidecar down', + ) + self.assertEqual(describe_config_error(_FakeRpcError(grpc.StatusCode.INTERNAL)), 'INTERNAL') + self.assertEqual(describe_config_error(ValueError('closed')), 'ValueError: closed') + class ConfigurationWatcherReconnectTests(unittest.TestCase): grpc_port = 50021 @@ -104,6 +154,10 @@ def setUp(self): server.config_stream_plans.clear() server.config_subscribe_requests.clear() server.config_unsubscribe_requests.clear() + server.config_get_requests.clear() + # No current values unless a test sets them, so reconnects add no handler calls. + server.config_values = {} + server.config_get_error = None self.delays: List[float] = [] self.updates: List[Tuple[str, str]] = [] @@ -184,11 +238,201 @@ def test_unsubscribe_after_reconnect_uses_new_id_and_stops(self): self.assertEqual(len(server.config_subscribe_requests), 2) def test_unsubscribe_unknown_id_is_sent_as_is(self): - self.assertTrue(self.client.unsubscribe_configuration(STORE, 'not-tracked')) + # The client does not track this id, so it goes to the sidecar unchanged and the + # sidecar's answer (ok=False: no such subscription) is returned. + self.assertFalse(self.client.unsubscribe_configuration(STORE, 'not-tracked')) self.assertEqual( [r.id for r in self._fake_dapr_server.config_unsubscribe_requests], ['not-tracked'] ) + def test_unsubscribe_while_reconnecting_stops_locally(self): + server = self._fake_dapr_server + server.config_stream_plans.extend( + [ + {'id': 'first', 'end': 'abort'}, + {'id': 'second', 'updates': [{'k': 'v2'}], 'end': 'hold'}, + ] + ) + waiting = threading.Event() + + def wait_until_stopped(watcher: ConfigurationWatcher, delay: float) -> bool: + waiting.set() + return watcher._stop_event.wait(WAIT_TIMEOUT_SECONDS) + + with patch.object(ConfigurationWatcher, '_wait_before_retry', wait_until_stopped): + subscription_id = self.subscribe() + watcher = self.watcher(subscription_id) + self.assertTrue(waiting.wait(WAIT_TIMEOUT_SECONDS)) + self.assertIsNone(watcher.live_stream_id()) + + self.assertTrue(self.client.unsubscribe_configuration(STORE, subscription_id)) + + # The dead stream's id would get ok=False from the sidecar, so nothing is sent. + self.assertEqual(server.config_unsubscribe_requests, []) + thread = watcher._thread + assert thread is not None + self.assertFalse(thread.is_alive()) + self.assertNotIn((STORE, subscription_id), self.client._config_watchers) + self.assertEqual(len(server.config_subscribe_requests), 1) + + def test_invalid_argument_after_the_subscription_was_established_is_retried(self): + # daprd answers INVALID_ARGUMENT when the store's Subscribe fails, for example while + # its backend is unreachable during the same outage that broke the stream. + self._fake_dapr_server.config_stream_plans.extend( + [ + {'id': 'first', 'updates': [{'k': 'v1'}], 'end': 'abort'}, + {'reject': grpc.StatusCode.INVALID_ARGUMENT}, + {'id': 'third', 'updates': [{'k': 'v3'}], 'end': 'hold'}, + ] + ) + + self.assertEqual(self.subscribe(), 'first') + + self.assertTrue(wait_until(lambda: len(self.updates) == 2), self.updates) + self.assertEqual(self.updates, [('first', 'v1'), ('first', 'v3')]) + self.assertEqual(len(self._fake_dapr_server.config_subscribe_requests), 3) + self.assertEqual(self.watcher('first').live_stream_id(), 'third') + + def test_reconnect_delivers_current_values(self): + server = self._fake_dapr_server + server.config_values = {'k': 'changed-while-down', 'other': 'x'} + server.config_stream_plans.extend( + [ + {'id': 'first', 'updates': [{'k': 'v1'}], 'end': 'abort'}, + {'id': 'second', 'end': 'hold'}, + ] + ) + + subscription_id = self.client.subscribe_configuration( + store_name=STORE, keys=['k'], handler=self.handler, config_metadata={'m': '1'} + ) + + self.assertTrue(wait_until(lambda: len(self.updates) == 2), self.updates) + self.assertEqual(self.updates, [('first', 'v1'), ('first', 'changed-while-down')]) + self.assertEqual(subscription_id, 'first') + self.assertEqual(len(server.config_get_requests), 1) + get_request = server.config_get_requests[0] + self.assertEqual(get_request.store_name, STORE) + self.assertEqual(list(get_request.keys), ['k']) + self.assertEqual(dict(get_request.metadata), {'m': '1'}) + + def test_first_subscribe_does_not_read_current_values(self): + server = self._fake_dapr_server + server.config_values = {'k': 'current'} + server.config_stream_plans.append({'id': 'first', 'updates': [{'k': 'v1'}], 'end': 'hold'}) + + self.subscribe() + + self.assertTrue(wait_until(lambda: len(self.updates) == 1), self.updates) + time.sleep(0.2) + self.assertEqual(self.updates, [('first', 'v1')]) + self.assertEqual(server.config_get_requests, []) + + def test_failed_current_values_read_keeps_the_subscription(self): + server = self._fake_dapr_server + server.config_get_error = grpc.StatusCode.INTERNAL + server.config_stream_plans.extend( + [ + {'id': 'first', 'updates': [{'k': 'v1'}], 'end': 'abort'}, + {'id': 'second', 'updates': [{'k': 'v2'}], 'end': 'hold'}, + ] + ) + + with self.assertLogs('dapr.clients.grpc._response', level='WARNING') as logs: + self.subscribe() + self.assertTrue(wait_until(lambda: len(self.updates) == 2), self.updates) + + self.assertEqual(self.updates, [('first', 'v1'), ('first', 'v2')]) + self.assertEqual(len(server.config_get_requests), 1) + self.assertEqual(len(server.config_subscribe_requests), 2) + self.assertTrue( + any('Could not read the current configuration' in line for line in logs.output), + logs.output, + ) + + def test_watcher_that_gives_up_is_removed_from_the_client(self): + release = threading.Event() + self._fake_dapr_server.config_stream_plans.append( + { + 'id': 'first', + 'end': 'abort', + 'code': grpc.StatusCode.PERMISSION_DENIED, + 'wait': release, + } + ) + subscription_id = self.subscribe() + self.assertIn((STORE, subscription_id), self.client._config_watchers) + + release.set() + + self.assertTrue( + wait_until(lambda: (STORE, subscription_id) not in self.client._config_watchers) + ) + + @patch('dapr.clients.grpc._response._CONFIG_WATCHER_JOIN_TIMEOUT_SECONDS', 30.0) + def test_stop_requested_while_opening_a_stream_cancels_it(self): + release = threading.Event() + self._fake_dapr_server.config_stream_plans.extend( + [ + {'id': 'first', 'end': 'abort', 'wait': release}, + {'id': 'second', 'end': 'hold'}, + ] + ) + subscription_id = self.subscribe() + watcher = self.watcher(subscription_id) + stub = self.client._stub + real_subscribe = stub.SubscribeConfigurationAlpha1 + opened: List[_CancelRecordingCall] = [] + + def subscribe_and_stop(req): + call = _CancelRecordingCall(real_subscribe(req)) + opened.append(call) + # stop() arrives while the watcher is opening the reconnect stream. + watcher.request_stop() + return call + + stub.SubscribeConfigurationAlpha1 = subscribe_and_stop + release.set() + + self.assertTrue(wait_until(lambda: watcher.exited)) + self.assertEqual(len(opened), 1) + self.assertTrue(opened[0].cancel_called) + self.assertIsNone(watcher._call) + started = time.monotonic() + self.client.close() + self.assertLess(time.monotonic() - started, 5.0) + + def test_reconnect_failures_are_logged_concisely_and_once_per_outage(self): + self._fake_dapr_server.config_stream_plans.extend( + [ + {'id': 'first', 'end': 'abort'}, + {'reject': grpc.StatusCode.UNAVAILABLE}, + {'reject': grpc.StatusCode.UNAVAILABLE}, + {'id': 'second', 'updates': [{'k': 'v2'}], 'end': 'hold'}, + ] + ) + + with self.assertLogs('dapr.clients.grpc._response', level='DEBUG') as logs: + self.subscribe() + self.assertTrue(wait_until(lambda: len(self.updates) == 1), self.updates) + + failures = [r for r in logs.records if 'failed, reconnecting' in r.getMessage()] + self.assertEqual( + [r.levelname for r in failures], ['WARNING', 'DEBUG', 'DEBUG'], logs.output + ) + for record in failures: + message = record.getMessage() + self.assertNotIn('\n', message) + self.assertNotIn('Rendezvous', message) + self.assertIn(': UNAVAILABLE: ', message) + self.assertTrue( + any( + r.levelname == 'INFO' and 'reconnected with new id second' in r.getMessage() + for r in logs.records + ), + logs.output, + ) + def test_non_retryable_error_stops_without_reconnecting(self): server = self._fake_dapr_server server.config_stream_plans.extend( diff --git a/tests/clients/test_configuration_watcher_async.py b/tests/clients/test_configuration_watcher_async.py index 49f5bd038..cd3a79e16 100644 --- a/tests/clients/test_configuration_watcher_async.py +++ b/tests/clients/test_configuration_watcher_async.py @@ -14,6 +14,7 @@ """ import asyncio +import threading import time import unittest from typing import Callable, List, Tuple @@ -61,6 +62,10 @@ async def asyncSetUp(self): server.config_stream_plans.clear() server.config_subscribe_requests.clear() server.config_unsubscribe_requests.clear() + server.config_get_requests.clear() + # No current values unless a test sets them, so reconnects add no handler calls. + server.config_values = {} + server.config_get_error = None self.delays: List[float] = [] self.updates: List[Tuple[str, str]] = [] @@ -152,7 +157,7 @@ async def test_non_retryable_error_stops_without_reconnecting(self): 'id': 'first', 'updates': [{'k': 'v1'}], 'end': 'abort', - 'code': grpc.StatusCode.NOT_FOUND, + 'code': grpc.StatusCode.UNIMPLEMENTED, }, {'id': 'second', 'updates': [{'k': 'v2'}], 'end': 'hold'}, ] @@ -227,6 +232,191 @@ async def test_backoff_keeps_growing_when_streams_close_right_after_the_id(self) self.assertGreaterEqual(delay, base) self.assertLessEqual(delay, base * 1.2) + async def test_unsubscribe_unknown_id_is_sent_as_is(self): + self.assertFalse(await self.client.unsubscribe_configuration(STORE, 'not-tracked')) + self.assertEqual( + [r.id for r in self._fake_dapr_server.config_unsubscribe_requests], ['not-tracked'] + ) + + async def test_unsubscribe_while_reconnecting_stops_locally(self): + server = self._fake_dapr_server + server.config_stream_plans.extend( + [ + {'id': 'first', 'end': 'abort'}, + {'id': 'second', 'updates': [{'k': 'v2'}], 'end': 'hold'}, + ] + ) + waiting = asyncio.Event() + + async def wait_forever(watcher: AsyncConfigurationWatcher, delay: float) -> None: + waiting.set() + await asyncio.sleep(WAIT_TIMEOUT_SECONDS) + + with patch.object(AsyncConfigurationWatcher, '_wait_before_retry', wait_forever): + subscription_id = await self.subscribe() + watcher = self.watcher(subscription_id) + await asyncio.wait_for(waiting.wait(), WAIT_TIMEOUT_SECONDS) + self.assertIsNone(watcher.live_stream_id()) + + self.assertTrue(await self.client.unsubscribe_configuration(STORE, subscription_id)) + + self.assertEqual(server.config_unsubscribe_requests, []) + task = watcher._task + assert task is not None + self.assertTrue(task.done()) + self.assertNotIn((STORE, subscription_id), self.client._config_watchers) + self.assertEqual(len(server.config_subscribe_requests), 1) + + async def test_invalid_argument_after_the_subscription_was_established_is_retried(self): + self._fake_dapr_server.config_stream_plans.extend( + [ + {'id': 'first', 'updates': [{'k': 'v1'}], 'end': 'abort'}, + {'reject': grpc.StatusCode.INVALID_ARGUMENT}, + {'id': 'third', 'updates': [{'k': 'v3'}], 'end': 'hold'}, + ] + ) + + self.assertEqual(await self.subscribe(), 'first') + + self.assertTrue(await wait_until(lambda: len(self.updates) == 2), self.updates) + self.assertEqual(self.updates, [('first', 'v1'), ('first', 'v3')]) + self.assertEqual(len(self._fake_dapr_server.config_subscribe_requests), 3) + self.assertEqual(self.watcher('first').live_stream_id(), 'third') + + async def test_reconnect_delivers_current_values(self): + server = self._fake_dapr_server + server.config_values = {'k': 'changed-while-down'} + server.config_stream_plans.extend( + [ + {'id': 'first', 'updates': [{'k': 'v1'}], 'end': 'abort'}, + {'id': 'second', 'end': 'hold'}, + ] + ) + + await self.client.subscribe_configuration( + store_name=STORE, keys=['k'], handler=self.handler, config_metadata={'m': '1'} + ) + + self.assertTrue(await wait_until(lambda: len(self.updates) == 2), self.updates) + self.assertEqual(self.updates, [('first', 'v1'), ('first', 'changed-while-down')]) + self.assertEqual(len(server.config_get_requests), 1) + get_request = server.config_get_requests[0] + self.assertEqual(get_request.store_name, STORE) + self.assertEqual(list(get_request.keys), ['k']) + self.assertEqual(dict(get_request.metadata), {'m': '1'}) + + async def test_first_subscribe_does_not_read_current_values(self): + server = self._fake_dapr_server + server.config_values = {'k': 'current'} + server.config_stream_plans.append({'id': 'first', 'updates': [{'k': 'v1'}], 'end': 'hold'}) + + await self.subscribe() + + self.assertTrue(await wait_until(lambda: len(self.updates) == 1), self.updates) + await asyncio.sleep(0.2) + self.assertEqual(self.updates, [('first', 'v1')]) + self.assertEqual(server.config_get_requests, []) + + async def test_failed_current_values_read_keeps_the_subscription(self): + server = self._fake_dapr_server + server.config_get_error = grpc.StatusCode.INTERNAL + server.config_stream_plans.extend( + [ + {'id': 'first', 'updates': [{'k': 'v1'}], 'end': 'abort'}, + {'id': 'second', 'updates': [{'k': 'v2'}], 'end': 'hold'}, + ] + ) + + with self.assertLogs('dapr.aio.clients.grpc._response', level='WARNING') as logs: + await self.subscribe() + self.assertTrue(await wait_until(lambda: len(self.updates) == 2), self.updates) + + self.assertEqual(self.updates, [('first', 'v1'), ('first', 'v2')]) + self.assertEqual(len(server.config_get_requests), 1) + self.assertEqual(len(server.config_subscribe_requests), 2) + self.assertTrue( + any('Could not read the current configuration' in line for line in logs.output), + logs.output, + ) + + async def test_watcher_that_gives_up_is_removed_from_the_client(self): + release = threading.Event() + self._fake_dapr_server.config_stream_plans.append( + { + 'id': 'first', + 'end': 'abort', + 'code': grpc.StatusCode.PERMISSION_DENIED, + 'wait': release, + } + ) + subscription_id = await self.subscribe() + self.assertIn((STORE, subscription_id), self.client._config_watchers) + + release.set() + + self.assertTrue( + await wait_until(lambda: (STORE, subscription_id) not in self.client._config_watchers) + ) + + async def test_stop_requested_while_opening_a_stream_cancels_it(self): + release = threading.Event() + self._fake_dapr_server.config_stream_plans.extend( + [ + {'id': 'first', 'end': 'abort', 'wait': release}, + {'id': 'second', 'end': 'hold'}, + ] + ) + subscription_id = await self.subscribe() + watcher = self.watcher(subscription_id) + stub = self.client._stub + real_subscribe = stub.SubscribeConfigurationAlpha1 + opened: List[grpc.aio.Call] = [] + + def subscribe_and_stop(req): + call = real_subscribe(req) + opened.append(call) + # stop() arrives while the watcher is opening the reconnect stream. + watcher.request_stop() + return call + + stub.SubscribeConfigurationAlpha1 = subscribe_and_stop + release.set() + + self.assertTrue(await wait_until(lambda: watcher.exited)) + self.assertEqual(len(opened), 1) + self.assertTrue(opened[0].cancelled()) + self.assertIsNone(watcher._call) + + async def test_reconnect_failures_are_logged_concisely_and_once_per_outage(self): + self._fake_dapr_server.config_stream_plans.extend( + [ + {'id': 'first', 'end': 'abort'}, + {'reject': grpc.StatusCode.UNAVAILABLE}, + {'reject': grpc.StatusCode.UNAVAILABLE}, + {'id': 'second', 'updates': [{'k': 'v2'}], 'end': 'hold'}, + ] + ) + + with self.assertLogs('dapr.aio.clients.grpc._response', level='DEBUG') as logs: + await self.subscribe() + self.assertTrue(await wait_until(lambda: len(self.updates) == 1), self.updates) + + failures = [r for r in logs.records if 'failed, reconnecting' in r.getMessage()] + self.assertEqual( + [r.levelname for r in failures], ['WARNING', 'DEBUG', 'DEBUG'], logs.output + ) + for record in failures: + message = record.getMessage() + self.assertNotIn('\n', message) + self.assertIn(': UNAVAILABLE: ', message) + self.assertTrue( + any( + r.levelname == 'INFO' and 'reconnected with new id second' in r.getMessage() + for r in logs.records + ), + logs.output, + ) + async def test_async_handler_is_awaited(self): self._fake_dapr_server.config_stream_plans.append( {'id': 'first', 'updates': [{'k': 'v1'}], 'end': 'hold'} diff --git a/tests/clients/test_dapr_grpc_client.py b/tests/clients/test_dapr_grpc_client.py index 9c8c7c0af..b3be54bc5 100644 --- a/tests/clients/test_dapr_grpc_client.py +++ b/tests/clients/test_dapr_grpc_client.py @@ -882,8 +882,9 @@ def handler(id: str, resp: ConfigurationResponse): def test_unsubscribe_configuration(self): dapr = DaprGrpcClient(f'{self.scheme}localhost:{self.grpc_port}') + # Like daprd, the fake sidecar answers ok=False for an id it has no subscription for. res = dapr.unsubscribe_configuration(store_name='configurationstore', id='k') - self.assertTrue(res) + self.assertFalse(res) def test_query_state(self): dapr = DaprGrpcClient(f'{self.scheme}localhost:{self.grpc_port}') diff --git a/tests/clients/test_dapr_grpc_client_async.py b/tests/clients/test_dapr_grpc_client_async.py index 42638d31e..4665bb2bc 100644 --- a/tests/clients/test_dapr_grpc_client_async.py +++ b/tests/clients/test_dapr_grpc_client_async.py @@ -856,8 +856,9 @@ def handler(id: str, resp: ConfigurationResponse): async def test_unsubscribe_configuration(self): dapr = DaprGrpcClientAsync(f'{self.scheme}localhost:{self.grpc_port}') + # Like daprd, the fake sidecar answers ok=False for an id it has no subscription for. res = await dapr.unsubscribe_configuration(store_name='configurationstore', id='k') - self.assertTrue(res) + self.assertFalse(res) async def test_query_state(self): dapr = DaprGrpcClientAsync(f'{self.scheme}localhost:{self.grpc_port}') From 6b2c024265b123147fb45e7d4ab153140f2f1b93 Mon Sep 17 00:00:00 2001 From: Casper Nielsen Date: Mon, 28 Sep 2026 23:25:48 +0200 Subject: [PATCH 4/6] fix(clients): tighten configuration subscription reconnects - Leave the subscribe-only pgNotifyChannel metadata out of the catch-up GetConfiguration read; the PostgreSQL store filters rows by every metadata entry, so the read matched nothing. - Log a WARNING again every CONFIG_OUTAGE_WARNING_EVERY_N_FAILURES failed reconnect attempts, with the attempt count and duration, so a long outage stays visible. - Treat FAILED_PRECONDITION on the first subscribe as final so subscribe_configuration returns None straight away. - Async client: run plain-function handlers in a worker thread with asyncio.to_thread instead of on the event loop; async handlers are still awaited on the loop, one call at a time. - Document that unsubscribing must go through the subscribing client and that values may repeat or briefly arrive out of order after a reconnect. - Tests for the above, for unsubscribing before a reconnect stream sent its id, and for a watcher that exits before it is registered. Co-Authored-By: Claude Opus 5.5 Signed-off-by: Casper Nielsen --- dapr/aio/clients/grpc/_response.py | 57 ++++-- dapr/aio/clients/grpc/client.py | 18 ++ dapr/clients/grpc/_response.py | 95 ++++++++-- dapr/clients/grpc/client.py | 13 ++ tests/clients/fake_dapr_server.py | 10 +- tests/clients/test_configuration_watcher.py | 144 ++++++++++++++ .../test_configuration_watcher_async.py | 176 +++++++++++++++++- 7 files changed, 479 insertions(+), 34 deletions(-) diff --git a/dapr/aio/clients/grpc/_response.py b/dapr/aio/clients/grpc/_response.py index ae19c46bd..10b617198 100644 --- a/dapr/aio/clients/grpc/_response.py +++ b/dapr/aio/clients/grpc/_response.py @@ -36,10 +36,12 @@ DaprResponse, TCryptoResponse, cancel_config_call, + config_catch_up_metadata, config_reconnect_delay, config_stream_was_stable, describe_config_error, is_retryable_config_error, + log_config_stream_failure, ) from dapr.proto import api_service_v1, api_v1 @@ -132,7 +134,20 @@ class AsyncConfigurationWatcher: After each reconnect (not after the first subscribe) the watcher reads the current values of the keys with GetConfiguration and passes them to the handler, if there are any, so changes made while the stream was down are delivered. This may repeat values that did not - change. If the read fails, a warning is logged and the subscription carries on. + change. If the read fails, a warning is logged and the subscription carries on. The read + leaves out subscribe-only metadata (see config_catch_up_metadata). + + Right after a reconnect, values may repeat and may briefly arrive out of order: an update + that reached the new stream before the read is delivered after the read's result, which + can already hold a newer value. So until the next update to a key, the last value + delivered for it is not guaranteed to be the newest. + + ``async def`` handlers are awaited on the event loop; plain functions run in a worker + thread (asyncio.to_thread) so they cannot block the loop. Either way, one handler call + finishes before the next starts, in the order the updates arrived. + + Failed reconnect attempts are logged like ConfigurationWatcher does (see + CONFIG_OUTAGE_WARNING_EVERY_N_FAILURES). ``on_exit`` is called with the watcher when its task exits, whether it was stopped or gave up on its own. @@ -224,12 +239,16 @@ async def _read_subscribe_config( handler: AsyncConfigurationHandler, ) -> None: attempt = 0 - outage_reported = False + # Consecutive failed attempts since the last stream that delivered an id. + failures = 0 + outage_started_at = 0.0 try: while not self._stopping: self._stream_established = False try: await self._consume_stream(stub, req, handler) + if self._stream_established: + failures = 0 if not self._stopping: logger.info( 'Configuration subscription stream for keys %s on store %s was ' @@ -250,18 +269,14 @@ async def _read_subscribe_config( describe_config_error(error), ) break - # Warn once per outage; the retries after that go to DEBUG until a new - # stream is up, which is logged at INFO. - first_failure = self._stream_established or not outage_reported - logger.log( - logging.WARNING if first_failure else logging.DEBUG, - 'Configuration subscription stream for keys %s on store %s failed, ' - 'reconnecting: %s', - self.keys, - self.store_name, - describe_config_error(error), + if self._stream_established: + failures = 0 + if failures == 0: + outage_started_at = time.monotonic() + failures += 1 + log_config_stream_failure( + logger, self.keys, self.store_name, error, failures, outage_started_at ) - outage_reported = True finally: self._call = None if self._stopping: @@ -334,7 +349,9 @@ async def _deliver_current_values( """Reads the current values of the subscribed keys and passes them to the handler, so changes made while the stream was down are not lost.""" get_req = api_v1.GetConfigurationRequest( - store_name=req.store_name, keys=req.keys, metadata=req.metadata + store_name=req.store_name, + keys=req.keys, + metadata=config_catch_up_metadata(req.metadata), ) try: response = await stub.GetConfiguration( @@ -360,11 +377,15 @@ async def _deliver( handler: AsyncConfigurationHandler, response: ConfigurationResponse, ) -> None: - # Accept both plain functions and ``async def`` handlers. + # ``async def`` handlers run on the loop; plain functions run in a worker thread so a + # blocking handler does not stall the loop. Awaiting each call keeps delivery in order. try: - result = handler(self.subscription_id, response) - if inspect.isawaitable(result): - await result + if inspect.iscoroutinefunction(handler): + await handler(self.subscription_id, response) + else: + result = await asyncio.to_thread(handler, self.subscription_id, response) + if inspect.isawaitable(result): + await result except Exception: logger.exception( 'Configuration handler for keys %s on store %s raised an exception', diff --git a/dapr/aio/clients/grpc/client.py b/dapr/aio/clients/grpc/client.py index f429f90ca..b942f7a68 100644 --- a/dapr/aio/clients/grpc/client.py +++ b/dapr/aio/clients/grpc/client.py @@ -1240,6 +1240,19 @@ async def subscribe_configuration( handler(func (key, ConfigurationResponse)): the callback function to be called config_metadata (Dict[str, str], optional): Dapr metadata for configuration + If the stream breaks or the sidecar closes it (for example when daprd restarts), the + client subscribes again with backoff until the subscription is unsubscribed through + this client, or this client is closed. The handler keeps receiving the id returned + here. Only unsubscribe through the client that subscribed (or close that client): if + an UnsubscribeConfiguration for the id is sent any other way, for example through a + different client, the stream ends as if the sidecar had restarted and this client + subscribes again. + + The handler may be a plain function or an ``async def`` function. Async handlers run + on the event loop. Plain functions run in a worker thread (asyncio.to_thread), so a + slow handler does not block the event loop. Handler calls run one at a time, in the + order the updates arrive. + Returns: id (str): subscription id, which can be used to unsubscribe later """ @@ -1270,6 +1283,11 @@ async def unsubscribe_configuration(self, store_name: str, id: str) -> bool: (waiting to reconnect), there is nothing to remove on the sidecar: the watcher is stopped locally and True is returned without calling the sidecar. + Unsubscribe through the client that subscribed. The subscription is re-established + whenever its stream ends without this client having unsubscribed, so an + UnsubscribeConfiguration sent through another client only ends the current stream + and this client subscribes again. + Args: store_name (str): the state store name to unsubscribe from id (str): the subscription id to unsubscribe diff --git a/dapr/clients/grpc/_response.py b/dapr/clients/grpc/_response.py index 2348eccb9..ec2083345 100644 --- a/dapr/clients/grpc/_response.py +++ b/dapr/clients/grpc/_response.py @@ -693,6 +693,7 @@ def items(self) -> Dict[Text, ConfigurationItem]: { StatusCode.INVALID_ARGUMENT, StatusCode.NOT_FOUND, + StatusCode.FAILED_PRECONDITION, StatusCode.PERMISSION_DENIED, StatusCode.UNAUTHENTICATED, StatusCode.UNIMPLEMENTED, @@ -711,6 +712,60 @@ def items(self) -> Dict[Text, ConfigurationItem]: } ) +# While reconnecting keeps failing, the failure is logged at WARNING on the first attempt and +# on every Nth consecutive failed attempt after it, and at DEBUG in between, so an outage that +# does not end (for example a store removed from the sidecar) stays visible in the logs. +CONFIG_OUTAGE_WARNING_EVERY_N_FAILURES = 10 +# Request metadata keys (compared case-insensitively) that only apply to subscribing and are +# left out of the GetConfiguration read after a reconnect. The PostgreSQL store turns every +# GetConfiguration metadata entry into a column filter, so passing its notify channel there +# would match no rows. +_CONFIG_SUBSCRIBE_ONLY_METADATA_KEYS = frozenset({'pgnotifychannel'}) + + +def config_catch_up_metadata(subscribe_metadata: Mapping[str, str]) -> Dict[str, str]: + """Returns the metadata for the GetConfiguration read after a reconnect: the subscribe + metadata without the subscribe-only keys.""" + return { + key: value + for key, value in subscribe_metadata.items() + if key.lower() not in _CONFIG_SUBSCRIBE_ONLY_METADATA_KEYS + } + + +def log_config_stream_failure( + log: logging.Logger, + keys: Optional[List[str]], + store_name: Optional[str], + error: BaseException, + failures: int, + outage_started_at: float, +) -> None: + """Logs failed reconnect attempt number ``failures`` (1-based) of the current outage, which + started at ``outage_started_at`` (time.monotonic()). See + CONFIG_OUTAGE_WARNING_EVERY_N_FAILURES for the log level.""" + warn = failures == 1 or failures % CONFIG_OUTAGE_WARNING_EVERY_N_FAILURES == 0 + level = logging.WARNING if warn else logging.DEBUG + if failures == 1: + log.log( + level, + 'Configuration subscription stream for keys %s on store %s failed, reconnecting: %s', + keys, + store_name, + describe_config_error(error), + ) + return + log.log( + level, + 'Configuration subscription stream for keys %s on store %s failed, reconnecting: %s ' + '(still failing after %d attempts over %.0f s)', + keys, + store_name, + describe_config_error(error), + failures, + time.monotonic() - outage_started_at, + ) + def config_reconnect_delay(attempt: int) -> float: """Returns the backoff delay in seconds before reconnect attempt ``attempt`` (0-based).""" @@ -770,7 +825,17 @@ class ConfigurationWatcher: (not after the first subscribe) the watcher reads the current values of the keys with GetConfiguration and passes them to the handler, if there are any. That delivers values that changed while the stream was down, and may repeat values that did not change. If the - read fails, a warning is logged and the subscription carries on. + read fails, a warning is logged and the subscription carries on. The read leaves out + subscribe-only metadata (see config_catch_up_metadata). + + Right after a reconnect, values may repeat and may briefly arrive out of order: an update + that reached the new stream before the read is delivered after the read's result, which + can already hold a newer value. So until the next update to a key, the last value + delivered for it is not guaranteed to be the newest. + + Every failed reconnect attempt is logged (see CONFIG_OUTAGE_WARNING_EVERY_N_FAILURES for + which ones are warnings). The watcher keeps retrying errors that may clear up, including a + store that is missing on the sidecar, for as long as the subscription is not stopped. ``on_exit`` is called with the watcher when its thread exits, whether it was stopped or gave up on its own. @@ -876,13 +941,17 @@ def _read_subscribe_config( handler: Callable[[Text, ConfigurationResponse], None], ) -> None: attempt = 0 - outage_reported = False + # Consecutive failed attempts since the last stream that delivered an id. + failures = 0 + outage_started_at = 0.0 try: while not self._stop_event.is_set(): with self._lock: self._stream_established = False try: self._consume_stream(stub, req, handler) + if self._stream_established: + failures = 0 if not self._stop_event.is_set(): logger.info( 'Configuration subscription stream for keys %s on store %s was ' @@ -901,18 +970,14 @@ def _read_subscribe_config( describe_config_error(error), ) break - # Warn once per outage; the retries after that go to DEBUG until a new - # stream is up, which is logged at INFO. - first_failure = self._stream_established or not outage_reported - logger.log( - logging.WARNING if first_failure else logging.DEBUG, - 'Configuration subscription stream for keys %s on store %s failed, ' - 'reconnecting: %s', - self.keys, - self.store_name, - describe_config_error(error), + if self._stream_established: + failures = 0 + if failures == 0: + outage_started_at = time.monotonic() + failures += 1 + log_config_stream_failure( + logger, self.keys, self.store_name, error, failures, outage_started_at ) - outage_reported = True finally: with self._lock: self._call = None @@ -988,7 +1053,9 @@ def _deliver_current_values( """Reads the current values of the subscribed keys and passes them to the handler, so changes made while the stream was down are not lost.""" get_req = api_v1.GetConfigurationRequest( - store_name=req.store_name, keys=req.keys, metadata=req.metadata + store_name=req.store_name, + keys=req.keys, + metadata=config_catch_up_metadata(req.metadata), ) try: response = stub.GetConfiguration(get_req, timeout=CONFIG_SUBSCRIBE_TIMEOUT_SECONDS) diff --git a/dapr/clients/grpc/client.py b/dapr/clients/grpc/client.py index 3897dbbdc..b8b3cb088 100644 --- a/dapr/clients/grpc/client.py +++ b/dapr/clients/grpc/client.py @@ -1313,6 +1313,14 @@ def subscribe_configuration( handler(func (key, ConfigurationResponse)): the callback function to be called config_metadata (Dict[str, str], optional): Dapr metadata for configuration + If the stream breaks or the sidecar closes it (for example when daprd restarts), the + client subscribes again with backoff until the subscription is unsubscribed through + this client, or this client is closed. The handler keeps receiving the id returned + here. Only unsubscribe through the client that subscribed (or close that client): if + an UnsubscribeConfiguration for the id is sent any other way, for example through a + different client, the stream ends as if the sidecar had restarted and this client + subscribes again. + Returns: id (str): subscription id, which can be used to unsubscribe later """ @@ -1358,6 +1366,11 @@ def unsubscribe_configuration(self, store_name: str, id: str) -> bool: (waiting to reconnect), there is nothing to remove on the sidecar: the watcher is stopped locally and True is returned without calling the sidecar. + Unsubscribe through the client that subscribed. The subscription is re-established + whenever its stream ends without this client having unsubscribed, so an + UnsubscribeConfiguration sent through another client only ends the current stream + and this client subscribes again. + Args: store_name (str): the state store name to unsubscribe from id (str): the subscription id to unsubscribe diff --git a/tests/clients/fake_dapr_server.py b/tests/clients/fake_dapr_server.py index 291a93df3..6e2f26caa 100644 --- a/tests/clients/fake_dapr_server.py +++ b/tests/clients/fake_dapr_server.py @@ -75,7 +75,9 @@ def __init__(self, grpc_port: int = 50001, http_port: int = 8080): # stream with 'code' (default UNAVAILABLE), 'eof' closes cleanly, 'hold' keeps the stream # open until the id is unsubscribed or the client goes away). 'reject' fails the call with # that status code before any response is sent. 'wait' (a threading.Event) delays the - # 'end' behaviour until the event is set. + # 'end' behaviour until the event is set. 'wait_before_id' (a threading.Event) keeps the + # stream open without sending anything, not even the id, until the event is set or the + # client goes away. self.config_stream_plans: List[Dict[str, Any]] = [] self.config_subscribe_requests: List[api_v1.SubscribeConfigurationRequest] = [] self.config_unsubscribe_requests: List[api_v1.UnsubscribeConfigurationRequest] = [] @@ -539,6 +541,12 @@ def SubscribeConfigurationAlpha1(self, request, context): if plan.get('reject') is not None: context.abort(plan['reject'], 'rejected by test plan') sub_id = plan.get('id', 'sub-id') + id_gate = plan.get('wait_before_id') + if id_gate is not None: + while not id_gate.is_set(): + if not context.is_active(): + return + id_gate.wait(0.01) with self._config_lock: self._config_active_ids.add(sub_id) try: diff --git a/tests/clients/test_configuration_watcher.py b/tests/clients/test_configuration_watcher.py index b960c5432..1a9e407e8 100644 --- a/tests/clients/test_configuration_watcher.py +++ b/tests/clients/test_configuration_watcher.py @@ -22,6 +22,7 @@ import grpc from dapr.clients.grpc._response import ( + CONFIG_OUTAGE_WARNING_EVERY_N_FAILURES, CONFIG_RECONNECT_INITIAL_BACKOFF_SECONDS, CONFIG_RECONNECT_MAX_BACKOFF_SECONDS, ConfigurationResponse, @@ -99,6 +100,7 @@ def test_retryable_classification(self): for code in ( grpc.StatusCode.INVALID_ARGUMENT, grpc.StatusCode.NOT_FOUND, + grpc.StatusCode.FAILED_PRECONDITION, grpc.StatusCode.PERMISSION_DENIED, grpc.StatusCode.UNAUTHENTICATED, grpc.StatusCode.UNIMPLEMENTED, @@ -573,6 +575,148 @@ def test_backoff_keeps_growing_when_streams_close_right_after_the_id(self): self.assertGreaterEqual(delay, base) self.assertLessEqual(delay, base * 1.2) + def test_catch_up_read_leaves_out_subscribe_only_metadata(self): + # The PostgreSQL store filters GetConfiguration rows by every metadata entry, so its + # subscribe-only notify channel must not be sent with the read after a reconnect. + server = self._fake_dapr_server + server.config_values = {'k': 'changed-while-down'} + server.config_stream_plans.extend( + [ + {'id': 'first', 'end': 'abort'}, + {'id': 'second', 'end': 'hold'}, + ] + ) + metadata = {'pgNotifyChannel': 'config', 'label': 'x'} + + self.client.subscribe_configuration( + store_name=STORE, keys=['k'], handler=self.handler, config_metadata=metadata + ) + + self.assertTrue(wait_until(lambda: len(self.updates) == 1), self.updates) + self.assertEqual(len(server.config_get_requests), 1) + self.assertEqual(dict(server.config_get_requests[0].metadata), {'label': 'x'}) + for request in server.config_subscribe_requests: + self.assertEqual(dict(request.metadata), metadata) + + def test_long_outage_warns_again_every_n_failures(self): + # A store removed from the sidecar fails every resubscribe with INVALID_ARGUMENT; that + # keeps being retried, and must keep showing up at WARNING level. + failing_attempts = 2 * CONFIG_OUTAGE_WARNING_EVERY_N_FAILURES + 1 + server = self._fake_dapr_server + server.config_stream_plans.append({'id': 'first', 'end': 'abort'}) + server.config_stream_plans.extend( + [{'reject': grpc.StatusCode.INVALID_ARGUMENT}] * (failing_attempts - 1) + ) + server.config_stream_plans.append({'id': 'second', 'updates': [{'k': 'v2'}], 'end': 'hold'}) + + with self.assertLogs('dapr.clients.grpc._response', level='DEBUG') as logs: + self.subscribe() + self.assertTrue(wait_until(lambda: len(self.updates) == 1), self.updates) + + failures = [r for r in logs.records if 'failed, reconnecting' in r.getMessage()] + self.assertEqual(len(failures), failing_attempts, logs.output) + warnings = [r.getMessage() for r in failures if r.levelname == 'WARNING'] + self.assertEqual(len(warnings), 3, warnings) + self.assertIn( + f'still failing after {CONFIG_OUTAGE_WARNING_EVERY_N_FAILURES} attempts', warnings[1] + ) + self.assertIn( + f'still failing after {2 * CONFIG_OUTAGE_WARNING_EVERY_N_FAILURES} attempts', + warnings[2], + ) + + def test_no_configuration_stores_on_first_subscribe_returns_none_quickly(self): + server = self._fake_dapr_server + server.config_stream_plans.extend( + [{'reject': grpc.StatusCode.FAILED_PRECONDITION}, {'id': 'unused', 'end': 'hold'}] + ) + started = time.monotonic() + + self.assertIsNone(self.subscribe()) + + self.assertLess(time.monotonic() - started, 2.0) + self.assertEqual(len(server.config_subscribe_requests), 1) + self.assertEqual(self.client._config_watchers, {}) + + def test_unsubscribe_before_the_reconnect_stream_sent_its_id_stops_locally(self): + server = self._fake_dapr_server + id_gate = threading.Event() + self.addCleanup(id_gate.set) + server.config_stream_plans.extend( + [ + {'id': 'first', 'end': 'abort'}, + {'id': 'second', 'end': 'hold', 'wait_before_id': id_gate}, + ] + ) + subscription_id = self.subscribe() + watcher = self.watcher(subscription_id) + # The reconnect stream is open, but the sidecar has not sent its id yet. + self.assertTrue( + wait_until( + lambda: len(server.config_subscribe_requests) == 2 and watcher._call is not None + ) + ) + self.assertIsNone(watcher.live_stream_id()) + + self.assertTrue(self.client.unsubscribe_configuration(STORE, subscription_id)) + + # The first stream's id is dead (the sidecar would answer ok=False) and the new one is + # not known yet, so nothing is sent. + self.assertEqual(server.config_unsubscribe_requests, []) + self.assertTrue(watcher.stopped) + self.assertNotIn((STORE, subscription_id), self.client._config_watchers) + # A stream wrapped by gRPC instrumentation cannot be cancelled, so the thread may still + # be waiting for the id; once it arrives the watcher exits without reconnecting. + id_gate.set() + thread = watcher._thread + assert thread is not None + thread.join(WAIT_TIMEOUT_SECONDS) + self.assertFalse(thread.is_alive()) + self.assertEqual(server.config_unsubscribe_requests, []) + self.assertEqual(len(server.config_subscribe_requests), 2) + + def test_watcher_that_exited_before_registration_is_not_tracked(self): + def watch_and_exit(watcher: ConfigurationWatcher, *args, **kwargs) -> str: + # The watcher delivered an id and gave up before the client could register it. + watcher.store_name = STORE + watcher.subscription_id = 'gone' + watcher._exited = True + return 'gone' + + with patch.object(ConfigurationWatcher, 'watch_configuration', watch_and_exit): + self.assertEqual(self.subscribe(), 'gone') + + self.assertEqual(self.client._config_watchers, {}) + + def test_each_outage_starts_with_a_warning(self): + # A stream that delivered its id ends the outage, whether it then fails or closes + # cleanly, so the next failure is the first of a new outage. + self._fake_dapr_server.config_stream_plans.extend( + [ + {'id': 'first', 'end': 'abort'}, + {'reject': grpc.StatusCode.UNAVAILABLE}, + {'id': 'second', 'end': 'abort'}, + {'reject': grpc.StatusCode.UNAVAILABLE}, + {'id': 'third', 'end': 'eof'}, + {'reject': grpc.StatusCode.UNAVAILABLE}, + {'id': 'fourth', 'updates': [{'k': 'v4'}], 'end': 'hold'}, + ] + ) + + with self.assertLogs('dapr.clients.grpc._response', level='DEBUG') as logs: + self.subscribe() + self.assertTrue(wait_until(lambda: len(self.updates) == 1), self.updates) + + failures = [r for r in logs.records if 'failed, reconnecting' in r.getMessage()] + self.assertEqual( + [r.levelname for r in failures], + ['WARNING', 'DEBUG', 'WARNING', 'DEBUG', 'WARNING'], + logs.output, + ) + self.assertFalse( + any('still failing' in r.getMessage() for r in failures if r.levelname == 'WARNING') + ) + if __name__ == '__main__': unittest.main() diff --git a/tests/clients/test_configuration_watcher_async.py b/tests/clients/test_configuration_watcher_async.py index cd3a79e16..dfad6df6e 100644 --- a/tests/clients/test_configuration_watcher_async.py +++ b/tests/clients/test_configuration_watcher_async.py @@ -24,7 +24,10 @@ from dapr.aio.clients.grpc._response import AsyncConfigurationWatcher from dapr.aio.clients.grpc.client import DaprGrpcClientAsync -from dapr.clients.grpc._response import ConfigurationResponse +from dapr.clients.grpc._response import ( + CONFIG_OUTAGE_WARNING_EVERY_N_FAILURES, + ConfigurationResponse, +) from dapr.conf import settings from .fake_dapr_server import FakeDaprSidecar @@ -435,6 +438,177 @@ async def async_handler(subscription_id: str, response: ConfigurationResponse) - self.assertTrue(await wait_until(lambda: len(received) == 1), received) self.assertEqual(received, [('first', 'v1')]) + async def test_catch_up_read_leaves_out_subscribe_only_metadata(self): + server = self._fake_dapr_server + server.config_values = {'k': 'changed-while-down'} + server.config_stream_plans.extend( + [ + {'id': 'first', 'end': 'abort'}, + {'id': 'second', 'end': 'hold'}, + ] + ) + metadata = {'pgNotifyChannel': 'config', 'label': 'x'} + + await self.client.subscribe_configuration( + store_name=STORE, keys=['k'], handler=self.handler, config_metadata=metadata + ) + + self.assertTrue(await wait_until(lambda: len(self.updates) == 1), self.updates) + self.assertEqual(len(server.config_get_requests), 1) + self.assertEqual(dict(server.config_get_requests[0].metadata), {'label': 'x'}) + for request in server.config_subscribe_requests: + self.assertEqual(dict(request.metadata), metadata) + + async def test_long_outage_warns_again_every_n_failures(self): + failing_attempts = 2 * CONFIG_OUTAGE_WARNING_EVERY_N_FAILURES + 1 + server = self._fake_dapr_server + server.config_stream_plans.append({'id': 'first', 'end': 'abort'}) + server.config_stream_plans.extend( + [{'reject': grpc.StatusCode.INVALID_ARGUMENT}] * (failing_attempts - 1) + ) + server.config_stream_plans.append({'id': 'second', 'updates': [{'k': 'v2'}], 'end': 'hold'}) + + with self.assertLogs('dapr.aio.clients.grpc._response', level='DEBUG') as logs: + await self.subscribe() + self.assertTrue(await wait_until(lambda: len(self.updates) == 1), self.updates) + + failures = [r for r in logs.records if 'failed, reconnecting' in r.getMessage()] + self.assertEqual(len(failures), failing_attempts, logs.output) + warnings = [r.getMessage() for r in failures if r.levelname == 'WARNING'] + self.assertEqual(len(warnings), 3, warnings) + self.assertIn( + f'still failing after {CONFIG_OUTAGE_WARNING_EVERY_N_FAILURES} attempts', warnings[1] + ) + self.assertIn( + f'still failing after {2 * CONFIG_OUTAGE_WARNING_EVERY_N_FAILURES} attempts', + warnings[2], + ) + + async def test_no_configuration_stores_on_first_subscribe_returns_none_quickly(self): + server = self._fake_dapr_server + server.config_stream_plans.extend( + [{'reject': grpc.StatusCode.FAILED_PRECONDITION}, {'id': 'unused', 'end': 'hold'}] + ) + started = time.monotonic() + + self.assertIsNone(await self.subscribe()) + + self.assertLess(time.monotonic() - started, 2.0) + self.assertEqual(len(server.config_subscribe_requests), 1) + self.assertEqual(self.client._config_watchers, {}) + + async def test_unsubscribe_before_the_reconnect_stream_sent_its_id_stops_locally(self): + server = self._fake_dapr_server + id_gate = threading.Event() + self.addCleanup(id_gate.set) + server.config_stream_plans.extend( + [ + {'id': 'first', 'end': 'abort'}, + {'id': 'second', 'end': 'hold', 'wait_before_id': id_gate}, + ] + ) + subscription_id = await self.subscribe() + watcher = self.watcher(subscription_id) + self.assertTrue( + await wait_until( + lambda: len(server.config_subscribe_requests) == 2 and watcher._call is not None + ) + ) + self.assertIsNone(watcher.live_stream_id()) + + self.assertTrue(await self.client.unsubscribe_configuration(STORE, subscription_id)) + + self.assertEqual(server.config_unsubscribe_requests, []) + task = watcher._task + assert task is not None + self.assertTrue(task.done()) + self.assertNotIn((STORE, subscription_id), self.client._config_watchers) + + async def test_watcher_that_exited_before_registration_is_not_tracked(self): + async def watch_and_exit(watcher: AsyncConfigurationWatcher, *args, **kwargs) -> str: + watcher.store_name = STORE + watcher.subscription_id = 'gone' + watcher._exited = True + return 'gone' + + with patch.object(AsyncConfigurationWatcher, 'watch_configuration', watch_and_exit): + self.assertEqual(await self.subscribe(), 'gone') + + self.assertEqual(self.client._config_watchers, {}) + + async def test_blocking_sync_handler_does_not_stall_the_event_loop(self): + self._fake_dapr_server.config_stream_plans.append( + {'id': 'first', 'updates': [{'k': 'v1'}, {'k': 'v2'}], 'end': 'hold'} + ) + handler_started = threading.Event() + handler_finished = threading.Event() + release = threading.Event() + self.addCleanup(release.set) + received: List[str] = [] + + def blocking_handler(subscription_id: str, response: ConfigurationResponse) -> None: + handler_started.set() + release.wait(2.0) + received.append(response.items['k'].value) + handler_finished.set() + + ticks = 0 + + async def ticker() -> None: + nonlocal ticks + while True: + ticks += 1 + await asyncio.sleep(0.01) + + ticker_task = asyncio.create_task(ticker()) + try: + await self.client.subscribe_configuration( + store_name=STORE, keys=['k'], handler=blocking_handler + ) + self.assertTrue(await wait_until(handler_started.is_set)) + ticks_before = ticks + await asyncio.sleep(0.2) + ticks_while_blocked = ticks - ticks_before + handler_was_blocked = not handler_finished.is_set() + finally: + release.set() + ticker_task.cancel() + + self.assertTrue(handler_was_blocked) + self.assertGreater(ticks_while_blocked, 0) + # Calls stay sequential and in order. + self.assertTrue(await wait_until(lambda: len(received) == 2), received) + self.assertEqual(received, ['v1', 'v2']) + + async def test_each_outage_starts_with_a_warning(self): + # A stream that delivered its id ends the outage, whether it then fails or closes + # cleanly, so the next failure is the first of a new outage. + self._fake_dapr_server.config_stream_plans.extend( + [ + {'id': 'first', 'end': 'abort'}, + {'reject': grpc.StatusCode.UNAVAILABLE}, + {'id': 'second', 'end': 'abort'}, + {'reject': grpc.StatusCode.UNAVAILABLE}, + {'id': 'third', 'end': 'eof'}, + {'reject': grpc.StatusCode.UNAVAILABLE}, + {'id': 'fourth', 'updates': [{'k': 'v4'}], 'end': 'hold'}, + ] + ) + + with self.assertLogs('dapr.aio.clients.grpc._response', level='DEBUG') as logs: + await self.subscribe() + self.assertTrue(await wait_until(lambda: len(self.updates) == 1), self.updates) + + failures = [r for r in logs.records if 'failed, reconnecting' in r.getMessage()] + self.assertEqual( + [r.levelname for r in failures], + ['WARNING', 'DEBUG', 'WARNING', 'DEBUG', 'WARNING'], + logs.output, + ) + self.assertFalse( + any('still failing' in r.getMessage() for r in failures if r.levelname == 'WARNING') + ) + if __name__ == '__main__': unittest.main() From edde66d0b33967382fa25c13da1ef9606308a4c2 Mon Sep 17 00:00:00 2001 From: Casper Nielsen Date: Tue, 29 Sep 2026 08:35:17 +0200 Subject: [PATCH 5/6] docs(clients): document handler threads and the instrumented-stream unsubscribe gap - Async client: plain-function handlers may run on a different worker thread each call, and one still running at stop/close isn't waited for, so keep them short. - Sync client: with gRPC client instrumentation (e.g. OpenTelemetry) the stream can't be cancelled, so an unsubscribe while a reconnect is waiting for the first response leaves the watcher thread blocked until it arrives. Co-Authored-By: Claude Opus 5.5 (1M context) Signed-off-by: Casper Nielsen --- dapr/aio/clients/grpc/_response.py | 4 +++- dapr/aio/clients/grpc/client.py | 6 +++++- dapr/clients/grpc/_response.py | 8 +++++++- dapr/clients/grpc/client.py | 7 +++++++ 4 files changed, 22 insertions(+), 3 deletions(-) diff --git a/dapr/aio/clients/grpc/_response.py b/dapr/aio/clients/grpc/_response.py index 10b617198..f167c9d0c 100644 --- a/dapr/aio/clients/grpc/_response.py +++ b/dapr/aio/clients/grpc/_response.py @@ -144,7 +144,9 @@ class AsyncConfigurationWatcher: ``async def`` handlers are awaited on the event loop; plain functions run in a worker thread (asyncio.to_thread) so they cannot block the loop. Either way, one handler call - finishes before the next starts, in the order the updates arrived. + finishes before the next starts, in the order the updates arrived. Plain-function calls + may each run on a different worker thread (don't rely on thread-locals), and one that is + still running when the watcher stops is not waited for, so keep them short. Failed reconnect attempts are logged like ConfigurationWatcher does (see CONFIG_OUTAGE_WARNING_EVERY_N_FAILURES). diff --git a/dapr/aio/clients/grpc/client.py b/dapr/aio/clients/grpc/client.py index b942f7a68..99d7e68e3 100644 --- a/dapr/aio/clients/grpc/client.py +++ b/dapr/aio/clients/grpc/client.py @@ -1251,7 +1251,11 @@ async def subscribe_configuration( The handler may be a plain function or an ``async def`` function. Async handlers run on the event loop. Plain functions run in a worker thread (asyncio.to_thread), so a slow handler does not block the event loop. Handler calls run one at a time, in the - order the updates arrive. + order the updates arrive. Each call of a plain function may run on a different worker + thread, so don't rely on thread-local state or thread affinity across calls. Keep + plain handlers short: one still running when the subscription is stopped or the + client is closed keeps running in its thread, and its result or exception is not + observed. Returns: id (str): subscription id, which can be used to unsubscribe later diff --git a/dapr/clients/grpc/_response.py b/dapr/clients/grpc/_response.py index ec2083345..bf1aa767f 100644 --- a/dapr/clients/grpc/_response.py +++ b/dapr/clients/grpc/_response.py @@ -910,7 +910,13 @@ def request_stop(self) -> None: self._stop_event.set() def stop(self) -> None: - """Stops reconnecting, cancels the current stream and waits briefly for the thread.""" + """Stops reconnecting, cancels the current stream and waits briefly for the thread. + + If gRPC client instrumentation (for example OpenTelemetry) wrapped the stream, it has no + cancel(); a stream still waiting for the sidecar's first response then keeps the thread + blocked until that response arrives, after which the watcher exits without subscribing + again. Closing the client's channel ends such streams immediately. + """ self.request_stop() self.cancel_stream() self.join() diff --git a/dapr/clients/grpc/client.py b/dapr/clients/grpc/client.py index b8b3cb088..9d3b464ae 100644 --- a/dapr/clients/grpc/client.py +++ b/dapr/clients/grpc/client.py @@ -1371,6 +1371,13 @@ def unsubscribe_configuration(self, store_name: str, id: str) -> bool: UnsubscribeConfiguration sent through another client only ends the current stream and this client subscribes again. + Known limitation: when gRPC client instrumentation (for example OpenTelemetry, which + the workflow extension enables on import) wraps the stream, it can't be cancelled from + the client. If this is called while a reconnect is still waiting for the sidecar's + first response, the watcher thread stays blocked until that response arrives and then + exits without subscribing again. close() is not affected, because closing the channel + ends the stream. + Args: store_name (str): the state store name to unsubscribe from id (str): the subscription id to unsubscribe From be3491f0d94bbfb9bef9ce7e567d3ad8a359568f Mon Sep 17 00:00:00 2001 From: Casper Nielsen Date: Tue, 13 Oct 2026 15:58:00 +0200 Subject: [PATCH 6/6] fix: stop the async configuration watcher when subscribe is cancelled If the caller cancels subscribe_configuration while it waits for the first subscription id, the watcher task was left running and unreachable. Co-Authored-By: Claude Opus 5.5 Signed-off-by: Casper Nielsen --- dapr/aio/clients/grpc/_response.py | 4 ++ .../test_configuration_watcher_async.py | 47 +++++++++++++++++++ 2 files changed, 51 insertions(+) diff --git a/dapr/aio/clients/grpc/_response.py b/dapr/aio/clients/grpc/_response.py index f167c9d0c..428220643 100644 --- a/dapr/aio/clients/grpc/_response.py +++ b/dapr/aio/clients/grpc/_response.py @@ -198,6 +198,10 @@ async def watch_configuration( self.keys, self.store_name, ) + except asyncio.CancelledError: + # The caller never gets the id, so nothing could unsubscribe or close this watcher. + await self.stop() + raise if not self.subscription_id: await self.stop() return None diff --git a/tests/clients/test_configuration_watcher_async.py b/tests/clients/test_configuration_watcher_async.py index dfad6df6e..88b8b96aa 100644 --- a/tests/clients/test_configuration_watcher_async.py +++ b/tests/clients/test_configuration_watcher_async.py @@ -610,5 +610,52 @@ async def test_each_outage_starts_with_a_warning(self): ) +class SilentSubscribeCall: + """A SubscribeConfigurationAlpha1 call that never sends a message until cancelled.""" + + def __init__(self) -> None: + self.cancelled = asyncio.Event() + + def cancel(self) -> bool: + self.cancelled.set() + return True + + def __aiter__(self) -> 'SilentSubscribeCall': + return self + + async def __anext__(self) -> None: + await self.cancelled.wait() + raise StopAsyncIteration + + +class SilentSubscribeStub: + def __init__(self) -> None: + self.calls: List[SilentSubscribeCall] = [] + + def SubscribeConfigurationAlpha1(self, request: object) -> SilentSubscribeCall: + call = SilentSubscribeCall() + self.calls.append(call) + return call + + +class AsyncConfigurationWatcherCancellationTests(unittest.IsolatedAsyncioTestCase): + async def test_cancelled_watch_configuration_stops_the_watcher_task(self): + stub = SilentSubscribeStub() + exited: List[AsyncConfigurationWatcher] = [] + watcher = AsyncConfigurationWatcher(on_exit=exited.append) + watch = watcher.watch_configuration(stub, STORE, ['k'], lambda *_: None) + + with self.assertRaises(asyncio.TimeoutError): + await asyncio.wait_for(watch, timeout=0.2) + + task = watcher._task + assert task is not None + self.assertTrue(task.done()) + self.assertTrue(watcher.stopped) + self.assertEqual(exited, [watcher]) + self.assertEqual(len(stub.calls), 1) + self.assertTrue(stub.calls[0].cancelled.is_set()) + + if __name__ == '__main__': unittest.main()