Semantics of `fold` apparently broken by `observe` in case of failure

Hi.

The following piece of code shows that in version 1.0.5, when a pipe given to observe fails, a folded stream fails to comply with its guarantee of single emission.

import cats.effect.{ExitCode, IO, IOApp}
import cats.implicits._
import fs2.Stream

object TestApp extends IOApp {
  def run(args: List[String]): IO[ExitCode] =
    stream.compile.drain.as(ExitCode.Success)

  private def stream =
    for {
      count <- subStream
      _ <- Stream.eval(IO(println(s"Count: $count")))
    } yield ()

  private def subStream =
    Stream.range[IO](1, 10)
      .observe(_ => Stream.raiseError[IO](new RuntimeException("Just for fun")))
      .fold(0)((c, _) => c + 1)
}

The result is the following.

Count: 0
java.lang.RuntimeException: Just for fun
	at com.softwaymedical.itodirectory.extraction.TestApp$.$anonfun$subStream$1(TestApp.scala:19)
	at com.softwaymedical.itodirectory.extraction.TestApp$.$anonfun$subStream$1$adapted(TestApp.scala:19)
	at fs2.Stream$.through$extension(Stream.scala:2521)
	at fs2.Stream$InvariantOps$.sinkStream$1(Stream.scala:3493)
	...

Since fold should emit a single value, I expect the stream to fail without emitting anything. This is indeed the observed behavior when the implementation of subStream is replaced with:

Stream.raiseError[IO](new RuntimeException("Just for fun")).fold(0)((c, _) => c + 1)

Instead, a value is emitted, and then the streams fails.

As a result, the later effects (materialized here with Stream.eval(IO(println(s"Count: $count")))) take place despite the failure, which in my case leads to dramatic consequences.

I am aware of the concerns expressed in #1206 and #1208 and I was wondering whether they could be the cause.

I use observe a lot, mainly because I often need to use fold to gather information at the end of an ingestion process and use it afterwards. It forces me to shift my main logic into a "secondary" pipe given to observe, the reporting becoming the main stream so that I can preserve the result.

By the way, if there exists a better way to implement this kind of behavior, I would be glad to know about it.

Start with the supplied minimal reproducer and inspect the semantics of Stream.observe and fold, including the related concerns in issues #1206 and #1208. Compare the failing observed stream with the direct Stream.raiseError case. Done means a failing observed pipe does not cause fold to emit a value before the stream fails, with regression coverage for the reported behavior.