Merge branch 'main' into wangbill/largepayloadv2

Author: YunchuWang

CHANGELOG.md

@@ -1,5 +1,8 @@
 # Changelog
-
+## v1.15.0
+- Abandon workitem if processing workitem failed by @YunchuWang in ([#467](https://github.com/microsoft/durabletask-dotnet/pull/467))
+- Extended Sessions for Isolated (Orchestrations) by @sophiatev in ([#449](https://github.com/microsoft/durabletask-dotnet/pull/449))
+  
 ## v1.14.0
 - Add RestartAsync API Support at DurableTaskClient ([#456](https://github.com/microsoft/durabletask-dotnet/pull/456))
 
@@ -292,3 +295,4 @@ Microsoft.DurableTask.Generators
 ## v0.4.1-beta
 
 Initial public release
+

Directory.Packages.props

@@ -9,6 +9,7 @@
 
   <!-- Microsoft.Extensions.* Packages -->
   <ItemGroup>
+    <PackageVersion Include="Microsoft.Extensions.Caching.Memory" Version="6.0.3" />
     <PackageVersion Include="Microsoft.Extensions.Configuration.Abstractions" Version="6.0.0" />
     <PackageVersion Include="Microsoft.Extensions.DependencyInjection" Version="6.0.2" />
     <PackageVersion Include="Microsoft.Extensions.DependencyInjection.Abstractions" Version="6.0.0" />
@@ -29,7 +30,7 @@
 
   <!-- DurableTask Packages -->
   <ItemGroup>
-    <PackageVersion Include="Microsoft.Azure.DurableTask.Core" Version="3.3.0" />
+    <PackageVersion Include="Microsoft.Azure.DurableTask.Core" Version="3.4.0" />
     <PackageVersion Include="Microsoft.Azure.Functions.Worker.Extensions.DurableTask" Version="1.2.2" />
   </ItemGroup>
 

eng/targets/Release.props

@@ -17,7 +17,7 @@
   </PropertyGroup>
 
   <PropertyGroup>
-    <VersionPrefix>1.14.0</VersionPrefix>
+    <VersionPrefix>1.15.0</VersionPrefix>
     <VersionSuffix></VersionSuffix>
   </PropertyGroup>
 

src/Grpc/orchestrator_service.proto

@@ -342,8 +342,10 @@ message OrchestratorResponse {
     // The number of work item events that were processed by the orchestrator.
     // This field is optional. If not set, the service should assume that the orchestrator processed all events.
     google.protobuf.Int32Value numEventsProcessed = 5;
-
     OrchestrationTraceContext orchestrationTraceContext = 6;
+
+    // Whether or not a history is required to complete the original OrchestratorRequest and none was provided.
+    bool requiresHistory = 7;
 }
 
 message CreateInstanceRequest {
@@ -678,6 +680,18 @@ message AbandonEntityTaskResponse {
     // Empty.
 }
 
+message SkipGracefulOrchestrationTerminationsRequest {
+    // A maximum of 500 instance IDs can be provided in this list.
+    repeated string instanceIds = 1;
+    google.protobuf.StringValue reason = 2;
+}
+
+message SkipGracefulOrchestrationTerminationsResponse {
+    // Those instances which could not be terminated because they had locked entities at the time of this termination call,
+    // are already in a terminal state (completed, failed, terminated, etc.), are not orchestrations, or do not exist (i.e. have been purged)
+    repeated string unterminatedInstanceIds = 1;
+}
+
 service TaskHubSidecarService {
     // Sends a hello request to the sidecar service.
     rpc Hello(google.protobuf.Empty) returns (google.protobuf.Empty);
@@ -751,6 +765,10 @@ service TaskHubSidecarService {
 
     // Abandon an entity work item
     rpc AbandonTaskEntityWorkItem(AbandonEntityTaskRequest) returns (AbandonEntityTaskResponse);
+
+    // "Skip" graceful termination of orchestrations by immediately changing their status in storage to "terminated".
+    // Note that a maximum of 500 orchestrations can be terminated at a time using this method.
+    rpc SkipGracefulOrchestrationTerminations(SkipGracefulOrchestrationTerminationsRequest) returns (SkipGracefulOrchestrationTerminationsResponse);
 }
 
 message GetWorkItemsRequest {

src/Grpc/versions.txt

@@ -1,2 +1,2 @@
-# The following files were downloaded from branch main at 2025-09-10 22:50:45 UTC
-https://raw.githubusercontent.com/microsoft/durabletask-protobuf/985035a0890575ae18be0eb2a3ac93c10824498a/protos/orchestrator_service.proto
+# The following files were downloaded from branch main at 2025-09-17 01:45:58 UTC
+https://raw.githubusercontent.com/microsoft/durabletask-protobuf/f5745e0d83f608d77871c1894d9260ceaae08967/protos/orchestrator_service.proto

src/Shared/Grpc/ProtoUtils.cs

@@ -279,18 +279,19 @@ internal static Timestamp ToTimestamp(this DateTime dateTime)
     /// </param>
     /// <param name="entityConversionState">The entity conversion state, or null if no conversion is required.</param>
     /// <param name="orchestrationActivity">The <see cref="Activity" /> that represents orchestration execution.</param>
+    /// <param name="requiresHistory">Whether or not a history is required to complete the orchestration request and none was provided.</param>
     /// <returns>The orchestrator response.</returns>
     /// <exception cref="NotSupportedException">When an orchestrator action is unknown.</exception>
     internal static P.OrchestratorResponse ConstructOrchestratorResponse(
         string instanceId,
         string executionId,
         string? customStatus,
-        IEnumerable<OrchestratorAction> actions,
+        IEnumerable<OrchestratorAction>? actions,
         string completionToken,
         EntityConversionState? entityConversionState,
-        Activity? orchestrationActivity)
+        Activity? orchestrationActivity,
+        bool requiresHistory = false)
     {
-        Check.NotNull(actions);
         var response = new P.OrchestratorResponse
         {
             InstanceId = instanceId,
@@ -302,8 +303,16 @@ internal static P.OrchestratorResponse ConstructOrchestratorResponse(
                     SpanID = orchestrationActivity?.SpanId.ToString(),
                     SpanStartTime = orchestrationActivity?.StartTimeUtc.ToTimestamp(),
                 },
+            RequiresHistory = requiresHistory,
         };
 
+        // If a history is required and the orchestration request was not completed, then there is no list of actions.
+        if (requiresHistory)
+        {
+            return response;
+        }
+
+        Check.NotNull(actions);
         foreach (OrchestratorAction action in actions)
         {
             var protoAction = new P.OrchestratorAction { Id = action.Id };

src/Worker/Core/ExtendedSessionState.cs

@@ -0,0 +1,40 @@
+// Copyright (c) Microsoft Corporation.
+// Licensed under the MIT License.
+
+using DurableTask.Core;
+
+namespace Microsoft.DurableTask.Worker;
+
+/// <summary>
+/// Represents the state of an extended session for an orchestration.
+/// </summary>
+public class ExtendedSessionState
+{
+    /// <summary>
+    /// Initializes a new instance of the <see cref="ExtendedSessionState"/> class.
+    /// </summary>
+    /// <param name="state">The orchestration's runtime state.</param>
+    /// <param name="taskOrchestration">The TaskOrchestration implementation of the orchestration.</param>
+    /// <param name="orchestrationExecutor">The TaskOrchestrationExecutor for the orchestration.</param>
+    public ExtendedSessionState(OrchestrationRuntimeState state, TaskOrchestration taskOrchestration, TaskOrchestrationExecutor orchestrationExecutor)
+    {
+        this.RuntimeState = state;
+        this.TaskOrchestration = taskOrchestration;
+        this.OrchestrationExecutor = orchestrationExecutor;
+    }
+
+    /// <summary>
+    /// Gets or sets the saved runtime state of the orchestration.
+    /// </summary>
+    public OrchestrationRuntimeState RuntimeState { get; set; }
+
+    /// <summary>
+    /// Gets or sets the saved TaskOrchestration implementation of the orchestration.
+    /// </summary>
+    public TaskOrchestration TaskOrchestration { get; set; }
+
+    /// <summary>
+    /// Gets or sets the saved TaskOrchestrationExecutor.
+    /// </summary>
+    public TaskOrchestrationExecutor OrchestrationExecutor { get; set; }
+}

src/Worker/Core/ExtendedSessionsCache.cs

@@ -0,0 +1,48 @@
+// Copyright (c) Microsoft Corporation.
+// Licensed under the MIT License.
+
+using Microsoft.Extensions.Caching.Memory;
+
+namespace Microsoft.DurableTask.Worker;
+
+/// <summary>
+/// A cache for extended sessions that wraps a <see cref="MemoryCache"/> instance.
+/// Responsible for holding <see cref="ExtendedSessionState"/> for orchestrations that are running within extended sessions.
+/// </summary>
+public class ExtendedSessionsCache : IDisposable
+{
+    MemoryCache? extendedSessions;
+
+    /// <summary>
+    /// Gets a value indicating whether returns whether or not the cache has been initialized.
+    /// </summary>
+    /// <returns>True if the cache has been initialized, false otherwise.</returns>
+    public bool IsInitialized => this.extendedSessions is not null;
+
+    /// <summary>
+    /// Dispose the cache and release all resources.
+    /// </summary>
+    public void Dispose()
+    {
+        this.extendedSessions?.Dispose();
+        GC.SuppressFinalize(this);
+    }
+
+    /// <summary>
+    /// Gets the cache for extended sessions if it has already been initialized, or otherwise initializes it with the given expiration scan frequency.
+    /// </summary>
+    /// <param name="expirationScanFrequencyInSeconds">
+    /// The expiration scan frequency of the cache, in seconds.
+    /// This specifies how often the cache checks for stale items, and evicts them.
+    /// </param>
+    /// <returns>The IMemoryCache that holds the cached <see cref="ExtendedSessionState"/>.</returns>
+    public MemoryCache GetOrInitializeCache(double expirationScanFrequencyInSeconds)
+    {
+        this.extendedSessions ??= new MemoryCache(new MemoryCacheOptions
+        {
+            ExpirationScanFrequency = TimeSpan.FromSeconds(expirationScanFrequencyInSeconds / 5),
+        });
+
+        return this.extendedSessions;
+    }
+}

src/Worker/Core/Worker.csproj

@@ -9,6 +9,7 @@ The worker is responsible for processing durable task work items.</PackageDescri
   </PropertyGroup>
 
   <ItemGroup>
+    <PackageReference Include="Microsoft.Extensions.Caching.Memory" />
     <PackageReference Include="Microsoft.Extensions.Hosting.Abstractions" />
     <PackageReference Include="Microsoft.Extensions.Options" />
     <PackageReference Include="System.Text.Json" />