A borrowed resource can postpone its termination

[iRevive]

keypool version: 0.4.7

When a resource is borrowed from the pool, it's not a part of the PoolMap anymore. Hence it will not be destroyed during the finalization of the pool.
The resource will be open until it's not completed itself. Thus IO.never as a part of the operation with the resource can hang the whole system.

Example:

sealed trait Key
case object Key extends Key

class Connection(val isOpen: Ref[IO, Boolean])

val pool = KeyPool
  .Builder(
    (_: Key) => IO.ref(true).map(new Connection(_)),
    (conn: Connection) => conn.isOpen.set(false)
  )
  .doOnCreate(conn => IO.println(s"created $conn"))
  .doOnDestroy(conn => IO.println(s"destroyed $conn"))
  .build
  .onFinalize(IO.println("pool was closed"))
  
def example(outerAwait: Deferred[IO, Unit]): IO[FiberIO[(Boolean, Boolean)]] =
  pool.use { p =>
    def job(innerAwait: Deferred[IO, Unit]): IO[(Boolean, Boolean)] =
      p.take(Key).use { managed =>
        val connection = managed.value
        
        for {
          status1 <- connection.isOpen.get
          _       <- IO.println("awaiting inner")
          _       <- innerAwait.complete(())
          _       <- IO.println("awaiting outer")
          _       <- outerAwait.get          // wait for the signal from the outside
          _       <- IO.println("after awaits")
          status2 <- connection.isOpen.get   // isOpen should be 'false' there
          // _ <- IO.never[Unit] // once uncommented the program will hang
        } yield (status1, status2)
      }
      
    for {
      await <- IO.deferred[Unit]
      fiber <- job(await).start
      _     <- await.get // make sure first part happens inside of the open pool
    } yield fiber
  }
  
for {
  await   <- IO.deferred[Unit]
  fiber   <- example(await)
  _       <- await.complete(()) // pool is closed and we continue execution of the job
  outcome <- fiber.join
  (status1, status2) <- outcome.embedNever
  _                  <- IO.println(s"(pool open): is connection open $status1")
  _                  <- IO.println(s"(pool closed): is connection open $status2")
} yield ()

Output:

created Connection@76426537
awaiting inner
awaiting outer
pool was closed
after awaits
destroyed Connection@76426537
(pool open): is connection open true
(pool closed): is connection open true

Expected output:

created Connection@76426537
awaiting inner
awaiting outer
destroyed Connection@76426537
pool was closed
after awaits
(pool open): is connection open true
(pool closed): is connection open false