Fix memory leak caused by capturing the user's coroutine context when…

app/build.gradle

@@ -73,4 +73,6 @@ dependencies {
 
     implementation libraries.coroutinesCore
     implementation libraries.coroutinesAndroid
+
+    debugImplementation libraries.leakcanary
 }

buildsystem/dependencies.gradle

@@ -36,6 +36,7 @@ ext.versions = [
         constraintLayout  : '1.1.3',
         rx2               : '2.2.19',
         rx3               : '3.0.3',
+        leakcanary        : '2.4',
 
         // Testing.
         junit             : '4.13',
@@ -71,6 +72,7 @@ ext.libraries = [
         coreKtx               : "androidx.core:core-ktx:$versions.coreKtx",
         rx2                   : "io.reactivex.rxjava2:rxjava:$versions.rx2",
         rx3                   : "io.reactivex.rxjava3:rxjava:$versions.rx3",
+        leakcanary            : "com.squareup.leakcanary:leakcanary-android:$versions.leakcanary",
 
         // Testing.
         junit                 : "junit:junit:$versions.junit",

store/src/main/java/com/dropbox/android/external/store4/impl/FetcherController.kt

@@ -24,6 +24,7 @@ import com.dropbox.flow.multicast.Multicaster
 import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.ExperimentalCoroutinesApi
 import kotlinx.coroutines.FlowPreview
+import kotlinx.coroutines.async
 import kotlinx.coroutines.flow.Flow
 import kotlinx.coroutines.flow.emitAll
 import kotlinx.coroutines.flow.flow
@@ -97,7 +98,7 @@ internal class FetcherController<Key : Any, Input : Any, Output : Any>(
 
     fun getFetcher(key: Key, piggybackOnly: Boolean = false): Flow<StoreResponse<Input>> {
         return flow {
-            val fetcher = fetchers.acquire(key)
+            val fetcher = acquireFetcher(key)
             try {
                 emitAll(fetcher.newDownstream(piggybackOnly))
             } finally {
@@ -106,6 +107,23 @@ internal class FetcherController<Key : Any, Input : Any, Output : Any>(
         }
     }
 
+    /**
+     * This functions goes to great length to prevent capturing the calling context from
+     * [getFetcher]. The reason being that the [Flow] returned by [getFetcher] is collected on the
+     * user's context and [acquireFetcher] will, optionally, launch a long running coroutine on the
+     * [FetcherController]'s [scope]. In order to avoid capturing a reference to the scope we need
+     * to:
+     * 1) Not inline this function as that will cause the lambda to capture a reference to the
+     * surrounding suspend lambda which, in turn, holds a reference to the user's coroutine context.
+     * 2) Use [async]-[await] instead of
+     * [kotlinx.coroutines.withContext] as [kotlinx.coroutines.withContext] will also hold onto a
+     * reference to the caller's context (the LHS parameter of the new context which is used to run
+     * the operation).
+     */
+    private suspend fun acquireFetcher(key: Key) = scope.async {
+        fetchers.acquire(key)
+    }.await()
+
     // visible for testing
     internal suspend fun fetcherSize() = fetchers.size()
 }