Internally managed coroutine scope for pollers (#1991)

SessionHero01 · web-flow · commit bca6f208916a · 2026-02-24T09:00:01.000+11:00
diff --git a/app/src/main/java/org/session/libsession/messaging/sending_receiving/pollers/BasePoller.kt b/app/src/main/java/org/session/libsession/messaging/sending_receiving/pollers/BasePoller.kt
@@ -3,7 +3,10 @@ package org.session.libsession.messaging.sending_receiving.pollers
 import kotlinx.coroutines.CancellationException
 import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.Deferred
+import kotlinx.coroutines.Dispatchers
+import kotlinx.coroutines.SupervisorJob
 import kotlinx.coroutines.async
+import kotlinx.coroutines.cancel
 import kotlinx.coroutines.channels.Channel
 import kotlinx.coroutines.channels.SendChannel
 import kotlinx.coroutines.delay
@@ -39,8 +42,8 @@ abstract class BasePoller<T>(
     private val debugLabel: String,
     private val networkConnectivity: NetworkConnectivity,
     private val appVisibilityManager: AppVisibilityManager,
-    private val scope: CoroutineScope,
 ) {
+    private val scope = CoroutineScope(SupervisorJob() + Dispatchers.Default)
 
     private val manualPollRequestSender: SendChannel<PollRequestCallback<T>>
 
@@ -80,8 +83,8 @@ abstract class BasePoller<T>(
                     pollOnce(pollReason)
                 }.onSuccess { numConsecutiveFailures = 0 }
                     .onFailure {
-                        if (it is CancellationException) {
-                            logE("Polling cancelled", it)
+                        if (it is PollerCancelledException) {
+                            log("Polling cancelled", it)
                             throw it
                         }
                         numConsecutiveFailures += 1
@@ -97,6 +100,12 @@ abstract class BasePoller<T>(
         }
     }
 
+    private class PollerCancelledException : CancellationException("Poller is cancelled")
+
+    fun cancel() {
+        scope.cancel(PollerCancelledException())
+    }
+
     private fun waitForRoutinePoll(minStartAt: TimeMark?): Deferred<Unit> {
         return scope.async {
             combine(
diff --git a/app/src/main/java/org/session/libsession/messaging/sending_receiving/pollers/OpenGroupPoller.kt b/app/src/main/java/org/session/libsession/messaging/sending_receiving/pollers/OpenGroupPoller.kt
@@ -58,11 +58,9 @@ class OpenGroupPoller @AssistedInject constructor(
     appVisibilityManager: AppVisibilityManager,
     private val json: Json,
     @Assisted private val server: String,
-    @Assisted private val scope: CoroutineScope,
     @Assisted private val pollerSemaphore: Semaphore,
 ): BasePoller<Unit>(
     networkConnectivity = networkConnectivity,
-    scope = scope,
     appVisibilityManager = appVisibilityManager,
     debugLabel = "OpenGroupPoller($server)"
 ) {
@@ -336,7 +334,6 @@ class OpenGroupPoller @AssistedInject constructor(
     interface Factory {
         fun create(
             server: String,
-            scope: CoroutineScope,
             pollerSemaphore: Semaphore
         ): OpenGroupPoller
     }
diff --git a/app/src/main/java/org/session/libsession/messaging/sending_receiving/pollers/OpenGroupPollerManager.kt b/app/src/main/java/org/session/libsession/messaging/sending_receiving/pollers/OpenGroupPollerManager.kt
@@ -1,11 +1,8 @@
 package org.session.libsession.messaging.sending_receiving.pollers
 
 import kotlinx.coroutines.CoroutineScope
-import kotlinx.coroutines.Dispatchers
-import kotlinx.coroutines.SupervisorJob
 import kotlinx.coroutines.async
 import kotlinx.coroutines.awaitAll
-import kotlinx.coroutines.cancel
 import kotlinx.coroutines.flow.SharingStarted
 import kotlinx.coroutines.flow.StateFlow
 import kotlinx.coroutines.flow.distinctUntilChanged
@@ -48,7 +45,7 @@ class OpenGroupPollerManager @Inject constructor(
 ) : OnAppStartupComponent {
     private val pollerSemaphore = Semaphore(3)
 
-    val pollers: StateFlow<Map<String, PollerHandle>> =
+    val pollers: StateFlow<Map<String, OpenGroupPoller>> =
         loginStateRepository.flowWithLoggedInState {
             configFactory
                 .userConfigsChanged(onlyConfigTypes = EnumSet.of(UserConfigType.USER_GROUPS))
@@ -61,25 +58,21 @@ class OpenGroupPollerManager @Inject constructor(
                 }
         }
             .distinctUntilChanged()
-            .scan(emptyMap<String, PollerHandle>()) { acc, value ->
+            .scan(emptyMap<String, OpenGroupPoller>()) { acc, value ->
                 if (acc.keys == value) {
                     acc // No change, return the same map
                 } else {
                     val newPollerStates = value.associateWith { baseUrl ->
                         acc[baseUrl] ?: run {
-                            val scope = CoroutineScope(SupervisorJob() + Dispatchers.Default)
                             Log.d(TAG, "Creating new poller for $baseUrl")
-                            PollerHandle(
-                                poller = pollerFactory.create(baseUrl, scope, pollerSemaphore),
-                                pollerScope = scope
-                            )
+                            pollerFactory.create(baseUrl, pollerSemaphore)
                         }
                     }
 
-                    for ((baseUrl, handle) in acc) {
+                    for ((baseUrl, poller) in acc) {
                         if (baseUrl !in value) {
                             Log.d(TAG, "Stopping poller for $baseUrl")
-                            handle.pollerScope.cancel()
+                            poller.cancel()
                         }
                     }
 
@@ -90,23 +83,18 @@ class OpenGroupPollerManager @Inject constructor(
 
     val isAllCaughtUp: Boolean
         get() = pollers.value.values.all {
-            it.poller.pollState.value is BasePoller.PollState.Polled
+            it.pollState.value is BasePoller.PollState.Polled
         }
 
 
     suspend fun pollAllOpenGroupsOnce() {
         Log.d(TAG, "Polling all open groups once")
         supervisorScope {
-            pollers.value.map { (_, handle) ->
+            pollers.value.map { (_, poller) ->
                 async {
-                    handle.poller.manualPollOnce()
+                    poller.manualPollOnce()
                 }
             }.awaitAll()
         }
     }
-
-    data class PollerHandle(
-        val poller: OpenGroupPoller,
-        val pollerScope: CoroutineScope
-    )
 }
diff --git a/app/src/main/java/org/session/libsession/messaging/sending_receiving/pollers/Poller.kt b/app/src/main/java/org/session/libsession/messaging/sending_receiving/pollers/Poller.kt
@@ -1,10 +1,6 @@
 package org.session.libsession.messaging.sending_receiving.pollers
 
-import dagger.assisted.Assisted
-import dagger.assisted.AssistedFactory
-import dagger.assisted.AssistedInject
 import kotlinx.coroutines.CancellationException
-import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.async
 import kotlinx.coroutines.launch
 import kotlinx.coroutines.supervisorScope
@@ -40,10 +36,11 @@ import org.thoughtcrime.securesms.preferences.PreferenceKey
 import org.thoughtcrime.securesms.preferences.PreferenceStorage
 import org.thoughtcrime.securesms.util.AppVisibilityManager
 import org.thoughtcrime.securesms.util.NetworkConnectivity
+import javax.inject.Inject
 import kotlin.time.Duration.Companion.days
 import kotlin.time.Duration.Companion.seconds
 
-class Poller @AssistedInject constructor(
+class Poller @Inject constructor(
     private val configFactory: ConfigFactoryProtocol,
     private val storage: StorageProtocol,
     private val lokiApiDatabase: LokiAPIDatabaseProtocol,
@@ -60,11 +57,9 @@ class Poller @AssistedInject constructor(
     private val swarmDirectory: SwarmDirectory,
     private val snodeApiExecutor: SnodeApiExecutor,
     appVisibilityManager: AppVisibilityManager,
-    @Assisted scope: CoroutineScope
 ) : BasePoller<Unit>(
     debugLabel = "MainPoller",
     networkConnectivity = networkConnectivity,
-    scope = scope,
     appVisibilityManager = appVisibilityManager
 ) {
     private val userPublicKey: String
@@ -76,11 +71,6 @@ class Poller @AssistedInject constructor(
     }
 
 
-    @AssistedFactory
-    interface Factory {
-        fun create(scope: CoroutineScope): Poller
-    }
-
     override suspend fun doPollOnce(isFirstPollSinceAppStarted: Boolean) {
         // Migrate to multipart config when needed
         if (isFirstPollSinceAppStarted && !prefs[hasMigratedToMultiPartConfigKey]) {
diff --git a/app/src/main/java/org/session/libsession/messaging/sending_receiving/pollers/PollerManager.kt b/app/src/main/java/org/session/libsession/messaging/sending_receiving/pollers/PollerManager.kt
@@ -1,12 +1,12 @@
 package org.session.libsession.messaging.sending_receiving.pollers
 
-import kotlinx.coroutines.coroutineScope
 import kotlinx.coroutines.flow.MutableStateFlow
 import kotlinx.coroutines.flow.filterNotNull
 import kotlinx.coroutines.flow.first
 import org.thoughtcrime.securesms.auth.AuthAwareComponent
 import org.thoughtcrime.securesms.auth.LoggedInState
 import javax.inject.Inject
+import javax.inject.Provider
 import javax.inject.Singleton
 
 /**
@@ -17,15 +17,19 @@ import javax.inject.Singleton
  */
 @Singleton
 class PollerManager @Inject constructor(
-    private val provider: Poller.Factory,
+    private val provider: Provider<Poller>,
 ) : AuthAwareComponent {
     private val currentPoller = MutableStateFlow<Poller?>(null)
 
-    override suspend fun doWhileLoggedIn(loggedInState: LoggedInState): Unit = coroutineScope {
-        val poller = provider.create(this)
+    override suspend fun doWhileLoggedIn(loggedInState: LoggedInState) {
+        val poller = provider.get()
         currentPoller.value = poller
     }
 
+    override fun onLoggedOut() {
+        currentPoller.value?.cancel()
+    }
+
 
     val isPolling: Boolean
         get() = currentPoller.value?.pollState?.value is BasePoller.PollState.Polling
diff --git a/app/src/main/java/org/thoughtcrime/securesms/groups/GroupPoller.kt b/app/src/main/java/org/thoughtcrime/securesms/groups/GroupPoller.kt
@@ -3,7 +3,6 @@ package org.thoughtcrime.securesms.groups
 import dagger.assisted.Assisted
 import dagger.assisted.AssistedFactory
 import dagger.assisted.AssistedInject
-import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.Deferred
 import kotlinx.coroutines.async
 import kotlinx.coroutines.awaitAll
@@ -39,7 +38,6 @@ import kotlin.coroutines.cancellation.CancellationException
 import kotlin.time.Duration.Companion.days
 
 class GroupPoller @AssistedInject constructor(
-    @Assisted scope: CoroutineScope,
     @Assisted private val groupId: AccountId,
     @Assisted private val pollSemaphore: Semaphore,
     private val configFactoryProtocol: ConfigFactoryProtocol,
@@ -58,7 +56,6 @@ class GroupPoller @AssistedInject constructor(
 ): BasePoller<GroupPoller.GroupPollResult>(
     networkConnectivity = networkConnectivity,
     appVisibilityManager = appVisibilityManager,
-    scope = scope,
     debugLabel = "GroupPoller(${groupId.truncatedForDisplay()})"
 ) {
     data class GroupPollResult(
@@ -337,6 +334,6 @@ class GroupPoller @AssistedInject constructor(
 
     @AssistedFactory
     interface Factory {
-        fun create(scope: CoroutineScope, groupId: AccountId, pollSemaphore: Semaphore): GroupPoller
+        fun create(groupId: AccountId, pollSemaphore: Semaphore): GroupPoller
     }
 }
diff --git a/app/src/main/java/org/thoughtcrime/securesms/groups/GroupPollerManager.kt b/app/src/main/java/org/thoughtcrime/securesms/groups/GroupPollerManager.kt
@@ -1,12 +1,9 @@
 package org.thoughtcrime.securesms.groups
 
 import kotlinx.coroutines.CoroutineScope
-import kotlinx.coroutines.Dispatchers
 import kotlinx.coroutines.ExperimentalCoroutinesApi
-import kotlinx.coroutines.SupervisorJob
 import kotlinx.coroutines.async
 import kotlinx.coroutines.awaitAll
-import kotlinx.coroutines.cancel
 import kotlinx.coroutines.flow.Flow
 import kotlinx.coroutines.flow.SharingStarted
 import kotlinx.coroutines.flow.StateFlow
@@ -64,7 +61,7 @@ class GroupPollerManager @Inject constructor(
     private val groupPollerSemaphore = Semaphore(20)
 
     @Suppress("OPT_IN_USAGE")
-    private val groupPollers: StateFlow<Map<AccountId, GroupPollerHandle>> =
+    private val groupPollers: StateFlow<Map<AccountId, GroupPoller>> =
         combine(
             connectivity.networkAvailable.debounce(200L),
             loginStateRepository.loggedInState,
@@ -95,12 +92,12 @@ class GroupPollerManager @Inject constructor(
             // This scan compares the previous active group pollers with the incoming set of groups
             // that should be polled now, to work out which pollers should be started or stopped,
             // and finally emits the new state
-            .scan(emptyMap<AccountId, GroupPollerHandle>()) { previous, newActiveGroupIDs ->
+            .scan(emptyMap<AccountId, GroupPoller>()) { previous, newActiveGroupIDs ->
                 // Go through previous pollers and stop those that are not in the new set
                 for ((groupId, poller) in previous) {
                     if (groupId !in newActiveGroupIDs) {
                         Log.d(TAG, "Stopping poller for $groupId")
-                        poller.scope.cancel()
+                        poller.cancel()
                     }
                 }
 
@@ -111,14 +108,9 @@ class GroupPollerManager @Inject constructor(
 
                     if (poller == null) {
                         Log.d(TAG, "Starting poller for $groupId")
-                        val scope = CoroutineScope(SupervisorJob() + Dispatchers.Default)
-                        poller = GroupPollerHandle(
-                            poller = pollFactory.create(
-                                scope = scope,
-                                groupId = groupId,
-                                pollSemaphore = groupPollerSemaphore,
-                            ),
-                            scope = scope
+                        poller = pollFactory.create(
+                            groupId = groupId,
+                            pollSemaphore = groupPollerSemaphore,
                         )
                     }
 
@@ -133,7 +125,7 @@ class GroupPollerManager @Inject constructor(
     fun watchGroupPollingState(groupId: AccountId): Flow<BasePoller.PollState<GroupPoller.GroupPollResult>> {
         return groupPollers
             .flatMapLatest { pollers ->
-                pollers[groupId]?.poller?.pollState ?: flowOf(BasePoller.PollState.Idle)
+                pollers[groupId]?.pollState ?: flowOf(BasePoller.PollState.Idle)
             }
             .distinctUntilChanged()
     }
@@ -145,7 +137,7 @@ class GroupPollerManager @Inject constructor(
                 // Merge all poller states into a single flow of (groupId, state) pairs
                 merge(
                     *pollers
-                        .map { (id, poller) -> poller.poller.pollState.map { state -> id to state } }
+                        .map { (id, poller) -> poller.pollState.map { state -> id to state } }
                         .toTypedArray()
                 )
             }
@@ -155,7 +147,7 @@ class GroupPollerManager @Inject constructor(
         supervisorScope {
             groupPollers.value.values.map {
                 async {
-                    it.poller.manualPollOnce()
+                    it.manualPollOnce()
                 }
             }.awaitAll()
         }
@@ -170,15 +162,9 @@ class GroupPollerManager @Inject constructor(
     suspend fun pollOnce(groupId: AccountId): GroupPoller.GroupPollResult {
         return groupPollers.mapNotNull { it[groupId] }
             .first()
-            .poller
             .manualPollOnce()
     }
 
-    data class GroupPollerHandle(
-        val poller: GroupPoller,
-        val scope: CoroutineScope,
-    )
-
     companion object {
         private const val TAG = "GroupPollerHandler"
     }
