Allow SyncObjectsPool entries to contain additional data

This is preparation for tombstoning an object per [1]; we'll need the
ObjectMessage's serialTimestamp.

[1] ably/specification#350

Author: lawrence-forooghian

Sources/AblyLiveObjects/Internal/InternalDefaultRealtimeObjects.swift

@@ -45,7 +45,7 @@ internal final class InternalDefaultRealtimeObjects: Sendable, LiveMapObjectPool
         internal var id: String
 
         /// The `ObjectMessage`s gathered during this sync sequence.
-        internal var syncObjectsPool: [ObjectState]
+        internal var syncObjectsPool: [SyncObjectsPoolEntry]
 
         /// `OBJECT` ProtocolMessages that were received during this sync sequence, to be applied once the sync sequence is complete, per RTO7a.
         internal var bufferedObjectOperations: [InboundObjectMessage]
@@ -276,7 +276,7 @@ internal final class InternalDefaultRealtimeObjects: Sendable, LiveMapObjectPool
             receivedObjectSyncProtocolMessagesContinuation.yield(objectMessages)
 
             // If populated, this contains a full set of sync data for the channel, and should be applied to the ObjectsPool.
-            let completedSyncObjectsPool: [ObjectState]?
+            let completedSyncObjectsPool: [SyncObjectsPoolEntry]?
             // If populated, this contains a set of buffered inbound OBJECT messages that should be applied.
             let completedSyncBufferedObjectOperations: [InboundObjectMessage]?
 
@@ -305,7 +305,13 @@ internal final class InternalDefaultRealtimeObjects: Sendable, LiveMapObjectPool
                 }
 
                 // RTO5b
-                updatedSyncSequence.syncObjectsPool.append(contentsOf: objectMessages.compactMap(\.object))
+                updatedSyncSequence.syncObjectsPool.append(contentsOf: objectMessages.compactMap { objectMessage in
+                    if let object = objectMessage.object {
+                        .init(state: object)
+                    } else {
+                        nil
+                    }
+                })
 
                 syncSequence = updatedSyncSequence
 
@@ -316,7 +322,13 @@ internal final class InternalDefaultRealtimeObjects: Sendable, LiveMapObjectPool
                 }
             } else {
                 // RTO5a5: The sync data is contained entirely within this single OBJECT_SYNC
-                completedSyncObjectsPool = objectMessages.compactMap(\.object)
+                completedSyncObjectsPool = objectMessages.compactMap { objectMessage in
+                    if let object = objectMessage.object {
+                        .init(state: object)
+                    } else {
+                        nil
+                    }
+                }
                 completedSyncBufferedObjectOperations = nil
             }
 

Sources/AblyLiveObjects/Internal/ObjectsPool.swift

@@ -177,7 +177,7 @@ internal struct ObjectsPool {
 
     /// Applies the objects gathered during an `OBJECT_SYNC` to this `ObjectsPool`, per RTO5c1 and RTO5c2.
     internal mutating func applySyncObjectsPool(
-        _ syncObjectsPool: [ObjectState],
+        _ syncObjectsPool: [SyncObjectsPoolEntry],
         logger: AblyPlugin.Logger,
         userCallbackQueue: DispatchQueue,
         clock: SimpleClock,
@@ -191,46 +191,46 @@ internal struct ObjectsPool {
         var updatesToExistingObjects: [ObjectsPool.Entry.DeferredUpdate] = []
 
         // RTO5c1: For each ObjectState member in the SyncObjectsPool list
-        for objectState in syncObjectsPool {
-            receivedObjectIds.insert(objectState.objectId)
+        for syncObjectsPoolEntry in syncObjectsPool {
+            receivedObjectIds.insert(syncObjectsPoolEntry.state.objectId)
 
             // RTO5c1a: If an object with ObjectState.objectId exists in the internal ObjectsPool
-            if let existingEntry = entries[objectState.objectId] {
-                logger.log("Updating existing object with ID: \(objectState.objectId)", level: .debug)
+            if let existingEntry = entries[syncObjectsPoolEntry.state.objectId] {
+                logger.log("Updating existing object with ID: \(syncObjectsPoolEntry.state.objectId)", level: .debug)
 
                 // RTO5c1a1: Override the internal data for the object as per RTLC6, RTLM6
-                let deferredUpdate = existingEntry.replaceData(using: objectState, objectsPool: &self)
+                let deferredUpdate = existingEntry.replaceData(using: syncObjectsPoolEntry.state, objectsPool: &self)
                 // RTO5c1a2: Store this update to emit at end
                 updatesToExistingObjects.append(deferredUpdate)
             } else {
                 // RTO5c1b: If an object with ObjectState.objectId does not exist in the internal ObjectsPool
-                logger.log("Creating new object with ID: \(objectState.objectId)", level: .debug)
+                logger.log("Creating new object with ID: \(syncObjectsPoolEntry.state.objectId)", level: .debug)
 
                 // RTO5c1b1: Create a new LiveObject using the data from ObjectState and add it to the internal ObjectsPool:
                 let newEntry: Entry?
 
-                if objectState.counter != nil {
+                if syncObjectsPoolEntry.state.counter != nil {
                     // RTO5c1b1a: If ObjectState.counter is present, create a zero-value LiveCounter,
                     // set its private objectId equal to ObjectState.objectId and override its internal data per RTLC6
-                    let counter = InternalDefaultLiveCounter.createZeroValued(objectID: objectState.objectId, logger: logger, userCallbackQueue: userCallbackQueue, clock: clock)
-                    _ = counter.replaceData(using: objectState)
+                    let counter = InternalDefaultLiveCounter.createZeroValued(objectID: syncObjectsPoolEntry.state.objectId, logger: logger, userCallbackQueue: userCallbackQueue, clock: clock)
+                    _ = counter.replaceData(using: syncObjectsPoolEntry.state)
                     newEntry = .counter(counter)
-                } else if let objectsMap = objectState.map {
+                } else if let objectsMap = syncObjectsPoolEntry.state.map {
                     // RTO5c1b1b: If ObjectState.map is present, create a zero-value LiveMap,
                     // set its private objectId equal to ObjectState.objectId, set its private semantics
                     // equal to ObjectState.map.semantics and override its internal data per RTLM6
-                    let map = InternalDefaultLiveMap.createZeroValued(objectID: objectState.objectId, semantics: objectsMap.semantics, logger: logger, userCallbackQueue: userCallbackQueue, clock: clock)
-                    _ = map.replaceData(using: objectState, objectsPool: &self)
+                    let map = InternalDefaultLiveMap.createZeroValued(objectID: syncObjectsPoolEntry.state.objectId, semantics: objectsMap.semantics, logger: logger, userCallbackQueue: userCallbackQueue, clock: clock)
+                    _ = map.replaceData(using: syncObjectsPoolEntry.state, objectsPool: &self)
                     newEntry = .map(map)
                 } else {
                     // RTO5c1b1c: Otherwise, log a warning that an unsupported object state message has been received, and discard the current ObjectState without taking any action
-                    logger.log("Unsupported object state message received for objectId: \(objectState.objectId)", level: .warn)
+                    logger.log("Unsupported object state message received for objectId: \(syncObjectsPoolEntry.state.objectId)", level: .warn)
                     newEntry = nil
                 }
 
                 if let newEntry {
                     // Note that we will never replace the root object here, and thus never break the RTO3b invariant that the root object is always a map. This is because the pool always contains a root object and thus we always go through the RTO5c1a branch of the `if` above.
-                    entries[objectState.objectId] = newEntry
+                    entries[syncObjectsPoolEntry.state.objectId] = newEntry
                 }
             }
         }

Sources/AblyLiveObjects/Internal/SyncObjectsPoolEntry.swift

@@ -0,0 +1,6 @@
+import Foundation
+
+/// The contents of the spec's `SyncObjectsPool` that is built during an `OBJECT_SYNC` sync sequence.
+internal struct SyncObjectsPoolEntry {
+    internal var state: ObjectState
+}

Tests/AblyLiveObjectsTests/ObjectsPoolTests.swift

@@ -100,7 +100,7 @@ struct ObjectsPoolTests {
                 entries: [key: entry],
             )
 
-            pool.applySyncObjectsPool([objectState], logger: logger, userCallbackQueue: .main, clock: MockSimpleClock())
+            pool.applySyncObjectsPool([.init(state: objectState)], logger: logger, userCallbackQueue: .main, clock: MockSimpleClock())
 
             // Verify the existing map was updated by checking side effects of InternalDefaultLiveMap.replaceData(using:)
             let updatedMap = try #require(pool.entries["map:hash@123"]?.mapValue)
@@ -135,7 +135,7 @@ struct ObjectsPoolTests {
                 count: 10,
             )
 
-            pool.applySyncObjectsPool([objectState], logger: logger, userCallbackQueue: .main, clock: MockSimpleClock())
+            pool.applySyncObjectsPool([.init(state: objectState)], logger: logger, userCallbackQueue: .main, clock: MockSimpleClock())
 
             // Verify the existing counter was updated by checking side effects of InternalDefaultLiveCounter.replaceData(using:)
             let updatedCounter = try #require(pool.entries["counter:hash@123"]?.counterValue)
@@ -163,7 +163,7 @@ struct ObjectsPoolTests {
                 count: 100,
             )
 
-            pool.applySyncObjectsPool([objectState], logger: logger, userCallbackQueue: .main, clock: MockSimpleClock())
+            pool.applySyncObjectsPool([.init(state: objectState)], logger: logger, userCallbackQueue: .main, clock: MockSimpleClock())
 
             // Verify a new counter was created and data was set by checking side effects of InternalDefaultLiveCounter.replaceData(using:)
             let newCounter = try #require(pool.entries["counter:hash@456"]?.counterValue)
@@ -191,7 +191,7 @@ struct ObjectsPoolTests {
                 entries: [key: entry],
             )
 
-            pool.applySyncObjectsPool([objectState], logger: logger, userCallbackQueue: .main, clock: MockSimpleClock())
+            pool.applySyncObjectsPool([.init(state: objectState)], logger: logger, userCallbackQueue: .main, clock: MockSimpleClock())
 
             // Verify a new map was created and data was set by checking side effects of InternalDefaultLiveMap.replaceData(using:)
             let newMap = try #require(pool.entries["map:hash@789"]?.mapValue)
@@ -218,7 +218,7 @@ struct ObjectsPoolTests {
 
             let invalidObjectState = TestFactories.objectState(objectId: "invalid")
 
-            pool.applySyncObjectsPool([invalidObjectState, validObjectState], logger: logger, userCallbackQueue: .main, clock: MockSimpleClock())
+            pool.applySyncObjectsPool([invalidObjectState, validObjectState].map { .init(state: $0) }, logger: logger, userCallbackQueue: .main, clock: MockSimpleClock())
 
             // Check that there's no entry for the key that we don't know how to handle, and that it didn't interfere with the insertion of the we one that we do know how to handle
             #expect(Set(pool.entries.keys) == ["root", "counter:hash@456"])
@@ -243,7 +243,7 @@ struct ObjectsPoolTests {
             // Only sync one of the existing objects
             let objectState = TestFactories.mapObjectState(objectId: "map:hash@1")
 
-            pool.applySyncObjectsPool([objectState], logger: logger, userCallbackQueue: .main, clock: MockSimpleClock())
+            pool.applySyncObjectsPool([.init(state: objectState)], logger: logger, userCallbackQueue: .main, clock: MockSimpleClock())
 
             // Verify only synced object and root remain
             #expect(pool.entries.count == 2) // root + map:hash@1
@@ -324,7 +324,7 @@ struct ObjectsPoolTests {
                 // Note: "map:toremove@1" is not in sync, so it should be removed
             ]
 
-            pool.applySyncObjectsPool(syncObjects, logger: logger, userCallbackQueue: .main, clock: MockSimpleClock())
+            pool.applySyncObjectsPool(syncObjects.map { .init(state: $0) }, logger: logger, userCallbackQueue: .main, clock: MockSimpleClock())
 
             // Verify final state
             #expect(pool.entries.count == 5) // root + 4 synced objects