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\"")); + } + } + } +}