feat: add guarded symbol switching

src/app.rs

@@ -15,9 +15,13 @@ use std::sync::Arc;
 use std::time::Duration;
 use wasm_bindgen::JsCast;
 
+use std::time::Duration;
+
 use crate::event_utils::{EventOptions, wheel_event_options, window_event_listener_with_options};
 use crate::global_signals;
-use crate::global_state::{ensure_chart, get_chart_signal, set_chart_in_ecs, view_state};
+use crate::global_state::{
+    connection_id, domain_state, ensure_chart, get_chart_signal, set_chart_in_ecs, view_state,
+};
 use crate::{
     domain::{
         chart::Chart,
@@ -1202,6 +1206,13 @@ pub async fn start_websocket_stream(set_status: WriteSignal<String>) {
     }
 
     let interval = current_interval().get_untracked();
+    let conn_id = connection_id().get_untracked() + 1;
+    connection_id().set(conn_id);
+    domain_state().update(|ds| {
+        ds.timeframe = Duration::from_millis(interval.duration_ms());
+        ds.candles = Arc::new(Vec::new());
+        ds.indicators = Arc::new(Vec::new());
+    });
 
     let rest_client_arc =
         Arc::new(Mutex::new(BinanceWebSocketClient::new(symbol.clone(), interval)));
@@ -1214,8 +1225,11 @@ pub async fn start_websocket_stream(set_status: WriteSignal<String>) {
 
     let hist_res = {
         let client = rest_client_arc.lock().await;
-        client.fetch_historical_data(500).await
+        client.fetch_historical_data(1000).await
     };
+    if conn_id != connection_id().get_untracked() {
+        return;
+    }
     match hist_res {
         Ok(historical_candles) => {
             get_logger().info(
@@ -1225,6 +1239,10 @@ pub async fn start_websocket_stream(set_status: WriteSignal<String>) {
 
             chart.update(|ch| ch.set_historical_data(historical_candles.clone()));
             chart.with_untracked(|c| set_chart_in_ecs(&symbol, c.clone()));
+            domain_state().update(|ds| {
+                ds.candles = Arc::new(historical_candles.clone());
+                ds.indicators = Arc::new(Vec::new());
+            });
             chart.with_untracked(|c| {
                 if c.get_candle_count() > 0
                     && with_global_renderer(|r| {
@@ -1295,8 +1313,11 @@ pub async fn start_websocket_stream(set_status: WriteSignal<String>) {
     let fut = futures::future::Abortable::new(
         async move {
             let handler_handle = handle_check.clone();
+            let connection_guard = conn_id;
             let handler = move |candle: Candle| {
-                if handler_handle.is_aborted() {
+                if handler_handle.is_aborted()
+                    || connection_guard != connection_id().get_untracked()
+                {
                     return;
                 }
                 global_current_price().set(candle.ohlcv.close.value());
@@ -1313,6 +1334,11 @@ pub async fn start_websocket_stream(set_status: WriteSignal<String>) {
                 });
                 chart.with_untracked(|c| set_chart_in_ecs(&symbol, c.clone()));
                 crate::global_state::push_realtime_candle(candle.clone());
+                domain_state().update(|ds| {
+                    let mut v = (*ds.candles).clone();
+                    v.push(candle.clone());
+                    ds.candles = Arc::new(v);
+                });
 
                 let count = chart.with(|c| c.get_candle_count());
                 global_candle_count().set(count);

src/global_state.rs

@@ -35,6 +35,7 @@ pub struct Globals {
     pub line_visibility: RwSignal<crate::infrastructure::rendering::renderer::LineVisibility>,
     pub domain_state: RwSignal<DomainState>,
     pub view_state: RwSignal<ViewState>,
+    pub connection_id: RwSignal<u64>,
 }
 
 // The `OnceCell` ensures this state is created at most once on demand.
@@ -63,6 +64,7 @@ pub fn globals() -> &'static Globals {
             Arc::new(Vec::new()),
         )),
         view_state: create_rw_signal(ViewState::new(5.0, 1.0, 20.0)),
+        connection_id: create_rw_signal(0),
     })
 }
 
@@ -103,6 +105,10 @@ pub fn view_state() -> RwSignal<ViewState> {
     globals().view_state
 }
 
+pub fn connection_id() -> RwSignal<u64> {
+    globals().connection_id
+}
+
 /// Add a candle to the ECS world and process systems.
 pub fn push_realtime_candle(candle: Candle) {
     use crate::ecs::components::CandleComponent;

tests/switch_symbol_loop.rs

@@ -0,0 +1,43 @@
+use futures::future::AbortHandle;
+use leptos::*;
+use price_chart_wasm::app::{abort_other_streams, current_symbol, stream_abort_handles};
+use price_chart_wasm::domain::market_data::Symbol;
+use price_chart_wasm::global_state::domain_state;
+use std::sync::Arc;
+use wasm_bindgen::prelude::*;
+use wasm_bindgen_test::*;
+
+wasm_bindgen_test_configure!(run_in_browser);
+
+fn heap_usage() -> f64 {
+    js_sys::Reflect::get(&js_sys::global(), &JsValue::from_str("performance"))
+        .ok()
+        .and_then(|perf| js_sys::Reflect::get(&perf, &JsValue::from_str("memory")).ok())
+        .and_then(|mem| js_sys::Reflect::get(&mem, &JsValue::from_str("usedJSHeapSize")).ok())
+        .and_then(|val| val.as_f64())
+        .unwrap_or(0.0)
+}
+
+#[wasm_bindgen_test]
+fn rapid_symbol_switch_no_leak() {
+    let start = heap_usage();
+    for i in 0..20 {
+        let sym = if i % 2 == 0 { "BTCUSDT" } else { "ETHUSDT" };
+        current_symbol().set(Symbol::from(sym));
+        let (handle, _) = AbortHandle::new_pair();
+        stream_abort_handles().update(|m| {
+            m.insert(Symbol::from(sym), handle);
+        });
+        abort_other_streams(&Symbol::from(sym));
+        domain_state().update(|ds| {
+            ds.candles = Arc::new(Vec::new());
+            ds.indicators = Arc::new(Vec::new());
+        });
+    }
+    assert_eq!(stream_abort_handles().with(|m| m.len()), 1);
+    let end = heap_usage();
+    if start > 0.0 && end > 0.0 {
+        let growth = (end - start) / start;
+        assert!(growth <= 0.05);
+    }
+}