diff --git a/README.md b/README.md index e56a0543..2764a1f8 100644 --- a/README.md +++ b/README.md @@ -799,6 +799,19 @@ public class MyHandler : TaskRequestHandler } ``` +#### Run context (Maestro) + +Maestro can start a workflow run with an execution context: key/value pairs such as the tenant +and user the run is for. It sends the context with every polled task, and the workflow cannot +change it, so read identity from there rather than from the task input: + +```csharp +var workspaceId = _context.Context.TryGetValue("workspaceId", out var id) ? id : null; +``` + +Set it when starting a run with `StartWorkflowRequest.Context`. Conductor sends no context, so +against Conductor `Context` is empty and the field is left out of start requests. + ## Patterns Package Additional built-in tasks and utilities: diff --git a/src/ConductorSharp.Client/ConductorSharp.Client.csproj b/src/ConductorSharp.Client/ConductorSharp.Client.csproj index 1a506640..946018b3 100644 --- a/src/ConductorSharp.Client/ConductorSharp.Client.csproj +++ b/src/ConductorSharp.Client/ConductorSharp.Client.csproj @@ -6,7 +6,7 @@ Codaxy Codaxy ConductorSharp.Client - 4.4.1 + 4.5.0 Client library for Netflix Conductor, with some additional quality of life features. https://github.com/codaxy/conductor-sharp netflix;conductor diff --git a/src/ConductorSharp.Client/Generated/ExecutionContext.cs b/src/ConductorSharp.Client/Generated/ExecutionContext.cs new file mode 100644 index 00000000..03953565 --- /dev/null +++ b/src/ConductorSharp.Client/Generated/ExecutionContext.cs @@ -0,0 +1,40 @@ +using Newtonsoft.Json; + +namespace ConductorSharp.Client.Generated +{ + // The execution context is a Maestro extension of the Conductor API: key/value pairs a workflow + // run is started with, kept unchanged for the life of the run and inherited by its sub-workflows. + // Conductor neither sends nor reads it. These partials live outside Models.cs so regenerating the + // models does not drop them. + + public partial class Task + { + /// + /// Execution context of the workflow run this task belongs to, as Maestro sends it with a + /// polled task. Null from Conductor. + /// + [JsonProperty("context", Required = Required.Default, NullValueHandling = NullValueHandling.Ignore)] + public IDictionary? Context { get; set; } + } + + public partial class Workflow + { + /// + /// Execution context the run was started with, as Maestro reports it on a workflow read. + /// Null from Conductor. + /// + [JsonProperty("context", Required = Required.Default, NullValueHandling = NullValueHandling.Ignore)] + public IDictionary? Context { get; set; } + } + + public partial class StartWorkflowRequest + { + /// + /// Execution context to start the run with. Maestro stores it with the run, hands it to every + /// task and sub-workflow, and never lets the workflow change it. Omitted from the request when + /// null, so a start against Conductor is unchanged. + /// + [JsonProperty("context", Required = Required.Default, NullValueHandling = NullValueHandling.Ignore)] + public IDictionary? Context { get; set; } + } +} diff --git a/src/ConductorSharp.Engine/ConductorSharp.Engine.csproj b/src/ConductorSharp.Engine/ConductorSharp.Engine.csproj index df478cee..d2e48908 100644 --- a/src/ConductorSharp.Engine/ConductorSharp.Engine.csproj +++ b/src/ConductorSharp.Engine/ConductorSharp.Engine.csproj @@ -6,7 +6,7 @@ Codaxy Codaxy ConductorSharp.Engine - 4.4.1 + 4.5.0 Client library for Netflix Conductor, with some additional quality of life features. https://github.com/codaxy/conductor-sharp netflix;conductor diff --git a/src/ConductorSharp.Engine/ExecutionManager.cs b/src/ConductorSharp.Engine/ExecutionManager.cs index 7eb18963..dcf21cf6 100644 --- a/src/ConductorSharp.Engine/ExecutionManager.cs +++ b/src/ConductorSharp.Engine/ExecutionManager.cs @@ -145,6 +145,12 @@ private async Task PollAndHandle(TaskToWorker scheduledWorker, CancellationToken return; } + // A transport other than HTTP may answer "no task" with null instead of a 204, as + // Maestro's in-process transport does. Handling it here keeps that from surfacing as a + // NullReferenceException in a poll nobody awaits. + if (pollResponse == null) + return; + await ProcessPolledTask(pollResponse, workerId, scheduledWorker, cancellationToken); } @@ -200,6 +206,7 @@ CancellationToken cancellationToken context.TaskReferenceName = pollResponse.ReferenceTaskName; context.WorkflowId = pollResponse.WorkflowInstanceId; context.CorrelationId = pollResponse.CorrelationId; + context.Context = ConductorSharpExecutionContext.CopyContext(pollResponse.Context); context.TaskId = pollResponse.TaskId; context.WorkerId = workerId; } diff --git a/src/ConductorSharp.Engine/TypePollSpreadingExecutionManager.cs b/src/ConductorSharp.Engine/TypePollSpreadingExecutionManager.cs index 550bf1ec..968044dd 100644 --- a/src/ConductorSharp.Engine/TypePollSpreadingExecutionManager.cs +++ b/src/ConductorSharp.Engine/TypePollSpreadingExecutionManager.cs @@ -154,6 +154,12 @@ private async Task PollAndHandle(TaskToWorker scheduledWorker, CancellationToken return; } + // A transport other than HTTP may answer "no task" with null instead of a 204, as + // Maestro's in-process transport does. Handling it here keeps that from surfacing as a + // NullReferenceException in a poll nobody awaits. + if (pollResponse == null) + return; + await ProcessPolledTask(pollResponse, workerId, scheduledWorker, cancellationToken); } @@ -209,6 +215,7 @@ CancellationToken cancellationToken context.TaskReferenceName = pollResponse.ReferenceTaskName; context.WorkflowId = pollResponse.WorkflowInstanceId; context.CorrelationId = pollResponse.CorrelationId; + context.Context = ConductorSharpExecutionContext.CopyContext(pollResponse.Context); context.TaskId = pollResponse.TaskId; context.WorkerId = workerId; } diff --git a/src/ConductorSharp.Engine/Util/ConductorSharpExecutionContext.cs b/src/ConductorSharp.Engine/Util/ConductorSharpExecutionContext.cs index 91f30f9b..ddad2062 100644 --- a/src/ConductorSharp.Engine/Util/ConductorSharpExecutionContext.cs +++ b/src/ConductorSharp.Engine/Util/ConductorSharpExecutionContext.cs @@ -1,11 +1,16 @@ using System; using System.Collections.Generic; +using System.Collections.ObjectModel; using System.Text; namespace ConductorSharp.Engine.Util { public class ConductorSharpExecutionContext { + private static readonly IReadOnlyDictionary NoContext = new ReadOnlyDictionary( + new Dictionary() + ); + public string WorkflowName { get; set; } public string WorkflowId { get; set; } public string TaskName { get; set; } @@ -13,5 +18,29 @@ public class ConductorSharpExecutionContext public string TaskReferenceName { get; set; } public string CorrelationId { get; set; } public string WorkerId { get; set; } + + /// + /// The execution context of the workflow run the task belongs to: key/value pairs the run was + /// started with, such as the tenant and user it runs for. Maestro sends it with every polled + /// task. Conductor sends none, and then this is empty. + /// + /// + /// Read identity from here rather than from the task's input. Maestro does not let the + /// workflow change the context after the start, while the input is whatever the workflow + /// definition puts there. + /// + public IReadOnlyDictionary Context { get; set; } = NoContext; + + /// + /// Copies the context from a polled task, so a handler cannot change what the next handler + /// sees by writing to the task's dictionary. + /// + internal static IReadOnlyDictionary CopyContext(IDictionary context) + { + if (context == null || context.Count == 0) + return NoContext; + + return new ReadOnlyDictionary(new Dictionary(context)); + } } } diff --git a/src/ConductorSharp.KafkaCancellationNotifier/ConductorSharp.KafkaCancellationNotifier.csproj b/src/ConductorSharp.KafkaCancellationNotifier/ConductorSharp.KafkaCancellationNotifier.csproj index 989e052e..a7c7a774 100644 --- a/src/ConductorSharp.KafkaCancellationNotifier/ConductorSharp.KafkaCancellationNotifier.csproj +++ b/src/ConductorSharp.KafkaCancellationNotifier/ConductorSharp.KafkaCancellationNotifier.csproj @@ -4,7 +4,7 @@ net6.0 enable enable - 4.4.1 + 4.5.0 Codaxy Codaxy diff --git a/src/ConductorSharp.Patterns/ConductorSharp.Patterns.csproj b/src/ConductorSharp.Patterns/ConductorSharp.Patterns.csproj index 70163e9e..f2273b42 100644 --- a/src/ConductorSharp.Patterns/ConductorSharp.Patterns.csproj +++ b/src/ConductorSharp.Patterns/ConductorSharp.Patterns.csproj @@ -7,7 +7,7 @@ False Codaxy Codaxy - 4.4.1 + 4.5.0 diff --git a/src/ConductorSharp.Toolkit/ConductorSharp.Toolkit.csproj b/src/ConductorSharp.Toolkit/ConductorSharp.Toolkit.csproj index 0b48430b..1d5c5a4e 100644 --- a/src/ConductorSharp.Toolkit/ConductorSharp.Toolkit.csproj +++ b/src/ConductorSharp.Toolkit/ConductorSharp.Toolkit.csproj @@ -7,7 +7,7 @@ disable true dotnet-conductorsharp - 4.4.1 + 4.5.0 diff --git a/test/ConductorSharp.Engine.Tests/Unit/ExecutionContextTests.cs b/test/ConductorSharp.Engine.Tests/Unit/ExecutionContextTests.cs new file mode 100644 index 00000000..3d9cdfcd --- /dev/null +++ b/test/ConductorSharp.Engine.Tests/Unit/ExecutionContextTests.cs @@ -0,0 +1,264 @@ +using ConductorSharp.Client.Generated; +using ConductorSharp.Client.Service; +using ConductorSharp.Client.Util; +using ConductorSharp.Engine.Builders.Metadata; +using ConductorSharp.Engine.Extensions; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.DependencyInjection.Extensions; +using ConductorTask = ConductorSharp.Client.Generated.Task; +using Task = System.Threading.Tasks.Task; + +namespace ConductorSharp.Engine.Tests.Unit +{ + /// + /// Maestro sends a workflow run's execution context with every polled task. A handler reads it + /// from . + /// + public class ExecutionContextTests + { + private const string ProbeTaskName = "CONTEXT_probe"; + + // The settings ConductorSharp's generated HTTP client serializes requests with. + private static readonly JsonSerializerSettings WireJson = new() { Converters = { new JsonElementConverter(), new JsonNodeConverter() } }; + + [Fact] + public void Polled_task_reads_the_context() + { + var task = JsonConvert.DeserializeObject("""{"taskId":"t1","context":{"workspaceId":"42","userId":"7"}}""", WireJson); + + Assert.Equal("42", task.Context["workspaceId"]); + Assert.Equal("7", task.Context["userId"]); + Assert.False(task.AdditionalProperties.ContainsKey("context")); + } + + [Fact] + public void Polled_task_from_conductor_has_no_context() + { + var task = JsonConvert.DeserializeObject("""{"taskId":"t1"}""", WireJson); + + Assert.Null(task.Context); + } + + /// + /// A start against Conductor must not change, so a request with no context leaves the field out. + /// + [Fact] + public void Start_request_without_context_omits_the_field() + { + var json = JsonConvert.SerializeObject(new StartWorkflowRequest { Name = "wf" }, WireJson); + + Assert.DoesNotContain("context", json); + } + + [Fact] + public void Start_request_carries_the_context() + { + var request = new StartWorkflowRequest + { + Name = "wf", + Context = new Dictionary { ["workspaceId"] = "42" }, + }; + + var json = JsonConvert.SerializeObject(request, WireJson); + + Assert.Contains("\"context\":{\"workspaceId\":\"42\"}", json); + } + + [Fact] + public async Task Handler_sees_the_context_of_the_polled_task() + { + var taskService = new ScriptedTaskService( + new ConductorTask + { + TaskId = "task-1", + TaskDefName = ProbeTaskName, + WorkflowInstanceId = "workflow-1", + InputData = new Dictionary(), + Context = new Dictionary { ["workspaceId"] = "42" }, + } + ); + + await RunUntilAsync(taskService, () => taskService.Updates.Count > 0); + + var update = Assert.Single(taskService.Updates); + Assert.Equal(TaskResultStatus.COMPLETED, update.Status); + Assert.Equal("42", update.OutputData["workspace_id"]?.ToString()); + } + + /// + /// A transport may answer an empty poll with null rather than a 204. The manager skips it and + /// keeps polling. + /// + /// + /// This pins the behaviour, but it cannot catch its loss. Without the null check the poll + /// throws inside a task the manager never awaits, so the failure is silent and the manager + /// keeps polling either way. + /// + [Fact] + public async Task Null_poll_response_is_skipped() + { + var taskService = new ScriptedTaskService(null); + + await RunUntilAsync(taskService, () => taskService.Polls >= 2); + + Assert.Empty(taskService.Updates); + } + + private static async Task RunUntilAsync(ScriptedTaskService taskService, Func condition) + { + var services = new ServiceCollection(); + services.AddLogging(); + services + .AddConductorSharp(baseUrl: "http://empty/empty") + .AddExecutionManager( + maxConcurrentWorkers: 1, + sleepInterval: 250, + longPollInterval: 500, + domain: null, + handlerAssemblies: typeof(ExecutionContextTests).Assembly + ); + services.RegisterWorkerTask(); + services.RemoveAll(); + services.AddSingleton(taskService); + + await using var provider = services.BuildServiceProvider(); + var manager = provider.GetRequiredService(); + + using var cancellation = new CancellationTokenSource(TimeSpan.FromSeconds(20)); + var run = manager.StartAsync(cancellation.Token); + + while (!condition()) + { + Assert.False(cancellation.IsCancellationRequested, "The execution manager did not get there in time"); + await Task.Delay(25); + } + + cancellation.Cancel(); + await Assert.ThrowsAnyAsync(() => run); + } + + public class ContextProbeRequest : IRequest { } + + public class ContextProbeResponse + { + public string WorkspaceId { get; set; } + } + + [OriginalName(ProbeTaskName)] + public class ContextProbeHandler(ConductorSharpExecutionContext context) : ITaskRequestHandler + { + public Task Handle(ContextProbeRequest request, CancellationToken cancellationToken) + { + context.Context.TryGetValue("workspaceId", out var workspaceId); + + return Task.FromResult(new ContextProbeResponse { WorkspaceId = workspaceId }); + } + } + + /// + /// Reports one queued probe task and answers every poll with the same scripted response. + /// + private sealed class ScriptedTaskService(ConductorTask pollResponse) : ITaskService + { + private int _polls; + + public List Updates { get; } = []; + + public int Polls => _polls; + + public Task> ListQueuesAsync(CancellationToken cancellationToken = default) => + Task.FromResult>(new Dictionary { [ProbeTaskName] = 1 }); + + public Task PollAsync(string taskType, string workerId, string domain, CancellationToken cancellationToken = default) + { + Interlocked.Increment(ref _polls); + + // Handed out once, as a real queue would. + var response = Updates.Count == 0 ? pollResponse : null; + + return Task.FromResult(response); + } + + public Task UpdateAsync(TaskResult updateRequest, CancellationToken cancellationToken = default) + { + lock (Updates) + { + Updates.Add(updateRequest); + } + + return Task.FromResult(updateRequest.TaskId); + } + + public Task GetAsync(string taskId, CancellationToken cancellationToken = default) => throw new NotImplementedException(); + + public Task GetQueueAsync( + string taskType, + string domain = null, + string isolationGroupId = null, + string executionNamespace = null, + CancellationToken cancellationToken = default + ) => throw new NotImplementedException(); + + public Task GetExternalStorageLocationAsync( + string path, + string operation, + string payloadType, + CancellationToken cancellationToken = default + ) => throw new NotImplementedException(); + + public Task> ListLogsAsync(string taskId, CancellationToken cancellationToken = default) => + throw new NotImplementedException(); + + public Task> GetPollDataAsync(string taskType, CancellationToken cancellationToken = default) => + throw new NotImplementedException(); + + public Task> ListPollDataAsync(CancellationToken cancellationToken = default) => + throw new NotImplementedException(); + + public Task>>> ListQueuesVerboseAsync( + CancellationToken cancellationToken = default + ) => throw new NotImplementedException(); + + public Task LogAsync(string taskId, string message, CancellationToken cancellationToken = default) => Task.CompletedTask; + + public Task> BatchPollAsync( + string taskType, + string workerId, + string domain, + int? count = null, + int? timeout = null, + CancellationToken cancellationToken = default + ) => throw new NotImplementedException(); + + public Task RequeuePendingAsync(string taskType, CancellationToken cancellationToken = default) => + throw new NotImplementedException(); + + public Task SearchAsync( + int? start = null, + int? size = null, + string sort = null, + string freeText = null, + string query = null, + CancellationToken cancellationToken = default + ) => throw new NotImplementedException(); + + public Task SearchV2Async( + int? start = null, + int? size = null, + string sort = null, + string freeText = null, + string query = null, + CancellationToken cancellationToken = default + ) => throw new NotImplementedException(); + + public Task UpdateTaskByRefnameAsync( + string workflowId, + string taskRefName, + Status status, + string workerid, + IDictionary body, + CancellationToken cancellationToken = default + ) => throw new NotImplementedException(); + } + } +}