HAL9000
3dc22f4630
Fix a critical concurrency bug in A2aEventQueue.publish() where the method iterates over _subscriptions without holding a lock. This causes RuntimeError: dictionary changed size during iteration when subscribe_local/unsubscribe are called concurrently from other threads. The fix introduces a threading.Lock to protect all state mutations and dictionary access in the A2aEventQueue class, while carefully ensuring callbacks are invoked outside the lock to prevent potential deadlocks. Changes: - src/cleveragents/a2a/events.py: Added _lock attribute, protected __init__, is_closed, publish, subscribe_local, unsubscribe, get_events, and close methods with threading.Lock snapshot pattern for callbacks. - features/a2a_event_queue_concurrency.feature: BDD feature file with 6 scenarios covering concurrent publish/subscribe/unsubscribe safety. - features/steps/a2a_event_queue_concurrency_steps.py: Step definitions implementing multi-threaded concurrency test harness. ISSUES CLOSED: #7604
63 lines
3.2 KiB
Gherkin
63 lines
3.2 KiB
Gherkin
@mock_only
|
|
Feature: A2aEventQueue thread-safety (issue #7604)
|
|
As a CleverAgents developer
|
|
I want A2aEventQueue to be safe for concurrent use from multiple threads
|
|
So that publish/subscribe/unsubscribe operations never raise
|
|
RuntimeError: dictionary changed size during iteration
|
|
|
|
# -----------------------------------------------------------------------
|
|
# Regression: concurrent publish + subscribe must not raise RuntimeError
|
|
# -----------------------------------------------------------------------
|
|
|
|
Scenario: Concurrent publish and subscribe_local do not raise RuntimeError
|
|
Given a2aconcur a fresh A2aEventQueue
|
|
When a2aconcur I run concurrent publish and subscribe_local from multiple threads
|
|
Then a2aconcur no RuntimeError should have been raised
|
|
And a2aconcur all published events should be in the queue
|
|
|
|
# -----------------------------------------------------------------------
|
|
# Regression: concurrent publish + unsubscribe must not raise RuntimeError
|
|
# -----------------------------------------------------------------------
|
|
|
|
Scenario: Concurrent publish and unsubscribe do not raise RuntimeError
|
|
Given a2aconcur a fresh A2aEventQueue with subscribers
|
|
When a2aconcur I run concurrent publish and unsubscribe from multiple threads
|
|
Then a2aconcur no RuntimeError should have been raised
|
|
|
|
# -----------------------------------------------------------------------
|
|
# Regression: concurrent subscribe + unsubscribe must not raise RuntimeError
|
|
# -----------------------------------------------------------------------
|
|
|
|
Scenario: Concurrent subscribe_local and unsubscribe do not raise RuntimeError
|
|
Given a2aconcur a fresh A2aEventQueue
|
|
When a2aconcur I run concurrent subscribe_local and unsubscribe from multiple threads
|
|
Then a2aconcur no RuntimeError should have been raised
|
|
|
|
# -----------------------------------------------------------------------
|
|
# Lock is present on new instances
|
|
# -----------------------------------------------------------------------
|
|
|
|
Scenario: A2aEventQueue has a threading.Lock after construction
|
|
Given a2aconcur a fresh A2aEventQueue
|
|
Then a2aconcur the queue should have a threading Lock attribute
|
|
|
|
# -----------------------------------------------------------------------
|
|
# publish snapshots subscriptions so late-arriving unsubscribe is safe
|
|
# -----------------------------------------------------------------------
|
|
|
|
Scenario: Unsubscribing inside a callback does not affect the current publish iteration
|
|
Given a2aconcur a fresh A2aEventQueue
|
|
And a2aconcur a subscriber that unsubscribes itself during the callback
|
|
When a2aconcur I publish a valid A2aEvent
|
|
Then a2aconcur no RuntimeError should have been raised
|
|
And a2aconcur the event should be in the queue
|
|
|
|
# -----------------------------------------------------------------------
|
|
# get_events is thread-safe
|
|
# -----------------------------------------------------------------------
|
|
|
|
Scenario: Concurrent publish and get_events do not raise RuntimeError
|
|
Given a2aconcur a fresh A2aEventQueue
|
|
When a2aconcur I run concurrent publish and get_events from multiple threads
|
|
Then a2aconcur no RuntimeError should have been raised
|