/** * @license * Copyright 2022-2026 Matter.js Authors * SPDX-License-Identifier: Apache-2.0 */ import { Behavior } from "#behavior/Behavior.js"; import type { ActionContext } from "#behavior/context/ActionContext.js"; import { DclBehavior } from "#behavior/system/dcl/DclBehavior.js"; import { NetworkClient } from "#behavior/system/network/NetworkClient.js"; import { OtaAnnouncements } from "#behavior/system/software-update/OtaAnnouncements.js"; import { OTA_PROGRESS_TIMEOUT } from "#behavior/system/software-update/OtaProgress.js"; import { BasicInformationClient } from "#behaviors/basic-information"; import { Endpoint } from "#endpoint/Endpoint.js"; import type { ClientNode } from "#node/ClientNode.js"; import { Node } from "#node/Node.js"; import type { ServerNode } from "#node/ServerNode.js"; import { Bytes, Diagnostic, Duration, EventEmitter, Hours, ImplementationError, InternalError, Logger, MatterError, Millis, Minutes, Observable, ObserverGroup, Seconds, Time, Timer, Timestamp, } from "@matter/general"; import type { Session } from "@matter/protocol"; import { BdxProtocol, ClientSubscriptions, DclOtaUpdateService, FabricAuthority, FileDesignator, hasRemoteActor, OtaUpdateError, OtaUpdateSource, PeerAddress, PeerAddressMap, RebootResubscribeArmer, SessionManager, StorageScope, } from "@matter/protocol"; import { VendorId } from "@matter/types"; import { OtaSoftwareUpdateProvider } from "@matter/types/clusters/ota-software-update-provider"; import { OtaSoftwareUpdateRequestor } from "@matter/types/clusters/ota-software-update-requestor"; const logger = Logger.get("SoftwareUpdateManager"); /** Rejects transfer options a BDX session could not honour, before any state changes on their behalf. */ function assertTransferOptions({ maxBdxBlockSize, bdxAdditionalMrpDelay }: SoftwareUpdateManager.UpdateTarget) { if (maxBdxBlockSize !== undefined && (!Number.isInteger(maxBdxBlockSize) || maxBdxBlockSize <= 0)) { throw new ImplementationError(`maxBdxBlockSize must be a positive integer, got ${maxBdxBlockSize}`); } if (bdxAdditionalMrpDelay !== undefined && (!Number.isFinite(bdxAdditionalMrpDelay) || bdxAdditionalMrpDelay < 0)) { throw new ImplementationError( `bdxAdditionalMrpDelay must be a finite, non-negative duration, got ${bdxAdditionalMrpDelay}`, ); } } interface UpdateConsent extends SoftwareUpdateManager.UpdateTarget { peerAddress: PeerAddress; } interface UpdateQueueEntry extends UpdateConsent { endpoint: Endpoint; lastProgressUpdateTime?: Timestamp; lastProgressStatus?: OtaUpdateStatus; lastBdxTransferredBytes?: number; /** With `Applying`, the delay the device waits out before it applies; progress is not expected meanwhile. */ applyDelay?: Duration; } interface UpdateConsentEntry extends DclOtaUpdateService.OtaUpdateListEntry { consentPeers: PeerAddress[]; } interface CollectedNodesUpdateInfo { vendorId: VendorId; productId: number; softwareVersion: number; softwareVersionString: string; otaEndpoints: Set<{ endpoint: Endpoint; peerAddress: PeerAddress }>; } export interface OtaUpdateAvailableDetails { newSoftwareVersion: number; newSoftwareVersionString: string; fileDesignator: FileDesignator; httpsUri?: string; consentRequired: boolean; metadataForRequestor?: Bytes; // If provided will be passed back with the QueryImage command } export enum OtaUpdateStatus { Unknown, WaitForConsent, Querying, Downloading, WaitForApply, Applying, Done, Cancelled, } export interface SoftwareUpdateInfo { /** VendorId of the updated device */ vendorId: VendorId; /** ProductId of the updated device */ productId: number; /** Software version of the updated device */ softwareVersion: number; /** Software Version String of the updated device */ softwareVersionString: string; /** ReleaseNotesUrl for the update */ releaseNotesUrl?: string; /** SpecificationVersion of the updated device */ specificationVersion?: number; /** Source of the update returned, whether it is a local file, downloaded from test or production DCL */ source: OtaUpdateSource; } export interface PendingUpdateInfo { peerAddress: PeerAddress; vendorId: VendorId; productId: number; targetSoftwareVersion: number; /** * - `"queued"` — waiting for its turn; no announcement has been sent yet. * - `"in-progress"` — the OTA provider has been announced to the node and we are waiting for it to complete. * - `"stalled"` — no status update and no BDX transfer activity within {@link OTA_PROGRESS_TIMEOUT}. The entry * will be automatically reset and retried. */ status: "queued" | "in-progress" | "stalled"; lastProgressStatus?: OtaUpdateStatus; lastProgressUpdateTime?: Timestamp; /** Bytes transferred by the BDX session of an in-progress update, undefined for any other status. */ bdxTransferredBytes?: number; /** Size of the transfer of an in-progress update, undefined for any other status or while it is unknown. */ bdxDataLength?: number; } /** * The Software Update Manager is the instance to bridge between the central OTA store and DCL service and manage the * updates for all peers of a node. It gets installed on the endpoint where the OtaSoftwareUpdateProvider behavior is * installed. It uses the generic DclOtaUpdateService, which exists globally, to request and get new updates. * * The following state allows configuring the behavior: * * {@link allowTestOtaImages}: When set to true, we also query the Test DCL additionally to the production DCL for update * and use this when a newer version is found. Default is false * * * {@link updateCheckInterval}: By default, we check the DCL for updates every 24h. This state value allows adjusting this. * * * {@link announceAsDefaultProvider}: By default, we announce ourselves as a default update provider to all nodes in the fabric. * Set to "false" if this is not wanted and updates are only pushed on availability */ export class SoftwareUpdateManager extends Behavior { static override readonly id = "softwareupdates"; declare readonly state: SoftwareUpdateManager.State; declare internal: SoftwareUpdateManager.Internal; declare readonly events: SoftwareUpdateManager.Events; override async initialize() { const rootNode = Node.forEndpoint(this.endpoint) as ServerNode; rootNode.behaviors.require(DclBehavior); this.internal.otaService = rootNode.agentFor(this.context).get(DclBehavior).otaUpdateService; await this.internal.otaService.construction; this.internal.otaStorageScope = this.internal.otaService.storage.scope; this.reactTo(rootNode.lifecycle.online, this.#nodeOnline); this.reactTo(rootNode.lifecycle.goingOffline, this.#nodeGoingOffline); if (rootNode.lifecycle.isOnline) { await this.#nodeOnline(); } this.reactTo(this.events.announceAsDefaultProvider$Changed, this.#updateAnnouncementSettings); this.reactTo(this.events.announcementInterval$Changed, this.#updateAnnouncementSettings); } async #nodeOnline() { this.internal.suppressUpdates = false; if (this.internal.announcements !== undefined) { await this.internal.announcements.close(); this.internal.announcements = undefined; } // For now let's just provide update functionality when we are the fabric owner // In theory we could also claim that for any other fabric but could conflict with the main controller of // the fabric that then also claims being "the one provider". const fabricAuthority = await this.env.load(FabricAuthority); const ownFabric = fabricAuthority.fabrics[0]; if (!ownFabric) { // Can only happen if the SoftwareUpdateManager is used without any commissioned nodes logger.info(`No commissioned peers yet, cannot check for OTA updates. Wait for Fabric being added.`); this.reactTo(fabricAuthority.fabricAdded, this.#nodeOnline, { once: true }); return; } this.internal.announcements = new OtaAnnouncements(this.endpoint, ownFabric); this.#updateAnnouncementSettings(); // Randomly force the first update check 5-10 minutes after startup const delay = Millis(Seconds(Math.floor(Math.random() * 300)) + Minutes(5)); logger.info(`Scheduling first OTA update check in ${Duration.format(delay)}`); this.internal.checkForUpdateTimer?.stop(); this.internal.checkForUpdateTimer = this.reactorTimer( "initializeUpdateCheck", delay, this.#initializeUpdateCheck, ).start(); } async #nodeGoingOffline() { this.internal.suppressUpdates = true; this.internal.checkForUpdateTimer?.stop(); // Announcements run on their own timers and write to peers, so the suppression flag alone does not stop them. // #nodeOnline installs a fresh instance. await this.internal.announcements?.close(); this.internal.announcements = undefined; // Bound to the ClientSubscriptions the stopping runtime closes; the getter creates a new armer for the next run this.internal.rebootResubscribeArmer?.[Symbol.dispose](); this.internal.rebootResubscribeArmer = undefined; } #updateAnnouncementSettings() { if (this.internal.announcements === undefined) { return; } if (this.state.announceAsDefaultProvider) { this.internal.announcements.interval = this.state.announcementInterval; } else { this.internal.announcements.interval = undefined; } } /** * Trigger the first check for updates some time after the node went online, and register a periodic interval * afterward. */ async #initializeUpdateCheck() { try { await this.checkForUpdates(); } catch (error) { logger.notice(`Error during initial OTA update check:`, error); } this.internal.checkForUpdateTimer.stop(); this.internal.checkForUpdateTimer = this.periodicReactorTimer( "checkAvailableUpdates", this.state.updateCheckInterval, this.checkForUpdates, ).start(); } get storage() { return this.internal.otaService.storage; } get #rebootResubscribeArmer() { if (this.internal.rebootResubscribeArmer === undefined) { this.internal.rebootResubscribeArmer = new RebootResubscribeArmer( this.env.get(SessionManager), this.env.get(ClientSubscriptions), ); } return this.internal.rebootResubscribeArmer; } /** * Returns a snapshot of the current update queue for introspection. */ get queuedUpdates(): PendingUpdateInfo[] { const now = Time.nowMs; return this.internal.updateQueue.map(entry => { const bdxSession = this.#bdxSessionFor(entry.peerAddress); let status: PendingUpdateInfo["status"]; if (entry.lastProgressUpdateTime === undefined) { status = "queued"; } else if ( entry.lastProgressUpdateTime + (entry.applyDelay ?? 0) + OTA_PROGRESS_TIMEOUT < now && !this.#isTransferring(entry, bdxSession?.transferredBytes) ) { status = "stalled"; } else { status = "in-progress"; } return { peerAddress: entry.peerAddress, vendorId: entry.vendorId, productId: entry.productId, targetSoftwareVersion: entry.targetSoftwareVersion, status, lastProgressStatus: entry.lastProgressStatus, lastProgressUpdateTime: entry.lastProgressUpdateTime, bdxTransferredBytes: status === "in-progress" ? bdxSession?.transferredBytes : undefined, bdxDataLength: status === "in-progress" ? bdxSession?.dataLength : undefined, }; }); } #bdxSessionFor(peerAddress: PeerAddress) { const { otaStorageScope } = this.internal; if (otaStorageScope === undefined || !this.env.has(BdxProtocol)) { return undefined; } return this.env.get(BdxProtocol).sessionFor(peerAddress, otaStorageScope); } /** * A retry of a failed download opens a fresh BDX session whose counter restarts below the one we recorded, so any * change of the counter counts as liveness, not only growth. */ #isTransferring(entry: UpdateQueueEntry, transferredBytes?: number) { return transferredBytes !== undefined && transferredBytes !== (entry.lastBdxTransferredBytes ?? 0); } /** * A download produces no OTA status transitions for as long as it runs, so a large image on a slow link looks * stalled to a timestamp that only status changes refresh. */ #refreshBdxProgress(entry: UpdateQueueEntry, now: Timestamp) { const transferredBytes = this.#bdxSessionFor(entry.peerAddress)?.transferredBytes; if (!this.#isTransferring(entry, transferredBytes)) { return false; } entry.lastBdxTransferredBytes = transferredBytes; entry.lastProgressUpdateTime = now; return true; } #resetEntryProgress(entry: UpdateQueueEntry) { entry.lastProgressUpdateTime = undefined; entry.lastProgressStatus = OtaUpdateStatus.Unknown; entry.lastBdxTransferredBytes = undefined; entry.applyDelay = undefined; } /** Validate that we know the peer the update is requested for and the details match to what we know */ async #validatePeerDetails( peerAddress: PeerAddress, details: { softwareVersion: number; vendorId: VendorId; productId: number }, ) { const { softwareVersion, vendorId, productId } = details; const peers = (Node.forEndpoint(this.endpoint) as ServerNode).peers; const node = peers.get(peerAddress); const basicInfo = node?.maybeStateOf(BasicInformationClient); if ( basicInfo?.softwareVersion === softwareVersion && basicInfo?.vendorId === vendorId && basicInfo?.productId === productId ) { return true; } return false; } /** * Used to determine if an update is existing in our storage for a peer with a certain software version. * * It uses the already checked details and does not check again on-demand. It considers consents already given, * but validates the peer data. */ async updateExistsFor( peerAddress: PeerAddress, { softwareVersion, vendorId, productId, requestorCanConsent }: OtaSoftwareUpdateProvider.QueryImageRequest, ): Promise { if (!(await this.#validatePeerDetails(peerAddress, { softwareVersion, vendorId, productId }))) { logger.info( `Peer details for node ${peerAddress.toString()} do not match values the update was requested for, ignoring`, ); return undefined; } // We just use what we know from our last check here, so no on-demand DCL checks for now const availableUpdates = await this.internal.otaService.find({ vendorId, productId }); const candidates = availableUpdates .filter( ({ softwareVersion: candidateSoftwareVersion, minApplicableSoftwareVersion = 0, maxApplicableSoftwareVersion = softwareVersion, mode, }) => softwareVersion < candidateSoftwareVersion && softwareVersion >= minApplicableSoftwareVersion && softwareVersion <= maxApplicableSoftwareVersion && (mode === "prod" || this.state.allowTestOtaImages), ) .sort((a, b) => b.softwareVersion - a.softwareVersion); if (candidates.length === 0) { return undefined; } const candidatesWithConsent: UpdateConsentEntry[] = candidates.map(candidate => { const consent = this.internal.consents .filter(consent => consent.vendorId === candidate.vendorId && consent.productId === candidate.productId) .filter(consent => consent.targetSoftwareVersion >= candidate.softwareVersion); // Consent for this or higher version applies return { ...candidate, consentPeers: consent.map(c => c.peerAddress), }; }); // We have a candidate list, now see about consent let candidate: UpdateConsentEntry | undefined; let consentRequired = false; if (requestorCanConsent) { // Ok, the request can get consent himself, so we just use the first candidate version and determine if // consent is needed to be requested or if we already have it candidate = candidatesWithConsent[0]; consentRequired = !candidate.consentPeers.some( ({ fabricIndex, nodeId }) => fabricIndex === peerAddress.fabricIndex && nodeId === peerAddress.nodeId, ); } else { // Find the first candidate where we have consent already, if any candidate = candidatesWithConsent.find(({ consentPeers }) => consentPeers.some( ({ fabricIndex, nodeId }) => fabricIndex === peerAddress.fabricIndex && nodeId === peerAddress.nodeId, ), ); } if (candidate) { const { softwareVersion, softwareVersionString, filename } = candidate; return { newSoftwareVersion: softwareVersion, newSoftwareVersionString: softwareVersionString, fileDesignator: new FileDesignator(`ota/${filename}`), // We do not provide any httpsUri here, as we expect the node to download from our DCL proxy via BDX consentRequired, }; } else { logger.info( `No update candidate found for node ${peerAddress.nodeId} in fabric ${peerAddress.fabricIndex} for any update`, ); } } /** * Check all commissioned peers for available updates and purge stored updates that are no longer needed. * * Runs automatically on a periodic timer but may also be invoked directly to force an immediate check. */ async checkForUpdates() { await this.queryUpdates({ includeStoredUpdates: true }); await this.#cleanupObsoleteUpdates(); } /** * Clean up stored OTA files that no node in the system needs anymore. * A stored version is obsolete when ALL nodes with that vendor/product ID are already at or above that version, * or when its mode is no longer permitted by the current settings (e.g. test-mode files after * {@link allowTestOtaImages} is turned off). Only cleans "prod" and "test" mode files; * "local" files are user-managed. */ async #cleanupObsoleteUpdates() { const rootNode = Node.forEndpoint(this.endpoint) as ServerNode; // Collect minimum software version per vendor/product across all nodes const nodeVersions = new Map(); for (const peer of rootNode.peers) { const basicInfo = peer.maybeStateOf(BasicInformationClient); if (basicInfo === undefined) { continue; } const { vendorId, productId, softwareVersion } = basicInfo; const key = `${vendorId}-${productId}`; const existing = nodeVersions.get(key); if (existing === undefined || softwareVersion < existing) { nodeVersions.set(key, softwareVersion); } } // Nothing can be cleaned up when there are no peers and test-mode files are still permitted — // skip the storage scan entirely. const purgeDisallowedTestFiles = !this.state.allowTestOtaImages; if (nodeVersions.size === 0 && !purgeDisallowedTestFiles) { return; } // Narrow the scan when only the test-purge path is active (no peers to evaluate obsolescence against). const storedUpdates = await this.internal.otaService.find(nodeVersions.size === 0 ? { mode: "test" } : {}); for (const update of storedUpdates) { if (update.mode === "local") { continue; // Never auto-delete user-added files } // Purge test-mode files when test OTA usage is off — they cannot be offered and would // otherwise pile up (e.g. after a user disables "Enable test-net DCL usage"). if (update.mode === "test" && purgeDisallowedTestFiles) { try { await this.internal.otaService.delete({ filename: update.filename }); logger.info(`Cleaned up test-mode OTA file ${update.filename} (allowTestOtaImages is disabled)`); } catch (error) { logger.warn(`Failed to clean up OTA file ${update.filename}:`, error); } continue; } if (nodeVersions.size === 0) { continue; } const key = `${update.vendorId}-${update.productId}`; const minNodeVersion = nodeVersions.get(key); if (minNodeVersion === undefined) { continue; // No known nodes for this vid/pid, keep the file } // If all nodes are at or above this stored version, it's obsolete if (update.softwareVersion <= minNodeVersion) { try { await this.internal.otaService.delete({ filename: update.filename }); logger.info( `Cleaned up obsolete OTA file ${update.filename} (all nodes at version >= ${minNodeVersion})`, ); } catch (error) { logger.warn(`Failed to clean up OTA file ${update.filename}:`, error); } } } } /** * Checks all nodes, or optionally a defined one, for available updates from the DCL OTA Update Service. * * Returns a list of peers for which updates are available along with the collected update info. * * If `includeStoredUpdates` is set to true, locally stored/imported OTA images are considered as update candidates * alongside the DCL. The DCL is always queried regardless, unless the OTA service is offline; across all candidates * the highest applicable and allowed version wins. */ async queryUpdates(options: { peerToCheck?: ClientNode; includeStoredUpdates?: boolean } = {}) { const { peerToCheck, includeStoredUpdates = false } = options; const rootNode = Node.forEndpoint(this.endpoint) as ServerNode; // Collect all client nodes and their versions, so we only need to check each version once const updateDetails = new Map(); for (const peer of rootNode.peers) { if (this.internal.suppressUpdates) { return []; } if (peerToCheck !== undefined && peerToCheck !== peer) { continue; } const details = this.#preparePeerForUpdate(peer); if (details === undefined) { continue; } const { otaEndpoint, peerAddress, vendorId, productId, softwareVersion, softwareVersionString } = details; const key = `${vendorId}-${productId}-${softwareVersion}`; const versionData = updateDetails.get(key) ?? { vendorId, productId, softwareVersion, softwareVersionString, otaEndpoints: new Set<{ endpoint: Endpoint; peerAddress: PeerAddress }>(), }; versionData.otaEndpoints.add({ endpoint: otaEndpoint, peerAddress }); updateDetails.set(key, versionData); } if (updateDetails.size === 0) { logger.info(`No updatable nodes found`); return []; } const peersWithUpdates = new Array<{ peerAddress: PeerAddress; info: SoftwareUpdateInfo }>(); for (const infos of updateDetails.values()) { if (this.internal.suppressUpdates) { return []; } try { const peers = await this.#checkProductForUpdates(infos, includeStoredUpdates); for (const peer of peers) { const info = this.internal.knownUpdates.get(peer); if (info === undefined) { continue; // Race condition should normally not happen } peersWithUpdates.push({ peerAddress: peer, info }); } } catch (error) { // We ignore any errors that occur, but log them logger.warn( `Error while checking for updates for product ${infos.vendorId}-${infos.productId}:`, error, ); } } return peersWithUpdates; } /** * Check storage and DCL for updates for the given product/version, downloads it, and notify nodes if an update was * found */ async #checkProductForUpdates(infos: CollectedNodesUpdateInfo, includeStoredUpdates: boolean) { const { vendorId, productId, softwareVersion, otaEndpoints } = infos; const updateDetails = await this.internal.otaService.checkForUpdate({ vendorId, productId, currentSoftwareVersion: softwareVersion, includeStoredUpdates, isProduction: this.state.allowTestOtaImages ? undefined : true, }); if (!updateDetails || this.internal.suppressUpdates) { return []; } const fd = await this.internal.otaService.downloadUpdate(updateDetails); logger.info( "OTA update available for", Diagnostic.dict({ vendorId, productId, softwareVersion, file: fd.text, peers: [...otaEndpoints.values()].map(({ peerAddress }) => peerAddress.toString()), }), ); // Request consent or notify peers about the update if we have consent for (const { endpoint, peerAddress } of otaEndpoints) { const { vid, pid, softwareVersion, softwareVersionString, releaseNotesUrl, specificationVersion, source } = updateDetails; const details: SoftwareUpdateInfo = { vendorId: vid, productId: pid, softwareVersion, softwareVersionString, releaseNotesUrl, specificationVersion, source, }; this.internal.knownUpdates.set(peerAddress, details); const consent = this.internal.consents.find( consent => consent.vendorId === vendorId && consent.productId === productId && consent.targetSoftwareVersion === updateDetails.softwareVersion && consent.peerAddress.fabricIndex === peerAddress.fabricIndex && consent.peerAddress.nodeId === peerAddress.nodeId, ); if (this.internal.suppressUpdates) { break; } if (consent !== undefined) { // We already have a consent for this update, so just announce the provider this.#queueUpdate({ endpoint, vendorId, productId, targetSoftwareVersion: updateDetails.softwareVersion, maxBdxBlockSize: consent.maxBdxBlockSize, bdxAdditionalMrpDelay: consent.bdxAdditionalMrpDelay, peerAddress, }); } else { // Inform the application that an update is available and we need consent this.requestConsentForUpdate(peerAddress, details); } } return [...otaEndpoints.values()].map(({ peerAddress }) => peerAddress); } /** * Determine if we can request an update for the given node and return node meta-data needed for the process. * The Node needs to be commissioned (have a peer address), being not disabled, and having an active subscription * (to not work with outdated data). * When a node is applicable for updates, it also subscribes to softwareVersion changes to be able to react */ #preparePeerForUpdate(peer: ClientNode) { const otaData = this.internal.announcements?.peerApplicableForUpdate(peer); if (otaData === undefined) { // We need more information on the node to request an update return; } const { peerAddress, otaEndpoint } = otaData; const { vendorId, productId, softwareVersion, softwareVersionString } = peer.stateOf(BasicInformationClient); // Todo sort out test vendors? // Node is applicable for update checks, register listener on softwareVersion to allow resetting update state const versionEvent = peer.eventsOf(BasicInformationClient).softwareVersion$Changed; if (!this.internal.versionUpdateObservers.observes(versionEvent)) { const that = this; function triggerVersionChange(newVersion: number) { that.#onSoftwareVersionChanged(peerAddress, newVersion); } this.internal.versionUpdateObservers.on(versionEvent, this.callback(triggerVersionChange)); } // Listen to startUp event to detect reboots after applying — if the device reboots with the old version // softwareVersion$Changed won't fire (same value), so we feed the version through the same handler const startUpEvent = peer.eventsOf(BasicInformationClient).startUp; if (!this.internal.versionUpdateObservers.observes(startUpEvent)) { const that = this; function triggerStartUp({ softwareVersion }: { softwareVersion: number }, context?: ActionContext) { that.#onSoftwareVersionChanged( peerAddress, softwareVersion, true, hasRemoteActor(context) ? context.session : undefined, ); } this.internal.versionUpdateObservers.on(startUpEvent, this.callback(triggerStartUp)); } // A device that applies without restarting sends neither a version change nor startUp; the controller's // resubscription then reports the version it still runs const subscriptionEvent = peer.eventsOf(NetworkClient).subscriptionStatusChanged; if (!this.internal.versionUpdateObservers.observes(subscriptionEvent)) { const that = this; function triggerSubscribed(isActive: boolean) { if (isActive) { that.#onSubscribedAfterApply(peerAddress); } } this.internal.versionUpdateObservers.on(subscriptionEvent, this.callback(triggerSubscribed)); } return { otaEndpoint, peerAddress, vendorId, productId, softwareVersion, softwareVersionString }; } /** * Handler for softwareVersion changes on a peer. * * Also called from the startUp event with `isStartUp = true`. When a device reboots after applying an update, * softwareVersion$Changed fires if the version actually changed (success). But if the device reboots with the * same version (failed apply), softwareVersion$Changed does NOT fire — so startUp feeds the version through * this same handler to detect the failure. A device that does not restart at all is caught by * {@link #onSubscribedAfterApply}. */ #onSoftwareVersionChanged( peerAddress: PeerAddress, newVersion: number, isStartUp = false, currentSession?: Session, ) { const entryIndex = this.internal.updateQueue.findIndex( e => e.peerAddress.fabricIndex === peerAddress.fabricIndex && e.peerAddress.nodeId === peerAddress.nodeId, ); if (isStartUp) { const suppressedSessionId = this.internal.pendingStartUpSuppress.get(peerAddress); if (suppressedSessionId !== undefined) { this.internal.pendingStartUpSuppress.delete(peerAddress); // Use session from event context when available (avoids extra lookup); fall back to // SessionManager for local/test contexts where there is no remote actor. const sessionId = currentSession?.id ?? this.env.get(SessionManager).maybeSessionFor(peerAddress)?.id; if (sessionId === suppressedSessionId) { // startUp is from the reboot we already handled via queryImage — suppress it return; } // Different session — process normally } } if (entryIndex === -1) { return; } const entry = this.internal.updateQueue[entryIndex]; const previousProgressStatus = entry.lastProgressStatus; if (entry.lastProgressUpdateTime !== undefined) { logger.info( `Clearing in-progress update for node ${peerAddress.toString()} due to software version change. Last State was ${OtaUpdateStatus[entry.lastProgressStatus!]}`, ); this.#resetEntryProgress(entry); } const expectedVersion = entry.targetSoftwareVersion; if (newVersion < expectedVersion) { // Device rebooted during apply with old version — the update failed if ( isStartUp && (previousProgressStatus === OtaUpdateStatus.Applying || previousProgressStatus === OtaUpdateStatus.WaitForApply) ) { logger.warn( `Device ${peerAddress.toString()} rebooted after applying update but reports softwareVersion ${newVersion} (expected >= ${expectedVersion}), update failed to apply`, ); this.#failAppliedUpdate(entryIndex); return; } logger.info( `Software version for node ${peerAddress.toString()} changed to ${newVersion}, but still below target version ${expectedVersion}, keeping in update queue`, ); return; } logger.info( `Software version changed to ${newVersion} (expected ${expectedVersion}) for node ${peerAddress.toString()}, removing from update queue`, ); this.internal.knownUpdates.delete(peerAddress); this.internal.pendingStartUpSuppress.delete(peerAddress); this.events.updateDone.emit(peerAddress); this.internal.updateQueue.splice(entryIndex, 1); // Also clean up consents this.internal.consents = this.internal.consents.filter( ({ peerAddress: consentAddress, targetSoftwareVersion }) => !PeerAddress.is(peerAddress, consentAddress) || targetSoftwareVersion > newVersion, ); this.#triggerQueuedUpdate(); } /** * Handler for a peer's subscription becoming active. * * An update in `Applying` whose peer reports a version below the target over a subscription established after the * device's restart deadline did not apply, so it is handled like a reboot into the old version. The deadline is * {@link RebootResubscribeArmer.returnTimeout}, after which the armer forces that resubscription. */ #onSubscribedAfterApply(peerAddress: PeerAddress) { const entryIndex = this.internal.updateQueue.findIndex( e => e.peerAddress.fabricIndex === peerAddress.fabricIndex && e.peerAddress.nodeId === peerAddress.nodeId, ); const softwareVersion = (Node.forEndpoint(this.endpoint) as ServerNode).peers .get(peerAddress) ?.maybeStateOf(BasicInformationClient)?.softwareVersion; if (entryIndex === -1 || softwareVersion === undefined) { return; } const { lastProgressStatus, lastProgressUpdateTime, applyDelay, targetSoftwareVersion } = this.internal.updateQueue[entryIndex]; if ( lastProgressStatus !== OtaUpdateStatus.Applying || lastProgressUpdateTime === undefined || Time.nowMs < lastProgressUpdateTime + RebootResubscribeArmer.returnTimeout(applyDelay) || softwareVersion >= targetSoftwareVersion ) { return; } logger.warn( `Device ${peerAddress.toString()} resubscribed after applying update but reports softwareVersion ${softwareVersion} (expected >= ${targetSoftwareVersion}), update failed to apply`, ); this.#failAppliedUpdate(entryIndex); } #failAppliedUpdate(entryIndex: number) { const [{ peerAddress }] = this.internal.updateQueue.splice(entryIndex, 1); this.internal.pendingStartUpSuppress.delete(peerAddress); this.events.updateFailed.emit(peerAddress); this.#triggerQueuedUpdate(); } /** * Notify the application that consent is needed for the given update on the given peer */ protected requestConsentForUpdate(peerAddress: PeerAddress, updateDetails: SoftwareUpdateInfo) { this.events.updateAvailable.emit(peerAddress, updateDetails); } /** * Add an update to the update queue and execute it. */ #queueUpdate(entry: UpdateQueueEntry) { // Check for existing entry for this peer to avoid duplicates const existing = this.internal.updateQueue.find( e => e.peerAddress.fabricIndex === entry.peerAddress.fabricIndex && e.peerAddress.nodeId === entry.peerAddress.nodeId, ); if (existing !== undefined) { if (existing.lastProgressUpdateTime !== undefined) { // Update is in progress — don't disrupt it; the new version will be picked up on the next cycle logger.info( `Update for node ${entry.peerAddress.toString()} already in progress, skipping queue update`, ); return; } // Not yet started — safe to update the queued entry existing.vendorId = entry.vendorId; existing.productId = entry.productId; existing.targetSoftwareVersion = entry.targetSoftwareVersion; existing.endpoint = entry.endpoint; existing.maxBdxBlockSize = entry.maxBdxBlockSize; existing.bdxAdditionalMrpDelay = entry.bdxAdditionalMrpDelay; logger.info(`Updated existing queued update for node ${entry.peerAddress.toString()}`); } else { logger.info(`Queuing update consent for node ${entry.peerAddress.toString()}`); this.internal.updateQueue.push(entry); } this.#triggerQueuedUpdate(); } /** * Triggers updates for queued entries in the update queue, ensuring proper prioritization and handling of stalled * updates. This method checks for ongoing updates, resets states for stalled updates, and processes the next entry * in the queue if no update is currently in progress. * * The queue is re-sorted to prioritize entries with an unknown status, and a periodic timer is activated to * monitor for stalled updates. */ #triggerQueuedUpdate() { if (this.internal.suppressUpdates) { return; } const now = Time.nowMs; const inProgressEntries = this.internal.updateQueue.filter( ({ lastProgressUpdateTime }) => lastProgressUpdateTime !== undefined, ); let inProgressCount = inProgressEntries.length; if (inProgressCount === 0 && this.internal.updateQueueTimer?.isRunning) { // No update in progress, stop timer this.internal.updateQueueTimer.stop(); } else if (inProgressCount > 0) { for (const entry of inProgressEntries) { if (this.#refreshBdxProgress(entry, now)) { continue; } if (entry.lastProgressUpdateTime! + (entry.applyDelay ?? 0) + OTA_PROGRESS_TIMEOUT < now) { logger.info( `Resetting stalled OTA update state for node ${entry.peerAddress.toString()} due to inactivity`, ); this.#resetEntryProgress(entry); inProgressCount--; } } } if (inProgressCount > 0) { // An update is already in progress return; } // Re-sort queue to deprioritize entries with Unknown status (sort them last). Entries without a // lastProgressStatus (undefined — never started) go first, while Unknown entries go last. Unknown // arises when a BDX-cancelled entry is reset for retry (Fix 1); deprioritizing it lets any // genuinely fresh entries go first before the retry is attempted. this.internal.updateQueue = this.internal.updateQueue.sort((a, b) => { const aStatus = a.lastProgressStatus === OtaUpdateStatus.Unknown ? 1 : 0; const bStatus = b.lastProgressStatus === OtaUpdateStatus.Unknown ? 1 : 0; return aStatus - bStatus; }); const nextEntry = this.internal.updateQueue[0]; if (nextEntry) { this.#triggerUpdateOnNode(nextEntry).catch(error => { logger.warn(`Error while triggering OTA update on node ${nextEntry.peerAddress.toString()}:`, error); }); if (!this.internal.updateQueueTimer?.isRunning) { // Start a periodic timer to check for stalled updates this.internal.updateQueueTimer = this.periodicReactorTimer( "checkQueuedUpdates", Minutes(5), this.#triggerQueuedUpdate, ).start(); } } } /** * Triggers an update on the given node. This method announces ourselves as OTA Provider to the node with an * available update and then triggers the update on the node. * The node usually calls queryImage as a result of this when it processes the announcement. */ async #triggerUpdateOnNode(entry: UpdateQueueEntry) { if (this.internal.announcements === undefined) { logger.info(`Not yet initialized with peers, can not trigger update on node`, entry.peerAddress); return; } const { endpoint, peerAddress } = entry; // Guard: skip the announcement if a BDX session is already active for this peer if (this.#bdxSessionFor(peerAddress) !== undefined) { logger.info(`BDX transfer still active for node ${peerAddress.toString()}, skipping announcement`); return; } try { logger.info(`Announcing OTA provider to node ${peerAddress.toString()}`); await this.internal.announcements.announceOtaProvider( endpoint, peerAddress, OtaSoftwareUpdateRequestor.AnnouncementReason.UpdateAvailable, ); // Only mark in-progress if the entry still exists in the queue (it may have been canceled during the await) if (this.internal.updateQueue.indexOf(entry) >= 0) { entry.lastProgressUpdateTime = Time.nowMs; } } catch (error) { logger.warn(`Failed to announce OTA provider to node ${peerAddress.toString()}:`, error); // Reset entry state so the queue doesn't stay blocked this.#resetEntryProgress(entry); this.#triggerQueuedUpdate(); } } /** * Forces an OTA update for a specific node identified by its peer address, vendor ID, product ID, * and target software version. This method will override any ongoing updates by explicitly adding an update * consent for the specified node and processing the update immediately if applicable. * This can be used when an exact timing of the update is needed. When the update can be executed in a delayed/queued * manner, please use `addUpdateConsent()`. */ async forceUpdate(peerAddress: PeerAddress, target: SoftwareUpdateManager.UpdateTarget) { const { vendorId, productId, targetSoftwareVersion } = target; assertTransferOptions(target); peerAddress = PeerAddress(peerAddress); const existingEntry = this.internal.updateQueue.find( e => e.peerAddress.fabricIndex === peerAddress.fabricIndex && e.peerAddress.nodeId === peerAddress.nodeId && e.lastProgressUpdateTime !== undefined, ); if (existingEntry !== undefined) { // Check whether a BDX session is still active for this peer const bdxProtocol = this.env.get(BdxProtocol); const activeSession = this.#bdxSessionFor(peerAddress); if (activeSession !== undefined) { // Update is legitimately in progress, don't disrupt it logger.info( `Force update for node ${peerAddress.toString()} skipped: BDX transfer is actively in progress`, ); return; } // No active BDX session — entry is stale, clean it up logger.info(`Cleaning up stale update entry for node ${peerAddress.toString()} before forcing new update`); const staleIndex = this.internal.updateQueue.indexOf(existingEntry); if (staleIndex >= 0) { this.internal.updateQueue.splice(staleIndex, 1); this.internal.pendingStartUpSuppress.delete(peerAddress); } try { await bdxProtocol.disablePeerForScope(peerAddress, this.storage, true); } catch (error) { MatterError.accept(error); logger.debug(`Error cleaning up stale BDX registration:`, error); } } const added = await this.addUpdateConsent(peerAddress, target); if (!added) { throw new OtaUpdateError(`Node at ${peerAddress.toString()} is not currently applicable for OTA updates`); } const entry = this.internal.updateQueue.find( e => e.peerAddress.fabricIndex === peerAddress.fabricIndex && e.peerAddress.nodeId === peerAddress.nodeId && e.vendorId === vendorId && e.productId === productId && e.targetSoftwareVersion === targetSoftwareVersion, ); if (!entry) { throw new InternalError(`Failed to find queued update after adding consent`); // Should not happen } if (this.internal.updateQueue.length === 1) { return; // Only entry, already being processed } await this.#triggerUpdateOnNode(entry); } /** Tries to cancel an ongoing OTA update for the given peer address. */ async #cancelUpdate(peerAddress: PeerAddress) { peerAddress = PeerAddress(peerAddress); const bdxProtocol = this.env.get(BdxProtocol); // Disable the Peer on BdxProtocol for the OTA scope if registered and also cancel an open BDX session, if any await bdxProtocol.disablePeerForScope(peerAddress, this.storage, true); const entryIndex = this.internal.updateQueue.findIndex( e => e.peerAddress.fabricIndex === peerAddress.fabricIndex && e.peerAddress.nodeId === peerAddress.nodeId, ); if (entryIndex < 0) { logger.warn(`No Ota update queued for node ${peerAddress.toString()}`); return; } const entry = this.internal.updateQueue[entryIndex]; if ( entry.lastProgressStatus === OtaUpdateStatus.Applying || entry.lastProgressStatus === OtaUpdateStatus.Done ) { // Too late, update is already applying or done logger.info(`Cannot cancel update for node ${peerAddress.toString()}, already applying or done`); return; } this.internal.updateQueue.splice(entryIndex, 1); // Disarm only on an actual cancel — a too-late (already Applying/Done) cancel returns above and must keep // the peer armed so the imminent reboot still gets fast re-subscription. this.internal.rebootResubscribeArmer?.disarm(peerAddress); // Clean up any pending startUp suppression for this peer — it will never arrive if the peer // is permanently removed from tracking (decommissioned / disconnected). this.internal.pendingStartUpSuppress.delete(peerAddress); logger.info(`Cancelled OTA update for node ${peerAddress.toString()}`); this.events.updateFailed.emit(peerAddress); this.#triggerQueuedUpdate(); } /** * How BDX should carry this peer's pending update: the options given with its consent, else the general defaults * in {@link State}. An absent value leaves the decision to the peer and its medium. * * A consent may exist before the update is queued, so both are consulted. */ bdxTransferOptionsFor(peerAddress: PeerAddress) { peerAddress = PeerAddress(peerAddress); const matches = (candidate: PeerAddress) => candidate.fabricIndex === peerAddress.fabricIndex && candidate.nodeId === peerAddress.nodeId; const pending = this.internal.updateQueue.find(entry => matches(entry.peerAddress)) ?? this.internal.consents.find(consent => matches(consent.peerAddress)); return { maxBlockSize: pending?.maxBdxBlockSize ?? this.state.maxBdxBlockSize, additionalMrpDelay: pending?.bdxAdditionalMrpDelay ?? this.state.bdxAdditionalMrpDelay, }; } /** * Adds or updates a consent for a given peer address, vendor ID, product ID, and target software version. * Filters out existing consents for the given peer address and replaces them with the new one. * If the node associated with the peer address is applicable for an update, it schedules the update to happen with * the next queue slot, so potentially delayed. * This can be used when the update can be executed in a delayed/queued manner and it does not matter exactly when. */ async addUpdateConsent(peerAddress: PeerAddress, target: SoftwareUpdateManager.UpdateTarget) { const { vendorId, productId, targetSoftwareVersion, maxBdxBlockSize, bdxAdditionalMrpDelay } = target; assertTransferOptions(target); peerAddress = PeerAddress(peerAddress); // Filter out all existing consents for this peer, they are replaced by the new one const consents = this.internal.consents.filter( consent => consent.peerAddress.fabricIndex !== peerAddress.fabricIndex || consent.peerAddress.nodeId !== peerAddress.nodeId, ); const node = await (Node.forEndpoint(this.endpoint) as ServerNode).peers.forAddress(peerAddress); const { otaEndpoint, vendorId: nodeVendorId, productId: nodeProductId, } = this.#preparePeerForUpdate(node) ?? {}; if (otaEndpoint !== undefined && (nodeVendorId !== vendorId || nodeProductId !== productId)) { throw new ImplementationError(`Node at ${peerAddress.toString()} does not match given vendorId/productId`); } consents.push({ vendorId, productId, targetSoftwareVersion, maxBdxBlockSize, bdxAdditionalMrpDelay, peerAddress, }); this.internal.consents = consents; if (otaEndpoint === undefined) { logger.info(`Node ${peerAddress.toString()} is currently not applicable for OTA updates, try again later`); return false; } this.#queueUpdate({ vendorId, productId, targetSoftwareVersion, maxBdxBlockSize, bdxAdditionalMrpDelay, peerAddress, endpoint: otaEndpoint, }); return true; } /** * Checks if consent exists for the given peer address and optionally for a specific target software version. */ hasConsent(peerAddress: PeerAddress, targetSoftwareVersion?: number) { return this.internal.consents.some( consent => consent.peerAddress.fabricIndex === peerAddress.fabricIndex && consent.peerAddress.nodeId === peerAddress.nodeId && (targetSoftwareVersion === undefined || consent.targetSoftwareVersion === targetSoftwareVersion), ); } /** * Checks for consent and removes it if present, also cancels if in progress. Use this to remove a formerly given * consent. */ removeConsent(peerAddress: PeerAddress, targetSoftwareVersion?: number) { const consentIndex = this.internal.consents.findIndex( consent => consent.peerAddress.fabricIndex === peerAddress.fabricIndex && consent.peerAddress.nodeId === peerAddress.nodeId && (targetSoftwareVersion === undefined || consent.targetSoftwareVersion === targetSoftwareVersion), ); if (consentIndex >= 0) { this.internal.consents.splice(consentIndex, 1); } else { logger.info( `No consent to remove found for node ${peerAddress.toString()}${targetSoftwareVersion !== undefined ? ` for version ${targetSoftwareVersion}` : ""}`, ); } return this.#cancelUpdate(peerAddress); } /** * Suppress the next startUp event from the given peer on the given session. * Called by OtaSoftwareUpdateProviderServer when a queryImage is detected as a post-reboot query * so the startUp event that follows is not treated as a new failure-after-apply. */ suppressNextStartUp(peerAddress: PeerAddress, sessionId: number) { this.internal.pendingStartUpSuppress.set(peerAddress, sessionId); } /** * Handles the status change of an OTA update for a given peer device. * * This method processes OTA update status notifications received from a specified device. * Based on the status, it updates the internal state of the update queue, logs relevant * messages, and triggers the necessary events. * * With `Applying`, `applyDelay` is the `DelayedActionTime` the provider allowed the apply with, which the device * waits out before it applies and restarts. */ onOtaStatusChange(peerAddress: PeerAddress, status: OtaUpdateStatus, toVersion?: number, applyDelay?: Duration) { if (this.internal.suppressUpdates) { // A post-dispose Applying would otherwise re-arm the already-disposed armer whose observers are // closed, leaving an entry that never gets a grace timer nor self-cleans. return; } peerAddress = PeerAddress(peerAddress); const entryIndex = this.internal.updateQueue.findIndex( e => e.peerAddress.fabricIndex === peerAddress.fabricIndex && e.peerAddress.nodeId === peerAddress.nodeId, ); if (entryIndex < 0) { if (status !== OtaUpdateStatus.Cancelled) { logger.warn( `Received OTA status update from unknown node ${peerAddress.toString()}, status=${OtaUpdateStatus[status]}`, ); } return; } const entry = this.internal.updateQueue[entryIndex]; if (status === OtaUpdateStatus.Done) { logger.info(`OTA update completed for node`, peerAddress.toString()); const completedVersion = toVersion ?? entry.targetSoftwareVersion; this.internal.updateQueue.splice(entryIndex, 1); this.internal.knownUpdates.delete(peerAddress); this.internal.pendingStartUpSuppress.delete(peerAddress); this.events.updateDone.emit(peerAddress); // Also clean up consents (mirrors the software-version-change path, which is not reached when the Done // status arrives directly instead of via a subsequent version-change event) this.internal.consents = this.internal.consents.filter( ({ peerAddress: consentAddress, targetSoftwareVersion }) => !PeerAddress.is(peerAddress, consentAddress) || targetSoftwareVersion > completedVersion, ); this.#triggerQueuedUpdate(); } else if (status === OtaUpdateStatus.Cancelled) { // BDX transport interruption: reset the entry so it can be retried. // The intentional cancel path (#cancelUpdate) handles removal and updateFailed independently. logger.info(`OTA update BDX cancelled for node ${peerAddress.toString()}, resetting for retry`); this.internal.rebootResubscribeArmer?.disarm(peerAddress); this.#resetEntryProgress(entry); this.#triggerQueuedUpdate(); } else { logger.info( `OTA update status for node ${peerAddress.toString()} changed to ${OtaUpdateStatus[status]}${ toVersion !== undefined ? ` for version ${toVersion}` : "" }`, ); const wasApplying = entry.lastProgressStatus === OtaUpdateStatus.Applying; entry.lastProgressUpdateTime = Time.nowMs; entry.lastProgressStatus = status; entry.applyDelay = status === OtaUpdateStatus.Applying ? applyDelay : undefined; // Once the device asked for (and we allowed) the final apply, it will reboot and return — arm fast // re-subscription so a non-persistent device is refreshed within the grace window rather than after the // full subscription timeout. notifyUpdateApplied (Done) must NOT disarm this — the subscription may // still be stale then; the armer resolves purely on new-session + data-flow. if (status === OtaUpdateStatus.Applying) { this.#rebootResubscribeArmer.arm(peerAddress, applyDelay); // A retransmitted ApplyUpdateRequest re-enters the provider's handler and reports Applying // again; the event says the peer reached this point, not how often it asked. if (!wasApplying) { this.events.updateApplying.emit(peerAddress); } } } } override async [Symbol.asyncDispose]() { this.internal.suppressUpdates = true; await this.internal.announcements?.close(); this.internal.versionUpdateObservers.close(); this.internal.rebootResubscribeArmer?.[Symbol.dispose](); await super[Symbol.asyncDispose]?.(); } } export namespace SoftwareUpdateManager { export class State { /** Set this to true to also allow updates from the Test DCL */ allowTestOtaImages = false; /** Default Update check Interval */ updateCheckInterval = Hours(24); /** Announce this controller as Update provider to all nodes */ announceAsDefaultProvider = false; /** Interval to Announces this controller as Update provider. Must not be lower than 24h, `Instant` disables. */ announcementInterval = Hours(24); /** * Caps the BDX block size offered for OTA transfers. Unset accepts whatever the peer requests, which is the * transport maximum. * * A smaller block multiplies round trips, and each of those is bounded by the peer's BDX response budget, so * lower this only for a peer that cannot handle full-size blocks. */ maxBdxBlockSize?: number = undefined; /** * Additive MRP retransmission margin for OTA transfers, overriding the margin the peer's medium implies. * * A BDX retransmission interval is capped so the whole schedule fits the peer's response budget, so values * above roughly 7s are inert. */ bdxAdditionalMrpDelay?: Duration = undefined; } /** Identifies the update a consent applies to, and how to transfer it. */ export interface UpdateTarget { vendorId: VendorId; productId: number; targetSoftwareVersion: number; /** * Caps the BDX block size for this transfer, overriding {@link State.maxBdxBlockSize}. Unset falls back to * that default. */ maxBdxBlockSize?: number; /** * Additive MRP retransmission margin for this transfer, overriding {@link State.bdxAdditionalMrpDelay} and the * margin the peer's medium implies. Unset falls back to that default. */ bdxAdditionalMrpDelay?: Duration; } export class Internal { /** Use this to pre-initialize consent to allow nodes to update automatically. The content will not be persisted! */ consents = new Array(); otaService!: DclOtaUpdateService; /** Scope of the OTA storage, captured so a lookup cannot depend on the service still being constructed. */ otaStorageScope?: StorageScope; checkForUpdateTimer!: Timer; updateQueue = new Array(); updateQueueTimer?: Timer; announcements?: OtaAnnouncements; versionUpdateObservers = new ObserverGroup(); knownUpdates = new PeerAddressMap(); /** * Peer → session ID on which the reboot-triggering queryImage was received. * * Intentionally ephemeral (not persisted): the suppress window is sub-second — between the * queryImage response and the subsequent startUp event that follows on the same session. * Entries are consumed immediately when the matching startUp fires in #onSoftwareVersionChanged, * or cleaned up in #cancelUpdate when the peer is permanently removed from tracking. */ pendingStartUpSuppress = new PeerAddressMap(); rebootResubscribeArmer?: RebootResubscribeArmer; /** Raised whenever the node is not online, so update work cannot start before the first online transition. */ suppressUpdates = true; } export class Events extends EventEmitter { /** Emitted when an update is available for a Peer and there is no consent stored and contains update details */ updateAvailable = Observable<[peer: PeerAddress, updateDetails: SoftwareUpdateInfo]>(); /** * Emitted when a Peer has asked to apply the image it downloaded and this provider allowed it. * * This is the last thing a Peer needs from its provider: the download is complete, the * `ApplyUpdateRequest` is answered, and what follows — applying and reporting back — is the * Peer's own. {@link updateDone} is further off, because it waits for the `NotifyUpdateApplied` * a device sends after it has restarted. */ updateApplying = Observable<[peer: PeerAddress]>(); /** Emitted when an update for a Peer is finished */ updateDone = Observable<[peer: PeerAddress]>(); /** Emitted when an update for a Peer has failed or was cancelled */ updateFailed = Observable<[peer: PeerAddress]>(); announceAsDefaultProvider$Changed = Observable<[announceAsDefaultProvider: boolean]>(); announcementInterval$Changed = Observable<[announcementInterval: Duration]>(); } }