fix(circuit_breaker): make probe election per-thread and synchronize local counters (#8323)

* fix(circuit_breaker): make probe election per-thread and synchronize local counters

The half-open probe owner id was minted once per process, so every thread
in a process passed the owner check once any sibling won the election and
all of them probed the recovering backend. The id is now a per-thread
uuid (pid-aware for forked workers); the existing DynamoDB conditional
write then elects one prober across threads and processes with no
protocol change.

The in-memory failure/success counters and observed-state map are now
guarded by a lock, with threshold crossings detected atomically so a
trip is persisted exactly once. Persistence settings are keyed per
circuit name instead of living as shared instance attributes, and the
local cache no longer raises into the protected call on concurrent
expiry.

* test(circuit_breaker): cover losing the expired-lease probe takeover election

The takeover's failure arm was untested: when another environment wins
the conditional election, the caller must get the open-circuit response
and the protected function must not run.

* fix(circuit_breaker): address review feedback

- Make the counter race test catch a missing lock: park readers mid-increment so an unsynchronized read-modify-write deterministically loses updates (fails 10/10 unlocked, passes 10/10 locked)
- Document that on_circuit_open and on_transition callbacks can run concurrently and must be thread-safe
- Use dict.get's default instead of `or` for the per-circuit settings lookup

* fix(circuit_breaker): avoid LRUDict.pop when evicting expired cache entries

- On Python 3.10, OrderedDict.pop re-enters the subclass __getitem__ after
  detaching the linked-list node, so LRUDict.pop raises KeyError for a present
  key and corrupts the dict (fixed in CPython 3.11) — this failed 14 tests on
  the 3.10 CI job only
- Evict with a guarded del instead, which never calls subclass hooks; the
  surrounding lock already makes the get-then-delete atomic, and the
  try/except keeps the never-raise-into-the-protected-call contract explicit

Author: Iamrodos

aws_lambda_powertools/utilities/circuit_breaker_alpha/base.py

@@ -10,6 +10,8 @@
 
 import datetime
 import logging
+import os
+import threading
 import uuid
 from typing import TYPE_CHECKING, Any
 
@@ -37,8 +39,31 @@
 # recovered), so stale local failure streaks can be invalidated.
 _LAST_OBSERVED_STATE: dict[str, CircuitState] = {}
 
-# Stable per-environment identifier used to claim the half-open probe lock.
-_ENVIRONMENT_ID = uuid.uuid4().hex
+# Guards the three dicts above. Increments are read-modify-write and a threshold
+# crossing must be observed by exactly one thread, so every access goes through this
+# lock. Held only while mutating the dicts, never across persistence writes or user
+# callbacks.
+_COUNTERS_LOCK = threading.Lock()
+
+# Identifier used to claim the half-open probe lock, unique per thread so the store's
+# conditional election picks a single prober across threads as well as processes.
+_PROBE_OWNER = threading.local()
+
+
+def _probe_owner_id() -> str:
+    """
+    Return this thread's stable probe-owner identifier, minting it on first use.
+
+    A uuid in thread-local storage rather than ``threading.get_ident()``: the OS reuses
+    thread ids, and a recycled id would let an unrelated thread pass the owner check and
+    probe alongside the real owner. The pid check re-mints the id in forked children,
+    which inherit the forking thread's local storage.
+    """
+    pid = os.getpid()
+    if getattr(_PROBE_OWNER, "pid", None) != pid:
+        _PROBE_OWNER.id = f"{pid}#{uuid.uuid4().hex}"
+        _PROBE_OWNER.pid = pid
+    return _PROBE_OWNER.id
 
 
 class CircuitBreakerHandler:
@@ -111,32 +136,35 @@ def handle(self) -> Any:
             # If we previously observed a non-CLOSED state and the circuit is now back to
             # CLOSED, another environment completed the recovery cycle. Reset local counters
             # so a stale partial failure streak doesn't immediately re-trip the circuit.
-            prev = _LAST_OBSERVED_STATE.get(self.name)
-            if prev is not None and prev != CircuitState.CLOSED:
-                _LOCAL_FAILURES[self.name] = 0
-            _LAST_OBSERVED_STATE[self.name] = CircuitState.CLOSED
+            with _COUNTERS_LOCK:
+                prev = _LAST_OBSERVED_STATE.get(self.name)
+                if prev is not None and prev != CircuitState.CLOSED:
+                    _LOCAL_FAILURES[self.name] = 0
+                _LAST_OBSERVED_STATE[self.name] = CircuitState.CLOSED
             return self._call_closed()
 
         if record.state == CircuitState.OPEN:
-            _LAST_OBSERVED_STATE[self.name] = CircuitState.OPEN
+            with _COUNTERS_LOCK:
+                _LAST_OBSERVED_STATE[self.name] = CircuitState.OPEN
             # ``opened_at`` may legitimately be 0 (epoch); treat only None as missing.
             opened_at = record.opened_at if record.opened_at is not None else self._now()
             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, _ENVIRONMENT_ID, opened_at):
+                if self.persistence_store.try_acquire_half_open(self.name, _probe_owner_id(), opened_at):
                     self._notify(CircuitState.OPEN, CircuitState.HALF_OPEN, opened_at=opened_at)
                     return self._call_probe()
             return self._open_response(record.to_circuit_info())
 
-        # HALF_OPEN: only the environment that owns the probe lock runs.
-        _LAST_OBSERVED_STATE[self.name] = CircuitState.HALF_OPEN
-        if record.half_open_owner == _ENVIRONMENT_ID:
+        # HALF_OPEN: only the thread that owns the probe lock runs.
+        with _COUNTERS_LOCK:
+            _LAST_OBSERVED_STATE[self.name] = CircuitState.HALF_OPEN
+        if record.half_open_owner == _probe_owner_id():
             return self._call_probe()
 
         # If the probe lease has expired (owner recycled mid-probe), take over.
         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, _ENVIRONMENT_ID, record.opened_at or 0):
+            if self.persistence_store.try_acquire_half_open(self.name, _probe_owner_id(), record.opened_at or 0):
                 return self._call_probe()
 
         return self._open_response(record.to_circuit_info())
@@ -148,9 +176,14 @@ def _call_closed(self) -> Any:
         except Exception as exc:
             if not self.config.counts_as_failure(exc):
                 raise
-            failures = _LOCAL_FAILURES.get(self.name, 0) + 1
-            _LOCAL_FAILURES[self.name] = failures
-            if failures >= self.config.failure_threshold:
+            # Increment and reset atomically so exactly one thread observes the threshold
+            # crossing; racing threads would otherwise lose increments (tripping late) or
+            # each persist the same transition.
+            with _COUNTERS_LOCK:
+                failures = _LOCAL_FAILURES.get(self.name, 0) + 1
+                tripped = failures >= self.config.failure_threshold
+                _LOCAL_FAILURES[self.name] = 0 if tripped else failures
+            if tripped:
                 logger.debug("Circuit '%s' tripping CLOSED to OPEN after %d failures.", self.name, failures)
                 opened_at = self._now()
                 self._safe_persist(
@@ -159,11 +192,11 @@ def _call_closed(self) -> Any:
                     failure_count=failures,
                     opened_at=opened_at,
                 )
-                _LOCAL_FAILURES[self.name] = 0
                 self._notify(CircuitState.CLOSED, CircuitState.OPEN, opened_at=opened_at)
             raise
         else:
-            _LOCAL_FAILURES[self.name] = 0
+            with _COUNTERS_LOCK:
+                _LOCAL_FAILURES[self.name] = 0
             return result
 
     def _call_probe(self) -> Any:
@@ -176,17 +209,20 @@ def _call_probe(self) -> Any:
             logger.debug("Circuit '%s' probe failed; reopening.", self.name)
             opened_at = self._now()
             self._safe_persist(self.persistence_store.save_reopen, self.name, opened_at=opened_at)
-            _LOCAL_SUCCESSES[self.name] = 0
+            with _COUNTERS_LOCK:
+                _LOCAL_SUCCESSES[self.name] = 0
             self._notify(CircuitState.HALF_OPEN, CircuitState.OPEN, opened_at=opened_at)
             raise
         else:
-            successes = _LOCAL_SUCCESSES.get(self.name, 0) + 1
-            _LOCAL_SUCCESSES[self.name] = successes
-            if successes >= self.config.success_threshold:
+            with _COUNTERS_LOCK:
+                successes = _LOCAL_SUCCESSES.get(self.name, 0) + 1
+                closed = successes >= self.config.success_threshold
+                _LOCAL_SUCCESSES[self.name] = 0 if closed else successes
+                if closed:
+                    _LOCAL_FAILURES[self.name] = 0
+            if closed:
                 logger.debug("Circuit '%s' closing after %d probe successes.", self.name, successes)
                 self._safe_persist(self.persistence_store.save_closed, self.name)
-                _LOCAL_SUCCESSES[self.name] = 0
-                _LOCAL_FAILURES[self.name] = 0
                 self._notify(CircuitState.HALF_OPEN, CircuitState.CLOSED)
             return result
 

aws_lambda_powertools/utilities/circuit_breaker_alpha/persistence/base.py

@@ -10,8 +10,9 @@
 
 import datetime
 import logging
+import threading
 from abc import ABC, abstractmethod
-from typing import TYPE_CHECKING
+from typing import TYPE_CHECKING, NamedTuple
 
 from aws_lambda_powertools.shared.cache_dict import LRUDict
 from aws_lambda_powertools.utilities.circuit_breaker_alpha.persistence.record import CircuitStateRecord
@@ -32,6 +33,17 @@
 PERSISTED_STATE_TTL_BUFFER = 3600
 
 
+class _CircuitSettings(NamedTuple):
+    """Per-circuit tunables the layer captures from :meth:`configure`."""
+
+    local_cache_max_age: int
+    recovery_timeout: int
+
+
+# Fallback for direct layer use before configure() has run; mirrors CircuitBreakerConfig defaults.
+_DEFAULT_SETTINGS = _CircuitSettings(local_cache_max_age=5, recovery_timeout=30)
+
+
 class CircuitBreakerExistingLockError(Exception):
     """Internal signal that a conditional half-open probe write lost the race."""
 
@@ -50,63 +62,86 @@ class CircuitBreakerPersistenceLayer(ABC):
 
     def __init__(self) -> None:
         """Initialize defaults; real configuration happens in :meth:`configure`."""
-        self.circuit_name: str = ""
-        self.local_cache_max_age: int = 5
-        self.recovery_timeout: int = 30
+        # Per-circuit tunables, keyed by circuit name. One persistence instance is shared
+        # by every circuit (and thread) using the same store, so these must never live in
+        # plain instance attributes: circuits with different configs would stamp each
+        # other's TTL and probe lease. A plain dict, not an LRUDict: evicting a live
+        # circuit's settings would silently swap in the defaults (wrong lease and TTL),
+        # whereas evicting a cache entry below only costs a store re-read.
+        self._settings: dict[str, _CircuitSettings] = {}
         # Maps circuit name -> the unix timestamp the locally cached record goes stale.
         # Kept separate from the record's durable ``expiry_timestamp`` (the store TTL) so
         # the short in-memory freshness window is never mistaken for the long store TTL.
         self._cache: LRUDict = LRUDict(max_items=LOCAL_CACHE_MAX_ITEMS)
+        # One lock for both maps: LRUDict reorders entries even on reads, so unguarded
+        # concurrent access can corrupt it or raise.
+        self._lock = threading.Lock()
 
     def configure(self, config: CircuitBreakerConfig, circuit_name: str) -> None:
         """
-        Bind the layer to a circuit and its configuration.
+        Bind a circuit's configuration to the layer.
 
-        Called once per invocation by the handler; the assignments are cheap and the
+        Called once per invocation by the handler; the assignment is cheap and the
         same persistence instance is reused across invocations within an environment.
 
         Parameters
         ----------
         config : CircuitBreakerConfig
             Configuration providing the local cache TTL and recovery timeout.
         circuit_name : str
-            The circuit this layer instance serves.
+            The circuit these settings apply to.
         """
-        self.circuit_name = circuit_name
-        self.local_cache_max_age = config.local_cache_max_age
-        self.recovery_timeout = config.recovery_timeout
+        with self._lock:
+            self._settings[circuit_name] = _CircuitSettings(
+                local_cache_max_age=config.local_cache_max_age,
+                recovery_timeout=config.recovery_timeout,
+            )
+
+    def _settings_for(self, name: str) -> _CircuitSettings:
+        """Return a circuit's configured settings, or the defaults if never configured."""
+        with self._lock:
+            return self._settings.get(name, _DEFAULT_SETTINGS)
 
     # ------------------------------------------------------------------ cache
 
     def _cache_key(self, name: str) -> str:
         return name
 
-    def _durable_ttl(self) -> int:
+    def _durable_ttl(self, name: str) -> int:
         """
         Compute the store TTL stamped on a persisted record.
 
         Sized to outlive a full recovery window so a live circuit is never reaped
         mid-cycle, while an abandoned circuit (no further writes) self-cleans soon after.
         """
-        return int(datetime.datetime.now().timestamp()) + self.recovery_timeout + PERSISTED_STATE_TTL_BUFFER
+        now = int(datetime.datetime.now().timestamp())
+        return now + self._settings_for(name).recovery_timeout + PERSISTED_STATE_TTL_BUFFER
 
     def _save_to_cache(self, record: CircuitStateRecord) -> None:
         """Cache a record locally with a short in-memory freshness window."""
-        local_expiry = int(datetime.datetime.now().timestamp()) + self.local_cache_max_age
-        self._cache[self._cache_key(record.name)] = (local_expiry, record)
+        local_expiry = int(datetime.datetime.now().timestamp()) + self._settings_for(record.name).local_cache_max_age
+        with self._lock:
+            self._cache[self._cache_key(record.name)] = (local_expiry, record)
 
     def _retrieve_from_cache(self, name: str) -> CircuitStateRecord | None:
         """Return a cached record if present and still within its local freshness window."""
-        cached = self._cache.get(self._cache_key(name))
-        if cached is None:
-            return None
-
-        local_expiry, record = cached
-        if int(datetime.datetime.now().timestamp()) >= local_expiry:
-            del self._cache[self._cache_key(name)]
-            return None
-
-        return record
+        with self._lock:
+            cached = self._cache.get(self._cache_key(name))
+            if cached is None:
+                return None
+
+            local_expiry, record = cached
+            if int(datetime.datetime.now().timestamp()) >= local_expiry:
+                # Guarded del, not pop: on Python 3.10 OrderedDict.pop re-enters the
+                # subclass __getitem__ after detaching the node, so LRUDict.pop raises
+                # KeyError for a *present* key and corrupts the dict (fixed in 3.11).
+                try:
+                    del self._cache[self._cache_key(name)]
+                except KeyError:
+                    pass
+                return None
+
+            return record
 
     # ------------------------------------------------------------- public API
 
@@ -175,45 +210,46 @@ def save_open(self, name: str, failure_count: int, opened_at: int) -> None:
             state=CircuitState.OPEN,
             failure_count=failure_count,
             opened_at=opened_at,
-            expiry_timestamp=self._durable_ttl(),
+            expiry_timestamp=self._durable_ttl(name),
         )
         self._put_record(record)
         self._save_to_cache(record)
 
     def try_acquire_half_open(self, name: str, owner: str, opened_at: int) -> bool:
         """
-        Atomically elect a single environment to run the half-open probe.
+        Atomically elect a single worker to run the half-open probe.
 
         The conditional write succeeds only when the circuit is OPEN with no existing
         lock owner AND the ``opened_at`` matches what the caller observed (guards against
         stale eventually-consistent reads). A lease expiry is stamped so that if the
-        winning environment is recycled before completing the probe, others can take over
+        winning worker is recycled before completing the probe, others can take over
         once the lease lapses.
 
         Parameters
         ----------
         name : str
             Circuit name.
         owner : str
-            Identifier of the environment attempting the probe.
+            Identifier of the worker (one thread in one execution environment)
+            attempting the probe.
         opened_at : int
             The ``opened_at`` the caller observed, kept stable across the transition.
 
         Returns
         -------
         bool
-            ``True`` if this environment won the probe lock, ``False`` if another
-            environment already holds it.
+            ``True`` if this worker won the probe lock, ``False`` if another
+            worker already holds it.
         """
         # Lease = recovery_timeout gives the probe a full cycle to complete.
-        probe_lease_expiry = int(datetime.datetime.now().timestamp()) + self.recovery_timeout
+        probe_lease_expiry = int(datetime.datetime.now().timestamp()) + self._settings_for(name).recovery_timeout
         record = CircuitStateRecord(
             name=name,
             state=CircuitState.HALF_OPEN,
             opened_at=opened_at,
             half_open_owner=owner,
             probe_lease_expiry=probe_lease_expiry,
-            expiry_timestamp=self._durable_ttl(),
+            expiry_timestamp=self._durable_ttl(name),
         )
         try:
             self._put_record(record, condition="half_open", expected_opened_at=opened_at)
@@ -228,7 +264,7 @@ def save_closed(self, name: str) -> None:
             name=name,
             state=CircuitState.CLOSED,
             failure_count=0,
-            expiry_timestamp=self._durable_ttl(),
+            expiry_timestamp=self._durable_ttl(name),
         )
         self._update_record(record)
         self._save_to_cache(record)
@@ -245,7 +281,7 @@ def save_reopen(self, name: str, opened_at: int) -> None:
             name=name,
             state=CircuitState.OPEN,
             opened_at=opened_at,
-            expiry_timestamp=self._durable_ttl(),
+            expiry_timestamp=self._durable_ttl(name),
         )
         self._update_record(record)
         self._save_to_cache(record)

aws_lambda_powertools/utilities/circuit_breaker_alpha/persistence/record.py

@@ -30,7 +30,8 @@ class CircuitStateRecord:
         Unix timestamp (seconds) the circuit opened. Anchors the recovery timeout;
         ``None`` while closed.
     half_open_owner : str | None
-        Identifier of the execution environment that won the half-open probe lock, if any.
+        Identifier of the worker (one thread in one execution environment) that won the
+        half-open probe lock, if any.
     expiry_timestamp : int | None
         Unix timestamp (seconds) for the store's TTL attribute.
     """

docs/utilities/circuit_breaker.md

@@ -20,6 +20,7 @@ The circuit breaker utility stops sending traffic to an unhealthy downstream dep
 * Hands rejected requests to an `on_circuit_open` callback so you decide what happens next (buffer, drop, return a cached value)
 * Tests recovery with an explicit half-open probe rather than blindly retrying everything at once
 * Shares circuit state across execution environments via Amazon DynamoDB
+* Safe under concurrency: one half-open probe across all threads and execution environments, and synchronized failure counting
 * Keeps the healthy path write-free: failures are counted in memory and only persisted on a state transition
 
 ## Terminology
@@ -182,6 +183,19 @@ Passing both raises `CircuitBreakerConfigError`. An exception that doesn't count
 
 After `recovery_timeout` seconds, the circuit moves to `HALF_OPEN` and elects a **single** execution environment (via a conditional DynamoDB write) to run a probe. If `success_threshold` consecutive probes succeed, the circuit closes; a single failing probe reopens it. This stops a thundering herd of every environment hammering a recovering backend at once.
 
+!!! note "Thread safety"
+    The utility is safe to share across threads: within a multi-threaded environment the probe election picks a single
+    thread, so the single-prober guarantee spans threads as well as environments, and the in-memory failure counter is
+    synchronized. Single-threaded functions (the normal Lambda model) are unaffected.
+
+    Probe ownership belongs to the thread that won the election. If that thread never runs the circuit again (for
+    example, a thread-per-request worker pool), recovery waits for the probe lease to expire before another thread or
+    environment takes over.
+
+!!! warning "Make your hooks thread-safe"
+    If your function runs multiple threads, `on_circuit_open` and `on_transition` callbacks can run concurrently
+    for the same circuit. Make them thread-safe.
+
 ### State coordination across environments
 
 The consecutive-failure counter lives in memory per execution environment, so a healthy circuit performs **no writes**. Only when an environment reaches `failure_threshold` does it persist `OPEN`. The shared state is cached locally for `local_cache_max_age` seconds to avoid a read per invocation. A cache miss (cold start or expired entry) forces a read-through before routing.