Skip to content

Commit 81925d8

Browse files
ecofrankieRobert Karpclaude
authored
fix: suppress APM error events for receive-loop cancellations during shutdown (5.7.5) (#116)
* 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> * fix: suppress auto-instrumented AmqpReceiver TaskCanceledException APM errors during shutdown (5.7.5) Root cause: Elastic APM creates "AzureServiceBus RECEIVE" transactions via DiagnosticSource auto-instrumentation. The Azure SDK ends its Activity before firing ProcessErrorAsync, so Agent.Tracer.CurrentTransaction is null in OnReceiveCancelled() — the TX ID is never tracked and the filter passes the error through. Fix: add culprit-based suppression for TaskCanceledException on the AmqpReceiver path (post-WebSockets fix, only fires on shutdown). Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com> * fix: address review feedback — TryRemove, AmqpReceiverCulprit const, cap comment, filter tests - Replace ContainsKey with TryRemove in the filter so matched transaction IDs are consumed on first use, keeping the dictionary lean and avoiding unnecessary cap pressure - Extract 'AmqpReceiver' magic string to private const AmqpReceiverCulprit with comment documenting the Azure SDK source and why the culprit path is safe post-WebSockets fix - Extract ShouldSuppressError as an internal static method for unit testability - Add inline comment at the CancelledTransactionIdCap guard explaining the culprit fallback - Add InternalsVisibleTo("Ev.ServiceBus.UnitTests") to Ev.ServiceBus.Apm - Add ApmTransactionManagerTests: 8 tests covering both filter paths, the TryRemove consume-once behaviour, and non-suppressed exception types Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com> * fix: address Michal's follow-up — add parens, TOCTOU comment, cap-exceeded test - Add explicit parentheses around the is-pattern in ShouldSuppressError to document operator precedence intent: (exceptionType is "TCE" or "OCE") && culprit.Contains(...) - Document the non-atomic Count+TryAdd in OnReceiveCancelled as an intentional soft cap - Add ShouldSuppressError_WhenCapExceeded_CulpritPathStillSuppresses test: seeds 1000 entries to simulate cap exhaustion, then verifies the culprit path still suppresses the error for an untracked transaction ID Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com> * fix: ensure APM error filter is registered before first message processing (5.7.5) The Service Bus hosted service starts before the Elastic APM hosted service (DI registration order in consuming apps), so Agent.IsConfigured is false when ApmTransactionManager is constructed and the constructor's filter registration is silently skipped. RegisterShutdownErrorFilter() is now also called: - At the start of RunWithInTransaction() when IsTraceEnabled() is true (agent is definitively up; O(1) no-op on all subsequent calls via Interlocked.CompareExchange guard) - Before the IsTraceEnabled() guard in OnReceiveCancelled() This guarantees the filter is in place before the first message processing completes and before any graceful-shutdown RECEIVE activity errors can fire. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com> * docs: clarify Case 1 type-safety and cap behaviour in ApmTransactionManager Address review comments from PR #116: - Case 1 has no exceptionType guard because _cancelledTransactionIds is populated exclusively via OnReceiveCancelled(), which ReceiverWrapper only calls for OperationCanceledException. - Cap overflow for non-AmqpReceiver paths won't be suppressed; this is documented as acceptable given the conditions required to reach it. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com> --------- Co-authored-by: Robert Karp <rkarp@ecovadisazure.onmicrosoft.com> Co-authored-by: Claude Sonnet 4.6 <noreply@anthropic.com>
1 parent f0e645f commit 81925d8

5 files changed

Lines changed: 244 additions & 3 deletions

File tree

‎docs/CHANGELOG.md‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -4,6 +4,12 @@ All notable changes to this project will be documented in this file.
44
The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
55
and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
66

7+
## 5.7.5
8+
- Fixed
9+
- `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.
10+
- **Registration reliability fix (5.7.5-preview4):** the constructor registration silently fails when the Elastic APM hosted service starts *after* the Service Bus hosted service (the typical DI registration order). `RegisterShutdownErrorFilter()` is now also called at the start of every `RunWithInTransaction()` call (guarded by `Interlocked.CompareExchange` — O(1) no-op after the first success) and before the `IsTraceEnabled()` guard in `OnReceiveCancelled()`, ensuring the filter is in place before the first message processing completes regardless of hosted-service start order.
11+
- 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.
12+
713
## 5.7.4
814
- Fixed
915
- 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‎

Lines changed: 108 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,8 @@
11
using System;
2+
using System.Collections.Concurrent;
23
using System.Collections.Generic;
34
using System.Diagnostics;
5+
using System.Threading;
46
using System.Threading.Tasks;
57
using Elastic.Apm;
68
using Elastic.Apm.Api;
@@ -14,10 +16,52 @@ namespace Ev.ServiceBus.Apm;
1416
/// </summary>
1517
public class ApmTransactionManager : ITransactionManager, ICancellationAwareTransactionManager
1618
{
19+
// Static: Agent.AddFilter is process-wide; all ApmTransactionManager instances (one per consumer)
20+
// must share one filter registration and one cancelled-transaction set.
21+
// Note for tests: static state is shared across test runs within the same process.
22+
// Call ResetForTests() in test setup/teardown to isolate tests from each other.
23+
24+
// Tracks transaction IDs for which the ASB ProcessErrorAsync callback fired an OperationCanceledException
25+
// (the standard signal that the receive loop is being stopped, most commonly during pod graceful shutdown).
26+
// The error filter below suppresses APM error events for these transactions so that
27+
// shutdown-induced TaskCanceledException entries do not appear in APM.
28+
//
29+
// IDs are removed from this set after the filter matches them (TryRemove) to keep the dictionary lean.
30+
// Cap behaviour: entries beyond CancelledTransactionIdCap are not tracked, so their error events
31+
// fall through to the culprit-based filter path (Case 2 in ShouldSuppressError). This is an accepted
32+
// tradeoff — reaching 1000 entries requires ~20 consecutive graceful processor stop/start cycles on
33+
// the same pod instance, which does not occur in normal Kubernetes rolling-deploy scenarios.
34+
// If the cap were reached by non-AmqpReceiver OperationCanceledException paths (i.e. user-code
35+
// cancellations unrelated to AMQP shutdown), those excess error events would not be suppressed by
36+
// either Case 1 or Case 2 and would reach the APM server. This is acceptable: the conditions
37+
// required to reach the cap via that path are not reachable in practice.
38+
private static readonly ConcurrentDictionary<string, byte> _cancelledTransactionIds = new();
39+
private const int CancelledTransactionIdCap = 1000;
40+
private static int _filterRegistered; // 0 = not registered, 1 = registered (Interlocked.CompareExchange requires int)
41+
42+
// Azure.Messaging.ServiceBus internal class name that appears in the APM error culprit when
43+
// the AMQP receive loop is cancelled. After switching to WebSockets transport (PR #202138),
44+
// this culprit only fires during pod graceful shutdown — not from network drops (which surface
45+
// as ServiceBusException, not TaskCanceledException).
46+
private const string AmqpReceiverCulprit = "AmqpReceiver";
47+
48+
public ApmTransactionManager()
49+
{
50+
// Best-effort registration at construction time. If the Elastic APM hosted service starts
51+
// AFTER the Service Bus hosted service (the typical registration order), Agent.IsConfigured
52+
// is still false here and registration is deferred to the first RunWithInTransaction call.
53+
RegisterShutdownErrorFilter();
54+
}
55+
1756
public async Task RunWithInTransaction(MessageExecutionContext executionContext, Func<Task> transaction)
1857
{
1958
if (IsTraceEnabled())
2059
{
60+
// Ensure the filter is registered before any message processing completes.
61+
// This is the reliable registration point: by the time IsTraceEnabled() returns true,
62+
// Agent.IsConfigured is guaranteed true — covering the case where the APM hosted service
63+
// started after the Service Bus hosted service and the constructor registration was skipped.
64+
RegisterShutdownErrorFilter();
2165
Agent.Tracer.CurrentTransaction.Name = executionContext.ExecutionName;
2266
Agent.Tracer.CurrentTransaction.SetLabel(
2367
nameof(executionContext.ClientType),
@@ -73,10 +117,71 @@ private static List<SpanLink> GetSpanLinks(string? diagnosticId)
73117

74118
public void OnReceiveCancelled()
75119
{
76-
if (IsTraceEnabled())
77-
Agent.Tracer.CurrentTransaction.Outcome = Outcome.Success;
120+
// Attempt registration before checking IsTraceEnabled — the filter must be in place even
121+
// for auto-instrumented "AzureServiceBus RECEIVE" transactions where CurrentTransaction is
122+
// null (Case 2). After this point it is too late for the current error batch, but
123+
// registering here ensures coverage if RunWithInTransaction was never reached.
124+
RegisterShutdownErrorFilter();
125+
126+
if (!IsTraceEnabled())
127+
return;
128+
129+
var tx = Agent.Tracer.CurrentTransaction;
130+
if (tx is null) return;
131+
tx.Outcome = Outcome.Success;
132+
// Count check and TryAdd are not atomic — under high concurrency the dictionary may reach
133+
// CancelledTransactionIdCap + N entries. The cap is a soft limit; this is intentional.
134+
if (_cancelledTransactionIds.Count < CancelledTransactionIdCap)
135+
_cancelledTransactionIds.TryAdd(tx.Id, 0);
136+
// If Count >= CancelledTransactionIdCap, this ID is not tracked here.
137+
// The culprit-based path (Case 2 in ShouldSuppressError) still suppresses the error.
78138
}
79139

140+
private static void RegisterShutdownErrorFilter()
141+
{
142+
if (!Agent.IsConfigured || Interlocked.CompareExchange(ref _filterRegistered, 1, 0) != 0)
143+
return;
144+
145+
// Returning null from the filter drops the error event before it reaches the APM server.
146+
Agent.AddFilter((IError error) =>
147+
ShouldSuppressError(error.TransactionId, error.Exception?.Type, error.Culprit)
148+
? null
149+
: error);
150+
}
151+
152+
// Extracted for unit testability. Called by the APM filter lambda.
153+
internal static bool ShouldSuppressError(string? transactionId, string? exceptionType, string? culprit)
154+
{
155+
// Case 1: transaction explicitly tracked via OnReceiveCancelled().
156+
// TryRemove keeps the dictionary lean — matched IDs are consumed on first use.
157+
// No exceptionType guard is applied here: _cancelledTransactionIds is populated exclusively
158+
// by OnReceiveCancelled(), which ReceiverWrapper only calls for OperationCanceledException.
159+
// Any ID present in this set therefore already originates from a cancellation path.
160+
if (transactionId is not null && _cancelledTransactionIds.TryRemove(transactionId, out _))
161+
return true;
162+
163+
// Case 2: Elastic APM auto-instrumented "AzureServiceBus RECEIVE" transactions.
164+
// The Azure SDK ends its Activity (and therefore the APM transaction) before firing
165+
// ProcessErrorAsync, so Agent.Tracer.CurrentTransaction is null when OnReceiveCancelled()
166+
// runs — the transaction ID is never added to _cancelledTransactionIds.
167+
// After switching to WebSockets transport, TaskCanceledException originating in
168+
// AmqpReceiver.ReceiveMessagesAsyncInternal only occurs during pod graceful shutdown.
169+
return (exceptionType is "System.Threading.Tasks.TaskCanceledException"
170+
or "System.OperationCanceledException") &&
171+
culprit?.Contains(AmqpReceiverCulprit, StringComparison.Ordinal) == true;
172+
}
173+
174+
// For test isolation only — resets static state between test runs in the same process.
175+
internal static void ResetForTests()
176+
{
177+
_cancelledTransactionIds.Clear();
178+
Interlocked.Exchange(ref _filterRegistered, 0);
179+
}
180+
181+
// For test setup only — seeds a transaction ID as if OnReceiveCancelled() had recorded it.
182+
internal static void AddCancelledTransactionIdForTests(string id) =>
183+
_cancelledTransactionIds.TryAdd(id, 0);
184+
80185
private static bool IsTraceEnabled()
81186
=> Agent.IsConfigured && Agent.Config.Enabled && Agent.Tracer is not null && Agent.Tracer.CurrentTransaction is not null;
82-
}
187+
}

‎src/Ev.ServiceBus.Apm/Ev.ServiceBus.Apm.csproj‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -18,4 +18,8 @@
1818
<PackageReference Include="Elastic.Apm" Version="1.19.0" />
1919
</ItemGroup>
2020

21+
<ItemGroup>
22+
<InternalsVisibleTo Include="Ev.ServiceBus.UnitTests" />
23+
</ItemGroup>
24+
2125
</Project>

‎tests/Ev.ServiceBus.UnitTests/ApmTransactionManagerTests.cs‎

Lines changed: 125 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,125 @@
1+
using System;
2+
using Ev.ServiceBus.Apm;
3+
using Xunit;
4+
5+
namespace Ev.ServiceBus.UnitTests;
6+
7+
public sealed class ApmTransactionManagerTests : IDisposable
8+
{
9+
public ApmTransactionManagerTests()
10+
{
11+
ApmTransactionManager.ResetForTests();
12+
}
13+
14+
public void Dispose()
15+
{
16+
ApmTransactionManager.ResetForTests();
17+
}
18+
19+
[Fact]
20+
public void ShouldSuppressError_WithTrackedTransactionId_ReturnsTrue()
21+
{
22+
ApmTransactionManager.AddCancelledTransactionIdForTests("tx-abc");
23+
24+
var result = ApmTransactionManager.ShouldSuppressError(
25+
"tx-abc",
26+
"System.Threading.Tasks.TaskCanceledException",
27+
"some.culprit");
28+
29+
Assert.True(result);
30+
}
31+
32+
[Fact]
33+
public void ShouldSuppressError_WithTrackedTransactionId_RemovesIdAfterMatch()
34+
{
35+
ApmTransactionManager.AddCancelledTransactionIdForTests("tx-abc");
36+
ApmTransactionManager.ShouldSuppressError("tx-abc", "System.Threading.Tasks.TaskCanceledException", "some.culprit");
37+
38+
var secondResult = ApmTransactionManager.ShouldSuppressError(
39+
"tx-abc",
40+
"System.Threading.Tasks.TaskCanceledException",
41+
"some.culprit");
42+
43+
Assert.False(secondResult);
44+
}
45+
46+
[Fact]
47+
public void ShouldSuppressError_WithAmqpReceiverCulpritAndTaskCanceledException_ReturnsTrue()
48+
{
49+
var result = ApmTransactionManager.ShouldSuppressError(
50+
"untracked-tx",
51+
"System.Threading.Tasks.TaskCanceledException",
52+
"Azure.Messaging.ServiceBus.Amqp.AmqpReceiver+<ReceiveMessagesAsyncInternal>d__45");
53+
54+
Assert.True(result);
55+
}
56+
57+
[Fact]
58+
public void ShouldSuppressError_WithAmqpReceiverCulpritAndOperationCanceledException_ReturnsTrue()
59+
{
60+
var result = ApmTransactionManager.ShouldSuppressError(
61+
"untracked-tx",
62+
"System.OperationCanceledException",
63+
"Azure.Messaging.ServiceBus.Amqp.AmqpReceiver+SomeInternalMethod");
64+
65+
Assert.True(result);
66+
}
67+
68+
[Fact]
69+
public void ShouldSuppressError_WithAmqpReceiverCulpritButServiceBusException_ReturnsFalse()
70+
{
71+
var result = ApmTransactionManager.ShouldSuppressError(
72+
"untracked-tx",
73+
"Azure.Messaging.ServiceBus.ServiceBusException",
74+
"Azure.Messaging.ServiceBus.Amqp.AmqpReceiver+SomeInternalMethod");
75+
76+
Assert.False(result);
77+
}
78+
79+
[Fact]
80+
public void ShouldSuppressError_WithNonAmqpCulpritAndTaskCanceledException_ReturnsFalse()
81+
{
82+
var result = ApmTransactionManager.ShouldSuppressError(
83+
null,
84+
"System.Threading.Tasks.TaskCanceledException",
85+
"Some.Other.Namespace.SomeClass+SomeMethod");
86+
87+
Assert.False(result);
88+
}
89+
90+
[Fact]
91+
public void ShouldSuppressError_WithUntrackedIdAndNonCancelledException_ReturnsFalse()
92+
{
93+
var result = ApmTransactionManager.ShouldSuppressError(
94+
"not-tracked",
95+
"System.InvalidOperationException",
96+
"Some.Class+SomeMethod");
97+
98+
Assert.False(result);
99+
}
100+
101+
[Fact]
102+
public void ShouldSuppressError_WithNullTransactionIdAndNullCulprit_ReturnsFalse()
103+
{
104+
var result = ApmTransactionManager.ShouldSuppressError(null, null, null);
105+
106+
Assert.False(result);
107+
}
108+
109+
[Fact]
110+
public void ShouldSuppressError_WhenCapExceeded_CulpritPathStillSuppresses()
111+
{
112+
// Simulate the cap-exceeded scenario: OnReceiveCancelled stops tracking IDs once the
113+
// dictionary is full. Errors for those untracked transactions must still be suppressed
114+
// by the culprit-based path (Case 2).
115+
for (var i = 0; i < 1000; i++)
116+
ApmTransactionManager.AddCancelledTransactionIdForTests($"capped-tx-{i}");
117+
118+
var result = ApmTransactionManager.ShouldSuppressError(
119+
"untracked-due-to-cap",
120+
"System.Threading.Tasks.TaskCanceledException",
121+
"Azure.Messaging.ServiceBus.Amqp.AmqpReceiver+<ReceiveMessagesAsyncInternal>d__45");
122+
123+
Assert.True(result);
124+
}
125+
}

‎tests/Ev.ServiceBus.UnitTests/Ev.ServiceBus.UnitTests.csproj‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -14,6 +14,7 @@
1414
<ItemGroup>
1515
<ProjectReference Include="..\..\src\Ev.ServiceBus.Abstractions\Ev.ServiceBus.Abstractions.csproj" />
1616
<ProjectReference Include="..\..\src\Ev.ServiceBus\Ev.ServiceBus.csproj" />
17+
<ProjectReference Include="..\..\src\Ev.ServiceBus.Apm\Ev.ServiceBus.Apm.csproj" />
1718
<ProjectReference Include="..\Ev.ServiceBus.TestHelpers\Ev.ServiceBus.TestHelpers.csproj" />
1819
</ItemGroup>
1920
</Project>

0 commit comments

Comments
 (0)