Release loser eagerly in Resource.race

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

@@ -303,7 +303,67 @@ sealed abstract class Resource[F[_], +A] extends Serializable {
   def race[B](
       that: Resource[F, B]
   )(implicit F: Concurrent[F]): Resource[F, Either[A, B]] =
-    Concurrent[Resource[F, *]].race(this, that)
+    Resource.applyFull { poll =>
+      def cancelLoser[C](f: Fiber[F, Throwable, (C, ExitCase => F[Unit])]): F[Unit] =
+        f.cancel *>
+          f.join
+            .flatMap(
+              _.fold(
+                F.unit,
+                _ => F.unit,
+                _.flatMap(_._2.apply(ExitCase.Canceled))
+              )
+            )
+
+      poll(F.racePair(this.allocatedCase, that.allocatedCase)).flatMap {
+        case Left((oc, f)) =>
+          oc match {
+            case Outcome.Succeeded(fa) =>
+              cancelLoser(f).start.flatMap { f =>
+                fa.map {
+                  case (a, fin) =>
+                    (
+                      Either.left[A, B](a),
+                      fin(_: ExitCase).guarantee(f.join.flatMap(_.embedNever)))
+                }
+              }
+            case Outcome.Errored(ea) =>
+              F.raiseError[(Either[A, B], ExitCase => F[Unit])](ea).guarantee(cancelLoser(f))
+            case Outcome.Canceled() =>
+              poll(f.join).onCancel(f.cancel).flatMap {
+                case Outcome.Succeeded(fb) =>
+                  fb.map { case (b, fin) => (Either.right[A, B](b), fin) }
+                case Outcome.Errored(eb) =>
+                  F.raiseError[(Either[A, B], ExitCase => F[Unit])](eb)
+                case Outcome.Canceled() =>
+                  poll(F.canceled) *> F.never[(Either[A, B], ExitCase => F[Unit])]
+              }
+          }
+        case Right((f, oc)) =>
+          oc match {
+            case Outcome.Succeeded(fb) =>
+              cancelLoser(f).start.flatMap { f =>
+                fb.map {
+                  case (b, fin) =>
+                    (
+                      Either.right[A, B](b),
+                      fin(_: ExitCase).guarantee(f.join.flatMap(_.embedNever)))
+                }
+              }
+            case Outcome.Errored(eb) =>
+              F.raiseError[(Either[A, B], ExitCase => F[Unit])](eb).guarantee(cancelLoser(f))
+            case Outcome.Canceled() =>
+              poll(f.join).onCancel(f.cancel).flatMap {
+                case Outcome.Succeeded(fa) =>
+                  fa.map { case (a, fin) => (Either.left[A, B](a), fin) }
+                case Outcome.Errored(ea) =>
+                  F.raiseError[(Either[A, B], ExitCase => F[Unit])](ea)
+                case Outcome.Canceled() =>
+                  poll(F.canceled) *> F.never[(Either[A, B], ExitCase => F[Unit])]
+              }
+          }
+      }
+    }
 
   /**
    * Implementation for the `flatMap` operation, as described via the `cats.Monad` type class.
@@ -1277,6 +1337,9 @@ abstract private[effect] class ResourceConcurrent[F[_]]
   override def both[A, B](fa: Resource[F, A], fb: Resource[F, B]): Resource[F, (A, B)] =
     fa.both(fb)
 
+  override def race[A, B](fa: Resource[F, A], fb: Resource[F, B]): Resource[F, Either[A, B]] =
+    fa.race(fb)
+
   override def memoize[A](fa: Resource[F, A]): Resource[F, Resource[F, A]] = {
     Resource.eval(F.ref(false)).flatMap { allocated =>
       val fa2 = F.uncancelable(poll => poll(fa.allocatedCase) <* allocated.set(true))

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

@@ -21,6 +21,7 @@ import cats.data.{Kleisli, OptionT}
 import cats.effect.implicits._
 import cats.effect.kernel.testkit.TestContext
 import cats.effect.laws.AsyncTests
+import cats.effect.testkit.TestControl
 import cats.kernel.laws.discipline.MonoidTests
 import cats.laws.discipline._
 import cats.laws.discipline.arbitrary._
@@ -826,6 +827,29 @@ class ResourceSpec extends BaseSpec with ScalaCheck with Discipline {
           winnerClosed must beTrue
           completed must beTrue
       }
+
+      "closes the loser eagerly" in real {
+        val go = (
+          IO.ref(false),
+          IO.ref(false),
+          IO.deferred[Either[Unit, Unit]]
+        ).flatMapN { (acquiredLeft, acquiredRight, loserReleased) =>
+          val left = Resource.make(acquiredLeft.set(true))(_ =>
+            loserReleased.complete(Left[Unit, Unit](())).void)
+
+          val right = Resource.make(acquiredRight.set(true))(_ =>
+            loserReleased.complete(Right[Unit, Unit](())).void)
+
+          Resource.race(left, right).use {
+            case Left(()) =>
+              acquiredRight.get.ifM(loserReleased.get.map(_ must beRight[Unit]), IO.unit)
+            case Right(()) =>
+              acquiredLeft.get.ifM(loserReleased.get.map(_ must beLeft[Unit]).void, IO.unit)
+          }
+        }
+
+        TestControl.executeEmbed(go.replicateA_(100)).as(true)
+      }
     }
 
     "start" >> {