fix: close six reachable contradictions in implicit state machines

CHANGELOG.md

@@ -106,6 +106,38 @@ archived by series under [docs/changelog/](docs/changelog/); see the
 
 ### Fixed
 
+- **An iPhone with Bluetooth off no longer tells the engine Bluetooth is up.**
+  `start()` reported Bluetooth available at once and again a second later,
+  whatever the radio's state. With Bluetooth off, unauthorized or unsupported,
+  the Bluetooth manager reports it down, and the second report then marked it
+  up again, so the engine routed sends to a radio that was off until the
+  radio next changed state. Neither report is made while the radio is
+  unavailable. Coming to the foreground likewise restated the Wi-Fi Direct
+  layer up when that transport was disabled, stopped or had failed to start;
+  it now does so only while the transport runs.
+
+- **An event callback may replace itself, and one that throws no longer
+  costs the polling queue an event.** Events the bindings raise themselves
+  (neighbor discovered and lost, signal strength, carrier status) reached the
+  callback with its slot locked and no isolation, unlike events from the
+  engine: a callback that called `set_event_callback` deadlocked, and one
+  that threw skipped the polling queue. Both kinds are now delivered the same
+  way.
+
+- **Deleting a 1:1 session drops the Welcome still waiting for it.**
+  `mls_get_pending_welcome` kept returning, after the session was deleted, an
+  invitation to the group that no longer existed.
+
+- **A browser Bluetooth link-up refused before `initialize` leaves no link
+  behind.** The address stayed recorded as linked, so its fragments were
+  accepted and the next link-up was taken for a relink and never announced
+  the peer.
+
+- **Android `start()` with no protocol created rejects instead of putting up
+  the mesh service.** It started the foreground service over no engine, which
+  a sticky restart then kept alive. The JavaScript wrapper never does this; a
+  direct caller of the native module could.
+
 - **A second SDK app on an Android phone gets its own LAN streams.** The
   peer-stream listener bound the fixed port 8988, so a second app on the
   same phone failed to bind it and still advertised 8988 on the LAN: every

bindings/react-native/android/src/main/java/com/offlineprotocol/OfflineProtocolModule.kt

@@ -1020,6 +1020,14 @@ class OfflineProtocolModule(reactContext: ReactApplicationContext) :
     @ReactMethod
     fun start(promise: Promise) {
         try {
+            // Refused before anything below runs. start() used to call
+            // protocol?.start() and then startForegroundService() regardless,
+            // so a start with no protocol put "Mesh Active" up and registered a
+            // host over no engine, which a sticky restart then read as
+            // KEEP_ALIVE. The JavaScript wrapper never does this; a caller of
+            // the native module directly could.
+            val proto = protocol ?: throw IllegalStateException("Protocol not initialized")
+
             // A one-shot event held from the *previous* mesh session is moot the
             // moment a new one comes up, and redelivering it would be the bug
             // this whole mechanism exists to prevent, inverted: an app told
@@ -1058,7 +1066,7 @@ class OfflineProtocolModule(reactContext: ReactApplicationContext) :
             // for the previous session belongs to whatever identity ran it.
             inboundEvents.beginSession()
             emitDiagnostic("info", "Starting protocol")
-            protocol?.start()
+            proto.start()
             emitDiagnostic("info", "Protocol core started")
 
             // Start foreground service to protect the process from being killed

bindings/react-native/ios/OfflineProtocolModule.swift

@@ -479,8 +479,16 @@ class OfflineProtocolModule: RCTEventEmitter {
         // transition) — and onto the same serial queue, which is what keeps
         // the false/true pair from reordering. No background assertion here:
         // the app is becoming active, not losing its runtime.
-        processQueue.async {
-            try? proto.wifiDirectStatusChanged(isConnected: true)
+        //
+        // Only when the peer-stream transport is running. The background edge
+        // reports the layer down unconditionally, which is always true; this
+        // edge used to report it up unconditionally too, so a transport that
+        // was disabled, stopped or had failed to start was restated as up on
+        // every foreground. The manager reports its own start and stop.
+        if wifiDirectManager?.state == .running {
+            processQueue.async {
+                try? proto.wifiDirectStatusChanged(isConnected: true)
+            }
         }
         // On `telemetryQueue`, not `processQueue`: it has to be the same queue
         // the background edge used, or `.active` could be delivered while the
@@ -1099,12 +1107,21 @@ class OfflineProtocolModule: RCTEventEmitter {
 
                     //  Ensure bleStatusChanged(true) is called immediately and as backup
                     // This ensures BLE transport is marked as Available for message sending
-                    do {
-                        try protocolInstance?.bleStatusChanged(isAvailable: true)
-                        print("[OfflineProtocolModule] ✅ Called protocol.bleStatusChanged(true) immediately")
-                        emitDiagnostic(level: "info", message: "BLE status set to available")
-                    } catch {
-                        print("[OfflineProtocolModule] Immediate bleStatusChanged failed: \(error.localizedDescription)")
+                    //
+                    // Never over a radio the manager has reported down. A
+                    // powered-off, unauthorized or unsupported radio moves the
+                    // manager to `.unavailable` and reports false; the backup
+                    // below used to land a second later and report true over
+                    // it, so the core routed sends to a radio that was off
+                    // until the next radio state change.
+                    if manager.state != .unavailable {
+                        do {
+                            try protocolInstance?.bleStatusChanged(isAvailable: true)
+                            print("[OfflineProtocolModule] ✅ Called protocol.bleStatusChanged(true) immediately")
+                            emitDiagnostic(level: "info", message: "BLE status set to available")
+                        } catch {
+                            print("[OfflineProtocolModule] Immediate bleStatusChanged failed: \(error.localizedDescription)")
+                        }
                     }
 
                     // Backup call in case timing is off. Held in a work item so
@@ -1122,7 +1139,11 @@ class OfflineProtocolModule: RCTEventEmitter {
                     // a backup that already started running, which a
                     // main-queue backup could dodge.
                     bleStatusBackupWorkItem?.cancel()
-                    let backup = DispatchWorkItem { [weak self] in
+                    let backup = DispatchWorkItem { [weak self, weak manager] in
+                        guard let manager, manager.state != .unavailable else {
+                            self?.emitDiagnostic(level: "info", message: "Backup bleStatusChanged(true) skipped: the radio is unavailable")
+                            return
+                        }
                         print("[OfflineProtocolModule] Backup bleStatusChanged(true) call")
                         self?.emitDiagnostic(level: "info", message: "Backup call to protocol.bleStatusChanged(true)")
                         try? self?.protocolInstance?.bleStatusChanged(isAvailable: true)

crates/offline-protocol-mls/src/manager.rs

@@ -1215,11 +1215,25 @@ impl MlsManager {
     /// still there, because callers clean up their own state for the peer
     /// only once this succeeds. Deleting the session first and then failing
     /// on the stamp left a peer recorded as confirmed with no session.
+    ///
+    /// A Welcome still pending for the peer goes last, and only after the
+    /// session is gone. It invites the peer into the group just deleted, so
+    /// leaving it let `get_pending_welcome` hand out an invitation to a group
+    /// this device no longer has. Clearing it is best-effort: by then the
+    /// session is deleted, and an `Err` here would claim it was not.
     pub fn delete_session(&self, other_user_id: &str) -> Result<()> {
         let session_id = self.session_manager.get_session_id(other_user_id)?;
         self.storage
             .delete(StorageKeyType::GroupMetadata.as_str(), session_id.as_str())?;
-        self.session_manager.delete_session(other_user_id)
+        self.session_manager.delete_session(other_user_id)?;
+        if let Err(error) = self.clear_pending_welcome(other_user_id) {
+            warn!(
+                other_user_id = %other_user_id,
+                error = %error,
+                "Session deleted but its pending Welcome could not be cleared"
+            );
+        }
+        Ok(())
     }
 
     // ========================================================================
@@ -4341,6 +4355,26 @@ mod tests {
         assert_eq!(alice.unacknowledged_seals(&bob), 0);
     }
 
+    /// A Welcome stored for a session must not outlive it: it invites the
+    /// peer into the group that was just deleted.
+    #[test]
+    fn deleting_a_session_drops_its_pending_welcome() {
+        let alice = create_test_manager("alice");
+        let bob = create_test_manager("bob");
+        let bob_kp = bob.generate_key_package().unwrap();
+        alice
+            .import_key_package(&addr("bob"), &bob_kp.key_package_data)
+            .unwrap();
+
+        // The FFI path creates the session on first send and parks its Welcome.
+        alice.encrypt_for_user(&addr("bob"), b"hello").unwrap();
+        assert!(alice.get_pending_welcome(&addr("bob")).unwrap().is_some());
+
+        alice.delete_session(&addr("bob")).unwrap();
+        assert!(!alice.has_session(&addr("bob")).unwrap());
+        assert!(alice.get_pending_welcome(&addr("bob")).unwrap().is_none());
+    }
+
     /// RFC 8032 §7.1 TEST 1 Ed25519 public key.
     const RFC8032_TV1_PK: [u8; 32] = [
         0xd7, 0x5a, 0x98, 0x01, 0x82, 0xb1, 0x0a, 0xb7, 0xd5, 0x4b, 0xfe, 0xd3, 0xc9, 0x64, 0x07,

crates/offline-protocol-uniffi/src/lib.rs

@@ -294,6 +294,52 @@ fn recover_rwlock_read<'a, T>(
     })
 }
 
+/// How many delivered events the polling queue keeps; the oldest goes first.
+const EVENT_QUEUE_CAPACITY: usize = 1000;
+
+/// Hands one event to the binding's callback and to the polling queue.
+///
+/// Every event the binding sees goes through here, whether the core engine
+/// raised it or the wrapper did. There used to be two copies, and the wrapper's
+/// called the callback while holding the slot's read guard and without
+/// isolating a throw: a callback that called `set_event_callback` deadlocked,
+/// and a throwing one unwound into the wrapper and skipped the queue.
+///
+/// The callback is cloned out of the slot before it is called, so it may
+/// re-enter the engine. A foreign callback that throws reaches here as a
+/// panic, because UniFFI has no error type for a void callback's exception;
+/// it is caught so the queue still gets the event, and the miss is named by
+/// type (#514).
+fn deliver_event(
+    callback_slot: &RwLock<Option<Arc<dyn EventCallback>>>,
+    queue: &Mutex<VecDeque<String>>,
+    event: &crate::CoreEvent,
+) {
+    let Ok(event_json) = event.to_json() else {
+        return;
+    };
+    let callback = recover_rwlock_read(callback_slot, "event_callback")
+        .as_ref()
+        .cloned();
+    if let Some(callback) = callback {
+        let delivered = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
+            callback.on_event(event_json.clone());
+        }));
+        if delivered.is_err() {
+            tracing::error!(
+                event = event.telemetry_name(),
+                "EventCallback.on_event threw; the binding missed this event",
+            );
+        }
+    }
+
+    let mut queue = recover_mutex(queue, "event_queue");
+    queue.push_back(event_json);
+    if queue.len() > EVENT_QUEUE_CAPACITY {
+        queue.pop_front();
+    }
+}
+
 fn recover_rwlock_write<'a, T>(
     lock: &'a RwLock<T>,
     name: &str,
@@ -2809,40 +2855,7 @@ impl OfflineProtocol {
         // Register event handler with core protocol to forward all events
         // This bridges events from the core protocol to JavaScript
         protocol.on_event(move |event| {
-            // Convert event to JSON
-            if let Ok(event_json) = event.to_json() {
-                // Clone callback Arc outside the lock to avoid holding the
-                // RwLock during callback invocation (prevents deadlock if the
-                // callback re-enters the protocol).
-                let callback_arc = recover_rwlock_read(&event_callback_clone, "event_callback")
-                    .as_ref()
-                    .cloned();
-                if let Some(callback) = callback_arc {
-                    // A foreign callback that throws reaches here as a panic:
-                    // UniFFI has no error type to lift a void callback's
-                    // exception into. Isolated here rather than left to the
-                    // engine's own isolation so the polling queue below still
-                    // gets the event, and the miss is named by type (#514).
-                    let delivered = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
-                        callback.on_event(event_json.clone());
-                    }));
-                    if delivered.is_err() {
-                        tracing::error!(
-                            event = event.telemetry_name(),
-                            "EventCallback.on_event threw; the binding missed this event",
-                        );
-                    }
-                }
-
-                // Add to event queue for polling
-                let mut queue = recover_mutex(&event_queue_clone, "event_queue");
-                queue.push_back(event_json);
-
-                // Limit queue size to prevent memory issues
-                if queue.len() > 1000 {
-                    queue.pop_front();
-                }
-            }
+            deliver_event(&event_callback_clone, &event_queue_clone, &event);
         });
 
         Ok(Self {
@@ -3261,24 +3274,7 @@ impl OfflineProtocol {
 
     /// Internal: Emit an event through the callback
     fn emit_event(&self, event: crate::CoreEvent) {
-        // Convert event to JSON
-        if let Ok(event_json) = event.to_json() {
-            // Call the callback if set
-            if let Some(callback) =
-                recover_rwlock_read(&self.event_callback, "event_callback").as_ref()
-            {
-                callback.on_event(event_json.clone());
-            }
-
-            // Also queue it for polling
-            let mut queue = recover_mutex(&self.event_queue, "event_queue");
-            queue.push_back(event_json);
-
-            // Limit queue size to prevent memory issues
-            if queue.len() > 1000 {
-                queue.pop_front();
-            }
-        }
+        deliver_event(&self.event_callback, &self.event_queue, &event);
     }
 
     /// Internal: single seam for all carrier-level reachability signals
@@ -9523,6 +9519,72 @@ mod tests {
         assert_eq!(protocol.ble_get_peer_count(), 1);
     }
 
+    /// An event the wrapper raises itself (here `NeighborDiscovered`) reaches
+    /// the callback with the slot released, so the callback may replace
+    /// itself. The wrapper's own path used to hold the read guard across the
+    /// call, and `set_event_callback` from inside it deadlocked.
+    #[test]
+    fn a_callback_may_replace_itself_from_a_wrapper_raised_event() {
+        use std::sync::mpsc;
+        use std::sync::Weak;
+
+        struct Noop;
+        impl EventCallback for Noop {
+            fn on_event(&self, _event_json: String) {}
+        }
+        struct ReplacesItself(Weak<OfflineProtocol>);
+        impl EventCallback for ReplacesItself {
+            fn on_event(&self, _event_json: String) {
+                if let Some(protocol) = self.0.upgrade() {
+                    protocol.set_event_callback(Box::new(Noop));
+                }
+            }
+        }
+
+        let protocol = Arc::new(OfflineProtocol::new(create_test_config()).unwrap());
+        protocol.set_event_callback(Box::new(ReplacesItself(Arc::downgrade(&protocol))));
+
+        // On a thread with a deadline, so a regression fails instead of hanging.
+        let (done, finished) = mpsc::channel();
+        let raiser = protocol.clone();
+        std::thread::spawn(move || {
+            let result = raiser.ble_peer_discovered("peer1".to_string(), -50);
+            let _ = done.send(result.is_ok());
+        });
+        assert_eq!(
+            finished.recv_timeout(std::time::Duration::from_secs(10)),
+            Ok(true),
+            "set_event_callback from inside the callback deadlocked"
+        );
+    }
+
+    /// A callback that throws on a wrapper-raised event does not unwind into
+    /// the wrapper, and the polling queue still gets the event.
+    #[test]
+    fn a_throwing_callback_does_not_cost_the_polling_queue_a_wrapper_raised_event() {
+        struct Throws;
+        impl EventCallback for Throws {
+            fn on_event(&self, _event_json: String) {
+                panic!("the binding's listener threw");
+            }
+        }
+
+        let protocol = OfflineProtocol::new(create_test_config()).unwrap();
+        protocol.set_event_callback(Box::new(Throws));
+        protocol
+            .ble_peer_discovered("peer1".to_string(), -50)
+            .unwrap();
+
+        let events = drained_events(&protocol);
+        assert!(
+            events
+                .iter()
+                .any(|event| event.contains("neighbor_discovered")
+                    || event.contains("NeighborDiscovered")),
+            "the queue lost the event: {events:?}"
+        );
+    }
+
     #[test]
     fn test_file_transfer_tracking() {
         let config = create_test_config();

crates/offline-protocol-web/src/ble_host.rs

@@ -128,6 +128,11 @@ impl Node {
     /// and has said hello: tells the transport its write payload, then the
     /// engine that the peer is here.
     pub fn ble_link_up(&mut self, peer: &str, max_payload: u32) -> NodeResult<()> {
+        // ORDERING: the slot before either set changes. Before `initialize`
+        // there is no BLE transport, and refusing after the insert below left
+        // the address in `linked`: the next link-up read as a relink and was
+        // never announced, and fragments from the address were accepted.
+        let transport = self.ble_slot()?;
         let host = self.ble_host_mut()?;
         let Some(at) = host.verified.iter().position(|held| held == peer) else {
             return Err(NodeError(format!(
@@ -141,7 +146,6 @@ impl Node {
         }
         host.verified.remove(at);
         let relinked = !host.linked.insert(peer.to_string());
-        let transport = self.ble_slot()?;
         // ORDERING: the payload before the peer, as every binding does.
         transport.set_peer_mtu(peer, max_payload as usize);
         if relinked {