From 24956082fed85f8a4a9cdd30cfe6caaccc93cb6c Mon Sep 17 00:00:00 2001 From: wangbill Date: Thu, 3 Sep 2026 14:43:02 -0700 Subject: [PATCH 1/3] Fix external event loss when canceled/abandoned waiter is only listener (#1676) Root cause: TaskOrchestrationContextWrapper.CompleteExternalEvent popped the top-of-stack waiter and called TrySetResult unconditionally, ignoring that a canceled/abandoned waiter's TrySetResult is a no-op that returns false. The event was still treated as consumed, so it was neither buffered nor forwarded to the next ContinueAsNew generation, resulting in silent event loss. Fix: IEventSource.TrySetResult now returns bool. CompleteExternalEvent walks the LIFO waiter stack, skipping dead waiters until a live one accepts the event; if none accept it, falls through to existing buffer/forward-on- ContinueAsNew logic. LIFO ordering, payload/entity deserialization, and late-event forwarding semantics are preserved. Added regression tests covering: single canceled waiter, canceled waiter with ContinueAsNew(preserveUnprocessedEvents:true) already scheduled, multiple consecutive canceled waiters, and a live waiter above a canceled one. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- ...OrchestrationContextWrapper.EventSource.cs | 8 +- .../Shims/TaskOrchestrationContextWrapper.cs | 21 +++- .../TaskOrchestrationContextWrapperTests.cs | 111 ++++++++++++++++++ 3 files changed, 136 insertions(+), 4 deletions(-) diff --git a/src/Worker/Core/Shims/TaskOrchestrationContextWrapper.EventSource.cs b/src/Worker/Core/Shims/TaskOrchestrationContextWrapper.EventSource.cs index 9e2e9c6a..69e81adf 100644 --- a/src/Worker/Core/Shims/TaskOrchestrationContextWrapper.EventSource.cs +++ b/src/Worker/Core/Shims/TaskOrchestrationContextWrapper.EventSource.cs @@ -30,7 +30,11 @@ interface IEventSource /// Tries to set the result on tcs. /// /// The result. - void TrySetResult(object result); + /// + /// true if the result was set successfully, meaning this waiter is a live consumer of the event; + /// false if the waiter was already completed, canceled, or abandoned and cannot accept the result. + /// + bool TrySetResult(object result); } class EventTaskCompletionSource : TaskCompletionSource, IEventSource @@ -42,7 +46,7 @@ class EventTaskCompletionSource : TaskCompletionSource, IEventSource public IEventSource? Next { get; set; } /// - void IEventSource.TrySetResult(object result) => this.TrySetResult((T)result); + bool IEventSource.TrySetResult(object result) => this.TrySetResult((T)result); } class NamedQueue diff --git a/src/Worker/Core/Shims/TaskOrchestrationContextWrapper.cs b/src/Worker/Core/Shims/TaskOrchestrationContextWrapper.cs index c92ee5d6..962aa05d 100644 --- a/src/Worker/Core/Shims/TaskOrchestrationContextWrapper.cs +++ b/src/Worker/Core/Shims/TaskOrchestrationContextWrapper.cs @@ -473,7 +473,7 @@ internal void ExitCriticalSectionIfNeeded() /// The serialized event payload. internal void CompleteExternalEvent(string eventName, string rawEventPayload) { - if (this.externalEventSources.TryGetValue(eventName, out IEventSource? waiter)) + while (this.externalEventSources.TryGetValue(eventName, out IEventSource? waiter)) { // Get the waiter at the top of the stack (most recent waiter) // If we're going to raise an event we should remove it from the pending collection @@ -501,7 +501,24 @@ internal void CompleteExternalEvent(string eventName, string rawEventPayload) value = this.DataConverter.Deserialize(rawEventPayload, waiter.EventType); } - waiter.TrySetResult(value); + if (waiter.TrySetResult(value)) + { + // The event was delivered to a live waiter. We're done. + return; + } + + // This waiter was already completed, canceled, or abandoned (e.g. the losing side of a + // Task.WhenAny) and cannot accept the event. It has already been popped off the stack above, + // so continue the loop to try the next waiter underneath it, if any. If there are no more + // waiters, fall through to the buffering/forwarding logic below as if no one was listening. + } + + if (this.preserveUnprocessedEventsOnContinueAsNew) + { + // ContinueAsNew has already been scheduled with event preservation enabled. + // Forward late-arriving events directly to the next execution instead of buffering + // them on the current wrapper instance, which is about to be discarded. + this.ForwardRawExternalEvent(eventName, rawEventPayload); } else { diff --git a/test/Worker/Core.Tests/Shims/TaskOrchestrationContextWrapperTests.cs b/test/Worker/Core.Tests/Shims/TaskOrchestrationContextWrapperTests.cs index e8335e79..d5c574a9 100644 --- a/test/Worker/Core.Tests/Shims/TaskOrchestrationContextWrapperTests.cs +++ b/test/Worker/Core.Tests/Shims/TaskOrchestrationContextWrapperTests.cs @@ -391,6 +391,117 @@ await wrapper.CallSubOrchestratorAsync( innerContext.LastSubOrchestrationVersion.Should().Be(string.Empty); } + [Fact] + public void CompleteExternalEvent_CanceledWaiter_BuffersEventForNextWaiter() + { + // Arrange: reproduces the sequence from Azure/azure-functions-durable-extension#1676. + // Two differently-named external event waits are registered with a shared cancellation + // token, similar to a Task.WhenAny loop. The waiter for "event_1" is canceled (e.g. + // because "event_0" won the race), but "event_1" is later raised while the canceled + // waiter is still the only registered listener for that name. + TrackingOrchestrationContext innerContext = new(); + OrchestrationInvocationContext invocationContext = new("Test", new(), NullLoggerFactory.Instance, null); + TaskOrchestrationContextWrapper wrapper = new(innerContext, invocationContext, "input"); + + using CancellationTokenSource cts = new(); + Task canceledWait = wrapper.WaitForExternalEvent("event_1", cts.Token); + cts.Cancel(); + + // Act: the event arrives after the only waiter for it has been canceled/abandoned. + InvokeCompleteExternalEvent(wrapper, "event_1", "\"payload\""); + + // Assert: the canceled waiter must not "consume" the event. It should be buffered so + // that a subsequent WaitForExternalEvent call (e.g. in the next ContinueAsNew generation) + // can still observe it. + canceledWait.IsCanceled.Should().BeTrue(); + Task nextWait = wrapper.WaitForExternalEvent("event_1"); + nextWait.IsCompletedSuccessfully.Should().BeTrue(); + nextWait.Result.Should().Be("payload"); + } + + [Fact] + public void CompleteExternalEvent_CanceledWaiter_WithContinueAsNewPreserved_ForwardsEventToNextExecution() + { + // Arrange: same as above, but the event arrives after ContinueAsNew(preserveUnprocessedEvents: true) + // has already been called (e.g. while an activity scheduled before ContinueAsNew is still + // completing). The event should be forwarded to the next execution, not silently dropped. + TrackingOrchestrationContext innerContext = new(); + OrchestrationInvocationContext invocationContext = new("Test", new(), NullLoggerFactory.Instance, null); + TaskOrchestrationContextWrapper wrapper = new(innerContext, invocationContext, "input"); + + using CancellationTokenSource cts = new(); + Task canceledWait = wrapper.WaitForExternalEvent("event_1", cts.Token); + cts.Cancel(); + + wrapper.ContinueAsNew("new-input", preserveUnprocessedEvents: true); + + // Act + InvokeCompleteExternalEvent(wrapper, "event_1", "\"payload\""); + + // Assert + canceledWait.IsCanceled.Should().BeTrue(); + innerContext.SentEvents.Should().ContainSingle(); + innerContext.SentEvents[0].InstanceId.Should().Be(wrapper.InstanceId); + innerContext.SentEvents[0].EventName.Should().Be("event_1"); + innerContext.SentEvents[0].EventData.Should().BeOfType().Which.Value.Should().Be("\"payload\""); + } + + [Fact] + public void CompleteExternalEvent_MultipleCanceledWaiters_BuffersEventAfterSkippingAll() + { + // Arrange: several waiters for the same event name are canceled/abandoned in a row + // (e.g. a loop that repeatedly races and abandons the same event). None of them should + // be able to consume the event. + TrackingOrchestrationContext innerContext = new(); + OrchestrationInvocationContext invocationContext = new("Test", new(), NullLoggerFactory.Instance, null); + TaskOrchestrationContextWrapper wrapper = new(innerContext, invocationContext, "input"); + + using CancellationTokenSource cts1 = new(); + using CancellationTokenSource cts2 = new(); + using CancellationTokenSource cts3 = new(); + Task wait1 = wrapper.WaitForExternalEvent("event_1", cts1.Token); + Task wait2 = wrapper.WaitForExternalEvent("event_1", cts2.Token); + Task wait3 = wrapper.WaitForExternalEvent("event_1", cts3.Token); + cts1.Cancel(); + cts2.Cancel(); + cts3.Cancel(); + + // Act + InvokeCompleteExternalEvent(wrapper, "event_1", "\"payload\""); + + // Assert + wait1.IsCanceled.Should().BeTrue(); + wait2.IsCanceled.Should().BeTrue(); + wait3.IsCanceled.Should().BeTrue(); + Task nextWait = wrapper.WaitForExternalEvent("event_1"); + nextWait.IsCompletedSuccessfully.Should().BeTrue(); + nextWait.Result.Should().Be("payload"); + } + + [Fact] + public void CompleteExternalEvent_ActiveWaiterAboveCanceledWaiter_DeliversToActiveWaiter() + { + // Arrange: the most recent (top-of-stack) waiter is still live, but an older waiter + // underneath it for the same event name was already canceled. The event should go to + // the live waiter, and the canceled one underneath should remain untouched (not consumed). + TrackingOrchestrationContext innerContext = new(); + OrchestrationInvocationContext invocationContext = new("Test", new(), NullLoggerFactory.Instance, null); + TaskOrchestrationContextWrapper wrapper = new(innerContext, invocationContext, "input"); + + using CancellationTokenSource cts1 = new(); + Task canceledWait = wrapper.WaitForExternalEvent("event_1", cts1.Token); + cts1.Cancel(); + Task activeWait = wrapper.WaitForExternalEvent("event_1"); + + // Act + InvokeCompleteExternalEvent(wrapper, "event_1", "\"payload\""); + + // Assert + canceledWait.IsCanceled.Should().BeTrue(); + activeWait.IsCompletedSuccessfully.Should().BeTrue(); + activeWait.Result.Should().Be("payload"); + } + static IReadOnlyDictionary GetLastScheduledTaskTags(TrackingOrchestrationContext innerContext) { PropertyInfo tagsProperty = innerContext.LastScheduledTaskOptions!.GetType().GetProperty("Tags")!; From d9d5a478c7c408e41a99d95c9e1a9488c51bc0e5 Mon Sep 17 00:00:00 2001 From: wangbill Date: Fri, 4 Sep 2026 10:02:39 -0700 Subject: [PATCH 2/3] Remove duplicate external event conditional Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- .../Shims/TaskOrchestrationContextWrapper.cs | 16 +++------------- 1 file changed, 3 insertions(+), 13 deletions(-) diff --git a/src/Worker/Core/Shims/TaskOrchestrationContextWrapper.cs b/src/Worker/Core/Shims/TaskOrchestrationContextWrapper.cs index 962aa05d..f712964e 100644 --- a/src/Worker/Core/Shims/TaskOrchestrationContextWrapper.cs +++ b/src/Worker/Core/Shims/TaskOrchestrationContextWrapper.cs @@ -522,19 +522,9 @@ internal void CompleteExternalEvent(string eventName, string rawEventPayload) } else { - if (this.preserveUnprocessedEventsOnContinueAsNew) - { - // ContinueAsNew has already been scheduled with event preservation enabled. - // Forward late-arriving events directly to the next execution instead of buffering - // them on the current wrapper instance, which is about to be discarded. - this.ForwardRawExternalEvent(eventName, rawEventPayload); - } - else - { - // The orchestrator isn't waiting for this event (yet?). Save it in case - // the orchestrator wants it later. - this.externalEventBuffer.Add(eventName, rawEventPayload); - } + // The orchestrator isn't waiting for this event (yet?). Save it in case + // the orchestrator wants it later. + this.externalEventBuffer.Add(eventName, rawEventPayload); } } From 14b175eadd6924b6fbd26f8f7cb2707ac8da60bb Mon Sep 17 00:00:00 2001 From: wangbill Date: Fri, 4 Sep 2026 10:59:27 -0700 Subject: [PATCH 3/3] Optimize external event deserialization Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> Copilot-Session: e86c9d9a-883b-4d06-a9df-f8f991c132a9 --- ...OrchestrationContextWrapper.EventSource.cs | 6 ++--- .../Shims/TaskOrchestrationContextWrapper.cs | 24 ++++++++++++------- .../TaskOrchestrationContextWrapperTests.cs | 21 +++++++++++++++- 3 files changed, 38 insertions(+), 13 deletions(-) diff --git a/src/Worker/Core/Shims/TaskOrchestrationContextWrapper.EventSource.cs b/src/Worker/Core/Shims/TaskOrchestrationContextWrapper.EventSource.cs index 69e81adf..b93657c9 100644 --- a/src/Worker/Core/Shims/TaskOrchestrationContextWrapper.EventSource.cs +++ b/src/Worker/Core/Shims/TaskOrchestrationContextWrapper.EventSource.cs @@ -34,10 +34,10 @@ interface IEventSource /// true if the result was set successfully, meaning this waiter is a live consumer of the event; /// false if the waiter was already completed, canceled, or abandoned and cannot accept the result. /// - bool TrySetResult(object result); + bool TrySetResult(object? result); } - class EventTaskCompletionSource : TaskCompletionSource, IEventSource + sealed class EventTaskCompletionSource : TaskCompletionSource, IEventSource { /// public Type EventType => typeof(T); @@ -46,7 +46,7 @@ class EventTaskCompletionSource : TaskCompletionSource, IEventSource public IEventSource? Next { get; set; } /// - bool IEventSource.TrySetResult(object result) => this.TrySetResult((T)result); + bool IEventSource.TrySetResult(object? result) => this.TrySetResult((T)result!); } class NamedQueue diff --git a/src/Worker/Core/Shims/TaskOrchestrationContextWrapper.cs b/src/Worker/Core/Shims/TaskOrchestrationContextWrapper.cs index f712964e..b0d65bd2 100644 --- a/src/Worker/Core/Shims/TaskOrchestrationContextWrapper.cs +++ b/src/Worker/Core/Shims/TaskOrchestrationContextWrapper.cs @@ -473,6 +473,7 @@ internal void ExitCriticalSectionIfNeeded() /// The serialized event payload. internal void CompleteExternalEvent(string eventName, string rawEventPayload) { + Dictionary? deserializedValues = null; while (this.externalEventSources.TryGetValue(eventName, out IEventSource? waiter)) { // Get the waiter at the top of the stack (most recent waiter) @@ -489,16 +490,21 @@ internal void CompleteExternalEvent(string eventName, string rawEventPayload) this.externalEventSources[eventName] = next; } - object? value; - if (waiter.EventType == typeof(OperationResult)) + deserializedValues ??= new Dictionary(); + if (!deserializedValues.TryGetValue(waiter.EventType, out object? value)) { - // use the framework-defined deserialization for entity responses, not the application-defined data converter, - // because we are just unwrapping the entity response without yet deserializing any application-defined data. - value = this.entityFeature!.EntityContext.DeserializeEntityResponseEvent(rawEventPayload); - } - else - { - value = this.DataConverter.Deserialize(rawEventPayload, waiter.EventType); + if (waiter.EventType == typeof(OperationResult)) + { + // use the framework-defined deserialization for entity responses, not the application-defined data converter, + // because we are just unwrapping the entity response without yet deserializing any application-defined data. + value = this.entityFeature!.EntityContext.DeserializeEntityResponseEvent(rawEventPayload); + } + else + { + value = this.DataConverter.Deserialize(rawEventPayload, waiter.EventType); + } + + deserializedValues[waiter.EventType] = value; } if (waiter.TrySetResult(value)) diff --git a/test/Worker/Core.Tests/Shims/TaskOrchestrationContextWrapperTests.cs b/test/Worker/Core.Tests/Shims/TaskOrchestrationContextWrapperTests.cs index d5c574a9..6854438c 100644 --- a/test/Worker/Core.Tests/Shims/TaskOrchestrationContextWrapperTests.cs +++ b/test/Worker/Core.Tests/Shims/TaskOrchestrationContextWrapperTests.cs @@ -453,7 +453,12 @@ public void CompleteExternalEvent_MultipleCanceledWaiters_BuffersEventAfterSkipp // (e.g. a loop that repeatedly races and abandons the same event). None of them should // be able to consume the event. TrackingOrchestrationContext innerContext = new(); - OrchestrationInvocationContext invocationContext = new("Test", new(), NullLoggerFactory.Instance, null); + CountingDataConverter converter = new(); + OrchestrationInvocationContext invocationContext = new( + "Test", + new DurableTaskWorkerOptions { DataConverter = converter }, + NullLoggerFactory.Instance, + null); TaskOrchestrationContextWrapper wrapper = new(innerContext, invocationContext, "input"); using CancellationTokenSource cts1 = new(); @@ -470,6 +475,7 @@ public void CompleteExternalEvent_MultipleCanceledWaiters_BuffersEventAfterSkipp InvokeCompleteExternalEvent(wrapper, "event_1", "\"payload\""); // Assert + converter.DeserializeCallCounts[typeof(string)].Should().Be(1); wait1.IsCanceled.Should().BeTrue(); wait2.IsCanceled.Should().BeTrue(); wait3.IsCanceled.Should().BeTrue(); @@ -513,6 +519,19 @@ static void InvokeCompleteExternalEvent(TaskOrchestrationContextWrapper wrapper, CompleteExternalEventMethod.Invoke(wrapper, [eventName, rawEventPayload]); } + sealed class CountingDataConverter : DataConverter + { + public Dictionary DeserializeCallCounts { get; } = []; + + public override object? Deserialize(string? data, Type targetType) + { + this.DeserializeCallCounts[targetType] = this.DeserializeCallCounts.GetValueOrDefault(targetType) + 1; + return targetType == typeof(string) ? "payload" : null; + } + + public override string? Serialize(object? value) => throw new NotSupportedException(); + } + sealed class TrackingOrchestrationContext : OrchestrationContext { public TrackingOrchestrationContext(string? version = null)