recover on exception

app/src/main/java/com/bitchat/android/mesh/BluetoothPacketBroadcaster.kt

@@ -114,47 +114,85 @@ class BluetoothPacketBroadcaster(
     private val transferJobs = ConcurrentHashMap<String, Job>()
     
     // Replace simple actor FIFO with a priority-aware channel + processor
-    private val broadcasterChannel = Channel<BroadcastRequest>(capacity = Channel.UNLIMITED)
+    @Volatile
+    private var broadcasterChannel = Channel<BroadcastRequest>(capacity = Channel.UNLIMITED)
     private val sequenceCounter = AtomicLong(0)
     
+    // Track processor state for recovery
+    @Volatile
+    private var processorJob: Job? = null
+    private val processorLock = Any()
+
     init {
         startPriorityProcessor()
     }
 
     private fun startPriorityProcessor() {
-        broadcasterScope.launch {
+        synchronized(processorLock) {
-            Log.d(TAG, "🎭 Created priority packet broadcaster processor")
+            // Cancel existing processor if running
-            // Min-heap by (primaryPriority, secondaryPriority, sequence)
+            processorJob?.cancel()
-            val queue = PriorityQueue<QueuedBroadcast>(11) { a, b ->
-                when {
-                    a.primaryPriority != b.primaryPriority -> a.primaryPriority - b.primaryPriority
-                    a.secondaryPriority != b.secondaryPriority -> a.secondaryPriority - b.secondaryPriority
-                    else -> (a.sequence - b.sequence).coerceIn(Int.MIN_VALUE.toLong(), Int.MAX_VALUE.toLong()).toInt()
-                }
-            }
             
-            try {
+            // Create new channel for fresh start
-                while (isActive) {
+            broadcasterChannel = Channel(capacity = Channel.UNLIMITED)
-                    // If queue is empty, suspend to receive at least one item
+            
-                    if (queue.isEmpty()) {
+            processorJob = broadcasterScope.launch {
-                        val first = broadcasterChannel.receiveCatching().getOrNull() ?: break
+                var processorException: Exception? = null
-                        queue.offer(computeQueuedBroadcast(first))
+                Log.d(TAG, "🎭 Priority packet broadcaster processor started")
+                
+                // Min-heap by (primaryPriority, secondaryPriority, sequence)
+                val priorityQueue = PriorityQueue<QueuedBroadcast>(11) { a, b ->
+                    when {
+                        a.primaryPriority != b.primaryPriority -> a.primaryPriority - b.primaryPriority
+                        a.secondaryPriority != b.secondaryPriority -> a.secondaryPriority - b.secondaryPriority
+                        else -> (a.sequence - b.sequence).coerceIn(Int.MIN_VALUE.toLong(), Int.MAX_VALUE.toLong()).toInt()
+                    }
+                }
+
+                try {
+                    while (isActive) {
+                        // If priority queue is empty, suspend to receive at least one item
+                        if (priorityQueue.isEmpty()) {
+                            val first = broadcasterChannel.receiveCatching().getOrNull() ?: break
+                            priorityQueue.offer(computeQueuedBroadcast(first))
+                        }
+
+                        // Drain any immediately available items without suspending
+                        while (true) {
+                            val received = broadcasterChannel.tryReceive().getOrNull() ?: break
+                            priorityQueue.offer(computeQueuedBroadcast(received))
+                        }
+
+                        // Process one highest-priority item
+                        val next = priorityQueue.poll() ?: continue
+                        try {
+                            broadcastSinglePacketInternal(next.request.routed, next.request.gattServer, next.request.characteristic)
+                        } catch (e: Exception) {
+                            // Log the broadcast failure - this is likely a transient BLE error
+                            Log.e(TAG, "❌ Broadcast failed, will attempt recovery: ${e.message}", e)
+                            processorException = e
+                            // Don't throw - instead break and trigger recovery
+                            break
+                        }
+                    }
+                } catch (e: Exception) {
+                    Log.e(TAG, "❌ Priority processor loop terminated due to exception: ${e.message}", e)
+                    processorException = e
+                } finally {
+                    // Close the current channel
+                    broadcasterChannel.close(processorException)
+                    Log.w(TAG, "🎭 Priority packet broadcaster processor terminated, attempting recovery...")
+                    
+                    // Schedule automatic recovery after a short delay
+                    if (processorException != null && broadcasterScope.isActive) {
+                        broadcasterScope.launch {
+                            delay(1000) // Wait 1 second before recovery
+                            Log.d(TAG, "🔄 Attempting to restart priority processor...")
+                            startPriorityProcessor()
+                        }
+                    } else {
+                        Log.d(TAG, "🎭 Priority packet broadcaster processor shut down (no recovery scheduled)")
                     }
-
-                    // Drain any immediately available items without suspending
-                    while (true) {
-                        val received = broadcasterChannel.tryReceive().getOrNull() ?: break
-                        queue.offer(computeQueuedBroadcast(received))
-                    }
-
-                    // Process one highest-priority item
-                    val next = queue.poll() ?: continue
-                    broadcastSinglePacketInternal(next.request.routed, next.request.gattServer, next.request.characteristic)
                 }
-            } catch (e: Exception) {
-                Log.w(TAG, "Priority processor loop ended: ${'$'}{e.message}")
-            } finally {
-                Log.d(TAG, "🎭 Priority packet broadcaster processor terminated")
             }
         }
     }