Http4s: AsyncHttpClient fails with the exception: received subscription in invalid state

[harpresing]

Background

I discovered this behaviour while updating the http4s version from 0.21.0 to 0.21.4.

This seems to have been introduced since the version 0.21.2.

Setup

Given I have two http4s clients

import org.http4s.client.Client
import org.http4s.client.jetty.JettyClient
import org.http4s.client.asynchttpclient.AsyncHttpClient
import cats.effect.{ContextShift, IO}
import scala.concurrent.ExecutionContext

private implicit val cs: ContextShift[IO] = IO.contextShift(ExecutionContext.global)
val asyncHttpClient: Client[IO] = AsyncHttpClient.allocate[IO]().unsafeRunSync()._1
val jettyHttpClient: Client[IO] = JettyClient.allocate[IO]().unsafeRunSync()._1

And I define a way to extract the body string from the response

  implicit class BodyOps(b: EntityBody[IO]) {
    def getContent: String = b.through(fs2.text.utf8Decode).compile.string.unsafeRunSync()
  }

Then I get the exception java.lang.Error: received subscription in invalid state with the AsyncHttpClient but not with the JettyClient when I try to convert the content of the response stream to a string.

  "Async Client" should "work as expected" in {
    val request = Request[IO](Method.GET, someUri)
    val response = asyncHttpClient.fetch[Response[IO]](request)(IO.pure).unsafeRunSync()

    response.status should be (Status.Ok)
    response.body.getContent should be ("some response") // Error
  }

Full stack trace:

java.lang.Error: received subscription in invalid state [Idle(com.typesafe.netty.HandlerPublisher$ChannelSubscription@cd1b01)]
    at fs2.interop.reactivestreams.StreamSubscriber$.$anonfun$fsm$1(StreamSubscriber.scala:103)
    at cats.effect.concurrent.Ref$SyncRef.spin$1(Ref.scala:249)
    at cats.effect.concurrent.Ref$SyncRef.$anonfun$modify$1(Ref.scala:253)
    at cats.effect.internals.IORunLoop$.cats$effect$internals$IORunLoop$$loop(IORunLoop.scala:87)
  | => eat cats.effect.internals.IORunLoop$.start(IORunLoop.scala:34)
    at cats.effect.IO.unsafeRunAsync(IO.scala:258)
    at cats.effect.IO.$anonfun$runAsync$1(IO.scala:178)
    at scala.runtime.java8.JFunction0$mcV$sp.apply(JFunction0$mcV$sp.java:12)
    at cats.effect.internals.IORunLoop$.step(IORunLoop.scala:190)
    at cats.effect.IO.unsafeRunTimed(IO.scala:321)
    at cats.effect.IO.unsafeRunSync(IO.scala:240)
    at cats.effect.SyncIO.unsafeRunSync(SyncIO.scala:51)
    at fs2.interop.reactivestreams.package$Runner.unsafeRunAsync(package.scala:71)
    at fs2.interop.reactivestreams.StreamSubscriber.onSubscribe(StreamSubscriber.scala:24)
    at com.typesafe.netty.HandlerPublisher.subscribe(HandlerPublisher.java:159)
    at org.http4s.client.asynchttpclient.AsyncHttpClient$$anon$1.$anonfun$onStream$2(AsyncHttpClient.scala:87)
    at scala.runtime.java8.JFunction0$mcV$sp.apply(JFunction0$mcV$sp.java:12)
    at cats.effect.internals.IORunLoop$.cats$effect$internals$IORunLoop$$loop(IORunLoop.scala:87)
    at cats.effect.internals.IORunLoop$.startCancelable(IORunLoop.scala:41)
    at cats.effect.internals.IOBracket$BracketStart.run(IOBracket.scala:88)
    at cats.effect.internals.Trampoline.cats$effect$internals$Trampoline$$immediateLoop(Trampoline.scala:67)
    at cats.effect.internals.Trampoline.startLoop(Trampoline.scala:35)
    at cats.effect.internals.TrampolineEC$JVMTrampoline.super$startLoop(TrampolineEC.scala:89)
    at cats.effect.internals.TrampolineEC$JVMTrampoline.$anonfun$startLoop$1(TrampolineEC.scala:89)
    at scala.runtime.java8.JFunction0$mcV$sp.apply(JFunction0$mcV$sp.java:12)
    at scala.concurrent.BlockContext$.withBlockContext(BlockContext.scala:81)
    at cats.effect.internals.TrampolineEC$JVMTrampoline.startLoop(TrampolineEC.scala:89)
    at cats.effect.internals.Trampoline.execute(Trampoline.scala:43)
    at cats.effect.internals.TrampolineEC.execute(TrampolineEC.scala:42)
    at cats.effect.internals.IOBracket$BracketStart.apply(IOBracket.scala:69)
    at cats.effect.internals.IOBracket$BracketStart.apply(IOBracket.scala:49)
    at cats.effect.internals.IORunLoop$.cats$effect$internals$IORunLoop$$loop(IORunLoop.scala:139)
    at cats.effect.internals.IORunLoop$.start(IORunLoop.scala:34)
    at cats.effect.internals.IOBracket$.$anonfun$apply$1(IOBracket.scala:42)
    at cats.effect.internals.IOBracket$.$anonfun$apply$1$adapted(IOBracket.scala:32)
    at cats.effect.internals.IORunLoop$RestartCallback.start(IORunLoop.scala:345)
    at cats.effect.internals.IORunLoop$.cats$effect$internals$IORunLoop$$loop(IORunLoop.scala:122)
    at cats.effect.internals.IORunLoop$.$anonfun$suspendAsync$1(IORunLoop.scala:258)
    at cats.effect.internals.IORunLoop$.$anonfun$suspendAsync$1$adapted(IORunLoop.scala:257)
    at cats.effect.internals.IORunLoop$RestartCallback.start(IORunLoop.scala:345)
    at cats.effect.internals.IORunLoop$.cats$effect$internals$IORunLoop$$loop(IORunLoop.scala:122)
    at cats.effect.internals.IORunLoop$.start(IORunLoop.scala:34)
    at cats.effect.IO.unsafeRunAsync(IO.scala:258)
    at cats.effect.internals.IOPlatform$.unsafeResync(IOPlatform.scala:38)
    at cats.effect.IO.unsafeRunTimed(IO.scala:325)
    at cats.effect.IO.unsafeRunSync(IO.scala:240)
    at example.Http4sClientSpec$BodyOps.getContent(Http4sClientSpec.scala:22)

[satorg]

I bet the issue here can be caused by the way how the Client.fetch method is used:

asyncHttpClient.fetch[Response[IO]](request)(IO.pure)

In short, it is not safe to attempt to get Response by simply enwrapping it into IO in the method callback.

There's an issue already (#3345) and a corresponding PR which deprecates this fetch because it's a very common mistake.

[rossabaker]

Yes, @satorg is right. But it will be possible to make an equivalent mistake post-deprecation:

client.run(req).use(IO.pure)

The problem is that when the IO passed fetch or use is complete, the connection is returned to the client's pool. You need to be sure to consume everything you need from the response body in that callback.

(All snippets below untested, but they're close...)

A concise way to do this is:

// returns an IO[String] on any 2xx response
// raises an UnexpectedStatusError on any non-2xx response
client.expect[String](req)

If you want to specifically handle OK:

client.run(req).use {
  case Ok(resp) => 
    resp.as[String] // return IO[String] if it's 200 OK
  case resp => 
    IO.raiseError(anyThrowableYouLike) // handles other statuses
} // This whole expression returns IO[String]

Let us know if that helps, or if you have any other questions. :smile:

[harpresing]

Thanks @satorg and @rossabaker for your help! 😄

So is it fair to say that the org.http4s.client.jetty.JettyClient isn't returning the underlying HTTP connection when the returned task completes since an attempt to read the response body afterward doesn't result in an error, which might be a potential problem since it isn't implementing the org.http4s.client.Client#fetch contract correctly?

[rossabaker]

I would say more that it's undefined behavior: we've returned the connection to Jetty, and Jetty's internal connection pooling can reuse it however it sees fit.

It would be nice if we could make it a compile time error to leak the stream out of fetch or .run(...).use, but the type system isn't powerful enough for that. If we can't do that, it would be nice if this usage pattern would consistently and immediately raise an error so you don't depend on undefined behavior, but we'd have to wrap all the streams, which would cost efficiency. So that's why we're in the current situation of undefined behavior, where it might work, but shouldn't ever be relied upon. It's a tradeoff.

[harpresing]

Interesting. Thanks for the explanation!