feat(ble-proxy): serialize BLE connects + structured slot-exhaustion errors

docs/ble-proxy-protocol.md

@@ -223,12 +223,14 @@ connections. It is used in all subsequent commands and binary frames for this co
 
 **Errors:**
 
-| Error Code          | Description                                 |
-|---------------------|---------------------------------------------|
-| `device_not_found`  | Peripheral not reachable or not advertising |
-| `connection_failed` | Connection attempt failed                   |
-| `already_connected` | Already connected to this address           |
-| `timeout`           | Connection timed out                        |
+| Error Code                 | Description                                              |
+|----------------------------|----------------------------------------------------------|
+| `device_not_found`         | Peripheral not reachable or not advertising              |
+| `connection_failed`        | Connection attempt failed                                |
+| `out_of_connection_slots`  | Proxy backend has no free connection slot for this address |
+| `connection_aborted`       | Connection aborted (interference/range, peer/local terminate) |
+| `already_connected`        | Already connected to this address                        |
+| `timeout`                  | Connection timed out                                     |
 
 ---
 
@@ -711,6 +713,8 @@ sequenceDiagram
 | `not_scanning`             | No scan is currently active                     |
 | `device_not_found`         | Peripheral not reachable or not advertising     |
 | `connection_failed`        | BLE connection attempt failed                   |
+| `out_of_connection_slots`  | Proxy backend out of connection slots; add proxies or raise connection_slots |
+| `connection_aborted`       | Connection aborted (interference/range, peer/local terminate)                 |
 | `already_connected`        | Already connected to this peripheral            |
 | `not_connected`            | No active connection with this handle           |
 | `timeout`                  | Operation timed out                             |

packages/ble-proxy/src/BleProxyHandler.ts

@@ -10,6 +10,7 @@ import type { createServer } from "node:http";
 import { WebSocket, WebSocketServer } from "ws";
 import {
     BLE_PROXY_PROTOCOL_VERSION,
+    BleProxyError,
     decodeBinaryFrame,
     encodeBinaryFrame,
     type BinaryFrame,
@@ -317,7 +318,7 @@ export class BleProxyHandler implements WebServerHandler {
         if (msg.success) {
             pending.resolver(msg.result);
         } else {
-            pending.rejecter(new Error(`${msg.error}: ${msg.message}`));
+            pending.rejecter(new BleProxyError(msg.error, msg.message));
         }
     }
 

packages/ble-proxy/src/BleProxyProtocol.ts

@@ -249,10 +249,41 @@ export const BleProxyErrorCode = {
     MtuRequestFailed: "mtu_request_failed",
     DiscoveryFailed: "discovery_failed",
     InternalError: "internal_error",
+    OutOfConnectionSlots: "out_of_connection_slots",
+    ConnectionAborted: "connection_aborted",
 } as const;
 
 export type BleProxyErrorCodeValue = (typeof BleProxyErrorCode)[keyof typeof BleProxyErrorCode];
 
+/** Substring markers sourced from bleak_retry_connector's `OUT_OF_SLOTS_ERRORS`. */
+export const OUT_OF_SLOTS_MESSAGE_MARKERS = [
+    "available connection",
+    "connection slot",
+    "ESP_GATT_CONN_CONN_CANCEL",
+] as const;
+
+/** True when a connect failure means the proxy backend ran out of connection slots. */
+export function isOutOfConnectionSlotsError(code: string | undefined, message: string | undefined): boolean {
+    if (code === BleProxyErrorCode.OutOfConnectionSlots) {
+        return true;
+    }
+    if (code === BleProxyErrorCode.ConnectionFailed && message) {
+        const lower = message.toLowerCase();
+        return OUT_OF_SLOTS_MESSAGE_MARKERS.some(marker => lower.includes(marker.toLowerCase()));
+    }
+    return false;
+}
+
+/** Error carrying the structured proxy-client error code alongside a readable message. */
+export class BleProxyError extends Error {
+    readonly code: string;
+    constructor(code: string, detail: string) {
+        super(`${code}: ${detail}`);
+        this.name = "BleProxyError";
+        this.code = code;
+    }
+}
+
 // ─── Binary Frame Opcodes ────────────────────────────────────────────────────
 
 export const BinaryFrameOpcode = {

packages/ble-proxy/src/ProxyBle.ts

@@ -54,6 +54,10 @@ export class ProxyBle extends Ble {
         return this.#bleScanner;
     }
 
+    abortPendingConnects(): void {
+        this.#bleCentralInterface?.abortPendingConnects();
+    }
+
     async close(): Promise<void> {
         if (this.#closed) return;
         this.#closed = true;

packages/ble-proxy/src/ProxyBleChannel.ts

@@ -5,20 +5,30 @@
  */
 
 import {
+    Abort,
     type Bytes,
     type Channel,
     ChannelType,
     createPromise,
     InternalError,
     Logger,
     NetworkError,
+    Seconds,
+    Semaphore,
     ServerAddress,
     Time,
     type Transport,
+    type WorkSlot,
 } from "@matter/main";
 import { BleChannel, BleError, BtpCodec, BtpFlowError, BtpSessionHandler, MatterBle } from "@matter/main/protocol";
 import type { BleProxyHandler } from "./BleProxyHandler.js";
-import { BinaryFrameOpcode, BleProxyCommand, BleProxyEvent } from "./BleProxyProtocol.js";
+import {
+    BinaryFrameOpcode,
+    BleProxyCommand,
+    BleProxyError,
+    BleProxyEvent,
+    isOutOfConnectionSlotsError,
+} from "./BleProxyProtocol.js";
 import type { ProxyBleScanner } from "./ProxyBleScanner.js";
 
 const logger = Logger.get("ProxyBleChannel");
@@ -28,6 +38,9 @@ const BTP_HANDSHAKE_RESPONSE_OPCODE_1 = 0x65;
 const BTP_HANDSHAKE_RESPONSE_OPCODE_2 = 0x6c;
 const BTP_HANDSHAKE_RESPONSE_LENGTH = 6;
 
+/** Safety-net bound on waiting for a connection slot. */
+const MAX_QUEUE_WAIT = Seconds(20);
+
 /**
  * Normalize any UUID form sent by a proxy client to the canonical dashed-uppercase form used by
  * Matter's MatterBle constants. Different BLE proxy clients deliver different formats:
@@ -60,13 +73,19 @@ export class ProxyBleCentralInterface implements Transport {
     readonly #handler: BleProxyHandler;
     #onMatterMessageListener: ((socket: Channel<Bytes>, data: Bytes) => void) | undefined;
     #closed = false;
+    readonly #connectSemaphore = new Semaphore("ble-proxy-connect", 1);
+    readonly #closeAbort = new Abort();
+    /** Channels the gate has handed out and that are still open. */
+    readonly #openChannels = new Set<ProxyBleChannel>();
+    /** Bumped by abortPendingConnects/close so a connect that completes afterwards tears itself down. */
+    #abortGeneration = 0;
 
     constructor(bleScanner: ProxyBleScanner, handler: BleProxyHandler) {
         this.#bleScanner = bleScanner;
         this.#handler = handler;
     }
 
-    async openChannel(address: ServerAddress): Promise<Channel<Bytes>> {
+    async openChannel(address: ServerAddress, options?: Transport.OpenChannelOptions): Promise<Channel<Bytes>> {
         if (this.#closed) {
             throw new NetworkError("Network interface is closed");
         }
@@ -84,12 +103,46 @@ export class ProxyBleCentralInterface implements Transport {
             throw new BleError("BLE proxy client not connected");
         }
 
+        // Serialize BLE connects: a single physical device advertises rotating MAC
+        // addresses, and matter.js fires one connect per address. Without this gate the
+        // proxy backend's connection slots are exhausted and commissioning fails.
+        const waitAbort = new Abort({ abort: [this.#closeAbort.signal, options?.abort], timeout: MAX_QUEUE_WAIT });
+        let slot: WorkSlot;
+        try {
+            slot = await this.#connectSemaphore.obtainSlot(waitAbort.signal);
+        } catch {
+            throw new BleError(`BLE connect to ${peripheralAddress} aborted before a connection slot was available`);
+        } finally {
+            waitAbort.close();
+        }
+
+        const abortGenAtStart = this.#abortGeneration;
+        let slotReleased = false;
+        const releaseSlot = () => {
+            if (!slotReleased) {
+                slotReleased = true;
+                slot.close();
+            }
+        };
+
         logger.debug(`Connecting to peripheral ${peripheralAddress} via proxy`);
 
         // 1. Connect
-        const { connection_handle, mtu: peripheralMtu } = await this.#handler.sendCommand(BleProxyCommand.Connect, {
-            address: peripheralAddress,
-        });
+        let connection_handle: number;
+        let peripheralMtu: number | undefined;
+        try {
+            ({ connection_handle, mtu: peripheralMtu } = await this.#handler.sendCommand(BleProxyCommand.Connect, {
+                address: peripheralAddress,
+            }));
+        } catch (error) {
+            releaseSlot();
+            if (error instanceof BleProxyError && isOutOfConnectionSlotsError(error.code, error.message)) {
+                throw new BleError(
+                    `BLE proxy backend is out of connection slots for ${peripheralAddress} — increase connection_slots on the proxy or add more proxies`,
+                );
+            }
+            throw error;
+        }
 
         let mtu = peripheralMtu ?? 0;
         if (mtu > MatterBle.MAXIMUM_BTP_MTU) {
@@ -269,8 +322,20 @@ export class ProxyBleCentralInterface implements Transport {
                 this.#handler.eventReceived.off(eventObserver);
             };
 
-            const proxyChannel = new ProxyBleChannel(peripheralAddress, btpSession, cleanupObservers);
+            const proxyChannel = new ProxyBleChannel(peripheralAddress, btpSession, cleanupObservers, releaseSlot);
             channelRef.channel = proxyChannel;
+
+            // If the gate was aborted (commissioning ended) or the interface closed while we were
+            // connecting, matter.js has already abandoned this attempt via its own abort. Returning
+            // the channel would orphan it — nobody closes it, so it would hold the single connection
+            // slot forever. Tear it down instead; the catch below sends the proxy Disconnect.
+            if (this.#closed || this.#abortGeneration !== abortGenAtStart) {
+                await proxyChannel.close();
+                throw new BleError(`BLE connect to ${peripheralAddress} aborted before completion`);
+            }
+
+            this.#openChannels.add(proxyChannel);
+            proxyChannel.onClose(() => this.#openChannels.delete(proxyChannel));
             return proxyChannel;
         } catch (error) {
             // Clean up on failure — best-effort tear-down of the peripheral on the proxy side
@@ -279,6 +344,7 @@ export class ProxyBleCentralInterface implements Transport {
             } catch (cleanupError) {
                 logger.debug(`Peripheral ${peripheralAddress}: Error during connect-failure cleanup`, cleanupError);
             }
+            releaseSlot();
             throw error;
         }
     }
@@ -290,8 +356,27 @@ export class ProxyBleCentralInterface implements Transport {
         };
     }
 
+    /**
+     * Release the connect gate when a commissioning attempt ends. BLE is only used during
+     * commissioning, so afterwards any queued connect is stale and any still-open channel is an
+     * orphan holding the connection slot. Rejects queued waiters, tears down open channels, and
+     * bumps the generation so an in-flight connect that completes later tears itself down too.
+     */
+    abortPendingConnects(): void {
+        this.#abortGeneration++;
+        this.#connectSemaphore.clear();
+        for (const channel of [...this.#openChannels]) {
+            channel.close().catch(error => logger.debug(`Error closing channel during connect abort`, error));
+        }
+    }
+
     async close() {
         this.#closed = true;
+        this.#closeAbort.abort();
+        this.#connectSemaphore.close();
+        for (const channel of [...this.#openChannels]) {
+            await channel.close().catch(error => logger.debug(`Error closing channel during interface close`, error));
+        }
     }
 
     supports(type: ChannelType, _address?: string) {
@@ -308,17 +393,24 @@ export class ProxyBleChannel extends BleChannel<Bytes> {
     readonly #peripheralAddress: string;
     readonly #btpSession: BtpSessionHandler;
     readonly #cleanupObservers: () => void;
+    readonly #releaseSlot: () => void;
     readonly #onBtpSessionClosed: () => void;
     readonly #closeListeners = new Set<() => void>();
     #iteratorQueue = new Array<Bytes>();
     #iteratorWaiter?: (value: IteratorResult<Bytes>) => void;
     #iteratorDone = false;
 
-    constructor(peripheralAddress: string, btpSession: BtpSessionHandler, cleanupObservers: () => void) {
+    constructor(
+        peripheralAddress: string,
+        btpSession: BtpSessionHandler,
+        cleanupObservers: () => void,
+        releaseSlot: () => void,
+    ) {
         super();
         this.#peripheralAddress = peripheralAddress;
         this.#btpSession = btpSession;
         this.#cleanupObservers = cleanupObservers;
+        this.#releaseSlot = releaseSlot;
         this.#onBtpSessionClosed = () => this.emitClosed();
         btpSession.closed.on(this.#onBtpSessionClosed);
     }
@@ -391,6 +483,7 @@ export class ProxyBleChannel extends BleChannel<Bytes> {
 
     async close() {
         this.#connected = false;
+        this.#releaseSlot();
         this.#cleanupObservers();
         this.#terminateIterator();
         for (const listener of this.#closeListeners) {