Fix the behavioural defects found in the OpenTelemetry tracing review

RELEASE.md

@@ -63,6 +63,20 @@ dotnet nuget push -k NUGET_API_KEY -s https://api.nuget.org/v3/index.json ./pack
 
 ## `main` (`7.x`) branch
 
+### Collect release-note prose
+
+`CHANGELOG.md` and `HISTORY.md` are generated by `tools/generate-changelog.sh` and only carry the mechanical PR and issue list. User-visible behaviour notes (dashboards moving, error rates changing, deprecations) live in the GitHub Release body instead.
+
+Before creating the release, sweep merged PRs in the milestone labeled `release-notes` and fold each PR's `## For the X.Y.Z release notes` section into a `## Notable behavioural changes` section of the release body:
+
+```
+gh pr list --repo rabbitmq/rabbitmq-dotnet-client \
+    --state merged --search 'milestone:X.Y.Z label:release-notes' \
+    --json number,title,url
+```
+
+### Cut the release
+
 * Close the appropriate milestone, and make a note of the link to the milestone with closed issues visible
 * Use the GitHub web UI or `gh release create` command to create the new release
 * GitHub actions will build and publish the release to NuGet

projects/RabbitMQ.Client.OpenTelemetry/TraceProviderBuilderExtensions.cs

@@ -30,6 +30,18 @@ public static TracerProviderBuilder AddRabbitMQInstrumentation(this TracerProvid
 
         private static ActivityContext OpenTelemetryContextExtractor(IReadOnlyBasicProperties props)
         {
+            /*
+             * A message with no headers at all has nothing to extract. Returning early
+             * matters: without it the getter below is called once per propagator field
+             * with a null carrier, and the correct result depends entirely on its
+             * catch block swallowing a NullReferenceException. This mirrors the
+             * null check in RabbitMQActivitySource.DefaultContextExtractor.
+             */
+            if (props.Headers is null)
+            {
+                return default;
+            }
+
             // Extract the PropagationContext of the upstream parent from the message headers.
             var parentContext = Propagators.DefaultTextMapPropagator.Extract(default, props.Headers, OpenTelemetryContextGetter);
             Baggage.Current = parentContext.Baggage;
@@ -38,16 +50,25 @@ private static ActivityContext OpenTelemetryContextExtractor(IReadOnlyBasicPrope
 
         private static IEnumerable<string> OpenTelemetryContextGetter(IDictionary<string, object> carrier, string key)
         {
+            /*
+             * Defensive only. The caller null-checks Headers, and a malformed value is
+             * handled by the `is byte[]` test rather than by throwing, so this catch is
+             * no longer load-bearing for any known input. It stays because a custom
+             * IDictionary implementation supplied through a header table could throw
+             * from TryGetValue, and a failed context extraction must not fail the
+             * delivery.
+             */
             try
             {
-                if (carrier.TryGetValue(key, out object value) && value is byte[] bytes)
+                if (carrier != null && carrier.TryGetValue(key, out object value) && value is byte[] bytes)
                 {
                     return new[] { Encoding.UTF8.GetString(bytes) };
                 }
             }
             catch (Exception)
             {
-                //this.logger.LogError(ex, "Failed to extract trace context.");
+                // Ignored: an unparseable carrier yields an unparented span, which is
+                // strictly better than propagating the failure to the consumer.
             }
 
             return Enumerable.Empty<string>();

projects/RabbitMQ.Client/ConnectionFactory.cs

@@ -575,24 +575,20 @@ public async Task<IConnection> CreateConnectionAsync(IEndpointResolver endpointR
                         .ConfigureAwait(false);
                 }
             }
+            catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested)
+            {
+                // Caller-initiated cancellation is not a connection failure.
+                throw;
+            }
             catch (OperationCanceledException ex)
             {
-                connectionActivity?.SetStatus(ActivityStatusCode.Error);
-                connectionActivity?.AddException(ex);
-                if (cancellationToken.IsCancellationRequested)
-                {
-                    throw;
-                }
-                else
-                {
-                    throw new BrokerUnreachableException(ex);
-                }
+                connectionActivity.SetActivityError(ex);
+                throw new BrokerUnreachableException(ex);
             }
             catch (Exception ex)
             {
                 var brokerUnreachableException = new BrokerUnreachableException(ex);
-                connectionActivity?.SetStatus(ActivityStatusCode.Error);
-                connectionActivity?.AddException(brokerUnreachableException);
+                connectionActivity.SetActivityError(brokerUnreachableException);
                 throw brokerUnreachableException;
             }
         }

projects/RabbitMQ.Client/ConsumerDispatching/AsyncConsumerDispatcher.cs

@@ -32,10 +32,26 @@ protected override async Task ProcessChannelAsync()
                                         using (Activity? activity = RabbitMQActivitySource.Deliver(work.RoutingKey!, work.Exchange!,
                                             work.DeliveryTag, work.BasicProperties!, work.Body.Size))
                                         {
-                                            await work.Consumer.HandleBasicDeliverAsync(
-                                                work.ConsumerTag!, work.DeliveryTag, work.Redelivered,
-                                                work.Exchange!, work.RoutingKey!, work.BasicProperties!, work.Body.Memory, work.CancellationToken)
-                                                .ConfigureAwait(false);
+                                            /*
+                                             * Record a throwing consumer callback on the deliver span
+                                             * before rethrowing to the reporting catch below. Without
+                                             * this the span is disposed on the way out and ends
+                                             * status=Unset with no exception event, so a consumer that
+                                             * throws on every message still traces as fully
+                                             * successful. See issue #1967.
+                                             */
+                                            try
+                                            {
+                                                await work.Consumer.HandleBasicDeliverAsync(
+                                                    work.ConsumerTag!, work.DeliveryTag, work.Redelivered,
+                                                    work.Exchange!, work.RoutingKey!, work.BasicProperties!, work.Body.Memory, work.CancellationToken)
+                                                    .ConfigureAwait(false);
+                                            }
+                                            catch (Exception e)
+                                            {
+                                                activity.SetActivityError(e);
+                                                throw;
+                                            }
                                         }
                                         break;
                                     case WorkType.Cancel:

projects/RabbitMQ.Client/IEndpointResolverExtensions.cs

@@ -46,16 +46,17 @@ public static async Task<T> SelectOneAsync<T>(this IEndpointResolver resolver,
             foreach (AmqpTcpEndpoint ep in resolver.All())
             {
                 cancellationToken.ThrowIfCancellationRequested();
-                using Activity? tcpConnection = RabbitMQActivitySource.OpenTcpConnection();
-                if (tcpConnection is { IsAllDataRequested: true })
-                {
-                    tcpConnection.SetServerTags(ep);
-                }
+                using Activity? tcpConnection = RabbitMQActivitySource.OpenTcpConnection(ep);
 
                 try
                 {
                     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)
                 {
                     /*
@@ -67,19 +68,12 @@ public static async Task<T> SelectOneAsync<T>(this IEndpointResolver resolver,
                      * later endpoint succeeds, the overall operation succeeded, and only the
                      * individual attempt failed.
                      */
-                    tcpConnection?.AddException(ex);
-                    tcpConnection?.SetStatus(ActivityStatusCode.Error);
-                    if (cancellationToken.IsCancellationRequested)
-                    {
-                        throw;
-                    }
-
+                    tcpConnection.SetActivityError(ex);
                     exceptions.Add(ex);
                 }
                 catch (Exception e)
                 {
-                    tcpConnection?.AddException(e);
-                    tcpConnection?.SetStatus(ActivityStatusCode.Error);
+                    tcpConnection.SetActivityError(e);
                     exceptions.Add(e);
                 }
             }

projects/RabbitMQ.Client/Impl/AutorecoveringConnection.Recovery.cs

@@ -324,8 +324,7 @@ await maybeNewInnerConnection.OpenAsync(cancellationToken)
             }
             catch (Exception e)
             {
-                connectionActivity?.AddException(e);
-                connectionActivity?.SetStatus(ActivityStatusCode.Error);
+                connectionActivity.SetActivityError(e);
                 ESLog.Error("Connection recovery exception.", e);
                 // Trigger recovery error events
                 if (!_connectionRecoveryErrorAsyncWrapper.IsEmpty)

projects/RabbitMQ.Client/Impl/Channel.BasicPublish.cs

@@ -97,17 +97,33 @@ private async ValueTask BasicPublishCoreAsync<TMethod, TProperties>(
                 RateLimitLease? lease =
                     await MaybeAcquirePublisherConfirmationLockAsync(cancellationToken)
                         .ConfigureAwait(false);
+                /*
+                 * The send activity is declared out here, rather than inside the try
+                 * below, so the catch can record the failure on it. With the `using`
+                 * scoped to the inner try the span was already disposed by the time
+                 * the catch ran, so no publish failure was ever reported: the span
+                 * ended status=Unset with no exception event, which tracing backends
+                 * read as a successful publish. See issue #1967.
+                 */
+                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();
 
                     await MaybeEnforceFlowControlAsync(cancellationToken)
                         .ConfigureAwait(false);
 
-                    using Activity? sendActivity = RabbitMQActivitySource.PublisherHasListeners
-                        ? RabbitMQActivitySource.BasicPublish(routingKey, exchange, body.Length, basicProperties)
-                        : default;
-
                     ulong publishSequenceNumber = publisherConfirmationInfo?.PublishSequenceNumber ?? 0;
 
                     BasicProperties? props = PopulateBasicPropertiesHeaders(basicProperties, sendActivity, publishSequenceNumber);
@@ -125,6 +141,34 @@ await ModelSendAsync(in cmd, in props, body, bodyOwner, cancellationToken)
                 }
                 catch (Exception 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;
+                    }
+
+                    /*
+                     * "Handled" here means the exception was routed onto the publisher
+                     * confirmation task, not that it was swallowed: the finally below
+                     * awaits that task and re-raises the same instance to the caller. So
+                     * recording the error above is correct even when exceptionWasHandled
+                     * is true - the publish failed and the caller sees it, just through
+                     * the confirmation channel rather than a throw from here. This is not
+                     * the spec's "handled or retried and completed gracefully" exemption,
+                     * which is for operations that recover; a faulted publish never does.
+                     * Every path that records Error is one the caller observes as a
+                     * failure. See issue #1967.
+                     */
                     bool exceptionWasHandled =
                         MaybeHandleExceptionWithEnabledPublisherConfirmations(publisherConfirmationInfo, ex);
                     if (!exceptionWasHandled)
@@ -135,8 +179,30 @@ await ModelSendAsync(in cmd, in props, body, bodyOwner, cancellationToken)
                 finally
                 {
                     MaybeReleasePublisherConfirmationLock(lease);
-                    await MaybeEndPublisherConfirmationTrackingAsync(publisherConfirmationInfo, cancellationToken)
-                        .ConfigureAwait(false);
+
+                    /*
+                     * This await is the one that surfaces a nack or an unroutable
+                     * mandatory publish (PublishException), so it is a publish failure
+                     * like any other and belongs on the span. It cannot simply be
+                     * wrapped by the catch above, because it runs in the finally: the
+                     * confirmation is only awaited once the send has been issued.
+                     */
+                    try
+                    {
+                        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);
+                        throw;
+                    }
                 }
             }
             finally

projects/RabbitMQ.Client/Impl/Connection.cs

@@ -31,7 +31,6 @@
 
 using System;
 using System.Collections.Generic;
-using System.Diagnostics;
 using System.Diagnostics.CodeAnalysis;
 using System.IO;
 using System.Runtime.CompilerServices;
@@ -553,10 +552,7 @@ internal Task OnCallbackExceptionAsync(CallbackExceptionEventArgs args)
 
         internal ValueTask WriteAsync(OutgoingFrame frames, CancellationToken cancellationToken)
         {
-            if (RabbitMQActivitySource.PublisherHasListeners)
-            {
-                Activity.Current.SetNetworkTags(_frameHandler);
-            }
+            RabbitMQActivitySource.SetNetworkTagsOnAmbientPublisherActivity(_frameHandler);
 
             return _frameHandler.WriteAsync(frames, cancellationToken);
         }