From 906a8e585ab28f43648a528d3f1f9a1f9a51aab3 Mon Sep 17 00:00:00 2001 From: Chris Gillum Date: Thu, 6 Nov 2025 21:51:59 -0800 Subject: [PATCH 1/3] Initial attempt to fix https://github.com/microsoft/durabletask-dotnet/issues/387 --- src/Shared/Grpc/ProtoUtils.cs | 2065 ++++++++--------- .../Core/Shims/DurableTaskShimFactory.cs | 15 + .../Shims/TaskOrchestrationContextWrapper.cs | 34 +- .../Core/Shims/TaskOrchestrationShim.cs | 29 + .../Grpc/GrpcDurableTaskWorker.Processor.cs | 18 +- src/Worker/Grpc/GrpcOrchestrationRunner.cs | 2 + .../OrchestrationPatterns.cs | 46 +- 7 files changed, 1160 insertions(+), 1049 deletions(-) diff --git a/src/Shared/Grpc/ProtoUtils.cs b/src/Shared/Grpc/ProtoUtils.cs index db803563..e7fb891c 100644 --- a/src/Shared/Grpc/ProtoUtils.cs +++ b/src/Shared/Grpc/ProtoUtils.cs @@ -1,309 +1,309 @@ -// Copyright (c) Microsoft Corporation. -// Licensed under the MIT License. - -using System.Buffers; -using System.Buffers.Text; +// Copyright (c) Microsoft Corporation. +// Licensed under the MIT License. + +using System.Buffers; +using System.Buffers.Text; using System.Collections; using System.Diagnostics; -using System.Diagnostics.CodeAnalysis; +using System.Diagnostics.CodeAnalysis; using System.Globalization; -using System.Text; +using System.Text; using System.Text.Json; -using DurableTask.Core; -using DurableTask.Core.Command; -using DurableTask.Core.Entities; -using DurableTask.Core.Entities.OperationFormat; -using DurableTask.Core.History; +using DurableTask.Core; +using DurableTask.Core.Command; +using DurableTask.Core.Entities; +using DurableTask.Core.Entities.OperationFormat; +using DurableTask.Core.History; using DurableTask.Core.Tracing; -using Google.Protobuf; +using Google.Protobuf; using Google.Protobuf.Collections; -using Google.Protobuf.WellKnownTypes; -using DTCore = DurableTask.Core; -using P = Microsoft.DurableTask.Protobuf; +using Google.Protobuf.WellKnownTypes; +using DTCore = DurableTask.Core; +using P = Microsoft.DurableTask.Protobuf; using TraceHelper = Microsoft.DurableTask.Tracing.TraceHelper; - -namespace Microsoft.DurableTask; - -/// -/// Protobuf utilities and helpers. -/// -static class ProtoUtils -{ - /// - /// Converts a history event from to . - /// - /// The proto history event to converter. - /// The converted history event. - /// When the provided history event type is not supported. - internal static HistoryEvent ConvertHistoryEvent(P.HistoryEvent proto) - { - return ConvertHistoryEvent(proto, conversionState: null); - } - - /// - /// Converts a history event from to , and performs - /// stateful conversions of entity-related events. - /// - /// The proto history event to converter. - /// State needed for converting entity-related history entries and actions. - /// The converted history event. - /// When the provided history event type is not supported. - internal static HistoryEvent ConvertHistoryEvent(P.HistoryEvent proto, EntityConversionState? conversionState) - { - Check.NotNull(proto); - HistoryEvent historyEvent; - switch (proto.EventTypeCase) - { - case P.HistoryEvent.EventTypeOneofCase.ContinueAsNew: - historyEvent = new ContinueAsNewEvent(proto.EventId, proto.ContinueAsNew.Input); - break; - case P.HistoryEvent.EventTypeOneofCase.ExecutionStarted: - OrchestrationInstance instance = proto.ExecutionStarted.OrchestrationInstance.ToCore(); - conversionState?.SetOrchestrationInstance(instance); - historyEvent = new ExecutionStartedEvent(proto.EventId, proto.ExecutionStarted.Input) - { - Name = proto.ExecutionStarted.Name, - Version = proto.ExecutionStarted.Version, - OrchestrationInstance = instance, - Tags = proto.ExecutionStarted.Tags, - ParentInstance = proto.ExecutionStarted.ParentInstance == null ? null : new ParentInstance - { - Name = proto.ExecutionStarted.ParentInstance.Name, - Version = proto.ExecutionStarted.ParentInstance.Version, - OrchestrationInstance = proto.ExecutionStarted.ParentInstance.OrchestrationInstance.ToCore(), - TaskScheduleId = proto.ExecutionStarted.ParentInstance.TaskScheduledId, - }, - ScheduledStartTime = proto.ExecutionStarted.ScheduledStartTimestamp?.ToDateTime(), - }; - break; - case P.HistoryEvent.EventTypeOneofCase.ExecutionCompleted: - historyEvent = new ExecutionCompletedEvent( - proto.EventId, - proto.ExecutionCompleted.Result, - proto.ExecutionCompleted.OrchestrationStatus.ToCore()); - break; - case P.HistoryEvent.EventTypeOneofCase.ExecutionTerminated: - historyEvent = new ExecutionTerminatedEvent(proto.EventId, proto.ExecutionTerminated.Input); - break; - case P.HistoryEvent.EventTypeOneofCase.ExecutionSuspended: - historyEvent = new ExecutionSuspendedEvent(proto.EventId, proto.ExecutionSuspended.Input); - break; - case P.HistoryEvent.EventTypeOneofCase.ExecutionResumed: - historyEvent = new ExecutionResumedEvent(proto.EventId, proto.ExecutionResumed.Input); - break; - case P.HistoryEvent.EventTypeOneofCase.TaskScheduled: - historyEvent = new TaskScheduledEvent( - proto.EventId, - proto.TaskScheduled.Name, - proto.TaskScheduled.Version, + +namespace Microsoft.DurableTask; + +/// +/// Protobuf utilities and helpers. +/// +static class ProtoUtils +{ + /// + /// Converts a history event from to . + /// + /// The proto history event to converter. + /// The converted history event. + /// When the provided history event type is not supported. + internal static HistoryEvent ConvertHistoryEvent(P.HistoryEvent proto) + { + return ConvertHistoryEvent(proto, conversionState: null); + } + + /// + /// Converts a history event from to , and performs + /// stateful conversions of entity-related events. + /// + /// The proto history event to converter. + /// State needed for converting entity-related history entries and actions. + /// The converted history event. + /// When the provided history event type is not supported. + internal static HistoryEvent ConvertHistoryEvent(P.HistoryEvent proto, EntityConversionState? conversionState) + { + Check.NotNull(proto); + HistoryEvent historyEvent; + switch (proto.EventTypeCase) + { + case P.HistoryEvent.EventTypeOneofCase.ContinueAsNew: + historyEvent = new ContinueAsNewEvent(proto.EventId, proto.ContinueAsNew.Input); + break; + case P.HistoryEvent.EventTypeOneofCase.ExecutionStarted: + OrchestrationInstance instance = proto.ExecutionStarted.OrchestrationInstance.ToCore(); + conversionState?.SetOrchestrationInstance(instance); + historyEvent = new ExecutionStartedEvent(proto.EventId, proto.ExecutionStarted.Input) + { + Name = proto.ExecutionStarted.Name, + Version = proto.ExecutionStarted.Version, + OrchestrationInstance = instance, + Tags = proto.ExecutionStarted.Tags, + ParentInstance = proto.ExecutionStarted.ParentInstance == null ? null : new ParentInstance + { + Name = proto.ExecutionStarted.ParentInstance.Name, + Version = proto.ExecutionStarted.ParentInstance.Version, + OrchestrationInstance = proto.ExecutionStarted.ParentInstance.OrchestrationInstance.ToCore(), + TaskScheduleId = proto.ExecutionStarted.ParentInstance.TaskScheduledId, + }, + ScheduledStartTime = proto.ExecutionStarted.ScheduledStartTimestamp?.ToDateTime(), + }; + break; + case P.HistoryEvent.EventTypeOneofCase.ExecutionCompleted: + historyEvent = new ExecutionCompletedEvent( + proto.EventId, + proto.ExecutionCompleted.Result, + proto.ExecutionCompleted.OrchestrationStatus.ToCore()); + break; + case P.HistoryEvent.EventTypeOneofCase.ExecutionTerminated: + historyEvent = new ExecutionTerminatedEvent(proto.EventId, proto.ExecutionTerminated.Input); + break; + case P.HistoryEvent.EventTypeOneofCase.ExecutionSuspended: + historyEvent = new ExecutionSuspendedEvent(proto.EventId, proto.ExecutionSuspended.Input); + break; + case P.HistoryEvent.EventTypeOneofCase.ExecutionResumed: + historyEvent = new ExecutionResumedEvent(proto.EventId, proto.ExecutionResumed.Input); + break; + case P.HistoryEvent.EventTypeOneofCase.TaskScheduled: + historyEvent = new TaskScheduledEvent( + proto.EventId, + proto.TaskScheduled.Name, + proto.TaskScheduled.Version, proto.TaskScheduled.Input) { Tags = proto.TaskScheduled.Tags, }; - break; - case P.HistoryEvent.EventTypeOneofCase.TaskCompleted: - historyEvent = new TaskCompletedEvent( - proto.EventId, - proto.TaskCompleted.TaskScheduledId, - proto.TaskCompleted.Result); - break; - case P.HistoryEvent.EventTypeOneofCase.TaskFailed: - historyEvent = new TaskFailedEvent( - proto.EventId, - proto.TaskFailed.TaskScheduledId, - reason: null, /* not supported */ - details: null, /* not supported */ - proto.TaskFailed.FailureDetails.ToCore()); - break; - case P.HistoryEvent.EventTypeOneofCase.SubOrchestrationInstanceCreated: - historyEvent = new SubOrchestrationInstanceCreatedEvent(proto.EventId) - { - Input = proto.SubOrchestrationInstanceCreated.Input, - InstanceId = proto.SubOrchestrationInstanceCreated.InstanceId, - Name = proto.SubOrchestrationInstanceCreated.Name, - Version = proto.SubOrchestrationInstanceCreated.Version, - }; - break; - case P.HistoryEvent.EventTypeOneofCase.SubOrchestrationInstanceCompleted: - historyEvent = new SubOrchestrationInstanceCompletedEvent( - proto.EventId, - proto.SubOrchestrationInstanceCompleted.TaskScheduledId, - proto.SubOrchestrationInstanceCompleted.Result); - break; - case P.HistoryEvent.EventTypeOneofCase.SubOrchestrationInstanceFailed: - historyEvent = new SubOrchestrationInstanceFailedEvent( - proto.EventId, - proto.SubOrchestrationInstanceFailed.TaskScheduledId, - reason: null /* not supported */, - details: null /* not supported */, - proto.SubOrchestrationInstanceFailed.FailureDetails.ToCore()); - break; - case P.HistoryEvent.EventTypeOneofCase.TimerCreated: - historyEvent = new TimerCreatedEvent( - proto.EventId, - proto.TimerCreated.FireAt.ToDateTime()); - break; - case P.HistoryEvent.EventTypeOneofCase.TimerFired: - historyEvent = new TimerFiredEvent( - eventId: -1, - proto.TimerFired.FireAt.ToDateTime()) - { - TimerId = proto.TimerFired.TimerId, - }; - break; - case P.HistoryEvent.EventTypeOneofCase.OrchestratorStarted: - historyEvent = new OrchestratorStartedEvent(proto.EventId); - break; - case P.HistoryEvent.EventTypeOneofCase.OrchestratorCompleted: - historyEvent = new OrchestratorCompletedEvent(proto.EventId); - break; - case P.HistoryEvent.EventTypeOneofCase.EventSent: - historyEvent = new EventSentEvent(proto.EventId) - { - InstanceId = proto.EventSent.InstanceId, - Name = proto.EventSent.Name, - Input = proto.EventSent.Input, - }; - break; - case P.HistoryEvent.EventTypeOneofCase.EventRaised: - historyEvent = new EventRaisedEvent(proto.EventId, proto.EventRaised.Input) - { - Name = proto.EventRaised.Name, - }; - break; - case P.HistoryEvent.EventTypeOneofCase.EntityOperationCalled: - historyEvent = EntityConversions.EncodeOperationCalled(proto, conversionState!.CurrentInstance); - conversionState?.EntityRequestIds.Add(proto.EntityOperationCalled.RequestId); - break; - case P.HistoryEvent.EventTypeOneofCase.EntityOperationSignaled: - historyEvent = EntityConversions.EncodeOperationSignaled(proto); - conversionState?.EntityRequestIds.Add(proto.EntityOperationSignaled.RequestId); - break; - case P.HistoryEvent.EventTypeOneofCase.EntityLockRequested: - historyEvent = EntityConversions.EncodeLockRequested(proto, conversionState!.CurrentInstance); - conversionState?.AddUnlockObligations(proto.EntityLockRequested); - break; - case P.HistoryEvent.EventTypeOneofCase.EntityUnlockSent: - historyEvent = EntityConversions.EncodeUnlockSent(proto, conversionState!.CurrentInstance); - conversionState?.RemoveUnlockObligation(proto.EntityUnlockSent.TargetInstanceId); - break; - case P.HistoryEvent.EventTypeOneofCase.EntityLockGranted: - historyEvent = EntityConversions.EncodeLockGranted(proto); - break; - case P.HistoryEvent.EventTypeOneofCase.EntityOperationCompleted: - historyEvent = EntityConversions.EncodeOperationCompleted(proto); - break; - case P.HistoryEvent.EventTypeOneofCase.EntityOperationFailed: - historyEvent = EntityConversions.EncodeOperationFailed(proto); - break; - case P.HistoryEvent.EventTypeOneofCase.GenericEvent: - historyEvent = new GenericEvent(proto.EventId, proto.GenericEvent.Data); - break; - case P.HistoryEvent.EventTypeOneofCase.HistoryState: - historyEvent = new HistoryStateEvent( - proto.EventId, - new OrchestrationState - { - OrchestrationInstance = new OrchestrationInstance - { - InstanceId = proto.HistoryState.OrchestrationState.InstanceId, - }, - Name = proto.HistoryState.OrchestrationState.Name, - Version = proto.HistoryState.OrchestrationState.Version, - ScheduledStartTime = proto.HistoryState.OrchestrationState.ScheduledStartTimestamp.ToDateTime(), - CreatedTime = proto.HistoryState.OrchestrationState.CreatedTimestamp.ToDateTime(), - LastUpdatedTime = proto.HistoryState.OrchestrationState.LastUpdatedTimestamp.ToDateTime(), - Input = proto.HistoryState.OrchestrationState.Input, - Output = proto.HistoryState.OrchestrationState.Output, - Status = proto.HistoryState.OrchestrationState.CustomStatus, - Tags = proto.HistoryState.OrchestrationState.Tags, - }); - break; + break; + case P.HistoryEvent.EventTypeOneofCase.TaskCompleted: + historyEvent = new TaskCompletedEvent( + proto.EventId, + proto.TaskCompleted.TaskScheduledId, + proto.TaskCompleted.Result); + break; + case P.HistoryEvent.EventTypeOneofCase.TaskFailed: + historyEvent = new TaskFailedEvent( + proto.EventId, + proto.TaskFailed.TaskScheduledId, + reason: null, /* not supported */ + details: null, /* not supported */ + proto.TaskFailed.FailureDetails.ToCore()); + break; + case P.HistoryEvent.EventTypeOneofCase.SubOrchestrationInstanceCreated: + historyEvent = new SubOrchestrationInstanceCreatedEvent(proto.EventId) + { + Input = proto.SubOrchestrationInstanceCreated.Input, + InstanceId = proto.SubOrchestrationInstanceCreated.InstanceId, + Name = proto.SubOrchestrationInstanceCreated.Name, + Version = proto.SubOrchestrationInstanceCreated.Version, + }; + break; + case P.HistoryEvent.EventTypeOneofCase.SubOrchestrationInstanceCompleted: + historyEvent = new SubOrchestrationInstanceCompletedEvent( + proto.EventId, + proto.SubOrchestrationInstanceCompleted.TaskScheduledId, + proto.SubOrchestrationInstanceCompleted.Result); + break; + case P.HistoryEvent.EventTypeOneofCase.SubOrchestrationInstanceFailed: + historyEvent = new SubOrchestrationInstanceFailedEvent( + proto.EventId, + proto.SubOrchestrationInstanceFailed.TaskScheduledId, + reason: null /* not supported */, + details: null /* not supported */, + proto.SubOrchestrationInstanceFailed.FailureDetails.ToCore()); + break; + case P.HistoryEvent.EventTypeOneofCase.TimerCreated: + historyEvent = new TimerCreatedEvent( + proto.EventId, + proto.TimerCreated.FireAt.ToDateTime()); + break; + case P.HistoryEvent.EventTypeOneofCase.TimerFired: + historyEvent = new TimerFiredEvent( + eventId: -1, + proto.TimerFired.FireAt.ToDateTime()) + { + TimerId = proto.TimerFired.TimerId, + }; + break; + case P.HistoryEvent.EventTypeOneofCase.OrchestratorStarted: + historyEvent = new OrchestratorStartedEvent(proto.EventId); + break; + case P.HistoryEvent.EventTypeOneofCase.OrchestratorCompleted: + historyEvent = new OrchestratorCompletedEvent(proto.EventId); + break; + case P.HistoryEvent.EventTypeOneofCase.EventSent: + historyEvent = new EventSentEvent(proto.EventId) + { + InstanceId = proto.EventSent.InstanceId, + Name = proto.EventSent.Name, + Input = proto.EventSent.Input, + }; + break; + case P.HistoryEvent.EventTypeOneofCase.EventRaised: + historyEvent = new EventRaisedEvent(proto.EventId, proto.EventRaised.Input) + { + Name = proto.EventRaised.Name, + }; + break; + case P.HistoryEvent.EventTypeOneofCase.EntityOperationCalled: + historyEvent = EntityConversions.EncodeOperationCalled(proto, conversionState!.CurrentInstance); + conversionState?.EntityRequestIds.Add(proto.EntityOperationCalled.RequestId); + break; + case P.HistoryEvent.EventTypeOneofCase.EntityOperationSignaled: + historyEvent = EntityConversions.EncodeOperationSignaled(proto); + conversionState?.EntityRequestIds.Add(proto.EntityOperationSignaled.RequestId); + break; + case P.HistoryEvent.EventTypeOneofCase.EntityLockRequested: + historyEvent = EntityConversions.EncodeLockRequested(proto, conversionState!.CurrentInstance); + conversionState?.AddUnlockObligations(proto.EntityLockRequested); + break; + case P.HistoryEvent.EventTypeOneofCase.EntityUnlockSent: + historyEvent = EntityConversions.EncodeUnlockSent(proto, conversionState!.CurrentInstance); + conversionState?.RemoveUnlockObligation(proto.EntityUnlockSent.TargetInstanceId); + break; + case P.HistoryEvent.EventTypeOneofCase.EntityLockGranted: + historyEvent = EntityConversions.EncodeLockGranted(proto); + break; + case P.HistoryEvent.EventTypeOneofCase.EntityOperationCompleted: + historyEvent = EntityConversions.EncodeOperationCompleted(proto); + break; + case P.HistoryEvent.EventTypeOneofCase.EntityOperationFailed: + historyEvent = EntityConversions.EncodeOperationFailed(proto); + break; + case P.HistoryEvent.EventTypeOneofCase.GenericEvent: + historyEvent = new GenericEvent(proto.EventId, proto.GenericEvent.Data); + break; + case P.HistoryEvent.EventTypeOneofCase.HistoryState: + historyEvent = new HistoryStateEvent( + proto.EventId, + new OrchestrationState + { + OrchestrationInstance = new OrchestrationInstance + { + InstanceId = proto.HistoryState.OrchestrationState.InstanceId, + }, + Name = proto.HistoryState.OrchestrationState.Name, + Version = proto.HistoryState.OrchestrationState.Version, + ScheduledStartTime = proto.HistoryState.OrchestrationState.ScheduledStartTimestamp.ToDateTime(), + CreatedTime = proto.HistoryState.OrchestrationState.CreatedTimestamp.ToDateTime(), + LastUpdatedTime = proto.HistoryState.OrchestrationState.LastUpdatedTimestamp.ToDateTime(), + Input = proto.HistoryState.OrchestrationState.Input, + Output = proto.HistoryState.OrchestrationState.Output, + Status = proto.HistoryState.OrchestrationState.CustomStatus, + Tags = proto.HistoryState.OrchestrationState.Tags, + }); + break; case P.HistoryEvent.EventTypeOneofCase.ExecutionRewound: historyEvent = new ExecutionRewoundEvent(proto.EventId); break; - default: - throw new NotSupportedException($"Deserialization of {proto.EventTypeCase} is not supported."); - } - - historyEvent.Timestamp = proto.Timestamp.ToDateTime(); - return historyEvent; - } - - /// - /// Converts a to a gRPC . - /// - /// The date-time to convert. - /// The gRPC timestamp. - internal static Timestamp ToTimestamp(this DateTime dateTime) - { - // The protobuf libraries require timestamps to be in UTC - if (dateTime.Kind == DateTimeKind.Unspecified) - { - dateTime = DateTime.SpecifyKind(dateTime, DateTimeKind.Utc); - } - else if (dateTime.Kind == DateTimeKind.Local) - { - dateTime = dateTime.ToUniversalTime(); - } - - return Timestamp.FromDateTime(dateTime); - } - - /// - /// Converts a to a gRPC . - /// - /// The date-time to convert. - /// The gRPC timestamp. - internal static Timestamp? ToTimestamp(this DateTime? dateTime) - => dateTime.HasValue ? dateTime.Value.ToTimestamp() : null; - - /// - /// Converts a to a gRPC . - /// - /// The date-time to convert. - /// The gRPC timestamp. - internal static Timestamp ToTimestamp(this DateTimeOffset dateTime) => Timestamp.FromDateTimeOffset(dateTime); - - /// - /// Converts a to a gRPC . - /// - /// The date-time to convert. - /// The gRPC timestamp. - internal static Timestamp? ToTimestamp(this DateTimeOffset? dateTime) - => dateTime.HasValue ? dateTime.Value.ToTimestamp() : null; - - /// - /// Constructs a . - /// - /// The orchestrator instance ID. + default: + throw new NotSupportedException($"Deserialization of {proto.EventTypeCase} is not supported."); + } + + historyEvent.Timestamp = proto.Timestamp.ToDateTime(); + return historyEvent; + } + + /// + /// Converts a to a gRPC . + /// + /// The date-time to convert. + /// The gRPC timestamp. + internal static Timestamp ToTimestamp(this DateTime dateTime) + { + // The protobuf libraries require timestamps to be in UTC + if (dateTime.Kind == DateTimeKind.Unspecified) + { + dateTime = DateTime.SpecifyKind(dateTime, DateTimeKind.Utc); + } + else if (dateTime.Kind == DateTimeKind.Local) + { + dateTime = dateTime.ToUniversalTime(); + } + + return Timestamp.FromDateTime(dateTime); + } + + /// + /// Converts a to a gRPC . + /// + /// The date-time to convert. + /// The gRPC timestamp. + internal static Timestamp? ToTimestamp(this DateTime? dateTime) + => dateTime.HasValue ? dateTime.Value.ToTimestamp() : null; + + /// + /// Converts a to a gRPC . + /// + /// The date-time to convert. + /// The gRPC timestamp. + internal static Timestamp ToTimestamp(this DateTimeOffset dateTime) => Timestamp.FromDateTimeOffset(dateTime); + + /// + /// Converts a to a gRPC . + /// + /// The date-time to convert. + /// The gRPC timestamp. + internal static Timestamp? ToTimestamp(this DateTimeOffset? dateTime) + => dateTime.HasValue ? dateTime.Value.ToTimestamp() : null; + + /// + /// Constructs a . + /// + /// The orchestrator instance ID. /// The orchestrator execution ID. - /// The orchestrator customer status or null if no custom status. - /// The orchestrator actions. - /// - /// The completion token for the work item. It must be the exact same - /// value that was provided by the corresponding that triggered the orchestrator execution. - /// - /// The entity conversion state, or null if no conversion is required. + /// The orchestrator customer status or null if no custom status. + /// The orchestrator actions. + /// + /// The completion token for the work item. It must be the exact same + /// value that was provided by the corresponding that triggered the orchestrator execution. + /// + /// The entity conversion state, or null if no conversion is required. /// The that represents orchestration execution. /// Whether or not a history is required to complete the orchestration request and none was provided. - /// The orchestrator response. - /// When an orchestrator action is unknown. - internal static P.OrchestratorResponse ConstructOrchestratorResponse( - string instanceId, + /// The orchestrator response. + /// When an orchestrator action is unknown. + internal static P.OrchestratorResponse ConstructOrchestratorResponse( + string instanceId, string executionId, - string? customStatus, + string? customStatus, IEnumerable? actions, - string completionToken, + string completionToken, EntityConversionState? entityConversionState, Activity? orchestrationActivity, bool requiresHistory = false) - { - var response = new P.OrchestratorResponse - { - InstanceId = instanceId, - CustomStatus = customStatus, - CompletionToken = completionToken, + { + var response = new P.OrchestratorResponse + { + InstanceId = instanceId, + CustomStatus = customStatus, + CompletionToken = completionToken, OrchestrationTraceContext = new() { @@ -311,8 +311,8 @@ internal static P.OrchestratorResponse ConstructOrchestratorResponse( SpanStartTime = orchestrationActivity?.StartTimeUtc.ToTimestamp(), }, RequiresHistory = requiresHistory, - }; - + }; + // If a history is required and the orchestration request was not completed, then there is no list of actions. if (requiresHistory) { @@ -320,10 +320,10 @@ internal static P.OrchestratorResponse ConstructOrchestratorResponse( } Check.NotNull(actions); - foreach (OrchestratorAction action in actions) - { - var protoAction = new P.OrchestratorAction { Id = action.Id }; - + foreach (OrchestratorAction action in actions) + { + var protoAction = new P.OrchestratorAction { Id = action.Id }; + P.TraceContext? CreateTraceContext() { if (orchestrationActivity is null) @@ -341,18 +341,18 @@ internal static P.OrchestratorResponse ConstructOrchestratorResponse( }; } - switch (action.OrchestratorActionType) - { - case OrchestratorActionType.ScheduleOrchestrator: - var scheduleTaskAction = (ScheduleTaskOrchestratorAction)action; + switch (action.OrchestratorActionType) + { + case OrchestratorActionType.ScheduleOrchestrator: + var scheduleTaskAction = (ScheduleTaskOrchestratorAction)action; - protoAction.ScheduleTask = new P.ScheduleTaskAction - { - Name = scheduleTaskAction.Name, - Version = scheduleTaskAction.Version, - Input = scheduleTaskAction.Input, + protoAction.ScheduleTask = new P.ScheduleTaskAction + { + Name = scheduleTaskAction.Name, + Version = scheduleTaskAction.Version, + Input = scheduleTaskAction.Input, ParentTraceContext = CreateTraceContext(), - }; + }; if (scheduleTaskAction.Tags != null) { @@ -362,114 +362,111 @@ internal static P.OrchestratorResponse ConstructOrchestratorResponse( } } - break; - case OrchestratorActionType.CreateSubOrchestration: - var subOrchestrationAction = (CreateSubOrchestrationAction)action; - protoAction.CreateSubOrchestration = new P.CreateSubOrchestrationAction - { - Input = subOrchestrationAction.Input, - InstanceId = subOrchestrationAction.InstanceId, - Name = subOrchestrationAction.Name, - Version = subOrchestrationAction.Version, + break; + case OrchestratorActionType.CreateSubOrchestration: + var subOrchestrationAction = (CreateSubOrchestrationAction)action; + protoAction.CreateSubOrchestration = new P.CreateSubOrchestrationAction + { + Input = subOrchestrationAction.Input, + InstanceId = subOrchestrationAction.InstanceId, + Name = subOrchestrationAction.Name, + Version = subOrchestrationAction.Version, ParentTraceContext = CreateTraceContext(), - }; - break; - case OrchestratorActionType.CreateTimer: - var createTimerAction = (CreateTimerOrchestratorAction)action; - protoAction.CreateTimer = new P.CreateTimerAction - { - FireAt = createTimerAction.FireAt.ToTimestamp(), - }; - break; - case OrchestratorActionType.SendEvent: - var sendEventAction = (SendEventOrchestratorAction)action; - if (sendEventAction.Instance == null) - { - throw new ArgumentException( - $"{nameof(SendEventOrchestratorAction)} cannot have a null Instance property!"); - } - - if (entityConversionState is not null - && DTCore.Common.Entities.IsEntityInstance(sendEventAction.Instance.InstanceId) - && sendEventAction.EventName is not null - && sendEventAction.EventData is not null) - { - P.SendEntityMessageAction sendAction = new P.SendEntityMessageAction(); - protoAction.SendEntityMessage = sendAction; - - EntityConversions.DecodeEntityMessageAction( - sendEventAction.EventName, - sendEventAction.EventData, - sendEventAction.Instance.InstanceId, - sendAction, - out string requestId); - - entityConversionState.EntityRequestIds.Add(requestId); - - switch (sendAction.EntityMessageTypeCase) - { - case P.SendEntityMessageAction.EntityMessageTypeOneofCase.EntityLockRequested: - entityConversionState.AddUnlockObligations(sendAction.EntityLockRequested); - break; - case P.SendEntityMessageAction.EntityMessageTypeOneofCase.EntityUnlockSent: - entityConversionState.RemoveUnlockObligation(sendAction.EntityUnlockSent.TargetInstanceId); - break; - default: - break; - } - } - else - { - protoAction.SendEvent = new P.SendEventAction - { - Instance = sendEventAction.Instance.ToProtobuf(), - Name = sendEventAction.EventName, - Data = sendEventAction.EventData, - }; + }; + break; + case OrchestratorActionType.CreateTimer: + var createTimerAction = (CreateTimerOrchestratorAction)action; + protoAction.CreateTimer = new P.CreateTimerAction + { + FireAt = createTimerAction.FireAt.ToTimestamp(), + }; + break; + case OrchestratorActionType.SendEvent: + var sendEventAction = (SendEventOrchestratorAction)action; + if (sendEventAction.Instance == null) + { + throw new ArgumentException( + $"{nameof(SendEventOrchestratorAction)} cannot have a null Instance property!"); + } + + if (entityConversionState is not null + && DTCore.Common.Entities.IsEntityInstance(sendEventAction.Instance.InstanceId) + && sendEventAction.EventName is not null + && sendEventAction.EventData is not null) + { + P.SendEntityMessageAction sendAction = new(); + protoAction.SendEntityMessage = sendAction; + + EntityConversions.DecodeEntityMessageAction( + sendEventAction.EventName, + sendEventAction.EventData, + sendEventAction.Instance.InstanceId, + sendAction, + out string requestId); + + entityConversionState.EntityRequestIds.Add(requestId); + + switch (sendAction.EntityMessageTypeCase) + { + case P.SendEntityMessageAction.EntityMessageTypeOneofCase.EntityLockRequested: + entityConversionState.AddUnlockObligations(sendAction.EntityLockRequested); + break; + case P.SendEntityMessageAction.EntityMessageTypeOneofCase.EntityUnlockSent: + entityConversionState.RemoveUnlockObligation(sendAction.EntityUnlockSent.TargetInstanceId); + break; + default: + break; + } + } + else + { + protoAction.SendEvent = new P.SendEventAction + { + Instance = sendEventAction.Instance.ToProtobuf(), + Name = sendEventAction.EventName, + Data = sendEventAction.EventData, + }; // Distributed Tracing: start a new trace activity derived from the orchestration // for an EventRaisedEvent (external event) using Activity? traceActivity = TraceHelper.StartTraceActivityForEventRaisedFromWorker(sendEventAction, instanceId, executionId); traceActivity?.Stop(); - } - - break; - case OrchestratorActionType.OrchestrationComplete: - - if (entityConversionState is not null) - { - // as a precaution, unlock any entities that were not unlocked for some reason, before - // completing the orchestration. - foreach ((string target, string criticalSectionId) in entityConversionState.ResetObligations()) - { - response.Actions.Add(new P.OrchestratorAction - { - Id = action.Id, - SendEntityMessage = new P.SendEntityMessageAction - { - EntityUnlockSent = new P.EntityUnlockSentEvent - { - CriticalSectionId = criticalSectionId, - TargetInstanceId = target, - ParentInstanceId = entityConversionState.CurrentInstance?.InstanceId, - }, - }, - }); - } - } - - var completeAction = (OrchestrationCompleteOrchestratorAction)action; - protoAction.CompleteOrchestration = new P.CompleteOrchestrationAction - { - CarryoverEvents = - { - // TODO - }, - Details = completeAction.Details, - NewVersion = completeAction.NewVersion, - OrchestrationStatus = completeAction.OrchestrationStatus.ToProtobuf(), - Result = completeAction.Result, + } + + break; + case OrchestratorActionType.OrchestrationComplete: + + if (entityConversionState is not null) + { + // as a precaution, unlock any entities that were not unlocked for some reason, before + // completing the orchestration. + foreach ((string target, string criticalSectionId) in entityConversionState.ResetObligations()) + { + response.Actions.Add(new P.OrchestratorAction + { + Id = action.Id, + SendEntityMessage = new P.SendEntityMessageAction + { + EntityUnlockSent = new P.EntityUnlockSentEvent + { + CriticalSectionId = criticalSectionId, + TargetInstanceId = target, + ParentInstanceId = entityConversionState.CurrentInstance?.InstanceId, + }, + }, + }); + } + } + + var completeAction = (OrchestrationCompleteOrchestratorAction)action; + protoAction.CompleteOrchestration = new P.CompleteOrchestrationAction + { + CarryoverEvents = { completeAction.CarryoverEvents.Select(ToProtobuf) }, + Details = completeAction.Details, + NewVersion = completeAction.NewVersion, + OrchestrationStatus = completeAction.OrchestrationStatus.ToProtobuf(), + Result = completeAction.Result, }; foreach (KeyValuePair tag in completeAction.Tags) @@ -477,96 +474,96 @@ internal static P.OrchestratorResponse ConstructOrchestratorResponse( protoAction.CompleteOrchestration.Tags[tag.Key] = tag.Value; } - if (completeAction.OrchestrationStatus == OrchestrationStatus.Failed) - { - protoAction.CompleteOrchestration.FailureDetails = completeAction.FailureDetails.ToProtobuf(); - } - - break; - default: - throw new NotSupportedException($"Unknown orchestrator action: {action.OrchestratorActionType}"); - } - - response.Actions.Add(protoAction); - } - - return response; - } - - /// - /// Converts a to a . - /// - /// The status to convert. - /// The converted status. - internal static OrchestrationStatus ToCore(this P.OrchestrationStatus status) - { - return (OrchestrationStatus)status; - } - - /// - /// Converts a to a . - /// - /// The status to convert. - /// The converted status. - [return: NotNullIfNotNull(nameof(status))] - internal static OrchestrationInstance? ToCore(this P.OrchestrationInstance? status) - { - if (status == null) - { - return null; - } - - return new OrchestrationInstance - { - InstanceId = status.InstanceId, - ExecutionId = status.ExecutionId, - }; - } - - /// - /// Converts a to a . - /// - /// The failure details to convert. - /// The converted failure details. - [return: NotNullIfNotNull(nameof(failureDetails))] - internal static TaskFailureDetails? ToTaskFailureDetails(this P.TaskFailureDetails? failureDetails) - { - if (failureDetails == null) - { - return null; - } - - return new TaskFailureDetails( - failureDetails.ErrorType, - failureDetails.ErrorMessage, - failureDetails.StackTrace, + if (completeAction.OrchestrationStatus == OrchestrationStatus.Failed) + { + protoAction.CompleteOrchestration.FailureDetails = completeAction.FailureDetails.ToProtobuf(); + } + + break; + default: + throw new NotSupportedException($"Unknown orchestrator action: {action.OrchestratorActionType}"); + } + + response.Actions.Add(protoAction); + } + + return response; + } + + /// + /// Converts a to a . + /// + /// The status to convert. + /// The converted status. + internal static OrchestrationStatus ToCore(this P.OrchestrationStatus status) + { + return (OrchestrationStatus)status; + } + + /// + /// Converts a to a . + /// + /// The status to convert. + /// The converted status. + [return: NotNullIfNotNull(nameof(status))] + internal static OrchestrationInstance? ToCore(this P.OrchestrationInstance? status) + { + if (status == null) + { + return null; + } + + return new OrchestrationInstance + { + InstanceId = status.InstanceId, + ExecutionId = status.ExecutionId, + }; + } + + /// + /// Converts a to a . + /// + /// The failure details to convert. + /// The converted failure details. + [return: NotNullIfNotNull(nameof(failureDetails))] + internal static TaskFailureDetails? ToTaskFailureDetails(this P.TaskFailureDetails? failureDetails) + { + if (failureDetails == null) + { + return null; + } + + return new TaskFailureDetails( + failureDetails.ErrorType, + failureDetails.ErrorMessage, + failureDetails.StackTrace, failureDetails.InnerFailure.ToTaskFailureDetails(), ConvertProperties(failureDetails.Properties)); - } - - /// - /// Converts a to . - /// - /// The exception to convert. + } + + /// + /// Converts a to . + /// + /// The exception to convert. /// Optional exception properties provider. - /// The task failure details. - [return: NotNullIfNotNull(nameof(e))] + /// The task failure details. + [return: NotNullIfNotNull(nameof(e))] internal static P.TaskFailureDetails? ToTaskFailureDetails(this Exception? e, DTCore.IExceptionPropertiesProvider? exceptionPropertiesProvider = null) - { - if (e == null) - { - return null; - } - + { + if (e == null) + { + return null; + } + IDictionary? properties = exceptionPropertiesProvider?.GetExceptionProperties(e); var taskFailureDetails = new P.TaskFailureDetails - { - ErrorType = e.GetType().FullName, - ErrorMessage = e.Message, - StackTrace = e.StackTrace, + { + ErrorType = e.GetType().FullName, + ErrorMessage = e.Message, + StackTrace = e.StackTrace, InnerFailure = e.InnerException.ToTaskFailureDetails(exceptionPropertiesProvider), - }; + }; if (properties != null) { @@ -575,263 +572,263 @@ internal static OrchestrationStatus ToCore(this P.OrchestrationStatus status) taskFailureDetails.Properties[kvp.Key] = ConvertObjectToValue(kvp.Value); } } - + return taskFailureDetails; } - /// - /// Converts a to a . - /// - /// The entity batch request to convert. - /// The converted entity batch request. - [return: NotNullIfNotNull(nameof(entityBatchRequest))] - internal static EntityBatchRequest? ToEntityBatchRequest(this P.EntityBatchRequest? entityBatchRequest) - { - if (entityBatchRequest == null) - { - return null; - } - - return new EntityBatchRequest() - { - EntityState = entityBatchRequest.EntityState, - InstanceId = entityBatchRequest.InstanceId, - Operations = entityBatchRequest.Operations.Select(r => r.ToOperationRequest()).ToList(), - }; - } - - /// - /// Converts a to a . - /// - /// The entity request to convert. - /// The converted request. - /// Additional info about each operation, required by DTS. - internal static void ToEntityBatchRequest( - this P.EntityRequest entityRequest, - out EntityBatchRequest batchRequest, - out List operationInfos) - { - batchRequest = new EntityBatchRequest() - { - EntityState = entityRequest.EntityState, - InstanceId = entityRequest.InstanceId, - Operations = [], // operations are added to this collection below - }; - - operationInfos = new(entityRequest.OperationRequests.Count); - - foreach (P.HistoryEvent? op in entityRequest.OperationRequests) - { - if (op.EntityOperationSignaled is not null) - { - batchRequest.Operations.Add(new OperationRequest - { - Id = Guid.Parse(op.EntityOperationSignaled.RequestId), - Operation = op.EntityOperationSignaled.Operation, - Input = op.EntityOperationSignaled.Input, - }); - operationInfos.Add(new P.OperationInfo - { - RequestId = op.EntityOperationSignaled.RequestId, - ResponseDestination = null, // means we don't send back a response to the caller - }); - } - else if (op.EntityOperationCalled is not null) - { - batchRequest.Operations.Add(new OperationRequest - { - Id = Guid.Parse(op.EntityOperationCalled.RequestId), - Operation = op.EntityOperationCalled.Operation, - Input = op.EntityOperationCalled.Input, - }); - operationInfos.Add(new P.OperationInfo - { - RequestId = op.EntityOperationCalled.RequestId, - ResponseDestination = new P.OrchestrationInstance - { - InstanceId = op.EntityOperationCalled.ParentInstanceId, - ExecutionId = op.EntityOperationCalled.ParentExecutionId, - }, - }); - } - } - } - - /// - /// Converts a to a . - /// - /// The operation request to convert. - /// The converted operation request. - [return: NotNullIfNotNull(nameof(operationRequest))] - internal static OperationRequest? ToOperationRequest(this P.OperationRequest? operationRequest) - { - if (operationRequest == null) - { - return null; - } - - return new OperationRequest() - { - Operation = operationRequest.Operation, - Input = operationRequest.Input, - Id = Guid.Parse(operationRequest.RequestId), + /// + /// Converts a to a . + /// + /// The entity batch request to convert. + /// The converted entity batch request. + [return: NotNullIfNotNull(nameof(entityBatchRequest))] + internal static EntityBatchRequest? ToEntityBatchRequest(this P.EntityBatchRequest? entityBatchRequest) + { + if (entityBatchRequest == null) + { + return null; + } + + return new EntityBatchRequest() + { + EntityState = entityBatchRequest.EntityState, + InstanceId = entityBatchRequest.InstanceId, + Operations = entityBatchRequest.Operations.Select(r => r.ToOperationRequest()).ToList(), + }; + } + + /// + /// Converts a to a . + /// + /// The entity request to convert. + /// The converted request. + /// Additional info about each operation, required by DTS. + internal static void ToEntityBatchRequest( + this P.EntityRequest entityRequest, + out EntityBatchRequest batchRequest, + out List operationInfos) + { + batchRequest = new EntityBatchRequest() + { + EntityState = entityRequest.EntityState, + InstanceId = entityRequest.InstanceId, + Operations = [], // operations are added to this collection below + }; + + operationInfos = new(entityRequest.OperationRequests.Count); + + foreach (P.HistoryEvent? op in entityRequest.OperationRequests) + { + if (op.EntityOperationSignaled is not null) + { + batchRequest.Operations.Add(new OperationRequest + { + Id = Guid.Parse(op.EntityOperationSignaled.RequestId), + Operation = op.EntityOperationSignaled.Operation, + Input = op.EntityOperationSignaled.Input, + }); + operationInfos.Add(new P.OperationInfo + { + RequestId = op.EntityOperationSignaled.RequestId, + ResponseDestination = null, // means we don't send back a response to the caller + }); + } + else if (op.EntityOperationCalled is not null) + { + batchRequest.Operations.Add(new OperationRequest + { + Id = Guid.Parse(op.EntityOperationCalled.RequestId), + Operation = op.EntityOperationCalled.Operation, + Input = op.EntityOperationCalled.Input, + }); + operationInfos.Add(new P.OperationInfo + { + RequestId = op.EntityOperationCalled.RequestId, + ResponseDestination = new P.OrchestrationInstance + { + InstanceId = op.EntityOperationCalled.ParentInstanceId, + ExecutionId = op.EntityOperationCalled.ParentExecutionId, + }, + }); + } + } + } + + /// + /// Converts a to a . + /// + /// The operation request to convert. + /// The converted operation request. + [return: NotNullIfNotNull(nameof(operationRequest))] + internal static OperationRequest? ToOperationRequest(this P.OperationRequest? operationRequest) + { + if (operationRequest == null) + { + return null; + } + + return new OperationRequest() + { + Operation = operationRequest.Operation, + Input = operationRequest.Input, + Id = Guid.Parse(operationRequest.RequestId), TraceContext = operationRequest.TraceContext != null ? new DistributedTraceContext( operationRequest.TraceContext.TraceParent, operationRequest.TraceContext.TraceState) : null, - }; - } - - /// - /// Converts a to a . - /// - /// The operation result to convert. - /// The converted operation result. - [return: NotNullIfNotNull(nameof(operationResult))] - internal static OperationResult? ToOperationResult(this P.OperationResult? operationResult) - { - if (operationResult == null) - { - return null; - } - - switch (operationResult.ResultTypeCase) - { - case P.OperationResult.ResultTypeOneofCase.Success: - return new OperationResult() - { - Result = operationResult.Success.Result, - StartTimeUtc = operationResult.Success.StartTimeUtc?.ToDateTime(), - EndTimeUtc = operationResult.Success.EndTimeUtc?.ToDateTime(), - }; - - case P.OperationResult.ResultTypeOneofCase.Failure: - return new OperationResult() - { - FailureDetails = operationResult.Failure.FailureDetails.ToCore(), - StartTimeUtc = operationResult.Failure.StartTimeUtc?.ToDateTime(), - EndTimeUtc = operationResult.Failure.EndTimeUtc?.ToDateTime(), - }; - - default: - throw new NotSupportedException($"Deserialization of {operationResult.ResultTypeCase} is not supported."); - } - } - - /// - /// Converts a to . - /// - /// The operation result to convert. - /// The converted operation result. - [return: NotNullIfNotNull(nameof(operationResult))] - internal static P.OperationResult? ToOperationResult(this OperationResult? operationResult) - { - if (operationResult == null) - { - return null; - } - - if (operationResult.FailureDetails == null) - { - return new P.OperationResult() - { - Success = new P.OperationResultSuccess() - { - Result = operationResult.Result, + }; + } + + /// + /// Converts a to a . + /// + /// The operation result to convert. + /// The converted operation result. + [return: NotNullIfNotNull(nameof(operationResult))] + internal static OperationResult? ToOperationResult(this P.OperationResult? operationResult) + { + if (operationResult == null) + { + return null; + } + + switch (operationResult.ResultTypeCase) + { + case P.OperationResult.ResultTypeOneofCase.Success: + return new OperationResult() + { + Result = operationResult.Success.Result, + StartTimeUtc = operationResult.Success.StartTimeUtc?.ToDateTime(), + EndTimeUtc = operationResult.Success.EndTimeUtc?.ToDateTime(), + }; + + case P.OperationResult.ResultTypeOneofCase.Failure: + return new OperationResult() + { + FailureDetails = operationResult.Failure.FailureDetails.ToCore(), + StartTimeUtc = operationResult.Failure.StartTimeUtc?.ToDateTime(), + EndTimeUtc = operationResult.Failure.EndTimeUtc?.ToDateTime(), + }; + + default: + throw new NotSupportedException($"Deserialization of {operationResult.ResultTypeCase} is not supported."); + } + } + + /// + /// Converts a to . + /// + /// The operation result to convert. + /// The converted operation result. + [return: NotNullIfNotNull(nameof(operationResult))] + internal static P.OperationResult? ToOperationResult(this OperationResult? operationResult) + { + if (operationResult == null) + { + return null; + } + + if (operationResult.FailureDetails == null) + { + return new P.OperationResult() + { + Success = new P.OperationResultSuccess() + { + Result = operationResult.Result, StartTimeUtc = operationResult.StartTimeUtc?.ToTimestamp(), EndTimeUtc = operationResult.EndTimeUtc?.ToTimestamp(), - }, - }; - } - else - { - return new P.OperationResult() - { - Failure = new P.OperationResultFailure() - { - FailureDetails = ToProtobuf(operationResult.FailureDetails), + }, + }; + } + else + { + return new P.OperationResult() + { + Failure = new P.OperationResultFailure() + { + FailureDetails = ToProtobuf(operationResult.FailureDetails), StartTimeUtc = operationResult.StartTimeUtc?.ToTimestamp(), EndTimeUtc = operationResult.EndTimeUtc?.ToTimestamp(), - }, - }; - } - } - - /// - /// Converts a to a . - /// - /// The operation action to convert. - /// The converted operation action. - [return: NotNullIfNotNull(nameof(operationAction))] - internal static OperationAction? ToOperationAction(this P.OperationAction? operationAction) - { - if (operationAction == null) - { - return null; - } - - switch (operationAction.OperationActionTypeCase) - { - case P.OperationAction.OperationActionTypeOneofCase.SendSignal: - - return new SendSignalOperationAction() - { - Name = operationAction.SendSignal.Name, - Input = operationAction.SendSignal.Input, - InstanceId = operationAction.SendSignal.InstanceId, - ScheduledTime = operationAction.SendSignal.ScheduledTime?.ToDateTime(), + }, + }; + } + } + + /// + /// Converts a to a . + /// + /// The operation action to convert. + /// The converted operation action. + [return: NotNullIfNotNull(nameof(operationAction))] + internal static OperationAction? ToOperationAction(this P.OperationAction? operationAction) + { + if (operationAction == null) + { + return null; + } + + switch (operationAction.OperationActionTypeCase) + { + case P.OperationAction.OperationActionTypeOneofCase.SendSignal: + + return new SendSignalOperationAction() + { + Name = operationAction.SendSignal.Name, + Input = operationAction.SendSignal.Input, + InstanceId = operationAction.SendSignal.InstanceId, + ScheduledTime = operationAction.SendSignal.ScheduledTime?.ToDateTime(), RequestTime = operationAction.SendSignal.RequestTime?.ToDateTimeOffset(), ParentTraceContext = operationAction.SendSignal.ParentTraceContext != null ? new DistributedTraceContext( operationAction.SendSignal.ParentTraceContext.TraceParent, operationAction.SendSignal.ParentTraceContext.TraceState) : null, - }; - - case P.OperationAction.OperationActionTypeOneofCase.StartNewOrchestration: - - return new StartNewOrchestrationOperationAction() - { - Name = operationAction.StartNewOrchestration.Name, - Input = operationAction.StartNewOrchestration.Input, - InstanceId = operationAction.StartNewOrchestration.InstanceId, - Version = operationAction.StartNewOrchestration.Version, - ScheduledStartTime = operationAction.StartNewOrchestration.ScheduledTime?.ToDateTime(), + }; + + case P.OperationAction.OperationActionTypeOneofCase.StartNewOrchestration: + + return new StartNewOrchestrationOperationAction() + { + Name = operationAction.StartNewOrchestration.Name, + Input = operationAction.StartNewOrchestration.Input, + InstanceId = operationAction.StartNewOrchestration.InstanceId, + Version = operationAction.StartNewOrchestration.Version, + ScheduledStartTime = operationAction.StartNewOrchestration.ScheduledTime?.ToDateTime(), RequestTime = operationAction.StartNewOrchestration.RequestTime?.ToDateTimeOffset(), ParentTraceContext = operationAction.StartNewOrchestration.ParentTraceContext != null ? new DistributedTraceContext( operationAction.StartNewOrchestration.ParentTraceContext.TraceParent, operationAction.StartNewOrchestration.ParentTraceContext.TraceState) : null, - }; - default: - throw new NotSupportedException($"Deserialization of {operationAction.OperationActionTypeCase} is not supported."); - } - } - - /// - /// Converts a to . - /// - /// The operation action to convert. - /// The converted operation action. - [return: NotNullIfNotNull(nameof(operationAction))] - internal static P.OperationAction? ToOperationAction(this OperationAction? operationAction) - { - if (operationAction == null) - { - return null; - } - - var action = new P.OperationAction(); - - switch (operationAction) - { - case SendSignalOperationAction sendSignalAction: - - action.SendSignal = new P.SendSignalAction() - { - Name = sendSignalAction.Name, - Input = sendSignalAction.Input, - InstanceId = sendSignalAction.InstanceId, - ScheduledTime = sendSignalAction.ScheduledTime?.ToTimestamp(), + }; + default: + throw new NotSupportedException($"Deserialization of {operationAction.OperationActionTypeCase} is not supported."); + } + } + + /// + /// Converts a to . + /// + /// The operation action to convert. + /// The converted operation action. + [return: NotNullIfNotNull(nameof(operationAction))] + internal static P.OperationAction? ToOperationAction(this OperationAction? operationAction) + { + if (operationAction == null) + { + return null; + } + + var action = new P.OperationAction(); + + switch (operationAction) + { + case SendSignalOperationAction sendSignalAction: + + action.SendSignal = new P.SendSignalAction() + { + Name = sendSignalAction.Name, + Input = sendSignalAction.Input, + InstanceId = sendSignalAction.InstanceId, + ScheduledTime = sendSignalAction.ScheduledTime?.ToTimestamp(), RequestTime = sendSignalAction.RequestTime?.ToTimestamp(), ParentTraceContext = sendSignalAction.ParentTraceContext != null ? new P.TraceContext @@ -840,18 +837,18 @@ internal static void ToEntityBatchRequest( TraceState = sendSignalAction.ParentTraceContext.TraceState, } : null, - }; - break; - - case StartNewOrchestrationOperationAction startNewOrchestrationAction: - - action.StartNewOrchestration = new P.StartNewOrchestrationAction() - { - Name = startNewOrchestrationAction.Name, - Input = startNewOrchestrationAction.Input, - Version = startNewOrchestrationAction.Version, - InstanceId = startNewOrchestrationAction.InstanceId, - ScheduledTime = startNewOrchestrationAction.ScheduledStartTime?.ToTimestamp(), + }; + break; + + case StartNewOrchestrationOperationAction startNewOrchestrationAction: + + action.StartNewOrchestration = new P.StartNewOrchestrationAction() + { + Name = startNewOrchestrationAction.Name, + Input = startNewOrchestrationAction.Input, + Version = startNewOrchestrationAction.Version, + InstanceId = startNewOrchestrationAction.InstanceId, + ScheduledTime = startNewOrchestrationAction.ScheduledStartTime?.ToTimestamp(), RequestTime = startNewOrchestrationAction.RequestTime?.ToTimestamp(), ParentTraceContext = startNewOrchestrationAction.ParentTraceContext != null ? new P.TraceContext @@ -860,170 +857,170 @@ internal static void ToEntityBatchRequest( TraceState = startNewOrchestrationAction.ParentTraceContext.TraceState, } : null, - }; - break; - } - - return action; - } - - /// - /// Converts a to a . - /// - /// The operation result to convert. - /// The converted operation result. - [return: NotNullIfNotNull(nameof(entityBatchResult))] - internal static EntityBatchResult? ToEntityBatchResult(this P.EntityBatchResult? entityBatchResult) - { - if (entityBatchResult == null) - { - return null; - } - - return new EntityBatchResult() - { - Actions = entityBatchResult.Actions.Select(operationAction => operationAction!.ToOperationAction()).ToList(), - EntityState = entityBatchResult.EntityState, - Results = entityBatchResult.Results.Select(operationResult => operationResult!.ToOperationResult()).ToList(), - FailureDetails = entityBatchResult.FailureDetails.ToCore(), - }; - } - - /// - /// Converts a to . - /// - /// The operation result to convert. - /// The completion token, or null for the older protocol. - /// Additional information about each operation, required by DTS. - /// The converted operation result. - [return: NotNullIfNotNull(nameof(entityBatchResult))] - internal static P.EntityBatchResult? ToEntityBatchResult( - this EntityBatchResult? entityBatchResult, - string? completionToken = null, - IEnumerable? operationInfos = null) - { - if (entityBatchResult == null) - { - return null; - } - - return new P.EntityBatchResult() - { - EntityState = entityBatchResult.EntityState, - FailureDetails = entityBatchResult.FailureDetails.ToProtobuf(), - Actions = { entityBatchResult.Actions?.Select(a => a.ToOperationAction()) ?? [] }, - Results = { entityBatchResult.Results?.Select(a => a.ToOperationResult()) ?? [] }, - CompletionToken = completionToken ?? string.Empty, - OperationInfos = { operationInfos ?? [] }, - }; - } - - /// - /// Converts the gRPC representation of orchestrator entity parameters to the DT.Core representation. - /// - /// The DT.Core representation. - /// The gRPC representation. - [return: NotNullIfNotNull(nameof(parameters))] - internal static TaskOrchestrationEntityParameters? ToCore(this P.OrchestratorEntityParameters? parameters) - { - if (parameters == null) - { - return null; - } - - return new TaskOrchestrationEntityParameters() - { - EntityMessageReorderWindow = parameters.EntityMessageReorderWindow.ToTimeSpan(), - }; - } - - /// - /// Gets the approximate byte count for a . - /// - /// The failure details. - /// The approximate byte count. - internal static int GetApproximateByteCount(this P.TaskFailureDetails failureDetails) - { - // Protobuf strings are always UTF-8: https://developers.google.com/protocol-buffers/docs/proto3#scalar - Encoding encoding = Encoding.UTF8; - - int byteCount = 0; - if (failureDetails.ErrorType != null) - { - byteCount += encoding.GetByteCount(failureDetails.ErrorType); - } - - if (failureDetails.ErrorMessage != null) - { - byteCount += encoding.GetByteCount(failureDetails.ErrorMessage); - } - - if (failureDetails.StackTrace != null) - { - byteCount += encoding.GetByteCount(failureDetails.StackTrace); - } - - if (failureDetails.InnerFailure != null) - { - byteCount += failureDetails.InnerFailure.GetApproximateByteCount(); - } - - return byteCount; - } - - /// - /// Decode a protobuf message from a base64 string. - /// - /// The type to decode to. - /// The message parser. - /// The base64 encoded message. - /// The decoded message. - /// If decoding fails. - internal static T Base64Decode(this MessageParser parser, string encodedMessage) where T : IMessage - { - // Decode the base64 in a way that doesn't allocate a byte[] on each request - int encodedByteCount = Encoding.UTF8.GetByteCount(encodedMessage); - byte[] buffer = ArrayPool.Shared.Rent(encodedByteCount); - try - { - // The Base64 APIs require first converting the string into UTF-8 bytes. We then - // do an in-place conversion from base64 UTF-8 bytes to protobuf bytes so that - // we can finally decode the protobuf request. - Encoding.UTF8.GetBytes(encodedMessage, 0, encodedMessage.Length, buffer, 0); - OperationStatus status = Base64.DecodeFromUtf8InPlace( - buffer.AsSpan(0, encodedByteCount), - out int bytesWritten); - if (status != OperationStatus.Done) - { - throw new ArgumentException( - $"Failed to base64-decode the '{typeof(T).Name}' payload: {status}", nameof(encodedMessage)); - } - - return (T)parser.ParseFrom(buffer, 0, bytesWritten); - } - finally - { - ArrayPool.Shared.Return(buffer); - } - } - - /// - /// Converts a grpc to a . - /// - /// The failure details to convert. - /// The converted failure details. - internal static FailureDetails? ToCore(this P.TaskFailureDetails? failureDetails) - { - if (failureDetails == null) - { - return null; - } - - return new FailureDetails( - failureDetails.ErrorType, - failureDetails.ErrorMessage, - failureDetails.StackTrace, - failureDetails.InnerFailure.ToCore(), + }; + break; + } + + return action; + } + + /// + /// Converts a to a . + /// + /// The operation result to convert. + /// The converted operation result. + [return: NotNullIfNotNull(nameof(entityBatchResult))] + internal static EntityBatchResult? ToEntityBatchResult(this P.EntityBatchResult? entityBatchResult) + { + if (entityBatchResult == null) + { + return null; + } + + return new EntityBatchResult() + { + Actions = entityBatchResult.Actions.Select(operationAction => operationAction!.ToOperationAction()).ToList(), + EntityState = entityBatchResult.EntityState, + Results = entityBatchResult.Results.Select(operationResult => operationResult!.ToOperationResult()).ToList(), + FailureDetails = entityBatchResult.FailureDetails.ToCore(), + }; + } + + /// + /// Converts a to . + /// + /// The operation result to convert. + /// The completion token, or null for the older protocol. + /// Additional information about each operation, required by DTS. + /// The converted operation result. + [return: NotNullIfNotNull(nameof(entityBatchResult))] + internal static P.EntityBatchResult? ToEntityBatchResult( + this EntityBatchResult? entityBatchResult, + string? completionToken = null, + IEnumerable? operationInfos = null) + { + if (entityBatchResult == null) + { + return null; + } + + return new P.EntityBatchResult() + { + EntityState = entityBatchResult.EntityState, + FailureDetails = entityBatchResult.FailureDetails.ToProtobuf(), + Actions = { entityBatchResult.Actions?.Select(a => a.ToOperationAction()) ?? [] }, + Results = { entityBatchResult.Results?.Select(a => a.ToOperationResult()) ?? [] }, + CompletionToken = completionToken ?? string.Empty, + OperationInfos = { operationInfos ?? [] }, + }; + } + + /// + /// Converts the gRPC representation of orchestrator entity parameters to the DT.Core representation. + /// + /// The DT.Core representation. + /// The gRPC representation. + [return: NotNullIfNotNull(nameof(parameters))] + internal static TaskOrchestrationEntityParameters? ToCore(this P.OrchestratorEntityParameters? parameters) + { + if (parameters == null) + { + return null; + } + + return new TaskOrchestrationEntityParameters() + { + EntityMessageReorderWindow = parameters.EntityMessageReorderWindow.ToTimeSpan(), + }; + } + + /// + /// Gets the approximate byte count for a . + /// + /// The failure details. + /// The approximate byte count. + internal static int GetApproximateByteCount(this P.TaskFailureDetails failureDetails) + { + // Protobuf strings are always UTF-8: https://developers.google.com/protocol-buffers/docs/proto3#scalar + Encoding encoding = Encoding.UTF8; + + int byteCount = 0; + if (failureDetails.ErrorType != null) + { + byteCount += encoding.GetByteCount(failureDetails.ErrorType); + } + + if (failureDetails.ErrorMessage != null) + { + byteCount += encoding.GetByteCount(failureDetails.ErrorMessage); + } + + if (failureDetails.StackTrace != null) + { + byteCount += encoding.GetByteCount(failureDetails.StackTrace); + } + + if (failureDetails.InnerFailure != null) + { + byteCount += failureDetails.InnerFailure.GetApproximateByteCount(); + } + + return byteCount; + } + + /// + /// Decode a protobuf message from a base64 string. + /// + /// The type to decode to. + /// The message parser. + /// The base64 encoded message. + /// The decoded message. + /// If decoding fails. + internal static T Base64Decode(this MessageParser parser, string encodedMessage) where T : IMessage + { + // Decode the base64 in a way that doesn't allocate a byte[] on each request + int encodedByteCount = Encoding.UTF8.GetByteCount(encodedMessage); + byte[] buffer = ArrayPool.Shared.Rent(encodedByteCount); + try + { + // The Base64 APIs require first converting the string into UTF-8 bytes. We then + // do an in-place conversion from base64 UTF-8 bytes to protobuf bytes so that + // we can finally decode the protobuf request. + Encoding.UTF8.GetBytes(encodedMessage, 0, encodedMessage.Length, buffer, 0); + OperationStatus status = Base64.DecodeFromUtf8InPlace( + buffer.AsSpan(0, encodedByteCount), + out int bytesWritten); + if (status != OperationStatus.Done) + { + throw new ArgumentException( + $"Failed to base64-decode the '{typeof(T).Name}' payload: {status}", nameof(encodedMessage)); + } + + return (T)parser.ParseFrom(buffer, 0, bytesWritten); + } + finally + { + ArrayPool.Shared.Return(buffer); + } + } + + /// + /// Converts a grpc to a . + /// + /// The failure details to convert. + /// The converted failure details. + internal static FailureDetails? ToCore(this P.TaskFailureDetails? failureDetails) + { + if (failureDetails == null) + { + return null; + } + + return new FailureDetails( + failureDetails.ErrorType, + failureDetails.ErrorMessage, + failureDetails.StackTrace, + failureDetails.InnerFailure.ToCore(), failureDetails.IsNonRetriable, ConvertProperties(failureDetails.Properties)); } @@ -1123,28 +1120,28 @@ internal static Value ConvertObjectToValue(object? obj) // Fallback: convert unlisted type to string. _ => Value.ForString(obj.ToString() ?? string.Empty), }; - } - - /// - /// Converts a to a grpc . - /// - /// The failure details to convert. - /// The converted failure details. - static P.TaskFailureDetails? ToProtobuf(this FailureDetails? failureDetails) - { - if (failureDetails == null) - { - return null; - } - + } + + /// + /// Converts a to a grpc . + /// + /// The failure details to convert. + /// The converted failure details. + static P.TaskFailureDetails? ToProtobuf(this FailureDetails? failureDetails) + { + if (failureDetails == null) + { + return null; + } + var taskFailureDetails = new P.TaskFailureDetails - { - ErrorType = failureDetails.ErrorType ?? "(unknown)", - ErrorMessage = failureDetails.ErrorMessage ?? "(unknown)", - StackTrace = failureDetails.StackTrace, - IsNonRetriable = failureDetails.IsNonRetriable, - InnerFailure = failureDetails.InnerFailure.ToProtobuf(), - }; + { + ErrorType = failureDetails.ErrorType ?? "(unknown)", + ErrorMessage = failureDetails.ErrorMessage ?? "(unknown)", + StackTrace = failureDetails.StackTrace, + IsNonRetriable = failureDetails.IsNonRetriable, + InnerFailure = failureDetails.InnerFailure.ToProtobuf(), + }; // Properly populate the MapField if (failureDetails.Properties != null) @@ -1156,121 +1153,121 @@ internal static Value ConvertObjectToValue(object? obj) } return taskFailureDetails; - } - - static P.OrchestrationStatus ToProtobuf(this OrchestrationStatus status) - { - return (P.OrchestrationStatus)status; - } - - static P.OrchestrationInstance ToProtobuf(this OrchestrationInstance instance) - { - return new P.OrchestrationInstance - { - InstanceId = instance.InstanceId, - ExecutionId = instance.ExecutionId, - }; - } - - /// - /// Tracks state required for converting orchestration histories containing entity-related events. - /// - internal class EntityConversionState - { - readonly bool insertMissingEntityUnlocks; - - OrchestrationInstance? instance; - HashSet? entityRequestIds; - Dictionary? unlockObligations; - - /// - /// Initializes a new instance of the class. - /// - /// Whether to insert missing unlock events in to the history - /// when the orchestration completes. - public EntityConversionState(bool insertMissingEntityUnlocks) - { - this.ConvertFromProto = (P.HistoryEvent e) => ProtoUtils.ConvertHistoryEvent(e, this); - this.insertMissingEntityUnlocks = insertMissingEntityUnlocks; - } - - /// - /// Gets a function that converts a history event in protobuf format to a core history event. - /// - public Func ConvertFromProto { get; } - - /// - /// Gets the orchestration instance of this history. - /// - public OrchestrationInstance? CurrentInstance => this.instance; - - /// - /// Gets the set of guids that have been used as entity request ids in this history. - /// - public HashSet EntityRequestIds => this.entityRequestIds ??= new(); - - /// - /// Records the orchestration instance, which may be needed for some conversions. - /// - /// The orchestration instance. - public void SetOrchestrationInstance(OrchestrationInstance instance) - { - this.instance = instance; - } - - /// - /// Adds unlock obligations for all entities that are being locked by this request. - /// - /// The lock request. - public void AddUnlockObligations(P.EntityLockRequestedEvent request) - { - if (!this.insertMissingEntityUnlocks) - { - return; - } - - this.unlockObligations ??= new(); - - foreach (string target in request.LockSet) - { - this.unlockObligations[target] = request.CriticalSectionId; - } - } - - /// - /// Removes an unlock obligation. - /// - /// The target entity. - public void RemoveUnlockObligation(string target) - { - if (!this.insertMissingEntityUnlocks) - { - return; - } - - this.unlockObligations?.Remove(target); - } - - /// - /// Returns the remaining unlock obligations, and clears the list. - /// - /// The unlock obligations. - public IEnumerable<(string Target, string CriticalSectionId)> ResetObligations() - { - if (!this.insertMissingEntityUnlocks) - { - yield break; - } - - if (this.unlockObligations is not null) - { - foreach (var kvp in this.unlockObligations) - { - yield return (kvp.Key, kvp.Value); - } - - this.unlockObligations = null; - } - } - } -} + } + + static P.OrchestrationStatus ToProtobuf(this OrchestrationStatus status) + { + return (P.OrchestrationStatus)status; + } + + static P.OrchestrationInstance ToProtobuf(this OrchestrationInstance instance) + { + return new P.OrchestrationInstance + { + InstanceId = instance.InstanceId, + ExecutionId = instance.ExecutionId, + }; + } + + /// + /// Tracks state required for converting orchestration histories containing entity-related events. + /// + internal class EntityConversionState + { + readonly bool insertMissingEntityUnlocks; + + OrchestrationInstance? instance; + HashSet? entityRequestIds; + Dictionary? unlockObligations; + + /// + /// Initializes a new instance of the class. + /// + /// Whether to insert missing unlock events in to the history + /// when the orchestration completes. + public EntityConversionState(bool insertMissingEntityUnlocks) + { + this.ConvertFromProto = (P.HistoryEvent e) => ProtoUtils.ConvertHistoryEvent(e, this); + this.insertMissingEntityUnlocks = insertMissingEntityUnlocks; + } + + /// + /// Gets a function that converts a history event in protobuf format to a core history event. + /// + public Func ConvertFromProto { get; } + + /// + /// Gets the orchestration instance of this history. + /// + public OrchestrationInstance? CurrentInstance => this.instance; + + /// + /// Gets the set of guids that have been used as entity request ids in this history. + /// + public HashSet EntityRequestIds => this.entityRequestIds ??= new(); + + /// + /// Records the orchestration instance, which may be needed for some conversions. + /// + /// The orchestration instance. + public void SetOrchestrationInstance(OrchestrationInstance instance) + { + this.instance = instance; + } + + /// + /// Adds unlock obligations for all entities that are being locked by this request. + /// + /// The lock request. + public void AddUnlockObligations(P.EntityLockRequestedEvent request) + { + if (!this.insertMissingEntityUnlocks) + { + return; + } + + this.unlockObligations ??= new(); + + foreach (string target in request.LockSet) + { + this.unlockObligations[target] = request.CriticalSectionId; + } + } + + /// + /// Removes an unlock obligation. + /// + /// The target entity. + public void RemoveUnlockObligation(string target) + { + if (!this.insertMissingEntityUnlocks) + { + return; + } + + this.unlockObligations?.Remove(target); + } + + /// + /// Returns the remaining unlock obligations, and clears the list. + /// + /// The unlock obligations. + public IEnumerable<(string Target, string CriticalSectionId)> ResetObligations() + { + if (!this.insertMissingEntityUnlocks) + { + yield break; + } + + if (this.unlockObligations is not null) + { + foreach (var kvp in this.unlockObligations) + { + yield return (kvp.Key, kvp.Value); + } + + this.unlockObligations = null; + } + } + } +} diff --git a/src/Worker/Core/Shims/DurableTaskShimFactory.cs b/src/Worker/Core/Shims/DurableTaskShimFactory.cs index 3ae1cbc7..f4f12e03 100644 --- a/src/Worker/Core/Shims/DurableTaskShimFactory.cs +++ b/src/Worker/Core/Shims/DurableTaskShimFactory.cs @@ -38,6 +38,21 @@ public DurableTaskShimFactory( /// public static DurableTaskShimFactory Default { get; } = new(null, null); + /// + /// Used to do post-processing after an orchestration has executed. + /// + /// The orchestration that finished executing. + /// An invalid was provided. + public static void OnPostExecute(TaskOrchestration orchestration) + { + // This method should only be called on orchestrations created by the CreateOrchestration method above. + TaskOrchestrationShim shim = orchestration as TaskOrchestrationShim + ?? throw new ArgumentException("Invalid orchestration type", nameof(orchestration)); + + // If the orchestration continued-as-new, we need to reschedule buffered external events. + ////shim.TryRescheduleBufferedExternalEvents(); + } + /// /// Creates a from a . /// diff --git a/src/Worker/Core/Shims/TaskOrchestrationContextWrapper.cs b/src/Worker/Core/Shims/TaskOrchestrationContextWrapper.cs index c7090e0b..f08a7912 100644 --- a/src/Worker/Core/Shims/TaskOrchestrationContextWrapper.cs +++ b/src/Worker/Core/Shims/TaskOrchestrationContextWrapper.cs @@ -113,6 +113,11 @@ public override TaskOrchestrationEntityFeature Entities /// internal DataConverter DataConverter => this.invocationContext.Options.DataConverter; + /// + /// Gets a value indicating whether the orchestration should preserve unprocessed events when continuing as new. + /// + internal bool PreserveUnprocessedEvents { get; private set; } + /// protected override ILoggerFactory LoggerFactory => this.invocationContext.LoggerFactory; @@ -328,18 +333,7 @@ public override void SetCustomStatus(object? customStatus) public override void ContinueAsNew(object? newInput = null, bool preserveUnprocessedEvents = true) { this.innerContext.ContinueAsNew(newInput); - - if (preserveUnprocessedEvents) - { - // Send all the buffered external events to ourself. - OrchestrationInstance instance = new() { InstanceId = this.InstanceId }; - foreach ((string eventName, string eventPayload) in this.externalEventBuffer.TakeAll()) - { -#pragma warning disable CS0618 // Type or member is obsolete -- 'internal' usage. - this.innerContext.SendEvent(instance, eventName, new RawInput(eventPayload)); -#pragma warning restore CS0618 // Type or member is obsolete - } - } + this.PreserveUnprocessedEvents = preserveUnprocessedEvents; } /// @@ -401,6 +395,22 @@ and the goal is to generate a deterministic Guid. We cannot update to SHA2-based return new Guid(newGuidByteArray); } + /// + /// Reschedules all buffered external events to the current orchestration instance. + /// This is intended to be called when the orchestration continues as new. + /// + internal void RescheduleBufferedExternalEvents() + { + // Send all the buffered external events to ourself. + OrchestrationInstance instance = new() { InstanceId = this.InstanceId }; + foreach ((string eventName, string eventPayload) in this.externalEventBuffer.TakeAll()) + { +#pragma warning disable CS0618 // Type or member is obsolete -- 'internal' usage. + this.innerContext.SendEvent(instance, eventName, new RawInput(eventPayload)); +#pragma warning restore CS0618 // Type or member is obsolete + } + } + /// /// exits the critical section, if currently within a critical section. Otherwise, this has no effect. /// diff --git a/src/Worker/Core/Shims/TaskOrchestrationShim.cs b/src/Worker/Core/Shims/TaskOrchestrationShim.cs index eb7a179b..f7f51647 100644 --- a/src/Worker/Core/Shims/TaskOrchestrationShim.cs +++ b/src/Worker/Core/Shims/TaskOrchestrationShim.cs @@ -53,6 +53,8 @@ public TaskOrchestrationShim( this.logger = Logs.CreateWorkerLogger(this.invocationContext.LoggerFactory, "Orchestrations"); } + internal OrchestrationInvocationContext InvocationContext => this.invocationContext; + DataConverter DataConverter => this.invocationContext.Options.DataConverter; /// @@ -81,6 +83,11 @@ public TaskOrchestrationShim( this.logger.OrchestrationCompleted(instanceId, this.invocationContext.Name); } + if (this.wrapperContext.PreserveUnprocessedEvents) + { + this.wrapperContext.RescheduleBufferedExternalEvents(); + } + // Return the output (if any) as a serialized string. return this.DataConverter.Serialize(output); } @@ -118,4 +125,26 @@ public override void RaiseEvent(OrchestrationContext context, string name, strin { this.wrapperContext?.CompleteExternalEvent(name, input); } + + /// + /// Reschedules buffered external events if the orchestration continued-as-new and has unprocessed external events + /// in its internal buffer. + /// + /// true if unprocessed external events were scheduled; false otherwise. + /// This method was called prior to . + internal bool TryRescheduleBufferedExternalEvents() + { + if (this.wrapperContext is null) + { + throw new InvalidOperationException("RescheduleBufferedExternalEvents can only be called after Execute."); + } + + if (this.wrapperContext.PreserveUnprocessedEvents) + { + this.wrapperContext.RescheduleBufferedExternalEvents(); + return true; + } + + return false; + } } diff --git a/src/Worker/Grpc/GrpcDurableTaskWorker.Processor.cs b/src/Worker/Grpc/GrpcDurableTaskWorker.Processor.cs index 4dfa70f7..5a025085 100644 --- a/src/Worker/Grpc/GrpcDurableTaskWorker.Processor.cs +++ b/src/Worker/Grpc/GrpcDurableTaskWorker.Processor.cs @@ -218,10 +218,24 @@ async ValueTask BuildRuntimeStateAsync( pastEvents = orchestratorRequest.PastEvents.Select(converter); } - IEnumerable newEvents = orchestratorRequest.NewEvents.Select(converter); + List pastEventsList = pastEvents.ToList(); + this.Logger.LogDebug( + "{InstanceId}: Processing {Count} past event(s): {EventList}", + orchestratorRequest.InstanceId, + pastEventsList.Count, + string.Join(", ", pastEventsList.Select(e => e.EventType.ToString()))); + + IList newEvents = orchestratorRequest.NewEvents.Select(converter).ToList(); + + this.Logger.LogDebug( + "{InstanceId}: Processing {Count} new event(s): {EventList}", + orchestratorRequest.InstanceId, + newEvents.Count, + string.Join(", ", newEvents.Select(e => e.EventType.ToString()))); + // Reconstruct the orchestration state in a way that correctly distinguishes new events from past events - var runtimeState = new OrchestrationRuntimeState(pastEvents.ToList()); + var runtimeState = new OrchestrationRuntimeState(pastEventsList); foreach (HistoryEvent e in newEvents) { // AddEvent() puts events into the NewEvents list. diff --git a/src/Worker/Grpc/GrpcOrchestrationRunner.cs b/src/Worker/Grpc/GrpcOrchestrationRunner.cs index d8d04dea..fc1575c5 100644 --- a/src/Worker/Grpc/GrpcOrchestrationRunner.cs +++ b/src/Worker/Grpc/GrpcOrchestrationRunner.cs @@ -232,6 +232,8 @@ public static string LoadAndRun( { extendedSessions?.Remove(request.InstanceId); } + + DurableTaskShimFactory.OnPostExecute(shim); } } diff --git a/test/Grpc.IntegrationTests/OrchestrationPatterns.cs b/test/Grpc.IntegrationTests/OrchestrationPatterns.cs index b03ff679..500a1952 100644 --- a/test/Grpc.IntegrationTests/OrchestrationPatterns.cs +++ b/test/Grpc.IntegrationTests/OrchestrationPatterns.cs @@ -1145,7 +1145,51 @@ public ValueTask IsOrchestrationValidAsync(OrchestrationFilterParameters i } } + [Theory] + [InlineData(true)] + [InlineData(false)] + public async Task ContinueAsNewEventsArePreserved(bool injectTimers) + { + const int EventCount = 10; + async Task OrchestratorFunc(TaskOrchestrationContext ctx, int counter) + { + await ctx.WaitForExternalEvent("Event"); + counter++; + + if (injectTimers) + { + await ctx.CreateTimer(TimeSpan.FromMilliseconds(1), CancellationToken.None); + } + + if (counter < EventCount) + { + ctx.ContinueAsNew(counter, preserveUnprocessedEvents: true); + } + + return counter; + } + + await using HostTestLifetime server = await this.StartWorkerAsync( + b => b.AddTasks( + tasks => tasks.AddOrchestratorFunc(nameof(OrchestratorFunc), OrchestratorFunc))); + + string instanceId = await server.Client.ScheduleNewOrchestrationInstanceAsync( + nameof(OrchestratorFunc), + input: 0); + + for (int i = 0; i < EventCount; i++) + { + await server.Client.RaiseEventAsync(instanceId, eventName: "Event"); + await Task.Delay(TimeSpan.FromMilliseconds(1), this.TimeoutToken); + } + + OrchestrationMetadata metadata = await server.Client.WaitForInstanceCompletionAsync( + instanceId, getInputsAndOutputs: true, this.TimeoutToken); + Assert.NotNull(metadata); + Assert.Equal(OrchestrationRuntimeStatus.Completed, metadata.RuntimeStatus); + Assert.Equal(EventCount, metadata.ReadOutputAs()); + } + // TODO: Test for multiple external events with the same name - // TODO: Test for ContinueAsNew with external events that carry over // TODO: Test for catching activity exceptions of specific types } From 86256c008d2dc4d58c7cd252373d5cb4a70cbb0d Mon Sep 17 00:00:00 2001 From: Sophia Tevosyan Date: Mon, 10 Nov 2025 17:08:24 -0800 Subject: [PATCH 2/3] fixing the test dispatcher --- .../Dispatcher/TaskOrchestrationDispatcher.cs | 56 +++++++++---------- src/Shared/Grpc/ProtoUtils.cs | 22 ++++++++ .../Core/Shims/DurableTaskShimFactory.cs | 15 ----- .../Shims/TaskOrchestrationContextWrapper.cs | 34 ++++------- .../Core/Shims/TaskOrchestrationShim.cs | 29 ---------- .../Grpc/GrpcDurableTaskWorker.Processor.cs | 18 +----- src/Worker/Grpc/GrpcOrchestrationRunner.cs | 2 - 7 files changed, 61 insertions(+), 115 deletions(-) diff --git a/src/InProcessTestHost/Sidecar/Dispatcher/TaskOrchestrationDispatcher.cs b/src/InProcessTestHost/Sidecar/Dispatcher/TaskOrchestrationDispatcher.cs index f19ca532..3dfe627b 100644 --- a/src/InProcessTestHost/Sidecar/Dispatcher/TaskOrchestrationDispatcher.cs +++ b/src/InProcessTestHost/Sidecar/Dispatcher/TaskOrchestrationDispatcher.cs @@ -111,8 +111,12 @@ protected override async Task ExecuteWorkItemAsync(TaskOrchestrationWorkItem wor { throw new ArgumentException($"Could not find an orchestration instance ID in the work item's runtime state.", nameof(workItem)); } - - // We loop for as long as the orchestrator does a ContinueAsNew + + var activityMessages = new List(); + var timerMessages = new List(); + var orchestratorMessages = new List(); + + // We loop for as long as the orchestrator does a ContinueAsNew while (true) { if (this.log.IsEnabled(LogLevel.Debug)) @@ -138,9 +142,9 @@ protected override async Task ExecuteWorkItemAsync(TaskOrchestrationWorkItem wor this.ApplyOrchestratorActions( result, ref workItem.OrchestrationRuntimeState, - out List activityMessages, - out List orchestratorMessages, - out List timerMessages, + activityMessages, + orchestratorMessages, + timerMessages, out OrchestrationState? updatedStatus, out bool continueAsNew); if (continueAsNew) @@ -247,9 +251,9 @@ static string GetShortHistoryEventDescription(HistoryEvent e) void ApplyOrchestratorActions( OrchestratorExecutionResult result, ref OrchestrationRuntimeState runtimeState, - out List activityMessages, // CA1859: Use concrete types for better performance - out List orchestratorMessages, // CA1859: Use concrete types for better performance - out List timerMessages, // CA1859: Use concrete types for better performance + List activityMessages, // CA1859: Use concrete types for better performance + List orchestratorMessages, // CA1859: Use concrete types for better performance + List timerMessages, // CA1859: Use concrete types for better performance out OrchestrationState? updatedStatus, out bool continueAsNew) { @@ -258,9 +262,6 @@ void ApplyOrchestratorActions( throw new ArgumentException($"The provided {nameof(OrchestrationRuntimeState)} doesn't contain an instance ID!", nameof(runtimeState)); } - List? newActivityMessages = null; // CA1859: Use concrete types for better performance - List? newTimerMessages = null; // CA1859: Use concrete types for better performance - List? newOrchestratorMessages = null; // CA1859: Use concrete types for better performance FailureDetails? failureDetails = null; continueAsNew = false; @@ -288,8 +289,7 @@ void ApplyOrchestratorActions( scheduledEvent.ParentTraceContext ??= new(grpcAction.ParentTraceContext.TraceParent, grpcAction.ParentTraceContext.TraceState); } - newActivityMessages ??= new List(); - newActivityMessages.Add(new TaskMessage + activityMessages.Add(new TaskMessage { Event = scheduledEvent, OrchestrationInstance = runtimeState.OrchestrationInstance, @@ -301,8 +301,7 @@ void ApplyOrchestratorActions( { TimerCreatedEvent timerEvent = new(timerAction.Id, timerAction.FireAt); - newTimerMessages ??= new List(); - newTimerMessages.Add(new TaskMessage + timerMessages.Add(new TaskMessage { Event = new TimerFiredEvent(-1, timerAction.FireAt) { @@ -346,8 +345,7 @@ void ApplyOrchestratorActions( Tags = subOrchestrationAction.Tags, }; - newOrchestratorMessages ??= new List(); - newOrchestratorMessages.Add(new TaskMessage + orchestratorMessages.Add(new TaskMessage { Event = startedEvent, OrchestrationInstance = startedEvent.OrchestrationInstance, @@ -367,13 +365,17 @@ void ApplyOrchestratorActions( Input = sendEventAction.EventData, }; - runtimeState.AddEvent(sendEvent); + runtimeState.AddEvent(sendEvent); + + EventRaisedEvent eventRaisedEvent = new(-1, sendEventAction.EventData) + { + Name = sendEventAction.EventName + }; - newOrchestratorMessages ??= new List(); - newOrchestratorMessages.Add(new TaskMessage + orchestratorMessages.Add(new TaskMessage { - Event = sendEvent, - OrchestrationInstance = runtimeState.OrchestrationInstance, + Event = eventRaisedEvent, + OrchestrationInstance = sendEventAction.Instance, }); } else if (action is OrchestrationCompleteOrchestratorAction completeAction) @@ -408,9 +410,6 @@ void ApplyOrchestratorActions( } runtimeState = newRuntimeState; - activityMessages = new List(); - orchestratorMessages = new List(); - timerMessages = new List(); continueAsNew = true; updatedStatus = null; return; @@ -457,8 +456,7 @@ void ApplyOrchestratorActions( completeAction.FailureDetails); } - newOrchestratorMessages ??= new List(); - newOrchestratorMessages.Add(new TaskMessage + orchestratorMessages.Add(new TaskMessage { Event = subOrchestratorCompletedEvent, OrchestrationInstance = runtimeState.ParentInstance.OrchestrationInstance, @@ -475,10 +473,6 @@ void ApplyOrchestratorActions( runtimeState.AddEvent(new OrchestratorCompletedEvent(-1)); - activityMessages = newActivityMessages ?? new List(); - timerMessages = newTimerMessages ?? new List(); - orchestratorMessages = newOrchestratorMessages ?? new List(); - updatedStatus = new OrchestrationState { OrchestrationInstance = runtimeState.OrchestrationInstance, diff --git a/src/Shared/Grpc/ProtoUtils.cs b/src/Shared/Grpc/ProtoUtils.cs index e7fb891c..e9abf330 100644 --- a/src/Shared/Grpc/ProtoUtils.cs +++ b/src/Shared/Grpc/ProtoUtils.cs @@ -1169,6 +1169,28 @@ static P.OrchestrationInstance ToProtobuf(this OrchestrationInstance instance) }; } + static P.HistoryEvent ToProtobuf(HistoryEvent e) + { + var payload = new P.HistoryEvent() + { + EventId = e.EventId, + Timestamp = Timestamp.FromDateTime(e.Timestamp), + }; + + if (e.EventType == EventType.EventRaised) + { + var eventRaised = (EventRaisedEvent)e; + payload.EventRaised = new P.EventRaisedEvent + { + Name = eventRaised.Name, + Input = eventRaised.Input, + }; + return payload; + } + + throw new ArgumentException("Unsupported event type"); + } + /// /// Tracks state required for converting orchestration histories containing entity-related events. /// diff --git a/src/Worker/Core/Shims/DurableTaskShimFactory.cs b/src/Worker/Core/Shims/DurableTaskShimFactory.cs index f4f12e03..3ae1cbc7 100644 --- a/src/Worker/Core/Shims/DurableTaskShimFactory.cs +++ b/src/Worker/Core/Shims/DurableTaskShimFactory.cs @@ -38,21 +38,6 @@ public DurableTaskShimFactory( /// public static DurableTaskShimFactory Default { get; } = new(null, null); - /// - /// Used to do post-processing after an orchestration has executed. - /// - /// The orchestration that finished executing. - /// An invalid was provided. - public static void OnPostExecute(TaskOrchestration orchestration) - { - // This method should only be called on orchestrations created by the CreateOrchestration method above. - TaskOrchestrationShim shim = orchestration as TaskOrchestrationShim - ?? throw new ArgumentException("Invalid orchestration type", nameof(orchestration)); - - // If the orchestration continued-as-new, we need to reschedule buffered external events. - ////shim.TryRescheduleBufferedExternalEvents(); - } - /// /// Creates a from a . /// diff --git a/src/Worker/Core/Shims/TaskOrchestrationContextWrapper.cs b/src/Worker/Core/Shims/TaskOrchestrationContextWrapper.cs index f08a7912..c7090e0b 100644 --- a/src/Worker/Core/Shims/TaskOrchestrationContextWrapper.cs +++ b/src/Worker/Core/Shims/TaskOrchestrationContextWrapper.cs @@ -113,11 +113,6 @@ public override TaskOrchestrationEntityFeature Entities /// internal DataConverter DataConverter => this.invocationContext.Options.DataConverter; - /// - /// Gets a value indicating whether the orchestration should preserve unprocessed events when continuing as new. - /// - internal bool PreserveUnprocessedEvents { get; private set; } - /// protected override ILoggerFactory LoggerFactory => this.invocationContext.LoggerFactory; @@ -333,7 +328,18 @@ public override void SetCustomStatus(object? customStatus) public override void ContinueAsNew(object? newInput = null, bool preserveUnprocessedEvents = true) { this.innerContext.ContinueAsNew(newInput); - this.PreserveUnprocessedEvents = preserveUnprocessedEvents; + + if (preserveUnprocessedEvents) + { + // Send all the buffered external events to ourself. + OrchestrationInstance instance = new() { InstanceId = this.InstanceId }; + foreach ((string eventName, string eventPayload) in this.externalEventBuffer.TakeAll()) + { +#pragma warning disable CS0618 // Type or member is obsolete -- 'internal' usage. + this.innerContext.SendEvent(instance, eventName, new RawInput(eventPayload)); +#pragma warning restore CS0618 // Type or member is obsolete + } + } } /// @@ -395,22 +401,6 @@ and the goal is to generate a deterministic Guid. We cannot update to SHA2-based return new Guid(newGuidByteArray); } - /// - /// Reschedules all buffered external events to the current orchestration instance. - /// This is intended to be called when the orchestration continues as new. - /// - internal void RescheduleBufferedExternalEvents() - { - // Send all the buffered external events to ourself. - OrchestrationInstance instance = new() { InstanceId = this.InstanceId }; - foreach ((string eventName, string eventPayload) in this.externalEventBuffer.TakeAll()) - { -#pragma warning disable CS0618 // Type or member is obsolete -- 'internal' usage. - this.innerContext.SendEvent(instance, eventName, new RawInput(eventPayload)); -#pragma warning restore CS0618 // Type or member is obsolete - } - } - /// /// exits the critical section, if currently within a critical section. Otherwise, this has no effect. /// diff --git a/src/Worker/Core/Shims/TaskOrchestrationShim.cs b/src/Worker/Core/Shims/TaskOrchestrationShim.cs index f7f51647..eb7a179b 100644 --- a/src/Worker/Core/Shims/TaskOrchestrationShim.cs +++ b/src/Worker/Core/Shims/TaskOrchestrationShim.cs @@ -53,8 +53,6 @@ public TaskOrchestrationShim( this.logger = Logs.CreateWorkerLogger(this.invocationContext.LoggerFactory, "Orchestrations"); } - internal OrchestrationInvocationContext InvocationContext => this.invocationContext; - DataConverter DataConverter => this.invocationContext.Options.DataConverter; /// @@ -83,11 +81,6 @@ public TaskOrchestrationShim( this.logger.OrchestrationCompleted(instanceId, this.invocationContext.Name); } - if (this.wrapperContext.PreserveUnprocessedEvents) - { - this.wrapperContext.RescheduleBufferedExternalEvents(); - } - // Return the output (if any) as a serialized string. return this.DataConverter.Serialize(output); } @@ -125,26 +118,4 @@ public override void RaiseEvent(OrchestrationContext context, string name, strin { this.wrapperContext?.CompleteExternalEvent(name, input); } - - /// - /// Reschedules buffered external events if the orchestration continued-as-new and has unprocessed external events - /// in its internal buffer. - /// - /// true if unprocessed external events were scheduled; false otherwise. - /// This method was called prior to . - internal bool TryRescheduleBufferedExternalEvents() - { - if (this.wrapperContext is null) - { - throw new InvalidOperationException("RescheduleBufferedExternalEvents can only be called after Execute."); - } - - if (this.wrapperContext.PreserveUnprocessedEvents) - { - this.wrapperContext.RescheduleBufferedExternalEvents(); - return true; - } - - return false; - } } diff --git a/src/Worker/Grpc/GrpcDurableTaskWorker.Processor.cs b/src/Worker/Grpc/GrpcDurableTaskWorker.Processor.cs index 5a025085..4dfa70f7 100644 --- a/src/Worker/Grpc/GrpcDurableTaskWorker.Processor.cs +++ b/src/Worker/Grpc/GrpcDurableTaskWorker.Processor.cs @@ -218,24 +218,10 @@ async ValueTask BuildRuntimeStateAsync( pastEvents = orchestratorRequest.PastEvents.Select(converter); } - List pastEventsList = pastEvents.ToList(); - this.Logger.LogDebug( - "{InstanceId}: Processing {Count} past event(s): {EventList}", - orchestratorRequest.InstanceId, - pastEventsList.Count, - string.Join(", ", pastEventsList.Select(e => e.EventType.ToString()))); - - IList newEvents = orchestratorRequest.NewEvents.Select(converter).ToList(); - - this.Logger.LogDebug( - "{InstanceId}: Processing {Count} new event(s): {EventList}", - orchestratorRequest.InstanceId, - newEvents.Count, - string.Join(", ", newEvents.Select(e => e.EventType.ToString()))); - + IEnumerable newEvents = orchestratorRequest.NewEvents.Select(converter); // Reconstruct the orchestration state in a way that correctly distinguishes new events from past events - var runtimeState = new OrchestrationRuntimeState(pastEventsList); + var runtimeState = new OrchestrationRuntimeState(pastEvents.ToList()); foreach (HistoryEvent e in newEvents) { // AddEvent() puts events into the NewEvents list. diff --git a/src/Worker/Grpc/GrpcOrchestrationRunner.cs b/src/Worker/Grpc/GrpcOrchestrationRunner.cs index fc1575c5..d8d04dea 100644 --- a/src/Worker/Grpc/GrpcOrchestrationRunner.cs +++ b/src/Worker/Grpc/GrpcOrchestrationRunner.cs @@ -232,8 +232,6 @@ public static string LoadAndRun( { extendedSessions?.Remove(request.InstanceId); } - - DurableTaskShimFactory.OnPostExecute(shim); } } From 63a5055372a0df18fe9f61b90b9f086de4c69e66 Mon Sep 17 00:00:00 2001 From: Sophia Tevosyan Date: Mon, 10 Nov 2025 17:11:25 -0800 Subject: [PATCH 3/3] trying to remove the large diff for the ProtoUtils file --- src/Shared/Grpc/ProtoUtils.cs | 2052 ++++++++++++++++----------------- 1 file changed, 1026 insertions(+), 1026 deletions(-) diff --git a/src/Shared/Grpc/ProtoUtils.cs b/src/Shared/Grpc/ProtoUtils.cs index e9abf330..f7036d28 100644 --- a/src/Shared/Grpc/ProtoUtils.cs +++ b/src/Shared/Grpc/ProtoUtils.cs @@ -1,309 +1,309 @@ -// Copyright (c) Microsoft Corporation. -// Licensed under the MIT License. - -using System.Buffers; -using System.Buffers.Text; +// Copyright (c) Microsoft Corporation. +// Licensed under the MIT License. + +using System.Buffers; +using System.Buffers.Text; using System.Collections; using System.Diagnostics; -using System.Diagnostics.CodeAnalysis; +using System.Diagnostics.CodeAnalysis; using System.Globalization; -using System.Text; +using System.Text; using System.Text.Json; -using DurableTask.Core; -using DurableTask.Core.Command; -using DurableTask.Core.Entities; -using DurableTask.Core.Entities.OperationFormat; -using DurableTask.Core.History; +using DurableTask.Core; +using DurableTask.Core.Command; +using DurableTask.Core.Entities; +using DurableTask.Core.Entities.OperationFormat; +using DurableTask.Core.History; using DurableTask.Core.Tracing; -using Google.Protobuf; +using Google.Protobuf; using Google.Protobuf.Collections; -using Google.Protobuf.WellKnownTypes; -using DTCore = DurableTask.Core; -using P = Microsoft.DurableTask.Protobuf; +using Google.Protobuf.WellKnownTypes; +using DTCore = DurableTask.Core; +using P = Microsoft.DurableTask.Protobuf; using TraceHelper = Microsoft.DurableTask.Tracing.TraceHelper; - -namespace Microsoft.DurableTask; - -/// -/// Protobuf utilities and helpers. -/// -static class ProtoUtils -{ - /// - /// Converts a history event from to . - /// - /// The proto history event to converter. - /// The converted history event. - /// When the provided history event type is not supported. - internal static HistoryEvent ConvertHistoryEvent(P.HistoryEvent proto) - { - return ConvertHistoryEvent(proto, conversionState: null); - } - - /// - /// Converts a history event from to , and performs - /// stateful conversions of entity-related events. - /// - /// The proto history event to converter. - /// State needed for converting entity-related history entries and actions. - /// The converted history event. - /// When the provided history event type is not supported. - internal static HistoryEvent ConvertHistoryEvent(P.HistoryEvent proto, EntityConversionState? conversionState) - { - Check.NotNull(proto); - HistoryEvent historyEvent; - switch (proto.EventTypeCase) - { - case P.HistoryEvent.EventTypeOneofCase.ContinueAsNew: - historyEvent = new ContinueAsNewEvent(proto.EventId, proto.ContinueAsNew.Input); - break; - case P.HistoryEvent.EventTypeOneofCase.ExecutionStarted: - OrchestrationInstance instance = proto.ExecutionStarted.OrchestrationInstance.ToCore(); - conversionState?.SetOrchestrationInstance(instance); - historyEvent = new ExecutionStartedEvent(proto.EventId, proto.ExecutionStarted.Input) - { - Name = proto.ExecutionStarted.Name, - Version = proto.ExecutionStarted.Version, - OrchestrationInstance = instance, - Tags = proto.ExecutionStarted.Tags, - ParentInstance = proto.ExecutionStarted.ParentInstance == null ? null : new ParentInstance - { - Name = proto.ExecutionStarted.ParentInstance.Name, - Version = proto.ExecutionStarted.ParentInstance.Version, - OrchestrationInstance = proto.ExecutionStarted.ParentInstance.OrchestrationInstance.ToCore(), - TaskScheduleId = proto.ExecutionStarted.ParentInstance.TaskScheduledId, - }, - ScheduledStartTime = proto.ExecutionStarted.ScheduledStartTimestamp?.ToDateTime(), - }; - break; - case P.HistoryEvent.EventTypeOneofCase.ExecutionCompleted: - historyEvent = new ExecutionCompletedEvent( - proto.EventId, - proto.ExecutionCompleted.Result, - proto.ExecutionCompleted.OrchestrationStatus.ToCore()); - break; - case P.HistoryEvent.EventTypeOneofCase.ExecutionTerminated: - historyEvent = new ExecutionTerminatedEvent(proto.EventId, proto.ExecutionTerminated.Input); - break; - case P.HistoryEvent.EventTypeOneofCase.ExecutionSuspended: - historyEvent = new ExecutionSuspendedEvent(proto.EventId, proto.ExecutionSuspended.Input); - break; - case P.HistoryEvent.EventTypeOneofCase.ExecutionResumed: - historyEvent = new ExecutionResumedEvent(proto.EventId, proto.ExecutionResumed.Input); - break; - case P.HistoryEvent.EventTypeOneofCase.TaskScheduled: - historyEvent = new TaskScheduledEvent( - proto.EventId, - proto.TaskScheduled.Name, - proto.TaskScheduled.Version, + +namespace Microsoft.DurableTask; + +/// +/// Protobuf utilities and helpers. +/// +static class ProtoUtils +{ + /// + /// Converts a history event from to . + /// + /// The proto history event to converter. + /// The converted history event. + /// When the provided history event type is not supported. + internal static HistoryEvent ConvertHistoryEvent(P.HistoryEvent proto) + { + return ConvertHistoryEvent(proto, conversionState: null); + } + + /// + /// Converts a history event from to , and performs + /// stateful conversions of entity-related events. + /// + /// The proto history event to converter. + /// State needed for converting entity-related history entries and actions. + /// The converted history event. + /// When the provided history event type is not supported. + internal static HistoryEvent ConvertHistoryEvent(P.HistoryEvent proto, EntityConversionState? conversionState) + { + Check.NotNull(proto); + HistoryEvent historyEvent; + switch (proto.EventTypeCase) + { + case P.HistoryEvent.EventTypeOneofCase.ContinueAsNew: + historyEvent = new ContinueAsNewEvent(proto.EventId, proto.ContinueAsNew.Input); + break; + case P.HistoryEvent.EventTypeOneofCase.ExecutionStarted: + OrchestrationInstance instance = proto.ExecutionStarted.OrchestrationInstance.ToCore(); + conversionState?.SetOrchestrationInstance(instance); + historyEvent = new ExecutionStartedEvent(proto.EventId, proto.ExecutionStarted.Input) + { + Name = proto.ExecutionStarted.Name, + Version = proto.ExecutionStarted.Version, + OrchestrationInstance = instance, + Tags = proto.ExecutionStarted.Tags, + ParentInstance = proto.ExecutionStarted.ParentInstance == null ? null : new ParentInstance + { + Name = proto.ExecutionStarted.ParentInstance.Name, + Version = proto.ExecutionStarted.ParentInstance.Version, + OrchestrationInstance = proto.ExecutionStarted.ParentInstance.OrchestrationInstance.ToCore(), + TaskScheduleId = proto.ExecutionStarted.ParentInstance.TaskScheduledId, + }, + ScheduledStartTime = proto.ExecutionStarted.ScheduledStartTimestamp?.ToDateTime(), + }; + break; + case P.HistoryEvent.EventTypeOneofCase.ExecutionCompleted: + historyEvent = new ExecutionCompletedEvent( + proto.EventId, + proto.ExecutionCompleted.Result, + proto.ExecutionCompleted.OrchestrationStatus.ToCore()); + break; + case P.HistoryEvent.EventTypeOneofCase.ExecutionTerminated: + historyEvent = new ExecutionTerminatedEvent(proto.EventId, proto.ExecutionTerminated.Input); + break; + case P.HistoryEvent.EventTypeOneofCase.ExecutionSuspended: + historyEvent = new ExecutionSuspendedEvent(proto.EventId, proto.ExecutionSuspended.Input); + break; + case P.HistoryEvent.EventTypeOneofCase.ExecutionResumed: + historyEvent = new ExecutionResumedEvent(proto.EventId, proto.ExecutionResumed.Input); + break; + case P.HistoryEvent.EventTypeOneofCase.TaskScheduled: + historyEvent = new TaskScheduledEvent( + proto.EventId, + proto.TaskScheduled.Name, + proto.TaskScheduled.Version, proto.TaskScheduled.Input) { Tags = proto.TaskScheduled.Tags, }; - break; - case P.HistoryEvent.EventTypeOneofCase.TaskCompleted: - historyEvent = new TaskCompletedEvent( - proto.EventId, - proto.TaskCompleted.TaskScheduledId, - proto.TaskCompleted.Result); - break; - case P.HistoryEvent.EventTypeOneofCase.TaskFailed: - historyEvent = new TaskFailedEvent( - proto.EventId, - proto.TaskFailed.TaskScheduledId, - reason: null, /* not supported */ - details: null, /* not supported */ - proto.TaskFailed.FailureDetails.ToCore()); - break; - case P.HistoryEvent.EventTypeOneofCase.SubOrchestrationInstanceCreated: - historyEvent = new SubOrchestrationInstanceCreatedEvent(proto.EventId) - { - Input = proto.SubOrchestrationInstanceCreated.Input, - InstanceId = proto.SubOrchestrationInstanceCreated.InstanceId, - Name = proto.SubOrchestrationInstanceCreated.Name, - Version = proto.SubOrchestrationInstanceCreated.Version, - }; - break; - case P.HistoryEvent.EventTypeOneofCase.SubOrchestrationInstanceCompleted: - historyEvent = new SubOrchestrationInstanceCompletedEvent( - proto.EventId, - proto.SubOrchestrationInstanceCompleted.TaskScheduledId, - proto.SubOrchestrationInstanceCompleted.Result); - break; - case P.HistoryEvent.EventTypeOneofCase.SubOrchestrationInstanceFailed: - historyEvent = new SubOrchestrationInstanceFailedEvent( - proto.EventId, - proto.SubOrchestrationInstanceFailed.TaskScheduledId, - reason: null /* not supported */, - details: null /* not supported */, - proto.SubOrchestrationInstanceFailed.FailureDetails.ToCore()); - break; - case P.HistoryEvent.EventTypeOneofCase.TimerCreated: - historyEvent = new TimerCreatedEvent( - proto.EventId, - proto.TimerCreated.FireAt.ToDateTime()); - break; - case P.HistoryEvent.EventTypeOneofCase.TimerFired: - historyEvent = new TimerFiredEvent( - eventId: -1, - proto.TimerFired.FireAt.ToDateTime()) - { - TimerId = proto.TimerFired.TimerId, - }; - break; - case P.HistoryEvent.EventTypeOneofCase.OrchestratorStarted: - historyEvent = new OrchestratorStartedEvent(proto.EventId); - break; - case P.HistoryEvent.EventTypeOneofCase.OrchestratorCompleted: - historyEvent = new OrchestratorCompletedEvent(proto.EventId); - break; - case P.HistoryEvent.EventTypeOneofCase.EventSent: - historyEvent = new EventSentEvent(proto.EventId) - { - InstanceId = proto.EventSent.InstanceId, - Name = proto.EventSent.Name, - Input = proto.EventSent.Input, - }; - break; - case P.HistoryEvent.EventTypeOneofCase.EventRaised: - historyEvent = new EventRaisedEvent(proto.EventId, proto.EventRaised.Input) - { - Name = proto.EventRaised.Name, - }; - break; - case P.HistoryEvent.EventTypeOneofCase.EntityOperationCalled: - historyEvent = EntityConversions.EncodeOperationCalled(proto, conversionState!.CurrentInstance); - conversionState?.EntityRequestIds.Add(proto.EntityOperationCalled.RequestId); - break; - case P.HistoryEvent.EventTypeOneofCase.EntityOperationSignaled: - historyEvent = EntityConversions.EncodeOperationSignaled(proto); - conversionState?.EntityRequestIds.Add(proto.EntityOperationSignaled.RequestId); - break; - case P.HistoryEvent.EventTypeOneofCase.EntityLockRequested: - historyEvent = EntityConversions.EncodeLockRequested(proto, conversionState!.CurrentInstance); - conversionState?.AddUnlockObligations(proto.EntityLockRequested); - break; - case P.HistoryEvent.EventTypeOneofCase.EntityUnlockSent: - historyEvent = EntityConversions.EncodeUnlockSent(proto, conversionState!.CurrentInstance); - conversionState?.RemoveUnlockObligation(proto.EntityUnlockSent.TargetInstanceId); - break; - case P.HistoryEvent.EventTypeOneofCase.EntityLockGranted: - historyEvent = EntityConversions.EncodeLockGranted(proto); - break; - case P.HistoryEvent.EventTypeOneofCase.EntityOperationCompleted: - historyEvent = EntityConversions.EncodeOperationCompleted(proto); - break; - case P.HistoryEvent.EventTypeOneofCase.EntityOperationFailed: - historyEvent = EntityConversions.EncodeOperationFailed(proto); - break; - case P.HistoryEvent.EventTypeOneofCase.GenericEvent: - historyEvent = new GenericEvent(proto.EventId, proto.GenericEvent.Data); - break; - case P.HistoryEvent.EventTypeOneofCase.HistoryState: - historyEvent = new HistoryStateEvent( - proto.EventId, - new OrchestrationState - { - OrchestrationInstance = new OrchestrationInstance - { - InstanceId = proto.HistoryState.OrchestrationState.InstanceId, - }, - Name = proto.HistoryState.OrchestrationState.Name, - Version = proto.HistoryState.OrchestrationState.Version, - ScheduledStartTime = proto.HistoryState.OrchestrationState.ScheduledStartTimestamp.ToDateTime(), - CreatedTime = proto.HistoryState.OrchestrationState.CreatedTimestamp.ToDateTime(), - LastUpdatedTime = proto.HistoryState.OrchestrationState.LastUpdatedTimestamp.ToDateTime(), - Input = proto.HistoryState.OrchestrationState.Input, - Output = proto.HistoryState.OrchestrationState.Output, - Status = proto.HistoryState.OrchestrationState.CustomStatus, - Tags = proto.HistoryState.OrchestrationState.Tags, - }); - break; + break; + case P.HistoryEvent.EventTypeOneofCase.TaskCompleted: + historyEvent = new TaskCompletedEvent( + proto.EventId, + proto.TaskCompleted.TaskScheduledId, + proto.TaskCompleted.Result); + break; + case P.HistoryEvent.EventTypeOneofCase.TaskFailed: + historyEvent = new TaskFailedEvent( + proto.EventId, + proto.TaskFailed.TaskScheduledId, + reason: null, /* not supported */ + details: null, /* not supported */ + proto.TaskFailed.FailureDetails.ToCore()); + break; + case P.HistoryEvent.EventTypeOneofCase.SubOrchestrationInstanceCreated: + historyEvent = new SubOrchestrationInstanceCreatedEvent(proto.EventId) + { + Input = proto.SubOrchestrationInstanceCreated.Input, + InstanceId = proto.SubOrchestrationInstanceCreated.InstanceId, + Name = proto.SubOrchestrationInstanceCreated.Name, + Version = proto.SubOrchestrationInstanceCreated.Version, + }; + break; + case P.HistoryEvent.EventTypeOneofCase.SubOrchestrationInstanceCompleted: + historyEvent = new SubOrchestrationInstanceCompletedEvent( + proto.EventId, + proto.SubOrchestrationInstanceCompleted.TaskScheduledId, + proto.SubOrchestrationInstanceCompleted.Result); + break; + case P.HistoryEvent.EventTypeOneofCase.SubOrchestrationInstanceFailed: + historyEvent = new SubOrchestrationInstanceFailedEvent( + proto.EventId, + proto.SubOrchestrationInstanceFailed.TaskScheduledId, + reason: null /* not supported */, + details: null /* not supported */, + proto.SubOrchestrationInstanceFailed.FailureDetails.ToCore()); + break; + case P.HistoryEvent.EventTypeOneofCase.TimerCreated: + historyEvent = new TimerCreatedEvent( + proto.EventId, + proto.TimerCreated.FireAt.ToDateTime()); + break; + case P.HistoryEvent.EventTypeOneofCase.TimerFired: + historyEvent = new TimerFiredEvent( + eventId: -1, + proto.TimerFired.FireAt.ToDateTime()) + { + TimerId = proto.TimerFired.TimerId, + }; + break; + case P.HistoryEvent.EventTypeOneofCase.OrchestratorStarted: + historyEvent = new OrchestratorStartedEvent(proto.EventId); + break; + case P.HistoryEvent.EventTypeOneofCase.OrchestratorCompleted: + historyEvent = new OrchestratorCompletedEvent(proto.EventId); + break; + case P.HistoryEvent.EventTypeOneofCase.EventSent: + historyEvent = new EventSentEvent(proto.EventId) + { + InstanceId = proto.EventSent.InstanceId, + Name = proto.EventSent.Name, + Input = proto.EventSent.Input, + }; + break; + case P.HistoryEvent.EventTypeOneofCase.EventRaised: + historyEvent = new EventRaisedEvent(proto.EventId, proto.EventRaised.Input) + { + Name = proto.EventRaised.Name, + }; + break; + case P.HistoryEvent.EventTypeOneofCase.EntityOperationCalled: + historyEvent = EntityConversions.EncodeOperationCalled(proto, conversionState!.CurrentInstance); + conversionState?.EntityRequestIds.Add(proto.EntityOperationCalled.RequestId); + break; + case P.HistoryEvent.EventTypeOneofCase.EntityOperationSignaled: + historyEvent = EntityConversions.EncodeOperationSignaled(proto); + conversionState?.EntityRequestIds.Add(proto.EntityOperationSignaled.RequestId); + break; + case P.HistoryEvent.EventTypeOneofCase.EntityLockRequested: + historyEvent = EntityConversions.EncodeLockRequested(proto, conversionState!.CurrentInstance); + conversionState?.AddUnlockObligations(proto.EntityLockRequested); + break; + case P.HistoryEvent.EventTypeOneofCase.EntityUnlockSent: + historyEvent = EntityConversions.EncodeUnlockSent(proto, conversionState!.CurrentInstance); + conversionState?.RemoveUnlockObligation(proto.EntityUnlockSent.TargetInstanceId); + break; + case P.HistoryEvent.EventTypeOneofCase.EntityLockGranted: + historyEvent = EntityConversions.EncodeLockGranted(proto); + break; + case P.HistoryEvent.EventTypeOneofCase.EntityOperationCompleted: + historyEvent = EntityConversions.EncodeOperationCompleted(proto); + break; + case P.HistoryEvent.EventTypeOneofCase.EntityOperationFailed: + historyEvent = EntityConversions.EncodeOperationFailed(proto); + break; + case P.HistoryEvent.EventTypeOneofCase.GenericEvent: + historyEvent = new GenericEvent(proto.EventId, proto.GenericEvent.Data); + break; + case P.HistoryEvent.EventTypeOneofCase.HistoryState: + historyEvent = new HistoryStateEvent( + proto.EventId, + new OrchestrationState + { + OrchestrationInstance = new OrchestrationInstance + { + InstanceId = proto.HistoryState.OrchestrationState.InstanceId, + }, + Name = proto.HistoryState.OrchestrationState.Name, + Version = proto.HistoryState.OrchestrationState.Version, + ScheduledStartTime = proto.HistoryState.OrchestrationState.ScheduledStartTimestamp.ToDateTime(), + CreatedTime = proto.HistoryState.OrchestrationState.CreatedTimestamp.ToDateTime(), + LastUpdatedTime = proto.HistoryState.OrchestrationState.LastUpdatedTimestamp.ToDateTime(), + Input = proto.HistoryState.OrchestrationState.Input, + Output = proto.HistoryState.OrchestrationState.Output, + Status = proto.HistoryState.OrchestrationState.CustomStatus, + Tags = proto.HistoryState.OrchestrationState.Tags, + }); + break; case P.HistoryEvent.EventTypeOneofCase.ExecutionRewound: historyEvent = new ExecutionRewoundEvent(proto.EventId); break; - default: - throw new NotSupportedException($"Deserialization of {proto.EventTypeCase} is not supported."); - } - - historyEvent.Timestamp = proto.Timestamp.ToDateTime(); - return historyEvent; - } - - /// - /// Converts a to a gRPC . - /// - /// The date-time to convert. - /// The gRPC timestamp. - internal static Timestamp ToTimestamp(this DateTime dateTime) - { - // The protobuf libraries require timestamps to be in UTC - if (dateTime.Kind == DateTimeKind.Unspecified) - { - dateTime = DateTime.SpecifyKind(dateTime, DateTimeKind.Utc); - } - else if (dateTime.Kind == DateTimeKind.Local) - { - dateTime = dateTime.ToUniversalTime(); - } - - return Timestamp.FromDateTime(dateTime); - } - - /// - /// Converts a to a gRPC . - /// - /// The date-time to convert. - /// The gRPC timestamp. - internal static Timestamp? ToTimestamp(this DateTime? dateTime) - => dateTime.HasValue ? dateTime.Value.ToTimestamp() : null; - - /// - /// Converts a to a gRPC . - /// - /// The date-time to convert. - /// The gRPC timestamp. - internal static Timestamp ToTimestamp(this DateTimeOffset dateTime) => Timestamp.FromDateTimeOffset(dateTime); - - /// - /// Converts a to a gRPC . - /// - /// The date-time to convert. - /// The gRPC timestamp. - internal static Timestamp? ToTimestamp(this DateTimeOffset? dateTime) - => dateTime.HasValue ? dateTime.Value.ToTimestamp() : null; - - /// - /// Constructs a . - /// - /// The orchestrator instance ID. + default: + throw new NotSupportedException($"Deserialization of {proto.EventTypeCase} is not supported."); + } + + historyEvent.Timestamp = proto.Timestamp.ToDateTime(); + return historyEvent; + } + + /// + /// Converts a to a gRPC . + /// + /// The date-time to convert. + /// The gRPC timestamp. + internal static Timestamp ToTimestamp(this DateTime dateTime) + { + // The protobuf libraries require timestamps to be in UTC + if (dateTime.Kind == DateTimeKind.Unspecified) + { + dateTime = DateTime.SpecifyKind(dateTime, DateTimeKind.Utc); + } + else if (dateTime.Kind == DateTimeKind.Local) + { + dateTime = dateTime.ToUniversalTime(); + } + + return Timestamp.FromDateTime(dateTime); + } + + /// + /// Converts a to a gRPC . + /// + /// The date-time to convert. + /// The gRPC timestamp. + internal static Timestamp? ToTimestamp(this DateTime? dateTime) + => dateTime.HasValue ? dateTime.Value.ToTimestamp() : null; + + /// + /// Converts a to a gRPC . + /// + /// The date-time to convert. + /// The gRPC timestamp. + internal static Timestamp ToTimestamp(this DateTimeOffset dateTime) => Timestamp.FromDateTimeOffset(dateTime); + + /// + /// Converts a to a gRPC . + /// + /// The date-time to convert. + /// The gRPC timestamp. + internal static Timestamp? ToTimestamp(this DateTimeOffset? dateTime) + => dateTime.HasValue ? dateTime.Value.ToTimestamp() : null; + + /// + /// Constructs a . + /// + /// The orchestrator instance ID. /// The orchestrator execution ID. - /// The orchestrator customer status or null if no custom status. - /// The orchestrator actions. - /// - /// The completion token for the work item. It must be the exact same - /// value that was provided by the corresponding that triggered the orchestrator execution. - /// - /// The entity conversion state, or null if no conversion is required. + /// The orchestrator customer status or null if no custom status. + /// The orchestrator actions. + /// + /// The completion token for the work item. It must be the exact same + /// value that was provided by the corresponding that triggered the orchestrator execution. + /// + /// The entity conversion state, or null if no conversion is required. /// The that represents orchestration execution. /// Whether or not a history is required to complete the orchestration request and none was provided. - /// The orchestrator response. - /// When an orchestrator action is unknown. - internal static P.OrchestratorResponse ConstructOrchestratorResponse( - string instanceId, + /// The orchestrator response. + /// When an orchestrator action is unknown. + internal static P.OrchestratorResponse ConstructOrchestratorResponse( + string instanceId, string executionId, - string? customStatus, + string? customStatus, IEnumerable? actions, - string completionToken, + string completionToken, EntityConversionState? entityConversionState, Activity? orchestrationActivity, bool requiresHistory = false) - { - var response = new P.OrchestratorResponse - { - InstanceId = instanceId, - CustomStatus = customStatus, - CompletionToken = completionToken, + { + var response = new P.OrchestratorResponse + { + InstanceId = instanceId, + CustomStatus = customStatus, + CompletionToken = completionToken, OrchestrationTraceContext = new() { @@ -311,8 +311,8 @@ internal static P.OrchestratorResponse ConstructOrchestratorResponse( SpanStartTime = orchestrationActivity?.StartTimeUtc.ToTimestamp(), }, RequiresHistory = requiresHistory, - }; - + }; + // If a history is required and the orchestration request was not completed, then there is no list of actions. if (requiresHistory) { @@ -320,10 +320,10 @@ internal static P.OrchestratorResponse ConstructOrchestratorResponse( } Check.NotNull(actions); - foreach (OrchestratorAction action in actions) - { - var protoAction = new P.OrchestratorAction { Id = action.Id }; - + foreach (OrchestratorAction action in actions) + { + var protoAction = new P.OrchestratorAction { Id = action.Id }; + P.TraceContext? CreateTraceContext() { if (orchestrationActivity is null) @@ -341,18 +341,18 @@ internal static P.OrchestratorResponse ConstructOrchestratorResponse( }; } - switch (action.OrchestratorActionType) - { - case OrchestratorActionType.ScheduleOrchestrator: - var scheduleTaskAction = (ScheduleTaskOrchestratorAction)action; + switch (action.OrchestratorActionType) + { + case OrchestratorActionType.ScheduleOrchestrator: + var scheduleTaskAction = (ScheduleTaskOrchestratorAction)action; - protoAction.ScheduleTask = new P.ScheduleTaskAction - { - Name = scheduleTaskAction.Name, - Version = scheduleTaskAction.Version, - Input = scheduleTaskAction.Input, + protoAction.ScheduleTask = new P.ScheduleTaskAction + { + Name = scheduleTaskAction.Name, + Version = scheduleTaskAction.Version, + Input = scheduleTaskAction.Input, ParentTraceContext = CreateTraceContext(), - }; + }; if (scheduleTaskAction.Tags != null) { @@ -362,111 +362,111 @@ internal static P.OrchestratorResponse ConstructOrchestratorResponse( } } - break; - case OrchestratorActionType.CreateSubOrchestration: - var subOrchestrationAction = (CreateSubOrchestrationAction)action; - protoAction.CreateSubOrchestration = new P.CreateSubOrchestrationAction - { - Input = subOrchestrationAction.Input, - InstanceId = subOrchestrationAction.InstanceId, - Name = subOrchestrationAction.Name, - Version = subOrchestrationAction.Version, + break; + case OrchestratorActionType.CreateSubOrchestration: + var subOrchestrationAction = (CreateSubOrchestrationAction)action; + protoAction.CreateSubOrchestration = new P.CreateSubOrchestrationAction + { + Input = subOrchestrationAction.Input, + InstanceId = subOrchestrationAction.InstanceId, + Name = subOrchestrationAction.Name, + Version = subOrchestrationAction.Version, ParentTraceContext = CreateTraceContext(), - }; - break; - case OrchestratorActionType.CreateTimer: - var createTimerAction = (CreateTimerOrchestratorAction)action; - protoAction.CreateTimer = new P.CreateTimerAction - { - FireAt = createTimerAction.FireAt.ToTimestamp(), - }; - break; - case OrchestratorActionType.SendEvent: - var sendEventAction = (SendEventOrchestratorAction)action; - if (sendEventAction.Instance == null) - { - throw new ArgumentException( - $"{nameof(SendEventOrchestratorAction)} cannot have a null Instance property!"); - } - - if (entityConversionState is not null - && DTCore.Common.Entities.IsEntityInstance(sendEventAction.Instance.InstanceId) - && sendEventAction.EventName is not null - && sendEventAction.EventData is not null) - { - P.SendEntityMessageAction sendAction = new(); - protoAction.SendEntityMessage = sendAction; - - EntityConversions.DecodeEntityMessageAction( - sendEventAction.EventName, - sendEventAction.EventData, - sendEventAction.Instance.InstanceId, - sendAction, - out string requestId); - - entityConversionState.EntityRequestIds.Add(requestId); - - switch (sendAction.EntityMessageTypeCase) - { - case P.SendEntityMessageAction.EntityMessageTypeOneofCase.EntityLockRequested: - entityConversionState.AddUnlockObligations(sendAction.EntityLockRequested); - break; - case P.SendEntityMessageAction.EntityMessageTypeOneofCase.EntityUnlockSent: - entityConversionState.RemoveUnlockObligation(sendAction.EntityUnlockSent.TargetInstanceId); - break; - default: - break; - } - } - else - { - protoAction.SendEvent = new P.SendEventAction - { - Instance = sendEventAction.Instance.ToProtobuf(), - Name = sendEventAction.EventName, - Data = sendEventAction.EventData, - }; + }; + break; + case OrchestratorActionType.CreateTimer: + var createTimerAction = (CreateTimerOrchestratorAction)action; + protoAction.CreateTimer = new P.CreateTimerAction + { + FireAt = createTimerAction.FireAt.ToTimestamp(), + }; + break; + case OrchestratorActionType.SendEvent: + var sendEventAction = (SendEventOrchestratorAction)action; + if (sendEventAction.Instance == null) + { + throw new ArgumentException( + $"{nameof(SendEventOrchestratorAction)} cannot have a null Instance property!"); + } + + if (entityConversionState is not null + && DTCore.Common.Entities.IsEntityInstance(sendEventAction.Instance.InstanceId) + && sendEventAction.EventName is not null + && sendEventAction.EventData is not null) + { + P.SendEntityMessageAction sendAction = new P.SendEntityMessageAction(); + protoAction.SendEntityMessage = sendAction; + + EntityConversions.DecodeEntityMessageAction( + sendEventAction.EventName, + sendEventAction.EventData, + sendEventAction.Instance.InstanceId, + sendAction, + out string requestId); + + entityConversionState.EntityRequestIds.Add(requestId); + + switch (sendAction.EntityMessageTypeCase) + { + case P.SendEntityMessageAction.EntityMessageTypeOneofCase.EntityLockRequested: + entityConversionState.AddUnlockObligations(sendAction.EntityLockRequested); + break; + case P.SendEntityMessageAction.EntityMessageTypeOneofCase.EntityUnlockSent: + entityConversionState.RemoveUnlockObligation(sendAction.EntityUnlockSent.TargetInstanceId); + break; + default: + break; + } + } + else + { + protoAction.SendEvent = new P.SendEventAction + { + Instance = sendEventAction.Instance.ToProtobuf(), + Name = sendEventAction.EventName, + Data = sendEventAction.EventData, + }; // Distributed Tracing: start a new trace activity derived from the orchestration // for an EventRaisedEvent (external event) using Activity? traceActivity = TraceHelper.StartTraceActivityForEventRaisedFromWorker(sendEventAction, instanceId, executionId); traceActivity?.Stop(); - } - - break; - case OrchestratorActionType.OrchestrationComplete: - - if (entityConversionState is not null) - { - // as a precaution, unlock any entities that were not unlocked for some reason, before - // completing the orchestration. - foreach ((string target, string criticalSectionId) in entityConversionState.ResetObligations()) - { - response.Actions.Add(new P.OrchestratorAction - { - Id = action.Id, - SendEntityMessage = new P.SendEntityMessageAction - { - EntityUnlockSent = new P.EntityUnlockSentEvent - { - CriticalSectionId = criticalSectionId, - TargetInstanceId = target, - ParentInstanceId = entityConversionState.CurrentInstance?.InstanceId, - }, - }, - }); - } - } - - var completeAction = (OrchestrationCompleteOrchestratorAction)action; - protoAction.CompleteOrchestration = new P.CompleteOrchestrationAction + } + + break; + case OrchestratorActionType.OrchestrationComplete: + + if (entityConversionState is not null) + { + // as a precaution, unlock any entities that were not unlocked for some reason, before + // completing the orchestration. + foreach ((string target, string criticalSectionId) in entityConversionState.ResetObligations()) + { + response.Actions.Add(new P.OrchestratorAction + { + Id = action.Id, + SendEntityMessage = new P.SendEntityMessageAction + { + EntityUnlockSent = new P.EntityUnlockSentEvent + { + CriticalSectionId = criticalSectionId, + TargetInstanceId = target, + ParentInstanceId = entityConversionState.CurrentInstance?.InstanceId, + }, + }, + }); + } + } + + var completeAction = (OrchestrationCompleteOrchestratorAction)action; + protoAction.CompleteOrchestration = new P.CompleteOrchestrationAction { - CarryoverEvents = { completeAction.CarryoverEvents.Select(ToProtobuf) }, - Details = completeAction.Details, - NewVersion = completeAction.NewVersion, - OrchestrationStatus = completeAction.OrchestrationStatus.ToProtobuf(), - Result = completeAction.Result, + CarryoverEvents = { completeAction.CarryoverEvents.Select(ToProtobuf) }, + Details = completeAction.Details, + NewVersion = completeAction.NewVersion, + OrchestrationStatus = completeAction.OrchestrationStatus.ToProtobuf(), + Result = completeAction.Result, }; foreach (KeyValuePair tag in completeAction.Tags) @@ -474,96 +474,96 @@ internal static P.OrchestratorResponse ConstructOrchestratorResponse( protoAction.CompleteOrchestration.Tags[tag.Key] = tag.Value; } - if (completeAction.OrchestrationStatus == OrchestrationStatus.Failed) - { - protoAction.CompleteOrchestration.FailureDetails = completeAction.FailureDetails.ToProtobuf(); - } - - break; - default: - throw new NotSupportedException($"Unknown orchestrator action: {action.OrchestratorActionType}"); - } - - response.Actions.Add(protoAction); - } - - return response; - } - - /// - /// Converts a to a . - /// - /// The status to convert. - /// The converted status. - internal static OrchestrationStatus ToCore(this P.OrchestrationStatus status) - { - return (OrchestrationStatus)status; - } - - /// - /// Converts a to a . - /// - /// The status to convert. - /// The converted status. - [return: NotNullIfNotNull(nameof(status))] - internal static OrchestrationInstance? ToCore(this P.OrchestrationInstance? status) - { - if (status == null) - { - return null; - } - - return new OrchestrationInstance - { - InstanceId = status.InstanceId, - ExecutionId = status.ExecutionId, - }; - } - - /// - /// Converts a to a . - /// - /// The failure details to convert. - /// The converted failure details. - [return: NotNullIfNotNull(nameof(failureDetails))] - internal static TaskFailureDetails? ToTaskFailureDetails(this P.TaskFailureDetails? failureDetails) - { - if (failureDetails == null) - { - return null; - } - - return new TaskFailureDetails( - failureDetails.ErrorType, - failureDetails.ErrorMessage, - failureDetails.StackTrace, + if (completeAction.OrchestrationStatus == OrchestrationStatus.Failed) + { + protoAction.CompleteOrchestration.FailureDetails = completeAction.FailureDetails.ToProtobuf(); + } + + break; + default: + throw new NotSupportedException($"Unknown orchestrator action: {action.OrchestratorActionType}"); + } + + response.Actions.Add(protoAction); + } + + return response; + } + + /// + /// Converts a to a . + /// + /// The status to convert. + /// The converted status. + internal static OrchestrationStatus ToCore(this P.OrchestrationStatus status) + { + return (OrchestrationStatus)status; + } + + /// + /// Converts a to a . + /// + /// The status to convert. + /// The converted status. + [return: NotNullIfNotNull(nameof(status))] + internal static OrchestrationInstance? ToCore(this P.OrchestrationInstance? status) + { + if (status == null) + { + return null; + } + + return new OrchestrationInstance + { + InstanceId = status.InstanceId, + ExecutionId = status.ExecutionId, + }; + } + + /// + /// Converts a to a . + /// + /// The failure details to convert. + /// The converted failure details. + [return: NotNullIfNotNull(nameof(failureDetails))] + internal static TaskFailureDetails? ToTaskFailureDetails(this P.TaskFailureDetails? failureDetails) + { + if (failureDetails == null) + { + return null; + } + + return new TaskFailureDetails( + failureDetails.ErrorType, + failureDetails.ErrorMessage, + failureDetails.StackTrace, failureDetails.InnerFailure.ToTaskFailureDetails(), ConvertProperties(failureDetails.Properties)); - } - - /// - /// Converts a to . - /// - /// The exception to convert. + } + + /// + /// Converts a to . + /// + /// The exception to convert. /// Optional exception properties provider. - /// The task failure details. - [return: NotNullIfNotNull(nameof(e))] + /// The task failure details. + [return: NotNullIfNotNull(nameof(e))] internal static P.TaskFailureDetails? ToTaskFailureDetails(this Exception? e, DTCore.IExceptionPropertiesProvider? exceptionPropertiesProvider = null) - { - if (e == null) - { - return null; - } - + { + if (e == null) + { + return null; + } + IDictionary? properties = exceptionPropertiesProvider?.GetExceptionProperties(e); var taskFailureDetails = new P.TaskFailureDetails - { - ErrorType = e.GetType().FullName, - ErrorMessage = e.Message, - StackTrace = e.StackTrace, + { + ErrorType = e.GetType().FullName, + ErrorMessage = e.Message, + StackTrace = e.StackTrace, InnerFailure = e.InnerException.ToTaskFailureDetails(exceptionPropertiesProvider), - }; + }; if (properties != null) { @@ -572,263 +572,263 @@ internal static OrchestrationStatus ToCore(this P.OrchestrationStatus status) taskFailureDetails.Properties[kvp.Key] = ConvertObjectToValue(kvp.Value); } } - + return taskFailureDetails; } - /// - /// Converts a to a . - /// - /// The entity batch request to convert. - /// The converted entity batch request. - [return: NotNullIfNotNull(nameof(entityBatchRequest))] - internal static EntityBatchRequest? ToEntityBatchRequest(this P.EntityBatchRequest? entityBatchRequest) - { - if (entityBatchRequest == null) - { - return null; - } - - return new EntityBatchRequest() - { - EntityState = entityBatchRequest.EntityState, - InstanceId = entityBatchRequest.InstanceId, - Operations = entityBatchRequest.Operations.Select(r => r.ToOperationRequest()).ToList(), - }; - } - - /// - /// Converts a to a . - /// - /// The entity request to convert. - /// The converted request. - /// Additional info about each operation, required by DTS. - internal static void ToEntityBatchRequest( - this P.EntityRequest entityRequest, - out EntityBatchRequest batchRequest, - out List operationInfos) - { - batchRequest = new EntityBatchRequest() - { - EntityState = entityRequest.EntityState, - InstanceId = entityRequest.InstanceId, - Operations = [], // operations are added to this collection below - }; - - operationInfos = new(entityRequest.OperationRequests.Count); - - foreach (P.HistoryEvent? op in entityRequest.OperationRequests) - { - if (op.EntityOperationSignaled is not null) - { - batchRequest.Operations.Add(new OperationRequest - { - Id = Guid.Parse(op.EntityOperationSignaled.RequestId), - Operation = op.EntityOperationSignaled.Operation, - Input = op.EntityOperationSignaled.Input, - }); - operationInfos.Add(new P.OperationInfo - { - RequestId = op.EntityOperationSignaled.RequestId, - ResponseDestination = null, // means we don't send back a response to the caller - }); - } - else if (op.EntityOperationCalled is not null) - { - batchRequest.Operations.Add(new OperationRequest - { - Id = Guid.Parse(op.EntityOperationCalled.RequestId), - Operation = op.EntityOperationCalled.Operation, - Input = op.EntityOperationCalled.Input, - }); - operationInfos.Add(new P.OperationInfo - { - RequestId = op.EntityOperationCalled.RequestId, - ResponseDestination = new P.OrchestrationInstance - { - InstanceId = op.EntityOperationCalled.ParentInstanceId, - ExecutionId = op.EntityOperationCalled.ParentExecutionId, - }, - }); - } - } - } - - /// - /// Converts a to a . - /// - /// The operation request to convert. - /// The converted operation request. - [return: NotNullIfNotNull(nameof(operationRequest))] - internal static OperationRequest? ToOperationRequest(this P.OperationRequest? operationRequest) - { - if (operationRequest == null) - { - return null; - } - - return new OperationRequest() - { - Operation = operationRequest.Operation, - Input = operationRequest.Input, - Id = Guid.Parse(operationRequest.RequestId), + /// + /// Converts a to a . + /// + /// The entity batch request to convert. + /// The converted entity batch request. + [return: NotNullIfNotNull(nameof(entityBatchRequest))] + internal static EntityBatchRequest? ToEntityBatchRequest(this P.EntityBatchRequest? entityBatchRequest) + { + if (entityBatchRequest == null) + { + return null; + } + + return new EntityBatchRequest() + { + EntityState = entityBatchRequest.EntityState, + InstanceId = entityBatchRequest.InstanceId, + Operations = entityBatchRequest.Operations.Select(r => r.ToOperationRequest()).ToList(), + }; + } + + /// + /// Converts a to a . + /// + /// The entity request to convert. + /// The converted request. + /// Additional info about each operation, required by DTS. + internal static void ToEntityBatchRequest( + this P.EntityRequest entityRequest, + out EntityBatchRequest batchRequest, + out List operationInfos) + { + batchRequest = new EntityBatchRequest() + { + EntityState = entityRequest.EntityState, + InstanceId = entityRequest.InstanceId, + Operations = [], // operations are added to this collection below + }; + + operationInfos = new(entityRequest.OperationRequests.Count); + + foreach (P.HistoryEvent? op in entityRequest.OperationRequests) + { + if (op.EntityOperationSignaled is not null) + { + batchRequest.Operations.Add(new OperationRequest + { + Id = Guid.Parse(op.EntityOperationSignaled.RequestId), + Operation = op.EntityOperationSignaled.Operation, + Input = op.EntityOperationSignaled.Input, + }); + operationInfos.Add(new P.OperationInfo + { + RequestId = op.EntityOperationSignaled.RequestId, + ResponseDestination = null, // means we don't send back a response to the caller + }); + } + else if (op.EntityOperationCalled is not null) + { + batchRequest.Operations.Add(new OperationRequest + { + Id = Guid.Parse(op.EntityOperationCalled.RequestId), + Operation = op.EntityOperationCalled.Operation, + Input = op.EntityOperationCalled.Input, + }); + operationInfos.Add(new P.OperationInfo + { + RequestId = op.EntityOperationCalled.RequestId, + ResponseDestination = new P.OrchestrationInstance + { + InstanceId = op.EntityOperationCalled.ParentInstanceId, + ExecutionId = op.EntityOperationCalled.ParentExecutionId, + }, + }); + } + } + } + + /// + /// Converts a to a . + /// + /// The operation request to convert. + /// The converted operation request. + [return: NotNullIfNotNull(nameof(operationRequest))] + internal static OperationRequest? ToOperationRequest(this P.OperationRequest? operationRequest) + { + if (operationRequest == null) + { + return null; + } + + return new OperationRequest() + { + Operation = operationRequest.Operation, + Input = operationRequest.Input, + Id = Guid.Parse(operationRequest.RequestId), TraceContext = operationRequest.TraceContext != null ? new DistributedTraceContext( operationRequest.TraceContext.TraceParent, operationRequest.TraceContext.TraceState) : null, - }; - } - - /// - /// Converts a to a . - /// - /// The operation result to convert. - /// The converted operation result. - [return: NotNullIfNotNull(nameof(operationResult))] - internal static OperationResult? ToOperationResult(this P.OperationResult? operationResult) - { - if (operationResult == null) - { - return null; - } - - switch (operationResult.ResultTypeCase) - { - case P.OperationResult.ResultTypeOneofCase.Success: - return new OperationResult() - { - Result = operationResult.Success.Result, + }; + } + + /// + /// Converts a to a . + /// + /// The operation result to convert. + /// The converted operation result. + [return: NotNullIfNotNull(nameof(operationResult))] + internal static OperationResult? ToOperationResult(this P.OperationResult? operationResult) + { + if (operationResult == null) + { + return null; + } + + switch (operationResult.ResultTypeCase) + { + case P.OperationResult.ResultTypeOneofCase.Success: + return new OperationResult() + { + Result = operationResult.Success.Result, StartTimeUtc = operationResult.Success.StartTimeUtc?.ToDateTime(), EndTimeUtc = operationResult.Success.EndTimeUtc?.ToDateTime(), - }; - - case P.OperationResult.ResultTypeOneofCase.Failure: - return new OperationResult() - { - FailureDetails = operationResult.Failure.FailureDetails.ToCore(), + }; + + case P.OperationResult.ResultTypeOneofCase.Failure: + return new OperationResult() + { + FailureDetails = operationResult.Failure.FailureDetails.ToCore(), StartTimeUtc = operationResult.Failure.StartTimeUtc?.ToDateTime(), EndTimeUtc = operationResult.Failure.EndTimeUtc?.ToDateTime(), - }; - - default: - throw new NotSupportedException($"Deserialization of {operationResult.ResultTypeCase} is not supported."); - } - } - - /// - /// Converts a to . - /// - /// The operation result to convert. - /// The converted operation result. - [return: NotNullIfNotNull(nameof(operationResult))] - internal static P.OperationResult? ToOperationResult(this OperationResult? operationResult) - { - if (operationResult == null) - { - return null; - } - - if (operationResult.FailureDetails == null) - { - return new P.OperationResult() - { - Success = new P.OperationResultSuccess() - { - Result = operationResult.Result, + }; + + default: + throw new NotSupportedException($"Deserialization of {operationResult.ResultTypeCase} is not supported."); + } + } + + /// + /// Converts a to . + /// + /// The operation result to convert. + /// The converted operation result. + [return: NotNullIfNotNull(nameof(operationResult))] + internal static P.OperationResult? ToOperationResult(this OperationResult? operationResult) + { + if (operationResult == null) + { + return null; + } + + if (operationResult.FailureDetails == null) + { + return new P.OperationResult() + { + Success = new P.OperationResultSuccess() + { + Result = operationResult.Result, StartTimeUtc = operationResult.StartTimeUtc?.ToTimestamp(), EndTimeUtc = operationResult.EndTimeUtc?.ToTimestamp(), - }, - }; - } - else - { - return new P.OperationResult() - { - Failure = new P.OperationResultFailure() - { - FailureDetails = ToProtobuf(operationResult.FailureDetails), + }, + }; + } + else + { + return new P.OperationResult() + { + Failure = new P.OperationResultFailure() + { + FailureDetails = ToProtobuf(operationResult.FailureDetails), StartTimeUtc = operationResult.StartTimeUtc?.ToTimestamp(), EndTimeUtc = operationResult.EndTimeUtc?.ToTimestamp(), - }, - }; - } - } - - /// - /// Converts a to a . - /// - /// The operation action to convert. - /// The converted operation action. - [return: NotNullIfNotNull(nameof(operationAction))] - internal static OperationAction? ToOperationAction(this P.OperationAction? operationAction) - { - if (operationAction == null) - { - return null; - } - - switch (operationAction.OperationActionTypeCase) - { - case P.OperationAction.OperationActionTypeOneofCase.SendSignal: - - return new SendSignalOperationAction() - { - Name = operationAction.SendSignal.Name, - Input = operationAction.SendSignal.Input, - InstanceId = operationAction.SendSignal.InstanceId, - ScheduledTime = operationAction.SendSignal.ScheduledTime?.ToDateTime(), + }, + }; + } + } + + /// + /// Converts a to a . + /// + /// The operation action to convert. + /// The converted operation action. + [return: NotNullIfNotNull(nameof(operationAction))] + internal static OperationAction? ToOperationAction(this P.OperationAction? operationAction) + { + if (operationAction == null) + { + return null; + } + + switch (operationAction.OperationActionTypeCase) + { + case P.OperationAction.OperationActionTypeOneofCase.SendSignal: + + return new SendSignalOperationAction() + { + Name = operationAction.SendSignal.Name, + Input = operationAction.SendSignal.Input, + InstanceId = operationAction.SendSignal.InstanceId, + ScheduledTime = operationAction.SendSignal.ScheduledTime?.ToDateTime(), RequestTime = operationAction.SendSignal.RequestTime?.ToDateTimeOffset(), ParentTraceContext = operationAction.SendSignal.ParentTraceContext != null ? new DistributedTraceContext( operationAction.SendSignal.ParentTraceContext.TraceParent, operationAction.SendSignal.ParentTraceContext.TraceState) : null, - }; - - case P.OperationAction.OperationActionTypeOneofCase.StartNewOrchestration: - - return new StartNewOrchestrationOperationAction() - { - Name = operationAction.StartNewOrchestration.Name, - Input = operationAction.StartNewOrchestration.Input, - InstanceId = operationAction.StartNewOrchestration.InstanceId, - Version = operationAction.StartNewOrchestration.Version, - ScheduledStartTime = operationAction.StartNewOrchestration.ScheduledTime?.ToDateTime(), + }; + + case P.OperationAction.OperationActionTypeOneofCase.StartNewOrchestration: + + return new StartNewOrchestrationOperationAction() + { + Name = operationAction.StartNewOrchestration.Name, + Input = operationAction.StartNewOrchestration.Input, + InstanceId = operationAction.StartNewOrchestration.InstanceId, + Version = operationAction.StartNewOrchestration.Version, + ScheduledStartTime = operationAction.StartNewOrchestration.ScheduledTime?.ToDateTime(), RequestTime = operationAction.StartNewOrchestration.RequestTime?.ToDateTimeOffset(), ParentTraceContext = operationAction.StartNewOrchestration.ParentTraceContext != null ? new DistributedTraceContext( operationAction.StartNewOrchestration.ParentTraceContext.TraceParent, operationAction.StartNewOrchestration.ParentTraceContext.TraceState) : null, - }; - default: - throw new NotSupportedException($"Deserialization of {operationAction.OperationActionTypeCase} is not supported."); - } - } - - /// - /// Converts a to . - /// - /// The operation action to convert. - /// The converted operation action. - [return: NotNullIfNotNull(nameof(operationAction))] - internal static P.OperationAction? ToOperationAction(this OperationAction? operationAction) - { - if (operationAction == null) - { - return null; - } - - var action = new P.OperationAction(); - - switch (operationAction) - { - case SendSignalOperationAction sendSignalAction: - - action.SendSignal = new P.SendSignalAction() - { - Name = sendSignalAction.Name, - Input = sendSignalAction.Input, - InstanceId = sendSignalAction.InstanceId, - ScheduledTime = sendSignalAction.ScheduledTime?.ToTimestamp(), + }; + default: + throw new NotSupportedException($"Deserialization of {operationAction.OperationActionTypeCase} is not supported."); + } + } + + /// + /// Converts a to . + /// + /// The operation action to convert. + /// The converted operation action. + [return: NotNullIfNotNull(nameof(operationAction))] + internal static P.OperationAction? ToOperationAction(this OperationAction? operationAction) + { + if (operationAction == null) + { + return null; + } + + var action = new P.OperationAction(); + + switch (operationAction) + { + case SendSignalOperationAction sendSignalAction: + + action.SendSignal = new P.SendSignalAction() + { + Name = sendSignalAction.Name, + Input = sendSignalAction.Input, + InstanceId = sendSignalAction.InstanceId, + ScheduledTime = sendSignalAction.ScheduledTime?.ToTimestamp(), RequestTime = sendSignalAction.RequestTime?.ToTimestamp(), ParentTraceContext = sendSignalAction.ParentTraceContext != null ? new P.TraceContext @@ -837,18 +837,18 @@ internal static void ToEntityBatchRequest( TraceState = sendSignalAction.ParentTraceContext.TraceState, } : null, - }; - break; - - case StartNewOrchestrationOperationAction startNewOrchestrationAction: - - action.StartNewOrchestration = new P.StartNewOrchestrationAction() - { - Name = startNewOrchestrationAction.Name, - Input = startNewOrchestrationAction.Input, - Version = startNewOrchestrationAction.Version, - InstanceId = startNewOrchestrationAction.InstanceId, - ScheduledTime = startNewOrchestrationAction.ScheduledStartTime?.ToTimestamp(), + }; + break; + + case StartNewOrchestrationOperationAction startNewOrchestrationAction: + + action.StartNewOrchestration = new P.StartNewOrchestrationAction() + { + Name = startNewOrchestrationAction.Name, + Input = startNewOrchestrationAction.Input, + Version = startNewOrchestrationAction.Version, + InstanceId = startNewOrchestrationAction.InstanceId, + ScheduledTime = startNewOrchestrationAction.ScheduledStartTime?.ToTimestamp(), RequestTime = startNewOrchestrationAction.RequestTime?.ToTimestamp(), ParentTraceContext = startNewOrchestrationAction.ParentTraceContext != null ? new P.TraceContext @@ -857,170 +857,170 @@ internal static void ToEntityBatchRequest( TraceState = startNewOrchestrationAction.ParentTraceContext.TraceState, } : null, - }; - break; - } - - return action; - } - - /// - /// Converts a to a . - /// - /// The operation result to convert. - /// The converted operation result. - [return: NotNullIfNotNull(nameof(entityBatchResult))] - internal static EntityBatchResult? ToEntityBatchResult(this P.EntityBatchResult? entityBatchResult) - { - if (entityBatchResult == null) - { - return null; - } - - return new EntityBatchResult() - { - Actions = entityBatchResult.Actions.Select(operationAction => operationAction!.ToOperationAction()).ToList(), - EntityState = entityBatchResult.EntityState, - Results = entityBatchResult.Results.Select(operationResult => operationResult!.ToOperationResult()).ToList(), - FailureDetails = entityBatchResult.FailureDetails.ToCore(), - }; - } - - /// - /// Converts a to . - /// - /// The operation result to convert. - /// The completion token, or null for the older protocol. - /// Additional information about each operation, required by DTS. - /// The converted operation result. - [return: NotNullIfNotNull(nameof(entityBatchResult))] - internal static P.EntityBatchResult? ToEntityBatchResult( - this EntityBatchResult? entityBatchResult, - string? completionToken = null, - IEnumerable? operationInfos = null) - { - if (entityBatchResult == null) - { - return null; - } - - return new P.EntityBatchResult() - { - EntityState = entityBatchResult.EntityState, - FailureDetails = entityBatchResult.FailureDetails.ToProtobuf(), - Actions = { entityBatchResult.Actions?.Select(a => a.ToOperationAction()) ?? [] }, - Results = { entityBatchResult.Results?.Select(a => a.ToOperationResult()) ?? [] }, - CompletionToken = completionToken ?? string.Empty, - OperationInfos = { operationInfos ?? [] }, - }; - } - - /// - /// Converts the gRPC representation of orchestrator entity parameters to the DT.Core representation. - /// - /// The DT.Core representation. - /// The gRPC representation. - [return: NotNullIfNotNull(nameof(parameters))] - internal static TaskOrchestrationEntityParameters? ToCore(this P.OrchestratorEntityParameters? parameters) - { - if (parameters == null) - { - return null; - } - - return new TaskOrchestrationEntityParameters() - { - EntityMessageReorderWindow = parameters.EntityMessageReorderWindow.ToTimeSpan(), - }; - } - - /// - /// Gets the approximate byte count for a . - /// - /// The failure details. - /// The approximate byte count. - internal static int GetApproximateByteCount(this P.TaskFailureDetails failureDetails) - { - // Protobuf strings are always UTF-8: https://developers.google.com/protocol-buffers/docs/proto3#scalar - Encoding encoding = Encoding.UTF8; - - int byteCount = 0; - if (failureDetails.ErrorType != null) - { - byteCount += encoding.GetByteCount(failureDetails.ErrorType); - } - - if (failureDetails.ErrorMessage != null) - { - byteCount += encoding.GetByteCount(failureDetails.ErrorMessage); - } - - if (failureDetails.StackTrace != null) - { - byteCount += encoding.GetByteCount(failureDetails.StackTrace); - } - - if (failureDetails.InnerFailure != null) - { - byteCount += failureDetails.InnerFailure.GetApproximateByteCount(); - } - - return byteCount; - } - - /// - /// Decode a protobuf message from a base64 string. - /// - /// The type to decode to. - /// The message parser. - /// The base64 encoded message. - /// The decoded message. - /// If decoding fails. - internal static T Base64Decode(this MessageParser parser, string encodedMessage) where T : IMessage - { - // Decode the base64 in a way that doesn't allocate a byte[] on each request - int encodedByteCount = Encoding.UTF8.GetByteCount(encodedMessage); - byte[] buffer = ArrayPool.Shared.Rent(encodedByteCount); - try - { - // The Base64 APIs require first converting the string into UTF-8 bytes. We then - // do an in-place conversion from base64 UTF-8 bytes to protobuf bytes so that - // we can finally decode the protobuf request. - Encoding.UTF8.GetBytes(encodedMessage, 0, encodedMessage.Length, buffer, 0); - OperationStatus status = Base64.DecodeFromUtf8InPlace( - buffer.AsSpan(0, encodedByteCount), - out int bytesWritten); - if (status != OperationStatus.Done) - { - throw new ArgumentException( - $"Failed to base64-decode the '{typeof(T).Name}' payload: {status}", nameof(encodedMessage)); - } - - return (T)parser.ParseFrom(buffer, 0, bytesWritten); - } - finally - { - ArrayPool.Shared.Return(buffer); - } - } - - /// - /// Converts a grpc to a . - /// - /// The failure details to convert. - /// The converted failure details. - internal static FailureDetails? ToCore(this P.TaskFailureDetails? failureDetails) - { - if (failureDetails == null) - { - return null; - } - - return new FailureDetails( - failureDetails.ErrorType, - failureDetails.ErrorMessage, - failureDetails.StackTrace, - failureDetails.InnerFailure.ToCore(), + }; + break; + } + + return action; + } + + /// + /// Converts a to a . + /// + /// The operation result to convert. + /// The converted operation result. + [return: NotNullIfNotNull(nameof(entityBatchResult))] + internal static EntityBatchResult? ToEntityBatchResult(this P.EntityBatchResult? entityBatchResult) + { + if (entityBatchResult == null) + { + return null; + } + + return new EntityBatchResult() + { + Actions = entityBatchResult.Actions.Select(operationAction => operationAction!.ToOperationAction()).ToList(), + EntityState = entityBatchResult.EntityState, + Results = entityBatchResult.Results.Select(operationResult => operationResult!.ToOperationResult()).ToList(), + FailureDetails = entityBatchResult.FailureDetails.ToCore(), + }; + } + + /// + /// Converts a to . + /// + /// The operation result to convert. + /// The completion token, or null for the older protocol. + /// Additional information about each operation, required by DTS. + /// The converted operation result. + [return: NotNullIfNotNull(nameof(entityBatchResult))] + internal static P.EntityBatchResult? ToEntityBatchResult( + this EntityBatchResult? entityBatchResult, + string? completionToken = null, + IEnumerable? operationInfos = null) + { + if (entityBatchResult == null) + { + return null; + } + + return new P.EntityBatchResult() + { + EntityState = entityBatchResult.EntityState, + FailureDetails = entityBatchResult.FailureDetails.ToProtobuf(), + Actions = { entityBatchResult.Actions?.Select(a => a.ToOperationAction()) ?? [] }, + Results = { entityBatchResult.Results?.Select(a => a.ToOperationResult()) ?? [] }, + CompletionToken = completionToken ?? string.Empty, + OperationInfos = { operationInfos ?? [] }, + }; + } + + /// + /// Converts the gRPC representation of orchestrator entity parameters to the DT.Core representation. + /// + /// The DT.Core representation. + /// The gRPC representation. + [return: NotNullIfNotNull(nameof(parameters))] + internal static TaskOrchestrationEntityParameters? ToCore(this P.OrchestratorEntityParameters? parameters) + { + if (parameters == null) + { + return null; + } + + return new TaskOrchestrationEntityParameters() + { + EntityMessageReorderWindow = parameters.EntityMessageReorderWindow.ToTimeSpan(), + }; + } + + /// + /// Gets the approximate byte count for a . + /// + /// The failure details. + /// The approximate byte count. + internal static int GetApproximateByteCount(this P.TaskFailureDetails failureDetails) + { + // Protobuf strings are always UTF-8: https://developers.google.com/protocol-buffers/docs/proto3#scalar + Encoding encoding = Encoding.UTF8; + + int byteCount = 0; + if (failureDetails.ErrorType != null) + { + byteCount += encoding.GetByteCount(failureDetails.ErrorType); + } + + if (failureDetails.ErrorMessage != null) + { + byteCount += encoding.GetByteCount(failureDetails.ErrorMessage); + } + + if (failureDetails.StackTrace != null) + { + byteCount += encoding.GetByteCount(failureDetails.StackTrace); + } + + if (failureDetails.InnerFailure != null) + { + byteCount += failureDetails.InnerFailure.GetApproximateByteCount(); + } + + return byteCount; + } + + /// + /// Decode a protobuf message from a base64 string. + /// + /// The type to decode to. + /// The message parser. + /// The base64 encoded message. + /// The decoded message. + /// If decoding fails. + internal static T Base64Decode(this MessageParser parser, string encodedMessage) where T : IMessage + { + // Decode the base64 in a way that doesn't allocate a byte[] on each request + int encodedByteCount = Encoding.UTF8.GetByteCount(encodedMessage); + byte[] buffer = ArrayPool.Shared.Rent(encodedByteCount); + try + { + // The Base64 APIs require first converting the string into UTF-8 bytes. We then + // do an in-place conversion from base64 UTF-8 bytes to protobuf bytes so that + // we can finally decode the protobuf request. + Encoding.UTF8.GetBytes(encodedMessage, 0, encodedMessage.Length, buffer, 0); + OperationStatus status = Base64.DecodeFromUtf8InPlace( + buffer.AsSpan(0, encodedByteCount), + out int bytesWritten); + if (status != OperationStatus.Done) + { + throw new ArgumentException( + $"Failed to base64-decode the '{typeof(T).Name}' payload: {status}", nameof(encodedMessage)); + } + + return (T)parser.ParseFrom(buffer, 0, bytesWritten); + } + finally + { + ArrayPool.Shared.Return(buffer); + } + } + + /// + /// Converts a grpc to a . + /// + /// The failure details to convert. + /// The converted failure details. + internal static FailureDetails? ToCore(this P.TaskFailureDetails? failureDetails) + { + if (failureDetails == null) + { + return null; + } + + return new FailureDetails( + failureDetails.ErrorType, + failureDetails.ErrorMessage, + failureDetails.StackTrace, + failureDetails.InnerFailure.ToCore(), failureDetails.IsNonRetriable, ConvertProperties(failureDetails.Properties)); } @@ -1120,28 +1120,28 @@ internal static Value ConvertObjectToValue(object? obj) // Fallback: convert unlisted type to string. _ => Value.ForString(obj.ToString() ?? string.Empty), }; - } - - /// - /// Converts a to a grpc . - /// - /// The failure details to convert. - /// The converted failure details. - static P.TaskFailureDetails? ToProtobuf(this FailureDetails? failureDetails) - { - if (failureDetails == null) - { - return null; - } - + } + + /// + /// Converts a to a grpc . + /// + /// The failure details to convert. + /// The converted failure details. + static P.TaskFailureDetails? ToProtobuf(this FailureDetails? failureDetails) + { + if (failureDetails == null) + { + return null; + } + var taskFailureDetails = new P.TaskFailureDetails - { - ErrorType = failureDetails.ErrorType ?? "(unknown)", - ErrorMessage = failureDetails.ErrorMessage ?? "(unknown)", - StackTrace = failureDetails.StackTrace, - IsNonRetriable = failureDetails.IsNonRetriable, - InnerFailure = failureDetails.InnerFailure.ToProtobuf(), - }; + { + ErrorType = failureDetails.ErrorType ?? "(unknown)", + ErrorMessage = failureDetails.ErrorMessage ?? "(unknown)", + StackTrace = failureDetails.StackTrace, + IsNonRetriable = failureDetails.IsNonRetriable, + InnerFailure = failureDetails.InnerFailure.ToProtobuf(), + }; // Properly populate the MapField if (failureDetails.Properties != null) @@ -1153,20 +1153,20 @@ internal static Value ConvertObjectToValue(object? obj) } return taskFailureDetails; - } - - static P.OrchestrationStatus ToProtobuf(this OrchestrationStatus status) - { - return (P.OrchestrationStatus)status; - } - - static P.OrchestrationInstance ToProtobuf(this OrchestrationInstance instance) - { - return new P.OrchestrationInstance - { - InstanceId = instance.InstanceId, - ExecutionId = instance.ExecutionId, - }; + } + + static P.OrchestrationStatus ToProtobuf(this OrchestrationStatus status) + { + return (P.OrchestrationStatus)status; + } + + static P.OrchestrationInstance ToProtobuf(this OrchestrationInstance instance) + { + return new P.OrchestrationInstance + { + InstanceId = instance.InstanceId, + ExecutionId = instance.ExecutionId, + }; } static P.HistoryEvent ToProtobuf(HistoryEvent e) @@ -1189,107 +1189,107 @@ static P.HistoryEvent ToProtobuf(HistoryEvent e) } throw new ArgumentException("Unsupported event type"); - } - - /// - /// Tracks state required for converting orchestration histories containing entity-related events. - /// - internal class EntityConversionState - { - readonly bool insertMissingEntityUnlocks; - - OrchestrationInstance? instance; - HashSet? entityRequestIds; - Dictionary? unlockObligations; - - /// - /// Initializes a new instance of the class. - /// - /// Whether to insert missing unlock events in to the history - /// when the orchestration completes. - public EntityConversionState(bool insertMissingEntityUnlocks) - { - this.ConvertFromProto = (P.HistoryEvent e) => ProtoUtils.ConvertHistoryEvent(e, this); - this.insertMissingEntityUnlocks = insertMissingEntityUnlocks; - } - - /// - /// Gets a function that converts a history event in protobuf format to a core history event. - /// - public Func ConvertFromProto { get; } - - /// - /// Gets the orchestration instance of this history. - /// - public OrchestrationInstance? CurrentInstance => this.instance; - - /// - /// Gets the set of guids that have been used as entity request ids in this history. - /// - public HashSet EntityRequestIds => this.entityRequestIds ??= new(); - - /// - /// Records the orchestration instance, which may be needed for some conversions. - /// - /// The orchestration instance. - public void SetOrchestrationInstance(OrchestrationInstance instance) - { - this.instance = instance; - } - - /// - /// Adds unlock obligations for all entities that are being locked by this request. - /// - /// The lock request. - public void AddUnlockObligations(P.EntityLockRequestedEvent request) - { - if (!this.insertMissingEntityUnlocks) - { - return; - } - - this.unlockObligations ??= new(); - - foreach (string target in request.LockSet) - { - this.unlockObligations[target] = request.CriticalSectionId; - } - } - - /// - /// Removes an unlock obligation. - /// - /// The target entity. - public void RemoveUnlockObligation(string target) - { - if (!this.insertMissingEntityUnlocks) - { - return; - } - - this.unlockObligations?.Remove(target); - } - - /// - /// Returns the remaining unlock obligations, and clears the list. - /// - /// The unlock obligations. - public IEnumerable<(string Target, string CriticalSectionId)> ResetObligations() - { - if (!this.insertMissingEntityUnlocks) - { - yield break; - } - - if (this.unlockObligations is not null) - { - foreach (var kvp in this.unlockObligations) - { - yield return (kvp.Key, kvp.Value); - } - - this.unlockObligations = null; - } - } - } -} + } + + /// + /// Tracks state required for converting orchestration histories containing entity-related events. + /// + internal class EntityConversionState + { + readonly bool insertMissingEntityUnlocks; + + OrchestrationInstance? instance; + HashSet? entityRequestIds; + Dictionary? unlockObligations; + + /// + /// Initializes a new instance of the class. + /// + /// Whether to insert missing unlock events in to the history + /// when the orchestration completes. + public EntityConversionState(bool insertMissingEntityUnlocks) + { + this.ConvertFromProto = (P.HistoryEvent e) => ProtoUtils.ConvertHistoryEvent(e, this); + this.insertMissingEntityUnlocks = insertMissingEntityUnlocks; + } + + /// + /// Gets a function that converts a history event in protobuf format to a core history event. + /// + public Func ConvertFromProto { get; } + + /// + /// Gets the orchestration instance of this history. + /// + public OrchestrationInstance? CurrentInstance => this.instance; + + /// + /// Gets the set of guids that have been used as entity request ids in this history. + /// + public HashSet EntityRequestIds => this.entityRequestIds ??= new(); + + /// + /// Records the orchestration instance, which may be needed for some conversions. + /// + /// The orchestration instance. + public void SetOrchestrationInstance(OrchestrationInstance instance) + { + this.instance = instance; + } + + /// + /// Adds unlock obligations for all entities that are being locked by this request. + /// + /// The lock request. + public void AddUnlockObligations(P.EntityLockRequestedEvent request) + { + if (!this.insertMissingEntityUnlocks) + { + return; + } + + this.unlockObligations ??= new(); + + foreach (string target in request.LockSet) + { + this.unlockObligations[target] = request.CriticalSectionId; + } + } + + /// + /// Removes an unlock obligation. + /// + /// The target entity. + public void RemoveUnlockObligation(string target) + { + if (!this.insertMissingEntityUnlocks) + { + return; + } + + this.unlockObligations?.Remove(target); + } + + /// + /// Returns the remaining unlock obligations, and clears the list. + /// + /// The unlock obligations. + public IEnumerable<(string Target, string CriticalSectionId)> ResetObligations() + { + if (!this.insertMissingEntityUnlocks) + { + yield break; + } + + if (this.unlockObligations is not null) + { + foreach (var kvp in this.unlockObligations) + { + yield return (kvp.Key, kvp.Value); + } + + this.unlockObligations = null; + } + } + } +}