Trying to interrupt a fiber in test hangs forever? about zio HOT 2 CLOSED

There are multiple issues with your code:

1. The workflow that uses the Clock service is ask but you provide the LiveClock to practically every other workflow in your test but not that one. You can use TestAspect.withLiveClock to run your entire test with the live clock or Live.live or Live.withLive to run a particular part with the live clock. But if you want to use the more granular operators you need to actually use them on the part of your test that you want to use the live clock.
2. You are forking an uninterruptible fiber that never completes and then attempting to interrupt it which never completes. You fork a fiber in the acquire action of ZIO.acquireRelease. The acquire action is performed uninterruptibly and fibers inherit the interruptibility of their parents so the forked fiber is interruptible. You then attempt to interrupt that fiber in the release action but that can never complete because the fiber is uninterruptible. You can use the interruptible operator on the workflow you are forking to designate that it should be interruptible even though it is forked in an uninterruptible region.

See a corrected example below. We have numerous forums such as Discord where users can get help. Please use Github issues for things that are not working or feature requests, not questions about how to do something.

import zio._
import zio.stream._
import zio.test._
import zio.test.TestAspect._

import java.util.concurrent.atomic.AtomicInteger

object EchoService {
  val nextId = new AtomicInteger(0)
  val manager = ZLayer.fromZIO(
    for {
      in  <- Queue.bounded[String](10)
      out <- Queue.bounded[(Int, String)](10)
    } yield Mgr(in, out)
  )

  val forked = ZLayer.fromZIO(
    ZIO
      .acquireRelease(
        for {
          mgr <- ZIO.service[Mgr]
          f   <- mgr.fork
        } yield f
      )(_.interrupt)
  )

  def ask(s: String) =
    ZIO.service[Mgr].flatMap(_.ask(s))

  case class Mgr(in: Queue[String], out: Queue[(Int, String)]) {
    def ask(s: String) = in.offer(s) *> out.take.timeout(1.second)
    def fork           = Svc(nextId.updateAndGet(_ + 1), in, out).serve.interruptible.fork
  }

  case class Svc(id: Int, in: Queue[String], out: Queue[(Int, String)]) {
    val serve = ZStream
      .fromQueue(in)
      .takeWhile(_.nonEmpty)
      .foreach(s => out.offer(id -> s))
  }
}

object Tests extends ZIOSpecDefault {

  def spec = suite("Tests")(
    suite("Forked service")(
      test("echo") {
        for {
          x <- EchoService.ask("foo")
        } yield assertTrue(x == Some((1, "foo")))
      }
    ).provideSomeLayerShared(EchoService.manager >+> EchoService.forked) @@ withLiveClock
  )
}

from zio.

The workflow that uses the Clock service is ask but you provide the LiveClock to practically every other workflow in your test but not that one.

Not sure what you mean by that. Is not providing live clock to ask?

def ask(s: String) = ZIO.service[Mgr].flatMap(_.ask(s)).withClock(ClockLive)

The acquire action is performed uninterruptibly and fibers inherit the interruptibility of their parents so the forked fiber is interruptible. You then attempt to interrupt that fiber in the release action but that can never complete because the fiber is uninterruptible. You can use the interruptible operator on the workflow you are forking to designate that it should be interruptible even though it is forked in an uninterruptible region.

This is pure gold! Exactly what I was missing. Thank you so much!!!

from zio.