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

[susliko]

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