feat: SSE #1487
feat: SSE #1487
Conversation
WalkthroughThe pull request introduces a new Server-Sent Events (SSE) broadcasting module for real-time event streaming, including session-based client management, message broadcasting, and connection pruning. Dependencies are updated to support the new functionality. Changes
Sequence DiagramsequenceDiagram
actor Client
participant Handler as register_sse_client()
participant Broadcaster
participant Stream as SSE Stream
Client->>Handler: HTTP request
Handler->>Broadcaster: extract session ID from request
Broadcaster->>Broadcaster: new_client(session)
Broadcaster->>Stream: create SSE channel
Stream-->>Client: SSE connection established
Note over Broadcaster: Ping loop runs periodically
Broadcaster->>Broadcaster: broadcast(message, sessions)
Broadcaster->>Stream: send event to registered clients
Stream-->>Client: receive event
Broadcaster->>Broadcaster: ping() → prune stale connections
alt Connection responsive
Client-->>Broadcaster: ack
else Connection dead
Broadcaster->>Broadcaster: remove from registry
end
Estimated code review effort🎯 4 (Complex) | ⏱️ ~45 minutes
Poem
Pre-merge checks and finishing touches❌ Failed checks (1 warning, 1 inconclusive)
✅ Passed checks (1 passed)
✨ Finishing touches
🧪 Generate unit tests (beta)
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: 4
🧹 Nitpick comments (1)
src/sse/mod.rs (1)
56-76: Consider cleaning up empty sessions.When all clients for a session disconnect, the session key remains in the map with an empty
Vec. This can lead to accumulation of stale session entries over time.Apply this diff to filter out empty sessions:
for (session, clients) in sse_inner.iter() { let mut ok_clients = Vec::new(); for client in clients { if client .send(sse::Event::Comment("ping".into())) .await .is_ok() { ok_clients.push(client.clone()) } } - ok_sessions.insert(*session, ok_clients); + if !ok_clients.is_empty() { + ok_sessions.insert(*session, ok_clients); + } }
📜 Review details
Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro
⛔ Files ignored due to path filters (1)
Cargo.lockis excluded by!**/*.lock
📒 Files selected for processing (3)
Cargo.toml(2 hunks)src/lib.rs(1 hunks)src/sse/mod.rs(1 hunks)
🧰 Additional context used
🧠 Learnings (1)
📚 Learning: 2025-06-18T06:39:04.775Z
Learnt from: nikhilsinhaparseable
Repo: parseablehq/parseable PR: 1340
File: src/query/mod.rs:64-66
Timestamp: 2025-06-18T06:39:04.775Z
Learning: In src/query/mod.rs, QUERY_SESSION_STATE and QUERY_SESSION serve different architectural purposes: QUERY_SESSION_STATE is used for stats calculation and allows dynamic registration of individual parquet files from the staging path (files created every minute), while QUERY_SESSION is used for object store queries with the global schema provider. Session contexts with schema providers don't support registering individual tables/parquets, so both session objects are necessary for their respective use cases.
Applied to files:
src/lib.rs
🧬 Code graph analysis (1)
src/sse/mod.rs (1)
src/utils/actix.rs (1)
extract_session_key_from_req(51-71)
⏰ Context from checks skipped due to timeout of 90000ms. You can increase the timeout in your CodeRabbit configuration to a maximum of 15 minutes (900000ms). (9)
- GitHub Check: Quest Smoke and Load Tests for Standalone deployments
- GitHub Check: Quest Smoke and Load Tests for Distributed deployments
- GitHub Check: coverage
- GitHub Check: Build Default aarch64-unknown-linux-gnu
- GitHub Check: Build Default x86_64-pc-windows-msvc
- GitHub Check: Build Kafka aarch64-apple-darwin
- GitHub Check: Build Kafka x86_64-unknown-linux-gnu
- GitHub Check: Build Default x86_64-unknown-linux-gnu
- GitHub Check: Build Default aarch64-apple-darwin
🔇 Additional comments (4)
src/lib.rs (1)
45-45: LGTM!The new
ssemodule is correctly exposed and follows the existing alphabetical ordering convention.Cargo.toml (2)
87-87: LGTM!Pinning tokio-stream to 0.1.17 for reproducibility is reasonable.
33-33: Theactix-web-labversion 0.24.3 is the latest stable release and is compatible with actix-web 4.9.0. No changes needed.src/sse/mod.rs (1)
20-40: LGTM!The
Broadcasterdesign is appropriate:Arcfor shared ownership across handlers,RwLockfor async read-write access, and spawning a cleanup loop on creation.
| if let Some(clients) = self.inner.write().await.clients.get_mut(session) { | ||
| clients.push(tx); | ||
| } else { | ||
| self.inner.write().await.clients.insert(*session, vec![tx]); | ||
| } |
There was a problem hiding this comment.
Fix race condition with double write lock acquisition.
The code acquires a write lock on line 87, checks if the session exists, then drops the lock. In the else branch on line 90, it acquires another write lock. Between these two acquisitions, another concurrent call to new_client for the same session could insert an entry, leading to lost client registrations.
Use the Entry API to handle this atomically:
- if let Some(clients) = self.inner.write().await.clients.get_mut(session) {
- clients.push(tx);
- } else {
- self.inner.write().await.clients.insert(*session, vec![tx]);
- }
+ self.inner
+ .write()
+ .await
+ .clients
+ .entry(*session)
+ .or_default()
+ .push(tx);📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| if let Some(clients) = self.inner.write().await.clients.get_mut(session) { | |
| clients.push(tx); | |
| } else { | |
| self.inner.write().await.clients.insert(*session, vec![tx]); | |
| } | |
| self.inner | |
| .write() | |
| .await | |
| .clients | |
| .entry(*session) | |
| .or_default() | |
| .push(tx); |
🤖 Prompt for AI Agents
In src/sse/mod.rs around lines 87-91, avoid acquiring the write lock twice by
holding a single write lock and using the HashMap Entry API to insert-or-get the
client Vec atomically: obtain let mut guard = self.inner.write().await once,
then call guard.clients.entry(key).or_insert_with(Vec::new).push(tx) (use the
appropriate key ownership/clone or dereference as needed) so concurrent
new_client calls cannot race and lose registrations.
| pub async fn broadcast(&self, msg: &str, sessions: Option<&[Ulid]>) { | ||
| let clients = self.inner.read().await.clients.clone(); | ||
| warn!(clients=?clients); |
There was a problem hiding this comment.
Change log level from warn to debug or trace.
Logging all connected clients at warn level on every broadcast will spam logs during normal operation. This appears to be debug logging left over from development.
- warn!(clients=?clients);
+ tracing::debug!(clients=?clients);Committable suggestion skipped: line range outside the PR's diff.
🤖 Prompt for AI Agents
In src/sse/mod.rs around lines 99 to 101, the code currently logs connected
clients at warn level (warn!(clients=?clients)); change this to a lower
verbosity (use debug! or trace! instead of warn!) so normal broadcasts don't
spam logs; simply replace the warn! macro with debug! (or trace! if you prefer
more verbosity) and keep the same structured field formatting (clients=?clients)
so only the log level changes.
| pub async fn register_sse_client( | ||
| broadcaster: actix_web::web::Data<Arc<Broadcaster>>, | ||
| req: HttpRequest, | ||
| ) -> Sse<InfallibleStream<ReceiverStream<sse::Event>>> { | ||
| let session = extract_session_key_from_req(&req).unwrap(); | ||
| let sessionid = match session { | ||
| SessionKey::SessionId(ulid) => ulid, | ||
| _ => unreachable!("SSE requires a valid session. Unable to register client."), | ||
| }; | ||
| broadcaster.new_client(&sessionid).await | ||
| } |
There was a problem hiding this comment.
Replace panics with proper error handling.
This function will panic on unauthenticated requests (.unwrap() on line 134) or when a user authenticates via BasicAuth instead of session cookie (unreachable!() on line 137). HTTP handlers should not panic; they should return appropriate error responses.
Consider returning a Result or an Actix error response:
-pub async fn register_sse_client(
+pub async fn register_sse_client(
broadcaster: actix_web::web::Data<Arc<Broadcaster>>,
req: HttpRequest,
-) -> Sse<InfallibleStream<ReceiverStream<sse::Event>>> {
- let session = extract_session_key_from_req(&req).unwrap();
- let sessionid = match session {
- SessionKey::SessionId(ulid) => ulid,
- _ => unreachable!("SSE requires a valid session. Unable to register client."),
- };
- broadcaster.new_client(&sessionid).await
+) -> Result<Sse<InfallibleStream<ReceiverStream<sse::Event>>>, actix_web::Error> {
+ let session = extract_session_key_from_req(&req)?;
+ let sessionid = match session {
+ SessionKey::SessionId(ulid) => ulid,
+ SessionKey::BasicAuth { .. } => {
+ return Err(actix_web::error::ErrorBadRequest(
+ "SSE requires session-based authentication, not BasicAuth",
+ ));
+ }
+ };
+ Ok(broadcaster.new_client(&sessionid).await)
}📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| pub async fn register_sse_client( | |
| broadcaster: actix_web::web::Data<Arc<Broadcaster>>, | |
| req: HttpRequest, | |
| ) -> Sse<InfallibleStream<ReceiverStream<sse::Event>>> { | |
| let session = extract_session_key_from_req(&req).unwrap(); | |
| let sessionid = match session { | |
| SessionKey::SessionId(ulid) => ulid, | |
| _ => unreachable!("SSE requires a valid session. Unable to register client."), | |
| }; | |
| broadcaster.new_client(&sessionid).await | |
| } | |
| pub async fn register_sse_client( | |
| broadcaster: actix_web::web::Data<Arc<Broadcaster>>, | |
| req: HttpRequest, | |
| ) -> Result<Sse<InfallibleStream<ReceiverStream<sse::Event>>>, actix_web::Error> { | |
| let session = extract_session_key_from_req(&req)?; | |
| let sessionid = match session { | |
| SessionKey::SessionId(ulid) => ulid, | |
| SessionKey::BasicAuth { .. } => { | |
| return Err(actix_web::error::ErrorBadRequest( | |
| "SSE requires session-based authentication, not BasicAuth", | |
| )); | |
| } | |
| }; | |
| Ok(broadcaster.new_client(&sessionid).await) | |
| } |
🤖 Prompt for AI Agents
In src/sse/mod.rs around lines 130–140, replace the panic-causing unwrap() and
unreachable!() with proper error handling and change the handler to return a
Result so unauthorized/invalid requests produce HTTP errors instead of panics:
update the function signature to return
Result<Sse<InfallibleStream<ReceiverStream<sse::Event>>>, actix_web::Error> (or
Result<_, HttpResponse>), call extract_session_key_from_req and map its None/Err
to an appropriate HttpResponse::Unauthorized/BadRequest, match SessionKey and
return an Err(HttpResponse::Unauthorized().finish()) when not a SessionId, and
finally return Ok(broadcaster.new_client(&sessionid).await) so any broadcaster
errors can be propagated or converted into an HTTP 500 response.
| #[derive(Serialize, Debug, Deserialize)] | ||
| #[serde(rename_all = "camelCase")] | ||
| pub struct SSEAlertInfo { | ||
| id: Ulid, | ||
| state: AlertState, | ||
| message: String, | ||
| } | ||
|
|
||
| #[derive(Serialize, Debug, Deserialize)] | ||
| #[serde(rename_all = "camelCase")] | ||
| pub struct ControlPlaneEvent { | ||
| message: String, | ||
| } |
There was a problem hiding this comment.
Private fields prevent external instantiation.
SSEAlertInfo and ControlPlaneEvent have private fields (Rust default), making them impossible to construct from outside this module. If these are intended for use by other modules (e.g., to create and broadcast alert events), either make the fields pub or provide constructor functions.
#[derive(Serialize, Debug, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct SSEAlertInfo {
- id: Ulid,
- state: AlertState,
- message: String,
+ pub id: Ulid,
+ pub state: AlertState,
+ pub message: String,
}
#[derive(Serialize, Debug, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct ControlPlaneEvent {
- message: String,
+ pub message: String,
}📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| #[derive(Serialize, Debug, Deserialize)] | |
| #[serde(rename_all = "camelCase")] | |
| pub struct SSEAlertInfo { | |
| id: Ulid, | |
| state: AlertState, | |
| message: String, | |
| } | |
| #[derive(Serialize, Debug, Deserialize)] | |
| #[serde(rename_all = "camelCase")] | |
| pub struct ControlPlaneEvent { | |
| message: String, | |
| } | |
| #[derive(Serialize, Debug, Deserialize)] | |
| #[serde(rename_all = "camelCase")] | |
| pub struct SSEAlertInfo { | |
| pub id: Ulid, | |
| pub state: AlertState, | |
| pub message: String, | |
| } | |
| #[derive(Serialize, Debug, Deserialize)] | |
| #[serde(rename_all = "camelCase")] | |
| pub struct ControlPlaneEvent { | |
| pub message: String, | |
| } |
🤖 Prompt for AI Agents
In src/sse/mod.rs around lines 165 to 177, the structs SSEAlertInfo and
ControlPlaneEvent define fields with private visibility which prevents external
modules from constructing instances; either make each field public (e.g., pub
id: Ulid, pub state: AlertState, pub message: String for SSEAlertInfo and pub
message: String for ControlPlaneEvent) or add public constructor functions
(e.g., pub fn new(...) -> Self) that accept the necessary parameters and return
the struct, and ensure serde attributes remain intact for serialization.
1 participant
Add backend for SSE
Fixes #XXXX.
Description
This PR has:
Summary by CodeRabbit
✏️ Tip: You can customize this high-level summary in your review settings.