Add a memoizedAcquire method to Resource

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

@@ -751,6 +751,24 @@ sealed abstract class Resource[F[_], +A] extends Serializable {
         K.combineK(allocate(this), allocate(that))
       }
 
+  /**
+   * A Resource where the acquire step is done lazily and memoized. This means that acquire
+   * happens only if and when the `F[A]` value is executed, instead of happening immediately
+   * upon `use()`. If the `F[A]` value is executed multiple times, acquire happens once only and
+   * the acquired resource is shared to all callers. The resource is released as normal at the
+   * end of `use` (whether normal termination, error, or cancelled), if it was acquired.
+   */
+  def memoizedAcquire[B >: A](implicit F: Concurrent[F]): Resource[F, F[B]] = {
+    Resource.eval(F.ref(List.empty[Resource.ExitCase => F[Unit]])).flatMap { release =>
+      val fa2 = F.uncancelable { poll =>
+        poll(allocatedCase).flatMap { case (a, r) => release.update(r :: _).as(a) }
+      }
+      Resource.makeCaseFull[F, F[B]](poll => poll(F.memoize(fa2).map(_.widen))) { (_, exit) =>
+        release.get.flatMap(_.foldMapM(_(exit)))
+      }
+    }
+  }
+
 }
 
 object Resource extends ResourceFOInstances0 with ResourceHOInstances0 with ResourcePlatform {
@@ -1377,18 +1395,8 @@ 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)
 
-  override def memoize[A](fa: Resource[F, A]): Resource[F, Resource[F, A]] = {
-    Resource.eval(F.ref(List.empty[Resource.ExitCase => F[Unit]])).flatMap { release =>
-      val fa2 = F.uncancelable { poll =>
-        poll(fa.allocatedCase).flatMap { case (a, r) => release.update(r :: _).as(a) }
-      }
-      Resource
-        .makeCaseFull[F, F[A]](poll => poll(F.memoize(fa2))) { (_, exit) =>
-          release.get.flatMap(_.foldMapM(_(exit)))
-        }
-        .map(memo => Resource.eval(memo))
-    }
-  }
+  override def memoize[A](fa: Resource[F, A]): Resource[F, Resource[F, A]] =
+    fa.memoizedAcquire.map(Resource.eval(_))
 }
 
 private[effect] trait ResourceClock[F[_]] extends Clock[Resource[F, *]] {

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

@@ -36,6 +36,7 @@ import scala.concurrent.ExecutionContext
 import scala.concurrent.duration._
 
 import java.util.concurrent.atomic.AtomicBoolean
+import cats.effect.unsafe.IORuntimeConfig
 
 class ResourceSpec extends BaseSpec with ScalaCheck with Discipline {
   // We need this for testing laws: prop runs can interfere with each other
@@ -1170,6 +1171,33 @@ class ResourceSpec extends BaseSpec with ScalaCheck with Discipline {
         }.void must completeAs(())
       }
     }
+
+    "does not leak if canceled right after delayed acquire is canceled" in ticked {
+      implicit ticker =>
+        (IO(new AtomicBoolean), IO.ref(false), IO.ref(false)).flatMapN {
+          (acquired, used, released) =>
+            val go = for {
+              fiber <- Resource
+                .make(IO(acquired.set(true)))(_ => released.set(true))
+                .memoizedAcquire
+                .use(_ *> used.set(true))
+                .start
+              _ <- IO.cede.untilM_(IO(acquired.get))
+              _ <- fiber.cancel
+            } yield ()
+            TestControl
+              .executeEmbed(go, IORuntimeConfig(1, 2))
+              .flatMap { _ =>
+                for {
+                  acquireRun <- IO(acquired.get)
+                  useRun <- used.get
+                  releaseRun <- released.get
+                } yield acquireRun && releaseRun
+              }
+              .replicateA(1000)
+              .map(_.forall(identity(_)))
+        } must completeAs(true)
+    }
   }
 
   "uncancelable" >> {