diff --git a/bitchat/Services/MessageRouter.swift b/bitchat/Services/MessageRouter.swift index e3ed1f54..257cdd84 100644 --- a/bitchat/Services/MessageRouter.swift +++ b/bitchat/Services/MessageRouter.swift @@ -587,13 +587,37 @@ final class MessageRouter { /// undecryptable remotely. Normal pre-handshake sends are intentionally /// absent from `secureTransmissions` because BLE already queues /// and drains them when authentication completes. + /// One retained message paired with the alias whose queue holds it, plus + /// the tie-breakers that make a merge across aliases deterministic. + private typealias OutboxCandidate = ( + peerID: PeerID, + message: QueuedMessage, + aliasOrder: Int, + queueOrder: Int + ) + + /// Merge candidates drawn from several alias queues into one chronological + /// stream, so the order the aliases happen to arrive in cannot send newer + /// mail ahead of older mail for the same conversation. + private static func chronologically( + _ candidates: [OutboxCandidate] + ) -> [OutboxCandidate] { + candidates.sorted { lhs, rhs in + if lhs.message.timestamp != rhs.message.timestamp { + return lhs.message.timestamp < rhs.message.timestamp + } + if lhs.aliasOrder != rhs.aliasOrder { + return lhs.aliasOrder < rhs.aliasOrder + } + if lhs.queueOrder != rhs.queueOrder { + return lhs.queueOrder < rhs.queueOrder + } + return lhs.message.messageID < rhs.message.messageID + } + } + func retrySecurePrivateMessagesAfterAuthentication(for peerIDAliases: [PeerID]) { - typealias Candidate = ( - peerID: PeerID, - message: QueuedMessage, - aliasOrder: Int, - queueOrder: Int - ) + typealias Candidate = OutboxCandidate var visitedPeerIDs = Set() var retriedMessageIDs = Set() @@ -625,28 +649,22 @@ final class MessageRouter { // ephemeral and stable outbox keys. Merge both queues into one // chronological stream so callback alias order cannot send newer mail // ahead of older mail. - candidates.sort { lhs, rhs in - if lhs.message.timestamp != rhs.message.timestamp { - return lhs.message.timestamp < rhs.message.timestamp - } - if lhs.aliasOrder != rhs.aliasOrder { - return lhs.aliasOrder < rhs.aliasOrder - } - if lhs.queueOrder != rhs.queueOrder { - return lhs.queueOrder < rhs.queueOrder - } - return lhs.message.messageID < rhs.message.messageID - } + candidates = Self.chronologically(candidates) for candidate in candidates { let peerID = candidate.peerID let message = candidate.message let key = PeerMessageKey(peerID: peerID, messageID: message.messageID) - guard retriedMessageIDs.insert(message.messageID).inserted, - secureTransmissions.contains(key), + // Claim the ID last. A synchronous ack from an earlier send in + // this loop is peer-scoped, so it can clear this copy while the + // twin under the other alias stays live and eligible — and a + // claim made before these checks would let the dead candidate + // suppress that twin, silently skipping a retry that was due. + guard secureTransmissions.contains(key), queuedMessage(message.messageID, for: peerID) != nil, let transport = connectedTransport(for: peerID), - transport.canDeliverSecurely(to: peerID) else { + transport.canDeliverSecurely(to: peerID), + retriedMessageIDs.insert(message.messageID).inserted else { continue } @@ -725,16 +743,10 @@ final class MessageRouter { /// of it — re-sending every message the retry just put on the air and /// burning a second attempt against the cap. Skipping that set makes the /// two passes disjoint by construction rather than by comment. - func flushOutbox(forAliases peerIDAliases: [PeerID], skippingSecurelyTransmitted: Bool = false) { - typealias Candidate = ( - peerID: PeerID, - message: QueuedMessage, - aliasOrder: Int, - queueOrder: Int - ) + func flushOutbox(forAliases peerIDAliases: [PeerID], skippingSecurelyTransmitted: Bool) { + typealias Candidate = OutboxCandidate - var visitedPeerIDs = Set() - for peerID in peerIDAliases { visitedPeerIDs.insert(peerID) } + let aliasSet = Set(peerIDAliases) // The retry that precedes a skipping flush covers a message ID once, // under whichever alias holds the securely-transmitted copy. So the @@ -745,12 +757,12 @@ final class MessageRouter { // precisely the double-send this flag exists to prevent. var retriedMessageIDs = Set() if skippingSecurelyTransmitted { - for key in secureTransmissions where visitedPeerIDs.contains(key.peerID) { + for key in secureTransmissions where aliasSet.contains(key.peerID) { retriedMessageIDs.insert(key.messageID) } } - visitedPeerIDs.removeAll() + var visitedPeerIDs = Set() var candidates: [Candidate] = [] for (aliasOrder, peerID) in peerIDAliases.enumerated() { @@ -769,18 +781,7 @@ final class MessageRouter { guard !candidates.isEmpty else { return } - candidates.sort { lhs, rhs in - if lhs.message.timestamp != rhs.message.timestamp { - return lhs.message.timestamp < rhs.message.timestamp - } - if lhs.aliasOrder != rhs.aliasOrder { - return lhs.aliasOrder < rhs.aliasOrder - } - if lhs.queueOrder != rhs.queueOrder { - return lhs.queueOrder < rhs.queueOrder - } - return lhs.message.messageID < rhs.message.messageID - } + candidates = Self.chronologically(candidates) SecureLogger.debug( "Flushing merged outbox for \(peerIDAliases.count) alias(es) count=\(candidates.count)", diff --git a/bitchatTests/Services/MessageRouterTests.swift b/bitchatTests/Services/MessageRouterTests.swift index 722d1287..497040eb 100644 --- a/bitchatTests/Services/MessageRouterTests.swift +++ b/bitchatTests/Services/MessageRouterTests.swift @@ -167,6 +167,51 @@ struct MessageRouterTests { #expect(transport.sentPrivateMessages.map(\.peerID) == [stablePeerID, shortPeerID]) } + /// The retry claims each message ID once across the alias set. It must not + /// claim one for a candidate that turns out to be dead: a synchronous ack + /// from an earlier send in the same loop is peer-scoped, so it can clear + /// one copy while the twin under the other alias stays live and eligible. + /// A claim made before that check would silently skip a retry that was due. + @Test @MainActor + func authenticationRetry_aDeadCandidateDoesNotSuppressTheLiveTwin() async { + let shortPeerID = PeerID(str: "0000000000000028") + let stablePeerID = PeerID(hexData: Data(repeating: 0x28, count: 32)) + let transport = MockTransport() + transport.connectedPeers = [shortPeerID, stablePeerID] + transport.securePeers = [shortPeerID, stablePeerID] + let clock = MutableTestClock() + let router = MessageRouter(transports: [transport], now: { clock.now }) + + // Oldest, so it is retried first and its ack lands mid-loop. + router.sendPrivate("First", to: shortPeerID, recipientNickname: "Peer", messageID: "r-first") + clock.now = clock.now.addingTimeInterval(1) + router.sendPrivate("Twin", to: shortPeerID, recipientNickname: "Peer", messageID: "r-twin") + clock.now = clock.now.addingTimeInterval(1) + router.sendPrivate("Twin", to: stablePeerID, recipientNickname: "Peer", messageID: "r-twin") + + // Flushing marks all three as securely transmitted, which is what puts + // them in the retry's candidate set. + router.flushOutbox(for: shortPeerID) + router.flushOutbox(for: stablePeerID) + transport.resetRecordings() + + // Retrying the first message synchronously acks the short-ID copy of + // the twin, scoped to that alias alone. + transport.onSendPrivateMessage = { messageID in + guard messageID == "r-first" else { return } + router.markDelivered("r-twin", for: [shortPeerID]) + } + + router.retrySecurePrivateMessagesAfterAuthentication(for: [shortPeerID, stablePeerID]) + transport.onSendPrivateMessage = nil + + #expect( + transport.sentPrivateMessages.map(\.messageID) == ["r-first", "r-twin"], + "a copy acked mid-retry claimed the message ID and suppressed the live twin" + ) + #expect(transport.sentPrivateMessages.map(\.peerID) == [shortPeerID, stablePeerID]) + } + /// Reconnect drains both the ephemeral and the stable outbox key. Draining /// them as two sequential `flushOutbox(for:)` calls would put the newer /// short-ID mail on the air first; the merged flush must order by @@ -189,7 +234,7 @@ struct MessageRouterTests { transport.connectedPeers = [shortPeerID, stablePeerID] transport.securePeers = [shortPeerID, stablePeerID] - router.flushOutbox(forAliases: [shortPeerID, stablePeerID]) + router.flushOutbox(forAliases: [shortPeerID, stablePeerID], skippingSecurelyTransmitted: false) #expect(transport.sentPrivateMessages.map(\.messageID) == ["flush-old", "flush-new"]) #expect(transport.sentPrivateMessages.map(\.peerID) == [stablePeerID, shortPeerID]) @@ -211,7 +256,7 @@ struct MessageRouterTests { transport.connectedPeers = [shortPeerID, stablePeerID] transport.securePeers = [shortPeerID, stablePeerID] - router.flushOutbox(forAliases: [shortPeerID, stablePeerID]) + router.flushOutbox(forAliases: [shortPeerID, stablePeerID], skippingSecurelyTransmitted: false) #expect(transport.sentPrivateMessages.map(\.messageID) == ["dup-1"]) @@ -220,7 +265,7 @@ struct MessageRouterTests { // holds the ID, so the copy the flush passed over goes with it. router.markDelivered("dup-1", for: [shortPeerID, stablePeerID]) transport.resetRecordings() - router.flushOutbox(forAliases: [shortPeerID, stablePeerID]) + router.flushOutbox(forAliases: [shortPeerID, stablePeerID], skippingSecurelyTransmitted: false) #expect( transport.sentPrivateMessages.isEmpty, "the copy the merged flush skipped was re-sent after the ack" @@ -336,7 +381,7 @@ struct MessageRouterTests { router.markDelivered("gone-1", for: [shortPeerID]) } - router.flushOutbox(forAliases: [shortPeerID, stablePeerID]) + router.flushOutbox(forAliases: [shortPeerID, stablePeerID], skippingSecurelyTransmitted: false) transport.onSendPrivateMessage = nil #expect(