fix: cancel routing task on drop, handle control result, tighten validation

src-tauri/src/ble/source.rs

@@ -1,11 +1,13 @@
 use std::pin::Pin;
 use std::sync::{Arc, Mutex};
+use std::task::{Context, Poll};
 use std::time::Duration;
 
 use async_trait::async_trait;
 use btleplug::api::{bleuuid::uuid_from_u16, Central, Peripheral as _, ValueNotification, WriteType};
 use btleplug::platform::{Adapter, Peripheral};
 use futures::{Stream, StreamExt};
+use tokio::sync::oneshot;
 use uuid::Uuid;
 
 use crate::ble::BleError;
@@ -69,6 +71,29 @@ impl BleSource {
     }
 }
 
+/// Stream wrapper that cancels the background notification router on drop, so a
+/// reconnect does not leak the previous cycle's routing task.
+struct CancellableStream {
+    inner: tokio_stream::wrappers::UnboundedReceiverStream<Telemetry>,
+    cancel: Option<oneshot::Sender<()>>,
+}
+
+impl Stream for CancellableStream {
+    type Item = Telemetry;
+
+    fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Telemetry>> {
+        Pin::new(&mut self.inner).poll_next(cx)
+    }
+}
+
+impl Drop for CancellableStream {
+    fn drop(&mut self) {
+        if let Some(cancel) = self.cancel.take() {
+            let _ = cancel.send(());
+        }
+    }
+}
+
 #[async_trait]
 impl SensorSource for BleSource {
     async fn scan(&self, seconds: u64) -> Result<Vec<Device>, BleError> {
@@ -106,19 +131,33 @@ impl SensorSource for BleSource {
         // `request_control` can await confirmation while telemetry is still being consumed.
         let (tx, rx) = tokio::sync::mpsc::unbounded_channel::<Telemetry>();
         let control = self.control.clone();
+        let (cancel_tx, cancel_rx) = oneshot::channel::<()>();
 
         tokio::spawn(async move {
             let mut notifications = notifications;
             let mut acc = CscAccumulator::new();
-            while let Some(n) = notifications.next().await {
+            let mut cancel_rx = cancel_rx;
-                for t in parse_notification(n, &mut acc, &control) {
+            loop {
-                    let _ = tx.send(t);
+                tokio::select! {
+                    n = notifications.next() => {
+                        match n {
+                            Some(n) => {
+                                for t in parse_notification(n, &mut acc, &control) {
+                                    let _ = tx.send(t);
+                                }
+                            }
+                            None => break,
+                        }
+                    }
+                    _ = &mut cancel_rx => break,
                 }
             }
         });
 
-        let stream: Pin<Box<dyn Stream<Item = Telemetry> + Send>> =
+        let stream: Pin<Box<dyn Stream<Item = Telemetry> + Send>> = Box::pin(CancellableStream {
-            Box::pin(tokio_stream::wrappers::UnboundedReceiverStream::new(rx));
+            inner: tokio_stream::wrappers::UnboundedReceiverStream::new(rx),
+            cancel: Some(cancel_tx),
+        });
         Ok(stream)
     }
 

src-tauri/src/commands.rs

@@ -73,11 +73,16 @@ pub(crate) enum ControlResult {
 }
 
 /// Parses a Fitness Machine Status (0x2ADA) notification for a Control Point
-/// Response. Layout: [opcode=0x80][request opcode][result code].
+/// Response to Request Control (opcode 0x00).
+/// Layout: [opcode=0x80][request opcode=0x00][result code].
 pub(crate) fn parse_control_result(bytes: &[u8]) -> Option<ControlResult> {
     if bytes.first() != Some(&0x80) {
         return None;
     }
+    // Only accept responses to Request Control; ignore responses to other commands.
+    if bytes.get(1) != Some(&0x00) {
+        return None;
+    }
     match bytes.get(2) {
         Some(0x01) => Some(ControlResult::Success),
         Some(0x02) => Some(ControlResult::OpCodeNotSupported),
@@ -112,6 +117,12 @@ mod tests {
         assert_eq!(parse_control_result(&[]), None);
     }
 
+    #[test]
+    fn test_parse_control_result_ignores_other_request_opcodes() {
+        // opcode 0x80 but request opcode != 0x00 (e.g. response to Set Target Power 0x05)
+        assert_eq!(parse_control_result(&[0x80, 0x05, 0x01]), None);
+    }
+
     #[test]
     fn test_encode_request_control() {
         assert_eq!(

src-tauri/src/lib.rs

@@ -129,10 +129,12 @@ async fn connect(app: AppHandle, device_id: String, category: String) -> Result<
         .unwrap()
         .insert(category.clone(), handle);
 
-    // Trainer: remember the device id. Control is requested (with confirmation)
+    // Trainer: remember the device id and reset the previous trainer's targets.
-    // inside run_watchdog, after the notification subscription is live.
+    // Control is requested (with confirmation) inside run_watchdog, after the
+    // notification subscription is live.
     if category == "trainer" {
         *state.trainer.lock().unwrap() = Some(device.id.clone());
+        *state.trainer_state.lock().unwrap() = TrainerState::default();
     }
 
     Ok(())
@@ -276,8 +278,10 @@ async fn run_watchdog(
 
         // Trainer: request control (with confirmation) and restore targets/start.
         if category == "trainer" {
-            let _ = source.request_control(&device).await;
+            match source.request_control(&device).await {
-            reapply_trainer_state(&app, &source, &device).await;
+                Ok(()) => reapply_trainer_state(&app, &source, &device).await,
+                Err(e) => eprintln!("request_control failed: {e}"),
+            }
         }
 
         let app_emit = app.clone();
@@ -310,7 +314,7 @@ async fn run_watchdog(
             app.state::<State>().streams.lock().unwrap().remove(&category);
             return;
         }
-        time::sleep(RECONNECT_BACKOFF).await;
+        time::sleep(RECONNECT_BACKOFF * silent_cycles.max(1)).await;
     }
 }
 
@@ -339,7 +343,9 @@ async fn reapply_trainer_state(
         let _ = source.command(device, Command::FtmsSetTargetPower(watts)).await;
     }
     if let Some(rpm) = cadence {
-        let _ = source.command(device, Command::FtmsSetTargetCadence(rpm * 2)).await;
+        if let Some(half_rpm) = rpm.checked_mul(2) {
+            let _ = source.command(device, Command::FtmsSetTargetCadence(half_rpm)).await;
+        }
     }
     if running {
         let _ = source.command(device, Command::FtmsStart).await;

src/main.js

@@ -139,7 +139,7 @@ once("devices-loaded", (event) => {
   renderGroup(groups.trainers, repo.trainers ?? []);
 });
 
-once("battery", (event) => {
+listen("battery", (event) => {
   const { device_id, percent } = event.payload;
   const card = deviceCards.get(device_id);
   if (card) card.querySelector(".device__head").append(batteryIcon(percent));
@@ -230,6 +230,7 @@ document.querySelector("#apply-btn").addEventListener("click", async () => {
   const watts = Number(powerRaw);
   const rpm = Number(cadenceRaw);
   if (!Number.isFinite(watts) || !Number.isFinite(rpm)) return;
+  if (watts < 0 || rpm < 0 || !Number.isInteger(watts) || !Number.isInteger(rpm)) return;
   try {
     await invoke("set_target_power", { watts });
     await invoke("set_target_cadence", { rpm });