From fa48d63135693641f5d56ad49a47199e23299c96 Mon Sep 17 00:00:00 2001 From: wangbill Date: Tue, 25 Aug 2026 14:42:26 -0400 Subject: [PATCH 1/3] Release abandoned orchestrator tasks when an executor is retired TaskOrchestrationContext creates a TaskCompletionSource per activity, sub-orchestration and timer, and the orchestrator awaits it. When an episode ends while the orchestrator is parked on one of those awaits, the TCS is deliberately abandoned in a pending state -- that is inherent to the replay model. When a debugger is attached, TaskAwaiter.OnCompletedInternal calls OutputWaitEtwEvents, which calls Task.AddToActiveTasks because Task.s_asyncDebuggingEnabled is true. That inserts the task into the process-wide, strongly-referenced static dictionary Task.s_currentActiveTasks. The matching Task.RemoveFromActiveTasks only runs inside the awaiter continuation wrapper, i.e. only when the awaited task completes. Abandoned tasks are therefore rooted forever, and each rooted Task pins its continuation -> async state machine -> TaskOrchestrationContext -> history/inputs/outputs. Only the AddToActiveTasks branch roots the task, which is why the leak is debugger-specific. Fix: cancel the open tasks once the executor is guaranteed never to run again, so the abandoned awaiter continuations run and unregister themselves. - TaskOrchestrationContext.ReleaseOpenTasks() cancels every open task. It is idempotent, sets isReleased before cancelling, and snapshots openTasks before iterating because resumed user code can mutate the dictionary. A ThrowIfReleased guard on ScheduleTaskInternal, CreateSubOrchestrationInstanceCore and CreateTimer stops orchestrator code that swallows the cancellation from scheduling new work. - TaskOrchestrationExecutor implements IDisposable. Dispose() restores the orchestrator ambient environment, releases the open tasks, and observes a faulted result so nothing surfaces as an UnobservedTaskException. - TaskOrchestrationDispatcher owns executor lifetime via workItem.Cursor, so it retires the executor from a new finally on OnProcessWorkItemSessionAsync and at the continue-as-new site. Extended sessions are unaffected: release only happens when the executor is retired, never between episodes. ExtendedSession_OpenTasksSurviveBetweenEpisodes covers this. Ref https://github.com/Azure/azure-functions-durable-extension/issues/340 Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- .../TaskOrchestrationExecutorTests.cs | 344 ++++++++++++++++++ .../TaskOrchestrationContext.cs | 77 ++++ .../TaskOrchestrationDispatcher.cs | 22 +- .../TaskOrchestrationExecutor.cs | 48 ++- 4 files changed, 489 insertions(+), 2 deletions(-) create mode 100644 Test/DurableTask.Core.Tests/TaskOrchestrationExecutorTests.cs diff --git a/Test/DurableTask.Core.Tests/TaskOrchestrationExecutorTests.cs b/Test/DurableTask.Core.Tests/TaskOrchestrationExecutorTests.cs new file mode 100644 index 000000000..b0e503053 --- /dev/null +++ b/Test/DurableTask.Core.Tests/TaskOrchestrationExecutorTests.cs @@ -0,0 +1,344 @@ +// ---------------------------------------------------------------------------------- +// Copyright Microsoft Corporation +// 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. +// ---------------------------------------------------------------------------------- + +namespace DurableTask.Core.Tests +{ + using System; + using System.Collections.Generic; + using System.Linq; + using System.Reflection; + using System.Threading.Tasks; + using DurableTask.Core.Command; + using DurableTask.Core.History; + using DurableTask.Core.Serializing; + using Microsoft.VisualStudio.TestTools.UnitTesting; + + /// + /// Tests for lifetime management. + /// + /// + /// An orchestrator parks on a for every activity, + /// sub-orchestration, and timer it is waiting on, and those tasks are abandoned in a pending state when + /// the episode ends. When a debugger is attached, the CLR keeps every awaited task in the process-wide + /// Task.s_currentActiveTasks dictionary until the task completes, so abandoned awaits permanently + /// root the orchestration object graph. Disposing the executor cancels the open tasks, which lets those + /// awaiter continuations run and unregister themselves. + /// Regression coverage for https://github.com/Azure/azure-functions-durable-extension/issues/340. + /// + [TestClass] + public class TaskOrchestrationExecutorTests + { + const string ActivityName = "SayHello"; + + [TestMethod] + public void Dispose_ResumesAbandonedOrchestratorContinuations() + { + var orchestration = new FanOutOrchestration(fanOut: 3); + using (var executor = CreateExecutor(orchestration)) + { + executor.Execute(); + + Assert.AreEqual(3, orchestration.StartedTaskCount, "The orchestrator should have scheduled 3 activities."); + Assert.AreEqual(0, orchestration.ReleasedTaskCount, "Abandoned awaits should still be pending at the end of the episode."); + } + + Assert.AreEqual( + 3, + orchestration.ReleasedTaskCount, + "Disposing the executor should resume every abandoned await so its continuation can unregister itself."); + } + + [TestMethod] + public void Dispose_WithAsyncDebuggingEnabled_DoesNotLeakActiveTasks() + { + // This is the actual bug: with a debugger attached, every abandoned await stays in + // Task.s_currentActiveTasks forever, which roots the entire orchestration object graph. + // + // Task.s_currentActiveTasks is process-wide, so the test host's own async plumbing shows up in + // it as well. Rather than trying to subtract that noise, this measures how the growth scales + // with the number of abandoned awaits: a leak is proportional to the fan-out, while unrelated + // noise is not. + const int Episodes = 20; + const int SmallFanOut = 1; + const int LargeFanOut = 25; + + using (AsyncDebuggingScope.Enable()) + { + // Warm up so that one-time allocations aren't counted against the measurement. + RunEpisodes(Episodes, SmallFanOut, dispose: true); + + int leakySensitivity = MeasureFanOutSensitivity(Episodes, SmallFanOut, LargeFanOut, dispose: false); + Assert.IsTrue( + leakySensitivity > Episodes * (LargeFanOut - SmallFanOut), + "Undisposed executors are expected to leak one entry per abandoned await; if they no longer " + + $"do, this test can no longer detect the regression. Measured {leakySensitivity}."); + + int fixedSensitivity = MeasureFanOutSensitivity(Episodes, SmallFanOut, LargeFanOut, dispose: true); + Assert.IsTrue( + fixedSensitivity * 20 < leakySensitivity, + "Disposing the executor must stop Task.s_currentActiveTasks from growing with the number of " + + $"abandoned awaits, but growth was still {fixedSensitivity} against {leakySensitivity} when " + + "the executors were left undisposed."); + } + } + + /// + /// Returns how much more the active task table grows for abandoned + /// awaits per episode than it does for . Constant per-episode overhead + /// and unrelated test host activity cancel out, leaving only growth caused by abandoned awaits. + /// + static int MeasureFanOutSensitivity(int episodes, int smallFanOut, int largeFanOut, bool dispose) + { + int small = RunEpisodes(episodes, smallFanOut, dispose); + int large = RunEpisodes(episodes, largeFanOut, dispose); + return large - small; + } + + static int RunEpisodes(int episodes, int fanOut, bool dispose) + { + int before = AsyncDebuggingScope.ActiveTaskCount; + for (int i = 0; i < episodes; i++) + { + TaskOrchestrationExecutor executor = CreateExecutor(new FanOutOrchestration(fanOut)); + executor.Execute(); + if (dispose) + { + executor.Dispose(); + } + } + + return AsyncDebuggingScope.ActiveTaskCount - before; + } + + [TestMethod] + public void Dispose_DoesNotChangeTheDecisionsAlreadyProduced() + { + var orchestration = new FanOutOrchestration(fanOut: 3); + using (var executor = CreateExecutor(orchestration)) + { + OrchestratorExecutionResult result = executor.Execute(); + List before = result.Actions.ToList(); + + executor.Dispose(); + + CollectionAssert.AreEqual( + before, + result.Actions.ToList(), + "Releasing abandoned tasks must not add or remove orchestrator actions."); + } + } + + [TestMethod] + public void Dispose_OrchestratorThatSwallowsCancellation_CannotScheduleMoreWork() + { + var orchestration = new SwallowsCancellationOrchestration(); + using (var executor = CreateExecutor(orchestration)) + { + executor.Execute(); + } + + Assert.IsInstanceOfType( + orchestration.RescheduleFailure, + typeof(OperationCanceledException), + "A released context must refuse to open new tasks, otherwise resumed orchestrator code can leak again."); + } + + [TestMethod] + public void Dispose_IsIdempotent() + { + var orchestration = new FanOutOrchestration(fanOut: 2); + var executor = CreateExecutor(orchestration); + executor.Execute(); + + executor.Dispose(); + executor.Dispose(); + + Assert.AreEqual(2, orchestration.ReleasedTaskCount, "Repeated disposal should not resume continuations more than once."); + } + + [TestMethod] + public void ExtendedSession_OpenTasksSurviveBetweenEpisodes() + { + // Extended sessions reuse the executor across episodes, so open tasks must stay pending + // until their results arrive. Only the end of the session may release them. + var orchestration = new FanOutOrchestration(fanOut: 1); + OrchestrationRuntimeState runtimeState = CreateRuntimeState(); + + using (var executor = new TaskOrchestrationExecutor(runtimeState, orchestration, BehaviorOnContinueAsNew.Carryover)) + { + OrchestratorExecutionResult firstEpisode = executor.Execute(); + Assert.AreEqual(1, firstEpisode.Actions.Count(), "The first episode should schedule the activity."); + Assert.IsFalse(executor.IsCompleted); + + runtimeState.NewEvents.Clear(); + runtimeState.AddEvent(new OrchestratorStartedEvent(-1)); + runtimeState.AddEvent(new TaskCompletedEvent(-1, taskScheduledId: 0, result: JsonDataConverter.Default.Serialize("Hello"))); + + OrchestratorExecutionResult secondEpisode = executor.ExecuteNewEvents(); + + Assert.IsTrue(executor.IsCompleted, "The activity result should have been delivered to the still-open task."); + Assert.AreEqual(1, orchestration.ReleasedTaskCount, "The await should have been resumed by its result, not by cancellation."); + Assert.IsTrue( + secondEpisode.Actions.OfType().Any(), + "The orchestration should have completed on the second episode."); + } + } + + static TaskOrchestrationExecutor CreateExecutor(TaskOrchestration orchestration) => + new TaskOrchestrationExecutor(CreateRuntimeState(), orchestration, BehaviorOnContinueAsNew.Carryover); + + static OrchestrationRuntimeState CreateRuntimeState() + { + var runtimeState = new OrchestrationRuntimeState(); + runtimeState.AddEvent(new OrchestratorStartedEvent(-1)); + runtimeState.AddEvent(new ExecutionStartedEvent(-1, null) + { + OrchestrationInstance = new OrchestrationInstance + { + InstanceId = Guid.NewGuid().ToString("N"), + ExecutionId = Guid.NewGuid().ToString("N"), + }, + Name = "TestOrchestration", + Version = string.Empty, + }); + + return runtimeState; + } + + /// + /// Fans out to several activities and then parks, which is the state an orchestrator is in at the + /// end of a typical episode. Each await records whether it was ever resumed. + /// + class FanOutOrchestration : TaskOrchestration + { + readonly int fanOut; + + public FanOutOrchestration(int fanOut) + { + this.fanOut = fanOut; + } + + public int StartedTaskCount { get; private set; } + + public int ReleasedTaskCount { get; private set; } + + public override async Task RunTask(OrchestrationContext context, string input) + { + var tasks = new List>(this.fanOut); + for (int i = 0; i < this.fanOut; i++) + { + this.StartedTaskCount++; + tasks.Add(this.AwaitActivityAsync(context)); + } + + await Task.WhenAll(tasks); + return string.Empty; + } + + async Task AwaitActivityAsync(OrchestrationContext context) + { + try + { + return await context.ScheduleTask(ActivityName, string.Empty); + } + finally + { + this.ReleasedTaskCount++; + } + } + } + + /// + /// Mimics orchestrator code with a catch-all handler: it swallows the cancellation raised while the + /// executor is being released and then tries to schedule more work. + /// + class SwallowsCancellationOrchestration : TaskOrchestration + { + public Exception RescheduleFailure { get; private set; } + + public override async Task RunTask(OrchestrationContext context, string input) + { + try + { + return await context.ScheduleTask(ActivityName, string.Empty); + } + catch (Exception) + { + try + { + return await context.ScheduleTask(ActivityName, string.Empty); + } + catch (Exception e) + { + this.RescheduleFailure = e; + throw; + } + } + } + } + + /// + /// Turns on the CLR's async debugging bookkeeping for the duration of a test, which is what a + /// attached debugger does, and exposes the size of the tracking dictionary. + /// + sealed class AsyncDebuggingScope : IDisposable + { + const BindingFlags StaticFlags = BindingFlags.Static | BindingFlags.NonPublic | BindingFlags.Public; + + static readonly FieldInfo EnabledField = typeof(Task).GetField("s_asyncDebuggingEnabled", StaticFlags); + static readonly FieldInfo ActiveTasksField = typeof(Task).GetField("s_currentActiveTasks", StaticFlags); + + readonly bool previousValue; + + AsyncDebuggingScope(bool previousValue) + { + this.previousValue = previousValue; + } + + public static AsyncDebuggingScope Enable() + { + if (EnabledField == null || ActiveTasksField == null) + { + Assert.Inconclusive("This runtime does not expose the async debugging state that this test relies on."); + } + + var scope = new AsyncDebuggingScope((bool)EnabledField.GetValue(null)); + EnabledField.SetValue(null, true); + return scope; + } + + /// + /// The size of the process-wide Task.s_currentActiveTasks table. + /// + public static int ActiveTaskCount + { + get + { + object activeTasks = ActiveTasksField.GetValue(null); + if (activeTasks == null) + { + return 0; + } + + PropertyInfo count = activeTasks.GetType().GetProperty("Count"); + lock (activeTasks) + { + return (int)count.GetValue(activeTasks); + } + } + } + + public void Dispose() => EnabledField.SetValue(null, this.previousValue); + } + } +} diff --git a/src/DurableTask.Core/TaskOrchestrationContext.cs b/src/DurableTask.Core/TaskOrchestrationContext.cs index e4846124c..cbb9bffd2 100644 --- a/src/DurableTask.Core/TaskOrchestrationContext.cs +++ b/src/DurableTask.Core/TaskOrchestrationContext.cs @@ -35,6 +35,7 @@ internal class TaskOrchestrationContext : OrchestrationContext private OrchestrationCompleteOrchestratorAction continueAsNew; private static readonly ContinueAsNewOptions DefaultContinueAsNewOptions = new ContinueAsNewOptions(); private bool executionCompletedOrTerminated; + private bool isReleased; private int idCounter; private readonly Queue eventsWhileSuspended; private readonly IDictionary suspendedActionsMap; @@ -75,6 +76,76 @@ public TaskOrchestrationContext( public bool HasOpenTasks => this.openTasks.Count > 0; + /// + /// Cancels every open task so that the orchestrator's abandoned await continuations are released. + /// + /// + /// + /// Orchestrator code parks on a for every activity, + /// sub-orchestration, and timer it is waiting on. When an episode ends, those tasks are simply + /// abandoned in a pending state, because their results are not yet known. That is harmless to the + /// garbage collector on its own, but when a debugger is attached the CLR records every awaited task + /// in the process-wide Task.s_currentActiveTasks dictionary and only removes the entry when + /// the awaited task completes. Abandoned tasks therefore stay rooted forever, and with them the + /// entire orchestration object graph (context, history, inputs, and outputs). + /// + /// + /// Cancelling the open tasks lets each awaiter continuation run and unregister itself, which is what + /// allows the graph to be collected. This is only safe once the executor is guaranteed never to be + /// used again, since a cancelled task can no longer receive a result on a subsequent episode. + /// + /// + internal void ReleaseOpenTasks() + { + if (this.isReleased) + { + return; + } + + // Set this before cancelling anything: cancellation resumes orchestrator code, and that code must + // not be able to open new tasks that would leak in exactly the same way. + this.isReleased = true; + + if (this.openTasks.Count == 0) + { + return; + } + + // Resumed orchestrator code can mutate openTasks (for example via a timer cancellation callback), + // so cancel from a snapshot rather than while enumerating the live dictionary. + List abandonedTasks = this.openTasks.Values.ToList(); + this.openTasks.Clear(); + + foreach (OpenTaskInfo info in abandonedTasks) + { + try + { + info.Result.TrySetCanceled(); + } + catch (Exception e) when (!Utils.IsFatal(e)) + { + // Orchestrator code observed the cancellation and threw. The episode is already over and + // its decisions have already been captured, so there is nothing to report. Swallow the + // exception and keep going so the remaining tasks still get released. + TraceHelper.TraceSession( + TraceEventType.Warning, + "TaskOrchestrationContext-ReleaseOpenTasks", + OrchestrationInstance?.InstanceId, + "Exception while releasing abandoned orchestrator tasks: {0}", + e); + } + } + } + + private void ThrowIfReleased() + { + if (this.isReleased) + { + throw new OperationCanceledException( + "This orchestration episode has ended and the orchestration context can no longer schedule work."); + } + } + internal void ClearPendingActions() { this.orchestratorActionsMap.Clear(); @@ -126,6 +197,8 @@ public async Task ScheduleTaskToWorker(string name, string ver public async Task ScheduleTaskInternal(string name, string version, string taskList, Type resultType, ScheduleTaskOptions options, params object[] parameters) { + ThrowIfReleased(); + int id = this.idCounter++; string serializedInput = this.MessageDataConverter.SerializeInternal(parameters); var scheduleTaskTaskAction = new ScheduleTaskOrchestratorAction @@ -189,6 +262,8 @@ async Task CreateSubOrchestrationInstanceCore( object input, IDictionary tags) { + ThrowIfReleased(); + int id = this.idCounter++; string serializedInput = this.MessageDataConverter.SerializeInternal(input); @@ -295,6 +370,8 @@ public override async Task CreateTimer(DateTime fireAt, T state, Cancellat paramName: nameof(state)); } + ThrowIfReleased(); + int id = this.idCounter++; var createTimerOrchestratorAction = new CreateTimerOrchestratorAction { diff --git a/src/DurableTask.Core/TaskOrchestrationDispatcher.cs b/src/DurableTask.Core/TaskOrchestrationDispatcher.cs index 649e7b47a..95474a122 100644 --- a/src/DurableTask.Core/TaskOrchestrationDispatcher.cs +++ b/src/DurableTask.Core/TaskOrchestrationDispatcher.cs @@ -308,6 +308,24 @@ async Task OnProcessWorkItemSessionAsync(TaskOrchestrationWorkItem workItem) TraceHelper.TraceInstance(TraceEventType.Warning, "TaskOrchestrationDispatcher-ExecutionAborted", instance, "{0}", e.Message); await this.orchestrationService.AbandonTaskOrchestrationWorkItemAsync(workItem); } + finally + { + // The session is over and the executor will never run again, so release the orchestrator + // continuations it abandoned while waiting on activities, sub-orchestrations, and timers. + // Leaving them pending leaks the whole orchestration object graph when a debugger is + // attached. See https://github.com/Azure/azure-functions-durable-extension/issues/340. + ReleaseCursor(ref workItem.Cursor); + } + } + + /// + /// Retires the executor held by and clears the reference. + /// + static void ReleaseCursor(ref OrchestrationExecutionCursor? cursor) + { + OrchestrationExecutionCursor? retiredCursor = cursor; + cursor = null; + retiredCursor?.OrchestrationExecutor?.Dispose(); } /// @@ -669,7 +687,9 @@ protected async Task OnProcessWorkItemAsync(TaskOrchestrationWorkItem work runtimeState.AddEvent(new OrchestratorCompletedEvent(-1)); workItem.OrchestrationRuntimeState = runtimeState; - workItem.Cursor = null; + // The continued-as-new execution gets a brand new executor, so retire this + // one instead of just dropping the reference. + ReleaseCursor(ref workItem.Cursor); traceActivity = RestartTraceActivityForContinueAsNewIfNeeded( traceActivity, diff --git a/src/DurableTask.Core/TaskOrchestrationExecutor.cs b/src/DurableTask.Core/TaskOrchestrationExecutor.cs index 540851e50..0f4f6545a 100644 --- a/src/DurableTask.Core/TaskOrchestrationExecutor.cs +++ b/src/DurableTask.Core/TaskOrchestrationExecutor.cs @@ -28,7 +28,7 @@ namespace DurableTask.Core /// /// Utility for executing task orchestrators. /// - public class TaskOrchestrationExecutor + public class TaskOrchestrationExecutor : IDisposable { readonly TaskOrchestrationContext context; readonly TaskScheduler decisionScheduler; @@ -135,6 +135,52 @@ public OrchestratorExecutionResult ExecuteNewEvents() newEvents: this.orchestrationRuntimeState.NewEvents); } + /// + /// Releases the orchestrator continuations that this executor abandoned while waiting on + /// activities, sub-orchestrations, or timers. + /// + /// + /// + /// Call this once the executor is guaranteed never to run again, i.e. when the orchestration + /// session ends. It must not be called between episodes of an extended session, because an open + /// task still needs to be able to receive its result on a later episode. + /// + /// + /// Skipping this call is not a correctness problem, but it leaks memory whenever a debugger is + /// attached: the CLR keeps every awaited task in a process-wide dictionary until that task + /// completes, so abandoned orchestrator awaits permanently root the orchestration object graph. + /// See https://github.com/Azure/azure-functions-durable-extension/issues/340. + /// + /// + public void Dispose() + { + SynchronizationContext prevCtx = SynchronizationContext.Current; + bool prevIsOrchestratorThread = OrchestrationContext.IsOrchestratorThread; + + try + { + // Cancelling the open tasks resumes orchestrator code, so give it the same ambient + // environment it sees during a normal episode. + SynchronizationContext.SetSynchronizationContext( + new TaskOrchestrationSynchronizationContext(this.decisionScheduler)); + OrchestrationContext.IsOrchestratorThread = true; + + this.context.ReleaseOpenTasks(); + } + finally + { + SynchronizationContext.SetSynchronizationContext(prevCtx); + OrchestrationContext.IsOrchestratorThread = prevIsOrchestratorThread; + + // Unwinding may have faulted the orchestrator's top-level task. Nothing observes it at this + // point, so observe it here to keep it from surfacing as an unobserved task exception. + if (this.result?.IsFaulted == true) + { + _ = this.result.Exception; + } + } + } + OrchestratorExecutionResult ExecuteCore(IEnumerable pastEvents, IEnumerable newEvents) { SynchronizationContext prevCtx = SynchronizationContext.Current; From da21249cf73ea21765560b712970cc2433594187 Mon Sep 17 00:00:00 2001 From: wangbill Date: Thu, 27 Aug 2026 17:44:27 -0400 Subject: [PATCH 2/3] Make the executor release hook internal instead of IDisposable TaskOrchestrationExecutor is a public type, so implementing IDisposable added public API surface that no caller outside DurableTask.Core can use correctly: executor lifetime is owned by TaskOrchestrationDispatcher, which is the only component that knows when an executor will never run again. It also risked tripping CA2000 in downstream repos that construct an executor directly. Replace the public Dispose() with an internal Release() and have ReleaseCursor call that instead. The cleanup behavior is unchanged: install the orchestrator ambient context, call context.ReleaseOpenTasks(), restore ambient state in a finally, and observe the top-level task exception so it cannot surface as an UnobservedTaskException. Cursor clearing and all lifecycle placement in the dispatcher are untouched. Tests call Release() explicitly rather than relying on a using block, and are renamed Dispose_* -> Release_*. All six behavioral guarantees are preserved, including ExtendedSession_OpenTasksSurviveBetweenEpisodes. The public surface of TaskOrchestrationExecutor is now exactly what it was before this PR: three constructors, Execute, ExecuteNewEvents, and IsCompleted. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> Copilot-Session: 019e7654-1037-4e02-b2fd-c03e09d6b7e2 --- .../TaskOrchestrationExecutorTests.cs | 114 +++++++++--------- .../TaskOrchestrationDispatcher.cs | 2 +- .../TaskOrchestrationExecutor.cs | 10 +- 3 files changed, 65 insertions(+), 61 deletions(-) diff --git a/Test/DurableTask.Core.Tests/TaskOrchestrationExecutorTests.cs b/Test/DurableTask.Core.Tests/TaskOrchestrationExecutorTests.cs index b0e503053..6d7403b10 100644 --- a/Test/DurableTask.Core.Tests/TaskOrchestrationExecutorTests.cs +++ b/Test/DurableTask.Core.Tests/TaskOrchestrationExecutorTests.cs @@ -31,7 +31,7 @@ namespace DurableTask.Core.Tests /// sub-orchestration, and timer it is waiting on, and those tasks are abandoned in a pending state when /// the episode ends. When a debugger is attached, the CLR keeps every awaited task in the process-wide /// Task.s_currentActiveTasks dictionary until the task completes, so abandoned awaits permanently - /// root the orchestration object graph. Disposing the executor cancels the open tasks, which lets those + /// root the orchestration object graph. Releasing the executor cancels the open tasks, which lets those /// awaiter continuations run and unregister themselves. /// Regression coverage for https://github.com/Azure/azure-functions-durable-extension/issues/340. /// @@ -41,25 +41,25 @@ public class TaskOrchestrationExecutorTests const string ActivityName = "SayHello"; [TestMethod] - public void Dispose_ResumesAbandonedOrchestratorContinuations() + public void Release_ResumesAbandonedOrchestratorContinuations() { var orchestration = new FanOutOrchestration(fanOut: 3); - using (var executor = CreateExecutor(orchestration)) - { - executor.Execute(); + TaskOrchestrationExecutor executor = CreateExecutor(orchestration); + executor.Execute(); - Assert.AreEqual(3, orchestration.StartedTaskCount, "The orchestrator should have scheduled 3 activities."); - Assert.AreEqual(0, orchestration.ReleasedTaskCount, "Abandoned awaits should still be pending at the end of the episode."); - } + Assert.AreEqual(3, orchestration.StartedTaskCount, "The orchestrator should have scheduled 3 activities."); + Assert.AreEqual(0, orchestration.ReleasedTaskCount, "Abandoned awaits should still be pending at the end of the episode."); + + executor.Release(); Assert.AreEqual( 3, orchestration.ReleasedTaskCount, - "Disposing the executor should resume every abandoned await so its continuation can unregister itself."); + "Releasing the executor should resume every abandoned await so its continuation can unregister itself."); } [TestMethod] - public void Dispose_WithAsyncDebuggingEnabled_DoesNotLeakActiveTasks() + public void Release_WithAsyncDebuggingEnabled_DoesNotLeakActiveTasks() { // This is the actual bug: with a debugger attached, every abandoned await stays in // Task.s_currentActiveTasks forever, which roots the entire orchestration object graph. @@ -75,20 +75,20 @@ public void Dispose_WithAsyncDebuggingEnabled_DoesNotLeakActiveTasks() using (AsyncDebuggingScope.Enable()) { // Warm up so that one-time allocations aren't counted against the measurement. - RunEpisodes(Episodes, SmallFanOut, dispose: true); + RunEpisodes(Episodes, SmallFanOut, release: true); - int leakySensitivity = MeasureFanOutSensitivity(Episodes, SmallFanOut, LargeFanOut, dispose: false); + int leakySensitivity = MeasureFanOutSensitivity(Episodes, SmallFanOut, LargeFanOut, release: false); Assert.IsTrue( leakySensitivity > Episodes * (LargeFanOut - SmallFanOut), - "Undisposed executors are expected to leak one entry per abandoned await; if they no longer " + + "Unreleased executors are expected to leak one entry per abandoned await; if they no longer " + $"do, this test can no longer detect the regression. Measured {leakySensitivity}."); - int fixedSensitivity = MeasureFanOutSensitivity(Episodes, SmallFanOut, LargeFanOut, dispose: true); + int fixedSensitivity = MeasureFanOutSensitivity(Episodes, SmallFanOut, LargeFanOut, release: true); Assert.IsTrue( fixedSensitivity * 20 < leakySensitivity, - "Disposing the executor must stop Task.s_currentActiveTasks from growing with the number of " + + "Releasing the executor must stop Task.s_currentActiveTasks from growing with the number of " + $"abandoned awaits, but growth was still {fixedSensitivity} against {leakySensitivity} when " + - "the executors were left undisposed."); + "the executors were left unreleased."); } } @@ -97,23 +97,23 @@ public void Dispose_WithAsyncDebuggingEnabled_DoesNotLeakActiveTasks() /// awaits per episode than it does for . Constant per-episode overhead /// and unrelated test host activity cancel out, leaving only growth caused by abandoned awaits. /// - static int MeasureFanOutSensitivity(int episodes, int smallFanOut, int largeFanOut, bool dispose) + static int MeasureFanOutSensitivity(int episodes, int smallFanOut, int largeFanOut, bool release) { - int small = RunEpisodes(episodes, smallFanOut, dispose); - int large = RunEpisodes(episodes, largeFanOut, dispose); + int small = RunEpisodes(episodes, smallFanOut, release); + int large = RunEpisodes(episodes, largeFanOut, release); return large - small; } - static int RunEpisodes(int episodes, int fanOut, bool dispose) + static int RunEpisodes(int episodes, int fanOut, bool release) { int before = AsyncDebuggingScope.ActiveTaskCount; for (int i = 0; i < episodes; i++) { TaskOrchestrationExecutor executor = CreateExecutor(new FanOutOrchestration(fanOut)); executor.Execute(); - if (dispose) + if (release) { - executor.Dispose(); + executor.Release(); } } @@ -121,31 +121,30 @@ static int RunEpisodes(int episodes, int fanOut, bool dispose) } [TestMethod] - public void Dispose_DoesNotChangeTheDecisionsAlreadyProduced() + public void Release_DoesNotChangeTheDecisionsAlreadyProduced() { var orchestration = new FanOutOrchestration(fanOut: 3); - using (var executor = CreateExecutor(orchestration)) - { - OrchestratorExecutionResult result = executor.Execute(); - List before = result.Actions.ToList(); + TaskOrchestrationExecutor executor = CreateExecutor(orchestration); - executor.Dispose(); + OrchestratorExecutionResult result = executor.Execute(); + List before = result.Actions.ToList(); - CollectionAssert.AreEqual( - before, - result.Actions.ToList(), - "Releasing abandoned tasks must not add or remove orchestrator actions."); - } + executor.Release(); + + CollectionAssert.AreEqual( + before, + result.Actions.ToList(), + "Releasing abandoned tasks must not add or remove orchestrator actions."); } [TestMethod] - public void Dispose_OrchestratorThatSwallowsCancellation_CannotScheduleMoreWork() + public void Release_OrchestratorThatSwallowsCancellation_CannotScheduleMoreWork() { var orchestration = new SwallowsCancellationOrchestration(); - using (var executor = CreateExecutor(orchestration)) - { - executor.Execute(); - } + TaskOrchestrationExecutor executor = CreateExecutor(orchestration); + executor.Execute(); + + executor.Release(); Assert.IsInstanceOfType( orchestration.RescheduleFailure, @@ -154,16 +153,16 @@ public void Dispose_OrchestratorThatSwallowsCancellation_CannotScheduleMoreWork( } [TestMethod] - public void Dispose_IsIdempotent() + public void Release_IsIdempotent() { var orchestration = new FanOutOrchestration(fanOut: 2); - var executor = CreateExecutor(orchestration); + TaskOrchestrationExecutor executor = CreateExecutor(orchestration); executor.Execute(); - executor.Dispose(); - executor.Dispose(); + executor.Release(); + executor.Release(); - Assert.AreEqual(2, orchestration.ReleasedTaskCount, "Repeated disposal should not resume continuations more than once."); + Assert.AreEqual(2, orchestration.ReleasedTaskCount, "Repeated release should not resume continuations more than once."); } [TestMethod] @@ -174,24 +173,23 @@ public void ExtendedSession_OpenTasksSurviveBetweenEpisodes() var orchestration = new FanOutOrchestration(fanOut: 1); OrchestrationRuntimeState runtimeState = CreateRuntimeState(); - using (var executor = new TaskOrchestrationExecutor(runtimeState, orchestration, BehaviorOnContinueAsNew.Carryover)) - { - OrchestratorExecutionResult firstEpisode = executor.Execute(); - Assert.AreEqual(1, firstEpisode.Actions.Count(), "The first episode should schedule the activity."); - Assert.IsFalse(executor.IsCompleted); + var executor = new TaskOrchestrationExecutor(runtimeState, orchestration, BehaviorOnContinueAsNew.Carryover); - runtimeState.NewEvents.Clear(); - runtimeState.AddEvent(new OrchestratorStartedEvent(-1)); - runtimeState.AddEvent(new TaskCompletedEvent(-1, taskScheduledId: 0, result: JsonDataConverter.Default.Serialize("Hello"))); + OrchestratorExecutionResult firstEpisode = executor.Execute(); + Assert.AreEqual(1, firstEpisode.Actions.Count(), "The first episode should schedule the activity."); + Assert.IsFalse(executor.IsCompleted); - OrchestratorExecutionResult secondEpisode = executor.ExecuteNewEvents(); + runtimeState.NewEvents.Clear(); + runtimeState.AddEvent(new OrchestratorStartedEvent(-1)); + runtimeState.AddEvent(new TaskCompletedEvent(-1, taskScheduledId: 0, result: JsonDataConverter.Default.Serialize("Hello"))); - Assert.IsTrue(executor.IsCompleted, "The activity result should have been delivered to the still-open task."); - Assert.AreEqual(1, orchestration.ReleasedTaskCount, "The await should have been resumed by its result, not by cancellation."); - Assert.IsTrue( - secondEpisode.Actions.OfType().Any(), - "The orchestration should have completed on the second episode."); - } + OrchestratorExecutionResult secondEpisode = executor.ExecuteNewEvents(); + + Assert.IsTrue(executor.IsCompleted, "The activity result should have been delivered to the still-open task."); + Assert.AreEqual(1, orchestration.ReleasedTaskCount, "The await should have been resumed by its result, not by cancellation."); + Assert.IsTrue( + secondEpisode.Actions.OfType().Any(), + "The orchestration should have completed on the second episode."); } static TaskOrchestrationExecutor CreateExecutor(TaskOrchestration orchestration) => diff --git a/src/DurableTask.Core/TaskOrchestrationDispatcher.cs b/src/DurableTask.Core/TaskOrchestrationDispatcher.cs index 95474a122..a15a4ce96 100644 --- a/src/DurableTask.Core/TaskOrchestrationDispatcher.cs +++ b/src/DurableTask.Core/TaskOrchestrationDispatcher.cs @@ -325,7 +325,7 @@ static void ReleaseCursor(ref OrchestrationExecutionCursor? cursor) { OrchestrationExecutionCursor? retiredCursor = cursor; cursor = null; - retiredCursor?.OrchestrationExecutor?.Dispose(); + retiredCursor?.OrchestrationExecutor?.Release(); } /// diff --git a/src/DurableTask.Core/TaskOrchestrationExecutor.cs b/src/DurableTask.Core/TaskOrchestrationExecutor.cs index 0f4f6545a..97b6a9d94 100644 --- a/src/DurableTask.Core/TaskOrchestrationExecutor.cs +++ b/src/DurableTask.Core/TaskOrchestrationExecutor.cs @@ -28,7 +28,7 @@ namespace DurableTask.Core /// /// Utility for executing task orchestrators. /// - public class TaskOrchestrationExecutor : IDisposable + public class TaskOrchestrationExecutor { readonly TaskOrchestrationContext context; readonly TaskScheduler decisionScheduler; @@ -151,8 +151,14 @@ public OrchestratorExecutionResult ExecuteNewEvents() /// completes, so abandoned orchestrator awaits permanently root the orchestration object graph. /// See https://github.com/Azure/azure-functions-durable-extension/issues/340. /// + /// + /// This is intentionally internal rather than an implementation. + /// Executor lifetime is owned by , which is the only + /// component that knows when an executor will never run again, so exposing this externally would + /// add public surface that no caller outside this assembly can use correctly. + /// /// - public void Dispose() + internal void Release() { SynchronizationContext prevCtx = SynchronizationContext.Current; bool prevIsOrchestratorThread = OrchestrationContext.IsOrchestratorThread; From dc976ecc313c67bdcc7663090883cca1265cb193 Mon Sep 17 00:00:00 2001 From: wangbill Date: Thu, 27 Aug 2026 19:10:13 -0400 Subject: [PATCH 3/3] Gate abandoned-task release on debugger presence Cancelling the abandoned orchestrator tasks is what lets the CLR drop its active-task roots, but it is not side-effect free: resuming an abandoned await runs the orchestrator's catch and finally blocks. There is no supported way to remove those roots without running the continuations, so paying that cost unconditionally changes production behavior for a leak that only exists when a debugger is attached. Scope the teardown to the case that needs it: - TaskOrchestrationDispatcher.ReleaseCursor always clears the cursor, but now only calls TaskOrchestrationExecutor.Release() when Debugger.IsAttached. The gate lives in the dispatcher because it owns executor lifetime, and it is lifted into an internal overload that takes the flag so both branches are unit-testable without an attached debugger. Both retirement sites (the outer session finally and continue-as-new) keep clearing the cursor unconditionally. - TaskOrchestrationContext.ThrowIfReleased now throws InvalidOperationException instead of OperationCanceledException. The first failure at the abandoned await necessarily remains an OperationCanceledException, but a retired executor is permanent, so the very common 'catch (OperationCanceledException) { retry; }' shape must not treat it as a transient cancellation and spin. Without a debugger the dispatcher behaves exactly as it did before the leak fix: the executor is dropped and collected, and no orchestrator continuation runs. See dotnet/runtime#26565 for the upstream proposal to make the active-task table weak, which would make this teardown unnecessary. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> Copilot-Session: 019e7654-1037-4e02-b2fd-c03e09d6b7e2 --- .../TaskOrchestrationExecutorTests.cs | 75 +++++++++++++++++-- .../TaskOrchestrationContext.cs | 15 +++- .../TaskOrchestrationDispatcher.cs | 35 ++++++++- .../TaskOrchestrationExecutor.cs | 6 ++ 4 files changed, 120 insertions(+), 11 deletions(-) diff --git a/Test/DurableTask.Core.Tests/TaskOrchestrationExecutorTests.cs b/Test/DurableTask.Core.Tests/TaskOrchestrationExecutorTests.cs index 6d7403b10..0fad630ef 100644 --- a/Test/DurableTask.Core.Tests/TaskOrchestrationExecutorTests.cs +++ b/Test/DurableTask.Core.Tests/TaskOrchestrationExecutorTests.cs @@ -32,7 +32,9 @@ namespace DurableTask.Core.Tests /// the episode ends. When a debugger is attached, the CLR keeps every awaited task in the process-wide /// Task.s_currentActiveTasks dictionary until the task completes, so abandoned awaits permanently /// root the orchestration object graph. Releasing the executor cancels the open tasks, which lets those - /// awaiter continuations run and unregister themselves. + /// awaiter continuations run and unregister themselves. The dispatcher only does that when a debugger is + /// attached, since the cancellation resumes user code; these tests drive the internal entry points + /// directly so both branches are covered without an attached debugger. /// Regression coverage for https://github.com/Azure/azure-functions-durable-extension/issues/340. /// [TestClass] @@ -146,10 +148,24 @@ public void Release_OrchestratorThatSwallowsCancellation_CannotScheduleMoreWork( executor.Release(); + // The first failure has to be an OperationCanceledException: that is simply what a cancelled + // TaskCompletionSource raises at the await that was abandoned. Assert.IsInstanceOfType( - orchestration.RescheduleFailure, + orchestration.FirstFailure, typeof(OperationCanceledException), + "Releasing an open task must surface as cancellation at the await that was abandoned."); + + // The retry must not. If retirement also looked like cancellation, the extremely common + // 'catch (OperationCanceledException) { retry; }' shape would treat a permanently retired + // executor as a transient failure and loop. + Assert.IsInstanceOfType( + orchestration.RescheduleFailure, + typeof(InvalidOperationException), "A released context must refuse to open new tasks, otherwise resumed orchestrator code can leak again."); + Assert.IsNotInstanceOfType( + orchestration.RescheduleFailure, + typeof(OperationCanceledException), + "Retirement must not be reported as cancellation, or cancellation-specific retry loops will spin."); } [TestMethod] @@ -165,6 +181,43 @@ public void Release_IsIdempotent() Assert.AreEqual(2, orchestration.ReleasedTaskCount, "Repeated release should not resume continuations more than once."); } + [TestMethod] + public void ReleaseCursor_WithTeardownEnabled_ClearsCursorAndResumesContinuations() + { + var orchestration = new FanOutOrchestration(fanOut: 3); + TaskOrchestrationExecutor executor = CreateExecutor(orchestration); + executor.Execute(); + + OrchestrationExecutionCursor cursor = CreateCursor(executor); + TaskOrchestrationDispatcher.ReleaseCursor(ref cursor, runContinuationTeardown: true); + + Assert.IsNull(cursor, "Retiring a cursor must always clear the reference."); + Assert.AreEqual( + 3, + orchestration.ReleasedTaskCount, + "With teardown enabled the abandoned awaits should be resumed so they can unregister themselves."); + } + + [TestMethod] + public void ReleaseCursor_WithTeardownDisabled_ClearsCursorWithoutRunningContinuations() + { + // Teardown is gated on Debugger.IsAttached because cancelling the open tasks necessarily runs + // user catch/finally blocks. Without a debugger there is no leak to fix, so production keeps the + // pre-existing behavior: the executor is simply dropped and collected. + var orchestration = new FanOutOrchestration(fanOut: 3); + TaskOrchestrationExecutor executor = CreateExecutor(orchestration); + executor.Execute(); + + OrchestrationExecutionCursor cursor = CreateCursor(executor); + TaskOrchestrationDispatcher.ReleaseCursor(ref cursor, runContinuationTeardown: false); + + Assert.IsNull(cursor, "The cursor must be cleared whether or not continuation teardown runs."); + Assert.AreEqual( + 0, + orchestration.ReleasedTaskCount, + "With teardown disabled no orchestrator continuation may run, so user code is unaffected."); + } + [TestMethod] public void ExtendedSession_OpenTasksSurviveBetweenEpisodes() { @@ -195,6 +248,13 @@ public void ExtendedSession_OpenTasksSurviveBetweenEpisodes() static TaskOrchestrationExecutor CreateExecutor(TaskOrchestration orchestration) => new TaskOrchestrationExecutor(CreateRuntimeState(), orchestration, BehaviorOnContinueAsNew.Carryover); + static OrchestrationExecutionCursor CreateCursor(TaskOrchestrationExecutor executor) => + new OrchestrationExecutionCursor( + CreateRuntimeState(), + orchestration: null, + executor: executor, + latestDecisions: Enumerable.Empty()); + static OrchestrationRuntimeState CreateRuntimeState() { var runtimeState = new OrchestrationRuntimeState(); @@ -257,11 +317,14 @@ async Task AwaitActivityAsync(OrchestrationContext context) } /// - /// Mimics orchestrator code with a catch-all handler: it swallows the cancellation raised while the - /// executor is being released and then tries to schedule more work. + /// Mimics orchestrator code that retries on cancellation: it catches the + /// raised while the executor is being released and then + /// tries to schedule more work. /// class SwallowsCancellationOrchestration : TaskOrchestration { + public Exception FirstFailure { get; private set; } + public Exception RescheduleFailure { get; private set; } public override async Task RunTask(OrchestrationContext context, string input) @@ -270,8 +333,10 @@ public override async Task RunTask(OrchestrationContext context, string { return await context.ScheduleTask(ActivityName, string.Empty); } - catch (Exception) + catch (OperationCanceledException firstFailure) { + this.FirstFailure = firstFailure; + try { return await context.ScheduleTask(ActivityName, string.Empty); diff --git a/src/DurableTask.Core/TaskOrchestrationContext.cs b/src/DurableTask.Core/TaskOrchestrationContext.cs index cbb9bffd2..6e5af2980 100644 --- a/src/DurableTask.Core/TaskOrchestrationContext.cs +++ b/src/DurableTask.Core/TaskOrchestrationContext.cs @@ -94,6 +94,13 @@ public TaskOrchestrationContext( /// allows the graph to be collected. This is only safe once the executor is guaranteed never to be /// used again, since a cancelled task can no longer receive a result on a subsequent episode. /// + /// + /// Cancellation is not side-effect free: resuming an abandoned await runs the orchestrator's + /// catch and finally blocks. There is no supported way to drop the CLR's active-task + /// roots without running those continuations, so callers gate this on debugger presence rather than + /// paying the cost in production. See dotnet/runtime#26565 for the upstream proposal to make the + /// active-task table weak, which would remove the need for this teardown entirely. + /// /// internal void ReleaseOpenTasks() { @@ -141,8 +148,12 @@ private void ThrowIfReleased() { if (this.isReleased) { - throw new OperationCanceledException( - "This orchestration episode has ended and the orchestration context can no longer schedule work."); + // Deliberately not an OperationCanceledException. The first cancellation surfaced at the + // abandoned await necessarily is one, and orchestrator code that catches it and retries would + // otherwise treat permanent retirement as a transient cancellation and loop. + throw new InvalidOperationException( + "This orchestration executor has been retired and its context can no longer schedule work. " + + "The orchestration episode has ended and any work scheduled now would never run."); } } diff --git a/src/DurableTask.Core/TaskOrchestrationDispatcher.cs b/src/DurableTask.Core/TaskOrchestrationDispatcher.cs index a15a4ce96..0abb3d1cd 100644 --- a/src/DurableTask.Core/TaskOrchestrationDispatcher.cs +++ b/src/DurableTask.Core/TaskOrchestrationDispatcher.cs @@ -310,10 +310,10 @@ async Task OnProcessWorkItemSessionAsync(TaskOrchestrationWorkItem workItem) } finally { - // The session is over and the executor will never run again, so release the orchestrator - // continuations it abandoned while waiting on activities, sub-orchestrations, and timers. - // Leaving them pending leaks the whole orchestration object graph when a debugger is - // attached. See https://github.com/Azure/azure-functions-durable-extension/issues/340. + // The session is over and the executor will never run again. Always drop the cursor; only tear + // down the abandoned orchestrator continuations when a debugger is attached, because that + // teardown necessarily runs user catch/finally blocks and the leak it prevents is + // debugger-only. See https://github.com/Azure/azure-functions-durable-extension/issues/340. ReleaseCursor(ref workItem.Cursor); } } @@ -321,10 +321,37 @@ async Task OnProcessWorkItemSessionAsync(TaskOrchestrationWorkItem workItem) /// /// Retires the executor held by and clears the reference. /// + /// + /// Continuation teardown is gated on . The leak this guards against + /// only exists when a debugger is attached, and the teardown is not free of side effects: cancelling + /// the abandoned tasks resumes orchestrator code, so user catch and finally blocks run. + /// There is no supported way to drop the CLR's active-task roots without running those continuations, + /// so the cost is confined to the situation that actually needs it. + /// static void ReleaseCursor(ref OrchestrationExecutionCursor? cursor) + { + ReleaseCursor(ref cursor, runContinuationTeardown: Debugger.IsAttached); + } + + /// + /// Retirement logic behind , with the + /// debugger check lifted into so both branches are testable. + /// + /// The cursor to retire. Always cleared, whatever the second argument is. + /// + /// Whether to also cancel the executor's abandoned orchestrator continuations. When false the executor + /// is simply dropped, which is what this dispatcher did before the leak fix. + /// + internal static void ReleaseCursor(ref OrchestrationExecutionCursor? cursor, bool runContinuationTeardown) { OrchestrationExecutionCursor? retiredCursor = cursor; cursor = null; + + if (!runContinuationTeardown) + { + return; + } + retiredCursor?.OrchestrationExecutor?.Release(); } diff --git a/src/DurableTask.Core/TaskOrchestrationExecutor.cs b/src/DurableTask.Core/TaskOrchestrationExecutor.cs index 97b6a9d94..161771eb3 100644 --- a/src/DurableTask.Core/TaskOrchestrationExecutor.cs +++ b/src/DurableTask.Core/TaskOrchestrationExecutor.cs @@ -157,6 +157,12 @@ public OrchestratorExecutionResult ExecuteNewEvents() /// component that knows when an executor will never run again, so exposing this externally would /// add public surface that no caller outside this assembly can use correctly. /// + /// + /// This is not side-effect free: cancelling the open tasks resumes orchestrator code, so the + /// orchestrator's catch and finally blocks run. There is no supported way to drop the + /// CLR's active-task roots without running those continuations, so the dispatcher only calls this + /// when a debugger is attached, which is the only situation in which the leak exists. + /// /// internal void Release() {