Implement new rules for discarding buffered object operations

lawrence-forooghian · lawrence-forooghian · commit 11de235461e9 · 2026-01-21T11:44:39.000-03:00
That is, do it when we get a discontinuity, not when a new sync sequence changes, per spec changes in [1]. [1] ably/specification#416
diff --git a/AblyLiveObjects.xcworkspace/xcshareddata/swiftpm/Package.resolved b/AblyLiveObjects.xcworkspace/xcshareddata/swiftpm/Package.resolved
diff --git a/Package.resolved b/Package.resolved
diff --git a/Package.swift b/Package.swift
@@ -18,13 +18,15 @@ let package = Package(
         ),
     ],
     dependencies: [
+        // TODO: Unpin before release
         .package(
             url: "https://github.com/ably/ably-cocoa",
-            from: "1.2.55",
+            revision: "ee0b1fabf2e30377d6a1d90f3d2b6614d16951d8",
         ),
+        // TODO: Unpin before release
         .package(
             url: "https://github.com/ably/ably-cocoa-plugin-support",
-            from: "1.0.0",
+            revision: "2e640ce2ef422cd5ef693a9e1111400e9985e088",
         ),
         .package(
             url: "https://github.com/apple/swift-argument-parser",
diff --git a/Sources/AblyLiveObjects/Internal/DefaultInternalPlugin.swift b/Sources/AblyLiveObjects/Internal/DefaultInternalPlugin.swift
@@ -114,8 +114,12 @@ internal final class DefaultInternalPlugin: NSObject, _AblyPluginSupportPrivate.
         return wireObjectMessage.toWireObject.toPluginSupportDataDictionary
     }
 
-    internal func nosync_onChannelAttached(_ channel: _AblyPluginSupportPrivate.RealtimeChannel, hasObjects: Bool) {
-        nosync_realtimeObjects(for: channel).nosync_onChannelAttached(hasObjects: hasObjects)
+    internal func nosync_onChannelAttached(_: _AblyPluginSupportPrivate.RealtimeChannel, hasObjects _: Bool) {
+        preconditionFailure("Expected a version of ably-cocoa that calls the variant that accepts a `resumed` argument")
+    }
+
+    internal func nosync_onChannelAttached(_ channel: _AblyPluginSupportPrivate.RealtimeChannel, hasObjects: Bool, resumed: Bool) {
+        nosync_realtimeObjects(for: channel).nosync_onChannelAttached(hasObjects: hasObjects, resumed: resumed)
     }
 
     internal func nosync_handleObjectProtocolMessage(withObjectMessages publicObjectMessages: [any _AblyPluginSupportPrivate.ObjectMessageProtocol], channel: _AblyPluginSupportPrivate.RealtimeChannel) {
diff --git a/Sources/AblyLiveObjects/Internal/InternalDefaultRealtimeObjects.swift b/Sources/AblyLiveObjects/Internal/InternalDefaultRealtimeObjects.swift
@@ -302,10 +302,11 @@ internal final class InternalDefaultRealtimeObjects: Sendable, LiveMapObjectsPoo
         }
     }
 
-    internal func nosync_onChannelAttached(hasObjects: Bool) {
+    internal func nosync_onChannelAttached(hasObjects: Bool, resumed: Bool) {
         mutableStateMutex.withoutSync { mutableState in
             mutableState.nosync_onChannelAttached(
                 hasObjects: hasObjects,
+                resumed: resumed,
                 logger: logger,
                 userCallbackQueue: userCallbackQueue,
             )
@@ -500,13 +501,20 @@ internal final class InternalDefaultRealtimeObjects: Sendable, LiveMapObjectsPoo
 
         internal mutating func nosync_onChannelAttached(
             hasObjects: Bool,
+            resumed: Bool,
             logger: Logger,
             userCallbackQueue: DispatchQueue,
         ) {
-            logger.log("onChannelAttached(hasObjects: \(hasObjects)", level: .debug)
+            logger.log("onChannelAttached(hasObjects: \(hasObjects), resumed: \(resumed)", level: .debug)
 
             onChannelAttachedHasObjects = hasObjects
 
+            // RTO4d: If the RESUMED flag is not set, clear any buffered operations
+            if !resumed, case let .syncing(syncingData) = state {
+                logger.log("Clearing buffered operations due to ATTACHED without RESUMED flag", level: .debug)
+                syncingData.bufferedObjectOperations = []
+            }
+
             // We will subsequently transition to .synced either by the completion of the RTO4a OBJECT_SYNC, or by the RTO4b no-HAS_OBJECTS case below
             if state.toObjectsSyncState != .syncing {
                 // RTO4c
@@ -558,9 +566,8 @@ internal final class InternalDefaultRealtimeObjects: Sendable, LiveMapObjectsPoo
                 // Figure out whether to continue any existing sync sequence or start a new one
                 let isNewSyncSequence = syncCursor == nil || syncingData.syncSequence?.id != syncCursor?.sequenceID
                 if isNewSyncSequence {
-                    // RTO5a2a, RTO5a2b: new sequence started, discard previous. Else we continue the existing sequence per RTO5a3
+                    // RTO5a2a: new sequence started, discard previous SyncObjectsPool. Else we continue the existing sequence per RTO5a3
                     syncingData.syncSequence = nil
-                    syncingData.bufferedObjectOperations = []
                 }
             }
 
diff --git a/Tests/AblyLiveObjectsTests/InternalDefaultRealtimeObjectsTests.swift b/Tests/AblyLiveObjectsTests/InternalDefaultRealtimeObjectsTests.swift
@@ -126,9 +126,8 @@ struct InternalDefaultRealtimeObjectsTests {
 
         // @spec RTO5a2
         // @spec RTO5a2a
-        // @spec RTO5a2b
         @Test
-        func newSequenceIdDiscardsInFlightSync() async throws {
+        func newSequenceIdDiscardsInFlightSyncButKeepsBufferedOperations() async throws {
             let internalQueue = TestFactories.createInternalQueue()
             let realtimeObjects = InternalDefaultRealtimeObjectsTests.createDefaultRealtimeObjects(internalQueue: internalQueue)
             let firstSequenceId = "seq1"
@@ -145,7 +144,8 @@ struct InternalDefaultRealtimeObjectsTests {
 
             #expect(realtimeObjects.testsOnly_hasSyncSequence)
 
-            // Inject an OBJECT; it will get buffered per RTO8a and subsequently discarded per RTO5a2b
+            // Inject an OBJECT; it will get buffered per RTO8a and applied after sync completion per RTO5c6
+            // (Note: RTO5a2b was deleted, so buffered operations are NOT cleared when a new sequence ID arrives)
             internalQueue.ably_syncNoDeadlock {
                 realtimeObjects.nosync_handleObjectProtocolMessage(objectMessages: [
                     TestFactories.mapCreateOperationMessage(objectId: "map:3@789"),
@@ -172,10 +172,11 @@ struct InternalDefaultRealtimeObjectsTests {
                 )
             }
 
-            // Verify only the second sequence's objects were applied (RTO5a2a - previous cleared)
+            // Verify the second sequence's objects were applied (RTO5a2a - SyncObjectsPool was cleared)
+            // but the buffered OBJECT operation was also applied (RTO5a2b deleted - BufferedObjectOperations NOT cleared)
             let pool = realtimeObjects.testsOnly_objectsPool
-            #expect(pool.entries["map:1@123"] == nil) // From discarded first sequence
-            #expect(pool.entries["map:3@789"] == nil) // Check we discarded the OBJECT that was buffered during discarded first sequence (RTO5a2b)
+            #expect(pool.entries["map:1@123"] == nil) // From discarded first sequence's SyncObjectsPool
+            #expect(pool.entries["map:3@789"] != nil) // Buffered OBJECT was applied (RTO5a2b deleted)
             #expect(pool.entries["map:2@456"] != nil) // From completed second sequence
             #expect(!realtimeObjects.testsOnly_hasSyncSequence)
         }
@@ -373,7 +374,7 @@ struct InternalDefaultRealtimeObjectsTests {
 
             // When: onChannelAttached is called with hasObjects = true
             internalQueue.ably_syncNoDeadlock {
-                realtimeObjects.nosync_onChannelAttached(hasObjects: true)
+                realtimeObjects.nosync_onChannelAttached(hasObjects: true, resumed: false)
             }
 
             // Then: Nothing should be modified
@@ -389,6 +390,96 @@ struct InternalDefaultRealtimeObjectsTests {
             #expect(realtimeObjects.testsOnly_hasSyncSequence)
         }
 
+        // MARK: - RTO4d Tests
+
+        // @spec RTO4d - Checks that buffered operations are cleared when ATTACHED is received without RESUMED flag
+        @Test
+        func clearsBufferedOperationsWhenResumedIsFalse() {
+            let internalQueue = TestFactories.createInternalQueue()
+            let realtimeObjects = InternalDefaultRealtimeObjectsTests.createDefaultRealtimeObjects(internalQueue: internalQueue)
+
+            // Start a sync sequence
+            internalQueue.ably_syncNoDeadlock {
+                realtimeObjects.nosync_handleObjectSyncProtocolMessage(
+                    objectMessages: [
+                        TestFactories.mapObjectMessage(objectId: "map:sync@123"),
+                    ],
+                    protocolMessageChannelSerial: "seq1:cursor1",
+                )
+            }
+
+            #expect(realtimeObjects.testsOnly_hasSyncSequence)
+
+            // Buffer some OBJECT operations during the sync
+            internalQueue.ably_syncNoDeadlock {
+                realtimeObjects.nosync_handleObjectProtocolMessage(objectMessages: [
+                    TestFactories.mapCreateOperationMessage(objectId: "map:buffered@456"),
+                ])
+            }
+
+            // When: onChannelAttached is called with resumed = false (RTO4d)
+            internalQueue.ably_syncNoDeadlock {
+                realtimeObjects.nosync_onChannelAttached(hasObjects: true, resumed: false)
+            }
+
+            // Continue and complete the sync sequence
+            internalQueue.ably_syncNoDeadlock {
+                realtimeObjects.nosync_handleObjectSyncProtocolMessage(
+                    objectMessages: [],
+                    protocolMessageChannelSerial: "seq1:",
+                )
+            }
+
+            // Then: The buffered OBJECT operation should NOT have been applied because it was cleared per RTO4d
+            let pool = realtimeObjects.testsOnly_objectsPool
+            #expect(pool.entries["map:sync@123"] != nil) // From the sync sequence
+            #expect(pool.entries["map:buffered@456"] == nil) // Buffered operation was cleared (RTO4d)
+        }
+
+        // @spec RTO4d - Checks that buffered operations are NOT cleared when ATTACHED is received WITH RESUMED flag
+        @Test
+        func preservesBufferedOperationsWhenResumedIsTrue() {
+            let internalQueue = TestFactories.createInternalQueue()
+            let realtimeObjects = InternalDefaultRealtimeObjectsTests.createDefaultRealtimeObjects(internalQueue: internalQueue)
+
+            // Start a sync sequence
+            internalQueue.ably_syncNoDeadlock {
+                realtimeObjects.nosync_handleObjectSyncProtocolMessage(
+                    objectMessages: [
+                        TestFactories.mapObjectMessage(objectId: "map:sync@123"),
+                    ],
+                    protocolMessageChannelSerial: "seq1:cursor1",
+                )
+            }
+
+            #expect(realtimeObjects.testsOnly_hasSyncSequence)
+
+            // Buffer some OBJECT operations during the sync
+            internalQueue.ably_syncNoDeadlock {
+                realtimeObjects.nosync_handleObjectProtocolMessage(objectMessages: [
+                    TestFactories.mapCreateOperationMessage(objectId: "map:buffered@456"),
+                ])
+            }
+
+            // When: onChannelAttached is called with resumed = true (state was preserved)
+            internalQueue.ably_syncNoDeadlock {
+                realtimeObjects.nosync_onChannelAttached(hasObjects: true, resumed: true)
+            }
+
+            // Continue and complete the sync sequence
+            internalQueue.ably_syncNoDeadlock {
+                realtimeObjects.nosync_handleObjectSyncProtocolMessage(
+                    objectMessages: [],
+                    protocolMessageChannelSerial: "seq1:",
+                )
+            }
+
+            // Then: The buffered OBJECT operation should have been applied because it was NOT cleared
+            let pool = realtimeObjects.testsOnly_objectsPool
+            #expect(pool.entries["map:sync@123"] != nil) // From the sync sequence
+            #expect(pool.entries["map:buffered@456"] != nil) // Buffered operation was preserved and applied
+        }
+
         // MARK: - RTO4b Tests
 
         // @spec RTO4b1
@@ -440,7 +531,7 @@ struct InternalDefaultRealtimeObjectsTests {
 
             // When: onChannelAttached is called with hasObjects = false
             internalQueue.ably_syncNoDeadlock {
-                realtimeObjects.nosync_onChannelAttached(hasObjects: false)
+                realtimeObjects.nosync_onChannelAttached(hasObjects: false, resumed: false)
             }
 
             // Then: Verify the expected behavior per RTO4b
@@ -476,15 +567,15 @@ struct InternalDefaultRealtimeObjectsTests {
 
             // First call with hasObjects = true (should do nothing)
             internalQueue.ably_syncNoDeadlock {
-                realtimeObjects.nosync_onChannelAttached(hasObjects: true)
+                realtimeObjects.nosync_onChannelAttached(hasObjects: true, resumed: false)
             }
             #expect(realtimeObjects.testsOnly_onChannelAttachedHasObjects == true)
             let originalPool = realtimeObjects.testsOnly_objectsPool
             let originalRoot = originalPool.root
 
             // Second call with hasObjects = false (should reset)
             internalQueue.ably_syncNoDeadlock {
-                realtimeObjects.nosync_onChannelAttached(hasObjects: false)
+                realtimeObjects.nosync_onChannelAttached(hasObjects: false, resumed: false)
             }
             #expect(realtimeObjects.testsOnly_onChannelAttachedHasObjects == false)
             let newPool = realtimeObjects.testsOnly_objectsPool
@@ -493,7 +584,7 @@ struct InternalDefaultRealtimeObjectsTests {
 
             // Third call with hasObjects = true again (should do nothing)
             internalQueue.ably_syncNoDeadlock {
-                realtimeObjects.nosync_onChannelAttached(hasObjects: true)
+                realtimeObjects.nosync_onChannelAttached(hasObjects: true, resumed: false)
             }
             #expect(realtimeObjects.testsOnly_onChannelAttachedHasObjects == true)
             let finalPool = realtimeObjects.testsOnly_objectsPool
@@ -530,7 +621,7 @@ struct InternalDefaultRealtimeObjectsTests {
 
             // When: onChannelAttached is called with hasObjects = false
             internalQueue.ably_syncNoDeadlock {
-                realtimeObjects.nosync_onChannelAttached(hasObjects: false)
+                realtimeObjects.nosync_onChannelAttached(hasObjects: false, resumed: false)
             }
 
             // Then: All sync data should be discarded
@@ -565,7 +656,7 @@ struct InternalDefaultRealtimeObjectsTests {
 
             // When: onChannelAttached is called with hasObjects = false
             internalQueue.ably_syncNoDeadlock {
-                realtimeObjects.nosync_onChannelAttached(hasObjects: false)
+                realtimeObjects.nosync_onChannelAttached(hasObjects: false, resumed: false)
             }
 
             // Then: Should still reset the pool correctly
@@ -583,7 +674,7 @@ struct InternalDefaultRealtimeObjectsTests {
 
             // When: onChannelAttached is called with hasObjects = false
             internalQueue.ably_syncNoDeadlock {
-                realtimeObjects.nosync_onChannelAttached(hasObjects: false)
+                realtimeObjects.nosync_onChannelAttached(hasObjects: false, resumed: false)
             }
 
             // Then: The new root should be properly initialized
@@ -611,7 +702,7 @@ struct InternalDefaultRealtimeObjectsTests {
 
             // Complete sync via ATTACHED with HAS_OBJECTS false (RTO4b)
             internalQueue.ably_syncNoDeadlock {
-                realtimeObjects.nosync_onChannelAttached(hasObjects: false)
+                realtimeObjects.nosync_onChannelAttached(hasObjects: false, resumed: false)
             }
 
             // getRoot should now complete
@@ -736,7 +827,7 @@ struct InternalDefaultRealtimeObjectsTests {
 
             // Complete sync first
             internalQueue.ably_syncNoDeadlock {
-                realtimeObjects.nosync_onChannelAttached(hasObjects: false)
+                realtimeObjects.nosync_onChannelAttached(hasObjects: false, resumed: false)
             }
 
             // getRoot should return
@@ -761,7 +852,7 @@ struct InternalDefaultRealtimeObjectsTests {
 
             // Complete sync first
             internalQueue.ably_syncNoDeadlock {
-                realtimeObjects.nosync_onChannelAttached(hasObjects: false)
+                realtimeObjects.nosync_onChannelAttached(hasObjects: false, resumed: false)
             }
 
             // Call getRoot
@@ -1602,7 +1693,8 @@ struct InternalDefaultRealtimeObjectsTests {
                 for channelEvent in scenario.channelEvents {
                     switch channelEvent {
                     case let .attached(hasObjects):
-                        realtimeObjects.nosync_onChannelAttached(hasObjects: hasObjects)
+                        // The value of `resumed` is arbitrary for these tests; they're focused on sync state transitions
+                        realtimeObjects.nosync_onChannelAttached(hasObjects: hasObjects, resumed: true)
                     case let .objectSync(channelSerial):
                         realtimeObjects.nosync_handleObjectSyncProtocolMessage(
                             objectMessages: [],

Original file line number	Diff line number	Diff line change
`@@ -114,8 +114,12 @@ internal final class DefaultInternalPlugin: NSObject, _AblyPluginSupportPrivate.`
`114`	`114`	`return wireObjectMessage.toWireObject.toPluginSupportDataDictionary`
`115`	`115`	`}`
`116`	`116`
`117`		`- internal func nosync_onChannelAttached(_ channel: _AblyPluginSupportPrivate.RealtimeChannel, hasObjects: Bool) {`
`118`		`- nosync_realtimeObjects(for: channel).nosync_onChannelAttached(hasObjects: hasObjects)`
	`117`	`+ internal func nosync_onChannelAttached(_: _AblyPluginSupportPrivate.RealtimeChannel, hasObjects _: Bool) {`
	`118`	+ preconditionFailure("Expected a version of ably-cocoa that calls the variant that accepts a `resumed` argument")
	`119`	`+ }`
	`120`	`+`
	`121`	`+ internal func nosync_onChannelAttached(_ channel: _AblyPluginSupportPrivate.RealtimeChannel, hasObjects: Bool, resumed: Bool) {`
	`122`	`+ nosync_realtimeObjects(for: channel).nosync_onChannelAttached(hasObjects: hasObjects, resumed: resumed)`
`119`	`123`	`}`
`120`	`124`
`121`	`125`	`internal func nosync_handleObjectProtocolMessage(withObjectMessages publicObjectMessages: [any _AblyPluginSupportPrivate.ObjectMessageProtocol], channel: _AblyPluginSupportPrivate.RealtimeChannel) {`