Merge main into azure-storage-v12

README.md

@@ -25,7 +25,13 @@ The core programming model for the Durable Task Framework is contained in the [D
 
 ## Learning more
 
-The associated wiki contains more details about the framework and how it can be used: https://github.com/Azure/durabletask/wiki. You can also find great information in [this blog series](https://abhikmitra.github.io/blog/durable-task/). In some cases, the [Durable Functions documentation](https://docs.microsoft.com/en-us/azure/azure-functions/durable/) can actually be useful in learning things about the underlying framework, although not everything will apply. Lastly, you can watch a video with some of the original maintainers in [this Channel 9 video](https://channel9.msdn.com/Shows/On-NET/Building-workflows-with-the-Durable-Task-Framework).
+There are several places where you can learn more about this framework. Note that some are external and not owned by Microsoft:
+
+- [This repo's wiki](https://github.com/Azure/durabletask/wiki), which contains more details about the framework and how it can be used.
+- The following blog series contains useful information: https://abhikmitra.github.io/blog/durable-task/
+- Several useful samples are available here: https://github.com/kaushiksk/durabletask-samples
+- You can watch a video with some of the original maintainers in [Building Workflows with the Durable Task Framework](https://learn.microsoft.com/shows/on-net/building-workflows-with-the-durable-task-framework).
+- In some cases, the [Azure Durable Functions documentation](https://learn.microsoft.com/azure/azure-functions/durable/) can actually be useful in learning things about the underlying framework, although not everything will apply.
 
 ## Development Notes
 
@@ -35,6 +41,8 @@ Unit tests also require [Azure Storage Emulator](https://docs.microsoft.com/azur
 
 > Note: While it's possible to use in tests a real Azure Storage account it is not recommended to do so because many tests will fail with a 409 Conflict error. This is because tests delete and quickly recreate the same storage tables, and Azure Storage doesn't do well in these conditions. If you really want to change Azure Storage connection string you can do so via the **StorageConnectionString** app.config value in the test project, or by defining a **DurableTaskTestStorageConnectionString** environment variable. 
 
+<!--
 There is a gitter for this repo, but it's not currently being monitored. We're leaving the link for it up for now and will update this message if anything changes.
 
 [![Join the chat at https://gitter.im/azure/durabletask](https://badges.gitter.im/Join%20Chat.svg)](https://gitter.im/azure/durabletask?utm_source=badge&utm_medium=badge&utm_campaign=pr-badge&utm_content=badge) 
+-->

azure-pipelines-release.yml

@@ -1,11 +1,28 @@
-trigger: none
+trigger:
+- main
+
 pr: none
 
 pool:
   name: '1ES-Hosted-DurableTaskFramework'
   demands:
     - ImageOverride -equals MMS2022TLS
 
+variables:
+
+  - name: VersionSuffix
+    # The `Build.Reason` env var gets populated with `IndividualCI` on an automatic run of the CI,
+    # such as when a commit is made to `main`. If the CI is manually run, it will get populated with
+    # "Manual". For more details on these `Build.X` vars, see: https://learn.microsoft.com/en-us/azure/devops/pipelines/build/variables?view=azure-devops&tabs=yaml#build-variables-devops-services
+    ${{ if eq(variables['Build.Reason'], 'IndividualCI') }}:
+      # The `Build.BuildNumber` env var is an automatically-generated "build ID"
+      # for any given ADO run. It is usually the date format yyddmm.<counter> where
+      # `yyddmm` is a date formatter, and <counter> is a daily counter in case of multiple
+      # builds on the same date.
+      value: 'ci.$(Build.BuildNumber)'
+    ${{ else }}:
+      value: ''
+
 steps:
 # Start by restoring all the dependencies. This needs to be its own task
 # from what I can tell. We specifically only target DurableTask.AzureStorage

docs/telemetry/traces/getting-started.md

@@ -0,0 +1,26 @@
+# Getting Started - Distributed Tracing
+
+> ⚠ Important: durable task distributed tracing is currently [experimental](https://github.com/open-telemetry/opentelemetry-specification/blob/main/specification/document-status.md). The schema is subject to changes until it is marked as stable. These changes may occur in any package update.
+
+> ⚠ Important: this guide only applies DurableTask users. For Durable Functions, please see [here](https://github.com/Azure/azure-functions-durable-extension/blob/dev/samples/distributed-tracing/v2/DistributedTracingSample/README.md)
+
+Distributed tracing in DurableTask uses the `ActivitySource` approach, it is both OpenTelemetry and Application Insights compatible.
+
+## OpenTelemetry
+
+Add the `"DurableTask.Core"` source to the OTel trace builder.
+
+``` CSharp
+Sdk.CreateTracerProviderBuilder()
+    .AddSource("DurableTask.Core")
+    .Build()
+```
+
+See [sample](../../../samples/DistributedTraceSample/OpenTelemetry)
+
+## Application Insights
+
+1. Add reference to [Microsoft.Azure.DurableTask.ApplicationInsights](https://www.nuget.org/packages/Microsoft.Azure.DurableTask.ApplicationInsights)
+2. Add the `DurableTelemetryModule` to AppInsights: `services.TryAddEnumerable(ServiceDescriptor.Singleton<ITelemetryModule, DurableTelemetryModule>());`
+
+See [sample](../../../samples/DistributedTraceSample/ApplicationInsights)

samples/DurableTask.Samples/DurableTask.Samples.csproj

@@ -11,7 +11,6 @@
     <PackageReference Include="CommandLineParser" version="1.9.71" />
     <PackageReference Include="EnterpriseLibrary.SemanticLogging" Version="2.0.1406.1" />
     <PackageReference Include="ncrontab" version="1.0.0" />
-    <PackageReference Include="Newtonsoft.Json" version="11.0.2" />
   </ItemGroup>
 
   <ItemGroup>

src/DurableTask.AzureServiceFabric/DurableTask.AzureServiceFabric.csproj

@@ -13,17 +13,16 @@
     <Description>Orchestration message and runtime state is stored in Azure Service Fabric reliable collections.</Description>
     <Authors>Microsoft</Authors>
     <Platforms>AnyCPU;x64</Platforms>
+    <NoWarn>$(NoWarn);NU5104</NoWarn>
  </PropertyGroup>
 
   <ItemGroup Condition="'$(TargetFramework)' == 'net462'">
     <PackageReference Include="ImpromptuInterface" Version="6.2.2" />
-    <PackageReference Include="Newtonsoft.Json" Version="7.0.1" />
     <Reference Include="System.Web" />
   </ItemGroup>
 
   <ItemGroup Condition="'$(TargetFramework)' == 'net472'">
     <PackageReference Include="ImpromptuInterface" Version="6.2.2" />
-    <PackageReference Include="Newtonsoft.Json" Version="7.0.1" />
     <Reference Include="System.Web" />
   </ItemGroup>
 

src/DurableTask.AzureServiceFabric/Remote/RemoteOrchestrationServiceClient.cs

@@ -378,7 +378,7 @@ private async Task<HttpResponseMessage> ExecuteRequestWithRetriesAsync(string in
                         return response;
                     }
                 }
-                catch (Exception ex) when (ex is SocketException || ex is WebException)
+                catch (Exception ex) when (ex is SocketException || ex is WebException || ex is HttpRequestException)
                 {
                     exception = ex;
                 }

src/DurableTask.AzureStorage/AzureStorageOrchestrationService.cs

@@ -31,6 +31,7 @@ namespace DurableTask.AzureStorage
     using DurableTask.AzureStorage.Storage;
     using DurableTask.AzureStorage.Tracking;
     using DurableTask.Core;
+    using DurableTask.Core.Entities;
     using DurableTask.Core.Exceptions;
     using DurableTask.Core.History;
     using DurableTask.Core.Query;
@@ -44,7 +45,8 @@ public sealed class AzureStorageOrchestrationService :
         IOrchestrationServiceClient,
         IDisposable,
         IOrchestrationServiceQueryClient,
-        IOrchestrationServicePurgeClient
+        IOrchestrationServicePurgeClient,
+        IEntityOrchestrationService
     {
         static readonly HistoryEvent[] EmptyHistoryEventList = new HistoryEvent[0];
 
@@ -280,6 +282,51 @@ public BehaviorOnContinueAsNew EventBehaviourForContinueAsNew
         /// <inheritdoc />
         public int TaskOrchestrationDispatcherCount { get; } = 1;
 
+        #region IEntityOrchestrationService
+
+        EntityBackendProperties IEntityOrchestrationService.EntityBackendProperties
+           => new EntityBackendProperties()
+           {
+               EntityMessageReorderWindow = TimeSpan.FromMinutes(this.settings.EntityMessageReorderWindowInMinutes),
+               MaxEntityOperationBatchSize = this.settings.MaxEntityOperationBatchSize,
+               MaxConcurrentTaskEntityWorkItems = this.settings.MaxConcurrentTaskEntityWorkItems,
+               SupportsImplicitEntityDeletion = false, // not supported by this backend
+               MaximumSignalDelayTime = TimeSpan.FromDays(6),
+               UseSeparateQueueForEntityWorkItems = this.settings.UseSeparateQueueForEntityWorkItems,
+           };
+
+        EntityBackendQueries IEntityOrchestrationService.EntityBackendQueries
+           => new EntityTrackingStoreQueries(
+                this.messageManager,
+                this.trackingStore,
+                this.EnsureTaskHubAsync,
+                ((IEntityOrchestrationService)this).EntityBackendProperties,
+                this.SendTaskOrchestrationMessageAsync);
+
+        Task<TaskOrchestrationWorkItem> IEntityOrchestrationService.LockNextOrchestrationWorkItemAsync(
+          TimeSpan receiveTimeout,
+          CancellationToken cancellationToken)
+        {
+            if (!this.settings.UseSeparateQueueForEntityWorkItems)
+            {
+                throw new InvalidOperationException("Internal configuration is inconsistent. Backend is using single queue for orchestration/entity dispatch, but frontend is pulling from individual queues.");
+            }
+            return this.LockNextTaskOrchestrationWorkItemAsync(false, cancellationToken);
+        }
+
+        Task<TaskOrchestrationWorkItem> IEntityOrchestrationService.LockNextEntityWorkItemAsync(
+           TimeSpan receiveTimeout,
+           CancellationToken cancellationToken)
+        {
+            if (!this.settings.UseSeparateQueueForEntityWorkItems)
+            {
+                throw new InvalidOperationException("Internal configuration is inconsistent. Backend is using single queue for orchestration/entity dispatch, but frontend is pulling from individual queues.");
+            }
+            return this.LockNextTaskOrchestrationWorkItemAsync(entitiesOnly: true, cancellationToken);
+        }
+
+        #endregion
+
         #region Management Operations (Create/Delete/Start/Stop)
         /// <summary>
         /// Deletes and creates the neccesary Azure Storage resources for the orchestration service.
@@ -628,9 +675,18 @@ static TaskHubInfo GetTaskHubInfo(string taskHub, int partitionCount)
 
         #region Orchestration Work Item Methods
         /// <inheritdoc />
-        public async Task<TaskOrchestrationWorkItem> LockNextTaskOrchestrationWorkItemAsync(
+        public Task<TaskOrchestrationWorkItem> LockNextTaskOrchestrationWorkItemAsync(
             TimeSpan receiveTimeout,
             CancellationToken cancellationToken)
+        {
+            if (this.settings.UseSeparateQueueForEntityWorkItems)
+            {
+                throw new InvalidOperationException("Internal configuration is inconsistent. Backend is using separate queues for orchestration/entity dispatch, but frontend is pulling from single queue.");
+            }
+            return LockNextTaskOrchestrationWorkItemAsync(entitiesOnly: false, cancellationToken);
+        }
+
+        async Task<TaskOrchestrationWorkItem> LockNextTaskOrchestrationWorkItemAsync(bool entitiesOnly, CancellationToken cancellationToken)
         {
             Guid traceActivityId = StartNewLogicalTraceScope(useExisting: true);
 
@@ -644,7 +700,7 @@ public async Task<TaskOrchestrationWorkItem> LockNextTaskOrchestrationWorkItemAs
                 try
                 {
                     // This call will block until the next session is ready
-                    session = await this.orchestrationSessionManager.GetNextSessionAsync(linkedCts.Token);
+                    session = await this.orchestrationSessionManager.GetNextSessionAsync(entitiesOnly, linkedCts.Token);
                     if (session == null)
                     {
                         return null;
@@ -1634,7 +1690,7 @@ public async Task CreateTaskOrchestrationAsync(TaskMessage creationMessage, Orch
                 // An instance in this state already exists.
                 if (this.settings.ThrowExceptionOnInvalidDedupeStatus)
                 {
-                    throw new InvalidOperationException($"An Orchestration instance with the status {existingInstance.State.OrchestrationStatus} already exists.");
+                    throw new OrchestrationAlreadyExistsException($"An Orchestration instance with the status {existingInstance.State.OrchestrationStatus} already exists.");
                 }
 
                 return;
@@ -2064,6 +2120,7 @@ private static OrchestrationInstanceStatusQueryCondition ToAzureStorageCondition
                 TaskHubNames = condition.TaskHubNames,
                 InstanceIdPrefix = condition.InstanceIdPrefix,
                 FetchInput = condition.FetchInputsAndOutputs,
+                ExcludeEntities = condition.ExcludeEntities,
             };
         }
 

src/DurableTask.AzureStorage/AzureStorageOrchestrationServiceSettings.cs

@@ -85,6 +85,12 @@ public class AzureStorageOrchestrationServiceSettings
         /// </summary>
         public int MaxConcurrentTaskOrchestrationWorkItems { get; set; } = 100;
 
+        /// <summary>
+        /// Gets or sets the maximum number of entity operation batches that can be processed concurrently on a single node.
+        /// The default value is 100.
+        /// </summary>
+        public int MaxConcurrentTaskEntityWorkItems { get; set; } = 100;
+
         /// <summary>
         /// Gets or sets the maximum number of concurrent storage operations that can be executed in the context
         /// of a single orchestration instance.
@@ -234,5 +240,31 @@ internal LogHelper Logger
                 return this.logHelper;
             }
         }
+
+        /// <summary>
+        /// Gets or sets the limit on the number of entity operations that should be processed as a single batch.
+        /// A null value indicates that no particular limit should be enforced.
+        /// </summary>
+        /// <remarks>
+        /// Limiting the batch size can help to avoid timeouts in execution environments that impose time limitations on work items.
+        /// If set to 1, batching is disabled, and each operation executes as a separate work item.
+        /// </remarks>
+        /// <value>
+        /// A positive integer, or null.
+        /// </value>
+        public int? MaxEntityOperationBatchSize { get; set; } = null;
+
+        /// <summary>
+        /// Gets or sets the time window within which entity messages get deduplicated and reordered.
+        /// If set to zero, there is no sorting or deduplication, and all messages are just passed through.
+        /// </summary>
+        public int EntityMessageReorderWindowInMinutes { get; set; } = 30;
+
+        /// <summary>
+        /// Whether to use separate work item queues for entities and orchestrators.
+        /// This defaults to false, to avoid issues when using this provider from code that does not support separate dispatch.
+        /// Consumers that require separate dispatch (such as the new out-of-proc v2 SDKs) must set this to true.
+        /// </summary>
+        public bool UseSeparateQueueForEntityWorkItems { get; set; } = false;
     }
 }

src/DurableTask.AzureStorage/DurableTask.AzureStorage.csproj

@@ -30,7 +30,13 @@
     <FileVersion Condition="'$(FileVersionRevision)' != ''">$(VersionPrefix).$(FileVersionRevision)</FileVersion>
     <!-- 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 -->
+  </PropertyGroup>
+
+   <!-- This version is used as the nuget package version -->
+  <PropertyGroup Condition="$(VersionSuffix) == ''">
+    <Version>$(VersionPrefix)</Version>
+  </PropertyGroup>
+  <PropertyGroup Condition="$(VersionSuffix) != ''">
     <Version>$(VersionPrefix)-$(VersionSuffix)</Version>
   </PropertyGroup>
 
@@ -45,7 +51,7 @@
 
   <ItemGroup Condition="'$(TargetFramework)' == 'net462'">
     <PackageReference Include="Microsoft.Extensions.Logging.Abstractions" Version="1.1.1" />
-    <PackageReference Include="Newtonsoft.Json" Version="7.0.1" />
+    <PackageReference Include="WindowsAzure.Storage" version="7.2.1" />
   </ItemGroup>
 
   <ItemGroup Condition="'$(TargetFramework)' == 'netstandard2.0'">