diff --git a/docs/features/sub-orchestrations.md b/docs/features/sub-orchestrations.md index 5305c580d..7b52bc51c 100644 --- a/docs/features/sub-orchestrations.md +++ b/docs/features/sub-orchestrations.md @@ -207,22 +207,46 @@ public class PhaseOrchestration : TaskOrchestration ### Auto-Generated IDs ```csharp -// ID is an automatically generated GUID +// The framework generates a distinct instance ID for this call var result = await context.CreateSubOrchestrationInstance( typeof(ChildOrchestration), input); ``` -### Custom IDs for Idempotency +### Custom IDs ```csharp -// Using custom ID ensures idempotency +// Choose an ID that is unique among concurrently running children var result = await context.CreateSubOrchestrationInstance( typeof(ChildOrchestration), instanceId: $"{context.OrchestrationInstance.InstanceId}:child:{input.ItemId}", input: input); ``` +### Detecting Duplicate Awaited Child IDs + +Concurrent sub-orchestration calls must use distinct instance IDs. Different names, versions, or inputs do not disambiguate the same ID, and an explicit ID does not merge calls or make their results idempotent. + +Core hosts can opt in to detection before starting the worker: + +```csharp +var worker = new TaskHubWorker(service) +{ + FailOnDuplicateSubOrchestrationInstanceIds = true +}; +await worker.StartAsync(); +``` + +The option defaults to `false` for compatibility. When enabled, a new awaited child start that conflicts with another pending awaited child in the same parent execution fails the **parent orchestration** with failure type `DuplicateSubOrchestrationInstanceId` and non-retriable failure details. The entire current decision batch is discarded before any of its activities, timers, events, or children are scheduled. Work already scheduled in previous batches is not cancelled. + +This is a terminal orchestration-level failure, not an exception that the offending orchestration can catch around an individual call. Use distinct IDs, omit explicit IDs, or await a child's completion or failure before reusing its ID. Comparison is ordinal and case-sensitive. + +Detection does not check fire-and-forget starts or uniqueness across parents. Existing duplicate history without a new conflicting start is not rejected, and already stranded orchestrations are not automatically repaired. This is a Core worker option; downstream hosts such as Azure Functions must separately expose and enable it in a release that includes this capability. + +The guard lazily indexes accepted child history once per runtime-state load, then maintains the pending index as events are accepted. Cold initialization reduces completed history before allocating index entries for the remaining pending children. Reused runtime states, including extended sessions, validate new batches without rescanning old history or copying all pending children. Cold loads and ordinary uncached orchestration replay still process history; this option does not eliminate those costs. Proposed actions are tracked separately until they are accepted into history, and a drained fan-out releases its index capacity. + +Custom middleware or providers that rewrite history should construct a new `OrchestrationRuntimeState`. If they instead directly edit `Events` or mutate accepted event IDs, child instance IDs, completion/failure task schedule IDs, or fire-and-forget tags, they must call `InvalidateSubOrchestrationInstanceIdIndex()` before the next guarded validation. Normal `AddEvent` calls maintain the index automatically. Arbitrary external mutations are not detected automatically, and invalidating this derived index does not repair other runtime-state metadata. + ### Naming Conventions ```csharp diff --git a/src/DurableTask.Core/OrchestrationRuntimeState.cs b/src/DurableTask.Core/OrchestrationRuntimeState.cs index ab7d76044..f65c72f4c 100644 --- a/src/DurableTask.Core/OrchestrationRuntimeState.cs +++ b/src/DurableTask.Core/OrchestrationRuntimeState.cs @@ -27,11 +27,17 @@ namespace DurableTask.Core public class OrchestrationRuntimeState { private OrchestrationStatus orchestrationStatus; + SubOrchestrationInstanceIdIndex? subOrchestrationInstanceIdIndex; /// /// List of all history events for this runtime state. /// Note that this list is frequently a combination of and , but not always. /// + /// + /// Use to append accepted events. Custom history rewriters should + /// construct a new runtime state, or call + /// after directly modifying this list or indexed fields of its events. + /// public IList Events { get; } /// @@ -210,6 +216,27 @@ public void AddEvent(HistoryEvent historyEvent) AddEvent(historyEvent, true); } + /// + /// Invalidates derived tracking used by the opt-in duplicate sub-orchestration instance ID guard. + /// + /// + /// Call this after directly editing , or changing an accepted child's event ID, + /// instance ID, fire-and-forget tags, or a completion/failure's task schedule ID in place. + /// Normal calls update the tracking automatically. + /// The next guarded child-start validation rebuilds it from . + /// This does not reconcile other runtime state; constructing a new runtime state is preferred + /// when rewriting history. + /// + public void InvalidateSubOrchestrationInstanceIdIndex() + { + this.subOrchestrationInstanceIdIndex = null; + } + + internal SubOrchestrationInstanceIdIndex GetSubOrchestrationInstanceIdIndex() + { + return this.subOrchestrationInstanceIdIndex ??= SubOrchestrationInstanceIdIndex.FromHistory(this.Events); + } + ExecutionStartedEvent GetExecutionStartedEventOrThrow() { ExecutionStartedEvent? executionStartedEvent = this.ExecutionStartedEvent; @@ -245,6 +272,7 @@ void AddEvent(HistoryEvent historyEvent, bool isNewEvent) } SetMarkerEvents(historyEvent); + this.subOrchestrationInstanceIdIndex?.AddEvent(historyEvent); } bool IsDuplicateEvent(HistoryEvent historyEvent) diff --git a/src/DurableTask.Core/SubOrchestrationInstanceIdIndex.cs b/src/DurableTask.Core/SubOrchestrationInstanceIdIndex.cs new file mode 100644 index 000000000..367f82b1f --- /dev/null +++ b/src/DurableTask.Core/SubOrchestrationInstanceIdIndex.cs @@ -0,0 +1,162 @@ +// ---------------------------------------------------------------------------------- +// 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. +// ---------------------------------------------------------------------------------- + +#nullable enable +namespace DurableTask.Core +{ + using System; + using System.Collections.Generic; + using DurableTask.Core.History; + + internal sealed class SubOrchestrationInstanceIdIndex + { + Dictionary? pendingTasks; + Dictionary? pendingInstances; + bool hadConcurrentChildren; + + internal static SubOrchestrationInstanceIdIndex FromHistory(IEnumerable history) + { + // Reduce completed history before allocating nodes: cold sequential replay needs + // only a small temporary map, not one linked node for every historical child. + Dictionary? pending = null; + foreach (HistoryEvent historyEvent in history) + { + switch (historyEvent) + { + case SubOrchestrationInstanceCreatedEvent created + when created.InstanceId != null && !OrchestrationTags.IsTaggedAsFireAndForget(created.Tags): + pending ??= new Dictionary(); + pending[created.EventId] = created.InstanceId; + break; + case SubOrchestrationInstanceCompletedEvent completed: + pending?.Remove(completed.TaskScheduledId); + break; + case SubOrchestrationInstanceFailedEvent failed: + pending?.Remove(failed.TaskScheduledId); + break; + } + } + + var index = new SubOrchestrationInstanceIdIndex(); + if (pending != null) + { + foreach (KeyValuePair child in pending) + { + index.AddPending(child.Key, child.Value); + } + } + + return index; + } + + internal bool TryGetPendingTaskId(string instanceId, out int taskId) + { + if (this.pendingInstances != null && this.pendingInstances.TryGetValue(instanceId, out PendingChild child)) + { + taskId = child.TaskId; + return true; + } + + taskId = default; + return false; + } + + internal void AddEvent(HistoryEvent historyEvent) + { + switch (historyEvent) + { + case SubOrchestrationInstanceCreatedEvent created + when created.InstanceId != null && !OrchestrationTags.IsTaggedAsFireAndForget(created.Tags): + this.AddPending(created.EventId, created.InstanceId); + break; + case SubOrchestrationInstanceCompletedEvent completed: + this.Remove(completed.TaskScheduledId); + break; + case SubOrchestrationInstanceFailedEvent failed: + this.Remove(failed.TaskScheduledId); + break; + } + } + + void AddPending(int taskId, string instanceId) + { + this.Remove(taskId); + this.pendingTasks ??= new Dictionary(); + this.pendingInstances ??= new Dictionary(StringComparer.Ordinal); + this.pendingInstances.TryGetValue(instanceId, out PendingChild previous); + var child = new PendingChild(taskId, instanceId) { Next = previous }; + if (previous != null) + { + previous.Previous = child; + } + + this.pendingTasks.Add(child.TaskId, child); + this.pendingInstances[child.InstanceId] = child; + this.hadConcurrentChildren |= this.pendingTasks.Count > 1; + } + + void Remove(int taskId) + { + if (this.pendingTasks == null || !this.pendingTasks.TryGetValue(taskId, out PendingChild child)) + { + return; + } + + if (child.Previous != null) + { + child.Previous.Next = child.Next; + } + else if (child.Next != null) + { + this.pendingInstances![child.InstanceId] = child.Next; + } + else + { + this.pendingInstances!.Remove(child.InstanceId); + } + + if (child.Next != null) + { + child.Next.Previous = child.Previous; + } + + this.pendingTasks.Remove(taskId); + if (this.pendingTasks.Count == 0 && this.hadConcurrentChildren) + { + // Release drained fan-out capacity, but reuse the small buffers for sequential children. + this.pendingTasks = null; + this.pendingInstances = null; + this.hadConcurrentChildren = false; + } + } + + // Legacy histories can contain several pending task IDs for one instance ID. + // Links allow removal of any matching task in O(1), without a HashSet per child. + sealed class PendingChild + { + internal PendingChild(int taskId, string instanceId) + { + this.TaskId = taskId; + this.InstanceId = instanceId; + } + + internal int TaskId { get; } + + internal string InstanceId { get; } + + internal PendingChild? Previous { get; set; } + + internal PendingChild? Next { get; set; } + } + } +} diff --git a/src/DurableTask.Core/SubOrchestrationInstanceIdValidator.cs b/src/DurableTask.Core/SubOrchestrationInstanceIdValidator.cs new file mode 100644 index 000000000..a788628d1 --- /dev/null +++ b/src/DurableTask.Core/SubOrchestrationInstanceIdValidator.cs @@ -0,0 +1,65 @@ +// ---------------------------------------------------------------------------------- +// 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. +// ---------------------------------------------------------------------------------- + +#nullable enable +namespace DurableTask.Core +{ + using System; + using System.Collections.Generic; + using DurableTask.Core.Command; + + internal static class SubOrchestrationInstanceIdValidator + { + internal static OrchestrationCompleteOrchestratorAction? GetFailure( + string parentInstanceId, + OrchestrationRuntimeState runtimeState, + IEnumerable decisions) + { + SubOrchestrationInstanceIdIndex? pendingInstances = null; + Dictionary? batchInstances = null; + foreach (OrchestratorAction decision in decisions) + { + if (decision is not CreateSubOrchestrationAction action + || action.InstanceId == null + || OrchestrationTags.IsTaggedAsFireAndForget(action.Tags)) + { + continue; + } + + pendingInstances ??= runtimeState.GetSubOrchestrationInstanceIdIndex(); + if (pendingInstances.TryGetPendingTaskId(action.InstanceId, out int priorTaskId) + || (batchInstances != null && batchInstances.TryGetValue(action.InstanceId, out priorTaskId))) + { + string message = $"Orchestration '{parentInstanceId}' attempted to start sub-orchestration " + + $"'{action.InstanceId}' with task ID {action.Id}, but task ID {priorTaskId} is still pending " + + "with the same instance ID. Use distinct instance IDs for concurrent sub-orchestrations, " + + "omit the instance ID to generate one automatically, or await completion before reusing an ID."; + + return new OrchestrationCompleteOrchestratorAction + { + Id = action.Id, + OrchestrationStatus = OrchestrationStatus.Failed, + Result = message, + FailureDetails = new FailureDetails("DuplicateSubOrchestrationInstanceId", message, null, null, true), + }; + } + + // Proposed actions are not accepted history: a rejected or split batch may never send them. + batchInstances ??= new Dictionary(StringComparer.Ordinal); + batchInstances.Add(action.InstanceId, action.Id); + } + + return null; + } + } +} diff --git a/src/DurableTask.Core/TaskHubWorker.cs b/src/DurableTask.Core/TaskHubWorker.cs index 65dfbb47e..a95d4c3a9 100644 --- a/src/DurableTask.Core/TaskHubWorker.cs +++ b/src/DurableTask.Core/TaskHubWorker.cs @@ -247,6 +247,20 @@ public TaskHubWorker( /// public ErrorPropagationMode ErrorPropagationMode { get; set; } + /// + /// Gets or sets whether to fail an orchestration that starts an awaited sub-orchestration + /// with the same instance ID as another pending awaited sub-orchestration in that execution. + /// + /// + /// Defaults to false for compatibility. Set this property before . + /// When enabled, a conflicting start fails the orchestration and discards its entire current + /// decision batch; it does not throw a catchable exception at the individual call site. + /// Instance IDs are compared ordinally and case-sensitively. Reuse after completion or failure + /// is allowed. Fire-and-forget starts and conflicts across different parents are not checked. + /// Existing duplicate history alone does not cause failure or repair stranded orchestrations. + /// + public bool FailOnDuplicateSubOrchestrationInstanceIds { get; set; } + /// /// Gets or sets the exception properties provider that extracts custom properties from exceptions /// when creating FailureDetails objects. @@ -303,7 +317,10 @@ public async Task StartAsync() this.logHelper, this.ErrorPropagationMode, this.versioningSettings, - this.ExceptionPropertiesProvider); + this.ExceptionPropertiesProvider) + { + FailOnDuplicateSubOrchestrationInstanceIds = this.FailOnDuplicateSubOrchestrationInstanceIds, + }; this.activityDispatcher = new TaskActivityDispatcher( this.orchestrationService, this.activityManager, diff --git a/src/DurableTask.Core/TaskOrchestrationDispatcher.cs b/src/DurableTask.Core/TaskOrchestrationDispatcher.cs index 649e7b47a..2c358c658 100644 --- a/src/DurableTask.Core/TaskOrchestrationDispatcher.cs +++ b/src/DurableTask.Core/TaskOrchestrationDispatcher.cs @@ -135,6 +135,8 @@ public async Task StopAsync(bool forced) /// public bool EntitiesEnabled { get; set; } + internal bool FailOnDuplicateSubOrchestrationInstanceIds { get; set; } + /// /// Method to get the next work item to process within supplied timeout /// @@ -467,6 +469,21 @@ protected async Task OnProcessWorkItemAsync(TaskOrchestrationWorkItem work } } + if (this.FailOnDuplicateSubOrchestrationInstanceIds) + { + OrchestrationCompleteOrchestratorAction? failure = + SubOrchestrationInstanceIdValidator.GetFailure( + runtimeState.OrchestrationInstance!.InstanceId, + runtimeState, + decisions); + if (failure != null) + { + // Validate the whole batch before creating any history or outbound messages, + // including when the provider would otherwise split the batch across episodes. + decisions = new[] { failure }; + } + } + this.logHelper.OrchestrationExecuted( runtimeState.OrchestrationInstance!, runtimeState.Name, diff --git a/test/DurableTask.AzureStorage.Tests/DuplicateSubOrchestrationInstanceIdTests.cs b/test/DurableTask.AzureStorage.Tests/DuplicateSubOrchestrationInstanceIdTests.cs new file mode 100644 index 000000000..71a55383a --- /dev/null +++ b/test/DurableTask.AzureStorage.Tests/DuplicateSubOrchestrationInstanceIdTests.cs @@ -0,0 +1,389 @@ +// ---------------------------------------------------------------------------------- +// 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.AzureStorage.Tests +{ + using System; + using System.Diagnostics; + using System.Linq; + using System.Threading; + using System.Threading.Tasks; + using Azure.Data.Tables; + using DurableTask.AzureStorage.Tracking; + using DurableTask.Core; + using DurableTask.Core.Exceptions; + using DurableTask.Core.History; + using Microsoft.VisualStudio.TestTools.UnitTesting; + using Newtonsoft.Json; + + [TestClass] + public class DuplicateSubOrchestrationInstanceIdTests + { + const string ChildInput = "private-child-input"; + const string ReleaseEvent = "release"; + const string ScheduleNextEvent = "schedule-next"; + static readonly TimeSpan TestTimeout = TimeSpan.FromSeconds(30); + + AzureStorageOrchestrationService service; + TaskHubWorker worker; + TaskHubClient client; + OrchestrationInstance parent; + string taskHubName; + bool workerStarted; + + public TestContext TestContext { get; set; } + + [DataTestMethod] + [DataRow(false)] + [DataRow(true)] + public async Task DuplicateAwaitedIds_SameBatch_FailsWithoutStartingAnyChild(bool extendedSessions) + { + await this.StartAsync(extendedSessions, "Duplicate"); + + // There is deliberately no release event: rejection must not depend on a child finishing. + await this.AssertDuplicateFailureAsync(); + HistoryEvent[] history = await this.GetHistoryAsync(this.parent); + Assert.AreEqual(0, history.OfType().Count()); + Assert.AreEqual(0, history.OfType().Count()); + Assert.AreEqual(1, history.OfType().Single().EventId); + + await this.StopWorkerAsync(); + var trackingStore = (AzureTableTrackingStore)this.service.TrackingStore; + TableEntity[] instances = await trackingStore.InstancesTable.ExecuteQueryAsync().ToArrayAsync(); + TableEntity[] persistedHistory = await trackingStore.HistoryTable.ExecuteQueryAsync().ToArrayAsync(); + CollectionAssert.AreEqual(new[] { this.parent.InstanceId }, instances.Select(e => e.PartitionKey).ToArray()); + Assert.IsTrue(persistedHistory.All(e => e.PartitionKey == this.parent.InstanceId), + "A rejected batch must not persist history for either a duplicate child or its distinct sibling."); + Assert.IsFalse(persistedHistory.Any(e => + e.TryGetValue("EventType", out object value) && value?.ToString() == nameof(EventType.SubOrchestrationInstanceCreated))); + Assert.IsNull(await this.client.GetOrchestrationStateAsync(this.parent.InstanceId + "-child")); + await this.AssertEmptyControlQueuesAsync(); + } + + [DataTestMethod] + [DataRow(false)] + [DataRow(true)] + public async Task DuplicateAwaitedIds_LaterEpisode_FailsWithoutReleasingPendingChild(bool extendedSessions) + { + await this.StartAsync(extendedSessions, "Pending"); + SubOrchestrationInstanceCreatedEvent first = await this.WaitForChildStartAsync(0); + OrchestrationState child = await this.WaitForRunningChildAsync(first.InstanceId, ChildInput + "-0"); + + await this.client.RaiseEventAsync(this.parent, ScheduleNextEvent, string.Empty); + await this.AssertDuplicateFailureAsync(); + await this.StopWorkerAsync(); + + HistoryEvent[] history = await this.GetHistoryAsync(this.parent); + Assert.AreEqual(1, history.OfType().Count(), + "The child committed in the earlier episode is retained, but the new duplicate must not be committed."); + Assert.AreEqual(0, history.OfType().Count()); + Assert.AreEqual(0, history.OfType().Count()); + Assert.AreEqual(1, history.OfType().Single().EventId); + + OrchestrationState remainingChild = await this.client.GetOrchestrationStateAsync(first.InstanceId); + Assert.AreEqual(OrchestrationStatus.Running, remainingChild.OrchestrationStatus); + Assert.AreEqual(child.OrchestrationInstance.ExecutionId, remainingChild.OrchestrationInstance.ExecutionId); + HistoryEvent[] childHistory = await this.GetHistoryAsync(remainingChild.OrchestrationInstance); + Assert.AreEqual(1, childHistory.OfType().Count()); + Assert.AreEqual(0, childHistory.OfType().Count(), + "The pending child must never receive a release event in this regression."); + await this.AssertEmptyControlQueuesAsync(); + } + + [DataTestMethod] + [DataRow(false, "Distinct")] + [DataRow(true, "Distinct")] + [DataRow(false, "Automatic")] + [DataRow(true, "Automatic")] + [DataRow(false, "Sequential")] + [DataRow(true, "Sequential")] + [DataRow(false, "SequentialAfterFailure")] + [DataRow(true, "SequentialAfterFailure")] + public async Task AllowedAwaitedIds_WithGuardEnabled_Completes(bool extendedSessions, string scenario) + { + await this.StartAsync(extendedSessions, scenario); + bool firstChildFails = scenario == "SequentialAfterFailure"; + SubOrchestrationInstanceCreatedEvent first = await this.WaitForChildStartAsync(0); + OrchestrationState firstChild = await this.WaitForRunningChildAsync( + first.InstanceId, firstChildFails ? "fail" : ChildInput + "-0"); + await this.client.RaiseEventAsync(firstChild.OrchestrationInstance, ReleaseEvent, string.Empty); + + SubOrchestrationInstanceCreatedEvent second = await this.WaitForChildStartAsync(1); + OrchestrationState secondChild = await this.WaitForRunningChildAsync(second.InstanceId, ChildInput + "-1"); + await this.client.RaiseEventAsync(secondChild.OrchestrationInstance, ReleaseEvent, string.Empty); + + OrchestrationState state = await this.client.WaitForOrchestrationAsync(this.parent, TestTimeout); + Assert.IsNotNull(state, "The valid parent did not finish after both children were released."); + Assert.AreEqual(OrchestrationStatus.Completed, state.OrchestrationStatus, state.Output); + Assert.AreEqual(JsonConvert.SerializeObject( + (firstChildFails ? "caught" : ChildInput + "-0") + "," + ChildInput + "-1"), state.Output); + + HistoryEvent[] history = await this.GetHistoryAsync(this.parent); + SubOrchestrationInstanceCreatedEvent[] starts = history.OfType().ToArray(); + Assert.AreEqual(2, starts.Length); + int[] completedTaskIds = history.OfType() + .Select(e => e.TaskScheduledId).ToArray(); + int[] failedTaskIds = history.OfType() + .Select(e => e.TaskScheduledId).ToArray(); + CollectionAssert.AreEqual(firstChildFails ? new[] { 1 } : new[] { 0, 1 }, completedTaskIds); + CollectionAssert.AreEqual(firstChildFails ? new[] { 0 } : Array.Empty(), failedTaskIds); + + if (scenario.StartsWith("Sequential", StringComparison.Ordinal)) + { + Assert.AreEqual(starts[0].InstanceId, starts[1].InstanceId); + Assert.AreNotEqual(firstChild.OrchestrationInstance.ExecutionId, secondChild.OrchestrationInstance.ExecutionId); + } + else + { + Assert.AreNotEqual(starts[0].InstanceId, starts[1].InstanceId); + if (scenario == "Distinct") + { + Assert.IsTrue(string.Equals(starts[0].InstanceId, starts[1].InstanceId, StringComparison.OrdinalIgnoreCase), + "Distinct IDs that differ only by case must retain ordinal identity."); + } + } + } + + [TestCleanup] + public async Task Cleanup() + { + if (this.service == null) + { + return; + } + + try + { + await this.StopWorkerAsync(); + var trackingStore = (AzureTableTrackingStore)this.service.TrackingStore; + this.TestContext.WriteLine("Task hub: " + this.taskHubName); + if (this.parent != null) + { + this.TestContext.WriteLine("Parent history: " + await this.client.GetOrchestrationHistoryAsync(this.parent)); + } + + this.TestContext.WriteLine("Persisted instances: " + JsonConvert.SerializeObject( + await trackingStore.InstancesTable.ExecuteQueryAsync().ToArrayAsync())); + this.TestContext.WriteLine("Persisted history: " + JsonConvert.SerializeObject( + await trackingStore.HistoryTable.ExecuteQueryAsync().ToArrayAsync())); + } + finally + { + try + { + await this.service.DeleteAsync(); + } + finally + { + this.worker?.Dispose(); + } + } + } + + async Task StartAsync(bool extendedSessions, string scenario) + { + this.taskHubName = "duplicateids" + Guid.NewGuid().ToString("N"); + AzureStorageOrchestrationServiceSettings settings = + TestHelpers.GetTestAzureStorageOrchestrationServiceSettings(extendedSessions, extendedSessionTimeoutInSeconds: 5); + settings.TaskHubName = this.taskHubName; + settings.PartitionCount = 1; + settings.MaxQueuePollingInterval = TimeSpan.FromMilliseconds(100); + this.service = new AzureStorageOrchestrationService(settings); + this.client = new TaskHubClient(this.service); + this.worker = new TaskHubWorker(this.service) + { + FailOnDuplicateSubOrchestrationInstanceIds = true, + }; + this.worker.AddTaskOrchestrations(typeof(DuplicateIdParent), typeof(EventGatedChild)); + await this.service.CreateAsync(); + await this.worker.StartAsync(); + this.workerStarted = true; + this.parent = await this.client.CreateOrchestrationInstanceAsync( + typeof(DuplicateIdParent), "parent-" + Guid.NewGuid().ToString("N"), scenario); + } + + async Task AssertDuplicateFailureAsync() + { + OrchestrationState state = await this.client.WaitForOrchestrationAsync(this.parent, TestTimeout); + Assert.IsNotNull(state, "The duplicate-ID parent hung instead of failing without a child release."); + Assert.AreEqual(OrchestrationStatus.Failed, state.OrchestrationStatus, state.Output); + // AzureStorage exposes the failure message in state, but retains typed details in history. + HistoryEvent[] history = await this.GetHistoryAsync(this.parent); + FailureDetails failure = history.OfType().Single().FailureDetails; + Assert.IsNotNull(failure); + Assert.AreEqual("DuplicateSubOrchestrationInstanceId", failure.ErrorType); + Assert.IsTrue(failure.IsNonRetriable); + Assert.AreEqual(failure.ToString(), state.Output); + StringAssert.Contains(failure.ErrorMessage, this.parent.InstanceId); + StringAssert.Contains(failure.ErrorMessage, this.parent.InstanceId + "-child"); + StringAssert.Contains(failure.ErrorMessage, "task ID 0"); + StringAssert.Contains(failure.ErrorMessage, "task ID 1"); + StringAssert.Contains(failure.ErrorMessage, "distinct instance IDs"); + Assert.IsFalse(failure.ErrorMessage.Contains(ChildInput), + "Duplicate-ID diagnostics must not disclose child input."); + this.TestContext.WriteLine("Failure: " + JsonConvert.SerializeObject(failure)); + } + + async Task GetHistoryAsync(OrchestrationInstance instance) + { + OrchestrationHistory history = await this.service.TrackingStore.GetHistoryEventsAsync( + instance.InstanceId, instance.ExecutionId, CancellationToken.None); + return history.Events.ToArray(); + } + + async Task WaitForChildStartAsync(int taskId) + { + Stopwatch stopwatch = Stopwatch.StartNew(); + do + { + HistoryEvent[] history = await this.GetHistoryAsync(this.parent); + SubOrchestrationInstanceCreatedEvent child = history.OfType() + .SingleOrDefault(e => e.EventId == taskId); + if (child != null) + { + return child; + } + + await Task.Delay(100); + } + while (stopwatch.Elapsed < TestTimeout); + + throw new TimeoutException($"Child task {taskId} was not persisted for parent {this.parent.InstanceId}."); + } + + async Task WaitForRunningChildAsync(string instanceId, string expectedInput) + { + Stopwatch stopwatch = Stopwatch.StartNew(); + do + { + OrchestrationState state = await this.client.GetOrchestrationStateAsync(instanceId); + if (state?.OrchestrationStatus == OrchestrationStatus.Running && + state.Input == JsonConvert.SerializeObject(expectedInput)) + { + return state; + } + + await Task.Delay(100); + } + while (stopwatch.Elapsed < TestTimeout); + + throw new TimeoutException($"Child {instanceId} did not reach its external-event wait."); + } + + async Task StopWorkerAsync() + { + if (this.workerStarted) + { + await this.worker.StopAsync(); + this.workerStarted = false; + } + } + + async Task AssertEmptyControlQueuesAsync() + { + var queues = this.service.AllControlQueues.ToArray(); + Assert.AreEqual(1, queues.Length, "The outbound queue assertion must inspect the task hub's actual control queue."); + foreach (var queue in queues) + { + int count = await queue.InnerQueue.GetApproximateMessagesCountAsync(); + this.TestContext.WriteLine($"Persisted queue {queue.Name}: {count} messages"); + Assert.AreEqual(0, count, "A rejected child start must not remain queued, including invisible messages."); + } + } + + public class DuplicateIdParent : TaskOrchestration + { + readonly TaskCompletionSource scheduleNext = new TaskCompletionSource(); + + public override async Task RunTask(OrchestrationContext context, string scenario) + { + string childId = context.OrchestrationInstance.InstanceId + "-child"; + bool firstChildFails = scenario == "SequentialAfterFailure"; + string firstInput = firstChildFails ? "fail" : ChildInput + "-0"; + Task first = scenario == "Automatic" + ? context.CreateSubOrchestrationInstance(typeof(EventGatedChild), firstInput) + : context.CreateSubOrchestrationInstance(typeof(EventGatedChild), childId, firstInput); + + if (scenario.StartsWith("Sequential", StringComparison.Ordinal)) + { + string firstResult; + try + { + firstResult = await first; + } + catch (SubOrchestrationFailedException) when (firstChildFails) + { + firstResult = "caught"; + } + + string secondResult = await context.CreateSubOrchestrationInstance( + typeof(EventGatedChild), childId, ChildInput + "-1"); + return firstResult + "," + secondResult; + } + + if (scenario == "Pending") + { + await this.scheduleNext.Task; + } + + Task second = scenario == "Automatic" + ? context.CreateSubOrchestrationInstance(typeof(EventGatedChild), ChildInput + "-1") + : context.CreateSubOrchestrationInstance( + typeof(EventGatedChild), + scenario == "Distinct" ? context.OrchestrationInstance.InstanceId + "-CHILD" : childId, + ChildInput + "-1"); + if (scenario == "Duplicate") + { + // This unrelated start must also be discarded with the invalid decision batch. + Task sibling = context.CreateSubOrchestrationInstance( + typeof(EventGatedChild), childId + "-sibling", ChildInput); + return string.Join(",", await Task.WhenAll(first, second, sibling)); + } + + return string.Join(",", await Task.WhenAll(first, second)); + } + + public override void OnEvent(OrchestrationContext context, string name, string input) + { + if (name == ScheduleNextEvent) + { + this.scheduleNext.TrySetResult(true); + } + } + } + + public class EventGatedChild : TaskOrchestration + { + readonly TaskCompletionSource release = new TaskCompletionSource(); + + public override async Task RunTask(OrchestrationContext context, string input) + { + await this.release.Task; + if (input == "fail") + { + throw new InvalidOperationException("Expected child failure before sequential ID reuse."); + } + + return input; + } + + public override void OnEvent(OrchestrationContext context, string name, string input) + { + if (name == ReleaseEvent) + { + this.release.TrySetResult(true); + } + } + } + } +} diff --git a/test/DurableTask.Core.Tests/DuplicateSubOrchestrationDispatcherTests.cs b/test/DurableTask.Core.Tests/DuplicateSubOrchestrationDispatcherTests.cs new file mode 100644 index 000000000..716aee564 --- /dev/null +++ b/test/DurableTask.Core.Tests/DuplicateSubOrchestrationDispatcherTests.cs @@ -0,0 +1,655 @@ +// ---------------------------------------------------------------------------------- +// 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.Concurrent; + using System.Collections.Generic; + using System.IO; + using System.Linq; + using System.Threading; + using System.Threading.Tasks; + using DurableTask.Core.Command; + using DurableTask.Core.Exceptions; + using DurableTask.Core.History; + using DurableTask.Core.Logging; + using DurableTask.Core.Middleware; + using DurableTask.Core.Tracing; + using DurableTask.Emulator; + using Microsoft.VisualStudio.TestTools.UnitTesting; + + [TestClass] + public class DuplicateSubOrchestrationDispatcherTests + { + [DataTestMethod] + [DataRow(ErrorPropagationMode.SerializeExceptions, false)] + [DataRow(ErrorPropagationMode.UseFailureDetails, false)] + [DataRow(ErrorPropagationMode.UseFailureDetails, true)] + public async Task RawDuplicateActionsFailBeforeSendingAnyMessages(ErrorPropagationMode errorMode, bool splitMessages) + { + using var service = new RecordingService { MaxMessages = splitMessages ? 1 : (int?)null }; + using var worker = new TaskHubWorker(service) + { + FailOnDuplicateSubOrchestrationInstanceIds = true, + ErrorPropagationMode = errorMode, + }; + worker.AddOrchestrationDispatcherMiddleware((context, next) => + { + var state = context.GetProperty(); + context.SetProperty(new OrchestratorExecutionResult + { + Actions = state.Name == "parent" + ? new OrchestratorAction[] + { + new ScheduleTaskOrchestratorAction { Id = 0, Name = "activity" }, + new CreateTimerOrchestratorAction { Id = 1, FireAt = DateTime.UtcNow.AddHours(1) }, + new SendEventOrchestratorAction + { + Id = 2, + Instance = new OrchestrationInstance { InstanceId = "recipient" }, + EventName = "signal", + }, + Child(3, "same-child"), + Child(4, "same-child"), + } + : Array.Empty(), + }); + return Task.CompletedTask; + }); + + await worker.StartAsync(); + try + { + Assert.IsTrue(worker.TaskOrchestrationDispatcher.FailOnDuplicateSubOrchestrationInstanceIds); + worker.FailOnDuplicateSubOrchestrationInstanceIds = false; + Assert.IsTrue(worker.TaskOrchestrationDispatcher.FailOnDuplicateSubOrchestrationInstanceIds, "The option is captured at startup."); + var client = new TaskHubClient(service); + OrchestrationInstance instance = await client.CreateOrchestrationInstanceAsync("parent", "", null); + Checkpoint checkpoint = await service.FirstCheckpointAsync(instance.InstanceId); + Assert.AreEqual(OrchestrationStatus.Failed, checkpoint.Status); + Assert.AreEqual(0, checkpoint.Messages.Length); + Assert.AreEqual(0, checkpoint.Events.OfType().Count()); + Assert.AreEqual(0, checkpoint.Events.OfType().Count()); + Assert.AreEqual(0, checkpoint.Events.OfType().Count()); + Assert.AreEqual(0, checkpoint.Events.OfType().Count()); + ExecutionCompletedEvent completed = checkpoint.Events.OfType().Single(); + Assert.AreEqual(4, completed.EventId); + AssertFailure(completed.FailureDetails); + StringAssert.Contains(completed.Result, instance.InstanceId); + StringAssert.Contains(completed.Result, "same-child"); + OrchestrationState persisted = await client.WaitForOrchestrationAsync(instance, TimeSpan.FromSeconds(10)); + Assert.IsNotNull(persisted); + Assert.AreEqual(OrchestrationStatus.Failed, persisted.OrchestrationStatus); + AssertFailure(persisted.FailureDetails); + } + finally + { + await worker.StopAsync(true); + } + } + + [TestMethod] + public async Task DefaultOffPreservesDuplicateStarts() + { + using var service = new RecordingService(); + using var worker = new TaskHubWorker(service); + Assert.IsFalse(worker.FailOnDuplicateSubOrchestrationInstanceIds); + worker.AddOrchestrationDispatcherMiddleware((context, next) => + { + context.SetProperty(new OrchestratorExecutionResult + { + Actions = context.GetProperty().Name == "parent" + ? new[] { Child(0, "same-child"), Child(1, "same-child") } + : Array.Empty(), + }); + return Task.CompletedTask; + }); + await worker.StartAsync(); + try + { + Assert.IsFalse(worker.TaskOrchestrationDispatcher.FailOnDuplicateSubOrchestrationInstanceIds); + var client = new TaskHubClient(service); + OrchestrationInstance instance = await client.CreateOrchestrationInstanceAsync("parent", "", null); + Checkpoint checkpoint = await service.FirstCheckpointAsync(instance.InstanceId); + Assert.AreEqual(OrchestrationStatus.Running, checkpoint.Status); + Assert.AreEqual(2, checkpoint.Events.OfType().Count()); + var starts = checkpoint.Messages.Select(m => m.Event).OfType().ToArray(); + Assert.AreEqual(2, starts.Length); + Assert.AreEqual(starts[0].OrchestrationInstance.InstanceId, starts[1].OrchestrationInstance.InstanceId); + Assert.AreNotEqual(starts[0].OrchestrationInstance.ExecutionId, starts[1].OrchestrationInstance.ExecutionId); + Assert.AreEqual(0, checkpoint.Events.OfType().Count()); + } + finally + { + await worker.StopAsync(true); + } + } + + [DataTestMethod] + [DataRow(false, false)] + [DataRow(true, false)] + [DataRow(false, true)] + [DataRow(true, true)] + public async Task PendingChildIsCheckedOnReplayAndResume(bool resume, bool completeFirstChild) + { + using var service = new RecordingService { ForwardCompletions = false }; + var pipeline = new DispatchMiddlewarePipeline(); + pipeline.Add((context, next) => + { + bool alreadyStarted = context.GetProperty() + .Events.OfType().Any(); + context.SetProperty(new OrchestratorExecutionResult + { + Actions = new[] { Child(alreadyStarted ? 1 : 0, "same-child") }, + }); + return Task.CompletedTask; + }); + var dispatcher = new TestDispatcher(service, pipeline); + TaskOrchestrationWorkItem workItem = NewWorkItem(); + Assert.IsFalse(await dispatcher.ProcessAsync(workItem)); + Assert.IsNotNull(workItem.Cursor); + + Advance(workItem, resume, completeFirstChild + ? (HistoryEvent)new SubOrchestrationInstanceCompletedEvent(-1, 0, "done") + : new EventRaisedEvent(-1, null) { Name = "probe" }); + Assert.AreEqual(!completeFirstChild, await dispatcher.ProcessAsync(workItem)); + Checkpoint checkpoint = service.Checkpoints.Last(); + if (completeFirstChild) + { + Assert.AreEqual(OrchestrationStatus.Running, checkpoint.Status); + Assert.AreEqual(1, checkpoint.Messages.Length); + Assert.AreEqual(2, checkpoint.Events.OfType().Count()); + } + else + { + Assert.AreEqual(OrchestrationStatus.Failed, checkpoint.Status); + Assert.AreEqual(0, checkpoint.Messages.Length); + Assert.AreEqual(1, checkpoint.Events.OfType().Count()); + AssertFailure(checkpoint.Events.OfType().Single().FailureDetails); + } + } + + [DataTestMethod] + [DataRow(false)] + [DataRow(true)] + public async Task ContinueAsNewUsesOnlyNewGenerationHistory(bool duplicateInNewGeneration) + { + using var service = new RecordingService { ForwardCompletions = false }; + var pipeline = new DispatchMiddlewarePipeline(); + pipeline.Add((context, next) => + { + OrchestrationRuntimeState state = context.GetProperty(); + context.SetProperty(new OrchestratorExecutionResult + { + Actions = state.Input == "next" + ? (duplicateInNewGeneration + ? new[] { Child(0, "same-child"), Child(1, "same-child") } + : new[] { Child(0, "same-child") }) + : new OrchestratorAction[] + { + new OrchestrationCompleteOrchestratorAction + { + Id = 1, OrchestrationStatus = OrchestrationStatus.ContinuedAsNew, Result = "next", + }, + }, + }); + return Task.CompletedTask; + }); + var dispatcher = new TestDispatcher(service, pipeline); + TaskOrchestrationWorkItem workItem = NewWorkItem(); + workItem.OrchestrationRuntimeState = new OrchestrationRuntimeState(new[] + { + workItem.NewMessages[0].Event, + new SubOrchestrationInstanceCreatedEvent(0) { InstanceId = "same-child" }, + }); + workItem.NewMessages = new[] { Message(workItem, new EventRaisedEvent(-1, null) { Name = "continue" }) }; + string previousExecutionId = workItem.OrchestrationRuntimeState.OrchestrationInstance.ExecutionId; + Assert.IsNotNull(SubOrchestrationInstanceIdValidator.GetFailure( + "parent-id", workItem.OrchestrationRuntimeState, new[] { Child(1, "same-child") })); + + Assert.AreEqual(duplicateInNewGeneration, await dispatcher.ProcessAsync(workItem)); + Checkpoint checkpoint = service.Checkpoints.Single(); + Assert.AreNotEqual(previousExecutionId, workItem.OrchestrationRuntimeState.OrchestrationInstance.ExecutionId); + Assert.AreEqual(duplicateInNewGeneration ? OrchestrationStatus.Failed : OrchestrationStatus.Running, checkpoint.Status); + Assert.AreEqual(duplicateInNewGeneration ? 0 : 1, checkpoint.Messages.Length); + Assert.AreEqual(duplicateInNewGeneration ? 0 : 1, checkpoint.Events.OfType().Count()); + } + + [TestMethod] + public async Task RewindRebuildsPendingIndexFromRewrittenHistory() + { + using var service = new RecordingService { ForwardCompletions = false }; + var dispatcher = new TestDispatcher(service, new DispatchMiddlewarePipeline()); + TaskOrchestrationWorkItem workItem = NewWorkItem(); + var started = (ExecutionStartedEvent)workItem.NewMessages[0].Event; + started.ParentTraceContext = new DistributedTraceContext("00-0123456789abcdef0123456789abcdef-0123456789abcdef-01"); + workItem.OrchestrationRuntimeState = new OrchestrationRuntimeState(new HistoryEvent[] + { + started, + new SubOrchestrationInstanceCreatedEvent(0) { InstanceId = "rewound-child" }, + new SubOrchestrationInstanceFailedEvent(-1, 0, "child failed", null), + new ExecutionCompletedEvent(1, "failed", OrchestrationStatus.Failed), + }); + OrchestrationRuntimeState failedState = workItem.OrchestrationRuntimeState; + Assert.IsNull(SubOrchestrationInstanceIdValidator.GetFailure( + "parent-id", failedState, new[] { Child(2, "rewound-child") })); + workItem.NewMessages = new[] { Message(workItem, new ExecutionRewoundEvent(-1, "retry")) }; + Assert.IsTrue(await dispatcher.ProcessAsync(workItem)); + Assert.AreNotSame(failedState, workItem.OrchestrationRuntimeState); + Assert.AreEqual(OrchestrationStatus.Running, workItem.OrchestrationRuntimeState.OrchestrationStatus); + Assert.IsFalse(workItem.OrchestrationRuntimeState.Events.OfType().Any()); + Assert.IsNotNull(SubOrchestrationInstanceIdValidator.GetFailure( + "parent-id", workItem.OrchestrationRuntimeState, new[] { Child(2, "rewound-child") })); + Assert.IsTrue(service.Checkpoints.Single().Messages.Single().Event is ExecutionRewoundEvent); + } + + [TestMethod] + public async Task SplitBatchOnlyIndexesActuallyScheduledChildren() + { + using var service = new RecordingService { ForwardCompletions = false, MaxMessages = 1 }; + var pipeline = new DispatchMiddlewarePipeline(); + pipeline.Add((context, next) => + { + var state = context.GetProperty(); + context.SetProperty(new OrchestratorExecutionResult + { + Actions = state.Events.OfType().Any() + ? new[] { Child(1, "second") } + : new[] { Child(0, "first"), Child(1, "second") }, + }); + return Task.CompletedTask; + }); + var dispatcher = new TestDispatcher(service, pipeline); + TaskOrchestrationWorkItem workItem = NewWorkItem(); + Assert.IsTrue(await dispatcher.ProcessAsync(workItem)); + Assert.AreEqual(1, workItem.OrchestrationRuntimeState.Events.OfType().Count()); + Assert.IsNull(SubOrchestrationInstanceIdValidator.GetFailure( + "parent-id", workItem.OrchestrationRuntimeState, new[] { Child(1, "second") })); + service.MaxMessages = null; + Advance(workItem, true, new TimerFiredEvent(-1) { TimerId = FrameworkConstants.FakeTimerIdToSplitDecision }); + Assert.IsFalse(await dispatcher.ProcessAsync(workItem)); + Assert.AreEqual(2, workItem.OrchestrationRuntimeState.Events.OfType().Count()); + Assert.AreEqual(OrchestrationStatus.Running, service.Checkpoints.Last().Status); + } + + [TestMethod] + public async Task AbandonedCheckpointDoesNotPoisonRetriedWorkItem() + { + using var service = new RecordingService { FailNextCheckpoint = true }; + using var worker = new TaskHubWorker(service) { FailOnDuplicateSubOrchestrationInstanceIds = true }; + worker.AddOrchestrationDispatcherMiddleware((context, next) => + { + context.SetProperty(new OrchestratorExecutionResult + { + Actions = context.GetProperty().Name == "parent" + ? new OrchestratorAction[] + { + Child(0, "child"), + new OrchestrationCompleteOrchestratorAction { Id = 1, OrchestrationStatus = OrchestrationStatus.Completed }, + } + : Array.Empty(), + }); + return Task.CompletedTask; + }); + await worker.StartAsync(); + try + { + var client = new TaskHubClient(service); + OrchestrationInstance instance = await client.CreateOrchestrationInstanceAsync("parent", "", null); + OrchestrationState state = await client.WaitForOrchestrationAsync(instance, TimeSpan.FromSeconds(15)); + Assert.IsNotNull(state); + Assert.AreEqual(OrchestrationStatus.Completed, state.OrchestrationStatus, state.Output); + Assert.AreEqual(1, service.Abandonments); + Checkpoint checkpoint = await service.FirstCheckpointAsync(instance.InstanceId); + Assert.AreEqual(1, checkpoint.Events.OfType().Count()); + } + finally + { + await worker.StopAsync(true); + } + } + + [DataTestMethod] + [DataRow(false)] + [DataRow(true)] + public async Task SuspendedOrchestrationRejectsConflictWhenResumed(bool extendedSession) + { + using var service = new RecordingService { ForwardCompletions = false }; + var manager = new NameVersionObjectManager(); + manager.Add(new DefaultObjectCreator(typeof(EventDrivenParent))); + var dispatcher = new TestDispatcher(service, new DispatchMiddlewarePipeline(), manager); + TaskOrchestrationWorkItem workItem = NewWorkItem(NameVersionHelper.GetDefaultName(typeof(EventDrivenParent))); + Assert.IsFalse(await dispatcher.ProcessAsync(workItem)); + Advance(workItem, extendedSession, new ExecutionSuspendedEvent(-1, "pause")); + Assert.IsFalse(await dispatcher.ProcessAsync(workItem)); + Advance(workItem, extendedSession, new EventRaisedEvent(-1, null) { Name = "start-another" }); + Assert.IsFalse(await dispatcher.ProcessAsync(workItem)); + Checkpoint suspended = service.Checkpoints.Last(); + Assert.AreEqual(OrchestrationStatus.Suspended, suspended.Status); + Assert.AreEqual(0, suspended.Messages.Length); + + Advance(workItem, extendedSession, new ExecutionResumedEvent(-1, "resume")); + Assert.IsTrue(await dispatcher.ProcessAsync(workItem)); + Checkpoint failed = service.Checkpoints.Last(); + Assert.AreEqual(OrchestrationStatus.Failed, failed.Status); + Assert.AreEqual(0, failed.Messages.Length); + Assert.AreEqual(1, failed.Events.OfType().Count()); + } + + [TestMethod] + public async Task FailureNotifiesParentAndDoesNotRetryGuardedOrchestration() + { + using var service = new RecordingService(); + using var worker = new TaskHubWorker(service) + { + FailOnDuplicateSubOrchestrationInstanceIds = true, + ErrorPropagationMode = ErrorPropagationMode.UseFailureDetails, + }; + worker.AddTaskOrchestrations(typeof(RetryingParent)); + worker.AddOrchestrationDispatcherMiddleware((context, next) => + { + string name = context.GetProperty().Name; + if (name == "guarded-child" || name == "child") + { + context.SetProperty(new OrchestratorExecutionResult + { + Actions = name == "guarded-child" + ? new[] { Child(0, "grandchild"), Child(1, "grandchild") } + : Array.Empty(), + }); + return Task.CompletedTask; + } + + return next(); + }); + await worker.StartAsync(); + try + { + var client = new TaskHubClient(service); + OrchestrationInstance instance = await client.CreateOrchestrationInstanceAsync(typeof(RetryingParent), null); + OrchestrationState result = await client.WaitForOrchestrationAsync(instance, TimeSpan.FromSeconds(10)); + Assert.IsNotNull(result); + Assert.AreEqual(OrchestrationStatus.Completed, result.OrchestrationStatus); + Assert.AreEqual("\"caught non-retriable failure\"", result.Output); + Checkpoint guarded = await service.FirstCheckpointAsync("guarded-child-id"); + Assert.AreEqual(OrchestrationStatus.Failed, guarded.Status); + Assert.AreEqual(1, guarded.Messages.Length); + if (guarded.Messages[0].Event is not SubOrchestrationInstanceFailedEvent notification) + { + throw new AssertFailedException("Expected a sub-orchestration failure notification."); + } + Assert.AreEqual(0, notification.TaskScheduledId); + Assert.AreEqual(instance.InstanceId, guarded.Messages[0].OrchestrationInstance.InstanceId); + AssertFailure(notification.FailureDetails); + Assert.AreEqual(0, guarded.Events.OfType().Count()); + Checkpoint parent = service.Checkpoints.Last(c => c.InstanceId == instance.InstanceId); + Assert.AreEqual(1, parent.Events.OfType().Count()); + Assert.AreEqual(0, parent.Events.OfType().Count(), "A non-retriable failure must not schedule a retry."); + } + finally + { + await worker.StopAsync(true); + } + } + + [DataTestMethod] + [DataRow(false)] + [DataRow(true)] + public async Task SequentialAndRetryCallsCanReuseAnInstanceId(bool retryAfterFailure) + { + using var service = new RecordingService(); + using var worker = new TaskHubWorker(service) + { + FailOnDuplicateSubOrchestrationInstanceIds = true, + ErrorPropagationMode = ErrorPropagationMode.UseFailureDetails, + }; + int attempts = 0; + worker.AddTaskOrchestrations(typeof(ReuseParent)); + worker.AddOrchestrationDispatcherMiddleware((context, next) => + { + if (context.GetProperty().Name != "reusable-child") + { + return next(); + } + + bool fail = Interlocked.Increment(ref attempts) == 1 && retryAfterFailure; + context.SetProperty(new OrchestratorExecutionResult + { + Actions = new[] + { + new OrchestrationCompleteOrchestratorAction + { + Id = 0, + OrchestrationStatus = fail ? OrchestrationStatus.Failed : OrchestrationStatus.Completed, + Result = fail ? "transient failure" : "\"ok\"", + FailureDetails = fail ? new FailureDetails("Transient", "transient failure", null, null, false) : null, + }, + }, + }); + return Task.CompletedTask; + }); + await worker.StartAsync(); + try + { + var client = new TaskHubClient(service); + OrchestrationInstance instance = await client.CreateOrchestrationInstanceAsync(typeof(ReuseParent), retryAfterFailure); + OrchestrationState result = await client.WaitForOrchestrationAsync(instance, TimeSpan.FromSeconds(10)); + Assert.IsNotNull(result); + Assert.AreEqual(OrchestrationStatus.Completed, result.OrchestrationStatus); + Assert.AreEqual("\"ok\"", result.Output); + Assert.AreEqual(2, attempts); + Checkpoint checkpoint = service.Checkpoints.Last(c => c.InstanceId == instance.InstanceId); + var starts = checkpoint.Events.OfType().ToArray(); + Assert.AreEqual(2, starts.Length); + Assert.AreEqual(starts[0].InstanceId, starts[1].InstanceId); + Assert.AreNotEqual(starts[0].EventId, starts[1].EventId); + Assert.AreEqual(retryAfterFailure ? 1 : 0, checkpoint.Events.OfType().Count()); + } + finally + { + await worker.StopAsync(true); + } + } + + static void AssertFailure(FailureDetails failure) + { + Assert.IsNotNull(failure); + Assert.AreEqual("DuplicateSubOrchestrationInstanceId", failure.ErrorType); + Assert.IsTrue(failure.IsNonRetriable); + } + + static TaskOrchestrationWorkItem NewWorkItem(string name = "parent") + { + var instance = new OrchestrationInstance { InstanceId = "parent-id", ExecutionId = Guid.NewGuid().ToString("N") }; + return new TaskOrchestrationWorkItem + { + InstanceId = instance.InstanceId, + LockedUntilUtc = DateTime.MaxValue, + OrchestrationRuntimeState = new OrchestrationRuntimeState(), + NewMessages = new[] + { + new TaskMessage + { + OrchestrationInstance = instance, + Event = new ExecutionStartedEvent(-1, null) { Name = name, Version = "", OrchestrationInstance = instance }, + }, + }, + }; + } + + static TaskMessage Message(TaskOrchestrationWorkItem workItem, HistoryEvent historyEvent) + => new TaskMessage { OrchestrationInstance = workItem.OrchestrationRuntimeState.OrchestrationInstance, Event = historyEvent }; + + static void Advance(TaskOrchestrationWorkItem workItem, bool resume, HistoryEvent historyEvent) + { + if (resume) + { + workItem.OrchestrationRuntimeState.NewEvents.Clear(); + } + else + { + workItem.OrchestrationRuntimeState = new OrchestrationRuntimeState(workItem.OrchestrationRuntimeState.Events); + workItem.Cursor = null; + } + + workItem.NewMessages = new[] { Message(workItem, historyEvent) }; + } + + static CreateSubOrchestrationAction Child(int taskId, string instanceId) + => new CreateSubOrchestrationAction { Id = taskId, InstanceId = instanceId, Name = "child", Version = "" }; + + public class EventDrivenParent : TaskOrchestration + { + readonly TaskCompletionSource signal = new TaskCompletionSource(); + + public override async Task RunTask(OrchestrationContext context, string input) + { + Task first = context.CreateSubOrchestrationInstance("child", "", "same-child", null); + await this.signal.Task; + Task second = context.CreateSubOrchestrationInstance("child", "", "same-child", null); + await Task.WhenAll(first, second); + return "unreachable"; + } + + public override void OnEvent(OrchestrationContext context, string name, string input) + => this.signal.TrySetResult(true); + } + + public class RetryingParent : TaskOrchestration + { + public override async Task RunTask(OrchestrationContext context, string input) + { + try + { + return await context.CreateSubOrchestrationInstanceWithRetry( + "guarded-child", "", "guarded-child-id", new RetryOptions(TimeSpan.FromSeconds(1), 3), null); + } + catch (SubOrchestrationFailedException exception) when (exception.FailureDetails?.IsNonRetriable == true) + { + return "caught non-retriable failure"; + } + } + } + + public class ReuseParent : TaskOrchestration + { + public override async Task RunTask(OrchestrationContext context, bool retryAfterFailure) + { + if (retryAfterFailure) + { + return await context.CreateSubOrchestrationInstanceWithRetry( + "reusable-child", "", "reused-id", new RetryOptions(TimeSpan.FromMilliseconds(1), 3), null); + } + + await context.CreateSubOrchestrationInstance("reusable-child", "", "reused-id", null); + return await context.CreateSubOrchestrationInstance("reusable-child", "", "reused-id", null); + } + } + + sealed class TestDispatcher : TaskOrchestrationDispatcher + { + public TestDispatcher( + RecordingService service, + DispatchMiddlewarePipeline pipeline, + INameVersionObjectManager manager = null) + : base(service, manager ?? new NameVersionObjectManager(), pipeline, + new LogHelper(null), ErrorPropagationMode.UseFailureDetails, null, null) + { + this.FailOnDuplicateSubOrchestrationInstanceIds = true; + } + + public Task ProcessAsync(TaskOrchestrationWorkItem workItem) => this.OnProcessWorkItemAsync(workItem); + } + + sealed class Checkpoint + { + public string InstanceId { get; set; } + + public OrchestrationStatus Status { get; set; } + + public HistoryEvent[] Events { get; set; } + + public TaskMessage[] Messages { get; set; } + } + + sealed class RecordingService : LocalOrchestrationService, IOrchestrationService + { + readonly ConcurrentDictionary> firstCheckpoints = + new ConcurrentDictionary>(); + + public ConcurrentQueue Checkpoints { get; } = new ConcurrentQueue(); + + public bool ForwardCompletions { get; set; } = true; + + public int? MaxMessages { get; set; } + + public bool FailNextCheckpoint { get; set; } + + public int Abandonments { get; private set; } + + public new Task AbandonTaskOrchestrationWorkItemAsync(TaskOrchestrationWorkItem workItem) + { + this.Abandonments++; + return base.AbandonTaskOrchestrationWorkItemAsync(workItem); + } + + public new bool IsMaxMessageCountExceeded(int currentMessageCount, OrchestrationRuntimeState runtimeState) + => this.MaxMessages.HasValue + ? currentMessageCount >= this.MaxMessages.Value + : base.IsMaxMessageCountExceeded(currentMessageCount, runtimeState); + + public async Task FirstCheckpointAsync(string instanceId) + { + Task task = this.GetCheckpointSource(instanceId).Task; + Assert.AreSame(task, await Task.WhenAny(task, Task.Delay(TimeSpan.FromSeconds(10))), "No checkpoint was committed."); + return await task; + } + + public new async Task CompleteTaskOrchestrationWorkItemAsync( + TaskOrchestrationWorkItem workItem, + OrchestrationRuntimeState newOrchestrationRuntimeState, + IList outboundMessages, + IList orchestratorMessages, + IList timerMessages, + TaskMessage continuedAsNewMessage, + OrchestrationState state) + { + if (this.FailNextCheckpoint) + { + this.FailNextCheckpoint = false; + throw new IOException("Simulated checkpoint failure before persistence."); + } + + var checkpoint = new Checkpoint + { + InstanceId = workItem.InstanceId, + Status = newOrchestrationRuntimeState.OrchestrationStatus, + Events = newOrchestrationRuntimeState.Events.ToArray(), + Messages = outboundMessages.Concat(orchestratorMessages).Concat(timerMessages).ToArray(), + }; + this.Checkpoints.Enqueue(checkpoint); + if (this.ForwardCompletions) + { + await base.CompleteTaskOrchestrationWorkItemAsync( + workItem, newOrchestrationRuntimeState, outboundMessages, orchestratorMessages, timerMessages, continuedAsNewMessage, state); + } + this.GetCheckpointSource(workItem.InstanceId).TrySetResult(checkpoint); + } + + TaskCompletionSource GetCheckpointSource(string instanceId) + => this.firstCheckpoints.GetOrAdd( + instanceId, _ => new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously)); + } + } +} diff --git a/test/DurableTask.Core.Tests/SubOrchestrationInstanceIdValidatorTests.cs b/test/DurableTask.Core.Tests/SubOrchestrationInstanceIdValidatorTests.cs new file mode 100644 index 000000000..f134bd79f --- /dev/null +++ b/test/DurableTask.Core.Tests/SubOrchestrationInstanceIdValidatorTests.cs @@ -0,0 +1,491 @@ +// ---------------------------------------------------------------------------------- +// Copyright Microsoft Corporation +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// http://www.apache.org/licenses/LICENSE-2.0 +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. +// ---------------------------------------------------------------------------------- + +namespace DurableTask.Core.Tests +{ + using System; + using System.Collections.Generic; + using System.Linq; + using System.Reflection; + using System.Threading.Tasks; + using DurableTask.Core.Command; + using DurableTask.Core.History; + using DurableTask.Core.Serializing; + using DurableTask.Core.Settings; + using Microsoft.VisualStudio.TestTools.UnitTesting; + using Newtonsoft.Json; + + [TestClass] + public class SubOrchestrationInstanceIdValidatorTests + { + [TestMethod] + public void SameBatchDuplicateProducesActionableNonRetriableFailure() + { + CreateSubOrchestrationAction first = Child(3, "child-id"); + first.Name = "first-name"; + first.Version = "v1"; + first.Input = "private-first-input"; + CreateSubOrchestrationAction second = Child(7, "child-id"); + second.Name = "different-name"; + second.Version = "v2"; + second.Input = "private-second-input"; + + OrchestrationCompleteOrchestratorAction failure = Validate(Array.Empty(), first, second); + + Assert.IsNotNull(failure); + Assert.AreEqual(7, failure.Id); + Assert.AreEqual(OrchestrationStatus.Failed, failure.OrchestrationStatus); + Assert.AreEqual("DuplicateSubOrchestrationInstanceId", failure.FailureDetails.ErrorType); + Assert.IsTrue(failure.FailureDetails.IsNonRetriable); + Assert.AreEqual(failure.Result, failure.FailureDetails.ErrorMessage); + Assert.IsNull(failure.FailureDetails.InnerFailure); + Assert.IsNull(failure.FailureDetails.StackTrace); + StringAssert.Contains(failure.Result, "parent-id"); + StringAssert.Contains(failure.Result, "child-id"); + StringAssert.Contains(failure.Result, "task ID 3"); + StringAssert.Contains(failure.Result, "task ID 7"); + StringAssert.Contains(failure.Result, "distinct instance IDs"); + StringAssert.Contains(failure.Result, "automatically"); + StringAssert.Contains(failure.Result, "await completion"); + Assert.IsFalse(failure.Result.Contains("private-")); + } + + [TestMethod] + public void PendingChildFromPreviousEpisodeConflicts() + { + var history = new HistoryEvent[] { Created(2, "child-id"), new OrchestratorCompletedEvent(-1) }; + OrchestrationCompleteOrchestratorAction failure = Validate(history, Child(5, "child-id")); + Assert.IsNotNull(failure); + StringAssert.Contains(failure.Result, "task ID 2"); + StringAssert.Contains(failure.Result, "task ID 5"); + } + + [DataTestMethod] + [DataRow(false, false)] + [DataRow(true, false)] + [DataRow(false, true)] + [DataRow(true, true)] + public void CompletedOrFailedChildCanBeReused(bool failed, bool completionIsNew) + { + HistoryEvent completion = Completion(2, failed); + var runtimeState = new OrchestrationRuntimeState(new HistoryEvent[] { Created(2, "child-id") }); + if (completionIsNew) + { + runtimeState.AddEvent(completion); + Assert.IsFalse(runtimeState.PastEvents.Contains(completion)); + Assert.IsTrue(runtimeState.NewEvents.Contains(completion)); + } + else + { + runtimeState = new OrchestrationRuntimeState(new HistoryEvent[] { Created(2, "child-id"), completion }); + } + + Assert.IsNull(Validate(runtimeState.Events, Child(5, "child-id"))); + } + + [DataTestMethod] + [DataRow(false)] + [DataRow(true)] + public void DuplicateOldCompletionDoesNotRemoveNewerPendingChild(bool failed) + { + var history = new HistoryEvent[] + { + Created(2, "child-id"), + Completion(2, failed), + Created(5, "child-id"), + Completion(2, failed), + }; + OrchestrationCompleteOrchestratorAction failure = Validate(history, Child(8, "child-id")); + Assert.IsNotNull(failure); + StringAssert.Contains(failure.Result, "task ID 5"); + } + + [TestMethod] + public void LegacyDuplicatesOnlyRejectNewConflictingStarts() + { + var history = new HistoryEvent[] + { + Created(0, "child-id"), + Created(1, "child-id"), + Completion(1, false), + }; + Assert.IsNull(Validate(history)); + Assert.IsNull(Validate(history, new OrchestrationCompleteOrchestratorAction { OrchestrationStatus = OrchestrationStatus.Completed })); + Assert.IsNull(Validate(history, Child(2, "different-id"))); + Assert.IsNotNull(Validate(history, Child(2, "child-id"))); + } + + [TestMethod] + public void LegacyDuplicateIdsBecomeAvailableOnlyAfterAllTasksComplete() + { + var history = new List + { + Created(0, "child-id"), + Created(1, "child-id"), + Completion(0, false), + }; + Assert.IsNotNull(Validate(history, Child(2, "child-id"))); + history.Add(Completion(1, true)); + Assert.IsNull(Validate(history, Child(2, "child-id"))); + } + + [TestMethod] + public void DistinctAndCaseSensitiveIdsAreAllowed() + { + Assert.IsNull(Validate( + new[] { Created(0, "child-id") }, + Child(1, "CHILD-ID"), + Child(2, "other-id"))); + } + + [TestMethod] + public void AutomaticallyGeneratedIdsAreDistinct() + { + var context = new TaskOrchestrationContext( + new OrchestrationInstance { InstanceId = "parent-id", ExecutionId = "execution-id" }, + TaskScheduler.Default); + Task first = context.CreateSubOrchestrationInstance("child", "", null); + Task second = context.CreateSubOrchestrationInstance("child", "", null); + Assert.IsFalse(first.IsCompleted); + Assert.IsFalse(second.IsCompleted); + var actions = context.OrchestratorActions.Cast().ToArray(); + Assert.AreEqual(2, actions.Length); + Assert.AreNotEqual(actions[0].InstanceId, actions[1].InstanceId); + Assert.IsNull(Validate(Array.Empty(), actions)); + } + + [TestMethod] + public void FireAndForgetStartsAreExcludedInHistoryAndDecisions() + { + var tags = new Dictionary { { OrchestrationTags.FireAndForget, "" } }; + SubOrchestrationInstanceCreatedEvent previous = Created(0, "child-id"); + previous.Tags = tags; + CreateSubOrchestrationAction detached = Child(1, "child-id"); + detached.Tags = tags; + + Assert.IsNull(Validate(new[] { previous }, Child(2, "child-id"))); + Assert.IsNull(Validate(new[] { Created(0, "child-id") }, detached)); + Assert.IsNull(Validate(Array.Empty(), detached, Child(2, "child-id"))); + Assert.IsNull(Validate(Array.Empty(), Child(2, "child-id"), detached)); + Assert.IsNull(Validate(Array.Empty(), detached, detached)); + Assert.IsNotNull(Validate(Array.Empty(), detached, Child(2, "child-id"), Child(3, "child-id"))); + } + + [TestMethod] + public void NoAwaitedStartsDoesNotEnumerateHistory() + { + var tags = new CountingTags(); + var created = Created(0, "pending-child"); + created.Tags = tags; + var state = new OrchestrationRuntimeState(new[] { created }); + CreateSubOrchestrationAction detached = Child(1, "child-id"); + detached.Tags = new Dictionary { { OrchestrationTags.FireAndForget, "" } }; + Assert.IsNull(Validate(state)); + Assert.IsNull(Validate(state, detached, new CreateTimerOrchestratorAction { Id = 2 })); + Assert.AreEqual(0, tags.Reads); + Assert.IsNull(typeof(OrchestrationRuntimeState) + .GetField("subOrchestrationInstanceIdIndex", BindingFlags.NonPublic | BindingFlags.Instance).GetValue(state)); + } + + [TestMethod] + public void RepeatedValidationDoesNotReadHistoricalChildTagsAgain() + { + var tags = new CountingTags(); + var created = Created(0, "pending-child"); + created.Tags = tags; + var runtimeState = new OrchestrationRuntimeState(new[] { created }); + Assert.IsNull(Validate(runtimeState, Child(1, "other-child"))); + Assert.AreEqual(1, tags.Reads); + Assert.IsNull(Validate(runtimeState, Child(1, "other-child"))); + Assert.AreEqual(1, tags.Reads, "An unchanged runtime state must not rescan historical child tags."); + } + + [DataTestMethod] + [DataRow(false)] + [DataRow(true)] + public void AcceptedEventsUpdateAnInitializedIndex(bool failed) + { + var tags = new CountingTags(); + var created = Created(0, "child-id"); + created.Tags = tags; + var state = new OrchestrationRuntimeState(new[] { created }); + Assert.IsNotNull(Validate(state, Child(1, "child-id"))); + state.AddEvent(Completion(0, failed)); + Assert.IsNull(Validate(state, Child(1, "child-id"))); + state.AddEvent(Created(1, "child-id")); + state.AddEvent(Completion(0, failed)); + OrchestrationCompleteOrchestratorAction failure = Validate(state, Child(2, "child-id")); + Assert.IsNotNull(failure); + StringAssert.Contains(failure.Result, "task ID 1"); + Assert.AreEqual(1, tags.Reads, "Incremental updates must not rebuild prior history."); + } + + [DataTestMethod] + [DataRow(0, 1, 2)] + [DataRow(1, 2, 0)] + [DataRow(2, 0, 1)] + public void LegacyDuplicateCompletionsRemoveOnlyTheirMatchingTask(int first, int second, int last) + { + var state = new OrchestrationRuntimeState(new[] + { + Created(0, "child-id"), Created(1, "child-id"), Created(2, "child-id"), + }); + Assert.IsNotNull(Validate(state, Child(3, "child-id"))); + state.AddEvent(Completion(first, false)); + Assert.IsNotNull(Validate(state, Child(3, "child-id"))); + state.AddEvent(Completion(second, true)); + state.AddEvent(Completion(first, false)); + OrchestrationCompleteOrchestratorAction failure = Validate(state, Child(3, "child-id")); + Assert.IsNotNull(failure); + StringAssert.Contains(failure.Result, $"task ID {last}"); + state.AddEvent(Completion(last, false)); + Assert.IsNull(Validate(state, Child(3, "child-id"))); + } + + [TestMethod] + public void ProposedAndRejectedActionsNeverBecomeAcceptedHistory() + { + var state = new OrchestrationRuntimeState(); + Assert.IsNull(Validate(state, Child(0, "first"), Child(1, "second"))); + Assert.IsNull(Validate(state, Child(0, "first"), Child(1, "second"))); + Assert.AreEqual(0, state.Events.Count); + state.AddEvent(Created(0, "first")); + Assert.IsNull(Validate(state, Child(1, "second"))); + Assert.IsNotNull(Validate(state, Child(1, "second"), Child(2, "second"))); + Assert.IsNull(Validate(state, Child(1, "second"))); + Assert.AreEqual(1, state.Events.Count, "Unsent actions from a split or rejected batch must not poison the index."); + } + + [TestMethod] + public void ReloadedHistoryDoesNotReuseThePreviousRuntimeIndex() + { + var state = new OrchestrationRuntimeState(new[] { Created(0, "child-id") }); + Assert.IsNotNull(Validate(state, Child(1, "child-id"))); + var restored = new OrchestrationRuntimeState(state.Events); + state.AddEvent(Completion(0, false)); + Assert.IsNull(Validate(state, Child(1, "child-id"))); + Assert.IsNotNull(Validate(restored, Child(1, "child-id"))); + restored.AddEvent(Completion(0, true)); + Assert.IsNull(Validate(restored, Child(1, "child-id"))); + } + + [TestMethod] + public void SameCountHistoryReplacementIsRebuiltAfterInvalidation() + { + var state = new OrchestrationRuntimeState(new[] { Created(0, "old-id") }); + Assert.IsNotNull(Validate(state, Child(1, "old-id"))); + state.Events[0] = Created(0, "new-id"); + state.InvalidateSubOrchestrationInstanceIdIndex(); + Assert.IsNull(Validate(state, Child(1, "old-id"))); + Assert.IsNotNull(Validate(state, Child(1, "new-id"))); + state.Events.Clear(); + state.InvalidateSubOrchestrationInstanceIdIndex(); + Assert.IsNull(Validate(state, Child(1, "new-id"))); + } + + [TestMethod] + public void InPlaceIdentityAndAliasedTagChangesAreRebuiltAfterInvalidation() + { + var tags = new Dictionary(); + SubOrchestrationInstanceCreatedEvent created = Created(0, "old-id"); + created.Tags = tags; + var state = new OrchestrationRuntimeState(new[] { created }); + Assert.IsNotNull(Validate(state, Child(1, "old-id"))); + created.InstanceId = "new-id"; + created.EventId = 2; + state.InvalidateSubOrchestrationInstanceIdIndex(); + state.AddEvent(Completion(0, false)); + Assert.IsNull(Validate(state, Child(3, "old-id"))); + Assert.IsNotNull(Validate(state, Child(3, "new-id"))); + + tags.Add(OrchestrationTags.FireAndForget, ""); + state.InvalidateSubOrchestrationInstanceIdIndex(); + Assert.IsNull(Validate(state, Child(3, "new-id"))); + tags.Remove(OrchestrationTags.FireAndForget); + state.InvalidateSubOrchestrationInstanceIdIndex(); + Assert.IsNotNull(Validate(state, Child(3, "new-id"))); + state.AddEvent(Completion(2, true)); + Assert.IsNull(Validate(state, Child(3, "new-id"))); + } + + [DataTestMethod] + [DataRow(false)] + [DataRow(true)] + public void CompletionCorrelationReplacementIsRebuiltAfterInvalidation(bool failed) + { + HistoryEvent completion = Completion(0, failed); + var state = new OrchestrationRuntimeState(new[] + { + Created(0, "first"), Created(1, "second"), completion, + }); + Assert.IsNull(Validate(state, Child(2, "first"))); + Assert.IsNotNull(Validate(state, Child(2, "second"))); + state.Events[2] = Completion(1, failed); + state.InvalidateSubOrchestrationInstanceIdIndex(); + Assert.IsNotNull(Validate(state, Child(2, "first"))); + Assert.IsNull(Validate(state, Child(2, "second"))); + } + + [DataTestMethod] + [DataRow(TypeNameHandling.Objects)] + [DataRow(TypeNameHandling.Auto)] + [DataRow(TypeNameHandling.All)] + public void DerivedIndexDoesNotChangeSerializedHistoryOrRuntimeState(TypeNameHandling typeNameHandling) + { + var state = new OrchestrationRuntimeState(new HistoryEvent[] + { + new ExecutionStartedEvent(-1, null) + { + Name = "parent", + Version = "", + OrchestrationInstance = new OrchestrationInstance { InstanceId = "parent-id", ExecutionId = "execution-id" }, + }, + Created(0, "child-id"), + }); + Assert.IsInstanceOfType(state.Events, typeof(List)); + var converter = new JsonDataConverter(new JsonSerializerSettings { TypeNameHandling = typeNameHandling }); + string history = converter.Serialize(new OrchestrationSessionState(state.Events)); + string runtime = converter.Serialize(state); + Assert.IsNotNull(Validate(state, Child(1, "child-id"))); + Assert.AreEqual(history, converter.Serialize(new OrchestrationSessionState(state.Events))); + Assert.AreEqual(runtime, converter.Serialize(state)); + state.InvalidateSubOrchestrationInstanceIdIndex(); + Assert.AreEqual(runtime, converter.Serialize(state)); + } + + [DataTestMethod] + [DataRow(false)] + [DataRow(true)] + public void RepeatedCreatedTaskIdsReplaceRatherThanDuplicateMembership(bool initializeBeforeDuplicates) + { + var state = new OrchestrationRuntimeState(new[] + { + Created(0, "first"), Created(1, "first"), + }); + if (initializeBeforeDuplicates) + { + Assert.IsNotNull(Validate(state, Child(2, "first"))); + } + + state.AddEvent(Created(0, "first")); + state.AddEvent(Created(0, "first")); + Assert.IsNotNull(Validate(state, Child(2, "first"))); + state.AddEvent(Completion(0, false)); + OrchestrationCompleteOrchestratorAction failure = Validate(state, Child(2, "first")); + Assert.IsNotNull(failure); + StringAssert.Contains(failure.Result, "task ID 1"); + state.AddEvent(Created(1, "second")); + Assert.IsNull(Validate(state, Child(2, "first"))); + Assert.IsNotNull(Validate(state, Child(2, "second"))); + state.AddEvent(Completion(0, false)); + Assert.IsNotNull(Validate(state, Child(2, "second"))); + state.AddEvent(Completion(1, true)); + Assert.IsNull(Validate(state, Child(2, "second"))); + } + + [TestMethod] + public void DrainedFanoutReleasesHistoricalPeakDictionaryCapacity() + { + var state = new OrchestrationRuntimeState(); + Assert.IsNull(Validate(state, Child(1000, "next"))); + for (int i = 0; i < 1000; i++) + { + state.AddEvent(Created(i, "child-" + i)); + } + + SubOrchestrationInstanceIdIndex index = state.GetSubOrchestrationInstanceIdIndex(); + for (int i = 0; i < 1000; i++) + { + state.AddEvent(Completion(i, false)); + } + + Assert.AreSame(index, state.GetSubOrchestrationInstanceIdIndex()); + Assert.IsNull(typeof(SubOrchestrationInstanceIdIndex) + .GetField("pendingTasks", BindingFlags.NonPublic | BindingFlags.Instance).GetValue(index)); + Assert.IsNull(typeof(SubOrchestrationInstanceIdIndex) + .GetField("pendingInstances", BindingFlags.NonPublic | BindingFlags.Instance).GetValue(index)); + Assert.IsNull(Validate(state, Child(1000, "child-0"))); + } + + [TestMethod] + public void ColdCompletedSequentialHistoryDoesNotAllocatePendingNodeStorage() + { + var history = new List(); + for (int i = 0; i < 1000; i++) + { + history.Add(Created(i, "child-" + i)); + history.Add(Completion(i, false)); + } + + var state = new OrchestrationRuntimeState(history); + Assert.IsNull(Validate(state, Child(1000, "next"))); + SubOrchestrationInstanceIdIndex index = state.GetSubOrchestrationInstanceIdIndex(); + Assert.IsNull(typeof(SubOrchestrationInstanceIdIndex) + .GetField("pendingTasks", BindingFlags.NonPublic | BindingFlags.Instance).GetValue(index)); + Assert.IsNull(typeof(SubOrchestrationInstanceIdIndex) + .GetField("pendingInstances", BindingFlags.NonPublic | BindingFlags.Instance).GetValue(index)); + } + + [DataTestMethod] + [DataRow(false)] + [DataRow(true)] + public async Task RuntimeStreamRestoreRebuildsPendingChildren(bool compressed) + { + var state = new OrchestrationRuntimeState(new[] { Created(0, "child-id") }); + Assert.IsNotNull(Validate(state, Child(1, "child-id"))); + using var stream = await RuntimeStateStreamConverter.OrchestrationRuntimeStateToRawStream( + state, state, JsonDataConverter.Default, compressed, new SessionSettings(), null, "parent-id"); + var restored = await RuntimeStateStreamConverter.RawStreamToRuntimeState( + stream, "parent-id", null, JsonDataConverter.Default); + Assert.IsNotNull(Validate(restored, Child(1, "child-id"))); + restored.AddEvent(Completion(0, true)); + Assert.IsNull(Validate(restored, Child(1, "child-id"))); + Assert.IsNotNull(Validate(state, Child(1, "child-id"))); + } + + sealed class SessionSettings : ISessionSettings + { + public int SessionMaxSizeInBytes { get; set; } = 1024 * 1024; + + public int SessionOverflowThresholdInBytes { get; set; } = 1024 * 1024; + } + + sealed class CountingTags : Dictionary, IDictionary + { + public int Reads { get; private set; } + + bool IDictionary.ContainsKey(string key) + { + this.Reads++; + return base.ContainsKey(key); + } + } + + static OrchestrationCompleteOrchestratorAction Validate(IEnumerable history, params OrchestratorAction[] decisions) + => Validate(new OrchestrationRuntimeState(history.ToList()), decisions); + + static OrchestrationCompleteOrchestratorAction Validate(OrchestrationRuntimeState state, params OrchestratorAction[] decisions) + => SubOrchestrationInstanceIdValidator.GetFailure("parent-id", state, decisions); + + static CreateSubOrchestrationAction Child(int taskId, string instanceId) + => new CreateSubOrchestrationAction { Id = taskId, InstanceId = instanceId, Name = "child", Version = "" }; + + static SubOrchestrationInstanceCreatedEvent Created(int taskId, string instanceId) + => new SubOrchestrationInstanceCreatedEvent(taskId) { InstanceId = instanceId }; + + static HistoryEvent Completion(int taskId, bool failed) + => failed + ? (HistoryEvent)new SubOrchestrationInstanceFailedEvent(-1, taskId, "failure", null) + : new SubOrchestrationInstanceCompletedEvent(-1, taskId, "result"); + } +}