diff --git a/src/Dapr.Workflow/Worker/Grpc/GrpcProtocolHandler.cs b/src/Dapr.Workflow/Worker/Grpc/GrpcProtocolHandler.cs index e77872e09..9123f2652 100644 --- a/src/Dapr.Workflow/Worker/Grpc/GrpcProtocolHandler.cs +++ b/src/Dapr.Workflow/Worker/Grpc/GrpcProtocolHandler.cs @@ -231,23 +231,36 @@ private async Task ProcessWorkflowAsync(OrchestratorRequest request, string comp { _logger.LogGrpcProtocolHandlerWorkflowProcessorStart(request.InstanceId, activeCount); - var result = await handler(request, completionToken); - - // Send the result back to Dapr - var grpcCallOptions = CreateCallOptions(cancellationToken); - await _grpcClient.CompleteOrchestratorTaskAsync(result, grpcCallOptions); - } - catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested) - { - _logger.LogGrpcProtocolHandlerWorkflowProcessorCanceled(request.InstanceId); - } - catch (Exception ex) - { + // Execute the orchestrator and determine the response (normal actions or application failure). + // This try/catch must NOT include the CompleteOrchestratorTaskAsync call below — a transport + // failure during delivery must not be converted into an orchestrator-level failure, as that + // would incorrectly mark a healthy workflow turn as failed. + OrchestratorResponse result; + try + { + result = await handler(request, completionToken); + } + catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested) + { + _logger.LogGrpcProtocolHandlerWorkflowProcessorCanceled(request.InstanceId); + return; + } + catch (Exception ex) + { + result = CreateWorkflowFailureResult(request, completionToken, ex); + } + + // Send the result back to Dapr. If delivery fails (e.g. transient gRPC error or + // "no such instance exists"), log and abandon — do NOT send a secondary failure + // response, as that would corrupt the workflow history. try { - var failureResult = CreateWorkflowFailureResult(request, completionToken, ex); var grpcCallOptions = CreateCallOptions(cancellationToken); - await _grpcClient.CompleteOrchestratorTaskAsync(failureResult, grpcCallOptions); + await _grpcClient.CompleteOrchestratorTaskAsync(result, grpcCallOptions); + } + catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested) + { + _logger.LogGrpcProtocolHandlerWorkflowProcessorCanceled(request.InstanceId); } catch (Exception resultEx) { @@ -274,26 +287,39 @@ private async Task ProcessActivityAsync(ActivityRequest request, string completi { _logger.LogGrpcProtocolHandlerActivityProcessorStart(request.OrchestrationInstance.InstanceId, request.Name, request.TaskId, activeCount); - var result = await handler(request, completionToken); - // Send the result back to Dapr - var grpcCallOptions = CreateCallOptions(cancellationToken); - await _grpcClient.CompleteActivityTaskAsync(result, grpcCallOptions); - } - catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested) - { - _logger.LogGrpcProtocolHandlerActivityProcessorCanceled(request.Name); - } - catch (Exception ex) - { - _logger.LogGrpcProtocolHandlerActivityProcessorError(ex, request.Name, - request.OrchestrationInstance?.InstanceId); + // Execute the activity and determine the result (success or application failure). + // This try/catch must NOT include the CompleteActivityTaskAsync call below — a transport + // failure during delivery must not be converted into an application-level activity failure, + // as that would incorrectly mark a successfully-executed activity as failed. + ActivityResponse result; + try + { + result = await handler(request, completionToken); + } + catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested) + { + _logger.LogGrpcProtocolHandlerActivityProcessorCanceled(request.Name); + return; + } + catch (Exception ex) + { + _logger.LogGrpcProtocolHandlerActivityProcessorError(ex, request.Name, + request.OrchestrationInstance?.InstanceId); + result = CreateActivityFailureResult(request, completionToken, ex); + } + // Send the result back to Dapr. If delivery fails (e.g. transient gRPC error or + // "no such instance exists"), log and abandon — do NOT send a secondary failure + // response, as that would corrupt the workflow history. try { - var failureResult = CreateActivityFailureResult(request, completionToken, ex); var grpcCallOptions = CreateCallOptions(cancellationToken); - await _grpcClient.CompleteActivityTaskAsync(failureResult, grpcCallOptions); + await _grpcClient.CompleteActivityTaskAsync(result, grpcCallOptions); + } + catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested) + { + _logger.LogGrpcProtocolHandlerActivityProcessorCanceled(request.Name); } catch (Exception resultEx) { diff --git a/test/Dapr.Workflow.Test/Worker/Grpc/GrpcProtocolHandlerTests.cs b/test/Dapr.Workflow.Test/Worker/Grpc/GrpcProtocolHandlerTests.cs index 79115bb8d..131d850a1 100644 --- a/test/Dapr.Workflow.Test/Worker/Grpc/GrpcProtocolHandlerTests.cs +++ b/test/Dapr.Workflow.Test/Worker/Grpc/GrpcProtocolHandlerTests.cs @@ -859,6 +859,129 @@ public async Task StartAsync_ShouldNotExceedMaxConcurrentActivityWorkItems() Assert.Equal(totalItems, Volatile.Read(ref completedCount)); } + /// + /// Regression test for: "Task failed: Status(StatusCode="Unknown", Detail="no such instance exists")" + /// + /// Root cause: the old code had a single try/catch that wrapped both the handler execution and + /// the CompleteActivityTaskAsync delivery call. When the delivery call threw an RpcException + /// (e.g. "no such instance exists" — a transient sidecar condition), the exception was caught and + /// a *secondary* CompleteActivityTaskAsync call was made carrying FailureDetails. The sidecar + /// recorded this as a TaskFailed event. On the next workflow replay, HandleFailedActivityFromHistory + /// threw WorkflowTaskFailedException("Task failed: Status(StatusCode=Unknown, Detail=no such + /// instance exists)"), propagating the transport error as a business-logic failure. + /// + /// Fix: the delivery call is now in its own try/catch so a transport failure is logged and + /// abandoned — it does NOT produce a secondary failure response. + /// + [Fact] + public async Task StartAsync_ShouldCallCompleteActivityTaskOnce_WhenActivitySucceeds_AndCompleteActivityTaskThrowsRpcException() + { + var grpcClientMock = CreateGrpcClientMock(); + + var workItems = new[] + { + new WorkItem + { + ActivityRequest = new ActivityRequest + { + Name = "act", + TaskId = 42, + OrchestrationInstance = new OrchestrationInstance { InstanceId = "i-1" } + } + } + }; + + grpcClientMock + .Setup(x => x.GetWorkItems(It.IsAny(), It.IsAny())) + .Returns(CreateServerStreamingCall(workItems)); + + var completeCallCount = 0; + ActivityResponse? capturedResponse = null; + + grpcClientMock + .Setup(x => x.CompleteActivityTaskAsync(It.IsAny(), It.IsAny())) + .Callback((r, _) => + { + Interlocked.Increment(ref completeCallCount); + capturedResponse = r; + }) + .Throws(new RpcException(new Status(StatusCode.Unknown, "no such instance exists"))); + + var handler = new GrpcProtocolHandler(grpcClientMock.Object, NullLoggerFactory.Instance); + + await RunHandlerUntilAsync( + handler, + workflowHandler: (_, _) => Task.FromResult(new OrchestratorResponse()), + activityHandler: (req, tok) => Task.FromResult(new ActivityResponse + { + InstanceId = req.OrchestrationInstance.InstanceId, + TaskId = req.TaskId, + Result = "success", + CompletionToken = tok + }), + untilCondition: () => Volatile.Read(ref completeCallCount) >= 1, + timeout: TimeSpan.FromSeconds(2)); + + // CompleteActivityTaskAsync must be called exactly once — with the success result. + // The old code would call it a second time with FailureDetails set, causing a TaskFailed + // history event that propagates the transport error as a workflow-level activity failure. + Assert.Equal(1, Volatile.Read(ref completeCallCount)); + Assert.NotNull(capturedResponse); + Assert.Null(capturedResponse!.FailureDetails); // success result, no failure details + Assert.Equal("success", capturedResponse.Result); + } + + /// + /// Regression test for the orchestrator path of the same bug. When a workflow turn completes + /// successfully but CompleteOrchestratorTaskAsync fails transiently, the handler must log and + /// abandon rather than sending a secondary failure response that would corrupt workflow history. + /// + [Fact] + public async Task StartAsync_ShouldCallCompleteOrchestratorTaskOnce_WhenWorkflowHandlerSucceeds_AndCompleteOrchestratorTaskThrowsRpcException() + { + var grpcClientMock = CreateGrpcClientMock(); + + var workItems = new[] + { + new WorkItem + { + OrchestratorRequest = new OrchestratorRequest { InstanceId = "i-1" } + } + }; + + grpcClientMock + .Setup(x => x.GetWorkItems(It.IsAny(), It.IsAny())) + .Returns(CreateServerStreamingCall(workItems)); + + var completeCallCount = 0; + OrchestratorResponse? capturedResponse = null; + + grpcClientMock + .Setup(x => x.CompleteOrchestratorTaskAsync(It.IsAny(), It.IsAny())) + .Callback((r, _) => + { + Interlocked.Increment(ref completeCallCount); + capturedResponse = r; + }) + .Throws(new RpcException(new Status(StatusCode.Unknown, "no such instance exists"))); + + var handler = new GrpcProtocolHandler(grpcClientMock.Object, NullLoggerFactory.Instance); + + await RunHandlerUntilAsync( + handler, + workflowHandler: (req, _) => Task.FromResult(new OrchestratorResponse { InstanceId = req.InstanceId }), + activityHandler: (_, _) => Task.FromResult(new ActivityResponse()), + untilCondition: () => Volatile.Read(ref completeCallCount) >= 1, + timeout: TimeSpan.FromSeconds(2)); + + // CompleteOrchestratorTaskAsync must be called exactly once — with the success result. + // The old code would call it a second time with OrchestrationStatus.Failed, corrupting history. + Assert.Equal(1, Volatile.Read(ref completeCallCount)); + Assert.NotNull(capturedResponse); + Assert.DoesNotContain(capturedResponse!.Actions, + a => a.CompleteOrchestration?.OrchestrationStatus == OrchestrationStatus.Failed); + } + [Fact] public async Task DelayOrStopAsync_ShouldSwallowCancellation_WhenTokenIsCanceled() {