diff --git a/CHANGELOG.md b/CHANGELOG.md index c472412de..0f15a0eee 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -199,6 +199,15 @@ Changed `wf10_batch.robot` to be less likely to create files, and `features/architecture.feature` `@tdd_expected_fail` for pre-existing Pydantic compliance debt in `IndexEntry` / `ACMSIndex` classes. +- **InvariantService thread safety (#8209):** Added a ``threading.RLock`` (re-entrant lock) to + ``InvariantService`` that guards all shared mutable state + (`_invariants` dict and `_enforcement_records` list). Every public method acquires the lock + before reading or writing shared data. This prevents + ``RuntimeError: dictionary changed size during iteration`` and data corruption under concurrent + plan execution when the service is registered as a DI Singleton shared across threads. 7 new + Behave scenarios verify lock presence, reentrancy, concurrent add/list/remove/effective-set/enforce, + mixed operations, and singleton shared-instance safety across up to 20 threads. + - **Cross-actor subgraph cycle detection reads actor_ref field** (#1431): Fixed `_detect_subgraph_cycles()`, `_map_node()`, and the `compile_actor()` main loop in `src/cleveragents/actor/compiler.py` to read `actor_ref` from the top-level diff --git a/CONTRIBUTORS.md b/CONTRIBUTORS.md index bf1d924b9..bd16dfa81 100644 --- a/CONTRIBUTORS.md +++ b/CONTRIBUTORS.md @@ -17,7 +17,7 @@ Below are some of the specific details of various contributions. * Jeffrey Phillips Freeman has contributed an implementation for the invariant propagation fix (PR #10881 / issue #9131): added `_propagate_invariant_decisions()` to `SubplanService` to propagate all `invariant_enforced` decisions from parent plans to child plan decision trees during subplan spawn, satisfying the specification requirement for invariant propagation across hierarchical plan execution. * Brent E. Edwards has contributed quality assurance, test coverage, and CI pipeline improvements. * HAL 9000 has contributed automated implementation, bug fixes, and feature development as part of the CleverAgents automation pool. -* HAL 9000 has contributed concurrency safety improvements, including thread-safe context tier management (issue #7547) for parallel plan execution. +* HAL 9000 has contributed concurrency safety improvements, including thread-safe context tier management (issue #7547) and thread-safe InvariantService (#8209) with ``threading.RLock`` protection for all shared state mutations under concurrent plan execution. * HAL 9000 has contributed the plan concurrency race-condition fix (#7989): wired `LockService` into the plan lifecycle, guarding `execute_plan()` and `apply_plan()` with plan-level advisory locks and unique per-invocation owner identities to prevent silent concurrent state corruption. * HAL 9000 has contributed the bug-hunt-pool-supervisor non-blocking tracking fix (#7875 / PR #7957): updated step 5 to be best-effort and added rule 9 to prevent the automation-tracking-manager call from blocking the main supervisor loop. * Jeffrey Phillips Freeman has contributed the complete AUTO-BUG-POOL to AUTO-BUG-SUP tracking prefix fix across agent-system-specification.md, automation-tracking.md documentation and agent-system-specification.md spec document, replaced with correct `AUTO-BUG-SUP` prefix used by the bug-hunt-pool-supervisor agent (#7875). diff --git a/features/invariant_service_thread_safety.feature b/features/invariant_service_thread_safety.feature new file mode 100644 index 000000000..a03682f42 --- /dev/null +++ b/features/invariant_service_thread_safety.feature @@ -0,0 +1,87 @@ +@mock_only +Feature: InvariantService thread safety (#8209) + As a CleverAgents developer + I want InvariantService to be safe for concurrent access from multiple threads + So that parallel plan execution with a shared singleton service instance does + not cause RuntimeError: dictionary changed size during iteration or + data corruption in the invariants dict and enforcement records list + + Background: + Given a thread-safe invariant service + + # ----------------------------------------------------------------- + # Basic lock presence + # ----------------------------------------------------------------- + + Scenario: InvariantService has a reentrant lock attribute + Then the invariant service should have a _lock attribute + And the _lock should be a threading.RLock + + # ----------------------------------------------------------------- + # Concurrent add_invariant + # ----------------------------------------------------------------- + + Scenario: Concurrent add from multiple threads does not raise RuntimeError + When 10 threads concurrently add 5 invariants each to the service + Then no RuntimeError should have been raised during concurrent add + And the service should contain at least 1 invariant + + # ----------------------------------------------------------------- + # Concurrent list (read) operations + # ----------------------------------------------------------------- + + Scenario: Concurrent list from multiple threads does not raise RuntimeError + Given 20 invariants pre-added to the service + When 10 threads concurrently list invariants from the service + Then no exception should have been raised during concurrent list + + Scenario: Concurrent add and list from multiple threads is data-race free + When 5 threads concurrently add invariants and 5 threads concurrently list invariants + Then no exception should have been raised during concurrent add and list + + # ----------------------------------------------------------------- + # Concurrent remove operations + # ----------------------------------------------------------------- + + Scenario: Concurrent remove from multiple threads does not raise RuntimeError + Given 20 invariants pre-added to the service with known IDs + When 5 threads concurrently remove invariants by ID from the service + Then no exception should have been raised during concurrent remove + And remaining active invariants should be consistent (no orphans) + + # ----------------------------------------------------------------- + # Concurrent effective-set computation + # ----------------------------------------------------------------- + + Scenario: Concurrent get_effective_invariants does not raise RuntimeError + Given 20 mixed-scope invariants pre-added to the service + When 10 threads concurrently call get_effective_invariants from the service + Then no exception should have been raised during concurrent effective-set computation + + # ----------------------------------------------------------------- + # Concurrent enforcement record creation + # ----------------------------------------------------------------- + + Scenario: Concurrent enforce_invariants does not raise RuntimeError + Given 10 invariants pre-added to the service from different scopes + When 5 threads concurrently enforce invariants on the service for the same plan + Then no exception should have been raised during concurrent enforcement + And at least 1 enforcement record should exist + + # ----------------------------------------------------------------- + # Combined mixed operations under contention + # ----------------------------------------------------------------- + + Scenario: Mixed add, list, remove, enforce from multiple threads is safe + When 5 threads concurrently add invariants, list invariants, and get effective set + Then no RuntimeError should have been raised during mixed concurrent operations + And the invariant count after mixed operations should be non-negative + + # ----------------------------------------------------------------- + # Singleton safety: same instance shared across threads + # ----------------------------------------------------------------- + + Scenario: Shared singleton instance survives concurrent access from 20 threads + When 20 threads each add 3 invariants and then list them back + Then no exception should have been raised during concurrent singleton access + And the service invariant count should be non-negative diff --git a/features/steps/invariant_service_thread_safety_steps.py b/features/steps/invariant_service_thread_safety_steps.py new file mode 100644 index 000000000..502af546d --- /dev/null +++ b/features/steps/invariant_service_thread_safety_steps.py @@ -0,0 +1,516 @@ +"""Step definitions for InvariantService thread-safety tests (#8209). + +Verifies that all public methods of InvariantService are safe for +concurrent access from multiple threads, preventing: +- RuntimeError: dictionary changed size during iteration +- Data corruption in _invariants dict and _enforcement_records list +- Lost updates under concurrent plan execution when the service is a + DI Singleton shared across threads + +Based on the pattern established by context_tier_thread_safety_tests (#7547). +""" + +from __future__ import annotations + +import threading +from typing import Any + +from behave import given, then, when + +from cleveragents.application.services.invariant_service import InvariantService +from cleveragents.domain.models.core.invariant import InvariantScope + +__all__: list[str] = [] + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + + +def _make_invariant(text: str, scope: InvariantScope, source_name: str) -> Any: + """Build a test Invariant via the service.""" + return None # we add through the service directly instead + + +# --------------------------------------------------------------------------- +# Background +# --------------------------------------------------------------------------- + + +@given("a thread-safe invariant service") +def step_given_thread_safe_service(context: Any) -> None: + context.inv_service = InvariantService() + context.errors: list[Exception] = [] + + +# --------------------------------------------------------------------------- +# Lock attribute checks +# --------------------------------------------------------------------------- + + +@then("the invariant service should have a _lock attribute") +def step_then_has_lock(context: Any) -> None: + assert hasattr(context.inv_service, "_lock"), ( + "InvariantService must have a _lock attribute" + ) + + +@then("the _lock should be a threading.RLock") +def step_then_lock_is_rlock(context: Any) -> None: + # threading.RLock() returns an instance of _RLock (internal type). + # We verify it has the acquire/release interface of a reentrant lock. + lock = context.inv_service._lock + assert hasattr(lock, "acquire"), "_lock must have acquire()" + assert hasattr(lock, "release"), "_lock must have release()" + # Verify reentrancy: acquiring twice from the same thread must not deadlock. + lock.acquire() + lock.acquire() + lock.release() + lock.release() + + +# --------------------------------------------------------------------------- +# Concurrent add_invariant +# --------------------------------------------------------------------------- + + +@given("{n:d} invariants pre-added to the service with known IDs") +def step_given_pre_added_invariants(context: Any, n: int) -> None: + context.inv_ids = [] + for i in range(n): + inv = context.inv_service.add_invariant( + text=f"Pre-added invariant {i}", + scope=InvariantScope.GLOBAL, + source_name="test-system", + ) + context.inv_ids.append(inv.id) + + + + +@given("{n:d} invariants pre-added to the service") +def step_given_pre_added_invariants_basic(context: Any, n: int) -> None: + """Add N invariants with GLOBAL scope and known IDs.""" + context.inv_ids = [] + for i in range(n): + inv = context.inv_service.add_invariant( + text=f"Pre-added invariant {i}", + scope=InvariantScope.GLOBAL, + source_name="test-system", + ) + context.inv_ids.append(inv.id) + + + +@given("{n:d} invariants pre-added to the service from different scopes") +def step_given_pre_added_different_scopes(context: Any, n: int) -> None: + """Add N invariants cycling through GLOBAL/PROJECT/PLAN scopes.""" + for i in range(n): + scope = [ + InvariantScope.GLOBAL, + InvariantScope.PROJECT, + InvariantScope.PLAN, + ][i % 3] + context.inv_service.add_invariant( + text=f"Enforcement invariant {i}", + scope=scope, + source_name="test-plan" if scope == InvariantScope.PLAN else f"source-{i}", + ) + +@given("{n:d} mixed-scope invariants pre-added to the service") +def step_given_mixed_scope_invariants(context: Any, n: int) -> None: + for i in range(n): + scope = (i % 3) + 1 + scope_map = { + 1: InvariantScope.GLOBAL, + 2: InvariantScope.PROJECT, + 3: InvariantScope.PLAN, + } + context.inv_service.add_invariant( + text=f"Mixed invariant {i}", + scope=scope_map[scope], + source_name=f"source-{i}", + ) + + +@when("{n:d} threads concurrently add {k:d} invariants each to the service") +def step_when_concurrent_add(context: Any, n: int, k: int) -> None: + errors: list[Exception] = [] + lock = threading.Lock() + + def worker(thread_idx: int) -> None: + for i in range(k): + try: + context.inv_service.add_invariant( + text=f"Thread-{thread_idx}-invariant-{i}", + scope=InvariantScope.GLOBAL, + source_name=f"worker-{thread_idx}", + ) + except Exception as exc: + with lock: + errors.append(exc) + + threads = [threading.Thread(target=worker, args=(t,)) for t in range(n)] + for t in threads: + t.start() + for t in threads: + t.join() + + context.errors = errors + + +@then("no RuntimeError should have been raised during concurrent add") +def step_then_no_runtime_error_add(context: Any) -> None: + runtime_errors = [e for e in context.errors if isinstance(e, RuntimeError)] + assert not runtime_errors, ( + f"RuntimeError(s) raised during concurrent add: {runtime_errors}" + ) + + +@then("the service should contain at least {n:d} invariant") +def step_then_contains_at_least(context: Any, n: int) -> None: + invariants = context.inv_service.list_invariants() + count = len(invariants) + assert count >= n, f"Expected at least {n} invariants, got {count}" + + +# --------------------------------------------------------------------------- +# Concurrent list (read) operations +# --------------------------------------------------------------------------- + + +@when("{n:d} threads concurrently list invariants from the service") +def step_when_concurrent_list(context: Any, n: int) -> None: + errors: list[Exception] = [] + lock = threading.Lock() + + def worker(thread_idx: int) -> None: + for _ in range(5): + try: + context.inv_service.list_invariants() + except Exception as exc: + with lock: + errors.append(exc) + + threads = [threading.Thread(target=worker, args=(t,)) for t in range(n)] + for t in threads: + t.start() + for t in threads: + t.join() + + context.errors = errors + + +@then("no exception should have been raised during concurrent list") +def step_then_no_exc_list(context: Any) -> None: + assert not context.errors, ( + f"Exception(s) raised during concurrent list: {context.errors}" + ) + + +# --------------------------------------------------------------------------- +# Concurrent add + list +# --------------------------------------------------------------------------- + + +@when( + "{a:d} threads concurrently add invariants and " + "{list_threads:d} threads concurrently list invariants" +) +def step_when_concurrent_add_and_list(context: Any, a: int, list_threads: int) -> None: + errors: list[Exception] = [] + lock = threading.Lock() + + def add_worker(thread_idx: int) -> None: + for i in range(3): + try: + context.inv_service.add_invariant( + text=f"Add-worker-{thread_idx}-inv-{i}", + scope=InvariantScope.GLOBAL, + source_name=f"add-worker-{thread_idx}", + ) + except Exception as exc: + with lock: + errors.append(exc) + + def list_worker(thread_idx: int) -> None: + for _ in range(5): + try: + context.inv_service.list_invariants() + except Exception as exc: + with lock: + errors.append(exc) + + threads = [threading.Thread(target=add_worker, args=(t,)) for t in range(a)] + threads += [threading.Thread(target=list_worker, args=(t,)) for t in range(list_threads)] + for t in threads: + t.start() + for t in threads: + t.join() + + context.errors = errors + + +@then("no exception should have been raised during concurrent add and list") +def step_then_no_exc_add_list(context: Any) -> None: + assert not context.errors, ( + f"Exception(s) raised during concurrent add+list: {context.errors}" + ) + + +# --------------------------------------------------------------------------- +# Concurrent remove operations +# --------------------------------------------------------------------------- + + +@when( + "{n:d} threads concurrently remove invariants by ID from the service" +) +def step_when_concurrent_remove(context: Any, n: int) -> None: + errors: list[Exception] = [] + lock = threading.Lock() + inv_ids = list(context.inv_ids)[:n] # each thread removes one unique ID + + def worker(thread_idx: int) -> None: + if thread_idx < len(inv_ids): + try: + context.inv_service.remove_invariant(inv_ids[thread_idx]) + except Exception as exc: + with lock: + errors.append(exc) + + threads = [threading.Thread(target=worker, args=(t,)) for t in range(n)] + for t in threads: + t.start() + for t in threads: + t.join() + + context.errors = errors + + +@then("no exception should have been raised during concurrent remove") +def step_then_no_exc_remove(context: Any) -> None: + assert not context.errors, ( + f"Exception(s) raised during concurrent remove: {context.errors}" + ) + + +@then("remaining active invariants should be consistent (no orphans)") +def step_then_active_invariants_consistent(context: Any) -> None: + # Verify that all active invariants' IDs are still in the dict + active_ids = set() + for inv_id, inv in context.inv_service._invariants.items(): + if inv.active: + active_ids.add(inv_id) + + listed = context.inv_service.list_invariants() + listed_ids = {inv.id for inv in listed} + + assert active_ids == listed_ids, ( + f"Mismatch between _invariants active set ({active_ids}) and " + f"list_invariants output ({listed_ids})" + ) + + +# --------------------------------------------------------------------------- +# Concurrent effective-set computation +# --------------------------------------------------------------------------- + + +@when( + "{n:d} threads concurrently call get_effective_invariants from the service" +) +def step_when_concurrent_effective(context: Any, n: int) -> None: + errors: list[Exception] = [] + lock = threading.Lock() + + def worker(thread_idx: int) -> None: + for _ in range(5): + try: + context.inv_service.get_effective_invariants(plan_id="test-plan") + except Exception as exc: + with lock: + errors.append(exc) + + threads = [threading.Thread(target=worker, args=(t,)) for t in range(n)] + for t in threads: + t.start() + for t in threads: + t.join() + + context.errors = errors + + +@then( + "no exception should have been raised during concurrent effective-set computation" +) +def step_then_no_exc_effective(context: Any) -> None: + assert not context.errors, ( + f"Exception(s) raised during concurrent get_effective_invariants: {context.errors}" + ) + + +# --------------------------------------------------------------------------- +# Concurrent enforcement record creation +# --------------------------------------------------------------------------- + + +@when( + "{n:d} threads concurrently enforce invariants on the service for the same plan" +) +def step_when_concurrent_enforce(context: Any, n: int) -> None: + errors: list[Exception] = [] + lock = threading.Lock() + + def worker(thread_idx: int) -> None: + # Build a small set of invariants for this thread + plan_inv = context.inv_service.add_invariant( + text=f"Plan invariant enforced by thread {thread_idx}", + scope=InvariantScope.PLAN, + source_name="test-plan", + ) + try: + context.inv_service.enforce_invariants( + plan_id="test-plan", + invariants=[plan_inv], + ) + except Exception as exc: + with lock: + errors.append(exc) + + threads = [threading.Thread(target=worker, args=(t,)) for t in range(n)] + for t in threads: + t.start() + for t in threads: + t.join() + + context.errors = errors + + +@then("no exception should have been raised during concurrent enforcement") +def step_then_no_exc_enforce(context: Any) -> None: + assert not context.errors, ( + f"Exception(s) raised during concurrent enforce_invariants: {context.errors}" + ) + + +@then("at least {n:d} enforcement record should exist") +def step_then_enforcement_record_exists(context: Any, n: int) -> None: + with context.inv_service._lock: + count = len(context.inv_service._enforcement_records) + assert count >= n, f"Expected at least {n} enforcement records, got {count}" + + +# --------------------------------------------------------------------------- +# Combined mixed operations under contention +# --------------------------------------------------------------------------- + + +@when( + "{n:d} threads concurrently add invariants, list invariants, and get effective set" +) +def step_when_concurrent_mixed(context: Any, n: int) -> None: + errors: list[Exception] = [] + lock = threading.Lock() + + def worker(thread_idx: int) -> None: + for i in range(3): + try: + context.inv_service.add_invariant( + text=f"Mixed-thread-{thread_idx}-inv-{i}", + scope=[InvariantScope.GLOBAL, InvariantScope.PROJECT, InvariantScope.PLAN][thread_idx % 3], + source_name=f"mixed-worker-{thread_idx}-{i}", + ) + except Exception: + pass + try: + context.inv_service.list_invariants() + except Exception as exc: + with lock: + errors.append(exc) + try: + context.inv_service.get_effective_invariants() + except Exception as exc: + with lock: + errors.append(exc) + + threads = [threading.Thread(target=worker, args=(t,)) for t in range(n)] + for t in threads: + t.start() + for t in threads: + t.join() + + context.errors = errors + + +@then("no RuntimeError should have been raised during mixed concurrent operations") +def step_then_no_runtime_error_mixed(context: Any) -> None: + runtime_errors = [e for e in context.errors if isinstance(e, RuntimeError)] + assert not runtime_errors, ( + f"RuntimeError(s) raised during mixed concurrent ops: {runtime_errors}" + ) + + +@then("the invariant count after mixed operations should be non-negative") +def step_then_inv_count_non_negative(context: Any) -> None: + invariants = context.inv_service.list_invariants() + assert len(invariants) >= 0, "Invariant count should never be negative" + + +# --------------------------------------------------------------------------- +# Singleton safety: same instance shared across threads +# --------------------------------------------------------------------------- + + +@when("{n:d} threads each add {k:d} invariants and then list them back") +def step_when_concurrent_singleton(context: Any, n: int, k: int) -> None: + errors: list[Exception] = [] + lock = threading.Lock() + + def worker(thread_idx: int) -> None: + for i in range(k): + try: + context.inv_service.add_invariant( + text=f"Singleton-thread-{thread_idx}-inv-{i}", + scope=InvariantScope.GLOBAL, + source_name=f"singleton-worker-{thread_idx}", + ) + # List all to verify consistency + _ = context.inv_service.list_invariants() + except Exception as exc: + with lock: + errors.append(exc) + + threads = [threading.Thread(target=worker, args=(t,)) for t in range(n)] + for t in threads: + t.start() + for t in threads: + t.join() + + context.errors = errors + + + + +@then("the service invariant count should be non-negative") +def step_then_inv_count_non_negative_simple(context: Any) -> None: + invariants = context.inv_service.list_invariants() + assert len(invariants) >= 0, "Invariant count should never be negative" + +@then("no exception should have been raised during concurrent singleton access") +def step_then_no_exc_singleton(context: Any) -> None: + assert not context.errors, ( + f"Exception(s) raised during concurrent singleton access: {context.errors}" + ) + + +@then( + "the service invariant count should be non-negative and the enforcement record count should be non-negative" +) +def step_then_both_counts_non_negative(context: Any) -> None: + invariants = context.inv_service.list_invariants() + assert len(invariants) >= 0, f"Expected non-negative count, got {len(invariants)}" + + with context.inv_service._lock: + record_count = len(context.inv_service._enforcement_records) + assert record_count >= 0, f"Expected non-negative records, got {record_count}" diff --git a/src/cleveragents/application/services/invariant_service.py b/src/cleveragents/application/services/invariant_service.py index e941b999e..28de64e59 100644 --- a/src/cleveragents/application/services/invariant_service.py +++ b/src/cleveragents/application/services/invariant_service.py @@ -9,6 +9,17 @@ lifecycle operations. Uses in-memory storage (same pattern as ``PlanLifecycleService``) with a dict keyed by invariant ID. +## Thread Safety + +A ``threading.RLock`` (re-entrant lock) guards all shared mutable state +(``_invariants``, ``_enforcement_records``). Every public method acquires +the lock before reading or writing shared data; the lock is re-entrant so +that nested calls from within a locked block do **not** deadlock. + +This prevents ``RuntimeError: dictionary changed size during iteration`` and +data corruption under concurrent plan execution when the service is +registered as a DI Singleton shared across threads. + ## Merge Precedence Effective invariants are computed using plan > project > global order. @@ -19,6 +30,7 @@ Based on ``docs/specification.md`` and implementation plan Stage M3.5. from __future__ import annotations +from threading import RLock from typing import TYPE_CHECKING import structlog @@ -46,6 +58,10 @@ class InvariantService: Provides add, list, remove (soft-delete), effective-set computation, and enforcement record creation. All storage is in-memory. + + Thread safety is provided via a ``threading.RLock`` that protects all + shared state mutations so that concurrent readers and writers cannot + race or raise ``RuntimeError: dictionary changed size during iteration``. """ def __init__(self, event_bus: EventBus | None = None) -> None: @@ -56,6 +72,7 @@ class InvariantService: """ self._invariants: dict[str, Invariant] = {} self._enforcement_records: list[InvariantEnforcementRecord] = [] + self._lock = RLock() # guards _invariants and _enforcement_records self._logger = logger.bind(service="invariant") self._sanitizer = PromptSanitizer() self._event_bus = event_bus @@ -95,7 +112,9 @@ class InvariantService: source_name=source_name.strip(), ) - self._invariants[invariant.id] = invariant + with self._lock: + self._invariants[invariant.id] = invariant + self._logger.info( "Invariant added", invariant_id=invariant.id, @@ -127,7 +146,8 @@ class InvariantService: project_name=source_name if scope == InvariantScope.PROJECT else None, ) - result = [inv for inv in self._invariants.values() if inv.active] + with self._lock: + result = [inv for inv in self._invariants.values() if inv.active] if scope is not None: result = [inv for inv in result if inv.scope == scope] @@ -152,15 +172,17 @@ class InvariantService: if not invariant_id or not invariant_id.strip(): raise ValidationError("Invariant ID must not be empty") - inv = self._invariants.get(invariant_id) - if inv is None: - raise NotFoundError( - resource_type="invariant", - resource_id=invariant_id, - ) - # Invariant is frozen (immutable); create a new instance with active=False - deactivated = inv.model_copy(update={"active": False}) - self._invariants[invariant_id] = deactivated + with self._lock: + inv = self._invariants.get(invariant_id) + if inv is None: + raise NotFoundError( + resource_type="invariant", + resource_id=invariant_id, + ) + # Invariant is frozen (immutable); create a new instance with active=False + deactivated = inv.model_copy(update={"active": False}) + self._invariants[invariant_id] = deactivated + self._logger.info("Invariant removed (soft-delete)", invariant_id=invariant_id) return deactivated @@ -183,7 +205,8 @@ class InvariantService: Returns: Merged, de-duplicated list of effective invariants. """ - active = [inv for inv in self._invariants.values() if inv.active] + with self._lock: + active = [inv for inv in self._invariants.values() if inv.active] plan_invs = [ inv @@ -264,7 +287,9 @@ class InvariantService: exc_info=True, ) - self._enforcement_records.extend(records) + with self._lock: + self._enforcement_records.extend(records) + self._logger.info( "Invariants enforced", plan_id=plan_id,