permissionlesstech/bitchat
1
0
Fork 0
mirror of https://github.com/permissionlesstech/bitchat.git synced 2026-07-25 23:05:20 +00:00
Code Issues Packages Projects Releases Wiki Activity

Invalidate queued BLE ingress during panic

Browse Source
This commit is contained in:
jack
2026-07-25 23:49:41 +02:00
parent d1394392ed
commit 6cd7a43b4b
2 changed files with 286 additions and 22 deletions
Download Patch File Download Diff File Expand all files Collapse all files
bitchat/Services/BLE/BLEService.swift
+132 -22
View File
@@ -104,6 +104,10 @@ final class BLEService: NSObject {
// Test-only tap on the outbound pipeline so multi-node tests can ferry
// packets between in-process service instances.
var _test_onOutboundPacket: ((BitchatPacket) -> Void)?
/// May block a synthetic CoreBluetooth receive callback immediately
/// before it hands a packet to `messageQueue`.
var _test_beforeReceivePacketHandoff: (() -> Void)?
var _test_onReceivePacketHandoff: (() -> Void)?
#endif
private var selfBroadcastTracker = BLESelfBroadcastTracker()
private let meshTopology = MeshTopologyTracker()
@@ -119,6 +123,7 @@ final class BLEService: NSObject {
private struct PendingMeshPing {
let peerID: PeerID
let sentAt: Date
let lifecycleGeneration: UInt64
let completion: @MainActor (MeshPingResult?) -> Void
let timeout: DispatchWorkItem
}
@@ -483,11 +488,15 @@ final class BLEService: NSObject {
setPanicSuspended(true)
gossipSyncManager?.stop()
gossipSyncManager = nil
// Wait out sends that passed admission before the gate closed. Work
// queued behind this barrier observes `isPanicSuspended` and exits,
// so the radio stop below is a true synchronous boundary.
messageQueue.sync(flags: .barrier) {}
// Stop the radio and drain CoreBluetooth's delegate queue first. A
// callback may already have passed its initial suspension check; the
// bleQueue drain forces its final messageQueue handoff to happen
// before the receive barrier below.
stopServicesImmediatelyForPanic()
// Drain every receive/send submitted by callbacks that finished ahead
// of the radio stop. Later callbacks observe the closed lifecycle, and
// generation-bound handoffs that raced this barrier reject themselves.
messageQueue.sync(flags: .barrier) {}
clearEmergencySessionState()
}
@@ -795,21 +804,30 @@ final class BLEService: NSObject {
private func clearEmergencySessionState() {
// Clear all sessions and peers
let cancelledTransfers: [(id: String, items: [DispatchWorkItem])] = collectionsQueue.sync(flags: .barrier) {
let entries = outboundFragmentTransfers.removeAll().map { ($0.id, $0.workItems) }
let cancelled = collectionsQueue.sync(flags: .barrier) {
let entries = outboundFragmentTransfers.removeAll().map {
(id: $0.id, items: $0.workItems)
}
let pingTimeouts = pendingMeshPings.values.map(\.timeout)
pendingMeshPings.removeAll()
meshPingResponseLimiter = SyncResponseRateLimiter(
maxResponses: TransportConfig.meshPingInboundMaxPerLink,
window: TransportConfig.meshPingInboundWindowSeconds
)
peerRegistry.removeAll()
fragmentAssemblyBuffer.removeAll()
sourceRouteFailures = BLESourceRouteFailureCache()
// Also clear pending message queues to avoid stale state across sessions
pendingNoiseSessionQueues.removeAll()
pendingDirectedRelays.removeAll()
return entries
return (transfers: entries, pingTimeouts: pingTimeouts)
}
for entry in cancelledTransfers {
for entry in cancelled.transfers {
entry.items.forEach { $0.cancel() }
TransferProgressManager.shared.cancel(id: entry.id)
}
cancelled.pingTimeouts.forEach { $0.cancel() }
// Clear processed messages
messageDeduplicator.reset()
@@ -1602,22 +1620,40 @@ final class BLEService: NSObject {
}
func collectArchivedPublicMessages(completion: @escaping @MainActor ([ArchivedPublicMessage]) -> Void) {
guard let generation = capturePanicLifecycleGeneration() else {
return
}
guard let sync = gossipSyncManager else {
Task { @MainActor in completion([]) }
notifyUI { [weak self] in
guard let self,
self.isCurrentPanicLifecycleGeneration(generation) else {
return
}
completion([])
}
return
}
sync.collectPublicMessagePackets { [weak self] packets in
guard let self = self else {
Task { @MainActor in completion([]) }
guard let self,
self.isCurrentPanicLifecycleGeneration(generation) else {
return
}
// Signature verification and registry lookups run on messageQueue
// like the live receive path.
self.messageQueue.async {
guard self.isCurrentPanicLifecycleGeneration(generation) else {
return
}
let decoded = packets
.compactMap { self.decodeArchivedPublicMessage($0) }
.sorted { $0.timestamp < $1.timestamp }
Task { @MainActor in completion(decoded) }
self.notifyUI { [weak self] in
guard let self,
self.isCurrentPanicLifecycleGeneration(generation) else {
return
}
completion(decoded)
}
}
}
}
@@ -2414,6 +2450,28 @@ private extension BLEService {
#if DEBUG
// Test-only helper to inject packets into the receive pipeline
extension BLEService {
/// Queues an event through the same MainActor hop as production receive
/// handlers so panic-boundary tests can deterministically invalidate it.
func _test_emitTransportEvent(_ event: TransportEvent) {
emitTransportEvent(event)
}
var _test_isPanicIngressOpen: Bool {
capturePanicLifecycleGeneration() != nil
}
/// Models a CoreBluetooth delegate callback without requiring a physical
/// peripheral. The callback itself runs on `bleQueue`, exactly where the
/// panic radio-stop barrier must linearize it.
func _test_handlePacketFromBLEQueue(
_ packet: BitchatPacket,
fromPeerID: PeerID
) {
bleQueue.async { [weak self] in
self?.handleReceivedPacket(packet, from: fromPeerID)
}
}
func _test_handlePacket(_ packet: BitchatPacket, fromPeerID: PeerID, preseedPeer: Bool = true, signingPublicKey: Data? = nil) {
if preseedPeer {
// Ensure the synthetic peer is known and marked verified for public-message tests
@@ -3149,8 +3207,18 @@ extension BLEService {
/// Notify UI on the MainActor to satisfy Swift concurrency isolation
private func notifyUI(_ block: @escaping @MainActor () -> Void) {
// Always hop onto the MainActor so calls to @MainActor delegates are safe
Task { @MainActor in
// Capture the panic lifecycle before queueing the MainActor hop. A
// receive callback can enqueue UI delivery immediately before panic
// clears application state; rechecking here prevents that stale work
// from repopulating the wiped conversation store afterward.
guard let generation = capturePanicLifecycleGeneration() else {
return
}
Task { @MainActor [weak self] in
guard let self,
self.isCurrentPanicLifecycleGeneration(generation) else {
return
}
block()
}
}
@@ -3315,14 +3383,24 @@ extension BLEService {
/// The completion fires exactly once on the main actor: with RTT/hops
/// when the matching pong returns, or nil after the timeout window.
func sendMeshPing(to peerID: PeerID, completion: @escaping @MainActor (MeshPingResult?) -> Void) {
guard let generation = capturePanicLifecycleGeneration() else {
return
}
messageQueue.async { [weak self] in
guard let self,
self.isCurrentPanicLifecycleGeneration(generation),
let recipientData = peerID.toShort().routingData,
let payload = MeshPingPayload(
nonce: Data((0..<MeshPingPayload.nonceLength).map { _ in UInt8.random(in: .min ... .max) }),
originTTL: self.messageTTL
) else {
Task { @MainActor in completion(nil) }
self?.notifyUI { [weak self] in
guard let self,
self.isCurrentPanicLifecycleGeneration(generation) else {
return
}
completion(nil)
}
return
}
let nonce = payload.nonce
@@ -3341,12 +3419,21 @@ extension BLEService {
self.pendingMeshPings.removeValue(forKey: nonce)
}
guard let expired else { return }
Task { @MainActor in expired.completion(nil) }
self.notifyUI { [weak self] in
guard let self,
self.isCurrentPanicLifecycleGeneration(
expired.lifecycleGeneration
) else {
return
}
expired.completion(nil)
}
}
self.collectionsQueue.sync(flags: .barrier) {
self.pendingMeshPings[nonce] = PendingMeshPing(
peerID: PeerID(hexData: recipientData),
sentAt: Date(),
lifecycleGeneration: generation,
completion: completion,
timeout: timeout
)
@@ -3413,7 +3500,15 @@ extension BLEService {
rttMs: max(0, rttMs),
hops: MeshPingPayload.hopCount(originTTL: pong.originTTL, receivedTTL: packet.ttl)
)
Task { @MainActor in pending.completion(result) }
notifyUI { [weak self] in
guard let self,
self.isCurrentPanicLifecycleGeneration(
pending.lifecycleGeneration
) else {
return
}
pending.completion(result)
}
}
/// Estimated intermediate hops toward `peerID`, BFS over gossiped
@@ -3897,7 +3992,7 @@ extension BLEService {
let store = courierStore
let policy = courierDepositPolicy
let metrics = sfMetrics
Task { @MainActor in
notifyUI {
guard let tier = policy(depositorKey, isVerifiedPeer) else {
SecureLogger.debug("📦 Courier deposit from \(peerID.id.prefix(8))… rejected (neither favorite nor verified)", category: .session)
return
@@ -3977,7 +4072,7 @@ extension BLEService {
}
}
let policy = courierDepositPolicy
Task { @MainActor in
notifyUI {
// Same trust gate as deposits: don't hand mail to a peer who
// would reject it from us.
guard policy(noiseKey, isVerifiedPeer) != nil else { return }
@@ -4869,8 +4964,24 @@ extension BLEService {
private func handleReceivedPacket(_ packet: BitchatPacket, from peerID: PeerID) {
// Call directly if already on messageQueue, otherwise dispatch
if DispatchQueue.getSpecific(key: messageQueueKey) == nil {
guard let lifecycleGeneration =
capturePanicLifecycleGeneration() else {
return
}
#if DEBUG
_test_beforeReceivePacketHandoff?()
#endif
messageQueue.async { [weak self] in
self?.handleReceivedPacket(packet, from: peerID)
guard let self,
self.isCurrentPanicLifecycleGeneration(
lifecycleGeneration
) else {
return
}
#if DEBUG
self._test_onReceivePacketHandoff?()
#endif
self.handleReceivedPacket(packet, from: peerID)
}
return
}
@@ -5714,8 +5825,7 @@ extension BLEService {
let transportPeers: [TransportPeerSnapshot] = collectionsQueue.sync {
peerRegistry.transportSnapshots(selfNickname: myNickname)
}
// Notify UI on MainActor via delegate
Task { @MainActor [weak self] in
notifyUI { [weak self] in
self?.peerEventsDelegate?.didUpdatePeerSnapshots(transportPeers)
}
}
bitchatTests/BLEServiceCoreTests.swift
+154
View File
@@ -592,6 +592,88 @@ struct BLEServiceCoreTests {
#expect(outbound.count(ofType: .message) == 1)
}
@Test @MainActor
func panicSuspension_invalidatesQueuedMainActorIngress() async {
let ble = makeService()
let delegate = TransportEventCaptureDelegate()
ble.eventDelegate = delegate
let message = BitchatMessage(
id: "pre-panic-ingress",
sender: "Peer",
content: "must not survive panic",
timestamp: Date(),
isRelay: false,
isPrivate: true,
recipientNickname: "Me",
senderPeerID: PeerID(str: "1122334455667788")
)
// The test already owns MainActor, so this task cannot run until the
// synchronous panic boundary below has invalidated its generation.
ble._test_emitTransportEvent(.messageReceived(message))
ble.suspendForPanicReset()
await Task.yield()
#expect(delegate.messageIDs.isEmpty)
ble.completePanicReset(restartServices: false)
ble._test_emitTransportEvent(.messageReceived(message))
await Task.yield()
#expect(delegate.messageIDs == [message.id])
}
@Test @MainActor
func panicSuspension_rejectsPausedBLEReceiveBeforeMessageQueueHandoff() async {
let ble = makeService()
let gate = ReceivePacketHandoffGate()
ble._test_beforeReceivePacketHandoff = gate.pause
ble._test_onReceivePacketHandoff = gate.recordHandoff
defer {
gate.release()
ble._test_beforeReceivePacketHandoff = nil
ble._test_onReceivePacketHandoff = nil
}
let sender = PeerID(str: "1122334455667788")
let packet = makePublicPacket(
content: "must not cross panic",
sender: sender,
timestamp: UInt64(Date().timeIntervalSince1970 * 1000)
)
ble._test_handlePacketFromBLEQueue(packet, fromPeerID: sender)
#expect(await TestHelpers.waitUntil(
{ gate.hasPaused },
timeout: TestConstants.longTimeout
))
// Panic closes the lifecycle before waiting for the paused bleQueue
// callback. Releasing it afterward lets the callback enqueue its
// messageQueue handoff, where the captured generation must be rejected
// before packet processing starts.
let panicIngressObserver = PanicIngressObserver(service: ble)
let didObservePanicClosure = await withCheckedContinuation { continuation in
DispatchQueue.global(qos: .userInitiated).async {
let didObserveClosure = panicIngressObserver.waitUntilClosed(
timeout: TestConstants.defaultTimeout
)
gate.release()
continuation.resume(returning: didObserveClosure)
}
ble.suspendForPanicReset()
}
#expect(didObservePanicClosure)
#expect(gate.handoffCount == 0)
// A packet captured under the reopened lifecycle still crosses the
// same handoff, proving the test did not merely disable the hook.
ble.completePanicReset(restartServices: false)
ble._test_handlePacketFromBLEQueue(packet, fromPeerID: sender)
#expect(await TestHelpers.waitUntil(
{ gate.handoffCount == 1 },
timeout: TestConstants.longTimeout
))
}
@Test
func modifiedServices_rediscoverWhenBitChatServiceIsInvalidated() async throws {
let otherService = CBUUID(string: "0000180F-0000-1000-8000-00805F9B34FB")
@@ -686,6 +768,68 @@ private final class OutboundPacketTap {
}
}
private final class ReceivePacketHandoffGate: @unchecked Sendable {
private let condition = NSCondition()
private var paused = false
private var released = false
private var recordedHandoffCount = 0
var hasPaused: Bool {
condition.lock()
defer { condition.unlock() }
return paused
}
var handoffCount: Int {
condition.lock()
defer { condition.unlock() }
return recordedHandoffCount
}
func pause() {
condition.lock()
paused = true
condition.broadcast()
while !released {
condition.wait()
}
condition.unlock()
}
func release() {
condition.lock()
released = true
condition.broadcast()
condition.unlock()
}
func recordHandoff() {
condition.lock()
recordedHandoffCount += 1
condition.unlock()
}
}
/// Lets a dedicated dispatch worker observe the lock-protected panic gate
/// without treating the full BLE service as generally Sendable.
private final class PanicIngressObserver: @unchecked Sendable {
private let service: BLEService
init(service: BLEService) {
self.service = service
}
func waitUntilClosed(timeout: TimeInterval) -> Bool {
let deadline = DispatchTime.now().uptimeNanoseconds
+ UInt64(timeout * 1_000_000_000)
while service._test_isPanicIngressOpen,
DispatchTime.now().uptimeNanoseconds < deadline {
Thread.sleep(forTimeInterval: 0.001)
}
return !service._test_isPanicIngressOpen
}
}
private func makeService() -> BLEService {
let keychain = MockKeychain()
let identityManager = MockIdentityManager(keychain)
@@ -744,3 +888,13 @@ private final class PublicCaptureDelegate: BitchatDelegate {
return publicMessages
}
}
@MainActor
private final class TransportEventCaptureDelegate: TransportEventDelegate {
private(set) var messageIDs: [String] = []
func didReceiveTransportEvent(_ event: TransportEvent) {
guard case .messageReceived(let message) = event else { return }
messageIDs.append(message.id)
}
}