Release v1.4.0

CHANGELOG.md

@@ -9,6 +9,12 @@ The envelope wire format is versioned separately by `meta.schema_version`
 
 ## [Unreleased]
 
+## [1.4.0] - 2026-11-11
+
+### Added
+
+- **`RedisPublisher.PublishEnvelopeAsync(envelope, headers, ...)`.** Sends an already-built `Envelope` exactly as given, with the bare / framed value 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.3.0] - 2026-11-03
 
 ### Added

README.md

@@ -153,16 +153,19 @@ using BabelQueue.Tracing;
 
 // produce: BabelQueue.Core fills the carrier with the active span's traceparent
 var headers = new Dictionary<string, string>();
+var publisher = new RedisPublisher(db, "orders");
 await Telemetry.PublishAsync("urn:babel:orders:created", data, headers,
-    env => new RedisPublisher(db, "orders").PublishWithHeadersAsync("urn:babel:orders:created", data, headers));
+    env => publisher.PublishEnvelopeAsync(env, headers), "orders", Telemetry.Systems.Redis);
+// PublishEnvelopeAsync sends the envelope Telemetry built as is, so the span's messaging.message.id
+// is the meta.id on the wire; the Telemetry queue argument becomes meta.queue.
 
 // consume: the handler receives the un-framed body + headers; wrap to link the span
 ["urn:babel:orders:created"] = (env, rawBody, headers, ct) =>
     Telemetry.Wrap(async e => { /* ... */ }, headers)(env)
 ```
 
 With no `traceparent` the consumer falls back to the v0.1 `trace_id` mapping. Requires
-`BabelQueue.Core 1.4.0`.
+`BabelQueue.Core 1.14.0`.
 
 ## Build & test
 

src/BabelQueue.Redis/BabelQueue.Redis.csproj

@@ -16,7 +16,7 @@
 
     <!-- NuGet package metadata -->
     <PackageId>BabelQueue.Redis</PackageId>
-    <Version>1.3.0</Version>
+    <Version>1.4.0</Version>
     <Authors>Muhammet Şafak</Authors>
     <Company>BabelQueue</Company>
     <Description>Redis transport for BabelQueue — a canonical-envelope publisher and a URN-routed consumer over StackExchange.Redis (the reliable-queue list pattern), on the framework-agnostic core.</Description>

src/BabelQueue.Redis/RedisPublisher.cs

@@ -109,6 +109,31 @@ public async Task<string> PublishWithHeadersAsync(
         cancellationToken.ThrowIfCancellationRequested();
 
         var envelope = EnvelopeCodec.Make(urn, data, _queue, traceId);
+        return await PublishEnvelopeAsync(envelope, headers, cancellationToken).ConfigureAwait(false);
+    }
+
+    /// <summary>
+    /// Pushes an <b>already-built</b> <paramref name="envelope"/> exactly as given (the same bare / framed value and
+    /// <c>RPUSH</c> 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 push; 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 (legacy layout only).</param>
+    /// <param name="cancellationToken">Cancellation token.</param>
+    public async Task<string> PublishEnvelopeAsync(
+        Envelope envelope,
+        IReadOnlyDictionary<string, string>? headers = null,
+        CancellationToken cancellationToken = default)
+    {
+        ArgumentNullException.ThrowIfNull(envelope);
+        cancellationToken.ThrowIfCancellationRequested();
 
         // Bare: the list element IS the envelope JSON byte-for-byte (§1.2). Framed (headers present):
         // the headers ride in a transport-owned frame beside the frozen envelope (ADR-0028).

tests/BabelQueue.Redis.Tests/RedisHeadersTests.cs

@@ -1,6 +1,8 @@
+using System.Diagnostics;
 using System.Text.Json;
 using BabelQueue;
 using BabelQueue.Redis;
+using BabelQueue.Tracing;
 using Moq;
 using StackExchange.Redis;
 using Xunit;
@@ -199,4 +201,109 @@ public async Task ExtractedHeadersRebuildTheRemoteParentForTelemetryWrap()
         Assert.Equal("0af7651916cd43dd8448eb211c80319c", parent!.Value.TraceId.ToHexString());
         Assert.Equal("b7ad6b7169203331", parent.Value.SpanId.ToHexString());
     }
+
+    // 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()
+    {
+        var (mock, pushed) = MockDatabase();
+        var publisher = new RedisPublisher(mock.Object, Queue);
+        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 stored = (string)pushed()!;
+        string body = stored;
+        string? traceparent = null;
+        using (var frame = JsonDocument.Parse(stored))
+        {
+            if (frame.RootElement.TryGetProperty("__bq_frame", out _))
+            {
+                body = frame.RootElement.GetProperty("body").GetString()!;
+                traceparent = frame.RootElement.GetProperty("headers").GetProperty("traceparent").GetString();
+            }
+        }
+        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()
+    {
+        var (mock, pushed) = MockDatabase();
+        var publisher = new RedisPublisher(mock.Object, Queue);
+        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);
+
+        // no headers: the stored value is the bare envelope, byte-for-byte (no frame)
+        var body = (string)pushed()!;
+        Assert.DoesNotContain("__bq_frame", body, StringComparison.Ordinal);
+        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());
+    }
+
+    // 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()
+    {
+        var (mock, pushed) = MockDatabase();
+        var publisher = new RedisPublisher(mock.Object, Queue);
+        await Telemetry.PublishAsync(
+            "urn:babel:orders:created", null, env => publisher.PublishEnvelopeAsync(env));
+
+        // no headers: the stored value is the bare envelope, byte-for-byte (no frame)
+        var body = (string)pushed()!;
+        Assert.DoesNotContain("__bq_frame", body, StringComparison.Ordinal);
+        Assert.Equal("default", EnvelopeCodec.Decode(body).Meta!.Queue);
+    }
 }