Fix/17 circuit braker event loop (#18)

* refactor: improve error handling and logging in OutboxDispatcher

* feat: add CircuitBreakerSettings configuration options

* feat: implement Circuit Breaker pattern for database operation resilience

* feat: enhance Circuit Breaker with logging and configuration updates

* test: update tests to assert no exceptions are thrown during event processing

* refactor: improve documentation and clarity in MissingEventsCoordinator

* chore: update CHANGELOG for v1.5.0 release with new Circuit Breaker features and enhancements

* refactor: update Outbox event keys in configuration for consistency

Author: OguzErdi

CHANGELOG.md

@@ -1,3 +1,17 @@
+## v1.5.0 (March 6, 2025)
+
+### Added:
+- Circuit breaker implementation for handling consecutive database failures
+- New configuration section `CircuitBreakerSettings` with the following options:
+    - `IsEnabled`: Controls circuit breaker functionality
+    - `Threshold`: Maximum number of consecutive failures
+    - `DurationSc`: Duration in seconds for circuit open state
+    - `HalfOpenMaxAttempts`: Maximum attempts in half-open state
+
+### Changed:
+- Enhanced error handling in offset setting operations
+- Improved logging for database operation failures
+
 ## v1.4.0 (December 17, 2024)
 
 ### Added:

PollingOutboxPublisher.sln

@@ -7,7 +7,6 @@ EndProject
 Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "SolutionItems", "SolutionItems", "{BCF067D7-9518-4177-AD14-C854D96ED0F0}"
 ProjectSection(SolutionItems) = preProject
 	Dockerfile = Dockerfile
-	.gitlab-ci.yml = .gitlab-ci.yml
 EndProjectSection
 EndProject
 Global

README.md

@@ -106,6 +106,15 @@ value).
 > If there will be multiple instances, the `MasterPodSettings.IsActive` should be set to `true`. Otherwise, messages can
 > be duplicated or not published.
 
+### Circuit Breaker
+
+The application uses circuit breakers to handle database operation failures gracefully. Each daemon (OutboxEventsDaemon and MissingEventsDaemon) has its own circuit breaker that:
+
+1. Opens after `Threshold` consecutive database failures (default: 3)
+2. Stays open for `DurationMs` (default: 30 seconds)
+3. Allows `HalfOpenMaxAttempts` (default: 1) operations when half-open
+4. Automatically closes if operations succeed in half-open state
+
 ## Configuration
 
 The application can be configured using the `config.json` and `secret.json` files. Here are the configurations you can
@@ -119,7 +128,7 @@ set:
 
 | **Key**                                      | **Type** | **Description**                                                                                                                                                    |
 |----------------------------------------------|----------|--------------------------------------------------------------------------------------------------------------------------------------------------------------------|
-| `Kafka.ReloadOnChange`                         | bool     | The flag indicating whether the Kafka configuration should be reloaded when the configuration file changes.                                                        |
+| `Kafka.ReloadOnChange`                       | bool     | The flag indicating whether the Kafka configuration should be reloaded when the configuration file changes.                                                        |
 | `Kafka.SaslUsername`                         | string   | The username for the SASL authentication of the Kafka cluster.                                                                                                     |
 | `Kafka.Brokers`                              | string   | The addresses of the Kafka brokers.                                                                                                                                |
 | `Kafka.SaslPassword`                         | string   | The password for the SASL authentication of the Kafka cluster.                                                                                                     |
@@ -162,6 +171,10 @@ set:
 | `Redis.Config`                               | string   | The configuration for the Redis instance.                                                                                                                          |
 | `Redis.Password`                             | string   | The password for the Redis instance.                                                                                                                               |
 | `Serilog`                                    | object   | The configuration for Serilog.                                                                                                                                     |
+| `CircuitBreakerSettings.IsEnabled`           | bool     | Enable/disable the circuit breaker feature. When enabled, protects against database operation failures.                                                            |
+| `CircuitBreakerSettings.Threshold`           | int      | Number of consecutive database operation failures before the circuit breaker opens.                                                                                |
+| `CircuitBreakerSettings.DurationSc`          | int      | Duration in seconds to keep the circuit breaker open before attempting to half-open.                                                                               |
+| `CircuitBreakerSettings.HalfOpenMaxAttempts` | int      | Maximum number of attempts allowed when the circuit breaker is in half-open state.                                                                                 |
 
 ## EXAMPLE
 

example/Couchbase/config/config.json

@@ -63,5 +63,11 @@
     "Endpoints": "",
     "DefaultDatabase": 1,
     "Config": ""
+  },
+  "CircuitBreakerSettings": {
+    "IsEnabled": true,
+    "Threshold": 3,
+    "DurationSc": 60,
+    "HalfOpenMaxAttempts": 1
   }
 }

example/MsSql/config/config.json

@@ -57,5 +57,11 @@
     "Endpoints": "",
     "DefaultDatabase": 1,
     "Config": ""
+  },
+  "CircuitBreakerSettings": {
+    "IsEnabled": true,
+    "Threshold": 3,
+    "DurationSc": 60,
+    "HalfOpenMaxAttempts": 1
   }
 }

example/Postgres/config/config.json

@@ -57,5 +57,11 @@
     "Endpoints": "",
     "DefaultDatabase": 1,
     "Config": ""
+  },
+  "CircuitBreakerSettings": {
+    "IsEnabled": true,
+    "Threshold": 3,
+    "DurationSc": 60,
+    "HalfOpenMaxAttempts": 1
   }
 }

src/ConfigOptions/CircuitBreakerSettings.cs

@@ -0,0 +1,12 @@
+using System.Diagnostics.CodeAnalysis;
+
+namespace PollingOutboxPublisher.ConfigOptions;
+
+[ExcludeFromCodeCoverage]
+public class CircuitBreakerSettings
+{
+    public bool IsEnabled { get; set; } = false;
+    public int Threshold { get; set; } = 3;
+    public int DurationSc { get; set; } = 600;
+    public int HalfOpenMaxAttempts { get; set; } = 1;
+} 

src/Coordinators/MissingCoordinator/MissingEventsCoordinator.cs

@@ -4,6 +4,7 @@
 using System.Threading.Tasks;
 using PollingOutboxPublisher.Coordinators.MissingCoordinator.Interfaces;
 using PollingOutboxPublisher.Coordinators.MissingCoordinator.Services.Interfaces;
+using PollingOutboxPublisher.Coordinators.Services;
 using PollingOutboxPublisher.Coordinators.Services.Interfaces;
 using PollingOutboxPublisher.Models;
 
@@ -15,37 +16,46 @@ public class MissingEventsCoordinator : IMissingEventsCoordinator
     private readonly IPollingMissingQueue _pollingMissingQueue;
     private readonly IMissingEventCleaner _missingEventCleaner;
     private readonly IMasterPodChecker _masterPodChecker;
+    private readonly ICircuitBreaker _circuitBreaker;
 
     public MissingEventsCoordinator(IOutboxDispatcher outboxDispatcher, IPollingMissingQueue pollingMissingQueue,
-        IMissingEventCleaner missingEventCleaner, IMasterPodChecker masterPodChecker)
+        IMissingEventCleaner missingEventCleaner, IMasterPodChecker masterPodChecker, ICircuitBreaker circuitBreaker)
     {
         _outboxDispatcher = outboxDispatcher;
         _pollingMissingQueue = pollingMissingQueue;
         _missingEventCleaner = missingEventCleaner;
         _masterPodChecker = masterPodChecker;
+        _circuitBreaker = circuitBreaker;
     }
 
     /**
-     * Because of our outbox algorithm and MsSql async transactions, there may be some times when we process last message which
-     * row id: 500, product-api still may working on the transaction it occurred for id: 450 and not commit it yet. In our EventWorker
-     * when we mark "500" as last id we processed, we are missing 450 which are committed after we processed 500. For this reason we are dedecting
-     * these missing ids while processing them in EventWorker and later recover them here. Since event orders are not important for our domain,
-     * we don't block bulk of messages because there is still missing an event.
-     * Here is the algorithm;
-     * 1. EventWorker detects gap in its batch process, e.g: for events (1,2,5,6) missing events are (3,4)
-     * 2. Missing events are written to MissingOutboxEvents table with the date it is missed
-     * 3. MissingEventsWorker reads missing events from MissingOutboxEvents table with configured batchSize
-     * 4. Filter out events which retry time is exceed. Max retry duration is configurable from application.properties
-     * 5. For each retry time exceed event:
-     *  5.1. Delete events from MissingOutboxEvents table
-     *  5.2. Publish event to earth.product.pim.product-event-publisher.missing-events.0 topic for logging purpose
-     * 6. For each retryable events which retry time is not exceed:
-     *  6.1. Publish event to corresponding outbox topic
-     *  6.2. Delete events from MissingOutboxEvents table
+     * Due to the nature of asynchronous transactions in the outbox pattern, gaps in event sequences can occur.
+     * For example: When processing event ID 500, event ID 450 might still be in an uncommitted transaction.
+     * If we mark ID 500 as processed, we could miss ID 450 when it commits later.
+     *
+     * Since event ordering is not critical for our domain, we handle these gaps asynchronously:
+     *
+     * 1. EventWorker detects sequence gaps during batch processing
+     *    Example: For events [1,2,5,6], missing events are [3,4]
+     *
+     * 2. Missing events are recorded in MissingOutboxEvents table with timestamp
+     *
+     * 3. MissingEventsWorker polls MissingOutboxEvents table using configured batch size
+     *
+     * 4. Events exceeding retry duration (configured in application.properties) are filtered
+     *
+     * 5. For expired events:
+     *    - Remove from MissingOutboxEvents table
+     *    - Log to 'missing-events' topic for auditing
+     *
+     * 6. For the valid event that has a matching outbox event:
+     *    - Publish to original outbox topic
+     *    - Remove from MissingOutboxEvents table
      */
     public async Task StartAsync(CancellationToken cancellationToken)
     {
-        while (!cancellationToken.IsCancellationRequested && await _masterPodChecker.IsMasterPodAsync(cancellationToken))
+        while (!cancellationToken.IsCancellationRequested &&
+               await _masterPodChecker.IsMasterPodAsync(cancellationToken))
         {
             var (retryableMissingEvents, missingEvents, mappedMissingEvents, outboxEvents) =
                 await _pollingMissingQueue.DequeueAsync(cancellationToken);
@@ -62,7 +72,7 @@ public async Task StartAsync(CancellationToken cancellationToken)
                 await _missingEventCleaner.CleanMissingEventsHaveOutboxEventAsync(missingEvents,
                     resultOfMappedMissingEvents.ToList());
             }
-                
+
             // move missing events that don't have match
             if (missingEvents is not null && outboxEvents is not null)
             {
@@ -75,6 +85,9 @@ await _missingEventCleaner.CleanMissingEventsHaveOutboxEventAsync(missingEvents,
             {
                 await _missingEventCleaner.CleanMissingEventsNotHaveOutboxEventAsync(missingEvents);
             }
+
+            // Reset circuit breaker after successful processing
+            _circuitBreaker.Reset();
         }
     }
 

src/Coordinators/OutboxCoordinator/OutboxCoordinator.cs

@@ -3,6 +3,7 @@
 using System.Threading.Tasks;
 using PollingOutboxPublisher.Coordinators.OutboxCoordinator.Interfaces;
 using PollingOutboxPublisher.Coordinators.OutboxCoordinator.Services.Interfaces;
+using PollingOutboxPublisher.Coordinators.Services;
 using PollingOutboxPublisher.Coordinators.Services.Interfaces;
 
 namespace PollingOutboxPublisher.Coordinators.OutboxCoordinator;
@@ -13,14 +14,17 @@ public class OutboxCoordinator : IOutboxCoordinator
     private readonly IPollingQueue _pollingQueue;
     private readonly IOffsetSetter _offsetSetter;
     private readonly IMasterPodChecker _masterPodChecker;
+    private readonly ICircuitBreaker _circuitBreaker;
 
     public OutboxCoordinator(IOutboxDispatcher outboxDispatcher, IPollingQueue pollingQueue,
-        IOffsetSetter offsetSetter, IMasterPodChecker masterPodChecker)
+        IOffsetSetter offsetSetter, IMasterPodChecker masterPodChecker,
+        ICircuitBreaker circuitBreaker)
     {
         _outboxDispatcher = outboxDispatcher;
         _pollingQueue = pollingQueue;
         _offsetSetter = offsetSetter;
         _masterPodChecker = masterPodChecker;
+        _circuitBreaker = circuitBreaker;
     }
 
     public async Task StartAsync(CancellationToken cancellationToken)
@@ -34,6 +38,8 @@ public async Task StartAsync(CancellationToken cancellationToken)
             await Task.WhenAll(taskToAwait);
 
             await _offsetSetter.SetLatestOffset(outboxEvents);
+            
+            _circuitBreaker.Reset();
         }
     }
 }

src/Coordinators/OutboxCoordinator/Services/OffsetSetter.cs

@@ -1,3 +1,4 @@
+using System;
 using System.Diagnostics.CodeAnalysis;
 using System.Linq;
 using System.Threading.Tasks;
@@ -6,6 +7,7 @@
 using PollingOutboxPublisher.Coordinators.OutboxCoordinator.Services.Interfaces;
 using PollingOutboxPublisher.Database.Repositories.Interfaces;
 using PollingOutboxPublisher.Models;
+using PollingOutboxPublisher.Exceptions;
 
 namespace PollingOutboxPublisher.Coordinators.OutboxCoordinator.Services;
 
@@ -24,8 +26,16 @@ public OffsetSetter(ILogger<OffsetSetter> logger, IOutboxOffsetRepository outbox
     [Trace]
     public async Task SetLatestOffset(OutboxEvent[] items)
     {
-        var latestOffSet = items.Max(row => row.Id); // or newestEventId
-        await _outboxOffsetRepository.UpdateOffsetAsync(latestOffSet);
-        _logger.LogInformation("LatestOffSet: {LatestOffSet}", latestOffSet);
+        try
+        {
+            var latestOffSet = items.Max(row => row.Id);
+            await _outboxOffsetRepository.UpdateOffsetAsync(latestOffSet);
+            _logger.LogInformation("LatestOffSet: {LatestOffSet}", latestOffSet);
+        }
+        catch (Exception ex)
+        {
+            _logger.LogError(ex, "Failed to update offset in database");
+            throw new DatabaseOperationException("Failed to update offset", ex);
+        }
     }
 }