feat(relay): raise MessageExpired from the cleanup sweep

Previously MessageExpired only fired from the delivery path, so messages that
expired while waiting in the queue and were removed by the background cleanup
sweep were dropped without notification.

- Change IRelayQueue.ClearExpiredAsync to return the newly-expired messages
  (IReadOnlyList<IRelayMessage>) instead of a count. Messages already in the
  Expired state (already reported via the delivery path) are removed but not
  returned, so each message is reported exactly once.
- Add RelayService.SweepExpiredMessagesAsync, which raises MessageExpired for
  each returned message; the cleanup loop now calls it.
- Update InMemoryRelayQueue, the QueueManagementExample and the relay README.
- Add tests for the sweep raising events and the empty-sweep case (12 total).

BREAKING CHANGE: IRelayQueue.ClearExpiredAsync now returns
IReadOnlyList<IRelayMessage> instead of int. Custom IRelayQueue
implementations and callers must be updated.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Author: Taiizor

examples/Zetian.Relay.Examples/QueueManagementExample.cs

@@ -401,8 +401,8 @@ private static async Task RescheduleMessage(Zetian.Relay.Abstractions.IRelayQueu
 
         private static async Task ClearExpiredMessages(Zetian.Relay.Abstractions.IRelayQueue queue)
         {
-            int count = await queue.ClearExpiredAsync();
-            Console.WriteLine($"[INFO] Cleared {count} expired messages");
+            var expired = await queue.ClearExpiredAsync();
+            Console.WriteLine($"[INFO] Cleared {expired.Count} expired messages");
         }
 
         private static async Task SendMoreMessages(SmtpClient client)

src/Zetian.Relay/Abstractions/IRelayQueue.cs

@@ -94,9 +94,12 @@ Task<IReadOnlyList<IRelayMessage>> GetByStatusAsync(
             CancellationToken cancellationToken = default);
 
         /// <summary>
-        /// Clears expired messages from the queue
+        /// Removes expired messages from the queue and returns the messages that newly
+        /// transitioned to the expired state during this call. Messages that were already
+        /// marked as expired (e.g. reported via the delivery path) are removed but not
+        /// returned, so callers can raise an expiry notification exactly once per message.
         /// </summary>
-        Task<int> ClearExpiredAsync(CancellationToken cancellationToken = default);
+        Task<IReadOnlyList<IRelayMessage>> ClearExpiredAsync(CancellationToken cancellationToken = default);
 
         /// <summary>
         /// Gets queue statistics

src/Zetian.Relay/Queue/InMemoryRelayQueue.cs

@@ -306,7 +306,7 @@ public Task<IReadOnlyList<IRelayMessage>> GetByStatusAsync(
             return Task.FromResult<IReadOnlyList<IRelayMessage>>(messages);
         }
 
-        public async Task<int> ClearExpiredAsync(CancellationToken cancellationToken = default)
+        public async Task<IReadOnlyList<IRelayMessage>> ClearExpiredAsync(CancellationToken cancellationToken = default)
         {
             await _queueLock.WaitAsync(cancellationToken).ConfigureAwait(false);
             try
@@ -315,15 +315,26 @@ public async Task<int> ClearExpiredAsync(CancellationToken cancellationToken = d
                     .Where(m => m.IsExpired)
                     .ToList();
 
+                List<IRelayMessage> newlyExpired = [];
+
                 foreach (RelayMessage message in expired)
                 {
+                    // Messages already in the Expired state were reported via the delivery
+                    // path; remove them but do not surface them again to avoid double-firing.
+                    bool alreadyExpired = message.Status == RelayStatus.Expired;
+
                     if (_messages.TryRemove(message.QueueId, out _))
                     {
                         if (message.Status == RelayStatus.InProgress)
                         {
                             Interlocked.Decrement(ref _activeDeliveries);
                         }
-                        message.MarkExpired();
+
+                        if (!alreadyExpired)
+                        {
+                            message.MarkExpired();
+                            newlyExpired.Add(message);
+                        }
                     }
                 }
 
@@ -332,7 +343,7 @@ public async Task<int> ClearExpiredAsync(CancellationToken cancellationToken = d
                     _logger.LogInformation("Cleared {Count} expired messages from queue", expired.Count);
                 }
 
-                return expired.Count;
+                return newlyExpired;
             }
             finally
             {

src/Zetian.Relay/README.MD

@@ -236,8 +236,9 @@ var deferredMessages = await relayService.Queue.GetByStatusAsync(RelayStatus.Def
 // Remove a message
 await relayService.Queue.RemoveAsync(queueId);
 
-// Clear expired messages
-var cleared = await relayService.Queue.ClearExpiredAsync();
+// Clear expired messages (returns the messages that newly expired)
+var expired = await relayService.Queue.ClearExpiredAsync();
+Console.WriteLine($"Cleared {expired.Count} expired messages");
 ```
 
 ### Message Priority

src/Zetian.Relay/Services/RelayService.cs

@@ -384,13 +384,7 @@ private async Task CleanupExpiredMessagesAsync(CancellationToken cancellationTok
                     await Task.Delay(Configuration.CleanupInterval, cancellationToken)
                         .ConfigureAwait(false);
 
-                    int count = await Queue.ClearExpiredAsync(cancellationToken)
-                        .ConfigureAwait(false);
-
-                    if (count > 0)
-                    {
-                        _logger.LogInformation("Cleaned up {Count} expired messages", count);
-                    }
+                    await SweepExpiredMessagesAsync(cancellationToken).ConfigureAwait(false);
                 }
                 catch (OperationCanceledException)
                 {
@@ -405,6 +399,29 @@ await Task.Delay(Configuration.CleanupInterval, cancellationToken)
             _logger.LogInformation("Cleanup task stopped");
         }
 
+        /// <summary>
+        /// Performs a single sweep of the queue for expired messages, raising
+        /// <see cref="MessageExpired"/> for each message that expired while waiting in the
+        /// queue (i.e. without being picked up by the delivery path, which reports expiry itself).
+        /// </summary>
+        internal async Task SweepExpiredMessagesAsync(CancellationToken cancellationToken)
+        {
+            IReadOnlyList<IRelayMessage> expiredMessages = await Queue.ClearExpiredAsync(cancellationToken)
+                .ConfigureAwait(false);
+
+            if (expiredMessages.Count == 0)
+            {
+                return;
+            }
+
+            _logger.LogInformation("Cleaned up {Count} expired messages", expiredMessages.Count);
+
+            foreach (IRelayMessage expiredMessage in expiredMessages)
+            {
+                OnMessageExpired(new RelayDeliveryEventArgs(expiredMessage) { Error = "Message expired" });
+            }
+        }
+
         private async Task<bool> CanRelayAsync(ISmtpSession session, ISmtpMessage message)
         {
             // Allow relay if authenticated

tests/Zetian.Relay.Tests/RelayServiceEventTests.cs

@@ -1,5 +1,6 @@
 using System;
 using System.Collections.Generic;
+using System.Linq;
 using System.Net.Mail;
 using System.Threading;
 using System.Threading.Tasks;
@@ -275,6 +276,47 @@ public async Task MessageExpired_IsRaised_WhenMessageHasExpired()
             client.Verify(c => c.SendAsync(It.IsAny<ISmtpMessage>(), It.IsAny<CancellationToken>()), Times.Never);
         }
 
+        [Fact]
+        public async Task MessageExpired_IsRaised_FromCleanupSweep_PerReturnedMessage()
+        {
+            Mock<IRelayMessage> first = new();
+            first.SetupGet(m => m.QueueId).Returns("queue-1");
+            Mock<IRelayMessage> second = new();
+            second.SetupGet(m => m.QueueId).Returns("queue-2");
+
+            Mock<IRelayQueue> queue = new();
+            queue.Setup(q => q.ClearExpiredAsync(It.IsAny<CancellationToken>()))
+                .ReturnsAsync(new List<IRelayMessage> { first.Object, second.Object });
+
+            RelayService service = new(CreateConfiguration(), queue.Object, logger: null);
+
+            List<RelayDeliveryEventArgs> raised = new();
+            service.MessageExpired += (_, e) => raised.Add(e);
+
+            await service.SweepExpiredMessagesAsync(CancellationToken.None);
+
+            Assert.Equal(2, raised.Count);
+            Assert.Equal(new[] { "queue-1", "queue-2" }, raised.Select(r => r.QueueId).ToArray());
+            Assert.All(raised, r => Assert.Equal("Message expired", r.Error));
+        }
+
+        [Fact]
+        public async Task MessageExpired_NotRaised_WhenSweepFindsNothing()
+        {
+            Mock<IRelayQueue> queue = new();
+            queue.Setup(q => q.ClearExpiredAsync(It.IsAny<CancellationToken>()))
+                .ReturnsAsync(Array.Empty<IRelayMessage>());
+
+            RelayService service = new(CreateConfiguration(), queue.Object, logger: null);
+
+            bool raised = false;
+            service.MessageExpired += (_, _) => raised = true;
+
+            await service.SweepExpiredMessagesAsync(CancellationToken.None);
+
+            Assert.False(raised);
+        }
+
         [Fact]
         public async Task EventHandlerException_DoesNotPropagate()
         {