mirror of
https://github.com/permissionlesstech/bitchat-android.git
synced 2026-07-25 11:25:21 +00:00
Plumtree sync (#393)
* wip plumtree * sync works * fix logging * ttl to 0 * fix send packet to one peer * spec * wip GCS instead of bloom * remove bloom filter remainders * clean * prune old announcements * remove announcements from sync after LEAVE * sync after 1 second * pruning * track own announcement and prune messages without announcements * fix pruning * getGcsMaxFilterBytes default value 400 bytes * parameters
This commit is contained in:
callebtc
@@ -0,0 +1,191 @@
|
||||
package com.bitchat.android.sync
|
||||
|
||||
import java.security.MessageDigest
|
||||
import kotlin.math.ceil
|
||||
import kotlin.math.ln
|
||||
|
||||
/**
|
||||
* Golomb-Coded Set (GCS) filter implementation for sync.
|
||||
*
|
||||
* Hashing:
|
||||
* - h64(id) = first 8 bytes of SHA-256 over the 16-byte PacketId (big-endian unsigned)
|
||||
* - Map to range [0, M) via (h64 % M)
|
||||
*
|
||||
* Encoding (v1):
|
||||
* - Sort mapped values ascending; encode deltas (first is v0, then vi - v{i-1}) as positive integers
|
||||
* - For each delta x >= 1, write Golomb-Rice code with parameter P:
|
||||
* q = (x - 1) >> P (unary q ones followed by a zero), then P low bits r = (x - 1) & ((1<<P)-1)
|
||||
* - Bitstream is packed MSB-first in each byte.
|
||||
*/
|
||||
object GCSFilter {
|
||||
data class Params(
|
||||
val p: Int, // Golomb-Rice parameter (>= 1)
|
||||
val m: Long, // Range M = N * 2^P
|
||||
val data: ByteArray // Encoded GR bitstream
|
||||
)
|
||||
|
||||
// Derive P from target FPR; FPR ~= 1 / 2^P
|
||||
fun deriveP(targetFpr: Double): Int {
|
||||
val f = targetFpr.coerceIn(0.000001, 0.25)
|
||||
return ceil(ln(1.0 / f) / ln(2.0)).toInt().coerceAtLeast(1)
|
||||
}
|
||||
|
||||
// Rough capacity estimate: expected bits per element ~= P + 2 (quotient unary ~ around 2 bits)
|
||||
fun estimateMaxElementsForSize(bytes: Int, p: Int): Int {
|
||||
val bits = (bytes * 8).coerceAtLeast(8)
|
||||
val per = (p + 2).coerceAtLeast(3)
|
||||
return (bits / per).coerceAtLeast(1)
|
||||
}
|
||||
|
||||
fun buildFilter(
|
||||
ids: List<ByteArray>, // 16-byte PacketId bytes
|
||||
maxBytes: Int,
|
||||
targetFpr: Double
|
||||
): Params {
|
||||
val p = deriveP(targetFpr)
|
||||
var nCap = estimateMaxElementsForSize(maxBytes, p)
|
||||
val n = ids.size.coerceAtMost(nCap)
|
||||
val selected = ids.take(n)
|
||||
// Map to [0, M)
|
||||
val m = (n.toLong() shl p)
|
||||
val mapped = selected.map { id -> (h64(id) % m) }.sorted()
|
||||
var encoded = encode(mapped, p)
|
||||
// If estimate was too optimistic, trim until it fits
|
||||
var trimmedN = n
|
||||
while (encoded.size > maxBytes && trimmedN > 0) {
|
||||
trimmedN = (trimmedN * 9) / 10 // drop 10%
|
||||
val mapped2 = mapped.take(trimmedN)
|
||||
encoded = encode(mapped2, p)
|
||||
}
|
||||
val finalM = (trimmedN.toLong() shl p)
|
||||
return Params(p = p, m = finalM, data = encoded)
|
||||
}
|
||||
|
||||
fun decodeToSortedSet(p: Int, m: Long, data: ByteArray): LongArray {
|
||||
val values = ArrayList<Long>()
|
||||
val reader = BitReader(data)
|
||||
var acc = 0L
|
||||
val mask = (1L shl p) - 1L
|
||||
while (!reader.eof()) {
|
||||
// Read unary quotient (q ones terminated by zero)
|
||||
var q = 0L
|
||||
while (true) {
|
||||
val b = reader.readBit() ?: break
|
||||
if (b == 1) q++ else break
|
||||
}
|
||||
if (reader.lastWasEOF) break
|
||||
// Read remainder
|
||||
val r = reader.readBits(p) ?: break
|
||||
val x = (q shl p) + r + 1
|
||||
acc += x
|
||||
if (acc >= m) break // out of range safeguard
|
||||
values.add(acc)
|
||||
}
|
||||
return values.toLongArray()
|
||||
}
|
||||
|
||||
fun contains(sortedValues: LongArray, candidate: Long): Boolean {
|
||||
var lo = 0
|
||||
var hi = sortedValues.size - 1
|
||||
while (lo <= hi) {
|
||||
val mid = (lo + hi) ushr 1
|
||||
val v = sortedValues[mid]
|
||||
if (v == candidate) return true
|
||||
if (v < candidate) lo = mid + 1 else hi = mid - 1
|
||||
}
|
||||
return false
|
||||
}
|
||||
|
||||
private fun h64(id16: ByteArray): Long {
|
||||
val md = MessageDigest.getInstance("SHA-256")
|
||||
md.update(id16)
|
||||
val d = md.digest()
|
||||
var x = 0L
|
||||
for (i in 0 until 8) {
|
||||
x = (x shl 8) or ((d[i].toLong() and 0xFF))
|
||||
}
|
||||
return x and 0x7fff_ffff_ffff_ffffL // positive
|
||||
}
|
||||
|
||||
private fun encode(sorted: List<Long>, p: Int): ByteArray {
|
||||
val bw = BitWriter()
|
||||
var prev = 0L
|
||||
val mask = (1L shl p) - 1L
|
||||
for (v in sorted) {
|
||||
val delta = v - prev
|
||||
prev = v
|
||||
val x = delta
|
||||
val q = (x - 1) ushr p
|
||||
val r = (x - 1) and mask
|
||||
// unary q ones then a zero
|
||||
repeat(q.toInt()) { bw.writeBit(1) }
|
||||
bw.writeBit(0)
|
||||
// then P bits of r (MSB-first)
|
||||
bw.writeBits(r, p)
|
||||
}
|
||||
return bw.toByteArray()
|
||||
}
|
||||
|
||||
// Simple MSB-first bit writer
|
||||
private class BitWriter {
|
||||
private val buf = ArrayList<Byte>()
|
||||
private var cur = 0
|
||||
private var nbits = 0
|
||||
fun writeBit(bit: Int) {
|
||||
cur = (cur shl 1) or (bit and 1)
|
||||
nbits++
|
||||
if (nbits == 8) {
|
||||
buf.add(cur.toByte())
|
||||
cur = 0; nbits = 0
|
||||
}
|
||||
}
|
||||
fun writeBits(value: Long, count: Int) {
|
||||
if (count <= 0) return
|
||||
for (i in count - 1 downTo 0) {
|
||||
val bit = ((value ushr i) and 1L).toInt()
|
||||
writeBit(bit)
|
||||
}
|
||||
}
|
||||
fun toByteArray(): ByteArray {
|
||||
if (nbits > 0) {
|
||||
val rem = cur shl (8 - nbits)
|
||||
buf.add(rem.toByte())
|
||||
cur = 0; nbits = 0
|
||||
}
|
||||
return buf.toByteArray()
|
||||
}
|
||||
}
|
||||
|
||||
// Simple MSB-first bit reader
|
||||
private class BitReader(private val data: ByteArray) {
|
||||
private var i = 0
|
||||
private var nleft = 8
|
||||
private var cur = if (data.isNotEmpty()) (data[0].toInt() and 0xFF) else 0
|
||||
var lastWasEOF: Boolean = false
|
||||
private set
|
||||
fun eof() = i >= data.size
|
||||
fun readBit(): Int? {
|
||||
if (i >= data.size) { lastWasEOF = true; return null }
|
||||
val bit = (cur ushr 7) and 1
|
||||
cur = (cur shl 1) and 0xFF
|
||||
nleft--
|
||||
if (nleft == 0) {
|
||||
i++
|
||||
if (i < data.size) {
|
||||
cur = data[i].toInt() and 0xFF
|
||||
nleft = 8
|
||||
}
|
||||
}
|
||||
return bit
|
||||
}
|
||||
fun readBits(count: Int): Long? {
|
||||
var v = 0L
|
||||
for (k in 0 until count) {
|
||||
val b = readBit() ?: return null
|
||||
v = (v shl 1) or b.toLong()
|
||||
}
|
||||
return v
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -0,0 +1,263 @@
|
||||
package com.bitchat.android.sync
|
||||
|
||||
import android.util.Log
|
||||
import com.bitchat.android.mesh.BluetoothPacketBroadcaster
|
||||
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 kotlinx.coroutines.*
|
||||
import java.util.concurrent.ConcurrentHashMap
|
||||
|
||||
/**
|
||||
* Gossip-based synchronization manager using on-demand GCS filters.
|
||||
* Tracks seen public packets (ANNOUNCE, broadcast MESSAGE) and periodically requests sync
|
||||
* from neighbors. Responds to REQUEST_SYNC by sending missing packets.
|
||||
*/
|
||||
class GossipSyncManager(
|
||||
private val myPeerID: String,
|
||||
private val scope: CoroutineScope,
|
||||
private val configProvider: ConfigProvider
|
||||
) {
|
||||
interface Delegate {
|
||||
fun sendPacket(packet: BitchatPacket)
|
||||
fun sendPacketToPeer(peerID: String, packet: BitchatPacket)
|
||||
fun signPacketForBroadcast(packet: BitchatPacket): BitchatPacket
|
||||
}
|
||||
|
||||
interface ConfigProvider {
|
||||
fun seenCapacity(): Int // max packets we sync per request (cap across types)
|
||||
fun gcsMaxBytes(): Int
|
||||
fun gcsTargetFpr(): Double // percent -> 0.0..1.0
|
||||
}
|
||||
|
||||
companion object { private const val TAG = "GossipSyncManager" }
|
||||
|
||||
var delegate: Delegate? = null
|
||||
|
||||
// Defaults (configurable constants)
|
||||
private val defaultMaxBytes = SyncDefaults.DEFAULT_FILTER_BYTES
|
||||
private val defaultFpr = SyncDefaults.DEFAULT_FPR_PERCENT
|
||||
|
||||
// Stored packets for sync:
|
||||
// - broadcast messages: keep up to seenCapacity() most recent, keyed by packetId
|
||||
private val messages = LinkedHashMap<String, BitchatPacket>()
|
||||
// - announcements: only keep latest per sender peerID
|
||||
private val latestAnnouncementByPeer = ConcurrentHashMap<String, Pair<String, BitchatPacket>>()
|
||||
|
||||
private var periodicJob: Job? = null
|
||||
fun start() {
|
||||
periodicJob?.cancel()
|
||||
periodicJob = scope.launch(Dispatchers.IO) {
|
||||
while (isActive) {
|
||||
try {
|
||||
delay(30_000)
|
||||
sendRequestSync()
|
||||
} catch (e: CancellationException) { throw e }
|
||||
catch (e: Exception) { Log.e(TAG, "Periodic sync error: ${e.message}") }
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
fun stop() {
|
||||
periodicJob?.cancel(); periodicJob = null
|
||||
}
|
||||
|
||||
fun scheduleInitialSync(delayMs: Long = 5_000L) {
|
||||
scope.launch(Dispatchers.IO) {
|
||||
delay(delayMs)
|
||||
sendRequestSync()
|
||||
}
|
||||
}
|
||||
|
||||
fun scheduleInitialSyncToPeer(peerID: String, delayMs: Long = 5_000L) {
|
||||
scope.launch(Dispatchers.IO) {
|
||||
delay(delayMs)
|
||||
sendRequestSyncToPeer(peerID)
|
||||
}
|
||||
}
|
||||
|
||||
fun onPublicPacketSeen(packet: BitchatPacket) {
|
||||
// Only ANNOUNCE or broadcast MESSAGE
|
||||
val mt = MessageType.fromValue(packet.type)
|
||||
val isBroadcastMessage = (mt == MessageType.MESSAGE && (packet.recipientID == null || packet.recipientID.contentEquals(SpecialRecipients.BROADCAST)))
|
||||
val isAnnouncement = (mt == MessageType.ANNOUNCE)
|
||||
if (!isBroadcastMessage && !isAnnouncement) return
|
||||
|
||||
val idBytes = PacketIdUtil.computeIdBytes(packet)
|
||||
val id = idBytes.joinToString("") { b -> "%02x".format(b) }
|
||||
|
||||
if (isBroadcastMessage) {
|
||||
synchronized(messages) {
|
||||
messages[id] = packet
|
||||
// Enforce capacity (remove oldest when exceeded)
|
||||
val cap = configProvider.seenCapacity().coerceAtLeast(1)
|
||||
while (messages.size > cap) {
|
||||
val it = messages.entries.iterator()
|
||||
if (it.hasNext()) { it.next(); it.remove() } else break
|
||||
}
|
||||
}
|
||||
} else if (isAnnouncement) {
|
||||
// senderID is fixed-size 8 bytes; map to hex string for key
|
||||
val sender = packet.senderID.joinToString("") { b -> "%02x".format(b) }
|
||||
latestAnnouncementByPeer[sender] = id to packet
|
||||
// Enforce capacity (remove oldest when exceeded)
|
||||
val cap = configProvider.seenCapacity().coerceAtLeast(1)
|
||||
while (latestAnnouncementByPeer.size > cap) {
|
||||
val it = latestAnnouncementByPeer.entries.iterator()
|
||||
if (it.hasNext()) { it.next(); it.remove() } else break
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private fun sendRequestSync() {
|
||||
val payload = buildGcsPayload()
|
||||
|
||||
val packet = BitchatPacket(
|
||||
type = MessageType.REQUEST_SYNC.value,
|
||||
senderID = hexStringToByteArray(myPeerID),
|
||||
timestamp = System.currentTimeMillis().toULong(),
|
||||
payload = payload,
|
||||
ttl = 0u // neighbors only
|
||||
)
|
||||
// Sign and broadcast
|
||||
val signed = delegate?.signPacketForBroadcast(packet) ?: packet
|
||||
delegate?.sendPacket(signed)
|
||||
}
|
||||
|
||||
private fun sendRequestSyncToPeer(peerID: String) {
|
||||
val payload = buildGcsPayload()
|
||||
|
||||
val packet = BitchatPacket(
|
||||
type = MessageType.REQUEST_SYNC.value,
|
||||
senderID = hexStringToByteArray(myPeerID),
|
||||
recipientID = hexStringToByteArray(peerID),
|
||||
timestamp = System.currentTimeMillis().toULong(),
|
||||
payload = payload,
|
||||
ttl = 0u // neighbor only
|
||||
)
|
||||
Log.d(TAG, "Sending sync request to $peerID (${payload.size} bytes)")
|
||||
// Sign and send directly to peer
|
||||
val signed = delegate?.signPacketForBroadcast(packet) ?: packet
|
||||
delegate?.sendPacketToPeer(peerID, signed)
|
||||
}
|
||||
|
||||
fun handleRequestSync(fromPeerID: String, request: RequestSyncPacket) {
|
||||
// Decode GCS into sorted set for membership checks
|
||||
val sorted = GCSFilter.decodeToSortedSet(request.p, request.m, request.data)
|
||||
fun mightContain(id: ByteArray): Boolean {
|
||||
val v = (GCSFilter.run {
|
||||
// reuse hashing method from GCSFilter
|
||||
val md = java.security.MessageDigest.getInstance("SHA-256");
|
||||
md.update(id); val d = md.digest();
|
||||
var x = 0L; for (i in 0 until 8) { x = (x shl 8) or (d[i].toLong() and 0xFF) }
|
||||
(x and 0x7fff_ffff_ffff_ffffL) % request.m
|
||||
})
|
||||
return GCSFilter.contains(sorted, v)
|
||||
}
|
||||
|
||||
// 1) Announcements: send latest per peerID if remote doesn't have them
|
||||
for ((_, pair) in latestAnnouncementByPeer.entries) {
|
||||
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 = 0u)
|
||||
delegate?.sendPacketToPeer(fromPeerID, toSend)
|
||||
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) {
|
||||
val idBytes = PacketIdUtil.computeIdBytes(pkt)
|
||||
if (!mightContain(idBytes)) {
|
||||
val toSend = pkt.copy(ttl = 0u)
|
||||
delegate?.sendPacketToPeer(fromPeerID, toSend)
|
||||
Log.d(TAG, "Sent sync message: Type ${toSend.type} to $fromPeerID packet id ${idBytes.toHexString()}")
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private fun hexStringToByteArray(hexString: String): ByteArray {
|
||||
val result = ByteArray(8) { 0 }
|
||||
var tempID = hexString
|
||||
var index = 0
|
||||
while (tempID.length >= 2 && index < 8) {
|
||||
val hexByte = tempID.substring(0, 2)
|
||||
val byte = hexByte.toIntOrNull(16)?.toByte()
|
||||
if (byte != null) result[index] = byte
|
||||
tempID = tempID.substring(2)
|
||||
index++
|
||||
}
|
||||
return result
|
||||
}
|
||||
|
||||
private fun hexToBytes(hex: String): ByteArray {
|
||||
val clean = if (hex.length % 2 == 0) hex else "0$hex"
|
||||
val out = ByteArray(clean.length / 2)
|
||||
var i = 0
|
||||
while (i < clean.length) {
|
||||
out[i/2] = clean.substring(i, i+2).toInt(16).toByte()
|
||||
i += 2
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
private fun buildGcsPayload(): ByteArray {
|
||||
// Collect candidates: latest announcement per peer + recent broadcast messages
|
||||
val list = ArrayList<BitchatPacket>()
|
||||
// announcements
|
||||
for ((_, pair) in latestAnnouncementByPeer) {
|
||||
list.add(pair.second)
|
||||
}
|
||||
// messages
|
||||
synchronized(messages) {
|
||||
list.addAll(messages.values)
|
||||
}
|
||||
// sort by timestamp desc, then take up to min(seenCapacity, fit capacity)
|
||||
list.sortByDescending { it.timestamp.toLong() }
|
||||
|
||||
val maxBytes = try { configProvider.gcsMaxBytes() } catch (_: Exception) { defaultMaxBytes }
|
||||
val fpr = try { configProvider.gcsTargetFpr() } catch (_: Exception) { defaultFpr }
|
||||
val p = GCSFilter.deriveP(fpr)
|
||||
val nMax = GCSFilter.estimateMaxElementsForSize(maxBytes, p)
|
||||
val cap = configProvider.seenCapacity().coerceAtLeast(1)
|
||||
val takeN = minOf(nMax, cap, list.size)
|
||||
if (takeN <= 0) {
|
||||
val p0 = GCSFilter.deriveP(fpr)
|
||||
return RequestSyncPacket(p = p0, m = 1, data = ByteArray(0)).encode()
|
||||
}
|
||||
val ids = list.take(takeN).map { pkt -> PacketIdUtil.computeIdBytes(pkt) }
|
||||
val params = GCSFilter.buildFilter(ids, maxBytes, fpr)
|
||||
val mVal = if (params.m <= 0L) 1 else params.m
|
||||
return RequestSyncPacket(p = params.p, m = mVal, data = params.data).encode()
|
||||
}
|
||||
|
||||
// Explicitly remove stored announcement for a given peer (hex ID)
|
||||
fun removeAnnouncementForPeer(peerID: String) {
|
||||
val key = peerID.lowercase()
|
||||
if (latestAnnouncementByPeer.remove(key) != null) {
|
||||
Log.d(TAG, "Removed stored announcement for peer $peerID")
|
||||
}
|
||||
|
||||
// Collect IDs to remove first to avoid modifying collection while iterating
|
||||
val idsToRemove = mutableListOf<String>()
|
||||
for ((id, message) in messages) {
|
||||
val sender = message.senderID.joinToString("") { b -> "%02x".format(b) }
|
||||
if (sender == key) {
|
||||
idsToRemove.add(id)
|
||||
}
|
||||
}
|
||||
|
||||
// Now remove the collected IDs
|
||||
for (id in idsToRemove) {
|
||||
messages.remove(id)
|
||||
}
|
||||
|
||||
if (idsToRemove.isNotEmpty()) {
|
||||
Log.d(TAG, "Pruned ${idsToRemove.size} messages with senders without announcements")
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,31 @@
|
||||
package com.bitchat.android.sync
|
||||
|
||||
import com.bitchat.android.protocol.BitchatPacket
|
||||
import java.security.MessageDigest
|
||||
|
||||
/**
|
||||
* Deterministic packet ID helper for sync purposes.
|
||||
* Uses SHA-256 over a canonical subset of packet fields:
|
||||
* [type | senderID | timestamp | payload] to generate a stable ID.
|
||||
* Returns a 16-byte (128-bit) truncated hash for compactness.
|
||||
*/
|
||||
object PacketIdUtil {
|
||||
fun computeIdBytes(packet: BitchatPacket): ByteArray {
|
||||
val md = MessageDigest.getInstance("SHA-256")
|
||||
md.update(packet.type.toByte())
|
||||
md.update(packet.senderID)
|
||||
// Timestamp as 8 bytes big-endian
|
||||
val ts = packet.timestamp.toLong()
|
||||
for (i in 7 downTo 0) {
|
||||
md.update(((ts ushr (i * 8)) and 0xFF).toByte())
|
||||
}
|
||||
md.update(packet.payload)
|
||||
val digest = md.digest()
|
||||
return digest.copyOf(16) // 128-bit ID
|
||||
}
|
||||
|
||||
fun computeIdHex(packet: BitchatPacket): String {
|
||||
return computeIdBytes(packet).joinToString("") { b -> "%02x".format(b) }
|
||||
}
|
||||
}
|
||||
|
||||
@@ -0,0 +1,11 @@
|
||||
package com.bitchat.android.sync
|
||||
|
||||
object SyncDefaults {
|
||||
// Default values used when debug prefs are unavailable
|
||||
const val DEFAULT_FILTER_BYTES: Int = 256
|
||||
const val DEFAULT_FPR_PERCENT: Double = 1.0
|
||||
|
||||
// Receiver-side hard cap to avoid DoS (also enforced in RequestSyncPacket)
|
||||
const val MAX_ACCEPT_FILTER_BYTES: Int = 1024
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user