mirror of
https://github.com/permissionlesstech/bitchat.git
synced 2026-08-29 07:27:16 +00:00
Merge 265028c712f16bdddb2504eef139e4696df67e83 into 1f59e814f90c3f489f48d68262cb1bf640bf6181
This commit is contained in:
Vincenzo Palazzo
GitHub
commit
ab630c1d47
@ -35,10 +35,10 @@ final class MessageRouter {
|
||||
typealias QueuedMessage = MessageOutboxStore.QueuedMessage
|
||||
|
||||
private struct PeerMessageKey: Hashable {
|
||||
// periphery:ignore - read only via the synthesized Hashable
|
||||
// conformance (dictionary-key identity), which the indexer
|
||||
// cannot attribute; see retain_codable_properties in .periphery.yml
|
||||
// for the same class of false positive.
|
||||
// Both properties are read directly now — `peerID` by the alias-scoped
|
||||
// sweep in `flushOutbox(forAliases:)`, `messageID` throughout — so the
|
||||
// ignore directive this once carried (for reads visible only through
|
||||
// the synthesized Hashable conformance) would itself be flagged.
|
||||
let peerID: PeerID
|
||||
let messageID: String
|
||||
}
|
||||
@ -48,6 +48,9 @@ final class MessageRouter {
|
||||
private let courierDirectory: CourierDirectory
|
||||
private let outboxStore: MessageOutboxStore?
|
||||
private let metrics: StoreAndForwardMetrics?
|
||||
// Stamps our npub onto favorite notifications so the recipient can learn
|
||||
// our Nostr identity. Optional so tests can omit it.
|
||||
private let idBridge: NostrIdentityBridge?
|
||||
|
||||
/// Invoked whenever a retained private message is dropped without a
|
||||
/// delivery ack (attempt cap, TTL expiry, or per-peer overflow eviction)
|
||||
@ -141,9 +144,11 @@ final class MessageRouter {
|
||||
now: @escaping () -> Date = Date.init,
|
||||
courierDirectory: CourierDirectory? = nil,
|
||||
outboxStore: MessageOutboxStore? = nil,
|
||||
metrics: StoreAndForwardMetrics? = nil
|
||||
metrics: StoreAndForwardMetrics? = nil,
|
||||
idBridge: NostrIdentityBridge? = nil
|
||||
) {
|
||||
self.transports = transports
|
||||
self.idBridge = idBridge
|
||||
self.now = now
|
||||
self.courierDirectory = courierDirectory ?? .favoritesBacked()
|
||||
self.outboxStore = outboxStore
|
||||
@ -574,11 +579,19 @@ final class MessageRouter {
|
||||
}
|
||||
|
||||
func sendFavoriteNotification(to peerID: PeerID, isFavorite: Bool) {
|
||||
if let transport = connectedTransport(for: peerID) {
|
||||
transport.sendFavoriteNotification(to: peerID, isFavorite: isFavorite)
|
||||
} else if let transport = reachableTransport(for: peerID) {
|
||||
transport.sendFavoriteNotification(to: peerID, isFavorite: isFavorite)
|
||||
// Route favorites through the outbox instead of fire-and-forget: a
|
||||
// toggle sent while the peer is offline or mid-handshake used to be
|
||||
// silently dropped, which for mesh-only peers meant a missed npub
|
||||
// exchange. The outbox retains until an ack, and the alias flush
|
||||
// drains the queue on reconnect/authentication no matter which PeerID
|
||||
// form the toggle was keyed under. The payload is the same
|
||||
// [FAVORITED]:<npub> content the transports built internally — the
|
||||
// receiver parses the prefix from PM content and never displays it.
|
||||
var content = isFavorite ? "[FAVORITED]" : "[UNFAVORITED]"
|
||||
if let identity = try? idBridge?.getCurrentNostrIdentity() {
|
||||
content += ":" + identity.npub
|
||||
}
|
||||
sendPrivate(content, to: peerID, recipientNickname: "", messageID: UUID().uuidString)
|
||||
}
|
||||
|
||||
/// Retries only messages that the router previously transmitted through
|
||||
@ -589,13 +602,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<PeerID>()
|
||||
var retriedMessageIDs = Set<String>()
|
||||
@ -627,28 +664,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
|
||||
}
|
||||
|
||||
@ -699,77 +730,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 {
|
||||
@ -777,6 +738,181 @@ 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) {
|
||||
typealias Candidate = OutboxCandidate
|
||||
|
||||
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
|
||||
// skip has to be by message ID across the whole alias set, not by
|
||||
// peer/message pair: the same ID can also sit under the *other* alias
|
||||
// without being in `secureTransmissions`, and filtering per pair would
|
||||
// let that copy sail through and put the message on the air twice —
|
||||
// precisely the double-send this flag exists to prevent.
|
||||
var retriedMessageIDs = Set<String>()
|
||||
if skippingSecurelyTransmitted {
|
||||
for key in secureTransmissions where aliasSet.contains(key.peerID) {
|
||||
retriedMessageIDs.insert(key.messageID)
|
||||
}
|
||||
}
|
||||
|
||||
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() {
|
||||
guard !retriedMessageIDs.contains(message.messageID) else { continue }
|
||||
candidates.append((
|
||||
peerID: peerID,
|
||||
message: message,
|
||||
aliasOrder: aliasOrder,
|
||||
queueOrder: queueOrder
|
||||
))
|
||||
}
|
||||
}
|
||||
|
||||
guard !candidates.isEmpty else { return }
|
||||
|
||||
candidates = Self.chronologically(candidates)
|
||||
|
||||
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 {
|
||||
// Claim the ID only for a candidate that is still live. Marking it
|
||||
// flushed first would let a copy removed by a synchronous ack
|
||||
// earlier in this loop suppress the live copy under the other
|
||||
// alias, which would silently drop mail rather than dedup it.
|
||||
guard queuedMessage(candidate.message.messageID, for: candidate.peerID) != nil else {
|
||||
continue
|
||||
}
|
||||
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) {
|
||||
@ -274,12 +280,38 @@ final class ChatTransportEventCoordinator {
|
||||
context.registerEphemeralSession(peerID: peerID)
|
||||
context.notifyUIChanged()
|
||||
|
||||
// Resolve the stable key robustly: unified-peer state may not be
|
||||
// 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 stablePeerID = PeerID(hexData: peer.noisePublicKey)
|
||||
context.cacheStablePeerID(stablePeerID, for: peerID)
|
||||
let resolved = PeerID(hexData: peer.noisePublicKey)
|
||||
context.cacheStablePeerID(resolved, for: peerID)
|
||||
stablePeerID = resolved
|
||||
} 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)
|
||||
// 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 {
|
||||
aliases.append(stablePeerID)
|
||||
}
|
||||
context.flushRouterOutbox(forAliases: aliases, skippingSecurelyTransmitted: false)
|
||||
context.retryCourierDeposits(via: peerID)
|
||||
}
|
||||
|
||||
|
||||
@ -45,6 +45,9 @@ protocol ChatVerificationContext: AnyObject {
|
||||
func resolveNickname(for peerID: PeerID) -> String
|
||||
func cachedStablePeerID(for shortPeerID: PeerID) -> PeerID?
|
||||
func cacheStablePeerID(_ stablePeerID: PeerID, for shortPeerID: 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).
|
||||
@ -78,7 +81,9 @@ extension ChatViewModel: ChatVerificationContext {
|
||||
// `isVerifiedFingerprint(_:)`, `setEncryptionStatus(_:for:)`,
|
||||
// `resolveNickname(for:)`, `cachedStablePeerID(for:)`,
|
||||
// `cacheStablePeerID(_:for:)`, `noiseSessionPublicKeyData(for:)`,
|
||||
// `hasEstablishedNoiseSession(with:)`, and `triggerHandshake(with:)` 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.
|
||||
@ -248,9 +253,9 @@ final class ChatVerificationCoordinator {
|
||||
// retry it now that this newly authenticated/replacement
|
||||
// session can actually decrypt it.
|
||||
var peerIDAliases = [peerID]
|
||||
if let stablePeerID = authenticatedStablePeerID
|
||||
?? self.context.cachedStablePeerID(for: peerID),
|
||||
stablePeerID != peerID {
|
||||
let stablePeerID = authenticatedStablePeerID
|
||||
?? self.context.cachedStablePeerID(for: peerID)
|
||||
if let stablePeerID, stablePeerID != peerID {
|
||||
// Conversations can migrate from the ephemeral BLE ID
|
||||
// to the authenticated Noise-key ID. Retry both aliases
|
||||
// because either may own the retained outbox entry.
|
||||
@ -258,6 +263,22 @@ final class ChatVerificationCoordinator {
|
||||
}
|
||||
self.context.retrySecurePrivateMessagesAfterAuthentication(for: peerIDAliases)
|
||||
|
||||
// The retry above only reaches messages already transmitted
|
||||
// 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(
|
||||
to: peerID,
|
||||
|
||||
@ -33,7 +33,8 @@ struct ChatViewModelServiceBundle {
|
||||
let messageRouter = MessageRouter(
|
||||
transports: [meshService, nostrTransport],
|
||||
outboxStore: outboxStore,
|
||||
metrics: sfMetrics
|
||||
metrics: sfMetrics,
|
||||
idBridge: idBridge
|
||||
)
|
||||
|
||||
self.commandProcessor = commandProcessor
|
||||
|
||||
@ -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))
|
||||
@ -343,7 +347,10 @@ struct ChatTransportEventCoordinatorContextTests {
|
||||
#expect(context.isConnected)
|
||||
#expect(context.registeredEphemeralSessions == [peerID])
|
||||
#expect(context.stablePeerIDCache[peerID] == PeerID(hexData: noiseKey))
|
||||
#expect(context.flushedOutboxPeerIDs == [peerID])
|
||||
// Both queues: the short id, then the stable 64-hex key that offline-
|
||||
// composed mail is queued under (#1408). Order matters only in that the
|
||||
// short-id flush is not delayed behind the stable-key resolution.
|
||||
#expect(context.flushedOutboxPeerIDs == [peerID, PeerID(hexData: noiseKey)])
|
||||
#expect(context.notifyUIChangedCount == 1)
|
||||
|
||||
// Their messages' read receipts are un-marked on disconnect so READ
|
||||
@ -480,4 +487,125 @@ struct ChatTransportEventCoordinatorContextTests {
|
||||
#expect(context.handledPrivateMessages.count == 1)
|
||||
#expect(context.meshDeliveryAcks.count == 1)
|
||||
}
|
||||
|
||||
/// #1408: a DM composed while the recipient was an offline favorite is
|
||||
/// queued in the router outbox under the peer's STABLE 64-hex Noise key.
|
||||
/// `flushOutbox` is keyed by exactly the id it is handed, so flushing only
|
||||
/// the short 16-hex id on connect never reaches that queue — the mail waits
|
||||
/// for an app relaunch, a favorite-status change, or the 24h TTL.
|
||||
///
|
||||
/// Reconnect must flush both.
|
||||
@Test @MainActor
|
||||
func didConnectToPeer_flushesTheStableKeyOutboxAsWellAsTheShortID() {
|
||||
let context = MockChatTransportEventContext()
|
||||
let coordinator = ChatTransportEventCoordinator(context: context)
|
||||
|
||||
let shortPeerID = PeerID(str: "1122334455667788")
|
||||
let noiseKey = Data((0..<32).map { UInt8(0xB0 &+ $0) })
|
||||
let stablePeerID = PeerID(hexData: noiseKey)
|
||||
#expect(stablePeerID != shortPeerID)
|
||||
|
||||
// The peer is known by its Noise key, as it is after a handshake.
|
||||
context.peersByID[shortPeerID] = BitchatPeer(
|
||||
peerID: shortPeerID,
|
||||
noisePublicKey: noiseKey,
|
||||
nickname: "alice"
|
||||
)
|
||||
|
||||
coordinator.didConnectToPeerSynchronously(shortPeerID)
|
||||
|
||||
// Both queues are drained, and the stable key is the one that carries
|
||||
// the offline-composed mail.
|
||||
#expect(context.flushedOutboxPeerIDs.contains(shortPeerID))
|
||||
#expect(
|
||||
context.flushedOutboxPeerIDs.contains(stablePeerID),
|
||||
"offline-queued mail under the stable key was never flushed"
|
||||
)
|
||||
}
|
||||
|
||||
/// The stable key must still resolve when unified-peer state has not been
|
||||
/// populated yet at connect time — otherwise the flush silently no-ops in
|
||||
/// exactly the case it is for. Falls back to the cache, then to the Noise
|
||||
/// session key.
|
||||
@Test @MainActor
|
||||
func didConnectToPeer_resolvesTheStableKeyWithoutUnifiedPeerState() {
|
||||
let shortPeerID = PeerID(str: "1122334455667788")
|
||||
let noiseKey = Data((0..<32).map { UInt8(0xC0 &+ $0) })
|
||||
let stablePeerID = PeerID(hexData: noiseKey)
|
||||
|
||||
// Cache only.
|
||||
let viaCache = MockChatTransportEventContext()
|
||||
viaCache.cacheStablePeerID(stablePeerID, for: shortPeerID)
|
||||
ChatTransportEventCoordinator(context: viaCache)
|
||||
.didConnectToPeerSynchronously(shortPeerID)
|
||||
#expect(viaCache.flushedOutboxPeerIDs.contains(stablePeerID))
|
||||
|
||||
// Noise session key only.
|
||||
let viaSession = MockChatTransportEventContext()
|
||||
viaSession.noiseSessionKeysByPeerID[shortPeerID] = noiseKey
|
||||
ChatTransportEventCoordinator(context: viaSession)
|
||||
.didConnectToPeerSynchronously(shortPeerID)
|
||||
#expect(viaSession.flushedOutboxPeerIDs.contains(stablePeerID))
|
||||
|
||||
// Nothing resolvable: the short-id flush still happens, and no bogus
|
||||
// second flush is issued.
|
||||
let unresolvable = MockChatTransportEventContext()
|
||||
ChatTransportEventCoordinator(context: unresolvable)
|
||||
.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"
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
@ -91,6 +91,13 @@ private final class MockChatVerificationContext: ChatVerificationContext {
|
||||
stablePeerIDCache[shortPeerID] = stablePeerID
|
||||
}
|
||||
|
||||
private(set) var flushedOutboxPeerIDs: [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)
|
||||
var establishedNoiseSessions: Set<PeerID> = []
|
||||
@ -300,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,230 @@ 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
|
||||
/// 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], skippingSecurelyTransmitted: false)
|
||||
|
||||
#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], skippingSecurelyTransmitted: false)
|
||||
|
||||
#expect(transport.sentPrivateMessages.map(\.messageID) == ["dup-1"])
|
||||
|
||||
// The skipped copy must not resurface as a duplicate once the ack
|
||||
// arrives: an ack scoped to the peer's aliases clears every key that
|
||||
// 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], skippingSecurelyTransmitted: false)
|
||||
#expect(
|
||||
transport.sentPrivateMessages.isEmpty,
|
||||
"the copy the merged flush skipped was re-sent after the ack"
|
||||
)
|
||||
}
|
||||
|
||||
/// 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"
|
||||
)
|
||||
}
|
||||
|
||||
/// The skip must be by message ID across the whole alias set, not by
|
||||
/// peer/message pair. A migrated conversation holds the same ID under both
|
||||
/// keys, but only the copy that was actually transmitted is in
|
||||
/// `secureTransmissions` — so a per-pair filter excludes that one and lets
|
||||
/// the untransmitted twin through, putting the message on the air twice in
|
||||
/// the very pass that was meant to prevent it.
|
||||
@Test @MainActor
|
||||
func mergedFlush_skippingSecurelyTransmitted_coversTheTwinUnderTheOtherAlias() async {
|
||||
let shortPeerID = PeerID(str: "0000000000000026")
|
||||
let stablePeerID = PeerID(hexData: Data(repeating: 0x26, count: 32))
|
||||
let transport = MockTransport()
|
||||
let router = MessageRouter(transports: [transport])
|
||||
|
||||
// The short-ID copy is transmitted securely, so it lands in
|
||||
// `secureTransmissions` — the retry's territory.
|
||||
transport.connectedPeers = [shortPeerID]
|
||||
transport.securePeers = [shortPeerID]
|
||||
router.sendPrivate("Migrated", to: shortPeerID, recipientNickname: "Peer", messageID: "twin-1")
|
||||
router.flushOutbox(for: shortPeerID)
|
||||
|
||||
// The same message also sits under the stable key, never transmitted.
|
||||
transport.connectedPeers = []
|
||||
transport.securePeers = []
|
||||
router.sendPrivate("Migrated", to: stablePeerID, recipientNickname: "Peer", messageID: "twin-1")
|
||||
|
||||
transport.connectedPeers = [shortPeerID, stablePeerID]
|
||||
transport.securePeers = [shortPeerID, stablePeerID]
|
||||
transport.resetRecordings()
|
||||
|
||||
router.flushOutbox(
|
||||
forAliases: [shortPeerID, stablePeerID],
|
||||
skippingSecurelyTransmitted: true
|
||||
)
|
||||
|
||||
#expect(
|
||||
transport.sentPrivateMessages.isEmpty,
|
||||
"the untransmitted twin was sent even though the retry already covered this ID"
|
||||
)
|
||||
}
|
||||
|
||||
/// A synchronous ack fired by an earlier send in the same flush removes an
|
||||
/// entry from the live outbox. The merged flush must not let that dead
|
||||
/// candidate claim the message ID, or the live copy under the other alias
|
||||
/// is silently dropped instead of deduped.
|
||||
@Test @MainActor
|
||||
func mergedFlush_aDeadFirstCandidateDoesNotSuppressTheLiveTwin() async {
|
||||
let shortPeerID = PeerID(str: "0000000000000027")
|
||||
let stablePeerID = PeerID(hexData: Data(repeating: 0x27, count: 32))
|
||||
let transport = MockTransport()
|
||||
let clock = MutableTestClock()
|
||||
let router = MessageRouter(transports: [transport], now: { clock.now })
|
||||
|
||||
// Oldest, so it is flushed first and its ack lands mid-loop.
|
||||
router.sendPrivate("First", to: shortPeerID, recipientNickname: "Peer", messageID: "first-1")
|
||||
clock.now = clock.now.addingTimeInterval(1)
|
||||
// "gone-1" queues under the short ID before the stable one, so it sorts
|
||||
// ahead of its twin and is the candidate that would claim the ID.
|
||||
router.sendPrivate("Gone", to: shortPeerID, recipientNickname: "Peer", messageID: "gone-1")
|
||||
clock.now = clock.now.addingTimeInterval(1)
|
||||
router.sendPrivate("Gone", to: stablePeerID, recipientNickname: "Peer", messageID: "gone-1")
|
||||
|
||||
transport.connectedPeers = [shortPeerID, stablePeerID]
|
||||
transport.securePeers = [shortPeerID, stablePeerID]
|
||||
transport.resetRecordings()
|
||||
|
||||
// Sending the first message synchronously acks the short-ID copy of
|
||||
// "gone-1" — scoped to that alias alone, which is the deliberate
|
||||
// behaviour for an ID that is also queued elsewhere. The candidate
|
||||
// list was snapshotted before this, so it still holds the dead copy.
|
||||
transport.onSendPrivateMessage = { messageID in
|
||||
guard messageID == "first-1" else { return }
|
||||
router.markDelivered("gone-1", for: [shortPeerID])
|
||||
}
|
||||
|
||||
router.flushOutbox(forAliases: [shortPeerID, stablePeerID], skippingSecurelyTransmitted: false)
|
||||
transport.onSendPrivateMessage = nil
|
||||
|
||||
#expect(
|
||||
transport.sentPrivateMessages.map(\.messageID) == ["first-1", "gone-1"],
|
||||
"a copy removed mid-flush claimed the message ID and suppressed the live twin"
|
||||
)
|
||||
#expect(transport.sentPrivateMessages.map(\.peerID) == [shortPeerID, stablePeerID])
|
||||
}
|
||||
|
||||
@Test @MainActor
|
||||
func authenticationRetry_doesNotDuplicateNormalPendingHandshakeSend() async {
|
||||
let peerID = PeerID(str: "0000000000000020")
|
||||
@ -429,7 +653,7 @@ struct MessageRouterTests {
|
||||
}
|
||||
|
||||
@Test @MainActor
|
||||
func sendFavoriteNotification_usesConnectedOrReachable() async {
|
||||
func sendFavoriteNotification_routesThroughOutboxAsPrivateMessage() async {
|
||||
let peerID = PeerID(str: "0000000000000004")
|
||||
let transport = MockTransport()
|
||||
transport.reachablePeers.insert(peerID)
|
||||
@ -437,7 +661,28 @@ struct MessageRouterTests {
|
||||
let router = MessageRouter(transports: [transport])
|
||||
router.sendFavoriteNotification(to: peerID, isFavorite: true)
|
||||
|
||||
#expect(transport.sentFavoriteNotifications.count == 1)
|
||||
// Favorites ride the outbox (sendPrivate) so they survive offline /
|
||||
// handshake gaps; the recipient parses the [FAVORITED] prefix.
|
||||
#expect(transport.sentFavoriteNotifications.isEmpty)
|
||||
#expect(transport.sentPrivateMessages.count == 1)
|
||||
#expect(transport.sentPrivateMessages.first?.content.hasPrefix("[FAVORITED]") == true)
|
||||
}
|
||||
|
||||
@Test @MainActor
|
||||
func sendFavoriteNotification_whileOffline_queuesAndFlushesOnReconnect() async {
|
||||
let peerID = PeerID(str: "0000000000000009")
|
||||
let transport = MockTransport()
|
||||
// Peer is neither connected nor reachable: the favorite must be retained.
|
||||
let router = MessageRouter(transports: [transport])
|
||||
router.sendFavoriteNotification(to: peerID, isFavorite: false)
|
||||
#expect(transport.sentPrivateMessages.isEmpty)
|
||||
|
||||
// Peer comes back: the queued favorite flushes.
|
||||
transport.connectedPeers.insert(peerID)
|
||||
router.flushOutbox(for: peerID)
|
||||
|
||||
#expect(transport.sentPrivateMessages.count == 1)
|
||||
#expect(transport.sentPrivateMessages.first?.content.hasPrefix("[UNFAVORITED]") == true)
|
||||
}
|
||||
|
||||
// MARK: - Courier deposits
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user