From 067ec90617671064feb74f86fd8528e4d045de54 Mon Sep 17 00:00:00 2001 From: tobias-tengler <45513122+tobias-tengler@users.noreply.github.com> Date: Tue, 20 Oct 2026 09:40:32 +0000 Subject: [PATCH 1/2] [Mocha] Fix the RabbitMQ transport losing the exact SentAt and DeliverBy --- .../RabbitMQMessageEnvelopeFormatter.cs | 32 ++++-- .../RabbitMQMessageEnvelopeParser.cs | 34 +++++- .../RabbitMQMessageHeaders.cs | 15 +++ .../RabbitMQMessageEnvelopeFormatterTests.cs | 52 ++++++++- .../RabbitMQMessageEnvelopeParserTests.cs | 107 ++++++++++++++++++ 5 files changed, 225 insertions(+), 15 deletions(-) diff --git a/src/Mocha/src/Mocha.Transport.RabbitMQ/RabbitMQMessageEnvelopeFormatter.cs b/src/Mocha/src/Mocha.Transport.RabbitMQ/RabbitMQMessageEnvelopeFormatter.cs index f45096cf8bd..7d173fe4fb7 100644 --- a/src/Mocha/src/Mocha.Transport.RabbitMQ/RabbitMQMessageEnvelopeFormatter.cs +++ b/src/Mocha/src/Mocha.Transport.RabbitMQ/RabbitMQMessageEnvelopeFormatter.cs @@ -8,8 +8,8 @@ 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 @@ -17,8 +17,8 @@ public static BasicProperties Format(MessageEnvelope envelope, TimeProvider time 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, @@ -26,10 +26,12 @@ public static BasicProperties Format(MessageEnvelope envelope, TimeProvider time }; } - private static IDictionary FormatHeaders(MessageEnvelope envelope) + private static IDictionary 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"; diff --git a/src/Mocha/src/Mocha.Transport.RabbitMQ/RabbitMQMessageEnvelopeParser.cs b/src/Mocha/src/Mocha.Transport.RabbitMQ/RabbitMQMessageEnvelopeParser.cs index 3abf21a3566..39489030bf3 100644 --- a/src/Mocha/src/Mocha.Transport.RabbitMQ/RabbitMQMessageEnvelopeParser.cs +++ b/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? fieldTable) { if (fieldTable is null || fieldTable.Count == 0) @@ -69,12 +79,32 @@ public static Headers ParseHeaders(this IDictionary? 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? fieldTable, + ContextDataKey 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 key) { var value = headers.GetValue(key.Key); diff --git a/src/Mocha/src/Mocha.Transport.RabbitMQ/RabbitMQMessageHeaders.cs b/src/Mocha/src/Mocha.Transport.RabbitMQ/RabbitMQMessageHeaders.cs index b0248017670..efa2fd0f98d 100644 --- a/src/Mocha/src/Mocha.Transport.RabbitMQ/RabbitMQMessageHeaders.cs +++ b/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. /// public static readonly ContextDataKey DeliveryCount = new("x-delivery-count"); + + /// + /// Header key for the time the message was sent, stored as Unix milliseconds. + /// + public static readonly ContextDataKey SentAt = new("x-sent-at"); + + /// + /// Header key for the time by which the message must be delivered, stored as Unix milliseconds. + /// + public static readonly ContextDataKey DeliverBy = new("x-deliver-by"); + + /// + /// Returns whether the header carries an envelope property and is therefore not a message header. + /// + public static bool IsEnvelopeProperty(string key) => key == SentAt.Key || key == DeliverBy.Key; } diff --git a/src/Mocha/test/Mocha.Transport.RabbitMQ.Tests/RabbitMQMessageEnvelopeFormatterTests.cs b/src/Mocha/test/Mocha.Transport.RabbitMQ.Tests/RabbitMQMessageEnvelopeFormatterTests.cs index 0888c475436..146692aaef8 100644 --- a/src/Mocha/test/Mocha.Transport.RabbitMQ.Tests/RabbitMQMessageEnvelopeFormatterTests.cs +++ b/src/Mocha/test/Mocha.Transport.RabbitMQ.Tests/RabbitMQMessageEnvelopeFormatterTests.cs @@ -52,6 +52,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 +314,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 +323,27 @@ public void Format_Should_ReturnEmptyHeaders_When_EnvelopeIsEmpty() var headers = RabbitMQMessageEnvelopeFormatter.Format(envelope, s_timeProvider).Headers!; // assert - Assert.Empty(headers); + Assert.Equal(new Dictionary { ["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 { ["x-sent-at"] = s_now.ToUnixTimeMilliseconds() }, headers); } [Fact] @@ -352,7 +391,8 @@ public void Format_Should_MapToTableValues_When_TypesHaveNoWireForm() ["element"] = new Dictionary { ["k"] = 1 }, ["node"] = new Dictionary { ["k"] = 1 }, ["nested"] = new Dictionary { ["inner"] = "00:00:00" }, - ["set"] = new List { 7L } + ["set"] = new List { 7L }, + ["x-sent-at"] = s_now.ToUnixTimeMilliseconds() }, headers); } @@ -422,7 +462,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 +494,8 @@ public void Format_Should_MapJsonToTableShapes_When_HeaderIsAJsonValue() ["object"] = new Dictionary { ["k"] = 1 }, ["array"] = new List { 1, 2 }, ["text"] = "abc", - ["number"] = 5 + ["number"] = 5, + ["x-sent-at"] = s_now.ToUnixTimeMilliseconds() }, headers); } diff --git a/src/Mocha/test/Mocha.Transport.RabbitMQ.Tests/RabbitMQMessageEnvelopeParserTests.cs b/src/Mocha/test/Mocha.Transport.RabbitMQ.Tests/RabbitMQMessageEnvelopeParserTests.cs index cb332824ed0..0c4c9582c43 100644 --- a/src/Mocha/test/Mocha.Transport.RabbitMQ.Tests/RabbitMQMessageEnvelopeParserTests.cs +++ b/src/Mocha/test/Mocha.Transport.RabbitMQ.Tests/RabbitMQMessageEnvelopeParserTests.cs @@ -113,6 +113,90 @@ public void Parse_Should_ReturnNullSentAt_When_TimestampZero() Assert.Null(envelope.SentAt); } + [Fact] + public void Parse_Should_PreferSentAtHeader_When_HeaderAndTimestampPresent() + { + // arrange + var args = CreateDeliverEventArgs(props => + { + props.Timestamp = new AmqpTimestamp(1700000000); + props.Headers = new Dictionary { ["x-sent-at"] = 1699999990750L }; + }); + + // act + var envelope = _parser.Parse(args, TimeProvider.System); + + // assert + Assert.Equal(DateTimeOffset.FromUnixTimeMilliseconds(1699999990750), envelope.SentAt); + } + + [Fact] + public void Parse_Should_FallbackToTimestamp_When_SentAtHeaderIsOutOfRange() + { + // arrange + var args = CreateDeliverEventArgs(props => + { + props.Timestamp = new AmqpTimestamp(1700000000); + props.Headers = new Dictionary { ["x-sent-at"] = long.MaxValue }; + }); + + // act + var envelope = _parser.Parse(args, TimeProvider.System); + + // assert + Assert.Equal(DateTimeOffset.FromUnixTimeSeconds(1700000000), envelope.SentAt); + } + + [Fact] + public void Parse_Should_HideTimeHeaders_When_HeadersPresent() + { + // arrange + var args = CreateDeliverEventArgs(props => + props.Headers = new Dictionary + { + ["x-sent-at"] = 1700000000750L, + ["x-deliver-by"] = 1700000060750L, + ["x-tenant"] = "acme"u8.ToArray() + }); + + // act + var envelope = _parser.Parse(args, TimeProvider.System); + + // assert + Assert.Collection( + envelope.Headers!, + header => + { + Assert.Equal("x-tenant", header.Key); + Assert.Equal("acme", header.Value); + }); + } + + [Fact] + public void Parse_Should_RestoreExactTimes_When_PublishedLaterThanSent() + { + // arrange + // an outbox or scheduler publishes the stored envelope after it was sent + var sentAt = new DateTimeOffset(2026, 8, 12, 10, 0, 0, 750, TimeSpan.Zero); + var deliverBy = sentAt.AddHours(2).AddMilliseconds(125); + var published = RabbitMQMessageEnvelopeFormatter.Format( + new MessageEnvelope { SentAt = sentAt, DeliverBy = deliverBy }, + new FixedTimeProvider(sentAt.AddHours(1))); + var args = CreateDeliverEventArgs(props => + { + props.Timestamp = published.Timestamp; + props.Expiration = published.Expiration; + props.Headers = published.Headers; + }); + + // act + var envelope = _parser.Parse(args, TimeProvider.System); + + // assert + Assert.Equal(sentAt, envelope.SentAt); + Assert.Equal(deliverBy, envelope.DeliverBy); + } + [Fact] public void Parse_Should_ExtractBody_When_BodyPresent() { @@ -454,6 +538,24 @@ public void Parse_Should_ComputeDeliverByFromSentAt_When_TimestampAndExpirationP Assert.Equal(sentAt.AddMilliseconds(60000), envelope.DeliverBy.Value); } + [Fact] + public void Parse_Should_PreferDeliverByHeader_When_HeaderAndExpirationPresent() + { + // arrange + var args = CreateDeliverEventArgs(props => + { + props.Timestamp = new AmqpTimestamp(1700000000); + props.Expiration = "60000"; + props.Headers = new Dictionary { ["x-deliver-by"] = 1700000060125L }; + }); + + // act + var envelope = _parser.Parse(args, TimeProvider.System); + + // assert + Assert.Equal(DateTimeOffset.FromUnixTimeMilliseconds(1700000060125), envelope.DeliverBy); + } + [Fact] public void Parse_Should_FallbackToUtcNow_When_ExpirationPresentButNoTimestamp() { @@ -545,4 +647,9 @@ private static BasicDeliverEventArgs CreateDeliverEventArgs( properties: props, body: body ?? ReadOnlyMemory.Empty); } + + private sealed class FixedTimeProvider(DateTimeOffset utcNow) : TimeProvider + { + public override DateTimeOffset GetUtcNow() => utcNow; + } } From bd304886bc07acb48f8f55eaca0b130b8db832fa Mon Sep 17 00:00:00 2001 From: tobias-tengler <45513122+tobias-tengler@users.noreply.github.com> Date: Tue, 20 Oct 2026 10:03:25 +0000 Subject: [PATCH 2/2] Use FakeTimeProvider in the RabbitMQ envelope tests --- .../Mocha.Transport.RabbitMQ.Tests.csproj | 1 + .../RabbitMQMessageEnvelopeFormatterTests.cs | 8 ++------ .../RabbitMQMessageEnvelopeParserTests.cs | 8 ++------ 3 files changed, 5 insertions(+), 12 deletions(-) diff --git a/src/Mocha/test/Mocha.Transport.RabbitMQ.Tests/Mocha.Transport.RabbitMQ.Tests.csproj b/src/Mocha/test/Mocha.Transport.RabbitMQ.Tests/Mocha.Transport.RabbitMQ.Tests.csproj index 1fb147ec10e..ceedb6bef2a 100644 --- a/src/Mocha/test/Mocha.Transport.RabbitMQ.Tests/Mocha.Transport.RabbitMQ.Tests.csproj +++ b/src/Mocha/test/Mocha.Transport.RabbitMQ.Tests/Mocha.Transport.RabbitMQ.Tests.csproj @@ -7,6 +7,7 @@ + diff --git a/src/Mocha/test/Mocha.Transport.RabbitMQ.Tests/RabbitMQMessageEnvelopeFormatterTests.cs b/src/Mocha/test/Mocha.Transport.RabbitMQ.Tests/RabbitMQMessageEnvelopeFormatterTests.cs index 146692aaef8..5ee2825bbf9 100644 --- a/src/Mocha/test/Mocha.Transport.RabbitMQ.Tests/RabbitMQMessageEnvelopeFormatterTests.cs +++ b/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() @@ -597,9 +598,4 @@ private enum TestPriority Low, High } - - private sealed class FixedTimeProvider(DateTimeOffset utcNow) : TimeProvider - { - public override DateTimeOffset GetUtcNow() => utcNow; - } } diff --git a/src/Mocha/test/Mocha.Transport.RabbitMQ.Tests/RabbitMQMessageEnvelopeParserTests.cs b/src/Mocha/test/Mocha.Transport.RabbitMQ.Tests/RabbitMQMessageEnvelopeParserTests.cs index 0c4c9582c43..eebc3ad0a45 100644 --- a/src/Mocha/test/Mocha.Transport.RabbitMQ.Tests/RabbitMQMessageEnvelopeParserTests.cs +++ b/src/Mocha/test/Mocha.Transport.RabbitMQ.Tests/RabbitMQMessageEnvelopeParserTests.cs @@ -1,3 +1,4 @@ +using Microsoft.Extensions.Time.Testing; using Mocha.Middlewares; using RabbitMQ.Client; using RabbitMQ.Client.Events; @@ -181,7 +182,7 @@ public void Parse_Should_RestoreExactTimes_When_PublishedLaterThanSent() var deliverBy = sentAt.AddHours(2).AddMilliseconds(125); var published = RabbitMQMessageEnvelopeFormatter.Format( new MessageEnvelope { SentAt = sentAt, DeliverBy = deliverBy }, - new FixedTimeProvider(sentAt.AddHours(1))); + new FakeTimeProvider(sentAt.AddHours(1))); var args = CreateDeliverEventArgs(props => { props.Timestamp = published.Timestamp; @@ -647,9 +648,4 @@ private static BasicDeliverEventArgs CreateDeliverEventArgs( properties: props, body: body ?? ReadOnlyMemory.Empty); } - - private sealed class FixedTimeProvider(DateTimeOffset utcNow) : TimeProvider - { - public override DateTimeOffset GetUtcNow() => utcNow; - } }