Protocol v6 and support for partial objects sync

See spec ably/specification#413 and a DR [1]

Resolves AIT-38

[1] https://ably.atlassian.net/wiki/x/AQBxCQE

Author: VeskeR

src/plugins/liveobjects/realtimeobject.ts

@@ -154,11 +154,11 @@ export class RealtimeObject {
     }
 
     // RTO5a3 - continue current sync sequence
-    this._syncObjectsPool.applyObjectSyncMessages(objectMessages); // RTO5b
+    this._syncObjectsPool.applyObjectSyncMessages(objectMessages); // RTO5f
 
     // RTO5a4 - if this is the last (or only) message in a sequence of sync updates, end the sync
     if (!syncCursor) {
-      this._endSync();
+      this._endSync(); // RTO5c
     }
   }
 
@@ -312,34 +312,34 @@ export class RealtimeObject {
     }[] = [];
 
     // RTO5c1
-    for (const [objectId, entry] of this._syncObjectsPool.entries()) {
+    for (const [objectId, objectMessage] of this._syncObjectsPool.entries()) {
       receivedObjectIds.add(objectId);
       const existingObject = this._objectsPool.get(objectId);
 
       // RTO5c1a
       if (existingObject) {
-        const update = existingObject.overrideWithObjectState(entry.objectMessage); // RTO5c1a1
+        const update = existingObject.overrideWithObjectState(objectMessage); // RTO5c1a1
         // store updates to call subscription callbacks for all of them once the sync sequence is completed.
         // this will ensure that clients get notified about the changes only once everything has been applied.
         existingObjectUpdates.push({ object: existingObject, update });
         continue;
       }
 
-      // RTO5c1b,
+      // RTO5c1b
       let newObject: LiveObject;
-      // assign to a variable so TS doesn't complain about 'never' type in the default case
-      const objectType = entry.objectType;
-      switch (objectType) {
-        case 'LiveCounter':
-          newObject = LiveCounter.fromObjectState(this, entry.objectMessage); // RTO5c1b1a
-          break;
-
-        case 'LiveMap':
-          newObject = LiveMap.fromObjectState(this, entry.objectMessage); // RTO5c1b1b
-          break;
-
-        default:
-          throw new this._client.ErrorInfo(`Unknown LiveObject type: ${objectType}`, 50000, 500); // RTO5c1b1c
+      if (objectMessage.object?.counter) {
+        newObject = LiveCounter.fromObjectState(this, objectMessage); // RTO5c1b1a
+      } else if (objectMessage.object?.map) {
+        newObject = LiveMap.fromObjectState(this, objectMessage); // RTO5c1b1b
+      } else {
+        // RTO5c1b1c
+        this._client.Logger.logAction(
+          this._client.logger,
+          this._client.Logger.LOG_MAJOR,
+          'RealtimeObject._applySync()',
+          `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}`,
+        );
+        continue;
       }
 
       this._objectsPool.set(objectId, newObject); // RTO5c1b1

src/plugins/liveobjects/syncobjectspool.ts

@@ -3,34 +3,19 @@ import type RealtimeChannel from 'common/lib/client/realtimechannel';
 import { ObjectMessage } from './objectmessage';
 import { RealtimeObject } from './realtimeobject';
 
-export interface LiveObjectDataEntry {
-  objectMessage: ObjectMessage;
-  objectType: 'LiveMap' | 'LiveCounter';
-}
-
-export interface LiveCounterDataEntry extends LiveObjectDataEntry {
-  objectType: 'LiveCounter';
-}
-
-export interface LiveMapDataEntry extends LiveObjectDataEntry {
-  objectType: 'LiveMap';
-}
-
-export type AnyDataEntry = LiveCounterDataEntry | LiveMapDataEntry;
-
 /**
  * @internal
- * @spec RTO5b
  */
 export class SyncObjectsPool {
   private _client: BaseClient;
   private _channel: RealtimeChannel;
-  private _pool: Map<string, AnyDataEntry>;
+  /** Used to accumulate object state during a sync sequence, keyed by object ID */
+  private _pool: Map<string, ObjectMessage>;
 
   constructor(private _realtimeObject: RealtimeObject) {
     this._client = this._realtimeObject.getClient();
     this._channel = this._realtimeObject.getChannel();
-    this._pool = new Map<string, AnyDataEntry>();
+    this._pool = new Map<string, ObjectMessage>();
   }
 
   entries() {
@@ -49,50 +34,83 @@ export class SyncObjectsPool {
     this._pool.clear();
   }
 
+  /** @spec RTO5f */
   applyObjectSyncMessages(objectMessages: ObjectMessage[]): void {
     for (const objectMessage of objectMessages) {
       if (!objectMessage.object) {
         this._client.Logger.logAction(
           this._client.logger,
           this._client.Logger.LOG_MAJOR,
           'SyncObjectsPool.applyObjectSyncMessages()',
-          `object message is received during OBJECT_SYNC without 'object' field, skipping message; message id: ${objectMessage.id}, channel: ${this._channel.name}`,
+          `received OBJECT_SYNC message without 'object' field, skipping message; message id: ${objectMessage.id}, channel: ${this._channel.name}`,
         );
         continue;
       }
 
       const objectState = objectMessage.object;
+      const objectId = objectState.objectId;
+      const existingEntry = this._pool.get(objectId);
 
+      if (!existingEntry) {
+        // RTO5f1 - no entry with this objectId exists yet, store it
+        this._pool.set(objectId, objectMessage);
+        continue;
+      }
+
+      // RTO5f2 - an object is split across multiple sync messages, merge the new state with the existing entry in the pool based on the object type
       if (objectState.counter) {
-        this._pool.set(objectState.objectId, this._createLiveCounterDataEntry(objectMessage));
-      } else if (objectState.map) {
-        this._pool.set(objectState.objectId, this._createLiveMapDataEntry(objectMessage));
-      } else {
+        // RTO5f2b - counter objects have a bounded size and should never be split
+        // across multiple sync messages. Skip the unexpected partial state.
         this._client.Logger.logAction(
           this._client.logger,
-          this._client.Logger.LOG_MAJOR,
+          this._client.Logger.LOG_ERROR,
           'SyncObjectsPool.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}`,
+          `received partial OBJECT_SYNC state for a counter object, skipping message; object id: ${objectId}, message id: ${objectMessage.id}, channel: ${this._channel.name}`,
         );
+        continue;
       }
-    }
-  }
 
-  private _createLiveCounterDataEntry(objectMessage: ObjectMessage): LiveCounterDataEntry {
-    const newEntry: LiveCounterDataEntry = {
-      objectMessage,
-      objectType: 'LiveCounter',
-    };
+      if (objectState.map) {
+        // RTO5f2a
+        this._mergeMapSyncState(existingEntry, objectMessage);
+        continue;
+      }
 
-    return newEntry;
+      // RTO5f2c
+      this._client.Logger.logAction(
+        this._client.logger,
+        this._client.Logger.LOG_MAJOR,
+        'SyncObjectsPool.applyObjectSyncMessages()',
+        `received OBJECT_SYNC message with unsupported object type, expected 'counter' or 'map' to be present, skipping message; message id: ${objectMessage.id}, channel: ${this._channel.name}`,
+      );
+    }
   }
 
-  private _createLiveMapDataEntry(objectMessage: ObjectMessage): LiveMapDataEntry {
-    const newEntry: LiveMapDataEntry = {
-      objectMessage,
-      objectType: 'LiveMap',
-    };
+  /**
+   * Merges map entries from a partial sync message into an existing entry in the pool.
+   * @spec RTO5f2a
+   */
+  private _mergeMapSyncState(existingEntry: ObjectMessage, newObjectMessage: ObjectMessage): void {
+    const existingObjectState = existingEntry.object!;
+    const newObjectState = newObjectMessage.object!;
+
+    if (newObjectState.tombstone) {
+      // RTO5f2a1 - a tombstone flag on any partial message takes precedence over previously accumulated entries
+      this._pool.set(existingObjectState.objectId, newObjectMessage);
+      return;
+    }
 
-    return newEntry;
+    // Other fields on the ObjectState envelope (such as siteTimeserials) and the map envelope
+    // (such as semantics) are identical across all partial messages for the same object,
+    // so only the entries need to be merged.
+    if (!existingObjectState.map!.entries) {
+      existingObjectState.map!.entries = {};
+    }
+
+    if (newObjectState.map?.entries) {
+      // RTO5f2a2 - 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);
+    }
   }
 }