diff --git a/http-core/src/main/scala/org/apache/pekko/http/impl/engine/server/HttpServerBluePrint.scala b/http-core/src/main/scala/org/apache/pekko/http/impl/engine/server/HttpServerBluePrint.scala index e08901e367..2948ceaed0 100644 --- a/http-core/src/main/scala/org/apache/pekko/http/impl/engine/server/HttpServerBluePrint.scala +++ b/http-core/src/main/scala/org/apache/pekko/http/impl/engine/server/HttpServerBluePrint.scala @@ -349,6 +349,13 @@ private[http] object HttpServerBluePrint { def onPull(): Unit = pull(responseIn) override def onDownstreamFinish(cause: Throwable): Unit = cancel(responseIn) }) + + override def postStop(): Unit = { + // the timeouts are scheduled on the materializer, so they outlive this stage unless they are cancelled here. + // Whatever is still open now belongs to a connection that is gone, so firing them cannot do anything useful. + openTimeouts.foreach(_.clear()) + openTimeouts = immutable.Queue.empty + } } } diff --git a/http-core/src/test/scala/org/apache/pekko/http/impl/engine/server/HttpServerWithExplicitSchedulerSpec.scala b/http-core/src/test/scala/org/apache/pekko/http/impl/engine/server/HttpServerWithExplicitSchedulerSpec.scala index c575c7ac0e..54fea9b178 100644 --- a/http-core/src/test/scala/org/apache/pekko/http/impl/engine/server/HttpServerWithExplicitSchedulerSpec.scala +++ b/http-core/src/test/scala/org/apache/pekko/http/impl/engine/server/HttpServerWithExplicitSchedulerSpec.scala @@ -19,10 +19,13 @@ import pekko.http.impl.util._ import pekko.http.scaladsl.model._ import pekko.http.scaladsl.model.headers.{ `Timeout-Access`, Connection } import pekko.stream.Materializer +import pekko.stream.scaladsl.{ BidiFlow, Flow, Keep } +import pekko.stream.testkit.scaladsl.{ TestSink, TestSource } import pekko.stream.testkit.Utils.assertAllStagesStopped import pekko.testkit.ExplicitlyTriggeredScheduler import org.scalatest.Inside +import scala.concurrent.Promise import scala.concurrent.duration._ /** Tests similar to HttpServerSpec that need ExplicitlyTriggeredScheduler */ @@ -171,6 +174,41 @@ class HttpServerWithExplicitSchedulerSpec extends PekkoSpecWithMaterializer( netOut.expectComplete() netIn.sendComplete() }) + + // the timeouts are scheduled on the materializer, so they are not tied to the lifetime of the stage that + // created them and have to be cancelled when a connection goes away with a request still open + "are cancelled when the stage that scheduled them stops" in assertAllStagesStopped { + val scheduler = system.scheduler.asInstanceOf[ExplicitlyTriggeredScheduler] + val timeoutHandlerCalled = Promise[Unit]() + + val timeoutSupport = + BidiFlow.fromGraph(new HttpServerBluePrint.RequestTimeoutSupport(400.millis, system.log)) + val applicationSide = + Flow.fromSinkAndSourceMat(TestSink[HttpRequest](), TestSource[HttpResponse]())(Keep.both) + val ((requestsIn, (applicationRequests, applicationResponses)), responsesOut) = + TestSource[HttpRequest]() + .viaMat(timeoutSupport.joinMat(applicationSide)(Keep.right))(Keep.both) + .toMat(TestSink[HttpResponse]())(Keep.both) + .run() + + responsesOut.request(1) + requestsIn.sendNext(HttpRequest()) + applicationRequests.requestNext().header[`Timeout-Access`].foreach( + _.timeoutAccess.updateHandler { (_: HttpRequest) => + timeoutHandlerCalled.trySuccess(()) + HttpResponse(StatusCodes.InternalServerError) + }) + + // the connection goes away while the application is still working on the response + requestsIn.sendComplete() + applicationRequests.expectComplete() + applicationResponses.sendComplete() + responsesOut.expectComplete() + + // nothing is left to run the timeout of a request that can no longer be answered + scheduler.timePasses(500.millis) + timeoutHandlerCalled.isCompleted shouldBe false + } } }