fix(worker): handle oversized gRPC payloads cleanly instead of hanging

An oversized worker<->controller message left the execution stuck in RUNNING,
in both directions. Now the task fails cleanly:
- worker->controller: TaskRun.fail() no longer throws on an immutable attempts
  list, so the RESOURCE_EXHAUSTED fallback can resend the failed result.
- controller->worker: the worker advertises its maxInboundMessageSize and the
  controller fails an oversized job before dispatch instead of sending a frame
  the worker can never receive.

Author: fhussonnois

core/src/main/java/io/kestra/core/models/executions/TaskRun.java

@@ -142,7 +142,9 @@ public TaskRun withStateAndAttempt(State.Type state) {
 
     public TaskRun fail() {
         var attempt = TaskRunAttempt.builder().state(new State(State.Type.FAILED)).build();
-        List<TaskRunAttempt> newAttempts = this.attempts == null ? new ArrayList<>(1) : this.attempts;
+        // Copy defensively: this.attempts may be immutable (e.g. after deserialization), and fail()
+        // must return a new TaskRun without mutating the caller's list in place.
+        List<TaskRunAttempt> newAttempts = this.attempts == null ? new ArrayList<>(1) : new ArrayList<>(this.attempts);
         newAttempts.add(attempt);
 
         return new TaskRun(

worker-controller/src/main/java/io/kestra/controller/grpc/services/GrpcWorkerControllerService.java

@@ -129,6 +129,7 @@ public void onNext(WorkerJobRequest request) {
                     WorkerStreamContext<WorkerJobResponse> context = new WorkerStreamContext<>(
                         workerId, workerGroupId, subscriptions, maxConcurrency, responseObserver, capacityPolicy
                     );
+                    context.setMaxInboundMessageSize(connInfo.getMaxInboundMessageSize());
                     contextRef.set(context);
 
                     // Register with dispatcher

worker-controller/src/main/java/io/kestra/controller/grpc/services/WorkerJobDispatcher.java

@@ -866,6 +866,25 @@ private void dispatchJobToWorker(WorkerStreamContext<WorkerJobResponse> context,
 
         // The permit and bucket slot were already atomically reserved in findAndReserveWorker.
 
+        WorkerJobResponse response = WorkerJobResponse.newBuilder()
+            .setHeader(RequestOrResponseHeaderFactory.create(context.getWorkerId()))
+            .addJobs(
+                WorkerJobPayload.newBuilder()
+                    .setJobId(jobId)
+                    .setJobData(MessageFormats.JSON.toByteString(job))
+                    .build()
+            )
+            .build();
+
+        // Reject a payload the worker's gRPC channel could never receive: dispatching it would
+        // trigger RESOURCE_EXHAUSTED on the worker stream, hot-loop reconnects, and hang the
+        // execution in RUNNING forever. Fail the job cleanly instead.
+        int workerLimit = context.getMaxInboundMessageSize();
+        if (workerLimit > 0 && response.getSerializedSize() > workerLimit) {
+            rejectOversizedJob(context, job, dispatchWorkerQueueId, bucket, response.getSerializedSize(), workerLimit);
+            return;
+        }
+
         // 1. PERSIST before sending (critical for recovery)
         persistJobToStateStore(context, job, dispatchWorkerQueueId);
 
@@ -874,16 +893,6 @@ private void dispatchJobToWorker(WorkerStreamContext<WorkerJobResponse> context,
 
         // 3. Send to worker
         try {
-            WorkerJobResponse response = WorkerJobResponse.newBuilder()
-                .setHeader(RequestOrResponseHeaderFactory.create(context.getWorkerId()))
-                .addJobs(
-                    WorkerJobPayload.newBuilder()
-                        .setJobId(jobId)
-                        .setJobData(MessageFormats.JSON.toByteString(job))
-                        .build()
-                )
-                .build();
-
             context.sendResponse(response);
             log.debug("Dispatched job {} to worker {}", jobId, context.getWorkerId());
             metricRegistry.counter(
@@ -941,6 +950,38 @@ private void handleDispatchFailure(WorkerStreamContext<WorkerJobResponse> contex
         requeue(originalEvent);
     }
 
+    /**
+     * Rejects a job whose serialized payload exceeds the worker's advertised gRPC inbound
+     * limit. Releases the reserved permit and bucket and fails the job cleanly so the
+     * execution terminates instead of hanging while the worker hot-loops reconnecting.
+     */
+    private void rejectOversizedJob(WorkerStreamContext<WorkerJobResponse> context, WorkerJob job,
+        String dispatchWorkerQueueId, String bucket, int payloadSize, int workerLimit) {
+        log.error("Job {} payload ({} bytes) exceeds worker {} max inbound gRPC message size ({} bytes); failing the job instead of dispatching",
+            job.uid(), payloadSize, context.getWorkerId(), workerLimit);
+
+        metricRegistry.counter(
+            MetricRegistry.METRIC_CONTROLLER_JOB_DISPATCH_FAILED_TOTAL,
+            MetricRegistry.METRIC_CONTROLLER_JOB_DISPATCH_FAILED_TOTAL_DESCRIPTION,
+            metricRegistry.workerGroupAndQueueTags(context.getWorkerGroupId(), dispatchWorkerQueueId)
+        ).increment();
+
+        // Release the permit + bucket reserved in findAndReserveWorker (the job is not dispatched).
+        context.addPermits(1);
+        context.releaseBucket(bucket);
+
+        // Fail the job cleanly so the execution reaches a terminal state.
+        if (job instanceof WorkerTask workerTask) {
+            try {
+                workerTaskResultQueue.emit(new WorkerTaskResult(workerTask.getTaskRun().fail()));
+            } catch (QueueException e) {
+                log.error("Failed to emit FAILED result for oversized job {}: {}", job.uid(), e.getMessage(), e);
+            }
+        } else if (job instanceof WorkerTrigger workerTrigger) {
+            triggerEventQueue.send(new TriggerEvaluated(workerTrigger.triggerId(), null));
+        }
+    }
+
     /**
      * Re-queues a job event back to the queue.
      */

worker-controller/src/main/java/io/kestra/controller/grpc/services/WorkerStreamContext.java

@@ -45,6 +45,9 @@ public class WorkerStreamContext<T> {
     private final StreamObserver<T> responseObserver;
     private final WorkerCapacityPolicy capacityPolicy;
 
+    /** Worker's advertised gRPC max inbound message size (bytes); 0 = not advertised. */
+    private volatile int maxInboundMessageSize;
+
     /**
      * Number of jobs the worker has capacity to receive.
      * Decremented when a job is sent, incremented when worker sends permits.
@@ -308,6 +311,11 @@ public int getInFlightCount() {
         return inFlightJobs.size();
     }
 
+    /** Records the worker's advertised gRPC max inbound message size (bytes). */
+    public void setMaxInboundMessageSize(int maxInboundMessageSize) {
+        this.maxInboundMessageSize = maxInboundMessageSize;
+    }
+
     /**
      * Lock to serialize onNext() calls on the response stream observer.
      * gRPC's StreamObserver is not thread-safe for concurrent onNext() calls.

worker-controller/src/main/proto/worker_controller.proto

@@ -48,6 +48,9 @@ message WorkerConnectionInfo {
 
   // Maximum concurrent jobs this worker can handle (based on thread count)
   int32 maxConcurrency = 3;
+
+  // Worker's gRPC max inbound message size (bytes).
+  int32 max_inbound_message_size = 4;
 }
 
 // Controller -> Worker: Jobs to be executed

worker/src/main/java/io/kestra/worker/fetchers/WorkerJobFetcher.java

@@ -13,6 +13,7 @@
 import com.google.protobuf.ByteString;
 
 import io.kestra.controller.GrpcChannelManager;
+import io.kestra.controller.config.GrpcConfiguration;
 import io.kestra.controller.grpc.WorkerConnectionInfo;
 import io.kestra.controller.grpc.WorkerControllerServiceGrpc.WorkerControllerServiceStub;
 import io.kestra.controller.grpc.WorkerJobPayload;
@@ -92,6 +93,7 @@ public class WorkerJobFetcher extends WorkerLoop implements JobFetcher {
     private final GrpcChannelManager channelManager;
     private final WorkerQueueRegistry workerQueueRegistry;
     private final ExecutionKilledManager executionKilledManager;
+    private final GrpcConfiguration grpcConfiguration;
     private final BroadcastQueueInterface<ClusterEvent> clusterEventQueue;
     private final List<WorkerMetadataChangeHandler> metadataChangeHandlers;
 
@@ -172,12 +174,14 @@ public WorkerJobFetcher(final WorkerControllerServiceStub workerControllerServic
         final WorkerQueueRegistry workerQueueRegistry,
         final ExecutionKilledManager executionKilledManager,
         @Nullable @Named(WORKER_LOCAL_CLUSTER_EVENTS) final BroadcastQueueInterface<ClusterEvent> clusterEventQueue,
-        final List<WorkerMetadataChangeHandler> metadataChangeHandlers) {
+        final List<WorkerMetadataChangeHandler> metadataChangeHandlers,
+        final GrpcConfiguration grpcConfiguration) {
         super(WorkerJobFetcher.class.getSimpleName());
         this.workerQueueRegistry = workerQueueRegistry;
         this.workerControllerServiceStub = workerControllerServiceStub;
         this.channelManager = channelManager;
         this.executionKilledManager = executionKilledManager;
+        this.grpcConfiguration = grpcConfiguration;
         this.clusterEventQueue = clusterEventQueue;
         this.metadataChangeHandlers = metadataChangeHandlers;
     }
@@ -308,6 +312,7 @@ private void sendInitialRequest(ClientCallStreamObserver<WorkerJobRequest> reque
                     .setWorkerId(workerContext.workerId())
                     .setWorkerGroupId(workerGroupId)
                     .setMaxConcurrency(maxConcurrency)
+                    .setMaxInboundMessageSize(grpcConfiguration.maxInboundMessageSize())
                     .build()
             );
         addPendingCompletions(requestBuilder);

worker/src/test/java/io/kestra/worker/fetchers/WorkerJobFetcherMetadataChangeTest.java

@@ -8,6 +8,7 @@
 import io.kestra.worker.queues.WorkerQueueRegistry;
 import io.kestra.worker.services.ExecutionKilledManager;
 import io.kestra.controller.GrpcChannelManager;
+import io.kestra.controller.config.GrpcConfiguration;
 import io.kestra.controller.grpc.WorkerControllerServiceGrpc.WorkerControllerServiceStub;
 import org.junit.jupiter.api.Test;
 
@@ -29,7 +30,8 @@ private WorkerJobFetcher newFetcher(List<WorkerMetadataChangeHandler> handlers)
             mock(WorkerQueueRegistry.class),
             mock(ExecutionKilledManager.class),
             (BroadcastQueueInterface<ClusterEvent>) mock(BroadcastQueueInterface.class),
-            handlers
+            handlers,
+            new GrpcConfiguration(false, 10485760)
         );
     }
 
@@ -93,7 +95,8 @@ void shouldStillRouteKillEvents() {
             mock(WorkerQueueRegistry.class),
             killed,
             null,
-            List.of(handler)
+            List.of(handler),
+            new GrpcConfiguration(false, 10485760)
         );
 
         io.kestra.core.models.executions.ExecutionKilled k = io.kestra.core.models.executions.ExecutionKilledExecution.builder()