From 38b12f2066c576ba25f382cf5f149b25d11a694b Mon Sep 17 00:00:00 2001
From: Ingo Fischer <github@fischer-ka.de>
Date: Sun, 15 Sep 2024 01:08:28 +0200
Subject: [PATCH] Limits the parallel open exchanges to 5

---
 CHANGELOG.md                                  |  3 +++
 .../matter.js/src/protocol/ExchangeManager.ts | 23 ++++++++++++++++---
 .../matter.js/src/protocol/MessageExchange.ts | 11 +++++++++
 3 files changed, 34 insertions(+), 3 deletions(-)

diff --git a/CHANGELOG.md b/CHANGELOG.md
index dfe9ee214f..159f71a56c 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -18,6 +18,9 @@ The main work (all changes without a GitHub username in brackets in the below li
     -   BREAKING: The "ByteArray" type is removed, replaced with native-JS Uint8Array and a small collection of utility functions in the "Bytes" namespace
     -   The Matter object model previously exported as @project-chip/matter.js/model now resides in @project-chip/matter.js-model
 
+-   matter.js-protocol:
+    -   Limits the number of parallel exchanges to 5 
+
 -   matter.js-nodejs:
     -   Node specialization is moved to matter.js-nodejs.  matter-node.js remains as a compatibility import.
     -   BREAKING: The previously deprecated re-exports in matter-node.js from matter.js are removed.
diff --git a/packages/matter.js/src/protocol/ExchangeManager.ts b/packages/matter.js/src/protocol/ExchangeManager.ts
index 7ce4ddfb38..7ac14fb262 100644
--- a/packages/matter.js/src/protocol/ExchangeManager.ts
+++ b/packages/matter.js/src/protocol/ExchangeManager.ts
@@ -32,6 +32,8 @@ import { SecureChannelMessenger } from "./securechannel/SecureChannelMessenger.j
 
 const logger = Logger.get("ExchangeManager");
 
+const MAXIMUM_CONCURRENT_EXCHANGES_PER_SESSION = 5;
+
 export class ChannelNotConnectedError extends MatterError {}
 
 export class MessageChannel<ContextT extends SessionContext> implements Channel<Message> {
@@ -284,9 +286,6 @@ export class ExchangeManager<ContextT extends SessionContext> {
                     );
                 }
             }
-
-            // 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.
         }
     }
 
@@ -358,6 +357,24 @@ export class ExchangeManager<ContextT extends SessionContext> {
     #addExchange(exchangeIndex: number, exchange: MessageExchange<ContextT>) {
         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??
     }
 }
 
diff --git a/packages/matter.js/src/protocol/MessageExchange.ts b/packages/matter.js/src/protocol/MessageExchange.ts
index a6fd2523fc..2be5142eeb 100644
--- a/packages/matter.js/src/protocol/MessageExchange.ts
+++ b/packages/matter.js/src/protocol/MessageExchange.ts
@@ -165,6 +165,7 @@ export class MessageExchange<ContextT extends SessionContext> {
     #retransmissionTimer: Timer | undefined;
     #retransmissionCounter = 0;
     #closeTimer: Timer | undefined;
+    #isClosing = false;
     #timedInteractionTimer: Timer | undefined;
 
     readonly #peerSessionId: number;
@@ -221,6 +222,14 @@ export class MessageExchange<ContextT extends SessionContext> {
         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.
@@ -613,6 +622,7 @@ export class MessageExchange<ContextT extends SessionContext> {
 
     async close() {
         if (this.#closeTimer !== undefined) return; // close was already called
+        this.#isClosing = true;
 
         if (this.#receivedMessageToAck !== undefined) {
             this.#receivedMessageAckTimer.stop();
@@ -643,6 +653,7 @@ export class MessageExchange<ContextT extends SessionContext> {
     }
 
     private async closeInternal() {
+        this.#isClosing = true;
         this.#retransmissionTimer?.stop();
         this.#closeTimer?.stop();
         this.#timedInteractionTimer?.stop();