Http4s: Response stream is not interrupted after mapping the effect type

[maximskripnik]

I am having troubles interrupting a stream which is sent in a response using http4s. It happens only if I try to convert the effect type of the routes:

_All imports are omitted to save the space. I am sorry for very fat code samples, but it's the best way I could narrow it down._

Here is my routes definition:

class TestController[F[_] : Concurrent : Timer](deferred: Deferred[F, Unit]) extends Http4sDsl[F] {

  private implicit val uuidEncoder: EntityEncoder[F, UUID] =
    EntityEncoder
        .stringEncoder[F]
        .contramap(_.toString + "\n")

  val routes: HttpRoutes[F] = HttpRoutes.of[F] {
    case GET -> Root / "stream" =>
      val stream = Stream
        .repeatEval(Concurrent[F].delay(UUID.randomUUID()).flatTap(_ => Timer[F].sleep(2.seconds)))
        .interruptWhen(deferred.get.attempt)
      Ok(stream)
    case POST -> Root / "complete" =>
      deferred.complete(()).flatMap(_ => Ok("stream is cancelled"))
  }

}

Here is how I instantiate those routes twice with multiple effects (IO[_] and EitherT[IO, String, _]) and then merge them into one route with final effect (IO[_]) and two prefixes:

object Routes {

  def createRoutes(
      deferred1: Deferred[IO, Unit],
      deferred2: Deferred[IO, Unit]
  )(implicit c: Concurrent[IO], timer: Timer[IO]): HttpRoutes[IO] = {

    type SpecialEff[T] = EitherT[IO, String, T]

    val ioToSpecial: IO ~> SpecialEff = EitherT.liftK
    val specialToIO: SpecialEff ~> IO =
      new ~>[SpecialEff, IO] {
        override def apply[A](fa: SpecialEff[A]): IO[A] = fa.value.flatMap {
          case Left(_)    => IO.raiseError(new RuntimeException("should never happen"))
          case Right(res) => IO.pure(res)
        }
      }

    val dsl = Http4sDsl[IO]
    import dsl._

    val ioTestRoutes: HttpRoutes[IO] = new TestController[IO](deferred1).routes
    val specialTestRoutes: HttpRoutes[SpecialEff] =
      new TestController[SpecialEff](deferred2.mapK(ioToSpecial)).routes

    val convertedSpecialRoutes: HttpRoutes[IO] = Kleisli { requestIO =>
      val requestSpecial =
        Request[SpecialEff]( // no mapK cause it uses Stream#translate for the body (which makes stream uninterruptible)
          method = requestIO.method,
          uri = requestIO.uri,
          httpVersion = requestIO.httpVersion,
          headers = requestIO.headers,
          body = requestIO.body.translateInterruptible(ioToSpecial),
          attributes = requestIO.attributes
        )

      val specialResult: OptionT[SpecialEff, Response[SpecialEff]] =
        specialTestRoutes(requestSpecial)

      val ioResult: IO[Option[Response[IO]]] = specialResult.value
          .map(_.map { specialResponse =>
            Response[IO]( // no mapK cause it uses Stream#translate for the body (which makes stream uninterruptible)
              status = specialResponse.status,
              httpVersion = specialResponse.httpVersion,
              headers = specialResponse.headers,
              body = specialResponse.body.translateInterruptible(specialToIO)
            )
          })
          .value
          .flatMap {
            case Right(maybeResp) =>
              IO.pure(maybeResp)
            case Left(error) =>
              BadRequest(error).map(Some(_))
          }

      OptionT(ioResult)
    }

    Router[IO]("/io" -> ioTestRoutes, "/special" -> convertedSpecialRoutes)
  }
}

I don't know why, but /io/complete requests interrupt the stream opened with /io/stream just fine, but /special/complete ones do not interrupt the stream opened with /special/stream. Even though in both cases the corresponding Deferred instance is set successfully. I suspect that somehow I made the response body stream uninterruptible while mapping SpecialEff to IO or vice versa.

In case if that is an expected behaviour, could you please point me in the right direction to fix this? I am not expert in fs2 streams, but I don't see any errors in the code.

Also, here is a sample project with this. It's just three scala files, two of which are above: https://github.com/maximskripnik/stream-cancel-issue

Thanks in advance!

[rossabaker]

Do you think this captures the essence of the problem? There's no HTTP in it, and it fails for me in the way your app does:

[Edit: previously pasted wrong version]

import java.util.UUID

import cats._
import cats.arrow._
import cats.data._
import cats.effect._
import cats.effect.concurrent.Deferred
import cats.syntax.applicativeError._
import cats.syntax.flatMap._
import fs2.Stream

import scala.concurrent.duration._

object WithoutHttp extends IOApp {
  type F[T] = EitherT[IO, String, T]
  val ioToSpecial: IO ~> F = EitherT.liftK
  val specialToIO: F ~> IO =
    new ~>[F, IO] {
      override def apply[A](fa: F[A]): IO[A] = fa.value.flatMap {
        case Left(_)    => IO.raiseError(new RuntimeException("should never happen"))
        case Right(res) => IO.pure(res)
      }
    }
  val F = Concurrent[F]

  def run(args: List[String]) = {
    val deferred = Deferred.unsafe[IO, Unit].mapK(ioToSpecial)
    val stream = Stream
      .repeatEval(F.delay(UUID.randomUUID()).flatTap(_ => Timer[F].sleep(2.seconds)))
      .evalMap(uuid => F.delay(println(uuid)))
      .interruptWhen(F.attempt(deferred.get))
      .translateInterruptible(specialToIO)
      .compile
      .drain

    for {
      fiber <- stream.start
      _ <- Timer[IO].sleep(5.seconds)
      _ <- specialToIO(deferred.complete())
      _ <- fiber.join
    } yield ExitCode.Success
  }
}

[maximskripnik]

@rossabaker Yes, I think it really does capture the same issue. It works for IO[_] and does not work for EitherT[IO, String, _] 😞

Thanks for narrowing it down to fs2 or cats. It's clear now that this is not http4s issue

[rossabaker]

I asked our resident fs2 experts on Gitter to see if they can spot it, and once we find the answer, it might be an interesting problem to wind it back into http4s. This one has my interest, too. :smile:

[rossabaker]

The dependencies are a bit old in the example repo, but I confirmed it's happening on http4s-0.21.9 and its transitive dependencies, which should be the latest fs2-2.4.

[mpilquist]

Not sure it's possible in this case, but one workaround is calling interruptAfter/interruptWhen after calling transalteInterrupt.

[maximskripnik]

@mpilquist Thank your for your attention to this. Unfortunately, our use case implies calling interruptWhen before mapping the effect type.
The idea is that we use one effect that is capable of capturing non-throwable errors (EitherT) for our business logic which includes creating the stream and calling interruptWhen on it. And _almost at the end of the world_ we map this effect to IO. The way we map it is a bit more complicated than in the provided example, but this forces us to handle all the possible errors that might be returned by classes that implement the business logic (and work over EitherT). Calling interruptWhen after translateInterrupt means that we move part of the business logic to one of the outermost layers. This code is also quite complicated with all the Deferred instances and other concurrency primitives we use to track the state, so it's very preferred to keep it somewhere in the innermost layers

[mpilquist]

@maximskripnik Can you give fs2 version 2.5-bdc810b a try and let me know whether that resolves this issue for you?

[maximskripnik]

@mpilquist Yes, the issue is not reproducible with "co.fs2" %% "fs2-core" % "2.5-bdc810b". The sample provided by @rossabaker also works as expected with this version.
Thanks!

[mpilquist]

@maximskripnik Cool, I released fs2 2.4.6 with that fix.