From 69ae37f33e08d4c6b3a88990af6a101ddf6ae711 Mon Sep 17 00:00:00 2001 From: Aleksandar Apostolov Date: Mon, 31 Aug 2026 14:46:17 +0200 Subject: [PATCH 1/4] fix(core): drop the pending message when runOn times out The posted Runnable stayed queued after the five second wait expired, so it ran once the looper freed up, long after the caller had already failed and moved on. Keep a reference to the message and remove it on timeout. --- .../android/core/api/utils/Threading.kt | 8 ++++- .../android/core/api/utils/ThreadingTest.kt | 33 +++++++++++++++++++ 2 files changed, 40 insertions(+), 1 deletion(-) diff --git a/stream-android-core/src/main/java/io/getstream/android/core/api/utils/Threading.kt b/stream-android-core/src/main/java/io/getstream/android/core/api/utils/Threading.kt index 2f9ae5f7..15e51324 100644 --- a/stream-android-core/src/main/java/io/getstream/android/core/api/utils/Threading.kt +++ b/stream-android-core/src/main/java/io/getstream/android/core/api/utils/Threading.kt @@ -163,7 +163,8 @@ public inline fun runOn(looper: Looper, crossinline block: () -> T): Result< } else { val latch = CountDownLatch(1) var result: Result? = 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 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() diff --git a/stream-android-core/src/test/java/io/getstream/android/core/api/utils/ThreadingTest.kt b/stream-android-core/src/test/java/io/getstream/android/core/api/utils/ThreadingTest.kt index 5471c22e..26925f16 100644 --- a/stream-android-core/src/test/java/io/getstream/android/core/api/utils/ThreadingTest.kt +++ b/stream-android-core/src/test/java/io/getstream/android/core/api/utils/ThreadingTest.kt @@ -17,10 +17,12 @@ 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 @@ -28,6 +30,7 @@ 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 { + 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() } From 7706f517964cba627898a958d95719ae76a8d76a Mon Sep 17 00:00:00 2001 From: Aleksandar Apostolov Date: Mon, 31 Aug 2026 14:46:17 +0200 Subject: [PATCH 2/4] fix(core): stop blocking the caller on the lifecycle main-looper hop StreamLifecycleMonitorImpl bridged addObserver/removeObserver with a blocking main-looper round trip. A caller that already occupies the main looper can never release it, so the wait could only end in the five second timeout. Post the attach/detach without awaiting completion, and run inline only when the caller is on the main looper with nothing of ours already queued -- an inline call behind a queued one would invert the two. --- .../lifecycle/StreamLifecycleMonitorImpl.kt | 44 ++++++++++- .../lifecycle/StreamLifecycleMonitorTest.kt | 78 +++++++++++++++---- 2 files changed, 103 insertions(+), 19 deletions(-) diff --git a/stream-android-core/src/main/java/io/getstream/android/core/internal/observers/lifecycle/StreamLifecycleMonitorImpl.kt b/stream-android-core/src/main/java/io/getstream/android/core/internal/observers/lifecycle/StreamLifecycleMonitorImpl.kt index 34c0d760..e0440f60 100644 --- a/stream-android-core/src/main/java/io/getstream/android/core/internal/observers/lifecycle/StreamLifecycleMonitorImpl.kt +++ b/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,6 +39,9 @@ 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) + override fun subscribe( listener: StreamLifecycleListener, options: StreamSubscriptionManager.Options, @@ -46,14 +51,47 @@ internal class StreamLifecycleMonitorImpl( if (!started.compareAndSet(false, true)) { return@runCatching } - runOnMainLooper { lifecycle.addObserver(this) }.getOrThrow() + onMainLooper { lifecycle.addObserver(this) } } override fun stop(): Result = runCatching { if (!started.compareAndSet(true, false)) { return@runCatching } - runOnMainLooper { lifecycle.removeObserver(this) }.getOrThrow() + onMainLooper { lifecycle.removeObserver(this) } + } + + /** + * 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. + */ + private fun onMainLooper(block: () -> Unit) { + val mainLooper = Looper.getMainLooper() ?: error("Main looper is not initialized") + if (Looper.myLooper() === mainLooper && pendingOnMain.get() == 0) { + block() + return + } + pendingOnMain.incrementAndGet() + val posted = + Handler(mainLooper).post { + try { + block() + } finally { + pendingOnMain.decrementAndGet() + } + } + if (!posted) { + pendingOnMain.decrementAndGet() + error("Main looper is no longer accepting messages") + } } override fun onResume(owner: LifecycleOwner) { diff --git a/stream-android-core/src/test/java/io/getstream/android/core/api/observers/lifecycle/StreamLifecycleMonitorTest.kt b/stream-android-core/src/test/java/io/getstream/android/core/api/observers/lifecycle/StreamLifecycleMonitorTest.kt index db3e6371..4798fd40 100644 --- a/stream-android-core/src/test/java/io/getstream/android/core/api/observers/lifecycle/StreamLifecycleMonitorTest.kt +++ b/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,53 @@ 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()) + } + private fun newSubscriptionManager(): StreamSubscriptionManager = StreamSubscriptionManager(TestLogger) @@ -240,16 +283,19 @@ class StreamLifecycleMonitorTest { private val registry = LifecycleRegistry(this) val addObserverThread = AtomicReference() val removeObserverThread = AtomicReference() + val calls = CopyOnWriteArrayList() 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) } From b58c434053143173634ad67bc142b0a249c88147 Mon Sep 17 00:00:00 2001 From: Aleksandar Apostolov Date: Thu, 3 Sep 2026 15:19:44 +0200 Subject: [PATCH 3/4] fix(core): serialise the started flip with the main-looper dispatch The compareAndSet on `started` and the inline-or-post decision were separate steps. A background start() that won the flip but had not yet incremented pendingOnMain looked like nothing was queued, so a stop() arriving on the main thread in that window detached inline and the attach landed after it, leaving the observer attached with started == false. Take both under one lock. The block itself still runs outside it: the inline path holds the main thread, so nothing of ours can overtake it there, and the lock stays clear of the listener callbacks the attach fans out to. --- .../lifecycle/StreamLifecycleMonitorImpl.kt | 42 +++++++++++++------ 1 file changed, 30 insertions(+), 12 deletions(-) diff --git a/stream-android-core/src/main/java/io/getstream/android/core/internal/observers/lifecycle/StreamLifecycleMonitorImpl.kt b/stream-android-core/src/main/java/io/getstream/android/core/internal/observers/lifecycle/StreamLifecycleMonitorImpl.kt index e0440f60..0954c785 100644 --- a/stream-android-core/src/main/java/io/getstream/android/core/internal/observers/lifecycle/StreamLifecycleMonitorImpl.kt +++ b/stream-android-core/src/main/java/io/getstream/android/core/internal/observers/lifecycle/StreamLifecycleMonitorImpl.kt @@ -42,27 +42,25 @@ internal class StreamLifecycleMonitorImpl( /** 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 = subscriptionManager.subscribe(listener, options) override fun start(): Result = runCatching { - if (!started.compareAndSet(false, true)) { - return@runCatching - } - onMainLooper { lifecycle.addObserver(this) } + transition(from = false, to = true) { lifecycle.addObserver(this) } } override fun stop(): Result = runCatching { - if (!started.compareAndSet(true, false)) { - return@runCatching - } - onMainLooper { lifecycle.removeObserver(this) } + transition(from = true, to = false) { lifecycle.removeObserver(this) } } /** - * Runs [block] on the main looper without waiting for it to complete. + * 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 @@ -72,13 +70,33 @@ internal class StreamLifecycleMonitorImpl( * 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 onMainLooper(block: () -> Unit) { + private fun transition(from: Boolean, to: Boolean, block: () -> Unit) { val mainLooper = Looper.getMainLooper() ?: error("Main looper is not initialized") - if (Looper.myLooper() === mainLooper && pendingOnMain.get() == 0) { + 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() - return } + } + + private fun postToMainLooper(mainLooper: Looper, block: () -> Unit) { pendingOnMain.incrementAndGet() val posted = Handler(mainLooper).post { From b028013664ca38e20cf04858e916bb0b327f329f Mon Sep 17 00:00:00 2001 From: Aleksandar Apostolov Date: Thu, 3 Sep 2026 15:19:44 +0200 Subject: [PATCH 4/4] test(core): cover the foreground replayed after start returns The attach now lands after start() returns, so a listener subscribed in between is already in place for the ON_RESUME that LifecycleRegistry replays to a new observer. Nothing acts on it -- the recovery evaluator needs an earlier successful connection -- but the delivery is behaviour now, so pin it. --- .../lifecycle/StreamLifecycleMonitorTest.kt | 46 +++++++++++++++++++ 1 file changed, 46 insertions(+) diff --git a/stream-android-core/src/test/java/io/getstream/android/core/api/observers/lifecycle/StreamLifecycleMonitorTest.kt b/stream-android-core/src/test/java/io/getstream/android/core/api/observers/lifecycle/StreamLifecycleMonitorTest.kt index 4798fd40..ca271fae 100644 --- a/stream-android-core/src/test/java/io/getstream/android/core/api/observers/lifecycle/StreamLifecycleMonitorTest.kt +++ b/stream-android-core/src/test/java/io/getstream/android/core/api/observers/lifecycle/StreamLifecycleMonitorTest.kt @@ -269,6 +269,52 @@ class StreamLifecycleMonitorTest { 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() + 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 = StreamSubscriptionManager(TestLogger)