From c396f5f20e851734a4338ce33e94f3edca4cea97 Mon Sep 17 00:00:00 2001 From: bspeth Date: Mon, 12 Oct 2026 15:16:30 +0200 Subject: [PATCH] fix(#491328): abandon message when cancellation is requested in MessageReceptionHandler (5.8.1) Co-Authored-By: Claude Sonnet 5.5 --- docs/CHANGELOG.md | 4 + .../Reception/MessageReceptionHandler.cs | 17 ++++ .../Reception/ReceptionTest.cs | 84 +++++++++++++++++++ 3 files changed, 105 insertions(+) diff --git a/docs/CHANGELOG.md b/docs/CHANGELOG.md index 704ee23..15fe2d3 100644 --- a/docs/CHANGELOG.md +++ b/docs/CHANGELOG.md @@ -4,6 +4,10 @@ All notable changes to this project will be documented in this file. The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/), and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html). +## 5.8.1 +- Fixed + - `MessageReceptionHandler` now explicitly abandons the message when cancellation is requested before the handler is invoked (#491328). Previously it returned normally, so the Azure SDK treated the message as successfully handled and, with `AutoCompleteMessages` enabled, completed (deleted) a message that was never processed. This silently lost messages during pod shutdown. The message is now abandoned (using `CancellationToken.None`, as the context token is already cancelled) so the broker redelivers it after the lock is released. + ## 5.8.0 - Added - `net10.0` added as an additional target framework across `Ev.ServiceBus`, `Ev.ServiceBus.Abstractions`, `Ev.ServiceBus.Apm`, `Ev.ServiceBus.AsyncApi`, `Ev.ServiceBus.HealthChecks`, `Ev.ServiceBus.Mvc`, and `Ev.ServiceBus.Prometheus`, alongside the existing `net8.0` (and `netstandard2.0`/`netstandard2.1` on the core packages). Non-breaking — existing net8.0 consumers are unaffected. diff --git a/src/Ev.ServiceBus/Reception/MessageReceptionHandler.cs b/src/Ev.ServiceBus/Reception/MessageReceptionHandler.cs index 8af7f7a..2207c1e 100644 --- a/src/Ev.ServiceBus/Reception/MessageReceptionHandler.cs +++ b/src/Ev.ServiceBus/Reception/MessageReceptionHandler.cs @@ -70,7 +70,11 @@ public async Task HandleMessageAsync(MessageContext context) return; if (context.CancellationToken.IsCancellationRequested) + { + // Returning normally would let the SDK auto-complete (delete) a message that was never processed. + await AbandonMessageAsync(context); return; + } var @event = _messagePayloadSerializer.DeSerializeBody(context.Message.Body.ToArray(), context.ReceptionRegistration!.PayloadType); var methodInfo = _callHandlerInfo.MakeGenericMethod(context.ReceptionRegistration.PayloadType); @@ -129,6 +133,19 @@ public async Task HandleMessageAsync(MessageContext context) } } + private static async Task AbandonMessageAsync(MessageContext context) + { + // CancellationToken.None: the context token is already cancelled and would make the abandon itself fail. + if (context.SessionArgs != null) + { + await context.SessionArgs.AbandonMessageAsync(context.Message, cancellationToken: CancellationToken.None); + } + else if (context.Args != null) + { + await context.Args.AbandonMessageAsync(context.Message, cancellationToken: CancellationToken.None); + } + } + private IDisposable? AddLoggingContext(MessageContext context) { var executionContext = context.ReadExecutionContext(); diff --git a/tests/Ev.ServiceBus.UnitTests/Reception/ReceptionTest.cs b/tests/Ev.ServiceBus.UnitTests/Reception/ReceptionTest.cs index 98ae4ad..968c80a 100644 --- a/tests/Ev.ServiceBus.UnitTests/Reception/ReceptionTest.cs +++ b/tests/Ev.ServiceBus.UnitTests/Reception/ReceptionTest.cs @@ -1,4 +1,5 @@ using System; +using System.Collections.Generic; using System.Linq; using System.Threading; using System.Threading.Tasks; @@ -193,6 +194,89 @@ public async Task CancellationTokenIsPassedDownToHandlers() await SimulateEventReception(client, tokenSource.Token); } + [Fact] + public async Task MessageIsAbandonedAndNotHandledWhenCancellationIsRequested() + { + var (client, eventStore) = await ComposeSubscribedReception(); + var receiver = new Mock(); + using var tokenSource = new CancellationTokenSource(); + tokenSource.Cancel(); + + await client.TriggerMessageReception(CreateSubscribedMessage(), tokenSource.Token, receiver.Object); + + using (new AssertionScope()) + { + receiver.Verify( + r => r.AbandonMessageAsync( + It.IsAny(), + It.IsAny>(), + CancellationToken.None), + Times.Once); + eventStore.Events.Should().BeEmpty(); + } + } + + [Fact] + public async Task MessageIsNotAbandonedWhenCancellationIsNotRequested() + { + var (client, eventStore) = await ComposeSubscribedReception(); + var receiver = new Mock(); + + await client.TriggerMessageReception(CreateSubscribedMessage(), CancellationToken.None, receiver.Object); + + using (new AssertionScope()) + { + receiver.Verify( + r => r.AbandonMessageAsync( + It.IsAny(), + It.IsAny>(), + It.IsAny()), + Times.Never); + eventStore.Events.Should().ContainSingle(o => o.HandlerType == typeof(SubscribedPayloadHandler)); + } + } + + private static async Task<(ProcessorMock Client, EventStore EventStore)> ComposeSubscribedReception() + { + var composer = new Composer(); + var eventStore = new EventStore(); + + composer.WithAdditionalServices( + services => + { + services.RegisterServiceBusReception() + .FromSubscription( + "testTopic", + "testSubscription", + builder => + { + builder.RegisterReception() + .CustomizePayloadTypeId("MyEvent"); + }); + + services.AddSingleton(eventStore); + }); + + await composer.Compose(); + return (composer.ClientFactory.GetProcessorMock("testTopic", "testSubscription"), eventStore); + } + + private static ServiceBusMessage CreateSubscribedMessage() + { + var parser = new TextJsonPayloadSerializer(); + var result = parser.SerializeBody(new { SomeString = "hello", SomeNumber = 36 }); + return new ServiceBusMessage(result.Body) + { + ContentType = result.ContentType, + Subject = "An integration event of type 'MyEvent'", + ApplicationProperties = + { + { UserProperties.MessageTypeProperty, "IntegrationEvent" }, + { UserProperties.PayloadTypeIdProperty, "MyEvent" } + } + }; + } + [Fact] public async Task MaxAutoRenewDurationIsSet() {