feat: EventQueue - unify implementation between python versions

[bartek-w]

Introduced a compatibility layer using the culsans library to backport asyncio.Queue.shutdown functionality to Python versions older than 3.13. Previous implementation was broken (deadlocks and inconsistent behaviour with 3.13 implementation). Culsans library allowed for unified code between versions.

EventConsumer now starts a background task to gracefully wait for queue to finish.

This is one of the steps towards better concurrency model in a2a python sdk.

Fixes #869

[gemini-code-assist]

Summary of Changes

Hello, I'm Gemini Code Assist¹! I'm currently reviewing this pull request and will post my feedback shortly. In the meantime, here's a summary to help you and other reviewers quickly get up to speed!

This pull request significantly enhances the event queue's shutdown mechanism, ensuring a graceful and robust closing process across various Python versions. By introducing a compatibility layer for asyncio.Queue.shutdown and refactoring the close logic, the system can now reliably drain events and prevent deadlocks during termination, improving overall stability and data integrity.

Highlights

• Python 3.13+ asyncio.Queue.shutdown compatibility: Introduced a compatibility layer using the culsans library to backport asyncio.Queue.shutdown functionality to Python versions older than 3.13, ensuring consistent graceful queue closing behavior across different Python environments.
• Refactored EventQueue.close method: The EventQueue.close method was significantly refactored to unify its behavior across Python versions, leveraging the new compatibility layer. It now supports graceful shutdown (waiting for events to drain) and immediate shutdown, with an added parameter to clear parent events to prevent deadlocks.
• EventConsumer graceful shutdown: The EventConsumer now initiates a graceful close of its associated EventQueue in the background when a final event is processed. This prevents data loss for slow consumers and avoids deadlocks by clearing trailing events.
• Enhanced EventQueue robustness: Improved the EventQueue to handle QueueShutDown exceptions during event enqueuing, preventing new events from being added to a closed queue and logging a warning instead. Dequeuing from a closed and empty queue now consistently raises QueueShutDown.
• Comprehensive testing for graceful close: Added extensive new tests to test_event_consumer.py and test_event_queue.py to thoroughly validate the graceful close mechanism, including scenarios for slow consumers, race conditions during enqueue/close, and proper propagation of shutdown signals.

Using Gemini Code Assist

The full guide for Gemini Code Assist can be found on our documentation page, here are some quick tips.

Invoking Gemini

You can request assistance from Gemini at any point by creating a comment using either /gemini <command> or @gemini-code-assist <command>. Below is a summary of the supported commands on the current page.

Feature	Command	Description
Code Review	/gemini review	Performs a code review for the current pull request in its current state.
Pull Request Summary	/gemini summary	Provides a summary of the current pull request in its current state.
Comment	@gemini-code-assist	Responds in comments when explicitly tagged, both in pull request comments and review comments.
Help	/gemini help	Displays a list of available commands.

Customization

To customize Gemini Code Assist for GitHub experience, repository maintainers can create a configuration file and/or provide a custom code review style guide (such as PEP-8 for Python) by creating and adding files to a .gemini/ folder in the base of the repository. Detailed instructions can be found here.

Limitations & Feedback

Gemini Code Assist may make mistakes. Please leave feedback on any instances where its feedback is incorrect or counter productive. You can react with 👍 and 👎 on @gemini-code-assist comments. If you're interested in giving your feedback about your experience with Gemini Code Assist for GitHub and other Google products, sign up here.

Footnotes

1. Review the Privacy Notices, Generative AI Prohibited Use Policy, Terms of Service, and learn how to configure Gemini Code Assist in GitHub here. Gemini can make mistakes, so double check it and use code with caution. ↩

[github-actions]

🧪 Code Coverage (vs 1.0-dev)

⬇️ Download Full Report

	Base	PR	Delta
src/a2a/server/events/event_consumer.py	90.91%	91.80%	🟢 +0.89%
src/a2a/server/events/event_queue.py	90.24%	95.37%	🟢 +5.13%
Total	90.69%	90.77%	🟢 +0.08%

Generated by coverage-comment.yml

[gemini-code-assist]

Code Review

This pull request significantly improves the EventQueue and EventConsumer by introducing graceful shutdown mechanisms and ensuring cross-Python version compatibility for asyncio.QueueShutDown behavior. The refactoring of the close method in EventQueue is well-executed, abstracting away version-specific logic. The addition of culsans for older Python versions and the updated test suite demonstrate a thorough approach to handling these complexities. The new tests adequately cover the graceful closing scenarios and race conditions, which is crucial for robust asynchronous event handling. Additionally, the docstring for the close method has been expanded for better clarity and maintainability.

[bartek-w]

/gemini review

[gemini-code-assist]

Code Review

This pull request introduces a significant improvement to the event queue's graceful shutdown mechanism by backporting asyncio.Queue.shutdown functionality using the culsans library. This change unifies the queue behavior across different Python versions, simplifies the EventQueue implementation by removing version-specific logic, and resolves potential deadlocks in the EventConsumer. The addition of comprehensive, parametrized tests for various shutdown scenarios is a notable enhancement to the test suite's robustness. The overall changes are well-implemented and improve both the correctness and maintainability of the concurrency model.