Handle DTFx Orchestration Failure send Message to Dead Letter Queue

[mariohdez]

Hi team,

I am hosting my durable task framework solution through webjob host. I am using the webjob's Service Bus Trigger to start the durable orchestration.

The scenario I would like to support is: have the service bus trigger handle the service bus message, and start the durable task framework orchestration. If orchestration fails, then dead letter the original service bus message.

I am calling the children activities of the orchestration with CallActivityWithRetryAsync. However, if retries are exhausted due to unforeseen issue, I am catching the FunctionFailedActivity exception. In this catch block, I am calling another failure activity. Here, I would like to dead letter queue the original service bus message to have observability to when fatal errors occur in the application level.

I tried to pass the MessageReceiver to the Ochestration's RunAsync function, but get a bind exception. I was wondering if this scenario is supported, or other recommendation workarounds/solutions?

[cgillum]

Let's move this issue into the Azure/azure-functions-durable-extension repo...

[mariohdez]

Thanks @cgillum. Please let me know what other information is needed on my side.

I ran into this stack overflow post where others had similar requirements but I don't think anyone found a good answer for how to achieve application-level dead lettering of the message that triggered the durable orchestration.

https://stackoverflow.com/questions/61279051/handling-service-bus-messages-in-durable-functions

the accepted answer to the stack overflow post was to pass the message receiver to the orchestration function, but as I mentioned earlier, when I did that I got a webjobs bind exception.

i am using versions:

    <PackageReference Include="Microsoft.Azure.WebJobs.Core" Version="3.0.27" />
    <PackageReference Include="Microsoft.Azure.WebJobs.Extensions" Version="4.0.1" />
    <PackageReference Include="Microsoft.Azure.WebJobs.Extensions.DurableTask" Version="2.5.0" />
    <PackageReference Include="Microsoft.Azure.WebJobs.Extensions.DurableTask.Analyzers" Version="0.4.0" />
    <PackageReference Include="Microsoft.Azure.WebJobs.Extensions.ServiceBus" Version="4.3.0" />
    <PackageReference Include="Microsoft.Azure.WebJobs.Extensions.Storage" Version="4.0.4" />
    <PackageReference Include="Microsoft.Azure.WebJobs.Script.ExtensionsMetadataGenerator" Version="1.2.3" />
    <PackageReference Include="Microsoft.Extensions.Hosting" Version="5.0.0" />
    <PackageReference Include="Microsoft.Azure.ServiceBus" Version="5.1.3" />

The current hack I am doing is creating a new service bus message with the same properties as the original, with a TTL of 1 millisecond such that the message gets automatically dead lettered by the service bus queue's TTL property.

[ConnorMcMahon]

In general, I think you have a few options:

1. Have the service-bus triggered webjob/function be responsible for keeping track of orchestration status and deadlettering the original queue message. This would only work if your orchestration has lower latency than the TTL of your service bus message.
2. Within an activity in your orchestrator, you could deadletter the message using MessageReceiver. As noted on the stack overflow message, you can't pass around the MessageReceiver directly, as Durable Task Framework requires all orchestrations/activity inputs and outputs to be serialized/deserialized. Instead, you would likely want to reconstruct the MessageReceiver inside of your activity function (or you could in theory use a static MessageReceiver on a per worker basis, but that is potentially risky). I don't know enough about the ServiceBus SDKs to know if reconstructing a new MessageReceiver and using that to deadletter a message received from a different MessageReceiver works, but you could try to proof of concept this.

[ConnorMcMahon]

Hmm, how long is your lock interval? In general, I'm not the expert on how the ServiceBus trigger works, and Durable Task should not have any impact on this. That being said, I would think ServiceBus would make it impossible for the webjobs host to reprocess the service bus trigger until the lock expires?

[mariohdez]

Hmmm. I see your point and I thought the same... but I guess what I'm experiencing is conflicting with my assumption.

my lock token is set at 30 secs.

my polling method is pretty much the same as the http poll you pointed me to.
Calling the polling method with timeout of 1 hr, retry interval of 20 seconds. the retry interval is 10 seconds less than the lock expiration (30secs) to allow for latency of the network call to get the status of the durable job + the renew lock token request.

        private static async Task<string> HandleServiceBusMessageLifecycle(IDurableOrchestrationClient client, string instanceId, MessageReceiver messageReceiver, TimeSpan timeout, TimeSpan retryInterval, Message message)
        {
            if (retryInterval > timeout)
            {
                throw new ArgumentException($"Total timeout {timeout.TotalSeconds} should be bigger than retry timeout {retryInterval.TotalSeconds}");
            }

            // Rehydrate the lock.
            await messageReceiver.RenewLockAsync(message).ConfigureAwait(false);

            var stopwatch = Stopwatch.StartNew();
            DurableOrchestrationStatus status;

            while (true)
            {
                 status = await client.GetStatusAsync(instanceId).ConfigureAwait(false);

                if (status != null)
                {
                    if (status.RuntimeStatus == OrchestrationRuntimeStatus.Completed)
                    {
                        await messageReceiver.CompleteAsync(message.SystemProperties.LockToken).ConfigureAwait(false);
                        return await Task.FromResult(string.Empty).ConfigureAwait(false);
                    }

                    if (status.RuntimeStatus == OrchestrationRuntimeStatus.Canceled ||
                        status.RuntimeStatus == OrchestrationRuntimeStatus.Failed ||
                        status.RuntimeStatus == OrchestrationRuntimeStatus.Terminated)
                    {
                        await messageReceiver.AbandonAsync(message.SystemProperties.LockToken).ConfigureAwait(false);
                        return await Task.FromResult(string.Empty).ConfigureAwait(false);
                    }
                }

                TimeSpan elapsed = stopwatch.Elapsed;
                if (elapsed < timeout)
                {
                    TimeSpan remainingTime = timeout.Subtract(elapsed);
                    await Task.Delay(remainingTime > retryInterval ? retryInterval : remainingTime).ConfigureAwait(false);
                }
                else
                {
                    await messageReceiver.AbandonAsync(message.SystemProperties.LockToken).ConfigureAwait(false);
                    return await Task.FromResult(string.Empty).ConfigureAwait(false);
                }
            }
        }

[mariohdez]

Ohh shoot. I think I know what's going on.

I think messageReceiver.AbandonAsync frees the message to be picked up by another receiver.

I think I should inject the queueclient or something else and use the abandon async method such that is shows I've acknowledged the message and then can say i can't process the message now go to DLQ.

[mariohdez]

Okay 1 final question @ConnorMcMahon :D

So I have configured the polling, and to call await messageReceiver.DeadLetterAsync(lockToken).ConfigureAwait(false); when the runtime status is not completed.

However, I am seeing a continued retry of my dead letter async operation:

info: DurableTask.AzureStorage[136]
      Fetched instance status for 06842e8615bf4aa6a35a04920ce7c219
warn: Microsoft.Azure.Squall.ExperimentManagementService.DeleteWorker.Triggers.DeleteExperimentServiceBusTriggerFunction[0]
      seding message with id: 0d3495555c4a49adb0a65d7f1f24b09b to the DLQ.
fail: Microsoft.Azure.Squall.ExperimentManagementService.DeleteWorker.Triggers.DeleteExperimentServiceBusTriggerFunction[0]
      we are seeing this exception: $Microsoft.Azure.ServiceBus.MessageLockLostException: The lock supplied is invalid. Either the lock expired, or the message has already been removed from the queue, or was received by a different receiver instance.
         at Microsoft.Azure.ServiceBus.Core.MessageReceiver.DisposeMessagesAsync(IEnumerable`1 lockTokens, Outcome outcome)
         at Microsoft.Azure.ServiceBus.RetryPolicy.RunOperation(Func`1 operation, TimeSpan operationTimeout)
         at Microsoft.Azure.ServiceBus.RetryPolicy.RunOperation(Func`1 operation, TimeSpan operationTimeout)
         at Microsoft.Azure.ServiceBus.Core.MessageReceiver.DeadLetterAsync(String lockToken, IDictionary`2 propertiesToModify)
         at Microsoft.Azure.Squall.ExperimentManagementService.DeleteWorker.Triggers.DeleteExperimentServiceBusTriggerFunction.HandleServiceBusMessageLifecycle(IDurableOrchestrationClient client, String instanceId, MessageReceiver messageReceiver, TimeSpan timeout, TimeSpan retryInterval, Message message, String lockToken) in C:\Users\marher\source\repos\Experiment-Management-Service\src\ExperimentManagementService\ExperimentManagementService.DeleteWorker\Triggers\DeleteExperimentServiceBusTriggerFunction.cs:line 192

It is important to note that the message does get sent to the DLQ but the lock token expiration exception gets repeatedly thrown.

any ideas here?

here are my service bus trigger options:

info: Microsoft.Azure.WebJobs.Hosting.OptionsLoggingService[0]
      ServiceBusOptions
      {
        "PrefetchCount": 0,
        "MessageHandlerOptions": {
          "AutoComplete": false,
          "MaxAutoRenewDuration": "00:05:00",
          "MaxConcurrentCalls": 128
        },
        "SessionHandlerOptions": {
          "AutoComplete": true,
          "MaxAutoRenewDuration": "00:05:00",
          "MaxConcurrentSessions": 2000,
          "MessageWaitTimeout": "00:01:00"
        },
        "BatchOptions": {
          "MaxMessageCount": 1000,
          "OperationTimeout": "00:01:00",
          "AutoComplete": true
        }

[mariohdez]

In addition, please note I am hosting this through webjobs and running this on a netcore 3.1 worker service solution.

[olitomlinson]

I’ve commented on this exact same thing, but with an alternative solution.

My recommendation is still the same, don’t try to defer completion of the SB message once it’s been passed to the orchestration.

#1670 (comment)

[olitomlinson]

@ConnorMcMahon

it seems quite simple to me.

Durable Functions becomes a recognised “target resource” and emits various “signals” - one of which could be an “orchestration failed” or “orchestration failed count”

https://docs.microsoft.com/en-us/azure/azure-monitor/alerts/alerts-overview

[mariohdez]

Hello, just wanted to update this thread that I was never able to get this 'polling' of the job to work quite right... Kept getting exceptions related the SB message's lock being invalid once the orchestration job had reached a terminal state.

Just wanted to reiterate that the idea of being able to dead letter a message due to application level errors in DTFx would add great value. My vision was that, if there was ever an infra outage I would like to dead letter the messages. Then, I could write a tool for our on call engineers to "drain" the dead letter queue, and re-queue the message so that once our system is healthy again we could have a no-touch experience.

Unfortunately, I had to deliver my work with emitting a custom metric that we will monitor and fire an alert if a message was unable to be processed by DTFx orchestration. The way we hunt down what the payload the message contained is through sifting through logs.

Are there any plans to allow for this in the future @cgillum and @ConnorMcMahon and @olitomlinson? Thanks!

[olitomlinson]

@mariohdez

Then, I could write a tool for our on call engineers to "drain" the dead letter queue, and re-queue the message so that once our system is healthy again we could have a no-touch experience.

this is the same principle to what I did, but without using SB DLQ.

I used lifecycle hooks to write the failed orchestration IDs to a separate ‘failed’ table.

then I made a little .exe utility that read from that failed table and used the Restart API for every orchestration in the table, and cleaned up the row after restart. This was used when triaging alerts during the day and on-call.

I also extended the utility to do other things like dump out the ‘input’ payloads of any failed orchestrations into a Json file. From here I could modify the input Params, and using another switch on the utility I could start 1 or more new orchestrations using that modified json as an input.

All the utility was doing was orchestrating over the various HTTP Management APIs that come with DF. No particular wizardry.

The hardest bit was educating on-call on how to use the utility correctly.

[mariohdez]

@olitomlinson I would love to pick your brain on how to leverage the lifecycle hooks on the orchestration.

Would you mind pasting some code snippets or links to help me get an understanding on how you're achieving this?

[olitomlinson]

@mariohdez hey!

Unfortunately I know longer work there so I don’t have access to the source.

But this is the doc that I used to hook up the orchestration lifecycle hooks to EventGrid.

https://docs.microsoft.com/en-us/azure/azure-functions/durable/durable-functions-event-publishing?tabs=csharp-script

I think built an EventGrid trigger (into the same Function App that used DF) to receive those events. The Event contains the Orchestration ID, which you can then use in order to retrieve the input payloads for that particular orchestration and store those payloads for restarting or modifying.

If you only want to restart those failed orchestrations (and not modify the input payload) it becomes a heck of a lot simpler as you don’t need to store the input payload anywhere, you can just hit the built-in “restart” management API endpoint.

#1545

I wrote any failed Orchestration IDs to a ‘failed’ table. I then wrote a Timer triggered function that polled that ‘failed’ table and counted the amount of rows in that table. This count was then published to Azure Monitor, each minute.

I then set up an Azure Monitor rule to fire an alert if that count ever became greater than zero. This would then alert me and the on-call team via the usual action groups in Azure Monitor.