diff --git a/CHANGELOG.md b/CHANGELOG.md index 589054d157..fb62f878a8 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -194,6 +194,8 @@ The main work (all changes without a GitHub username in brackets in the below li - Fix: Giving up on a peripheral reports the last connection failure as cause - Fix: A disconnect that never completes no longer leaves the channel request pending forever - Fix: Aborting a BLE connection attempt now stops its retries and releases a link the attempt already established + - Fix: A peripheral that completes the BTP handshake but then never responds to data is retried once with the minimum BTP segment size instead of failing commissioning + - Fix: A pending ATT_MTU exchange is awaited briefly, so a late MTU no longer pins the BTP segment size to the minimum - Fix: Stopping an advertisement that was still waiting for the Bluetooth adapter retracts it, so the adapter powering on no longer starts an advertisement that was already given up on - @matter/nodejs-shell @@ -209,6 +211,9 @@ The main work (all changes without a GitHub username in brackets in the below li - Enhancement: New `CertificateAuthority.erase()` discards the authority's key material, persisted and in memory - Enhancement: A discovered commissionable device reports the `hostname` its SRV record names - Enhancement: Commissioning accepts `caseConnectionTimeout`, bounding how long it waits for the operational CASE connection that follows it; defaults to the previous fixed 4m15s + - Enhancement: `BtpSessionHandler.stalledAfterHandshake` reports a peer that answers the handshake and then nothing else, carrying the messages it never acknowledged; a session nobody observes closes on the acknowledgement timeout as before + - Enhancement: New `BtpCodec.isHandshakeResponse()` identifies a BTP handshake response without a session to decode against + - Fix: A central BTP session no longer accepts a segment size larger than the one it offered; `createAsCentral` requires the offered size - Fix: Cancelling BLE commissioning aborts the in-flight channel open - Fix: A concrete subscription path is reported only when that attribute changed; it was previously reported whenever any other attribute of the same cluster changed - Fix: A subscription's `maxIntervalCeiling` is transmitted exactly as requested; jitter now applies only when we derive the ceiling ourselves diff --git a/packages/nodejs-ble/src/NobleBleChannel.ts b/packages/nodejs-ble/src/NobleBleChannel.ts index 98e1d13c4b..be52c51a2c 100644 --- a/packages/nodejs-ble/src/NobleBleChannel.ts +++ b/packages/nodejs-ble/src/NobleBleChannel.ts @@ -23,15 +23,7 @@ import { createPromise, withTimeout, } from "@matter/general"; -import { - BleChannel, - BleDisconnectedError, - BleError, - BtpCodec, - BtpFlowError, - BtpSessionHandler, - MatterBle, -} from "@matter/protocol"; +import { BleChannel, BleDisconnectedError, BleError, BtpCodec, BtpSessionHandler, MatterBle } from "@matter/protocol"; import type { Characteristic, Peripheral } from "@stoprocent/noble"; import { BleScanner } from "./BleScanner.js"; import { nobleDisconnectReason } from "./NobleBleClient.js"; @@ -41,6 +33,9 @@ const logger = Logger.get("BleChannel"); /** noble waits for a disconnect event that a vanished peripheral never sends, so the wait needs its own bound. */ const BLE_DISCONNECT_TIMEOUT = Seconds(5); +/** How long to wait for a pending ATT_MTU exchange before deriving the BTP segment size without it. */ +const ATT_MTU_SETTLE_TIMEOUT = Seconds(2); + /** * Detect noble errors that indicate the BLE connection is no longer usable. * On macOS/Linux noble throws errors starting with "Disconnected". @@ -397,6 +392,7 @@ export class NobleBleCentralInterface implements Transport { characteristicC2ForSubscribe, this.#onMatterMessageListener, additionalCommissioningRelatedData, + abort, ); clearConnectionGuard(); releaseSlot(); @@ -516,6 +512,148 @@ export class NobleBleCentralInterface implements Transport { } } +/** + * Resolve the peripheral's ATT_MTU, waiting briefly when the exchange is still in flight. + * + * noble reports the negotiated MTU through an event and leaves `Peripheral.mtu` null until it arrives, which can be + * after the interview completes. Deriving the BTP segment size from an unknown MTU would pin the session to the 20-byte + * minimum for its whole life, so give the exchange a moment to land. + */ +async function attMtuOf(peripheral: Peripheral) { + if (peripheral.mtu !== null) { + return peripheral.mtu; + } + + const { promise, resolver } = createPromise(); + let settled = false; + const settle = (mtu?: number) => { + if (settled) { + return; + } + settled = true; + settleTimeout.stop(); + peripheral.removeListener("mtu", onMtu); + peripheral.removeListener("disconnect", onDisconnect); + resolver(mtu); + }; + const onMtu = (mtu: number) => settle(mtu); + const onDisconnect = () => settle(); + const settleTimeout = Time.getTimer("BLE ATT_MTU exchange", ATT_MTU_SETTLE_TIMEOUT, () => settle()).start(); + + peripheral.on("mtu", onMtu); + peripheral.on("disconnect", onDisconnect); + + return await promise; +} + +/** + * Unsubscribe from C2, which is how a GATT client closes a BTP session (§4.19.3.3). Bounded because noble leaves the + * operation pending when the peripheral has vanished without a disconnect event. + */ +async function unsubscribeC2( + peripheral: Peripheral, + characteristicC2ForSubscribe: Characteristic, + onFailure: "throw" | "log" = "log", +) { + try { + await withTimeout(BLE_DISCONNECT_TIMEOUT, characteristicC2ForSubscribe.unsubscribeAsync()); + } catch (error) { + if (onFailure === "throw") { + throw error; + } + if (!isNobleDisconnectError(error)) { + logger.warn(`Peripheral ${peripheral.address}: Error while unsubscribing from C2`, error); + } + } +} + +/** + * Run the BTP session handshake over an established GATT connection and return the peripheral's handshake response. + */ +async function performBtpHandshake( + peripheral: Peripheral, + characteristicC1ForWrite: Characteristic, + characteristicC2ForSubscribe: Characteristic, + segmentSize: number, + abort?: AbortSignal, +): Promise { + const { address: peripheralAddress } = peripheral; + if (abort?.aborted) { + throw new AbortedError(`Peripheral ${peripheralAddress}: BTP handshake was aborted`); + } + + const { + promise: handshakeResponseReceivedPromise, + resolver: handshakeResolver, + rejecter: handshakeRejecter, + } = createPromise(); + + const handshakeHandler = (data: Buffer, isNotification: boolean) => { + if (BtpCodec.isHandshakeResponse(data)) { + logger.info( + `Peripheral ${peripheralAddress}: Received Matter handshake response: ${data.toString("hex")}.`, + ); + btpHandshakeTimeout.stop(); + handshakeResolver(data); + } else { + logger.debug( + `Peripheral ${peripheralAddress}: Received first data on C2: ${data.toString("hex")} (isNotification: ${isNotification}) - No handshake response, ignoring`, + ); + } + }; + + const btpHandshakeTimeout = Time.getTimer("BLE handshake timeout", MatterBle.BTP_CONN_RSP_TIMEOUT, async () => { + logger.debug(`Peripheral ${peripheralAddress}: Handshake Response not received. Disconnect from peripheral`); + + // Reject before unsubscribing: the caller's bound must not depend on an unsubscribe that can hang + handshakeRejecter(new BleError(`Peripheral ${peripheralAddress}: Handshake Response not received`)); + + if (peripheral.state === "connected") { + await unsubscribeC2(peripheral, characteristicC2ForSubscribe); + } + }).start(); + + const onAbort = () => { + btpHandshakeTimeout.stop(); + handshakeRejecter(new AbortedError(`Peripheral ${peripheralAddress}: BTP handshake was aborted`)); + }; + abort?.addEventListener("abort", onAbort, { once: true }); + + const btpHandshakeRequest = BtpCodec.encodeBtpHandshakeRequest({ + versions: MatterBle.BTP_SUPPORTED_VERSIONS, + attMtu: segmentSize, + clientWindowSize: MatterBle.BTP_MAXIMUM_WINDOW_SIZE, + }); + + logger.debug( + `Peripheral ${peripheralAddress}: Sending BTP handshake request: ${Diagnostic.json(btpHandshakeRequest)}`, + ); + + try { + await characteristicC1ForWrite.writeAsync(Buffer.from(Bytes.of(btpHandshakeRequest)), false); + + characteristicC2ForSubscribe.on("data", handshakeHandler); + + logger.debug(`Peripheral ${peripheralAddress}: Subscribing to C2 characteristic`); + // Awaited together: a subscribe noble leaves pending must not strand the handshake rejection + const [, response] = await Promise.all([ + characteristicC2ForSubscribe.subscribeAsync(), + handshakeResponseReceivedPromise, + ]); + + return new Uint8Array(response); + } catch (error) { + btpHandshakeTimeout.stop(); + if (isNobleDisconnectError(error)) { + throw new BleDisconnectedError(error.message, { cause: error }); + } + throw error; + } finally { + abort?.removeEventListener("abort", onAbort); + characteristicC2ForSubscribe.removeListener("data", handshakeHandler); + } +} + export class NobleBleChannel extends BleChannel { static async create( peripheral: Peripheral, @@ -523,88 +661,111 @@ export class NobleBleChannel extends BleChannel { characteristicC2ForSubscribe: Characteristic, onMatterMessageListener: (socket: Channel, data: Bytes) => void, _additionalCommissioningRelatedData?: Bytes, + abort?: AbortSignal, ): Promise { - const { address: peripheralAddress } = peripheral; - const mtu = MatterBle.btpSegmentSizeFromAttMtu(peripheral.mtu ?? 0); - logger.debug( - `Peripheral ${peripheralAddress}: Using BTP segment size=${mtu} bytes (Peripheral ATT_MTU up to ${peripheral.mtu} bytes)`, - ); - - const { - promise: handshakeResponseReceivedPromise, - resolver: handshakeResolver, - rejecter: handshakeRejecter, - } = createPromise(); - - const handshakeHandler = (data: Buffer, isNotification: boolean) => { - if (data[0] === 0x65 && data[1] === 0x6c && data.length === 6) { - // Check if the first two bytes and length match the Matter handshake - logger.info( - `Peripheral ${peripheralAddress}: Received Matter handshake response: ${data.toString("hex")}.`, - ); - btpHandshakeTimeout.stop(); - handshakeResolver(data); - } else { - logger.debug( - `Peripheral ${peripheralAddress}: Received first data on C2: ${data.toString("hex")} (isNotification: ${isNotification}) - No handshake response, ignoring`, - ); - } - }; - - const btpHandshakeTimeout = Time.getTimer("BLE handshake timeout", MatterBle.BTP_CONN_RSP_TIMEOUT, async () => { - characteristicC2ForSubscribe.removeListener("data", handshakeHandler); - - if (peripheral.state === "connected") { - await characteristicC2ForSubscribe.unsubscribeAsync().catch(error => { - if (!isNobleDisconnectError(error)) { - logger.warn(`Peripheral ${peripheralAddress}: Error while unsubscribing`, error); - } - }); - } - + const attMtu = await attMtuOf(peripheral); + const segmentSize = + attMtu === undefined ? MatterBle.MINIMUM_ATT_MTU : MatterBle.btpSegmentSizeFromAttMtu(attMtu); + if (attMtu === undefined) { + logger.info( + `Peripheral ${peripheral.address}: ATT_MTU still unknown, using the minimum BTP segment size of ${segmentSize} bytes`, + ); + } else { logger.debug( - `Peripheral ${peripheralAddress}: Handshake Response not received. Disconnect from peripheral`, + `Peripheral ${peripheral.address}: Using BTP segment size=${segmentSize} bytes (Peripheral ATT_MTU up to ${attMtu} bytes)`, ); + } - handshakeRejecter(new BleError(`Peripheral ${peripheralAddress}: Handshake Response not received`)); - }).start(); - - const btpHandshakeRequest = BtpCodec.encodeBtpHandshakeRequest({ - versions: MatterBle.BTP_SUPPORTED_VERSIONS, - attMtu: mtu, - clientWindowSize: MatterBle.BTP_MAXIMUM_WINDOW_SIZE, - }); - - logger.debug( - `Peripheral ${peripheralAddress}: Sending BTP handshake request: ${Diagnostic.json(btpHandshakeRequest)}`, + const handshakeResponse = await performBtpHandshake( + peripheral, + characteristicC1ForWrite, + characteristicC2ForSubscribe, + segmentSize, + abort, ); + const channel = new NobleBleChannel( + peripheral, + characteristicC1ForWrite, + characteristicC2ForSubscribe, + onMatterMessageListener, + ); try { - await characteristicC1ForWrite.writeAsync(Buffer.from(Bytes.of(btpHandshakeRequest)), false); + await channel.#adoptSession(handshakeResponse, segmentSize); + } catch (error) { + // The peripheral outlives a rejected attempt and is reused by the next one, so a channel nobody receives + // must leave no listener behind + channel.#releasePeripheralListener(); + throw error; + } + return channel; + } + + #connected = true; + readonly #closeListeners = new Set<() => void>(); + #iteratorQueue = new Array(); + #iteratorWaiter?: (value: IteratorResult) => void; + #iteratorDone = false; - characteristicC2ForSubscribe.on("data", handshakeHandler); + #btpSession?: BtpSessionHandler; + #c2DataHandler?: (data: Buffer, isNotification: boolean) => void; + #renegotiation?: Promise; + #closing = false; + readonly #lifetime = new AbortController(); + readonly #onPeripheralDisconnect: (reason: unknown) => void; - logger.debug(`Peripheral ${peripheralAddress}: Subscribing to C2 characteristic`); - await characteristicC2ForSubscribe.subscribeAsync(); - } catch (error) { - btpHandshakeTimeout.stop(); - characteristicC2ForSubscribe.removeListener("data", handshakeHandler); - if (isNobleDisconnectError(error)) { - throw new BleDisconnectedError(error.message, { cause: error }); + private constructor( + private readonly peripheral: Peripheral, + private readonly characteristicC1ForWrite: Characteristic, + private readonly characteristicC2ForSubscribe: Characteristic, + private readonly onMatterMessageListener: (socket: Channel, data: Bytes) => void, + ) { + super(); + this.#onPeripheralDisconnect = (reason: unknown) => { + logger.debug( + `Disconnected from peripheral ${peripheral.address} (reason ${nobleDisconnectReason(reason)}). Closing BTP session`, + ); + this.#connected = false; + // Same reason as in close(): a renegotiation parked on the handshake would otherwise hold its timer, and + // any send waiting on it, until the handshake times out + this.#lifetime.abort(); + this.#detachDataHandler(); + this.#terminateIterator(); + for (const listener of this.#closeListeners) { + listener(); } - throw error; + this.#btpSession?.close().catch(error => { + logger.debug(`Peripheral ${peripheral.address}: Error closing BTP session on disconnect`, error); + }); + this.emitClosed(); + }; + peripheral.once("disconnect", this.#onPeripheralDisconnect); + } + + #releasePeripheralListener() { + this.peripheral.removeListener("disconnect", this.#onPeripheralDisconnect); + } + + /** The session is installed before {@link create} hands the channel out, so absence means an internal error. */ + get #session() { + if (this.#btpSession === undefined) { + throw new InternalError(`Peripheral ${this.peripheral.address}: No BTP session initialized`); } + return this.#btpSession; + } - const handshakeResponse = await handshakeResponseReceivedPromise; - characteristicC2ForSubscribe.removeListener("data", handshakeHandler); + /** Install a freshly handshaken BTP session and route incoming C2 data to it. */ + async #adoptSession(handshakeResponse: Bytes, requestedSegmentSize: number) { + const { address: peripheralAddress } = this.peripheral; + this.#detachDataHandler(); - const btpSession = await BtpSessionHandler.createAsCentral( - new Uint8Array(handshakeResponse), + const session = await BtpSessionHandler.createAsCentral( + handshakeResponse, // callback to write data to characteristic C1; translates noble's disconnect/transport // errors into BleDisconnectedError so BtpSessionHandler can handle them specifically async (data: Bytes) => { try { - return await characteristicC1ForWrite.writeAsync(Buffer.from(Bytes.of(data)), false); + return await this.characteristicC1ForWrite.writeAsync(Buffer.from(Bytes.of(data)), false); } catch (error) { if (isNobleDisconnectError(error)) { throw new BleDisconnectedError(error.message, { cause: error }); @@ -614,20 +775,14 @@ export class NobleBleChannel extends BleChannel { }, // callback to disconnect the BLE connection async () => { - if (peripheral.state !== "connected" || !nobleChannel.connected) return; + if (this.peripheral.state !== "connected" || !this.connected) return; logger.debug(`Peripheral ${peripheralAddress}: Disconnect from peripheral because btp session closed`); - characteristicC2ForSubscribe - .unsubscribeAsync() - .catch(error => { - if (!isNobleDisconnectError(error)) { - logger.debug(`Peripheral ${peripheralAddress}: Error while unsubscribing from C2`, error); - } - }) + unsubscribeC2(this.peripheral, this.characteristicC2ForSubscribe) .then(() => { - if (peripheral.state !== "connected") { + if (this.peripheral.state !== "connected") { return; } - return peripheral.disconnectAsync().then( + return this.peripheral.disconnectAsync().then( () => logger.debug(`Peripheral ${peripheralAddress}: Disconnected from peripheral`), error => logger.debug(`Peripheral ${peripheralAddress}: Error while disconnecting`, error), ); @@ -639,63 +794,112 @@ export class NobleBleChannel extends BleChannel { // callback to forward decoded and de-assembled Matter messages async (data: Bytes) => { - if (onMatterMessageListener === undefined) { + if (this.onMatterMessageListener === undefined) { throw new InternalError(`No listener registered for Matter messages`); } - nobleChannel.pushMessage(data); - onMatterMessageListener(nobleChannel, data); + this.pushMessage(data); + this.onMatterMessageListener(this, data); }, + requestedSegmentSize, ); + if (this.#closing || !this.connected) { + // The disconnect that ended the channel ran before this session existed, so nothing else would stop its + // timers or remove a data listener we attached now + session.suspend(); + throw new BleDisconnectedError( + `Peripheral ${peripheralAddress}: Channel was lost while establishing the BTP session`, + ); + } + + this.#btpSession = session; + + // Forward BTP-initiated close (e.g. ack-receive timeout) to our Observable. + session.closed.once(() => this.emitClosed()); + session.stalledAfterHandshake.once(messagesToReplay => this.#startRenegotiation(messagesToReplay)); + const c2DataHandler = (data: Buffer, isNotification: boolean) => { logger.debug( `Peripheral ${peripheralAddress}: received data on C2: ${data.toString("hex")} (isNotification: ${isNotification})`, ); - btpSession.handleIncomingBleData(new Uint8Array(data)).catch(error => { + session.handleIncomingBleData(new Uint8Array(data)).catch(error => { logger.info(`Peripheral ${peripheralAddress}: Error handling incoming BLE data`, error); }); }; - characteristicC2ForSubscribe.on("data", c2DataHandler); + this.#c2DataHandler = c2DataHandler; + this.characteristicC2ForSubscribe.on("data", c2DataHandler); + } - const nobleChannel = new NobleBleChannel(peripheral, btpSession, () => { - characteristicC2ForSubscribe.removeListener("data", c2DataHandler); - }); - return nobleChannel; + #detachDataHandler() { + if (this.#c2DataHandler !== undefined) { + this.characteristicC2ForSubscribe.removeListener("data", this.#c2DataHandler); + this.#c2DataHandler = undefined; + } } - #connected = true; - readonly #closeListeners = new Set<() => void>(); - #iteratorQueue = new Array(); - #iteratorWaiter?: (value: IteratorResult) => void; - #iteratorDone = false; + /** + * Establish a fresh BTP session with the smallest permitted segment size, replaying what the peer never + * acknowledged. See {@link BtpSessionHandler.stalledAfterHandshake} for why, and for the fact that this is an + * interop workaround rather than specified behaviour. + * + * Runs at most once per channel: a session already at the minimum segment size never reports the condition. + */ + #startRenegotiation(messagesToReplay: readonly Bytes[]) { + if (this.#renegotiation !== undefined) { + return; + } + // A send must be able to observe the renegotiation before its first step runs, or it reaches the session that + // was just suspended + this.#renegotiation = Promise.resolve() + .then(() => this.#renegotiate(messagesToReplay)) + .catch(async error => { + logger.warn(`Peripheral ${this.peripheral.address}: Renegotiating the BTP session failed`, error); + // Every send awaits this promise, so it must not settle rejected + await this.close().catch(closeError => + logger.debug( + `Peripheral ${this.peripheral.address}: Error closing after a failed renegotiation`, + closeError, + ), + ); + }); + } - readonly #cleanupDataListener: () => void; + async #renegotiate(messagesToReplay: readonly Bytes[]) { + this.#detachDataHandler(); - constructor( - private readonly peripheral: Peripheral, - private readonly btpSession: BtpSessionHandler, - cleanupDataListener: () => void, - ) { - super(); - this.#cleanupDataListener = cleanupDataListener; - peripheral.once("disconnect", reason => { - logger.debug( - `Disconnected from peripheral ${peripheral.address} (reason ${nobleDisconnectReason(reason)}). Closing BTP session`, + logger.info( + `Peripheral ${this.peripheral.address}: Peer did not respond to any BTP packet, retrying with a ${MatterBle.MINIMUM_ATT_MTU} byte BTP segment size`, + ); + + // §4.19.3.3: unsubscribing from C2 closes the BTP session for the peripheral; the BLE connection is unaffected. + // A failure here means the peer still holds the old session, so the new handshake would be rejected anyway + await unsubscribeC2(this.peripheral, this.characteristicC2ForSubscribe, "throw"); + this.#assertRenegotiable(); + + const handshakeResponse = await performBtpHandshake( + this.peripheral, + this.characteristicC1ForWrite, + this.characteristicC2ForSubscribe, + MatterBle.MINIMUM_ATT_MTU, + this.#lifetime.signal, + ); + this.#assertRenegotiable(); + + await this.#adoptSession(handshakeResponse, MatterBle.MINIMUM_ATT_MTU); + this.#assertRenegotiable(); + + for (const message of messagesToReplay) { + await this.#session.sendMatterMessage(message); + } + } + + #assertRenegotiable() { + if (this.#closing || !this.connected) { + throw new BleDisconnectedError( + `Peripheral ${this.peripheral.address}: Channel was lost while renegotiating the BTP session`, ); - this.#connected = false; - this.#cleanupDataListener(); - this.#terminateIterator(); - for (const listener of this.#closeListeners) { - listener(); - } - this.btpSession.close().catch(error => { - logger.debug(`Peripheral ${peripheral.address}: Error closing BTP session on disconnect`, error); - }); - this.emitClosed(); - }); - // Forward BTP-initiated close (e.g. ack-receive timeout) to our Observable. - this.btpSession.closed.once(() => this.emitClosed()); + } } get connected() { @@ -708,17 +912,14 @@ export class NobleBleChannel extends BleChannel { * @param data */ async send(data: Bytes) { - if (!this.connected) { + // A renegotiation replaces the session, so sending into the outgoing one would be rejected as inactive + await this.#renegotiation; + if (this.#closing || !this.connected) { throw new BleDisconnectedError( `Peripheral ${this.peripheral.address}: Cannot send data because not connected to peripheral.`, ); } - if (this.btpSession === undefined) { - throw new BtpFlowError( - `Peripheral ${this.peripheral.address}: Cannot send data, no BTP session initialized`, - ); - } - await this.btpSession.sendMatterMessage(data); + await this.#session.sendMatterMessage(data); } // Channel @@ -771,10 +972,17 @@ export class NobleBleChannel extends BleChannel { } async close() { - this.#cleanupDataListener(); + // Connectivity is decided up front: the flags below make a parked send see the channel as gone, which would + // otherwise also suppress the disconnect this method owes the peripheral + const wasConnected = this.connected; + this.#closing = true; + this.#connected = false; + // Interrupts a renegotiation parked on the handshake, whose timer would otherwise outlive the channel + this.#lifetime.abort(); + this.#detachDataHandler(); this.#terminateIterator(); - await this.btpSession.close(); - if (this.connected) { + await this.#btpSession?.close(); + if (wasConnected && this.peripheral.state === "connected") { this.peripheral.disconnectAsync().catch(error => { if (!isNobleDisconnectError(error)) { logger.warn(`Peripheral ${this.peripheral.address}: Error while disconnecting`, error); diff --git a/packages/nodejs-ble/test/NobleBleChannelTest.ts b/packages/nodejs-ble/test/NobleBleChannelTest.ts index bdf3e581f4..50745ad132 100644 --- a/packages/nodejs-ble/test/NobleBleChannelTest.ts +++ b/packages/nodejs-ble/test/NobleBleChannelTest.ts @@ -5,7 +5,7 @@ */ import { asError, Bytes, MatterError, ServerAddress } from "@matter/general"; -import { BtpCodec, MatterBle } from "@matter/protocol"; +import { BleDisconnectedError, BtpCodec, MatterBle } from "@matter/protocol"; import type { Peripheral, PeripheralState, Service } from "@stoprocent/noble"; import { EventEmitter } from "node:events"; import type { BleScanner } from "../src/BleScanner.js"; @@ -28,7 +28,9 @@ const CONNECT_ERROR = "le-connection-abort-by-local"; */ class FakePeripheral extends EventEmitter { readonly address = PERIPHERAL_ADDRESS; - readonly mtu = null; + + /** noble reports null until the ATT_MTU exchange completes; tests covering that path set it back to null. */ + mtu: number | null = MatterBle.MAXIMUM_ATT_MTU; state: PeripheralState = "disconnected"; connectAttempts = 0; serviceDiscoveries = 0; @@ -203,6 +205,96 @@ function respondingMatterService(onSubscribe?: () => void) { }); } +/** + * Matter service that completes every handshake with the segment size it was offered but never answers a data packet, + * which is how a peripheral behaves whose link layer cannot carry a segment spread over several link-layer packets. + */ +function handshakeOnlyMatterService(onUnsubscribe?: () => void) { + const c2 = new EventEmitter(); + const handshakeSegmentSizes = new Array(); + const dataWrites = new Array(); + const dataWaiters = new Map void>(); + const handshakeWaiters = new Map void>(); + let unsubscribes = 0; + const state = { failUnsubscribe: false, withholdAfter: Infinity, onWithheld: () => {} }; + + const service = matterService({ + c1: { + uuid: nobleUuid(MatterBle.C1_CHARACTERISTIC_UUID), + properties: [], + async writeAsync(data: Buffer) { + const written = new Uint8Array(data); + if (BtpCodec.isHandshakeResponse(written) || written.length === 9) { + handshakeSegmentSizes.push(BtpCodec.decodeBtpHandshakeRequest(written).attMtu); + handshakeWaiters.get(handshakeSegmentSizes.length)?.(); + } else { + dataWrites.push(written); + dataWaiters.get(dataWrites.length)?.(); + } + }, + }, + c2: Object.assign(c2, { + uuid: nobleUuid(MatterBle.C2_CHARACTERISTIC_UUID), + properties: [], + async subscribeAsync() { + if (handshakeSegmentSizes.length > state.withholdAfter) { + queueMicrotask(() => state.onWithheld()); + return; + } + const attMtu = handshakeSegmentSizes[handshakeSegmentSizes.length - 1]; + const response = Buffer.from( + Bytes.of(BtpCodec.encodeBtpHandshakeResponse({ version: 4, attMtu, windowSize: 4 })), + ); + queueMicrotask(() => c2.emit("data", response, true)); + }, + async unsubscribeAsync() { + unsubscribes++; + onUnsubscribe?.(); + if (state.failUnsubscribe) { + throw new Error("Unsubscribe failed"); + } + }, + }), + }); + + return { + service, + handshakeSegmentSizes, + dataWrites, + set failUnsubscribe(fail: boolean) { + state.failUnsubscribe = fail; + }, + /** Leave every handshake past the given count unanswered, so the client waits out its handshake timer. */ + withholdHandshakeResponseAfter(count: number, onWithheld: () => void) { + state.withholdAfter = count; + state.onWithheld = onWithheld; + }, + /** Resolves once the given number of handshake requests have been written. */ + whenHandshake(count: number) { + return new Promise(resolve => { + if (handshakeSegmentSizes.length >= count) { + resolve(); + } else { + handshakeWaiters.set(count, resolve); + } + }); + }, + get unsubscribes() { + return unsubscribes; + }, + /** Resolves once the given number of BTP data packets have been written. */ + whenDataWrite(count: number) { + return new Promise(resolve => { + if (dataWrites.length >= count) { + resolve(); + } else { + dataWaiters.set(count, resolve); + } + }); + }, + }; +} + /** Wraps a signal so a test can observe the listeners a channel attempt registers on it. */ function observedSignal(controller: AbortController) { const listeners = new Set(); @@ -488,6 +580,231 @@ describe("NobleBleCentralInterface", () => { await central.close(); }); + it("derives the segment size from an ATT_MTU that arrives after the interview", async () => { + MockTime.init(); + const peripheral = new FakePeripheral(p => p.completeConnect()); + peripheral.mtu = null; + // The exchange lands only once someone waits for it + peripheral.on("newListener", event => { + if (event === "mtu") { + queueMicrotask(() => peripheral.emit("mtu", MatterBle.MAXIMUM_ATT_MTU)); + } + }); + const peer = handshakeOnlyMatterService(); + peripheral.services = [peer.service]; + const central = centralInterfaceFor(peripheral); + + const channel = await MockTime.resolve(central.openChannel(ADDRESS), { stepMs: 100 }); + + expect(peer.handshakeSegmentSizes).deep.equal([MatterBle.MAXIMUM_BTP_MTU]); + + await channel.close(); + await central.close(); + }); + + it("falls back to the minimum segment size when the ATT_MTU exchange never completes", async () => { + MockTime.init(); + const peripheral = new FakePeripheral(p => p.completeConnect()); + peripheral.mtu = null; + const peer = handshakeOnlyMatterService(); + peripheral.services = [peer.service]; + const central = centralInterfaceFor(peripheral); + + const channel = await MockTime.resolve(central.openChannel(ADDRESS), { stepMs: 500 }); + + expect(peer.handshakeSegmentSizes).deep.equal([MatterBle.MINIMUM_ATT_MTU]); + + await channel.close(); + await central.close(); + }); + + it("renegotiates the BTP session with the minimum segment size when the peer answers no data packet", async () => { + MockTime.init(); + const peripheral = new FakePeripheral(p => p.completeConnect()); + peripheral.mtu = MatterBle.MAXIMUM_ATT_MTU; + const peer = handshakeOnlyMatterService(); + peripheral.services = [peer.service]; + const central = centralInterfaceFor(peripheral); + + const channel = await central.openChannel(ADDRESS); + expect(peer.handshakeSegmentSizes).deep.equal([MatterBle.MAXIMUM_BTP_MTU]); + + // One segment at 244 bytes, three at 20, so the replay only reassembles if the size actually dropped + const message = Bytes.fromHex("a1".repeat(50)); + await channel.send(message); + expect(peer.dataWrites.length).equal(1); + + await MockTime.resolve(peer.whenDataWrite(4), { stepMs: 1000 }); + + expect(peer.handshakeSegmentSizes).deep.equal([MatterBle.MAXIMUM_BTP_MTU, MatterBle.MINIMUM_ATT_MTU]); + expect(peer.unsubscribes).equal(1); + expect(peripheral.state).equals("connected"); + + const replay = peer.dataWrites.slice(1); + expect(replay.length).equal(3); + let reassembled: Bytes = new Uint8Array(0); + for (const packet of replay) { + expect(packet.byteLength).most(MatterBle.MINIMUM_ATT_MTU); + reassembled = Bytes.concat(reassembled, BtpCodec.decodeBtpPacket(packet).payload.segmentPayload); + } + expect(reassembled).deep.equal(message); + + await channel.close(); + await central.close(); + }); + + it("closes the channel when the renegotiation cannot close the peer's BTP session", async () => { + MockTime.init(); + const peripheral = new FakePeripheral(p => p.completeConnect()); + peripheral.mtu = MatterBle.MAXIMUM_ATT_MTU; + const peer = handshakeOnlyMatterService(); + peer.failUnsubscribe = true; + peripheral.services = [peer.service]; + const central = centralInterfaceFor(peripheral); + + const channel = await central.openChannel(ADDRESS); + const disconnected = peripheral.whenDisconnected(); + await channel.send(Bytes.fromHex("00112233445566778899")); + + await MockTime.resolve(disconnected, { stepMs: 1000 }); + + expect(peer.handshakeSegmentSizes).deep.equal([MatterBle.MAXIMUM_BTP_MTU]); + expect(peer.dataWrites.length).equal(1); + + await central.close(); + }); + + it("closes the channel when the renegotiated session is not answered either", async () => { + MockTime.init(); + const peripheral = new FakePeripheral(p => p.completeConnect()); + const peer = handshakeOnlyMatterService(); + peripheral.services = [peer.service]; + const central = centralInterfaceFor(peripheral); + + const channel = await central.openChannel(ADDRESS); + const disconnected = peripheral.whenDisconnected(); + await channel.send(Bytes.fromHex("00112233445566778899")); + + await MockTime.resolve(disconnected, { stepMs: 1000 }); + + // One renegotiation, then the peer is given up on rather than retried forever + expect(peer.handshakeSegmentSizes).deep.equal([MatterBle.MAXIMUM_BTP_MTU, MatterBle.MINIMUM_ATT_MTU]); + + await central.close(); + }); + + it("reports channel loss to a send parked on a renegotiation that fails", async () => { + MockTime.init(); + const peripheral = new FakePeripheral(p => p.completeConnect()); + + let parked: Promise | undefined; + const peer = handshakeOnlyMatterService(() => { + parked ??= channel.send(Bytes.fromHex("aabb")).then( + () => undefined, + (error: unknown) => error, + ); + }); + peer.failUnsubscribe = true; + peripheral.services = [peer.service]; + const central = centralInterfaceFor(peripheral); + + const channel = await central.openChannel(ADDRESS); + const disconnected = peripheral.whenDisconnected(); + await channel.send(Bytes.fromHex("00112233445566778899")); + + const failure = await MockTime.resolve( + disconnected.then(() => parked), + { stepMs: 1000 }, + ); + + // BtpFlowError from the suspended session would not read as channel loss downstream + expect(failure).instanceOf(BleDisconnectedError); + + await central.close(); + }); + + it("abandons a renegotiation when the channel is closed while it waits for the handshake", async () => { + MockTime.init(); + const peripheral = new FakePeripheral(p => p.completeConnect()); + + let closing: Promise | undefined; + const peer = handshakeOnlyMatterService(); + // The renegotiation parks here: the peer never answers the second handshake + peer.withholdHandshakeResponseAfter(1, () => { + closing ??= channel.close(); + }); + peripheral.services = [peer.service]; + const central = centralInterfaceFor(peripheral); + + const channel = await central.openChannel(ADDRESS); + await channel.send(Bytes.fromHex("00112233445566778899")); + + await MockTime.resolve( + MockTime.resolve(peer.whenHandshake(2), { stepMs: 1000 }).then(() => closing), + { stepMs: 1000 }, + ); + + // Without the abort the handshake timer would still be armed for its full BTP_CONN_RSP_TIMEOUT + expect(MockTime.timerCountFor("BLE handshake timeout")).equal(0); + + await central.close(); + }); + + it("abandons a renegotiation when the peripheral disconnects while it waits for the handshake", async () => { + MockTime.init(); + const peripheral = new FakePeripheral(p => p.completeConnect()); + + let parked: Promise | undefined; + const peer = handshakeOnlyMatterService(); + peer.withholdHandshakeResponseAfter(1, () => { + parked ??= channel.send(Bytes.fromHex("aabb")).then( + () => undefined, + (error: unknown) => error, + ); + peripheral.dropConnection(undefined); + }); + peripheral.services = [peer.service]; + const central = centralInterfaceFor(peripheral); + + const channel = await central.openChannel(ADDRESS); + await channel.send(Bytes.fromHex("00112233445566778899")); + + await MockTime.resolve(peer.whenHandshake(2), { stepMs: 1000 }); + + // No further time passes: the disconnect must settle this, not the handshake's own timeout + await MockTime.yield3(); + expect(MockTime.timerCountFor("BLE handshake timeout")).equal(0); + expect(await parked).instanceOf(BleDisconnectedError); + + await central.close(); + }); + + it("holds a send issued while the BTP session is being renegotiated", async () => { + MockTime.init(); + const peripheral = new FakePeripheral(p => p.completeConnect()); + peripheral.mtu = MatterBle.MAXIMUM_ATT_MTU; + + let sendDuringRenegotiation: Promise | undefined; + const during = Bytes.fromHex("aabbccdd"); + const peer = handshakeOnlyMatterService(() => { + sendDuringRenegotiation ??= channel.send(during); + }); + peripheral.services = [peer.service]; + const central = centralInterfaceFor(peripheral); + + const channel = await central.openChannel(ADDRESS); + await channel.send(Bytes.fromHex("00112233445566778899")); + + await MockTime.resolve(peer.whenDataWrite(3), { stepMs: 1000 }); + await sendDuringRenegotiation; + + // The replay goes out first, then the message queued while the session was gone + expect(Bytes.toHex(peer.dataWrites[2]).endsWith(Bytes.toHex(during))).equal(true); + + await channel.close(); + await central.close(); + }); + it("bounds a disconnect that never completes after a failed channel setup", async () => { MockTime.init(); const peripheral = new FakePeripheral(p => p.completeConnect()); diff --git a/packages/protocol/src/ble/BtpSessionHandler.ts b/packages/protocol/src/ble/BtpSessionHandler.ts index c92a2fce46..3dd4dfbcf3 100644 --- a/packages/protocol/src/ble/BtpSessionHandler.ts +++ b/packages/protocol/src/ble/BtpSessionHandler.ts @@ -39,12 +39,37 @@ export class BtpSessionHandler { await this.close(); }); readonly #closed = Observable<[]>(); + readonly #stalledAfterHandshake = Observable<[messagesToReplay: readonly Bytes[]]>(); + + /** + * Matter messages handed to {@link sendMatterMessage} while the peer has not acknowledged anything yet. Dropped as + * soon as any BTP packet arrives, because from then on {@link stalledAfterHandshake} can no longer fire, and the + * ack timeout bounds how much can accumulate. + */ + #messagesPendingFirstAck: Bytes[] | undefined = new Array(); /** Emitted exactly once when the session transitions to closed. */ get closed() { return this.#closed; } + /** + * Emitted instead of {@link closed} when the peer completed the handshake and then never responded to any data + * packet, carrying the Matter messages the peer never acknowledged in submission order. + * + * A peripheral whose link layer cannot carry a BTP segment spanning several link-layer packets fails exactly this + * way. Recovering from it is an interop workaround with no basis in the specification: the transport may establish + * a fresh session with a smaller segment size and resend the messages, which the specification neither describes + * nor forbids. + * + * The session is suspended when this fires: it no longer touches the transport, and the transport owns what happens + * next. A session nobody observes closes on the acknowledgement timeout as it always did, so a transport that + * cannot renegotiate needs no changes. + */ + get stalledAfterHandshake() { + return this.#stalledAfterHandshake; + } + /** Factory method to create a new BTPSessionHandler from a received handshake request */ static async createFromHandshakeRequest( maxDataSize: number | undefined, @@ -118,18 +143,24 @@ export class BtpSessionHandler { return btpSession; } + /** + * @param requestedSegmentSize The segment size offered in our handshake request. A peripheral must not select more + * than we offered, so it also bounds the session; without it a peripheral that echoes an oversized value would + * defeat a deliberate reduction of the segment size. + */ static async createAsCentral( handshakeResponsePayload: Bytes, writeBleCallback: (data: Bytes) => Promise, disconnectBleCallback: () => Promise, handleMatterMessagePayload: (data: Bytes) => Promise, + requestedSegmentSize: number, ) { const handshakeRequest = BtpCodec.decodeBtpHandshakeResponsePayload(handshakeResponsePayload); logger.debug("Handshake request", Diagnostic.dict(handshakeRequest)); const { version, attMtu: handshakeMtu, windowSize } = handshakeRequest; - const fragmentSize = Math.min(handshakeMtu, MatterBle.MAXIMUM_BTP_MTU); + const fragmentSize = Math.min(handshakeMtu, requestedSegmentSize, MatterBle.MAXIMUM_BTP_MTU); return new BtpSessionHandler( "central", @@ -229,6 +260,7 @@ export class BtpSessionHandler { throw new BtpProtocolError("Expected and actual BTP packets sequence number does not match"); } this.prevIncomingSequenceNumber = sequenceNumber; + this.#messagesPendingFirstAck = undefined; if (!this.sendAckTimer.isRunning) { this.sendAckTimer.start(); @@ -336,6 +368,7 @@ export class BtpSessionHandler { throw new BtpFlowError("BTP packet must not be empty"); } const dataReader = new DataReader(data, Endian.Little); + this.#messagesPendingFirstAck?.push(data); this.queuedOutgoingMatterMessages.push(dataReader); await this.processSendQueue(); } @@ -436,6 +469,14 @@ export class BtpSessionHandler { return; } + if (!this.isActive) { + // A suspend during the write hands the transport to someone else; further segments of this session + // would reach a peer that has already renegotiated + this.queuedOutgoingMatterMessages.length = 0; + this.sendInProgress = false; + return; + } + if (!this.ackReceiveTimer.isRunning) { this.ackReceiveTimer.start(); // starts the timer } @@ -464,9 +505,7 @@ export class BtpSessionHandler { * Close the BTP session. This method is called when the BLE transport is disconnected and so the BTP session gets closed. */ public async close() { - this.sendAckTimer.stop(); - this.ackReceiveTimer.stop(); - this.idleTimeout.stop(); + this.#stopTimers(); if (this.isActive) { logger.debug(`Closing BTP session`); this.isActive = false; @@ -480,6 +519,21 @@ export class BtpSessionHandler { } } + /** + * End the session without touching the transport, so the same BLE connection can carry a renegotiated session. + * Unlike {@link close} this emits neither {@link closed} nor a disconnect. + */ + suspend() { + this.#stopTimers(); + this.isActive = false; + } + + #stopTimers() { + this.sendAckTimer.stop(); + this.ackReceiveTimer.stop(); + this.idleTimeout.stop(); + } + /** * If this timer expires and the peer has a pending acknowledgement, the peer SHALL immediately send that * acknowledgement @@ -558,10 +612,33 @@ export class BtpSessionHandler { * the peer SHALL close the BTP session and report an error to the application. */ private async btpAckTimeoutTriggered() { - if (this.prevIncomingAckNumber !== this.sequenceNumber) { - logger.warn("Acknowledgement for the sent sequence number was not received ... disconnect"); - await this.close(); + if (!this.isActive || this.prevIncomingAckNumber === this.sequenceNumber) { + return; } + if ( + this.#messagesPendingFirstAck !== undefined && + this.#stalledAfterHandshake.isObserved && + this.#isStalledAfterHandshake() + ) { + logger.warn( + `No BTP response at all since the handshake with a segment size of ${this.fragmentSize} bytes ... renegotiate`, + ); + const messagesToReplay = this.#messagesPendingFirstAck; + this.#messagesPendingFirstAck = undefined; + this.suspend(); + this.#stalledAfterHandshake.emit(messagesToReplay); + return; + } + logger.warn("Acknowledgement for the sent sequence number was not received ... disconnect"); + await this.close(); + } + + /** + * True when we sent data and the peer has since sent nothing at all — neither an acknowledgement nor a packet of + * its own. A smaller segment size is the only lever left, so a session already at the minimum is excluded. + */ + #isStalledAfterHandshake() { + return this.role === "central" && this.fragmentSize > MatterBle.MINIMUM_ATT_MTU; } /** diff --git a/packages/protocol/src/codec/BtpCodec.ts b/packages/protocol/src/codec/BtpCodec.ts index 11ec842cfe..244f60dd2f 100644 --- a/packages/protocol/src/codec/BtpCodec.ts +++ b/packages/protocol/src/codec/BtpCodec.ts @@ -65,7 +65,23 @@ export enum BtpOpcode { const HANDSHAKE_HEADER = 0b01100101; +/** Header, opcode, version, 16-bit ATT_MTU and window size. */ +const HANDSHAKE_RESPONSE_LENGTH = 6; + export class BtpCodec { + /** + * Whether the payload is a BTP handshake response, so a client can tell it apart from data arriving on C2 before it + * has a session to decode against. + */ + static isHandshakeResponse(data: Bytes) { + const bytes = Bytes.of(data); + return ( + bytes.length === HANDSHAKE_RESPONSE_LENGTH && + bytes[0] === HANDSHAKE_HEADER && + bytes[1] === BtpOpcode.HandshakeManagementOpcode + ); + } + static decodeBtpHandshakeRequest(data: Bytes): BtpHandshakeRequest { const reader = new DataReader(data, Endian.Little); return this.decodeHandshakeRequestPayload(reader); diff --git a/packages/protocol/test/ble/BtpSessionHandlerTest.ts b/packages/protocol/test/ble/BtpSessionHandlerTest.ts index 8348a83868..57c63663b3 100644 --- a/packages/protocol/test/ble/BtpSessionHandlerTest.ts +++ b/packages/protocol/test/ble/BtpSessionHandlerTest.ts @@ -764,6 +764,7 @@ describe("BtpSessionHandler", () => { async () => { throw new Error("Should not be called"); }, + MatterBle.MAXIMUM_BTP_MTU, ); for (let i = 0; i < 6; i++) { @@ -866,6 +867,7 @@ describe("BtpSessionHandler", () => { async () => { throw new Error("Should not be called"); }, + MatterBle.MAXIMUM_BTP_MTU, ); let raised: unknown; @@ -1067,6 +1069,7 @@ describe("BtpSessionHandler", () => { async () => { throw new Error("Should not be called"); }, + MatterBle.MAXIMUM_BTP_MTU, ); // Send enough single-fragment messages that the central's sequence number wraps past 255. @@ -1121,6 +1124,7 @@ describe("BtpSessionHandler", () => { async () => { throw new Error("Should not be called"); }, + MatterBle.MAXIMUM_BTP_MTU, ); const message = Bytes.fromHex("aabbccdd".repeat(50)); // 200 bytes, far exceeds window * fragment size @@ -1133,6 +1137,198 @@ describe("BtpSessionHandler", () => { await peripheral.close(); }); + describe("Test stalled peer detection", () => { + function centralFor( + attMtu: number, + written: Bytes[], + disconnected: { value: boolean }, + requestedSegmentSize = MatterBle.MAXIMUM_BTP_MTU, + ) { + return BtpSessionHandler.createAsCentral( + BtpCodec.encodeBtpHandshakeResponse({ version: 4, attMtu, windowSize: 8 }), + async data => { + written.push(data); + }, + async () => { + disconnected.value = true; + }, + async () => { + throw new Error("Should not be called"); + }, + requestedSegmentSize, + ); + } + + it("reports a stall instead of closing when the peer never answers a data packet", async () => { + const written = new Array(); + const disconnected = { value: false }; + const central = await centralFor(244, written, disconnected); + + let stalled = 0; + let closed = 0; + let replayed: readonly Bytes[] | undefined; + central.stalledAfterHandshake.on(messagesToReplay => { + stalled++; + replayed = messagesToReplay; + }); + central.closed.on(() => { + closed++; + }); + + const first = Bytes.fromHex("0102030405"); + const second = Bytes.fromHex("060708"); + await central.sendMatterMessage(first); + await central.sendMatterMessage(second); + expect(written.length).equal(2); + + await MockTime.advance(15000); + + expect(stalled).equal(1); + expect(closed).equal(0); + expect(disconnected.value).equal(false); + expect(replayed).deep.equal([first, second]); + }); + + it("closes when nobody observes the stall", async () => { + const written = new Array(); + const disconnected = { value: false }; + const central = await centralFor(244, written, disconnected); + + const { promise: closedPromise, resolver: closedResolver } = createPromise(); + central.closed.on(() => closedResolver()); + + await central.sendMatterMessage(Bytes.fromHex("0102030405")); + await MockTime.advance(15000); + await closedPromise; + + // A transport that cannot renegotiate must still see the session close + expect(disconnected.value).equal(true); + }); + + it("closes when a session already at the minimum segment size is not answered", async () => { + const written = new Array(); + const disconnected = { value: false }; + const central = await centralFor(MatterBle.MINIMUM_ATT_MTU, written, disconnected); + + let stalled = 0; + central.stalledAfterHandshake.on(() => { + stalled++; + }); + const { promise: closedPromise, resolver: closedResolver } = createPromise(); + central.closed.on(() => closedResolver()); + + await central.sendMatterMessage(Bytes.fromHex("0102030405")); + await MockTime.advance(15000); + await closedPromise; + + expect(stalled).equal(0); + expect(disconnected.value).equal(true); + }); + + it("closes once the peer has sent anything at all", async () => { + const written = new Array(); + const disconnected = { value: false }; + const central = await BtpSessionHandler.createAsCentral( + BtpCodec.encodeBtpHandshakeResponse({ version: 4, attMtu: 244, windowSize: 8 }), + async data => { + written.push(data); + }, + async () => { + disconnected.value = true; + }, + async () => {}, + MatterBle.MAXIMUM_BTP_MTU, + ); + + let stalled = 0; + let replayed: readonly Bytes[] | undefined; + central.stalledAfterHandshake.on(messagesToReplay => { + stalled++; + replayed = messagesToReplay; + }); + const { promise: closedPromise, resolver: closedResolver } = createPromise(); + central.closed.on(() => closedResolver()); + + await central.sendMatterMessage(Bytes.fromHex("0102030405")); + + const segmentPayload = Bytes.fromHex("aabb"); + await central.handleIncomingBleData( + BtpCodec.encodeBtpPacket({ + header: { + isHandshakeRequest: false, + hasManagementOpcode: false, + hasAckNumber: false, + isBeginningSegment: true, + isContinuingSegment: false, + isEndingSegment: true, + }, + payload: { + ackNumber: undefined, + sequenceNumber: 1, + messageLength: segmentPayload.byteLength, + segmentPayload, + }, + }), + ); + + await MockTime.advance(15000); + await closedPromise; + + expect(stalled).equal(0); + expect(replayed).equal(undefined); + }); + + it("stops writing the remaining fragments of a suspended session", async () => { + const written = new Array(); + let releaseWrite: (() => void) | undefined; + const central = await BtpSessionHandler.createAsCentral( + BtpCodec.encodeBtpHandshakeResponse({ version: 4, attMtu: 244, windowSize: 8 }), + async data => { + written.push(data); + if (written.length === 2) { + await new Promise(resolve => (releaseWrite = resolve)); + } + }, + async () => {}, + async () => {}, + MatterBle.MAXIMUM_BTP_MTU, + ); + + let stalled = 0; + central.stalledAfterHandshake.on(() => { + stalled++; + }); + + // The first message starts the acknowledgement timer; the second stalls mid-write while it runs out + await central.sendMatterMessage(Bytes.fromHex("0102030405")); + const sending = central.sendMatterMessage(Bytes.fromHex("bb".repeat(500))); + expect(written.length).equal(2); + + await MockTime.advance(15000); + expect(stalled).equal(1); + + releaseWrite?.(); + await sending; + + expect(written.length).equal(2); + }); + + it("keeps the segment size we offered when the peer answers with a larger one", async () => { + const written = new Array(); + const disconnected = { value: false }; + const central = await centralFor(244, written, disconnected, MatterBle.MINIMUM_ATT_MTU); + + await central.sendMatterMessage(Bytes.fromHex("aa".repeat(60))); + + expect(written.length).greaterThan(1); + for (const packet of written) { + expect(packet.byteLength).most(MatterBle.MINIMUM_ATT_MTU); + } + + await central.close(); + }); + }); + describe("Test ATT_MTU fragment derivation", () => { // §4.19.3.1.4: the BTP segment size is the negotiated ATT_MTU minus the 3-byte GATT header, // clamped to the supported range. diff --git a/packages/react-native/src/ble/ReactNativeBleChannel.ts b/packages/react-native/src/ble/ReactNativeBleChannel.ts index eb28b77719..c0d8632bb5 100644 --- a/packages/react-native/src/ble/ReactNativeBleChannel.ts +++ b/packages/react-native/src/ble/ReactNativeBleChannel.ts @@ -269,6 +269,7 @@ export class ReactNativeBleChannel extends BleChannel { bleChannel.#pushMessage(data); onMatterMessageListener(bleChannel, data); }, + mtu, ); const bleChannel = new ReactNativeBleChannel(peripheral, btpSession);