let tests own the worker coroutine scopes

The repositories and the playback controller each created a hidden SupervisorJob scope and exposed close() only so tests could join it before closing the DB; production never called close(). Inject the scope instead and have each test cancel and join the scope it passes, so no test-only teardown lives in the app. Koin now provides the app/io/playback scopes.

Also read upcomingItem under the controller lock, and give the wall-clock test waits one shared 30s budget instead of 5s per file: a loaded runner could starve the controller's single thread past the old deadline.

app/src/main/kotlin/com/subtracks/data/download/DownloadNotifier.kt

@@ -13,11 +13,8 @@ import com.subtracks.R
 import com.subtracks.data.repo.DownloadRepository
 import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.Dispatchers
-import kotlinx.coroutines.Job
 import kotlinx.coroutines.SupervisorJob
-import kotlinx.coroutines.cancel
 import kotlinx.coroutines.launch
-import kotlinx.coroutines.runBlocking
 
 /**
  * Our own aggregated progress notification for the downloads the platform engine runs. The engine's
@@ -27,8 +24,8 @@ import kotlinx.coroutines.runBlocking
 class DownloadNotifier(
     private val context: Context,
     private val repository: DownloadRepository,
+    private val scope: CoroutineScope = CoroutineScope(SupervisorJob() + Dispatchers.Default),
 ) {
-    private val scope = CoroutineScope(SupervisorJob() + Dispatchers.Default)
     private val manager = NotificationManagerCompat.from(context)
 
     fun start() {
@@ -45,12 +42,6 @@ class DownloadNotifier(
         }
     }
 
-    internal fun close() {
-        val job = scope.coroutineContext[Job]
-        scope.cancel()
-        runBlocking { job?.join() }
-    }
-
     private fun build(content: DownloadNotification): Notification =
         NotificationCompat
             .Builder(context, CHANNEL_ID)

app/src/main/kotlin/com/subtracks/data/repo/DownloadRepository.kt

@@ -24,7 +24,6 @@ import kotlinx.coroutines.Dispatchers
 import kotlinx.coroutines.ExperimentalCoroutinesApi
 import kotlinx.coroutines.Job
 import kotlinx.coroutines.SupervisorJob
-import kotlinx.coroutines.cancel
 import kotlinx.coroutines.delay
 import kotlinx.coroutines.flow.Flow
 import kotlinx.coroutines.flow.MutableStateFlow
@@ -35,7 +34,6 @@ import kotlinx.coroutines.flow.flowOf
 import kotlinx.coroutines.flow.flowOn
 import kotlinx.coroutines.flow.map
 import kotlinx.coroutines.launch
-import kotlinx.coroutines.runBlocking
 import kotlinx.coroutines.sync.Mutex
 import kotlinx.coroutines.sync.withLock
 import kotlinx.coroutines.withContext
@@ -51,8 +49,8 @@ class DownloadRepository(
     private val artworkFetcher: ArtworkFetcher,
     private val showMessage: (String) -> Unit = {},
     private val dispatcher: CoroutineDispatcher = Dispatchers.IO,
+    private val scope: CoroutineScope = CoroutineScope(SupervisorJob() + dispatcher),
 ) {
-    private val scope = CoroutineScope(SupervisorJob() + dispatcher)
     private val mutex = Mutex()
     private val loaded = CompletableDeferred<Unit>()
     private var pollJob: Job? = null
@@ -83,12 +81,6 @@ class DownloadRepository(
         }
     }
 
-    fun close() {
-        val job = scope.coroutineContext[Job]
-        scope.cancel()
-        runBlocking { job?.join() }
-    }
-
     fun states(): StateFlow<Map<String, SongDownload>> = statesFlow
 
     @OptIn(ExperimentalCoroutinesApi::class)

app/src/main/kotlin/com/subtracks/data/repo/LibraryRepository.kt

@@ -34,6 +34,7 @@ class LibraryRepository(
     private val sourceRepository: SourceRepository,
     private val serverActions: ServerActionSink,
     private val showMessage: (String) -> Unit = {},
+    private val scope: CoroutineScope = CoroutineScope(SupervisorJob() + Dispatchers.IO),
 ) {
     val activeSourceId: Flow<Long?> = sourceRepository.activeSourceId()
 
@@ -280,21 +281,20 @@ class LibraryRepository(
         }
 
     private val starLock = Mutex()
-    private val starScope = CoroutineScope(SupervisorJob() + Dispatchers.IO)
 
     fun star(
         type: StarType,
         id: String,
         starred: Boolean,
     ) {
-        starScope.launch { setStar(type, id, starred) }
+        scope.launch { setStar(type, id, starred) }
     }
 
     fun toggleStar(
         type: StarType,
         id: String,
     ) {
-        starScope.launch {
+        scope.launch {
             starLock.withLock {
                 val sourceId = sourceRepository.activeSourceIdOnce() ?: return@withLock
                 setStarLocked(type, id, starredValue(sourceId, type, id) == null)

app/src/main/kotlin/com/subtracks/data/repo/SourceRepository.kt

@@ -16,9 +16,7 @@ import com.subtracks.data.source.streamLengthSuffix
 import com.subtracks.data.source.subsonic.SubsonicClient
 import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.Dispatchers
-import kotlinx.coroutines.Job
 import kotlinx.coroutines.SupervisorJob
-import kotlinx.coroutines.cancel
 import kotlinx.coroutines.flow.Flow
 import kotlinx.coroutines.flow.MutableStateFlow
 import kotlinx.coroutines.flow.StateFlow
@@ -26,7 +24,6 @@ import kotlinx.coroutines.flow.combine
 import kotlinx.coroutines.flow.first
 import kotlinx.coroutines.flow.flowOf
 import kotlinx.coroutines.launch
-import kotlinx.coroutines.runBlocking
 import kotlinx.coroutines.withContext
 import okhttp3.HttpUrl.Companion.toHttpUrl
 import okhttp3.OkHttpClient
@@ -39,9 +36,8 @@ class SourceRepository(
     private val artworkStore: ArtworkStore,
     networkMode: Flow<NetworkMode> = flowOf(NetworkMode.Wifi),
     private val showMessage: (String) -> Unit = {},
+    private val scope: CoroutineScope = CoroutineScope(SupervisorJob() + Dispatchers.Default),
 ) {
-    private val scope = CoroutineScope(SupervisorJob() + Dispatchers.Default)
-
     private data class ActiveSource(
         val id: Long,
         val source: SubsonicMusicSource,
@@ -101,12 +97,6 @@ class SourceRepository(
         }
     }
 
-    fun close() {
-        val job = scope.coroutineContext[Job]
-        scope.cancel()
-        runBlocking { job?.join() }
-    }
-
     fun sources(): Flow<List<Source>> = db.sourcesDao().sources()
 
     fun activeSourceId(): Flow<Long?> = db.sourcesDao().activeSourceId()

app/src/main/kotlin/com/subtracks/data/sync/SyncManager.kt

@@ -7,14 +7,11 @@ import com.subtracks.data.repo.SourceRepository
 import kotlinx.coroutines.CancellationException
 import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.Dispatchers
-import kotlinx.coroutines.Job
 import kotlinx.coroutines.SupervisorJob
-import kotlinx.coroutines.cancel
 import kotlinx.coroutines.channels.Channel
 import kotlinx.coroutines.flow.MutableStateFlow
 import kotlinx.coroutines.flow.StateFlow
 import kotlinx.coroutines.launch
-import kotlinx.coroutines.runBlocking
 
 sealed interface SyncStatus {
     data object Idle : SyncStatus
@@ -33,8 +30,8 @@ class SyncManager(
     private val sourceRepository: SourceRepository,
     private val queueRepository: QueueRepository,
     private val showMessage: (String) -> Unit = {},
+    private val scope: CoroutineScope = CoroutineScope(SupervisorJob() + Dispatchers.Default),
 ) {
-    private val scope = CoroutineScope(SupervisorJob() + Dispatchers.Default)
     private val requests = Channel<Unit>(Channel.CONFLATED)
     private val _status = MutableStateFlow<SyncStatus>(SyncStatus.Idle)
     val status: StateFlow<SyncStatus> = _status
@@ -49,12 +46,6 @@ class SyncManager(
         requests.trySend(Unit)
     }
 
-    fun close() {
-        val job = scope.coroutineContext[Job]
-        scope.cancel()
-        runBlocking { job?.join() }
-    }
-
     private suspend fun runSync() {
         if (sourceRepository.offline.value) {
             showMessage("Offline mode is on")

app/src/main/kotlin/com/subtracks/di/AppModule.kt

@@ -37,8 +37,12 @@ import com.subtracks.ui.library.PlaylistDetailViewModel
 import com.subtracks.ui.playback.QueueViewModel
 import com.subtracks.ui.settings.AddSourceViewModel
 import com.subtracks.ui.settings.SettingsViewModel
+import kotlinx.coroutines.CoroutineScope
+import kotlinx.coroutines.Dispatchers
+import kotlinx.coroutines.SupervisorJob
 import okhttp3.OkHttpClient
 import org.koin.core.module.dsl.viewModel
+import org.koin.core.qualifier.named
 import org.koin.dsl.module
 import java.io.File
 
@@ -53,11 +57,24 @@ fun appModule(
     single { createAndroidDatabase(context) }
     single { http }
     single { createUserPreferences(context) }
-    single { SourceRepository(get(), get(), get(), get(), networkMode = networkMode(context.applicationContext), showMessage = toast) }
+    single { CoroutineScope(SupervisorJob() + Dispatchers.Default) }
+    single(named("io")) { CoroutineScope(SupervisorJob() + Dispatchers.IO) }
+    single(named("playback")) { CoroutineScope(SupervisorJob() + Dispatchers.Main.immediate) }
+    single {
+        SourceRepository(
+            get(),
+            get(),
+            get(),
+            get(),
+            networkMode = networkMode(context.applicationContext),
+            showMessage = toast,
+            scope = get(),
+        )
+    }
     single<ServerActionSink> { NetworkServerActionSink(get()) }
-    single { LibraryRepository(get(), get(), get(), toast) }
+    single { LibraryRepository(get(), get(), get(), toast, scope = get(named("io"))) }
     single<ArtworkSeedStore> { ArtworkSeedRepository(get()) }
-    single { SyncManager(get(), get(), get(), showMessage = toast) }
+    single { SyncManager(get(), get(), get(), showMessage = toast, scope = get()) }
     single { QueueRepository(get()) }
     single { ArtworkStore(downloadsRoot(context)) }
     single<ArtworkFetcher> { OkHttpArtworkFetcher(get()) }
@@ -71,16 +88,19 @@ fun appModule(
             artworkStore = get(),
             artworkFetcher = get(),
             showMessage = toast,
+            dispatcher = Dispatchers.IO,
+            scope = get(named("io")),
         ).also { it.start() }
     }
-    single(createdAtStart = true) { DownloadNotifier(context.applicationContext, get()).also { it.start() } }
+    single(createdAtStart = true) { DownloadNotifier(context.applicationContext, get(), scope = get()).also { it.start() } }
     single<PlayerConnection> { MediaSessionConnection(context.applicationContext, get(), get()) }
-    single { PlaybackController(get(), get(), get(), get(), showMessage = toast) }
+    single { PlaybackController(get(), get(), get(), get(), showMessage = toast, scope = get(named("playback"))) }
     single(createdAtStart = true) {
         val playback = get<PlaybackController>()
         Scrobbler(
             sink = get(),
             enabled = get<UserPreferences>().scrobbling(),
+            scope = get(),
             recordPlay = get<LibraryRepository>()::recordPlay,
         ).also {
             it.attach(playback.state, playback.positionMs)

app/src/main/kotlin/com/subtracks/playback/PlaybackController.kt

@@ -11,12 +11,10 @@ import com.subtracks.data.repo.QueueWindowItem
 import com.subtracks.data.repo.Shuffle
 import com.subtracks.data.repo.SourceRepository
 import kotlinx.coroutines.CancellationException
-import kotlinx.coroutines.CoroutineDispatcher
 import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.Dispatchers
 import kotlinx.coroutines.Job
 import kotlinx.coroutines.SupervisorJob
-import kotlinx.coroutines.cancel
 import kotlinx.coroutines.delay
 import kotlinx.coroutines.flow.MutableStateFlow
 import kotlinx.coroutines.flow.StateFlow
@@ -111,9 +109,8 @@ class PlaybackController(
     private val connection: PlayerConnection,
     private val downloads: DownloadRepository,
     private val showMessage: (String) -> Unit = {},
-    dispatcher: CoroutineDispatcher = Dispatchers.Main.immediate,
+    private val scope: CoroutineScope = CoroutineScope(SupervisorJob() + Dispatchers.Main.immediate),
 ) {
-    private val scope = CoroutineScope(SupervisorJob() + dispatcher)
     private val _state = MutableStateFlow(PlaybackState())
     val state: StateFlow<PlaybackState> = _state
 
@@ -169,14 +166,6 @@ class PlaybackController(
         }
     }
 
-    fun close() {
-        scope.cancel()
-    }
-
-    internal suspend fun awaitStopped() {
-        scope.coroutineContext[Job]?.join()
-    }
-
     private val _ready = MutableStateFlow(false)
     val ready: StateFlow<Boolean> = _ready
 
@@ -879,11 +868,12 @@ class PlaybackController(
         thumbnail: Boolean = false,
     ): CoverArtRef? = sourceRepository.coverArt(item?.coverArtId, thumbnail)
 
-    suspend fun upcomingItem(): QueueItem? {
-        val snap = snapshot ?: return null
-        val position = currentPosition() ?: return null
-        return queueRepository.itemAt(snap, position + 1)?.toQueueItem()
-    }
+    suspend fun upcomingItem(): QueueItem? =
+        startLock.withLock {
+            val snap = snapshot ?: return@withLock null
+            val position = currentPosition() ?: return@withLock null
+            queueRepository.itemAt(snap, position + 1)?.toQueueItem()
+        }
 
     private fun play(
         entries: List<QueueEntry>,