Retries on connect timeout

[raghavan-arvind]

Currently connect timeouts are not retried for POST/PATCH requests which are not considered idempotent. I agree that on connection reset we can't tell if the server has already started processing the request or not, but I wonder if there is any reason to not allow retries up to max-retries on connection timeout errors (no response to SYN packet) even for POST/PATCH

[johanandren]

Sounds like it could make sense. From a quick look at the retry implementation for idempotent calls, it cannot immediately be piggybacked on that so would require a bit of effort to add.

[johanandren]

I tried a few things around this out today. To me it seems it is already working as requested.

1. A server accepting connections but not sending anything back, so triggering the underlying Akka TCP idle connection timeout of 60s: this is retried up to max-retries times.
2. A server accepting the connection but then issuing a connection reset, this is retried max-retries and then fails the request with Failure(akka.stream.StreamTcpException: The connection closed with error: Connection reset)

Sample test app to try it out with

  // to get some output on each try to connect
  class LoggingTransport(actualTransport: ClientTransport) extends ClientTransport {
    private val log = LoggerFactory.getLogger(getClass)

    override def connectTo(host: String, port: Int, settings: ClientConnectionSettings)(implicit system: actor.ActorSystem): scaladsl.Flow[ByteString, ByteString, Future[Http.OutgoingConnection]] =
      actualTransport.connectTo(host, port, settings)(system).mapMaterializedValue { mat =>
        log.info("Connecting to {}", host)
        mat
      }
  }

  def main(args: Array[String]): Unit = {
    implicit val system: ActorSystem[Nothing] = ActorSystem[Nothing](Behaviors.empty, "HelloAkkaHttpServer")

    // accept but fail, connection reset
    Tcp(system).bind("localhost", 9777).runWith(Sink.foreach(connection =>
        connection.handleWith(Flow[ByteString].map(_ => throw new RuntimeException("boom")))
    ))
    // accept but then stall
    // Tcp(system).bind("localhost", 9777).runWith(Sink.foreach(connection =>
    //  connection.handleWith(Flow.fromSinkAndSource(Sink.never, Source.never))
    // ))
    
    system.log.info("Issuing request")
    Http()(system).singleRequest(HttpRequest(uri = "http://localhost:9777"), settings =
      ConnectionPoolSettings(system).withUpdatedConnectionSettings(s => s.withTransport(new LoggingTransport(s.transport)))
      )
      .onComplete(t => system.log.info("Result: {}", t))(system.executionContext)
  }

[raghavan-arvind]

Thanks for the response! The issue we encountered was that the server was not establishing the TCP connection, and we hit the connect timeout (I believe this is distinct from accepting the connection but not sending data and failing due to the idle timeout).

For us, we are routing through an LB, so simply retrying is usually enough to find a healthy upstream; you can imagine we could also try to re-resolve.

[johanandren]

If you could try to provide a reproducer or steps to reproduce locally that would be great.

[raghavan-arvind]

Here's a version that reproduces for me (our versions are old and we're on scala 2.12 so that might be an issue as well).

First, find some endpoint where we get connection reset (I noticed it's the same for connection reset and black hole). localhost:8080 works for me locally since there is nothing running there and Mac OS automatically resets localhost (for me, worth double checking).

In one shell:

sudo tcpdump -n -i any dst port 8080

In another shell:

curl localhost:8080

In tcpdump I see:

22:41:12.703339 IP6 ::1.59527 > ::1.8080: Flags [S], seq 3489025623, win 65535, options [mss 16324,nop,wscale 6,nop,nop,TS val 1047038633 ecr 0,sackOK,eol], length 0
22:41:12.703359 IP6 ::1.59527 > ::1.8080: Flags [S], seq 3489025623, win 65535, options [mss 16324,nop,wscale 6,nop,nop,TS val 1047038633 ecr 0,sackOK,eol], length 0
22:41:12.703503 IP 127.0.0.1.59528 > 127.0.0.1.8080: Flags [S], seq 2125315631, win 65535, options [mss 16344,nop,wscale 6,nop,nop,TS val 1425853684 ecr 0,sackOK,eol], length 0
22:41:12.703538 IP 127.0.0.1.59528 > 127.0.0.1.8080: Flags [S], seq 2125315631, win 65535, options [mss 16344,nop,wscale 6,nop,nop,TS val 1425853684 ecr 0,sackOK,eol], length 0

A little surprising but it looks like curl is retrying the tcp connection; two SYN packets get sent and then it actually tries again with another source port. I get the same thing with curl -X POST localhost:8080.

The scala example is, build.sbt:

name := "AkkaHttpMinimalClient"

version := "0.1"

scalaVersion := "2.12.19"

val akkaVersion = "2.7.0"
val akkaHttpVersion = "10.4.0"

libraryDependencies ++= Seq(
  "com.typesafe.akka" %% "akka-actor"  % akkaVersion,
  "com.typesafe.akka" %% "akka-stream" % akkaVersion,
  "com.typesafe.akka" %% "akka-http"   % akkaHttpVersion
)

Main.scala:

import akka.actor.ActorSystem
import akka.stream.ActorMaterializer
import akka.stream.scaladsl.{Sink, Source}
import akka.http.scaladsl.Http
import akka.http.scaladsl.model._
import akka.http.scaladsl.settings.ConnectionPoolSettings

import scala.util.{Success, Failure}
import scala.concurrent.{Await, Future}
import scala.concurrent.duration._

object Main extends App {
  implicit val system = ActorSystem("HttpClientSystem")
  implicit val materializer = ActorMaterializer()
  import system.dispatcher

  // Set connection pool settings programmatically
  val poolSettings = ConnectionPoolSettings(system).withMaxRetries(3)

  val pool = Http().cachedHostConnectionPool[Unit]("127.0.0.42", 8080, settings = poolSettings)

  val requests = Source.single((HttpRequest(uri = "/"), ()))

  val responseFuture: Future[Unit] =
    requests
      .via(pool)
      .runWith(Sink.head)
      .flatMap {
        case (Success(response), _) =>
          println(s"Got response: ${response.status}")
          response.entity.dataBytes.runForeach(chunk => println(chunk.utf8String)).map(_ => ())

        case (Failure(ex), _) =>
          println(s"Request failed: ${ex.getMessage}")
          Future.successful(())
      }
      .andThen { case _ => system.terminate() }

  Await.result(responseFuture, 30.seconds)
}

This initiates a GET request with max-retries = 3. In the tcpdump I see four bursts of packets, each burst has a different source port. This is great; different source port means the LB will send it to a different downstream, and we can configure connect timeout such that it retries as fast as we want (based on expected round-trip latency) -- amazing!! for our usecase.

Here's the same thing, three retries, this time with POST (same build.sbt):

import akka.actor.ActorSystem
import akka.stream.ActorMaterializer
import akka.stream.scaladsl.{Sink, Source}
import akka.http.scaladsl.Http
import akka.http.scaladsl.model._
import akka.http.scaladsl.settings.ConnectionPoolSettings
import akka.util.ByteString

import scala.util.{Success, Failure}
import scala.concurrent.{Await, Future}
import scala.concurrent.duration._

object Main extends App {
  implicit val system = ActorSystem("HttpClientSystem")
  implicit val materializer = ActorMaterializer()
  import system.dispatcher

  // Connection pool settings with retry support
  val poolSettings = ConnectionPoolSettings(system).withMaxRetries(3)

  val pool = Http().cachedHostConnectionPool[Unit]("127.0.0.1", 8080, settings = poolSettings)

  // Define the POST request
  val jsonPayload = """{"message": "hello"}"""
  val entity = HttpEntity(ContentTypes.`application/json`, ByteString(jsonPayload))

  val postRequest = HttpRequest(
    method = HttpMethods.POST,
    uri = "/",
    entity = entity
  )

  val requests = Source.single((postRequest, ()))

  val responseFuture: Future[Unit] =
    requests
      .via(pool)
      .runWith(Sink.head)
      .flatMap {
        case (Success(response), _) =>
          println(s"Got response: ${response.status}")
          response.entity.dataBytes.runForeach(chunk => println(chunk.utf8String)).map(_ => ())

        case (Failure(ex), _) =>
          println(s"Request failed: ${ex.getMessage}")
          Future.successful(())
      }
      .andThen { case _ => system.terminate() }

  Await.result(responseFuture, 30.seconds)
}

This time I see:

22:47:00.949906 IP 127.0.0.1.60013 > 127.0.0.1.8080: Flags [S], seq 3353561783, win 65535, options [mss 16344,nop,wscale 6,nop,nop,TS val 2697315485 ecr 0,sackOK,eol], length 0
22:47:00.949928 IP 127.0.0.1.60013 > 127.0.0.1.8080: Flags [S], seq 3353561783, win 65535, options [mss 16344,nop,wscale 6,nop,nop,TS val 2697315485 ecr 0,sackOK,eol], length 0

so immediate failure and this gets printed:

Request failed: Tcp command [Connect(127.0.0.1:8080,None,List(),Some(10 seconds),true)] failed because of java.net.ConnectException: Connection refused

If we can make the POST behavior mimic the GET behavior, then we can fully delegate all of our TCP layer to akka-http which would be awesome.