diff --git a/dapr/ext/workflow/aio/dapr_workflow_client.py b/dapr/ext/workflow/aio/dapr_workflow_client.py index 5ce1e14ef..c8001db98 100644 --- a/dapr/ext/workflow/aio/dapr_workflow_client.py +++ b/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,6 +78,7 @@ 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, @@ -70,6 +86,7 @@ def __init__( log_handler=options.log_handler, log_formatter=options.log_formatter, interceptors=[DaprClientTimeoutInterceptorAsync()], + channel_options=channel_options, ) async def schedule_new_workflow( diff --git a/dapr/ext/workflow/dapr_workflow_client.py b/dapr/ext/workflow/dapr_workflow_client.py index a9d70c711..c72adef12 100644 --- a/dapr/ext/workflow/dapr_workflow_client.py +++ b/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,6 +80,7 @@ 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, @@ -72,6 +88,7 @@ def __init__( log_handler=options.log_handler, log_formatter=options.log_formatter, interceptors=[DaprClientTimeoutInterceptor()], + channel_options=channel_options, ) def schedule_new_workflow( diff --git a/dapr/ext/workflow/util.py b/dapr/ext/workflow/util.py index 3199e2558..47c7b33a2 100644 --- a/dapr/ext/workflow/util.py +++ b/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 ( diff --git a/dapr/ext/workflow/workflow_runtime.py b/dapr/ext/workflow/workflow_runtime.py index 4f70f953b..8d4e86505 100644 --- a/dapr/ext/workflow/workflow_runtime.py +++ b/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,6 +165,7 @@ 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, @@ -144,6 +173,7 @@ def __init__( 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, diff --git a/tests/ext/workflow/test_workflow_client.py b/tests/ext/workflow/test_workflow_client.py index 0267fca3b..fa2796a85 100644 --- a/tests/ext/workflow/test_workflow_client.py +++ b/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}') diff --git a/tests/ext/workflow/test_workflow_client_aio.py b/tests/ext/workflow/test_workflow_client_aio.py index 27897976f..63a653c93 100644 --- a/tests/ext/workflow/test_workflow_client_aio.py +++ b/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}') diff --git a/tests/ext/workflow/test_workflow_runtime.py b/tests/ext/workflow/test_workflow_runtime.py index b3558e8f8..3d333fd77 100644 --- a/tests/ext/workflow/test_workflow_runtime.py +++ b/tests/ext/workflow/test_workflow_runtime.py @@ -20,6 +20,7 @@ import grpc from pydantic import BaseModel, ValidationError +from dapr.conf import settings from dapr.ext.workflow.dapr_workflow_context import DaprWorkflowContext from dapr.ext.workflow.workflow_activity_context import WorkflowActivityContext from dapr.ext.workflow.workflow_runtime import WorkflowRuntime, alternate_name @@ -830,3 +831,63 @@ def my_act(ctx, order: Optional[Order]): wrapper = self.fake_registry._activity_fns['optional_no_default_act'] self.assertIsNone(wrapper(mock.MagicMock(), None)) + + +class WorkflowRuntimeChannelOptionsTest(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.worker.TaskHubGrpcWorker' + ) as mock_worker_cls: + WorkflowRuntime(max_grpc_message_length=8 * 1024 * 1024) + channel_options = mock_worker_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.worker.TaskHubGrpcWorker' + ) as mock_worker_cls: + WorkflowRuntime() + channel_options = mock_worker_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.worker.TaskHubGrpcWorker' + ) as mock_worker_cls: + WorkflowRuntime(max_grpc_message_length=8 * 1024 * 1024) + channel_options = mock_worker_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.worker.TaskHubGrpcWorker' + ) as mock_worker_cls: + WorkflowRuntime() + channel_options = mock_worker_cls.call_args[1]['channel_options'] + self.assertIsNone(channel_options) + + +if __name__ == '__main__': + unittest.main() diff --git a/tests/ext/workflow/test_workflow_util.py b/tests/ext/workflow/test_workflow_util.py index 476126858..9a195abc1 100644 --- a/tests/ext/workflow/test_workflow_util.py +++ b/tests/ext/workflow/test_workflow_util.py @@ -2,7 +2,7 @@ from unittest.mock import patch from dapr.conf import settings -from dapr.ext.workflow.util import getAddress +from dapr.ext.workflow.util import get_grpc_channel_options, getAddress class DaprWorkflowUtilTest(unittest.TestCase): @@ -27,3 +27,42 @@ def test_get_address_with_constructor_arguments_and_env_variable(self): @patch.object(settings, 'DAPR_GRPC_ENDPOINT', 'https://domain1.com:5000') def test_get_address_with_env_variable(self): self.assertEqual('https://domain1.com:5000', getAddress()) + + +class GetGrpcChannelOptionsTest(unittest.TestCase): + @patch.object(settings, 'DAPR_GRPC_MAX_INBOUND_MESSAGE_SIZE_BYTES', 0) + def test_explicit_kwarg_sets_both_directions(self): + options = get_grpc_channel_options(8 * 1024 * 1024) + self.assertEqual( + [ + ('grpc.max_send_message_length', 8 * 1024 * 1024), + ('grpc.max_receive_message_length', 8 * 1024 * 1024), + ], + options, + ) + + @patch.object(settings, 'DAPR_GRPC_MAX_INBOUND_MESSAGE_SIZE_BYTES', 16 * 1024 * 1024) + def test_env_var_sets_both_directions(self): + options = get_grpc_channel_options() + self.assertEqual( + [ + ('grpc.max_send_message_length', 16 * 1024 * 1024), + ('grpc.max_receive_message_length', 16 * 1024 * 1024), + ], + options, + ) + + @patch.object(settings, 'DAPR_GRPC_MAX_INBOUND_MESSAGE_SIZE_BYTES', 16 * 1024 * 1024) + def test_kwarg_takes_precedence_over_env_var(self): + options = get_grpc_channel_options(8 * 1024 * 1024) + self.assertEqual( + [ + ('grpc.max_send_message_length', 8 * 1024 * 1024), + ('grpc.max_receive_message_length', 8 * 1024 * 1024), + ], + options, + ) + + @patch.object(settings, 'DAPR_GRPC_MAX_INBOUND_MESSAGE_SIZE_BYTES', 0) + def test_neither_set_returns_none(self): + self.assertIsNone(get_grpc_channel_options())