镜像站点 · 本页由第三方 GitHub 只读镜像提供,非 GitHub 官方站点,不接受任何登录或凭据输入。前往 github.com
Skip to content
Draft
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
18 changes: 18 additions & 0 deletions docs/features/retries.md
Original file line number Diff line number Diff line change
Expand Up @@ -169,6 +169,24 @@ var result = await context.CreateSubOrchestrationInstanceWithRetry<Result>(

## 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
Expand Down
12 changes: 12 additions & 0 deletions docs/telemetry/application-insights.md
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
40 changes: 40 additions & 0 deletions docs/telemetry/traces/semantic-conventions.md
Original file line number Diff line number Diff line change
Expand Up @@ -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.
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -20,7 +20,8 @@
<!-- The assembly version is only the major/minor pair, making it easier to do in-place upgrades -->
<AssemblyVersion>$(MajorVersion).$(MinorVersion).0.0</AssemblyVersion>
<!-- This version is used as the nuget package version -->
<Version>$(VersionPrefix)</Version>
<Version Condition="'$(VersionSuffix)' == ''">$(VersionPrefix)</Version>
<Version Condition="'$(VersionSuffix)' != ''">$(VersionPrefix)-$(VersionSuffix)</Version>
</PropertyGroup>

<ItemGroup>
Expand Down
41 changes: 41 additions & 0 deletions src/DurableTask.ApplicationInsights/DurableTelemetryModule.cs
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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<string, object> 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)
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -48,5 +48,14 @@ public class ScheduleTaskOrchestratorAction : OrchestratorAction
/// Gets or sets a dictionary of tags of string, string
/// </summary>
public IDictionary<string, string>? Tags { get; set; }

/// <summary>
/// Gets or sets durable retry metadata, or null for an activity without a retry policy.
/// </summary>
[Newtonsoft.Json.JsonProperty(NullValueHandling = Newtonsoft.Json.NullValueHandling.Ignore)]
public RetryMetadata? RetryMetadata { get; set; }

[Newtonsoft.Json.JsonIgnore]
internal string? RetryFailureType { get; set; }
}
}
10 changes: 10 additions & 0 deletions src/DurableTask.Core/History/TaskScheduledEvent.cs
Original file line number Diff line number Diff line change
Expand Up @@ -91,5 +91,15 @@ public TaskScheduledEvent(int eventId)
/// </summary>
[DataMember]
public IDictionary<string, string>? Tags { get; set; }

/// <summary>
/// Gets or sets durable retry metadata, or null for uninstrumented or non-retry activity calls.
/// </summary>
[DataMember(EmitDefaultValue = false)]
[Newtonsoft.Json.JsonProperty(NullValueHandling = Newtonsoft.Json.NullValueHandling.Ignore)]
public RetryMetadata? RetryMetadata { get; set; }

[Newtonsoft.Json.JsonIgnore]
internal string? RetryFailureType { get; set; }
}
}
8 changes: 8 additions & 0 deletions src/DurableTask.Core/OrchestrationContext.cs
Original file line number Diff line number Diff line change
Expand Up @@ -31,6 +31,14 @@ public abstract class OrchestrationContext
/// </summary>
private static readonly ProxyGenerator ProxyGenerator = new ProxyGenerator();

readonly System.Threading.AsyncLocal<ActivityRetryContext> activityRetryContext = new System.Threading.AsyncLocal<ActivityRetryContext>();

internal ActivityRetryContext ActivityRetryContext
{
get => this.activityRetryContext.Value;
set => this.activityRetryContext.Value = value;
}

/// <summary>
/// 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.
Expand Down
28 changes: 27 additions & 1 deletion src/DurableTask.Core/RetryInterceptor.cs
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Comment on lines +62 to +63
ActivityRetryContext? previousContext = this.context.ActivityRetryContext;
Task<T> attempt;
try
{
this.context.ActivityRetryContext = retryContext;
attempt = this.retryCall();
}
finally
{
this.context.ActivityRetryContext = previousContext;
}

return await attempt;
}
catch (Exception e) when (!Utils.IsFatal(e))
{
Expand Down Expand Up @@ -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; }
}
}
70 changes: 70 additions & 0 deletions src/DurableTask.Core/RetryMetadata.cs
Original file line number Diff line number Diff line change
@@ -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;

/// <summary>
/// Identifies an activity attempt scheduled by a durable retry policy.
/// </summary>
[DataContract]
public sealed class RetryMetadata
{
/// <summary>
/// Initializes validated metadata for an activity attempt.
/// </summary>
/// <param name="attemptNumber">The one-based durable attempt number.</param>
/// <param name="maxAttempts">The maximum number of attempts, including the initial attempt.</param>
/// <param name="retryGroupId">The replay-stable identifier shared by all attempts of the logical activity.</param>
[JsonConstructor]
public RetryMetadata(int attemptNumber, int maxAttempts, string retryGroupId)
{
this.AttemptNumber = attemptNumber;
this.MaxAttempts = maxAttempts;
this.RetryGroupId = retryGroupId;
this.Validate();
}

/// <summary>
/// Gets the one-based durable attempt number, independent of message redelivery.
/// </summary>
[DataMember]
public int AttemptNumber { get; private set; }

/// <summary>
/// Gets the policy's maximum number of attempts, including the initial attempt.
/// </summary>
[DataMember]
public int MaxAttempts { get; private set; }

/// <summary>
/// Gets the identifier shared by attempts of the logical activity across replay.
/// </summary>
[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");
}
}
}
}
1 change: 1 addition & 0 deletions src/DurableTask.Core/TaskActivityDispatcher.cs
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down
6 changes: 6 additions & 0 deletions src/DurableTask.Core/TaskContext.cs
Original file line number Diff line number Diff line change
Expand Up @@ -69,6 +69,12 @@ public TaskContext(OrchestrationInstance orchestrationInstance, string name, str
/// </remarks>
public long? DeliveryAttempt { get; internal set; }

/// <summary>
/// Gets durable retry metadata, or null when the activity was scheduled without it.
/// Message redelivery and worker restarts do not change this value.
/// </summary>
public RetryMetadata? RetryMetadata { get; internal set; }

/// <summary>
/// Gets or sets a value indicating how to propagate unhandled exception metadata.
/// </summary>
Expand Down
23 changes: 22 additions & 1 deletion src/DurableTask.Core/TaskOrchestrationContext.cs
Original file line number Diff line number Diff line change
Expand Up @@ -94,7 +94,13 @@ public override async Task<TResult> ScheduleTask<TResult>(string name, string ve
{
if (options.RetryOptions != null)
{
Task<TResult> RetryCall() => ScheduleTask<TResult>(name, version, ScheduleTaskOptions.CreateBuilder().WithTags(options.Tags).Build(), parameters);
var attemptOptions = ScheduleTaskOptions.CreateBuilder();
if (options.Tags != null)
{
attemptOptions.WithTags(options.Tags);
}

Task<TResult> RetryCall() => ScheduleTask<TResult>(name, version, attemptOptions.Build(), parameters);
var retryInterceptor = new RetryInterceptor<TResult>(this, options.RetryOptions, RetryCall);
return await retryInterceptor.Invoke();
}
Expand Down Expand Up @@ -138,6 +144,19 @@ public async Task<object> 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<string>();
Expand Down Expand Up @@ -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);
}

Expand Down
Loading
Loading