From 08520f52b8acc73578613f1fabebc9f57f78446a Mon Sep 17 00:00:00 2001 From: Alan Deutscher Date: Tue, 18 Aug 2026 11:38:12 -0700 Subject: [PATCH 1/5] Initial Cursor output --- .../Extensions/ServiceCollectionExtensions.cs | 5 +- .../Models/ActiveMqExceptionArbiterReport.cs | 9 + .../Services/ActiveMqJobSource.cs | 147 +++++---- .../ActiveMqExceptionArbiterService.cs | 118 +++++++ .../Resilience/ActiveMqRetryWrapperService.cs | 210 ++++++++++++ .../ServiceCollectionExtensionsTests.cs | 32 ++ .../Tests/Services/ActiveMqJobSourceTests.cs | 38 ++- .../Services/ActiveMqRetryTestHelpers.cs | 21 ++ .../ActiveMqExceptionArbiterServiceTests.cs | 249 ++++++++++++++ .../ActiveMqRetryWrapperServiceTests.cs | 307 ++++++++++++++++++ 10 files changed, 1048 insertions(+), 88 deletions(-) create mode 100644 src/RedShirt.Example.JobWorker.JobManagement.ActiveMq/Models/ActiveMqExceptionArbiterReport.cs create mode 100644 src/RedShirt.Example.JobWorker.JobManagement.ActiveMq/Services/Resilience/ActiveMqExceptionArbiterService.cs create mode 100644 src/RedShirt.Example.JobWorker.JobManagement.ActiveMq/Services/Resilience/ActiveMqRetryWrapperService.cs create mode 100644 test/RedShirt.Example.JobWorker.JobManagement.ActiveMq.UnitTests/Tests/Extensions/ServiceCollectionExtensionsTests.cs create mode 100644 test/RedShirt.Example.JobWorker.JobManagement.ActiveMq.UnitTests/Tests/Services/ActiveMqRetryTestHelpers.cs create mode 100644 test/RedShirt.Example.JobWorker.JobManagement.ActiveMq.UnitTests/Tests/Services/Resilience/ActiveMqExceptionArbiterServiceTests.cs create mode 100644 test/RedShirt.Example.JobWorker.JobManagement.ActiveMq.UnitTests/Tests/Services/Resilience/ActiveMqRetryWrapperServiceTests.cs diff --git a/src/RedShirt.Example.JobWorker.JobManagement.ActiveMq/Extensions/ServiceCollectionExtensions.cs b/src/RedShirt.Example.JobWorker.JobManagement.ActiveMq/Extensions/ServiceCollectionExtensions.cs index 05ac7ad2..607edbcb 100644 --- a/src/RedShirt.Example.JobWorker.JobManagement.ActiveMq/Extensions/ServiceCollectionExtensions.cs +++ b/src/RedShirt.Example.JobWorker.JobManagement.ActiveMq/Extensions/ServiceCollectionExtensions.cs @@ -3,6 +3,7 @@ using RedShirt.Example.JobWorker.Core.Services.Abstractions; using RedShirt.Example.JobWorker.JobManagement.ActiveMq.Factories; using RedShirt.Example.JobWorker.JobManagement.ActiveMq.Services; +using RedShirt.Example.JobWorker.JobManagement.ActiveMq.Services.Resilience; namespace RedShirt.Example.JobWorker.JobManagement.ActiveMq.Extensions; @@ -21,6 +22,8 @@ public static IServiceCollection AddActiveMqJobManagement(this IServiceCollectio configuration.GetSection("JobSource:ActiveMq")) .AddSingleton() .AddSingleton() - .AddSingleton(); + .AddSingleton() + .AddSingleton() + .AddSingleton(); } } \ No newline at end of file diff --git a/src/RedShirt.Example.JobWorker.JobManagement.ActiveMq/Models/ActiveMqExceptionArbiterReport.cs b/src/RedShirt.Example.JobWorker.JobManagement.ActiveMq/Models/ActiveMqExceptionArbiterReport.cs new file mode 100644 index 00000000..1b6d8a74 --- /dev/null +++ b/src/RedShirt.Example.JobWorker.JobManagement.ActiveMq/Models/ActiveMqExceptionArbiterReport.cs @@ -0,0 +1,9 @@ +namespace RedShirt.Example.JobWorker.JobManagement.ActiveMq.Models; + +internal sealed class ActiveMqExceptionArbiterReport +{ + public required bool AlreadyHandled { get; init; } + public required bool IsExpected { get; init; } + public required bool CouldBeExternallySolvable { get; init; } + public required bool CouldBeTransient { get; init; } +} diff --git a/src/RedShirt.Example.JobWorker.JobManagement.ActiveMq/Services/ActiveMqJobSource.cs b/src/RedShirt.Example.JobWorker.JobManagement.ActiveMq/Services/ActiveMqJobSource.cs index ce8068ba..496658cb 100644 --- a/src/RedShirt.Example.JobWorker.JobManagement.ActiveMq/Services/ActiveMqJobSource.cs +++ b/src/RedShirt.Example.JobWorker.JobManagement.ActiveMq/Services/ActiveMqJobSource.cs @@ -7,71 +7,102 @@ using RedShirt.Example.JobWorker.JobManagement.ActiveMq.Exceptions; using RedShirt.Example.JobWorker.JobManagement.ActiveMq.Factories; using RedShirt.Example.JobWorker.JobManagement.ActiveMq.Models; +using RedShirt.Example.JobWorker.JobManagement.ActiveMq.Services.Resilience; namespace RedShirt.Example.JobWorker.JobManagement.ActiveMq.Services; internal class ActiveMqJobSource : IJobSource { private readonly IOptions _configuration; - - // ReSharper disable once PrivateFieldCanBeConvertedToLocalVariable - private readonly Lazy> _connection; + private readonly IActiveMqConnectionFactory _connectionFactory; private readonly ILogger _logger; + private readonly IActiveMqRetryWrapperService _retryWrapperService; + private IMessageConsumer? _messageConsumer; + + private async Task FetchJobsAsync(int batchSize, CancellationToken cancellationToken) + { + try + { + var consumer = await GetConsumerAsync(cancellationToken); + var getJobsResponseItems = new List(); + + while (getJobsResponseItems.Count < batchSize) + { + var result = await consumer.ReceiveAsync(TimeSpan.FromMilliseconds(100)); + + if (result is null) + // Nothing more to grab at the moment. + { + break; + } + + // Got a message, add it to return set. + getJobsResponseItems.Add(new ActiveMqRawJobModel + { + Message = result, + MessageId = result.NMSMessageId, // Not really used by this framework, but why not + CreatedAtUtc = DateTime.UtcNow + }); + } + + return new JobSourceResponse + { + Items = getJobsResponseItems + }; + } + catch + { + Reset(); + throw; + } + } - // ReSharper disable once PrivateFieldCanBeConvertedToLocalVariable - private readonly Lazy> _messageConsumer; + private async Task GetConsumerAsync(CancellationToken cancellationToken) + { + if (_messageConsumer is not null) + { + return _messageConsumer; + } - // ReSharper disable once PrivateFieldCanBeConvertedToLocalVariable - private readonly Lazy> _queue; - private readonly Lazy> _session; + var connection = await _connectionFactory.GetConnectionAsync(cancellationToken); + connection.Start(); + var session = await connection.CreateSessionAsync(AcknowledgementMode.ClientAcknowledge); + var queue = await session.GetQueueAsync(_configuration.Value.QueueName); + + if (queue is null) + { + throw new CouldNotLoadQueueException(); + } + + var consumer = await session.CreateConsumerAsync(queue); + _messageConsumer = consumer; + return consumer; + } + + private void Reset() + { + _messageConsumer = null; + } public ActiveMqJobSource(IActiveMqConnectionFactory connectionFactory, + IActiveMqRetryWrapperService retryWrapperService, IOptions configuration, ILogger logger) { + _connectionFactory = connectionFactory; + _retryWrapperService = retryWrapperService; _configuration = configuration; _logger = logger; - _connection = new Lazy>(async () => - { - var connection = await connectionFactory.GetConnectionAsync(); - connection.Start(); - return connection; - }); - _session = new Lazy>(async () => - { - var connection = await _connection.Value; - return await connection.CreateSessionAsync(AcknowledgementMode.ClientAcknowledge); - }); - _queue = new Lazy>(async () => - { - var session = await _session.Value; - return await session.GetQueueAsync(_configuration.Value.QueueName); - }); - _messageConsumer = new Lazy>(async () => - { - var queue = await _queue.Value; - - if (queue is null) - { - throw new CouldNotLoadQueueException(); - } - - var session = await _session.Value; - return await session.CreateConsumerAsync(queue); - }); } public int RecommendedHeartbeatIntervalSeconds => 0; -#pragma warning disable S2325 - public Task AcknowledgeAsync(IRawJobModel message, CoreJobResult result, + public async Task AcknowledgeAsync(IRawJobModel message, CoreJobResult result, CancellationToken cancellationToken = default) -#pragma warning restore S2325 { - // ReSharper disable once ConvertIfStatementToReturnStatement if (message is not ActiveMqRawJobModel jobModel) { - return Task.CompletedTask; + return; } // Intentionally not using result @@ -80,7 +111,9 @@ public Task AcknowledgeAsync(IRawJobModel message, CoreJobResult result, // Acknowledge whether successful, recoverable, or unrecoverable // (ActiveMQ client API has no direct dead-letter call here). - return jobModel.Message.AcknowledgeAsync(); + await _retryWrapperService.RunAsync( + _ => jobModel.Message.AcknowledgeAsync(), + cancellationToken); } public async Task GetJobsAsync(int batchSize, CancellationToken cancellationToken = default) @@ -90,33 +123,9 @@ public async Task GetJobsAsync(int batchSize, CancellationTo _logger.LogTrace("Fetching up to {EffectiveBatchSize} messages from ActiveMQ Queue: {QueueName}", batchSize, _configuration.Value.QueueName); - var getJobsResponseItems = new List(); - - var consumer = await _messageConsumer.Value; - - while (getJobsResponseItems.Count < batchSize) - { - var result = await consumer.ReceiveAsync(TimeSpan.FromMilliseconds(100)); - - if (result is null) - // Nothing more to grab at the moment. - { - break; - } - - // Got a message, add it to return set. - getJobsResponseItems.Add(new ActiveMqRawJobModel - { - Message = result, - MessageId = result.NMSMessageId, // Not really used by this framework, but why not - CreatedAtUtc = DateTime.UtcNow - }); - } - - return new JobSourceResponse - { - Items = getJobsResponseItems - }; + return await _retryWrapperService.RunAsync( + ct => FetchJobsAsync(batchSize, ct), + cancellationToken); } public Task HeartbeatAsync(IRawJobModel message, CancellationToken cancellationToken = default) diff --git a/src/RedShirt.Example.JobWorker.JobManagement.ActiveMq/Services/Resilience/ActiveMqExceptionArbiterService.cs b/src/RedShirt.Example.JobWorker.JobManagement.ActiveMq/Services/Resilience/ActiveMqExceptionArbiterService.cs new file mode 100644 index 00000000..f77c4a5a --- /dev/null +++ b/src/RedShirt.Example.JobWorker.JobManagement.ActiveMq/Services/Resilience/ActiveMqExceptionArbiterService.cs @@ -0,0 +1,118 @@ +using Apache.NMS; +using Apache.NMS.ActiveMQ; +using RedShirt.Example.JobWorker.Core.Exceptions; +using RedShirt.Example.JobWorker.JobManagement.ActiveMq.Exceptions; +using RedShirt.Example.JobWorker.JobManagement.ActiveMq.Models; +using System.Net.Sockets; +using ActiveMqIoException = Apache.NMS.ActiveMQ.IOException; + +namespace RedShirt.Example.JobWorker.JobManagement.ActiveMq.Services.Resilience; + +/// +/// Classifies ActiveMQ / NMS client exceptions for retry decisions. +/// +internal interface IActiveMqExceptionArbiterService +{ + ActiveMqExceptionArbiterReport GetReport(Exception exception); +} + +/// +/// ActiveMQ-oriented exception arbiter modelled after the Kafka / Redis Streams / Pulsar arbiters: +/// known infrastructure failures may be transient; auth, cancel, and bad arguments are not. +/// +internal class ActiveMqExceptionArbiterService : IActiveMqExceptionArbiterService +{ + private static ActiveMqExceptionArbiterReport Fresh( + bool isExpected, + bool couldBeTransient, + bool couldBeExternallySolvable) + { + return new ActiveMqExceptionArbiterReport + { + AlreadyHandled = false, + IsExpected = isExpected, + CouldBeTransient = couldBeTransient, + CouldBeExternallySolvable = couldBeExternallySolvable + }; + } + + private static ActiveMqExceptionArbiterReport Handled( + bool isExpected, + bool couldBeTransient, + bool couldBeExternallySolvable) + { + return new ActiveMqExceptionArbiterReport + { + AlreadyHandled = true, + IsExpected = isExpected, + CouldBeTransient = couldBeTransient, + CouldBeExternallySolvable = couldBeExternallySolvable + }; + } + + public ActiveMqExceptionArbiterReport GetReport(Exception exception) + { + ArgumentNullException.ThrowIfNull(exception); + + while (exception is AggregateException {InnerExceptions.Count: 1, InnerException: not null} aggregate) + { + exception = aggregate.InnerException!; + } + + return exception switch + { + // Already classified/wrapped by an earlier job-source layer — do not wrap again. + // Only allow further retry when the prior wrapper has not already exhausted retries. + WorkerJobSourceException workerJobSource => + Handled( + true, + workerJobSource is {IsHandled: false, CouldBeTransient: true}, + workerJobSource.CouldBeExternallySolvable), + // Queue lookup returned null — ops can create the destination without a worker restart. + CouldNotLoadQueueException => Fresh(true, false, true), + // Unsupported / unreadable payload — a local data issue, not retryable. + CouldNotRetrieveMessageBodyException => Fresh(true, false, false), + // Auth failures — ops can grant credentials / ACLs externally. + NMSSecurityException => Fresh(true, false, true), + // Missing / invalid destination — ops can create or restore the queue externally. + InvalidDestinationException => Fresh(true, false, true), + // Bad local client identity or selector — requires a config change, not an external fix. + InvalidClientIDException + or InvalidSelectorException => Fresh(true, false, false), + // Payload / cursor issues — not retryable and not externally solvable. + MessageEOFException + or MessageFormatException + or MessageNotReadableException + or MessageNotWriteableException => Fresh(true, false, false), + // Nested transaction — a local client-state problem. + TransactionInProgressException => Fresh(true, false, false), + // Broker rolled back — a brief conflict that can clear on retry / broker recovery. + TransactionRolledBackException => Fresh(true, true, true), + // Timeouts and broker resource pressure — infra blips ops can clear. + RequestTimedOutException + or ResourceAllocationException => Fresh(true, true, true), + // Connection / consumer lifecycle blips — reconnecting or restarting the broker can clear them. + NMSConnectionException + or ConnectionClosedException + or ConnectionFailedException + or ConsumerClosedException + or IllegalStateException => Fresh(true, true, true), + // Transport IO failures from the OpenWire client. + ActiveMqIoException => Fresh(true, true, true), + // Remaining NMS failures (including BrokerException) are expected broker/client issues. + NMSException => Fresh(true, true, true), + TimeoutException + or SocketException + or System.IO.IOException => Fresh(true, true, true), + // HttpClient-style timeouts sometimes surface as TaskCanceledException. + // Must be matched before OperationCanceledException (TCE derives from OCE). + TaskCanceledException => Fresh(true, true, true), + // Explicit CancellationToken cancellation from the caller — do not retry; not externally solvable. + OperationCanceledException => Fresh(true, false, false), + // Client-side argument validation — not retryable and not externally solvable. + ArgumentException => Fresh(true, false, false), + // Unrecognized exception type — treat as unexpected so callers surface the raw failure. + _ => Fresh(false, false, false) + }; + } +} diff --git a/src/RedShirt.Example.JobWorker.JobManagement.ActiveMq/Services/Resilience/ActiveMqRetryWrapperService.cs b/src/RedShirt.Example.JobWorker.JobManagement.ActiveMq/Services/Resilience/ActiveMqRetryWrapperService.cs new file mode 100644 index 00000000..76e9af5e --- /dev/null +++ b/src/RedShirt.Example.JobWorker.JobManagement.ActiveMq/Services/Resilience/ActiveMqRetryWrapperService.cs @@ -0,0 +1,210 @@ +using Microsoft.Extensions.Logging; +using Polly; +using Polly.Retry; +using RedShirt.Example.JobWorker.Common.Services.Utility; +using RedShirt.Example.JobWorker.Core.Exceptions; + +namespace RedShirt.Example.JobWorker.JobManagement.ActiveMq.Services.Resilience; + +/// +/// Retries ActiveMQ client operations that fail with expected transient exceptions, +/// then surfaces remaining failures as with +/// set so Core does not retry again. +/// +internal interface IActiveMqRetryWrapperService +{ + /// + /// Executes with retry for expected transient ActiveMQ failures. + /// + /// The result type produced by . + /// + /// The operation to execute. Receives the same used for retries and backoff delays. + /// + /// + /// Token used to cancel the operation, retry attempts, and backoff delays. + /// + /// The successful result of . + /// + /// Propagated when is cancelled. + /// + /// + /// Thrown when ultimately fails. + /// reflects the arbiter judgement for the final exception; + /// is true. + /// + Task RunAsync(Func> func, CancellationToken cancellationToken = default); + + /// + /// Executes with retry for expected transient ActiveMQ failures. + /// + /// + /// The operation to execute. Receives the same used for retries and backoff delays. + /// + /// + /// Token used to cancel the operation, retry attempts, and backoff delays. + /// + /// + /// Propagated when is cancelled. + /// + /// + /// Thrown when ultimately fails. + /// reflects the arbiter judgement for the final exception; + /// is true. + /// + Task RunAsync(Func func, CancellationToken cancellationToken = default); +} + +/// +/// Polly v8-based retry wrapper for ActiveMQ client calls. +/// Retries when reports an expected transient failure, +/// using exponential backoff via . +/// +/// Classifies ActiveMQ-related exceptions as expected/transient. +/// Logs each retry attempt. +/// Provides cancellable backoff delays between retry attempts. +internal class ActiveMqRetryWrapperService( + IActiveMqExceptionArbiterService exceptionArbiterService, + ILogger logger, + ISleepService sleepService) + : IActiveMqRetryWrapperService +{ + private const int ActiveMqRetryCount = 3; + + /// + /// Lazily built Polly v8 shared across invocations. + /// + private ResiliencePipeline? _retryPipeline; + + /// + /// Creates (once) the retry pipeline: arbiter-driven ShouldHandle, zero Polly delay, + /// and exponential backoff performed in OnRetry through . + /// + private ResiliencePipeline GetRetryPipeline() + { + return _retryPipeline ??= new ResiliencePipelineBuilder() + .AddRetry(new RetryStrategyOptions + { + MaxRetryAttempts = ActiveMqRetryCount, + ShouldHandle = args => + { + if (args.Outcome.Exception is not { } exception) + { + return PredicateResult.False(); + } + + // Cancellation is honoured via ResilienceContext. + if (args.Context.CancellationToken.IsCancellationRequested) + { + return PredicateResult.False(); + } + + var report = exceptionArbiterService.GetReport(exception); + return report is {IsExpected: true, CouldBeTransient: true} + ? PredicateResult.True() + : PredicateResult.False(); + }, + // Do not use Polly-based delays between attempts + DelayGenerator = static _ => new ValueTask(TimeSpan.Zero), + OnRetry = async args => + { + logger.LogWarning(args.Outcome.Exception, + "Retrying ActiveMQ operation after attempt {AttemptNumber}", + args.AttemptNumber); + // Delay is performed via ISleepService in OnRetry so tests can mock sleeps. + await sleepService.DelayAsync(TimeSpan.FromSeconds(Math.Pow(2, args.AttemptNumber)), + args.Context.CancellationToken); + } + }) + .Build(); + } + + /// + /// Try to get the wrapped exception. + /// + /// Exception to be judged. + /// + /// If wrapping was appropriate, then will be wrapped around the + /// . + /// If wrapping was not appropriate, then will be null. + /// + /// true if the exception was wrapped, else false + private bool TryGetWrappedException(Exception exception, out Exception? wrappedException) + { + wrappedException = null; + var report = exceptionArbiterService.GetReport(exception); + + // ReSharper disable once DuplicatedSequentialIfBodies + if (report.AlreadyHandled && exception is WorkerJobSourceException) + { + return false; + } + + if (!report.IsExpected) + { + /* + * Unexpected / unrecognized. + * Unexpected failures stay raw so they raise attention and get classified. + */ + return false; + } + + wrappedException = new WorkerJobSourceException(exception) + { + CouldBeTransient = report.CouldBeTransient, + IsHandled = true, + CouldBeExternallySolvable = report.CouldBeExternallySolvable + }; + return true; + } + + /// + public async Task RunAsync(Func> func, + CancellationToken cancellationToken = default) + { + try + { + return await GetRetryPipeline().ExecuteAsync( + async token => await func(token), + cancellationToken); + } + catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested) + { + throw; + } + catch (Exception exception) + { + if (TryGetWrappedException(exception, out var wrappedException) && wrappedException is not null) + { + throw wrappedException; + } + + // Do a flat throw to preserve stack trace + throw; + } + } + + /// + public async Task RunAsync(Func func, CancellationToken cancellationToken = default) + { + try + { + await GetRetryPipeline().ExecuteAsync( + async token => await func(token), + cancellationToken); + } + catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested) + { + throw; + } + catch (Exception exception) + { + if (TryGetWrappedException(exception, out var wrappedException) && wrappedException is not null) + { + throw wrappedException; + } + + // Do a flat throw to preserve stack trace + throw; + } + } +} diff --git a/test/RedShirt.Example.JobWorker.JobManagement.ActiveMq.UnitTests/Tests/Extensions/ServiceCollectionExtensionsTests.cs b/test/RedShirt.Example.JobWorker.JobManagement.ActiveMq.UnitTests/Tests/Extensions/ServiceCollectionExtensionsTests.cs new file mode 100644 index 00000000..4cc6a938 --- /dev/null +++ b/test/RedShirt.Example.JobWorker.JobManagement.ActiveMq.UnitTests/Tests/Extensions/ServiceCollectionExtensionsTests.cs @@ -0,0 +1,32 @@ +using Microsoft.Extensions.Configuration; +using Microsoft.Extensions.DependencyInjection; +using RedShirt.Example.JobWorker.Core.Services.Abstractions; +using RedShirt.Example.JobWorker.JobManagement.ActiveMq.Extensions; +using RedShirt.Example.JobWorker.JobManagement.ActiveMq.Services; +using RedShirt.Example.JobWorker.JobManagement.ActiveMq.Services.Resilience; + +namespace RedShirt.Example.JobWorker.JobManagement.ActiveMq.UnitTests.Tests.Extensions; + +public class ServiceCollectionExtensionsTests +{ + [Fact] + public void AddActiveMqJobManagement_RegistersExpectedServices() + { + var services = new ServiceCollection(); + var configuration = new ConfigurationBuilder() + .AddInMemoryCollection(new Dictionary + { + ["JobSource:ActiveMq:QueueName"] = "jobs" + }) + .Build(); + + services.AddActiveMqJobManagement(configuration); + + Assert.Contains(services, d => d.ServiceType == typeof(IJobSource) + && d.ImplementationType == typeof(ActiveMqJobSource)); + Assert.Contains(services, d => d.ServiceType == typeof(IJobFailureHandler) + && d.ImplementationType == typeof(NoReactionFailureHandler)); + Assert.Contains(services, d => d.ServiceType == typeof(IActiveMqExceptionArbiterService)); + Assert.Contains(services, d => d.ServiceType == typeof(IActiveMqRetryWrapperService)); + } +} diff --git a/test/RedShirt.Example.JobWorker.JobManagement.ActiveMq.UnitTests/Tests/Services/ActiveMqJobSourceTests.cs b/test/RedShirt.Example.JobWorker.JobManagement.ActiveMq.UnitTests/Tests/Services/ActiveMqJobSourceTests.cs index 408e6385..b778d2b9 100644 --- a/test/RedShirt.Example.JobWorker.JobManagement.ActiveMq.UnitTests/Tests/Services/ActiveMqJobSourceTests.cs +++ b/test/RedShirt.Example.JobWorker.JobManagement.ActiveMq.UnitTests/Tests/Services/ActiveMqJobSourceTests.cs @@ -12,6 +12,17 @@ namespace RedShirt.Example.JobWorker.JobManagement.ActiveMq.UnitTests.Tests.Serv public class ActiveMqJobSourceTests { + private static ActiveMqJobSource CreateJobSource( + IActiveMqConnectionFactory? factory, + ActiveMqJobSource.ConfigurationModel configuration) + { + return new ActiveMqJobSource( + factory!, + ActiveMqRetryTestHelpers.CreatePassthroughRetryWrapper().Object, + Options.Create(configuration), + new NullLogger()); + } + [Fact] public async Task Test_AcknowledgeAsync() { @@ -22,8 +33,7 @@ public async Task Test_AcknowledgeAsync() QueueName = null! }; - var activeMqJobSource = new ActiveMqJobSource(null!, Options.Create(configuration), - new NullLogger()); + var activeMqJobSource = CreateJobSource(null, configuration); var jobModel = new ActiveMqRawJobModel { @@ -54,8 +64,7 @@ public async Task Test_AcknowledgeAsync_AlwaysAcknowledges(CoreJobResult result) QueueName = null! }; - var activeMqJobSource = new ActiveMqJobSource(null!, Options.Create(configuration), - new NullLogger()); + var activeMqJobSource = CreateJobSource(null, configuration); var jobModel = new ActiveMqRawJobModel { @@ -80,8 +89,7 @@ public async Task Test_AcknowledgeAsync_Incompatible() QueueName = null! }; - var activeMqJobSource = new ActiveMqJobSource(null!, Options.Create(configuration), - new NullLogger()); + var activeMqJobSource = CreateJobSource(null, configuration); await activeMqJobSource.AcknowledgeAsync(job.Object, CoreJobResult.Success, TestContext.Current.CancellationToken); @@ -120,8 +128,7 @@ public async Task Test_GetJobs_GetNoJobs() .Setup(c => c.ReceiveAsync(It.IsAny())) .ReturnsAsync(() => null); - var jobSource = new ActiveMqJobSource(activeConnectionFactory.Object, Options.Create(configuration), - new NullLogger()); + var jobSource = CreateJobSource(activeConnectionFactory.Object, configuration); var jobResponse = await jobSource.GetJobsAsync(10, TestContext.Current.CancellationToken); @@ -160,8 +167,7 @@ public async Task Test_GetJobs_GetNoQueue() activeConnectionFactory.Setup(f => f.GetConnectionAsync(It.IsAny())) .ReturnsAsync(mockConnection.Object); - var jobSource = new ActiveMqJobSource(activeConnectionFactory.Object, Options.Create(configuration), - new NullLogger()); + var jobSource = CreateJobSource(activeConnectionFactory.Object, configuration); await Assert.ThrowsAsync(() => jobSource.GetJobsAsync(1, TestContext.Current.CancellationToken)); @@ -215,8 +221,7 @@ public async Task Test_GetJobs_GotJob() .Setup(c => c.ReceiveAsync(It.IsAny())) .ReturnsAsync(() => mockChannelQueue.TryDequeue(out var job) ? job : null); - var jobSource = new ActiveMqJobSource(activeConnectionFactory.Object, Options.Create(configuration), - new NullLogger()); + var jobSource = CreateJobSource(activeConnectionFactory.Object, configuration); var jobResponse = await jobSource.GetJobsAsync(1, TestContext.Current.CancellationToken); @@ -275,8 +280,7 @@ public async Task Test_GetJobs_GotJob_BatchSizeZero() .Setup(c => c.ReceiveAsync(It.IsAny())) .ReturnsAsync(() => mockChannelQueue.TryDequeue(out var job) ? job : null); - var jobSource = new ActiveMqJobSource(activeConnectionFactory.Object, Options.Create(configuration), - new NullLogger()); + var jobSource = CreateJobSource(activeConnectionFactory.Object, configuration); var jobResponse = await jobSource.GetJobsAsync(0, TestContext.Current.CancellationToken); @@ -342,8 +346,7 @@ public async Task Test_GetJobs_GotJobs_MultipleJobs(int batchSize) .Setup(c => c.ReceiveAsync(It.IsAny())) .ReturnsAsync(() => mockChannelQueue.TryDequeue(out var job) ? job : null); - var jobSource = new ActiveMqJobSource(activeConnectionFactory.Object, Options.Create(configuration), - new NullLogger()); + var jobSource = CreateJobSource(activeConnectionFactory.Object, configuration); var jobResponse = await jobSource.GetJobsAsync(batchSize, TestContext.Current.CancellationToken); @@ -369,8 +372,7 @@ public async Task Test_HeartbeatAsync() QueueName = null! }; - var jobSource = new ActiveMqJobSource(null!, Options.Create(configuration), - new NullLogger()); + var jobSource = CreateJobSource(null, configuration); await jobSource.HeartbeatAsync(null!, TestContext.Current.CancellationToken); } diff --git a/test/RedShirt.Example.JobWorker.JobManagement.ActiveMq.UnitTests/Tests/Services/ActiveMqRetryTestHelpers.cs b/test/RedShirt.Example.JobWorker.JobManagement.ActiveMq.UnitTests/Tests/Services/ActiveMqRetryTestHelpers.cs new file mode 100644 index 00000000..872b0b54 --- /dev/null +++ b/test/RedShirt.Example.JobWorker.JobManagement.ActiveMq.UnitTests/Tests/Services/ActiveMqRetryTestHelpers.cs @@ -0,0 +1,21 @@ +using RedShirt.Example.JobWorker.Core.Models; +using RedShirt.Example.JobWorker.JobManagement.ActiveMq.Services.Resilience; + +namespace RedShirt.Example.JobWorker.JobManagement.ActiveMq.UnitTests.Tests.Services; + +internal static class ActiveMqRetryTestHelpers +{ + public static Mock CreatePassthroughRetryWrapper() + { + var retry = new Mock(MockBehavior.Strict); + retry + .Setup(r => r.RunAsync(It.IsAny>(), It.IsAny())) + .Returns, CancellationToken>((func, token) => func(token)); + retry + .Setup(r => r.RunAsync(It.IsAny>>(), + It.IsAny())) + .Returns>, CancellationToken>((func, token) => + func(token)); + return retry; + } +} diff --git a/test/RedShirt.Example.JobWorker.JobManagement.ActiveMq.UnitTests/Tests/Services/Resilience/ActiveMqExceptionArbiterServiceTests.cs b/test/RedShirt.Example.JobWorker.JobManagement.ActiveMq.UnitTests/Tests/Services/Resilience/ActiveMqExceptionArbiterServiceTests.cs new file mode 100644 index 00000000..8bc97f01 --- /dev/null +++ b/test/RedShirt.Example.JobWorker.JobManagement.ActiveMq.UnitTests/Tests/Services/Resilience/ActiveMqExceptionArbiterServiceTests.cs @@ -0,0 +1,249 @@ +using Apache.NMS; +using Apache.NMS.ActiveMQ; +using RedShirt.Example.JobWorker.Core.Exceptions; +using RedShirt.Example.JobWorker.JobManagement.ActiveMq.Exceptions; +using RedShirt.Example.JobWorker.JobManagement.ActiveMq.Services.Resilience; +using System.Net.Sockets; +using ActiveMqIoException = Apache.NMS.ActiveMQ.IOException; + +namespace RedShirt.Example.JobWorker.JobManagement.ActiveMq.UnitTests.Tests.Services.Resilience; + +public class ActiveMqExceptionArbiterServiceTests +{ + private readonly ActiveMqExceptionArbiterService _sut = new(); + + [Fact] + public void GetReport_ArgumentException_IsExpectedAndNotTransient() + { + var report = _sut.GetReport(new ArgumentException("bad queue", "queue")); + + Assert.False(report.AlreadyHandled); + Assert.True(report.IsExpected); + Assert.False(report.CouldBeTransient); + Assert.False(report.CouldBeExternallySolvable); + } + + [Fact] + public void GetReport_ConnectionClosedException_IsExpectedAndTransient() + { + var report = _sut.GetReport(new ConnectionClosedException("connection closed")); + + Assert.False(report.AlreadyHandled); + Assert.True(report.IsExpected); + Assert.True(report.CouldBeTransient); + Assert.True(report.CouldBeExternallySolvable); + } + + [Fact] + public void GetReport_CouldNotLoadQueueException_IsExpectedAndNotTransient() + { + var report = _sut.GetReport(new CouldNotLoadQueueException()); + + Assert.False(report.AlreadyHandled); + Assert.True(report.IsExpected); + Assert.False(report.CouldBeTransient); + Assert.True(report.CouldBeExternallySolvable); + } + + [Fact] + public void GetReport_CouldNotRetrieveMessageBodyException_IsExpectedAndNotTransient() + { + var report = _sut.GetReport(new CouldNotRetrieveMessageBodyException()); + + Assert.False(report.AlreadyHandled); + Assert.True(report.IsExpected); + Assert.False(report.CouldBeTransient); + Assert.False(report.CouldBeExternallySolvable); + } + + [Fact] + public void GetReport_InvalidDestinationException_IsExpectedAndNotTransient() + { + var report = _sut.GetReport(new InvalidDestinationException("no such queue")); + + Assert.False(report.AlreadyHandled); + Assert.True(report.IsExpected); + Assert.False(report.CouldBeTransient); + Assert.True(report.CouldBeExternallySolvable); + } + + [Fact] + public void GetReport_IoException_IsExpectedAndTransient() + { + var report = _sut.GetReport(new ActiveMqIoException("transport failed")); + + Assert.False(report.AlreadyHandled); + Assert.True(report.IsExpected); + Assert.True(report.CouldBeTransient); + Assert.True(report.CouldBeExternallySolvable); + } + + [Fact] + public void GetReport_MultiInnerAggregateException_IsNotExpected() + { + var exception = new AggregateException( + new NMSConnectionException("disconnected"), + new SocketException((int) SocketError.TimedOut)); + + var report = _sut.GetReport(exception); + + Assert.False(report.AlreadyHandled); + Assert.False(report.IsExpected); + Assert.False(report.CouldBeTransient); + Assert.False(report.CouldBeExternallySolvable); + } + + [Fact] + public void GetReport_NmsConnectionException_IsExpectedAndTransient() + { + var report = _sut.GetReport(new NMSConnectionException("broker unavailable")); + + Assert.False(report.AlreadyHandled); + Assert.True(report.IsExpected); + Assert.True(report.CouldBeTransient); + Assert.True(report.CouldBeExternallySolvable); + } + + [Fact] + public void GetReport_NmsException_IsExpectedAndTransient() + { + var report = _sut.GetReport(new NMSException("generic nms failure")); + + Assert.False(report.AlreadyHandled); + Assert.True(report.IsExpected); + Assert.True(report.CouldBeTransient); + Assert.True(report.CouldBeExternallySolvable); + } + + [Fact] + public void GetReport_NmsSecurityException_IsExpectedAndNotTransient() + { + var report = _sut.GetReport(new NMSSecurityException("bad credentials")); + + Assert.False(report.AlreadyHandled); + Assert.True(report.IsExpected); + Assert.False(report.CouldBeTransient); + Assert.True(report.CouldBeExternallySolvable); + } + + [Fact] + public void GetReport_NullException_ThrowsArgumentNullException() + { + Assert.Throws(() => _sut.GetReport(null!)); + } + + [Fact] + public void GetReport_OperationCanceledException_IsExpectedAndNotTransient() + { + var report = _sut.GetReport(new OperationCanceledException("caller cancelled")); + + Assert.False(report.AlreadyHandled); + Assert.True(report.IsExpected); + Assert.False(report.CouldBeTransient); + Assert.False(report.CouldBeExternallySolvable); + } + + [Fact] + public void GetReport_RequestTimedOutException_IsExpectedAndTransient() + { + var report = _sut.GetReport(new RequestTimedOutException(TimeSpan.FromSeconds(1))); + + Assert.False(report.AlreadyHandled); + Assert.True(report.IsExpected); + Assert.True(report.CouldBeTransient); + Assert.True(report.CouldBeExternallySolvable); + } + + [Fact] + public void GetReport_SingleInnerAggregateException_Unwraps() + { + var exception = new AggregateException(new NMSConnectionException("timeout")); + + var report = _sut.GetReport(exception); + + Assert.False(report.AlreadyHandled); + Assert.True(report.IsExpected); + Assert.True(report.CouldBeTransient); + Assert.True(report.CouldBeExternallySolvable); + } + + [Fact] + public void GetReport_SocketException_IsExpectedAndTransient() + { + var report = _sut.GetReport(new SocketException((int) SocketError.TimedOut)); + + Assert.False(report.AlreadyHandled); + Assert.True(report.IsExpected); + Assert.True(report.CouldBeTransient); + Assert.True(report.CouldBeExternallySolvable); + } + + [Fact] + public void GetReport_TaskCanceledException_IsExpectedAndTransient() + { + var report = _sut.GetReport(new TaskCanceledException()); + + Assert.False(report.AlreadyHandled); + Assert.True(report.IsExpected); + Assert.True(report.CouldBeTransient); + Assert.True(report.CouldBeExternallySolvable); + } + + [Fact] + public void GetReport_TimeoutException_IsExpectedAndTransient() + { + var report = _sut.GetReport(new TimeoutException("timed out")); + + Assert.False(report.AlreadyHandled); + Assert.True(report.IsExpected); + Assert.True(report.CouldBeTransient); + Assert.True(report.CouldBeExternallySolvable); + } + + [Fact] + public void GetReport_UnrecognizedException_IsNotExpected() + { + var report = _sut.GetReport(new InvalidOperationException("boom")); + + Assert.False(report.AlreadyHandled); + Assert.False(report.IsExpected); + Assert.False(report.CouldBeTransient); + Assert.False(report.CouldBeExternallySolvable); + } + + [Fact] + public void GetReport_WorkerJobSourceException_Handled_DoesNotRetry() + { + var exception = new WorkerJobSourceException("already handled") + { + IsHandled = true, + CouldBeTransient = true, + CouldBeExternallySolvable = true + }; + + var report = _sut.GetReport(exception); + + Assert.True(report.AlreadyHandled); + Assert.True(report.IsExpected); + Assert.False(report.CouldBeTransient); + Assert.True(report.CouldBeExternallySolvable); + } + + [Fact] + public void GetReport_WorkerJobSourceException_UnhandledTransient_MayRetry() + { + var exception = new WorkerJobSourceException("transient") + { + IsHandled = false, + CouldBeTransient = true, + CouldBeExternallySolvable = true + }; + + var report = _sut.GetReport(exception); + + Assert.True(report.AlreadyHandled); + Assert.True(report.IsExpected); + Assert.True(report.CouldBeTransient); + Assert.True(report.CouldBeExternallySolvable); + } +} diff --git a/test/RedShirt.Example.JobWorker.JobManagement.ActiveMq.UnitTests/Tests/Services/Resilience/ActiveMqRetryWrapperServiceTests.cs b/test/RedShirt.Example.JobWorker.JobManagement.ActiveMq.UnitTests/Tests/Services/Resilience/ActiveMqRetryWrapperServiceTests.cs new file mode 100644 index 00000000..ac96ddec --- /dev/null +++ b/test/RedShirt.Example.JobWorker.JobManagement.ActiveMq.UnitTests/Tests/Services/Resilience/ActiveMqRetryWrapperServiceTests.cs @@ -0,0 +1,307 @@ +using Microsoft.Extensions.Logging.Abstractions; +using RedShirt.Example.JobWorker.Common.Services.Utility; +using RedShirt.Example.JobWorker.Core.Exceptions; +using RedShirt.Example.JobWorker.JobManagement.ActiveMq.Models; +using RedShirt.Example.JobWorker.JobManagement.ActiveMq.Services.Resilience; + +namespace RedShirt.Example.JobWorker.JobManagement.ActiveMq.UnitTests.Tests.Services.Resilience; + +public class ActiveMqRetryWrapperServiceTests +{ + private static ActiveMqExceptionArbiterReport TransientReport() + { + return new ActiveMqExceptionArbiterReport + { + AlreadyHandled = false, + IsExpected = true, + CouldBeTransient = true, + CouldBeExternallySolvable = true + }; + } + + private static ActiveMqExceptionArbiterReport PermanentReport() + { + return new ActiveMqExceptionArbiterReport + { + AlreadyHandled = false, + IsExpected = true, + CouldBeTransient = false, + CouldBeExternallySolvable = false + }; + } + + private static ActiveMqExceptionArbiterReport CriticalReport() + { + return new ActiveMqExceptionArbiterReport + { + AlreadyHandled = false, + IsExpected = false, + CouldBeTransient = false, + CouldBeExternallySolvable = false + }; + } + + private static ActiveMqExceptionArbiterReport AlreadyHandledReport(bool couldBeTransient) + { + return new ActiveMqExceptionArbiterReport + { + AlreadyHandled = true, + IsExpected = true, + CouldBeTransient = couldBeTransient, + CouldBeExternallySolvable = false + }; + } + + private static Mock CreateSleepService(IList? capturedDelays = null) + { + var sleep = new Mock(MockBehavior.Strict); + sleep.Setup(s => s.DelayAsync(It.IsAny(), It.IsAny())) + .Returns((delay, _) => + { + capturedDelays?.Add(delay); + return Task.CompletedTask; + }); + return sleep; + } + + [Fact] + public async Task RunAsync_NonGeneric_WhenFuncSucceeds_CompletesWithoutSleeping() + { + var arbiter = new Mock(MockBehavior.Strict); + var sleep = CreateSleepService(); + var wrapper = new ActiveMqRetryWrapperService(arbiter.Object, + NullLogger.Instance, + sleep.Object); + var ran = false; + + await wrapper.RunAsync(_ => + { + ran = true; + return Task.CompletedTask; + }, TestContext.Current.CancellationToken); + + Assert.True(ran); + arbiter.VerifyNoOtherCalls(); + Assert.Empty(sleep.Invocations); + } + + [Fact] + public async Task RunAsync_NonGeneric_WhenOperationCanceledAndTokenCancelled_PropagatesWithoutWrapping() + { + using var cts = new CancellationTokenSource(); + await cts.CancelAsync(); + + var arbiter = new Mock(MockBehavior.Strict); + var sleep = CreateSleepService(); + var wrapper = new ActiveMqRetryWrapperService(arbiter.Object, + NullLogger.Instance, + sleep.Object); + + await Assert.ThrowsAnyAsync(() => wrapper.RunAsync( + _ => throw new OperationCanceledException(cts.Token), + cts.Token)); + + arbiter.VerifyNoOtherCalls(); + Assert.Empty(sleep.Invocations); + } + + [Fact] + public async Task RunAsync_NonGeneric_WhenTransientFailuresExhaustRetries_Wraps() + { + var attempts = 0; + var inner = new TimeoutException("still failing"); + + var arbiter = new Mock(MockBehavior.Strict); + arbiter.Setup(a => a.GetReport(It.IsAny())).Returns(TransientReport()); + + var sleep = CreateSleepService(); + var wrapper = new ActiveMqRetryWrapperService(arbiter.Object, + NullLogger.Instance, + sleep.Object); + + var thrown = await Assert.ThrowsAsync(() => wrapper.RunAsync( + _ => + { + attempts++; + throw inner; + }, + TestContext.Current.CancellationToken)); + + Assert.Same(inner, thrown.InnerException); + Assert.True(thrown.IsHandled); + Assert.True(thrown.CouldBeExternallySolvable); + Assert.Equal(4, attempts); + } + + [Fact] + public async Task RunAsync_WhenAlreadyHandled_RethrowsWithoutWrapping() + { + var inner = new WorkerJobSourceException("already wrapped") + {CouldBeTransient = false, IsHandled = true, CouldBeExternallySolvable = false}; + + var arbiter = new Mock(MockBehavior.Strict); + arbiter.Setup(a => a.GetReport(inner)).Returns(AlreadyHandledReport(false)); + + var sleep = CreateSleepService(); + var wrapper = new ActiveMqRetryWrapperService(arbiter.Object, + NullLogger.Instance, + sleep.Object); + + var thrown = await Assert.ThrowsAsync(() => + wrapper.RunAsync(_ => throw inner, TestContext.Current.CancellationToken)); + + Assert.Same(inner, thrown); + Assert.Empty(sleep.Invocations); + } + + [Fact] + public async Task RunAsync_WhenCriticalFailure_ThrowsRaw() + { + var inner = new InvalidOperationException("critical"); + + var arbiter = new Mock(MockBehavior.Strict); + arbiter.Setup(a => a.GetReport(inner)).Returns(CriticalReport()); + + var sleep = CreateSleepService(); + var wrapper = new ActiveMqRetryWrapperService(arbiter.Object, + NullLogger.Instance, + sleep.Object); + + var thrown = await Assert.ThrowsAsync(() => + wrapper.RunAsync(_ => throw inner, TestContext.Current.CancellationToken)); + + Assert.Same(inner, thrown); + Assert.Empty(sleep.Invocations); + } + + [Fact] + public async Task RunAsync_WhenFuncSucceeds_ReturnsResultWithoutSleeping() + { + var arbiter = new Mock(MockBehavior.Strict); + var sleep = CreateSleepService(); + var wrapper = new ActiveMqRetryWrapperService(arbiter.Object, + NullLogger.Instance, + sleep.Object); + + var result = await wrapper.RunAsync(_ => Task.FromResult(42), TestContext.Current.CancellationToken); + + Assert.Equal(42, result); + arbiter.VerifyNoOtherCalls(); + Assert.Empty(sleep.Invocations); + } + + [Fact] + public async Task RunAsync_WhenOperationCanceledAndTokenCancelled_PropagatesWithoutWrapping() + { + using var cts = new CancellationTokenSource(); + await cts.CancelAsync(); + + var arbiter = new Mock(MockBehavior.Strict); + var sleep = CreateSleepService(); + var wrapper = new ActiveMqRetryWrapperService(arbiter.Object, + NullLogger.Instance, + sleep.Object); + + await Assert.ThrowsAnyAsync(() => wrapper.RunAsync( + _ => throw new OperationCanceledException(cts.Token), + cts.Token)); + + arbiter.VerifyNoOtherCalls(); + Assert.Empty(sleep.Invocations); + } + + [Fact] + public async Task RunAsync_WhenPermanentFailure_WrapsWithoutRetry() + { + var attempts = 0; + var inner = new ArgumentException("bad"); + + var arbiter = new Mock(MockBehavior.Strict); + arbiter.Setup(a => a.GetReport(inner)).Returns(PermanentReport()); + + var sleep = CreateSleepService(); + var wrapper = new ActiveMqRetryWrapperService(arbiter.Object, + NullLogger.Instance, + sleep.Object); + + var thrown = await Assert.ThrowsAsync(() => wrapper.RunAsync( + _ => + { + attempts++; + throw inner; + }, + TestContext.Current.CancellationToken)); + + Assert.Equal(1, attempts); + Assert.Same(inner, thrown.InnerException); + Assert.False(thrown.CouldBeTransient); + Assert.True(thrown.IsHandled); + Assert.False(thrown.CouldBeExternallySolvable); + Assert.Empty(sleep.Invocations); + } + + [Fact] + public async Task RunAsync_WhenTransientFailuresExhaustRetries_WrapsAsWorkerJobSourceException() + { + var attempts = 0; + var delays = new List(); + var inner = new TimeoutException("still failing"); + + var arbiter = new Mock(MockBehavior.Strict); + arbiter.Setup(a => a.GetReport(It.IsAny())).Returns(TransientReport()); + + var sleep = CreateSleepService(delays); + var wrapper = new ActiveMqRetryWrapperService(arbiter.Object, + NullLogger.Instance, + sleep.Object); + + var thrown = await Assert.ThrowsAsync(() => wrapper.RunAsync( + _ => + { + attempts++; + throw inner; + }, + TestContext.Current.CancellationToken)); + + Assert.Same(inner, thrown.InnerException); + Assert.True(thrown.CouldBeTransient); + Assert.True(thrown.IsHandled); + Assert.True(thrown.CouldBeExternallySolvable); + Assert.Equal(4, attempts); + Assert.Equal( + [TimeSpan.FromSeconds(1), TimeSpan.FromSeconds(2), TimeSpan.FromSeconds(4)], + delays); + } + + [Fact] + public async Task RunAsync_WhenTransientThenSucceeds_RetriesWithBackoff() + { + var attempts = 0; + var delays = new List(); + + var arbiter = new Mock(MockBehavior.Strict); + arbiter.Setup(a => a.GetReport(It.IsAny())).Returns(TransientReport()); + + var sleep = CreateSleepService(delays); + var wrapper = new ActiveMqRetryWrapperService(arbiter.Object, + NullLogger.Instance, + sleep.Object); + + var result = await wrapper.RunAsync( + _ => + { + attempts++; + if (attempts == 1) + { + throw new TimeoutException("timeout"); + } + + return Task.FromResult("ok"); + }, + TestContext.Current.CancellationToken); + + Assert.Equal("ok", result); + Assert.Equal(2, attempts); + Assert.Equal([TimeSpan.FromSeconds(1)], delays); + } +} From 9bb32fbe4877d5a7027b9681d95edc01ea9fc63b Mon Sep 17 00:00:00 2001 From: Alan Deutscher Date: Thu, 20 Aug 2026 23:56:14 -0700 Subject: [PATCH 2/5] Refine application of retry wrapper --- .../Services/ActiveMqJobSource.cs | 56 +++++++++---------- .../Tests/Services/ActiveMqJobSourceTests.cs | 21 ++++--- .../Services/ActiveMqRetryTestHelpers.cs | 21 +++++-- 3 files changed, 56 insertions(+), 42 deletions(-) diff --git a/src/RedShirt.Example.JobWorker.JobManagement.ActiveMq/Services/ActiveMqJobSource.cs b/src/RedShirt.Example.JobWorker.JobManagement.ActiveMq/Services/ActiveMqJobSource.cs index 8664d9d3..7cf87d1d 100644 --- a/src/RedShirt.Example.JobWorker.JobManagement.ActiveMq/Services/ActiveMqJobSource.cs +++ b/src/RedShirt.Example.JobWorker.JobManagement.ActiveMq/Services/ActiveMqJobSource.cs @@ -11,24 +11,27 @@ namespace RedShirt.Example.JobWorker.JobManagement.ActiveMq.Services; -internal class ActiveMqJobSource : IJobSource +internal class ActiveMqJobSource( + IActiveMqConnectionFactory connectionFactory, + IActiveMqRetryWrapperService retryWrapperService, + IOptions configuration, + ILogger logger) + : IJobSource { - private readonly IOptions _configuration; - private readonly IActiveMqConnectionFactory _connectionFactory; - private readonly ILogger _logger; - private readonly IActiveMqRetryWrapperService _retryWrapperService; private IMessageConsumer? _messageConsumer; private async Task FetchJobsAsync(int batchSize, CancellationToken cancellationToken) { try { - var consumer = await GetConsumerAsync(cancellationToken); + var consumer = await retryWrapperService.RunAsync(GetConsumerAsync, cancellationToken); var getJobsResponseItems = new List(); while (getJobsResponseItems.Count < batchSize) { - var result = await consumer.ReceiveAsync(TimeSpan.FromMilliseconds(100)); + var result = + await retryWrapperService.RunAsync(_ => consumer.ReceiveAsync(TimeSpan.FromMilliseconds(100)), + cancellationToken); if (result is null) // Nothing more to grab at the moment. @@ -52,11 +55,18 @@ private async Task FetchJobsAsync(int batchSize, Cancellation } catch { - Reset(); + ResetConsumer(); throw; } } + /// + /// Get a cached consumer or get a new one from the connection factory. + /// Confirming that the invocation of this method should be already covered by the retry wrapper service. + /// + /// + /// + /// private async Task GetConsumerAsync(CancellationToken cancellationToken) { if (_messageConsumer is not null) @@ -64,10 +74,10 @@ private async Task GetConsumerAsync(CancellationToken cancella return _messageConsumer; } - var connection = await _connectionFactory.GetConnectionAsync(cancellationToken); + var connection = await connectionFactory.GetConnectionAsync(cancellationToken); await connection.StartAsync(); var session = await connection.CreateSessionAsync(AcknowledgementMode.ClientAcknowledge); - var queue = await session.GetQueueAsync(_configuration.Value.QueueName); + var queue = await session.GetQueueAsync(configuration.Value.QueueName); if (queue is null) { @@ -75,26 +85,18 @@ private async Task GetConsumerAsync(CancellationToken cancella } var consumer = await session.CreateConsumerAsync(queue); + + // Cache for later _messageConsumer = consumer; + return consumer; } - private void Reset() + private void ResetConsumer() { _messageConsumer = null; } - public ActiveMqJobSource(IActiveMqConnectionFactory connectionFactory, - IActiveMqRetryWrapperService retryWrapperService, - IOptions configuration, - ILogger logger) - { - _connectionFactory = connectionFactory; - _retryWrapperService = retryWrapperService; - _configuration = configuration; - _logger = logger; - } - public int RecommendedHeartbeatIntervalSeconds => 0; public bool IsSubscriptionSource => false; @@ -114,7 +116,7 @@ public async Task AcknowledgeAsync(IRawJobModel message, CoreJobResult result, // Acknowledge whether successful, recoverable, or unrecoverable // (ActiveMQ client API has no direct dead-letter call here). - await _retryWrapperService.RunAsync( + await retryWrapperService.RunAsync( _ => jobModel.Message.AcknowledgeAsync(), cancellationToken); } @@ -123,12 +125,10 @@ public async Task GetJobsAsync(int batchSize, CancellationTo { batchSize = Math.Max(1, batchSize); - _logger.LogTrace("Fetching up to {EffectiveBatchSize} messages from ActiveMQ Queue: {QueueName}", - batchSize, _configuration.Value.QueueName); + logger.LogTrace("Fetching up to {EffectiveBatchSize} messages from ActiveMQ Queue: {QueueName}", + batchSize, configuration.Value.QueueName); - return await _retryWrapperService.RunAsync( - ct => FetchJobsAsync(batchSize, ct), - cancellationToken); + return await FetchJobsAsync(batchSize, cancellationToken); } public Task HeartbeatAsync(IRawJobModel message, CancellationToken cancellationToken = default) diff --git a/test/RedShirt.Example.JobWorker.JobManagement.ActiveMq.UnitTests/Tests/Services/ActiveMqJobSourceTests.cs b/test/RedShirt.Example.JobWorker.JobManagement.ActiveMq.UnitTests/Tests/Services/ActiveMqJobSourceTests.cs index b778d2b9..7f0ebf7c 100644 --- a/test/RedShirt.Example.JobWorker.JobManagement.ActiveMq.UnitTests/Tests/Services/ActiveMqJobSourceTests.cs +++ b/test/RedShirt.Example.JobWorker.JobManagement.ActiveMq.UnitTests/Tests/Services/ActiveMqJobSourceTests.cs @@ -116,7 +116,8 @@ public async Task Test_GetJobs_GetNoJobs() .ReturnsAsync(consumer.Object); var mockConnection = new Mock(MockBehavior.Strict); - mockConnection.Setup(c => c.Start()); + mockConnection.Setup(c => c.StartAsync()) + .Returns(Task.CompletedTask); mockConnection.Setup(c => c.CreateSessionAsync(AcknowledgementMode.ClientAcknowledge)) .ReturnsAsync(mockSession.Object); @@ -137,7 +138,7 @@ public async Task Test_GetJobs_GetNoJobs() Assert.Single(activeConnectionFactory.Invocations); Assert.Equal(2, mockConnection.Invocations.Count); - mockConnection.Verify(c => c.Start(), Times.Once); + mockConnection.Verify(c => c.StartAsync(), Times.Once); mockConnection.Verify(c => c.CreateSessionAsync(AcknowledgementMode.ClientAcknowledge), Times.Once); Assert.Equal(2, mockSession.Invocations.Count); mockSession.Verify(s => s.GetQueueAsync(queueName), Times.Once); @@ -159,7 +160,8 @@ public async Task Test_GetJobs_GetNoQueue() .ReturnsAsync((IQueue?) null); var mockConnection = new Mock(MockBehavior.Strict); - mockConnection.Setup(c => c.Start()); + mockConnection.Setup(c => c.StartAsync()) + .Returns(Task.CompletedTask); mockConnection.Setup(c => c.CreateSessionAsync(AcknowledgementMode.ClientAcknowledge)) .ReturnsAsync(mockSession.Object); @@ -174,7 +176,7 @@ await Assert.ThrowsAsync(() => Assert.Single(activeConnectionFactory.Invocations); Assert.Equal(2, mockConnection.Invocations.Count); - mockConnection.Verify(c => c.Start(), Times.Once); + mockConnection.Verify(c => c.StartAsync(), Times.Once); mockConnection.Verify(c => c.CreateSessionAsync(AcknowledgementMode.ClientAcknowledge), Times.Once); Assert.Single(mockSession.Invocations); mockSession.Verify(s => s.GetQueueAsync(queueName), Times.Once); @@ -201,7 +203,8 @@ public async Task Test_GetJobs_GotJob() .ReturnsAsync(consumer.Object); var mockConnection = new Mock(MockBehavior.Strict); - mockConnection.Setup(c => c.Start()); + mockConnection.Setup(c => c.StartAsync()) + .Returns(Task.CompletedTask); mockConnection.Setup(c => c.CreateSessionAsync(AcknowledgementMode.ClientAcknowledge)) .ReturnsAsync(mockSession.Object); @@ -232,7 +235,7 @@ public async Task Test_GetJobs_GotJob() Assert.Single(activeConnectionFactory.Invocations); Assert.Equal(2, mockConnection.Invocations.Count); - mockConnection.Verify(c => c.Start(), Times.Once); + mockConnection.Verify(c => c.StartAsync(), Times.Once); mockConnection.Verify(c => c.CreateSessionAsync(AcknowledgementMode.ClientAcknowledge), Times.Once); Assert.Equal(2, mockSession.Invocations.Count); mockSession.Verify(s => s.GetQueueAsync(queueName), Times.Once); @@ -260,7 +263,8 @@ public async Task Test_GetJobs_GotJob_BatchSizeZero() .ReturnsAsync(consumer.Object); var mockConnection = new Mock(MockBehavior.Strict); - mockConnection.Setup(c => c.Start()); + mockConnection.Setup(c => c.StartAsync()) + .Returns(Task.CompletedTask); mockConnection.Setup(c => c.CreateSessionAsync(AcknowledgementMode.ClientAcknowledge)) .ReturnsAsync(mockSession.Object); @@ -314,7 +318,8 @@ public async Task Test_GetJobs_GotJobs_MultipleJobs(int batchSize) .ReturnsAsync(consumer.Object); var mockConnection = new Mock(MockBehavior.Strict); - mockConnection.Setup(c => c.Start()); + mockConnection.Setup(c => c.StartAsync()) + .Returns(Task.CompletedTask); mockConnection.Setup(c => c.CreateSessionAsync(AcknowledgementMode.ClientAcknowledge)) .ReturnsAsync(mockSession.Object); diff --git a/test/RedShirt.Example.JobWorker.JobManagement.ActiveMq.UnitTests/Tests/Services/ActiveMqRetryTestHelpers.cs b/test/RedShirt.Example.JobWorker.JobManagement.ActiveMq.UnitTests/Tests/Services/ActiveMqRetryTestHelpers.cs index 872b0b54..f31b453b 100644 --- a/test/RedShirt.Example.JobWorker.JobManagement.ActiveMq.UnitTests/Tests/Services/ActiveMqRetryTestHelpers.cs +++ b/test/RedShirt.Example.JobWorker.JobManagement.ActiveMq.UnitTests/Tests/Services/ActiveMqRetryTestHelpers.cs @@ -1,3 +1,4 @@ +using Apache.NMS; using RedShirt.Example.JobWorker.Core.Models; using RedShirt.Example.JobWorker.JobManagement.ActiveMq.Services.Resilience; @@ -5,17 +6,25 @@ namespace RedShirt.Example.JobWorker.JobManagement.ActiveMq.UnitTests.Tests.Serv internal static class ActiveMqRetryTestHelpers { + private static void SetupPassthrough(Mock retry) + { + retry + .Setup(r => r.RunAsync(It.IsAny>>(), It.IsAny())) + .Returns>, CancellationToken>((func, token) => func(token)); + } + public static Mock CreatePassthroughRetryWrapper() { var retry = new Mock(MockBehavior.Strict); retry .Setup(r => r.RunAsync(It.IsAny>(), It.IsAny())) .Returns, CancellationToken>((func, token) => func(token)); - retry - .Setup(r => r.RunAsync(It.IsAny>>(), - It.IsAny())) - .Returns>, CancellationToken>((func, token) => - func(token)); + + // ActiveMqJobSource wraps GetConsumerAsync / ReceiveAsync. + SetupPassthrough(retry); + SetupPassthrough(retry); + SetupPassthrough(retry); + return retry; } -} +} \ No newline at end of file From d74736464f9c55427f6a2fa921ad6610e0742bbb Mon Sep 17 00:00:00 2001 From: Alan Deutscher Date: Thu, 20 Aug 2026 23:59:16 -0700 Subject: [PATCH 3/5] Handle WorkerSecretManagerException --- .../ActiveMqExceptionArbiterService.cs | 10 +++++-- .../ActiveMqExceptionArbiterServiceTests.cs | 28 ++++++++++++++++++- 2 files changed, 35 insertions(+), 3 deletions(-) diff --git a/src/RedShirt.Example.JobWorker.JobManagement.ActiveMq/Services/Resilience/ActiveMqExceptionArbiterService.cs b/src/RedShirt.Example.JobWorker.JobManagement.ActiveMq/Services/Resilience/ActiveMqExceptionArbiterService.cs index f77c4a5a..569f8587 100644 --- a/src/RedShirt.Example.JobWorker.JobManagement.ActiveMq/Services/Resilience/ActiveMqExceptionArbiterService.cs +++ b/src/RedShirt.Example.JobWorker.JobManagement.ActiveMq/Services/Resilience/ActiveMqExceptionArbiterService.cs @@ -1,10 +1,12 @@ using Apache.NMS; using Apache.NMS.ActiveMQ; +using RedShirt.Example.JobWorker.Common.SecretManagers.Core.Exceptions; using RedShirt.Example.JobWorker.Core.Exceptions; using RedShirt.Example.JobWorker.JobManagement.ActiveMq.Exceptions; using RedShirt.Example.JobWorker.JobManagement.ActiveMq.Models; using System.Net.Sockets; using ActiveMqIoException = Apache.NMS.ActiveMQ.IOException; +using IOException = System.IO.IOException; namespace RedShirt.Example.JobWorker.JobManagement.ActiveMq.Services.Resilience; @@ -68,6 +70,10 @@ public ActiveMqExceptionArbiterReport GetReport(Exception exception) true, workerJobSource is {IsHandled: false, CouldBeTransient: true}, workerJobSource.CouldBeExternallySolvable), + // Secret-manager failures (e.g. credential fetch) — already wrapped; propagate the + // secret layer's transient / externally-solvable classification for upstream decisions. + WorkerSecretManagerException workerSecretManager => + Handled(true, workerSecretManager.CouldBeTransient, workerSecretManager.CouldBeExternallySolvable), // Queue lookup returned null — ops can create the destination without a worker restart. CouldNotLoadQueueException => Fresh(true, false, true), // Unsupported / unreadable payload — a local data issue, not retryable. @@ -103,7 +109,7 @@ or ConsumerClosedException NMSException => Fresh(true, true, true), TimeoutException or SocketException - or System.IO.IOException => Fresh(true, true, true), + or IOException => Fresh(true, true, true), // HttpClient-style timeouts sometimes surface as TaskCanceledException. // Must be matched before OperationCanceledException (TCE derives from OCE). TaskCanceledException => Fresh(true, true, true), @@ -115,4 +121,4 @@ or SocketException _ => Fresh(false, false, false) }; } -} +} \ No newline at end of file diff --git a/test/RedShirt.Example.JobWorker.JobManagement.ActiveMq.UnitTests/Tests/Services/Resilience/ActiveMqExceptionArbiterServiceTests.cs b/test/RedShirt.Example.JobWorker.JobManagement.ActiveMq.UnitTests/Tests/Services/Resilience/ActiveMqExceptionArbiterServiceTests.cs index 8bc97f01..cc2a7c5c 100644 --- a/test/RedShirt.Example.JobWorker.JobManagement.ActiveMq.UnitTests/Tests/Services/Resilience/ActiveMqExceptionArbiterServiceTests.cs +++ b/test/RedShirt.Example.JobWorker.JobManagement.ActiveMq.UnitTests/Tests/Services/Resilience/ActiveMqExceptionArbiterServiceTests.cs @@ -1,5 +1,6 @@ using Apache.NMS; using Apache.NMS.ActiveMQ; +using RedShirt.Example.JobWorker.Common.SecretManagers.Core.Exceptions; using RedShirt.Example.JobWorker.Core.Exceptions; using RedShirt.Example.JobWorker.JobManagement.ActiveMq.Exceptions; using RedShirt.Example.JobWorker.JobManagement.ActiveMq.Services.Resilience; @@ -246,4 +247,29 @@ public void GetReport_WorkerJobSourceException_UnhandledTransient_MayRetry() Assert.True(report.CouldBeTransient); Assert.True(report.CouldBeExternallySolvable); } -} + + [Theory] + [InlineData(true, true, true)] + [InlineData(true, false, false)] + [InlineData(false, true, true)] + [InlineData(false, false, false)] + public void GetReport_WorkerSecretManagerException_IsAlreadyHandledWithFlags( + bool isHandled, + bool couldBeTransient, + bool couldBeExternallySolvable) + { + var exception = new WorkerSecretManagerException("secret failure") + { + IsHandled = isHandled, + CouldBeTransient = couldBeTransient, + CouldBeExternallySolvable = couldBeExternallySolvable + }; + + var report = _sut.GetReport(exception); + + Assert.True(report.AlreadyHandled); + Assert.True(report.IsExpected); + Assert.Equal(couldBeTransient, report.CouldBeTransient); + Assert.Equal(couldBeExternallySolvable, report.CouldBeExternallySolvable); + } +} \ No newline at end of file From d6bb95b5450aa59cd602fa6cee6cd28ef57e0e1b Mon Sep 17 00:00:00 2001 From: Alan Deutscher Date: Thu, 20 Aug 2026 23:59:56 -0700 Subject: [PATCH 4/5] cleanup sweep of ActiveMQ projects --- .../Models/ActiveMqExceptionArbiterReport.cs | 2 +- .../Services/Resilience/ActiveMqRetryWrapperService.cs | 2 +- .../Tests/Extensions/ServiceCollectionExtensionsTests.cs | 2 +- .../Services/Resilience/ActiveMqRetryWrapperServiceTests.cs | 2 +- 4 files changed, 4 insertions(+), 4 deletions(-) diff --git a/src/RedShirt.Example.JobWorker.JobManagement.ActiveMq/Models/ActiveMqExceptionArbiterReport.cs b/src/RedShirt.Example.JobWorker.JobManagement.ActiveMq/Models/ActiveMqExceptionArbiterReport.cs index 1b6d8a74..9af261e5 100644 --- a/src/RedShirt.Example.JobWorker.JobManagement.ActiveMq/Models/ActiveMqExceptionArbiterReport.cs +++ b/src/RedShirt.Example.JobWorker.JobManagement.ActiveMq/Models/ActiveMqExceptionArbiterReport.cs @@ -6,4 +6,4 @@ internal sealed class ActiveMqExceptionArbiterReport public required bool IsExpected { get; init; } public required bool CouldBeExternallySolvable { get; init; } public required bool CouldBeTransient { get; init; } -} +} \ No newline at end of file diff --git a/src/RedShirt.Example.JobWorker.JobManagement.ActiveMq/Services/Resilience/ActiveMqRetryWrapperService.cs b/src/RedShirt.Example.JobWorker.JobManagement.ActiveMq/Services/Resilience/ActiveMqRetryWrapperService.cs index 76e9af5e..760f0f13 100644 --- a/src/RedShirt.Example.JobWorker.JobManagement.ActiveMq/Services/Resilience/ActiveMqRetryWrapperService.cs +++ b/src/RedShirt.Example.JobWorker.JobManagement.ActiveMq/Services/Resilience/ActiveMqRetryWrapperService.cs @@ -207,4 +207,4 @@ await GetRetryPipeline().ExecuteAsync( throw; } } -} +} \ No newline at end of file diff --git a/test/RedShirt.Example.JobWorker.JobManagement.ActiveMq.UnitTests/Tests/Extensions/ServiceCollectionExtensionsTests.cs b/test/RedShirt.Example.JobWorker.JobManagement.ActiveMq.UnitTests/Tests/Extensions/ServiceCollectionExtensionsTests.cs index 4cc6a938..c58df5df 100644 --- a/test/RedShirt.Example.JobWorker.JobManagement.ActiveMq.UnitTests/Tests/Extensions/ServiceCollectionExtensionsTests.cs +++ b/test/RedShirt.Example.JobWorker.JobManagement.ActiveMq.UnitTests/Tests/Extensions/ServiceCollectionExtensionsTests.cs @@ -29,4 +29,4 @@ public void AddActiveMqJobManagement_RegistersExpectedServices() Assert.Contains(services, d => d.ServiceType == typeof(IActiveMqExceptionArbiterService)); Assert.Contains(services, d => d.ServiceType == typeof(IActiveMqRetryWrapperService)); } -} +} \ No newline at end of file diff --git a/test/RedShirt.Example.JobWorker.JobManagement.ActiveMq.UnitTests/Tests/Services/Resilience/ActiveMqRetryWrapperServiceTests.cs b/test/RedShirt.Example.JobWorker.JobManagement.ActiveMq.UnitTests/Tests/Services/Resilience/ActiveMqRetryWrapperServiceTests.cs index ac96ddec..776c63b4 100644 --- a/test/RedShirt.Example.JobWorker.JobManagement.ActiveMq.UnitTests/Tests/Services/Resilience/ActiveMqRetryWrapperServiceTests.cs +++ b/test/RedShirt.Example.JobWorker.JobManagement.ActiveMq.UnitTests/Tests/Services/Resilience/ActiveMqRetryWrapperServiceTests.cs @@ -304,4 +304,4 @@ public async Task RunAsync_WhenTransientThenSucceeds_RetriesWithBackoff() Assert.Equal(2, attempts); Assert.Equal([TimeSpan.FromSeconds(1)], delays); } -} +} \ No newline at end of file From 15ffcacb0cb5208865eec113dcae380e3a146c0e Mon Sep 17 00:00:00 2001 From: Alan Deutscher Date: Fri, 21 Aug 2026 00:39:31 -0700 Subject: [PATCH 5/5] Revise local ActiveMQ testing documentation, disk usage tolerance --- test/local/docker-compose.yaml | 28 +++++ test/local/make-local-activemq-resources.py | 118 ++++++++++++++++++ test/local/readme.md | 66 ++++++---- test/local/send-activemq-message.py | 131 +++++++++++++++++--- 4 files changed, 299 insertions(+), 44 deletions(-) create mode 100755 test/local/make-local-activemq-resources.py diff --git a/test/local/docker-compose.yaml b/test/local/docker-compose.yaml index 86e49e97..b40391b6 100644 --- a/test/local/docker-compose.yaml +++ b/test/local/docker-compose.yaml @@ -19,6 +19,34 @@ services: environment: ARTEMIS_USER: admin ARTEMIS_PASSWORD: admin + # Raise max-disk-usage so local publishes still work on nearly-full developer disks + # (stock Artemis defaults to 90% and blocks all producers beyond that). + entrypoint: ["/bin/bash", "-c"] + command: + - | + set -euo pipefail + cd /var/lib/artemis-instance + if [[ "$${ANONYMOUS_LOGIN,,}" == "true" ]]; then + LOGIN_OPTION="--allow-anonymous" + else + LOGIN_OPTION="--require-login" + fi + if ! [ -f ./etc/broker.xml ]; then + /opt/activemq-artemis/bin/artemis create \ + --user "$${ARTEMIS_USER}" \ + --password "$${ARTEMIS_PASSWORD}" \ + --silent \ + $${LOGIN_OPTION} \ + $${EXTRA_ARGS:-} \ + . + if [ -d ./etc-override ]; then + for file in $$(ls ./etc-override); do + cp "./etc-override/$$file" ./etc || true + done + fi + fi + sed -i 's#[0-9]*#99#' ./etc/broker.xml + exec ./bin/artemis run nats: image: nats diff --git a/test/local/make-local-activemq-resources.py b/test/local/make-local-activemq-resources.py new file mode 100755 index 00000000..a5ec6537 --- /dev/null +++ b/test/local/make-local-activemq-resources.py @@ -0,0 +1,118 @@ +#!/usr/bin/env python3 + +"""Create the local ActiveMQ Artemis address/queue used by the job worker. + +Uses the Artemis Jolokia HTTP management API (stdlib only; no extra packages). +Safe to re-run if the queue already exists. +""" + +import base64 +import json +import sys +import time +import urllib.error +import urllib.request + +JOLOKIA_URL = "http://localhost:8161/console/jolokia/" +BROKER_MBEAN = 'org.apache.activemq.artemis:broker="0.0.0.0"' +USERNAME = "admin" +PASSWORD = "admin" +QUEUE_NAME = "/queue/ActiveQueue" + + +def _auth_header() -> str: + token = base64.b64encode(f"{USERNAME}:{PASSWORD}".encode()).decode() + return f"Basic {token}" + + +def jolokia_exec(operation: str, arguments: list) -> dict: + payload = json.dumps( + { + "type": "exec", + "mbean": BROKER_MBEAN, + "operation": operation, + "arguments": arguments, + } + ).encode() + request = urllib.request.Request( + JOLOKIA_URL, + data=payload, + method="POST", + headers={ + "Content-Type": "application/json", + "Origin": "http://localhost", + "Authorization": _auth_header(), + }, + ) + with urllib.request.urlopen(request, timeout=10) as response: + return json.loads(response.read().decode()) + + +def wait_for_jolokia(timeout_seconds: int = 120) -> None: + deadline = time.time() + timeout_seconds + last_error = None + while time.time() < deadline: + try: + request = urllib.request.Request( + f"{JOLOKIA_URL}version", + headers={ + "Origin": "http://localhost", + "Authorization": _auth_header(), + }, + ) + with urllib.request.urlopen(request, timeout=2) as response: + if response.status == 200: + return + except Exception as exc: # noqa: BLE001 - local bootstrap helper + last_error = exc + time.sleep(2) + raise RuntimeError(f"Artemis Jolokia was not ready at {JOLOKIA_URL}: {last_error}") + + +def ensure_queue() -> None: + queue_config = json.dumps( + { + "name": QUEUE_NAME, + "address": QUEUE_NAME, + "routing-type": "ANYCAST", + "durable": True, + } + ) + try: + result = jolokia_exec( + "createQueue(java.lang.String,boolean)", + [queue_config, True], + ) + except urllib.error.HTTPError as exc: + body = exc.read().decode("utf-8", errors="replace") + if "already exists" in body.lower(): + print(f"Queue '{QUEUE_NAME}' already exists.") + return + raise RuntimeError(f"Failed to create queue '{QUEUE_NAME}': {body}") from exc + + if result.get("status") == 200: + value = result.get("value") + if isinstance(value, str) and "id" not in value and QUEUE_NAME in value: + # ignoreIfExists path often returns a slim JSON without a new id. + print(f"Ensured queue '{QUEUE_NAME}' exists.") + else: + print(f"Created queue '{QUEUE_NAME}'.") + return + + error = str(result.get("error", "")) + if "already exists" in error.lower(): + print(f"Queue '{QUEUE_NAME}' already exists.") + return + + raise RuntimeError(f"Failed to create queue '{QUEUE_NAME}': {result}") + + +def main() -> int: + wait_for_jolokia() + ensure_queue() + print(json.dumps({"queue": QUEUE_NAME, "jolokia_url": JOLOKIA_URL})) + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/test/local/readme.md b/test/local/readme.md index fa054a41..be0c6892 100644 --- a/test/local/readme.md +++ b/test/local/readme.md @@ -289,7 +289,7 @@ To initialize RabbitMQ and queue messages: ActiveMQ Artemis takes a few more steps to set up than the other input sources. -To initialize RabbitMQ and queue messages: +To initialize ActiveMQ and queue messages: 1. Bring up ministack and Redis: @@ -309,39 +309,23 @@ To initialize RabbitMQ and queue messages: docker compose up -d activemq ``` -4. Go to http://localhost:8161/ - -5. Sign in with the username 'admin' and password 'admin'. - -6. Select the 'Addresses' tab. - -7. Create a new multicast address named `/queue/ActiveQueue`. - -8. As the list of addresses does not automatically refresh, navigate away from the 'Addresses' tab and then return to the 'Addresses' tab. - -9. Go to Artemis JMX by selecting it as an option in the menu generated by clicking the 3-dot icon for the newly-created multi-cast address. - -10. In Artemis JMX, select the newly-created `/queue/ActiveQueue` address. - -11. Within the Artemis JMX menu for `/queue/ActiveQueue` address, go to the Create Queue tab. + The local compose service raises Artemis `max-disk-usage` to `99` so publishes are not + blocked when the host disk is already past the stock `90` threshold. -12. Create an anycast queue named `/queue/ActiveQueue`. - -13. To insert a new message, you can do one of the following: - - * Use the `send-activemq-message.py` script (requires the `stomp.py` Python library) +4. Create the ActiveMQ anycast queue (safe to re-run if it already exists). Uses the Artemis Jolokia HTTP API (stdlib only; no extra Python packages): ```bash - ./send-activemq-message.py 12 + ./make-local-activemq-resources.py ``` - * In Artemis JMX, select the ActiveQueue address and then select the 'Send Message' tab. Example of a message JSON: +5. Use the `send-activemq-message.py` script to publish a message to the `/queue/ActiveQueue` queue. This requires the `stomp.py` Python module. Specify the number of seconds the worker should sleep for in the first argument. You may optionally provide a second argument to set the STOMP `correlation-id` header for idempotency testing: - ```json - {"SleepDurationSeconds": 12} + ```bash + ./send-activemq-message.py 12 + ./send-activemq-message.py 12 example-idempotency-key ``` -14. Before starting the worker, make sure that neither the `USE_ACTIVEMQ` is set to `1` and that other `USE_` environment variables are not set to `1`. +6. Before starting the worker, make sure that `USE_ACTIVEMQ` is set to `1` and that other `USE_` environment variables are not set to `1`. Unset `COMMON__DISTRIBUTED__REDIS__CONNECTION_STRING_PATH` so compose uses the default SSM path (`/common/redis`): ```bash @@ -359,12 +343,40 @@ To initialize RabbitMQ and queue messages: unset COMMON__DISTRIBUTED__REDIS__CONNECTION_STRING_PATH ``` -15. Bring up the worker: +7. Bring up the worker: ```bash docker compose up worker ``` +#### Alternative: Web Console Testing + +If you prefer not to use the Python scripts, you can create the address/queue and send messages from the Artemis web console: + +1. Go to http://localhost:8161/ + +2. Sign in with the username `admin` and password `admin`. + +3. Select the **Addresses** tab. + +4. Create a new multicast address named `/queue/ActiveQueue`. + +5. As the list of addresses does not automatically refresh, navigate away from the **Addresses** tab and then return to the **Addresses** tab. + +6. Go to Artemis JMX by selecting it as an option in the menu generated by clicking the 3-dot icon for the newly-created multi-cast address. + +7. In Artemis JMX, select the newly-created `/queue/ActiveQueue` address. + +8. Within the Artemis JMX menu for the `/queue/ActiveQueue` address, go to the **Create Queue** tab. + +9. Create an anycast queue named `/queue/ActiveQueue`. + +10. To insert a new message, in Artemis JMX select the ActiveQueue address and then select the **Send Message** tab. Example message JSON: + + ```json + {"SleepDurationSeconds": 12} + ``` + ### NATS NATS takes a few more steps to set up than the other input sources. It will also require the installation of the `nats` command. diff --git a/test/local/send-activemq-message.py b/test/local/send-activemq-message.py index d8fa4efd..14b73be8 100755 --- a/test/local/send-activemq-message.py +++ b/test/local/send-activemq-message.py @@ -1,24 +1,121 @@ -#!/usr/bin/env python +#!/usr/bin/env python3 +"""Publish a sleep-job message to the local ActiveMQ Artemis queue. + +Requires the `stomp.py` Python module (`pip install stomp.py`). +""" + +import base64 import json -import stomp # pip install stomp.py import sys +import time +import urllib.error +import urllib.request + +import stomp + +QUEUE = "/queue/ActiveQueue" +HOST = "localhost" +PORT = 61616 +USERNAME = "admin" +PASSWORD = "admin" +JOLOKIA_URL = "http://localhost:8161/console/jolokia/" +BROKER_MBEAN = 'org.apache.activemq.artemis:broker="0.0.0.0"' + + +class _ErrorListener(stomp.ConnectionListener): + def __init__(self) -> None: + self.error: str | None = None + + def on_error(self, frame) -> None: # noqa: ANN001 - stomp.py frame type + body = frame.body or "" + headers = getattr(frame, "headers", {}) or {} + message = headers.get("message") or body or "unknown STOMP error" + self.error = message + + +def _auth_header() -> str: + token = base64.b64encode(f"{USERNAME}:{PASSWORD}".encode()).decode() + return f"Basic {token}" + + +def _disk_usage_hint() -> str | None: + """Return a hint when Artemis is blocking producers due to max-disk-usage.""" + try: + payload = json.dumps( + { + "type": "read", + "mbean": BROKER_MBEAN, + "attribute": ["DiskStoreUsage", "MaxDiskUsage"], + } + ).encode() + request = urllib.request.Request( + JOLOKIA_URL, + data=payload, + method="POST", + headers={ + "Content-Type": "application/json", + "Origin": "http://localhost", + "Authorization": _auth_header(), + }, + ) + with urllib.request.urlopen(request, timeout=3) as response: + result = json.loads(response.read().decode()) + values = result.get("value") or {} + usage = float(values.get("DiskStoreUsage", 0)) + max_usage = float(values.get("MaxDiskUsage", 90)) + usage_pct = usage * 100 if usage <= 1 else usage + if usage_pct >= max_usage: + return ( + f"Artemis is blocking producers: disk usage {usage_pct:.1f}% " + f">= max-disk-usage {max_usage:.0f}%. " + "Recreate the local activemq compose service " + "(it raises max-disk-usage to 99 for local use), or free disk space." + ) + except (urllib.error.URLError, TimeoutError, ValueError, TypeError, KeyError): + return None + return None + + +def main() -> int: + if len(sys.argv) < 2: + print( + f"Usage: {sys.argv[0]} [message-id]", + file=sys.stderr, + ) + return 1 + + body = {"SleepDurationSeconds": int(sys.argv[1])} + headers = { + "content-type": "application/json", + "persistent": "true", + "receipt": "send-1", + } + if len(sys.argv) > 2: + headers["correlation-id"] = sys.argv[2] -# Connection configuration -# Default STOMP port for ActiveMQ is 61616 -conn = stomp.Connection([('localhost', 61616)]) + listener = _ErrorListener() + conn = stomp.Connection([(HOST, PORT)]) + conn.set_listener("errors", listener) + conn.connect(USERNAME, PASSWORD, wait=True) + try: + # Destination must be prefixed with /queue/ for STOMP anycast. + conn.send(body=json.dumps(body), destination=QUEUE, headers=headers) + # Allow ERROR frames (e.g. broker disk full) to arrive before disconnect. + time.sleep(0.25) + if listener.error: + hint = _disk_usage_hint() + print(f"Failed to publish to {QUEUE}: {listener.error}", file=sys.stderr) + if hint: + print(hint, file=sys.stderr) + return 1 + finally: + if conn.is_connected(): + conn.disconnect() -# Connect with credentials -conn.connect('admin', 'admin', wait=True) + print(f"Published to {QUEUE}: {json.dumps(body)}") + return 0 -# Send a message to a queue -# The destination must be prefixed with '/queue/' or '/topic/' -body = {'SleepDurationSeconds': int(sys.argv[1])} -queue = '/queue/ActiveQueue' -# Note: This technically sends a binary payload. -# The worker is still able to handle this. -conn.send(body=json.dumps(body), destination=queue) -# Disconnect -conn.disconnect() -print("Message sent successfully") +if __name__ == "__main__": + sys.exit(main())