diff --git a/aws_lambda_powertools/utilities/circuit_breaker/base.py b/aws_lambda_powertools/utilities/circuit_breaker/base.py index 7e476828491..3458a67aae8 100644 --- a/aws_lambda_powertools/utilities/circuit_breaker/base.py +++ b/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: diff --git a/tests/functional/circuit_breaker/test_circuit_breaker.py b/tests/functional/circuit_breaker/test_circuit_breaker.py index 1db372ee900..75b39fd5c9f 100644 --- a/tests/functional/circuit_breaker/test_circuit_breaker.py +++ b/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)