Merge pull request #306 from Vulthil/fix/messaging-wire-fidelity

fix(messaging): Fault<T> payload shape, consume-side header fidelity, and TTL expiration

Author: Vulthil

docs/articles/messaging.md

@@ -131,6 +131,17 @@ public sealed class OrderCreatedConsumer : IConsumer<OrderCreatedEvent>
 `IMessageContext.CancellationToken` exposes the delivery's cancellation token for
 handlers that want to observe it alongside the explicit method parameter.
 
+### Header values on the consume side
+
+Custom headers travel as JSON, and the consume side normalizes them so every path — a Vulthil
+envelope, a bare-AMQP message from a non-Vulthil producer, or an outbox-relayed publish — surfaces
+the same CLR primitives in `IMessageContext.Headers`: strings arrive as `string`, booleans as
+`bool`, and numbers as `int`, `long`, or `double` (the narrowest that represents the value). A
+header published as `AddHeader("tenant", "acme")` is the string `"acme"` again in the consumer.
+Values without a JSON primitive form keep their JSON shape: objects and arrays surface as
+`JsonElement`, and types JSON serializes as strings (e.g. `Guid`, `DateTimeOffset`) surface as
+that string.
+
 ## Point-to-point Send
 
 `IPublisher.PublishAsync` fans a message out via its per-type exchange to any number of
@@ -441,6 +452,15 @@ public record Fault<TMessage> where TMessage : notnull
 }
 ```
 
+`Message` is the faulted message's own payload — for envelope-wrapped (Vulthil-produced) deliveries
+the wire envelope is unwrapped before the fault is built — so a subscriber reads the original fields
+with a plain deserialization:
+
+```csharp
+var fault = JsonSerializer.Deserialize<Fault<OrderCreatedEvent>>(body, jsonOptions)!;
+var orderId = fault.Message.OrderId;
+```
+
 The fault exchange is a diagnostics/observability broadcast — drain it with a monitoring service or
 any AMQP consumer bound to the exchange — rather than a typed `IConsumer<Fault<T>>` endpoint.
 

docs/articles/outbox-pattern.md

@@ -156,6 +156,13 @@ Capture is relational-only (it enlists in the ambient transaction); the relay wo
 the general publish/send **filter pipeline** (`IPublishFilter`, registered via `AddPublishFilter<T>()`), which is the
 publish-side counterpart to consume filters and can host other cross-cutting concerns.
 
+The capture preserves the publish faithfully: the message id, correlation id, routing key, send destination, custom
+headers, and the current trace context (`TraceParent`/`TraceState`, when tracing is enabled) are stored with the row
+and re-applied on relay — the relayed publish runs under an activity linked to the capturing trace. Headers persist
+as JSON, so a relayed message and a directly-published one carry identical header bytes on the wire and surface
+identically to consumers; only CLR types with no JSON primitive form (e.g. a `Guid` header value) are re-published
+in their JSON-serialized form rather than as the original CLR type.
+
 On startup the relay waits for the broker transport to finish declaring its subscriber topology (exchanges, queues,
 and bindings) before its first publish — otherwise the commit-time trigger could relay a message before the
 subscriber queues exist, and a pub/sub message with no bound queue is silently dropped. This is wired automatically

src/Vulthil.Messaging.Abstractions/Consumers/IMessageContext.cs

@@ -55,6 +55,16 @@ public interface IMessageContext
     /// <summary>
     /// Gets the transport headers associated with the message, containing custom metadata.
     /// </summary>
+    /// <remarks>
+    /// Values are normalized when the context is built, so every consume path (envelope, bare AMQP, outbox
+    /// relay) surfaces the same CLR primitives: strings arrive as <see cref="string"/>, booleans as
+    /// <see cref="bool"/>, and numbers as <see cref="int"/>, <see cref="long"/>, or <see cref="double"/> —
+    /// the narrowest of the three that represents the value. Header values published as non-primitive
+    /// objects have no CLR type on the wire and surface in their JSON form: as a
+    /// <see cref="System.Text.Json.JsonElement"/> when published as an object or array, or as a
+    /// <see cref="string"/> for types JSON represents as strings (e.g. <see cref="Guid"/>,
+    /// <see cref="DateTimeOffset"/>).
+    /// </remarks>
     IReadOnlyDictionary<string, object?> Headers { get; }
 
     // --- Timing & Lifecycle ---

src/Vulthil.Messaging.Outbox/BrokerOutboxMetadata.cs

@@ -16,6 +16,12 @@ internal sealed record BrokerOutboxMetadata
 
     public string? DestinationAddress { get; init; }
 
+    /// <summary>
+    /// The captured publish headers. Values persist as JSON, so the relay reproduces JSON-primitive values
+    /// exactly and republishes anything else as its serialized JSON form — a relayed message's headers match
+    /// a directly-published message's on the wire, but CLR types with no JSON representation (e.g.
+    /// <see cref="Guid"/>) are not rematerialized as their original type.
+    /// </summary>
     [JsonConverter(typeof(BrokerOutboxMetadataHeadersConverter))]
     public Dictionary<string, object?>? Headers { get; init; }
 }

src/Vulthil.Messaging.Outbox/BrokerOutboxMetadataHeadersConverter.cs

@@ -3,6 +3,13 @@
 
 namespace Vulthil.Messaging.Outbox;
 
+/// <summary>
+/// Round-trips <see cref="BrokerOutboxMetadata.Headers"/> through the persisted metadata JSON so a relayed
+/// publish re-serializes to the same wire bytes as the captured one: primitives rematerialize as the CLR
+/// types the consume side normalizes to (<see cref="string"/>, <see cref="bool"/>, and <see cref="int"/>/
+/// <see cref="long"/>/<see cref="double"/> — the narrowest that represents the value), and non-primitive
+/// values rematerialize as detached <see cref="JsonElement"/>s, which serialize back verbatim.
+/// </summary>
 internal sealed class BrokerOutboxMetadataHeadersConverter : JsonConverter<Dictionary<string, object?>>
 {
     public override Dictionary<string, object?> Read(ref Utf8JsonReader reader, Type typeToConvert, JsonSerializerOptions options)
@@ -55,16 +62,21 @@ public override void Write(Utf8JsonWriter writer, Dictionary<string, object?> va
             case JsonTokenType.Null:
                 return null;
             case JsonTokenType.Number:
-                if (reader.TryGetInt64(out var integer))
+                if (reader.TryGetInt32(out var intValue))
                 {
-                    return integer;
+                    return intValue;
+                }
+
+                if (reader.TryGetInt64(out var longValue))
+                {
+                    return longValue;
                 }
 
                 return reader.GetDouble();
             default:
                 using (var document = JsonDocument.ParseValue(ref reader))
                 {
-                    return document.RootElement.GetRawText();
+                    return document.RootElement.Clone();
                 }
         }
     }

src/Vulthil.Messaging.RabbitMq/Consumers/AmqpHeaderValueNormalizer.cs

@@ -0,0 +1,32 @@
+using System.Text;
+
+namespace Vulthil.Messaging.RabbitMq.Consumers;
+
+/// <summary>
+/// Normalizes received AMQP header values so the bare-AMQP compat path surfaces the same CLR primitives as the
+/// envelope path: the client wire-encodes every string header as a UTF-8 byte array, so byte arrays decode back
+/// to <see cref="string"/>, and nested tables/arrays are normalized recursively. Values the client already
+/// surfaces as typed primitives (<see cref="int"/>, <see cref="long"/>, <see cref="bool"/>, ...) pass through
+/// unchanged.
+/// </summary>
+internal static class AmqpHeaderValueNormalizer
+{
+    public static Dictionary<string, object?> Normalize(IDictionary<string, object?> headers)
+    {
+        var normalized = new Dictionary<string, object?>(headers.Count);
+        foreach (var (key, value) in headers)
+        {
+            normalized[key] = NormalizeValue(value);
+        }
+
+        return normalized;
+    }
+
+    private static object? NormalizeValue(object? value) => value switch
+    {
+        byte[] bytes => Encoding.UTF8.GetString(bytes),
+        List<object?> list => list.ConvertAll(NormalizeValue),
+        IDictionary<string, object?> table => Normalize(table),
+        _ => value,
+    };
+}

src/Vulthil.Messaging.RabbitMq/Consumers/MessageContextFactory.cs

@@ -1,3 +1,4 @@
+using RabbitMQ.Client;
 using RabbitMQ.Client.Events;
 using Vulthil.Messaging.Abstractions.Consumers;
 using Vulthil.Messaging.Abstractions.Publishers;
@@ -8,7 +9,10 @@ namespace Vulthil.Messaging.RabbitMq.Consumers;
 /// <summary>
 /// Builds <see cref="MessageContext"/> instances from RabbitMQ deliveries: the AMQP property/header paths plus a
 /// serializable snapshot for faults. The envelope receive path delegates the transport-agnostic mapping to
-/// <see cref="MessageContext.CreateFromEnvelope{TMessage}"/>.
+/// <see cref="MessageContext.CreateFromEnvelope{TMessage}"/>. Header values are normalized via
+/// <see cref="AmqpHeaderValueNormalizer"/> so both receive paths honor the <see cref="IMessageContext.Headers"/>
+/// contract, and a per-message TTL is anchored to the delivery's timestamp when one is present (the AMQP
+/// expiration is relative to publish, not to consumption).
 /// </summary>
 internal static class MessageContextFactory
 {
@@ -83,13 +87,14 @@ private static MessageContext BuildMetadata(BasicDeliverEventArgs ea)
     {
         var props = ea.BasicProperties;
         var headers = props.Headers ?? new Dictionary<string, object?>();
+        var sentTime = GetSentTime(props);
         return new MessageContext
         {
             MessageId = props.MessageId,
             CorrelationId = props.CorrelationId,
             RequestId = props.CorrelationId,
             RoutingKey = ea.RoutingKey,
-            Headers = headers.ToDictionary(),
+            Headers = AmqpHeaderValueNormalizer.Normalize(headers),
             Redelivered = ea.Redelivered,
             RetryCount = RabbitMqConstants.GetRetryCount(headers),
             ConversationId = RabbitMqConstants.GetHeaderString(headers, "ConversationId"),
@@ -99,8 +104,8 @@ private static MessageContext BuildMetadata(BasicDeliverEventArgs ea)
             ResponseAddress = RabbitMqConstants.GetHeaderUri(headers, "ResponseAddress")
                 ?? (string.IsNullOrEmpty(props.ReplyTo) ? null : new Uri($"queue:{props.ReplyTo}")),
             FaultAddress = RabbitMqConstants.GetHeaderUri(headers, "FaultAddress"),
-            SentTime = props.Timestamp.UnixTime > 0 ? DateTimeOffset.FromUnixTimeSeconds(props.Timestamp.UnixTime) : null,
-            ExpirationTime = RabbitMqConstants.TryParseExpiration(props.Expiration)
+            SentTime = sentTime,
+            ExpirationTime = RabbitMqConstants.TryParseExpiration(props.Expiration, sentTime)
         };
     }
 
@@ -113,6 +118,7 @@ private static MessageContext<TMessage> BuildTypedMetadata<TMessage>(
     {
         var props = ea.BasicProperties;
         var headers = props.Headers ?? new Dictionary<string, object?>();
+        var sentTime = GetSentTime(props);
         return new MessageContext<TMessage>
         {
             Message = message,
@@ -123,7 +129,7 @@ private static MessageContext<TMessage> BuildTypedMetadata<TMessage>(
             CorrelationId = props.CorrelationId,
             RequestId = props.CorrelationId,
             RoutingKey = ea.RoutingKey,
-            Headers = headers.ToDictionary(),
+            Headers = AmqpHeaderValueNormalizer.Normalize(headers),
             Redelivered = ea.Redelivered,
             RetryCount = RabbitMqConstants.GetRetryCount(headers),
             ConversationId = RabbitMqConstants.GetHeaderString(headers, "ConversationId"),
@@ -133,8 +139,11 @@ private static MessageContext<TMessage> BuildTypedMetadata<TMessage>(
             ResponseAddress = RabbitMqConstants.GetHeaderUri(headers, "ResponseAddress")
                 ?? (string.IsNullOrEmpty(props.ReplyTo) ? null : new Uri($"queue:{props.ReplyTo}")),
             FaultAddress = RabbitMqConstants.GetHeaderUri(headers, "FaultAddress"),
-            SentTime = props.Timestamp.UnixTime > 0 ? DateTimeOffset.FromUnixTimeSeconds(props.Timestamp.UnixTime) : null,
-            ExpirationTime = RabbitMqConstants.TryParseExpiration(props.Expiration)
+            SentTime = sentTime,
+            ExpirationTime = RabbitMqConstants.TryParseExpiration(props.Expiration, sentTime)
         };
     }
+
+    private static DateTimeOffset? GetSentTime(IReadOnlyBasicProperties props)
+        => props.Timestamp.UnixTime > 0 ? DateTimeOffset.FromUnixTimeSeconds(props.Timestamp.UnixTime) : null;
 }

src/Vulthil.Messaging.RabbitMq/Consumers/RabbitMqConsumerWorker.cs

@@ -181,7 +181,7 @@ private async Task ProcessAsync(PreparedDelivery prepared, BasicDeliverEventArgs
 
             activity?.SetStatus(ActivityStatusCode.Error, ex.Message);
             activity?.AddException(ex);
-            await HandleFailureAsync(ex, ea, prepared.DiagnosticTypeName);
+            await HandleFailureAsync(ex, ea, prepared.Envelope, prepared.DiagnosticTypeName);
         }
     }
 
@@ -216,7 +216,7 @@ private async Task ExecuteWithInMemoryRetryAsync(RetryPolicyDefinition policy, P
                     activity?.SetStatus(ActivityStatusCode.Error, ex.Message);
                     activity?.AddException(ex);
                     MessagingLog.ConsumerFailed(_logger, ex, _queueDefinition.Name, prepared.DiagnosticTypeName, ea.RoutingKey);
-                    await PublishFaultAsync(ex, ea, ea.BasicProperties.Headers ?? new Dictionary<string, object?>(), prepared.DiagnosticTypeName);
+                    await PublishFaultAsync(ex, ea, ea.BasicProperties.Headers ?? new Dictionary<string, object?>(), prepared.Envelope, prepared.DiagnosticTypeName);
                     await NackAsync(ea);
                     return;
                 }
@@ -262,7 +262,7 @@ private async Task ExecuteWithInMemoryRetryAsync(RetryPolicyDefinition policy, P
         return activity;
     }
 
-    private async Task HandleFailureAsync(Exception ex, BasicDeliverEventArgs ea, string messageTypeName)
+    private async Task HandleFailureAsync(Exception ex, BasicDeliverEventArgs ea, MessageEnvelope? envelope, string messageTypeName)
     {
         var plan = _typeCache.GetPlan(messageTypeName);
         var policy = GetPolicy(plan, _queueDefinition);
@@ -272,7 +272,7 @@ private async Task HandleFailureAsync(Exception ex, BasicDeliverEventArgs ea, st
         if (policy is null)
         {
             MessagingLog.ConsumerFailed(_logger, ex, _queueDefinition.Name, messageTypeName, ea.RoutingKey);
-            await PublishFaultAsync(ex, ea, headers, messageTypeName);
+            await PublishFaultAsync(ex, ea, headers, envelope, messageTypeName);
             await NackAsync(ea);
             return;
         }
@@ -296,28 +296,32 @@ private async Task HandleFailureAsync(Exception ex, BasicDeliverEventArgs ea, st
         }
 
         MessagingLog.ConsumerFailed(_logger, ex, _queueDefinition.Name, messageTypeName, ea.RoutingKey);
-        await PublishFaultAsync(ex, ea, headers, messageTypeName);
+        await PublishFaultAsync(ex, ea, headers, envelope, messageTypeName);
         await NackAsync(ea);
     }
 
     /// <summary>
     /// Publishes a <see cref="Fault{TMessage}"/> for a terminally-failed delivery. When the delivery carries an
     /// explicit <c>FaultAddress</c> the fault is routed point-to-point to that address (via the broker's default
     /// exchange); otherwise it is published by convention to the shared fault exchange with the faulted message's
-    /// URN as the routing key. Exactly one fault is emitted per failure. Publishing is best-effort: a failure to
-    /// publish the fault is logged and never disrupts settling the original delivery.
+    /// URN as the routing key. Exactly one fault is emitted per failure. The fault's <c>Message</c> is the original
+    /// message payload: for an envelope-wrapped delivery the wire envelope is unwrapped (re-parsing the body when
+    /// the caller has no parsed envelope at hand), so a subscriber can deserialize the fault as
+    /// <see cref="Fault{TMessage}"/> of the faulted message type. Publishing is best-effort: a failure to publish
+    /// the fault is logged and never disrupts settling the original delivery.
     /// </summary>
-    private async Task PublishFaultAsync(Exception ex, BasicDeliverEventArgs ea, IDictionary<string, object?> headers, string messageTypeName)
+    private async Task PublishFaultAsync(Exception ex, BasicDeliverEventArgs ea, IDictionary<string, object?> headers, MessageEnvelope? envelope, string messageTypeName)
     {
         var (exchange, routingKey) = ResolveFaultRoute(headers, _messageConfigurationProvider.FaultExchangeName, messageTypeName);
 
         try
         {
-            var originalBody = JsonSerializer.Deserialize<JsonElement>(ea.Body.Span, _jsonOptions);
+            var faultedEnvelope = envelope ?? TryParseEnvelope(ea.Body, _jsonOptions);
+            var originalMessage = faultedEnvelope?.Message ?? JsonSerializer.Deserialize<JsonElement>(ea.Body.Span, _jsonOptions);
 
             var fault = new Fault<JsonElement>
             {
-                Message = originalBody,
+                Message = originalMessage,
                 ExceptionMessage = ex.Message,
                 StackTrace = ex.StackTrace,
                 ExceptionType = ex.GetType().FullName ?? "Unknown",

src/Vulthil.Messaging.RabbitMq/RabbitMqConstants.cs

@@ -22,13 +22,18 @@ public static int GetRetryCount(IDictionary<string, object?>? headers)
         return 0;
     }
 
-    public static DateTimeOffset? TryParseExpiration(string? expiration)
+    /// <summary>
+    /// Maps a per-message AMQP TTL to an absolute expiration instant. The TTL is relative to when the message
+    /// was published, so it is anchored to <paramref name="sentTime"/> when the delivery carries a timestamp;
+    /// without one the consume-side clock is the only anchor available and the result is an upper bound.
+    /// </summary>
+    public static DateTimeOffset? TryParseExpiration(string? expiration, DateTimeOffset? sentTime)
     {
         if (!string.IsNullOrWhiteSpace(expiration) && long.TryParse(expiration, out var ms))
         {
             try
             {
-                return DateTimeOffset.UtcNow.AddMilliseconds(ms);
+                return (sentTime ?? DateTimeOffset.UtcNow).AddMilliseconds(ms);
             }
             catch (ArgumentOutOfRangeException)
             {

src/Vulthil.Messaging/Transport/HeaderValueNormalizer.cs

@@ -0,0 +1,44 @@
+using System.Text.Json;
+
+namespace Vulthil.Messaging.Transport;
+
+/// <summary>
+/// Normalizes header values that crossed the JSON wire so consumers observe stable CLR primitives instead of
+/// <see cref="JsonElement"/> wrappers: strings, booleans, and numbers unwrap to <see cref="string"/>,
+/// <see cref="bool"/>, and <see cref="int"/>/<see cref="long"/>/<see cref="double"/> (the narrowest of the
+/// three that represents the value). Objects and arrays stay <see cref="JsonElement"/>, and values that never
+/// crossed the wire (in-memory transports) pass through unchanged, so the normalization is idempotent.
+/// </summary>
+internal static class HeaderValueNormalizer
+{
+    public static object? Normalize(object? value) => value switch
+    {
+        JsonElement element => NormalizeElement(element),
+        _ => value,
+    };
+
+    private static object? NormalizeElement(JsonElement element) => element.ValueKind switch
+    {
+        JsonValueKind.String => element.GetString(),
+        JsonValueKind.True => true,
+        JsonValueKind.False => false,
+        JsonValueKind.Null or JsonValueKind.Undefined => null,
+        JsonValueKind.Number => NormalizeNumber(element),
+        _ => element,
+    };
+
+    private static object NormalizeNumber(JsonElement element)
+    {
+        if (element.TryGetInt32(out var intValue))
+        {
+            return intValue;
+        }
+
+        if (element.TryGetInt64(out var longValue))
+        {
+            return longValue;
+        }
+
+        return element.GetDouble();
+    }
+}