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.

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,81 @@ 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) or _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
+        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:
-            del self._cache[self._cache_key(name)]
-            return None
+            local_expiry, record = cached
+            if int(datetime.datetime.now().timestamp()) >= local_expiry:
+                # pop, not del: this must never raise into the protected call.
+                self._cache.pop(self._cache_key(name), None)
+                return None
 
-        return record
+            return record
 
     # ------------------------------------------------------------- public API
 
@@ -175,45 +205,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 +259,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 +276,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,16 @@ 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. `on_circuit_open` and
+    `on_transition` hooks may run concurrently from multiple threads.
+
+    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.
+
 ### 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.