fix(circuit_breaker): start a fresh probe success streak on each probe election (#8646)

_LOCAL_SUCCESSES is module-level and was only reset when a probe failed
in the same environment or when this environment closed the circuit. A
streak left from an earlier recovery window (this environment probed,
lost its lease, and another environment reopened the circuit) counted
towards the next window, so the circuit could close after fewer than
success_threshold probes.

Reset the streak right after this worker wins try_acquire_half_open, on
both the OPEN path and the expired-lease takeover path. Add regression
tests for both paths.

Closes #8645

Co-authored-by: Leandro Damascena <lcdama@amazon.pt>

Author: unknownhackerworld

aws_lambda_powertools/utilities/circuit_breaker/base.py

@@ -151,6 +151,7 @@ def handle(self) -> Any:
             if self._now() >= opened_at + self.config.recovery_timeout:
                 # Recovery window elapsed: try to become the single prober.
                 if self.persistence_store.try_acquire_half_open(self.name, _probe_owner_id(), opened_at):
+                    self._reset_probe_successes()
                     self._notify(CircuitState.OPEN, CircuitState.HALF_OPEN, opened_at=opened_at)
                     return self._call_probe()
             return self._open_response(record.to_circuit_info())
@@ -165,10 +166,23 @@ def handle(self) -> Any:
         if record.probe_lease_expiry is not None and self._now() >= record.probe_lease_expiry:
             logger.debug("Circuit '%s' probe lease expired; attempting takeover.", self.name)
             if self.persistence_store.try_acquire_half_open(self.name, _probe_owner_id(), record.opened_at or 0):
+                self._reset_probe_successes()
                 return self._call_probe()
 
         return self._open_response(record.to_circuit_info())
 
+    def _reset_probe_successes(self) -> None:
+        """
+        Start a fresh probe success streak after this worker wins a probe election.
+
+        ``_LOCAL_SUCCESSES`` lives at module level, so a streak left over from an earlier
+        recovery window (this environment probed, then lost the lease, or another
+        environment finished that cycle) would otherwise count towards the new one and
+        close the circuit after fewer than ``success_threshold`` probes.
+        """
+        with _COUNTERS_LOCK:
+            _LOCAL_SUCCESSES[self.name] = 0
+
     def _call_closed(self) -> Any:
         """Run the protected call while the circuit is closed, tracking failures."""
         try:

tests/functional/circuit_breaker/test_circuit_breaker.py

@@ -442,6 +442,60 @@ def call():
     assert base_module._LOCAL_FAILURES["c"] == 1  # fresh count, not 3
 
 
+def test_new_probe_election_does_not_reuse_successes_from_an_earlier_recovery_window(store, now):
+    """A probe success streak left from an earlier recovery window must not close a new one early."""
+    config = CircuitBreakerConfig(recovery_timeout=30, success_threshold=3, local_cache_max_age=0)
+    store.db["c"] = CircuitStateRecord(name="c", state=CircuitState.OPEN, failure_count=5, opened_at=now - 60)
+
+    @circuit_breaker(name="c", persistence_store=store, config=config)
+    def call():
+        return "ok"
+
+    # This environment wins the probe election and gets 2 of the 3 successes it needs.
+    call()
+    call()
+    assert store.db["c"].state == CircuitState.HALF_OPEN
+    assert base_module._LOCAL_SUCCESSES["c"] == 2
+
+    # Simulate: this environment goes idle, its probe lease lapses, another environment
+    # takes over, its probe fails and the circuit reopens. That environment cannot reset
+    # this one's module-level counter.
+    store.db["c"] = CircuitStateRecord(name="c", state=CircuitState.OPEN, failure_count=0, opened_at=now - 60)
+
+    # The next recovery window is won by this environment again.
+    call()
+
+    # One success must not close the circuit: the new window needs its own 3 in a row.
+    assert store.db["c"].state == CircuitState.HALF_OPEN
+    assert base_module._LOCAL_SUCCESSES["c"] == 1
+    call()
+    call()
+    assert store.db["c"].state == CircuitState.CLOSED
+
+
+def test_probe_lease_takeover_starts_a_fresh_success_streak(store, now):
+    """Taking over an expired probe lease must not inherit the previous owner's streak."""
+    config = CircuitBreakerConfig(recovery_timeout=30, success_threshold=3, local_cache_max_age=0)
+    store.db["c"] = CircuitStateRecord(
+        name="c",
+        state=CircuitState.HALF_OPEN,
+        opened_at=now - 200,
+        half_open_owner="recycled-owner",
+        probe_lease_expiry=now - 10,  # expired
+    )
+    # The previous owner shared this environment and left 2 successes behind.
+    base_module._LOCAL_SUCCESSES["c"] = 2
+
+    @circuit_breaker(name="c", persistence_store=store, config=config)
+    def call():
+        return "ok"
+
+    call()
+
+    assert store.db["c"].state == CircuitState.HALF_OPEN
+    assert base_module._LOCAL_SUCCESSES["c"] == 1
+
+
 def test_store_write_failure_does_not_mask_downstream_result(store, now):
     """Bug 2: if save_closed() fails, the successful probe result must still be returned."""
     config = CircuitBreakerConfig(failure_threshold=3, recovery_timeout=30, success_threshold=1, local_cache_max_age=0)