Protocol v6 and support for partial sync of objects

See spec ably/specification#413

Resolves AIT-38

Author: VeskeR

src/common/lib/util/defaults.ts

@@ -91,7 +91,7 @@ const Defaults = {
   maxMessageSize: 65536,
 
   version,
-  protocolVersion: 5,
+  protocolVersion: 6,
   agent,
   getPort,
   getHttpScheme,

src/plugins/liveobjects/realtimeobject.ts

@@ -43,8 +43,8 @@ export class RealtimeObject {
   // related to RTC10, should have a separate EventEmitter for users of the library
   private _eventEmitterPublic: EventEmitter;
   private _objectsPool: ObjectsPool; // RTO3
-  /** An array of ObjectMessages received during the sync sequence */
-  private _syncObjectsPool: ObjectMessage[];
+  /** Used to accumulate object state during a sync sequence, keyed by object ID */
+  private _syncObjectsPool: Map<string, ObjectMessage>;
   private _currentSyncId: string | undefined;
   private _currentSyncCursor: string | undefined;
   private _bufferedObjectOperations: ObjectMessage[];
@@ -60,7 +60,7 @@ export class RealtimeObject {
     this._eventEmitterInternal = new this._client.EventEmitter(this._client.logger);
     this._eventEmitterPublic = new this._client.EventEmitter(this._client.logger);
     this._objectsPool = new ObjectsPool(this);
-    this._syncObjectsPool = [];
+    this._syncObjectsPool = new Map<string, ObjectMessage>();
     this._bufferedObjectOperations = [];
     this._pathObjectSubscriptionRegister = new PathObjectSubscriptionRegister(this);
     // use server-provided objectsGCGracePeriod if available, and subscribe to new connectionDetails that can be emitted as part of the RTN24
@@ -198,7 +198,7 @@ export class RealtimeObject {
       // if no HAS_OBJECTS flag received on attach, we can end sync sequence immediately and treat it as no objects on a channel.
       // reset the objects pool to its initial state, and emit update events so subscribers to root object get notified about changes.
       this._objectsPool.resetToInitialPool(true); // RTO4b1, RTO4b2
-      this._syncObjectsPool = []; // RTO4b3
+      this._syncObjectsPool.clear(); // RTO4b3
       this._endSync(); // RTO4b4
     }
   }
@@ -216,7 +216,7 @@ export class RealtimeObject {
       case 'failed':
         // do not emit data update events as the actual current state of Objects data is unknown when we're in these channel states
         this._objectsPool.clearObjectsData(false);
-        this._syncObjectsPool = [];
+        this._syncObjectsPool.clear();
         break;
     }
   }
@@ -261,7 +261,7 @@ export class RealtimeObject {
   private _startNewSync(syncId?: string, syncCursor?: string): void {
     // need to discard all buffered object operation messages on new sync start
     this._bufferedObjectOperations = [];
-    this._syncObjectsPool = [];
+    this._syncObjectsPool.clear();
     this._currentSyncId = syncId;
     this._currentSyncCursor = syncCursor;
     this._stateChange(ObjectsState.syncing);
@@ -275,7 +275,7 @@ export class RealtimeObject {
     this._applyObjectMessages(this._bufferedObjectOperations);
 
     this._bufferedObjectOperations = [];
-    this._syncObjectsPool = []; // RTO5c4
+    this._syncObjectsPool.clear(); // RTO5c4
     this._currentSyncId = undefined; // RTO5c3
     this._currentSyncCursor = undefined; // RTO5c3
     this._stateChange(ObjectsState.synced);
@@ -301,7 +301,7 @@ export class RealtimeObject {
   }
 
   private _applySync(): void {
-    if (this._syncObjectsPool.length === 0) {
+    if (this._syncObjectsPool.size === 0) {
       return;
     }
 
@@ -312,8 +312,7 @@ export class RealtimeObject {
     }[] = [];
 
     // RTO5c1
-    for (const objectMessage of this._syncObjectsPool) {
-      const objectId = objectMessage.object?.objectId!;
+    for (const [objectId, objectMessage] of this._syncObjectsPool) {
       receivedObjectIds.add(objectId);
       const existingObject = this._objectsPool.get(objectId);
 
@@ -370,7 +369,57 @@ export class RealtimeObject {
         continue;
       }
 
-      this._syncObjectsPool.push(objectMessage);
+      // RTO5b2 - partial object sync handling
+      const objectState = objectMessage.object;
+      const objectId = objectState.objectId;
+
+      if (objectState.counter) {
+        // RTO5b2a, RTO5b2b2 - counter objects have a bounded size and will never be split
+        // across multiple sync messages, so they can be stored directly without merging.
+        this._syncObjectsPool.set(objectId, objectMessage);
+        continue;
+      }
+
+      if (objectState.map) {
+        const existingEntry = this._syncObjectsPool.get(objectId);
+
+        if (!existingEntry) {
+          // RTO5b2a - no ObjectState with the given objectId exists yet, store it
+          this._syncObjectsPool.set(objectId, objectMessage);
+        } else {
+          // RTO5b2b, RTO5b2b1 - merge entries from the new ObjectState into the existing one
+          this._mergeMapSyncState(existingEntry, objectMessage);
+        }
+        continue;
+      }
+
+      this._client.Logger.logAction(
+        this._client.logger,
+        this._client.Logger.LOG_MAJOR,
+        'RealtimeObject._applyObjectSyncMessages()',
+        `received unsupported object state message during OBJECT_SYNC, expected 'counter' or 'map' to be present, skipping message; message id: ${objectMessage.id}, channel: ${this._channel.name}`,
+      );
+    }
+  }
+
+  /**
+   * Merges map entries from a partial sync message into an existing entry.
+   * Other fields on the ObjectState envelope are identical across all partial messages
+   * for the same object, so only the entries need to be merged.
+   * @spec RTO5b2b1
+   */
+  private _mergeMapSyncState(existingEntry: ObjectMessage, newObjectMessage: ObjectMessage): void {
+    const existingObjectState = existingEntry.object!;
+    const newObjectState = newObjectMessage.object!;
+
+    if (!existingObjectState.map!.entries) {
+      existingObjectState.map!.entries = {};
+    }
+
+    if (newObjectState.map?.entries) {
+      // During partial sync, no two messages contain the same map key,
+      // so entries can be merged directly without conflict checking.
+      Object.assign(existingObjectState.map!.entries, newObjectState.map.entries);
     }
   }
 

test/realtime/init.test.js

@@ -43,7 +43,7 @@ define(['ably', 'shared_helper', 'chai'], function (Ably, Helper, chai) {
                     return transport.recvRequest.recvUri;
                   })();
               try {
-                expect(connectUri.indexOf('v=5') > -1, 'Check uri includes v=5').to.be.ok;
+                expect(connectUri.indexOf('v=6') > -1, 'Check uri includes v=6').to.be.ok;
               } catch (err) {
                 helper.closeAndFinish(done, realtime, err);
                 return;

test/realtime/liveobjects.test.js

@@ -804,6 +804,52 @@ define(['ably', 'shared_helper', 'chai', 'liveobjects', 'liveobjects_helper'], f
           },
         },
 
+        {
+          description: 'OBJECT_SYNC does not break when receiving an unknown object type',
+          action: async (ctx) => {
+            const { channel, objectsHelper } = ctx;
+
+            // first message: unknown object type (no counter or map field set)
+            await objectsHelper.processObjectStateMessageOnChannel({
+              channel,
+              syncSerial: 'serial:cursor',
+              state: [
+                {
+                  object: {
+                    objectId: 'unknown:object123',
+                    siteTimeserials: { aaa: lexicoTimeserial('aaa', 0, 0) },
+                    tombstone: false,
+                    // intentionally not setting counter or map fields
+                  },
+                },
+              ],
+            });
+
+            // second message: root with a key, ends sync sequence
+            await objectsHelper.processObjectStateMessageOnChannel({
+              channel,
+              syncSerial: 'serial:',
+              state: [
+                objectsHelper.mapObject({
+                  objectId: 'root',
+                  siteTimeserials: { aaa: lexicoTimeserial('aaa', 0, 0) },
+                  initialEntries: {
+                    foo: { timeserial: lexicoTimeserial('aaa', 0, 0), data: { string: 'bar' } },
+                  },
+                }),
+              ],
+            });
+
+            const root = await channel.object.get();
+
+            // verify root has the expected key - SDK should not break due to unknown object type
+            expect(root.get('foo').value()).to.equal(
+              'bar',
+              'Check root has correct value after unknown object type in sync',
+            );
+          },
+        },
+
         {
           description: 'OBJECT_SYNC sequence with "tombstone=true" for an object creates tombstoned object',
           action: async (ctx) => {
@@ -1132,6 +1178,161 @@ define(['ably', 'shared_helper', 'chai', 'liveobjects', 'liveobjects_helper'], f
             ).to.be.true;
           },
         },
+
+        {
+          description: 'partial OBJECT_SYNC builds object tree across multiple messages',
+          action: async (ctx) => {
+            const { channel, objectsHelper, entryPathObject } = ctx;
+
+            const counterId = objectsHelper.fakeCounterObjectId();
+            const mapId = objectsHelper.fakeMapObjectId();
+
+            // send three separate OBJECT_SYNC messages: one for root, one for counter, one for map
+            await objectsHelper.processObjectStateMessageOnChannel({
+              channel,
+              syncSerial: 'serial:cursor1',
+              state: [
+                objectsHelper.mapObject({
+                  objectId: 'root',
+                  siteTimeserials: { aaa: lexicoTimeserial('aaa', 0, 0) },
+                  initialEntries: {
+                    stringKey: { timeserial: lexicoTimeserial('aaa', 0, 0), data: { string: 'hello' } },
+                    counter: { timeserial: lexicoTimeserial('aaa', 0, 0), data: { objectId: counterId } },
+                    map: { timeserial: lexicoTimeserial('aaa', 0, 0), data: { objectId: mapId } },
+                  },
+                }),
+              ],
+            });
+
+            await objectsHelper.processObjectStateMessageOnChannel({
+              channel,
+              syncSerial: 'serial:cursor2',
+              state: [
+                objectsHelper.counterObject({
+                  objectId: counterId,
+                  siteTimeserials: { aaa: lexicoTimeserial('aaa', 0, 0) },
+                  initialCount: 10,
+                  materialisedCount: 5,
+                }),
+              ],
+            });
+
+            await objectsHelper.processObjectStateMessageOnChannel({
+              channel,
+              syncSerial: 'serial:', // end sync sequence
+              state: [
+                objectsHelper.mapObject({
+                  objectId: mapId,
+                  siteTimeserials: { aaa: lexicoTimeserial('aaa', 0, 0) },
+                  initialEntries: {
+                    foo: { timeserial: lexicoTimeserial('aaa', 0, 0), data: { string: 'bar' } },
+                  },
+                  materialisedEntries: {
+                    baz: { timeserial: lexicoTimeserial('bbb', 0, 0), data: { string: 'qux' } },
+                  },
+                }),
+              ],
+            });
+
+            expect(entryPathObject.get('stringKey').value()).to.equal('hello', 'Check root has correct string value');
+            expect(entryPathObject.get('counter').value()).to.equal(15, 'Check counter has correct aggregated value');
+            expect(entryPathObject.get('map').get('foo').value()).to.equal('bar', 'Check map has initial entries');
+            expect(entryPathObject.get('map').get('baz').value()).to.equal('qux', 'Check map has materialised entries');
+          },
+        },
+
+        {
+          description: 'partial OBJECT_SYNC merges map entries across multiple messages for the same objectId',
+          action: async (ctx) => {
+            const { channel, objectsHelper, entryPathObject } = ctx;
+
+            const mapId = objectsHelper.fakeMapObjectId();
+
+            await objectsHelper.processObjectStateMessageOnChannel({
+              channel,
+              syncSerial: 'serial:cursor1',
+              state: [
+                objectsHelper.mapObject({
+                  objectId: 'root',
+                  siteTimeserials: { aaa: lexicoTimeserial('aaa', 0, 0) },
+                  initialEntries: {
+                    map: { timeserial: lexicoTimeserial('aaa', 0, 0), data: { objectId: mapId } },
+                  },
+                }),
+              ],
+            });
+
+            await objectsHelper.processObjectStateMessageOnChannel({
+              channel,
+              syncSerial: 'serial:cursor2',
+              state: [
+                objectsHelper.mapObject({
+                  objectId: mapId,
+                  siteTimeserials: { aaa: lexicoTimeserial('aaa', 0, 0) },
+                  // initialEntries are the same across all partial messages
+                  initialEntries: {
+                    initialKey: { timeserial: lexicoTimeserial('aaa', 0, 0), data: { string: 'initial' } },
+                  },
+                  // materialisedEntries are merged across partial messages
+                  materialisedEntries: {
+                    key1: { timeserial: lexicoTimeserial('aaa', 0, 0), data: { number: 1 } },
+                    key2: { timeserial: lexicoTimeserial('aaa', 0, 0), data: { string: 'two' } },
+                  },
+                }),
+              ],
+            });
+
+            await objectsHelper.processObjectStateMessageOnChannel({
+              channel,
+              syncSerial: 'serial:cursor3',
+              state: [
+                objectsHelper.mapObject({
+                  objectId: mapId,
+                  siteTimeserials: { aaa: lexicoTimeserial('aaa', 0, 0) },
+                  initialEntries: {
+                    initialKey: { timeserial: lexicoTimeserial('aaa', 0, 0), data: { string: 'initial' } },
+                  },
+                  materialisedEntries: {
+                    key3: { timeserial: lexicoTimeserial('aaa', 0, 0), data: { number: 3 } },
+                    key4: { timeserial: lexicoTimeserial('aaa', 0, 0), data: { boolean: true } },
+                  },
+                }),
+              ],
+            });
+
+            await objectsHelper.processObjectStateMessageOnChannel({
+              channel,
+              syncSerial: 'serial:', // end sync sequence
+              state: [
+                objectsHelper.mapObject({
+                  objectId: mapId,
+                  siteTimeserials: { aaa: lexicoTimeserial('aaa', 0, 0) },
+                  initialEntries: {
+                    initialKey: { timeserial: lexicoTimeserial('aaa', 0, 0), data: { string: 'initial' } },
+                  },
+                  materialisedEntries: {
+                    key5: { timeserial: lexicoTimeserial('aaa', 0, 0), data: { string: 'five' } },
+                  },
+                }),
+              ],
+            });
+
+            const map = entryPathObject.get('map');
+
+            // verify initial entries are applied
+            expect(map.get('initialKey').value()).to.equal(
+              'initial',
+              'Check keys from the create operations are present',
+            );
+
+            // verify all materialised entries were merged
+            expect(map.get('key1').value()).to.equal(1, 'Check key1 from first partial sync');
+            expect(map.get('key2').value()).to.equal('two', 'Check key2 from first partial sync');
+            expect(map.get('key3').value()).to.equal(3, 'Check key3 from second partial sync');
+            expect(map.get('key4').value()).to.equal(true, 'Check key4 from second partial sync');
+            expect(map.get('key5').value()).to.equal('five', 'Check key5 from third partial sync');
+          },
+        },
       ];
 
       const applyOperationsScenarios = [

test/rest/http.test.js

@@ -34,7 +34,7 @@ define(['ably', 'shared_helper', 'chai'], function (Ably, Helper, chai) {
 
         // This test should not directly validate version against Defaults.version, as
         // ultimately the version header has been derived from that value.
-        expect(headers['X-Ably-Version']).to.equal('5', 'Verify current version number');
+        expect(headers['X-Ably-Version']).to.equal('6', 'Verify current version number');
         helper.recordPrivateApi('read.Defaults.version');
         expect(headers['Ably-Agent'].indexOf('ably-js/' + Defaults.version) > -1, 'Verify agent').to.be.ok;
         expect(headers['Ably-Agent'].indexOf('custom-agent/0.1.2') > -1, 'Verify custom agent').to.be.ok;