mirror of
https://github.com/permissionlesstech/bitchat.git
synced 2026-07-24 21:45:20 +00:00
Add field correctness diagnostics: store invariant audit, drop and bounds proofs
ConversationStore.auditInvariants() verifies the per-conversation and store-level message-ID indexes, caps, timestamp ordering, unread-set membership, and selection validity - wired to the existing read-receipt cleanup cadence, loud (.error) on violation, sampled heartbeat when healthy (~2.8ms per audit at 5k messages, benchmarked and floored). Router drops log both outcomes (marked failed / skipped by no-downgrade guard); relay cap evictions, age sweeps, and jittered reconnect delays log their counts; mirrored republishes get a sampled proof line. 11 new invariant tests corrupt store state through DEBUG-only hooks since the single-writer lockdown makes those states unreachable via intents. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
jack
@@ -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<ConversationID>] = [:]
|
||||
|
||||
/// 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<ConversationChange, Never>()
|
||||
|
||||
// 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 {
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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] {
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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..<conversation.cap {
|
||||
store.append(makeMessage(id: "m\(index)", timestamp: TimeInterval(index)), to: .mesh)
|
||||
}
|
||||
#expect(store.auditInvariants().isEmpty)
|
||||
|
||||
store._testAppendBypassingCap(
|
||||
makeMessage(id: "over-cap", timestamp: TimeInterval(conversation.cap)),
|
||||
to: .mesh
|
||||
)
|
||||
|
||||
let violations = store.auditInvariants()
|
||||
#expect(violations.contains { $0.contains("exceeds cap") })
|
||||
// The hook keeps the map exact: only the cap invariant fires.
|
||||
#expect(violations.count == 1)
|
||||
}
|
||||
|
||||
@Test("audit flags unread entries for nonexistent conversations")
|
||||
@MainActor
|
||||
func auditFlagsStaleUnreadEntry() {
|
||||
let store = Self.makeExercisedStore()
|
||||
store._testInsertUnreadConversationID(makeDirectConversationID("gone"))
|
||||
|
||||
let violations = store.auditInvariants()
|
||||
#expect(violations.contains { $0.contains("unreadConversations contains unknown conversation") })
|
||||
}
|
||||
|
||||
@Test("audit flags a selection pointing at a nonexistent conversation")
|
||||
@MainActor
|
||||
func auditFlagsDanglingSelection() {
|
||||
let store = Self.makeExercisedStore()
|
||||
store._testSetSelectedConversationID(makeDirectConversationID("gone"))
|
||||
|
||||
let violations = store.auditInvariants()
|
||||
#expect(violations.contains { $0.contains("selectedConversationID") && $0.contains("has no conversation") })
|
||||
}
|
||||
|
||||
@Test("appendCount counts every insertion, not duplicates or updates")
|
||||
@MainActor
|
||||
func appendCountTracksInsertions() {
|
||||
let store = ConversationStore()
|
||||
let source = makeDirectConversationID("ephemeral")
|
||||
let destination = makeDirectConversationID("stable")
|
||||
|
||||
store.append(makeMessage(id: "m1", timestamp: 1), to: .mesh)
|
||||
store.append(makeMessage(id: "m1", timestamp: 1), to: .mesh) // duplicate: not counted
|
||||
store.upsertByID(makeMessage(id: "m2", timestamp: 2), in: .mesh) // upsert-append: counted
|
||||
store.upsertByID(makeMessage(id: "m2", timestamp: 2, content: "edit"), in: .mesh) // in-place: not counted
|
||||
#expect(store.appendCount == 2)
|
||||
|
||||
store.append(makeMessage(id: "dm-1", timestamp: 1, isPrivate: true), to: source)
|
||||
store.migrateConversation(from: source, to: destination) // migration insert: counted
|
||||
#expect(store.appendCount == 4)
|
||||
|
||||
// Removal does not decrement: the counter is a throughput odometer.
|
||||
store.removeMessage(withID: "dm-1", from: destination)
|
||||
#expect(store.appendCount == 4)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -496,6 +496,34 @@ final class PerformanceBaselineTests: XCTestCase {
|
||||
reportThroughput("store.append", samples: samples, operations: messageCount, unit: "messages")
|
||||
}
|
||||
|
||||
// MARK: - 8. ConversationStore invariant audit (field observability)
|
||||
|
||||
/// `ConversationStore.auditInvariants()` over a realistic 5k-message
|
||||
/// corpus (mesh + geohash + 75 private chats). The audit runs in the
|
||||
/// field on the read-receipt cleanup cadence
|
||||
/// (`ChatViewModel.auditConversationStore`), so this measures the
|
||||
/// per-audit cost that piggybacks on peer-list updates — it must stay
|
||||
/// trivially cheap relative to that cadence.
|
||||
func testConversationStoreAudit() {
|
||||
let context = PerfDeliveryContext.makeCorpus(publicCount: 2000, peerCount: 75, messagesPerPeer: 40)
|
||||
let store = context.store
|
||||
XCTAssertEqual(store.totalMessageCount, 5000)
|
||||
let repsPerPass = 20
|
||||
var samples: [TimeInterval] = []
|
||||
|
||||
measure {
|
||||
let start = Date()
|
||||
var violationCount = 0
|
||||
for _ in 0..<repsPerPass {
|
||||
violationCount += store.auditInvariants().count
|
||||
}
|
||||
samples.append(Date().timeIntervalSince(start))
|
||||
XCTAssertEqual(violationCount, 0, "healthy corpus must audit clean")
|
||||
}
|
||||
|
||||
reportThroughput("store.audit", samples: samples, operations: repsPerPass, unit: "audits")
|
||||
}
|
||||
|
||||
/// Spins the main run loop in small slices (draining main-queue tasks and
|
||||
/// timers) until `condition` holds or `timeout` elapses.
|
||||
private func spinMainRunLoop(timeout: TimeInterval, until condition: () -> Bool) -> Bool {
|
||||
|
||||
@@ -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
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user