Gossip routing tmp with connection limit fixed (#569)

* fix: r8 exception for LocationManager (#566) * fragmented packets inherit route * update spec * fix connection limits * fix: deserialization issue with routed packets * log * dynamic fragment size * fragment size * add tests
2026-07-25 10:05:21 +00:00 · 2026-01-09 22:24:07 +07:00
parent a74e587123
commit 856fe5eb3e
9 changed files with 444 additions and 135 deletions
@@ -109,11 +109,19 @@ class BluetoothConnectionManager(
            }
            // Connection caps: enforce on change
            connectionScope.launch {
-                dbg.maxConnectionsOverall.collect {
+                dbg.maxConnectionsOverall.collect { maxOverall ->
                    if (!isActive) return@collect
+                    // 1. Enforce client limits (handled by tracker)
                    connectionTracker.enforceConnectionLimits()
-                    // Also enforce server side best-effort
-                    serverManager.enforceServerLimit(dbg.maxServerConnections.value)
+                    
+                    // 2. Enforce overall limit on server connections if needed
+                    // (Tracker knows about all connections but can't disconnect servers directly)
+                    val maxServer = dbg.maxServerConnections.value
+                    val excessServers = connectionTracker.getExcessServerConnections(maxServer, maxOverall)
+                    excessServers.forEach { device ->
+                        Log.d(TAG, "Disconnecting server ${device.address} due to overall cap")
+                        serverManager.disconnectDevice(device)
+                    }
                }
            }
            connectionScope.launch {
@@ -123,9 +131,18 @@ class BluetoothConnectionManager(
                }
            }
            connectionScope.launch {
-                dbg.maxServerConnections.collect {
+                dbg.maxServerConnections.collect { maxServer ->
                    if (!isActive) return@collect
-                    serverManager.enforceServerLimit(dbg.maxServerConnections.value)
+                    // Enforce server specific limit
+                    serverManager.enforceServerLimit(maxServer)
+                    
+                    // Also check if this change puts us over the overall limit
+                    val maxOverall = dbg.maxConnectionsOverall.value
+                    val excessServers = connectionTracker.getExcessServerConnections(maxServer, maxOverall)
+                    excessServers.forEach { device ->
+                        Log.d(TAG, "Disconnecting server ${device.address} due to overall cap")
+                        serverManager.disconnectDevice(device)
+                    }
                }
            }
        } catch (_: Exception) { }
@@ -229,11 +229,17 @@ class BluetoothConnectionTracker(
     */
    fun getConnectedDeviceCount(): Int = connectedDevices.size
    
+    /**
+     * Check if connection limit is reached
+     */
    /**
     * Check if connection limit is reached
     */
    fun isConnectionLimitReached(): Boolean {
-        return connectedDevices.size >= powerManager.getMaxConnections()
+        // Respect debug override if set
+        val dbg = try { com.bitchat.android.ui.debug.DebugSettingsManager.getInstance() } catch (_: Exception) { null }
+        val maxConnections = dbg?.maxConnectionsOverall?.value ?: powerManager.getMaxConnections()
+        return connectedDevices.size >= maxConnections
    }
    
    /**
@@ -244,10 +250,9 @@ class BluetoothConnectionTracker(
        val dbg = try { com.bitchat.android.ui.debug.DebugSettingsManager.getInstance() } catch (_: Exception) { null }
        val maxOverall = dbg?.maxConnectionsOverall?.value ?: powerManager.getMaxConnections()
        val maxClient = dbg?.maxClientConnections?.value ?: maxOverall
-        val maxServer = dbg?.maxServerConnections?.value ?: maxOverall
+        // Note: maxServer is handled by GattServerManager, but we need to respect overall limit here too

        val clients = connectedDevices.values.filter { it.isClient }
-        val servers = connectedDevices.values.filter { !it.isClient }

        // Enforce client cap first (we can actively disconnect)
        if (clients.size > maxClient) {
@@ -259,22 +264,56 @@ class BluetoothConnectionTracker(
            }
        }

-        // Note: server cap enforced in GattServerManager (we don't have server handle here)
-
-        // Enforce overall cap by disconnecting oldest client connections
+        // Re-check overall cap after client cleanup
        if (connectedDevices.size > maxOverall) {
            Log.i(TAG, "Enforcing overall cap: ${connectedDevices.size} > $maxOverall")
            val excess = connectedDevices.size - maxOverall
-            val toDisconnect = connectedDevices.values
-                .filter { it.isClient } // only clients from here
+            
+            // Prefer disconnecting clients first to satisfy overall cap
+            val clientsToDisconnect = connectedDevices.values
+                .filter { it.isClient }
                .sortedBy { it.connectedAt }
                .take(excess)
-            toDisconnect.forEach { dc ->
+                
+            clientsToDisconnect.forEach { dc ->
                Log.d(TAG, "Disconnecting client ${dc.device.address} due to overall cap")
                dc.gatt?.disconnect()
            }
        }
    }
+
+    /**
+     * Get excess server connections that should be disconnected to satisfy limits.
+     * This allows the Manager to coordinate server disconnects since Tracker doesn't control the server.
+     */
+    fun getExcessServerConnections(maxServer: Int, maxOverall: Int): List<BluetoothDevice> {
+        val servers = connectedDevices.values.filter { !it.isClient }
+        val excessList = mutableListOf<BluetoothDevice>()
+
+        // 1. Check server specific limit
+        if (servers.size > maxServer) {
+            val excessCount = servers.size - maxServer
+            val toRemove = servers.sortedBy { it.connectedAt }.take(excessCount)
+            excessList.addAll(toRemove.map { it.device })
+        }
+
+        // 2. Check overall limit (considering we might have already removed some above)
+        // We need to count how many connections we will have after the above removals
+        val currentTotal = connectedDevices.size
+        val plannedRemovals = excessList.size
+        val projectedTotal = currentTotal - plannedRemovals
+        
+        if (projectedTotal > maxOverall) {
+            val furtherExcess = projectedTotal - maxOverall
+            // We can only remove servers here. Clients are handled in enforceConnectionLimits.
+            // Filter out devices we already planned to remove
+            val remainingServers = servers.filter { s -> excessList.none { it.address == s.device.address } }
+            val toRemove = remainingServers.sortedBy { it.connectedAt }.take(furtherExcess)
+            excessList.addAll(toRemove.map { it.device })
+        }
+
+        return excessList
+    }
    
    /**
     * Clean up a specific device connection
@@ -49,15 +49,28 @@ class BluetoothGattServerManager(
    fun enforceServerLimit(maxServer: Int) {
        if (maxServer <= 0) return
        try {
-            val subs = connectionTracker.getSubscribedDevices()
-            if (subs.size > maxServer) {
-                val excess = subs.size - maxServer
-                subs.take(excess).forEach { d ->
-                    try { gattServer?.cancelConnection(d) } catch (_: Exception) { }
+            // Use connection tracker to get actual connected server devices
+            val servers = connectionTracker.getConnectedDevices().values.filter { !it.isClient }
+            if (servers.size > maxServer) {
+                val excess = servers.size - maxServer
+                // Disconnect oldest
+                servers.sortedBy { it.connectedAt }.take(excess).forEach { d ->
+                    try { gattServer?.cancelConnection(d.device) } catch (_: Exception) { }
                }
            }
        } catch (_: Exception) { }
    }
+
+    /**
+     * Disconnect a specific device (used by ConnectionManager to enforce overall limits)
+     */
+    fun disconnectDevice(device: BluetoothDevice) {
+        try {
+            gattServer?.cancelConnection(device)
+        } catch (e: Exception) {
+            Log.w(TAG, "Error disconnecting device ${device.address}: ${e.message}")
+        }
+    }
    
    /**
     * Start GATT server
@@ -122,9 +135,10 @@ class BluetoothGattServerManager(
            
            // Try to cancel any active connections explicitly before closing
            try {
-                val devices = connectionTracker.getSubscribedDevices()
-                devices.forEach { d ->
-                    try { gattServer?.cancelConnection(d) } catch (_: Exception) { }
+                // Disconnect ALL server connections
+                val servers = connectionTracker.getConnectedDevices().values.filter { !it.isClient }
+                servers.forEach { d ->
+                    try { gattServer?.cancelConnection(d.device) } catch (_: Exception) { }
                }
            } catch (_: Exception) { }
            
@@ -378,7 +378,7 @@ class BluetoothPacketBroadcaster(
        val subscribedDevices = connectionTracker.getSubscribedDevices()
        val connectedDevices = connectionTracker.getConnectedDevices()
        
-        Log.i(TAG, "Broadcasting packet type ${packet.type} to ${subscribedDevices.size} server + ${connectedDevices.size} client connections")
+        Log.i(TAG, "Broadcasting packet v${packet.version} type ${packet.type} to ${subscribedDevices.size} server + ${connectedDevices.size} client connections")

        val senderID = String(packet.senderID).replace("\u0000", "")        
        
@@ -77,8 +77,31 @@ class FragmentManager {
        val fragmentID = FragmentPayload.generateFragmentID()
        
        // iOS: stride(from: 0, to: fullData.count, by: maxFragmentSize)
-        val fragmentChunks = stride(0, fullData.size, MAX_FRAGMENT_SIZE) { offset ->
-            val endOffset = minOf(offset + MAX_FRAGMENT_SIZE, fullData.size)
+        // Calculate dynamic fragment size to fit in MTU (512)
+        // Packet = Header + Sender + Recipient + Route + FragmentHeader + Payload + PaddingBuffer
+        val hasRoute = packet.route != null
+        val version = if (hasRoute) 2 else 1
+        val headerSize = if (version == 2) 15 else 13
+        val senderSize = 8
+        val recipientSize = if (packet.recipientID != null) 8 else 0
+        // Route: 1 byte count + 8 bytes per hop
+        val routeSize = if (hasRoute) (1 + (packet.route?.size ?: 0) * 8) else 0
+        val fragmentHeaderSize = 13 // FragmentPayload header
+        val paddingBuffer = 16 // MessagePadding.optimalBlockSize adds 16 bytes overhead
+
+        // 512 - Overhead
+        val packetOverhead = headerSize + senderSize + recipientSize + routeSize + fragmentHeaderSize + paddingBuffer
+        val maxDataSize = (512 - packetOverhead).coerceAtMost(MAX_FRAGMENT_SIZE)
+        
+        if (maxDataSize <= 0) {
+            Log.e(TAG, "❌ Calculated maxDataSize is non-positive ($maxDataSize). Route too large?")
+            return emptyList()
+        }
+
+        Log.d(TAG, "📏 Dynamic fragment size: $maxDataSize (MAX: $MAX_FRAGMENT_SIZE, Overhead: $packetOverhead)")
+
+        val fragmentChunks = stride(0, fullData.size, maxDataSize) { offset ->
+            val endOffset = minOf(offset + maxDataSize, fullData.size)
            fullData.sliceArray(offset..<endOffset)
        }
        
@@ -98,13 +121,16 @@ class FragmentManager {
            )
            
            // iOS: MessageType.fragment.rawValue (single fragment type)
+            // Fix: Fragments must inherit source route and use v2 if routed
            val fragmentPacket = BitchatPacket(
+                version = if (packet.route != null) 2u else 1u,
                type = MessageType.FRAGMENT.value,
                ttl = packet.ttl,
                senderID = packet.senderID,
                recipientID = packet.recipientID,
                timestamp = packet.timestamp,
                payload = fragmentPayload.encode(),
+                route = packet.route,
                signature = null // iOS: signature: nil
            )
            
@@ -219,7 +219,10 @@ object BinaryProtocol {
            val signatureBytes = if (packet.signature != null) SIGNATURE_SIZE else 0
            val sizeFieldBytes = if (isCompressed) (if (packet.version >= 2u.toUByte()) 4 else 2) else 0
            val payloadBytes = payload.size + sizeFieldBytes
-            val capacity = headerSize + SENDER_ID_SIZE + recipientBytes + payloadBytes + signatureBytes + 16 // small slack
+            val routeBytes = if (!packet.route.isNullOrEmpty() && packet.version >= 2u.toUByte()) {
+                1 + (packet.route!!.size.coerceAtMost(255) * SENDER_ID_SIZE)
+            } else 0
+            val capacity = headerSize + SENDER_ID_SIZE + recipientBytes + payloadBytes + signatureBytes + routeBytes + 16 // small slack
            val buffer = ByteBuffer.allocate(capacity.coerceAtLeast(512)).apply { order(ByteOrder.BIG_ENDIAN) }
            
            // Header
@@ -368,9 +371,16 @@ object BinaryProtocol {
            var routeCount = 0
            if (hasRoute) {
                // Peek count (1 byte) without consuming buffer for now
-                val mark = buffer.position()
-                if (raw.size >= mark + 1) {
-                    routeCount = raw[mark].toUByte().toInt()
+                // The buffer is currently positioned at the start of SenderID (after fixed header)
+                // We must skip SenderID and RecipientID (if present) to find the route count
+                val currentPos = buffer.position()
+                var routeOffset = currentPos + SENDER_ID_SIZE
+                if (hasRecipient) {
+                    routeOffset += RECIPIENT_ID_SIZE
+                }
+
+                if (raw.size >= routeOffset + 1) {
+                    routeCount = raw[routeOffset].toUByte().toInt()
                }
                expectedSize += 1 + (routeCount * SENDER_ID_SIZE)
            }