[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,16 +313,30 @@ 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 -> {
- // 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()
+ 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
+ if (state != ChannelState.suspended) {
+ // 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()
+ }
  }
-
  else -> {
  // No action needed for other states
  }

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,48 @@ 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)
+ syncCompletionWaiter = null
  }
 
  /**
@@ -162,7 +188,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 +206,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 +273,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 +285,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
+}