feat: grace period for HTTP/2 GOAWAY (#4528)

Author: johanandren

akka-http-core/src/main/mima-filters/10.7.3.backwards.excludes/goaway-grace-time.excludes

@@ -0,0 +1,3 @@
+# not for user extension
+ProblemFilters.exclude[ReversedMissingMethodProblem]("akka.http.scaladsl.settings.Http2ClientSettings.goawayGracePeriod")
+ProblemFilters.exclude[ReversedMissingMethodProblem]("akka.http.scaladsl.settings.Http2ServerSettings.goawayGracePeriod")

akka-http-core/src/main/resources/reference.conf

@@ -315,6 +315,13 @@ akka.http {
       # Protects against rapid reset attacks. If a connection goes over the limit, it is closed with HTTP/2 protocol error ENHANCE_YOUR_CALM
       max-resets = 400
       max-resets-interval = 10s
+
+      # After sending a GOAWAY frame during graceful termination, wait this long before closing the
+      # TCP connection. This gives the client time to receive the GOAWAY frame before the connection
+      # is torn down, avoiding a TCP RST on the client's next read. This is especially important for
+      # idle or nearly-idle connections where the stage would otherwise complete almost immediately
+      # after queuing the GOAWAY.
+      goaway-grace-period = 250ms
     }
 
     websocket {
@@ -508,6 +515,11 @@ akka.http {
       # requests to complete.
       completion-timeout = 3s
 
+      # After sending a GOAWAY frame, wait this long before closing the TCP connection. This gives the
+      # peer time to receive the GOAWAY frame before the connection is torn down. Set to 0 to not wait but instead
+      # immediately close the connection.
+      goaway-grace-period = 250ms
+
     }
 
     #client-settings

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

@@ -217,6 +217,7 @@ private[http2] abstract class Http2Demux(http2Settings: Http2CommonSettings, ini
       logic =>
 
       import Http2Demux.CompletionTimeout
+      import Http2Demux.GoAwayGracePeriod
 
       def wrapTrailingHeaders(headers: ParsedHeadersFrame): Option[HttpEntity.ChunkStreamPart] = stage.wrapTrailingHeaders(headers)
 
@@ -233,6 +234,7 @@ private[http2] abstract class Http2Demux(http2Settings: Http2CommonSettings, ini
 
       private val terminationPromise = Promise[Http.HttpTerminated]()
       private var terminating: Boolean = false
+      private var goAwayGracePeriodElapsed: Boolean = false
       private var lastIdBeforeTermination: Int = 0
       private val terminateCallback = getAsyncCallback[FiniteDuration](triggerTermination)
       override def terminate(deadline: FiniteDuration)(implicit ex: ExecutionContext): Future[Http.HttpTerminated] = {
@@ -242,11 +244,24 @@ private[http2] abstract class Http2Demux(http2Settings: Http2CommonSettings, ini
       private def triggerTermination(deadline: FiniteDuration): Unit =
         // check if we are already terminating, otherwise start termination
         if (!terminating) {
-          log.debug(s"Termination of this connection was triggered. Sending GOAWAY and waiting for open requests to complete for $CompletionTimeout.")
-          terminating = true
-          pushGOAWAY(ErrorCode.NO_ERROR, "Voluntary connection close.")
-          lastIdBeforeTermination = lastStreamId()
-          completeIfDone()
+          if (deadline == Duration.Zero) {
+            log.debug("Termination of this connection was triggered. Sending GOAWAY and waiting for open requests to complete for {}.", CompletionTimeout)
+            terminating = true
+            goAwayGracePeriodElapsed = true
+            pushGOAWAY(ErrorCode.NO_ERROR, "Voluntary connection close.")
+            lastIdBeforeTermination = lastStreamId()
+            completeIfDone()
+          } else {
+            log.debug("Termination of this connection was triggered. Sending GOAWAY and waiting for open requests to complete for {}.", deadline)
+            terminating = true
+            // First GOAWAY per RFC 7540 §6.8: use last-stream-id = Int.MaxValue to signal graceful
+            // shutdown intent. The client must stop initiating new streams, but streams already in
+            // flight before the client receives this frame may still arrive. We accept those during
+            // the grace period by keeping lastIdBeforeTermination at Int.MaxValue for now.
+            multiplexer.pushControlFrame(GoAwayFrame(Int.MaxValue, ErrorCode.NO_ERROR, ByteString("Voluntary connection close.")))
+            lastIdBeforeTermination = Int.MaxValue
+            scheduleOnce(GoAwayGracePeriod, http2Settings.goawayGracePeriod)
+          }
           if (!isClosed(frameOut))
             scheduleOnce(CompletionTimeout, deadline)
         }
@@ -297,7 +312,6 @@ private[http2] abstract class Http2Demux(http2Settings: Http2CommonSettings, ini
       override def pushGOAWAY(errorCode: ErrorCode, debug: String): Unit = {
         val frame = GoAwayFrame(lastStreamId(), errorCode, ByteString(debug))
         multiplexer.pushControlFrame(frame)
-        // FIXME: handle the connection closing according to the specification
       }
       private[this] var allowReadingIncomingFrames: Boolean = true
       override def allowReadingIncomingFrames(allow: Boolean): Unit = {
@@ -415,7 +429,10 @@ private[http2] abstract class Http2Demux(http2Settings: Http2CommonSettings, ini
       private def completeIfDone(): Unit = {
         val noMoreOutgoingStreams = (terminating || isClosed(substreamIn)) && activeStreamCount() == 0
         def allOutgoingDataFlushed = isClosed(frameOut) || multiplexer.hasFlushedAllData
-        if (noMoreOutgoingStreams && allOutgoingDataFlushed) {
+        // When terminating via GOAWAY, delay closure until the grace period has elapsed so the
+        // client has time to receive the GOAWAY frame before we tear down the TCP connection.
+        val gracePeriodDone = !terminating || goAwayGracePeriodElapsed
+        if (noMoreOutgoingStreams && allOutgoingDataFlushed && gracePeriodDone) {
           log.debug("Closing connection after all streams are done and all data has been flushed.")
           if (isServer)
             completeStage()
@@ -488,6 +505,18 @@ private[http2] abstract class Http2Demux(http2Settings: Http2CommonSettings, ini
       }
 
       override protected def onTimer(timerKey: Any): Unit = timerKey match {
+        case GoAwayGracePeriod =>
+          // Second GOAWAY per RFC 7540 §6.8: now that the grace period has elapsed, confirm the
+          // actual last stream ID that was processed. The client must retry any streams above this
+          // value. Only send if the connection is still open — it may have already been closed by
+          // CompletionTimeout or a peer-initiated close during the grace period.
+          if (!isClosed(frameOut)) {
+            lastIdBeforeTermination = lastStreamId()
+            pushGOAWAY(ErrorCode.NO_ERROR, "Voluntary connection close.")
+          }
+          goAwayGracePeriodElapsed = true
+          completeIfDone()
+
         case ConfigurablePing.Tick =>
           // don't do anything unless there are active streams
           if (activeStreamCount() > 0) {
@@ -524,4 +553,5 @@ private[http2] abstract class Http2Demux(http2Settings: Http2CommonSettings, ini
 @InternalApi
 private[akka] object Http2Demux {
   case object CompletionTimeout
+  case object GoAwayGracePeriod
 }

akka-http-core/src/main/scala/akka/http/javadsl/settings/Http2ClientSettings.scala

@@ -46,4 +46,7 @@ trait Http2ClientSettings { self: scaladsl.settings.Http2ClientSettings.Http2Cli
   def getMaxConnectionBackoff: Duration = Duration.ofMillis(maxConnectionBackoff.toMillis)
   def withMaxConnectionBackoff(backoff: Duration): Http2ClientSettings = copy(maxConnectionBackoff = backoff.toMillis.millis)
 
+  def getGoawayGracePeriod: Duration = Duration.ofMillis(goawayGracePeriod.toMillis)
+  def withGoawayGracePeriod(duration: Duration): Http2ClientSettings = copy(goawayGracePeriod = duration.toMillis.millis)
+
 }

akka-http-core/src/main/scala/akka/http/javadsl/settings/Http2ServerSettings.scala

@@ -48,6 +48,10 @@ trait Http2ServerSettings { self: scaladsl.settings.Http2ServerSettings with akk
 
   def withMaxResetsInterval(interval: Duration): Http2ServerSettings = copy(maxResetsInterval = interval.toMillis.millis)
 
+  def getGoawayGracePeriod: Duration = Duration.ofMillis(goawayGracePeriod.toMillis)
+
+  def withGoawayGracePeriod(duration: Duration): Http2ServerSettings = copy(goawayGracePeriod = duration.toMillis.millis)
+
 }
 object Http2ServerSettings extends SettingsCompanion[Http2ServerSettings] {
   def create(config: Config): Http2ServerSettings = scaladsl.settings.Http2ServerSettings(config)

akka-http-core/src/main/scala/akka/http/scaladsl/settings/Http2ServerSettings.scala

@@ -34,6 +34,8 @@ private[http] trait Http2CommonSettings {
 
   def pingInterval: FiniteDuration
   def pingTimeout: FiniteDuration
+
+  def goawayGracePeriod: FiniteDuration
 }
 
 /**
@@ -96,6 +98,9 @@ trait Http2ServerSettings extends javadsl.settings.Http2ServerSettings with Http
 
   def withMaxResetsInterval(interval: FiniteDuration): Http2ServerSettings = copy(maxResetsInterval = interval)
 
+  def goawayGracePeriod: FiniteDuration
+  def withGoawayGracePeriod(duration: FiniteDuration): Http2ServerSettings = copy(goawayGracePeriod = duration)
+
   @InternalApi
   private[http] def internalSettings: Option[Http2InternalServerSettings]
   @InternalApi
@@ -120,6 +125,7 @@ object Http2ServerSettings extends SettingsCompanion[Http2ServerSettings] {
     pingTimeout:                       FiniteDuration,
     maxResets:                         Int,
     maxResetsInterval:                 FiniteDuration,
+    goawayGracePeriod:                 FiniteDuration,
     internalSettings:                  Option[Http2InternalServerSettings]
   )
     extends Http2ServerSettings {
@@ -147,6 +153,7 @@ object Http2ServerSettings extends SettingsCompanion[Http2ServerSettings] {
       pingTimeout = c.getFiniteDuration("ping-timeout"),
       maxResets = c.getInt("max-resets"),
       maxResetsInterval = c.getFiniteDuration("max-resets-interval"),
+      goawayGracePeriod = c.getFiniteDuration("goaway-grace-period"),
       internalSettings = None, // no possibility to configure internal settings with config
     )
   }
@@ -194,6 +201,9 @@ trait Http2ClientSettings extends javadsl.settings.Http2ClientSettings with Http
   def completionTimeout: FiniteDuration
   def withCompletionTimeout(timeout: FiniteDuration): Http2ClientSettings = copy(completionTimeout = timeout)
 
+  def goawayGracePeriod: FiniteDuration
+  def withGoawayGracePeriod(duration: FiniteDuration): Http2ClientSettings = copy(goawayGracePeriod = duration)
+
   def baseConnectionBackoff: FiniteDuration
   def withBaseConnectionBackoff(backoff: FiniteDuration): Http2ClientSettings = copy(baseConnectionBackoff = backoff)
 
@@ -225,6 +235,7 @@ object Http2ClientSettings extends SettingsCompanion[Http2ClientSettings] {
     completionTimeout:                 FiniteDuration,
     baseConnectionBackoff:             FiniteDuration,
     maxConnectionBackoff:              FiniteDuration,
+    goawayGracePeriod:                 FiniteDuration,
     internalSettings:                  Option[Http2InternalClientSettings])
     extends Http2ClientSettings with javadsl.settings.Http2ClientSettings {
     require(maxConcurrentStreams >= 0, "max-concurrent-streams must be >= 0")
@@ -252,6 +263,7 @@ object Http2ClientSettings extends SettingsCompanion[Http2ClientSettings] {
       completionTimeout = c.getFiniteDuration("completion-timeout"),
       baseConnectionBackoff = c.getFiniteDuration("base-connection-backoff"),
       maxConnectionBackoff = c.getFiniteDuration("max-connection-backoff"),
+      goawayGracePeriod = c.getFiniteDuration("goaway-grace-period"),
       internalSettings = None // no possibility to configure internal settings with config
     )
   }