Add worker drain signal#3454
Conversation
There was a problem hiding this comment.
Pull request overview
This PR adds a “worker draining” signal intended to prevent Durable executions from being permanently marked as failed when the underlying Functions worker is shutting down, by aborting the current dispatch so the work item can be retried on a healthy worker.
Changes:
- Worker-side: detect draining via
IHostApplicationLifetime.ApplicationStoppingand either (a) throw a sentinelWorkerDrainingExceptionfor activities or (b) stampIsWorkerDrainingonto orchestrator/entity protobuf responses. - Host-side: update
OutOfProcMiddlewareto treat worker-draining activity failures (via sentinel marker) and orchestrator/entity failures (viaIsWorkerDrainingresponse flag) as retryable by throwingSessionAbortedException. - Tests: expand
OutOfProcMiddlewareTeststo cover the new draining behaviors for activity/orchestrator/entity paths.
Reviewed changes
Copilot reviewed 7 out of 7 changed files in this pull request and generated 3 comments.
Show a summary per file
| File | Description |
|---|---|
| test/FunctionsV2/Tests/Unit/OutOfProcMiddlewareTests.cs | Adds unit coverage for the new worker-draining retry behaviors. |
| src/WebJobs.Extensions.DurableTask/OutOfProcMiddleware.cs | Detects worker-draining conditions and forces durable retries instead of committing failures. |
| src/Worker.Extensions.DurableTask/Execution/DurableFunctionExecutor.cs | Introduces worker-draining detection via IHostApplicationLifetime. |
| src/Worker.Extensions.DurableTask/Execution/DurableFunctionExecutor.Activity.cs | Throws a sentinel exception after activity execution when draining. |
| src/Worker.Extensions.DurableTask/Execution/DurableFunctionExecutor.Orchestration.cs | Stamps IsWorkerDraining into the orchestrator response when draining. |
| src/Worker.Extensions.DurableTask/Execution/DurableFunctionExecutor.Entity.cs | Stamps IsWorkerDraining into the entity batch result when draining. |
| src/Worker.Extensions.DurableTask/Exceptions/WorkerDrainingException.cs | Adds a sentinel exception + stable marker string for host-side detection. |
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
| byte[] triggerReturnValueBytes = Convert.FromBase64String(triggerReturnValue); | ||
| P.OrchestratorResponse response = P.OrchestratorResponse.Parser.ParseFrom(triggerReturnValueBytes); | ||
|
|
||
| workerRequiresHistory = response.RequiresHistory; | ||
| isWorkerDraining = response.IsWorkerDraining; | ||
|
|
| 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); |
| [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); |
| // 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."); |
There was a problem hiding this comment.
We shouldn't be throwing System.Exception as a general practice. Why not SessionAbortedException like we do elsewhere?
| // 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."); |
There was a problem hiding this comment.
Same comment about the type of exception we throw.
| this.Options.HubName, | ||
| functionName.Name, | ||
| instance.InstanceId, | ||
| "The worker is shutting down. The activity execution will be aborted and retried.", |
There was a problem hiding this comment.
Consider saying "out-of-process worker" instead of just "worker" in all of these messages to disambiguate from the durable task worker.
| 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}") |
There was a problem hiding this comment.
nit: it probably makes sense to put the sentinel in the beginning of the message. That also allows us to do a more efficient StartsWith string comparison instead of a Contains.
| await this.RunDirectActivityAsync(context, triggerBinding); | ||
| return; | ||
| } | ||
| else |
There was a problem hiding this comment.
What is this else branch change for? It seems to be adding unnecessary complexity to the diff and the code overall.
| private DataConverter Converter => options.Value.DataConverter; | ||
|
|
||
| private bool IsWorkerDraining => | ||
| hostApplicationLifetime?.ApplicationStopping.IsCancellationRequested ?? false; |
There was a problem hiding this comment.
The DurableFunctionsExecutor runs really late in the execution pipeline so it will only work in cases where execution actually makes it here. If I understand correctly, the person who opened the original issue observed an exception that happened somewhere else in the pipeline, suggesting that cancellation-related issues can occur before or after we get to this middleware. So, it seems that this will catch some cases but not all. A more robust approach would be to add this kind of logic directly in the .NET Isolated worker code rather than trying to put it in the Durable Functions worker extension code.
Summary
What changed?
This PR introduces a new worker drain signal in the case of Function execution - if this signal is set and the Function failed, we will retry the execution rather than permanently failing it.
Why is this change needed?
We have this signal for host shutdown but this is decoupled from worker shutdown.
Issues / work items
Project checklist
pending_docs.mdrelease_notes.mdWebJobs.Extensions.DurableTaskpackage/src/Worker.Extensions.DurableTask/AssemblyInfo.csEventSourcelogsv2.xbranchWebJobs.Extensions.DurableTaskv3.x and will be retained only in thedevandmainbranchesAI-assisted code disclosure (required)
Was an AI tool used? (select one)
If AI was used:
AI verification (required if AI was used):
Testing
Automated tests
Manual validation (only if runtime/behavior changed)
1.
2.
3.
Notes for reviewers