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 c31fdc371..f73259bcf 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(); @@ -132,7 +141,21 @@ await ModelSendAsync(in cmd, in props, body, bodyOwner, cancellationToken) } catch (Exception ex) { - sendActivity.SetActivityError(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); @@ -157,7 +180,13 @@ await ModelSendAsync(in cmd, in props, body, bodyOwner, cancellationToken) await MaybeEndPublisherConfirmationTrackingAsync(publisherConfirmationInfo, cancellationToken) .ConfigureAwait(false); } - catch (Exception ex) + 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); throw; 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) 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]