Release v1.5.0

CHANGELOG.md

@@ -9,6 +9,12 @@ The envelope wire format is versioned separately by `meta.schema_version`
 
 ## [Unreleased]
 
+## [1.5.0] - 2026-11-11
+
+### Added
+
+- **`PulsarPublisher.PublishEnvelopeAsync(envelope, headers, ...)`.** Sends an already-built `Envelope` exactly as given, with `delay` and the §5 property projection and the optional out-of-band `headers`, and returns `meta.id`. Use it as the `send` callback of `Telemetry.PublishAsync(..., headers, env => publisher.PublishEnvelopeAsync(env, headers), queue)`: the producer span's `messaging.message.id` was the `meta.id` of the envelope `Telemetry` built, while `PublishWithHeadersAsync(urn, data, ...)` rebuilt a different one, so the span id never matched the wire. Now it does. `PublishWithHeadersAsync` and `PublishAsync` delegate to it; their behaviour is unchanged. Envelope unchanged (`schema_version` stays `1`); no new `BabelQueue.Core` requirement.
+
 ## [1.4.0] - 2026-11-03
 
 ### Added

README.md

@@ -125,6 +125,25 @@ await using var babel = new PulsarConsumer(consumer, handlers, new PulsarConsume
 first `await` escapes the consumer's try/catch and can crash the process. Keep hooks
 synchronous, or hand async work off yourself and observe its failures.
 
+## Tracing a publish (ADR-0028, ADR-0038)
+
+`PublishEnvelopeAsync(envelope, headers, ...)` sends an **already-built** envelope exactly as given (the same
+§5 property projection, header overlay and delayed delivery as `PublishWithHeadersAsync`, no rebuild). Use it as the `send` callback of
+`BabelQueue.Core`'s `Telemetry.PublishAsync`, so the producer span's `messaging.message.id` is the `meta.id` that
+reaches the wire and the span's `traceparent` rides beside the envelope:
+
+```csharp
+using BabelQueue.Tracing;
+
+var headers = new Dictionary<string, string>();
+await Telemetry.PublishAsync("urn:babel:orders:created", data, headers,
+    env => publisher.PublishEnvelopeAsync(env, headers), "orders", Telemetry.Systems.Pulsar);
+```
+
+The `queue` argument you give `Telemetry.PublishAsync` becomes `meta.queue`; `PublishEnvelopeAsync` does not change
+the envelope. The envelope is unchanged (`schema_version` stays `1`). Needs no `BabelQueue.Core` newer than the
+one this package already requires.
+
 ## Build & test
 
 ```bash

src/BabelQueue.Pulsar/BabelQueue.Pulsar.csproj

@@ -16,7 +16,7 @@
 
     <!-- NuGet package metadata -->
     <PackageId>BabelQueue.Pulsar</PackageId>
-    <Version>1.4.0</Version>
+    <Version>1.5.0</Version>
     <Authors>Muhammet Şafak</Authors>
     <Company>BabelQueue</Company>
     <Description>Apache Pulsar transport for BabelQueue — a canonical-envelope publisher and a URN-routed consumer over DotPulsar, on the framework-agnostic core.</Description>

src/BabelQueue.Pulsar/PulsarPublisher.cs

@@ -63,6 +63,32 @@ public async Task<string> PublishWithHeadersAsync(
         CancellationToken cancellationToken = default)
     {
         var envelope = EnvelopeCodec.Make(urn, data, TopicName(_producer.Topic), traceId);
+        return await PublishEnvelopeAsync(envelope, headers, delay, cancellationToken).ConfigureAwait(false);
+    }
+
+    /// <summary>
+    /// Sends an <b>already-built</b> <paramref name="envelope"/> exactly as given (the same §5 property projection,
+    /// header overlay and delayed delivery as <see cref="PublishWithHeadersAsync"/>) without rebuilding it:
+    /// <c>meta.id</c>, <c>trace_id</c> and every extension field reach the wire unchanged. Use it as the <c>send</c>
+    /// callback of <c>Telemetry.PublishAsync(..., headers, env =&gt; publisher.PublishEnvelopeAsync(env, headers), queue)</c>,
+    /// so the producer span's <c>messaging.message.id</c> is the <c>meta.id</c> on the wire. Returns <c>meta.id</c>.
+    /// </summary>
+    /// <param name="envelope">The envelope to send; its <c>meta.queue</c> is the caller's responsibility.</param>
+    /// <remarks>
+    /// <c>meta.queue</c> is whatever the envelope carries; it is not rewritten to this publisher's target. An envelope
+    /// built by <c>Telemetry.PublishAsync</c> carries that call's <c>queue</c> argument (default <c>"default"</c>), so pass
+    /// the queue/topic name there, or the wire <c>meta.queue</c> and the span's destination will not name it.
+    /// </remarks>
+    /// <param name="headers">The out-of-band transport headers to ride beside the envelope.</param>
+    /// <param name="delay">A positive delay schedules native delayed delivery.</param>
+    /// <param name="cancellationToken">Cancellation token.</param>
+    public async Task<string> PublishEnvelopeAsync(
+        Envelope envelope,
+        IReadOnlyDictionary<string, string>? headers = null,
+        TimeSpan? delay = null,
+        CancellationToken cancellationToken = default)
+    {
+        ArgumentNullException.ThrowIfNull(envelope);
 
         var metadata = new MessageMetadata();
         var properties = new Dictionary<string, string>(PulsarProperties.Of(envelope), StringComparer.Ordinal);

tests/BabelQueue.Pulsar.Tests/PulsarPublisherTests.cs

@@ -268,4 +268,103 @@ public void ExtractLeavesOutTheContractProperties()
 
         Assert.Equal(new[] { "traceparent", "tracestate" }, new SortedSet<string>(map.Keys, StringComparer.Ordinal));
     }
+
+    // O6: the envelope Telemetry.PublishAsync builds is the one that is sent, so the producer span's
+    // messaging.message.id is the meta.id on the wire (and the traceparent rides beside the envelope).
+    [Fact]
+    public async Task TelemetryPublishEnvelopeRecordsTheWireMessageId()
+    {
+        MessageMetadata? meta = null;
+        byte[]? payload = null;
+        var publisher = new PulsarPublisher(Producer("orders", (m, b) => { meta = m; payload = b; }).Object);
+        const string envelopeUrn = "urn:babel:orders:envelope-id";
+        var activities = new List<Activity>();
+        using var listener = new ActivityListener
+        {
+            ShouldListenTo = source => source.Name == Telemetry.ActivitySourceName,
+            Sample = (ref ActivityCreationOptions<ActivityContext> _) => ActivitySamplingResult.AllData,
+            ActivityStopped = activity =>
+            {
+                if (activity.Kind == ActivityKind.Producer && Equals(activity.GetTagItem("babelqueue.urn"), envelopeUrn))
+                {
+                    lock (activities)
+                    {
+                        activities.Add(activity);
+                    }
+                }
+            },
+        };
+        ActivitySource.AddActivityListener(listener);
+
+        var headers = new Dictionary<string, string>();
+        var id = await Telemetry.PublishAsync(
+            envelopeUrn,
+            new Dictionary<string, object?> { ["order_id"] = 7 },
+            headers,
+            env => publisher.PublishEnvelopeAsync(env, headers),
+            "orders");
+
+        var producerSpan = Assert.Single(activities);
+        var body = Encoding.UTF8.GetString(payload!);
+        var traceparent = string.IsNullOrEmpty(meta!["traceparent"]) ? null : meta["traceparent"];
+        var wire = EnvelopeCodec.Decode(body);
+
+        Assert.Equal(id, wire.Meta!.Id);
+        Assert.Equal(wire.Meta.Id, producerSpan.GetTagItem("messaging.message.id"));
+        Assert.Equal(Telemetry.UuidOf(producerSpan.TraceId), wire.TraceId);
+        Assert.Equal(wire.TraceId, producerSpan.GetTagItem("messaging.message.conversation_id"));
+        Assert.Equal(EnvelopeCodec.SchemaVersion, wire.Meta.SchemaVersion);
+        Assert.Equal(headers["traceparent"], traceparent);
+        Assert.Contains(producerSpan.TraceId.ToHexString(), traceparent, StringComparison.Ordinal);
+        Assert.DoesNotContain("traceparent", body, StringComparison.Ordinal);
+    }
+
+    [Fact]
+    public async Task PublishEnvelopeSendsTheGivenEnvelopeUnchanged()
+    {
+        MessageMetadata? meta = null;
+        byte[]? payload = null;
+        var publisher = new PulsarPublisher(Producer("orders", (m, b) => { meta = m; payload = b; }).Object);
+        var made = EnvelopeCodec.Make(
+            "urn:babel:orders:created", new Dictionary<string, object?> { ["order_id"] = 7 }, "orders", "trace-fixed");
+        var envelope = made with
+        {
+            Extras = new Dictionary<string, JsonElement> { ["x_tenant"] = JsonSerializer.SerializeToElement("t1") },
+            Meta = made.Meta! with
+            {
+                Extras = new Dictionary<string, JsonElement> { ["x_region"] = JsonSerializer.SerializeToElement("eu") },
+            },
+        };
+
+        var id = await publisher.PublishEnvelopeAsync(envelope);
+
+        var body = Encoding.UTF8.GetString(payload!);
+        var traceparent = string.IsNullOrEmpty(meta!["traceparent"]) ? null : meta["traceparent"];
+        Assert.Equal(envelope.Meta!.Id, id);
+        Assert.Equal(EnvelopeCodec.Encode(envelope), body);
+        var wire = EnvelopeCodec.Decode(body);
+        Assert.Equal("trace-fixed", wire.TraceId);
+        Assert.Equal(EnvelopeCodec.SchemaVersion, wire.Meta!.SchemaVersion);
+        Assert.Equal(envelope.Meta.Id, wire.Meta.Id);
+        Assert.Equal("t1", wire.Extras!["x_tenant"].GetString());
+        Assert.Equal("eu", wire.Meta.Extras!["x_region"].GetString());
+        Assert.Null(traceparent);
+    }
+
+    // Documents the PublishEnvelopeAsync contract: meta.queue is whatever the envelope carries. Telemetry stamps
+    // its queue argument (default "default"), not this publisher's target, so callers must pass the queue.
+    [Fact]
+    public async Task TelemetryPublishWithoutAQueueArgumentStampsDefaultMetaQueue()
+    {
+        MessageMetadata? meta = null;
+        byte[]? payload = null;
+        var publisher = new PulsarPublisher(Producer("orders", (m, b) => { meta = m; payload = b; }).Object);
+        await Telemetry.PublishAsync(
+            "urn:babel:orders:created", null, env => publisher.PublishEnvelopeAsync(env));
+
+        var body = Encoding.UTF8.GetString(payload!);
+        var traceparent = string.IsNullOrEmpty(meta!["traceparent"]) ? null : meta["traceparent"];
+        Assert.Equal("default", EnvelopeCodec.Decode(body).Meta!.Queue);
+        Assert.Null(traceparent);
+    }
 }