PG-215 Check for old processing events and move them to the dead letters queue

Author: DavidHogue

src/Learning.EventStore/Learning.EventStore.csproj

@@ -2,7 +2,7 @@
 
   <PropertyGroup>
     <AssemblyTitle>Learning.EventStore</AssemblyTitle>
-    <VersionPrefix>11.2.1</VersionPrefix>
+    <VersionPrefix>11.3.0-beta.10</VersionPrefix>
     <TargetFrameworks>netstandard2.0</TargetFrameworks>
     <AssemblyName>Learning.EventStore</AssemblyName>
     <PackageId>Learning.EventStore</PackageId>

src/Learning.MessageQueue/Messages/RetryableAsyncRedisSubscription.cs

@@ -1,7 +1,9 @@
 using System;
 using System.Threading.Tasks;
+using Learning.EventStore.Common.Sql;
 using Learning.MessageQueue.Logging;
 using Learning.MessageQueue.Repository;
+using Microsoft.Extensions.Logging;
 using Newtonsoft.Json;
 using StackExchange.Redis;
 
@@ -37,6 +39,8 @@ protected RetryableAsyncRedisSubscription(IEventSubscriber subscriber, IMessageQ
 
         public virtual async Task RetryAsync()
         {
+            await MoveStaleProcessingEventsToDeadLetters();
+
             var eventType = typeof(T).Name;
             var listLength = await _messageQueueRepository.GetDeadLetterListLength<T>().ConfigureAwait(false);
             var eventsProcessed = 0;
@@ -120,5 +124,27 @@ protected virtual async Task<bool> ShouldRetry(IMessage @event, RedisValue event
 
             return true;
         }
+
+        private async Task MoveStaleProcessingEventsToDeadLetters()
+        {
+            const int eventCount = 10;
+            var waitTime = TimeSpan.FromMinutes(5);
+            var staleTimeStamp = DateTimeOffset.UtcNow - waitTime;
+
+            var unprocessedEvents = await _messageQueueRepository.GetOldestProcessingEvents<T>(eventCount);
+
+            for (var i = 0; i < unprocessedEvents.Length; i++)
+            {
+                var unprocessedEvent = unprocessedEvents[i];
+                var @event = JsonConvert.DeserializeObject<T>(unprocessedEvent);
+
+                if (@event.TimeStamp < staleTimeStamp)
+                {
+                    _logger.Info($"Moving {typeof(T).Name} {@event.Id} with timestamp @{@event.TimeStamp} to dead letter queue.");
+
+                    await _messageQueueRepository.MoveProcessingEventToDeadLetterQueue<T>(unprocessedEvent, @event);
+                }
+            }
+        }
     }
 }

src/Learning.MessageQueue/Messages/RetryableRedisSubscription.cs

@@ -37,6 +37,8 @@ protected RetryableRedisSubscription(IEventSubscriber subscriber, IMessageQueueR
 
         public virtual async Task RetryAsync()
         {
+            await MoveStaleProcessingEventsToDeadLetters();
+
             var eventType = typeof(T).Name;
             var listLength = await _messageQueueRepository.GetDeadLetterListLength<T>().ConfigureAwait(false);
             var eventsProcessed = 0;
@@ -123,5 +125,27 @@ protected virtual async Task<bool> ShouldRetry(IMessage @event, RedisValue event
 
             return true;
         }
+
+        private async Task MoveStaleProcessingEventsToDeadLetters()
+        {
+            const int eventCount = 10;
+            var waitTime = TimeSpan.FromMinutes(5);
+            var staleTimeStamp = DateTimeOffset.UtcNow - waitTime;
+
+            var unprocessedEvents = await _messageQueueRepository.GetOldestProcessingEvents<T>(eventCount);
+
+            for (var i = 0; i < unprocessedEvents.Length; i++)
+            {
+                var unprocessedEvent = unprocessedEvents[i];
+                var @event = JsonConvert.DeserializeObject<T>(unprocessedEvent);
+
+                if (@event.TimeStamp < staleTimeStamp)
+                {
+                    _logger.Info($"Moving {typeof(T).Name} {@event.Id} with timestamp @{@event.TimeStamp} to dead letter queue.");
+
+                    await _messageQueueRepository.MoveProcessingEventToDeadLetterQueue<T>(unprocessedEvent, @event);
+                }
+            }
+        }
     }
 }

src/Learning.MessageQueue/Repository/IMessageQueueRepository.cs

@@ -10,6 +10,7 @@ public interface IMessageQueueRepository
     {
         Task<long> GetDeadLetterListLength<T>() where T : IMessage;
 
+        // TODO: Rename to GetDeadLetterMessage
         Task<RedisValue> GetUnprocessedMessage<T>(int index) where T : IMessage;
 
         void AddToDeadLetterQueue<T>(RedisValue eventData, IMessage @event, Exception exception) where T : IMessage;
@@ -19,5 +20,9 @@ public interface IMessageQueueRepository
         Task UpdateRetryData(IMessage @event, string exceptionMessage);
 
         Task<RetryData> GetRetryData(IMessage @event);
+
+        Task<RedisValue[]> GetOldestProcessingEvents<T>(int count) where T : IMessage;
+
+        Task MoveProcessingEventToDeadLetterQueue<T>(RedisValue eventData, IMessage @event) where T : IMessage;
     }
 }

src/Learning.MessageQueue/Repository/MessageQueueRepository.cs

@@ -1,7 +1,9 @@
 using System;
 using System.Linq;
+using System.Reflection;
 using System.Threading.Tasks;
 using Learning.EventStore.Common.Redis;
+using Learning.EventStore.Common.Sql;
 using Learning.MessageQueue.Messages;
 using StackExchange.Redis;
 
@@ -33,6 +35,7 @@ public async Task<long> GetDeadLetterListLength<T>() where T : IMessage
             return listLength;
         }
 
+        // TODO: Rename to GetDeadLetterMessage
         public async Task<RedisValue> GetUnprocessedMessage<T>(int index) where T : IMessage
         {
             var deadLetterListKey = GetDeadLetterListKey<T>();
@@ -95,6 +98,41 @@ public async Task<RetryData> GetRetryData(IMessage @event)
             return data;
         }
 
+        public async Task<RedisValue[]> GetOldestProcessingEvents<T>(int count) where T : IMessage
+        {
+            var processingEventsListKey = GetProcessingEventsListKey<T>();
+            var listLength = await _redisClient.ListLengthAsync(processingEventsListKey).ConfigureAwait(false);
+            long startIndex;
+            
+            if (listLength == 0)
+            {
+                return new RedisValue[0];
+            }
+            if (listLength > count)
+            {
+                startIndex = listLength - count;
+            }
+            else
+            {
+                startIndex = 0;
+            }
+
+            var unprocessedEvents = await _redisClient.ListRangeAsync(processingEventsListKey, startIndex, -1).ConfigureAwait(false);
+            return unprocessedEvents;
+        }
+
+        public async Task MoveProcessingEventToDeadLetterQueue<T>(RedisValue eventData, IMessage @event) where T : IMessage
+        {
+            var deadLetterListKey = GetDeadLetterListKey<T>();
+            var processingEventsListKey = GetProcessingEventsListKey<T>();
+            var tran = _redisClient.CreateTransaction();
+
+            tran.ListLeftPushAsync(deadLetterListKey, eventData);
+            tran.ListRemoveAsync(processingEventsListKey, eventData, -1);
+
+            await ExcecuteTransaction(tran, @event.Id).ConfigureAwait(false);
+        }
+
         private string GetDeadLetterListKey<T>() where T : IMessage
         {
             var eventType = typeof(T).Name;
@@ -104,6 +142,15 @@ private string GetDeadLetterListKey<T>() where T : IMessage
             return processingListKey;
         }
 
+        private string GetProcessingEventsListKey<T>() where T : IMessage
+        {
+            var eventType = typeof(T).Name;
+            var eventKey = $"{_environment}:{eventType}";
+            var processingListKey = $"{_applicationName}:{{{eventKey}}}:ProcessingEvents";
+
+            return processingListKey;
+        }
+
         private string GetRetryDataHashKey(IMessage @event)
         {
             var eventType = @event.GetType().Name;
@@ -122,5 +169,6 @@ private async Task ExcecuteTransaction(ITransaction tran, string aggregateId)
                 throw new Exception($"Redis transaction failed for AggregateId {aggregateId}");
             }
         }
+
     }
 }