From 94c5fa8820ee31153a84445f93763c21d3001ec4 Mon Sep 17 00:00:00 2001 From: tobias-tengler <45513122+tobias-tengler@users.noreply.github.com> Date: Wed, 21 Oct 2026 14:06:38 +0000 Subject: [PATCH] [Mocha] Fix PostgreSQL timestamps and outbox expiration --- .../Outbox/OutboxProcessor.cs | 13 ++ .../PostgresMessageEnvelopeParser.cs | 17 +- .../PostgresMessageHeaders.cs | 5 + .../PostgresMessageHeadersWriter.cs | 7 +- .../PostgresOutboxExpirationTests.cs | 175 ++++++++++++++++++ .../PostgresDispatchHeaderTests.cs | 57 ++++++ .../PostgresMessageEnvelopeParserTests.cs | 49 ++++- 7 files changed, 317 insertions(+), 6 deletions(-) create mode 100644 src/Mocha/test/Mocha.EntityFrameworkCore.Postgres.Tests/PostgresOutboxExpirationTests.cs diff --git a/src/Mocha/src/Mocha.EntityFrameworkCore.Postgres/Outbox/OutboxProcessor.cs b/src/Mocha/src/Mocha.EntityFrameworkCore.Postgres/Outbox/OutboxProcessor.cs index 3fb5a798fe3..40a7e4314c9 100644 --- a/src/Mocha/src/Mocha.EntityFrameworkCore.Postgres/Outbox/OutboxProcessor.cs +++ b/src/Mocha/src/Mocha.EntityFrameworkCore.Postgres/Outbox/OutboxProcessor.cs @@ -19,6 +19,7 @@ public sealed class PostgresOutboxProcessor private readonly IOutboxSignal _signal; private readonly ObjectPool _contextPool; private readonly PostgresMessageOutboxOptions _options; + private readonly TimeProvider _timeProvider; /// /// Initializes a new instance of the class. @@ -55,6 +56,7 @@ internal PostgresOutboxProcessor( _signal = signal; _contextPool = pools.DispatchContext; _options = options; + _timeProvider = services.GetService() ?? TimeProvider.System; } /// @@ -170,6 +172,14 @@ private async Task ProcessEventAsync(NpgsqlConnection connection, Cancella return true; } + if (envelope.DeliverBy is { } deadline && deadline <= _timeProvider.GetUtcNow()) + { + await reader.CloseAsync(); + await DeleteEventAsync(connection, id, transaction, cancellationToken); + _logger.OutboxMessageExpired(id); + return true; + } + var messageType = GetMessageType(envelope.MessageType); if (messageType is null) @@ -397,6 +407,9 @@ internal static partial class Logs LogLevel.Critical, "Message with ID {Id} exhausted all {MaxAttempts} attempts and will not be retried.")] public static partial void OutboxMessageExhausted(this ILogger logger, Guid id, int maxAttempts); + + [LoggerMessage(8, LogLevel.Information, "Expired outbox message with ID {Id} was discarded.")] + public static partial void OutboxMessageExpired(this ILogger logger, Guid id); } file static class Serializer diff --git a/src/Mocha/src/Mocha.Transport.Postgres/PostgresMessageEnvelopeParser.cs b/src/Mocha/src/Mocha.Transport.Postgres/PostgresMessageEnvelopeParser.cs index 6c32c021b9b..f3b9074770c 100644 --- a/src/Mocha/src/Mocha.Transport.Postgres/PostgresMessageEnvelopeParser.cs +++ b/src/Mocha/src/Mocha.Transport.Postgres/PostgresMessageEnvelopeParser.cs @@ -29,6 +29,7 @@ public MessageEnvelope Parse(PostgresMessageItem messageItem) string? faultAddress = null; string? contentType = null; string? messageType = null; + DateTimeOffset? sentAt = null; DateTimeOffset? deliverBy = null; DateTimeOffset? scheduledTime = null; ImmutableArray? enclosedMessageTypes = null; @@ -117,6 +118,20 @@ public MessageEnvelope Parse(PostgresMessageItem messageItem) reader.Skip(); } } + else if (reader.ValueTextEquals(PostgresMessageHeaders.SentAt)) + { + reader.Read(); + + if (reader.TokenType == JsonTokenType.String + && reader.TryGetDateTimeOffset(out var parsed)) + { + sentAt = parsed; + } + else + { + reader.Skip(); + } + } else if (reader.ValueTextEquals(PostgresMessageHeaders.DeliverBy)) { reader.Read(); @@ -186,7 +201,7 @@ public MessageEnvelope Parse(PostgresMessageItem messageItem) MessageType = messageType, EnclosedMessageTypes = enclosedMessageTypes, Headers = customHeaders ?? Headers.Empty(), - SentAt = new DateTimeOffset(messageItem.SentTime, TimeSpan.Zero), + SentAt = sentAt ?? new DateTimeOffset(messageItem.SentTime, TimeSpan.Zero), DeliverBy = deliverBy, ScheduledTime = scheduledTime, DeliveryCount = messageItem.DeliveryCount, diff --git a/src/Mocha/src/Mocha.Transport.Postgres/PostgresMessageHeaders.cs b/src/Mocha/src/Mocha.Transport.Postgres/PostgresMessageHeaders.cs index eb4eaf98700..c0266a932da 100644 --- a/src/Mocha/src/Mocha.Transport.Postgres/PostgresMessageHeaders.cs +++ b/src/Mocha/src/Mocha.Transport.Postgres/PostgresMessageHeaders.cs @@ -60,6 +60,11 @@ internal static class PostgresMessageHeaders /// public static ReadOnlySpan EnclosedMessageTypes => "enclosedMessageTypes"u8; + /// + /// Header key for the time the original publication was created. + /// + public static ReadOnlySpan SentAt => "sentAt"u8; + /// /// Header key for the deadline by which the message must be delivered and processed. /// diff --git a/src/Mocha/src/Mocha.Transport.Postgres/PostgresMessageHeadersWriter.cs b/src/Mocha/src/Mocha.Transport.Postgres/PostgresMessageHeadersWriter.cs index ae27966ebd7..c3d9f24a140 100644 --- a/src/Mocha/src/Mocha.Transport.Postgres/PostgresMessageHeadersWriter.cs +++ b/src/Mocha/src/Mocha.Transport.Postgres/PostgresMessageHeadersWriter.cs @@ -73,6 +73,11 @@ public static ReadOnlyMemory Write(JsonHeadersFeature feature, MessageEnve writer.WriteEndArray(); } + if (envelope.SentAt is { } sentAt) + { + writer.WriteString(PostgresMessageHeaders.SentAt, sentAt); + } + if (envelope.DeliverBy is { } deliverBy) { writer.WriteString(PostgresMessageHeaders.DeliverBy, deliverBy.ToString("O")); @@ -87,7 +92,7 @@ public static ReadOnlyMemory Write(JsonHeadersFeature feature, MessageEnve { foreach (var header in envelope.Headers) { - if (header.Value is not null) + if (header.Value is not null && header.Key != "sentAt") { writer.WritePropertyName(header.Key); JsonSerializer.Serialize(writer, header.Value, header.Value.GetType()); diff --git a/src/Mocha/test/Mocha.EntityFrameworkCore.Postgres.Tests/PostgresOutboxExpirationTests.cs b/src/Mocha/test/Mocha.EntityFrameworkCore.Postgres.Tests/PostgresOutboxExpirationTests.cs new file mode 100644 index 00000000000..343f94b7a66 --- /dev/null +++ b/src/Mocha/test/Mocha.EntityFrameworkCore.Postgres.Tests/PostgresOutboxExpirationTests.cs @@ -0,0 +1,175 @@ +using System.Text.Json; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Time.Testing; +using Mocha.EntityFrameworkCore.Postgres.Tests.Helpers; +using Mocha.Middlewares; +using Mocha.Outbox; +using Mocha.Transport.InMemory; + +namespace Mocha.EntityFrameworkCore.Postgres.Tests; + +public sealed class PostgresOutboxExpirationTests(PostgresFixture fixture) : IClassFixture +{ + private static readonly TimeSpan s_timeout = TimeSpan.FromSeconds(30); + private static readonly DateTimeOffset s_now = new(2050, 1, 1, 12, 0, 0, TimeSpan.Zero); + + private static CancellationToken TestToken => TestContext.Current.CancellationToken; + + [Theory] + [InlineData(-1, false)] + [InlineData(0, false)] + [InlineData(-1, true)] + [InlineData(0, true)] + public async Task ProcessAsync_Should_DiscardExpiredMessage_When_DeadlineHasBeenReached( + int seconds, + bool unresolvable) + { + // arrange + var connectionString = await fixture.CreateDatabaseAsync(); + var recorder = new DeliveryRecorder(); + await using var provider = await CreateProviderAsync( + connectionString, new FakeTimeProvider(s_now), recorder); + var envelope = CreateEnvelope(provider, s_now.AddSeconds(seconds), unresolvable); + var id = await MessageRows.InsertOutboxAsync(connectionString, envelope, TestToken); + + // act + await using var host = await MessagingTestHost.StartAsync(provider, TestToken); + var remainingAttempts = await MessageRows.WaitForOutboxRetryAsync( + connectionString, id, s_timeout, TestToken); + + // assert + Assert.Null(remainingAttempts); + Assert.Equal(0, recorder.Attempts); + Assert.False(recorder.Delivered.Task.IsCompleted); + } + + [Theory] + [InlineData(false)] + [InlineData(true)] + public async Task ProcessAsync_Should_DeliverMessage_When_DeadlineHasNotBeenReached(bool hasDeadline) + { + // arrange + var connectionString = await fixture.CreateDatabaseAsync(); + var recorder = new DeliveryRecorder(); + await using var provider = await CreateProviderAsync( + connectionString, new FakeTimeProvider(s_now), recorder); + var envelope = CreateEnvelope(provider, hasDeadline ? s_now.AddHours(1) : null, unresolvable: false); + var id = await MessageRows.InsertOutboxAsync(connectionString, envelope, TestToken); + + // act + await using var host = await MessagingTestHost.StartAsync(provider, TestToken); + var delivered = await recorder.Delivered.Task.WaitAsync(s_timeout, TestToken); + var remainingAttempts = await MessageRows.WaitForOutboxRetryAsync( + connectionString, id, s_timeout, TestToken); + + // assert + Assert.Null(remainingAttempts); + Assert.Equal(1, recorder.Attempts); + Assert.Equal("pending", delivered.Value); + } + + [Fact] + public async Task ProcessAsync_Should_DiscardExpiredMessage_When_DeadlinePassesAfterFailedDelivery() + { + // arrange + var connectionString = await fixture.CreateDatabaseAsync(); + var clock = new FakeTimeProvider(s_now); + var recorder = new DeliveryRecorder + { + OnFirstAttempt = () => + { + clock.Advance(TimeSpan.FromHours(2)); + throw new InvalidOperationException("The first delivery failed deliberately."); + } + }; + await using var provider = await CreateProviderAsync(connectionString, clock, recorder); + var id = await MessageRows.InsertOutboxAsync( + connectionString, CreateEnvelope(provider, s_now.AddHours(1), unresolvable: false), TestToken); + + // act + await using var host = await MessagingTestHost.StartAsync(provider, TestToken); + var remainingAttempts = await MessageRows.WaitForOutboxRetryAsync( + connectionString, id, s_timeout, TestToken); + + // assert + Assert.Null(remainingAttempts); + Assert.Equal(1, recorder.Attempts); + Assert.False(recorder.Delivered.Task.IsCompleted); + } + + private static MessageEnvelope CreateEnvelope( + IServiceProvider provider, + DateTimeOffset? deadline, + bool unresolvable) + { + var runtime = provider.GetRequiredService(); + var type = runtime.GetMessageType(typeof(PendingEvent)); + return new MessageEnvelope + { + MessageId = Guid.NewGuid().ToString(), + MessageType = unresolvable ? "urn:message:unknown" : type.Identity, + ContentType = MessageContentType.Json.Value, + DestinationAddress = unresolvable ? "unknown://destination" : runtime.GetPublishEndpoint(type).Address.ToString(), + DeliverBy = deadline, + Body = JsonSerializer.SerializeToUtf8Bytes(new PendingEvent { Value = "pending" }) + }; + } + + private static async Task CreateProviderAsync( + string connectionString, + TimeProvider clock, + DeliveryRecorder recorder) + { + var services = new ServiceCollection(); + services.AddSingleton(clock); + services.AddSingleton(recorder); + services.AddSingleton(recorder); + services.AddDbContext(options => options.UseTestNpgsql(connectionString)); + var builder = services.AddMessageBus(); + builder.AddEntityFramework(options => options.UsePostgresOutbox()); + builder.AddEventHandler(); + builder.AddInMemory(); + + var provider = services.BuildServiceProvider(); + await ((MessagingRuntime)provider.GetRequiredService()).StartAsync(TestToken); + await using var scope = provider.CreateAsyncScope(); + await scope.ServiceProvider.GetRequiredService().Database.EnsureCreatedAsync(TestToken); + return provider; + } + + public sealed class PendingEvent + { + public required string Value { get; init; } + } + + public sealed class PendingEventHandler(DeliveryRecorder recorder) : IEventHandler + { + public ValueTask HandleAsync(PendingEvent message, CancellationToken cancellationToken) + { + recorder.Delivered.TrySetResult(message); + return default; + } + } + + public sealed class DeliveryRecorder : MessagingDiagnosticEventListener + { + private int _attempts; + + public int Attempts => Volatile.Read(ref _attempts); + + public Action? OnFirstAttempt { get; init; } + + public TaskCompletionSource Delivered { get; } = + new(TaskCreationOptions.RunContinuationsAsynchronously); + + public override IDisposable Dispatch(IDispatchContext context) + { + if (Interlocked.Increment(ref _attempts) == 1) + { + OnFirstAttempt?.Invoke(); + } + + return EmptyScope; + } + } +} diff --git a/src/Mocha/test/Mocha.Transport.Postgres.Tests/PostgresDispatchHeaderTests.cs b/src/Mocha/test/Mocha.Transport.Postgres.Tests/PostgresDispatchHeaderTests.cs index f9e49793a93..2765863bae6 100644 --- a/src/Mocha/test/Mocha.Transport.Postgres.Tests/PostgresDispatchHeaderTests.cs +++ b/src/Mocha/test/Mocha.Transport.Postgres.Tests/PostgresDispatchHeaderTests.cs @@ -117,6 +117,63 @@ public void Headers_Should_RoundTrip_DeliverBy_When_Set() Assert.Equal(deliverBy, parsed.DeliverBy.Value); } + [Fact] + public void Headers_Should_PreservePublicationTime_When_TransportSendOccursLater() + { + // arrange + var sentAt = new DateTimeOffset(2024, 1, 15, 12, 0, 0, TimeSpan.FromMinutes(330)).AddTicks(1234567); + var envelope = new MessageEnvelope { SentAt = sentAt }; + + // act + var parsed = RoundTrip(envelope); + + // assert + Assert.Equal(sentAt, parsed.SentAt); + Assert.Equal(sentAt.Offset, parsed.SentAt!.Value.Offset); + Assert.Empty(parsed.Headers!); + } + + [Fact] + public void Headers_Should_PreservePublicationTime_When_CustomHeaderUsesReservedName() + { + // arrange + var sentAt = new DateTimeOffset(2024, 1, 15, 12, 0, 0, TimeSpan.Zero); + var envelope = new MessageEnvelope + { + SentAt = sentAt, + Headers = new Headers( + [ + new HeaderValue { Key = "sentAt", Value = "2030-01-01T00:00:00Z" }, + new HeaderValue { Key = "x-tenant", Value = "acme" } + ]) + }; + + // act + var parsed = RoundTrip(envelope); + + // assert + Assert.Equal(sentAt, parsed.SentAt); + var header = Assert.Single(parsed.Headers!); + Assert.Equal("x-tenant", header.Key); + Assert.Equal("acme", header.Value); + } + + [Fact] + public void Headers_Should_OmitReservedCustomHeader_When_PublicationTimeAbsent() + { + // arrange + var envelope = new MessageEnvelope + { + Headers = new Headers([new HeaderValue { Key = "sentAt", Value = "2030-01-01T00:00:00Z" }]) + }; + + // act + var headerBytes = WriteHeadersJson(envelope); + + // assert + Assert.True(headerBytes.IsEmpty); + } + [Fact] public void Headers_Should_RoundTrip_CustomHeaders_When_Set() { diff --git a/src/Mocha/test/Mocha.Transport.Postgres.Tests/PostgresMessageEnvelopeParserTests.cs b/src/Mocha/test/Mocha.Transport.Postgres.Tests/PostgresMessageEnvelopeParserTests.cs index 7503f01aed5..57315bda2ec 100644 --- a/src/Mocha/test/Mocha.Transport.Postgres.Tests/PostgresMessageEnvelopeParserTests.cs +++ b/src/Mocha/test/Mocha.Transport.Postgres.Tests/PostgresMessageEnvelopeParserTests.cs @@ -152,18 +152,59 @@ public void Parse_Should_ExtractBody_When_BodyPresent() } [Fact] - public void Parse_Should_ExtractSentAt_When_SentTimePresent() + public void Parse_Should_UseTransportSentTime_When_SentAtHeaderAbsent() { // arrange var sentTime = new DateTime(2024, 1, 15, 12, 0, 0, DateTimeKind.Utc); - var item = CreateMessageItem(sentTime: sentTime); + var item = CreateMessageItem(headers: new { messageId = "legacy" }, sentTime: sentTime); // act var envelope = _parser.Parse(item); // assert - Assert.NotNull(envelope.SentAt); - Assert.Equal(new DateTimeOffset(sentTime, TimeSpan.Zero), envelope.SentAt.Value); + Assert.Equal(new DateTimeOffset(sentTime, TimeSpan.Zero), envelope.SentAt); + } + + [Fact] + public void Parse_Should_PreservePublicationTime_When_TransportSendOccursLater() + { + // arrange + var sentAt = new DateTimeOffset(2024, 1, 15, 12, 0, 0, TimeSpan.FromHours(2)); + var item = CreateMessageItem( + headers: new { sentAt }, + sentTime: new DateTime(2024, 1, 15, 13, 0, 0, DateTimeKind.Utc)); + + // act + var envelope = _parser.Parse(item); + + // assert + Assert.Equal(sentAt, envelope.SentAt); + Assert.Equal(sentAt.Offset, envelope.SentAt!.Value.Offset); + Assert.Empty(envelope.Headers!); + } + + [Theory] + [InlineData("null")] + [InlineData("123")] + [InlineData("true")] + [InlineData("\"invalid\"")] + [InlineData("{}")] + [InlineData("[]")] + public void Parse_Should_UseTransportSentTime_When_SentAtHeaderInvalid(string sentAt) + { + // arrange + var sentTime = new DateTime(2024, 1, 15, 13, 0, 0, DateTimeKind.Utc); + using var headers = JsonDocument.Parse( + $$"""{"sentAt": {{sentAt}}, "messageId": "after-timestamp"}"""); + var item = CreateMessageItem(headers: headers.RootElement, sentTime: sentTime); + + // act + var envelope = _parser.Parse(item); + + // assert + Assert.Equal(new DateTimeOffset(sentTime, TimeSpan.Zero), envelope.SentAt); + Assert.Equal("after-timestamp", envelope.MessageId); + Assert.Empty(envelope.Headers!); } [Fact]