ServerBinding termination does not close HTTP/2 connections

[bdmendes]

Calling the ServerBinding terminate() method appears to not close open HTTP/2 connections reused by a client for different requests, even after the hardDeadline timeout is exceeded. In fact, new requests are still handled successfully even after the termination future completes.

Steps to reproduce (in 1.1.0):

1. Use enable-http2=on in the Typesafe Config.
2. Create a binding with Http().newServerAt(...).bind(...).
3. Make an HTTP request to this service with a client. I used Unirest for this.
4. Request binding termination and await for it.
5. Make more HTTP requests with the same connection. Unirest does that behind the scenes.
6. Verify that the service still responds, even though it was supposed to be terminated.

[pjfanning]

And http 1.1 behaves ok?

[bdmendes]

Yes, everything behaves as expected with HTTP 1.1.

[pjfanning]

@bdmendes if you have the code you used to reproduce this can you share it? Helps avoid having someone write the test case from scratch.

I suspect that the Unirest client is keeping an active HTTP/2 connection open and that when you use the client again after the ServerBinding close that the connection is still open. If this is the case, we would need to rework ServerBinding close to force any open HTTP/2 connections to be closed.

[He-Pin]

is it by designed ? @jrudolph cc.

[jrudolph]

In theory, it should work. It was implemented in 4d16224, so it might be a bug or an edge case? A reproducer or at least debug logs would help a lot.

[bdmendes]

@bdmendes if you have the code you used to reproduce this can you share it? Helps avoid having someone write the test case from scratch.

I suspect that the Unirest client is keeping an active HTTP/2 connection open and that when you use the client again after the ServerBinding close that the connection is still open. If this is the case, we would need to rework ServerBinding close to force any open HTTP/2 connections to be closed.

I have a Pekko actor that does:

abstract class HttpActor extends Actor {
  var bindingOpt = Option.empty[ServerBinding]
  override def preStart() = {
    implicit val sys: ActorSystem = context.system
    val builder = Http().newServerAt(interface = "0.0.0.0", port = 4321)
    builder.bind(...).pipeTo(self)
  }
  override def postStop(): Unit = {
    bindingOpt.foreach(
      _.terminate(hardDeadline =
        CoordinatedShutdown(context.system).timeout(CoordinatedShutdown.PhaseServiceRequestsDone)
    )
  }
  def receive = { case binding: ServerBinding =>
    logger.info(s"Bound to ${binding.localAddress}.$timeoutOverridden")
    bindingOpt = Some(binding)
  }
}

Then I have a spec that does:

def get(req: String) = Unirest.get(req).headers(Map()).exec(10.seconds)
val actor = system.actorOf(Props(new TestHttpServiceActor())) // of type HttpActor
get(s"http://localhost:4321/hello").getBody must beEqualTo("world").eventually
system.stop(actor)
get(s"http://localhost:4321/hello").isSuccess must beFalse.eventually

Enabling HTTP/2 fails the last assertion.
Thank you for the feedback.

[pjfanning]

I've put a POC at https://github.com/pjfanning/pekko-http-sample/tree/binding-terminate (a branch of my pekko-http-sample project).

sbt run is enough to run the POC.

When http2 is not enabled, the HTTP request after the binding terminate fails.
With http2 enabled, the HTTP request after the binding terminate runs ok and returns hello.

Issue happens with pekko-http 1.0.1 and 1.1.0.

I did an extra test where I did a Thread.sleep of 30s and then tried another HTTP request and that failed with a connection exception. It does look like the server binding terminate does eventually close all the open connections but that the Future returned by binding.terminate completes too quickly.

A shorter sleep of 5s was not enough.

[jrudolph]

Thanks for putting that together @pjfanning. I only looked quickly into it, it might be that the issue only happens during the uncommon non-tls Http 1.1 => Http 2 upgrade request.

Does the issue show for you in a full Http 2 over Https setting, @bdmendes?

[jrudolph]

In some way, it's no wonder that this is not implemented as the h2c 101 upgrade flow is the most complicated of all the h2 server setups (and also not supported on the client side). What would need to be done to support this is to change Http2.handleUpgradeRequest to somehow surface the ServerTerminator from the HTTP2 stage (which will only be created later) and combine it with the static one created for the native http1 variant created in

pekko-http/http-core/src/main/scala/org/apache/pekko/http/impl/engine/http2/Http2.scala

Line 97 in 5427d0c

.joinMat(GracefulTerminatorStage(system, settings).atop(http.serverLayer(settings, log = log)))(Keep.right)

.

Not sure if that is worth it for a very uncommon setup outside of testing (h2c using upgrade headers).

WDYT, @raboof?

[jrudolph]

My idea was to try to invert the flow using a joinable ServerTerminator implementation like this:

private class JoinableServerTermination extends ServerTerminator {
  private val terminationInitiated = Promise[Done]()
  private val terminationCompleted = new AtomicReference[Future[Http.HttpTerminated]](Future.successful(Http.HttpServerTerminated))

  def onTerminationInitiated(f: () => Future[Http.HttpTerminated]): Unit = {
    val promise = Promise[Http.HttpTerminated]()

    def chain(f: Future[Http.HttpTerminated]): Unit = {
      val p = terminationCompleted.get()
      if (!terminationCompleted.compareAndSet(p,p.flatMap(_ => promise.future)(ExecutionContexts.parasitic)))
        chain(f)
    }

    chain(promise.future)
    terminationInitiated.future.foreach { _ => promise.completeWith(f())
    }(ExecutionContexts.parasitic)
  }

  override def terminate(deadline: FiniteDuration)(implicit ex: ExecutionContext): Future[Http.HttpTerminated] = {
    terminationInitiated.tryComplete(Success(Done))
    terminationCompleted.get()
  }
}

The next question would be how to get it into mapAsync(...)(handleUpgradeRequest) which could work something like this:

val http1: HttpImplementation =
        Flow.lazyFlow { () =>
          val termination = new JoinableServerTermination
          Flow[HttpRequest]
          .mapAsync(settings.pipeliningLimit)(handleUpgradeRequests(handler, settings, log, termination))
            .mapMaterializedValue(_ => termination)
        }
        .joinMat(GracefulTerminatorStage(system, settings).atop(http.serverLayer(settings, log = log)))( /* join http1 termination here */)

In handleUpgradeRequest the ServerTerminator coming out of Http2.serverStack would have to be joined to the joinable termination.

[bdmendes]

Thanks for putting that together @pjfanning. I only looked quickly into it, it might be that the issue only happens during the uncommon non-tls Http 1.1 => Http 2 upgrade request.

Does the issue show for you in a full Http 2 over Https setting, @bdmendes?

Sorry for the delay. Considering there is no explicit http2 setting in the Unirest client, I'd imagine it indeed defaults to http/1 and then the server handles the upgrade. I didn't confirm that, though.

[jrudolph]

The question was also about whether it happens when TLS is enabled (as it is common for http2).