Merge pull request #2223 from ably/path-subscribe-dedup

Propose an alternative to `PathEvent.bubbles` to make intent clearer

Author: lawrence-forooghian

src/plugins/liveobjects/liveobject.ts

@@ -299,40 +299,44 @@ export abstract class LiveObject<
 
   /**
    * Notifies path-based subscriptions about changes to this object.
-   * For LiveMapUpdate events, also creates non-bubbling events for each updated key.
+   * For LiveMapUpdate events, each updated key also contributes a candidate
+   * path one segment deeper than this object's own path.
    */
   private _notifyPathSubscriptions(update: TUpdate): void {
-    const paths = this.getFullPaths();
+    const pathsToThis = this.getFullPaths();
 
-    if (paths.length === 0) {
+    if (pathsToThis.length === 0) {
       // No paths to this object, skip notification
       return;
     }
 
     // Do not expose object sync messages as they do not represent a single operation on an object
     const operationObjectMessage = update.objectMessage?.isOperationMessage() ? update.objectMessage : undefined;
-    const pathEvents: PathEvent[] = paths.map((path) => ({
-      path,
-      message: operationObjectMessage,
-      bubbles: true,
-    }));
-
-    // For LiveMapUpdate, also create non-bubbling events for each updated key
-    if (update._type === 'LiveMapUpdate') {
-      const updatedKeys = Object.keys(update.update);
-
-      for (const key of updatedKeys) {
-        for (const basePath of paths) {
-          pathEvents.push({
-            path: [...basePath, key],
-            message: operationObjectMessage,
-            bubbles: false,
-          });
+
+    // Call notifyPathEvent() once for each path-to-this. Since
+    // notifyPathEvent() emits at most one event on each subscription, this
+    // means that we emit at most one event per path-to-this.
+    for (const pathToThis of pathsToThis) {
+      const priorityOrderedCandidatePaths: Path[] = [pathToThis];
+
+      // For LiveMapUpdate, also add a candidate path per updated key. We insert these after
+      // pathToThis so that notifyPathEvent() picks pathToThis in the case where a given subscription
+      // covers multiple candidate paths (that is, we favour the shorter path).
+      if (update._type === 'LiveMapUpdate') {
+        const updatedKeys = Object.keys(update.update);
+
+        for (const key of updatedKeys) {
+          priorityOrderedCandidatePaths.push([...pathToThis, key]);
         }
       }
-    }
 
-    this._realtimeObject.getPathObjectSubscriptionRegister().notifyPathEvents(pathEvents);
+      const pathEvent: PathEvent = {
+        priorityOrderedCandidatePaths,
+        message: operationObjectMessage,
+      };
+
+      this._realtimeObject.getPathObjectSubscriptionRegister().notifyPathEvent(pathEvent);
+    }
   }
 
   private _isNoopUpdate(update: TUpdate | LiveObjectUpdateNoop): update is LiveObjectUpdateNoop {

src/plugins/liveobjects/pathobjectsubscriptionregister.ts

@@ -22,12 +22,14 @@ export interface SubscriptionEntry {
  * Event data that LiveObjects provide when notifying of changes
  */
 export interface PathEvent {
-  /** The path where the event occurred */
-  path: Path;
+  /**
+   * Candidate paths for surfacing this event to subscriptions, in order of
+   * decreasing priority. For a given subscription, the first candidate path
+   * it covers is used as the path of `event.object` passed to its listener.
+   */
+  priorityOrderedCandidatePaths: Path[];
   /** Object message that caused this event */
   message?: ObjectMessage;
-  /** Whether this event should bubble up to parent paths. Defaults to true if not specified. */
-  bubbles?: boolean;
 }
 
 /**
@@ -87,28 +89,26 @@ export class PathObjectSubscriptionRegister {
   }
 
   /**
-   * Notifies all matching subscriptions about an event that occurred at the specified path(s).
-   *
-   * @param events - Array of path events to process
-   */
-  notifyPathEvents(events: PathEvent[]): void {
-    for (const event of events) {
-      this._processEvent(event);
-    }
-  }
-
-  /**
-   * Processes a single path event and calls all matching subscription listeners.
+   * Dispatches a {@link PathEvent} to subscriptions. Each subscription that
+   * covers any of the event's {@link PathEvent.priorityOrderedCandidatePaths}
+   * receives at most one notification, at the first covered path.
    */
-  private _processEvent(event: PathEvent): void {
+  notifyPathEvent(event: PathEvent): void {
     for (const subscription of this._subscriptions.values()) {
-      if (!this._shouldNotifySubscription(subscription, event)) {
+      const chosenCoveredPath = event.priorityOrderedCandidatePaths.find((path) =>
+        this._subscriptionCoversPath(subscription, path),
+      );
+      if (chosenCoveredPath === undefined) {
         continue;
       }
 
       try {
         const subscriptionEvent: PathObjectSubscriptionEvent = {
-          object: new DefaultPathObject(this._realtimeObject, this._realtimeObject.getPool().getRoot(), event.path),
+          object: new DefaultPathObject(
+            this._realtimeObject,
+            this._realtimeObject.getPool().getRoot(),
+            chosenCoveredPath,
+          ),
           message: event.message?.toUserFacingMessage(this._realtimeObject.getChannel()),
         };
 
@@ -118,46 +118,35 @@ export class PathObjectSubscriptionRegister {
         this._client.Logger.logAction(
           this._client.logger,
           this._client.Logger.LOG_MINOR,
-          'PathObjectSubscriptionRegister._processEvent()',
-          `Error in PathObject subscription listener; path=${JSON.stringify(event.path)}, error=${error}`,
+          'PathObjectSubscriptionRegister.notifyPathEvent()',
+          `Error in PathObject subscription listener; path=${JSON.stringify(chosenCoveredPath)}, error=${error}`,
         );
       }
     }
   }
 
   /**
-   * Determines if a subscription should be notified about an event at the given path.
-   * Implements depth-based filtering logic and bubbling control.
-   *
-   * Depth examples (when event.bubbles is true):
-   * - subscription at ["users"] with depth=undefined: matches ["users"], ["users", "emma"], ["users", "emma", "visits"], etc.
-   * - subscription at ["users"] with depth=1: matches ["users"] only
-   * - subscription at ["users"] with depth=2: matches ["users"], ["users", "emma"] only
-   * - subscription at ["users"] with depth=3: matches ["users"], ["users", "emma"], ["users", "emma", "visits"] only
+   * Returns true if the given path falls within the area covered by the
+   * subscription — that is, it starts with the subscription's path, and
+   * extends it by at most `depth − 1` further segments.
    *
-   * Non-bubbling examples (when event.bubbles is false):
-   * - Event at ["users", "emma"] with bubbles=false:
-   *   - subscription at ["users"]: NOT triggered (no bubbling to parent)
-   *   - subscription at ["users", "emma"]: triggered (exact path match)
+   * Coverage examples:
+   * - subscription at ["users"] with depth=undefined: covers ["users"], ["users", "emma"], ["users", "emma", "visits"], etc.
+   * - subscription at ["users"] with depth=1: covers ["users"] only
+   * - subscription at ["users"] with depth=2: covers ["users"], ["users", "emma"] only
+   * - subscription at ["users"] with depth=3: covers ["users"], ["users", "emma"], ["users", "emma", "visits"] only
    *
    * The depth calculation is: eventPath.length - subscriptionPath.length + 1
    * This means:
    * - Same level (["users"] -> ["users"]): 1 - 1 + 1 = 1 (depth=1)
    * - One level deeper (["users"] -> ["users", "emma"]): 2 - 1 + 1 = 2 (depth=2)
    * - Two levels deeper (["users"] -> ["users", "emma", "visits"]): 3 - 1 + 1 = 3 (depth=3)
    */
-  private _shouldNotifySubscription(subscription: SubscriptionEntry, event: PathEvent): boolean {
+  private _subscriptionCoversPath(subscription: SubscriptionEntry, eventPath: Path): boolean {
     const subPath = subscription.path;
-    const eventPath = event.path;
     const depth = subscription.options.depth;
-    const bubbles = event.bubbles !== false; // Default to true if not specified
 
-    // If event doesn't bubble, only match exact paths
-    if (!bubbles) {
-      return this._pathsAreEqual(eventPath, subPath);
-    }
-
-    // Otherwise check if the event path starts with the subscription path
+    // Check if the event path starts with the subscription path
     if (!this._pathStartsWith(eventPath, subPath)) {
       return false;
     }
@@ -194,15 +183,4 @@ export class PathObjectSubscriptionRegister {
 
     return true;
   }
-
-  /**
-   * Checks if two paths are exactly equal.
-   *
-   * @param path1 - First path to compare
-   * @param path2 - Second path to compare
-   * @returns true if paths are exactly equal
-   */
-  private _pathsAreEqual(path1: Path, path2: Path): boolean {
-    return this._client.Utils.arrEquals(path1, path2);
-  }
 }

test/realtime/liveobjects.test.js

@@ -5590,6 +5590,64 @@ define(['ably', 'shared_helper', 'chai', 'liveobjects', 'liveobjects_helper'], f
           },
         },
 
+        {
+          description:
+            'PathObject.subscribe() fires once per path when the same object is referenced from multiple paths',
+          action: async (ctx) => {
+            const { entryPathObject, entryInstance, objectsHelper, channel } = ctx;
+
+            // Inject ObjectMessages directly because the Realtime public API doesn't expose a way to
+            // alias the same underlying object at multiple paths.
+            // Create a single counter and reference it from two keys on root, so it has two full paths.
+            const counterId = objectsHelper.fakeCounterObjectId();
+            await objectsHelper.processObjectOperationMessageOnChannel({
+              channel,
+              serial: lexicoTimeserial('aaa', 0, 0),
+              siteCode: 'aaa',
+              state: [objectsHelper.counterCreateOp({ objectId: counterId })],
+            });
+            const keyAUpdatedPromise = waitForMapKeyUpdate(entryInstance, 'counterA');
+            await objectsHelper.processObjectOperationMessageOnChannel({
+              channel,
+              serial: lexicoTimeserial('aaa', 1, 0),
+              siteCode: 'aaa',
+              state: [objectsHelper.mapSetOp({ objectId: 'root', key: 'counterA', data: { objectId: counterId } })],
+            });
+            await keyAUpdatedPromise;
+            const keyBUpdatedPromise = waitForMapKeyUpdate(entryInstance, 'counterB');
+            await objectsHelper.processObjectOperationMessageOnChannel({
+              channel,
+              serial: lexicoTimeserial('aaa', 2, 0),
+              siteCode: 'aaa',
+              state: [objectsHelper.mapSetOp({ objectId: 'root', key: 'counterB', data: { objectId: counterId } })],
+            });
+            await keyBUpdatedPromise;
+
+            // subscribe at root with unlimited depth — the increment below should be reported once per path
+            const receivedPaths = [];
+            entryPathObject.subscribe((event) => {
+              receivedPaths.push(event.object.path());
+            });
+
+            // a single COUNTER_INC op produces a single LiveObjectUpdate on a single counter object,
+            // but that counter is reachable from two paths, so the subscription should fire twice
+            const counterUpdatedPromise = waitForCounterUpdate(entryInstance.get('counterA'));
+            await objectsHelper.processObjectOperationMessageOnChannel({
+              channel,
+              serial: lexicoTimeserial('aaa', 3, 0),
+              siteCode: 'aaa',
+              state: [objectsHelper.counterIncOp({ objectId: counterId, amount: 1 })],
+            });
+            await counterUpdatedPromise;
+
+            expect(receivedPaths).to.have.lengthOf(2, 'Check one event was received per path');
+            expect(receivedPaths.slice().sort()).to.deep.equal(
+              ['counterA', 'counterB'],
+              'Check events were received at both paths',
+            );
+          },
+        },
+
         {
           description: 'PathObject.subscribe() on LiveCounter path receives increment/decrement events',
           action: async (ctx) => {