`.interruptWhen` sometimes prevents error handler in `.handleErrorWith` from running

[vladimir-lu]

This seems to be a concurrency issue introduced by .interruptWhen. This test reproduces the behavior: https://github.com/vladimir-lu/fs2-interruptwhen-concurrency-issue/blob/master/src/test/scala/com/example/InterruptHandleBug.scala

A simpler example:

(1 to 50).foreach { _ =>
      val stopSignal = Signal.constant[IO, Boolean](false)
      val gotError = new AtomicBoolean()
      Stream(42)
        .covary[IO]
        .evalMap(_ => IO.raiseError(new RuntimeException("BOOOM")))
        .interruptWhen(stopSignal)
        .handleErrorWith { _ =>
          gotError.set(true); Stream.empty
        }
        .compile
        .drain
        .unsafeRunSync()

      gotError.get() should equal(true)
}

If you remove .interruptWhen the test always passes. Otherwise errors are intermittent.

To me this seems quite similar to the issues described in #1021 and some of the comments in #1063

I would have a look at the implementation of .interruptWhen but I'm not familiar with internals yet.

[pchlupacek]

This seems to be related to invalid definition of Signal.constant, and essentially, the program races between constant stream to terminate immediately and other stream to fail.

This shall be fixed in 1.0.

However, I think that we shall backport fix to Signal.constant to 0.10 as well to prevent issues like these. Any thoughts @SystemFw @mpilquist ?

[vladimir-lu]

Indeed, not using a constant Signal makes the issue disappear.

Out of curiosity, what fixes this in 1.0? Is it because .discrete is defined as a stream of a single element?

[pchlupacek]

Hm, interesting, I thought that fix was already merged there. Could you perhaps close this and make another issue that Signal.constant is buggy and link this reference to it ? Thanks a lot for this help and findings.

[vladimir-lu]

Closing in favour of #1171