Implement the OBJECT_SYNC specification

Based on [1] at dd25dca.

The development approach here was roughly the following: I decided the
internal types and interfaces that I wanted to exist, and some of the
implementation details (e.g. the mutations that DefaultRealtimeObjects
performs during a sync sequence). I then asked Cursor to fill in the
most of the implementation and tests (providing it with the relevant
spec text). I have done some tweaking of the code that Cursor generated,
but that doesn't mean that all of the code that's here is exactly the
same as code that I would write, and I'm going to leave it that way.

I have looked through all (and changed some) of the tests that Cursor
generated, and in doing so also fixed the spec conformance tags (because
it got those very wrong), so I am _fairly_ confident that the tests are
testing what they claim to test. But I have not looked at them in minute
detail (in particular, the exact data created by the canned data
factories that it created). I also think that tests are one of those
things where there's not always one right way to write them, so I'm not
going to get hung up on "could it have done this test differently?" or
"are the tests consistent with each other?" at the moment.

There are some outstanding questions on the spec PR at the moment; have
implemented to best of my understanding, and will update later once
those are answered.

Not yet implemented:

- Checking of channel modes before performing operations (RTO2 and the
  points that refer to it); there's an outstanding question about this
  at [2]

[1] ably/specification#333
[2] https://github.com/ably/specification/pull/333/files#r2152297442

Author: lawrence-forooghian

Sources/AblyLiveObjects/DefaultRealtimeObjects.swift

@@ -2,7 +2,7 @@ import Ably
 internal import AblyPlugin
 
 /// The class that provides the public API for interacting with LiveObjects, via the ``ARTRealtimeChannel/objects`` property.
-internal final class DefaultRealtimeObjects: RealtimeObjects {
+internal final class DefaultRealtimeObjects: RealtimeObjects, LiveMapObjectPoolDelegate {
     // Used for synchronizing access to all of this instance's mutable state. This is a temporary solution just to allow us to implement `Sendable`, and we'll revisit it in https://github.com/ably/ably-cocoa-liveobjects-plugin/issues/3.
     private let mutex = NSLock()
 
@@ -15,17 +15,91 @@ internal final class DefaultRealtimeObjects: RealtimeObjects {
     private let receivedObjectSyncProtocolMessages: AsyncStream<[InboundObjectMessage]>
     private let receivedObjectSyncProtocolMessagesContinuation: AsyncStream<[InboundObjectMessage]>.Continuation
 
+    private nonisolated(unsafe) var objectsPool: ObjectsPool!
+    internal var testsOnly_objectsPool: ObjectsPool {
+        mutex.withLock {
+            objectsPool
+        }
+    }
+
+    private nonisolated(unsafe) var syncSequence: SyncSequence?
+    /// If this returns false, it means that there is currently no stored sync sequence ID or SyncObjectsPool
+    internal var testsOnly_hasSyncSequence: Bool {
+        mutex.withLock {
+            syncSequence != nil
+        }
+    }
+
+    private nonisolated(unsafe) var syncStatus = SyncStatus()
+    // These drive the testsOnly_waitingForSyncEvents property that informs the test suite when `getRoot()` is waiting for the object sync sequence to complete per RTO1c.
+    private let waitingForSyncEvents: AsyncStream<Void>
+    private let waitingForSyncEventsContinuation: AsyncStream<Void>.Continuation
+    /// Emits an element whenever `getRoot()` starts waiting for the object sync sequence to complete per RTO1c.
+    internal var testsOnly_waitingForSyncEvents: AsyncStream<Void> {
+        waitingForSyncEvents
+    }
+
+    /// Tracks whether an object sync sequence has happened yet. This allows us to wait for a sync before returning from `getRoot()`, per RTO1c.
+    private struct SyncStatus {
+        private(set) var isSyncComplete = false
+        private let syncCompletionEvents: AsyncStream<Void>
+        private let syncCompletionContinuation: AsyncStream<Void>.Continuation
+
+        internal init() {
+            (syncCompletionEvents, syncCompletionContinuation) = AsyncStream.makeStream()
+        }
+
+        internal mutating func signalSyncComplete() {
+            isSyncComplete = true
+            syncCompletionContinuation.yield()
+        }
+
+        internal func waitForSyncCompletion() async {
+            await syncCompletionEvents.first { _ in true }
+        }
+    }
+
     internal init(coreSDK: CoreSDK, logger: AblyPlugin.Logger) {
         self.coreSDK = coreSDK
         self.logger = logger
         (receivedObjectProtocolMessages, receivedObjectProtocolMessagesContinuation) = AsyncStream.makeStream()
         (receivedObjectSyncProtocolMessages, receivedObjectSyncProtocolMessagesContinuation) = AsyncStream.makeStream()
+        (waitingForSyncEvents, waitingForSyncEventsContinuation) = AsyncStream.makeStream()
+        objectsPool = .init(rootDelegate: self, rootCoreSDK: coreSDK)
+    }
+
+    // MARK: - LiveMapObjectPoolDelegate
+
+    internal func getObjectFromPool(id: String) -> ObjectsPool.Entry? {
+        mutex.withLock {
+            objectsPool.entries[id]
+        }
     }
 
     // MARK: `RealtimeObjects` protocol
 
     internal func getRoot() async throws(ARTErrorInfo) -> any LiveMap {
-        notYetImplemented()
+        // RTO1b: If the channel is in the DETACHED or FAILED state, the library should indicate an error with code 90001
+        let currentChannelState = coreSDK.channelState
+        if currentChannelState == .detached || currentChannelState == .failed {
+            throw ARTErrorInfo.create(withCode: Int(ARTErrorCode.channelOperationFailedInvalidState.rawValue), message: "getRoot operation failed (invalid channel state: \(currentChannelState))")
+        }
+
+        let isSyncComplete = mutex.withLock {
+            syncStatus.isSyncComplete
+        }
+        if !isSyncComplete {
+            // RTO1c
+            waitingForSyncEventsContinuation.yield()
+            logger.log("getRoot started waiting for sync sequence to complete", level: .debug)
+            await syncStatus.waitForSyncCompletion()
+            logger.log("getRoot completed waiting for sync sequence to complete", level: .debug)
+        }
+
+        return mutex.withLock {
+            // RTO1d
+            objectsPool.root
+        }
     }
 
     internal func createMap(entries _: [String: LiveMapValue]) async throws(ARTErrorInfo) -> any LiveMap {
@@ -67,7 +141,24 @@ internal final class DefaultRealtimeObjects: RealtimeObjects {
 
     internal func onChannelAttached(hasObjects: Bool) {
         mutex.withLock {
+            logger.log("onChannelAttached(hasObjects: \(hasObjects)", level: .debug)
+
             onChannelAttachedHasObjects = hasObjects
+
+            // We only care about the case where HAS_OBJECTS is not set (RTO4b); if it is set then we're going to shortly receive an OBJECT_SYNC instead (RTO4a)
+            guard !hasObjects else {
+                return
+            }
+
+            // RTO4b1, RTO4b2: Reset the ObjectsPool to have a single empty root object
+            // TODO: this one is unclear (are we meant to replace the root or just clear its data?) https://github.com/ably/specification/pull/333/files#r2183493458
+            objectsPool = .init(rootDelegate: self, rootCoreSDK: coreSDK)
+
+            // I have, for now, not directly implemented the "perform the actions for object sync completion" of RTO4b4 since my implementation doesn't quite match the model given there; here you only have a SyncObjectsPool if you have an OBJECT_SYNC in progress, which you might not have upon receiving an ATTACHED. Instead I've just implemented what seem like the relevant side effects. Can revisit this if "the actions for object sync completion" get more complex.
+
+            // RTO4b3, RTO4b4, RTO5c3, RTO5c4
+            syncSequence = nil
+            syncStatus.signalSyncComplete()
         }
     }
 
@@ -83,8 +174,87 @@ internal final class DefaultRealtimeObjects: RealtimeObjects {
         receivedObjectSyncProtocolMessages
     }
 
-    internal func handleObjectSyncProtocolMessage(objectMessages: [InboundObjectMessage], protocolMessageChannelSerial _: String?) {
-        receivedObjectSyncProtocolMessagesContinuation.yield(objectMessages)
+    /// Contains the data gathered during an `OBJECT_SYNC` sequence.
+    private struct SyncSequence {
+        /// The sync sequence ID, per RTO5a1.
+        internal var id: String
+
+        /// The `ObjectMessage`s gathered during this sync sequence.
+        internal var syncObjectsPool: [ObjectState]
+    }
+
+    /// Implements the `OBJECT_SYNC` handling of RTO5.
+    internal func handleObjectSyncProtocolMessage(objectMessages: [InboundObjectMessage], protocolMessageChannelSerial: String?) {
+        mutex.withLock {
+            logger.log("handleObjectSyncProtocolMessage(objectMessages: \(objectMessages), protocolMessageChannelSerial: \(String(describing: protocolMessageChannelSerial)))", level: .debug)
+
+            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]?
+
+            if let protocolMessageChannelSerial {
+                let syncCursor: SyncCursor
+                do {
+                    // RTO5a
+                    syncCursor = try SyncCursor(channelSerial: protocolMessageChannelSerial)
+                } catch {
+                    logger.log("Failed to parse sync cursor: \(error)", level: .error)
+                    return
+                }
+
+                // Figure out whether to continue any existing sync sequence or start a new one
+                var updatedSyncSequence: SyncSequence = if let syncSequence {
+                    if syncCursor.sequenceID == syncSequence.id {
+                        // RTO5a3: Continue existing sync sequence
+                        syncSequence
+                    } else {
+                        // RTO5a2: new sequence started, discard previous
+                        .init(id: syncCursor.sequenceID, syncObjectsPool: [])
+                    }
+                } else {
+                    // There's no current sync sequence; start one
+                    .init(id: syncCursor.sequenceID, syncObjectsPool: [])
+                }
+
+                // RTO5b
+                updatedSyncSequence.syncObjectsPool.append(contentsOf: objectMessages.compactMap(\.object))
+
+                syncSequence = updatedSyncSequence
+
+                completedSyncObjectsPool = if syncCursor.isEndOfSequence {
+                    updatedSyncSequence.syncObjectsPool
+                } else {
+                    nil
+                }
+            } else {
+                // RTO5a5: The sync data is contained entirely within this single OBJECT_SYNC
+                completedSyncObjectsPool = objectMessages.compactMap(\.object)
+            }
+
+            if let completedSyncObjectsPool {
+                // RTO5c
+                objectsPool.applySyncObjectsPool(
+                    completedSyncObjectsPool,
+                    mapDelegate: self,
+                    coreSDK: coreSDK,
+                    logger: logger,
+                )
+                // RTO5c3, RTO5c4
+                syncSequence = nil
+
+                syncStatus.signalSyncComplete()
+            }
+        }
+    }
+
+    /// Creates a zero-value LiveObject in the object pool for this object ID.
+    ///
+    /// Intended as a way for tests to populate the object pool.
+    internal func testsOnly_createZeroValueLiveObject(forObjectID objectID: String, coreSDK: CoreSDK) -> ObjectsPool.Entry? {
+        mutex.withLock {
+            objectsPool.createZeroValueObject(forObjectID: objectID, mapDelegate: self, coreSDK: coreSDK)
+        }
     }
 
     // MARK: - Sending `OBJECT` ProtocolMessage
@@ -93,4 +263,17 @@ internal final class DefaultRealtimeObjects: RealtimeObjects {
     internal func testsOnly_sendObject(objectMessages: [OutboundObjectMessage]) async throws(InternalError) {
         try await coreSDK.sendObject(objectMessages: objectMessages)
     }
+
+    // MARK: - Testing
+
+    /// Finishes the following streams, to allow a test to perform assertions about which elements the streams have emitted to this moment:
+    ///
+    /// - testsOnly_receivedObjectProtocolMessages
+    /// - testsOnly_receivedObjectStateProtocolMessages
+    /// - testsOnly_waitingForSyncEvents
+    internal func testsOnly_finishAllTestHelperStreams() {
+        receivedObjectProtocolMessagesContinuation.finish()
+        receivedObjectSyncProtocolMessagesContinuation.finish()
+        waitingForSyncEventsContinuation.finish()
+    }
 }

Sources/AblyLiveObjects/Internal/CoreSDK.swift

@@ -6,6 +6,9 @@ internal import AblyPlugin
 /// This provides us with a mockable interface to ably-cocoa, and it also allows internal components and their tests not to need to worry about some of the boring details of how we bridge Swift types to AblyPlugin's Objective-C API (i.e. boxing).
 internal protocol CoreSDK: AnyObject, Sendable {
     func sendObject(objectMessages: [OutboundObjectMessage]) async throws(InternalError)
+
+    /// Returns the current state of the Realtime channel that this wraps.
+    var channelState: ARTRealtimeChannelState { get }
 }
 
 internal final class DefaultCoreSDK: CoreSDK {
@@ -41,4 +44,8 @@ internal final class DefaultCoreSDK: CoreSDK {
             pluginAPI: pluginAPI,
         )
     }
+
+    internal var channelState: ARTRealtimeChannelState {
+        channel.state
+    }
 }

Sources/AblyLiveObjects/Internal/DefaultLiveCounter.swift

@@ -3,12 +3,80 @@ import Foundation
 
 /// Our default implementation of ``LiveCounter``.
 internal final class DefaultLiveCounter: LiveCounter {
-    internal init() {}
+    // Used for synchronizing access to all of this instance's mutable state. This is a temporary solution just to allow us to implement `Sendable`, and we'll revisit it in https://github.com/ably/ably-cocoa-liveobjects-plugin/issues/3.
+    private let mutex = NSLock()
+
+    private nonisolated(unsafe) var mutableState: MutableState
+
+    internal var testsOnly_siteTimeserials: [String: String]? {
+        mutex.withLock {
+            mutableState.siteTimeserials
+        }
+    }
+
+    internal var testsOnly_createOperationIsMerged: Bool? {
+        mutex.withLock {
+            mutableState.createOperationIsMerged
+        }
+    }
+
+    internal var testsOnly_objectID: String? {
+        mutex.withLock {
+            mutableState.objectID
+        }
+    }
+
+    private let coreSDK: CoreSDK
+
+    // MARK: - Initialization
+
+    internal convenience init(
+        testsOnly_data data: Double,
+        objectID: String?,
+        coreSDK: CoreSDK
+    ) {
+        self.init(data: data, objectID: objectID, coreSDK: coreSDK)
+    }
+
+    private init(
+        data: Double,
+        objectID: String?,
+        coreSDK: CoreSDK
+    ) {
+        mutableState = .init(data: data, objectID: objectID)
+        self.coreSDK = coreSDK
+    }
+
+    /// Creates a "zero-value LiveCounter", per RTLC4.
+    ///
+    /// - Parameters:
+    ///   - objectID: The value for the "private objectId field" of RTO5c1b1a.
+    internal static func createZeroValued(
+        objectID: String? = nil,
+        coreSDK: CoreSDK,
+    ) -> Self {
+        .init(
+            data: 0,
+            objectID: objectID,
+            coreSDK: coreSDK,
+        )
+    }
 
     // MARK: - LiveCounter conformance
 
     internal var value: Double {
-        notYetImplemented()
+        get throws(ARTErrorInfo) {
+            // RTLC5b: If the channel is in the DETACHED or FAILED state, the library should indicate an error with code 90001
+            let currentChannelState = coreSDK.channelState
+            if currentChannelState == .detached || currentChannelState == .failed {
+                throw ARTErrorInfo.create(withCode: Int(ARTErrorCode.channelOperationFailedInvalidState.rawValue), message: "LiveCounter.value operation failed (invalid channel state: \(currentChannelState))")
+            }
+
+            return mutex.withLock {
+                // RTLC5c
+                mutableState.data
+            }
+        }
     }
 
     internal func increment(amount _: Double) async throws(ARTErrorInfo) {
@@ -34,4 +102,51 @@ internal final class DefaultLiveCounter: LiveCounter {
     internal func offAll() {
         notYetImplemented()
     }
+
+    // MARK: - Data manipulation
+
+    /// Replaces the internal data of this counter with the provided ObjectState, per RTLC6.
+    internal func replaceData(using state: ObjectState) {
+        mutex.withLock {
+            mutableState.replaceData(using: state)
+        }
+    }
+
+    // MARK: - Mutable state and the operations that affect it
+
+    private struct MutableState {
+        /// The internal data that this map holds, per RTLC3.
+        internal var data: Double
+
+        /// The site timeserials for this counter, per RTLC6a.
+        internal var siteTimeserials: [String: String]?
+
+        /// Whether the create operation has been merged, per RTLC6b and RTLC6d2.
+        internal var createOperationIsMerged: Bool?
+
+        /// The "private `objectId` field" of RTO5c1b1a.
+        internal var objectID: String?
+
+        /// Replaces the internal data of this counter with the provided ObjectState, per RTLC6.
+        internal mutating func replaceData(using state: ObjectState) {
+            // RTLC6a: Replace the private siteTimeserials with the value from ObjectState.siteTimeserials
+            siteTimeserials = state.siteTimeserials
+
+            // RTLC6b: Set the private flag createOperationIsMerged to false
+            createOperationIsMerged = false
+
+            // RTLC6c: Set data to the value of ObjectState.counter.count, or to 0 if it does not exist
+            data = state.counter?.count?.doubleValue ?? 0
+
+            // RTLC6d: If ObjectState.createOp is present
+            if let createOp = state.createOp {
+                // RTLC6d1: Add ObjectState.createOp.counter.count to data, if it exists
+                if let createOpCount = createOp.counter?.count?.doubleValue {
+                    data += createOpCount
+                }
+                // RTLC6d2: Set the private flag createOperationIsMerged to true
+                createOperationIsMerged = true
+            }
+        }
+    }
 }