fix: fail instead of throw when a transform on a completed FastFuture throws

http-core/src/main/scala/org/apache/pekko/http/scaladsl/util/FastFuture.scala

@@ -90,7 +90,7 @@ object FastFuture {
     def ready(atMost: Duration)(implicit permit: CanAwait) = this
     def transform[S](f: scala.util.Try[A] => scala.util.Try[S])(
         implicit executor: scala.concurrent.ExecutionContext): scala.concurrent.Future[S] =
-      FastFuture(f(Success(a)))
+      strictTransform(Success(a), f)
     def transformWith[S](f: scala.util.Try[A] => scala.concurrent.Future[S])(
         implicit executor: scala.concurrent.ExecutionContext): scala.concurrent.Future[S] =
       new FastFuture(this).transformWith(f)
@@ -103,12 +103,20 @@ object FastFuture {
     def ready(atMost: Duration)(implicit permit: CanAwait) = this
     def transform[S](f: scala.util.Try[Nothing] => scala.util.Try[S])(
         implicit executor: scala.concurrent.ExecutionContext): scala.concurrent.Future[S] =
-      FastFuture(f(Failure(error)))
+      strictTransform(Failure(error), f)
     def transformWith[S](f: scala.util.Try[Nothing] => scala.concurrent.Future[S])(
         implicit executor: scala.concurrent.ExecutionContext): scala.concurrent.Future[S] =
       new FastFuture(this).transformWith(f)
   }
 
+  /**
+   * Applies `f` to an already available result, turning an exception thrown by `f` into a failed future
+   * the way [[scala.concurrent.Future.transform]] does for asynchronously completed futures.
+   */
+  private def strictTransform[A, S](value: Try[A], f: Try[A] => Try[S]): Future[S] =
+    try FastFuture(f(value))
+    catch { case NonFatal(e) => ErrorFuture(e) }
+
   implicit class EnhancedFuture[T](val future: Future[T]) extends AnyVal {
     def fast: FastFuture[T] = new FastFuture[T](future)
   }

http-core/src/test/scala/org/apache/pekko/http/scaladsl/util/FastFutureSpec.scala

@@ -171,6 +171,37 @@ class FastFutureSpec extends AnyFreeSpec with Matchers {
     }
   }
 
+  "FastFuture-produced futures should fail instead of throwing when a standard Future combinator's function throws" - {
+    // The already completed futures override `Future.transform`, which the standard `filter`, `collect` and
+    // `transform` combinators are built on, so an exception thrown by the user function must become a
+    // failed future exactly as it does for a `Future.successful` / `Future.failed` value.
+    "transform on a successful future" in {
+      testStdlib(Success(23), _.transform(_ => throw TheException))
+    }
+    "transform on a failed future" in {
+      testStdlib(Failure(UnexpectedException), _.transform(_ => throw TheException))
+    }
+    "filter with a throwing predicate" in {
+      testStdlib(Success(23), _.filter(_ => throw TheException))
+    }
+    "collect with a throwing partial function" in {
+      testStdlib(Success(23), _.collect { case _ => throw TheException })
+    }
+    "map with a throwing function" in {
+      testStdlib(Success(23), _.map(_ => throw TheException))
+    }
+    "recover with a throwing partial function" in {
+      testStdlib(Failure(UnexpectedException), _.recover { case _ => throw TheException })
+    }
+  }
+
+  def testStdlib(result: Try[Int], op: Future[Int] => Future[Int]): Unit = {
+    val f = FastFuture(result)
+    val transformed = op(f)
+    Await.ready(transformed, 500.millis)
+    transformed.value shouldEqual Some(Failure(TheException))
+  }
+
   def test(result: Try[Int], op: FastFuture[Int] => Future[Int])(check: Try[Int] => Unit): Unit = {
     def testStrictly(): Unit = {
       val f = FastFuture(result)