kotlinx.coroutines.JobCancellationException: FlowSubscription was cancelled

[mhernand40]

I recently migrated a class that uses https://github.com/f2prateek/rx-preferences to now use FlowPreferences. In our app, when the user logs out, we clear all SharedPreferences. During the logout, the app is crashing with the following exception:

kotlinx.coroutines.JobCancellationException: FlowSubscription was cancelled; job=FlowSubscription{Cancelled}@bed4c69

There isn't really any more details regarding the stack trace. I suspect I am running into Kotlin/kotlinx.coroutines#974 which may make sense given FlowSharedPreferences.keyFlow is implemented as such:

  val keyFlow = callbackFlow {
    val listener = SharedPreferences.OnSharedPreferenceChangeListener { _, key -> offer(key) }
    sharedPreferences.registerOnSharedPreferenceChangeListener(listener)
    awaitClose { sharedPreferences.unregisterOnSharedPreferenceChangeListener(listener) }
  }

Something tells me that the offer(key) is being performed after the Channel has been closed.

[mhernand40]

Might also be related to Kotlin/kotlinx.coroutines#1762.

[tfcporciuncula]

Hey @mhernand40, I missed the notification for this issue, sorry. Do you understand what's different in your logout flow that might be related to the issue? It'd be great if we had a minimal working example to dig deeper.

[mhernand40]

Hi @tfcporciuncula, no worries. I have since reverted back to https://github.com/f2prateek/rx-preferences and then convert the Observable to a Flow to unblock myself.

I believe the issue is that we are subscribed to Streams from both Rx and Flow in our MainActivity. Normally when this activity gets destroyed, we dispose/cancel these streams. However, for the logout scenario, I believe our SharedPreferences get cleared before MainActivity gets destroyed, which is probably something we should fix on our end, but that's a separate issue. For some reason with Flow Preferences, we were running into a JobCancellationException which was crashing our app.

I feel that what's happening is:

1. Flow gets cancelled
2. SharedPreferences.OnSharedPreferenceChangeListener fires event, triggering the offer() which then throws the JobCancellationException
3. For some reason, awaitClose does not run on time and remove the listener.

I still have to come up with a minimal repro to verify what exactly is going wrong.

[mhernand40]

I finally have a minimal repro and have confirmed that the crash still occurs with:

• Kotlin 1.3.70
• Coroutines 1.3.4
• Flow Preferences 1.1.1

The repro is a race condition and it actually involves converting the Flow to an RxJava 2 Observable using the kotlinx-coroutines-rx2 library, and then subscribing to the Observable. Should a SharedPreferences update trigger the Flow emission concurrently while the Disposable gets disposed, the crash occurs.

Please see the attached repro app.
ReproApp.zip

[tfcporciuncula]

Thanks @mhernand40! I'll take a look. But my initial thinking is that this might be outside of the scope of this library since it's an issue that only happens when the Observable conversion is involved. It's certainly still relevant, though, so thanks for reporting and providing an example.

[mhernand40]

I decided that I really only require basic Flow extension functions for SharedPreferences and have decided to build my own in-house solution. FWIW, in my implementation, I wrapped the offer() call within a runCatching {} block to mitigate the risk of crashing of the app due to a JobCancellationException. I borrowed the idea from workarounds suggested in the ongoing conversation of Kotlin/kotlinx.coroutines#974.

[tfcporciuncula]

Thanks for letting me know @mhernand40. I'll keep this issue open since it might be relevant for other people using the library and will close it once there's an actual fix in place.

[tfcporciuncula]

Potential relevant update about this here: Kotlin/kotlinx.coroutines#974 (comment)

[tfcporciuncula]

Now that the underlying coroutines issue is fixed, I'm reverting the fix we have for this in favor of the new trySend() method from 1.5.0. Everything should continue to work just fine, but I thought I'd mention this here just in case.