diff --git a/CHANGELOG.md b/CHANGELOG.md index f57ba3e..a9f9826 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/README.md b/README.md index cb66133..582e21f 100644 --- a/README.md +++ b/README.md @@ -153,8 +153,11 @@ using BabelQueue.Tracing; // produce: BabelQueue.Core fills the carrier with the active span's traceparent var headers = new Dictionary(); +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) => @@ -162,7 +165,7 @@ await Telemetry.PublishAsync("urn:babel:orders:created", data, headers, ``` 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 diff --git a/src/BabelQueue.Redis/BabelQueue.Redis.csproj b/src/BabelQueue.Redis/BabelQueue.Redis.csproj index 5c527b9..373edc4 100644 --- a/src/BabelQueue.Redis/BabelQueue.Redis.csproj +++ b/src/BabelQueue.Redis/BabelQueue.Redis.csproj @@ -16,7 +16,7 @@ BabelQueue.Redis - 1.3.0 + 1.4.0 Muhammet Şafak BabelQueue 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. diff --git a/src/BabelQueue.Redis/RedisPublisher.cs b/src/BabelQueue.Redis/RedisPublisher.cs index 0c4cf45..a4e96a3 100644 --- a/src/BabelQueue.Redis/RedisPublisher.cs +++ b/src/BabelQueue.Redis/RedisPublisher.cs @@ -109,6 +109,31 @@ public async Task PublishWithHeadersAsync( cancellationToken.ThrowIfCancellationRequested(); var envelope = EnvelopeCodec.Make(urn, data, _queue, traceId); + return await PublishEnvelopeAsync(envelope, headers, cancellationToken).ConfigureAwait(false); + } + + /// + /// Pushes an already-built exactly as given (the same bare / framed value and + /// RPUSH as ) without rebuilding it: meta.id, trace_id + /// and every extension field reach the wire unchanged. Use it as the send callback of + /// Telemetry.PublishAsync(..., headers, env => publisher.PublishEnvelopeAsync(env, headers), queue), so the + /// producer span's messaging.message.id is the meta.id on the wire. Returns meta.id. + /// + /// The envelope to push; its meta.queue is the caller's responsibility. + /// + /// meta.queue is whatever the envelope carries; it is not rewritten to this publisher's target. An envelope + /// built by Telemetry.PublishAsync carries that call's queue argument (default "default"), so pass + /// the queue/topic name there, or the wire meta.queue and the span's destination will not name it. + /// + /// The out-of-band transport headers to ride beside the envelope (legacy layout only). + /// Cancellation token. + public async Task PublishEnvelopeAsync( + Envelope envelope, + IReadOnlyDictionary? 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). diff --git a/tests/BabelQueue.Redis.Tests/RedisHeadersTests.cs b/tests/BabelQueue.Redis.Tests/RedisHeadersTests.cs index b01a17f..d6ef3e3 100644 --- a/tests/BabelQueue.Redis.Tests/RedisHeadersTests.cs +++ b/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(); + using var listener = new ActivityListener + { + ShouldListenTo = source => source.Name == Telemetry.ActivitySourceName, + Sample = (ref ActivityCreationOptions _) => 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(); + var id = await Telemetry.PublishAsync( + envelopeUrn, + new Dictionary { ["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 { ["order_id"] = 7 }, "orders", "trace-fixed"); + var envelope = made with + { + Extras = new Dictionary { ["x_tenant"] = JsonSerializer.SerializeToElement("t1") }, + Meta = made.Meta! with + { + Extras = new Dictionary { ["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); + } }