[BUG] vert.x streaming endpoints hang whole application when VertxCatsServerInterpreter is used with ZIO

Tapir version: 1.13.31

Scala version: 3.7.3

Describe the bug

When a cats-effect Dispatcher for vert.x interpreter is created on a default executor, then sending CPU-count parallel requests to a streaming endpoint (doing unsafeRunSync under the hood) actually blocks every thread of that executor, making the application freeze.

How to reproduce?

//> using scala 3.7.3
//> using dep com.softwaremill.sttp.tapir::tapir-vertx-server:1.13.31
//> using dep com.softwaremill.sttp.tapir::tapir-vertx-server-cats:1.13.31
//> using dep dev.zio::zio-interop-cats:23.1.0.13
//> using dep dev.zio::zio:2.1.26
//> using dep io.vertx:vertx-web:5.1.6

import zio._
import zio.interop.catz._
import cats.effect.std.Dispatcher
import fs2.Stream
import io.vertx.core.Vertx
import io.vertx.core.http.HttpServerOptions
import io.vertx.ext.web.Router
import sttp.capabilities.fs2.Fs2Streams
import sttp.tapir._
import sttp.tapir.server.vertx.cats.{VertxCatsServerInterpreter, VertxCatsServerOptions}

import java.net.URI
import java.net.http.{HttpClient, HttpRequest, HttpResponse}
import java.util.concurrent.Executors

object Repro extends ZIOAppDefault:

  override val bootstrap =
    Runtime.setExecutor(zio.Executor.fromJavaExecutor(Executors.newFixedThreadPool(2)))

  val streamEndpoint =
    endpoint.get.in("stream")
      .out(streamTextBody(Fs2Streams[Task])(CodecFormat.TextPlain(), None))
      .serverLogic[Task](_ => ZIO.sleep(200.millis).as(Right(Stream.emits("hello, world!".getBytes.toIndexedSeq))))

  val run =
    Dispatcher.parallel[Task].use { dispatcher =>
      val options = VertxCatsServerOptions.default[Task](dispatcher)
      val interpreter = VertxCatsServerInterpreter(options)
      val vertx = Vertx.vertx()
      val router = Router.router(vertx)
      interpreter.route(streamEndpoint)(router)

      for {
        server <- ZIO.fromCompletionStage(
          vertx.createHttpServer(new HttpServerOptions().setPort(0)).requestHandler(router).listen(0).toCompletionStage)
        port = server.actualPort()
        _ <- Console.printLine(s"port=$port")
        _ <- ZIO.foreachParDiscard(1 to 4) { i =>
          ZIO.attemptBlocking {
            val client = HttpClient.newHttpClient()
            val req = HttpRequest.newBuilder(URI.create(s"http://127.0.0.1:$port/stream")).GET().build()
            val resp = client.send(req, HttpResponse.BodyHandlers.ofString())
            println(s"req $i status=${resp.statusCode()} body=${resp.body()}")
          }
        }.timeoutFail(new Exception("deadlock"))(20.seconds)
        _ <- Console.printLine("SUCCESS")
      } yield ()
    }

Additional information

jstack output showing both executor threads blocked

"pool-1-thread-1" #62 prio=5 waiting on condition
   java.lang.Thread.State: WAITING (parking)
	at jdk.internal.misc.Unsafe.park(Native Method)
	- parking to wait for  <0x000000061ecd69a0> (a scala.concurrent.impl.CompletionLatch)
	at java.util.concurrent.locks.LockSupport.park(LockSupport.java:221)
	...
	at cats.effect.std.Dispatcher$$anon$2.unsafeRunSync(Dispatcher.scala:287)
	at sttp.tapir.server.vertx.cats.streams.fs2$$anon$1.mapToReadStream(fs2.scala:45)
	at sttp.tapir.server.vertx.cats.streams.fs2$$anon$1.asReadStream(fs2.scala:40)
	at sttp.tapir.server.vertx.encoders.VertxToResponseBody.fromStreamValue$$anonfun$1(VertxToResponseBody.scala:54)
	...

"pool-1-thread-2" #66 prio=5 waiting on condition
   java.lang.Thread.State: WAITING (parking)
	at jdk.internal.misc.Unsafe.park(Native Method)
	- parking to wait for  <0x000000061ed0c368> (a scala.concurrent.impl.CompletionLatch)
	at java.util.concurrent.locks.LockSupport.park(LockSupport.java:221)
	...
	at cats.effect.std.Dispatcher$$anon$2.unsafeRunSync(Dispatcher.scala:287)
	at sttp.tapir.server.vertx.cats.streams.fs2$$anon$1.mapToReadStream(fs2.scala:45)
	at sttp.tapir.server.vertx.cats.streams.fs2$$anon$1.asReadStream(fs2.scala:40)
	at sttp.tapir.server.vertx.encoders.VertxToResponseBody.fromStreamValue$$anonfun$1(VertxToResponseBody.scala:54)

A workaround is to create the Dispatcher on a blocking thread-pool via ZIO.blocking(Dispatcher.parallel.toScopedZIO)

This works correct if cats-effect runtime is used, because of their own implementation of the work-stealing thread-pool, which is capable of detecting blocking actions and spawning additional threads.

I wonder if .unsafeRunSync could be refactored to Async[F].async that does not involve thread blocking

Run the supplied Scala reproducer with four parallel streaming requests and inspect tapir-vertx-server-cats streams/fs2.scala, especially mapToReadStream and asReadStream, together with VertxToResponseBody.scala. Trace how Dispatcher.unsafeRunSync is used for the response stream and compare it with the reported ZIO blocking behavior. Done means the reproducer completes without freezing the executor, while preserving streaming responses.