http4s client tries to re-run body stream, possibly uses disposed connection

[fiadliel]

_executeRequest is defined like this:

    def _executeRequest[T, U](
      method: String,
      path: Seq[String],
      queryParameters: Seq[(String, String)] = Nil,
      requestHeaders: Seq[(String, String)] = Nil,
      body: Option[T] = None
    )(handler: org.http4s.Response[F] => F[U]
    )(implicit encoder: io.circe.Encoder[T]): F[U] = {

and uses the fetch method of the http4s client:

      httpClient.fetch(modifyRequest(authBody))(handler)

If we look at the scaladoc for fetch:

def fetch[A](req: Request[F])(f: (Response[F]) ⇒ F[A]): F[A]
Submits a request, and provides a callback to process the response.

req
The request to submit

f
A callback for the response to req. The underlying HTTP connection is disposed when the returned task completes. Attempts to read the response body afterward will result in an error.

returns
The result of applying f to the response to req

This means you should not access the body after this point in the code.

However, when an expected error occurs, the response is given back, including a call which tries to parse JSON in a lazy val. This is fundamentally unsafe for two reasons:

1. the connection could already have been disposed before the lazy val is evaluated (which is outside of the fetch block).
2. if the body has already been consumed, in a streaming situation, you cannot restart streaming. This might not have been done in some cases, but it's a fiddly thing to keep track of.

[mchimirev]

We ran into a case, most likely caused by this issue, where when we try to decode a large (49,084 characters) response we get an exception:
java.lang.RuntimeException: org.http4s.InvalidBodyException: Received premature EOF.

Function looks like this:

  override def getOrdersByOrderId(_req: Request[IO], organization: com.hbc.mobile.common.models.v0.models.Organization, orderId: Long, billingZipCode: String): IO[GetOrdersByOrderIdResponse] = {
    val response: IO[GetOrdersByOrderIdResponse] = client.accounts.getOrdersByOrderId(orderId, billingZipCode, getSessionId(_req)).map { accountOrderTrackingResponseWrapper =>
      GetOrdersByOrderIdResponse.HTTP200(accountOrderTrackingResponseWrapper.response.results.order.map(toAccountOrder).getOrElse {
        throw new RuntimeException("No order present")
      })
    }
    response.recoverWith {
      case wrapperResponse: client.errors.AccountOrderStatusResponseWrapperResponse => {
        wrapperResponse.response.status.code match {
          case 404 => {
            wrapperResponse.accountOrderStatusResponseWrapper.attempt.map {
              case Left(e) =>
                throw new RuntimeException(e)
              case Right(accountOrderStatusResponseWrapper) =>
                GetOrdersByOrderIdResponse.HTTP422(ValidationErrors(cleanupValidationErrors(accountOrderStatusResponseWrapper.errors.map(toValidationError))))
            }
          }
          case undefined =>
            throw new RuntimeException(s"undefined response code ${undefined}")
        }
      }
    }
  }

And we are getting a 404 back but we are not able to decode the response.

@amarrella If you could look at it tomorrow, that would be great, it's blocking some of our work.

[amarrella]

The best thing would be to return a fs2.Stream in all the functions, but that's a pretty big breaking change.
Second best thing would be to do something like this in _executeRequest:

httpClient
        .streaming(modifyRequest(authBody))(response => fs2.Stream.emit(handler(response)))
        .compile.foldMonoid.flatten

but this would add the requirement for F[U] to be a monoid (even if we build the proof in the generated code, this means that we need to require U to be a monoid at least). Again, this would be a breaking change.

I think the only option we have for now is replacing the fetch call with this:

      for {
        disposableResponse <- httpClient.open(request)
        handledResponse <- handler(disposableResponse.response)
        _ <- Sync[F].delay(disposableResponse.dispose)
      } yield handledResponse

@fiadliel what do you think?

[amarrella]

OK, bad news @mchimirev. After talking with @fiadliel we came to the conclusion that client is broken in this case and the only way to fix it is a pretty big breaking change (the ones suggested above won't solve the issue). My suggestion for now is to unfortunately not use the client, do the client manually and use the apibuilder ModelsJson.

@gheine @fiadliel any suggestion on how to move forward with this?

[gheine]

So the problem is confined to non-unit error responses and their lazily parsing the response body. A quick fix could be to make the error type non-lazy, ie. deserialize when the error response case class is constructed, similar to what happens when non-error response types are deserialized.

[gheine]

ie change from lazy val to val in:

case class ValidationErrorsResponse(
      response: org.http4s.Response[F],
      message: Option[String] = None
    ) extends Exception(message.getOrElse(response.status.code + ": " + response.body)){
      lazy val validationErrors = _root_.com.foo.v0.Client.parseJson[F, com.foo.v0.models.ValidationErrors]("com.foo.v0.models.ValidationErrors", response)
    }

[mchimirev]

Tested using mobile-middleware (see branch issue-391-fix), and every thing is working correctly. I especially love not having to .attempt. response from the client.

For example, this was old code to handle 409:

    response.recoverWith {
      case wrapperResponse: client.errors.UserMessageResponse => {
        wrapperResponse.response.status.code match {
          case 409 => {
            wrapperResponse.userMessage.attempt.map {
              case Left(e) =>
                throw new RuntimeException(e)
              case Right(userMessage) =>
                GetResponse.HTTP409(toUserMessage(userMessage))
            }
          }
          case undefined =>
            throw new RuntimeException(s"undefined response code ${undefined}")
        }
      }
    }

And this is new code:

    response.recover {
      case wrapperResponse: client.errors.UserMessageResponse => {
        wrapperResponse.response.status.code match {
          case 409 => {
            GetResponse.HTTP409(toUserMessage(wrapperResponse.userMessage))
          }
          case undefined =>
            throw new RuntimeException(s"undefined response code ${undefined}")
        }
      }
    }