Release v1.4.0 (ADR-0038, ADR-0039)

CHANGELOG.md

@@ -9,6 +9,19 @@ The envelope wire format is versioned separately by `meta.schema_version`
 
 ## [Unreleased]
 
+## [1.4.0] - 2026-11-03
+
+### Added
+
+- **Dead-letter observer (ADR-0039 section 5.1).** `PulsarConsumerOptions.DeadLetterObserver` takes the core's optional `BabelQueue.IDeadLetterObserver` (needs a `BabelQueue.Core` that ships it). `PulsarConsumer` calls it once per dead-letter the SDK itself carried out: a message written through `DeadLetterProducer`, right after the write is confirmed: an exhausted `failed` / `unknown_urn` message, a poison body, and a raw message rejected by the schema-version gate (empty `Reason`). `Queue` is `meta.queue` when the envelope carries one, else the consumed topic's queue name; `DeadLetterQueue` is the dead-letter producer's topic. It is not called when the write fails (the message is redelivered), for a retry, or when dead-lettering is off. The call goes through `DeadLetterObservers.NotifySafely`, so an observer exception is caught and logged and never changes the settle outcome. Without an observer nothing changes. Envelope unchanged (`schema_version` stays 1).
+- **W3C `traceparent` propagation (ADR-0028).** `PulsarPublisher.PublishWithHeadersAsync(urn, data, headers, traceId, delay)` overlays an out-of-band header carrier (e.g. the one `Telemetry.PublishAsync(..., headers, ...)` fills from the producer span) as message properties, **beside** the frozen envelope and the `bq-` / `bq_` contract projection, never inside the body. The transport-owned `bq-*` properties (including `bq-delay`) are always dropped from the carrier (parity with the Go reference), keys already present are never overwritten and empty keys/values are skipped; a `null`/empty carrier is identical to `PublishAsync`.
+- `PulsarTraceHeaders.Extract(...)`: surfaces a received message's message properties as a flat map to hand to `Telemetry.Wrap(handler, headers)`, so the consumer span becomes a true child of the producer span. The transport-owned contract keys are left out of the map, so it can be handed back to `PublishWithHeadersAsync` safely. With no `traceparent` the core falls back to the v0.1 `trace_id` mapping. The `BabelHandler` signature is unchanged; call `PulsarTraceHeaders.Extract(message)` inside the handler.
+- Envelope unchanged (`schema_version` stays 1); additive, no wire change.
+
+### Changed
+
+- `BabelQueue.Core` dependency raised from `1.13.0` to `1.14.0` (`IDeadLetterObserver`).
+
 ## [1.3.0] - 2026-10-26
 
 Implements ADR-0037 decision 1 (version-gate parity, broker-bindings §5.7). The envelope stays at `schema_version: 1`;

src/BabelQueue.Pulsar/BabelHandler.cs

@@ -9,4 +9,9 @@ namespace BabelQueue.Pulsar;
 /// <see cref="PulsarConsumerOptions.RetryProducer"/> is set) until <see cref="PulsarConsumerOptions.MaxAttempts"/>
 /// deliveries, after which it is dead-lettered or dropped (ADR-0036).
 /// </summary>
+/// <remarks>
+/// To continue a W3C <c>traceparent</c> that rode beside the envelope (ADR-0028), hand
+/// <c>PulsarTraceHeaders.Extract(message)</c> to <c>Telemetry.Wrap(handler, headers)</c>; the consumer span then becomes a true child of the
+/// producer span. With no <c>traceparent</c> in the message properties the core falls back to the v0.1 <c>trace_id</c> mapping.
+/// </remarks>
 public delegate Task BabelHandler(Envelope envelope, IMessage<byte[]> message, CancellationToken cancellationToken);

src/BabelQueue.Pulsar/BabelQueue.Pulsar.csproj

@@ -16,7 +16,7 @@
 
     <!-- NuGet package metadata -->
     <PackageId>BabelQueue.Pulsar</PackageId>
-    <Version>1.3.0</Version>
+    <Version>1.4.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>
@@ -33,7 +33,7 @@
 
   <ItemGroup>
     <PackageReference Include="DotPulsar" Version="5.3.1" />
-    <PackageReference Include="BabelQueue.Core" Version="1.13.0" />
+    <PackageReference Include="BabelQueue.Core" Version="1.14.0" />
   </ItemGroup>
 
   <ItemGroup>

src/BabelQueue.Pulsar/PulsarConsumer.cs

@@ -293,6 +293,8 @@ await DeadLetterOrRedeliverAsync(
                     return (properties, Encoding.UTF8.GetBytes(EnvelopeCodec.Encode(annotated)));
                 },
                 envelope,
+                reason,
+                envelope,
                 cancellationToken).ConfigureAwait(false);
         }
         else
@@ -329,6 +331,8 @@ await DeadLetterOrRedeliverAsync(
                 message,
                 () => (PulsarProperties.Copy(message.Properties), message.Value() ?? Array.Empty<byte>()),
                 envelope,
+                string.Empty, // a raw body: no dead_letter block, so no reason
+                null,         // never decoded: no meta.queue
                 cancellationToken).ConfigureAwait(false);
         }
 
@@ -363,6 +367,8 @@ await DeadLetterOrRedeliverAsync(
                 return (PulsarProperties.Copy(message.Properties), record);
             },
             envelope,
+            "poison",
+            null, // the body is not a trustworthy envelope: the source queue is the consumed topic
             cancellationToken).ConfigureAwait(false);
         Report(cause, envelope, message);
     }
@@ -428,11 +434,17 @@ private async Task RetryAsync(
     /// <summary>
     /// Writes the dead-letter message and acknowledges the original only after that succeeded; a failed write leaves the
     /// original unacknowledged (negative ack, so it is redelivered and the write is tried again) and is reported.
+    /// The dead-letter observer (ADR-0039 section 5.1) is told once, right after the confirmed write, with
+    /// <paramref name="reason"/> (empty for a raw version-gated body); a failed write never reaches it.
+    /// <paramref name="queueSource"/> is the envelope whose <c>meta.queue</c> names the source queue, or <c>null</c>
+    /// when the body is not a decoded envelope (the consumed topic is used).
     /// </summary>
     private async Task DeadLetterOrRedeliverAsync(
         IMessage<byte[]> message,
         Func<(IReadOnlyDictionary<string, string> Properties, byte[] Payload)> build,
         Envelope envelope,
+        string reason,
+        Envelope? queueSource,
         CancellationToken cancellationToken)
     {
         Exception? writeError = null;
@@ -456,6 +468,11 @@ private async Task DeadLetterOrRedeliverAsync(
 
         if (writeError is null)
         {
+            var metaQueue = queueSource?.Meta?.Queue;
+            _options.DeadLetterObserver.NotifySafely(new DeadLetterEvent(
+                string.IsNullOrWhiteSpace(metaQueue) ? QueueName(message) : metaQueue,
+                _deadLetterProducer!.Topic,
+                reason));
             await AcknowledgeAsync(message, cancellationToken).ConfigureAwait(false);
             return;
         }

src/BabelQueue.Pulsar/PulsarConsumerOptions.cs

@@ -57,6 +57,16 @@ public sealed class PulsarConsumerOptions
     /// </summary>
     public IProducer<byte[]>? DeadLetterProducer { get; set; }
 
+    /// <summary>
+    /// Optional dead-letter observer (ADR-0039 section 5.1), called once per message the consumer wrote to the
+    /// <see cref="DeadLetterProducer"/>, right after the write is confirmed: an exhausted <c>failed</c> or
+    /// <c>unknown_urn</c> message, a poison body, or a raw message rejected by the schema-version gate (empty reason).
+    /// Not called when the write fails (the message is redelivered), for a retry, or when dead-lettering is off. An
+    /// observer exception is caught and logged; it never changes the settle outcome. Needs a <c>BabelQueue.Core</c>
+    /// that ships <see cref="IDeadLetterObserver"/>.
+    /// </summary>
+    public IDeadLetterObserver? DeadLetterObserver { get; set; }
+
     /// <summary>
     /// A producer bound to the consumer's own topic, used to retry a failed message the Go way (broker-bindings §5.5): the
     /// message is re-published with <c>bq-attempts</c> + 1 (and the body's <c>attempts</c> in sync) and the original is

src/BabelQueue.Pulsar/PulsarProperties.cs

@@ -11,6 +11,18 @@ namespace BabelQueue.Pulsar;
 /// </summary>
 internal static class PulsarProperties
 {
+    /// <summary>
+    /// The properties the transport owns: the §5.2 projection plus <c>bq-delay</c>. They are never accepted from, nor
+    /// surfaced as, out-of-band transport headers (parity with the Go reference).
+    /// </summary>
+    private static readonly HashSet<string> Contract = new(StringComparer.Ordinal)
+    {
+        "bq-job", "bq-trace-id", "bq-message-id", "bq-schema-version", "bq-source-lang", "bq-attempts", "bq-delay",
+    };
+
+    /// <summary>True when <paramref name="key"/> is a contract property the transport owns.</summary>
+    public static bool IsContract(string key) => Contract.Contains(key);
+
     public static IReadOnlyDictionary<string, string> Of(Envelope envelope)
     {
         var props = new Dictionary<string, string>(StringComparer.Ordinal);
@@ -35,6 +47,33 @@ public static IReadOnlyDictionary<string, string> Copy(IReadOnlyDictionary<strin
             ? new Dictionary<string, string>(StringComparer.Ordinal)
             : new Dictionary<string, string>(properties, StringComparer.Ordinal);
 
+    /// <summary>
+    /// Overlays the out-of-band string <paramref name="headers"/> (e.g. a W3C <c>traceparent</c>, ADR-0028) onto an
+    /// outgoing message's properties, beside the <c>bq-</c> contract projection. An existing property is never
+    /// overwritten, a contract property name (<see cref="IsContract"/>, including <c>bq-delay</c>) is always dropped,
+    /// and empty keys/values are skipped. Keys are applied in
+    /// ordinal order so the result is deterministic. A <c>null</c>/empty map is a no-op.
+    /// </summary>
+    public static void Merge(IDictionary<string, string> properties, IReadOnlyDictionary<string, string>? headers)
+    {
+        ArgumentNullException.ThrowIfNull(properties);
+        if (headers is null || headers.Count == 0)
+        {
+            return;
+        }
+
+        foreach (var key in headers.Keys.OrderBy(static k => k, StringComparer.Ordinal))
+        {
+            var value = headers[key];
+            if (string.IsNullOrEmpty(key) || string.IsNullOrEmpty(value) || IsContract(key) || properties.ContainsKey(key))
+            {
+                continue;
+            }
+
+            properties[key] = value;
+        }
+    }
+
     private static void Put(Dictionary<string, string> props, string key, string? value)
     {
         if (!string.IsNullOrEmpty(value))
@@ -43,3 +82,45 @@ private static void Put(Dictionary<string, string> props, string key, string? va
         }
     }
 }
+
+/// <summary>
+/// Out-of-band transport-header carrier for Pulsar (ADR-0028). A W3C <c>traceparent</c> (and
+/// <c>tracestate</c>) rides as a message property <b>beside</b> the frozen envelope — never inside it — next to the
+/// §5 <c>bq-*</c> projection. This is the consume-side half: it surfaces a received message's properties as a flat
+/// map a handler hands to <c>Telemetry.Wrap(handler, headers)</c>, so the consumer span becomes a true child of the
+/// producer span. With no <c>traceparent</c> the core falls back to the v0.1 <c>trace_id</c> mapping.
+/// </summary>
+public static class PulsarTraceHeaders
+{
+    /// <summary>
+    /// The properties of a received message as a flat map (empty when it has none). Only
+    /// out-of-band headers (e.g. <c>traceparent</c>) surface: the <c>bq-</c> contract properties the transport owns
+    /// are left out, so handing the map back to <c>PublishWithHeadersAsync</c> cannot re-inject them. Empty values
+    /// are skipped.
+    /// </summary>
+    public static Dictionary<string, string> Extract(DotPulsar.Abstractions.IMessage<byte[]> message)
+    {
+        ArgumentNullException.ThrowIfNull(message);
+        return Extract(message.Properties);
+    }
+
+    /// <summary>The same flat map, from a raw property dictionary (or <c>null</c>).</summary>
+    public static Dictionary<string, string> Extract(IReadOnlyDictionary<string, string>? properties)
+    {
+        var headers = new Dictionary<string, string>(StringComparer.Ordinal);
+        if (properties is null)
+        {
+            return headers;
+        }
+
+        foreach (var (key, value) in properties)
+        {
+            if (!string.IsNullOrEmpty(value) && !PulsarProperties.IsContract(key))
+            {
+                headers[key] = value;
+            }
+        }
+
+        return headers;
+    }
+}

src/BabelQueue.Pulsar/PulsarPublisher.cs

@@ -32,25 +32,52 @@ public PulsarPublisher(IProducer<byte[]> producer)
     /// projection, and returns the message id (<c>meta.id</c>). A positive <paramref name="delay"/>
     /// schedules native delayed delivery via <c>DeliverAtTime</c> and mirrors <c>bq-delay</c>.
     /// </summary>
-    public async Task<string> PublishAsync(
+    public Task<string> PublishAsync(
         string urn,
         IReadOnlyDictionary<string, object?>? data = null,
         string? traceId = null,
         TimeSpan? delay = null,
         CancellationToken cancellationToken = default)
+        => PublishWithHeadersAsync(urn, data, headers: null, traceId, delay, cancellationToken);
+
+    /// <summary>
+    /// The header-aware (ADR-0028) counterpart of
+    /// <see cref="PublishAsync(string, IReadOnlyDictionary{string, object?}?, string?, TimeSpan?, CancellationToken)"/>:
+    /// in addition to the §5 property projection it overlays the out-of-band <paramref name="headers"/> (e.g. a W3C
+    /// <c>traceparent</c> from <c>Telemetry.PublishAsync(..., headers, ...)</c>) as message properties. The headers
+    /// ride <b>beside</b> the frozen envelope, never inside it, and never overwrite a <c>bq-</c> contract property.
+    /// A <c>null</c>/empty header map is identical to <c>PublishAsync</c>.
+    /// </summary>
+    /// <param name="urn">The message URN to publish.</param>
+    /// <param name="data">The message payload, or <c>null</c> for an empty body.</param>
+    /// <param name="headers">The out-of-band transport headers to ride beside the envelope.</param>
+    /// <param name="traceId">An existing trace id to continue, or <c>null</c> to mint one.</param>
+    /// <param name="delay">A positive delay schedules native delayed delivery.</param>
+    /// <param name="cancellationToken">Cancellation token.</param>
+    public async Task<string> PublishWithHeadersAsync(
+        string urn,
+        IReadOnlyDictionary<string, object?>? data,
+        IReadOnlyDictionary<string, string>? headers,
+        string? traceId = null,
+        TimeSpan? delay = null,
+        CancellationToken cancellationToken = default)
     {
         var envelope = EnvelopeCodec.Make(urn, data, TopicName(_producer.Topic), traceId);
 
         var metadata = new MessageMetadata();
-        foreach (var property in PulsarProperties.Of(envelope))
-        {
-            metadata[property.Key] = property.Value;
-        }
+        var properties = new Dictionary<string, string>(PulsarProperties.Of(envelope), StringComparer.Ordinal);
 
         if (delay is { } window && window > TimeSpan.Zero)
         {
             metadata.DeliverAtTimeAsDateTimeOffset = DateTimeOffset.UtcNow + window;
-            metadata["bq-delay"] = ((long)window.TotalMilliseconds).ToString(CultureInfo.InvariantCulture);
+            properties["bq-delay"] = ((long)window.TotalMilliseconds).ToString(CultureInfo.InvariantCulture);
+        }
+
+        // ADR-0028: out-of-band headers (traceparent) ride beside the contract properties, never over them.
+        PulsarProperties.Merge(properties, headers);
+        foreach (var property in properties)
+        {
+            metadata[property.Key] = property.Value;
         }
 
         var payload = Encoding.UTF8.GetBytes(EnvelopeCodec.Encode(envelope));

tests/BabelQueue.Pulsar.Tests/PulsarConsumerTests.cs

@@ -667,4 +667,41 @@ public async Task RunBreaksWhenReceiveIsCancelled()
 
         consumer.Verify(c => c.Receive(It.IsAny<CancellationToken>()), Times.AtLeastOnce);
     }
+
+    [Fact]
+    public async Task HandlerReceivesThePublishedTraceparentThroughTheConsumer()
+    {
+        const string traceparent = "00-0af7651916cd43dd8448eb211c80319c-b7ad6b7169203331-01";
+        MessageMetadata? meta = null;
+        byte[]? body = null;
+        var producer = new Mock<IProducer<byte[]>>();
+        producer.SetupGet(p => p.Topic).Returns("orders");
+        producer
+            .Setup(p => p.Send(It.IsAny<MessageMetadata>(), It.IsAny<byte[]>(), It.IsAny<CancellationToken>()))
+            .Callback<MessageMetadata, byte[], CancellationToken>((m, b, _) => { meta = m; body = b; })
+            .Returns(new ValueTask<MessageId>(new MessageId(0, 0, 0, 0, "orders")));
+        await new PulsarPublisher(producer.Object).PublishWithHeadersAsync(
+            Urn, null, new Dictionary<string, string> { ["traceparent"] = traceparent }, "trace-1");
+
+        var message = Message(0, body!, new Dictionary<string, string>
+        {
+            ["bq-job"] = meta!["bq-job"]!,
+            ["traceparent"] = meta["traceparent"]!,
+        });
+        IReadOnlyDictionary<string, string>? seen = null;
+        var handlers = new Dictionary<string, BabelHandler>
+        {
+            [Urn] = (_, received, _) =>
+            {
+                seen = PulsarTraceHeaders.Extract(received);
+                return Task.CompletedTask;
+            },
+        };
+
+        await new PulsarConsumer(ConsumerWith(message.Object).Object, handlers).ReceiveAsync();
+
+        Assert.NotNull(seen);
+        Assert.Equal(traceparent, seen!["traceparent"]);
+        Assert.DoesNotContain("bq-job", seen.Keys);
+    }
 }