Perf: avoid repeated orchestration-history scans for tracing

src/Shared/Grpc/Tracing/TraceHelper.cs

@@ -20,6 +20,17 @@ static class TraceHelper
 
     static readonly ActivitySource ActivityTraceSource = new ActivitySource(Source);
 
+    /// <summary>
+    /// Gets a value indicating whether any listener is currently registered for the Durable Task
+    /// <see cref="ActivitySource"/>.
+    /// </summary>
+    /// <remarks>
+    /// 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
+    /// <see cref="Activity"/> would be discarded anyway.
+    /// </remarks>
+    public static bool HasListeners => ActivityTraceSource.HasListeners();
+
     /// <summary>
     /// Starts a new trace activity for scheduling an orchestration from the client.
     /// </summary>

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;
+
+/// <summary>
+/// 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").
+/// </summary>
+/// <remarks>
+/// 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.
+/// </remarks>
+sealed class TraceHistoryEventLookup
+{
+    readonly IEnumerable<P.HistoryEvent> pastEvents;
+
+    Dictionary<int, P.HistoryEvent>? taskScheduledEventsByEventId;
+    Dictionary<int, P.HistoryEvent>? subOrchestrationInstanceCreatedEventsByEventId;
+
+    /// <summary>
+    /// Initializes a new instance of the <see cref="TraceHistoryEventLookup"/> class.
+    /// </summary>
+    /// <param name="pastEvents">The past history events for the current orchestrator work item.</param>
+    public TraceHistoryEventLookup(IEnumerable<P.HistoryEvent> pastEvents)
+    {
+        this.pastEvents = pastEvents;
+    }
+
+    /// <summary>
+    /// Gets the "TaskScheduled" history event with the given event ID, if any.
+    /// </summary>
+    /// <param name="eventId">The event ID to look up.</param>
+    /// <returns>The matching event, or <see langword="null"/> if none is found.</returns>
+    /// <remarks>
+    /// If more than one "TaskScheduled" event shares the given event ID, the last one encountered (in history
+    /// order) is returned, matching the original <c>LastOrDefault</c> lookup semantics.
+    /// </remarks>
+    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;
+    }
+
+    /// <summary>
+    /// Gets the "SubOrchestrationInstanceCreated" history event with the given event ID, if any.
+    /// </summary>
+    /// <param name="eventId">The event ID to look up.</param>
+    /// <returns>The matching event, or <see langword="null"/> if none is found.</returns>
+    /// <remarks>
+    /// If more than one "SubOrchestrationInstanceCreated" event shares the given event ID, the first one
+    /// encountered (in history order) is returned, matching the original <c>FirstOrDefault</c> lookup semantics.
+    /// </remarks>
+    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<int, P.HistoryEvent> BuildIndex(
+        IEnumerable<P.HistoryEvent> events, P.HistoryEvent.EventTypeOneofCase eventType, bool keepFirst)
+    {
+        Dictionary<int, P.HistoryEvent> 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;
+    }
+}

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;