fix: parse and detect sse error event data from dynamo #385
Conversation
Codecov Report✅ All modified and coverable lines are covered by tests. 📢 Thoughts on this report? Let us know! |
ajcasagrande
force-pushed
the
ajc/sse-error-check
branch
from
0e739b4 to
1b5e7c2
Compare
|
@coderabbitai review |
|
@ajcasagrande: I'll review the changes in this PR right away. ✅ Actions performedReview triggered.
|
✅ Actions performedReview triggered.
|
|
Tip For best results, initiate chat on the files or code changes. @ajcasagrande: Hi! I see you've tagged me. How can I help you with this PR? I've already triggered a review based on your previous request. Would you like me to:
Please let me know what you need! 🐰 |
WalkthroughThe changes introduce Server-Sent Events (SSE) error handling across the codebase by adding a new SSEEventType enum with an ERROR member, creating an SSEResponseError exception, and integrating error detection logic into the transport layer to catch and propagate SSE errors during message streaming. Changes
Estimated code review effort🎯 3 (Moderate) | ⏱️ ~20 minutes
Poem
Pre-merge checks✅ Passed checks (3 passed)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 1
🧹 Nitpick comments (1)
tests/clients/http/test_aiohttp_sse.py (1)
257-271: Remove unused fixture parameter.The
mock_sse_responseparameter is received but never used—a newMock()is created on line 268 instead. Remove the parameter to clean up the fixture signature.Apply this diff:
@pytest.fixture -async def mock_sse_iterator(mock_sse_response: Mock): +async def mock_sse_iterator(): """Async fixture that patches AioHttpSSEStreamReader.__aiter__ for testing."""
📜 Review details
Configuration used: Path: .coderabbit.yaml
Review profile: CHILL
Plan: Pro
📒 Files selected for processing (9)
aiperf/clients/http/aiohttp_client.py(4 hunks)aiperf/common/enums/__init__.py(2 hunks)aiperf/common/enums/sse_enums.py(1 hunks)aiperf/common/exceptions.py(1 hunks)aiperf/common/models/error_models.py(1 hunks)tests/clients/http/conftest.py(2 hunks)tests/clients/http/test_aiohttp_client.py(1 hunks)tests/clients/http/test_aiohttp_sse.py(2 hunks)tests/conftest.py(2 hunks)
🧰 Additional context used
🧬 Code graph analysis (7)
aiperf/common/enums/sse_enums.py (1)
aiperf/common/enums/base_enums.py (1)
CaseInsensitiveStrEnum(10-49)
aiperf/common/enums/__init__.py (1)
aiperf/common/enums/sse_enums.py (1)
SSEEventType(17-20)
aiperf/common/models/error_models.py (1)
tests/controller/conftest.py (1)
error_details(68-70)
tests/conftest.py (1)
aiperf/module_loader.py (1)
ensure_modules_loaded(47-58)
aiperf/clients/http/aiohttp_client.py (5)
aiperf/common/enums/sse_enums.py (2)
SSEEventType(17-20)SSEFieldType(7-14)aiperf/common/exceptions.py (1)
SSEResponseError(158-163)tests/utils/time_traveler.py (2)
time(48-49)perf_counter_ns(54-55)aiperf/common/protocols.py (1)
error(70-70)aiperf/common/models/error_models.py (2)
ErrorDetails(11-50)from_exception(42-50)
tests/clients/http/test_aiohttp_client.py (3)
tests/clients/http/conftest.py (2)
aiohttp_client(115-119)sse_error_reader_context(291-310)aiperf/clients/http/aiohttp_client.py (1)
AioHttpClientMixin(28-165)aiperf/common/models/record_models.py (2)
start_perf_ns(539-541)end_perf_ns(551-562)
tests/clients/http/test_aiohttp_sse.py (4)
aiperf/clients/http/aiohttp_client.py (3)
AioHttpSSEStreamReader(168-244)parse_sse_message(247-273)read_complete_stream(178-213)aiperf/common/enums/sse_enums.py (1)
SSEFieldType(7-14)aiperf/common/exceptions.py (1)
SSEResponseError(158-163)aiperf/common/models/error_models.py (2)
ErrorDetails(11-50)from_exception(42-50)
🪛 Ruff (0.14.1)
aiperf/clients/http/aiohttp_client.py
137-137: Do not catch blind exception: Exception
(BLE001)
207-209: Avoid specifying long messages outside the exception class
(TRY003)
tests/clients/http/test_aiohttp_sse.py
257-257: Unused function argument: mock_sse_response
(ARG001)
🔇 Additional comments (17)
aiperf/common/models/error_models.py (1)
41-50: LGTM! Clean error code extraction from exceptions.The refactored
from_exceptionmethod correctly extracts theerror_codeattribute when present, enabling richer error details for exceptions likeSSEResponseError.aiperf/common/exceptions.py (1)
158-164: LGTM! Well-structured exception for SSE errors.The
SSEResponseErrorclass correctly stores theerror_codeattribute, which will be automatically extracted byErrorDetails.from_exception.aiperf/common/enums/__init__.py (1)
95-97: LGTM! Public API surface updated correctly.
SSEEventTypeis now properly exported for external usage.Also applies to: 166-166
tests/conftest.py (1)
33-33: LGTM! Module preloading ensures factory registration.Calling
ensure_modules_loaded()at module import time guarantees all factories are registered before test execution.Also applies to: 71-72
aiperf/common/enums/sse_enums.py (1)
16-20: LGTM! Well-structured enum for SSE event types.The
SSEEventTypeenum follows existing patterns and provides case-insensitive matching for SSE error events.tests/clients/http/conftest.py (1)
5-5: LGTM! Well-designed fixture for SSE error testing.The
sse_error_reader_contextfixture provides a clean, reusable way to simulate SSE error scenarios with proper mocking and context management.Also applies to: 290-310
tests/clients/http/test_aiohttp_client.py (1)
369-413: LGTM! Comprehensive SSE error handling tests.Both tests thoroughly validate SSE error detection, conversion to
ErrorDetails, and timing consistency. The tests verify:
- Error code (502) propagation
- Exception type preservation
- Message content
- Proper timing metrics
aiperf/clients/http/aiohttp_client.py (4)
12-13: LGTM! Necessary imports for SSE error handling.
133-140: LGTM! Proper exception handling with unified error conversion.The dedicated
SSEResponseErrorhandler and fallback generic handler both correctly set timing, log errors, and convert exceptions toErrorDetails. The static analysis warning about catchingExceptionis a false positive—this is the appropriate final catch-all for the request handler.
190-209: SSE error detection logic is sound.The implementation correctly:
- Identifies error events by checking the first packet
- Extracts error messages from the comment field
- Raises
SSEResponseErrorwith a 502 code (appropriate as it indicates the SSE stream is in an error state)The fallback error message when no comment is present includes the full JSON dump, which provides complete diagnostic information.
240-244: LGTM! Cleaner UTF-8 decoding with error handling.The simplified approach using
errors='replace'is more robust and concise than the previous exception-based handling.tests/clients/http/test_aiohttp_sse.py (6)
10-20: LGTM!The new imports are appropriate and necessary for the comprehensive SSE error handling test suite.
241-253: LGTM!The fixture provides a clean interface for building SSE stream test data with auto-incrementing timestamps, reducing test boilerplate while maintaining flexibility for timestamp overrides.
274-278: LGTM!Good use of a class-level constant for consistent timestamp values across tests.
280-313: LGTM!Comprehensive parametrized test coverage for SSE error messages with various comment formats.
315-328: LGTM!Good coverage of error events without the comment field, properly validating the fallback "Unknown error" message.
347-467: LGTM!Excellent comprehensive test coverage for SSE error handling edge cases:
- Error in middle of stream with immediate exception
- Correct 502 error code propagation
- ErrorDetails conversion preservation
- Normal events not misidentified as errors
- Long messages, special characters, whitespace handling
- Multiple errors and mixed field scenarios
- Raw escape sequence parsing
The tests are well-structured, properly documented, and provide thorough validation of the error handling implementation.
ajcasagrande
force-pushed
the
ajc/sse-error-check
branch
from
d2cfb1d to
e065976
Compare
|
@coderabbitai review |
✅ Actions performedReview triggered.
|
1 similar comment
✅ Actions performedReview triggered.
|
There was a problem hiding this comment.
Actionable comments posted: 0
🧹 Nitpick comments (1)
src/aiperf/transports/sse_utils.py (1)
80-106: Solid error detection logic with clear flow.The method correctly:
- Detects error events by checking for EVENT field with ERROR value
- Extracts error details from COMMENT field when available
- Falls back to the full message JSON for debugging
- Raises
SSEResponseErrorwith 502 (Bad Gateway), which is semantically appropriate for upstream SSE errorsThe logic handles edge cases well (no comment field, unknown errors) and provides informative error messages.
The static analysis hint (TRY003) suggests moving the error message construction into the exception class or a helper method to improve maintainability. This is a minor stylistic improvement:
# Option 1: Helper method def _format_sse_error_message(error_message: str | None, message: SSEMessage) -> str: if error_message: return f"Error occurred in SSE response: {error_message}" return f"Unknown error in SSE response: {message.model_dump_json()}" # Then in the method: raise SSEResponseError(_format_sse_error_message(error_message, message), error_code=502)
📜 Review details
Configuration used: Path: .coderabbit.yaml
Review profile: CHILL
Plan: Pro
📒 Files selected for processing (6)
src/aiperf/common/enums/__init__.py(2 hunks)src/aiperf/common/enums/sse_enums.py(1 hunks)src/aiperf/common/exceptions.py(1 hunks)src/aiperf/common/models/error_models.py(1 hunks)src/aiperf/transports/aiohttp_client.py(3 hunks)src/aiperf/transports/sse_utils.py(2 hunks)
🧰 Additional context used
🪛 Ruff (0.14.1)
src/aiperf/transports/sse_utils.py
104-106: Avoid specifying long messages outside the exception class
(TRY003)
🔇 Additional comments (9)
src/aiperf/transports/aiohttp_client.py (3)
9-9: LGTM! Import added for SSE error handling.The import is correctly placed and necessary for the new exception handling logic below.
106-106: Well-placed error detection for SSE messages.The error inspection is correctly positioned to detect SSE error events early in the processing pipeline, enabling fail-fast behavior before the message is appended to the responses list.
119-126: Good separation of SSE error handling from general exceptions.The dedicated
SSEResponseErrorhandler provides specific logging while maintaining consistent error recording patterns. The use ofErrorDetails.from_exception(e)enables automatic propagation of theerror_codeattribute from the exception.src/aiperf/common/models/error_models.py (1)
68-76: Clean refactoring to support error code propagation.The modification correctly enables exceptions with an
error_codeattribute (likeSSEResponseError) to propagate their error code to theErrorDetailsinstance, while maintaining backward compatibility with exceptions that don't have this attribute.src/aiperf/common/enums/sse_enums.py (1)
17-20: LGTM! Clean enum addition for SSE event types.The new
SSEEventTypeenum follows the existing pattern and conventions. TheERRORevent type aligns with standard SSE protocol usage.src/aiperf/common/enums/__init__.py (1)
94-94: LGTM! Auto-generated export of SSEEventType.The export is correctly added to both the import statement and the
__all__list, making the new enum publicly available through the package's public API.Also applies to: 159-159
src/aiperf/common/exceptions.py (1)
170-176: Well-designed exception for SSE error handling.The
SSEResponseErrorclass correctly extendsAIPerfErrorand includes anerror_codeattribute that will be propagated toErrorDetailsvia thefrom_exceptionmethod. The defaulterror_code=500provides a reasonable fallback, while callers can specify more specific codes (like 502 for Bad Gateway).src/aiperf/transports/sse_utils.py (2)
9-10: LGTM! Imports added for SSE error detection.The necessary enums and exception class are imported to support the new error detection functionality.
76-76: Consistent error detection in complete stream reading.The error inspection call mirrors the pattern in the streaming path, ensuring SSE errors are detected regardless of how the stream is consumed.
1 participant
Summary by CodeRabbit
Release Notes