Harden Central::connect against Android 133 zombies and hangs

CHANGELOG.md

@@ -5,8 +5,52 @@ All notable changes to `blew` are documented here. Format follows
 
 ## [Unreleased]
 
+### Added
+
+- `CentralConfig::connect_timeout: Option<Duration>` — deadline applied to
+  `Central::connect()`. `Default` sets it to 15s; `None` restores pre-0.3
+  unbounded-wait behavior. Applied uniformly across Apple, Linux, and Android
+  backends; emits `CentralEvent::DeviceDisconnected { cause: Timeout }` when
+  it fires.
+- `BlewError::ConnectTimedOut(DeviceId)` — returned from `Central::connect`
+  when the new deadline elapses. Distinct from `BlewError::Timeout`, which
+  remains reserved for adapter-readiness waits (`wait_ready` / `wait_powered`).
+- `BlewError::ConnectInFlight(DeviceId)` — returned when `Central::connect`
+  is called for a device that already has a connect in flight. Previously the
+  second caller silently stole the first caller's completion.
+
+### Changed
+
+- **Android `Central::disconnect` now awaits the `onConnectionStateChange`
+  callback** (with a 2-second fallback) instead of returning `Ok` immediately
+  after dispatching the JNI call. This prevents zombie `BluetoothGatt`
+  handles when the callback never arrives — the fallback path calls a new
+  Kotlin `forceClose(addr)` that flushes the service cache (`refresh()`) and
+  closes the GATT handle synchronously, freeing the client-IF slot.
+- **Android post-133 cleanup.** `onConnectionStateChange(DISCONNECTED,
+  status=133)` now calls the hidden `BluetoothGatt.refresh()` before
+  `close()` to flush the client-side service cache. The `connect()`
+  stale-cleanup path also calls `refresh()` on the old handle and waits
+  ~300ms before issuing the fresh `connectGatt()`, matching the canonical
+  Android BLE back-off recommendation.
+- Linux `Central::connect` deadline is now config-driven (via
+  `CentralConfig::connect_timeout`) and returns `BlewError::ConnectTimedOut`
+  on elapse. Previously it was hardcoded to 30s and returned the generic
+  `BlewError::Timeout`.
+
 ### Fixed
 
+- `Central::connect()` could hang indefinitely on Android if
+  `onConnectionStateChange` never fired (e.g. radio busy, adapter thrash,
+  status-133 zombie). The new `connect_timeout` default now bounds this;
+  callers that need the old behavior can opt in with
+  `CentralConfig { connect_timeout: None, .. }`.
+- Overlapping `Central::connect()` calls for the same device no longer
+  silently orphan the first caller's completion channel. Android, Apple:
+  the second caller receives `BlewError::ConnectInFlight` immediately.
+- Apple `Central::connect` silently evicted any pending completion channel
+  when called twice concurrently for the same device, leaving the first
+  caller hanging. Now rejected with `ConnectInFlight`.
 - `tauri-plugin-blew` no longer fails to locate the Android Kotlin sources when
   consumed from crates.io. Previously `build.rs` resolved `../blew/android`
   relative to its own manifest, which works in the workspace but points to a
@@ -177,6 +221,61 @@ These were `pub use`'d from `util` but are removed. If you reached into them
 
 ---
 
+## Upgrade guide — 0.2.x → Unreleased
+
+**If you were constructing `CentralConfig` as a struct literal**, it gains a
+new field:
+
+```rust
+// Before
+let config = CentralConfig { restore_identifier: Some("...".into()) };
+
+// After — either spread the default or set the new field explicitly
+let config = CentralConfig {
+    restore_identifier: Some("...".into()),
+    ..CentralConfig::default()
+};
+```
+
+**If you were matching on `BlewError::Timeout` for a connect failure on
+Linux**, the variant narrowed to `BlewError::ConnectTimedOut(DeviceId)`:
+
+```rust
+// Before — Linux connect timeout returned the generic variant
+match central.connect(&id).await {
+    Err(BlewError::Timeout) => /* ... */,
+    _ => /* ... */,
+}
+
+// After
+match central.connect(&id).await {
+    Err(BlewError::ConnectTimedOut(_)) => /* ... */,
+    _ => /* ... */,
+}
+```
+
+`BlewError::Timeout` is still used by `Central::wait_ready`,
+`Peripheral::wait_ready`, and `wait_powered`.
+
+**If you relied on unbounded connect waits**, set
+`CentralConfig.connect_timeout = None` when constructing the central:
+
+```rust
+let central: Central = Central::with_config(CentralConfig {
+    connect_timeout: None,
+    ..CentralConfig::default()
+}).await?;
+```
+
+**If overlapping `connect()` calls were previously "works by luck"**: the
+second caller now receives `BlewError::ConnectInFlight(DeviceId)` immediately
+on Apple and Android. If you need shared-completion fan-out semantics,
+implement them at your layer (e.g. wrap the shared future in
+`futures::future::Shared`). Linux still permits concurrent connects via
+bluer's own state machine.
+
+---
+
 ## [0.1.0]
 
 Initial release.

crates/blew/android/src/main/java/org/jakebot/blew/BleCentralManager.kt

@@ -16,6 +16,7 @@ import kotlinx.coroutines.CancellationException
 import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.Dispatchers
 import kotlinx.coroutines.SupervisorJob
+import kotlinx.coroutines.delay
 import kotlinx.coroutines.launch
 import java.util.UUID
 import java.util.concurrent.ConcurrentHashMap
@@ -309,6 +310,14 @@ object BleCentralManager {
                     noResponseHandled.remove(addr)
                     pendingNonces.keys.removeIf { it.startsWith("$addr:") }
                     gattQueues.remove(addr)?.close(CancellationException("device $addr disconnected"))
+                    // Status 133 is the Android BLE zombie signal. Flush the
+                    // client-side service cache before close() so the next
+                    // connectGatt() on this address starts with a clean slate.
+                    if (status == 133) {
+                        if (refreshGatt(gatt)) {
+                            Log.d(TAG, "flushed GATT cache for $addr after status=133")
+                        }
+                    }
                     gatt.close()
                     nativeOnConnectionStateChanged(addr, false, status)
                 }
@@ -413,16 +422,34 @@ object BleCentralManager {
             }
 
         // Close any stale GATT connection to avoid leaking clientIf slots.
-        // Android has a limit of ~7 concurrent GATT clients.
-        gattConnections.remove(deviceAddr)?.let { oldGatt ->
-            oldGatt.disconnect()
-            oldGatt.close()
+        // Android has a limit of ~7 concurrent GATT clients. If we had a stale
+        // handle, flush its service cache and give the stack ~300ms to release
+        // the client-IF before the next connectGatt — back-to-back attempts on
+        // the same address can be silently dropped on some vendors.
+        val stale = gattConnections.remove(deviceAddr)
+        if (stale != null) {
+            refreshGatt(stale)
+            stale.disconnect()
+            stale.close()
             Log.d(TAG, "closed stale GATT for $deviceAddr")
+            scope.launch {
+                delay(300)
+                openGatt(ctx, deviceAddr)
+            }
+            return
         }
 
+        openGatt(ctx, deviceAddr)
+    }
+
+    private fun openGatt(
+        ctx: Context,
+        deviceAddr: String,
+    ) {
         val device =
             adapter?.getRemoteDevice(deviceAddr) ?: run {
                 Log.e(TAG, "could not get remote device $deviceAddr")
+                nativeOnConnectionStateChanged(deviceAddr, false, 0)
                 return
             }
         // TRANSPORT_LE ensures we connect over BLE, not classic Bluetooth.
@@ -443,6 +470,46 @@ object BleCentralManager {
         }
     }
 
+    /**
+     * Synchronously tear down the GATT handle for [deviceAddr] without waiting
+     * for [BluetoothGattCallback.onConnectionStateChange]. Called from Rust
+     * when the normal disconnect callback path cannot be trusted (connect
+     * timeout, disconnect whose callback never arrived, status-133 zombie).
+     *
+     * Flushes the client-side service cache with `refresh()` before closing
+     * so the next connectGatt() starts clean. Emits a synthetic
+     * [nativeOnConnectionStateChanged]`(addr, false, 0)` so any Rust-side
+     * state waiting on the disconnect callback unblocks.
+     */
+    @JvmStatic
+    fun forceClose(deviceAddr: String) {
+        val gatt = gattConnections.remove(deviceAddr)
+        mtuMap.remove(deviceAddr)
+        noResponseHandled.remove(deviceAddr)
+        pendingNonces.keys.removeIf { it.startsWith("$deviceAddr:") }
+        gattQueues.remove(deviceAddr)?.close(CancellationException("device $deviceAddr force-closed"))
+        if (gatt != null) {
+            refreshGatt(gatt)
+            try {
+                gatt.disconnect()
+            } catch (e: Exception) {
+                Log.w(TAG, "forceClose: disconnect threw for $deviceAddr: ${e.message}")
+            }
+            gatt.close()
+            Log.d(TAG, "forceClose: tore down GATT for $deviceAddr")
+        }
+        nativeOnConnectionStateChanged(deviceAddr, false, 0)
+    }
+
+    private fun refreshGatt(gatt: BluetoothGatt): Boolean =
+        try {
+            val method = gatt.javaClass.getMethod("refresh")
+            method.invoke(gatt) as Boolean
+        } catch (e: Exception) {
+            Log.w(TAG, "refresh failed: ${e.message}")
+            false
+        }
+
     /**
      * Clear the GATT service cache for [deviceAddr] by invoking the hidden
      * `BluetoothGatt.refresh()` method via reflection. Returns false if no
@@ -453,13 +520,7 @@ object BleCentralManager {
     @JvmStatic
     fun refresh(deviceAddr: String): Boolean {
         val gatt = gattConnections[deviceAddr] ?: return false
-        return try {
-            val method = gatt.javaClass.getMethod("refresh")
-            method.invoke(gatt) as Boolean
-        } catch (e: Exception) {
-            Log.w(TAG, "refresh failed for $deviceAddr: ${e.message}")
-            false
-        }
+        return refreshGatt(gatt)
     }
 
     // ── GATT operations (serialized via per-device queue) ──

crates/blew/examples/restore.rs

@@ -51,6 +51,7 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
 
     let central: Central = Central::with_config(CentralConfig {
         restore_identifier: Some(CENTRAL_RESTORE_ID.into()),
+        ..CentralConfig::default()
     })
     .await?;
 

crates/blew/src/central/types.rs

@@ -1,14 +1,34 @@
 use crate::types::{BleDevice, DeviceId};
 use bytes::Bytes;
+use std::time::Duration;
 use uuid::Uuid;
 
+/// Default deadline for [`Central::connect`](crate::central::Central::connect).
+/// Covers ACL link-up, optional SMP re-bond, service discovery, and the
+/// per-backend post-connect steps (Android's MTU negotiation in particular).
+pub const DEFAULT_CONNECT_TIMEOUT: Duration = Duration::from_secs(15);
+
 /// Configuration for initialising the central role.
-#[derive(Debug, Clone, Default)]
+#[derive(Debug, Clone)]
 pub struct CentralConfig {
     /// On Apple platforms, passed as `CBCentralManagerOptionRestoreIdentifierKey` to
     /// `initWithDelegate:queue:options:`, enabling state restoration for the app's
     /// background BLE central session. Ignored on all other platforms.
     pub restore_identifier: Option<String>,
+    /// Deadline applied to [`Central::connect`](crate::central::Central::connect).
+    /// `None` disables the deadline (pre-0.3 behaviour); `Some(d)` returns
+    /// [`BlewError::ConnectTimedOut`](crate::error::BlewError::ConnectTimedOut)
+    /// if the connection has not completed within `d`.
+    pub connect_timeout: Option<Duration>,
+}
+
+impl Default for CentralConfig {
+    fn default() -> Self {
+        Self {
+            restore_identifier: None,
+            connect_timeout: Some(DEFAULT_CONNECT_TIMEOUT),
+        }
+    }
 }
 
 /// Reason a peripheral disconnected. Used by [`CentralEvent::DeviceDisconnected`].

crates/blew/src/error.rs

@@ -40,6 +40,12 @@ pub enum BlewError {
     #[error("operation timed out")]
     Timeout,
 
+    #[error("connect to {0} timed out")]
+    ConnectTimedOut(DeviceId),
+
+    #[error("connect to {0} already in flight")]
+    ConnectInFlight(DeviceId),
+
     #[error("GATT busy on device {0} (another operation in flight)")]
     GattBusy(DeviceId),