[automatic failover] CAE-1861: Atomic lock-free metrics reset on CircuitBreaker state transitions (#3527)

* abstract clock for easy testing

* Improve LockFreeSlidingWindowMetrics: fix bugs and add tests

Bug Fixes:
- Fix: Ensure snapshot metrics remain accurate after a full window rotation
- Fix: events recorded exactly at bucket boundaries were miscounted
- Enforce window size % bucket size == 0
- Move LockFreeSlidingWindowMetricsUnitTests to correct package
  (io.lettuce.core.failover.metrics)

* remove unused reset methods

* extract interface for MetricsSnapshot

   - remove snapshotTime - not used & not correctly calcualted
   - remove reset metrics - unused as of now

* add LockFreeSlidingWindowMetrics benchmark test

* performance tests moved to metrics package

* replace with port from reselience4j

* update copyrights

* format

* clean up javadocs

* clean up
   - fix incorrect javadoc
   - fix failing benchmark

* [automatic failover] Hide failover metrics implementation

 - CircuitBreakerMetrics, MetricsSnapshot - public
 - metrics implementation details stay inside io.lettuce.core.failover.metrics
 - Update CircuitBreaker to obtain its metrics via CircuitBreakerMetricsFactory.createLockFree()

* Apply suggestions from code review

Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>

* rename createLockFree -> createDefaultMetrics

* address review comments by @atakavci

    - remove CircuitBreakerMetrics, CircuitBreakerMetricsImpl
    - rename SlidingWindowMetrics -> CircuitBreakerMetrics

* format

* Enforce min-window size of 2 buckets

  Current implementation requires at least 2 buckets window
        With windowSize=1, only one node is created with next=null
        When updateWindow() advances the window it sets HEAD to headNext, which is null for a single-node window
        On the next call to updateWindow(), tries to access head.next but head is now null, causing:
        NullPointerException: Cannot read field "next" because "head" is null

* Clean-up benchmark

   - benchmark matrix
       threads (1,4)
       window_size ("2", "30", "180")
   - performs 1_000_000 ops in simulated 5min test window
   - benchmark record events
   - benchmark record & read snapshot

* remove MetricsPerformanceTests.java

  - no reliable way to assert on performance, instead added basic benchmark test to benchmark  recording/snapshot reading average times
 - gc benchmarks are available for local testing

* reset method removed

* reset circuit breaker metrics on state transition

* fix test : shouldMaintainMetricsAfterSwitch()

CB metrics are updated async on command completion, meaning waiting on command completion threads might proceed before metrics snapshot is updated.

* format

* evaluateMetrics - javadocs & make it package private

* format

---------

Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>

Author: ggivo

src/main/java/io/lettuce/core/failover/CircuitBreaker.java

@@ -8,7 +8,7 @@
 import java.util.Set;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.TimeoutException;
-import java.util.function.Predicate;
+import java.util.concurrent.atomic.AtomicReference;
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -23,6 +23,9 @@
 /**
  * Circuit breaker for tracking command metrics and managing circuit breaker state. Wraps CircuitBreakerMetrics and exposes it
  * via {@link #getMetrics()}.
+ * <p>
+ * State transitions and metrics replacement are atomic and lock-free using {@link AtomicReference}. When the circuit breaker
+ * transitions to a new state, a fresh metrics instance is created atomically.
  *
  * @author Ali Takavci
  * @since 7.1
@@ -31,11 +34,9 @@ public class CircuitBreaker implements Closeable {
 
     private static final Logger log = LoggerFactory.getLogger(CircuitBreaker.class);
 
-    private final CircuitBreakerMetrics metrics;
-
     private final CircuitBreakerConfig config;
 
-    private volatile State currentState = State.CLOSED;
+    private final AtomicReference<CircuitBreakerStateHolder> stateRef;
 
     private final Set<CircuitBreakerStateListener> listeners = ConcurrentHashMap.newKeySet();
 
@@ -45,9 +46,10 @@ public class CircuitBreaker implements Closeable {
      * Create a circuit breaker instance.
      */
     public CircuitBreaker(CircuitBreakerConfig config) {
-        this.metrics = MetricsFactory.createDefaultMetrics();
         this.config = config;
         this.trackedExceptions = new HashSet<>(config.trackedExceptions);
+        this.stateRef = new AtomicReference<>(
+                new CircuitBreakerStateHolder(State.CLOSED, MetricsFactory.createDefaultMetrics()));
     }
 
     /**
@@ -58,7 +60,7 @@ public CircuitBreaker(CircuitBreakerConfig config) {
      * @return the circuit breaker metrics
      */
     CircuitBreakerMetrics getMetrics() {
-        return metrics;
+        return stateRef.get().metrics;
     }
 
     /**
@@ -68,12 +70,13 @@ CircuitBreakerMetrics getMetrics() {
      * @return an immutable snapshot of current metrics
      */
     public MetricsSnapshot getSnapshot() {
-        return metrics.getSnapshot();
+        return stateRef.get().metrics.getSnapshot();
     }
 
     @Override
     public String toString() {
-        return "CircuitBreaker{" + "metrics=" + metrics + ", config=" + config + '}';
+        CircuitBreakerStateHolder current = stateRef.get();
+        return "CircuitBreaker{" + "state=" + current.state + ", metrics=" + current.metrics + ", config=" + config + '}';
     }
 
     public boolean isCircuitBreakerTrackedException(Throwable throwable) {
@@ -95,16 +98,27 @@ public void recordResult(Throwable error) {
     }
 
     public void recordFailure() {
-        metrics.recordFailure();
+        stateRef.get().metrics.recordFailure();
         evaluateMetrics();
     }
 
     public void recordSuccess() {
-        metrics.recordSuccess();
+        stateRef.get().metrics.recordSuccess();
     }
 
-    public MetricsSnapshot evaluateMetrics() {
-        MetricsSnapshot snapshot = metrics.getSnapshot();
+    /**
+     * Evaluate the current metrics to determine if the circuit breaker should transition to a new state.
+     *
+     * <p>
+     * This method checks the failure rate and failure count against the configured thresholds. If the thresholds are met, the
+     * circuit breaker transitions to the OPEN state. Metrics are reset when the state changes.
+     * </p>
+     *
+     * @return an immutable snapshot of current metrics
+     */
+    MetricsSnapshot evaluateMetrics() {
+        CircuitBreakerStateHolder current = stateRef.get();
+        MetricsSnapshot snapshot = current.metrics.getSnapshot();
         boolean evaluationResult = snapshot.getFailureRate() >= config.getFailureRateThreshold()
                 && snapshot.getFailureCount() >= config.getMinimumNumberOfFailures();
         if (evaluationResult) {
@@ -113,16 +127,56 @@ public MetricsSnapshot evaluateMetrics() {
         return snapshot;
     }
 
+    /**
+     * Switch the circuit breaker to the specified state. This method is used to force the circuit breaker to a specific state.
+     *
+     * <p>
+     * This method does not evaluate the metrics to determine if the state transition is valid. It simply transitions to the
+     * specified state. Metrics are reset when the state changes.
+     * </p>
+     *
+     * @param newState the target state
+     */
+    public void transitionTo(State newState) {
+        stateTransitionTo(newState);
+    }
+
+    /**
+     * Atomically transition to a new state with fresh metrics.
+     * <p>
+     * This method uses lock-free CAS to ensure that state change and metrics reset happen atomically. Whenever the circuit
+     * breaker transitions to a new state, a fresh metrics instance is created, providing a clean slate for tracking metrics in
+     * the new state.
+     * <p>
+     * If the state is already the target state, no transition occurs.
+     *
+     * @param newState the target state
+     */
     private void stateTransitionTo(State newState) {
-        State previousState = this.currentState;
-        if (previousState != newState) {
-            this.currentState = newState;
-            fireStateChanged(previousState, newState);
+        while (true) {
+            CircuitBreakerStateHolder current = stateRef.get();
+
+            // No transition needed if already in target state
+            if (current.state == newState) {
+                return;
+            }
+
+            // Always create fresh metrics on state transition
+            CircuitBreakerMetrics nextMetrics = MetricsFactory.createDefaultMetrics();
+
+            CircuitBreakerStateHolder next = new CircuitBreakerStateHolder(newState, nextMetrics);
+
+            // Atomically swap if current state hasn't changed
+            if (stateRef.compareAndSet(current, next)) {
+                fireStateChanged(current.state, newState);
+                return;
+            }
+            // CAS failed, retry with updated current state
         }
     }
 
     public State getCurrentState() {
-        return currentState;
+        return stateRef.get().state;
     }
 
     /**
@@ -166,10 +220,29 @@ public void close() {
         listeners.clear();
     }
 
-    public static enum State {
+    public enum State {
         CLOSED, OPEN
     }
 
+    /**
+     * Immutable holder for circuit breaker state and metrics.
+     * <p>
+     * This class enables atomic updates of both state and metrics using {@link AtomicReference}. When a state transition
+     * occurs, a new instance is created with fresh metrics.
+     */
+    private static final class CircuitBreakerStateHolder {
+
+        final State state;
+
+        final CircuitBreakerMetrics metrics;
+
+        CircuitBreakerStateHolder(State state, CircuitBreakerMetrics metrics) {
+            this.state = state;
+            this.metrics = metrics;
+        }
+
+    }
+
     public static class CircuitBreakerConfig {
 
         private final static float DEFAULT_FAILURE_RATE_THRESHOLD = 10;

src/main/java/io/lettuce/core/failover/metrics/MetricsSnapshotImpl.java

@@ -100,4 +100,21 @@ public String toString() {
                 + ", failureRate=" + String.format("%.2f", getFailureRate()) + "%" + '}';
     }
 
+    @Override
+    public boolean equals(Object o) {
+        if (this == o) {
+            return true;
+        }
+        if (o == null || getClass() != o.getClass()) {
+            return false;
+        }
+
+        MetricsSnapshotImpl that = (MetricsSnapshotImpl) o;
+
+        if (successCount != that.successCount) {
+            return false;
+        }
+        return failureCount == that.failureCount;
+    }
+
 }

src/test/java/io/lettuce/core/failover/CircuitBreakerMetricsIntegrationTests.java

@@ -2,6 +2,7 @@
 
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.awaitility.Awaitility.await;
 import static org.junit.jupiter.api.Assertions.assertNotNull;
 
 import java.util.List;
@@ -10,6 +11,7 @@
 
 import javax.inject.Inject;
 
+import io.lettuce.core.failover.metrics.MetricsSnapshot;
 import org.junit.jupiter.api.Tag;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.extension.ExtendWith;
@@ -111,34 +113,54 @@ void shouldThrowExceptionForUnknownEndpoint() {
 
     @Test
     void shouldMaintainMetricsAfterSwitch() {
+        // Given: Connection with multiple endpoints
         StatefulRedisMultiDbConnection<String, String> connection = multiDbClient.connect();
         RedisURI firstEndpoint = connection.getCurrentEndpoint();
 
-        // Execute command on first endpoint
-        connection.sync().set("key1", "value1");
-        CircuitBreaker cb1Before = connection.getCircuitBreaker(firstEndpoint);
-        long successes1Before = cb1Before.getSnapshot().getSuccessCount();
+        // When: Record successful command on first endpoint
+        MetricsSnapshot metricsBefore = recordSuccessfulCommand(connection, "key1", "value1");
 
-        // Switch to second endpoint
+        // When: Switch to second endpoint
         List<RedisURI> endpoints = StreamSupport.stream(connection.getEndpoints().spliterator(), false)
                 .collect(Collectors.toList());
         RedisURI secondEndpoint = endpoints.stream().filter(uri -> !uri.equals(firstEndpoint)).findFirst()
                 .orElseThrow(() -> new IllegalStateException("No second endpoint found"));
         connection.switchToDatabase(secondEndpoint);
 
-        // Execute command on second endpoint
-        connection.sync().set("key2", "value2");
+        // When: Record successful commands on second endpoint
+        recordSuccessfulCommand(connection, "key2", "value2");
+        recordSuccessfulCommand(connection, "key3", "value3");
 
-        // Switch back to first endpoint
+        // When: Switch back to first endpoint
         connection.switchToDatabase(firstEndpoint);
 
-        // Verify metrics for first endpoint are unchanged
+        // Then: Circuit breaker metrics on first endpoint should be maintained
         CircuitBreaker cb1After = connection.getCircuitBreaker(firstEndpoint);
-        assertThat(cb1After.getSnapshot().getSuccessCount()).isEqualTo(successes1Before);
+        assertThat(cb1After.getSnapshot()).isEqualTo(metricsBefore);
 
         connection.close();
     }
 
+    /**
+     * Helper method to record a successful command and wait for metrics to update.
+     *
+     * <p>
+     * Metrics are updated asynchronously post command completion, we need to wait for the metrics to update before proceeding.
+     * </p>
+     *
+     * @param connection
+     * @param key
+     * @param value
+     * @return final success count
+     */
+    private MetricsSnapshot recordSuccessfulCommand(StatefulRedisMultiDbConnection<String, String> connection, String key,
+            String value) {
+        CircuitBreaker cb = connection.getCircuitBreaker(connection.getCurrentEndpoint());
+        MetricsSnapshot metrics = cb.getSnapshot();
+        connection.sync().set(key, value);
+        return await().until(cb::getSnapshot, snapshot -> snapshot.getSuccessCount() > metrics.getSuccessCount());
+    }
+
     @Test
     void shouldExposeMetricsViaCircuitBreaker() {
         StatefulRedisMultiDbConnection<String, String> connection = multiDbClient.connect();

src/test/java/io/lettuce/core/failover/CircuitBreakerUnitTests.java

@@ -23,6 +23,7 @@
  * Comprehensive unit tests for {@link CircuitBreaker} functionality.
  *
  * @author Ali Takavci
+ * @author Ivo Gaydajiev
  * @since 7.1
  */
 @Tag(UNIT_TEST)
@@ -238,6 +239,50 @@ void shouldNotTransitionWhenAlreadyInTargetState() {
             assertThat(circuitBreaker.getCurrentState()).isEqualTo(CircuitBreaker.State.OPEN);
         }
 
+        @Test
+        @DisplayName("Should reset metrics on forceful state transition")
+        void shouldResetMetricsOnStateTransition() {
+            // Given: some recorded events
+            assertThat(circuitBreaker.getCurrentState()).isEqualTo(CircuitBreaker.State.CLOSED);
+            circuitBreaker.recordSuccess();
+            circuitBreaker.recordFailure();
+
+            // When: force transition to OPEN
+            circuitBreaker.transitionTo(CircuitBreaker.State.OPEN);
+            assertThat(circuitBreaker.getCurrentState()).isEqualTo(CircuitBreaker.State.OPEN);
+
+            // Then: metrics should be reset
+            assertThat(circuitBreaker.getSnapshot().getSuccessCount()).isEqualTo(0);
+            assertThat(circuitBreaker.getSnapshot().getFailureCount()).isEqualTo(0);
+
+            // When: record some more successes and failures
+            circuitBreaker.recordSuccess();
+            circuitBreaker.recordFailure();
+
+            // Then: metrics should reflect the new events
+            assertThat(circuitBreaker.getSnapshot().getSuccessCount()).isEqualTo(1);
+            assertThat(circuitBreaker.getSnapshot().getFailureCount()).isEqualTo(1);
+        }
+
+        @Test
+        @DisplayName("Should reset metrics on automatic state transition")
+        void shouldResetMetricsOnAutomaticStateTransition() {
+            // Given: some recorded events
+            for (int i = 0; i < 4; i++) {
+                circuitBreaker.recordSuccess();
+                circuitBreaker.recordFailure();
+            }
+            assertThat(circuitBreaker.getCurrentState()).isEqualTo(CircuitBreaker.State.CLOSED);
+
+            // When: record 1 more failures to meet the minimumNumberOfFailures threshold
+            circuitBreaker.recordFailure();
+            assertThat(circuitBreaker.getCurrentState()).isEqualTo(CircuitBreaker.State.OPEN);
+
+            // Then: metrics should be reset
+            assertThat(circuitBreaker.getSnapshot().getSuccessCount()).isEqualTo(0);
+            assertThat(circuitBreaker.getSnapshot().getFailureCount()).isEqualTo(0);
+        }
+
     }
 
     @Nested
@@ -452,7 +497,9 @@ void shouldHandleLargeNumberOfOperations() {
 
             // Should open because failure count >= 1000 and failure rate >= 1.0%
             assertThat(circuitBreaker.getCurrentState()).isEqualTo(CircuitBreaker.State.OPEN);
-            assertThat(circuitBreaker.getSnapshot().getFailureCount()).isGreaterThanOrEqualTo(1000);
+            // After state transition, metrics are reset to fresh instance
+            assertThat(circuitBreaker.getSnapshot().getFailureCount()).isEqualTo(0);
+            assertThat(circuitBreaker.getSnapshot().getSuccessCount()).isEqualTo(0);
         }
 
     }