diff --git a/framework/src/Volo.Abp.EntityFrameworkCore/Volo/Abp/EntityFrameworkCore/DistributedEvents/DbContextEventInbox.cs b/framework/src/Volo.Abp.EntityFrameworkCore/Volo/Abp/EntityFrameworkCore/DistributedEvents/DbContextEventInbox.cs index 9a5ac3f154..856b8612be 100644 --- a/framework/src/Volo.Abp.EntityFrameworkCore/Volo/Abp/EntityFrameworkCore/DistributedEvents/DbContextEventInbox.cs +++ b/framework/src/Volo.Abp.EntityFrameworkCore/Volo/Abp/EntityFrameworkCore/DistributedEvents/DbContextEventInbox.cs @@ -66,7 +66,7 @@ public class DbContextEventInbox : IDbContextEventInbox { 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 : IDbContextEventInbox { 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] diff --git a/framework/src/Volo.Abp.EntityFrameworkCore/Volo/Abp/EntityFrameworkCore/DistributedEvents/IncomingEventRecord.cs b/framework/src/Volo.Abp.EntityFrameworkCore/Volo/Abp/EntityFrameworkCore/DistributedEvents/IncomingEventRecord.cs index 4f01dfbe37..3f06bf4ef9 100644 --- a/framework/src/Volo.Abp.EntityFrameworkCore/Volo/Abp/EntityFrameworkCore/DistributedEvents/IncomingEventRecord.cs +++ b/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) diff --git a/framework/src/Volo.Abp.EventBus.Abstractions/Volo/Abp/EventBus/Distributed/IncomingEventInfo.cs b/framework/src/Volo.Abp.EventBus.Abstractions/Volo/Abp/EventBus/Distributed/IncomingEventInfo.cs index 855d4a7302..23e59ad61a 100644 --- a/framework/src/Volo.Abp.EventBus.Abstractions/Volo/Abp/EventBus/Distributed/IncomingEventInfo.cs +++ b/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(); } diff --git a/framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/Distributed/InboxProcessor.cs b/framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/Distributed/InboxProcessor.cs index bbd203011a..88f9e18a29 100644 --- a/framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/Distributed/InboxProcessor.cs +++ b/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); } diff --git a/framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/Distributed/InboxProcessorOptions.cs b/framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/Distributed/InboxProcessorOptions.cs index f3972fd1b3..7767945208 100644 --- a/framework/src/Volo.Abp.EventBus/Volo/Abp/EventBus/Distributed/InboxProcessorOptions.cs +++ b/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; + /// + /// 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. + /// public int MaxRetryCount { get; set; } = 10; } diff --git a/framework/src/Volo.Abp.MongoDB/Volo/Abp/MongoDB/DistributedEvents/IncomingEventRecord.cs b/framework/src/Volo.Abp.MongoDB/Volo/Abp/MongoDB/DistributedEvents/IncomingEventRecord.cs index 9ef7c87167..0a3a1d4dd2 100644 --- a/framework/src/Volo.Abp.MongoDB/Volo/Abp/MongoDB/DistributedEvents/IncomingEventRecord.cs +++ b/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) diff --git a/framework/src/Volo.Abp.MongoDB/Volo/Abp/MongoDB/DistributedEvents/MongoDbContextEventInbox.cs b/framework/src/Volo.Abp.MongoDB/Volo/Abp/MongoDB/DistributedEvents/MongoDbContextEventInbox.cs index 1925f8280a..112f6d7cf7 100644 --- a/framework/src/Volo.Abp.MongoDB/Volo/Abp/MongoDB/DistributedEvents/MongoDbContextEventInbox.cs +++ b/framework/src/Volo.Abp.MongoDB/Volo/Abp/MongoDB/DistributedEvents/MongoDbContextEventInbox.cs @@ -81,7 +81,7 @@ public class MongoDbContextEventInbox : IMongoDbContextEventInb var dbContext = await DbContextProvider.GetDbContextAsync(); var filter = Builders.Filter.Eq(x => x.Id, id); - var update = Builders.Update.Set(x => x.Status == IncomingEventStatus.Processed, true).Set(x => x.DiscardedOrProcessedTime, Clock.Now); + var update = Builders.Update.Set(x => x.Status, IncomingEventStatus.Processed).Set(x => x.HandledTime, Clock.Now); if (dbContext.SessionHandle != null) { @@ -117,7 +117,7 @@ public class MongoDbContextEventInbox : IMongoDbContextEventInb var dbContext = await DbContextProvider.GetDbContextAsync(); var filter = Builders.Filter.Eq(x => x.Id, id); - var update = Builders.Update.Set(x => x.Status == IncomingEventStatus.Discarded, true).Set(x => x.DiscardedOrProcessedTime, Clock.Now); + var update = Builders.Update.Set(x => x.Status, IncomingEventStatus.Discarded).Set(x => x.HandledTime, Clock.Now); if (dbContext.SessionHandle != null) {