diff --git a/bitchat/App/ConversationStore.swift b/bitchat/App/ConversationStore.swift index 73b57b34..88d72085 100644 --- a/bitchat/App/ConversationStore.swift +++ b/bitchat/App/ConversationStore.swift @@ -16,6 +16,7 @@ // import BitFoundation +import BitLogger import Combine import Foundation @@ -185,6 +186,39 @@ final class Conversation: ObservableObject, Identifiable { indexByMessageID.removeAll() } + // MARK: Diagnostics + + /// Appends human-readable invariant violations for this conversation + /// (empty when healthy): the ID index must be the exact inverse of the + /// messages array, the cap must hold, and timestamps must be + /// non-decreasing (equal timestamps keep arrival order, so only strict + /// inversions are violations). O(messages); allocates only on violation. + fileprivate func collectInvariantViolations(into violations: inout [String], label: String) { + if indexByMessageID.count != messages.count { + violations.append("\(label): index has \(indexByMessageID.count) entries for \(messages.count) messages") + } + if messages.count > cap { + violations.append("\(label): \(messages.count) messages exceeds cap \(cap)") + } + var previousTimestamp: Date? + for position in messages.indices { + let message = messages[position] + // Count equality + every message resolving to its own position + // proves the index is exactly the inverse map (no stale extras). + if let index = indexByMessageID[message.id] { + if index != position { + violations.append("\(label): message \(message.id.prefix(8))… at \(position) indexed at \(index)") + } + } else { + violations.append("\(label): message \(message.id.prefix(8))… at \(position) missing from index") + } + if let previousTimestamp, message.timestamp < previousTimestamp { + violations.append("\(label): timestamp order violated at \(position)") + } + previousTimestamp = message.timestamp + } + } + // MARK: Internals static func shouldSkipStatusUpdate(current: DeliveryStatus?, new: DeliveryStatus) -> Bool { @@ -293,6 +327,16 @@ final class ConversationStore: ObservableObject { /// copies. private var conversationIDsByMessageID: [String: Set] = [:] + /// Monotonic count of messages inserted into any conversation (appends, + /// upsert-appends, migration inserts). Field-observability only: the + /// periodic store audit folds the delta into its heartbeat line so logs + /// carry throughput context. Never read on a hot path. + private(set) var appendCount: Int = 0 + + /// Sample counter for the mirrored-republish debug log in the ID-only + /// `setDeliveryStatus` fan-out (first + every Nth occurrence). + private var mirroredRepublishLogCount = 0 + let changes = PassthroughSubject() // MARK: Intent API @@ -379,6 +423,16 @@ final class ConversationStore: ObservableObject { guard let conversation = conversationsByID[id], conversation.message(withID: messageID)?.deliveryStatus == status, conversation.republishMessage(withID: messageID) else { continue } + // Field proof the mirrored-copy republish path actually fires; + // sampled (first + every Nth) so mirrored chats can't spam logs. + mirroredRepublishLogCount += 1 + if mirroredRepublishLogCount == 1 + || mirroredRepublishLogCount.isMultiple(of: TransportConfig.conversationStoreMirroredRepublishLogInterval) { + SecureLogger.debug( + "mirrored republish #\(mirroredRepublishLogCount) for \(messageID.prefix(8))… in \(id.auditDescription)", + category: .session + ) + } changes.send(.statusChanged(id, messageID: messageID, status)) } return true @@ -567,10 +621,89 @@ final class ConversationStore: ObservableObject { } } + // MARK: Diagnostics + + /// Total messages across all conversations. O(#conversations) — heartbeat + /// logging only, never a hot path. + var totalMessageCount: Int { + conversationsByID.values.reduce(0) { $0 + $1.messages.count } + } + + /// Number of distinct message IDs in the store-level membership map. + var messageIDMapCount: Int { + conversationIDsByMessageID.count + } + + /// Verifies the store's correctness invariants and returns human-readable + /// violations (empty = healthy). Intended for a periodic field audit: + /// O(total messages) and allocation-free while healthy. Checks: + /// - the `conversationIDs` ordering array matches `conversationsByID` + /// - per conversation: ID index exact, cap held, timestamp order + /// (see `Conversation.collectInvariantViolations`) + /// - the message-ID → conversation map matches reality exactly: every + /// mapped membership points at a live conversation actually holding + /// the message, and total memberships equal total messages (with the + /// forward check, equality proves no conversation message is missing + /// from the map) + /// - `unreadConversations` only references existing conversations + /// - `selectedConversationID`, when set, references an existing + /// conversation (`select(_:)` creates on selection and + /// `removeConversation`/`clearAll` clear it, so existence is the + /// invariant for both the channel-derived and direct-peer cases) + func auditInvariants() -> [String] { + var violations: [String] = [] + + if conversationIDs.count != conversationsByID.count { + violations.append("conversationIDs lists \(conversationIDs.count) conversations but dictionary holds \(conversationsByID.count)") + } + for id in conversationIDs where conversationsByID[id] == nil { + violations.append("conversationIDs lists \(id.auditDescription) but no conversation exists") + } + + var totalMessages = 0 + for (id, conversation) in conversationsByID { + totalMessages += conversation.messages.count + conversation.collectInvariantViolations(into: &violations, label: id.auditDescription) + } + + var totalMappedMemberships = 0 + for (messageID, ids) in conversationIDsByMessageID { + totalMappedMemberships += ids.count + if ids.isEmpty { + violations.append("message map: \(messageID.prefix(8))… has an empty membership set") + } + for id in ids { + guard let conversation = conversationsByID[id] else { + violations.append("message map: \(messageID.prefix(8))… claims unknown conversation \(id.auditDescription)") + continue + } + if !conversation.containsMessage(withID: messageID) { + violations.append("message map: \(messageID.prefix(8))… not present in claimed conversation \(id.auditDescription)") + } + } + } + if totalMappedMemberships != totalMessages { + violations.append("message map holds \(totalMappedMemberships) memberships but conversations hold \(totalMessages) messages") + } + + for id in unreadConversations where conversationsByID[id] == nil { + violations.append("unreadConversations contains unknown conversation \(id.auditDescription)") + } + + if let selected = selectedConversationID, conversationsByID[selected] == nil { + violations.append("selectedConversationID \(selected.auditDescription) has no conversation") + } + + return violations + } + // MARK: Internals private func registerMessageID(_ messageID: String, in id: ConversationID) { conversationIDsByMessageID[messageID, default: []].insert(id) + // Single choke point for every successful insertion (append, upsert + // append, migration insert) — the audit heartbeat's throughput delta. + appendCount += 1 } private func unregisterMessageID(_ messageID: String, from id: ConversationID) { @@ -675,6 +808,96 @@ extension ConversationStore { } } +// MARK: - Diagnostics support + +extension ConversationID { + /// Short, log-safe description for audit/diagnostic lines. Direct + /// conversations truncate the handle so full peer keys never hit logs. + fileprivate var auditDescription: String { + switch self { + case .mesh: + return "mesh" + case .geohash(let geohash): + return "geo:\(geohash)" + case .direct(let handle): + return "direct:\(handle.id.prefix(13))…" + } + } +} + +#if DEBUG +// Test-only corruption hooks for `auditInvariants()` tests. The store is the +// sole writer by design — `Conversation`'s mutators are fileprivate and the +// store's backing collections are private — so the inconsistent states the +// audit exists to catch CANNOT be manufactured through the intent API. These +// DEBUG-only hooks deliberately bypass that lockdown to inject exactly those +// impossible states. Never call them outside tests. +extension Conversation { + /// Points an existing message's index entry at the wrong position + /// (positions 0 and 1 swap their index entries). Requires >= 2 messages. + func _testCorruptIndexEntries() { + guard messages.count >= 2 else { return } + indexByMessageID[messages[0].id] = 1 + indexByMessageID[messages[1].id] = 0 + } + + /// Drops a message's index entry entirely (count mismatch + missing). + func _testRemoveIndexEntry(forMessageID messageID: String) { + indexByMessageID.removeValue(forKey: messageID) + } + + /// Swaps the first and last messages while keeping the index consistent, + /// so ONLY the timestamp-order invariant is violated (requires the two + /// messages to have distinct timestamps). + func _testCorruptOrderingPreservingIndex() { + guard messages.count >= 2 else { return } + messages.swapAt(0, messages.count - 1) + indexByMessageID[messages[0].id] = 0 + indexByMessageID[messages[messages.count - 1].id] = messages.count - 1 + } +} + +extension ConversationStore { + /// Adds a map membership that the conversation does not actually hold. + func _testRegisterPhantomMessageID(_ messageID: String, in id: ConversationID) { + conversationIDsByMessageID[messageID, default: []].insert(id) + } + + /// Drops a real map membership (conversation message missing from map). + func _testUnregisterMessageID(_ messageID: String, from id: ConversationID) { + conversationIDsByMessageID[messageID]?.remove(id) + if conversationIDsByMessageID[messageID]?.isEmpty == true { + conversationIDsByMessageID.removeValue(forKey: messageID) + } + } + + /// Appends past the conversation cap, bypassing trim (map kept exact so + /// only the cap invariant is violated). + func _testAppendBypassingCap(_ message: BitchatMessage, to id: ConversationID) { + let conversation = conversation(for: id) + conversation._testAppendBypassingTrim(message) + conversationIDsByMessageID[message.id, default: []].insert(id) + } + + /// Marks a nonexistent conversation unread without creating it. + func _testInsertUnreadConversationID(_ id: ConversationID) { + unreadConversations.insert(id) + } + + /// Sets the selection directly, without `select(_:)`'s create-on-select. + func _testSetSelectedConversationID(_ id: ConversationID?) { + selectedConversationID = id + } +} + +extension Conversation { + fileprivate func _testAppendBypassingTrim(_ message: BitchatMessage) { + messages.append(message) + indexByMessageID[message.id] = messages.count - 1 + } +} +#endif + // MARK: - Public timeline derived views extension ConversationStore { diff --git a/bitchat/Nostr/NostrRelayManager.swift b/bitchat/Nostr/NostrRelayManager.swift index b2eaa53f..aa9562c4 100644 --- a/bitchat/Nostr/NostrRelayManager.swift +++ b/bitchat/Nostr/NostrRelayManager.swift @@ -656,11 +656,16 @@ final class NostrRelayManager: ObservableObject { // durable intent stays in subscriptionRequestState, so an evicted REQ // is still replayed if its subscription is active when the relay // (re)connects. + var evictedCount = 0 while map.count > TransportConfig.nostrPendingSubscriptionsPerRelayCap, let oldest = map.min(by: { $0.value.sequence < $1.value.sequence }) { map.removeValue(forKey: oldest.key) + evictedCount += 1 + } + if evictedCount > 0 { + // Bounds proof: the cap eviction actually removed entries. SecureLogger.warning( - "📋 Pending subscription overflow for \(url) - evicted oldest id=\(oldest.key)", + "📋 Evicted \(evictedCount) pending sub(s) over cap for \(url)", category: .session ) } @@ -678,8 +683,10 @@ final class NostrRelayManager: ObservableObject { 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)", + // Bounds proof: the age sweep actually removed entries. Warning + // (not debug) — stale pending REQs mean a relay never came up. + SecureLogger.warning( + "📋 Swept \(map.count - fresh.count) stale pending sub(s) for \(url)", category: .session ) pendingSubscriptions[url] = fresh.isEmpty ? nil : fresh @@ -1053,8 +1060,14 @@ final class NostrRelayManager: ObservableObject { let nextReconnectTime = dependencies.now().addingTimeInterval(backoffInterval) relays[index].nextReconnectTime = nextReconnectTime - - + + // Reconnects are bounded by maxReconnectAttempts and exponentially + // backed off, so this is low-frequency: plain debug, no sampling. + SecureLogger.debug( + "🔄 Reconnect \(relayUrl) in \(String(format: "%.1f", backoffInterval))s (base \(String(format: "%.1f", baseBackoffInterval))s, attempt \(relays[index].reconnectAttempts)/\(maxReconnectAttempts))", + category: .session + ) + // Schedule reconnection with exponential backoff let gen = connectionGeneration dependencies.scheduleAfter(backoffInterval) { [weak self] in diff --git a/bitchat/Services/TransportConfig.swift b/bitchat/Services/TransportConfig.swift index 87255245..d4c9d431 100644 --- a/bitchat/Services/TransportConfig.swift +++ b/bitchat/Services/TransportConfig.swift @@ -50,6 +50,14 @@ enum TransportConfig { // Sample interval for per-event debug logs on the inbound hot path. static let nostrInboundEventLogInterval: Int = 100 + // Conversation store diagnostics (field observability) + // Sample interval for the periodic store-audit "OK" heartbeat line + // (first + every Nth audit); violations always log at error level. + static let conversationStoreAuditLogInterval: Int = 10 + // Sample interval for the mirrored-republish debug line in the ID-only + // delivery fan-out (first + every Nth republish). + static let conversationStoreMirroredRepublishLogInterval: Int = 25 + // UI thresholds static let uiProcessedNostrEventsCap: Int = 2000 static let uiChannelInactivityThresholdSeconds: TimeInterval = 9 * 60 diff --git a/bitchat/ViewModels/ChatViewModel.swift b/bitchat/ViewModels/ChatViewModel.swift index 93f98484..8bf4222a 100644 --- a/bitchat/ViewModels/ChatViewModel.swift +++ b/bitchat/ViewModels/ChatViewModel.swift @@ -445,6 +445,12 @@ final class ChatViewModel: ObservableObject, BitchatDelegate, TransportEventDele // Track app startup phase to prevent marking old messages as unread var isStartupPhase = true + + // ConversationStore field audit bookkeeping (see auditConversationStore()): + // runs on the read-receipt cleanup cadence, heartbeat sampled first + + // every `TransportConfig.conversationStoreAuditLogInterval`th audit. + private var storeAuditCount = 0 + private var storeAuditLastAppendCount = 0 // Announce Tor initial readiness once per launch to avoid duplicates var torInitialReadyAnnounced: Bool = false @@ -1482,6 +1488,35 @@ final class ChatViewModel: ObservableObject, BitchatDelegate, TransportEventDele @MainActor func cleanupOldReadReceipts() { deliveryCoordinator.cleanupOldReadReceipts() + auditConversationStore() + } + + /// Periodic on-device verification of the `ConversationStore`'s + /// correctness invariants, piggybacked on the read-receipt cleanup + /// cadence (peer-list updates) so no extra timer exists. Loud on + /// violation (one error line each), near-silent when healthy (sampled + /// heartbeat: first + every Nth audit). The audit is O(total messages) + /// and allocation-free while healthy — measured ~0.5 ms at 5k messages + /// (see `PerformanceBaselineTests.testConversationStoreAudit`), cheap + /// relative to its cadence, so it always runs. + @MainActor + private func auditConversationStore() { + storeAuditCount += 1 + let violations = conversations.auditInvariants() + guard violations.isEmpty else { + for violation in violations { + SecureLogger.error("🚨 ConversationStore invariant violated: \(violation)", category: .session) + } + return + } + let appendCount = conversations.appendCount + if storeAuditCount == 1 || storeAuditCount.isMultiple(of: TransportConfig.conversationStoreAuditLogInterval) { + SecureLogger.debug( + "Store audit OK: \(conversations.conversationsByID.count) conversations, \(conversations.totalMessageCount) messages, map=\(conversations.messageIDMapCount), appends since last audit=\(appendCount - storeAuditLastAppendCount)", + category: .session + ) + } + storeAuditLastAppendCount = appendCount } func parseMentions(from content: String) -> [String] { diff --git a/bitchat/ViewModels/ChatViewModelBootstrapper.swift b/bitchat/ViewModels/ChatViewModelBootstrapper.swift index ac919c1f..e2615cf9 100644 --- a/bitchat/ViewModels/ChatViewModelBootstrapper.swift +++ b/bitchat/ViewModels/ChatViewModelBootstrapper.swift @@ -83,12 +83,22 @@ private extension ChatViewModelBootstrapper { // 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 + viewModel.messageRouter.onMessageDropped = { [weak viewModel] messageID, peerID in guard let viewModel else { return } switch viewModel.conversations.deliveryStatus(forMessageID: messageID) { case .delivered, .read: - return + // Field proof of the no-downgrade guard: the drop arrived + // after a confirmed receipt, so the `.failed` write is + // deliberately skipped. + SecureLogger.warning( + "📤 Router dropped message \(messageID.prefix(8))… for \(peerID.id.prefix(8))… → .failed skipped (already delivered/read)", + category: .session + ) default: + SecureLogger.warning( + "📤 Router dropped message \(messageID.prefix(8))… for \(peerID.id.prefix(8))… → marked failed", + category: .session + ) viewModel.conversations.setDeliveryStatus( .failed(reason: "Not delivered"), forMessageID: messageID diff --git a/bitchatTests/ConversationStoreTests.swift b/bitchatTests/ConversationStoreTests.swift index 52598db9..3c9ef6fd 100644 --- a/bitchatTests/ConversationStoreTests.swift +++ b/bitchatTests/ConversationStoreTests.swift @@ -754,4 +754,180 @@ struct ConversationStoreTests { #expect(publishedIDs.isEmpty) #expect(statusChangedIDs.isEmpty) } + + // MARK: - Invariant audit (field observability) + + /// A store exercised through every intent family: public + geohash + + /// mirrored direct conversations, out-of-order and equal-timestamp + /// appends, upserts, delivery updates, removal, migration, unread, and + /// selection. Used as the healthy baseline for audit tests. + @MainActor + private static func makeExercisedStore() -> ConversationStore { + let store = ConversationStore() + let stable = makeDirectConversationID("stable") + let ephemeral = makeDirectConversationID("ephemeral") + + store.append(makeMessage(id: "m1", timestamp: 10), to: .mesh) + store.append(makeMessage(id: "m3", timestamp: 30), to: .mesh) + store.append(makeMessage(id: "m2", timestamp: 20), to: .mesh) // out of order + store.append(makeMessage(id: "m2b", timestamp: 20), to: .mesh) // equal timestamp + store.append(makeMessage(id: "g1", timestamp: 5), to: .geohash("u4pruyd")) + store.upsertByID(makeMessage(id: "g1", timestamp: 5, content: "edited"), in: .geohash("u4pruyd")) + + // Mirrored private copy (shared instance) across two direct chats. + let mirrored = makeMessage(id: "dm-1", timestamp: 1, isPrivate: true, deliveryStatus: .sent) + store.upsertByID(mirrored, in: stable) + store.upsertByID(mirrored, in: ephemeral) + store.setDeliveryStatus(.delivered(to: "bob", at: Date(timeIntervalSince1970: 2)), forMessageID: "dm-1") + + store.append(makeMessage(id: "dm-2", timestamp: 3, isPrivate: true), to: ephemeral) + store.migrateConversation(from: ephemeral, to: stable) + store.append(makeMessage(id: "dm-gone", timestamp: 4, isPrivate: true), to: stable) + store.removeMessage(withID: "dm-gone", from: stable) + + store.markUnread(stable) + store.setActiveChannel(.mesh) + return store + } + + @Test("audit reports no violations for a healthy, well-exercised store") + @MainActor + func auditHealthyStoreIsClean() { + let store = Self.makeExercisedStore() + #expect(store.auditInvariants().isEmpty) + + // Selection through both axes stays healthy. + store.setSelectedPrivatePeer(PeerID(str: "peer-stable")) + #expect(store.auditInvariants().isEmpty) + store.setSelectedPrivatePeer(nil) + #expect(store.auditInvariants().isEmpty) + } + + @Test("audit flags index entries pointing at the wrong position") + @MainActor + func auditFlagsCorruptIndexEntries() { + let store = Self.makeExercisedStore() + store.conversation(for: .mesh)._testCorruptIndexEntries() + + let violations = store.auditInvariants() + #expect(!violations.isEmpty) + #expect(violations.contains { $0.contains("mesh") && $0.contains("indexed at") }) + } + + @Test("audit flags a message missing from the per-conversation index") + @MainActor + func auditFlagsMissingIndexEntry() { + let store = Self.makeExercisedStore() + store.conversation(for: .mesh)._testRemoveIndexEntry(forMessageID: "m1") + + let violations = store.auditInvariants() + #expect(violations.contains { $0.contains("missing from index") }) + #expect(violations.contains { $0.contains("index has") }) // count mismatch + } + + @Test("audit flags timestamp-order violations") + @MainActor + func auditFlagsOrderingViolation() { + let store = Self.makeExercisedStore() + store.conversation(for: .mesh)._testCorruptOrderingPreservingIndex() + + let violations = store.auditInvariants() + #expect(violations.contains { $0.contains("timestamp order violated") }) + // The hook keeps the index consistent: no index violations leak in. + #expect(!violations.contains { $0.contains("indexed at") || $0.contains("missing from index") }) + } + + @Test("audit flags map memberships the conversation does not hold") + @MainActor + func auditFlagsPhantomMapMembership() { + let store = Self.makeExercisedStore() + store._testRegisterPhantomMessageID("ghost", in: .mesh) + + let violations = store.auditInvariants() + #expect(violations.contains { $0.contains("not present in claimed conversation") }) + #expect(violations.contains { $0.contains("memberships but conversations hold") }) + } + + @Test("audit flags map memberships claiming unknown conversations") + @MainActor + func auditFlagsUnknownConversationMembership() { + let store = Self.makeExercisedStore() + store._testRegisterPhantomMessageID("ghost", in: makeDirectConversationID("nope")) + + let violations = store.auditInvariants() + #expect(violations.contains { $0.contains("claims unknown conversation") }) + } + + @Test("audit flags conversation messages missing from the map") + @MainActor + func auditFlagsMissingMapMembership() { + let store = Self.makeExercisedStore() + store._testUnregisterMessageID("m1", from: .mesh) + + let violations = store.auditInvariants() + #expect(violations.contains { $0.contains("memberships but conversations hold") }) + } + + @Test("audit flags a conversation exceeding its cap") + @MainActor + func auditFlagsCapViolation() { + let store = ConversationStore() + let conversation = store.conversation(for: .mesh) + for index in 0.. Bool) -> Bool { diff --git a/bitchatTests/Performance/perf-floors.json b/bitchatTests/Performance/perf-floors.json index 41517400..af349819 100644 --- a/bitchatTests/Performance/perf-floors.json +++ b/bitchatTests/Performance/perf-floors.json @@ -24,7 +24,8 @@ "formatting.formatMessage": 12261, "pipeline.privateIngest": 24848, "pipeline.publicIngest": 13102, - "store.append": 213201 + "store.append": 213201, + "store.audit": 362 }, "floors": { "nostrInbound.fresh": 530, @@ -36,6 +37,7 @@ "formatting.formatMessage": 3000, "pipeline.privateIngest": 6000, "pipeline.publicIngest": 3200, - "store.append": 53000 + "store.append": 53000, + "store.audit": 90 } }