diff --git a/src/DurableTask.AzureStorage/Tracking/AzureTableTrackingStore.cs b/src/DurableTask.AzureStorage/Tracking/AzureTableTrackingStore.cs index 167a3311..4540f2c3 100644 --- a/src/DurableTask.AzureStorage/Tracking/AzureTableTrackingStore.cs +++ b/src/DurableTask.AzureStorage/Tracking/AzureTableTrackingStore.cs @@ -48,6 +48,9 @@ class AzureTableTrackingStore : TrackingStoreBase const string SentinelRowKey = "sentinel"; const string IsCheckpointCompleteProperty = "IsCheckpointComplete"; const string CheckpointCompletedTimestampProperty = "CheckpointCompletedTimestamp"; + const string RewoundReasonPrefix = "Rewound: "; + const string RewoundExecutionCompletedReason = RewoundReasonPrefix + nameof(EventType.ExecutionCompleted); + const string RewoundSubOrchestrationFailedReason = RewoundReasonPrefix + nameof(EventType.SubOrchestrationInstanceFailed); // See https://docs.microsoft.com/en-us/rest/api/storageservices/understanding-the-table-service-data-model#property-types const int MaxTablePropertySizeInBytes = 60 * 1024; // 60KB to give buffer @@ -73,6 +76,74 @@ class AzureTableTrackingStore : TrackingStoreBase readonly IReadOnlyDictionary eventTypeMap; readonly MessageManager messageManager; + readonly struct RewindRecoveryEdge : IEquatable + { + public RewindRecoveryEdge( + string parentInstanceId, + string parentExecutionId, + int taskScheduleId, + string childInstanceId, + string childExecutionId) + { + this.ParentInstanceId = parentInstanceId; + this.ParentExecutionId = parentExecutionId; + this.TaskScheduleId = taskScheduleId; + this.ChildInstanceId = childInstanceId; + this.ChildExecutionId = childExecutionId; + } + + string ParentInstanceId { get; } + + string ParentExecutionId { get; } + + int TaskScheduleId { get; } + + string ChildInstanceId { get; } + + string ChildExecutionId { get; } + + public bool Equals(RewindRecoveryEdge other) + { + return string.Equals(this.ParentInstanceId, other.ParentInstanceId, StringComparison.Ordinal) && + string.Equals(this.ParentExecutionId, other.ParentExecutionId, StringComparison.Ordinal) && + this.TaskScheduleId == other.TaskScheduleId && + string.Equals(this.ChildInstanceId, other.ChildInstanceId, StringComparison.Ordinal) && + string.Equals(this.ChildExecutionId, other.ChildExecutionId, StringComparison.Ordinal); + } + + public override bool Equals(object obj) + { + return obj is RewindRecoveryEdge other && this.Equals(other); + } + + public override int GetHashCode() + { + unchecked + { + int hash = 17; + hash = (hash * 31) + StringComparer.Ordinal.GetHashCode(this.ParentInstanceId ?? string.Empty); + hash = (hash * 31) + StringComparer.Ordinal.GetHashCode(this.ParentExecutionId ?? string.Empty); + hash = (hash * 31) + this.TaskScheduleId; + hash = (hash * 31) + StringComparer.Ordinal.GetHashCode(this.ChildInstanceId ?? string.Empty); + hash = (hash * 31) + StringComparer.Ordinal.GetHashCode(this.ChildExecutionId ?? string.Empty); + return hash; + } + } + } + + sealed class RewindRecoveryState + { + public RewindRecoveryState(bool suppressesParent, IEnumerable emittedTargets) + { + this.SuppressesParent = suppressesParent; + this.EmittedTargets = new HashSet(emittedTargets, StringComparer.Ordinal); + } + + public bool SuppressesParent { get; set; } + + public HashSet EmittedTargets { get; } + } + public AzureTableTrackingStore( AzureStorageClient azureStorageClient, MessageManager messageManager) @@ -271,7 +342,24 @@ async Task> QueryHistoryAsync(string filter, string i return entities; } - public override async IAsyncEnumerable RewindHistoryAsync(string instanceId, [EnumeratorCancellation] CancellationToken cancellationToken = default) + public override IAsyncEnumerable RewindHistoryAsync( + string instanceId, + CancellationToken cancellationToken = default) + { + return this.RewindHistoryAsync( + instanceId, + rewindStartEntity: null, + resetCurrentInstance: true, + handledEdges: new Dictionary(), + cancellationToken); + } + + async IAsyncEnumerable RewindHistoryAsync( + string instanceId, + TableEntity rewindStartEntity, + bool resetCurrentInstance, + Dictionary handledEdges, + [EnumeratorCancellation] CancellationToken cancellationToken) { ////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////// // REWIND ALGORITHM: @@ -294,17 +382,35 @@ public override async IAsyncEnumerable RewindHistoryAsync(string instanc string executionId = recentStartRow[0].GetString(nameof(OrchestrationInstance.ExecutionId)); DateTime instanceTimestamp = recentStartRow[0].Timestamp.GetValueOrDefault().DateTime; + // Capture the instance version before changing history so the reset can fence a lagging + // failure write without overwriting a state advanced by another rewind. + rewindStartEntity ??= await this.GetInstanceEntityForRewindAsync(instanceId, cancellationToken); + EnsureRewindExecutionMatches(rewindStartEntity, instanceId, executionId); + ETag rewindStartETag = rewindStartEntity.ETag; + // 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(')'); + if (resetCurrentInstance) + { + 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( + $" or ({AzureTableQueryFilter.ColumnEquals(nameof(HistoryEvent.EventType), nameof(EventType.GenericEvent))}" + + $" and {AzureTableQueryFilter.ColumnEquals(nameof(SubOrchestrationInstanceFailedEvent.Reason), RewoundSubOrchestrationFailedReason)})"); + updateFilterBuilder.Append(')'); + } + else + { + updateFilterBuilder.Append( + $" and {AzureTableQueryFilter.ColumnEquals(nameof(HistoryEvent.EventType), nameof(EventType.GenericEvent))}" + + $" and {AzureTableQueryFilter.ColumnEquals(nameof(SubOrchestrationInstanceFailedEvent.Reason), RewoundSubOrchestrationFailedReason)}"); + } IReadOnlyList entitiesToClear = await this.QueryHistoryAsync(updateFilterBuilder.ToString(), instanceId, cancellationToken); foreach (TableEntity entity in entitiesToClear) @@ -321,6 +427,10 @@ public override async IAsyncEnumerable RewindHistoryAsync(string instanc } int? taskScheduledId = entity.GetInt32(nameof(TaskCompletedEvent.TaskScheduledId)); + string eventType = entity.GetString(nameof(HistoryEvent.EventType)); + bool isRewoundSubOrchestrationFailure = + eventType == nameof(EventType.GenericEvent) && + entity.GetString(nameof(SubOrchestrationInstanceFailedEvent.Reason)) == RewoundSubOrchestrationFailedReason; var eventFilterBuilder = new StringBuilder(); eventFilterBuilder.Append($"{partitionFilter}"); @@ -335,46 +445,267 @@ public override async IAsyncEnumerable RewindHistoryAsync(string instanc 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.Reason)] = + RewoundReasonPrefix + 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; - eventFilterBuilder.Append($" and {nameof(HistoryEvent.EventType)} eq '{nameof(EventType.SubOrchestrationInstanceCreated)}'"); IReadOnlyList subOrchesratrationEntities = await this.QueryHistoryAsync(eventFilterBuilder.ToString(), instanceId, cancellationToken); - // 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)); + string childInstanceId = soEntity.GetString(nameof(OrchestrationInstance.InstanceId)); + // The SubOrchestrationCreatedEvent is still healthy and will not be overwritten, just marked as rewound. + soEntity[nameof(SubOrchestrationInstanceFailedEvent.Reason)] = + RewoundReasonPrefix + soEntity.GetString(nameof(HistoryEvent.EventType)); await this.HistoryTable.ReplaceEntityAsync(soEntity, soEntity.ETag, cancellationToken); - // recursive call to clear out failure events on child instances - await foreach (string childInstanceId in this.RewindHistoryAsync(soEntity.GetString(nameof(OrchestrationInstance.InstanceId)), cancellationToken)) + TableEntity liveChildRewindStartEntity = + await this.GetInstanceEntityForRewindAsync(childInstanceId, cancellationToken); + var liveEdge = new RewindRecoveryEdge( + instanceId, + executionId, + taskScheduledId.GetValueOrDefault(), + childInstanceId, + liveChildRewindStartEntity.GetString(nameof(OrchestrationInstance.ExecutionId))); + handledEdges.TryGetValue(liveEdge, out RewindRecoveryState liveEdgeState); + var previouslyEmittedTargets = liveEdgeState == null + ? new HashSet(StringComparer.Ordinal) + : new HashSet(liveEdgeState.EmittedTargets, StringComparer.Ordinal); + + var liveTargets = new List(); + await foreach (string failedLeafInstanceId in this.RewindHistoryAsync( + childInstanceId, + liveChildRewindStartEntity, + resetCurrentInstance: true, + handledEdges, + cancellationToken)) + { + liveTargets.Add(failedLeafInstanceId); + } + + // A live child failure suppresses the parent even if its recursive branch + // unexpectedly produces no target, preserving the existing rewind behavior. + if (liveEdgeState == null) + { + liveEdgeState = new RewindRecoveryState( + suppressesParent: true, + emittedTargets: liveTargets); + handledEdges.Add(liveEdge, liveEdgeState); + } + else + { + liveEdgeState.SuppressesParent = true; + liveEdgeState.EmittedTargets.UnionWith(liveTargets); + } + + hasFailedSubOrchestrations = true; + foreach (string liveTarget in liveTargets) { - yield return childInstanceId; + if (!previouslyEmittedTargets.Contains(liveTarget)) + { + yield return liveTarget; + } } break; + + case nameof(EventType.GenericEvent) when isRewoundSubOrchestrationFailure: + if (!taskScheduledId.HasValue) + { + continue; + } + + eventFilterBuilder.Append($" and {nameof(HistoryEvent.EventType)} eq '{nameof(EventType.SubOrchestrationInstanceCreated)}'"); + IReadOnlyList rewoundSubOrchestrationEntities = + await this.QueryHistoryAsync(eventFilterBuilder.ToString(), instanceId, cancellationToken); + + TableEntity rewoundSubOrchestration = rewoundSubOrchestrationEntities[0]; + string rewoundChildInstanceId = + rewoundSubOrchestration.GetString(nameof(OrchestrationInstance.InstanceId)); + (TableEntity childRewindStartEntity, bool resetChildInstance) = + await this.GetRewindRecoveryContextAsync( + rewoundChildInstanceId, + instanceId, + executionId, + taskScheduledId.Value, + cancellationToken); + if (childRewindStartEntity == null) + { + continue; + } + + var recoveredEdge = new RewindRecoveryEdge( + instanceId, + executionId, + taskScheduledId.Value, + rewoundChildInstanceId, + childRewindStartEntity.GetString(nameof(OrchestrationInstance.ExecutionId))); + if (handledEdges.TryGetValue(recoveredEdge, out RewindRecoveryState recoveredEdgeState)) + { + hasFailedSubOrchestrations |= recoveredEdgeState.SuppressesParent; + continue; + } + + // Buffer this recovery branch so an active intermediate with no stranded + // descendants does not suppress reviving the current parent. + var recoveredTargets = new List(); + await foreach (string recoveredTarget in this.RewindHistoryAsync( + rewoundChildInstanceId, + childRewindStartEntity, + resetChildInstance, + handledEdges, + cancellationToken)) + { + recoveredTargets.Add(recoveredTarget); + } + + handledEdges.Add( + recoveredEdge, + new RewindRecoveryState( + suppressesParent: recoveredTargets.Count > 0, + emittedTargets: recoveredTargets)); + if (recoveredTargets.Count == 0) + { + continue; + } + + hasFailedSubOrchestrations = true; + foreach (string recoveredTarget in recoveredTargets) + { + yield return recoveredTarget; + } + + // Keep the existing marker idempotent so another ancestor retry can recover it. + continue; + } + + if (eventType == nameof(EventType.GenericEvent) && + entity.GetString(nameof(TaskFailedEvent.Reason))?.StartsWith(RewoundReasonPrefix, StringComparison.Ordinal) == true) + { + continue; } // "clear" failure event by making RewindEvent: replay ignores row while dummy event preserves rowKey - entity[nameof(TaskFailedEvent.Reason)] = "Rewound: " + entity.GetString(nameof(HistoryEvent.EventType)); + entity[nameof(TaskFailedEvent.Reason)] = RewoundReasonPrefix + eventType; entity[nameof(TaskFailedEvent.EventType)] = nameof(EventType.GenericEvent); await this.HistoryTable.ReplaceEntityAsync(entity, entity.ETag, cancellationToken); } - // reset orchestration status in instance store table - await this.UpdateStatusForRewindAsync(instanceId, cancellationToken); + if (resetCurrentInstance) + { + // reset orchestration status in instance store table + await this.UpdateStatusForRewindAsync(instanceId, executionId, rewindStartETag, cancellationToken); + + if (!hasFailedSubOrchestrations) + { + yield return instanceId; + } + } + } + + async Task<(TableEntity RewindStartEntity, bool ResetCurrentInstance)> GetRewindRecoveryContextAsync( + string instanceId, + string expectedParentInstanceId, + string expectedParentExecutionId, + int expectedTaskScheduleId, + CancellationToken cancellationToken) + { + string instanceFilter = + $"{AzureTableQueryFilter.PartitionKeyEquals(instanceId)} and " + + $"{AzureTableQueryFilter.ColumnEquals(RowKeyProperty, string.Empty)}"; + TableEntity instanceEntity = await this.InstancesTable + .ExecuteQueryAsync(instanceFilter, 1, cancellationToken: cancellationToken) + .FirstOrDefaultAsync(cancellationToken); + if (instanceEntity == null) + { + return default; + } + + string executionId = instanceEntity.GetString(nameof(OrchestrationInstance.ExecutionId)); + if (string.IsNullOrEmpty(executionId)) + { + return default; + } + + string runtimeStatus = instanceEntity.GetString("RuntimeStatus"); + bool isPending = + runtimeStatus == nameof(OrchestrationStatus.Pending) && + !instanceEntity.ContainsKey(OutputProperty); + bool isFailed = runtimeStatus == nameof(OrchestrationStatus.Failed); + bool isActive = + runtimeStatus == nameof(OrchestrationStatus.Running) || + runtimeStatus == nameof(OrchestrationStatus.Suspended); + if (!isPending && !isFailed && !isActive) + { + return default; + } + + string rewoundCompletionFilter = + $"{AzureTableQueryFilter.PartitionKeyEquals(instanceId)} and " + + $"{AzureTableQueryFilter.ColumnEquals(nameof(OrchestrationInstance.ExecutionId), executionId)} and " + + $"{AzureTableQueryFilter.ColumnEquals(nameof(HistoryEvent.EventType), nameof(EventType.GenericEvent))} and " + + $"{AzureTableQueryFilter.ColumnEquals(nameof(TaskFailedEvent.Reason), RewoundExecutionCompletedReason)} and " + + $"{AzureTableQueryFilter.ColumnEquals(nameof(ExecutionCompletedEvent.OrchestrationStatus), nameof(OrchestrationStatus.Failed))}"; + IReadOnlyList rewoundCompletions = + await this.QueryHistoryAsync(rewoundCompletionFilter, instanceId, cancellationToken); + if (rewoundCompletions.Count == 0) + { + return default; + } + + string executionStartedFilter = + $"{AzureTableQueryFilter.PartitionKeyEquals(instanceId)} and " + + $"{AzureTableQueryFilter.ColumnEquals(nameof(OrchestrationInstance.ExecutionId), executionId)} and " + + $"{AzureTableQueryFilter.ColumnEquals(nameof(HistoryEvent.EventType), nameof(EventType.ExecutionStarted))}"; + IReadOnlyList executionStartedEntities = + await this.QueryHistoryAsync(executionStartedFilter, instanceId, cancellationToken); + if (executionStartedEntities.Count != 1) + { + return default; + } + + var executionStarted = (ExecutionStartedEvent)TableEntityConverter.Deserialize( + executionStartedEntities[0], + typeof(ExecutionStartedEvent)); + ParentInstance parent = executionStarted.ParentInstance; + if (!string.Equals( + executionStarted.OrchestrationInstance?.InstanceId, + instanceId, + StringComparison.Ordinal) || + !string.Equals( + executionStarted.OrchestrationInstance?.ExecutionId, + executionId, + StringComparison.Ordinal) || + !string.Equals( + parent?.OrchestrationInstance?.InstanceId, + expectedParentInstanceId, + StringComparison.Ordinal) || + !string.Equals( + parent?.OrchestrationInstance?.ExecutionId, + expectedParentExecutionId, + StringComparison.Ordinal) || + parent?.TaskScheduleId != expectedTaskScheduleId) + { + return default; + } - if (!hasFailedSubOrchestrations) + string currentCompletionFilter = + $"{AzureTableQueryFilter.PartitionKeyEquals(instanceId)} and " + + $"{AzureTableQueryFilter.ColumnEquals(nameof(OrchestrationInstance.ExecutionId), executionId)} and " + + $"{AzureTableQueryFilter.ColumnEquals(nameof(HistoryEvent.EventType), nameof(EventType.ExecutionCompleted))}"; + IReadOnlyList currentCompletions = + await this.QueryHistoryAsync(currentCompletionFilter, instanceId, cancellationToken); + if (currentCompletions.Count > 0) { - yield return instanceId; + return default; } + + return (instanceEntity, !isActive); } /// @@ -854,17 +1185,43 @@ public override async Task SetNewExecutionAsync( } /// - public override async Task UpdateStatusForRewindAsync(string instanceId, CancellationToken cancellationToken = default) + public override async Task UpdateStatusForRewindAsync( + string instanceId, + string executionId, + ETag rewindStartETag, + CancellationToken cancellationToken = default) { - string sanitizedInstanceId = KeySanitation.EscapePartitionKey(instanceId); - TableEntity entity = new TableEntity(sanitizedInstanceId, "") + TableEntity entity = await this.GetInstanceEntityForRewindAsync(instanceId, cancellationToken); + EnsureRewindExecutionMatches(entity, instanceId, executionId); + + bool changedSinceRewindStarted = entity.ETag != rewindStartETag; + bool needsWriteFence = !changedSinceRewindStarted && IsPreFailureProjection(entity); + if (IsEquivalentRewindState(entity) && !needsWriteFence) { - ["RuntimeStatus"] = OrchestrationStatus.Pending.ToString("G"), - ["LastUpdatedTime"] = DateTime.UtcNow, - }; + return; + } + + // Merge cannot remove a table property, so replace the complete row using its current ETag. + entity.Remove(OutputProperty); + entity["RuntimeStatus"] = OrchestrationStatus.Pending.ToString("G"); + entity["LastUpdatedTime"] = DateTime.UtcNow; Stopwatch stopwatch = Stopwatch.StartNew(); - await this.InstancesTable.MergeEntityAsync(entity, ETag.All, cancellationToken); + try + { + await this.InstancesTable.ReplaceEntityAsync(entity, entity.ETag, cancellationToken); + } + catch (DurableTaskStorageException ex) when (ex.HttpStatusCode == (int)HttpStatusCode.PreconditionFailed) + { + TableEntity currentEntity = await this.GetInstanceEntityForRewindAsync(instanceId, cancellationToken); + EnsureRewindExecutionMatches(currentEntity, instanceId, executionId); + if (IsEquivalentRewindState(currentEntity)) + { + return; + } + + throw; + } // We don't have enough information to get the episode number. // It's also not important to have for this particular trace. @@ -874,12 +1231,64 @@ public override async Task UpdateStatusForRewindAsync(string instanceId, Cancell this.storageAccountName, this.taskHubName, instanceId, - string.Empty, + executionId, OrchestrationStatus.Pending, currentEpisodeNumber, stopwatch.ElapsedMilliseconds); } + async Task GetInstanceEntityForRewindAsync( + string instanceId, + CancellationToken cancellationToken) + { + string filter = $"{AzureTableQueryFilter.PartitionKeyEquals(instanceId)} and {AzureTableQueryFilter.ColumnEquals(RowKeyProperty, string.Empty)}"; + TableEntity entity = await this.InstancesTable + .ExecuteQueryAsync(filter, 1, cancellationToken: cancellationToken) + .FirstOrDefaultAsync(cancellationToken); + + return entity ?? + throw new DurableTaskStorageException($"The orchestration instance '{instanceId}' does not exist."); + } + + static void EnsureRewindExecutionMatches( + TableEntity entity, + string instanceId, + string expectedExecutionId) + { + string currentExecutionId = entity.GetString("ExecutionId"); + if (!string.Equals(currentExecutionId, expectedExecutionId, StringComparison.Ordinal)) + { + throw new DurableTaskStorageException( + $"Rewind conflict for orchestration instance '{instanceId}': expected execution " + + $"'{expectedExecutionId}', but the current execution is '{currentExecutionId ?? "(missing)"}'."); + } + } + + static bool IsPreFailureProjection(TableEntity entity) + { + return Enum.TryParse(entity.GetString("RuntimeStatus"), out OrchestrationStatus runtimeStatus) && + (runtimeStatus == OrchestrationStatus.Pending || + runtimeStatus == OrchestrationStatus.Running || + runtimeStatus == OrchestrationStatus.Suspended); + } + + static bool IsEquivalentRewindState(TableEntity entity) + { + if (!Enum.TryParse(entity.GetString("RuntimeStatus"), out OrchestrationStatus runtimeStatus)) + { + return false; + } + + if (runtimeStatus == OrchestrationStatus.Pending) + { + return !entity.ContainsKey(OutputProperty); + } + + // A competing rewind can advance beyond Pending before an ETag conflict is observed. + // Preserve that newer state and allow the caller to enqueue any targets it already found. + return runtimeStatus != OrchestrationStatus.Failed; + } + /// public override async Task UpdateStatusForTerminationAsync( string instanceId, diff --git a/src/DurableTask.AzureStorage/Tracking/ITrackingStore.cs b/src/DurableTask.AzureStorage/Tracking/ITrackingStore.cs index 6fc2aa8b..42ab1f05 100644 --- a/src/DurableTask.AzureStorage/Tracking/ITrackingStore.cs +++ b/src/DurableTask.AzureStorage/Tracking/ITrackingStore.cs @@ -161,8 +161,14 @@ interface ITrackingStore /// Used to update a state in the tracking store to pending whenever a rewind is initiated from the client /// /// The instance being rewound - /// The token to monitor for cancellation requests. The default value is . - Task UpdateStatusForRewindAsync(string instanceId, CancellationToken cancellationToken = default); + /// The execution being rewound + /// The instance ETag captured before the rewind changed history. + /// The token to monitor for cancellation requests. The default value is . + Task UpdateStatusForRewindAsync( + string instanceId, + string executionId, + ETag rewindStartETag, + CancellationToken cancellationToken = default); /// /// Used to update the instance status to "Terminated" when a pending orchestration is terminated. diff --git a/src/DurableTask.AzureStorage/Tracking/TrackingStoreBase.cs b/src/DurableTask.AzureStorage/Tracking/TrackingStoreBase.cs index d02a729c..cf3f29a1 100644 --- a/src/DurableTask.AzureStorage/Tracking/TrackingStoreBase.cs +++ b/src/DurableTask.AzureStorage/Tracking/TrackingStoreBase.cs @@ -95,7 +95,11 @@ public virtual Task PurgeInstanceHistoryAsync(DateTime creat public abstract Task SetNewExecutionAsync(ExecutionStartedEvent executionStartedEvent, ETag? eTag, string inputStatusOverride, CancellationToken cancellationToken = default); /// - public virtual Task UpdateStatusForRewindAsync(string instanceId, CancellationToken cancellationToken = default) + public virtual Task UpdateStatusForRewindAsync( + string instanceId, + string executionId, + ETag rewindStartETag, + CancellationToken cancellationToken = default) { throw new NotSupportedException(); } diff --git a/test/DurableTask.AzureStorage.Tests/AzureStorageScenarioTests.cs b/test/DurableTask.AzureStorage.Tests/AzureStorageScenarioTests.cs index d07babea..010b9309 100644 --- a/test/DurableTask.AzureStorage.Tests/AzureStorageScenarioTests.cs +++ b/test/DurableTask.AzureStorage.Tests/AzureStorageScenarioTests.cs @@ -13,9 +13,12 @@ namespace DurableTask.AzureStorage.Tests { + using Azure.Core; + using Azure.Core.Pipeline; using Azure.Data.Tables; using Azure.Storage.Blobs; using Azure.Storage.Blobs.Models; + using Azure.Storage.Queues; using DurableTask.AzureStorage.Storage; using DurableTask.AzureStorage.Tracking; using DurableTask.Core; @@ -33,6 +36,7 @@ namespace DurableTask.AzureStorage.Tests using System.IO; using System.Linq; using System.Net; + using System.Net.Http; using System.Reflection; using System.Runtime.Serialization; using System.Text; @@ -1518,6 +1522,950 @@ public async Task RewindActivityFail() } } + [TestMethod] + public async Task RewindLargeFailure_RetainsHistoryPayloadBlobs() + { + using (TestOrchestrationHost host = TestHelpers.GetTestOrchestrationHost(enableExtendedSessions: false)) + { + Orchestrations.RewindLargeFailure.ShouldFail = true; + host.ErrorPropagationMode = ErrorPropagationMode.UseFailureDetails; + await host.StartAsync(); + + string failureMessage = this.GenerateMediumRandomStringPayload().ToString(); + TestOrchestrationClient client = await host.StartOrchestrationAsync( + typeof(Orchestrations.RewindLargeFailure), + input: failureMessage); + OrchestrationState failed = await client.WaitForCompletionAsync(StandardTimeout); + Assert.IsNotNull(failed); + Assert.IsNotNull(failed.OrchestrationInstance); + Assert.AreEqual(OrchestrationStatus.Failed, failed.OrchestrationStatus); + + var trackingStore = (AzureTableTrackingStore)host.service.TrackingStore; + string instanceFilter = $"{AzureTableQueryFilter.PartitionKeyEquals(client.InstanceId)} and " + + $"{AzureTableQueryFilter.ColumnEquals(nameof(ITableEntity.RowKey), string.Empty)}"; + TableEntity failedInstance = await trackingStore.InstancesTable + .ExecuteQueryAsync(instanceFilter, 1) + .FirstOrDefaultAsync(); + string outputBlobUrl = failedInstance.GetString("Output"); + Assert.IsTrue(Uri.IsWellFormedUriString(outputBlobUrl, UriKind.Absolute)); + + string failedCompletionFilter = $"{AzureTableQueryFilter.PartitionKeyEquals(client.InstanceId)} and " + + $"{AzureTableQueryFilter.ColumnEquals(nameof(OrchestrationInstance.ExecutionId), failed.OrchestrationInstance.ExecutionId)} and " + + $"{AzureTableQueryFilter.ColumnEquals(nameof(HistoryEvent.EventType), nameof(EventType.ExecutionCompleted))}"; + TableEntity failedCompletion = (await trackingStore.HistoryTable + .ExecuteQueryAsync(failedCompletionFilter) + .ToListAsync()) + .Single(); + string resultBlobName = failedCompletion.GetString("ResultBlobName"); + string failureDetailsBlobName = failedCompletion.GetString("FailureDetailsBlobName"); + Assert.IsNotNull(resultBlobName); + Assert.IsNotNull(failureDetailsBlobName); + Assert.IsTrue(new Uri(outputBlobUrl).AbsolutePath.EndsWith(resultBlobName, StringComparison.Ordinal)); + + BlobContainerClient container = new BlobServiceClient(TestHelpers.GetTestStorageAccountConnectionString()) + .GetBlobContainerClient($"{host.TaskHub.ToLowerInvariant()}-largemessages"); + BlobClient resultBlob = container.GetBlobClient(resultBlobName); + BlobClient failureDetailsBlob = container.GetBlobClient(failureDetailsBlobName); + Assert.IsTrue((await resultBlob.ExistsAsync()).Value); + Assert.IsTrue((await failureDetailsBlob.ExistsAsync()).Value); + + CollectionAssert.AreEqual( + new[] { client.InstanceId }, + await trackingStore.RewindHistoryAsync(client.InstanceId).ToListAsync()); + + TableEntity rewoundInstance = await trackingStore.InstancesTable + .ExecuteQueryAsync(instanceFilter, 1) + .FirstOrDefaultAsync(); + Assert.AreEqual(OrchestrationStatus.Pending.ToString(), rewoundInstance.GetString("RuntimeStatus")); + Assert.IsFalse(rewoundInstance.ContainsKey("Output")); + + string historyFilter = $"{AzureTableQueryFilter.PartitionKeyEquals(client.InstanceId)} and " + + $"{AzureTableQueryFilter.ColumnEquals(nameof(ITableEntity.RowKey), failedCompletion.RowKey)}"; + TableEntity rewoundCompletion = (await trackingStore.HistoryTable + .ExecuteQueryAsync(historyFilter) + .ToListAsync()) + .Single(); + Assert.AreEqual(nameof(EventType.GenericEvent), rewoundCompletion.GetString(nameof(HistoryEvent.EventType))); + AssertHistoryPropertyUnchanged(failedCompletion, rewoundCompletion, "Result"); + AssertHistoryPropertyUnchanged(failedCompletion, rewoundCompletion, "ResultBlobName"); + AssertHistoryPropertyUnchanged(failedCompletion, rewoundCompletion, "FailureDetails"); + AssertHistoryPropertyUnchanged(failedCompletion, rewoundCompletion, "FailureDetailsBlobName"); + Assert.IsTrue((await resultBlob.ExistsAsync()).Value); + Assert.IsTrue((await failureDetailsBlob.ExistsAsync()).Value); + + CollectionAssert.AreEqual( + new[] { client.InstanceId }, + await trackingStore.RewindHistoryAsync(client.InstanceId).ToListAsync()); + Assert.IsTrue((await resultBlob.ExistsAsync()).Value); + Assert.IsTrue((await failureDetailsBlob.ExistsAsync()).Value); + + Orchestrations.RewindLargeFailure.ShouldFail = false; + await client.RewindAsync("Resume after retaining the failure payload."); + OrchestrationState completed = await client.WaitForCompletionAsync(StandardTimeout); + Assert.AreEqual(OrchestrationStatus.Completed, completed?.OrchestrationStatus); + Assert.AreEqual("\"Done\"", completed?.Output); + + TableEntity completedInstance = await trackingStore.InstancesTable + .ExecuteQueryAsync(instanceFilter, 1) + .FirstOrDefaultAsync(); + Assert.AreEqual(OrchestrationStatus.Completed.ToString(), completedInstance.GetString("RuntimeStatus")); + Assert.AreEqual("\"Done\"", completedInstance.GetString("Output")); + + await host.StopAsync(); + } + } + + [TestMethod] + public async Task RewindWhileStateReadIsDownloadingOutput_RetainsReadableBlob() + { + string connectionString = TestHelpers.GetTestStorageAccountConnectionString(); + var defaultProvider = new StorageAccountClientProvider(connectionString); + using var blobBarrier = new OneShotRequestBarrierHandler(); + using var blobClientProvider = + new TransportClientProvider( + defaultProvider.Blob, + blobBarrier); + var provider = new StorageAccountClientProvider( + blobClientProvider, + defaultProvider.Queue, + defaultProvider.Table); + + using (TestOrchestrationHost host = TestHelpers.GetTestOrchestrationHost( + enableExtendedSessions: false, + modifySettingsAction: settings => settings.StorageAccountClientProvider = provider)) + { + Orchestrations.RewindLargeFailure.ShouldFail = true; + host.ErrorPropagationMode = ErrorPropagationMode.UseFailureDetails; + await host.StartAsync(); + + string failureMessage = this.GenerateMediumRandomStringPayload().ToString(); + TestOrchestrationClient client = await host.StartOrchestrationAsync( + typeof(Orchestrations.RewindLargeFailure), + input: failureMessage); + OrchestrationState failed = await client.WaitForCompletionAsync(StandardTimeout); + Assert.IsNotNull(failed); + Assert.IsNotNull(failed.OrchestrationInstance); + Assert.AreEqual(OrchestrationStatus.Failed, failed.OrchestrationStatus); + + var trackingStore = (AzureTableTrackingStore)host.service.TrackingStore; + string instanceFilter = $"{AzureTableQueryFilter.PartitionKeyEquals(client.InstanceId)} and " + + $"{AzureTableQueryFilter.ColumnEquals(nameof(ITableEntity.RowKey), string.Empty)}"; + TableEntity failedInstance = await trackingStore.InstancesTable + .ExecuteQueryAsync(instanceFilter, 1) + .FirstOrDefaultAsync(); + var outputBlobUri = new Uri(failedInstance.GetString("Output")); + string failedCompletionFilter = $"{AzureTableQueryFilter.PartitionKeyEquals(client.InstanceId)} and " + + $"{AzureTableQueryFilter.ColumnEquals(nameof(OrchestrationInstance.ExecutionId), failed.OrchestrationInstance.ExecutionId)} and " + + $"{AzureTableQueryFilter.ColumnEquals(nameof(HistoryEvent.EventType), nameof(EventType.ExecutionCompleted))}"; + TableEntity failedCompletion = (await trackingStore.HistoryTable + .ExecuteQueryAsync(failedCompletionFilter) + .ToListAsync()) + .Single(); + BlobClient outputBlob = new BlobServiceClient(connectionString) + .GetBlobContainerClient($"{host.TaskHub.ToLowerInvariant()}-largemessages") + .GetBlobClient(failedCompletion.GetString("ResultBlobName")); + + await host.StopAsync(); + blobBarrier.Arm(request => + request.Method == HttpMethod.Get && + request.RequestUri.AbsolutePath == outputBlobUri.AbsolutePath); + + Task statusRead = client.GetStatusAsync(); + await blobBarrier.WaitUntilBlockedAsync(); + + try + { + await client.RewindAsync("Race a status read with persisted-output cleanup."); + + TableEntity rewoundInstance = await trackingStore.InstancesTable + .ExecuteQueryAsync(instanceFilter, 1) + .FirstOrDefaultAsync(); + Assert.IsFalse(rewoundInstance.ContainsKey("Output")); + Assert.IsTrue((await outputBlob.ExistsAsync()).Value); + + blobBarrier.Release(); + OrchestrationState staleSnapshot = await statusRead; + Assert.AreEqual(OrchestrationStatus.Failed, staleSnapshot.OrchestrationStatus); + Assert.AreEqual(failed.Output, staleSnapshot.Output); + } + finally + { + blobBarrier.Release(); + } + } + } + + [TestMethod] + public async Task RewindOldExecution_DoesNotResetNewExecution() + { + string connectionString = TestHelpers.GetTestStorageAccountConnectionString(); + var defaultProvider = new StorageAccountClientProvider(connectionString); + using var tableBarrier = new OneShotRequestBarrierHandler(); + using var tableClientProvider = + new TransportClientProvider( + defaultProvider.Table, + tableBarrier); + var provider = new StorageAccountClientProvider( + defaultProvider.Blob, + defaultProvider.Queue, + tableClientProvider); + + using (TestOrchestrationHost host = TestHelpers.GetTestOrchestrationHost( + enableExtendedSessions: false, + allowReplayingTerminalInstances: true, + modifySettingsAction: settings => settings.StorageAccountClientProvider = provider)) + { + Orchestrations.RewindLargeFailure.ShouldFail = true; + host.ErrorPropagationMode = ErrorPropagationMode.UseFailureDetails; + await host.StartAsync(); + + string instanceId = $"reuse-race-{Guid.NewGuid():N}"; + TestOrchestrationClient firstClient = await host.StartOrchestrationAsync( + typeof(Orchestrations.RewindLargeFailure), + input: this.GenerateMediumRandomStringPayload().ToString(), + instanceId: instanceId); + OrchestrationState firstFailure = await firstClient.WaitForCompletionAsync(StandardTimeout); + Assert.IsNotNull(firstFailure); + Assert.IsNotNull(firstFailure.OrchestrationInstance); + Assert.AreEqual(OrchestrationStatus.Failed, firstFailure.OrchestrationStatus); + + var trackingStore = (AzureTableTrackingStore)host.service.TrackingStore; + string instanceFilter = $"{AzureTableQueryFilter.PartitionKeyEquals(instanceId)} and " + + $"{AzureTableQueryFilter.ColumnEquals(nameof(ITableEntity.RowKey), string.Empty)}"; + tableBarrier.Arm(request => + request.Method == HttpMethod.Get && + request.RequestUri.AbsolutePath.IndexOf( + trackingStore.InstancesTable.Name, + StringComparison.OrdinalIgnoreCase) >= 0 && + Uri.UnescapeDataString(request.RequestUri.Query).Contains(instanceId), + matchingRequestsToSkip: 1); + + Task rewindFirstExecution = firstClient.RewindAsync("Pause before replacing the Instances row."); + await tableBarrier.WaitUntilBlockedAsync(); + + Orchestrations.RewindLargeFailure.ShouldFail = false; + TestOrchestrationClient secondClient = await host.StartOrchestrationAsync( + typeof(Orchestrations.RewindLargeFailure), + input: "second execution", + instanceId: instanceId); + OrchestrationState secondCompletion = await secondClient.WaitForCompletionAsync(StandardTimeout); + Assert.IsNotNull(secondCompletion); + Assert.IsNotNull(secondCompletion.OrchestrationInstance); + Assert.AreEqual(OrchestrationStatus.Completed, secondCompletion.OrchestrationStatus); + Assert.AreNotEqual( + firstFailure.OrchestrationInstance.ExecutionId, + secondCompletion.OrchestrationInstance.ExecutionId); + + TableEntity beforeRelease = await trackingStore.InstancesTable + .ExecuteQueryAsync(instanceFilter, 1) + .FirstOrDefaultAsync(); + Assert.AreEqual(secondCompletion.OrchestrationInstance.ExecutionId, beforeRelease.GetString("ExecutionId")); + Assert.AreEqual(OrchestrationStatus.Completed.ToString(), beforeRelease.GetString("RuntimeStatus")); + Assert.AreEqual("\"Done\"", beforeRelease.GetString("Output")); + + await host.StopAsync(); + tableBarrier.Release(); + DurableTaskStorageException conflict = + await Assert.ThrowsExceptionAsync(() => rewindFirstExecution); + StringAssert.Contains(conflict.Message, firstFailure.OrchestrationInstance.ExecutionId); + StringAssert.Contains(conflict.Message, secondCompletion.OrchestrationInstance.ExecutionId); + + TableEntity afterRelease = await trackingStore.InstancesTable + .ExecuteQueryAsync(instanceFilter, 1) + .FirstOrDefaultAsync(); + Assert.AreEqual(secondCompletion.OrchestrationInstance.ExecutionId, afterRelease.GetString("ExecutionId")); + Assert.AreEqual(OrchestrationStatus.Completed.ToString(), afterRelease.GetString("RuntimeStatus")); + Assert.AreEqual("\"Done\"", afterRelease.GetString("Output")); + + string secondCompletionFilter = $"{AzureTableQueryFilter.PartitionKeyEquals(instanceId)} and " + + $"{AzureTableQueryFilter.ColumnEquals(nameof(OrchestrationInstance.ExecutionId), secondCompletion.OrchestrationInstance.ExecutionId)} and " + + $"{AzureTableQueryFilter.ColumnEquals(nameof(HistoryEvent.EventType), nameof(EventType.ExecutionCompleted))}"; + TableEntity secondHistory = (await trackingStore.HistoryTable + .ExecuteQueryAsync(secondCompletionFilter) + .ToListAsync()) + .Single(); + Assert.AreEqual("\"Done\"", secondHistory.GetString("Result")); + } + } + + [TestMethod] + public async Task ConcurrentParentRewinds_PreserveChildRevivalTarget() + { + string connectionString = TestHelpers.GetTestStorageAccountConnectionString(); + var defaultProvider = new StorageAccountClientProvider(connectionString); + using var tableBarrier = new OneShotRequestBarrierHandler(); + using var queueRecorder = new RecordingRequestHandler(); + using var queueClientProvider = + new TransportClientProvider( + defaultProvider.Queue, + queueRecorder); + using var tableClientProvider = + new TransportClientProvider( + defaultProvider.Table, + tableBarrier); + var provider = new StorageAccountClientProvider( + defaultProvider.Blob, + queueClientProvider, + tableClientProvider); + AzureStorageOrchestrationServiceSettings settings = null; + + using (TestOrchestrationHost host = TestHelpers.GetTestOrchestrationHost( + enableExtendedSessions: false, + modifySettingsAction: configuredSettings => + { + configuredSettings.PartitionCount = 1; + configuredSettings.StorageAccountClientProvider = provider; + settings = configuredSettings; + })) + { + Orchestrations.ChildWorkflowSubOrchestrationFail.ShouldFail1 = true; + Orchestrations.ChildWorkflowSubOrchestrationFail.ShouldFail2 = true; + await host.StartAsync(); + + string parentInstanceId = $"parent-race-{Guid.NewGuid():N}"; + TestOrchestrationClient parentClient = await host.StartOrchestrationAsync( + typeof(Orchestrations.ParentWorkflowSubOrchestrationFail), + input: true, + instanceId: parentInstanceId); + OrchestrationState parentFailure = await parentClient.WaitForCompletionAsync(StandardTimeout); + Assert.AreEqual(OrchestrationStatus.Failed, parentFailure?.OrchestrationStatus); + + var trackingStore = (AzureTableTrackingStore)host.service.TrackingStore; + string childCreatedFilter = + $"{AzureTableQueryFilter.PartitionKeyEquals(parentInstanceId)} and " + + $"{AzureTableQueryFilter.ColumnEquals(nameof(HistoryEvent.EventType), nameof(EventType.SubOrchestrationInstanceCreated))}"; + TableEntity childCreated = (await trackingStore.HistoryTable + .ExecuteQueryAsync(childCreatedFilter) + .ToListAsync()) + .Single(); + string childInstanceId = childCreated.GetString(nameof(OrchestrationInstance.InstanceId)); + string childInstanceFilter = + $"{AzureTableQueryFilter.PartitionKeyEquals(childInstanceId)} and " + + $"{AzureTableQueryFilter.ColumnEquals(nameof(ITableEntity.RowKey), string.Empty)}"; + string parentInstanceFilter = + $"{AzureTableQueryFilter.PartitionKeyEquals(parentInstanceId)} and " + + $"{AzureTableQueryFilter.ColumnEquals(nameof(ITableEntity.RowKey), string.Empty)}"; + + Orchestrations.ChildWorkflowSubOrchestrationFail.ShouldFail1 = false; + Orchestrations.ChildWorkflowSubOrchestrationFail.ShouldFail2 = false; + await host.StopAsync(); + string controlQueueName = AzureStorageOrchestrationService.GetControlQueueName(host.TaskHub, 0); + queueRecorder.Clear(); + + tableBarrier.Arm(request => + request.Method == HttpMethod.Put && + request.RequestUri.AbsolutePath.IndexOf( + trackingStore.InstancesTable.Name, + StringComparison.OrdinalIgnoreCase) >= 0 && + Uri.UnescapeDataString(request.RequestUri.AbsoluteUri).Contains(parentInstanceId)); + + Task rewindA = parentClient.RewindAsync("Concurrent rewind A."); + await tableBarrier.WaitUntilBlockedAsync(); + + TableEntity pendingChild = await trackingStore.InstancesTable + .ExecuteQueryAsync(childInstanceFilter, 1) + .FirstOrDefaultAsync(); + Assert.AreEqual(OrchestrationStatus.Pending.ToString(), pendingChild.GetString("RuntimeStatus")); + + await parentClient.RewindAsync("Concurrent rewind B."); + + tableBarrier.Release(); + await rewindA; + + var messageManager = new MessageManager( + settings, + new AzureStorageClient(settings), + $"{host.TaskHub.ToLowerInvariant()}-largemessages"); + string[] queuedRewindTargets = queueRecorder + .GetQueueMessageBodies(controlQueueName) + .Select(body => DeserializeQueueMessageBody(messageManager, body)) + .Where(message => message.TaskMessage.Event is GenericEvent) + .Select(message => message.TaskMessage.OrchestrationInstance.InstanceId) + .Distinct() + .OrderBy(instanceId => instanceId) + .ToArray(); + CollectionAssert.AreEqual( + new[] { childInstanceId }, + queuedRewindTargets); + + var resumeService = new AzureStorageOrchestrationService(settings); + using var resumeWorker = new TaskHubWorker(resumeService, loggerFactory: settings.LoggerFactory); + resumeWorker.AddTaskOrchestrations( + typeof(Orchestrations.ParentWorkflowSubOrchestrationFail), + typeof(Orchestrations.ChildWorkflowSubOrchestrationFail)); + resumeWorker.AddTaskActivities(typeof(Activities.Hello)); + await resumeWorker.StartAsync(); + try + { + TableEntity completedParent = await WaitForInstanceStatusAsync( + trackingStore.InstancesTable, + parentInstanceFilter, + OrchestrationStatus.Completed); + TableEntity completedChild = await WaitForInstanceStatusAsync( + trackingStore.InstancesTable, + childInstanceFilter, + OrchestrationStatus.Completed); + Assert.IsTrue(completedParent.ContainsKey("Output")); + Assert.IsTrue(completedChild.ContainsKey("Output")); + } + finally + { + await resumeWorker.StopAsync(isForced: true); + } + + Orchestrations.ChildWorkflowSubOrchestrationFail.ShouldFail1 = true; + Orchestrations.ChildWorkflowSubOrchestrationFail.ShouldFail2 = true; + } + } + + [DataTestMethod] + [DataRow(1, false)] + [DataRow(3, false)] + [DataRow(1, true)] + public async Task TerminalRepairConflicts_PreserveChildForParentRetry( + int conflictCount, + bool repairChildAfterReset) + { + string connectionString = TestHelpers.GetTestStorageAccountConnectionString(); + var defaultProvider = new StorageAccountClientProvider(connectionString); + using var tableBarrier = new OneShotRequestBarrierHandler(); + using var queueRecorder = new RecordingRequestHandler(); + using var queueClientProvider = + new TransportClientProvider( + defaultProvider.Queue, + queueRecorder); + using var tableClientProvider = + new TransportClientProvider( + defaultProvider.Table, + tableBarrier); + var provider = new StorageAccountClientProvider( + defaultProvider.Blob, + queueClientProvider, + tableClientProvider); + AzureStorageOrchestrationServiceSettings settings = null; + + using (TestOrchestrationHost host = TestHelpers.GetTestOrchestrationHost( + enableExtendedSessions: false, + modifySettingsAction: configuredSettings => + { + configuredSettings.PartitionCount = 1; + configuredSettings.StorageAccountClientProvider = provider; + settings = configuredSettings; + })) + { + Orchestrations.ChildWorkflowSubOrchestrationFail.ShouldFail1 = true; + Orchestrations.ChildWorkflowSubOrchestrationFail.ShouldFail2 = true; + await host.StartAsync(); + + try + { + string parentInstanceId = $"parent-repair-race-{Guid.NewGuid():N}"; + TestOrchestrationClient parentClient = await host.StartOrchestrationAsync( + typeof(Orchestrations.ParentWorkflowSubOrchestrationFail), + input: true, + instanceId: parentInstanceId); + OrchestrationState parentFailure = await parentClient.WaitForCompletionAsync(StandardTimeout); + Assert.IsNotNull(parentFailure); + Assert.IsNotNull(parentFailure.OrchestrationInstance); + Assert.AreEqual(OrchestrationStatus.Failed, parentFailure.OrchestrationStatus); + + var trackingStore = (AzureTableTrackingStore)host.service.TrackingStore; + string parentExecutionId = parentFailure.OrchestrationInstance.ExecutionId; + OrchestrationHistory failedParentHistory = + await trackingStore.GetHistoryEventsAsync(parentInstanceId, parentExecutionId); + var capturedFailedRuntimeState = + new OrchestrationRuntimeState(failedParentHistory.Events); + Assert.AreEqual(OrchestrationStatus.Failed, capturedFailedRuntimeState.OrchestrationStatus); + Assert.AreEqual( + parentExecutionId, + capturedFailedRuntimeState.OrchestrationInstance.ExecutionId); + + string childCreatedFilter = + $"{AzureTableQueryFilter.PartitionKeyEquals(parentInstanceId)} and " + + $"{AzureTableQueryFilter.ColumnEquals(nameof(HistoryEvent.EventType), nameof(EventType.SubOrchestrationInstanceCreated))}"; + TableEntity childCreated = (await trackingStore.HistoryTable + .ExecuteQueryAsync(childCreatedFilter) + .ToListAsync()) + .Single(); + string childInstanceId = childCreated.GetString(nameof(OrchestrationInstance.InstanceId)); + string childInstanceFilter = + $"{AzureTableQueryFilter.PartitionKeyEquals(childInstanceId)} and " + + $"{AzureTableQueryFilter.ColumnEquals(nameof(ITableEntity.RowKey), string.Empty)}"; + string parentInstanceFilter = + $"{AzureTableQueryFilter.PartitionKeyEquals(parentInstanceId)} and " + + $"{AzureTableQueryFilter.ColumnEquals(nameof(ITableEntity.RowKey), string.Empty)}"; + string rewoundChildMarkerFilter = + $"{AzureTableQueryFilter.PartitionKeyEquals(parentInstanceId)} and " + + $"{AzureTableQueryFilter.ColumnEquals(nameof(HistoryEvent.EventType), nameof(EventType.GenericEvent))} and " + + $"{AzureTableQueryFilter.ColumnEquals(nameof(SubOrchestrationInstanceFailedEvent.Reason), "Rewound: " + nameof(EventType.SubOrchestrationInstanceFailed))}"; + TableEntity failedChild = await trackingStore.InstancesTable + .ExecuteQueryAsync(childInstanceFilter, 1) + .FirstOrDefaultAsync(); + string childExecutionId = failedChild.GetString("ExecutionId"); + OrchestrationHistory failedChildHistory = + await trackingStore.GetHistoryEventsAsync(childInstanceId, childExecutionId); + var capturedFailedChildRuntimeState = + new OrchestrationRuntimeState(failedChildHistory.Events); + Assert.AreEqual( + OrchestrationStatus.Failed, + capturedFailedChildRuntimeState.OrchestrationStatus); + + Orchestrations.ChildWorkflowSubOrchestrationFail.ShouldFail1 = false; + Orchestrations.ChildWorkflowSubOrchestrationFail.ShouldFail2 = false; + await host.StopAsync(); + string controlQueueName = AzureStorageOrchestrationService.GetControlQueueName(host.TaskHub, 0); + queueRecorder.Clear(); + + Func isParentInstanceReplace = request => + request.Method == HttpMethod.Put && + request.RequestUri.AbsolutePath.IndexOf( + trackingStore.InstancesTable.Name, + StringComparison.OrdinalIgnoreCase) >= 0 && + Uri.UnescapeDataString(request.RequestUri.AbsoluteUri).Contains(parentInstanceId); + + for (int conflict = 0; conflict < conflictCount; conflict++) + { + tableBarrier.Arm(isParentInstanceReplace); + Task rewind = parentClient.RewindAsync( + $"Rewind with terminal-state repair conflict {conflict + 1}."); + await tableBarrier.WaitUntilBlockedAsync(); + + TableEntity pendingChild = await trackingStore.InstancesTable + .ExecuteQueryAsync(childInstanceFilter, 1) + .FirstOrDefaultAsync(); + Assert.AreEqual( + OrchestrationStatus.Pending.ToString(), + pendingChild.GetString("RuntimeStatus")); + Assert.IsFalse(pendingChild.ContainsKey("Output")); + + TableEntity parentBeforeRepair = await trackingStore.InstancesTable + .ExecuteQueryAsync(parentInstanceFilter, 1) + .FirstOrDefaultAsync(); + Assert.AreEqual( + OrchestrationStatus.Failed.ToString(), + parentBeforeRepair.GetString("RuntimeStatus")); + + await trackingStore.UpdateInstanceStatusForCompletedOrchestrationAsync( + parentInstanceId, + parentExecutionId, + capturedFailedRuntimeState, + instanceEntityExists: true); + + TableEntity parentAfterRepair = await trackingStore.InstancesTable + .ExecuteQueryAsync(parentInstanceFilter, 1) + .FirstOrDefaultAsync(); + Assert.AreEqual( + OrchestrationStatus.Failed.ToString(), + parentAfterRepair.GetString("RuntimeStatus")); + Assert.AreNotEqual( + parentBeforeRepair.ETag.ToString(), + parentAfterRepair.ETag.ToString(), + "The production terminal-state repair must advance the parent Instances ETag."); + + tableBarrier.Release(); + HttpStatusCode resetStatus = await tableBarrier.WaitUntilCompletedAsync(); + DurableTaskStorageException rewindFailure = + await Assert.ThrowsExceptionAsync(() => rewind); + Assert.AreEqual(HttpStatusCode.PreconditionFailed, resetStatus); + Assert.AreEqual( + (int)HttpStatusCode.PreconditionFailed, + rewindFailure.HttpStatusCode); + Assert.AreEqual( + 0, + queueRecorder.GetQueueMessageBodies(controlQueueName).Count, + "A failed rewind must not enqueue targets before the whole tree is reset."); + Assert.AreEqual( + 1, + (await trackingStore.HistoryTable + .ExecuteQueryAsync(rewoundChildMarkerFilter) + .ToListAsync()) + .Count, + "The converted child-failure marker must remain recoverable."); + } + + if (repairChildAfterReset) + { + await trackingStore.UpdateInstanceStatusForCompletedOrchestrationAsync( + childInstanceId, + childExecutionId, + capturedFailedChildRuntimeState, + instanceEntityExists: true); + TableEntity repairedChild = await trackingStore.InstancesTable + .ExecuteQueryAsync(childInstanceFilter, 1) + .FirstOrDefaultAsync(); + Assert.AreEqual( + OrchestrationStatus.Failed.ToString(), + repairedChild.GetString("RuntimeStatus")); + Assert.IsTrue(repairedChild.ContainsKey("Output")); + } + + await parentClient.RewindAsync("Retry after terminal-state repair conflicts stop."); + + var messageManager = new MessageManager( + settings, + new AzureStorageClient(settings), + $"{host.TaskHub.ToLowerInvariant()}-largemessages"); + string[] retryTargets = queueRecorder + .GetQueueMessageBodies(controlQueueName) + .Select(body => DeserializeQueueMessageBody(messageManager, body)) + .Where(message => message.TaskMessage.Event is GenericEvent) + .Select(message => message.TaskMessage.OrchestrationInstance.InstanceId) + .Distinct() + .OrderBy(instanceId => instanceId) + .ToArray(); + CollectionAssert.AreEqual(new[] { childInstanceId }, retryTargets); + + var resumeService = new AzureStorageOrchestrationService(settings); + using var resumeWorker = new TaskHubWorker( + resumeService, + loggerFactory: settings.LoggerFactory); + resumeWorker.AddTaskOrchestrations( + typeof(Orchestrations.ParentWorkflowSubOrchestrationFail), + typeof(Orchestrations.ChildWorkflowSubOrchestrationFail)); + resumeWorker.AddTaskActivities(typeof(Activities.Hello)); + await resumeWorker.StartAsync(); + try + { + TableEntity completedParent = await WaitForInstanceStatusAsync( + trackingStore.InstancesTable, + parentInstanceFilter, + OrchestrationStatus.Completed); + TableEntity completedChild = await WaitForInstanceStatusAsync( + trackingStore.InstancesTable, + childInstanceFilter, + OrchestrationStatus.Completed); + Assert.IsTrue(completedParent.ContainsKey("Output")); + Assert.IsTrue(completedChild.ContainsKey("Output")); + } + finally + { + await resumeWorker.StopAsync(isForced: true); + } + } + finally + { + tableBarrier.Release(); + Orchestrations.ChildWorkflowSubOrchestrationFail.ShouldFail1 = true; + Orchestrations.ChildWorkflowSubOrchestrationFail.ShouldFail2 = true; + } + } + } + + [TestMethod] + public async Task RewindAfterLaterChildFailure_DoesNotReviveCompletedChild() + { + string connectionString = TestHelpers.GetTestStorageAccountConnectionString(); + var defaultProvider = new StorageAccountClientProvider(connectionString); + using var queueRecorder = new RecordingRequestHandler(); + using var queueClientProvider = + new TransportClientProvider( + defaultProvider.Queue, + queueRecorder); + var provider = new StorageAccountClientProvider( + defaultProvider.Blob, + queueClientProvider, + defaultProvider.Table); + AzureStorageOrchestrationServiceSettings settings = null; + + using (TestOrchestrationHost host = TestHelpers.GetTestOrchestrationHost( + enableExtendedSessions: false, + modifySettingsAction: configuredSettings => + { + configuredSettings.PartitionCount = 1; + configuredSettings.StorageAccountClientProvider = provider; + settings = configuredSettings; + })) + { + Orchestrations.RewindSequentialChild.FailingInput = 0; + await host.StartAsync(); + + try + { + string parentInstanceId = $"parent-sequential-{Guid.NewGuid():N}"; + TestOrchestrationClient parentClient = await host.StartOrchestrationAsync( + typeof(Orchestrations.RewindSequentialParent), + input: 2, + instanceId: parentInstanceId); + OrchestrationState firstFailure = + await parentClient.WaitForCompletionAsync(StandardTimeout); + Assert.IsNotNull(firstFailure); + Assert.AreEqual(OrchestrationStatus.Failed, firstFailure.OrchestrationStatus); + + var trackingStore = (AzureTableTrackingStore)host.service.TrackingStore; + string childCreatedFilter = + $"{AzureTableQueryFilter.PartitionKeyEquals(parentInstanceId)} and " + + $"{AzureTableQueryFilter.ColumnEquals(nameof(HistoryEvent.EventType), nameof(EventType.SubOrchestrationInstanceCreated))}"; + + Orchestrations.RewindSequentialChild.FailingInput = 1; + await parentClient.RewindAsync("Allow the first child and fail the second child."); + OrchestrationState secondFailure = + await parentClient.WaitForCompletionAsync(StandardTimeout); + Assert.IsNotNull(secondFailure); + Assert.AreEqual(OrchestrationStatus.Failed, secondFailure.OrchestrationStatus); + + TableEntity[] childCreated = (await trackingStore.HistoryTable + .ExecuteQueryAsync(childCreatedFilter) + .ToListAsync()) + .OrderBy(entity => entity.RowKey) + .ToArray(); + Assert.AreEqual(2, childCreated.Length); + string completedChildInstanceId = + childCreated[0].GetString(nameof(OrchestrationInstance.InstanceId)); + string failedChildInstanceId = + childCreated[1].GetString(nameof(OrchestrationInstance.InstanceId)); + string completedChildFilter = + $"{AzureTableQueryFilter.PartitionKeyEquals(completedChildInstanceId)} and " + + $"{AzureTableQueryFilter.ColumnEquals(nameof(ITableEntity.RowKey), string.Empty)}"; + string failedChildFilter = + $"{AzureTableQueryFilter.PartitionKeyEquals(failedChildInstanceId)} and " + + $"{AzureTableQueryFilter.ColumnEquals(nameof(ITableEntity.RowKey), string.Empty)}"; + string parentInstanceFilter = + $"{AzureTableQueryFilter.PartitionKeyEquals(parentInstanceId)} and " + + $"{AzureTableQueryFilter.ColumnEquals(nameof(ITableEntity.RowKey), string.Empty)}"; + + TableEntity completedChildBefore = await trackingStore.InstancesTable + .ExecuteQueryAsync(completedChildFilter, 1) + .FirstOrDefaultAsync(); + Assert.AreEqual( + OrchestrationStatus.Completed.ToString(), + completedChildBefore.GetString("RuntimeStatus")); + Assert.AreEqual("\"0\"", completedChildBefore.GetString("Output")); + + await host.StopAsync(); + queueRecorder.Clear(); + Orchestrations.RewindSequentialChild.FailingInput = -1; + await parentClient.RewindAsync("Recover only the later failed child."); + + var messageManager = new MessageManager( + settings, + new AzureStorageClient(settings), + $"{host.TaskHub.ToLowerInvariant()}-largemessages"); + string[] retryTargets = queueRecorder + .GetQueueMessageBodies( + AzureStorageOrchestrationService.GetControlQueueName(host.TaskHub, 0)) + .Select(body => DeserializeQueueMessageBody(messageManager, body)) + .Where(message => message.TaskMessage.Event is GenericEvent) + .Select(message => message.TaskMessage.OrchestrationInstance.InstanceId) + .Distinct() + .ToArray(); + CollectionAssert.AreEqual(new[] { failedChildInstanceId }, retryTargets); + + TableEntity completedChildAfter = await trackingStore.InstancesTable + .ExecuteQueryAsync(completedChildFilter, 1) + .FirstOrDefaultAsync(); + Assert.AreEqual(completedChildBefore.ETag, completedChildAfter.ETag); + Assert.AreEqual( + OrchestrationStatus.Completed.ToString(), + completedChildAfter.GetString("RuntimeStatus")); + Assert.AreEqual("\"0\"", completedChildAfter.GetString("Output")); + + var resumeService = new AzureStorageOrchestrationService(settings); + using var resumeWorker = new TaskHubWorker( + resumeService, + loggerFactory: settings.LoggerFactory); + resumeWorker.AddTaskOrchestrations( + typeof(Orchestrations.RewindSequentialParent), + typeof(Orchestrations.RewindSequentialChild)); + await resumeWorker.StartAsync(); + try + { + TableEntity completedParent = await WaitForInstanceStatusAsync( + trackingStore.InstancesTable, + parentInstanceFilter, + OrchestrationStatus.Completed); + TableEntity completedLaterChild = await WaitForInstanceStatusAsync( + trackingStore.InstancesTable, + failedChildFilter, + OrchestrationStatus.Completed); + Assert.AreEqual("\"0,1\"", completedParent.GetString("Output")); + Assert.AreEqual("\"1\"", completedLaterChild.GetString("Output")); + } + finally + { + await resumeWorker.StopAsync(isForced: true); + } + } + finally + { + Orchestrations.RewindSequentialChild.FailingInput = 0; + } + } + } + + [DataTestMethod] + [DataRow(false)] + [DataRow(true)] + public async Task RewindRejectsLaggingChildFailureWrite(bool suspendedProjection) + { + string connectionString = TestHelpers.GetTestStorageAccountConnectionString(); + var defaultProvider = new StorageAccountClientProvider(connectionString); + using var tableBarrier = new OneShotRequestBarrierHandler(); + using var queueBarrier = new OneShotRequestBarrierHandler(); + using var queueClientProvider = + new TransportClientProvider( + defaultProvider.Queue, + queueBarrier); + using var tableClientProvider = + new TransportClientProvider( + defaultProvider.Table, + tableBarrier); + var provider = new StorageAccountClientProvider( + defaultProvider.Blob, + queueClientProvider, + tableClientProvider); + AzureStorageOrchestrationServiceSettings settings = null; + + using (TestOrchestrationHost host = TestHelpers.GetTestOrchestrationHost( + enableExtendedSessions: false, + modifySettingsAction: configuredSettings => + { + configuredSettings.PartitionCount = 1; + configuredSettings.StorageAccountClientProvider = provider; + configuredSettings.UseInstanceTableEtag = true; + settings = configuredSettings; + })) + { + string parentInstanceId = $"parent-lagging-{Guid.NewGuid():N}"; + string childInstanceId = $"child-lagging-{Guid.NewGuid():N}"; + Func isChildInstanceUpdate = request => + (request.Method == HttpMethod.Put || + request.Method.Method == "MERGE" || + request.Method.Method == "PATCH") && + Uri.UnescapeDataString(request.RequestUri.AbsoluteUri).Contains(childInstanceId); + if (!suspendedProjection) + { + tableBarrier.Arm(isChildInstanceUpdate); + } + + Orchestrations.RewindLaggingWriterChild.ShouldFail = true; + Activities.RewindLaggingWriterBlocking.Reset(); + await host.StartAsync(); + + try + { + var trackingStore = (AzureTableTrackingStore)host.service.TrackingStore; + string childInstanceFilter = + $"{AzureTableQueryFilter.PartitionKeyEquals(childInstanceId)} and " + + $"{AzureTableQueryFilter.ColumnEquals(nameof(ITableEntity.RowKey), string.Empty)}"; + TestOrchestrationClient parentClient = await host.StartOrchestrationAsync( + typeof(Orchestrations.RewindLaggingWriterParent), + input: new Orchestrations.RewindLaggingWriterInput + { + ChildInstanceId = childInstanceId, + SuspendBeforeFailure = suspendedProjection, + }, + instanceId: parentInstanceId); + + if (suspendedProjection) + { + Task activityStarted = Activities.RewindLaggingWriterBlocking.Started; + Task activityStartWait = await Task.WhenAny(activityStarted, Task.Delay(StandardTimeout)); + Assert.AreSame(activityStarted, activityStartWait, "The blocking child activity did not start."); + await activityStarted; + + var childInstance = new OrchestrationInstance { InstanceId = childInstanceId }; + var childControlClient = new TaskHubClient( + host.service, + loggerFactory: settings.LoggerFactory); + await childControlClient.SuspendInstanceAsync(childInstance, "suspend-before-failure"); + TableEntity suspendedChild = await WaitForInstanceStatusAsync( + trackingStore.InstancesTable, + childInstanceFilter, + OrchestrationStatus.Suspended); + Assert.AreEqual("suspend-before-failure", suspendedChild.GetString("Output")); + + Activities.RewindLaggingWriterBlocking.Release(); + string taskCompletedFilter = + $"{AzureTableQueryFilter.PartitionKeyEquals(childInstanceId)} and " + + $"{AzureTableQueryFilter.ColumnEquals(nameof(HistoryEvent.EventType), nameof(EventType.TaskCompleted))}"; + await WaitForTableEntityAsync( + trackingStore.HistoryTable, + taskCompletedFilter, + entity => true, + "the child activity result to be buffered while suspended"); + await WaitForTableEntityAsync( + trackingStore.InstancesTable, + childInstanceFilter, + entity => + entity.ETag.ToString() != suspendedChild.ETag.ToString() && + entity.GetString("RuntimeStatus") == OrchestrationStatus.Suspended.ToString(), + "the suspended child checkpoint containing the buffered activity result"); + + tableBarrier.Arm(isChildInstanceUpdate); + await childControlClient.ResumeInstanceAsync(childInstance, "resume-into-failure"); + } + + OrchestrationState parentFailure = await parentClient.WaitForCompletionAsync(StandardTimeout); + Assert.AreEqual(OrchestrationStatus.Failed, parentFailure?.OrchestrationStatus); + await tableBarrier.WaitUntilBlockedAsync(); + + TableEntity laggingChild = await trackingStore.InstancesTable + .ExecuteQueryAsync(childInstanceFilter, 1) + .FirstOrDefaultAsync(); + if (suspendedProjection) + { + Assert.AreEqual( + OrchestrationStatus.Suspended.ToString(), + laggingChild.GetString("RuntimeStatus")); + Assert.AreEqual("suspend-before-failure", laggingChild.GetString("Output")); + } + else + { + Assert.IsTrue( + laggingChild.GetString("RuntimeStatus") == OrchestrationStatus.Pending.ToString() || + laggingChild.GetString("RuntimeStatus") == OrchestrationStatus.Running.ToString(), + $"Expected a pre-failure projection but found {laggingChild.GetString("RuntimeStatus")}."); + Assert.IsFalse(laggingChild.ContainsKey("Output")); + } + + Orchestrations.RewindLaggingWriterChild.ShouldFail = false; + string controlQueueName = AzureStorageOrchestrationService.GetControlQueueName(host.TaskHub, 0); + queueBarrier.Arm(request => + request.Method == HttpMethod.Post && + request.RequestUri.AbsolutePath.IndexOf( + $"/{controlQueueName}/messages", + StringComparison.OrdinalIgnoreCase) >= 0); + + Task rewindTask = parentClient.RewindAsync("Rewind while the child failure write is delayed."); + await queueBarrier.WaitUntilBlockedAsync(); + + tableBarrier.Release(); + HttpStatusCode staleWriteStatus = await tableBarrier.WaitUntilCompletedAsync(); + + Assert.AreEqual(HttpStatusCode.PreconditionFailed, staleWriteStatus); + TableEntity rewoundChild = await trackingStore.InstancesTable + .ExecuteQueryAsync(childInstanceFilter, 1) + .FirstOrDefaultAsync(); + Assert.AreEqual( + OrchestrationStatus.Pending.ToString(), + rewoundChild.GetString("RuntimeStatus")); + Assert.IsFalse(rewoundChild.ContainsKey("Output")); + + queueBarrier.Release(); + await rewindTask; + + OrchestrationState parentCompletion = + await parentClient.WaitForCompletionAsync(StandardTimeout); + Assert.IsNotNull(parentCompletion); + Assert.AreEqual(OrchestrationStatus.Completed, parentCompletion.OrchestrationStatus); + Assert.AreEqual("\"Hello, lagging!\"", parentCompletion.Output); + + TableEntity completedChild = await WaitForInstanceStatusAsync( + trackingStore.InstancesTable, + childInstanceFilter, + OrchestrationStatus.Completed); + Assert.AreEqual("\"Hello, lagging!\"", completedChild.GetString("Output")); + } + finally + { + tableBarrier.Release(); + queueBarrier.Release(); + Orchestrations.RewindLaggingWriterChild.ShouldFail = true; + Activities.RewindLaggingWriterBlocking.Release(); + } + } + } + [TestMethod] public async Task RewindMultipleActivityFail() { @@ -5049,6 +5997,297 @@ public async Task OpenTelemetry_ExternalEvent_SendEvent(bool enableExtendedSessi } #endif + static void AssertHistoryPropertyUnchanged( + TableEntity beforeRewind, + TableEntity afterRewind, + string propertyName) + { + bool beforeContainsProperty = beforeRewind.TryGetValue(propertyName, out object beforeValue); + bool afterContainsProperty = afterRewind.TryGetValue(propertyName, out object afterValue); + Assert.AreEqual( + beforeContainsProperty, + afterContainsProperty, + $"The presence of history property '{propertyName}' changed during rewind."); + if (beforeContainsProperty) + { + Assert.AreEqual(beforeValue, afterValue); + } + } + + static MessageData DeserializeQueueMessageBody(MessageManager messageManager, string body) + { + if (!body.StartsWith("{", StringComparison.Ordinal)) + { + body = Encoding.UTF8.GetString(Convert.FromBase64String(body)); + } + + MessageData data = messageManager.DeserializeMessageData(body); + Assert.IsTrue( + string.IsNullOrEmpty(data.CompressedBlobName), + "Rewind control messages are expected to fit directly in the control queue."); + return data; + } + + static async Task WaitForInstanceStatusAsync( + Table table, + string filter, + OrchestrationStatus expectedStatus) + { + return await WaitForTableEntityAsync( + table, + filter, + entity => entity.GetString("RuntimeStatus") == expectedStatus.ToString(), + $"the instance to reach {expectedStatus}"); + } + + static async Task WaitForTableEntityAsync( + Table table, + string filter, + Func predicate, + string expectation) + { + Stopwatch timeout = Stopwatch.StartNew(); + do + { + TableEntity entity = await table + .ExecuteQueryAsync(filter, 1) + .FirstOrDefaultAsync(); + if (entity != null && predicate(entity)) + { + return entity; + } + + await Task.Delay(TimeSpan.FromMilliseconds(100)); + } + while (timeout.Elapsed < StandardTimeout); + + Assert.Fail($"Timed out waiting for {expectation} within {StandardTimeout}."); + return null; + } + + sealed class TransportClientProvider : + IStorageServiceClientProvider, + IDisposable + where TOptions : ClientOptions + { + readonly IStorageServiceClientProvider inner; + readonly HttpClientTransport transport; + + public TransportClientProvider( + IStorageServiceClientProvider inner, + HttpMessageHandler handler) + { + this.inner = inner; + this.transport = new HttpClientTransport( + new HttpClient(handler, disposeHandler: false)); + } + + public TOptions CreateOptions() + { + TOptions options = this.inner.CreateOptions(); + options.Transport = this.transport; + return options; + } + + public TClient CreateClient(TOptions options) => this.inner.CreateClient(options); + + public void Dispose() + { + this.transport.Dispose(); + } + } + + sealed class OneShotRequestBarrierHandler : DelegatingHandler + { + readonly object sync = new object(); + Func predicate; + TaskCompletionSource blocked; + TaskCompletionSource release; + TaskCompletionSource completed; + int matchesToSkip; + bool claimed; + + public OneShotRequestBarrierHandler() + : base(new HttpClientHandler()) + { + } + + public void Arm( + Func predicate, + int matchingRequestsToSkip = 0) + { + lock (this.sync) + { + this.predicate = predicate; + this.blocked = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + this.release = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + this.completed = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + this.matchesToSkip = matchingRequestsToSkip; + this.claimed = false; + } + } + + public async Task WaitUntilBlockedAsync() + { + Task blockedTask; + lock (this.sync) + { + blockedTask = this.blocked.Task; + } + + Task completedTask = await Task.WhenAny(blockedTask, Task.Delay(StandardTimeout)); + Assert.AreSame(blockedTask, completedTask, "The expected storage request did not reach the barrier."); + await blockedTask; + } + + public async Task WaitUntilCompletedAsync() + { + Task completionTask; + lock (this.sync) + { + completionTask = this.completed.Task; + } + + Task completedTask = await Task.WhenAny(completionTask, Task.Delay(StandardTimeout)); + Assert.AreSame(completionTask, completedTask, "The blocked request did not complete."); + return await completionTask; + } + + public void Release() + { + lock (this.sync) + { + this.release?.TrySetResult(null); + } + } + + protected override void Dispose(bool disposing) + { + if (disposing) + { + this.Release(); + } + + base.Dispose(disposing); + } + + protected override async Task SendAsync( + HttpRequestMessage request, + CancellationToken cancellationToken) + { + Task releaseTask = null; + TaskCompletionSource completionSignal = null; + lock (this.sync) + { + if (!this.claimed && this.predicate?.Invoke(request) == true) + { + if (this.matchesToSkip > 0) + { + this.matchesToSkip--; + } + else + { + this.claimed = true; + this.blocked.TrySetResult(null); + releaseTask = this.release.Task; + completionSignal = this.completed; + } + } + } + + if (releaseTask != null) + { + await releaseTask; + try + { + HttpResponseMessage response = await base.SendAsync(request, cancellationToken); + completionSignal.TrySetResult(response.StatusCode); + return response; + } + catch (Exception exception) + { + completionSignal.TrySetException(exception); + throw; + } + } + + return await base.SendAsync(request, cancellationToken); + } + } + + sealed class RecordingRequestHandler : DelegatingHandler + { + readonly object sync = new object(); + readonly List requests = new List(); + + public RecordingRequestHandler() + : base(new HttpClientHandler()) + { + } + + public void Clear() + { + lock (this.sync) + { + this.requests.Clear(); + } + } + + public IReadOnlyList GetQueueMessageBodies(string queueName) + { + lock (this.sync) + { + return this.requests + .Where(request => + request.Method == HttpMethod.Post && + request.Uri.AbsolutePath.IndexOf( + $"/{queueName}/messages", + StringComparison.OrdinalIgnoreCase) >= 0) + .Select(request => + System.Xml.Linq.XDocument.Parse(request.Body) + .Descendants() + .Single(element => element.Name.LocalName == "MessageText") + .Value) + .ToList(); + } + } + + protected override async Task SendAsync( + HttpRequestMessage request, + CancellationToken cancellationToken) + { + string body = request.Content == null + ? null + : await request.Content.ReadAsStringAsync(); + HttpResponseMessage response = await base.SendAsync(request, cancellationToken); + if (response.IsSuccessStatusCode && body != null) + { + lock (this.sync) + { + this.requests.Add(new RecordedRequest(request.Method, request.RequestUri, body)); + } + } + + return response; + } + + sealed class RecordedRequest + { + public RecordedRequest(HttpMethod method, Uri uri, string body) + { + this.Method = method; + this.Uri = uri; + this.Body = body; + } + + public HttpMethod Method { get; } + + public Uri Uri { get; } + + public string Body { get; } + } + } + static class Orchestrations { internal class SayHelloInline : TaskOrchestration @@ -5272,6 +6511,89 @@ public override async Task RunTask(OrchestrationContext context, int inp } } + [KnownType(typeof(RewindLaggingWriterChild))] + [KnownType(typeof(Activities.Hello))] + [KnownType(typeof(Activities.RewindLaggingWriterBlocking))] + public class RewindLaggingWriterParent : TaskOrchestration + { + public override Task RunTask( + OrchestrationContext context, + RewindLaggingWriterInput input) + { + return context.CreateSubOrchestrationInstance( + typeof(RewindLaggingWriterChild), + input.ChildInstanceId, + input.SuspendBeforeFailure); + } + } + + [KnownType(typeof(Activities.Hello))] + [KnownType(typeof(Activities.RewindLaggingWriterBlocking))] + public class RewindLaggingWriterChild : TaskOrchestration + { + public static bool ShouldFail = true; + + public override async Task RunTask( + OrchestrationContext context, + bool suspendBeforeFailure) + { + Type activityType = suspendBeforeFailure + ? typeof(Activities.RewindLaggingWriterBlocking) + : typeof(Activities.Hello); + string result = await context.ScheduleTask(activityType, "lagging"); + if (ShouldFail) + { + throw new Exception("Simulating a delayed child failure write."); + } + + return result; + } + } + + public class RewindLaggingWriterInput + { + public string ChildInstanceId { get; set; } + + public bool SuspendBeforeFailure { get; set; } + } + + [KnownType(typeof(RewindSequentialChild))] + public class RewindSequentialParent : TaskOrchestration + { + public override async Task RunTask( + OrchestrationContext context, + int childCount) + { + var results = new string[childCount]; + for (int i = 0; i < childCount; i++) + { + results[i] = await context.CreateSubOrchestrationInstance( + typeof(RewindSequentialChild), + i); + } + + return string.Join(",", results); + } + } + + [KnownType(typeof(RewindSequentialParent))] + public class RewindSequentialChild : TaskOrchestration + { + public static int FailingInput = 0; + + public override Task RunTask( + OrchestrationContext context, + int input) + { + if (input == FailingInput) + { + throw new Exception($"Simulating child {input} failure."); + } + + return Task.FromResult(input.ToString()); + } + } + [KnownType(typeof(Orchestrations.ParentWorkflowSubOrchestrationActivityFail))] [KnownType(typeof(Activities.HelloFailSubOrchestrationActivity))] public class ChildWorkflowSubOrchestrationActivityFail : TaskOrchestration @@ -5559,6 +6881,21 @@ public override async Task RunTask(OrchestrationContext context, string } } + internal class RewindLargeFailure : TaskOrchestration + { + public static bool ShouldFail = true; + + public override Task RunTask(OrchestrationContext context, string message) + { + if (ShouldFail) + { + throw new Exception(message); + } + + return Task.FromResult("Done"); + } + } + [KnownType(typeof(Activities.Throw))] internal class Throw : TaskOrchestration { @@ -5887,6 +7224,36 @@ public override Task RunTask(OrchestrationContext context, string inpu static class Activities { + internal class RewindLaggingWriterBlocking : TaskActivity + { + static TaskCompletionSource started; + static ManualResetEventSlim release; + + public static Task Started => started.Task; + + public static void Reset() + { + started = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + release = new ManualResetEventSlim(initialState: false); + } + + public static void Release() + { + release?.Set(); + } + + protected override string Execute(TaskContext context, string input) + { + started.TrySetResult(null); + if (!release.Wait(StandardTimeout)) + { + throw new TimeoutException("Timed out waiting to release the lagging-writer activity."); + } + + return $"Hello, {input}!"; + } + } + internal class HelloFailActivity : TaskActivity { public static bool ShouldFail = true; diff --git a/test/DurableTask.AzureStorage.Tests/AzureTableTrackingStoreTest.cs b/test/DurableTask.AzureStorage.Tests/AzureTableTrackingStoreTest.cs index 5c1b402d..df8a005e 100644 --- a/test/DurableTask.AzureStorage.Tests/AzureTableTrackingStoreTest.cs +++ b/test/DurableTask.AzureStorage.Tests/AzureTableTrackingStoreTest.cs @@ -127,6 +127,569 @@ public async Task QueryStatus_WithContinuationToken_NoInputToken() Assert.IsNull(actual[2].ParentInstance); } + [TestMethod] + public async Task UpdateStatusForRewind_ReplacesFullEntityUsingCurrentEtag() + { + const string TableName = "MockTable"; + const string ConnectionString = "UseDevelopmentStorage=true"; + const string InstanceId = "rewind-instance"; + const string ExecutionId = "execution-1"; + const string PreservedProperty = "preserved"; + using var tokenSource = new CancellationTokenSource(); + + var settings = new AzureStorageOrchestrationServiceSettings + { + StorageAccountClientProvider = new StorageAccountClientProvider(ConnectionString), + }; + + var azureStorageClient = new AzureStorageClient(settings); + var tableServiceClient = new Mock(MockBehavior.Strict, ConnectionString); + var tableClient = new Mock(MockBehavior.Loose, ConnectionString, TableName); + tableClient.Setup(t => t.Name).Returns(TableName); + tableServiceClient.Setup(t => t.GetTableClient(TableName)).Returns(tableClient.Object); + + var storedEntity = new TableEntity(InstanceId, string.Empty) + { + ETag = new ETag("current-etag"), + ["ExecutionId"] = ExecutionId, + ["RuntimeStatus"] = OrchestrationStatus.Failed.ToString(), + ["Output"] = "stale output", + ["PreservedProperty"] = PreservedProperty, + }; + tableClient + .Setup(t => t.QueryAsync( + It.IsAny(), + It.IsAny(), + It.IsAny>(), + tokenSource.Token)) + .Returns(AsyncPageable.FromPages( + new[] + { + Page.FromValues( + new[] { storedEntity }, + continuationToken: null, + new Mock().Object), + })); + + TableEntity replacedEntity = null; + ETag replaceEtag = default; + TableUpdateMode updateMode = default; + tableClient + .Setup(t => t.UpdateEntityAsync( + It.IsAny(), + It.IsAny(), + It.IsAny(), + tokenSource.Token)) + .Callback((entity, etag, mode, _) => + { + replacedEntity = entity; + replaceEtag = etag; + updateMode = mode; + }) + .ReturnsAsync(new Mock().Object); + + var table = new Table(azureStorageClient, tableServiceClient.Object, TableName); + var trackingStore = new AzureTableTrackingStore(new AzureStorageOrchestrationServiceStats(), table); + + await trackingStore.UpdateStatusForRewindAsync( + InstanceId, + ExecutionId, + storedEntity.ETag, + tokenSource.Token); + + Assert.AreEqual(TableUpdateMode.Replace, updateMode); + Assert.AreEqual(storedEntity.ETag, replaceEtag); + Assert.AreEqual(PreservedProperty, replacedEntity["PreservedProperty"]); + Assert.AreEqual(OrchestrationStatus.Pending.ToString(), replacedEntity["RuntimeStatus"]); + Assert.IsFalse(replacedEntity.ContainsKey("Output")); + } + + [DataTestMethod] + [DataRow(OrchestrationStatus.Pending)] + [DataRow(OrchestrationStatus.Running)] + [DataRow(OrchestrationStatus.Suspended)] + public async Task UpdateStatusForRewind_ResetsUnchangedPreFailureProjection( + OrchestrationStatus currentStatus) + { + const string TableName = "MockTable"; + const string ConnectionString = "UseDevelopmentStorage=true"; + const string InstanceId = "rewind-instance"; + const string ExecutionId = "execution-1"; + using var tokenSource = new CancellationTokenSource(); + + var settings = new AzureStorageOrchestrationServiceSettings + { + StorageAccountClientProvider = new StorageAccountClientProvider(ConnectionString), + }; + var azureStorageClient = new AzureStorageClient(settings); + var tableServiceClient = new Mock(MockBehavior.Strict, ConnectionString); + var tableClient = new Mock(MockBehavior.Loose, ConnectionString, TableName); + tableClient.Setup(t => t.Name).Returns(TableName); + tableServiceClient.Setup(t => t.GetTableClient(TableName)).Returns(tableClient.Object); + + var currentEntity = new TableEntity(InstanceId, string.Empty) + { + ETag = new ETag("rewind-start-etag"), + ["ExecutionId"] = ExecutionId, + ["RuntimeStatus"] = currentStatus.ToString(), + }; + if (currentStatus != OrchestrationStatus.Pending) + { + currentEntity["Output"] = "pre-failure output"; + } + + tableClient + .Setup(t => t.QueryAsync( + It.IsAny(), + It.IsAny(), + It.IsAny>(), + tokenSource.Token)) + .Returns(AsyncPageableFromEntity(currentEntity)); + + TableEntity replacedEntity = null; + tableClient + .Setup(t => t.UpdateEntityAsync( + It.IsAny(), + currentEntity.ETag, + TableUpdateMode.Replace, + tokenSource.Token)) + .Callback( + (entity, _, _, _) => replacedEntity = entity) + .ReturnsAsync(new Mock().Object); + + var table = new Table(azureStorageClient, tableServiceClient.Object, TableName); + var trackingStore = new AzureTableTrackingStore(new AzureStorageOrchestrationServiceStats(), table); + + await trackingStore.UpdateStatusForRewindAsync( + InstanceId, + ExecutionId, + currentEntity.ETag, + tokenSource.Token); + + Assert.IsNotNull(replacedEntity); + Assert.AreEqual(OrchestrationStatus.Pending.ToString(), replacedEntity["RuntimeStatus"]); + Assert.IsFalse(replacedEntity.ContainsKey("Output")); + } + + [DataTestMethod] + [DataRow(OrchestrationStatus.Pending, true, false)] + [DataRow(OrchestrationStatus.Running, true, true)] + [DataRow(OrchestrationStatus.Suspended, true, true)] + [DataRow(OrchestrationStatus.Completed, true, true)] + [DataRow(OrchestrationStatus.Completed, false, true)] + [DataRow(OrchestrationStatus.ContinuedAsNew, true, true)] + [DataRow(OrchestrationStatus.ContinuedAsNew, false, true)] + [DataRow(OrchestrationStatus.Canceled, true, true)] + [DataRow(OrchestrationStatus.Canceled, false, true)] + [DataRow(OrchestrationStatus.Terminated, true, true)] + [DataRow(OrchestrationStatus.Terminated, false, true)] + public async Task UpdateStatusForRewind_PreservesAdvancedState( + OrchestrationStatus currentStatus, + bool changedSinceRewindStarted, + bool hasOutput) + { + const string TableName = "MockTable"; + const string ConnectionString = "UseDevelopmentStorage=true"; + const string InstanceId = "rewind-instance"; + const string ExecutionId = "execution-1"; + var rewindStartETag = new ETag("rewind-start-etag"); + using var tokenSource = new CancellationTokenSource(); + + var settings = new AzureStorageOrchestrationServiceSettings + { + StorageAccountClientProvider = new StorageAccountClientProvider(ConnectionString), + }; + var azureStorageClient = new AzureStorageClient(settings); + var tableServiceClient = new Mock(MockBehavior.Strict, ConnectionString); + var tableClient = new Mock(MockBehavior.Loose, ConnectionString, TableName); + tableClient.Setup(t => t.Name).Returns(TableName); + tableServiceClient.Setup(t => t.GetTableClient(TableName)).Returns(tableClient.Object); + + var currentEntity = new TableEntity(InstanceId, string.Empty) + { + ETag = changedSinceRewindStarted ? new ETag("advanced-etag") : rewindStartETag, + ["ExecutionId"] = ExecutionId, + ["RuntimeStatus"] = currentStatus.ToString(), + }; + if (hasOutput) + { + currentEntity["Output"] = "new output"; + } + + tableClient + .Setup(t => t.QueryAsync( + It.IsAny(), + It.IsAny(), + It.IsAny>(), + tokenSource.Token)) + .Returns(AsyncPageableFromEntity(currentEntity)); + + var table = new Table(azureStorageClient, tableServiceClient.Object, TableName); + var trackingStore = new AzureTableTrackingStore(new AzureStorageOrchestrationServiceStats(), table); + + await trackingStore.UpdateStatusForRewindAsync( + InstanceId, + ExecutionId, + rewindStartETag, + tokenSource.Token); + + Assert.AreEqual(hasOutput, currentEntity.ContainsKey("Output")); + tableClient.Verify( + t => t.UpdateEntityAsync( + It.IsAny(), + It.IsAny(), + It.IsAny(), + It.IsAny()), + Times.Never); + } + + [DataTestMethod] + [DataRow(OrchestrationStatus.Failed)] + [DataRow(OrchestrationStatus.Pending)] + public async Task UpdateStatusForRewind_ResetsChangedNonEquivalentProjection( + OrchestrationStatus currentStatus) + { + const string TableName = "MockTable"; + const string ConnectionString = "UseDevelopmentStorage=true"; + const string InstanceId = "rewind-instance"; + const string ExecutionId = "execution-1"; + var rewindStartETag = new ETag("rewind-start-etag"); + using var tokenSource = new CancellationTokenSource(); + + var settings = new AzureStorageOrchestrationServiceSettings + { + StorageAccountClientProvider = new StorageAccountClientProvider(ConnectionString), + }; + var azureStorageClient = new AzureStorageClient(settings); + var tableServiceClient = new Mock(MockBehavior.Strict, ConnectionString); + var tableClient = new Mock(MockBehavior.Loose, ConnectionString, TableName); + tableClient.Setup(t => t.Name).Returns(TableName); + tableServiceClient.Setup(t => t.GetTableClient(TableName)).Returns(tableClient.Object); + + var currentEntity = new TableEntity(InstanceId, string.Empty) + { + ETag = new ETag("changed-etag"), + ["ExecutionId"] = ExecutionId, + ["RuntimeStatus"] = currentStatus.ToString(), + ["Output"] = "concurrent output", + }; + tableClient + .Setup(t => t.QueryAsync( + It.IsAny(), + It.IsAny(), + It.IsAny>(), + tokenSource.Token)) + .Returns(AsyncPageableFromEntity(currentEntity)); + TableEntity replacedEntity = null; + tableClient + .Setup(t => t.UpdateEntityAsync( + It.IsAny(), + currentEntity.ETag, + TableUpdateMode.Replace, + tokenSource.Token)) + .Callback( + (entity, _, _, _) => replacedEntity = entity) + .ReturnsAsync(new Mock().Object); + + var table = new Table(azureStorageClient, tableServiceClient.Object, TableName); + var trackingStore = new AzureTableTrackingStore(new AzureStorageOrchestrationServiceStats(), table); + + await trackingStore.UpdateStatusForRewindAsync( + InstanceId, + ExecutionId, + rewindStartETag, + tokenSource.Token); + + Assert.IsNotNull(replacedEntity); + Assert.AreEqual(OrchestrationStatus.Pending.ToString(), replacedEntity["RuntimeStatus"]); + Assert.IsFalse(replacedEntity.ContainsKey("Output")); + } + + [DataTestMethod] + [DataRow(OrchestrationStatus.Failed)] + [DataRow(OrchestrationStatus.Pending)] + public async Task UpdateStatusForRewind_PropagatesEtagConflict( + OrchestrationStatus currentStatus) + { + const string TableName = "MockTable"; + const string ConnectionString = "UseDevelopmentStorage=true"; + const string InstanceId = "rewind-instance"; + const string ExecutionId = "execution-1"; + using var tokenSource = new CancellationTokenSource(); + + var settings = new AzureStorageOrchestrationServiceSettings + { + StorageAccountClientProvider = new StorageAccountClientProvider(ConnectionString), + }; + + var azureStorageClient = new AzureStorageClient(settings); + var tableServiceClient = new Mock(MockBehavior.Strict, ConnectionString); + var tableClient = new Mock(MockBehavior.Loose, ConnectionString, TableName); + tableClient.Setup(t => t.Name).Returns(TableName); + tableServiceClient.Setup(t => t.GetTableClient(TableName)).Returns(tableClient.Object); + + var staleEntity = new TableEntity(InstanceId, string.Empty) + { + ETag = new ETag("stale-etag"), + ["ExecutionId"] = ExecutionId, + ["RuntimeStatus"] = OrchestrationStatus.Failed.ToString(), + ["Output"] = "stale output", + }; + var currentEntity = new TableEntity(InstanceId, string.Empty) + { + ETag = new ETag("current-etag"), + ["ExecutionId"] = ExecutionId, + ["RuntimeStatus"] = currentStatus.ToString(), + ["Output"] = "current failure output", + }; + tableClient + .SetupSequence(t => t.QueryAsync( + It.IsAny(), + It.IsAny(), + It.IsAny>(), + tokenSource.Token)) + .Returns(AsyncPageableFromEntity(staleEntity)) + .Returns(AsyncPageableFromEntity(currentEntity)); + tableClient + .Setup(t => t.UpdateEntityAsync( + It.IsAny(), + staleEntity.ETag, + TableUpdateMode.Replace, + tokenSource.Token)) + .ThrowsAsync(new RequestFailedException(412, "The entity changed.")); + + var table = new Table(azureStorageClient, tableServiceClient.Object, TableName); + var trackingStore = new AzureTableTrackingStore(new AzureStorageOrchestrationServiceStats(), table); + + await Assert.ThrowsExceptionAsync( + () => trackingStore.UpdateStatusForRewindAsync( + InstanceId, + ExecutionId, + staleEntity.ETag, + tokenSource.Token)); + tableClient.Verify( + t => t.UpdateEntityAsync( + It.IsAny(), + staleEntity.ETag, + TableUpdateMode.Replace, + tokenSource.Token), + Times.Once); + tableClient.Verify( + t => t.UpdateEntityAsync( + It.IsAny(), + ETag.All, + It.IsAny(), + It.IsAny()), + Times.Never); + } + + [TestMethod] + public async Task UpdateStatusForRewind_RejectsDifferentExecutionBeforeWrite() + { + const string TableName = "MockTable"; + const string ConnectionString = "UseDevelopmentStorage=true"; + const string InstanceId = "rewind-instance"; + const string ExpectedExecutionId = "execution-1"; + const string CurrentExecutionId = "execution-2"; + using var tokenSource = new CancellationTokenSource(); + + var settings = new AzureStorageOrchestrationServiceSettings + { + StorageAccountClientProvider = new StorageAccountClientProvider(ConnectionString), + }; + var azureStorageClient = new AzureStorageClient(settings); + var tableServiceClient = new Mock(MockBehavior.Strict, ConnectionString); + var tableClient = new Mock(MockBehavior.Loose, ConnectionString, TableName); + tableClient.Setup(t => t.Name).Returns(TableName); + tableServiceClient.Setup(t => t.GetTableClient(TableName)).Returns(tableClient.Object); + + var currentEntity = new TableEntity(InstanceId, string.Empty) + { + ETag = new ETag("current-etag"), + ["ExecutionId"] = CurrentExecutionId, + ["RuntimeStatus"] = OrchestrationStatus.Completed.ToString(), + ["Output"] = "new output", + }; + tableClient + .Setup(t => t.QueryAsync( + It.IsAny(), + It.IsAny(), + It.IsAny>(), + tokenSource.Token)) + .Returns(AsyncPageableFromEntity(currentEntity)); + + var table = new Table(azureStorageClient, tableServiceClient.Object, TableName); + var trackingStore = new AzureTableTrackingStore(new AzureStorageOrchestrationServiceStats(), table); + + DurableTaskStorageException conflict = + await Assert.ThrowsExceptionAsync( + () => trackingStore.UpdateStatusForRewindAsync( + InstanceId, + ExpectedExecutionId, + new ETag("rewind-start-etag"), + tokenSource.Token)); + + StringAssert.Contains(conflict.Message, ExpectedExecutionId); + StringAssert.Contains(conflict.Message, CurrentExecutionId); + tableClient.Verify( + t => t.UpdateEntityAsync( + It.IsAny(), + It.IsAny(), + It.IsAny(), + It.IsAny()), + Times.Never); + } + + [DataTestMethod] + [DataRow(OrchestrationStatus.Pending, false)] + [DataRow(OrchestrationStatus.Running, true)] + [DataRow(OrchestrationStatus.Completed, true)] + public async Task UpdateStatusForRewind_AcceptsEquivalentEtagConflict( + OrchestrationStatus currentStatus, + bool hasNewOutput) + { + const string TableName = "MockTable"; + const string ConnectionString = "UseDevelopmentStorage=true"; + const string InstanceId = "rewind-instance"; + const string ExecutionId = "execution-1"; + using var tokenSource = new CancellationTokenSource(); + + var settings = new AzureStorageOrchestrationServiceSettings + { + StorageAccountClientProvider = new StorageAccountClientProvider(ConnectionString), + }; + var azureStorageClient = new AzureStorageClient(settings); + var tableServiceClient = new Mock(MockBehavior.Strict, ConnectionString); + var tableClient = new Mock(MockBehavior.Loose, ConnectionString, TableName); + tableClient.Setup(t => t.Name).Returns(TableName); + tableServiceClient.Setup(t => t.GetTableClient(TableName)).Returns(tableClient.Object); + + var staleEntity = new TableEntity(InstanceId, string.Empty) + { + ETag = new ETag("stale-etag"), + ["ExecutionId"] = ExecutionId, + ["RuntimeStatus"] = OrchestrationStatus.Failed.ToString(), + ["Output"] = "stale output", + }; + var currentEntity = new TableEntity(InstanceId, string.Empty) + { + ETag = new ETag("current-etag"), + ["ExecutionId"] = ExecutionId, + ["RuntimeStatus"] = currentStatus.ToString(), + }; + if (hasNewOutput) + { + currentEntity["Output"] = "new output"; + } + + tableClient + .SetupSequence(t => t.QueryAsync( + It.IsAny(), + It.IsAny(), + It.IsAny>(), + tokenSource.Token)) + .Returns(AsyncPageableFromEntity(staleEntity)) + .Returns(AsyncPageableFromEntity(currentEntity)); + tableClient + .Setup(t => t.UpdateEntityAsync( + It.IsAny(), + staleEntity.ETag, + TableUpdateMode.Replace, + tokenSource.Token)) + .ThrowsAsync(new RequestFailedException(412, "The entity changed.")); + + var table = new Table(azureStorageClient, tableServiceClient.Object, TableName); + var trackingStore = new AzureTableTrackingStore(new AzureStorageOrchestrationServiceStats(), table); + + await trackingStore.UpdateStatusForRewindAsync( + InstanceId, + ExecutionId, + staleEntity.ETag, + tokenSource.Token); + + Assert.AreEqual(hasNewOutput, currentEntity.ContainsKey("Output")); + tableClient.Verify( + t => t.UpdateEntityAsync( + It.IsAny(), + It.IsAny(), + It.IsAny(), + It.IsAny()), + Times.Once); + } + + [TestMethod] + public async Task UpdateStatusForRewind_RejectsExecutionChangeAfterEtagConflict() + { + const string TableName = "MockTable"; + const string ConnectionString = "UseDevelopmentStorage=true"; + const string InstanceId = "rewind-instance"; + const string ExpectedExecutionId = "execution-1"; + const string CurrentExecutionId = "execution-2"; + using var tokenSource = new CancellationTokenSource(); + + var settings = new AzureStorageOrchestrationServiceSettings + { + StorageAccountClientProvider = new StorageAccountClientProvider(ConnectionString), + }; + var azureStorageClient = new AzureStorageClient(settings); + var tableServiceClient = new Mock(MockBehavior.Strict, ConnectionString); + var tableClient = new Mock(MockBehavior.Loose, ConnectionString, TableName); + tableClient.Setup(t => t.Name).Returns(TableName); + tableServiceClient.Setup(t => t.GetTableClient(TableName)).Returns(tableClient.Object); + + var staleEntity = new TableEntity(InstanceId, string.Empty) + { + ETag = new ETag("stale-etag"), + ["ExecutionId"] = ExpectedExecutionId, + ["RuntimeStatus"] = OrchestrationStatus.Failed.ToString(), + ["Output"] = "stale output", + }; + var currentEntity = new TableEntity(InstanceId, string.Empty) + { + ETag = new ETag("current-etag"), + ["ExecutionId"] = CurrentExecutionId, + ["RuntimeStatus"] = OrchestrationStatus.Completed.ToString(), + ["Output"] = "new output", + }; + tableClient + .SetupSequence(t => t.QueryAsync( + It.IsAny(), + It.IsAny(), + It.IsAny>(), + tokenSource.Token)) + .Returns(AsyncPageableFromEntity(staleEntity)) + .Returns(AsyncPageableFromEntity(currentEntity)); + tableClient + .Setup(t => t.UpdateEntityAsync( + It.IsAny(), + staleEntity.ETag, + TableUpdateMode.Replace, + tokenSource.Token)) + .ThrowsAsync(new RequestFailedException(412, "The entity changed.")); + + var table = new Table(azureStorageClient, tableServiceClient.Object, TableName); + var trackingStore = new AzureTableTrackingStore(new AzureStorageOrchestrationServiceStats(), table); + + DurableTaskStorageException conflict = + await Assert.ThrowsExceptionAsync( + () => trackingStore.UpdateStatusForRewindAsync( + InstanceId, + ExpectedExecutionId, + staleEntity.ETag, + tokenSource.Token)); + + StringAssert.Contains(conflict.Message, ExpectedExecutionId); + StringAssert.Contains(conflict.Message, CurrentExecutionId); + Assert.AreEqual("new output", currentEntity["Output"]); + tableClient.Verify( + t => t.UpdateEntityAsync( + It.IsAny(), + It.IsAny(), + It.IsAny(), + It.IsAny()), + Times.Once); + } + [TestMethod] public async Task InstanceStoreBackedTrackingStore_PersistsParentOnCreation() { @@ -164,5 +727,17 @@ public async Task InstanceStoreBackedTrackingStore_PersistsParentOnCreation() Assert.AreSame(startedEvent.ParentInstance, writtenState.State.ParentInstance); Assert.AreEqual(ParentInstanceId, writtenState.State.ParentInstance.OrchestrationInstance.InstanceId); } + + static AsyncPageable AsyncPageableFromEntity(TableEntity entity) + { + return AsyncPageable.FromPages( + new[] + { + Page.FromValues( + new[] { entity }, + continuationToken: null, + new Mock().Object), + }); + } } } diff --git a/test/DurableTask.AzureStorage.Tests/RewindOutputTrackingStoreTests.cs b/test/DurableTask.AzureStorage.Tests/RewindOutputTrackingStoreTests.cs new file mode 100644 index 00000000..9e2e65c3 --- /dev/null +++ b/test/DurableTask.AzureStorage.Tests/RewindOutputTrackingStoreTests.cs @@ -0,0 +1,1432 @@ +// ---------------------------------------------------------------------------------- +// 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.Collections.Generic; + using System.Linq; + using System.Net.Http; + using System.Threading; + using System.Threading.Tasks; + using Azure; + using Azure.Core; + using Azure.Core.Pipeline; + using Azure.Data.Tables; + using DurableTask.AzureStorage.Storage; + using DurableTask.AzureStorage.Tracking; + using DurableTask.Core; + using DurableTask.Core.History; + using Microsoft.VisualStudio.TestTools.UnitTesting; + + [TestClass] + public class RewindOutputTrackingStoreTests + { + const string PreservedProperty = "PreservedProperty"; + + string taskHubName; + AzureTableTrackingStore trackingStore; + RecordingRequestHandler tableRequestRecorder; + TransportClientProvider tableClientProvider; + + [TestInitialize] + public async Task Initialize() + { + this.taskHubName = "rewind" + Guid.NewGuid().ToString("N").Substring(0, 9); + AzureStorageOrchestrationServiceSettings settings = + TestHelpers.GetTestAzureStorageOrchestrationServiceSettings(enableExtendedSessions: false); + settings.TaskHubName = this.taskHubName; + var defaultProvider = settings.StorageAccountClientProvider; + this.tableRequestRecorder = new RecordingRequestHandler(); + this.tableClientProvider = + new TransportClientProvider( + defaultProvider.Table, + this.tableRequestRecorder); + settings.StorageAccountClientProvider = new StorageAccountClientProvider( + defaultProvider.Blob, + defaultProvider.Queue, + this.tableClientProvider); + + var azureStorageClient = new AzureStorageClient(settings); + var messageManager = new MessageManager( + settings, + azureStorageClient, + $"{this.taskHubName}-largemessages".ToLowerInvariant()); + this.trackingStore = new AzureTableTrackingStore(azureStorageClient, messageManager); + await this.trackingStore.CreateAsync(); + } + + [TestCleanup] + public async Task Cleanup() + { + try + { + if (this.trackingStore != null) + { + await this.trackingStore.DeleteAsync(); + } + } + finally + { + this.tableClientProvider?.Dispose(); + this.tableRequestRecorder?.Dispose(); + } + } + + [TestMethod] + public async Task UpdateStatusForRewind_RemovesPersistedOutput() + { + string instanceId = $"output-{Guid.NewGuid():N}"; + await this.SeedInstanceRowAsync(instanceId, OrchestrationStatus.Failed, output: "old failure"); + + TableEntity failed = await this.GetRawEntityAsync(instanceId); + Assert.AreEqual("old failure", failed["Output"]); + + await this.trackingStore.UpdateStatusForRewindAsync( + instanceId, + "execution-1", + failed.ETag); + + TableEntity rewound = await this.GetRawEntityAsync(instanceId); + Assert.AreEqual(OrchestrationStatus.Pending.ToString(), rewound["RuntimeStatus"]); + Assert.IsFalse(rewound.ContainsKey("Output")); + Assert.AreEqual("preserve me", rewound[PreservedProperty]); + } + + [TestMethod] + public async Task UpdateStatusForRewind_IsIdempotentWhenOutputIsMissing() + { + string instanceId = $"missing-{Guid.NewGuid():N}"; + await this.SeedInstanceRowAsync(instanceId, OrchestrationStatus.Failed, output: null); + + TableEntity failed = await this.GetRawEntityAsync(instanceId); + await this.trackingStore.UpdateStatusForRewindAsync( + instanceId, + "execution-1", + failed.ETag); + TableEntity pending = await this.GetRawEntityAsync(instanceId); + await this.trackingStore.UpdateStatusForRewindAsync( + instanceId, + "execution-1", + pending.ETag); + + TableEntity rewound = await this.GetRawEntityAsync(instanceId); + Assert.AreEqual(OrchestrationStatus.Pending.ToString(), rewound["RuntimeStatus"]); + Assert.IsFalse(rewound.ContainsKey("Output")); + } + + [DataTestMethod] + [DataRow(OrchestrationStatus.Completed)] + [DataRow(OrchestrationStatus.Failed)] + public async Task TerminalWriteAfterRewind_PersistsNewOutput(OrchestrationStatus terminalStatus) + { + string instanceId = $"complete-{Guid.NewGuid():N}"; + const string ExecutionId = "execution-1"; + await this.SeedInstanceRowAsync(instanceId, OrchestrationStatus.Failed, output: "old failure"); + TableEntity failed = await this.GetRawEntityAsync(instanceId); + await this.trackingStore.UpdateStatusForRewindAsync(instanceId, ExecutionId, failed.ETag); + + var runtimeState = new OrchestrationRuntimeState(); + runtimeState.AddEvent(CreateExecutionStartedEvent(instanceId, ExecutionId)); + runtimeState.AddEvent(new ExecutionCompletedEvent(-1, "new output", terminalStatus)); + + await this.trackingStore.UpdateInstanceStatusForCompletedOrchestrationAsync( + instanceId, + ExecutionId, + runtimeState, + instanceEntityExists: true); + + TableEntity completed = await this.GetRawEntityAsync(instanceId); + Assert.AreEqual(terminalStatus.ToString(), completed["RuntimeStatus"]); + Assert.AreEqual("new output", completed["Output"]); + } + + [TestMethod] + public async Task SetNewExecution_ReplacesPersistedOutput() + { + string instanceId = $"reuse-{Guid.NewGuid():N}"; + await this.SeedInstanceRowAsync(instanceId, OrchestrationStatus.Completed, output: "old output"); + TableEntity existing = await this.GetRawEntityAsync(instanceId); + + bool created = await this.trackingStore.SetNewExecutionAsync( + CreateExecutionStartedEvent(instanceId, "execution-2"), + existing.ETag, + inputPayloadOverride: null); + + Assert.IsTrue(created); + TableEntity pending = await this.GetRawEntityAsync(instanceId); + Assert.AreEqual(OrchestrationStatus.Pending.ToString(), pending["RuntimeStatus"]); + Assert.IsFalse(pending.ContainsKey("Output")); + } + + [DataTestMethod] + [DataRow(OrchestrationStatus.Pending, true, false, false, true)] + [DataRow(OrchestrationStatus.Pending, false, false, false, false)] + [DataRow(OrchestrationStatus.Pending, true, false, true, false)] + [DataRow(OrchestrationStatus.Pending, true, true, false, false)] + [DataRow(OrchestrationStatus.Running, true, false, true, false)] + [DataRow(OrchestrationStatus.Suspended, true, false, true, false)] + [DataRow(OrchestrationStatus.Completed, true, false, true, false)] + [DataRow(OrchestrationStatus.Canceled, true, false, true, false)] + [DataRow(OrchestrationStatus.Terminated, true, false, true, false)] + [DataRow(OrchestrationStatus.ContinuedAsNew, true, false, true, false)] + [DataRow(OrchestrationStatus.Failed, true, false, true, true)] + [DataRow(OrchestrationStatus.Failed, true, true, true, false)] + public async Task RewindHistory_RediscoversOnlyCurrentRewoundChild( + OrchestrationStatus childStatus, + bool markerMatchesCurrentExecution, + bool hasCurrentFailure, + bool hasOutput, + bool expectsChildTarget) + { + string parentInstanceId = $"parent-{Guid.NewGuid():N}"; + string childInstanceId = $"child-{Guid.NewGuid():N}"; + const string ParentExecutionId = "parent-execution"; + const string ChildExecutionId = "child-execution"; + const int TaskScheduledId = 0; + + await this.SeedInstanceRowAsync( + parentInstanceId, + OrchestrationStatus.Failed, + output: "parent failure", + executionId: ParentExecutionId); + await this.SeedHistoryRowAsync( + parentInstanceId, + "0000000000000000", + ParentExecutionId, + EventType.OrchestratorStarted); + await this.SeedHistoryRowAsync( + parentInstanceId, + "0000000000000001", + ParentExecutionId, + EventType.SubOrchestrationInstanceCreated, + eventId: TaskScheduledId, + childInstanceId: childInstanceId); + await this.SeedHistoryRowAsync( + parentInstanceId, + "0000000000000002", + ParentExecutionId, + EventType.GenericEvent, + taskScheduledId: TaskScheduledId, + reason: "Rewound: " + nameof(EventType.SubOrchestrationInstanceFailed)); + await this.SeedHistoryRowAsync( + parentInstanceId, + "0000000000000003", + ParentExecutionId, + EventType.GenericEvent, + reason: "Rewound: " + nameof(EventType.ExecutionCompleted), + orchestrationStatus: OrchestrationStatus.Failed); + + await this.SeedInstanceRowAsync( + childInstanceId, + childStatus, + output: hasOutput ? "child output" : null, + executionId: ChildExecutionId); + await this.SeedExecutionStartedRowAsync( + childInstanceId, + ChildExecutionId, + parentInstanceId, + ParentExecutionId, + TaskScheduledId); + await this.SeedHistoryRowAsync( + childInstanceId, + "0000000000000000", + ChildExecutionId, + EventType.OrchestratorStarted); + await this.SeedHistoryRowAsync( + childInstanceId, + "0000000000000001", + markerMatchesCurrentExecution ? ChildExecutionId : "old-child-execution", + EventType.GenericEvent, + reason: "Rewound: " + nameof(EventType.ExecutionCompleted), + orchestrationStatus: OrchestrationStatus.Failed); + if (hasCurrentFailure) + { + await this.SeedHistoryRowAsync( + childInstanceId, + "0000000000000002", + ChildExecutionId, + EventType.ExecutionCompleted, + reason: "new failure", + orchestrationStatus: OrchestrationStatus.Failed); + } + + TableEntity childBefore = await this.GetRawEntityAsync(childInstanceId); + string[] targets = (await this.trackingStore + .RewindHistoryAsync(parentInstanceId) + .ToListAsync()) + .ToArray(); + TableEntity childAfter = await this.GetRawEntityAsync(childInstanceId); + + CollectionAssert.AreEqual( + expectsChildTarget ? new[] { childInstanceId } : new[] { parentInstanceId }, + targets); + Assert.AreEqual( + expectsChildTarget ? OrchestrationStatus.Pending.ToString() : childStatus.ToString(), + childAfter["RuntimeStatus"]); + Assert.AreEqual( + expectsChildTarget, + childBefore.ETag != childAfter.ETag, + "Only a proven stranded child should receive a new write fence."); + if (expectsChildTarget || !hasOutput) + { + Assert.IsFalse(childAfter.ContainsKey("Output")); + } + else + { + Assert.AreEqual("child output", childAfter["Output"]); + } + } + + [DataTestMethod] + [DataRow(OrchestrationStatus.Running, false)] + [DataRow(OrchestrationStatus.Suspended, false)] + [DataRow(OrchestrationStatus.Failed, true)] + public async Task RewindHistory_TraversesActiveIntermediateToPendingDescendant( + OrchestrationStatus intermediateStatus, + bool resetsIntermediate) + { + string parentInstanceId = $"parent-{Guid.NewGuid():N}"; + string intermediateInstanceId = $"intermediate-{Guid.NewGuid():N}"; + string leafInstanceId = $"leaf-{Guid.NewGuid():N}"; + const string ParentExecutionId = "parent-execution"; + const string IntermediateExecutionId = "intermediate-execution"; + const string LeafExecutionId = "leaf-execution"; + + await this.SeedInstanceRowAsync( + parentInstanceId, + OrchestrationStatus.Failed, + output: "parent failure", + executionId: ParentExecutionId); + await this.SeedRewoundParentHistoryAsync( + parentInstanceId, + ParentExecutionId, + intermediateInstanceId); + + await this.SeedInstanceRowAsync( + intermediateInstanceId, + intermediateStatus, + output: "intermediate progress", + executionId: IntermediateExecutionId); + await this.SeedExecutionStartedRowAsync( + intermediateInstanceId, + IntermediateExecutionId, + parentInstanceId, + ParentExecutionId, + taskScheduleId: 0); + await this.SeedRewoundParentHistoryAsync( + intermediateInstanceId, + IntermediateExecutionId, + leafInstanceId); + + await this.SeedInstanceRowAsync( + leafInstanceId, + OrchestrationStatus.Pending, + output: null, + executionId: LeafExecutionId); + await this.SeedExecutionStartedRowAsync( + leafInstanceId, + LeafExecutionId, + intermediateInstanceId, + IntermediateExecutionId, + taskScheduleId: 0); + await this.SeedHistoryRowAsync( + leafInstanceId, + "0000000000000000", + LeafExecutionId, + EventType.OrchestratorStarted); + await this.SeedHistoryRowAsync( + leafInstanceId, + "0000000000000001", + LeafExecutionId, + EventType.GenericEvent, + reason: "Rewound: " + nameof(EventType.ExecutionCompleted), + orchestrationStatus: OrchestrationStatus.Failed); + + TableEntity intermediateBefore = + await this.GetRawEntityAsync(intermediateInstanceId); + TableEntity leafBefore = await this.GetRawEntityAsync(leafInstanceId); + + string[] targets = (await this.trackingStore + .RewindHistoryAsync(parentInstanceId) + .ToListAsync()) + .ToArray(); + + CollectionAssert.AreEqual(new[] { leafInstanceId }, targets); + TableEntity parentAfter = await this.GetRawEntityAsync(parentInstanceId); + Assert.AreEqual(OrchestrationStatus.Pending.ToString(), parentAfter["RuntimeStatus"]); + Assert.IsFalse(parentAfter.ContainsKey("Output")); + + TableEntity intermediateAfter = + await this.GetRawEntityAsync(intermediateInstanceId); + Assert.AreEqual(resetsIntermediate, intermediateBefore.ETag != intermediateAfter.ETag); + Assert.AreEqual( + resetsIntermediate ? OrchestrationStatus.Pending.ToString() : intermediateStatus.ToString(), + intermediateAfter["RuntimeStatus"]); + if (resetsIntermediate) + { + Assert.IsFalse(intermediateAfter.ContainsKey("Output")); + } + else + { + Assert.AreEqual("intermediate progress", intermediateAfter["Output"]); + } + + TableEntity leafAfter = await this.GetRawEntityAsync(leafInstanceId); + Assert.AreNotEqual(leafBefore.ETag, leafAfter.ETag); + Assert.AreEqual(OrchestrationStatus.Pending.ToString(), leafAfter["RuntimeStatus"]); + Assert.IsFalse(leafAfter.ContainsKey("Output")); + } + + [DataTestMethod] + [DataRow(OrchestrationStatus.Running)] + [DataRow(OrchestrationStatus.Suspended)] + public async Task RewindHistory_ActiveIntermediateWithoutRecoverableDescendantFallsBackToParent( + OrchestrationStatus intermediateStatus) + { + string parentInstanceId = $"parent-{Guid.NewGuid():N}"; + string intermediateInstanceId = $"intermediate-{Guid.NewGuid():N}"; + string completedLeafInstanceId = $"leaf-{Guid.NewGuid():N}"; + const string ParentExecutionId = "parent-execution"; + const string IntermediateExecutionId = "intermediate-execution"; + const string LeafExecutionId = "leaf-execution"; + + await this.SeedInstanceRowAsync( + parentInstanceId, + OrchestrationStatus.Failed, + output: "parent failure", + executionId: ParentExecutionId); + await this.SeedRewoundParentHistoryAsync( + parentInstanceId, + ParentExecutionId, + intermediateInstanceId); + + await this.SeedInstanceRowAsync( + intermediateInstanceId, + intermediateStatus, + output: "intermediate progress", + executionId: IntermediateExecutionId); + await this.SeedExecutionStartedRowAsync( + intermediateInstanceId, + IntermediateExecutionId, + parentInstanceId, + ParentExecutionId, + taskScheduleId: 0); + await this.SeedRewoundParentHistoryAsync( + intermediateInstanceId, + IntermediateExecutionId, + completedLeafInstanceId); + + await this.SeedInstanceRowAsync( + completedLeafInstanceId, + OrchestrationStatus.Completed, + output: "completed output", + executionId: LeafExecutionId); + await this.SeedExecutionStartedRowAsync( + completedLeafInstanceId, + LeafExecutionId, + intermediateInstanceId, + IntermediateExecutionId, + taskScheduleId: 0); + await this.SeedHistoryRowAsync( + completedLeafInstanceId, + "0000000000000000", + LeafExecutionId, + EventType.OrchestratorStarted); + await this.SeedHistoryRowAsync( + completedLeafInstanceId, + "0000000000000001", + LeafExecutionId, + EventType.GenericEvent, + reason: "Rewound: " + nameof(EventType.ExecutionCompleted), + orchestrationStatus: OrchestrationStatus.Failed); + + TableEntity intermediateBefore = + await this.GetRawEntityAsync(intermediateInstanceId); + TableEntity completedLeafBefore = + await this.GetRawEntityAsync(completedLeafInstanceId); + + string[] targets = (await this.trackingStore + .RewindHistoryAsync(parentInstanceId) + .ToListAsync()) + .ToArray(); + + CollectionAssert.AreEqual(new[] { parentInstanceId }, targets); + TableEntity parentAfter = await this.GetRawEntityAsync(parentInstanceId); + Assert.AreEqual(OrchestrationStatus.Pending.ToString(), parentAfter["RuntimeStatus"]); + Assert.IsFalse(parentAfter.ContainsKey("Output")); + + TableEntity intermediateAfter = + await this.GetRawEntityAsync(intermediateInstanceId); + Assert.AreEqual(intermediateBefore.ETag, intermediateAfter.ETag); + Assert.AreEqual(intermediateStatus.ToString(), intermediateAfter["RuntimeStatus"]); + Assert.AreEqual("intermediate progress", intermediateAfter["Output"]); + + TableEntity completedLeafAfter = + await this.GetRawEntityAsync(completedLeafInstanceId); + Assert.AreEqual(completedLeafBefore.ETag, completedLeafAfter.ETag); + Assert.AreEqual(OrchestrationStatus.Completed.ToString(), completedLeafAfter["RuntimeStatus"]); + Assert.AreEqual("completed output", completedLeafAfter["Output"]); + } + + [DataTestMethod] + [DataRow(OrchestrationStatus.Running, OrchestrationStatus.Failed)] + [DataRow(OrchestrationStatus.Running, OrchestrationStatus.Completed)] + [DataRow(OrchestrationStatus.Suspended, OrchestrationStatus.Failed)] + [DataRow(OrchestrationStatus.Suspended, OrchestrationStatus.Completed)] + public async Task RewindHistory_DoesNotTraverseActiveIntermediateWithFreshCompletion( + OrchestrationStatus intermediateStatus, + OrchestrationStatus completionStatus) + { + string parentInstanceId = $"parent-{Guid.NewGuid():N}"; + string intermediateInstanceId = $"intermediate-{Guid.NewGuid():N}"; + string oldLeafInstanceId = $"leaf-{Guid.NewGuid():N}"; + const string ParentExecutionId = "parent-execution"; + const string IntermediateExecutionId = "intermediate-execution"; + const string LeafExecutionId = "leaf-execution"; + + await this.SeedInstanceRowAsync( + parentInstanceId, + OrchestrationStatus.Failed, + output: "parent failure", + executionId: ParentExecutionId); + await this.SeedRewoundParentHistoryAsync( + parentInstanceId, + ParentExecutionId, + intermediateInstanceId); + + await this.SeedInstanceRowAsync( + intermediateInstanceId, + intermediateStatus, + output: "intermediate progress", + executionId: IntermediateExecutionId); + await this.SeedExecutionStartedRowAsync( + intermediateInstanceId, + IntermediateExecutionId, + parentInstanceId, + ParentExecutionId, + taskScheduleId: 0); + await this.SeedRewoundParentHistoryAsync( + intermediateInstanceId, + IntermediateExecutionId, + oldLeafInstanceId); + await this.SeedHistoryRowAsync( + intermediateInstanceId, + "0000000000000004", + IntermediateExecutionId, + EventType.ExecutionCompleted, + reason: "new completion", + orchestrationStatus: completionStatus); + + await this.SeedInstanceRowAsync( + oldLeafInstanceId, + OrchestrationStatus.Pending, + output: null, + executionId: LeafExecutionId); + await this.SeedExecutionStartedRowAsync( + oldLeafInstanceId, + LeafExecutionId, + intermediateInstanceId, + IntermediateExecutionId, + taskScheduleId: 0); + await this.SeedHistoryRowAsync( + oldLeafInstanceId, + "0000000000000000", + LeafExecutionId, + EventType.OrchestratorStarted); + await this.SeedHistoryRowAsync( + oldLeafInstanceId, + "0000000000000001", + LeafExecutionId, + EventType.GenericEvent, + reason: "Rewound: " + nameof(EventType.ExecutionCompleted), + orchestrationStatus: OrchestrationStatus.Failed); + + TableEntity intermediateBefore = + await this.GetRawEntityAsync(intermediateInstanceId); + TableEntity oldLeafBefore = + await this.GetRawEntityAsync(oldLeafInstanceId); + + string[] targets = (await this.trackingStore + .RewindHistoryAsync(parentInstanceId) + .ToListAsync()) + .ToArray(); + + CollectionAssert.AreEqual(new[] { parentInstanceId }, targets); + TableEntity intermediateAfter = + await this.GetRawEntityAsync(intermediateInstanceId); + Assert.AreEqual(intermediateBefore.ETag, intermediateAfter.ETag); + Assert.AreEqual(intermediateStatus.ToString(), intermediateAfter["RuntimeStatus"]); + Assert.AreEqual("intermediate progress", intermediateAfter["Output"]); + + TableEntity oldLeafAfter = await this.GetRawEntityAsync(oldLeafInstanceId); + Assert.AreEqual(oldLeafBefore.ETag, oldLeafAfter.ETag); + Assert.AreEqual(OrchestrationStatus.Pending.ToString(), oldLeafAfter["RuntimeStatus"]); + Assert.IsFalse(oldLeafAfter.ContainsKey("Output")); + } + + [DataTestMethod] + [DataRow(OrchestrationStatus.Completed)] + [DataRow(OrchestrationStatus.ContinuedAsNew)] + [DataRow(OrchestrationStatus.Terminated)] + public async Task RewindHistory_DoesNotTraverseTerminalOrReusedIntermediate( + OrchestrationStatus intermediateStatus) + { + string parentInstanceId = $"parent-{Guid.NewGuid():N}"; + string intermediateInstanceId = $"intermediate-{Guid.NewGuid():N}"; + string strandedLeafInstanceId = $"leaf-{Guid.NewGuid():N}"; + const string ParentExecutionId = "parent-execution"; + const string IntermediateExecutionId = "intermediate-execution"; + const string LeafExecutionId = "leaf-execution"; + + await this.SeedInstanceRowAsync( + parentInstanceId, + OrchestrationStatus.Failed, + output: "parent failure", + executionId: ParentExecutionId); + await this.SeedRewoundParentHistoryAsync( + parentInstanceId, + ParentExecutionId, + intermediateInstanceId); + + await this.SeedInstanceRowAsync( + intermediateInstanceId, + intermediateStatus, + output: "new execution output", + executionId: IntermediateExecutionId); + await this.SeedExecutionStartedRowAsync( + intermediateInstanceId, + IntermediateExecutionId, + parentInstanceId: "unrelated-parent", + parentExecutionId: "unrelated-parent-execution", + taskScheduleId: 0); + await this.SeedRewoundParentHistoryAsync( + intermediateInstanceId, + "old-intermediate-execution", + strandedLeafInstanceId); + + await this.SeedInstanceRowAsync( + strandedLeafInstanceId, + OrchestrationStatus.Pending, + output: null, + executionId: LeafExecutionId); + await this.SeedExecutionStartedRowAsync( + strandedLeafInstanceId, + LeafExecutionId, + intermediateInstanceId, + "old-intermediate-execution", + taskScheduleId: 0); + await this.SeedHistoryRowAsync( + strandedLeafInstanceId, + "0000000000000000", + LeafExecutionId, + EventType.OrchestratorStarted); + await this.SeedHistoryRowAsync( + strandedLeafInstanceId, + "0000000000000001", + LeafExecutionId, + EventType.GenericEvent, + reason: "Rewound: " + nameof(EventType.ExecutionCompleted), + orchestrationStatus: OrchestrationStatus.Failed); + + TableEntity intermediateBefore = + await this.GetRawEntityAsync(intermediateInstanceId); + TableEntity strandedLeafBefore = + await this.GetRawEntityAsync(strandedLeafInstanceId); + + string[] targets = (await this.trackingStore + .RewindHistoryAsync(parentInstanceId) + .ToListAsync()) + .ToArray(); + + CollectionAssert.AreEqual(new[] { parentInstanceId }, targets); + TableEntity intermediateAfter = + await this.GetRawEntityAsync(intermediateInstanceId); + Assert.AreEqual(intermediateBefore.ETag, intermediateAfter.ETag); + Assert.AreEqual(intermediateStatus.ToString(), intermediateAfter["RuntimeStatus"]); + Assert.AreEqual("new execution output", intermediateAfter["Output"]); + + TableEntity strandedLeafAfter = + await this.GetRawEntityAsync(strandedLeafInstanceId); + Assert.AreEqual(strandedLeafBefore.ETag, strandedLeafAfter.ETag); + Assert.AreEqual(OrchestrationStatus.Pending.ToString(), strandedLeafAfter["RuntimeStatus"]); + Assert.IsFalse(strandedLeafAfter.ContainsKey("Output")); + } + + [DataTestMethod] + [DataRow(false, true, true)] + [DataRow(true, false, true)] + [DataRow(true, true, false)] + public async Task RewindHistory_DoesNotRecoverChildOwnedByDifferentParentEdge( + bool parentInstanceMatches, + bool parentExecutionMatches, + bool taskScheduleMatches) + { + string parentInstanceId = $"parent-{Guid.NewGuid():N}"; + string childInstanceId = $"child-{Guid.NewGuid():N}"; + const string ParentExecutionId = "parent-execution"; + const string ChildExecutionId = "reused-child-execution"; + const int TaskScheduledId = 0; + + await this.SeedInstanceRowAsync( + parentInstanceId, + OrchestrationStatus.Failed, + output: "parent failure", + executionId: ParentExecutionId); + await this.SeedRewoundParentHistoryAsync( + parentInstanceId, + ParentExecutionId, + childInstanceId); + + await this.SeedInstanceRowAsync( + childInstanceId, + OrchestrationStatus.Pending, + output: null, + executionId: ChildExecutionId); + await this.SeedExecutionStartedRowAsync( + childInstanceId, + ChildExecutionId, + parentInstanceMatches ? parentInstanceId : "unrelated-parent", + parentExecutionMatches ? ParentExecutionId : "unrelated-parent-execution", + taskScheduleMatches ? TaskScheduledId : TaskScheduledId + 1); + await this.SeedHistoryRowAsync( + childInstanceId, + "0000000000000000", + ChildExecutionId, + EventType.OrchestratorStarted); + const string RewoundCompletionRowKey = "0000000000000001"; + await this.SeedHistoryRowAsync( + childInstanceId, + RewoundCompletionRowKey, + ChildExecutionId, + EventType.GenericEvent, + reason: "Rewound: " + nameof(EventType.ExecutionCompleted), + orchestrationStatus: OrchestrationStatus.Failed); + + TableEntity childBefore = await this.GetRawEntityAsync(childInstanceId); + TableEntity completionBefore = + await this.GetRawHistoryEntityAsync(childInstanceId, RewoundCompletionRowKey); + + string[] targets = (await this.trackingStore + .RewindHistoryAsync(parentInstanceId) + .ToListAsync()) + .ToArray(); + + CollectionAssert.AreEqual(new[] { parentInstanceId }, targets); + TableEntity childAfter = await this.GetRawEntityAsync(childInstanceId); + Assert.AreEqual(childBefore.ETag, childAfter.ETag); + Assert.AreEqual(ChildExecutionId, childAfter["ExecutionId"]); + Assert.AreEqual(OrchestrationStatus.Pending.ToString(), childAfter["RuntimeStatus"]); + Assert.IsFalse(childAfter.ContainsKey("Output")); + + TableEntity completionAfter = + await this.GetRawHistoryEntityAsync(childInstanceId, RewoundCompletionRowKey); + Assert.AreEqual(completionBefore.ETag, completionAfter.ETag); + Assert.AreEqual(nameof(EventType.GenericEvent), completionAfter["EventType"]); + Assert.AreEqual( + "Rewound: " + nameof(EventType.ExecutionCompleted), + completionAfter["Reason"]); + } + + [TestMethod] + public async Task RewindHistory_DeduplicatesRepeatedRecoveredEdgesBeforeRecursing() + { + string parentInstanceId = $"parent-{Guid.NewGuid():N}"; + string intermediateInstanceId = $"intermediate-{Guid.NewGuid():N}"; + string leafInstanceId = $"leaf-{Guid.NewGuid():N}"; + const string ParentExecutionId = "parent-execution"; + const string IntermediateExecutionId = "intermediate-execution"; + const string LeafExecutionId = "leaf-execution"; + + await this.SeedInstanceRowAsync( + parentInstanceId, + OrchestrationStatus.Failed, + output: "parent failure", + executionId: ParentExecutionId); + await this.SeedRewoundParentHistoryAsync( + parentInstanceId, + ParentExecutionId, + intermediateInstanceId); + await this.SeedHistoryRowAsync( + parentInstanceId, + "0000000000000004", + ParentExecutionId, + EventType.GenericEvent, + taskScheduledId: 0, + reason: "Rewound: " + nameof(EventType.SubOrchestrationInstanceFailed)); + + await this.SeedInstanceRowAsync( + intermediateInstanceId, + OrchestrationStatus.Running, + output: "intermediate progress", + executionId: IntermediateExecutionId); + await this.SeedExecutionStartedRowAsync( + intermediateInstanceId, + IntermediateExecutionId, + parentInstanceId, + ParentExecutionId, + taskScheduleId: 0); + await this.SeedRewoundParentHistoryAsync( + intermediateInstanceId, + IntermediateExecutionId, + leafInstanceId); + await this.SeedHistoryRowAsync( + intermediateInstanceId, + "0000000000000004", + IntermediateExecutionId, + EventType.GenericEvent, + taskScheduledId: 0, + reason: "Rewound: " + nameof(EventType.SubOrchestrationInstanceFailed)); + + await this.SeedInstanceRowAsync( + leafInstanceId, + OrchestrationStatus.Pending, + output: null, + executionId: LeafExecutionId); + await this.SeedExecutionStartedRowAsync( + leafInstanceId, + LeafExecutionId, + intermediateInstanceId, + IntermediateExecutionId, + taskScheduleId: 0); + await this.SeedHistoryRowAsync( + leafInstanceId, + "0000000000000000", + LeafExecutionId, + EventType.OrchestratorStarted); + await this.SeedHistoryRowAsync( + leafInstanceId, + "0000000000000001", + LeafExecutionId, + EventType.GenericEvent, + reason: "Rewound: " + nameof(EventType.ExecutionCompleted), + orchestrationStatus: OrchestrationStatus.Failed); + + this.tableRequestRecorder.Clear(); + string[] targets = (await this.trackingStore + .RewindHistoryAsync(parentInstanceId) + .ToListAsync()) + .ToArray(); + + int leafOrchestratorQueries = this.tableRequestRecorder.CountRequests(request => + request.Method == HttpMethod.Get && + request.Uri.AbsolutePath.IndexOf( + this.trackingStore.HistoryTable.Name, + StringComparison.OrdinalIgnoreCase) >= 0 && + Uri.UnescapeDataString(request.Uri.Query).Contains(leafInstanceId) && + Uri.UnescapeDataString(request.Uri.Query).Contains(nameof(EventType.OrchestratorStarted))); + int leafResets = this.tableRequestRecorder.CountRequests(request => + request.Method == HttpMethod.Put && + request.Uri.AbsolutePath.IndexOf( + this.trackingStore.InstancesTable.Name, + StringComparison.OrdinalIgnoreCase) >= 0 && + Uri.UnescapeDataString(request.Uri.AbsoluteUri).Contains(leafInstanceId)); + + Assert.AreEqual( + 1, + leafOrchestratorQueries, + $"Expected one leaf traversal, observed {leafOrchestratorQueries} queries, {leafResets} resets, and {targets.Length} targets."); + Assert.AreEqual(1, leafResets); + CollectionAssert.AreEqual(new[] { leafInstanceId }, targets); + } + + [DataTestMethod] + [DataRow(false)] + [DataRow(true)] + public async Task RewindHistory_DeduplicatesMixedLiveAndRecoveredEdge(bool liveFailureFirst) + { + string parentInstanceId = $"parent-{Guid.NewGuid():N}"; + string childInstanceId = $"child-{Guid.NewGuid():N}"; + const string ParentExecutionId = "parent-execution"; + const string ChildExecutionId = "child-execution"; + const string FirstFailureRowKey = "0000000000000002"; + const string SecondFailureRowKey = "0000000000000003"; + + await this.SeedInstanceRowAsync( + parentInstanceId, + OrchestrationStatus.Failed, + output: "parent failure", + executionId: ParentExecutionId); + await this.SeedHistoryRowAsync( + parentInstanceId, + "0000000000000000", + ParentExecutionId, + EventType.OrchestratorStarted); + await this.SeedHistoryRowAsync( + parentInstanceId, + "0000000000000001", + ParentExecutionId, + EventType.SubOrchestrationInstanceCreated, + eventId: 0, + childInstanceId: childInstanceId); + await this.SeedHistoryRowAsync( + parentInstanceId, + FirstFailureRowKey, + ParentExecutionId, + liveFailureFirst ? EventType.SubOrchestrationInstanceFailed : EventType.GenericEvent, + taskScheduledId: 0, + reason: liveFailureFirst + ? "current failure" + : "Rewound: " + nameof(EventType.SubOrchestrationInstanceFailed)); + await this.SeedHistoryRowAsync( + parentInstanceId, + SecondFailureRowKey, + ParentExecutionId, + liveFailureFirst ? EventType.GenericEvent : EventType.SubOrchestrationInstanceFailed, + taskScheduledId: 0, + reason: liveFailureFirst + ? "Rewound: " + nameof(EventType.SubOrchestrationInstanceFailed) + : "current failure"); + await this.SeedHistoryRowAsync( + parentInstanceId, + "0000000000000004", + ParentExecutionId, + EventType.GenericEvent, + reason: "Rewound: " + nameof(EventType.ExecutionCompleted), + orchestrationStatus: OrchestrationStatus.Failed); + + await this.SeedInstanceRowAsync( + childInstanceId, + OrchestrationStatus.Pending, + output: null, + executionId: ChildExecutionId); + await this.SeedExecutionStartedRowAsync( + childInstanceId, + ChildExecutionId, + parentInstanceId, + ParentExecutionId, + taskScheduleId: 0); + await this.SeedHistoryRowAsync( + childInstanceId, + "0000000000000000", + ChildExecutionId, + EventType.OrchestratorStarted); + await this.SeedHistoryRowAsync( + childInstanceId, + "0000000000000001", + ChildExecutionId, + EventType.GenericEvent, + reason: "Rewound: " + nameof(EventType.ExecutionCompleted), + orchestrationStatus: OrchestrationStatus.Failed); + + this.tableRequestRecorder.Clear(); + string[] targets = (await this.trackingStore + .RewindHistoryAsync(parentInstanceId) + .ToListAsync()) + .ToArray(); + + int childOrchestratorQueries = this.tableRequestRecorder.CountRequests(request => + request.Method == HttpMethod.Get && + request.Uri.AbsolutePath.IndexOf( + this.trackingStore.HistoryTable.Name, + StringComparison.OrdinalIgnoreCase) >= 0 && + Uri.UnescapeDataString(request.Uri.Query).Contains(childInstanceId) && + Uri.UnescapeDataString(request.Uri.Query).Contains(nameof(EventType.OrchestratorStarted))); + int childResets = this.tableRequestRecorder.CountRequests(request => + request.Method == HttpMethod.Put && + request.Uri.AbsolutePath.IndexOf( + this.trackingStore.InstancesTable.Name, + StringComparison.OrdinalIgnoreCase) >= 0 && + Uri.UnescapeDataString(request.Uri.AbsoluteUri).Contains(childInstanceId)); + + int expectedChildRewinds = liveFailureFirst ? 1 : 2; + Assert.AreEqual(expectedChildRewinds, childOrchestratorQueries); + Assert.AreEqual(expectedChildRewinds, childResets); + CollectionAssert.AreEqual(new[] { childInstanceId }, targets); + + string liveFailureRowKey = liveFailureFirst ? FirstFailureRowKey : SecondFailureRowKey; + TableEntity liveFailure = + await this.GetRawHistoryEntityAsync(parentInstanceId, liveFailureRowKey); + Assert.AreEqual(nameof(EventType.GenericEvent), liveFailure["EventType"]); + Assert.AreEqual( + "Rewound: " + nameof(EventType.SubOrchestrationInstanceFailed), + liveFailure["Reason"]); + } + + [DataTestMethod] + [DataRow(false)] + [DataRow(true)] + public async Task RewindHistory_LiveFailureProcessesFreshCheckpointAfterRecoveredEdge( + bool recoveredEdgeProducedTarget) + { + string parentInstanceId = $"parent-{Guid.NewGuid():N}"; + string childInstanceId = $"child-{Guid.NewGuid():N}"; + const string ParentExecutionId = "parent-execution"; + const string ChildExecutionId = "child-execution"; + const string FreshCompletionRowKey = "0000000000000002"; + + await this.SeedInstanceRowAsync( + parentInstanceId, + OrchestrationStatus.Failed, + output: "parent failure", + executionId: ParentExecutionId); + await this.SeedHistoryRowAsync( + parentInstanceId, + "0000000000000000", + ParentExecutionId, + EventType.OrchestratorStarted); + await this.SeedHistoryRowAsync( + parentInstanceId, + "0000000000000001", + ParentExecutionId, + EventType.SubOrchestrationInstanceCreated, + eventId: 0, + childInstanceId: childInstanceId); + await this.SeedHistoryRowAsync( + parentInstanceId, + "0000000000000002", + ParentExecutionId, + EventType.GenericEvent, + taskScheduledId: 0, + reason: "Rewound: " + nameof(EventType.SubOrchestrationInstanceFailed)); + await this.SeedHistoryRowAsync( + parentInstanceId, + "0000000000000003", + ParentExecutionId, + EventType.SubOrchestrationInstanceFailed, + taskScheduledId: 0, + reason: "current failure"); + await this.SeedHistoryRowAsync( + parentInstanceId, + "0000000000000004", + ParentExecutionId, + EventType.GenericEvent, + reason: "Rewound: " + nameof(EventType.ExecutionCompleted), + orchestrationStatus: OrchestrationStatus.Failed); + + await this.SeedInstanceRowAsync( + childInstanceId, + recoveredEdgeProducedTarget + ? OrchestrationStatus.Pending + : OrchestrationStatus.Running, + output: null, + executionId: ChildExecutionId); + await this.SeedExecutionStartedRowAsync( + childInstanceId, + ChildExecutionId, + parentInstanceId, + ParentExecutionId, + taskScheduleId: 0); + await this.SeedHistoryRowAsync( + childInstanceId, + "0000000000000000", + ChildExecutionId, + EventType.OrchestratorStarted); + await this.SeedHistoryRowAsync( + childInstanceId, + "0000000000000001", + ChildExecutionId, + EventType.GenericEvent, + reason: "Rewound: " + nameof(EventType.ExecutionCompleted), + orchestrationStatus: OrchestrationStatus.Failed); + + this.tableRequestRecorder.Arm(request => + request.Method == HttpMethod.Put && + request.RequestUri.AbsolutePath.IndexOf( + this.trackingStore.HistoryTable.Name, + StringComparison.OrdinalIgnoreCase) >= 0 && + Uri.UnescapeDataString(request.RequestUri.AbsoluteUri).Contains(parentInstanceId)); + Task rewind = RewindAsync(); + + try + { + await this.tableRequestRecorder.WaitUntilBlockedAsync(); + + TableEntity runningChild = await this.GetRawEntityAsync(childInstanceId); + runningChild["RuntimeStatus"] = OrchestrationStatus.Running.ToString(); + runningChild["Output"] = "fresh in-flight output"; + await this.trackingStore.InstancesTable.ReplaceEntityAsync( + runningChild, + runningChild.ETag); + await this.SeedHistoryRowAsync( + childInstanceId, + FreshCompletionRowKey, + ChildExecutionId, + EventType.ExecutionCompleted, + reason: "fresh failure", + orchestrationStatus: OrchestrationStatus.Failed); + + // Count only work performed after the fresh child checkpoint is visible. + this.tableRequestRecorder.Clear(); + this.tableRequestRecorder.Release(); + + string[] targets = await rewind; + int childOrchestratorQueries = this.tableRequestRecorder.CountRequests(request => + request.Method == HttpMethod.Get && + request.Uri.AbsolutePath.IndexOf( + this.trackingStore.HistoryTable.Name, + StringComparison.OrdinalIgnoreCase) >= 0 && + Uri.UnescapeDataString(request.Uri.Query).Contains(childInstanceId) && + Uri.UnescapeDataString(request.Uri.Query).Contains(nameof(EventType.OrchestratorStarted))); + int childResets = this.tableRequestRecorder.CountRequests(request => + request.Method == HttpMethod.Put && + request.Uri.AbsolutePath.IndexOf( + this.trackingStore.InstancesTable.Name, + StringComparison.OrdinalIgnoreCase) >= 0 && + Uri.UnescapeDataString(request.Uri.AbsoluteUri).Contains(childInstanceId)); + + Assert.AreEqual(1, childOrchestratorQueries); + Assert.AreEqual(1, childResets); + CollectionAssert.AreEqual(new[] { childInstanceId }, targets); + + TableEntity childAfter = await this.GetRawEntityAsync(childInstanceId); + Assert.AreEqual(OrchestrationStatus.Pending.ToString(), childAfter["RuntimeStatus"]); + Assert.IsFalse(childAfter.ContainsKey("Output")); + + TableEntity freshCompletion = + await this.GetRawHistoryEntityAsync(childInstanceId, FreshCompletionRowKey); + Assert.AreEqual(nameof(EventType.GenericEvent), freshCompletion["EventType"]); + Assert.AreEqual( + "Rewound: " + nameof(EventType.ExecutionCompleted), + freshCompletion["Reason"]); + } + finally + { + this.tableRequestRecorder.Release(); + } + + async Task RewindAsync() + { + return (await this.trackingStore + .RewindHistoryAsync(parentInstanceId) + .ToListAsync()) + .ToArray(); + } + } + + async Task SeedRewoundParentHistoryAsync( + string instanceId, + string executionId, + string childInstanceId) + { + await this.SeedHistoryRowAsync( + instanceId, + "0000000000000000", + executionId, + EventType.OrchestratorStarted); + await this.SeedHistoryRowAsync( + instanceId, + "0000000000000001", + executionId, + EventType.SubOrchestrationInstanceCreated, + eventId: 0, + childInstanceId: childInstanceId); + await this.SeedHistoryRowAsync( + instanceId, + "0000000000000002", + executionId, + EventType.GenericEvent, + taskScheduledId: 0, + reason: "Rewound: " + nameof(EventType.SubOrchestrationInstanceFailed)); + await this.SeedHistoryRowAsync( + instanceId, + "0000000000000003", + executionId, + EventType.GenericEvent, + reason: "Rewound: " + nameof(EventType.ExecutionCompleted), + orchestrationStatus: OrchestrationStatus.Failed); + } + + async Task SeedExecutionStartedRowAsync( + string instanceId, + string executionId, + string parentInstanceId, + string parentExecutionId, + int taskScheduleId) + { + var executionStarted = new ExecutionStartedEvent(-1, "input") + { + Name = "TestOrchestration", + OrchestrationInstance = new OrchestrationInstance + { + InstanceId = instanceId, + ExecutionId = executionId, + }, + ParentInstance = new ParentInstance + { + OrchestrationInstance = new OrchestrationInstance + { + InstanceId = parentInstanceId, + ExecutionId = parentExecutionId, + }, + TaskScheduleId = taskScheduleId, + }, + }; + TableEntity entity = TableEntityConverter.Serialize(executionStarted); + entity.PartitionKey = KeySanitation.EscapePartitionKey(instanceId); + entity.RowKey = "execution-start"; + entity[nameof(OrchestrationInstance.ExecutionId)] = executionId; + await this.trackingStore.HistoryTable.InsertEntityAsync(entity); + } + + async Task SeedInstanceRowAsync( + string instanceId, + OrchestrationStatus status, + string output, + string executionId = "execution-1") + { + var entity = new TableEntity(KeySanitation.EscapePartitionKey(instanceId), string.Empty) + { + ["Name"] = "TestOrchestration", + ["RuntimeStatus"] = status.ToString(), + ["CreatedTime"] = DateTime.UtcNow, + ["LastUpdatedTime"] = DateTime.UtcNow, + ["TaskHubName"] = this.taskHubName, + ["ExecutionId"] = executionId, + [PreservedProperty] = "preserve me", + }; + + if (output != null) + { + entity["Output"] = output; + } + + await this.trackingStore.InstancesTable.InsertEntityAsync(entity); + } + + async Task SeedHistoryRowAsync( + string instanceId, + string rowKey, + string executionId, + EventType eventType, + int? eventId = null, + int? taskScheduledId = null, + string childInstanceId = null, + string reason = null, + OrchestrationStatus? orchestrationStatus = null) + { + var entity = new TableEntity(KeySanitation.EscapePartitionKey(instanceId), rowKey) + { + [nameof(OrchestrationInstance.ExecutionId)] = executionId, + [nameof(HistoryEvent.EventType)] = eventType.ToString(), + }; + + if (eventId.HasValue) + { + entity[nameof(HistoryEvent.EventId)] = eventId.Value; + } + + if (taskScheduledId.HasValue) + { + entity[nameof(TaskCompletedEvent.TaskScheduledId)] = taskScheduledId.Value; + } + + if (childInstanceId != null) + { + entity[nameof(OrchestrationInstance.InstanceId)] = childInstanceId; + } + + if (reason != null) + { + entity[nameof(TaskFailedEvent.Reason)] = reason; + } + + if (orchestrationStatus.HasValue) + { + entity[nameof(ExecutionCompletedEvent.OrchestrationStatus)] = + orchestrationStatus.Value.ToString(); + } + + await this.trackingStore.HistoryTable.InsertEntityAsync(entity); + } + + async Task GetRawEntityAsync(string instanceId) + { + string filter = $"{AzureTableQueryFilter.PartitionKeyEquals(instanceId)} and " + + $"{AzureTableQueryFilter.ColumnEquals(nameof(ITableEntity.RowKey), string.Empty)}"; + return await this.trackingStore.InstancesTable + .ExecuteQueryAsync(filter, 1) + .FirstOrDefaultAsync(); + } + + async Task GetRawHistoryEntityAsync(string instanceId, string rowKey) + { + string filter = $"{AzureTableQueryFilter.PartitionKeyEquals(instanceId)} and " + + $"{AzureTableQueryFilter.ColumnEquals(nameof(ITableEntity.RowKey), rowKey)}"; + return await this.trackingStore.HistoryTable + .ExecuteQueryAsync(filter, 1) + .FirstOrDefaultAsync(); + } + + static ExecutionStartedEvent CreateExecutionStartedEvent(string instanceId, string executionId) + { + return new ExecutionStartedEvent(-1, "input") + { + Name = "TestOrchestration", + Version = string.Empty, + OrchestrationInstance = new OrchestrationInstance + { + InstanceId = instanceId, + ExecutionId = executionId, + }, + }; + } + + sealed class TransportClientProvider : + IStorageServiceClientProvider, + IDisposable + where TOptions : ClientOptions + { + readonly IStorageServiceClientProvider inner; + readonly HttpClientTransport transport; + + public TransportClientProvider( + IStorageServiceClientProvider inner, + HttpMessageHandler handler) + { + this.inner = inner; + this.transport = new HttpClientTransport( + new HttpClient(handler, disposeHandler: false)); + } + + public TOptions CreateOptions() + { + TOptions options = this.inner.CreateOptions(); + options.Transport = this.transport; + return options; + } + + public TClient CreateClient(TOptions options) => this.inner.CreateClient(options); + + public void Dispose() + { + this.transport.Dispose(); + } + } + + sealed class RecordingRequestHandler : DelegatingHandler + { + readonly object sync = new object(); + readonly List requests = new List(); + Func barrierPredicate; + TaskCompletionSource blocked; + TaskCompletionSource release; + bool barrierClaimed; + + public RecordingRequestHandler() + : base(new HttpClientHandler()) + { + } + + public void Clear() + { + lock (this.sync) + { + this.requests.Clear(); + } + } + + public void Arm(Func predicate) + { + lock (this.sync) + { + this.barrierPredicate = predicate; + this.blocked = new TaskCompletionSource( + TaskCreationOptions.RunContinuationsAsynchronously); + this.release = new TaskCompletionSource( + TaskCreationOptions.RunContinuationsAsynchronously); + this.barrierClaimed = false; + } + } + + public async Task WaitUntilBlockedAsync() + { + Task blockedTask; + lock (this.sync) + { + blockedTask = this.blocked.Task; + } + + Task completedTask = await Task.WhenAny( + blockedTask, + Task.Delay(TimeSpan.FromSeconds(30))); + Assert.AreSame( + blockedTask, + completedTask, + "The expected table request did not reach the barrier."); + await blockedTask; + } + + public void Release() + { + lock (this.sync) + { + this.release?.TrySetResult(null); + } + } + + public int CountRequests(Func predicate) + { + lock (this.sync) + { + return this.requests.Count(predicate); + } + } + + protected override async Task SendAsync( + HttpRequestMessage request, + CancellationToken cancellationToken) + { + Task releaseTask = null; + lock (this.sync) + { + if (!this.barrierClaimed && this.barrierPredicate?.Invoke(request) == true) + { + this.barrierClaimed = true; + this.blocked.TrySetResult(null); + releaseTask = this.release.Task; + } + } + + if (releaseTask != null) + { + await releaseTask; + } + + HttpResponseMessage response = await base.SendAsync(request, cancellationToken); + if (response.IsSuccessStatusCode) + { + lock (this.sync) + { + this.requests.Add(new RecordedRequest(request.Method, request.RequestUri)); + } + } + + return response; + } + + protected override void Dispose(bool disposing) + { + if (disposing) + { + this.Release(); + } + + base.Dispose(disposing); + } + + public sealed class RecordedRequest + { + public RecordedRequest(HttpMethod method, Uri uri) + { + this.Method = method; + this.Uri = uri; + } + + public HttpMethod Method { get; } + + public Uri Uri { get; } + } + } + } +} diff --git a/test/DurableTask.AzureStorage.Tests/TestOrchestrationHost.cs b/test/DurableTask.AzureStorage.Tests/TestOrchestrationHost.cs index 34210c00..f855c31b 100644 --- a/test/DurableTask.AzureStorage.Tests/TestOrchestrationHost.cs +++ b/test/DurableTask.AzureStorage.Tests/TestOrchestrationHost.cs @@ -49,6 +49,12 @@ public TestOrchestrationHost(AzureStorageOrchestrationServiceSettings settings, public string TaskHub => this.settings.TaskHubName; + public ErrorPropagationMode ErrorPropagationMode + { + get => this.worker.ErrorPropagationMode; + set => this.worker.ErrorPropagationMode = value; + } + public void Dispose() { this.worker.Dispose();