/** * @license * Copyright 2022-2026 Matter.js Authors * SPDX-License-Identifier: Apache-2.0 */ import { type ClusterBehavior } from "#behavior/cluster/ClusterBehavior.js"; import { LocalActorContext } from "#behavior/context/server/LocalActorContext.js"; import { CommissioningClient } from "#behavior/system/commissioning/CommissioningClient.js"; import { RemoteDescriptor } from "#behavior/system/commissioning/RemoteDescriptor.js"; import { CommissioningDiscovery } from "#behavior/system/controller/discovery/CommissioningDiscovery.js"; import { ContinuousDiscovery } from "#behavior/system/controller/discovery/ContinuousDiscovery.js"; import { Discovery } from "#behavior/system/controller/discovery/Discovery.js"; import { InstanceDiscovery } from "#behavior/system/controller/discovery/InstanceDiscovery.js"; import { PaseDiscovery } from "#behavior/system/controller/discovery/PaseDiscovery.js"; import { IcdClient } from "#behavior/system/icd/IcdClient.js"; import { NetworkClient } from "#behavior/system/network/NetworkClient.js"; import { NetworkServer } from "#behavior/system/network/NetworkServer.js"; import { BasicInformationClient } from "#behaviors/basic-information"; import { BridgedDeviceBasicInformationClient } from "#behaviors/bridged-device-basic-information"; import { DescriptorClient } from "#behaviors/descriptor"; import { IcdManagementClient } from "#behaviors/icd-management"; import { OperationalCredentialsClient } from "#behaviors/operational-credentials"; import { Endpoint } from "#endpoint/Endpoint.js"; import { EndpointContainer } from "#endpoint/properties/EndpointContainer.js"; import { EndpointType } from "#endpoint/type/EndpointType.js"; import { RootEndpoint } from "#endpoints/root"; import { ClientGroup } from "#node/ClientGroup.js"; import { InteractionServer } from "#node/server/InteractionServer.js"; import { ServerNodeStore } from "#storage/server/ServerNodeStore.js"; import { Bytes, CancelablePromise, createPromise, Diagnostic, Duration, ImplementationError, Lifecycle, Logger, MatterAggregateError, MatterError, MaybePromise, Minutes, Mutex, Observable, Seconds, Time, Timestamp, UninitializedDependencyError, } from "@matter/general"; import { ClientInteraction, ClientSubscriptionHandler, ClientSubscriptions, CommissioningError, DiscoveryData, FabricAuthority, FabricManager, Invoke, Peer, PeerAddress, PeerLeftError, ResultStatus, SessionManager, SessionParameters, } from "@matter/protocol"; import { EndpointNumber, FabricIndex, NodeId, Status } from "@matter/types"; import { Descriptor } from "@matter/types/clusters/descriptor"; import { GeneralCommissioning } from "@matter/types/clusters/general-commissioning"; import { ClientNode } from "../ClientNode.js"; import type { ServerNode } from "../ServerNode.js"; import { ClientNodeFactory } from "./ClientNodeFactory.js"; import { ClientStructure } from "./ClientStructure.js"; import { ClientStructureEvents } from "./ClientStructureEvents.js"; const logger = Logger.get("Peers"); const DEFAULT_TTL = Minutes(15); const EXPIRATION_INTERVAL = Minutes.one; type FabricOperation = "commission" | "decommission"; /** * Thrown when a commission or decommission of a {@link ClientNode} is requested while one is already in progress. */ export class FabricOperationInProgressError extends CommissioningError {} /** * Manages the set of known remote nodes. * * Remote nodes are either peers (commissioned into a fabric we share) or commissionable. */ export class Peers extends EndpointContainer { #expirationInterval?: CancelablePromise; #installedSubscriptionHandler?: ClientSubscriptionHandler; #mutex = new Mutex(this); #closed = false; #fabricOperations = new Map }>(); #instrumented = new WeakSet(); #bridgedInstrumented = new WeakSet(); constructor(owner: ServerNode) { super(owner); if (!owner.env.has(ClientNodeFactory)) { owner.env.set(ClientNodeFactory, new Factory(this)); } owner.env.applyTo(InteractionServer, this.#configureInteractionServer.bind(this)); this.added.on(this.#handlePeerAdded.bind(this)); this.deleted.on(this.#manageExpiration.bind(this)); this.clusterInstalled(BasicInformationClient).on(this.#instrumentBasicInformation.bind(this)); this.clusterInstalled(IcdManagementClient).on(this.#installIcdClient.bind(this)); this.clusterInstalled(BridgedDeviceBasicInformationClient).on( this.#instrumentBridgedConfigurationVersion.bind(this), ); owner.env.get(ClientStructureEvents).structureApplied.on(node => { if (node instanceof ClientNode) { this.#evaluateSeeded(node); } }); const lifecycle = owner.lifecycle; lifecycle.online.on(this.#nodeOnline.bind(this)); lifecycle.offline.on(this.#nodeOffline.bind(this)); } /** * Load nodes. Invoked automatically by owner. */ initialize() { const factory = this.owner.env.get(ClientNodeFactory); const clientStores = this.owner.env.maybeGet(ServerNodeStore)?.clientStores; if (clientStores === undefined) { throw new UninitializedDependencyError( "Peers", "are not available because ServerNode initialization is incomplete", ); } // Group nodes have an in-memory only store, so all nodes restored here are ClientNode for (const id of clientStores.knownIds) { const node = factory.create({ id, owner: this.owner, }); // Claiming the construction error keeps a peer we cannot load from surfacing as an unhandled rejection, // and names the peer the failure belongs to. Construction reports a failure that occurs before it // returns here itself, so adopting that one too would report it twice if (node.construction.status === Lifecycle.Status.Initializing) { node.construction.onError(error => logger.warn("Peer", Diagnostic.strong(id), "is unavailable because it failed to load:", error), ); } this.add(node); } } async #nodeOnline() { if (this.owner.stateOf(NetworkServer).autoStartCommissionedPeers) { for (const peer of this) { if (!peer.lifecycle.isCommissioned || peer.maybeStateOf("network")?.isDisabled) { continue; } try { await peer.start(); } catch (e) { MatterError.accept(e); logger.warn(`Error starting peer ${peer}:`, e); } } } this.#manageExpiration(); } async #nodeOffline() { this.#cancelExpiration(); for (const peer of this) { await peer.stop(); } } /** * Find a specific commissionable node, or, if no discovery options are provided, returns the first discovered node. * TODO: Allow to provide an array of options for multiple discoveries (e.g. from a Multi QR code). */ locate(options?: Discovery.Options) { return new InstanceDiscovery(this.owner, options); } /** * Employ discovery to find a set of commissionable nodes, the options can be used to limit the discovered devices * (e.g. just a specific vendor). * TODO: Allow to provide multiple identifiers for multiple discoveries (e.g. from a Multi QR code). * * If you do not provide a timeout value, will search until canceled, and you need to add a listener to * {@link Discovery#discovered} or {@link added} to receive discovered nodes. */ discover(options?: Discovery.Options) { return new ContinuousDiscovery(this.owner, options); } /** * Find a specific commissionable node and commission. */ commission(options: CommissioningDiscovery.Options) { return new CommissioningDiscovery(this.owner, options); } /** * Find a specific commissionable node and establish a PASE session without commissioning. * * Useful for split-commissioning scenarios where one controller establishes PASE and another * performs the commissioning flow, or for raw PASE channel establishment. */ pase(options: PaseDiscovery.Options) { return new PaseDiscovery(this.owner, options); } /** * Complete a commissioning another commissioner started and handed off before the operational (CASE) step * (see {@link CommissioningClient.CommissioningOptions.finalizeCommissioning}). Operationally discovers and * connects the node, sends {@link GeneralCommissioning} `CommissioningComplete` to disarm the failsafe and * finalize, reads the node's structure and (unless `autoSubscribe` is false) subscribes — exactly as * {@link commission} does, so the peer is seeded rather than a blind commissioned node — then registers it. * `discoveryData` and `sessionParameters` (from the hand-off) seed operational discovery and the CASE session * establishment; `options` mirror the matching {@link commission} options. Throws {@link CommissioningError} and * removes the peer entry if discovery, connection, or `CommissioningComplete` fails; rejects as * {@link runCommissioning} describes if the node is gone or busy. */ async completeCommissioning( nodeId: NodeId, discoveryData?: DiscoveryData, sessionParameters?: Partial, options?: Pick< CommissioningClient.CommissioningOptions, "autoSubscribe" | "defaultSubscription" | "autoStateInitialize" >, ): Promise { // Split commissioning can only finalize over THE fabric the initiating commissioner established (the one the // device's NOC belongs to). Require it to already exist rather than auto-creating an empty fabric, which would // mask a misconfigured controller that can never succeed. Match by CA root certificate, as FabricAuthority does. const fabricAuthority = await this.owner.env.load(FabricAuthority); const fabric = fabricAuthority.fabrics.find(f => Bytes.areEqual(f.rootCert, fabricAuthority.ca.rootCert)); if (fabric === undefined) { throw new ImplementationError( `Cannot complete commissioning for ${nodeId}: the controller does not hold the shared fabric it must be finalized on (import it first)`, ); } const address = fabric.addressOf(nodeId); // forAddress() below returns the existing node for an already-commissioned address rather than creating one, // and the catch block deletes that node on any failure (including a non-throwing CommissioningComplete // errorCode). Without this guard a second completeCommissioning() call for the same peer — e.g. a duplicate // hand-off or a retry — would resolve to the live peer and could delete it. Must check before forAddress() // since that call sets peerAddress, which flips lifecycle.isCommissioned true immediately. const existing = this.get(address); if (existing?.lifecycle.isCommissioned) { throw new ImplementationError(`${existing} is already commissioned`); } // Both-or-nothing: persist the fabric before forAddress() durably writes the node, so a persist failure aborts // cleanly and never leaves a node referencing an unpersisted fabric. await fabric.persist(); const node = await this.forAddress(address); // Serialize like commission() so parallel finalize attempts on the same node cannot both send // CommissioningComplete (the loser races on device failsafe state and would delete the winner's node). return this.runCommissioning(node, async () => { try { if (discoveryData !== undefined) { // forAddress() already bound the live Peer (CommissioningClient#bindPeer fires as soon as // peerAddress is set), snapshotting an empty descriptor. Seed the Peer directly so operational // discovery sees it; writing CommissioningClient.descriptor only updates state that nothing // re-reads. node.env.get(Peer).descriptor.discoveryData = discoveryData; } if (sessionParameters !== undefined) { node.env.get(Peer).descriptor.reportedSessionParameters = sessionParameters; } // Mirror commission()'s post-commission setup so start() reads the node's structure (latching // `seeded`) and, unless opted out, establishes the sustained subscription — otherwise the finalized // peer would be commissioned but blind, and `lifecycle.seeded` would never emit. await node.act(agent => { const network = agent.get(NetworkClient); network.state.defaultSubscription = options?.defaultSubscription; network.state.autoSubscribe = options?.autoSubscribe !== false; network.state.autoStateInitialize = options?.autoStateInitialize; network.internal.isNewlyCommissioned = true; }); await node.start(); // Low-level invoke so the finalize round-trip gets the extended failsafe response timeout, exactly as // ControllerCommissioningFlow does; the generated command method cannot pass invoke options. const invoke = Invoke({ commands: [ Invoke.ConcreteCommandRequest({ endpoint: node, cluster: GeneralCommissioning, command: "commissioningComplete", fields: undefined, }), ], useExtendedFailSafeMessageResponseTimeout: true, }); let response: GeneralCommissioning.CommissioningCompleteResponse | undefined; for await (const chunk of (node.interaction as ClientInteraction).invoke(invoke)) { for (const entry of chunk) { if (entry.kind === "cmd-status" && entry.status !== Status.Success) { throw new CommissioningError( `CommissioningComplete failed for ${nodeId}: status ${entry.status}`, { cause: ResultStatus.errorOf(entry), }, ); } if (entry.kind === "cmd-response") { // DecodedCommandResponse.data is `any`; CommissioningComplete answers with errorCode/debugText. response = entry.data as GeneralCommissioning.CommissioningCompleteResponse; } } } if (response === undefined) { throw new CommissioningError(`CommissioningComplete for ${nodeId} returned no response`); } if (response.errorCode !== GeneralCommissioning.CommissioningError.Ok) { throw new CommissioningError( `CommissioningComplete failed for ${nodeId}: ${response.errorCode} ${response.debugText}`, ); } } catch (error) { // lifecycle.isCommissioned flips true the moment forAddress() sets peerAddress, before start() or // CommissioningComplete run, so any failure up to this point (thrown or via CommissioningError above) // would otherwise leave a phantom commissioned-looking node behind. Once CommissioningComplete // returns Ok below, the device has genuinely disarmed its failsafe and joined the fabric, so a // failure persisting that fact locally must NOT delete the node. try { await node.delete(); } catch (deleteError) { logger.warn(`Error deleting ${node} after failed commissioning completion:`, deleteError); } throw error; } await node.setStateOf(CommissioningClient, { commissionedAt: Time.nowMs }); return node; }); } /** * Find or create a {@link ClientNode} for a device described by {@link descriptor}. * * If a matching node already exists in the peer collection, returns it after refreshing its addresses and * discovery data from the supplied descriptor. Otherwise creates a new node using the descriptor. A matching node * that is being decommissioned is awaited first: if the decommission removes it, a new node is created, otherwise * the kept node is returned. * * After calling {@link forDescriptor}, commission the returned node via {@link ClientNode.commission}. */ async forDescriptor(descriptor: RemoteDescriptor): Promise { const factory = this.owner.env.get(ClientNodeFactory); let node = factory.find(descriptor); while (node !== undefined) { const running = this.#fabricOperations.get(node); if (running?.operation !== "decommission") { break; } await running.settled; node = factory.find(descriptor); } if (node !== undefined) { // Refresh addresses and discovery data from the new descriptor await node.act(agent => { agent.commissioning.descriptor = descriptor; }); return node; } return factory.create({ commissioning: { descriptor } }); } /** * Obtain an {@link Observable} that emits when a specific type of endpoint initializes for a peer. * * This is useful for initializing general behavior on any peer endpoint of the specified type. */ endpointInstalled(type: T): Observable<[endpoint: Endpoint]> { return this.owner.env.get(ClientStructureEvents).endpointInstalled(type); } /** * Obtain a {@link Observable} that emits when a specific type of cluster initializes for a peer. * * This is useful for initializing general behavior on any peer endpoint with the specified cluster. */ clusterInstalled(type: T) { return this.owner.env.get(ClientStructureEvents).clusterInstalled(type); } /** Commissioned operational peers. Excludes commissionable discoveries and group nodes. */ get commissioned(): ClientNode[] { return [...this].filter(node => node.nodeType === "client" && node.lifecycle.isCommissioned); } /** * Look up a peer by numeric/string id or {@link PeerAddress}. A {@link PeerAddress} matches the commissioned * peer whose {@link CommissioningClient} peer address equals it; otherwise the container's id lookup is used. * * Look up by {@link PeerAddress} for anything held while a peer is commissioned: a local id may be reissued to a * different device once the peer it named is gone, so the same string can resolve to a peer that never saw the * work it is being used for. An address is stable for as long as its peer stays commissioned; afterwards it * may name another device, under the rules `ControllerBehavior.allocatePeerAddress` documents. So a lookup * that *succeeds* is not proof this is the peer the caller meant. */ override get(id: number | string | PeerAddress) { if (typeof id !== "string" && typeof id !== "number") { const address = PeerAddress(id); for (const node of this) { if (node.behaviors.active.some(behavior => behavior.id === "commissioning")) { const nodeAddress = node.maybeStateOf("commissioning")?.peerAddress as PeerAddress | undefined; if (nodeAddress !== undefined && PeerAddress.is(nodeAddress, address)) { return node; } } } return undefined; } return super.get(id); } override get owner() { return super.owner as ServerNode; } override add(node: ClientNode) { node.owner = this.owner; super.add(node); } /** * Get or create a client node for the given peer address. * * This is mainly used to communicate to other known nodes on the fabric without having a formal commissioning * process. */ async forAddress(peerAddress: PeerAddress, options: Omit = {}) { if (!this.owner.env.get(FabricManager).has(peerAddress)) { throw new ImplementationError("Cannot register a peer address for a fabric we do not belong to"); } const existing = this.get(peerAddress); if (existing) { return existing; } // Serialize creation through the same mutex the cull/leave paths use. Two concurrent callers for the same // address would otherwise both pass the check and both create a ClientNode (the awaits below yield between // the check and the add). return this.#mutex.produce(async () => { let node = this.get(peerAddress); if (node) { return node; } if (options.id !== undefined) { // We want to initialize a node with a provided id. This could be an injected node, so ensure the // ClientNodeStore is constructed. Without id the storage is empty anyway because id is newly assigned // TODO: Remove when we remove legacy controller const store = this.owner.env.get(ServerNodeStore).clientStores.storeForNode(options.id); await store.construction; } // We do not have that node till now, also not persisted, so create it const factory = this.owner.env.get(ClientNodeFactory); node = factory.create(options, peerAddress); await node.construction; this.add(node); // Nodes we do not commission are not auto-subscribed but enabled // But we add the peer address await node.set({ commissioning: { peerAddress: PeerAddress(peerAddress) }, }); return node; }); } /** * Delete every known node. * * A node we cannot tear down cleanly is closed and dropped instead, so it cannot block the factory reset this * serves; its persisted data outlives this call and is the caller's to discard. Failures are logged rather than * raised for the same reason. */ async erase() { await this.#mutex.produce(async () => { for (const node of [...this]) { try { await node.delete(); continue; } catch (error) { logger.warn(`Error deleting ${node}:`, error); } try { await node.close(); } catch (error) { logger.warn(`Error closing ${node}:`, error); } this.delete(node); } }); } override async close() { this.#closed = true; await MatterAggregateError.settleSeries( [ () => this.#installedSubscriptionHandler?.close(), () => this.#cancelExpiration(), () => this.#mutex, () => super.close(), ], `Error closing peers of ${this.owner}`, ); } #cancelExpiration() { if (this.#expirationInterval) { this.#expirationInterval.cancel(); this.#expirationInterval = undefined; } } #handlePeerAdded(peer: ClientNode) { this.#watchSeeding(peer); if (this.owner.env.has(InteractionServer)) { this.#configureInteractionServer(); } this.#manageExpiration(); } /** * Evaluates seeding wherever the node's structure may have become complete: after the structure applies data (see * the constructor), when the node becomes active with its cached structure, and when endpoints or behaviors change * outside of the structure, as with `ClientNodeEndpoints.require()`. */ #watchSeeding(peer: ClientNode) { peer.construction.change.on(status => { if (status === Lifecycle.Status.Active) { this.#evaluateSeeded(peer); } }); peer.lifecycle.changed.on(() => this.#evaluateSeeded(peer)); } /** * If required, installs a listener in the environment's {@link InteractionServer} to handle subscription responses. */ #configureInteractionServer() { if (this.#closed || !this.owner.env.has(InteractionServer)) { return; } // A node restart replaces the InteractionServer and ClientSubscriptions, so each new server needs a handler // bound to the current subscriptions const interactionServer = this.owner.env.get(InteractionServer); if ( this.#installedSubscriptionHandler !== undefined && interactionServer.clientHandler === this.#installedSubscriptionHandler ) { return; } this.#installedSubscriptionHandler = new ClientSubscriptionHandler(this.owner.env.get(ClientSubscriptions)); interactionServer.clientHandler = this.#installedSubscriptionHandler; } /** * Run a commission attempt on {@link node} while protecting the node from the expired-node cull. * * Rejects with `DestroyedDependencyError` or `CrashedDependencyError` if the node is gone (its closing or deletion * began, or it crashed), so the attempt fails fast instead of crashing later when the closed backing is accessed, and with {@link FabricOperationInProgressError} if a commission or decommission attempt on it * is already in progress, as parallel attempts on the same {@link ClientNode} would race on device-side state. * Calling {@link ClientNode.commission} inside {@link fn} therefore rejects. */ async runCommissioning(node: ClientNode, fn: () => MaybePromise): Promise { return this.#runFabricOperation(node, "commission", fn); } /** * Run a decommission attempt on {@link node}. While it runs, the expired-node cull and leave events do not delete * the node, and {@link forDescriptor} waits for it to end before handing out a node for the device. * * Rejects like {@link runCommissioning}: with `DestroyedDependencyError` or `CrashedDependencyError` if the node is * gone, with {@link FabricOperationInProgressError} if a commission or decommission of it is already in progress. */ async runDecommissioning(node: ClientNode, fn: () => MaybePromise): Promise { return this.#runFabricOperation(node, "decommission", fn); } /** * Registration runs as a {@link #mutex} task, so no cull or leave handling is in flight for the node once * {@link fn} starts, and every later one sees the registration. */ async #runFabricOperation(node: ClientNode, operation: FabricOperation, fn: () => MaybePromise): Promise { const { promise: settled, resolver: settle } = createPromise(); await this.#mutex.produce(async () => { node.lifecycle.assertNotGone(); const running = this.#fabricOperations.get(node); if (running !== undefined) { throw new FabricOperationInProgressError( `Cannot ${operation} ${node.toString()} because a ${running.operation} attempt is already in progress`, ); } this.#fabricOperations.set(node, { operation, settled }); }); try { return await fn(); } finally { this.#fabricOperations.delete(node); settle(); } } /** * Enables or disables the expiration timer that culls expired uncommissioned nodes. */ #manageExpiration() { if (this.#closed) { return; } if (this.#expirationInterval) { if (!this.size) { this.#cancelExpiration(); } return; } if (!this.size) { return; } this.#expirationInterval = Time.sleep("client node expiration", EXPIRATION_INTERVAL).then( this.#onExpirationIntervalElapsed.bind(this), ); } #onExpirationIntervalElapsed() { if (this.#closed) { return; } this.#mutex.run(() => this.#cullExpiredNodesAndAddresses() .catch(error => { logger.warn("Error culling expired nodes", error); }) .finally(() => { this.#manageExpiration(); }), ); } async #cullExpiredNodesAndAddresses() { try { const now = Time.nowMs; for (const node of this) { if (!node.lifecycle.isReady) { continue; } if (this.#fabricOperations.has(node)) { continue; } const state = node.maybeStateOf(CommissioningClient); if (state === undefined) { continue; } const { addresses } = state; const isCommissioned = state.peerAddress !== undefined; // Shortcut for conditions we know no change is possible if (addresses === undefined || (isCommissioned && addresses.length === 1)) { continue; } // Remove expired addresses let newAddresses = addresses.filter(addr => { const exp = expirationOf(addr); if (exp === undefined) { return true; } return exp > now; }); // Cull commissionable nodes that have expired if (!isCommissioned) { if (!newAddresses?.length || (expirationOf(state) ?? 0) <= now) { await node.delete(); continue; } } // If the node is commissioned, do not remove the last address. Instead keep the "least expired" addresses if (isCommissioned && addresses.length && !newAddresses.length) { if (addresses.length === 1) { continue; } const freshestExp = addresses.reduce((freshestExp, addr) => { return Math.max(freshestExp, expirationOf(addr)!); }, 0); newAddresses = addresses.filter(addr => expirationOf(addr) === freshestExp); } // Apply new addresses if changed if (addresses.length !== newAddresses.length) { await node.set({ commissioning: { addresses: newAddresses } }); } } } finally { this.#expirationInterval = undefined; } } #instrumentBasicInformation(node: Endpoint, type: typeof BasicInformationClient) { if (!(node instanceof ClientNode)) { return; } // clusterInstalled re-emits per node (endpoint-install, cluster-add, structure rebuilds), so register the // device-event handlers below once per node to avoid duplicate side effects on repeat emits. if (this.#instrumented.has(node)) { return; } this.#instrumented.add(node); node.eventsOf(type).leave?.on(({ fabricIndex }) => this.#onLeave(node, fabricIndex)); node.eventsOf(type).shutDown?.on(() => this.#onShutdown(node)); node.eventsOf(type).startUp?.on(() => this.#onStartUp(node)); node.eventsOf(type).configurationVersion$Changed?.on(() => node.lifecycle.configurationVersionChanged.emit()); } /** * A node is seeded once its structure is known: BasicInformation is present and every endpoint the PartsList the * root endpoint reported names is installed. An empty PartsList means the node has only its root endpoint. A node * whose PartsList is unknown, such as one built with `ClientNodeEndpoints.require()`, needs an endpoint beyond * the root. */ #evaluateSeeded(node: ClientNode) { if (node.lifecycle.isSeeded || !node.lifecycle.isReadable) { return; } const structure = node.env.maybeGet(ClientStructure); if (structure?.isApplying || node.maybeStateOf(BasicInformationClient) === undefined) { return; } // Values outside the endpoint number range name nothing the structure could place const descriptor = node.maybeStateOf(DescriptorClient); const parts = structure?.isDescriptorListReported(EndpointNumber(0), Descriptor.attributes.partsList.id) ? descriptor?.partsList.filter(number => EndpointNumber.isValid(number)) : undefined; const endpoints = node.endpoints; if (parts === undefined ? endpoints.size <= 1 : parts.some(number => !endpoints.has(number))) { return; } node.lifecycle.markSeeded(LocalActorContext.ReadOnly); if ( parts !== undefined && !parts.length && structure?.isDescriptorListReported(EndpointNumber(0), Descriptor.attributes.deviceTypeList.id) && descriptor?.deviceTypeList.every(({ deviceType }) => deviceType === RootEndpoint.deviceType) ) { logger.warn( `Peer ${node} has no endpoint besides its root and no device type besides Root Node,`, "so it offers no function beyond node management", ); } } /** * Bridged endpoints carry their own `ConfigurationVersion` on `BridgedDeviceBasicInformation`; surface its changes * through the same endpoint lifecycle event used for the node's `BasicInformation`. */ #instrumentBridgedConfigurationVersion(endpoint: Endpoint, type: typeof BridgedDeviceBasicInformationClient) { // clusterInstalled re-emits per endpoint; register once if (this.#bridgedInstrumented.has(endpoint)) { return; } this.#bridgedInstrumented.add(endpoint); endpoint .eventsOf(type) .configurationVersion$Changed?.on(() => endpoint.lifecycle.configurationVersionChanged.emit()); } #installIcdClient(node: Endpoint) { if (!(node instanceof ClientNode) || node.behaviors.has(IcdClient)) { return; } node.behaviors.inject(IcdClient); } #onLeave(node: ClientNode, fabricIndex: FabricIndex) { if (!node.lifecycle.isReady) { return; } this.#mutex.run(async () => { if (!node.lifecycle.isReady || this.#fabricOperations.get(node)?.operation === "decommission") { return; } const { fabrics: peerFabrics } = node.maybeStateOf(OperationalCredentialsClient) ?? {}; const peerFabric = peerFabrics?.find(fabric => fabric.fabricIndex === fabricIndex); if (!peerFabric) { return; } const peerAddress = node.maybeStateOf(CommissioningClient)?.peerAddress; if (!peerAddress) { return; } const localFabrics = this.owner.env.get(FabricManager); const localFabric = localFabrics.forDescriptor(peerFabric); if (!localFabric || localFabric.fabricIndex !== peerAddress.fabricIndex) { return; } // Ignore leave events received during an initial subscription establishment as they may be stale events // from before the device was re-commissioned with the same identifier. // The reason is that we saw such cases and should prevent discarding a node directly after commissioning. // This solution still has some holes that could prevent removing nodes automatically, but best-effort // variant for now until we know how often that happens in practice. if (!(await node.act(agent => agent.get(NetworkClient).subscriptionActive))) { logger.info( "Leave event for peer", Diagnostic.strong(node.id), " received without active subscription. Ignoring.", ); return; } logger.notice("Peer", Diagnostic.strong(node.id), "has left the fabric"); node.lifecycle.decommissioned.emit(LocalActorContext.ReadOnly); try { await node.env.maybeGet(Peer)?.disconnect(new PeerLeftError()); } catch (error) { logger.warn(`Error force-closing sessions for ${node.id} on Leave:`, error); } await node.delete(); }); } async #onShutdown(node: ClientNode) { if (!node.lifecycle.isReady || !node.lifecycle.isOnline) { return; } // Ignore shutdown events received during initial subscription establishment as they may be stale events // from before the device was restarted. This mirrors the same guard in #onLeave. if (!(await node.act(agent => agent.get(NetworkClient).subscriptionActive))) { logger.debug( "Shutdown event for peer", Diagnostic.strong(node.id), "received without active subscription. Ignoring.", ); return; } const peerAddress = node.maybeStateOf(CommissioningClient)?.peerAddress; if (peerAddress !== undefined) { // Shutdown event means the device reboots, handle it like a peer loss and remove all sessions await this.owner.env.get(SessionManager).handlePeerShutdown(peerAddress); } } /** * Process startup events to detect node reboots in real-time. This allows us to terminate sessions * that would otherwise linger until timed out and cause interaction delays. */ async #onStartUp(node: ClientNode) { // isOnline is checked because startUp arrives via an active subscription channel — the node // must be online for the subscription to deliver events. isReady guards against races during // node initialization where the event might fire before the node is fully set up. if (!node.lifecycle.isReady || !node.lifecycle.isOnline) { return; } // Ignore startup events received during the initial subscription establishment // as they may be stale events from before the device was restarted. if (!(await node.act(agent => agent.get(NetworkClient).subscriptionActive))) { logger.debug( "Startup event for peer", Diagnostic.strong(node.id), "received without active subscription. Ignoring.", ); return; } const peerAddress = node.maybeStateOf(CommissioningClient)?.peerAddress; if (peerAddress === undefined) { return; } // Use the current session's createdAt as asOf so it (and newer sessions) are preserved // while older sessions (from before the reboot) are closed. If the currentSession is // undefined (no known session), handlePeerShutdown uses the current Time.nowUs value as its cutoff. const sessionManager = this.owner.env.get(SessionManager); await sessionManager.handlePeerShutdown(peerAddress, sessionManager.maybeSessionFor(peerAddress)?.createdAt); } } class Factory extends ClientNodeFactory { #owner: Peers; #groupIdCounter = 0; /** * Creation descriptors for {@link find}, which cannot read a node's state before its first construction completes. */ #descriptorsUnderConstruction = new WeakMap(); constructor(owner: Peers) { super(); this.#owner = owner; } create(options: ClientNode.Options, peerAddress?: PeerAddress) { let node: ClientNode; if (peerAddress !== undefined && PeerAddress.isGroup(peerAddress)) { node = new ClientGroup({ ...options, id: options.id ?? `group${++this.#groupIdCounter}`, owner: this.#owner.owner, }); } else { node = new ClientNode({ ...options, id: options.id ?? this.#owner.owner.env.get(ServerNodeStore).clientStores.allocateId(), owner: this.#owner.owner, }); } const descriptor = options.commissioning?.descriptor; if (descriptor !== undefined) { this.#descriptorsUnderConstruction.set(node, descriptor); const forget = (status: Lifecycle.Status) => { if (status !== Lifecycle.Status.Initializing) { this.#descriptorsUnderConstruction.delete(node); node.construction.change.off(forget); } }; node.construction.change.on(forget); } node.construction.start(); return node; } find(descriptor: RemoteDescriptor) { for (const node of this.#owner) { if (!node.lifecycle.isReadable && node.construction.status !== Lifecycle.Status.Initializing) { continue; } const known = this.#descriptorsUnderConstruction.get(node) ?? node.state.commissioning; if (RemoteDescriptor.is(known, descriptor)) { return node; } } } get nodes() { return this.#owner; } } function expirationOf( lifespan: T, ): T extends { discoveredAt: Timestamp } ? Timestamp : Timestamp | undefined { if (lifespan.discoveredAt !== undefined) { return Timestamp(lifespan.discoveredAt + (Seconds(lifespan.ttl) ?? DEFAULT_TTL)); } return undefined as unknown as Timestamp; }