Fix event subscriptions stopping after a reconnect

ui/component/core/build.gradle

@@ -9,3 +9,8 @@ tasks.register('prepareUi') {
 tasks.register('publishUi') {
   dependsOn clean, npmPublish
 }
+
+tasks.register('npmTest', Exec) {
+  dependsOn getYarnInstallTask()
+  commandLine npmCommand("yarn"), "run", "test"
+}

ui/component/core/package.json

@@ -16,7 +16,7 @@
   ],
   "scripts": {
     "build": "npx tsc -b",
-    "test": "echo \"No tests\" && exit 0",
+    "test": "npx tsc -b && npx playwright test",
     "prepack": "npx rspack"
   },
   "author": "OpenRemote",

ui/component/core/playwright.config.ts

@@ -0,0 +1,21 @@
+/*
+ * Copyright 2026, OpenRemote Inc.
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Affero General Public License as
+ * published by the Free Software Foundation, either version 3 of the
+ * License, or (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+ * GNU Affero General Public License for more details.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program. If not, see <https://www.gnu.org/licenses/>.
+ *
+ * SPDX-License-Identifier: AGPL-3.0-or-later
+ */
+import { defineCtConfig } from "@openremote/test/component.config";
+
+export default defineCtConfig(__dirname);

ui/component/core/src/event.ts

@@ -90,6 +90,8 @@ interface EventSubscriptionInfo<T extends SharedEvent> {
   eventSubscription: EventSubscription<T>;
   callback: (event: T) => void;
   deferred: Deferred<string> | null;
+  // Unsubscribed while the server has yet to confirm it
+  cancelled?: boolean;
 }
 
 interface AssetSubscriptionInfo {
@@ -229,8 +231,12 @@ abstract class EventProviderImpl implements EventProvider {
           const subscription = this._pendingSubscription;
           this._pendingSubscription = null;
 
-          // Store subscriptionId and callback
-          this._subscriptionMap.set(subscriptionId, subscription);
+          if (subscription.cancelled) {
+            this._doUnsubscribe(subscriptionId);
+          } else {
+            // Store subscriptionId and callback
+            this._subscriptionMap.set(subscriptionId, subscription);
+          }
 
           this._processNextSubscription();
           const deferred = subscription.deferred;
@@ -266,6 +272,21 @@ abstract class EventProviderImpl implements EventProvider {
       return;
     }
 
+    // Restored subscriptions are not in the map until the server has confirmed them again
+    const queuedIndex = this._queuedSubscriptions.findIndex(
+      (info) => info.eventSubscription.subscriptionId === subscriptionId
+    );
+    if (queuedIndex !== -1) {
+      const [queued] = this._queuedSubscriptions.splice(queuedIndex, 1);
+      this._resolveUnsubscribed(queued);
+      return;
+    }
+
+    if (this._pendingSubscription?.eventSubscription.subscriptionId === subscriptionId) {
+      this._pendingSubscription.cancelled = true;
+      return;
+    }
+
     if (this._assetEventCallbackMap.delete(subscriptionId)) {
       if (this._assetEventPromise && this._assetEventCallbackMap.size === 0) {
         // Remove the asset subscription
@@ -330,8 +351,12 @@ abstract class EventProviderImpl implements EventProvider {
       //         assetIds: assetIds
       //     };
       // }
+      // A failed subscription must not be reused
       this._assetEventPromise = this.subscribe(subscription, (event) => {
         this._assetEventCallbackMap.forEach((callback) => callback(event));
+      }).catch((reason) => {
+        this._assetEventPromise = undefined;
+        throw reason;
       });
     }
 
@@ -352,6 +377,9 @@ abstract class EventProviderImpl implements EventProvider {
 
     this._assetEventCallbackMap.set(subscriptionId, eventFilter);
 
+    // The caller doesn't get an ID to unsubscribe with when this fails
+    this._assetEventPromise.catch(() => this._assetEventCallbackMap.delete(subscriptionId));
+
     return this._assetEventPromise.then(() => {
       try {
         // Get the current state of the assets if requested
@@ -418,8 +446,12 @@ abstract class EventProviderImpl implements EventProvider {
         eventType: "attribute",
         filter: assetFilter,
       };
+      // A failed subscription must not be reused
       this._attributeEventPromise = this.subscribe(subscription, (event) => {
         this._attributeEventCallbackMap.forEach((callback) => callback(event));
+      }).catch((reason) => {
+        this._attributeEventPromise = undefined;
+        throw reason;
       });
     }
 
@@ -447,6 +479,9 @@ abstract class EventProviderImpl implements EventProvider {
 
     this._attributeEventCallbackMap.set(subscriptionId, eventFilter);
 
+    // The caller doesn't get an ID to unsubscribe with when this fails
+    this._attributeEventPromise.catch(() => this._attributeEventCallbackMap.delete(subscriptionId));
+
     return this._attributeEventPromise.then(() => {
       try {
         // Get the current state of the attributes if requested
@@ -551,14 +586,8 @@ abstract class EventProviderImpl implements EventProvider {
   }
 
   protected _onConnect() {
-    if (Object.keys(this._subscriptionMap).length > 0) {
-      for (const subscriptionId in this._subscriptionMap) {
-        if (this._subscriptionMap.has(subscriptionId)) {
-          this._queuedSubscriptions.unshift(this._subscriptionMap.get(subscriptionId)!);
-        }
-      }
-      this._subscriptionMap.clear();
-    }
+    this._subscriptionMap.forEach((subscription) => this._queuedSubscriptions.unshift(subscription));
+    this._subscriptionMap.clear();
 
     this._processNextSubscription();
   }
@@ -568,14 +597,26 @@ abstract class EventProviderImpl implements EventProvider {
       this._onStatusChanged(EventProviderStatus.DISCONNECTED);
     }
     if (this._pendingSubscription) {
-      this._queuedSubscriptions.unshift(this._pendingSubscription);
+      const pending = this._pendingSubscription;
       this._pendingSubscription = null;
+
+      if (pending.cancelled) {
+        this._resolveUnsubscribed(pending);
+      } else {
+        this._queuedSubscriptions.unshift(pending);
+      }
     }
 
     this._onStatusChanged(EventProviderStatus.CONNECTING);
     this._scheduleReconnect();
   }
 
+  // Resolves rather than rejects, since the subscribe didn't fail but was unsubscribed
+  protected _resolveUnsubscribed(subscription: EventSubscriptionInfo<SharedEvent>) {
+    subscription.deferred?.resolve(subscription.eventSubscription.subscriptionId!);
+    subscription.deferred = null;
+  }
+
   protected _scheduleReconnect() {
     if (this._reconnectTimer) {
       return;
@@ -646,6 +687,8 @@ export class WebSocketEventProvider extends EventProviderImpl {
   }
 
   protected _doConnect(): Promise<boolean> {
+    this._discardWebSocket();
+
     let authorisedUrl = this._endpointUrl + "?Realm=" + manager.config.realm;
     this._connectDeferred = new Deferred();
 
@@ -741,7 +784,23 @@ export class WebSocketEventProvider extends EventProviderImpl {
     return this._connectDeferred.promise;
   }
 
+  // Only the current socket may change the state of the provider
+  protected _discardWebSocket(): void {
+    if (!this._webSocket) {
+      return;
+    }
+
+    this._webSocket.onopen = null;
+    this._webSocket.onerror = null;
+    this._webSocket.onclose = null;
+    this._webSocket.onmessage = null;
+    this._webSocket.close();
+    this._webSocket = undefined;
+  }
+
   protected _beforeDisconnect(): void {
+    // No reply can arrive anymore and the pending subscription is sent again after reconnecting
+    this._subscribeDeferred = null;
     this._onDisconnect();
   }