From 582417a8e400b79af6843d9c4275c64130f56cd9 Mon Sep 17 00:00:00 2001 From: kvfasil Date: Wed, 8 Jul 2026 13:29:14 -0400 Subject: [PATCH 01/10] RDKEMW-21222: Implement connection-level and session-level cleanup mechanisms - Add CleanupType enum with Connection(cid) and Session(session_id) variants - Update BrokerCleaner with cleanup_connection() and cleanup_session() methods - Add cleanup_for_connection() for WebSocket disconnect (cleans only that connection) - Add cleanup_for_session() for app unload (cleans ALL subscriptions for session) - Update subscription_map to use cid as key when available via get_subscription_key() - Wire session cleanup in end_session() in delegated_launcher_handler - Remove unused cleanup_for_app() and UnloadSession command - Add unit tests for cleanup functionality --- core/main/src/broker/endpoint_broker.rs | 96 +++++++++++--- core/main/src/broker/thunder_broker.rs | 117 ++++++++++++++---- core/main/src/broker/websocket_broker.rs | 23 +++- core/main/src/firebolt/firebolt_gateway.rs | 16 +-- .../apps/delegated_launcher_handler.rs | 11 +- 5 files changed, 207 insertions(+), 56 deletions(-) diff --git a/core/main/src/broker/endpoint_broker.rs b/core/main/src/broker/endpoint_broker.rs index ae2c85d16..760a6d318 100644 --- a/core/main/src/broker/endpoint_broker.rs +++ b/core/main/src/broker/endpoint_broker.rs @@ -81,19 +81,50 @@ pub struct BrokerSender { pub sender: Sender, } +/// Enum to distinguish between connection-level and session-level cleanup +#[derive(Clone, Debug)] +pub enum CleanupType { + /// Cleanup by connection_id - used when a single WebSocket connection disconnects + Connection(String), + /// Cleanup by session_id - used when an app session is unloaded/terminated + Session(String), +} + #[derive(Clone, Debug, Default)] pub struct BrokerCleaner { - pub cleaner: Option>, + pub cleaner: Option>, } impl BrokerCleaner { - async fn cleanup_session(&self, appid: &str) -> Result { + /// Cleanup subscriptions for a specific connection (WebSocket disconnect) + pub async fn cleanup_connection(&self, connection_id: &str) -> Result { if let Some(cleaner) = self.cleaner.clone() { - if let Err(e) = cleaner.send(appid.to_owned()).await { - error!("Could not clean up {} {:?}", appid, e); + if let Err(e) = cleaner + .send(CleanupType::Connection(connection_id.to_owned())) + .await + { + error!( + "Could not clean up connection {} {:?}", + connection_id, e + ); return Err(RippleError::SendFailure); } - return Ok(appid.to_owned()); + return Ok(connection_id.to_owned()); + } + Err(RippleError::NotAvailable) + } + + /// Cleanup all subscriptions for a session (app unload/termination) + pub async fn cleanup_session(&self, session_id: &str) -> Result { + if let Some(cleaner) = self.cleaner.clone() { + if let Err(e) = cleaner + .send(CleanupType::Session(session_id.to_owned())) + .await + { + error!("Could not clean up session {} {:?}", session_id, e); + return Err(RippleError::SendFailure); + } + return Ok(session_id.to_owned()); } Err(RippleError::NotAvailable) } @@ -1094,20 +1125,31 @@ impl EndpointBrokerState { } } - // Method to cleanup all subscription on App termination - pub async fn cleanup_for_app(&self, id: &str) { + /// Cleanup subscriptions for a specific connection (WebSocket disconnect). + /// This only removes subscriptions made by this specific connection, + /// not all subscriptions for the session. + pub async fn cleanup_for_connection(&self, connection_id: &str) { let cleaners = { self.cleaner_list.read().unwrap().clone() }; for cleaner in cleaners { - /* - for now, just eat the error - the return type was mainly added to prepate for future refactoring/testability - */ - let _ = cleaner.cleanup_session(id).await; + let _ = cleaner.cleanup_connection(connection_id).await; } // Clean up subscription entries from request_map and extension_request_map - // that belong to this app. These are never removed on disconnect otherwise. - self.cleanup_request_maps(id); + self.cleanup_request_maps(connection_id); + } + + /// Cleanup all subscriptions for a session (app unload/termination). + /// This removes ALL subscriptions associated with this session_id. + pub async fn cleanup_for_session(&self, session_id: &str) { + let cleaners = { self.cleaner_list.read().unwrap().clone() }; + + for cleaner in cleaners { + let _ = cleaner.cleanup_session(session_id).await; + } + + // Clean up subscription entries from request_map and extension_request_map + self.cleanup_request_maps(session_id); } /// Remove subscription/event entries from request_map and extension_request_map @@ -3454,7 +3496,7 @@ mod endpoint_broker_tests { mod cleaner { use ripple_sdk::tokio::{self, sync::mpsc}; - use crate::broker::endpoint_broker::BrokerCleaner; + use crate::broker::endpoint_broker::{BrokerCleaner, CleanupType}; #[tokio::test] async fn test_cleanup_session_with_cleaner() { @@ -3463,7 +3505,23 @@ mod endpoint_broker_tests { assert!(cleaner.cleanup_session("test_app").await.is_ok()); let received = rx.recv().await; - assert_eq!(received, Some("test_app".to_string())); + assert!(matches!( + received, + Some(CleanupType::Session(ref s)) if s == "test_app" + )); + } + + #[tokio::test] + async fn test_cleanup_connection_with_cleaner() { + let (tx, mut rx) = mpsc::channel(1); + let cleaner = BrokerCleaner { cleaner: Some(tx) }; + + assert!(cleaner.cleanup_connection("test_cid").await.is_ok()); + let received = rx.recv().await; + assert!(matches!( + received, + Some(CleanupType::Connection(ref c)) if c == "test_cid" + )); } #[tokio::test] @@ -3473,6 +3531,14 @@ mod endpoint_broker_tests { // Should not panic or send anything assert!(cleaner.cleanup_session("test_app").await.is_err()); } + + #[tokio::test] + async fn test_cleanup_connection_without_cleaner() { + let cleaner = BrokerCleaner { cleaner: None }; + + // Should not panic or send anything + assert!(cleaner.cleanup_connection("test_cid").await.is_err()); + } } #[cfg(test)] mod workflow { diff --git a/core/main/src/broker/thunder_broker.rs b/core/main/src/broker/thunder_broker.rs index 34903c17d..7b7b690e0 100644 --- a/core/main/src/broker/thunder_broker.rs +++ b/core/main/src/broker/thunder_broker.rs @@ -17,7 +17,7 @@ use super::{ endpoint_broker::{ BrokerCallback, BrokerCleaner, BrokerConnectRequest, BrokerOutput, BrokerRequest, - BrokerSender, BrokerSubMap, EndpointBroker, EndpointBrokerState, + BrokerSender, BrokerSubMap, CleanupType, EndpointBroker, EndpointBrokerState, BROKER_CHANNEL_BUFFER_SIZE, }, thunder::thunder_plugins_status_mgr::StatusManager, @@ -198,7 +198,7 @@ impl ThunderBroker { ) -> Self { let endpoint = request.endpoint.clone(); let (broker_request_tx, mut broker_request_rx) = mpsc::channel(BROKER_CHANNEL_BUFFER_SIZE); - let (c_tx, mut c_tr) = mpsc::channel(2); + let (c_tx, mut c_tr) = mpsc::channel::(2); let broker_sender = BrokerSender { sender: broker_request_tx, }; @@ -369,13 +369,62 @@ impl ThunderBroker { }, Some(cleanup_request) = c_tr.recv() => { - let value = { - broker_for_cleanup.subscription_map.write().unwrap().remove(&cleanup_request) + // Collect subscriptions to clean up based on cleanup type + let subscriptions_to_cleanup: Vec = match &cleanup_request { + CleanupType::Connection(connection_id) => { + // Connection cleanup: remove only subscriptions with matching cid + let mut sub_map = broker_for_cleanup.subscription_map.write().unwrap(); + let mut to_cleanup = Vec::new(); + + // Iterate all entries and remove matching subscriptions + let keys: Vec = sub_map.keys().cloned().collect(); + for key in keys { + if let Some(mut requests) = sub_map.remove(&key) { + let (matching, remaining): (Vec<_>, Vec<_>) = requests + .drain(..) + .partition(|req| req.rpc.ctx.cid.as_deref() == Some(connection_id)); + to_cleanup.extend(matching); + if !remaining.is_empty() { + sub_map.insert(key, remaining); + } + } + } + debug!( + "BrokerCleaner: connection cleanup for cid={}, found {} subscription(s)", + connection_id, to_cleanup.len() + ); + to_cleanup + } + CleanupType::Session(session_id) => { + // Session cleanup: remove all subscriptions for this session_id + let mut sub_map = broker_for_cleanup.subscription_map.write().unwrap(); + let mut to_cleanup = Vec::new(); + + // Iterate all entries and remove matching subscriptions + let keys: Vec = sub_map.keys().cloned().collect(); + for key in keys { + if let Some(mut requests) = sub_map.remove(&key) { + let (matching, remaining): (Vec<_>, Vec<_>) = requests + .drain(..) + .partition(|req| req.rpc.ctx.session_id == *session_id); + to_cleanup.extend(matching); + if !remaining.is_empty() { + sub_map.insert(key, remaining); + } + } + } + debug!( + "BrokerCleaner: session cleanup for session_id={}, found {} subscription(s)", + session_id, to_cleanup.len() + ); + to_cleanup + } }; - if let Some(mut cleanup) = value { + + if !subscriptions_to_cleanup.is_empty() { debug!( "BrokerCleaner: unsubscribing {} subscription(s)", - cleanup.len() + subscriptions_to_cleanup.len() ); // Send unregister calls directly to Thunder via the WebSocket. // We must NOT route through the broker sender + prepare_request(), @@ -383,7 +432,7 @@ impl ThunderBroker { // subscription_map — but we already removed entries above. let binding = ws_tx_wrap.clone(); let mut ws_tx = binding.lock().await; - while let Some(v) = cleanup.pop() { + for v in subscriptions_to_cleanup { let (callsign, method) = ThunderBroker::get_callsign_and_method_from_alias(&v.rule.alias); if let Some(method) = method { @@ -420,7 +469,7 @@ impl ThunderBroker { } } else { debug!( - "BrokerCleaner: no subscriptions found, skipping" + "BrokerCleaner: no subscriptions found for cleanup, skipping" ); } } @@ -500,56 +549,72 @@ impl ThunderBroker { let callsign = collection.join("."); (callsign, method) } + + /// Get the subscription map key from the request context. + /// Uses connection_id (cid) if available, falls back to session_id. + fn get_subscription_key(request: &BrokerRequest) -> String { + request + .rpc + .ctx + .cid + .clone() + .unwrap_or_else(|| request.rpc.ctx.session_id.clone()) + } + fn unsubscribe(&self, request: &BrokerRequest) -> Option { let mut sub_map = self.subscription_map.write().unwrap(); + let sub_key = Self::get_subscription_key(request); trace!( - "Unsubscribing a listen request for session id: {:?}", - request.rpc.ctx.session_id + "Unsubscribing a listen request for key: {:?} (session_id: {:?}, cid: {:?})", + sub_key, + request.rpc.ctx.session_id, + request.rpc.ctx.cid ); - let app_id = &request.rpc.ctx.session_id; let method = &request.rpc.ctx.method; let mut existing_request = None; - if let Some(mut existing_requests) = sub_map.remove(app_id) { + if let Some(mut existing_requests) = sub_map.remove(&sub_key) { if let Some(i) = existing_requests .iter() .position(|x| x.rpc.ctx.method.eq_ignore_ascii_case(method)) { existing_request = Some(existing_requests.remove(i)); } - let _ = sub_map.insert(app_id.clone(), existing_requests); + if !existing_requests.is_empty() { + let _ = sub_map.insert(sub_key.clone(), existing_requests); + } } existing_request } fn subscribe(&self, request: &BrokerRequest) -> Option { let mut sub_map = self.subscription_map.write().unwrap(); - let app_id = &request.rpc.ctx.session_id; + let sub_key = Self::get_subscription_key(request); let method = &request.rpc.ctx.method; let listen = request.rpc.is_listening(); let mut response = None; debug!( - "Initial subscription map of {:?} app_id {:?}", - sub_map, app_id + "Initial subscription map of {:?} sub_key {:?}", + sub_map, sub_key ); - if let Some(mut v) = sub_map.remove(app_id) { - debug!("Subscription map after removing app {:?}", v); + if let Some(mut v) = sub_map.remove(&sub_key) { + debug!("Subscription map after removing key {:?}", v); if let Some(i) = v .iter() .position(|x| x.rpc.ctx.method.eq_ignore_ascii_case(method)) { debug!( - "Removing subscription for method {} for app {}", - method, app_id + "Removing subscription for method {} for key {}", + method, sub_key ); response = Some(v.remove(i)); } if listen { v.push(request.clone()); } - let _ = sub_map.insert(app_id.clone(), v); - } else { - let _ = sub_map.insert(app_id.clone(), vec![request.clone()]); + let _ = sub_map.insert(sub_key.clone(), v); + } else if listen { + let _ = sub_map.insert(sub_key.clone(), vec![request.clone()]); } response } @@ -1380,8 +1445,8 @@ mod tests { thunder_broker.unsubscribe(&unsubscribe_request); let subscription_map = thunder_broker.subscription_map.write().unwrap(); - // TBD : Check. Unsubscribe is re-inserting the subscription back to the map - // let _ = sub_map.insert(app_id.clone(), existing_requests); - assert_eq!(subscription_map.len(), 1); + // After unsubscribing the only subscription for this key, the map entry should be removed + // (we no longer keep empty Vec entries in the map) + assert_eq!(subscription_map.len(), 0); } } diff --git a/core/main/src/broker/websocket_broker.rs b/core/main/src/broker/websocket_broker.rs index 15fadf3e5..7c31e3d2e 100644 --- a/core/main/src/broker/websocket_broker.rs +++ b/core/main/src/broker/websocket_broker.rs @@ -17,7 +17,7 @@ use super::endpoint_broker::{ BrokerCallback, BrokerCleaner, BrokerConnectRequest, BrokerOutputForwarder, BrokerRequest, - BrokerSender, EndpointBroker, + BrokerSender, CleanupType, EndpointBroker, }; use crate::broker::endpoint_broker::EndpointBrokerState; use crate::state::platform_state::PlatformState; @@ -42,7 +42,7 @@ impl WebsocketBroker { fn start(request: BrokerConnectRequest, callback: BrokerCallback) -> Self { let endpoint = request.endpoint.clone(); let (tx, mut tr) = mpsc::channel(10); - let (cleaner_tx, mut cleaner_tr) = mpsc::channel::(1); + let (cleaner_tx, mut cleaner_tr) = mpsc::channel::(1); let non_json_rpc_map: Arc>>>> = Arc::new(RwLock::new(HashMap::new())); let map_clone = non_json_rpc_map.clone(); @@ -106,13 +106,18 @@ impl WebsocketBroker { } else { let cleaner_clone = non_json_rpc_map.clone(); tokio::spawn(async move { - while let Some(v) = cleaner_tr.recv().await { + while let Some(cleanup_type) = cleaner_tr.recv().await { + // Extract the ID from the cleanup type + let id = match cleanup_type { + CleanupType::Connection(ref cid) => cid.clone(), + CleanupType::Session(ref sid) => sid.clone(), + }; { if let Some(cleaner_list) = - { cleaner_clone.write().unwrap().remove(&v) } + { cleaner_clone.write().unwrap().remove(&id) } { for sender in cleaner_list { - if sender.try_send(v.clone()).is_err() { + if sender.try_send(id.clone()).is_err() { error!("Cleaning up listener"); } } @@ -391,7 +396,13 @@ mod tests { broker.sender.send(request).await.unwrap(); - broker.cleaner.cleaner.unwrap().send(id).await.unwrap(); + broker + .cleaner + .cleaner + .unwrap() + .send(CleanupType::Connection(id)) + .await + .unwrap(); // See if ws is closed assert!(tr.recv().await.unwrap()) } diff --git a/core/main/src/firebolt/firebolt_gateway.rs b/core/main/src/firebolt/firebolt_gateway.rs index 4470cd34a..948b3c5e0 100644 --- a/core/main/src/firebolt/firebolt_gateway.rs +++ b/core/main/src/firebolt/firebolt_gateway.rs @@ -82,6 +82,7 @@ pub enum FireboltGatewayCommand { session_id: String, session: Session, }, + /// Cleanup when a WebSocket connection disconnects UnregisterSession { session_id: String, cid: String, @@ -141,7 +142,8 @@ impl FireboltGateway { } UnregisterSession { session_id, cid } => { info!( - "Cleanup: app disconnect - removing event listeners, broker subs, session" + "Cleanup: WebSocket disconnect - cid={}, session_id={}", + cid, session_id ); // Clean event listeners by session_id AppEvents::remove_session(&self.state.platform_state, session_id.clone()); @@ -149,17 +151,15 @@ impl FireboltGateway { AppEvents::cleanup_by_connection_id(&self.state.platform_state, &cid); ProviderBroker::unregister_session(&self.state.platform_state, cid.clone()) .await; + + // Cleanup broker subscriptions by connection_id (cid) + // This removes only subscriptions made by this specific WebSocket connection self.state .platform_state .endpoint_state - .cleanup_for_app(&cid) - .await; - // Also cleanup broker subscriptions by session_id (subscription_map uses session_id as key) - self.state - .platform_state - .endpoint_state - .cleanup_for_app(&session_id) + .cleanup_for_connection(&cid) .await; + // Resolve app_id from session state BEFORE clearing the session, // because ThunderEventProcessor stores listeners by app_id (e.g. "epg"), // not by cid (a UUID). diff --git a/core/main/src/service/apps/delegated_launcher_handler.rs b/core/main/src/service/apps/delegated_launcher_handler.rs index 91d14aa54..952d3c7c3 100644 --- a/core/main/src/service/apps/delegated_launcher_handler.rs +++ b/core/main/src/service/apps/delegated_launcher_handler.rs @@ -1308,10 +1308,19 @@ impl DelegatedLauncherHandler { async fn end_session(&mut self, app_id: &str) -> Result { debug!("end_session: entry: app_id={}", app_id); let app = self.platform_state.app_manager_state.remove(app_id); - if app.is_some() { + if let Some(app) = app { if let Some(timer) = self.timer_map.remove(app_id) { timer.cancel(); } + // Clean up all subscriptions for this session when the app session ends + debug!( + "end_session: cleaning up subscriptions for session_id={}", + app.session_id + ); + self.platform_state + .endpoint_state + .cleanup_for_session(&app.session_id) + .await; } else { error!("end_session app_id={} Not found", app_id); return Err(AppError::NotFound); From 09fa1d3ba8e86ebc8f7b5aec33f19264ef8e73d3 Mon Sep 17 00:00:00 2001 From: kvfasil Date: Thu, 9 Jul 2026 13:53:22 -0400 Subject: [PATCH 02/10] RDKEMW-21222: Fix event listener over-cleanup in multi-connection secure mode - Remove redundant AppEvents::remove_session(session_id) call from disconnect handler - Add comprehensive unit test test_cleanup_by_connection_id_shared_session_multi_connection to verify multi-connection shared session scenario and prevent regression. This prevents correctness bugs in secure connection mode where multiple clients share the same logical session. --- core/main/src/firebolt/firebolt_gateway.rs | 6 +-- core/main/src/service/apps/app_events.rs | 57 ++++++++++++++++++++++ 2 files changed, 60 insertions(+), 3 deletions(-) diff --git a/core/main/src/firebolt/firebolt_gateway.rs b/core/main/src/firebolt/firebolt_gateway.rs index 948b3c5e0..d02e06fa1 100644 --- a/core/main/src/firebolt/firebolt_gateway.rs +++ b/core/main/src/firebolt/firebolt_gateway.rs @@ -145,9 +145,9 @@ impl FireboltGateway { "Cleanup: WebSocket disconnect - cid={}, session_id={}", cid, session_id ); - // Clean event listeners by session_id - AppEvents::remove_session(&self.state.platform_state, session_id.clone()); - // Also clean event listeners by connection_id (cid) in case session_id != cid + // Clean event listeners by connection_id (cid). + // All WebSocket RPC handlers populate cid, so we clean only this connection's listeners. + // removes only this connection's listeners, leaving others intact. AppEvents::cleanup_by_connection_id(&self.state.platform_state, &cid); ProviderBroker::unregister_session(&self.state.platform_state, cid.clone()) .await; diff --git a/core/main/src/service/apps/app_events.rs b/core/main/src/service/apps/app_events.rs index 0158aaa8c..fe5da94e5 100644 --- a/core/main/src/service/apps/app_events.rs +++ b/core/main/src/service/apps/app_events.rs @@ -635,4 +635,61 @@ pub mod tests { assert_eq!(state.app_events_state.listeners.read().unwrap().len(), 1); } + + #[tokio::test] + async fn test_cleanup_by_connection_id_shared_session_multi_connection() { + // CRITICAL TEST: Secure mode scenario where multiple connections share the same session_id. + // This verifies that cleanup_by_connection_id(cid) only removes the specific connection's listeners, + // NOT all listeners for the shared session_id. + // + // Background: In secure mode (ws://127.0.0.1:3473?session=), + // multiple WebSocket connections can share the same session_id but have unique cid values. + // If cleanup incorrectly removes by session_id instead of cid, it would over-clean. + let state = PlatformState::mock(); + let session_abc = "session-abc"; + let app = "app1"; + + // Connection 1 and 2 both join the same secure session "session-abc" + let ctx1 = make_ctx(session_abc, app, Some("conn-1-uuid")); + let ctx2 = make_ctx(session_abc, app, Some("conn-2-uuid")); + + // Both subscribe to the same event + add_listener_direct(&state, "onNameChanged", ctx1, None); + add_listener_direct(&state, "onNameChanged", ctx2, None); + + // Verify both listeners are registered + { + let listeners = state.app_events_state.listeners.read().unwrap(); + assert_eq!( + listeners.get("onNameChanged").unwrap().get(&None).unwrap().len(), + 2 + ); + } + + // Connection 1 disconnects: cleanup only by conn-1-uuid, NOT by session_id + AppEvents::cleanup_by_connection_id(&state, "conn-1-uuid"); + + // Verify ONLY Connection 1's listener was removed, Connection 2's remains + { + let listeners = state.app_events_state.listeners.read().unwrap(); + assert_eq!( + listeners.get("onNameChanged").unwrap().get(&None).unwrap().len(), + 1 + ); + assert_eq!( + listeners.get("onNameChanged").unwrap().get(&None).unwrap()[0] + .call_ctx + .cid + .as_deref(), + Some("conn-2-uuid") + ); + // Verify Connection 2 still has the correct session_id + assert_eq!( + listeners.get("onNameChanged").unwrap().get(&None).unwrap()[0] + .call_ctx + .session_id, + session_abc + ); + } + } } From ed72775b412ef0a1b5e301b2592760789515cdea Mon Sep 17 00:00:00 2001 From: kvfasil Date: Thu, 9 Jul 2026 16:43:17 -0400 Subject: [PATCH 03/10] RDKEMW-21222: Add info logs for Thunder register/unregister subscription requests - Log 'Sending Thunder register for .' when subscribing to Thunder events - Log 'Sending Thunder unregister for .' when unsubscribing from Thunder events - These logs enable runtime monitoring of subscription lifecycle for debugging memory leaks - Example: 'Sending Thunder unregister for org.rdk.TextToSpeech.onnetworkerror' --- core/main/src/broker/thunder_broker.rs | 111 +++++++++++++------------ 1 file changed, 59 insertions(+), 52 deletions(-) diff --git a/core/main/src/broker/thunder_broker.rs b/core/main/src/broker/thunder_broker.rs index 7b7b690e0..ffa86ce35 100644 --- a/core/main/src/broker/thunder_broker.rs +++ b/core/main/src/broker/thunder_broker.rs @@ -592,13 +592,13 @@ impl ThunderBroker { let method = &request.rpc.ctx.method; let listen = request.rpc.is_listening(); let mut response = None; - debug!( - "Initial subscription map of {:?} sub_key {:?}", - sub_map, sub_key - ); + // debug!( + // "Initial subscription map of {:?} sub_key {:?}", + // sub_map, sub_key + // ); if let Some(mut v) = sub_map.remove(&sub_key) { - debug!("Subscription map after removing key {:?}", v); + // debug!("Subscription map after removing key {:?}", v); if let Some(i) = v .iter() .position(|x| x.rpc.ctx.method.eq_ignore_ascii_case(method)) @@ -714,54 +714,61 @@ impl EndpointBroker for ThunderBroker { if rpc_request.rpc.is_subscription() && !rpc_request.rpc.is_unlisten() { let listen = rpc_request.rpc.is_listening(); // If there was an existing app and method combo for the same subscription just unregister that - if let Some(cleanup) = self.subscribe(rpc_request) { - requests.push( - json!({ - "jsonrpc": "2.0", - "id": cleanup.rpc.ctx.call_id, - "method": format!("{}.unregister", callsign), - "params": { - "event": method, - "id": format!("{}", cleanup.rpc.ctx.call_id) - } - }) - .to_string(), - ) - } - - // Given unregistration is already performed by previous step just do registration - if listen { - requests.push( - json!({ - "jsonrpc": "2.0", - "id": id, - "method": format!("{}.register", callsign), - "params": json!({ - "event": method, - "id": format!("{}", id) - }) - }) - .to_string(), - ) - } + if let Some(cleanup) = self.subscribe(rpc_request) { + info!( + "Sending Thunder unregister for {}.{}", + callsign, method + ); + requests.push( + json!({ + "jsonrpc": "2.0", + "id": cleanup.rpc.ctx.call_id, + "method": format!("{}.unregister", callsign), + "params": { + "event": method, + "id": format!("{}", cleanup.rpc.ctx.call_id) + } + }) + .to_string(), + ) + } + + // Given unregistration is already performed by previous step just do registration + if listen { + info!( + "Sending Thunder register for {}.{}", + callsign, method + ); + requests.push( + json!({ + "jsonrpc": "2.0", + "id": id, + "method": format!("{}.register", callsign), + "params": json!({ + "event": method, + "id": format!("{}", id) + }) + }) + .to_string(), + ) + } } else if rpc_request.rpc.is_unlisten() { - if let Some(cleanup) = self.unsubscribe(rpc_request) { - trace!( - "Unregistering thunder listener for call_id {} and method {}", - cleanup.rpc.ctx.call_id, - method - ); - requests.push( - json!({ - "jsonrpc": "2.0", - "id": cleanup.rpc.ctx.call_id, - "method": format!("{}.unregister", callsign), - "params": { - "event": method, - "id": format!("{}", cleanup.rpc.ctx.call_id) - } - }) - .to_string(), + if let Some(cleanup) = self.unsubscribe(rpc_request) { + info!( + "Sending Thunder unregister for {}.{}", + callsign, method + ); + requests.push( + json!({ + "jsonrpc": "2.0", + "id": cleanup.rpc.ctx.call_id, + "method": format!("{}.unregister", callsign), + "params": { + "event": method, + "id": format!("{}", cleanup.rpc.ctx.call_id) + } + }) + .to_string(), ) } } else { From 013e03c8eb6a323ec8fb14dc756b252f8c9e1afa Mon Sep 17 00:00:00 2001 From: kvfasil Date: Thu, 9 Jul 2026 20:52:53 -0400 Subject: [PATCH 04/10] Fix: fmt Error. --- core/main/src/broker/endpoint_broker.rs | 5 +- core/main/src/broker/thunder_broker.rs | 103 ++++++++++----------- core/main/src/firebolt/firebolt_gateway.rs | 4 +- core/main/src/service/apps/app_events.rs | 16 +++- 4 files changed, 63 insertions(+), 65 deletions(-) diff --git a/core/main/src/broker/endpoint_broker.rs b/core/main/src/broker/endpoint_broker.rs index 760a6d318..970aae465 100644 --- a/core/main/src/broker/endpoint_broker.rs +++ b/core/main/src/broker/endpoint_broker.rs @@ -103,10 +103,7 @@ impl BrokerCleaner { .send(CleanupType::Connection(connection_id.to_owned())) .await { - error!( - "Could not clean up connection {} {:?}", - connection_id, e - ); + error!("Could not clean up connection {} {:?}", connection_id, e); return Err(RippleError::SendFailure); } return Ok(connection_id.to_owned()); diff --git a/core/main/src/broker/thunder_broker.rs b/core/main/src/broker/thunder_broker.rs index ffa86ce35..3df89c672 100644 --- a/core/main/src/broker/thunder_broker.rs +++ b/core/main/src/broker/thunder_broker.rs @@ -375,7 +375,7 @@ impl ThunderBroker { // Connection cleanup: remove only subscriptions with matching cid let mut sub_map = broker_for_cleanup.subscription_map.write().unwrap(); let mut to_cleanup = Vec::new(); - + // Iterate all entries and remove matching subscriptions let keys: Vec = sub_map.keys().cloned().collect(); for key in keys { @@ -399,7 +399,7 @@ impl ThunderBroker { // Session cleanup: remove all subscriptions for this session_id let mut sub_map = broker_for_cleanup.subscription_map.write().unwrap(); let mut to_cleanup = Vec::new(); - + // Iterate all entries and remove matching subscriptions let keys: Vec = sub_map.keys().cloned().collect(); for key in keys { @@ -714,61 +714,52 @@ impl EndpointBroker for ThunderBroker { if rpc_request.rpc.is_subscription() && !rpc_request.rpc.is_unlisten() { let listen = rpc_request.rpc.is_listening(); // If there was an existing app and method combo for the same subscription just unregister that - if let Some(cleanup) = self.subscribe(rpc_request) { - info!( - "Sending Thunder unregister for {}.{}", - callsign, method - ); - requests.push( - json!({ - "jsonrpc": "2.0", - "id": cleanup.rpc.ctx.call_id, - "method": format!("{}.unregister", callsign), - "params": { - "event": method, - "id": format!("{}", cleanup.rpc.ctx.call_id) - } - }) - .to_string(), - ) - } - - // Given unregistration is already performed by previous step just do registration - if listen { - info!( - "Sending Thunder register for {}.{}", - callsign, method - ); - requests.push( - json!({ - "jsonrpc": "2.0", - "id": id, - "method": format!("{}.register", callsign), - "params": json!({ - "event": method, - "id": format!("{}", id) - }) - }) - .to_string(), - ) - } + if let Some(cleanup) = self.subscribe(rpc_request) { + info!("Sending Thunder unregister for {}.{}", callsign, method); + requests.push( + json!({ + "jsonrpc": "2.0", + "id": cleanup.rpc.ctx.call_id, + "method": format!("{}.unregister", callsign), + "params": { + "event": method, + "id": format!("{}", cleanup.rpc.ctx.call_id) + } + }) + .to_string(), + ) + } + + // Given unregistration is already performed by previous step just do registration + if listen { + info!("Sending Thunder register for {}.{}", callsign, method); + requests.push( + json!({ + "jsonrpc": "2.0", + "id": id, + "method": format!("{}.register", callsign), + "params": json!({ + "event": method, + "id": format!("{}", id) + }) + }) + .to_string(), + ) + } } else if rpc_request.rpc.is_unlisten() { - if let Some(cleanup) = self.unsubscribe(rpc_request) { - info!( - "Sending Thunder unregister for {}.{}", - callsign, method - ); - requests.push( - json!({ - "jsonrpc": "2.0", - "id": cleanup.rpc.ctx.call_id, - "method": format!("{}.unregister", callsign), - "params": { - "event": method, - "id": format!("{}", cleanup.rpc.ctx.call_id) - } - }) - .to_string(), + if let Some(cleanup) = self.unsubscribe(rpc_request) { + info!("Sending Thunder unregister for {}.{}", callsign, method); + requests.push( + json!({ + "jsonrpc": "2.0", + "id": cleanup.rpc.ctx.call_id, + "method": format!("{}.unregister", callsign), + "params": { + "event": method, + "id": format!("{}", cleanup.rpc.ctx.call_id) + } + }) + .to_string(), ) } } else { diff --git a/core/main/src/firebolt/firebolt_gateway.rs b/core/main/src/firebolt/firebolt_gateway.rs index d02e06fa1..d6993b80e 100644 --- a/core/main/src/firebolt/firebolt_gateway.rs +++ b/core/main/src/firebolt/firebolt_gateway.rs @@ -151,7 +151,7 @@ impl FireboltGateway { AppEvents::cleanup_by_connection_id(&self.state.platform_state, &cid); ProviderBroker::unregister_session(&self.state.platform_state, cid.clone()) .await; - + // Cleanup broker subscriptions by connection_id (cid) // This removes only subscriptions made by this specific WebSocket connection self.state @@ -159,7 +159,7 @@ impl FireboltGateway { .endpoint_state .cleanup_for_connection(&cid) .await; - + // Resolve app_id from session state BEFORE clearing the session, // because ThunderEventProcessor stores listeners by app_id (e.g. "epg"), // not by cid (a UUID). diff --git a/core/main/src/service/apps/app_events.rs b/core/main/src/service/apps/app_events.rs index fe5da94e5..a58cc7f40 100644 --- a/core/main/src/service/apps/app_events.rs +++ b/core/main/src/service/apps/app_events.rs @@ -641,7 +641,7 @@ pub mod tests { // CRITICAL TEST: Secure mode scenario where multiple connections share the same session_id. // This verifies that cleanup_by_connection_id(cid) only removes the specific connection's listeners, // NOT all listeners for the shared session_id. - // + // // Background: In secure mode (ws://127.0.0.1:3473?session=), // multiple WebSocket connections can share the same session_id but have unique cid values. // If cleanup incorrectly removes by session_id instead of cid, it would over-clean. @@ -661,7 +661,12 @@ pub mod tests { { let listeners = state.app_events_state.listeners.read().unwrap(); assert_eq!( - listeners.get("onNameChanged").unwrap().get(&None).unwrap().len(), + listeners + .get("onNameChanged") + .unwrap() + .get(&None) + .unwrap() + .len(), 2 ); } @@ -673,7 +678,12 @@ pub mod tests { { let listeners = state.app_events_state.listeners.read().unwrap(); assert_eq!( - listeners.get("onNameChanged").unwrap().get(&None).unwrap().len(), + listeners + .get("onNameChanged") + .unwrap() + .get(&None) + .unwrap() + .len(), 1 ); assert_eq!( From 581b77a9c55d848e2b89697d4c72e000e287be23 Mon Sep 17 00:00:00 2001 From: kvfasil Date: Fri, 10 Jul 2026 12:47:40 -0400 Subject: [PATCH 05/10] Remove DeviceEvent::Cleanup and related Thunder extension cleanup - Remove DeviceEvent::Cleanup enum variant from device_events.rs - Remove Cleanup handling from thunder_events.rs processor - Remove Cleanup handling from thunder_event_processor.rs - Remove cleanup dispatch from firebolt_gateway.rs - Remove related unit tests (test_cleanup_event_request_payload_roundtrip, test_cleanup_event_contract) - Remove unused 'log::info' import Note: Connection-level cleanup via cleanup_for_connection() remains and handles all necessary broker-layer subscription cleanup. This is sufficient for proper system cleanup without requiring the extension layer cleanup. --- core/main/src/broker/endpoint_broker.rs | 4 +- core/main/src/broker/service_broker.rs | 3 +- core/main/src/broker/thunder_broker.rs | 19 ++++---- core/main/src/firebolt/firebolt_gateway.rs | 33 +------------ core/sdk/src/api/device/device_events.rs | 47 +------------------ .../src/events/thunder_event_processor.rs | 4 -- .../src/processors/thunder_events.rs | 15 +----- 7 files changed, 18 insertions(+), 107 deletions(-) diff --git a/core/main/src/broker/endpoint_broker.rs b/core/main/src/broker/endpoint_broker.rs index 970aae465..50f5aed21 100644 --- a/core/main/src/broker/endpoint_broker.rs +++ b/core/main/src/broker/endpoint_broker.rs @@ -1070,7 +1070,7 @@ impl EndpointBrokerState { Ok(response) } RenderedRequest::BrokerRequest(request) => { - info!( + trace!( "Sending broker_request json rpc response to endpoint {:?}", request ); @@ -1176,7 +1176,7 @@ impl EndpointBrokerState { for id in &removed_ids { extn_map.remove(id); } - debug!( + trace!( "cleanup_request_maps: removed {} request_map and extension_request_map entries", removed_ids.len() ); diff --git a/core/main/src/broker/service_broker.rs b/core/main/src/broker/service_broker.rs index 50a217336..3cd05be73 100644 --- a/core/main/src/broker/service_broker.rs +++ b/core/main/src/broker/service_broker.rs @@ -19,6 +19,7 @@ use super::endpoint_broker::{ EndpointBroker, EndpointBrokerState, BROKER_CHANNEL_BUFFER_SIZE, }; use crate::state::platform_state::PlatformState; +use jsonrpsee::tracing::trace; use ripple_sdk::{ api::{gateway::rpc_gateway_api::JsonRpcApiError, observability::log_signal::LogSignal}, log::{error, info}, @@ -171,7 +172,7 @@ impl ServiceBroker { fn update_service_request(broker_request: &BrokerRequest) -> Result { let v = Self::apply_request_rule(broker_request)?; - info!("transformed request {:?}", v); + trace!("transformed request {:?}", v); // Create a ServiceMessage let mut request = ServiceMessage::new_request( diff --git a/core/main/src/broker/thunder_broker.rs b/core/main/src/broker/thunder_broker.rs index 3df89c672..aade01e4a 100644 --- a/core/main/src/broker/thunder_broker.rs +++ b/core/main/src/broker/thunder_broker.rs @@ -422,7 +422,7 @@ impl ThunderBroker { }; if !subscriptions_to_cleanup.is_empty() { - debug!( + trace!( "BrokerCleaner: unsubscribing {} subscription(s)", subscriptions_to_cleanup.len() ); @@ -444,7 +444,7 @@ impl ThunderBroker { "id": format!("{}", v.rpc.ctx.call_id) } }); - debug!( + trace!( "BrokerCleaner: sending Thunder unregister for {}.{}", callsign, method ); @@ -468,7 +468,7 @@ impl ThunderBroker { ); } } else { - debug!( + trace!( "BrokerCleaner: no subscriptions found for cleanup, skipping" ); } @@ -603,9 +603,10 @@ impl ThunderBroker { .iter() .position(|x| x.rpc.ctx.method.eq_ignore_ascii_case(method)) { - debug!( + trace!( "Removing subscription for method {} for key {}", - method, sub_key + method, + sub_key ); response = Some(v.remove(i)); } @@ -702,7 +703,7 @@ impl EndpointBroker for ThunderBroker { let mut requests = Vec::new(); let method = method.unwrap(); - debug!( + trace!( "Preparing request for method {} and callsign {} for {} subscription {}", method, callsign, @@ -715,7 +716,7 @@ impl EndpointBroker for ThunderBroker { let listen = rpc_request.rpc.is_listening(); // If there was an existing app and method combo for the same subscription just unregister that if let Some(cleanup) = self.subscribe(rpc_request) { - info!("Sending Thunder unregister for {}.{}", callsign, method); + trace!("Sending Thunder unregister for {}.{}", callsign, method); requests.push( json!({ "jsonrpc": "2.0", @@ -732,7 +733,7 @@ impl EndpointBroker for ThunderBroker { // Given unregistration is already performed by previous step just do registration if listen { - info!("Sending Thunder register for {}.{}", callsign, method); + trace!("Sending Thunder register for {}.{}", callsign, method); requests.push( json!({ "jsonrpc": "2.0", @@ -748,7 +749,7 @@ impl EndpointBroker for ThunderBroker { } } else if rpc_request.rpc.is_unlisten() { if let Some(cleanup) = self.unsubscribe(rpc_request) { - info!("Sending Thunder unregister for {}.{}", callsign, method); + trace!("Sending Thunder unregister for {}.{}", callsign, method); requests.push( json!({ "jsonrpc": "2.0", diff --git a/core/main/src/firebolt/firebolt_gateway.rs b/core/main/src/firebolt/firebolt_gateway.rs index d6993b80e..1ee8f73b9 100644 --- a/core/main/src/firebolt/firebolt_gateway.rs +++ b/core/main/src/firebolt/firebolt_gateway.rs @@ -18,7 +18,6 @@ use jsonrpsee::{core::server::rpc_module::Methods, types::TwoPointZero}; use ripple_sdk::{ api::{ - device::device_events::{DeviceEvent, DeviceEventCallback, DeviceEventRequest}, firebolt::{ fb_capabilities::JSON_RPC_STANDARD_ERROR_INVALID_PARAMS, fb_openrpc::FireboltOpenRpcMethod, @@ -33,7 +32,7 @@ use ripple_sdk::{ }, chrono::Utc, extn::extn_client_message::ExtnMessage, - log::{debug, error, info, trace, warn}, + log::{error, info, trace, warn}, serde_json::{self, Value}, service::service_message::{JsonRpcMessage as JsonRpcServiceMessage, ServiceMessage}, tokio::{self, runtime::Handle, sync::mpsc::Sender}, @@ -160,38 +159,8 @@ impl FireboltGateway { .cleanup_for_connection(&cid) .await; - // Resolve app_id from session state BEFORE clearing the session, - // because ThunderEventProcessor stores listeners by app_id (e.g. "epg"), - // not by cid (a UUID). - let app_id_for_cleanup = self - .state - .platform_state - .session_state - .get_app_id(cid.clone()); // Clear session from session_map by connection_id (the key used during registration) self.state.platform_state.session_state.clear_session(&cid); - // Send cleanup to ThunderEventProcessor (extension side) to remove - // event_map and last_event entries for this app - if let Some(app_id) = app_id_for_cleanup { - let cleanup_request = DeviceEventRequest { - event: DeviceEvent::Cleanup, - subscribe: false, - callback_type: DeviceEventCallback::FireboltAppEvent(app_id.clone()), - }; - if let Err(e) = self - .state - .platform_state - .get_client() - .send_extn_request_transient(cleanup_request) - { - warn!( - "Failed to send ThunderEventProcessor cleanup for app_id={}: {:?}", - app_id, e - ); - } - } else { - debug!("Could not resolve app_id, skipping ThunderEventProcessor cleanup"); - } } HandleRpc { request } => self.handle(request, None).await, HandleRpcForExtn { msg } => { diff --git a/core/sdk/src/api/device/device_events.rs b/core/sdk/src/api/device/device_events.rs index 52eb57e02..29858860e 100644 --- a/core/sdk/src/api/device/device_events.rs +++ b/core/sdk/src/api/device/device_events.rs @@ -39,9 +39,6 @@ pub const VOICE_GUIDANCE_SPEED_CHANGED: &str = "voiceguidance.onSpeedChanged"; pub enum DeviceEvent { InputChanged, AudioChanged, - /// Signals that an app has disconnected and its event subscriptions should be cleaned up. - /// The callback_type.get_id() carries the app_id to clean. - Cleanup, } impl FromStr for DeviceEvent { @@ -51,7 +48,6 @@ impl FromStr for DeviceEvent { match s { "device.onHdcpChanged" => Ok(Self::InputChanged), "device.onAudioChanged" => Ok(Self::AudioChanged), - "device.cleanup" => Ok(Self::Cleanup), _ => Err(()), } } @@ -95,7 +91,6 @@ impl ExtnPayloadProvider for DeviceEventRequest { match self.event { DeviceEvent::InputChanged => RippleContract::DeviceEvents(EventAdjective::Input), DeviceEvent::AudioChanged => RippleContract::DeviceEvents(EventAdjective::Audio), - DeviceEvent::Cleanup => RippleContract::DeviceEvents(EventAdjective::Input), } } @@ -112,7 +107,6 @@ mod tests { #[rstest(input, expected, case("device.onHdcpChanged", Ok(DeviceEvent::InputChanged)), - case("device.cleanup", Ok(DeviceEvent::Cleanup)), case("invalid_event", Err(())), )] fn test_from_str(input: &str, expected: Result) { @@ -120,50 +114,13 @@ mod tests { } #[test] - fn test_cleanup_event_request_payload_roundtrip() { + fn test_device_event_request_payload_roundtrip() { let request = DeviceEventRequest { - event: DeviceEvent::Cleanup, - subscribe: false, - callback_type: DeviceEventCallback::FireboltAppEvent("epg".to_string()), - }; - let payload = request.get_extn_payload(); - let recovered = DeviceEventRequest::get_from_payload(payload).unwrap(); - assert_eq!(recovered.event, DeviceEvent::Cleanup); - assert_eq!(recovered.callback_type.get_id(), "epg"); - } - - #[test] - fn test_cleanup_event_contract() { - let request = DeviceEventRequest { - event: DeviceEvent::Cleanup, - subscribe: true, - callback_type: DeviceEventCallback::FireboltAppEvent("app1".to_string()), - }; - // Cleanup routes through the same contract as Input events - assert_eq!( - request.get_contract(), - RippleContract::DeviceEvents(EventAdjective::Input) - ); - } - - #[rstest] - #[case( - DeviceEventCallback::FireboltAppEvent("app_event".to_string()), - "app_event" - )] - #[case(DeviceEventCallback::ExtnEvent, "internal")] - fn test_get_id(#[case] callback: DeviceEventCallback, #[case] expected_id: &str) { - assert_eq!(callback.get_id(), expected_id); - } - - #[test] - fn test_extn_request_device_event() { - let device_event_request = DeviceEventRequest { event: DeviceEvent::InputChanged, subscribe: true, callback_type: DeviceEventCallback::FireboltAppEvent("id".to_string()), }; let contract_type: RippleContract = RippleContract::DeviceEvents(EventAdjective::Input); - test_extn_payload_provider(device_event_request, contract_type); + test_extn_payload_provider(request, contract_type); } } diff --git a/device/thunder_ripple_sdk/src/events/thunder_event_processor.rs b/device/thunder_ripple_sdk/src/events/thunder_event_processor.rs index 9c68c75a2..52efecff9 100644 --- a/device/thunder_ripple_sdk/src/events/thunder_event_processor.rs +++ b/device/thunder_ripple_sdk/src/events/thunder_event_processor.rs @@ -93,10 +93,6 @@ impl ThunderEventMessage { value.clone(), ))) } - DeviceEvent::Cleanup => { - // Cleanup is not a real Thunder event, skip - return None; - } } } else { debug!( diff --git a/device/thunder_ripple_sdk/src/processors/thunder_events.rs b/device/thunder_ripple_sdk/src/processors/thunder_events.rs index f0128bc6b..43a62329b 100644 --- a/device/thunder_ripple_sdk/src/processors/thunder_events.rs +++ b/device/thunder_ripple_sdk/src/processors/thunder_events.rs @@ -16,7 +16,7 @@ // use ripple_sdk::{ - api::session::EventAdjective, framework::ripple_contract::RippleContract, log::info, + api::session::EventAdjective, framework::ripple_contract::RippleContract, utils::error::RippleError, }; @@ -106,19 +106,6 @@ impl ExtnRequestProcessor for ThunderOpenEventsProcessor { id.clone(), HDCPEventHandler::provide(id, callback_type), )), - DeviceEvent::Cleanup => { - // Clean up all event subscriptions for this app - let removed_events = state.event_processor.cleanup_by_app_id(&id); - if !removed_events.is_empty() { - info!( - "ThunderEventProcessor cleanup: removed {} event(s) for app {}: {:?}", - removed_events.len(), - id, - removed_events - ); - } - return true; - } } { v.await; Self::ack(state.get_client(), msg).await.is_ok() From 771c0f45ce99cc2e92a5655d6178c03cd49d1adb Mon Sep 17 00:00:00 2001 From: kvfasil Date: Fri, 10 Jul 2026 13:37:49 -0400 Subject: [PATCH 06/10] Fix non-JSON-RPC WebSocket broker cleanup keying bug Use cid (connection_id) as storage key with fallback to session_id, matching Thunder broker pattern. This ensures cleanup_for_connection(cid) can properly close brokers when WebSocket disconnects. --- core/main/src/broker/websocket_broker.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/core/main/src/broker/websocket_broker.rs b/core/main/src/broker/websocket_broker.rs index 7c31e3d2e..34f38558c 100644 --- a/core/main/src/broker/websocket_broker.rs +++ b/core/main/src/broker/websocket_broker.rs @@ -127,7 +127,7 @@ impl WebsocketBroker { }); while let Some(v) = tr.recv().await { - let id = v.get_id(); + let id = v.rpc.ctx.cid.clone().unwrap_or_else(|| v.rpc.ctx.session_id.clone()); let cleaner = WSNotificationBroker::start( v.clone(), callback.clone(), From efb2a38e4c4a0457ed40a271aa6675c1f3639e8f Mon Sep 17 00:00:00 2001 From: kvfasil Date: Fri, 10 Jul 2026 14:39:00 -0400 Subject: [PATCH 07/10] Fix: fmt Error. --- core/main/src/broker/websocket_broker.rs | 7 ++++++- 1 file changed, 6 insertions(+), 1 deletion(-) diff --git a/core/main/src/broker/websocket_broker.rs b/core/main/src/broker/websocket_broker.rs index 34f38558c..72413082e 100644 --- a/core/main/src/broker/websocket_broker.rs +++ b/core/main/src/broker/websocket_broker.rs @@ -127,7 +127,12 @@ impl WebsocketBroker { }); while let Some(v) = tr.recv().await { - let id = v.rpc.ctx.cid.clone().unwrap_or_else(|| v.rpc.ctx.session_id.clone()); + let id = v + .rpc + .ctx + .cid + .clone() + .unwrap_or_else(|| v.rpc.ctx.session_id.clone()); let cleaner = WSNotificationBroker::start( v.clone(), callback.clone(), From 37c17497559a1a5bce558533afdfe1b5a04e3dfc Mon Sep 17 00:00:00 2001 From: kvfasil Date: Fri, 10 Jul 2026 14:43:20 -0400 Subject: [PATCH 08/10] Prevent empty Vec re-insertion in Thunder broker subscribe() Only re-insert Vec into subscription_map if it's non-empty after removing matching subscription. This prevents map bloat from accumulating empty entries when requests with listen=false remove the last subscription for a key, avoiding unnecessary memory usage and cleanup iterations. --- core/main/src/broker/thunder_broker.rs | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/core/main/src/broker/thunder_broker.rs b/core/main/src/broker/thunder_broker.rs index aade01e4a..ebcd04e23 100644 --- a/core/main/src/broker/thunder_broker.rs +++ b/core/main/src/broker/thunder_broker.rs @@ -613,7 +613,10 @@ impl ThunderBroker { if listen { v.push(request.clone()); } - let _ = sub_map.insert(sub_key.clone(), v); + // Only re-insert if the Vec is non-empty to avoid map bloat from empty entries + if !v.is_empty() { + let _ = sub_map.insert(sub_key.clone(), v); + } } else if listen { let _ = sub_map.insert(sub_key.clone(), vec![request.clone()]); } From 2adc9e3063ac9bb9934f962f3d8ed8e7ef4e8ea1 Mon Sep 17 00:00:00 2001 From: kvfasil Date: Mon, 13 Jul 2026 07:47:43 -0400 Subject: [PATCH 09/10] Addressed review coments. --- core/main/src/broker/thunder_broker.rs | 5 ----- 1 file changed, 5 deletions(-) diff --git a/core/main/src/broker/thunder_broker.rs b/core/main/src/broker/thunder_broker.rs index ebcd04e23..199324741 100644 --- a/core/main/src/broker/thunder_broker.rs +++ b/core/main/src/broker/thunder_broker.rs @@ -592,13 +592,8 @@ impl ThunderBroker { let method = &request.rpc.ctx.method; let listen = request.rpc.is_listening(); let mut response = None; - // debug!( - // "Initial subscription map of {:?} sub_key {:?}", - // sub_map, sub_key - // ); if let Some(mut v) = sub_map.remove(&sub_key) { - // debug!("Subscription map after removing key {:?}", v); if let Some(i) = v .iter() .position(|x| x.rpc.ctx.method.eq_ignore_ascii_case(method)) From 54833a49eb326629ca2dde0345b70323697248cd Mon Sep 17 00:00:00 2001 From: kvfasil Date: Mon, 13 Jul 2026 16:32:28 -0400 Subject: [PATCH 10/10] fix : Updated log level for service broker. --- core/main/src/broker/endpoint_broker.rs | 2 +- core/main/src/broker/rules/rules_engine.rs | 6 ++++-- core/main/src/broker/service_broker.rs | 4 ++-- core/main/src/broker/thunder_broker.rs | 2 +- core/sdk/src/service/service_client.rs | 4 ++-- 5 files changed, 10 insertions(+), 8 deletions(-) diff --git a/core/main/src/broker/endpoint_broker.rs b/core/main/src/broker/endpoint_broker.rs index 50f5aed21..0fd0c28a9 100644 --- a/core/main/src/broker/endpoint_broker.rs +++ b/core/main/src/broker/endpoint_broker.rs @@ -1528,7 +1528,7 @@ impl BrokerOutputForwarder { .emit_debug(); let session_id = rpc_request.ctx.get_id(); if let Some(workflow_callback) = workflow_callback { - debug!("sending to workflow callback {:?}", response); + trace!("sending to workflow callback {:?}", response); LogSignal::new( "forward_response".to_string(), "sending to workflow callback".to_string(), diff --git a/core/main/src/broker/rules/rules_engine.rs b/core/main/src/broker/rules/rules_engine.rs index 24be1a24c..e0102c666 100644 --- a/core/main/src/broker/rules/rules_engine.rs +++ b/core/main/src/broker/rules/rules_engine.rs @@ -539,9 +539,11 @@ fn get_parse_ctx() -> MutexGuard<'static, ParseCtx> { } pub fn jq_compile(input: Value, filter: &str, reference: String) -> Result { - info!( + trace!( "Jq rule {} input {:?}, reference {}", - filter, input, reference + filter, + input, + reference ); let start = Utc::now().timestamp_millis(); // start out only from core filters, diff --git a/core/main/src/broker/service_broker.rs b/core/main/src/broker/service_broker.rs index 3cd05be73..006eaaab4 100644 --- a/core/main/src/broker/service_broker.rs +++ b/core/main/src/broker/service_broker.rs @@ -22,7 +22,7 @@ use crate::state::platform_state::PlatformState; use jsonrpsee::tracing::trace; use ripple_sdk::{ api::{gateway::rpc_gateway_api::JsonRpcApiError, observability::log_signal::LogSignal}, - log::{error, info}, + log::error, service::service_message::{Id, ServiceMessage}, tokio::{self, sync::mpsc}, tokio_tungstenite::tungstenite::Message, @@ -122,7 +122,7 @@ impl ServiceBroker { } let message = Message::Text(request.clone()); - info!("Sending request to service {}: {:#?}", service_id, message); + trace!("Sending request to service {}: {:#?}", service_id, message); if let Err(err) = service_sender.try_send(message) { error!( diff --git a/core/main/src/broker/thunder_broker.rs b/core/main/src/broker/thunder_broker.rs index 199324741..37dd83a75 100644 --- a/core/main/src/broker/thunder_broker.rs +++ b/core/main/src/broker/thunder_broker.rs @@ -291,7 +291,7 @@ impl ThunderBroker { }, Some(mut request) = broker_request_rx.recv() => { - debug!("Got request from receiver for broker {:?}", request); + trace!("Got request from receiver for broker {:?}", request); diagnostic_context.lock().await.replace(request.clone()); match broker_c.check_and_generate_plugin_activation_request(&request) { diff --git a/core/sdk/src/service/service_client.rs b/core/sdk/src/service/service_client.rs index 5416c94e4..eb651b505 100644 --- a/core/sdk/src/service/service_client.rs +++ b/core/sdk/src/service/service_client.rs @@ -245,7 +245,7 @@ impl ServiceClient { let handle_ws_message = |msg: Message| { if let Message::Text(message) = msg.clone() { // Service message - debug!("Received Service Message: {:#?}", message); + trace!("Received Service Message: {:#?}", message); if let Ok(sm) = serde_json::from_str::(&message) { match sm.message { JsonRpcMessage::Request(ref _json_rpc_request) => { @@ -263,7 +263,7 @@ impl ServiceClient { } } JsonRpcMessage::Notification(ref json_rpc_notification) => { - debug!( + trace!( "Received Service Notification: {:?}", json_rpc_notification, );