fix(transport/ble): give the seam a stop_scanning, so a stop reaches the radio

`BleIo` had `stop_advertising` and no counterpart for scanning, so
`stop_async` stopped the advertisement and left the scan running. On BlueZ
that is harmless, because discovery ends when the scanner's event stream is
dropped. On a backend whose radio the embedder owns it is not: the transport
aborted its own scan task, nothing told the radio, and a scan started by a
transport that has since stopped ran for the life of the process. On a phone
that is battery and a continued broadcast of the user's presence, both after
the feature was switched off.

`AndroidRadio::stop_scanning` already existed and was called from nowhere.

Add `stop_scanning` to `BleIo` beside `stop_advertising` and call it from
`stop_async`. `BluerIo` implements it as a no-op that says why. `AndroidIo`
clears the scan intent and tells the radio through a new
`AndroidBleBridge::end_scanning`, mirroring `end_advertising`. Clearing the
intent matters as much as stopping the radio: without it a radio installed
after the transport stopped would be told to scan by `RadioIntent::apply`,
with nothing left to consume the adverts.

Putting it in the seam rather than in a `Drop` on the Android scanner is what
makes it available to the next backend. macOS has the same shape as Android
here -- CoreBluetooth owns the radio and has its own `stopScan` -- so this is
one more thing `io_macos.rs` inherits rather than rediscovers.

Both tests were checked by breaking what they guard: with the `stop_async`
call removed, and separately with `end_scanning`'s radio call removed, each
fails on the assertion that names the behaviour.

src/transport/ble/io.rs

@@ -163,6 +163,16 @@ pub trait BleIo: Send + Sync + 'static {
         &self,
     ) -> impl std::future::Future<Output = Result<Self::Scanner, TransportError>> + Send;
 
+    /// Stop scanning.
+    ///
+    /// The counterpart to [`Self::stop_advertising`], and needed for the same
+    /// reason: dropping the transport's scan task stops *us* reading adverts,
+    /// but on a backend whose radio is owned elsewhere it does not stop the
+    /// radio. A backend whose scan ends when its `Scanner` is dropped
+    /// implements this as a no-op and says so.
+    fn stop_scanning(&self)
+    -> impl std::future::Future<Output = Result<(), TransportError>> + Send;
+
     /// Get the adapter's BLE address.
     fn local_addr(&self) -> Result<BleAddr, TransportError>;
 
@@ -288,6 +298,8 @@ pub struct MockBleIo {
     bound_psm: std::sync::Mutex<Option<u16>>,
     /// PSM most recently passed to `start_advertising`.
     advertised_psm: std::sync::Mutex<Option<u16>>,
+    /// Number of times `stop_scanning` has been called.
+    stop_scans: std::sync::atomic::AtomicUsize,
 }
 
 impl MockBleIo {
@@ -305,9 +317,15 @@ impl MockBleIo {
             connect_handler: std::sync::Mutex::new(None),
             bound_psm: std::sync::Mutex::new(None),
             advertised_psm: std::sync::Mutex::new(None),
+            stop_scans: std::sync::atomic::AtomicUsize::new(0),
         }
     }
 
+    /// How many times the transport has asked this backend to stop scanning.
+    pub fn stop_scan_calls(&self) -> usize {
+        self.stop_scans.load(std::sync::atomic::Ordering::Relaxed)
+    }
+
     /// Inject an inbound connection (simulates a remote device connecting).
     pub async fn inject_inbound(&self, stream: MockBleStream) {
         let _ = self.accept_tx.send(stream).await;
@@ -393,6 +411,12 @@ impl BleIo for MockBleIo {
         Ok(())
     }
 
+    async fn stop_scanning(&self) -> Result<(), TransportError> {
+        self.stop_scans
+            .fetch_add(1, std::sync::atomic::Ordering::Relaxed);
+        Ok(())
+    }
+
     async fn start_scanning(&self) -> Result<Self::Scanner, TransportError> {
         let rx = self
             .scan_rx

src/transport/ble/io_android.rs

@@ -421,6 +421,18 @@ impl AndroidBleBridge {
         }
     }
 
+    /// Stop scanning, so a later request starts it again.
+    ///
+    /// The embedder's radio outlives this transport, so a scan nobody stops
+    /// runs for the life of the process: on a phone that is battery and a
+    /// broadcast of the user's presence, both after the feature was switched
+    /// off.
+    fn end_scanning(&self) {
+        if self.scanning.swap(false, Ordering::AcqRel) {
+            self.radio.stop_scanning();
+        }
+    }
+
     /// Allocate a channel, registering the bridge-side half and returning the
     /// stream-side half.
     fn make_channel(&self, remote: BleAddr, send_mtu: u16, recv_mtu: u16) -> StreamEndpoints {
@@ -986,6 +998,17 @@ impl BleIo for AndroidIo {
         Ok(())
     }
 
+    /// Clearing the intent matters as much as stopping the radio: without it a
+    /// radio installed after the transport stopped would be told to scan by
+    /// `RadioIntent::apply`, with nothing left to consume the adverts.
+    async fn stop_scanning(&self) -> Result<(), TransportError> {
+        self.intent.scan.store(false, Ordering::Relaxed);
+        if let Some(bridge) = self.slot.current() {
+            bridge.end_scanning();
+        }
+        Ok(())
+    }
+
     async fn start_scanning(&self) -> Result<AndroidScanner, TransportError> {
         self.intent.scan.store(true, Ordering::Relaxed);
         let mut seen = None;
@@ -1040,6 +1063,7 @@ mod tests {
         advertised_psm: AtomicU16,
         advertise_calls: AtomicU32,
         scan_calls: AtomicU32,
+        stop_scan_calls: AtomicU32,
         closed_channels: Mutex<Vec<i64>>,
         dials: Mutex<Vec<(i64, BleAddr, u16)>>,
     }
@@ -1075,7 +1099,9 @@ mod tests {
         fn start_scanning(&self) {
             self.scan_calls.fetch_add(1, Ordering::Relaxed);
         }
-        fn stop_scanning(&self) {}
+        fn stop_scanning(&self) {
+            self.stop_scan_calls.fetch_add(1, Ordering::Relaxed);
+        }
         fn close_channel(&self, ch_id: i64) {
             self.closed_channels.lock().unwrap().push(ch_id);
         }
@@ -1707,4 +1733,43 @@ mod tests {
         reader.read_exact(&mut tail).await.unwrap();
         assert_eq!(&tail, b"456789");
     }
+
+    /// The embedder's radio outlives the transport, so stopping the transport
+    /// has to reach the radio. Nothing did: `stop_scanning` was declared on
+    /// `AndroidRadio` and called from nowhere, so a scan started by a
+    /// transport that has since stopped ran for the life of the process.
+    #[tokio::test]
+    async fn stopping_the_transport_stops_the_radio_scanning() {
+        let radio = MockRadio::with_psm(0x0081);
+        let slot = Arc::new(BleRadioSlot::new());
+        slot.install(AndroidBleBridge::new(
+            Arc::clone(&radio) as Arc<dyn AndroidRadio>
+        ));
+        let io = AndroidIo::new(Arc::clone(&slot));
+
+        let _scanner = io.start_scanning().await.unwrap();
+        assert_eq!(radio.scan_calls.load(Ordering::Relaxed), 1);
+        assert_eq!(radio.stop_scan_calls.load(Ordering::Relaxed), 0);
+
+        io.stop_scanning().await.unwrap();
+        assert_eq!(
+            radio.stop_scan_calls.load(Ordering::Relaxed),
+            1,
+            "the radio must be told, not just the scanner dropped"
+        );
+
+        // The intent is cleared too, so a radio installed after the stop is
+        // not told to scan with nothing left to read the adverts.
+        let later = MockRadio::with_psm(0x0081);
+        slot.install(AndroidBleBridge::new(
+            Arc::clone(&later) as Arc<dyn AndroidRadio>
+        ));
+        let mut seen = None;
+        resolve(&slot, &mut seen, &io.intent);
+        assert_eq!(
+            later.scan_calls.load(Ordering::Relaxed),
+            0,
+            "a radio installed after the stop must not be told to scan"
+        );
+    }
 }

src/transport/ble/io_linux.rs

@@ -368,6 +368,14 @@ impl BleIo for BluerIo {
         Ok(())
     }
 
+    /// A no-op: BlueZ discovery ends when [`BluerScanner`]'s event stream is
+    /// dropped, which happens when the transport drops the scanner. There is
+    /// no separate adapter-level stop to issue.
+    async fn stop_scanning(&self) -> Result<(), TransportError> {
+        debug!("BLE scanning stops with the scanner");
+        Ok(())
+    }
+
     async fn start_scanning(&self) -> Result<Self::Scanner, TransportError> {
         // Clear cached devices so BlueZ fires DeviceAdded for every
         // advertisement. Without this, already-known devices only

src/transport/ble/mod.rs

@@ -326,6 +326,11 @@ impl<I: BleIo> BleTransport<I> {
         // Stop advertising
         let _ = self.io.stop_advertising().await;
 
+        // Stop scanning. Aborting the scan task below stops us reading
+        // adverts; on a backend whose radio the embedder owns, only this
+        // stops the radio.
+        let _ = self.io.stop_scanning().await;
+
         // Abort accept loop
         if let Some(task) = self.accept_task.take() {
             task.abort();
@@ -1938,6 +1943,34 @@ mod tests {
         assert_eq!(transport.state(), TransportState::Down);
     }
 
+    /// `stop_async` has always stopped advertising; it must stop scanning
+    /// too. Aborting the scan task only stops the transport reading adverts —
+    /// on a backend whose radio the embedder owns, the radio keeps scanning
+    /// until it is told, which on a phone costs battery and keeps
+    /// broadcasting after the feature was switched off.
+    #[tokio::test]
+    async fn stop_async_tells_the_backend_to_stop_scanning() {
+        let io = MockBleIo::new("hci0", test_addr(1));
+        let config = BleConfig {
+            adapter: Some("hci0".to_string()),
+            scan: Some(true),
+            advertise: Some(false),
+            accept_connections: Some(false),
+            ..Default::default()
+        };
+        let (tx, _rx) = tokio::sync::mpsc::channel(64);
+        let mut transport = BleTransport::new(TransportId::new(1), None, config, io, tx);
+        transport.start_async().await.unwrap();
+        assert_eq!(transport.io.stop_scan_calls(), 0);
+
+        transport.stop_async().await.unwrap();
+        assert_eq!(
+            transport.io.stop_scan_calls(),
+            1,
+            "stopping the transport must reach the backend's scan"
+        );
+    }
+
     #[tokio::test(start_paused = true)]
     async fn test_scan_discovers_peers() {
         let io = MockBleIo::new("hci0", test_addr(1));