Fix flaky SSE tests in akka/pekko server modules (#5387)

## Problem

The "Send and receive SSE" test in `AkkaHttpServerTest` flaked on CI
(seen on PR #5381's CI run, 2026-07-07, ci 2.13 / JVM 11) with only:

```
org.scalatest.exceptions.TestFailedException was thrown. (AkkaHttpServerTest.scala:82)
```

No message, no cause. An identical sibling test exists in
`PekkoHttpServerTest`.

Two things went wrong:

1. **Swallowed exception**: the test ended with `.transform(sse => sse
shouldBe ..., ex => fail(ex))`. `fail(ex)` wraps the real exception as
the *cause* of a message-less `TestFailedException`, and the test
reporter prints neither — hence the bare failure line.
2. **The actual flake**: the sttp client backend was created with
default `ConnectionPoolSettings`, whose
`response-entity-subscription-timeout` is **1s**. The test consumes the
response via `asStreamUnsafe` and only materializes the SSE-parsing
`runFold` in a later future callback. On a loaded CI runner that gap can
exceed 1s, at which point the akka/pekko-http client pool fails the
entity stream ("Response entity was not subscribed after 1 second...").

## Fix

- Build the per-test client backend with
`ConnectionPoolSettings(actorSystem).withResponseEntitySubscriptionTimeout(1.minute)`
so a slow runner can't hit the subscription timeout.
- Replace the `.transform(..., ex => fail(ex))` with a plain `.map(...)`
so if the test ever fails again, the real exception (message + type)
propagates to the report instead of a bare `TestFailedException`.
- Close the per-test client backend after the request completes (it was
previously leaked).

Same changes applied to both the akka and pekko variants.

## Force-repro proof

- A
`-Dakka.http.host-connection-pool.response-entity-subscription-timeout=1ms`
override propagates into the (non-forked) test JVM but does *not*
reproduce locally: on an idle machine the stream is subscribed before
the scheduler-tick-granularity timeout task can fire.
- Simulating CI load instead — inserting `Thread.sleep(1500)` before the
`runFold` materialization on unmodified master (default 1s timeout) —
reproduces the **exact** CI symptom: `TestFailedException was thrown.
(AkkaHttpServerTest.scala:82)`.
- The same 1500ms delay with this fix applied: test **passes**.

## Verification

- Both SSE tests (akka + pekko) run 3x each: all green.
- `akkaHttpServer2_12/Test/compile`, `pekkoHttpServer2_12/Test/compile`,
`pekkoHttpServer3/Test/compile`: all pass.
- Full `AkkaHttpServerTest` suite: 301 tests, 0 failures.

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

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

Author: adamw

server/akka-http-server/src/test/scala/sttp/tapir/server/akkahttp/AkkaHttpServerTest.scala

@@ -3,6 +3,7 @@ package sttp.tapir.server.akkahttp
 import akka.actor.ActorSystem
 import akka.http.scaladsl.model.HttpEntity
 import akka.http.scaladsl.server.{Directives, RequestContext}
+import akka.http.scaladsl.settings.ConnectionPoolSettings
 import akka.stream.scaladsl.{Flow, Sink, Source}
 import cats.effect.unsafe.implicits.global
 import cats.effect.{IO, Resource}
@@ -27,6 +28,7 @@ import sttp.tapir.tests.{Test, TestSuite}
 import java.util.UUID
 import java.util.concurrent.atomic.AtomicInteger
 import scala.concurrent.Future
+import scala.concurrent.duration._
 import scala.util.Random
 
 class AkkaHttpServerTest extends TestSuite with EitherValues {
@@ -64,6 +66,12 @@ class AkkaHttpServerTest extends TestSuite with EitherValues {
               Future.successful(Source(List(sse1, sse2)))
             })
           val route = AkkaHttpServerInterpreter().toRoute(e)
+          // the default 1s response-entity-subscription-timeout can be exceeded on a loaded CI runner before the
+          // SSE-parsing stream is materialized, failing the entity stream; hence raising the timeout
+          val clientBackend = AkkaHttpBackend.usingActorSystem(
+            actorSystem,
+            customConnectionPoolSettings = Some(ConnectionPoolSettings(actorSystem).withResponseEntitySubscriptionTimeout(1.minute))
+          )
           interpreter
             .server(route)
             .use { port =>
@@ -78,8 +86,9 @@ class AkkaHttpServerTest extends TestSuite with EitherValues {
                         )
                       )
                     )
-                    .send(AkkaHttpBackend.usingActorSystem(actorSystem))
-                    .flatMap(_.body.value.transform(sse => sse shouldBe List(sse1, sse2), ex => fail(ex)))
+                    .send(clientBackend)
+                    .flatMap(_.body.value.map(sse => sse shouldBe List(sse1, sse2)))
+                    .andThen { case _ => clientBackend.close() }
                 )
               }
             }

server/pekko-http-server/src/test/scala/sttp/tapir/server/pekkohttp/PekkoHttpServerTest.scala

@@ -3,6 +3,7 @@ package sttp.tapir.server.pekkohttp
 import org.apache.pekko.actor.ActorSystem
 import org.apache.pekko.http.scaladsl.model.HttpEntity
 import org.apache.pekko.http.scaladsl.server.{Directives, RequestContext}
+import org.apache.pekko.http.scaladsl.settings.ConnectionPoolSettings
 import org.apache.pekko.stream.scaladsl.{Flow, Sink, Source}
 import cats.effect.unsafe.implicits.global
 import cats.effect.{IO, Resource}
@@ -28,6 +29,7 @@ import sttp.tapir.tests.{Test, TestSuite}
 import java.util.UUID
 import java.util.concurrent.atomic.AtomicInteger
 import scala.concurrent.Future
+import scala.concurrent.duration._
 import scala.util.Random
 
 class PekkoHttpServerTest extends TestSuite with EitherValues {
@@ -65,6 +67,12 @@ class PekkoHttpServerTest extends TestSuite with EitherValues {
               Future.successful(Source(List(sse1, sse2)))
             })
           val route = PekkoHttpServerInterpreter().toRoute(e)
+          // the default 1s response-entity-subscription-timeout can be exceeded on a loaded CI runner before the
+          // SSE-parsing stream is materialized, failing the entity stream; hence raising the timeout
+          val clientBackend = PekkoHttpBackend.usingActorSystem(
+            actorSystem,
+            customConnectionPoolSettings = Some(ConnectionPoolSettings(actorSystem).withResponseEntitySubscriptionTimeout(1.minute))
+          )
           interpreter
             .server(route)
             .use { port =>
@@ -79,8 +87,9 @@ class PekkoHttpServerTest extends TestSuite with EitherValues {
                         )
                       )
                     )
-                    .send(PekkoHttpBackend.usingActorSystem(actorSystem))
-                    .flatMap(_.body.value.transform(sse => sse shouldBe List(sse1, sse2), ex => fail(ex)))
+                    .send(clientBackend)
+                    .flatMap(_.body.value.map(sse => sse shouldBe List(sse1, sse2)))
+                    .andThen { case _ => clientBackend.close() }
                 )
               }
             }