From 02bd5ef6c07da0252b664efcaaf6feabe50b1248 Mon Sep 17 00:00:00 2001 From: wangbill Date: Tue, 25 Aug 2026 15:44:53 -0400 Subject: [PATCH] Fix rewind non-determinism by scrubbing history episode-aware Rewind rebuilt the orchestration history by filtering on event type plus failed-task IDs. That removed the failed task, but kept everything the orchestrator scheduled *because* it observed the failure - an activity invoked from a catch block, a sub-orchestration, a sent event, or the delay timer RetryInterceptor always creates after the final failed attempt of ScheduleWithRetry. Those leftover scheduling events carry sequence IDs the replayed orchestrator can never reach, because after the rewind the failure is invisible and the orchestrator blocks awaiting the re-scheduled task. Replay then hits the orphan and throws NonDeterministicOrchestrationException, which TaskOrchestrationExecutor converts into a fail-orchestration action - so rewind appeared to "always return" the non-determinism error. Fixes Azure/azure-functions-durable-extension#444. The scrub is now episode-aware. History is divided into episodes delimited by OrchestratorStartedEvent; everything scheduled at or after the episode in which a failure was first observed is removed, along with the events carrying those results (a stale result could otherwise satisfy a different task assigned the same sequence ID). All four event types replay matches against the orchestrator's sequence-ID counter are covered: TaskScheduled, SubOrchestrationInstanceCreated, TimerCreated and EventSent. Fan-out/fan-in is unaffected: parallel branches are scheduled in an episode before the failure is observed, so they are retained. Failed sub-orchestrations are likewise created before the episode that delivers their failure, so their creation event is retained and the child rewind message is still emitted. Applied in both live rewind implementations: - TaskOrchestrationDispatcher.ProcessRewindOrchestrationDecision (SDK layer) - AzureTableTrackingStore.RewindHistoryAsync (Azure Storage) Out-of-repo backends that replicate the SDK-layer scrub server-side (e.g. the Durable Task Scheduler) must apply the same rule; the WARNING comment on ProcessRewindOrchestrationDecision now spells out the contract. Tests: new Test/DurableTask.Core.Tests/RewindTests.cs drives real orchestrations through real episodes, rewinds, and replays (8 cases; the 4 regression cases fail without this change). Two end-to-end scenario tests added for the Azure Storage path, both of which reproduce the issue-444 error without the fix. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- Test/DurableTask.Core.Tests/RewindTests.cs | 537 ++++++++++++++++++ .../Tracking/AzureTableTrackingStore.cs | 196 +++++-- .../TaskOrchestrationDispatcher.cs | 227 ++++++-- .../AzureStorageScenarioTests.cs | 143 +++++ 4 files changed, 990 insertions(+), 113 deletions(-) create mode 100644 Test/DurableTask.Core.Tests/RewindTests.cs 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;