Avoid spurious Resource timeout error reports

kernel/shared/src/main/scala/cats/effect/kernel/Resource.scala

@@ -1479,6 +1479,91 @@ abstract private[effect] class ResourceConcurrent[F[_]]
   override def race[A, B](fa: Resource[F, A], fb: Resource[F, B]): Resource[F, Either[A, B]] =
     fa.race(fb)
 
+  /*
+   * The fiber returned for the side still running is not canceled when the scope
+   * containing `racePair` closes, matching `start`:
+   * 1. If it finished allocating before the scope closes, its finalizers run when
+   *    the scope closes
+   * 2. If it is still allocating when the scope closes, it keeps running and releases
+   *    what it allocated as soon as it finishes; `join` then results in `Canceled()`
+   * 3. If it never finishes allocating, the fiber leaks, so cancel it before the
+   *    scope closes if that can happen
+   */
+  override def racePair[A, B](fa: Resource[F, A], fb: Resource[F, B]): Resource[
+    F,
+    Either[
+      (Outcome[Resource[F, *], Throwable, A], Fiber[Resource[F, *], Throwable, B]),
+      (Fiber[Resource[F, *], Throwable, A], Outcome[Resource[F, *], Throwable, B])]
+  ] = {
+    type Fin = Resource.ExitCase => F[Unit]
+    type Result = Either[
+      (Outcome[Resource[F, *], Throwable, A], Fiber[Resource[F, *], Throwable, B]),
+      (Fiber[Resource[F, *], Throwable, A], Outcome[Resource[F, *], Throwable, B])]
+
+    // (scope closed, finalizer to run when it closes) for each side
+    val open: (Boolean, Fin) = (false, _ => F.unit)
+    val closed: (Boolean, Fin) = (true, _ => F.unit)
+
+    // the loser may finish allocating after the scope has closed: in that case it releases
+    // what it allocated right away and reports cancelation, as `start` does
+    def allocate[C](rc: Resource[F, C], state: Ref[F, (Boolean, Fin)]): F[(C, Fin)] =
+      F.uncancelable { poll =>
+        poll(rc.allocatedCase).flatMap {
+          case (c, fin) =>
+            state.modify {
+              case (true, _) =>
+                (
+                  closed,
+                  fin(Resource.ExitCase.Canceled) *> poll(F.canceled) *> F.never[(C, Fin)])
+              case (false, _) => ((false, fin), F.pure((c, fin)))
+            }.flatten
+        }
+      }
+
+    def release(state: Ref[F, (Boolean, Fin)], ec: Resource.ExitCase): F[Unit] =
+      state.modify { case (_, fin) => (closed, fin(ec)) }.flatten
+
+    def liftOutcome[C](
+        oc: Outcome[F, Throwable, (C, Fin)],
+        state: Ref[F, (Boolean, Fin)]): F[Outcome[Resource[F, *], Throwable, C]] =
+      oc match {
+        case Outcome.Succeeded(fc) =>
+          state.get.flatMap {
+            case (true, _) => F.pure(Outcome.canceled[Resource[F, *], Throwable, C])
+            case (false, _) =>
+              fc.map {
+                case (c, _) => Outcome.succeeded[Resource[F, *], Throwable, C](Resource.pure(c))
+              }
+          }
+        case Outcome.Errored(e) => F.pure(Outcome.errored[Resource[F, *], Throwable, C](e))
+        case Outcome.Canceled() => F.pure(Outcome.canceled[Resource[F, *], Throwable, C])
+      }
+
+    def liftFiber[C](
+        f: Fiber[F, Throwable, (C, Fin)],
+        state: Ref[F, (Boolean, Fin)]): Fiber[Resource[F, *], Throwable, C] =
+      new Fiber[Resource[F, *], Throwable, C] {
+        def cancel = Resource.eval(f.cancel)
+        def join = Resource.eval(f.join.flatMap(liftOutcome(_, state)))
+      }
+
+    Resource.applyFull { poll =>
+      (F.ref(open), F.ref(open)).flatMapN { (left, right) =>
+        val finalizeBoth: Fin = ec => release(left, ec).guarantee(release(right, ec))
+
+        poll(F.racePair(allocate(fa, left), allocate(fb, right)))
+          .flatMap[Result] {
+            case Left((oc, f)) =>
+              liftOutcome(oc, left).map(oc => Left((oc, liftFiber(f, right))))
+            case Right((f, oc)) =>
+              liftOutcome(oc, right).map(oc => Right((liftFiber(f, left), oc)))
+          }
+          .onCancel(finalizeBoth(Resource.ExitCase.Canceled))
+          .map(result => (result, finalizeBoth))
+      }
+    }
+  }
+
   override def memoize[A](fa: Resource[F, A]): Resource[F, Resource[F, A]] =
     fa.memoizedAcquire.map(Resource.eval(_))
 }

tests/shared/src/test/scala/cats/effect/ResourceSuite.scala

@@ -1211,6 +1211,133 @@ class ResourceSuite extends BaseScalaCheckSuite with DisciplineSuite {
     }
   }
 
+  for {
+    acquisition <- List("make", "eval")
+    operation <- List("timeout", "racePair left", "racePair right")
+  } {
+    ticked(s"$operation - does not report handled $acquisition acquisition errors") {
+      implicit ticker =>
+        val error = new RuntimeException("acquisition failed")
+        var reported: List[Throwable] = Nil
+
+        val action = IO.executionContext.flatMap { ec =>
+          val recording = new ExecutionContext {
+            def reportFailure(t: Throwable): Unit = reported ::= t
+            def execute(r: Runnable): Unit = ec.execute(r)
+          }
+          val fail = IO.cede *> IO.raiseError[Unit](error)
+          val resource = acquisition match {
+            case "make" => Resource.make(fail)(_ => IO.unit)
+            case _ => Resource.eval(fail)
+          }
+          val raced = operation match {
+            case "timeout" => resource.timeout(1.hour)
+            case "racePair left" =>
+              Spawn[Resource[IO, *]].racePair(resource, Resource.never[IO, Unit]).flatMap {
+                case Left((outcome, _)) => outcome.embedNever
+                case Right((_, outcome)) => outcome.embedNever
+              }
+            case _ =>
+              Spawn[Resource[IO, *]].racePair(Resource.never[IO, Unit], resource).flatMap {
+                case Left((outcome, _)) => outcome.embedNever
+                case Right((_, outcome)) => outcome.embedNever
+              }
+          }
+          raced.use_.attempt.evalOn(recording)
+        }
+
+        assertCompleteAs(action, Left(error))
+        ticker.ctx.tick()
+        assertEquals(reported, Nil)
+    }
+  }
+
+  for (winnerOnLeft <- List(true, false)) {
+    ticked(s"racePair - retains both successful resources until scope exit ($winnerOnLeft)") {
+      implicit ticker =>
+        val action = IO.ref(List.empty[Int]).flatMap { released =>
+          val winner = Resource.make(IO.pure(1))(i => released.update(i :: _))
+          val loser = Resource.make(IO.sleep(1.second).as(2))(i => released.update(i :: _))
+          val raced =
+            if (winnerOnLeft)
+              Spawn[Resource[IO, *]].racePair(winner, loser)
+            else
+              Spawn[Resource[IO, *]].racePair(loser, winner)
+
+          raced.use { result =>
+            val (outcome, fiber) = result.fold(identity, { case (f, oc) => (oc, f) })
+            outcome.embedNever.use(IO.pure).flatMap(i => IO(assertEquals(i, 1))) *>
+              fiber.joinWithNever.use(IO.pure).flatMap(i => IO(assertEquals(i, 2))) *>
+              released.get.flatMap(rs => IO(assertEquals(rs, Nil)))
+          } *> released.get.map(_.sorted)
+        }
+
+        assertCompleteAs(action, List(1, 2))
+    }
+
+    ticked(s"racePair - finalizes masked loser after scope exit ($winnerOnLeft)") {
+      implicit ticker =>
+        val action = IO.ref(List.empty[Int]).flatMap { released =>
+          val winner = Resource.make(IO.sleep(100.millis).as(1))(i => released.update(i :: _))
+          val loser = Resource.make(IO.sleep(1.second).as(2))(i => released.update(i :: _))
+          val raced =
+            if (winnerOnLeft)
+              Spawn[Resource[IO, *]].racePair(winner, loser)
+            else
+              Spawn[Resource[IO, *]].racePair(loser, winner)
+
+          raced.use(result => IO.pure(result.fold(_._2, _._1))).flatMap { fiber =>
+            released.get.flatMap(rs => IO(assertEquals(rs, List(1)))) *>
+              IO.sleep(2.seconds) *>
+              fiber.join.use(oc => IO(assert(oc.isCanceled))) *>
+              released.get.map(_.sorted)
+          }
+        }
+
+        assertCompleteAs(action, List(1, 2))
+    }
+  }
+
+  ticked("racePair - canceling the loser preserves its canceled join outcome") {
+    implicit ticker =>
+      var released: List[Int] = Nil
+      val winner = Resource.make(IO.sleep(100.millis))(_ => IO { released ::= 1 })
+      val loser = Resource.make(IO.unit)(_ => IO { released ::= 2 }) *> Resource.never[IO, Unit]
+      val action = Spawn[Resource[IO, *]].racePair(winner, loser).use {
+        case Left((_, fiber)) =>
+          (fiber.cancel *> fiber.join).use { oc =>
+            IO {
+              assert(oc.isCanceled)
+              assertEquals(released, List(2))
+            }
+          }
+        case Right(_) => IO(fail("the pending resource won"))
+      }
+
+      assertCompleteAs(action, ())
+      assertEquals(released.sorted, List(1, 2))
+  }
+
+  ticked("racePair - cancellation releases both pending acquisitions") { implicit ticker =>
+    var released = 0
+    val action = (IO.deferred[Unit], IO.deferred[Unit]).tupled.flatMap {
+      case (leftStarted, rightStarted) =>
+        val left = Resource.make(leftStarted.complete(()))(_ => IO { released += 1 }) *>
+          Resource.never[IO, Unit]
+        val right = Resource.make(rightStarted.complete(()))(_ => IO { released += 1 }) *>
+          Resource.never[IO, Unit]
+        for {
+          fiber <- Spawn[Resource[IO, *]].racePair(left, right).use_.start
+          _ <- leftStarted.get *> rightStarted.get
+          _ <- fiber.cancel
+          outcome <- fiber.join
+        } yield assert(outcome.isCanceled)
+    }
+
+    assertCompleteAs(action, ())
+    assertEquals(released, 2)
+  }
+
   // issue #4489 repro (1)
   real("timeout finalizer (start/release race)") {
     val test: IO[Unit] = IO.ref(false).flatMap { ref =>