[Mocha] Fix the RabbitMQ transport losing the exact SentAt and DeliverBy

src/Mocha/src/Mocha.Transport.RabbitMQ/RabbitMQMessageEnvelopeFormatter.cs

@@ -8,28 +8,30 @@ internal static class RabbitMQMessageEnvelopeFormatter
 {
     public static BasicProperties Format(MessageEnvelope envelope, TimeProvider timeProvider)
     {
-        var sentAt = timeProvider.GetUtcNow();
-        var headers = FormatHeaders(envelope);
+        var publishedAt = timeProvider.GetUtcNow();
+        var headers = FormatHeaders(envelope, envelope.SentAt ?? publishedAt);
         var messageType = envelope.MessageType ?? headers.Get(MessageHeaders.Transport.MessageType);
 
         return new BasicProperties
         {
             MessageId = envelope.MessageId,
             CorrelationId = envelope.CorrelationId,
             Type = messageType,
-            Timestamp = sentAt.ToAmqpTimestamp(),
-            Expiration = envelope.DeliverBy.ToExpiration(sentAt),
+            Timestamp = publishedAt.ToAmqpTimestamp(),
+            Expiration = envelope.DeliverBy.ToExpiration(publishedAt),
             ReplyTo = envelope.ResponseAddress,
             Headers = headers,
             ContentType = envelope.ContentType,
             DeliveryMode = DeliveryModes.Persistent
         };
     }
 
-    private static IDictionary<string, object?> FormatHeaders(MessageEnvelope envelope)
+    private static IDictionary<string, object?> FormatHeaders(MessageEnvelope envelope, DateTimeOffset sentAt)
     {
         var headerCount =
-            (envelope.ConversationId is not null ? 1 : 0)
+            1 // x-sent-at
+            + (envelope.DeliverBy is not null ? 1 : 0)
+            + (envelope.ConversationId is not null ? 1 : 0)
             + (envelope.CausationId is not null ? 1 : 0)
             + (envelope.SourceAddress is not null ? 1 : 0)
             + (envelope.DestinationAddress is not null ? 1 : 0)
@@ -44,6 +46,11 @@ public static BasicProperties Format(MessageEnvelope envelope, TimeProvider time
         {
             foreach (var header in envelope.Headers)
             {
+                if (RabbitMQMessageHeaders.IsEnvelopeProperty(header.Key))
+                {
+                    continue;
+                }
+
                 headers[header.Key] = header.Value.ToFieldTableValue(header.Key);
             }
         }
@@ -83,20 +90,29 @@ public static BasicProperties Format(MessageEnvelope envelope, TimeProvider time
             headers.Set(MessageHeaders.Transport.MessageType, envelope.MessageType);
         }
 
+        // AMQP timestamps have whole seconds and expirations are relative to the publish,
+        // so the exact times travel as headers.
+        headers.Set(RabbitMQMessageHeaders.SentAt, sentAt.ToUnixTimeMilliseconds());
+
+        if (envelope.DeliverBy is { } deliverBy)
+        {
+            headers.Set(RabbitMQMessageHeaders.DeliverBy, deliverBy.ToUnixTimeMilliseconds());
+        }
+
         return headers;
     }
 }
 
 file static class Extensions
 {
-    public static string? ToExpiration(this DateTimeOffset? deliverBy, DateTimeOffset sentAt)
+    public static string? ToExpiration(this DateTimeOffset? deliverBy, DateTimeOffset publishedAt)
     {
         if (deliverBy is null)
         {
             return null;
         }
 
-        var timeToLive = deliverBy.Value - sentAt;
+        var timeToLive = deliverBy.Value - publishedAt;
         if (timeToLive <= TimeSpan.Zero)
         {
             return "0";

src/Mocha/src/Mocha.Transport.RabbitMQ/RabbitMQMessageEnvelopeParser.cs

@@ -21,11 +21,17 @@ internal sealed class RabbitMQMessageEnvelopeParser
     public MessageEnvelope Parse(BasicDeliverEventArgs eventArgs, TimeProvider timeProvider)
     {
         var props = eventArgs.BasicProperties;
-        var sentAt =
+        var publishedAt =
             props.Timestamp.UnixTime > 0
                 ? DateTimeOffset.FromUnixTimeSeconds(props.Timestamp.UnixTime)
                 : (DateTimeOffset?)null;
 
+        // other producers only send the AMQP timestamp and an expiration relative to it
+        var sentAt = props.Headers.GetUnixTimeMilliseconds(RabbitMQMessageHeaders.SentAt) ?? publishedAt;
+        var deliverBy =
+            props.Headers.GetUnixTimeMilliseconds(RabbitMQMessageHeaders.DeliverBy)
+            ?? props.Expiration.ParseExpiration(publishedAt, timeProvider);
+
         var headers = props.Headers.ParseHeaders();
 
         var envelope = new MessageEnvelope
@@ -41,7 +47,7 @@ public MessageEnvelope Parse(BasicDeliverEventArgs eventArgs, TimeProvider timeP
             ContentType = props.ContentType,
             MessageType = props.Type ?? headers.GetString(MessageHeaders.Transport.MessageType),
             SentAt = sentAt,
-            DeliverBy = props.Expiration.ParseExpiration(sentAt, timeProvider),
+            DeliverBy = deliverBy,
             DeliveryCount = headers.GetDeliveryCount(eventArgs.Redelivered),
             Headers = headers,
             EnclosedMessageTypes = headers.GetStringArray(MessageHeaders.Transport.EnclosedMessageTypes),
@@ -59,6 +65,10 @@ public MessageEnvelope Parse(BasicDeliverEventArgs eventArgs, TimeProvider timeP
 
 file static class Extensions
 {
+    private static readonly long s_minUnixTimeMilliseconds = DateTimeOffset.MinValue.ToUnixTimeMilliseconds();
+
+    private static readonly long s_maxUnixTimeMilliseconds = DateTimeOffset.MaxValue.ToUnixTimeMilliseconds();
+
     public static Headers ParseHeaders(this IDictionary<string, object?>? fieldTable)
     {
         if (fieldTable is null || fieldTable.Count == 0)
@@ -69,12 +79,32 @@ public static Headers ParseHeaders(this IDictionary<string, object?>? fieldTable
         var headers = new Headers(fieldTable.Count);
         foreach (var (key, value) in fieldTable)
         {
+            if (RabbitMQMessageHeaders.IsEnvelopeProperty(key))
+            {
+                continue;
+            }
+
             headers.Set(key, value.FromFieldTableValue(key));
         }
 
         return headers;
     }
 
+    public static DateTimeOffset? GetUnixTimeMilliseconds(
+        this IDictionary<string, object?>? fieldTable,
+        ContextDataKey<long> key)
+    {
+        if (fieldTable is not null
+            && fieldTable.TryGet(key, out var milliseconds)
+            && milliseconds >= s_minUnixTimeMilliseconds
+            && milliseconds <= s_maxUnixTimeMilliseconds)
+        {
+            return DateTimeOffset.FromUnixTimeMilliseconds(milliseconds);
+        }
+
+        return null;
+    }
+
     public static string? GetString(this Headers headers, ContextDataKey<string> key)
     {
         var value = headers.GetValue(key.Key);

src/Mocha/src/Mocha.Transport.RabbitMQ/RabbitMQMessageHeaders.cs

@@ -19,4 +19,19 @@ internal static class RabbitMQMessageHeaders
     /// Header key for the delivery count maintained by RabbitMQ quorum queues.
     /// </summary>
     public static readonly ContextDataKey<long> DeliveryCount = new("x-delivery-count");
+
+    /// <summary>
+    /// Header key for the time the message was sent, stored as Unix milliseconds.
+    /// </summary>
+    public static readonly ContextDataKey<long> SentAt = new("x-sent-at");
+
+    /// <summary>
+    /// Header key for the time by which the message must be delivered, stored as Unix milliseconds.
+    /// </summary>
+    public static readonly ContextDataKey<long> DeliverBy = new("x-deliver-by");
+
+    /// <summary>
+    /// Returns whether the header carries an envelope property and is therefore not a message header.
+    /// </summary>
+    public static bool IsEnvelopeProperty(string key) => key == SentAt.Key || key == DeliverBy.Key;
 }

src/Mocha/test/Mocha.Transport.RabbitMQ.Tests/Mocha.Transport.RabbitMQ.Tests.csproj

@@ -7,6 +7,7 @@
   </PropertyGroup>
 
   <ItemGroup>
+    <PackageReference Include="Microsoft.Extensions.TimeProvider.Testing" />
     <PackageReference Include="Moq" />
   </ItemGroup>
 

src/Mocha/test/Mocha.Transport.RabbitMQ.Tests/RabbitMQMessageEnvelopeFormatterTests.cs

@@ -2,6 +2,7 @@
 using System.Collections.Immutable;
 using System.Text.Json;
 using System.Text.Json.Nodes;
+using Microsoft.Extensions.Time.Testing;
 using Mocha.Middlewares;
 using RabbitMQ.Client;
 
@@ -10,7 +11,7 @@ namespace Mocha.Transport.RabbitMQ.Tests;
 public class RabbitMQMessageEnvelopeFormatterTests
 {
     private static readonly DateTimeOffset s_now = new(2026, 8, 12, 10, 30, 0, TimeSpan.Zero);
-    private static readonly TimeProvider s_timeProvider = new FixedTimeProvider(s_now);
+    private static readonly TimeProvider s_timeProvider = new FakeTimeProvider(s_now);
 
     [Fact]
     public void Format_Should_SetBasicProperties_When_EnvelopeContainsMetadata()
@@ -52,6 +53,25 @@ public void Format_Should_SetExpiration_When_EnvelopeHasDeliverBy()
         Assert.Equal("60000", properties.Expiration);
     }
 
+    [Fact]
+    public void Format_Should_SendExactTimes_When_EnvelopeWasSentBeforePublish()
+    {
+        // arrange
+        // an outbox or scheduler publishes the stored envelope after it was sent
+        var sentAt = s_now.AddHours(-1).AddMilliseconds(750);
+        var deliverBy = s_now.AddHours(1).AddMilliseconds(125);
+        var envelope = new MessageEnvelope { SentAt = sentAt, DeliverBy = deliverBy };
+
+        // act
+        var properties = RabbitMQMessageEnvelopeFormatter.Format(envelope, s_timeProvider);
+
+        // assert
+        Assert.Equal(s_now.ToUnixTimeSeconds(), properties.Timestamp.UnixTime);
+        Assert.Equal("3600125", properties.Expiration);
+        Assert.Equal(sentAt.ToUnixTimeMilliseconds(), properties.Headers!["x-sent-at"]);
+        Assert.Equal(deliverBy.ToUnixTimeMilliseconds(), properties.Headers!["x-deliver-by"]);
+    }
+
     [Fact]
     public void Format_Should_ConvertDateTimeOffset_When_HeaderContainsDateTimeOffset()
     {
@@ -295,7 +315,7 @@ public void Format_Should_SetMessageType_When_EnvelopeHasMessageType()
     }
 
     [Fact]
-    public void Format_Should_ReturnEmptyHeaders_When_EnvelopeIsEmpty()
+    public void Format_Should_OnlySetSentAt_When_EnvelopeIsEmpty()
     {
         // arrange
         var envelope = new MessageEnvelope();
@@ -304,7 +324,27 @@ public void Format_Should_ReturnEmptyHeaders_When_EnvelopeIsEmpty()
         var headers = RabbitMQMessageEnvelopeFormatter.Format(envelope, s_timeProvider).Headers!;
 
         // assert
-        Assert.Empty(headers);
+        Assert.Equal(new Dictionary<string, object?> { ["x-sent-at"] = s_now.ToUnixTimeMilliseconds() }, headers);
+    }
+
+    [Fact]
+    public void Format_Should_IgnoreTimeHeaders_When_EnvelopeHeadersContainThem()
+    {
+        // arrange
+        var envelope = new MessageEnvelope
+        {
+            Headers = new Headers(
+            [
+                new HeaderValue { Key = "x-sent-at", Value = 1L },
+                new HeaderValue { Key = "x-deliver-by", Value = 2L }
+            ])
+        };
+
+        // act
+        var headers = RabbitMQMessageEnvelopeFormatter.Format(envelope, s_timeProvider).Headers!;
+
+        // assert
+        Assert.Equal(new Dictionary<string, object?> { ["x-sent-at"] = s_now.ToUnixTimeMilliseconds() }, headers);
     }
 
     [Fact]
@@ -352,7 +392,8 @@ public void Format_Should_MapToTableValues_When_TypesHaveNoWireForm()
                 ["element"] = new Dictionary<string, object?> { ["k"] = 1 },
                 ["node"] = new Dictionary<string, object?> { ["k"] = 1 },
                 ["nested"] = new Dictionary<string, object?> { ["inner"] = "00:00:00" },
-                ["set"] = new List<object?> { 7L }
+                ["set"] = new List<object?> { 7L },
+                ["x-sent-at"] = s_now.ToUnixTimeMilliseconds()
             },
             headers);
     }
@@ -422,7 +463,8 @@ public void Format_Should_CarryDecimalAsText_When_ItExceedsTheTableMantissa()
             {
                 ["fits"] = 9.95m,
                 ["exceeds"] = "21474836.48",
-                ["negative"] = "-21474836.48"
+                ["negative"] = "-21474836.48",
+                ["x-sent-at"] = s_now.ToUnixTimeMilliseconds()
             },
             headers);
     }
@@ -453,7 +495,8 @@ public void Format_Should_MapJsonToTableShapes_When_HeaderIsAJsonValue()
                 ["object"] = new Dictionary<string, object?> { ["k"] = 1 },
                 ["array"] = new List<object?> { 1, 2 },
                 ["text"] = "abc",
-                ["number"] = 5
+                ["number"] = 5,
+                ["x-sent-at"] = s_now.ToUnixTimeMilliseconds()
             },
             headers);
     }
@@ -555,9 +598,4 @@ private enum TestPriority
         Low,
         High
     }
-
-    private sealed class FixedTimeProvider(DateTimeOffset utcNow) : TimeProvider
-    {
-        public override DateTimeOffset GetUtcNow() => utcNow;
-    }
 }