[AIT-276] refactor: fixes based on review comments

Author: ttypic

liveobjects/src/main/kotlin/io/ably/lib/objects/DefaultRealtimeObjects.kt

@@ -319,15 +319,14 @@ internal class DefaultRealtimeObjects(internal val channelName: String, internal
         ChannelState.detached,
         ChannelState.suspended,
         ChannelState.failed -> {
-          // RTO20e1 - fail any publishAndApply operations waiting for sync
           val errorReason = try { adapter.getChannel(channelName).reason } catch (e: Exception) { null }
           val error = ablyException(
             "publishAndApply could not be applied locally: channel entered $state whilst waiting for objects sync",
             ErrorCode.PublishAndApplyFailedDueToChannelState,
             HttpStatusCode.BadRequest,
             cause = errorReason?.let { AblyException.fromErrorInfo(it) }
           )
-          objectsManager.failBufferedAcks(error)
+          objectsManager.failBufferedAcks(error) // RTO20e1
           // do not emit data update events as the actual current state of Objects data is unknown when we're in these channel states
           objectsPool.clearObjectsData(false)
           objectsManager.clearSyncObjectsDataPool()

liveobjects/src/main/kotlin/io/ably/lib/objects/ObjectsManager.kt

@@ -23,10 +23,7 @@ internal class ObjectsManager(private val realtimeObjects: DefaultRealtimeObject
    * @spec RTO7 - Buffered object operations during sync
    */
   private val bufferedObjectOperations = mutableListOf<ObjectMessage>() // RTO7a
-  /**
-   * @spec RTO22 - ACK results buffered during sync, with deferred for caller waiting
-   */
-  private val bufferedAcks = mutableListOf<Pair<List<ObjectMessage>, CompletableDeferred<Unit>>>()
+  private var syncCompletionWaiter: CompletableDeferred<Unit>? = null
 
   /**
    * Handles object messages (non-sync messages).
@@ -83,11 +80,8 @@ internal class ObjectsManager(private val realtimeObjects: DefaultRealtimeObject
     // need to discard all buffered object operation messages on new sync start
     bufferedObjectOperations.clear() // RTO5a2b
     syncObjectsDataPool.clear() // RTO5a2a
-    // RTO21b - clear ACK tracking state on new sync (safety guard; RTO20e1 should have already failed deferreds)
-    for ((_, deferred) in bufferedAcks) { deferred.cancel() }
-    bufferedAcks.clear()
-    realtimeObjects.appliedOnAckSerials.clear()
     currentSyncId = syncId
+    syncCompletionWaiter = CompletableDeferred()
     stateChange(ObjectsState.Syncing, false)
   }
 
@@ -98,47 +92,36 @@ internal class ObjectsManager(private val realtimeObjects: DefaultRealtimeObject
    */
   internal fun endSync(deferStateEvent: Boolean) {
     Log.v(tag, "Ending sync sequence")
-    applySync()
-    realtimeObjects.appliedOnAckSerials.clear() // RTO5c9 - sync replaces state
-
-    // RTO5c6b: apply buffered ACKs before buffered OBJECT messages (proper dedup ordering)
-    for ((messages, deferred) in bufferedAcks) {
-      applyObjectMessages(messages, ObjectsOperationSource.LOCAL)
-      deferred.complete(Unit) // signal publishAndApply to resume
-    }
-    bufferedAcks.clear()
-
+    applySync()                                                                    // RTO5c1/2/7
     applyObjectMessages(bufferedObjectOperations, ObjectsOperationSource.CHANNEL) // RTO5c6
-    bufferedObjectOperations.clear() // RTO5c5
-    syncObjectsDataPool.clear() // RTO5c4
-    currentSyncId = null // RTO5c3
-    stateChange(ObjectsState.Synced, deferStateEvent) // RTO5c8
+    bufferedObjectOperations.clear()                                               // RTO5c5
+    syncObjectsDataPool.clear()                                                    // RTO5c4
+    currentSyncId = null                                                           // RTO5c3
+    realtimeObjects.appliedOnAckSerials.clear()                                    // RTO5c9
+    stateChange(ObjectsState.Synced, deferStateEvent)                              // RTO5c8
+    syncCompletionWaiter?.complete(Unit)
+    syncCompletionWaiter = null
   }
 
   /**
    * Called from publishAndApply (via withContext sequentialScope).
    * If SYNCED: apply immediately with LOCAL source.
-   * If SYNCING: buffer, suspend deferred until endSync processes it (RTO20e).
+   * If not SYNCED: suspend until endSync transitions to SYNCED (RTO20e), then apply.
    */
   internal suspend fun applyAckResult(messages: List<ObjectMessage>) {
-    if (realtimeObjects.state == ObjectsState.Synced) {
-      applyObjectMessages(messages, ObjectsOperationSource.LOCAL) // RTO20f
-    } else {
-      val deferred = CompletableDeferred<Unit>()
-      bufferedAcks.add(Pair(messages, deferred))
-      deferred.await() // RTO20e - suspends until endSync completes or channel fails (RTO20e1)
+    if (realtimeObjects.state != ObjectsState.Synced) {
+      if (syncCompletionWaiter == null) syncCompletionWaiter = CompletableDeferred()
+      syncCompletionWaiter?.await() // suspends; resumes after endSync transitions to SYNCED (RTO20e1)
     }
+    applyObjectMessages(messages, ObjectsOperationSource.LOCAL) // RTO20f
   }
 
   /**
-   * Fails all buffered ACK deferreds.
+   * Fails all pending apply waiters.
    * Called when the channel enters DETACHED/SUSPENDED/FAILED (RTO20e1).
    */
   internal fun failBufferedAcks(error: AblyException) {
-    for ((_, deferred) in bufferedAcks) {
-      deferred.completeExceptionally(error)
-    }
-    bufferedAcks.clear()
+    syncCompletionWaiter?.completeExceptionally(error)
   }
 
   /**
@@ -301,8 +284,7 @@ internal class ObjectsManager(private val realtimeObjects: DefaultRealtimeObject
   }
 
   internal fun dispose() {
-    for ((_, deferred) in bufferedAcks) { deferred.cancel() }
-    bufferedAcks.clear()
+    syncCompletionWaiter?.cancel()
     syncObjectsDataPool.clear()
     bufferedObjectOperations.clear()
     disposeObjectsStateListeners()

liveobjects/src/test/kotlin/io/ably/lib/objects/unit/TestHelpers.kt

@@ -17,6 +17,7 @@ import io.ably.lib.types.ClientOptions
 import io.mockk.every
 import io.mockk.mockk
 import io.mockk.spyk
+import kotlinx.coroutines.CompletableDeferred
 
 internal fun getMockRealtimeChannel(
   channelName: String,
@@ -73,8 +74,8 @@ internal val ObjectsManager.SyncObjectsDataPool: Map<String, ObjectState>
 internal val ObjectsManager.BufferedObjectOperations: List<ObjectMessage>
   get() = this.getPrivateField("bufferedObjectOperations")
 
-internal val ObjectsManager.BufferedAcks: List<Pair<List<ObjectMessage>, *>>
-  get() = this.getPrivateField("bufferedAcks")
+internal val ObjectsManager.SyncCompletionWaiter: CompletableDeferred<Unit>?
+  get() = this.getPrivateField("syncCompletionWaiter")
 
 internal var DefaultRealtimeObjects.ObjectsManager: ObjectsManager
   get() = this.getPrivateField("objectsManager")