fix: Fix leaky Frigate subscriptions/unsubscriptions (#2513)

src/camera-manager/camera.ts

@@ -58,6 +58,7 @@ export class Camera {
   protected _capabilities?: Capabilities;
   protected _eventCallback?: CameraEventCallback;
   protected _destroyCallbacks: DestroyCallback[] = [];
+  protected _destroyed = false;
   protected _entity: Entity | null = null;
 
   constructor(
@@ -91,7 +92,7 @@ export class Camera {
       await this._getTriggerEntities(options);
       this._config.triggers.entities = uniq(this._config.triggers.entities);
 
-      // Subscribe to state based triggers.
+      // Subscribe to state based triggers (sync; no race with destroy).
       options.stateWatcher.subscribe(
         this._stateChangeHandler,
         this._config.triggers.entities,
@@ -101,14 +102,34 @@ export class Camera {
       // Subscribe to event based triggers.
       for (const event of this._config.triggers.events) {
         const request = this._buildEventSubscriptionRequest(event);
-        await options.eventWatcher.subscribe(options.hass, request);
-        this._onDestroy(() => options.eventWatcher.unsubscribe(request));
+        await this._setupSubscription(
+          () => options.eventWatcher.subscribe(options.hass, request),
+          () => options.eventWatcher.unsubscribe(request),
+        );
       }
     }
 
     return this;
   }
 
+  /**
+   * Wire up an async subscription with its teardown. Registers the unsubscribe
+   * callback synchronously before awaiting subscribe, so a destroy during the
+   * await reliably triggers cleanup; short-circuits if destroy has already
+   * run, so the cleanup callback can't fire (and enqueue an unsubscribe)
+   * before the subscribe runs.
+   */
+  protected async _setupSubscription(
+    subscribe: () => Promise<void>,
+    unsubscribe: () => void | Promise<void>,
+  ): Promise<void> {
+    if (this._destroyed) {
+      return;
+    }
+    this._onDestroy(unsubscribe);
+    await subscribe();
+  }
+
   private _buildEventSubscriptionRequest(event: TriggerEvent): EventSubscriptionRequest {
     const filter = event.event_data;
     return {
@@ -246,7 +267,10 @@ export class Camera {
   }
 
   public async destroy(): Promise<void> {
-    await Promise.all(this._destroyCallbacks.map((callback) => callback()));
+    this._destroyed = true;
+    const callbacks = this._destroyCallbacks;
+    this._destroyCallbacks = [];
+    await Promise.all(callbacks.map((callback) => callback()));
   }
 
   public getConfig(): CameraConfig {

src/camera-manager/frigate/camera.ts

@@ -473,8 +473,10 @@ export class FrigateCamera extends Camera {
         event.after.camera === config.frigate.camera_name,
     };
 
-    await frigateEventWatcher.subscribe(hass, request);
-    this._onDestroy(() => frigateEventWatcher.unsubscribe(request));
+    await this._setupSubscription(
+      () => frigateEventWatcher.subscribe(hass, request),
+      () => frigateEventWatcher.unsubscribe(request),
+    );
   }
 
   private _frigateEventHandler = (ev: FrigateEventChange): void => {
@@ -548,8 +550,10 @@ export class FrigateCamera extends Camera {
         review.after.camera === config.frigate.camera_name,
     };
 
-    await frigateReviewWatcher.subscribe(hass, request);
-    this._onDestroy(() => frigateReviewWatcher.unsubscribe(request));
+    await this._setupSubscription(
+      () => frigateReviewWatcher.subscribe(hass, request),
+      () => frigateReviewWatcher.unsubscribe(request),
+    );
   }
 
   private _frigateReviewHandler = (review: FrigateReviewChange): void => {

src/camera-manager/frigate/watcher.ts

@@ -1,5 +1,6 @@
 import { z } from 'zod';
-import { HomeAssistant, SubscriptionUnsubscribe } from '../../ha/types';
+import { HomeAssistant } from '../../ha/types';
+import { KeyedSubscriptionManager } from '../../utils/keyed-subscription-manager';
 import {
   FrigateEventChange,
   FrigateReviewChange,
@@ -17,53 +18,39 @@ export interface FrigateWatcherRequest<T> {
 // Generic subscription interface
 export interface FrigateWatcherSubscriptionInterface<T> {
   subscribe(hass: HomeAssistant, request: FrigateWatcherRequest<T>): Promise<void>;
-  unsubscribe(request: FrigateWatcherRequest<T>): void;
+  unsubscribe(request: FrigateWatcherRequest<T>): Promise<void>;
 }
 
 /**
- * Base class for Frigate WebSocket watchers.
- * Handles subscription management and message routing to callbacks.
+ * Base class for Frigate WebSocket watchers. Counted per `instanceID`: the
+ * first subscriber for an instance opens the WS subscription, the last to
+ * unsubscribe tears it down. Each message is parsed, schema-validated, and
+ * fanned out to every registered request whose `instanceID` matches and whose
+ * `matcher` accepts the payload.
  */
 abstract class FrigateWatcher<T> implements FrigateWatcherSubscriptionInterface<T> {
   protected abstract _type: string;
   protected abstract _schema: z.ZodType<T>;
 
-  protected _requests: FrigateWatcherRequest<T>[] = [];
-  protected _unsubscribeCallback: Record<string, SubscriptionUnsubscribe> = {};
+  private _subscriptions = new KeyedSubscriptionManager<
+    string,
+    FrigateWatcherRequest<T>
+  >((request) => request.instanceID);
 
   public async subscribe(
     hass: HomeAssistant,
     request: FrigateWatcherRequest<T>,
   ): Promise<void> {
-    const shouldSubscribe = !this._hasSubscribers(request.instanceID);
-    this._requests.push(request);
-    if (shouldSubscribe) {
-      this._unsubscribeCallback[request.instanceID] =
-        await hass.connection.subscribeMessage<string>(
-          (data) => this._receiveHandler(request.instanceID, data),
-          { type: this._type, instance_id: request.instanceID },
-        );
-    }
+    await this._subscriptions.subscribe(request, () =>
+      hass.connection.subscribeMessage<string>(
+        (data) => this._receiveHandler(request.instanceID, data),
+        { type: this._type, instance_id: request.instanceID },
+      ),
+    );
   }
 
   public async unsubscribe(request: FrigateWatcherRequest<T>): Promise<void> {
-    this._requests = this._requests.filter(
-      (existingRequest) => existingRequest !== request,
-    );
-
-    if (!this._hasSubscribers(request.instanceID)) {
-      const callback = this._unsubscribeCallback[request.instanceID];
-      delete this._unsubscribeCallback[request.instanceID];
-
-      // Callback may be undefined if unsubscribe is called while subscribe is
-      // still awaiting the Home Assistant connection.
-      await callback?.();
-    }
-  }
-
-  protected _hasSubscribers(instanceID: string): boolean {
-    return !!this._requests.filter((request) => request.instanceID === instanceID)
-      .length;
+    await this._subscriptions.unsubscribe(request);
   }
 
   protected _receiveHandler(instanceID: string, data: string): void {
@@ -82,11 +69,8 @@ abstract class FrigateWatcher<T> implements FrigateWatcherSubscriptionInterface<
       return;
     }
 
-    for (const request of this._requests) {
-      if (
-        request.instanceID === instanceID &&
-        (!request.matcher || request.matcher(parseResult.data))
-      ) {
+    for (const request of this._subscriptions.getRequestsForKey(instanceID)) {
+      if (!request.matcher || request.matcher(parseResult.data)) {
         request.callback(parseResult.data);
       }
     }

src/camera-manager/manager.ts

@@ -256,7 +256,7 @@ export class CameraManager {
     );
 
     const destroyCameras = async () => {
-      cameras.forEach((camera) => camera.destroy());
+      await allPromises(cameras, (camera) => camera.destroy());
     };
     const cameraIDs: Set<string> = new Set();
 

src/card-controller/call/ringtone.ts

@@ -12,9 +12,10 @@ import { WestminsterTone } from './tones/westminster';
 // card placed twice), all of which may independently react to the same trigger
 // state change -- with no lock, every instance would start its own AudioContext
 // and the audio would layer. First-to-start wins; subsequent `start()` calls
-// from other holders are no-ops until the active one releases via `stop()`. The
-// lock auto-recovers from a holder that forgot to release (e.g. a controller
-// GC'd without disconnect cleanup) via the `isPlaying()` sweep below.
+// from other holders are no-ops until the active one releases via `stop()`.
+// The `isPlaying()` sweep below is defensive against a future code path that
+// clears `_tone` without removing from the lock -- today every such path keeps
+// them in sync, but the sweep prevents a regression from wedging the lock.
 const sharedLock = new Set<Ringtone>();
 
 export class Ringtone {

src/card-controller/hass/event-watcher.ts

@@ -1,5 +1,6 @@
 import { HassEvent } from 'home-assistant-js-websocket';
-import { HomeAssistant, SubscriptionUnsubscribe } from '../../ha/types';
+import { HomeAssistant } from '../../ha/types';
+import { KeyedSubscriptionManager } from '../../utils/keyed-subscription-manager';
 
 export interface EventSubscriptionRequest {
   event_type: string;
@@ -23,50 +24,30 @@ export interface EventWatcherSubscriptionInterface {
  * payload.
  */
 export class EventWatcher implements EventWatcherSubscriptionInterface {
-  private _requests: EventSubscriptionRequest[] = [];
-
-  // Stored as a promise so an unsubscribe that races against an in-flight
-  // subscribe can await completion before tearing down -- otherwise the unsub
-  // func is unavailable and the subscription would leak (via hass.connection's
-  // internal subscription map).
-  private _unsubscribers = new Map<string, Promise<SubscriptionUnsubscribe>>();
+  private _subscriptions = new KeyedSubscriptionManager<
+    string,
+    EventSubscriptionRequest
+  >((request) => request.event_type);
 
   public async subscribe(
     hass: HomeAssistant,
     request: EventSubscriptionRequest,
   ): Promise<void> {
-    const isFirst = !this._hasSubscribers(request.event_type);
-    this._requests.push(request);
-    if (isFirst) {
-      const pendingSubscription = hass.connection.subscribeEvents<HassEvent>(
+    await this._subscriptions.subscribe(request, () =>
+      hass.connection.subscribeEvents<HassEvent>(
         (event) => this._receiveEvent(event),
         request.event_type,
-      );
-      this._unsubscribers.set(request.event_type, pendingSubscription);
-      await pendingSubscription;
-    }
+      ),
+    );
   }
 
   public async unsubscribe(request: EventSubscriptionRequest): Promise<void> {
-    this._requests = this._requests.filter((r) => r !== request);
-    if (!this._hasSubscribers(request.event_type)) {
-      const pendingSubscription = this._unsubscribers.get(request.event_type);
-      this._unsubscribers.delete(request.event_type);
-      const unsubscribeCallback = await pendingSubscription;
-      await unsubscribeCallback?.();
-    }
-  }
-
-  private _hasSubscribers(eventType: string): boolean {
-    return this._requests.some((r) => r.event_type === eventType);
+    await this._subscriptions.unsubscribe(request);
   }
 
   private _receiveEvent(event: HassEvent): void {
-    for (const request of this._requests) {
-      if (
-        request.event_type === event.event_type &&
-        (!request.matcher || request.matcher(event.data))
-      ) {
+    for (const request of this._subscriptions.getRequestsForKey(event.event_type)) {
+      if (!request.matcher || request.matcher(event.data)) {
         request.callback(event.data);
       }
     }

src/ha/types.ts

@@ -243,8 +243,6 @@ export interface HassStateDifference {
   newState: HassEntity;
 }
 
-export type SubscriptionUnsubscribe = () => Promise<void>;
-
 // *************************************************************************
 //                     Home Assistant API types.
 // *************************************************************************

src/utils/keyed-subscription-manager.ts

@@ -0,0 +1,65 @@
+import PQueue from 'p-queue';
+
+type UnsubscribeFn = () => Promise<void>;
+type SubscribeFn = () => Promise<UnsubscribeFn>;
+
+/**
+ * Manages subscriptions keyed by `K`: the first subscriber for a key invokes
+ * `subscribeFn` to establish the underlying connection, subsequent subscribers
+ * for the same key piggyback on it, and the last to unsubscribe tears it down.
+ *
+ * Operations for a given key run through a single-concurrency queue, so
+ * subscribe and unsubscribe cannot interleave for the same key -- no race
+ * windows by construction. This aims to a leak-proof reusable subscription
+ * manager.
+ */
+export class KeyedSubscriptionManager<K, R> {
+  private _requests: R[] = [];
+  private _unsubscribers = new Map<K, UnsubscribeFn>();
+  private _queues = new Map<K, PQueue>();
+  private _getKeyFn: (request: R) => K;
+
+  constructor(getKeyFn: (request: R) => K) {
+    this._getKeyFn = getKeyFn;
+  }
+
+  public async subscribe(request: R, subscribeFn: SubscribeFn): Promise<void> {
+    const key = this._getKeyFn(request);
+    await this._queueFor(key).add(async () => {
+      this._requests.push(request);
+      if (!this._unsubscribers.has(key)) {
+        const unsubscribe = await subscribeFn();
+        this._unsubscribers.set(key, unsubscribe);
+      }
+    });
+  }
+
+  public async unsubscribe(request: R): Promise<void> {
+    const key = this._getKeyFn(request);
+    await this._queueFor(key).add(async () => {
+      this._requests = this._requests.filter((r) => r !== request);
+      if (!this._hasSubscribers(key)) {
+        const unsubscribe = this._unsubscribers.get(key);
+        this._unsubscribers.delete(key);
+        await unsubscribe?.();
+      }
+    });
+  }
+
+  public getRequestsForKey(key: K): readonly R[] {
+    return this._requests.filter((r) => this._getKeyFn(r) === key);
+  }
+
+  private _queueFor(key: K): PQueue {
+    let queue = this._queues.get(key);
+    if (!queue) {
+      queue = new PQueue({ concurrency: 1 });
+      this._queues.set(key, queue);
+    }
+    return queue;
+  }
+
+  private _hasSubscribers(key: K): boolean {
+    return this._requests.some((r) => this._getKeyFn(r) === key);
+  }
+}