From e82c1db5100c5f2aee9332ce127adde3064e43fc Mon Sep 17 00:00:00 2001 From: "bojan.malinic" Date: Fri, 17 Jul 2026 12:53:31 +0200 Subject: [PATCH 1/4] Apply fix from master branch --- src/ConductorSharp.Engine/AssemblyInfo.cs | 3 + src/ConductorSharp.Engine/ExecutionManager.cs | 8 +- .../Extensions/ConductorSharpBuilder.cs | 8 +- .../Service/TaskQueuePollingService.cs | 97 +++++++++++++++++++ .../TypePollSpreadingExecutionManager.cs | 8 +- .../Unit/TaskQueuePollingServiceTests.cs | 97 +++++++++++++++++++ test/ConductorSharp.Engine.Tests/Usings.cs | 1 + 7 files changed, 215 insertions(+), 7 deletions(-) create mode 100644 src/ConductorSharp.Engine/AssemblyInfo.cs create mode 100644 src/ConductorSharp.Engine/Service/TaskQueuePollingService.cs create mode 100644 test/ConductorSharp.Engine.Tests/Unit/TaskQueuePollingServiceTests.cs diff --git a/src/ConductorSharp.Engine/AssemblyInfo.cs b/src/ConductorSharp.Engine/AssemblyInfo.cs new file mode 100644 index 00000000..f7591f45 --- /dev/null +++ b/src/ConductorSharp.Engine/AssemblyInfo.cs @@ -0,0 +1,3 @@ +using System.Runtime.CompilerServices; + +[assembly: InternalsVisibleTo("ConductorSharp.Engine.Tests")] diff --git a/src/ConductorSharp.Engine/ExecutionManager.cs b/src/ConductorSharp.Engine/ExecutionManager.cs index 259b4aba..5288c5b4 100644 --- a/src/ConductorSharp.Engine/ExecutionManager.cs +++ b/src/ConductorSharp.Engine/ExecutionManager.cs @@ -11,6 +11,7 @@ using ConductorSharp.Engine.Interface; using ConductorSharp.Engine.Model; using ConductorSharp.Engine.Polling; +using ConductorSharp.Engine.Service; using ConductorSharp.Engine.Util; using MediatR; using Microsoft.Extensions.DependencyInjection; @@ -32,6 +33,7 @@ internal class ExecutionManager : IExecutionManager private readonly IPollTimingStrategy _pollTimingStrategy; private readonly IPollOrderStrategy _pollOrderStrategy; private readonly ICancellationNotifier _cancellationNotifier; + private readonly TaskQueuePollingService _taskQueuePollingService; public ExecutionManager( WorkerSetConfig options, @@ -42,7 +44,8 @@ public ExecutionManager( IServiceScopeFactory lifetimeScope, IPollTimingStrategy pollTimingStrategy, IPollOrderStrategy pollOrderStrategy, - ICancellationNotifier cancellationNotifier + ICancellationNotifier cancellationNotifier, + TaskQueuePollingService taskQueuePollingService ) { _configuration = options; @@ -55,6 +58,7 @@ ICancellationNotifier cancellationNotifier _pollOrderStrategy = pollOrderStrategy; _cancellationNotifier = cancellationNotifier; _externalPayloadService = externalPayloadService; + _taskQueuePollingService = taskQueuePollingService; } public async Task StartAsync(CancellationToken cancellationToken) @@ -63,7 +67,7 @@ public async Task StartAsync(CancellationToken cancellationToken) while (!cancellationToken.IsCancellationRequested) { - var queuedTasks = (await _taskManager.ListQueuesAsync(cancellationToken)) + var queuedTasks = (await _taskQueuePollingService.ListQueuesAsync(cancellationToken)) .Where(a => a.Value > 0) .ToDictionary(a => a.Key, a => a.Value); diff --git a/src/ConductorSharp.Engine/Extensions/ConductorSharpBuilder.cs b/src/ConductorSharp.Engine/Extensions/ConductorSharpBuilder.cs index 5c59e1e7..912acb50 100644 --- a/src/ConductorSharp.Engine/Extensions/ConductorSharpBuilder.cs +++ b/src/ConductorSharp.Engine/Extensions/ConductorSharpBuilder.cs @@ -43,7 +43,9 @@ params Assembly[] handlerAssemblies Builder.AddTransient(); - Builder.AddSingleton(); + Builder.AddSingleton(); + + Builder.AddSingleton(); Builder.AddScoped(); @@ -62,10 +64,10 @@ params Assembly[] handlerAssemblies public IExecutionManagerBuilder UseBetaExecutionManager() { - Builder.AddSingleton(); + Builder.AddSingleton(); return this; } - + public IExecutionManagerBuilder AddPipelines(Action behaviorBuilder) { var pipelineBuilder = new PipelineBuilder(Builder); diff --git a/src/ConductorSharp.Engine/Service/TaskQueuePollingService.cs b/src/ConductorSharp.Engine/Service/TaskQueuePollingService.cs new file mode 100644 index 00000000..f293217c --- /dev/null +++ b/src/ConductorSharp.Engine/Service/TaskQueuePollingService.cs @@ -0,0 +1,97 @@ +using System; +using System.Collections.Generic; +using System.Net.Http; +using System.Threading; +using System.Threading.Tasks; +using ConductorSharp.Client.Generated; +using ConductorSharp.Client.Service; +using Microsoft.Extensions.Logging; +using Task = System.Threading.Tasks.Task; + +namespace ConductorSharp.Engine.Service +{ + internal class TaskQueuePollingService + { + private const int DefaultMaxAttempts = 5; + private static readonly TimeSpan DefaultInitialRetryDelay = TimeSpan.FromSeconds(1); + private static readonly TimeSpan DefaultMaxRetryDelay = TimeSpan.FromSeconds(30); + private static readonly TimeSpan DefaultMaxJitter = TimeSpan.FromMilliseconds(500); + + private readonly ITaskService _taskService; + private readonly ILogger _logger; + private readonly int _maxAttempts; + private readonly TimeSpan _initialRetryDelay; + private readonly TimeSpan _maxRetryDelay; + private readonly TimeSpan _maxJitter; + private readonly Func _delay; + private readonly Func _jitter; + + public TaskQueuePollingService(ITaskService taskService, ILogger logger) + : this( + taskService, + logger, + DefaultMaxAttempts, + DefaultInitialRetryDelay, + DefaultMaxRetryDelay, + DefaultMaxJitter, + Task.Delay, + Random.Shared.NextDouble + ) { } + + internal TaskQueuePollingService( + ITaskService taskService, + ILogger logger, + int maxAttempts, + TimeSpan initialRetryDelay, + TimeSpan maxRetryDelay, + TimeSpan maxJitter, + Func delay, + Func jitter + ) + { + _taskService = taskService; + _logger = logger; + _maxAttempts = maxAttempts; + _initialRetryDelay = initialRetryDelay; + _maxRetryDelay = maxRetryDelay; + _maxJitter = maxJitter; + _delay = delay; + _jitter = jitter; + } + + public async Task> ListQueuesAsync(CancellationToken cancellationToken) + { + var retryDelay = _initialRetryDelay; + + for (var attempt = 1; ; attempt++) + { + try + { + return await _taskService.ListQueuesAsync(cancellationToken); + } + catch (Exception exception) when (!cancellationToken.IsCancellationRequested && IsTransient(exception) && attempt < _maxAttempts) + { + var delay = retryDelay + TimeSpan.FromMilliseconds(_jitter() * _maxJitter.TotalMilliseconds); + + _logger.LogWarning( + exception, + "Failed to read Conductor task queues. Attempt {Attempt}/{MaxAttempts}; retrying in {RetryDelay}", + attempt, + _maxAttempts, + delay + ); + + await _delay(delay, cancellationToken); + retryDelay = TimeSpan.FromMilliseconds(Math.Min(retryDelay.TotalMilliseconds * 2, _maxRetryDelay.TotalMilliseconds)); + } + } + } + + private static bool IsTransient(Exception exception) + { + return exception is HttpRequestException + || exception is TaskCanceledException + || exception is ApiException apiException && (apiException.StatusCode is 408 or 429 || apiException.StatusCode >= 500); + } + } +} diff --git a/src/ConductorSharp.Engine/TypePollSpreadingExecutionManager.cs b/src/ConductorSharp.Engine/TypePollSpreadingExecutionManager.cs index fc12ac34..9c18eb52 100644 --- a/src/ConductorSharp.Engine/TypePollSpreadingExecutionManager.cs +++ b/src/ConductorSharp.Engine/TypePollSpreadingExecutionManager.cs @@ -11,6 +11,7 @@ using ConductorSharp.Engine.Interface; using ConductorSharp.Engine.Model; using ConductorSharp.Engine.Polling; +using ConductorSharp.Engine.Service; using ConductorSharp.Engine.Util; using MediatR; using Microsoft.Extensions.DependencyInjection; @@ -32,6 +33,7 @@ internal class TypePollSpreadingExecutionManager : IExecutionManager private readonly IPollTimingStrategy _pollTimingStrategy; private readonly IPollOrderStrategy _pollOrderStrategy; private readonly ICancellationNotifier _cancellationNotifier; + private readonly TaskQueuePollingService _taskQueuePollingService; public TypePollSpreadingExecutionManager( WorkerSetConfig options, @@ -42,7 +44,8 @@ public TypePollSpreadingExecutionManager( IServiceScopeFactory lifetimeScope, IPollTimingStrategy pollTimingStrategy, IPollOrderStrategy pollOrderStrategy, - ICancellationNotifier cancellationNotifier + ICancellationNotifier cancellationNotifier, + TaskQueuePollingService taskQueuePollingService ) { _configuration = options; @@ -55,6 +58,7 @@ ICancellationNotifier cancellationNotifier _pollOrderStrategy = pollOrderStrategy; _cancellationNotifier = cancellationNotifier; _externalPayloadService = externalPayloadService; + _taskQueuePollingService = taskQueuePollingService; } public async Task StartAsync(CancellationToken cancellationToken) @@ -63,7 +67,7 @@ public async Task StartAsync(CancellationToken cancellationToken) while (!cancellationToken.IsCancellationRequested) { - var queuedTasks = (await _taskManager.ListQueuesAsync(cancellationToken)) + var queuedTasks = (await _taskQueuePollingService.ListQueuesAsync(cancellationToken)) .Where(a => a.Value > 0) .ToDictionary(a => a.Key, a => a.Value); diff --git a/test/ConductorSharp.Engine.Tests/Unit/TaskQueuePollingServiceTests.cs b/test/ConductorSharp.Engine.Tests/Unit/TaskQueuePollingServiceTests.cs new file mode 100644 index 00000000..38db4d24 --- /dev/null +++ b/test/ConductorSharp.Engine.Tests/Unit/TaskQueuePollingServiceTests.cs @@ -0,0 +1,97 @@ +using System.Net; +using System.Text; +using ConductorSharp.Client.Generated; +using ConductorSharp.Client.Service; +using ConductorSharp.Engine.Service; +using Microsoft.Extensions.Logging.Abstractions; +using Task = System.Threading.Tasks.Task; + +namespace ConductorSharp.Engine.Tests.Unit; + +public class TaskQueuePollingServiceTests +{ + [Fact] + public async Task ListQueuesAsync_RetriesTransientFailuresAndReturnsQueues() + { + var handler = new SequenceHandler(HttpStatusCode.InternalServerError, HttpStatusCode.ServiceUnavailable, HttpStatusCode.OK); + var delays = new List(); + var service = CreateService(handler, delays); + + var queues = await service.ListQueuesAsync(CancellationToken.None); + + Assert.Equal(3, handler.RequestCount); + Assert.Equal(2, delays.Count); + Assert.Equal(2, queues["test-task"]); + } + + [Fact] + public async Task ListQueuesAsync_DoesNotRetryNonTransientApiErrors() + { + var handler = new SequenceHandler(HttpStatusCode.BadRequest, HttpStatusCode.OK); + var delays = new List(); + var service = CreateService(handler, delays); + + var exception = await Assert.ThrowsAsync(() => service.ListQueuesAsync(CancellationToken.None)); + + Assert.Equal(400, exception.StatusCode); + Assert.Equal(1, handler.RequestCount); + Assert.Empty(delays); + } + + [Fact] + public async Task ListQueuesAsync_RethrowsAfterRetryLimit() + { + var handler = new SequenceHandler( + HttpStatusCode.InternalServerError, + HttpStatusCode.InternalServerError, + HttpStatusCode.InternalServerError, + HttpStatusCode.InternalServerError, + HttpStatusCode.InternalServerError + ); + var delays = new List(); + var service = CreateService(handler, delays); + + var exception = await Assert.ThrowsAsync(() => service.ListQueuesAsync(CancellationToken.None)); + + Assert.Equal(500, exception.StatusCode); + Assert.Equal(5, handler.RequestCount); + Assert.Equal(4, delays.Count); + } + + private static TaskQueuePollingService CreateService(HttpMessageHandler handler, ICollection delays) + { + var httpClient = new HttpClient(handler) { BaseAddress = new Uri("http://conductor/") }; + var taskService = new TaskService(httpClient); + + return new TaskQueuePollingService( + taskService, + NullLogger.Instance, + maxAttempts: 5, + initialRetryDelay: TimeSpan.FromSeconds(1), + maxRetryDelay: TimeSpan.FromSeconds(30), + maxJitter: TimeSpan.FromMilliseconds(500), + delay: (delay, _) => + { + delays.Add(delay); + return Task.CompletedTask; + }, + jitter: () => 0 + ); + } + + private sealed class SequenceHandler(params HttpStatusCode[] statuses) : HttpMessageHandler + { + private readonly Queue _statuses = new(statuses); + + public int RequestCount { get; private set; } + + protected override Task SendAsync(HttpRequestMessage request, CancellationToken cancellationToken) + { + RequestCount++; + var status = _statuses.Dequeue(); + var content = status == HttpStatusCode.OK ? """{"test-task":2}""" : "{}"; + + return Task.FromResult(new HttpResponseMessage(status) { Content = new StringContent(content, Encoding.UTF8, "application/json"), }); + } + } +} diff --git a/test/ConductorSharp.Engine.Tests/Usings.cs b/test/ConductorSharp.Engine.Tests/Usings.cs index eb6fe6a9..12d909d5 100644 --- a/test/ConductorSharp.Engine.Tests/Usings.cs +++ b/test/ConductorSharp.Engine.Tests/Usings.cs @@ -6,3 +6,4 @@ global using MediatR; global using Newtonsoft.Json; global using Xunit; +global using EmbeddedFileHelper = ConductorSharp.Engine.Tests.Util.EmbeddedFileHelper; From 3b755b33adfa900829c2ae393fcad771e3e7d625 Mon Sep 17 00:00:00 2001 From: "bojan.malinic" Date: Fri, 17 Jul 2026 12:55:12 +0200 Subject: [PATCH 2/4] Bump version --- src/ConductorSharp.Client/ConductorSharp.Client.csproj | 2 +- src/ConductorSharp.Engine/ConductorSharp.Engine.csproj | 2 +- .../ConductorSharp.KafkaCancellationNotifier.csproj | 2 +- src/ConductorSharp.Patterns/ConductorSharp.Patterns.csproj | 2 +- 4 files changed, 4 insertions(+), 4 deletions(-) diff --git a/src/ConductorSharp.Client/ConductorSharp.Client.csproj b/src/ConductorSharp.Client/ConductorSharp.Client.csproj index 89a783bd..797a4b47 100644 --- a/src/ConductorSharp.Client/ConductorSharp.Client.csproj +++ b/src/ConductorSharp.Client/ConductorSharp.Client.csproj @@ -6,7 +6,7 @@ Codaxy Codaxy ConductorSharp.Client - 3.7.2 + 3.7.3-alpha.1 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/ConductorSharp.Engine.csproj b/src/ConductorSharp.Engine/ConductorSharp.Engine.csproj index ef306da3..2c589416 100644 --- a/src/ConductorSharp.Engine/ConductorSharp.Engine.csproj +++ b/src/ConductorSharp.Engine/ConductorSharp.Engine.csproj @@ -6,7 +6,7 @@ Codaxy Codaxy ConductorSharp.Engine - 3.7.2 + 3.7.3-alpha.1 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.KafkaCancellationNotifier/ConductorSharp.KafkaCancellationNotifier.csproj b/src/ConductorSharp.KafkaCancellationNotifier/ConductorSharp.KafkaCancellationNotifier.csproj index 0cce80b7..242162a1 100644 --- a/src/ConductorSharp.KafkaCancellationNotifier/ConductorSharp.KafkaCancellationNotifier.csproj +++ b/src/ConductorSharp.KafkaCancellationNotifier/ConductorSharp.KafkaCancellationNotifier.csproj @@ -4,7 +4,7 @@ net6.0 enable enable - 3.7.2 + 3.7.3-alpha.1 Codaxy Codaxy diff --git a/src/ConductorSharp.Patterns/ConductorSharp.Patterns.csproj b/src/ConductorSharp.Patterns/ConductorSharp.Patterns.csproj index 2d92ff76..622fc52b 100644 --- a/src/ConductorSharp.Patterns/ConductorSharp.Patterns.csproj +++ b/src/ConductorSharp.Patterns/ConductorSharp.Patterns.csproj @@ -7,7 +7,7 @@ False Codaxy Codaxy - 3.7.2 + 3.7.3-alpha.1 From ba990e296b307acee5309b628b46243c70aeb3ee Mon Sep 17 00:00:00 2001 From: "bojan.malinic" Date: Fri, 17 Jul 2026 14:50:50 +0200 Subject: [PATCH 3/4] Bump to stable version --- src/ConductorSharp.Client/ConductorSharp.Client.csproj | 2 +- src/ConductorSharp.Engine/ConductorSharp.Engine.csproj | 2 +- .../ConductorSharp.KafkaCancellationNotifier.csproj | 2 +- src/ConductorSharp.Patterns/ConductorSharp.Patterns.csproj | 2 +- 4 files changed, 4 insertions(+), 4 deletions(-) diff --git a/src/ConductorSharp.Client/ConductorSharp.Client.csproj b/src/ConductorSharp.Client/ConductorSharp.Client.csproj index 797a4b47..9419daea 100644 --- a/src/ConductorSharp.Client/ConductorSharp.Client.csproj +++ b/src/ConductorSharp.Client/ConductorSharp.Client.csproj @@ -6,7 +6,7 @@ Codaxy Codaxy ConductorSharp.Client - 3.7.3-alpha.1 + 3.7.3 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/ConductorSharp.Engine.csproj b/src/ConductorSharp.Engine/ConductorSharp.Engine.csproj index 2c589416..410fee8b 100644 --- a/src/ConductorSharp.Engine/ConductorSharp.Engine.csproj +++ b/src/ConductorSharp.Engine/ConductorSharp.Engine.csproj @@ -6,7 +6,7 @@ Codaxy Codaxy ConductorSharp.Engine - 3.7.3-alpha.1 + 3.7.3 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.KafkaCancellationNotifier/ConductorSharp.KafkaCancellationNotifier.csproj b/src/ConductorSharp.KafkaCancellationNotifier/ConductorSharp.KafkaCancellationNotifier.csproj index 242162a1..4da07891 100644 --- a/src/ConductorSharp.KafkaCancellationNotifier/ConductorSharp.KafkaCancellationNotifier.csproj +++ b/src/ConductorSharp.KafkaCancellationNotifier/ConductorSharp.KafkaCancellationNotifier.csproj @@ -4,7 +4,7 @@ net6.0 enable enable - 3.7.3-alpha.1 + 3.7.3 Codaxy Codaxy diff --git a/src/ConductorSharp.Patterns/ConductorSharp.Patterns.csproj b/src/ConductorSharp.Patterns/ConductorSharp.Patterns.csproj index 622fc52b..268fe0e6 100644 --- a/src/ConductorSharp.Patterns/ConductorSharp.Patterns.csproj +++ b/src/ConductorSharp.Patterns/ConductorSharp.Patterns.csproj @@ -7,7 +7,7 @@ False Codaxy Codaxy - 3.7.3-alpha.1 + 3.7.3 From e61f2b8d472228130d51c056d9ef4a9e454e5c11 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Andrej=20=C5=A0imi=C4=87?= Date: Tue, 21 Jul 2026 08:51:49 +0200 Subject: [PATCH 4/4] CxODEV-1730: StructuredErrorException + structured_error task-output contract (#216) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * CxODEV-1730: add StructuredErrorException + structured_error task-output contract Adds a framework capability for workers to attach a sanitized, structured error classification to a failed task's output: - StructuredErrorException(code, reason, referenceError?) - StructuredError DTO and ErrorOutput.StructuredError (omitted when null → back-compat) - StructuredErrorSerializer: OutputKey, ToOutputData (for signal senders) and tolerant TryParse (consumer side); a round-trip contract test pins the shape/key - both ExecutionManager and TypePollSpreadingExecutionManager populate structured_error when a StructuredErrorException is caught; plain exceptions still emit only error_message - version bumped 3.7.3 -> 3.8.0 (additive, on the v3 line) --------- Co-authored-by: Claude Opus 4.8 --- .../ConductorSharp.Client.csproj | 2 +- .../ConductorSharp.Engine.csproj | 2 +- .../Exceptions/StructuredErrorException.cs | 40 ++++++ src/ConductorSharp.Engine/ExecutionManager.cs | 13 ++ .../Model/ErrorOutput.cs | 7 + .../Model/StructuredError.cs | 24 ++++ .../TypePollSpreadingExecutionManager.cs | 13 ++ .../Util/StructuredErrorSerializer.cs | 75 +++++++++++ ...ctorSharp.KafkaCancellationNotifier.csproj | 2 +- .../ConductorSharp.Patterns.csproj | 2 +- .../Unit/StructuredErrorTests.cs | 127 ++++++++++++++++++ 11 files changed, 303 insertions(+), 4 deletions(-) create mode 100644 src/ConductorSharp.Engine/Exceptions/StructuredErrorException.cs create mode 100644 src/ConductorSharp.Engine/Model/StructuredError.cs create mode 100644 src/ConductorSharp.Engine/Util/StructuredErrorSerializer.cs create mode 100644 test/ConductorSharp.Engine.Tests/Unit/StructuredErrorTests.cs diff --git a/src/ConductorSharp.Client/ConductorSharp.Client.csproj b/src/ConductorSharp.Client/ConductorSharp.Client.csproj index 9419daea..7b8d5de8 100644 --- a/src/ConductorSharp.Client/ConductorSharp.Client.csproj +++ b/src/ConductorSharp.Client/ConductorSharp.Client.csproj @@ -6,7 +6,7 @@ Codaxy Codaxy ConductorSharp.Client - 3.7.3 + 3.8.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/ConductorSharp.Engine.csproj b/src/ConductorSharp.Engine/ConductorSharp.Engine.csproj index 410fee8b..d566f860 100644 --- a/src/ConductorSharp.Engine/ConductorSharp.Engine.csproj +++ b/src/ConductorSharp.Engine/ConductorSharp.Engine.csproj @@ -6,7 +6,7 @@ Codaxy Codaxy ConductorSharp.Engine - 3.7.3 + 3.8.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/Exceptions/StructuredErrorException.cs b/src/ConductorSharp.Engine/Exceptions/StructuredErrorException.cs new file mode 100644 index 00000000..86d1dd92 --- /dev/null +++ b/src/ConductorSharp.Engine/Exceptions/StructuredErrorException.cs @@ -0,0 +1,40 @@ +using System; + +namespace ConductorSharp.Engine.Exceptions +{ + /// + /// Thrown by a worker to attach a structured, sanitized error classification to the failed task's output. + /// When caught by the execution manager, the // + /// are serialized under the structured_error output key (see + /// ), in addition to the plain + /// error_message, so downstream consumers can read a stable classification without parsing free-text + /// reasons. Plain exceptions are unaffected and keep producing only error_message. + /// + public class StructuredErrorException : Exception + { + /// Stable, opaque classification code. Consumers map this to a failure response. + public string Code { get; } + + /// Human-readable, sanitized reason. Safe to surface across a layer boundary. + public string Reason { get; } + + /// Optional URI pointing at the entity where the failure originated (drill-down link). + public string ReferenceError { get; } + + public StructuredErrorException(string code, string reason, string referenceError = null) + : base(reason) + { + Code = code; + Reason = reason; + ReferenceError = referenceError; + } + + public StructuredErrorException(string code, string reason, string referenceError, Exception innerException) + : base(reason, innerException) + { + Code = code; + Reason = reason; + ReferenceError = referenceError; + } + } +} diff --git a/src/ConductorSharp.Engine/ExecutionManager.cs b/src/ConductorSharp.Engine/ExecutionManager.cs index 5288c5b4..2ef451cf 100644 --- a/src/ConductorSharp.Engine/ExecutionManager.cs +++ b/src/ConductorSharp.Engine/ExecutionManager.cs @@ -8,6 +8,7 @@ using ConductorSharp.Client.Generated; using ConductorSharp.Client.Service; using ConductorSharp.Client.Util; +using ConductorSharp.Engine.Exceptions; using ConductorSharp.Engine.Interface; using ConductorSharp.Engine.Model; using ConductorSharp.Engine.Polling; @@ -248,6 +249,18 @@ await _taskManager.UpdateAsync( var errorMessage = new ErrorOutput { ErrorMessage = exception.Message }; + // A worker may throw a StructuredErrorException to attach a sanitized, stable classification to the + // failed task's output. Plain exceptions keep producing only error_message, preserving backward compatibility. + if (exception is StructuredErrorException structuredException) + { + errorMessage.StructuredError = new StructuredError + { + Code = structuredException.Code, + Reason = structuredException.Reason, + ReferenceError = structuredException.ReferenceError + }; + } + // TODO: We should verify that this is alright, it is possible that when executed concurrently, // the updates caused by LogAsync will be discarded because the call of UpdateAsync(TaskResult...) // sets the logs to null. Not sure how this is implemented in the backend, also, would have expected this to be a diff --git a/src/ConductorSharp.Engine/Model/ErrorOutput.cs b/src/ConductorSharp.Engine/Model/ErrorOutput.cs index add9e806..9fb4096e 100644 --- a/src/ConductorSharp.Engine/Model/ErrorOutput.cs +++ b/src/ConductorSharp.Engine/Model/ErrorOutput.cs @@ -7,5 +7,12 @@ namespace ConductorSharp.Engine.Model public class ErrorOutput { public string ErrorMessage { get; set; } + + /// + /// Optional structured error classification. Null for plain (unclassified) failures, in which case it is + /// omitted from serialized output (NullValueHandling.Ignore), preserving backward compatibility with + /// consumers that only read . + /// + public StructuredError StructuredError { get; set; } } } diff --git a/src/ConductorSharp.Engine/Model/StructuredError.cs b/src/ConductorSharp.Engine/Model/StructuredError.cs new file mode 100644 index 00000000..1457b0ee --- /dev/null +++ b/src/ConductorSharp.Engine/Model/StructuredError.cs @@ -0,0 +1,24 @@ +namespace ConductorSharp.Engine.Model +{ + /// + /// Sanitized, structured error classification transported across the structured_error task-output key. + /// The field shape is versioned via so it can evolve without silent misparses. + /// + public class StructuredError + { + /// Current structured-error payload shape version. + public const int CurrentVersion = 1; + + /// Stable, opaque classification code (e.g. an implementation-defined code, or UNCLASSIFIED). + public string Code { get; set; } + + /// Human-readable, sanitized reason. + public string Reason { get; set; } + + /// Optional URI pointing at the entity where the failure originated (drill-down link). + public string ReferenceError { get; set; } + + /// Payload shape version marker. Defaults to . + public int Version { get; set; } = CurrentVersion; + } +} diff --git a/src/ConductorSharp.Engine/TypePollSpreadingExecutionManager.cs b/src/ConductorSharp.Engine/TypePollSpreadingExecutionManager.cs index 9c18eb52..6fcab408 100644 --- a/src/ConductorSharp.Engine/TypePollSpreadingExecutionManager.cs +++ b/src/ConductorSharp.Engine/TypePollSpreadingExecutionManager.cs @@ -8,6 +8,7 @@ using ConductorSharp.Client.Generated; using ConductorSharp.Client.Service; using ConductorSharp.Client.Util; +using ConductorSharp.Engine.Exceptions; using ConductorSharp.Engine.Interface; using ConductorSharp.Engine.Model; using ConductorSharp.Engine.Polling; @@ -257,6 +258,18 @@ await _taskManager.UpdateAsync( var errorMessage = new ErrorOutput { ErrorMessage = exception.Message }; + // A worker may throw a StructuredErrorException to attach a sanitized, stable classification to the + // failed task's output. Plain exceptions keep producing only error_message, preserving backward compatibility. + if (exception is StructuredErrorException structuredException) + { + errorMessage.StructuredError = new StructuredError + { + Code = structuredException.Code, + Reason = structuredException.Reason, + ReferenceError = structuredException.ReferenceError + }; + } + // TODO: We should verify that this is alright, it is possible that when executed concurrently, // the updates caused by LogAsync will be discarded because the call of UpdateAsync(TaskResult...) // sets the logs to null. Not sure how this is implemented in the backend, also, would have expected this to be a diff --git a/src/ConductorSharp.Engine/Util/StructuredErrorSerializer.cs b/src/ConductorSharp.Engine/Util/StructuredErrorSerializer.cs new file mode 100644 index 00000000..c640c957 --- /dev/null +++ b/src/ConductorSharp.Engine/Util/StructuredErrorSerializer.cs @@ -0,0 +1,75 @@ +using System.Collections.Generic; +using ConductorSharp.Client; +using ConductorSharp.Engine.Model; +using Newtonsoft.Json; + +namespace ConductorSharp.Engine.Util +{ + /// + /// Defines the structured_error task-output contract (key + shape) and the read side used by all consumers + /// (). There are two producers, both emitting the same shape because they serialize the same + /// type with the same serializer settings: + /// + /// the execution-manager catch block, which serializes when a + /// worker throws a ; and + /// external signal senders (which have no exception to catch), which render via . + /// + /// + public static class StructuredErrorSerializer + { + /// Well-known task-output key carrying the structured error payload. + public const string OutputKey = "structured_error"; + + /// + /// Renders a to an output-data fragment ({ "structured_error": { ... } }) + /// using the standard snake_case IO serializer settings. Returns an empty dictionary for a null error. + /// + public static IDictionary ToOutputData(StructuredError error) + { + if (error == null) + return new Dictionary(); + + return new Dictionary { [OutputKey] = ToOutputValue(error) }; + } + + /// Renders just the value placed under , in the canonical snake_case shape. + public static object ToOutputValue(StructuredError error) + { + var json = JsonConvert.SerializeObject(error, ConductorConstants.IoJsonSerializerSettings); + return JsonConvert.DeserializeObject>(json, ConductorConstants.IoJsonSerializerSettings); + } + + /// + /// Tolerantly extracts a from a failed task's output data. Presence-checks the + /// single and deserializes only that subtree. A missing, malformed, or code-less + /// payload returns false and never throws, so a parse problem degrades error quality (falling back to + /// the generic path) rather than failing the failure workflow. + /// + public static bool TryParse(IDictionary taskOutput, out StructuredError error) + { + error = null; + + if (taskOutput == null || !taskOutput.TryGetValue(OutputKey, out var raw) || raw == null) + return false; + + try + { + // raw may be a JObject (Newtonsoft round-trip), a nested dictionary, or a raw JSON string. + var json = raw is string s ? s : JsonConvert.SerializeObject(raw, ConductorConstants.IoJsonSerializerSettings); + var parsed = JsonConvert.DeserializeObject(json, ConductorConstants.IoJsonSerializerSettings); + + // A structured error is only meaningful with a classification code; anything else is treated as + // unstructured and degraded to the generic fallback by the caller. + if (parsed == null || string.IsNullOrEmpty(parsed.Code)) + return false; + + error = parsed; + return true; + } + catch (JsonException) + { + return false; + } + } + } +} diff --git a/src/ConductorSharp.KafkaCancellationNotifier/ConductorSharp.KafkaCancellationNotifier.csproj b/src/ConductorSharp.KafkaCancellationNotifier/ConductorSharp.KafkaCancellationNotifier.csproj index 4da07891..ddf0c885 100644 --- a/src/ConductorSharp.KafkaCancellationNotifier/ConductorSharp.KafkaCancellationNotifier.csproj +++ b/src/ConductorSharp.KafkaCancellationNotifier/ConductorSharp.KafkaCancellationNotifier.csproj @@ -4,7 +4,7 @@ net6.0 enable enable - 3.7.3 + 3.8.0 Codaxy Codaxy diff --git a/src/ConductorSharp.Patterns/ConductorSharp.Patterns.csproj b/src/ConductorSharp.Patterns/ConductorSharp.Patterns.csproj index 268fe0e6..cd3d1cd8 100644 --- a/src/ConductorSharp.Patterns/ConductorSharp.Patterns.csproj +++ b/src/ConductorSharp.Patterns/ConductorSharp.Patterns.csproj @@ -7,7 +7,7 @@ False Codaxy Codaxy - 3.7.3 + 3.8.0 diff --git a/test/ConductorSharp.Engine.Tests/Unit/StructuredErrorTests.cs b/test/ConductorSharp.Engine.Tests/Unit/StructuredErrorTests.cs new file mode 100644 index 00000000..6fd7856e --- /dev/null +++ b/test/ConductorSharp.Engine.Tests/Unit/StructuredErrorTests.cs @@ -0,0 +1,127 @@ +using System.Collections.Generic; +using ConductorSharp.Client; +using ConductorSharp.Client.Util; +using ConductorSharp.Engine.Exceptions; +using ConductorSharp.Engine.Model; +using ConductorSharp.Engine.Util; +using Newtonsoft.Json; +using Newtonsoft.Json.Linq; +using Xunit; + +namespace ConductorSharp.Engine.Tests.Unit +{ + public class StructuredErrorTests + { + // Mirrors the execution-manager catch block: builds the ErrorOutput (setting StructuredError for a + // StructuredErrorException) and serializes it. TryParse below asserts this output round-trips through the + // shared serializer, pinning the property-derived key/shape to StructuredErrorSerializer.OutputKey. + private static IDictionary SerializeCatchOutput(System.Exception exception) + { + var output = new ErrorOutput { ErrorMessage = exception.Message }; + + if (exception is StructuredErrorException structuredException) + { + output.StructuredError = new StructuredError + { + Code = structuredException.Code, + Reason = structuredException.Reason, + ReferenceError = structuredException.ReferenceError + }; + } + + return SerializationHelper.ObjectToDictionary(output, ConductorConstants.IoJsonSerializerSettings); + } + + [Fact] + public void StructuredErrorException_produces_snake_case_structured_error() + { + var exception = new StructuredErrorException("RESOURCE_UNAVAILABLE", "No port available", "https://rom/resourceOrder/42"); + + var dict = SerializeCatchOutput(exception); + + Assert.True(dict.ContainsKey("error_message")); + Assert.True(dict.ContainsKey(StructuredErrorSerializer.OutputKey)); + + var structured = JObject.Parse(JsonConvert.SerializeObject(dict))["structured_error"]; + Assert.Equal("RESOURCE_UNAVAILABLE", (string)structured["code"]); + Assert.Equal("No port available", (string)structured["reason"]); + Assert.Equal("https://rom/resourceOrder/42", (string)structured["reference_error"]); + Assert.Equal(StructuredError.CurrentVersion, (int)structured["version"]); + } + + [Fact] + public void PlainException_output_is_backward_compatible() + { + var dict = SerializeCatchOutput(new System.InvalidOperationException("boom")); + + Assert.True(dict.ContainsKey("error_message")); + Assert.Equal("boom", (string)dict["error_message"]); + Assert.False(dict.ContainsKey(StructuredErrorSerializer.OutputKey)); + Assert.Single(dict); + } + + [Fact] + public void RoundTrip_helper_output_is_parsed_back() + { + var error = new StructuredError + { + Code = "UNCLASSIFIED", + Reason = "generic failure", + ReferenceError = "https://rom/resourceOrder/7" + }; + + var outputData = StructuredErrorSerializer.ToOutputData(error); + + Assert.True(StructuredErrorSerializer.TryParse(outputData, out var parsed)); + Assert.Equal(error.Code, parsed.Code); + Assert.Equal(error.Reason, parsed.Reason); + Assert.Equal(error.ReferenceError, parsed.ReferenceError); + Assert.Equal(error.Version, parsed.Version); + } + + [Fact] + public void RoundTrip_catch_block_output_is_parsed_back() + { + var dict = SerializeCatchOutput(new StructuredErrorException("RESOURCE_UNAVAILABLE", "No port available")); + + Assert.True(StructuredErrorSerializer.TryParse(dict, out var parsed)); + Assert.Equal("RESOURCE_UNAVAILABLE", parsed.Code); + Assert.Equal("No port available", parsed.Reason); + } + + [Fact] + public void TryParse_returns_false_when_key_absent() + { + var dict = new Dictionary { ["error_message"] = "boom" }; + + Assert.False(StructuredErrorSerializer.TryParse(dict, out var parsed)); + Assert.Null(parsed); + } + + [Fact] + public void TryParse_returns_false_on_null_input() + { + Assert.False(StructuredErrorSerializer.TryParse(null, out var parsed)); + Assert.Null(parsed); + } + + [Fact] + public void TryParse_returns_false_on_malformed_payload() + { + var dict = new Dictionary { [StructuredErrorSerializer.OutputKey] = "not-a-structured-error" }; + + Assert.False(StructuredErrorSerializer.TryParse(dict, out _)); + } + + [Fact] + public void TryParse_returns_false_when_code_missing() + { + var dict = new Dictionary + { + [StructuredErrorSerializer.OutputKey] = new Dictionary { ["reason"] = "no code here" } + }; + + Assert.False(StructuredErrorSerializer.TryParse(dict, out _)); + } + } +}