Dequeue successfully processed queue messages before shutting down in drain mode #50426

zahalzel · 2025-06-04T19:35:06Z

We have a bug in FlexConsumption scenarios where drain mode triggers a cancellation token in CompleteProcessingMessageAsync. As a result, messages that are successfully processed are not dequeued and therefore end up triggering further executions.

A previous attempt to resolve this bug was only partially successful, because the call to put the host in drain mode is still sending a cancellation token that gets passed into the DeleteMessage call. Any time that cancellation token is active, the queue processor will not delete the message, even if the message was successfully processed.

Related discussions / incidents:

https://learn.microsoft.com/en-us/answers/questions/2184443/problems-with-flex-consumption-function
https://portal.microsofticm.com/imp/v5/incidents/details/636224894/summary

This PR removes explicit CancellationToken parameters from several overloads of CompleteProcessingMessageAsync and related queue message operations, updating both production and test code accordingly.

The idea is that any time a message has been processed, we should always make a best-case attempt to dequeue it before shutting down the queue listener. Furthermore, there is not a compelling reason to send cancellation tokens to the calls to delete the message, so we remove the cancellation token parameter altogether.

Removed CancellationToken parameters from calls to CompleteProcessingMessageAsync and associated helper methods in production code.
Updated unit tests and mock setups in QueueProcessorTests.cs and QueueListenerTests.cs to align with the modified API signature.
Introduced new tests to verify cancellation token behavior during shutdown scenarios.

Copilot

Pull Request Overview

This PR removes explicit CancellationToken parameters from several overloads of CompleteProcessingMessageAsync and related queue message operations, updating both production and test code accordingly.

Removed CancellationToken parameters from calls to CompleteProcessingMessageAsync and associated helper methods in production code.
Updated unit tests and mock setups in QueueProcessorTests.cs and QueueListenerTests.cs to align with the modified API signature.
Introduced new tests to verify cancellation token behavior during shutdown scenarios.

Reviewed Changes

Copilot reviewed 4 out of 4 changed files in this pull request and generated 1 comment.

File	Description
sdk/storage/Microsoft.Azure.WebJobs.Extensions.Storage.Queues/tests/QueueProcessorTests.cs	Removed cancellation token arguments from processor method calls in tests.
sdk/storage/Microsoft.Azure.WebJobs.Extensions.Storage.Queues/tests/QueueListenerTests.cs	Adjusted mock setups and added tests for shutdown token behavior.
sdk/storage/Microsoft.Azure.WebJobs.Extensions.Storage.Common/src/Shared/Queues/QueueProcessor.cs	Updated API calls by removing cancellation token parameters in message handling methods.
sdk/storage/Microsoft.Azure.WebJobs.Extensions.Storage.Common/src/Shared/Queues/QueueListener.cs	Refined cancellation token usage and improved log messages in StopAsync and ProcessMessageAsync.

Comments suppressed due to low confidence (1)

sdk/storage/Microsoft.Azure.WebJobs.Extensions.Storage.Common/src/Shared/Queues/QueueProcessor.cs:132

Using CancellationToken.None here instead of a propagating token changes the cancellation behavior during poison message handling. Verify that a best-effort message addition is acceptable in scenarios where cancellation might be desired.

await poisonQueue.AddMessageAndCreateIfNotExistsAsync(message.Body, CancellationToken.None).ConfigureAwait(false);

sdk/storage/Microsoft.Azure.WebJobs.Extensions.Storage.Queues/tests/QueueListenerTests.cs

mathewc · 2025-06-05T16:58:29Z

...storage/Microsoft.Azure.WebJobs.Extensions.Storage.Common/src/Shared/Queues/QueueListener.cs

@@ -113,7 +126,7 @@ public QueueListener(QueueClient queue,

            // if the function runs longer than this, the invisibility will be updated
            // on a timer periodically for the duration of the function execution
-            _visibilityTimeout = TimeSpan.FromMinutes(10);
+            _visibilityTimeout = _queueOptions.VisibilityTimeout;


I don't think you need to make these changes in this PR. I agree this is somewhat confusing but is discussed more in detail in Azure/azure-webjobs-sdk#1040. Summary is, that the configured QueuesOptions.VisibilityTimeout IS used when messages fail processing (see code). This hardcoded value is just the initial timeout used when dequeuing messages. If we want to make any changes to this, I think we should discuss and do that separately, because it's not really related to the cancellation token changes right?

That's fair, it is indeed a separate scenario, I'll remove from this PR :)

But I do appreciate the attempt to fix a perceived issue :)

…tests/QueueListenerTests.cs Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>

mathewc · 2025-06-05T17:31:19Z

Please update this PR with a more descriptive title than just "updates", and also provide some context on why the changes are being made. I know there's an internal teams thread with the details, but people on the Azure SDK team who are looking at PRs in this repo won't have that context.

mathewc · 2025-06-05T17:43:19Z

...storage/Microsoft.Azure.WebJobs.Extensions.Storage.Common/src/Shared/Queues/QueueListener.cs

+                   ThrowIfDisposed();
+                   _timer.Cancel();
+                   await Task.WhenAll(_processing).ConfigureAwait(false);
+                   await _timer.StopAsync(cancellationToken).ConfigureAwait(false);
+                   _logger.LogDebug($"Drain mode is not enabled, storage queue listener stopped ({_details})");


This is introducing a lot of code duplication between these two branches. Instead, just isolate the lines that differ in your if/else block, and have the common code outside.

To simplify, probably you can just remove the "using (cancellationToken.Register" - I don't think the shutdown tasks need to be within a using scope

@mathewc How about this refactor? I can also remove the using scope entirely, but this is maybe a good compromise while preserving the existing behavior?

github-actions · 2025-06-05T21:09:34Z

API Change Check

APIView identified API level changes in this PR and created the following API reviews

Microsoft.Azure.WebJobs.Extensions.Storage.Common

mathewc · 2025-06-09T21:55:37Z

...storage/Microsoft.Azure.WebJobs.Extensions.Storage.Common/src/Shared/Queues/QueueListener.cs

            {
-                // Cancel the execution token when drain mode is not enabled or drain mode manager is not set.
+                _logger.LogDebug($"Drain mode is not enabled, storage queue listener will stop immediately ({_details})");


This "will stop immediately" log isn't really true - all you're doing differently here in this branch is cancelling the token passed into functions. That's just a signal to functions that shutdown is happening, it doesn't forcefully terminate any of the individual invocations. I.e. below the Task.WhenAll will still be gated on all the invocations completing, and that's up to the function implementation.

mathewc · 2025-06-09T22:03:26Z

...torage/Microsoft.Azure.WebJobs.Extensions.Storage.Common/src/Shared/Queues/QueueProcessor.cs

@@ -89,17 +89,17 @@ internal protected virtual async Task CompleteProcessingMessageAsync(QueueMessag
        {
            if (result.Succeeded)
            {
-                await DeleteMessageAsync(message, cancellationToken).ConfigureAwait(false);
+                await DeleteMessageAsync(message, CancellationToken.None).ConfigureAwait(false);


I don't think we should be ignoring the cancellationTokens passed into these methods - it should instead be up to the caller to decide to pass None in for operations that shouldn't be cancelled. I believe you're doing that in the QueueProcessor already, passing CancellationToken.None to CompleteProcessingMessageAsync.

mathewc · 2025-06-09T22:59:48Z

...storage/Microsoft.Azure.WebJobs.Extensions.Storage.Common/src/Shared/Queues/QueueListener.cs

-                    // Only cancel completion or update of the message if a non-graceful shutdown is requested via _shutdownCancellationTokenSource.
-                    await _queueProcessor.CompleteProcessingMessageAsync(message, result, _shutdownCancellationTokenSource.Token).ConfigureAwait(false);
+                    // No cancellation token for message completion: always best-effort to complete
+                    await _queueProcessor.CompleteProcessingMessageAsync(message, result, CancellationToken.None).ConfigureAwait(false);


I do agree with your general assertion that if a message has been processed by the user function, we don't want to cancel the completion actions. E.g. if the method succeeded, we need to delete the processed message from the queue.

However, it looks like _shutdownCancellationTokenSource is only canceled in 2 cases. 1) when the listener is disposed here, and 2) if the listener StopAsync call is cancelled here. Are you saying you think one of these is happening? I don't see how.

Updates

bf8619a

Copilot AI review requested due to automatic review settings June 4, 2025 19:35

zahalzel requested review from seanmcc-msft, amnguye, jaschrep-msft, tg-msft, fabiocav, mathewc, jalauzon-msft and nickliu-msft as code owners June 4, 2025 19:35

github-actions bot added the Storage Storage Service (Queues, Blobs, Files) label Jun 4, 2025

Copilot AI reviewed Jun 4, 2025

View reviewed changes

sdk/storage/Microsoft.Azure.WebJobs.Extensions.Storage.Queues/tests/QueueListenerTests.cs Outdated Show resolved Hide resolved

mathewc reviewed Jun 5, 2025

View reviewed changes

zahalzel and others added 2 commits June 5, 2025 13:12

Revert timeout change

a502ccc

Update sdk/storage/Microsoft.Azure.WebJobs.Extensions.Storage.Queues/…

370bf00

…tests/QueueListenerTests.cs Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>

zahalzel changed the title ~~Updates~~ Dequeue successfully processed queue messages before shutting down in drain mode Jun 5, 2025

Merge branch 'main' into queueCancellation

587cf18

mathewc reviewed Jun 5, 2025

View reviewed changes

zahalzel added 11 commits June 5, 2025 15:27

Update StopAsync()

641d6cc

Fix test

39dabf5

Revert signatures

25afd74

Revert signature

63b88aa

Fix tests

03e7abc

Fix tests

041cac9

Update doc strings

0b98e07

Revert usings

193e25a

Revert tests

ab139a5

Revert signature

b95373c

Fix doc strings

27693b6

nit spacing

a6de18f

mathewc reviewed Jun 9, 2025

View reviewed changes

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Uh oh!

Dequeue successfully processed queue messages before shutting down in drain mode #50426

Dequeue successfully processed queue messages before shutting down in drain mode #50426

Uh oh!

zahalzel commented Jun 4, 2025 •

edited

Loading

Uh oh!

Copilot AI left a comment

Uh oh!

Uh oh!

mathewc Jun 5, 2025

Uh oh!

zahalzel Jun 5, 2025

Uh oh!

mathewc Jun 5, 2025

Uh oh!

mathewc commented Jun 5, 2025

Uh oh!

mathewc Jun 5, 2025

Uh oh!

zahalzel Jun 5, 2025

Uh oh!

github-actions bot commented Jun 5, 2025

Uh oh!

mathewc Jun 9, 2025

Uh oh!

mathewc Jun 9, 2025 •

edited

Loading

Uh oh!

mathewc Jun 9, 2025 •

edited

Loading

Uh oh!

Uh oh!

pFad - (p)hone/(F)rame/(a)nonymizer/(d)eclutterfier! Saves Data!

Dequeue successfully processed queue messages before shutting down in drain mode #50426

Are you sure you want to change the base?

Dequeue successfully processed queue messages before shutting down in drain mode #50426

Uh oh!

Conversation

zahalzel commented Jun 4, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

Pull Request Overview

Reviewed Changes

Uh oh!

Uh oh!

mathewc Jun 5, 2025

Choose a reason for hiding this comment

Uh oh!

zahalzel Jun 5, 2025

Choose a reason for hiding this comment

Uh oh!

mathewc Jun 5, 2025

Choose a reason for hiding this comment

Uh oh!

mathewc commented Jun 5, 2025

Uh oh!

mathewc Jun 5, 2025

Choose a reason for hiding this comment

Uh oh!

zahalzel Jun 5, 2025

Choose a reason for hiding this comment

Uh oh!

github-actions bot commented Jun 5, 2025

API Change Check

Uh oh!

mathewc Jun 9, 2025

Choose a reason for hiding this comment

Uh oh!

mathewc Jun 9, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

mathewc Jun 9, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

pFad - (p)hone/(F)rame/(a)nonymizer/(d)eclutterfier! Saves Data!

zahalzel commented Jun 4, 2025 •

edited

Loading

mathewc Jun 9, 2025 •

edited

Loading

mathewc Jun 9, 2025 •

edited

Loading