diff --git a/Test/DurableTask.Core.Tests/RewindTests.cs b/Test/DurableTask.Core.Tests/RewindTests.cs
new file mode 100644
index 00000000..7882a95f
--- /dev/null
+++ b/Test/DurableTask.Core.Tests/RewindTests.cs
@@ -0,0 +1,537 @@
+// ----------------------------------------------------------------------------------
+// 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.Threading.Tasks;
+ using DurableTask.Core.Command;
+ using DurableTask.Core.History;
+ using Microsoft.VisualStudio.TestTools.UnitTesting;
+
+ ///
+ /// Tests for the history scrubbing performed by
+ /// .
+ ///
+ ///
+ /// Each test drives a real through real episodes with
+ /// until it fails, rewinds the resulting history, then
+ /// replays the rewound history and asserts on the actions the orchestrator produces. A rewound
+ /// history that is not replayable surfaces as a fail-orchestration action carrying a
+ /// "Non-Deterministic workflow detected" message, because
+ /// converts
+ /// into that action.
+ ///
+ [TestClass]
+ public class RewindTests
+ {
+ const string FailingActivity = "FailingActivity";
+
+ ///
+ /// Regression test for https://github.com/Azure/azure-functions-durable-extension/issues/444.
+ /// The orchestrator catches the activity failure and schedules another activity before
+ /// rethrowing. That second activity only exists because the orchestrator observed the
+ /// failure, so it must not survive the rewind.
+ ///
+ [TestMethod]
+ public void Rewind_CatchBlockSchedulesActivity_ProducesReplayableHistory()
+ {
+ RewindResult result = RunRewindTest(() => new CatchAndScheduleActivityOrchestration());
+
+ AssertReplayable(result);
+
+ // 'Cleanup' was scheduled from the catch block, so neither it nor its result may survive.
+ Assert.IsFalse(
+ result.RewoundHistory.OfType().Any(e => e.Name == "Cleanup"),
+ "The activity scheduled from the catch block should have been removed from the history.");
+ Assert.AreEqual(
+ 0,
+ result.RewoundHistory.OfType().Count(e => e.TaskScheduledId == 3),
+ "The result of the activity scheduled from the catch block should have been removed too.");
+
+ // The two activities that succeeded before the failure are retained, so only the failed
+ // activity is rescheduled.
+ CollectionAssert.AreEqual(
+ new[] { "First", "Second" },
+ result.RewoundHistory.OfType().Select(e => e.Name).ToArray());
+ CollectionAssert.AreEqual(
+ new[] { FailingActivity },
+ result.ReplayScheduledActivities);
+ }
+
+ ///
+ /// An activity scheduled through
+ /// always leaves a delay timer in the history (see ), including
+ /// one created after the final failed attempt. Those timers are created only because the
+ /// orchestrator observed a failure, so they must not survive the rewind either.
+ ///
+ [TestMethod]
+ public void Rewind_ScheduleWithRetry_RemovesRetryTimers()
+ {
+ RewindResult result = RunRewindTest(() => new RetryOrchestration());
+
+ AssertReplayable(result);
+
+ Assert.AreEqual(
+ 0,
+ result.RewoundHistory.OfType().Count(),
+ "Retry timers should have been removed from the history.");
+ Assert.AreEqual(
+ 0,
+ result.RewoundHistory.OfType().Count(),
+ "Retry timer results should have been removed from the history.");
+ Assert.AreEqual(
+ 0,
+ result.RewoundHistory.OfType().Count(),
+ "Every attempt of the failed activity should have been removed from the history.");
+ CollectionAssert.AreEqual(
+ new[] { FailingActivity },
+ result.ReplayScheduledActivities);
+ }
+
+ ///
+ /// Same as , but the
+ /// catch block creates a sub-orchestration rather than scheduling an activity.
+ ///
+ [TestMethod]
+ public void Rewind_CatchBlockCreatesSubOrchestration_ProducesReplayableHistory()
+ {
+ RewindResult result = RunRewindTest(() => new CatchAndCreateSubOrchestrationOrchestration());
+
+ AssertReplayable(result);
+
+ Assert.AreEqual(
+ 0,
+ result.RewoundHistory.OfType().Count(),
+ "The sub-orchestration created from the catch block should have been removed.");
+ Assert.AreEqual(
+ 0,
+ result.RewoundHistory.OfType().Count(),
+ "The result of the sub-orchestration created from the catch block should have been removed.");
+ CollectionAssert.AreEqual(
+ new[] { FailingActivity },
+ result.ReplayScheduledActivities);
+
+ // No failed sub-orchestration means this is a terminal leaf, so a dummy rewind message is
+ // emitted to force the orchestration to rerun.
+ Assert.AreEqual(1, result.RewindMessages.Count);
+ Assert.IsInstanceOfType(result.RewindMessages[0].Event, typeof(ExecutionRewoundEvent));
+ }
+
+ ///
+ /// Same as , but the
+ /// catch block sends an external event, which is also matched against the orchestrator's
+ /// sequence-ID counter during replay.
+ ///
+ [TestMethod]
+ public void Rewind_CatchBlockSendsEvent_ProducesReplayableHistory()
+ {
+ RewindResult result = RunRewindTest(() => new CatchAndSendEventOrchestration());
+
+ AssertReplayable(result);
+
+ Assert.AreEqual(
+ 0,
+ result.RewoundHistory.OfType().Count(),
+ "The event sent from the catch block should have been removed from the history.");
+ CollectionAssert.AreEqual(
+ new[] { FailingActivity },
+ result.ReplayScheduledActivities);
+ }
+
+ ///
+ /// Control case: an orchestration that does no work in response to the failure must rewind
+ /// exactly as it did before this change.
+ ///
+ [TestMethod]
+ public void Rewind_SimpleActivityFailure_ReschedulesOnlyTheFailedActivity()
+ {
+ RewindResult result = RunRewindTest(() => new SimpleFailureOrchestration());
+
+ AssertReplayable(result);
+
+ CollectionAssert.AreEqual(
+ new[] { "First" },
+ result.RewoundHistory.OfType().Select(e => e.Name).ToArray());
+ CollectionAssert.AreEqual(
+ new[] { FailingActivity },
+ result.ReplayScheduledActivities);
+ }
+
+ ///
+ /// Fan-out/fan-in: the parallel branches are all scheduled in an episode before the failure is
+ /// observed, so they must be retained and only the failed branch rescheduled.
+ ///
+ [TestMethod]
+ public void Rewind_FanOutFanIn_RetainsSuccessfulBranches()
+ {
+ RewindResult result = RunRewindTest(() => new FanOutFanInOrchestration());
+
+ AssertReplayable(result);
+
+ CollectionAssert.AreEquivalent(
+ new[] { "Branch1", "Branch3" },
+ result.RewoundHistory.OfType().Select(e => e.Name).ToArray());
+ CollectionAssert.AreEqual(
+ new[] { FailingActivity },
+ result.ReplayScheduledActivities);
+ }
+
+ ///
+ /// A sub-orchestration is always created in an episode before the one that delivers its
+ /// failure, so its creation event is retained and a rewind message is emitted for the child.
+ ///
+ [TestMethod]
+ public void Rewind_FailedSubOrchestration_RetainsCreationAndEmitsChildRewindMessage()
+ {
+ RewindResult result = RunRewindTest(() => new SubOrchestrationFailureOrchestration());
+
+ AssertReplayable(result);
+
+ SubOrchestrationInstanceCreatedEvent createdEvent =
+ result.RewoundHistory.OfType().SingleOrDefault();
+ Assert.IsNotNull(createdEvent, "The failed sub-orchestration's creation event should be retained.");
+ Assert.AreEqual("ChildInstance", createdEvent.InstanceId);
+
+ Assert.AreEqual(1, result.RewindMessages.Count);
+ Assert.AreEqual("ChildInstance", result.RewindMessages[0].OrchestrationInstance.InstanceId);
+ Assert.IsInstanceOfType(result.RewindMessages[0].Event, typeof(ExecutionRewoundEvent));
+
+ // The parent replays up to the sub-orchestration and then waits for the rewound child.
+ Assert.AreEqual(0, result.ReplayScheduledActivities.Count);
+ }
+
+ ///
+ /// The rewound history must always carry a fresh execution ID.
+ ///
+ [TestMethod]
+ public void Rewind_AssignsNewExecutionId()
+ {
+ RewindResult result = RunRewindTest(() => new SimpleFailureOrchestration());
+
+ ExecutionStartedEvent executionStartedEvent =
+ result.RewoundHistory.OfType().Single();
+ Assert.AreNotEqual(
+ InitialExecutionId,
+ executionStartedEvent.OrchestrationInstance.ExecutionId,
+ "The rewound history should carry a new execution ID.");
+ Assert.AreEqual(0, result.RewoundHistory.OfType().Count());
+ }
+
+ #region Test orchestrations
+
+ class SimpleFailureOrchestration : TaskOrchestration
+ {
+ public override async Task RunTask(OrchestrationContext context, string input)
+ {
+ await context.ScheduleTask("First", string.Empty);
+ await context.ScheduleTask(FailingActivity, string.Empty);
+ return "done";
+ }
+ }
+
+ class CatchAndScheduleActivityOrchestration : TaskOrchestration
+ {
+ public override async Task RunTask(OrchestrationContext context, string input)
+ {
+ await context.ScheduleTask("First", string.Empty);
+ await context.ScheduleTask("Second", string.Empty);
+ try
+ {
+ await context.ScheduleTask(FailingActivity, string.Empty);
+ }
+ catch (Exception)
+ {
+ await context.ScheduleTask("Cleanup", string.Empty);
+ throw;
+ }
+
+ return "done";
+ }
+ }
+
+ class CatchAndCreateSubOrchestrationOrchestration : TaskOrchestration
+ {
+ public override async Task RunTask(OrchestrationContext context, string input)
+ {
+ try
+ {
+ await context.ScheduleTask(FailingActivity, string.Empty);
+ }
+ catch (Exception)
+ {
+ await context.CreateSubOrchestrationInstance("Notify", string.Empty, "NotifyInstance", null);
+ throw;
+ }
+
+ return "done";
+ }
+ }
+
+ class CatchAndSendEventOrchestration : TaskOrchestration
+ {
+ public override async Task RunTask(OrchestrationContext context, string input)
+ {
+ try
+ {
+ await context.ScheduleTask(FailingActivity, string.Empty);
+ }
+ catch (Exception)
+ {
+ context.SendEvent(
+ new OrchestrationInstance { InstanceId = "Listener" },
+ "Failed",
+ "payload");
+ throw;
+ }
+
+ return "done";
+ }
+ }
+
+ class RetryOrchestration : TaskOrchestration
+ {
+ public override async Task RunTask(OrchestrationContext context, string input)
+ {
+ var retryOptions = new RetryOptions(TimeSpan.FromSeconds(1), 2);
+ return await context.ScheduleWithRetry(FailingActivity, string.Empty, retryOptions);
+ }
+ }
+
+ class FanOutFanInOrchestration : TaskOrchestration
+ {
+ public override async Task RunTask(OrchestrationContext context, string input)
+ {
+ var tasks = new List>
+ {
+ context.ScheduleTask("Branch1", string.Empty),
+ context.ScheduleTask(FailingActivity, string.Empty),
+ context.ScheduleTask("Branch3", string.Empty),
+ };
+
+ await Task.WhenAll(tasks);
+ return "done";
+ }
+ }
+
+ class SubOrchestrationFailureOrchestration : TaskOrchestration
+ {
+ public override async Task RunTask(OrchestrationContext context, string input)
+ {
+ return await context.CreateSubOrchestrationInstance(
+ FailingActivity,
+ string.Empty,
+ "ChildInstance",
+ null);
+ }
+ }
+
+ #endregion
+
+ #region Harness
+
+ const string InstanceId = "TestInstance";
+ const string InitialExecutionId = "TestExecution";
+
+ class RewindResult
+ {
+ public List HistoryAtFailure { get; set; }
+
+ public List RewoundHistory { get; set; }
+
+ public List RewindMessages { get; set; }
+
+ public IReadOnlyList ReplayActions { get; set; }
+
+ public List ReplayScheduledActivities { get; set; }
+
+ public string ReplayFailureReason { get; set; }
+ }
+
+ static void AssertReplayable(RewindResult result)
+ {
+ Assert.IsNull(
+ result.ReplayFailureReason,
+ "Replaying the rewound history should not fail the orchestration. Actual failure: "
+ + result.ReplayFailureReason);
+ }
+
+ static RewindResult RunRewindTest(Func orchestrationFactory)
+ {
+ List historyAtFailure = RunToFailure(orchestrationFactory());
+
+ var runtimeState = new OrchestrationRuntimeState(historyAtFailure);
+ runtimeState.AddEvent(new OrchestratorStartedEvent(-1));
+ runtimeState.AddEvent(new ExecutionRewoundEvent(-1, "rewind requested"));
+
+ TaskOrchestrationDispatcher.ProcessRewindOrchestrationDecision(
+ runtimeState,
+ out List rewindMessages,
+ out OrchestrationRuntimeState rewoundState);
+
+ List rewoundHistory = rewoundState.Events.ToList();
+
+ // The dispatcher persists the rewound history and the orchestration is then picked up by a
+ // fresh work item.
+ var replayState = new OrchestrationRuntimeState(rewoundHistory);
+ replayState.AddEvent(new OrchestratorStartedEvent(-1));
+
+ var executor = new TaskOrchestrationExecutor(
+ replayState,
+ orchestrationFactory(),
+ BehaviorOnContinueAsNew.Carryover);
+ OrchestratorExecutionResult replayResult = executor.Execute();
+
+ OrchestrationCompleteOrchestratorAction failureAction = replayResult.Actions
+ .OfType()
+ .FirstOrDefault(a => a.OrchestrationStatus == OrchestrationStatus.Failed);
+
+ return new RewindResult
+ {
+ HistoryAtFailure = historyAtFailure,
+ RewoundHistory = rewoundHistory,
+ RewindMessages = rewindMessages,
+ ReplayActions = replayResult.Actions.ToList(),
+ ReplayScheduledActivities = replayResult.Actions
+ .OfType()
+ .Select(a => a.Name)
+ .ToList(),
+ ReplayFailureReason = failureAction == null
+ ? null
+ : failureAction.FailureDetails?.ErrorMessage ?? failureAction.Result,
+ };
+ }
+
+ ///
+ /// Drives the orchestration through real episodes until it completes, failing every task named
+ /// and completing everything else.
+ ///
+ static List RunToFailure(TaskOrchestration orchestration)
+ {
+ var instance = new OrchestrationInstance
+ {
+ InstanceId = InstanceId,
+ ExecutionId = InitialExecutionId,
+ };
+
+ var history = new List();
+ var inbox = new List
+ {
+ new ExecutionStartedEvent(-1, "\"input\"")
+ {
+ OrchestrationInstance = instance,
+ Name = "TestOrchestration",
+ Version = string.Empty,
+ },
+ };
+
+ for (int episode = 0; episode < 25; episode++)
+ {
+ var runtimeState = new OrchestrationRuntimeState(history);
+ runtimeState.AddEvent(new OrchestratorStartedEvent(-1));
+ foreach (HistoryEvent inboundEvent in inbox)
+ {
+ runtimeState.AddEvent(inboundEvent);
+ }
+
+ inbox = new List();
+
+ var executor = new TaskOrchestrationExecutor(
+ runtimeState,
+ orchestration,
+ BehaviorOnContinueAsNew.Carryover);
+ OrchestratorExecutionResult result = executor.Execute();
+
+ bool completed = false;
+ foreach (OrchestratorAction action in result.Actions)
+ {
+ switch (action)
+ {
+ case ScheduleTaskOrchestratorAction scheduleTask:
+ runtimeState.AddEvent(new TaskScheduledEvent(
+ scheduleTask.Id,
+ scheduleTask.Name,
+ scheduleTask.Version,
+ scheduleTask.Input));
+ inbox.Add(scheduleTask.Name == FailingActivity
+ ? (HistoryEvent)new TaskFailedEvent(-1, scheduleTask.Id, "failure", "details")
+ : new TaskCompletedEvent(-1, scheduleTask.Id, "\"ok\""));
+ break;
+
+ case CreateSubOrchestrationAction createSubOrchestration:
+ runtimeState.AddEvent(new SubOrchestrationInstanceCreatedEvent(createSubOrchestration.Id)
+ {
+ Name = createSubOrchestration.Name,
+ Version = createSubOrchestration.Version,
+ InstanceId = createSubOrchestration.InstanceId,
+ Input = createSubOrchestration.Input,
+ });
+ inbox.Add(createSubOrchestration.Name == FailingActivity
+ ? (HistoryEvent)new SubOrchestrationInstanceFailedEvent(
+ -1,
+ createSubOrchestration.Id,
+ "failure",
+ "details")
+ : new SubOrchestrationInstanceCompletedEvent(-1, createSubOrchestration.Id, "\"ok\""));
+ break;
+
+ case CreateTimerOrchestratorAction createTimer:
+ runtimeState.AddEvent(new TimerCreatedEvent(createTimer.Id)
+ {
+ FireAt = createTimer.FireAt,
+ });
+ inbox.Add(new TimerFiredEvent(-1, createTimer.FireAt)
+ {
+ TimerId = createTimer.Id,
+ });
+ break;
+
+ case SendEventOrchestratorAction sendEvent:
+ runtimeState.AddEvent(new EventSentEvent(sendEvent.Id)
+ {
+ InstanceId = sendEvent.Instance.InstanceId,
+ Name = sendEvent.EventName,
+ Input = sendEvent.EventData,
+ });
+ break;
+
+ case OrchestrationCompleteOrchestratorAction complete:
+ runtimeState.AddEvent(new ExecutionCompletedEvent(
+ -1,
+ complete.Result,
+ complete.OrchestrationStatus,
+ complete.FailureDetails));
+ completed = true;
+ break;
+
+ default:
+ throw new InvalidOperationException(
+ "Unexpected orchestrator action: " + action.GetType().Name);
+ }
+ }
+
+ history = runtimeState.Events.ToList();
+ if (completed)
+ {
+ return history;
+ }
+ }
+
+ throw new InvalidOperationException("The test orchestration never completed.");
+ }
+
+ #endregion
+ }
+}
diff --git a/src/DurableTask.AzureStorage/Tracking/AzureTableTrackingStore.cs b/src/DurableTask.AzureStorage/Tracking/AzureTableTrackingStore.cs
index 6cecb9d6..151b6eea 100644
--- a/src/DurableTask.AzureStorage/Tracking/AzureTableTrackingStore.cs
+++ b/src/DurableTask.AzureStorage/Tracking/AzureTableTrackingStore.cs
@@ -276,12 +276,16 @@ public override async IAsyncEnumerable RewindHistoryAsync(string instanc
// REWIND ALGORITHM:
// 1. Finds failed execution of specified orchestration instance to rewind
// 2. Finds failure entities to clear and over-writes them (as well as corresponding trigger events)
- // 3. Identifies sub-orchestration failure(s) from parent instance and calls RewindHistoryAsync recursively on failed sub-orchestration child instance(s)
- // 4. Resets orchestration status of rewound instance in instance store table to prepare it to be restarted
- // 5. Returns "failedLeaves", a list of the deepest failed instances on each failed branch to revive with RewindEvent messages
+ // 3. Over-writes everything the orchestrator scheduled at or after the episode in which it first observed a failure,
+ // together with the entities carrying those results, since the replayed orchestrator can no longer reach them
+ // 4. Identifies sub-orchestration failure(s) from parent instance and calls RewindHistoryAsync recursively on failed sub-orchestration child instance(s)
+ // 5. Resets orchestration status of rewound instance in instance store table to prepare it to be restarted
+ // 6. Returns "failedLeaves", a list of the deepest failed instances on each failed branch to revive with RewindEvent messages
+ //
+ // NOTE: this must be kept in sync with TaskOrchestrationDispatcher.ProcessRewindOrchestrationDecision, which performs the
+ // equivalent scrub over an OrchestrationRuntimeState for backends that rewind at the SDK layer.
////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
- bool hasFailedSubOrchestrations = false;
string partitionFilter = AzureTableQueryFilter.PartitionKeyEquals(instanceId);
string orchestratorStartedFilter = $"{partitionFilter} and {nameof(HistoryEvent.EventType)} eq '{nameof(EventType.OrchestratorStarted)}'";
@@ -296,86 +300,176 @@ public override async IAsyncEnumerable RewindHistoryAsync(string instanc
// Use parameterized filter to prevent OData injection via crafted execution IDs
string executionIdFilter = AzureTableQueryFilter.ColumnEquals(nameof(OrchestrationInstance.ExecutionId), executionId);
- var updateFilterBuilder = new StringBuilder();
- updateFilterBuilder.Append($"{partitionFilter}");
- updateFilterBuilder.Append($" and {executionIdFilter}");
- updateFilterBuilder.Append(" and (");
- updateFilterBuilder.Append($"{nameof(ExecutionCompletedEvent.OrchestrationStatus)} eq '{nameof(OrchestrationStatus.Failed)}'");
- updateFilterBuilder.Append($" or {nameof(HistoryEvent.EventType)} eq '{nameof(EventType.TaskFailed)}'");
- updateFilterBuilder.Append($" or {nameof(HistoryEvent.EventType)} eq '{nameof(EventType.SubOrchestrationInstanceFailed)}'");
- updateFilterBuilder.Append(')');
+ // The full history of the current execution. Row keys are the chronological sequence number of the event formatted as
+ // fixed-width hex, and Azure Table Storage returns entities ordered by row key within a partition, so this list is in
+ // chronological order.
+ IReadOnlyList historyEntities = await this.QueryHistoryAsync(
+ $"{partitionFilter} and {executionIdFilter}",
+ instanceId,
+ cancellationToken);
- IReadOnlyList entitiesToClear = await this.QueryHistoryAsync(updateFilterBuilder.ToString(), instanceId, cancellationToken);
- foreach (TableEntity entity in entitiesToClear)
+ // Determine the task IDs of the failed tasks and suborchestrations, along with the earliest episode in which the
+ // orchestrator observed one of those failures. Episodes are delimited by OrchestratorStarted events.
+ var failedTaskIds = new HashSet();
+ int failureEpisode = int.MaxValue;
+ int episode = -1;
+
+ foreach (TableEntity entity in historyEntities)
{
- if (entity.GetString(nameof(OrchestrationInstance.ExecutionId)) != executionId)
+ string eventType = entity.GetString(nameof(HistoryEvent.EventType));
+ if (eventType == nameof(EventType.OrchestratorStarted))
{
- // the remaining entities are from a previous generation and can be discarded.
- break;
+ episode++;
+ }
+ else if (eventType == nameof(EventType.TaskFailed) || eventType == nameof(EventType.SubOrchestrationInstanceFailed))
+ {
+ failedTaskIds.Add(entity.GetInt32(nameof(TaskFailedEvent.TaskScheduledId)).GetValueOrDefault());
+ failureEpisode = Math.Min(failureEpisode, episode);
}
+ }
+ // Anything the orchestrator scheduled from the failure episode onwards was scheduled only because the orchestrator
+ // observed the failure (for example an activity invoked from a catch block, or the delay timer that RetryInterceptor
+ // creates between attempts). After the rewind the failure is no longer visible, so the replayed orchestrator can never
+ // reach those sequence IDs and would fail with a NonDeterministicOrchestrationException. Note that this deliberately
+ // errs on the side of removing too much: within the failure episode we cannot tell a consequence of the failure apart
+ // from unrelated work that happened to be batched into the same work item without replaying the orchestrator code. The
+ // cost is that a small number of already-successful tasks may be re-executed; the benefit is a history that replays.
+ var consequenceTaskIds = new HashSet();
+ episode = -1;
+
+ foreach (TableEntity entity in historyEntities)
+ {
+ string eventType = entity.GetString(nameof(HistoryEvent.EventType));
+ if (eventType == nameof(EventType.OrchestratorStarted))
+ {
+ episode++;
+ }
+ else if (episode >= failureEpisode && IsScheduledEventType(eventType))
+ {
+ consequenceTaskIds.Add(entity.GetInt32(nameof(HistoryEvent.EventId)).GetValueOrDefault());
+ }
+ }
+
+ // Decide what to do with each entity before touching the table, so that no entity is written twice.
+ var entitiesToClear = new List();
+ var failedSubOrchestrationEntities = new List();
+
+ foreach (TableEntity entity in historyEntities)
+ {
if (entity.RowKey == SentinelRowKey)
{
continue;
}
- int? taskScheduledId = entity.GetInt32(nameof(TaskCompletedEvent.TaskScheduledId));
+ if (entity.GetString(nameof(OrchestrationInstance.ExecutionId)) != executionId)
+ {
+ // the entity is from a previous generation and can be discarded.
+ continue;
+ }
- var eventFilterBuilder = new StringBuilder();
- eventFilterBuilder.Append($"{partitionFilter}");
- eventFilterBuilder.Append($" and {executionIdFilter}");
- eventFilterBuilder.Append($" and {nameof(HistoryEvent.EventId)} eq {taskScheduledId.GetValueOrDefault()}");
+ string eventType = entity.GetString(nameof(HistoryEvent.EventType));
+ int eventId = entity.GetInt32(nameof(HistoryEvent.EventId)).GetValueOrDefault();
- switch (entity.GetString(nameof(HistoryEvent.EventType)))
+ // the failure events themselves, including the failed ExecutionCompleted event
+ if (eventType == nameof(EventType.TaskFailed)
+ || eventType == nameof(EventType.SubOrchestrationInstanceFailed)
+ || entity.GetString(nameof(ExecutionCompletedEvent.OrchestrationStatus)) == nameof(OrchestrationStatus.Failed))
{
- // delete TaskScheduled corresponding to TaskFailed event
- case nameof(EventType.TaskFailed):
- eventFilterBuilder.Append($" and {nameof(HistoryEvent.EventType)} eq '{nameof(EventType.TaskScheduled)}'");
- IReadOnlyList taskScheduledEntities = await this.QueryHistoryAsync(eventFilterBuilder.ToString(), instanceId, cancellationToken);
-
- TableEntity tsEntity = taskScheduledEntities[0];
- tsEntity[nameof(TaskFailedEvent.Reason)] = "Rewound: " + tsEntity.GetString(nameof(HistoryEvent.EventType));
- tsEntity[nameof(TaskFailedEvent.EventType)] = nameof(EventType.GenericEvent);
- await this.HistoryTable.ReplaceEntityAsync(tsEntity, tsEntity.ETag, cancellationToken);
- break;
-
- // delete SubOrchestratorCreated corresponding to SubOrchestraionInstanceFailed event
- case nameof(EventType.SubOrchestrationInstanceFailed):
- hasFailedSubOrchestrations = true;
+ entitiesToClear.Add(entity);
+ }
- eventFilterBuilder.Append($" and {nameof(HistoryEvent.EventType)} eq '{nameof(EventType.SubOrchestrationInstanceCreated)}'");
- IReadOnlyList subOrchesratrationEntities = await this.QueryHistoryAsync(eventFilterBuilder.ToString(), instanceId, cancellationToken);
+ // the TaskScheduled event corresponding to a TaskFailed event, so that the task gets rescheduled
+ else if (eventType == nameof(EventType.TaskScheduled) && failedTaskIds.Contains(eventId))
+ {
+ entitiesToClear.Add(entity);
+ }
- // the SubOrchestrationCreatedEvent is still healthy and will not be overwritten, just marked as rewound
- TableEntity soEntity = subOrchesratrationEntities[0];
- soEntity[nameof(SubOrchestrationInstanceFailedEvent.Reason)] = "Rewound: " + soEntity.GetString(nameof(HistoryEvent.EventType));
- await this.HistoryTable.ReplaceEntityAsync(soEntity, soEntity.ETag, cancellationToken);
+ // the SubOrchestrationInstanceCreated event corresponding to a SubOrchestrationInstanceFailed event is still
+ // healthy and will not be overwritten, just marked as rewound, so long as it is not itself being removed
+ else if (eventType == nameof(EventType.SubOrchestrationInstanceCreated)
+ && failedTaskIds.Contains(eventId)
+ && !consequenceTaskIds.Contains(eventId))
+ {
+ failedSubOrchestrationEntities.Add(entity);
+ }
- // recursive call to clear out failure events on child instances
- await foreach (string childInstanceId in this.RewindHistoryAsync(soEntity.GetString(nameof(OrchestrationInstance.InstanceId)), cancellationToken))
- {
- yield return childInstanceId;
- }
+ // everything the orchestrator scheduled as a consequence of observing the failure
+ else if (IsScheduledEventType(eventType) && consequenceTaskIds.Contains(eventId))
+ {
+ entitiesToClear.Add(entity);
+ }
- break;
+ // and the results of those scheduled events. Leaving them behind would allow a stale result to satisfy a different
+ // task that gets assigned the same sequence ID once the orchestration resumes.
+ else if (TryGetCompletedTaskId(entity, eventType, out int completedTaskId) && consequenceTaskIds.Contains(completedTaskId))
+ {
+ entitiesToClear.Add(entity);
}
+ }
- // "clear" failure event by making RewindEvent: replay ignores row while dummy event preserves rowKey
+ foreach (TableEntity entity in entitiesToClear)
+ {
+ // "clear" the event by making it a GenericEvent: replay ignores the row while the dummy event preserves the rowKey
entity[nameof(TaskFailedEvent.Reason)] = "Rewound: " + entity.GetString(nameof(HistoryEvent.EventType));
entity[nameof(TaskFailedEvent.EventType)] = nameof(EventType.GenericEvent);
await this.HistoryTable.ReplaceEntityAsync(entity, entity.ETag, cancellationToken);
}
+ foreach (TableEntity entity in failedSubOrchestrationEntities)
+ {
+ entity[nameof(SubOrchestrationInstanceFailedEvent.Reason)] = "Rewound: " + entity.GetString(nameof(HistoryEvent.EventType));
+ await this.HistoryTable.ReplaceEntityAsync(entity, entity.ETag, cancellationToken);
+
+ // recursive call to clear out failure events on child instances
+ await foreach (string childInstanceId in this.RewindHistoryAsync(entity.GetString(nameof(OrchestrationInstance.InstanceId)), cancellationToken))
+ {
+ yield return childInstanceId;
+ }
+ }
+
// reset orchestration status in instance store table
await this.UpdateStatusForRewindAsync(instanceId, cancellationToken);
- if (!hasFailedSubOrchestrations)
+ if (failedSubOrchestrationEntities.Count == 0)
{
yield return instanceId;
}
}
+ ///
+ /// Determines whether the event type is one that replay matches against the orchestrator's sequence-ID counter. If one of
+ /// these events has no counterpart in the replayed execution, the orchestration fails with a
+ /// .
+ ///
+ static bool IsScheduledEventType(string eventType) =>
+ eventType == nameof(EventType.TaskScheduled)
+ || eventType == nameof(EventType.SubOrchestrationInstanceCreated)
+ || eventType == nameof(EventType.TimerCreated)
+ || eventType == nameof(EventType.EventSent);
+
+ ///
+ /// Gets the sequence ID of the scheduled task whose result the entity carries.
+ ///
+ static bool TryGetCompletedTaskId(TableEntity entity, string eventType, out int taskId)
+ {
+ if (eventType == nameof(EventType.TaskCompleted) || eventType == nameof(EventType.SubOrchestrationInstanceCompleted))
+ {
+ taskId = entity.GetInt32(nameof(TaskCompletedEvent.TaskScheduledId)).GetValueOrDefault();
+ return true;
+ }
+
+ if (eventType == nameof(EventType.TimerFired))
+ {
+ taskId = entity.GetInt32(nameof(TimerFiredEvent.TimerId)).GetValueOrDefault();
+ return true;
+ }
+
+ taskId = -1;
+ return false;
+ }
+
///
public override async IAsyncEnumerable GetStateAsync(string instanceId, bool allExecutions, bool fetchInput, [EnumeratorCancellation] CancellationToken cancellationToken = default)
{
diff --git a/src/DurableTask.Core/TaskOrchestrationDispatcher.cs b/src/DurableTask.Core/TaskOrchestrationDispatcher.cs
index 649e7b47..1247617a 100644
--- a/src/DurableTask.Core/TaskOrchestrationDispatcher.cs
+++ b/src/DurableTask.Core/TaskOrchestrationDispatcher.cs
@@ -551,7 +551,7 @@ protected async Task OnProcessWorkItemAsync(TaskOrchestrationWorkItem work
isCompleted = !continuedAsNew;
break;
case OrchestratorActionType.RewindOrchestration:
- this.ProcessRewindOrchestrationDecision(
+ ProcessRewindOrchestrationDecision(
runtimeState,
out List subOrchestrationRewindMessages,
out OrchestrationRuntimeState newRuntimeState);
@@ -1379,7 +1379,57 @@ TaskMessage ProcessSendEventDecision(
};
}
- void ProcessRewindOrchestrationDecision(
+ ///
+ /// Gets the sequence ID of a history event that the orchestrator matches against its own
+ /// sequence-ID counter during replay. If one of these events has no counterpart in the
+ /// replayed execution, throws a
+ /// .
+ ///
+ static bool TryGetScheduledTaskId(HistoryEvent historyEvent, out int taskId)
+ {
+ switch (historyEvent)
+ {
+ case TaskScheduledEvent taskScheduledEvent:
+ taskId = taskScheduledEvent.EventId;
+ return true;
+ case SubOrchestrationInstanceCreatedEvent subOrchestrationInstanceCreatedEvent:
+ taskId = subOrchestrationInstanceCreatedEvent.EventId;
+ return true;
+ case TimerCreatedEvent timerCreatedEvent:
+ taskId = timerCreatedEvent.EventId;
+ return true;
+ case EventSentEvent eventSentEvent:
+ taskId = eventSentEvent.EventId;
+ return true;
+ default:
+ taskId = -1;
+ return false;
+ }
+ }
+
+ ///
+ /// Gets the sequence ID of the scheduled task that a history event carries the result of.
+ ///
+ static bool TryGetCompletedTaskId(HistoryEvent historyEvent, out int taskId)
+ {
+ switch (historyEvent)
+ {
+ case TaskCompletedEvent taskCompletedEvent:
+ taskId = taskCompletedEvent.TaskScheduledId;
+ return true;
+ case SubOrchestrationInstanceCompletedEvent subOrchestrationInstanceCompletedEvent:
+ taskId = subOrchestrationInstanceCompletedEvent.TaskScheduledId;
+ return true;
+ case TimerFiredEvent timerFiredEvent:
+ taskId = timerFiredEvent.TimerId;
+ return true;
+ default:
+ taskId = -1;
+ return false;
+ }
+ }
+
+ internal static void ProcessRewindOrchestrationDecision(
OrchestrationRuntimeState runtimeState,
out List subOrchestrationRewindMessages,
out OrchestrationRuntimeState newRuntimeState)
@@ -1388,9 +1438,15 @@ void ProcessRewindOrchestrationDecision(
/* WARNING!!!:
* If any changes are made to how this method modifies the orchestration's history, then corresponding changes *must*
* be made in the backend implementations that rely on this method for executing a rewind.
+ *
+ * The rewind is "episode-aware": history is divided into episodes delimited by OrchestratorStartedEvent, and every
+ * task the orchestrator scheduled at or after the episode in which a failure was first observed is removed along with
+ * the failure itself. Backends replicating this logic must remove all four of the event types that replay matches
+ * against the orchestrator's sequence-ID counter (TaskScheduled, SubOrchestrationInstanceCreated, TimerCreated and
+ * EventSent), plus their result events. See AzureTableTrackingStore.RewindHistoryAsync for the equivalent
+ * implementation over Azure Table rows.
*/
- HashSet failedTaskIds = new();
subOrchestrationRewindMessages = new();
newRuntimeState = new()
@@ -1398,16 +1454,50 @@ void ProcessRewindOrchestrationDecision(
Status = runtimeState.Status
};
- // Determine the task IDs of the failed tasks and suborchestrations
+ // Determine the task IDs of the failed tasks and suborchestrations, along with the earliest episode in which the
+ // orchestrator observed one of those failures.
+ HashSet failedTaskIds = new();
+ int failureEpisode = int.MaxValue;
+ int episode = -1;
+
foreach (var evt in runtimeState.Events)
{
- if (evt is TaskFailedEvent taskFailedEvent)
+ if (evt is OrchestratorStartedEvent)
+ {
+ episode++;
+ }
+ else if (evt is TaskFailedEvent taskFailedEvent)
{
failedTaskIds.Add(taskFailedEvent.TaskScheduledId);
+ failureEpisode = Math.Min(failureEpisode, episode);
}
else if (evt is SubOrchestrationInstanceFailedEvent subOrchestrationInstanceFailedEvent)
{
failedTaskIds.Add(subOrchestrationInstanceFailedEvent.TaskScheduledId);
+ failureEpisode = Math.Min(failureEpisode, episode);
+ }
+ }
+
+ // Anything the orchestrator scheduled from the failure episode onwards was scheduled only because the orchestrator
+ // observed the failure (for example an activity invoked from a catch block, or the delay timer that
+ // RetryInterceptor creates between attempts). After the rewind the failure is no longer visible, so the replayed
+ // orchestrator can never reach those sequence IDs and would fail with a NonDeterministicOrchestrationException.
+ // Note that this deliberately errs on the side of removing too much: within the failure episode we cannot tell a
+ // consequence of the failure apart from unrelated work that happened to be batched into the same work item without
+ // replaying the orchestrator code, which the rewind path (and the backends replicating it) cannot do. The cost is
+ // that a small number of already-successful tasks may be re-executed; the benefit is a history that always replays.
+ HashSet consequenceTaskIds = new();
+ episode = -1;
+
+ foreach (var evt in runtimeState.Events)
+ {
+ if (evt is OrchestratorStartedEvent)
+ {
+ episode++;
+ }
+ else if (episode >= failureEpisode && TryGetScheduledTaskId(evt, out int scheduledTaskId))
+ {
+ consequenceTaskIds.Add(scheduledTaskId);
}
}
@@ -1417,74 +1507,87 @@ void ProcessRewindOrchestrationDecision(
// Copy the existing history, removing the failed task/suborchestration events and generating rewind events for each of the failed suborchestrations.
foreach (var evt in runtimeState.Events)
{
- // Do not add the TaskScheduledEvents for the failed tasks so that they get rescheduled, and do not add any of
- // the failed task/suborchestration/execution events to the new history.
- if (!(evt is TaskScheduledEvent taskScheduledEvent && failedTaskIds.Contains(taskScheduledEvent.EventId))
- && evt is not TaskFailedEvent
- && evt is not SubOrchestrationInstanceFailedEvent
- && evt is not ExecutionCompletedEvent)
+ // Do not add any of the failed task/suborchestration/execution events to the new history.
+ if (evt is TaskFailedEvent || evt is SubOrchestrationInstanceFailedEvent || evt is ExecutionCompletedEvent)
{
- HistoryEvent eventToAdd = evt;
+ continue;
+ }
- if (evt is ExecutionStartedEvent executionStartedEvent)
- {
- // Copy all information from the old ExecutionStartedEvent except for the ExecutionId, since we create a new one
- var newExecutionStartedEvent = new ExecutionStartedEvent(executionStartedEvent);
- newExecutionStartedEvent.OrchestrationInstance.ExecutionId = newExecutionId;
+ // Do not add the TaskScheduledEvents for the failed tasks so that they get rescheduled, and do not add anything
+ // the orchestrator scheduled as a consequence of observing the failure.
+ if (TryGetScheduledTaskId(evt, out int taskId)
+ && (consequenceTaskIds.Contains(taskId) || (evt is TaskScheduledEvent && failedTaskIds.Contains(taskId))))
+ {
+ continue;
+ }
- // If this is a suborchestration, we also need to update the ParentInstance's ExecutionId to match the new ExecutionId of the rewinding parent orchestration
- if (!string.IsNullOrEmpty(executionRewoundEvent.ParentExecutionId))
- {
- newExecutionStartedEvent.ParentInstance.OrchestrationInstance.ExecutionId = executionRewoundEvent.ParentExecutionId;
- }
- eventToAdd = newExecutionStartedEvent;
+ // Do not add the results of the removed tasks either. Leaving them behind would allow a stale result to satisfy
+ // a different task that gets assigned the same sequence ID once the orchestration resumes.
+ if (TryGetCompletedTaskId(evt, out int completedTaskId) && consequenceTaskIds.Contains(completedTaskId))
+ {
+ continue;
+ }
+
+ HistoryEvent eventToAdd = evt;
+
+ if (evt is ExecutionStartedEvent executionStartedEvent)
+ {
+ // Copy all information from the old ExecutionStartedEvent except for the ExecutionId, since we create a new one
+ var newExecutionStartedEvent = new ExecutionStartedEvent(executionStartedEvent);
+ newExecutionStartedEvent.OrchestrationInstance.ExecutionId = newExecutionId;
+
+ // If this is a suborchestration, we also need to update the ParentInstance's ExecutionId to match the new ExecutionId of the rewinding parent orchestration
+ if (!string.IsNullOrEmpty(executionRewoundEvent.ParentExecutionId))
+ {
+ newExecutionStartedEvent.ParentInstance.OrchestrationInstance.ExecutionId = executionRewoundEvent.ParentExecutionId;
}
+ eventToAdd = newExecutionStartedEvent;
+ }
+
+ // For each of the failed suborchestrations we are keeping, generate a rewind event
+ else if (evt is SubOrchestrationInstanceCreatedEvent subOrchestrationInstanceCreatedEvent
+ && failedTaskIds.Contains(subOrchestrationInstanceCreatedEvent.EventId))
+ {
+ var childExecutionRewoundEvent = new ExecutionRewoundEvent(-1, executionRewoundEvent!.Reason)
+ {
+ ParentExecutionId = newExecutionId,
+ InstanceId = subOrchestrationInstanceCreatedEvent.InstanceId
+ };
- // For each of the failed suborchestrations, generate a rewind event
- else if (evt is SubOrchestrationInstanceCreatedEvent subOrchestrationInstanceCreatedEvent
- && failedTaskIds.Contains(subOrchestrationInstanceCreatedEvent.EventId))
- {
- var childExecutionRewoundEvent = new ExecutionRewoundEvent(-1, executionRewoundEvent!.Reason)
+ if (runtimeState.ExecutionStartedEvent.TryGetParentTraceContext(out ActivityContext parentTraceContext))
+ {
+ // We set a new client span ID here so that the execution of the rewound suborchestration is not tied to the
+ // old parent.
+ var newClientSpanId = ActivitySpanId.CreateRandom();
+ var newSubOrchestrationInstanceCreatedEvent = new SubOrchestrationInstanceCreatedEvent(subOrchestrationInstanceCreatedEvent)
{
- ParentExecutionId = newExecutionId,
- InstanceId = subOrchestrationInstanceCreatedEvent.InstanceId
+ ClientSpanId = newClientSpanId.ToString()
};
+ eventToAdd = newSubOrchestrationInstanceCreatedEvent;
+
+ ActivityContext childActivityContext = new(
+ parentTraceContext.TraceId,
+ newClientSpanId,
+ parentTraceContext.TraceFlags,
+ parentTraceContext.TraceState);
+ childExecutionRewoundEvent.SetParentTraceContext(childActivityContext);
+ }
- if (runtimeState.ExecutionStartedEvent.TryGetParentTraceContext(out ActivityContext parentTraceContext))
- {
- // We set a new client span ID here so that the execution of the rewound suborchestration is not tied to the
- // old parent.
- var newClientSpanId = ActivitySpanId.CreateRandom();
- var newSubOrchestrationInstanceCreatedEvent = new SubOrchestrationInstanceCreatedEvent(subOrchestrationInstanceCreatedEvent)
+ subOrchestrationRewindMessages.Add
+ (
+ new TaskMessage
{
- ClientSpanId = newClientSpanId.ToString()
- };
- eventToAdd = newSubOrchestrationInstanceCreatedEvent;
-
- ActivityContext childActivityContext = new(
- parentTraceContext.TraceId,
- newClientSpanId,
- parentTraceContext.TraceFlags,
- parentTraceContext.TraceState);
- childExecutionRewoundEvent.SetParentTraceContext(childActivityContext);
- }
-
- subOrchestrationRewindMessages.Add
- (
- new TaskMessage
+ Event = childExecutionRewoundEvent,
+ OrchestrationInstance = new OrchestrationInstance
{
- Event = childExecutionRewoundEvent,
- OrchestrationInstance = new OrchestrationInstance
- {
- InstanceId = subOrchestrationInstanceCreatedEvent.InstanceId
- },
- }
- );
- }
-
- // Finally, add the event to the new history
- newRuntimeState.AddEvent(eventToAdd);
+ InstanceId = subOrchestrationInstanceCreatedEvent.InstanceId
+ },
+ }
+ );
}
+
+ // Finally, add the event to the new history
+ newRuntimeState.AddEvent(eventToAdd);
}
// If this is a "terminal leaf" with no suborchestrations, we need to add an outbound message to it to force it to rerun.
diff --git a/test/DurableTask.AzureStorage.Tests/AzureStorageScenarioTests.cs b/test/DurableTask.AzureStorage.Tests/AzureStorageScenarioTests.cs
index 07b064ac..2b59166d 100644
--- a/test/DurableTask.AzureStorage.Tests/AzureStorageScenarioTests.cs
+++ b/test/DurableTask.AzureStorage.Tests/AzureStorageScenarioTests.cs
@@ -1438,6 +1438,79 @@ public async Task RewindActivityFail()
}
}
+ ///
+ /// End-to-end test which validates that an orchestration that schedules more work in response to an activity
+ /// failure can be rewound. Regression test for
+ /// https://github.com/Azure/azure-functions-durable-extension/issues/444.
+ ///
+ [TestMethod]
+ public async Task RewindActivityFailWithCleanupActivity()
+ {
+ using (TestOrchestrationHost host = TestHelpers.GetTestOrchestrationHost(enableExtendedSessions: true))
+ {
+ Activities.HelloFailCleanupActivity.ShouldFail = true;
+ await host.StartAsync();
+
+ string singletonInstanceId = $"Test_{Guid.NewGuid():N}";
+
+ var client = await host.StartOrchestrationAsync(
+ typeof(Orchestrations.SayHelloWithActivityFailAndCleanup),
+ input: "World",
+ instanceId: singletonInstanceId);
+
+ var statusFail = await client.WaitForCompletionAsync(TimeSpan.FromSeconds(30));
+
+ Assert.AreEqual(OrchestrationStatus.Failed, statusFail?.OrchestrationStatus);
+
+ Activities.HelloFailCleanupActivity.ShouldFail = false;
+
+ await client.RewindAsync("Rewind orchestrator that scheduled an activity from its catch block.");
+
+ var statusRewind = await client.WaitForCompletionAsync(TimeSpan.FromSeconds(30));
+
+ Assert.AreEqual(OrchestrationStatus.Completed, statusRewind?.OrchestrationStatus);
+ Assert.AreEqual("\"Hello, World!\"", statusRewind?.Output);
+
+ await host.StopAsync();
+ }
+ }
+
+ ///
+ /// End-to-end test which validates that an orchestration whose retried activity ultimately failed can be rewound.
+ /// The retry delay timers are also a consequence of the failure and must not survive the rewind.
+ ///
+ [TestMethod]
+ public async Task RewindActivityFailWithRetry()
+ {
+ using (TestOrchestrationHost host = TestHelpers.GetTestOrchestrationHost(enableExtendedSessions: true))
+ {
+ Activities.HelloFailRetryActivity.ShouldFail = true;
+ await host.StartAsync();
+
+ string singletonInstanceId = $"Test_{Guid.NewGuid():N}";
+
+ var client = await host.StartOrchestrationAsync(
+ typeof(Orchestrations.SayHelloWithActivityFailAndRetry),
+ input: "World",
+ instanceId: singletonInstanceId);
+
+ var statusFail = await client.WaitForCompletionAsync(TimeSpan.FromSeconds(60));
+
+ Assert.AreEqual(OrchestrationStatus.Failed, statusFail?.OrchestrationStatus);
+
+ Activities.HelloFailRetryActivity.ShouldFail = false;
+
+ await client.RewindAsync("Rewind orchestrator with a retried activity.");
+
+ var statusRewind = await client.WaitForCompletionAsync(TimeSpan.FromSeconds(60));
+
+ Assert.AreEqual(OrchestrationStatus.Completed, statusRewind?.OrchestrationStatus);
+ Assert.AreEqual("\"Hello, World!\"", statusRewind?.Output);
+
+ await host.StopAsync();
+ }
+ }
+
[TestMethod]
public async Task RewindMultipleActivityFail()
{
@@ -4303,6 +4376,38 @@ public override Task RunTask(OrchestrationContext context, string input)
}
}
+ [KnownType(typeof(Activities.HelloFailCleanupActivity))]
+ [KnownType(typeof(Activities.Hello))]
+ internal class SayHelloWithActivityFailAndCleanup : TaskOrchestration
+ {
+ public override async Task RunTask(OrchestrationContext context, string input)
+ {
+ try
+ {
+ return await context.ScheduleTask(typeof(Activities.HelloFailCleanupActivity), input);
+ }
+ catch (Exception)
+ {
+ // The cleanup activity exists only because the orchestrator observed the failure, so rewind has to
+ // remove it from the history. See https://github.com/Azure/azure-functions-durable-extension/issues/444.
+ await context.ScheduleTask(typeof(Activities.Hello), "Cleanup");
+ throw;
+ }
+ }
+ }
+
+ [KnownType(typeof(Activities.HelloFailRetryActivity))]
+ internal class SayHelloWithActivityFailAndRetry : TaskOrchestration
+ {
+ public override Task RunTask(OrchestrationContext context, string input)
+ {
+ // Retries leave delay timers in the history, including one created after the final failed attempt, which
+ // rewind has to remove along with the failed attempts themselves.
+ var retryOptions = new RetryOptions(TimeSpan.FromSeconds(1), maxNumberOfAttempts: 2);
+ return context.ScheduleWithRetry(typeof(Activities.HelloFailRetryActivity), retryOptions, input);
+ }
+ }
+
[KnownType(typeof(Activities.Multiply))]
internal class Factorial : TaskOrchestration
{
@@ -5087,6 +5192,44 @@ protected override string Execute(TaskContext context, string input)
}
}
+ internal class HelloFailCleanupActivity : TaskActivity
+ {
+ public static bool ShouldFail = true;
+ protected override string Execute(TaskContext context, string input)
+ {
+ if (string.IsNullOrEmpty(input))
+ {
+ throw new ArgumentNullException(nameof(input));
+ }
+
+ if (ShouldFail)
+ {
+ throw new Exception("Simulating unhandled activity function failure...");
+ }
+
+ return $"Hello, {input}!";
+ }
+ }
+
+ internal class HelloFailRetryActivity : TaskActivity
+ {
+ public static bool ShouldFail = true;
+ protected override string Execute(TaskContext context, string input)
+ {
+ if (string.IsNullOrEmpty(input))
+ {
+ throw new ArgumentNullException(nameof(input));
+ }
+
+ if (ShouldFail)
+ {
+ throw new Exception("Simulating unhandled activity function failure...");
+ }
+
+ return $"Hello, {input}!";
+ }
+ }
+
internal class HelloFailFanOut : TaskActivity
{
public static bool ShouldFail1 = true;