diff --git a/app/src/main/java/com/bitchat/android/mesh/BluetoothConnectionManager.kt b/app/src/main/java/com/bitchat/android/mesh/BluetoothConnectionManager.kt index 0cc7f967..19c103a3 100644 --- a/app/src/main/java/com/bitchat/android/mesh/BluetoothConnectionManager.kt +++ b/app/src/main/java/com/bitchat/android/mesh/BluetoothConnectionManager.kt @@ -247,6 +247,19 @@ class BluetoothConnectionManager( serverManager.getCharacteristic() ) } + + /** + * Send a packet directly to a specific peer, without broadcasting to others. + */ + fun sendPacketToPeer(peerID: String, packet: BitchatPacket): Boolean { + if (!isActive) return false + return packetBroadcaster.sendPacketToPeer( + RoutedPacket(packet), + peerID, + serverManager.getGattServer(), + serverManager.getCharacteristic() + ) + } // Expose role controls for debug UI diff --git a/app/src/main/java/com/bitchat/android/mesh/BluetoothMeshService.kt b/app/src/main/java/com/bitchat/android/mesh/BluetoothMeshService.kt index f6e5ac64..8d6b357f 100644 --- a/app/src/main/java/com/bitchat/android/mesh/BluetoothMeshService.kt +++ b/app/src/main/java/com/bitchat/android/mesh/BluetoothMeshService.kt @@ -10,6 +10,8 @@ import com.bitchat.android.model.IdentityAnnouncement import com.bitchat.android.protocol.BitchatPacket import com.bitchat.android.protocol.MessageType import com.bitchat.android.protocol.SpecialRecipients +import com.bitchat.android.model.RequestSyncPacket +import com.bitchat.android.sync.GossipSyncManager import com.bitchat.android.util.toHexString import kotlinx.coroutines.* import java.util.* @@ -49,6 +51,7 @@ class BluetoothMeshService(private val context: Context) { private val messageHandler = MessageHandler(myPeerID) internal val connectionManager = BluetoothConnectionManager(context, myPeerID, fragmentManager) // Made internal for access private val packetProcessor = PacketProcessor(myPeerID) + private lateinit var gossipSyncManager: GossipSyncManager // Service state management private var isActive = false @@ -63,6 +66,38 @@ class BluetoothMeshService(private val context: Context) { setupDelegates() messageHandler.packetProcessor = packetProcessor //startPeriodicDebugLogging() + + // Initialize sync manager (needs serviceScope) + gossipSyncManager = GossipSyncManager( + myPeerID = myPeerID, + scope = serviceScope, + configProvider = object : GossipSyncManager.ConfigProvider { + override fun seenCapacity(): Int = try { + com.bitchat.android.ui.debug.DebugPreferenceManager.getSeenPacketCapacity(500) + } catch (_: Exception) { 500 } + + override fun gcsMaxBytes(): Int = try { + com.bitchat.android.ui.debug.DebugPreferenceManager.getGcsMaxFilterBytes(400) + } catch (_: Exception) { 400 } + + override fun gcsTargetFpr(): Double = try { + com.bitchat.android.ui.debug.DebugPreferenceManager.getGcsFprPercent(1.0) / 100.0 + } catch (_: Exception) { 0.01 } + } + ) + + // Wire sync manager delegate + gossipSyncManager.delegate = object : GossipSyncManager.Delegate { + override fun sendPacket(packet: BitchatPacket) { + connectionManager.broadcastPacket(RoutedPacket(packet)) + } + override fun sendPacketToPeer(peerID: String, packet: BitchatPacket) { + connectionManager.sendPacketToPeer(peerID, packet) + } + override fun signPacketForBroadcast(packet: BitchatPacket): BitchatPacket { + return signPacketBeforeBroadcast(packet) + } + } } /** @@ -113,6 +148,9 @@ class BluetoothMeshService(private val context: Context) { override fun onPeerListUpdated(peerIDs: List) { delegate?.didUpdatePeerList(peerIDs) } + override fun onPeerRemoved(peerID: String) { + try { gossipSyncManager.removeAnnouncementForPeer(peerID) } catch (_: Exception) { } + } } // SecurityManager delegate for key exchange notifications @@ -380,13 +418,26 @@ class BluetoothMeshService(private val context: Context) { } catch (_: Exception) { } } } catch (_: Exception) { } + + // Schedule initial sync for this new directly connected peer only + try { gossipSyncManager.scheduleInitialSyncToPeer(pid, 1_000) } catch (_: Exception) { } } } + // Track for sync + try { gossipSyncManager.onPublicPacketSeen(routed.packet) } catch (_: Exception) { } } } override fun handleMessage(routed: RoutedPacket) { serviceScope.launch { messageHandler.handleMessage(routed) } + // Track broadcast messages for sync + try { + val pkt = routed.packet + val isBroadcast = (pkt.recipientID == null || pkt.recipientID.contentEquals(SpecialRecipients.BROADCAST)) + if (isBroadcast && pkt.type == MessageType.MESSAGE.value) { + gossipSyncManager.onPublicPacketSeen(pkt) + } + } catch (_: Exception) { } } override fun handleLeave(routed: RoutedPacket) { @@ -408,6 +459,13 @@ class BluetoothMeshService(private val context: Context) { override fun relayPacket(routed: RoutedPacket) { connectionManager.broadcastPacket(routed) } + + override fun handleRequestSync(routed: RoutedPacket) { + // Decode request and respond with missing packets + val fromPeer = routed.peerID ?: return + val req = RequestSyncPacket.decode(routed.packet.payload) ?: return + gossipSyncManager.handleRequestSync(fromPeer, req) + } } // BluetoothConnectionManager delegates @@ -480,6 +538,8 @@ class BluetoothMeshService(private val context: Context) { // Start periodic announcements for peer discovery and connectivity sendPeriodicBroadcastAnnounce() Log.d(TAG, "Started periodic broadcast announcements (every 30 seconds)") + // Start periodic syncs + gossipSyncManager.start() } else { Log.e(TAG, "Failed to start Bluetooth services") } @@ -504,6 +564,7 @@ class BluetoothMeshService(private val context: Context) { delay(200) // Give leave message time to send // Stop all components + gossipSyncManager.stop() connectionManager.stopServices() peerManager.shutdown() fragmentManager.shutdown() @@ -537,6 +598,8 @@ class BluetoothMeshService(private val context: Context) { // Sign the packet before broadcasting val signedPacket = signPacketBeforeBroadcast(packet) connectionManager.broadcastPacket(RoutedPacket(signedPacket)) + // Track our own broadcast message for sync + try { gossipSyncManager.onPublicPacketSeen(signedPacket) } catch (_: Exception) { } } } @@ -708,6 +771,8 @@ class BluetoothMeshService(private val context: Context) { connectionManager.broadcastPacket(RoutedPacket(signedPacket)) Log.d(TAG, "Sent iOS-compatible signed TLV announce (${tlvPayload.size} bytes)") + // Track announce for sync + try { gossipSyncManager.onPublicPacketSeen(signedPacket) } catch (_: Exception) { } } } @@ -756,6 +821,9 @@ class BluetoothMeshService(private val context: Context) { connectionManager.broadcastPacket(RoutedPacket(signedPacket)) peerManager.markPeerAsAnnouncedTo(peerID) Log.d(TAG, "Sent iOS-compatible signed TLV peer announce to $peerID (${tlvPayload.size} bytes)") + + // Track announce for sync + try { gossipSyncManager.onPublicPacketSeen(signedPacket) } catch (_: Exception) { } } /** diff --git a/app/src/main/java/com/bitchat/android/mesh/BluetoothPacketBroadcaster.kt b/app/src/main/java/com/bitchat/android/mesh/BluetoothPacketBroadcaster.kt index 1fc52bb4..188e9d2e 100644 --- a/app/src/main/java/com/bitchat/android/mesh/BluetoothPacketBroadcaster.kt +++ b/app/src/main/java/com/bitchat/android/mesh/BluetoothPacketBroadcaster.kt @@ -139,6 +139,47 @@ class BluetoothPacketBroadcaster( broadcastSinglePacket(routed, gattServer, characteristic) } + /** + * Send a packet to a specific peer only, without broadcasting. + * Returns true if a direct path was found and used. + */ + fun sendPacketToPeer( + routed: RoutedPacket, + targetPeerID: String, + gattServer: BluetoothGattServer?, + characteristic: BluetoothGattCharacteristic? + ): Boolean { + val packet = routed.packet + val data = packet.toBinaryData() ?: return false + val typeName = MessageType.fromValue(packet.type)?.name ?: packet.type.toString() + val incomingAddr = routed.relayAddress + val incomingPeer = incomingAddr?.let { connectionTracker.addressPeerMap[it] } + val senderPeerID = routed.peerID ?: packet.senderID.toHexString() + val senderNick = senderPeerID.let { pid -> nicknameResolver?.invoke(pid) } + + // Prefer server-side subscriptions + val serverTarget = connectionTracker.getSubscribedDevices() + .firstOrNull { connectionTracker.addressPeerMap[it.address] == targetPeerID } + if (serverTarget != null) { + if (notifyDevice(serverTarget, data, gattServer, characteristic)) { + logPacketRelay(typeName, senderPeerID, senderNick, incomingPeer, incomingAddr, targetPeerID, serverTarget.address, packet.ttl) + return true + } + } + + // Then client connections + val clientTarget = connectionTracker.getConnectedDevices().values + .firstOrNull { connectionTracker.addressPeerMap[it.device.address] == targetPeerID } + if (clientTarget != null) { + if (writeToDeviceConn(clientTarget, data)) { + logPacketRelay(typeName, senderPeerID, senderNick, incomingPeer, incomingAddr, targetPeerID, clientTarget.device.address, packet.ttl) + return true + } + } + + return false + } + /** * Public entry point for broadcasting - submits request to actor for serialization diff --git a/app/src/main/java/com/bitchat/android/mesh/PacketProcessor.kt b/app/src/main/java/com/bitchat/android/mesh/PacketProcessor.kt index 54ecffbe..fca35ca1 100644 --- a/app/src/main/java/com/bitchat/android/mesh/PacketProcessor.kt +++ b/app/src/main/java/com/bitchat/android/mesh/PacketProcessor.kt @@ -146,6 +146,7 @@ class PacketProcessor(private val myPeerID: String) { MessageType.MESSAGE -> handleMessage(routed) MessageType.LEAVE -> handleLeave(routed) MessageType.FRAGMENT -> handleFragment(routed) + MessageType.REQUEST_SYNC -> handleRequestSync(routed) else -> { // Handle private packet types (address check required) if (packetRelayManager.isPacketAddressedToMe(packet)) { @@ -232,6 +233,15 @@ class PacketProcessor(private val myPeerID: String) { // Fragment relay is now handled by centralized PacketRelayManager } + + /** + * Handle REQUEST_SYNC packets (public, TTL=1) + */ + private suspend fun handleRequestSync(routed: RoutedPacket) { + val peerID = routed.peerID ?: "unknown" + Log.d(TAG, "Processing REQUEST_SYNC from ${formatPeerForLog(peerID)}") + delegate?.handleRequestSync(routed) + } /** * Handle delivery acknowledgment @@ -305,6 +315,7 @@ interface PacketProcessorDelegate { fun handleMessage(routed: RoutedPacket) fun handleLeave(routed: RoutedPacket) fun handleFragment(packet: BitchatPacket): BitchatPacket? + fun handleRequestSync(routed: RoutedPacket) // Communication fun sendAnnouncementToPeer(peerID: String) diff --git a/app/src/main/java/com/bitchat/android/mesh/PacketRelayManager.kt b/app/src/main/java/com/bitchat/android/mesh/PacketRelayManager.kt index f292534b..bc401daa 100644 --- a/app/src/main/java/com/bitchat/android/mesh/PacketRelayManager.kt +++ b/app/src/main/java/com/bitchat/android/mesh/PacketRelayManager.kt @@ -41,7 +41,7 @@ class PacketRelayManager(private val myPeerID: String) { val packet = routed.packet val peerID = routed.peerID ?: "unknown" - Log.d(TAG, "Evaluating relay for packet type ${'$'}{packet.type} from ${'$'}peerID (TTL: ${'$'}{packet.ttl})") + Log.d(TAG, "Evaluating relay for packet type ${packet.type} from ${peerID} (TTL: ${packet.ttl})") // Double-check this packet isn't addressed to us if (isPacketAddressedToMe(packet)) { @@ -63,7 +63,7 @@ class PacketRelayManager(private val myPeerID: String) { // Decrement TTL by 1 val relayPacket = packet.copy(ttl = (packet.ttl - 1u).toUByte()) - Log.d(TAG, "Decremented TTL from ${'$'}{packet.ttl} to ${'$'}{relayPacket.ttl}") + Log.d(TAG, "Decremented TTL from ${packet.ttl} to ${relayPacket.ttl}") // Apply relay logic based on packet type and debug switch val shouldRelay = isRelayEnabled() && shouldRelayPacket(relayPacket, peerID) @@ -71,7 +71,7 @@ class PacketRelayManager(private val myPeerID: String) { if (shouldRelay) { relayPacket(RoutedPacket(relayPacket, peerID, routed.relayAddress)) } else { - Log.d(TAG, "Relay decision: NOT relaying packet type ${'$'}{packet.type}") + Log.d(TAG, "Relay decision: NOT relaying packet type ${packet.type}") } } @@ -103,7 +103,7 @@ class PacketRelayManager(private val myPeerID: String) { private fun shouldRelayPacket(packet: BitchatPacket, fromPeerID: String): Boolean { // Always relay if TTL is high enough (indicates important message) if (packet.ttl >= 4u) { - Log.d(TAG, "High TTL (${ '$' }{packet.ttl}), relaying") + Log.d(TAG, "High TTL (${packet.ttl}), relaying") return true } @@ -112,7 +112,7 @@ class PacketRelayManager(private val myPeerID: String) { // Small networks always relay to ensure connectivity if (networkSize <= 3) { - Log.d(TAG, "Small network (${ '$' }networkSize peers), relaying") + Log.d(TAG, "Small network (${networkSize} peers), relaying") return true } @@ -126,52 +126,16 @@ class PacketRelayManager(private val myPeerID: String) { } val shouldRelay = Random.nextDouble() < relayProb - Log.d(TAG, "Network size: ${'$'}networkSize, Relay probability: ${'$'}relayProb, Decision: ${'$'}shouldRelay") + Log.d(TAG, "Network size: ${networkSize}, Relay probability: ${relayProb}, Decision: ${shouldRelay}") return shouldRelay } - /** - * Relay message with adaptive probability and timing (same as iOS) - * Moved from MessageHandler.kt - */ - suspend fun relayMessage(routed: RoutedPacket) { - val packet = routed.packet - - if (packet.ttl == 0u.toUByte()) { - Log.d(TAG, "TTL expired, not relaying message") - return - } - - val relayPacket = packet.copy(ttl = (packet.ttl - 1u).toUByte()) - - // Check network size and apply adaptive relay probability - val networkSize = delegate?.getNetworkSize() ?: 1 - val relayProb = when { - networkSize <= 10 -> 1.0 - networkSize <= 30 -> 0.85 - networkSize <= 50 -> 0.7 - networkSize <= 100 -> 0.55 - else -> 0.4 - } - - val shouldRelay = relayPacket.ttl >= 4u || networkSize <= 3 || Random.nextDouble() < relayProb - - if (shouldRelay) { - val delay = Random.nextLong(50, 500) // Random delay like iOS - Log.d(TAG, "Relaying message after ${'$'}delay ms delay") - delay(delay) - relayPacket(routed.copy(packet = relayPacket)) - } else { - Log.d(TAG, "Relay decision: NOT relaying message (network size: ${'$'}networkSize, prob: ${'$'}relayProb)") - } - } - /** * Actually broadcast the packet for relay */ private fun relayPacket(routed: RoutedPacket) { - Log.d(TAG, "🔄 Relaying packet type ${'$'}{routed.packet.type} with TTL ${'$'}{routed.packet.ttl}") + Log.d(TAG, "🔄 Relaying packet type ${routed.packet.type} with TTL ${routed.packet.ttl}") delegate?.broadcastPacket(routed) } @@ -181,9 +145,9 @@ class PacketRelayManager(private val myPeerID: String) { fun getDebugInfo(): String { return buildString { appendLine("=== Packet Relay Manager Debug Info ===") - appendLine("Relay Scope Active: ${'$'}{relayScope.isActive}") - appendLine("My Peer ID: ${'$'}myPeerID") - appendLine("Network Size: ${'$'}{delegate?.getNetworkSize() ?: \"unknown\"}") + appendLine("Relay Scope Active: ${relayScope.isActive}") + appendLine("My Peer ID: ${myPeerID}") + appendLine("Network Size: ${delegate?.getNetworkSize() ?: "unknown"}") } } diff --git a/app/src/main/java/com/bitchat/android/mesh/PeerManager.kt b/app/src/main/java/com/bitchat/android/mesh/PeerManager.kt index 4b8a16dc..68a539eb 100644 --- a/app/src/main/java/com/bitchat/android/mesh/PeerManager.kt +++ b/app/src/main/java/com/bitchat/android/mesh/PeerManager.kt @@ -262,6 +262,8 @@ class PeerManager { fingerprintManager.removePeer(peerID) if (notifyDelegate && removed != null) { + // Notify specific removal event then list update + try { delegate?.onPeerRemoved(peerID) } catch (_: Exception) {} notifyPeerListUpdate() } } @@ -529,4 +531,5 @@ class PeerManager { */ interface PeerManagerDelegate { fun onPeerListUpdated(peerIDs: List) + fun onPeerRemoved(peerID: String) } diff --git a/app/src/main/java/com/bitchat/android/mesh/SecurityManager.kt b/app/src/main/java/com/bitchat/android/mesh/SecurityManager.kt index 89f0197c..f763424a 100644 --- a/app/src/main/java/com/bitchat/android/mesh/SecurityManager.kt +++ b/app/src/main/java/com/bitchat/android/mesh/SecurityManager.kt @@ -50,12 +50,6 @@ class SecurityManager(private val encryptionService: EncryptionService, private return false } - // TTL check - if (packet.ttl == 0u.toUByte()) { - Log.d(TAG, "Dropping packet with TTL 0") - return false - } - // Validate packet payload if (packet.payload.isEmpty()) { Log.d(TAG, "Dropping packet with empty payload") @@ -67,11 +61,11 @@ class SecurityManager(private val encryptionService: EncryptionService, private val packetTime = packet.timestamp.toLong() val timeDiff = kotlin.math.abs(currentTime - packetTime) - if (timeDiff > MESSAGE_TIMEOUT) { - Log.d(TAG, "Dropping old packet from $peerID, time diff: ${timeDiff/1000}s") - return false - } - +// if (timeDiff > MESSAGE_TIMEOUT) { +// Log.d(TAG, "Dropping old packet from $peerID, time diff: ${timeDiff/1000}s") +// return false +// } + // Duplicate detection val messageID = generateMessageID(packet, peerID) if (processedMessages.contains(messageID)) { diff --git a/app/src/main/java/com/bitchat/android/model/RequestSyncPacket.kt b/app/src/main/java/com/bitchat/android/model/RequestSyncPacket.kt new file mode 100644 index 00000000..ab52f071 --- /dev/null +++ b/app/src/main/java/com/bitchat/android/model/RequestSyncPacket.kt @@ -0,0 +1,82 @@ +package com.bitchat.android.model + +import com.bitchat.android.sync.SyncDefaults + +/** + * REQUEST_SYNC payload using GCS (Golomb-Coded Set) parameters. + * TLV (type, length16, value), types: + * - 0x01: P (uint8) — Golomb-Rice parameter + * - 0x02: M (uint32, big-endian) — hash range (N * 2^P) + * - 0x03: data (opaque) — GR bitstream bytes + */ +data class RequestSyncPacket( + val p: Int, + val m: Long, + val data: ByteArray +) { + fun encode(): ByteArray { + val out = ArrayList() + fun putTLV(t: Int, v: ByteArray) { + out.add(t.toByte()) + val len = v.size + out.add(((len ushr 8) and 0xFF).toByte()) + out.add((len and 0xFF).toByte()) + out.addAll(v.toList()) + } + // P + putTLV(0x01, byteArrayOf(p.toByte())) + // M (uint32) + val m32 = m.coerceAtMost(0xffff_ffffL) + putTLV( + 0x02, + byteArrayOf( + ((m32 ushr 24) and 0xFF).toByte(), + ((m32 ushr 16) and 0xFF).toByte(), + ((m32 ushr 8) and 0xFF).toByte(), + (m32 and 0xFF).toByte() + ) + ) + // data + putTLV(0x03, data) + return out.toByteArray() + } + + companion object { + // Receiver-side safety limit (configurable constant) + const val MAX_ACCEPT_FILTER_BYTES: Int = SyncDefaults.MAX_ACCEPT_FILTER_BYTES + + fun decode(data: ByteArray): RequestSyncPacket? { + var off = 0 + var p: Int? = null + var m: Long? = null + var payload: ByteArray? = null + + while (off + 3 <= data.size) { + val t = (data[off].toInt() and 0xFF); off += 1 + val len = ((data[off].toInt() and 0xFF) shl 8) or (data[off+1].toInt() and 0xFF); off += 2 + if (off + len > data.size) return null + val v = data.copyOfRange(off, off + len); off += len + when (t) { + 0x01 -> if (len == 1) p = (v[0].toInt() and 0xFF) + 0x02 -> if (len == 4) { + val mm = ((v[0].toLong() and 0xFF) shl 24) or + ((v[1].toLong() and 0xFF) shl 16) or + ((v[2].toLong() and 0xFF) shl 8) or + (v[3].toLong() and 0xFF) + m = mm + } + 0x03 -> { + if (v.size > MAX_ACCEPT_FILTER_BYTES) return null + payload = v + } + } + } + + val pp = p ?: return null + val mm = m ?: return null + val dd = payload ?: return null + if (pp < 1 || mm <= 0L) return null + return RequestSyncPacket(pp, mm, dd) + } + } +} diff --git a/app/src/main/java/com/bitchat/android/protocol/BinaryProtocol.kt b/app/src/main/java/com/bitchat/android/protocol/BinaryProtocol.kt index 3dcc47a2..7724ca04 100644 --- a/app/src/main/java/com/bitchat/android/protocol/BinaryProtocol.kt +++ b/app/src/main/java/com/bitchat/android/protocol/BinaryProtocol.kt @@ -15,7 +15,8 @@ enum class MessageType(val value: UByte) { LEAVE(0x03u), NOISE_HANDSHAKE(0x10u), // Noise handshake NOISE_ENCRYPTED(0x11u), // Noise encrypted transport message - FRAGMENT(0x20u); // Fragmentation for large packets + FRAGMENT(0x20u), // Fragmentation for large packets + REQUEST_SYNC(0x21u); // GCS-based sync request companion object { fun fromValue(value: UByte): MessageType? { diff --git a/app/src/main/java/com/bitchat/android/sync/GCSFilter.kt b/app/src/main/java/com/bitchat/android/sync/GCSFilter.kt new file mode 100644 index 00000000..3cc64c58 --- /dev/null +++ b/app/src/main/java/com/bitchat/android/sync/GCSFilter.kt @@ -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<= 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, // 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() + 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, 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() + 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 + } + } +} + diff --git a/app/src/main/java/com/bitchat/android/sync/GossipSyncManager.kt b/app/src/main/java/com/bitchat/android/sync/GossipSyncManager.kt new file mode 100644 index 00000000..38d23a91 --- /dev/null +++ b/app/src/main/java/com/bitchat/android/sync/GossipSyncManager.kt @@ -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() + // - announcements: only keep latest per sender peerID + private val latestAnnouncementByPeer = ConcurrentHashMap>() + + 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() + // 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() + 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") + } + } +} diff --git a/app/src/main/java/com/bitchat/android/sync/PacketIdUtil.kt b/app/src/main/java/com/bitchat/android/sync/PacketIdUtil.kt new file mode 100644 index 00000000..4396ff5a --- /dev/null +++ b/app/src/main/java/com/bitchat/android/sync/PacketIdUtil.kt @@ -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) } + } +} + diff --git a/app/src/main/java/com/bitchat/android/sync/SyncDefaults.kt b/app/src/main/java/com/bitchat/android/sync/SyncDefaults.kt new file mode 100644 index 00000000..970c8afb --- /dev/null +++ b/app/src/main/java/com/bitchat/android/sync/SyncDefaults.kt @@ -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 +} + diff --git a/app/src/main/java/com/bitchat/android/ui/GeohashViewModel.kt b/app/src/main/java/com/bitchat/android/ui/GeohashViewModel.kt index 08865c1e..9ecb94f5 100644 --- a/app/src/main/java/com/bitchat/android/ui/GeohashViewModel.kt +++ b/app/src/main/java/com/bitchat/android/ui/GeohashViewModel.kt @@ -153,7 +153,7 @@ class GeohashViewModel( isRelay = false ) fun startGeohashDM(pubkeyHex: String, onStartPrivateChat: (String) -> Unit) { - val convKey = "nostr_${'$'}{pubkeyHex.take(16)}" + val convKey = "nostr_${pubkeyHex.take(16)}" repo.putNostrKeyMapping(convKey, pubkeyHex) // Record the conversation's geohash using the currently selected location channel (if any) val current = state.selectedLocationChannel.value @@ -163,7 +163,7 @@ class GeohashViewModel( com.bitchat.android.nostr.GeohashConversationRegistry.set(convKey, gh) } onStartPrivateChat(convKey) - Log.d(TAG, "🗨️ Started geohash DM with ${'$'}pubkeyHex -> ${'$'}convKey (geohash=${'$'}gh)") + Log.d(TAG, "🗨️ Started geohash DM with ${pubkeyHex} -> ${convKey} (geohash=${gh})") } messageManager.addMessage(sysMsg) diff --git a/app/src/main/java/com/bitchat/android/ui/debug/DebugPreferenceManager.kt b/app/src/main/java/com/bitchat/android/ui/debug/DebugPreferenceManager.kt index 84b624f7..e2f9d966 100644 --- a/app/src/main/java/com/bitchat/android/ui/debug/DebugPreferenceManager.kt +++ b/app/src/main/java/com/bitchat/android/ui/debug/DebugPreferenceManager.kt @@ -16,6 +16,10 @@ object DebugPreferenceManager { private const val KEY_MAX_CONN_OVERALL = "max_connections_overall" private const val KEY_MAX_CONN_SERVER = "max_connections_server" private const val KEY_MAX_CONN_CLIENT = "max_connections_client" + private const val KEY_SEEN_PACKET_CAP = "seen_packet_capacity" + // GCS keys (no migration/back-compat) + private const val KEY_GCS_MAX_BYTES = "gcs_max_filter_bytes" + private const val KEY_GCS_FPR = "gcs_filter_fpr_percent" private lateinit var prefs: SharedPreferences @@ -74,4 +78,26 @@ object DebugPreferenceManager { fun setMaxConnectionsClient(value: Int) { if (ready()) prefs.edit().putInt(KEY_MAX_CONN_CLIENT, value).apply() } + + // Sync/GCS settings + fun getSeenPacketCapacity(default: Int = 500): Int = + if (ready()) prefs.getInt(KEY_SEEN_PACKET_CAP, default) else default + + fun setSeenPacketCapacity(value: Int) { + if (ready()) prefs.edit().putInt(KEY_SEEN_PACKET_CAP, value).apply() + } + + fun getGcsMaxFilterBytes(default: Int = 400): Int = + if (ready()) prefs.getInt(KEY_GCS_MAX_BYTES, default) else default + + fun setGcsMaxFilterBytes(value: Int) { + if (ready()) prefs.edit().putInt(KEY_GCS_MAX_BYTES, value).apply() + } + + fun getGcsFprPercent(default: Double = 1.0): Double = + if (ready()) java.lang.Double.longBitsToDouble(prefs.getLong(KEY_GCS_FPR, java.lang.Double.doubleToRawLongBits(default))) else default + + fun setGcsFprPercent(value: Double) { + if (ready()) prefs.edit().putLong(KEY_GCS_FPR, java.lang.Double.doubleToRawLongBits(value)).apply() + } } diff --git a/app/src/main/java/com/bitchat/android/ui/debug/DebugSettingsManager.kt b/app/src/main/java/com/bitchat/android/ui/debug/DebugSettingsManager.kt index 81538058..9095d765 100644 --- a/app/src/main/java/com/bitchat/android/ui/debug/DebugSettingsManager.kt +++ b/app/src/main/java/com/bitchat/android/ui/debug/DebugSettingsManager.kt @@ -201,6 +201,37 @@ class DebugSettingsManager private constructor() { fun updateRelayStats(stats: PacketRelayStats) { _relayStats.value = stats } + + // Sync/GCS settings (UI-configurable) + private val _seenPacketCapacity = MutableStateFlow(DebugPreferenceManager.getSeenPacketCapacity(500)) + val seenPacketCapacity: StateFlow = _seenPacketCapacity.asStateFlow() + + private val _gcsMaxBytes = MutableStateFlow(DebugPreferenceManager.getGcsMaxFilterBytes(400)) + val gcsMaxBytes: StateFlow = _gcsMaxBytes.asStateFlow() + + private val _gcsFprPercent = MutableStateFlow(DebugPreferenceManager.getGcsFprPercent(1.0)) + val gcsFprPercent: StateFlow = _gcsFprPercent.asStateFlow() + + fun setSeenPacketCapacity(value: Int) { + val clamped = value.coerceIn(10, 1000) + DebugPreferenceManager.setSeenPacketCapacity(clamped) + _seenPacketCapacity.value = clamped + addDebugMessage(DebugMessage.SystemMessage("🧩 max packets per sync set to $clamped")) + } + + fun setGcsMaxBytes(value: Int) { + val clamped = value.coerceIn(128, 1024) + DebugPreferenceManager.setGcsMaxFilterBytes(clamped) + _gcsMaxBytes.value = clamped + addDebugMessage(DebugMessage.SystemMessage("🌸 max GCS filter size set to $clamped bytes")) + } + + fun setGcsFprPercent(value: Double) { + val clamped = value.coerceIn(0.1, 5.0) + DebugPreferenceManager.setGcsFprPercent(clamped) + _gcsFprPercent.value = clamped + addDebugMessage(DebugMessage.SystemMessage("🎯 GCS FPR set to ${String.format("%.2f", clamped)}%")) + } // MARK: - Debug Message Creation Helpers diff --git a/app/src/main/java/com/bitchat/android/ui/debug/DebugSettingsSheet.kt b/app/src/main/java/com/bitchat/android/ui/debug/DebugSettingsSheet.kt index 03453aaf..23bfc524 100644 --- a/app/src/main/java/com/bitchat/android/ui/debug/DebugSettingsSheet.kt +++ b/app/src/main/java/com/bitchat/android/ui/debug/DebugSettingsSheet.kt @@ -48,6 +48,9 @@ fun DebugSettingsSheet( val scanResults by manager.scanResults.collectAsState() val connectedDevices by manager.connectedDevices.collectAsState() val relayStats by manager.relayStats.collectAsState() + val seenCapacity by manager.seenPacketCapacity.collectAsState() + val gcsMaxBytes by manager.gcsMaxBytes.collectAsState() + val gcsFpr by manager.gcsFprPercent.collectAsState() // Push live connected devices from mesh service whenever sheet is visible LaunchedEffect(isPresented) { @@ -284,6 +287,27 @@ fun DebugSettingsSheet( } } + // Connected devices + item { + Surface(shape = RoundedCornerShape(12.dp), color = colorScheme.surfaceVariant.copy(alpha = 0.2f)) { + Column(Modifier.padding(16.dp), verticalArrangement = Arrangement.spacedBy(10.dp)) { + Row(verticalAlignment = Alignment.CenterVertically, horizontalArrangement = Arrangement.spacedBy(8.dp)) { + Icon(Icons.Filled.SettingsEthernet, contentDescription = null, tint = Color(0xFF9C27B0)) + Text("sync settings", fontFamily = FontFamily.Monospace, fontSize = 14.sp, fontWeight = FontWeight.Medium) + } + Text("max packets per sync: $seenCapacity", fontFamily = FontFamily.Monospace, fontSize = 11.sp, color = colorScheme.onSurface.copy(alpha = 0.7f)) + Slider(value = seenCapacity.toFloat(), onValueChange = { manager.setSeenPacketCapacity(it.toInt()) }, valueRange = 10f..1000f, steps = 99) + Text("max GCS filter size: $gcsMaxBytes bytes (128–1024)", fontFamily = FontFamily.Monospace, fontSize = 11.sp, color = colorScheme.onSurface.copy(alpha = 0.7f)) + Slider(value = gcsMaxBytes.toFloat(), onValueChange = { manager.setGcsMaxBytes(it.toInt()) }, valueRange = 128f..1024f, steps = 0) + Text("target FPR: ${String.format("%.2f", gcsFpr)}%", fontFamily = FontFamily.Monospace, fontSize = 11.sp, color = colorScheme.onSurface.copy(alpha = 0.7f)) + Slider(value = gcsFpr.toFloat(), onValueChange = { manager.setGcsFprPercent(it.toDouble()) }, valueRange = 0.1f..5.0f, steps = 49) + val p = remember(gcsFpr) { com.bitchat.android.sync.GCSFilter.deriveP(gcsFpr / 100.0) } + val nmax = remember(gcsFpr, gcsMaxBytes) { com.bitchat.android.sync.GCSFilter.estimateMaxElementsForSize(gcsMaxBytes, p) } + Text("derived P: $p • est. max elements: $nmax", fontFamily = FontFamily.Monospace, fontSize = 11.sp, color = colorScheme.onSurface.copy(alpha = 0.7f)) + } + } + } + // Connected devices item { Surface(shape = RoundedCornerShape(12.dp), color = colorScheme.surfaceVariant.copy(alpha = 0.2f)) { diff --git a/docs/sync.md b/docs/sync.md new file mode 100644 index 00000000..0bbd0735 --- /dev/null +++ b/docs/sync.md @@ -0,0 +1,149 @@ +# GCS Filter Sync (REQUEST_SYNC) + +This document specifies the gossip-based synchronization feature for BitChat, inspired by Plumtree. It ensures eventual consistency of public packets (ANNOUNCE and broadcast MESSAGE) across nodes via periodic sync requests containing a compact Golomb‑Coded Set (GCS) of recently seen packets. + +## Overview + +- Each node maintains a rolling set of public BitChat packets it has seen recently: + - Broadcast messages (MessageType.MESSAGE where recipient is broadcast) + - Identity announcements (MessageType.ANNOUNCE) + - Default retention is 100 recent packets (configurable in the debug sheet). This value is the maximum number of packets that are synchronized per request (across both types combined). +- Nodes do not maintain a rolling Bloom filter. Instead, they compute a GCS filter on demand when sending a REQUEST_SYNC. +- Every 30 seconds, a node sends a REQUEST_SYNC packet to all immediate neighbors (local only; not relayed). +- Additionally, 5 seconds after the first announcement from a newly directly connected peer is detected, a node sends a REQUEST_SYNC only to that peer (unicast; local only). +- The receiver checks which packets are not in the sender’s filter and sends those packets back. For announcements, only the latest announcement per peerID is sent; for broadcast messages, all missing ones are sent. + +This synchronization is strictly local (not relayed), ensuring only immediate neighbors participate and preventing wide-area flooding while converging content across the mesh. + +## Packet ID + +To compare packets across peers, a deterministic packet ID is used: + +- ID = first 16 bytes of SHA-256 over: [type | senderID | timestamp | payload] +- This yields a 128-bit ID used in the filter. + +Implementation: `com.bitchat.android.sync.PacketIdUtil`. + +## GCS Filter (On-demand) + +Implementation: `com.bitchat.android.sync.GCSFilter`. + +- Parameters (configurable): + - size: 128–1024 bytes (default 256) + - target false positive rate (FPR): default 1% (range 0.1%–5%) +- Derivations: + - P = ceil(log2(1/FPR)) + - Maximum number of elements that fit into the filter is estimated as: N_max ≈ floor((8 * sizeBytes) / (P + 2)) + - This estimate is used to cap the set; the actual encoder will trim further if needed to stay within the configured size. +- What goes into the set: + - Combine the following and sort by packet timestamp (descending): + - Broadcast messages (MessageType 1) + - The most recent ANNOUNCE per peer + - Take at most `min(N_max, maxPacketsPerSync)` items from this ordered list. + - Compute the 16-byte Packet ID (see below), then for hashing use the first 8 bytes of SHA‑256 over the 16‑byte ID. + - Map each hash to [0, M) with M = N * 2^P; sort ascending and encode deltas with Golomb‑Rice parameter P. + +Hashing scheme (fixed for cross‑impl compatibility): +- Packet ID: first 16 bytes of SHA‑256 over [type | senderID | timestamp | payload]. +- GCS hash: h64 = first 8 bytes of SHA‑256 over the 16‑byte Packet ID, interpreted as an unsigned 64‑bit integer. Value = h64 % M. + +## REQUEST_SYNC Packet + +MessageType: `REQUEST_SYNC (0x21)` + +- Header: normal BitChat header with TTL indicating “local-only” semantics. Implementations SHOULD set TTL=0 to prevent any relay; neighbors still receive the packet over the direct link-layer. For periodic sync, recipient is broadcast; for per-peer initial sync, recipient is the specific peer. +- Payload: TLV with 16‑bit big‑endian length fields (type, length16, value) + - 0x01: P (uint8) — Golomb‑Rice parameter + - 0x02: M (uint32) — hash range N * 2^P + - 0x03: data (opaque) — GCS bitstream (MSB‑first bit packing) + +Notes: +- The GCS bitstream uses MSB‑first packing (bit 7 is the first bit in each byte). +- Receivers MUST reject filters with data length exceeding the local maximum (default 1024 bytes) to avoid DoS. + +Encode/Decode implementation: `com.bitchat.android.model.RequestSyncPacket`. + +## Behavior + +Sender behavior: +- Periodic: every 30 seconds, send REQUEST_SYNC with a freshly computed GCS snapshot, broadcast to immediate neighbors, and mark as local‑only (TTL=0 recommended; do not relay). +- Initial per-peer: upon receiving the first ANNOUNCE from a new directly connected peer, send a REQUEST_SYNC only to that peer after ~5 seconds (unicast; TTL=0 recommended; do not relay). + +Receiver behavior: +- Decode the REQUEST_SYNC payload and reconstruct the sorted set of mapped values using the provided P, M, and bitstream. +- For each locally stored public packet ID: + - Compute h64(ID) % M and check if it is in the reconstructed set; if NOT present, send the original packet back with `ttl=0` to the requester only. + - For announcements, send only the latest announcement per (sender peerID). + - For broadcast messages, send all missing ones. + +Announcement retention and pruning (consensus): +- Store only the most recent announcement per peerID for sync purposes. +- Age-out policy: announcements older than 60 seconds MUST be removed from the sync candidate set. +- Pruning cadence: run pruning every 15 seconds to drop expired announcements. +- LEAVE handling: upon receiving a LEAVE message from a peer, immediately remove that peer’s stored announcement from the sync candidate set. +- Stale/offline peer handling: when a peer is considered stale/offline (e.g., last announcement older than 60 seconds), immediately remove that peer’s stored announcement from the sync candidate set. + +Important: original packets are sent unmodified to preserve original signatures (e.g., ANNOUNCE). They MUST NOT be relayed beyond immediate neighbors. Implementations SHOULD send these response packets with TTL=0 (local-only) and, when possible, route them only to the requesting peer without altering the original packet contents. + +## Scope and Types Included + +Included in sync: +- Public broadcast messages: `MessageType.MESSAGE` with BROADCAST recipient (or null recipient). +- Identity announcements: `MessageType.ANNOUNCE`. +- Both packets produced by other peers and packets produced by the requester itself MUST be represented in the requester’s GCS; the responder MUST track and consider its own produced public packets as candidates to return when they are missing on the requester. +- Announcements included in the GCS MUST be at most 60 seconds old at the time of filter construction; older announcements are excluded by pruning. + +Not included: +- Private messages and any packets addressed to a non-broadcast recipient. + +## Configuration (Debug Sheet) + +Exposed under “sync settings” in the debug settings sheet: +- Max packets per sync (default 100) +- Max GCS filter size in bytes (default 256, min 128, max 1024) +- GCS target FPR in percent (default 1%, 0.1%–5%) +- Derived values (display only): P and the estimated maximum number of elements that fit into the filter. + +Backed by `DebugPreferenceManager` getters and setters: +- `getSeenPacketCapacity` / `setSeenPacketCapacity` +- `getGcsMaxFilterBytes` / `setGcsMaxFilterBytes` +- `getGcsFprPercent` / `setGcsFprPercent` + +## Android Integration + +- New/updated types and classes: + - `MessageType.REQUEST_SYNC` (0x21) in `BinaryProtocol.kt` + - `RequestSyncPacket` in `model/RequestSyncPacket.kt` + - `GCSFilter` and `PacketIdUtil` in `sync/` + - `GossipSyncManager` in `sync/` +- `BluetoothMeshService` wires and starts the sync manager, schedules per-peer initial (unicast) and periodic (broadcast) syncs, and forwards seen public packets (including our own) to the manager. +- `PacketProcessor` handles REQUEST_SYNC and forwards to `BluetoothMeshService` which responds via the sync manager with responses targeted only to the requester. + +## Compatibility Notes + +- GCS hashing and TLV structures are fully specified above; other implementations should use the same hashing scheme and payload layout for interoperability. +- REQUEST_SYNC and responses are local-only and MUST NOT be relayed. Implementations SHOULD use TTL=0 to prevent relaying. If an implementation requires TTL>0 for local delivery, it MUST still ensure that REQUEST_SYNC and responses are not relayed beyond direct neighbors (e.g., by special-casing these types in relay logic). + +## Consensus vs. Configurable + +The following items require consensus across all implementations to ensure interoperability: + +- Packet ID recipe: first 16 bytes of SHA‑256(type | senderID | timestamp | payload). +- GCS hashing function and mapping to [0, M) as specified above (v1), and MSB‑first bit packing for the bitstream. +- Payload encoding: TLV with 16‑bit big‑endian lengths; TLV types 0x01 = P (uint8), 0x02 = M (uint32), 0x03 = data (opaque). +- Packet type and scope: REQUEST_SYNC = 0x21; local-only (not relayed); only ANNOUNCE and broadcast MESSAGE are synchronized; ANNOUNCE de‑dupe is “latest per sender peerID”. + +The following are requester‑defined and communicated or local policy (no global agreement required): + +- GCS parameters: P and M are carried in the REQUEST_SYNC and must be used by the receiver for membership tests. The sender chooses size and FPR; receivers MUST cap accepted data length for DoS protection. +- Local storage policy: how many packets to consider and how you determine the “latest” announcement per peer. +- Sync cadence: how often to send REQUEST_SYNC and initial delay after new neighbor connection; whether to use unicast for initial per-peer sync versus broadcast for periodic sync. The number of packets included is bounded by the debug setting and filter capacity. + +Validation and limits (recommended): + +- Reject malformed REQUEST_SYNC payloads (e.g., P < 1, M <= 0, or data length too large for local limits). +- Practical bounds: data length in [0, 1024]; P in [1, 24]; M up to 2^32‑1. + +Versioning: + +- This document defines a fixed GCS hashing scheme (“v1”) with no explicit version field in the payload. Changing the hashing or ID recipe would require a new message or an additional TLV in a future revision; current deployments must adhere to the constants above.