Silence gRPC error spam on EAGAIN

Signed-off-by: Sergio Herrera <627709+seherv@users.noreply.github.com>

Author: seherv

ext/dapr-ext-workflow/AGENTS.md

@@ -122,6 +122,8 @@ Workflow (orchestrator) functions must remain generators (`def` with `yield`). T
 
 **Concurrency sizing and load characterization.** See `docs/concurrency.md` for sizing recommendations (`maximum_concurrent_activity_work_items`, `maximum_thread_pool_workers`) and an async-vs-sync decision tree. The `benchmarks/` directory ships `bench_async_activities.py`; re-run it locally before claiming a perf regression. The generated `RESULTS.md` is gitignored because numbers are machine-specific; see `docs/concurrency.md` for the regen command.
 
+**grpc.aio poller log noise.** The async client can emit benign `BlockingIOError: [Errno 11]` ERROR lines from `grpc.aio`'s `PollerCompletionQueue` under load. It is harmless and retried. `get_grpc_aio_channel` installs an internal `asyncio`-logger filter (`_silence_grpc_aio_poller_noise`) that drops only those records, so the SDK suppresses it automatically with no user action.
+
 
 ### DaprWorkflowClient (`dapr_workflow_client.py`)
 

ext/dapr-ext-workflow/dapr/ext/workflow/_durabletask/aio/internal/shared.py

@@ -9,6 +9,7 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+import logging
 from typing import Optional, Sequence, Union
 
 import grpc
@@ -27,6 +28,30 @@
     grpc_aio.StreamStreamClientInterceptor,
 ]
 
+_POLLER_NOISE_MARKER = 'PollerCompletionQueue._handle_events'
+
+
+class _GrpcAioPollerNoiseFilter(logging.Filter):
+    """Drops the harmless grpc.aio poller BlockingIOError (EAGAIN) records.
+
+    The poller does a non-blocking read on its wake-up fd and can get EAGAIN, which
+    asyncio logs at ERROR even though the read is retried and nothing is lost.
+    """
+
+    def filter(self, record: logging.LogRecord) -> bool:
+        exc = record.exc_info[1] if record.exc_info else None
+        is_poller_noise = isinstance(exc, BlockingIOError) and (
+            _POLLER_NOISE_MARKER in record.getMessage()
+        )
+        return not is_poller_noise
+
+
+def _silence_grpc_aio_poller_noise() -> None:
+    """Install the poller-noise filter on the asyncio logger if not already present."""
+    asyncio_logger = logging.getLogger('asyncio')
+    if not any(isinstance(f, _GrpcAioPollerNoiseFilter) for f in asyncio_logger.filters):
+        asyncio_logger.addFilter(_GrpcAioPollerNoiseFilter())
+
 
 def get_grpc_aio_channel(
     host_address: Optional[str],
@@ -42,6 +67,8 @@ def get_grpc_aio_channel(
         interceptors: Optional sequence of client interceptors to apply to the channel.
         options: Optional sequence of gRPC channel options as (key, value) tuples. Keys defined in https://grpc.github.io/grpc/core/group__grpc__arg__keys.html
     """
+    _silence_grpc_aio_poller_noise()
+
     if host_address is None:
         host_address = get_default_host_address()
 

ext/dapr-ext-workflow/tests/durabletask/test_grpc_aio_log_filter.py

@@ -0,0 +1,71 @@
+# 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 logging
+import sys
+from unittest.mock import patch
+
+from dapr.ext.workflow._durabletask.aio.internal import shared
+
+HOST_ADDRESS = 'localhost:50051'
+
+
+def _record(msg: str, exc: BaseException | None) -> logging.LogRecord:
+    exc_info = None
+    if exc is not None:
+        try:
+            raise exc
+        except BaseException:
+            exc_info = sys.exc_info()
+    return logging.LogRecord('asyncio', logging.ERROR, __file__, 1, msg, (), exc_info)
+
+
+def test_filter_drops_poller_eagain_record():
+    record = _record(
+        'Exception in callback PollerCompletionQueue._handle_events()',
+        BlockingIOError(11, 'Resource temporarily unavailable'),
+    )
+    assert shared._GrpcAioPollerNoiseFilter().filter(record) is False
+
+
+def test_filter_keeps_record_without_exception():
+    assert shared._GrpcAioPollerNoiseFilter().filter(_record('some other error', None)) is True
+
+
+def test_filter_keeps_blockingioerror_without_marker():
+    record = _record('unrelated message', BlockingIOError(11, 'nope'))
+    assert shared._GrpcAioPollerNoiseFilter().filter(record) is True
+
+
+def test_get_grpc_aio_channel_installs_filter_on_asyncio_logger():
+    asyncio_logger = logging.getLogger('asyncio')
+    for existing in [
+        f for f in asyncio_logger.filters if isinstance(f, shared._GrpcAioPollerNoiseFilter)
+    ]:
+        asyncio_logger.removeFilter(existing)
+
+    with patch('dapr.ext.workflow._durabletask.aio.internal.shared.grpc_aio.insecure_channel'):
+        shared.get_grpc_aio_channel(HOST_ADDRESS, False)
+
+    installed = [
+        f for f in asyncio_logger.filters if isinstance(f, shared._GrpcAioPollerNoiseFilter)
+    ]
+    assert len(installed) == 1  # installed once, not duplicated
+
+
+def test_install_is_idempotent():
+    asyncio_logger = logging.getLogger('asyncio')
+    shared._silence_grpc_aio_poller_noise()
+    shared._silence_grpc_aio_poller_noise()
+    installed = [
+        f for f in asyncio_logger.filters if isinstance(f, shared._GrpcAioPollerNoiseFilter)
+    ]
+    assert len(installed) == 1