Http4s: Ember client error when polling an url

[BalmungSan]

I am getting the below exception when using an ember client to constantly fetch data from AWS SQS using the HTTP API.

org.http4s.ember.core.EmberException$ParseError: Incomplete Header received (sz = 0): Right()

The error is somewhat sporadic, since sometimes it happens almost immediately and sometimes it takes roughly a minute.
But, after a couple of tries, it had never lasted more than 2 minutes before dying with the error.
Whereas, if using any other client it keeps running without any problem _(expect with the netty client, more info below)_.

I am using this library for polling SQS.
As such, I am not sure if the problem is with the library or with ember _(@d2a4u you may want to take a look to this)_; but since it doesn't happen with almost all other clients, it feels it should be something with ember.
Also, it means I can not create a reproducible example which doesn't involve the library and SQS _(sorry!)_

Ammonite script to reproduce the problem.

// scala 2.13.3

import $repo.`https://dl.bintray.com/d2a4u/sqs4s`

import $ivy.`org.typelevel::cats-effect:2.1.4`
import $ivy.`co.fs2::fs2-core:2.4.2`
import $ivy.`co.fs2::fs2-io:2.4.2`
import $ivy.`org.http4s::http4s-core:0.21.6`
import $ivy.`org.http4s::http4s-async-http-client:0.21.6`
import $ivy.`org.http4s::http4s-blaze-client:0.21.6`
import $ivy.`org.http4s::http4s-ember-client:0.21.6`
import $ivy.`org.http4s::http4s-jdk-http-client:0.3.0`
import $ivy.`org.http4s::http4s-jetty-client:0.21.6`
import $ivy.`org.http4s::http4s-netty-client:0.1.0`
import $ivy.`org.http4s::http4s-okhttp-client:0.21.6`
import $ivy.`io.sqs4s::sqs4s-native:1.0.0`
import $ivy.`io.chrisdavenport::log4cats-slf4j:1.1.1`
import $ivy.`ch.qos.logback:logback-classic:1.2.3`

import cats.effect.{Blocker, IO, Resource}
import fs2.Stream
import fs2.io.tls.TLSContext
import io.chrisdavenport.log4cats.slf4j.Slf4jLogger
import org.http4s.client.JavaNetClientBuilder
import org.http4s.client.asynchttpclient.AsyncHttpClient
import org.http4s.client.blaze.BlazeClientBuilder
import org.http4s.client.jdkhttpclient.JdkHttpClient
import org.http4s.client.jetty.JettyClient
import org.http4s.client.okhttp.OkHttpBuilder
import org.http4s.ember.client.EmberClientBuilder
import org.http4s.netty.client.NettyClientBuilder
import org.http4s.syntax.literals._

import sqs4s.serialization.instances._
import sqs4s.api.{AwsAuth, SqsSettings}
import sqs4s.api.lo.ReceiveMessage

import org.eclipse.jetty.client.{HttpClient => JettyHttpClient}
import org.eclipse.jetty.util.ssl.SslContextFactory.{Client => JettyHttpClientSslContext}

import scala.concurrent.duration._

object Main {
  implicit val cs = IO.contextShift(scala.concurrent.ExecutionContext.global)
  implicit val timer = IO.timer(scala.concurrent.ExecutionContext.global)

  // Clients ------------------------------------------------------------------
  val clientNet = for {
    blocker <- Blocker[IO]
    client <- JavaNetClientBuilder[IO](blocker).resource
  } yield client

  val clientJdk = Resource.liftF(JdkHttpClient.simple[IO])

  val clientAsync = AsyncHttpClient.resource[IO]()

  val clientOkHttp = for {
    blocker <- Blocker[IO]
    builder <- OkHttpBuilder.withDefaultClient[IO](blocker)
    client <- builder.resource
  } yield client

  val clientJetty = JettyClient.resource[IO](new JettyHttpClient(new JettyHttpClientSslContext()))

  val clientNetty = NettyClientBuilder[IO].withDefaultSSLContext.withIdleTimeout(3.seconds).resource

  val clientBlaze = BlazeClientBuilder[IO](scala.concurrent.ExecutionContext.global).resource

  val clientEmber = for {
    blocker <- Blocker[IO]
    logger <- Resource.liftF(Slf4jLogger.create[IO])
    tlsContext <- Resource.liftF(TLSContext.system[IO](blocker))
    client <- EmberClientBuilder.default[IO].withTLSContext(tlsContext).withBlocker(blocker).withLogger(logger).withTimeout(3.seconds).build
  } yield client
  // --------------------------------------------------------------------------

  // Here change the client to use.
  val events = Stream.resource(clientNetty).flatMap { client =>
    def logData(data: String): IO[Unit] = IO {
      println("-------------------- DATA --------------------")
      println(data)
      println("-------------------- DATA --------------------")
    }

    val config = SqsSettings(
      queue = uri"your/sqs/uri",
      auth = AwsAuth(
        accessKey = "YOUR_ACCESS_KEY",
        secretKey = "YOUR_SECRET_KEY",
        region = "YOUR_AWS_REGION"
      )
    )

    val request = client.expect[String](ReceiveMessage[IO, String]().mkRequest(config))

    Stream.repeatEval(request).evalMap(logData).metered(2.seconds)
  }

  val program = events.compile.drain

  def run(): Unit = {
    program.unsafeRunSync()
  }
}

Main.run()

---

Additional information:

• Scala version: 2.13.3
• cats-effect version: 2.1.4
• fs2 version: 2.4.2
• http4s version: 0.21.6
• Java version: OpenJDK GraalVM CE 20.1.0 (build 11.0.7+10-jvmci-20.1-b02)
• OS: WSL 2 - Ubuntu 20.04 LTS (4.19.104-microsoft-standard)

---

Let me know if I can do something else to help debug this problem.

---

Edit

Updated the snippet to use more clients.

Netty

For some reason, the netty client fails immediately after doing the first request with a 403 _(Forbidden)_.
Apparently, because the request wasn't properly signed.

This is probably an error either with the library or with how netty does the request; since this doesn't happen with other clients _(not even ember)_. So, I am pretty sure it is not related, but I guess I lose nothing by mentioning it here too.

[d2a4u]

hi, I have always used Blaze and never tried Ember. The error looks similar with this issue, however, it should be fixed in 0.21.6. I will also have a look and see if I can reproduce the error.

[d2a4u]

I've tried some stuff to get more info about this issue so just want to give some updates on my findings. The code I have used to reproduce this error:

Stream.repeatEval(
    client.expect[String](ReceiveMessage[IO, String]().mkRequest(config))
).evalMap(stt => IO(println(stt))).metered(2.seconds)

ReceiveMessage[IO, String]().mkRequest(config) just returns F[Request[F]]. So the snippet above it just sends a request repeatedly every 2 seconds. However, if I use client.status, it works perfectly fine, so I think there is an issue with the way Ember handles the response.

I also had a look into TCP level with wireshark, the last TCP package right before this error occurs is when AWS server sends a [FIN, ACK] to terminate the connection. This happens to Blaze as well but Blaze creates a new connection straight away but Ember just crashes.

[BalmungSan]

@d2a4u would you be able to join the http4s gitter channel? So we can discuss this more quickly than in issue comments.

[ChristopherDavenport]

Closed By #3598