Modifies WorkItem renew logic to start running after item is fetched. #1168

sig5 · 2024-09-23T18:47:43Z

Fixes #1150
The summary of changes is as follows:

Adds a WorkItemBase to extract common properties for Work Items.
Pulls the logic to renew work item lock in WorkItemDispatcher.

Open Question:
Do we want to add a renewal policy for Tracking store items too? If not, why?

sig5 · 2024-09-23T18:48:32Z

@microsoft-github-policy-service agree company="Microsoft"

sig5 · 2024-10-04T10:12:16Z

Bumping this up for review :)

davidmrdavid

Sorry for the delay. Left some notes, overall the work makes sense based on what I had originally suggested here, but now that I'm reviewing the code and looking at the logic more closely, I don't think this really gets to the root problem. Sorry, I think I misunderstood the original problem.

Let me explain. In WorkItemDispatcher.DispatchAsync, we have we in line 227 the logic for ensuring we don't exceed our concurrency limits:

if (!await this.concurrencyLock.WaitAsync(TimeSpan.FromSeconds(5)))

In line 262 we have the FetchWorkItem invocation. At that point ,we have already guaranteed that we're not going over the concurrency limit, so it doesn't make a big difference whether we start renewing there, or a bit later in ProcessWorkItemAsync. In both cases, we have already satisfied the check in line 227 (the .WaitAsync). So it's "too late".

I think the issue is the FetchWorkItem async is simply pulling of messages already read from the queues, and it is those messages that may have already expired (because they weren't getting renewe until after we call FetchWorkItem). So it feels to me that the solution really is to guarantee the renewing is ocurring as soon as messages are obtained from storage, in GetMessageAsync here:

durabletask/src/DurableTask.AzureStorage/Storage/Queue.cs

Lines 88 to 99 in 2795981

    
           public async Task<QueueMessage?> GetMessageAsync(TimeSpan visibilityTimeout, CancellationToken cancellationToken = default) 
        
           { 
        
               QueueMessage message = await this.queueClient.ReceiveMessageAsync(visibilityTimeout, cancellationToken).DecorateFailure(); 
        
               if (message == null) 
        
               { 
        
                   return null; 
        
               } 
        
               this.stats.MessagesRead.Increment(); 
        
               return message; 
        
           }

Please let me know if this doesn't make sense, I'm happy to clarify. In any case, I need to think about the design a bit more, if it ends up being too complicated, I may not want to tackle this edge case just yet. I'm happy to hear your ideas if the problem makes sense though, please let me know

davidmrdavid · 2024-10-21T17:43:24Z

UpgradeLog.htm

+<!DOCTYPE html>
+<!-- saved from url=(0014)about:internet -->


Hmm, what is this file?
In any case, can we please delete it. Doesn't seem to be part of our codebase. I must admit I'm not super comfortable with the strangely encoded URIs referenced in here, but they may be benign.

davidmrdavid · 2024-10-21T17:46:41Z

src/DurableTask.Core/TaskOrchestrationWorkItem.cs

-        /// <summary>
-        /// The datetime this orchestration work item is locked until
-        /// </summary>
-        public DateTime LockedUntilUtc;


Since this is a public class, I don't think we can remove types here without breaking our dependent packages, and possibly some customers. We can choose to deprecate some of these fields, but not remove them.

nevermind, I see this was added to WorkItemBase

davidmrdavid · 2024-10-21T17:48:56Z

src/DurableTask.Core/TrackingWorkItem.cs

-        /// <summary>
-        /// The datetime this work item is locked until
-        /// </summary>
-        public DateTime LockedUntilUtc;


same note as before - we should be careful not to remove fields since this is a public class, that would be a breaking change

nevermind, I see this was added to WorkItemBase

davidmrdavid · 2024-10-21T18:16:14Z

src/DurableTask.ServiceBus/ServiceBusOrchestrationService.cs

@@ -221,7 +221,8 @@ public ServiceBusOrchestrationService(
                    "TrackingDispatcher",
                    item => item == null ? string.Empty : item.InstanceId,
                    FetchTrackingWorkItemAsync,
-                    ProcessTrackingWorkItemAsync)
+                    ProcessTrackingWorkItemAsync,
+                    (T,token) => Task.CompletedTask)


can you please explain why this is the implementation for ServiceBus? Not sure I follow why we can just return Task.CompletedTask

davidmrdavid · 2024-10-21T18:17:19Z

src/DurableTask.Core/TaskOrchestrationDispatcher.cs

@@ -635,6 +613,7 @@ static OrchestrationExecutionContext GetOrchestrationExecutionContext(Orchestrat
        static TimeSpan MinRenewalInterval = TimeSpan.FromSeconds(5); // prevents excessive retries if clocks are off
        static TimeSpan MaxRenewalInterval = TimeSpan.FromSeconds(30);



please remove this newline

davidmrdavid · 2024-10-21T18:18:27Z

src/DurableTask.Core/WorkItemDispatcher.cs

            Func<T, string> workItemIdentifier,
            Func<TimeSpan, CancellationToken, Task<T>> fetchWorkItem,
-            Func<T, Task> processWorkItem)
+            Func<T, Task> processWorkItem,
+            Func<T, CancellationToken, Task> renewWorkItem)


since this is a public class, we should avoid adding a new manadatory parameter, that will break existing implementations. Please make this optional, or create another constructor

sig5 added 2 commits September 23, 2024 18:54

Move renewal hook to WorkItemDispatcher

0631cb6

nit

3783680

sig5 added 3 commits September 24, 2024 13:36

Add additional cancellation condition pn failure.

20f1256

undo change

c7b401a

fixes Azure#1150

e5e3161

sig5 marked this pull request as ready for review September 24, 2024 12:41

sig5 mentioned this pull request Sep 24, 2024

Consider starting the renewal of messages in DTFx.Core as soon as they are fetched #1150

Open

davidmrdavid reviewed Oct 21, 2024

View reviewed changes

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Modifies WorkItem renew logic to start running after item is fetched. #1168

Modifies WorkItem renew logic to start running after item is fetched. #1168

sig5 commented Sep 23, 2024 •

edited

Loading

sig5 commented Sep 23, 2024

sig5 commented Oct 4, 2024

davidmrdavid left a comment •

edited

Loading

davidmrdavid Oct 21, 2024

davidmrdavid Oct 21, 2024

davidmrdavid Oct 21, 2024

davidmrdavid Oct 21, 2024

davidmrdavid Oct 21, 2024

davidmrdavid Oct 21, 2024

davidmrdavid Oct 21, 2024

davidmrdavid Oct 21, 2024

	public async Task<QueueMessage?> GetMessageAsync(TimeSpan visibilityTimeout, CancellationToken cancellationToken = default)
	{
	QueueMessage message = await this.queueClient.ReceiveMessageAsync(visibilityTimeout, cancellationToken).DecorateFailure();

	if (message == null)
	{
	return null;
	}

	this.stats.MessagesRead.Increment();
	return message;
	}

		@@ -635,6 +613,7 @@ static OrchestrationExecutionContext GetOrchestrationExecutionContext(Orchestrat
		static TimeSpan MinRenewalInterval = TimeSpan.FromSeconds(5); // prevents excessive retries if clocks are off
		static TimeSpan MaxRenewalInterval = TimeSpan.FromSeconds(30);

Modifies WorkItem renew logic to start running after item is fetched. #1168

Are you sure you want to change the base?

Modifies WorkItem renew logic to start running after item is fetched. #1168

Conversation

sig5 commented Sep 23, 2024 • edited Loading

sig5 commented Sep 23, 2024

sig5 commented Oct 4, 2024

davidmrdavid left a comment • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

sig5 commented Sep 23, 2024 •

edited

Loading

davidmrdavid left a comment •

edited

Loading