Nexus cancellation sample

README.md

@@ -26,6 +26,7 @@ Prerequisites:
 * [Encryption](src/Encryption) - End-to-end encryption with Temporal payload codecs.
 * [EnvConfig](src/EnvConfig) - Load client configuration from TOML files with programmatic overrides
 * [Mutex](src/Mutex) - How to implement a mutex as a workflow. Demonstrates how to avoid race conditions or parallel mutually exclusive operations on the same resource.
+* [NexusCancellation](src/NexusCancellation) - Demonstrates how to cancel a running Nexus operation from a caller workflow.
 * [NexusContextPropagation](src/NexusContextPropagation) - Context propagation through Nexus services.
 * [NexusMultiArg](src/NexusMultiArg) - Nexus service implementation calling a workflow with multiple arguments.
 * [NexusSimple](src/NexusSimple) - Simple Nexus service implementation.

TemporalioSamples.sln

@@ -103,6 +103,10 @@ Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "TemporalioSamples.Updatable
 EndProject
 Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "TemporalioSamples.Dsl", "src\Dsl\TemporalioSamples.Dsl.csproj", "{AF077751-E4B9-4696-93CB-74653F0BB6C4}"
 EndProject
+Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "TemporalioSamples.NexusCancellation", "src\NexusCancellation\TemporalioSamples.NexusCancellation.csproj", "{6D0BE4C4-9C4F-4A3D-78F1-B0B761568559}"
+EndProject
+Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "NexusCancellation", "NexusCancellation", "{7123C63D-3158-4C9A-8EAD-6D4F1295BC04}"
+EndProject
 Global
 	GlobalSection(SolutionConfigurationPlatforms) = preSolution
 		Debug|Any CPU = Debug|Any CPU
@@ -617,6 +621,18 @@ Global
 		{AF077751-E4B9-4696-93CB-74653F0BB6C4}.Release|x64.Build.0 = Release|Any CPU
 		{AF077751-E4B9-4696-93CB-74653F0BB6C4}.Release|x86.ActiveCfg = Release|Any CPU
 		{AF077751-E4B9-4696-93CB-74653F0BB6C4}.Release|x86.Build.0 = Release|Any CPU
+		{6D0BE4C4-9C4F-4A3D-78F1-B0B761568559}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
+		{6D0BE4C4-9C4F-4A3D-78F1-B0B761568559}.Debug|Any CPU.Build.0 = Debug|Any CPU
+		{6D0BE4C4-9C4F-4A3D-78F1-B0B761568559}.Debug|x64.ActiveCfg = Debug|Any CPU
+		{6D0BE4C4-9C4F-4A3D-78F1-B0B761568559}.Debug|x64.Build.0 = Debug|Any CPU
+		{6D0BE4C4-9C4F-4A3D-78F1-B0B761568559}.Debug|x86.ActiveCfg = Debug|Any CPU
+		{6D0BE4C4-9C4F-4A3D-78F1-B0B761568559}.Debug|x86.Build.0 = Debug|Any CPU
+		{6D0BE4C4-9C4F-4A3D-78F1-B0B761568559}.Release|Any CPU.ActiveCfg = Release|Any CPU
+		{6D0BE4C4-9C4F-4A3D-78F1-B0B761568559}.Release|Any CPU.Build.0 = Release|Any CPU
+		{6D0BE4C4-9C4F-4A3D-78F1-B0B761568559}.Release|x64.ActiveCfg = Release|Any CPU
+		{6D0BE4C4-9C4F-4A3D-78F1-B0B761568559}.Release|x64.Build.0 = Release|Any CPU
+		{6D0BE4C4-9C4F-4A3D-78F1-B0B761568559}.Release|x86.ActiveCfg = Release|Any CPU
+		{6D0BE4C4-9C4F-4A3D-78F1-B0B761568559}.Release|x86.Build.0 = Release|Any CPU
 	EndGlobalSection
 	GlobalSection(SolutionProperties) = preSolution
 		HideSolutionNode = FALSE
@@ -668,5 +684,7 @@ Global
 		{B37B3E98-4B04-48B8-9017-F0EDEDC7BD98} = {1A647B41-53D0-4638-AE5A-6630BAAE45FC}
 		{5D02DFEA-DC08-4B7B-8E26-EDAC1942D347} = {1A647B41-53D0-4638-AE5A-6630BAAE45FC}
 		{AF077751-E4B9-4696-93CB-74653F0BB6C4} = {1A647B41-53D0-4638-AE5A-6630BAAE45FC}
+		{6D0BE4C4-9C4F-4A3D-78F1-B0B761568559} = {7123C63D-3158-4C9A-8EAD-6D4F1295BC04}
+		{7123C63D-3158-4C9A-8EAD-6D4F1295BC04} = {1A647B41-53D0-4638-AE5A-6630BAAE45FC}
 	EndGlobalSection
 EndGlobal

src/NexusCancellation/Caller/HelloCallerWorkflow.workflow.cs

@@ -0,0 +1,64 @@
+namespace TemporalioSamples.NexusCancellation.Caller;
+
+using Microsoft.Extensions.Logging;
+using Temporalio.Exceptions;
+using Temporalio.Workflows;
+
+[Workflow]
+public class HelloCallerWorkflow
+{
+    private static readonly IHelloService.HelloLanguage[] Languages =
+        [IHelloService.HelloLanguage.En, IHelloService.HelloLanguage.Fr, IHelloService.HelloLanguage.De,
+        IHelloService.HelloLanguage.Es, IHelloService.HelloLanguage.Tr];
+
+    [WorkflowRun]
+    public async Task<string> RunAsync(string name)
+    {
+        using var cts = CancellationTokenSource.CreateLinkedTokenSource(Workflow.CancellationToken);
+        var client = Workflow.CreateNexusClient<IHelloService>(IHelloService.EndpointName);
+
+        // Concurrently execute an operation per language.
+        var tasks = Languages.Select(lang =>
+            client.ExecuteNexusOperationAsync(
+                svc => svc.SayHello(new IHelloService.HelloInput(name, lang)),
+                new NexusOperationOptions
+                {
+                    // We set the CancellationType to WaitCancellationRequested, which means the caller waits
+                    // for the request to be received by the handler before proceeding with the cancellation.
+                    //
+                    // The default CancellationType is WaitCancellationCompleted, where the caller would wait
+                    // until the operation is completed.
+                    CancellationType = NexusOperationCancellationType.WaitCancellationRequested,
+                    CancellationToken = cts.Token,
+                })).ToList();
+
+        var firstTask = await Workflow.WhenAnyAsync(tasks);
+
+        Workflow.Logger.LogInformation("First operation completed, cancelling remaining operations");
+
+        // Now that the first operation has won the race, we are going to cancel the other operations.
+#pragma warning disable CA1849, VSTHRD103 // CancelAsync() is non-deterministic in workflows.
+        cts.Cancel();
+#pragma warning restore CA1849, VSTHRD103
+
+        // Wait for all tasks to resolve. Once the workflow completes, the server will stop trying to cancel any of
+        // the operations that have not yet received cancellation, letting them run to completion. We are using the
+        // CancellationType of WaitCancellationRequested so these tasks will return as soon as the operation has received
+        // the cancellation request.
+        foreach (var task in tasks)
+        {
+            try
+            {
+                await task;
+            }
+            // Only throw an error if an operation errored out not due to cancellation.
+            catch (Exception ex) when (TemporalException.IsCanceledException(ex))
+            {
+                Workflow.Logger.LogInformation("Operation was cancelled");
+            }
+        }
+
+        var result = await firstTask;
+        return result?.Message ?? throw new ApplicationFailureException("No successful result");
+    }
+}

src/NexusCancellation/Handler/HelloHandlerWorkflow.workflow.cs

@@ -0,0 +1,52 @@
+namespace TemporalioSamples.NexusCancellation.Handler;
+
+using Microsoft.Extensions.Logging;
+using Temporalio.Exceptions;
+using Temporalio.Workflows;
+
+[Workflow]
+public class HelloHandlerWorkflow
+{
+    [WorkflowRun]
+    public async Task<IHelloService.HelloOutput> RunAsync(IHelloService.HelloInput input)
+    {
+        Workflow.Logger.LogInformation(
+            "HelloHandlerWorkflow started for {Name} in {Language}",
+            input.Name,
+            input.Language);
+
+        // Sleep for a random duration to simulate some work
+        var duration = TimeSpan.FromSeconds(Workflow.Random.Next(5));
+
+        try
+        {
+            await Workflow.DelayAsync(duration);
+        }
+        catch (Exception ex) when (TemporalException.IsCanceledException(ex))
+        {
+            // Simulate cleanup work after cancellation is requested.
+            // Use CancellationToken.None to create a "disconnected" context.
+            var cleanupDuration = TimeSpan.FromSeconds(Workflow.Random.Next(5));
+            await Workflow.DelayAsync(cleanupDuration, CancellationToken.None);
+
+            Workflow.Logger.LogInformation(
+                "HelloHandlerWorkflow for {Name} in {Language} was cancelled after {Duration} of work, performed {CleanupDuration} of cleanup",
+                input.Name,
+                input.Language,
+                duration,
+                cleanupDuration);
+            throw; // Re-throw the cancellation after cleanup
+        }
+
+        return input.Language switch
+        {
+            IHelloService.HelloLanguage.En => new($"Hello {input.Name} 👋"),
+            IHelloService.HelloLanguage.Fr => new($"Bonjour {input.Name} 👋"),
+            IHelloService.HelloLanguage.De => new($"Hallo {input.Name} 👋"),
+            IHelloService.HelloLanguage.Es => new($"¡Hola! {input.Name} 👋"),
+            IHelloService.HelloLanguage.Tr => new($"Merhaba {input.Name} 👋"),
+            _ => throw new ApplicationFailureException(
+                $"Unsupported language: {input.Language}", errorType: "UNSUPPORTED_LANGUAGE"),
+        };
+    }
+}

src/NexusCancellation/Handler/HelloService.cs

@@ -0,0 +1,21 @@
+namespace TemporalioSamples.NexusCancellation.Handler;
+
+using NexusRpc.Handlers;
+using Temporalio.Nexus;
+
+[NexusServiceHandler(typeof(IHelloService))]
+public class HelloService
+{
+    [NexusOperationHandler]
+    public IOperationHandler<IHelloService.HelloInput, IHelloService.HelloOutput> SayHello() =>
+        // This Nexus service operation is backed by a workflow run
+        WorkflowRunOperationHandler.FromHandleFactory(
+            (WorkflowRunOperationContext context, IHelloService.HelloInput input) =>
+                context.StartWorkflowAsync(
+                    (HelloHandlerWorkflow wf) => wf.RunAsync(input),
+                    // Workflow IDs should typically be business meaningful IDs and are used to
+                    // dedupe workflow starts. For this example, we're using the request ID
+                    // allocated by Temporal when the caller workflow schedules the operation,
+                    // this ID is guaranteed to be stable across retries of this operation.
+                    new() { Id = context.HandlerContext.RequestId }));
+}

src/NexusCancellation/IHelloService.cs

@@ -0,0 +1,25 @@
+namespace TemporalioSamples.NexusCancellation;
+
+using NexusRpc;
+
+[NexusService]
+public interface IHelloService
+{
+    static readonly string EndpointName = "nexus-cancellation-endpoint";
+
+    [NexusOperation]
+    HelloOutput SayHello(HelloInput input);
+
+    public record HelloInput(string Name, HelloLanguage Language);
+
+    public record HelloOutput(string Message);
+
+    public enum HelloLanguage
+    {
+        En,
+        Fr,
+        De,
+        Es,
+        Tr,
+    }
+}

src/NexusCancellation/Program.cs

@@ -0,0 +1,94 @@
+using Microsoft.Extensions.Logging;
+using Temporalio.Client;
+using Temporalio.Common.EnvConfig;
+using Temporalio.Worker;
+using TemporalioSamples.NexusCancellation.Caller;
+using TemporalioSamples.NexusCancellation.Handler;
+
+using var loggerFactory = LoggerFactory.Create(builder =>
+    builder.
+        AddSimpleConsole(options => options.TimestampFormat = "[HH:mm:ss] ").
+        SetMinimumLevel(LogLevel.Information));
+var logger = loggerFactory.CreateLogger<Program>();
+
+// Cancellation token cancelled on ctrl+c
+using var tokenSource = new CancellationTokenSource();
+Console.CancelKeyPress += (_, eventArgs) =>
+{
+    tokenSource.Cancel();
+    eventArgs.Cancel = true;
+};
+
+Task<TemporalClient> ConnectClientAsync(string temporalNamespace)
+{
+    var connectOptions = ClientEnvConfig.LoadClientConnectOptions();
+    connectOptions.TargetHost ??= "localhost:7233";
+    connectOptions.Namespace = temporalNamespace;
+    connectOptions.LoggerFactory = loggerFactory;
+    return TemporalClient.ConnectAsync(connectOptions);
+}
+
+async Task RunHandlerWorkerAsync()
+{
+    // Run worker until cancelled
+    logger.LogInformation("Running handler worker");
+    using var worker = new TemporalWorker(
+        await ConnectClientAsync("nexus-cancellation-handler-namespace"),
+        new TemporalWorkerOptions(taskQueue: "nexus-cancellation-handler-sample").
+            AddNexusService(new HelloService()).
+            AddWorkflow<HelloHandlerWorkflow>());
+    try
+    {
+        await worker.ExecuteAsync(tokenSource.Token);
+    }
+    catch (OperationCanceledException)
+    {
+        logger.LogInformation("Handler worker cancelled");
+    }
+}
+
+async Task RunCallerWorkerAsync()
+{
+    // Run worker until cancelled
+    logger.LogInformation("Running caller worker");
+    using var worker = new TemporalWorker(
+        await ConnectClientAsync("nexus-cancellation-caller-namespace"),
+        new TemporalWorkerOptions(taskQueue: "nexus-cancellation-caller-sample")
+            .AddWorkflow<HelloCallerWorkflow>());
+    try
+    {
+        await worker.ExecuteAsync(tokenSource.Token);
+    }
+    catch (OperationCanceledException)
+    {
+        logger.LogInformation("Caller worker cancelled");
+    }
+}
+
+async Task ExecuteCallerWorkflowAsync()
+{
+    logger.LogInformation("Executing caller hello workflow");
+
+    var client = await ConnectClientAsync("nexus-cancellation-caller-namespace");
+
+    var result = await client.ExecuteWorkflowAsync(
+        (HelloCallerWorkflow wf) => wf.RunAsync("Temporal"),
+        new(id: "nexus-cancellation-hello-id", taskQueue: "nexus-cancellation-caller-sample"));
+    logger.LogInformation("Workflow result: {Result}", result);
+}
+
+switch (args.ElementAtOrDefault(0))
+{
+    case "handler-worker":
+        await RunHandlerWorkerAsync();
+        break;
+    case "caller-worker":
+        await RunCallerWorkerAsync();
+        break;
+    case "caller-workflow":
+        await ExecuteCallerWorkflowAsync();
+        break;
+    default:
+        throw new ArgumentException(
+            "Must pass 'handler-worker', 'caller-worker', or 'caller-workflow' as the single argument");
+}