diff --git a/dapr/ext/workflow/_durabletask/worker.py b/dapr/ext/workflow/_durabletask/worker.py index c1166bb8d..77ec07e41 100644 --- a/dapr/ext/workflow/_durabletask/worker.py +++ b/dapr/ext/workflow/_durabletask/worker.py @@ -478,7 +478,9 @@ def __init__( self._shutdown = Event() self._is_running = False self._secure_channel = secure_channel - self._channel_options = channel_options + # The sidecar bounds work items by its max request body size. A lower receive limit here + # only makes the worker refuse, re-fetch and refuse the same work item forever. + self._channel_options = [('grpc.max_receive_message_length', -1), *(channel_options or [])] self._stop_timeout = stop_timeout self._keepalive_interval = keepalive_interval self._current_channel: Optional[grpc.Channel] = None # Store channel reference for cleanup diff --git a/dapr/ext/workflow/workflow_runtime.py b/dapr/ext/workflow/workflow_runtime.py index def4da211..5abc5e5d1 100644 --- a/dapr/ext/workflow/workflow_runtime.py +++ b/dapr/ext/workflow/workflow_runtime.py @@ -148,7 +148,8 @@ def __init__( 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 + then the gRPC defaults: no send limit, and the worker receives work + items of any size the sidecar sends. ``0`` in either source means "no opinion" and falls through to the next source. disable_stateful_history: Stop advertising ``WORKER_CAPABILITY_STATEFUL_HISTORY`` to the sidecar, so every diff --git a/tests/ext/workflow/durabletask/test_worker_large_work_item.py b/tests/ext/workflow/durabletask/test_worker_large_work_item.py new file mode 100644 index 000000000..4eff54073 --- /dev/null +++ b/tests/ext/workflow/durabletask/test_worker_large_work_item.py @@ -0,0 +1,101 @@ +# -*- 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. + +"""A work item larger than gRPC's 4MB default receive limit still reaches the worker.""" + +import json +import threading +import time +from collections.abc import Iterator +from concurrent import futures +from unittest.mock import patch + +import grpc +from google.protobuf import empty_pb2, wrappers_pb2 + +import dapr.ext.workflow._durabletask.internal.orchestrator_service_pb2_grpc as stubs +import dapr.ext.workflow._durabletask.internal.protos as pb +import dapr.ext.workflow._durabletask.internal.shared as shared +from dapr.ext.workflow._durabletask import task +from dapr.ext.workflow._durabletask.worker import TaskHubGrpcWorker + +# A fan-out of five 1MB activity results is a work item of about 6.3MB. +_PAYLOAD_BYTES = 6 * 1024 * 1024 + + +def payload_length(ctx: task.ActivityContext, data: str) -> int: + return len(data) + + +class _Sidecar(stubs.TaskHubSidecarServiceServicer): + """Serves one large activity work item on every stream until its result arrives.""" + + def __init__(self, work_item: pb.WorkItem): + self._work_item = work_item + self.streams = 0 + self.response: pb.ActivityResponse | None = None + self.done = threading.Event() + + def Hello(self, request: empty_pb2.Empty, context: grpc.ServicerContext) -> empty_pb2.Empty: # noqa: N802 + return empty_pb2.Empty() + + def GetWorkItems( # noqa: N802 + self, request: pb.GetWorkItemsRequest, context: grpc.ServicerContext + ) -> Iterator[pb.WorkItem]: + self.streams += 1 + if not self.done.is_set(): + yield self._work_item + while context.is_active() and not self.done.is_set(): + time.sleep(0.05) + + def CompleteActivityTask( # noqa: N802 + self, request: pb.ActivityResponse, context: grpc.ServicerContext + ) -> pb.CompleteTaskResponse: + self.response = request + self.done.set() + return pb.CompleteTaskResponse() + + +def test_work_item_larger_than_default_receive_limit_is_processed() -> None: + activity_input = wrappers_pb2.StringValue(value=json.dumps('x' * _PAYLOAD_BYTES)) + activity_request = pb.ActivityRequest( + name=task.get_name(payload_length), + taskId=1, + workflowInstance=pb.WorkflowInstance(instanceId='instance'), + input=activity_input, + ) + sidecar = _Sidecar(pb.WorkItem(activityRequest=activity_request)) + server = grpc.server(futures.ThreadPoolExecutor(max_workers=4)) + stubs.add_TaskHubSidecarServiceServicer_to_server(sidecar, server) + port = server.add_insecure_port('127.0.0.1:0') + server.start() + worker = TaskHubGrpcWorker(host_address=f'127.0.0.1:{port}') + worker.add_activity(payload_length) + worker.start() + try: + assert sidecar.done.wait(15), ( + f'no activity result after {sidecar.streams} work item streams: ' + 'the worker could not receive the work item' + ) + assert sidecar.response is not None + assert sidecar.response.result.value == str(_PAYLOAD_BYTES) + finally: + worker.stop() + server.stop(0) + + +def test_explicit_receive_limit_is_kept() -> None: + worker = TaskHubGrpcWorker(channel_options=[('grpc.max_receive_message_length', 1024)]) + with patch.object(grpc, 'insecure_channel') as insecure_channel: + shared.get_grpc_channel(worker._host_address, options=worker._channel_options) + options_merged = dict(insecure_channel.call_args.kwargs['options']) + assert options_merged['grpc.max_receive_message_length'] == 1024