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

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 {

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
@@ -73,9 +72,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 fromRealtimeObject(realtimeObject: BaseRealtimeObject): LiveMapValue {