From 97a8185051f2f5bcdd0526a7f6a6f3f7a0675ef3 Mon Sep 17 00:00:00 2001 From: Garth Vander Houwen Date: Thu, 16 Jul 2026 23:04:35 -0700 Subject: [PATCH 1/3] perf(ingest): recycle the MeshPackets actor periodically; throttle the DM unread recount MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The ingest actor's ModelContext registers every entity it inserts or faults and never releases them. recreateShared() exists for exactly this ('call ... periodically to release accumulated memory') but was only invoked at connect time, so a session that stays connected under sustained traffic accumulates registered objects without bound — a long TCP stress-replay run reached millions of live model objects and multi-gigabyte RSS, with the SwiftData observation machinery alone holding hundreds of megabytes. didReceive now flushes pending debounced saves and recycles the actor every 20,000 packets while subscribed — between packets, where no in-flight handler holds the retiring instance, and never during the node-DB retrieval phase. On a busy mesh that's every few minutes; on a quiet mesh it may never fire, which is fine because accumulation is proportional to packets processed. Also rate-limits the per-DM unreadMessages recount to ~1/sec, matching the throttle the channel-badge path already has: the recount is O(unread), so under a DM burst the per-message recompute turned the badge update into quadratic work. --- .../AccessoryManager+Connect.swift | 1 + .../Accessory Manager/AccessoryManager.swift | 22 +++++++++++++++++++ Meshtastic/Helpers/MeshPackets.swift | 17 ++++++++++++-- 3 files changed, 38 insertions(+), 2 deletions(-) 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..87d0380ae 100644 --- a/Meshtastic/Accessory/Accessory Manager/AccessoryManager.swift +++ b/Meshtastic/Accessory/Accessory Manager/AccessoryManager.swift @@ -233,6 +233,18 @@ 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. On a busy mesh (~140 pkt/s) this is a few minutes; + /// on a quiet mesh it may never fire — which is fine, since accumulation is proportional + /// to packets processed. Recycling costs one context teardown + cold caches on the next + /// few fetches, so keep it infrequent. + static let ingestRecycleInterval = 20_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 +549,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/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 From cbafc94fe8e19d9c6f38a5ea35a54f1b594dd6a0 Mon Sep 17 00:00:00 2001 From: Garth Vander Houwen Date: Thu, 16 Jul 2026 23:53:52 -0700 Subject: [PATCH 2/3] perf(ingest): bound the connection event stream; add a DEBUG TCP auto-connect hook MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The transport event streams used AsyncStream's default unbounded buffering, so a producer faster than the main-actor consumer (a TCP radio or replay can sustain >100 packets/s) queued every unprocessed frame indefinitely: under a sustained stress replay the backlog grew to effectively every packet of the session — hundreds of thousands of live protobufs — while the app fell minutes behind real time. All three transports now use .bufferingNewest(4096): under sustained overload the OLDEST frames drop first (stale mesh traffic, exactly what a saturated real radio sheds), a node-DB dump fits well inside the bound, and error/teardown events are always the newest when they occur. The TCP reader counts and logs drops so backpressure is visible instead of silent. Also drops the ingest-recycle interval to 5,000 packets — at the consumer's real throughput 20,000 meant many minutes between recycles, letting hundreds of MB of registered objects pile up between them. Adds a DEBUG-only launch argument for automated stress testing: -meshtastic-connect-tcp connects straight to a TCP radio at startup, so perf runs are fully scriptable (simctl launch + replay server) with no UI automation. --- .../Accessory Manager/AccessoryManager.swift | 12 ++++---- .../Bluetooth Low Energy/BLEConnection.swift | 5 +++- .../Transports/Serial/SerialConnection.swift | 4 ++- .../Transports/TCP/TCPConnection.swift | 29 +++++++++++++++++-- Meshtastic/MeshtasticApp.swift | 18 ++++++++++++ 5 files changed, 58 insertions(+), 10 deletions(-) diff --git a/Meshtastic/Accessory/Accessory Manager/AccessoryManager.swift b/Meshtastic/Accessory/Accessory Manager/AccessoryManager.swift index 87d0380ae..eb4fa7c20 100644 --- a/Meshtastic/Accessory/Accessory Manager/AccessoryManager.swift +++ b/Meshtastic/Accessory/Accessory Manager/AccessoryManager.swift @@ -239,11 +239,13 @@ class AccessoryManager: ObservableObject, MqttClientProxyManagerDelegate { /// 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. On a busy mesh (~140 pkt/s) this is a few minutes; - /// on a quiet mesh it may never fire — which is fine, since accumulation is proportional - /// to packets processed. Recycling costs one context teardown + cold caches on the next - /// few fetches, so keep it infrequent. - static let ingestRecycleInterval = 20_000 + /// 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 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..febcde0e9 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,34 @@ actor TCPConnection: Connection { // For TCP, reader is already started } + /// Data events dropped by the bounded stream buffer since connect (see `getPacketStream`). + private var droppedDataEventCount = 0 + + /// Yields a decoded frame into the event stream, counting buffer-overflow drops so + /// sustained backpressure is visible in the logs instead of silent. + private func yieldDataEvent(_ fromRadio: FromRadio) { + guard let continuation = connectionStreamContinuation else { return } + if case .dropped = continuation.yield(.data(fromRadio)) { + droppedDataEventCount += 1 + if droppedDataEventCount == 1 || droppedDataEventCount % 1_000 == 0 { + Logger.transport.warning("🌊 [TCP] Event buffer full — dropped oldest frame (\(self.droppedDataEventCount) dropped 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/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 } } From 4e1f5583341f3b5f7ae42809376634ab53e0aefe Mon Sep 17 00:00:00 2001 From: Garth Vander Houwen Date: Fri, 17 Jul 2026 00:09:19 -0700 Subject: [PATCH 3/3] review: buffer saturation surfaces as .enqueued(remaining: 0), not .dropped MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit With .bufferingNewest the new element is always enqueued and the OLDEST buffered frame is evicted, so yield never returns .dropped and the backpressure warning was dead code. Count yields that report zero remaining capacity instead — each one evicts an oldest frame (to within the one yield that exactly fills the buffer). Verified live: the warning now fires under a saturating replay. --- .../Transports/TCP/TCPConnection.swift | 20 +++++++++++-------- 1 file changed, 12 insertions(+), 8 deletions(-) diff --git a/Meshtastic/Accessory/Transports/TCP/TCPConnection.swift b/Meshtastic/Accessory/Transports/TCP/TCPConnection.swift index febcde0e9..f6ef6bdc1 100644 --- a/Meshtastic/Accessory/Transports/TCP/TCPConnection.swift +++ b/Meshtastic/Accessory/Transports/TCP/TCPConnection.swift @@ -204,17 +204,21 @@ actor TCPConnection: Connection { // For TCP, reader is already started } - /// Data events dropped by the bounded stream buffer since connect (see `getPacketStream`). - private var droppedDataEventCount = 0 + /// 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 buffer-overflow drops so - /// sustained backpressure is visible in the logs instead of silent. + /// 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 .dropped = continuation.yield(.data(fromRadio)) { - droppedDataEventCount += 1 - if droppedDataEventCount == 1 || droppedDataEventCount % 1_000 == 0 { - Logger.transport.warning("🌊 [TCP] Event buffer full — dropped oldest frame (\(self.droppedDataEventCount) dropped this session); consumer is behind the radio's packet rate") + 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") } } }