From e4998cefa0722ca1dbfa94bfde42e234068f88bc Mon Sep 17 00:00:00 2001 From: HAL9000 Date: Sat, 9 May 2026 10:23:45 +0000 Subject: [PATCH] fix(concurrency): add thread safety to InvariantService (lint and Behave fixes) Address all review feedback from HAL9001: - Remove unused 'from typing import cast' import (F401) - Replace unused variables with _ prefix (F841) - Use contextlib.suppress(Exception) instead of try/except/pass (SIM105) - Fix ScenarioOutline decorators to use curly-brace {param:d} syntax - Convert And-Then steps to proper Then/assertion steps for Behave compatibility - Rename branch from pr_fix/8209 to bugfix/m3-invariant-service-thread-safety Fixes: resolves PR #11051 review comments --- .../invariant_service_thread_safety.feature | 42 +++++++++---------- .../invariant_service_thread_safety_steps.py | 38 ++++++++--------- 2 files changed, 39 insertions(+), 41 deletions(-) diff --git a/features/invariant_service_thread_safety.feature b/features/invariant_service_thread_safety.feature index abe44ca9d..b36363a01 100644 --- a/features/invariant_service_thread_safety.feature +++ b/features/invariant_service_thread_safety.feature @@ -15,9 +15,9 @@ Feature: TDD Issue #7524 - InvariantService Thread Safety Scenario Outline: Concurrent add_invariant() must not corrupt the dict Given an invariant service - When threads concurrently add d invariants through a barrier - And the service has no RuntimeError raised during concurrent access - Then the service should have exactly d active invariants + When {n} threads concurrently add {count:d} invariants through a barrier + Then the service has no RuntimeError raised during concurrent access + And the service should have exactly {total:d} active invariants And all added invariants should be retrievable via get_invariant_snapshot Examples: @@ -28,10 +28,10 @@ Feature: TDD Issue #7524 - InvariantService Thread Safety | 10| 2 | 20 | Scenario Outline: Concurrent list_invariants() must not raise during iteration - Given an invariant service with d pre-existing invariants + Given an invariant service with {count:d} pre-existing invariants When 5 threads concurrently list all invariants through a barrier - And no thread raised RuntimeError during concurrent reads - Then the caller should receive a valid list of at least d active invariants + Then no thread raised RuntimeError during concurrent reads + And the caller should receive a valid list of at least {min:d} active invariants Examples: | count | min | @@ -41,9 +41,9 @@ Feature: TDD Issue #7524 - InvariantService Thread Safety Scenario Outline: Concurrent remove_invariant() must not race with add Given an invariant service - When threads concurrently add invariants while concurrent threads remove different ones through a barrier - And no thread raised RuntimeError during mixed access - Then the service should have a consistent count of active invariants + When {n} threads concurrently add invariants while {m:d} concurrent threads remove different ones through a barrier + Then no thread raised RuntimeError during mixed access + And the service should have a consistent count of active invariants Examples: | n | m | @@ -52,22 +52,22 @@ Feature: TDD Issue #7524 - InvariantService Thread Safety | 8 | 4 | Scenario Outline: Concurrent enforce_invariants() must not corrupt the record list - Given an invariant service with d invariants to enforce - When threads concurrently call enforce_invariants on the same set through a barrier - And no thread raised RuntimeError during enforcement - Then each thread should have received exactly d enforcement records + Given an invariant service with {count:d} invariants to enforce + When {n:d} threads concurrently call enforce_invariants on the same set through a barrier + Then no thread raised RuntimeError during enforcement + And each thread should have received exactly {total:d} enforcement records Examples: - | count | n | - | 3 | 4 | - | 5 | 8 | - | 2 | 10| + | count | n | total | + | 3 | 4 | 3 | + | 5 | 8 | 5 | + | 2 | 10| 2 | Scenario Outline: Concurrent enforce and list must both succeed - Given an invariant service with d invariants already enforced by a previous thread - When threads concurrently enforce new invariants while other threads list them through a barrier - And no thread raised RuntimeError during mixed enforcement and listing - Then the enforcement record count should be consistent across all threads + Given an invariant service with {count:d} invariants already enforced by a previous thread + When {n:d} threads concurrently enforce new invariants while other threads list them through a barrier + Then no thread raised RuntimeError during mixed enforcement and listing + And the enforcement record count should be consistent across all threads Examples: | count | n | diff --git a/features/steps/invariant_service_thread_safety_steps.py b/features/steps/invariant_service_thread_safety_steps.py index 7d8bc70d1..56267263d 100644 --- a/features/steps/invariant_service_thread_safety_steps.py +++ b/features/steps/invariant_service_thread_safety_steps.py @@ -10,8 +10,8 @@ See: https://git.cleverthis.com/cleveragents/cleveragents-core/issues/7524 from __future__ import annotations +import contextlib import threading -from typing import cast from behave import given, then, when # type: ignore[import-untyped] from behave.runner import Context @@ -64,7 +64,7 @@ def step_service_with_invariants(context: Context, count: int) -> None: """Create an InvariantService populated with specific invariants for enforcement.""" context.service = InvariantService() for i in range(count): - inv = context.service.add_invariant( + _ = context.service.add_invariant( text=f"Enforced constraint {i + 1}", scope=InvariantScope.GLOBAL, source_name="enforcement-test", @@ -101,11 +101,11 @@ def step_service_with_enforced_invariants(context: Context, count: int) -> None: @when( - " threads concurrently add d invariants through a barrier", + "{n:d} threads concurrently add {count:d} invariants through a barrier", ) def step_concurrent_add(context: Context, n: int, count: int) -> None: - """Launch n thread groups, each adding {count} invariants at once.""" - total = n * count # actual total added by this scenario + """Launch n thread groups, each adding count invariants at once.""" + _ = n * count # actual total added by this scenario barrier, threads = _create_barrier_and_threads(n) def worker(thread_index: int) -> None: @@ -210,13 +210,13 @@ def step_no_runtime_error_lists(context: Context) -> None: @then("the caller should receive a valid list of at least {min:d} active invariants") -def step_list_produces_valid_result(context: Context, min: int) -> None: +def step_list_produces_valid_result(context: Context, min_: int) -> None: """Each thread's list result should be consistent.""" # All threads should have seen the same count (snapshot consistency) for results in context._list_results: - assert len(results) >= min, ( + assert len(results) >= min_, ( f"Thread saw only {len(results)} active invariants " - f"(expected at least {min}). " + f"(expected at least {min_}). " "A reader may have seen a partially-written dict." ) @@ -227,11 +227,11 @@ def step_list_produces_valid_result(context: Context, min: int) -> None: @when( - " threads concurrently add invariants while concurrent " + "{n:d} threads concurrently add invariants while {m:d} concurrent " "threads remove different ones through a barrier", ) def step_concurrent_mixed(context: Context, n: int, m: int) -> None: - """Launch {n} add-threads and {m} remove-threads simultaneously.""" + """Launch n add-threads and m remove-threads simultaneously.""" total_workers = n + m barrier, threads = _create_barrier_and_threads(total_workers) @@ -254,10 +254,8 @@ def step_concurrent_mixed(context: Context, n: int, m: int) -> None: invs = context.service.list_invariants() if len(invs) > 0: target_id = invs[idx % len(invs)].id - try: + with contextlib.suppress(Exception): context.service.remove_invariant(target_id) - except Exception: - pass # already removed by another worker - expected race except Exception as exc: with getattr(context, "_mix_lock", threading.Lock()): context.concurrent_errors.append(exc) @@ -268,7 +266,7 @@ def step_concurrent_mixed(context: Context, n: int, m: int) -> None: for i in range(n): threads.append(threading.Thread(target=add_worker, args=(i,), daemon=True)) - remove_workers = n # index offset into the list; add workers occupy first `n` slots + _ = n # index offset into the list; add workers occupy first `n` slots for i in range(m): threads.append(threading.Thread(target=remove_worker, args=(i,), daemon=True)) @@ -306,17 +304,17 @@ def step_consistent_mixed_count(context: Context) -> None: @when( - " threads concurrently call enforce_invariants on the same set " + "{n:d} threads concurrently call enforce_invariants on the same set " "through a barrier", ) def step_concurrent_enforce(context: Context, n: int) -> None: - """Launch {n} threads that all enforce the same invariants simultaneously.""" + """Launch n threads that all enforce the same invariants simultaneously.""" barrier = threading.Barrier(n) context.concurrent_errors = [] context._enforcement_results: dict[int, list] = {} # Re-read invariants fresh inside each thread to get stable snapshot. - invariants = context.service.list_invariants() + _ = context.service.list_invariants() def worker(thread_idx: int) -> None: try: @@ -368,16 +366,16 @@ def step_no_runtime_error_enforce(context: Context) -> None: @when( - " threads concurrently enforce new invariants while other threads " + "{n:d} threads concurrently enforce new invariants while other threads " "list them through a barrier", ) def step_concurrent_enforce_list(context: Context, n: int) -> None: - """Launch {n} enforcers and {n} lister workers simultaneously.""" + """Launch n enforcers and n lister workers simultaneously.""" total_workers = n + n # enforcers + listers barrier, threads = _create_barrier_and_threads(total_workers) context.concurrent_errors = [] - invariants = context.service.list_invariants() + _ = context.service.list_invariants() def enforce_worker(idx: int) -> None: try: