StreamSubscriber hangs in some scenarios

import cats.effect.{ ContextShift, IO, BracketThrow }

import monix.eval.Task
import monix.execution.{ ExecutionModel, Scheduler }
import monix.execution.Scheduler.DefaultScheduledExecutor
import monix.execution.schedulers.AsyncScheduler
import monix.reactive.Observable

import fs2.{ Chunk, Stream }
import fs2.interop.reactivestreams.PublisherOps

import scala.concurrent.ExecutionContext

implicit val scheduler: Scheduler =
  AsyncScheduler(DefaultScheduledExecutor, ExecutionContext.global, ExecutionModel.Default)

implicit val contextShift: ContextShift[IO] =
  IO.contextShift(ExecutionContext.global)

val program: IO[Vector[Byte]] =
  Observable
    .fromIterator(Task.async[Iterator[Byte]] { cb => cb(Right(Iterator[Byte](1, 2, 3, 4, 5))) })
    .map(Array(_))
    .map(Chunk.bytes(_))
    .toReactivePublisher[Chunk[Byte]](Scheduler.global)
    .toStream[Task]
    .flatMap(Stream.chunk(_))
    .compile
    .toVector
    .to[IO]

implicitly[BracketThrow[IO]].bracket(IO.unit)(_ => program)(_ => IO.unit).unsafeRunSync()

• FS2 Version 2.4.4 (works) https://scastie.scala-lang.org/o56q3I7RRuSia4wBgEI6SQ
• FS2 Version 2.4.6 (hangs) https://scastie.scala-lang.org/MWy3e0z1T3mK6GjewxvGNg

This commit causes the issue https://github.com/typelevel/fs2/commit/a9ff266ea23c67e9feb43ea8a672242a7bf99e5d. Just in case this is the only change needed to recreate the issue:

// Before
def onSubscribe(s: Subscription): Unit = {
  nonNull(s)
  sub.onSubscribe(s).unsafeRunAsync()
}

// After
def onSubscribe(s: Subscription): Unit = {
  nonNull(s)
  sub.onSubscribe(s).unsafeRunSync()
}

IO Bracket and Task.async are needed in order to recreate the issue. I tried to isolate the problem as much as I could. It seems to be hanging in cats.effect.internals.ForwardCancelable#complete which is used by IO.cancelable and that's the function ConcurrentEffect[F].toIO calls.

Start with the StreamSubscriber onSubscribe path and compare the behavior introduced by commit a9ff266ea23c67e9feb43ea8a672242a7bf99e5d. Use the supplied FS2 2.4.4 and 2.4.6 reproductions, then inspect cats.effect.internals.ForwardCancelable#complete, IO.cancelable, and ConcurrentEffect[F].toIO. Done means the bracketed program completes without hanging on the 2.4.6 path.