From 31a528227536041754b3ad8783e2606578357db8 Mon Sep 17 00:00:00 2001 From: Bernd Verst Date: Fri, 24 Jul 2026 15:16:24 -0700 Subject: [PATCH 1/5] Avoid repeated orchestration-history scans for tracing (#769) OnRunOrchestratorAsync previously scanned orchestration history to correlate tracing events unconditionally, and for every qualifying new event it rescanned the full past-events list to find the originating TaskScheduled/SubOrchestrationInstanceCreated event - O(new events x past events) work, even when no tracing listener was registered. - Add TraceHelper.HasListeners, a cheap check backed by ActivitySource.HasListeners(), and gate all tracing-correlation work in OnRunOrchestratorAsync behind it so nothing is scanned when no listener is registered. - Replace the LINQ Concat-based scan for the ExecutionStarted event with a single-pass local function. - Add TraceHistoryEventLookup, which builds one dictionary index per work item (built lazily, once) for O(1) repeated lookups instead of rescanning PastEvents per new event, while preserving the exact original first-wins (SubOrchestrationInstanceCreated) and last-wins (TaskScheduled) semantics for duplicate event IDs. No public API changes; trace output, error handling, and replay determinism are unaffected. Adds regression tests for the no-listener fast path, first/last-wins correlation semantics end-to-end, and direct unit tests for TraceHistoryEventLookup. Fixes #769 Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> Copilot-Session: 9c538f3e-308d-48a3-af7f-b3baf90e97b2 --- src/Shared/Grpc/Tracing/TraceHelper.cs | 11 ++ .../Grpc/Tracing/TraceHistoryEventLookup.cs | 92 +++++++++ .../Grpc/GrpcDurableTaskWorker.Processor.cs | 177 +++++++++--------- .../Grpc.Tests/GrpcDurableTaskWorkerTests.cs | 150 +++++++++++++++ .../TraceHistoryEventLookupTests.cs | 134 +++++++++++++ 5 files changed, 477 insertions(+), 87 deletions(-) create mode 100644 src/Shared/Grpc/Tracing/TraceHistoryEventLookup.cs create mode 100644 test/Worker/Grpc.Tests/TraceHistoryEventLookupTests.cs diff --git a/src/Shared/Grpc/Tracing/TraceHelper.cs b/src/Shared/Grpc/Tracing/TraceHelper.cs index 1283ff12..ec5ecf5f 100644 --- a/src/Shared/Grpc/Tracing/TraceHelper.cs +++ b/src/Shared/Grpc/Tracing/TraceHelper.cs @@ -20,6 +20,17 @@ static class TraceHelper static readonly ActivitySource ActivityTraceSource = new ActivitySource(Source); + /// + /// Gets a value indicating whether any listener is currently registered for the Durable Task + /// . + /// + /// + /// This is a cheap check that callers can use to skip trace-event lookup work (such as scanning + /// orchestration history to correlate scheduling events) when no listener is registered and any resulting + /// would be discarded anyway. + /// + public static bool HasListeners => ActivityTraceSource.HasListeners(); + /// /// Starts a new trace activity for scheduling an orchestration from the client. /// diff --git a/src/Shared/Grpc/Tracing/TraceHistoryEventLookup.cs b/src/Shared/Grpc/Tracing/TraceHistoryEventLookup.cs new file mode 100644 index 00000000..2e7c4ec6 --- /dev/null +++ b/src/Shared/Grpc/Tracing/TraceHistoryEventLookup.cs @@ -0,0 +1,92 @@ +// Copyright (c) Microsoft Corporation. +// Licensed under the MIT License. + +using P = Microsoft.DurableTask.Protobuf; + +namespace Microsoft.DurableTask.Tracing; + +/// +/// Provides indexed lookups of past history events by event ID, used to correlate new completion/failure +/// events (e.g. "TaskCompleted") back to the history event that scheduled them (e.g. "TaskScheduled"). +/// +/// +/// The indexes are built lazily, at most once per instance, and cached for the lifetime of the instance. This +/// avoids re-scanning the full set of past events for every new event being processed in a work item, which +/// would otherwise be O(new events x past events) for work items with many new events. +/// +sealed class TraceHistoryEventLookup +{ + readonly IEnumerable pastEvents; + + Dictionary? taskScheduledEventsByEventId; + Dictionary? subOrchestrationInstanceCreatedEventsByEventId; + + /// + /// Initializes a new instance of the class. + /// + /// The past history events for the current orchestrator work item. + public TraceHistoryEventLookup(IEnumerable pastEvents) + { + this.pastEvents = pastEvents; + } + + /// + /// Gets the "TaskScheduled" history event with the given event ID, if any. + /// + /// The event ID to look up. + /// The matching event, or if none is found. + /// + /// If more than one "TaskScheduled" event shares the given event ID, the last one encountered (in history + /// order) is returned, matching the original LastOrDefault lookup semantics. + /// + public P.HistoryEvent? GetTaskScheduledEvent(int eventId) + { + this.taskScheduledEventsByEventId ??= BuildIndex( + this.pastEvents, P.HistoryEvent.EventTypeOneofCase.TaskScheduled, keepFirst: false); + return this.taskScheduledEventsByEventId.TryGetValue(eventId, out P.HistoryEvent? historyEvent) + ? historyEvent + : null; + } + + /// + /// Gets the "SubOrchestrationInstanceCreated" history event with the given event ID, if any. + /// + /// The event ID to look up. + /// The matching event, or if none is found. + /// + /// If more than one "SubOrchestrationInstanceCreated" event shares the given event ID, the first one + /// encountered (in history order) is returned, matching the original FirstOrDefault lookup semantics. + /// + public P.HistoryEvent? GetSubOrchestrationInstanceCreatedEvent(int eventId) + { + this.subOrchestrationInstanceCreatedEventsByEventId ??= BuildIndex( + this.pastEvents, P.HistoryEvent.EventTypeOneofCase.SubOrchestrationInstanceCreated, keepFirst: true); + return this.subOrchestrationInstanceCreatedEventsByEventId.TryGetValue(eventId, out P.HistoryEvent? historyEvent) + ? historyEvent + : null; + } + + static Dictionary BuildIndex( + IEnumerable events, P.HistoryEvent.EventTypeOneofCase eventType, bool keepFirst) + { + Dictionary index = new(); + foreach (P.HistoryEvent historyEvent in events) + { + if (historyEvent.EventTypeCase != eventType) + { + continue; + } + + if (keepFirst && index.ContainsKey(historyEvent.EventId)) + { + // Preserve first-match-wins semantics for duplicate event IDs. + continue; + } + + // Last write wins for duplicate event IDs, preserving last-match-wins semantics. + index[historyEvent.EventId] = historyEvent; + } + + return index; + } +} diff --git a/src/Worker/Grpc/GrpcDurableTaskWorker.Processor.cs b/src/Worker/Grpc/GrpcDurableTaskWorker.Processor.cs index 6d63f6af..c4a7080e 100644 --- a/src/Worker/Grpc/GrpcDurableTaskWorker.Processor.cs +++ b/src/Worker/Grpc/GrpcDurableTaskWorker.Processor.cs @@ -586,107 +586,110 @@ async Task OnRunOrchestratorAsync( string completionToken, CancellationToken cancellationToken) { - var executionStartedEvent = - request - .NewEvents - .Concat(request.PastEvents) - .Where(e => e.EventTypeCase == P.HistoryEvent.EventTypeOneofCase.ExecutionStarted) - .Select(e => e.ExecutionStarted) - .FirstOrDefault(); - - Activity? traceActivity = TraceHelper.StartTraceActivityForOrchestrationExecution( - executionStartedEvent, - request.OrchestrationTraceContext); - - if (executionStartedEvent is not null) + // Avoid the cost of scanning orchestration history for tracing purposes (potentially O(new events x + // past events) for work items with many new events) when no listener is registered for the Durable + // Task ActivitySource. In that case any Activity created below would be discarded anyway. + Activity? traceActivity = null; + if (TraceHelper.HasListeners) { - P.HistoryEvent? GetSuborchestrationInstanceCreatedEvent(int eventId) - { - var subOrchestrationEvent = - request - .PastEvents - .Where(x => x.EventTypeCase == P.HistoryEvent.EventTypeOneofCase.SubOrchestrationInstanceCreated) - .FirstOrDefault(x => x.EventId == eventId); + P.ExecutionStartedEvent? executionStartedEvent = FindExecutionStartedEvent(request); - return subOrchestrationEvent; - } + traceActivity = TraceHelper.StartTraceActivityForOrchestrationExecution( + executionStartedEvent, + request.OrchestrationTraceContext); - P.HistoryEvent? GetTaskScheduledEvent(int eventId) + if (executionStartedEvent is not null) { - var taskScheduledEvent = - request - .PastEvents - .Where(x => x.EventTypeCase == P.HistoryEvent.EventTypeOneofCase.TaskScheduled) - .LastOrDefault(x => x.EventId == eventId); + // Build lookups once per work item instead of rescanning PastEvents for every new event. + TraceHistoryEventLookup historyLookup = new(request.PastEvents); - return taskScheduledEvent; - } - - foreach (var newEvent in request.NewEvents) - { - switch (newEvent.EventTypeCase) + foreach (var newEvent in request.NewEvents) { - case P.HistoryEvent.EventTypeOneofCase.SubOrchestrationInstanceCompleted: - { - P.HistoryEvent? subOrchestrationInstanceCreatedEvent = - GetSuborchestrationInstanceCreatedEvent( - newEvent.SubOrchestrationInstanceCompleted.TaskScheduledId); - - TraceHelper.EmitTraceActivityForSubOrchestrationCompleted( - request.InstanceId, - subOrchestrationInstanceCreatedEvent, - subOrchestrationInstanceCreatedEvent?.SubOrchestrationInstanceCreated); - break; - } - - case P.HistoryEvent.EventTypeOneofCase.SubOrchestrationInstanceFailed: - { - P.HistoryEvent? subOrchestrationInstanceCreatedEvent = - GetSuborchestrationInstanceCreatedEvent( - newEvent.SubOrchestrationInstanceFailed.TaskScheduledId); - - TraceHelper.EmitTraceActivityForSubOrchestrationFailed( - request.InstanceId, - subOrchestrationInstanceCreatedEvent, - subOrchestrationInstanceCreatedEvent?.SubOrchestrationInstanceCreated, - newEvent.SubOrchestrationInstanceFailed); - break; - } + switch (newEvent.EventTypeCase) + { + case P.HistoryEvent.EventTypeOneofCase.SubOrchestrationInstanceCompleted: + { + P.HistoryEvent? subOrchestrationInstanceCreatedEvent = + historyLookup.GetSubOrchestrationInstanceCreatedEvent( + newEvent.SubOrchestrationInstanceCompleted.TaskScheduledId); + + TraceHelper.EmitTraceActivityForSubOrchestrationCompleted( + request.InstanceId, + subOrchestrationInstanceCreatedEvent, + subOrchestrationInstanceCreatedEvent?.SubOrchestrationInstanceCreated); + break; + } + + case P.HistoryEvent.EventTypeOneofCase.SubOrchestrationInstanceFailed: + { + P.HistoryEvent? subOrchestrationInstanceCreatedEvent = + historyLookup.GetSubOrchestrationInstanceCreatedEvent( + newEvent.SubOrchestrationInstanceFailed.TaskScheduledId); + + TraceHelper.EmitTraceActivityForSubOrchestrationFailed( + request.InstanceId, + subOrchestrationInstanceCreatedEvent, + subOrchestrationInstanceCreatedEvent?.SubOrchestrationInstanceCreated, + newEvent.SubOrchestrationInstanceFailed); + break; + } + + case P.HistoryEvent.EventTypeOneofCase.TaskCompleted: + { + P.HistoryEvent? taskScheduledEvent = + historyLookup.GetTaskScheduledEvent(newEvent.TaskCompleted.TaskScheduledId); - case P.HistoryEvent.EventTypeOneofCase.TaskCompleted: - { - P.HistoryEvent? taskScheduledEvent = - GetTaskScheduledEvent(newEvent.TaskCompleted.TaskScheduledId); + TraceHelper.EmitTraceActivityForTaskCompleted( + request.InstanceId, + taskScheduledEvent, + taskScheduledEvent?.TaskScheduled); + break; + } - TraceHelper.EmitTraceActivityForTaskCompleted( + case P.HistoryEvent.EventTypeOneofCase.TaskFailed: + { + P.HistoryEvent? taskScheduledEvent = + historyLookup.GetTaskScheduledEvent(newEvent.TaskFailed.TaskScheduledId); + + TraceHelper.EmitTraceActivityForTaskFailed( + request.InstanceId, + taskScheduledEvent, + taskScheduledEvent?.TaskScheduled, + newEvent.TaskFailed); + break; + } + + case P.HistoryEvent.EventTypeOneofCase.TimerFired: + TraceHelper.EmitTraceActivityForTimer( request.InstanceId, - taskScheduledEvent, - taskScheduledEvent?.TaskScheduled); + executionStartedEvent.Name, + newEvent.Timestamp.ToDateTime(), + newEvent.TimerFired); break; - } + } + } + } + } - case P.HistoryEvent.EventTypeOneofCase.TaskFailed: - { - P.HistoryEvent? taskScheduledEvent = - GetTaskScheduledEvent(newEvent.TaskFailed.TaskScheduledId); + static P.ExecutionStartedEvent? FindExecutionStartedEvent(P.OrchestratorRequest request) + { + foreach (P.HistoryEvent newEvent in request.NewEvents) + { + if (newEvent.EventTypeCase == P.HistoryEvent.EventTypeOneofCase.ExecutionStarted) + { + return newEvent.ExecutionStarted; + } + } - TraceHelper.EmitTraceActivityForTaskFailed( - request.InstanceId, - taskScheduledEvent, - taskScheduledEvent?.TaskScheduled, - newEvent.TaskFailed); - break; - } - - case P.HistoryEvent.EventTypeOneofCase.TimerFired: - TraceHelper.EmitTraceActivityForTimer( - request.InstanceId, - executionStartedEvent.Name, - newEvent.Timestamp.ToDateTime(), - newEvent.TimerFired); - break; + foreach (P.HistoryEvent pastEvent in request.PastEvents) + { + if (pastEvent.EventTypeCase == P.HistoryEvent.EventTypeOneofCase.ExecutionStarted) + { + return pastEvent.ExecutionStarted; } } + + return null; } OrchestratorExecutionResult? result = null; diff --git a/test/Worker/Grpc.Tests/GrpcDurableTaskWorkerTests.cs b/test/Worker/Grpc.Tests/GrpcDurableTaskWorkerTests.cs index bc9faab6..8f5d6e55 100644 --- a/test/Worker/Grpc.Tests/GrpcDurableTaskWorkerTests.cs +++ b/test/Worker/Grpc.Tests/GrpcDurableTaskWorkerTests.cs @@ -2,12 +2,14 @@ // Licensed under the MIT License. using System.Collections.Concurrent; +using System.Diagnostics; using System.IO; using System.Reflection; using Google.Protobuf.WellKnownTypes; using Grpc.Core; using Microsoft.DurableTask; using Microsoft.DurableTask.Tests.Logging; +using Microsoft.DurableTask.Tracing; using Microsoft.DurableTask.Worker; using Microsoft.DurableTask.Worker.Grpc.Internal; using Microsoft.Extensions.Logging; @@ -359,6 +361,154 @@ public async Task DispatchWorkItem_ActivityRequest_NotificationFailure_Completes logs.Should().Contain(log => log.Message.Contains("Activity notification callback failed for phase 'Completed'")); } + // The following two tests both touch the process-wide "Microsoft.DurableTask" ActivitySource used by + // TraceHelper, so they are kept in this class (whose test methods xunit runs sequentially by default) to + // avoid flaky interference between them. + [Fact] + public async Task DispatchWorkItem_OrchestratorRequest_NoActivityListeners_SkipsTracingWorkAndCompletes() + { + // Arrange: verify no listener is registered for the Durable Task ActivitySource, so that + // OnRunOrchestratorAsync takes the fast path that skips all trace-event lookup work. + TraceHelper.HasListeners.Should().BeFalse(); + + P.WorkItem orchestratorWorkItem = CreateOrchestratorWorkItemWithDuplicateEventIds(); + + TaskCompletionSource completed = new(TaskCreationOptions.RunContinuationsAsynchronously); + GrpcDurableTaskWorker worker = CreateActivityWorker(new GrpcDurableTaskWorkerOptions()); + Mock clientMock = new( + MockBehavior.Strict, + new object[] { Mock.Of() }); + clientMock + .Setup(client => client.CompleteOrchestratorTaskAsync( + It.IsAny(), + It.IsAny(), + It.IsAny(), + It.IsAny())) + .Callback(() => completed.TrySetResult()) + .Returns(CreateUnaryCall(Task.FromResult(new P.CompleteTaskResponse()))); + object processor = CreateProcessor(worker, clientMock.Object); + + // Act + InvokeDispatchWorkItem(processor, orchestratorWorkItem, CancellationToken.None); + await completed.Task.WaitAsync(TimeSpan.FromSeconds(5)); + + // Assert: work item processing still completes normally with no listener registered. + clientMock.VerifyAll(); + } + + [Fact] + public async Task DispatchWorkItem_OrchestratorRequest_WithActivityListener_UsesFirstAndLastWinsSemantics() + { + // Arrange: register a listener so OnRunOrchestratorAsync performs the tracing-correlation work, and + // craft history containing duplicate event IDs to verify the last/first-wins lookup semantics are + // preserved by the new indexed TraceHistoryEventLookup. + ConcurrentQueue stoppedActivities = new(); + using ActivityListener listener = new() + { + ShouldListenTo = source => source.Name == "Microsoft.DurableTask", + Sample = (ref ActivityCreationOptions options) => ActivitySamplingResult.AllDataAndRecorded, + ActivityStopped = activity => stoppedActivities.Enqueue(activity), + }; + ActivitySource.AddActivityListener(listener); + TraceHelper.HasListeners.Should().BeTrue(); + + P.WorkItem orchestratorWorkItem = CreateOrchestratorWorkItemWithDuplicateEventIds(); + + TaskCompletionSource completed = new(TaskCreationOptions.RunContinuationsAsynchronously); + GrpcDurableTaskWorker worker = CreateActivityWorker(new GrpcDurableTaskWorkerOptions()); + Mock clientMock = new( + MockBehavior.Strict, + new object[] { Mock.Of() }); + clientMock + .Setup(client => client.CompleteOrchestratorTaskAsync( + It.IsAny(), + It.IsAny(), + It.IsAny(), + It.IsAny())) + .Callback(() => completed.TrySetResult()) + .Returns(CreateUnaryCall(Task.FromResult(new P.CompleteTaskResponse()))); + object processor = CreateProcessor(worker, clientMock.Object); + + // Act + InvokeDispatchWorkItem(processor, orchestratorWorkItem, CancellationToken.None); + await completed.Task.WaitAsync(TimeSpan.FromSeconds(5)); + + // Assert: the "TaskScheduled" event with EventId=1 was duplicated; the last one (by history order) + // should win, matching the original LastOrDefault lookup semantics. + Activity taskActivity = stoppedActivities.Should().ContainSingle( + a => a.GetTagItem(Schema.Task.Type) as string == TraceActivityConstants.Activity).Subject; + taskActivity.GetTagItem(Schema.Task.Name).Should().Be("SecondScheduled"); + + // The "SubOrchestrationInstanceCreated" event with EventId=2 was duplicated; the first one (by history + // order) should win, matching the original FirstOrDefault lookup semantics. + Activity subOrchestrationActivity = stoppedActivities.Should().ContainSingle( + a => a.GetTagItem(Schema.Task.Type) as string == TraceActivityConstants.Orchestration + && a.OperationName.Contains("FirstSub", StringComparison.Ordinal)).Subject; + subOrchestrationActivity.GetTagItem(Schema.Task.Name).Should().Be("FirstSub"); + } + + static P.WorkItem CreateOrchestratorWorkItemWithDuplicateEventIds() + { + P.OrchestratorRequest request = new() + { + InstanceId = "instance1", + ExecutionId = "execution1", + }; + request.PastEvents.Add(new P.HistoryEvent + { + EventId = -1, + ExecutionStarted = new P.ExecutionStartedEvent + { + Name = "TestOrchestration", + OrchestrationInstance = new P.OrchestrationInstance { InstanceId = "instance1", ExecutionId = "execution1" }, + }, + }); + request.PastEvents.Add(new P.HistoryEvent + { + EventId = 1, + TaskScheduled = new P.TaskScheduledEvent { Name = "FirstScheduled" }, + }); + request.PastEvents.Add(new P.HistoryEvent + { + EventId = 1, + TaskScheduled = new P.TaskScheduledEvent { Name = "SecondScheduled" }, + }); + request.PastEvents.Add(new P.HistoryEvent + { + EventId = 2, + SubOrchestrationInstanceCreated = new P.SubOrchestrationInstanceCreatedEvent + { + InstanceId = "sub1", + Name = "FirstSub", + }, + }); + request.PastEvents.Add(new P.HistoryEvent + { + EventId = 2, + SubOrchestrationInstanceCreated = new P.SubOrchestrationInstanceCreatedEvent + { + InstanceId = "sub2", + Name = "SecondSub", + }, + }); + request.NewEvents.Add(new P.HistoryEvent + { + EventId = 10, + TaskCompleted = new P.TaskCompletedEvent { TaskScheduledId = 1 }, + }); + request.NewEvents.Add(new P.HistoryEvent + { + EventId = 11, + SubOrchestrationInstanceCompleted = new P.SubOrchestrationInstanceCompletedEvent { TaskScheduledId = 2 }, + }); + + return new P.WorkItem + { + OrchestratorRequest = request, + CompletionToken = "completion1", + }; + } + [Fact] public async Task ProcessorExecuteAsync_HelloDeadlineExceeded_ReturnsChannelRecreateRequested() { diff --git a/test/Worker/Grpc.Tests/TraceHistoryEventLookupTests.cs b/test/Worker/Grpc.Tests/TraceHistoryEventLookupTests.cs new file mode 100644 index 00000000..ab9f6544 --- /dev/null +++ b/test/Worker/Grpc.Tests/TraceHistoryEventLookupTests.cs @@ -0,0 +1,134 @@ +// Copyright (c) Microsoft Corporation. +// Licensed under the MIT License. + +using Microsoft.DurableTask.Tracing; +using P = Microsoft.DurableTask.Protobuf; + +namespace Microsoft.DurableTask.Worker.Grpc.Tests; + +public class TraceHistoryEventLookupTests +{ + [Fact] + public void GetTaskScheduledEvent_DuplicateEventIds_ReturnsLastMatch() + { + // Arrange + List pastEvents = + [ + CreateTaskScheduled(eventId: 1, name: "FirstScheduled"), + CreateTaskScheduled(eventId: 1, name: "SecondScheduled"), + ]; + TraceHistoryEventLookup lookup = new(pastEvents); + + // Act + P.HistoryEvent? result = lookup.GetTaskScheduledEvent(1); + + // Assert + result.Should().NotBeNull(); + result!.TaskScheduled.Name.Should().Be("SecondScheduled"); + } + + [Fact] + public void GetSubOrchestrationInstanceCreatedEvent_DuplicateEventIds_ReturnsFirstMatch() + { + // Arrange + List pastEvents = + [ + CreateSubOrchestrationInstanceCreated(eventId: 2, name: "FirstSub"), + CreateSubOrchestrationInstanceCreated(eventId: 2, name: "SecondSub"), + ]; + TraceHistoryEventLookup lookup = new(pastEvents); + + // Act + P.HistoryEvent? result = lookup.GetSubOrchestrationInstanceCreatedEvent(2); + + // Assert + result.Should().NotBeNull(); + result!.SubOrchestrationInstanceCreated.Name.Should().Be("FirstSub"); + } + + [Fact] + public void GetTaskScheduledEvent_NoMatch_ReturnsNull() + { + // Arrange + List pastEvents = [CreateTaskScheduled(eventId: 1, name: "Scheduled")]; + TraceHistoryEventLookup lookup = new(pastEvents); + + // Act + P.HistoryEvent? result = lookup.GetTaskScheduledEvent(99); + + // Assert + result.Should().BeNull(); + } + + [Fact] + public void GetSubOrchestrationInstanceCreatedEvent_NoMatch_ReturnsNull() + { + // Arrange + List pastEvents = [CreateSubOrchestrationInstanceCreated(eventId: 2, name: "Sub")]; + TraceHistoryEventLookup lookup = new(pastEvents); + + // Act + P.HistoryEvent? result = lookup.GetSubOrchestrationInstanceCreatedEvent(99); + + // Assert + result.Should().BeNull(); + } + + [Fact] + public void GetTaskScheduledEvent_IgnoresOtherEventTypesWithSameEventId() + { + // Arrange: a SubOrchestrationInstanceCreated event shares the event ID of the TaskScheduled event we + // look up, but must not be returned since it is a different history event type. + List pastEvents = + [ + CreateSubOrchestrationInstanceCreated(eventId: 5, name: "Sub"), + CreateTaskScheduled(eventId: 5, name: "Scheduled"), + ]; + TraceHistoryEventLookup lookup = new(pastEvents); + + // Act + P.HistoryEvent? taskScheduled = lookup.GetTaskScheduledEvent(5); + P.HistoryEvent? subOrchestrationCreated = lookup.GetSubOrchestrationInstanceCreatedEvent(5); + + // Assert + taskScheduled.Should().NotBeNull(); + taskScheduled!.TaskScheduled.Name.Should().Be("Scheduled"); + subOrchestrationCreated.Should().NotBeNull(); + subOrchestrationCreated!.SubOrchestrationInstanceCreated.Name.Should().Be("Sub"); + } + + [Fact] + public void GetTaskScheduledEvent_EmptyPastEvents_ReturnsNull() + { + // Arrange + TraceHistoryEventLookup lookup = new([]); + + // Act + P.HistoryEvent? result = lookup.GetTaskScheduledEvent(0); + + // Assert + result.Should().BeNull(); + } + + static P.HistoryEvent CreateTaskScheduled(int eventId, string name) + { + return new P.HistoryEvent + { + EventId = eventId, + TaskScheduled = new P.TaskScheduledEvent { Name = name }, + }; + } + + static P.HistoryEvent CreateSubOrchestrationInstanceCreated(int eventId, string name) + { + return new P.HistoryEvent + { + EventId = eventId, + SubOrchestrationInstanceCreated = new P.SubOrchestrationInstanceCreatedEvent + { + InstanceId = $"sub-{eventId}", + Name = name, + }, + }; + } +} From e736e30b78426cfe64cb462a9c24e50880c039a6 Mon Sep 17 00:00:00 2001 From: Bernd Verst Date: Wed, 12 Aug 2026 18:58:04 -0700 Subject: [PATCH 2/5] Address tracing lookup review feedback Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> Copilot-Session: 01dc5b7c-c742-45b7-ae8c-5cda9bcfbd2c --- .../Grpc/Tracing/TraceHistoryEventLookup.cs | 41 ++++++------- .../TracingIntegrationTests.cs | 59 +++++++++++++++++++ .../Grpc.Tests/GrpcDurableTaskWorkerTests.cs | 48 +++++++-------- .../TraceHistoryEventLookupTests.cs | 16 ++--- 4 files changed, 109 insertions(+), 55 deletions(-) diff --git a/src/Shared/Grpc/Tracing/TraceHistoryEventLookup.cs b/src/Shared/Grpc/Tracing/TraceHistoryEventLookup.cs index 2e7c4ec6..dc04557c 100644 --- a/src/Shared/Grpc/Tracing/TraceHistoryEventLookup.cs +++ b/src/Shared/Grpc/Tracing/TraceHistoryEventLookup.cs @@ -10,9 +10,9 @@ namespace Microsoft.DurableTask.Tracing; /// events (e.g. "TaskCompleted") back to the history event that scheduled them (e.g. "TaskScheduled"). /// /// -/// The indexes are built lazily, at most once per instance, and cached for the lifetime of the instance. This -/// avoids re-scanning the full set of past events for every new event being processed in a work item, which -/// would otherwise be O(new events x past events) for work items with many new events. +/// The indexes are built lazily, at most once per event type, and cached for the lifetime of the orchestrator +/// work item. This avoids re-scanning the full set of past events for every new event being processed in a work +/// item, which would otherwise be O(new events x past events) for work items with many new events. /// sealed class TraceHistoryEventLookup { @@ -35,14 +35,13 @@ public TraceHistoryEventLookup(IEnumerable pastEvents) /// /// The event ID to look up. /// The matching event, or if none is found. - /// - /// If more than one "TaskScheduled" event shares the given event ID, the last one encountered (in history - /// order) is returned, matching the original LastOrDefault lookup semantics. - /// + /// + /// Thrown when more than one "TaskScheduled" event has the same event ID. + /// public P.HistoryEvent? GetTaskScheduledEvent(int eventId) { this.taskScheduledEventsByEventId ??= BuildIndex( - this.pastEvents, P.HistoryEvent.EventTypeOneofCase.TaskScheduled, keepFirst: false); + this.pastEvents, P.HistoryEvent.EventTypeOneofCase.TaskScheduled); return this.taskScheduledEventsByEventId.TryGetValue(eventId, out P.HistoryEvent? historyEvent) ? historyEvent : null; @@ -53,21 +52,20 @@ public TraceHistoryEventLookup(IEnumerable pastEvents) /// /// The event ID to look up. /// The matching event, or if none is found. - /// - /// If more than one "SubOrchestrationInstanceCreated" event shares the given event ID, the first one - /// encountered (in history order) is returned, matching the original FirstOrDefault lookup semantics. - /// + /// + /// Thrown when more than one "SubOrchestrationInstanceCreated" event has the same event ID. + /// public P.HistoryEvent? GetSubOrchestrationInstanceCreatedEvent(int eventId) { this.subOrchestrationInstanceCreatedEventsByEventId ??= BuildIndex( - this.pastEvents, P.HistoryEvent.EventTypeOneofCase.SubOrchestrationInstanceCreated, keepFirst: true); + this.pastEvents, P.HistoryEvent.EventTypeOneofCase.SubOrchestrationInstanceCreated); return this.subOrchestrationInstanceCreatedEventsByEventId.TryGetValue(eventId, out P.HistoryEvent? historyEvent) ? historyEvent : null; } static Dictionary BuildIndex( - IEnumerable events, P.HistoryEvent.EventTypeOneofCase eventType, bool keepFirst) + IEnumerable events, P.HistoryEvent.EventTypeOneofCase eventType) { Dictionary index = new(); foreach (P.HistoryEvent historyEvent in events) @@ -77,14 +75,17 @@ public TraceHistoryEventLookup(IEnumerable pastEvents) continue; } - if (keepFirst && index.ContainsKey(historyEvent.EventId)) + try { - // Preserve first-match-wins semantics for duplicate event IDs. - continue; + index.Add(historyEvent.EventId, historyEvent); + } + catch (ArgumentException exception) + { + throw new InvalidOperationException( + $"Past orchestration history contains multiple '{eventType}' events with event ID " + + $"'{historyEvent.EventId}'.", + exception); } - - // Last write wins for duplicate event IDs, preserving last-match-wins semantics. - index[historyEvent.EventId] = historyEvent; } return index; diff --git a/test/Grpc.IntegrationTests/TracingIntegrationTests.cs b/test/Grpc.IntegrationTests/TracingIntegrationTests.cs index 041c76de..91432b21 100644 --- a/test/Grpc.IntegrationTests/TracingIntegrationTests.cs +++ b/test/Grpc.IntegrationTests/TracingIntegrationTests.cs @@ -38,6 +38,65 @@ static ActivityListener CreateListener(string[] sources, ConcurrentBag static readonly ActivitySource TestActivitySource = new(TestActivitySourceName); + [Fact] + public async Task HistoryEventLookupCorrelatesDistinctScheduledOperations() + { + // Arrange + ConcurrentBag activities = new(); + using ActivityListener listener = CreateListener(ActivitySourceNames, activities); + + string orchestratorName = nameof(HistoryEventLookupCorrelatesDistinctScheduledOperations); + string firstActivityName = "FirstActivity"; + string secondActivityName = "SecondActivity"; + string subOrchestratorName = "SubOrchestration"; + + await using HostTestLifetime server = await this.StartWorkerAsync(b => + { + b.AddTasks(tasks => tasks + .AddOrchestratorFunc( + orchestratorName, + async (ctx, input) => + { + await ctx.CallActivityAsync(firstActivityName, input); + await ctx.CallActivityAsync(secondActivityName, input); + await ctx.CallSubOrchestratorAsync(subOrchestratorName, input: input); + return true; + }) + .AddOrchestratorFunc(subOrchestratorName, (ctx, input) => input) + .AddActivityFunc(firstActivityName, (ctx, input) => input) + .AddActivityFunc(secondActivityName, (ctx, input) => input)); + }); + + // Act + OrchestrationMetadata metadata; + using (TestActivitySource.StartActivity("Test")) + { + string instanceId = await server.Client.ScheduleNewOrchestrationInstanceAsync( + orchestratorName, + input: true, + cancellation: this.TimeoutToken); + metadata = await server.Client.WaitForInstanceCompletionAsync( + instanceId, + getInputsAndOutputs: true, + this.TimeoutToken); + } + + // Assert + metadata.RuntimeStatus.Should().Be(OrchestrationRuntimeStatus.Completed); + activities.Should().ContainSingle( + activity => activity.Kind == ActivityKind.Client + && activity.Source.Name == CoreActivitySourceName + && activity.OperationName == $"activity:{firstActivityName}"); + activities.Should().ContainSingle( + activity => activity.Kind == ActivityKind.Client + && activity.Source.Name == CoreActivitySourceName + && activity.OperationName == $"activity:{secondActivityName}"); + activities.Should().ContainSingle( + activity => activity.Kind == ActivityKind.Client + && activity.Source.Name == CoreActivitySourceName + && activity.OperationName == $"orchestration:{subOrchestratorName}"); + } + [Fact] public async Task MultiTaskOrchestration() { diff --git a/test/Worker/Grpc.Tests/GrpcDurableTaskWorkerTests.cs b/test/Worker/Grpc.Tests/GrpcDurableTaskWorkerTests.cs index 0acb0756..f5402fdd 100644 --- a/test/Worker/Grpc.Tests/GrpcDurableTaskWorkerTests.cs +++ b/test/Worker/Grpc.Tests/GrpcDurableTaskWorkerTests.cs @@ -370,10 +370,10 @@ public async Task DispatchWorkItem_ActivityRequest_NotificationFailure_Completes // TraceHelper, so they are kept in this class (whose test methods xunit runs sequentially by default) to // avoid flaky interference between them. [Fact] - public async Task DispatchWorkItem_OrchestratorRequest_NoActivityListeners_SkipsTracingWorkAndCompletes() + public async Task DispatchWorkItem_OrchestratorRequest_NoActivityListeners_DoesNotBuildHistoryIndexes() { - // Arrange: verify no listener is registered for the Durable Task ActivitySource, so that - // OnRunOrchestratorAsync takes the fast path that skips all trace-event lookup work. + // Arrange: duplicate event IDs cause TraceHistoryEventLookup to throw when it builds an index. If this + // invalid history still completes, the no-listener fast path did not invoke either lookup method. TraceHelper.HasListeners.Should().BeFalse(); P.WorkItem orchestratorWorkItem = CreateOrchestratorWorkItemWithDuplicateEventIds(); @@ -402,54 +402,48 @@ public async Task DispatchWorkItem_OrchestratorRequest_NoActivityListeners_Skips } [Fact] - public async Task DispatchWorkItem_OrchestratorRequest_WithActivityListener_UsesFirstAndLastWinsSemantics() + public async Task DispatchWorkItem_OrchestratorRequest_WithActivityListener_DuplicateEventIds_AbandonsWorkItem() { - // Arrange: register a listener so OnRunOrchestratorAsync performs the tracing-correlation work, and - // craft history containing duplicate event IDs to verify the last/first-wins lookup semantics are - // preserved by the new indexed TraceHistoryEventLookup. - ConcurrentQueue stoppedActivities = new(); + // Arrange using ActivityListener listener = new() { ShouldListenTo = source => source.Name == "Microsoft.DurableTask", Sample = (ref ActivityCreationOptions options) => ActivitySamplingResult.AllDataAndRecorded, - ActivityStopped = activity => stoppedActivities.Enqueue(activity), }; ActivitySource.AddActivityListener(listener); TraceHelper.HasListeners.Should().BeTrue(); P.WorkItem orchestratorWorkItem = CreateOrchestratorWorkItemWithDuplicateEventIds(); - TaskCompletionSource completed = new(TaskCreationOptions.RunContinuationsAsynchronously); + TaskCompletionSource abandoned = new(TaskCreationOptions.RunContinuationsAsynchronously); GrpcDurableTaskWorker worker = CreateActivityWorker(new GrpcDurableTaskWorkerOptions()); Mock clientMock = new( MockBehavior.Strict, new object[] { Mock.Of() }); clientMock - .Setup(client => client.CompleteOrchestratorTaskAsync( - It.IsAny(), + .Setup(client => client.AbandonTaskOrchestratorWorkItemAsync( + It.Is( + request => request.CompletionToken == orchestratorWorkItem.CompletionToken), It.IsAny(), It.IsAny(), It.IsAny())) - .Callback(() => completed.TrySetResult()) - .Returns(CreateUnaryCall(Task.FromResult(new P.CompleteTaskResponse()))); + .Callback(() => abandoned.TrySetResult()) + .Returns(CreateUnaryCall(Task.FromResult(new P.AbandonOrchestrationTaskResponse()))); object processor = CreateProcessor(worker, clientMock.Object); // Act InvokeDispatchWorkItem(processor, orchestratorWorkItem, CancellationToken.None); - await completed.Task.WaitAsync(TimeSpan.FromSeconds(5)); + await abandoned.Task.WaitAsync(TimeSpan.FromSeconds(5)); - // Assert: the "TaskScheduled" event with EventId=1 was duplicated; the last one (by history order) - // should win, matching the original LastOrDefault lookup semantics. - Activity taskActivity = stoppedActivities.Should().ContainSingle( - a => a.GetTagItem(Schema.Task.Type) as string == TraceActivityConstants.Activity).Subject; - taskActivity.GetTagItem(Schema.Task.Name).Should().Be("SecondScheduled"); - - // The "SubOrchestrationInstanceCreated" event with EventId=2 was duplicated; the first one (by history - // order) should win, matching the original FirstOrDefault lookup semantics. - Activity subOrchestrationActivity = stoppedActivities.Should().ContainSingle( - a => a.GetTagItem(Schema.Task.Type) as string == TraceActivityConstants.Orchestration - && a.OperationName.Contains("FirstSub", StringComparison.Ordinal)).Subject; - subOrchestrationActivity.GetTagItem(Schema.Task.Name).Should().Be("FirstSub"); + // Assert + clientMock.VerifyAll(); + clientMock.Verify( + client => client.CompleteOrchestratorTaskAsync( + It.IsAny(), + It.IsAny(), + It.IsAny(), + It.IsAny()), + Times.Never); } static P.WorkItem CreateOrchestratorWorkItemWithDuplicateEventIds() diff --git a/test/Worker/Grpc.Tests/TraceHistoryEventLookupTests.cs b/test/Worker/Grpc.Tests/TraceHistoryEventLookupTests.cs index ab9f6544..f7a2cfe4 100644 --- a/test/Worker/Grpc.Tests/TraceHistoryEventLookupTests.cs +++ b/test/Worker/Grpc.Tests/TraceHistoryEventLookupTests.cs @@ -9,7 +9,7 @@ namespace Microsoft.DurableTask.Worker.Grpc.Tests; public class TraceHistoryEventLookupTests { [Fact] - public void GetTaskScheduledEvent_DuplicateEventIds_ReturnsLastMatch() + public void GetTaskScheduledEvent_DuplicateEventIds_Throws() { // Arrange List pastEvents = @@ -20,15 +20,15 @@ public void GetTaskScheduledEvent_DuplicateEventIds_ReturnsLastMatch() TraceHistoryEventLookup lookup = new(pastEvents); // Act - P.HistoryEvent? result = lookup.GetTaskScheduledEvent(1); + Action act = () => lookup.GetTaskScheduledEvent(1); // Assert - result.Should().NotBeNull(); - result!.TaskScheduled.Name.Should().Be("SecondScheduled"); + act.Should().Throw() + .WithMessage("*'TaskScheduled'*event ID '1'*"); } [Fact] - public void GetSubOrchestrationInstanceCreatedEvent_DuplicateEventIds_ReturnsFirstMatch() + public void GetSubOrchestrationInstanceCreatedEvent_DuplicateEventIds_Throws() { // Arrange List pastEvents = @@ -39,11 +39,11 @@ public void GetSubOrchestrationInstanceCreatedEvent_DuplicateEventIds_ReturnsFir TraceHistoryEventLookup lookup = new(pastEvents); // Act - P.HistoryEvent? result = lookup.GetSubOrchestrationInstanceCreatedEvent(2); + Action act = () => lookup.GetSubOrchestrationInstanceCreatedEvent(2); // Assert - result.Should().NotBeNull(); - result!.SubOrchestrationInstanceCreated.Name.Should().Be("FirstSub"); + act.Should().Throw() + .WithMessage("*'SubOrchestrationInstanceCreated'*event ID '2'*"); } [Fact] From 5ff5dcddb6ddc9d9c431a4581944c24cd03c1514 Mon Sep 17 00:00:00 2001 From: Bernd Verst Date: Wed, 12 Aug 2026 19:23:58 -0700 Subject: [PATCH 3/5] Isolate tracing integration test names Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> Copilot-Session: 01dc5b7c-c742-45b7-ae8c-5cda9bcfbd2c --- test/Grpc.IntegrationTests/TracingIntegrationTests.cs | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/test/Grpc.IntegrationTests/TracingIntegrationTests.cs b/test/Grpc.IntegrationTests/TracingIntegrationTests.cs index 91432b21..f85d3ac0 100644 --- a/test/Grpc.IntegrationTests/TracingIntegrationTests.cs +++ b/test/Grpc.IntegrationTests/TracingIntegrationTests.cs @@ -46,9 +46,9 @@ public async Task HistoryEventLookupCorrelatesDistinctScheduledOperations() using ActivityListener listener = CreateListener(ActivitySourceNames, activities); string orchestratorName = nameof(HistoryEventLookupCorrelatesDistinctScheduledOperations); - string firstActivityName = "FirstActivity"; - string secondActivityName = "SecondActivity"; - string subOrchestratorName = "SubOrchestration"; + string firstActivityName = $"{orchestratorName}.FirstActivity"; + string secondActivityName = $"{orchestratorName}.SecondActivity"; + string subOrchestratorName = $"{orchestratorName}.SubOrchestration"; await using HostTestLifetime server = await this.StartWorkerAsync(b => { From 3830a05c1ceb598548c6aef7c1c1f7021b4cc4da Mon Sep 17 00:00:00 2001 From: Bernd Verst Date: Wed, 12 Aug 2026 20:05:30 -0700 Subject: [PATCH 4/5] Bound tracing history index storage Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> Copilot-Session: 01dc5b7c-c742-45b7-ae8c-5cda9bcfbd2c --- .../Grpc/Tracing/TraceHistoryEventLookup.cs | 144 ++++++++++--- .../Grpc/GrpcDurableTaskWorker.Processor.cs | 4 +- .../TraceHistoryEventLookupTests.cs | 195 +++++++++++++++++- 3 files changed, 305 insertions(+), 38 deletions(-) diff --git a/src/Shared/Grpc/Tracing/TraceHistoryEventLookup.cs b/src/Shared/Grpc/Tracing/TraceHistoryEventLookup.cs index dc04557c..52d3b093 100644 --- a/src/Shared/Grpc/Tracing/TraceHistoryEventLookup.cs +++ b/src/Shared/Grpc/Tracing/TraceHistoryEventLookup.cs @@ -10,24 +10,60 @@ namespace Microsoft.DurableTask.Tracing; /// events (e.g. "TaskCompleted") back to the history event that scheduled them (e.g. "TaskScheduled"). /// /// -/// The indexes are built lazily, at most once per event type, and cached for the lifetime of the orchestrator -/// work item. This avoids re-scanning the full set of past events for every new event being processed in a work -/// item, which would otherwise be O(new events x past events) for work items with many new events. +/// The indexes contain only scheduling event IDs referenced by the current work item's new completion/failure +/// events. They are built together in one lazy scan and cached for the lifetime of the orchestrator work item. +/// This avoids both re-scanning the full set of past events for every new event and retaining unrelated history +/// events, bounding lookup storage by the number of correlation IDs in the current work item. /// sealed class TraceHistoryEventLookup { readonly IEnumerable pastEvents; - Dictionary? taskScheduledEventsByEventId; - Dictionary? subOrchestrationInstanceCreatedEventsByEventId; + Dictionary? taskScheduledEventsByEventId; + Dictionary? subOrchestrationInstanceCreatedEventsByEventId; + + HashSet? duplicateTaskScheduledEventIds; + HashSet? duplicateSubOrchestrationInstanceCreatedEventIds; + bool indexesBuilt; /// /// Initializes a new instance of the class. /// /// The past history events for the current orchestrator work item. - public TraceHistoryEventLookup(IEnumerable pastEvents) + /// The new history events whose correlation IDs may be looked up. + public TraceHistoryEventLookup( + IEnumerable pastEvents, + IEnumerable newEvents) { this.pastEvents = pastEvents; + + foreach (P.HistoryEvent newEvent in newEvents) + { + switch (newEvent.EventTypeCase) + { + case P.HistoryEvent.EventTypeOneofCase.TaskCompleted: + this.taskScheduledEventsByEventId ??= new(); + this.taskScheduledEventsByEventId[newEvent.TaskCompleted.TaskScheduledId] = null; + break; + + case P.HistoryEvent.EventTypeOneofCase.TaskFailed: + this.taskScheduledEventsByEventId ??= new(); + this.taskScheduledEventsByEventId[newEvent.TaskFailed.TaskScheduledId] = null; + break; + + case P.HistoryEvent.EventTypeOneofCase.SubOrchestrationInstanceCompleted: + this.subOrchestrationInstanceCreatedEventsByEventId ??= new(); + this.subOrchestrationInstanceCreatedEventsByEventId[ + newEvent.SubOrchestrationInstanceCompleted.TaskScheduledId] = null; + break; + + case P.HistoryEvent.EventTypeOneofCase.SubOrchestrationInstanceFailed: + this.subOrchestrationInstanceCreatedEventsByEventId ??= new(); + this.subOrchestrationInstanceCreatedEventsByEventId[ + newEvent.SubOrchestrationInstanceFailed.TaskScheduledId] = null; + break; + } + } } /// @@ -40,9 +76,14 @@ public TraceHistoryEventLookup(IEnumerable pastEvents) /// public P.HistoryEvent? GetTaskScheduledEvent(int eventId) { - this.taskScheduledEventsByEventId ??= BuildIndex( - this.pastEvents, P.HistoryEvent.EventTypeOneofCase.TaskScheduled); - return this.taskScheduledEventsByEventId.TryGetValue(eventId, out P.HistoryEvent? historyEvent) + this.BuildIndexes(); + ThrowIfDuplicate( + this.duplicateTaskScheduledEventIds, + eventId, + P.HistoryEvent.EventTypeOneofCase.TaskScheduled); + + return this.taskScheduledEventsByEventId is not null + && this.taskScheduledEventsByEventId.TryGetValue(eventId, out P.HistoryEvent? historyEvent) ? historyEvent : null; } @@ -57,37 +98,80 @@ public TraceHistoryEventLookup(IEnumerable pastEvents) /// public P.HistoryEvent? GetSubOrchestrationInstanceCreatedEvent(int eventId) { - this.subOrchestrationInstanceCreatedEventsByEventId ??= BuildIndex( - this.pastEvents, P.HistoryEvent.EventTypeOneofCase.SubOrchestrationInstanceCreated); - return this.subOrchestrationInstanceCreatedEventsByEventId.TryGetValue(eventId, out P.HistoryEvent? historyEvent) + this.BuildIndexes(); + ThrowIfDuplicate( + this.duplicateSubOrchestrationInstanceCreatedEventIds, + eventId, + P.HistoryEvent.EventTypeOneofCase.SubOrchestrationInstanceCreated); + + return this.subOrchestrationInstanceCreatedEventsByEventId is not null + && this.subOrchestrationInstanceCreatedEventsByEventId.TryGetValue( + eventId, out P.HistoryEvent? historyEvent) ? historyEvent : null; } - static Dictionary BuildIndex( - IEnumerable events, P.HistoryEvent.EventTypeOneofCase eventType) + static void IndexRequestedEvent( + Dictionary? index, + ref HashSet? duplicateEventIds, + P.HistoryEvent historyEvent) { - Dictionary index = new(); - foreach (P.HistoryEvent historyEvent in events) + if (index is null + || !index.TryGetValue(historyEvent.EventId, out P.HistoryEvent? existingEvent)) { - if (historyEvent.EventTypeCase != eventType) - { - continue; - } + return; + } - try - { - index.Add(historyEvent.EventId, historyEvent); - } - catch (ArgumentException exception) + if (existingEvent is null) + { + index[historyEvent.EventId] = historyEvent; + } + else + { + duplicateEventIds ??= new(); + duplicateEventIds.Add(historyEvent.EventId); + } + } + + static void ThrowIfDuplicate( + HashSet? duplicateEventIds, + int eventId, + P.HistoryEvent.EventTypeOneofCase eventType) + { + if (duplicateEventIds?.Contains(eventId) == true) + { + throw new InvalidOperationException( + $"Past orchestration history contains multiple '{eventType}' events with event ID '{eventId}'."); + } + } + + void BuildIndexes() + { + if (this.indexesBuilt) + { + return; + } + + foreach (P.HistoryEvent historyEvent in this.pastEvents) + { + switch (historyEvent.EventTypeCase) { - throw new InvalidOperationException( - $"Past orchestration history contains multiple '{eventType}' events with event ID " - + $"'{historyEvent.EventId}'.", - exception); + case P.HistoryEvent.EventTypeOneofCase.TaskScheduled: + IndexRequestedEvent( + this.taskScheduledEventsByEventId, + ref this.duplicateTaskScheduledEventIds, + historyEvent); + break; + + case P.HistoryEvent.EventTypeOneofCase.SubOrchestrationInstanceCreated: + IndexRequestedEvent( + this.subOrchestrationInstanceCreatedEventsByEventId, + ref this.duplicateSubOrchestrationInstanceCreatedEventIds, + historyEvent); + break; } } - return index; + this.indexesBuilt = true; } } diff --git a/src/Worker/Grpc/GrpcDurableTaskWorker.Processor.cs b/src/Worker/Grpc/GrpcDurableTaskWorker.Processor.cs index 44c403a4..54dfab0c 100644 --- a/src/Worker/Grpc/GrpcDurableTaskWorker.Processor.cs +++ b/src/Worker/Grpc/GrpcDurableTaskWorker.Processor.cs @@ -616,8 +616,8 @@ async Task OnRunOrchestratorAsync( if (executionStartedEvent is not null) { - // Build lookups once per work item instead of rescanning PastEvents for every new event. - TraceHistoryEventLookup historyLookup = new(request.PastEvents); + // Index only correlation IDs referenced by this work item's new events in one lazy history pass. + TraceHistoryEventLookup historyLookup = new(request.PastEvents, request.NewEvents); foreach (var newEvent in request.NewEvents) { diff --git a/test/Worker/Grpc.Tests/TraceHistoryEventLookupTests.cs b/test/Worker/Grpc.Tests/TraceHistoryEventLookupTests.cs index f7a2cfe4..8bcf4fab 100644 --- a/test/Worker/Grpc.Tests/TraceHistoryEventLookupTests.cs +++ b/test/Worker/Grpc.Tests/TraceHistoryEventLookupTests.cs @@ -17,7 +17,7 @@ public void GetTaskScheduledEvent_DuplicateEventIds_Throws() CreateTaskScheduled(eventId: 1, name: "FirstScheduled"), CreateTaskScheduled(eventId: 1, name: "SecondScheduled"), ]; - TraceHistoryEventLookup lookup = new(pastEvents); + TraceHistoryEventLookup lookup = CreateLookup(pastEvents, taskScheduledEventIds: [1]); // Act Action act = () => lookup.GetTaskScheduledEvent(1); @@ -36,7 +36,8 @@ public void GetSubOrchestrationInstanceCreatedEvent_DuplicateEventIds_Throws() CreateSubOrchestrationInstanceCreated(eventId: 2, name: "FirstSub"), CreateSubOrchestrationInstanceCreated(eventId: 2, name: "SecondSub"), ]; - TraceHistoryEventLookup lookup = new(pastEvents); + TraceHistoryEventLookup lookup = CreateLookup( + pastEvents, subOrchestrationInstanceCreatedEventIds: [2]); // Act Action act = () => lookup.GetSubOrchestrationInstanceCreatedEvent(2); @@ -51,7 +52,7 @@ public void GetTaskScheduledEvent_NoMatch_ReturnsNull() { // Arrange List pastEvents = [CreateTaskScheduled(eventId: 1, name: "Scheduled")]; - TraceHistoryEventLookup lookup = new(pastEvents); + TraceHistoryEventLookup lookup = CreateLookup(pastEvents, taskScheduledEventIds: [99]); // Act P.HistoryEvent? result = lookup.GetTaskScheduledEvent(99); @@ -65,7 +66,8 @@ public void GetSubOrchestrationInstanceCreatedEvent_NoMatch_ReturnsNull() { // Arrange List pastEvents = [CreateSubOrchestrationInstanceCreated(eventId: 2, name: "Sub")]; - TraceHistoryEventLookup lookup = new(pastEvents); + TraceHistoryEventLookup lookup = CreateLookup( + pastEvents, subOrchestrationInstanceCreatedEventIds: [99]); // Act P.HistoryEvent? result = lookup.GetSubOrchestrationInstanceCreatedEvent(99); @@ -84,7 +86,10 @@ public void GetTaskScheduledEvent_IgnoresOtherEventTypesWithSameEventId() CreateSubOrchestrationInstanceCreated(eventId: 5, name: "Sub"), CreateTaskScheduled(eventId: 5, name: "Scheduled"), ]; - TraceHistoryEventLookup lookup = new(pastEvents); + TraceHistoryEventLookup lookup = CreateLookup( + pastEvents, + taskScheduledEventIds: [5], + subOrchestrationInstanceCreatedEventIds: [5]); // Act P.HistoryEvent? taskScheduled = lookup.GetTaskScheduledEvent(5); @@ -101,7 +106,7 @@ public void GetTaskScheduledEvent_IgnoresOtherEventTypesWithSameEventId() public void GetTaskScheduledEvent_EmptyPastEvents_ReturnsNull() { // Arrange - TraceHistoryEventLookup lookup = new([]); + TraceHistoryEventLookup lookup = CreateLookup([], taskScheduledEventIds: [0]); // Act P.HistoryEvent? result = lookup.GetTaskScheduledEvent(0); @@ -110,6 +115,184 @@ public void GetTaskScheduledEvent_EmptyPastEvents_ReturnsNull() result.Should().BeNull(); } + [Fact] + public void GetEvents_IndexesOnlyIdsReferencedByNewEvents() + { + // Arrange + List pastEvents = + [ + CreateTaskScheduled(eventId: 1, name: "UnreferencedTask1"), + CreateTaskScheduled(eventId: 1, name: "UnreferencedTask2"), + CreateSubOrchestrationInstanceCreated(eventId: 2, name: "UnreferencedSub1"), + CreateSubOrchestrationInstanceCreated(eventId: 2, name: "UnreferencedSub2"), + CreateTaskScheduled(eventId: 3, name: "ReferencedTask"), + CreateSubOrchestrationInstanceCreated(eventId: 4, name: "ReferencedSub"), + ]; + TraceHistoryEventLookup lookup = CreateLookup( + pastEvents, + taskScheduledEventIds: [3], + subOrchestrationInstanceCreatedEventIds: [4]); + + // Act + P.HistoryEvent? taskScheduled = lookup.GetTaskScheduledEvent(3); + P.HistoryEvent? subOrchestrationCreated = lookup.GetSubOrchestrationInstanceCreatedEvent(4); + + // Assert + taskScheduled!.TaskScheduled.Name.Should().Be("ReferencedTask"); + subOrchestrationCreated!.SubOrchestrationInstanceCreated.Name.Should().Be("ReferencedSub"); + } + + [Fact] + public void GetEventsOfBothTypes_EnumeratesPastEventsOnce() + { + // Arrange + int enumerationCount = 0; + TraceHistoryEventLookup lookup = CreateLookup( + EnumeratePastEvents(), + taskScheduledEventIds: [1], + subOrchestrationInstanceCreatedEventIds: [2]); + + // Act + P.HistoryEvent? taskScheduled = lookup.GetTaskScheduledEvent(1); + P.HistoryEvent? subOrchestrationCreated = lookup.GetSubOrchestrationInstanceCreatedEvent(2); + + // Assert + taskScheduled.Should().NotBeNull(); + subOrchestrationCreated.Should().NotBeNull(); + enumerationCount.Should().Be(1); + + IEnumerable EnumeratePastEvents() + { + enumerationCount++; + yield return CreateTaskScheduled(eventId: 1, name: "Scheduled"); + yield return CreateSubOrchestrationInstanceCreated(eventId: 2, name: "Sub"); + } + } + + [Fact] + public void Constructor_DoesNotEnumeratePastEvents() + { + // Arrange + int enumerationCount = 0; + + // Act + TraceHistoryEventLookup lookup = CreateLookup( + EnumeratePastEvents(), taskScheduledEventIds: [1]); + + // Assert + lookup.Should().NotBeNull(); + enumerationCount.Should().Be(0); + + IEnumerable EnumeratePastEvents() + { + enumerationCount++; + yield return CreateTaskScheduled(eventId: 1, name: "Scheduled"); + } + } + + [Fact] + public void GetEvents_RegistersFailureCorrelationIds() + { + // Arrange + List pastEvents = + [ + CreateTaskScheduled(eventId: 1, name: "Scheduled"), + CreateSubOrchestrationInstanceCreated(eventId: 2, name: "Sub"), + ]; + TraceHistoryEventLookup lookup = CreateLookup( + pastEvents, + taskScheduledEventIds: [1], + subOrchestrationInstanceCreatedEventIds: [2], + useFailureEvents: true); + + // Act + P.HistoryEvent? taskScheduled = lookup.GetTaskScheduledEvent(1); + P.HistoryEvent? subOrchestrationCreated = lookup.GetSubOrchestrationInstanceCreatedEvent(2); + + // Assert + taskScheduled!.TaskScheduled.Name.Should().Be("Scheduled"); + subOrchestrationCreated!.SubOrchestrationInstanceCreated.Name.Should().Be("Sub"); + } + + [Fact] + public void GetEvents_DuplicateRequestedId_ThrowsOnlyForThatIdAndType() + { + // Arrange + List pastEvents = + [ + CreateTaskScheduled(eventId: 1, name: "DuplicateTask1"), + CreateTaskScheduled(eventId: 1, name: "DuplicateTask2"), + CreateTaskScheduled(eventId: 3, name: "ValidTask"), + CreateSubOrchestrationInstanceCreated(eventId: 2, name: "DuplicateSub1"), + CreateSubOrchestrationInstanceCreated(eventId: 2, name: "DuplicateSub2"), + CreateSubOrchestrationInstanceCreated(eventId: 4, name: "ValidSub"), + ]; + TraceHistoryEventLookup lookup = CreateLookup( + pastEvents, + taskScheduledEventIds: [1, 3], + subOrchestrationInstanceCreatedEventIds: [2, 4]); + + // Act + P.HistoryEvent? validTask = lookup.GetTaskScheduledEvent(3); + P.HistoryEvent? validSub = lookup.GetSubOrchestrationInstanceCreatedEvent(4); + Action getDuplicateTask = () => lookup.GetTaskScheduledEvent(1); + Action getDuplicateSub = () => lookup.GetSubOrchestrationInstanceCreatedEvent(2); + + // Assert + validTask!.TaskScheduled.Name.Should().Be("ValidTask"); + validSub!.SubOrchestrationInstanceCreated.Name.Should().Be("ValidSub"); + getDuplicateTask.Should().Throw() + .WithMessage("*'TaskScheduled'*event ID '1'*"); + getDuplicateSub.Should().Throw() + .WithMessage("*'SubOrchestrationInstanceCreated'*event ID '2'*"); + } + + static TraceHistoryEventLookup CreateLookup( + IEnumerable pastEvents, + IEnumerable? taskScheduledEventIds = null, + IEnumerable? subOrchestrationInstanceCreatedEventIds = null, + bool useFailureEvents = false) + { + List newEvents = []; + if (taskScheduledEventIds is not null) + { + foreach (int eventId in taskScheduledEventIds) + { + P.HistoryEvent newEvent = useFailureEvents + ? new P.HistoryEvent + { + TaskFailed = new P.TaskFailedEvent { TaskScheduledId = eventId }, + } + : new P.HistoryEvent + { + TaskCompleted = new P.TaskCompletedEvent { TaskScheduledId = eventId }, + }; + newEvents.Add(newEvent); + } + } + + if (subOrchestrationInstanceCreatedEventIds is not null) + { + foreach (int eventId in subOrchestrationInstanceCreatedEventIds) + { + P.HistoryEvent newEvent = useFailureEvents + ? new P.HistoryEvent + { + SubOrchestrationInstanceFailed = + new P.SubOrchestrationInstanceFailedEvent { TaskScheduledId = eventId }, + } + : new P.HistoryEvent + { + SubOrchestrationInstanceCompleted = + new P.SubOrchestrationInstanceCompletedEvent { TaskScheduledId = eventId }, + }; + newEvents.Add(newEvent); + } + } + + return new TraceHistoryEventLookup(pastEvents, newEvents); + } + static P.HistoryEvent CreateTaskScheduled(int eventId, string name) { return new P.HistoryEvent From 0792af02ff1ac51afc0231bf02e25efdf3e8f798 Mon Sep 17 00:00:00 2001 From: Bernd Verst Date: Wed, 12 Aug 2026 20:18:36 -0700 Subject: [PATCH 5/5] Address tracing index code quality feedback Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> Copilot-Session: 01dc5b7c-c742-45b7-ae8c-5cda9bcfbd2c --- .../Grpc/Tracing/TraceHistoryEventLookup.cs | 4 ++-- .../Grpc.Tests/TraceHistoryEventLookupTests.cs | 18 ++++++------------ 2 files changed, 8 insertions(+), 14 deletions(-) diff --git a/src/Shared/Grpc/Tracing/TraceHistoryEventLookup.cs b/src/Shared/Grpc/Tracing/TraceHistoryEventLookup.cs index 52d3b093..497f314b 100644 --- a/src/Shared/Grpc/Tracing/TraceHistoryEventLookup.cs +++ b/src/Shared/Grpc/Tracing/TraceHistoryEventLookup.cs @@ -19,8 +19,8 @@ sealed class TraceHistoryEventLookup { readonly IEnumerable pastEvents; - Dictionary? taskScheduledEventsByEventId; - Dictionary? subOrchestrationInstanceCreatedEventsByEventId; + readonly Dictionary? taskScheduledEventsByEventId; + readonly Dictionary? subOrchestrationInstanceCreatedEventsByEventId; HashSet? duplicateTaskScheduledEventIds; HashSet? duplicateSubOrchestrationInstanceCreatedEventIds; diff --git a/test/Worker/Grpc.Tests/TraceHistoryEventLookupTests.cs b/test/Worker/Grpc.Tests/TraceHistoryEventLookupTests.cs index 8bcf4fab..960c01f5 100644 --- a/test/Worker/Grpc.Tests/TraceHistoryEventLookupTests.cs +++ b/test/Worker/Grpc.Tests/TraceHistoryEventLookupTests.cs @@ -256,9 +256,8 @@ static TraceHistoryEventLookup CreateLookup( List newEvents = []; if (taskScheduledEventIds is not null) { - foreach (int eventId in taskScheduledEventIds) - { - P.HistoryEvent newEvent = useFailureEvents + newEvents.AddRange( + taskScheduledEventIds.Select(eventId => useFailureEvents ? new P.HistoryEvent { TaskFailed = new P.TaskFailedEvent { TaskScheduledId = eventId }, @@ -266,16 +265,13 @@ static TraceHistoryEventLookup CreateLookup( : new P.HistoryEvent { TaskCompleted = new P.TaskCompletedEvent { TaskScheduledId = eventId }, - }; - newEvents.Add(newEvent); - } + })); } if (subOrchestrationInstanceCreatedEventIds is not null) { - foreach (int eventId in subOrchestrationInstanceCreatedEventIds) - { - P.HistoryEvent newEvent = useFailureEvents + newEvents.AddRange( + subOrchestrationInstanceCreatedEventIds.Select(eventId => useFailureEvents ? new P.HistoryEvent { SubOrchestrationInstanceFailed = @@ -285,9 +281,7 @@ static TraceHistoryEventLookup CreateLookup( { SubOrchestrationInstanceCompleted = new P.SubOrchestrationInstanceCompletedEvent { TaskScheduledId = eventId }, - }; - newEvents.Add(newEvent); - } + })); } return new TraceHistoryEventLookup(pastEvents, newEvents);