Refactor event handling to use HandledTime property

framework/src/Volo.Abp.EntityFrameworkCore/Volo/Abp/EntityFrameworkCore/DistributedEvents/DbContextEventInbox.cs

@@ -66,7 +66,7 @@ public class DbContextEventInbox<TDbContext> : IDbContextEventInbox<TDbContext>
     {
         var dbContext = await DbContextProvider.GetDbContextAsync();
         await dbContext.IncomingEvents.Where(x => x.Id == id).ExecuteUpdateAsync(x =>
-            x.SetProperty(p => p.Status, _ => IncomingEventStatus.Processed).SetProperty(p => p.DiscardedOrProcessedTime, _ => Clock.Now));
+            x.SetProperty(p => p.Status, _ => IncomingEventStatus.Processed).SetProperty(p => p.HandledTime, _ => Clock.Now));
     }
 
     [UnitOfWork]
@@ -84,7 +84,7 @@ public class DbContextEventInbox<TDbContext> : IDbContextEventInbox<TDbContext>
     {
         var dbContext = await DbContextProvider.GetDbContextAsync();
         await dbContext.IncomingEvents.Where(x => x.Id == id).ExecuteUpdateAsync(x =>
-            x.SetProperty(p => p.Status, _ => IncomingEventStatus.Discarded).SetProperty(p => p.DiscardedOrProcessedTime, _ => Clock.Now));
+            x.SetProperty(p => p.Status, _ => IncomingEventStatus.Discarded).SetProperty(p => p.HandledTime, _ => Clock.Now));
     }
 
     [UnitOfWork]

framework/src/Volo.Abp.EntityFrameworkCore/Volo/Abp/EntityFrameworkCore/DistributedEvents/IncomingEventRecord.cs

@@ -26,7 +26,7 @@ public class IncomingEventRecord :
 
     public IncomingEventStatus Status { get; set; } = IncomingEventStatus.Pending;
 
-    public DateTime? DiscardedOrProcessedTime { get; set; }
+    public DateTime? HandledTime { get; set; }
 
     public int RetryCount { get; set; } = 0;
 
@@ -62,7 +62,11 @@ public class IncomingEventRecord :
             MessageId,
             EventName,
             EventData,
-            CreationTime
+            CreationTime,
+            Status,
+            HandledTime,
+            RetryCount,
+            NextRetryTime
         );
 
         foreach (var property in ExtraProperties)
@@ -76,13 +80,13 @@ public class IncomingEventRecord :
     public void MarkAsProcessed(DateTime processedTime)
     {
         Status = IncomingEventStatus.Processed;
-        DiscardedOrProcessedTime = processedTime;
+        HandledTime = processedTime;
     }
 
     public void MarkAsDiscarded(DateTime discardedTime)
     {
         Status = IncomingEventStatus.Discarded;
-        DiscardedOrProcessedTime = discardedTime;
+        HandledTime = discardedTime;
     }
 
     public void RetryLater(int retryCount, DateTime nextRetryTime)

framework/src/Volo.Abp.EventBus.Abstractions/Volo/Abp/EventBus/Distributed/IncomingEventInfo.cs

@@ -22,7 +22,7 @@ public class IncomingEventInfo : IIncomingEventInfo
 
     public IncomingEventStatus Status { get; set; } = IncomingEventStatus.Pending;
 
-    public DateTime? DiscardedOrProcessedTime { get; set; }
+    public DateTime? HandledTime { get; set; }
 
     public int RetryCount { get; set; } = 0;
 
@@ -39,14 +39,21 @@ public class IncomingEventInfo : IIncomingEventInfo
         string messageId,
         string eventName,
         byte[] eventData,
-        DateTime creationTime)
+        DateTime creationTime,
+        IncomingEventStatus status = IncomingEventStatus.Pending,
+        DateTime? handledTime = null,
+        int retryCount = 0,
+        DateTime? nextRetryTime = null)
     {
         Id = id;
         MessageId = messageId;
         EventName = Check.NotNullOrWhiteSpace(eventName, nameof(eventName), MaxEventNameLength);
         EventData = eventData;
         CreationTime = creationTime;
-
+        Status = status;
+        HandledTime = handledTime;
+        RetryCount = retryCount;
+        NextRetryTime = nextRetryTime;
         ExtraProperties = new ExtraPropertyDictionary();
         this.SetDefaultsForExtraProperties();
     }

framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/Distributed/InboxProcessor.cs

@@ -105,6 +105,8 @@ public class InboxProcessor : IInboxProcessor, ITransientDependency
 
                     foreach (var waitingEvent in waitingEvents)
                     {
+                        Logger.LogInformation($"Start processing the incoming event with id = {waitingEvent.Id:N}");
+
                         if (waitingEvent.NextRetryTime.HasValue && waitingEvent.NextRetryTime.Value > Clock.Now)
                         {
                             Logger.LogInformation($"Event with id = {waitingEvent.Id:N} is not ready to be processed yet. Next retry time: {waitingEvent.NextRetryTime.Value}");
@@ -139,7 +141,7 @@ public class InboxProcessor : IInboxProcessor, ITransientDependency
                             {
                                 using (var uow = UnitOfWorkManager.Begin(isTransactional: true, requiresNew: true))
                                 {
-                                    if (waitingEvent.RetryCount >= InboxProcessorOptions.MaxRetryCount)
+                                    if (waitingEvent.RetryCount > InboxProcessorOptions.MaxRetryCount)
                                     {
                                         Logger.LogWarning($"Max retry count reached for event with id = {waitingEvent.Id:N}. Discarding the event.");
 
@@ -163,6 +165,7 @@ public class InboxProcessor : IInboxProcessor, ITransientDependency
                                 using (var uow = UnitOfWorkManager.Begin(isTransactional: true, requiresNew: true))
                                 {
                                     Logger.LogInformation($"Discarding event with id = {waitingEvent.Id:N} due to an error.");
+
                                     await Inbox.MarkAsDiscardAsync(waitingEvent.Id);
                                     await uow.CompleteAsync(StoppingToken);
                                 }

framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/Distributed/InboxProcessorOptions.cs

@@ -4,5 +4,12 @@ public class InboxProcessorOptions
 {
     public InboxProcessorFailurePolicy FailurePolicy { get; set; } = InboxProcessorFailurePolicy.Retry;
 
+    /// <summary>
+    /// Retry intervals follow an exponential backoff strategy:
+    /// The delay for each retry is twice the previous one,
+    /// starting with an initial delay of 1 second.
+    /// For example: 1s, 2s, 4s, 8s, 16s, 32s ...
+    /// Maximum of 10 retries.
+    /// </summary>
     public int MaxRetryCount { get; set; } = 10;
 }

framework/src/Volo.Abp.MongoDB/Volo/Abp/MongoDB/DistributedEvents/IncomingEventRecord.cs

@@ -23,9 +23,9 @@ public class IncomingEventRecord :
 
     public DateTime CreationTime { get; private set; }
 
-    public IncomingEventStatus Status { get; set; }
+    public IncomingEventStatus Status { get; set; } = IncomingEventStatus.Pending;
 
-    public DateTime? DiscardedOrProcessedTime { get; set; }
+    public DateTime? HandledTime { get; set; }
 
     public int RetryCount { get; set; } = 0;
 
@@ -33,7 +33,6 @@ public class IncomingEventRecord :
 
     protected IncomingEventRecord()
     {
-        Status = IncomingEventStatus.Pending;
         ExtraProperties = new ExtraPropertyDictionary();
         this.SetDefaultsForExtraProperties();
     }
@@ -46,7 +45,10 @@ public class IncomingEventRecord :
         EventName = eventInfo.EventName;
         EventData = eventInfo.EventData;
         CreationTime = eventInfo.CreationTime;
-        Status = IncomingEventStatus.Pending;
+        Status = eventInfo.Status;
+        HandledTime = eventInfo.HandledTime;
+        RetryCount = eventInfo.RetryCount;
+        NextRetryTime = eventInfo.NextRetryTime;
         ExtraProperties = new ExtraPropertyDictionary();
         this.SetDefaultsForExtraProperties();
         foreach (var property in eventInfo.ExtraProperties)
@@ -62,7 +64,11 @@ public class IncomingEventRecord :
             MessageId,
             EventName,
             EventData,
-            CreationTime
+            CreationTime,
+            Status,
+            HandledTime,
+            RetryCount,
+            NextRetryTime
         );
 
         foreach (var property in ExtraProperties)
@@ -76,13 +82,13 @@ public class IncomingEventRecord :
     public void MarkAsProcessed(DateTime processedTime)
     {
         Status = IncomingEventStatus.Processed;
-        DiscardedOrProcessedTime = processedTime;
+        HandledTime = processedTime;
     }
 
     public void MarkAsDiscarded(DateTime discardedTime)
     {
         Status = IncomingEventStatus.Discarded;
-        DiscardedOrProcessedTime = discardedTime;
+        HandledTime = discardedTime;
     }
 
     public void RetryLater(int retryCount, DateTime nextRetryTime)

framework/src/Volo.Abp.MongoDB/Volo/Abp/MongoDB/DistributedEvents/MongoDbContextEventInbox.cs

@@ -81,7 +81,7 @@ public class MongoDbContextEventInbox<TMongoDbContext> : IMongoDbContextEventInb
         var dbContext = await DbContextProvider.GetDbContextAsync();
 
         var filter = Builders<IncomingEventRecord>.Filter.Eq(x => x.Id, id);
-        var update = Builders<IncomingEventRecord>.Update.Set(x => x.Status == IncomingEventStatus.Processed, true).Set(x => x.DiscardedOrProcessedTime, Clock.Now);
+        var update = Builders<IncomingEventRecord>.Update.Set(x => x.Status, IncomingEventStatus.Processed).Set(x => x.HandledTime, Clock.Now);
 
         if (dbContext.SessionHandle != null)
         {
@@ -117,7 +117,7 @@ public class MongoDbContextEventInbox<TMongoDbContext> : IMongoDbContextEventInb
         var dbContext = await DbContextProvider.GetDbContextAsync();
 
         var filter = Builders<IncomingEventRecord>.Filter.Eq(x => x.Id, id);
-        var update = Builders<IncomingEventRecord>.Update.Set(x => x.Status == IncomingEventStatus.Discarded, true).Set(x => x.DiscardedOrProcessedTime, Clock.Now);
+        var update = Builders<IncomingEventRecord>.Update.Set(x => x.Status, IncomingEventStatus.Discarded).Set(x => x.HandledTime, Clock.Now);
 
         if (dbContext.SessionHandle != null)
         {