Phase 7 SP3: làn bulk qua Bluetooth (kênh 0x02)

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

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)

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

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)
+}

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,
+}

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)
+}

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
+    }
+}

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
 }

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<String>()
 
+    /** 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<Boolean>()
+
     /** Phong bì JSON (0.5.1) trên kênh điều khiển `0x01`; tự chọn làn ngắn hay dài theo 0.4.5 "Control channel". */
     fun enqueueControlEnvelope(
         type: String,
@@ -52,23 +55,33 @@ class BluetoothLinkScheduler(
 
     fun hasPending(): Boolean = short.isNotEmpty() || long.isNotEmpty() || bulk.isNotEmpty()
 
-    /** Mảnh plaintext kế tiếp theo ưu tiên 0.4.5 "Priority", hay `null` nếu không còn gì để gửi ngay. */
+    /**
+     * Mảnh plaintext kế tiếp theo ưu tiên 0.4.5 "Priority", hay `null` nếu không còn gì để gửi ngay. Bên gọi viết
+     * mảnh này lên dây rồi gọi [onDrained] đúng một lần cho nó (vòng ghi của `BluetoothPeerSocket` làm vậy, tuần
+     * tự) — [inflightCompletesBulkEnvelope] phải nhận một mục cho MỌI mảnh trả về ở đây, không riêng dài/bulk, để
+     * khớp thứ tự với [onDrained].
+     */
     fun poll(): ByteArray? =
-        short.removeFirstOrNull()
-            ?: if (longBulkInflight >= bulkInflight) {
-                null
-            } else {
-                long.removeFirstOrNull()?.let { dispatch(it) }
-                    ?: bulk.removeFirstOrNull()?.let { dispatch(it) }
-            }
+        short.removeFirstOrNull()?.also { inflightCompletesBulkEnvelope.addLast(false) }
+            ?: nextLongOrBulk()?.let { dispatch(it) }
+
+    /** Làn dài trước làn bulk, và chỉ khi còn chỗ trong [bulkInflight]; hết chỗ thì `null` dù hàng đợi còn mảnh. */
+    private fun nextLongOrBulk(): QueuedFrame? =
+        if (longBulkInflight >= bulkInflight) null else long.removeFirstOrNull() ?: bulk.removeFirstOrNull()
 
-    /** Gọi khi một bản ghi dài/bulk đã [poll] trước đó thật sự xả khỏi bộ đệm ghi (mở lại một chỗ inflight). */
-    fun onDrained() {
+    /**
+     * Gọi khi một bản ghi dài/bulk đã [poll] trước đó thật sự xả khỏi bộ đệm ghi (mở lại một chỗ inflight).
+     * Trả `true` khi bản ghi đó là mảnh cuối của một envelope bulk — envelope ấy giờ đã "fully written" (0.4.7
+     * bulk rule 4), bên gọi (chủ của [enqueueBulk]) có thể mở thêm một chỗ hàng đợi `BT_BULK_QUEUED`.
+     */
+    fun onDrained(): Boolean {
         if (longBulkInflight > 0) longBulkInflight--
+        return inflightCompletesBulkEnvelope.removeFirstOrNull() == true
     }
 
     private fun dispatch(frame: QueuedFrame): ByteArray {
         longBulkInflight++
+        inflightCompletesBulkEnvelope.addLast(frame.isLastOfMessage && frame.type == null)
         if (frame.isLastOfMessage) frame.type?.let { typesPendingLong -= it }
         return frame.bytes
     }

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<Frame>,
+    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)
+        }
+    }
+}

core/transport/src/main/kotlin/app/handlive/android/core/transport/bluetooth/BluetoothPeerSocket.kt

@@ -4,11 +4,10 @@ import app.handlive.android.core.protocol.bluetooth.BtLinkConstants
 import app.handlive.android.core.protocol.bluetooth.BtLinkException
 import app.handlive.android.core.protocol.bluetooth.BtTransportFrame
 import app.handlive.android.core.protocol.envelope.Envelope
-import app.handlive.android.core.protocol.envelope.EnvelopeCodec
+import app.handlive.android.core.transport.BulkLaneSocket
 import io.ktor.websocket.Frame
 import io.ktor.websocket.WebSocketExtension
 import io.ktor.websocket.WebSocketSession
-import io.ktor.websocket.readText
 import kotlinx.coroutines.CancellationException
 import kotlinx.coroutines.CompletableDeferred
 import kotlinx.coroutines.CoroutineScope
@@ -27,9 +26,9 @@ import kotlin.coroutines.CoroutineContext
  * Một phiên `/v1/ctl` qua liên kết Bluetooth đã qua bắt tay bảo mật liên kết (0.4.5 bước 4-5, #191): như
  * `RelayPeerSocket` của relay, `ControlServer.serveBluetoothPeer` nhận socket ảo này mà không đụng tới dây L2CAP
  * bên dưới — E2E `session/hello` (0.6.3) chạy y nguyên bên trong (0.4.5 bước 6). [sendCipher]/[receiveCipher] là
- * hai nửa của `Split()`; việc cắt mảnh/ưu tiên là [BluetoothLinkScheduler], ghép lại làn dài là
- * [BluetoothLongLaneReassembler] (cả hai của #191). Kênh bulk (`chan` = 0x02) bỏ qua ở đây — nội dung của nó
- * thuộc 0.4.7 (A7.4), ngoài phạm vi thẻ A7.1b. [onFirstRecordOpened]/[onFirstRecordRejected] báo bước 5 cho
+ * hai nửa của `Split()`; việc cắt mảnh/ưu tiên là [BluetoothLinkScheduler] (vòng ghi: [BluetoothLinkWritePump]),
+ * ghép lại làn dài là [BluetoothLongLaneReassembler], làn bulk (`chan` = 0x02, 0.4.7) là [BluetoothBulkLane] và
+ * [sendBulk] (cả ba của #191/A7.4). [onFirstRecordOpened]/[onFirstRecordRejected] báo bước 5 cho
  * `BluetoothLinkAcceptor` (admission, `BT_PREAUTH_TIMEOUT`).
  */
 internal data class BluetoothPeerSocketCallbacks(
@@ -44,14 +43,21 @@ internal class BluetoothPeerSocket(
     private val receiveCipher: BtTransportCipher,
     parent: CoroutineScope,
     private val callbacks: BluetoothPeerSocketCallbacks = BluetoothPeerSocketCallbacks(),
-) : WebSocketSession {
+) : WebSocketSession,
+    BulkLaneSocket {
     override val coroutineContext: CoroutineContext =
         parent.coroutineContext + SupervisorJob(parent.coroutineContext.job)
 
     private val incomingFrames = Channel<Frame>(Channel.UNLIMITED)
     private val outgoingFrames = Channel<Frame>(Channel.UNLIMITED)
-    private val scheduler = BluetoothLinkScheduler()
     private val reassembler = BluetoothLongLaneReassembler()
+    private val bulkLane = BluetoothBulkLane()
+    private val writePump =
+        BluetoothLinkWritePump(
+            outgoing = outgoingFrames,
+            writePlaintext = { io.writeRecord(sendCipher.seal(it)) },
+            onBulkEnvelopeWritten = bulkLane::onEnvelopeWritten,
+        )
     private var anyRecordOpened = false
 
     /** Vòng ghi đã dừng: mọi frame xếp trước một `Frame.Close` đã lên dây (hoặc dây đã đứt). */
@@ -72,10 +78,15 @@ internal class BluetoothPeerSocket(
     private suspend fun readLoop() {
         try {
             while (true) {
-                val body = io.readRecord()
-                val plaintext = openRecord(body)
-                val frame = BtTransportFrame.decode(plaintext)
-                if (frame.chan == BtLinkConstants.CHAN_CONTROL) deliverControlFrame(frame)
+                val frame = BtTransportFrame.decode(openRecord(io.readRecord()))
+                // Làn bulk có đệm riêng (0.4.7 bulk rule 5-6); tin đủ của cả hai kênh lên cùng pipeline `/v1/ctl`.
+                val message =
+                    when {
+                        frame.chan != BtLinkConstants.CHAN_CONTROL -> bulkLane.accept(frame)
+                        frame.long -> reassembler.accept(frame)
+                        else -> frame.data
+                    }
+                message?.let { incomingFrames.trySend(Frame.Text(it.toString(Charsets.UTF_8))) }
             }
         } catch (e: CancellationException) {
             throw e
@@ -102,21 +113,15 @@ internal class BluetoothPeerSocket(
         return plaintext
     }
 
-    private fun deliverControlFrame(frame: BtTransportFrame) {
-        val message = if (frame.long) reassembler.accept(frame) else frame.data
-        message?.let { incomingFrames.trySend(Frame.Text(it.toString(Charsets.UTF_8))) }
+    /** Làn bulk (0.4.7): [BulkLaneSocket.sendBulk] gọi trực tiếp, không qua [outgoingFrames] như làn điều khiển. */
+    override suspend fun sendBulk(envelope: String) {
+        bulkLane.awaitQueueSlot()
+        outgoingFrames.send(Frame.Binary(true, envelope.toByteArray(Charsets.UTF_8)))
     }
 
     private suspend fun writeLoop() {
         try {
-            for (frame in outgoingFrames) {
-                when (frame) {
-                    is Frame.Text -> enqueueEnvelope(frame.readText())
-                    is Frame.Close -> break
-                    else -> Unit
-                }
-                drainScheduler()
-            }
+            writePump.run()
         } catch (e: CancellationException) {
             throw e
         } catch (_: BtLinkException) {
@@ -135,20 +140,6 @@ internal class BluetoothPeerSocket(
         withTimeoutOrNull(timeoutMillis) { writesFinished.await() }
     }
 
-    private fun enqueueEnvelope(json: String) {
-        val type = EnvelopeCodec.decode(json).type
-        scheduler.enqueueControlEnvelope(type, json.toByteArray(Charsets.UTF_8))
-    }
-
-    private suspend fun drainScheduler() {
-        var plaintext = scheduler.poll()
-        while (plaintext != null) {
-            io.writeRecord(sendCipher.seal(plaintext))
-            scheduler.onDrained()
-            plaintext = scheduler.poll()
-        }
-    }
-
     /** Dây L2CAP đứt hoặc phiên đóng: đóng cả hai hàng đợi và dây vật lý bên dưới. */
     fun end() {
         incomingFrames.close()

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
                 }