Merge pull request #2053 from ably/PUB-1702/objects-server-gc

[PUB-1702] Use server-provided GC grace period for tombstone removal

Author: VeskeR

src/plugins/objects/defaults.ts

@@ -1,6 +1,9 @@
 export const DEFAULTS = {
   gcInterval: 1000 * 60 * 5, // 5 minutes
   /**
+   * The SDK will attempt to use the `objectsGCGracePeriod` value provided by the server in the `connectionDetails` object of the `CONNECTED` event.
+   * If the server does not provide this value, the SDK will fall back to this default value.
+   *
    * Must be > 2 minutes to ensure we keep tombstones long enough to avoid the possibility of receiving an operation
    * with an earlier serial that would not have been applied if the tombstone still existed.
    *

src/plugins/objects/livemap.ts

@@ -2,7 +2,6 @@ import { dequal } from 'dequal';
 
 import type { Bufferlike } from 'common/platform';
 import type * as API from '../../../ably';
-import { DEFAULTS } from './defaults';
 import { LiveObject, LiveObjectData, LiveObjectUpdate, LiveObjectUpdateNoop } from './liveobject';
 import { ObjectId } from './objectid';
 import {
@@ -556,7 +555,7 @@ export class LiveMap<T extends API.LiveMapType> extends LiveObject<LiveMapData,
 
     const keysToDelete: string[] = [];
     for (const [key, value] of this._dataRef.data.entries()) {
-      if (value.tombstone === true && Date.now() - value.tombstonedAt! >= DEFAULTS.gcGracePeriod) {
+      if (value.tombstone === true && Date.now() - value.tombstonedAt! >= this._objects.gcGracePeriod) {
         keysToDelete.push(key);
       }
     }

src/plugins/objects/objects.ts

@@ -37,6 +37,8 @@ export interface OnObjectsEventResponse {
 export type BatchCallback = (batchContext: BatchContext) => void;
 
 export class Objects {
+  gcGracePeriod: number;
+
   private _client: BaseClient;
   private _channel: RealtimeChannel;
   private _state: ObjectsState;
@@ -63,6 +65,12 @@ export class Objects {
     this._objectsPool = new ObjectsPool(this);
     this._syncObjectsDataPool = new SyncObjectsDataPool(this);
     this._bufferedObjectOperations = [];
+    // use server-provided objectsGCGracePeriod if available, and subscribe to new connectionDetails that can be emitted as part of the RTN24
+    this.gcGracePeriod =
+      this._channel.connectionManager.connectionDetails?.objectsGCGracePeriod ?? DEFAULTS.gcGracePeriod;
+    this._channel.connectionManager.on('connectiondetails', (details: Record<string, any>) => {
+      this.gcGracePeriod = details.objectsGCGracePeriod ?? DEFAULTS.gcGracePeriod;
+    });
   }
 
   /**

src/plugins/objects/objectspool.ts

@@ -109,7 +109,7 @@ export class ObjectsPool {
       // tombstoned objects should be removed from the pool if they have been tombstoned for longer than grace period.
       // by removing them from the local pool, Objects plugin no longer keeps a reference to those objects, allowing JS's
       // Garbage Collection to eventually free the memory for those objects, provided the user no longer references them either.
-      if (obj.isTombstoned() && Date.now() - obj.tombstonedAt()! >= DEFAULTS.gcGracePeriod) {
+      if (obj.isTombstoned() && Date.now() - obj.tombstonedAt()! >= this._objects.gcGracePeriod) {
         toDelete.push(objectId);
         continue;
       }

test/common/modules/private_api_recorder.js

@@ -80,6 +80,8 @@ define(['test/support/output_directory_paths'], function (outputDirectoryPaths)
     'read.Defaults.version',
     'read.LiveMap._dataRef.data',
     'read.EventEmitter.events',
+    'read.Objects._DEFAULTS.gcGracePeriod',
+    'read.Objects.gcGracePeriod',
     'read.Platform.Config.push',
     'read.ProtocolMessage.channelSerial',
     'read.Realtime._transports',
@@ -141,8 +143,8 @@ define(['test/support/output_directory_paths'], function (outputDirectoryPaths)
     'write.Defaults.ENDPOINT',
     'write.Defaults.ENVIRONMENT',
     'write.Defaults.wsConnectivityCheckUrl',
-    'write.Objects._DEFAULTS.gcGracePeriod',
     'write.Objects._DEFAULTS.gcInterval',
+    'write.Objects.gcGracePeriod',
     'write.Platform.Config.push', // This implies using a mock implementation of the internal IPlatformPushConfig interface. Our mock (in push_channel_transport.js) then interacts with internal objects and private APIs of public objects to implement this interface; I haven’t added annotations for that private API usage, since there wasn’t an easy way to pass test context information into the mock. I think that for now we can just say that if we wanted to get rid of this private API usage, then we’d need to remove this mock entirely.
     'write.auth.authOptions.requestHeaders',
     'write.auth.key',

test/realtime/objects.test.js

@@ -15,7 +15,6 @@ define(['ably', 'shared_helper', 'chai', 'objects', 'objects_helper'], function
   const objectsFixturesChannel = 'objects_fixtures';
   const nextTick = Ably.Realtime.Platform.Config.nextTick;
   const gcIntervalOriginal = ObjectsPlugin.Objects._DEFAULTS.gcInterval;
-  const gcGracePeriodOriginal = ObjectsPlugin.Objects._DEFAULTS.gcGracePeriod;
 
   function RealtimeWithObjects(helper, options) {
     return helper.AblyRealtime({ ...options, plugins: { Objects: ObjectsPlugin } });
@@ -4525,6 +4524,99 @@ define(['ably', 'shared_helper', 'chai', 'objects', 'objects_helper'], function
         }, client);
       });
 
+      it('gcGracePeriod is set from connectionDetails.objectsGCGracePeriod', async function () {
+        const helper = this.test.helper;
+        const client = RealtimeWithObjects(helper);
+
+        await helper.monitorConnectionThenCloseAndFinishAsync(async () => {
+          await client.connection.once('connected');
+
+          const channel = client.channels.get('channel', channelOptionsWithObjects());
+          const objects = channel.objects;
+          const connectionManager = client.connection.connectionManager;
+          const connectionDetails = connectionManager.connectionDetails;
+
+          // gcGracePeriod should be set after the initial connection
+          helper.recordPrivateApi('read.Objects.gcGracePeriod');
+          expect(
+            objects.gcGracePeriod,
+            'Check gcGracePeriod is set after initial connection from connectionDetails.objectsGCGracePeriod',
+          ).to.exist;
+          helper.recordPrivateApi('read.Objects.gcGracePeriod');
+          expect(objects.gcGracePeriod).to.equal(
+            connectionDetails.objectsGCGracePeriod,
+            'Check gcGracePeriod is set to equal connectionDetails.objectsGCGracePeriod',
+          );
+
+          const connectionDetailsPromise = connectionManager.once('connectiondetails');
+
+          helper.recordPrivateApi('call.connectionManager.activeProtocol.getTransport');
+          helper.recordPrivateApi('call.transport.onProtocolMessage');
+          helper.recordPrivateApi('call.makeProtocolMessageFromDeserialized');
+          connectionManager.activeProtocol.getTransport().onProtocolMessage(
+            createPM({
+              action: 4, // CONNECTED
+              connectionDetails: {
+                ...connectionDetails,
+                objectsGCGracePeriod: 999,
+              },
+            }),
+          );
+
+          helper.recordPrivateApi('listen.connectionManager.connectiondetails');
+          await connectionDetailsPromise;
+          // wait for next tick to ensure the connectionDetails event was processed by Objects plugin
+          await new Promise((res) => nextTick(res));
+
+          helper.recordPrivateApi('read.Objects.gcGracePeriod');
+          expect(objects.gcGracePeriod).to.equal(999, 'Check gcGracePeriod is updated on new CONNECTED event');
+        }, client);
+      });
+
+      it('gcGracePeriod has a default value if connectionDetails.objectsGCGracePeriod is missing', async function () {
+        const helper = this.test.helper;
+        const client = RealtimeWithObjects(helper);
+
+        await helper.monitorConnectionThenCloseAndFinishAsync(async () => {
+          await client.connection.once('connected');
+
+          const channel = client.channels.get('channel', channelOptionsWithObjects());
+          const objects = channel.objects;
+          const connectionManager = client.connection.connectionManager;
+          const connectionDetails = connectionManager.connectionDetails;
+
+          helper.recordPrivateApi('read.Objects._DEFAULTS.gcGracePeriod');
+          helper.recordPrivateApi('write.Objects.gcGracePeriod');
+          // set gcGracePeriod to a value different from the default
+          objects.gcGracePeriod = ObjectsPlugin.Objects._DEFAULTS.gcGracePeriod + 1;
+
+          const connectionDetailsPromise = connectionManager.once('connectiondetails');
+
+          // send a CONNECTED event without objectsGCGracePeriod, it should use the default value instead
+          helper.recordPrivateApi('call.connectionManager.activeProtocol.getTransport');
+          helper.recordPrivateApi('call.transport.onProtocolMessage');
+          helper.recordPrivateApi('call.makeProtocolMessageFromDeserialized');
+          connectionManager.activeProtocol.getTransport().onProtocolMessage(
+            createPM({
+              action: 4, // CONNECTED
+              connectionDetails,
+            }),
+          );
+
+          helper.recordPrivateApi('listen.connectionManager.connectiondetails');
+          await connectionDetailsPromise;
+          // wait for next tick to ensure the connectionDetails event was processed by Objects plugin
+          await new Promise((res) => nextTick(res));
+
+          helper.recordPrivateApi('read.Objects._DEFAULTS.gcGracePeriod');
+          helper.recordPrivateApi('read.Objects.gcGracePeriod');
+          expect(objects.gcGracePeriod).to.equal(
+            ObjectsPlugin.Objects._DEFAULTS.gcGracePeriod,
+            'Check gcGracePeriod is set to a default value if connectionDetails.objectsGCGracePeriod is missing',
+          );
+        }, client);
+      });
+
       const tombstonesGCScenarios = [
         // for the next tests we need to access the private API of Objects plugin in order to verify that tombstoned entities were indeed deleted after the GC grace period.
         // public API hides that kind of information from the user and returns undefined for tombstoned entities even if realtime client still keeps a reference to them.
@@ -4631,8 +4723,6 @@ define(['ably', 'shared_helper', 'chai', 'objects', 'objects_helper'], function
         try {
           helper.recordPrivateApi('write.Objects._DEFAULTS.gcInterval');
           ObjectsPlugin.Objects._DEFAULTS.gcInterval = 500;
-          helper.recordPrivateApi('write.Objects._DEFAULTS.gcGracePeriod');
-          ObjectsPlugin.Objects._DEFAULTS.gcGracePeriod = 250;
 
           const objectsHelper = new ObjectsHelper(helper);
           const client = RealtimeWithObjects(helper, clientOptions);
@@ -4644,6 +4734,11 @@ define(['ably', 'shared_helper', 'chai', 'objects', 'objects_helper'], function
             await channel.attach();
             const root = await objects.getRoot();
 
+            helper.recordPrivateApi('read.Objects.gcGracePeriod');
+            const gcGracePeriodOriginal = objects.gcGracePeriod;
+            helper.recordPrivateApi('write.Objects.gcGracePeriod');
+            objects.gcGracePeriod = 250;
+
             // helper function to spy on the GC interval callback and wait for a specific number of GC cycles.
             // returns a promise which will resolve when required number of cycles have happened.
             const waitForGCCycles = (cycles) => {
@@ -4674,12 +4769,13 @@ define(['ably', 'shared_helper', 'chai', 'objects', 'objects_helper'], function
               helper,
               waitForGCCycles,
             });
+
+            helper.recordPrivateApi('write.Objects.gcGracePeriod');
+            objects.gcGracePeriod = gcGracePeriodOriginal;
           }, client);
         } finally {
           helper.recordPrivateApi('write.Objects._DEFAULTS.gcInterval');
           ObjectsPlugin.Objects._DEFAULTS.gcInterval = gcIntervalOriginal;
-          helper.recordPrivateApi('write.Objects._DEFAULTS.gcGracePeriod');
-          ObjectsPlugin.Objects._DEFAULTS.gcGracePeriod = gcGracePeriodOriginal;
         }
       });
 
@@ -4920,7 +5016,7 @@ define(['ably', 'shared_helper', 'chai', 'objects', 'objects_helper'], function
        */
       it('object message publish respects connectionDetails.maxMessageSize', async function () {
         const helper = this.test.helper;
-        const client = RealtimeWithObjects(helper, { clientId: 'test' });
+        const client = RealtimeWithObjects(helper);
 
         await helper.monitorConnectionThenCloseAndFinishAsync(async () => {
           await client.connection.once('connected');