From dda77d40f29b4aca1877df1163a9296d490083e0 Mon Sep 17 00:00:00 2001 From: wangbill Date: Thu, 3 Sep 2026 14:40:05 -0700 Subject: [PATCH 1/3] Clear persisted output when rewinding Replace the complete Azure Table instance row with its current ETag so rewind removes the terminal Output property without clobbering concurrent fields. Add unit and Azurite coverage for persisted state, retries, terminal rewrites, instance reuse, and shared large-output blobs. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- .../Tracking/AzureTableTrackingStore.cs | 20 +- .../AzureStorageScenarioTests.cs | 48 +++++ .../AzureTableTrackingStoreTest.cs | 139 ++++++++++++++ .../RewindOutputTrackingStoreTests.cs | 179 ++++++++++++++++++ 4 files changed, 380 insertions(+), 6 deletions(-) create mode 100644 test/DurableTask.AzureStorage.Tests/RewindOutputTrackingStoreTests.cs diff --git a/src/DurableTask.AzureStorage/Tracking/AzureTableTrackingStore.cs b/src/DurableTask.AzureStorage/Tracking/AzureTableTrackingStore.cs index 167a3311..7f63f60c 100644 --- a/src/DurableTask.AzureStorage/Tracking/AzureTableTrackingStore.cs +++ b/src/DurableTask.AzureStorage/Tracking/AzureTableTrackingStore.cs @@ -856,15 +856,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. diff --git a/test/DurableTask.AzureStorage.Tests/AzureStorageScenarioTests.cs b/test/DurableTask.AzureStorage.Tests/AzureStorageScenarioTests.cs index a2372034..89a23473 100644 --- a/test/DurableTask.AzureStorage.Tests/AzureStorageScenarioTests.cs +++ b/test/DurableTask.AzureStorage.Tests/AzureStorageScenarioTests.cs @@ -1510,6 +1510,54 @@ public async Task RewindActivityFail() } } + [TestMethod] + public async Task RewindLargeFailure_RemovesInstanceOutputButRetainsSharedHistoryBlob() + { + using (TestOrchestrationHost host = TestHelpers.GetTestOrchestrationHost(enableExtendedSessions: false)) + { + await host.StartAsync(); + + string failureMessage = this.GenerateMediumRandomStringPayload().ToString(); + var client = await host.StartOrchestrationAsync( + typeof(Orchestrations.ThrowException), + input: failureMessage); + OrchestrationState failed = await client.WaitForCompletionAsync(TimeSpan.FromSeconds(30)); + 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)); + + await host.StopAsync(); + await client.RewindAsync("Remove the stale persisted output."); + + 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(ExecutionCompletedEvent.OrchestrationStatus), nameof(OrchestrationStatus.Failed))}"; + TableEntity rewoundCompletion = (await trackingStore.HistoryTable + .ExecuteQueryAsync(historyFilter) + .ToListAsync()) + .Single(entity => entity.GetString("ResultBlobName") != null); + string resultBlobName = rewoundCompletion.GetString("ResultBlobName"); + + Assert.AreEqual(nameof(EventType.GenericEvent), rewoundCompletion.GetString(nameof(HistoryEvent.EventType))); + Assert.IsTrue(new Uri(outputBlobUrl).AbsolutePath.EndsWith(resultBlobName, StringComparison.Ordinal)); + var blobServiceClient = new BlobServiceClient(TestHelpers.GetTestStorageAccountConnectionString()); + BlobContainerClient container = blobServiceClient.GetBlobContainerClient($"{host.TaskHub.ToLowerInvariant()}-largemessages"); + Assert.IsTrue((await container.GetBlobClient(resultBlobName).ExistsAsync()).Value); + } + } + [TestMethod] public async Task RewindMultipleActivityFail() { 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/RewindOutputTrackingStoreTests.cs b/test/DurableTask.AzureStorage.Tests/RewindOutputTrackingStoreTests.cs new file mode 100644 index 00000000..22e08ad3 --- /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"), + new ETag(existing.ETag.ToString()), + 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, + }, + }; + } + } +} From d33ee4ab966c1b8f061fcc9bd776fb8a65304b6d Mon Sep 17 00:00:00 2001 From: wangbill Date: Fri, 4 Sep 2026 10:28:29 -0700 Subject: [PATCH 2/3] Delete stale output blobs when rewinding Remove terminal history payload references before clearing the instance output and deleting the uniquely owned large-message blobs. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> Copilot-Session: cf71194c-b467-4684-b5f6-f31ff9a2e60e --- .../Tracking/AzureTableTrackingStore.cs | 25 ++++++ .../AzureStorageScenarioTests.cs | 82 ++++++++++++++++--- .../TestOrchestrationHost.cs | 6 ++ 3 files changed, 102 insertions(+), 11 deletions(-) diff --git a/src/DurableTask.AzureStorage/Tracking/AzureTableTrackingStore.cs b/src/DurableTask.AzureStorage/Tracking/AzureTableTrackingStore.cs index 7f63f60c..4778de59 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,6 +379,9 @@ 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 Task.WhenAll(blobsToDelete.Select(blobName => this.messageManager.DeleteBlobAsync(blobName, cancellationToken))); + if (!hasFailedSubOrchestrations) { yield return instanceId; @@ -1386,6 +1397,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 89a23473..d598156e 100644 --- a/test/DurableTask.AzureStorage.Tests/AzureStorageScenarioTests.cs +++ b/test/DurableTask.AzureStorage.Tests/AzureStorageScenarioTests.cs @@ -1511,15 +1511,17 @@ public async Task RewindActivityFail() } [TestMethod] - public async Task RewindLargeFailure_RemovesInstanceOutputButRetainsSharedHistoryBlob() + public async Task RewindLargeFailure_RemovesPersistedOutputAndBlobs() { using (TestOrchestrationHost host = TestHelpers.GetTestOrchestrationHost(enableExtendedSessions: false)) { + Orchestrations.RewindLargeFailure.ShouldFail = true; + host.ErrorPropagationMode = ErrorPropagationMode.UseFailureDetails; await host.StartAsync(); string failureMessage = this.GenerateMediumRandomStringPayload().ToString(); var client = await host.StartOrchestrationAsync( - typeof(Orchestrations.ThrowException), + typeof(Orchestrations.RewindLargeFailure), input: failureMessage); OrchestrationState failed = await client.WaitForCompletionAsync(TimeSpan.FromSeconds(30)); Assert.AreEqual(OrchestrationStatus.Failed, failed?.OrchestrationStatus); @@ -1533,8 +1535,35 @@ public async Task RewindLargeFailure_RemovesInstanceOutputButRetainsSharedHistor string outputBlobUrl = failedInstance.GetString("Output"); Assert.IsTrue(Uri.IsWellFormedUriString(outputBlobUrl, UriKind.Absolute)); - await host.StopAsync(); - await client.RewindAsync("Remove the stale persisted 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(); + 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); + } + + Orchestrations.RewindLargeFailure.ShouldFail = false; + CollectionAssert.AreEqual( + new[] { client.InstanceId }, + await trackingStore.RewindHistoryAsync(client.InstanceId).ToListAsync()); TableEntity rewoundInstance = await trackingStore.InstancesTable .ExecuteQueryAsync(instanceFilter, 1) @@ -1543,18 +1572,34 @@ public async Task RewindLargeFailure_RemovesInstanceOutputButRetainsSharedHistor Assert.IsFalse(rewoundInstance.ContainsKey("Output")); string historyFilter = $"{AzureTableQueryFilter.PartitionKeyEquals(client.InstanceId)} and " + - $"{AzureTableQueryFilter.ColumnEquals(nameof(ExecutionCompletedEvent.OrchestrationStatus), nameof(OrchestrationStatus.Failed))}"; + $"{AzureTableQueryFilter.ColumnEquals(nameof(ITableEntity.RowKey), failedCompletion.RowKey)}"; TableEntity rewoundCompletion = (await trackingStore.HistoryTable .ExecuteQueryAsync(historyFilter) .ToListAsync()) - .Single(entity => entity.GetString("ResultBlobName") != null); - string resultBlobName = rewoundCompletion.GetString("ResultBlobName"); + .Single(); Assert.AreEqual(nameof(EventType.GenericEvent), rewoundCompletion.GetString(nameof(HistoryEvent.EventType))); - Assert.IsTrue(new Uri(outputBlobUrl).AbsolutePath.EndsWith(resultBlobName, StringComparison.Ordinal)); - var blobServiceClient = new BlobServiceClient(TestHelpers.GetTestStorageAccountConnectionString()); - BlobContainerClient container = blobServiceClient.GetBlobContainerClient($"{host.TaskHub.ToLowerInvariant()}-largemessages"); - Assert.IsTrue((await container.GetBlobClient(resultBlobName).ExistsAsync()).Value); + Assert.IsFalse(rewoundCompletion.ContainsKey("Result")); + Assert.IsFalse(rewoundCompletion.ContainsKey("ResultBlobName")); + Assert.IsFalse(rewoundCompletion.ContainsKey("FailureDetails")); + Assert.IsFalse(rewoundCompletion.ContainsKey("FailureDetailsBlobName")); + foreach (string blobName in outputBlobNames) + { + Assert.IsFalse((await container.GetBlobClient(blobName).ExistsAsync()).Value); + } + + 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(); } } @@ -4905,6 +4950,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/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(); From d090a7502a39a715594dfcb574cc68ebd6def392 Mon Sep 17 00:00:00 2001 From: wangbill Date: Fri, 4 Sep 2026 12:11:27 -0700 Subject: [PATCH 3/3] Make rewind blob cleanup best effort Normalize ordinary blob retry and timeout failures, log them after table dereference, and continue producing rewind targets while preserving caller cancellation. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> Copilot-Session: b4b381ba-7e3f-4548-ab7d-73eac7a4466a --- .../Logging/LogHelper.cs | 5 +- .../MessageManager.cs | 15 +- .../Tracking/AzureTableTrackingStore.cs | 27 +++- .../AzureStorageScenarioTests.cs | 134 ++++++++++++++---- .../MessageManagerTests.cs | 54 +++++++ .../RewindOutputTrackingStoreTests.cs | 2 +- 6 files changed, 205 insertions(+), 32 deletions(-) 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 4778de59..e4c5fecb 100644 --- a/src/DurableTask.AzureStorage/Tracking/AzureTableTrackingStore.cs +++ b/src/DurableTask.AzureStorage/Tracking/AzureTableTrackingStore.cs @@ -380,7 +380,7 @@ public override async IAsyncEnumerable RewindHistoryAsync(string instanc await this.UpdateStatusForRewindAsync(instanceId, cancellationToken); // Delete only after both the history pointers and the Instances-table Output reference are gone. - await Task.WhenAll(blobsToDelete.Select(blobName => this.messageManager.DeleteBlobAsync(blobName, cancellationToken))); + await this.DeleteRewindBlobsAsync(instanceId, blobsToDelete, cancellationToken); if (!hasFailedSubOrchestrations) { @@ -388,6 +388,31 @@ public override async IAsyncEnumerable RewindHistoryAsync(string instanc } } + 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) { diff --git a/test/DurableTask.AzureStorage.Tests/AzureStorageScenarioTests.cs b/test/DurableTask.AzureStorage.Tests/AzureStorageScenarioTests.cs index d598156e..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,10 +1513,20 @@ public async Task RewindActivityFail() } } - [TestMethod] - public async Task RewindLargeFailure_RemovesPersistedOutputAndBlobs() + [DataTestMethod] + [DataRow(false)] + [DataRow(true)] + public async Task RewindLargeFailure_HandlesBlobDeletionFailure(bool failBlobDeletion) { - using (TestOrchestrationHost host = TestHelpers.GetTestOrchestrationHost(enableExtendedSessions: false)) + 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; @@ -1524,7 +1537,8 @@ public async Task RewindLargeFailure_RemovesPersistedOutputAndBlobs() typeof(Orchestrations.RewindLargeFailure), input: failureMessage); OrchestrationState failed = await client.WaitForCompletionAsync(TimeSpan.FromSeconds(30)); - Assert.AreEqual(OrchestrationStatus.Failed, failed?.OrchestrationStatus); + Assert.IsNotNull(failed); + Assert.AreEqual(OrchestrationStatus.Failed, failed.OrchestrationStatus); var trackingStore = (AzureTableTrackingStore)host.service.TrackingStore; string instanceFilter = $"{AzureTableQueryFilter.PartitionKeyEquals(client.InstanceId)} and " + @@ -1560,35 +1574,103 @@ public async Task RewindLargeFailure_RemovesPersistedOutputAndBlobs() 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; - CollectionAssert.AreEqual( - new[] { client.InstanceId }, - await trackingStore.RewindHistoryAsync(client.InstanceId).ToListAsync()); + 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(); - Assert.AreEqual(OrchestrationStatus.Pending.ToString(), rewoundInstance.GetString("RuntimeStatus")); - Assert.IsFalse(rewoundInstance.ContainsKey("Output")); + 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(); + 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(); + } + } - 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")); - foreach (string blobName in outputBlobNames) + if (!failBlobDeletion) { - Assert.IsFalse((await container.GetBlobClient(blobName).ExistsAsync()).Value); + await client.RewindAsync("Retry the persisted-output cleanup."); } - 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); 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 index 22e08ad3..cad19076 100644 --- a/test/DurableTask.AzureStorage.Tests/RewindOutputTrackingStoreTests.cs +++ b/test/DurableTask.AzureStorage.Tests/RewindOutputTrackingStoreTests.cs @@ -123,7 +123,7 @@ public async Task SetNewExecution_ReplacesPersistedOutput() bool created = await this.trackingStore.SetNewExecutionAsync( CreateExecutionStartedEvent(instanceId, "execution-2"), - new ETag(existing.ETag.ToString()), + existing.ETag, inputPayloadOverride: null); Assert.IsTrue(created);