Perf: avoid repeated orchestration-history scans for tracing

src/Shared/Grpc/Tracing/TraceHelper.cs

@@ -20,6 +20,84 @@ 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>
+    /// Correlates new completion events with past scheduling events and emits their trace activities.
+    /// </summary>
+    /// <param name="instanceId">The ID of the associated orchestration.</param>
+    /// <param name="orchestrationName">The orchestration name used for timer activities.</param>
+    /// <param name="pastEvents">The past history events containing scheduling events.</param>
+    /// <param name="newEvents">The new history events to trace.</param>
+    public static void EmitTraceActivitiesForNewEvents(
+        string instanceId,
+        string orchestrationName,
+        IEnumerable<P.HistoryEvent> pastEvents,
+        IEnumerable<P.HistoryEvent> newEvents)
+    {
+        if (!HasListeners)
+        {
+            return;
+        }
+
+        TraceHistoryEventLookup historyLookup = new(pastEvents);
+        foreach (P.HistoryEvent newEvent in newEvents)
+        {
+            switch (newEvent.EventTypeCase)
+            {
+                case P.HistoryEvent.EventTypeOneofCase.SubOrchestrationInstanceCompleted:
+                    {
+                        P.HistoryEvent? scheduledEvent = historyLookup.GetSubOrchestrationInstanceCreatedEvent(
+                            newEvent.SubOrchestrationInstanceCompleted.TaskScheduledId);
+                        EmitTraceActivityForSubOrchestrationCompleted(
+                            instanceId, scheduledEvent, scheduledEvent?.SubOrchestrationInstanceCreated);
+                        break;
+                    }
+
+                case P.HistoryEvent.EventTypeOneofCase.SubOrchestrationInstanceFailed:
+                    {
+                        P.HistoryEvent? scheduledEvent = historyLookup.GetSubOrchestrationInstanceCreatedEvent(
+                            newEvent.SubOrchestrationInstanceFailed.TaskScheduledId);
+                        EmitTraceActivityForSubOrchestrationFailed(
+                            instanceId, scheduledEvent, scheduledEvent?.SubOrchestrationInstanceCreated,
+                            newEvent.SubOrchestrationInstanceFailed);
+                        break;
+                    }
+
+                case P.HistoryEvent.EventTypeOneofCase.TaskCompleted:
+                    {
+                        P.HistoryEvent? scheduledEvent = historyLookup.GetTaskScheduledEvent(
+                            newEvent.TaskCompleted.TaskScheduledId);
+                        EmitTraceActivityForTaskCompleted(instanceId, scheduledEvent, scheduledEvent?.TaskScheduled);
+                        break;
+                    }
+
+                case P.HistoryEvent.EventTypeOneofCase.TaskFailed:
+                    {
+                        P.HistoryEvent? scheduledEvent = historyLookup.GetTaskScheduledEvent(
+                            newEvent.TaskFailed.TaskScheduledId);
+                        EmitTraceActivityForTaskFailed(
+                            instanceId, scheduledEvent, scheduledEvent?.TaskScheduled, newEvent.TaskFailed);
+                        break;
+                    }
+
+                case P.HistoryEvent.EventTypeOneofCase.TimerFired:
+                    EmitTraceActivityForTimer(
+                        instanceId, orchestrationName, newEvent.Timestamp.ToDateTime(), newEvent.TimerFired);
+                    break;
+            }
+        }
+    }
+
     /// <summary>
     /// Starts a new trace activity for scheduling an orchestration from the client.
     /// </summary>

src/Shared/Grpc/Tracing/TraceHistoryEventLookup.cs

@@ -0,0 +1,84 @@
+// 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>
+/// Both indexes contain all past scheduling events of their respective types. They are populated together
+/// on the first lookup in one lazy scan and cached for the lifetime of the orchestrator work item.
+/// </remarks>
+sealed class TraceHistoryEventLookup
+{
+    readonly IEnumerable<P.HistoryEvent> pastEvents;
+
+    readonly Dictionary<int, P.HistoryEvent> taskScheduledEventsByEventId = new();
+    readonly Dictionary<int, P.HistoryEvent> subOrchestrationInstanceCreatedEventsByEventId = new();
+
+    bool indexesBuilt;
+
+    /// <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>
+    public P.HistoryEvent? GetTaskScheduledEvent(int eventId)
+    {
+        this.BuildIndexes();
+        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>
+    public P.HistoryEvent? GetSubOrchestrationInstanceCreatedEvent(int eventId)
+    {
+        this.BuildIndexes();
+        return this.subOrchestrationInstanceCreatedEventsByEventId.TryGetValue(
+                eventId, out P.HistoryEvent? historyEvent)
+            ? historyEvent
+            : null;
+    }
+
+    void BuildIndexes()
+    {
+        if (this.indexesBuilt)
+        {
+            return;
+        }
+
+        foreach (P.HistoryEvent historyEvent in this.pastEvents)
+        {
+            switch (historyEvent.EventTypeCase)
+            {
+                case P.HistoryEvent.EventTypeOneofCase.TaskScheduled:
+                    this.taskScheduledEventsByEventId.Add(historyEvent.EventId, historyEvent);
+                    break;
+
+                case P.HistoryEvent.EventTypeOneofCase.SubOrchestrationInstanceCreated:
+                    this.subOrchestrationInstanceCreatedEventsByEventId.Add(historyEvent.EventId, historyEvent);
+                    break;
+            }
+        }
+
+        this.indexesBuilt = true;
+    }
+}

src/Worker/Grpc/GrpcDurableTaskWorker.Processor.cs

@@ -659,13 +659,12 @@ async Task OnRunOrchestratorAsync(
             }
 
             IReadOnlyList<P.HistoryEvent> pastEvents = materializedPastEvents ?? request.PastEvents;
-            var executionStartedEvent =
-                request
-                    .NewEvents
-                    .Concat(pastEvents)
-                    .Where(e => e.EventTypeCase == P.HistoryEvent.EventTypeOneofCase.ExecutionStarted)
-                    .Select(e => e.ExecutionStarted)
-                    .FirstOrDefault();
+            bool hasTraceListeners = TraceHelper.HasListeners;
+            // Rewind validation needs the start event even without tracing. Ordinary work items
+            // skip this tracing-only lookup when no listener is registered.
+            P.ExecutionStartedEvent? executionStartedEvent = isInitialRewind || hasTraceListeners
+                ? FindExecutionStartedEvent(request, pastEvents)
+                : null;
 
             if (isInitialRewind)
             {
@@ -684,9 +683,11 @@ async Task OnRunOrchestratorAsync(
             // A rewind starts a new orchestration span instead of continuing the failed execution's stored span.
             P.OrchestrationTraceContext? orchestrationTraceContext =
                 isInitialRewind ? null : request.OrchestrationTraceContext;
-            Activity? traceActivity = TraceHelper.StartTraceActivityForOrchestrationExecution(
-                executionStartedEvent,
-                orchestrationTraceContext);
+            Activity? traceActivity = hasTraceListeners
+                ? TraceHelper.StartTraceActivityForOrchestrationExecution(
+                    executionStartedEvent,
+                    orchestrationTraceContext)
+                : null;
 
             if (isInitialRewind)
             {
@@ -701,93 +702,42 @@ await this.CompleteOrchestratorTaskWithChunkingAsync(
                 return;
             }
 
-            if (executionStartedEvent is not null)
+            try
             {
-                P.HistoryEvent? GetSuborchestrationInstanceCreatedEvent(int eventId)
+                if (hasTraceListeners && executionStartedEvent is not null)
                 {
-                    var subOrchestrationEvent =
-                        pastEvents
-                            .Where(x => x.EventTypeCase == P.HistoryEvent.EventTypeOneofCase.SubOrchestrationInstanceCreated)
-                            .FirstOrDefault(x => x.EventId == eventId);
-
-                    return subOrchestrationEvent;
+                    TraceHelper.EmitTraceActivitiesForNewEvents(
+                        request.InstanceId, executionStartedEvent.Name, pastEvents, request.NewEvents);
                 }
+            }
+            catch
+            {
+                // Correlation failures abandon the work item before normal trace cleanup is reached.
+                traceActivity?.Dispose();
+                throw;
+            }
 
-                P.HistoryEvent? GetTaskScheduledEvent(int eventId)
+            static P.ExecutionStartedEvent? FindExecutionStartedEvent(
+                P.OrchestratorRequest request,
+                IReadOnlyList<P.HistoryEvent> pastEvents)
+            {
+                foreach (P.HistoryEvent newEvent in request.NewEvents)
                 {
-                    var taskScheduledEvent =
-                        pastEvents
-                            .Where(x => x.EventTypeCase == P.HistoryEvent.EventTypeOneofCase.TaskScheduled)
-                            .LastOrDefault(x => x.EventId == eventId);
-
-                    return taskScheduledEvent;
+                    if (newEvent.EventTypeCase == P.HistoryEvent.EventTypeOneofCase.ExecutionStarted)
+                    {
+                        return newEvent.ExecutionStarted;
+                    }
                 }
 
-                foreach (var newEvent in request.NewEvents)
+                foreach (P.HistoryEvent pastEvent in pastEvents)
                 {
-                    switch (newEvent.EventTypeCase)
+                    if (pastEvent.EventTypeCase == P.HistoryEvent.EventTypeOneofCase.ExecutionStarted)
                     {
-                        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;
-                            }
-
-                        case P.HistoryEvent.EventTypeOneofCase.TaskCompleted:
-                            {
-                                P.HistoryEvent? taskScheduledEvent =
-                                    GetTaskScheduledEvent(newEvent.TaskCompleted.TaskScheduledId);
-
-                                TraceHelper.EmitTraceActivityForTaskCompleted(
-                                    request.InstanceId,
-                                    taskScheduledEvent,
-                                    taskScheduledEvent?.TaskScheduled);
-                                break;
-                            }
-
-                        case P.HistoryEvent.EventTypeOneofCase.TaskFailed:
-                            {
-                                P.HistoryEvent? taskScheduledEvent =
-                                    GetTaskScheduledEvent(newEvent.TaskFailed.TaskScheduledId);
-
-                                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;
+                        return pastEvent.ExecutionStarted;
                     }
                 }
+
+                return null;
             }
 
             OrchestratorExecutionResult? result = null;