From 86fae2067887a8287f845c29eafbed883b0f9265 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Muhammet=20=C5=9Eafak?= Date: Wed, 11 Nov 2026 09:28:38 +0300 Subject: [PATCH 1/2] feat(pulsar): add PublishEnvelopeAsync so a Telemetry-built envelope is sent as is Telemetry.PublishAsync records the built envelope's meta.id on the producer span, but the send callback had to call PublishWithHeadersAsync(urn, data), which rebuilt a different envelope, so messaging.message.id never matched the wire meta.id. PublishEnvelopeAsync sends the given envelope unchanged; PublishWithHeadersAsync now delegates to it. --- CHANGELOG.md | 4 + README.md | 19 ++++ src/BabelQueue.Pulsar/PulsarPublisher.cs | 26 +++++ .../PulsarPublisherTests.cs | 99 +++++++++++++++++++ 4 files changed, 148 insertions(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index ca636ed..e8840f9 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,10 @@ The envelope wire format is versioned separately by `meta.schema_version` ## [Unreleased] +### 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 diff --git a/README.md b/README.md index 2804ecd..372e5b8 100644 --- a/README.md +++ b/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(); +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 diff --git a/src/BabelQueue.Pulsar/PulsarPublisher.cs b/src/BabelQueue.Pulsar/PulsarPublisher.cs index 39adae9..f31e46c 100644 --- a/src/BabelQueue.Pulsar/PulsarPublisher.cs +++ b/src/BabelQueue.Pulsar/PulsarPublisher.cs @@ -63,6 +63,32 @@ public async Task PublishWithHeadersAsync( CancellationToken cancellationToken = default) { var envelope = EnvelopeCodec.Make(urn, data, TopicName(_producer.Topic), traceId); + return await PublishEnvelopeAsync(envelope, headers, delay, cancellationToken).ConfigureAwait(false); + } + + /// + /// Sends an already-built exactly as given (the same §5 property projection, + /// header overlay and delayed delivery 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 send; 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. + /// A positive delay schedules native delayed delivery. + /// Cancellation token. + public async Task PublishEnvelopeAsync( + Envelope envelope, + IReadOnlyDictionary? headers = null, + TimeSpan? delay = null, + CancellationToken cancellationToken = default) + { + ArgumentNullException.ThrowIfNull(envelope); var metadata = new MessageMetadata(); var properties = new Dictionary(PulsarProperties.Of(envelope), StringComparer.Ordinal); diff --git a/tests/BabelQueue.Pulsar.Tests/PulsarPublisherTests.cs b/tests/BabelQueue.Pulsar.Tests/PulsarPublisherTests.cs index b88a32e..26ca327 100644 --- a/tests/BabelQueue.Pulsar.Tests/PulsarPublisherTests.cs +++ b/tests/BabelQueue.Pulsar.Tests/PulsarPublisherTests.cs @@ -268,4 +268,103 @@ public void ExtractLeavesOutTheContractProperties() Assert.Equal(new[] { "traceparent", "tracestate" }, new SortedSet(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(); + 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 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 { ["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); + + 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); + } } From 8bf57a4cae8fa82f9a1afac2ed4bd4db2a43a815 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Muhammet=20=C5=9Eafak?= Date: Wed, 11 Nov 2026 14:16:19 +0300 Subject: [PATCH 2/2] Release v1.5.0 --- CHANGELOG.md | 2 ++ src/BabelQueue.Pulsar/BabelQueue.Pulsar.csproj | 2 +- 2 files changed, 3 insertions(+), 1 deletion(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index e8840f9..f142921 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,8 @@ 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. diff --git a/src/BabelQueue.Pulsar/BabelQueue.Pulsar.csproj b/src/BabelQueue.Pulsar/BabelQueue.Pulsar.csproj index bb2d9ff..8d63a28 100644 --- a/src/BabelQueue.Pulsar/BabelQueue.Pulsar.csproj +++ b/src/BabelQueue.Pulsar/BabelQueue.Pulsar.csproj @@ -16,7 +16,7 @@ BabelQueue.Pulsar - 1.4.0 + 1.5.0 Muhammet Şafak BabelQueue Apache Pulsar transport for BabelQueue — a canonical-envelope publisher and a URN-routed consumer over DotPulsar, on the framework-agnostic core.