Purposeful Cached Resource Leak - Like Memoize But can Move into SyncIO

Leak.scala

object ResourceLeak {

  def leak[F[_]: Sync, G[_]: Async, A](resource: Resource[G, A]): F[G[A]] =  {
    Ref.in[F, G, Option[Deferred[G, A]]](None).map{ref => 
      def initiateOrGet(withDef: Option[Deferred[G, A]] = None): G[A] = Concurrent[G].uncancelable(poll => 
          ref.modify{
            case s@Some(a) => (s: Option[Deferred[G, A]], a.get)
            case None => withDef match {
              case s@Some(deferred) => (s, poll(resource.allocated.flatTap{ case (a, _) => deferred.complete(a).void}.map(_._1)).onCancel(ref.set(None)).onError{ case _ => ref.set(None)})
              case None => (None, poll(Deferred[G, A].map(_.some).flatMap(initiateOrGet(_))))
            }
          }.flatten
      )
      initiateOrGet(None)
    }
  }
  
  // Allow Memoization via SyncIO
  import cats._
  import cats.syntax.all._
  import cats.effect._
  import cats.effect.syntax.all._
  import java.util.concurrent.CancellationException

 def externalOnce[F[_]: Sync, G[_]: Async, A](resource: G[A]): F[G[A]] =  {
    Ref.in[F, G, Option[Deferred[G, Outcome[G, Throwable, A]]]](None).map{ref => 
      def initiateOrGet(withDef: Option[Deferred[G, Outcome[G, Throwable, A]]] = None): G[A] = Concurrent[G].uncancelable(poll => 
          ref.modify{
            case s@Some(a) => (s, a.get.flatMap(fromOutcome(_)))
            case None => withDef match {
              case s@Some(deferred) => (s, poll(resource).guaranteeCase(o => deferred.complete(o).void))
              case None => (None, poll(Deferred[G, Outcome[G, Throwable, A]].map(_.some).flatMap(initiateOrGet(_))))
            }
          }.flatten
      )
      initiateOrGet(None)
    }
  }
  
  def fromOutcome[F[_]: ApplicativeThrow, A](outcome: Outcome[F, Throwable, A]): F[A] = outcome match {
    case Outcome.Succeeded(fa) => fa
    case Outcome.Errored(e) => e.raiseError[F, A]
    case Outcome.Canceled() => new CancellationException().raiseError[F, A]
  }

}