diff --git a/src/ConductorSharp.Client/ConductorSharp.Client.csproj b/src/ConductorSharp.Client/ConductorSharp.Client.csproj index 89a783bd..51431bb1 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.8.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/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/ConductorSharp.Engine.csproj b/src/ConductorSharp.Engine/ConductorSharp.Engine.csproj index ef306da3..0319acbe 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.8.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/Exceptions/StructuredErrorException.cs b/src/ConductorSharp.Engine/Exceptions/StructuredErrorException.cs new file mode 100644 index 00000000..5a928f17 --- /dev/null +++ b/src/ConductorSharp.Engine/Exceptions/StructuredErrorException.cs @@ -0,0 +1,75 @@ +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 // + /// and the diagnostic message 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. + /// + /// + /// There is deliberately no Message property here: the diagnostic message is carried by the inherited + /// , which the message-taking constructors set. When no message is supplied it + /// falls back to , matching the behaviour of the original constructors. + /// + public class StructuredErrorException : Exception + { + /// Stable, opaque classification code. Consumers map this to a failure response. + public string Code { get; } + + /// Short, stable, 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; + } + + /// + /// Declares a diagnostic distinct from the short, stable + /// . Pass null for to fall back to the reason, + /// and null for when there is no entity to drill down into. + /// + /// + /// trails rather than following + /// on purpose. Overload resolution cannot choose between + /// this constructor and the one when the fourth argument is an untyped null, + /// so the nullable parameter is placed third, where it is typed the same either way. The only call this + /// leaves ambiguous is (code, reason, referenceError, null) — a declared-but-null inner exception, + /// which the three-argument constructor already expresses. Disambiguate with a named argument if needed. + /// + public StructuredErrorException(string code, string reason, string referenceError, string message) + : base(message ?? reason) + { + Code = code; + Reason = reason; + ReferenceError = referenceError; + } + + /// + public StructuredErrorException(string code, string reason, string referenceError, string message, Exception innerException) + : base(message ?? reason, innerException) + { + Code = code; + Reason = reason; + ReferenceError = referenceError; + } + } +} diff --git a/src/ConductorSharp.Engine/ExecutionManager.cs b/src/ConductorSharp.Engine/ExecutionManager.cs index 259b4aba..7eb18963 100644 --- a/src/ConductorSharp.Engine/ExecutionManager.cs +++ b/src/ConductorSharp.Engine/ExecutionManager.cs @@ -8,9 +8,11 @@ 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; +using ConductorSharp.Engine.Service; using ConductorSharp.Engine.Util; using MediatR; using Microsoft.Extensions.DependencyInjection; @@ -32,6 +34,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 +45,8 @@ public ExecutionManager( IServiceScopeFactory lifetimeScope, IPollTimingStrategy pollTimingStrategy, IPollOrderStrategy pollOrderStrategy, - ICancellationNotifier cancellationNotifier + ICancellationNotifier cancellationNotifier, + TaskQueuePollingService taskQueuePollingService ) { _configuration = options; @@ -55,6 +59,7 @@ ICancellationNotifier cancellationNotifier _pollOrderStrategy = pollOrderStrategy; _cancellationNotifier = cancellationNotifier; _externalPayloadService = externalPayloadService; + _taskQueuePollingService = taskQueuePollingService; } public async Task StartAsync(CancellationToken cancellationToken) @@ -63,7 +68,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); @@ -242,7 +247,9 @@ await _taskManager.UpdateAsync( pollResponse.WorkflowInstanceId ); - 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. + var errorMessage = new ErrorOutput { ErrorMessage = exception.Message, StructuredError = StructuredError.FromException(exception) }; // 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...) 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/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..e61bbeae --- /dev/null +++ b/src/ConductorSharp.Engine/Model/StructuredError.cs @@ -0,0 +1,57 @@ +using System; +using ConductorSharp.Engine.Exceptions; + +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; } + + /// Short, stable, sanitized reason. Consumers may key off this text, so keep it terse. + public string Reason { get; set; } + + /// + /// Optional diagnostic detail, longer and more specific than — the explanation an + /// operator needs, kept out of so that stays short and stable. Null when the producer + /// supplied nothing distinct from the reason, in which case it is omitted from serialized output + /// (NullValueHandling.Ignore) and the payload is unchanged from before this field existed. + /// + public string Message { 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; + + /// + /// Maps a thrown exception onto the payload, returning null for anything that is not a + /// so plain exceptions keep producing only error_message. + /// This is the single exception-to-payload mapping: both execution managers and the contract tests call it, + /// so the two poll strategies cannot drift apart as the shape evolves. + /// + public static StructuredError FromException(Exception exception) + { + if (exception is not StructuredErrorException structuredException) + return null; + + return new StructuredError + { + Code = structuredException.Code, + Reason = structuredException.Reason, + // Exception.Message falls back to Reason when the thrower supplied no distinct detail, so only + // carry it when it actually adds something. Existing call sites keep their exact payload. + Message = structuredException.Message == structuredException.Reason ? null : structuredException.Message, + ReferenceError = structuredException.ReferenceError + }; + } + } +} 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..550bf1ec 100644 --- a/src/ConductorSharp.Engine/TypePollSpreadingExecutionManager.cs +++ b/src/ConductorSharp.Engine/TypePollSpreadingExecutionManager.cs @@ -8,9 +8,11 @@ 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; +using ConductorSharp.Engine.Service; using ConductorSharp.Engine.Util; using MediatR; using Microsoft.Extensions.DependencyInjection; @@ -32,6 +34,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 +45,8 @@ public TypePollSpreadingExecutionManager( IServiceScopeFactory lifetimeScope, IPollTimingStrategy pollTimingStrategy, IPollOrderStrategy pollOrderStrategy, - ICancellationNotifier cancellationNotifier + ICancellationNotifier cancellationNotifier, + TaskQueuePollingService taskQueuePollingService ) { _configuration = options; @@ -55,6 +59,7 @@ ICancellationNotifier cancellationNotifier _pollOrderStrategy = pollOrderStrategy; _cancellationNotifier = cancellationNotifier; _externalPayloadService = externalPayloadService; + _taskQueuePollingService = taskQueuePollingService; } public async Task StartAsync(CancellationToken cancellationToken) @@ -63,7 +68,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); @@ -251,7 +256,9 @@ await _taskManager.UpdateAsync( pollResponse.WorkflowInstanceId ); - 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. + var errorMessage = new ErrorOutput { ErrorMessage = exception.Message, StructuredError = StructuredError.FromException(exception) }; // 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...) 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 0cce80b7..5cb93853 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.8.1 Codaxy Codaxy diff --git a/src/ConductorSharp.Patterns/ConductorSharp.Patterns.csproj b/src/ConductorSharp.Patterns/ConductorSharp.Patterns.csproj index 2d92ff76..3fe05855 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.8.1 diff --git a/test/ConductorSharp.Engine.Tests/Unit/StructuredErrorTests.cs b/test/ConductorSharp.Engine.Tests/Unit/StructuredErrorTests.cs new file mode 100644 index 00000000..c4ab097a --- /dev/null +++ b/test/ConductorSharp.Engine.Tests/Unit/StructuredErrorTests.cs @@ -0,0 +1,208 @@ +using System.Collections.Generic; +using System.Linq; +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. It calls StructuredError.FromException — the same mapping both + // ExecutionManager and TypePollSpreadingExecutionManager use — rather than reimplementing it, so a field + // added to the payload cannot pass here while being dropped by one of the managers. + private static IDictionary SerializeCatchOutput(System.Exception exception) + { + var output = new ErrorOutput { ErrorMessage = exception.Message, StructuredError = StructuredError.FromException(exception) }; + + return SerializationHelper.ObjectToDictionary(output, ConductorConstants.IoJsonSerializerSettings); + } + + private static JToken StructuredErrorOf(IDictionary dict) => + JObject.Parse(JsonConvert.SerializeObject(dict))[StructuredErrorSerializer.OutputKey]; + + [Fact] + public void StructuredErrorException_produces_snake_case_structured_error() + { + var exception = new StructuredErrorException("RESOURCE_UNAVAILABLE", "No port available", "https://example.org/entity/42"); + + var dict = SerializeCatchOutput(exception); + + Assert.True(dict.ContainsKey("error_message")); + Assert.True(dict.ContainsKey(StructuredErrorSerializer.OutputKey)); + + var structured = StructuredErrorOf(dict); + Assert.Equal("RESOURCE_UNAVAILABLE", (string)structured["code"]); + Assert.Equal("No port available", (string)structured["reason"]); + Assert.Equal("https://example.org/entity/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 Message_is_carried_under_snake_case_message_key() + { + var exception = new StructuredErrorException( + "VALIDATION_FAILED", + "Input field not recognized", + "https://example.org/entity/7", + "Field 'widget_id' is not present in schema 'default'." + ); + + var structured = StructuredErrorOf(SerializeCatchOutput(exception)); + + Assert.Equal("VALIDATION_FAILED", (string)structured["code"]); + Assert.Equal("Input field not recognized", (string)structured["reason"]); + Assert.Equal("Field 'widget_id' is not present in schema 'default'.", (string)structured["message"]); + Assert.Equal("https://example.org/entity/7", (string)structured["reference_error"]); + } + + [Fact] + public void Message_reaches_error_message_and_reason_for_incompletion() + { + // error_message is set from Exception.Message, which the message-taking constructor overrides. The same + // value is what the execution manager sends as TaskResult.ReasonForIncompletion (the Conductor UI banner). + var exception = new StructuredErrorException("CODE", "Short reason", null, "Long diagnostic detail"); + + Assert.Equal("Long diagnostic detail", exception.Message); + Assert.Equal("Long diagnostic detail", (string)SerializeCatchOutput(exception)["error_message"]); + } + + [Fact] + public void Message_is_omitted_when_no_message_was_supplied() + { + // Guards the backward-compatibility promise: pre-existing call sites must keep their exact payload. + var structured = StructuredErrorOf( + SerializeCatchOutput(new StructuredErrorException("CODE", "Short reason", "https://example.org/entity/1")) + ); + + Assert.Null(structured["message"]); + Assert.Equal( + new[] { "code", "reason", "reference_error", "version" }, + ((JObject)structured).Properties().Select(p => p.Name).OrderBy(n => n) + ); + } + + [Fact] + public void Message_is_omitted_when_it_only_repeats_the_reason() + { + var exception = new StructuredErrorException("CODE", "Same text", null, "Same text"); + + Assert.Null(StructuredErrorOf(SerializeCatchOutput(exception))["message"]); + } + + [Fact] + public void Message_survives_the_round_trip() + { + var dict = SerializeCatchOutput( + new StructuredErrorException("CODE", "Short reason", "https://example.org/entity/9", "Long diagnostic detail") + ); + + Assert.True(StructuredErrorSerializer.TryParse(dict, out var parsed)); + Assert.Equal("CODE", parsed.Code); + Assert.Equal("Short reason", parsed.Reason); + Assert.Equal("Long diagnostic detail", parsed.Message); + Assert.Equal("https://example.org/entity/9", parsed.ReferenceError); + } + + [Fact] + public void FromException_returns_null_for_a_plain_exception() + { + Assert.Null(StructuredError.FromException(new System.InvalidOperationException("boom"))); + } + + [Fact] + public void RoundTrip_helper_output_is_parsed_back() + { + // The signal-sender producer: no exception to catch, so the payload is rendered from the model directly. + var error = new StructuredError + { + Code = "UNCLASSIFIED", + Reason = "generic failure", + Message = "downstream call failed: connection refused", + ReferenceError = "https://example.org/entity/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.Message, parsed.Message); + 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 _)); + } + + [Fact] + public void TryParse_tolerates_a_message_only_payload_by_degrading() + { + // A message without a code is still unstructured: the caller must fall back to the generic path. + var dict = new Dictionary + { + [StructuredErrorSerializer.OutputKey] = new Dictionary { ["message"] = "detail but no code" } + }; + + Assert.False(StructuredErrorSerializer.TryParse(dict, out _)); + } + } +} 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;