Fs2: Errors are not always catched with .handleErrorWith

[nikiforo]

Take a look at https://github.com/nikiforo/fs-handle/pull/1
The example is minimized, but can be minimized even further.

A way to reproduce.
On one tab launch the client sbt runMain name.nikiforo.Client
On the other tab launch the server

sbt
> runMain name.nikiforo.Server
acquired localhost 20111
got 0 messages

Wait a minute

<...>
got 538 messages
got 485 messages
exception handling socket Connection reset
got 320 messages
exception handling socket Connection reset
<...>
got 1 messages
DISASTER: Multiple exceptions were thrown (45), first java.io.IOException: Connection reset

Despite some errors being catched at https://github.com/nikiforo/fs-handle/pull/1/files#diff-3e571cc6ca077b3bd8dc2bde7f5f8e9787724d1f98e64d56a34e7b263957bfd5R47 others propagate to the encompassing stream, halt it, are catched only at https://github.com/nikiforo/fs-handle/pull/1/files#diff-4a1eb1d4e3714fb61d4adc5f666a0a5c435065c20bfd0433f08697a0a2c96715R27

At this point I want to differentiate whether or not the provided behavior is expected.

@mpilquist

[mpilquist]

I looked at this some today and I'm pretty sure the problem is due to partitionEither.
Consider:

    socket
      .reads(8192)
      .through(pipe)
      .flatMap {
        case OutputResponse(v) => Stream.eval_(socket.write(Chunk.bytes(v)))
        case other => Stream.emit(other)
      }
      .onComplete(Stream.eval_(socket.endOfOutput))
      .handleErrorWith(handler)

The pipe value has a partitionEither deep inside its definition. In essence, partitionEither is causing the source stream exception to occur while publishing to the underlying Broadcast, which is not directly accessible when using the returned stream. That's why calling .handleErrorWith before calling .through(pipe) works fine, whereas attaching a handleErrorWith to the result of partitionEither doesn't.

We should be able to minimize this a bunch by combining a source that raises an exception through a Broadcast. Something like Stream.raiseError[IO](new Err).through(Broadcast(1)).flatten

[mpilquist]

This works okay, so it's not quite as straightforward as I had hoped:

scala> implicit class PartitionEither[F[_], O](self: Stream[F, O]) {
     |   def partitionEither[L, R](f: O => Either[L,R])(implicit c: Concurrent[F]): Stream[F, (Stream[F, L], Stream[F, R])] =
     |     self.map(f).through(Broadcast(2)).pull.take(2).void.streamNoScope.foldMap(List(_)).map { broadIn =>
     |       val in1 = broadIn.head
     |       val in2 = broadIn.tail.head
     |       val ls = in1.flatMap(_.fold(Stream.emit, _ => Stream.empty))
     |       val rs = in2.flatMap(_.fold(_ => Stream.empty, Stream.emit))
     |       (ls, rs)
     |     }
     | }
class PartitionEither

scala> Stream(1, 2, 3, 4).append(Stream.raiseError[IO](new RuntimeException)).partitionEither(i => if (i % 2 == 0) Left(i) else Right(i * 10)).map { case (ls, rs) => ls.merge(rs) }.handleErrorWith(t => Stream.eval_(IO(println("handled")))).flatten.compile.toLi
st.attempt.unsafeRunSync()
handled
val res26: Either[Throwable,List[Int]] = Right(List(10, 2))

[mpilquist]

Interesting observation: if we comment out case OutputResponse(v) => Stream.eval_(socket.write(Chunk.bytes(v))) in Tcp2.scala, everything works. This makes me wonder if we're writing stuff to the underlying AsynchronousSocketChannel and somehow an exception is getting raised somewhere else.

Adding .attempt -- that is, case OutputResponse(v) => Stream.eval_(socket.write(Chunk.bytes(v)).attempt) -- results in the original crashing behavior, indicating maybe we're on to something here and the write is causing a non-local exception getting raised.

[mpilquist]

This minimization does NOT have the problem, which points back to something funky going on as a result of partitionEither, etc.

import cats.effect._
import fs2.Stream
import fs2.io.tcp._
import java.net.InetSocketAddress
import scala.concurrent.duration._

object Server extends IOApp {
  def run(args: List[String]) =
    Blocker[IO].use { blocker =>
      SocketGroup[IO](blocker).use { sg =>
        sg.server[IO](new InetSocketAddress("127.0.0.1", 20111)).map { client =>
          Stream.resource(client).flatMap { socket =>
            println("accepted connection")
            socket.reads(8192).chunks.evalMap(socket.write(_)).drain
          }
            .handleErrorWith(t => Stream.eval_(IO(println("caught client error: " + t.getMessage))))
            .onFinalize(IO(println("dropped connection")))
        }.parJoinUnbounded.compile.drain.as(ExitCode.Success)
      }
    }
}

object Client extends IOApp {
  def run(args: List[String]) =
    Blocker[IO].use { blocker =>
      SocketGroup[IO](blocker).use { sg =>
        Stream.constant {
          Stream.resource(sg.client[IO](new InetSocketAddress("127.0.0.1", 20111))).flatMap { client =>
            Stream.emits(Array.fill[Byte](100)(1)).covary[IO].metered(1.milli).through(client.writes())
          }.handleErrorWith(t => Stream.eval_(IO(println("terminated: " + t.getMessage))))
        }.parJoin(1000).compile.drain.as(ExitCode.Success)
      }
    }
}

[nikiforo]

Great! I think we agree, that the behavior isn't expected and should be further investigated.
A couple of facts:

• It was found in production, my minimizations from scratch didn't show the error, thus I cut out the example from the codebase.
• It was found on fs2 2.1.0. To get rid of the error I've migrated to 2.2.0, 2.3.0, 2.4.6, behavior hasn't changed.
• I've seen, that not only commenting write, but also excluding other parts hide the error.

So, now I'll minimize the sample even further.

[nikiforo]

This one is easier to reproduce. https://github.com/nikiforo/fs-handle/pull/3

Interestingly, if you change 2424000000000e000000000000000000000000000000002421 to 2421, exceptions will be thrown and cought.

If you change parJoin(1000) to parJoin(1) in Client, exceptions will still halt the encompassing stream. So... this isn't about concurrency?

Currently, this gives me no hint on what is going on.

[mpilquist]

While minimizing, Is there a simple way to remove the throughFlatResponse and use a pipe that does not use partitionEither? I'd like to rule out weird scoping issues as a result of turning a Stream[F, O] in to a Stream[F, Stream[F, O]]

[nikiforo]

I think that partitionEither & throughFlatResponse are the main villains here. The bug is induced by an unlucky combination of throughFlatResponse with something else. So, I think the bug wouldn't be reproducible without it. Anyway, I'll do my best to get rid of throughFlatResponse.
However, shouldn't fs2 be used for such scenarios? I mean I think intentions to map over specific member of sealed family while preserving other members are natural(similarly to either, future, writer, etc). Moreover, supplying mapper with pipe is more general and can solve broader class of problems.

[mpilquist]

Some more findings:

• In Tcp2, I removed the parJoinUnbounded in favor of processing each client sequentially via streams.flatMap(handleClient) - problem still occurs
• I commented out the .through(pipe) and instead just wrote random bytes to the client socket in response - problem did not occur

So more evidence that the cause is partitionEither/throughFlatResponse.

However, shouldn't fs2 be used for such scenarios?

We definitely should support such scenarios. At this point, I'm not sure we're looking at a bug in the interpreter which should be fixed or something more fundamentally limiting about our current approach to interpretation which will require a workaround in user code. Either way, we should get to the bottom of it.

[mpilquist]

Another note: problem occurs on CE3/FS2 3 as well.

[nikiforo]

https://github.com/nikiforo/fs-handle/pull/3

I've significantly minimized example. However, I won't be able to continue until tomorrow, thus sharing as is for now
.

[nikiforo]

https://github.com/nikiforo/fs-handle/pull/3

I consider minimization as completed.

[nikiforo]

Copied from the discussion of pull request:

Stream(Stream(0, 1))
      .repeat
      .flatMap { s =>
        s.flatMap {
          case 0 => Stream(())
          case 1 => Stream.raiseError[IO](new RuntimeException)
        }
        .through(Broadcast(1))
        .pull
        .take(1)
        .void
        .streamNoScope
        .foldMap(List(_))
        .flatMap(_.head)
        .recoverWith { ex => Stream.eval_(IO.delay(println(s"catched: ${ex.getMessage}")))}
      }
      .evalMap(_ => IO.unit)
      .compile
      .drain
      .recoverWith { ex => IO.delay(println(s"DISASTER: ${ex.getMessage}")) }
      .as(ExitCode.Success)

[mpilquist]

New minimization:

import cats.effect._
import cats.syntax.all._
import fs2.Stream

object Simulate extends IOApp {

  def run(args : List[String]): IO[ExitCode] =
    Stream(Stream(()) ++ Stream.raiseError[IO](new RuntimeException))
      .repeat
      .flatMap { s =>
        s.prefetch.handleErrorWith(t => Stream.empty)
      }
      .flatMap(_ => Stream.empty)
      .compile
      .drain
      .recoverWith { ex => IO.delay(println(s"DISASTER: ${ex.getMessage}")) }
      .as(ExitCode.Success)
}

[SystemFw]

no broadcast nor partition 😮

[mpilquist]

Yeah, but there's a hidden concurrently. By the way, this fails intermittently - sometimes takes as long as 1 minute before failing. Still not really sure of what's happening -- if it was as simple as a race condition in concurrently, we wouldn't need the .flatMap(_ => Stream.empty).

[mpilquist]

Same program on CE3 side-channels / drops the error and hence terminates successfully. As in, some number of the inner exceptions are caught by the inner handleErrorWith and then when one escapes, it doesn't bubble up to the outer recoverWith but instead the stream terminates successfully.