Unreliable stream interruption

When a stream is interrupted, I expect based on these docs that the interruption is final and can only be handled with Stream.bracket (or something built on Stream.bracket).

So, I would not expect the following code, which attempts to restart a stream indefinitely, to work when the provided stream is interrupted:

def resume[F[_] : Concurrent : RaiseThrowable, A, B](mk: A => Stream[F, B], checkpoint: B => A)(start: A): Stream[F, B] = {
  def go(s: Stream[F, Either[Throwable, B]], watermark: A): Pull[F, B, Unit] = s.pull.uncons1.flatMap {
    case Some((Right(b), rest)) => Pull.output1(b) >> go(rest, checkpoint(b))
    case Some((Left(_), _)) => go(mk(watermark).attempt, watermark)
    case None => go(mk(watermark).attempt, watermark)
  }

  go(mk(start).attempt, start).stream
}

However, I find that it sometimes works, and sometimes doesn't. In the following test:

// How many times should we interrupt the stream?
val Interrupts: Int = 1

// Interrupt the stream after five items, up to a max number of times
def interrupter[A](deferred: Deferred[IO, Unit], interruptCount: Ref[IO, Int]): Pipe[IO, A, A] = {
  input: Stream[IO, A] =>
    input.zipWithIndex
      .evalTap {
        case (_, 5) => interruptCount.getAndUpdate(_ + 1).flatMap { i =>
          if (i < Interrupts) {
            deferred.complete(())
          } else IO.unit
        }
        case _ => IO.unit
      }
      .map(_._1)
      .interruptWhen(deferred.get.attempt)
}

val stream: Int => Stream[IO, Int] = Stream.iterate(_)(_ + 1)

val assertion = for {
  interruptCount <- Ref.of[IO, Int](0)
  _ <- resume[Int, Int](
    start => Stream.eval(Deferred[IO, Unit]).flatMap(d => stream(start).through(interrupter(d, interruptCount))),
    _ + 1
  )(0)
    .take(1000)
    .compile
    .toList
    .map(lst => assert(lst == List.range(0, 1000))
} yield ()

if I only interrupt the stream a small number of times (e.g. once), most likely the restart function works and I get the 1000 elements from .take. But if I allow many interruptions (e.g. 10), typically the stream I get is truncated.

This is very surprising! Is this nondeterminism a bug?

Use the linked interruption documentation and the provided resume/interrupter reproduction as entry points. Run the assertion with one and many interruptions, then trace interruption handling to determine whether truncation is expected or a bug. Done means the nondeterminism is explained and the behavior is covered by the assertion or a focused regression test.