[feat](trx-rs): make spectrum affordable over a slow link

Spectrum dominates the server↔client connection, and all three things that
govern its cost were working against a poor link.

**It was polled, one round trip per frame.** The client asked for a frame every
50 ms on a dedicated connection and waited for the reply, so the frame rate was
capped at 1/RTT — on a 200 ms link, five frames a second no matter what was
configured.  Add SubscribeSpectrum alongside the existing SubscribeMeter: the
server pushes frames from a per-rig broadcast that rig_task fills only while
somebody is subscribed.  A server too old to know the command answers with an
error and leaves the connection usable, so the client falls back to polling on
the same connection without reconnecting.

**Bins were JSON floats.** 1024 bins spelled out as decimal text is around
10 KB a frame, ~200 KB/s at full rate — while the very next hop, client to
browser, already sends the same information as base64 i8 in about 1.4 KB.  Bins
now travel base64-encoded whole dBFS, the resolution the display draws at
anyway.  Decoding still accepts the old array form.

**Nothing was tunable.** [sdr].spectrum_fft_size and [sdr].spectrum_interval_ms
replace the compile-time FFT size and cadence; [[remotes]].spectrum_interval_ms
lets the client ask for less.  512 bins at 5 frames/s is roughly 3.5 KB/s
against roughly 200 KB/s before.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01SyX26FCpMQxiBoC7r5K1A7
Signed-off-by: Stan Grams <sjg@haxx.space>

src/trx-client/src/main.rs

@@ -288,6 +288,7 @@ async fn async_init() -> DynResult<AppState> {
         let token = cli.token.clone().or_else(|| cfg.remote.auth.token.clone());
         let poll_interval_ms = cli.poll_interval_ms.unwrap_or(cfg.remote.poll_interval_ms);
         vec![RemoteEntry {
+            spectrum_interval_ms: cfg.remote.spectrum_interval_ms,
             name,
             url: url.clone(),
             rig_id,
@@ -483,6 +484,13 @@ async fn async_init() -> DynResult<AppState> {
             .map(|e| e.poll_interval_ms)
             .min()
             .unwrap_or(750);
+        // Entries sharing a server share its connections, so the most frequent
+        // request wins: whoever wants spectrum fastest sets the rate.
+        let spectrum_interval = entries
+            .iter()
+            .map(|e| e.spectrum_interval_ms)
+            .min()
+            .unwrap_or_else(|| remote_client::DEFAULT_SPECTRUM_INTERVAL.as_millis() as u64);
 
         let (server_tx, server_rx) = mpsc::channel::<RigRequest>(RIG_TASK_CHANNEL_BUFFER);
         for entry in entries {
@@ -496,6 +504,8 @@ async fn async_init() -> DynResult<AppState> {
             known_rigs: frontend_runtime.routing.remote_rigs.clone(),
             rig_states: frontend_runtime.routing.rig_states.clone(),
             poll_interval: Duration::from_millis(poll_interval),
+            spectrum_interval: Duration::from_millis(spectrum_interval),
+            spectrum_stream_unsupported: Arc::new(std::sync::atomic::AtomicBool::new(false)),
             spectrum: frontend_runtime.spectrum.sender.clone(),
             rig_spectrums: frontend_runtime.spectrum.per_rig.clone(),
             server_connected: frontend_runtime.routing.server_connected.clone(),

src/trx-client/src/remote_client.rs

@@ -20,7 +20,7 @@ use trx_core::{RigError, RigResult};
 use trx_frontend::{RemoteRigEntry, SharedSpectrum};
 use trx_protocol::rig_command_to_client;
 use trx_protocol::types::RigEntry;
-use trx_protocol::{ClientCommand, ClientEnvelope, ClientResponse, MeterUpdate};
+use trx_protocol::{ClientCommand, ClientEnvelope, ClientResponse, MeterUpdate, SpectrumFrame};
 
 // Endpoint parsing lives in `trx-config` so config validation and the
 // connection code agree on what a URL means.
@@ -32,9 +32,10 @@ const SPECTRUM_IO_TIMEOUT: Duration = Duration::from_secs(3);
 const MAX_JSON_LINE_BYTES: usize = 256 * 1024;
 const MAX_CONSECUTIVE_POLL_FAILURES: u32 = 3;
 
-// Keep remote spectrum reasonably responsive without returning to the old
-// timeout churn caused by a much tighter request cadence.
-const SPECTRUM_POLL_INTERVAL: Duration = Duration::from_millis(50);
+// Default spectrum cadence when a config does not specify one.  Both the push
+// stream and the poll fallback run at the configured rate; see
+// `[[remotes]].spectrum_interval_ms`.
+pub const DEFAULT_SPECTRUM_INTERVAL: Duration = Duration::from_millis(50);
 
 #[derive(Clone)]
 pub struct RemoteClientConfig {
@@ -43,6 +44,12 @@ pub struct RemoteClientConfig {
     pub selected_rig_id: Arc<Mutex<Option<String>>>,
     pub known_rigs: Arc<Mutex<Vec<RemoteRigEntry>>>,
     pub poll_interval: Duration,
+    /// How often spectrum frames are wanted.  Drives the poll fallback and is
+    /// the rate the client asks the server to push at.
+    pub spectrum_interval: Duration,
+    /// Set once a server has rejected `SubscribeSpectrum`, so later
+    /// connections to it go straight to polling instead of asking again.
+    pub spectrum_stream_unsupported: Arc<AtomicBool>,
     /// Spectrum watch sender; spectrum task publishes here, SSE clients subscribe.
     pub spectrum: Arc<watch::Sender<SharedSpectrum>>,
     /// Shared flag: `true` while a TCP connection to trx-server is active.
@@ -491,6 +498,100 @@ async fn send_get_sat_passes_on(
     ))
 }
 
+/// What ended a spectrum stream attempt.
+enum SpectrumStreamOutcome {
+    /// The stream ran and is over; the connection is spent.
+    Finished,
+    /// The server rejected the subscription.  The connection is still usable,
+    /// so the caller can poll on it.
+    Unsupported,
+}
+
+/// Subscribe to the server's spectrum push for one rig and publish frames as
+/// they arrive.
+///
+/// The server answers either with frames or, when it is too old to know the
+/// command, with an error response — which leaves the connection usable for
+/// polling, so falling back costs no reconnect.
+async fn run_spectrum_stream(
+    config: &RemoteClientConfig,
+    writer: &mut (impl AsyncWriteExt + Unpin),
+    reader: &mut (impl AsyncBufRead + Unpin),
+    short_name: &str,
+    shutdown_rx: &mut watch::Receiver<bool>,
+) -> RigResult<SpectrumStreamOutcome> {
+    let envelope = build_envelope(
+        config,
+        ClientCommand::SubscribeSpectrum,
+        Some(short_name.to_string()),
+    );
+    let mut payload = serde_json::to_string(&envelope)
+        .map_err(|e| RigError::communication(format!("JSON serialize failed: {e}")))?;
+    payload.push('\n');
+    time::timeout(SPECTRUM_IO_TIMEOUT, writer.write_all(payload.as_bytes()))
+        .await
+        .map_err(|_| RigError::communication("spectrum subscribe write timed out".to_string()))?
+        .map_err(|e| RigError::communication(format!("spectrum subscribe write failed: {e}")))?;
+    time::timeout(SPECTRUM_IO_TIMEOUT, writer.flush())
+        .await
+        .map_err(|_| RigError::communication("spectrum subscribe flush timed out".to_string()))?
+        .map_err(|e| RigError::communication(format!("spectrum subscribe flush failed: {e}")))?;
+
+    // Re-check what the UI wants: switching rigs has to end this stream so the
+    // connection can be rebuilt for the new one.
+    let mut supervisor = time::interval(Duration::from_millis(500));
+    supervisor.tick().await;
+
+    loop {
+        tokio::select! {
+            changed = shutdown_rx.changed() => {
+                match changed {
+                    Ok(()) if *shutdown_rx.borrow() => return Ok(SpectrumStreamOutcome::Finished),
+                    Ok(()) => {}
+                    Err(_) => return Ok(SpectrumStreamOutcome::Finished),
+                }
+            }
+            _ = supervisor.tick() => {
+                let wanted = active_spectrum_rig_ids(config);
+                if wanted.len() != 1 || wanted[0] != short_name {
+                    return Ok(SpectrumStreamOutcome::Finished);
+                }
+            }
+            line = read_limited_line(reader, MAX_JSON_LINE_BYTES) => {
+                let line = line
+                    .map_err(|e| RigError::communication(format!("spectrum read failed: {e}")))?
+                    .ok_or_else(|| {
+                        RigError::communication("spectrum connection closed".to_string())
+                    })?;
+                let trimmed = line.trim_end();
+                if trimmed.is_empty() {
+                    continue;
+                }
+                match serde_json::from_str::<SpectrumFrame>(trimmed) {
+                    Ok(frame) => publish_spectrum_frame(config, short_name, frame),
+                    // Anything that is not a frame means the server would rather
+                    // answer than stream: an older build rejecting the command.
+                    Err(_) => return Ok(SpectrumStreamOutcome::Unsupported),
+                }
+            }
+        }
+    }
+}
+
+/// Publish one pushed frame to the per-rig and selected-rig watch channels.
+fn publish_spectrum_frame(config: &RemoteClientConfig, short_name: &str, frame: SpectrumFrame) {
+    if let Ok(map) = config.rig_spectrums.read() {
+        if let Some(tx) = map.get(short_name) {
+            tx.send_modify(|s| s.set(Some(frame.spectrum.clone()), frame.vchan_rds.clone()));
+        }
+    }
+    if selected_rig_id(config).as_deref() == Some(short_name) {
+        config
+            .spectrum
+            .send_modify(|s| s.set(Some(frame.spectrum), frame.vchan_rds));
+    }
+}
+
 async fn handle_spectrum_connection(
     config: &RemoteClientConfig,
     stream: TcpStream,
@@ -498,7 +599,33 @@ async fn handle_spectrum_connection(
 ) -> RigResult<()> {
     let (reader, mut writer) = stream.into_split();
     let mut reader = BufReader::new(reader);
-    let mut interval = time::interval(SPECTRUM_POLL_INTERVAL);
+
+    // Prefer the push stream: polling costs a round trip per frame, so on a
+    // high-latency link the frame rate is 1/RTT no matter what interval is
+    // configured.  It only works for one rig per connection, and only against
+    // a server new enough to understand the command.
+    let streamable = active_spectrum_rig_ids(config);
+    if streamable.len() == 1 && !config.spectrum_stream_unsupported.load(Ordering::Relaxed) {
+        match run_spectrum_stream(
+            config,
+            &mut writer,
+            &mut reader,
+            &streamable[0],
+            shutdown_rx,
+        )
+        .await?
+        {
+            SpectrumStreamOutcome::Finished => return Ok(()),
+            SpectrumStreamOutcome::Unsupported => {
+                info!("Server does not support spectrum streaming; falling back to polling");
+                config
+                    .spectrum_stream_unsupported
+                    .store(true, Ordering::Relaxed);
+            }
+        }
+    }
+
+    let mut interval = time::interval(config.spectrum_interval);
     // Cache the token outside the poll loop to avoid cloning it every 50ms.
     let cached_token = config.token.clone();
 
@@ -1187,6 +1314,7 @@ mod tests {
     use super::{has_short_names, resolve_server_rig_id, resolve_short_name};
     use super::{
         parse_audio_url, parse_remote_url, RemoteClientConfig, RemoteEndpoint, SharedSpectrum,
+        DEFAULT_SPECTRUM_INTERVAL,
     };
     use std::collections::HashMap;
     use std::sync::atomic::AtomicBool;
@@ -1381,6 +1509,8 @@ mod tests {
                 selected_rig_id: Arc::new(Mutex::new(None)),
                 known_rigs: Arc::new(Mutex::new(Vec::new())),
                 poll_interval: Duration::from_millis(100),
+                spectrum_interval: DEFAULT_SPECTRUM_INTERVAL,
+                spectrum_stream_unsupported: Arc::new(AtomicBool::new(false)),
                 spectrum: Arc::new(spectrum_tx),
                 server_connected: Arc::new(AtomicBool::new(false)),
                 rig_server_connected: Arc::new(RwLock::new(HashMap::new())),
@@ -1426,6 +1556,8 @@ mod tests {
             selected_rig_id: Arc::new(Mutex::new(Some("sdr".to_string()))),
             known_rigs: Arc::new(Mutex::new(Vec::new())),
             poll_interval: Duration::from_millis(500),
+            spectrum_interval: DEFAULT_SPECTRUM_INTERVAL,
+            spectrum_stream_unsupported: Arc::new(AtomicBool::new(false)),
             spectrum: Arc::new(spectrum_tx),
             server_connected: Arc::new(AtomicBool::new(false)),
             rig_server_connected: Arc::new(RwLock::new(HashMap::new())),
@@ -1441,6 +1573,130 @@ mod tests {
         assert_eq!(envelope.rig_id.as_deref(), Some("sdr"));
     }
 
+    fn stream_test_config(spectrum_tx: watch::Sender<SharedSpectrum>) -> super::RemoteClientConfig {
+        super::RemoteClientConfig {
+            addr: "127.0.0.1:4530".to_string(),
+            token: None,
+            selected_rig_id: Arc::new(Mutex::new(Some("sdr".to_string()))),
+            known_rigs: Arc::new(Mutex::new(Vec::new())),
+            poll_interval: Duration::from_millis(500),
+            spectrum_interval: DEFAULT_SPECTRUM_INTERVAL,
+            spectrum_stream_unsupported: Arc::new(AtomicBool::new(false)),
+            spectrum: Arc::new(spectrum_tx),
+            server_connected: Arc::new(AtomicBool::new(false)),
+            rig_server_connected: Arc::new(RwLock::new(HashMap::new())),
+            rig_states: Arc::new(RwLock::new(HashMap::new())),
+            rig_spectrums: Arc::new(RwLock::new(HashMap::new())),
+            rig_id_to_short_name: HashMap::new(),
+            short_name_to_rig_id: Arc::new(RwLock::new(HashMap::new())),
+            sat_passes: Arc::new(RwLock::new(None)),
+            rig_meters: Arc::new(RwLock::new(HashMap::new())),
+        }
+    }
+
+    /// A pushed frame reaches the watch channel the UI reads, with no request
+    /// from the client beyond the initial subscribe.
+    #[tokio::test]
+    async fn spectrum_stream_publishes_pushed_frames() {
+        use tokio::io::{AsyncBufReadExt, BufReader};
+
+        let (spectrum_tx, mut spectrum_rx) = watch::channel(SharedSpectrum::default());
+        let config = stream_test_config(spectrum_tx);
+        let (shutdown_tx, mut shutdown_rx) = watch::channel(false);
+
+        let (client_io, mut server_io) = tokio::io::duplex(64 * 1024);
+        let (client_read, mut client_write) = tokio::io::split(client_io);
+        let mut client_read = BufReader::new(client_read);
+
+        let task = tokio::spawn(async move {
+            super::run_spectrum_stream(
+                &config,
+                &mut client_write,
+                &mut client_read,
+                "sdr",
+                &mut shutdown_rx,
+            )
+            .await
+        });
+
+        // The server sees the subscribe, then pushes without being asked.
+        let mut server = BufReader::new(&mut server_io);
+        let mut subscribe = String::new();
+        server.read_line(&mut subscribe).await.expect("subscribe");
+        assert!(
+            subscribe.contains("subscribe_spectrum"),
+            "unexpected command: {subscribe}"
+        );
+
+        let frame = trx_protocol::SpectrumFrame {
+            rig_id: "sdr".to_string(),
+            spectrum: trx_core::rig::state::SpectrumData {
+                bins: vec![-70.0, -30.0],
+                center_hz: 14_200_000,
+                sample_rate: 1_920_000,
+                rds: None,
+            },
+            vchan_rds: None,
+        };
+        let mut line = serde_json::to_string(&frame).unwrap();
+        line.push('\n');
+        server_io.write_all(line.as_bytes()).await.expect("push");
+        server_io.flush().await.expect("flush");
+
+        spectrum_rx.changed().await.expect("spectrum published");
+        let published = spectrum_rx.borrow().clone();
+        let spectrum = published.frame.expect("spectrum present");
+        assert_eq!(spectrum.center_hz, 14_200_000);
+        assert_eq!(spectrum.bins, vec![-70.0, -30.0]);
+
+        let _ = shutdown_tx.send(true);
+        let _ = task.await;
+    }
+
+    /// An older server answers the unknown command with an error instead of
+    /// frames.  That has to read as "poll instead", not as a dead connection.
+    #[tokio::test]
+    async fn spectrum_stream_falls_back_when_unsupported() {
+        use tokio::io::{AsyncBufReadExt, BufReader};
+
+        let (spectrum_tx, _spectrum_rx) = watch::channel(SharedSpectrum::default());
+        let config = stream_test_config(spectrum_tx);
+        let (_shutdown_tx, mut shutdown_rx) = watch::channel(false);
+
+        let (client_io, mut server_io) = tokio::io::duplex(64 * 1024);
+        let (client_read, mut client_write) = tokio::io::split(client_io);
+        let mut client_read = BufReader::new(client_read);
+
+        let task = tokio::spawn(async move {
+            super::run_spectrum_stream(
+                &config,
+                &mut client_write,
+                &mut client_read,
+                "sdr",
+                &mut shutdown_rx,
+            )
+            .await
+        });
+
+        let mut server = BufReader::new(&mut server_io);
+        let mut subscribe = String::new();
+        server.read_line(&mut subscribe).await.expect("subscribe");
+
+        server_io
+            .write_all(
+                b"{\"success\":false,\"state\":null,\"error\":\"Invalid JSON: unknown variant\"}\n",
+            )
+            .await
+            .expect("error response");
+        server_io.flush().await.expect("flush");
+
+        let outcome = task.await.expect("join").expect("stream result");
+        assert!(
+            matches!(outcome, super::SpectrumStreamOutcome::Unsupported),
+            "an error response should fall back to polling"
+        );
+    }
+
     #[test]
     fn build_envelope_translates_short_name_to_server_rig_id() {
         let (spectrum_tx, _spectrum_rx) = watch::channel(SharedSpectrum::default());
@@ -1454,6 +1710,8 @@ mod tests {
             selected_rig_id: Arc::new(Mutex::new(Some("home-hf".to_string()))),
             known_rigs: Arc::new(Mutex::new(Vec::new())),
             poll_interval: Duration::from_millis(500),
+            spectrum_interval: DEFAULT_SPECTRUM_INTERVAL,
+            spectrum_stream_unsupported: Arc::new(AtomicBool::new(false)),
             spectrum: Arc::new(spectrum_tx),
             server_connected: Arc::new(AtomicBool::new(false)),
             rig_server_connected: Arc::new(RwLock::new(HashMap::new())),
@@ -1486,6 +1744,8 @@ mod tests {
             selected_rig_id: Arc::new(Mutex::new(None)),
             known_rigs: Arc::new(Mutex::new(Vec::new())),
             poll_interval: Duration::from_millis(500),
+            spectrum_interval: DEFAULT_SPECTRUM_INTERVAL,
+            spectrum_stream_unsupported: Arc::new(AtomicBool::new(false)),
             spectrum: Arc::new(spectrum_tx),
             server_connected: Arc::new(AtomicBool::new(false)),
             rig_server_connected: Arc::new(RwLock::new(HashMap::new())),
@@ -1510,6 +1770,8 @@ mod tests {
             selected_rig_id: Arc::new(Mutex::new(None)),
             known_rigs: Arc::new(Mutex::new(Vec::new())),
             poll_interval: Duration::from_millis(500),
+            spectrum_interval: DEFAULT_SPECTRUM_INTERVAL,
+            spectrum_stream_unsupported: Arc::new(AtomicBool::new(false)),
             spectrum: Arc::new(spectrum_tx),
             server_connected: Arc::new(AtomicBool::new(false)),
             rig_server_connected: Arc::new(RwLock::new(HashMap::new())),
@@ -1550,6 +1812,8 @@ mod tests {
             selected_rig_id: Arc::new(Mutex::new(None)),
             known_rigs: known_rigs.clone(),
             poll_interval: Duration::from_millis(500),
+            spectrum_interval: DEFAULT_SPECTRUM_INTERVAL,
+            spectrum_stream_unsupported: Arc::new(AtomicBool::new(false)),
             spectrum: Arc::new(spectrum_tx),
             server_connected: Arc::new(AtomicBool::new(false)),
             rig_server_connected: Arc::new(RwLock::new(HashMap::new())),
@@ -1623,6 +1887,8 @@ mod tests {
             selected_rig_id,
             known_rigs,
             poll_interval: Duration::from_millis(500),
+            spectrum_interval: DEFAULT_SPECTRUM_INTERVAL,
+            spectrum_stream_unsupported: Arc::new(AtomicBool::new(false)),
             spectrum: Arc::new(spectrum_tx),
             server_connected: Arc::new(AtomicBool::new(false)),
             rig_server_connected: Arc::new(RwLock::new(HashMap::new())),

src/trx-client/trx-frontend/trx-frontend-http/frontend/src/api/generated.ts

@@ -78,6 +78,10 @@ export type RdsData = { pi?: number | null, program_service?: string | null, rad
 export type SpectrumData = { 
 /**
  * FFT magnitude bins in dBFS, FFT-shifted so DC (centre frequency) is at index N/2.
+ *
+ * On the wire these are base64-encoded `i8` whole dBFS (see
+ * `spectrum_wire`), which is what the display draws anyway; the TypeScript
+ * type describes the decoded array the browser receives over SSE.
  */
 bins: Array<number>, 
 /**