Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -34,6 +34,10 @@ class BluetoothGattServerManager(
// Self-healing advertising recovery tuning
private const val ADVERTISE_RETRY_BASE_MS = 3_000L // base backoff for transient advertise failures
private const val ADVERTISE_MAX_RETRY_DELAY_MS = 30_000L // cap on backoff delay
// ATT error code 0x09 "Prepare Queue Full" — Android's BluetoothGatt has no
// constant for it (it exposes GATT_INVALID_OFFSET but not this), so it is
// defined here. Returned when a prepared-write chunk can't be buffered.
private const val ATT_ERROR_PREPARE_QUEUE_FULL = 0x09
}

// Core Bluetooth components
Expand All @@ -52,6 +56,11 @@ class BluetoothGattServerManager(
// State management
private var isActive = false

// Reassembles BLE prepared ("reliable"/long) writes. iOS uses these to send
// packets larger than the negotiated ATT MTU: the payload arrives as several
// preparedWrite chunks and is finalized by onExecuteWrite.
private val preparedWriteBuffer = GattPreparedWriteBuffer()

private fun isBleTransportEnabled(): Boolean {
return try {
com.bitchat.android.ui.debug.DebugSettingsManager.getInstance().bleEnabled.value
Expand Down Expand Up @@ -119,6 +128,10 @@ class BluetoothGattServerManager(
* Stop GATT server
*/
fun stop() {
// Drop any half-finished prepared writes. A disconnect after the server is marked
// inactive returns early before the per-device cancel path, so clear them here.
preparedWriteBuffer.clearAll()

if (!isActive) {
// Idempotent stop
stopAdvertising()
Expand Down Expand Up @@ -161,6 +174,24 @@ class BluetoothGattServerManager(
* Get characteristic instance
*/
fun getCharacteristic(): BluetoothGattCharacteristic? = characteristic

/**
* Parse a fully-received payload and hand it to the delegate. Shared by the
* non-prepared write path and the reassembled prepared-write path so both
* behave identically once a complete payload is in hand.
*/
private fun handleReceivedPacket(device: BluetoothDevice, value: ByteArray, linkID: String) {
Log.i(TAG, "Server: Received packet from ${device.address}, size: ${value.size} bytes")
val packet = BitchatPacket.fromBinaryData(value)
if (packet != null) {
val peerID = packet.senderID.take(8).toByteArray().joinToString("") { "%02x".format(it) }
Log.d(TAG, "Server: Parsed packet type ${packet.type} from $peerID")
delegate?.onPacketReceived(packet, peerID, device, linkID)
} else {
Log.w(TAG, "Server: Failed to parse packet from ${device.address}, size: ${value.size} bytes")
Log.w(TAG, "Server: Packet data: ${value.joinToString(" ") { "%02x".format(it) }}")
}
}

/**
* Setup GATT server with proper sequencing
Expand Down Expand Up @@ -206,6 +237,8 @@ class BluetoothGattServerManager(
if (linkID != null) {
connectionTracker.cleanupDeviceConnectionIfCurrent(device.address, linkID)
}
// Drop any in-flight prepared-write buffer for this device to avoid leaks
preparedWriteBuffer.cancel(device.address)

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P2 Badge Clear prepared writes when stopping the server

When stop() sets isActive = false before cancelling connections, the resulting disconnect callback returns at the shutdown guard and never reaches this cancellation. Consequently, an in-flight prepared write survives a server stop/restart and may remain allocated or be executed after the same address reconnects. Clear all prepared-write buffers directly in stop(), or perform this per-device cleanup before the callback's isActive guard.

AGENTS.md reference: AGENTS.md:L63-L66

Useful? React with 👍 / 👎.

Copy link
Copy Markdown
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Already handled: stop() calls preparedWriteBuffer.clearAll() before the isActive guard (commit e16ca70), so a server stop clears every in-flight prepared write regardless of the disconnect-callback timing. The normal disconnect path also cancels the per-device buffer directly at STATE_DISCONNECTED (not behind the guard), so a prepared write can't survive a stop/restart.

// Notify delegate about device disconnection so higher layers can update direct flags
delegate?.onDeviceDisconnected(device, linkID)
}
Expand Down Expand Up @@ -238,6 +271,33 @@ class BluetoothGattServerManager(
}

if (characteristic.uuid == AppConstants.Mesh.Gatt.CHARACTERISTIC_UUID) {
if (preparedWrite) {
// BLE reliable/long write (used by iOS for packets larger than the
// negotiated ATT MTU): buffer this chunk by offset and defer parsing
// until onExecuteWrite reassembles the full payload.
val accepted = preparedWriteBuffer.append(device.address, offset, value)
if (accepted) {
Log.d(TAG, "Server: Buffered prepared-write chunk from ${device.address} (offset=$offset, size=${value.size})")
} else {
val reason = if (offset < 0) "invalid offset" else "buffer cap exceeded"
Log.w(TAG, "Server: Rejected prepared-write chunk from ${device.address} (offset=$offset, size=${value.size}); $reason")
}
// On success the prepared-write protocol echoes the offset and value
// back; on rejection send the matching ATT error so the peer can
// abort or resend, instead of only discovering the loss at execute
// (where the chunk is silently dropped).
if (responseNeeded) {
val status = when {
accepted -> BluetoothGatt.GATT_SUCCESS
offset < 0 -> BluetoothGatt.GATT_INVALID_OFFSET
else -> ATT_ERROR_PREPARE_QUEUE_FULL
}
gattServer?.sendResponse(device, requestId, status, offset, if (accepted) value else null)
}
return
}

// Reject writes from a connection we no longer track (stale link).
val linkID = serverLinkIDs[device.address]
if (linkID == null) {
Log.d(TAG, "Server: Dropping packet from stale connection ${device.address}")
Expand All @@ -252,19 +312,46 @@ class BluetoothGattServerManager(
}
return
}
val packet = BitchatPacket.fromBinaryData(value)
if (packet != null) {
val peerID = packet.senderID.take(8).toByteArray().joinToString("") { "%02x".format(it) }
delegate?.onPacketReceived(packet, peerID, device, linkID)
} else {
Log.d(TAG, "Server: Failed to parse packet from ${device.address}, size: ${value.size} bytes")
}


handleReceivedPacket(device, value, linkID)

if (responseNeeded) {
gattServer?.sendResponse(device, requestId, BluetoothGatt.GATT_SUCCESS, 0, null)
}
}
}

override fun onExecuteWrite(device: BluetoothDevice, requestId: Int, execute: Boolean) {
// Guard against callbacks after service shutdown
if (!isActive) {
Log.d(TAG, "Server: Ignoring execute write after shutdown")
preparedWriteBuffer.cancel(device.address)
return
}

if (execute) {
val assembled = preparedWriteBuffer.execute(device.address)
if (assembled != null) {
// Same stale-connection guard as the non-prepared write path: only
// dispatch the reassembled payload if we still track this link.
val linkID = serverLinkIDs[device.address]
if (linkID != null) {
Log.i(TAG, "Server: Reassembled prepared write from ${device.address}, size: ${assembled.size} bytes")
handleReceivedPacket(device, assembled, linkID)
} else {
Log.d(TAG, "Server: Dropping reassembled prepared write from stale connection ${device.address}")
}
} else {
Log.w(TAG, "Server: Execute write from ${device.address} with no or oversized buffered data; dropped")
}
} else {
// Client cancelled the long write; discard the buffered chunks
preparedWriteBuffer.cancel(device.address)
}

// An execute-write always expects a response
gattServer?.sendResponse(device, requestId, BluetoothGatt.GATT_SUCCESS, 0, null)
}

override fun onDescriptorWriteRequest(
device: BluetoothDevice,
Expand Down
145 changes: 145 additions & 0 deletions app/src/main/java/com/bitchat/android/mesh/GattPreparedWriteBuffer.kt
Original file line number Diff line number Diff line change
@@ -0,0 +1,145 @@
package com.bitchat.android.mesh

import java.util.concurrent.ConcurrentHashMap

/**
* Reassembles BLE prepared ("reliable" / long) writes on the GATT server side.
*
* When a peer (notably iOS) needs to send a payload larger than the negotiated
* ATT MTU, it uses the ATT prepared-write procedure: the payload arrives as a
* series of `onCharacteristicWriteRequest` callbacks with `preparedWrite == true`
* and increasing `offset`s, and is finalized by a single `onExecuteWrite`. Each
* individual chunk is only a slice of the packet, so parsing a chunk on its own
* fails. This buffer collects the chunks per device and returns the concatenated
* payload once the write is executed.
*
* All operations are thread-safe: these GATT callbacks can arrive on binder
* threads and several devices may be writing concurrently. Each device is keyed
* by a stable string (its BLE address) so device buffers are isolated.
*
* A per-device size cap protects against a broken or malicious peer streaming an
* unbounded prepared write to exhaust memory; once exceeded the buffer is dropped
* and [execute] returns null.
*/
class GattPreparedWriteBuffer(
private val maxPayloadSize: Int = DEFAULT_MAX_PAYLOAD_SIZE
) {
companion object {
/**
* Generous upper bound for a single reassembled payload. Real bitchat
* packets are far smaller (well under the low-KB range), so this only
* ever trips for abusive peers.
*/
const val DEFAULT_MAX_PAYLOAD_SIZE = 512 * 1024 // 512 KiB
}

private class DeviceBuffer {
var data: ByteArray = ByteArray(0)
var length: Int = 0
var overflowed: Boolean = false
// Byte ranges [start, end) actually written, kept sorted and merged. `data`
// is grown to the highest offset seen, so without this an unwritten gap
// would reassemble as zero-filled and parse as a complete packet. Tracking
// coverage lets execute() tell "not yet written" from "wrote zeros".
val covered: MutableList<IntArray> = mutableListOf()
}

private val buffers = ConcurrentHashMap<String, DeviceBuffer>()

/**
* Buffer a single prepared-write chunk for [deviceKey], placing [value] at
* [offset] within that device's growing payload.
*
* Chunks are written at their byte offset, so reassembly is correct even if
* chunks are delivered out of order. Returns true if the chunk was accepted,
* or false if it was rejected: a negative offset, or a write that would push
* the payload past [maxPayloadSize]. On overflow the device's buffer is
* marked so that [execute] yields null and the oversized payload is dropped.
*/
fun append(deviceKey: String, offset: Int, value: ByteArray): Boolean {
if (offset < 0) return false
val buf = buffers.getOrPut(deviceKey) { DeviceBuffer() }
synchronized(buf) {
if (buf.overflowed) return false
val end = offset.toLong() + value.size.toLong()
if (end > maxPayloadSize.toLong()) {
// Drop what we have and remember the overflow until the write is
// finalized/cancelled, so we never allocate beyond the cap.
buf.overflowed = true
buf.data = ByteArray(0)
buf.length = 0
buf.covered.clear()
return false
}
val endInt = end.toInt()
if (endInt > buf.data.size) {
val newCapacity = maxOf(endInt, buf.data.size * 2).coerceAtMost(maxPayloadSize)
buf.data = buf.data.copyOf(maxOf(newCapacity, endInt))
}
System.arraycopy(value, 0, buf.data, offset, value.size)
if (endInt > buf.length) buf.length = endInt
if (value.isNotEmpty()) addCovered(buf.covered, offset, endInt)
return true
}
}

/**
* Merge the range [start, end) into [ranges], keeping it sorted and coalesced
* so [execute] can check whether the payload was fully covered.
*/
private fun addCovered(ranges: MutableList<IntArray>, start: Int, end: Int) {
if (end <= start) return
ranges.add(intArrayOf(start, end))
ranges.sortBy { it[0] }
val merged = ArrayList<IntArray>(ranges.size)
for (r in ranges) {
val last = merged.lastOrNull()
if (last != null && r[0] <= last[1]) {
last[1] = maxOf(last[1], r[1])
} else {
merged.add(r)
}
}
ranges.clear()
ranges.addAll(merged)
}

/**
* Finalize the prepared write for [deviceKey] and return the reassembled
* payload, or null if nothing was buffered or the buffer overflowed. The
* device's buffer is always removed.
*/
fun execute(deviceKey: String): ByteArray? {
val buf = buffers.remove(deviceKey) ?: return null
synchronized(buf) {
if (buf.overflowed || buf.length == 0) return null
// Reject a payload with a hole: a missing chunk would otherwise be
// handed up as zero-filled and parse as if it were complete. Require a
// single contiguous run covering [0, length).
if (buf.covered.size != 1 || buf.covered[0][0] != 0 || buf.covered[0][1] != buf.length) {
return null
}
return buf.data.copyOf(buf.length)
}
}

/**
* Discard any buffered chunks for [deviceKey]. Used when a prepared write is
* cancelled or when the device disconnects, to avoid leaking buffers.
*/
fun cancel(deviceKey: String) {
buffers.remove(deviceKey)
}

/**
* Drop every device's buffered chunks. Used when the GATT server shuts down, since a
* disconnect that happens after the server is already marked inactive never reaches the
* per-device cancel path.
*/
fun clearAll() {
buffers.clear()
}

/** Number of devices currently holding buffered chunks (diagnostics/tests). */
fun activeDeviceCount(): Int = buffers.size
}
Loading