[AIT-276] fix: clear sync objects data pool only when channel state is not suspended

Author: ttypic

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

@@ -319,19 +319,24 @@ internal class DefaultRealtimeObjects(internal val channelName: String, internal
         ChannelState.detached,
         ChannelState.suspended,
         ChannelState.failed -> {
-          val errorReason = try { adapter.getChannel(channelName).reason } catch (e: Exception) { null }
+          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()
+          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/ObjectsManager.kt

@@ -122,6 +122,7 @@ internal class ObjectsManager(private val realtimeObjects: DefaultRealtimeObject
    */
   internal fun failBufferedAcks(error: AblyException) {
     syncCompletionWaiter?.completeExceptionally(error)
+    syncCompletionWaiter = null
   }
 
   /**

liveobjects/src/test/kotlin/io/ably/lib/objects/unit/objects/DefaultRealtimeObjectsTest.kt

@@ -16,14 +16,19 @@ import io.ably.lib.objects.unit.BufferedObjectOperations
 import io.ably.lib.objects.unit.ObjectsManager
 import io.ably.lib.objects.unit.SyncObjectsDataPool
 import io.ably.lib.objects.unit.getDefaultRealtimeObjectsWithMockedDeps
+import io.ably.lib.objects.unit.getMockRealtimeChannel
 import io.ably.lib.objects.unit.size
 import io.ably.lib.realtime.ChannelState
+import io.ably.lib.types.AblyException
+import io.ably.lib.types.ErrorInfo
 import io.ably.lib.types.ProtocolMessage
+import io.mockk.every
 import io.mockk.verify
+import kotlinx.coroutines.CompletableDeferred
 import kotlinx.coroutines.test.runTest
 import org.junit.Test
 import kotlin.test.assertEquals
-import io.mockk.every
+import kotlin.test.assertNotNull
 
 class DefaultRealtimeObjectsTest {
 
@@ -155,6 +160,104 @@ class DefaultRealtimeObjectsTest {
     }
   }
 
+  @Test
+  fun `(RTO20e1) handleStateChange(DETACHED) fails pending ACK waiters with error 92008`() = runTest {
+    val defaultRealtimeObjects = getDefaultRealtimeObjectsWithMockedDeps()
+
+    // Capture the error passed to failBufferedAcks via a CompletableDeferred
+    val capturedError = CompletableDeferred<AblyException>()
+    every { defaultRealtimeObjects.ObjectsManager.failBufferedAcks(any()) } answers {
+      capturedError.complete(firstArg())
+      callOriginal()
+    }
+
+    defaultRealtimeObjects.handleStateChange(ChannelState.detached, false)
+
+    val error = capturedError.await()
+    assertEquals(92008, error.errorInfo.code) // PublishAndApplyFailedDueToChannelState
+  }
+
+  @Test
+  fun `(RTO20e1) handleStateChange(SUSPENDED) fails pending ACK waiters with error 92008`() = runTest {
+    val defaultRealtimeObjects = getDefaultRealtimeObjectsWithMockedDeps()
+
+    val capturedError = CompletableDeferred<AblyException>()
+    every { defaultRealtimeObjects.ObjectsManager.failBufferedAcks(any()) } answers {
+      capturedError.complete(firstArg())
+      callOriginal()
+    }
+
+    defaultRealtimeObjects.handleStateChange(ChannelState.suspended, false)
+
+    val error = capturedError.await()
+    assertEquals(92008, error.errorInfo.code) // PublishAndApplyFailedDueToChannelState
+  }
+
+  @Test
+  fun `(RTO20e1) handleStateChange(FAILED) fails pending ACK waiters and propagates channel reason`() = runTest {
+    val defaultRealtimeObjects = getDefaultRealtimeObjectsWithMockedDeps()
+
+    // Override the channel returned by the adapter to carry a non-null reason
+    val channelReason = ErrorInfo("channel failed due to auth error", 40100, 401)
+    val channelWithReason = getMockRealtimeChannel("testChannelName")
+    channelWithReason.reason = channelReason
+    every { defaultRealtimeObjects.adapter.getChannel(any()) } returns channelWithReason
+
+    val capturedError = CompletableDeferred<AblyException>()
+    every { defaultRealtimeObjects.ObjectsManager.failBufferedAcks(any()) } answers {
+      capturedError.complete(firstArg())
+      callOriginal()
+    }
+
+    defaultRealtimeObjects.handleStateChange(ChannelState.failed, false)
+
+    val error = capturedError.await()
+    assertEquals(92008, error.errorInfo.code)
+    val causeException = error.cause as? AblyException
+    assertNotNull(causeException, "Error cause must include the channel's reason")
+    assertEquals(channelReason.code, causeException.errorInfo.code)
+    assertEquals(channelReason.message, causeException.errorInfo.message)
+  }
+
+  @Test
+  fun `(RTO4) handleStateChange(SUSPENDED) does NOT clear objects data`() = runTest {
+    val defaultRealtimeObjects = getDefaultRealtimeObjectsWithMockedDeps()
+
+    // Use the failBufferedAcks call as a signal that the state-change coroutine has run to completion
+    val failCalled = CompletableDeferred<Unit>()
+    every { defaultRealtimeObjects.ObjectsManager.failBufferedAcks(any()) } answers {
+      callOriginal()
+      failCalled.complete(Unit)
+    }
+
+    defaultRealtimeObjects.handleStateChange(ChannelState.suspended, false)
+
+    // For SUSPENDED, the coroutine ends immediately after failBufferedAcks (no clear calls)
+    failCalled.await()
+
+    verify(exactly = 0) { defaultRealtimeObjects.objectsPool.clearObjectsData(any()) }
+    verify(exactly = 0) { defaultRealtimeObjects.ObjectsManager.clearSyncObjectsDataPool() }
+  }
+
+  @Test
+  fun `(RTO4) handleStateChange(DETACHED) clears objects data and sync pool`() = runTest {
+    val defaultRealtimeObjects = getDefaultRealtimeObjectsWithMockedDeps()
+
+    // Use clearSyncObjectsDataPool (the last operation in the coroutine) as the completion signal
+    val syncPoolCleared = CompletableDeferred<Unit>()
+    every { defaultRealtimeObjects.ObjectsManager.clearSyncObjectsDataPool() } answers {
+      callOriginal()
+      syncPoolCleared.complete(Unit)
+    }
+
+    defaultRealtimeObjects.handleStateChange(ChannelState.detached, false)
+
+    syncPoolCleared.await()
+
+    verify(exactly = 1) { defaultRealtimeObjects.objectsPool.clearObjectsData(false) }
+    verify(exactly = 1) { defaultRealtimeObjects.ObjectsManager.clearSyncObjectsDataPool() }
+  }
+
   @Test
   fun `(OM2) Populate objectMessage missing id, timestamp and connectionId from protocolMessage`() = runTest {
     val defaultRealtimeObjects = getDefaultRealtimeObjectsWithMockedDeps()