Update protobuf, add suspend and resume event support (#97)

Author: jviau

CHANGELOG.md

@@ -2,6 +2,7 @@
 
 ## v1.0.0
 
+- Added `SuspendInstanceAsync` and `ResumeInstanceAsync` to `DurableTaskClient`.
 - `TaskOrchestrationContext.CreateReplaySafeLogger` now creates `ILogger` directly (as opposed to wrapping an existing `ILogger`).
 - Durable Functions class-based syntax now resolves `ITaskActivity` instances from `IServiceProvider`, if available there.
 - `DurableTaskClient` methods have been touched up to ensure `CancellationToken` is included, as well as is the last parameter.

src/Client/Core/DurableTaskClient.cs

@@ -178,6 +178,33 @@ public virtual Task<OrchestrationMetadata> WaitForInstanceStartAsync(
         string instanceId, CancellationToken cancellation)
         => this.WaitForInstanceStartAsync(instanceId, false, cancellation);
 
+    /// <summary>
+    /// Suspends an orchestration instance, halting processing of it until <see cref="ResumeInstanceAsync" /> is used
+    /// to resume the orchestration.
+    /// </summary>
+    /// <param name="instanceId">The instance ID of the orchestration to suspend.</param>
+    /// <param name="reason">The optional suspension reason.</param>
+    /// <param name="cancellation">
+    /// A <see cref="CancellationToken"/> that can be used to cancel the suspend operation. Note, cancelling this token
+    /// does <b>not</b> resume the orchestration if suspend was successful.
+    /// </param>
+    /// <returns>A task that completes when the suspend has been committed to the backend.</returns>
+    public abstract Task SuspendInstanceAsync(
+        string instanceId, string? reason = null, CancellationToken cancellation = default);
+
+    /// <summary>
+    /// Resumes an orchestration instance that was suspended via <see cref="SuspendInstanceAsync" />.
+    /// </summary>
+    /// <param name="instanceId">The instance ID of the orchestration to resume.</param>
+    /// <param name="reason">The optional resume reason.</param>
+    /// <param name="cancellation">
+    /// A <see cref="CancellationToken"/> that can be used to cancel the resume operation. Note, cancelling this token
+    /// does <b>not</b> re-suspend the orchestration if resume was successful.
+    /// </param>
+    /// <returns>A task that completes when the resume has been committed to the backend.</returns>
+    public abstract Task ResumeInstanceAsync(
+        string instanceId, string? reason = null, CancellationToken cancellation = default);
+
     /// <summary>
     /// Waits for an orchestration to start running and returns a <see cref="OrchestrationMetadata"/>
     /// object that contains metadata about the started instance.

src/Client/Core/OrchestrationRuntimeStatus.cs

@@ -44,4 +44,9 @@ public enum OrchestrationRuntimeStatus
     /// The orchestration was scheduled but hasn't started running.
     /// </summary>
     Pending,
+
+    /// <summary>
+    /// The orchestration is in a suspended state.
+    /// </summary>
+    Suspended,
 }

src/Client/Grpc/Client.Grpc.csproj

@@ -13,8 +13,7 @@
   </PropertyGroup>
 
   <ItemGroup>
-    <PackageReference Include="Grpc.Core" Version="2.39.1" />
-    <PackageReference Include="Microsoft.DurableTask.Sidecar.Protobuf" Version="0.3.1" />
+    <PackageReference Include="Microsoft.DurableTask.Sidecar.Protobuf" Version="1.0.0" />
     <PackageReference Include="Microsoft.Extensions.Configuration.Abstractions" Version="6.0.0" />
     <PackageReference Include="Microsoft.Extensions.DependencyInjection.Abstractions" Version="6.0.0" />
   </ItemGroup>

src/Client/Grpc/GrpcDurableTaskClient.cs

@@ -136,6 +136,50 @@ await this.sidecarClient.TerminateInstanceAsync(
             cancellationToken: cancellation);
     }
 
+    /// <inheritdoc/>
+    public override async Task SuspendInstanceAsync(
+        string instanceId, string? reason = null, CancellationToken cancellation = default)
+    {
+        P.SuspendRequest request = new()
+        {
+            InstanceId = instanceId,
+            Reason = reason,
+        };
+
+        try
+        {
+            await this.sidecarClient.SuspendInstanceAsync(
+                request, cancellationToken: cancellation);
+        }
+        catch (RpcException e) when (e.StatusCode == StatusCode.Cancelled)
+        {
+            throw new OperationCanceledException(
+                $"The {nameof(this.SuspendInstanceAsync)} operation was canceled.", e, cancellation);
+        }
+    }
+
+    /// <inheritdoc/>
+    public override async Task ResumeInstanceAsync(
+        string instanceId, string? reason = null, CancellationToken cancellation = default)
+    {
+        P.ResumeRequest request = new()
+        {
+            InstanceId = instanceId,
+            Reason = reason,
+        };
+
+        try
+        {
+            await this.sidecarClient.ResumeInstanceAsync(
+                request, cancellationToken: cancellation);
+        }
+        catch (RpcException e) when (e.StatusCode == StatusCode.Cancelled)
+        {
+            throw new OperationCanceledException(
+                $"The {nameof(this.ResumeInstanceAsync)} operation was canceled.", e, cancellation);
+        }
+    }
+
     /// <inheritdoc/>
     public override async Task<OrchestrationMetadata?> GetInstanceMetadataAsync(
         string instanceId, bool getInputsAndOutputs = false, CancellationToken cancellation = default)

src/Client/Grpc/ProtoUtils.cs

@@ -71,6 +71,7 @@ internal static P.OrchestrationStatus ToGrpcStatus(this OrchestrationRuntimeStat
             OrchestrationRuntimeStatus.Pending => P.OrchestrationStatus.Pending,
             OrchestrationRuntimeStatus.Running => P.OrchestrationStatus.Running,
             OrchestrationRuntimeStatus.Terminated => P.OrchestrationStatus.Terminated,
+            OrchestrationRuntimeStatus.Suspended => P.OrchestrationStatus.Suspended,
             _ => throw new ArgumentOutOfRangeException(nameof(status), "Unexpected value"),
         };
 #pragma warning restore 0618 // Referencing Obsolete member.

src/Shared/Grpc/ProtoUtils.cs

@@ -60,6 +60,12 @@ internal static HistoryEvent ConvertHistoryEvent(P.HistoryEvent proto)
             case P.HistoryEvent.EventTypeOneofCase.ExecutionTerminated:
                 historyEvent = new ExecutionTerminatedEvent(proto.EventId, proto.ExecutionTerminated.Input);
                 break;
+            case P.HistoryEvent.EventTypeOneofCase.ExecutionSuspended:
+                historyEvent = new ExecutionSuspendedEvent(proto.EventId, proto.ExecutionSuspended.Input);
+                break;
+            case P.HistoryEvent.EventTypeOneofCase.ExecutionResumed:
+                historyEvent = new ExecutionResumedEvent(proto.EventId, proto.ExecutionResumed.Input);
+                break;
             case P.HistoryEvent.EventTypeOneofCase.TaskScheduled:
                 historyEvent = new TaskScheduledEvent(
                     proto.EventId,

src/Worker/Grpc/Worker.Grpc.csproj

@@ -13,10 +13,9 @@
   </PropertyGroup>
 
   <ItemGroup>
-    <PackageReference Include="Grpc.Core" Version="2.39.1" />
     <PackageReference Include="Microsoft.AspNetCore.Server.Kestrel" Version="2.2.0" />
     <PackageReference Include="Microsoft.AspNetCore.Hosting" Version="2.2.7" />
-    <PackageReference Include="Microsoft.DurableTask.Sidecar.Protobuf" Version="0.3.1" />
+    <PackageReference Include="Microsoft.DurableTask.Sidecar.Protobuf" Version="1.0.0" />
     <PackageReference Include="System.Collections.Immutable" Version="6.0.0" />
   </ItemGroup>
 

test/Client/Core.Tests/DependencyInjection/DefaultDurableTaskClientBuilderTests.cs

@@ -106,6 +106,12 @@ public override Task RaiseEventAsync(
             throw new NotImplementedException();
         }
 
+        public override Task ResumeInstanceAsync(
+            string instanceId, string? reason = null, CancellationToken cancellation = default)
+        {
+            throw new NotImplementedException();
+        }
+
         public override Task<string> ScheduleNewOrchestrationInstanceAsync(
             TaskName orchestratorName,
             object? input = null,
@@ -115,6 +121,12 @@ public override Task<string> ScheduleNewOrchestrationInstanceAsync(
             throw new NotImplementedException();
         }
 
+        public override Task SuspendInstanceAsync(
+            string instanceId, string? reason = null, CancellationToken cancellation = default)
+        {
+            throw new NotImplementedException();
+        }
+
         public override Task TerminateAsync(
             string instanceId, object? output = null, CancellationToken cancellation = default)
         {

test/Client/Core.Tests/DependencyInjection/DurableTaskClientBuilderExtensionsTests.cs

@@ -136,6 +136,12 @@ public override Task RaiseEventAsync(
             throw new NotImplementedException();
         }
 
+        public override Task ResumeInstanceAsync(
+            string instanceId, string? reason = null, CancellationToken cancellation = default)
+        {
+            throw new NotImplementedException();
+        }
+
         public override Task<string> ScheduleNewOrchestrationInstanceAsync(
             TaskName orchestratorName,
             object? input = null,
@@ -145,6 +151,12 @@ public override Task<string> ScheduleNewOrchestrationInstanceAsync(
             throw new NotImplementedException();
         }
 
+        public override Task SuspendInstanceAsync(
+            string instanceId, string? reason = null, CancellationToken cancellation = default)
+        {
+            throw new NotImplementedException();
+        }
+
         public override Task TerminateAsync(
             string instanceId, object? output = null, CancellationToken cancellation = default)
         {