Limits the parallel open exchanges to 5 (#1192)

CHANGELOG.md

@@ -46,6 +46,9 @@ The main work (all changes without a GitHub username in brackets in the below li
 -   @matter.js/nodejs-ble
     -   The BLE specialization for Node.js is moved here.  `@project-chip/matter-node-ble.js` remains as a compatibility import.
 
+-   matter.js-protocol:
+    -   Limits the number of parallel exchanges to 5
+
 -   @matter.js/main:
     -   This package is a new "one-and-done" dependency for applications.  It automatically loads platform specialization and reexports pacakages above as appropriate
 

packages/protocol/src/protocol/ExchangeManager.ts

@@ -31,6 +31,8 @@ import { ProtocolHandler } from "./ProtocolHandler.js";
 
 const logger = Logger.get("ExchangeManager");
 
+const MAXIMUM_CONCURRENT_EXCHANGES_PER_SESSION = 5;
+
 export class ChannelNotConnectedError extends MatterError {}
 
 export class MessageChannel implements Channel<Message> {
@@ -283,9 +285,6 @@ export class ExchangeManager {
                     );
                 }
             }
-
-            // TODO A node SHOULD limit itself to a maximum of 5 concurrent exchanges over a unicast session. This is
-            //  to prevent a node from exhausting the message counter window of the peer node.
         }
     }
 
@@ -357,6 +356,24 @@ export class ExchangeManager {
     #addExchange(exchangeIndex: number, exchange: MessageExchange) {
         exchange.closed.on(() => this.deleteExchange(exchangeIndex));
         this.exchanges.set(exchangeIndex, exchange);
+
+        // A node SHOULD limit itself to a maximum of 5 concurrent exchanges over a unicast session. This is
+        // to prevent a node from exhausting the message counter window of the peer node.
+        // TODO Make sure Group sessions are handled differently
+        this.#cleanupSessionExchanges(exchange.session.id);
+    }
+
+    #cleanupSessionExchanges(sessionId: number) {
+        const sessionExchanges = Array.from(this.exchanges.values()).filter(
+            exchange => exchange.session.id === sessionId && !exchange.isClosing,
+        );
+        if (sessionExchanges.length <= MAXIMUM_CONCURRENT_EXCHANGES_PER_SESSION) {
+            return;
+        }
+        // let's use the first entry in the Map as the oldest exchange and close it
+        const exchangeToClose = sessionExchanges[0];
+        logger.debug(`Closing oldest exchange ${exchangeToClose.id} for session ${sessionId}`);
+        exchangeToClose.close().catch(error => logger.error("Error closing exchange", error)); // TODO Promise??
     }
 }
 

packages/protocol/src/protocol/MessageExchange.ts

@@ -156,6 +156,7 @@ export class MessageExchange {
     #retransmissionTimer: Timer | undefined;
     #retransmissionCounter = 0;
     #closeTimer: Timer | undefined;
+    #isClosing = false;
     #timedInteractionTimer: Timer | undefined;
 
     readonly #peerSessionId: number;
@@ -212,6 +213,14 @@ export class MessageExchange {
         return this.#closed;
     }
 
+    get isClosing() {
+        return this.#isClosing;
+    }
+
+    get id() {
+        return this.#exchangeId;
+    }
+
     /**
      * Max Payload size of the exchange which bases on the maximum payload size of the channel reduced by Matter
      * protocol overhead.
@@ -605,6 +614,7 @@ export class MessageExchange {
 
     async close() {
         if (this.#closeTimer !== undefined) return; // close was already called
+        this.#isClosing = true;
 
         if (this.#receivedMessageToAck !== undefined) {
             this.#receivedMessageAckTimer.stop();
@@ -635,6 +645,7 @@ export class MessageExchange {
     }
 
     private async closeInternal() {
+        this.#isClosing = true;
         this.#retransmissionTimer?.stop();
         this.#closeTimer?.stop();
         this.#timedInteractionTimer?.stop();