Persist parent orchestration metadata in Azure Storage

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; }

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<OrchestrationState> 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<bool> 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);

src/DurableTask.AzureStorage/Tracking/InstanceStoreBackedTrackingStore.cs

@@ -108,6 +108,7 @@ public override async Task<bool> SetNewExecutionAsync(
                 Name = executionStartedEvent.Name,
                 Version = executionStartedEvent.Version,
                 OrchestrationInstance = executionStartedEvent.OrchestrationInstance,
+                ParentInstance = executionStartedEvent.ParentInstance,
                 OrchestrationStatus = OrchestrationStatus.Pending,
                 Input = inputStatusOverride ?? executionStartedEvent.Input,
                 Tags = executionStartedEvent.Tags,

test/DurableTask.AzureStorage.Tests/AzureStorageScenarioTests.cs

@@ -154,6 +154,78 @@ public async Task ParentOfSequentialOrchestration()
             }
         }
 
+        /// <summary>
+        /// 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
+        /// <see cref="ParentInstanceIdTrackingStoreTests.CompletedOrchestrationRepair_PersistsParentInstanceId"/>
+        /// for that.
+        /// </summary>
+        [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();
+            }
+        }
+
         /// <summary>
         /// End-to-end test which runs a slow orchestrator that causes work item renewal
         /// </summary>
@@ -4875,6 +4947,51 @@ public override Task<int> RunTask(OrchestrationContext context, int input)
                 }
             }
 
+            [KnownType(typeof(InlineChild))]
+            internal class ParentOfInlineChild : TaskOrchestration<string, string>
+            {
+                public override Task<string> RunTask(OrchestrationContext context, string input)
+                {
+                    return context.CreateSubOrchestrationInstance<string>(
+                        typeof(InlineChild),
+                        context.OrchestrationInstance.InstanceId + ":child",
+                        input);
+                }
+            }
+
+            internal class InlineChild : TaskOrchestration<string, string>
+            {
+                public override Task<string> RunTask(OrchestrationContext context, string input)
+                {
+                    return Task.FromResult(input);
+                }
+            }
+
+            [KnownType(typeof(ContinueAsNewChild))]
+            internal class ParentOfContinueAsNewChild : TaskOrchestration<int, int>
+            {
+                public override Task<int> RunTask(OrchestrationContext context, int input)
+                {
+                    return context.CreateSubOrchestrationInstance<int>(
+                        typeof(ContinueAsNewChild),
+                        context.OrchestrationInstance.InstanceId + ":child",
+                        input);
+                }
+            }
+
+            internal class ContinueAsNewChild : TaskOrchestration<int, int>
+            {
+                public override Task<int> RunTask(OrchestrationContext context, int input)
+                {
+                    if (input == 0)
+                    {
+                        context.ContinueAsNew(1);
+                    }
+
+                    return Task.FromResult(input);
+                }
+            }
+
             [KnownType(typeof(Activities.Hello))]
             internal class DoubleFanOut : TaskOrchestration<string, string>
             {

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<IOrchestrationServiceInstanceStore>(MockBehavior.Strict);
+            instanceStore
+                .Setup(store => store.WriteEntitiesAsync(It.IsAny<IEnumerable<InstanceEntityBase>>()))
+                .Callback<IEnumerable<InstanceEntityBase>>(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);
         }
     }
 }

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));
         }
 
+        /// <summary>
+        /// 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.
+        /// </summary>
+        [TestMethod]
+        public void OrchestrationInstanceQuery_ProjectionRetainsParentInstanceId()
+        {
+            var condition = new OrchestrationInstanceStatusQueryCondition
+            {
+                RuntimeStatus = new OrchestrationStatus[] { OrchestrationStatus.Running },
+                FetchInput = false,
+                FetchOutput = false,
+            };
+
+            IEnumerable<string> 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()
         {