Some streams merged with `merge` stop emitting when downstream does certain actions

[lukestephenson]

Hello, I came across a bug in an application of mine where we stop seeing updates on one stream when multiple streams are merged with mergeHaltBoth. I've come up with a fairly minimal reproduction of the issue. Our application will sometimes inspect the first element of the stream to determine how to process the remaining elements of the stream. We make use of the splitHead function used in this example.

Here is the code which demonstrates this issue:

package bug

import cats.effect.{IO, IOApp}
import fs2.{Chunk, Stream}

import scala.concurrent.duration.DurationInt

object FS2Bug extends IOApp.Simple {

  sealed trait Data

  case class Item(value: Int) extends Data
  case object Tick1 extends Data
  case object Tick2 extends Data

  override def run: IO[Unit] = {
    val source = Stream.emits(1 to 100).evalMap(i => IO(Item(i)).delayBy(2.seconds))

    val timer = fs2.Stream.awakeEvery[IO](1.second).evalMap(_ => IO(()).as(Tick1))
    val timer2 = fs2.Stream.awakeEvery[IO](1.second).evalMap(_ => IO(()).as(Tick2))

    val sources = timer2.mergeHaltBoth(source.mergeHaltBoth(timer))

    val writer = splitHead(sources)
      .flatMap { case (head, tail) =>
        splitHead(tail).flatMap { case (head2, tail) =>
          Stream.emit(head) ++ Stream.emit(head2) ++ tail
        }.parEvalMap(3) { i =>
          IO(println(s"Writing ${i.toString}"))
        }
      }

    writer.compile.drain
  }

  def splitHead[F[_], O](in: fs2.Stream[F, O]): fs2.Stream[F, (O, fs2.Stream[F, O])] =
    in.pull.uncons1
      .flatMap {
        case Some((head, tail)) => fs2.Pull.output(Chunk((head, tail)))
        case None => fs2.Pull.done
      }
      .stream
}

The program merges 3 streams together. It splits the head of the stream from the remaining elements, before merging it back together. The source stream will terminate after 100 elements are emitted, the timer streams never terminate.

When run, this program generates output like:

Writing Tick2
Writing Tick1
Writing Item(1)
Writing Tick1
Writing Tick1
Writing Tick1
Writing Item(2)
Writing Tick1
Writing Tick1
Writing Item(3)
Writing Tick1
Writing Tick1
Writing Item(4)
Writing Tick1
Writing Tick1
Writing Item(5)
Writing Tick1
Writing Tick1
Writing Item(6)
Writing Tick1
Writing Tick1

Note how Tick2 from the timer2 stream is only printed once. This should happen at the same frequency as the Tick1 outputs.

This affects fs2 versions:

• 3.12.0
• 2.5.10

I'll take a look at getting a PR up with a test to highlight the issue. Thanks

[benhutchison]

Hi @lukestephenson I spent a bit of time investigating this with the help of GPT5.

I (..we 😝) suspect a problem in the splitHead method. It emits one head+tail pair and then halts.

Here's a variant that preserves the timer2 ticks in output, by not halting the stream in splitHead (by recursing on the pull). Not sure it's exacty what you need but may be a clue on how to tweak your splitHead...

//> using scala "3.7.1"
//> using lib "co.fs2::fs2-core:3.12.0"
//> using lib "org.typelevel::cats-effect:3.6.0"

import cats.effect.{IO, IOApp}
import fs2.{Stream, Pull}
import scala.concurrent.duration._

object FS2BugPeekRecursive extends IOApp.Simple {

  sealed trait Data
  final case class Item(value: Int) extends Data
  case object Tick1 extends Data
  case object Tick2 extends Data

  /**
   * Recursively peek at the head of the stream without halting.
   * Emits a sequence of (head, tail) pairs for each element.
   */
  def peekAll[F[_], O](s: Stream[F, O]): Stream[F, (O, Stream[F, O])] = {

    def go(stream: Stream[F, O]): Pull[F, (O, Stream[F, O]), Unit] =
      stream.pull.uncons1.flatMap {
        case Some((head, tail)) =>
          // emit current (head, tail) and recurse on the tail
          Pull.output1((head, tail)) >> go(tail)
        case None =>
          Pull.done
      }

    go(s).stream
  }

  override def run: IO[Unit] = {
    val source = Stream.emits(1 to 100).evalMap(i => IO(Item(i)).delayBy(2.seconds))
    val timer  = Stream.awakeEvery[IO](1.second).evalMap(_ => IO(Tick1))
    val timer2 = Stream.awakeEvery[IO](1.second).evalMap(_ => IO(Tick2))

    val sources = timer2.mergeHaltBoth(source.mergeHaltBoth(timer))

    val writer =
      peekAll(sources).parEvalMap(3) { case (head, tail) =>
        IO(println(s"Writing $head"))
      }

    writer.compile.drain
  }
}

[lukestephenson]

Hey @benhutchison. Thanks for taking a look.

I'm not following what the problem is with the original splitHead implementation. It provides the first element of the stream, plus the tail which is a Stream that can be evaluated if you decide to do so. In the example we append the tail into a subsequent stream so the intention is to consume it.

If the problem was in splitHead halting prematurely, I'd expect the original code to not work in many cases. But if you replace the parEvalMap in the original example with an evalMap then everything works as expected. Note in my example, the parEvalMap isn't required for any reason, the upstream source is much slower than the processing of the elements downstream with IO(println(s"Writing ${i.toString}")). However, I put that in to highlight that some strange combination of the merge / split / parallelism breaks things. If the splitHead implementation was to blame, shouldn't it also be broken if the parallelism is removed?

[benhutchison]

But if you replace the parEvalMap in the original example with an evalMap then everything works as expected. .. If the splitHead implementation was to blame, shouldn't it also be broken if the parallelism is removed?

Right. That seems a fairly compelling argument that the problem is not simply in splitHead..