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

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)