mirror of
https://github.com/permissionlesstech/bitchat.git
synced 2026-07-25 01:05:19 +00:00
Surface router drops as failed status; bound relay pending subscriptions; jitter backoff
MessageRouter outbox drops (attempt cap, TTL expiry in flush and cleanup, per-peer overflow eviction) now invoke onMessageDropped, wired to mark the message .failed in the ConversationStore - guarded so a late failure never downgrades an already delivered/read status. NostrRelayManager pending subscriptions gain a per-relay cap (64, oldest-by-sequence eviction; durable intent still replays from subscriptionRequestState) and a 10-minute age sweep on the existing connect path. Reconnect backoff gets injectable +/-20% jitter so recovering relays don't thundering-herd. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
@@ -66,6 +66,9 @@ struct NostrRelayManagerDependencies {
|
|||||||
var makeSession: () -> NostrRelaySessionProtocol
|
var makeSession: () -> NostrRelaySessionProtocol
|
||||||
var scheduleAfter: @Sendable (TimeInterval, @escaping @Sendable () -> Void) -> Void
|
var scheduleAfter: @Sendable (TimeInterval, @escaping @Sendable () -> Void) -> Void
|
||||||
var now: () -> Date
|
var now: () -> Date
|
||||||
|
/// Uniform random value in [0, 1) used to jitter reconnect backoff.
|
||||||
|
/// Injectable so tests can pin or sweep the jitter deterministically.
|
||||||
|
var jitterUnit: () -> Double
|
||||||
}
|
}
|
||||||
|
|
||||||
private extension NostrRelayManagerDependencies {
|
private extension NostrRelayManagerDependencies {
|
||||||
@@ -93,7 +96,8 @@ private extension NostrRelayManagerDependencies {
|
|||||||
scheduleAfter: { delay, action in
|
scheduleAfter: { delay, action in
|
||||||
DispatchQueue.main.asyncAfter(deadline: .now() + delay, execute: action)
|
DispatchQueue.main.asyncAfter(deadline: .now() + delay, execute: action)
|
||||||
},
|
},
|
||||||
now: Date.init
|
now: Date.init,
|
||||||
|
jitterUnit: { Double.random(in: 0..<1) }
|
||||||
)
|
)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@@ -140,7 +144,16 @@ final class NostrRelayManager: ObservableObject {
|
|||||||
private var hasLocationPermission: Bool = false
|
private var hasLocationPermission: Bool = false
|
||||||
private var connections: [String: NostrRelayConnectionProtocol] = [:]
|
private var connections: [String: NostrRelayConnectionProtocol] = [:]
|
||||||
private var subscriptions: [String: Set<String>] = [:] // relay URL -> active subscription IDs
|
private var subscriptions: [String: Set<String>] = [:] // relay URL -> active subscription IDs
|
||||||
private var pendingSubscriptions: [String: [String: String]] = [:] // relay URL -> (subscription id -> encoded REQ JSON)
|
// Not-yet-flushed REQs per relay, bounded by a per-relay cap (oldest by
|
||||||
|
// insertion order evicted) and an age sweep on connect attempts. Dicts are
|
||||||
|
// unordered, so each entry carries an insertion sequence and queue time.
|
||||||
|
private struct PendingSubscription {
|
||||||
|
let messageString: String // encoded REQ JSON
|
||||||
|
let queuedAt: Date
|
||||||
|
let sequence: UInt64
|
||||||
|
}
|
||||||
|
private var pendingSubscriptions: [String: [String: PendingSubscription]] = [:] // relay URL -> (subscription id -> pending REQ)
|
||||||
|
private var pendingSubscriptionSequence: UInt64 = 0
|
||||||
private var messageHandlers: [String: (NostrEvent) -> Void] = [:]
|
private var messageHandlers: [String: (NostrEvent) -> Void] = [:]
|
||||||
private struct InboundEventKey: Hashable {
|
private struct InboundEventKey: Hashable {
|
||||||
let subscriptionID: String
|
let subscriptionID: String
|
||||||
@@ -432,9 +445,7 @@ final class NostrRelayManager: ObservableObject {
|
|||||||
existingSet.insert(url)
|
existingSet.insert(url)
|
||||||
}
|
}
|
||||||
for url in urls {
|
for url in urls {
|
||||||
var map = self.pendingSubscriptions[url] ?? [:]
|
queuePendingSubscription(id: id, messageString: messageString, for: url)
|
||||||
map[id] = messageString
|
|
||||||
self.pendingSubscriptions[url] = map
|
|
||||||
}
|
}
|
||||||
// Initialize EOSE tracking if requested
|
// Initialize EOSE tracking if requested
|
||||||
if let onEOSE = onEOSE {
|
if let onEOSE = onEOSE {
|
||||||
@@ -550,6 +561,7 @@ final class NostrRelayManager: ObservableObject {
|
|||||||
|
|
||||||
private func connectToRelays(_ relayUrls: [String], shouldLog: Bool = false) {
|
private func connectToRelays(_ relayUrls: [String], shouldLog: Bool = false) {
|
||||||
guard dependencies.activationAllowed() else { return }
|
guard dependencies.activationAllowed() else { return }
|
||||||
|
sweepStalePendingSubscriptions()
|
||||||
let targets = allowedRelayList(from: relayUrls).filter {
|
let targets = allowedRelayList(from: relayUrls).filter {
|
||||||
connections[$0] == nil && !isPermanentlyFailed($0)
|
connections[$0] == nil && !isPermanentlyFailed($0)
|
||||||
}
|
}
|
||||||
@@ -627,11 +639,53 @@ final class NostrRelayManager: ObservableObject {
|
|||||||
private func subscriptionStateExists(id: String, requestState: SubscriptionRequestState) -> Bool {
|
private func subscriptionStateExists(id: String, requestState: SubscriptionRequestState) -> Bool {
|
||||||
guard !requestState.relayURLs.isEmpty else { return true }
|
guard !requestState.relayURLs.isEmpty else { return true }
|
||||||
return requestState.relayURLs.allSatisfy { url in
|
return requestState.relayURLs.allSatisfy { url in
|
||||||
pendingSubscriptions[url]?[id] == requestState.messageString ||
|
pendingSubscriptions[url]?[id]?.messageString == requestState.messageString ||
|
||||||
subscriptions[url]?.contains(id) == true
|
subscriptions[url]?.contains(id) == true
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
private func queuePendingSubscription(id: String, messageString: String, for url: String) {
|
||||||
|
var map = pendingSubscriptions[url] ?? [:]
|
||||||
|
pendingSubscriptionSequence &+= 1
|
||||||
|
map[id] = PendingSubscription(
|
||||||
|
messageString: messageString,
|
||||||
|
queuedAt: dependencies.now(),
|
||||||
|
sequence: pendingSubscriptionSequence
|
||||||
|
)
|
||||||
|
// Bound per-relay pending REQs; evict oldest by insertion order. The
|
||||||
|
// durable intent stays in subscriptionRequestState, so an evicted REQ
|
||||||
|
// is still replayed if its subscription is active when the relay
|
||||||
|
// (re)connects.
|
||||||
|
while map.count > TransportConfig.nostrPendingSubscriptionsPerRelayCap,
|
||||||
|
let oldest = map.min(by: { $0.value.sequence < $1.value.sequence }) {
|
||||||
|
map.removeValue(forKey: oldest.key)
|
||||||
|
SecureLogger.warning(
|
||||||
|
"📋 Pending subscription overflow for \(url) - evicted oldest id=\(oldest.key)",
|
||||||
|
category: .session
|
||||||
|
)
|
||||||
|
}
|
||||||
|
pendingSubscriptions[url] = map
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Drop pending REQs older than the TTL. Runs on connect attempts (the
|
||||||
|
/// natural maintenance path: connect/ensureConnections/reconnects all
|
||||||
|
/// funnel through connectToRelays) so stale entries for relays that never
|
||||||
|
/// come up cannot accumulate without bound.
|
||||||
|
private func sweepStalePendingSubscriptions() {
|
||||||
|
let now = dependencies.now()
|
||||||
|
for (url, map) in pendingSubscriptions {
|
||||||
|
let fresh = map.filter {
|
||||||
|
now.timeIntervalSince($0.value.queuedAt) <= TransportConfig.nostrPendingSubscriptionTTLSeconds
|
||||||
|
}
|
||||||
|
guard fresh.count != map.count else { continue }
|
||||||
|
SecureLogger.debug(
|
||||||
|
"📋 Swept \(map.count - fresh.count) stale pending subscription(s) for \(url)",
|
||||||
|
category: .session
|
||||||
|
)
|
||||||
|
pendingSubscriptions[url] = fresh.isEmpty ? nil : fresh
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
private func startEOSETracking(id: String, relayURLs: Set<String>, callback: @escaping () -> Void) {
|
private func startEOSETracking(id: String, relayURLs: Set<String>, callback: @escaping () -> Void) {
|
||||||
eoseTrackerEpoch += 1
|
eoseTrackerEpoch += 1
|
||||||
let epoch = eoseTrackerEpoch
|
let epoch = eoseTrackerEpoch
|
||||||
@@ -770,7 +824,7 @@ final class NostrRelayManager: ObservableObject {
|
|||||||
/// active subscription targeting this relay must be re-sent.
|
/// active subscription targeting this relay must be re-sent.
|
||||||
private func flushPendingSubscriptions(for relayUrl: String) {
|
private func flushPendingSubscriptions(for relayUrl: String) {
|
||||||
guard let connection = connections[relayUrl] else { return }
|
guard let connection = connections[relayUrl] else { return }
|
||||||
var toSend = pendingSubscriptions[relayUrl] ?? [:]
|
var toSend = (pendingSubscriptions[relayUrl] ?? [:]).mapValues(\.messageString)
|
||||||
for (id, state) in subscriptionRequestState where state.relayURLs.contains(relayUrl) && toSend[id] == nil {
|
for (id, state) in subscriptionRequestState where state.relayURLs.contains(relayUrl) && toSend[id] == nil {
|
||||||
toSend[id] = state.messageString
|
toSend[id] = state.messageString
|
||||||
}
|
}
|
||||||
@@ -987,11 +1041,15 @@ final class NostrRelayManager: ObservableObject {
|
|||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
|
||||||
// Calculate backoff interval
|
// Calculate backoff interval with ±jitterRatio random jitter so relays
|
||||||
let backoffInterval = min(
|
// that dropped together don't all reconnect at the same instant.
|
||||||
|
let baseBackoffInterval = min(
|
||||||
initialBackoffInterval * pow(backoffMultiplier, Double(relays[index].reconnectAttempts - 1)),
|
initialBackoffInterval * pow(backoffMultiplier, Double(relays[index].reconnectAttempts - 1)),
|
||||||
maxBackoffInterval
|
maxBackoffInterval
|
||||||
)
|
)
|
||||||
|
let jitterRatio = TransportConfig.nostrRelayBackoffJitterRatio
|
||||||
|
let jitterFactor = 1.0 + (dependencies.jitterUnit() * 2.0 - 1.0) * jitterRatio
|
||||||
|
let backoffInterval = baseBackoffInterval * jitterFactor
|
||||||
|
|
||||||
let nextReconnectTime = dependencies.now().addingTimeInterval(backoffInterval)
|
let nextReconnectTime = dependencies.now().addingTimeInterval(backoffInterval)
|
||||||
relays[index].nextReconnectTime = nextReconnectTime
|
relays[index].nextReconnectTime = nextReconnectTime
|
||||||
@@ -1054,6 +1112,11 @@ final class NostrRelayManager: ObservableObject {
|
|||||||
pendingSubscriptions[relayUrl]?.count ?? 0
|
pendingSubscriptions[relayUrl]?.count ?? 0
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func debugPendingSubscriptionIDs(for relayUrl: String) -> Set<String> {
|
||||||
|
guard let map = pendingSubscriptions[relayUrl] else { return [] }
|
||||||
|
return Set(map.keys)
|
||||||
|
}
|
||||||
|
|
||||||
var debugDuplicateInboundEventDropCount: Int {
|
var debugDuplicateInboundEventDropCount: Int {
|
||||||
duplicateInboundEventDropCount
|
duplicateInboundEventDropCount
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -6,6 +6,13 @@ import Foundation
|
|||||||
@MainActor
|
@MainActor
|
||||||
final class MessageRouter {
|
final class MessageRouter {
|
||||||
private let transports: [Transport]
|
private let transports: [Transport]
|
||||||
|
private let now: () -> Date
|
||||||
|
|
||||||
|
/// Invoked whenever a retained private message is dropped without a
|
||||||
|
/// delivery ack (attempt cap, TTL expiry, or per-peer overflow eviction)
|
||||||
|
/// so the UI can surface the failure instead of leaving the message in a
|
||||||
|
/// stale "sending/sent" state forever.
|
||||||
|
var onMessageDropped: ((_ messageID: String, _ peerID: PeerID) -> Void)?
|
||||||
|
|
||||||
// Outbox entry with timestamp for TTL-based eviction
|
// Outbox entry with timestamp for TTL-based eviction
|
||||||
private struct QueuedMessage {
|
private struct QueuedMessage {
|
||||||
@@ -25,8 +32,9 @@ final class MessageRouter {
|
|||||||
// get a delivery ack (e.g. peer on an old client that doesn't ack).
|
// get a delivery ack (e.g. peer on an old client that doesn't ack).
|
||||||
private static let maxSendAttempts = 8
|
private static let maxSendAttempts = 8
|
||||||
|
|
||||||
init(transports: [Transport]) {
|
init(transports: [Transport], now: @escaping () -> Date = Date.init) {
|
||||||
self.transports = transports
|
self.transports = transports
|
||||||
|
self.now = now
|
||||||
|
|
||||||
// Observe favorites changes to learn Nostr mapping and flush queued messages
|
// Observe favorites changes to learn Nostr mapping and flush queued messages
|
||||||
NotificationCenter.default.addObserver(
|
NotificationCenter.default.addObserver(
|
||||||
@@ -72,7 +80,7 @@ final class MessageRouter {
|
|||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
|
||||||
let message = QueuedMessage(content: content, nickname: recipientNickname, messageID: messageID, timestamp: Date(), sendAttempts: 1)
|
let message = QueuedMessage(content: content, nickname: recipientNickname, messageID: messageID, timestamp: now(), sendAttempts: 1)
|
||||||
if let transport = reachableTransport(for: peerID) {
|
if let transport = reachableTransport(for: peerID) {
|
||||||
// Reachability without a connection is a freshness heuristic (e.g.
|
// Reachability without a connection is a freshness heuristic (e.g.
|
||||||
// the mesh retention window), so the send can silently go nowhere.
|
// the mesh retention window), so the send can silently go nowhere.
|
||||||
@@ -108,6 +116,7 @@ final class MessageRouter {
|
|||||||
if queue.count > Self.maxMessagesPerPeer {
|
if queue.count > Self.maxMessagesPerPeer {
|
||||||
let evicted = queue.removeFirst()
|
let evicted = queue.removeFirst()
|
||||||
SecureLogger.warning("📤 Outbox overflow for \(peerID.id.prefix(8))… - evicted oldest message: \(evicted.messageID.prefix(8))…", category: .session)
|
SecureLogger.warning("📤 Outbox overflow for \(peerID.id.prefix(8))… - evicted oldest message: \(evicted.messageID.prefix(8))…", category: .session)
|
||||||
|
onMessageDropped?(evicted.messageID, peerID)
|
||||||
}
|
}
|
||||||
outbox[peerID] = queue
|
outbox[peerID] = queue
|
||||||
}
|
}
|
||||||
@@ -142,13 +151,14 @@ final class MessageRouter {
|
|||||||
guard let queued = outbox[peerID], !queued.isEmpty else { return }
|
guard let queued = outbox[peerID], !queued.isEmpty else { return }
|
||||||
SecureLogger.debug("Flushing outbox for \(peerID.id.prefix(8))… count=\(queued.count)", category: .session)
|
SecureLogger.debug("Flushing outbox for \(peerID.id.prefix(8))… count=\(queued.count)", category: .session)
|
||||||
|
|
||||||
let now = Date()
|
let now = now()
|
||||||
var remaining: [QueuedMessage] = []
|
var remaining: [QueuedMessage] = []
|
||||||
|
|
||||||
for message in queued {
|
for message in queued {
|
||||||
// Skip expired messages (TTL exceeded)
|
// Skip expired messages (TTL exceeded)
|
||||||
if now.timeIntervalSince(message.timestamp) > Self.messageTTLSeconds {
|
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)
|
SecureLogger.debug("⏰ Expired queued message for \(peerID.id.prefix(8))… id=\(message.messageID.prefix(8))… (age: \(Int(now.timeIntervalSince(message.timestamp)))s)", category: .session)
|
||||||
|
onMessageDropped?(message.messageID, peerID)
|
||||||
continue
|
continue
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -161,6 +171,7 @@ final class MessageRouter {
|
|||||||
// bounded by attempt count for peers that never ack.
|
// bounded by attempt count for peers that never ack.
|
||||||
guard message.sendAttempts < Self.maxSendAttempts else {
|
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)
|
SecureLogger.warning("📤 Dropping unacked PM for \(peerID.id.prefix(8))… id=\(message.messageID.prefix(8))… after \(message.sendAttempts) attempts", category: .session)
|
||||||
|
onMessageDropped?(message.messageID, peerID)
|
||||||
continue
|
continue
|
||||||
}
|
}
|
||||||
SecureLogger.debug("Outbox -> \(type(of: transport)) (reachable) for \(peerID.id.prefix(8))… id=\(message.messageID.prefix(8))…", category: .session)
|
SecureLogger.debug("Outbox -> \(type(of: transport)) (reachable) for \(peerID.id.prefix(8))… id=\(message.messageID.prefix(8))…", category: .session)
|
||||||
@@ -186,12 +197,21 @@ final class MessageRouter {
|
|||||||
|
|
||||||
/// Periodically clean up expired messages from all outboxes
|
/// Periodically clean up expired messages from all outboxes
|
||||||
func cleanupExpiredMessages() {
|
func cleanupExpiredMessages() {
|
||||||
let now = Date()
|
let now = now()
|
||||||
for peerID in Array(outbox.keys) {
|
for peerID in Array(outbox.keys) {
|
||||||
outbox[peerID]?.removeAll { now.timeIntervalSince($0.timestamp) > Self.messageTTLSeconds }
|
var expiredMessageIDs: [String] = []
|
||||||
|
outbox[peerID]?.removeAll { message in
|
||||||
|
guard now.timeIntervalSince(message.timestamp) > Self.messageTTLSeconds else { return false }
|
||||||
|
expiredMessageIDs.append(message.messageID)
|
||||||
|
return true
|
||||||
|
}
|
||||||
if outbox[peerID]?.isEmpty == true {
|
if outbox[peerID]?.isEmpty == true {
|
||||||
outbox.removeValue(forKey: peerID)
|
outbox.removeValue(forKey: peerID)
|
||||||
}
|
}
|
||||||
|
for messageID in expiredMessageIDs {
|
||||||
|
SecureLogger.debug("⏰ Expired queued message for \(peerID.id.prefix(8))… id=\(messageID.prefix(8))…", category: .session)
|
||||||
|
onMessageDropped?(messageID, peerID)
|
||||||
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -148,11 +148,19 @@ enum TransportConfig {
|
|||||||
static let nostrRelayMaxBackoffSeconds: TimeInterval = 300.0
|
static let nostrRelayMaxBackoffSeconds: TimeInterval = 300.0
|
||||||
static let nostrRelayBackoffMultiplier: Double = 2.0
|
static let nostrRelayBackoffMultiplier: Double = 2.0
|
||||||
static let nostrRelayMaxReconnectAttempts: Int = 10
|
static let nostrRelayMaxReconnectAttempts: Int = 10
|
||||||
|
// Reconnect delays get ±20% random jitter so relays that dropped together
|
||||||
|
// (e.g. a network blip) don't thundering-herd the same reconnect instant.
|
||||||
|
static let nostrRelayBackoffJitterRatio: Double = 0.2
|
||||||
static let nostrRelayDefaultFetchLimit: Int = 100
|
static let nostrRelayDefaultFetchLimit: Int = 100
|
||||||
// How many consecutive Tor-readiness waits (each bounded by TorManager's
|
// How many consecutive Tor-readiness waits (each bounded by TorManager's
|
||||||
// bootstrap deadline) to attempt before unblocking pending EOSE callers.
|
// bootstrap deadline) to attempt before unblocking pending EOSE callers.
|
||||||
static let nostrTorReadyMaxWaitAttempts: Int = 3
|
static let nostrTorReadyMaxWaitAttempts: Int = 3
|
||||||
static let nostrPendingSendQueueCap: Int = 200
|
static let nostrPendingSendQueueCap: Int = 200
|
||||||
|
// Pending (not-yet-flushed) REQs are bounded per relay: oldest-by-insertion
|
||||||
|
// eviction at the cap, plus an age sweep on connect attempts. Durable
|
||||||
|
// subscription intent survives in subscriptionRequestState either way.
|
||||||
|
static let nostrPendingSubscriptionsPerRelayCap: Int = 64
|
||||||
|
static let nostrPendingSubscriptionTTLSeconds: TimeInterval = 600.0
|
||||||
// Fallback deadline for treating a subscription's initial fetch as complete
|
// Fallback deadline for treating a subscription's initial fetch as complete
|
||||||
// when a relay never sends EOSE (generous to cover Tor circuit setup).
|
// when a relay never sends EOSE (generous to cover Tor circuit setup).
|
||||||
static let nostrSubscriptionEOSEFallbackSeconds: TimeInterval = 10.0
|
static let nostrSubscriptionEOSEFallbackSeconds: TimeInterval = 10.0
|
||||||
|
|||||||
@@ -78,6 +78,23 @@ private extension ChatViewModelBootstrapper {
|
|||||||
viewModel.privateChatManager.messageRouter = viewModel.messageRouter
|
viewModel.privateChatManager.messageRouter = viewModel.messageRouter
|
||||||
viewModel.privateChatManager.unifiedPeerService = viewModel.unifiedPeerService
|
viewModel.privateChatManager.unifiedPeerService = viewModel.unifiedPeerService
|
||||||
viewModel.unifiedPeerService.messageRouter = viewModel.messageRouter
|
viewModel.unifiedPeerService.messageRouter = viewModel.messageRouter
|
||||||
|
// Surface silent outbox drops (attempt cap, TTL expiry, overflow
|
||||||
|
// eviction) as a visible failure. The store's no-downgrade rule does
|
||||||
|
// not cover `.failed` over confirmed receipts, so guard here: a drop
|
||||||
|
// of an already-delivered/read message (e.g. a stale retained copy)
|
||||||
|
// must not downgrade its status.
|
||||||
|
viewModel.messageRouter.onMessageDropped = { [weak viewModel] messageID, _ in
|
||||||
|
guard let viewModel else { return }
|
||||||
|
switch viewModel.conversations.deliveryStatus(forMessageID: messageID) {
|
||||||
|
case .delivered, .read:
|
||||||
|
return
|
||||||
|
default:
|
||||||
|
viewModel.conversations.setDeliveryStatus(
|
||||||
|
.failed(reason: "Not delivered"),
|
||||||
|
forMessageID: messageID
|
||||||
|
)
|
||||||
|
}
|
||||||
|
}
|
||||||
viewModel.commandProcessor.contextProvider = viewModel
|
viewModel.commandProcessor.contextProvider = viewModel
|
||||||
viewModel.commandProcessor.meshService = viewModel.meshService
|
viewModel.commandProcessor.meshService = viewModel.meshService
|
||||||
viewModel.participantTracker.configure(context: viewModel)
|
viewModel.participantTracker.configure(context: viewModel)
|
||||||
|
|||||||
@@ -349,6 +349,80 @@ struct ChatViewModelDeliveryStatusTests {
|
|||||||
#expect(isRead(viewModel.privateChats[peerID]?.last?.deliveryStatus))
|
#expect(isRead(viewModel.privateChats[peerID]?.last?.deliveryStatus))
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// MARK: - MessageRouter Drop Wiring Tests
|
||||||
|
|
||||||
|
/// Drives a real outbox drop (per-peer overflow eviction with no
|
||||||
|
/// reachable transport) and proves the bootstrapper wiring marks the
|
||||||
|
/// dropped message `.failed` in the conversation store.
|
||||||
|
@Test @MainActor
|
||||||
|
func messageRouterDrop_marksMessageFailedInStore() async {
|
||||||
|
let (viewModel, transport) = makeTestableViewModel()
|
||||||
|
let peerID = PeerID(str: "0102030405060708")
|
||||||
|
let droppedID = "router-drop-0"
|
||||||
|
|
||||||
|
let message = BitchatMessage(
|
||||||
|
id: droppedID,
|
||||||
|
sender: viewModel.nickname,
|
||||||
|
content: "Will be dropped",
|
||||||
|
timestamp: Date(),
|
||||||
|
isRelay: false,
|
||||||
|
isPrivate: true,
|
||||||
|
recipientNickname: "Peer",
|
||||||
|
senderPeerID: transport.myPeerID,
|
||||||
|
deliveryStatus: .sent
|
||||||
|
)
|
||||||
|
viewModel.seedPrivateChat([message], for: peerID)
|
||||||
|
|
||||||
|
// No transport is reachable, so every send is queued; the 101st
|
||||||
|
// enqueue for this peer evicts the oldest queued message.
|
||||||
|
viewModel.messageRouter.sendPrivate("Will be dropped", to: peerID, recipientNickname: "Peer", messageID: droppedID)
|
||||||
|
for i in 1...100 {
|
||||||
|
viewModel.messageRouter.sendPrivate("Filler \(i)", to: peerID, recipientNickname: "Peer", messageID: "router-drop-\(i)")
|
||||||
|
}
|
||||||
|
|
||||||
|
let status = viewModel.conversations.deliveryStatus(forMessageID: droppedID)
|
||||||
|
#expect({
|
||||||
|
if case .failed = status { return true }
|
||||||
|
return false
|
||||||
|
}())
|
||||||
|
}
|
||||||
|
|
||||||
|
/// The store's no-downgrade rule does not cover `.failed` over confirmed
|
||||||
|
/// receipts, so the wiring guards it: a drop of an already-delivered
|
||||||
|
/// message must not downgrade its status.
|
||||||
|
@Test @MainActor
|
||||||
|
func messageRouterDrop_doesNotDowngradeDeliveredStatus() async {
|
||||||
|
let (viewModel, transport) = makeTestableViewModel()
|
||||||
|
let peerID = PeerID(str: "0102030405060708")
|
||||||
|
let droppedID = "router-drop-delivered"
|
||||||
|
|
||||||
|
let message = BitchatMessage(
|
||||||
|
id: droppedID,
|
||||||
|
sender: viewModel.nickname,
|
||||||
|
content: "Already delivered",
|
||||||
|
timestamp: Date(),
|
||||||
|
isRelay: false,
|
||||||
|
isPrivate: true,
|
||||||
|
recipientNickname: "Peer",
|
||||||
|
senderPeerID: transport.myPeerID,
|
||||||
|
deliveryStatus: .delivered(to: "Peer", at: Date())
|
||||||
|
)
|
||||||
|
viewModel.seedPrivateChat([message], for: peerID)
|
||||||
|
|
||||||
|
// Same eviction-driven drop as above, but the store already recorded
|
||||||
|
// a delivery confirmation for the message.
|
||||||
|
viewModel.messageRouter.sendPrivate("Already delivered", to: peerID, recipientNickname: "Peer", messageID: droppedID)
|
||||||
|
for i in 1...100 {
|
||||||
|
viewModel.messageRouter.sendPrivate("Filler \(i)", to: peerID, recipientNickname: "Peer", messageID: "router-keep-\(i)")
|
||||||
|
}
|
||||||
|
|
||||||
|
let status = viewModel.conversations.deliveryStatus(forMessageID: droppedID)
|
||||||
|
#expect({
|
||||||
|
if case .delivered = status { return true }
|
||||||
|
return false
|
||||||
|
}())
|
||||||
|
}
|
||||||
|
|
||||||
// MARK: - Status Rank Tests (for deduplication)
|
// MARK: - Status Rank Tests (for deduplication)
|
||||||
|
|
||||||
@Test @MainActor
|
@Test @MainActor
|
||||||
|
|||||||
@@ -110,6 +110,98 @@ struct MessageRouterTests {
|
|||||||
#expect(transport.sentPrivateMessages.count == 8)
|
#expect(transport.sentPrivateMessages.count == 8)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// MARK: - Drop visibility (onMessageDropped)
|
||||||
|
|
||||||
|
@Test @MainActor
|
||||||
|
func flushOutbox_attemptCapDropInvokesOnMessageDropped() async {
|
||||||
|
let peerID = PeerID(str: "0000000000000009")
|
||||||
|
let transport = MockTransport()
|
||||||
|
transport.reachablePeers.insert(peerID)
|
||||||
|
|
||||||
|
let router = MessageRouter(transports: [transport])
|
||||||
|
var dropped: [(messageID: String, peerID: PeerID)] = []
|
||||||
|
router.onMessageDropped = { dropped.append(($0, $1)) }
|
||||||
|
|
||||||
|
router.sendPrivate("Hello", to: peerID, recipientNickname: "Peer", messageID: "m9")
|
||||||
|
for _ in 0..<10 {
|
||||||
|
router.flushOutbox(for: peerID)
|
||||||
|
}
|
||||||
|
|
||||||
|
#expect(dropped.count == 1)
|
||||||
|
#expect(dropped.first?.messageID == "m9")
|
||||||
|
#expect(dropped.first?.peerID == peerID)
|
||||||
|
}
|
||||||
|
|
||||||
|
@Test @MainActor
|
||||||
|
func flushOutbox_ttlExpiryInvokesOnMessageDroppedAndDoesNotResend() async {
|
||||||
|
let peerID = PeerID(str: "000000000000000a")
|
||||||
|
let transport = MockTransport()
|
||||||
|
let clock = MutableTestClock()
|
||||||
|
|
||||||
|
let router = MessageRouter(transports: [transport], now: { clock.now })
|
||||||
|
var dropped: [String] = []
|
||||||
|
router.onMessageDropped = { messageID, _ in dropped.append(messageID) }
|
||||||
|
|
||||||
|
// No reachable transport: the message is queued, never sent.
|
||||||
|
router.sendPrivate("Hello", to: peerID, recipientNickname: "Peer", messageID: "m10")
|
||||||
|
#expect(transport.sentPrivateMessages.isEmpty)
|
||||||
|
|
||||||
|
// Past the 24h TTL the flush must drop it (visibly), not send it.
|
||||||
|
clock.now = clock.now.addingTimeInterval(24 * 60 * 60 + 1)
|
||||||
|
transport.reachablePeers.insert(peerID)
|
||||||
|
router.flushOutbox(for: peerID)
|
||||||
|
|
||||||
|
#expect(dropped == ["m10"])
|
||||||
|
#expect(transport.sentPrivateMessages.isEmpty)
|
||||||
|
|
||||||
|
// The drop is final: nothing is retained for later flushes.
|
||||||
|
router.flushOutbox(for: peerID)
|
||||||
|
#expect(dropped == ["m10"])
|
||||||
|
}
|
||||||
|
|
||||||
|
@Test @MainActor
|
||||||
|
func cleanupExpiredMessages_invokesOnMessageDroppedForExpiredOnly() async {
|
||||||
|
let peerID = PeerID(str: "000000000000000b")
|
||||||
|
let transport = MockTransport()
|
||||||
|
let clock = MutableTestClock()
|
||||||
|
|
||||||
|
let router = MessageRouter(transports: [transport], now: { clock.now })
|
||||||
|
var dropped: [String] = []
|
||||||
|
router.onMessageDropped = { messageID, _ in dropped.append(messageID) }
|
||||||
|
|
||||||
|
router.sendPrivate("Old", to: peerID, recipientNickname: "Peer", messageID: "m11-old")
|
||||||
|
clock.now = clock.now.addingTimeInterval(24 * 60 * 60 - 60)
|
||||||
|
router.sendPrivate("Fresh", to: peerID, recipientNickname: "Peer", messageID: "m11-fresh")
|
||||||
|
clock.now = clock.now.addingTimeInterval(120)
|
||||||
|
|
||||||
|
router.cleanupExpiredMessages()
|
||||||
|
|
||||||
|
#expect(dropped == ["m11-old"])
|
||||||
|
|
||||||
|
// The fresh message survived and still flushes once reachable.
|
||||||
|
transport.reachablePeers.insert(peerID)
|
||||||
|
router.flushOutbox(for: peerID)
|
||||||
|
#expect(transport.sentPrivateMessages.map(\.messageID) == ["m11-fresh"])
|
||||||
|
}
|
||||||
|
|
||||||
|
@Test @MainActor
|
||||||
|
func enqueue_perPeerOverflowEvictionInvokesOnMessageDropped() async {
|
||||||
|
let peerID = PeerID(str: "000000000000000c")
|
||||||
|
let transport = MockTransport()
|
||||||
|
|
||||||
|
let router = MessageRouter(transports: [transport])
|
||||||
|
var dropped: [String] = []
|
||||||
|
router.onMessageDropped = { messageID, _ in dropped.append(messageID) }
|
||||||
|
|
||||||
|
// No reachable transport: everything queues. The cap is 100 per peer,
|
||||||
|
// so the 101st enqueue evicts the oldest.
|
||||||
|
for i in 0...100 {
|
||||||
|
router.sendPrivate("Hello \(i)", to: peerID, recipientNickname: "Peer", messageID: "q\(i)")
|
||||||
|
}
|
||||||
|
|
||||||
|
#expect(dropped == ["q0"])
|
||||||
|
}
|
||||||
|
|
||||||
@Test @MainActor
|
@Test @MainActor
|
||||||
func sendReadReceipt_usesReachableTransport() async {
|
func sendReadReceipt_usesReachableTransport() async {
|
||||||
let peerID = PeerID(str: "0000000000000003")
|
let peerID = PeerID(str: "0000000000000003")
|
||||||
@@ -135,3 +227,9 @@ struct MessageRouterTests {
|
|||||||
#expect(transport.sentFavoriteNotifications.count == 1)
|
#expect(transport.sentFavoriteNotifications.count == 1)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// Mutable wall clock injected into `MessageRouter` so TTL expiry is testable
|
||||||
|
/// without real waiting.
|
||||||
|
private final class MutableTestClock {
|
||||||
|
var now = Date(timeIntervalSince1970: 1_700_000_000)
|
||||||
|
}
|
||||||
|
|||||||
@@ -1195,6 +1195,98 @@ final class NostrRelayManagerTests: XCTestCase {
|
|||||||
XCTAssertTrue(reconnected)
|
XCTAssertTrue(reconnected)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func test_pendingSubscriptions_perRelayCapEvictsOldestByInsertionOrder() async {
|
||||||
|
let relayURL = "wss://pending-cap.example"
|
||||||
|
// Tor stalled: nothing flushes, so every REQ stays pending.
|
||||||
|
let context = makeContext(permission: .denied, userTorEnabled: true, torEnforced: true, torIsReady: false)
|
||||||
|
let cap = TransportConfig.nostrPendingSubscriptionsPerRelayCap
|
||||||
|
|
||||||
|
for i in 0..<(cap + 3) {
|
||||||
|
context.manager.subscribe(filter: makeFilter(), id: "cap-sub-\(i)", relayUrls: [relayURL], handler: { _ in })
|
||||||
|
}
|
||||||
|
|
||||||
|
XCTAssertEqual(context.manager.debugPendingSubscriptionCount(for: relayURL), cap)
|
||||||
|
let pendingIDs = context.manager.debugPendingSubscriptionIDs(for: relayURL)
|
||||||
|
// The three oldest entries were evicted; the newest survive.
|
||||||
|
for i in 0..<3 {
|
||||||
|
XCTAssertFalse(pendingIDs.contains("cap-sub-\(i)"), "expected cap-sub-\(i) to be evicted")
|
||||||
|
}
|
||||||
|
for i in 3..<(cap + 3) {
|
||||||
|
XCTAssertTrue(pendingIDs.contains("cap-sub-\(i)"), "expected cap-sub-\(i) to be retained")
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func test_pendingSubscriptions_staleEntriesSweptOnConnectAttempt() async {
|
||||||
|
let relayURL = "wss://pending-sweep.example"
|
||||||
|
// Tor stalled: the REQ stays pending and no socket ever opens.
|
||||||
|
let context = makeContext(permission: .denied, userTorEnabled: true, torEnforced: true, torIsReady: false)
|
||||||
|
|
||||||
|
context.manager.subscribe(filter: makeFilter(), id: "stale-pending-sub", relayUrls: [relayURL], handler: { _ in })
|
||||||
|
XCTAssertEqual(context.manager.debugPendingSubscriptionCount(for: relayURL), 1)
|
||||||
|
|
||||||
|
// Just under the TTL the entry survives a connect attempt.
|
||||||
|
context.clock.now = context.clock.now.addingTimeInterval(TransportConfig.nostrPendingSubscriptionTTLSeconds - 1)
|
||||||
|
context.manager.ensureConnections(to: [relayURL])
|
||||||
|
XCTAssertEqual(context.manager.debugPendingSubscriptionCount(for: relayURL), 1)
|
||||||
|
|
||||||
|
// Past the TTL the next connect attempt sweeps it.
|
||||||
|
context.clock.now = context.clock.now.addingTimeInterval(2)
|
||||||
|
context.manager.ensureConnections(to: [relayURL])
|
||||||
|
XCTAssertEqual(context.manager.debugPendingSubscriptionCount(for: relayURL), 0)
|
||||||
|
}
|
||||||
|
|
||||||
|
func test_reconnectBackoff_appliesJitterWithinConfiguredBounds() async {
|
||||||
|
let relayURL = "wss://jitter-bounds.example"
|
||||||
|
// Pin the jitter source to the extremes and the midpoint of [0, 1).
|
||||||
|
let jitter = JitterSequence([0.0, 1.0.nextDown, 0.25])
|
||||||
|
let context = makeContext(permission: .denied, jitterUnit: { jitter.next() })
|
||||||
|
// Persistent ping failure: every connect attempt fails and schedules
|
||||||
|
// the next reconnect with an increasing attempt count.
|
||||||
|
context.sessionFactory.pingErrorByURL[relayURL] = NSError(domain: NSURLErrorDomain, code: NSURLErrorTimedOut)
|
||||||
|
|
||||||
|
context.manager.ensureConnections(to: [relayURL])
|
||||||
|
|
||||||
|
var delays: [TimeInterval] = []
|
||||||
|
for attempt in 1...3 {
|
||||||
|
let scheduled = await waitUntil { context.scheduler.scheduled.count == 1 }
|
||||||
|
XCTAssertTrue(scheduled, "reconnect for attempt \(attempt) was not scheduled")
|
||||||
|
delays.append(context.scheduler.scheduled[0].delay)
|
||||||
|
context.scheduler.runNext()
|
||||||
|
}
|
||||||
|
|
||||||
|
// Bases: 1s, 2s, 4s. Jitter factors: 0.8, ~1.2, 0.9.
|
||||||
|
XCTAssertEqual(delays[0], 0.8 * TransportConfig.nostrRelayInitialBackoffSeconds, accuracy: 1e-9)
|
||||||
|
XCTAssertEqual(delays[1], 1.2 * TransportConfig.nostrRelayInitialBackoffSeconds * TransportConfig.nostrRelayBackoffMultiplier, accuracy: 1e-6)
|
||||||
|
XCTAssertEqual(delays[2], 0.9 * TransportConfig.nostrRelayInitialBackoffSeconds * pow(TransportConfig.nostrRelayBackoffMultiplier, 2), accuracy: 1e-9)
|
||||||
|
}
|
||||||
|
|
||||||
|
func test_reconnectBackoff_realRandomJitterStaysInBoundsAndVaries() async {
|
||||||
|
let relayURL = "wss://jitter-random.example"
|
||||||
|
let context = makeContext(permission: .denied, jitterUnit: { Double.random(in: 0..<1) })
|
||||||
|
context.sessionFactory.pingErrorByURL[relayURL] = NSError(domain: NSURLErrorDomain, code: NSURLErrorTimedOut)
|
||||||
|
|
||||||
|
context.manager.ensureConnections(to: [relayURL])
|
||||||
|
|
||||||
|
var factors: [Double] = []
|
||||||
|
for attempt in 1...5 {
|
||||||
|
let scheduled = await waitUntil { context.scheduler.scheduled.count == 1 }
|
||||||
|
XCTAssertTrue(scheduled, "reconnect for attempt \(attempt) was not scheduled")
|
||||||
|
let base = min(
|
||||||
|
TransportConfig.nostrRelayInitialBackoffSeconds * pow(TransportConfig.nostrRelayBackoffMultiplier, Double(attempt - 1)),
|
||||||
|
TransportConfig.nostrRelayMaxBackoffSeconds
|
||||||
|
)
|
||||||
|
let factor = context.scheduler.scheduled[0].delay / base
|
||||||
|
XCTAssertGreaterThanOrEqual(factor, 1.0 - TransportConfig.nostrRelayBackoffJitterRatio)
|
||||||
|
XCTAssertLessThan(factor, 1.0 + TransportConfig.nostrRelayBackoffJitterRatio)
|
||||||
|
factors.append(factor)
|
||||||
|
context.scheduler.runNext()
|
||||||
|
}
|
||||||
|
|
||||||
|
// A real RNG must not produce a constant delay across attempts
|
||||||
|
// (5 identical uniform doubles is probability ~0).
|
||||||
|
XCTAssertGreaterThan(Set(factors).count, 1)
|
||||||
|
}
|
||||||
|
|
||||||
private func makeContext(
|
private func makeContext(
|
||||||
permission: LocationChannelManager.PermissionState,
|
permission: LocationChannelManager.PermissionState,
|
||||||
favorites: Set<Data> = [],
|
favorites: Set<Data> = [],
|
||||||
@@ -1202,7 +1294,8 @@ final class NostrRelayManagerTests: XCTestCase {
|
|||||||
userTorEnabled: Bool = false,
|
userTorEnabled: Bool = false,
|
||||||
torEnforced: Bool = false,
|
torEnforced: Bool = false,
|
||||||
torIsReady: Bool = true,
|
torIsReady: Bool = true,
|
||||||
torIsForeground: Bool = true
|
torIsForeground: Bool = true,
|
||||||
|
jitterUnit: @escaping () -> Double = { 0.5 } // 0.5 -> jitter factor 1.0 (no jitter)
|
||||||
) -> RelayManagerTestContext {
|
) -> RelayManagerTestContext {
|
||||||
let permissionSubject = CurrentValueSubject<LocationChannelManager.PermissionState, Never>(permission)
|
let permissionSubject = CurrentValueSubject<LocationChannelManager.PermissionState, Never>(permission)
|
||||||
let favoritesSubject = CurrentValueSubject<Set<Data>, Never>(favorites)
|
let favoritesSubject = CurrentValueSubject<Set<Data>, Never>(favorites)
|
||||||
@@ -1228,7 +1321,8 @@ final class NostrRelayManagerTests: XCTestCase {
|
|||||||
scheduleAfter: { delay, action in
|
scheduleAfter: { delay, action in
|
||||||
scheduler.schedule(delay: delay, action: action)
|
scheduler.schedule(delay: delay, action: action)
|
||||||
},
|
},
|
||||||
now: { clock.now }
|
now: { clock.now },
|
||||||
|
jitterUnit: jitterUnit
|
||||||
)
|
)
|
||||||
)
|
)
|
||||||
return RelayManagerTestContext(
|
return RelayManagerTestContext(
|
||||||
@@ -1305,6 +1399,20 @@ private final class MutableClock {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// Deterministic jitter source: returns the queued values in order, then a
|
||||||
|
/// neutral 0.5 (jitter factor 1.0) once exhausted.
|
||||||
|
private final class JitterSequence {
|
||||||
|
private var values: [Double]
|
||||||
|
|
||||||
|
init(_ values: [Double]) {
|
||||||
|
self.values = values
|
||||||
|
}
|
||||||
|
|
||||||
|
func next() -> Double {
|
||||||
|
values.isEmpty ? 0.5 : values.removeFirst()
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
private final class MutableBool {
|
private final class MutableBool {
|
||||||
var value: Bool
|
var value: Bool
|
||||||
|
|
||||||
|
|||||||
Reference in New Issue
Block a user