[ECO-5056] use server-provided GC grace period for garbage collection

- Add support for server-configured objectsGCGracePeriod from connection details,
with fallback to default 24-hour period. Updates ObjectsPool and all LiveObjects
- Implementations to use the server-provided value for tombstone cleanup timing.

Author: sacOO7

lib/src/main/java/io/ably/lib/objects/Adapter.java

@@ -2,7 +2,7 @@
 
 import io.ably.lib.realtime.AblyRealtime;
 import io.ably.lib.realtime.ChannelBase;
-import io.ably.lib.transport.ConnectionManager;
+import io.ably.lib.realtime.Connection;
 import io.ably.lib.types.AblyException;
 import io.ably.lib.types.ClientOptions;
 import io.ably.lib.types.ErrorInfo;
@@ -23,8 +23,8 @@ public Adapter(@NotNull AblyRealtime ably) {
     }
 
     @Override
-    public @NotNull ConnectionManager getConnectionManager() {
-        return ably.connection.connectionManager;
+    public @NotNull Connection getConnection() {
+        return ably.connection;
     }
 
     @Override

lib/src/main/java/io/ably/lib/objects/ObjectsAdapter.java

@@ -1,7 +1,7 @@
 package io.ably.lib.objects;
 
 import io.ably.lib.realtime.ChannelBase;
-import io.ably.lib.transport.ConnectionManager;
+import io.ably.lib.realtime.Connection;
 import io.ably.lib.types.AblyException;
 import io.ably.lib.types.ClientOptions;
 import org.jetbrains.annotations.Blocking;
@@ -18,13 +18,13 @@ public interface ObjectsAdapter {
     @NotNull ClientOptions getClientOptions();
 
     /**
-     * Retrieves the connection manager for handling connection state and operations.
+     * Retrieves the connection instance for handling connection state and operations.
      * Used to check connection status, obtain error information, and manage
      * message transmission across the Ably connection.
      *
-     * @return the connection manager instance
+     * @return the connection instance
      */
-    @NotNull ConnectionManager getConnectionManager();
+    @NotNull Connection getConnection();
 
     /**
      * Retrieves the current time in milliseconds from the Ably server.

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

@@ -100,6 +100,7 @@ public class ConnectionManager implements ConnectListener {
      * This field is initialized only if the LiveObjects plugin is present in the classpath.
      */
     private final LiveObjectsPlugin liveObjectsPlugin;
+    public Long objectsGCGracePeriod = null;
 
     /**
      * Methods on the channels map owned by the {@link AblyRealtime} instance
@@ -1297,6 +1298,7 @@ private synchronized void onConnected(ProtocolMessage message) {
         maxIdleInterval = connectionDetails.maxIdleInterval;
         connectionStateTtl = connectionDetails.connectionStateTtl;
         maxMessageSize = connectionDetails.maxMessageSize;
+        objectsGCGracePeriod = connectionDetails.objectsGCGracePeriod;
 
         /* set the clientId resolved from token, if any */
         String clientId = connectionDetails.clientId;

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

@@ -74,6 +74,11 @@ public class ConnectionDetails {
      */
     public Long connectionStateTtl;
 
+    /**
+     * The duration in milliseconds used to retain tombstoned objects at client side.
+     */
+    public Long objectsGCGracePeriod;
+
     ConnectionDetails() {
         maxIdleInterval = Defaults.maxIdleInterval;
         connectionStateTtl = Defaults.connectionStateTtl;
@@ -114,6 +119,9 @@ ConnectionDetails readMsgpack(MessageUnpacker unpacker) throws IOException {
                 case "connectionStateTtl":
                     connectionStateTtl = unpacker.unpackLong();
                     break;
+                case "objectsGCGracePeriod":
+                    objectsGCGracePeriod = unpacker.unpackLong();
+                    break;
                 default:
                     Log.v(TAG, "Unexpected field: " + fieldName);
                     unpacker.skipValue();

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

@@ -2,6 +2,7 @@ package io.ably.lib.objects
 
 import io.ably.lib.realtime.ChannelState
 import io.ably.lib.realtime.CompletionListener
+import io.ably.lib.realtime.ConnectionEvent
 import io.ably.lib.types.ChannelMode
 import io.ably.lib.types.ErrorInfo
 import io.ably.lib.types.ProtocolMessage
@@ -15,7 +16,7 @@ import kotlin.coroutines.resumeWithException
  */
 internal suspend fun ObjectsAdapter.sendAsync(message: ProtocolMessage) = suspendCancellableCoroutine { continuation ->
   try {
-    connectionManager.send(message, clientOptions.queueMessages, object : CompletionListener {
+    connection.connectionManager.send(message, clientOptions.queueMessages, object : CompletionListener {
       override fun onSuccess() {
         continuation.resume(Unit)
       }
@@ -45,6 +46,17 @@ internal suspend fun ObjectsAdapter.attachAsync(channelName: String) = suspendCa
   }
 }
 
+internal fun ObjectsAdapter.retrieveObjectsGCGracePeriod(block : (Long?) -> Unit) {
+  val connectionManager = connection.connectionManager
+  if (connectionManager.objectsGCGracePeriod != null) {
+    block(connectionManager.objectsGCGracePeriod)
+    return
+  }
+  connection.once(ConnectionEvent.connected) {
+    block(connectionManager.objectsGCGracePeriod)
+  }
+}
+
 /**
  * Retrieves the channel modes for a specific channel.
  * This method returns the modes that are set for the specified channel.
@@ -76,7 +88,7 @@ internal fun ObjectsAdapter.getChannelModes(channelName: String): Array<ChannelM
  * Spec: RTO15d
  */
 internal fun ObjectsAdapter.ensureMessageSizeWithinLimit(objectMessages: Array<ObjectMessage>) {
-  val maximumAllowedSize = connectionManager.maxMessageSize
+  val maximumAllowedSize = connection.connectionManager.maxMessageSize
   val objectsTotalMessageSize = objectMessages.sumOf { it.size() }
   if (objectsTotalMessageSize > maximumAllowedSize) {
     throw ablyException("ObjectMessages size $objectsTotalMessageSize exceeds maximum allowed size of $maximumAllowedSize bytes",
@@ -131,8 +143,8 @@ internal fun ObjectsAdapter.throwIfInvalidWriteApiConfiguration(channelName: Str
 }
 
 internal fun ObjectsAdapter.throwIfUnpublishableState(channelName: String) {
-  if (!connectionManager.isActive) {
-    throw ablyException(connectionManager.stateErrorInfo)
+  if (!connection.connectionManager.isActive) {
+    throw ablyException(connection.connectionManager.stateErrorInfo)
   }
   throwIfInChannelState(channelName, arrayOf(ChannelState.failed, ChannelState.suspended))
 }

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

@@ -14,6 +14,9 @@ import java.util.concurrent.ConcurrentHashMap
 internal object ObjectsPoolDefaults {
   const val GC_INTERVAL_MS = 1000L * 60 * 5 // 5 minutes
   /**
+   * The SDK will attempt to use the `objectsGCGracePeriod` value provided by the server in the `connectionDetails`
+   * object of the `CONNECTED` event.
+   * If the server does not provide this value, the SDK will fall back to this default value.
    * Must be > 2 minutes to ensure we keep tombstones long enough to avoid the possibility of receiving an operation
    * with an earlier serial that would not have been applied if the tombstone still existed.
    *
@@ -49,10 +52,19 @@ internal class ObjectsPool(
   private val gcScope = CoroutineScope(Dispatchers.Default + SupervisorJob())
   private var gcJob: Job // Job for the garbage collection coroutine
 
+  @Volatile
+  private var gcGracePeriod = ObjectsPoolDefaults.GC_GRACE_PERIOD_MS
+
   init {
     // RTO3b - Initialize pool with root object
     pool[ROOT_OBJECT_ID] = DefaultLiveMap.zeroValue(ROOT_OBJECT_ID, realtimeObjects)
-    // Start garbage collection coroutine
+    // Start garbage collection coroutine with server-provided grace period if available
+    realtimeObjects.adapter.retrieveObjectsGCGracePeriod { period ->
+      period?.let {
+        gcGracePeriod = it
+        Log.i(tag, "Using objectsGCGracePeriod from server: $gcGracePeriod ms")
+      } ?: Log.i(tag, "Server did not provide objectsGCGracePeriod, using default: $gcGracePeriod ms")
+    }
     gcJob = startGCJob()
   }
 
@@ -123,9 +135,9 @@ internal class ObjectsPool(
    */
   private fun onGCInterval() {
     pool.entries.removeIf { (_, obj) ->
-      if (obj.isEligibleForGc()) { true } // Remove from pool
+      if (obj.isEligibleForGc(gcGracePeriod)) { true } // Remove from pool
       else {
-        obj.onGCInterval()
+        obj.onGCInterval(gcGracePeriod)
         false  // Keep in pool
       }
     }

liveobjects/src/main/kotlin/io/ably/lib/objects/type/BaseRealtimeObject.kt

@@ -3,7 +3,6 @@ package io.ably.lib.objects.type
 import io.ably.lib.objects.ObjectMessage
 import io.ably.lib.objects.ObjectOperation
 import io.ably.lib.objects.ObjectState
-import io.ably.lib.objects.ObjectsPoolDefaults
 import io.ably.lib.objects.objectError
 import io.ably.lib.objects.type.livecounter.noOpCounterUpdate
 import io.ably.lib.objects.type.livemap.noOpMapUpdate
@@ -136,10 +135,20 @@ internal abstract class BaseRealtimeObject(
 
   /**
    * Checks if the object is eligible for garbage collection.
+   * 
+   * An object is eligible for garbage collection if it has been tombstoned and 
+   * the time since tombstoning exceeds the specified grace period.
+   * 
+   * @param gcGracePeriod The grace period in milliseconds that tombstoned objects
+   *                      should be kept before being eligible for collection.
+   *                      This value is retrieved from the server's connection details
+   *                      or defaults to 24 hours if not provided by the server.
+   * @return true if the object is tombstoned and the grace period has elapsed,
+   *         false otherwise
    */
-  internal fun isEligibleForGc(): Boolean {
+  internal fun isEligibleForGc(gcGracePeriod: Long): Boolean {
     val currentTime = System.currentTimeMillis()
-    return isTombstoned && tombstonedAt?.let { currentTime - it >= ObjectsPoolDefaults.GC_GRACE_PERIOD_MS } == true
+    return isTombstoned && tombstonedAt?.let { currentTime - it >= gcGracePeriod } == true
   }
 
   /**
@@ -195,12 +204,22 @@ internal abstract class BaseRealtimeObject(
   /**
    * Called during garbage collection intervals to clean up expired entries.
    *
+   * This method is invoked periodically (every 5 minutes) by the ObjectsPool
+   * to perform cleanup of tombstoned data that has exceeded the grace period.
+   *
    * This method should identify and remove entries that:
    * - Have been marked as tombstoned
-   * - Have a tombstone timestamp older than the configured grace period
+   * - Have a tombstone timestamp older than the specified grace period
+   *
+   * @param gcGracePeriod The grace period in milliseconds that tombstoned entries
+   *                      should be kept before being eligible for removal.
+   *                      This value is retrieved from the server's connection details
+   *                      or defaults to 24 hours if not provided by the server.
+   *                      Must be greater than 2 minutes to ensure proper operation
+   *                      ordering and avoid issues with delayed operations.
    *
    * Implementations typically use single-pass removal techniques to
    * efficiently clean up expired data without creating temporary collections.
    */
-  abstract fun onGCInterval()
+  abstract fun onGCInterval(gcGracePeriod: Long)
 }

liveobjects/src/main/kotlin/io/ably/lib/objects/type/livecounter/DefaultLiveCounter.kt

@@ -109,7 +109,7 @@ internal class DefaultLiveCounter private constructor(
     liveCounterManager.notify(update as LiveCounterUpdate)
   }
 
-  override fun onGCInterval() {
+  override fun onGCInterval(gcGracePeriod: Long) {
     // Nothing to GC for a counter object
     return
   }

liveobjects/src/main/kotlin/io/ably/lib/objects/type/livemap/DefaultLiveMap.kt

@@ -182,8 +182,8 @@ internal class DefaultLiveMap private constructor(
     liveMapManager.notify(update as LiveMapUpdate)
   }
 
-  override fun onGCInterval() {
-    data.entries.removeIf { (_, entry) -> entry.isEligibleForGc() }
+  override fun onGCInterval(gcGracePeriod: Long) {
+    data.entries.removeIf { (_, entry) -> entry.isEligibleForGc(gcGracePeriod) }
   }
 
   companion object {

liveobjects/src/main/kotlin/io/ably/lib/objects/type/livemap/LiveMapEntry.kt

@@ -3,7 +3,6 @@ package io.ably.lib.objects.type.livemap
 import io.ably.lib.objects.*
 import io.ably.lib.objects.ObjectData
 import io.ably.lib.objects.ObjectsPool
-import io.ably.lib.objects.ObjectsPoolDefaults
 import io.ably.lib.objects.type.BaseRealtimeObject
 import io.ably.lib.objects.type.ObjectType
 import io.ably.lib.objects.type.counter.LiveCounter
@@ -61,9 +60,9 @@ internal fun LiveMapEntry.getResolvedValue(objectsPool: ObjectsPool): LiveMapVal
 /**
  * Extension function to check if a LiveMapEntry is expired and ready for garbage collection
  */
-internal fun LiveMapEntry.isEligibleForGc(): Boolean {
+internal fun LiveMapEntry.isEligibleForGc(gcGracePeriod: Long): Boolean {
   val currentTime = System.currentTimeMillis()
-  return isTombstoned && tombstonedAt?.let { currentTime - it >= ObjectsPoolDefaults.GC_GRACE_PERIOD_MS } == true
+  return isTombstoned && tombstonedAt?.let { currentTime - it >= gcGracePeriod } == true
 }
 
 private fun fromObjectValue(objValue: ObjectValue): LiveMapValue {