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 5dd18d52..54dfab0c 100644 --- a/src/Worker/Grpc/GrpcDurableTaskWorker.Processor.cs +++ b/src/Worker/Grpc/GrpcDurableTaskWorker.Processor.cs @@ -602,107 +602,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); + // Index only correlation IDs referenced by this work item's new events in one lazy history pass. + TraceHistoryEventLookup historyLookup = new(request.PastEvents, request.NewEvents); - 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/Grpc.IntegrationTests/TracingIntegrationTests.cs b/test/Grpc.IntegrationTests/TracingIntegrationTests.cs index 041c76de..f85d3ac0 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 = $"{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); + 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 816dfb87..f5402fdd 100644 --- a/test/Worker/Grpc.Tests/GrpcDurableTaskWorkerTests.cs +++ b/test/Worker/Grpc.Tests/GrpcDurableTaskWorkerTests.cs @@ -2,6 +2,7 @@ // Licensed under the MIT License. using System.Collections.Concurrent; +using System.Diagnostics; using System.IO; using System.Reflection; using DurableTask.Core; @@ -10,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.Logging; @@ -364,6 +366,148 @@ 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 + using ActivityListener listener = new() + { + ShouldListenTo = source => source.Name == "Microsoft.DurableTask", + Sample = (ref ActivityCreationOptions options) => ActivitySamplingResult.AllDataAndRecorded, + }; + ActivitySource.AddActivityListener(listener); + TraceHelper.HasListeners.Should().BeTrue(); + + P.WorkItem orchestratorWorkItem = CreateOrchestratorWorkItemWithDuplicateEventIds(); + + 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 + 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() { 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, + }, + }; + } +}