diff --git a/README.md b/README.md index ef98644..0284366 100644 --- a/README.md +++ b/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. diff --git a/src/SqsPoller.Extensions.Publisher/SqsPoller.Extensions.Publisher.csproj b/src/SqsPoller.Extensions.Publisher/SqsPoller.Extensions.Publisher.csproj index 62f9994..078867f 100644 --- a/src/SqsPoller.Extensions.Publisher/SqsPoller.Extensions.Publisher.csproj +++ b/src/SqsPoller.Extensions.Publisher/SqsPoller.Extensions.Publisher.csproj @@ -3,7 +3,7 @@ netstandard2.0 SqsPoller.Extensions.Publisher - 2.0.1$(VersionSuffix) + 2.1.0$(VersionSuffix) Alexey Bogdan Alexey Bogdan A small library that helps ASP.NET Core applications easily consume messages from a SQS queue diff --git a/src/SqsPoller/SqsPoller.csproj b/src/SqsPoller/SqsPoller.csproj index bca3829..bafc24a 100644 --- a/src/SqsPoller/SqsPoller.csproj +++ b/src/SqsPoller/SqsPoller.csproj @@ -3,7 +3,7 @@ netstandard2.0 SqsPoller - 2.0.1$(VersionSuffix) + 2.1.0$(VersionSuffix) Alexey Bogdan Alexey Bogdan A small library that helps ASP.NET Core applications easily consume messages from a SQS queue diff --git a/src/SqsPoller/SqsPollerConfig.cs b/src/SqsPoller/SqsPollerConfig.cs index 244e8d7..82c5cbf 100644 --- a/src/SqsPoller/SqsPollerConfig.cs +++ b/src/SqsPoller/SqsPollerConfig.cs @@ -70,6 +70,33 @@ public class SqsPollerConfig /// public int MaxNumberOfParallelism { get; set; } = 100; + /// + /// (Optional) Gets and sets the per-message processing budget (default null, no limit). + /// + /// 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. + /// + /// + /// 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. + /// + /// + /// A consumer that ignores the token is stopped being awaited after this timeout plus + /// . 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. + /// + /// + public TimeSpan? MessageProcessingTimeout { get; set; } + + /// + /// Gets and sets how long to wait, after has fired, + /// for a consumer to honour its cancellation token before it is abandoned (default 5 seconds). + /// Only used when is set. + /// + public TimeSpan MessageProcessingAbandonGracePeriod { get; set; } = TimeSpan.FromSeconds(5); + /// /// Gets and sets a log level for default log messages when an exception is caught /// @@ -80,6 +107,15 @@ public class SqsPollerConfig /// /// A callback with an exception and messageId public Action? OnHandleMessageException { get; set; } + + /// + /// (Optional) A callback invoked when a message exceeds , + /// 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. + /// + /// A callback with the messageId and whether the consumer was abandoned + /// (it ignored the cancellation token and was still running after the grace period) + public Action? OnHandleMessageTimeout { get; set; } /// /// (Optional) A custom exception handler that overrides DeleteMessage default exception handler. diff --git a/src/SqsPoller/SqsPollerHostedService.cs b/src/SqsPoller/SqsPollerHostedService.cs index f49fb18..f50b189 100644 --- a/src/SqsPoller/SqsPollerHostedService.cs +++ b/src/SqsPoller/SqsPollerHostedService.cs @@ -27,6 +27,13 @@ public SqsPollerHostedService( _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) @@ -75,7 +82,15 @@ public async Task HandleMessage(Message message, CancellationToken cancellationT 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}"); @@ -99,6 +114,69 @@ public async Task HandleMessage(Message message, CancellationToken cancellationT } } + /// + /// Runs the consumer under a token that fires after 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. + /// + private async Task 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 diff --git a/test/SqsPoller.Tests.Unit/SqsPollerHostedServiceTimeoutTests.cs b/test/SqsPoller.Tests.Unit/SqsPollerHostedServiceTimeoutTests.cs new file mode 100644 index 0000000..6ef6809 --- /dev/null +++ b/test/SqsPoller.Tests.Unit/SqsPollerHostedServiceTimeoutTests.cs @@ -0,0 +1,288 @@ +#nullable enable +using System; +using System.Collections.Generic; +using System.Linq; +using System.Threading; +using System.Threading.Tasks; +using Amazon.SQS; +using Amazon.SQS.Model; +using Microsoft.Extensions.Logging; +using NSubstitute; +using Shouldly; +using Xunit; + +namespace SqsPoller.Tests.Unit; + +public class SqsPollerHostedServiceTimeoutTests +{ + private const string TimeoutLogText = "was not processed within MessageProcessingTimeout"; + private static readonly TimeSpan Timeout = TimeSpan.FromMilliseconds(100); + private static readonly TimeSpan Grace = TimeSpan.FromMilliseconds(100); + // Upper bound for any single wait in these tests; far above Timeout + Grace. + private static readonly TimeSpan TestGuard = TimeSpan.FromSeconds(10); + + [Fact] + public async Task HandleMessage_ConsumerIgnoresTokenAndNeverCompletes_AbandonsItReleasesSlotAndDoesNotDelete() + { + //Arrange + var sqsClient = Substitute.For(); + string? timedOutMessageId = null; + bool? abandoned = null; + var config = TimeoutConfig(); + config.OnHandleMessageTimeout = (id, a) => { timedOutMessageId = id; abandoned = a; }; + var consumerResolver = Substitute.For(); + consumerResolver.Resolve(Arg.Any(), Arg.Any()) + .Returns(new TaskCompletionSource().Task); + var logger = Substitute.For>(); + var hostedService = new SqsPollerHostedService(sqsClient, config, consumerResolver, logger); + var semaphore = new SemaphoreSlim(1); + await semaphore.WaitAsync(); + + //Act + await hostedService.HandleMessage(TestMessage(), CancellationToken.None, semaphore, "") + .WaitAsync(TestGuard); + + //Assert + semaphore.CurrentCount.ShouldBe(1); + await sqsClient.DidNotReceive().DeleteMessageAsync(Arg.Any(), Arg.Any()); + CountLogs(logger, LogLevel.Error, TimeoutLogText).ShouldBe(1); + timedOutMessageId.ShouldBe("test-message-id"); + abandoned.ShouldBe(true); + } + + [Fact] + public async Task HandleMessage_ConsumerHonoursToken_TimesOutWithoutAbandoningAndDoesNotDelete() + { + //Arrange + var sqsClient = Substitute.For(); + bool? abandoned = null; + var config = TimeoutConfig(); + config.MessageProcessingAbandonGracePeriod = TestGuard; + config.OnHandleMessageTimeout = (_, a) => abandoned = a; + var consumerResolver = Substitute.For(); + consumerResolver.Resolve(Arg.Any(), Arg.Any()) + .Returns(ci => Task.Delay(System.Threading.Timeout.Infinite, ci.Arg())); + var logger = Substitute.For>(); + var hostedService = new SqsPollerHostedService(sqsClient, config, consumerResolver, logger); + var semaphore = new SemaphoreSlim(0); + + //Act + await hostedService.HandleMessage(TestMessage(), CancellationToken.None, semaphore, "") + .WaitAsync(TestGuard); + + //Assert + semaphore.CurrentCount.ShouldBe(1); + await sqsClient.DidNotReceive().DeleteMessageAsync(Arg.Any(), Arg.Any()); + CountLogs(logger, LogLevel.Error, TimeoutLogText).ShouldBe(1); + CountLogs(logger, LogLevel.Error, "Failed to handle message").ShouldBe(0); + abandoned.ShouldBe(false); + } + + [Fact] + public async Task HandleMessage_ConsumerCompletesInTime_DeletesMessage() + { + //Arrange + var sqsClient = Substitute.For(); + var timeoutReported = false; + var config = TimeoutConfig(); + config.OnHandleMessageTimeout = (_, _) => timeoutReported = true; + var consumerResolver = Substitute.For(); + consumerResolver.Resolve(Arg.Any(), Arg.Any()).Returns(Task.CompletedTask); + var logger = Substitute.For>(); + var hostedService = new SqsPollerHostedService(sqsClient, config, consumerResolver, logger); + + //Act + await hostedService.HandleMessage(TestMessage(), CancellationToken.None, new SemaphoreSlim(0), "") + .WaitAsync(TestGuard); + + //Assert + await sqsClient.Received(1).DeleteMessageAsync( + Arg.Is(r => r.ReceiptHandle == "test-receipt-handle"), Arg.Any()); + timeoutReported.ShouldBeFalse(); + CountLogs(logger, LogLevel.Error, TimeoutLogText).ShouldBe(0); + } + + [Fact] + public async Task HandleMessage_ShutdownCancelsConsumer_NotReportedAsTimeoutAndNotDeleted() + { + //Arrange + var sqsClient = Substitute.For(); + var timeoutReported = false; + var config = TimeoutConfig(); + config.MessageProcessingTimeout = TestGuard; + config.OnHandleMessageTimeout = (_, _) => timeoutReported = true; + var consumerStarted = new TaskCompletionSource(); + var consumerResolver = Substitute.For(); + consumerResolver.Resolve(Arg.Any(), Arg.Any()) + .Returns(ci => + { + consumerStarted.TrySetResult(true); + return Task.Delay(System.Threading.Timeout.Infinite, ci.Arg()); + }); + var logger = Substitute.For>(); + var hostedService = new SqsPollerHostedService(sqsClient, config, consumerResolver, logger); + using var stopping = new CancellationTokenSource(); + var semaphore = new SemaphoreSlim(0); + + //Act + var handling = hostedService.HandleMessage(TestMessage(), stopping.Token, semaphore, ""); + await consumerStarted.Task.WaitAsync(TestGuard); + stopping.Cancel(); + await handling.WaitAsync(TestGuard); + + //Assert + semaphore.CurrentCount.ShouldBe(1); + timeoutReported.ShouldBeFalse(); + CountLogs(logger, LogLevel.Error, TimeoutLogText).ShouldBe(0); + // Unchanged pre-timeout behaviour on shutdown: the cancellation goes to the regular handler. + CountLogs(logger, LogLevel.Error, "Failed to handle message").ShouldBe(1); + await sqsClient.DidNotReceive().DeleteMessageAsync(Arg.Any(), Arg.Any()); + } + + [Fact] + public async Task HandleMessage_AbandonedConsumerFailsLater_ExceptionIsObserved() + { + //Arrange + var sqsClient = Substitute.For(); + var config = TimeoutConfig(); + var hung = new TaskCompletionSource(); + var consumerResolver = Substitute.For(); + consumerResolver.Resolve(Arg.Any(), Arg.Any()).Returns(hung.Task); + var logger = Substitute.For>(); + var hostedService = new SqsPollerHostedService(sqsClient, config, consumerResolver, logger); + await hostedService.HandleMessage(TestMessage(), CancellationToken.None, new SemaphoreSlim(0), "") + .WaitAsync(TestGuard); + + //Act + hung.SetException(new InvalidOperationException("late failure")); + + //Assert + await WaitUntil(() => CountLogs(logger, LogLevel.Debug, "Abandoned consumer of message") == 1); + } + + [Fact] + public async Task ExecuteAsync_SingleSlotAndHungConsumer_NextMessageIsStillProcessed() + { + //Arrange + var sqsClient = Substitute.For(); + var receives = new Queue(new[] + { + new Message { MessageId = "hung", ReceiptHandle = "hung-rh" }, + new Message { MessageId = "next", ReceiptHandle = "next-rh" } + }); + sqsClient.ReceiveMessageAsync(Arg.Any(), Arg.Any()) + .Returns(async ci => + { + lock (receives) + { + if (receives.Count > 0) + return new ReceiveMessageResponse { Messages = new List { receives.Dequeue() } }; + } + await Task.Delay(20, ci.Arg()); + return new ReceiveMessageResponse { Messages = new List() }; + }); + var config = TimeoutConfig(); + config.QueueUrl = "http://localhost/queue"; + config.MaxNumberOfParallelism = 1; + var consumerResolver = Substitute.For(); + consumerResolver.Resolve(Arg.Any(), Arg.Any()) + .Returns(ci => ci.Arg().MessageId == "hung" + ? new TaskCompletionSource().Task + : Task.CompletedTask); + var logger = Substitute.For>(); + var hostedService = new SqsPollerHostedService(sqsClient, config, consumerResolver, logger); + + //Act + await hostedService.StartAsync(CancellationToken.None); + try + { + await WaitUntil(() => DeletedReceiptHandles(sqsClient).Contains("next-rh")); + } + finally + { + await hostedService.StopAsync(CancellationToken.None); + } + + //Assert + DeletedReceiptHandles(sqsClient).ShouldNotContain("hung-rh"); + CountLogs(logger, LogLevel.Error, TimeoutLogText).ShouldBe(1); + } + + [Fact] + public async Task HandleMessage_TimeoutNotSet_ConsumerGetsCallerTokenAndIsAwaitedWithoutLimit() + { + //Arrange + var sqsClient = Substitute.For(); + var config = new SqsPollerConfig(); + var hung = new TaskCompletionSource(); + var consumerResolver = Substitute.For(); + consumerResolver.Resolve(Arg.Any(), Arg.Any()).Returns(hung.Task); + var logger = Substitute.For>(); + var hostedService = new SqsPollerHostedService(sqsClient, config, consumerResolver, logger); + using var stopping = new CancellationTokenSource(); + var semaphore = new SemaphoreSlim(0); + + //Act + var handling = hostedService.HandleMessage(TestMessage(), stopping.Token, semaphore, ""); + var finishedEarly = await Task.WhenAny(handling, Task.Delay(Timeout + Grace + Grace)) == handling; + hung.SetResult(true); + await handling.WaitAsync(TestGuard); + + //Assert + config.MessageProcessingTimeout.ShouldBeNull(); + finishedEarly.ShouldBeFalse(); + await consumerResolver.Received(1).Resolve(Arg.Any(), stopping.Token); + await sqsClient.Received(1).DeleteMessageAsync(Arg.Any(), Arg.Any()); + semaphore.CurrentCount.ShouldBe(1); + CountLogs(logger, LogLevel.Error, TimeoutLogText).ShouldBe(0); + } + + [Theory] + [InlineData(0)] + [InlineData(-1)] + public void Constructor_NonPositiveTimeout_Throws(int milliseconds) + { + var config = new SqsPollerConfig { MessageProcessingTimeout = TimeSpan.FromMilliseconds(milliseconds) }; + + Should.Throw(() => new SqsPollerHostedService( + Substitute.For(), config, Substitute.For(), + Substitute.For>())); + } + + private static SqsPollerConfig TimeoutConfig() => new SqsPollerConfig + { + MessageProcessingTimeout = Timeout, + MessageProcessingAbandonGracePeriod = Grace + }; + + private static Message TestMessage() => new Message + { + MessageId = "test-message-id", + ReceiptHandle = "test-receipt-handle" + }; + + private static int CountLogs(ILogger logger, LogLevel level, string text) => + logger.ReceivedCalls() + .Where(c => c.GetMethodInfo().Name == nameof(ILogger.Log)) + .Select(c => c.GetArguments()) + .Count(args => (LogLevel)args[0]! == level && args[2]?.ToString()?.Contains(text) == true); + + private static List DeletedReceiptHandles(IAmazonSQS sqsClient) => + sqsClient.ReceivedCalls() + .Where(c => c.GetMethodInfo().Name == nameof(IAmazonSQS.DeleteMessageAsync)) + .Select(c => c.GetArguments()[0]) + .OfType() + .Select(r => r.ReceiptHandle) + .ToList(); + + private static async Task WaitUntil(Func condition) + { + var deadline = DateTime.UtcNow + TestGuard; + while (!condition()) + { + if (DateTime.UtcNow > deadline) + throw new TimeoutException("Condition was not met in time"); + await Task.Delay(10); + } + } +}