From 9ec7b78e98090d17ff6cd717e12a9960e31b599e Mon Sep 17 00:00:00 2001 From: lambda Date: Mon, 26 Oct 2026 18:19:41 +0300 Subject: [PATCH] fix: cancel routing task on drop, handle control result, tighten validation --- src-tauri/src/ble/source.rs | 49 +++++++++++++++++++++++++++++++++---- src-tauri/src/commands.rs | 13 +++++++++- src-tauri/src/lib.rs | 18 +++++++++----- src/main.js | 3 ++- 4 files changed, 70 insertions(+), 13 deletions(-) diff --git a/src-tauri/src/ble/source.rs b/src-tauri/src/ble/source.rs index 06a1704..09a36fd 100644 --- a/src-tauri/src/ble/source.rs +++ b/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, + cancel: Option>, +} + +impl Stream for CancellableStream { + type Item = Telemetry; + + fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { + 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, 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::(); 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 { - for t in parse_notification(n, &mut acc, &control) { - let _ = tx.send(t); + let mut cancel_rx = cancel_rx; + loop { + 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 + Send>> = - Box::pin(tokio_stream::wrappers::UnboundedReceiverStream::new(rx)); + let stream: Pin + Send>> = Box::pin(CancellableStream { + inner: tokio_stream::wrappers::UnboundedReceiverStream::new(rx), + cancel: Some(cancel_tx), + }); Ok(stream) } diff --git a/src-tauri/src/commands.rs b/src-tauri/src/commands.rs index af2bf08..220e03d 100644 --- a/src-tauri/src/commands.rs +++ b/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 { 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!( diff --git a/src-tauri/src/lib.rs b/src-tauri/src/lib.rs index 1da84fd..2c76843 100644 --- a/src-tauri/src/lib.rs +++ b/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) - // inside run_watchdog, after the notification subscription is live. + // Trainer: remember the device id and reset the previous trainer's targets. + // 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; - reapply_trainer_state(&app, &source, &device).await; + match source.request_control(&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::().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; diff --git a/src/main.js b/src/main.js index 73aad92..fa15a92 100644 --- a/src/main.js +++ b/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 });