Incorrect termination of a cancelled stream

fs2 version: 2.4.6

I'm running N tasks (the task is uncancelable) in parallel and fs2 does not wait for the completion of the tasks after interruption.

Since the task is uncancelable I'm expecting the following behavior:

1. Interurruption signal appeared
2. Stop peeking pending elements from the inner streams
3. Complete processing of all elements that were peeked before the interruption signal appeared
4. Discard extra results (after post-interruption signal)

In fact, the post-termination in-flight tasks are not chained properly into the cancelation flow and being completed somewhere in the thread pool. Therefore, acquired resources being released earlier than they should.

Expected behavior: since the tasks are uncancelable, the cancelation needs to wait until the in-flight tasks are done.

The code scastie:

import cats.effect.{ExitCase, ExitCode, Resource, IO, IOApp}
import cats.effect.concurrent.Ref
import cats.syntax.functor._
import cats.syntax.flatMap._
import fs2.Stream

import scala.concurrent.duration._

object Main extends IOApp {

  def run(args: List[String]): IO[ExitCode] = {

    def log(msg: String) =
      IO.delay(println(msg))

    def job(idx: Int, isOpenRef: Ref[IO, Boolean]): IO[Int] = {
      val io = 
        for {
          isOpen <- isOpenRef.get
          _      <- log(s"Started $idx. $isOpen")
          _      <- IO.sleep(100.millis)
        } yield idx
  
      io
        .guaranteeCase {
          case ExitCase.Completed => isOpenRef.get.flatMap(isOpen => log(s"Completed $idx. $isOpen"))
          case ExitCase.Error(e)  => isOpenRef.get.flatMap(isOpen => log(s"Error $idx $e. $isOpen"))
          case ExitCase.Canceled  => isOpenRef.get.flatMap(isOpen => log(s"Cancelled $idx. $isOpen"))
        }
        .uncancelable
    }
    
    val isOpenResource = Resource.make(Ref.of[IO, Boolean](true))(bool => bool.set(false))

   
    val process = isOpenResource.use { isOpenRef =>
      Stream.range(1, 100).covary[IO]
      .parEvalMapUnordered(10)(idx => job(idx, isOpenRef))
      .takeWhile(result => result <= 50, takeFailure = true)
      .takeRight(1) // getting the first value (which finishes the stream)
      .compile
      .last
      .flatMap(lastOpt => isOpenRef.get.flatMap(isOpen => log(s"Last $lastOpt. $isOpen")).as(ExitCode.Success))
    }
    
    
    process.start >> IO.never
  }
  
}

The output:

Completed 60. true
Started 62. true
Last Some(51). true
Completed 61. false <- the resource was released, but there are still some in-flight tasks in progress
Completed 62. false

The expected output:

Completed 60. true
Started 62. true
Completed 61. true
Completed 62. true
Last Some(51). true

fs2 3.0.0-M6 version works correctly https://scastie.scala-lang.org/HQq6v4ESSnGWY4KI2W3VxA.

Start with the parEvalMapUnordered cancellation behavior in fs2 2.4.6 and run the linked Scastie reproduction. Compare it with fs2 3.0.0-M6, which reportedly works correctly. Done means uncancelable in-flight tasks finish before the resource is released, matching the expected output.