Fix the five second stall in lifecycle monitor start and stop

stream-android-core/src/main/java/io/getstream/android/core/api/utils/Threading.kt

@@ -163,7 +163,8 @@ public inline fun <T> runOn(looper: Looper, crossinline block: () -> T): Result<
         } else {
             val latch = CountDownLatch(1)
             var result: Result<T>? = null
-            Handler(looper).post {
+            val handler = Handler(looper)
+            val runnable = Runnable {
                 try {
                     result = Result.success(block())
                 } catch (t: Throwable) {
@@ -172,8 +173,13 @@ public inline fun <T> runOn(looper: Looper, crossinline block: () -> T): Result<
                     latch.countDown()
                 }
             }
+            handler.post(runnable)
 
             if (!latch.await(5, TimeUnit.SECONDS)) {
+                // Drop the message instead of leaving it queued. It would otherwise run once
+                // the looper frees up, long after this call has already failed and the caller
+                // has moved on, applying [block] against state it no longer expects.
+                handler.removeCallbacks(runnable)
                 throw IllegalStateException("Timed out waiting to post to main thread")
             }
             result!!.getOrThrow()

stream-android-core/src/main/java/io/getstream/android/core/internal/observers/lifecycle/StreamLifecycleMonitorImpl.kt

@@ -16,6 +16,8 @@
 
 package io.getstream.android.core.internal.observers.lifecycle
 
+import android.os.Handler
+import android.os.Looper
 import androidx.lifecycle.DefaultLifecycleObserver
 import androidx.lifecycle.Lifecycle
 import androidx.lifecycle.LifecycleOwner
@@ -25,8 +27,8 @@ import io.getstream.android.core.api.observers.lifecycle.StreamLifecycleListener
 import io.getstream.android.core.api.observers.lifecycle.StreamLifecycleMonitor
 import io.getstream.android.core.api.subscribe.StreamSubscription
 import io.getstream.android.core.api.subscribe.StreamSubscriptionManager
-import io.getstream.android.core.api.utils.runOnMainLooper
 import java.util.concurrent.atomic.AtomicBoolean
+import java.util.concurrent.atomic.AtomicInteger
 import kotlinx.coroutines.ExperimentalCoroutinesApi
 
 internal class StreamLifecycleMonitorImpl(
@@ -37,23 +39,77 @@ internal class StreamLifecycleMonitorImpl(
 
     private val started = AtomicBoolean(false)
 
+    /** Attach/detach messages posted to the main looper but not yet executed. */
+    private val pendingOnMain = AtomicInteger(0)
+
+    /** Guards the [started] flip together with the decision to run inline or to post. */
+    private val transitionLock = Any()
+
     override fun subscribe(
         listener: StreamLifecycleListener,
         options: StreamSubscriptionManager.Options,
     ): Result<StreamSubscription> = subscriptionManager.subscribe(listener, options)
 
     override fun start(): Result<Unit> = runCatching {
-        if (!started.compareAndSet(false, true)) {
-            return@runCatching
-        }
-        runOnMainLooper { lifecycle.addObserver(this) }.getOrThrow()
+        transition(from = false, to = true) { lifecycle.addObserver(this) }
     }
 
     override fun stop(): Result<Unit> = runCatching {
-        if (!started.compareAndSet(true, false)) {
-            return@runCatching
+        transition(from = true, to = false) { lifecycle.removeObserver(this) }
+    }
+
+    /**
+     * Flips [started] from [from] to [to] and runs [block] on the main looper without waiting for
+     * it to complete.
+     *
+     * Attaching and detaching the observer has to happen on the main thread, but *waiting* for it
+     * must not: a caller that already occupies the main looper — a suspend `disconnect()` bridged
+     * with `runBlocking`, for example — would block the very thread it is waiting on, and the wait
+     * could only ever end in a timeout.
+     *
+     * Running inline when the caller is already on the main looper is only safe while nothing of
+     * ours is queued. Otherwise this call would jump ahead of an attach/detach posted earlier from
+     * another thread and invert the two, leaving [started] and the observer disagreeing.
+     *
+     * The flip and that decision are taken under [transitionLock] as one step. Apart they leave a
+     * window: a background `start()` that has won the flip but not yet posted looks like nothing is
+     * queued, so a `stop()` arriving on the main thread in between detaches inline and the attach
+     * lands after it — observer attached, [started] false. [block] itself runs outside the lock;
+     * the inline path holds the main thread, so nothing of ours can overtake it there anyway, and
+     * the lock stays clear of the listener callbacks the attach fans out to.
+     */
+    private fun transition(from: Boolean, to: Boolean, block: () -> Unit) {
+        val mainLooper = Looper.getMainLooper() ?: error("Main looper is not initialized")
+        val runInline =
+            synchronized(transitionLock) {
+                if (!started.compareAndSet(from, to)) {
+                    return
+                }
+                val inline = Looper.myLooper() === mainLooper && pendingOnMain.get() == 0
+                if (!inline) {
+                    postToMainLooper(mainLooper, block)
+                }
+                inline
+            }
+        if (runInline) {
+            block()
+        }
+    }
+
+    private fun postToMainLooper(mainLooper: Looper, block: () -> Unit) {
+        pendingOnMain.incrementAndGet()
+        val posted =
+            Handler(mainLooper).post {
+                try {
+                    block()
+                } finally {
+                    pendingOnMain.decrementAndGet()
+                }
+            }
+        if (!posted) {
+            pendingOnMain.decrementAndGet()
+            error("Main looper is no longer accepting messages")
         }
-        runOnMainLooper { lifecycle.removeObserver(this) }.getOrThrow()
     }
 
     override fun onResume(owner: LifecycleOwner) {

stream-android-core/src/test/java/io/getstream/android/core/api/observers/lifecycle/StreamLifecycleMonitorTest.kt

@@ -27,12 +27,14 @@ import io.getstream.android.core.api.subscribe.StreamSubscriptionManager
 import io.getstream.android.core.api.subscribe.StreamSubscriptionManager.Options
 import io.getstream.android.core.api.subscribe.StreamSubscriptionManager.Options.Retention
 import io.getstream.android.core.testing.TestLogger
+import java.util.concurrent.CopyOnWriteArrayList
 import java.util.concurrent.CountDownLatch
 import java.util.concurrent.TimeUnit
 import java.util.concurrent.atomic.AtomicReference
 import kotlin.concurrent.thread
 import kotlin.test.Test
 import kotlin.test.assertEquals
+import kotlin.test.assertNull
 import kotlin.test.assertTrue
 import org.junit.runner.RunWith
 import org.robolectric.RobolectricTestRunner
@@ -145,10 +147,9 @@ class StreamLifecycleMonitorTest {
             completed.countDown()
         }
 
-        // Continuously process main looper tasks while waiting for thread to complete
-        while (!completed.await(10, TimeUnit.MILLISECONDS)) {
-            shadowLooper.idle()
-        }
+        // start() queues the registration and returns; it must not wait on the main looper.
+        assertTrue(completed.await(5, TimeUnit.SECONDS))
+        shadowLooper.idle()
 
         assertEquals(mainLooper.thread, owner.addObserverThread.get())
 
@@ -173,10 +174,9 @@ class StreamLifecycleMonitorTest {
             completed.countDown()
         }
 
-        // Continuously process main looper tasks while waiting for thread to complete
-        while (!completed.await(10, TimeUnit.MILLISECONDS)) {
-            shadowLooper.idle()
-        }
+        // stop() queues the removal and returns; it must not wait on the main looper.
+        assertTrue(completed.await(5, TimeUnit.SECONDS))
+        shadowLooper.idle()
 
         assertEquals(mainLooper.thread, owner.removeObserverThread.get())
     }
@@ -196,10 +196,8 @@ class StreamLifecycleMonitorTest {
                 startCompleted.countDown()
             }
 
-            // Continuously process main looper tasks while waiting for thread to complete
-            while (!startCompleted.await(10, TimeUnit.MILLISECONDS)) {
-                shadowLooper.idle()
-            }
+            assertTrue(startCompleted.await(5, TimeUnit.SECONDS))
+            shadowLooper.idle()
 
             assertEquals(
                 mainLooper.thread,
@@ -213,10 +211,8 @@ class StreamLifecycleMonitorTest {
                 stopCompleted.countDown()
             }
 
-            // Continuously process main looper tasks while waiting for thread to complete
-            while (!stopCompleted.await(10, TimeUnit.MILLISECONDS)) {
-                shadowLooper.idle()
-            }
+            assertTrue(stopCompleted.await(5, TimeUnit.SECONDS))
+            shadowLooper.idle()
 
             assertEquals(
                 mainLooper.thread,
@@ -226,6 +222,99 @@ class StreamLifecycleMonitorTest {
         }
     }
 
+    // Regression for AND-1468: start() used to block the caller on a five second latch while
+    // waiting for the main looper to run the registration. A caller that is itself holding the
+    // main looper can never release it, so the wait could only ever end in the timeout. Here
+    // the looper is simply never idled, which has the same shape.
+    @Test
+    fun `start returns without waiting for the main looper to run`() {
+        val owner = RecordingLifecycleOwner()
+        val monitor = StreamLifecycleMonitor(TestLogger, owner.lifecycle, newSubscriptionManager())
+        val completed = CountDownLatch(1)
+        val mainLooper = Looper.getMainLooper()
+
+        thread(start = true, name = "StreamLifecycleMonitorTest-unblocked") {
+            monitor.start().getOrThrow()
+            completed.countDown()
+        }
+
+        assertTrue(completed.await(2, TimeUnit.SECONDS), "start must not wait on the main looper")
+        assertNull(owner.addObserverThread.get(), "the registration is queued, not yet run")
+
+        Shadows.shadowOf(mainLooper).idle()
+
+        assertEquals(mainLooper.thread, owner.addObserverThread.get())
+    }
+
+    // A registration that is already queued must not be overtaken by a later call that happens
+    // to run on the main thread, or the observer ends up in the opposite state to `started`.
+    @Test
+    fun `a main thread call does not overtake a registration queued from another thread`() {
+        val owner = RecordingLifecycleOwner()
+        val monitor = StreamLifecycleMonitor(TestLogger, owner.lifecycle, newSubscriptionManager())
+        val started = CountDownLatch(1)
+        val shadowLooper = Shadows.shadowOf(Looper.getMainLooper())
+
+        thread(start = true, name = "StreamLifecycleMonitorTest-attach") {
+            monitor.start().getOrThrow()
+            started.countDown()
+        }
+        assertTrue(started.await(5, TimeUnit.SECONDS))
+
+        // The attach is still queued, so this stop() has to queue behind it rather than run
+        // inline on the main thread.
+        monitor.stop().getOrThrow()
+        shadowLooper.idle()
+
+        assertEquals(listOf("add", "remove"), owner.calls.toList())
+    }
+
+    // The attach lands after start() returns, so a listener subscribed in between — which is what
+    // StreamNetworkAndLifecycleMonitorImpl.start() does — is already in place for the ON_RESUME
+    // that LifecycleRegistry replays to a newly added observer. Nothing acts on that replay: the
+    // recovery evaluator needs an earlier successful connection before it will reconnect, covered
+    // by StreamConnectionRecoveryEvaluatorImplTest. The delivery itself is behaviour now, so pin it
+    // here rather than leave it to be rediscovered.
+    @Test
+    fun `a listener subscribed before start receives the replayed foreground`() {
+        val owner = TestLifecycleOwner()
+        owner.registry.handleLifecycleEvent(Lifecycle.Event.ON_CREATE)
+        owner.registry.handleLifecycleEvent(Lifecycle.Event.ON_START)
+        owner.registry.handleLifecycleEvent(Lifecycle.Event.ON_RESUME)
+
+        val monitor = StreamLifecycleMonitor(TestLogger, owner.lifecycle, newSubscriptionManager())
+        val received = CopyOnWriteArrayList<String>()
+        val listener =
+            object : StreamLifecycleListener {
+                override fun onForeground() {
+                    received += "fg"
+                }
+
+                override fun onBackground() {
+                    received += "bg"
+                }
+            }
+        val subscription =
+            monitor
+                .subscribe(listener, Options(retention = Retention.KEEP_UNTIL_CANCELLED))
+                .getOrThrow()
+
+        val completed = CountDownLatch(1)
+        thread(start = true, name = "StreamLifecycleMonitorTest-replay") {
+            monitor.start().getOrThrow()
+            completed.countDown()
+        }
+        assertTrue(completed.await(5, TimeUnit.SECONDS))
+
+        assertEquals(emptyList(), received.toList(), "the attach is still queued")
+
+        Shadows.shadowOf(Looper.getMainLooper()).idle()
+
+        assertEquals(listOf("fg"), received.toList())
+
+        subscription.cancel()
+    }
+
     private fun newSubscriptionManager(): StreamSubscriptionManager<StreamLifecycleListener> =
         StreamSubscriptionManager(TestLogger)
 
@@ -240,16 +329,19 @@ class StreamLifecycleMonitorTest {
         private val registry = LifecycleRegistry(this)
         val addObserverThread = AtomicReference<Thread?>()
         val removeObserverThread = AtomicReference<Thread?>()
+        val calls = CopyOnWriteArrayList<String>()
 
         override val lifecycle: Lifecycle =
             object : Lifecycle() {
                 override fun addObserver(observer: LifecycleObserver) {
                     addObserverThread.set(Thread.currentThread())
+                    calls.add("add")
                     registry.addObserver(observer)
                 }
 
                 override fun removeObserver(observer: LifecycleObserver) {
                     removeObserverThread.set(Thread.currentThread())
+                    calls.add("remove")
                     registry.removeObserver(observer)
                 }
 

stream-android-core/src/test/java/io/getstream/android/core/api/utils/ThreadingTest.kt

@@ -17,17 +17,20 @@
 package io.getstream.android.core.api.utils
 
 import android.os.Build
+import android.os.Handler
 import android.os.HandlerThread
 import android.os.Looper
 import java.util.concurrent.CountDownLatch
 import java.util.concurrent.TimeUnit
+import java.util.concurrent.atomic.AtomicBoolean
 import java.util.concurrent.atomic.AtomicInteger
 import java.util.concurrent.atomic.AtomicReference
 import kotlin.concurrent.thread
 import kotlin.coroutines.cancellation.CancellationException
 import kotlin.test.Test
 import kotlin.test.assertEquals
 import kotlin.test.assertFailsWith
+import kotlin.test.assertFalse
 import kotlin.test.assertNotNull
 import kotlin.test.assertNull
 import kotlin.test.assertSame
@@ -450,6 +453,36 @@ class ThreadingTest {
         assertTrue(exception.message?.contains("Timed out") == true)
     }
 
+    @Test
+    fun `runOn drops the pending message when it times out`() {
+        val handlerThread = HandlerThread("ThreadingTest-busy").apply { start() }
+        try {
+            val looper = handlerThread.looper
+            val release = CountDownLatch(1)
+            val executed = AtomicBoolean(false)
+
+            // Occupy the looper for longer than runOn's five second bound.
+            Handler(looper).post { release.await(30, TimeUnit.SECONDS) }
+
+            val exception =
+                assertFailsWith<IllegalStateException> {
+                    runOn(looper) { executed.set(true) }.getOrThrow()
+                }
+            assertTrue(exception.message?.contains("Timed out") == true)
+
+            // Free the looper and let anything still queued run. The drain message is posted
+            // after the timed-out one, so it cannot overtake it.
+            release.countDown()
+            val drained = CountDownLatch(1)
+            Handler(looper).post { drained.countDown() }
+            assertTrue(drained.await(5, TimeUnit.SECONDS))
+
+            assertFalse(executed.get(), "a timed-out block must not run afterwards")
+        } finally {
+            handlerThread.quitSafely()
+        }
+    }
+
     @Test
     fun `runOn handles concurrent calls to same looper`() {
         val handlerThread = HandlerThread("ThreadingTest-concurrent").apply { start() }