played by rules with durable functions adding orchstrators and activity

Author: gbelenky

AsyncJob.cs

@@ -4,15 +4,20 @@
 using Microsoft.Azure.Functions.Worker.Http;
 using Microsoft.DurableTask;
 using Microsoft.DurableTask.Client;
+using Microsoft.DurableTask.Entities;
 using Microsoft.Extensions.Logging;
 
 
 namespace gbelenky.AsyncJob;
 
+// Define the state of the JobEntity
+public record JobEntityState(string ThirdPartyJobId, string JobName);
+
+
 public static class AsyncJob
 {
-    [Function(nameof(AsyncJobOrchestrator))]
-    public static async Task<List<string>> AsyncJobOrchestrator(
+    [Function(nameof(JobStartOrchestrator))] // Renamed
+    public static async Task<List<string>> JobStartOrchestrator( // Renamed
         [OrchestrationTrigger] TaskOrchestrationContext context)
     {
         ILogger logger = context.CreateReplaySafeLogger(nameof(AsyncJob));
@@ -46,55 +51,95 @@ public static async Task<List<string>> AsyncJobOrchestrator(
         return outputs;
     }
 
+    public record StatusPayload(string? JobStatus);
+
+    [Function(nameof(GetJobStatusActivity))]
+    public static async Task<string> GetJobStatusActivity(
+        [ActivityTrigger] string targetInstanceId,
+        [DurableClient] DurableTaskClient durableTaskClient, // Inject DurableTaskClient
+        FunctionContext executionContext)
+    {
+        ILogger logger = executionContext.GetLogger(nameof(GetJobStatusActivity));
+        logger.LogInformation("Activity: Attempting to fetch status for instance ID: {TargetInstanceId}", targetInstanceId);
+
+        OrchestrationMetadata? instanceMetadata = await durableTaskClient.GetInstanceAsync(targetInstanceId, getInputsAndOutputs: true);
+
+        if (instanceMetadata == null)
+        {
+            logger.LogWarning("Activity: Instance not found: {TargetInstanceId}", targetInstanceId);
+            return "NotFound";
+        }
+
+        // The custom status is a JToken, deserialize it to string.
+        // Corrected access to custom status via SerializedCustomStatus and ensuring it's not null before deserializing.
+        string? customStatus = null;
+        if (instanceMetadata.SerializedCustomStatus != null)
+        {
+            customStatus = System.Text.Json.JsonSerializer.Deserialize<string>(instanceMetadata.SerializedCustomStatus);
+        }
+
+        logger.LogInformation("Activity: Successfully fetched status for {TargetInstanceId}. Status: '{CustomStatus}'", targetInstanceId, customStatus ?? "null");
+        return customStatus;
+    }
 
 
+    [Function(nameof(JobStatusOrchestrator))]
+    public static async Task<string> JobStatusOrchestrator(
+        [OrchestrationTrigger] TaskOrchestrationContext context)
+    {
+        string instanceId = context.GetInput<string>() ?? string.Empty;
+        var customStatus = await context.CallActivityAsync<string>(nameof(GetJobStatusActivity), instanceId);
+        return customStatus ?? "Unknown";
+    }
 
     // add a function to retuen the status of the job
-    [Function(nameof(AsyncJobStatus))]
-    public static async Task<HttpResponseData> AsyncJobStatus(
+    [Function(nameof(JobStatusTrigger))] // Renamed
+    public static async Task<HttpResponseData> JobStatusTrigger( // Renamed
         [HttpTrigger(AuthorizationLevel.Anonymous, "get", Route = "job-status/{jobName}")] HttpRequestData req,
-        [DurableClient] DurableTaskClient client,
-        FunctionContext executionContext, string jobName)
+        [DurableClient] DurableTaskClient client, // This is the DurableTaskClient
+        FunctionContext executionContext,
+        string jobName)
     {
-        ILogger logger = executionContext.GetLogger("AsyncJobStatus");
-        string instanceId = $"job-{jobName}";
+        ILogger logger = executionContext.GetLogger(nameof(JobStatusTrigger));
+        string targetJobInstanceId = $"job-{jobName}"; // Instance ID of the job we want to get status for
 
-        var host = req.Url.Host;
+        logger.LogInformation("Trigger: Received request to get status for job: {JobName} (Instance ID: {TargetJobInstanceId})", jobName, targetJobInstanceId);
 
-        // Build the Durable Functions status URL correctly using string interpolation
-        string statusUrl = $"{req.Url.Scheme}://{host}:{req.Url.Port}/runtime/webhooks/durabletask/instances/{instanceId}";
-        HttpClient httpClient = new HttpClient();
-        var response = await httpClient.GetAsync(statusUrl);
-        var responseContent = await response.Content.ReadAsStringAsync();
-        string? customStatus = null;
-        try
-        {
-            using var doc = System.Text.Json.JsonDocument.Parse(responseContent);
-            if (doc.RootElement.TryGetProperty("customStatus", out var statusProp))
-            {
-                customStatus = statusProp.GetString();
-            }
-        }
-        catch
+        // Start the orchestrator that will fetch the status.
+        // The input to JobStatusOrchestrator is the instance ID of the job whose status is being queried.
+        string statusQueryOrchestrationId = await client.ScheduleNewOrchestrationInstanceAsync(
+            nameof(JobStatusOrchestrator),
+            targetJobInstanceId);
+
+        logger.LogInformation("Trigger: Started status query orchestration with ID = '{StatusQueryOrchestrationId}' to get status for job '{JobName}'.", statusQueryOrchestrationId, jobName);
+
+        // Wait for the orchestration to complete (timeout after 30 seconds)
+        var timeout = TimeSpan.FromSeconds(30);
+        using var cts = new CancellationTokenSource(timeout);
+        var orchestrationStatus = await client.WaitForInstanceCompletionAsync(statusQueryOrchestrationId, true, cts.Token);
+
+        var response = req.CreateResponse(HttpStatusCode.OK);
+        if (orchestrationStatus == null)
         {
-            // Optionally log or handle JSON parse errors
+            await response.WriteAsJsonAsync(new { status = "Timeout while retrieving job status" });
+            return response;
         }
-        var httpResponse = req.CreateResponse(HttpStatusCode.OK);
-        string json = System.Text.Json.JsonSerializer.Serialize(new { jobStatus = customStatus });
-        logger.LogInformation("Job {JobId} status: {Status}", instanceId, json);
-        await httpResponse.WriteStringAsync(json);
-        return httpResponse;
-    }
 
+        var statusPayload = orchestrationStatus.ReadOutputAs<string>();
+        await response.WriteAsJsonAsync(new {
+            status = statusPayload ?? "Unknown"
+        });
+        return response;
+    }
 
-    [Function("AsyncJobTrigger")]
-    public static async Task<HttpResponseData> AsyncJobTrigger(
+    [Function("JobStartTrigger")] // Renamed from AsyncJobTrigger
+    public static async Task<HttpResponseData> JobStartTrigger( // Renamed from AsyncJobTrigger
         [HttpTrigger(AuthorizationLevel.Anonymous, "get", Route = "job-start/{jobName}")] HttpRequestData req,
         [DurableClient] DurableTaskClient client,
         FunctionContext executionContext, string jobName)
     {
         // The function input comes from the request content.
-        ILogger logger = executionContext.GetLogger("AsyncJobTrigger");
+        ILogger logger = executionContext.GetLogger("JobStartTrigger"); // Updated logger category
         // extract ASYNC_JOB_QUEUED_DURATION_SEC from the environment variable
         double queuedDuration = double.TryParse(
             Environment.GetEnvironmentVariable("ASYNC_JOB_QUEUED_DURATION_SEC"),
@@ -119,7 +164,7 @@ public static async Task<HttpResponseData> AsyncJobTrigger(
 
             // Start new orchestration instance, specifying the desired instance ID via options
             await client.ScheduleNewOrchestrationInstanceAsync(
-                nameof(AsyncJobOrchestrator),
+                nameof(JobStartOrchestrator), // Updated to new name
                 orchParams,
                 new StartOrchestrationOptions { InstanceId = instanceId });
             logger.LogInformation("Started orchestration with ID = '{InstanceId}' for job '{JobName}'.", instanceId, jobName);
@@ -139,4 +184,6 @@ private record OrchParams(
         string ThirdPartyJobId,
         double QueuedDuration,
         double InProgressDuration);
+        
+    
 }

README.md

@@ -12,7 +12,7 @@ lightweight async job implementation
 2. **Durable Task Framework**: The project employs the Durable Task framework for managing long-running workflows and state persistence.
 3. **Customizable Durations**: Job states such as `Queued` and `InProgress` can be customized using environment variables.
 4. **State Management**: Provides detailed logging and state transitions for each job instance.
-5. **HTTP Trigger Support**: The `AsyncJobTrigger` function allows initiating jobs via HTTP requests.
+5. **HTTP Trigger Support**: The `JobStartTrigger` function allows initiating jobs via HTTP requests.
 
 ### Environment Variables:
 The project uses the following environment variables for configuration:
@@ -37,11 +37,11 @@ You can query the current status of a job instance using the AsyncJobStatus HTTP
 /job-status/{jobName}
 ```
 
-This endpoint returns the current status and details of the specified job instance. The response includes the job's name, instance ID, runtime status, input parameters, custom status, output, and timestamps. Example response:
+This endpoint returns the current status of the specified job instance. Example response:
 
 ```json
 {
-  "jobStatus": "Queued"
+  "status": "Queued"
 }
 ```
 

test.http

@@ -1,7 +1,7 @@
-GET http://localhost:7071/api/job-start/gb4722 HTTP/1.1
+GET http://localhost:7071/api/job-start/gb4730 HTTP/1.1
 
 ###
-GET http://localhost:7071/api/job-status/gb4722  HTTP/1.1
+GET http://localhost:7071/api/job-status/gb4730  HTTP/1.1
 
 ###
 GET http://localhost:7071/api/delay HTTP/1.1