From 43b5b95bbd072f4e206ba58279c1e8765f7b9b99 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?H=E1=BB=93=20Xu=C3=A2n=20D=C5=A9ng?= Date: Fri, 30 Oct 2026 12:38:56 +0700 Subject: [PATCH 1/9] feat(transport): bulk lane over Bluetooth for CLIP-03 (0.4.7, A7.4) The clipboard engine now picks EnvelopeLane (CONTROL/BULK) on every send path for chunked 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 0x01 when the session is over Bluetooth (ignored on LAN/relay, single lane). BluetoothPeerSocket implements BulkLaneSocket.sendBulk, gated by a Semaphore(BT_BULK_QUEUED=2) that releases once the scheduler reports a bulk envelope's last fragment fully written; a new BluetoothBulkLaneReassembler reassembles channel 0x02 in its own buffer, separate from the long control lane's. BtTransportFrame.decode closes the link when long=1 is set on channel 0x02 (bulk rule 5). A duplicate request's kept ack (0.5.1 rule 2) now resends on the lane it was first sent on, tracked through ReplayWindow/SessionCipher. Until Wi-Fi Direct (A7.5) ships, every Bluetooth transfer runs on the bulk lane regardless of size. --- CHANGELOG.md | 11 ++ .../protocol/bluetooth/BtTransportFrame.kt | 2 + .../bluetooth/BtTransportFrameTest.kt | 12 ++ .../android/core/transport/BulkLaneSocket.kt | 7 + .../android/core/transport/EnvelopeLane.kt | 7 + .../bluetooth/BluetoothBulkLaneReassembler.kt | 30 ++++ .../bluetooth/BluetoothLinkConstants.kt | 3 + .../bluetooth/BluetoothLinkScheduler.kt | 38 +++-- .../bluetooth/BluetoothPeerSocket.kt | 28 +++- .../server/ControlConnectionHandler.kt | 2 +- .../core/transport/server/ControlSession.kt | 11 +- .../server/EncryptedEnvelopeChannel.kt | 31 +++- .../core/transport/session/ReplayWindow.kt | 32 ++-- .../core/transport/session/SessionCipher.kt | 12 +- .../BluetoothBulkLaneReassemblerTest.kt | 45 ++++++ .../bluetooth/BluetoothLinkSchedulerTest.kt | 29 ++++ .../bluetooth/BluetoothPeerSocketTest.kt | 61 +++++++ .../transport/session/ReplayWindowTest.kt | 16 ++ .../session/SessionCipherReplayTest.kt | 14 ++ .../call/system/CallPermissionNotifierTest.kt | 2 +- .../android/feature/call/testing/CallFakes.kt | 2 +- .../feature/clipboard/module/ClipReceiver.kt | 41 +++-- .../clipboard/module/IncomingTransfers.kt | 7 +- .../clipboard/module/OutgoingTransfers.kt | 16 +- .../clipboard/module/BulkLaneContractTest.kt | 150 ++++++++++++++++++ .../clipboard/testing/ClipboardTestKit.kt | 20 ++- .../feature/connection/session/PeerSession.kt | 24 ++- .../connection/session/SessionTable.kt | 2 +- .../connection/session/SessionRouterTest.kt | 4 +- .../pairing/devices/DeviceListModelTest.kt | 2 +- .../pairing/revoke/UnpairControllerTest.kt | 2 +- .../sms/system/SmsSendLimitNotifierTest.kt | 2 +- 32 files changed, 590 insertions(+), 75 deletions(-) create mode 100644 core/transport/src/main/kotlin/app/handlive/android/core/transport/BulkLaneSocket.kt create mode 100644 core/transport/src/main/kotlin/app/handlive/android/core/transport/EnvelopeLane.kt create mode 100644 core/transport/src/main/kotlin/app/handlive/android/core/transport/bluetooth/BluetoothBulkLaneReassembler.kt create mode 100644 core/transport/src/test/kotlin/app/handlive/android/core/transport/bluetooth/BluetoothBulkLaneReassemblerTest.kt create mode 100644 feature/clipboard/src/test/kotlin/app/handlive/android/feature/clipboard/module/BulkLaneContractTest.kt 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/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..b491be0 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,36 @@ 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. */ - fun poll(): ByteArray? = - short.removeFirstOrNull() - ?: if (longBulkInflight >= bulkInflight) { - null - } else { - long.removeFirstOrNull()?.let { dispatch(it) } - ?: bulk.removeFirstOrNull()?.let { dispatch(it) } - } + /** + * 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()?.let { + inflightCompletesBulkEnvelope.addLast(false) + return it + } + if (longBulkInflight >= bulkInflight) return null + long.removeFirstOrNull()?.let { return dispatch(it) } + bulk.removeFirstOrNull()?.let { return dispatch(it) } + return null + } - /** 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/BluetoothPeerSocket.kt b/core/transport/src/main/kotlin/app/handlive/android/core/transport/bluetooth/BluetoothPeerSocket.kt index 5da84cb..f3d0b47 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 @@ -5,6 +5,7 @@ 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 @@ -19,6 +20,7 @@ import kotlinx.coroutines.channels.ReceiveChannel import kotlinx.coroutines.channels.SendChannel import kotlinx.coroutines.job import kotlinx.coroutines.launch +import kotlinx.coroutines.sync.Semaphore import kotlinx.coroutines.withTimeoutOrNull import java.security.GeneralSecurityException import kotlin.coroutines.CoroutineContext @@ -28,8 +30,8 @@ import kotlin.coroutines.CoroutineContext * `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 + * [BluetoothLongLaneReassembler], làn bulk (`chan` = 0x02, 0.4.7) là [BluetoothBulkLaneReassembler] 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,7 +46,7 @@ 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) @@ -52,6 +54,10 @@ internal class BluetoothPeerSocket( private val outgoingFrames = Channel(Channel.UNLIMITED) private val scheduler = BluetoothLinkScheduler() private val reassembler = BluetoothLongLaneReassembler() + private val bulkReassembler = BluetoothBulkLaneReassembler() + + /** `BT_BULK_QUEUED` (0.4.7 bulk rule 4): [sendBulk] chờ ở đây khi đã đủ envelope bulk chưa xả hết. */ + private val bulkQueueGate = Semaphore(BluetoothLinkConstants.BULK_QUEUED) 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). */ @@ -75,7 +81,7 @@ internal class BluetoothPeerSocket( val body = io.readRecord() val plaintext = openRecord(body) val frame = BtTransportFrame.decode(plaintext) - if (frame.chan == BtLinkConstants.CHAN_CONTROL) deliverControlFrame(frame) + if (frame.chan == BtLinkConstants.CHAN_CONTROL) deliverControlFrame(frame) else deliverBulkFrame(frame) } } catch (e: CancellationException) { throw e @@ -107,11 +113,23 @@ internal class BluetoothPeerSocket( message?.let { incomingFrames.trySend(Frame.Text(it.toString(Charsets.UTF_8))) } } + /** Đệm riêng của làn bulk (0.4.7 bulk rule 5-6): tin hoàn chỉnh lên cùng pipeline `/v1/ctl` như làn điều khiển. */ + private fun deliverBulkFrame(frame: BtTransportFrame) { + bulkReassembler.accept(frame)?.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) { + bulkQueueGate.acquire() + 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.Binary -> scheduler.enqueueBulk(frame.data) is Frame.Close -> break else -> Unit } @@ -144,7 +162,7 @@ internal class BluetoothPeerSocket( var plaintext = scheduler.poll() while (plaintext != null) { io.writeRecord(sendCipher.seal(plaintext)) - scheduler.onDrained() + if (scheduler.onDrained()) bulkQueueGate.release() plaintext = scheduler.poll() } } 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/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/module/ClipReceiver.kt b/feature/clipboard/src/main/kotlin/app/handlive/android/feature/clipboard/module/ClipReceiver.kt index 57c5dd3..7908eec 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,18 @@ class ClipReceiver( ("html" to 1).takeIf { push.html != null }, ), ) + // 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. + val lane = if (push.transfer != null) EnvelopeLane.BULK else EnvelopeLane.CONTROL 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 +83,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 +175,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 +189,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( @@ -191,7 +201,7 @@ class ClipReceiver( state.latest?.asLocalChange(), now, ) - if (decision != ConflictDecision.APPLY) return keepLocal(session, pushId, push, content, decision) + if (decision != ConflictDecision.APPLY) return keepLocal(session, pushId, push, content, decision, lane) val written = runCatching { val writer = context.platform.writer @@ -217,6 +227,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 +244,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 +258,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). */ @@ -260,24 +272,27 @@ class ClipReceiver( push: ClipboardPushData, content: ClipContent, decision: ConflictDecision, + lane: EnvelopeLane, ) { (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 ?: "-"), 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..65c49a5 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) 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..3a7bb8c 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) } } 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..ccf21ce --- /dev/null +++ b/feature/clipboard/src/test/kotlin/app/handlive/android/feature/clipboard/module/BulkLaneContractTest.kt @@ -0,0 +1,150 @@ +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 transferId = checkNotNull(h.mac.pushes().single().second.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()) + } +} 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..182bb60 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 @@ -180,6 +181,7 @@ class FakeClient( val type: MessageType, val plaintext: ByteArray, val id: String, + val lane: EnvelopeLane, ) val sent = mutableListOf() @@ -192,9 +194,9 @@ class FakeClient( PeerSession.Channel.LAN, 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 +213,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 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 }, ) From 55ccf6f12544f66abb2c07eeafedf843d360dcab Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?H=E1=BB=93=20Xu=C3=A2n=20D=C5=A9ng?= Date: Wed, 4 Nov 2026 10:21:45 +0700 Subject: [PATCH 2/9] refactor(transport): move bulk lane state out of BluetoothPeerSocket BluetoothBulkLane now holds the BT_BULK_QUEUED gate and the channel 0x02 reassembler; BluetoothLinkScheduler.poll keeps short > long > bulk and the BT_BULK_INFLIGHT cap in a single expression. ClipReceiver derives the reply lane from the push, and the SMS test harness takes the lane argument added to EnvelopeSender. Clears detekt and ktlint on check. --- .../transport/bluetooth/BluetoothBulkLane.kt | 25 ++++++++++ .../bluetooth/BluetoothLinkScheduler.kt | 17 +++---- .../bluetooth/BluetoothPeerSocket.kt | 49 ++++++++----------- .../feature/clipboard/module/ClipReceiver.kt | 11 +++-- .../android/feature/sms/testing/SmsHarness.kt | 2 +- 5 files changed, 60 insertions(+), 44 deletions(-) create mode 100644 core/transport/src/main/kotlin/app/handlive/android/core/transport/bluetooth/BluetoothBulkLane.kt 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/BluetoothLinkScheduler.kt b/core/transport/src/main/kotlin/app/handlive/android/core/transport/bluetooth/BluetoothLinkScheduler.kt index b491be0..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 @@ -61,16 +61,13 @@ class BluetoothLinkScheduler( * 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()?.let { - inflightCompletesBulkEnvelope.addLast(false) - return it - } - if (longBulkInflight >= bulkInflight) return null - long.removeFirstOrNull()?.let { return dispatch(it) } - bulk.removeFirstOrNull()?.let { return dispatch(it) } - return null - } + fun poll(): ByteArray? = + 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). 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 f3d0b47..defc152 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 @@ -20,7 +20,6 @@ import kotlinx.coroutines.channels.ReceiveChannel import kotlinx.coroutines.channels.SendChannel import kotlinx.coroutines.job import kotlinx.coroutines.launch -import kotlinx.coroutines.sync.Semaphore import kotlinx.coroutines.withTimeoutOrNull import java.security.GeneralSecurityException import kotlin.coroutines.CoroutineContext @@ -30,8 +29,8 @@ import kotlin.coroutines.CoroutineContext * `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], làn bulk (`chan` = 0x02, 0.4.7) là [BluetoothBulkLaneReassembler] và - * [sendBulk] (cả ba của #191/A7.4). [onFirstRecordOpened]/[onFirstRecordRejected] báo bước 5 cho + * [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( @@ -46,7 +45,8 @@ internal class BluetoothPeerSocket( private val receiveCipher: BtTransportCipher, parent: CoroutineScope, private val callbacks: BluetoothPeerSocketCallbacks = BluetoothPeerSocketCallbacks(), -) : WebSocketSession, BulkLaneSocket { +) : WebSocketSession, + BulkLaneSocket { override val coroutineContext: CoroutineContext = parent.coroutineContext + SupervisorJob(parent.coroutineContext.job) @@ -54,10 +54,7 @@ internal class BluetoothPeerSocket( private val outgoingFrames = Channel(Channel.UNLIMITED) private val scheduler = BluetoothLinkScheduler() private val reassembler = BluetoothLongLaneReassembler() - private val bulkReassembler = BluetoothBulkLaneReassembler() - - /** `BT_BULK_QUEUED` (0.4.7 bulk rule 4): [sendBulk] chờ ở đây khi đã đủ envelope bulk chưa xả hết. */ - private val bulkQueueGate = Semaphore(BluetoothLinkConstants.BULK_QUEUED) + private val bulkLane = BluetoothBulkLane() 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). */ @@ -78,10 +75,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) else deliverBulkFrame(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 @@ -108,19 +110,9 @@ 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))) } - } - - /** Đệm riêng của làn bulk (0.4.7 bulk rule 5-6): tin hoàn chỉnh lên cùng pipeline `/v1/ctl` như làn điều khiển. */ - private fun deliverBulkFrame(frame: BtTransportFrame) { - bulkReassembler.accept(frame)?.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) { - bulkQueueGate.acquire() + bulkLane.awaitQueueSlot() outgoingFrames.send(Frame.Binary(true, envelope.toByteArray(Charsets.UTF_8))) } @@ -128,7 +120,7 @@ internal class BluetoothPeerSocket( try { for (frame in outgoingFrames) { when (frame) { - is Frame.Text -> enqueueEnvelope(frame.readText()) + is Frame.Text -> enqueueControlEnvelope(frame.readText()) is Frame.Binary -> scheduler.enqueueBulk(frame.data) is Frame.Close -> break else -> Unit @@ -153,16 +145,15 @@ 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)) - } + /** Phong bì JSON của làn điều khiển: [BluetoothLinkScheduler] tự chọn làn ngắn hay dài theo `type` và cỡ. */ + private fun enqueueControlEnvelope(json: String) = + scheduler.enqueueControlEnvelope(EnvelopeCodec.decode(json).type, json.toByteArray(Charsets.UTF_8)) private suspend fun drainScheduler() { var plaintext = scheduler.poll() while (plaintext != null) { io.writeRecord(sendCipher.seal(plaintext)) - if (scheduler.onDrained()) bulkQueueGate.release() + if (scheduler.onDrained()) bulkLane.onEnvelopeWritten() plaintext = scheduler.poll() } } 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 7908eec..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 @@ -64,8 +64,7 @@ class ClipReceiver( ("html" to 1).takeIf { push.html != null }, ), ) - // 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. - val lane = if (push.transfer != null) EnvelopeLane.BULK else EnvelopeLane.CONTROL + val lane = push.replyLane val rejection = PushValidator.check(push, acceptance(session)) when { rejection != null -> { @@ -201,7 +200,7 @@ class ClipReceiver( state.latest?.asLocalChange(), now, ) - if (decision != ConflictDecision.APPLY) return keepLocal(session, pushId, push, content, decision, lane) + if (decision != ConflictDecision.APPLY) return keepLocal(session, pushId, push, content, decision) val written = runCatching { val writer = context.platform.writer @@ -272,8 +271,8 @@ class ClipReceiver( push: ClipboardPushData, content: ClipContent, decision: ConflictDecision, - lane: EnvelopeLane, ) { + 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, lane) @@ -306,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/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, ) From 091cdac9dda28324c67cef8fbd92e8079ad35552 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?H=E1=BB=93=20Xu=C3=A2n=20D=C5=A9ng?= Date: Wed, 4 Nov 2026 10:21:45 +0700 Subject: [PATCH 3/9] test(clipboard): Bluetooth progress from BULK_DIRECT_MIN (0.4.7) Over a Bluetooth session an image of at least BULK_DIRECT_MIN shows sending and receiving progress flagged for the Bluetooth note (CLIP-03 fields 2 and 10); below it, and on LAN under 1 MiB, nothing shows. --- .../clipboard/module/BulkLaneContractTest.kt | 55 ++++++++++++++++++- .../clipboard/testing/ClipboardTestKit.kt | 6 +- 2 files changed, 59 insertions(+), 2 deletions(-) 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 index ccf21ce..bc6b9a9 100644 --- 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 @@ -106,7 +106,8 @@ class BulkLaneContractTest { test { h -> h.connect(h.mac) h.readImage(Random(3).nextBytes(100_000)) - val transferId = checkNotNull(h.mac.pushes().single().second.transfer).transferId + 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()) @@ -147,4 +148,56 @@ class BulkLaneContractTest { 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 182bb60..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 @@ -176,6 +176,7 @@ class FakeClient( val deviceId: String, val pairId: String, clock: () -> Long, + channel: PeerSession.Channel = PeerSession.Channel.LAN, ) { class Sent( val type: MessageType, @@ -191,7 +192,7 @@ class FakeClient( val session = PeerSession( PeerSession.PeerInfo(pairId, deviceId, name, PeerPlatform.MACOS), - PeerSession.Channel.LAN, + channel, effective, capability, { type, plaintext, id, lane -> @@ -289,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() } From 3f11939e37d53fded004c490afc3abe4caf1cbe2 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?H=E1=BB=93=20Xu=C3=A2n=20D=C5=A9ng?= Date: Wed, 4 Nov 2026 10:21:45 +0700 Subject: [PATCH 4/9] feat(clipboard): show Bluetooth transfer progress from BULK_DIRECT_MIN TransferProgress carries overBluetooth so the notifier can add the CLIP-03 field 10 line; LAN and relay keep the 1 MiB threshold. --- .../android/feature/clipboard/ClipNotices.kt | 7 ++++- .../feature/clipboard/engine/ClipLimits.kt | 3 +++ .../clipboard/module/IncomingTransfers.kt | 11 ++++++-- .../clipboard/module/OutgoingTransfers.kt | 27 +++++++++++++++++-- 4 files changed, 43 insertions(+), 5 deletions(-) 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..0dbff8b 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. */ +/** + * Progress of an image over 1 MiB, or of at least `BULK_DIRECT_MIN` over Bluetooth (CLIP-03 fields 2–4); + * `percent = null` removes it. [overBluetooth]: the transfer runs on the Bluetooth bulk lane (0.4.7), so CLIP-03 + * field 10 shows under the progress. + */ 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/engine/ClipLimits.kt b/feature/clipboard/src/main/kotlin/app/handlive/android/feature/clipboard/engine/ClipLimits.kt index 0c900fa..b59badb 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): over Bluetooth, images of at least 128 KiB show progress (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/IncomingTransfers.kt b/feature/clipboard/src/main/kotlin/app/handlive/android/feature/clipboard/module/IncomingTransfers.kt index 65c49a5..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 @@ -185,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 3a7bb8c..7dfa9d9 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 @@ -122,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: an image shows progress above `PROGRESS_MIN_BYTES`, or from `BULK_DIRECT_MIN` up when the session + * is over Bluetooth (0.4.7 "Timeouts and progress": about 10 s at 12.9 KB/s, so the user sees it is still running). + */ +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, From be88bbe10d3c738f708d88abc13f0539b00e0a6f Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?H=E1=BB=93=20Xu=C3=A2n=20D=C5=A9ng?= Date: Wed, 4 Nov 2026 10:43:11 +0700 Subject: [PATCH 5/9] test(clipboard): Bluetooth note under the transfer progress Over the Bluetooth link the progress notification carries the CLIP-03 field 10 line under the percent, in English and Vietnamese; LAN progress has none. --- .../clipboard/component/ClipNotifierTest.kt | 41 +++++++++++++++++++ 1 file changed, 41 insertions(+) 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")) From 0a29dd00ef3d90c77ea2740845f310923557588b Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?H=E1=BB=93=20Xu=C3=A2n=20D=C5=A9ng?= Date: Wed, 4 Nov 2026 10:45:35 +0700 Subject: [PATCH 6/9] feat(clipboard): Bluetooth note under the transfer progress When a transfer runs over the Bluetooth link the progress notification expands to the percent line followed by clipboard.bulk_bluetooth_sending or _receiving (CLIP-03 field 10); LAN and relay progress is unchanged. --- .../clipboard/component/ClipNotifier.kt | 20 ++++++++++++++++--- 1 file changed, 17 insertions(+), 3 deletions(-) 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() = From 2cf835d61789cfb60ccb698309bbf05c462afc9e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?H=E1=BB=93=20Xu=C3=A2n=20D=C5=A9ng?= Date: Wed, 4 Nov 2026 10:49:07 +0700 Subject: [PATCH 7/9] test(transport): short control written mid-bulk arrives first (0.4.5 Priority) --- .../BluetoothPeerSocketPriorityTest.kt | 82 +++++++++++++++++++ 1 file changed, 82 insertions(+) create mode 100644 core/transport/src/test/kotlin/app/handlive/android/core/transport/bluetooth/BluetoothPeerSocketPriorityTest.kt 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..aef32c1 --- /dev/null +++ b/core/transport/src/test/kotlin/app/handlive/android/core/transport/bluetooth/BluetoothPeerSocketPriorityTest.kt @@ -0,0 +1,82 @@ +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 + } +} From d2c37e24b8c8513dbddf318420e248f3321134b5 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?H=E1=BB=93=20Xu=C3=A2n=20D=C5=A9ng?= Date: Wed, 4 Nov 2026 10:54:45 +0700 Subject: [PATCH 8/9] fix(transport): let short control cut in between bulk fragments (0.4.5 Priority) --- .../bluetooth/BluetoothLinkWritePump.kt | 69 +++++++++++++++++++ .../bluetooth/BluetoothPeerSocket.kt | 38 +++------- .../BluetoothPeerSocketPriorityTest.kt | 8 ++- 3 files changed, 86 insertions(+), 29 deletions(-) create mode 100644 core/transport/src/main/kotlin/app/handlive/android/core/transport/bluetooth/BluetoothLinkWritePump.kt 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 defc152..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,12 +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 @@ -28,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], 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 + * 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( @@ -52,9 +50,14 @@ internal class BluetoothPeerSocket( 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). */ @@ -118,15 +121,7 @@ internal class BluetoothPeerSocket( private suspend fun writeLoop() { try { - for (frame in outgoingFrames) { - when (frame) { - is Frame.Text -> enqueueControlEnvelope(frame.readText()) - is Frame.Binary -> scheduler.enqueueBulk(frame.data) - is Frame.Close -> break - else -> Unit - } - drainScheduler() - } + writePump.run() } catch (e: CancellationException) { throw e } catch (_: BtLinkException) { @@ -145,19 +140,6 @@ internal class BluetoothPeerSocket( withTimeoutOrNull(timeoutMillis) { writesFinished.await() } } - /** Phong bì JSON của làn điều khiển: [BluetoothLinkScheduler] tự chọn làn ngắn hay dài theo `type` và cỡ. */ - private fun enqueueControlEnvelope(json: String) = - scheduler.enqueueControlEnvelope(EnvelopeCodec.decode(json).type, json.toByteArray(Charsets.UTF_8)) - - private suspend fun drainScheduler() { - var plaintext = scheduler.poll() - while (plaintext != null) { - io.writeRecord(sendCipher.seal(plaintext)) - if (scheduler.onDrained()) bulkLane.onEnvelopeWritten() - 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/test/kotlin/app/handlive/android/core/transport/bluetooth/BluetoothPeerSocketPriorityTest.kt b/core/transport/src/test/kotlin/app/handlive/android/core/transport/bluetooth/BluetoothPeerSocketPriorityTest.kt index aef32c1..688f367 100644 --- 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 @@ -33,7 +33,13 @@ class BluetoothPeerSocketPriorityTest { type: String, bytes: Int, ) = EnvelopeCodec.encode( - Envelope(Envelope.VERSION, type, ids.next(), System.currentTimeMillis(), Base64Codecs.encodeB64(ByteArray(bytes))), + Envelope( + Envelope.VERSION, + type, + ids.next(), + System.currentTimeMillis(), + Base64Codecs.encodeB64(ByteArray(bytes)), + ), ) @Test From cf64ef54f39cc8163dc2e7664fb32898485fb696 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?H=E1=BB=93=20Xu=C3=A2n=20D=C5=A9ng?= Date: Wed, 4 Nov 2026 10:54:45 +0700 Subject: [PATCH 9/9] style(clipboard): Vietnamese comments for the Bluetooth bulk progress --- .../app/handlive/android/feature/clipboard/ClipNotices.kt | 6 +++--- .../handlive/android/feature/clipboard/engine/ClipLimits.kt | 2 +- .../android/feature/clipboard/module/OutgoingTransfers.kt | 4 ++-- 3 files changed, 6 insertions(+), 6 deletions(-) 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 0dbff8b..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 @@ -45,9 +45,9 @@ sealed interface ClipMessage { } /** - * Progress of an image over 1 MiB, or of at least `BULK_DIRECT_MIN` over Bluetooth (CLIP-03 fields 2–4); - * `percent = null` removes it. [overBluetooth]: the transfer runs on the Bluetooth bulk lane (0.4.7), so CLIP-03 - * field 10 shows under the progress. + * 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, 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 b59badb..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,7 +17,7 @@ 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): over Bluetooth, images of at least 128 KiB show progress (CLIP-03 field 2, 10). */ + /** `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). */ 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 7dfa9d9..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 @@ -137,8 +137,8 @@ class OutgoingTransfers( } /** - * CLIP-03 field 2: an image shows progress above `PROGRESS_MIN_BYTES`, or from `BULK_DIRECT_MIN` up when the session - * is over Bluetooth (0.4.7 "Timeouts and progress": about 10 s at 12.9 KB/s, so the user sees it is still running). + * 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,