fix: suppress APM error events for receive-loop cancellations during shutdown (5.7.5)

Setting Outcome=Success (5.7.4) was insufficient: Elastic APM captures error
events at the DiagnosticSource level before ReceiverWrapper runs, so the error
document was already queued regardless of the outcome override.

Registers a one-time Agent.AddFilter(IError) that drops error events whose
TransactionId matches a cancelled-receive transaction, preventing them from
reaching the APM server.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

Author: Robert Karp

docs/CHANGELOG.md

@@ -4,6 +4,11 @@ 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.7.5
+- Fixed
+  - `ApmTransactionManager` now registers the APM error filter **at construction time** (application startup) instead of lazily inside `OnReceiveCancelled()`. The lazy approach lost a race: during pod graceful shutdown the APM agent flushes its internal buffer concurrently with Service Bus processor teardown, so error events could be sent to APM before `ReceiverWrapper.OnExceptionOccured` ran and had a chance to register the filter. Registering at construction time — before any message processing starts — closes this window. A fallback call in `OnReceiveCancelled()` handles the edge case where the APM agent was not yet configured at construction.
+  - The filter now also suppresses `TaskCanceledException` / `OperationCanceledException` errors whose culprit originates in `AmqpReceiver.ReceiveMessagesAsyncInternal`. These error events are produced by Elastic APM's **auto-instrumented** Azure Service Bus transactions (`"AzureServiceBus RECEIVE from …"`): the Azure SDK ends its underlying `Activity` (and therefore the APM transaction) before calling `ProcessErrorAsync`, so `Agent.Tracer.CurrentTransaction` is already `null` when `OnReceiveCancelled()` runs — the transaction ID is never added to `_cancelledTransactionIds` and the transaction-ID-based filter path cannot suppress them. After switching to WebSockets transport, `TaskCanceledException` from this code path only occurs during pod graceful shutdown.
+
 ## 5.7.4
 - Fixed
   - Prevented `OperationCanceledException` during pod graceful shutdown from being recorded as APM errors. Added `ICancellationAwareTransactionManager` — an optional interface that `ITransactionManager` implementations can implement to react to receive-loop cancellations. `ApmTransactionManager` implements it by setting the current Elastic APM transaction outcome to `Success`, overriding the error state set by the Azure SDK's auto-instrumentation. `ReceiverWrapper` calls `OnReceiveCancelled()` via a runtime cast before logging the shutdown warning.

src/Ev.ServiceBus.Apm/ApmTransactionManager.cs

@@ -1,6 +1,8 @@
 using System;
+using System.Collections.Concurrent;
 using System.Collections.Generic;
 using System.Diagnostics;
+using System.Threading;
 using System.Threading.Tasks;
 using Elastic.Apm;
 using Elastic.Apm.Api;
@@ -14,6 +16,31 @@ namespace Ev.ServiceBus.Apm;
 /// </summary>
 public class ApmTransactionManager : ITransactionManager, ICancellationAwareTransactionManager
 {
+    // Static: Agent.AddFilter is process-wide; all ApmTransactionManager instances (one per consumer)
+    // must share one filter registration and one cancelled-transaction set.
+
+    // Tracks transaction IDs for which the ASB ProcessErrorAsync callback fired an OperationCanceledException
+    // (the standard signal that the receive loop is being stopped, most commonly during pod graceful shutdown).
+    // The error filter below suppresses APM error events for these transactions so that
+    // shutdown-induced TaskCanceledException entries do not appear in APM.
+    //
+    // Cap behaviour: entries beyond CancelledTransactionIdCap are not tracked, so their error events
+    // pass through the filter. This is an accepted tradeoff — reaching 1000 entries requires ~20
+    // consecutive graceful processor stop/start cycles on the same pod instance, which does not occur
+    // in normal Kubernetes rolling-deploy scenarios where the pod is replaced after each shutdown.
+    private static readonly ConcurrentDictionary<string, byte> _cancelledTransactionIds = new();
+    private const int CancelledTransactionIdCap = 1000;
+    private static int _filterRegistered; // 0 = not registered, 1 = registered
+
+    public ApmTransactionManager()
+    {
+        // Register the shutdown-cancellation error filter at construction time (application startup),
+        // not lazily on first OnReceiveCancelled(). During pod graceful shutdown the APM agent flushes
+        // its buffer concurrently with Service Bus processor teardown — registering the filter after the
+        // first OperationCanceledException fires loses that race and lets error events escape to APM.
+        RegisterShutdownErrorFilter();
+    }
+
     public async Task RunWithInTransaction(MessageExecutionContext executionContext, Func<Task> transaction)
     {
         if (IsTraceEnabled())
@@ -73,8 +100,45 @@ private static List<SpanLink> GetSpanLinks(string? diagnosticId)
 
     public void OnReceiveCancelled()
     {
-        if (IsTraceEnabled())
-            Agent.Tracer.CurrentTransaction.Outcome = Outcome.Success;
+        if (!IsTraceEnabled())
+            return;
+
+        var tx = Agent.Tracer.CurrentTransaction;
+        if (tx is null) return;
+        tx.Outcome = Outcome.Success;
+        if (_cancelledTransactionIds.Count < CancelledTransactionIdCap)
+            _cancelledTransactionIds.TryAdd(tx.Id, 0);
+
+        // Fallback: if the agent was not yet configured when the constructor ran, register now.
+        RegisterShutdownErrorFilter();
+    }
+
+    private static void RegisterShutdownErrorFilter()
+    {
+        if (!Agent.IsConfigured || Interlocked.CompareExchange(ref _filterRegistered, 1, 0) != 0)
+            return;
+
+        // Returning null from the filter drops the error event before it reaches the APM server.
+        Agent.AddFilter((IError error) =>
+        {
+            // Case 1: transaction explicitly tracked via OnReceiveCancelled().
+            if (error.TransactionId is not null && _cancelledTransactionIds.ContainsKey(error.TransactionId))
+                return null;
+
+            // Case 2: Elastic APM auto-instrumented "AzureServiceBus RECEIVE" transactions.
+            // The Azure SDK ends its Activity (and therefore the APM transaction) before firing
+            // ProcessErrorAsync, so Agent.Tracer.CurrentTransaction is null by the time
+            // OnReceiveCancelled() runs — the transaction ID is never added to
+            // _cancelledTransactionIds. Identify these by culprit pattern instead.
+            // After switching to WebSockets transport, TaskCanceledException originating in
+            // AmqpReceiver.ReceiveMessagesAsyncInternal only occurs during pod graceful shutdown.
+            if (error.Exception?.Type is "System.Threading.Tasks.TaskCanceledException"
+                                       or "System.OperationCanceledException" &&
+                error.Culprit?.Contains("AmqpReceiver", StringComparison.Ordinal) == true)
+                return null;
+
+            return error;
+        });
     }
 
     private static bool IsTraceEnabled()