diff --git a/src/DurableTask.AzureStorage/Logging/LogHelper.cs b/src/DurableTask.AzureStorage/Logging/LogHelper.cs index e4ecaae1..b6334002 100644 --- a/src/DurableTask.AzureStorage/Logging/LogHelper.cs +++ b/src/DurableTask.AzureStorage/Logging/LogHelper.cs @@ -759,14 +759,15 @@ internal void GeneralWarning( string account, string taskHub, string details, - string instanceId = null) + string instanceId = null, + Exception exception = null) { var logEvent = new LogEvents.GeneralWarning( account, taskHub, details, instanceId ?? string.Empty); - this.WriteStructuredLog(logEvent); + this.WriteStructuredLog(logEvent, exception); } internal void SplitBrainDetected( diff --git a/src/DurableTask.AzureStorage/MessageManager.cs b/src/DurableTask.AzureStorage/MessageManager.cs index 1fc6d078..a19f28c6 100644 --- a/src/DurableTask.AzureStorage/MessageManager.cs +++ b/src/DurableTask.AzureStorage/MessageManager.cs @@ -253,10 +253,21 @@ public Task DownloadAndDecompressAsBytesAsync(Uri blobUri, CancellationT return DownloadAndDecompressAsBytesAsync(blob, cancellationToken); } - public Task DeleteBlobAsync(string blobName, CancellationToken cancellationToken = default) + public async Task DeleteBlobAsync(string blobName, CancellationToken cancellationToken = default) { Blob blob = this.blobContainer.GetBlobReference(blobName); - return blob.DeleteIfExistsAsync(cancellationToken); + try + { + return await blob.DeleteIfExistsAsync(cancellationToken); + } + catch (AggregateException ex) when (!cancellationToken.IsCancellationRequested) + { + throw new DurableTaskStorageException("Azure Storage retries failed while deleting a blob.", ex); + } + catch (OperationCanceledException ex) when (!cancellationToken.IsCancellationRequested) + { + throw new DurableTaskStorageException("Azure Storage timed out while deleting a blob.", ex); + } } private async Task DownloadAndDecompressAsBytesAsync(Blob blob, CancellationToken cancellationToken = default) diff --git a/src/DurableTask.AzureStorage/Tracking/AzureTableTrackingStore.cs b/src/DurableTask.AzureStorage/Tracking/AzureTableTrackingStore.cs index 167a3311..e4c5fecb 100644 --- a/src/DurableTask.AzureStorage/Tracking/AzureTableTrackingStore.cs +++ b/src/DurableTask.AzureStorage/Tracking/AzureTableTrackingStore.cs @@ -283,6 +283,7 @@ public override async IAsyncEnumerable RewindHistoryAsync(string instanc //////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////// bool hasFailedSubOrchestrations = false; + var blobsToDelete = new List(); string partitionFilter = AzureTableQueryFilter.PartitionKeyEquals(instanceId); string orchestratorStartedFilter = $"{partitionFilter} and {nameof(HistoryEvent.EventType)} eq '{nameof(EventType.OrchestratorStarted)}'"; @@ -361,6 +362,13 @@ public override async IAsyncEnumerable RewindHistoryAsync(string instanc break; } + if (entity.GetString(nameof(HistoryEvent.EventType)) == nameof(EventType.ExecutionCompleted)) + { + // GenericEvent replay ignores the terminal payload, so remove its blob references in the same ETag-guarded replace. + RemovePropertyAndTrackBlob(entity, nameof(ExecutionCompletedEvent.Result), blobsToDelete); + RemovePropertyAndTrackBlob(entity, nameof(ExecutionCompletedEvent.FailureDetails), blobsToDelete); + } + // "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.EventType)] = nameof(EventType.GenericEvent); @@ -371,12 +379,40 @@ public override async IAsyncEnumerable RewindHistoryAsync(string instanc // reset orchestration status in instance store table await this.UpdateStatusForRewindAsync(instanceId, cancellationToken); + // Delete only after both the history pointers and the Instances-table Output reference are gone. + await this.DeleteRewindBlobsAsync(instanceId, blobsToDelete, cancellationToken); + if (!hasFailedSubOrchestrations) { yield return instanceId; } } + async Task DeleteRewindBlobsAsync(string instanceId, IEnumerable blobNames, CancellationToken cancellationToken) + { + foreach (string blobName in blobNames) + { + cancellationToken.ThrowIfCancellationRequested(); + + try + { + await this.messageManager.DeleteBlobAsync(blobName, cancellationToken); + } + catch (DurableTaskStorageException ex) + { + this.settings.Logger.GeneralWarning( + this.azureStorageClient.BlobAccountName, + this.settings.TaskHubName, + $"Failed to delete unreferenced rewind blob '{blobName}'. The blob will remain until the orchestration is purged. " + + $"Storage status code: {ex.HttpStatusCode}; error code: '{ex.ErrorCode}'.", + instanceId, + ex); + } + } + + cancellationToken.ThrowIfCancellationRequested(); + } + /// public override async IAsyncEnumerable GetStateAsync(string instanceId, bool allExecutions, bool fetchInput, [EnumeratorCancellation] CancellationToken cancellationToken = default) { @@ -856,15 +892,23 @@ public override async Task SetNewExecutionAsync( /// public override async Task UpdateStatusForRewindAsync(string instanceId, CancellationToken cancellationToken = default) { - string sanitizedInstanceId = KeySanitation.EscapePartitionKey(instanceId); - TableEntity entity = new TableEntity(sanitizedInstanceId, "") + string filter = $"{AzureTableQueryFilter.PartitionKeyEquals(instanceId)} and {AzureTableQueryFilter.ColumnEquals(RowKeyProperty, string.Empty)}"; + TableEntity entity = await this.InstancesTable + .ExecuteQueryAsync(filter, 1, cancellationToken: cancellationToken) + .FirstOrDefaultAsync(); + + if (entity == null) { - ["RuntimeStatus"] = OrchestrationStatus.Pending.ToString("G"), - ["LastUpdatedTime"] = DateTime.UtcNow, - }; + throw new DurableTaskStorageException($"The orchestration instance '{instanceId}' does not exist."); + } + + // 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); + await this.InstancesTable.ReplaceEntityAsync(entity, entity.ETag, cancellationToken); // We don't have enough information to get the episode number. // It's also not important to have for this particular trace. @@ -1378,6 +1422,20 @@ static string GetBlobPropertyName(string originalPropertyName) return originalPropertyName + "BlobName"; } + static void RemovePropertyAndTrackBlob(TableEntity entity, string propertyName, List blobsToDelete) + { + string blobPropertyName = GetBlobPropertyName(propertyName); + if (entity.TryGetValue(blobPropertyName, out object value) && + value is string blobName && + !string.IsNullOrEmpty(blobName)) + { + blobsToDelete.Add(blobName); + } + + entity.Remove(propertyName); + entity.Remove(blobPropertyName); + } + static string GetBlobName(TableEntity entity, string property) { string sanitizedInstanceId = entity.PartitionKey; diff --git a/test/DurableTask.AzureStorage.Tests/AzureStorageScenarioTests.cs b/test/DurableTask.AzureStorage.Tests/AzureStorageScenarioTests.cs index a2372034..65fda267 100644 --- a/test/DurableTask.AzureStorage.Tests/AzureStorageScenarioTests.cs +++ b/test/DurableTask.AzureStorage.Tests/AzureStorageScenarioTests.cs @@ -16,6 +16,8 @@ namespace DurableTask.AzureStorage.Tests using Azure.Data.Tables; using Azure.Storage.Blobs; using Azure.Storage.Blobs.Models; + using Azure.Storage.Blobs.Specialized; + using DurableTask.AzureStorage.Logging; using DurableTask.AzureStorage.Storage; using DurableTask.AzureStorage.Tracking; using DurableTask.Core; @@ -23,6 +25,7 @@ namespace DurableTask.AzureStorage.Tests using DurableTask.Core.History; using DurableTask.Core.Settings; using Microsoft.Practices.EnterpriseLibrary.SemanticLogging.Utility; + using Microsoft.Extensions.Logging; using Microsoft.VisualStudio.TestTools.UnitTesting; using Moq; using Newtonsoft.Json; @@ -1510,6 +1513,178 @@ public async Task RewindActivityFail() } } + [DataTestMethod] + [DataRow(false)] + [DataRow(true)] + public async Task RewindLargeFailure_HandlesBlobDeletionFailure(bool failBlobDeletion) + { + var logger = new Mock(); + var loggerFactory = new Mock(); + loggerFactory + .Setup(factory => factory.CreateLogger(It.IsAny())) + .Returns(logger.Object); + + using (TestOrchestrationHost host = TestHelpers.GetTestOrchestrationHost( + enableExtendedSessions: false, + modifySettingsAction: settings => settings.LoggerFactory = loggerFactory.Object)) + { + Orchestrations.RewindLargeFailure.ShouldFail = true; + host.ErrorPropagationMode = ErrorPropagationMode.UseFailureDetails; + await host.StartAsync(); + + string failureMessage = this.GenerateMediumRandomStringPayload().ToString(); + var client = await host.StartOrchestrationAsync( + typeof(Orchestrations.RewindLargeFailure), + input: failureMessage); + OrchestrationState failed = await client.WaitForCompletionAsync(TimeSpan.FromSeconds(30)); + Assert.IsNotNull(failed); + 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"); + Assert.IsNotNull(resultBlobName); + Assert.IsTrue(new Uri(outputBlobUrl).AbsolutePath.EndsWith(resultBlobName, StringComparison.Ordinal)); + string failureDetailsBlobName = failedCompletion.GetString("FailureDetailsBlobName"); + Assert.IsNotNull(failureDetailsBlobName); + + string[] outputBlobNames = new[] + { + resultBlobName, + failureDetailsBlobName, + }; + var blobServiceClient = new BlobServiceClient(TestHelpers.GetTestStorageAccountConnectionString()); + BlobContainerClient container = blobServiceClient.GetBlobContainerClient($"{host.TaskHub.ToLowerInvariant()}-largemessages"); + foreach (string blobName in outputBlobNames) + { + Assert.IsTrue((await container.GetBlobClient(blobName).ExistsAsync()).Value); + } + + if (!failBlobDeletion) + { + using var canceled = new CancellationTokenSource(); + canceled.Cancel(); + await Assert.ThrowsExceptionAsync( + async () => await trackingStore.RewindHistoryAsync(client.InstanceId, canceled.Token).ToListAsync()); + + TableEntity instanceAfterCancellation = await trackingStore.InstancesTable + .ExecuteQueryAsync(instanceFilter, 1) + .FirstOrDefaultAsync(); + Assert.AreEqual(OrchestrationStatus.Failed.ToString(), instanceAfterCancellation.GetString("RuntimeStatus")); + Assert.AreEqual(outputBlobUrl, instanceAfterCancellation.GetString("Output")); + } + + BlobClient resultBlob = container.GetBlobClient(resultBlobName); + BlobLeaseClient resultBlobLease = null; + if (failBlobDeletion) + { + resultBlobLease = resultBlob.GetBlobLeaseClient(); + await resultBlobLease.AcquireAsync(TimeSpan.FromSeconds(15)); + } + + Orchestrations.RewindLargeFailure.ShouldFail = false; + try + { + if (failBlobDeletion) + { + await client.RewindAsync("Retry despite the persisted-output cleanup failure."); + } + else + { + CollectionAssert.AreEqual( + new[] { client.InstanceId }, + await trackingStore.RewindHistoryAsync(client.InstanceId).ToListAsync()); + } + + TableEntity rewoundInstance = await trackingStore.InstancesTable + .ExecuteQueryAsync(instanceFilter, 1) + .FirstOrDefaultAsync(); + if (failBlobDeletion) + { + Assert.AreNotEqual(outputBlobUrl, rewoundInstance.GetString("Output")); + } + else + { + 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))); + Assert.IsFalse(rewoundCompletion.ContainsKey("Result")); + Assert.IsFalse(rewoundCompletion.ContainsKey("ResultBlobName")); + Assert.IsFalse(rewoundCompletion.ContainsKey("FailureDetails")); + Assert.IsFalse(rewoundCompletion.ContainsKey("FailureDetailsBlobName")); + + Assert.AreEqual(failBlobDeletion, (await resultBlob.ExistsAsync()).Value); + Assert.IsFalse((await container.GetBlobClient(failureDetailsBlobName).ExistsAsync()).Value); + + var cleanupWarnings = logger.Invocations + .Where(invocation => invocation.Arguments.Count > 2) + .Where(invocation => invocation.Arguments[2] is LogEvents.GeneralWarning warning && + warning.Details.Contains(resultBlobName)) + .ToList(); + if (failBlobDeletion) + { + Assert.AreEqual(1, cleanupWarnings.Count); + var cleanupWarning = (LogEvents.GeneralWarning)cleanupWarnings[0].Arguments[2]; + Assert.AreEqual(client.InstanceId, cleanupWarning.InstanceId); + StringAssert.Contains(cleanupWarning.Details, "LeaseIdMissing"); + Assert.IsInstanceOfType(cleanupWarnings[0].Arguments[3], typeof(DurableTaskStorageException)); + } + else + { + Assert.AreEqual(0, cleanupWarnings.Count); + } + } + finally + { + if (resultBlobLease != null) + { + await resultBlobLease.ReleaseAsync(); + await resultBlob.DeleteIfExistsAsync(); + } + } + + if (!failBlobDeletion) + { + await client.RewindAsync("Retry the persisted-output cleanup."); + } + + OrchestrationState completed = await client.WaitForCompletionAsync(TimeSpan.FromSeconds(30)); + 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 RewindMultipleActivityFail() { @@ -4857,6 +5032,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 { diff --git a/test/DurableTask.AzureStorage.Tests/AzureTableTrackingStoreTest.cs b/test/DurableTask.AzureStorage.Tests/AzureTableTrackingStoreTest.cs index 5c1b402d..feac3aed 100644 --- a/test/DurableTask.AzureStorage.Tests/AzureTableTrackingStoreTest.cs +++ b/test/DurableTask.AzureStorage.Tests/AzureTableTrackingStoreTest.cs @@ -127,6 +127,145 @@ 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 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"), + ["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, 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")); + } + + [TestMethod] + public async Task UpdateStatusForRewind_PropagatesEtagConflict() + { + const string TableName = "MockTable"; + const string ConnectionString = "UseDevelopmentStorage=true"; + const string InstanceId = "rewind-instance"; + 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("stale-etag"), + ["RuntimeStatus"] = OrchestrationStatus.Failed.ToString(), + ["Output"] = "stale output", + }; + 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), + })); + tableClient + .Setup(t => t.UpdateEntityAsync( + It.IsAny(), + storedEntity.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, tokenSource.Token)); + tableClient.Verify( + t => t.UpdateEntityAsync( + It.IsAny(), + storedEntity.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 InstanceStoreBackedTrackingStore_PersistsParentOnCreation() { diff --git a/test/DurableTask.AzureStorage.Tests/MessageManagerTests.cs b/test/DurableTask.AzureStorage.Tests/MessageManagerTests.cs index 008a9eac..173b7d8e 100644 --- a/test/DurableTask.AzureStorage.Tests/MessageManagerTests.cs +++ b/test/DurableTask.AzureStorage.Tests/MessageManagerTests.cs @@ -13,12 +13,15 @@ #nullable enable namespace DurableTask.AzureStorage.Tests { + using Azure.Storage.Blobs; using DurableTask.AzureStorage.Storage; using DurableTask.Core.History; using Microsoft.VisualStudio.TestTools.UnitTesting; using Newtonsoft.Json; using System; using System.Collections.Generic; + using System.Threading; + using System.Threading.Tasks; [TestClass] public class MessageManagerTests @@ -87,6 +90,28 @@ public void GetBlobUrlEscaped(string blob, string blobUrl) Assert.AreEqual(expected, manager.GetBlobUrl(blob)); } + [TestMethod] + public async Task DeleteBlobAsync_NormalizesRetryExhaustion() + { + MessageManager manager = SetupUnavailableBlobMessageManager(); + + DurableTaskStorageException failure = await Assert.ThrowsExceptionAsync( + async () => await manager.DeleteBlobAsync("blob")); + + Assert.IsInstanceOfType(failure.InnerException, typeof(AggregateException)); + } + + [TestMethod] + public async Task DeleteBlobAsync_PropagatesCallerCancellation() + { + MessageManager manager = SetupUnavailableBlobMessageManager(); + using var cancellation = new CancellationTokenSource(); + cancellation.Cancel(); + + await Assert.ThrowsExceptionAsync( + async () => await manager.DeleteBlobAsync("blob", cancellation.Token)); + } + private string GetMessage(string dictionaryType) => "{\"$type\":\"DurableTask.AzureStorage.MessageData\",\"ActivityId\":\"5406d369-4369-4673-afae-6671a2fa1e57\",\"TaskMessage\":{\"$type\":\"DurableTask.Core.TaskMessage\",\"Event\":{\"$type\":\"DurableTask.Core.History.ExecutionStartedEvent\",\"OrchestrationInstance\":{\"$type\":\"DurableTask.Core.OrchestrationInstance\",\"InstanceId\":\"2.2-34a2c9d4-306e-4467-8470-a8018b2e4f11\",\"ExecutionId\":\"aae324dcc8f943e490b37ec5e5bbf9da\"},\"EventType\":0,\"ParentInstance\":null,\"Name\":\"OrchestrationName\",\"Version\":\"2.0\",\"Input\":\"input\",\"Tags\":{\"$type\":\"" + dictionaryType @@ -105,6 +130,35 @@ private MessageManager SetupMessageManager(ICustomTypeBinder binder) azureStorageClient, "$root"); } + + static MessageManager SetupUnavailableBlobMessageManager() + { + var developmentStorage = new StorageAccountClientProvider("UseDevelopmentStorage=true"); + var settings = new AzureStorageOrchestrationServiceSettings + { + StorageAccountClientProvider = new StorageAccountClientProvider( + new UnavailableBlobServiceClientProvider(), + developmentStorage.Queue, + developmentStorage.Table), + }; + + return new MessageManager(settings, new AzureStorageClient(settings), "unavailable"); + } + } + + sealed class UnavailableBlobServiceClientProvider : IStorageServiceClientProvider + { + public BlobClientOptions CreateOptions() + { + var options = new BlobClientOptions(); + options.Retry.MaxRetries = 1; + options.Retry.Delay = TimeSpan.FromMilliseconds(10); + options.Retry.NetworkTimeout = TimeSpan.FromSeconds(1); + return options; + } + + public BlobServiceClient CreateClient(BlobClientOptions options) => + new BlobServiceClient(new Uri("http://127.0.0.1:1"), options); } internal class KnownTypeBinder : ICustomTypeBinder diff --git a/test/DurableTask.AzureStorage.Tests/RewindOutputTrackingStoreTests.cs b/test/DurableTask.AzureStorage.Tests/RewindOutputTrackingStoreTests.cs new file mode 100644 index 00000000..cad19076 --- /dev/null +++ b/test/DurableTask.AzureStorage.Tests/RewindOutputTrackingStoreTests.cs @@ -0,0 +1,179 @@ +// ---------------------------------------------------------------------------------- +// 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.Linq; + using System.Threading.Tasks; + using Azure; + 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; + + [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 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() + { + if (this.trackingStore != null) + { + await this.trackingStore.DeleteAsync(); + } + } + + [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); + + 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); + + await this.trackingStore.UpdateStatusForRewindAsync(instanceId); + await this.trackingStore.UpdateStatusForRewindAsync(instanceId); + + 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"); + await this.trackingStore.UpdateStatusForRewindAsync(instanceId); + + 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")); + } + + async Task SeedInstanceRowAsync(string instanceId, OrchestrationStatus status, string output) + { + var entity = new TableEntity(KeySanitation.EscapePartitionKey(instanceId), string.Empty) + { + ["Name"] = "TestOrchestration", + ["RuntimeStatus"] = status.ToString(), + ["CreatedTime"] = DateTime.UtcNow, + ["LastUpdatedTime"] = DateTime.UtcNow, + ["TaskHubName"] = this.taskHubName, + ["ExecutionId"] = "execution-1", + [PreservedProperty] = "preserve me", + }; + + if (output != null) + { + entity["Output"] = output; + } + + await this.trackingStore.InstancesTable.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(); + } + + static ExecutionStartedEvent CreateExecutionStartedEvent(string instanceId, string executionId) + { + return new ExecutionStartedEvent(-1, "input") + { + Name = "TestOrchestration", + Version = string.Empty, + OrchestrationInstance = new OrchestrationInstance + { + InstanceId = instanceId, + ExecutionId = executionId, + }, + }; + } + } +} 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();