parEvalMap* runs resource finaliser before usage

[nikiforo]

Found by: @lavrov
Original message

Example:

//> using lib "co.fs2::fs2-core::3.2.4"

import cats.effect.*
import fs2.*
import scala.concurrent.duration.*

object Main extends IOApp.Simple:

  def run =
    Stream
      .resource(Resource.make(IO.println("acquire"))(_ => IO.println("release")))
      .parEvalMap(2)(id =>
        IO.sleep(1.second) >> IO.println("use")
      )
      .compile
      .drain

Expected logs:

acquire
use
release

Actual logs:

acquire
release
use

[armanbilge]

@diesalbla identified the issue to here:

fs2/core/shared/src/main/scala/fs2/Stream.scala

Line 2201 in 57032ff

foreground.onFinalize(stop.complete(()) *> end.get).concurrently(background)

The problem is if the background stream (which is derived from the resource-acquiring stream) completes before the foreground stream then it will release the resources.

I am not sure if there is concurrently-like solution, but I was imagining we could do something like:

Stream.resource(background.compile.resource.something).flatMap { _ =>
  foreground ...
}

The precise details are a bit fuzzy :)

[tothpeti]

@armanbilge If possible, I would like to work on this issue.

[tothpeti]

Unfortunately, I haven't been able to make any noticable progression with the ticket since I picked it up. This is why I will unassign myself from it.