From 17be2969f0d828f96632a74d30da48a6f757dffc Mon Sep 17 00:00:00 2001 From: Varshi Bachu Date: Wed, 11 Nov 2026 20:55:47 -0800 Subject: [PATCH] Add durable activity retry metadata and telemetry Carry validated replay-stable retry metadata through scheduled actions, history, and activity execution. Project retry events at orchestration completion with original timestamps and correlated allowlisted Application Insights telemetry. Add serialization, replay, dispatcher, and export regressions and document the existing generation boundary. --- docs/features/retries.md | 18 + docs/telemetry/application-insights.md | 12 + docs/telemetry/traces/semantic-conventions.md | 40 ++ .../DurableTask.ApplicationInsights.csproj | 3 +- .../DurableTelemetryModule.cs | 41 ++ .../Command/ScheduleTaskOrchestratorAction.cs | 9 + .../History/TaskScheduledEvent.cs | 10 + src/DurableTask.Core/OrchestrationContext.cs | 8 + src/DurableTask.Core/RetryInterceptor.cs | 28 +- src/DurableTask.Core/RetryMetadata.cs | 70 ++++ .../TaskActivityDispatcher.cs | 1 + src/DurableTask.Core/TaskContext.cs | 6 + .../TaskOrchestrationContext.cs | 23 +- .../TaskOrchestrationDispatcher.cs | 10 +- src/DurableTask.Core/Tracing/Schema.cs | 5 + src/DurableTask.Core/Tracing/TraceHelper.cs | 81 ++++ .../DataContractJsonConverterTests.cs | 32 ++ .../TableEntityConverterTests.cs | 30 ++ .../DispatcherMiddlewareTests.cs | 184 ++++++++- .../DurableTask.Core.Tests.csproj | 1 + .../DurableTelemetryModuleTests.cs | 222 ++++++++++ .../RetryMetadataTests.cs | 383 ++++++++++++++++++ 22 files changed, 1210 insertions(+), 7 deletions(-) create mode 100644 src/DurableTask.Core/RetryMetadata.cs create mode 100644 test/DurableTask.Core.Tests/DurableTelemetryModuleTests.cs create mode 100644 test/DurableTask.Core.Tests/RetryMetadataTests.cs diff --git a/docs/features/retries.md b/docs/features/retries.md index 2baad2623..8b69b5729 100644 --- a/docs/features/retries.md +++ b/docs/features/retries.md @@ -169,6 +169,24 @@ var result = await context.CreateSubOrchestrationInstanceWithRetry( ## Retry Behavior +### Activity Attempt Metadata + +For activity calls with a durable retry policy, `TaskContext.RetryMetadata` is available +starting with the initial attempt. `AttemptNumber` is one-based, `MaxAttempts` includes +the initial attempt, and `RetryGroupId` identifies the same logical activity across +attempts and orchestration replay. The group identifier is opaque; do not parse it. +Timeouts, exception filters, and non-retriable errors can stop before `MaxAttempts`. + +The metadata is stored on the scheduled action and `TaskScheduledEvent`, so a worker +restart or message redelivery preserves it. It is independent of `DeliveryAttempt`, +which counts transport deliveries rather than durable retries. Calls without a retry +policy and previously recorded histories without metadata expose `null`, not an +inferred first attempt. Sub-orchestration retry metadata is not exposed. + +`RetryMetadata` validates positive 32-bit attempt counts, `AttemptNumber <= MaxAttempts`, +and a nonempty group identifier. Malformed metadata is a contract error, not a reason +to substitute default values. + ### What Happens During Retry 1. Activity throws an exception diff --git a/docs/telemetry/application-insights.md b/docs/telemetry/application-insights.md index bab234285..166f4d432 100644 --- a/docs/telemetry/application-insights.md +++ b/docs/telemetry/application-insights.md @@ -109,6 +109,18 @@ The `DurableTelemetryModule` automatically tracks: ### Distributed Tracing +`DurableTelemetryModule` maps `durabletask.task.retry.scheduled` orchestration-span +events to correlated `EventTelemetry` (`customEvents`). These are exported only +when the orchestration span completes, retaining the original retry scheduling +timestamps; they are not live notifications while the orchestration runs. +Only the six retry-event attributes in the +[trace semantic conventions](traces/semantic-conventions.md#activity-retry-metadata) +are copied. Inputs, error messages, stack traces, and arbitrary event tags/baggage +are not added to these retry events. Existing request and dependency export is unchanged. +Reconstruction is limited to retained generation history. The existing inherited-trace +`ContinueAsNew` history reset does not retain prior-generation orchestration retry +events for later export; it does not change activity metadata or activity spans. + Operations are automatically correlated: ```text diff --git a/docs/telemetry/traces/semantic-conventions.md b/docs/telemetry/traces/semantic-conventions.md index f7c633fdc..4e6616f21 100644 --- a/docs/telemetry/traces/semantic-conventions.md +++ b/docs/telemetry/traces/semantic-conventions.md @@ -41,6 +41,46 @@ durabletask.event.instance_id | string | The target orchestration instance ID of [2] When instance ID is not available (e.g. activities) use the parent orchestration's instance ID ### Spans + +#### Activity retry metadata + +Activity client and server spans include the following attributes only when the +scheduled activity carries durable retry metadata, including on attempt 1: + +| Attribute | Type | Meaning | +|---|---|---| +| `durabletask.task.retry.attempt` | integer | One-based durable attempt number | +| `durabletask.task.retry.group_id` | string | Opaque replay-stable logical activity identifier | +| `durabletask.task.retry.max_attempts` | integer | Maximum attempts, including the initial attempt | + +The orchestration span includes one `durabletask.task.retry.scheduled` event for +each persisted retry activity schedule, excluding attempt 1. Events are projected +from the saved history when the orchestration span is exported at completion, +using the original `TaskScheduledEvent.Timestamp`. They are **not visible while +the orchestration is still running**. Replay does not export extra events, and +stopping at a retry limit, exception filter, or timeout without scheduling another +activity does not produce an event. Its attributes are `durabletask.task.name`, +`durabletask.task.task_id` (the next scheduled task ID), the three retry attributes +above (describing the next attempt), and `durabletask.task.failure.type` +(the previous attempt's error type). Retry telemetry does not include activity +inputs, failure messages, or stack traces. Transport redelivery does not increment +the durable attempt number. + +Projection uses only the generation history retained at span export. The existing +`ContinueAsNew` history reset and span lifecycle are unchanged: when the same trace +is inherited, prior-generation orchestration retry events are not retained for a +later final-span reconstruction. With `ContinueAsNewTraceBehavior.StartNewTrace`, +the completed generation's events are projected before its existing span stops. +Activity retry metadata and activity spans are independent of this +orchestration-event retention boundary; no cross-generation telemetry state is added. + +The failure type comes from the previous attempt's structured `FailureDetails`, +or from the sanitized exception type captured by normal orchestration replay. +If neither is available (for example, an opaque legacy error whose original type +was not propagated), the value is `DurableTask.Core.Exceptions.TaskFailedException`. +It does not claim to identify the original user exception in that case. Telemetry +never deserializes legacy error payloads or parses messages/stacks to infer a type. + #### Client: Starting an Orchestration Represents enqueueing a new orchestration via TaskHubClient to the backend. diff --git a/src/DurableTask.ApplicationInsights/DurableTask.ApplicationInsights.csproj b/src/DurableTask.ApplicationInsights/DurableTask.ApplicationInsights.csproj index 8e82bdadd..0107364c2 100644 --- a/src/DurableTask.ApplicationInsights/DurableTask.ApplicationInsights.csproj +++ b/src/DurableTask.ApplicationInsights/DurableTask.ApplicationInsights.csproj @@ -20,7 +20,8 @@ $(MajorVersion).$(MinorVersion).0.0 - $(VersionPrefix) + $(VersionPrefix) + $(VersionPrefix)-$(VersionSuffix) diff --git a/src/DurableTask.ApplicationInsights/DurableTelemetryModule.cs b/src/DurableTask.ApplicationInsights/DurableTelemetryModule.cs index bb585c708..22a16ba0c 100644 --- a/src/DurableTask.ApplicationInsights/DurableTelemetryModule.cs +++ b/src/DurableTask.ApplicationInsights/DurableTelemetryModule.cs @@ -16,6 +16,7 @@ namespace DurableTask.ApplicationInsights using System; using System.Collections.Generic; using System.Diagnostics; + using System.Globalization; using System.Threading; using System.Threading.Tasks; using Microsoft.ApplicationInsights; @@ -58,6 +59,46 @@ void OnEndActivity(Activity activity) OperationTelemetry telemetry = CreateTelemetry(activity); this.telemetryClient.Track(telemetry); + + if (activity.Kind != ActivityKind.Server + || !Equals(activity.GetTagItem("durabletask.type"), "orchestration")) + { + return; + } + + foreach (ActivityEvent activityEvent in activity.Events) + { + if (activityEvent.Name != "durabletask.task.retry.scheduled") + { + continue; + } + + var retryEvent = new EventTelemetry(activityEvent.Name) + { + Timestamp = activityEvent.Timestamp, + }; + retryEvent.Context.Operation.Id = activity.TraceId.ToString(); + retryEvent.Context.Operation.ParentId = activity.SpanId.ToString(); + foreach (KeyValuePair tag in activityEvent.Tags) + { + switch (tag.Key) + { + case "durabletask.task.name": + case "durabletask.task.task_id": + case "durabletask.task.retry.attempt": + case "durabletask.task.retry.group_id": + case "durabletask.task.retry.max_attempts": + case "durabletask.task.failure.type": + if (tag.Value != null) + { + retryEvent.Properties[tag.Key] = Convert.ToString(tag.Value, CultureInfo.InvariantCulture); + } + break; + } + } + + this.telemetryClient.TrackEvent(retryEvent); + } } static OperationTelemetry CreateTelemetry(Activity activity) diff --git a/src/DurableTask.Core/Command/ScheduleTaskOrchestratorAction.cs b/src/DurableTask.Core/Command/ScheduleTaskOrchestratorAction.cs index 4e3c0bf85..191a4f4fb 100644 --- a/src/DurableTask.Core/Command/ScheduleTaskOrchestratorAction.cs +++ b/src/DurableTask.Core/Command/ScheduleTaskOrchestratorAction.cs @@ -48,5 +48,14 @@ public class ScheduleTaskOrchestratorAction : OrchestratorAction /// Gets or sets a dictionary of tags of string, string /// public IDictionary? Tags { get; set; } + + /// + /// Gets or sets durable retry metadata, or null for an activity without a retry policy. + /// + [Newtonsoft.Json.JsonProperty(NullValueHandling = Newtonsoft.Json.NullValueHandling.Ignore)] + public RetryMetadata? RetryMetadata { get; set; } + + [Newtonsoft.Json.JsonIgnore] + internal string? RetryFailureType { get; set; } } } \ No newline at end of file diff --git a/src/DurableTask.Core/History/TaskScheduledEvent.cs b/src/DurableTask.Core/History/TaskScheduledEvent.cs index 73f84ce15..74dfb5fe1 100644 --- a/src/DurableTask.Core/History/TaskScheduledEvent.cs +++ b/src/DurableTask.Core/History/TaskScheduledEvent.cs @@ -91,5 +91,15 @@ public TaskScheduledEvent(int eventId) /// [DataMember] public IDictionary? Tags { get; set; } + + /// + /// Gets or sets durable retry metadata, or null for uninstrumented or non-retry activity calls. + /// + [DataMember(EmitDefaultValue = false)] + [Newtonsoft.Json.JsonProperty(NullValueHandling = Newtonsoft.Json.NullValueHandling.Ignore)] + public RetryMetadata? RetryMetadata { get; set; } + + [Newtonsoft.Json.JsonIgnore] + internal string? RetryFailureType { get; set; } } } \ No newline at end of file diff --git a/src/DurableTask.Core/OrchestrationContext.cs b/src/DurableTask.Core/OrchestrationContext.cs index 1f405a6ec..10172ad58 100644 --- a/src/DurableTask.Core/OrchestrationContext.cs +++ b/src/DurableTask.Core/OrchestrationContext.cs @@ -31,6 +31,14 @@ public abstract class OrchestrationContext /// private static readonly ProxyGenerator ProxyGenerator = new ProxyGenerator(); + readonly System.Threading.AsyncLocal activityRetryContext = new System.Threading.AsyncLocal(); + + internal ActivityRetryContext ActivityRetryContext + { + get => this.activityRetryContext.Value; + set => this.activityRetryContext.Value = value; + } + /// /// Thread-static variable used to signal whether the calling thread is the orchestrator thread. /// The primary use case is for detecting illegal async usage in orchestration code. diff --git a/src/DurableTask.Core/RetryInterceptor.cs b/src/DurableTask.Core/RetryInterceptor.cs index 2f5617ab2..c598e6f30 100644 --- a/src/DurableTask.Core/RetryInterceptor.cs +++ b/src/DurableTask.Core/RetryInterceptor.cs @@ -53,12 +53,27 @@ public RetryInterceptor(OrchestrationContext context, RetryOptions retryOptions, { Exception? lastException = null; DateTime firstAttempt = this.context.CurrentUtcDateTime; + var retryContext = new ActivityRetryContext { MaxAttempts = this.retryOptions.MaxNumberOfAttempts }; for (var retryCount = 0; retryCount < this.retryOptions.MaxNumberOfAttempts; retryCount++) { try { - return await this.retryCall(); + retryContext.AttemptNumber = retryCount + 1; + retryContext.PreviousFailure = lastException; + ActivityRetryContext? previousContext = this.context.ActivityRetryContext; + Task attempt; + try + { + this.context.ActivityRetryContext = retryContext; + attempt = this.retryCall(); + } + finally + { + this.context.ActivityRetryContext = previousContext; + } + + return await attempt; } catch (Exception e) when (!Utils.IsFatal(e)) { @@ -129,4 +144,15 @@ TimeSpan ComputeNextDelay(int attempt, DateTime firstAttempt, Exception failure) return nextDelay; } } + + internal sealed class ActivityRetryContext + { + internal int AttemptNumber { get; set; } + + internal int MaxAttempts { get; set; } + + internal string? RetryGroupId { get; set; } + + internal Exception? PreviousFailure { get; set; } + } } \ No newline at end of file diff --git a/src/DurableTask.Core/RetryMetadata.cs b/src/DurableTask.Core/RetryMetadata.cs new file mode 100644 index 000000000..746da042b --- /dev/null +++ b/src/DurableTask.Core/RetryMetadata.cs @@ -0,0 +1,70 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. + +#nullable enable +namespace DurableTask.Core +{ + using System; + using System.Runtime.Serialization; + using Newtonsoft.Json; + + /// + /// Identifies an activity attempt scheduled by a durable retry policy. + /// + [DataContract] + public sealed class RetryMetadata + { + /// + /// Initializes validated metadata for an activity attempt. + /// + /// The one-based durable attempt number. + /// The maximum number of attempts, including the initial attempt. + /// The replay-stable identifier shared by all attempts of the logical activity. + [JsonConstructor] + public RetryMetadata(int attemptNumber, int maxAttempts, string retryGroupId) + { + this.AttemptNumber = attemptNumber; + this.MaxAttempts = maxAttempts; + this.RetryGroupId = retryGroupId; + this.Validate(); + } + + /// + /// Gets the one-based durable attempt number, independent of message redelivery. + /// + [DataMember] + public int AttemptNumber { get; private set; } + + /// + /// Gets the policy's maximum number of attempts, including the initial attempt. + /// + [DataMember] + public int MaxAttempts { get; private set; } + + /// + /// Gets the identifier shared by attempts of the logical activity across replay. + /// + [DataMember] + public string RetryGroupId { get; private set; } + + [OnDeserialized] + void OnDeserialized(StreamingContext context) => this.Validate(); + + void Validate() + { + if (this.MaxAttempts <= 0) + { + throw new ArgumentOutOfRangeException("maxAttempts", "The maximum attempt count must be positive."); + } + + if (this.AttemptNumber <= 0 || this.AttemptNumber > this.MaxAttempts) + { + throw new ArgumentOutOfRangeException("attemptNumber", "The attempt must be between one and the maximum attempt count."); + } + + if (string.IsNullOrWhiteSpace(this.RetryGroupId)) + { + throw new ArgumentException("A retry group identifier is required.", "retryGroupId"); + } + } + } +} diff --git a/src/DurableTask.Core/TaskActivityDispatcher.cs b/src/DurableTask.Core/TaskActivityDispatcher.cs index 6494b59ac..267cc376f 100644 --- a/src/DurableTask.Core/TaskActivityDispatcher.cs +++ b/src/DurableTask.Core/TaskActivityDispatcher.cs @@ -208,6 +208,7 @@ await this.dispatchPipeline.RunAsync(dispatchContext, async _ => scheduledEvent.EventId) { DeliveryAttempt = deliveryAttempt, + RetryMetadata = scheduledEvent.RetryMetadata, }; context.ErrorPropagationMode = this.errorPropagationMode; context.ExceptionPropertiesProvider = this.exceptionPropertiesProvider; diff --git a/src/DurableTask.Core/TaskContext.cs b/src/DurableTask.Core/TaskContext.cs index 975e05327..b16ec8ff6 100644 --- a/src/DurableTask.Core/TaskContext.cs +++ b/src/DurableTask.Core/TaskContext.cs @@ -69,6 +69,12 @@ public TaskContext(OrchestrationInstance orchestrationInstance, string name, str /// public long? DeliveryAttempt { get; internal set; } + /// + /// Gets durable retry metadata, or null when the activity was scheduled without it. + /// Message redelivery and worker restarts do not change this value. + /// + public RetryMetadata? RetryMetadata { get; internal set; } + /// /// Gets or sets a value indicating how to propagate unhandled exception metadata. /// diff --git a/src/DurableTask.Core/TaskOrchestrationContext.cs b/src/DurableTask.Core/TaskOrchestrationContext.cs index fc3b77e7e..34c8c6a67 100644 --- a/src/DurableTask.Core/TaskOrchestrationContext.cs +++ b/src/DurableTask.Core/TaskOrchestrationContext.cs @@ -94,7 +94,13 @@ public override async Task ScheduleTask(string name, string ve { if (options.RetryOptions != null) { - Task RetryCall() => ScheduleTask(name, version, ScheduleTaskOptions.CreateBuilder().WithTags(options.Tags).Build(), parameters); + var attemptOptions = ScheduleTaskOptions.CreateBuilder(); + if (options.Tags != null) + { + attemptOptions.WithTags(options.Tags); + } + + Task RetryCall() => ScheduleTask(name, version, attemptOptions.Build(), parameters); var retryInterceptor = new RetryInterceptor(this, options.RetryOptions, RetryCall); return await retryInterceptor.Invoke(); } @@ -138,6 +144,19 @@ public async Task ScheduleTaskInternal(string name, string version, stri Tags = options.Tags, }; + ActivityRetryContext retryContext = this.ActivityRetryContext; + if (retryContext != null) + { + retryContext.RetryGroupId ??= this.OrchestrationInstance.ExecutionId + ":" + id.ToString(CultureInfo.InvariantCulture); + scheduleTaskTaskAction.RetryMetadata = new RetryMetadata( + retryContext.AttemptNumber, + retryContext.MaxAttempts, + retryContext.RetryGroupId); + scheduleTaskTaskAction.RetryFailureType = retryContext.PreviousFailure == null + ? null + : TraceHelper.GetTaskFailureType(retryContext.PreviousFailure); + } + this.orchestratorActionsMap.Add(id, scheduleTaskTaskAction); var tcs = new TaskCompletionSource(); @@ -355,6 +374,8 @@ public void HandleTaskScheduledEvent(TaskScheduledEvent scheduledEvent) + "orchestrator code after this instance had already started running?"); } + // Keep only the sanitized type in memory for completion-time telemetry projection. + scheduledEvent.RetryFailureType = currentReplayAction.RetryFailureType; this.orchestratorActionsMap.Remove(taskId); } diff --git a/src/DurableTask.Core/TaskOrchestrationDispatcher.cs b/src/DurableTask.Core/TaskOrchestrationDispatcher.cs index 492cd2e44..9a6ad67f3 100644 --- a/src/DurableTask.Core/TaskOrchestrationDispatcher.cs +++ b/src/DurableTask.Core/TaskOrchestrationDispatcher.cs @@ -1140,6 +1140,7 @@ internal static bool ReconcileMessagesWithState(TaskOrchestrationWorkItem workIt private void ResetDistributedTraceActivity(OrchestrationRuntimeState runtimeState) { TraceHelper.SetRuntimeStatusTag(runtimeState.OrchestrationStatus.ToString()); + TraceHelper.AddTaskRetryEvents(DistributedTraceActivity.Current, runtimeState.Events); DistributedTraceActivity.Current?.Stop(); DistributedTraceActivity.Current = null; } @@ -1155,6 +1156,7 @@ private void ResetDistributedTraceActivity(OrchestrationRuntimeState runtimeStat } TraceHelper.SetRuntimeStatusTag(completedRuntimeState.OrchestrationStatus.ToString()); + TraceHelper.AddTaskRetryEvents(currentTraceActivity, completedRuntimeState.Events); currentTraceActivity?.Stop(); DistributedTraceActivity.Current = null; @@ -1207,7 +1209,9 @@ TaskMessage ProcessScheduleTaskDecision( version: scheduleTaskOrchestratorAction.Version, input: scheduleTaskOrchestratorAction.Input) { - Tags = scheduleTaskOrchestratorAction.Tags + Tags = scheduleTaskOrchestratorAction.Tags, + RetryMetadata = scheduleTaskOrchestratorAction.RetryMetadata, + RetryFailureType = scheduleTaskOrchestratorAction.RetryFailureType, }; ActivitySpanId clientSpanId = ActivitySpanId.CreateRandom(); @@ -1229,7 +1233,9 @@ TaskMessage ProcessScheduleTaskDecision( name: scheduleTaskOrchestratorAction.Name, version: scheduleTaskOrchestratorAction.Version) { - Tags = scheduleTaskOrchestratorAction.Tags + Tags = scheduleTaskOrchestratorAction.Tags, + RetryMetadata = scheduleTaskOrchestratorAction.RetryMetadata, + RetryFailureType = scheduleTaskOrchestratorAction.RetryFailureType, }; if (parentTraceActivity != null) diff --git a/src/DurableTask.Core/Tracing/Schema.cs b/src/DurableTask.Core/Tracing/Schema.cs index 647726221..c10ba8e11 100644 --- a/src/DurableTask.Core/Tracing/Schema.cs +++ b/src/DurableTask.Core/Tracing/Schema.cs @@ -29,6 +29,11 @@ internal static class Task internal const string Operation = "durabletask.task.operation"; internal const string ScheduledTime = "durabletask.task.scheduled_time"; internal const string ErrorMessage = "durabletask.entity.error_message"; + internal const string RetryAttempt = "durabletask.task.retry.attempt"; + internal const string RetryGroupId = "durabletask.task.retry.group_id"; + internal const string RetryMaxAttempts = "durabletask.task.retry.max_attempts"; + internal const string FailureType = "durabletask.task.failure.type"; + internal const string RetryScheduled = "durabletask.task.retry.scheduled"; } internal static class Status diff --git a/src/DurableTask.Core/Tracing/TraceHelper.cs b/src/DurableTask.Core/Tracing/TraceHelper.cs index 13a7a0cb9..08190f66a 100644 --- a/src/DurableTask.Core/Tracing/TraceHelper.cs +++ b/src/DurableTask.Core/Tracing/TraceHelper.cs @@ -25,6 +25,7 @@ namespace DurableTask.Core.Tracing using DurableTask.Core.Entities; using DurableTask.Core.Entities.OperationFormat; using System.Linq; + using DurableTask.Core.Exceptions; /// /// Helper class for logging/tracing @@ -212,6 +213,7 @@ private static DateTimeOffset GetActivityStartTime(DateTime timestamp) newActivity.SetTag(Schema.Task.Name, scheduledEvent.Name); newActivity.SetTag(Schema.Task.InstanceId, instance.InstanceId); newActivity.SetTag(Schema.Task.TaskId, scheduledEvent.EventId); + SetRetryTags(newActivity, scheduledEvent.RetryMetadata); if (!string.IsNullOrEmpty(scheduledEvent.Version)) { @@ -262,6 +264,7 @@ private static DateTimeOffset GetActivityStartTime(DateTime timestamp) newActivity.AddTag(Schema.Task.Name, taskScheduledEvent.Name); newActivity.AddTag(Schema.Task.InstanceId, instance?.InstanceId); newActivity.AddTag(Schema.Task.TaskId, taskScheduledEvent.EventId); + SetRetryTags(newActivity, taskScheduledEvent.RetryMetadata); if (!string.IsNullOrEmpty(taskScheduledEvent.Version)) { @@ -271,6 +274,84 @@ private static DateTimeOffset GetActivityStartTime(DateTime timestamp) return newActivity; } + static void SetRetryTags(Activity activity, RetryMetadata? metadata) + { + if (metadata != null) + { + activity.SetTag(Schema.Task.RetryAttempt, metadata.AttemptNumber); + activity.SetTag(Schema.Task.RetryGroupId, metadata.RetryGroupId); + activity.SetTag(Schema.Task.RetryMaxAttempts, metadata.MaxAttempts); + } + } + + internal static string? GetTaskFailureType(Exception failure) => + (failure as OrchestrationException)?.FailureDetails?.ErrorType + ?? failure.InnerException?.GetType().FullName + ?? failure.GetType().FullName; + + internal static void AddTaskRetryEvents(Activity? activity, IEnumerable history) + { + if (activity == null) + { + return; + } + + var emitted = new HashSet<(int TaskId, string GroupId)>(); + foreach (ActivityEvent existing in activity.Events) + { + if (existing.Name == Schema.Task.RetryScheduled) + { + var tags = existing.Tags.ToDictionary(tag => tag.Key, tag => tag.Value); + if (tags.TryGetValue(Schema.Task.TaskId, out object? taskId) && taskId is int id + && tags.TryGetValue(Schema.Task.RetryGroupId, out object? groupId) && groupId is string group) + { + emitted.Add((id, group)); + } + } + } + + var previousAttempts = new Dictionary(StringComparer.Ordinal); + var failures = new Dictionary(); + foreach (HistoryEvent historyEvent in history) + { + if (historyEvent is TaskFailedEvent failed) + { + failures[failed.TaskScheduledId] = failed; + } + else if (historyEvent is TaskScheduledEvent scheduled && scheduled.RetryMetadata is RetryMetadata metadata) + { + if (metadata.AttemptNumber > 1 && emitted.Add((scheduled.EventId, metadata.RetryGroupId))) + { + TaskFailedEvent? previousFailure = null; + if (previousAttempts.TryGetValue(metadata.RetryGroupId, out TaskScheduledEvent? previous) + && previous.RetryMetadata!.AttemptNumber == metadata.AttemptNumber - 1) + { + failures.TryGetValue(previous.EventId, out previousFailure); + } + + // Never deserialize opaque legacy Details just for telemetry. + string? failureType = previousFailure?.FailureDetails?.ErrorType + ?? scheduled.RetryFailureType + ?? typeof(TaskFailedException).FullName; + activity.AddEvent(new ActivityEvent( + Schema.Task.RetryScheduled, + timestamp: GetActivityStartTime(scheduled.Timestamp), + tags: new ActivityTagsCollection + { + { Schema.Task.Name, scheduled.Name }, + { Schema.Task.TaskId, scheduled.EventId }, + { Schema.Task.RetryAttempt, metadata.AttemptNumber }, + { Schema.Task.RetryGroupId, metadata.RetryGroupId }, + { Schema.Task.RetryMaxAttempts, metadata.MaxAttempts }, + { Schema.Task.FailureType, failureType }, + })); + } + + previousAttempts[metadata.RetryGroupId] = scheduled; + } + } + } + /// /// Emits a new trace activity for a (task) activity that successfully completes. /// diff --git a/test/DurableTask.AzureStorage.Tests/DataContractJsonConverterTests.cs b/test/DurableTask.AzureStorage.Tests/DataContractJsonConverterTests.cs index fa28622f6..7bce36841 100644 --- a/test/DurableTask.AzureStorage.Tests/DataContractJsonConverterTests.cs +++ b/test/DurableTask.AzureStorage.Tests/DataContractJsonConverterTests.cs @@ -71,6 +71,38 @@ public void ReadWrite_TaskMessage_Succeeds() VerifyEqual(instance, actual); } + [TestMethod] + public void ReadWrite_ActivityRetryMetadata_PreservesAttemptOnRedelivery() + { + var message = new TaskMessage + { + Event = new TaskScheduledEvent(17, "work") + { + RetryMetadata = new RetryMetadata(2, 5, "execution:0"), + }, + OrchestrationInstance = new OrchestrationInstance { InstanceId = "instance", ExecutionId = "execution" }, + }; + for (int delivery = 0; delivery < 3; delivery++) + { + message = Deserialize(Serialize(message)); + var scheduled = (TaskScheduledEvent)message.Event; + Assert.AreEqual(2, scheduled.RetryMetadata.AttemptNumber); + Assert.AreEqual(5, scheduled.RetryMetadata.MaxAttempts); + Assert.AreEqual("execution:0", scheduled.RetryMetadata.RetryGroupId); + } + } + + [TestMethod] + public void Read_ActivityRetryMetadata_RejectsInvalidContract() + { + var scheduled = new TaskScheduledEvent(17, "work") + { + RetryMetadata = new RetryMetadata(2, 5, "execution:0"), + }; + string json = Serialize(scheduled).Replace("\"AttemptNumber\":2", "\"AttemptNumber\":0"); + Assert.ThrowsException(() => Deserialize(json)); + } + private static void VerifyEqual(TaskMessage expected, TaskMessage actual) { Assert.IsNotNull(actual); diff --git a/test/DurableTask.AzureStorage.Tests/TableEntityConverterTests.cs b/test/DurableTask.AzureStorage.Tests/TableEntityConverterTests.cs index f0c4a5c1f..076346911 100644 --- a/test/DurableTask.AzureStorage.Tests/TableEntityConverterTests.cs +++ b/test/DurableTask.AzureStorage.Tests/TableEntityConverterTests.cs @@ -21,12 +21,42 @@ namespace DurableTask.AzureStorage.Tests using Azure.Data.Tables; using DurableTask.AzureStorage.Tests.Obsolete; using DurableTask.AzureStorage.Tracking; + using DurableTask.Core; + using DurableTask.Core.History; using Microsoft.VisualStudio.TestTools.UnitTesting; using Microsoft.WindowsAzure.Storage; [TestClass] public class TableEntityConverterTests { + [TestMethod] + public void ActivityRetryMetadataRoundTripsInHistoryTable() + { + var expected = new TaskScheduledEvent(42, "work") + { + RetryMetadata = new RetryMetadata(2, 5, "execution:17"), + }; + TableEntity entity = TableEntityConverter.Serialize(expected); + Assert.IsTrue(entity.GetString(nameof(TaskScheduledEvent.RetryMetadata)).Contains("\"AttemptNumber\":2")); + var actual = (TaskScheduledEvent)TableEntityConverter.Deserialize(entity, typeof(TaskScheduledEvent)); + Assert.AreEqual(2, actual.RetryMetadata.AttemptNumber); + Assert.AreEqual(5, actual.RetryMetadata.MaxAttempts); + Assert.AreEqual("execution:17", actual.RetryMetadata.RetryGroupId); + + entity.Remove(nameof(TaskScheduledEvent.RetryMetadata)); + actual = (TaskScheduledEvent)TableEntityConverter.Deserialize(entity, typeof(TaskScheduledEvent)); + Assert.IsNull(actual.RetryMetadata); + } + + [TestMethod] + public void ActivityRetryMetadataRejectsCorruptHistory() + { + var entity = TableEntityConverter.Serialize(new TaskScheduledEvent(42, "work")); + entity[nameof(TaskScheduledEvent.RetryMetadata)] = + "{\"AttemptNumber\":0,\"MaxAttempts\":5,\"RetryGroupId\":\"execution:17\"}"; + Assert.ThrowsException(() => TableEntityConverter.Deserialize(entity, typeof(TaskScheduledEvent))); + } + [TestMethod] public void DeserializeNonNull() { diff --git a/test/DurableTask.Core.Tests/DispatcherMiddlewareTests.cs b/test/DurableTask.Core.Tests/DispatcherMiddlewareTests.cs index cf05a71f4..18f46e3bd 100644 --- a/test/DurableTask.Core.Tests/DispatcherMiddlewareTests.cs +++ b/test/DurableTask.Core.Tests/DispatcherMiddlewareTests.cs @@ -27,6 +27,8 @@ namespace DurableTask.Core.Tests using DurableTask.Core.Command; using DurableTask.Core.History; using DurableTask.Core.Tracing; + using DurableTask.ApplicationInsights; + using Microsoft.ApplicationInsights.DataContracts; using DurableTask.Emulator; using DurableTask.Test.Orchestrations; using Microsoft.Extensions.Logging; @@ -42,6 +44,7 @@ public class DispatcherMiddlewareTests TaskHubClient client = null!; DeliveryAttemptOrchestrationService service = null!; DeliveryAttemptActivity deliveryAttemptActivity = null!; + RetryMetadataActivity retryMetadataActivity = null!; [TestInitialize] public void InitializeTests() @@ -54,6 +57,7 @@ public void InitializeTests() }); this.service = new DeliveryAttemptOrchestrationService(); this.deliveryAttemptActivity = new DeliveryAttemptActivity(); + this.retryMetadataActivity = new RetryMetadataActivity(); this.worker = new TaskHubWorker(this.service, loggerFactory); // We use `GetAwaiter().GetResult()` because otherwise this method will fail with: @@ -64,12 +68,13 @@ public void InitializeTests() typeof(ParentWorkflow), typeof(ChildWorkflow), typeof(ActivityStatusResetOrchestration), - typeof(DeliveryAttemptOrchestration)) + typeof(DeliveryAttemptOrchestration), + typeof(RetryMetadataOrchestration)) .AddTaskActivities( typeof(SimplestGetUserTask), typeof(SimplestSendGreetingTask), typeof(ActivityStatusResetActivity)) - .AddTaskActivities(this.deliveryAttemptActivity) + .AddTaskActivities(this.deliveryAttemptActivity, this.retryMetadataActivity) .StartAsync().GetAwaiter().GetResult(); this.client = new TaskHubClient(this.service); @@ -149,6 +154,139 @@ public void TaskActivityDeliveryAttemptDefaultsToNull() Assert.IsNull(metadata.DeliveryAttempt); } + [DataTestMethod] + [DataRow(false, false, false)] + [DataRow(true, false, false)] + [DataRow(false, true, false)] + [DataRow(true, true, false)] + [DataRow(false, false, true)] + [DataRow(true, true, true)] + public async Task DurableRetryMetadataFlowsThroughDispatchersAndTelemetry(bool includeParameters, bool includeFailureDetails, bool fail) + { + using var collector = new DurableActivityCollector(); + using var channel = new CapturingTelemetryChannel(); + using var configuration = DurableTelemetryModuleTests.CreateConfiguration(channel); + await using var module = new DurableTelemetryModule(); + module.Initialize(configuration); + this.worker.TaskOrchestrationDispatcher.IncludeParameters = includeParameters; + this.worker.TaskActivityDispatcher.IncludeDetails = includeFailureDetails; + this.service.ActivityDeliveryAttempt = 42; + var completionGate = new TaskCompletionSource(); + this.retryMetadataActivity.CompletionGate = completionGate.Task; + var history = new ConcurrentDictionary(); + this.worker.AddOrchestrationDispatcherMiddleware(async (context, next) => + { + foreach (TaskScheduledEvent scheduled in context.GetProperty().Events.OfType()) + { + history[scheduled.EventId] = scheduled; + } + + await next(); + }); + + OrchestrationInstance instance = await this.client.CreateOrchestrationInstanceAsync( + typeof(RetryMetadataOrchestration), fail ? "fail" : "secret input"); + try + { + Task started = this.retryMetadataActivity.FinalAttemptStarted.Task; + Assert.AreSame(started, await Task.WhenAny(started, Task.Delay(TimeSpan.FromSeconds(15)))); + Assert.IsFalse(channel.Items.OfType().Any(), "Retry events are not live notifications."); + Assert.IsFalse(collector.Activities.Any(a => a.Kind == ActivityKind.Server + && Equals(a.GetTagItem(Schema.Task.Type), TraceActivityConstants.Orchestration)), + "Do not prematurely stop or duplicate orchestration spans."); + } + finally + { + completionGate.TrySetResult(true); + } + + OrchestrationState state = await this.client.WaitForOrchestrationAsync(instance, TimeSpan.FromSeconds(15)); + Assert.AreEqual(fail ? OrchestrationStatus.Failed : OrchestrationStatus.Completed, state.OrchestrationStatus); + + TaskContext[] contexts = this.retryMetadataActivity.Contexts.OrderBy(c => c.RetryMetadata!.AttemptNumber).ToArray(); + Assert.AreEqual(3, contexts.Length); + Assert.AreEqual(3, history.Count); + string group = contexts[0].RetryMetadata!.RetryGroupId; + for (int i = 0; i < contexts.Length; i++) + { + Assert.AreEqual(i + 1, contexts[i].RetryMetadata!.AttemptNumber); + Assert.AreEqual(3, contexts[i].RetryMetadata!.MaxAttempts); + Assert.AreEqual(group, contexts[i].RetryMetadata!.RetryGroupId); + Assert.AreEqual(42L, contexts[i].DeliveryAttempt); + TaskScheduledEvent scheduled = history[contexts[i].TaskId]; + Assert.AreEqual(group, scheduled.RetryMetadata!.RetryGroupId); + Assert.AreEqual(i + 1, scheduled.RetryMetadata.AttemptNumber); + Assert.AreEqual(includeParameters, scheduled.Input != null); + } + + ActivityEvent[] events = collector.Activities + .SelectMany(a => a.Events) + .Where(e => e.Name == Schema.Task.RetryScheduled) + .OrderBy(e => (int)e.Tags.Single(t => t.Key == Schema.Task.RetryAttempt).Value!) + .ToArray(); + Assert.AreEqual(2, events.Length, "Replay must not duplicate retry events."); + Activity orchestrationSpan = collector.Activities.Single(a => + a.Kind == ActivityKind.Server && Equals(a.GetTagItem(Schema.Task.Type), TraceActivityConstants.Orchestration)); + Assert.AreEqual(2, orchestrationSpan.Events.Count(e => e.Name == Schema.Task.RetryScheduled)); + EventTelemetry[] exportedEvents = channel.Items.OfType() + .OrderBy(e => e.Properties[Schema.Task.RetryAttempt]).ToArray(); + Assert.AreEqual(2, exportedEvents.Length); + for (int i = 0; i < events.Length; i++) + { + Assert.AreEqual(contexts[i + 1].TaskId, events[i].Tags.Single(t => t.Key == Schema.Task.TaskId).Value); + Assert.AreEqual(new DateTimeOffset(history[contexts[i + 1].TaskId].Timestamp), events[i].Timestamp); + Assert.AreEqual(includeFailureDetails ? typeof(InvalidOperationException).FullName : typeof(DurableTask.Core.Exceptions.TaskFailedException).FullName, + events[i].Tags.Single(t => t.Key == Schema.Task.FailureType).Value); + Assert.IsFalse(events[i].Tags.Any(t => t.Value?.ToString()?.Contains("secret") == true)); + Assert.AreEqual(events[i].Timestamp, exportedEvents[i].Timestamp); + Assert.AreEqual(orchestrationSpan.TraceId.ToString(), exportedEvents[i].Context.Operation.Id); + Assert.AreEqual(orchestrationSpan.SpanId.ToString(), exportedEvents[i].Context.Operation.ParentId); + Assert.AreEqual(6, exportedEvents[i].Properties.Count); + } + + Activity[] serverSpans = collector.Activities + .Where(a => a.Kind == ActivityKind.Server && Equals(a.GetTagItem(Schema.Task.Type), TraceActivityConstants.Activity)) + .OrderBy(a => (int)a.GetTagItem(Schema.Task.RetryAttempt)!) + .ToArray(); + Assert.AreEqual(3, serverSpans.Length); + for (int i = 0; i < serverSpans.Length; i++) + { + Assert.AreEqual(i + 1, serverSpans[i].GetTagItem(Schema.Task.RetryAttempt)); + Assert.AreEqual(3, serverSpans[i].GetTagItem(Schema.Task.RetryMaxAttempts)); + Assert.AreEqual(group, serverSpans[i].GetTagItem(Schema.Task.RetryGroupId)); + } + } + + [TestMethod] + public async Task DurableRetryEventsAreExportedOnTerminationWithoutCompletingThePendingActivity() + { + using var collector = new DurableActivityCollector(); + using var channel = new CapturingTelemetryChannel(); + using var configuration = DurableTelemetryModuleTests.CreateConfiguration(channel); + await using var module = new DurableTelemetryModule(); + module.Initialize(configuration); + var completionGate = new TaskCompletionSource(); + this.retryMetadataActivity.CompletionGate = completionGate.Task; + OrchestrationInstance instance = await this.client.CreateOrchestrationInstanceAsync( + typeof(RetryMetadataOrchestration), "secret input"); + try + { + Task started = this.retryMetadataActivity.FinalAttemptStarted.Task; + Assert.AreSame(started, await Task.WhenAny(started, Task.Delay(TimeSpan.FromSeconds(15)))); + Assert.AreEqual(0, channel.Items.OfType().Count()); + await this.client.TerminateInstanceAsync(instance, "stop"); + OrchestrationState state = await this.client.WaitForOrchestrationAsync(instance, TimeSpan.FromSeconds(15)); + Assert.AreEqual(OrchestrationStatus.Terminated, state.OrchestrationStatus); + Assert.AreEqual(2, channel.Items.OfType().Count()); + Assert.AreEqual(1, collector.Activities.Count(a => a.Kind == ActivityKind.Server + && Equals(a.GetTagItem(Schema.Task.Type), TraceActivityConstants.Orchestration))); + } + finally + { + completionGate.TrySetResult(true); + } + } + [TestMethod] public async Task TaskActivityDeliveryAttemptFlowsToTaskContextAndMiddleware() { @@ -580,6 +718,8 @@ public DurableActivityCollector() .LastOrDefault(); } + public Activity[] Activities => this.activities.ToArray(); + public void Dispose() { this.listener.Dispose(); @@ -624,6 +764,46 @@ protected override string Execute(TaskContext context, string input) } } + private sealed class RetryMetadataOrchestration : TaskOrchestration + { + public override Task RunTask(OrchestrationContext context, string input) + { + return context.ScheduleWithRetry(typeof(RetryMetadataActivity), + new RetryOptions(TimeSpan.FromMilliseconds(1), 3), input); + } + } + + private sealed class RetryMetadataActivity : AsyncTaskActivity + { + internal ConcurrentBag Contexts { get; } = new ConcurrentBag(); + + internal Task? CompletionGate { get; set; } + + internal TaskCompletionSource FinalAttemptStarted { get; } = new TaskCompletionSource(); + + protected override async Task ExecuteAsync(TaskContext context, string input) + { + this.Contexts.Add(context); + if (context.RetryMetadata!.AttemptNumber < 3) + { + throw new InvalidOperationException("secret error message"); + } + + this.FinalAttemptStarted.TrySetResult(true); + if (this.CompletionGate != null) + { + await this.CompletionGate; + } + + if (input == "fail") + { + throw new InvalidOperationException("secret error message"); + } + + return input; + } + } + private sealed class DeliveryAttemptOrchestrationService : LocalOrchestrationService, IOrchestrationService { public long? ActivityDeliveryAttempt { get; set; } diff --git a/test/DurableTask.Core.Tests/DurableTask.Core.Tests.csproj b/test/DurableTask.Core.Tests/DurableTask.Core.Tests.csproj index 6068df4e2..f4758d008 100644 --- a/test/DurableTask.Core.Tests/DurableTask.Core.Tests.csproj +++ b/test/DurableTask.Core.Tests/DurableTask.Core.Tests.csproj @@ -17,6 +17,7 @@ + diff --git a/test/DurableTask.Core.Tests/DurableTelemetryModuleTests.cs b/test/DurableTask.Core.Tests/DurableTelemetryModuleTests.cs new file mode 100644 index 000000000..e930507b1 --- /dev/null +++ b/test/DurableTask.Core.Tests/DurableTelemetryModuleTests.cs @@ -0,0 +1,222 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. + +#nullable enable +namespace DurableTask.Core.Tests +{ + using System; + using System.Collections.Concurrent; + using System.Diagnostics; + using System.Linq; + using System.Threading.Tasks; + using DurableTask.ApplicationInsights; + using DurableTask.Core.Exceptions; + using DurableTask.Core.History; + using DurableTask.Core.Tracing; + using Microsoft.ApplicationInsights.Channel; + using Microsoft.ApplicationInsights.DataContracts; + using Microsoft.ApplicationInsights.Extensibility; + using Microsoft.VisualStudio.TestTools.UnitTesting; + + [TestClass] + [DoNotParallelize] + public class DurableTelemetryModuleTests + { + [DataTestMethod] + [DataRow(true)] + [DataRow(false)] + public async Task ProjectedRetryEventsReachApplicationInsightsWithOriginalTimeAndCorrelation(bool structuredFailure) + { + using var channel = new CapturingTelemetryChannel(); + using var configuration = CreateConfiguration(channel); + await using var module = new DurableTelemetryModule(); + module.Initialize(configuration); + using var source = new ActivitySource("DurableTask.Core"); + Activity span = source.StartActivity("orchestration:work", ActivityKind.Server)!; + Assert.IsNotNull(span); + span.SetTag(Schema.Task.Type, "orchestration"); + DateTime scheduledAt = DateTime.UtcNow.AddMinutes(-5); + var history = new HistoryEvent[] + { + new TaskScheduledEvent(10, "work") { RetryMetadata = new RetryMetadata(1, 3, "group") }, + new TaskFailedEvent(-1, 10, "secret failure message", "opaque secret stack", + structuredFailure ? new FailureDetails("Example.Error", "secret", "secret stack", null, false) : null), + new TaskScheduledEvent(20, "work") + { + Timestamp = scheduledAt, + Input = "secret input", + RetryMetadata = new RetryMetadata(2, 3, "group"), + }, + }; + + TraceHelper.AddTaskRetryEvents(span, history); + TraceHelper.AddTaskRetryEvents(span, history); + Assert.AreEqual(0, channel.Items.Count, "Telemetry must not be sent before the orchestration span stops."); + span.Stop(); + span.Stop(); + + EventTelemetry retryEvent = channel.Items.OfType().Single(); + Assert.AreEqual(2, channel.Items.Count); + Assert.AreEqual(1, channel.Items.OfType().Count(), "Do not duplicate orchestration spans."); + Assert.AreEqual(Schema.Task.RetryScheduled, retryEvent.Name); + Assert.AreEqual(new DateTimeOffset(scheduledAt), retryEvent.Timestamp); + Assert.AreEqual(span.TraceId.ToString(), retryEvent.Context.Operation.Id); + Assert.AreEqual(span.SpanId.ToString(), retryEvent.Context.Operation.ParentId); + Assert.AreEqual(6, retryEvent.Properties.Count); + Assert.AreEqual("work", retryEvent.Properties[Schema.Task.Name]); + Assert.AreEqual("20", retryEvent.Properties[Schema.Task.TaskId]); + Assert.AreEqual("2", retryEvent.Properties[Schema.Task.RetryAttempt]); + Assert.AreEqual("3", retryEvent.Properties[Schema.Task.RetryMaxAttempts]); + Assert.AreEqual("group", retryEvent.Properties[Schema.Task.RetryGroupId]); + Assert.AreEqual(structuredFailure ? "Example.Error" : typeof(TaskFailedException).FullName, + retryEvent.Properties[Schema.Task.FailureType]); + Assert.IsFalse(retryEvent.Properties.Values.Any(value => value.Contains("secret"))); + } + + [TestMethod] + public async Task RetryEventMappingOnlyExportsAllowlistedFieldsFromOrchestrationEvents() + { + using var channel = new CapturingTelemetryChannel(); + using var configuration = CreateConfiguration(channel); + await using var module = new DurableTelemetryModule(); + module.Initialize(configuration); + using var source = new ActivitySource("DurableTask.Core"); + using (Activity span = source.StartActivity("orchestration:work", ActivityKind.Server)!) + { + span.SetTag(Schema.Task.Type, "orchestration"); + span.AddBaggage("input", "secret baggage"); + span.AddEvent(new ActivityEvent("exception", tags: new ActivityTagsCollection + { + { "exception.message", "secret message" }, + })); + span.AddEvent(new ActivityEvent(Schema.Task.RetryScheduled, tags: new ActivityTagsCollection + { + { Schema.Task.Name, "work" }, + { Schema.Task.TaskId, 20 }, + { Schema.Task.RetryAttempt, 2 }, + { Schema.Task.RetryGroupId, "group" }, + { Schema.Task.RetryMaxAttempts, 3 }, + { Schema.Task.FailureType, "Example.Error" }, + { "exception.message", "secret message" }, + { "exception.stacktrace", "secret stack" }, + { "input", "secret input" }, + })); + } + + using (Activity span = source.StartActivity("activity:work", ActivityKind.Server)!) + { + span.SetTag(Schema.Task.Type, "activity"); + span.AddEvent(new ActivityEvent(Schema.Task.RetryScheduled)); + } + + EventTelemetry retryEvent = channel.Items.OfType().Single(); + Assert.AreEqual(6, retryEvent.Properties.Count); + Assert.IsFalse(retryEvent.Properties.Values.Any(value => value.Contains("secret"))); + Assert.AreEqual(2, channel.Items.OfType().Count(), "Existing request export is unchanged."); + } + + [TestMethod] + public async Task StartingFreshTraceExportsTheCompletedGenerationsRetryEvents() + { + using var channel = new CapturingTelemetryChannel(); + using var configuration = CreateConfiguration(channel); + await using var module = new DurableTelemetryModule(); + module.Initialize(configuration); + using var source = new ActivitySource("DurableTask.Core"); + Activity span = source.StartActivity("orchestration:work", ActivityKind.Server)!; + span.SetTag(Schema.Task.Type, "orchestration"); + var state = new OrchestrationRuntimeState(new HistoryEvent[] + { + new ExecutionStartedEvent(-1, null) + { + Name = "work", + OrchestrationInstance = new OrchestrationInstance { InstanceId = "instance", ExecutionId = "generation1" }, + }, + new TaskScheduledEvent(10, "work") { RetryMetadata = new RetryMetadata(1, 3, "group") }, + new TaskFailedEvent(-1, 10, "secret", null, new FailureDetails("Example.Error", "secret", null, null, false)), + new TaskScheduledEvent(20, "work") { RetryMetadata = new RetryMetadata(2, 3, "group") }, + new ContinueAsNewEvent(-1, null), + }); + var nextGeneration = new ExecutionStartedEvent(-1, null) + { + GenerateNewTrace = true, + Name = "work", + OrchestrationInstance = new OrchestrationInstance { InstanceId = "instance", ExecutionId = "generation2" }, + }; + + Activity next = TaskOrchestrationDispatcher.RestartTraceActivityForContinueAsNewIfNeeded( + span, state, nextGeneration)!; + try + { + EventTelemetry retryEvent = channel.Items.OfType().Single(); + Assert.AreEqual(span.TraceId.ToString(), retryEvent.Context.Operation.Id); + Assert.AreNotEqual(span.TraceId, next.TraceId); + Assert.AreEqual(0, next.Events.Count()); + Assert.AreEqual(1, channel.Items.OfType().Count(), "Only the completed generation is stopped."); + } + finally + { + next.Stop(); + DistributedTraceActivity.Current = null!; + } + } + + [TestMethod] + public async Task ProjectionKeepsConcurrentGroupsSeparateAndDoesNotInventMissingRetries() + { + using var channel = new CapturingTelemetryChannel(); + using var configuration = CreateConfiguration(channel); + await using var module = new DurableTelemetryModule(); + module.Initialize(configuration); + using var source = new ActivitySource("DurableTask.Core"); + Activity span = source.StartActivity("orchestration:work", ActivityKind.Server)!; + span.SetTag(Schema.Task.Type, "orchestration"); + var history = new HistoryEvent[] + { + new TaskScheduledEvent(1, "left") { RetryMetadata = new RetryMetadata(1, 3, "left") }, + new TaskScheduledEvent(2, "right") { RetryMetadata = new RetryMetadata(1, 3, "right") }, + new TaskFailedEvent(-1, 2, "secret", null, new FailureDetails("Right.Error", "secret", null, null, false)), + new TaskFailedEvent(-1, 1, "secret", null, new FailureDetails("Left.Error", "secret", null, null, false)), + new TaskScheduledEvent(3, "right") { RetryMetadata = new RetryMetadata(2, 3, "right") }, + new TaskScheduledEvent(4, "left") { RetryMetadata = new RetryMetadata(2, 3, "left") }, + new TaskScheduledEvent(5, "old"), + new TaskFailedEvent(-1, 5, "secret", null), + new TaskScheduledEvent(6, "old"), + new TaskScheduledEvent(7, "initial") { RetryMetadata = new RetryMetadata(1, 1, "single") }, + new TaskFailedEvent(-1, 7, "secret", null), + }; + TraceHelper.AddTaskRetryEvents(span, history); + TraceHelper.AddTaskRetryEvents(span, history); + span.Stop(); + + EventTelemetry[] events = channel.Items.OfType().ToArray(); + Assert.AreEqual(2, events.Length); + Assert.AreEqual("Right.Error", events[0].Properties[Schema.Task.FailureType]); + Assert.AreEqual("right", events[0].Properties[Schema.Task.RetryGroupId]); + Assert.AreEqual("Left.Error", events[1].Properties[Schema.Task.FailureType]); + Assert.AreEqual("left", events[1].Properties[Schema.Task.RetryGroupId]); + } + + internal static TelemetryConfiguration CreateConfiguration(CapturingTelemetryChannel channel) + { + var configuration = TelemetryConfiguration.CreateDefault(); + configuration.ConnectionString = "InstrumentationKey=00000000-0000-0000-0000-000000000001"; + configuration.TelemetryChannel = channel; + return configuration; + } + } + + internal sealed class CapturingTelemetryChannel : ITelemetryChannel + { + internal ConcurrentQueue Items { get; } = new ConcurrentQueue(); + + public bool? DeveloperMode { get; set; } + + public string EndpointAddress { get; set; } = string.Empty; + + public void Send(ITelemetry item) => this.Items.Enqueue(item); + + public void Flush() { } + + public void Dispose() { } + } +} diff --git a/test/DurableTask.Core.Tests/RetryMetadataTests.cs b/test/DurableTask.Core.Tests/RetryMetadataTests.cs new file mode 100644 index 000000000..a39a7e2a5 --- /dev/null +++ b/test/DurableTask.Core.Tests/RetryMetadataTests.cs @@ -0,0 +1,383 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. + +#nullable enable +namespace DurableTask.Core.Tests +{ + using System; + using System.Diagnostics; + using System.IO; + using System.Linq; + using System.Runtime.Serialization; + using System.Threading.Tasks; + using DurableTask.Core.Command; + using DurableTask.Core.Exceptions; + using DurableTask.Core.History; + using DurableTask.Core.Serializing; + using DurableTask.Core.Tracing; + using Microsoft.VisualStudio.TestTools.UnitTesting; + using Newtonsoft.Json; + + [TestClass] + public class RetryMetadataTests + { + [DataTestMethod] + [DataRow(0, 3)] + [DataRow(-1, 3)] + [DataRow(1, 0)] + [DataRow(1, -1)] + [DataRow(4, 3)] + public void ConstructorRejectsInvalidAttempts(int attempt, int max) + { + Assert.ThrowsException(() => new RetryMetadata(attempt, max, "group")); + } + + [DataTestMethod] + [DataRow(null)] + [DataRow("")] + [DataRow(" ")] + public void ConstructorRejectsMissingGroup(string group) + { + Assert.ThrowsException(() => new RetryMetadata(1, 3, group)); + } + + [DataTestMethod] + [DataRow(1, 1)] + [DataRow(int.MaxValue, int.MaxValue)] + public void MetadataRoundTripsAcrossSerializers(int attempt, int max) + { + var metadata = new RetryMetadata(attempt, max, "execution:42"); + string json = JsonConvert.SerializeObject(metadata); + AssertMetadata(metadata, JsonConvert.DeserializeObject(json)); + AssertMetadata(metadata, System.Text.Json.JsonSerializer.Deserialize(json)); + AssertMetadata(metadata, JsonDataConverter.Default.Deserialize(JsonDataConverter.Default.Serialize(metadata))); + + var serializer = new DataContractSerializer(typeof(RetryMetadata)); + using var stream = new MemoryStream(); + serializer.WriteObject(stream, metadata); + stream.Position = 0; + AssertMetadata(metadata, (RetryMetadata)serializer.ReadObject(stream)!); + } + + [DataTestMethod] + [DataRow("{}", typeof(ArgumentOutOfRangeException))] + [DataRow("{\"AttemptNumber\":0,\"MaxAttempts\":3,\"RetryGroupId\":\"g\"}", typeof(ArgumentOutOfRangeException))] + [DataRow("{\"AttemptNumber\":4,\"MaxAttempts\":3,\"RetryGroupId\":\"g\"}", typeof(ArgumentOutOfRangeException))] + [DataRow("{\"AttemptNumber\":1,\"MaxAttempts\":3,\"RetryGroupId\":\"\"}", typeof(ArgumentException))] + [DataRow("{\"AttemptNumber\":2147483648,\"MaxAttempts\":3,\"RetryGroupId\":\"g\"}", typeof(JsonReaderException))] + public void JsonRejectsInvalidMetadata(string json, Type exceptionType) + { + AssertContractError(() => JsonConvert.DeserializeObject(json), exceptionType); + AssertContractError(() => JsonConvert.DeserializeObject("{\"RetryMetadata\":" + json + "}"), exceptionType); + } + + static void AssertContractError(Action deserialize, Type exceptionType) + { + if (exceptionType == typeof(ArgumentOutOfRangeException)) + { + Assert.ThrowsException(deserialize); + } + else if (exceptionType == typeof(ArgumentException)) + { + Assert.ThrowsException(deserialize); + } + else + { + Assert.ThrowsException(deserialize); + } + } + + [TestMethod] + public void HistoryAndActionsRoundTripOptionalMetadata() + { + var metadata = new RetryMetadata(2, 3, "execution:42"); + var action = new ScheduleTaskOrchestratorAction + { + Id = 42, Name = "work", RetryMetadata = metadata, RetryFailureType = "Captured.Error", + }; + Assert.IsFalse(JsonConvert.SerializeObject(action).Contains("Captured.Error")); + AssertMetadata(metadata, JsonConvert.DeserializeObject( + JsonConvert.SerializeObject(action))!.RetryMetadata); + AssertMetadata(metadata, System.Text.Json.JsonSerializer.Deserialize( + System.Text.Json.JsonSerializer.Serialize(action))!.RetryMetadata); + var scheduled = new TaskScheduledEvent(42, "work") { RetryMetadata = metadata, RetryFailureType = "Captured.Error" }; + Assert.IsFalse(JsonDataConverter.Default.Serialize(scheduled).Contains("Captured.Error")); + AssertMetadata(metadata, JsonDataConverter.Default.Deserialize( + JsonDataConverter.Default.Serialize(scheduled)).RetryMetadata); + + var serializer = new DataContractSerializer(typeof(TaskMessage)); + using var stream = new MemoryStream(); + serializer.WriteObject(stream, new TaskMessage { Event = scheduled }); + stream.Position = 0; + AssertMetadata(metadata, ((TaskScheduledEvent)((TaskMessage)serializer.ReadObject(stream)!).Event).RetryMetadata); + + scheduled.RetryMetadata = null; + Assert.IsFalse(JsonConvert.SerializeObject(scheduled).Contains("RetryMetadata")); + Assert.IsNull(JsonConvert.DeserializeObject("{\"EventId\":42,\"Name\":\"work\"}")!.RetryMetadata); + Assert.IsNull(new TaskContext(new OrchestrationInstance()).RetryMetadata); + } + + [DataTestMethod] + [DataRow(false)] + [DataRow(true)] + public async Task RetrySchedulingIsReplayStableAndIsolated(bool withOptions) + { + string firstGroup = await RunAttempts(withOptions, replay: false); + string replayGroup = await RunAttempts(withOptions, replay: true); + Assert.AreEqual(firstGroup, replayGroup); + } + + static async Task RunAttempts(bool withOptions, bool replay) + { + var context = new ImmediateTimerContext { IsReplaying = replay }; + using var span = new Activity("orchestration").Start(); + Activity? previous = DistributedTraceActivity.Current; + DistributedTraceActivity.Current = span; + try + { + var options = new RetryOptions(TimeSpan.FromMilliseconds(1), 3); + Task result = withOptions + ? context.ScheduleTask("work", "", ScheduleTaskOptions.CreateBuilder().WithRetryOptions(options).Build()) + : context.ScheduleWithRetry("work", "", options); + ScheduleTaskOrchestratorAction first = context.Actions.Single(); + Assert.AreEqual(1, first.RetryMetadata!.AttemptNumber); + Assert.AreEqual(3, first.RetryMetadata.MaxAttempts); + Assert.AreEqual(0, span.Events.Count()); + string group = first.RetryMetadata.RetryGroupId; + + Task ordinary = context.ScheduleTask("ordinary", ""); + Assert.IsNull(context.Actions.Single(a => a.Name == "ordinary").RetryMetadata); + context.Complete(context.Actions.Single(a => a.Name == "ordinary")); + await ordinary; + context.Fail(first); + ScheduleTaskOrchestratorAction second = context.Actions.Single(); + Assert.AreEqual(2, second.RetryMetadata!.AttemptNumber); + Assert.AreEqual(group, second.RetryMetadata.RetryGroupId); + context.Fail(second); + ScheduleTaskOrchestratorAction third = context.Actions.Single(); + Assert.AreEqual(3, third.RetryMetadata!.AttemptNumber); + Assert.AreEqual(group, third.RetryMetadata.RetryGroupId); + context.Complete(third); + Assert.AreEqual("done", await result); + Assert.AreEqual(2, context.TimerCalls); + + Assert.AreEqual(0, span.Events.Count(), "Retry events are projected from persisted history only at completion."); + var history = new HistoryEvent[] + { + ToScheduled(first), + new TaskFailedEvent(-1, first.Id, "secret", null, new FailureDetails("Example.Failure", "secret", null, null, false)), + ToScheduled(second), + new TaskFailedEvent(-1, second.Id, "secret", null, new FailureDetails("Example.Failure", "secret", null, null, false)), + ToScheduled(third), + }; + TraceHelper.AddTaskRetryEvents(span, history); + TraceHelper.AddTaskRetryEvents(span, history); + ActivityEvent[] events = span.Events.ToArray(); + Assert.AreEqual(2, events.Length); + for (int i = 0; i < events.Length; i++) + { + Assert.AreEqual(Schema.Task.RetryScheduled, events[i].Name); + var tags = events[i].Tags.ToDictionary(t => t.Key, t => t.Value); + Assert.AreEqual(6, tags.Count); + Assert.AreEqual("work", tags[Schema.Task.Name]); + Assert.AreEqual(i == 0 ? second.Id : third.Id, tags[Schema.Task.TaskId]); + Assert.AreEqual(i + 2, tags[Schema.Task.RetryAttempt]); + Assert.AreEqual(3, tags[Schema.Task.RetryMaxAttempts]); + Assert.AreEqual(group, tags[Schema.Task.RetryGroupId]); + Assert.AreEqual("Example.Failure", tags[Schema.Task.FailureType]); + Assert.IsFalse(tags.Values.Any(v => v?.ToString()?.Contains("secret") == true)); + } + + return group; + } + + finally + { + DistributedTraceActivity.Current = previous!; + } + } + + static TaskScheduledEvent ToScheduled(ScheduleTaskOrchestratorAction action) => + new TaskScheduledEvent(action.Id, action.Name!) + { + RetryMetadata = action.RetryMetadata, + RetryFailureType = action.RetryFailureType, + }; + + [TestMethod] + public async Task ConcurrentRetryGroupsDoNotLeakAndOldHistoryIsAccepted() + { + var context = new ImmediateTimerContext(); + var options = new RetryOptions(TimeSpan.FromMilliseconds(1), 2); + Task leftResult = context.ScheduleWithRetry("left", "", options); + Task rightResult = context.ScheduleWithRetry("right", "", options); + ScheduleTaskOrchestratorAction left = context.Actions.Single(a => a.Name == "left"); + ScheduleTaskOrchestratorAction right = context.Actions.Single(a => a.Name == "right"); + Assert.AreNotEqual(left.RetryMetadata!.RetryGroupId, right.RetryMetadata!.RetryGroupId); + context.Fail(right); + ScheduleTaskOrchestratorAction rightRetry = context.Actions.Single(a => a.Name == "right"); + Assert.AreEqual(right.RetryMetadata.RetryGroupId, rightRetry.RetryMetadata!.RetryGroupId); + context.Fail(left); + ScheduleTaskOrchestratorAction leftRetry = context.Actions.Single(a => a.Name == "left"); + Assert.AreEqual(left.RetryMetadata.RetryGroupId, leftRetry.RetryMetadata!.RetryGroupId); + context.Complete(rightRetry); + context.Complete(leftRetry); + await Task.WhenAll(leftResult, rightResult); + } + + [DataTestMethod] + [DataRow("exhausted")] + [DataRow("filter")] + [DataRow("timeout")] + [DataRow("nonretriable")] + public async Task EarlyStopDoesNotScheduleAnotherAttempt(string stop) + { + var context = new ImmediateTimerContext(); + using var span = new Activity("orchestration").Start(); + Activity? previous = DistributedTraceActivity.Current; + DistributedTraceActivity.Current = span; + try + { + var options = new RetryOptions(TimeSpan.FromMilliseconds(1), stop == "exhausted" ? 1 : 3) + { + Handle = _ => stop != "filter", + RetryTimeout = stop == "timeout" ? TimeSpan.Zero : TimeSpan.MaxValue, + }; + Task result = context.ScheduleWithRetry("work", "", options); + context.Fail(context.Actions.Single(), nonRetriable: stop == "nonretriable"); + await Assert.ThrowsExceptionAsync(async () => await result); + Assert.AreEqual(0, context.Actions.Count()); + Assert.AreEqual(0, span.Events.Count()); + Assert.AreEqual(stop == "exhausted" ? 1 : 0, context.TimerCalls); + } + finally + { + DistributedTraceActivity.Current = previous!; + } + } + + [TestMethod] + public async Task AsyncRetryDelegateRetainsItsScopeWithoutLeakingToCaller() + { + var context = new ImmediateTimerContext(); + var gate = new TaskCompletionSource(); + var interceptor = new RetryInterceptor(context, new RetryOptions(TimeSpan.FromMilliseconds(1), 1), async () => + { + await gate.Task; + return await context.ScheduleTask("retry", ""); + }); + Task result = interceptor.Invoke(); + Task ordinary = context.ScheduleTask("ordinary", ""); + ScheduleTaskOrchestratorAction action = context.Actions.Single(); + Assert.IsNull(action.RetryMetadata); + context.Complete(action); + await ordinary; + gate.SetResult(true); + action = context.Actions.Single(); + Assert.AreEqual(1, action.RetryMetadata!.AttemptNumber); + context.Complete(action); + await result; + } + + [TestMethod] + public async Task RetryableProxySchedulesMetadata() + { + var context = new ImmediateTimerContext(); + var proxy = context.CreateRetryableClient(new RetryOptions(TimeSpan.FromMilliseconds(1), 2)); + Task result = proxy.Run("input"); + ScheduleTaskOrchestratorAction action = context.Actions.Single(); + Assert.AreEqual(1, action.RetryMetadata!.AttemptNumber); + context.Fail(action); + action = context.Actions.Single(); + Assert.AreEqual(2, action.RetryMetadata!.AttemptNumber); + context.Complete(action); + await result; + } + + [TestMethod] + public async Task SubOrchestrationRetriesDoNotAnnotateUnrelatedActivities() + { + var context = new ImmediateTimerContext(); + Task child = context.CreateSubOrchestrationInstanceWithRetry( + "child", "", new RetryOptions(TimeSpan.FromMilliseconds(1), 2), null); + var action = context.OrchestratorActions.OfType().Single(); + context.HandleSubOrchestrationCreatedEvent(new SubOrchestrationInstanceCreatedEvent(action.Id) + { + Name = action.Name, + InstanceId = action.InstanceId, + }); + context.HandleSubOrchestrationInstanceCompletedEvent(new SubOrchestrationInstanceCompletedEvent(-1, action.Id, "\"child result\"")); + await child; + Task ordinary = context.ScheduleTask("ordinary", ""); + ScheduleTaskOrchestratorAction scheduled = context.Actions.Single(); + Assert.IsNull(scheduled.RetryMetadata); + context.Complete(scheduled); + await ordinary; + } + + [TestMethod] + public async Task ExhaustedRetryPreservesTheCompatibilityTimer() + { + var context = new ImmediateTimerContext(); + Task result = context.ScheduleWithRetry("work", "", + new RetryOptions(TimeSpan.FromMilliseconds(1), 3)); + for (int attempt = 1; attempt <= 3; attempt++) + { + ScheduleTaskOrchestratorAction scheduled = context.Actions.Single(); + Assert.AreEqual(attempt, scheduled.RetryMetadata!.AttemptNumber); + context.Fail(scheduled); + } + + await Assert.ThrowsExceptionAsync(async () => await result); + Assert.AreEqual(0, context.Actions.Count()); + Assert.AreEqual(3, context.TimerCalls); + } + + public interface IRetryActivity + { + Task Run(string input); + } + + internal static void AssertMetadata(RetryMetadata expected, RetryMetadata? actual) + { + Assert.IsNotNull(actual); + Assert.AreEqual(expected.AttemptNumber, actual!.AttemptNumber); + Assert.AreEqual(expected.MaxAttempts, actual.MaxAttempts); + Assert.AreEqual(expected.RetryGroupId, actual.RetryGroupId); + } + + sealed class ImmediateTimerContext : TaskOrchestrationContext + { + internal ImmediateTimerContext() + : base(new OrchestrationInstance { InstanceId = "instance", ExecutionId = "execution" }, TaskScheduler.Default, + errorPropagationMode: ErrorPropagationMode.UseFailureDetails) + { + this.CurrentUtcDateTime = new DateTime(2026, 1, 1, 0, 0, 0, DateTimeKind.Utc); + } + + internal System.Collections.Generic.IEnumerable Actions => + this.OrchestratorActions.OfType(); + + internal int TimerCalls { get; private set; } + + public override Task CreateTimer(DateTime fireAt, T state) + { + this.TimerCalls++; + this.CurrentUtcDateTime = fireAt; + return Task.FromResult(state); + } + + internal void Fail(ScheduleTaskOrchestratorAction action, bool nonRetriable = false) + { + this.HandleTaskScheduledEvent(new TaskScheduledEvent(action.Id, action.Name!)); + this.HandleTaskFailedEvent(new TaskFailedEvent(-1, action.Id, "secret message", "secret stack", + new FailureDetails("Example.Failure", "secret message", "secret stack", null, nonRetriable))); + } + + internal void Complete(ScheduleTaskOrchestratorAction action) + { + this.HandleTaskScheduledEvent(new TaskScheduledEvent(action.Id, action.Name!)); + this.HandleTaskCompletedEvent(new TaskCompletedEvent(-1, action.Id, "\"done\"")); + } + } + } +}