fix(ext-workflow): honor DAPR_GRPC_MAX_INBOUND_MESSAGE_SIZE_BYTES on workflow channels

dapr/ext/workflow/aio/dapr_workflow_client.py

@@ -28,7 +28,7 @@
 from dapr.ext.workflow._durabletask import client
 from dapr.ext.workflow._durabletask.aio import client as aioclient
 from dapr.ext.workflow.logger import Logger, LoggerOptions
-from dapr.ext.workflow.util import getAddress
+from dapr.ext.workflow.util import get_grpc_channel_options, getAddress
 from dapr.ext.workflow.workflow_context import Workflow
 from dapr.ext.workflow.workflow_state import WorkflowState
 
@@ -49,7 +49,22 @@ def __init__(
         host: Optional[str] = None,
         port: Optional[str] = None,
         logger_options: Optional[LoggerOptions] = None,
+        max_grpc_message_length: Optional[int] = None,
     ):
+        """Initializes the async workflow client.
+
+        Args:
+            host: Dapr sidecar gRPC hostname. Defaults to
+                ``settings.DAPR_RUNTIME_HOST`` (or ``DAPR_GRPC_ENDPOINT`` when set).
+            port: Dapr sidecar gRPC port. Defaults to ``settings.DAPR_GRPC_PORT``.
+            logger_options: Configuration for the client's internal logger.
+            max_grpc_message_length: Maximum gRPC message size in bytes for the
+                workflow channel, applied symmetrically to both send and receive
+                directions. Precedence: this kwarg (if non-zero), then the
+                ``DAPR_GRPC_MAX_INBOUND_MESSAGE_SIZE_BYTES`` env var (if non-zero),
+                then the gRPC default (4 MiB). ``0`` in either source means
+                "no opinion" and falls through to the next source.
+        """
         address = getAddress(host, port)
 
         try:
@@ -63,13 +78,15 @@ def __init__(
         if settings.DAPR_API_TOKEN:
             metadata = ((DAPR_API_TOKEN_HEADER, settings.DAPR_API_TOKEN),)
         options = self._logger.get_options()
+        channel_options = get_grpc_channel_options(max_grpc_message_length)
         self.__obj = aioclient.AsyncTaskHubGrpcClient(
             host_address=uri.endpoint,
             metadata=metadata,
             secure_channel=uri.tls,
             log_handler=options.log_handler,
             log_formatter=options.log_formatter,
             interceptors=[DaprClientTimeoutInterceptorAsync()],
+            channel_options=channel_options,
         )
 
     async def schedule_new_workflow(

dapr/ext/workflow/dapr_workflow_client.py

@@ -27,7 +27,7 @@
 from dapr.conf.helpers import GrpcEndpoint
 from dapr.ext.workflow._durabletask import client
 from dapr.ext.workflow.logger import Logger, LoggerOptions
-from dapr.ext.workflow.util import getAddress
+from dapr.ext.workflow.util import get_grpc_channel_options, getAddress
 from dapr.ext.workflow.workflow_context import Workflow
 from dapr.ext.workflow.workflow_state import WorkflowState
 
@@ -51,7 +51,22 @@ def __init__(
         host: Optional[str] = None,
         port: Optional[str] = None,
         logger_options: Optional[LoggerOptions] = None,
+        max_grpc_message_length: Optional[int] = None,
     ):
+        """Initializes the sync workflow client.
+
+        Args:
+            host: Dapr sidecar gRPC hostname. Defaults to
+                ``settings.DAPR_RUNTIME_HOST`` (or ``DAPR_GRPC_ENDPOINT`` when set).
+            port: Dapr sidecar gRPC port. Defaults to ``settings.DAPR_GRPC_PORT``.
+            logger_options: Configuration for the client's internal logger.
+            max_grpc_message_length: Maximum gRPC message size in bytes for the
+                workflow channel, applied symmetrically to both send and receive
+                directions. Precedence: this kwarg (if non-zero), then the
+                ``DAPR_GRPC_MAX_INBOUND_MESSAGE_SIZE_BYTES`` env var (if non-zero),
+                then the gRPC default (4 MiB). ``0`` in either source means
+                "no opinion" and falls through to the next source.
+        """
         address = getAddress(host, port)
 
         try:
@@ -65,13 +80,15 @@ def __init__(
         if settings.DAPR_API_TOKEN:
             metadata = ((DAPR_API_TOKEN_HEADER, settings.DAPR_API_TOKEN),)
         options = self._logger.get_options()
+        channel_options = get_grpc_channel_options(max_grpc_message_length)
         self.__obj = client.TaskHubGrpcClient(
             host_address=uri.endpoint,
             metadata=metadata,
             secure_channel=uri.tls,
             log_handler=options.log_handler,
             log_formatter=options.log_formatter,
             interceptors=[DaprClientTimeoutInterceptor()],
+            channel_options=channel_options,
         )
 
     def schedule_new_workflow(

dapr/ext/workflow/util.py

@@ -13,11 +13,46 @@
 limitations under the License.
 """
 
-from typing import Optional
+from typing import Any, Optional, Sequence
 
 from dapr.conf import settings
 
 
+def get_grpc_channel_options(
+    max_grpc_message_length: Optional[int] = None,
+) -> Optional[Sequence[tuple[str, Any]]]:
+    """Resolves gRPC channel options for the workflow message-size limit.
+
+    Precedence: explicit ``max_grpc_message_length`` kwarg (if non-zero),
+    else ``settings.DAPR_GRPC_MAX_INBOUND_MESSAGE_SIZE_BYTES`` (if non-zero),
+    else the gRPC default. A ``0`` in either source is interpreted as
+    "no opinion / use default" and falls through to the next source — this
+    matches ``global_settings.DAPR_GRPC_MAX_INBOUND_MESSAGE_SIZE_BYTES = 0``
+    being the documented "unset" sentinel.
+
+    Sets BOTH send and receive limits symmetrically because workflow activity
+    payloads cross the channel in both directions. Returns ``None`` when no
+    explicit limit is configured so callers leave the channel unconfigured and
+    the gRPC default applies.
+
+    Args:
+        max_grpc_message_length: Explicit max gRPC message size in bytes.
+            ``0`` or ``None`` means "no opinion" and falls through to the
+            env var.
+
+    Returns:
+        A sequence of ``(option, value)`` tuples setting both send and receive
+        limits, or ``None`` when no limit is configured.
+    """
+    size = max_grpc_message_length or settings.DAPR_GRPC_MAX_INBOUND_MESSAGE_SIZE_BYTES
+    if not size:
+        return None
+    return [
+        ('grpc.max_send_message_length', size),
+        ('grpc.max_receive_message_length', size),
+    ]
+
+
 def getAddress(host: Optional[str] = None, port: Optional[str] = None) -> str:
     if not host and not port:
         address = settings.DAPR_GRPC_ENDPOINT or (

dapr/ext/workflow/workflow_runtime.py

@@ -29,7 +29,7 @@
 from dapr.ext.workflow._durabletask.internal.shared import is_async_callable as _is_async_callable
 from dapr.ext.workflow.dapr_workflow_context import DaprWorkflowContext
 from dapr.ext.workflow.logger import Logger, LoggerOptions
-from dapr.ext.workflow.util import getAddress
+from dapr.ext.workflow.util import get_grpc_channel_options, getAddress
 from dapr.ext.workflow.workflow_activity_context import Activity, WorkflowActivityContext
 from dapr.ext.workflow.workflow_context import Workflow
 
@@ -118,7 +118,35 @@ def __init__(
         maximum_concurrent_orchestration_work_items: Optional[int] = None,
         maximum_thread_pool_workers: Optional[int] = None,
         worker_ready_timeout: Optional[float] = None,
+        max_grpc_message_length: Optional[int] = None,
     ):
+        """Initializes the workflow runtime.
+
+        Args:
+            host: Dapr sidecar gRPC hostname. Defaults to
+                ``settings.DAPR_RUNTIME_HOST`` (or ``DAPR_GRPC_ENDPOINT`` when set).
+            port: Dapr sidecar gRPC port. Defaults to ``settings.DAPR_GRPC_PORT``.
+            logger_options: Configuration for the runtime's internal logger.
+            interceptors: Additional client gRPC interceptors. The built-in
+                ``DaprClientTimeoutInterceptor`` is always appended after these.
+            maximum_concurrent_activity_work_items: Maximum number of activity
+                work items the worker dispatches concurrently. ``None`` lets the
+                durabletask worker pick a default.
+            maximum_concurrent_orchestration_work_items: Maximum number of
+                orchestration work items the worker dispatches concurrently.
+                ``None`` lets the durabletask worker pick a default.
+            maximum_thread_pool_workers: Size of the worker's thread pool for
+                executing sync activities. ``None`` lets the durabletask worker
+                pick a default.
+            worker_ready_timeout: Seconds to wait in :meth:`start` for the
+                worker's gRPC stream to be ready. Defaults to 30s.
+            max_grpc_message_length: Maximum gRPC message size in bytes for the
+                workflow channel, applied symmetrically to both send and receive
+                directions. Precedence: this kwarg (if non-zero), then the
+                ``DAPR_GRPC_MAX_INBOUND_MESSAGE_SIZE_BYTES`` env var (if non-zero),
+                then the gRPC default (4 MiB). ``0`` in either source means
+                "no opinion" and falls through to the next source.
+        """
         self._logger = Logger('WorkflowRuntime', logger_options)
         self._worker_ready_timeout = 30.0 if worker_ready_timeout is None else worker_ready_timeout
 
@@ -137,13 +165,15 @@ def __init__(
         if interceptors:
             all_interceptors.extend(interceptors)
         all_interceptors.append(DaprClientTimeoutInterceptor())
+        channel_options = get_grpc_channel_options(max_grpc_message_length)
         self.__worker = worker.TaskHubGrpcWorker(
             host_address=uri.endpoint,
             metadata=metadata,
             secure_channel=uri.tls,
             log_handler=options.log_handler,
             log_formatter=options.log_formatter,
             interceptors=all_interceptors,
+            channel_options=channel_options,
             concurrency_options=worker.ConcurrencyOptions(
                 maximum_concurrent_activity_work_items=maximum_concurrent_activity_work_items,
                 maximum_concurrent_orchestration_work_items=maximum_concurrent_orchestration_work_items,

tests/ext/workflow/test_workflow_client.py

@@ -20,6 +20,7 @@
 
 from grpc import RpcError
 
+from dapr.conf import settings
 from dapr.ext.workflow._durabletask import client
 from dapr.ext.workflow.dapr_workflow_client import DaprWorkflowClient
 from dapr.ext.workflow.dapr_workflow_context import DaprWorkflowContext
@@ -127,6 +128,62 @@ def test_timeout_interceptor_is_passed_to_client(self):
             self.assertIsInstance(interceptors[0], DaprClientTimeoutInterceptor)
 
 
+class WorkflowClientChannelOptionsTest(unittest.TestCase):
+    @mock.patch.object(settings, 'DAPR_GRPC_MAX_INBOUND_MESSAGE_SIZE_BYTES', 0)
+    def test_explicit_kwarg_sets_symmetric_channel_options(self):
+        with mock.patch(
+            'dapr.ext.workflow._durabletask.client.TaskHubGrpcClient'
+        ) as mock_client_cls:
+            DaprWorkflowClient(max_grpc_message_length=8 * 1024 * 1024)
+            channel_options = mock_client_cls.call_args[1]['channel_options']
+            self.assertEqual(
+                [
+                    ('grpc.max_send_message_length', 8 * 1024 * 1024),
+                    ('grpc.max_receive_message_length', 8 * 1024 * 1024),
+                ],
+                channel_options,
+            )
+
+    @mock.patch.object(settings, 'DAPR_GRPC_MAX_INBOUND_MESSAGE_SIZE_BYTES', 16 * 1024 * 1024)
+    def test_env_var_sets_symmetric_channel_options(self):
+        with mock.patch(
+            'dapr.ext.workflow._durabletask.client.TaskHubGrpcClient'
+        ) as mock_client_cls:
+            DaprWorkflowClient()
+            channel_options = mock_client_cls.call_args[1]['channel_options']
+            self.assertEqual(
+                [
+                    ('grpc.max_send_message_length', 16 * 1024 * 1024),
+                    ('grpc.max_receive_message_length', 16 * 1024 * 1024),
+                ],
+                channel_options,
+            )
+
+    @mock.patch.object(settings, 'DAPR_GRPC_MAX_INBOUND_MESSAGE_SIZE_BYTES', 16 * 1024 * 1024)
+    def test_kwarg_takes_precedence_over_env_var(self):
+        with mock.patch(
+            'dapr.ext.workflow._durabletask.client.TaskHubGrpcClient'
+        ) as mock_client_cls:
+            DaprWorkflowClient(max_grpc_message_length=8 * 1024 * 1024)
+            channel_options = mock_client_cls.call_args[1]['channel_options']
+            self.assertEqual(
+                [
+                    ('grpc.max_send_message_length', 8 * 1024 * 1024),
+                    ('grpc.max_receive_message_length', 8 * 1024 * 1024),
+                ],
+                channel_options,
+            )
+
+    @mock.patch.object(settings, 'DAPR_GRPC_MAX_INBOUND_MESSAGE_SIZE_BYTES', 0)
+    def test_neither_set_passes_none(self):
+        with mock.patch(
+            'dapr.ext.workflow._durabletask.client.TaskHubGrpcClient'
+        ) as mock_client_cls:
+            DaprWorkflowClient()
+            channel_options = mock_client_cls.call_args[1]['channel_options']
+            self.assertIsNone(channel_options)
+
+
 class WorkflowClientTest(unittest.TestCase):
     def mock_client_wf(ctx: DaprWorkflowContext, input):
         print(f'{input}')

tests/ext/workflow/test_workflow_client_aio.py

@@ -20,6 +20,7 @@
 
 from grpc.aio import AioRpcError
 
+from dapr.conf import settings
 from dapr.ext.workflow._durabletask import client
 from dapr.ext.workflow.aio import DaprWorkflowClient
 from dapr.ext.workflow.dapr_workflow_context import DaprWorkflowContext
@@ -128,6 +129,62 @@ async def test_timeout_interceptor_is_passed_to_client(self):
             self.assertIsInstance(interceptors[0], DaprClientTimeoutInterceptorAsync)
 
 
+class WorkflowClientAioChannelOptionsTest(unittest.TestCase):
+    @mock.patch.object(settings, 'DAPR_GRPC_MAX_INBOUND_MESSAGE_SIZE_BYTES', 0)
+    def test_explicit_kwarg_sets_symmetric_channel_options(self):
+        with mock.patch(
+            'dapr.ext.workflow._durabletask.aio.client.AsyncTaskHubGrpcClient'
+        ) as mock_client_cls:
+            DaprWorkflowClient(max_grpc_message_length=8 * 1024 * 1024)
+            channel_options = mock_client_cls.call_args[1]['channel_options']
+            self.assertEqual(
+                [
+                    ('grpc.max_send_message_length', 8 * 1024 * 1024),
+                    ('grpc.max_receive_message_length', 8 * 1024 * 1024),
+                ],
+                channel_options,
+            )
+
+    @mock.patch.object(settings, 'DAPR_GRPC_MAX_INBOUND_MESSAGE_SIZE_BYTES', 16 * 1024 * 1024)
+    def test_env_var_sets_symmetric_channel_options(self):
+        with mock.patch(
+            'dapr.ext.workflow._durabletask.aio.client.AsyncTaskHubGrpcClient'
+        ) as mock_client_cls:
+            DaprWorkflowClient()
+            channel_options = mock_client_cls.call_args[1]['channel_options']
+            self.assertEqual(
+                [
+                    ('grpc.max_send_message_length', 16 * 1024 * 1024),
+                    ('grpc.max_receive_message_length', 16 * 1024 * 1024),
+                ],
+                channel_options,
+            )
+
+    @mock.patch.object(settings, 'DAPR_GRPC_MAX_INBOUND_MESSAGE_SIZE_BYTES', 16 * 1024 * 1024)
+    def test_kwarg_takes_precedence_over_env_var(self):
+        with mock.patch(
+            'dapr.ext.workflow._durabletask.aio.client.AsyncTaskHubGrpcClient'
+        ) as mock_client_cls:
+            DaprWorkflowClient(max_grpc_message_length=8 * 1024 * 1024)
+            channel_options = mock_client_cls.call_args[1]['channel_options']
+            self.assertEqual(
+                [
+                    ('grpc.max_send_message_length', 8 * 1024 * 1024),
+                    ('grpc.max_receive_message_length', 8 * 1024 * 1024),
+                ],
+                channel_options,
+            )
+
+    @mock.patch.object(settings, 'DAPR_GRPC_MAX_INBOUND_MESSAGE_SIZE_BYTES', 0)
+    def test_neither_set_passes_none(self):
+        with mock.patch(
+            'dapr.ext.workflow._durabletask.aio.client.AsyncTaskHubGrpcClient'
+        ) as mock_client_cls:
+            DaprWorkflowClient()
+            channel_options = mock_client_cls.call_args[1]['channel_options']
+            self.assertIsNone(channel_options)
+
+
 class WorkflowClientAioTest(unittest.IsolatedAsyncioTestCase):
     def mock_client_wf(ctx: DaprWorkflowContext, input):
         print(f'{input}')