mirror of
https://github.com/permissionlesstech/bitchat.git
synced 2026-08-29 07:27:16 +00:00
Merge the two outbox keys into one flush instead of draining them in turn
Flushing the short ID and then the stable key as two calls let newer mail overtake older mail: `flushOutbox(for:)` is keyed by exactly the ID it is handed, so the short-ID queue drained first and a message composed moments ago went out ahead of the offline-composed DM that had been waiting under the stable key. `retrySecurePrivateMessagesAfterAuthentication` already solved this shape for its own alias set; `flushOutbox(forAliases:)` now does the same for the flush — merge the queues, sort by timestamp, and send each message ID once no matter how many keys hold a copy. Two smaller repairs on the way: Prefer the live Noise session key over the stable-ID cache when unified-peer state is not populated yet. Short BLE IDs are ephemeral and get recycled, so a cache entry left by a previous owner of the same ID would name the wrong peer, and the flush would drain — and transmit under — their queue. The session key is the identity of the link we just brought up. Make the authentication-path flush disjoint from the retry that precedes it. On an authenticated link the flush is a strict superset of the retry, so without `skippingSecurelyTransmitted` every retried message went out twice back-to-back and burned two attempts against the cap. Each guard is mutation-proven: dropping the sort fails only the FIFO test (and shows the pre-fix order), neutralising the skip fails only the double-send test, preferring the cache fails only the recycled-ID test, and removing the authentication flush fails only its own test. 1953 tests pass. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
This commit is contained in:
ecgang
parent
623f3855c5
commit
9c473f3841
@ -697,77 +697,7 @@ final class MessageRouter {
|
||||
var outboxChanged = false
|
||||
|
||||
for message in queued {
|
||||
// A synchronous ack from an earlier send in this flush may have
|
||||
// removed an entry from the live outbox. The snapshot is only an
|
||||
// iteration order; never use it to recreate removed messages.
|
||||
guard queuedMessage(message.messageID, for: peerID) != nil else { continue }
|
||||
|
||||
// Skip expired messages (TTL exceeded)
|
||||
if now.timeIntervalSince(message.timestamp) > Self.messageTTLSeconds {
|
||||
SecureLogger.debug("⏰ Expired queued message for \(peerID.id.prefix(8))… id=\(message.messageID.prefix(8))… (age: \(Int(now.timeIntervalSince(message.timestamp)))s)", category: .session)
|
||||
if removeQueuedMessage(message.messageID, for: peerID) {
|
||||
dropMessage(message.messageID, for: peerID)
|
||||
outboxChanged = true
|
||||
}
|
||||
continue
|
||||
}
|
||||
|
||||
if let transport = connectedTransport(for: peerID), transport.canDeliverSecurely(to: peerID) {
|
||||
// A secure session is meaningful enough to retry, but not
|
||||
// proof that this particular ciphertext reached the peer: the
|
||||
// remote app may have restarted while our old session still
|
||||
// looked established. Retain until an ack, while bounding
|
||||
// actual secure transmissions for peers that never ack.
|
||||
guard message.sendAttempts < Self.maxSendAttempts else {
|
||||
SecureLogger.warning("📤 Dropping unacked PM for \(peerID.id.prefix(8))… id=\(message.messageID.prefix(8))… after \(message.sendAttempts) attempts", category: .session)
|
||||
if removeQueuedMessage(message.messageID, for: peerID) {
|
||||
dropMessage(message.messageID, for: peerID)
|
||||
outboxChanged = true
|
||||
}
|
||||
continue
|
||||
}
|
||||
SecureLogger.debug("Outbox -> \(type(of: transport)) (connected) for \(peerID.id.prefix(8))… id=\(message.messageID.prefix(8))…", category: .session)
|
||||
secureTransmissions.insert(
|
||||
PeerMessageKey(peerID: peerID, messageID: message.messageID)
|
||||
)
|
||||
transport.sendPrivateMessage(message.content, to: peerID, recipientNickname: message.nickname, messageID: message.messageID)
|
||||
metrics?.record(.outboxResent)
|
||||
outboxChanged = incrementSendAttemptsIfQueued(message.messageID, for: peerID) || outboxChanged
|
||||
} else if let transport = connectedTransport(for: peerID) {
|
||||
// "Connected" without a secure session — possibly a stolen
|
||||
// binding from a replayed announce: send (a genuine link
|
||||
// finishes the handshake and delivers) but keep retaining
|
||||
// until an ack clears it. These flushes do NOT count toward
|
||||
// the attempt-cap drop: the message was transmitted over a
|
||||
// live link, so a peer whose handshake stalls across
|
||||
// reconnect flapping must not burn through the cap and lose
|
||||
// the store-and-forward copy this retention exists to
|
||||
// preserve. Retention stays bounded by the 24h outbox TTL
|
||||
// and the per-peer FIFO cap.
|
||||
SecureLogger.debug("Outbox -> \(type(of: transport)) (connected, no secure session) for \(peerID.id.prefix(8))… id=\(message.messageID.prefix(8))…", category: .session)
|
||||
secureTransmissions.remove(
|
||||
PeerMessageKey(peerID: peerID, messageID: message.messageID)
|
||||
)
|
||||
transport.sendPrivateMessage(message.content, to: peerID, recipientNickname: message.nickname, messageID: message.messageID)
|
||||
metrics?.record(.outboxResent)
|
||||
} else if let transport = reachableTransport(for: peerID) {
|
||||
// Reachability without a connection is a freshness heuristic,
|
||||
// so the send can silently go nowhere: send but keep retaining
|
||||
// until an ack clears it, bounded by attempt count for peers
|
||||
// that never ack.
|
||||
guard message.sendAttempts < Self.maxSendAttempts else {
|
||||
SecureLogger.warning("📤 Dropping unacked PM for \(peerID.id.prefix(8))… id=\(message.messageID.prefix(8))… after \(message.sendAttempts) attempts", category: .session)
|
||||
if removeQueuedMessage(message.messageID, for: peerID) {
|
||||
dropMessage(message.messageID, for: peerID)
|
||||
outboxChanged = true
|
||||
}
|
||||
continue
|
||||
}
|
||||
SecureLogger.debug("Outbox -> \(type(of: transport)) (reachable) for \(peerID.id.prefix(8))… id=\(message.messageID.prefix(8))…", category: .session)
|
||||
transport.sendPrivateMessage(message.content, to: peerID, recipientNickname: message.nickname, messageID: message.messageID)
|
||||
metrics?.record(.outboxResent)
|
||||
outboxChanged = incrementSendAttemptsIfQueued(message.messageID, for: peerID) || outboxChanged
|
||||
}
|
||||
outboxChanged = flushQueuedMessage(message, for: peerID, now: now) || outboxChanged
|
||||
}
|
||||
|
||||
if outboxChanged {
|
||||
@ -775,6 +705,179 @@ final class MessageRouter {
|
||||
}
|
||||
}
|
||||
|
||||
/// Flush several outbox keys belonging to the same peer as one
|
||||
/// chronological stream.
|
||||
///
|
||||
/// A conversation can leave retained mail split across the ephemeral BLE
|
||||
/// ID and the stable Noise-key ID: a DM composed while the recipient was
|
||||
/// an offline favorite is queued under the stable key, while mail composed
|
||||
/// after they appeared is queued under the short one. `flushOutbox(for:)`
|
||||
/// is keyed by exactly the ID it is handed, so draining the two keys one
|
||||
/// after another would let newer mail on the first key go out ahead of
|
||||
/// older mail on the second. Merge them the way
|
||||
/// `retrySecurePrivateMessagesAfterAuthentication` already does, and send
|
||||
/// each message ID once no matter how many keys hold a copy.
|
||||
///
|
||||
/// Pass `skippingSecurelyTransmitted` when a caller has just run
|
||||
/// `retrySecurePrivateMessagesAfterAuthentication` over the same aliases.
|
||||
/// That retry covers exactly the entries in `secureTransmissions`, and on
|
||||
/// an authenticated link this flush would otherwise be a strict superset
|
||||
/// 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
|
||||
)
|
||||
|
||||
var visitedPeerIDs = Set<PeerID>()
|
||||
var candidates: [Candidate] = []
|
||||
|
||||
for (aliasOrder, peerID) in peerIDAliases.enumerated() {
|
||||
guard visitedPeerIDs.insert(peerID).inserted else { continue }
|
||||
guard let queued = outbox[peerID], !queued.isEmpty else { continue }
|
||||
for (queueOrder, message) in queued.enumerated() {
|
||||
if skippingSecurelyTransmitted,
|
||||
secureTransmissions.contains(
|
||||
PeerMessageKey(peerID: peerID, messageID: message.messageID)
|
||||
) {
|
||||
continue
|
||||
}
|
||||
candidates.append((
|
||||
peerID: peerID,
|
||||
message: message,
|
||||
aliasOrder: aliasOrder,
|
||||
queueOrder: queueOrder
|
||||
))
|
||||
}
|
||||
}
|
||||
|
||||
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
|
||||
}
|
||||
|
||||
SecureLogger.debug(
|
||||
"Flushing merged outbox for \(peerIDAliases.count) alias(es) count=\(candidates.count)",
|
||||
category: .session
|
||||
)
|
||||
|
||||
let now = now()
|
||||
var outboxChanged = false
|
||||
var flushedMessageIDs = Set<String>()
|
||||
|
||||
for candidate in candidates {
|
||||
guard flushedMessageIDs.insert(candidate.message.messageID).inserted else { continue }
|
||||
outboxChanged = flushQueuedMessage(
|
||||
candidate.message,
|
||||
for: candidate.peerID,
|
||||
now: now
|
||||
) || outboxChanged
|
||||
}
|
||||
|
||||
if outboxChanged {
|
||||
persistOutbox()
|
||||
}
|
||||
}
|
||||
|
||||
/// Send one queued message, or drop it if it is past TTL or the attempt
|
||||
/// cap. Returns whether the outbox changed and needs persisting; the
|
||||
/// caller owns the `persistOutbox()` so a whole flush costs one write.
|
||||
private func flushQueuedMessage(
|
||||
_ message: QueuedMessage,
|
||||
for peerID: PeerID,
|
||||
now: Date
|
||||
) -> Bool {
|
||||
var outboxChanged = false
|
||||
|
||||
// A synchronous ack from an earlier send in this flush may have
|
||||
// removed an entry from the live outbox. The snapshot is only an
|
||||
// iteration order; never use it to recreate removed messages.
|
||||
guard queuedMessage(message.messageID, for: peerID) != nil else { return false }
|
||||
|
||||
// Skip expired messages (TTL exceeded)
|
||||
if now.timeIntervalSince(message.timestamp) > Self.messageTTLSeconds {
|
||||
SecureLogger.debug("⏰ Expired queued message for \(peerID.id.prefix(8))… id=\(message.messageID.prefix(8))… (age: \(Int(now.timeIntervalSince(message.timestamp)))s)", category: .session)
|
||||
if removeQueuedMessage(message.messageID, for: peerID) {
|
||||
dropMessage(message.messageID, for: peerID)
|
||||
outboxChanged = true
|
||||
}
|
||||
return outboxChanged
|
||||
}
|
||||
|
||||
if let transport = connectedTransport(for: peerID), transport.canDeliverSecurely(to: peerID) {
|
||||
// A secure session is meaningful enough to retry, but not
|
||||
// proof that this particular ciphertext reached the peer: the
|
||||
// remote app may have restarted while our old session still
|
||||
// looked established. Retain until an ack, while bounding
|
||||
// actual secure transmissions for peers that never ack.
|
||||
guard message.sendAttempts < Self.maxSendAttempts else {
|
||||
SecureLogger.warning("📤 Dropping unacked PM for \(peerID.id.prefix(8))… id=\(message.messageID.prefix(8))… after \(message.sendAttempts) attempts", category: .session)
|
||||
if removeQueuedMessage(message.messageID, for: peerID) {
|
||||
dropMessage(message.messageID, for: peerID)
|
||||
outboxChanged = true
|
||||
}
|
||||
return outboxChanged
|
||||
}
|
||||
SecureLogger.debug("Outbox -> \(type(of: transport)) (connected) for \(peerID.id.prefix(8))… id=\(message.messageID.prefix(8))…", category: .session)
|
||||
secureTransmissions.insert(
|
||||
PeerMessageKey(peerID: peerID, messageID: message.messageID)
|
||||
)
|
||||
transport.sendPrivateMessage(message.content, to: peerID, recipientNickname: message.nickname, messageID: message.messageID)
|
||||
metrics?.record(.outboxResent)
|
||||
outboxChanged = incrementSendAttemptsIfQueued(message.messageID, for: peerID) || outboxChanged
|
||||
} else if let transport = connectedTransport(for: peerID) {
|
||||
// "Connected" without a secure session — possibly a stolen
|
||||
// binding from a replayed announce: send (a genuine link
|
||||
// finishes the handshake and delivers) but keep retaining
|
||||
// until an ack clears it. These flushes do NOT count toward
|
||||
// the attempt-cap drop: the message was transmitted over a
|
||||
// live link, so a peer whose handshake stalls across
|
||||
// reconnect flapping must not burn through the cap and lose
|
||||
// the store-and-forward copy this retention exists to
|
||||
// preserve. Retention stays bounded by the 24h outbox TTL
|
||||
// and the per-peer FIFO cap.
|
||||
SecureLogger.debug("Outbox -> \(type(of: transport)) (connected, no secure session) for \(peerID.id.prefix(8))… id=\(message.messageID.prefix(8))…", category: .session)
|
||||
secureTransmissions.remove(
|
||||
PeerMessageKey(peerID: peerID, messageID: message.messageID)
|
||||
)
|
||||
transport.sendPrivateMessage(message.content, to: peerID, recipientNickname: message.nickname, messageID: message.messageID)
|
||||
metrics?.record(.outboxResent)
|
||||
} else if let transport = reachableTransport(for: peerID) {
|
||||
// Reachability without a connection is a freshness heuristic,
|
||||
// so the send can silently go nowhere: send but keep retaining
|
||||
// until an ack clears it, bounded by attempt count for peers
|
||||
// that never ack.
|
||||
guard message.sendAttempts < Self.maxSendAttempts else {
|
||||
SecureLogger.warning("📤 Dropping unacked PM for \(peerID.id.prefix(8))… id=\(message.messageID.prefix(8))… after \(message.sendAttempts) attempts", category: .session)
|
||||
if removeQueuedMessage(message.messageID, for: peerID) {
|
||||
dropMessage(message.messageID, for: peerID)
|
||||
outboxChanged = true
|
||||
}
|
||||
return outboxChanged
|
||||
}
|
||||
SecureLogger.debug("Outbox -> \(type(of: transport)) (reachable) for \(peerID.id.prefix(8))… id=\(message.messageID.prefix(8))…", category: .session)
|
||||
transport.sendPrivateMessage(message.content, to: peerID, recipientNickname: message.nickname, messageID: message.messageID)
|
||||
metrics?.record(.outboxResent)
|
||||
outboxChanged = incrementSendAttemptsIfQueued(message.messageID, for: peerID) || outboxChanged
|
||||
}
|
||||
|
||||
return outboxChanged
|
||||
}
|
||||
|
||||
func flushAllOutbox() {
|
||||
for key in Array(outbox.keys) { flushOutbox(for: key) }
|
||||
}
|
||||
|
||||
@ -56,7 +56,10 @@ protocol ChatTransportEventContext: AnyObject {
|
||||
func cachedStablePeerID(for shortPeerID: PeerID) -> PeerID?
|
||||
|
||||
// MARK: Routing & acknowledgements
|
||||
func flushRouterOutbox(for peerID: PeerID)
|
||||
/// Drains the message router's disk outbox for every alias of one peer
|
||||
/// as a single chronological stream, so mail split across the ephemeral
|
||||
/// and stable keys cannot be delivered out of order.
|
||||
func flushRouterOutbox(forAliases peerIDAliases: [PeerID], skippingSecurelyTransmitted: Bool)
|
||||
/// Offer queued mail for *other* peers to this newly connected courier.
|
||||
func retryCourierDeposits(via peerID: PeerID)
|
||||
func sendMeshDeliveryAck(for messageID: String, to peerID: PeerID)
|
||||
@ -114,8 +117,11 @@ extension ChatViewModel: ChatTransportEventContext {
|
||||
meshService.noiseSessionPublicKeyData(for: peerID)
|
||||
}
|
||||
|
||||
func flushRouterOutbox(for peerID: PeerID) {
|
||||
messageRouter.flushOutbox(for: peerID)
|
||||
func flushRouterOutbox(forAliases peerIDAliases: [PeerID], skippingSecurelyTransmitted: Bool) {
|
||||
messageRouter.flushOutbox(
|
||||
forAliases: peerIDAliases,
|
||||
skippingSecurelyTransmitted: skippingSecurelyTransmitted
|
||||
)
|
||||
}
|
||||
|
||||
func retryCourierDeposits(via peerID: PeerID) {
|
||||
@ -275,32 +281,38 @@ final class ChatTransportEventCoordinator {
|
||||
context.notifyUIChanged()
|
||||
|
||||
// Resolve the stable key robustly: unified-peer state may not be
|
||||
// populated yet at connect time, so fall back to the cache and then to
|
||||
// the noise session key (mirroring the disconnect path).
|
||||
// populated yet at connect time, so fall back to the live noise
|
||||
// session key and only then to the cache. Order matters — short BLE
|
||||
// IDs are ephemeral and get recycled, so a cache entry left by a
|
||||
// previous owner of this ID would name the wrong peer, while the
|
||||
// session key is the identity of the link we just brought up.
|
||||
var stablePeerID: PeerID?
|
||||
if let peer = context.unifiedPeer(for: peerID) {
|
||||
let resolved = PeerID(hexData: peer.noisePublicKey)
|
||||
context.cacheStablePeerID(resolved, for: peerID)
|
||||
stablePeerID = resolved
|
||||
} else if let cached = context.cachedStablePeerID(for: peerID) {
|
||||
stablePeerID = cached
|
||||
} else if let key = context.noiseSessionPublicKeyData(for: peerID) {
|
||||
let derived = PeerID(hexData: key)
|
||||
context.cacheStablePeerID(derived, for: peerID)
|
||||
stablePeerID = derived
|
||||
} else if let cached = context.cachedStablePeerID(for: peerID) {
|
||||
stablePeerID = cached
|
||||
}
|
||||
|
||||
context.flushRouterOutbox(for: peerID)
|
||||
context.retryCourierDeposits(via: peerID)
|
||||
|
||||
// Also flush under the stable 64-hex key. `flushOutbox` is keyed by
|
||||
// exactly the id it is handed, and a DM composed while the recipient
|
||||
// was an offline favorite is queued under their stable key — so the
|
||||
// short-id flush above never reaches it, and absent a courier it waits
|
||||
// for a relaunch, a favorite-status change, or the 24h TTL.
|
||||
// Flush the short ID and the stable 64-hex key together. `flushOutbox`
|
||||
// is keyed by exactly the ID it is handed, and a DM composed while the
|
||||
// recipient was an offline favorite is queued under their stable key —
|
||||
// so a short-ID flush alone never reaches it, and absent a courier it
|
||||
// waits for a relaunch, a favorite-status change, or the 24h TTL.
|
||||
// Passing both as aliases lets the router merge the two queues
|
||||
// chronologically, so recent mail on the short ID cannot overtake the
|
||||
// older mail that has been waiting under the stable key.
|
||||
var aliases = [peerID]
|
||||
if let stablePeerID, stablePeerID != peerID {
|
||||
context.flushRouterOutbox(for: stablePeerID)
|
||||
aliases.append(stablePeerID)
|
||||
}
|
||||
context.flushRouterOutbox(forAliases: aliases, skippingSecurelyTransmitted: false)
|
||||
context.retryCourierDeposits(via: peerID)
|
||||
}
|
||||
|
||||
func didDisconnectFromPeer(_ peerID: PeerID) {
|
||||
|
||||
@ -45,9 +45,9 @@ protocol ChatVerificationContext: AnyObject {
|
||||
func resolveNickname(for peerID: PeerID) -> String
|
||||
func cachedStablePeerID(for shortPeerID: PeerID) -> PeerID?
|
||||
func cacheStablePeerID(_ stablePeerID: PeerID, for shortPeerID: PeerID)
|
||||
/// Flushes the message router's disk outbox for the given (stable) key so
|
||||
/// mail queued while the peer was offline delivers once it authenticates.
|
||||
func flushRouterOutbox(for peerID: PeerID)
|
||||
/// Drains the message router's disk outbox for every alias of one peer so
|
||||
/// mail queued while they were offline delivers once they authenticate.
|
||||
func flushRouterOutbox(forAliases peerIDAliases: [PeerID], skippingSecurelyTransmitted: Bool)
|
||||
|
||||
// MARK: Noise sessions & verification transport
|
||||
/// Installs the Noise service's session callbacks (single registration point).
|
||||
@ -81,8 +81,9 @@ extension ChatViewModel: ChatVerificationContext {
|
||||
// `isVerifiedFingerprint(_:)`, `setEncryptionStatus(_:for:)`,
|
||||
// `resolveNickname(for:)`, `cachedStablePeerID(for:)`,
|
||||
// `cacheStablePeerID(_:for:)`, `noiseSessionPublicKeyData(for:)`,
|
||||
// `hasEstablishedNoiseSession(with:)`, `triggerHandshake(with:)`, and
|
||||
// `flushRouterOutbox(for:)` (shared with `ChatTransportEventContext`) are
|
||||
// `hasEstablishedNoiseSession(with:)`, `triggerHandshake(with:)`,
|
||||
// and `flushRouterOutbox(forAliases:skippingSecurelyTransmitted:)`
|
||||
// (shared with `ChatTransportEventContext`) are
|
||||
// shared requirements with the other contexts or satisfied by existing
|
||||
// `ChatViewModel` members. The members below flatten nested service
|
||||
// accesses into intent-named calls.
|
||||
@ -260,16 +261,20 @@ final class ChatVerificationCoordinator {
|
||||
self.context.retrySecurePrivateMessagesAfterAuthentication(for: peerIDAliases)
|
||||
|
||||
// The retry above only reaches messages already transmitted
|
||||
// through a secure session (it filters on `sendAttempts > 0`).
|
||||
// A DM composed while this peer was offline was never
|
||||
// transmitted at all — it sits in the outbox under the
|
||||
// stable 64-hex key — so it is in neither that set nor the
|
||||
// short-id flush on connect. Flush it here, now that the
|
||||
// link is authenticated and the stable identity is known,
|
||||
// rather than leaving it for the TTL.
|
||||
if let stablePeerID, stablePeerID != peerID {
|
||||
self.context.flushRouterOutbox(for: stablePeerID)
|
||||
}
|
||||
// through a secure session. A DM composed while this peer
|
||||
// was offline was never transmitted at all — it sits in
|
||||
// the outbox under the stable 64-hex key — so it is in
|
||||
// neither that set nor, when the stable key was still
|
||||
// unresolvable at connect time, the flush on connect.
|
||||
// Flush it here, now that the link is authenticated and
|
||||
// the stable identity is known, rather than leaving it for
|
||||
// the TTL. `skippingSecurelyTransmitted` keeps this
|
||||
// disjoint from the retry above instead of re-sending
|
||||
// everything the retry just put on the air.
|
||||
self.context.flushRouterOutbox(
|
||||
forAliases: peerIDAliases,
|
||||
skippingSecurelyTransmitted: true
|
||||
)
|
||||
|
||||
if var pending = self.pendingQRVerifications[peerID], pending.sent == false {
|
||||
self.context.sendVerifyChallenge(
|
||||
|
||||
@ -124,7 +124,11 @@ private final class MockChatTransportEventContext: ChatTransportEventContext {
|
||||
private(set) var courierRetryPeerIDs: [PeerID] = []
|
||||
private(set) var meshDeliveryAcks: [(messageID: String, peerID: PeerID)] = []
|
||||
|
||||
func flushRouterOutbox(for peerID: PeerID) { flushedOutboxPeerIDs.append(peerID) }
|
||||
private(set) var flushedSkippingSecurelyTransmitted: [Bool] = []
|
||||
func flushRouterOutbox(forAliases peerIDAliases: [PeerID], skippingSecurelyTransmitted: Bool) {
|
||||
flushedOutboxPeerIDs.append(contentsOf: peerIDAliases)
|
||||
flushedSkippingSecurelyTransmitted.append(skippingSecurelyTransmitted)
|
||||
}
|
||||
func retryCourierDeposits(via peerID: PeerID) { courierRetryPeerIDs.append(peerID) }
|
||||
func sendMeshDeliveryAck(for messageID: String, to peerID: PeerID) {
|
||||
meshDeliveryAcks.append((messageID, peerID))
|
||||
@ -550,4 +554,58 @@ struct ChatTransportEventCoordinatorContextTests {
|
||||
.didConnectToPeerSynchronously(shortPeerID)
|
||||
#expect(unresolvable.flushedOutboxPeerIDs == [shortPeerID])
|
||||
}
|
||||
|
||||
/// Both keys must be handed to the router in a *single* call. Two
|
||||
/// sequential single-key flushes would drain the short-ID queue first, so
|
||||
/// mail composed moments ago could be delivered ahead of the older mail
|
||||
/// that has been waiting under the stable key. Only the merged call lets
|
||||
/// the router order the two queues by timestamp.
|
||||
@Test @MainActor
|
||||
func didConnectToPeer_flushesBothKeysInOneMergedCall() {
|
||||
let context = MockChatTransportEventContext()
|
||||
let shortPeerID = PeerID(str: "1122334455667788")
|
||||
let noiseKey = Data((0..<32).map { UInt8(0xD0 &+ $0) })
|
||||
let stablePeerID = PeerID(hexData: noiseKey)
|
||||
|
||||
context.peersByID[shortPeerID] = BitchatPeer(
|
||||
peerID: shortPeerID,
|
||||
noisePublicKey: noiseKey,
|
||||
nickname: "alice"
|
||||
)
|
||||
|
||||
ChatTransportEventCoordinator(context: context)
|
||||
.didConnectToPeerSynchronously(shortPeerID)
|
||||
|
||||
#expect(
|
||||
context.flushedSkippingSecurelyTransmitted.count == 1,
|
||||
"the two keys must be merged into one flush, not drained in sequence"
|
||||
)
|
||||
#expect(context.flushedOutboxPeerIDs == [shortPeerID, stablePeerID])
|
||||
// Connect has no preceding retry pass, so nothing may be skipped.
|
||||
#expect(context.flushedSkippingSecurelyTransmitted == [false])
|
||||
}
|
||||
|
||||
/// Short BLE IDs are ephemeral and get recycled. A cache entry left by a
|
||||
/// previous owner of the same short ID must lose to the identity of the
|
||||
/// link we just brought up, or the flush drains — and transmits under —
|
||||
/// the wrong peer's queue.
|
||||
@Test @MainActor
|
||||
func didConnectToPeer_prefersTheLiveSessionKeyOverAStaleCacheEntry() {
|
||||
let context = MockChatTransportEventContext()
|
||||
let shortPeerID = PeerID(str: "1122334455667788")
|
||||
|
||||
let staleKey = Data(repeating: 0xAA, count: 32)
|
||||
let liveKey = Data(repeating: 0xBB, count: 32)
|
||||
context.cacheStablePeerID(PeerID(hexData: staleKey), for: shortPeerID)
|
||||
context.noiseSessionKeysByPeerID[shortPeerID] = liveKey
|
||||
|
||||
ChatTransportEventCoordinator(context: context)
|
||||
.didConnectToPeerSynchronously(shortPeerID)
|
||||
|
||||
#expect(context.flushedOutboxPeerIDs == [shortPeerID, PeerID(hexData: liveKey)])
|
||||
#expect(
|
||||
!context.flushedOutboxPeerIDs.contains(PeerID(hexData: staleKey)),
|
||||
"a recycled short ID flushed the previous owner's outbox"
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
@ -92,7 +92,11 @@ private final class MockChatVerificationContext: ChatVerificationContext {
|
||||
}
|
||||
|
||||
private(set) var flushedOutboxPeerIDs: [PeerID] = []
|
||||
func flushRouterOutbox(for peerID: PeerID) { flushedOutboxPeerIDs.append(peerID) }
|
||||
private(set) var flushedSkippingSecurelyTransmitted: [Bool] = []
|
||||
func flushRouterOutbox(forAliases peerIDAliases: [PeerID], skippingSecurelyTransmitted: Bool) {
|
||||
flushedOutboxPeerIDs.append(contentsOf: peerIDAliases)
|
||||
flushedSkippingSecurelyTransmitted.append(skippingSecurelyTransmitted)
|
||||
}
|
||||
|
||||
// Noise sessions & verification transport
|
||||
var myNoiseStaticKey = Data(repeating: 0x42, count: 32)
|
||||
@ -303,6 +307,35 @@ struct ChatVerificationCoordinatorContextTests {
|
||||
#expect(context.encryptionStatuses[peerID] == .noiseHandshaking)
|
||||
}
|
||||
|
||||
/// A DM composed while the peer was offline sits in the outbox under their
|
||||
/// stable 64-hex key and was never transmitted, so
|
||||
/// `retrySecurePrivateMessagesAfterAuthentication` — which only covers
|
||||
/// messages already sent through a secure session — cannot reach it. When
|
||||
/// the stable key was still unresolvable at connect time, authentication is
|
||||
/// the first moment it can be flushed at all.
|
||||
@Test @MainActor
|
||||
func peerAuthentication_flushesTheOutboxForBothAliases() async {
|
||||
let context = MockChatVerificationContext()
|
||||
let coordinator = ChatVerificationCoordinator(context: context)
|
||||
let peerID = PeerID(str: "1122334455667788")
|
||||
let noiseKey = Data(repeating: 0x55, count: 32)
|
||||
let stablePeerID = PeerID(hexData: noiseKey)
|
||||
context.noiseSessionKeysByPeerID[peerID] = noiseKey
|
||||
|
||||
coordinator.setupNoiseCallbacks()
|
||||
context.installedCallbacks?.onPeerAuthenticated(peerID, "fp-unverified")
|
||||
await waitForMainQueue()
|
||||
|
||||
#expect(
|
||||
context.flushedOutboxPeerIDs == [peerID, stablePeerID],
|
||||
"offline-queued mail under the stable key was never flushed on authentication"
|
||||
)
|
||||
// The retry pass just transmitted everything in `secureTransmissions`;
|
||||
// flushing that set again would double-send it and burn a second
|
||||
// attempt against the cap.
|
||||
#expect(context.flushedSkippingSecurelyTransmitted == [true])
|
||||
}
|
||||
|
||||
@Test @MainActor
|
||||
func handleVerifyChallengePayload_postsMutualVerificationToastOncePerMinute() async {
|
||||
let context = MockChatVerificationContext()
|
||||
|
||||
@ -167,6 +167,90 @@ struct MessageRouterTests {
|
||||
#expect(transport.sentPrivateMessages.map(\.peerID) == [stablePeerID, shortPeerID])
|
||||
}
|
||||
|
||||
/// 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
|
||||
/// timestamp, exactly as the authentication retry above does.
|
||||
@Test @MainActor
|
||||
func mergedFlush_preservesFIFOAcrossSplitAliases() async {
|
||||
let shortPeerID = PeerID(str: "0000000000000023")
|
||||
let stablePeerID = PeerID(hexData: Data(repeating: 0x23, count: 32))
|
||||
let transport = MockTransport()
|
||||
let clock = MutableTestClock()
|
||||
let router = MessageRouter(transports: [transport], now: { clock.now })
|
||||
|
||||
// Both queue while the peer is unreachable: the older one under the
|
||||
// stable key (composed while they were an offline favorite), the newer
|
||||
// under the ephemeral ID.
|
||||
router.sendPrivate("Older", to: stablePeerID, recipientNickname: "Peer", messageID: "flush-old")
|
||||
clock.now = clock.now.addingTimeInterval(1)
|
||||
router.sendPrivate("Newer", to: shortPeerID, recipientNickname: "Peer", messageID: "flush-new")
|
||||
#expect(transport.sentPrivateMessages.isEmpty)
|
||||
|
||||
transport.connectedPeers = [shortPeerID, stablePeerID]
|
||||
transport.securePeers = [shortPeerID, stablePeerID]
|
||||
router.flushOutbox(forAliases: [shortPeerID, stablePeerID])
|
||||
|
||||
#expect(transport.sentPrivateMessages.map(\.messageID) == ["flush-old", "flush-new"])
|
||||
#expect(transport.sentPrivateMessages.map(\.peerID) == [stablePeerID, shortPeerID])
|
||||
}
|
||||
|
||||
/// The same message ID can sit under both keys after a conversation
|
||||
/// migrates from the ephemeral ID to the stable one. The merged flush must
|
||||
/// put it on the air once, not once per key.
|
||||
@Test @MainActor
|
||||
func mergedFlush_sendsAMessageHeldUnderTwoKeysOnlyOnce() async {
|
||||
let shortPeerID = PeerID(str: "0000000000000024")
|
||||
let stablePeerID = PeerID(hexData: Data(repeating: 0x24, count: 32))
|
||||
let transport = MockTransport()
|
||||
let router = MessageRouter(transports: [transport])
|
||||
|
||||
router.sendPrivate("Migrated", to: shortPeerID, recipientNickname: "Peer", messageID: "dup-1")
|
||||
router.sendPrivate("Migrated", to: stablePeerID, recipientNickname: "Peer", messageID: "dup-1")
|
||||
#expect(transport.sentPrivateMessages.isEmpty)
|
||||
|
||||
transport.connectedPeers = [shortPeerID, stablePeerID]
|
||||
transport.securePeers = [shortPeerID, stablePeerID]
|
||||
router.flushOutbox(forAliases: [shortPeerID, stablePeerID])
|
||||
|
||||
#expect(transport.sentPrivateMessages.map(\.messageID) == ["dup-1"])
|
||||
}
|
||||
|
||||
/// On the authentication path the flush runs straight after
|
||||
/// `retrySecurePrivateMessagesAfterAuthentication`. Without the skip, the
|
||||
/// flush is a strict superset of that retry on an authenticated link, so
|
||||
/// every retried message goes out twice and burns two attempts against the
|
||||
/// cap. Never-transmitted mail must still go out.
|
||||
@Test @MainActor
|
||||
func mergedFlush_skippingSecurelyTransmitted_doesNotResendTheRetriedSet() async {
|
||||
let peerID = PeerID(str: "0000000000000025")
|
||||
let transport = MockTransport()
|
||||
transport.connectedPeers = [peerID]
|
||||
transport.securePeers = [peerID]
|
||||
let router = MessageRouter(transports: [transport])
|
||||
|
||||
// Transmitted through a secure session: this is the retry's territory.
|
||||
// The flush is what marks it as securely transmitted.
|
||||
router.sendPrivate("Transmitted", to: peerID, recipientNickname: "Peer", messageID: "sec-1")
|
||||
router.flushOutbox(for: peerID)
|
||||
#expect(transport.sentPrivateMessages.allSatisfy { $0.messageID == "sec-1" })
|
||||
|
||||
// Never transmitted: queued while the peer was unreachable.
|
||||
transport.connectedPeers = []
|
||||
transport.securePeers = []
|
||||
router.sendPrivate("Offline", to: peerID, recipientNickname: "Peer", messageID: "off-1")
|
||||
transport.connectedPeers = [peerID]
|
||||
transport.securePeers = [peerID]
|
||||
transport.resetRecordings()
|
||||
|
||||
router.flushOutbox(forAliases: [peerID], skippingSecurelyTransmitted: true)
|
||||
|
||||
#expect(
|
||||
transport.sentPrivateMessages.map(\.messageID) == ["off-1"],
|
||||
"the flush re-sent mail the authentication retry already put on the air"
|
||||
)
|
||||
}
|
||||
|
||||
@Test @MainActor
|
||||
func authenticationRetry_doesNotDuplicateNormalPendingHandshakeSend() async {
|
||||
let peerID = PeerID(str: "0000000000000020")
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user