fixes from Claude

Author: lawrence-forooghian

Sources/AblyLiveObjects/Internal/InternalDefaultRealtimeObjects.swift

@@ -803,18 +803,14 @@ internal final class InternalDefaultRealtimeObjects: Sendable, InternalRealtimeO
                 }
                 var updatedSyncSequence = syncSequenceToContinue ?? .init(id: syncCursor.sequenceID, syncObjectsPool: .init())
                 // RTO5f
-                for objectMessage in objectMessages {
-                    updatedSyncSequence.syncObjectsPool.accumulate(objectMessage: objectMessage, logger: logger)
-                }
+                updatedSyncSequence.syncObjectsPool.accumulate(objectMessages, logger: logger)
                 syncSequenceForSyncingState = updatedSyncSequence
 
                 completedSyncObjectsPool = syncCursor.isEndOfSequence ? updatedSyncSequence.syncObjectsPool : nil
             } else {
                 // RTO5a5: The sync data is contained entirely within this single OBJECT_SYNC
                 var pool = SyncObjectsPool()
-                for objectMessage in objectMessages {
-                    pool.accumulate(objectMessage: objectMessage, logger: logger)
-                }
+                pool.accumulate(objectMessages, logger: logger)
                 completedSyncObjectsPool = pool
                 syncSequenceForSyncingState = nil
             }

Sources/AblyLiveObjects/Internal/ObjectsPool.swift

@@ -293,69 +293,71 @@ internal struct ObjectsPool {
         var updatesToExistingObjects: [ObjectsPool.Entry.DeferredUpdate] = []
 
         // RTO5c1: For each ObjectState member in the SyncObjectsPool list
-        for syncObjectsPoolEntry in syncObjectsPool {
-            receivedObjectIds.insert(syncObjectsPoolEntry.state.objectId)
+        for objectMessage in syncObjectsPool {
+            // Every message yielded by SyncObjectsPool is guaranteed to have a non-nil `.object` with `.map` or `.counter`.
+            let state = objectMessage.object!
+            receivedObjectIds.insert(state.objectId)
 
             // RTO5c1a: If an object with ObjectState.objectId exists in the internal ObjectsPool
-            if let existingEntry = entries[syncObjectsPoolEntry.state.objectId] {
-                logger.log("Updating existing object with ID: \(syncObjectsPoolEntry.state.objectId)", level: .debug)
+            if let existingEntry = entries[state.objectId] {
+                logger.log("Updating existing object with ID: \(state.objectId)", level: .debug)
 
                 // RTO5c1a1: Override the internal data for the object as per RTLC6, RTLM6
                 let deferredUpdate = existingEntry.nosync_replaceData(
-                    using: syncObjectsPoolEntry.state,
-                    objectMessageSerialTimestamp: syncObjectsPoolEntry.objectMessageSerialTimestamp,
+                    using: state,
+                    objectMessageSerialTimestamp: objectMessage.serialTimestamp,
                     objectsPool: &self,
                     userCallbackQueue: userCallbackQueue,
                 )
                 // 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: \(syncObjectsPoolEntry.state.objectId)", level: .debug)
+                logger.log("Creating new object with ID: \(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 syncObjectsPoolEntry.state.counter != nil {
+                if 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: syncObjectsPoolEntry.state.objectId,
+                        objectID: state.objectId,
                         logger: logger,
                         internalQueue: internalQueue,
                         userCallbackQueue: userCallbackQueue,
                         clock: clock,
                     )
                     _ = counter.nosync_replaceData(
-                        using: syncObjectsPoolEntry.state,
-                        objectMessageSerialTimestamp: syncObjectsPoolEntry.objectMessageSerialTimestamp,
+                        using: state,
+                        objectMessageSerialTimestamp: objectMessage.serialTimestamp,
                     )
                     newEntry = .counter(counter)
-                } else if let objectsMap = syncObjectsPoolEntry.state.map {
+                } else if let objectsMap = 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: syncObjectsPoolEntry.state.objectId,
+                        objectID: state.objectId,
                         semantics: objectsMap.semantics,
                         logger: logger,
                         internalQueue: internalQueue,
                         userCallbackQueue: userCallbackQueue,
                         clock: clock,
                     )
                     _ = map.nosync_replaceData(
-                        using: syncObjectsPoolEntry.state,
-                        objectMessageSerialTimestamp: syncObjectsPoolEntry.objectMessageSerialTimestamp,
+                        using: state,
+                        objectMessageSerialTimestamp: objectMessage.serialTimestamp,
                         objectsPool: &self,
                     )
                     newEntry = .map(map)
                 } else {
-                    // See SyncObjectsPool.Entry.state documentation.
-                    preconditionFailure("SyncObjectsPool entry for objectId \(syncObjectsPoolEntry.state.objectId) has neither counter nor map")
+                    // SyncObjectsPool guarantees every yielded message has `.map` or `.counter`.
+                    preconditionFailure("SyncObjectsPool entry for objectId \(state.objectId) has neither counter nor map")
                 }
 
                 // 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[syncObjectsPoolEntry.state.objectId] = newEntry
+                entries[state.objectId] = newEntry
             }
         }
 

Sources/AblyLiveObjects/Internal/SyncObjectsPool.swift

@@ -2,29 +2,30 @@ import Foundation
 
 /// The RTO5f collection of objects gathered during an `OBJECT_SYNC` sequence, ready to be applied to the `ObjectsPool`.
 ///
-/// Internally stores `InboundObjectMessage` values keyed by `objectId`. The `accumulate` method implements the RTO5f
-/// merge logic for partial object sync.
+/// Every stored message is guaranteed to have a non-nil `.object` with either `.map` or `.counter` populated.
 internal struct SyncObjectsPool: Sequence {
-    /// A computed view of a stored `InboundObjectMessage`, yielded during iteration.
-    ///
-    /// Preserves backward compatibility with the consumption side in `ObjectsPool.nosync_applySyncObjectsPool`.
-    internal struct Entry {
-        /// Guaranteed to have either `.map` or `.counter` populated.
-        internal var state: ObjectState
-        /// The `serialTimestamp` of the `ObjectMessage` that generated this entry.
-        internal var objectMessageSerialTimestamp: Date?
-    }
-
+    /// Keyed by `objectId`. Every value has a non-nil `.object` with either `.map` or `.counter` populated; the
+    /// `accumulate` method enforces this invariant.
     private var objectMessages: [String: InboundObjectMessage]
 
     /// Creates an empty pool.
     internal init() {
         objectMessages = [:]
     }
 
-    /// Accumulates an `ObjectMessage` into the pool per RTO5f.
+    /// Accumulates object messages into the pool per RTO5f.
     internal mutating func accumulate(
-        objectMessage: InboundObjectMessage,
+        _ objectMessages: [InboundObjectMessage],
+        logger: Logger,
+    ) {
+        for objectMessage in objectMessages {
+            accumulate(objectMessage, logger: logger)
+        }
+    }
+
+    /// Accumulates a single `ObjectMessage` into the pool per RTO5f.
+    private mutating func accumulate(
+        _ objectMessage: InboundObjectMessage,
         logger: Logger,
     ) {
         guard let object = objectMessage.object else {
@@ -33,7 +34,8 @@ internal struct SyncObjectsPool: Sequence {
 
         let objectId = object.objectId
 
-        // RTO5f3: Reject unsupported object types before pool lookup. This provides the guarantee documented on Entry.state.
+        // RTO5f3: Reject unsupported object types before pool lookup. Only messages whose `.object` has `.map` or `.counter`
+        // are stored, which callers of the iteration can rely on.
         guard object.map != nil || object.counter != nil else {
             logger.log("Skipping unsupported object type during sync for objectId \(objectId)", level: .warn)
             return
@@ -51,7 +53,13 @@ internal struct SyncObjectsPool: Sequence {
                     var merged = existing
                     if let incomingEntries = object.map?.entries {
                         var mergedObject = merged.object!
-                        var mergedMap = mergedObject.map!
+                        guard var mergedMap = mergedObject.map else {
+                            // Not a specified scenario — the server won't send a map and a non-map for the same
+                            // objectId in practice. Guard defensively rather than force-unwrapping.
+                            logger.log("Existing entry for objectId \(objectId) is not a map; replacing with incoming message", level: .error)
+                            objectMessages[objectId] = objectMessage
+                            return
+                        }
                         var mergedEntries = mergedMap.entries ?? [:]
                         mergedEntries.merge(incomingEntries) { _, new in new }
                         mergedMap.entries = mergedEntries
@@ -82,16 +90,8 @@ internal struct SyncObjectsPool: Sequence {
             self.underlying = underlying
         }
 
-        internal mutating func next() -> Entry? {
-            guard let message = underlying.next() else {
-                return nil
-            }
-
-            // We only store messages whose `object` is non-nil (see `accumulate`).
-            return Entry(
-                state: message.object!,
-                objectMessageSerialTimestamp: message.serialTimestamp,
-            )
+        internal mutating func next() -> InboundObjectMessage? {
+            underlying.next()
         }
     }
 

Tests/AblyLiveObjectsTests/ObjectsPoolTests.swift

@@ -2,21 +2,21 @@ import _AblyPluginSupportPrivate
 @testable import AblyLiveObjects
 import Testing
 
-extension SyncObjectsPool {
+private extension SyncObjectsPool {
     /// Test-only convenience to create a `SyncObjectsPool` from an array of `(state, serialTimestamp)` pairs,
     /// wrapping each in an `InboundObjectMessage` and calling `accumulate`.
     static func testsOnly_fromStates(
         _ states: [(state: ObjectState, serialTimestamp: Date?)],
         logger: AblyLiveObjects.Logger = TestLogger(),
     ) -> SyncObjectsPool {
         var pool = SyncObjectsPool()
-        for pair in states {
-            let message = TestFactories.inboundObjectMessage(
+        let messages = states.map { pair in
+            TestFactories.inboundObjectMessage(
                 object: pair.state,
                 serialTimestamp: pair.serialTimestamp,
             )
-            pool.accumulate(objectMessage: message, logger: logger)
         }
+        pool.accumulate(messages, logger: logger)
         return pool
     }
 }