diff --git a/CHANGELOG.md b/CHANGELOG.md index 80b3914..d4d7fb0 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,17 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/). ## [Unreleased] +### Added + +- **Bulk lane on the Bluetooth link (0.4.7, A7.4):** the clipboard engine now picks `EnvelopeLane` (`CONTROL`/`BULK`) + on every send path for chunked (CLIP-03) transfers — `clipboard/push` with `transfer`, every `clipboard/chunk`, + `clipboard/cancel` from either side, and the matching `ack`/`clipboard/conflict` all travel on channel `0x02` + instead of the control lane `0x01`. `BluetoothPeerSocket` now implements `BulkLaneSocket.sendBulk`, reassembles + channel `0x02` in its own buffer (`BluetoothBulkLaneReassembler`), and backpressures the sender at `BT_BULK_QUEUED` + (2) envelopes queued and not fully written. A duplicate request's kept `ack` (0.5.1 rule 2) now resends on the lane + it was first sent on. Until Wi-Fi Direct (A7.5) ships, every bulk transfer over Bluetooth runs in the background + regardless of size. + ## [0.2.0] — 2026-10-30 ### Added diff --git a/core/protocol/src/main/kotlin/app/handlive/android/core/protocol/bluetooth/BtTransportFrame.kt b/core/protocol/src/main/kotlin/app/handlive/android/core/protocol/bluetooth/BtTransportFrame.kt index 55516c7..050560d 100644 --- a/core/protocol/src/main/kotlin/app/handlive/android/core/protocol/bluetooth/BtTransportFrame.kt +++ b/core/protocol/src/main/kotlin/app/handlive/android/core/protocol/bluetooth/BtTransportFrame.kt @@ -79,6 +79,8 @@ class BtTransportFrame( // Luật "more=1 với long=0 → đóng kênh" chỉ áp dụng cho làn ngắn của control (chan=0x01); bulk // (chan=0x02) không dùng bit `long` nhưng vẫn hợp lệ cắt mảnh nhiều `more=1` (0.4.7). if (chan == BtLinkConstants.CHAN_CONTROL && more && !long) fail("more without long") + // 0.4.7 bulk rule 5: chan 0x02 never sets long=1 (it has its own reassembly buffer, not the long lane's). + if (chan == BtLinkConstants.CHAN_BULK && long) fail("long set on bulk channel") val dlen = ((plaintext[2].toInt() and BYTE_MASK) shl BITS_PER_BYTE) or (plaintext[DLEN_LOW_BYTE_INDEX].toInt() and BYTE_MASK) diff --git a/core/protocol/src/test/kotlin/app/handlive/android/core/protocol/bluetooth/BtTransportFrameTest.kt b/core/protocol/src/test/kotlin/app/handlive/android/core/protocol/bluetooth/BtTransportFrameTest.kt index e84530c..a976f71 100644 --- a/core/protocol/src/test/kotlin/app/handlive/android/core/protocol/bluetooth/BtTransportFrameTest.kt +++ b/core/protocol/src/test/kotlin/app/handlive/android/core/protocol/bluetooth/BtTransportFrameTest.kt @@ -67,6 +67,18 @@ class BtTransportFrameTest { assertThrows(BtLinkException::class.java) { BtTransportFrame.decode(padded) } } + @Test + fun bulkChanWithLongBitSetClosesTheLink() { + // 0.4.7 bulk rule 5: bulk (chan 0x02) never sets long=1; không có vector sẵn cho tổ hợp này nên dựng thủ công + // như bulkFragmentOverMaxClosesTheLink. + val padded = ByteArray(256) + padded[0] = BtLinkConstants.CHAN_BULK.toByte() + padded[1] = 0x02 // flags: long = 1, more = 0 + padded[2] = 0 + padded[3] = 0 + assertThrows(BtLinkException::class.java) { BtTransportFrame.decode(padded) } + } + @Test fun maxDlenMatchesEachLane() { val control = BtLinkConstants.CHAN_CONTROL diff --git a/core/transport/src/main/kotlin/app/handlive/android/core/transport/BulkLaneSocket.kt b/core/transport/src/main/kotlin/app/handlive/android/core/transport/BulkLaneSocket.kt new file mode 100644 index 0000000..c750ad7 --- /dev/null +++ b/core/transport/src/main/kotlin/app/handlive/android/core/transport/BulkLaneSocket.kt @@ -0,0 +1,7 @@ +package app.handlive.android.core.transport + +/** A session socket with a separate bulk lane (the Bluetooth link); other sockets carry both lanes as text. */ +interface BulkLaneSocket { + /** Queues [envelope] on channel `0x02`; suspends while 0.4.7 bulk rule 4 holds the queue full. */ + suspend fun sendBulk(envelope: String) +} diff --git a/core/transport/src/main/kotlin/app/handlive/android/core/transport/EnvelopeLane.kt b/core/transport/src/main/kotlin/app/handlive/android/core/transport/EnvelopeLane.kt new file mode 100644 index 0000000..a65a2fa --- /dev/null +++ b/core/transport/src/main/kotlin/app/handlive/android/core/transport/EnvelopeLane.kt @@ -0,0 +1,7 @@ +package app.handlive.android.core.transport + +/** Which physical lane an envelope travels on (0.4.7); ignored on a LAN or relay session ("single lane"). */ +enum class EnvelopeLane { + CONTROL, + BULK, +} diff --git a/core/transport/src/main/kotlin/app/handlive/android/core/transport/bluetooth/BluetoothBulkLane.kt b/core/transport/src/main/kotlin/app/handlive/android/core/transport/bluetooth/BluetoothBulkLane.kt new file mode 100644 index 0000000..da9ad84 --- /dev/null +++ b/core/transport/src/main/kotlin/app/handlive/android/core/transport/bluetooth/BluetoothBulkLane.kt @@ -0,0 +1,25 @@ +package app.handlive.android.core.transport.bluetooth + +import app.handlive.android.core.protocol.bluetooth.BtTransportFrame +import kotlinx.coroutines.sync.Semaphore + +/** + * Trạng thái làn bulk (`chan` = 0x02, 0.4.7) của một `BluetoothPeerSocket`: cổng hàng đợi `BT_BULK_QUEUED` (bulk + * rule 4) cho chiều gửi và đệm ráp mảnh riêng (bulk rule 5-6) cho chiều nhận. Việc cắt mảnh và xếp sau làn điều + * khiển vẫn là của [BluetoothLinkScheduler]; lớp này không đụng dây. + */ +internal class BluetoothBulkLane { + private val reassembler = BluetoothBulkLaneReassembler() + + /** Đủ `BT_BULK_QUEUED` envelope bulk chưa ghi xong thì [awaitQueueSlot] chờ tới khi một envelope xả hết. */ + private val queueGate = Semaphore(BluetoothLinkConstants.BULK_QUEUED) + + /** Giữ một chỗ trong hàng đợi bulk trước khi xếp một envelope; trả lại bằng [onEnvelopeWritten]. */ + suspend fun awaitQueueSlot() = queueGate.acquire() + + /** Mảnh cuối của một envelope bulk đã lên dây (`BluetoothLinkScheduler.onDrained` trả `true`). */ + fun onEnvelopeWritten() = queueGate.release() + + /** Một mảnh kênh 0x02; trả envelope hoàn chỉnh khi đó là mảnh cuối, vượt `BT_MESSAGE_MAX` thì ném lỗi liên kết. */ + fun accept(frame: BtTransportFrame): ByteArray? = reassembler.accept(frame) +} diff --git a/core/transport/src/main/kotlin/app/handlive/android/core/transport/bluetooth/BluetoothBulkLaneReassembler.kt b/core/transport/src/main/kotlin/app/handlive/android/core/transport/bluetooth/BluetoothBulkLaneReassembler.kt new file mode 100644 index 0000000..3893d1c --- /dev/null +++ b/core/transport/src/main/kotlin/app/handlive/android/core/transport/bluetooth/BluetoothBulkLaneReassembler.kt @@ -0,0 +1,30 @@ +package app.handlive.android.core.transport.bluetooth + +import app.handlive.android.core.protocol.bluetooth.BtLinkConstants +import app.handlive.android.core.protocol.bluetooth.BtLinkException +import app.handlive.android.core.protocol.bluetooth.BtTransportFrame +import java.io.ByteArrayOutputStream + +/** + * Ghép lại làn bulk của kênh `0x02` (0.4.7 bulk rule 5): một đệm riêng, khác đệm làn dài của kênh điều khiển + * `0x01` — một tin một lúc mỗi hướng (bulk rule 3). Đóng liên kết khi tổng byte tích luỹ vượt `BT_MESSAGE_MAX`, + * kiểm ở MỌI mảnh như `BluetoothLongLaneReassembler`. Bulk không dùng bit `long`; `BtTransportFrame.decode` đã + * đóng liên kết khi nó bằng 1 trên kênh này (bulk rule 5), nên [accept] chỉ còn ghép theo `more`. + */ +class BluetoothBulkLaneReassembler( + private val messageMax: Int = BtLinkConstants.MESSAGE_MAX, +) { + private val buffer = ByteArrayOutputStream() + + /** Trả tin hoàn chỉnh khi [frame] là mảnh cuối (`more` = false), `null` nếu còn chờ mảnh tiếp. */ + fun accept(frame: BtTransportFrame): ByteArray? { + require(frame.chan == BtLinkConstants.CHAN_BULK) { "not a bulk-lane frame" } + val total = buffer.size() + frame.data.size + if (total > messageMax) throw BtLinkException("bulk message exceeds BT_MESSAGE_MAX") + buffer.write(frame.data) + if (frame.more) return null + val result = buffer.toByteArray() + buffer.reset() + return result + } +} diff --git a/core/transport/src/main/kotlin/app/handlive/android/core/transport/bluetooth/BluetoothLinkConstants.kt b/core/transport/src/main/kotlin/app/handlive/android/core/transport/bluetooth/BluetoothLinkConstants.kt index b2928f3..cee060d 100644 --- a/core/transport/src/main/kotlin/app/handlive/android/core/transport/bluetooth/BluetoothLinkConstants.kt +++ b/core/transport/src/main/kotlin/app/handlive/android/core/transport/bluetooth/BluetoothLinkConstants.kt @@ -20,6 +20,9 @@ object BluetoothLinkConstants { /** Số bản ghi làn dài + bulk cùng lúc đã gửi mà chưa xả (0.4.5 "Priority"). */ const val BULK_INFLIGHT = 2 + /** `BT_BULK_QUEUED` (0.4.7 bulk rule 4): số envelope bulk tối đa xếp hàng mà chưa xả xong hết; `sendBulk` chờ. */ + const val BULK_QUEUED = 2 + /** `BT_ENABLE_PROMPT_INTERVAL` (SET-01 API 10): tối đa một lần xin bật Bluetooth tự động trong khoảng này. */ val ENABLE_PROMPT_INTERVAL: Duration = 24.hours } diff --git a/core/transport/src/main/kotlin/app/handlive/android/core/transport/bluetooth/BluetoothLinkScheduler.kt b/core/transport/src/main/kotlin/app/handlive/android/core/transport/bluetooth/BluetoothLinkScheduler.kt index 1ddd977..985e5d3 100644 --- a/core/transport/src/main/kotlin/app/handlive/android/core/transport/bluetooth/BluetoothLinkScheduler.kt +++ b/core/transport/src/main/kotlin/app/handlive/android/core/transport/bluetooth/BluetoothLinkScheduler.kt @@ -29,6 +29,9 @@ class BluetoothLinkScheduler( private var longBulkInflight = 0 private val typesPendingLong = mutableSetOf() + /** Song song với thứ tự [dispatch]: `true` khi mảnh đó là mảnh cuối của một envelope bulk, cho [onDrained]. */ + private val inflightCompletesBulkEnvelope = ArrayDeque() + /** Phong bì JSON (0.5.1) trên kênh điều khiển `0x01`; tự chọn làn ngắn hay dài theo 0.4.5 "Control channel". */ fun enqueueControlEnvelope( type: String, @@ -52,23 +55,33 @@ class BluetoothLinkScheduler( fun hasPending(): Boolean = short.isNotEmpty() || long.isNotEmpty() || bulk.isNotEmpty() - /** Mảnh plaintext kế tiếp theo ưu tiên 0.4.5 "Priority", hay `null` nếu không còn gì để gửi ngay. */ + /** + * Mảnh plaintext kế tiếp theo ưu tiên 0.4.5 "Priority", hay `null` nếu không còn gì để gửi ngay. Bên gọi viết + * mảnh này lên dây rồi gọi [onDrained] đúng một lần cho nó (vòng ghi của `BluetoothPeerSocket` làm vậy, tuần + * tự) — [inflightCompletesBulkEnvelope] phải nhận một mục cho MỌI mảnh trả về ở đây, không riêng dài/bulk, để + * khớp thứ tự với [onDrained]. + */ fun poll(): ByteArray? = - short.removeFirstOrNull() - ?: if (longBulkInflight >= bulkInflight) { - null - } else { - long.removeFirstOrNull()?.let { dispatch(it) } - ?: bulk.removeFirstOrNull()?.let { dispatch(it) } - } + short.removeFirstOrNull()?.also { inflightCompletesBulkEnvelope.addLast(false) } + ?: nextLongOrBulk()?.let { dispatch(it) } + + /** Làn dài trước làn bulk, và chỉ khi còn chỗ trong [bulkInflight]; hết chỗ thì `null` dù hàng đợi còn mảnh. */ + private fun nextLongOrBulk(): QueuedFrame? = + if (longBulkInflight >= bulkInflight) null else long.removeFirstOrNull() ?: bulk.removeFirstOrNull() - /** Gọi khi một bản ghi dài/bulk đã [poll] trước đó thật sự xả khỏi bộ đệm ghi (mở lại một chỗ inflight). */ - fun onDrained() { + /** + * Gọi khi một bản ghi dài/bulk đã [poll] trước đó thật sự xả khỏi bộ đệm ghi (mở lại một chỗ inflight). + * Trả `true` khi bản ghi đó là mảnh cuối của một envelope bulk — envelope ấy giờ đã "fully written" (0.4.7 + * bulk rule 4), bên gọi (chủ của [enqueueBulk]) có thể mở thêm một chỗ hàng đợi `BT_BULK_QUEUED`. + */ + fun onDrained(): Boolean { if (longBulkInflight > 0) longBulkInflight-- + return inflightCompletesBulkEnvelope.removeFirstOrNull() == true } private fun dispatch(frame: QueuedFrame): ByteArray { longBulkInflight++ + inflightCompletesBulkEnvelope.addLast(frame.isLastOfMessage && frame.type == null) if (frame.isLastOfMessage) frame.type?.let { typesPendingLong -= it } return frame.bytes } diff --git a/core/transport/src/main/kotlin/app/handlive/android/core/transport/bluetooth/BluetoothLinkWritePump.kt b/core/transport/src/main/kotlin/app/handlive/android/core/transport/bluetooth/BluetoothLinkWritePump.kt new file mode 100644 index 0000000..798ed59 --- /dev/null +++ b/core/transport/src/main/kotlin/app/handlive/android/core/transport/bluetooth/BluetoothLinkWritePump.kt @@ -0,0 +1,69 @@ +package app.handlive.android.core.transport.bluetooth + +import app.handlive.android.core.protocol.envelope.EnvelopeCodec +import io.ktor.websocket.Frame +import io.ktor.websocket.readText +import kotlinx.coroutines.channels.ReceiveChannel + +/** + * Vòng ghi của một `BluetoothPeerSocket`: lấy frame từ [outgoing], xếp vào [BluetoothLinkScheduler] rồi viết từng + * mảnh lên dây qua [writePlaintext] (bên gọi niêm phong và ghi bản ghi). Sau MỖI mảnh, vòng này rút ngay các frame + * đã chờ sẵn trong [outgoing] trước lần `poll()` kế, để envelope điều khiển ngắn gửi giữa lúc bulk đang chạy chen + * vào ở ranh giới mảnh kế tiếp, không phải đợi cả envelope bulk xả xong (0.4.5 "Priority", 0.4.7). + */ +internal class BluetoothLinkWritePump( + private val outgoing: ReceiveChannel, + private val writePlaintext: suspend (ByteArray) -> Unit, + private val onBulkEnvelopeWritten: () -> Unit, +) { + private val scheduler = BluetoothLinkScheduler() + + /** Đã gặp `Frame.Close`: không nhận thêm frame nào, chỉ xả nốt phần đã xếp rồi dừng. */ + private var closeRequested = false + + /** Chạy tới khi [outgoing] đóng hoặc gặp `Frame.Close`; mọi frame xếp trước `Frame.Close` đều lên dây. */ + suspend fun run() { + for (frame in outgoing) { + accept(frame) + drain() + if (closeRequested) return + } + } + + private fun accept(frame: Frame) { + when (frame) { + // Phong bì JSON của làn điều khiển: scheduler tự chọn làn ngắn hay dài theo `type` và cỡ. + is Frame.Text -> { + val json = frame.readText() + scheduler.enqueueControlEnvelope(EnvelopeCodec.decode(json).type, json.toByteArray(Charsets.UTF_8)) + } + + is Frame.Binary -> { + scheduler.enqueueBulk(frame.data) + } + + is Frame.Close -> { + closeRequested = true + } + + else -> {} + } + } + + private suspend fun drain() { + var plaintext = scheduler.poll() + while (plaintext != null) { + writePlaintext(plaintext) + if (scheduler.onDrained()) onBulkEnvelopeWritten() + acceptPending() + plaintext = scheduler.poll() + } + } + + /** Ranh giới mảnh: nhận mọi frame đang chờ mà không treo, dừng ở `Frame.Close`. */ + private fun acceptPending() { + while (!closeRequested) { + accept(outgoing.tryReceive().getOrNull() ?: return) + } + } +} diff --git a/core/transport/src/main/kotlin/app/handlive/android/core/transport/bluetooth/BluetoothPeerSocket.kt b/core/transport/src/main/kotlin/app/handlive/android/core/transport/bluetooth/BluetoothPeerSocket.kt index 5da84cb..8f3b7c0 100644 --- a/core/transport/src/main/kotlin/app/handlive/android/core/transport/bluetooth/BluetoothPeerSocket.kt +++ b/core/transport/src/main/kotlin/app/handlive/android/core/transport/bluetooth/BluetoothPeerSocket.kt @@ -4,11 +4,10 @@ import app.handlive.android.core.protocol.bluetooth.BtLinkConstants import app.handlive.android.core.protocol.bluetooth.BtLinkException import app.handlive.android.core.protocol.bluetooth.BtTransportFrame import app.handlive.android.core.protocol.envelope.Envelope -import app.handlive.android.core.protocol.envelope.EnvelopeCodec +import app.handlive.android.core.transport.BulkLaneSocket import io.ktor.websocket.Frame import io.ktor.websocket.WebSocketExtension import io.ktor.websocket.WebSocketSession -import io.ktor.websocket.readText import kotlinx.coroutines.CancellationException import kotlinx.coroutines.CompletableDeferred import kotlinx.coroutines.CoroutineScope @@ -27,9 +26,9 @@ import kotlin.coroutines.CoroutineContext * Một phiên `/v1/ctl` qua liên kết Bluetooth đã qua bắt tay bảo mật liên kết (0.4.5 bước 4-5, #191): như * `RelayPeerSocket` của relay, `ControlServer.serveBluetoothPeer` nhận socket ảo này mà không đụng tới dây L2CAP * bên dưới — E2E `session/hello` (0.6.3) chạy y nguyên bên trong (0.4.5 bước 6). [sendCipher]/[receiveCipher] là - * hai nửa của `Split()`; việc cắt mảnh/ưu tiên là [BluetoothLinkScheduler], ghép lại làn dài là - * [BluetoothLongLaneReassembler] (cả hai của #191). Kênh bulk (`chan` = 0x02) bỏ qua ở đây — nội dung của nó - * thuộc 0.4.7 (A7.4), ngoài phạm vi thẻ A7.1b. [onFirstRecordOpened]/[onFirstRecordRejected] báo bước 5 cho + * hai nửa của `Split()`; việc cắt mảnh/ưu tiên là [BluetoothLinkScheduler] (vòng ghi: [BluetoothLinkWritePump]), + * ghép lại làn dài là [BluetoothLongLaneReassembler], làn bulk (`chan` = 0x02, 0.4.7) là [BluetoothBulkLane] và + * [sendBulk] (cả ba của #191/A7.4). [onFirstRecordOpened]/[onFirstRecordRejected] báo bước 5 cho * `BluetoothLinkAcceptor` (admission, `BT_PREAUTH_TIMEOUT`). */ internal data class BluetoothPeerSocketCallbacks( @@ -44,14 +43,21 @@ internal class BluetoothPeerSocket( private val receiveCipher: BtTransportCipher, parent: CoroutineScope, private val callbacks: BluetoothPeerSocketCallbacks = BluetoothPeerSocketCallbacks(), -) : WebSocketSession { +) : WebSocketSession, + BulkLaneSocket { override val coroutineContext: CoroutineContext = parent.coroutineContext + SupervisorJob(parent.coroutineContext.job) private val incomingFrames = Channel(Channel.UNLIMITED) private val outgoingFrames = Channel(Channel.UNLIMITED) - private val scheduler = BluetoothLinkScheduler() private val reassembler = BluetoothLongLaneReassembler() + private val bulkLane = BluetoothBulkLane() + private val writePump = + BluetoothLinkWritePump( + outgoing = outgoingFrames, + writePlaintext = { io.writeRecord(sendCipher.seal(it)) }, + onBulkEnvelopeWritten = bulkLane::onEnvelopeWritten, + ) private var anyRecordOpened = false /** Vòng ghi đã dừng: mọi frame xếp trước một `Frame.Close` đã lên dây (hoặc dây đã đứt). */ @@ -72,10 +78,15 @@ internal class BluetoothPeerSocket( private suspend fun readLoop() { try { while (true) { - val body = io.readRecord() - val plaintext = openRecord(body) - val frame = BtTransportFrame.decode(plaintext) - if (frame.chan == BtLinkConstants.CHAN_CONTROL) deliverControlFrame(frame) + val frame = BtTransportFrame.decode(openRecord(io.readRecord())) + // Làn bulk có đệm riêng (0.4.7 bulk rule 5-6); tin đủ của cả hai kênh lên cùng pipeline `/v1/ctl`. + val message = + when { + frame.chan != BtLinkConstants.CHAN_CONTROL -> bulkLane.accept(frame) + frame.long -> reassembler.accept(frame) + else -> frame.data + } + message?.let { incomingFrames.trySend(Frame.Text(it.toString(Charsets.UTF_8))) } } } catch (e: CancellationException) { throw e @@ -102,21 +113,15 @@ internal class BluetoothPeerSocket( return plaintext } - private fun deliverControlFrame(frame: BtTransportFrame) { - val message = if (frame.long) reassembler.accept(frame) else frame.data - message?.let { incomingFrames.trySend(Frame.Text(it.toString(Charsets.UTF_8))) } + /** Làn bulk (0.4.7): [BulkLaneSocket.sendBulk] gọi trực tiếp, không qua [outgoingFrames] như làn điều khiển. */ + override suspend fun sendBulk(envelope: String) { + bulkLane.awaitQueueSlot() + outgoingFrames.send(Frame.Binary(true, envelope.toByteArray(Charsets.UTF_8))) } private suspend fun writeLoop() { try { - for (frame in outgoingFrames) { - when (frame) { - is Frame.Text -> enqueueEnvelope(frame.readText()) - is Frame.Close -> break - else -> Unit - } - drainScheduler() - } + writePump.run() } catch (e: CancellationException) { throw e } catch (_: BtLinkException) { @@ -135,20 +140,6 @@ internal class BluetoothPeerSocket( withTimeoutOrNull(timeoutMillis) { writesFinished.await() } } - private fun enqueueEnvelope(json: String) { - val type = EnvelopeCodec.decode(json).type - scheduler.enqueueControlEnvelope(type, json.toByteArray(Charsets.UTF_8)) - } - - private suspend fun drainScheduler() { - var plaintext = scheduler.poll() - while (plaintext != null) { - io.writeRecord(sendCipher.seal(plaintext)) - scheduler.onDrained() - plaintext = scheduler.poll() - } - } - /** Dây L2CAP đứt hoặc phiên đóng: đóng cả hai hàng đợi và dây vật lý bên dưới. */ fun end() { incomingFrames.close() diff --git a/core/transport/src/main/kotlin/app/handlive/android/core/transport/server/ControlConnectionHandler.kt b/core/transport/src/main/kotlin/app/handlive/android/core/transport/server/ControlConnectionHandler.kt index 11914ea..3b009cd 100644 --- a/core/transport/src/main/kotlin/app/handlive/android/core/transport/server/ControlConnectionHandler.kt +++ b/core/transport/src/main/kotlin/app/handlive/android/core/transport/server/ControlConnectionHandler.kt @@ -259,7 +259,7 @@ internal class ControlConnectionHandler( // 0.5.1 rule 2: the earlier ack again if one was sent; never processed twice. opened.replayed -> { - opened.earlierAck?.let { session.channel.resendAck(it) } + opened.earlierAck?.let { session.channel.resendAck(it, opened.earlierLane) } null } diff --git a/core/transport/src/main/kotlin/app/handlive/android/core/transport/server/ControlSession.kt b/core/transport/src/main/kotlin/app/handlive/android/core/transport/server/ControlSession.kt index 41bf184..0902bbb 100644 --- a/core/transport/src/main/kotlin/app/handlive/android/core/transport/server/ControlSession.kt +++ b/core/transport/src/main/kotlin/app/handlive/android/core/transport/server/ControlSession.kt @@ -8,6 +8,7 @@ import app.handlive.android.core.protocol.envelope.MessageType import app.handlive.android.core.protocol.envelope.PlaintextCodec import app.handlive.android.core.protocol.session.SessionByeData import app.handlive.android.core.protocol.session.SessionOp +import app.handlive.android.core.transport.EnvelopeLane import app.handlive.android.core.transport.WsCloseCode import app.handlive.android.core.transport.capability.Feature import app.handlive.android.core.transport.session.SessionCipher @@ -92,13 +93,19 @@ class ControlSession internal constructor( /** * Gửi một envelope ứng dụng đã mã hóa; trả `id` của envelope. A caller-chosen UUIDv7 [id] lets a request's `ack` - * waiter exist before the envelope leaves. + * waiter exist before the envelope leaves. [lane] (0.4.7): ignored on a LAN or relay session (single lane). */ suspend fun send( type: MessageType, plaintext: ByteArray, id: String? = null, - ): String = if (id == null) channel.send(type.wire, plaintext) else channel.send(type.wire, plaintext, id) + lane: EnvelopeLane = EnvelopeLane.CONTROL, + ): String = + if (id == null) { + channel.send(type.wire, plaintext, lane = lane) + } else { + channel.send(type.wire, plaintext, id, lane) + } /** Gửi `capability/update` (ảnh chụp đầy đủ) khi cấu hình Android đổi, và tính lại tính năng hiệu lực. */ suspend fun sendCapabilityUpdate() { diff --git a/core/transport/src/main/kotlin/app/handlive/android/core/transport/server/EncryptedEnvelopeChannel.kt b/core/transport/src/main/kotlin/app/handlive/android/core/transport/server/EncryptedEnvelopeChannel.kt index f44a5b4..b7257ac 100644 --- a/core/transport/src/main/kotlin/app/handlive/android/core/transport/server/EncryptedEnvelopeChannel.kt +++ b/core/transport/src/main/kotlin/app/handlive/android/core/transport/server/EncryptedEnvelopeChannel.kt @@ -11,6 +11,8 @@ import app.handlive.android.core.protocol.envelope.PlaintextCodec import app.handlive.android.core.protocol.id.UuidV7Generator import app.handlive.android.core.protocol.session.SessionOp import app.handlive.android.core.protocol.session.SessionRekeyData +import app.handlive.android.core.transport.BulkLaneSocket +import app.handlive.android.core.transport.EnvelopeLane import app.handlive.android.core.transport.TransportConstants import app.handlive.android.core.transport.session.RekeyRequestResult import app.handlive.android.core.transport.session.SessionCipher @@ -43,16 +45,18 @@ internal class EncryptedEnvelopeChannel( /** * Mã hóa và gửi; nếu đến ngưỡng rekey và chưa có rekey đang chờ thì gửi `session/rekey` trước. [id] lets a - * caller that waits for an `ack` register its waiter before the envelope leaves. + * caller that waits for an `ack` register its waiter before the envelope leaves. [lane] (0.4.7): bên gọi + * (mô-đun tính năng) chọn, không phải transport tự đoán; bị bỏ qua khi [socket] không phải [BulkLaneSocket]. */ suspend fun send( type: String, plaintext: ByteArray, id: String = ids.next(), + lane: EnvelopeLane = EnvelopeLane.CONTROL, ): String = lock.withLock { startRekeyIfDueLocked() - id.also { writeLocked(type, plaintext, it) } + id.also { writeLocked(type, plaintext, it, lane) } } suspend fun startRekeyIfDue() = lock.withLock { startRekeyIfDueLocked() } @@ -101,18 +105,33 @@ internal class EncryptedEnvelopeChannel( } } - /** 0.5.1 rule 2: a duplicate gets the earlier `ack` again, sealed anew under the current keys. */ - suspend fun resendAck(ack: ByteArray) = lock.withLock { writeLocked(MessageType.ACK.wire, ack, ids.next()) } + /** + * 0.5.1 rule 2: a duplicate gets the earlier `ack` again, sealed anew under the current keys, on [lane] — the + * lane it was first sent on (0.4.7 bulk rule 2), so it cannot overtake a bulk envelope still queued. + */ + suspend fun resendAck( + ack: ByteArray, + lane: EnvelopeLane = EnvelopeLane.CONTROL, + ) = lock.withLock { writeLocked(MessageType.ACK.wire, ack, ids.next(), lane) } private suspend fun writeLocked( type: String, plaintext: ByteArray, id: String, + lane: EnvelopeLane = EnvelopeLane.CONTROL, ) { if (type == MessageType.ACK.wire) { - runCatching { PlaintextCodec.decodeAck(plaintext).re }.getOrNull()?.let { cipher.recordAck(it, plaintext) } + runCatching { PlaintextCodec.decodeAck(plaintext).re }.getOrNull()?.let { + cipher.recordAck(it, plaintext, lane) + } } val envelope = cipher.seal(EnvelopeHeader(type, id, clock()), plaintext) - socket.send(Frame.Text(EnvelopeCodec.encode(envelope))) + val encoded = EnvelopeCodec.encode(envelope) + val bulkSocket = socket as? BulkLaneSocket + if (lane == EnvelopeLane.BULK && bulkSocket != null) { + bulkSocket.sendBulk(encoded) + } else { + socket.send(Frame.Text(encoded)) + } } } diff --git a/core/transport/src/main/kotlin/app/handlive/android/core/transport/session/ReplayWindow.kt b/core/transport/src/main/kotlin/app/handlive/android/core/transport/session/ReplayWindow.kt index 1dcb6ac..8276498 100644 --- a/core/transport/src/main/kotlin/app/handlive/android/core/transport/session/ReplayWindow.kt +++ b/core/transport/src/main/kotlin/app/handlive/android/core/transport/session/ReplayWindow.kt @@ -1,31 +1,40 @@ package app.handlive.android.core.transport.session +import app.handlive.android.core.transport.EnvelopeLane import app.handlive.android.core.transport.TransportConstants import java.util.UUID /** * The receiving side's `DEDUP_WINDOW` (0.5.1 rule 2, 0.10): every `id` accepted in the current key epoch and, while * the previous key is still accepted, the previous epoch's, each with the `ack` this side sent for it so a duplicate - * gets that `ack` again for the whole epoch. A direction that reaches [maxIds] ids (the rekey did not complete) - * overflows; the session is then closed 4410. Stored `ack` bodies are also bounded by [maxAckBytes], oldest dropped - * first — a duplicate whose `ack` was dropped is still never processed again. Not thread-safe (the channel's lock). + * gets that `ack` again for the whole epoch, on the lane it was first sent on (0.4.7 bulk rule 2). A direction that + * reaches [maxIds] ids (the rekey did not complete) overflows; the session is then closed 4410. Stored `ack` bodies + * are also bounded by [maxAckBytes], oldest dropped first — a duplicate whose `ack` was dropped is still never + * processed again. Not thread-safe (the channel's lock). */ internal class ReplayWindow( private val maxIds: Int = TransportConstants.MAX_TRACKED_IDS, private val maxAckBytes: Long = TransportConstants.MAX_TRACKED_ACK_BYTES, ) { + /** The `ack` kept for an accepted `id`, and the lane it went out on. */ + private class StoredAck( + val ack: ByteArray, + val lane: EnvelopeLane, + ) + /** What [record] found: a repeat and its earlier `ack` if kept, or a direction over its cap. */ class Seen( val replayed: Boolean, val earlierAck: ByteArray?, + val earlierLane: EnvelopeLane, val overflow: Boolean, ) - private var current = LinkedHashMap() - private var previous: LinkedHashMap? = null + private var current = LinkedHashMap() + private var previous: LinkedHashMap? = null /** Stored acks, oldest first, each with the epoch that holds it (an id may be in both epochs). */ - private val ackOrder = ArrayDeque, UUID>>() + private val ackOrder = ArrayDeque, UUID>>() private var ackBytes = 0L val size: Int get() = current.size + (previous?.size ?: 0) @@ -43,16 +52,17 @@ internal class ReplayWindow( val earlier = current[id] ?: previous?.get(id) val epoch = if (withPrevious) checkNotNull(previous) else current if (!epoch.containsKey(id)) epoch[id] = null - return Seen(known, earlier, current.size >= maxIds) + return Seen(known, earlier?.ack, earlier?.lane ?: EnvelopeLane.CONTROL, current.size >= maxIds) } - /** The `ack` this side sent for request [re]; kept only for an `id` it accepted. */ + /** The `ack` this side sent for request [re], on [lane]; kept only for an `id` it accepted. */ fun recordAck( re: UUID, ack: ByteArray, + lane: EnvelopeLane = EnvelopeLane.CONTROL, ) { val epoch = listOfNotNull(current, previous).firstOrNull { it.containsKey(re) && it[re] == null } ?: return - epoch[re] = ack + epoch[re] = StoredAck(ack, lane) ackOrder.addLast(epoch to re) ackBytes += ack.size while (ackBytes > maxAckBytes && ackOrder.isNotEmpty()) { @@ -77,10 +87,10 @@ internal class ReplayWindow( /** Forgets the ack of [id] in [holder]; always `true`, so it can drive a `removeAll`. */ private fun release( - holder: MutableMap, + holder: MutableMap, id: UUID, ): Boolean { - holder[id]?.let { ackBytes -= it.size } + holder[id]?.let { ackBytes -= it.ack.size } holder[id] = null return true } diff --git a/core/transport/src/main/kotlin/app/handlive/android/core/transport/session/SessionCipher.kt b/core/transport/src/main/kotlin/app/handlive/android/core/transport/session/SessionCipher.kt index f0306fa..a2f920d 100644 --- a/core/transport/src/main/kotlin/app/handlive/android/core/transport/session/SessionCipher.kt +++ b/core/transport/src/main/kotlin/app/handlive/android/core/transport/session/SessionCipher.kt @@ -6,6 +6,7 @@ import app.handlive.android.core.crypto.message.EnvelopeCipher import app.handlive.android.core.protocol.ProtocolException import app.handlive.android.core.protocol.envelope.Envelope import app.handlive.android.core.protocol.envelope.EnvelopeHeader +import app.handlive.android.core.transport.EnvelopeLane import app.handlive.android.core.transport.TransportConstants import java.util.UUID import kotlin.time.Duration @@ -48,12 +49,14 @@ class SessionCipher( /** * A decrypted envelope. [replayed]: its `id` was already accepted in this epoch or the previous one; [earlierAck] - * is the `ack` plaintext sent for it, if kept. [overflow]: the direction reached `MAX_TRACKED_IDS` (close 4410). + * is the `ack` plaintext sent for it, if kept, on [earlierLane] (0.4.7 bulk rule 2: resent on the lane it was + * first sent on). [overflow]: the direction reached `MAX_TRACKED_IDS` (close 4410). */ class Opened( val plaintext: ByteArray, val replayed: Boolean, val earlierAck: ByteArray? = null, + val earlierLane: EnvelopeLane = EnvelopeLane.CONTROL, val overflow: Boolean = false, ) @@ -78,15 +81,16 @@ class SessionCipher( fun accept(envelope: Envelope): Opened { val (plaintext, withPrevious) = decrypt(envelope) val seen = window.record(UUID.fromString(envelope.id), withPrevious) - return Opened(plaintext, seen.replayed, seen.earlierAck, seen.overflow) + return Opened(plaintext, seen.replayed, seen.earlierAck, seen.earlierLane, seen.overflow) } - /** The `ack` this side sends for request [re], kept to answer a duplicate for the whole epoch. */ + /** The `ack` this side sends for request [re], kept to answer a duplicate for the whole epoch, on [lane]. */ fun recordAck( re: String, ack: ByteArray, + lane: EnvelopeLane = EnvelopeLane.CONTROL, ) { - runCatching { UUID.fromString(re) }.getOrNull()?.let { window.recordAck(it, ack) } + runCatching { UUID.fromString(re) }.getOrNull()?.let { window.recordAck(it, ack, lane) } } private fun decrypt(envelope: Envelope): Pair { diff --git a/core/transport/src/test/kotlin/app/handlive/android/core/transport/bluetooth/BluetoothBulkLaneReassemblerTest.kt b/core/transport/src/test/kotlin/app/handlive/android/core/transport/bluetooth/BluetoothBulkLaneReassemblerTest.kt new file mode 100644 index 0000000..7d83308 --- /dev/null +++ b/core/transport/src/test/kotlin/app/handlive/android/core/transport/bluetooth/BluetoothBulkLaneReassemblerTest.kt @@ -0,0 +1,45 @@ +package app.handlive.android.core.transport.bluetooth + +import app.handlive.android.core.protocol.bluetooth.BtLinkException +import app.handlive.android.core.protocol.bluetooth.BtTransportFrame +import org.junit.Assert.assertArrayEquals +import org.junit.Assert.assertNull +import org.junit.Assert.assertThrows +import org.junit.Test + +/** 0.4.7 bulk rule 5: ghép làn bulk của kênh `0x02`, đệm riêng của làn dài; đóng liên kết khi vượt `BT_MESSAGE_MAX`. */ +class BluetoothBulkLaneReassemblerTest { + @Test + fun reassemblesTwoBulkFragmentsOfOneMessage() { + val reassembler = BluetoothBulkLaneReassembler() + val first = BtTransportFrame(chan = 0x02, more = true, long = false, data = byteArrayOf(1, 2, 3)) + val second = BtTransportFrame(chan = 0x02, more = false, long = false, data = byteArrayOf(4, 5)) + assertNull(reassembler.accept(first)) + val message = reassembler.accept(second) + assertArrayEquals(byteArrayOf(1, 2, 3, 4, 5), message) + } + + @Test + fun aSingleFragmentMessageCompletesImmediately() { + val reassembler = BluetoothBulkLaneReassembler() + val frame = BtTransportFrame(chan = 0x02, more = false, long = false, data = byteArrayOf(9)) + assertArrayEquals(byteArrayOf(9), reassembler.accept(frame)) + } + + @Test + fun messageOverMaxClosesTheLinkEvenMidStream() { + val reassembler = BluetoothBulkLaneReassembler(messageMax = 10) + val first = BtTransportFrame(chan = 0x02, more = true, long = false, data = ByteArray(8)) + assertNull(reassembler.accept(first)) + val second = BtTransportFrame(chan = 0x02, more = false, long = false, data = ByteArray(3)) + assertThrows(BtLinkException::class.java) { reassembler.accept(second) } + } + + @Test + fun aSecondMessageStartsWithAFreshBuffer() { + val reassembler = BluetoothBulkLaneReassembler() + reassembler.accept(BtTransportFrame(chan = 0x02, more = false, long = false, data = byteArrayOf(1))) + val frame = BtTransportFrame(chan = 0x02, more = false, long = false, data = byteArrayOf(2)) + assertArrayEquals(byteArrayOf(2), reassembler.accept(frame)) + } +} diff --git a/core/transport/src/test/kotlin/app/handlive/android/core/transport/bluetooth/BluetoothLinkSchedulerTest.kt b/core/transport/src/test/kotlin/app/handlive/android/core/transport/bluetooth/BluetoothLinkSchedulerTest.kt index ea1b37b..257bba1 100644 --- a/core/transport/src/test/kotlin/app/handlive/android/core/transport/bluetooth/BluetoothLinkSchedulerTest.kt +++ b/core/transport/src/test/kotlin/app/handlive/android/core/transport/bluetooth/BluetoothLinkSchedulerTest.kt @@ -3,6 +3,7 @@ package app.handlive.android.core.transport.bluetooth import app.handlive.android.core.protocol.bluetooth.BtLinkConstants import app.handlive.android.core.protocol.bluetooth.BtTransportFrame import org.junit.Assert.assertEquals +import org.junit.Assert.assertFalse import org.junit.Assert.assertNotNull import org.junit.Assert.assertNull import org.junit.Assert.assertTrue @@ -98,6 +99,34 @@ class BluetoothLinkSchedulerTest { assertTrue(frame.long) } + @Test + fun onDrainedReportsTheLastFragmentOfABulkEnvelopeOnlyAtTheVeryLastFragment() { + // 0.4.7 bulk rule 4: the signal `BluetoothPeerSocket` uses to free a `BT_BULK_QUEUED` slot. + val scheduler = BluetoothLinkScheduler(bulkInflight = 100) + scheduler.enqueueBulk(ByteArray(BtLinkConstants.BULK_FRAGMENT_MAX * 2 + 1)) // 3 fragments + requireNotNull(scheduler.poll()) + assertFalse(scheduler.onDrained()) + requireNotNull(scheduler.poll()) + assertFalse(scheduler.onDrained()) + requireNotNull(scheduler.poll()) + assertTrue(scheduler.onDrained()) + } + + @Test + fun onDrainedOfAShortFrameNeverSignalsABulkCompletion() { + // poll() must push an entry for EVERY frame it returns (short included) to stay in lockstep with + // onDrained(), since the real write loop calls onDrained() once per poll() — not just for long/bulk. + val scheduler = BluetoothLinkScheduler() + scheduler.enqueueBulk(ByteArray(10)) // 1 fragment, last of its envelope + scheduler.enqueueControlEnvelope("ping", ByteArray(5)) + val first = BtTransportFrame.decode(scheduler.poll()!!) + assertEquals(false, first.long) + assertFalse(scheduler.onDrained()) + val second = BtTransportFrame.decode(scheduler.poll()!!) + assertEquals(BtLinkConstants.CHAN_BULK, second.chan) + assertTrue(scheduler.onDrained()) + } + @Test fun hasPendingReflectsAllThreeQueues() { val scheduler = BluetoothLinkScheduler() diff --git a/core/transport/src/test/kotlin/app/handlive/android/core/transport/bluetooth/BluetoothPeerSocketPriorityTest.kt b/core/transport/src/test/kotlin/app/handlive/android/core/transport/bluetooth/BluetoothPeerSocketPriorityTest.kt new file mode 100644 index 0000000..688f367 --- /dev/null +++ b/core/transport/src/test/kotlin/app/handlive/android/core/transport/bluetooth/BluetoothPeerSocketPriorityTest.kt @@ -0,0 +1,88 @@ +package app.handlive.android.core.transport.bluetooth + +import app.handlive.android.core.crypto.primitives.SecureRandomBytes +import app.handlive.android.core.protocol.encoding.Base64Codecs +import app.handlive.android.core.protocol.envelope.Envelope +import app.handlive.android.core.protocol.envelope.EnvelopeCodec +import app.handlive.android.core.protocol.id.UuidV7Generator +import app.handlive.android.core.transport.bluetooth.testing.InMemoryBluetoothLinkIo +import io.ktor.websocket.Frame +import io.ktor.websocket.readText +import kotlinx.coroutines.CompletableDeferred +import kotlinx.coroutines.CoroutineScope +import kotlinx.coroutines.Dispatchers +import kotlinx.coroutines.SupervisorJob +import kotlinx.coroutines.cancel +import kotlinx.coroutines.launch +import kotlinx.coroutines.runBlocking +import kotlinx.coroutines.withTimeout +import org.junit.After +import org.junit.Assert.assertEquals +import org.junit.Test +import kotlin.time.Duration.Companion.seconds + +/** Envelope điều khiển ngắn gửi giữa lúc bulk đang chạy phải tới trước phần bulk còn lại (0.4.5 Priority). */ +class BluetoothPeerSocketPriorityTest { + private val scope = CoroutineScope(SupervisorJob() + Dispatchers.Default) + private val ids = UuidV7Generator() + + @After + fun tearDown() = scope.cancel() + + private fun envelope( + type: String, + bytes: Int, + ) = EnvelopeCodec.encode( + Envelope( + Envelope.VERSION, + type, + ids.next(), + System.currentTimeMillis(), + Base64Codecs.encodeB64(ByteArray(bytes)), + ), + ) + + @Test + fun shortControlSentMidBulkArrivesBeforeTheBulkEnds() = + runBlocking { + val c1 = SecureRandomBytes.next(32) + val c2 = SecureRandomBytes.next(32) + val (rawA, ioB) = InMemoryBluetoothLinkIo.pair() + val ioA = HoldSecondRecordLinkIo(rawA) + val a = BluetoothPeerSocket(ioA, BtTransportCipher(c1), BtTransportCipher(c2), scope) + val b = BluetoothPeerSocket(ioB, BtTransportCipher(c2), BtTransportCipher(c1), scope) + val bulk = envelope("clipboard", 180_000) + val ping = envelope("ping", 8) + + launch { a.sendBulk(bulk) } + // Mảnh bulk đầu đã lên dây, mảnh thứ hai đang bị giữ: bulk chắc chắn đang giữa chừng. + withTimeout(WAIT) { ioA.firstRecordWritten.await() } + a.outgoing.send(Frame.Text(ping)) + ioA.release.complete(Unit) + + val first = withTimeout(WAIT) { (b.incoming.receive() as Frame.Text).readText() } + val second = withTimeout(WAIT) { (b.incoming.receive() as Frame.Text).readText() } + assertEquals("control phải tới trước bulk", ping, first) + assertEquals("bulk vẫn tới trọn vẹn sau control", bulk, second) + } + + /** Ống giả giữ bản ghi thứ hai tới khi test mở [release], để control được gửi đúng lúc bulk còn dở. */ + private class HoldSecondRecordLinkIo( + private val inner: BluetoothLinkIo, + ) : BluetoothLinkIo by inner { + val firstRecordWritten = CompletableDeferred() + val release = CompletableDeferred() + private var writes = 0 + + override suspend fun writeRecord(body: ByteArray) { + writes++ + if (writes == 2) release.await() + inner.writeRecord(body) + if (writes == 1) firstRecordWritten.complete(Unit) + } + } + + private companion object { + val WAIT = 10.seconds + } +} diff --git a/core/transport/src/test/kotlin/app/handlive/android/core/transport/bluetooth/BluetoothPeerSocketTest.kt b/core/transport/src/test/kotlin/app/handlive/android/core/transport/bluetooth/BluetoothPeerSocketTest.kt index e63e706..5d2e523 100644 --- a/core/transport/src/test/kotlin/app/handlive/android/core/transport/bluetooth/BluetoothPeerSocketTest.kt +++ b/core/transport/src/test/kotlin/app/handlive/android/core/transport/bluetooth/BluetoothPeerSocketTest.kt @@ -13,6 +13,7 @@ import kotlinx.coroutines.CoroutineScope import kotlinx.coroutines.Dispatchers import kotlinx.coroutines.SupervisorJob import kotlinx.coroutines.cancel +import kotlinx.coroutines.launch import kotlinx.coroutines.runBlocking import kotlinx.coroutines.withTimeout import org.junit.After @@ -91,6 +92,66 @@ class BluetoothPeerSocketTest { assertEquals(sent, received.readText()) } + @Test + fun aBulkEnvelopeRoundTripsOnChannel0x02() = + runBlocking { + // 0.4.7: BulkLaneSocket.sendBulk, separate from the control lane's Frame.Text(outgoing.send). + val (a, b) = link() + val sent = envelope("clipboard", payloadBytes = 8) + a.sendBulk(sent) + val received = withTimeout(WAIT) { b.incoming.receive() as Frame.Text } + assertEquals(sent, received.readText()) + } + + @Test + fun aBulkEnvelopeOverFragmentMaxReassemblesWholeOnTheOtherSide() = + runBlocking { + val (a, b) = link() + val sent = envelope("clipboard", payloadBytes = BtLinkConstants.BULK_FRAGMENT_MAX * 3) + a.sendBulk(sent) + val received = withTimeout(WAIT) { b.incoming.receive() as Frame.Text } + assertEquals(sent, received.readText()) + } + + @Test + fun sendBulkNeverHangsPastBtBulkQueuedConsecutiveEnvelopes() = + runBlocking { + // 0.4.7 bulk rule 4 (BT_BULK_QUEUED = 2): the gate that backpressures sendBulk must always release + // once an earlier envelope fully writes — more envelopes than the queue depth must still all land, + // not deadlock. BluetoothLinkSchedulerTest covers the exact drain-signal accounting in isolation. + val (a, b) = link() + val sent = (1..5).map { envelope("clipboard", payloadBytes = 8) } + val sender = launch { sent.forEach { a.sendBulk(it) } } + val received = (1..5).map { withTimeout(WAIT) { (b.incoming.receive() as Frame.Text).readText() } } + withTimeout(WAIT) { sender.join() } + assertEquals(sent, received) + } + + @Test + fun aBulkFragmentWithLongBitSetClosesTheLink() = + runBlocking { + // BtTransportFrame.decode already enforces 0.4.7 bulk rule 5 (long=1 on chan 0x02); here end-to-end: + // a malformed record written straight to the raw link (no real sender ever builds this) must still + // make the receiving BluetoothPeerSocket close instead of delivering garbage. + val c1 = SecureRandomBytes.next(KEY_SIZE) + val c2 = SecureRandomBytes.next(KEY_SIZE) + val (ioA, ioB) = InMemoryBluetoothLinkIo.pair() + val b = BluetoothPeerSocket(ioB, BtTransportCipher(c2), BtTransportCipher(c1), scope) + val plaintext = ByteArray(256) + plaintext[0] = BtLinkConstants.CHAN_BULK.toByte() + plaintext[1] = 0x02 // flags: long = 1 + ioA.writeRecord(BtTransportCipher(c1).seal(plaintext)) + withTimeout(WAIT) { + var closed = false + try { + b.incoming.receive() + } catch (_: Exception) { + closed = true + } + assertTrue(closed || b.incoming.isClosedForReceive) + } + } + @Test fun closingOneEndClosesTheOtherSidesIncoming() = runBlocking { diff --git a/core/transport/src/test/kotlin/app/handlive/android/core/transport/session/ReplayWindowTest.kt b/core/transport/src/test/kotlin/app/handlive/android/core/transport/session/ReplayWindowTest.kt index d601a6f..b73a9b7 100644 --- a/core/transport/src/test/kotlin/app/handlive/android/core/transport/session/ReplayWindowTest.kt +++ b/core/transport/src/test/kotlin/app/handlive/android/core/transport/session/ReplayWindowTest.kt @@ -1,5 +1,6 @@ package app.handlive.android.core.transport.session +import app.handlive.android.core.transport.EnvelopeLane import org.junit.Assert.assertArrayEquals import org.junit.Assert.assertEquals import org.junit.Assert.assertNull @@ -48,6 +49,21 @@ class ReplayWindowTest { assertEquals(1, window.storedAcks) } + @Test + fun theEarlierAckIsResentOnTheLaneItWasFirstSentOn() { + // 0.4.7 bulk rule 2: an ack kept to answer duplicates is resent on the lane it was first sent on. + val id = accepted() + window.recordAck(id, ByteArray(4), EnvelopeLane.BULK) + val seen = window.record(id, false) + assertArrayEquals(ByteArray(4), seen.earlierAck) + assertEquals(EnvelopeLane.BULK, seen.earlierLane) + } + + @Test + fun anUnseenIdHasNoEarlierLaneEither() { + assertEquals(EnvelopeLane.CONTROL, window.record(UUID.randomUUID(), false).earlierLane) + } + @Test fun anIdInBothEpochsKeepsTheCurrentAckWhenThePreviousGoes() { val id = accepted() diff --git a/core/transport/src/test/kotlin/app/handlive/android/core/transport/session/SessionCipherReplayTest.kt b/core/transport/src/test/kotlin/app/handlive/android/core/transport/session/SessionCipherReplayTest.kt index e2f28db..f7ead49 100644 --- a/core/transport/src/test/kotlin/app/handlive/android/core/transport/session/SessionCipherReplayTest.kt +++ b/core/transport/src/test/kotlin/app/handlive/android/core/transport/session/SessionCipherReplayTest.kt @@ -7,8 +7,10 @@ import app.handlive.android.core.protocol.ProtocolException import app.handlive.android.core.protocol.envelope.Envelope import app.handlive.android.core.protocol.envelope.EnvelopeHeader import app.handlive.android.core.protocol.id.UuidV7Generator +import app.handlive.android.core.transport.EnvelopeLane import app.handlive.android.core.transport.TransportConstants import org.junit.Assert.assertArrayEquals +import org.junit.Assert.assertEquals import org.junit.Assert.assertFalse import org.junit.Assert.assertNull import org.junit.Assert.assertThrows @@ -93,6 +95,18 @@ class SessionCipherReplayTest { assertNull(server.accept(unknown).earlierAck) } + @Test + fun aDuplicateGetsItsEarlierAckOnTheLaneItWasFirstSentOn() { + // 0.4.7 bulk rule 2. + val request = seal() + server.accept(request) + val ack = "{\"re\":\"${request.id}\",\"ok\":true}".toByteArray() + server.recordAck(request.id, ack, EnvelopeLane.BULK) + val replay = server.accept(request) + assertArrayEquals(ack, replay.earlierAck) + assertEquals(EnvelopeLane.BULK, replay.earlierLane) + } + @Test fun aDirectionThatReaches20000IdsWithoutARekeyOverflows() { repeat(TransportConstants.MAX_TRACKED_IDS - 1) { assertFalse(server.accept(seal()).overflow) } diff --git a/feature/call/src/test/kotlin/app/handlive/android/feature/call/system/CallPermissionNotifierTest.kt b/feature/call/src/test/kotlin/app/handlive/android/feature/call/system/CallPermissionNotifierTest.kt index 65dbee8..293c07b 100644 --- a/feature/call/src/test/kotlin/app/handlive/android/feature/call/system/CallPermissionNotifierTest.kt +++ b/feature/call/src/test/kotlin/app/handlive/android/feature/call/system/CallPermissionNotifierTest.kt @@ -39,7 +39,7 @@ class CallPermissionNotifierTest { PeerSession.Channel.LAN, MutableStateFlow(emptySet()), MutableStateFlow(null), - { _, _, _ -> }, + { _, _, _, _ -> }, { now }, ) diff --git a/feature/call/src/test/kotlin/app/handlive/android/feature/call/testing/CallFakes.kt b/feature/call/src/test/kotlin/app/handlive/android/feature/call/testing/CallFakes.kt index 51b3088..1d986e9 100644 --- a/feature/call/src/test/kotlin/app/handlive/android/feature/call/testing/CallFakes.kt +++ b/feature/call/src/test/kotlin/app/handlive/android/feature/call/testing/CallFakes.kt @@ -223,7 +223,7 @@ class FakeClient( PeerSession.Channel.LAN, effective, MutableStateFlow(null), - { type, plaintext, _ -> sent += Sent(type, plaintext) }, + { type, plaintext, _, _ -> sent += Sent(type, plaintext) }, clock, ) diff --git a/feature/clipboard/src/main/kotlin/app/handlive/android/feature/clipboard/ClipNotices.kt b/feature/clipboard/src/main/kotlin/app/handlive/android/feature/clipboard/ClipNotices.kt index 931b2b4..a4f8a75 100644 --- a/feature/clipboard/src/main/kotlin/app/handlive/android/feature/clipboard/ClipNotices.kt +++ b/feature/clipboard/src/main/kotlin/app/handlive/android/feature/clipboard/ClipNotices.kt @@ -44,13 +44,18 @@ sealed interface ClipMessage { ) : ClipMessage } -/** Progress of an image over 1 MiB (CLIP-03 fields 2–4); `percent = null` removes it. */ +/** + * Tiến độ của một ảnh trên 1 MiB, hoặc từ `BULK_DIRECT_MIN` trở lên khi qua Bluetooth (CLIP-03 field 2–4); + * `percent = null` thì gỡ nó đi. [overBluetooth]: transfer chạy trên làn bulk Bluetooth (0.4.7), nên dòng chữ + * CLIP-03 field 10 hiện dưới tiến độ. + */ data class TransferProgress( val transferId: String, val pairId: String, val deviceName: String, val sending: Boolean, val percent: Int?, + val overBluetooth: Boolean = false, ) /** diff --git a/feature/clipboard/src/main/kotlin/app/handlive/android/feature/clipboard/component/ClipNotifier.kt b/feature/clipboard/src/main/kotlin/app/handlive/android/feature/clipboard/component/ClipNotifier.kt index 7d701cd..eb6325c 100644 --- a/feature/clipboard/src/main/kotlin/app/handlive/android/feature/clipboard/component/ClipNotifier.kt +++ b/feature/clipboard/src/main/kotlin/app/handlive/android/feature/clipboard/component/ClipNotifier.kt @@ -129,7 +129,7 @@ class ClipNotifier( NumberFormat.getPercentInstance().format(percent / FULL.toDouble()), ) val cancel = ClipActionReceiver.cancelTransfer(context, progress.pairId, progress.transferId, progress.sending) - val notification = + val builder = builder() .setSmallIcon(DesignR.drawable.ic_symbol_image) .setContentText(text) @@ -137,8 +137,22 @@ class ClipNotifier( .setOngoing(true) .setOnlyAlertOnce(true) .addAction(0, context.getString(R.string.common_cancel), cancel) - .build() - post(ID_PROGRESS, progress.transferId, notification) + // CLIP-03 field 10: qua liên kết Bluetooth (0.4.7) thêm dòng báo có thể mất vài phút, ngay dưới dòng phần trăm. + if (progress.overBluetooth) { + val noteId = + if (progress.sending) { + R.string.clipboard_bulk_bluetooth_sending + } else { + R.string.clipboard_bulk_bluetooth_receiving + } + val bigText = + buildString { + appendLine(text) + append(context.getString(noteId)) + } + builder.setStyle(NotificationCompat.BigTextStyle().bigText(bigText)) + } + post(ID_PROGRESS, progress.transferId, builder.build()) } private fun builder() = diff --git a/feature/clipboard/src/main/kotlin/app/handlive/android/feature/clipboard/engine/ClipLimits.kt b/feature/clipboard/src/main/kotlin/app/handlive/android/feature/clipboard/engine/ClipLimits.kt index 0c900fa..0cc27d3 100644 --- a/feature/clipboard/src/main/kotlin/app/handlive/android/feature/clipboard/engine/ClipLimits.kt +++ b/feature/clipboard/src/main/kotlin/app/handlive/android/feature/clipboard/engine/ClipLimits.kt @@ -17,6 +17,9 @@ object ClipLimits { /** Images above 1 MiB show progress (CLIP-03 field 2). */ const val PROGRESS_MIN_BYTES = 1_048_576L + /** `BULK_DIRECT_MIN` (0.4.7): qua Bluetooth, ảnh từ 128 KiB trở lên hiện tiến độ (CLIP-03 field 2, 10). */ + const val BULK_DIRECT_MIN_BYTES = 128 * 1024L + /** `CLIP_STALE_AFTER`: replay window (QC7), sensitive content and "Send Again" lifetime (QC3, CLIP-01 field 13). */ const val STALE_AFTER_MILLIS = 120_000L diff --git a/feature/clipboard/src/main/kotlin/app/handlive/android/feature/clipboard/module/ClipReceiver.kt b/feature/clipboard/src/main/kotlin/app/handlive/android/feature/clipboard/module/ClipReceiver.kt index 57c5dd3..fb9d84c 100644 --- a/feature/clipboard/src/main/kotlin/app/handlive/android/feature/clipboard/module/ClipReceiver.kt +++ b/feature/clipboard/src/main/kotlin/app/handlive/android/feature/clipboard/module/ClipReceiver.kt @@ -10,6 +10,7 @@ import app.handlive.android.core.protocol.clipboard.ClipboardPushData import app.handlive.android.core.protocol.clipboard.ClipboardValues import app.handlive.android.core.protocol.clipboard.HtmlClipSanitizer import app.handlive.android.core.protocol.envelope.MessageType +import app.handlive.android.core.transport.EnvelopeLane import app.handlive.android.core.transport.capability.Feature import app.handlive.android.feature.clipboard.engine.Acceptance import app.handlive.android.feature.clipboard.engine.Clip @@ -38,7 +39,10 @@ class ClipReceiver( private val outgoing: OutgoingTransfers, ) { private val state = context.state - val incoming = IncomingTransfers(context, ::reply, ::onVerified) + + /** CLIP-03 là chủ của làn bulk (0.4.7 bulk rule 1): IncomingTransfers chỉ xử lý push có `transfer`. */ + private val replyBulk: suspend (PeerSession, Ack, String) -> Unit = { s, a, c -> reply(s, a, c, EnvelopeLane.BULK) } + val incoming = IncomingTransfers(context, replyBulk, ::onVerified) suspend fun onPush( session: PeerSession, @@ -60,15 +64,17 @@ class ClipReceiver( ("html" to 1).takeIf { push.html != null }, ), ) + val lane = push.replyLane val rejection = PushValidator.check(push, acceptance(session)) when { rejection != null -> { state.ledger.recordRejected(push.clipId) - reply(session, ClipWire.rejected(pushId, push.clipId, rejection), push.clipId) + reply(session, ClipWire.rejected(pushId, push.clipId, rejection), push.clipId, lane) } state.ledger.isDuplicate(push.clipId) -> { - reply(session, ClipWire.ignored(pushId, push.clipId, ClipboardValues.REASON_DUPLICATE), push.clipId) + val ignored = ClipWire.ignored(pushId, push.clipId, ClipboardValues.REASON_DUPLICATE) + reply(session, ignored, push.clipId, lane) } else -> { @@ -76,7 +82,7 @@ class ClipReceiver( if (text == null) { incoming.start(session, pushId, push) } else { - apply(session, pushId, push, ClipContent.Text(text, sanitizedHtml(push.html))) + apply(session, pushId, push, ClipContent.Text(text, sanitizedHtml(push.html)), lane) } } } @@ -168,9 +174,11 @@ class ClipReceiver( Rejection(ErrorCode.BAD_REQUEST, "text is not valid UTF-8"), ), push.clipId, + EnvelopeLane.BULK, ) } else { - apply(done.session, done.pushId, push, content) + // onVerified chỉ chạy sau một transfer CLIP-03 hoàn tất: ack của nó luôn đi làn bulk (bulk rule 1). + apply(done.session, done.pushId, push, content, EnvelopeLane.BULK) } } @@ -180,8 +188,9 @@ class ClipReceiver( pushId: String, push: ClipboardPushData, content: ClipContent, + lane: EnvelopeLane, ) { - if (state.loopGuard.alreadyHolds(content.sha256)) return acknowledgeHeld(session, pushId, push, content) + if (state.loopGuard.alreadyHolds(content.sha256)) return acknowledgeHeld(session, pushId, push, content, lane) val now = context.clock.wall() val decision = ConflictPolicy.decide( @@ -217,6 +226,7 @@ class ClipReceiver( session, ClipWire.rejected(pushId, push.clipId, Rejection(ErrorCode.INTERNAL, "clipboard write failed")), push.clipId, + lane, ) } else { BenchLog.event(BenchEvent.CLIP_APPLIED, "clip" to push.clipId) @@ -233,7 +243,7 @@ class ClipReceiver( ) state.remember(clip) state.ledger.recordFinal(push.clipId) - reply(session, ClipWire.applied(pushId, push.clipId), push.clipId) + reply(session, ClipWire.applied(pushId, push.clipId), push.clipId, lane) sender.distribute(clip, manual = false) } } @@ -247,10 +257,11 @@ class ClipReceiver( pushId: String, push: ClipboardPushData, content: ClipContent, + lane: EnvelopeLane, ) { (content as? ClipContent.FileBacked)?.file?.delete() state.ledger.recordFinal(push.clipId) - reply(session, ClipWire.applied(pushId, push.clipId), push.clipId) + reply(session, ClipWire.applied(pushId, push.clipId), push.clipId, lane) } /** QC8: `ignored`/`conflict`; case (a) also tells the sender which device kept its content (API 6 logic 1). */ @@ -261,23 +272,26 @@ class ClipReceiver( content: ClipContent, decision: ConflictDecision, ) { + val lane = push.replyLane (content as? ClipContent.FileBacked)?.file?.delete() state.ledger.recordFinal(push.clipId) - reply(session, ClipWire.ignored(pushId, push.clipId, ClipboardValues.REASON_CONFLICT), push.clipId) + reply(session, ClipWire.ignored(pushId, push.clipId, ClipboardValues.REASON_CONFLICT), push.clipId, lane) if (decision == ConflictDecision.KEEP_LOCAL_AND_NOTIFY) { + // 0.4.7 bulk rule 1: this conflict follows the ack above, same lane, so it cannot overtake it. val local = context.network.local() val conflict = ClipboardConflictData(push.clipId, push.originDeviceId, local.deviceId, local.name) - runCatching { session.send(MessageType.CLIPBOARD, ClipWire.conflict(conflict)) } + runCatching { session.send(MessageType.CLIPBOARD, ClipWire.conflict(conflict), lane) } } } - /** Sends the `ack` of a push (and keeps it for duplicates); a closed session is not an error here. */ + /** Sends the `ack` of a push on [lane] (and keeps it for duplicates); a closed session is not an error here. */ private suspend fun reply( session: PeerSession, ack: Ack, clipId: String?, + lane: EnvelopeLane = EnvelopeLane.CONTROL, ) { - runCatching { session.sendAck(ack) } + runCatching { session.sendAck(ack, lane) } BenchLog.event( BenchEvent.ACK_SENT, "clip" to (clipId ?: "-"), @@ -291,6 +305,10 @@ class ClipReceiver( } } +/** 0.4.7 bulk rule 1: the ack of a push with `transfer` (and chunks/cancel) is bulk; an inline push's ack isn't. */ +private val ClipboardPushData.replyLane: EnvelopeLane + get() = if (transfer != null) EnvelopeLane.BULK else EnvelopeLane.CONTROL + /** * The receiver never trusts the sender's sanitizing: `html` is sanitized again before any write. Escaping can grow * the output past [ClipLimits.MAX_HTML_BYTES]; then the text is written alone. diff --git a/feature/clipboard/src/main/kotlin/app/handlive/android/feature/clipboard/module/IncomingTransfers.kt b/feature/clipboard/src/main/kotlin/app/handlive/android/feature/clipboard/module/IncomingTransfers.kt index d1d892b..769ac1f 100644 --- a/feature/clipboard/src/main/kotlin/app/handlive/android/feature/clipboard/module/IncomingTransfers.kt +++ b/feature/clipboard/src/main/kotlin/app/handlive/android/feature/clipboard/module/IncomingTransfers.kt @@ -6,6 +6,7 @@ import app.handlive.android.core.protocol.clipboard.ClipboardChunkPlaintext import app.handlive.android.core.protocol.clipboard.ClipboardPushData import app.handlive.android.core.protocol.clipboard.ClipboardValues import app.handlive.android.core.protocol.envelope.MessageType +import app.handlive.android.core.transport.EnvelopeLane import app.handlive.android.feature.clipboard.ClipMessage import app.handlive.android.feature.clipboard.TransferProgress import app.handlive.android.feature.clipboard.engine.ClipLimits @@ -148,7 +149,11 @@ class IncomingTransfers( ) { finish(incoming) val transferId = incoming.transfer.transfer.transferId - cancel?.let { runCatching { incoming.session.send(MessageType.CLIPBOARD, ClipWire.cancel(transferId, it)) } } + cancel?.let { + runCatching { + incoming.session.send(MessageType.CLIPBOARD, ClipWire.cancel(transferId, it), EnvelopeLane.BULK) + } + } // A cancelled clip may come again (a timeout is "not received", QC7), so it is not a final outcome. context.state.ledger.recordRejected(incoming.push.clipId) reply(incoming.session, ClipWire.ignored(incoming.pushId, incoming.push.clipId, reason), incoming.push.clipId) @@ -180,8 +185,15 @@ class IncomingTransfers( push: ClipboardPushData, ): TransferProgress? { val transfer = checkNotNull(push.transfer) - return if (push.kind == ClipboardValues.KIND_IMAGE && transfer.size > ClipLimits.PROGRESS_MIN_BYTES) { - TransferProgress(transfer.transferId, session.pairId, session.peerName, sending = false, percent = null) + return if (showsProgress(session, push.kind, transfer.size)) { + TransferProgress( + transfer.transferId, + session.pairId, + session.peerName, + sending = false, + percent = null, + overBluetooth = session.channel == PeerSession.Channel.BLUETOOTH, + ) } else { null } diff --git a/feature/clipboard/src/main/kotlin/app/handlive/android/feature/clipboard/module/OutgoingTransfers.kt b/feature/clipboard/src/main/kotlin/app/handlive/android/feature/clipboard/module/OutgoingTransfers.kt index 6f29c48..8486dbc 100644 --- a/feature/clipboard/src/main/kotlin/app/handlive/android/feature/clipboard/module/OutgoingTransfers.kt +++ b/feature/clipboard/src/main/kotlin/app/handlive/android/feature/clipboard/module/OutgoingTransfers.kt @@ -5,6 +5,7 @@ import app.handlive.android.core.protocol.clipboard.ClipboardChunkPlaintext import app.handlive.android.core.protocol.clipboard.ClipboardValues import app.handlive.android.core.protocol.envelope.MessageType import app.handlive.android.core.protocol.id.UuidV7Generator +import app.handlive.android.core.transport.EnvelopeLane import app.handlive.android.feature.clipboard.ClipNotices import app.handlive.android.feature.clipboard.TransferProgress import app.handlive.android.feature.clipboard.engine.ChunkSource @@ -46,12 +47,17 @@ class OutgoingTransfers( val progress = progressOf(session, clip, transferId) return try { ChunkSource.open(clip.content).use { source -> - session.request(MessageType.CLIPBOARD, ClipWire.chunkedPush(clip, transferId)) { + session.request( + MessageType.CLIPBOARD, + ClipWire.chunkedPush(clip, transferId), + lane = EnvelopeLane.BULK, + ) { onPushSent() for (index in 0 until source.chunkCount) { session.send( MessageType.CLIPBOARD, ClipboardChunkPlaintext(transferId, index, source.next()).encode(), + EnvelopeLane.BULK, ) progress?.let { notices.progress(it.copy(percent = percentOf(index + 1, source.chunkCount))) } } @@ -71,7 +77,10 @@ class OutgoingTransfers( val entry = active[session.pairId]?.takeIf { it.transferId == transferId } ?: return stop(session.pairId, entry) entry.clip.deliveredTo += session.pairId - runCatching { session.send(MessageType.CLIPBOARD, ClipWire.cancel(transferId, ClipboardValues.CANCEL_USER)) } + runCatching { + val cancel = ClipWire.cancel(transferId, ClipboardValues.CANCEL_USER) + session.send(MessageType.CLIPBOARD, cancel, EnvelopeLane.BULK) + } } /** The receiver cancelled (`user` or `timeout`, API 5 rule 2); `user` means the clip is not replayed to it. */ @@ -95,7 +104,8 @@ class OutgoingTransfers( val previous = active[session.pairId] ?: return stop(session.pairId, previous) runCatching { - session.send(MessageType.CLIPBOARD, ClipWire.cancel(previous.transferId, ClipboardValues.CANCEL_SUPERSEDED)) + val cancel = ClipWire.cancel(previous.transferId, ClipboardValues.CANCEL_SUPERSEDED) + session.send(MessageType.CLIPBOARD, cancel, EnvelopeLane.BULK) } } @@ -112,13 +122,36 @@ class OutgoingTransfers( clip: Clip, transferId: String, ): TransferProgress? = - if (clip.kind == ClipboardValues.KIND_IMAGE && clip.content.size > ClipLimits.PROGRESS_MIN_BYTES) { - TransferProgress(transferId, session.pairId, session.peerName, sending = true, percent = null) + if (showsProgress(session, clip.kind, clip.content.size)) { + TransferProgress( + transferId, + session.pairId, + session.peerName, + sending = true, + percent = null, + overBluetooth = session.channel == PeerSession.Channel.BLUETOOTH, + ) } else { null } } +/** + * CLIP-03 field 2: ảnh hiện tiến độ khi lớn hơn `PROGRESS_MIN_BYTES`, hoặc từ `BULK_DIRECT_MIN` trở lên khi phiên + * đi qua Bluetooth (0.4.7 "Timeouts and progress": khoảng 10 s ở 12,9 KB/s, để người dùng thấy việc vẫn đang chạy). + */ +internal fun showsProgress( + session: PeerSession, + kind: String, + size: Long, +): Boolean = + kind == ClipboardValues.KIND_IMAGE && + if (session.channel == PeerSession.Channel.BLUETOOTH) { + size >= ClipLimits.BULK_DIRECT_MIN_BYTES + } else { + size > ClipLimits.PROGRESS_MIN_BYTES + } + /** Progress in whole percent (CLIP-03 API 3 rule 6: chunks done / `chunk_count`). */ internal fun percentOf( done: Int, diff --git a/feature/clipboard/src/test/kotlin/app/handlive/android/feature/clipboard/component/ClipNotifierTest.kt b/feature/clipboard/src/test/kotlin/app/handlive/android/feature/clipboard/component/ClipNotifierTest.kt index dbf0457..6bbbfbf 100644 --- a/feature/clipboard/src/test/kotlin/app/handlive/android/feature/clipboard/component/ClipNotifierTest.kt +++ b/feature/clipboard/src/test/kotlin/app/handlive/android/feature/clipboard/component/ClipNotifierTest.kt @@ -28,6 +28,11 @@ class ClipNotifierTest { private fun posted(): List = shadowOf(manager).allNotifications + private fun bluetoothProgress( + transferId: String, + sending: Boolean, + ) = TransferProgress(transferId, "pair-mac", "Lan's Pixel", sending, percent = 10, overBluetooth = true) + @Test fun sensitiveContentIsSilentOnTheClipboardChannelWithSendAnywayForTwoMinutes() { notifier.sensitiveBlocked() @@ -77,6 +82,42 @@ class ClipNotifierTest { assertTrue(posted().isEmpty()) } + @Test + fun bluetoothProgressAddsTheBluetoothNoteUnderThePercent() { + // CLIP-03 field 10: the line under field 2 when the transfer runs over the Bluetooth link (0.4.7). + val sending = bluetoothProgress("t-2", sending = true) + notifier.progress(sending) + val notification = posted().single() + assertEquals("Sending image to Lan's Pixel — 10%", NotificationCompat.getContentText(notification)) + assertEquals( + "Sending image to Lan's Pixel — 10%\nSending over Bluetooth. This can take a few minutes.", + notification.extras.getCharSequence(NotificationCompat.EXTRA_BIG_TEXT).toString(), + ) + notifier.progress(sending.copy(transferId = "t-3", sending = false)) + val bigTexts = posted().map { it.extras.getCharSequence(NotificationCompat.EXTRA_BIG_TEXT).toString() } + assertTrue(bigTexts.any { it.endsWith("Receiving over Bluetooth. This can take a few minutes.") }) + } + + @Test + fun lanProgressHasNoBluetoothNote() { + notifier.progress(TransferProgress("t-4", "pair-mac", "Lan's Pixel", sending = true, percent = 10)) + assertEquals(null, posted().single().extras.getCharSequence(NotificationCompat.EXTRA_BIG_TEXT)) + } + + @Test + @Config(qualifiers = "vi") + fun bluetoothNoteInVietnamese() { + notifier.progress(bluetoothProgress("t-5", sending = true)) + assertTrue( + posted() + .single() + .extras + .getCharSequence(NotificationCompat.EXTRA_BIG_TEXT) + .toString() + .endsWith("Đang gửi qua Bluetooth. Có thể mất vài phút."), + ) + } + @Test fun refusalsAndSkippedSendsNameTheDevice() { notifier.show(ClipMessage.WriteFailedOnDevice("MacBook của Lan")) diff --git a/feature/clipboard/src/test/kotlin/app/handlive/android/feature/clipboard/module/BulkLaneContractTest.kt b/feature/clipboard/src/test/kotlin/app/handlive/android/feature/clipboard/module/BulkLaneContractTest.kt new file mode 100644 index 0000000..bc6b9a9 --- /dev/null +++ b/feature/clipboard/src/test/kotlin/app/handlive/android/feature/clipboard/module/BulkLaneContractTest.kt @@ -0,0 +1,203 @@ +package app.handlive.android.feature.clipboard.module + +import app.handlive.android.core.protocol.clipboard.ClipboardPushData +import app.handlive.android.core.protocol.clipboard.ClipboardTransfer +import app.handlive.android.core.protocol.clipboard.ClipboardValues +import app.handlive.android.core.protocol.encoding.Base64Codecs +import app.handlive.android.core.transport.EnvelopeLane +import app.handlive.android.feature.clipboard.engine.ChunkPlan +import app.handlive.android.feature.clipboard.engine.ClipLimits +import app.handlive.android.feature.clipboard.testing.ClipboardHarness +import kotlinx.coroutines.test.TestScope +import kotlinx.coroutines.test.runTest +import org.junit.Assert.assertEquals +import org.junit.Assert.assertTrue +import org.junit.Rule +import org.junit.Test +import org.junit.rules.TemporaryFolder +import java.security.MessageDigest +import kotlin.random.Random + +/** + * 0.4.7 "Bulk lane on the Bluetooth link": the clipboard engine picks `EnvelopeLane` on every send path (bulk rule + * 2). `ClipboardHarness` runs on `PeerSession.Channel.LAN` where the lane argument is ignored on the wire (bulk + * rule 1), but the engine must still choose the right one so a real Bluetooth session carries it correctly. + */ +class BulkLaneContractTest { + @get:Rule + val folder = TemporaryFolder() + + private fun test(body: suspend TestScope.(ClipboardHarness) -> Unit) = + runTest { + body(ClipboardHarness(this, folder.root)) + } + + private fun imagePush( + h: ClipboardHarness, + bytes: ByteArray, + sha256: ByteArray = MessageDigest.getInstance("SHA-256").digest(bytes), + clipId: String = h.newId(), + ) = ClipboardPushData( + clipId = clipId, + kind = ClipboardValues.KIND_IMAGE, + mime = ClipboardValues.MIME_PNG, + transfer = + ClipboardTransfer( + transferId = h.newId(), + size = bytes.size.toLong(), + sha256 = Base64Codecs.encodeB64u(sha256), + chunkSize = ChunkPlan.CHUNK_SIZE, + chunkCount = ChunkPlan.chunkCount(bytes.size.toLong()), + ), + width = 2880, + height = 1800, + sensitive = false, + originTs = h.wall(), + source = "mac", + originDeviceId = "5b1f8c2e-9a4d-8e6f-a1b2-c3d4e5f60718", + ) + + private suspend fun sendChunks( + h: ClipboardHarness, + push: ClipboardPushData, + bytes: ByteArray, + ) { + val transferId = checkNotNull(push.transfer).transferId + val count = ChunkPlan.chunkCount(bytes.size.toLong()) + (0 until count).forEach { index -> + val from = index * ChunkPlan.CHUNK_SIZE + h.chunk(h.mac, transferId, index, bytes.copyOfRange(from, minOf(bytes.size, from + ChunkPlan.CHUNK_SIZE))) + } + } + + @Test + fun outgoingPushWithTransferAndItsChunksAreBulk() = + test { h -> + // Bulk rule 1: `clipboard/push` with `transfer`, and every `clipboard/chunk`. + h.connect(h.mac) + h.readImage(Random(1).nextBytes(1_500_000)) + val (pushId, _) = h.mac.pushes().single() + assertEquals(EnvelopeLane.BULK, h.mac.laneOf(pushId)) + assertTrue(h.mac.chunkLanes().isNotEmpty()) + assertTrue(h.mac.chunkLanes().all { it == EnvelopeLane.BULK }) + } + + @Test + fun outgoingInlineTextPushIsControl() = + test { h -> + // Bulk rule 1: an inline `clipboard/push` stays on control. + h.connect(h.mac) + h.readText("hello there") + val (pushId, _) = h.mac.pushes().single() + assertEquals(EnvelopeLane.CONTROL, h.mac.laneOf(pushId)) + } + + @Test + fun aSupersededCancelOfARunningTransferIsBulk() = + test { h -> + h.connect(h.mac) + h.readImage(Random(2).nextBytes(100_000)) + h.readText("newer text") + assertEquals(listOf(EnvelopeLane.BULK), h.mac.cancelLanes()) + } + + @Test + fun aUserCancelOfASendingTransferIsBulk() = + test { h -> + h.connect(h.mac) + h.readImage(Random(3).nextBytes(100_000)) + val (_, push) = h.mac.pushes().single() + val transferId = checkNotNull(push.transfer).transferId + h.module.cancelTransfer("pair-mac", transferId, sending = true) + h.run() + assertEquals(listOf(EnvelopeLane.BULK), h.mac.cancelLanes()) + } + + @Test + fun aReceiverTimeoutCancelOfAnIncomingTransferIsBulk() = + test { h -> + h.connect(h.mac) + val bytes = Random(4).nextBytes(200_000) + val push = imagePush(h, bytes) + h.push(h.mac, push) + testScheduler.advanceTimeBy(ClipLimits.TRANSFER_IDLE_MILLIS + 1) + h.run() + assertEquals(listOf(EnvelopeLane.BULK), h.mac.cancelLanes()) + } + + @Test + fun ackOfAPushWithTransferIsBulkWhetherAppliedOrRejected() = + test { h -> + h.connect(h.mac) + val bytes = Random(5).nextBytes(100_000) + val push = imagePush(h, bytes) + h.push(h.mac, push) + sendChunks(h, push, bytes) + assertEquals(listOf(EnvelopeLane.BULK), h.mac.ackLanes()) + + val bad = imagePush(h, bytes, sha256 = ByteArray(32)) + h.push(h.mac, bad) + sendChunks(h, bad, bytes) + assertEquals(listOf(EnvelopeLane.BULK, EnvelopeLane.BULK), h.mac.ackLanes()) + } + + @Test + fun ackOfAnInlinePushIsControl() = + test { h -> + h.connect(h.mac) + h.push(h.mac, h.macText("hi")) + assertEquals(listOf(EnvelopeLane.CONTROL), h.mac.ackLanes()) + } + + @Test + fun anImageOfBulkDirectMinOverBluetoothShowsSendingProgressWithTheBluetoothNote() = + test { h -> + // 0.4.7 "Timeouts and progress": over Bluetooth, progress from `BULK_DIRECT_MIN` (CLIP-03 field 2, 10). + h.connect(h.macOverBluetooth) + h.readImage(Random(6).nextBytes(ClipLimits.BULK_DIRECT_MIN_BYTES.toInt())) + val progress = h.notices.progress.filter { it.sending } + assertTrue(progress.isNotEmpty()) + assertTrue(progress.all { it.overBluetooth }) + assertTrue(progress.any { it.percent != null }) + } + + @Test + fun anImageUnderBulkDirectMinOverBluetoothShowsNoProgress() = + test { h -> + h.connect(h.macOverBluetooth) + h.readImage(Random(7).nextBytes(ClipLimits.BULK_DIRECT_MIN_BYTES.toInt() - 1)) + assertTrue(h.mac.pushes().isEmpty()) + assertEquals(1, h.macOverBluetooth.pushes().size) + assertTrue(h.notices.progress.isEmpty()) + } + + @Test + fun theSameImageOnLanKeepsTheOneMebibyteThreshold() = + test { h -> + // LAN keeps CLIP-03 field 2 as before: images over 1 MiB only. + h.connect(h.mac) + h.readImage(Random(8).nextBytes(ClipLimits.BULK_DIRECT_MIN_BYTES.toInt())) + assertEquals(1, h.mac.pushes().size) + assertTrue(h.notices.progress.isEmpty()) + } + + @Test + fun anIncomingImageOfBulkDirectMinOverBluetoothShowsReceivingProgressWithTheBluetoothNote() = + test { h -> + h.connect(h.macOverBluetooth) + val bytes = Random(9).nextBytes(ClipLimits.BULK_DIRECT_MIN_BYTES.toInt()) + val push = imagePush(h, bytes) + h.push(h.macOverBluetooth, push) + val transferId = checkNotNull(push.transfer).transferId + val count = ChunkPlan.chunkCount(bytes.size.toLong()) + (0 until count).forEach { index -> + val from = index * ChunkPlan.CHUNK_SIZE + val part = bytes.copyOfRange(from, minOf(bytes.size, from + ChunkPlan.CHUNK_SIZE)) + h.chunk(h.macOverBluetooth, transferId, index, part) + } + val progress = h.notices.progress.filter { !it.sending } + assertTrue(progress.all { it.overBluetooth }) + assertEquals(0, progress.first().percent) + assertEquals(null, progress.last().percent) + } +} diff --git a/feature/clipboard/src/test/kotlin/app/handlive/android/feature/clipboard/testing/ClipboardTestKit.kt b/feature/clipboard/src/test/kotlin/app/handlive/android/feature/clipboard/testing/ClipboardTestKit.kt index ca4c1bd..ddeefc4 100644 --- a/feature/clipboard/src/test/kotlin/app/handlive/android/feature/clipboard/testing/ClipboardTestKit.kt +++ b/feature/clipboard/src/test/kotlin/app/handlive/android/feature/clipboard/testing/ClipboardTestKit.kt @@ -18,6 +18,7 @@ import app.handlive.android.core.protocol.clipboard.ClipboardValues import app.handlive.android.core.protocol.envelope.MessageType import app.handlive.android.core.protocol.envelope.PlaintextCodec import app.handlive.android.core.protocol.id.UuidV7Generator +import app.handlive.android.core.transport.EnvelopeLane import app.handlive.android.core.transport.capability.Feature import app.handlive.android.core.transport.server.InboundEnvelope import app.handlive.android.feature.clipboard.ClipMessage @@ -175,11 +176,13 @@ class FakeClient( val deviceId: String, val pairId: String, clock: () -> Long, + channel: PeerSession.Channel = PeerSession.Channel.LAN, ) { class Sent( val type: MessageType, val plaintext: ByteArray, val id: String, + val lane: EnvelopeLane, ) val sent = mutableListOf() @@ -189,12 +192,12 @@ class FakeClient( val session = PeerSession( PeerSession.PeerInfo(pairId, deviceId, name, PeerPlatform.MACOS), - PeerSession.Channel.LAN, + channel, effective, capability, - { type, plaintext, id -> + { type, plaintext, id, lane -> check(connected) { "session closed" } - sent += Sent(type, plaintext, id) + sent += Sent(type, plaintext, id, lane) }, clock, ) @@ -211,6 +214,20 @@ class FakeClient( fun cancels(): List = ops(ClipboardOp.CANCEL, ClipboardCancelData.serializer()).map { it.second } + /** 0.4.7: the lane of the envelope with [id] — `sent` keeps every envelope this client has received. */ + fun laneOf(id: String): EnvelopeLane = sent.first { it.id == id }.lane + + /** Lane of every `clipboard/cancel`, same order as [cancels]. */ + fun cancelLanes(): List = + ops(ClipboardOp.CANCEL, ClipboardCancelData.serializer()).map { laneOf(it.first) } + + /** Lane of every `clipboard/chunk`, same order as [chunks]. */ + fun chunkLanes(): List = + sent.filter { it.type == MessageType.CLIPBOARD && it.plaintext.first() == 0.toByte() }.map { it.lane } + + /** Lane of every `ack`, same order as [acks]. */ + fun ackLanes(): List = sent.filter { it.type == MessageType.ACK }.map { it.lane } + fun conflicts(): List = ops(ClipboardOp.CONFLICT, ClipboardConflictData.serializer()).map { it.second @@ -273,6 +290,9 @@ class ClipboardHarness( val mac = FakeClient("MacBook của Lan", MAC_ID, "pair-mac", wall) val ipad = FakeClient("iPad của Lan", IPAD_ID, "pair-ipad", wall) + /** The same Mac reached over the Bluetooth link (0.4.5); connect it instead of [mac], never both. */ + val macOverBluetooth = FakeClient("MacBook của Lan", MAC_ID, "pair-mac", wall, PeerSession.Channel.BLUETOOTH) + init { module.start() } diff --git a/feature/connection/src/main/kotlin/app/handlive/android/feature/connection/session/PeerSession.kt b/feature/connection/src/main/kotlin/app/handlive/android/feature/connection/session/PeerSession.kt index 7467266..be374f5 100644 --- a/feature/connection/src/main/kotlin/app/handlive/android/feature/connection/session/PeerSession.kt +++ b/feature/connection/src/main/kotlin/app/handlive/android/feature/connection/session/PeerSession.kt @@ -7,6 +7,7 @@ import app.handlive.android.core.protocol.capability.CapabilityData import app.handlive.android.core.protocol.envelope.MessageType import app.handlive.android.core.protocol.envelope.PlaintextCodec import app.handlive.android.core.protocol.id.UuidV7Generator +import app.handlive.android.core.transport.EnvelopeLane import app.handlive.android.core.transport.TransportConstants import app.handlive.android.core.transport.capability.Feature import kotlinx.coroutines.CompletableDeferred @@ -21,6 +22,7 @@ fun interface EnvelopeSender { type: MessageType, plaintext: ByteArray, id: String, + lane: EnvelopeLane, ) } @@ -66,28 +68,30 @@ class PeerSession( fun isEffective(feature: Feature): Boolean = feature in effectiveFeatures.value - /** Sends an event (no `ack`); returns its `id`. */ + /** Sends an event (no `ack`) on [lane] (0.4.7); returns its `id`. */ suspend fun send( type: MessageType, plaintext: ByteArray, - ): String = ids.next().also { sender.send(type, plaintext, it) } + lane: EnvelopeLane = EnvelopeLane.CONTROL, + ): String = ids.next().also { sender.send(type, plaintext, it, lane) } /** * Sends a request and waits for its `ack` (0.5.1 rule 1); throws [AckTimeoutException] after [timeout], counted * from the end of [afterSend] — a chunked clip sends its chunks there and the 10 s start after the last one - * (CLIP-03 API 3 rule 3). [afterSend] gets the request's `id`. + * (CLIP-03 API 3 rule 3). [afterSend] gets the request's `id`. [lane] (0.4.7): the caller picks it. */ suspend fun request( type: MessageType, plaintext: ByteArray, timeout: Duration = TransportConstants.REQUEST_TIMEOUT, + lane: EnvelopeLane = EnvelopeLane.CONTROL, afterSend: suspend (String) -> Unit = {}, ): Ack { val id = ids.next() val waiter = CompletableDeferred() pendingAcks[id] = waiter return try { - sender.send(type, plaintext, id) + sender.send(type, plaintext, id, lane) afterSend(id) withTimeoutOrNull(timeout) { waiter.await() } ?: throw AckTimeoutException() } finally { @@ -95,17 +99,21 @@ class PeerSession( } } - /** Sends the `ack` of request [re] and keeps it for duplicates. */ - suspend fun sendAck(ack: Ack) { + /** Sends the `ack` of request [re] on [lane] and keeps it for duplicates. */ + suspend fun sendAck( + ack: Ack, + lane: EnvelopeLane = EnvelopeLane.CONTROL, + ) { ledger.recordAck(ack) - send(MessageType.ACK, PlaintextCodec.encodeAck(ack)) + send(MessageType.ACK, PlaintextCodec.encodeAck(ack), lane) } suspend fun sendError( re: String, code: ErrorCode, message: String, - ) = sendAck(Ack.failure(re, code, message)) + lane: EnvelopeLane = EnvelopeLane.CONTROL, + ) = sendAck(Ack.failure(re, code, message), lane) /** An `ack` for no waiter (late, after a timeout) is dropped. */ internal fun completeAck(ack: Ack) { diff --git a/feature/connection/src/main/kotlin/app/handlive/android/feature/connection/session/SessionTable.kt b/feature/connection/src/main/kotlin/app/handlive/android/feature/connection/session/SessionTable.kt index 0e58ac5..c93cd4d 100644 --- a/feature/connection/src/main/kotlin/app/handlive/android/feature/connection/session/SessionTable.kt +++ b/feature/connection/src/main/kotlin/app/handlive/android/feature/connection/session/SessionTable.kt @@ -58,7 +58,7 @@ class SessionTable( channel = channel, effectiveFeatures = control.effectiveFeatures, peerCapability = control.peerCapability, - sender = { type, plaintext, id -> control.send(type, plaintext, id) }, + sender = { type, plaintext, id, lane -> control.send(type, plaintext, id, lane) }, clock = clock, ).also { it.ledger = ledgers.getOrPut(control.pairId) { EnvelopeLedger(clock) } } sessionsFlow.update { it + (control.pairId to peer) } diff --git a/feature/connection/src/test/kotlin/app/handlive/android/feature/connection/session/SessionRouterTest.kt b/feature/connection/src/test/kotlin/app/handlive/android/feature/connection/session/SessionRouterTest.kt index b458021..7cedb78 100644 --- a/feature/connection/src/test/kotlin/app/handlive/android/feature/connection/session/SessionRouterTest.kt +++ b/feature/connection/src/test/kotlin/app/handlive/android/feature/connection/session/SessionRouterTest.kt @@ -45,7 +45,7 @@ class SessionRouterTest { channel = PeerSession.Channel.LAN, effectiveFeatures = MutableStateFlow(emptySet()), peerCapability = MutableStateFlow(null), - sender = { type, plaintext, id -> sent += Sent(type, plaintext, id) }, + sender = { type, plaintext, id, _ -> sent += Sent(type, plaintext, id) }, clock = { now }, ) private val router = SessionRouter(clock = { now }) @@ -122,7 +122,7 @@ class SessionRouterTest { channel = PeerSession.Channel.RELAY, effectiveFeatures = MutableStateFlow(emptySet()), peerCapability = MutableStateFlow(null), - sender = { type, plaintext, id -> sent += Sent(type, plaintext, id) }, + sender = { type, plaintext, id, _ -> sent += Sent(type, plaintext, id) }, clock = { now }, ) relayed.ledger = session.ledger diff --git a/feature/pairing/src/test/kotlin/app/handlive/android/feature/pairing/devices/DeviceListModelTest.kt b/feature/pairing/src/test/kotlin/app/handlive/android/feature/pairing/devices/DeviceListModelTest.kt index d86f12c..bdf74c2 100644 --- a/feature/pairing/src/test/kotlin/app/handlive/android/feature/pairing/devices/DeviceListModelTest.kt +++ b/feature/pairing/src/test/kotlin/app/handlive/android/feature/pairing/devices/DeviceListModelTest.kt @@ -159,7 +159,7 @@ class DeviceListModelTest { ), ), ), - sender = { _, _, _ -> }, + sender = { _, _, _, _ -> }, clock = { 0 }, ) } diff --git a/feature/pairing/src/test/kotlin/app/handlive/android/feature/pairing/revoke/UnpairControllerTest.kt b/feature/pairing/src/test/kotlin/app/handlive/android/feature/pairing/revoke/UnpairControllerTest.kt index 69af57b..c0908d5 100644 --- a/feature/pairing/src/test/kotlin/app/handlive/android/feature/pairing/revoke/UnpairControllerTest.kt +++ b/feature/pairing/src/test/kotlin/app/handlive/android/feature/pairing/revoke/UnpairControllerTest.kt @@ -53,7 +53,7 @@ class UnpairControllerTest { channel = channel, effectiveFeatures = MutableStateFlow(emptySet()), peerCapability = MutableStateFlow(null), - sender = { type, plaintext, id -> + sender = { type, plaintext, id, _ -> sent += type to plaintext if (type == MessageType.PAIR && answerRevoke) { // The Mac acknowledges before it cleans up (API 1 rule 1). diff --git a/feature/sms/src/test/kotlin/app/handlive/android/feature/sms/system/SmsSendLimitNotifierTest.kt b/feature/sms/src/test/kotlin/app/handlive/android/feature/sms/system/SmsSendLimitNotifierTest.kt index 7aa849d..872ae97 100644 --- a/feature/sms/src/test/kotlin/app/handlive/android/feature/sms/system/SmsSendLimitNotifierTest.kt +++ b/feature/sms/src/test/kotlin/app/handlive/android/feature/sms/system/SmsSendLimitNotifierTest.kt @@ -29,7 +29,7 @@ class SmsSendLimitNotifierTest { PeerSession.Channel.LAN, MutableStateFlow(emptySet()), MutableStateFlow(null), - { _, _, _ -> }, + { _, _, _, _ -> }, { 0L }, ) diff --git a/feature/sms/src/test/kotlin/app/handlive/android/feature/sms/testing/SmsHarness.kt b/feature/sms/src/test/kotlin/app/handlive/android/feature/sms/testing/SmsHarness.kt index 7782b3a..e011367 100644 --- a/feature/sms/src/test/kotlin/app/handlive/android/feature/sms/testing/SmsHarness.kt +++ b/feature/sms/src/test/kotlin/app/handlive/android/feature/sms/testing/SmsHarness.kt @@ -139,7 +139,7 @@ class FakeClient( PeerSession.Channel.LAN, effective, capability, - { type, plaintext, _ -> sent += Sent(type, plaintext) }, + { type, plaintext, _, _ -> sent += Sent(type, plaintext) }, clock, )