diff --git a/durabletask-client/src/main/java/io/dapr/durabletask/DurableTaskGrpcWorker.java b/durabletask-client/src/main/java/io/dapr/durabletask/DurableTaskGrpcWorker.java index ce63c27a96..96bbe322fe 100644 --- a/durabletask-client/src/main/java/io/dapr/durabletask/DurableTaskGrpcWorker.java +++ b/durabletask-client/src/main/java/io/dapr/durabletask/DurableTaskGrpcWorker.java @@ -101,7 +101,10 @@ public final class DurableTaskGrpcWorker implements AutoCloseable { this.tracer = GlobalOpenTelemetry.getTracer("dapr-workflow"); - this.sidecarClient = TaskHubSidecarServiceGrpc.newBlockingStub(sidecarGrpcChannel); + // The sidecar bounds work items and histories 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. + this.sidecarClient = TaskHubSidecarServiceGrpc.newBlockingStub(sidecarGrpcChannel) + .withMaxInboundMessageSize(Integer.MAX_VALUE); this.dataConverter = builder.dataConverter != null ? builder.dataConverter : new JacksonDataConverter(); this.maximumTimerInterval = builder.maximumTimerInterval != null ? builder.maximumTimerInterval : DEFAULT_MAXIMUM_TIMER_INTERVAL; diff --git a/durabletask-client/src/test/java/io/dapr/durabletask/DurableTaskGrpcWorkerLargeWorkItemTest.java b/durabletask-client/src/test/java/io/dapr/durabletask/DurableTaskGrpcWorkerLargeWorkItemTest.java new file mode 100644 index 0000000000..8b6190bddd --- /dev/null +++ b/durabletask-client/src/test/java/io/dapr/durabletask/DurableTaskGrpcWorkerLargeWorkItemTest.java @@ -0,0 +1,134 @@ +/* + * 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. + */ + +package io.dapr.durabletask; + +import com.google.protobuf.StringValue; +import io.dapr.durabletask.implementation.protobuf.Orchestration; +import io.dapr.durabletask.implementation.protobuf.OrchestratorService; +import io.dapr.durabletask.implementation.protobuf.TaskHubSidecarServiceGrpc; +import io.grpc.ManagedChannel; +import io.grpc.ManagedChannelBuilder; +import io.grpc.Server; +import io.grpc.netty.NettyServerBuilder; +import io.grpc.stub.StreamObserver; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.Test; + +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; +import java.util.concurrent.atomic.AtomicInteger; + +import static org.junit.jupiter.api.Assertions.assertEquals; + +/** + * Verifies DurableTaskGrpcWorker accepts a work item larger than gRPC's 4MB default receive limit. + * + *
daprd sends work items up to its own max body size, so a fan-out of five 1MB activity results + * is a work item of about 6MB. The channel is built like the SDK's own, whose inbound limit defaults + * to 4MB (DAPR_GRPC_MAX_INBOUND_MESSAGE_SIZE_BYTES). Uses a real TCP server because the in-process + * transport does not enforce message sizes.
+ */ +class DurableTaskGrpcWorkerLargeWorkItemTest { + + private static final int DEFAULT_MAX_INBOUND_BYTES = 4 * 1024 * 1024; + private static final int PAYLOAD_CHARS = 6 * 1024 * 1024; + private static final String ACTIVITY_NAME = "PayloadLength"; + + private DurableTaskGrpcWorker worker; + private ManagedChannel channel; + private Server server; + + @AfterEach + void tearDown() throws Exception { + if (worker != null) { + worker.close(); + } + if (channel != null) { + channel.shutdownNow().awaitTermination(5, TimeUnit.SECONDS); + } + if (server != null) { + server.shutdownNow().awaitTermination(5, TimeUnit.SECONDS); + } + } + + @Test + void workerProcessesWorkItemLargerThanDefaultReceiveLimit() throws Exception { + OrchestratorService.ActivityRequest activityRequest = OrchestratorService.ActivityRequest.newBuilder() + .setName(ACTIVITY_NAME) + .setTaskId(1) + .setWorkflowInstance(Orchestration.WorkflowInstance.newBuilder().setInstanceId("instance").build()) + .setInput(StringValue.of("\"" + "x".repeat(PAYLOAD_CHARS) + "\"")) + .build(); + OrchestratorService.WorkItem workItem = OrchestratorService.WorkItem.newBuilder() + .setActivityRequest(activityRequest) + .build(); + + AtomicInteger streams = new AtomicInteger(); + CompletableFuture