mirror of
https://github.com/permissionlesstech/bitchat-android.git
synced 2026-09-19 04:59:59 +00:00
Fix courier custody and sync interoperability
This commit is contained in:
callebtc
parent
d83fb82aae
commit
39b3eaa999
@ -1632,22 +1632,35 @@ class BluetoothMeshService(private val context: Context) : TransportBridgeServic
|
||||
}
|
||||
if (direct) {
|
||||
courierStore.sprayCopiesFor(noiseKey).forEach { envelope ->
|
||||
val payload = envelope.encode()
|
||||
if (payload == null) {
|
||||
courierStore.cancelSpray(envelope, noiseKey)
|
||||
return@forEach
|
||||
}
|
||||
val packet = BitchatPacket(
|
||||
type = MessageType.COURIER_ENVELOPE.value,
|
||||
senderID = hexStringToByteArray(myPeerID),
|
||||
recipientID = hexStringToByteArray(peerID),
|
||||
timestamp = System.currentTimeMillis().toULong(),
|
||||
payload = envelope.encode() ?: return@forEach,
|
||||
payload = payload,
|
||||
ttl = MAX_TTL
|
||||
)
|
||||
serviceScope.launch {
|
||||
if (connectionManager.sendToPeerAndAwaitCompletion(
|
||||
peerID,
|
||||
RoutedPacket(signPacketBeforeBroadcast(packet))
|
||||
)
|
||||
) {
|
||||
courierStore.commitSpray(envelope, noiseKey)
|
||||
var committed = false
|
||||
try {
|
||||
if (connectionManager.sendToPeerAndAwaitCompletion(
|
||||
peerID,
|
||||
RoutedPacket(signPacketBeforeBroadcast(packet))
|
||||
)
|
||||
) {
|
||||
committed = courierStore.commitSpray(envelope, noiseKey)
|
||||
}
|
||||
} finally {
|
||||
if (!committed) courierStore.cancelSpray(envelope, noiseKey)
|
||||
}
|
||||
}.invokeOnCompletion {
|
||||
// A coroutine cancelled before its body starts never reaches finally.
|
||||
courierStore.cancelSpray(envelope, noiseKey)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@ -27,6 +27,12 @@ internal class CourierStore(
|
||||
var lastRemoteHandoverAtMs: Long = 0
|
||||
)
|
||||
|
||||
private data class SprayReservation(
|
||||
val envelopeKey: String,
|
||||
val courierKey: String,
|
||||
val copies: Int
|
||||
)
|
||||
|
||||
companion object {
|
||||
private const val KEY_ALIAS = "bitchat_courier_store_v1"
|
||||
private val AAD = "bitchat-courier-store-v1".toByteArray(Charsets.UTF_8)
|
||||
@ -41,6 +47,7 @@ internal class CourierStore(
|
||||
private val gson = Gson()
|
||||
private val file = File(context.applicationContext.filesDir, "courier-store.sealed")
|
||||
private val stored = load()
|
||||
private val sprayReservations = mutableListOf<SprayReservation>()
|
||||
|
||||
@Synchronized
|
||||
fun deposit(envelope: CourierEnvelope, depositorNoiseKey: ByteArray, tier: CourierDepositTier): Boolean {
|
||||
@ -82,10 +89,14 @@ internal class CourierStore(
|
||||
|
||||
@Synchronized
|
||||
fun remove(envelope: CourierEnvelope): Boolean {
|
||||
val envelopeKey = envelope.ciphertext.storageKey()
|
||||
val removed = stored.removeAll {
|
||||
it.envelope()?.ciphertext?.contentEquals(envelope.ciphertext) == true
|
||||
}
|
||||
if (removed) persist()
|
||||
if (removed) {
|
||||
sprayReservations.removeAll { it.envelopeKey == envelopeKey }
|
||||
persist()
|
||||
}
|
||||
return removed
|
||||
}
|
||||
|
||||
@ -118,10 +129,18 @@ internal class CourierStore(
|
||||
val result = mutableListOf<CourierEnvelope>()
|
||||
stored.forEach { record ->
|
||||
val envelope = record.envelope() ?: return@forEach
|
||||
val envelopeKey = envelope.ciphertext.storageKey()
|
||||
if (record.copies <= 1 || record.depositorKey == key || key in record.sprayedTo ||
|
||||
sprayReservations.any { it.envelopeKey == envelopeKey && it.courierKey == key } ||
|
||||
courierTags.any { it.contentEquals(envelope.recipientTag) }
|
||||
) return@forEach
|
||||
val given = record.copies / 2
|
||||
val reserved = sprayReservations
|
||||
.filter { it.envelopeKey == envelopeKey }
|
||||
.sumOf { it.copies }
|
||||
val available = record.copies - reserved
|
||||
if (available <= 1) return@forEach
|
||||
val given = available / 2
|
||||
sprayReservations += SprayReservation(envelopeKey, key, given)
|
||||
result += envelope.copy(copies = given.toUByte())
|
||||
}
|
||||
return result
|
||||
@ -130,28 +149,55 @@ internal class CourierStore(
|
||||
@Synchronized
|
||||
fun commitSpray(envelope: CourierEnvelope, courierNoiseKey: ByteArray): Boolean {
|
||||
val key = courierNoiseKey.toHex()
|
||||
val envelopeKey = envelope.ciphertext.storageKey()
|
||||
val reservationIndex = sprayReservations.indexOfFirst {
|
||||
it.envelopeKey == envelopeKey && it.courierKey == key && it.copies == envelope.copies.toInt()
|
||||
}
|
||||
if (reservationIndex < 0) return false
|
||||
val reservation = sprayReservations[reservationIndex]
|
||||
val record = stored.firstOrNull {
|
||||
it.envelope()?.ciphertext?.contentEquals(envelope.ciphertext) == true
|
||||
} ?: return false
|
||||
if (record.copies <= 1 || key in record.sprayedTo) return false
|
||||
val given = record.copies / 2
|
||||
if (given != envelope.copies.toInt()) return false
|
||||
record.copies -= given
|
||||
}
|
||||
val otherReservedCopies = sprayReservations.withIndex()
|
||||
.filter { (index, candidate) -> index != reservationIndex && candidate.envelopeKey == envelopeKey }
|
||||
.sumOf { it.value.copies }
|
||||
if (record == null || key in record.sprayedTo ||
|
||||
record.copies - otherReservedCopies <= reservation.copies
|
||||
) {
|
||||
sprayReservations.removeAt(reservationIndex)
|
||||
return false
|
||||
}
|
||||
record.copies -= reservation.copies
|
||||
record.sprayedTo += key
|
||||
sprayReservations.removeAt(reservationIndex)
|
||||
persist()
|
||||
return true
|
||||
}
|
||||
|
||||
@Synchronized
|
||||
fun cancelSpray(envelope: CourierEnvelope, courierNoiseKey: ByteArray): Boolean {
|
||||
val envelopeKey = envelope.ciphertext.storageKey()
|
||||
val courierKey = courierNoiseKey.toHex()
|
||||
return sprayReservations.removeAll {
|
||||
it.envelopeKey == envelopeKey && it.courierKey == courierKey && it.copies == envelope.copies.toInt()
|
||||
}
|
||||
}
|
||||
|
||||
@Synchronized
|
||||
fun wipe() {
|
||||
stored.clear()
|
||||
sprayReservations.clear()
|
||||
file.delete()
|
||||
cipher.destroyKey()
|
||||
}
|
||||
|
||||
private fun pruneExpired() {
|
||||
val nowMs = now().toULong()
|
||||
if (stored.removeAll { (it.envelope()?.expiry ?: 0u) <= nowMs }) persist()
|
||||
if (stored.removeAll { (it.envelope()?.expiry ?: 0u) <= nowMs }) {
|
||||
val retainedEnvelopeKeys = stored.mapNotNull { it.envelope()?.ciphertext?.storageKey() }.toSet()
|
||||
sprayReservations.removeAll { it.envelopeKey !in retainedEnvelopeKeys }
|
||||
persist()
|
||||
}
|
||||
}
|
||||
|
||||
private fun Stored.envelope(): CourierEnvelope? = try {
|
||||
@ -190,4 +236,6 @@ internal class CourierStore(
|
||||
}
|
||||
|
||||
private fun ByteArray.toHex(): String = joinToString("") { "%02x".format(it) }
|
||||
|
||||
private fun ByteArray.storageKey(): String = Base64.getEncoder().encodeToString(this)
|
||||
}
|
||||
|
||||
@ -10,7 +10,8 @@ data class CourierEnvelope(
|
||||
val recipientTag: ByteArray,
|
||||
val expiry: ULong,
|
||||
val ciphertext: ByteArray,
|
||||
val copies: UByte = 1u
|
||||
val copies: UByte = 1u,
|
||||
val prekeyID: UInt? = null
|
||||
) {
|
||||
companion object {
|
||||
const val TAG_LENGTH = 16
|
||||
@ -36,6 +37,7 @@ data class CourierEnvelope(
|
||||
var expiry: ULong? = null
|
||||
var ciphertext: ByteArray? = null
|
||||
var copies: UByte = 1u
|
||||
var prekeyID: UInt? = null
|
||||
while (offset + 3 <= data.size) {
|
||||
val type = data[offset].toUByte()
|
||||
val length = ((data[offset + 1].toInt() and 0xff) shl 8) or
|
||||
@ -53,6 +55,11 @@ data class CourierEnvelope(
|
||||
copies = value[0].toUByte()
|
||||
if (copies !in 2u.toUByte()..MAX_COPIES.toUByte()) return null
|
||||
}
|
||||
5 -> if (length == 4 && prekeyID == null) {
|
||||
prekeyID = ByteBuffer.wrap(value).order(ByteOrder.BIG_ENDIAN).int.toUInt()
|
||||
} else {
|
||||
return null
|
||||
}
|
||||
}
|
||||
}
|
||||
if (offset != data.size) return null
|
||||
@ -60,7 +67,7 @@ data class CourierEnvelope(
|
||||
val requiredExpiry = expiry ?: return null
|
||||
val requiredCiphertext = ciphertext ?: return null
|
||||
if (requiredCiphertext.isEmpty() || requiredCiphertext.size > MAX_CIPHERTEXT_BYTES) return null
|
||||
return CourierEnvelope(requiredTag, requiredExpiry, requiredCiphertext, copies)
|
||||
return CourierEnvelope(requiredTag, requiredExpiry, requiredCiphertext, copies, prekeyID)
|
||||
}
|
||||
}
|
||||
|
||||
@ -73,6 +80,9 @@ data class CourierEnvelope(
|
||||
fields += 2 to ByteBuffer.allocate(8).order(ByteOrder.BIG_ENDIAN).putLong(expiry.toLong()).array()
|
||||
fields += 3 to ciphertext
|
||||
if (copies > 1u) fields += 4 to byteArrayOf(copies.toByte())
|
||||
prekeyID?.let {
|
||||
fields += 5 to ByteBuffer.allocate(4).order(ByteOrder.BIG_ENDIAN).putInt(it.toInt()).array()
|
||||
}
|
||||
val size = fields.sumOf { 3 + it.second.size }
|
||||
val buffer = ByteBuffer.allocate(size).order(ByteOrder.BIG_ENDIAN)
|
||||
fields.forEach { (type, value) ->
|
||||
@ -92,8 +102,14 @@ data class CourierEnvelope(
|
||||
|
||||
override fun equals(other: Any?): Boolean = other is CourierEnvelope &&
|
||||
recipientTag.contentEquals(other.recipientTag) && expiry == other.expiry &&
|
||||
ciphertext.contentEquals(other.ciphertext) && copies == other.copies
|
||||
ciphertext.contentEquals(other.ciphertext) && copies == other.copies && prekeyID == other.prekeyID
|
||||
|
||||
override fun hashCode(): Int = 31 * (31 * recipientTag.contentHashCode() + expiry.hashCode()) +
|
||||
31 * ciphertext.contentHashCode() + copies.hashCode()
|
||||
override fun hashCode(): Int {
|
||||
var result = recipientTag.contentHashCode()
|
||||
result = 31 * result + expiry.hashCode()
|
||||
result = 31 * result + ciphertext.contentHashCode()
|
||||
result = 31 * result + copies.hashCode()
|
||||
result = 31 * result + (prekeyID?.hashCode() ?: 0)
|
||||
return result
|
||||
}
|
||||
}
|
||||
|
||||
@ -1,6 +1,8 @@
|
||||
package com.bitchat.android.model
|
||||
|
||||
import com.bitchat.android.sync.SyncDefaults
|
||||
import com.bitchat.android.sync.GCSFilter
|
||||
import com.bitchat.android.sync.SyncTypeFlags
|
||||
|
||||
/**
|
||||
* REQUEST_SYNC payload using GCS (Golomb-Coded Set) parameters.
|
||||
@ -8,11 +10,15 @@ import com.bitchat.android.sync.SyncDefaults
|
||||
* - 0x01: P (uint8) — Golomb-Rice parameter
|
||||
* - 0x02: M (uint32, big-endian) — hash range (N * 2^P)
|
||||
* - 0x03: data (opaque) — GR bitstream bytes
|
||||
* - 0x04: types (compact little-endian SyncTypeFlags) — packet types covered by the filter
|
||||
* - 0x05: sinceTimestamp (uint64, big-endian) — oldest timestamp covered by the filter
|
||||
*/
|
||||
data class RequestSyncPacket(
|
||||
val p: Int,
|
||||
val m: Long,
|
||||
val data: ByteArray
|
||||
val data: ByteArray,
|
||||
val types: SyncTypeFlags? = null,
|
||||
val sinceTimestamp: ULong? = null
|
||||
) {
|
||||
fun encode(): ByteArray {
|
||||
val out = ArrayList<Byte>()
|
||||
@ -38,6 +44,15 @@ data class RequestSyncPacket(
|
||||
)
|
||||
// data
|
||||
putTLV(0x03, data)
|
||||
types?.encode()?.let { putTLV(0x04, it) }
|
||||
sinceTimestamp?.let { timestamp ->
|
||||
putTLV(
|
||||
0x05,
|
||||
ByteArray(8) { index ->
|
||||
(timestamp shr ((7 - index) * 8) and 0xffu).toByte()
|
||||
}
|
||||
)
|
||||
}
|
||||
return out.toByteArray()
|
||||
}
|
||||
|
||||
@ -50,6 +65,8 @@ data class RequestSyncPacket(
|
||||
var p: Int? = null
|
||||
var m: Long? = null
|
||||
var payload: ByteArray? = null
|
||||
var types: SyncTypeFlags? = null
|
||||
var sinceTimestamp: ULong? = null
|
||||
|
||||
while (off + 3 <= data.size) {
|
||||
val t = (data[off].toInt() and 0xFF); off += 1
|
||||
@ -69,14 +86,20 @@ data class RequestSyncPacket(
|
||||
if (v.size > MAX_ACCEPT_FILTER_BYTES) return null
|
||||
payload = v
|
||||
}
|
||||
0x04 -> SyncTypeFlags.decode(v)?.let { types = it }
|
||||
0x05 -> if (v.size == 8) {
|
||||
var timestamp = 0uL
|
||||
v.forEach { byte -> timestamp = (timestamp shl 8) or byte.toUByte().toULong() }
|
||||
sinceTimestamp = timestamp
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
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)
|
||||
if (pp !in 1..GCSFilter.MAX_P || mm <= 0L) return null
|
||||
return RequestSyncPacket(pp, mm, dd, types, sinceTimestamp)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@ -21,9 +21,12 @@ 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
|
||||
val data: ByteArray, // Encoded GR bitstream
|
||||
val includedCount: Int // Number of newest-first input IDs actually encoded
|
||||
)
|
||||
|
||||
const val MAX_P = 32
|
||||
|
||||
// Derive P from target FPR; FPR ~= 1 / 2^P
|
||||
fun deriveP(targetFpr: Double): Int {
|
||||
val f = targetFpr.coerceIn(0.000001, 0.25)
|
||||
@ -66,7 +69,12 @@ object GCSFilter {
|
||||
encoded = encode(mapped, p)
|
||||
}
|
||||
|
||||
return Params(p = p, m = finalM, data = encoded)
|
||||
return Params(
|
||||
p = p,
|
||||
m = finalM,
|
||||
data = encoded,
|
||||
includedCount = if (encoded.isEmpty()) 0 else trimmedN
|
||||
)
|
||||
}
|
||||
|
||||
fun decodeToSortedSet(p: Int, m: Long, data: ByteArray): LongArray {
|
||||
@ -196,4 +204,3 @@ object GCSFilter {
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@ -175,22 +175,27 @@ class GossipSyncManager(
|
||||
}
|
||||
|
||||
private fun sendRequestSync() {
|
||||
val payload = buildGcsPayload()
|
||||
|
||||
val packet = BitchatPacket(
|
||||
type = MessageType.REQUEST_SYNC.value,
|
||||
senderID = hexStringToByteArray(myPeerID),
|
||||
timestamp = System.currentTimeMillis().toULong(),
|
||||
payload = payload,
|
||||
ttl = com.bitchat.android.util.AppConstants.SYNC_TTL_HOPS // neighbors only
|
||||
)
|
||||
// Sign and broadcast
|
||||
val signed = delegate?.signPacketForBroadcast(packet) ?: packet
|
||||
delegate?.sendPacket(signed)
|
||||
listOf(
|
||||
SyncTypeFlags.PUBLIC_MESSAGES,
|
||||
SyncTypeFlags.FRAGMENT,
|
||||
SyncTypeFlags.FILE_TRANSFER
|
||||
).forEach { types ->
|
||||
val payload = buildGcsPayload(types)
|
||||
val packet = BitchatPacket(
|
||||
type = MessageType.REQUEST_SYNC.value,
|
||||
senderID = hexStringToByteArray(myPeerID),
|
||||
timestamp = System.currentTimeMillis().toULong(),
|
||||
payload = payload,
|
||||
ttl = com.bitchat.android.util.AppConstants.SYNC_TTL_HOPS // neighbors only
|
||||
)
|
||||
val signed = delegate?.signPacketForBroadcast(packet) ?: packet
|
||||
delegate?.sendPacket(signed)
|
||||
}
|
||||
}
|
||||
|
||||
private fun sendRequestSyncToPeer(peerID: String) {
|
||||
val payload = buildGcsPayload()
|
||||
val types = SyncTypeFlags.PUBLIC_MESSAGES.union(SyncTypeFlags.FRAGMENTS_AND_FILES)
|
||||
val payload = buildGcsPayload(types)
|
||||
|
||||
val packet = BitchatPacket(
|
||||
type = MessageType.REQUEST_SYNC.value,
|
||||
@ -207,6 +212,8 @@ class GossipSyncManager(
|
||||
}
|
||||
|
||||
fun handleRequestSync(fromPeerID: String, request: RequestSyncPacket) {
|
||||
val requestedTypes = request.types ?: SyncTypeFlags.PUBLIC_MESSAGES
|
||||
val sinceTimestamp = request.sinceTimestamp
|
||||
// Decode GCS into sorted set for membership checks
|
||||
val sorted = GCSFilter.decodeToSortedSet(request.p, request.m, request.data)
|
||||
fun mightContain(id: ByteArray): Boolean {
|
||||
@ -215,30 +222,38 @@ class GossipSyncManager(
|
||||
return GCSFilter.contains(sorted, nonZeroV)
|
||||
}
|
||||
|
||||
// 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 = com.bitchat.android.util.AppConstants.SYNC_TTL_HOPS)
|
||||
delegate?.sendPacketToPeer(fromPeerID, toSend)
|
||||
Log.d(TAG, "Sent sync announce: Type ${toSend.type} from ${toSend.senderID.toHexString()} to $fromPeerID packet id ${idBytes.toHexString()}")
|
||||
// Announcements are exempt from the cursor: only the latest per peer is retained,
|
||||
// and peers need their signing keys before they can verify other sync responses.
|
||||
if (requestedTypes.contains(MessageType.ANNOUNCE)) {
|
||||
for ((_, pair) in latestAnnouncementByPeer.entries) {
|
||||
val (id, pkt) = pair
|
||||
val idBytes = hexToBytes(id)
|
||||
if (!mightContain(idBytes)) {
|
||||
val toSend = pkt.copy(ttl = com.bitchat.android.util.AppConstants.SYNC_TTL_HOPS)
|
||||
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 = com.bitchat.android.util.AppConstants.SYNC_TTL_HOPS)
|
||||
delegate?.sendPacketToPeer(fromPeerID, toSend)
|
||||
Log.d(TAG, "Sent sync message: Type ${toSend.type} to $fromPeerID packet id ${idBytes.toHexString()}")
|
||||
if (requestedTypes.contains(MessageType.MESSAGE)) {
|
||||
val toSendMsgs = synchronized(messages) { messages.values.toList() }
|
||||
for (pkt in toSendMsgs) {
|
||||
if (sinceTimestamp != null && pkt.timestamp < sinceTimestamp) continue
|
||||
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)
|
||||
Log.d(TAG, "Sent sync message: Type ${toSend.type} to $fromPeerID packet id ${idBytes.toHexString()}")
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
val toSendFragments = synchronized(fragments) { fragments.values.toList() }
|
||||
for (pkt in toSendFragments) {
|
||||
val type = MessageType.fromValue(pkt.type) ?: continue
|
||||
if (!requestedTypes.contains(type)) continue
|
||||
if (sinceTimestamp != null && pkt.timestamp < sinceTimestamp) continue
|
||||
val idBytes = PacketIdUtil.computeIdBytes(pkt)
|
||||
if (!mightContain(idBytes)) {
|
||||
delegate?.sendPacketToPeer(fromPeerID, pkt.copy(ttl = com.bitchat.android.util.AppConstants.SYNC_TTL_HOPS))
|
||||
@ -271,18 +286,24 @@ class GossipSyncManager(
|
||||
return out
|
||||
}
|
||||
|
||||
private fun buildGcsPayload(): ByteArray {
|
||||
// Collect candidates: latest announcement per peer + recent broadcast messages
|
||||
internal fun buildGcsPayload(types: SyncTypeFlags): ByteArray {
|
||||
// Collect only the packet types represented by this filter.
|
||||
val list = ArrayList<BitchatPacket>()
|
||||
// announcements
|
||||
for ((_, pair) in latestAnnouncementByPeer) {
|
||||
list.add(pair.second)
|
||||
if (types.contains(MessageType.ANNOUNCE)) {
|
||||
for ((_, pair) in latestAnnouncementByPeer) {
|
||||
list.add(pair.second)
|
||||
}
|
||||
}
|
||||
// messages
|
||||
synchronized(messages) {
|
||||
list.addAll(messages.values)
|
||||
if (types.contains(MessageType.MESSAGE)) {
|
||||
synchronized(messages) {
|
||||
list.addAll(messages.values)
|
||||
}
|
||||
}
|
||||
synchronized(fragments) {
|
||||
list.addAll(fragments.values.filter { packet ->
|
||||
MessageType.fromValue(packet.type)?.let(types::contains) == true
|
||||
})
|
||||
}
|
||||
synchronized(fragments) { list.addAll(fragments.values) }
|
||||
// sort by timestamp desc, then take up to min(seenCapacity, fit capacity)
|
||||
list.sortByDescending { it.timestamp.toLong() }
|
||||
|
||||
@ -294,12 +315,21 @@ class GossipSyncManager(
|
||||
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()
|
||||
return RequestSyncPacket(p = p0, m = 1, data = ByteArray(0), types = types).encode()
|
||||
}
|
||||
val ids = list.take(takeN).map { pkt -> PacketIdUtil.computeIdBytes(pkt) }
|
||||
val included = list.take(takeN)
|
||||
val ids = included.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()
|
||||
val covered = params.includedCount
|
||||
val sinceTimestamp = if (covered in 1 until list.size) included[covered - 1].timestamp else null
|
||||
return RequestSyncPacket(
|
||||
p = params.p,
|
||||
m = mVal,
|
||||
data = params.data,
|
||||
types = types,
|
||||
sinceTimestamp = sinceTimestamp
|
||||
).encode()
|
||||
}
|
||||
|
||||
// Announcements age out quickly; public history remains independently sync-able for six hours.
|
||||
|
||||
68
app/src/main/java/com/bitchat/android/sync/SyncTypeFlags.kt
Normal file
68
app/src/main/java/com/bitchat/android/sync/SyncTypeFlags.kt
Normal file
@ -0,0 +1,68 @@
|
||||
package com.bitchat.android.sync
|
||||
|
||||
import com.bitchat.android.protocol.MessageType
|
||||
|
||||
/** Compact little-endian bitfield matching iOS SyncTypeFlags. */
|
||||
@JvmInline
|
||||
value class SyncTypeFlags private constructor(val rawValue: ULong) {
|
||||
companion object {
|
||||
private const val KNOWN_TYPE_MASK: ULong = 0xffu
|
||||
|
||||
val ANNOUNCE = fromMessageTypes(MessageType.ANNOUNCE)
|
||||
val MESSAGE = fromMessageTypes(MessageType.MESSAGE)
|
||||
val FRAGMENT = fromMessageTypes(MessageType.FRAGMENT)
|
||||
val FILE_TRANSFER = fromMessageTypes(MessageType.FILE_TRANSFER)
|
||||
val PUBLIC_MESSAGES = fromMessageTypes(MessageType.ANNOUNCE, MessageType.MESSAGE)
|
||||
val FRAGMENTS_AND_FILES = fromMessageTypes(MessageType.FRAGMENT, MessageType.FILE_TRANSFER)
|
||||
|
||||
fun fromRawValue(rawValue: ULong): SyncTypeFlags = SyncTypeFlags(rawValue and KNOWN_TYPE_MASK)
|
||||
|
||||
fun fromMessageTypes(vararg types: MessageType): SyncTypeFlags {
|
||||
var rawValue = 0uL
|
||||
types.forEach { type ->
|
||||
bitIndex(type)?.let { bit -> rawValue = rawValue or (1uL shl bit) }
|
||||
}
|
||||
return fromRawValue(rawValue)
|
||||
}
|
||||
|
||||
fun decode(data: ByteArray): SyncTypeFlags? {
|
||||
if (data.size !in 1..8) return null
|
||||
var rawValue = 0uL
|
||||
data.forEachIndexed { index, byte ->
|
||||
rawValue = rawValue or (byte.toUByte().toULong() shl (index * 8))
|
||||
}
|
||||
return fromRawValue(rawValue)
|
||||
}
|
||||
|
||||
private fun bitIndex(type: MessageType): Int? = when (type) {
|
||||
MessageType.ANNOUNCE -> 0
|
||||
MessageType.MESSAGE -> 1
|
||||
MessageType.LEAVE -> 2
|
||||
MessageType.NOISE_HANDSHAKE -> 3
|
||||
MessageType.NOISE_ENCRYPTED -> 4
|
||||
MessageType.FRAGMENT -> 5
|
||||
MessageType.REQUEST_SYNC -> 6
|
||||
MessageType.FILE_TRANSFER -> 7
|
||||
MessageType.COURIER_ENVELOPE,
|
||||
MessageType.VOICE_FRAME -> null
|
||||
}
|
||||
}
|
||||
|
||||
fun contains(type: MessageType): Boolean {
|
||||
val bit = bitIndex(type) ?: return false
|
||||
return (rawValue and (1uL shl bit)) != 0uL
|
||||
}
|
||||
|
||||
fun union(other: SyncTypeFlags): SyncTypeFlags = fromRawValue(rawValue or other.rawValue)
|
||||
|
||||
fun encode(): ByteArray? {
|
||||
if (rawValue == 0uL) return null
|
||||
var remaining = rawValue
|
||||
val bytes = ArrayList<Byte>(8)
|
||||
while (remaining != 0uL && bytes.size < 8) {
|
||||
bytes += (remaining and 0xffu).toByte()
|
||||
remaining = remaining shr 8
|
||||
}
|
||||
return bytes.toByteArray()
|
||||
}
|
||||
}
|
||||
@ -9,6 +9,7 @@ import com.bitchat.android.model.NoisePayloadType
|
||||
import com.bitchat.android.model.PeerCapabilities
|
||||
import com.bitchat.android.model.PrivateMessagePacket
|
||||
import com.bitchat.android.model.RequestSyncPacket
|
||||
import com.bitchat.android.sync.SyncTypeFlags
|
||||
import com.bitchat.android.model.UnknownAnnouncementTLV
|
||||
import com.bitchat.android.protocol.BinaryProtocol
|
||||
import com.bitchat.android.protocol.BitchatPacket
|
||||
@ -203,9 +204,14 @@ class ClientRewriteWireContractTest {
|
||||
val request = RequestSyncPacket(
|
||||
p = 19,
|
||||
m = 0x01020304L,
|
||||
data = hex("aabb")
|
||||
data = hex("aabb"),
|
||||
types = SyncTypeFlags.FRAGMENTS_AND_FILES,
|
||||
sinceTimestamp = 0x0102030405060708u
|
||||
)
|
||||
val wire = hex(
|
||||
"0100011302000401020304030002aabb" +
|
||||
"040001a00500080102030405060708"
|
||||
)
|
||||
val wire = hex("0100011302000401020304030002aabb")
|
||||
|
||||
assertArrayEquals(wire, request.encode())
|
||||
assertSyncRequestEquals(request, RequestSyncPacket.decode(wire))
|
||||
@ -297,5 +303,7 @@ class ClientRewriteWireContractTest {
|
||||
assertEquals(expected.p, actual!!.p)
|
||||
assertEquals(expected.m, actual.m)
|
||||
assertArrayEquals(expected.data, actual.data)
|
||||
assertEquals(expected.types, actual.types)
|
||||
assertEquals(expected.sinceTimestamp, actual.sinceTimestamp)
|
||||
}
|
||||
}
|
||||
|
||||
@ -71,6 +71,49 @@ class CourierStoreTest {
|
||||
assertEquals(2u.toUByte(), reloaded().sprayCopiesFor(courier).single().copies)
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `concurrent spray reservations cannot over allocate custody`() {
|
||||
val store = newStore()
|
||||
assertTrue(store.deposit(envelope(1, copies = 4u), verified, CourierDepositTier.VERIFIED))
|
||||
val firstCourier = ByteArray(32) { 12 }
|
||||
val secondCourier = ByteArray(32) { 13 }
|
||||
val thirdCourier = ByteArray(32) { 14 }
|
||||
|
||||
val first = store.sprayCopiesFor(firstCourier).single()
|
||||
val second = store.sprayCopiesFor(secondCourier).single()
|
||||
|
||||
assertEquals(2u.toUByte(), first.copies)
|
||||
assertEquals(1u.toUByte(), second.copies)
|
||||
assertTrue(store.sprayCopiesFor(firstCourier).isEmpty())
|
||||
assertTrue(store.sprayCopiesFor(thirdCourier).isEmpty())
|
||||
assertTrue(store.commitSpray(second, secondCourier))
|
||||
assertTrue(store.commitSpray(first, firstCourier))
|
||||
assertTrue(reloaded().sprayCopiesFor(thirdCourier).isEmpty())
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `cancelled spray reservation makes its copies eligible again`() {
|
||||
val store = newStore()
|
||||
assertTrue(store.deposit(envelope(1, copies = 4u), verified, CourierDepositTier.VERIFIED))
|
||||
val courier = ByteArray(32) { 15 }
|
||||
|
||||
val first = store.sprayCopiesFor(courier).single()
|
||||
|
||||
assertTrue(store.cancelSpray(first, courier))
|
||||
assertEquals(2u.toUByte(), store.sprayCopiesFor(courier).single().copies)
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `stored prekey envelope retains its prekey id through spray`() {
|
||||
val store = newStore()
|
||||
val envelope = envelope(1, copies = 4u).copy(prekeyID = 0x11223344u)
|
||||
assertTrue(store.deposit(envelope, verified, CourierDepositTier.VERIFIED))
|
||||
|
||||
val spray = store.sprayCopiesFor(ByteArray(32) { 16 }).single()
|
||||
|
||||
assertEquals(0x11223344u, spray.prekeyID)
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `wipe deletes sealed custody and destroys key`() {
|
||||
val store = newStore()
|
||||
|
||||
@ -12,13 +12,15 @@ class CourierEnvelopeTest {
|
||||
recipientTag = ByteArray(16) { it.toByte() },
|
||||
expiry = 0x0102030405060708u,
|
||||
ciphertext = byteArrayOf(0xaa.toByte(), 0xbb.toByte(), 0xcc.toByte()),
|
||||
copies = 4u
|
||||
copies = 4u,
|
||||
prekeyID = 0x11223344u
|
||||
)
|
||||
assertEquals(
|
||||
"010010000102030405060708090a0b0c0d0e0f" +
|
||||
"0200080102030405060708" +
|
||||
"030003aabbcc" +
|
||||
"04000104",
|
||||
"04000104" +
|
||||
"05000411223344",
|
||||
envelope.encode()!!.toHex()
|
||||
)
|
||||
}
|
||||
@ -40,6 +42,33 @@ class CourierEnvelopeTest {
|
||||
assertEquals(1u.toUByte(), CourierEnvelope.decode(legacy.encode()!!)!!.copies)
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `prekey id survives decode re-encode and copy changes`() {
|
||||
val envelope = CourierEnvelope(
|
||||
recipientTag = ByteArray(16) { it.toByte() },
|
||||
expiry = 123456789u,
|
||||
ciphertext = ByteArray(96) { (it + 1).toByte() },
|
||||
copies = 4u,
|
||||
prekeyID = 0xfedcba98u
|
||||
)
|
||||
|
||||
val decoded = CourierEnvelope.decode(envelope.encode()!!)!!
|
||||
|
||||
assertEquals(0xfedcba98u, decoded.prekeyID)
|
||||
assertArrayEquals(envelope.encode(), decoded.copy(copies = 4u).encode())
|
||||
assertEquals(0xfedcba98u, decoded.copy(copies = 2u).prekeyID)
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `invalid or duplicate prekey fields are rejected`() {
|
||||
val envelope = CourierEnvelope(ByteArray(16), 1u, ByteArray(32) { 1 }, prekeyID = 7u)
|
||||
val encoded = envelope.encode()!!
|
||||
val prekeyField = encoded.copyOfRange(encoded.size - 7, encoded.size)
|
||||
|
||||
assertNull(CourierEnvelope.decode(encoded + prekeyField))
|
||||
assertNull(CourierEnvelope.decode(encoded.copyOf(encoded.size - 1)))
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `tag rotates daily and matches adjacent day for clock skew`() {
|
||||
val key = ByteArray(32) { 0x2a }
|
||||
|
||||
@ -18,6 +18,7 @@ class GCSFilterTest {
|
||||
|
||||
// Build filter with plenty of bytes (no trimming)
|
||||
val params = GCSFilter.buildFilter(ids, maxBytes = 400, targetFpr = 0.01)
|
||||
assertEquals(ids.size, params.includedCount)
|
||||
val sorted = GCSFilter.decodeToSortedSet(params.p, params.m, params.data)
|
||||
|
||||
for (id in ids) {
|
||||
@ -47,7 +48,7 @@ class GCSFilterTest {
|
||||
val sorted = GCSFilter.decodeToSortedSet(params.p, params.m, params.data)
|
||||
|
||||
// Let's verify that the first trimmedN elements in ids are all matched
|
||||
val trimmedN = (params.m ushr params.p).toInt()
|
||||
val trimmedN = params.includedCount
|
||||
assertTrue("At least some elements should have been encoded", trimmedN > 0)
|
||||
|
||||
val retainedIds = ids.take(trimmedN)
|
||||
|
||||
@ -36,7 +36,15 @@ class GossipSyncManagerTest {
|
||||
manager.onPublicPacketSeen(packet(MessageType.FRAGMENT, now - 20 * 60 * 1000L, 4))
|
||||
manager.onPublicPacketSeen(packet(MessageType.FILE_TRANSFER, now - 10 * 60 * 1000L, 5))
|
||||
|
||||
manager.handleRequestSync("peer", RequestSyncPacket(p = 1, m = 1, data = byteArrayOf()))
|
||||
manager.handleRequestSync(
|
||||
"peer",
|
||||
RequestSyncPacket(
|
||||
p = 1,
|
||||
m = 1,
|
||||
data = byteArrayOf(),
|
||||
types = SyncTypeFlags.PUBLIC_MESSAGES.union(SyncTypeFlags.FRAGMENTS_AND_FILES)
|
||||
)
|
||||
)
|
||||
|
||||
assertEquals(listOf(1, 3, 5), sent.map { it.payload.single().toInt() })
|
||||
}
|
||||
@ -57,11 +65,97 @@ class GossipSyncManagerTest {
|
||||
manager.onPublicPacketSeen(packet(MessageType.MESSAGE, now + 11 * 60_000L, 4))
|
||||
manager.onPublicPacketSeen(packet(MessageType.FRAGMENT, now + 11 * 60_000L, 5))
|
||||
|
||||
manager.handleRequestSync("peer", RequestSyncPacket(p = 1, m = 1, data = byteArrayOf()))
|
||||
manager.handleRequestSync(
|
||||
"peer",
|
||||
RequestSyncPacket(
|
||||
p = 1,
|
||||
m = 1,
|
||||
data = byteArrayOf(),
|
||||
types = SyncTypeFlags.PUBLIC_MESSAGES.union(SyncTypeFlags.FRAGMENTS_AND_FILES)
|
||||
)
|
||||
)
|
||||
|
||||
assertEquals(listOf(1, 2, 3), sent.map { it.payload.single().toInt() })
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `type scoped fragment request does not replay other packet classes`() {
|
||||
val sent = mutableListOf<BitchatPacket>()
|
||||
val manager = GossipSyncManager("1111222233334444", TestScope(), config)
|
||||
manager.delegate = recordingDelegate(sent)
|
||||
val now = System.currentTimeMillis()
|
||||
manager.onPublicPacketSeen(packet(MessageType.MESSAGE, now, 1))
|
||||
manager.onPublicPacketSeen(packet(MessageType.FRAGMENT, now, 2))
|
||||
manager.onPublicPacketSeen(packet(MessageType.FILE_TRANSFER, now, 3))
|
||||
|
||||
manager.handleRequestSync(
|
||||
"peer",
|
||||
RequestSyncPacket(
|
||||
p = 1,
|
||||
m = 1,
|
||||
data = byteArrayOf(),
|
||||
types = SyncTypeFlags.FRAGMENT
|
||||
)
|
||||
)
|
||||
|
||||
assertEquals(listOf(2), sent.map { it.payload.single().toInt() })
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `coverage cursor prevents replay of an identical history tail`() {
|
||||
val tinyFilterConfig = object : GossipSyncManager.ConfigProvider {
|
||||
override fun seenCapacity() = 100
|
||||
override fun gcsMaxBytes() = 1
|
||||
override fun gcsTargetFpr() = 0.01
|
||||
}
|
||||
val requester = GossipSyncManager("1111222233334444", TestScope(), tinyFilterConfig)
|
||||
val responder = GossipSyncManager("5555666677778888", TestScope(), tinyFilterConfig)
|
||||
val now = System.currentTimeMillis()
|
||||
val history = listOf(
|
||||
packet(MessageType.MESSAGE, now - 100, 1),
|
||||
packet(MessageType.MESSAGE, now - 200, 2),
|
||||
packet(MessageType.MESSAGE, now - 300, 3)
|
||||
)
|
||||
history.forEach {
|
||||
requester.onPublicPacketSeen(it)
|
||||
responder.onPublicPacketSeen(it)
|
||||
}
|
||||
val request = RequestSyncPacket.decode(
|
||||
requester.buildGcsPayload(SyncTypeFlags.PUBLIC_MESSAGES)
|
||||
)!!
|
||||
val sent = mutableListOf<BitchatPacket>()
|
||||
responder.delegate = recordingDelegate(sent)
|
||||
|
||||
responder.handleRequestSync("peer", request)
|
||||
|
||||
assertEquals(SyncTypeFlags.PUBLIC_MESSAGES, request.types)
|
||||
assertEquals(history.first().timestamp, request.sinceTimestamp)
|
||||
assertEquals(emptyList<Int>(), sent.map { it.payload.single().toInt() })
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `legacy request without type metadata remains public message only`() {
|
||||
val sent = mutableListOf<BitchatPacket>()
|
||||
val manager = GossipSyncManager("1111222233334444", TestScope(), config)
|
||||
manager.delegate = recordingDelegate(sent)
|
||||
val now = System.currentTimeMillis()
|
||||
manager.onPublicPacketSeen(packet(MessageType.MESSAGE, now, 1))
|
||||
manager.onPublicPacketSeen(packet(MessageType.FRAGMENT, now, 2))
|
||||
|
||||
manager.handleRequestSync("peer", RequestSyncPacket(p = 1, m = 1, data = byteArrayOf()))
|
||||
|
||||
assertEquals(listOf(1), sent.map { it.payload.single().toInt() })
|
||||
}
|
||||
|
||||
private fun recordingDelegate(sent: MutableList<BitchatPacket>) =
|
||||
object : GossipSyncManager.Delegate {
|
||||
override fun sendPacket(packet: BitchatPacket) = Unit
|
||||
override fun sendPacketToPeer(peerID: String, packet: BitchatPacket) {
|
||||
sent += packet
|
||||
}
|
||||
override fun signPacketForBroadcast(packet: BitchatPacket) = packet
|
||||
}
|
||||
|
||||
private fun packet(type: MessageType, timestamp: Long, marker: Int) = BitchatPacket(
|
||||
type = type.value,
|
||||
senderID = ByteArray(8) { 1 },
|
||||
|
||||
@ -0,0 +1,36 @@
|
||||
package com.bitchat.android.sync
|
||||
|
||||
import com.bitchat.android.protocol.MessageType
|
||||
import org.junit.Assert.assertArrayEquals
|
||||
import org.junit.Assert.assertEquals
|
||||
import org.junit.Assert.assertFalse
|
||||
import org.junit.Assert.assertNull
|
||||
import org.junit.Assert.assertTrue
|
||||
import org.junit.Test
|
||||
|
||||
class SyncTypeFlagsTest {
|
||||
@Test
|
||||
fun `wire encoding uses compact little endian iOS bit positions`() {
|
||||
val flags = SyncTypeFlags.PUBLIC_MESSAGES.union(SyncTypeFlags.FRAGMENTS_AND_FILES)
|
||||
|
||||
assertArrayEquals(byteArrayOf(0xa3.toByte()), flags.encode())
|
||||
assertEquals(flags, SyncTypeFlags.decode(byteArrayOf(0xa3.toByte())))
|
||||
assertTrue(flags.contains(MessageType.ANNOUNCE))
|
||||
assertTrue(flags.contains(MessageType.FILE_TRANSFER))
|
||||
assertFalse(flags.contains(MessageType.NOISE_HANDSHAKE))
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `unknown extended bits are normalized away`() {
|
||||
val decoded = SyncTypeFlags.decode(byteArrayOf(0x03, 0xfc.toByte()))!!
|
||||
|
||||
assertEquals(SyncTypeFlags.PUBLIC_MESSAGES, decoded)
|
||||
assertArrayEquals(byteArrayOf(0x03), decoded.encode())
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `empty and oversized fields are rejected`() {
|
||||
assertNull(SyncTypeFlags.decode(byteArrayOf()))
|
||||
assertNull(SyncTypeFlags.decode(ByteArray(9)))
|
||||
}
|
||||
}
|
||||
@ -15,9 +15,9 @@ The remaining implementation work and milestone progress are tracked in
|
||||
| Outer mesh packet | v1/v2 header widths, big-endian fields, flags, section order, route placement, signature placement, padding, compression, signing bytes | `BinaryProtocolTest`, `ClientRewriteWireContractTest` |
|
||||
| Chat payload | Flag bits, millisecond timestamp, UTF-8 byte lengths, encrypted-content substitution, optional-field order | `ClientRewriteWireContractTest` |
|
||||
| Inner payloads | Noise type bytes, private-message TLVs, peer-state TLVs, file-transfer TLVs, live-voice bursts, fragment header, sync request TLVs | `ClientRewriteWireContractTest`, `AuthenticatedPeerStateTest`, `PrivateMediaTransferPreparerTest`, `VoiceBurstPacketTest`, `FragmentManagerTest` |
|
||||
| Store and forward | Courier type `0x04`, rotating HMAC recipient tags, Noise X seals, copy-budget TLV, 24-hour expiry, bounded tiered custody | `CourierEnvelopeTest`, `NoiseCourierTest`, `MessageRouterTest` |
|
||||
| Store and forward | Courier type `0x04`, rotating HMAC recipient tags, Noise X seals, copy-budget and prekey-ID TLVs, 24-hour expiry, bounded tiered custody | `CourierEnvelopeTest`, `NoiseCourierTest`, `MessageRouterTest` |
|
||||
| Identity/security | Announcement extensions, capability bitfield endianness, Noise static-key binding, handshake identity binding, signatures | `IdentityAnnouncementTest`, `NoiseSessionManagerIdentityBindingTest`, `ClientRewritePrimitiveContractTest` |
|
||||
| Sync/routing | Stable packet IDs, GCS bitstream, replay collapse, TTL handling, relay choice, confirmed graph edges | `ClientRewritePrimitiveContractTest`, `GCSFilterTest`, `PacketRelayManagerTest`, `MeshGraphServiceTest`, `TransportBridgeServiceTest` |
|
||||
| Sync/routing | Stable packet IDs, GCS bitstream, type-scoped filters, bounded-history cursors, replay collapse, TTL handling, relay choice, confirmed graph edges | `ClientRewriteWireContractTest`, `ClientRewritePrimitiveContractTest`, `GCSFilterTest`, `GossipSyncManagerTest`, `PacketRelayManagerTest`, `MeshGraphServiceTest`, `TransportBridgeServiceTest` |
|
||||
| Nostr | Bech32, secp256k1 key derivation, NIP-01 event IDs/signatures, NIP-44 authenticated encryption, NIP-13 PoW, authenticated NIP-17 seals | `ClientRewriteNostrContractTest`, `NostrProtocolTest` |
|
||||
| Application state | Peer unions, canonical private conversations, chronological history, delivery/read behavior, media migration policy | `AppStateStoreTest`, `PrivateChatManagerTest`, `MediaSendingManagerMigrationTest` |
|
||||
|
||||
|
||||
32
docs/sync.md
32
docs/sync.md
@ -1,17 +1,19 @@
|
||||
# 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.
|
||||
This document specifies the gossip-based synchronization feature for BitChat, inspired by Plumtree. It ensures eventual consistency of public packets across nodes via periodic, type-scoped 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).
|
||||
- Broadcast fragments (MessageType.FRAGMENT)
|
||||
- Broadcast file transfers (MessageType.FILE_TRANSFER)
|
||||
- Default retention is 100 recent packets per local store (configurable in the debug sheet).
|
||||
- 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.
|
||||
- Every 30 seconds, Android sends separate public-message, fragment, and file-transfer REQUEST_SYNC rounds to immediate neighbors (local only; not relayed). Keeping the filters type-scoped prevents one high-volume class from crowding another out.
|
||||
- Additionally, 5 seconds after the first announcement from a newly directly connected peer is detected, a node sends a combined type-scoped REQUEST_SYNC only to that peer (unicast; local only).
|
||||
- The receiver checks only the packet types named by the request and sends packets absent from the filter. For announcements, only the latest announcement per peerID is sent.
|
||||
|
||||
This synchronization is strictly local (not relayed), ensuring only immediate neighbors participate and preventing wide-area flooding while converging content across the mesh.
|
||||
|
||||
@ -36,9 +38,7 @@ Implementation: `com.bitchat.android.sync.GCSFilter`.
|
||||
- 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
|
||||
- Select only the classes named by the request's `types` field, then sort matching packets by timestamp (descending). Android stores broadcast messages, the most recent ANNOUNCE per peer, broadcast fragments, and broadcast file transfers.
|
||||
- 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.
|
||||
@ -56,6 +56,10 @@ MessageType: `REQUEST_SYNC (0x21)`
|
||||
- 0x01: P (uint8) — Golomb‑Rice parameter
|
||||
- 0x02: M (uint32) — hash range N * 2^P
|
||||
- 0x03: data (opaque) — GCS bitstream (MSB‑first bit packing)
|
||||
- 0x04: types (1–8 byte compact little-endian bitfield) — packet classes represented by this filter
|
||||
- 0x05: sinceTimestamp (uint64, big-endian) — timestamp of the oldest packet represented when the candidate tail did not fit
|
||||
|
||||
`types` uses the cross-client bit mapping: ANNOUNCE=bit 0, MESSAGE=bit 1, LEAVE=bit 2, NOISE_HANDSHAKE=bit 3, NOISE_ENCRYPTED=bit 4, FRAGMENT=bit 5, REQUEST_SYNC=bit 6, and FILE_TRANSFER=bit 7. Implementations ignore bits they do not recognize. When `types` is absent, receivers use the legacy scope of ANNOUNCE plus broadcast MESSAGE.
|
||||
|
||||
Notes:
|
||||
- The GCS bitstream uses MSB‑first packing (bit 7 is the first bit in each byte).
|
||||
@ -71,7 +75,9 @@ Sender behavior:
|
||||
|
||||
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:
|
||||
- Restrict the diff to the packet classes in `types` (or the legacy public-message scope when absent).
|
||||
- For each locally stored packet ID in that scope:
|
||||
- If `sinceTimestamp` is present, skip messages, fragments, and files older than it because they were outside the requester's bounded filter rather than missing. ANNOUNCE is exempt so a peer can still obtain current verification keys.
|
||||
- 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.
|
||||
@ -90,6 +96,8 @@ Important: original packets are sent unmodified to preserve original signatures
|
||||
Included in sync:
|
||||
- Public broadcast messages: `MessageType.MESSAGE` with BROADCAST recipient (or null recipient).
|
||||
- Identity announcements: `MessageType.ANNOUNCE`.
|
||||
- Broadcast fragments: `MessageType.FRAGMENT` with BROADCAST recipient (or null recipient).
|
||||
- Broadcast files: `MessageType.FILE_TRANSFER` with BROADCAST recipient (or null recipient).
|
||||
- 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.
|
||||
|
||||
@ -130,8 +138,8 @@ The following items require consensus across all implementations to ensure inter
|
||||
|
||||
- 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”.
|
||||
- Payload encoding: TLV with 16‑bit big‑endian lengths; TLV types 0x01 = P (uint8), 0x02 = M (uint32), 0x03 = data (opaque), 0x04 = compact little-endian type flags, and 0x05 = big-endian coverage cursor.
|
||||
- Packet type and scope: REQUEST_SYNC = 0x21; local-only (not relayed); only requested, supported public packet classes are synchronized; ANNOUNCE de‑dupe is “latest per sender peerID”.
|
||||
|
||||
The following are requester‑defined and communicated or local policy (no global agreement required):
|
||||
|
||||
@ -142,7 +150,7 @@ The following are requester‑defined and communicated or local policy (no globa
|
||||
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.
|
||||
- Practical bounds: data length in [0, 1024]; P in [1, 32]; M up to 2^32‑1.
|
||||
|
||||
Versioning:
|
||||
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user