http2: add client-side cancellation propagation delay (#4027)

Author: jrudolph

akka-http-core/src/main/scala/akka/http/impl/engine/http2/Http2.scala

@@ -234,6 +234,7 @@ private[http] final class Http2Ext(implicit val system: ActorSystem)
       TLS(createEngine _, closing = TLSClosing.eagerClose)
 
     stack.joinMat(clientConnectionSettings.transport.connectTo(host, port, clientConnectionSettings)(system.classicSystem))(Keep.right)
+      .addAttributes(Http.cancellationStrategyAttributeForDelay(clientConnectionSettings.streamCancellationDelay))
   }
 
   def outgoingConnectionPriorKnowledge(host: String, port: Int, clientConnectionSettings: ClientConnectionSettings, log: LoggingAdapter): Flow[HttpRequest, HttpResponse, Future[OutgoingConnection]] = {
@@ -243,6 +244,7 @@ private[http] final class Http2Ext(implicit val system: ActorSystem)
       TLSPlacebo()
 
     stack.joinMat(clientConnectionSettings.transport.connectTo(host, port, clientConnectionSettings)(system.classicSystem))(Keep.right)
+      .addAttributes(Http.cancellationStrategyAttributeForDelay(clientConnectionSettings.streamCancellationDelay))
   }
 
   private def prepareClientAttributes(serverHost: String, port: Int): Attributes =

akka-http-core/src/main/scala/akka/http/impl/engine/http2/client/PersistentConnection.scala

@@ -13,7 +13,7 @@ import akka.http.scaladsl.settings.Http2ClientSettings
 import akka.stream.scaladsl.{ Flow, Keep, Source }
 import akka.stream.stage.TimerGraphStageLogic
 import akka.stream.stage.{ GraphStage, GraphStageLogic, InHandler, OutHandler, StageLogging }
-import akka.stream.{ Attributes, FlowShape, Inlet, Outlet }
+import akka.stream.{ Attributes, FlowShape, Inlet, Outlet, StreamTcpException }
 import akka.util.PrettyDuration
 
 import java.util.concurrent.ThreadLocalRandom
@@ -133,8 +133,10 @@ private[http2] object PersistentConnection {
           }
         }
         val onFailed = getAsyncCallback[Throwable] { cause =>
+          // If the materialized value is failed, then the stream should be broken by design.
+          // Nevertheless also kick our ends of the stream.
           responseIn.cancel()
-          requestOut.fail(new RuntimeException("connection broken", cause))
+          requestOut.fail(new StreamTcpException("connection broken"))
 
           if (connectsLeft.contains(0)) {
             failStage(new RuntimeException(s"Connection failed after $maxAttempts attempts", cause))

akka-http2-support/src/test/scala/akka/http/impl/engine/http2/Http2ClientServerSpec.scala

@@ -13,6 +13,7 @@ import akka.http.scaladsl.settings.ClientConnectionSettings
 import akka.http.scaladsl.unmarshalling.Unmarshal
 import akka.http.scaladsl.Http
 import akka.http.scaladsl.settings.ServerSettings
+import akka.stream.StreamTcpException
 import akka.stream.scaladsl.{ Sink, Source }
 import akka.stream.testkit.{ TestPublisher, TestSubscriber }
 import akka.testkit.TestProbe
@@ -98,7 +99,8 @@ class Http2ClientServerSpec extends AkkaSpecWithMaterializer(
       clientResponsesIn.ensureSubscription()
       Thread.sleep(500)
       clientRequestsOut.expectCancellation()
-      clientResponsesIn.expectComplete()
+      // expect idle timeout connection abort exception to propagate to user
+      clientResponsesIn.expectError() shouldBe a[StreamTcpException]
     }
     "support client-side idle-timeout" in new TestSetup {
       override def clientSettings: ClientConnectionSettings = super.clientSettings.withIdleTimeout(100.millis)

akka-http2-support/src/test/scala/akka/http/impl/engine/http2/Http2PersistentClientSpec.scala

@@ -15,7 +15,7 @@ import akka.http.scaladsl.settings.Http2ClientSettings
 import akka.http.scaladsl.unmarshalling.Unmarshal
 import akka.http.scaladsl.{ ClientTransport, Http }
 import akka.http.scaladsl.settings.ServerSettings
-import akka.stream.{ KillSwitches, UniqueKillSwitch }
+import akka.stream.{ KillSwitches, StreamTcpException, UniqueKillSwitch }
 import akka.stream.scaladsl.{ Flow, Keep, Sink, Source }
 import akka.stream.testkit.scaladsl.StreamTestKit
 import akka.stream.testkit.{ TestPublisher, TestSubscriber }
@@ -364,7 +364,7 @@ abstract class Http2PersistentClientSpec(tls: Boolean) extends AkkaSpecWithMater
       })
 
     val killProbe = TestProbe()
-    def killConnection(): Unit = killProbe.expectMsgType[UniqueKillSwitch].abort(new RuntimeException("connection was killed"))
+    def killConnection(): Unit = killProbe.expectMsgType[UniqueKillSwitch].abort(new StreamTcpException("connection was killed"))
 
     object server {
       private lazy val requestProbe = TestProbe()