[ATF-2881] Add optional per-message processing timeout

README.md

@@ -34,3 +34,32 @@ Create a message object and consumer that consumes the message
 # Constraint
 
 Each message should have a `MessageType` header and value should the name of the type that you are going to consume (i.e. `UserCreated`). 
+
+# Per-message processing timeout
+
+By default a consumer runs for as long as it takes. A consumer that hangs (for example on a dead
+connection) keeps its concurrency slot forever, and when all `MaxNumberOfParallelism` slots are stuck
+the poller stops processing without any error.
+
+Set `MessageProcessingTimeout` to bound each message:
+
+```
+  services.AddSqsPoller(new SqsPollerConfig
+  {
+      QueueUrl = "...",
+      // Keep it below the queue's visibility timeout.
+      MessageProcessingTimeout = TimeSpan.FromMinutes(2),
+      OnHandleMessageTimeout = (messageId, abandoned) => { /* e.g. increment a metric */ }
+  }, new[] { typeof(SendConfirmationMessageConsumer) });
+```
+
+When the timeout fires:
+
+* the consumer's `CancellationToken` is cancelled;
+* the message is not deleted, so SQS redelivers it after its visibility timeout;
+* an error is logged with the message id, and `OnHandleMessageTimeout` is called if set;
+* if the consumer ignores the token, it stops being awaited after `MessageProcessingAbandonGracePeriod`
+  (default 5 seconds) and its slot is released. It keeps running in the background, so consumers must
+  be idempotent.
+
+Shutdown of the host is not reported as a timeout.

src/SqsPoller.Extensions.Publisher/SqsPoller.Extensions.Publisher.csproj

@@ -3,7 +3,7 @@
     <PropertyGroup>
         <TargetFramework>netstandard2.0</TargetFramework>
         <PackageId>SqsPoller.Extensions.Publisher</PackageId>
-        <Version>2.0.1$(VersionSuffix)</Version>
+        <Version>2.1.0$(VersionSuffix)</Version>
         <Authors>Alexey Bogdan</Authors>
         <Company>Alexey Bogdan</Company>
         <Description>A small library that helps ASP.NET Core applications easily consume messages from a SQS queue</Description>

src/SqsPoller/SqsPoller.csproj

@@ -3,7 +3,7 @@
     <PropertyGroup>
         <TargetFramework>netstandard2.0</TargetFramework>
         <PackageId>SqsPoller</PackageId>
-        <Version>2.0.1$(VersionSuffix)</Version>
+        <Version>2.1.0$(VersionSuffix)</Version>
         <Authors>Alexey Bogdan</Authors>
         <Company>Alexey Bogdan</Company>
         <Description>A small library that helps ASP.NET Core applications easily consume messages from a SQS queue</Description>

src/SqsPoller/SqsPollerConfig.cs

@@ -70,6 +70,33 @@ public class SqsPollerConfig
         /// </summary>
         public int MaxNumberOfParallelism { get; set; } = 100;
 
+        /// <summary>
+        /// (Optional) Gets and sets the per-message processing budget (default null, no limit).
+        /// <para>
+        /// When set, the consumer receives a cancellation token that fires after this time. A
+        /// message whose consumer does not finish in time is NOT deleted, so SQS redelivers it
+        /// once its visibility timeout expires, and its concurrency slot is released.
+        /// </para>
+        /// <para>
+        /// Set it below the queue's visibility timeout. Otherwise SQS makes the message visible
+        /// again while the first attempt is still running, and a second consumer picks it up.
+        /// </para>
+        /// <para>
+        /// A consumer that ignores the token is stopped being awaited after this timeout plus
+        /// <see cref="MessageProcessingAbandonGracePeriod"/>. It keeps running in the
+        /// background, but no longer occupies a slot. Consumers must be idempotent: an abandoned
+        /// consumer can still finish its work after the message has been redelivered.
+        /// </para>
+        /// </summary>
+        public TimeSpan? MessageProcessingTimeout { get; set; }
+
+        /// <summary>
+        /// Gets and sets how long to wait, after <see cref="MessageProcessingTimeout"/> has fired,
+        /// for a consumer to honour its cancellation token before it is abandoned (default 5 seconds).
+        /// Only used when <see cref="MessageProcessingTimeout"/> is set.
+        /// </summary>
+        public TimeSpan MessageProcessingAbandonGracePeriod { get; set; } = TimeSpan.FromSeconds(5);
+
         /// <summary>
         /// Gets and sets a log level for default log messages when an exception is caught
         /// </summary>
@@ -80,6 +107,15 @@ public class SqsPollerConfig
         /// </summary>
         /// <returns>A callback with an exception and messageId</returns>
         public Action<Exception, string>? OnHandleMessageException { get; set; }
+
+        /// <summary>
+        /// (Optional) A callback invoked when a message exceeds <see cref="MessageProcessingTimeout"/>,
+        /// for example to count timeouts in a metric. The timeout is always logged at Error level as well.
+        /// It is not invoked when the host is shutting down.
+        /// </summary>
+        /// <returns>A callback with the messageId and whether the consumer was abandoned
+        /// (it ignored the cancellation token and was still running after the grace period)</returns>
+        public Action<string, bool>? OnHandleMessageTimeout { get; set; }
 
         /// <summary>
         /// (Optional) A custom exception handler that overrides DeleteMessage default exception handler.

src/SqsPoller/SqsPollerHostedService.cs

@@ -27,6 +27,13 @@
             _consumerResolver = consumerResolver;
             _logger = logger;
             _config = config;
+
+            if (config.MessageProcessingTimeout is { } timeout && timeout <= TimeSpan.Zero)
+                throw new ArgumentOutOfRangeException(nameof(config), timeout,
+                    "SqsPollerConfig.MessageProcessingTimeout must be positive, or null to disable it.");
+            if (config.MessageProcessingAbandonGracePeriod < TimeSpan.Zero)
+                throw new ArgumentOutOfRangeException(nameof(config), config.MessageProcessingAbandonGracePeriod,
+                    "SqsPollerConfig.MessageProcessingAbandonGracePeriod must not be negative.");
         }
 
         protected override async Task ExecuteAsync(CancellationToken stoppingToken)
@@ -64,7 +71,7 @@
                     "Start processing the message with id {message_id} and ReceiptHandle {receipt_handle}");
 
                 await semaphore.WaitAsync(cancellationToken);
                 HandleMessage(message, cancellationToken, semaphore, queueUrl);
                 if (cancellationToken.IsCancellationRequested)
                     return;
             }
@@ -75,11 +82,19 @@
             try
             {
                 _logger.LogTrace("Handling message {message_id} {receipt_handle}");
-                await _consumerResolver.Resolve(message, cancellationToken);
+                if (_config.MessageProcessingTimeout is { } timeout)
+                {
+                    if (!await ResolveWithTimeout(message, timeout, cancellationToken))
+                        return; // Not deleted: SQS redelivers it after the visibility timeout.
+                }
+                else
+                {
+                    await _consumerResolver.Resolve(message, cancellationToken);
+                }
                 _logger.LogTrace("Message {message_id} {receipt_handle} has been handled");
 
                 _logger.LogTrace("Deleting the message with id {message_id} and ReceiptHandle {receipt_handle}");
                 DeleteMessage(queueUrl, message, cancellationToken);
                 _logger.LogTrace("Message {message_id} {receipt_handle} has been deleted");
             }
             catch (Exception e)
@@ -99,6 +114,69 @@
             }
         }
 
+        /// <summary>
+        /// Runs the consumer under a token that fires after <paramref name="timeout"/> or on shutdown.
+        /// Returns true when the consumer completed and the message may be deleted, false when it
+        /// timed out or was abandoned. Other consumer exceptions, including cancellation on shutdown,
+        /// propagate to the caller exactly as without a timeout.
+        /// </summary>
+        private async Task<bool> ResolveWithTimeout(Message message, TimeSpan timeout, CancellationToken stoppingToken)
+        {
+            using var timeoutCts = CancellationTokenSource.CreateLinkedTokenSource(stoppingToken);
+            timeoutCts.CancelAfter(timeout);
+
+            // Task.Run: a consumer that blocks synchronously before its first await would otherwise
+            // block this method, and the deadline below could never be observed.
+            var consumerTask = Task.Run(() => _consumerResolver.Resolve(message, timeoutCts.Token));
+
+            using var deadlineCts = new CancellationTokenSource();
+            var deadline = Task.Delay(timeout + _config.MessageProcessingAbandonGracePeriod, deadlineCts.Token);
+            if (await Task.WhenAny(consumerTask, deadline) == consumerTask)
+            {
+                deadlineCts.Cancel();
+                try
+                {
+                    await consumerTask;
+                    return true;
+                }
+                // Shutdown wins: once the stopping token has fired, the exception propagates to
+                // HandleMessage's exception handler and is not reported as a timeout.
+                catch (Exception e) when (timeoutCts.IsCancellationRequested && !stoppingToken.IsCancellationRequested)
+                {
+                    ReportTimeout(message, timeout, abandoned: false, e);
+                    return false;
+                }
+            }
+
+            // The consumer ignored its token for the whole grace period. Stop awaiting it so the
+            // semaphore slot is released. It keeps running; observe its eventual exception so it
+            // does not surface as an unobserved task exception.
+            timeoutCts.Cancel();
+            _ = consumerTask.ContinueWith(
+                t => _logger.LogDebug(t.Exception, "Abandoned consumer of message {message_id} failed", message.MessageId),
+                CancellationToken.None,
+                TaskContinuationOptions.OnlyOnFaulted | TaskContinuationOptions.ExecuteSynchronously,
+                TaskScheduler.Default);
+
+            if (stoppingToken.IsCancellationRequested)
+            {
+                _logger.LogDebug("Consumer of message {message_id} ignored shutdown and was abandoned", message.MessageId);
+                return false;
+            }
+
+            ReportTimeout(message, timeout, abandoned: true, null);
+            return false;
+        }
+
+        private void ReportTimeout(Message message, TimeSpan timeout, bool abandoned, Exception? exception)
+        {
+            _logger.LogError(
+                exception,
+                "Message {message_id} was not processed within MessageProcessingTimeout {timeout} and is left in the queue for redelivery. Consumer abandoned: {abandoned}",
+                message.MessageId, timeout, abandoned);
+            _config.OnHandleMessageTimeout?.Invoke(message.MessageId, abandoned);
+        }
+
         private async Task DeleteMessage(string queueUrl, Message message, CancellationToken cancellationToken)
         {
             try