NullReferenceException in OrchestrationEntityContext.RecoverLockAfterCall

[mllab-nl]

Description

I observe null reference in /_/src/DurableTask.Core/Entities/OrchestrationEntityContext.cs:line 193

As it looks like things are not going as expected.
Could you please check if it is not a result of "Incorrect usage"

Simplified use case:
We need to generate reports.
To speed up, scale out and reuse(cache) the report is broken into logically independent pieces
Each peace is calculated separately as a separate activity and its results are cached into Stateful entity.
Before calculating specific piece the corresponding entity is locked to avoid work duplication.

Thus the orchestration creates a lot of tasks.
Each task reads the cached entity and if it is not calculated yet, locks it, calls the heavy activity, updates the cached entity with the result.
The orchestration combines the result (lightweight operation, last part of the orchestation)

Often, when some of the pieces are not yet in the cache the orchestration will fail with the Null Reference in the Core.

1. Would you suggest to implement the use case in this way?
2. Do we use it correctly?

Expected behavior

It works

Actual behavior

Null Reference Exception in the Core framework.

Relevant source code snippets

using Microsoft.Azure.Functions.Worker;
using Microsoft.Azure.Functions.Worker.Http;
using Microsoft.DurableTask;
using Microsoft.DurableTask.Client;
using Microsoft.DurableTask.Entities;
using Microsoft.Extensions.Logging;
using System.Collections.Concurrent;

namespace ReportBuilder.Orchestations
{
    public static class EntityLockNullReferenceOrchestation
    {
        [Function(nameof(EntityLockNullReferenceOrchestation))]
        public static async Task<string> RunOrchestrator([OrchestrationTrigger] TaskOrchestrationContext context, string[] tags)
        {
            ILogger logger = context.CreateReplaySafeLogger(nameof(EntityLockNullReferenceOrchestation));
            var orchestrationId = context.InstanceId;

            var tagsData = new ConcurrentQueue<Tuple<string, string>>();
            var categoriesReportsTasks = tags.Select(tag => GetOrBuildTagData(context, tag, tagsData)).ToArray();
            await Task.WhenAll(categoriesReportsTasks);
            return string.Join(", ", tagsData.Select(t => $"{t.Item1}: {t.Item2}"));
        }
        private static async Task GetOrBuildTagData(TaskOrchestrationContext context, string tag, ConcurrentQueue<Tuple<string, string>> tagsData)
        {
            var dataSet1Id = new EntityInstanceId(nameof(DummyTagCacheEntity), tag);
            var tagResult = await context.Entities.CallEntityAsync<string>(dataSet1Id, nameof(DummyTagCacheEntity.Get));
            if (string.IsNullOrEmpty(tagResult))
            {
                await using (await context.Entities.LockEntitiesAsync(dataSet1Id))
                {
                    tagResult = await context.CallActivityAsync<string>(nameof(DummyTagReport), tag);
                    await context.Entities.CallEntityAsync(dataSet1Id, nameof(DummyTagCacheEntity.Set), tagResult);
                }
            }
            tagsData.Enqueue(new(tag, tagResult));
        }
        
        [Function(nameof(DummyTagReport))]
        public static async Task<string> DummyTagReport([ActivityTrigger] string tag, FunctionContext executionContext)
        {
            ILogger logger = executionContext.GetLogger(nameof(DummyTagReport));
            logger.LogInformation("Generating report for tag '{tag}'", tag);

            // Simulate async call sending notification
            await Task.Delay(TimeSpan.FromSeconds(5));

            return $"{tag}.Result";
        }


        [Function($"{nameof(EntityLockNullReferenceOrchestation)}{nameof(HttpStart)}")]
        public static async Task<HttpResponseData> HttpStart(
            [HttpTrigger(AuthorizationLevel.Anonymous, "get", "post")] HttpRequestData req,
            [DurableClient] DurableTaskClient client,
            FunctionContext executionContext)
        {
            var tags = req.Query["tags"] ?? "tag1,tag2,";
            var tagsArray = tags.Split(',', StringSplitOptions.RemoveEmptyEntries).Select(tag => tag.Trim()).ToArray();

            
            ILogger logger = executionContext.GetLogger(nameof(EntityLockNullReferenceOrchestation) + nameof(HttpStart));

            string instanceId = await client.ScheduleNewOrchestrationInstanceAsync(nameof(EntityLockNullReferenceOrchestation), tagsArray);
            logger.LogInformation("Started orchestration with ID = '{instanceId}'.", instanceId);
            return await client.CreateCheckStatusResponseAsync(req, instanceId);
        }

    }
    public class DummyTagCacheEntity: TaskEntity<string>
    {
        public string Get() => State;

        public void Set(string value) => State = value;
        protected override string InitializeState(TaskEntityOperation entityOperation)
        {
            return string.Empty;
        }
        [Function(nameof(DummyTagCacheEntity))]
        public Task RunEntityAsync([EntityTrigger] TaskEntityDispatcher dispatcher)
        {
            return dispatcher.DispatchAsync(this);
        }
    }
}

Known workarounds

Provide a description of any known workarounds you used.

App Details

Local
All latest versions

<FrameworkReference Include="Microsoft.AspNetCore.App" />
<PackageReference Include="Microsoft.ApplicationInsights.WorkerService" Version="2.23.0" />
	<PackageReference Include="Microsoft.Azure.Functions.Worker.Extensions.Http" Version="3.3.0" />
	<PackageReference Include="Microsoft.Azure.Functions.Worker.Extensions.Timer" Version="4.3.1" />
	<PackageReference Include="Microsoft.Azure.Functions.Worker.Extensions.DurableTask" Version="1.7.1" />
	<PackageReference Include="Microsoft.Azure.Functions.Worker" Version="2.0.0" />
	<PackageReference Include="Microsoft.Azure.Functions.Worker.Sdk" Version="2.0.5" />
	<PackageReference Include="Microsoft.VisualStudio.Azure.Containers.Tools.Targets" Version="1.22.1" />
	<PackageReference Include="Microsoft.Azure.Functions.Worker.ApplicationInsights" Version="2.0.0" />
<PackageReference Include="Microsoft.Azure.Functions.Worker.Extensions.Http.AspNetCore" Version="2.0.2" />

Screenshots

If applicable, add screenshots to help explain your problem.

[andystaples]

Hi @mllab-nl
This pattern is actually unsupported. Yielding Task.WhenAll as part of a Durable orchestration is only supported when the tasks are single Durable operations. Calling Task.WhenAll on a method that is comprised of multiple operations (like entity calls and locks in your example with EntityLockNullReferenceOrchestration calling GetOrBuildTagData) can result in many problems, including non-determinism exceptions. In your case, the actual issue is that you are effectively trying to create two entity locks in parallel, when an orchestration can only hold one lock at a time.
We should not be throwing a NullReferenceException here, and I have opened a PR in DurableTask.Core that avoids the NullReferenceException and exposes the real issue, which is attempting to hold multiple entity locks simultaneously: Azure/durabletask#1248

You can accomplish your goals with sub-orchestrators. Here is the example you provided modified to use sub-orchestrators, and it runs successfully:

        [Function(nameof(EntityLockNullReferenceOrchestation))]
        public static async Task<string> RunOrchestrator([OrchestrationTrigger] TaskOrchestrationContext context, string[] tags)
        {
            ILogger logger = context.CreateReplaySafeLogger(nameof(EntityLockNullReferenceOrchestation));
            var orchestrationId = context.InstanceId;

            var categoriesReportsTasks = tags.Select(tag => context.CallSubOrchestratorAsync<Tuple<string, string>>(nameof(GetOrBuildTagData), tag)).ToArray();
            var tagsData = await Task.WhenAll(categoriesReportsTasks);
            return string.Join(", ", tagsData.Select(t => $"{t.Item1}: {t.Item2}"));
        }

        [Function(nameof(GetOrBuildTagData))]
        public static async Task<Tuple<string, string>> GetOrBuildTagData([OrchestrationTrigger] TaskOrchestrationContext context, string tag)
        {
            var dataSet1Id = new EntityInstanceId(nameof(DummyTagCacheEntity), tag);
            var tagResult = await context.Entities.CallEntityAsync<string>(dataSet1Id, nameof(DummyTagCacheEntity.Get));
            if (string.IsNullOrEmpty(tagResult))
            {
                await using (await context.Entities.LockEntitiesAsync(dataSet1Id))
                {
                    tagResult = await context.CallActivityAsync<string>(nameof(DummyTagReport), tag);
                    await context.Entities.CallEntityAsync(dataSet1Id, nameof(DummyTagCacheEntity.Set), tagResult);
                }
            }
            return new(tag, tagResult);
        }

[mllab-nl]

Hello @andystaples !
Thank you for the confirmation, proper example and the fix!

I was trying to find the info on what you can fan out/in, but have not found it. Most likely I have missed it ..
We went with a simpler and more performant approach of making the activities multistep idempotent.
In our case activities have properties that allow as to implement the behavior in them, efficiently.

It is nice to have the general solution and no null references! 😊
If you find a refence in the docs on this would be nice to add it here for future reference.

Thx!

[andystaples]

I think the best reference doc for this is here: https://learn.microsoft.com/en-us/azure/azure-functions/durable/durable-functions-code-constraints?tabs=csharp#async-apis where we explain that orchestrators should not await any Task except Durable SDK APIs. We don't explicitly mention composite tasks being disallowed, if there is some additional clarity in the language you feel would help there, feedback is welcome.