From 34e0baec242276d7ef5b5bbd4a0d897e45343ac3 Mon Sep 17 00:00:00 2001 From: PJ Fanning Date: Wed, 16 Sep 2026 11:23:40 +0100 Subject: [PATCH] fix: fail instead of throw when a transform on a completed FastFuture throws Motivation: `FastFuture.successful` / `FastFuture.failed` return already completed futures that override `Future.transform` to apply the function directly. Unlike `DefaultPromise`, that override did not catch exceptions thrown by the function, so the standard `filter`, `collect` and `transform` combinators on such a future threw synchronously at the call site instead of returning a failed future. Callers cannot tell a FastFuture-produced `Future` from any other, and pekko-grpc now returns them from its public APIs. Modification: Route both `transform` overrides through a helper that converts a `NonFatal` exception from the function into an `ErrorFuture`, matching the contract of `Future.transform`. Result: Standard combinators on a completed FastFuture return a failed future when their function throws, the same as for `Future.successful` / `Future.failed`. Tests: - sbt "http-core/Test/testOnly org.apache.pekko.http.scaladsl.util.FastFutureSpec" - new transform/filter/collect cases fail without the fix, 32/32 pass with it - sbt "http-core/mimaReportBinaryIssues" - no issues (against 1.0.0) - scalafmt --mode diff-ref=upstream/main - git diff --check References: None - found while reviewing FastFuture stack and exception semantics for pekko-grpc --- .../pekko/http/scaladsl/util/FastFuture.scala | 12 +++++-- .../http/scaladsl/util/FastFutureSpec.scala | 31 +++++++++++++++++++ 2 files changed, 41 insertions(+), 2 deletions(-) diff --git a/http-core/src/main/scala/org/apache/pekko/http/scaladsl/util/FastFuture.scala b/http-core/src/main/scala/org/apache/pekko/http/scaladsl/util/FastFuture.scala index ca5dc1f208..bb1807626b 100644 --- a/http-core/src/main/scala/org/apache/pekko/http/scaladsl/util/FastFuture.scala +++ b/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) } diff --git a/http-core/src/test/scala/org/apache/pekko/http/scaladsl/util/FastFutureSpec.scala b/http-core/src/test/scala/org/apache/pekko/http/scaladsl/util/FastFutureSpec.scala index b8b78b4c1b..fb3e0e474f 100644 --- a/http-core/src/test/scala/org/apache/pekko/http/scaladsl/util/FastFutureSpec.scala +++ b/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)