Make ordinary Noise reconnects atomic and race-safe (#1463)

Replaces destroy-then-rebuild Noise re-handshakes with an atomic reconnect protocol: prepared XX message-1 handoff tokens (claim-once, invalidated by crossed inbound initiations), receive-only quarantine of the established transport while an inbound replacement proves identity (promote on success, rollback+cooldown on failure/timeout), deterministic lower-peerID-wins crossed-initiator resolution, and a per-link-epoch BLE revalidation policy that re-proves cached sessions inside the same bleQueue critical section as the link rebind. On main, a fresh msg1 simply destroys an established session and rekey spans two non-atomic barriers.

Includes the review fix: timeout-restores defer outbound queue draining until the convergence retry completes (restore reason plumbed end-to-end), so DMs are never drained under keys a restarted peer already discarded — with a deterministic interleaving test. Identity-mismatch restores drain immediately. Full local suite 1768 tests green.

bitchatTests/BLEServiceCoreTests.swift

@@ -502,14 +502,26 @@ struct BLEServiceCoreTests {
         )
         let replay = try #require(victim.signPacket(unsigned), "Failed to sign replayed announce")
         #expect(ble._test_recordIngressIfNew(packet: replay, linkID: attackerLink))
+        let rebindGate = VerifiedDirectRebindGate()
+        ble._test_afterVerifiedDirectRebindEnqueued = rebindGate.pause
+        defer {
+            rebindGate.release()
+            ble._test_afterVerifiedDirectRebindEnqueued = nil
+        }
         ble._test_handlePacket(replay, fromPeerID: victimPeerID, preseedPeer: false)
 
-        let rebound = await TestHelpers.waitUntil(
-            { ble._test_centralBinding(attackerLink) == victimPeerID },
+        let announcePaused = await TestHelpers.waitUntil(
+            { rebindGate.hasPaused },
             timeout: TestConstants.longTimeout
         )
-        #expect(rebound)
-        #expect(ble.canDeliverSecurely(to: victimPeerID))
+        try #require(announcePaused)
+
+        // Rebind and ordinary reconnect preparation are one bleQueue
+        // critical section. Once the binding is visible, stale sending keys
+        // must already be unavailable.
+        #expect(ble._test_centralBinding(attackerLink) == victimPeerID)
+        #expect(!ble.canDeliverSecurely(to: victimPeerID))
+        rebindGate.release()
 
         let outbound = OutboundPacketTap()
         ble._test_onOutboundPacket = { outbound.record($0) }
@@ -537,20 +549,26 @@ struct BLEServiceCoreTests {
 
         // Preserve a working victim session while an unauthenticated
         // replacement candidate arrives on a newly bound physical link.
-        let message1 = try ble._test_noiseInitiateHandshake(with: victimPeerID)
+        // Establish BLE as responder so the replacement candidate below is
+        // not coalesced by the initiator-completion grace path.
+        let message1 = try victim.initiateHandshake(with: ble.myPeerID)
         let message2 = try #require(
-            try victim.processHandshakeMessage(from: ble.myPeerID, message: message1)
-        )
-        let message3 = try #require(
             try ble._test_noiseProcessHandshakeMessage(
                 from: victimPeerID,
+                message: message1
+            )
+        )
+        let message3 = try #require(
+            try victim.processHandshakeMessage(
+                from: ble.myPeerID,
                 message: message2
             )
         )
-        _ = try victim.processHandshakeMessage(
-            from: ble.myPeerID,
+        _ = try ble._test_noiseProcessHandshakeMessage(
+            from: victimPeerID,
             message: message3
         )
+        await ble._test_drainNoiseMessagePipeline()
         #expect(ble.canDeliverSecurely(to: victimPeerID))
 
         let centralUUID = "central-replacement-xx-message-one"
@@ -614,7 +632,330 @@ struct BLEServiceCoreTests {
                 for: victimPeerID
             )
         )
-        #expect(ble.canDeliverSecurely(to: victimPeerID))
+        // Ordinary reconnect hardening quarantines the cached transport while
+        // this candidate proves the claimed identity. It must be unavailable
+        // for sending as well as unable to authenticate this ingress link.
+        #expect(!ble.canDeliverSecurely(to: victimPeerID))
+    }
+
+    @Test
+    func failedInboundReconnectRestoresAndDrainsTypedPayloadQueue() async throws {
+        let ble = makeService()
+        let alice = NoiseEncryptionService(keychain: MockKeychain())
+        let mallory = NoiseEncryptionService(keychain: MockKeychain())
+        let alicePeerID = PeerID(publicKey: alice.getStaticPublicKeyData())
+
+        // Establish BLE as responder so the following inbound reconnect is
+        // not intentionally coalesced by the initiator-completion grace path.
+        let message1 = try alice.initiateHandshake(with: ble.myPeerID)
+        let message2 = try #require(
+            try ble._test_noiseProcessHandshakeMessage(
+                from: alicePeerID,
+                message: message1
+            )
+        )
+        let message3 = try #require(
+            try alice.processHandshakeMessage(
+                from: ble.myPeerID,
+                message: message2
+            )
+        )
+        _ = try ble._test_noiseProcessHandshakeMessage(
+            from: alicePeerID,
+            message: message3
+        )
+        await ble._test_drainNoiseMessagePipeline()
+        #expect(ble.canDeliverSecurely(to: alicePeerID))
+
+        let outbound = OutboundPacketTap()
+        ble._test_onOutboundPacket = outbound.record
+        let forgedMessage1 = try mallory.initiateHandshake(with: ble.myPeerID)
+        let firstPacket = BitchatPacket(
+            type: MessageType.noiseHandshake.rawValue,
+            senderID: Data(hexString: alicePeerID.id) ?? Data(),
+            recipientID: Data(hexString: ble.myPeerID.id),
+            timestamp: UInt64(Date().timeIntervalSince1970 * 1_000),
+            payload: forgedMessage1,
+            signature: nil,
+            ttl: 7
+        )
+        ble._test_handlePacket(firstPacket, fromPeerID: alicePeerID)
+
+        let responseReady = await TestHelpers.waitUntil(
+            {
+                outbound.snapshot().contains {
+                    $0.type == MessageType.noiseHandshake.rawValue
+                        && PeerID(hexData: $0.senderID) == ble.myPeerID
+                        && $0.payload.count
+                            != NoiseSecurityConstants.xxInitialMessageSize
+                }
+            },
+            timeout: TestConstants.longTimeout
+        )
+        try #require(responseReady)
+        let forgedMessage2 = try #require(
+            outbound.snapshot().first {
+                $0.type == MessageType.noiseHandshake.rawValue
+                    && PeerID(hexData: $0.senderID) == ble.myPeerID
+                    && $0.payload.count
+                        != NoiseSecurityConstants.xxInitialMessageSize
+            }?.payload
+        )
+        #expect(!ble.canDeliverSecurely(to: alicePeerID))
+
+        // Typed control traffic must queue behind the ordinary responder,
+        // rather than attempting encryption and disappearing.
+        let privateMessageID = "quarantine-pm-\(UUID().uuidString)"
+        ble.sendPrivateMessage(
+            "queued private message",
+            to: alicePeerID,
+            recipientNickname: "Alice",
+            messageID: privateMessageID
+        )
+        ble.sendGroupInvite(Data("queued-during-quarantine".utf8), to: alicePeerID)
+        await ble._test_drainNoiseMessagePipeline()
+        #expect(outbound.count(ofType: .noiseEncrypted) == 0)
+
+        let forgedMessage3 = try #require(
+            try mallory.processHandshakeMessage(
+                from: ble.myPeerID,
+                message: forgedMessage2
+            )
+        )
+        let thirdPacket = BitchatPacket(
+            type: MessageType.noiseHandshake.rawValue,
+            senderID: Data(hexString: alicePeerID.id) ?? Data(),
+            recipientID: Data(hexString: ble.myPeerID.id),
+            timestamp: UInt64(Date().timeIntervalSince1970 * 1_000) + 1,
+            payload: forgedMessage3,
+            signature: nil,
+            ttl: 7
+        )
+        ble._test_handlePacket(thirdPacket, fromPeerID: alicePeerID)
+
+        // Restore re-enters the generation-bound authentication transition:
+        // authenticated state and both outbound queues drain exactly once.
+        let drained = await TestHelpers.waitUntil(
+            { outbound.count(ofType: .noiseEncrypted) >= 3 },
+            timeout: TestConstants.longTimeout
+        )
+        try #require(drained)
+        await ble._test_drainNoiseMessagePipeline()
+        let plaintexts = try outbound.snapshot()
+            .filter { $0.type == MessageType.noiseEncrypted.rawValue }
+            .map { try alice.decrypt($0.payload, from: ble.myPeerID) }
+        #expect(plaintexts.count == 3)
+        #expect(
+            plaintexts.filter {
+                $0.first == NoisePayloadType.authenticatedPeerState.rawValue
+            }.count == 1
+        )
+        #expect(
+            plaintexts.filter {
+                $0.first == NoisePayloadType.privateMessage.rawValue
+            }.count == 1
+        )
+        #expect(
+            plaintexts.filter {
+                $0.first == NoisePayloadType.groupInvite.rawValue
+            }.count == 1
+        )
+    }
+
+    /// The message-loss interleaving behind a legitimate peer restart: the
+    /// remote completes a replacement handshake (discarding the old keys)
+    /// but its completion never arrives, so the local responder timeout
+    /// restores the quarantined OLD generation and requests one convergence
+    /// retry — both dispatched unordered. When the restore handler wins the
+    /// race, it must NOT drain the pending private-message/typed-payload
+    /// queues under the restored keys the remote no longer holds; the drain
+    /// has to wait for the convergence handshake and use its new session.
+    @Test
+    func timeoutRestoredSessionDefersQueueDrainUntilConvergence() async throws {
+        let ble = makeService(noiseResponderHandshakeTimeout: 0.3)
+        let alice = NoiseEncryptionService(keychain: MockKeychain())
+        let mallory = NoiseEncryptionService(keychain: MockKeychain())
+        let alicePeerID = PeerID(publicKey: alice.getStaticPublicKeyData())
+
+        // Establish BLE as responder so the inbound reconnect below is not
+        // coalesced by the initiator-completion grace path.
+        let message1 = try alice.initiateHandshake(with: ble.myPeerID)
+        let message2 = try #require(
+            try ble._test_noiseProcessHandshakeMessage(
+                from: alicePeerID,
+                message: message1
+            )
+        )
+        let message3 = try #require(
+            try alice.processHandshakeMessage(
+                from: ble.myPeerID,
+                message: message2
+            )
+        )
+        _ = try ble._test_noiseProcessHandshakeMessage(
+            from: alicePeerID,
+            message: message3
+        )
+        await ble._test_drainNoiseMessagePipeline()
+        #expect(ble.canDeliverSecurely(to: alicePeerID))
+
+        // The convergence retry only prepares for reachable peers.
+        ble._test_seedConnectedPeer(alicePeerID, nickname: "Alice")
+
+        let reconciled = SessionReconcileCounter()
+        ble._test_onPrivateMediaSessionReconciled = reconciled.record
+        // Park the convergence-recovery callback on its global-queue thread
+        // before it can enqueue onto messageQueue: the restore handler
+        // deterministically wins the dispatch race this test exercises.
+        let recoveryGate = HandshakeRecoveryEnqueueGate()
+        defer { recoveryGate.release() }
+        ble._test_beforeHandshakeRecoveryEnqueued = { _ in recoveryGate.pause() }
+        let outbound = OutboundPacketTap()
+        ble._test_onOutboundPacket = outbound.record
+
+        // Park the traffic the race would lose directly in the pending
+        // queues — the same place live sends land during quarantine — so no
+        // assertion below depends on outrunning the responder deadline under
+        // parallel test load. Nothing drains these queues while the current
+        // session stays untouched.
+        ble._test_enqueuePendingPrivateMessage(
+            content: "deferred private message",
+            messageID: "deferred-pm-\(UUID().uuidString)",
+            for: alicePeerID
+        )
+        ble._test_enqueuePendingNoisePayload(
+            NoisePayload(
+                type: .groupInvite,
+                data: Data("queued-during-quarantine".utf8)
+            ).encode(),
+            transferId: "deferred-invite-\(UUID().uuidString)",
+            for: alicePeerID
+        )
+
+        // An unauthenticated message 1 quarantines the established transport.
+        // Its message 3 never arrives, modeling the restarted peer whose
+        // completion was lost after it already discarded the old keys.
+        let reconnectMessage1 = try mallory.initiateHandshake(with: ble.myPeerID)
+        let reconnectPacket = BitchatPacket(
+            type: MessageType.noiseHandshake.rawValue,
+            senderID: Data(hexString: alicePeerID.id) ?? Data(),
+            recipientID: Data(hexString: ble.myPeerID.id),
+            timestamp: UInt64(Date().timeIntervalSince1970 * 1_000),
+            payload: reconnectMessage1,
+            signature: nil,
+            ttl: 7
+        )
+        ble._test_handlePacket(reconnectPacket, fromPeerID: alicePeerID)
+        // The responder's message 2 is a monotonic quarantine signal; the
+        // secure-delivery dip itself only lasts until the responder deadline,
+        // which parallel test load can outrun. (The recovery gate keeps the
+        // convergence retry's message 1 out of the tap until released.)
+        let responderReady = await TestHelpers.waitUntil(
+            { outbound.count(ofType: .noiseHandshake) >= 1 },
+            timeout: TestConstants.longTimeout
+        )
+        try #require(responderReady)
+        #expect(outbound.count(ofType: .noiseEncrypted) == 0)
+
+        // The responder timeout restores the quarantined generation; the
+        // gate guarantees its handler runs before the convergence retry.
+        let restoreRan = await TestHelpers.waitUntil(
+            { reconciled.count(for: alicePeerID) == 1 },
+            timeout: TestConstants.longTimeout
+        )
+        try #require(restoreRan)
+        #expect(ble.canDeliverSecurely(to: alicePeerID))
+        await ble._test_drainNoiseMessagePipeline()
+        // The parked queues must not have been encrypted under the restored
+        // old generation the remote may no longer be able to read.
+        #expect(outbound.count(ofType: .noiseEncrypted) == 0)
+
+        // Release the mandatory convergence retry: it retires the restored
+        // session and starts a fresh XX exchange with the live peer.
+        recoveryGate.release()
+        let retryStarted = await TestHelpers.waitUntil(
+            {
+                outbound.snapshot().contains {
+                    $0.type == MessageType.noiseHandshake.rawValue
+                        && PeerID(hexData: $0.senderID) == ble.myPeerID
+                        && $0.payload.count
+                            == NoiseSecurityConstants.xxInitialMessageSize
+                }
+            },
+            timeout: TestConstants.longTimeout
+        )
+        try #require(retryStarted)
+        #expect(outbound.count(ofType: .noiseEncrypted) == 0)
+        let retryMessage1 = try #require(
+            outbound.snapshot().last {
+                $0.type == MessageType.noiseHandshake.rawValue
+                    && PeerID(hexData: $0.senderID) == ble.myPeerID
+                    && $0.payload.count
+                        == NoiseSecurityConstants.xxInitialMessageSize
+            }?.payload
+        )
+
+        // Alice answers the retry as the restarted peer she models: her old
+        // session is gone, so the retry is a fresh responder exchange (and
+        // never the initiator-completion grace deferral, whose lower-peerID
+        // arm would otherwise coalesce the retry for random key orderings).
+        alice.clearSession(for: ble.myPeerID)
+        let retryMessage2 = try #require(
+            try alice.processHandshakeMessage(
+                from: ble.myPeerID,
+                message: retryMessage1
+            )
+        )
+        let retryPacket = BitchatPacket(
+            type: MessageType.noiseHandshake.rawValue,
+            senderID: Data(hexString: alicePeerID.id) ?? Data(),
+            recipientID: Data(hexString: ble.myPeerID.id),
+            timestamp: UInt64(Date().timeIntervalSince1970 * 1_000) + 1,
+            payload: retryMessage2,
+            signature: nil,
+            ttl: 7
+        )
+        ble._test_handlePacket(retryPacket, fromPeerID: alicePeerID)
+        let drained = await TestHelpers.waitUntil(
+            { outbound.count(ofType: .noiseEncrypted) >= 3 },
+            timeout: TestConstants.longTimeout
+        )
+        try #require(drained)
+        await ble._test_drainNoiseMessagePipeline()
+
+        // Alice completes with message 3, then must be able to decrypt every
+        // drained payload — proving nothing left under the old generation.
+        let retryMessage3 = try #require(
+            outbound.snapshot().last {
+                $0.type == MessageType.noiseHandshake.rawValue
+                    && PeerID(hexData: $0.senderID) == ble.myPeerID
+                    && $0.payload.count
+                        != NoiseSecurityConstants.xxInitialMessageSize
+            }?.payload
+        )
+        _ = try alice.processHandshakeMessage(
+            from: ble.myPeerID,
+            message: retryMessage3
+        )
+        let plaintexts = try outbound.snapshot()
+            .filter { $0.type == MessageType.noiseEncrypted.rawValue }
+            .map { try alice.decrypt($0.payload, from: ble.myPeerID) }
+        #expect(plaintexts.count == 3)
+        #expect(
+            plaintexts.filter {
+                $0.first == NoisePayloadType.authenticatedPeerState.rawValue
+            }.count == 1
+        )
+        #expect(
+            plaintexts.filter {
+                $0.first == NoisePayloadType.privateMessage.rawValue
+            }.count == 1
+        )
+        #expect(
+            plaintexts.filter {
+                $0.first == NoisePayloadType.groupInvite.rawValue
+            }.count == 1
+        )
     }
 
     /// A legitimate rotation announce necessarily arrives on a link still
@@ -943,6 +1284,79 @@ private final class OutboundPacketTap {
         lock.lock(); defer { lock.unlock() }
         return packets.filter { $0.type == type.rawValue }.count
     }
+
+    func snapshot() -> [BitchatPacket] {
+        lock.lock(); defer { lock.unlock() }
+        return packets
+    }
+}
+
+/// Blocks the convergence-recovery callback on its global-queue thread so a
+/// test can prove the quarantine-restore handler wins the messageQueue race.
+private final class HandshakeRecoveryEnqueueGate: @unchecked Sendable {
+    private let condition = NSCondition()
+    private var released = false
+
+    func pause() {
+        condition.lock()
+        while !released {
+            condition.wait()
+        }
+        condition.unlock()
+    }
+
+    func release() {
+        condition.lock()
+        released = true
+        condition.broadcast()
+        condition.unlock()
+    }
+}
+
+/// Thread-safe counter for `_test_onPrivateMediaSessionReconciled` firings.
+private final class SessionReconcileCounter: @unchecked Sendable {
+    private let lock = NSLock()
+    private var reconciles: [PeerID] = []
+
+    func record(_ peerID: PeerID) {
+        lock.lock()
+        reconciles.append(peerID)
+        lock.unlock()
+    }
+
+    func count(for peerID: PeerID) -> Int {
+        lock.lock(); defer { lock.unlock() }
+        return reconciles.filter { $0 == peerID }.count
+    }
+}
+
+private final class VerifiedDirectRebindGate: @unchecked Sendable {
+    private let condition = NSCondition()
+    private var paused = false
+    private var released = false
+
+    var hasPaused: Bool {
+        condition.lock()
+        defer { condition.unlock() }
+        return paused
+    }
+
+    func pause() {
+        condition.lock()
+        paused = true
+        condition.broadcast()
+        while !released {
+            condition.wait()
+        }
+        condition.unlock()
+    }
+
+    func release() {
+        condition.lock()
+        released = true
+        condition.broadcast()
+        condition.unlock()
+    }
 }
 
 private final class ReceivePacketHandoffGate: @unchecked Sendable {
@@ -1007,7 +1421,10 @@ private final class PanicIngressObserver: @unchecked Sendable {
     }
 }
 
-private func makeService() -> BLEService {
+private func makeService(
+    noiseResponderHandshakeTimeout: TimeInterval =
+        NoiseSecurityConstants.ordinaryResponderHandshakeTimeout
+) -> BLEService {
     let keychain = MockKeychain()
     let identityManager = MockIdentityManager(keychain)
     let idBridge = NostrIdentityBridge(keychain: MockKeychainHelper())
@@ -1015,7 +1432,8 @@ private func makeService() -> BLEService {
         keychain: keychain,
         idBridge: idBridge,
         identityManager: identityManager,
-        initializeBluetoothManagers: false
+        initializeBluetoothManagers: false,
+        noiseResponderHandshakeTimeout: noiseResponderHandshakeTimeout
     )
 }