From 44c0f0761c1ba3761388e2591644fc78f6c2659d Mon Sep 17 00:00:00 2001 From: Daniel Marbach Date: Thu, 30 Jul 2026 20:32:44 +0200 Subject: [PATCH 1/4] Dedupe exception events on the publish failure path (#1967) When the send throws and publisher confirmations are enabled, MaybeHandleExceptionWithEnabledPublisherConfirmations faults the confirm TCS with the exception and the finally's confirmation await re-raises that same instance. The send catch and the finally's catch therefore both called SetActivityError on the same exception, so a publish on a closed connection recorded AlreadyClosedException twice. Track the exception recorded by the send catch and gate the finally's catch on a ReferenceEquals filter, so the same instance is recorded at most once while a distinct exception from the confirmation await is still captured. Filtering on reference equality is safe here: the TCS is faulted with the original exception object and awaited via ExceptionDispatchInfo, so the re-raised instance is the one stored. The conservative failure mode is a missed dedupe (two events), never a dropped real error. --- .../RabbitMQ.Client/Impl/Channel.BasicPublish.cs | 12 +++++++++++- 1 file changed, 11 insertions(+), 1 deletion(-) diff --git a/projects/RabbitMQ.Client/Impl/Channel.BasicPublish.cs b/projects/RabbitMQ.Client/Impl/Channel.BasicPublish.cs index c31fdc371..308d13693 100644 --- a/projects/RabbitMQ.Client/Impl/Channel.BasicPublish.cs +++ b/projects/RabbitMQ.Client/Impl/Channel.BasicPublish.cs @@ -108,6 +108,15 @@ await MaybeAcquirePublisherConfirmationLockAsync(cancellationToken) using Activity? sendActivity = RabbitMQActivitySource.PublisherHasListeners ? RabbitMQActivitySource.BasicPublish(routingKey, exchange, body.Length, basicProperties) : default; + /* + * Tracks the exception (if any) already recorded on sendActivity by the + * catch below, so the finally's confirmation-await catch does not record + * the same instance twice. When MaybeHandleExceptionWithEnabledPublisherConfirmations + * faults the confirm TCS, the finally's await re-raises that exception; + * without this guard a publish whose send failed (e.g. on a closed + * connection) recorded the same exception twice. See issue #1967. + */ + Exception? recordedSendError = null; try { publisherConfirmationInfo = MaybeStartPublisherConfirmationTracking(); @@ -133,6 +142,7 @@ await ModelSendAsync(in cmd, in props, body, bodyOwner, cancellationToken) catch (Exception ex) { sendActivity.SetActivityError(ex); + recordedSendError = ex; bool exceptionWasHandled = MaybeHandleExceptionWithEnabledPublisherConfirmations(publisherConfirmationInfo, ex); @@ -157,7 +167,7 @@ await ModelSendAsync(in cmd, in props, body, bodyOwner, cancellationToken) await MaybeEndPublisherConfirmationTrackingAsync(publisherConfirmationInfo, cancellationToken) .ConfigureAwait(false); } - catch (Exception ex) + catch (Exception ex) when (!ReferenceEquals(ex, recordedSendError)) { sendActivity.SetActivityError(ex); throw; From 27200908001471650c1ac8a28d200278339dd718 Mon Sep 17 00:00:00 2001 From: Daniel Marbach Date: Thu, 30 Jul 2026 20:33:03 +0200 Subject: [PATCH 2/4] Apply all three SetActivityError signals consistently error.type was gated behind IsAllDataRequested while the exception event and Error status were set unconditionally. With a listener sampling only PropagationData (IsAllDataRequested == false) the span received the exception event and the Error status but not error.type - the expensive signals recorded and the cheap string tag dropped. error.type is a single SetTag of a fully-qualified type name, so it is no more expensive than the signals already firing; gate them together or not at all. Set it unconditionally so a sampled failure still reports the one Stable messaging attribute. See issue #1967. --- .../Impl/RabbitMQActivitySource.cs | 16 +++++++++++----- 1 file changed, 11 insertions(+), 5 deletions(-) diff --git a/projects/RabbitMQ.Client/Impl/RabbitMQActivitySource.cs b/projects/RabbitMQ.Client/Impl/RabbitMQActivitySource.cs index cc3ffe480..11b7fd11e 100644 --- a/projects/RabbitMQ.Client/Impl/RabbitMQActivitySource.cs +++ b/projects/RabbitMQ.Client/Impl/RabbitMQActivitySource.cs @@ -345,13 +345,19 @@ internal static void SetActivityError(this Activity? activity, Exception excepti return; } + /* + * All three signals fire together so they stay consistent across sampling + * levels. AddException and SetStatus are cheap and already execute when + * IsAllDataRequested is false (a listener sampling PropagationData still + * receives the event and the status), so gating error.type - a single + * string tag - would record the expensive signals and drop the cheap one. + * That left a span marked Error with an exception event but no error.type, + * which is the only Stable attribute in the messaging convention. See + * issue #1967. + */ activity.AddException(exception); activity.SetStatus(ActivityStatusCode.Error, exception.Message); - - if (activity.IsAllDataRequested) - { - activity.SetTag(ErrorType, exception.GetType().FullName); - } + activity.SetTag(ErrorType, exception.GetType().FullName); } internal static void SetNetworkTags(this Activity? activity, IFrameHandler frameHandler) From 17f82c523c3dfa3185c7a5d4528c465a80eb37ce Mon Sep 17 00:00:00 2001 From: Daniel Marbach Date: Thu, 30 Jul 2026 20:33:51 +0200 Subject: [PATCH 3/4] Do not record caller-initiated cancellation as an error A caller cancelling its own token is not a messaging or connection failure, but the error paths recorded OperationCanceledException on the span all the same: the publish span (now widened to cover the confirmation await, so the finally's await surfaces it), the connection span, and each tcp-connection-attempt span. That inflated error rates for graceful shutdown and client-side timeouts. Use exception filters (when cancellationToken.IsCancellationRequested) on OperationCanceledException so caller cancellation rethrows without being recorded, while genuine timeouts and broker failures still set all three error signals. The publish send-catch uses an inline guard rather than a filter because confirmation tracking cleanup (faulting the TCS, decrementing the sequence number) must still run on cancellation, and a filter that skipped the catch would skip that cleanup too. See issue #1967. --- projects/RabbitMQ.Client/ConnectionFactory.cs | 14 +++++------ .../IEndpointResolverExtensions.cs | 10 ++++---- .../Impl/Channel.BasicPublish.cs | 23 +++++++++++++++++-- 3 files changed, 32 insertions(+), 15 deletions(-) diff --git a/projects/RabbitMQ.Client/ConnectionFactory.cs b/projects/RabbitMQ.Client/ConnectionFactory.cs index 29365f548..9147140b2 100644 --- a/projects/RabbitMQ.Client/ConnectionFactory.cs +++ b/projects/RabbitMQ.Client/ConnectionFactory.cs @@ -575,17 +575,15 @@ public async Task CreateConnectionAsync(IEndpointResolver endpointR .ConfigureAwait(false); } } + catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested) + { + // Caller-initiated cancellation is not a connection failure. + throw; + } catch (OperationCanceledException ex) { connectionActivity.SetActivityError(ex); - if (cancellationToken.IsCancellationRequested) - { - throw; - } - else - { - throw new BrokerUnreachableException(ex); - } + throw new BrokerUnreachableException(ex); } catch (Exception ex) { diff --git a/projects/RabbitMQ.Client/IEndpointResolverExtensions.cs b/projects/RabbitMQ.Client/IEndpointResolverExtensions.cs index 2d141ab7d..1b2f43c63 100644 --- a/projects/RabbitMQ.Client/IEndpointResolverExtensions.cs +++ b/projects/RabbitMQ.Client/IEndpointResolverExtensions.cs @@ -52,6 +52,11 @@ public static async Task SelectOneAsync(this IEndpointResolver resolver, { return await selector(ep, cancellationToken).ConfigureAwait(false); } + catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested) + { + // Caller-initiated cancellation is not a connection attempt failure. + throw; + } catch (OperationCanceledException ex) { /* @@ -64,11 +69,6 @@ public static async Task SelectOneAsync(this IEndpointResolver resolver, * individual attempt failed. */ tcpConnection.SetActivityError(ex); - if (cancellationToken.IsCancellationRequested) - { - throw; - } - exceptions.Add(ex); } catch (Exception e) diff --git a/projects/RabbitMQ.Client/Impl/Channel.BasicPublish.cs b/projects/RabbitMQ.Client/Impl/Channel.BasicPublish.cs index 308d13693..f73259bcf 100644 --- a/projects/RabbitMQ.Client/Impl/Channel.BasicPublish.cs +++ b/projects/RabbitMQ.Client/Impl/Channel.BasicPublish.cs @@ -141,8 +141,21 @@ await ModelSendAsync(in cmd, in props, body, bodyOwner, cancellationToken) } catch (Exception ex) { - sendActivity.SetActivityError(ex); - recordedSendError = ex; + /* + * Caller-initiated cancellation is not a publish failure, so it is + * not recorded on the span. Confirmation tracking still needs the + * cleanup below (faulting the TCS, decrementing the sequence number), + * which is why this is an inline guard rather than a `when` filter: + * a filter that skipped this catch would skip the cleanup too. + * See issue #1967. + */ + bool isCallerCancellation = + ex is OperationCanceledException && cancellationToken.IsCancellationRequested; + if (!isCallerCancellation) + { + sendActivity.SetActivityError(ex); + recordedSendError = ex; + } bool exceptionWasHandled = MaybeHandleExceptionWithEnabledPublisherConfirmations(publisherConfirmationInfo, ex); @@ -167,6 +180,12 @@ await ModelSendAsync(in cmd, in props, body, bodyOwner, cancellationToken) await MaybeEndPublisherConfirmationTrackingAsync(publisherConfirmationInfo, cancellationToken) .ConfigureAwait(false); } + catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested) + { + // Caller-initiated cancellation during the confirmation await is + // not a publish failure. See issue #1967. + throw; + } catch (Exception ex) when (!ReferenceEquals(ex, recordedSendError)) { sendActivity.SetActivityError(ex); From 5f6bdaa20c75c5b70411a46a7b5a8efe8ae85eff Mon Sep 17 00:00:00 2001 From: Daniel Marbach Date: Thu, 30 Jul 2026 20:35:23 +0200 Subject: [PATCH 4/4] Strengthen failure-recording test assertions and pin publish parenting HasRecordedException read Events.First(), so it could not detect a span that recorded the same exception twice - exactly the duplicate the publish failure path produced. Filter to exception events and assert exactly one, so any future duplicate recording fails the test. The ambient-pollution test set VerifyParent = false on all recorders, leaving parenting unasserted. The publish it performs runs after the app activity's scope has ended, so its parent is root; assert that, so a parenting regression on the frame-writing path (the same path the ownership fix touches) would surface here rather than only in TestOpenTelemetry. See issue #1967. --- projects/Test/Common/ActivityRecorder.cs | 12 ++++++++++-- .../Test/SequentialIntegration/TestActivitySource.cs | 9 +++++++++ 2 files changed, 19 insertions(+), 2 deletions(-) diff --git a/projects/Test/Common/ActivityRecorder.cs b/projects/Test/Common/ActivityRecorder.cs index bc6d81765..f44f25b68 100644 --- a/projects/Test/Common/ActivityRecorder.cs +++ b/projects/Test/Common/ActivityRecorder.cs @@ -147,8 +147,16 @@ public static void HasRecordedException(this Activity activity, Exception except public static void HasRecordedException(this Activity activity, string exceptionTypeName) { - var exceptionEvent = activity.Events.First(); - Assert.Equal("exception", exceptionEvent.Name); + /* + * Assert exactly one exception event so duplicate recordings are caught. + * A publish whose send failed on a closed connection used to record the + * same exception twice (once in the send catch, once when the confirmation + * await re-raised it), which Events.First() alone does not detect. + * See issue #1967. + */ + var exceptionEvents = activity.Events.Where(e => e.Name == "exception").ToList(); + Assert.Single(exceptionEvents); + ActivityEvent exceptionEvent = exceptionEvents[0]; Assert.Equal(exceptionTypeName, exceptionEvent.Tags.SingleOrDefault(t => t.Key == "exception.type").Value); } diff --git a/projects/Test/SequentialIntegration/TestActivitySource.cs b/projects/Test/SequentialIntegration/TestActivitySource.cs index 7f3256ffc..6a21815cd 100644 --- a/projects/Test/SequentialIntegration/TestActivitySource.cs +++ b/projects/Test/SequentialIntegration/TestActivitySource.cs @@ -480,6 +480,15 @@ public async Task TestAmqpOperationsDoNotTagAnUnrelatedAmbientActivity_GH1967() publishActivity.HasTag("messaging.message.envelope.size"); publishActivity.HasTag("server.port"); publishActivity.HasTag("network.peer.address"); + + /* + * The publish runs after the app activity's scope has ended, so it has no + * parent. Asserting that pins the one parenting relationship this test can + * speak to: the library's publish span must not attach to an ambient span + * it does not belong to. The ownership fix is about tags, but a parenting + * regression on the frame-writing path would show up here. See issue #1967. + */ + Assert.Null(publishActivity.Parent); } [Fact]