fix(#491328): abandon message when cancellation is requested in Messa…

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.

src/Ev.ServiceBus/Reception/MessageReceptionHandler.cs

@@ -70,7 +70,11 @@
                     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);
@@ -78,8 +82,8 @@
 
                 ServiceBusMeter.RecordDeliveryCount(
                    context.Message.DeliveryCount,
                         executionContext.ClientType,
                         executionContext.ResourceId,
                         executionContext.PayloadTypeId);
 
                 ServiceBusMeter.RecordMessageQueueLatency(
@@ -129,6 +133,19 @@
         }
     }
 
+    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();

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<ServiceBusReceiver>();
+        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<ServiceBusReceivedMessage>(),
+                    It.IsAny<IDictionary<string, object>>(),
+                    CancellationToken.None),
+                Times.Once);
+            eventStore.Events.Should().BeEmpty();
+        }
+    }
+
+    [Fact]
+    public async Task MessageIsNotAbandonedWhenCancellationIsNotRequested()
+    {
+        var (client, eventStore) = await ComposeSubscribedReception();
+        var receiver = new Mock<ServiceBusReceiver>();
+
+        await client.TriggerMessageReception(CreateSubscribedMessage(), CancellationToken.None, receiver.Object);
+
+        using (new AssertionScope())
+        {
+            receiver.Verify(
+                r => r.AbandonMessageAsync(
+                    It.IsAny<ServiceBusReceivedMessage>(),
+                    It.IsAny<IDictionary<string, object>>(),
+                    It.IsAny<CancellationToken>()),
+                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<SubscribedEvent, SubscribedPayloadHandler>()
+                                .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()
     {