diff --git a/crates/rns-interface/src/android_usb.rs b/crates/rns-interface/src/android_usb.rs index fe4a821..b0d2c34 100644 --- a/crates/rns-interface/src/android_usb.rs +++ b/crates/rns-interface/src/android_usb.rs @@ -16,7 +16,9 @@ use crate::android_usb_lifecycle::{ }; #[cfg(test)] use crate::kiss; -use crate::rnode::{self, RNodeRuntimeReason, RNodeTransportClass, SpawnedRNodeInterface}; +use crate::rnode::{ + self, RNodeDriverShutdown, RNodeRuntimeReason, RNodeTransportClass, SpawnedRNodeInterface, +}; use crate::rnode_protocol::RNodeProtocolTarget; use crate::traits::{ InterfaceDirection, InterfaceError, InterfaceHandle, InterfaceId, InterfaceMode, @@ -66,6 +68,7 @@ fn android_usb_rnode_stop_registry() -> &'static AndroidUsbRNodeStopRegistry { struct AndroidUsbRNodeStopRegistryGuard { id: InterfaceId, + stop_tx: mpsc::Sender<()>, status: watch::Receiver, } @@ -79,10 +82,15 @@ impl Drop for AndroidUsbRNodeStopRegistryGuard { // quarantine separately blocks a competing reopen. return; } - android_usb_rnode_stop_registry() + let mut registry = android_usb_rnode_stop_registry() .lock() - .unwrap_or_else(std::sync::PoisonError::into_inner) - .remove(&self.id); + .unwrap_or_else(std::sync::PoisonError::into_inner); + let owns_entry = registry + .get(&self.id) + .is_some_and(|registered| registered.stop_tx.same_channel(&self.stop_tx)); + if owns_entry { + registry.remove(&self.id); + } } } @@ -95,9 +103,16 @@ fn register_android_usb_rnode_stop( android_usb_rnode_stop_registry() .lock() .unwrap_or_else(std::sync::PoisonError::into_inner) - .insert(id, AndroidUsbStopHandle { stop_tx, status }); + .insert( + id, + AndroidUsbStopHandle { + stop_tx: stop_tx.clone(), + status, + }, + ); AndroidUsbRNodeStopRegistryGuard { id, + stop_tx, status: guard_status, } } @@ -124,8 +139,12 @@ fn request_android_usb_rnode_stop(id: InterfaceId) -> Option(64); let (stop_tx, mut stop_rx) = mpsc::channel::<()>(1); + let (mut snapshot_publisher, driver) = rnode::new_rnode_driver_observation_with_shutdown( + RNodeTransportClass::Usb, + RNodeDriverShutdown::from_stop_sender(stop_tx.clone()), + ); + snapshot_publisher.connection_established(); let (shutdown_status_tx, shutdown_status_rx) = watch::channel(AndroidUsbShutdownStatus::Running); let stop_guard = register_android_usb_rnode_stop(id, stop_tx, shutdown_status_rx); @@ -1427,6 +1447,47 @@ mod tests { let _usb_transport: RNodeTransportClass = RNodeTransportClass::Usb; } + #[test] + fn android_usb_exact_shutdown_and_cleanup_resist_same_id_aba() { + let id: InterfaceId = 0xA11D_0001; + let (old_tx, mut old_rx) = mpsc::channel::<()>(2); + let (_old_status_tx, old_status_rx) = watch::channel(AndroidUsbShutdownStatus::Running); + let (_old_publisher, old_driver) = rnode::new_rnode_driver_observation_with_shutdown( + RNodeTransportClass::Usb, + RNodeDriverShutdown::from_stop_sender(old_tx.clone()), + ); + let old_guard = register_android_usb_rnode_stop(id, old_tx, old_status_rx); + + let (new_tx, mut new_rx) = mpsc::channel::<()>(2); + let (_new_status_tx, new_status_rx) = watch::channel(AndroidUsbShutdownStatus::Running); + let (_new_publisher, _new_driver) = rnode::new_rnode_driver_observation_with_shutdown( + RNodeTransportClass::Usb, + RNodeDriverShutdown::from_stop_sender(new_tx.clone()), + ); + let new_guard = register_android_usb_rnode_stop(id, new_tx, new_status_rx); + + drop(old_guard); + old_driver.request_shutdown(); + assert!(old_rx.try_recv().is_ok()); + assert!( + matches!(new_rx.try_recv(), Err(mpsc::error::TryRecvError::Empty)), + "the retired handle must not stop the newer same-ID USB driver" + ); + + stop_android_usb_rnode_interface(id); + assert!( + new_rx.try_recv().is_ok(), + "retired guard cleanup must preserve the newer compatibility entry" + ); + drop(new_guard); + assert!( + !android_usb_rnode_stop_registry() + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .contains_key(&id) + ); + } + fn fragmented_decode(wire: &[u8]) -> Vec<(u8, Vec)> { let mut deframer = kiss::RawKissDeframer::new(); let fragment_widths = [1, 2, 1, 3, 2, 4]; diff --git a/crates/rns-interface/src/ble_rnode.rs b/crates/rns-interface/src/ble_rnode.rs index 26db681..3feccf6 100644 --- a/crates/rns-interface/src/ble_rnode.rs +++ b/crates/rns-interface/src/ble_rnode.rs @@ -29,8 +29,8 @@ pub fn is_btleplug_initialized() -> bool { use crate::kiss; use crate::rnode::{ - self, RNodeResponse, RNodeRuntimeReason, RNodeSnapshotPublisher, RNodeTransportClass, - SpawnedRNodeInterface, + self, RNodeDriverShutdown, RNodeResponse, RNodeRuntimeReason, RNodeSnapshotPublisher, + RNodeTransportClass, SpawnedRNodeInterface, }; use crate::rnode_protocol::{RNodeProtocolState, RNodeProtocolTarget}; use crate::traits::{ @@ -76,14 +76,23 @@ fn register_running(id: InterfaceId) -> Arc { flag } -fn unregister_running(id: InterfaceId) { +fn unregister_running(id: InterfaceId, running: &Arc) { if let Ok(mut map) = running_map().lock() { - map.remove(&id); + let owns_entry = map + .get(&id) + .is_some_and(|registered| Arc::ptr_eq(registered, running)); + if owns_entry { + map.remove(&id); + } } } -/// Idempotent. Safe to call before or after deregistering from the -/// transport actor. +/// Compatibility facade requesting shutdown of the currently registered BLE +/// RNode for `id`. +/// +/// New owners should retain [`crate::rnode::RNodeDriverHandle`] and call +/// [`crate::rnode::RNodeDriverHandle::request_shutdown`] so later ID reuse +/// cannot redirect the request. pub fn stop_ble_rnode_interface(id: InterfaceId) { if let Ok(map) = running_map().lock() { if let Some(flag) = map.get(&id) { @@ -1300,8 +1309,6 @@ pub async fn spawn_ble_rnode_interface_with_driver( config.coding_rate, config.tx_power, ); - let (snapshot_publisher, driver) = - rnode::new_rnode_driver_observation(RNodeTransportClass::Ble); let online = Arc::new(AtomicBool::new(false)); let online_handle = online.clone(); let shared_rxb = Arc::new(AtomicU64::new(0)); @@ -1311,6 +1318,10 @@ pub async fn spawn_ble_rnode_interface_with_driver( let (tx, rx) = mpsc::channel::(256); let rx = Arc::new(tokio::sync::Mutex::new(rx)); let running = register_running(id); + let (snapshot_publisher, driver) = rnode::new_rnode_driver_observation_with_shutdown( + RNodeTransportClass::Ble, + RNodeDriverShutdown::from_running_flag(running.clone()), + ); let bitrate = rnode::calculate_bitrate( config.spreading_factor, @@ -1336,13 +1347,13 @@ pub async fn spawn_ble_rnode_interface_with_driver( // Drop guard: every early return must clear the running-flag map // entry, or stale entries confuse later spawns reusing the id. - struct Cleanup(InterfaceId); + struct Cleanup(InterfaceId, Arc); impl Drop for Cleanup { fn drop(&mut self) { - unregister_running(self.0); + unregister_running(self.0, &self.1); } } - let _cleanup = Cleanup(id); + let _cleanup = Cleanup(id, running_task.clone()); loop { if !running_task.load(Ordering::SeqCst) { @@ -1763,8 +1774,6 @@ pub async fn spawn_ble_rnode_interface_native_with_driver( config.coding_rate, config.tx_power, ); - let (snapshot_publisher, driver) = - rnode::new_rnode_driver_observation(RNodeTransportClass::Ble); let online = Arc::new(AtomicBool::new(false)); let online_handle = online.clone(); let shared_rxb = Arc::new(AtomicU64::new(0)); @@ -1789,6 +1798,10 @@ pub async fn spawn_ble_rnode_interface_native_with_driver( let ble_uri = config.ble_uri.clone(); let log_name = name.clone(); let running = register_running(id); + let (snapshot_publisher, driver) = rnode::new_rnode_driver_observation_with_shutdown( + RNodeTransportClass::Ble, + RNodeDriverShutdown::from_running_flag(running.clone()), + ); let running_task = running.clone(); let read_task = tokio::spawn(async move { @@ -1797,13 +1810,13 @@ pub async fn spawn_ble_rnode_interface_native_with_driver( let mut backoff = RECONNECT_WAIT; let mut initial_attempt = true; - struct Cleanup(InterfaceId); + struct Cleanup(InterfaceId, Arc); impl Drop for Cleanup { fn drop(&mut self) { - unregister_running(self.0); + unregister_running(self.0, &self.1); } } - let _cleanup = Cleanup(id); + let _cleanup = Cleanup(id, running_task.clone()); loop { if !running_task.load(Ordering::SeqCst) { @@ -2218,7 +2231,7 @@ mod tests { let flag = register_running(id); assert!(is_registered(id)); assert!(flag.load(Ordering::SeqCst)); - unregister_running(id); + unregister_running(id, &flag); assert!(!is_registered(id)); } @@ -2230,7 +2243,7 @@ mod tests { stop_ble_rnode_interface(id); assert!(!flag.load(Ordering::SeqCst)); // Map entry survives until the owning task's Drop runs; clean up. - unregister_running(id); + unregister_running(id, &flag); } #[test] @@ -2241,6 +2254,41 @@ mod tests { assert!(!is_registered(id)); } + #[test] + fn test_ble_exact_shutdown_and_cleanup_resist_same_id_aba() { + let id: InterfaceId = 0xDEAD_BEEF_0000_0004; + let old_running = register_running(id); + let (_old_publisher, old_driver) = rnode::new_rnode_driver_observation_with_shutdown( + RNodeTransportClass::Ble, + RNodeDriverShutdown::from_running_flag(old_running.clone()), + ); + + let new_running = register_running(id); + let (_new_publisher, _new_driver) = rnode::new_rnode_driver_observation_with_shutdown( + RNodeTransportClass::Ble, + RNodeDriverShutdown::from_running_flag(new_running.clone()), + ); + + unregister_running(id, &old_running); + assert!( + is_registered(id), + "retired task cleanup must preserve the newer compatibility entry" + ); + old_driver.request_shutdown(); + assert!(!old_running.load(Ordering::SeqCst)); + assert!( + new_running.load(Ordering::SeqCst), + "the retired handle must not stop the newer same-ID BLE driver" + ); + + stop_ble_rnode_interface(id); + assert!( + !new_running.load(Ordering::SeqCst), + "the compatibility facade must still stop the current registration" + ); + unregister_running(id, &new_running); + } + #[tokio::test] async fn test_wait_or_shutdown_returns_false_when_flag_stays_set() { // Short real-time wait; flag stays true → full duration elapses. diff --git a/crates/rns-interface/src/rnode.rs b/crates/rns-interface/src/rnode.rs index 80ca2e2..6a79a70 100644 --- a/crates/rns-interface/src/rnode.rs +++ b/crates/rns-interface/src/rnode.rs @@ -30,14 +30,17 @@ use crate::{rnode_protocol::RNodeProtocolTarget, traits::InterfaceDirection}; use std::collections::HashMap; use std::sync::Arc; #[cfg(any(feature = "serial", feature = "rnode-tcp"))] -use std::sync::atomic::{AtomicBool, AtomicU64, Ordering}; +use std::sync::atomic::AtomicU64; +use std::sync::atomic::{AtomicBool, Ordering}; #[cfg(any(feature = "serial", feature = "rnode-tcp"))] use std::sync::{Mutex, OnceLock}; #[cfg(any(feature = "serial", feature = "rnode-tcp"))] use std::time::Duration; -use tokio::sync::watch; +#[cfg(any(feature = "serial", feature = "rnode-tcp", target_os = "android", test))] +use tokio::sync::mpsc; #[cfg(any(feature = "serial", feature = "rnode-tcp"))] -use tokio::sync::{mpsc, oneshot}; +use tokio::sync::oneshot; +use tokio::sync::watch; pub const CMD_FREQUENCY: u8 = 0x01; pub const CMD_BANDWIDTH: u8 = 0x02; @@ -425,10 +428,93 @@ fn project_rnode_protocol_effect( *snapshot != before } -/// Cloneable, observation-only handle for a generic RNode driver. +enum RNodeDriverShutdownSignal { + #[cfg(test)] + InertTest, + #[cfg(any(feature = "serial", feature = "rnode-tcp", target_os = "android", test))] + StopSender(mpsc::Sender<()>), + #[cfg(feature = "ble")] + RunningFlag(Arc), +} + +struct RNodeDriverShutdownInner { + requested: AtomicBool, + signal: RNodeDriverShutdownSignal, +} + +/// Exact-instance shutdown request shared by every clone of a driver handle. +/// +/// This stays crate-private so observed drivers can expose only the one +/// bounded lifecycle action, never arbitrary RNode controls. +#[derive(Clone)] +pub(crate) struct RNodeDriverShutdown { + inner: Arc, +} + +impl RNodeDriverShutdown { + #[cfg(test)] + fn inert_test() -> Self { + Self::new(RNodeDriverShutdownSignal::InertTest) + } + + #[cfg(any(feature = "serial", feature = "rnode-tcp", target_os = "android", test))] + pub(crate) fn from_stop_sender(stop_tx: mpsc::Sender<()>) -> Self { + Self::new(RNodeDriverShutdownSignal::StopSender(stop_tx)) + } + + #[cfg(feature = "ble")] + pub(crate) fn from_running_flag(running: Arc) -> Self { + Self::new(RNodeDriverShutdownSignal::RunningFlag(running)) + } + + #[cfg(any( + feature = "serial", + feature = "rnode-tcp", + feature = "ble", + target_os = "android", + test + ))] + fn new(signal: RNodeDriverShutdownSignal) -> Self { + Self { + inner: Arc::new(RNodeDriverShutdownInner { + requested: AtomicBool::new(false), + signal, + }), + } + } + + fn request(&self) { + if self.inner.requested.swap(true, Ordering::SeqCst) { + return; + } + match &self.inner.signal { + #[cfg(test)] + RNodeDriverShutdownSignal::InertTest => {} + #[cfg(any(feature = "serial", feature = "rnode-tcp", target_os = "android", test))] + RNodeDriverShutdownSignal::StopSender(stop_tx) => { + let _ = stop_tx.try_send(()); + } + #[cfg(feature = "ble")] + RNodeDriverShutdownSignal::RunningFlag(running) => { + running.store(false, Ordering::SeqCst); + } + #[cfg(not(any( + feature = "serial", + feature = "rnode-tcp", + feature = "ble", + target_os = "android", + test + )))] + _ => unreachable!("no RNode driver transport can construct a shutdown primitive"), + } + } +} + +/// Cloneable, privacy-safe lifecycle handle for one exact RNode driver. #[derive(Clone)] pub struct RNodeDriverHandle { state: watch::Receiver>, + shutdown: RNodeDriverShutdown, } /// Clone-only subscription to generic RNode driver observations. @@ -453,6 +539,16 @@ impl RNodeDriverHandle { state: self.state.clone(), } } + + /// Request shutdown of this exact spawned driver instance. + /// + /// The request is idempotent across every clone. It never resolves an + /// interface ID or another global registry entry, so later reuse of the + /// same ID cannot redirect it to a different session. Completion remains + /// owned by the spawned interface task and its normal join path. + pub fn request_shutdown(&self) { + self.shutdown.request(); + } } impl RNodeDriverSubscription { @@ -668,6 +764,13 @@ impl Drop for RNodeSnapshotPublisher { } } +#[cfg(test)] +pub(crate) fn new_rnode_driver_observation( + transport: RNodeTransportClass, +) -> (RNodeSnapshotPublisher, RNodeDriverHandle) { + new_rnode_driver_observation_with_shutdown(transport, RNodeDriverShutdown::inert_test()) +} + #[cfg(any( feature = "serial", feature = "rnode-tcp", @@ -675,13 +778,17 @@ impl Drop for RNodeSnapshotPublisher { target_os = "android", test ))] -pub(crate) fn new_rnode_driver_observation( +pub(crate) fn new_rnode_driver_observation_with_shutdown( transport: RNodeTransportClass, + shutdown: RNodeDriverShutdown, ) -> (RNodeSnapshotPublisher, RNodeDriverHandle) { let (state_tx, state_rx) = watch::channel(Arc::new(RNodeRuntimeSnapshot::initial(transport))); ( RNodeSnapshotPublisher::new(state_tx), - RNodeDriverHandle { state: state_rx }, + RNodeDriverHandle { + state: state_rx, + shutdown, + }, ) } @@ -697,15 +804,21 @@ fn rnode_stop_registry() -> &'static RNodeStopRegistry { #[cfg(any(feature = "serial", feature = "rnode-tcp"))] struct RNodeStopRegistryGuard { id: InterfaceId, + stop_tx: mpsc::Sender<()>, } #[cfg(any(feature = "serial", feature = "rnode-tcp"))] impl Drop for RNodeStopRegistryGuard { fn drop(&mut self) { - rnode_stop_registry() + let mut registry = rnode_stop_registry() .lock() - .expect("rnode_stop_registry mutex poisoned") - .remove(&self.id); + .expect("rnode_stop_registry mutex poisoned"); + let owns_entry = registry + .get(&self.id) + .is_some_and(|registered| registered.same_channel(&self.stop_tx)); + if owns_entry { + registry.remove(&self.id); + } } } @@ -714,12 +827,16 @@ fn register_rnode_stop(id: InterfaceId, stop_tx: mpsc::Sender<()>) -> RNodeStopR rnode_stop_registry() .lock() .expect("rnode_stop_registry mutex poisoned") - .insert(id, stop_tx); - RNodeStopRegistryGuard { id } + .insert(id, stop_tx.clone()); + RNodeStopRegistryGuard { id, stop_tx } } -/// Ask a serial/TCP RNode interface to send upstream's detach sequence before -/// runtime teardown aborts the task. Idempotent; unknown ids are ignored. +/// Compatibility facade requesting shutdown of the currently registered +/// serial/TCP RNode for `id`. +/// +/// New owners should retain [`RNodeDriverHandle`] and call +/// [`RNodeDriverHandle::request_shutdown`] so later ID reuse cannot redirect +/// the request. Unknown IDs are ignored. #[cfg(any(feature = "serial", feature = "rnode-tcp"))] pub fn stop_rnode_interface(id: InterfaceId) { let stop_tx = rnode_stop_registry() @@ -2406,7 +2523,6 @@ pub async fn spawn_rnode_interface_with_driver( }; let port = open_configured_rnode_stream(&config, &port_cfg).await?; - let (initial_snapshot_publisher, driver) = new_rnode_driver_observation(transport); let bitrate = calculate_bitrate( config.spreading_factor, @@ -2449,6 +2565,10 @@ pub async fn spawn_rnode_interface_with_driver( start_rnode_generation(port, &config, id, &online, &shared_txb, &beacon).await?; let (stop_tx, mut stop_rx) = mpsc::channel::<()>(1); + let (initial_snapshot_publisher, driver) = new_rnode_driver_observation_with_shutdown( + transport, + RNodeDriverShutdown::from_stop_sender(stop_tx.clone()), + ); let stop_guard = register_rnode_stop(id, stop_tx); let online_r = online.clone(); let rxb_r = shared_rxb.clone(); @@ -3297,6 +3417,26 @@ mod tests { ); } + #[test] + fn test_driver_shutdown_is_idempotent_across_clones() { + let (stop_tx, mut stop_rx) = mpsc::channel::<()>(4); + let (_publisher, driver) = new_rnode_driver_observation_with_shutdown( + RNodeTransportClass::Tcp, + RNodeDriverShutdown::from_stop_sender(stop_tx), + ); + let clone = driver.clone(); + + driver.request_shutdown(); + clone.request_shutdown(); + driver.request_shutdown(); + + assert!(stop_rx.try_recv().is_ok()); + assert!( + matches!(stop_rx.try_recv(), Err(mpsc::error::TryRecvError::Empty)), + "all handle clones must share one shutdown request" + ); + } + #[cfg(any(feature = "serial", feature = "rnode-tcp"))] #[test] fn test_stop_rnode_interface_signals_registered_driver() { @@ -3311,6 +3451,40 @@ mod tests { stop_rnode_interface(id); } + #[cfg(any(feature = "serial", feature = "rnode-tcp"))] + #[test] + fn test_exact_shutdown_and_registry_cleanup_resist_same_id_aba() { + let id = 0x0BAD_5701; + let (old_tx, mut old_rx) = mpsc::channel::<()>(2); + let (_old_publisher, old_driver) = new_rnode_driver_observation_with_shutdown( + RNodeTransportClass::Tcp, + RNodeDriverShutdown::from_stop_sender(old_tx.clone()), + ); + let old_guard = register_rnode_stop(id, old_tx); + + let (new_tx, mut new_rx) = mpsc::channel::<()>(2); + let (_new_publisher, _new_driver) = new_rnode_driver_observation_with_shutdown( + RNodeTransportClass::Tcp, + RNodeDriverShutdown::from_stop_sender(new_tx.clone()), + ); + let new_guard = register_rnode_stop(id, new_tx); + + drop(old_guard); + old_driver.request_shutdown(); + assert!(old_rx.try_recv().is_ok()); + assert!( + matches!(new_rx.try_recv(), Err(mpsc::error::TryRecvError::Empty)), + "the retired handle must not stop the newer same-ID driver" + ); + + stop_rnode_interface(id); + assert!( + new_rx.try_recv().is_ok(), + "retired guard cleanup must preserve the newer compatibility entry" + ); + drop(new_guard); + } + #[cfg(any(feature = "serial", feature = "rnode-tcp"))] #[tokio::test] async fn test_rnode_writer_startup_has_two_flush_acked_stages() { @@ -4909,7 +5083,10 @@ mod tests { assert_eq!(snapshot.phase, RNodeRuntimePhase::Ready); let (state_tx, state_rx) = watch::channel(Arc::new(snapshot.clone())); - let driver = RNodeDriverHandle { state: state_rx }; + let driver = RNodeDriverHandle { + state: state_rx, + shutdown: RNodeDriverShutdown::inert_test(), + }; let publisher = RNodeSnapshotPublisher::new(state_tx); let subscription = driver.watch(); let retained = subscription.snapshot();