mirror of
https://github.com/permissionlesstech/bitchat.git
synced 2026-08-22 07:16:03 +00:00
Merge 9e95374bca5f2d278469fc57075a54bb1453593f into 1f59e814f90c3f489f48d68262cb1bf640bf6181
This commit is contained in:
ecgang
GitHub
commit
43629946cd
@ -39,6 +39,15 @@ final class MessageRouter {
|
||||
// conformance (dictionary-key identity), which the indexer
|
||||
// cannot attribute; see retain_codable_properties in .periphery.yml
|
||||
// for the same class of false positive.
|
||||
//
|
||||
// The alias-scoped sweep in `flushOutbox(forAliases:)` used to read
|
||||
// this directly, which is why the suppression was dropped in a450204.
|
||||
// That sweep was the bug: it derived the flush's skip set from every
|
||||
// `secureTransmissions` entry under the aliases with no liveness
|
||||
// check, so removing it brought the false positive back. Do not delete
|
||||
// the property to satisfy the scanner — it is what scopes
|
||||
// `secureTransmissions` and `dropMessage` per peer, and collapsing it
|
||||
// would let one alias's drop clear its twin under the other.
|
||||
let peerID: PeerID
|
||||
let messageID: String
|
||||
}
|
||||
@ -589,16 +598,52 @@ final class MessageRouter {
|
||||
/// undecryptable remotely. Normal pre-handshake sends are intentionally
|
||||
/// absent from `secureTransmissions` because BLE already queues
|
||||
/// and drains them when authentication completes.
|
||||
func retrySecurePrivateMessagesAfterAuthentication(for peerIDAliases: [PeerID]) {
|
||||
typealias Candidate = (
|
||||
peerID: PeerID,
|
||||
message: QueuedMessage,
|
||||
aliasOrder: Int,
|
||||
queueOrder: Int
|
||||
)
|
||||
/// 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
|
||||
}
|
||||
}
|
||||
|
||||
/// Returns the message IDs this pass actually put on the air, so a flush
|
||||
/// that follows can skip exactly those and nothing more. Deriving that set
|
||||
/// from `secureTransmissions` instead would over-skip: an alias with no
|
||||
/// live secure transport is abandoned wholesale below, and its entries are
|
||||
/// in that set while never having been sent.
|
||||
///
|
||||
/// Deliberately not `@discardableResult`. A caller that drops this set and
|
||||
/// passes an empty one to `flushOutbox` re-sends everything this pass just
|
||||
/// put on the air and burns a second attempt against the cap. An unused
|
||||
/// result is the one thing that catches that at compile time, so the tests
|
||||
/// that genuinely do not care spell it `_ =`.
|
||||
func retrySecurePrivateMessagesAfterAuthentication(for peerIDAliases: [PeerID]) -> Set<String> {
|
||||
typealias Candidate = OutboxCandidate
|
||||
|
||||
var visitedPeerIDs = Set<PeerID>()
|
||||
var retriedMessageIDs = Set<String>()
|
||||
var transmittedMessageIDs = Set<String>()
|
||||
var outboxChanged = false
|
||||
let currentDate = now()
|
||||
var candidates: [Candidate] = []
|
||||
@ -627,25 +672,16 @@ 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),
|
||||
// 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.
|
||||
guard secureTransmissions.contains(key),
|
||||
queuedMessage(message.messageID, for: peerID) != nil,
|
||||
let transport = connectedTransport(for: peerID),
|
||||
transport.canDeliverSecurely(to: peerID) else {
|
||||
@ -672,6 +708,16 @@ final class MessageRouter {
|
||||
continue
|
||||
}
|
||||
|
||||
// Claim the ID only now — past every check that can drop this
|
||||
// candidate instead of sending it. Claiming in the guard chain
|
||||
// above let a copy that was expired or past the attempt cap take
|
||||
// the ID, drop itself, and suppress the live twin under the other
|
||||
// alias, which then went unsent for this whole pass. The flush
|
||||
// that follows still recovered it, but the drop had already
|
||||
// reported the message failed to the UI while it was in fact
|
||||
// about to deliver.
|
||||
guard retriedMessageIDs.insert(message.messageID).inserted else { continue }
|
||||
|
||||
SecureLogger.debug(
|
||||
"Auth retry -> \(type(of: transport)) for \(peerID.id.prefix(8))… id=\(message.messageID.prefix(8))…",
|
||||
category: .session
|
||||
@ -683,12 +729,15 @@ final class MessageRouter {
|
||||
messageID: message.messageID
|
||||
)
|
||||
metrics?.record(.outboxResent)
|
||||
transmittedMessageIDs.insert(message.messageID)
|
||||
outboxChanged = incrementSendAttemptsIfQueued(message.messageID, for: peerID) || outboxChanged
|
||||
}
|
||||
|
||||
if outboxChanged {
|
||||
persistOutbox()
|
||||
}
|
||||
|
||||
return transmittedMessageIDs
|
||||
}
|
||||
|
||||
func flushOutbox(for peerID: PeerID) {
|
||||
@ -699,76 +748,94 @@ 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 }
|
||||
outboxChanged = flushQueuedMessage(message, for: peerID, now: now).outboxChanged || outboxChanged
|
||||
}
|
||||
|
||||
// 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 outboxChanged {
|
||||
persistOutbox()
|
||||
}
|
||||
}
|
||||
|
||||
/// 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 the set returned by `retrySecurePrivateMessagesAfterAuthentication`
|
||||
/// as `skippingMessageIDs` when a caller has just run it over the same
|
||||
/// aliases. Those messages are already on the air, and re-sending them here
|
||||
/// would burn a second attempt against the cap.
|
||||
///
|
||||
/// It has to be the IDs that retry *transmitted*, not the entries in
|
||||
/// `secureTransmissions` matching these aliases. Those differ: the retry
|
||||
/// abandons an alias wholesale when it has no live secure transport, so a
|
||||
/// message under that alias is in `secureTransmissions` yet was never sent.
|
||||
/// Skipping the wider set left such a message neither retried nor flushed,
|
||||
/// waiting on the next reconnect or the 24h TTL — the exact delay this
|
||||
/// flush exists to remove.
|
||||
///
|
||||
/// The skip is by message ID across the whole alias set rather than by
|
||||
/// peer/message pair, because the same ID can sit under both aliases and
|
||||
/// filtering per pair would put it on the air twice.
|
||||
func flushOutbox(forAliases peerIDAliases: [PeerID], skippingMessageIDs: Set<String>) {
|
||||
typealias Candidate = OutboxCandidate
|
||||
|
||||
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 !skippingMessageIDs.contains(message.messageID) else { continue }
|
||||
candidates.append((
|
||||
peerID: peerID,
|
||||
message: message,
|
||||
aliasOrder: aliasOrder,
|
||||
queueOrder: queueOrder
|
||||
))
|
||||
}
|
||||
}
|
||||
|
||||
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
|
||||
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 once this alias actually put the message on a
|
||||
// transport. Claiming any earlier — before the liveness check, or
|
||||
// merely because the candidate was still queued — lets an alias
|
||||
// that sends nothing suppress the live twin under the other alias,
|
||||
// silently dropping mail instead of deduping it. A copy removed by
|
||||
// a synchronous ack earlier in this loop, or an alias with no
|
||||
// transport at all, both fail to send and so leave the twin
|
||||
// eligible.
|
||||
guard !flushedMessageIDs.contains(candidate.message.messageID) else { continue }
|
||||
let attempt = flushQueuedMessage(
|
||||
candidate.message,
|
||||
for: candidate.peerID,
|
||||
now: now
|
||||
)
|
||||
outboxChanged = attempt.outboxChanged || outboxChanged
|
||||
if attempt.sent {
|
||||
flushedMessageIDs.insert(candidate.message.messageID)
|
||||
}
|
||||
}
|
||||
|
||||
@ -777,6 +844,108 @@ final class MessageRouter {
|
||||
}
|
||||
}
|
||||
|
||||
/// What one flush attempt did. The two facts are independent and neither
|
||||
/// implies the other: a message past TTL is dropped without being sent
|
||||
/// (`outboxChanged`, not `sent`), while a send over a connected link with
|
||||
/// no secure session deliberately does not touch the attempt count
|
||||
/// (`sent`, not `outboxChanged`). A caller deduping by message ID needs
|
||||
/// `sent`; a caller deciding whether to persist needs `outboxChanged`.
|
||||
private struct FlushAttempt {
|
||||
let sent: Bool
|
||||
let outboxChanged: Bool
|
||||
}
|
||||
|
||||
/// Send one queued message, or drop it if it is past TTL or the attempt
|
||||
/// cap. The caller owns the `persistOutbox()` so a whole flush costs one
|
||||
/// write.
|
||||
private func flushQueuedMessage(
|
||||
_ message: QueuedMessage,
|
||||
for peerID: PeerID,
|
||||
now: Date
|
||||
) -> FlushAttempt {
|
||||
var outboxChanged = false
|
||||
var sent = 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 FlushAttempt(sent: false, outboxChanged: 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 FlushAttempt(sent: false, outboxChanged: 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 FlushAttempt(sent: false, outboxChanged: 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)
|
||||
sent = true
|
||||
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)
|
||||
sent = true
|
||||
} 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 FlushAttempt(sent: false, outboxChanged: 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)
|
||||
sent = true
|
||||
outboxChanged = incrementSendAttemptsIfQueued(message.messageID, for: peerID) || outboxChanged
|
||||
}
|
||||
|
||||
return FlushAttempt(sent: sent, outboxChanged: 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], skippingMessageIDs: Set<String>)
|
||||
/// 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], skippingMessageIDs: Set<String>) {
|
||||
messageRouter.flushOutbox(
|
||||
forAliases: peerIDAliases,
|
||||
skippingMessageIDs: skippingMessageIDs
|
||||
)
|
||||
}
|
||||
|
||||
func retryCourierDeposits(via peerID: PeerID) {
|
||||
@ -274,12 +280,42 @@ final class ChatTransportEventCoordinator {
|
||||
context.registerEphemeralSession(peerID: peerID)
|
||||
context.notifyUIChanged()
|
||||
|
||||
// Resolve the stable key from evidence about *this* link only:
|
||||
// unified-peer state, or the live noise session key. Both name the peer
|
||||
// we just brought up.
|
||||
//
|
||||
// Deliberately no cache fallback. Short BLE IDs are ephemeral and get
|
||||
// recycled, so a cache entry left by a previous owner of this ID names
|
||||
// the wrong peer — and reaching the fallback means both live sources
|
||||
// were absent, which is exactly when there is nothing to catch the
|
||||
// mistake. Flushing `[shortID, wrongStableID]` would drain a stranger's
|
||||
// queue and silently skip the right one. When neither live source has
|
||||
// resolved yet, the short-ID flush below still runs, and authentication
|
||||
// flushes both aliases once the identity is known.
|
||||
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
|
||||
}
|
||||
|
||||
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, skippingMessageIDs: [])
|
||||
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], skippingMessageIDs: Set<String>)
|
||||
|
||||
// MARK: Noise sessions & verification transport
|
||||
/// Installs the Noise service's session callbacks (single registration point).
|
||||
@ -62,7 +65,7 @@ protocol ChatVerificationContext: AnyObject {
|
||||
/// Retries only private messages previously transmitted through a secure
|
||||
/// session and still pending an ack. Both ephemeral and stable aliases
|
||||
/// are supplied because either can own the outbox entry.
|
||||
func retrySecurePrivateMessagesAfterAuthentication(for peerIDAliases: [PeerID])
|
||||
func retrySecurePrivateMessagesAfterAuthentication(for peerIDAliases: [PeerID]) -> Set<String>
|
||||
func sendVerifyChallenge(to peerID: PeerID, noiseKeyHex: String, nonceA: Data)
|
||||
func sendVerifyResponse(to peerID: PeerID, noiseKeyHex: String, nonceA: Data)
|
||||
|
||||
@ -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:skippingMessageIDs:)`
|
||||
// (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.
|
||||
@ -125,7 +130,7 @@ extension ChatViewModel: ChatVerificationContext {
|
||||
mediaTransferCoordinator.peerDidAuthenticate(peerID.toShort())
|
||||
}
|
||||
|
||||
func retrySecurePrivateMessagesAfterAuthentication(for peerIDAliases: [PeerID]) {
|
||||
func retrySecurePrivateMessagesAfterAuthentication(for peerIDAliases: [PeerID]) -> Set<String> {
|
||||
messageRouter.retrySecurePrivateMessagesAfterAuthentication(for: peerIDAliases)
|
||||
}
|
||||
|
||||
@ -248,15 +253,32 @@ 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.
|
||||
peerIDAliases.append(stablePeerID)
|
||||
}
|
||||
self.context.retrySecurePrivateMessagesAfterAuthentication(for: peerIDAliases)
|
||||
let retried = 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. Skipping exactly what the retry transmitted
|
||||
// keeps the two passes disjoint without stranding a
|
||||
// message the retry skipped for want of a live transport.
|
||||
self.context.flushRouterOutbox(
|
||||
forAliases: peerIDAliases,
|
||||
skippingMessageIDs: retried
|
||||
)
|
||||
|
||||
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 flushedSkippingMessageIDs: [Set<String>] = []
|
||||
func flushRouterOutbox(forAliases peerIDAliases: [PeerID], skippingMessageIDs: Set<String>) {
|
||||
flushedOutboxPeerIDs.append(contentsOf: peerIDAliases)
|
||||
flushedSkippingMessageIDs.append(skippingMessageIDs)
|
||||
}
|
||||
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,130 @@ 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. It resolves from the live Noise session key,
|
||||
/// never from the cache alone.
|
||||
@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, with no live evidence for this link. Short BLE IDs are
|
||||
// recycled, so the entry may belong to a previous owner of this ID —
|
||||
// flushing it would drain a stranger's queue and skip the right one.
|
||||
let viaCache = MockChatTransportEventContext()
|
||||
viaCache.cacheStablePeerID(stablePeerID, for: shortPeerID)
|
||||
ChatTransportEventCoordinator(context: viaCache)
|
||||
.didConnectToPeerSynchronously(shortPeerID)
|
||||
#expect(
|
||||
viaCache.flushedOutboxPeerIDs == [shortPeerID],
|
||||
"a cache entry with no live corroboration named the stable peer"
|
||||
)
|
||||
|
||||
// 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.flushedSkippingMessageIDs.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.flushedSkippingMessageIDs == [[]])
|
||||
}
|
||||
|
||||
/// 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 flushedSkippingMessageIDs: [Set<String>] = []
|
||||
func flushRouterOutbox(forAliases peerIDAliases: [PeerID], skippingMessageIDs: Set<String>) {
|
||||
flushedOutboxPeerIDs.append(contentsOf: peerIDAliases)
|
||||
flushedSkippingMessageIDs.append(skippingMessageIDs)
|
||||
}
|
||||
|
||||
// Noise sessions & verification transport
|
||||
var myNoiseStaticKey = Data(repeating: 0x42, count: 32)
|
||||
var establishedNoiseSessions: Set<PeerID> = []
|
||||
@ -119,8 +126,10 @@ private final class MockChatVerificationContext: ChatVerificationContext {
|
||||
privateMediaAuthenticatedPeers.append(peerID)
|
||||
}
|
||||
|
||||
func retrySecurePrivateMessagesAfterAuthentication(for peerIDAliases: [PeerID]) {
|
||||
var securePrivateMessageRetryResult: Set<String> = []
|
||||
func retrySecurePrivateMessagesAfterAuthentication(for peerIDAliases: [PeerID]) -> Set<String> {
|
||||
securePrivateMessageRetryAliases.append(peerIDAliases)
|
||||
return securePrivateMessageRetryResult
|
||||
}
|
||||
|
||||
func sendVerifyChallenge(to peerID: PeerID, noiseKeyHex: String, nonceA: Data) {
|
||||
@ -300,6 +309,39 @@ 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
|
||||
context.securePrivateMessageRetryResult = ["retried-1"]
|
||||
|
||||
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 flush must skip exactly what the retry reported transmitting.
|
||||
// Re-deriving that set from `secureTransmissions` would also skip mail
|
||||
// the retry passed over for want of a live secure session, stranding it.
|
||||
#expect(
|
||||
context.flushedSkippingMessageIDs == [["retried-1"]],
|
||||
"the flush did not skip exactly the set the retry transmitted"
|
||||
)
|
||||
}
|
||||
|
||||
@Test @MainActor
|
||||
func handleVerifyChallengePayload_postsMutualVerificationToastOncePerMinute() async {
|
||||
let context = MockChatVerificationContext()
|
||||
|
||||
@ -513,7 +513,7 @@ struct ChatViewModelDeliveryStatusTests {
|
||||
transport.connectedPeers.insert(peerID)
|
||||
transport.securePeers = [peerID]
|
||||
viewModel.messageRouter.flushOutbox(for: peerID)
|
||||
viewModel.messageRouter.retrySecurePrivateMessagesAfterAuthentication(for: [peerID])
|
||||
_ = viewModel.messageRouter.retrySecurePrivateMessagesAfterAuthentication(for: [peerID])
|
||||
#expect(transport.sentPrivateMessages.isEmpty)
|
||||
|
||||
// The clear reached the durable snapshot: the next relaunch restores
|
||||
|
||||
@ -119,11 +119,11 @@ struct MessageRouterTests {
|
||||
|
||||
// A newly authenticated/replacement session retries the retained
|
||||
// message instead of losing the first ciphertext to a stale session.
|
||||
router.retrySecurePrivateMessagesAfterAuthentication(for: [peerID])
|
||||
_ = router.retrySecurePrivateMessagesAfterAuthentication(for: [peerID])
|
||||
#expect(transport.sentPrivateMessages.count == 2)
|
||||
|
||||
router.markDelivered("m7")
|
||||
router.retrySecurePrivateMessagesAfterAuthentication(for: [peerID])
|
||||
_ = router.retrySecurePrivateMessagesAfterAuthentication(for: [peerID])
|
||||
#expect(transport.sentPrivateMessages.count == 2)
|
||||
}
|
||||
|
||||
@ -138,7 +138,7 @@ struct MessageRouterTests {
|
||||
let router = MessageRouter(transports: [transport])
|
||||
router.sendPrivate("Hello", to: stablePeerID, recipientNickname: "Peer", messageID: "alias-retry")
|
||||
|
||||
router.retrySecurePrivateMessagesAfterAuthentication(for: [shortPeerID, stablePeerID, stablePeerID])
|
||||
_ = router.retrySecurePrivateMessagesAfterAuthentication(for: [shortPeerID, stablePeerID, stablePeerID])
|
||||
|
||||
#expect(transport.sentPrivateMessages.map(\.messageID) == ["alias-retry", "alias-retry"])
|
||||
#expect(transport.sentPrivateMessages.allSatisfy { $0.peerID == stablePeerID })
|
||||
@ -161,12 +161,388 @@ struct MessageRouterTests {
|
||||
router.sendPrivate("Newer", to: shortPeerID, recipientNickname: "Peer", messageID: "fifo-new")
|
||||
transport.resetRecordings()
|
||||
|
||||
router.retrySecurePrivateMessagesAfterAuthentication(for: [shortPeerID, stablePeerID])
|
||||
_ = router.retrySecurePrivateMessagesAfterAuthentication(for: [shortPeerID, stablePeerID])
|
||||
|
||||
#expect(transport.sentPrivateMessages.map(\.messageID) == ["fifo-old", "fifo-new"])
|
||||
#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], skippingMessageIDs: [])
|
||||
|
||||
#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], skippingMessageIDs: [])
|
||||
|
||||
#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], skippingMessageIDs: [])
|
||||
#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_skippingRetriedIDs_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], skippingMessageIDs: ["sec-1"])
|
||||
|
||||
#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_skippingRetriedIDs_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],
|
||||
skippingMessageIDs: ["twin-1"]
|
||||
)
|
||||
|
||||
#expect(
|
||||
transport.sentPrivateMessages.isEmpty,
|
||||
"the untransmitted twin was sent even though the retry already covered this ID"
|
||||
)
|
||||
}
|
||||
|
||||
/// A message ID must be claimed only once an alias actually put it on a
|
||||
/// transport. The same ID can sit under both the ephemeral and the stable
|
||||
/// key; if the first alias visited has no transport at all, its flush is a
|
||||
/// no-op, and claiming the ID there would suppress the twin under the
|
||||
/// alias that *can* deliver — dropping the message rather than deduping it.
|
||||
@Test @MainActor
|
||||
func mergedFlush_deadAliasDoesNotSuppressTheDeliverableTwin() async {
|
||||
let deadPeerID = PeerID(str: "0000000000000028")
|
||||
let livePeerID = PeerID(hexData: Data(repeating: 0x28, count: 32))
|
||||
let transport = MockTransport()
|
||||
let router = MessageRouter(transports: [transport])
|
||||
|
||||
// Queue the same message ID under both aliases while neither is
|
||||
// reachable, so nothing is sent yet.
|
||||
transport.connectedPeers = []
|
||||
transport.securePeers = []
|
||||
router.sendPrivate("Twin", to: deadPeerID, recipientNickname: "Peer", messageID: "twin-2")
|
||||
router.sendPrivate("Twin", to: livePeerID, recipientNickname: "Peer", messageID: "twin-2")
|
||||
|
||||
// Only the stable alias comes up. The ephemeral one stays dark, so its
|
||||
// flush sends nothing.
|
||||
transport.connectedPeers = [livePeerID]
|
||||
transport.securePeers = [livePeerID]
|
||||
transport.resetRecordings()
|
||||
|
||||
router.flushOutbox(forAliases: [deadPeerID, livePeerID], skippingMessageIDs: [])
|
||||
|
||||
#expect(
|
||||
transport.sentPrivateMessages.map(\.messageID) == ["twin-2"],
|
||||
"the dead alias claimed the ID and suppressed the twin that could deliver"
|
||||
)
|
||||
}
|
||||
|
||||
/// The skip set must be what the retry *transmitted*, not every entry in
|
||||
/// `secureTransmissions` under these aliases. Those differ whenever an
|
||||
/// alias is connected but has no live secure session: the retry abandons
|
||||
/// that alias wholesale, so its messages are in `secureTransmissions`
|
||||
/// having never been sent. Deriving the skip set from that map left them
|
||||
/// neither retried nor flushed — stranded until the next reconnect or the
|
||||
/// 24h TTL, which is the exact delay this flush exists to remove.
|
||||
@Test @MainActor
|
||||
func mergedFlush_deliversMailTheRetrySkippedForWantOfASecureSession() async {
|
||||
let peerID = PeerID(str: "0000000000000027")
|
||||
let transport = MockTransport()
|
||||
transport.connectedPeers = [peerID]
|
||||
transport.securePeers = [peerID]
|
||||
let router = MessageRouter(transports: [transport])
|
||||
|
||||
// Transmitted securely, so it lands in `secureTransmissions` and stays
|
||||
// queued pending an ack.
|
||||
router.sendPrivate("Stranded", to: peerID, recipientNickname: "Peer", messageID: "stranded-1")
|
||||
router.flushOutbox(for: peerID)
|
||||
#expect(transport.sentPrivateMessages.allSatisfy { $0.messageID == "stranded-1" })
|
||||
|
||||
// The link comes back without a secure session: still connected, but
|
||||
// `canDeliverSecurely` is false, so the retry abandons this alias.
|
||||
transport.securePeers = []
|
||||
transport.resetRecordings()
|
||||
|
||||
let retried = router.retrySecurePrivateMessagesAfterAuthentication(for: [peerID])
|
||||
|
||||
#expect(
|
||||
retried.isEmpty,
|
||||
"the retry reported transmitting a message it never sent"
|
||||
)
|
||||
#expect(
|
||||
transport.sentPrivateMessages.isEmpty,
|
||||
"the retry sent over a link that cannot deliver securely"
|
||||
)
|
||||
|
||||
router.flushOutbox(forAliases: [peerID], skippingMessageIDs: retried)
|
||||
|
||||
#expect(
|
||||
transport.sentPrivateMessages.map(\.messageID) == ["stranded-1"],
|
||||
"the flush skipped a message the retry never transmitted, stranding it until TTL"
|
||||
)
|
||||
}
|
||||
|
||||
/// The returned set is the entire contract with the flush that follows, so
|
||||
/// it has to be pinned on the path where the retry actually sends. Every
|
||||
/// other test either ignores the return or exercises the empty case, which
|
||||
/// left the line that records a transmission unguarded: deleting it kept
|
||||
/// the suite green while reopening the double-send it exists to prevent.
|
||||
@Test @MainActor
|
||||
func authenticationRetry_reportsExactlyTheIDsItPutOnTheAir() async {
|
||||
let peerID = PeerID(str: "0000000000000029")
|
||||
let transport = MockTransport()
|
||||
let router = MessageRouter(transports: [transport])
|
||||
|
||||
// Composed while the peer was offline: queued, never transmitted, so
|
||||
// never in `secureTransmissions` and never this pass's business. It is
|
||||
// here so the assertion below pins the set exactly rather than merely
|
||||
// proving it is non-empty — an over-inclusive bug would name this one.
|
||||
transport.connectedPeers = []
|
||||
transport.securePeers = []
|
||||
router.sendPrivate("Offline", to: peerID, recipientNickname: "Peer", messageID: "off-1")
|
||||
|
||||
// Transmitted securely, so it lands in `secureTransmissions` and stays
|
||||
// queued pending an ack — exactly what the retry pass re-sends.
|
||||
transport.connectedPeers = [peerID]
|
||||
transport.securePeers = [peerID]
|
||||
router.sendPrivate("On the air", to: peerID, recipientNickname: "Peer", messageID: "air-1")
|
||||
transport.resetRecordings()
|
||||
|
||||
let retried = router.retrySecurePrivateMessagesAfterAuthentication(for: [peerID])
|
||||
|
||||
#expect(
|
||||
transport.sentPrivateMessages.map(\.messageID) == ["air-1"],
|
||||
"the retry sent mail that was never securely transmitted"
|
||||
)
|
||||
#expect(
|
||||
retried == ["air-1"],
|
||||
"the retry must report exactly what it put on the air: under-reporting makes the flush send it again and burn a second attempt against the cap, over-reporting strands the message it wrongly named"
|
||||
)
|
||||
}
|
||||
|
||||
/// The retry must claim a message ID only past every check that can drop
|
||||
/// the candidate rather than send it. Claiming in the guard chain let an
|
||||
/// expired copy take the ID, drop itself, and suppress the live twin under
|
||||
/// the other alias — which then went unsent for the whole pass. The flush
|
||||
/// afterwards still recovered it, but the drop had already reported the
|
||||
/// message failed to the UI while it was in fact about to deliver.
|
||||
@Test @MainActor
|
||||
func authenticationRetry_anExpiredCandidateDoesNotSuppressTheLiveTwin() async {
|
||||
let shortPeerID = PeerID(str: "0000000000000030")
|
||||
let stablePeerID = PeerID(hexData: Data(repeating: 0x30, count: 32))
|
||||
let transport = MockTransport()
|
||||
transport.connectedPeers = [shortPeerID, stablePeerID]
|
||||
transport.securePeers = [shortPeerID, stablePeerID]
|
||||
let clock = MutableTestClock()
|
||||
let router = MessageRouter(transports: [transport], now: { clock.now })
|
||||
|
||||
// The ephemeral copy ages past the TTL. The stable copy of the same ID
|
||||
// is composed fresh, so only one of the twins is droppable.
|
||||
router.sendPrivate("Twin", to: shortPeerID, recipientNickname: "Peer", messageID: "ttl-twin")
|
||||
clock.now = clock.now.addingTimeInterval(25 * 60 * 60)
|
||||
router.sendPrivate("Twin", to: stablePeerID, recipientNickname: "Peer", messageID: "ttl-twin")
|
||||
transport.resetRecordings()
|
||||
|
||||
let retried = router.retrySecurePrivateMessagesAfterAuthentication(for: [shortPeerID, stablePeerID])
|
||||
|
||||
#expect(
|
||||
transport.sentPrivateMessages.map(\.messageID) == ["ttl-twin"],
|
||||
"the expired copy claimed the ID and suppressed the twin that was still deliverable"
|
||||
)
|
||||
#expect(
|
||||
transport.sentPrivateMessages.map(\.peerID) == [stablePeerID],
|
||||
"the retry sent under the expired alias rather than the live one"
|
||||
)
|
||||
#expect(retried == ["ttl-twin"])
|
||||
}
|
||||
|
||||
/// 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], skippingMessageIDs: [])
|
||||
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")
|
||||
@ -182,7 +558,7 @@ struct MessageRouterTests {
|
||||
// the session becomes secure, the router's targeted auth retry must
|
||||
// stay silent instead of producing a second copy.
|
||||
transport.securePeers = [peerID]
|
||||
router.retrySecurePrivateMessagesAfterAuthentication(for: [peerID])
|
||||
_ = router.retrySecurePrivateMessagesAfterAuthentication(for: [peerID])
|
||||
#expect(transport.sentPrivateMessages.count == 1)
|
||||
|
||||
router.markDelivered("normal-handshake")
|
||||
@ -231,10 +607,10 @@ struct MessageRouterTests {
|
||||
transport.resetRecordings()
|
||||
transport.securePeers = [securePeer, pendingPeer]
|
||||
|
||||
router.retrySecurePrivateMessagesAfterAuthentication(for: [pendingPeer])
|
||||
_ = router.retrySecurePrivateMessagesAfterAuthentication(for: [pendingPeer])
|
||||
#expect(transport.sentPrivateMessages.isEmpty)
|
||||
|
||||
router.retrySecurePrivateMessagesAfterAuthentication(for: [securePeer])
|
||||
_ = router.retrySecurePrivateMessagesAfterAuthentication(for: [securePeer])
|
||||
#expect(transport.sentPrivateMessages.count == 2)
|
||||
#expect(Set(transport.sentPrivateMessages.map(\.messageID)) == [promotedID, clearedID])
|
||||
#expect(transport.sentPrivateMessages.allSatisfy { $0.peerID == securePeer })
|
||||
@ -259,7 +635,7 @@ struct MessageRouterTests {
|
||||
#expect(transport.sentPrivateMessages.count == 2)
|
||||
|
||||
transport.securePeers = [peerID]
|
||||
router.retrySecurePrivateMessagesAfterAuthentication(for: [peerID])
|
||||
_ = router.retrySecurePrivateMessagesAfterAuthentication(for: [peerID])
|
||||
#expect(transport.sentPrivateMessages.count == 2)
|
||||
|
||||
router.markDelivered("session-lost")
|
||||
@ -618,7 +994,7 @@ struct MessageRouterTests {
|
||||
|
||||
router.sendPrivate("Hello", to: peerID, recipientNickname: "Peer", messageID: "secure-retry")
|
||||
for _ in 0..<10 {
|
||||
router.retrySecurePrivateMessagesAfterAuthentication(for: [peerID])
|
||||
_ = router.retrySecurePrivateMessagesAfterAuthentication(for: [peerID])
|
||||
}
|
||||
|
||||
#expect(dropped == ["secure-retry"])
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user