diff --git a/Test/DurableTask.Core.Tests/TaskOrchestrationExecutorTests.cs b/Test/DurableTask.Core.Tests/TaskOrchestrationExecutorTests.cs
new file mode 100644
index 00000000..0fad630e
--- /dev/null
+++ b/Test/DurableTask.Core.Tests/TaskOrchestrationExecutorTests.cs
@@ -0,0 +1,407 @@
+// ----------------------------------------------------------------------------------
+// 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. Releasing the executor cancels the open tasks, which lets those
+ /// 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]
+ public class TaskOrchestrationExecutorTests
+ {
+ const string ActivityName = "SayHello";
+
+ [TestMethod]
+ public void Release_ResumesAbandonedOrchestratorContinuations()
+ {
+ var orchestration = new FanOutOrchestration(fanOut: 3);
+ 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.");
+
+ executor.Release();
+
+ Assert.AreEqual(
+ 3,
+ orchestration.ReleasedTaskCount,
+ "Releasing the executor should resume every abandoned await so its continuation can unregister itself.");
+ }
+
+ [TestMethod]
+ 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.
+ //
+ // 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, release: true);
+
+ int leakySensitivity = MeasureFanOutSensitivity(Episodes, SmallFanOut, LargeFanOut, release: false);
+ Assert.IsTrue(
+ leakySensitivity > Episodes * (LargeFanOut - SmallFanOut),
+ "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, release: true);
+ Assert.IsTrue(
+ fixedSensitivity * 20 < leakySensitivity,
+ "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 unreleased.");
+ }
+ }
+
+ ///
+ /// 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 release)
+ {
+ int small = RunEpisodes(episodes, smallFanOut, release);
+ int large = RunEpisodes(episodes, largeFanOut, release);
+ return large - small;
+ }
+
+ 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 (release)
+ {
+ executor.Release();
+ }
+ }
+
+ return AsyncDebuggingScope.ActiveTaskCount - before;
+ }
+
+ [TestMethod]
+ public void Release_DoesNotChangeTheDecisionsAlreadyProduced()
+ {
+ var orchestration = new FanOutOrchestration(fanOut: 3);
+ TaskOrchestrationExecutor executor = CreateExecutor(orchestration);
+
+ OrchestratorExecutionResult result = executor.Execute();
+ List before = result.Actions.ToList();
+
+ executor.Release();
+
+ CollectionAssert.AreEqual(
+ before,
+ result.Actions.ToList(),
+ "Releasing abandoned tasks must not add or remove orchestrator actions.");
+ }
+
+ [TestMethod]
+ public void Release_OrchestratorThatSwallowsCancellation_CannotScheduleMoreWork()
+ {
+ var orchestration = new SwallowsCancellationOrchestration();
+ TaskOrchestrationExecutor executor = CreateExecutor(orchestration);
+ executor.Execute();
+
+ 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.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]
+ public void Release_IsIdempotent()
+ {
+ var orchestration = new FanOutOrchestration(fanOut: 2);
+ TaskOrchestrationExecutor executor = CreateExecutor(orchestration);
+ executor.Execute();
+
+ executor.Release();
+ executor.Release();
+
+ 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()
+ {
+ // 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();
+
+ 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 OrchestrationExecutionCursor CreateCursor(TaskOrchestrationExecutor executor) =>
+ new OrchestrationExecutionCursor(
+ CreateRuntimeState(),
+ orchestration: null,
+ executor: executor,
+ latestDecisions: Enumerable.Empty());
+
+ 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 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)
+ {
+ try
+ {
+ return await context.ScheduleTask(ActivityName, string.Empty);
+ }
+ catch (OperationCanceledException firstFailure)
+ {
+ this.FirstFailure = firstFailure;
+
+ 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 e4846124..6e5af298 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,87 @@ 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.
+ ///
+ ///
+ /// 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()
+ {
+ 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)
+ {
+ // 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.");
+ }
+ }
+
internal void ClearPendingActions()
{
this.orchestratorActionsMap.Clear();
@@ -126,6 +208,8 @@ public async Task ScheduleTaskToWorker(string name, string ver
public async Task