Merge branch 'master' into add-security-gates

Author: OguzErdi

.github/workflows/scorecard.yml

@@ -47,15 +47,17 @@ jobs:
           results_format: sarif
           publish_results: true
 
-      - name: "Upload artifact"
-        uses: actions/upload-artifact@97a0fba1372883ab732affbe8f94b823f91727db
+     - name: "Upload artifact"
+        uses: actions/upload-artifact@65c4c4a1ddee5b72f698fdd19549f0f0fb45cf08 # v4.6.0
         with:
           name: SARIF file
           path: results.sarif
           retention-days: 5
 
+      # Upload the results to GitHub's code scanning dashboard (optional).
+      # Commenting out will disable upload of results to your repo's Code Scanning dashboard
       - name: "Upload to code-scanning"
-        uses: github/codeql-action/upload-sarif@v3
+        uses: github/codeql-action/upload-sarif@f6091c0113d1dcf9b98e269ee48e8a7e51b7bdd4 # v3.28.5
         with:
           sarif_file: results.sarif
 

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

@@ -108,6 +108,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 `DurationSc` (default: 600 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
@@ -121,7 +130,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.                                                                                                     |
@@ -164,6 +173,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();
         }
     }
 }