Support for partial objects sync

VeskeR · VeskeR · commit 1052acfc987a · 2026-02-24T17:03:58.000Z
See spec ably/specification#413 and a DR [1] Resolves AIT-38 [1] https://ably.atlassian.net/wiki/x/AQBxCQE
diff --git a/src/plugins/liveobjects/realtimeobject.ts b/src/plugins/liveobjects/realtimeobject.ts
@@ -163,11 +163,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
     }
   }
 
@@ -443,34 +443,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
diff --git a/src/plugins/liveobjects/syncobjectspool.ts b/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);
+    }
   }
 }
diff --git a/test/realtime/liveobjects.test.js b/test/realtime/liveobjects.test.js
@@ -961,6 +961,86 @@ define(['ably', 'shared_helper', 'chai', 'liveobjects', 'liveobjects_helper'], f
           },
         },
 
+        {
+          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();
+
+            // assign map object to root
+            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 } },
+                  },
+                }),
+              ],
+            });
+
+            // send partial sync messages for the same map object, each with different materialised entries.
+            // initialEntries are identical across all partial messages for the same object - a server guarantee.
+            const partialMessages = [
+              {
+                syncSerial: 'serial:cursor2',
+                materialisedEntries: {
+                  key1: { timeserial: lexicoTimeserial('aaa', 0, 0), data: { number: 1 } },
+                  key2: { timeserial: lexicoTimeserial('aaa', 0, 0), data: { string: 'two' } },
+                },
+              },
+              {
+                syncSerial: 'serial:cursor3',
+                materialisedEntries: {
+                  key3: { timeserial: lexicoTimeserial('aaa', 0, 0), data: { number: 3 } },
+                  key4: { timeserial: lexicoTimeserial('aaa', 0, 0), data: { boolean: true } },
+                },
+              },
+              {
+                syncSerial: 'serial:', // end sync sequence
+                materialisedEntries: {
+                  key5: { timeserial: lexicoTimeserial('aaa', 0, 0), data: { string: 'five' } },
+                },
+              },
+            ];
+
+            for (const partial of partialMessages) {
+              await objectsHelper.processObjectStateMessageOnChannel({
+                channel,
+                syncSerial: partial.syncSerial,
+                state: [
+                  objectsHelper.mapObject({
+                    objectId: mapId,
+                    siteTimeserials: { aaa: lexicoTimeserial('aaa', 0, 0) },
+                    initialEntries: {
+                      initialKey: { timeserial: lexicoTimeserial('aaa', 0, 0), data: { string: 'initial' } },
+                    },
+                    materialisedEntries: partial.materialisedEntries,
+                  }),
+                ],
+              });
+            }
+
+            const map = entryPathObject.get('map');
+
+            expect(map.get('initialKey').value()).to.equal(
+              'initial',
+              'Check keys from the create operation are present',
+            );
+
+            // check that materialised entries from all partial messages 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');
+          },
+        },
+
         {
           allTransportsAndProtocols: true,
           description: 'LiveCounter is initialized with initial value from OBJECT_SYNC sequence',
