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,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.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.
+
 ## 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,25 @@ namespace Ev.ServiceBus.Apm;
 /// </summary>
 public class ApmTransactionManager : ITransactionManager, ICancellationAwareTransactionManager
 {
+    // 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.
+    // Capped at 1000 entries as a safety net: normal pod shutdown produces ~50 entries; in any
+    // edge case where the processor is stopped and restarted mid-lifecycle the cap prevents unbounded growth.
+    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 +94,32 @@ 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;
+        // Soft cap: Count + TryAdd are not atomic, so the dict can slightly exceed the limit under
+        // concurrent shutdown. This is intentional — the cap is a safety net against unbounded growth
+        // in edge cases, not a strict hard limit. Normal shutdown adds ~50 entries at most.
+        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) =>
+            error.TransactionId is not null && _cancelledTransactionIds.ContainsKey(error.TransactionId)
+                ? null
+                : error);
     }
 
     private static bool IsTraceEnabled()