perf(ingest): bound memory under sustained packet load — actor recycling, bounded event stream, DM badge throttle

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

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))

Meshtastic/Accessory/Transports/Bluetooth Low Energy/BLEConnection.swift

@@ -194,7 +194,10 @@ actor BLEConnection: Connection {
 	}
 
 	func getPacketStream() -> AsyncStream<ConnectionEvent> {
-		AsyncStream<ConnectionEvent> { 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<ConnectionEvent>(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()

Meshtastic/Accessory/Transports/Serial/SerialConnection.swift

@@ -253,7 +253,9 @@ actor SerialConnection: Connection {
 
 	// MARK: - Stream Management
 	private func getPacketStream() -> AsyncStream<ConnectionEvent> {
-		AsyncStream<ConnectionEvent> { continuation in
+		// Bounded like TCPConnection's stream: drop the oldest events under sustained overload
+		// instead of queueing them without limit.
+		AsyncStream<ConnectionEvent>(bufferingPolicy: .bufferingNewest(4096)) { continuation in
 			self.eventStreamContinuation = continuation
 			continuation.onTermination = { _ in
 				Task {

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<ConnectionEvent> {
 		self.connectionStreamContinuation?.finish()
 		self.connectionStreamContinuation = nil
-
-		return AsyncStream<ConnectionEvent> { 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<ConnectionEvent>(bufferingPolicy: .bufferingNewest(4096)) { continuation in
 			self.connectionStreamContinuation = continuation
 			continuation.onTermination = { [weak self] termination in
 				guard let self else { return }

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

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 <host[:port]>`, 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
 		}
 	}