Task cancellation support (#200)

boma96 · web-flow · commit f31072744bb6 · 2025-03-24T09:23:12.000+01:00
* Move task processing into separate method

* Add initial implementation

* Add initial implementation

* Add debug logs

* Do not create cts on kafka event

* Make background service more robust

* Remove package, add null checks

* Suppress warnings

* Create experimental version

* Add github action step to build KafkaCancellationNotifier

* Fix log messages

* Bump version

* Bump version
diff --git a/.github/workflows/release.yml b/.github/workflows/release.yml
@@ -29,7 +29,9 @@ jobs:
     - name: Create ConductorSharp.Patterns package
       run: dotnet pack --no-restore ./src/ConductorSharp.Patterns -c Release 
     - name: Create ConductorSharp.Toolkit package
-      run: dotnet pack --no-restore ./src/ConductorSharp.Toolkit -c Release 
+      run: dotnet pack --no-restore ./src/ConductorSharp.Toolkit -c Release
+    - name: Create ConductorSharp.KafkaCancellationNotifier package
+      run: dotnet pack --no-restore ./src/ConductorSharp.KafkaCancellationNotifier -c Release 
 
     - name: Publish ConductorSharp.Client package
       run: dotnet nuget push ./src/ConductorSharp.Client/bin/Release/ConductorSharp.Client*.nupkg --skip-duplicate --api-key ${{secrets.NUGET_API_KEY}} --source https://api.nuget.org/v3/index.json
@@ -39,3 +41,5 @@ jobs:
       run: dotnet nuget push ./src/ConductorSharp.Patterns/bin/Release/ConductorSharp.Patterns*.nupkg --skip-duplicate --api-key ${{secrets.NUGET_API_KEY}} --source https://api.nuget.org/v3/index.json
     - name: Publish ConductorSharp.Toolkit package
       run: dotnet nuget push ./src/ConductorSharp.Toolkit/bin/Release/ConductorSharp.Toolkit*.nupkg --skip-duplicate --api-key ${{secrets.NUGET_API_KEY}} --source https://api.nuget.org/v3/index.json
+    - name: Publish ConductorSharp.KafkaCancellationNotifier package
+      run: dotnet nuget push ./src/ConductorSharp.KafkaCancellationNotifier/bin/Release/ConductorSharp.KafkaCancellationNotifier*.nupkg --skip-duplicate --api-key ${{secrets.NUGET_API_KEY}} --source https://api.nuget.org/v3/index.json
diff --git a/ConductorSharp.sln b/ConductorSharp.sln
@@ -26,7 +26,9 @@ Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "Solution Items", "Solution
 		.editorconfig = .editorconfig
 	EndProjectSection
 EndProject
-Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "ConductorSharp.Patterns", "src\ConductorSharp.Patterns\ConductorSharp.Patterns.csproj", "{9A55F1CF-2553-4F44-A8BC-0063A402582F}"
+Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "ConductorSharp.Patterns", "src\ConductorSharp.Patterns\ConductorSharp.Patterns.csproj", "{9A55F1CF-2553-4F44-A8BC-0063A402582F}"
+EndProject
+Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "ConductorSharp.KafkaCancellationNotifier", "src\ConductorSharp.KafkaCancellationNotifier\ConductorSharp.KafkaCancellationNotifier.csproj", "{A94EE48D-17F3-432A-A47D-BCB9B1EF2670}"
 EndProject
 Global
 	GlobalSection(SolutionConfigurationPlatforms) = preSolution
@@ -70,6 +72,10 @@ Global
 		{9A55F1CF-2553-4F44-A8BC-0063A402582F}.Debug|Any CPU.Build.0 = Debug|Any CPU
 		{9A55F1CF-2553-4F44-A8BC-0063A402582F}.Release|Any CPU.ActiveCfg = Release|Any CPU
 		{9A55F1CF-2553-4F44-A8BC-0063A402582F}.Release|Any CPU.Build.0 = Release|Any CPU
+		{A94EE48D-17F3-432A-A47D-BCB9B1EF2670}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
+		{A94EE48D-17F3-432A-A47D-BCB9B1EF2670}.Debug|Any CPU.Build.0 = Debug|Any CPU
+		{A94EE48D-17F3-432A-A47D-BCB9B1EF2670}.Release|Any CPU.ActiveCfg = Release|Any CPU
+		{A94EE48D-17F3-432A-A47D-BCB9B1EF2670}.Release|Any CPU.Build.0 = Release|Any CPU
 	EndGlobalSection
 	GlobalSection(SolutionProperties) = preSolution
 		HideSolutionNode = FALSE
diff --git a/examples/ConductorSharp.NoApi/ConductorSharp.NoApi.csproj b/examples/ConductorSharp.NoApi/ConductorSharp.NoApi.csproj
@@ -39,6 +39,7 @@
 
   <ItemGroup>
     <ProjectReference Include="..\..\src\ConductorSharp.Engine\ConductorSharp.Engine.csproj" />
+    <ProjectReference Include="..\..\src\ConductorSharp.KafkaCancellationNotifier\ConductorSharp.KafkaCancellationNotifier.csproj" />
     <ProjectReference Include="..\..\src\ConductorSharp.Patterns\ConductorSharp.Patterns.csproj" />
   </ItemGroup>
 
diff --git a/examples/ConductorSharp.NoApi/Program.cs b/examples/ConductorSharp.NoApi/Program.cs
@@ -1,5 +1,6 @@
 ﻿using ConductorSharp.Engine.Extensions;
 using ConductorSharp.Engine.Health;
+using ConductorSharp.KafkaCancellationNotifier.Extensions;
 using ConductorSharp.NoApi.Behaviors;
 using ConductorSharp.NoApi.Handlers;
 using ConductorSharp.Patterns.Extensions;
@@ -39,6 +40,11 @@
                     pipelines.AddCustomBehavior<PrepareEmailBehavior, PrepareEmailRequest, PrepareEmailResponse>();
                 })
                 .AddConductorSharpPatterns();
+            //.AddKafkaCancellationNotifier(
+            //    kafkaBootstrapServers: configuration.GetValue<string>(
+            //        "Conductor:KafkaCancellationNotifier:BootstrapServers"),
+            //    topicName: configuration.GetValue<string>("Conductor:KafkaCancellationNotifier:TopicName"),
+            //    groupId: configuration.GetValue<string>("Conductor:KafkaCancellationNotifier:GroupId"));
 
             services.RegisterWorkerTask<GetCustomerHandler>();
             services.RegisterWorkerTask<PrepareEmailHandler>();
diff --git a/src/ConductorSharp.Client/ConductorSharp.Client.csproj b/src/ConductorSharp.Client/ConductorSharp.Client.csproj
@@ -6,7 +6,7 @@
 		<Authors>Codaxy</Authors>
 		<Company>Codaxy</Company>
 		<PackageId>ConductorSharp.Client</PackageId>
-		<Version>3.1.1</Version>
+		<Version>3.2.0</Version>
 		<Description>Client library for Netflix Conductor, with some additional quality of life features.</Description>
 		<RepositoryUrl>https://github.com/codaxy/conductor-sharp</RepositoryUrl>
 		<PackageTags>netflix;conductor</PackageTags>
diff --git a/src/ConductorSharp.Engine/Behaviors/RequestResponseLoggingBehavior.cs b/src/ConductorSharp.Engine/Behaviors/RequestResponseLoggingBehavior.cs
@@ -52,6 +52,16 @@ public async Task<TResponse> Handle(TRequest request, RequestHandlerDelegate<TRe
 
                 return response;
             }
+            catch (TaskCanceledException)
+            {
+                _logger.LogWarning(
+                    $"Request {{Request}} cancelled with payload {{@{requestName}}} and with id {{RequestId}}",
+                    requestName,
+                    request,
+                    requestId
+                );
+                throw;
+            }
             catch (Exception exc)
             {
                 stopwatch.Stop();
diff --git a/src/ConductorSharp.Engine/ConductorSharp.Engine.csproj b/src/ConductorSharp.Engine/ConductorSharp.Engine.csproj
@@ -6,7 +6,7 @@
     <Authors>Codaxy</Authors>
 	<Company>Codaxy</Company>
 	<PackageId>ConductorSharp.Engine</PackageId>
-	<Version>3.1.1</Version>
+	<Version>3.2.0</Version>
     <Description>Client library for Netflix Conductor, with some additional quality of life features.</Description>
 	<RepositoryUrl>https://github.com/codaxy/conductor-sharp</RepositoryUrl>
 	<PackageTags>netflix;conductor</PackageTags>
diff --git a/src/ConductorSharp.Engine/ExecutionManager.cs b/src/ConductorSharp.Engine/ExecutionManager.cs
@@ -3,6 +3,7 @@
 using System.IO;
 using System.Linq;
 using System.Threading;
+using System.Threading.Tasks;
 using ConductorSharp.Client;
 using ConductorSharp.Client.Generated;
 using ConductorSharp.Client.Service;
@@ -30,6 +31,7 @@ internal class ExecutionManager
         private readonly IServiceScopeFactory _lifetimeScopeFactory;
         private readonly IPollTimingStrategy _pollTimingStrategy;
         private readonly IPollOrderStrategy _pollOrderStrategy;
+        private readonly ICancellationNotifier _cancellationNotifier;
 
         public ExecutionManager(
             WorkerSetConfig options,
@@ -39,7 +41,8 @@ public ExecutionManager(
             IExternalPayloadService externalPayloadService,
             IServiceScopeFactory lifetimeScope,
             IPollTimingStrategy pollTimingStrategy,
-            IPollOrderStrategy pollOrderStrategy
+            IPollOrderStrategy pollOrderStrategy,
+            ICancellationNotifier cancellationNotifier
         )
         {
             _configuration = options;
@@ -50,6 +53,7 @@ IPollOrderStrategy pollOrderStrategy
             _lifetimeScopeFactory = lifetimeScope;
             _pollTimingStrategy = pollTimingStrategy;
             _pollOrderStrategy = pollOrderStrategy;
+            _cancellationNotifier = cancellationNotifier;
             _externalPayloadService = externalPayloadService;
         }
 
@@ -110,18 +114,58 @@ private static Type GetInputType(Type workerType)
 
         private async Task PollAndHandle(TaskToWorker scheduledWorker, CancellationToken cancellationToken)
         {
-            Client.Generated.Task pollResponse = null;
+            Client.Generated.Task pollResponse;
+
+            // TODO: Maybe this should be configurable
+            var workerId = Guid.NewGuid().ToString();
             try
             {
-                var workerId = Guid.NewGuid().ToString();
-
                 pollResponse = await _taskManager.PollAsync(
                     scheduledWorker.TaskName,
                     workerId,
                     scheduledWorker.TaskDomain ?? _configuration.Domain,
                     cancellationToken
                 );
+            }
+            catch (ApiException exception) when (exception.StatusCode == 204)
+            {
+                // This handles the case when PollAsync throws exception in case there are no tasks in queue
+                // Even though Conductor reports 1 task in queue for particular task type this endpoint won't return scheduled task immmediately
+                // We skip the further handling as task will be handled in next call to this method
+                return;
+            }
+            catch (Exception exception)
+            {
+                _logger.LogError(exception, "Exception during the task polling");
+                return;
+            }
+
+            try
+            {
+                using var tokenHolder = _cancellationNotifier.GetCancellationToken(pollResponse.TaskId, cancellationToken);
+                await ProcessPolledTask(pollResponse, workerId, scheduledWorker, tokenHolder.CancellationToken);
+            }
+            catch (TaskCanceledException)
+            {
+                _logger.LogWarning(
+                    "Polled task {Task}(id={TaskId}) of workflow {Workflow}(id={WorkflowId}) is cancelled",
+                    pollResponse.TaskDefName,
+                    pollResponse.TaskId,
+                    pollResponse.WorkflowType,
+                    pollResponse.WorkflowInstanceId
+                );
+            }
+        }
 
+        private async Task ProcessPolledTask(
+            Client.Generated.Task pollResponse,
+            string workerId,
+            TaskToWorker scheduledWorker,
+            CancellationToken cancellationToken
+        )
+        {
+            try
+            {
                 if (!string.IsNullOrEmpty(pollResponse.ExternalInputPayloadStoragePath))
                 {
                     _logger.LogDebug("Fetching storage {location}", pollResponse.ExternalInputPayloadStoragePath);
@@ -180,20 +224,19 @@ await _taskManager.UpdateAsync(
                     cancellationToken
                 );
             }
-            catch (ApiException exception) when (exception.StatusCode == 204)
+            catch (TaskCanceledException)
             {
-                // This handles the case when PollAsync throws exception in case there are no tasks in queue
-                // Even though Conductor reports 1 task in queue for particular task type this endpoint won't return scheduled task immmediately
-                // This causes NullReferenceException in logging code below hence why we ignore the exception
+                // Propagate this exception to outer handler
+                throw;
             }
             catch (Exception exception)
             {
                 _logger.LogError(
                     "{@Exception} while executing {Task} as part of {Workflow} with id {WorkflowId}",
                     exception,
-                    pollResponse?.TaskDefName,
-                    pollResponse?.WorkflowType,
-                    pollResponse?.WorkflowInstanceId
+                    pollResponse.TaskDefName,
+                    pollResponse.WorkflowType,
+                    pollResponse.WorkflowInstanceId
                 );
 
                 var errorMessage = new ErrorOutput { ErrorMessage = exception.Message };
@@ -207,16 +250,16 @@ await Task.WhenAll(
                         _taskManager.UpdateAsync(
                             new TaskResult
                             {
-                                TaskId = pollResponse?.TaskId,
+                                TaskId = pollResponse.TaskId,
                                 Status = TaskResultStatus.FAILED,
                                 ReasonForIncompletion = exception.Message,
                                 OutputData = SerializationHelper.ObjectToDictionary(errorMessage, ConductorConstants.IoJsonSerializerSettings),
                                 WorkflowInstanceId = pollResponse?.WorkflowInstanceId
                             },
                             cancellationToken
                         ),
-                        _taskManager.LogAsync(pollResponse?.TaskId, exception.Message, cancellationToken),
-                        _taskManager.LogAsync(pollResponse?.TaskId, exception.StackTrace, cancellationToken)
+                        _taskManager.LogAsync(pollResponse.TaskId, exception.Message, cancellationToken),
+                        _taskManager.LogAsync(pollResponse.TaskId, exception.StackTrace, cancellationToken)
                     ]
                 );
             }
diff --git a/src/ConductorSharp.Engine/Extensions/ConductorSharpBuilder.cs b/src/ConductorSharp.Engine/Extensions/ConductorSharpBuilder.cs
@@ -53,6 +53,8 @@ params Assembly[] handlerAssemblies
 
             Builder.AddTransient<IPollOrderStrategy, RandomOrdering>();
 
+            Builder.AddSingleton<ICancellationNotifier, NoOpCancellationNotifier>();
+
             Builder.AddMediatR(cfg => cfg.RegisterServicesFromAssemblies(handlerAssemblies));
 
             return this;
diff --git a/src/ConductorSharp.Engine/Interface/ICancellationNotifier.cs b/src/ConductorSharp.Engine/Interface/ICancellationNotifier.cs
@@ -0,0 +1,15 @@
+﻿using System;
+using System.Threading;
+
+namespace ConductorSharp.Engine.Interface
+{
+    public interface ICancellationNotifier
+    {
+        public interface ICancellationTokenHolder : IDisposable
+        {
+            CancellationToken CancellationToken { get; }
+        }
+
+        ICancellationTokenHolder GetCancellationToken(string taskId, CancellationToken engineCancellationToken);
+    }
+}
diff --git a/src/ConductorSharp.Engine/Service/NoOpCancellationNotifier.cs b/src/ConductorSharp.Engine/Service/NoOpCancellationNotifier.cs
@@ -0,0 +1,18 @@
+﻿using System.Threading;
+using ConductorSharp.Engine.Interface;
+
+namespace ConductorSharp.Engine.Service;
+
+// By default, only engine shutdown can cancel task execution
+internal class NoOpCancellationNotifier : ICancellationNotifier
+{
+    internal class PassthroughCancellationTokenHolder(CancellationToken cancellationToken) : ICancellationNotifier.ICancellationTokenHolder
+    {
+        public CancellationToken CancellationToken { get; } = cancellationToken;
+
+        public void Dispose() { }
+    }
+
+    public ICancellationNotifier.ICancellationTokenHolder GetCancellationToken(string taskId, CancellationToken engineCancellationToken) =>
+        new PassthroughCancellationTokenHolder(engineCancellationToken);
+}
diff --git a/src/ConductorSharp.KafkaCancellationNotifier/ConductorSharp.KafkaCancellationNotifier.csproj b/src/ConductorSharp.KafkaCancellationNotifier/ConductorSharp.KafkaCancellationNotifier.csproj
@@ -0,0 +1,20 @@
+﻿<Project Sdk="Microsoft.NET.Sdk">
+
+  <PropertyGroup>
+    <TargetFramework>net6.0</TargetFramework>
+    <ImplicitUsings>enable</ImplicitUsings>
+    <Nullable>enable</Nullable>
+    <Version>3.2.0</Version>
+    <Authors>Codaxy</Authors>
+    <Company>Codaxy</Company>
+  </PropertyGroup>
+
+  <ItemGroup>
+    <PackageReference Include="Confluent.Kafka" Version="2.8.0" />
+  </ItemGroup>
+
+  <ItemGroup>
+    <ProjectReference Include="..\ConductorSharp.Engine\ConductorSharp.Engine.csproj" />
+  </ItemGroup>
+
+</Project>
diff --git a/src/ConductorSharp.KafkaCancellationNotifier/Extensions/ExecutionManagerBuilderExtensions.cs b/src/ConductorSharp.KafkaCancellationNotifier/Extensions/ExecutionManagerBuilderExtensions.cs
@@ -0,0 +1,36 @@
+﻿using ConductorSharp.Engine.Extensions;
+using ConductorSharp.Engine.Interface;
+using ConductorSharp.KafkaCancellationNotifier.Service;
+using Microsoft.Extensions.DependencyInjection;
+
+namespace ConductorSharp.KafkaCancellationNotifier.Extensions
+{
+    public static class ExecutionManagerBuilderExtensions
+    {
+        public static IExecutionManagerBuilder AddKafkaCancellationNotifier(
+            this IExecutionManagerBuilder builder,
+            string kafkaBootstrapServers,
+            string topicName,
+            string groupId
+        )
+        {
+            ArgumentNullException.ThrowIfNull(kafkaBootstrapServers);
+            ArgumentNullException.ThrowIfNull(topicName);
+            ArgumentNullException.ThrowIfNull(groupId);
+
+            builder
+                .Builder.AddOptions<KafkaOptions>()
+                .Configure(opts =>
+                {
+                    opts.BootstrapServers = kafkaBootstrapServers;
+                    opts.GroupId = groupId;
+                    opts.TopicName = topicName;
+                });
+
+            builder.Builder.AddSingleton<ICancellationNotifier, Service.KafkaCancellationNotifier>();
+            builder.Builder.AddHostedService<KafkaConsumerBackgroundService>();
+
+            return builder;
+        }
+    }
+}
diff --git a/src/ConductorSharp.KafkaCancellationNotifier/KafkaOptions.cs b/src/ConductorSharp.KafkaCancellationNotifier/KafkaOptions.cs
@@ -0,0 +1,12 @@
+﻿using System.ComponentModel.DataAnnotations;
+using Newtonsoft.Json;
+
+namespace ConductorSharp.KafkaCancellationNotifier
+{
+    internal class KafkaOptions
+    {
+        public string BootstrapServers { get; set; } = null!;
+        public string TopicName { get; set; } = null!;
+        public string GroupId { get; set; } = null!;
+    }
+}
diff --git a/src/ConductorSharp.KafkaCancellationNotifier/Model/TaskStatusModel.cs b/src/ConductorSharp.KafkaCancellationNotifier/Model/TaskStatusModel.cs
@@ -0,0 +1,12 @@
+﻿using ConductorSharp.Client.Generated;
+using TaskStatus = ConductorSharp.Client.Generated.TaskStatus;
+
+namespace ConductorSharp.KafkaCancellationNotifier.Model;
+
+internal class TaskStatusModel
+{
+    public WorkflowTask WorkflowTask { get; set; } = null!;
+    public TaskStatus Status { get; set; }
+    public string TaskId { get; set; } = null!;
+    public string WorkflowInstanceId { get; set; } = null!;
+}
diff --git a/src/ConductorSharp.KafkaCancellationNotifier/Service/KafkaCancellationNotifier.cs b/src/ConductorSharp.KafkaCancellationNotifier/Service/KafkaCancellationNotifier.cs
diff --git a/src/ConductorSharp.KafkaCancellationNotifier/Service/KafkaConsumerBackgroundService.cs b/src/ConductorSharp.KafkaCancellationNotifier/Service/KafkaConsumerBackgroundService.cs
diff --git a/src/ConductorSharp.Patterns/ConductorSharp.Patterns.csproj b/src/ConductorSharp.Patterns/ConductorSharp.Patterns.csproj
