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..497f314b --- /dev/null +++ b/src/Shared/Grpc/Tracing/TraceHistoryEventLookup.cs @@ -0,0 +1,177 @@ +// 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 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; + + readonly Dictionary? taskScheduledEventsByEventId; + readonly 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. + /// 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; + } + } + } + + /// + /// 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. + /// + /// Thrown when more than one "TaskScheduled" event has the same event ID. + /// + public P.HistoryEvent? GetTaskScheduledEvent(int eventId) + { + 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; + } + + /// + /// 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. + /// + /// Thrown when more than one "SubOrchestrationInstanceCreated" event has the same event ID. + /// + public P.HistoryEvent? GetSubOrchestrationInstanceCreatedEvent(int eventId) + { + 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 void IndexRequestedEvent( + Dictionary? index, + ref HashSet? duplicateEventIds, + P.HistoryEvent historyEvent) + { + if (index is null + || !index.TryGetValue(historyEvent.EventId, out P.HistoryEvent? existingEvent)) + { + return; + } + + 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) + { + 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; + } + } + + this.indexesBuilt = true; + } +} diff --git a/src/Worker/Grpc/GrpcDurableTaskWorker.Processor.cs b/src/Worker/Grpc/GrpcDurableTaskWorker.Processor.cs index abfba18a..0d31213c 100644 --- a/src/Worker/Grpc/GrpcDurableTaskWorker.Processor.cs +++ b/src/Worker/Grpc/GrpcDurableTaskWorker.Processor.cs @@ -659,13 +659,12 @@ async Task OnRunOrchestratorAsync( } IReadOnlyList 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,108 @@ 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); + // Index only correlation IDs referenced by this work item's new events in one lazy history pass. + TraceHistoryEventLookup historyLookup = new(pastEvents, request.NewEvents); - return subOrchestrationEvent; - } - - P.HistoryEvent? GetTaskScheduledEvent(int eventId) - { - var taskScheduledEvent = - pastEvents - .Where(x => x.EventTypeCase == P.HistoryEvent.EventTypeOneofCase.TaskScheduled) - .LastOrDefault(x => x.EventId == eventId); - - 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; - } + } + } + } + } + catch + { + // Correlation failures abandon the work item before normal trace cleanup is reached. + traceActivity?.Dispose(); + throw; + } - case P.HistoryEvent.EventTypeOneofCase.TaskFailed: - { - P.HistoryEvent? taskScheduledEvent = - GetTaskScheduledEvent(newEvent.TaskFailed.TaskScheduledId); + static P.ExecutionStartedEvent? FindExecutionStartedEvent( + P.OrchestratorRequest request, + IReadOnlyList pastEvents) + { + 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 pastEvents) + { + if (pastEvent.EventTypeCase == P.HistoryEvent.EventTypeOneofCase.ExecutionStarted) + { + return pastEvent.ExecutionStarted; } } + + return null; } OrchestratorExecutionResult? result = null; diff --git a/test/Grpc.IntegrationTests/TracingIntegrationTests.cs b/test/Grpc.IntegrationTests/TracingIntegrationTests.cs index 041c76de..0fc0b753 100644 --- a/test/Grpc.IntegrationTests/TracingIntegrationTests.cs +++ b/test/Grpc.IntegrationTests/TracingIntegrationTests.cs @@ -38,6 +38,85 @@ 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 = $"{orchestratorName}.FirstActivity"; + string secondActivityName = $"{orchestratorName}.SecondActivity"; + string subOrchestratorName = $"{orchestratorName}.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); + Activity orchestrationActivity = activities.First( + activity => activity.Kind == ActivityKind.Server + && activity.Source.Name == CoreActivitySourceName + && activity.OperationName == $"orchestration:{orchestratorName}"); + List scheduledActivities = []; + foreach (string operationName in new[] + { + $"activity:{firstActivityName}", + $"activity:{secondActivityName}", + $"orchestration:{subOrchestratorName}", + }) + { + Activity scheduledActivity = activities.Should().ContainSingle( + activity => activity.Kind == ActivityKind.Client + && activity.Source.Name == CoreActivitySourceName + && activity.OperationName == operationName).Subject; + scheduledActivities.Add(scheduledActivity); + scheduledActivity.ParentSpanId.Should().Be(orchestrationActivity.SpanId); + scheduledActivity.TraceId.Should().Be(orchestrationActivity.TraceId); + + activities.Where( + activity => activity.Kind == ActivityKind.Server + && activity.Source.Name == CoreActivitySourceName + && activity.OperationName == operationName) + .Should().NotBeEmpty().And.AllSatisfy(activity => + { + activity.ParentSpanId.Should().Be(scheduledActivity.SpanId); + activity.TraceId.Should().Be(scheduledActivity.TraceId); + }); + } + + scheduledActivities.Select(activity => activity.SpanId).Distinct().Should().HaveCount(3); + } + [Fact] public async Task MultiTaskOrchestration() { diff --git a/test/Worker/Grpc.Tests/GrpcDurableTaskWorkerTests.cs b/test/Worker/Grpc.Tests/GrpcDurableTaskWorkerTests.cs index 61c3cdf8..ec3f3acd 100644 --- a/test/Worker/Grpc.Tests/GrpcDurableTaskWorkerTests.cs +++ b/test/Worker/Grpc.Tests/GrpcDurableTaskWorkerTests.cs @@ -11,6 +11,7 @@ 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.DependencyInjection; @@ -372,6 +373,162 @@ 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_DoesNotBuildHistoryIndexes() + { + // 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(); + + 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_DuplicateEventIds_AbandonsWorkItem() + { + // Arrange + TaskCompletionSource stoppedOrchestrationActivity = new(TaskCreationOptions.RunContinuationsAsynchronously); + using ActivityListener listener = new() + { + ShouldListenTo = source => source.Name == "Microsoft.DurableTask", + Sample = (ref ActivityCreationOptions options) => ActivitySamplingResult.AllDataAndRecorded, + ActivityStopped = activity => + { + if (activity.OperationName == "orchestration:TestOrchestration") + { + stoppedOrchestrationActivity.TrySetResult(activity); + } + }, + }; + ActivitySource.AddActivityListener(listener); + TraceHelper.HasListeners.Should().BeTrue(); + + P.WorkItem orchestratorWorkItem = CreateOrchestratorWorkItemWithDuplicateEventIds(); + orchestratorWorkItem.OrchestratorRequest.PastEvents[0].ExecutionStarted.ParentTraceContext = new P.TraceContext + { + TraceParent = "00-11111111111111111111111111111111-2222222222222222-01", + }; + + TaskCompletionSource abandoned = new(TaskCreationOptions.RunContinuationsAsynchronously); + GrpcDurableTaskWorker worker = CreateActivityWorker(new GrpcDurableTaskWorkerOptions()); + Mock clientMock = new( + MockBehavior.Strict, + new object[] { Mock.Of() }); + clientMock + .Setup(client => client.AbandonTaskOrchestratorWorkItemAsync( + It.Is( + request => request.CompletionToken == orchestratorWorkItem.CompletionToken), + It.IsAny(), + It.IsAny(), + It.IsAny())) + .Callback(() => abandoned.TrySetResult()) + .Returns(CreateUnaryCall(Task.FromResult(new P.AbandonOrchestrationTaskResponse()))); + object processor = CreateProcessor(worker, clientMock.Object); + + // Act + InvokeDispatchWorkItem(processor, orchestratorWorkItem, CancellationToken.None); + await abandoned.Task.WaitAsync(TimeSpan.FromSeconds(5)); + + // Assert + Activity stoppedActivity = await stoppedOrchestrationActivity.Task.WaitAsync(TimeSpan.FromSeconds(5)); + stoppedActivity.Kind.Should().Be(ActivityKind.Server); + clientMock.VerifyAll(); + clientMock.Verify( + client => client.CompleteOrchestratorTaskAsync( + It.IsAny(), + It.IsAny(), + It.IsAny(), + It.IsAny()), + Times.Never); + } + + 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() { @@ -672,14 +829,15 @@ public async Task BuildRuntimeStateAsync_MultiChunkHistoryStream_MaterializesEve runtimeState.ExecutionStartedEvent!.Name.Should().Be("TestOrchestration"); } - [Fact] - public async Task OnRunOrchestratorAsync_StreamingHistory_StartsTraceFromExecutionStarted() + [Theory] + [InlineData(false)] + [InlineData(true)] + public async Task OnRunOrchestratorAsync_StreamingHistory_StartsTraceAndCorrelatesEvents(bool correlateEvents) { // Arrange const string ParentTraceId = "11111111111111111111111111111111"; const string ParentSpanId = "2222222222222222"; - SequenceAsyncStreamReader historyReader = new( - new P.HistoryChunk + P.HistoryChunk startChunk = new() { Events = { @@ -702,7 +860,31 @@ public async Task OnRunOrchestratorAsync_StreamingHistory_StartsTraceFromExecuti }, }, }, - }); + }; + P.HistoryChunk schedulingChunk = new() + { + Events = + { + new P.HistoryEvent + { + EventId = 1, + Timestamp = Timestamp.FromDateTime(new DateTime(2025, 1, 1, 0, 0, 0, DateTimeKind.Utc)), + TaskScheduled = new P.TaskScheduledEvent { Name = "StreamedActivity" }, + }, + new P.HistoryEvent + { + EventId = 2, + Timestamp = Timestamp.FromDateTime(new DateTime(2025, 1, 1, 0, 0, 0, DateTimeKind.Utc)), + SubOrchestrationInstanceCreated = new P.SubOrchestrationInstanceCreatedEvent + { + Name = "StreamedSubOrchestration", + InstanceId = "child-instance", + }, + }, + }, + }; + SequenceAsyncStreamReader historyReader = new( + correlateEvents ? [startChunk, schedulingChunk] : [startChunk]); Mock clientMock = new( MockBehavior.Strict, @@ -753,13 +935,30 @@ public async Task OnRunOrchestratorAsync_StreamingHistory_StartsTraceFromExecuti }, }; - Activity? orchestrationActivity = null; + if (correlateEvents) + { + request.NewEvents.Add(new P.HistoryEvent + { + EventId = 3, + Timestamp = Timestamp.FromDateTime(new DateTime(2025, 1, 1, 0, 0, 1, DateTimeKind.Utc)), + TaskCompleted = new P.TaskCompletedEvent { TaskScheduledId = 1 }, + }); + request.NewEvents.Add(new P.HistoryEvent + { + EventId = 4, + Timestamp = Timestamp.FromDateTime(new DateTime(2025, 1, 1, 0, 0, 1, DateTimeKind.Utc)), + SubOrchestrationInstanceCompleted = + new P.SubOrchestrationInstanceCompletedEvent { TaskScheduledId = 2 }, + }); + } + + List startedActivities = []; using ActivityListener listener = new() { ShouldListenTo = source => source.Name == "Microsoft.DurableTask", Sample = (ref ActivityCreationOptions options) => ActivitySamplingResult.AllDataAndRecorded, - ActivityStarted = activity => orchestrationActivity = activity, + ActivityStarted = activity => startedActivities.Add(activity), }; ActivitySource.AddActivityListener(listener); @@ -771,9 +970,28 @@ await InvokeOnRunOrchestratorAsync( CancellationToken.None); // Assert - orchestrationActivity.Should().NotBeNull(); - orchestrationActivity!.TraceId.Should().Be(ActivityTraceId.CreateFromString(ParentTraceId.AsSpan())); + Activity orchestrationActivity = startedActivities.Should().ContainSingle( + activity => activity.Kind == ActivityKind.Server + && activity.OperationName == "orchestration:TestOrchestration").Subject; + orchestrationActivity.TraceId.Should().Be(ActivityTraceId.CreateFromString(ParentTraceId.AsSpan())); orchestrationActivity.ParentSpanId.Should().Be(ActivitySpanId.CreateFromString(ParentSpanId.AsSpan())); + if (correlateEvents) + { + foreach (string operationName in new[] { "activity:StreamedActivity", "orchestration:StreamedSubOrchestration" }) + { + Activity scheduledActivity = startedActivities.Should().ContainSingle( + activity => activity.Kind == ActivityKind.Client + && activity.OperationName == operationName).Subject; + scheduledActivity.ParentSpanId.Should().Be(orchestrationActivity.SpanId); + scheduledActivity.TraceId.Should().Be(orchestrationActivity.TraceId); + } + } + + clientMock.Verify(client => client.StreamInstanceHistory( + It.IsAny(), + It.IsAny(), + It.IsAny(), + It.IsAny()), Times.Once); clientMock.VerifyAll(); factoryMock.VerifyAll(); } @@ -1097,7 +1315,7 @@ await InvokeOnRunOrchestratorAsync( action.OrchestratorActionTypeCase.Should().Be( P.OrchestratorAction.OrchestratorActionTypeOneofCase.RewindOrchestration); P.HistoryEvent[] newHistory = action.RewindOrchestration.NewHistory.ToArray(); - + // Confirm the shape of the new history (failed tasks removed, etc.) newHistory.Should().HaveCount(8); newHistory.Should().NotContain(e => @@ -1118,7 +1336,7 @@ await InvokeOnRunOrchestratorAsync( newHistory.Should().ContainSingle(e => e.EventTypeCase == P.HistoryEvent.EventTypeOneofCase.OrchestratorStarted && e.EventId == 9); - + // Confirm the new execution ID and trace state of the rewritten ExecutionStartedEvent P.ExecutionStartedEvent rewrittenStart = newHistory .Single(e => e.EventTypeCase == P.HistoryEvent.EventTypeOneofCase.ExecutionStarted) @@ -1136,7 +1354,7 @@ await InvokeOnRunOrchestratorAsync( rewrittenStart.ParentTraceContext.TraceState.Should().Be( rewindParentTraceContext?.TraceState ?? executionStarted.ExecutionStarted.ParentTraceContext.TraceState); - + // Confirm the new trace information of the failed suborchestration P.SubOrchestrationInstanceCreatedEvent rewrittenFailedChild = newHistory .Single(e => @@ -1165,8 +1383,8 @@ await InvokeOnRunOrchestratorAsync( && e.EventId == 4) .SubOrchestrationInstanceCreated; rewrittenSuccessfulChild.ParentTraceContext.TraceParent - .Should().Be(successfulChildCreated.SubOrchestrationInstanceCreated.ParentTraceContext.TraceParent); - + .Should().Be(successfulChildCreated.SubOrchestrationInstanceCreated.ParentTraceContext.TraceParent); + // Confirm the presence of the rewind event and its reason P.HistoryEvent rewindEvent = newHistory.Should().ContainSingle(e => e.EventTypeCase == P.HistoryEvent.EventTypeOneofCase.ExecutionRewound diff --git a/test/Worker/Grpc.Tests/TraceHistoryEventLookupTests.cs b/test/Worker/Grpc.Tests/TraceHistoryEventLookupTests.cs new file mode 100644 index 00000000..960c01f5 --- /dev/null +++ b/test/Worker/Grpc.Tests/TraceHistoryEventLookupTests.cs @@ -0,0 +1,311 @@ +// 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_Throws() + { + // Arrange + List pastEvents = + [ + CreateTaskScheduled(eventId: 1, name: "FirstScheduled"), + CreateTaskScheduled(eventId: 1, name: "SecondScheduled"), + ]; + TraceHistoryEventLookup lookup = CreateLookup(pastEvents, taskScheduledEventIds: [1]); + + // Act + Action act = () => lookup.GetTaskScheduledEvent(1); + + // Assert + act.Should().Throw() + .WithMessage("*'TaskScheduled'*event ID '1'*"); + } + + [Fact] + public void GetSubOrchestrationInstanceCreatedEvent_DuplicateEventIds_Throws() + { + // Arrange + List pastEvents = + [ + CreateSubOrchestrationInstanceCreated(eventId: 2, name: "FirstSub"), + CreateSubOrchestrationInstanceCreated(eventId: 2, name: "SecondSub"), + ]; + TraceHistoryEventLookup lookup = CreateLookup( + pastEvents, subOrchestrationInstanceCreatedEventIds: [2]); + + // Act + Action act = () => lookup.GetSubOrchestrationInstanceCreatedEvent(2); + + // Assert + act.Should().Throw() + .WithMessage("*'SubOrchestrationInstanceCreated'*event ID '2'*"); + } + + [Fact] + public void GetTaskScheduledEvent_NoMatch_ReturnsNull() + { + // Arrange + List pastEvents = [CreateTaskScheduled(eventId: 1, name: "Scheduled")]; + TraceHistoryEventLookup lookup = CreateLookup(pastEvents, taskScheduledEventIds: [99]); + + // 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 = CreateLookup( + pastEvents, subOrchestrationInstanceCreatedEventIds: [99]); + + // 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 = CreateLookup( + pastEvents, + taskScheduledEventIds: [5], + subOrchestrationInstanceCreatedEventIds: [5]); + + // 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 = CreateLookup([], taskScheduledEventIds: [0]); + + // Act + P.HistoryEvent? result = lookup.GetTaskScheduledEvent(0); + + // Assert + 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) + { + newEvents.AddRange( + taskScheduledEventIds.Select(eventId => useFailureEvents + ? new P.HistoryEvent + { + TaskFailed = new P.TaskFailedEvent { TaskScheduledId = eventId }, + } + : new P.HistoryEvent + { + TaskCompleted = new P.TaskCompletedEvent { TaskScheduledId = eventId }, + })); + } + + if (subOrchestrationInstanceCreatedEventIds is not null) + { + newEvents.AddRange( + subOrchestrationInstanceCreatedEventIds.Select(eventId => useFailureEvents + ? new P.HistoryEvent + { + SubOrchestrationInstanceFailed = + new P.SubOrchestrationInstanceFailedEvent { TaskScheduledId = eventId }, + } + : new P.HistoryEvent + { + SubOrchestrationInstanceCompleted = + new P.SubOrchestrationInstanceCompletedEvent { TaskScheduledId = eventId }, + })); + } + + return new TraceHistoryEventLookup(pastEvents, newEvents); + } + + 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, + }, + }; + } +}