fix(subscriptions): link to the restored trace context, don't parent

src/Core/src/Eventuous.Diagnostics/MetadataExtensions.cs

@@ -21,8 +21,8 @@ public Metadata AddActivityTags(Activity? activity) {
 
         [MethodImpl(MethodImplOptions.AggressiveInlining)]
         Metadata AddTracingMeta(TracingMeta tracingMeta)
-            => metadata.ContainsKey(TraceId) || tracingMeta.TraceId == EmptyId
-                ? metadata // don't override existing tracing data
+            => metadata.ContainsKey(TraceId) || !tracingMeta.IsValid()
+                ? metadata // don't override existing tracing data, and don't persist an unusable one
                 : metadata
                     .AddNotNull(TraceId, tracingMeta.TraceId)
                     .AddNotNull(SpanId, tracingMeta.SpanId);
@@ -31,6 +31,4 @@ Metadata AddTracingMeta(TracingMeta tracingMeta)
         public TracingMeta GetTracingMeta()
             => new(metadata.GetString(TraceId), metadata.GetString(SpanId));
     }
-
-    const string EmptyId = "0000000000000000";
 }

src/Core/src/Eventuous.Diagnostics/Tags/TracingMeta.cs

@@ -4,20 +4,34 @@
 namespace Eventuous.Diagnostics;
 
 public record TracingMeta(string? TraceId, string? SpanId) {
-    bool IsValid() => TraceId != null && SpanId != null;
+    /// <summary>
+    /// Whether the metadata carries a tracing context that can be used: both ids present and neither all-zero.
+    /// An all-zero id is what an activity reports when it was never given a real one, and persisting or
+    /// restoring that is worse than having nothing, so both ends of the round trip check it here.
+    /// </summary>
+    internal bool IsValid() => TraceId is not (null or EmptyTraceId) && SpanId is not (null or EmptySpanId);
 
+    /// <summary>
+    /// Restores an activity context from the persisted tracing metadata, or <c>null</c> when the metadata carries
+    /// no usable context. Callers check for null to decide whether there is anything to correlate with, so an
+    /// absent or all-zero context must not come back as a zeroed <seealso cref="ActivityContext"/>.
+    /// </summary>
     public ActivityContext? ToActivityContext(bool isRemote) {
         try {
-            return IsValid() ?
-                new ActivityContext(
+            return IsValid()
+                ? new ActivityContext(
                     ActivityTraceId.CreateFromString(TraceId),
                     ActivitySpanId.CreateFromString(SpanId),
                     ActivityTraceFlags.Recorded,
                     isRemote: isRemote
-                ) : default;
+                )
+                : null;
         }
         catch (Exception) {
-            return default;
+            return null;
         }
     }
-}
+
+    const string EmptyTraceId = "00000000000000000000000000000000";
+    const string EmptySpanId  = "0000000000000000";
+}

src/Core/src/Eventuous.Subscriptions/Diagnostics/SubscriptionActivity.cs

@@ -15,8 +15,12 @@ static class SubscriptionActivity {
             IMessageConsumeContext                      context,
             IEnumerable<KeyValuePair<string, object?>>? tags = null
         ) {
-        context.ParentContext = GetParentContext(context);
-        var activity = Create(name, activityKind, context.ParentContext, tags);
+        var (parentContext, link) = GetParentOrLink(context);
+        context.ParentContext     = parentContext;
+
+        var activity = link == null
+            ? Create(name, activityKind, parentContext, tags)
+            : CreateLinkedRoot(name, activityKind, link.Value, tags);
 
         return activity?.SetContextTags(context);
     }
@@ -42,25 +46,82 @@ static class SubscriptionActivity {
             .SetOrCopyParentTag(TelemetryTags.Messaging.CorrelationId, context.Metadata?.GetCorrelationId());
     }
 
-    static ActivityContext? GetParentContext(IBaseConsumeContext context) {
+    /// <summary>
+    /// Decides how the consume span relates to what came before it: a parent, when the causality is real
+    /// in-process causality inside a single consume, or a link, when the only thing available is the tracing
+    /// context persisted in the message metadata.
+    /// </summary>
+    static (ActivityContext? ParentContext, ActivityLink? Link) GetParentOrLink(IBaseConsumeContext context) {
         // The current activity is only trusted as a parent when Eventuous created it: that's the
         // message's own pipeline activity (handler, then filters nested under it). A foreign ambient
         // activity — a test framework's per-test span, a client library's delivery span — must not
-        // override the remote context propagated in the message metadata, or the consumer span gets
-        // detached from the producer trace.
+        // override the remote context propagated in the message metadata.
         if (Activity.Current?.Source.Name.StartsWith(EventuousDiagnostics.InstrumentationName, StringComparison.Ordinal) == true) {
-            return Activity.Current.Context;
+            return (Activity.Current.Context, null);
         }
 
         if (context.Items.TryGetItem<Activity>(ContextItemKeys.Activity, out var parentActivity)) {
-            return parentActivity?.Context;
+            return (parentActivity?.Context, null);
+        }
+
+        // The trace context restored from message metadata is durable data on the message, so it outlives the
+        // process that produced it. Parenting to it would put every redelivery of that message — a replay, a
+        // checkpoint reset, a hot resubscribe loop — into the same trace, forever: there is no live root to end
+        // it and no sampling decision left to take, so the trace grows without any bound a collector can impose.
+        // A link records the same causality without joining the trace, which is also what the OpenTelemetry
+        // messaging conventions prescribe for an asynchronous "process" operation.
+        if (context.Metadata?.GetTracingMeta().ToActivityContext(true) is { } remoteContext) {
+            return (null, new ActivityLink(remoteContext));
         }
 
-        var tracingData = context.Metadata?.GetTracingMeta();
+        return (Activity.Current?.Context, null);
+    }
 
-        if (tracingData?.ToActivityContext(true) is { } remoteContext) return remoteContext;
+    static Activity? CreateLinkedRoot(
+            string                                      name,
+            ActivityKind                                activityKind,
+            ActivityLink                                link,
+            IEnumerable<KeyValuePair<string, object?>>? tags
+        ) {
+        // .NET offers no "explicitly parentless" argument: a default parent context makes Activity.Current the
+        // parent. Suppressing it keeps an unrelated ambient span out of both the parentage and the sampling
+        // decision, which has to be taken as a root. Assigning Activity.Current copies the execution context,
+        // so it is only touched when there is actually something to suppress: this runs per consumed message.
+        var ambient = Activity.Current;
 
-        return Activity.Current?.Context;
+        if (ambient != null) Activity.Current = null;
+
+        try {
+            var activity = EventuousDiagnostics.ActivitySource.CreateActivity(
+                name,
+                activityKind,
+                parentContext: default,
+                tags,
+                links: [link],
+                idFormat: ActivityIdFormat.W3C
+            );
+
+            if (activity == null) return null;
+
+            // .NET only materialises the trace id at creation if the sampler actually read it — a ratio-based
+            // sampler does, an always-on one does not — and an unstarted activity reports an all-zero id
+            // otherwise. Reuse the sampler's id when there is one, so the decision belongs to the span it was
+            // taken for, and generate one when there isn't: pinning zeroes below would make the id permanent.
+            var traceId = activity.TraceId;
+
+            // Activity binds its parent from Activity.Current when it is started, and the async handling path
+            // starts this activity further down the pipe, where something else may be ambient. Pinning the trace
+            // id with a zero parent span id makes the span a root that a late Start cannot re-parent. The trace
+            // flags are carried over, otherwise the span would silently stop being recorded.
+            return activity.SetParentId(
+                traceId == default ? ActivityTraceId.CreateRandom() : traceId,
+                default,
+                activity.ActivityTraceFlags
+            );
+        }
+        finally {
+            if (ambient != null) Activity.Current = ambient;
+        }
     }
 
     public static Activity? Create(