Connection leak with http4s websocket subscriptions

[enriquerodbe]

Hi!

I have a cats-effect app with http4s blaze server and Caliban. This is a simplified version of how I instantiate the server:

Resource
  .eval(
    graphQL(RootResolver(queryResolver, mutationResolver, subscriptionResolver))
      .interpreterAsync[IO]
  )
  .flatMap { interpreter =>
    BlazeServerBuilder[IO]
      .bindHttp(port = config.port, host = config.host)
      .withHttpWebSocketApp { wsBuilder =>
        Router(
          "api/ws" -> Http4sAdapter.makeWebSocketServiceF(
            wsBuilder,
            WebSocketInterpreter(interpreter)
          )
        ).orNotFound
      }
      .resource
  }

Everything more or less like it's described in the documentation.

The problem is that when a client subscribes using a websocket connection, then the connection resource is never released, even when the client sends a "type": "stop" graphql message and a close message to the websocket channel.
When the server accepts enough connections and reaches the maxConnections limit, it stops accepting more connections, even when the rest are all idle (and even when clients explicitly tried to close the connection).

[kyri-petrou]

Hi @enriquerodbe, thanks for reporting this issue! I'm guessing since that you used "type":"stop" as the termination message, you're using the legacy WS protocol is that correct? To help us narrow the cause of it down, is it possible to check whether the issue also exists if the client connects using the graphql-ws protocol or is it not possible to configure the client to use it?

[enriquerodbe]

Thanks for the quick response! I'm testing manually with a UI client so I can try both. I just realized I was using a legacy protocol, but it seems that the outcome is the same. Just to be sure, the way to use the new protocol is to send header Sec-WebSocket-Protocol: graphql-transport-ws with the initial connection, right?
I was using caliban 2.5.0 and tried with 2.5.2 now and the result is the same.

[enriquerodbe]

Confirmed with the newer protocol using "type": "complete" I still get the same effect.

[kyri-petrou]

Just to be sure, the way to use the new protocol is to send header Sec-WebSocket-Protocol: graphql-transport-ws

@enriquerodbe Yeap that's the correct way. OK I'll take a look at what might be happening. Just to make sure that I replicate it correctly, do you use some specific automated / semi-automated way to identify that the connections aren't being closed or do you just open a bunch and close them manually?

[enriquerodbe]

My method is pretty rudimentary 😆 I just set blaze server maxActiveConnections to 1, and then I try to connect, disconnect, and connect again. The second time it just hangs.
I suppose I could write a test if needed.

[kyri-petrou]

@enriquerodbe I think the issue you're facing might be related to this recent fix in zio-interop-cats. I managed to replicate it with version 2.1.0.0 but works fine with 2.1.0.1. Can you try adding the following dependency and see if it resolves the issue?

"dev.zio" %% "zio-interop-cats" % "2.1.0.1"

[kyri-petrou]

Re-opening until fix is confirmed

[enriquerodbe]

It did not work for me. I notice you consistently write 2.1.0.1 but that version doesn't exist, I updated from 23.1.0.0 to 23.1.0.1

[enriquerodbe]

I minimized my problem to this:

package ws_test

import caliban.*
import caliban.interop.cats.implicits.*
import caliban.interop.tapir.WebSocketInterpreter
import caliban.schema.Schema
import cats.effect.std.Dispatcher
import cats.effect.{IO, IOApp, Resource}
import org.http4s.blaze.server.BlazeServerBuilder
import org.http4s.server.{Router, Server}
import sttp.tapir.json.circe.*
import zio.Runtime
import zio.stream.interop.fs2z.*
import zio.stream.{Stream, ZStream}

object Main extends IOApp.Simple {

  private final case class Subscription(test: Stream[Throwable, String])
  private final case class Query(test: String = "test query")
  private final case class Mutation(test: String = "test mutation")

  override def run: IO[Unit] = {
    given Runtime[Any] = Runtime.default

    val resource = for {
      given Dispatcher[IO] <- Dispatcher.parallel[IO]
      interpreter <- Resource.eval(buildInterpreter)
      server <- buildBlazeServer(interpreter)
    } yield server

    resource.useForever
  }

  private def buildInterpreter(using Runtime[Any]): IO[GraphQLInterpreter[Any, CalibanError]] = {
    given Schema[Any, Subscription] = Schema.gen
    given Schema[Any, Query] = Schema.gen
    given Schema[Any, Mutation] = Schema.gen
    val query = Query()
    val mutation = Mutation()
    val subscription = Subscription(ZStream("test1", "test2"))
    graphQL(RootResolver(query, mutation, subscription)).interpreterAsync[IO]
  }

  private def buildBlazeServer(
      interpreter: GraphQLInterpreter[Any, CalibanError]
  )(using Dispatcher[IO], Runtime[Any]): Resource[IO, Server] =
    BlazeServerBuilder[IO]
      .bindHttp(port = 8088, host = "0.0.0.0")
      .withHttpWebSocketApp { wsBuilder =>
        Router(
          "ws" -> Http4sAdapter.makeWebSocketServiceF(wsBuilder, WebSocketInterpreter(interpreter))
        ).orNotFound
      }
      .withMaxConnections(1)
      .resource
}

And this build.sbt:

name := "blaze-caliban"
scalaVersion := "3.3.1"

fork := true

libraryDependencies ++= Seq(
  "org.http4s" %% "http4s-blaze-server" % "0.23.16",
  "com.github.ghostdogpr" %% "caliban-http4s" % "2.5.2",
  "com.softwaremill.sttp.tapir" %% "tapir-json-circe" % "1.9.6",
  "ch.qos.logback" % "logback-classic" % "1.4.14"
)

dependencyOverrides += "dev.zio" %% "zio-interop-cats" % "23.1.0.1"

To reproduce the steps are:

1. Run the app with sbt run
2. Connect to ws using client (I'm using Insomnia)
3. Send connection_init and then subscribe messages as in the graphql WS protocol
4. Observe that I receive the two next messages with test strings and a complete message
5. Send another complete message just in case
6. (WS connection is still open) Close the websocket connection
7. Try to make another connection - Fails here, just hangs

[enriquerodbe]

My workaround was to switch to ember. It seems to be an issue with blaze.

[ghostdogpr]

Since it works with Ember, it is likely to be an issue with Blaze so I'm closing this.