[Mocha] Fix PostgreSQL timestamps and outbox expiration

src/Mocha/src/Mocha.EntityFrameworkCore.Postgres/Outbox/OutboxProcessor.cs

@@ -19,6 +19,7 @@ public sealed class PostgresOutboxProcessor
     private readonly IOutboxSignal _signal;
     private readonly ObjectPool<DispatchContext> _contextPool;
     private readonly PostgresMessageOutboxOptions _options;
+    private readonly TimeProvider _timeProvider;
 
     /// <summary>
     /// Initializes a new instance of the <see cref="PostgresOutboxProcessor"/> class.
@@ -55,6 +56,7 @@ internal PostgresOutboxProcessor(
         _signal = signal;
         _contextPool = pools.DispatchContext;
         _options = options;
+        _timeProvider = services.GetService<TimeProvider>() ?? TimeProvider.System;
     }
 
     /// <summary>
@@ -170,6 +172,14 @@ private async Task<bool> 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

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<string>? 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,

src/Mocha/src/Mocha.Transport.Postgres/PostgresMessageHeaders.cs

@@ -60,6 +60,11 @@ internal static class PostgresMessageHeaders
     /// </summary>
     public static ReadOnlySpan<byte> EnclosedMessageTypes => "enclosedMessageTypes"u8;
 
+    /// <summary>
+    /// Header key for the time the original publication was created.
+    /// </summary>
+    public static ReadOnlySpan<byte> SentAt => "sentAt"u8;
+
     /// <summary>
     /// Header key for the deadline by which the message must be delivered and processed.
     /// </summary>

src/Mocha/src/Mocha.Transport.Postgres/PostgresMessageHeadersWriter.cs

@@ -73,6 +73,11 @@ public static ReadOnlyMemory<byte> 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<byte> 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());

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<PostgresFixture>
+{
+    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<IMessagingRuntime>();
+        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<ServiceProvider> CreateProviderAsync(
+        string connectionString,
+        TimeProvider clock,
+        DeliveryRecorder recorder)
+    {
+        var services = new ServiceCollection();
+        services.AddSingleton(clock);
+        services.AddSingleton(recorder);
+        services.AddSingleton<IMessagingDiagnosticEventListener>(recorder);
+        services.AddDbContext<TestDbContext>(options => options.UseTestNpgsql(connectionString));
+        var builder = services.AddMessageBus();
+        builder.AddEntityFramework<TestDbContext>(options => options.UsePostgresOutbox());
+        builder.AddEventHandler<PendingEventHandler>();
+        builder.AddInMemory();
+
+        var provider = services.BuildServiceProvider();
+        await ((MessagingRuntime)provider.GetRequiredService<IMessagingRuntime>()).StartAsync(TestToken);
+        await using var scope = provider.CreateAsyncScope();
+        await scope.ServiceProvider.GetRequiredService<TestDbContext>().Database.EnsureCreatedAsync(TestToken);
+        return provider;
+    }
+
+    public sealed class PendingEvent
+    {
+        public required string Value { get; init; }
+    }
+
+    public sealed class PendingEventHandler(DeliveryRecorder recorder) : IEventHandler<PendingEvent>
+    {
+        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<PendingEvent> Delivered { get; } =
+            new(TaskCreationOptions.RunContinuationsAsynchronously);
+
+        public override IDisposable Dispatch(IDispatchContext context)
+        {
+            if (Interlocked.Increment(ref _attempts) == 1)
+            {
+                OnFirstAttempt?.Invoke();
+            }
+
+            return EmptyScope;
+        }
+    }
+}

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()
     {