diff --git a/Meshtastic/Accessory/Accessory Manager/AccessoryManager+Connect.swift b/Meshtastic/Accessory/Accessory Manager/AccessoryManager+Connect.swift index 2386dec77..700d48ef7 100644 --- a/Meshtastic/Accessory/Accessory Manager/AccessoryManager+Connect.swift +++ b/Meshtastic/Accessory/Accessory Manager/AccessoryManager+Connect.swift @@ -38,6 +38,7 @@ extension AccessoryManager { self.activeDeviceNum = nil packetsSent = 0 packetsReceived = 0 + packetsAtLastIngestRecycle = 0 expectedNodeDBSize = nil self.allowDisconnect = true diff --git a/Meshtastic/Accessory/Accessory Manager/AccessoryManager.swift b/Meshtastic/Accessory/Accessory Manager/AccessoryManager.swift index 00ded983b..eb4fa7c20 100644 --- a/Meshtastic/Accessory/Accessory Manager/AccessoryManager.swift +++ b/Meshtastic/Accessory/Accessory Manager/AccessoryManager.swift @@ -233,6 +233,20 @@ class AccessoryManager: ObservableObject, MqttClientProxyManagerDelegate { var packetsSent: Int = 0 var packetsReceived: Int = 0 + /// Packet count at the last periodic MeshPackets recycle (see `didReceive`). The ingest + /// actor's ModelContext registers every entity it inserts or faults and never lets go, so a + /// long high-traffic session accumulates them without bound (a sustained TCP stress replay + /// reached millions of live model objects / multi-GB RSS). Recreating the actor releases + /// them; connect-time recreation alone doesn't help a session that stays connected. + var packetsAtLastIngestRecycle: Int = 0 + /// How many packets between recycles. Each processed packet leaves a handful of registered + /// objects behind, so this bounds the ingest context's working set to a few hundred MB at + /// worst. Under a saturating TCP replay this fires every couple of minutes; on a busy real + /// mesh every ~10 minutes; on a quiet mesh it may never fire — which is fine, since + /// accumulation is proportional to packets processed. Recycling costs one context teardown + /// plus cold caches on the next few fetches. + static let ingestRecycleInterval = 5_000 + // Debug counter: MQTT client-proxy downlink packets dropped before forwarding // to the device because they carried no payload (see MqttForwardFilter). NOT // @Published — read only for debug logging, so it needn't drive view updates. @@ -537,6 +551,16 @@ class AccessoryManager: ObservableObject, MqttClientProxyManagerDelegate { } // Logger.transport.info("✅ [Accessory] didReceive: \(fromRadio.payloadVariant.debugDescription)") await self.processFromRadio(fromRadio) + // Periodically recycle the ingest actor so its ModelContext releases accumulated + // registered objects (see packetsAtLastIngestRecycle). Only while subscribed — + // never mid node-DB retrieval — and only here, between packets, where no in-flight + // handler still holds the retiring instance. Flush first: recreateShared's + // invalidate deliberately drops a retired instance's pending writes. + if case .subscribed = state, packetsReceived - packetsAtLastIngestRecycle >= Self.ingestRecycleInterval { + packetsAtLastIngestRecycle = packetsReceived + await MeshPackets.shared.flushDebouncedSaves() + MeshPackets.recreateShared() + } Task { await self.heartbeatResponseTimer?.cancel(withReason: "Data packet received") await self.heartbeatTimer?.reset(delay: .seconds(Self.heartbeatInterval)) diff --git a/Meshtastic/Accessory/Transports/Bluetooth Low Energy/BLEConnection.swift b/Meshtastic/Accessory/Transports/Bluetooth Low Energy/BLEConnection.swift index 6650c2f8e..e1583c275 100644 --- a/Meshtastic/Accessory/Transports/Bluetooth Low Energy/BLEConnection.swift +++ b/Meshtastic/Accessory/Transports/Bluetooth Low Energy/BLEConnection.swift @@ -194,7 +194,10 @@ actor BLEConnection: Connection { } func getPacketStream() -> AsyncStream { - AsyncStream { continuation in + // Bounded like TCPConnection's stream: drop the oldest events under sustained overload + // instead of queueing them without limit. BLE can't reach rates that fill this in + // practice; the bound is a backstop so no transport can balloon memory. + AsyncStream(bufferingPolicy: .bufferingNewest(4096)) { continuation in // Finish any previous stream so its consumer's `for await` loop terminates cleanly // instead of hanging indefinitely on the abandoned continuation. self.connectionStreamContinuation?.finish() diff --git a/Meshtastic/Accessory/Transports/Serial/SerialConnection.swift b/Meshtastic/Accessory/Transports/Serial/SerialConnection.swift index c02e01e64..2775e7530 100644 --- a/Meshtastic/Accessory/Transports/Serial/SerialConnection.swift +++ b/Meshtastic/Accessory/Transports/Serial/SerialConnection.swift @@ -253,7 +253,9 @@ actor SerialConnection: Connection { // MARK: - Stream Management private func getPacketStream() -> AsyncStream { - AsyncStream { continuation in + // Bounded like TCPConnection's stream: drop the oldest events under sustained overload + // instead of queueing them without limit. + AsyncStream(bufferingPolicy: .bufferingNewest(4096)) { continuation in self.eventStreamContinuation = continuation continuation.onTermination = { _ in Task { diff --git a/Meshtastic/Accessory/Transports/TCP/TCPConnection.swift b/Meshtastic/Accessory/Transports/TCP/TCPConnection.swift index e865f5961..f6ef6bdc1 100644 --- a/Meshtastic/Accessory/Transports/TCP/TCPConnection.swift +++ b/Meshtastic/Accessory/Transports/TCP/TCPConnection.swift @@ -91,7 +91,7 @@ actor TCPConnection: Connection { let payload = try await receiveData(min: Int(length), max: Int(length)) switch FromRadioDecoder.classify(payload) { case .decoded(let fromRadio): - await connectionStreamContinuation?.yield(.data(fromRadio)) + await self.yieldDataEvent(fromRadio) case .skipInvalidUTF8(let error): // A string field failed UTF-8 validation; skip this frame and keep reading // rather than tearing down an otherwise healthy connection over one @@ -204,11 +204,38 @@ actor TCPConnection: Connection { // For TCP, reader is already started } + /// Yields into a full stream buffer since connect (see `getPacketStream`). With + /// `.bufferingNewest` each such yield evicts the oldest buffered frame, so this counts + /// evictions to within ±1 (the yield that exactly fills the buffer also reports 0 remaining). + private var saturatedYieldCount = 0 + + /// Yields a decoded frame into the event stream, counting full-buffer yields so sustained + /// backpressure is visible in the logs instead of silent. `.bufferingNewest` never reports + /// `.dropped` — the NEW element is always enqueued and the OLDEST is evicted — so a full + /// buffer surfaces as `.enqueued(remaining: 0)`. + private func yieldDataEvent(_ fromRadio: FromRadio) { + guard let continuation = connectionStreamContinuation else { return } + if case .enqueued(let remaining) = continuation.yield(.data(fromRadio)), remaining == 0 { + saturatedYieldCount += 1 + if saturatedYieldCount == 1 || saturatedYieldCount % 1_000 == 0 { + Logger.transport.warning("🌊 [TCP] Event buffer full — oldest frames are being evicted (\(self.saturatedYieldCount) saturated yields this session); consumer is behind the radio's packet rate") + } + } + } + private func getPacketStream() -> AsyncStream { self.connectionStreamContinuation?.finish() self.connectionStreamContinuation = nil - - return AsyncStream { continuation in + + // Bounded buffer: the default unbounded policy let a fast producer (a TCP radio can + // sustain >100 packets/s) queue every frame the main-actor consumer hadn't processed + // yet — under a stress replay that backlog reached ~all packets of the session, holding + // their protobufs live (multi-hundred-MB) while the app fell minutes behind real time. + // Keeping the newest 4,096 events drops the OLDEST first under sustained overload — + // stale mesh traffic, exactly what a saturated real radio would shed — while a node-DB + // dump (one event per node) fits well inside the bound, and error/teardown events are + // always the newest when they occur. + return AsyncStream(bufferingPolicy: .bufferingNewest(4096)) { continuation in self.connectionStreamContinuation = continuation continuation.onTermination = { [weak self] termination in guard let self else { return } diff --git a/Meshtastic/Helpers/MeshPackets.swift b/Meshtastic/Helpers/MeshPackets.swift index 2602192c9..44df85352 100644 --- a/Meshtastic/Helpers/MeshPackets.swift +++ b/Meshtastic/Helpers/MeshPackets.swift @@ -196,6 +196,17 @@ actor MeshPackets { return true } + /// Last time the direct-message unread badge was recomputed. Same O(unread) scan and same + /// burst hazard as the channel badge, so it gets the same ~1/sec rate limit — under a DM + /// flood the badge tolerates a brief lag and resyncs on app-active and on read. + private var lastDirectUnreadRecompute: ContinuousClock.Instant? + func shouldRecomputeDirectUnread() -> Bool { + let now = ContinuousClock.now + if let last = lastDirectUnreadRecompute, now - last < .seconds(1) { return false } + lastDirectUnreadRecompute = now + return true + } + func shouldPrunePositionHistory(for nodeNum: Int64) -> Bool { let nextCount = (positionInsertsSincePrune[nodeNum] ?? 0) + 1 if nextCount >= Self.positionPruneInterval { @@ -1412,8 +1423,10 @@ actor MeshPackets { return } if newMessage.fromUser != nil && newMessage.toUser != nil { - // Set Unread Message Indicators - if packet.to == connectedNode { + // Set Unread Message Indicators. unreadMessages is O(unread); like the + // channel badge, recomputing it for every incoming DM turns a burst into + // quadratic work, so it gets the same ~1/sec rate limit. + if packet.to == connectedNode, shouldRecomputeDirectUnread() { let unreadCount = await newMessage.toUser?.unreadMessages(context: modelContext, skipLastMessageCheck: true) ?? 0 // skipLastMessageCheck=true because we don't update lastMessage on our own connected node Task { @MainActor in appState?.unreadDirectMessages = unreadCount diff --git a/Meshtastic/MeshtasticApp.swift b/Meshtastic/MeshtasticApp.swift index 565b499b9..d907a5541 100644 --- a/Meshtastic/MeshtasticApp.swift +++ b/Meshtastic/MeshtasticApp.swift @@ -156,6 +156,24 @@ struct MeshtasticAppleApp: App { accessoryManager.startDiscovery() } } +#if DEBUG + // Automated perf/stress testing: connect straight to a TCP radio (or replay + // server) with `-meshtastic-connect-tcp `, skipping the Connect + // tab entirely. DEBUG-only, like the other automation hooks above. + let arguments = ProcessInfo.processInfo.arguments + if let flagIndex = arguments.firstIndex(of: "-meshtastic-connect-tcp"), + arguments.indices.contains(flagIndex + 1), + let tcpTransport = accessoryManager.transportForType(.tcp), + let device = tcpTransport.device(forManualConnection: arguments[flagIndex + 1]) { + let manager = accessoryManager + Task { + // Give startup (container, transports, discovery) a beat to settle. + try? await Task.sleep(for: .seconds(2)) + Logger.services.info("🧪 [App] Auto-connecting to TCP device \(device.identifier, privacy: .public) (launch argument)") + try? await manager.connect(to: device) + } + } +#endif } }