Tolerate Vert.x 5 pause/resume on ended requests (fixes silent test hangs) (#5383)

## Problem

The Vert.x 4 → 5 upgrade (80c5e37, 2026-06-10) changed
`HttpServerRequestImpl.pause()`/`fetch()`/`resume()` semantics: they now
`checkEnded()` and **throw** `IllegalStateException("Request has already
been read")` once the request body END event has been delivered. In
Vert.x 4 these calls were safe no-ops after end.

Tapir's vertx streaming bridge invokes `pause()`/`resume()` from
fire-and-forget fibers and Vert.x callbacks:

-
`server/vertx-server/cats/src/main/scala/sttp/tapir/server/vertx/cats/streams/fs2.scala`
— back-pressure loop in a started-and-never-joined fiber
-
`server/vertx-server/zio/src/main/scala/sttp/tapir/server/vertx/zio/streams/zio.scala`
— same pattern, `forkDaemon`-ed
-
`server/vertx-server/src/main/scala/sttp/tapir/server/vertx/streams/Pipe.scala`
— drain handlers / pipe start
-
`server/vertx-server/src/main/scala/sttp/tapir/server/vertx/handlers/package.scala`
— `streamPauseHandler` for every streaming/WS endpoint

Under CI load, a `pause`/`resume` racing the END event now throws,
silently killing the back-pressure fiber (or skipping cleanup such as
`socket.close()`), leaving stream consumers waiting forever. This is the
suspected root cause of the intermittent **silent hangs** in the vertx
cats/zio suites seen on dependency PRs #5373 / #5362, where CI went
quiet until the 15-minute retry timeout.

## Fix

Restore Vert.x 4 semantics at tapir's call sites by ignoring the
post-end `IllegalStateException`: after END there is nothing left to
pause/resume, so dropping the call is semantically correct. No
fibers/streams were restructured — only the pause/resume calls on the
request/read-stream are guarded (`.attempt.void` in fs2, `.ignore` in
zio, a small `ignoringReadEnded` helper in `Pipe.scala`, and a try/catch
in `streamPauseHandler`).

Companion PR: #5382 adds defensive timeouts + a thread-dump watchdog on
the CI side; this PR is the root-cause side.

## Verification

All variants compile (2.12 / 2.13 / 3): `vertxServer{,2_12,3}`,
`vertxServerCats{,2_12,3}`, `vertxServerZio{,2_12,3}`.

Test suites:
- `vertxServerCats2_12/test`: Tests: succeeded 292, failed 0 — all
passed
- `vertxServerZio3/test`: Tests: succeeded 295, failed 0 — all passed
- `vertxServer/test` (2.13): Tests: succeeded 554, failed 0 — all passed

Stress run of the formerly racy path — `vertxServerCats2_12/testOnly
sttp.tapir.server.vertx.cats.CatsVertxServerTest` 5 times in a loop: 5/5
passed (285 tests each).

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-authored-by: Claude Fable 5 <noreply@anthropic.com>

Author: adamw

server/vertx-server/cats/src/main/scala/sttp/tapir/server/vertx/cats/streams/fs2.scala

@@ -154,8 +154,9 @@ object fs2 {
                   } yield result.map((_, ()))
                 })
                 .evalMap({
-                  case Pause  => Sync[F].delay(readStream.pause())
-                  case Resume => Sync[F].delay(readStream.resume())
+                  // Vert.x 5 throws an IllegalStateException when pausing/resuming a fully-read request; it's safe to ignore (Vert.x 4 treated it as a no-op)
+                  case Pause  => Sync[F].delay(readStream.pause()).attempt.void
+                  case Resume => Sync[F].delay(readStream.resume()).attempt.void
                 })
                 .compile
                 .drain

server/vertx-server/src/main/scala/sttp/tapir/server/vertx/handlers/package.scala

@@ -21,7 +21,9 @@ package object handlers {
   }
 
   private[vertx] lazy val streamPauseHandler: Handler[RoutingContext] = { rc =>
-    rc.request.pause()
+    // Vert.x 5 throws an IllegalStateException when pausing a fully-read request; it's safe to ignore (Vert.x 4 treated it as a no-op)
+    try rc.request.pause()
+    catch { case _: IllegalStateException => () }
     rc.next()
   }
 

server/vertx-server/src/main/scala/sttp/tapir/server/vertx/streams/Pipe.scala

@@ -18,6 +18,12 @@ object Pipe {
   private case object Pause extends Command
   private case object Resume extends Command
 
+  // Vert.x 5 throws an IllegalStateException when pausing/resuming a fully-read request; it's safe to ignore (Vert.x 4 treated it as a no-op)
+  private def ignoringReadEnded[T](f: => T): Unit = {
+    try { val _ = f }
+    catch { case _: IllegalStateException => () }
+  }
+
   @tailrec
   def modify[A, B](ref: AtomicReference[A], f: A => (A, B)): B = {
     val oldA = ref.get
@@ -68,7 +74,7 @@ object Pipe {
       case Stop =>
         ()
       case act =>
-        if (act == Resume) request.resume() else if (act == Pause) request.pause() else ()
+        if (act == Resume) ignoringReadEnded(request.resume()) else if (act == Pause) ignoringReadEnded(request.pause()) else ()
         ref updateAndGet {
           case BackpressureState(true, i, commands) => BackpressureState(inProgress = false, i, commands)
           case unexpected                           => throw new Exception(s"Unexpected state $unexpected")
@@ -116,7 +122,7 @@ object Pipe {
       ()
     }
 
-    request.resume()
+    ignoringReadEnded(request.resume())
     ()
   }
 
@@ -170,7 +176,7 @@ object Pipe {
       ()
     }
 
-    request.resume()
+    ignoringReadEnded(request.resume())
     ()
   }
 }

server/vertx-server/zio/src/main/scala/sttp/tapir/server/vertx/zio/streams/zio.scala

@@ -144,10 +144,11 @@ package object streams {
             } yield result.map((_, ()))
           })
           .mapZIO({
+            // Vert.x 5 throws an IllegalStateException when pausing/resuming a fully-read request; it's safe to ignore (Vert.x 4 treated it as a no-op)
             case Pause =>
-              ZIO.attempt(readStream.pause())
+              ZIO.attempt(readStream.pause()).ignore
             case Resume =>
-              ZIO.attempt(readStream.resume())
+              ZIO.attempt(readStream.resume()).ignore
           })
           .runDrain
           .forkDaemon