Re-add ScheduledStartTime to orchestration scheduling in .NET isolated (

#2805) Also adds E2E tests for ScheduledStartTime in isolated Co-authored-by: Andy Staples <[email protected]>
Azure · Jan 15, 2025 · ca44d7e · ca44d7e
1 parent 89b743e
commit ca44d7e
Show file tree

Hide file tree

Showing 5 changed files with 150 additions and 17 deletions.
diff --git a/src/WebJobs.Extensions.DurableTask/LocalGrpcListener.cs b/src/WebJobs.Extensions.DurableTask/LocalGrpcListener.cs
@@ -154,11 +154,26 @@ public override Task<Empty> Hello(Empty request, ServerCallContext context)
             {
                 try
                 {
-                    string instanceId = await this.GetClient(context).StartNewAsync(
-                        request.Name, request.InstanceId, Raw(request.Input));
+                    string instanceId = request.InstanceId ?? Guid.NewGuid().ToString("N");
+                    TaskHubClient taskhubClient = new TaskHubClient(this.GetDurabilityProvider(context));
+                    OrchestrationInstance instance;
+
+                    // TODO: Ideally, we'd have a single method in the taskhubClient that can handle both scheduled and non-scheduled starts.
+                    // TODO: the type of `ScheduledStartTimestamp` is not nullable. Can we change it to `DateTime?` in the proto file?
+                    if (request.ScheduledStartTimestamp != null)
+                    {
+                        instance = await taskhubClient.CreateScheduledOrchestrationInstanceAsync(
+                            name: request.Name, version: request.Version, instanceId: instanceId, input: Raw(request.Input), startAt: request.ScheduledStartTimestamp.ToDateTime());
+                    }
+                    else
+                    {
+                        instance = await taskhubClient.CreateOrchestrationInstanceAsync(request.Name, request.Version, instanceId, Raw(request.Input));
+                    }
+
+                    // TODO: should this not include the ExecutionId and other elements of the taskhubClient response?
                     return new P.CreateInstanceResponse
                     {
-                        InstanceId = instanceId,
+                        InstanceId = instance.InstanceId,
                     };
                 }
                 catch (OrchestrationAlreadyExistsException)
@@ -231,13 +246,13 @@ public override Task<Empty> Hello(Empty request, ServerCallContext context)
                 EntityBackendQueries.EntityQueryResult result = await entityOrchestrationService.EntityBackendQueries!.QueryEntitiesAsync(
                     new EntityBackendQueries.EntityQuery()
                     {
-                         InstanceIdStartsWith = query.InstanceIdStartsWith,
-                         LastModifiedFrom = query.LastModifiedFrom?.ToDateTime(),
-                         LastModifiedTo = query.LastModifiedTo?.ToDateTime(),
-                         IncludeTransient = query.IncludeTransient,
-                         IncludeState = query.IncludeState,
-                         ContinuationToken = query.ContinuationToken,
-                         PageSize = query.PageSize,
+                        InstanceIdStartsWith = query.InstanceIdStartsWith,
+                        LastModifiedFrom = query.LastModifiedFrom?.ToDateTime(),
+                        LastModifiedTo = query.LastModifiedTo?.ToDateTime(),
+                        IncludeTransient = query.IncludeTransient,
+                        IncludeState = query.IncludeState,
+                        ContinuationToken = query.ContinuationToken,
+                        PageSize = query.PageSize,
                     },
                     context.CancellationToken);
 

diff --git a/test/e2e/Apps/BasicDotNetIsolated/HelloCities.cs b/test/e2e/Apps/BasicDotNetIsolated/HelloCities.cs
@@ -54,5 +54,27 @@ public static async Task<HttpResponseData> HttpStart(
             // See https://learn.microsoft.com/azure/azure-functions/durable/durable-functions-http-api#start-orchestration
             return await client.CreateCheckStatusResponseAsync(req, instanceId);
         }
+
+        [Function("HelloCities_HttpStart_Scheduled")]
+        public static async Task<HttpResponseData> HttpStartScheduled(
+            [HttpTrigger(AuthorizationLevel.Anonymous, "get", "post")] HttpRequestData req,
+            [DurableClient] DurableTaskClient client,
+            FunctionContext executionContext,
+            DateTime scheduledStartTime)
+        {
+            ILogger logger = executionContext.GetLogger("HelloCities_HttpStart");
+
+            var startOptions = new StartOrchestrationOptions(StartAt: scheduledStartTime);
+
+            // Function input comes from the request content.
+            string instanceId = await client.ScheduleNewOrchestrationInstanceAsync(
+                nameof(HelloCities), startOptions);
+
+            logger.LogInformation("Started orchestration with ID = '{instanceId}'.", instanceId);
+
+            // Returns an HTTP 202 response with an instance management payload.
+            // See https://learn.microsoft.com/azure/azure-functions/durable/durable-functions-http-api#start-orchestration
+            return await client.CreateCheckStatusResponseAsync(req, instanceId);
+        }
     }
 }
diff --git a/test/e2e/Tests/E2ETests.csproj b/test/e2e/Tests/E2ETests.csproj
@@ -1,4 +1,4 @@
-<Project Sdk="Microsoft.NET.Sdk">
+<Project Sdk="Microsoft.NET.Sdk">
 
   <PropertyGroup>
     <TargetFramework>net8.0</TargetFramework>

diff --git a/test/e2e/Tests/Helpers/DurableHelpers.cs b/test/e2e/Tests/Helpers/DurableHelpers.cs
@@ -0,0 +1,59 @@
+// Copyright (c) .NET Foundation. All rights reserved.
+// Licensed under the MIT License. See License.txt in the project root for license information.
+
+using System.Text.Json.Nodes;
+
+namespace Microsoft.Azure.Durable.Tests.DotnetIsolatedE2E;
+
+internal class DurableHelpers
+{
+    static readonly HttpClient _httpClient = new HttpClient();
+
+    internal class OrchestrationStatusDetails
+    {
+        public string RuntimeStatus { get; set; } = string.Empty;
+        public string Input { get; set; } = string.Empty;
+        public string Output { get; set; } = string.Empty;
+        public DateTime CreatedTime { get; set; }
+        public DateTime LastUpdatedTime { get; set; }
+        public OrchestrationStatusDetails(string statusQueryResponse)
+        {
+            JsonNode? statusQueryJsonNode = JsonNode.Parse(statusQueryResponse);
+            if (statusQueryJsonNode == null)
+            {
+                return;
+            }
+            this.RuntimeStatus = statusQueryJsonNode["runtimeStatus"]?.GetValue<string>() ?? string.Empty;
+            this.Input = statusQueryJsonNode["input"]?.ToString() ?? string.Empty;
+            this.Output = statusQueryJsonNode["output"]?.ToString() ?? string.Empty;
+            this.CreatedTime = DateTime.Parse(statusQueryJsonNode["createdTime"]?.GetValue<string>() ?? string.Empty).ToUniversalTime();
+            this.LastUpdatedTime = DateTime.Parse(statusQueryJsonNode["lastUpdatedTime"]?.GetValue<string>() ?? string.Empty).ToUniversalTime();
+        }
+    }
+
+    internal static string ParseStatusQueryGetUri(HttpResponseMessage invocationStartResponse)
+    {
+        string? responseString = invocationStartResponse.Content?.ReadAsStringAsync().Result;
+
+        if (string.IsNullOrEmpty(responseString))
+        {
+            return string.Empty;
+        }
+        JsonNode? responseJsonNode = JsonNode.Parse(responseString);
+        if (responseJsonNode == null)
+        {
+            return string.Empty;
+        }
+
+        string? statusQueryGetUri = responseJsonNode["StatusQueryGetUri"]?.GetValue<string>();
+        return statusQueryGetUri ?? string.Empty;
+    }
+    internal static OrchestrationStatusDetails GetRunningOrchestrationDetails(string statusQueryGetUri)
+    {
+        var statusQueryResponse = _httpClient.GetAsync(statusQueryGetUri);
+
+        string? statusQueryResponseString = statusQueryResponse.Result.Content.ReadAsStringAsync().Result;
+
+        return new OrchestrationStatusDetails(statusQueryResponseString);
+    }
+}
diff --git a/test/e2e/Tests/Tests/HelloCitiesTest.cs b/test/e2e/Tests/Tests/HelloCitiesTest.cs
@@ -7,29 +7,66 @@
 
 namespace Microsoft.Azure.Durable.Tests.DotnetIsolatedE2E;
 
-[Collection(Constants.FunctionAppCollectionName )]
+[Collection(Constants.FunctionAppCollectionName)]
 public class HttpEndToEndTests
 {
     private readonly FunctionAppFixture _fixture;
+    private readonly ITestOutputHelper _output;
 
     public HttpEndToEndTests(FunctionAppFixture fixture, ITestOutputHelper testOutputHelper)
     {
         _fixture = fixture;
         _fixture.TestLogs.UseTestLogger(testOutputHelper);
+        _output = testOutputHelper;
     }
 
     [Theory]
-    [InlineData("HelloCities_HttpStart", "", HttpStatusCode.Accepted, "")]
-    public async Task HttpTriggerTests(string functionName, string queryString, HttpStatusCode expectedStatusCode, string expectedMessage)
+    [InlineData("HelloCities_HttpStart", HttpStatusCode.Accepted, "Hello Tokyo!")]
+    public async Task HttpTriggerTests(string functionName, HttpStatusCode expectedStatusCode, string partialExpectedOutput)
     {
-        using HttpResponseMessage response = await HttpHelpers.InvokeHttpTrigger(functionName, queryString);
+        using HttpResponseMessage response = await HttpHelpers.InvokeHttpTrigger(functionName, "");
         string actualMessage = await response.Content.ReadAsStringAsync();
 
         Assert.Equal(expectedStatusCode, response.StatusCode);
+        string statusQueryGetUri = DurableHelpers.ParseStatusQueryGetUri(response);
+        Thread.Sleep(1000);
+        var orchestrationDetails = DurableHelpers.GetRunningOrchestrationDetails(statusQueryGetUri);
+        Assert.Equal("Completed", orchestrationDetails.RuntimeStatus);
+        Assert.Contains(partialExpectedOutput, orchestrationDetails.Output);
+    }
+
+    [Theory]
+    [InlineData("HelloCities_HttpStart_Scheduled", 5, HttpStatusCode.Accepted)]
+    [InlineData("HelloCities_HttpStart_Scheduled", -5, HttpStatusCode.Accepted)]
+    public async Task ScheduledStartTests(string functionName, int startDelaySeconds, HttpStatusCode expectedStatusCode)
+    {
+        var testStartTime = DateTime.UtcNow;
+        var scheduledStartTime = testStartTime + TimeSpan.FromSeconds(startDelaySeconds);
+        string urlQueryString = $"?ScheduledStartTime={scheduledStartTime.ToString("o")}";
+
+        using HttpResponseMessage response = await HttpHelpers.InvokeHttpTrigger(functionName, urlQueryString);
+        string actualMessage = await response.Content.ReadAsStringAsync();
 
-        if (!string.IsNullOrEmpty(expectedMessage))
+        string statusQueryGetUri = DurableHelpers.ParseStatusQueryGetUri(response);
+
+        Assert.Equal(expectedStatusCode, response.StatusCode);
+
+        var orchestrationDetails = DurableHelpers.GetRunningOrchestrationDetails(statusQueryGetUri);
+        while (DateTime.UtcNow < scheduledStartTime)
         {
-            Assert.False(string.IsNullOrEmpty(actualMessage));
+            _output.WriteLine($"Test scheduled for {scheduledStartTime}, current time {DateTime.Now}");
+            orchestrationDetails = DurableHelpers.GetRunningOrchestrationDetails(statusQueryGetUri);
+            Assert.Equal("Pending", orchestrationDetails.RuntimeStatus);
+            Thread.Sleep(3000);
         }
+
+        // Give a small amount of time for the orchestration to complete, even if scheduled to run immediately
+        Thread.Sleep(1000);
+        _output.WriteLine($"Test scheduled for {scheduledStartTime}, current time {DateTime.Now}, looking for completed");
+
+        var finalOrchestrationDetails = DurableHelpers.GetRunningOrchestrationDetails(statusQueryGetUri);
+        Assert.Equal("Completed", finalOrchestrationDetails.RuntimeStatus);
+
+        Assert.True(finalOrchestrationDetails.LastUpdatedTime > scheduledStartTime);
     }
 }