Upgrade to zio 2.0.0-RC5, zio http to 2.0.0-RC7

[gavares]

No description provided.

[adamw]

I'm afraid the given memory setting will cause a container out of memory (killing the process), as the VMs that github actions give us have 7GB RAM total. So the heap must be somewhat smaller than that, but let's see :)

[gavares]

At least it's giving compile errors now. I'll see about fixing those.

[gavares]

It appears this change is likely bigger than I expected. I think that sttp-shared needs to have it's zio version upgraded before tapir can be upgraded. I tried upgrading to the latest sttp-shared which depends on ZIO 2.0.0-RC3 but I'm getting macro errors in Endpoint.scala which is a little more than I'm currently able to tackle. I can look into an MR for sttp-shared to upgrade to the latest ZIO version and see how far I can get with that.

[gavares]

After forking sttp-shared I noticed that it has already been upgraded to ZIO 2.0.0-RC5. I guess we just need a release of sttp-shared that we can pull in to tapir.

[adamw]

@gavares sttp-shared 1.3.4 released :)

[gavares]

There appear to be some errors compiling macros and I'm a little out of my depth with macros in general. @adamw do you have any intuition about errors like this:

[error] /Users/ggavares/workspace/tapir/core/src/main/scala/sttp/tapir/EndpointIO.scala:10:19: encountered unrecoverable cycle resolving import.
[error] Note: this is often due in part to a class depending on a definition nested within its companion.
[error] If applicable, you may wish to try moving some members into another object.
[error] import sttp.tapir.macros.{EndpointInputMacros, EndpointOutputMacros, EndpointTransputMacros}
[error]                   ^
[error] /Users/ggavares/workspace/tapir/core/src/main/scala/sttp/tapir/EndpointIO.scala:47:42: not found: type EndpointTransputMacros
[error] sealed trait EndpointTransput[T] extends EndpointTransputMacros[T] {
[error]                                          ^
[error] /Users/ggavares/workspace/tapir/core/src/main/scala/sttp/tapir/Endpoint.scala:9:19: encountered unrecoverable cycle resolving import.
[error] Note: this is often due in part to a class depending on a definition nested within its companion.
[error] If applicable, you may wish to try moving some members into another object.
[error] import sttp.tapir.macros.{EndpointErrorOutputsMacros, EndpointInputsMacros, EndpointOutputsMacros, EndpointSecurityInputsMacros}
[error]                   ^
[error] /Users/ggavares/workspace/tapir/core/src/main/scala/sttp/tapir/Endpoint.scala:143:55: not found: type EndpointErrorOutputsMacros
[error] trait EndpointErrorOutputsOps[A, I, E, O, -R] extends EndpointErrorOutputsMacros[A, I, E, O, R] {
[error]                                                       ^
[error] /Users/ggavares/workspace/tapir/core/src/main/scala/sttp/tapir/Endpoint.scala:103:49: not found: type EndpointInputsMacros
[error] trait EndpointInputsOps[A, I, E, O, -R] extends EndpointInputsMacros[A, I, E, O, R] {
[error]                                                 ^
[error] /Users/ggavares/workspace/tapir/core/src/main/scala/sttp/tapir/Endpoint.scala:246:50: not found: type EndpointOutputsMacros
[error] trait EndpointOutputsOps[A, I, E, O, -R] extends EndpointOutputsMacros[A, I, E, O, R] {
[error]                                                  ^
[error] /Users/ggavares/workspace/tapir/core/src/main/scala/sttp/tapir/Endpoint.scala:82:57: not found: type EndpointSecurityInputsMacros
[error] trait EndpointSecurityInputsOps[A, I, E, O, -R] extends EndpointSecurityInputsMacros[A, I, E, O, R] {
[error]                                                         ^

[adamw]

@gavares yeah this happens when there are changes in core, core/clean helps :)

[gavares]

Yep that fixed it. I realized I also need to upgrade sttp to ZIO RC5. Working on a PR for that now.

[gavares]

This update is now waiting on: softwaremill/sttp#1410

[onsoup42]

@gavares I believe all the dependencies for this has been updated. Will you up date this PR? Otherwise I'd be happy to put up a PR.

[gavares]

Hi @onsoup42 , I've pushed up my latest changes. Feel free to help out. I've been working in my spare cycles on this MR but have been a little too busy to give it the attention I'd like the last 4-5 days.

I left off trying to get zioHttpServer/Test/compile to work. I'm having some issue with toResource in ZioHttpTestServerInterpreter.scala

[gavares]

I wonder if the updates to zio-cats-interop might be contributing to the issues I'm seeing. Basically the implicit conversion from a ZIO to a Resource isn't being resolved in the test classes.

[adamw]

@gavares I've fixed resource usages with Resource.scoped (thanks to help from zio discord)

though now I'm getting (when running the tests):

[info] sttp.tapir.server.ziohttp.ZioHttpServerTest *** ABORTED *** (2 milliseconds)
[info]   java.lang.NoSuchMethodError: 'zio.ZLayer zio.ZIO.toLayer(zio.package$Tag, java.lang.Object)'
[info]   at zhttp.service.server.ServerChannelFactory$.auto(ServerChannelFactory.scala:15)
[info]   at sttp.tapir.server.ziohttp.ZioHttpServerTest.$anonfun$tests$1(ZioHttpServerTest.scala:20)
[info]   at cats.effect.kernel.Resource.$anonfun$allocatedCase$2(Resource.scala:421)
[info]   at cats.effect.IOFiber.succeeded(IOFiber.scala:1185)
[info]   at cats.effect.IOFiber.runLoop(IOFiber.scala:366)
[info]   at cats.effect.IOFiber.execR(IOFiber.scala:1332)
[info]   at cats.effect.IOFiber.run(IOFiber.scala:139)
[info]   at cats.effect.unsafe.WorkerThread.run(WorkerThread.scala:523)
[info]   ...

I think we'll need to wait for zio http which depends on RC5 (see zio/zio-http#1217)

[gavares]

Thanks @adamw

After reading your Resource.scoped change I feel like I really should have thought of that myself.

[adamw]

@gavares I didn't as well, that's a tip I got from discord :) though I guess it falls in line nicely with ZIO.scoped

[gavares]

zio-http R7 released. I'll submit an update in the next couple hours.

[adamw]

I think scalajs is transitively updated here so we'll have to wait for scalatest/scalatest#2116 as well

[adamw]

@gavares zio http is already updated :)

[gavares]

You're too fast @adamw :)

[gavares]

Just pushed another change to fix some js compile errors

[sandeepkota]

Can't wait for this to be merged, my team is waiting for this is so badly.

[adamw]

@sandeepkota I'd like to as well, but there's something wrong with the tests, they hang (the JVM build gets cancelled). So we'll need to investigate and see what's happening.

[adamw]

It's the zio-http4s tests that hang - specifically, the web socket ones. And they only hang, if streaming tests are run beforehand ... this is going to be "fun" to debug ;)

[gavares]

Did the zio-http RC7 release have an issue related to websockets as well? This sounds a bit different but websockets seems to cause issues in both.

I'll have some time to take a look later today hopefully.

[adamw]

@gavares no, it's only the http4s-zio backend. Must be something in zio/zio-streams/zio-cats-interop 2 as the zio1 version works fine.

[gavares]

I split each of the ServerWebsocketTests out so I could run them in isolation. The tests that are failing to terminate are:

• "json client-terminated echo"
• "string server-terminated echo"

These are the only two tests which use functionToPipe in their definition.

They seem to work when run in isolation (no other tests are run via testOnly) but fail when they are run after any other tests in ServerWebsocketTests. All the other tests run just fine when run with the other tests in the class. (see comments at the end of scala code pasted in my next message).

Basically, including test3 or test4 with any other tests causes the jvm to hang.

If test3 is run before test1, then it's test1 that hangs.

[gavares]

package sttp.tapir.server.tests

import cats.effect.IO
import cats.syntax.all._
import io.circe.generic.auto._
import org.scalatest.matchers.should.Matchers._
import sttp.capabilities.{Streams, WebSockets}
import sttp.client3._
import sttp.monad.MonadError
import sttp.tapir._
import sttp.tapir.generic.auto._
import sttp.tapir.json.circe._
import sttp.tapir.server.interceptor.CustomiseInterceptors
import sttp.tapir.server.interceptor.metrics.MetricsRequestInterceptor
import sttp.tapir.server.tests.ServerMetricsTest._
import sttp.tapir.tests.Test
import sttp.tapir.tests.data.Fruit
import sttp.ws.{WebSocket, WebSocketFrame}

abstract class ServerWebSocketTests[F[_], S <: Streams[S], OPTIONS, ROUTE](
    createServerTest: CreateServerTest[F, S with WebSockets, OPTIONS, ROUTE],
    val streams: S
)(implicit
    m: MonadError[F]
) {
  import createServerTest._

  def functionToPipe[A, B](f: A => B): streams.Pipe[A, B]
  def emptyPipe[A, B]: streams.Pipe[A, B]

  private def stringWs = webSocketBody[String, CodecFormat.TextPlain, String, CodecFormat.TextPlain].apply(streams)
  private def stringEcho = functionToPipe((s: String) => s"echo: $s")

  val test1 = testServer(endpoint.out(stringWs), "string client-terminated echo")((_: Unit) => pureResult(stringEcho.asRight[Unit])) {
    (backend, baseUri) =>
      basicRequest
        .response(asWebSocket { (ws: WebSocket[IO]) =>
          for {
            _ <- ws.sendText("test1")
            _ <- ws.sendText("test2")
            m1 <- ws.receiveText()
            m2 <- ws.receiveText()
          } yield List(m1, m2)
        })
        .get(baseUri.scheme("ws"))
        .send(backend)
        .map(_.body shouldBe Right(List("echo: test1", "echo: test2")))
  }

  val test2 = {

    val reqCounter = newRequestCounter[F]
    val resCounter = newResponseCounter[F]
    val metrics = new MetricsRequestInterceptor[F](List(reqCounter, resCounter), Seq.empty)

    testServer(
      endpoint.out(stringWs).name("metrics"),
      interceptors = (ci: CustomiseInterceptors[F, OPTIONS]) => ci.metricsInterceptor(metrics)
    )((_: Unit) => pureResult(stringEcho.asRight[Unit])) { (backend, baseUri) =>
      basicRequest
        .response(asWebSocket { (ws: WebSocket[IO]) =>
          for {
            _ <- ws.sendText("test1")
            m <- ws.receiveText()
          } yield List(m)
        })
        .get(baseUri.scheme("ws"))
        .send(backend)
        .map { r =>
          r.body shouldBe Right(List("echo: test1"))
          reqCounter.metric.value.get() shouldBe 1
          resCounter.metric.value.get() shouldBe 1
        }
    }
  }

  val test3 =
    testServer(endpoint.out(webSocketBody[Fruit, CodecFormat.Json, Fruit, CodecFormat.Json](streams)), "json client-terminated echo")(
      (_: Unit) => pureResult(functionToPipe((f: Fruit) => Fruit(s"echo: ${f.f}")).asRight[Unit])
    ) { (backend, baseUri) =>
      basicRequest
        .response(asWebSocket { (ws: WebSocket[IO]) =>
          for {
            _ <- ws.sendText("""{"f":"apple"}""")
            _ <- ws.sendText("""{"f":"orange"}""")
            m1 <- ws.receiveText()
            m2 <- ws.receiveText()
          } yield List(m1, m2)
        })
        .get(baseUri.scheme("ws"))
        .send(backend)
        .map(_.body shouldBe Right(List("""{"f":"echo: apple"}""", """{"f":"echo: orange"}""")))
    }

  val test4 =
    testServer(
      endpoint.out(webSocketBody[String, CodecFormat.TextPlain, Option[String], CodecFormat.TextPlain](streams)),
      "string server-terminated echo"
    )((_: Unit) =>
      pureResult(functionToPipe[String, Option[String]] {
        case "end" => None
        case msg   => Some(s"echo: $msg")
      }.asRight[Unit])
    ) { (backend, baseUri) =>
      basicRequest
        .response(asWebSocket { (ws: WebSocket[IO]) =>
          for {
            _ <- ws.sendText("test1")
            _ <- ws.sendText("test2")
            _ <- ws.sendText("end")
            m1 <- ws.eitherClose(ws.receiveText())
            m2 <- ws.eitherClose(ws.receiveText())
            m3 <- ws.eitherClose(ws.receiveText())
          } yield List(m1, m2, m3)
        })
        .get(baseUri.scheme("ws"))
        .send(backend)
        .map(
          _.body.map(_.map(_.left.map(_.statusCode))) shouldBe Right(
            List(Right("echo: test1"), Right("echo: test2"), Left(WebSocketFrame.close.statusCode))
          )
        )
    }

  val test5 =
    testServer(
      endpoint.out(webSocketBody[String, CodecFormat.TextPlain, String, CodecFormat.TextPlain](streams)),
      "empty client stream"
    )((_: Unit) => pureResult(emptyPipe.asRight[Unit])) { (backend, baseUri) =>
      basicRequest
        .response(asWebSocketAlways { (ws: WebSocket[IO]) => ws.eitherClose(ws.receiveText()) })
        .get(baseUri.scheme("ws"))
        .send(backend)
        .map(_.body.left.map(_.statusCode) shouldBe Left(WebSocketFrame.close.statusCode))
    }

  val test6 = testServer(
    endpoint
      .in(isWebSocket)
      .errorOut(stringBody)
      .out(stringWs),
    "non web-socket request"
  )(isWS => if (isWS) pureResult(stringEcho.asRight) else pureResult("Not a WS!".asLeft)) { (backend, baseUri) =>
    basicRequest
      .response(asString)
      .get(baseUri.scheme("http"))
      .send(backend)
      .map(_.body shouldBe Left("Not a WS!"))
  }

  /**   
   *   - works: 3
   *   - works: 4
   *   - works: 1,5
   *   - works: 1,2,6
   *   - works: 1,2,5,6
   *   - DOESN't WORK: 1,3
    *  - DOESN't WORK: 1,4
    *  - DOESN'T WORK: 1,2,3
    *  - DOESN'T WORK: 2,3
    *  - DOESN'T WORK: 2,4
    */
  def tests(): List[Test] = List(
//    test1,
//    test2,
//    test3
    test4
//    test5
//    test6
  )

  // TODO: tests for ping/pong (control frames handling)
}

[gavares]

After adding some logging, it's pretty clear the server is being shutdown before the 2nd read in the the tests mentioned above is executing. If we modify test3 in my code above to look like this:

val test3 =
    testServer(endpoint.out(webSocketBody[Fruit, CodecFormat.Json, Fruit, CodecFormat.Json](streams)), "json client-terminated echo")(
      (_: Unit) => pureResult(functionToPipe((f: Fruit) => Fruit(s"echo: ${f.f}")).asRight[Unit])
    ) { (backend, baseUri) =>
      basicRequest
        .response(asWebSocket { (ws: WebSocket[IO]) =>
          for {
            _ <- IO.consoleForIO.println("Sending apple")
            _ <- ws.sendText("""{"f":"apple"}""")
            _ <- IO.consoleForIO.println("Sending orange")
            _ <- ws.sendText("""{"f":"orange"}""")
            _ <- IO.consoleForIO.println("reading apple")
            m1 <- ws.receiveText(false)
            _ <- IO.consoleForIO.println("reading orange")
            m2 <- ws.receiveText()
          } yield List(m1) // List(m1, m2)
        })
        .get(baseUri.scheme("ws"))
        .send(backend)
        .map(_.body shouldBe Right(List("""{"f":"echo: apple"}""", """{"f":"echo: orange"}""")))
    }

The logs show the server is shutdown before we log reading orange:

Sending apple
Sending orange
reading apple
2022-04-24 14:06:08,823 [scala-execution-context-global-519] DEBUG o.h.b.s.Http1ServerStage$$anon$1 - Shutting down HttpPipeline
2022-04-24 14:06:08,825 [scala-execution-context-global-519] DEBUG o.h.b.s.Http1ServerStage$$anon$1 - Shutting down.
2022-04-24 14:06:08,825 [scala-execution-context-global-519] DEBUG o.http4s.blazecore.IdleTimeoutStage - Stage IdleTimeoutStage sending inbound command: Connected
2022-04-24 14:06:08,825 [scala-execution-context-global-519] DEBUG o.h.blaze.server.WebSocketDecoder - Starting up.
2022-04-24 14:06:08,825 [scala-execution-context-global-519] DEBUG o.h.blaze.server.WebSocketDecoder - Stage WebSocketDecoder sending inbound command: Connected
2022-04-24 14:06:08,825 [scala-execution-context-global-519] DEBUG o.h.blaze.server.WSFrameAggregator - Starting up.
2022-04-24 14:06:08,825 [scala-execution-context-global-519] DEBUG o.h.blaze.server.WSFrameAggregator - Stage WSFrameAggregator sending inbound command: Connected
2022-04-24 14:06:08,827 [scala-execution-context-global-519] DEBUG o.h.b.websocket.Http4sWSStage - Starting up.
reading orange

I suspect there is some shared state/resource between the test services instances since each test succeeds when run independently and changing the order of the problematic test always allows the first test to succeed while subsequent tests hang indefinitely.

[adamw]

Thanks for the debug! :) I've narrowed it down further to running:

    import zio.stream.interop.fs2z._
    zio.Runtime.global.unsafeRunSync(
      fs2.Stream("pen pineapple apple pen".getBytes: _*).toZStream().runCollect
    )

plus a single web socket test. I think the server isn't shut down prematurely, it's the way http4s logs switching from handling http traffic to handling WS traffic.

[adamw]

@gavares I think I minimised it further, see: zio/interop-cats#533