diff --git a/src/WebJobs.Extensions.DurableTask/OutOfProcMiddleware.cs b/src/WebJobs.Extensions.DurableTask/OutOfProcMiddleware.cs index b8aafa31b..e45102218 100644 --- a/src/WebJobs.Extensions.DurableTask/OutOfProcMiddleware.cs +++ b/src/WebJobs.Extensions.DurableTask/OutOfProcMiddleware.cs @@ -24,6 +24,7 @@ internal class OutOfProcMiddleware private const string NoProcessAssociatedMessage = "No process is associated"; private const string NoWorkerInitializedMessage = "Did not find any initialized language workers"; private const string AssemblyNotLoadedMessage = "Could not load file or assembly"; + private const string WorkerDrainingMessageMarker = "[DurableTask:WorkerDraining]"; private readonly DurableTaskExtension extension; @@ -124,6 +125,7 @@ await this.LifeCycleNotificationHelper.OrchestratorStartingAsync( workItemMetadata.IncludeState); bool workerRequiresHistory = false; + bool isWorkerDraining = false; var input = new TriggeredFunctionData { @@ -153,6 +155,7 @@ await this.LifeCycleNotificationHelper.OrchestratorStartingAsync( P.OrchestratorResponse response = P.OrchestratorResponse.Parser.ParseFrom(triggerReturnValueBytes); workerRequiresHistory = response.RequiresHistory; + isWorkerDraining = response.IsWorkerDraining; // TrySetResult may throw if a platform-level error is encountered (like an out of memory exception). context.SetResult( @@ -177,9 +180,15 @@ await this.LifeCycleNotificationHelper.OrchestratorStartingAsync( { // Shutdown can surface as a completed invocation in a failed state. // Re-throw so we can abort this invocation. - this.HostLifetimeService.OnStopping.ThrowIfCancellationRequested(); - } - + this.HostLifetimeService.OnStopping.ThrowIfCancellationRequested(); + + // The Function may have failed because the worker is draining, so we want to retry its execution in this case + if (isWorkerDraining) + { + throw new Exception("Worker is shutting down. The orchestration execution will be aborted and retried."); + } + } + // we abort the invocation on "platform level errors" such as: // - a timeout // - an out of memory exception @@ -193,7 +202,9 @@ await this.LifeCycleNotificationHelper.OrchestratorStartingAsync( { string reason = this.HostLifetimeService.OnStopping.IsCancellationRequested ? "The Functions/WebJobs runtime is shutting down!" : - $"Unhandled exception in the Functions/WebJobs runtime: {hostRuntimeException}"; + isWorkerDraining ? + "The worker is shutting down. The orchestration execution will be aborted and retried." : + $"Unhandled exception in the Functions/WebJobs runtime: {hostRuntimeException}"; this.TraceHelper.FunctionAborted( this.Options.HubName, @@ -365,6 +376,7 @@ void SetErrorResult(FailureDetails failureDetails) workItemMetadata.IncludeState); bool workerRequiresEntityState = false; + bool isWorkerDraining = false; var input = new TriggeredFunctionData { @@ -393,6 +405,7 @@ void SetErrorResult(FailureDetails failureDetails) byte[] triggerReturnValueBytes = Convert.FromBase64String(triggerReturnValue); P.EntityBatchResult response = P.EntityBatchResult.Parser.ParseFrom(triggerReturnValueBytes); workerRequiresEntityState = response.RequiresState; + isWorkerDraining = response.IsWorkerDraining; context.Result = response.ToEntityBatchResult(this.Options.DefaultVersion); context.ThrowIfFailed(); @@ -418,6 +431,12 @@ void SetErrorResult(FailureDetails failureDetails) throw functionResult.Exception; } + // The Function may have failed because the worker is draining, so we want to retry its execution in this case + if (isWorkerDraining) + { + throw new Exception("Worker is shutting down. The entity execution will be aborted and retried."); + } + // Shutdown can surface as a completed invocation in a failed state. // Re-throw so we can abort this invocation. this.HostLifetimeService.OnStopping.ThrowIfCancellationRequested(); @@ -427,7 +446,9 @@ void SetErrorResult(FailureDetails failureDetails) { string reason = this.HostLifetimeService.OnStopping.IsCancellationRequested ? "The Functions/WebJobs runtime is shutting down!" : - $"Unhandled exception in the Functions/WebJobs runtime: {hostRuntimeException}"; + isWorkerDraining ? + "The worker is shutting down. The entity execution will be aborted and retried." : + $"Unhandled exception in the Functions/WebJobs runtime: {hostRuntimeException}"; this.TraceHelper.FunctionAborted( this.Options.HubName, @@ -596,6 +617,19 @@ public async Task CallActivityAsync(DispatchMiddlewareContext dispatchContext, F // This will abort the current execution and force an durable retry throw new SessionAbortedException(reason); + } + + // The Function may have failed because the worker is draining, so we want to retry its execution in this case + if (!result.Succeeded && IsWorkerDrainingException(result.Exception)) + { + this.TraceHelper.FunctionAborted( + this.Options.HubName, + functionName.Name, + instance.InstanceId, + "The worker is shutting down. The activity execution will be aborted and retried.", + functionType: FunctionType.Activity); + + throw new SessionAbortedException("The worker is shutting down. The activity execution will be aborted and retried."); } ActivityExecutionResult activityResult; @@ -681,6 +715,19 @@ private static bool IsWorkerNotFullyInitializedException(Exception? exception) { return (exception?.InnerException is InvalidOperationException ioe && ioe.Message.Contains(NoWorkerInitializedMessage, StringComparison.Ordinal)) || (exception?.InnerException is FileNotFoundException fnfe && fnfe.Message.Contains(AssemblyNotLoadedMessage, StringComparison.Ordinal)); + } + + private static bool IsWorkerDrainingException(Exception? exception) + { + for (Exception? current = exception; current != null; current = current.InnerException) + { + if (current.Message != null && current.Message.Contains(WorkerDrainingMessageMarker, StringComparison.Ordinal)) + { + return true; + } + } + + return false; } private static FailureDetails GetFailureDetails(Exception e, out bool fromSerializedException) diff --git a/src/Worker.Extensions.DurableTask/Exceptions/WorkerDrainingException.cs b/src/Worker.Extensions.DurableTask/Exceptions/WorkerDrainingException.cs new file mode 100644 index 000000000..bdd9e8b66 --- /dev/null +++ b/src/Worker.Extensions.DurableTask/Exceptions/WorkerDrainingException.cs @@ -0,0 +1,24 @@ +// Copyright (c) .NET Foundation. All rights reserved. +// Licensed under the MIT License. See License.txt in the project root for license information. + +using System; + +#nullable enable +namespace Microsoft.Azure.Functions.Worker.Extensions.DurableTask.Exceptions; + +/// +/// Thrown by the worker when a Function completes while the worker is draining (shutting down). +/// +internal sealed class WorkerDrainingException : Exception +{ + /// + /// Stable marker the host matches on to detect this exception after it crosses the gRPC boundary + /// as a serialized string. The host keeps an identical copy of this literal; keep them in sync. + /// + internal const string Sentinel = "[DurableTask:WorkerDraining]"; + + public WorkerDrainingException() + : base($"The worker is shutting down and will not commit this result; the work item should be retried on another worker. {Sentinel}") + { + } +} diff --git a/src/Worker.Extensions.DurableTask/Execution/DurableFunctionExecutor.Activity.cs b/src/Worker.Extensions.DurableTask/Execution/DurableFunctionExecutor.Activity.cs index 163941855..b4fe8cb84 100644 --- a/src/Worker.Extensions.DurableTask/Execution/DurableFunctionExecutor.Activity.cs +++ b/src/Worker.Extensions.DurableTask/Execution/DurableFunctionExecutor.Activity.cs @@ -28,11 +28,16 @@ private async ValueTask RunActivityAsync(FunctionContext context, BindingMetadat if (context.FunctionDefinition.EntryPoint == ActivityEntryPoint) { await this.RunDirectActivityAsync(context, triggerBinding); - return; + } + else + { + await inner.ExecuteAsync(context); } - await inner.ExecuteAsync(context); - return; + if (this.IsWorkerDraining) + { + throw new WorkerDrainingException(); + } } catch (Exception ex) { diff --git a/src/Worker.Extensions.DurableTask/Execution/DurableFunctionExecutor.Entity.cs b/src/Worker.Extensions.DurableTask/Execution/DurableFunctionExecutor.Entity.cs index 32e198421..b8de5443d 100644 --- a/src/Worker.Extensions.DurableTask/Execution/DurableFunctionExecutor.Entity.cs +++ b/src/Worker.Extensions.DurableTask/Execution/DurableFunctionExecutor.Entity.cs @@ -3,9 +3,11 @@ using System; using System.Threading.Tasks; +using Google.Protobuf; using Microsoft.DurableTask.Entities; using Microsoft.DurableTask.Worker; using Microsoft.DurableTask.Worker.Grpc; +using P = Microsoft.DurableTask.Protobuf; namespace Microsoft.Azure.Functions.Worker.Extensions.DurableTask.Execution; @@ -40,7 +42,15 @@ private async ValueTask RunEntityAsync(FunctionContext context, BindingMetadata TaskEntityDispatcher dispatcher = new(encodedEntityBatch, context.InstanceServices, extendedSessionsCache); triggerInputData.Value = dispatcher; await inner.ExecuteAsync(context); - context.GetInvocationResult().Value = dispatcher.Result; + + string entityResult = dispatcher.Result; + + if (this.IsWorkerDraining) + { + entityResult = FlagEntityDraining(entityResult); + } + + context.GetInvocationResult().Value = entityResult; } private async Task RunDirectEntityAsync( @@ -61,6 +71,20 @@ private async Task RunDirectEntityAsync( string result = await GrpcEntityRunner.LoadAndRunAsync( encodedEntityBatch, entity, context.InstanceServices); + + if (this.IsWorkerDraining) + { + result = FlagEntityDraining(result); + } + context.GetInvocationResult().Value = result; } + + private static string FlagEntityDraining(string encodedEntityBatchResult) + { + byte[] resultBytes = Convert.FromBase64String(encodedEntityBatchResult); + P.EntityBatchResult result = P.EntityBatchResult.Parser.ParseFrom(resultBytes); + result.IsWorkerDraining = true; + return Convert.ToBase64String(result.ToByteArray()); + } } diff --git a/src/Worker.Extensions.DurableTask/Execution/DurableFunctionExecutor.Orchestration.cs b/src/Worker.Extensions.DurableTask/Execution/DurableFunctionExecutor.Orchestration.cs index dc64c2872..c328d66a2 100644 --- a/src/Worker.Extensions.DurableTask/Execution/DurableFunctionExecutor.Orchestration.cs +++ b/src/Worker.Extensions.DurableTask/Execution/DurableFunctionExecutor.Orchestration.cs @@ -3,8 +3,10 @@ using System; using System.Threading.Tasks; +using Google.Protobuf; using Microsoft.DurableTask; using Microsoft.DurableTask.Worker.Grpc; +using P = Microsoft.DurableTask.Protobuf; namespace Microsoft.Azure.Functions.Worker.Extensions.DurableTask.Execution; @@ -34,10 +36,23 @@ private async ValueTask RunOrchestrationAsync(FunctionContext context, BindingMe string orchestratorOutput = GrpcOrchestrationRunner.LoadAndRun( encodedOrchestratorState, orchestrator, extendedSessionsCache, context.InstanceServices); + if (this.IsWorkerDraining) + { + orchestratorOutput = FlagOrchestratorDraining(orchestratorOutput); + } + // Send the encoded orchestrator output as the return value seen by the functions host extension context.GetInvocationResult().Value = orchestratorOutput; } + private static string FlagOrchestratorDraining(string encodedOrchestratorOutput) + { + byte[] responseBytes = Convert.FromBase64String(encodedOrchestratorOutput); + P.OrchestratorResponse response = P.OrchestratorResponse.Parser.ParseFrom(responseBytes); + response.IsWorkerDraining = true; + return Convert.ToBase64String(response.ToByteArray()); + } + private IDisposableOrchestrator CreateOrchestrator( FunctionContext context, InputBindingData triggerInputData) { diff --git a/src/Worker.Extensions.DurableTask/Execution/DurableFunctionExecutor.cs b/src/Worker.Extensions.DurableTask/Execution/DurableFunctionExecutor.cs index facdb7db4..a18c52b25 100644 --- a/src/Worker.Extensions.DurableTask/Execution/DurableFunctionExecutor.cs +++ b/src/Worker.Extensions.DurableTask/Execution/DurableFunctionExecutor.cs @@ -6,6 +6,7 @@ using Microsoft.Azure.Functions.Worker.Invocation; using Microsoft.DurableTask; using Microsoft.DurableTask.Worker; +using Microsoft.Extensions.Hosting; using Microsoft.Extensions.Options; namespace Microsoft.Azure.Functions.Worker.Extensions.DurableTask.Execution; @@ -15,11 +16,15 @@ internal partial class DurableFunctionExecutor( ExtendedSessionsCache extendedSessionsCache, IDurableTaskFactory factory, IOptions options, + IHostApplicationLifetime? hostApplicationLifetime = null, IExceptionPropertiesProvider? exceptionPropertiesProvider = null) : IFunctionExecutor { private DataConverter Converter => options.Value.DataConverter; + private bool IsWorkerDraining => + hostApplicationLifetime?.ApplicationStopping.IsCancellationRequested ?? false; + public virtual ValueTask ExecuteAsync(FunctionContext context) { if (context is null) diff --git a/test/FunctionsV2/Tests/Unit/OutOfProcMiddlewareTests.cs b/test/FunctionsV2/Tests/Unit/OutOfProcMiddlewareTests.cs index d3a476ccf..b03df0984 100644 --- a/test/FunctionsV2/Tests/Unit/OutOfProcMiddlewareTests.cs +++ b/test/FunctionsV2/Tests/Unit/OutOfProcMiddlewareTests.cs @@ -12,19 +12,23 @@ using DurableTask.Core.Exceptions; using DurableTask.Core.History; using DurableTask.Core.Middleware; +using Google.Protobuf; using Microsoft.Azure.WebJobs.Host; using Microsoft.Azure.WebJobs.Host.Executors; using Microsoft.Extensions.Logging.Abstractions; using Microsoft.Extensions.Options; using Moq; using Xunit; +using P = Microsoft.DurableTask.Protobuf; +#nullable enable namespace Microsoft.Azure.WebJobs.Extensions.DurableTask.Tests { public class OutOfProcMiddlewareTests { private const string NoWorkerInitializedMessage = "Did not find any initialized language workers"; private const string AssemblyNotLoadedMessage = "Could not load file or assembly"; + private const string WorkerDrainingMessageMarker = "[DurableTask:WorkerDraining]"; [Fact] [Trait("Category", PlatformSpecificHelpers.TestCategory)] @@ -143,6 +147,187 @@ await Assert.ThrowsAsync( () => middleware.CallActivityAsync(dispatchContext, () => Task.CompletedTask)); } + [Fact] + [Trait("Category", PlatformSpecificHelpers.TestCategory)] + public async Task CallActivityAsync_WorkerDrainingException_ThrowsSessionAbortedException() + { + // Arrange: an activity failure whose message contains the worker-draining marker indicates the + // worker completed the activity while shutting down. The activity should be aborted and retried + // on a healthy worker rather than recording a result produced during shutdown. + var exception = new Exception( + $"The worker is shutting down and will not commit this result. {WorkerDrainingMessageMarker}"); + + (OutOfProcMiddleware middleware, DispatchMiddlewareContext dispatchContext) = this.SetupActivityTest(exception); + + await Assert.ThrowsAsync( + () => middleware.CallActivityAsync(dispatchContext, () => Task.CompletedTask)); + } + + [Fact] + [Trait("Category", PlatformSpecificHelpers.TestCategory)] + public async Task CallActivityAsync_WorkerDrainingException_NestedInInnerException_ThrowsSessionAbortedException() + { + // Arrange: the worker-draining marker may be nested in an inner exception after crossing the + // gRPC boundary. The host should still detect it by walking the inner-exception chain. + var exception = new Exception( + "Function invocation failed.", + new InvalidOperationException( + $"The worker is shutting down and will not commit this result. {WorkerDrainingMessageMarker}")); + + (OutOfProcMiddleware middleware, DispatchMiddlewareContext dispatchContext) = this.SetupActivityTest(exception); + + await Assert.ThrowsAsync( + () => middleware.CallActivityAsync(dispatchContext, () => Task.CompletedTask)); + } + + [Fact] + [Trait("Category", PlatformSpecificHelpers.TestCategory)] + public async Task CallActivityAsync_UnrelatedFailure_DoesNotThrowSessionAbortedException() + { + // Arrange: an ordinary activity failure (no worker-draining marker, not a platform-level error) + // should be reported as a failed activity result, NOT trigger the draining retry path. + var exception = new Exception("Function 'TestActivity' failed with an unhandled exception."); + + (OutOfProcMiddleware middleware, DispatchMiddlewareContext dispatchContext) = this.SetupActivityTest(exception); + + // Act: should NOT throw — the activity failure flows through to a failed ActivityExecutionResult. + await middleware.CallActivityAsync(dispatchContext, () => Task.CompletedTask); + + // Assert: the middleware should have set an activity result on the dispatch context. + ActivityExecutionResult result = dispatchContext.GetProperty(); + Assert.NotNull(result); + } + + [Fact] + [Trait("Category", PlatformSpecificHelpers.TestCategory)] + public async Task CallOrchestratorAsync_WorkerDrainingResponseWithFailedResult_ThrowsSessionAbortedException() + { + // Arrange: the orchestration failed AND the worker flagged the response as produced while + // draining (shutting down). The host should abort and requeue so it is retried on a healthy + // worker rather than recording the failure. + var response = new P.OrchestratorResponse + { + InstanceId = "test-instance-id", + IsWorkerDraining = true, + }; + + (OutOfProcMiddleware middleware, DispatchMiddlewareContext dispatchContext) = + this.SetupOrchestratorTestWithResponse(EncodeProtobuf(response), succeeded: false); + + await Assert.ThrowsAsync( + () => middleware.CallOrchestratorAsync(dispatchContext, () => Task.CompletedTask)); + } + + [Fact] + [Trait("Category", PlatformSpecificHelpers.TestCategory)] + public async Task CallOrchestratorAsync_WorkerDrainingResponseWithSucceededResult_DoesNotThrowSessionAbortedException() + { + // Arrange: the worker flagged draining but the orchestration still completed successfully. + // The draining retry only applies to failed results, so a successful result should be committed. + var response = new P.OrchestratorResponse + { + InstanceId = "test-instance-id", + IsWorkerDraining = true, + }; + + (OutOfProcMiddleware middleware, DispatchMiddlewareContext dispatchContext) = + this.SetupOrchestratorTestWithResponse(EncodeProtobuf(response), succeeded: true); + + // Act: should NOT throw — a successful result is committed even when the draining flag is set. + await middleware.CallOrchestratorAsync(dispatchContext, () => Task.CompletedTask); + + // Assert: the middleware should have set an orchestrator result on the dispatch context. + OrchestratorExecutionResult result = dispatchContext.GetProperty(); + Assert.NotNull(result); + } + + [Fact] + [Trait("Category", PlatformSpecificHelpers.TestCategory)] + public async Task CallOrchestratorAsync_NotDrainingResponse_DoesNotThrowSessionAbortedException() + { + // Arrange: a normal (non-draining) orchestrator response should be committed, not aborted. + var response = new P.OrchestratorResponse + { + InstanceId = "test-instance-id", + IsWorkerDraining = false, + }; + + (OutOfProcMiddleware middleware, DispatchMiddlewareContext dispatchContext) = + this.SetupOrchestratorTestWithResponse(EncodeProtobuf(response)); + + // Act: should NOT throw — the orchestration result flows through to the dispatch pipeline. + await middleware.CallOrchestratorAsync(dispatchContext, () => Task.CompletedTask); + + // Assert: the middleware should have set an orchestrator result on the dispatch context. + OrchestratorExecutionResult result = dispatchContext.GetProperty(); + Assert.NotNull(result); + } + + [Fact] + [Trait("Category", PlatformSpecificHelpers.TestCategory)] + public async Task CallEntityAsync_WorkerDrainingResponseWithFailedResult_ThrowsSessionAbortedException() + { + // Arrange: the entity batch failed AND the worker flagged the response as produced while + // draining (shutting down). The host should abort and requeue so it is retried on a healthy + // worker rather than recording the failure. + var response = new P.EntityBatchResult + { + CompletionToken = "test-completion-token", + IsWorkerDraining = true, + }; + + (OutOfProcMiddleware middleware, DispatchMiddlewareContext dispatchContext) = + this.SetupEntityTestWithResponse(EncodeProtobuf(response), succeeded: false); + + await Assert.ThrowsAsync( + () => middleware.CallEntityAsync(dispatchContext, () => Task.CompletedTask)); + } + + [Fact] + [Trait("Category", PlatformSpecificHelpers.TestCategory)] + public async Task CallEntityAsync_WorkerDrainingResponseWithSucceededResult_DoesNotThrowSessionAbortedException() + { + // Arrange: the worker flagged draining but the entity batch still completed successfully. + // The draining retry only applies to failed results, so a successful result should be committed. + var response = new P.EntityBatchResult + { + CompletionToken = "test-completion-token", + IsWorkerDraining = true, + }; + + (OutOfProcMiddleware middleware, DispatchMiddlewareContext dispatchContext) = + this.SetupEntityTestWithResponse(EncodeProtobuf(response), succeeded: true); + + // Act: should NOT throw — a successful result is committed even when the draining flag is set. + await middleware.CallEntityAsync(dispatchContext, () => Task.CompletedTask); + + // Assert: the middleware should have set an entity batch result on the dispatch context. + EntityBatchResult result = dispatchContext.GetProperty(); + Assert.NotNull(result); + } + + [Fact] + [Trait("Category", PlatformSpecificHelpers.TestCategory)] + public async Task CallEntityAsync_NotDrainingResponse_DoesNotThrowSessionAbortedException() + { + // Arrange: a normal (non-draining) entity batch result should be committed, not aborted. + var response = new P.EntityBatchResult + { + CompletionToken = "test-completion-token", + IsWorkerDraining = false, + }; + + (OutOfProcMiddleware middleware, DispatchMiddlewareContext dispatchContext) = + this.SetupEntityTestWithResponse(EncodeProtobuf(response)); + + // Act: should NOT throw — the entity result flows through to the dispatch pipeline. + await middleware.CallEntityAsync(dispatchContext, () => Task.CompletedTask); + + // Assert: the middleware should have set an entity batch result on the dispatch context. + EntityBatchResult result = dispatchContext.GetProperty(); + Assert.NotNull(result); + } + public static IEnumerable PlatformLevelExceptions() { // FunctionTimeoutException (top-level) @@ -215,6 +400,37 @@ public static IEnumerable PlatformLevelExceptions() return (middleware, dispatchContext); } + private (OutOfProcMiddleware middleware, DispatchMiddlewareContext context) SetupOrchestratorTestWithResponse(string base64Response, bool succeeded = true) + { + (OutOfProcMiddleware middleware, DispatchMiddlewareContext dispatchContext) + = this.CreateMiddlewareWithInvocation(base64Response, "TestOrchestrator", FunctionType.Orchestrator, succeeded); + + var orchestrationState = new OrchestrationRuntimeState( + [ + new ExecutionStartedEvent(-1, null) { Name = "TestOrchestrator" }, + ]); + + dispatchContext.SetProperty(orchestrationState); + dispatchContext.SetProperty(new OrchestrationInstance { InstanceId = "test-instance-id" }); + + return (middleware, dispatchContext); + } + + private (OutOfProcMiddleware middleware, DispatchMiddlewareContext context) SetupEntityTestWithResponse(string base64Response, bool succeeded = true) + { + (OutOfProcMiddleware middleware, DispatchMiddlewareContext dispatchContext) + = this.CreateMiddlewareWithInvocation(base64Response, "TestEntity", FunctionType.Entity, succeeded); + + dispatchContext.SetProperty(new EntityBatchRequest + { + InstanceId = "@TestEntity@test-key", + EntityState = null, + Operations = new List(), + }); + + return (middleware, dispatchContext); + } + private (OutOfProcMiddleware middleware, DispatchMiddlewareContext context) CreateMiddleware( Exception executorException, string functionName, FunctionType functionType) { @@ -225,6 +441,40 @@ public static IEnumerable PlatformLevelExceptions() .Setup(e => e.TryExecuteAsync(It.IsAny(), It.IsAny())) .ReturnsAsync(new FunctionResult(false, executorException)); + return (new OutOfProcMiddleware(RegisterAndGetExtension(extension, functionName, functionType, mockExecutor)), CreateDispatchContext(functionType)); + } + + // Creates a middleware that drives the InvokeHandler with the given base64-encoded protobuf + // response, exercising the host's response-parsing path (including the IsWorkerDraining flag read) + // the same way a real out-of-proc worker invocation would. The succeeded flag controls whether the + // resulting FunctionResult reports success or failure, since the worker-draining retry only triggers + // when the function result is failed. + private (OutOfProcMiddleware middleware, DispatchMiddlewareContext context) CreateMiddlewareWithInvocation( + string base64Response, string functionName, FunctionType functionType, bool succeeded = true) + { + DurableTaskExtension extension = CreateDurableTaskExtension(); + + var mockExecutor = new Mock(); + mockExecutor + .Setup(e => e.TryExecuteAsync(It.IsAny(), It.IsAny())) + .Returns(async (data, ct) => + { + // The middleware supplies an InvokeHandler that expects the user-code invoker to return + // a Task whose result is the base64 protobuf response string. +#pragma warning disable CS0618 // Type or member is obsolete (approved for use by this extension) + await data.InvokeHandler(() => Task.FromResult(base64Response)); +#pragma warning restore CS0618 + return succeeded + ? new FunctionResult(true) + : new FunctionResult(false, new Exception("Function invocation failed.")); + }); + + return (new OutOfProcMiddleware(RegisterAndGetExtension(extension, functionName, functionType, mockExecutor)), CreateDispatchContext(functionType)); + } + + private static DurableTaskExtension RegisterAndGetExtension( + DurableTaskExtension extension, string functionName, FunctionType functionType, Mock mockExecutor) + { var name = new FunctionName(functionName); switch (functionType) @@ -240,6 +490,11 @@ public static IEnumerable PlatformLevelExceptions() break; } + return extension; + } + + private static DispatchMiddlewareContext CreateDispatchContext(FunctionType functionType) + { var dispatchContext = new DispatchMiddlewareContext(); // Orchestrators and entities require WorkItemMetadata; activities do not. @@ -248,7 +503,12 @@ public static IEnumerable PlatformLevelExceptions() dispatchContext.SetProperty(CreateWorkItemMetadata(isExtendedSession: false, includeState: false)); } - return (new OutOfProcMiddleware(extension), dispatchContext); + return dispatchContext; + } + + private static string EncodeProtobuf(IMessage message) + { + return Convert.ToBase64String(message.ToByteArray()); } private static DurableTaskExtension CreateDurableTaskExtension() @@ -279,7 +539,7 @@ private static DurableTaskExtension CreateDurableTaskExtension() private static WorkItemMetadata CreateWorkItemMetadata(bool isExtendedSession, bool includeState) { // WorkItemMetadata has an internal constructor, so we use reflection to create it. - ConstructorInfo ctor = typeof(WorkItemMetadata).GetConstructor( + ConstructorInfo? ctor = typeof(WorkItemMetadata).GetConstructor( BindingFlags.Instance | BindingFlags.NonPublic, binder: null, [typeof(bool), typeof(bool)],