ActiveMQ resilience

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<IActiveMqServerConfigurationSource, ActiveMqServerConfigurationSource>()
             .AddSingleton<IInnerActiveMqConnectionFactory, InnerActiveMqConnectionFactory>()
-            .AddSingleton<IActiveMqConnectionFactory, ActiveMqConnectionFactory>();
+            .AddSingleton<IActiveMqConnectionFactory, ActiveMqConnectionFactory>()
+            .AddSingleton<IActiveMqExceptionArbiterService, ActiveMqExceptionArbiterService>()
+            .AddSingleton<IActiveMqRetryWrapperService, ActiveMqRetryWrapperService>();
     }
 }

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; }
+}

src/RedShirt.Example.JobWorker.JobManagement.ActiveMq/Services/ActiveMqJobSource.cs

@@ -7,73 +7,107 @@
 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
+internal class ActiveMqJobSource(
+    IActiveMqConnectionFactory connectionFactory,
+    IActiveMqRetryWrapperService retryWrapperService,
+    IOptions<ActiveMqJobSource.ConfigurationModel> configuration,
+    ILogger<ActiveMqJobSource> logger)
+    : IJobSource
 {
-    private readonly IOptions<ConfigurationModel> _configuration;
+    private IMessageConsumer? _messageConsumer;
 
-    // ReSharper disable once PrivateFieldCanBeConvertedToLocalVariable
-    private readonly Lazy<Task<IConnection>> _connection;
-    private readonly ILogger<ActiveMqJobSource> _logger;
+    private async Task<JobSourceResponse> FetchJobsAsync(int batchSize, CancellationToken cancellationToken)
+    {
+        try
+        {
+            var consumer = await retryWrapperService.RunAsync(GetConsumerAsync, cancellationToken);
+            var getJobsResponseItems = new List<IRawJobModel>();
 
-    // ReSharper disable once PrivateFieldCanBeConvertedToLocalVariable
-    private readonly Lazy<Task<IMessageConsumer>> _messageConsumer;
+            while (getJobsResponseItems.Count < batchSize)
+            {
+                var result =
+                    await retryWrapperService.RunAsync(_ => consumer.ReceiveAsync(TimeSpan.FromMilliseconds(100)),
+                        cancellationToken);
+
+                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
+                });
+            }
 
-    // ReSharper disable once PrivateFieldCanBeConvertedToLocalVariable
-    private readonly Lazy<Task<IQueue?>> _queue;
-    private readonly Lazy<Task<ISession>> _session;
+            return new JobSourceResponse
+            {
+                Items = getJobsResponseItems
+            };
+        }
+        catch
+        {
+            ResetConsumer();
+            throw;
+        }
+    }
 
-    public ActiveMqJobSource(IActiveMqConnectionFactory connectionFactory,
-        IOptions<ConfigurationModel> configuration,
-        ILogger<ActiveMqJobSource> logger)
+    /// <summary>
+    ///     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.
+    /// </summary>
+    /// <param name="cancellationToken"></param>
+    /// <returns></returns>
+    /// <exception cref="CouldNotLoadQueueException"></exception>
+    private async Task<IMessageConsumer> GetConsumerAsync(CancellationToken cancellationToken)
     {
-        _configuration = configuration;
-        _logger = logger;
-        _connection = new Lazy<Task<IConnection>>(async () =>
-        {
-            var connection = await connectionFactory.GetConnectionAsync();
-            connection.Start();
-            return connection;
-        });
-        _session = new Lazy<Task<ISession>>(async () =>
+        if (_messageConsumer is not null)
         {
-            var connection = await _connection.Value;
-            return await connection.CreateSessionAsync(AcknowledgementMode.ClientAcknowledge);
-        });
-        _queue = new Lazy<Task<IQueue?>>(async () =>
-        {
-            var session = await _session.Value;
-            return await session.GetQueueAsync(_configuration.Value.QueueName);
-        });
-        _messageConsumer = new Lazy<Task<IMessageConsumer>>(async () =>
+            return _messageConsumer;
+        }
+
+        var connection = await connectionFactory.GetConnectionAsync(cancellationToken);
+        await connection.StartAsync();
+        var session = await connection.CreateSessionAsync(AcknowledgementMode.ClientAcknowledge);
+        var queue = await session.GetQueueAsync(configuration.Value.QueueName);
+
+        if (queue is null)
         {
-            var queue = await _queue.Value;
+            throw new CouldNotLoadQueueException();
+        }
 
-            if (queue is null)
-            {
-                throw new CouldNotLoadQueueException();
-            }
+        var consumer = await session.CreateConsumerAsync(queue);
+
+        // Cache for later
+        _messageConsumer = consumer;
+
+        return consumer;
+    }
 
-            var session = await _session.Value;
-            return await session.CreateConsumerAsync(queue);
-        });
+    private void ResetConsumer()
+    {
+        _messageConsumer = null;
     }
 
     public int RecommendedHeartbeatIntervalSeconds => 0;
 
     public bool IsSubscriptionSource => false;
 
 #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
@@ -82,43 +116,19 @@ 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<IJobSourceResponse> GetJobsAsync(int batchSize, CancellationToken cancellationToken = default)
     {
         batchSize = Math.Max(1, batchSize);
 
-        _logger.LogTrace("Fetching up to {EffectiveBatchSize} messages from ActiveMQ Queue: {QueueName}",
-            batchSize, _configuration.Value.QueueName);
-
-        var getJobsResponseItems = new List<IRawJobModel>();
-
-        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;
-            }
+        logger.LogTrace("Fetching up to {EffectiveBatchSize} messages from ActiveMQ Queue: {QueueName}",
+            batchSize, configuration.Value.QueueName);
 
-            // 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 FetchJobsAsync(batchSize, cancellationToken);
     }
 
     public Task HeartbeatAsync(IRawJobModel message, CancellationToken cancellationToken = default)

src/RedShirt.Example.JobWorker.JobManagement.ActiveMq/Services/Resilience/ActiveMqExceptionArbiterService.cs

@@ -0,0 +1,124 @@
+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;
+
+/// <summary>
+///     Classifies ActiveMQ / NMS client exceptions for retry decisions.
+/// </summary>
+internal interface IActiveMqExceptionArbiterService
+{
+    ActiveMqExceptionArbiterReport GetReport(Exception exception);
+}
+
+/// <summary>
+///     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.
+/// </summary>
+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),
+            // 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.
+            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 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)
+        };
+    }
+}