Flow.zip breaks if Flow1 impl catchs & rethrows a CancellationException, but not if Flow2 does it

[sameb]

Describe the bug

Flow.zip breaks if Flow1 impl catchs & rethrows a CancellationException, but not if Flow2 does it.

The problem boils down to the zip impl, where it cancels the Job with an AbortFlowException, but then expects the flow usercode that exception passed through to surface that exact same AbortFlowException. That isn't a fair assumption, which is especially true as the second flow is not held to the same expectation. The implementation should instead check what exception the job was cancelled with. (Note that in doing so, it should be careful to preserve exceptions thrown from the transform lambda, as well as any exceptions thrown from the first coroutine in scenarios where the job wasn't cancelled due to zip's intentional abort.)

Provide a Reproducer

Given the code:

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*
import kotlin.time.Duration.Companion.seconds

fun main() = runBlocking {
    val f1 = flow {
        try {
          emit(1)
          emit(2)
          delay(1.seconds)
          emit(3)
        } catch(ex: CancellationException) {
           //if(true) throw ex // Re-throwing the exception as-is makes things work
           // .. but throwing a _different_ exception breaks the zip impl
           throw InstrumentedCancellationException(ex)  
        }
    }
    val f2 = flow {
        emit("one")
        emit("two")
    }
    
    try {
      val z = f1.zip(f2, { v1, v2 -> v1 to v2 })
      println("f1.zip(f2) == ${z.toList()}")
    } catch (ex: Exception) {
      println("f1.zip(f2) failed")
      ex.printStackTrace()
    }
    
    
    try {
      val z = f2.zip(f1, { v1, v2 -> v1 to v2 })
      println("f2.zip(f1) == ${z.toList()}")
    } catch (ex: Exception) {
      println("f2.zip(f1) failed")
      ex.printStackTrace()
    }
}

class InstrumentedCancellationException(override val cause: Throwable) : CancellationException()

... the f1.zip(f2) fails and an exception is printed out, but f2.zip(f1) succeeds. The stack trace printed out from f1.zip(f2) is:

InstrumentedCancellationException
	at FileKt$main$1$f1$1.invokeSuspend(File.kt:14)
	at kotlin.coroutines.jvm.internal.BaseContinuationImpl.resumeWith(ContinuationImpl.kt:33)
	at kotlinx.coroutines.DispatchedTask.run(DispatchedTask.kt:106)
	at kotlinx.coroutines.EventLoopImplBase.processNextEvent(EventLoop.common.kt:280)
	at kotlinx.coroutines.BlockingCoroutine.joinBlocking(Builders.kt:85)
	at kotlinx.coroutines.BuildersKt__BuildersKt.runBlocking(Builders.kt:59)
	at kotlinx.coroutines.BuildersKt.runBlocking(Unknown Source)
	at kotlinx.coroutines.BuildersKt__BuildersKt.runBlocking$default(Builders.kt:38)
	at kotlinx.coroutines.BuildersKt.runBlocking$default(Unknown Source)
	at FileKt.main(File.kt:5)
	at FileKt.main(File.kt)
	at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
	at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke(Unknown Source)
	at java.base/jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(Unknown Source)
	at java.base/java.lang.reflect.Method.invoke(Unknown Source)
	at executors.JavaRunnerExecutor$Companion.main(JavaRunnerExecutor.kt:27)
	at executors.JavaRunnerExecutor.main(JavaRunnerExecutor.kt)
Caused by: kotlinx.coroutines.flow.internal.AbortFlowException: Flow was aborted, no more elements needed
	at kotlinx.coroutines.flow.internal.CombineKt$zipImpl$1$1$1.invoke(Combine.kt:108)
	at kotlinx.coroutines.flow.internal.CombineKt$zipImpl$1$1$1.invoke(Combine.kt:106)
	at kotlinx.coroutines.channels.BufferedChannel.invokeCloseHandler(BufferedChannel.kt:1823)
	at kotlinx.coroutines.channels.BufferedChannel.closeOrCancelImpl(BufferedChannel.kt:1800)
	at kotlinx.coroutines.channels.BufferedChannel.close(BufferedChannel.kt:1754)
	at kotlinx.coroutines.channels.SendChannel$DefaultImpls.close$default(Channel.kt:98)
	at kotlinx.coroutines.channels.ProducerCoroutine.onCompleted(Produce.kt:143)
	at kotlinx.coroutines.channels.ProducerCoroutine.onCompleted(Produce.kt:136)
	at kotlinx.coroutines.AbstractCoroutine.onCompletionInternal(AbstractCoroutine.kt:93)
	at kotlinx.coroutines.JobSupport.tryFinalizeSimpleState(JobSupport.kt:296)
	at kotlinx.coroutines.JobSupport.tryMakeCompleting(JobSupport.kt:860)
	at kotlinx.coroutines.JobSupport.makeCompletingOnce$kotlinx_coroutines_core(JobSupport.kt:832)
	at kotlinx.coroutines.AbstractCoroutine.resumeWith(AbstractCoroutine.kt:100)
	at kotlin.coroutines.jvm.internal.BaseContinuationImpl.resumeWith(ContinuationImpl.kt:46)
	at kotlinx.coroutines.flow.internal.SafeCollector.invokeSuspend(SafeCollector.kt:48)
	at kotlin.coroutines.jvm.internal.BaseContinuationImpl.resumeWith(ContinuationImpl.kt:33)
	at kotlinx.coroutines.DispatchedTask.run(DispatchedTask.kt:108)
	... 14 more

(This was filed as https://youtrack.jetbrains.com/issue/KT-80239/Flow.zip-breaks-if-Flow1-impl-catchs-rethrows-CancellationException-but-not-if-Flow2-does-it, but a colleague told me that reactive stuff is done outside the mainline kt stdlib and is tracked here.)

[murfel]

Minimized

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*

class InstrumentedCancellationException(override val cause: Throwable) : CancellationException()

fun main() = runBlocking {
    val f1 = flow {
        try {
            emit(1)
            emit(2)
            emit(3)
        } catch (e: CancellationException) {
            throw InstrumentedCancellationException(e);
        }
    }
    val f2 = flow {
        emit("one")
        emit("two")
    }

    println(f2.zip(f1, { v1, v2 -> v1 to v2 }).toList())  // ok

    println(f1.zip(f2, { v1, v2 -> v1 to v2 }).toList())  // throws
}

Output (edited)

[(one, 1), (two, 2)]
Exception in thread "main" InstrumentedCancellationException
  at ...
Caused by: kotlinx.coroutines.flow.internal.AbortFlowException: Flow was aborted, no more elements needed
  at ...

https://pl.kotl.in/D3CcKY748

[murfel]

Hey Sam!

Your report highlights two behaviours:

1. zip behaviour is not symmetric when an instrumented CancellationException is propagated
2. zip completes exceptionally (unexpectedly) if you instument a CancellationException

What's your use case requiring instrumentation of a CancellationException? Do you need to instrument any CancellationException, or only AbortFlowException?

[sameb]

Thanks for taking a look at this, @murfel !

The fundamental problem here is that the flow is arbitrary usercode that has free reign to do whatever it wants. The flow impl can weave through lots of third-party libs, unknown code, etc.. it's unrealistic (and impossible) to tell the user they can't catch and instrument/tweak exceptions.

Our use-case happens to be an infra lib that users must delegate to, and we'd like our infra to rethrow an "instrumented" cancellation exception that happens to have a custom stack trace related to other files. (We also instrument non-cancellation too, but we special-case cancellation to ensure that the rethrown type is also a cancellation).

However: that particular usage is kind of immaterial, because really anything could happen here. Aside from my infra lib, there's also tons of usercode that happens to try/catch(Exception) and rethrow as custom things, etc. While I agree that the particular try/catch(Exception) is a bad practice, I wouldn't go so far as agreeing that doing so should break zip.

Consider also that if the user adds a .buffer(100 /* some arbitrary big number */) on the zipped flow inputs, then zip suddenly starts working... presumably because the buffer is eating the exception and the job-cancel lets the AbortFlowException passthru. The fact that it isn't symmetrical and that buffering changes the behavior makes me inclined to say that this is truly a bug with the zip impl: it shouldn't be reliant on usercode passing the exception through as-is.

[murfel]

What's your bottom line for instrumenting the CancellationException?

[sameb]

Sorry, I don't understand the question. Can you clarify what you mean by "what's your bottom line"?

[murfel]

Why are you instrumenting the CancellationException?

[sameb]

To elaborate on the earlier descriptions: the infra lib in this particular use-case is effectively an async orchestration framework. Users write separate files describing connections between datas, and the infra connects the dots between them. It instruments exceptions so stack traces can point to the correct lines/symbols in the separate files users wrote that drove the work.

But I want to again point out: this is just one use-case, and in reality there's also a significant amount of arbitrary user code that can literally do anything, including try/catch/rethrow arbitrary Exceptions.

It isn't safe for the zip impl to assume a job cancelled with a particular exception instance will see that same exception instance when routed through user code. Other usages of AbortFlowException don't make this assumption (it is thrown and caught from areas directly in their control and can't weave back through user code).

[qwwdfsad]

I was thinking of closing it "as designed", but that's at the very least a documentation issue because I cannot point to a specific place that would describe the current behaviour.

In general, catching CancellationException without rethrowing is error-prone. Flow suffer from that even more as it is partially illegal to wrap CE into other CE. See also: #3658, #4073

When doing so, all bets are off -- and various implementation details start leaking into observable behaviour.

Other usages of AbortFlowException don't make this assumption

Unfortunately, it is not the case.
For example, any operator is not really different from your example:

val flow = flow {
    try {
        emit(1)
        emit(2)
    } catch (e: CancellationException) {
        throw CancellationException("Done", e)
    }
}

flow.any { true } // Will fail if the exception is wrapped, will succeed if the exception is just rethrown

While the other code indeed can do arbitrary things with exceptions (here is a nice example of a discussion that lead us into changing some of the behaviours: #2860), unconditional rewrapping of exceptions poses an unsolvable choice for the library:

• Consider the library has thrown a CancellationException (or AbortFlowException) to abort an upstream (flow)
• The termination sequence of the flow executes an arbitrary code in its catch and finally blocks. I.e. it can do finally { fileFromWhichElementsWereEmited.closeThatMightThrow() }
• These termination sequences indeed can throw an arbitrary exception
• Now, the library has to guess: is a new rethrown exception something that we can safely ignore?
  • If we ignore something that we shouldn't, then we silently hide a failure (potentially a critical one) from the developer
  • If we do not ignore something that we should (like in your case) then it leads to a very annoying and totally unclear exception

So far, we've been doing a safe bet on "let's always rethrow" and I am not really sure if we can do much better here

[qwwdfsad]

Yet while typing that, there is a fresh idea.

We cannot ignore non-CE exceptions that suppress our AbortFlowException.
We likely cannot really ignore CE as well because of this, for example (though it's still an open question #4159):

val flow = flow<Int> {
    throw CancellationException("Done")
}
flow.any { true } //Should definitely fail with CE even if there is a parallelism involved (i.e. buffer() inbetween)

Still, we can statically detect situations when we threw AFE and got the other CE in response. At the very least, we can throw a descriptive exception in such a situation on what's going on and why it is not recommended to wrap AFE

[sameb]

Thanks for this discussion!

Stepping away from the immediate problem at hand... One of the bigger issues here is that KT treats CancellationException as "special" and dislikes wrapping or suppressing it. However, there's no library support for catching everything except CancellationException. Thus, users need to always remember to special-case cancellations, and bugs creep in when they inevitably forget.

The situation would be improved if KT concurrency docs outlined how CEs should be "left alone", and if the stdlibs offered mechanisms for leaving them alone.