interleaveOrdered unexpected behavior

[aartigao]

Version

3.6.1

Bug description

While investigating an issue with fs2 and Kafka I found a strange behavior for interleaveOrdered. In the following code:

import cats.effect.std.Queue
import cats.effect.{IO, IOApp}
import fs2.Stream

import scala.concurrent.duration.DurationInt

object StreamInterleave extends IOApp.Simple {

  override def run: IO[Unit] = pures *> queues

  def pures = IO.println(Stream(1, 2, 3).interleaveOrdered(Stream.empty).toList)

  def queues =
    for {
      q1 <- Queue.unbounded[IO, Option[Int]]
      q2 <- Queue.unbounded[IO, Option[Int]]
      _ <- q1.tryOfferN(List(Some(1), Some(2), Some(3), None))
      _ <- q2.offer(None)
      r <- Stream.fromQueueNoneTerminated(q1).interleaveOrdered(Stream.fromQueueNoneTerminated(q2).debug()).debug().timeout(5.seconds).compile.toList
      _ <- IO.println(r)
    } yield ()

}

pures runs fine and prints

List(1, 2, 3)

whereas queues times out and prints nothing:

java.util.concurrent.TimeoutException: Timed out after 5 seconds
	at fs2.Stream.timeout(Stream.scala:2814)
	at com.stuart.routing.state.StreamInterleave$.$anonfun$queues$4(StreamInterleave.scala:21)
	at get @ fs2.internal.Scope.openScope(Scope.scala:281)
	at flatMap @ fs2.Compiler$Target.flatMap(Compiler.scala:163)
	at flatMap @ fs2.Pull$.$anonfun$compile$21(Pull.scala:1214)
	at update @ fs2.internal.Scope.releaseChildScope(Scope.scala:224)
	at flatMap @ fs2.Compiler$Target.flatMap(Compiler.scala:163)
	at flatMap @ fs2.Compiler$Target.flatMap(Compiler.scala:163)
	at flatMap @ fs2.Compiler$Target.flatMap(Compiler.scala:163)
	at flatMap @ fs2.Compiler$Target.flatMap(Compiler.scala:163)
	at flatMap @ fs2.Compiler$Target.flatMap(Compiler.scala:163)
	at flatMap @ fs2.Compiler$Target.flatMap(Compiler.scala:163)
	at modify @ fs2.internal.Scope.close(Scope.scala:262)
	at flatMap @ fs2.Compiler$Target.flatMap(Compiler.scala:163)
	at flatMap @ fs2.Pull$.$anonfun$compile$18(Pull.scala:1213)
	at handleErrorWith @ fs2.Compiler$Target.handleErrorWith(Compiler.scala:161)
	at flatMap @ fs2.Pull$.goCloseScope$1(Pull.scala:1199)
	at map @ fs2.internal.InterruptContext.$anonfun$eval$2(InterruptContext.scala:99)

Scastie

[armanbilge]

Ping @ivan-klass if you want to take a look :)

[aartigao]

Just in case someone is in the same situation as me, I've created a naive merger using queues:

https://gist.github.com/aartigao/43f321f4231c066daf5539ab8be9a1c4

I can imagine this is far from ideal from the performance POV, but it's working for my use case 🙃