fix(protocol): apply the peer's MRP margin to inbound exchanges

CHANGELOG.md

@@ -9,6 +9,11 @@ The main work (all changes without a GitHub username in brackets in the below li
 	## __WORK IN PROGRESS__
 -->
 
+## __WORK IN PROGRESS__
+
+- @matter/protocol
+    - Fix: The peer's medium-specific MRP retransmission margin now applies to peer-initiated exchanges (e.g. subscription data reports, BDX transfers) and to exchanges created without peer context
+
 ## 0.17.7 (2026-07-27)
 
 - @matter/general

packages/node/test/node/ClientTuningTest.ts

@@ -12,7 +12,14 @@ import { Endpoint } from "#endpoint/Endpoint.js";
 import { SecondaryNetworkInterfaceEndpoint } from "#endpoints/secondary-network-interface";
 import { ServerNode } from "#index.js";
 import { Bytes, causedBy, Crypto, Millis, MockCrypto, MockNetwork, Network, Seconds } from "@matter/general";
-import { NetworkProfiles, PeerSet, PeerTimingParameters, PeerUnreachableError, SessionManager } from "@matter/protocol";
+import {
+    MdnsService,
+    NetworkProfiles,
+    PeerSet,
+    PeerTimingParameters,
+    PeerUnreachableError,
+    SessionManager,
+} from "@matter/protocol";
 import { NetworkCommissioning } from "@matter/types/clusters/network-commissioning";
 import { ThreadNetworkDiagnostics } from "@matter/types/clusters/thread-network-diagnostics";
 import { MockServerNode } from "./mock-server-node.js";
@@ -73,6 +80,37 @@ describe("ClientTuningTest", () => {
         expect(peer.network.additionalMrpDelay).equals(Seconds(1));
     });
 
+    it("exposes the peer's MRP margin on the session so peer-initiated exchanges inherit it", async () => {
+        await using site = new MockSite();
+        const controller = await site.addController();
+        const device = await site.addNode(WifiRoot, { device: OnOffLightDevice });
+
+        await commission(controller, device);
+
+        const peer = [...controller.env.get(PeerSet)][0];
+        expect(peer.newestSession()?.peerAdditionalMrpDelay).equals(Seconds(1));
+    });
+
+    it("adopts sessions that predate its construction", async () => {
+        await using site = new MockSite();
+        const controller = await site.addController();
+        const device = await site.addNode(WifiRoot, { device: OnOffLightDevice });
+
+        await commission(controller, device);
+
+        const { env } = controller;
+        const late = new PeerSet({
+            lifetime: env,
+            sessions: env.get(SessionManager),
+            names: env.get(MdnsService).names,
+            networks: env.get(NetworkProfiles),
+        });
+
+        const peer = [...late][0];
+        expect(peer?.address).deep.equals([...env.get(PeerSet)][0].address);
+        expect(peer.sessions.size).equals(1);
+    });
+
     it("uses default timing when not configured", async () => {
         await using site = new MockSite();
         const { controller } = await site.addCommissionedPair();

packages/protocol/src/peer/Peer.ts

@@ -166,6 +166,11 @@ export class Peer {
 
             // Ensure session parameters reflect those most recently reported by peer
             this.#descriptor.sessionParameters = session.parameters;
+
+            // Only pad when we actually know the peer's medium; the "unknown" fallback is deliberately not applied
+            // here because it would inflate responder timing for every peer of a node that never characterizes them.
+            session.peerAdditionalMrpDelayResolver = () =>
+                this.#physicalProperties === undefined ? undefined : this.network.additionalMrpDelay;
         });
     }
 

packages/protocol/src/peer/PeerSet.ts

@@ -7,6 +7,7 @@
 import { MdnsService } from "#mdns/MdnsService.js";
 import { PeerAddress } from "#peer/PeerAddress.js";
 import { ExchangeManager } from "#protocol/ExchangeManager.js";
+import type { NodeSession } from "#session/NodeSession.js";
 import { Session } from "#session/Session.js";
 import { SessionManager } from "#session/SessionManager.js";
 import {
@@ -29,7 +30,6 @@ import {
     RetrySchedule,
     ServerAddressIp,
 } from "@matter/general";
-import { FabricIndex } from "@matter/types";
 import { NetworkProfiles } from "./NetworkProfile.js";
 import { Peer } from "./Peer.js";
 import { PeerConnection } from "./PeerConnection.js";
@@ -91,32 +91,36 @@ export class PeerSet implements ImmutableSet<Peer>, ObservableSet<Peer> {
             join: name => this.#lifetime.join(name),
         };
 
-        this.#peers.added.on(peer => {
+        this.#observers.on(this.#peers.added, peer => {
+            // Scoped to the peer's own observable, so it needs no cleanup of ours
             peer.sessions.deleted.on(() => {
                 if (!peer.sessions.size) {
                     this.#disconnected.emit(peer);
                 }
             });
         });
 
-        this.#sessions.sessions.added.on(session => {
-            if (session.peerAddress.fabricIndex === FabricIndex.NO_FABRIC || session.isClosed) {
-                return;
-            }
+        this.#observers.on(this.#sessions.sessions.added, session => this.#adoptSession(session));
 
-            this.addKnownPeer({
-                address: session.peerAddress,
-                operationalAddress: operationalAddressOf(session),
-            });
-        });
+        // Sessions may predate us: nothing guarantees we are instantiated before the node establishes its first
+        // operational session
+        for (const session of this.#sessions.sessions) {
+            this.#adoptSession(session);
+        }
+    }
 
-        this.#observers.on(this.#sessions.sessions.added, session => {
-            if (session.fabric === undefined) {
-                return;
-            }
+    /**
+     * Associate an operational session with its peer, creating the peer if we do not know it yet.
+     */
+    #adoptSession(session: NodeSession) {
+        if (session.fabric === undefined || session.isClosed) {
+            return;
+        }
 
-            this.for(session.peerAddress).sessions.add(session);
-        });
+        this.addKnownPeer({
+            address: session.peerAddress,
+            operationalAddress: operationalAddressOf(session),
+        }).sessions.add(session);
     }
 
     set exchanges(exchanges: ExchangeManager | undefined) {

packages/protocol/src/protocol/ExchangeProvider.ts

@@ -130,8 +130,6 @@ export class DedicatedChannelExchangeProvider extends ExchangeProvider {
     }
 
     async initiateExchange(options?: NewExchangeOptions): Promise<MessageExchange> {
-        // This provider has no peer/medium context, so the medium-derived margin is unavailable; only an explicit
-        // per-call override can apply here.
         return this.exchangeManager.initiateExchangeForSession(this.#session, INTERACTION_PROTOCOL_ID, {
             peerAdditionalMrpDelay: options?.additionalMrpDelay,
         });

packages/protocol/src/protocol/MessageExchange.ts

@@ -1046,7 +1046,10 @@ export class MessageExchange {
 
     /** Amplified backoff addition applied to our sends: the larger of our own and the peer's network-profile delay. */
     get #sendAdditionalDelay() {
-        return Duration.max(this.#context.localAdditionalMrpDelay, this.#peerAdditionalMrpDelay ?? Millis(0));
+        return Duration.max(
+            this.#context.localAdditionalMrpDelay,
+            this.#peerAdditionalMrpDelay ?? this.session.peerAdditionalMrpDelay ?? Millis(0),
+        );
     }
 
     /**
@@ -1111,6 +1114,9 @@ export namespace MessageExchange {
         /**
          * Additive MRP retransmission margin for the peer's network medium.  Sourced independently of
          * {@link network} so concurrency overrides cannot strip the medium-correct margin (e.g. thread's).
+         *
+         * Overrides {@link Session.peerAdditionalMrpDelay}, which supplies the margin for exchanges created without
+         * this option (notably peer-initiated ones).
          */
         peerAdditionalMrpDelay?: Duration;
 

packages/protocol/src/session/Session.ts

@@ -55,6 +55,7 @@ export abstract class Session {
     activeTimestamp: Timestamp = 0;
     abstract type: SessionType;
 
+    #peerAdditionalMrpDelay?: () => Duration | undefined;
     #closing = ObservableValue();
     #gracefulClose = AsyncObservable<[]>();
     readonly #exchanges = new Set<MessageExchange>();
@@ -199,6 +200,23 @@ export abstract class Session {
         };
     }
 
+    /**
+     * Additive MRP retransmission margin for the peer's network medium, or undefined if the medium is unknown.
+     *
+     * Applies to every exchange on the session, including those the peer initiates.
+     */
+    get peerAdditionalMrpDelay(): Duration | undefined {
+        return this.#peerAdditionalMrpDelay?.();
+    }
+
+    /**
+     * Installs the resolver for {@link peerAdditionalMrpDelay}.  A resolver rather than a value because the peer's
+     * medium often becomes known (or changes, e.g. a new thread channel) only after the session exists.
+     */
+    set peerAdditionalMrpDelayResolver(resolver: () => Duration | undefined) {
+        this.#peerAdditionalMrpDelay = resolver;
+    }
+
     /**
      * Allows updating the Session timing parameters based on received information from the peer during PASE/CASE initialization
      */

packages/protocol/test/protocol/MessageExchangeTest.ts

@@ -269,6 +269,8 @@ describe("MessageExchange", () => {
             localAdditionalMrpDelay: Duration;
             localFixedMrpBackoff?: Duration;
             peerAdditionalMrpDelay?: Duration;
+            sessionPeerAdditionalMrpDelay?: Duration;
+            peerInitiated?: boolean;
             network?: NetworkProfile;
         }): Promise<{ additionalDelay?: Duration; fixedBackoff?: Duration }> {
             // MRP only engages on unreliable transports; the default mock channel is reliable.
@@ -288,19 +290,26 @@ describe("MessageExchange", () => {
                 throw new NetworkError("captured");
             };
 
-            const exchange = MessageExchange.initiate(
-                {
-                    session,
-                    localSessionParameters: SessionParameters(SessionParameters.defaults),
-                    localAdditionalMrpDelay: options.localAdditionalMrpDelay,
-                    localFixedMrpBackoff: options.localFixedMrpBackoff ?? Millis(0),
-                    async peerLost() {},
-                    retry() {},
-                },
-                1,
-                SECURE_CHANNEL_PROTOCOL_ID,
-                { network: options.network, peerAdditionalMrpDelay: options.peerAdditionalMrpDelay },
-            );
+            if (options.sessionPeerAdditionalMrpDelay !== undefined) {
+                session.peerAdditionalMrpDelayResolver = () => options.sessionPeerAdditionalMrpDelay;
+            }
+
+            const context = {
+                session,
+                localSessionParameters: SessionParameters(SessionParameters.defaults),
+                localAdditionalMrpDelay: options.localAdditionalMrpDelay,
+                localFixedMrpBackoff: options.localFixedMrpBackoff ?? Millis(0),
+                async peerLost() {},
+                retry() {},
+            };
+            const exchangeOptions = {
+                network: options.network,
+                peerAdditionalMrpDelay: options.peerAdditionalMrpDelay,
+            };
+
+            const exchange = options.peerInitiated
+                ? MessageExchange.fromInitialMessage(context, fakeInboundMessage(), exchangeOptions)
+                : MessageExchange.initiate(context, 1, SECURE_CHANNEL_PROTOCOL_ID, exchangeOptions);
 
             await expect(exchange.send(1, Bytes.empty, { requiresAck: true })).to.be.rejectedWith("captured");
             return captured;
@@ -336,6 +345,35 @@ describe("MessageExchange", () => {
             expect(captured.additionalDelay).equals(Millis(0));
         });
 
+        it("applies the session's peer margin to peer-initiated exchanges", async () => {
+            const captured = await captureAdditionalDelay({
+                localAdditionalMrpDelay: Millis(0),
+                sessionPeerAdditionalMrpDelay: Seconds(1.5),
+                peerInitiated: true,
+            });
+
+            expect(captured.additionalDelay).equals(Seconds(1.5));
+        });
+
+        it("applies the session's peer margin to initiated exchanges without an explicit margin", async () => {
+            const captured = await captureAdditionalDelay({
+                localAdditionalMrpDelay: Millis(0),
+                sessionPeerAdditionalMrpDelay: Seconds(1.5),
+            });
+
+            expect(captured.additionalDelay).equals(Seconds(1.5));
+        });
+
+        it("prefers the explicit peer margin over the session's", async () => {
+            const captured = await captureAdditionalDelay({
+                localAdditionalMrpDelay: Millis(0),
+                peerAdditionalMrpDelay: Millis(0),
+                sessionPeerAdditionalMrpDelay: Seconds(1.5),
+            });
+
+            expect(captured.additionalDelay).equals(Millis(0));
+        });
+
         it("passes localFixedMrpBackoff through as the fixed backoff pad, separate from additionalDelay", async () => {
             const captured = await captureAdditionalDelay({
                 localAdditionalMrpDelay: Millis(0),