Merge pull request #1158 from ably/feature/liveobjects-use-server-gc-period

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

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

@@ -104,6 +104,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
@@ -1319,6 +1320,7 @@ private synchronized void onConnected(ProtocolMessage message) {
         connectionStateTtl = connectionDetails.connectionStateTtl;
         maxMessageSize = connectionDetails.maxMessageSize;
         siteCode = connectionDetails.siteCode; // CD2j
+        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

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

lib/src/test/java/io/ably/lib/test/realtime/RealtimeChannelTest.java

@@ -2617,6 +2617,29 @@ public void connect_should_not_rewrite_immediate_attach() throws AblyException {
         }
     }
 
+    @Test
+    public void channel_get_objects_throws_exception() throws AblyException {
+        ClientOptions opts = createOptions(testVars.keys[0].keyStr);
+        try (AblyRealtime ably = new AblyRealtime(opts)) {
+
+            /* wait until connected */
+            new ConnectionWaiter(ably.connection).waitFor(ConnectionState.connected);
+            assertEquals("Verify connected state reached", ably.connection.state, ConnectionState.connected);
+
+            /* create a channel and attach */
+            final Channel channel = ably.channels.get("channel");
+            channel.attach();
+            new ChannelWaiter(channel).waitFor(ChannelState.attached);
+            assertEquals("Verify attached state reached", channel.state, ChannelState.attached);
+
+            AblyException exception = assertThrows(AblyException.class, channel::getObjects);
+            assertNotNull(exception);
+            assertEquals(40019, exception.errorInfo.code);
+            assertEquals(400, exception.errorInfo.statusCode);
+            assertTrue(exception.errorInfo.message.contains("LiveObjects plugin hasn't been installed"));
+        }
+    }
+
     static class DetachingProtocolListener implements DebugOptions.RawProtocolListener {
 
         public Channel theChannel;

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

@@ -3,6 +3,8 @@ package io.ably.lib.objects
 import io.ably.lib.realtime.ChannelState
 import io.ably.lib.realtime.CompletionListener
 import io.ably.lib.types.Callback
+import io.ably.lib.realtime.ConnectionEvent
+import io.ably.lib.realtime.ConnectionStateListener
 import io.ably.lib.types.ChannelMode
 import io.ably.lib.types.ErrorInfo
 import io.ably.lib.types.ProtocolMessage
@@ -12,6 +14,8 @@ import kotlinx.coroutines.suspendCancellableCoroutine
 import kotlin.coroutines.resume
 import kotlin.coroutines.resumeWithException
 
+internal val ObjectsAdapter.connectionManager get() = connection.connectionManager
+
 /**
  * Spec: RTO15g
  */
@@ -47,6 +51,16 @@ internal suspend fun ObjectsAdapter.attachAsync(channelName: String) = suspendCa
   }
 }
 
+internal fun ObjectsAdapter.onGCGracePeriodUpdated(block : (Long?) -> Unit) : ObjectsSubscription {
+  connectionManager.objectsGCGracePeriod?.let { block(it) }
+  // Return new objectsGCGracePeriod whenever connection state changes to connected
+  val listener: (_: ConnectionStateListener.ConnectionStateChange) -> Unit = {
+    block(connectionManager.objectsGCGracePeriod)
+  }
+  connection.on(ConnectionEvent.connected, listener)
+  return ObjectsSubscription { connection.off(listener) }
+}
+
 /**
  * Retrieves the channel modes for a specific channel.
  * This method returns the modes that are set for the specified channel.

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
+  private var gcPeriodSubscription: ObjectsSubscription
+
   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
+    gcPeriodSubscription = realtimeObjects.adapter.onGCGracePeriodUpdated { 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
       }
     }
@@ -152,6 +164,7 @@ internal class ObjectsPool(
    * Should be called when the pool is no longer needed.
    */
   fun dispose() {
+    gcPeriodSubscription.unsubscribe()
     gcJob.cancel()
     gcScope.cancel()
     pool.clear()

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

@@ -4,7 +4,6 @@ import io.ably.lib.objects.ObjectMessage
 import io.ably.lib.objects.ObjectOperation
 import io.ably.lib.objects.ObjectState
 import io.ably.lib.objects.ObjectsOperationSource
-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
@@ -138,10 +137,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
   }
 
   /**
@@ -198,12 +207,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

@@ -111,7 +111,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

@@ -190,8 +190,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 {