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 @@ -654,7 +654,7 @@ class BluetoothMeshService(private val context: Context) : TransportBridgeServic
// Decode request and respond with missing packets
val fromPeer = routed.peerID ?: return
val req = RequestSyncPacket.decode(routed.packet.payload) ?: return
gossipSyncManager.handleRequestSync(fromPeer, req)
gossipSyncManager.handleRequestSync(fromPeer, req, routed.ingressLinkID)
}
}

Expand Down
2 changes: 1 addition & 1 deletion app/src/main/java/com/bitchat/android/mesh/MeshCore.kt
Original file line number Diff line number Diff line change
Expand Up @@ -525,7 +525,7 @@ class MeshCore(
override fun handleRequestSync(routed: RoutedPacket) {
val fromPeer = routed.peerID ?: return
val req = RequestSyncPacket.decode(routed.packet.payload) ?: return
gossipSyncManager.handleRequestSync(fromPeer, req)
gossipSyncManager.handleRequestSync(fromPeer, req, routed.ingressLinkID)
}
}
}
Expand Down
108 changes: 106 additions & 2 deletions app/src/main/java/com/bitchat/android/sync/GossipSyncManager.kt
Original file line number Diff line number Diff line change
Expand Up @@ -6,8 +6,10 @@ import com.bitchat.android.model.RequestSyncPacket
import com.bitchat.android.protocol.BitchatPacket
import com.bitchat.android.protocol.MessageType
import com.bitchat.android.protocol.SpecialRecipients
import com.bitchat.android.util.AppConstants
import kotlinx.coroutines.*
import java.util.concurrent.ConcurrentHashMap
import kotlin.math.floor

/**
* Gossip-based synchronization manager using on-demand GCS filters.
Expand All @@ -17,7 +19,8 @@ import java.util.concurrent.ConcurrentHashMap
class GossipSyncManager(
private val myPeerID: String,
private val scope: CoroutineScope,
private val configProvider: ConfigProvider
private val configProvider: ConfigProvider,
private val nowMs: () -> Long = { System.currentTimeMillis() }
) {
interface Delegate {
fun sendPacket(packet: BitchatPacket)
Expand Down Expand Up @@ -47,6 +50,25 @@ class GossipSyncManager(
// - announcements: only keep latest per sender peerID
private val latestAnnouncementByPeer = ConcurrentHashMap<String, Pair<String, BitchatPacket>>()

/**
* Per-requester token buckets. `requestTokens` bounds how often we are willing to decode a
* peer's filter and hash our stored packets against it; `responseTokens` bounds how many
* packets that peer can pull out of us. Both refill with elapsed time.
*/
private class RequesterBudget(
var requestTokens: Double,
var responseTokens: Double,
var lastRefillMs: Long
)

// Access-ordered so the eldest entry is the least recently seen requester.
private val requesterBudgets =
object : LinkedHashMap<String, RequesterBudget>(16, 0.75f, true) {
override fun removeEldestEntry(
eldest: MutableMap.MutableEntry<String, RequesterBudget>
): Boolean = size > AppConstants.Sync.MAX_TRACKED_REQUESTERS
}

private var periodicJob: Job? = null
private var cleanupJob: Job? = null
fun start() {
Expand Down Expand Up @@ -85,6 +107,9 @@ class GossipSyncManager(
messages.clear()
}
latestAnnouncementByPeer.clear()
synchronized(requesterBudgets) {
requesterBudgets.clear()
}
Log.d(TAG, "Cleared all gossip sync messages and announcements")
}

Expand Down Expand Up @@ -174,7 +199,80 @@ class GossipSyncManager(
delegate?.sendPacketToPeer(peerID, signed)
}

fun handleRequestSync(fromPeerID: String, request: RequestSyncPacket) {
/**
* Charges [budgetKey] for one request and returns how many packets may be sent in response,
* or 0 when the request must be dropped. Called before any decoding or hashing so a flood
* costs us the admission check and nothing else.
*/
private fun admitRequest(budgetKey: String): Int {
val capacity = try {
configProvider.seenCapacity()
} catch (_: Exception) {
0
}.coerceAtLeast(1).toDouble()
val now = nowMs()

synchronized(requesterBudgets) {
val budget = requesterBudgets.getOrPut(budgetKey) {
RequesterBudget(
requestTokens = AppConstants.Sync.REQUEST_BURST.toDouble(),
responseTokens = capacity,
lastRefillMs = now
)
}

// A clock that jumped backwards must not hand out free tokens.
val elapsedMs = (now - budget.lastRefillMs).coerceAtLeast(0L).toDouble()
budget.lastRefillMs = now
budget.requestTokens = (
budget.requestTokens + elapsedMs / AppConstants.Sync.REQUEST_INTERVAL_MS
).coerceAtMost(AppConstants.Sync.REQUEST_BURST.toDouble())
budget.responseTokens = (
budget.responseTokens +
capacity * elapsedMs / AppConstants.Sync.RESPONSE_REFILL_WINDOW_MS
).coerceAtMost(capacity)

if (budget.requestTokens < 1.0) return 0
val allowance = floor(budget.responseTokens).toInt()
if (allowance <= 0) return 0

budget.requestTokens -= 1.0
return allowance
}
}

internal fun trackedRequesterCount(): Int = synchronized(requesterBudgets) {
requesterBudgets.size
}

private fun chargeResponses(budgetKey: String, sent: Int) {
if (sent <= 0) return
synchronized(requesterBudgets) {
val budget = requesterBudgets[budgetKey] ?: return
budget.responseTokens = (budget.responseTokens - sent).coerceAtLeast(0.0)
}
}

/**
* @param ingressLinkID locally-assigned identity of the link the request arrived on. REQUEST_SYNC
* is not in the set of types `SecurityManager.verifyPacketSignature` authenticates, so the
* sender ID on the wire is unauthenticated and free to rotate — budgeting by it would let a
* peer mint a fresh allowance per request. This value is assigned by the transport and never
* serialized onto the mesh, so a flooder is bounded by the links it can actually hold open.
*/
fun handleRequestSync(
fromPeerID: String,
request: RequestSyncPacket,
ingressLinkID: String? = null
) {
val budgetKey = ingressLinkID ?: fromPeerID
var remaining = admitRequest(budgetKey)
if (remaining <= 0) {
Log.d(TAG, "Dropping REQUEST_SYNC from $fromPeerID: over budget")
return
}
val allowance = remaining

// Decode GCS into sorted set for membership checks
val sorted = GCSFilter.decodeToSortedSet(request.p, request.m, request.data)
fun mightContain(id: ByteArray): Boolean {
Expand All @@ -185,26 +283,32 @@ class GossipSyncManager(

// 1) Announcements: send latest per peerID if remote doesn't have them
for ((_, pair) in latestAnnouncementByPeer.entries) {
if (remaining <= 0) break
val (id, pkt) = pair
val idBytes = hexToBytes(id)
if (!mightContain(idBytes)) {
// Send original packet unchanged to requester only (keep local TTL)
val toSend = pkt.copy(ttl = com.bitchat.android.util.AppConstants.SYNC_TTL_HOPS)
delegate?.sendPacketToPeer(fromPeerID, toSend)
remaining--
Log.d(TAG, "Sent sync announce: Type ${toSend.type} from ${toSend.senderID.toHexString()} to $fromPeerID packet id ${idBytes.toHexString()}")
}
}

// 2) Broadcast messages: send all they lack
val toSendMsgs = synchronized(messages) { messages.values.toList() }
for (pkt in toSendMsgs) {
if (remaining <= 0) break
val idBytes = PacketIdUtil.computeIdBytes(pkt)
if (!mightContain(idBytes)) {
val toSend = pkt.copy(ttl = com.bitchat.android.util.AppConstants.SYNC_TTL_HOPS)
delegate?.sendPacketToPeer(fromPeerID, toSend)
remaining--
Log.d(TAG, "Sent sync message: Type ${toSend.type} to $fromPeerID packet id ${idBytes.toHexString()}")
}
}

chargeResponses(budgetKey, allowance - remaining)
}

private fun hexStringToByteArray(hexString: String): ByteArray {
Expand Down
19 changes: 19 additions & 0 deletions app/src/main/java/com/bitchat/android/util/AppConstants.kt
Original file line number Diff line number Diff line change
Expand Up @@ -32,6 +32,25 @@ object AppConstants {

object Sync {
const val CLEANUP_INTERVAL_MS: Long = 60_000L

// REQUEST_SYNC is a public, unauthenticated packet, so a neighbor decides how often we
// decode a filter, hash every stored packet, and transmit whatever it claims to lack.
// The budgets below bound that work without touching what an honest peer needs.

// Shortest spacing between two serviced requests from the same peer. Honest peers ask on
// a 30s timer plus one scheduled first request, so only floods are shed.
const val REQUEST_INTERVAL_MS: Long = 5_000L

// Lets the scheduled first sync and a periodic one land back to back.
const val REQUEST_BURST: Int = 2

// Window over which a peer's response allowance refills to seenCapacity() packets — the
// most a peer can legitimately be missing in one round.
const val RESPONSE_REFILL_WINDOW_MS: Long = 30_000L

// Requester IDs come off the wire, so the bookkeeping itself has to be bounded or the
// mitigation becomes its own memory exhaustion vector.
const val MAX_TRACKED_REQUESTERS: Int = 256
}

object Fragmentation {
Expand Down
Loading