[AIT-38] Add support for partial sync of objects - protocol v6

[VeskeR]

This PR is based on #2159, please review that one first.

Spec for this ably/specification#413, DR https://ably.atlassian.net/wiki/x/AQBxCQE

Resolves AIT-38.

This also removes unnecessary SyncObjectsDataPool class and simplifies a sync logic a bit by moving it to RealtimeObject.

Summary by CodeRabbit

• New Features

  • New synchronization pool implementation to accumulate and merge object sync messages during sync.

• Refactor

  • Internal replacement of the previous sync data pool with the new pool implementation (no public API changes).

• Bug Fixes

  • Unsupported object shapes are now logged and skipped instead of causing errors.
  • Improved merging for partial OBJECT_SYNC sequences, preserving map entries across messages.

• Tests

  • Added OBJECT_SYNC tests covering multi-message builds, partial merges, and unknown-type resilience.

[coderabbitai]

Note

Reviews paused

It looks like this branch is under active development. To avoid overwhelming you with review comments due to an influx of new commits, CodeRabbit has automatically paused this review. You can configure this behavior by changing the reviews.auto_review.auto_pause_after_reviewed_commits setting.

Use the following commands to manage reviews:

• @coderabbitai resume to resume automatic reviews.
• @coderabbitai review to trigger a single review.

Use the checkboxes below for quick actions:

• ▶️ Resume reviews
• 🔍 Trigger review

Walkthrough

Replaces the legacy SyncObjectsDataPool with a new SyncObjectsPool, moves pool logic to shape-based detection (counter vs map), adds map-state merging for partial OBJECT_SYNC messages, updates RealtimeObject to use the new pool, deletes the old module, and adds tests for multi-message sync scenarios.

Changes

Cohort / File(s)	Summary
Module Report & Bundling scripts/moduleReport.ts	Switched allowed LiveObjects bundle file reference from src/plugins/liveobjects/syncobjectsdatapool.ts to src/plugins/liveobjects/syncobjectspool.ts.
New Pool Implementation src/plugins/liveobjects/syncobjectspool.ts	Added SyncObjectsPool class: accumulates ObjectMessage entries by objectId, validates shapes, merges partial map state via _mergeMapSyncState(), and exposes entries/size/isEmpty/clear/applyObjectSyncMessages.
Removed Legacy Pool src/plugins/liveobjects/syncobjectsdatapool.ts	Deleted entire module and its public types/classes (SyncObjectsDataPool, LiveObjectDataEntry, LiveCounterDataEntry, LiveMapDataEntry, AnyDataEntry).
RealtimeObject Integration src/plugins/liveobjects/realtimeobject.ts	Replaced SyncObjectsDataPool with SyncObjectsPool across imports, field names, initialization, clear/isEmpty/entries calls, and _applySync() logic; switched from type-driven creation to shape-based creation (check object?.counter vs object?.map) and guarded unsupported shapes.
Tests test/realtime/liveobjects.test.js	Added OBJECT_SYNC tests covering multi-message object tree construction, resilience to unknown object types, and partial map entry merging across multiple sync messages (≈+188 lines).

Sequence Diagram

sequenceDiagram
    participant RO as RealtimeObject
    participant Pool as SyncObjectsPool
    participant Map as LiveMap
    participant Counter as LiveCounter

    RO->>Pool: applyObjectSyncMessages(objectMessages)
    activate Pool

    loop each ObjectMessage
        Pool->>Pool: validate object exists
        alt first message for objectId
            Pool->>Pool: store ObjectMessage
        else existing entry
            alt object has counter
                Pool->>Pool: log error, skip partial counter
            else object has map
                Pool->>Pool: _mergeMapSyncState(existing, new)
            end
        end
    end
    deactivate Pool

    RO->>Pool: entries()
    loop each [objectId, objectMessage]
        alt objectMessage.object?.counter
            RO->>Counter: LiveCounter.fromObjectState(objectMessage)
        else objectMessage.object?.map
            RO->>Map: LiveMap.fromObjectState(objectMessage)
        else unknown shape
            RO->>RO: log skip
        end
    end

Loading

Estimated code review effort

🎯 4 (Complex) | ⏱️ ~45 minutes

Poem

🐇 I hopped through syncs both near and far,
Collected objects, stitched each map and star,
Counters kept whole, maps merged piece by piece,
Old pool retired — the new one brings us peace,
A rabbit cheers for tidy sync release!

🚥 Pre-merge checks | ✅ 2 | ❌ 1

❌ Failed checks (1 warning)

Check name	Status	Explanation	Resolution
Docstring Coverage	⚠️ Warning	Docstring coverage is 0.00% which is insufficient. The required threshold is 80.00%.	Write docstrings for the functions missing them to satisfy the coverage threshold.

✅ Passed checks (2 passed)

Check name	Status	Explanation
Description Check	✅ Passed	Check skipped - CodeRabbit’s high-level summary is enabled.
Title check	✅ Passed	The title 'AIT-38 Add support for partial sync of objects - protocol v6' clearly and concisely describes the main change: adding support for partial object synchronization in protocol v6, which aligns with the primary objective of the PR.

_{✏️ Tip: You can configure your own custom pre-merge checks in the settings.}

✨ Finishing Touches

🧪 Generate unit tests (beta)

• Create PR with unit tests
• Post copyable unit tests in a comment
• Commit unit tests in branch AIT-38/partial-objects-sync

---

Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out.

❤️ Share

• X
• Mastodon
• Reddit
• LinkedIn

_{Comment @coderabbitai help to get the list of available commands and usage tips.}

[coderabbitai]

Actionable comments posted: 1

🤖 Fix all issues with AI agents

In `@src/plugins/liveobjects/realtimeobject.ts`:
- Around line 372-381: The code uses objectState.objectId as a Map key without
validation which can cause messages to pile up under undefined; inside the loop
in realtimeobject.ts where objectState and objectId are extracted (symbols:
objectState, objectId, this._syncObjectsPool), add a guard that checks if
objectId is null/undefined and if so skip storing/merging that message and emit
a warning/error (use the existing logger on the class, e.g., this.logger.warn or
console.warn) before continue; ensure no entry is created in _syncObjectsPool
for missing objectId and include the offending message or minimal context in the
log to aid debugging.

🧹 Nitpick comments (1)

test/realtime/liveobjects.test.js (1)

1182-1242: Consider asserting on a post-sync root to avoid timing races.
The assertions currently use entryPathObject created before the partial sync; if OBJECT_SYNC application ever becomes async, this can become flaky. Grabbing the root after the final sync message guarantees you’re asserting on the fresh state.

♻️ Suggested tweak

-            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');
+            const root = await channel.object.get(); // ensure OBJECT_SYNC has fully applied
+            expect(root.get('stringKey').value()).to.equal('hello', 'Check root has correct string value');
+            expect(root.get('counter').value()).to.equal(15, 'Check counter has correct aggregated value');
+            expect(root.get('map').get('foo').value()).to.equal('bar', 'Check map has initial entries');
+            expect(root.get('map').get('baz').value()).to.equal('qux', 'Check map has materialised entries');

[lawrence-forooghian]

LGTM, some minor comments. I'd suggest we finalise the spec (have left some comments there) and update the spec point references before merging.

[coderabbitai]

🧹 Nitpick comments (1)

src/plugins/liveobjects/realtimeobject.ts (1)

411-423: Consider extracting typed locals to avoid repeated non-null assertions.

The ! assertions on Lines 412, 413, 415, 416, 422 are safe because callers guarantee .object and .map are set, but they repeat 5 times and obscure that guarantee for future readers. Extracting to typed locals at the top makes the precondition explicit once.

Proposed refactor

  private _mergeMapSyncState(existingEntry: ObjectMessage, newObjectMessage: ObjectMessage): void {
-   const existingObjectState = existingEntry.object!;
-   const newObjectState = newObjectMessage.object!;
+   const existingMap = existingEntry.object!.map!;
+   const newMap = newObjectMessage.object!.map;

-   if (!existingObjectState.map!.entries) {
-     existingObjectState.map!.entries = {};
+   if (!existingMap.entries) {
+     existingMap.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);
+   if (newMap?.entries) {
+     Object.assign(existingMap.entries, newMap.entries);
    }
  }

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In `@src/plugins/liveobjects/realtimeobject.ts` around lines 411 - 423, In
_mergeMapSyncState, remove repeated non-null assertions by extracting typed
local variables for existingEntry.object and newObjectMessage.object and their
.map properties at the top (e.g., const existingObject = existingEntry.object;
const existingMap = existingObject.map; const newObject =
newObjectMessage.object; const newMap = newObject.map), ensure you initialize
existingMap.entries when missing and then use existingMap.entries and
newMap.entries in the subsequent logic (including the Object.assign call) so the
function no longer repeats the `!` assertions.

🤖 Prompt for all review comments with AI agents

Verify each finding against the current code and only fix it if needed.

Duplicate comments:
In `@src/plugins/liveobjects/realtimeobject.ts`:
- Around line 372-381: The code currently reads objectId = objectState.objectId
and uses it as a Map key (this._syncObjectsPool.set) without validating it; add
an early guard after extracting objectState (the same place where you check
!objectMessage.object) to ensure objectId is present and valid (e.g., if
(!objectId) { log/warn and continue; }) so malformed messages do not insert
under undefined; update the handling around objectState/objectId in the same
block to skip and optionally log the offending objectMessage when objectId is
missing.

---

Nitpick comments:
In `@src/plugins/liveobjects/realtimeobject.ts`:
- Around line 411-423: In _mergeMapSyncState, remove repeated non-null
assertions by extracting typed local variables for existingEntry.object and
newObjectMessage.object and their .map properties at the top (e.g., const
existingObject = existingEntry.object; const existingMap = existingObject.map;
const newObject = newObjectMessage.object; const newMap = newObject.map), ensure
you initialize existingMap.entries when missing and then use existingMap.entries
and newMap.entries in the subsequent logic (including the Object.assign call) so
the function no longer repeats the `!` assertions.

[VeskeR]

I have removed spec references temporarily as I'm going to rewrite the spec next. Will re-add them once the spec is agreed

[coderabbitai]

Actionable comments posted: 1

🧹 Nitpick comments (2)

test/rest/http.test.js (1)

35-37: Optional: clarify the comment – it references the wrong Defaults property.

The comment says the assertion should not validate against Defaults.version, but X-Ably-Version is set from Defaults.protocolVersion (an integer), not Defaults.version (the SDK semver string). The real reason to hardcode '6' is to avoid a circular assertion against Defaults.protocolVersion.toString() itself.

✏️ Suggested comment clarification

-        // This test should not directly validate version against Defaults.version, as
-        // ultimately the version header has been derived from that value.
+        // Hardcoded rather than using Defaults.protocolVersion.toString() to avoid a
+        // circular assertion — the header value is derived directly from that constant.
         expect(headers['X-Ably-Version']).to.equal('6', 'Verify current version number');

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In `@test/rest/http.test.js` around lines 35 - 37, Update the test comment to
reference Defaults.protocolVersion (an integer) instead of Defaults.version and
clarify why the header is hardcoded: the assertion compares
headers['X-Ably-Version'] to '6' to avoid a circular check against
Defaults.protocolVersion.toString(); modify the comment around the expect(...)
line (the assertion for headers['X-Ably-Version']) to state that the value
originates from Defaults.protocolVersion and is intentionally hardcoded in the
test to prevent validating against the source of the header itself.

src/plugins/liveobjects/syncobjectspool.ts (1)

54-82: First message for an objectId is stored unconditionally, but subsequent messages assume type consistency.

Line 56 stores the first ObjectMessage for a given objectId regardless of whether it carries a valid counter or map (i.e., an unsupported type is also stored). Subsequent messages for the same objectId are then routed by type—counters are rejected (Line 60-69), maps are merged (Line 72-74), and unsupported types are skipped (Line 77-82). This means the routing for the second message depends on the second message's type, not the first's, which could lead to inconsistent merging (e.g., first message is a counter, second is a map → merges into a counter entry).

Consider validating the type on the first store as well, or at least validating type consistency between existing and new entries before merging.

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In `@src/plugins/liveobjects/syncobjectspool.ts` around lines 54 - 82, The code
stores the first ObjectMessage into this._pool unconditionally which allows a
later message of a different type to be merged into an incompatible entry;
update SyncObjectsPool.applyObjectSyncMessages() to enforce type validation:
when inserting the initial entry (the branch that currently does
this._pool.set(objectId, objectMessage)) only accept/store messages that contain
a known type (objectState.counter or objectState.map), otherwise log and skip,
and additionally before calling _mergeMapSyncState or rejecting as a counter,
compare the existingEntry's type (inspect existingEntry.objectState.map/counter)
against the incoming objectMessage.objectState and if they differ log a clear
error and skip the message to avoid cross-type merges.

🤖 Prompt for all review comments with AI agents

Verify each finding against the current code and only fix it if needed.

Inline comments:
In `@src/plugins/liveobjects/syncobjectspool.ts`:
- Around line 89-110: The _mergeMapSyncState function assumes
existingEntry.object.map exists; add a guard so if existingObjectState.map is
undefined (i.e., the existing entry is a non-map), initialize it to a map
envelope with an empty entries object before touching entries; then proceed to
merge newObjectState.map.entries as before (use existingObjectState.map = {
entries: {} } or equivalent to preserve type expectations), ensuring you only
call Object.assign when newObjectState.map?.entries is present.

---

Duplicate comments:
In `@src/plugins/liveobjects/syncobjectspool.ts`:
- Around line 50-57: The code uses objectState.objectId as a Map key without
validating it, which can insert undefined/empty keys into this._pool; add a
guard after const objectId = objectState.objectId to validate it (non-null,
non-undefined, non-empty string/invalid type) and skip processing this
objectMessage when invalid (e.g., log/warn and continue) so you never call
this._pool.get/set with a bogus key; update the block around objectId,
existingEntry and this._pool to perform this validation before accessing the
Map.

---

Nitpick comments:
In `@src/plugins/liveobjects/syncobjectspool.ts`:
- Around line 54-82: The code stores the first ObjectMessage into this._pool
unconditionally which allows a later message of a different type to be merged
into an incompatible entry; update SyncObjectsPool.applyObjectSyncMessages() to
enforce type validation: when inserting the initial entry (the branch that
currently does this._pool.set(objectId, objectMessage)) only accept/store
messages that contain a known type (objectState.counter or objectState.map),
otherwise log and skip, and additionally before calling _mergeMapSyncState or
rejecting as a counter, compare the existingEntry's type (inspect
existingEntry.objectState.map/counter) against the incoming
objectMessage.objectState and if they differ log a clear error and skip the
message to avoid cross-type merges.

In `@test/rest/http.test.js`:
- Around line 35-37: Update the test comment to reference
Defaults.protocolVersion (an integer) instead of Defaults.version and clarify
why the header is hardcoded: the assertion compares headers['X-Ably-Version'] to
'6' to avoid a circular check against Defaults.protocolVersion.toString();
modify the comment around the expect(...) line (the assertion for
headers['X-Ably-Version']) to state that the value originates from
Defaults.protocolVersion and is intentionally hardcoded in the test to prevent
validating against the source of the header itself.

[lawrence-forooghian]

I have removed spec references temporarily as I'm going to rewrite the spec next. Will re-add them once the spec is agreed

I've approved the spec PR — please could you add these references here?

[VeskeR]

I've approved the spec PR — please could you add these references here?

Added spec references in https://github.com/ably/ably-js/compare/1ff0df106e844a91ab2c695f56466b3d3edff34a..83e7d8f206be609721bf1f017b0ce960e51d9fab

[lawrence-forooghian]

LGTM, happy to approve once rebased and squashed

[VeskeR]

LGTM, happy to approve once rebased and squashed

Rebased on top of #2159 (currently d348a98) and squashed. Please approve if you're happy, I'm going to merge once #2159 lands