Let a chart page again after it reloads from another source

A chart kept its end-of-history flag through an automatic source change, so a new source with more history than the first load returned never got asked for older bars. A full load now marks its History response, and the chart takes that response's end of history instead of keeping the old one. Live updates and gap refreshes keep it as before.

Cargo.lock

@@ -4280,6 +4280,7 @@ name = "terminal"
 version = "0.1.1"
 dependencies = [
  "anyhow",
+ "async-trait",
  "chrono",
  "eframe",
  "egui",

app/Cargo.toml

@@ -60,6 +60,10 @@ web-sys = { version = "0.3", features = [
     "Response",
     "Storage",
     "Url",
+    "UrlSearchParams",
     "Window",
     "console",
 ] }
+
+[dev-dependencies]
+async-trait = "0.1"

app/src/app.rs

@@ -957,6 +957,7 @@ impl TerminalApp {
                     prepended,
                     from_cache,
                     exhausted,
+                    full,
                     live,
                 } => {
                     let worker = &self.worker;
@@ -964,7 +965,7 @@ impl TerminalApp {
                         && c.symbol == symbol
                         && c.tf == tf
                     {
-                        c.on_history(generation, bars, gpu, prepended, from_cache, exhausted, live, worker);
+                        c.on_history(generation, bars, gpu, prepended, from_cache, exhausted, full, live, worker);
                     }
                 }
                 Response::HistoryError { pane, generation, error } => {

app/src/app/chrome.rs

@@ -108,7 +108,7 @@ impl TerminalApp {
         p.rect_filled(chip, 9.0, theme::control());
         p.circle_filled(pos2(chip.left() + 8.0, cy), 3.0, if live { theme::up() } else { amber() });
         p.galley(pos2(chip.left() + 14.0, cy - chip_text.size().y / 2.0), chip_text, muted());
-        ui.interact(chip, Id::new("source-chip"), Sense::hover()).on_hover_text(self.provider_label());
+        ui.interact(chip, Id::new("source-chip"), Sense::hover()).on_hover_text(provider_label(&self.provider));
         x = chip.right() + 8.0;
 
         // Which account trades: DEMO (paper) or LIVE (real money, in red). Click to switch.
@@ -294,21 +294,6 @@ impl TerminalApp {
         }
     }
 
-    fn provider_label(&self) -> String {
-        let n = self.provider.to_lowercase();
-        if n == "sample" {
-            "Offline · sample data".into()
-        } else if n.contains("sample") {
-            // The browser build (tr_data::web::WebMarkets).
-            "Live crypto from Binance and Coinbase; every other market is sample data in the browser".into()
-        } else if let Some(rest) = self.provider.strip_prefix("server+") {
-            // TR_SERVER_URL is set (tr_data::remote::RemoteMarkets).
-            format!("Live · crypto from Hyperliquid perps through the Terminal server; other markets from {rest}")
-        } else {
-            format!("Live · {}", self.provider)
-        }
-    }
-
     /// The Bloomberg command line, drawn as a rounded macOS search field.
     fn command_field(&mut self, ui: &mut Ui, field: Rect) -> bool {
         let suggestions = self.suggestions();
@@ -991,3 +976,41 @@ fn nav_row(ui: &mut Ui, icon: Icon, label: &str, trailing: Option<&str>, selecte
     }
     resp
 }
+
+/// What the data-source chip says on hover, from the provider's name.
+fn provider_label(provider: &str) -> String {
+    let n = provider.to_lowercase();
+    if n == "sample" {
+        "Offline · sample data".into()
+    } else if let Some(rest) = provider.strip_prefix("server+") {
+        // A Terminal server (tr_data::remote::RemoteMarkets), in front of the browser's sources
+        // (tr_data::web::WebMarkets) or the desktop's.
+        if rest.contains("sample") {
+            "Live crypto from Hyperliquid perps through the Terminal server, or from Binance and Coinbase while \
+             it is unreachable; every other market is sample data in the browser"
+                .into()
+        } else {
+            format!("Live · crypto from Hyperliquid perps through the Terminal server; other markets from {rest}")
+        }
+    } else if n.contains("sample") {
+        // The browser build (tr_data::web::WebMarkets).
+        "Live crypto from Binance and Coinbase; every other market is sample data in the browser".into()
+    } else {
+        format!("Live · {provider}")
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::provider_label;
+
+    #[test]
+    fn the_source_tooltip_names_the_terminal_server() {
+        assert_eq!(provider_label("sample"), "Offline · sample data");
+        assert!(provider_label("binance+sample").starts_with("Live crypto from Binance"));
+        let web = provider_label("server+binance+sample");
+        assert!(web.contains("Terminal server") && web.contains("sample data in the browser"), "{web}");
+        let desktop = provider_label("server+markets");
+        assert!(desktop.contains("Terminal server") && desktop.ends_with("other markets from markets"), "{desktop}");
+    }
+}

app/src/chart_view.rs

@@ -397,6 +397,7 @@ impl ChartState {
         prepended: usize,
         from_cache: bool,
         exhausted: bool,
+        full: bool,
         live: bool,
         w: &Worker,
     ) {
@@ -421,7 +422,7 @@ impl ChartState {
         self.bars = bars;
         self.gpu = Some(gpu);
         self.paging = false;
-        self.exhausted = exhausted || self.exhausted && prepended == 0;
+        self.exhausted = end_of_history(self.exhausted, exhausted, prepended, full);
         if !from_cache {
             self.loading = false;
         }
@@ -1435,10 +1436,26 @@ fn hover_card(ui: &Ui, at: Pos2, t: &Trade) {
     }
 }
 
+/// Whether the chart has reached the start of its history after a response. A full load decides
+/// it, since it may come from another source than the bars before it. A page decides it too. Live
+/// updates and refreshes keep what the chart knew.
+fn end_of_history(was: bool, exhausted: bool, prepended: usize, full: bool) -> bool {
+    if full { exhausted } else { exhausted || was && prepended == 0 }
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
 
+    #[test]
+    fn a_full_load_from_a_new_source_can_page_again() {
+        assert!(!end_of_history(true, false, 0, true), "a reload from a source with more history");
+        assert!(end_of_history(false, true, 0, true));
+        assert!(end_of_history(true, false, 0, false), "a live update keeps the end");
+        assert!(!end_of_history(true, false, 5, false), "a page that added bars");
+        assert!(end_of_history(false, true, 0, false), "an empty page");
+    }
+
     #[test]
     fn glide_pans_linearly_and_zooms_evenly() {
         let from = Viewport { right: 100.0, bar_px: 2.0, width_px: 800.0 };

app/src/web.rs

@@ -3,6 +3,10 @@
 //!
 //! Query flags: `?sample` serves offline sample data only (like `TR_PROVIDER=sample`), and
 //! `?fps` shows the frame-time readout and repaints continuously (like `TR_FPS`).
+//!
+//! Crypto markets come from the Terminal server that `TR_SERVER_URL` named when the page was built
+//! (scripts/build-web.sh). `?server=off` turns it off, and `?server=ws://localhost:8787/ws` uses a
+//! server on this machine instead (see `tr_data::web::server_url`).
 
 use std::cell::Cell;
 
@@ -19,6 +23,13 @@ pub fn start() {
     let query = web_sys::window().and_then(|w| w.location().search().ok()).unwrap_or_default();
     let offline = has_flag(&query, "sample");
     let perf = has_flag(&query, "fps");
+    let requested = web_sys::UrlSearchParams::new_with_str(&query).ok().and_then(|p| p.get("server"));
+    let server = tr_data::web::server_url(option_env!("TR_SERVER_URL"), requested.as_deref());
+    if let Some(url) = server.as_deref().filter(|_| !offline) {
+        tracing::info!(
+            "crypto markets come from the Terminal server at {url}, or Binance and Coinbase while it's down"
+        );
+    }
     wasm_bindgen_futures::spawn_local(async move {
         let Some(canvas) = canvas() else { return fail("the page has no #terminal canvas") };
         let started = eframe::WebRunner::new()
@@ -29,7 +40,7 @@ pub fn start() {
                     // The first frame needs fonts; set them before it (set_fonts applies from the next pass).
                     cc.egui_ctx.set_fonts(crate::theme::base_fonts());
                     wasm_bindgen_futures::spawn_local(load_emoji_fonts(cc.egui_ctx.clone()));
-                    let mut app = crate::create_app(cc, tr_data::for_browser(offline), 1, t0)?;
+                    let mut app = crate::create_app(cc, tr_data::for_browser(offline, server.as_deref()), 1, t0)?;
                     if perf {
                         app.show_perf();
                     }

app/src/worker.rs

@@ -34,6 +34,9 @@ pub type PaneId = u64;
 /// Minutes between Congress disclosure checks (`TR_CONGRESS_REFRESH_MIN`; 0 disables).
 pub const CONGRESS_REFRESH_MIN: u64 = 30;
 
+/// The least time between a chart's reloads after its provider ended the live feed.
+const RELOAD_AFTER_FEED_ENDS: Duration = Duration::from_secs(5);
+
 pub enum Request {
     /// Load the brokerage portfolio through SnapTrade (`None` = the simulated sample).
     Portfolio(Option<tr_broker::Credentials>),
@@ -238,6 +241,9 @@ pub enum Response {
         prepended: usize,
         from_cache: bool,
         exhausted: bool,
+        /// A full load from the provider. Its `exhausted` replaces the chart's, which may have
+        /// come from the source the chart showed before.
+        full: bool,
         /// A live-feed update of the latest bar.
         live: bool,
     },
@@ -305,6 +311,9 @@ struct Shared {
     /// Authoritative bars per chart pane (history, paging and live ticks all merge here),
     /// tagged with the pane's load generation.
     series: std::sync::Mutex<SeriesMap>,
+    /// Counts each chart pane's reloads for a changed data source, up once as one starts and again
+    /// when it's done (odd while it runs), so a page of older bars that spans one is dropped.
+    sources: std::sync::Mutex<HashMap<PaneId, u64>>,
     #[cfg(not(target_arch = "wasm32"))]
     cache: std::sync::Mutex<Option<BarCache>>,
     db: Mutex<Db>,
@@ -367,6 +376,7 @@ impl Worker {
         let shared = Arc::new(Shared {
             provider,
             series: Default::default(),
+            sources: Default::default(),
             cache: std::sync::Mutex::new(cache),
             db: Mutex::new(db),
             out,
@@ -407,8 +417,14 @@ impl Worker {
     pub fn spawn(ctx: egui::Context, provider: Provider) -> anyhow::Result<Self> {
         let (tx, mut req_rx) = tokio::sync::mpsc::unbounded_channel::<Request>();
         let (out, rx) = mpsc::channel();
-        let shared =
-            Arc::new(Shared { provider, series: Default::default(), db: Mutex::new(Db::open_in_memory()?), out, ctx });
+        let shared = Arc::new(Shared {
+            provider,
+            series: Default::default(),
+            sources: Default::default(),
+            db: Mutex::new(Db::open_in_memory()?),
+            out,
+            ctx,
+        });
         rt::spawn(async move {
             while let Some(req) = req_rx.recv().await {
                 let s = shared.clone();
@@ -571,6 +587,7 @@ async fn handle(s: &Arc<Shared>, req: Request) {
                     prepended: 0,
                     from_cache: true,
                     exhausted: false,
+                    full: false,
                     live: false,
                 });
             }
@@ -593,9 +610,10 @@ async fn handle(s: &Arc<Shared>, req: Request) {
                         prepended: 0,
                         from_cache: false,
                         exhausted,
+                        full: true,
                         live: false,
                     });
-                    start_live(s.clone(), pane, generation, symbol, tf);
+                    start_live(s.clone(), pane, generation, symbol, tf, limit);
                 }
                 Ok(_) => s.send(Response::HistoryError { pane, generation, error: format!("no data for {symbol}") }),
                 Err(e) => s.send(Response::HistoryError { pane, generation, error: format!("{e:#}") }),
@@ -605,6 +623,12 @@ async fn handle(s: &Arc<Shared>, req: Request) {
             // Page from the worker's copy (it may hold newer live bars than the UI's).
             let base = current_series(s, pane, generation).filter(|b| !b.is_empty()).unwrap_or(existing);
             let Some(first) = base.first().copied() else { return };
+            // A page that overlaps a reload for another source is dropped. The reload answers the
+            // pane, and the chart pages again from the new source.
+            let source = source_switches(s, pane);
+            if source % 2 == 1 {
+                return;
+            }
             match s.provider.history(&symbol, tf, Some(first.ts), limit).await {
                 Ok(older) => {
                     let older: Vec<Bar> = older.into_iter().filter(|b| b.ts < first.ts).collect();
@@ -613,6 +637,9 @@ async fn handle(s: &Arc<Shared>, req: Request) {
                         let mut map = s.series.lock().expect("series lock");
                         let Some((g, cur)) = map.get_mut(&pane).filter(|(g, _)| *g == generation) else { return };
                         let _ = g;
+                        if source_switches(s, pane) != source {
+                            return;
+                        }
                         let mut merged = older;
                         merged.extend_from_slice(cur);
                         *cur = Arc::new(merged);
@@ -629,10 +656,14 @@ async fn handle(s: &Arc<Shared>, req: Request) {
                         prepended,
                         from_cache: false,
                         exhausted: prepended == 0,
+                        full: false,
                         live: false,
                     });
                 }
-                Err(e) => s.send(Response::HistoryError { pane, generation, error: format!("{e:#}") }),
+                Err(e) if source_switches(s, pane) == source => {
+                    s.send(Response::HistoryError { pane, generation, error: format!("{e:#}") })
+                }
+                Err(_) => {}
             }
         }
         Request::Indicators { pane, generation, bars, tf, specs, palette } => {
@@ -828,6 +859,10 @@ fn store_series(s: &Shared, pane: PaneId, generation: u64, bars: Arc<Vec<Bar>>)
     }
 }
 
+fn source_switches(s: &Shared, pane: PaneId) -> u64 {
+    s.sources.lock().expect("sources lock").get(&pane).copied().unwrap_or(0)
+}
+
 fn current_series(s: &Shared, pane: PaneId, generation: u64) -> Option<Arc<Vec<Bar>>> {
     s.series.lock().expect("series lock").get(&pane).filter(|(g, _)| *g == generation).map(|(_, b)| b.clone())
 }
@@ -849,16 +884,21 @@ fn apply_live_bar(bars: &mut Vec<Bar>, bar: Bar) -> Applied {
     Applied { changed, inserted: bars.len() > len && bars.last().map(|b| b.ts) == last }
 }
 
-fn start_live(s: Arc<Shared>, pane: PaneId, generation: u64, symbol: String, tf: Timeframe) {
+fn start_live(s: Arc<Shared>, pane: PaneId, generation: u64, symbol: String, tf: Timeframe, limit: usize) {
     let Some(mut feed) = s.provider.live(&symbol, tf) else { return };
     rt::spawn(async move {
+        let started = web_time::Instant::now();
         let mut tick = rt::interval(Duration::from_millis(250));
         let mut dirty = false;
         let mut inserted = false;
+        let mut ended = false;
         loop {
             tokio::select! {
                 bar = feed.recv() => {
-                    let Some(bar) = bar else { break };
+                    let Some(bar) = bar else {
+                        ended = true;
+                        break;
+                    };
                     let mut map = s.series.lock().expect("series lock");
                     match map.get_mut(&pane) {
                         Some((g, bars)) if *g == generation => {
@@ -889,6 +929,7 @@ fn start_live(s: Arc<Shared>, pane: PaneId, generation: u64, symbol: String, tf:
                         prepended: 0,
                         from_cache: false,
                         exhausted: false,
+                        full: false,
                         // A bar slotted into a gap shifts everything after it: send it as a
                         // refresh so the chart rebuilds its compare line and indicators.
                         live: !std::mem::take(&mut inserted),
@@ -897,6 +938,17 @@ fn start_live(s: Arc<Shared>, pane: PaneId, generation: u64, symbol: String, tf:
             }
         }
         tracing::debug!("live feed for pane {pane} ({symbol}) stopped");
+        // A provider ends a feed the pane still shows when the chart's source changes (the Terminal
+        // server went away or came back): reload the chart from wherever its data comes from now.
+        if ended {
+            rt::sleep(RELOAD_AFTER_FEED_ENDS.saturating_sub(started.elapsed())).await;
+            if current_series(&s, pane, generation).is_some() {
+                let switch = || *s.sources.lock().expect("sources lock").entry(pane).or_default() += 1;
+                switch();
+                handle(&s, Request::History { pane, generation, symbol, tf, limit }).await;
+                switch();
+            }
+        }
     });
 }
 
@@ -1022,6 +1074,7 @@ mod tests {
         let s = Shared {
             provider: std::sync::Arc::new(tr_data::sample::SampleProvider::at(0, 10)),
             series: Default::default(),
+            sources: Default::default(),
             cache: std::sync::Mutex::new(None),
             db: Mutex::new(Db::open_in_memory().unwrap()),
             out,
@@ -1036,6 +1089,120 @@ mod tests {
         assert!(current_series(&s, 1, 1).is_none());
     }
 
+    /// Bars whose close counts the loads. Reloads wait for `reload`, and older pages for `page`.
+    /// The first chart's live feed ends at once, the way `RemoteMarkets` ends one when the chart's
+    /// source changes.
+    #[derive(Default)]
+    struct SourceChanges {
+        loads: std::sync::atomic::AtomicUsize,
+        open: std::sync::Mutex<Vec<tokio::sync::mpsc::Sender<Bar>>>,
+        reload: tokio::sync::Notify,
+        page: tokio::sync::Notify,
+    }
+
+    #[async_trait::async_trait]
+    impl tr_data::MarketData for SourceChanges {
+        fn name(&self) -> &str {
+            "test"
+        }
+
+        async fn history(&self, _: &str, _: Timeframe, end: Option<i64>, _: usize) -> anyhow::Result<Vec<Bar>> {
+            if end.is_some() {
+                self.page.notified().await;
+                return Ok(vec![Bar::new(-60, 1.0, 1.0, 1.0, 1.0, 1.0)]);
+            }
+            let n = self.loads.fetch_add(1, std::sync::atomic::Ordering::SeqCst) + 1;
+            if n > 1 {
+                self.reload.notified().await;
+            }
+            Ok(vec![Bar::new(0, 1.0, 1.0, 1.0, n as f64, 1.0)])
+        }
+
+        fn live(&self, _: &str, _: Timeframe) -> Option<tr_data::LiveFeed> {
+            let (tx, rx) = tokio::sync::mpsc::channel(1);
+            if self.loads.load(std::sync::atomic::Ordering::SeqCst) > 1 {
+                self.open.lock().unwrap().push(tx);
+            }
+            Some(rx)
+        }
+    }
+
+    /// A worker on `SourceChanges` with pane 1 loading `BTCUSD Curncy`, and its responses.
+    async fn switching_chart() -> (Arc<Shared>, mpsc::Receiver<Response>, Arc<SourceChanges>) {
+        let (out, rx) = mpsc::channel();
+        let provider = Arc::new(SourceChanges::default());
+        let s = Arc::new(Shared {
+            provider: provider.clone(),
+            series: Default::default(),
+            sources: Default::default(),
+            cache: std::sync::Mutex::new(None),
+            db: Mutex::new(Db::open_in_memory().unwrap()),
+            out,
+            ctx: egui::Context::default(),
+        });
+        let symbol = "BTCUSD Curncy".to_string();
+        handle(&s, Request::History { pane: 1, generation: 1, symbol, tf: Timeframe::D1, limit: 10 }).await;
+        (s, rx, provider)
+    }
+
+    /// The last close of each chart update, until there are `n`.
+    async fn closes(rx: &mpsc::Receiver<Response>, n: usize) -> Vec<f64> {
+        let t0 = std::time::Instant::now();
+        let mut closes = Vec::new();
+        while closes.len() < n {
+            assert!(t0.elapsed() < Duration::from_secs(15), "no reload: {closes:?}");
+            match rx.try_recv() {
+                Ok(Response::History { generation: 1, bars, .. }) => closes.push(bars.last().unwrap().close),
+                Ok(_) => {}
+                Err(_) => tokio::time::sleep(Duration::from_millis(50)).await,
+            }
+        }
+        closes
+    }
+
+    #[tokio::test]
+    async fn charts_reload_when_the_provider_ends_their_live_feed() {
+        let t0 = std::time::Instant::now();
+        let (_s, rx, provider) = switching_chart().await;
+        provider.reload.notify_one();
+        assert_eq!(closes(&rx, 2).await, [1.0, 2.0], "the reload replaced the chart");
+        assert!(t0.elapsed() >= RELOAD_AFTER_FEED_ENDS, "not before the pause");
+        assert_eq!(provider.loads.load(std::sync::atomic::Ordering::SeqCst), 2);
+    }
+
+    #[tokio::test]
+    async fn pages_that_overlap_a_source_switch_are_dropped() {
+        let (s, rx, provider) = switching_chart().await;
+        let more = || Request::MoreHistory {
+            pane: 1,
+            generation: 1,
+            symbol: "BTCUSD Curncy".into(),
+            tf: Timeframe::D1,
+            existing: Arc::new(Vec::new()),
+            limit: 10,
+        };
+        // One page starts before the switch, and its bars arrive after it.
+        let before = tokio::spawn({
+            let (s, req) = (s.clone(), more());
+            async move { handle(&s, req).await }
+        });
+        let t0 = std::time::Instant::now();
+        while source_switches(&s, 1) != 1 {
+            assert!(t0.elapsed() < Duration::from_secs(15), "no reload");
+            tokio::time::sleep(Duration::from_millis(20)).await;
+        }
+        // Another starts while the chart reloads, and gives up at once.
+        let during = tokio::time::timeout(Duration::from_secs(1), handle(&s, more())).await;
+        assert!(during.is_ok(), "a page started during the reload went ahead");
+        provider.reload.notify_one();
+        assert_eq!(closes(&rx, 2).await, [1.0, 2.0]);
+        provider.page.notify_one();
+        before.await.unwrap();
+        let ts: Vec<i64> = current_series(&s, 1, 1).unwrap().iter().map(|b| b.ts).collect();
+        assert_eq!(ts, [0], "the old source's page joined the new source's bars");
+        assert!(rx.try_iter().all(|r| !matches!(r, Response::History { prepended: 1.., .. })));
+    }
+
     #[test]
     fn join_all_runs_everything() {
         let rt = tokio::runtime::Builder::new_current_thread().enable_time().build().unwrap();

crates/data/Cargo.toml

@@ -6,6 +6,7 @@ edition.workspace = true
 
 [dependencies]
 tr-core.workspace = true
+tr-proto.workspace = true
 anyhow.workspace = true
 chrono.workspace = true
 serde.workspace = true
@@ -17,7 +18,6 @@ futures = "0.3"
 web-time = "1"
 
 [target.'cfg(not(target_arch = "wasm32"))'.dependencies]
-tr-proto.workspace = true
 tokio.workspace = true
 rusqlite.workspace = true
 tokio-tungstenite = { version = "0.30", default-features = false, features = ["connect", "rustls-tls-webpki-roots"] }

crates/data/src/lib.rs

@@ -18,7 +18,8 @@
 //!
 //! The browser build (`wasm32`) can only reach sources that allow cross-origin requests:
 //! [`for_browser`] serves live crypto through [`web::WebMarkets`] and sample data for the
-//! rest. It has no SQLite bar cache.
+//! rest. It has no SQLite bar cache. When the page was built with a Terminal server URL
+//! ([`web::server_url`]), crypto comes through the server first, as on the desktop.
 
 #[cfg(not(target_arch = "wasm32"))]
 pub mod alpaca;
@@ -27,7 +28,6 @@ pub mod cache;
 pub mod crypto;
 #[cfg(not(target_arch = "wasm32"))]
 pub mod markets;
-#[cfg(not(target_arch = "wasm32"))]
 pub mod remote;
 pub mod rss;
 pub mod rt;
@@ -435,11 +435,18 @@ pub fn from_env() -> Provider {
 }
 
 /// The browser build's provider: sample data only when `offline`, otherwise live crypto
-/// with sample data for everything else.
-pub fn for_browser(offline: bool) -> Provider {
+/// with sample data for everything else. With a Terminal `server` URL, crypto comes through
+/// that server, and from Binance and Coinbase whenever the server can't serve it.
+pub fn for_browser(offline: bool, server: Option<&str>) -> Provider {
     if !offline {
         match web::WebMarkets::new() {
-            Ok(m) => return Arc::new(Fallback::new(m)),
+            Ok(m) => {
+                let m = Fallback::new(m);
+                return match server {
+                    Some(url) => Arc::new(remote::RemoteMarkets::new(url, m)),
+                    None => Arc::new(m),
+                };
+            }
             Err(e) => tracing::warn!("crypto sources unavailable, using sample data: {e:#}"),
         }
     }

crates/data/src/remote.rs

@@ -5,23 +5,24 @@
 //! reconnects with backoff, resubscribes everything (each subscription then gets a fresh
 //! `Snapshot`), checks each subscription's `seq` and asks for a `Resync` on a gap. A
 //! subscription the server gave up on because of its upstream is subscribed again, with backoff.
+//! It runs on either target: tokio on the desktop, the page's event loop in the browser build.
 
 use std::collections::{HashMap, HashSet};
 use std::sync::atomic::{AtomicU32, Ordering};
 use std::sync::{Arc, Mutex, OnceLock};
-use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};
+use std::time::Duration;
 
 use anyhow::{Context, Result, anyhow, bail};
 use async_trait::async_trait;
-use futures::{SinkExt, StreamExt};
 use tokio::sync::{mpsc, oneshot, watch};
-use tokio_tungstenite::tungstenite::Message;
 use tr_core::{AssetClass, Bar, Timeframe};
 use tr_proto::{
     ApiError, BarDelta, ClientInfo, ClientMsg, Encoding, ErrorCode, Frame, MarketKey, PROTOCOL_VERSION, ReqId,
     ResyncReason, Rpc, RpcResult, ServerMsg, SubId, TopicData, TopicDelta, TopicReq, TopicStatus, UnavailableReason,
 };
+use web_time::{Instant, SystemTime, UNIX_EPOCH};
 
+use crate::ws::{Message, Socket};
 use crate::{LiveFeed, MarketData, NewsItem, OptionChain, OrderBook, Quote, SearchHit, YieldCurves, rt, security};
 
 /// What a subscription receives.
@@ -49,7 +50,12 @@ pub struct Update {
 #[derive(Debug, Clone, PartialEq, Eq)]
 pub enum ConnState {
     Connecting,
-    Up { session: u64 },
+    /// Connected. The server numbers the `session`; `connection` counts this client's connections
+    /// from 1.
+    Up {
+        session: u64,
+        connection: u64,
+    },
     Down(String),
 }
 
@@ -59,7 +65,8 @@ enum Cmd {
     Request { id: ReqId, req: Rpc, reply: oneshot::Sender<Result<RpcResult, ApiError>> },
 }
 
-/// A shared connection to one server. Cheap to clone. Create it inside a tokio runtime.
+/// A shared connection to one server. Cheap to clone. On the desktop, create it inside a tokio
+/// runtime.
 #[derive(Clone)]
 pub struct Client {
     cmd: mpsc::UnboundedSender<Cmd>,
@@ -103,8 +110,8 @@ pub fn normalize_url(url: &str) -> String {
 
 fn to_message(f: Frame) -> Message {
     match f {
-        Frame::Binary(b) => Message::Binary(b.into()),
-        Frame::Text(t) => Message::Text(t.into()),
+        Frame::Binary(b) => Message::Binary(b),
+        Frame::Text(t) => Message::Text(t),
     }
 }
 
@@ -118,7 +125,7 @@ impl Client {
         let (cmd, rx) = mpsc::unbounded_channel();
         let (state_tx, state) = watch::channel(ConnState::Connecting);
         let info = ClientInfo { name: name.into(), version: env!("CARGO_PKG_VERSION").into() };
-        tokio::spawn(run(normalize_url(url), encoding, info, rx, state_tx));
+        rt::spawn(run(normalize_url(url), encoding, info, rx, state_tx));
         Client { cmd, next_id: Arc::new(AtomicU32::new(1)), state }
     }
 
@@ -139,7 +146,7 @@ impl Client {
                 }
             }
         };
-        tokio::time::timeout(timeout, up).await.is_ok()
+        rt::timeout(timeout, up).await.is_ok()
     }
 
     pub fn subscribe(&self, topic: TopicReq) -> Subscription {
@@ -158,7 +165,7 @@ impl Client {
         let id = self.next_id.fetch_add(1, Ordering::Relaxed);
         let (reply, rx) = oneshot::channel();
         self.cmd.send(Cmd::Request { id, req, reply }).map_err(|_| err("client stopped"))?;
-        match tokio::time::timeout(Duration::from_secs(30), rx).await {
+        match rt::timeout(Duration::from_secs(30), rx).await {
             Ok(Ok(r)) => r,
             Ok(Err(_)) => Err(err("connection lost")),
             Err(_) => Err(err("request timed out")),
@@ -189,21 +196,19 @@ impl SubState {
     }
 }
 
-type Ws = tokio_tungstenite::WebSocketStream<tokio_tungstenite::MaybeTlsStream<tokio::net::TcpStream>>;
-
-async fn hello(url: &str, enc: Encoding, info: &ClientInfo) -> Result<(Ws, u64)> {
-    let (mut ws, _) = tokio::time::timeout(Duration::from_secs(10), tokio_tungstenite::connect_async(url))
+async fn hello(url: &str, enc: Encoding, info: &ClientInfo) -> Result<(Socket, u64)> {
+    let mut ws = rt::timeout(Duration::from_secs(10), Socket::connect(url))
         .await
         .context("connect timed out")?
         .context("connect")?;
     let hello =
         ClientMsg::Hello { version: PROTOCOL_VERSION, encoding: enc, client: info.clone(), token: None, resume: None };
-    ws.send(to_message(tr_proto::encode(&hello, enc)?)).await?;
-    let welcome = tokio::time::timeout(Duration::from_secs(10), async {
-        while let Some(m) = ws.next().await {
+    ws.send_message(to_message(tr_proto::encode(&hello, enc)?)).await?;
+    let welcome = rt::timeout(Duration::from_secs(10), async {
+        while let Some(m) = ws.next_message().await {
             match decode(&m?) {
-                Some(Ok(ServerMsg::Welcome { session, .. })) => return Ok(session.0),
-                Some(Ok(ServerMsg::Error { code, message })) => bail!("{code:?}: {message}"),
+                Ok(ServerMsg::Welcome { session, .. }) => return Ok(session.0),
+                Ok(ServerMsg::Error { code, message }) => bail!("{code:?}: {message}"),
                 _ => {}
             }
         }
@@ -214,11 +219,10 @@ async fn hello(url: &str, enc: Encoding, info: &ClientInfo) -> Result<(Ws, u64)>
     Ok((ws, welcome))
 }
 
-fn decode(m: &Message) -> Option<Result<ServerMsg, tr_proto::CodecError>> {
+fn decode(m: &Message) -> Result<ServerMsg, tr_proto::CodecError> {
     match m {
-        Message::Binary(b) => Some(tr_proto::decode_binary(b)),
-        Message::Text(t) => Some(tr_proto::decode_text(t.as_str())),
-        _ => None,
+        Message::Binary(b) => tr_proto::decode_binary(b),
+        Message::Text(t) => tr_proto::decode_text(t),
     }
 }
 
@@ -231,11 +235,13 @@ async fn run(
 ) {
     let mut subs: HashMap<SubId, SubState> = HashMap::new();
     let mut backoff = Duration::from_millis(500);
+    let mut connection = 0;
     loop {
         state.send_replace(ConnState::Connecting);
         match hello(&url, enc, &info).await {
             Ok((ws, session)) => {
-                state.send_replace(ConnState::Up { session });
+                connection += 1;
+                state.send_replace(ConnState::Up { session, connection });
                 backoff = Duration::from_millis(500);
                 if !connected(ws, enc, &mut subs, &mut cmds).await {
                     return;
@@ -255,7 +261,7 @@ async fn run(
         }
         // Back off, still taking commands: subscriptions wait for the next connection,
         // requests fail now.
-        let sleep = tokio::time::sleep(backoff);
+        let sleep = rt::sleep(backoff);
         tokio::pin!(sleep);
         loop {
             tokio::select! {
@@ -280,37 +286,35 @@ async fn run(
 
 /// Serve one connection. Returns false once every handle to the client is gone.
 async fn connected(
-    ws: Ws,
+    mut ws: Socket,
     enc: Encoding,
     subs: &mut HashMap<SubId, SubState>,
     cmds: &mut mpsc::UnboundedReceiver<Cmd>,
 ) -> bool {
-    let (mut tx, mut rx) = ws.split();
     let mut pending: HashMap<ReqId, oneshot::Sender<Result<RpcResult, ApiError>>> = HashMap::new();
     let mut queue: Vec<ClientMsg> =
         subs.iter().map(|(id, s)| ClientMsg::Subscribe { id: *id, topic: s.topic.clone() }).collect();
-    let mut ping = tokio::time::interval(Duration::from_secs(15));
+    let mut ping = rt::interval(Duration::from_secs(15));
     loop {
         for m in queue.drain(..) {
             let Ok(f) = tr_proto::encode(&m, enc) else { continue };
-            if tx.send(to_message(f)).await.is_err() {
+            if ws.send_message(to_message(f)).await.is_err() {
                 return true;
             }
         }
         let retry = subs.values().filter_map(|s| s.retry_at).min();
         tokio::select! {
-            m = rx.next() => {
+            m = ws.next_message() => {
                 let received = SystemTime::now();
                 let Some(Ok(m)) = m else { return true };
                 match decode(&m) {
-                    Some(Ok(msg)) => route(msg, received, subs, &mut pending, &mut queue),
-                    Some(Err(e)) => tracing::debug!("terminal server sent an unreadable frame: {e}"),
-                    None => {}
+                    Ok(msg) => route(msg, received, subs, &mut pending, &mut queue),
+                    Err(e) => tracing::debug!("terminal server sent an unreadable frame: {e}"),
                 }
             }
             cmd = cmds.recv() => match cmd {
                 None => {
-                    let _ = tx.close().await;
+                    ws.close().await;
                     return false;
                 }
                 Some(Cmd::Subscribe { id, topic, tx: sub }) => {
@@ -328,7 +332,7 @@ async fn connected(
                 }
             },
             _ = ping.tick() => queue.push(ClientMsg::Ping { t: now_ms() }),
-            _ = async { tokio::time::sleep_until(retry.expect("guarded").into()).await }, if retry.is_some() => {
+            _ = async { rt::sleep(retry.expect("guarded").saturating_duration_since(Instant::now())).await }, if retry.is_some() => {
                 resubscribe_due(subs, &mut queue);
             }
         }
@@ -428,6 +432,9 @@ pub struct RemoteMarkets<P> {
     unknown: Mutex<HashSet<String>>,
     /// Markets whose latest history came from the server, so paging and the live feed must too.
     served: Mutex<HashSet<String>>,
+    /// Markets the server serves whose latest history came from `inner` instead, with the
+    /// connection that was up then (0 for none). A newer connection moves their charts back.
+    fallbacks: Mutex<HashMap<String, u64>>,
     books: Arc<Mutex<HashMap<String, BookFeed>>>,
     sparks: Mutex<HashMap<String, (Instant, Vec<f64>)>>,
 }
@@ -444,11 +451,13 @@ impl<P: MarketData + 'static> RemoteMarkets<P> {
             served: Mutex::default(),
             books: Arc::default(),
             sparks: Mutex::default(),
+            fallbacks: Mutex::default(),
         }
     }
 
     fn client(&self) -> &Client {
-        self.client.get_or_init(|| Client::connect(&self.url, Encoding::Postcard, "terminal-desktop"))
+        let name = if cfg!(target_arch = "wasm32") { "terminal-web" } else { "terminal-desktop" };
+        self.client.get_or_init(|| Client::connect(&self.url, Encoding::Postcard, name))
     }
 
     fn routed(&self, symbol: &str) -> bool {
@@ -502,7 +511,7 @@ impl<P: MarketData + 'static> RemoteMarkets<P> {
         }
         let topic = TopicReq::Quotes { keys: keys.iter().map(MarketKey::new).collect(), max_hz: 1 };
         let mut sub = client.subscribe(topic);
-        let quotes = tokio::time::timeout(Duration::from_secs(5), async {
+        let quotes = rt::timeout(Duration::from_secs(5), async {
             while let Some(u) = sub.next().await {
                 match u.event {
                     Event::Snapshot(TopicData::Quotes(q)) => return Ok(q),
@@ -548,7 +557,7 @@ impl<P: MarketData + 'static> RemoteMarkets<P> {
                 }
             }
         };
-        let got = tokio::time::timeout(Duration::from_secs(3), latest.wait_for(Option::is_some))
+        let got = rt::timeout(Duration::from_secs(3), latest.wait_for(Option::is_some))
             .await
             .context("no book from the Terminal server")?
             .map_err(|_| anyhow!("book feed stopped"))?
@@ -569,7 +578,7 @@ async fn book_feed(
     used: Arc<Mutex<Instant>>,
     books: Arc<Mutex<HashMap<String, BookFeed>>>,
 ) {
-    let mut check = tokio::time::interval(Duration::from_secs(5));
+    let mut check = rt::interval(Duration::from_secs(5));
     loop {
         tokio::select! {
             u = sub.next() => {
@@ -580,6 +589,11 @@ async fn book_feed(
                         tx.send_replace(Some(Err(detail)));
                         continue;
                     }
+                    // The last book goes stale while the connection is down; the next snapshot replaces it.
+                    Event::Reconnecting => {
+                        tx.send_replace(Some(Err("the Terminal server is not reachable".into())));
+                        continue;
+                    }
                     _ => continue,
                 };
                 let level = |l: &tr_proto::Level| (l.px, l.sz);
@@ -611,19 +625,24 @@ async fn history_since(client: &Client, key: &MarketKey, tf: Timeframe, from: i6
 
 type Fetch = std::pin::Pin<Box<dyn Future<Output = Result<Vec<Bar>, ApiError>> + Send>>;
 
+/// How long a chart's live feed from the server waits for a dropped connection to come back
+/// before it ends, so the worker reloads the chart from the fallback.
+const OUTAGE_GRACE: Duration = Duration::from_secs(10);
+
 /// The server's bar stream, onto the worker's live feed. Live bars go straight through. After an
 /// outage the stream picks up at the current bar, so the bars from the last one sent up to it
 /// come from history: one request at a time per feed, retried with backoff, every outstanding
-/// gap merged into one range. The worker slots those bars into the gap. Everything stops when
-/// the worker drops the feed.
+/// gap merged into one range. The worker slots those bars into the gap. A connection that stays
+/// down past `OUTAGE_GRACE` ends the feed instead. Everything stops when the worker drops the feed.
 async fn live_bridge(client: Client, key: MarketKey, tf: Timeframe, mut sub: Subscription, tx: mpsc::Sender<Bar>) {
-    // The newest bar sent on, and whether the stream broke since.
+    // The newest bar sent on, whether the stream broke since, and when the connection dropped.
     let mut last: Option<i64> = None;
     let mut broken = false;
+    let mut away: Option<Instant> = None;
     // Bars still to fetch, `[from, until)`, and the range the request in flight covers.
     let mut pending: Option<(i64, i64)> = None;
     let mut fetching: Option<((i64, i64), Fetch)> = None;
-    let mut retry_at: Option<tokio::time::Instant> = None;
+    let mut retry_at: Option<Instant> = None;
     let mut wait = Duration::from_secs(1);
     let union = |a: Option<(i64, i64)>, b: (i64, i64)| Some(a.map_or(b, |a| (a.0.min(b.0), a.1.max(b.1))));
     loop {
@@ -634,15 +653,25 @@ async fn live_bridge(client: Client, key: MarketKey, tf: Timeframe, mut sub: Sub
             let (client, key) = (client.clone(), key.clone());
             fetching = Some((range, Box::pin(async move { history_since(&client, &key, tf, range.0).await })));
         }
+        let retry_in = retry_at.map(|t| t.saturating_duration_since(Instant::now()));
+        let give_up_in = away.map(|t| OUTAGE_GRACE.saturating_sub(t.elapsed()));
         tokio::select! {
             u = sub.next() => {
                 let Some(u) = u else { return };
                 let bars = match u.event {
-                    Event::Resync(_) | Event::Reconnecting | Event::Resubscribing => {
+                    Event::Resync(_) | Event::Resubscribing => {
                         broken = true;
                         continue;
                     }
-                    Event::Snapshot(TopicData::Bars(b)) => b,
+                    Event::Reconnecting => {
+                        broken = true;
+                        away.get_or_insert_with(Instant::now);
+                        continue;
+                    }
+                    Event::Snapshot(TopicData::Bars(b)) => {
+                        away = None;
+                        b
+                    }
                     // A `Closed` bar carries the bar's final values; merging it into the forming
                     // bar the worker holds gives the same bar (the server's forming bars are
                     // full states, never partial ones).
@@ -677,14 +706,49 @@ async fn live_bridge(client: Client, key: MarketKey, tf: Timeframe, mut sub: Sub
                     Err(e) => {
                         tracing::debug!("terminal server, backfilling {key}: {e}; retrying in {wait:?}");
                         pending = union(pending, range);
-                        retry_at = Some(tokio::time::Instant::now() + wait);
+                        retry_at = Some(Instant::now() + wait);
                         wait = (wait * 2).min(Duration::from_secs(30));
                     }
                 }
             }
-            _ = async { tokio::time::sleep_until(retry_at.expect("guarded")).await }, if retry_at.is_some() => {
+            _ = async { rt::sleep(retry_in.expect("guarded")).await }, if retry_in.is_some() => {
                 retry_at = None;
             }
+            _ = async { rt::sleep(give_up_in.expect("guarded")).await }, if give_up_in.is_some() => {
+                tracing::debug!("terminal server still unreachable; ending the {key} {tf} feed");
+                return;
+            }
+            _ = tx.closed() => return,
+        }
+    }
+}
+
+/// A fallback's live bars for a market the server serves, passed on until a connection newer than
+/// `after` is up, one the chart didn't fall back on. Then the feed ends, so the worker reloads the
+/// chart from the server.
+async fn until_server_up(mut state: watch::Receiver<ConnState>, after: u64, mut feed: LiveFeed, tx: mpsc::Sender<Bar>) {
+    let newer = |s: &ConnState| matches!(s, ConnState::Up { connection, .. } if *connection > after);
+    if newer(&state.borrow_and_update()) {
+        return;
+    }
+    let mut watching = true;
+    loop {
+        tokio::select! {
+            b = feed.recv() => {
+                let Some(b) = b else { return };
+                if tx.send(b).await.is_err() {
+                    return;
+                }
+            }
+            changed = state.changed(), if watching => {
+                if changed.is_err() {
+                    watching = false;
+                    continue;
+                }
+                if newer(&state.borrow_and_update()) {
+                    return;
+                }
+            }
             _ = tx.closed() => return,
         }
     }
@@ -714,6 +778,11 @@ impl<P: MarketData + 'static> MarketData for RemoteMarkets<P> {
                 Ok(_) => {}
                 Err(e) => self.note(symbol, &e),
             }
+            let connection = match self.client().state() {
+                ConnState::Up { connection, .. } => connection,
+                _ => 0,
+            };
+            self.fallbacks.lock().expect("fallbacks lock").insert(symbol.to_string(), connection);
         }
         self.served.lock().expect("served lock").remove(symbol);
         self.inner.history(symbol, tf, end, limit).await
@@ -746,7 +815,14 @@ impl<P: MarketData + 'static> MarketData for RemoteMarkets<P> {
 
     fn live(&self, symbol: &str, tf: Timeframe) -> Option<LiveFeed> {
         if !self.served(symbol) {
-            return self.inner.live(symbol, tf);
+            let feed = self.inner.live(symbol, tf)?;
+            if !self.routed(symbol) {
+                return Some(feed);
+            }
+            let after = self.fallbacks.lock().expect("fallbacks lock").get(symbol).copied().unwrap_or(0);
+            let (tx, rx) = mpsc::channel(256);
+            rt::spawn(until_server_up(self.client().state.clone(), after, feed, tx));
+            return Some(rx);
         }
         let (tx, rx) = mpsc::channel(256);
         let client = self.client().clone();
@@ -860,13 +936,15 @@ mod tests {
     /// `Unavailable { Upstream }`, then nothing. Like the real one, it refuses an id that's still
     /// subscribed, and serves a fresh subscription.
     async fn older_gateway(listener: tokio::net::TcpListener) {
+        use futures::{SinkExt, StreamExt};
+        use tokio_tungstenite::tungstenite::Message as WsMessage;
         use tr_proto::{Limits, SessionId};
         let (tcp, _) = listener.accept().await.unwrap();
         let mut ws = tokio_tungstenite::accept_async(tcp).await.unwrap();
         let mut subscribed = HashSet::new();
         let mut gave_up = false;
         while let Some(Ok(m)) = ws.next().await {
-            let Message::Binary(b) = m else { continue };
+            let WsMessage::Binary(b) = m else { continue };
             let reply = match tr_proto::decode_binary::<ClientMsg>(&b).unwrap() {
                 ClientMsg::Hello { .. } => ServerMsg::Welcome {
                     version: PROTOCOL_VERSION,
@@ -897,7 +975,8 @@ mod tests {
                 }
                 _ => continue,
             };
-            ws.send(to_message(tr_proto::encode(&reply, Encoding::Postcard).unwrap())).await.unwrap();
+            let Frame::Binary(f) = tr_proto::encode(&reply, Encoding::Postcard).unwrap() else { unreachable!() };
+            ws.send(WsMessage::Binary(f.into())).await.unwrap();
         }
     }
 

crates/data/src/rt.rs

@@ -37,16 +37,51 @@ pub async fn sleep(d: Duration) {
     tokio::time::sleep(d).await;
 }
 
-/// Resolves after `d`, through `setTimeout`.
+/// Resolves after `d`, through a `setTimeout` set on the first poll and cleared if the sleep is
+/// dropped before it fires.
 #[cfg(target_arch = "wasm32")]
-pub async fn sleep(d: Duration) {
-    let ms = d.as_millis().min(i32::MAX as u128) as i32;
-    let promise = js_sys::Promise::new(&mut |resolve, _| {
-        if let Some(w) = web_sys::window() {
-            let _ = w.set_timeout_with_callback_and_timeout_and_arguments_0(&resolve, ms);
+pub fn sleep(d: Duration) -> impl Future<Output = ()> + Send {
+    struct Timer(Option<i32>);
+    impl Drop for Timer {
+        fn drop(&mut self) {
+            if let (Some(id), Some(w)) = (self.0, web_sys::window()) {
+                w.clear_timeout_with_handle(id);
+            }
         }
-    });
-    let _ = wasm_bindgen_futures::JsFuture::from(promise).await;
+    }
+    let ms = d.as_millis().min(i32::MAX as u128) as i32;
+    // A promise isn't `Send`. The web build has one thread, so wrapping it is sound, and the provider
+    // traits' `Send` futures can then wait on it.
+    send_wrapper::SendWrapper::new(async move {
+        let mut timer = Timer(None);
+        let promise = js_sys::Promise::new(&mut |resolve, _| {
+            timer.0 = web_sys::window()
+                .and_then(|w| w.set_timeout_with_callback_and_timeout_and_arguments_0(&resolve, ms).ok());
+        });
+        let _ = wasm_bindgen_futures::JsFuture::from(promise).await;
+        timer.0 = None; // it fired
+    })
+}
+
+/// `fut` ran past its deadline.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub struct Elapsed;
+
+impl std::fmt::Display for Elapsed {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.write_str("deadline has elapsed")
+    }
+}
+
+impl std::error::Error for Elapsed {}
+
+/// Run `fut` for at most `d`.
+pub async fn timeout<F: Future>(d: Duration, fut: F) -> Result<F::Output, Elapsed> {
+    let (fut, deadline) = (std::pin::pin!(fut), std::pin::pin!(sleep(d)));
+    match futures::future::select(fut, deadline).await {
+        futures::future::Either::Left((out, _)) => Ok(out),
+        futures::future::Either::Right(_) => Err(Elapsed),
+    }
 }
 
 /// A periodic timer: the first tick completes at once, later ones every `period`. A late
@@ -86,4 +121,12 @@ mod tests {
         assert!(t0.elapsed() >= Duration::from_millis(40));
         assert_eq!(blocking(|| 6 * 7).await, Some(42));
     }
+
+    #[tokio::test]
+    async fn timeout_stops_at_the_deadline() {
+        assert_eq!(timeout(Duration::from_secs(5), async { 7 }).await, Ok(7));
+        let t0 = Instant::now();
+        assert_eq!(timeout(Duration::from_millis(20), std::future::pending::<()>()).await, Err(Elapsed));
+        assert!(t0.elapsed() >= Duration::from_millis(20));
+    }
 }

crates/data/src/web.rs

@@ -5,6 +5,10 @@
 //! Treasury.gov, Alpaca and CoinDesk do not, so every other asset class comes from the
 //! sample generator and there are no headlines or network search.
 //!
+//! When the page has a Terminal server ([`server_url`]), [`crate::for_browser`] puts
+//! [`crate::remote::RemoteMarkets`] in front of these, so crypto comes from the server and
+//! from Binance and Coinbase only while the server can't serve it.
+//!
 //! TODO(web proxy): route equities, FX, indices, futures, Treasuries and news through a
 //! same-origin proxy to those sources, then use [`crate::markets::Markets`] here instead.
 
@@ -27,6 +31,38 @@ impl WebMarkets {
     }
 }
 
+/// The Terminal server a page uses for crypto markets: `built`, the URL set when the page was
+/// built, unless the page's `?server=` parameter (`requested`) turns it off (`off`) or names a
+/// server on the same machine, such as `ws://localhost:8787/ws`. Other servers are refused, so a
+/// link can't point someone's page at a server that isn't ours.
+pub fn server_url(built: Option<&str>, requested: Option<&str>) -> Option<String> {
+    let built = built.map(str::trim).filter(|u| !u.is_empty()).map(str::to_string);
+    let Some(requested) = requested.map(str::trim) else { return built };
+    if requested.eq_ignore_ascii_case("off") {
+        return None;
+    }
+    if is_local(requested) {
+        return Some(requested.to_string());
+    }
+    tracing::warn!("?server= takes a server on this machine (ws://localhost:8787/ws) or off; ignoring it");
+    built
+}
+
+/// A `ws://` or `wss://` URL whose host is this machine. Anything unusual (user info, other
+/// spellings of the address) counts as not local.
+fn is_local(url: &str) -> bool {
+    let Some(rest) = url.strip_prefix("ws://").or_else(|| url.strip_prefix("wss://")) else { return false };
+    let authority = rest.split(['/', '?', '#']).next().unwrap_or_default();
+    if authority.contains('@') {
+        return false;
+    }
+    let host = match authority.strip_prefix('[') {
+        Some(v6) => v6.split(']').next().unwrap_or_default(),
+        None => authority.split(':').next().unwrap_or_default(),
+    };
+    host.eq_ignore_ascii_case("localhost") || host == "127.0.0.1" || host == "::1"
+}
+
 fn is_crypto(symbol: &str) -> bool {
     security(symbol).class == AssetClass::Crypto
 }
@@ -93,4 +129,35 @@ mod tests {
         assert_eq!(m.yield_curves().await.unwrap().current.points.len(), 11);
         assert!(is_crypto("BTCUSD Curncy") && !is_crypto("US10Y Govt"));
     }
+
+    #[test]
+    fn server_url_from_the_build_or_a_local_override() {
+        let prod = Some("wss://terminal-gateway.style.dev/ws");
+        assert_eq!(server_url(prod, None).as_deref(), prod);
+        assert_eq!(server_url(Some(" "), None), None, "an empty build setting means no server");
+        assert_eq!(server_url(None, None), None);
+        assert_eq!(server_url(prod, Some("off")), None);
+        for local in ["ws://localhost:8787/ws", "ws://127.0.0.1:9", "wss://[::1]:8787/ws", "ws://LOCALHOST/ws"] {
+            assert_eq!(server_url(prod, Some(local)).as_deref(), Some(local));
+            assert_eq!(server_url(None, Some(local)).as_deref(), Some(local));
+        }
+        for refused in [
+            "wss://evil.example/ws",
+            "ws://localhost.evil.example/ws",
+            "ws://[email protected]/ws",
+            "ws://evil.example:80@localhost/ws",
+            "http://localhost:8787/ws",
+            "",
+        ] {
+            assert_eq!(server_url(prod, Some(refused)).as_deref(), prod, "{refused}");
+            assert_eq!(server_url(None, Some(refused)), None, "{refused}");
+        }
+    }
+
+    #[tokio::test]
+    async fn the_server_goes_in_front_of_the_browser_sources() {
+        assert_eq!(crate::for_browser(false, None).name(), "binance+sample");
+        assert_eq!(crate::for_browser(false, Some("ws://127.0.0.1:9/ws")).name(), "server+binance+sample");
+        assert_eq!(crate::for_browser(true, Some("ws://127.0.0.1:9/ws")).name(), "sample");
+    }
 }

crates/data/src/ws.rs

@@ -1,18 +1,45 @@
-//! A text-message websocket client: tokio-tungstenite on the desktop, the browser's
-//! `WebSocket` on the web.
+//! A websocket client: tokio-tungstenite on the desktop, the browser's `WebSocket` on the web.
+
+use anyhow::Result;
 
 #[cfg(not(target_arch = "wasm32"))]
 pub use native::Socket;
 #[cfg(target_arch = "wasm32")]
 pub use web::Socket;
 
+/// A data message. Control frames stay inside the socket.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub enum Message {
+    Text(String),
+    Binary(Vec<u8>),
+}
+
+impl Socket {
+    pub async fn send(&mut self, text: String) -> Result<()> {
+        self.send_message(Message::Text(text)).await
+    }
+
+    /// The next text message, skipping binary ones; `None` once the server closes the connection.
+    pub async fn next(&mut self) -> Option<Result<String>> {
+        loop {
+            match self.next_message().await? {
+                Ok(Message::Text(t)) => return Some(Ok(t)),
+                Ok(Message::Binary(_)) => {}
+                Err(e) => return Some(Err(e)),
+            }
+        }
+    }
+}
+
 #[cfg(not(target_arch = "wasm32"))]
 mod native {
     use anyhow::{Context, Result};
     use futures::{SinkExt, StreamExt};
-    use tokio_tungstenite::tungstenite::Message;
+    use tokio_tungstenite::tungstenite::Message as WsMessage;
     use tokio_tungstenite::{MaybeTlsStream, WebSocketStream};
 
+    use super::Message;
+
     pub struct Socket(WebSocketStream<MaybeTlsStream<tokio::net::TcpStream>>);
 
     impl Socket {
@@ -21,22 +48,32 @@ mod native {
             Ok(Self(ws))
         }
 
-        pub async fn send(&mut self, text: String) -> Result<()> {
-            Ok(self.0.send(Message::Text(text.into())).await?)
+        pub async fn send_message(&mut self, m: Message) -> Result<()> {
+            let m = match m {
+                Message::Text(t) => WsMessage::Text(t.into()),
+                Message::Binary(b) => WsMessage::Binary(b.into()),
+            };
+            Ok(self.0.send(m).await?)
         }
 
-        /// The next text message; `None` once the server closes the connection.
-        pub async fn next(&mut self) -> Option<Result<String>> {
+        /// The next data message; `None` once the server closes the connection.
+        pub async fn next_message(&mut self) -> Option<Result<Message>> {
             while let Some(msg) = self.0.next().await {
                 match msg {
-                    Ok(Message::Text(t)) => return Some(Ok(t.as_str().to_owned())),
-                    Ok(Message::Close(_)) => return None,
+                    Ok(WsMessage::Text(t)) => return Some(Ok(Message::Text(t.as_str().to_owned()))),
+                    Ok(WsMessage::Binary(b)) => return Some(Ok(Message::Binary(b.into()))),
+                    Ok(WsMessage::Close(_)) => return None,
                     Ok(_) => {}
                     Err(e) => return Some(Err(e.into())),
                 }
             }
             None
         }
+
+        /// Send a close frame.
+        pub async fn close(&mut self) {
+            let _ = self.0.close(None).await;
+        }
     }
 }
 
@@ -46,11 +83,13 @@ mod web {
     use tokio::sync::mpsc;
     use wasm_bindgen::JsCast;
     use wasm_bindgen::prelude::*;
-    use web_sys::{CloseEvent, MessageEvent, WebSocket};
+    use web_sys::{BinaryType, CloseEvent, MessageEvent, WebSocket};
+
+    use super::Message;
 
     enum Event {
         Open,
-        Text(String),
+        Message(Message),
         Error,
         Closed,
     }
@@ -68,6 +107,8 @@ mod web {
     impl Socket {
         pub async fn connect(url: &str) -> Result<Self> {
             let ws = WebSocket::new(url).map_err(|e| anyhow!("websocket {url}: {e:?}"))?;
+            // Binary frames arrive as ArrayBuffers rather than Blobs, which can only be read asynchronously.
+            ws.set_binary_type(BinaryType::Arraybuffer);
             let (tx, rx) = mpsc::unbounded_channel();
             let t = tx.clone();
             let on_open = Closure::<dyn FnMut()>::new(move || {
@@ -75,9 +116,15 @@ mod web {
             });
             let t = tx.clone();
             let on_message = Closure::<dyn FnMut(MessageEvent)>::new(move |e: MessageEvent| {
-                if let Some(text) = e.data().as_string() {
-                    let _ = t.send(Event::Text(text));
-                }
+                let data = e.data();
+                let msg = if let Some(text) = data.as_string() {
+                    Message::Text(text)
+                } else if let Some(buf) = data.dyn_ref::<js_sys::ArrayBuffer>() {
+                    Message::Binary(js_sys::Uint8Array::new(buf).to_vec())
+                } else {
+                    return;
+                };
+                let _ = t.send(Event::Message(msg));
             });
             let t = tx.clone();
             let on_error = Closure::<dyn FnMut()>::new(move || {
@@ -98,18 +145,27 @@ mod web {
             }
         }
 
-        pub async fn send(&mut self, text: String) -> Result<()> {
-            self.ws.send_with_str(&text).map_err(|e| anyhow!("websocket send: {e:?}"))
+        pub async fn send_message(&mut self, m: Message) -> Result<()> {
+            match m {
+                Message::Text(t) => self.ws.send_with_str(&t),
+                Message::Binary(b) => self.ws.send_with_u8_array(&b),
+            }
+            .map_err(|e| anyhow!("websocket send: {e:?}"))
         }
 
-        /// The next text message; `None` once the server closes the connection.
-        pub async fn next(&mut self) -> Option<Result<String>> {
+        /// The next data message; `None` once the server closes the connection.
+        pub async fn next_message(&mut self) -> Option<Result<Message>> {
             match self.rx.recv().await? {
-                Event::Text(t) => Some(Ok(t)),
+                Event::Message(m) => Some(Ok(m)),
                 Event::Error => Some(Err(anyhow!("websocket error"))),
                 Event::Open | Event::Closed => None,
             }
         }
+
+        /// Send a close frame.
+        pub async fn close(&mut self) {
+            let _ = self.ws.close();
+        }
     }
 
     impl Drop for Socket {

docs/architecture/server.md

@@ -17,7 +17,8 @@ On 2026-10-29 Micky approved starting on Hyperliquid's public API, because a nod
 - Hyperliquid closes the whole websocket when asked for a coin it doesn't list, with no error message [observed 2026-10-29]. The gateway only subscribes coins that have a live mid, and keeps the permanent mids on a connection of their own.
 - History comes from `candleSnapshot`, so it stops at the most recent 5000 bars per interval [doc: INFO]. For 1m bars that is about 3.5 days. Node-built candles (§5.3) lift the cap.
 - It makes read-only public calls only. No `exchange` endpoint, no user subscriptions, no orders.
-- The desktop opts in with `TR_SERVER_URL` (§4.6). Crypto keys such as `BTCUSD Curncy` then come from Hyperliquid core perps, and the order book pane names the source. Everything else is unchanged. The web build doesn't use the server yet.
+- The desktop opts in with `TR_SERVER_URL` (§4.6). Crypto keys such as `BTCUSD Curncy` then come from Hyperliquid core perps, and the order book pane names the source. Everything else is unchanged.
+- The web build uses the server for the same crypto markets. `scripts/build-web.sh` builds in `wss://terminal-gateway.style.dev/ws` unless `TR_SERVER_URL` says otherwise. While the server is unreachable, the page falls back to Binance and Coinbase as before.
 - It runs on one Freestyle VM (2 vCPU, 4 GiB) at `wss://terminal-gateway.style.dev/ws`, behind Freestyle's TLS edge. `deploy/freestyle/README.md` covers deploys, health checks and cost, which is about $80 a month above the free allowance.
 
 How facts are marked:
@@ -263,7 +264,7 @@ For comparison, the public `l2Book` feed is pushed "on each block that is at lea
 
 - **`tr-data::remote::RemoteMarkets`** implements `MarketData` over one shared connection. `history` maps to an RPC, `quotes` to a short-lived or shared `Quotes` subscription, and `live` to a `Bars` subscription bridged into the existing `mpsc::Receiver<Bar>`.
 - **Selection:** `TR_SERVER_URL` (or a settings toggle) picks `RemoteMarkets`. The direct providers stay as the offline and dev fallback (`TR_PROVIDER=sample` keeps working).
-- **Web build:** `for_browser()` switches to `RemoteMarkets`. This closes the `TODO(web proxy)` stubs in `crates/data/src/web.rs` and `crates/congress/src/db_web.rs`.
+- **Web build:** `for_browser()` puts `RemoteMarkets` in front of the browser's own sources when the page was built with a server URL (`TR_SERVER_URL`, read by `scripts/build-web.sh`; production uses `wss://terminal-gateway.style.dev/ws`). The client runs on the page's event loop over the browser's `WebSocket`, with postcard in binary frames like the desktop. `?server=ws://localhost:8787/ws` points a page at a server on the same machine, and `?server=off` turns the server off. Crypto is the only asset class the server serves so far, so the `TODO(web proxy)` stubs in `crates/data/src/web.rs` and `crates/congress/src/db_web.rs` stay until it serves the rest.
 
 ---
 

scripts/build-web.sh

@@ -12,6 +12,11 @@
 # cache everything under immutable/ for good (site/vercel.json), and a page can't pair one
 # build's JS with another build's wasm.
 #
+# The app gets crypto markets from the Terminal data server at TR_SERVER_URL, which is built
+# in. It defaults to the production server; set it to another ws:// or wss:// URL, or to
+# nothing (TR_SERVER_URL=) for no server. A page can also ask for a server on the same machine
+# with ?server=ws://localhost:8787/ws, or for none with ?server=off.
+#
 # Needs the wasm32-unknown-unknown target (rustup target add wasm32-unknown-unknown),
 # wasm-bindgen-cli at the version pinned in Cargo.lock, and wasm-opt from binaryen.
 set -euo pipefail
@@ -19,7 +24,17 @@ cd "$(dirname "$0")/.."
 
 out="${1:-site/static/app}"
 target=wasm32-unknown-unknown
-wasm="target/$target/web/terminal.wasm"
+wasm="${CARGO_TARGET_DIR:-target}/$target/web/terminal.wasm"
+
+export TR_SERVER_URL="${TR_SERVER_URL-wss://terminal-gateway.style.dev/ws}"
+case "$TR_SERVER_URL" in
+"") echo "no Terminal server: crypto markets come from Binance and Coinbase" ;;
+ws://* | wss://*) echo "Terminal server: $TR_SERVER_URL" ;;
+*)
+	echo "TR_SERVER_URL must be a ws:// or wss:// URL, or empty for no server" >&2
+	exit 1
+	;;
+esac
 
 want=$(awk '/^name = "wasm-bindgen"$/ { getline; gsub(/version = |"/, ""); print; exit }' Cargo.lock)
 have=$(wasm-bindgen --version 2>/dev/null | awk '{ print $2 }' || true)

server/README.md

@@ -16,6 +16,8 @@ The desktop app uses it for crypto markets when `TR_SERVER_URL` is set:
 TR_SERVER_URL=ws://localhost:8787 cargo run --release
 ```
 
+The browser build uses it too. `scripts/build-web.sh` builds in `wss://terminal-gateway.style.dev/ws`, or `TR_SERVER_URL` when it is set (empty for none). To try a local gateway with a built page, open the page with `?server=ws://localhost:8787/ws`.
+
 The gateway serves plain HTTP and websockets. For `wss://` on 443, put a TLS-terminating edge in front of it and bind to an address the edge can reach, for example `TR_GATEWAY_BIND=0.0.0.0:8080`. Clients accept `wss://` URLs. The gateway pings each client every 20 s so idle connections survive proxies. The hosted one runs on Freestyle; see `deploy/freestyle/README.md`.
 
 ## Restarts

server/tests/gateway.rs

@@ -602,6 +602,86 @@ async fn shutdown_closes_upstream_for_good() {
     while !matches!(next(&mut book).await, Event::Reconnecting) {}
 }
 
+#[tokio::test]
+async fn order_books_fall_back_while_the_server_is_away() {
+    let s = Setup::new(|_| {}).await;
+    let m = RemoteMarkets::new(&format!("ws://{}", s.gw.addr), tr_data::sample::SampleProvider::default());
+    let push = async {
+        s.mock.until("l2Book:BTC upstream", |mk| mk.subs().contains("l2Book:BTC")).await;
+        s.mock.push("l2Book:BTC", "l2Book", book_msg("BTC", "100.0"));
+    };
+    let (book, ()) = tokio::join!(m.order_book("BTCUSD Curncy", 10), push);
+    let book = book.unwrap();
+    assert_eq!((book.source.as_str(), book.bids[0].0), (tr_data::remote::BOOK_SOURCE, 100.0));
+    // The server goes away: the pane gets the fallback's book, not the last one the server sent.
+    let Setup { mock: _mock, gw } = s;
+    tokio::time::timeout(WAIT, gw.shutdown()).await.expect("shutdown finishes");
+    let t0 = tokio::time::Instant::now();
+    loop {
+        let book = m.order_book("BTCUSD Curncy", 10).await.unwrap();
+        if book.source != tr_data::remote::BOOK_SOURCE {
+            assert!(book.simulated, "the sample provider's book");
+            break;
+        }
+        assert!(t0.elapsed() < WAIT, "still the server's last book {:?} after it went away", book.bids[0]);
+        tokio::time::sleep(Duration::from_millis(100)).await;
+    }
+}
+
+#[tokio::test]
+async fn charts_hand_off_to_the_fallback_and_back_when_the_server_goes_and_returns() {
+    let s = Setup::new(|_| {}).await;
+    let addr = s.gw.addr;
+    let m = RemoteMarkets::new(&format!("ws://{addr}"), tr_data::sample::SampleProvider::default().with_live_ticks());
+    assert_eq!(m.history("BTCUSD Curncy", Timeframe::M1, None, 10).await.unwrap().len(), 10, "from the server");
+    let mut feed = m.live("BTCUSD Curncy", Timeframe::M1).expect("the server's feed");
+    s.mock.until("candle:BTC:1m upstream", |mk| mk.subs().contains("candle:BTC:1m")).await;
+    // The server goes away and stays away: the feed ends, so the worker reloads the chart...
+    let Setup { mock: _mock, gw } = s;
+    tokio::time::timeout(WAIT, gw.shutdown()).await.expect("shutdown finishes");
+    let gone = tokio::time::Instant::now();
+    while tokio::time::timeout(Duration::from_secs(20), feed.recv()).await.expect("the feed ends").is_some() {}
+    assert!(gone.elapsed() >= Duration::from_secs(9), "it gave the server a while to come back first");
+    // ...from the fallback.
+    let bars = m.history("BTCUSD Curncy", Timeframe::M1, None, 10).await.unwrap();
+    assert_eq!(bars.len(), 10, "sample bars");
+    let mut feed = m.live("BTCUSD Curncy", Timeframe::M1).expect("the fallback's feed");
+    assert!(recv_bar(&mut feed).await.close > 0.0, "the fallback's live bars come through");
+    // The server comes back on the same address: the fallback's feed ends, so the worker reloads
+    // the chart from the server.
+    let back = Setup::new(|c| c.bind = addr).await;
+    while tokio::time::timeout(Duration::from_secs(30), feed.recv()).await.expect("the fallback feed ends").is_some() {}
+    assert_eq!(m.history("BTCUSD Curncy", Timeframe::M1, None, 10).await.unwrap().len(), 10);
+    let _feed = m.live("BTCUSD Curncy", Timeframe::M1).expect("the server's feed again");
+    back.mock.until("candle:BTC:1m upstream again", |mk| mk.subs().contains("candle:BTC:1m")).await;
+}
+
+#[tokio::test]
+async fn charts_on_the_fallback_move_back_on_a_newer_connection_only() {
+    // Nothing listens there yet, so the chart loads from the fallback.
+    let addr = std::net::TcpListener::bind((Ipv4Addr::LOCALHOST, 0)).unwrap().local_addr().unwrap();
+    let m = RemoteMarkets::new(&format!("ws://{addr}"), tr_data::sample::SampleProvider::default().with_live_ticks());
+    assert_eq!(m.history("BTCUSD Curncy", Timeframe::M1, None, 10).await.unwrap().len(), 10, "sample bars");
+    // The server comes up before the chart starts its live feed...
+    let s = Setup::new(|c| c.bind = addr).await;
+    let t0 = tokio::time::Instant::now();
+    while m.quotes(&["BTCUSD Curncy".into()]).await.unwrap()[0].last != 100.5 {
+        assert!(t0.elapsed() < Duration::from_secs(20), "the client never connected");
+    }
+    // ...so the fallback's feed ends at once, and the chart reloads from the server.
+    let mut feed = m.live("BTCUSD Curncy", Timeframe::M1).expect("the fallback's feed");
+    let ended = tokio::time::timeout(Duration::from_secs(2), async { while feed.recv().await.is_some() {} }).await;
+    assert!(ended.is_ok(), "the fallback's feed kept going with the server up");
+    // A chart that falls back while the server is up (it failed the request) stays on the fallback:
+    // reloading would only fail again.
+    *s.mock.fail_candles.lock().unwrap() = 1000;
+    assert_eq!(m.history("BTCUSD Curncy", Timeframe::M1, None, 10).await.unwrap().len(), 10, "sample bars");
+    assert!(*s.mock.fail_candles.lock().unwrap() < 1000, "the server was asked and failed");
+    let mut feed = m.live("BTCUSD Curncy", Timeframe::M1).expect("the fallback's feed");
+    let ended = tokio::time::timeout(Duration::from_secs(2), async { while feed.recv().await.is_some() {} }).await;
+    assert!(ended.is_err(), "the fallback's feed ended with no new connection");
+}
+
 #[tokio::test]
 async fn desktop_charts_backfill_bars_missed_in_an_outage() {
     let s = Setup::new(|_| {}).await;