diff --git a/src/DurableTask.AzureStorage/OrchestrationInstanceStatus.cs b/src/DurableTask.AzureStorage/OrchestrationInstanceStatus.cs index 29dea2c4..9cef3822 100644 --- a/src/DurableTask.AzureStorage/OrchestrationInstanceStatus.cs +++ b/src/DurableTask.AzureStorage/OrchestrationInstanceStatus.cs @@ -23,6 +23,7 @@ namespace DurableTask.AzureStorage class OrchestrationInstanceStatus : ITableEntity { public string ExecutionId { get; set; } + public string ParentInstanceId { get; set; } public string Name { get; set; } public string Version { get; set; } public string Input { get; set; } diff --git a/src/DurableTask.AzureStorage/Tracking/AzureTableTrackingStore.cs b/src/DurableTask.AzureStorage/Tracking/AzureTableTrackingStore.cs index 6cecb9d6..7b6ff313 100644 --- a/src/DurableTask.AzureStorage/Tracking/AzureTableTrackingStore.cs +++ b/src/DurableTask.AzureStorage/Tracking/AzureTableTrackingStore.cs @@ -38,6 +38,7 @@ namespace DurableTask.AzureStorage.Tracking class AzureTableTrackingStore : TrackingStoreBase { const string NameProperty = "Name"; + const string ParentInstanceIdProperty = "ParentInstanceId"; const string InputProperty = "Input"; const string ResultProperty = "Result"; const string OutputProperty = "Output"; @@ -461,6 +462,16 @@ async Task ConvertFromAsync(OrchestrationInstanceStatus orch InstanceId = instanceId, ExecutionId = orchestrationInstanceStatus.ExecutionId, }; + if (!string.IsNullOrEmpty(orchestrationInstanceStatus.ParentInstanceId)) + { + orchestrationState.ParentInstance = new ParentInstance + { + OrchestrationInstance = new OrchestrationInstance + { + InstanceId = orchestrationInstanceStatus.ParentInstanceId, + }, + }; + } orchestrationState.Name = orchestrationInstanceStatus.Name; orchestrationState.Version = orchestrationInstanceStatus.Version; @@ -799,6 +810,7 @@ public override async Task SetNewExecutionAsync( ["Generation"] = executionStartedEvent.Generation, ["Tags"] = TagsSerializer.Serialize(executionStartedEvent.Tags), }; + SetParentInstanceId(entity, executionStartedEvent.ParentInstance); // It is possible that the queue message was small enough to be written directly to a queue message, // not a blob, but is too large to be written to a table property. @@ -990,6 +1002,7 @@ public override async Task UpdateStateAsync( instanceEntity["RuntimeStatus"] = OrchestrationStatus.Running.ToString(); instanceEntity["Tags"] = TagsSerializer.Serialize(executionStartedEvent.Tags); instanceEntity["Generation"] = executionStartedEvent.Generation; + SetParentInstanceId(instanceEntity, executionStartedEvent.ParentInstance); if (executionStartedEvent.ScheduledStartTime.HasValue) { instanceEntity["ScheduledStartTime"] = executionStartedEvent.ScheduledStartTime; @@ -1151,6 +1164,7 @@ public override async Task UpdateInstanceStatusForCompletedOrchestrationAsync( ["Tags"] = TagsSerializer.Serialize(executionStartedEvent.Tags), ["TaskHubName"] = this.settings.TaskHubName, }; + SetParentInstanceId(instanceEntity, executionStartedEvent.ParentInstance); if (runtimeState.ExecutionStartedEvent.ScheduledStartTime.HasValue) { instanceEntity["ScheduledStartTime"] = executionStartedEvent.ScheduledStartTime; @@ -1248,6 +1262,16 @@ static int GetEstimatedByteCount(TableEntity entity) return estimatedByteCount; } + // The value is always assigned, including when there is no parent. Several of the Instances + // table writes use merge semantics, so omitting the property would let a top-level or newly + // recreated orchestration inherit a stale parent ID from a previous row with the same + // instance ID. An empty string is used rather than null because merge semantics for null + // properties are ambiguous; reads treat empty and missing identically. + static void SetParentInstanceId(TableEntity entity, ParentInstance parentInstance) + { + entity[ParentInstanceIdProperty] = parentInstance?.OrchestrationInstance?.InstanceId ?? string.Empty; + } + Type GetTypeForTableEntity(TableEntity tableEntity) { string propertyName = nameof(HistoryEvent.EventType); diff --git a/src/DurableTask.AzureStorage/Tracking/InstanceStoreBackedTrackingStore.cs b/src/DurableTask.AzureStorage/Tracking/InstanceStoreBackedTrackingStore.cs index f719c92d..a7c43988 100644 --- a/src/DurableTask.AzureStorage/Tracking/InstanceStoreBackedTrackingStore.cs +++ b/src/DurableTask.AzureStorage/Tracking/InstanceStoreBackedTrackingStore.cs @@ -108,6 +108,7 @@ public override async Task SetNewExecutionAsync( Name = executionStartedEvent.Name, Version = executionStartedEvent.Version, OrchestrationInstance = executionStartedEvent.OrchestrationInstance, + ParentInstance = executionStartedEvent.ParentInstance, OrchestrationStatus = OrchestrationStatus.Pending, Input = inputStatusOverride ?? executionStartedEvent.Input, Tags = executionStartedEvent.Tags, diff --git a/test/DurableTask.AzureStorage.Tests/AzureStorageScenarioTests.cs b/test/DurableTask.AzureStorage.Tests/AzureStorageScenarioTests.cs index 07b064ac..a2372034 100644 --- a/test/DurableTask.AzureStorage.Tests/AzureStorageScenarioTests.cs +++ b/test/DurableTask.AzureStorage.Tests/AzureStorageScenarioTests.cs @@ -154,6 +154,78 @@ public async Task ParentOfSequentialOrchestration() } } + /// + /// Verifies that the normal checkpoint write path records a child's parent, and that the value is + /// returned by both a direct instance lookup and a status query. This does not cover the + /// terminal-history repair path; see + /// + /// for that. + /// + [TestMethod] + public async Task ParentMetadataIsReturnedByDirectGetAndQuery() + { + using (TestOrchestrationHost host = TestHelpers.GetTestOrchestrationHost( + enableExtendedSessions: false, + modifySettingsAction: settings => settings.TaskHubName = "pmf" + Guid.NewGuid().ToString("N").Substring(0, 12))) + { + string parentInstanceId = $"parent-{Guid.NewGuid():N}"; + string childInstanceId = parentInstanceId + ":child"; + await host.StartAsync(); + + var client = await host.StartOrchestrationAsync( + typeof(Orchestrations.ParentOfInlineChild), + "input", + parentInstanceId); + OrchestrationState completed = await client.WaitForCompletionAsync(StandardTimeout); + + Assert.AreEqual(OrchestrationStatus.Completed, completed?.OrchestrationStatus); + + OrchestrationState parent = await host.service.GetOrchestrationStateAsync(parentInstanceId, executionId: null); + OrchestrationState child = await host.service.GetOrchestrationStateAsync(childInstanceId, executionId: null); + Assert.IsNull(parent.ParentInstance); + Assert.AreEqual(parentInstanceId, child.ParentInstance?.OrchestrationInstance.InstanceId); + + DurableStatusQueryResult queryResult = await host.service.GetOrchestrationStateAsync( + new OrchestrationInstanceStatusQueryCondition { InstanceIdPrefix = parentInstanceId }, + top: 10, + continuationToken: null); + OrchestrationState queriedChild = queryResult.OrchestrationState.Single(state => + state.OrchestrationInstance.InstanceId == childInstanceId); + Assert.AreEqual(parentInstanceId, queriedChild.ParentInstance?.OrchestrationInstance.InstanceId); + OrchestrationState queriedParent = queryResult.OrchestrationState.Single(state => + state.OrchestrationInstance.InstanceId == parentInstanceId); + Assert.IsNull(queriedParent.ParentInstance); + + await host.StopAsync(); + } + } + + [TestMethod] + public async Task ParentMetadataSurvivesChildContinueAsNew() + { + using (TestOrchestrationHost host = TestHelpers.GetTestOrchestrationHost( + enableExtendedSessions: false, + modifySettingsAction: settings => settings.TaskHubName = "pmc" + Guid.NewGuid().ToString("N").Substring(0, 12))) + { + string parentInstanceId = $"parent-{Guid.NewGuid():N}"; + string childInstanceId = parentInstanceId + ":child"; + await host.StartAsync(); + + var client = await host.StartOrchestrationAsync( + typeof(Orchestrations.ParentOfContinueAsNewChild), + 0, + parentInstanceId); + OrchestrationState completed = await client.WaitForCompletionAsync(StandardTimeout); + + Assert.AreEqual(OrchestrationStatus.Completed, completed?.OrchestrationStatus); + OrchestrationState child = await host.service.GetOrchestrationStateAsync(childInstanceId, executionId: null); + Assert.AreEqual(1, JToken.Parse(child.Input)); + Assert.AreEqual(parentInstanceId, child.ParentInstance?.OrchestrationInstance.InstanceId); + + await host.StopAsync(); + } + } + /// /// End-to-end test which runs a slow orchestrator that causes work item renewal /// @@ -4875,6 +4947,51 @@ public override Task RunTask(OrchestrationContext context, int input) } } + [KnownType(typeof(InlineChild))] + internal class ParentOfInlineChild : TaskOrchestration + { + public override Task RunTask(OrchestrationContext context, string input) + { + return context.CreateSubOrchestrationInstance( + typeof(InlineChild), + context.OrchestrationInstance.InstanceId + ":child", + input); + } + } + + internal class InlineChild : TaskOrchestration + { + public override Task RunTask(OrchestrationContext context, string input) + { + return Task.FromResult(input); + } + } + + [KnownType(typeof(ContinueAsNewChild))] + internal class ParentOfContinueAsNewChild : TaskOrchestration + { + public override Task RunTask(OrchestrationContext context, int input) + { + return context.CreateSubOrchestrationInstance( + typeof(ContinueAsNewChild), + context.OrchestrationInstance.InstanceId + ":child", + input); + } + } + + internal class ContinueAsNewChild : TaskOrchestration + { + public override Task RunTask(OrchestrationContext context, int input) + { + if (input == 0) + { + context.ContinueAsNew(1); + } + + return Task.FromResult(input); + } + } + [KnownType(typeof(Activities.Hello))] internal class DoubleFanOut : TaskOrchestration { diff --git a/test/DurableTask.AzureStorage.Tests/AzureTableTrackingStoreTest.cs b/test/DurableTask.AzureStorage.Tests/AzureTableTrackingStoreTest.cs index f5f43180..5c1b402d 100644 --- a/test/DurableTask.AzureStorage.Tests/AzureTableTrackingStoreTest.cs +++ b/test/DurableTask.AzureStorage.Tests/AzureTableTrackingStoreTest.cs @@ -25,6 +25,8 @@ namespace DurableTask.AzureStorage.Tests using DurableTask.AzureStorage.Storage; using DurableTask.AzureStorage.Tracking; using DurableTask.Core; + using DurableTask.Core.History; + using DurableTask.Core.Tracking; using Microsoft.VisualStudio.TestTools.UnitTesting; using Moq; @@ -66,16 +68,21 @@ public async Task QueryStatus_WithContinuationToken_NoInputToken() { new OrchestrationInstanceStatus { + PartitionKey = "child", + ParentInstanceId = "parent", Name = "foo", RuntimeStatus = "Running" }, new OrchestrationInstanceStatus { + PartitionKey = "top-level", + ParentInstanceId = "", Name = "bar", RuntimeStatus = "Completed" }, new OrchestrationInstanceStatus { + PartitionKey = "legacy", Name = "baz", RuntimeStatus = "Failed" } @@ -111,6 +118,51 @@ public async Task QueryStatus_WithContinuationToken_NoInputToken() Assert.AreEqual(expected[i].Name, actual[i].Name); Assert.AreEqual(Enum.Parse(typeof(OrchestrationStatus), expected[i].RuntimeStatus), actual[i].OrchestrationStatus); } + + // Child rows resolve to their parent; rows written for a top-level orchestration store an + // empty value to clear any stale parent, and legacy rows omit the property entirely. The + // latter two must both surface as a null ParentInstance. + Assert.AreEqual("parent", actual[0].ParentInstance.OrchestrationInstance.InstanceId); + Assert.IsNull(actual[1].ParentInstance); + Assert.IsNull(actual[2].ParentInstance); + } + + [TestMethod] + public async Task InstanceStoreBackedTrackingStore_PersistsParentOnCreation() + { + const string ParentInstanceId = "parent"; + OrchestrationStateInstanceEntity writtenState = null; + var instanceStore = new Mock(MockBehavior.Strict); + instanceStore + .Setup(store => store.WriteEntitiesAsync(It.IsAny>())) + .Callback>(entities => writtenState = entities.Single() as OrchestrationStateInstanceEntity) + .ReturnsAsync(new object()); + + var trackingStore = new InstanceStoreBackedTrackingStore(instanceStore.Object); + var startedEvent = new ExecutionStartedEvent(0, null) + { + Name = "child", + OrchestrationInstance = new OrchestrationInstance + { + InstanceId = "child", + ExecutionId = "execution", + }, + ParentInstance = new ParentInstance + { + OrchestrationInstance = new OrchestrationInstance + { + InstanceId = ParentInstanceId, + ExecutionId = "parent-execution", + }, + }, + }; + + bool created = await trackingStore.SetNewExecutionAsync(startedEvent, null, null); + + Assert.IsTrue(created); + Assert.IsNotNull(writtenState); + Assert.AreSame(startedEvent.ParentInstance, writtenState.State.ParentInstance); + Assert.AreEqual(ParentInstanceId, writtenState.State.ParentInstance.OrchestrationInstance.InstanceId); } } } diff --git a/test/DurableTask.AzureStorage.Tests/OrchestrationInstanceStatusQueryConditionTest.cs b/test/DurableTask.AzureStorage.Tests/OrchestrationInstanceStatusQueryConditionTest.cs index 9c6b20d4..8eb5081f 100644 --- a/test/DurableTask.AzureStorage.Tests/OrchestrationInstanceStatusQueryConditionTest.cs +++ b/test/DurableTask.AzureStorage.Tests/OrchestrationInstanceStatusQueryConditionTest.cs @@ -15,6 +15,7 @@ namespace DurableTask.AzureStorage.Tests { using System; using System.Collections.Generic; + using System.Linq; using DurableTask.AzureStorage.Tracking; using DurableTask.Core; using Microsoft.VisualStudio.TestTools.UnitTesting; @@ -108,6 +109,29 @@ public void OrchestrationInstanceQuery_NoParameter() Assert.IsTrue(string.IsNullOrWhiteSpace(condition.ToOData().Filter)); } + /// + /// When inputs and outputs are excluded, the query switches from "select everything" to an explicit + /// column projection. ParentInstanceId must stay in that projection, otherwise status queries that + /// omit inputs/outputs would silently return a null ParentInstance. + /// + [TestMethod] + public void OrchestrationInstanceQuery_ProjectionRetainsParentInstanceId() + { + var condition = new OrchestrationInstanceStatusQueryCondition + { + RuntimeStatus = new OrchestrationStatus[] { OrchestrationStatus.Running }, + FetchInput = false, + FetchOutput = false, + }; + + IEnumerable select = condition.ToOData().Select; + + Assert.IsNotNull(select, "Excluding input and output should produce an explicit projection."); + CollectionAssert.Contains(select.ToList(), nameof(OrchestrationInstanceStatus.ParentInstanceId)); + CollectionAssert.DoesNotContain(select.ToList(), nameof(OrchestrationInstanceStatus.Input)); + CollectionAssert.DoesNotContain(select.ToList(), nameof(OrchestrationInstanceStatus.Output)); + } + [TestMethod] public void OrchestrationInstanceQuery_MultipleRuntimeStatus() { diff --git a/test/DurableTask.AzureStorage.Tests/ParentInstanceIdTrackingStoreTests.cs b/test/DurableTask.AzureStorage.Tests/ParentInstanceIdTrackingStoreTests.cs new file mode 100644 index 00000000..e21a2e40 --- /dev/null +++ b/test/DurableTask.AzureStorage.Tests/ParentInstanceIdTrackingStoreTests.cs @@ -0,0 +1,368 @@ +// ---------------------------------------------------------------------------------- +// 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.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; + + /// + /// Tests that exercise the ParentInstanceId Instances-table property against a real storage + /// account, so that the actual Azure Table update semantics (InsertOrMerge / Merge) are covered. + /// Assertions are made on the raw stored property in addition to the public read conversion, + /// because a merge-based write that omits the property leaves a previous value intact and would + /// otherwise be invisible to a test that only inspects converted state. + /// + [TestClass] + public class ParentInstanceIdTrackingStoreTests + { + const string ParentInstanceIdProperty = "ParentInstanceId"; + + string taskHubName; + AzureStorageOrchestrationServiceSettings settings; + AzureStorageClient azureStorageClient; + AzureTableTrackingStore trackingStore; + + [TestInitialize] + public async Task Initialize() + { + // A unique task hub per test keeps the Instances/History tables isolated, so a leftover row + // from another test cannot mask a missing write. + this.taskHubName = "pid" + Guid.NewGuid().ToString("N").Substring(0, 12); + this.settings = TestHelpers.GetTestAzureStorageOrchestrationServiceSettings(enableExtendedSessions: false); + this.settings.TaskHubName = this.taskHubName; + + this.azureStorageClient = new AzureStorageClient(this.settings); + var messageManager = new MessageManager(this.settings, this.azureStorageClient, $"{this.taskHubName}-largemessages".ToLowerInvariant()); + this.trackingStore = new AzureTableTrackingStore(this.azureStorageClient, messageManager); + await this.trackingStore.CreateAsync(); + } + + [TestCleanup] + public async Task Cleanup() + { + // Delete the per-test tables so repeated runs do not leak storage resources. + if (this.trackingStore != null) + { + await this.trackingStore.DeleteAsync(); + } + } + + /// + /// Verifies that a no-parent write clears a parent ID left behind by a previous row with the same + /// instance ID. This covers the terminal-history repair write, which is unconditionally + /// InsertOrMerge, so a helper that skips the property when there is no parent would leave the + /// stale value in place. + /// + [TestMethod] + public async Task CompletedOrchestrationRepair_ClearsStaleParentInstanceId_WithInsertOrMerge() + { + string instanceId = $"stale-{Guid.NewGuid():N}"; + + await this.SeedInstanceRowAsync(instanceId, "stale-parent"); + Assert.AreEqual("stale-parent", await this.GetRawParentInstanceIdAsync(instanceId), "Seeded row should carry the stale parent."); + + // Drive a genuine no-parent write through the same production path that uses InsertOrMerge. + await this.trackingStore.UpdateInstanceStatusForCompletedOrchestrationAsync( + instanceId, + executionId: "execution-1", + runtimeState: CreateCompletedRuntimeState(instanceId, "execution-1", parentInstanceId: null), + instanceEntityExists: true); + + Assert.AreEqual( + string.Empty, + await this.GetRawParentInstanceIdAsync(instanceId), + "A no-parent write must clear the stored property, otherwise merge semantics retain the stale parent."); + + InstanceStatus status = await this.trackingStore.FetchInstanceStatusAsync(instanceId); + Assert.IsNull(status.State.ParentInstance, "A cleared parent must read back as a null ParentInstance."); + } + + /// + /// The clearing behavior must also hold for the checkpoint write in + /// , which routes through + /// UpdateInstanceTableAsync. That method picks InsertOrMerge when UseInstanceTableEtag is false and + /// Merge (with the supplied ETag) when it is true. Both are merge operations, so an omitted + /// property is retained and a previous parent would survive into an unrelated orchestration that + /// reused the instance ID. Seeding a row and passing its ETag is what makes the true case reach + /// MergeEntityAsync rather than the insert branch. + /// + [DataTestMethod] + [DataRow(false)] + [DataRow(true)] + public async Task CheckpointWrite_ClearsStaleParentInstanceId_ForBothEtagModes(bool useInstanceTableEtag) + { + this.settings.UseInstanceTableEtag = useInstanceTableEtag; + string instanceId = $"etag-{useInstanceTableEtag}-{Guid.NewGuid():N}"; + + await this.SeedInstanceRowAsync(instanceId, "stale-parent"); + Assert.AreEqual("stale-parent", await this.GetRawParentInstanceIdAsync(instanceId), "Seeded row should carry the stale parent."); + + // Passing the seeded row's ETag forces the UseInstanceTableEtag=true case down the + // MergeEntityAsync branch; a null ETag there would insert instead of merging. + OrchestrationInstanceStatus seeded = await this.GetRawEntityAsync(instanceId); + var eTags = new OrchestrationETags + { + InstanceETag = useInstanceTableEtag ? new ETag(seeded.ETag.ToString()) : (ETag?)null, + }; + + var runtimeState = new OrchestrationRuntimeState(); + runtimeState.AddEvent(CreateExecutionStartedEvent(instanceId, "execution-1", parentInstanceId: null)); + + await this.trackingStore.UpdateStateAsync( + runtimeState, + runtimeState, + instanceId, + "execution-1", + eTags, + await this.GetTrackingStoreContextAsync(instanceId)); + + Assert.AreEqual( + string.Empty, + await this.GetRawParentInstanceIdAsync(instanceId), + "The checkpoint write must clear the stored property, otherwise merge semantics retain the stale parent."); + + InstanceStatus status = await this.trackingStore.FetchInstanceStatusAsync(instanceId); + Assert.IsNull(status.State.ParentInstance, "A cleared parent must read back as a null ParentInstance."); + } + + /// + /// The checkpoint write must also persist a non-null parent, for both update modes. + /// + [DataTestMethod] + [DataRow(false)] + [DataRow(true)] + public async Task CheckpointWrite_PersistsParentInstanceId_ForBothEtagModes(bool useInstanceTableEtag) + { + this.settings.UseInstanceTableEtag = useInstanceTableEtag; + string parentInstanceId = $"parent-{Guid.NewGuid():N}"; + string childInstanceId = $"{parentInstanceId}:child"; + + await this.SeedInstanceRowAsync(childInstanceId, parentInstanceId: null); + + OrchestrationInstanceStatus seeded = await this.GetRawEntityAsync(childInstanceId); + var eTags = new OrchestrationETags + { + InstanceETag = useInstanceTableEtag ? new ETag(seeded.ETag.ToString()) : (ETag?)null, + }; + + var runtimeState = new OrchestrationRuntimeState(); + runtimeState.AddEvent(CreateExecutionStartedEvent(childInstanceId, "execution-1", parentInstanceId)); + + await this.trackingStore.UpdateStateAsync( + runtimeState, + runtimeState, + childInstanceId, + "execution-1", + eTags, + await this.GetTrackingStoreContextAsync(childInstanceId)); + + Assert.AreEqual(parentInstanceId, await this.GetRawParentInstanceIdAsync(childInstanceId)); + + InstanceStatus status = await this.trackingStore.FetchInstanceStatusAsync(childInstanceId); + Assert.AreEqual(parentInstanceId, status.State.ParentInstance?.OrchestrationInstance.InstanceId); + } + + /// + /// Verifies that the terminal-history repair path persists the parent ID when it recreates an + /// Instances row that no longer exists. This is the projection-repair case that runs when a worker + /// fails after writing history but before updating the Instances table, which is common for + /// sub-orchestrations that complete within a single execution. + /// + [TestMethod] + public async Task CompletedOrchestrationRepair_PersistsParentInstanceId() + { + string parentInstanceId = $"parent-{Guid.NewGuid():N}"; + string childInstanceId = $"{parentInstanceId}:child"; + + // No row is seeded: this mirrors a sub-orchestration whose Instances projection was never written. + await this.trackingStore.UpdateInstanceStatusForCompletedOrchestrationAsync( + childInstanceId, + executionId: "execution-1", + runtimeState: CreateCompletedRuntimeState(childInstanceId, "execution-1", parentInstanceId), + instanceEntityExists: false); + + Assert.AreEqual(parentInstanceId, await this.GetRawParentInstanceIdAsync(childInstanceId)); + + InstanceStatus status = await this.trackingStore.FetchInstanceStatusAsync(childInstanceId); + Assert.AreEqual(parentInstanceId, status.State.ParentInstance?.OrchestrationInstance.InstanceId); + } + + /// + /// Verifies the repair path also overwrites a stale parent on an Instances row that already exists. + /// + [TestMethod] + public async Task CompletedOrchestrationRepair_OverwritesStaleParentInstanceId() + { + string parentInstanceId = $"parent-{Guid.NewGuid():N}"; + string childInstanceId = $"{parentInstanceId}:child"; + + await this.SeedInstanceRowAsync(childInstanceId, "stale-parent"); + + await this.trackingStore.UpdateInstanceStatusForCompletedOrchestrationAsync( + childInstanceId, + executionId: "execution-1", + runtimeState: CreateCompletedRuntimeState(childInstanceId, "execution-1", parentInstanceId), + instanceEntityExists: true); + + Assert.AreEqual(parentInstanceId, await this.GetRawParentInstanceIdAsync(childInstanceId)); + } + + /// + /// Verifies the initial instance-creation write persists a non-null parent. This is the write that + /// happens when an orchestration is created through the client creation path with a parent supplied + /// on the ExecutionStartedEvent. + /// + [TestMethod] + public async Task SetNewExecution_PersistsNonNullParentInstanceId() + { + string parentInstanceId = $"parent-{Guid.NewGuid():N}"; + string childInstanceId = $"{parentInstanceId}:child"; + + bool created = await this.trackingStore.SetNewExecutionAsync( + CreateExecutionStartedEvent(childInstanceId, "execution-1", parentInstanceId), + eTag: null, + inputPayloadOverride: null); + + Assert.IsTrue(created); + Assert.AreEqual(parentInstanceId, await this.GetRawParentInstanceIdAsync(childInstanceId)); + + InstanceStatus status = await this.trackingStore.FetchInstanceStatusAsync(childInstanceId); + Assert.AreEqual(parentInstanceId, status.State.ParentInstance?.OrchestrationInstance.InstanceId); + } + + /// + /// Verifies the initial creation write clears a stale parent when the new execution has none. A + /// re-created instance reuses the partition key, and this write path can replace an earlier row. + /// + [TestMethod] + public async Task SetNewExecution_ClearsStaleParentInstanceId() + { + string instanceId = $"recreate-{Guid.NewGuid():N}"; + + await this.SeedInstanceRowAsync(instanceId, "stale-parent"); + OrchestrationInstanceStatus seeded = await this.GetRawEntityAsync(instanceId); + + bool created = await this.trackingStore.SetNewExecutionAsync( + CreateExecutionStartedEvent(instanceId, "execution-2", parentInstanceId: null), + eTag: new ETag(seeded.ETag.ToString()), + inputPayloadOverride: null); + + Assert.IsTrue(created); + Assert.AreEqual(string.Empty, await this.GetRawParentInstanceIdAsync(instanceId)); + } + + /// + /// Seeds an Instances row that already carries a parent ID, simulating a row written by a previous + /// orchestration that reused the same instance ID. + /// + async Task SeedInstanceRowAsync(string instanceId, string parentInstanceId) + { + var entity = new TableEntity(KeySanitation.EscapePartitionKey(instanceId), string.Empty) + { + ["Name"] = "SeededOrchestration", + ["RuntimeStatus"] = OrchestrationStatus.Running.ToString(), + ["CreatedTime"] = DateTime.UtcNow, + ["LastUpdatedTime"] = DateTime.UtcNow, + ["TaskHubName"] = this.taskHubName, + ["ExecutionId"] = "execution-0", + }; + + // A null parent seeds a row with no property at all, which is also the legacy row shape. + if (parentInstanceId != null) + { + entity[ParentInstanceIdProperty] = parentInstanceId; + } + + await this.trackingStore.InstancesTable.InsertOrMergeEntityAsync(entity); + } + + /// + /// UpdateStateAsync casts the tracking-store context to a private type, so the only legitimate way + /// to obtain one is from the production history read, which is exactly what the dispatcher does. + /// + async Task GetTrackingStoreContextAsync(string instanceId) + { + OrchestrationHistory history = await this.trackingStore.GetHistoryEventsAsync(instanceId, expectedExecutionId: null); + return history.TrackingStoreContext; + } + + async Task GetRawEntityAsync(string instanceId) + { + string filter = AzureTableQueryFilter.PartitionKeyEquals(KeySanitation.EscapePartitionKey(instanceId)); + await foreach (OrchestrationInstanceStatus entity in this.trackingStore.InstancesTable.ExecuteQueryAsync(filter)) + { + return entity; + } + + return null; + } + + /// + /// Reads the stored property directly rather than the converted state, so that a value which was + /// merely left untouched by a merge is still visible to the assertion. + /// + async Task GetRawParentInstanceIdAsync(string instanceId) + { + OrchestrationInstanceStatus entity = await this.GetRawEntityAsync(instanceId); + Assert.IsNotNull(entity, $"Expected an Instances row for '{instanceId}'."); + return entity.ParentInstanceId; + } + + static ExecutionStartedEvent CreateExecutionStartedEvent(string instanceId, string executionId, string parentInstanceId) + { + var executionStartedEvent = new ExecutionStartedEvent(-1, "input") + { + Name = "TestOrchestration", + Version = string.Empty, + OrchestrationInstance = new OrchestrationInstance + { + InstanceId = instanceId, + ExecutionId = executionId, + }, + }; + + if (parentInstanceId != null) + { + executionStartedEvent.ParentInstance = new ParentInstance + { + OrchestrationInstance = new OrchestrationInstance + { + InstanceId = parentInstanceId, + ExecutionId = "parent-execution", + }, + Name = "ParentOrchestration", + Version = string.Empty, + TaskScheduleId = 1, + }; + } + + return executionStartedEvent; + } + + static OrchestrationRuntimeState CreateCompletedRuntimeState(string instanceId, string executionId, string parentInstanceId) + { + var runtimeState = new OrchestrationRuntimeState(); + runtimeState.AddEvent(CreateExecutionStartedEvent(instanceId, executionId, parentInstanceId)); + runtimeState.AddEvent(new ExecutionCompletedEvent(-1, "output", OrchestrationStatus.Completed)); + return runtimeState; + } + } +}