Merge pull request typelevel#3473 from ValdemarGr/main

Fix resource leak in `Scope`.

core/shared/src/main/scala/fs2/internal/Scope.scala

@@ -259,10 +259,13 @@ private[fs2] final class Scope[F[_]] private (
     * more details.
     */
   def close(ec: Resource.ExitCase): F[Either[Throwable, Unit]] =
+    F.uncancelable(_ => close_(ec))
+
+  private def close_(ec: Resource.ExitCase): F[Either[Throwable, Unit]] =
     state.modify(s => Scope.State.closed -> s).flatMap {
       case previous: Scope.State.Open[F] =>
         for {
-          resultChildren <- traverseError[Scope[F]](previous.children, _.close(ec))
+          resultChildren <- traverseError[Scope[F]](previous.children, _.close_(ec))
           resultResources <- traverseError[ScopedResource[F]](previous.resources, _.release(ec))
           _ <- self.interruptible.map(_.cancelParent).getOrElse(F.unit)
           _ <- self.parent.fold(F.unit)(_.releaseChildScope(self.id))

core/shared/src/test/scala/fs2/BracketSuite.scala

@@ -371,4 +371,22 @@ class BracketSuite extends Fs2Suite {
           .flatMap(_.join) *> released.get.assert
       }
   }
+
+  val activeFibers = if (isJVM) 10000 else 100
+  test(
+    s"#3473 Scope.close frees it's children and it's parent's reference to itself uncancelably ($activeFibers fibers)"
+  ) {
+    (0 until 4).toList.traverse_ { _ =>
+      val fa = IO.ref(0).flatMap { ref =>
+        val res = Stream.resource(Resource.make(ref.update(_ + 1))(_ => ref.update(_ - 1)))
+
+        (0 to activeFibers).toList.parTraverse_ { _ =>
+          val fa = res.evalMap(_ => IO.sleep(10.millis)).take(10).compile.drain
+          IO.race(fa, IO.sleep(90.millis))
+        } >> ref.get.map(assertEquals(_, 0))
+      }
+
+      fa
+    }
+  }
 }