Fix courier BLE custody and broadcast clock skew

2026-10-25 07:40:36 +00:00 · 2026-08-13 17:08:18 +03:00 · 2026-08-13 17:08:18 +03:00 · d83fb82aae
commit d83fb82aae
parent a09e0cbcd7
10 changed files with 267 additions and 15 deletions
--- a/app/src/main/java/com/bitchat/android/mesh/BluetoothConnectionManager.kt
+++ b/app/src/main/java/com/bitchat/android/mesh/BluetoothConnectionManager.kt
@ -377,6 +377,16 @@ class BluetoothConnectionManager(
        )
    }

+    suspend fun sendToPeerAndAwaitCompletion(peerID: String, routed: RoutedPacket): Boolean {
+        if (!isActive || !isBleTransportEnabled()) return false
+        return packetBroadcaster.sendPacketToPeerAndAwaitCompletion(
+            routed,
+            peerID,
+            serverManager.getGattServer(),
+            serverManager.getCharacteristic()
+        )
+    }
+
    fun cancelTransfer(transferId: String): Boolean {
        return packetBroadcaster.cancelTransfer(transferId)
    }
--- a/app/src/main/java/com/bitchat/android/mesh/BluetoothMeshService.kt
+++ b/app/src/main/java/com/bitchat/android/mesh/BluetoothMeshService.kt
@ -1620,7 +1620,7 @@ class BluetoothMeshService(private val context: Context) : TransportBridgeServic
            )
            if (direct) {
                serviceScope.launch {
-                    if (connectionManager.sendToPeerAndAwaitAcceptance(
+                    if (connectionManager.sendToPeerAndAwaitCompletion(
                            peerID,
                            RoutedPacket(signPacketBeforeBroadcast(packet))
                        )
@ -1641,7 +1641,7 @@ class BluetoothMeshService(private val context: Context) : TransportBridgeServic
                    ttl = MAX_TTL
                )
                serviceScope.launch {
-                    if (connectionManager.sendToPeerAndAwaitAcceptance(
+                    if (connectionManager.sendToPeerAndAwaitCompletion(
                            peerID,
                            RoutedPacket(signPacketBeforeBroadcast(packet))
                        )
--- a/app/src/main/java/com/bitchat/android/mesh/BluetoothPacketBroadcaster.kt
+++ b/app/src/main/java/com/bitchat/android/mesh/BluetoothPacketBroadcaster.kt
@ -22,6 +22,7 @@ import kotlinx.coroutines.cancel
 import kotlinx.coroutines.delay
 import kotlinx.coroutines.isActive
 import kotlinx.coroutines.launch
+import kotlinx.coroutines.withTimeoutOrNull
 import kotlinx.coroutines.channels.actor
 import java.util.ArrayDeque

@ -55,6 +56,7 @@ class BluetoothPacketBroadcaster(
        private const val MAX_PENDING_BYTES_PER_LINK = 1_048_576
        private const val SEND_RETRY_DELAY_MS = 15L
        private const val MAX_CALLBACK_RETRIES = 3
+        private const val SEND_COMPLETION_TIMEOUT_MS = 30_000L
    }

    // Optional nickname resolver injected by higher layer (peerID -> nickname?)
@ -140,6 +142,7 @@ class BluetoothPacketBroadcaster(
        val gatt: BluetoothGatt? = null,
        val gattServer: BluetoothGattServer? = null,
        val characteristic: BluetoothGattCharacteristic,
+        val completion: CompletableDeferred<Boolean>? = null,
        var callbackFailures: Int = 0
    )

@ -216,6 +219,18 @@ class BluetoothPacketBroadcaster(
        }
    }

+    suspend fun sendPacketToPeerAndAwaitCompletion(
+        routed: RoutedPacket,
+        targetPeerID: String,
+        gattServer: BluetoothGattServer?,
+        characteristic: BluetoothGattCharacteristic?
+    ): Boolean {
+        if (!hasPeerConnection(targetPeerID)) return false
+        return fragmentingSender.sendAndAwaitAcceptance(routed, "BLE peer ${targetPeerID.take(8)}") { packet ->
+            sendSinglePacketToPeerAndAwaitCompletion(packet, targetPeerID, gattServer, characteristic)
+        }
+    }
+
    fun sendPacketToLink(
        routed: RoutedPacket,
        deviceAddress: String,
@ -279,6 +294,35 @@ class BluetoothPacketBroadcaster(
        return false
    }

+    private suspend fun sendSinglePacketToPeerAndAwaitCompletion(
+        routed: RoutedPacket,
+        targetPeerID: String,
+        gattServer: BluetoothGattServer?,
+        characteristic: BluetoothGattCharacteristic?
+    ): Boolean {
+        val packet = routed.packet
+        val data = packet.toBinaryData(
+            padding = BLEPacketPaddingPolicy.shouldPadForBLE(packet.type)
+        ) ?: return false
+        val completion = CompletableDeferred<Boolean>()
+
+        val serverTarget = connectionTracker.getSubscribedDevices()
+            .firstOrNull { connectionTracker.addressPeerMap[it.address] == targetPeerID }
+        val queuedOnServer = serverTarget != null &&
+            notifyDevice(serverTarget, data, gattServer, characteristic, completion)
+        val queued = if (queuedOnServer) {
+            true
+        } else {
+            val clientTarget = connectionTracker.getConnectedDevices().values
+                .firstOrNull { connectionTracker.addressPeerMap[it.device.address] == targetPeerID }
+                ?: return false
+            writeToDeviceConn(clientTarget, data, completion)
+        }
+        if (!queued) return false
+
+        return withTimeoutOrNull(SEND_COMPLETION_TIMEOUT_MS) { completion.await() } ?: false
+    }
+
    
    /**
     * Public entry point for broadcasting - submits request to actor for serialization
@ -486,7 +530,8 @@ class BluetoothPacketBroadcaster(
        device: BluetoothDevice, 
        data: ByteArray,
        gattServer: BluetoothGattServer?,
-        characteristic: BluetoothGattCharacteristic?
+        characteristic: BluetoothGattCharacteristic?,
+        completion: CompletableDeferred<Boolean>? = null
    ): Boolean {
        val server = gattServer ?: return false
        val char = characteristic ?: return false
@ -496,7 +541,13 @@ class BluetoothPacketBroadcaster(
            ?: return false
        return enqueueSend(
            SendKey(device.address, linkID, SendDirection.SERVER_NOTIFICATION),
-            PendingSend(data.copyOf(), device, gattServer = server, characteristic = char)
+            PendingSend(
+                data.copyOf(),
+                device,
+                gattServer = server,
+                characteristic = char,
+                completion = completion
+            )
        )
    }

@ -505,13 +556,20 @@ class BluetoothPacketBroadcaster(
     */
    private fun writeToDeviceConn(
        deviceConn: BluetoothConnectionTracker.DeviceConnection, 
-        data: ByteArray
+        data: ByteArray,
+        completion: CompletableDeferred<Boolean>? = null
    ): Boolean {
        val gatt = deviceConn.gatt ?: return false
        val char = deviceConn.characteristic ?: return false
        return enqueueSend(
            SendKey(deviceConn.device.address, deviceConn.linkID, SendDirection.CLIENT_WRITE),
-            PendingSend(data.copyOf(), deviceConn.device, gatt = gatt, characteristic = char)
+            PendingSend(
+                data.copyOf(),
+                deviceConn.device,
+                gatt = gatt,
+                characteristic = char,
+                completion = completion
+            )
        )
    }

@ -643,6 +701,7 @@ class BluetoothPacketBroadcaster(
                }
                state.pending.removeFirst()
                state.pendingBytes -= head.data.size
+                head.completion?.complete(status == BluetoothGatt.GATT_SUCCESS)
                if (state.pending.isEmpty()) {
                    sendStates.remove(key)
                    false
@ -657,8 +716,13 @@ class BluetoothPacketBroadcaster(

    fun onLinkDisconnected(deviceAddress: String, linkID: String?) {
        synchronized(sendLock) {
-            sendStates.keys.removeAll { key ->
-                key.deviceAddress == deviceAddress && (linkID == null || key.linkID == linkID)
+            val iterator = sendStates.entries.iterator()
+            while (iterator.hasNext()) {
+                val (key, state) = iterator.next()
+                if (key.deviceAddress == deviceAddress && (linkID == null || key.linkID == linkID)) {
+                    state.pending.forEach { it.completion?.complete(false) }
+                    iterator.remove()
+                }
            }
        }
    }
@ -679,7 +743,12 @@ class BluetoothPacketBroadcaster(
     * Shutdown the broadcaster actor gracefully
     */
    fun shutdown() {
-        synchronized(sendLock) { sendStates.clear() }
+        synchronized(sendLock) {
+            sendStates.values.forEach { state ->
+                state.pending.forEach { it.completion?.complete(false) }
+            }
+            sendStates.clear()
+        }
        // Close the actor gracefully
        broadcasterActor.close()

--- a/app/src/main/java/com/bitchat/android/mesh/FragmentingPacketSender.kt
+++ b/app/src/main/java/com/bitchat/android/mesh/FragmentingPacketSender.kt
@ -115,7 +115,7 @@ class FragmentingPacketSender(
    suspend fun sendAndAwaitAcceptance(
        routed: RoutedPacket,
        description: String,
-        sendSingle: (RoutedPacket) -> Boolean
+        sendSingle: suspend (RoutedPacket) -> Boolean
    ): Boolean {
        val packets = packetsForTransport(routed) ?: return false
        Log.d(logTag, "Sending ${packets.size} packet(s) for $description")
--- a/app/src/main/java/com/bitchat/android/mesh/MessageHandler.kt
+++ b/app/src/main/java/com/bitchat/android/mesh/MessageHandler.kt
@ -481,7 +481,10 @@ class MessageHandler(private val myPeerID: String, private val appContext: andro
        val peerID = routed.peerID ?: "unknown"
        if (packet.timestamp > Long.MAX_VALUE.toULong()) return
        val ageMs = System.currentTimeMillis() - packet.timestamp.toLong()
-        if (ageMs !in 0..com.bitchat.android.sync.GossipSyncManager.PUBLIC_MESSAGE_MAX_AGE_MS) return
+        if (ageMs !in
+            -com.bitchat.android.sync.GossipSyncManager.PUBLIC_PACKET_FUTURE_SKEW_MS..
+                com.bitchat.android.sync.GossipSyncManager.PUBLIC_MESSAGE_MAX_AGE_MS
+        ) return
        
        // Enforce: only accept public messages from verified peers we know
        val peerInfo = delegate?.getPeerInfo(peerID)
--- a/app/src/main/java/com/bitchat/android/mesh/PacketProcessor.kt
+++ b/app/src/main/java/com/bitchat/android/mesh/PacketProcessor.kt
@ -132,7 +132,9 @@ class PacketProcessor(private val myPeerID: String) {
                    com.bitchat.android.sync.GossipSyncManager.FRAGMENT_MAX_AGE_MS
                else -> null
            }
-            if (maxAgeMs != null && ageMs !in 0..maxAgeMs) return
+            if (maxAgeMs != null && ageMs !in
+                -com.bitchat.android.sync.GossipSyncManager.PUBLIC_PACKET_FUTURE_SKEW_MS..maxAgeMs
+            ) return
        } else if (isBroadcast && packet.timestamp > Long.MAX_VALUE.toULong()) {
            return
        }
--- a/app/src/main/java/com/bitchat/android/sync/GossipSyncManager.kt
+++ b/app/src/main/java/com/bitchat/android/sync/GossipSyncManager.kt
@ -39,6 +39,7 @@ class GossipSyncManager(
        private const val TAG = "GossipSyncManager"
        const val PUBLIC_MESSAGE_MAX_AGE_MS = 6 * 60 * 60 * 1000L
        const val FRAGMENT_MAX_AGE_MS = 15 * 60 * 1000L
+        const val PUBLIC_PACKET_FUTURE_SKEW_MS = 10 * 60 * 1000L
        private const val ARCHIVE_FILE = "gossip-public-history.bin"
    }

@ -132,7 +133,7 @@ class GossipSyncManager(
        if (isBroadcastMessage) {
            val now = System.currentTimeMillis()
            val age = now - packet.timestamp.toLong()
-            if (age !in 0..PUBLIC_MESSAGE_MAX_AGE_MS) return
+            if (age !in -PUBLIC_PACKET_FUTURE_SKEW_MS..PUBLIC_MESSAGE_MAX_AGE_MS) return
            synchronized(messages) {
                messages[id] = packet
                // Enforce capacity (remove oldest when exceeded)
@ -145,7 +146,7 @@ class GossipSyncManager(
            if (!restoringArchive) persistArchive()
        } else if (isFragment) {
            val age = System.currentTimeMillis() - packet.timestamp.toLong()
-            if (age !in 0..FRAGMENT_MAX_AGE_MS) return
+            if (age !in -PUBLIC_PACKET_FUTURE_SKEW_MS..FRAGMENT_MAX_AGE_MS) return
            synchronized(fragments) {
                fragments[id] = packet
                while (fragments.size > configProvider.seenCapacity().coerceAtLeast(1)) {
--- a/app/src/test/kotlin/com/bitchat/android/mesh/BluetoothPacketBroadcasterCompletionTest.kt
+++ b/app/src/test/kotlin/com/bitchat/android/mesh/BluetoothPacketBroadcasterCompletionTest.kt
@ -0,0 +1,110 @@
+package com.bitchat.android.mesh
+
+import android.bluetooth.BluetoothDevice
+import android.bluetooth.BluetoothGatt
+import android.bluetooth.BluetoothGattCharacteristic
+import android.os.Build
+import com.bitchat.android.model.RoutedPacket
+import com.bitchat.android.protocol.BitchatPacket
+import com.bitchat.android.protocol.MessageType
+import kotlinx.coroutines.CoroutineScope
+import kotlinx.coroutines.Dispatchers
+import kotlinx.coroutines.SupervisorJob
+import kotlinx.coroutines.async
+import kotlinx.coroutines.cancel
+import kotlinx.coroutines.runBlocking
+import org.junit.After
+import org.junit.Assert.assertFalse
+import org.junit.Assert.assertTrue
+import org.junit.Test
+import org.junit.runner.RunWith
+import org.mockito.kotlin.any
+import org.mockito.kotlin.mock
+import org.mockito.kotlin.timeout
+import org.mockito.kotlin.verify
+import org.mockito.kotlin.whenever
+import org.robolectric.RobolectricTestRunner
+import org.robolectric.annotation.Config
+
+@RunWith(RobolectricTestRunner::class)
+@Config(sdk = [Build.VERSION_CODES.P], manifest = Config.NONE)
+class BluetoothPacketBroadcasterCompletionTest {
+    private val scope = CoroutineScope(Dispatchers.Default + SupervisorJob())
+    private val tracker = BluetoothConnectionTracker(scope, mock())
+    private val broadcaster = BluetoothPacketBroadcaster(scope, tracker, null, MY_PEER_ID)
+
+    @After
+    fun tearDown() {
+        broadcaster.shutdown()
+        scope.cancel()
+    }
+
+    @Test
+    fun `queue admission waits for successful GATT completion`() = runBlocking {
+        val connection = connectedPeer()
+
+        val result = async(Dispatchers.Default) {
+            broadcaster.sendPacketToPeerAndAwaitCompletion(packet(), PEER_ID, null, null)
+        }
+
+        verify(connection.gatt!!, timeout(1_000)).writeCharacteristic(connection.characteristic!!)
+        assertFalse(result.isCompleted)
+
+        broadcaster.onGattClientWriteComplete(DEVICE_ADDRESS, LINK_ID, BluetoothGatt.GATT_SUCCESS)
+        assertTrue(result.await())
+    }
+
+    @Test
+    fun `disconnect fails an admitted send before custody can be released`() = runBlocking {
+        val connection = connectedPeer()
+
+        val result = async(Dispatchers.Default) {
+            broadcaster.sendPacketToPeerAndAwaitCompletion(packet(), PEER_ID, null, null)
+        }
+
+        verify(connection.gatt!!, timeout(1_000)).writeCharacteristic(connection.characteristic!!)
+        assertFalse(result.isCompleted)
+
+        broadcaster.onLinkDisconnected(DEVICE_ADDRESS, LINK_ID)
+        assertFalse(result.await())
+    }
+
+    private fun connectedPeer(): BluetoothConnectionTracker.DeviceConnection {
+        val device = mock<BluetoothDevice>()
+        val gatt = mock<BluetoothGatt>()
+        val characteristic = mock<BluetoothGattCharacteristic>()
+        whenever(device.address).thenReturn(DEVICE_ADDRESS)
+        whenever(gatt.writeCharacteristic(any())).thenReturn(true)
+        val connection = BluetoothConnectionTracker.DeviceConnection(
+            device = device,
+            gatt = gatt,
+            characteristic = characteristic,
+            isClient = true,
+            linkID = LINK_ID
+        )
+        tracker.addDeviceConnection(DEVICE_ADDRESS, connection)
+        tracker.observePeerIfCurrent(DEVICE_ADDRESS, LINK_ID, PEER_ID)
+        return connection
+    }
+
+    private fun packet() = RoutedPacket(
+        BitchatPacket(
+            type = MessageType.COURIER_ENVELOPE.value,
+            senderID = MY_PEER_ID.hexToBytes(),
+            recipientID = PEER_ID.hexToBytes(),
+            timestamp = System.currentTimeMillis().toULong(),
+            payload = byteArrayOf(1, 2, 3),
+            ttl = 7u
+        )
+    )
+
+    private fun String.hexToBytes(): ByteArray =
+        chunked(2).map { it.toInt(16).toByte() }.toByteArray()
+
+    private companion object {
+        const val MY_PEER_ID = "1111222233334444"
+        const val PEER_ID = "aaaabbbbccccdddd"
+        const val DEVICE_ADDRESS = "00:11:22:33:44:55"
+        const val LINK_ID = "synthetic-link"
+    }
+}
--- a/app/src/test/kotlin/com/bitchat/android/mesh/PacketProcessorAnnounceSideEffectTest.kt
+++ b/app/src/test/kotlin/com/bitchat/android/mesh/PacketProcessorAnnounceSideEffectTest.kt
@ -64,6 +64,26 @@ class PacketProcessorAnnounceSideEffectTest {
        assertEquals(PEER_ID, withTimeout(1_000) { delegate.lastSeen.await() })
    }

+    @Test
+    fun `broadcast message within future skew is handled`() = runBlocking {
+        val delegate = RecordingDelegate(acceptAnnounce = true)
+        val processor = processor(delegate)
+
+        processor.processPacket(message(timestampOffsetMs = 60_000))
+
+        withTimeout(1_000) { delegate.messageHandled.await() }
+    }
+
+    @Test
+    fun `broadcast message beyond future skew is rejected`() = runBlocking {
+        val delegate = RecordingDelegate(acceptAnnounce = true)
+        val processor = processor(delegate)
+
+        processor.processPacket(message(timestampOffsetMs = 11 * 60_000L))
+
+        assertNull(withTimeoutOrNull(250) { delegate.messageHandled.await() })
+    }
+
    private fun processor(delegate: RecordingDelegate): PacketProcessor =
        PacketProcessor(MY_PEER_ID).also {
            it.delegate = delegate
@ -96,12 +116,26 @@ class PacketProcessorAnnounceSideEffectTest {
        return RoutedPacket(packet, PEER_ID, "direct-link")
    }

+    private fun message(timestampOffsetMs: Long): RoutedPacket {
+        val packet = BitchatPacket(
+            version = 1u,
+            type = MessageType.MESSAGE.value,
+            senderID = PEER_ID.hexToBytes(),
+            recipientID = SpecialRecipients.BROADCAST,
+            timestamp = (System.currentTimeMillis() + timestampOffsetMs).toULong(),
+            payload = byteArrayOf(0x01),
+            ttl = 7u
+        )
+        return RoutedPacket(packet, PEER_ID, "direct-link")
+    }
+
    private class RecordingDelegate(
        private val acceptAnnounce: Boolean,
        private val acceptHandshake: Boolean = false
    ) : PacketProcessorDelegate {
        val handled = CompletableDeferred<Unit>()
        val handshakeHandled = CompletableDeferred<Unit>()
+        val messageHandled = CompletableDeferred<Unit>()
        val lastSeen = CompletableDeferred<String>()
        @Volatile var relayCount = 0

@ -121,7 +155,9 @@ class PacketProcessorAnnounceSideEffectTest {
            handled.complete(Unit)
            return acceptAnnounce
        }
-        override fun handleMessage(routed: RoutedPacket) = Unit
+        override fun handleMessage(routed: RoutedPacket) {
+            messageHandled.complete(Unit)
+        }
        override fun handleLeave(routed: RoutedPacket) = Unit
        override fun handleFragment(packet: BitchatPacket): BitchatPacket? = null
        override fun handleRequestSync(routed: RoutedPacket) = Unit
--- a/app/src/test/kotlin/com/bitchat/android/sync/GossipSyncManagerTest.kt
+++ b/app/src/test/kotlin/com/bitchat/android/sync/GossipSyncManagerTest.kt
@ -41,6 +41,27 @@ class GossipSyncManagerTest {
        assertEquals(listOf(1, 3, 5), sent.map { it.payload.single().toInt() })
    }

+    @Test
+    fun `public history tolerates bounded future clock skew`() {
+        val sent = mutableListOf<BitchatPacket>()
+        val manager = GossipSyncManager("1111222233334444", TestScope(), config)
+        manager.delegate = object : GossipSyncManager.Delegate {
+            override fun sendPacket(packet: BitchatPacket) = Unit
+            override fun sendPacketToPeer(peerID: String, packet: BitchatPacket) { sent += packet }
+            override fun signPacketForBroadcast(packet: BitchatPacket) = packet
+        }
+        val now = System.currentTimeMillis()
+        manager.onPublicPacketSeen(packet(MessageType.MESSAGE, now + 60_000L, 1))
+        manager.onPublicPacketSeen(packet(MessageType.FRAGMENT, now + 60_000L, 2))
+        manager.onPublicPacketSeen(packet(MessageType.FILE_TRANSFER, now + 60_000L, 3))
+        manager.onPublicPacketSeen(packet(MessageType.MESSAGE, now + 11 * 60_000L, 4))
+        manager.onPublicPacketSeen(packet(MessageType.FRAGMENT, now + 11 * 60_000L, 5))
+
+        manager.handleRequestSync("peer", RequestSyncPacket(p = 1, m = 1, data = byteArrayOf()))
+
+        assertEquals(listOf(1, 2, 3), sent.map { it.payload.single().toInt() })
+    }
+
    private fun packet(type: MessageType, timestamp: Long, marker: Int) = BitchatPacket(
        type = type.value,
        senderID = ByteArray(8) { 1 },