fix: LiveQuery subscriptions leak when a client reuses a subscribe requestId (#10499)

Author: mtrezza

spec/ParseLiveQuery.spec.js

@@ -1485,3 +1485,212 @@ describe('ParseLiveQuery', function () {
     });
   });
 });
+
+describe('ParseLiveQuery duplicate requestId handling', function () {
+  const WebSocket = require('ws');
+
+  const waitFor = async predicate => {
+    const deadline = Date.now() + 4000;
+    while (Date.now() < deadline) {
+      if (predicate()) {
+        return;
+      }
+      await sleep(20);
+    }
+    throw new Error('timed out waiting for condition');
+  };
+
+  let sockets;
+
+  beforeEach(() => {
+    Parse.CoreManager.getLiveQueryController().setDefaultLiveQueryClient(null);
+    sockets = [];
+  });
+
+  afterEach(() => {
+    for (const socket of sockets) {
+      if (socket.readyState === WebSocket.OPEN) {
+        socket.close();
+      }
+    }
+    sockets = [];
+  });
+
+  const configureServer = async () => {
+    const parseServer = await reconfigureServer({
+      liveQuery: { classNames: ['LQDupA', 'LQDupB'] },
+      startLiveQueryServer: true,
+      verbose: false,
+      silent: true,
+    });
+    return parseServer.liveQueryServer;
+  };
+
+  // Opens a raw LiveQuery WebSocket client and returns a small protocol helper.
+  const openClient = async () => {
+    const socket = new WebSocket('ws://localhost:8378/1');
+    sockets.push(socket);
+    const messages = [];
+    socket.on('message', data => messages.push(JSON.parse(data.toString())));
+    await new Promise((resolve, reject) => {
+      socket.on('open', resolve);
+      socket.on('error', reject);
+    });
+    socket.send(JSON.stringify({ op: 'connect', applicationId: Parse.applicationId }));
+    const client = {
+      socket,
+      messages,
+      subscribe(requestId, className, where) {
+        socket.send(JSON.stringify({ op: 'subscribe', requestId, query: { className, where } }));
+      },
+      update(requestId, className, where) {
+        socket.send(JSON.stringify({ op: 'update', requestId, query: { className, where } }));
+      },
+      countOp(op) {
+        return messages.filter(message => message.op === op).length;
+      },
+      waitForOpCount(op, count) {
+        return waitFor(() => this.countOp(op) === count);
+      },
+    };
+    await waitFor(() => messages.some(message => message.op === 'connected'));
+    return client;
+  };
+
+  it('replaces rather than leaks subscriptions when a client reuses a requestId with different queries', async () => {
+    const lqServer = await configureServer();
+    const client = await openClient();
+
+    for (let i = 0; i < 5; i++) {
+      client.subscribe(7, 'LQDupA', { marker: `ws-${i}` });
+    }
+    await client.waitForOpCount('subscribed', 5);
+
+    // Reusing one requestId must keep a single active subscription, not one per frame.
+    expect(lqServer.subscriptions.get('LQDupA').size).toBe(1);
+
+    client.socket.close();
+    await waitFor(() => lqServer.clients.size === 0);
+
+    // No stale subscriptions may survive the disconnect.
+    expect(lqServer.subscriptions.get('LQDupA')?.size ?? 0).toBe(0);
+  });
+
+  it('does not leak subscriptions when a client reuses a requestId with the same query', async () => {
+    const lqServer = await configureServer();
+    const client = await openClient();
+
+    for (let i = 0; i < 5; i++) {
+      client.subscribe(7, 'LQDupA', { marker: 'same' });
+    }
+    await client.waitForOpCount('subscribed', 5);
+
+    expect(lqServer.subscriptions.get('LQDupA').size).toBe(1);
+
+    client.socket.close();
+    await waitFor(() => lqServer.clients.size === 0);
+
+    expect(lqServer.subscriptions.get('LQDupA')?.size ?? 0).toBe(0);
+  });
+
+  it('cleans up the prior subscription when a client reuses a requestId on a different class', async () => {
+    const lqServer = await configureServer();
+    const client = await openClient();
+
+    client.subscribe(7, 'LQDupA', { marker: 'a' });
+    await client.waitForOpCount('subscribed', 1);
+    client.subscribe(7, 'LQDupB', { marker: 'b' });
+    await client.waitForOpCount('subscribed', 2);
+    client.subscribe(7, 'LQDupA', { marker: 'a2' });
+    await client.waitForOpCount('subscribed', 3);
+
+    // Only the most recent subscription survives; the prior class entry is pruned.
+    expect(lqServer.subscriptions.get('LQDupA').size).toBe(1);
+    expect(lqServer.subscriptions.has('LQDupB')).toBe(false);
+
+    client.socket.close();
+    await waitFor(() => lqServer.clients.size === 0);
+
+    expect(lqServer.subscriptions.get('LQDupA')?.size ?? 0).toBe(0);
+    expect(lqServer.subscriptions.has('LQDupB')).toBe(false);
+  });
+
+  it('does not tear down a subscription still held by another client when a client reuses a requestId', async () => {
+    const lqServer = await configureServer();
+    const clientA = await openClient();
+    const clientB = await openClient();
+
+    // Both clients share the same query, so they share one Subscription.
+    clientA.subscribe(7, 'LQDupA', { marker: 'shared' });
+    await clientA.waitForOpCount('subscribed', 1);
+    clientB.subscribe(9, 'LQDupA', { marker: 'shared' });
+    await clientB.waitForOpCount('subscribed', 1);
+    expect(lqServer.subscriptions.get('LQDupA').size).toBe(1);
+
+    // Client A reuses its requestId with a different query.
+    clientA.subscribe(7, 'LQDupA', { marker: 'other' });
+    await clientA.waitForOpCount('subscribed', 2);
+
+    // The shared subscription must survive (B still holds it), alongside A's new one.
+    expect(lqServer.subscriptions.get('LQDupA').size).toBe(2);
+
+    // The shared subscription still delivers events to B, but not to A anymore.
+    const shared = new Parse.Object('LQDupA');
+    shared.set('marker', 'shared');
+    await shared.save(null, { useMasterKey: true });
+    await clientB.waitForOpCount('create', 1);
+    expect(clientB.countOp('create')).toBe(1);
+    expect(clientA.countOp('create')).toBe(0);
+
+    clientA.socket.close();
+    clientB.socket.close();
+    await waitFor(() => lqServer.clients.size === 0);
+    expect(lqServer.subscriptions.get('LQDupA')?.size ?? 0).toBe(0);
+  });
+
+  it('delivers events only for the replacement query after a client reuses a requestId', async () => {
+    const lqServer = await configureServer();
+    const client = await openClient();
+
+    client.subscribe(7, 'LQDupA', { marker: 'old' });
+    await client.waitForOpCount('subscribed', 1);
+    client.subscribe(7, 'LQDupA', { marker: 'new' });
+    await client.waitForOpCount('subscribed', 2);
+    expect(lqServer.subscriptions.get('LQDupA').size).toBe(1);
+
+    const oldObject = new Parse.Object('LQDupA');
+    oldObject.set('marker', 'old');
+    await oldObject.save(null, { useMasterKey: true });
+
+    const newObject = new Parse.Object('LQDupA');
+    newObject.set('marker', 'new');
+    await newObject.save(null, { useMasterKey: true });
+
+    await client.waitForOpCount('create', 1);
+    // Only the replacement query (marker 'new') may produce an event.
+    expect(client.countOp('create')).toBe(1);
+    expect(client.messages.find(message => message.op === 'create').object.marker).toBe('new');
+  });
+
+  it('keeps the update op working after the duplicate-subscribe cleanup', async () => {
+    const lqServer = await configureServer();
+    const client = await openClient();
+
+    client.subscribe(7, 'LQDupA', { marker: 'old' });
+    await client.waitForOpCount('subscribed', 1);
+    client.update(7, 'LQDupA', { marker: 'new' });
+    await client.waitForOpCount('subscribed', 2);
+
+    expect(lqServer.subscriptions.get('LQDupA').size).toBe(1);
+
+    const updated = new Parse.Object('LQDupA');
+    updated.set('marker', 'new');
+    await updated.save(null, { useMasterKey: true });
+    await client.waitForOpCount('create', 1);
+    expect(client.countOp('create')).toBe(1);
+
+    client.socket.close();
+    await waitFor(() => lqServer.clients.size === 0);
+    expect(lqServer.subscriptions.get('LQDupA')?.size ?? 0).toBe(0);
+  });
+});

spec/helper.js

@@ -657,12 +657,18 @@ global.fdescribe_only = validator => {
 
 const libraryCache = {};
 jasmine.mockLibrary = function (library, name, mock) {
-  const original = require(library)[name];
   if (!libraryCache[library]) {
     libraryCache[library] = {};
   }
+  // Cache the original implementation only the first time an export is mocked.
+  // Re-mocking the same export (e.g. swapping the mock mid-test) must not
+  // overwrite the cached original with another mock, otherwise restoreLibrary
+  // would restore a mock instead of the real implementation and leak it into
+  // later specs.
+  if (!(name in libraryCache[library])) {
+    libraryCache[library][name] = require(library)[name];
+  }
   require(library)[name] = mock;
-  libraryCache[library][name] = original;
 };
 
 jasmine.restoreLibrary = function (library, name) {

src/LiveQuery/ParseLiveQueryServer.ts

@@ -523,12 +523,14 @@ class ParseLiveQueryServer {
 
         // If there is no client which is subscribing this subscription, remove it from subscriptions
         const classSubscriptions = this.subscriptions.get(subscription.className);
-        if (!subscription.hasSubscribingClient()) {
-          classSubscriptions.delete(subscription.hash);
-        }
-        // If there is no subscriptions under this class, remove it from subscriptions
-        if (classSubscriptions.size === 0) {
-          this.subscriptions.delete(subscription.className);
+        if (classSubscriptions) {
+          if (!subscription.hasSubscribingClient()) {
+            classSubscriptions.delete(subscription.hash);
+          }
+          // If there is no subscriptions under this class, remove it from subscriptions
+          if (classSubscriptions.size === 0) {
+            this.subscriptions.delete(subscription.className);
+          }
         }
       }
 
@@ -1164,6 +1166,28 @@ class ParseLiveQueryServer {
       // Validate regex patterns in the subscription query
       this._validateQueryConstraints(request.query.where);
 
+      // If this client already has a subscription registered under this
+      // requestId, replace it by tearing down the previous subscription before
+      // creating the new one. The client-side metadata map is keyed only by
+      // requestId, so a duplicate `subscribe` frame would otherwise overwrite it
+      // while the previous Subscription stays in the server-wide map, leaking it
+      // for the lifetime of the process (disconnect cleanup only walks the
+      // surviving client metadata and never reaches the orphaned subscription).
+      const previousSubscriptionInfo = client.getSubscriptionInfo(request.requestId);
+      if (previousSubscriptionInfo) {
+        const previousSubscription = previousSubscriptionInfo.subscription;
+        previousSubscription.deleteClientSubscription(parseWebsocket.clientId, request.requestId);
+        const previousClassSubscriptions = this.subscriptions.get(previousSubscription.className);
+        if (previousClassSubscriptions) {
+          if (!previousSubscription.hasSubscribingClient()) {
+            previousClassSubscriptions.delete(previousSubscription.hash);
+          }
+          if (previousClassSubscriptions.size === 0) {
+            this.subscriptions.delete(previousSubscription.className);
+          }
+        }
+      }
+
       // Get subscription from subscriptions, create one if necessary
       const subscriptionHash = queryHash(request.query);
       // Add className to subscriptions if necessary
@@ -1286,12 +1310,14 @@ class ParseLiveQueryServer {
     subscription.deleteClientSubscription(parseWebsocket.clientId, requestId);
     // If there is no client which is subscribing this subscription, remove it from subscriptions
     const classSubscriptions = this.subscriptions.get(className);
-    if (!subscription.hasSubscribingClient()) {
-      classSubscriptions.delete(subscription.hash);
-    }
-    // If there is no subscriptions under this class, remove it from subscriptions
-    if (classSubscriptions.size === 0) {
-      this.subscriptions.delete(className);
+    if (classSubscriptions) {
+      if (!subscription.hasSubscribingClient()) {
+        classSubscriptions.delete(subscription.hash);
+      }
+      // If there is no subscriptions under this class, remove it from subscriptions
+      if (classSubscriptions.size === 0) {
+        this.subscriptions.delete(className);
+      }
     }
     runLiveQueryEventHandlers({
       client,

src/LiveQuery/Subscription.js

@@ -22,7 +22,11 @@ class Subscription {
       this.clientRequestIds.set(clientId, []);
     }
     const requestIds = this.clientRequestIds.get(clientId);
-    requestIds.push(requestId);
+    // Keep (clientId, requestId) pairs unique so a duplicate registration cannot
+    // leave a residual entry that survives cleanup.
+    if (!requestIds.includes(requestId)) {
+      requestIds.push(requestId);
+    }
   }
 
   deleteClientSubscription(clientId: number, requestId: number): void {