[AIT-276] feat: introduce ACK-based local application of LiveObjects ops

lib/src/main/java/io/ably/lib/transport/ConnectionManager.java

@@ -1318,6 +1318,7 @@ private synchronized void onConnected(ProtocolMessage message) {
         maxIdleInterval = connectionDetails.maxIdleInterval;
         connectionStateTtl = connectionDetails.connectionStateTtl;
         maxMessageSize = connectionDetails.maxMessageSize;
+        siteCode = connectionDetails.siteCode; // CD2j
 
         /* set the clientId resolved from token, if any */
         String clientId = connectionDetails.clientId;
@@ -2033,6 +2034,7 @@ private boolean isFatalError(ErrorInfo err) {
     private CMConnectivityListener connectivityListener;
     private long connectionStateTtl = Defaults.connectionStateTtl;
     public int maxMessageSize = Defaults.maxMessageSize;
+    public String siteCode; // CD2j
     long maxIdleInterval = Defaults.maxIdleInterval;
     private int disconnectedRetryAttempt = 0;
 

lib/src/main/java/io/ably/lib/types/ConnectionDetails.java

@@ -73,6 +73,13 @@ public class ConnectionDetails {
      * Spec: CD2f, RTN14e, DF1a
      */
     public Long connectionStateTtl;
+    /**
+     * An opaque string identifying the server instance that the client is connected to.
+     * Used as a key in siteTimeserials maps for LiveObjects operations.
+     * <p>
+     * Spec: CD2j
+     */
+    public String siteCode;
 
     ConnectionDetails() {
         maxIdleInterval = Defaults.maxIdleInterval;
@@ -114,6 +121,9 @@ ConnectionDetails readMsgpack(MessageUnpacker unpacker) throws IOException {
                 case "connectionStateTtl":
                     connectionStateTtl = unpacker.unpackLong();
                     break;
+                case "siteCode":
+                    siteCode = unpacker.unpackString();
+                    break;
                 default:
                     Log.v(TAG, "Unexpected field: " + fieldName);
                     unpacker.skipValue();

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

@@ -12,6 +12,7 @@ import io.ably.lib.objects.type.map.LiveMapValue
 import io.ably.lib.realtime.ChannelState
 import io.ably.lib.types.AblyException
 import io.ably.lib.types.ProtocolMessage
+import io.ably.lib.types.PublishResult
 import io.ably.lib.util.Log
 import kotlinx.coroutines.*
 import kotlinx.coroutines.channels.Channel.Factory.UNLIMITED
@@ -31,6 +32,12 @@ internal class DefaultRealtimeObjects(internal val channelName: String, internal
 
   internal var state = ObjectsState.Initialized
 
+  /**
+   * Set of serials for operations applied locally upon ACK, awaiting deduplication of the server echo.
+   * @spec RTO7b, RTO7b1
+   */
+  internal val appliedOnAckSerials = mutableSetOf<String>()
+
   /**
    * @spec RTO4 - Used for handling object messages and object sync messages
    */
@@ -125,13 +132,12 @@ internal class DefaultRealtimeObjects(internal val channelName: String, internal
       )
     )
 
-    // RTO11g - Publish the message
-    publish(arrayOf(msg))
+    // RTO11i - publish and apply locally on ACK
+    publishAndApply(arrayOf(msg))
 
-    // RTO11h - Check if object already exists in pool, otherwise create a zero-value object using the sequential scope
-    return objectsPool.get(objectId) as? LiveMap ?: withContext(sequentialScope.coroutineContext) {
-      objectsPool.createZeroValueObjectIfNotExists(objectId) as LiveMap
-    }
+    // RTO11h2 - Return existing object if found after apply
+    return objectsPool.get(objectId) as? LiveMap
+      ?: throw serverError("createMap: MAP_CREATE was not applied as expected; objectId=$objectId") // RTO11h3d
   }
 
   private suspend fun createCounterAsync(initialValue: Number): LiveCounter {
@@ -161,13 +167,12 @@ internal class DefaultRealtimeObjects(internal val channelName: String, internal
       )
     )
 
-    // RTO12g - Publish the message
-    publish(arrayOf(msg))
+    // RTO12i - publish and apply locally on ACK
+    publishAndApply(arrayOf(msg))
 
-    // RTO12h - Check if object already exists in pool, otherwise create a zero-value object using the sequential scope
-    return objectsPool.get(objectId) as? LiveCounter ?: withContext(sequentialScope.coroutineContext) {
-      objectsPool.createZeroValueObjectIfNotExists(objectId) as LiveCounter
-    }
+    // RTO12h2 - Return existing object if found after apply
+    return objectsPool.get(objectId) as? LiveCounter
+      ?: throw serverError("createCounter: COUNTER_CREATE was not applied as expected; objectId=$objectId") // RTO12h3d
   }
 
   /**
@@ -182,15 +187,55 @@ internal class DefaultRealtimeObjects(internal val channelName: String, internal
   /**
    * Spec: RTO15
    */
-  internal suspend fun publish(objectMessages: Array<ObjectMessage>) {
+  internal suspend fun publish(objectMessages: Array<ObjectMessage>): PublishResult {
     // RTO15b, RTL6c - Ensure that the channel is in a valid state for publishing
     adapter.throwIfUnpublishableState(channelName)
     adapter.ensureMessageSizeWithinLimit(objectMessages)
     // RTO15e - Must construct the ProtocolMessage as per RTO15e1, RTO15e2, RTO15e3
     val protocolMessage = ProtocolMessage(ProtocolMessage.Action.`object`, channelName)
     protocolMessage.state = objectMessages
     // RTO15f, RTO15g - Send the ProtocolMessage using the adapter and capture success/failure
-    adapter.sendAsync(protocolMessage)
+    return adapter.sendAsync(protocolMessage) // RTO15h
+  }
+
+  /**
+   * Publishes the given object messages and, upon receiving the ACK, immediately applies them
+   * locally as synthetic inbound messages using the assigned serial and connection's siteCode.
+   *
+   * Spec: RTO20
+   */
+  internal suspend fun publishAndApply(objectMessages: Array<ObjectMessage>) {
+    // RTO20b - publish, propagate failure
+    val publishResult = publish(objectMessages)
+
+    // RTO20c - validate required info
+    val siteCode = adapter.connectionManager.siteCode
+    if (siteCode == null) {
+      Log.e(tag, "RTO20c1: siteCode not available; operations will be applied when echoed")
+      return
+    }
+    val serials = publishResult.serials
+    if (serials == null || serials.size != objectMessages.size) {
+      Log.e(tag, "RTO20c2: PublishResult.serials unavailable or wrong length; operations will be applied when echoed")
+      return
+    }
+
+    // RTO20d - create synthetic inbound ObjectMessages
+    val syntheticMessages = mutableListOf<ObjectMessage>()
+    objectMessages.forEachIndexed { i, msg ->
+      val serial = serials[i]
+      if (serial == null) {
+        Log.d(tag, "RTO20d1: serial null at index $i (conflated), skipping")
+        return@forEachIndexed
+      }
+      syntheticMessages.add(msg.copy(serial = serial, siteCode = siteCode)) // RTO20d2a, RTO20d2b, RTO20d3
+    }
+    if (syntheticMessages.isEmpty()) return
+
+    // RTO20e, RTO20f - dispatch to sequential scope for ordering
+    withContext(sequentialScope.coroutineContext) {
+      objectsManager.applyAckResult(syntheticMessages) // suspends if SYNCING (RTO20e), applies on SYNCED (RTO20f)
+    }
   }
 
   /**
@@ -268,11 +313,20 @@ internal class DefaultRealtimeObjects(internal val channelName: String, internal
             objectsManager.clearBufferedObjectOperations() // RTO4b5
             // defer the state change event until the next tick if we started a new sequence just now due to being in initialized state.
             // this allows any event listeners to process the start of the new sequence event that was emitted earlier during this event loop.
-            objectsManager.endSync(fromInitializedState) // RTO4b4
+            objectsManager.endSync() // RTO4b4
           }
         }
         ChannelState.detached,
+        ChannelState.suspended,
         ChannelState.failed -> {
+          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) // 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/ErrorCodes.kt

@@ -11,6 +11,7 @@ internal enum class ErrorCode(public val code: Int) {
   // Channel mode and state validation error codes
   ChannelModeRequired(40_024),
   ChannelStateError(90_001),
+  PublishAndApplyFailedDueToChannelState(92_008),
 }
 
 internal enum class HttpStatusCode(public val code: Int) {

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

@@ -4,7 +4,9 @@ import io.ably.lib.objects.type.BaseRealtimeObject
 import io.ably.lib.objects.type.ObjectUpdate
 import io.ably.lib.objects.type.livecounter.DefaultLiveCounter
 import io.ably.lib.objects.type.livemap.DefaultLiveMap
+import io.ably.lib.types.AblyException
 import io.ably.lib.util.Log
+import kotlinx.coroutines.CompletableDeferred
 
 /**
  * @spec RTO5 - Processes OBJECT and OBJECT_SYNC messages during sync sequences
@@ -21,6 +23,7 @@ internal class ObjectsManager(private val realtimeObjects: DefaultRealtimeObject
    * @spec RTO7 - Buffered object operations during sync
    */
   private val bufferedObjectOperations = mutableListOf<ObjectMessage>() // RTO7a
+  private var syncCompletionWaiter: CompletableDeferred<Unit>? = null
 
   /**
    * Handles object messages (non-sync messages).
@@ -39,7 +42,7 @@ internal class ObjectsManager(private val realtimeObjects: DefaultRealtimeObject
     }
 
     // Apply messages immediately if synced
-    applyObjectMessages(objectMessages) // RTO8b
+    applyObjectMessages(objectMessages, ObjectsOperationSource.CHANNEL) // RTO8b
   }
 
   /**
@@ -62,7 +65,7 @@ internal class ObjectsManager(private val realtimeObjects: DefaultRealtimeObject
     if (syncTracker.hasSyncEnded()) {
       // defer the state change event until the next tick if this was a new sync sequence
       // to allow any event listeners to process the start of the new sequence event that was emitted earlier during this event loop.
-      endSync(isNewSync)
+      endSync()
     }
   }
 
@@ -78,25 +81,47 @@ internal class ObjectsManager(private val realtimeObjects: DefaultRealtimeObject
     bufferedObjectOperations.clear() // RTO5a2b
     syncObjectsDataPool.clear() // RTO5a2a
     currentSyncId = syncId
-    stateChange(ObjectsState.Syncing, false)
+    syncCompletionWaiter = CompletableDeferred()
+    stateChange(ObjectsState.Syncing)
   }
 
   /**
    * Ends the current sync sequence.
    *
    * @spec RTO5c - Applies sync data and buffered operations
    */
-  internal fun endSync(deferStateEvent: Boolean) {
+  internal fun endSync() {
     Log.v(tag, "Ending sync sequence")
-    applySync()
-    // should apply buffered object operations after we applied the sync.
-    // can use regular non-sync object.operation logic
-    applyObjectMessages(bufferedObjectOperations) // RTO5c6
-
-    bufferedObjectOperations.clear() // RTO5c5
-    syncObjectsDataPool.clear() // RTO5c4
-    currentSyncId = null // RTO5c3
-    stateChange(ObjectsState.Synced, deferStateEvent)
+    applySync()                                                                    // RTO5c1/2/7
+    applyObjectMessages(bufferedObjectOperations, ObjectsOperationSource.CHANNEL) // RTO5c6
+    bufferedObjectOperations.clear()                                               // RTO5c5
+    syncObjectsDataPool.clear()                                                    // RTO5c4
+    currentSyncId = null                                                           // RTO5c3
+    realtimeObjects.appliedOnAckSerials.clear()                                    // RTO5c9
+    stateChange(ObjectsState.Synced)                              // RTO5c8
+    syncCompletionWaiter?.complete(Unit)
+    syncCompletionWaiter = null
+  }
+
+  /**
+   * Called from publishAndApply (via withContext sequentialScope).
+   * If SYNCED: apply immediately with LOCAL source.
+   * If not SYNCED: suspend until endSync transitions to SYNCED (RTO20e), then apply.
+   */
+  internal suspend fun applyAckResult(messages: List<ObjectMessage>) {
+    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 pending apply waiters.
+   * Called when the channel enters DETACHED/SUSPENDED/FAILED (RTO20e1).
+   */
+  internal fun failBufferedAcks(error: AblyException) {
+    syncCompletionWaiter?.completeExceptionally(error)
   }
 
   /**
@@ -162,7 +187,10 @@ internal class ObjectsManager(private val realtimeObjects: DefaultRealtimeObject
    *
    * @spec RTO9 - Creates zero-value objects if they don't exist
    */
-  private fun applyObjectMessages(objectMessages: List<ObjectMessage>) {
+  private fun applyObjectMessages(
+    objectMessages: List<ObjectMessage>,
+    source: ObjectsOperationSource = ObjectsOperationSource.CHANNEL,
+  ) {
     // RTO9a
     for (objectMessage in objectMessages) {
       if (objectMessage.operation == null) {
@@ -177,14 +205,30 @@ internal class ObjectsManager(private val realtimeObjects: DefaultRealtimeObject
         Log.w(tag, "Object operation action is unknown, skipping message: ${objectMessage.id}")
         continue
       }
+
+      // RTO9a3 - skip operations already applied on ACK (discard without taking any further action).
+      // This check comes before zero-value object creation (RTO9a2a1) so that no zero-value object is
+      // created for an objectId not yet in the pool when the echo is being discarded.
+      // Note: siteTimeserials is NOT updated here intentionally — updating it to the echo's serial would
+      // incorrectly reject older-but-unprocessed operations from the same site that arrive after the echo.
+      if (objectMessage.serial != null &&
+        realtimeObjects.appliedOnAckSerials.contains(objectMessage.serial)) {
+        Log.d(tag, "RTO9a3: serial ${objectMessage.serial} already applied on ACK; discarding echo")
+        realtimeObjects.appliedOnAckSerials.remove(objectMessage.serial)
+        continue // discard without taking any further action
+      }
+
       // RTO9a2a - we can receive an op for an object id we don't have yet in the pool. instead of buffering such operations,
       // we can create a zero-value object for the provided object id and apply the operation to that zero-value object.
       // this also means that all objects are capable of applying the corresponding *_CREATE ops on themselves,
       // since they need to be able to eventually initialize themselves from that *_CREATE op.
       // so to simplify operations handling, we always try to create a zero-value object in the pool first,
       // and then we can always apply the operation on the existing object in the pool.
       val obj = realtimeObjects.objectsPool.createZeroValueObjectIfNotExists(objectOperation.objectId) // RTO9a2a1
-      obj.applyObject(objectMessage) // RTO9a2a2, RTO9a2a3
+      val applied = obj.applyObject(objectMessage, source) // RTO9a2a2, RTO9a2a3
+      if (source == ObjectsOperationSource.LOCAL && applied && objectMessage.serial != null) {
+        realtimeObjects.appliedOnAckSerials.add(objectMessage.serial) // RTO9a2a4
+      }
     }
   }
 
@@ -228,7 +272,7 @@ internal class ObjectsManager(private val realtimeObjects: DefaultRealtimeObject
    *
    * @spec RTO2 - Emits state change events for syncing and synced states
    */
-  private fun stateChange(newState: ObjectsState, deferEvent: Boolean) {
+  private fun stateChange(newState: ObjectsState) {
     if (realtimeObjects.state == newState) {
       return
     }
@@ -240,6 +284,7 @@ internal class ObjectsManager(private val realtimeObjects: DefaultRealtimeObject
   }
 
   internal fun dispose() {
+    syncCompletionWaiter?.cancel()
     syncObjectsDataPool.clear()
     bufferedObjectOperations.clear()
     disposeObjectsStateListeners()

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

@@ -0,0 +1,7 @@
+package io.ably.lib.objects
+
+/** @spec RTO22 */
+internal enum class ObjectsOperationSource {
+    LOCAL,   // RTO22a - applied upon receipt of ACK
+    CHANNEL  // RTO22b - received over a Realtime channel
+}