Add resource-only scopes (resourceScope / ResourceScope)

core/src/main/scala/ox/Ox.scala

@@ -8,6 +8,29 @@ import ox.internal.ThreadHerd
 import java.util.concurrent.atomic.AtomicReference
 import scala.annotation.implicitNotFound
 
+/** Capability granted by a [[resourceScope]] and, via subtyping, by any concurrency scope ([[supervised]], [[supervisedError]],
+  * [[unsupervised]]).
+  *
+  * Represents a capability to register resources (e.g. using [[useInScope]] or [[releaseAfterScope]]) to be released when the scope
+  * completes. Does not allow forking.
+  *
+  * @see
+  *   [[OxUnsupervised]], [[Ox]]
+  */
+@implicitNotFound(
+  "This operation must be run within a `resourceScope`, or any concurrency scope (`supervised`, `supervisedError` or `unsupervised`). " +
+    "Alternatively, you must require that the enclosing method is run within a scope, by adding a `using ResourceScope` parameter list."
+)
+trait ResourceScope:
+  // contains null once the scope's finalizers have been run; registration then throws (see addFinalizer)
+  private[ox] def finalizers: AtomicReference[List[() => Unit]]
+  private[ox] def addFinalizer(f: () => Unit): Unit =
+    finalizers.updateAndGet {
+      case null => throw new IllegalStateException("Cannot register a resource: the scope to which it would be attached has already ended")
+      case fs   => f :: fs
+    }.discard
+end ResourceScope
+
 /** Capability granted by an [[unsupervised]] concurrency scope (as well as, via subtyping, by [[supervised]] and [[supervisedError]]).
   *
   * Represents a capability to:
@@ -21,16 +44,14 @@ import scala.annotation.implicitNotFound
 @implicitNotFound(
   "This operation must be run within a `supervised`, `supervisedError` or `unsupervised` block. Alternatively, you must require that the enclosing method is run within a scope, by adding a `using OxUnsupervised` parameter list."
 )
-trait OxUnsupervised:
+trait OxUnsupervised extends ResourceScope:
   private[ox] def herd: ThreadHerd
-  private[ox] def finalizers: AtomicReference[List[() => Unit]]
   private[ox] def supervisor: Supervisor[Nothing]
-  private[ox] def addFinalizer(f: () => Unit): Unit = finalizers.updateAndGet(f :: _).discard
   private[ox] def parent: Option[OxUnsupervised]
   private[ox] def locals: ForkLocalMap
 end OxUnsupervised
 
-/** Capability granted by an [[supervised]] or [[supervisedError]] concurrency scope.
+/** Capability granted by a [[supervised]] or [[supervisedError]] concurrency scope.
   *
   * Represents a capability to:
   *   - fork supervised or unsupervised, asynchronously running computations in a concurrency scope. Such forks can be created using

core/src/main/scala/ox/resource.scala

@@ -1,38 +1,93 @@
 package ox
 
-/** Use the given resource in the current concurrency scope. The resource is allocated using `acquire`, and released after the all forks in
-  * the scope complete (either successfully or with an error), using `release`. Releasing is [[uninterruptible]].
+import java.util.concurrent.atomic.AtomicReference
+import scala.annotation.implicitNotFound
+import scala.annotation.targetName
+import scala.util.NotGiven
+
+@implicitNotFound(
+  "resourceScope cannot be started here: a concurrency scope is visible, and forks started within the resource scope could outlive it. " +
+    "Extract the resourceScope usage to a method which doesn't take a `using Ox` parameter."
+)
+opaque type NoEnclosingConcurrencyScope = Unit
+
+object NoEnclosingConcurrencyScope:
+  // in the companion, so that it's found via the implicit scope of the type, without any imports
+  given noEnclosingConcurrencyScope(using NotGiven[OxUnsupervised]): NoEnclosingConcurrencyScope = ()
+
+/** Starts a new resource scope: within the given code block `f`, resources can be registered using [[useInScope]] and
+  * [[releaseAfterScope]]. They are released, in reverse registration order, once `f` completes (either successfully or with an exception).
+  * Releasing is [[uninterruptible]]. A resource scope is not a concurrency scope: no forks can be started and no [[ForkLocal]] values can
+  * be bound. The stdlib's analogue is `scala.util.Using.Manager`.
+  *
+  * Any concurrency scope ([[supervised]], [[supervisedError]], [[unsupervised]]) is also a resource scope, so within one you can register
+  * resources directly. Starting a resource scope there is disallowed (verified at compile-time), because forks started in a lexically
+  * visible resource scope could outlive it, using or registering resources after they've been released. For the same reason, the
+  * [[ResourceScope]] capability must not leak out of the scope: registration after the scope ends throws an [[IllegalStateException]].
+  *
+  * Finalizers run with the [[ForkLocal]] values in effect where `resourceScope` was called — the same values the body sees. A finalizer
+  * registered through a leaked capability from a nested [[ForkLocal]] binding does not see that nested binding.
+  */
+def resourceScope[T](f: ResourceScope ?=> T)(using NoEnclosingConcurrencyScope): T =
+  val scope = new ResourceScope:
+    private[ox] val finalizers = new AtomicReference[List[() => Unit]](Nil)
+  val result =
+    try Right(f(using scope))
+    catch case e: Throwable => Left(e)
+  runFinalizers(scope, result)
+
+/** Use the given resource in the current scope. The resource is allocated using `acquire`, and released using `release` when the scope
+  * completes, in reverse registration order. For concurrency scopes, release happens after all forks started within the scope have
+  * completed (either successfully or with an exception). Releasing is [[uninterruptible]].
+  *
+  * If the scope has already ended (which can only happen when using a leaked, explicitly-passed capability), the resource is acquired,
+  * immediately released — so that cleanup is never lost — and an [[IllegalStateException]] is thrown.
   */
-def useInScope[T](acquire: => T)(release: T => Unit)(using OxUnsupervised): T =
+def useInScope[T](acquire: => T)(release: T => Unit)(using rs: ResourceScope): T =
   val t = acquire
-  summon[OxUnsupervised].addFinalizer(() => release(t))
+  try rs.addFinalizer(() => release(t))
+  catch
+    case e: Throwable =>
+      try uninterruptible(release(t))
+      catch case e2: Throwable => e.addSuppressed(e2)
+      throw e
   t
+end useInScope
 
-/** Use the given resource, which implements [[AutoCloseable]], in the current concurrency scope. The resource is allocated using `acquire`,
-  * and released after the all forks in the scope complete (either successfully or with an error), using [[AutoCloseable.close()]].
-  * Releasing is [[uninterruptible]].
-  */
-def useCloseableInScope[T <: AutoCloseable](c: => T)(using OxUnsupervised): T = useInScope(c)(_.close())
+/** As [[useInScope]], but the resource, which implements [[AutoCloseable]], is released using [[AutoCloseable.close()]]. */
+def useCloseableInScope[T <: AutoCloseable](c: => T)(using rs: ResourceScope): T = useInScope(c)(_.close())
 
-/** Release the given resource, by running the `release` code block. Releasing is done after all the forks in the scope complete (either
-  * successfully or with an error), but before the current concurrency scope completes. Releasing is [[uninterruptible]].
-  */
-def releaseAfterScope(release: => Unit)(using OxUnsupervised): Unit = useInScope(())(_ => release)
+/** As [[useInScope]], but nothing is acquired — only the `release` code block is registered, to be run when the scope completes. */
+def releaseAfterScope(release: => Unit)(using rs: ResourceScope): Unit = useInScope(())(_ => release)
 
-/** Release the given resource, which implements [[AutoCloseable]], by running its `.close()` method. Releasing is done after all the forks
-  * in the scope complete (either successfully or with an error), but before the current concurrency scope completes. Releasing is
-  * [[uninterruptible]].
-  */
-def releaseCloseableAfterScope(toRelease: AutoCloseable)(using OxUnsupervised): Unit = useInScope(())(_ => toRelease.close())
+/** As [[releaseAfterScope]], but closes the given [[AutoCloseable]] resource. */
+def releaseCloseableAfterScope(toRelease: AutoCloseable)(using rs: ResourceScope): Unit = useInScope(())(_ => toRelease.close())
+
+// binary-compatibility bridges for callers compiled against previous ox versions; remove in 2.0. The Scala-level
+// names differ from the originals (to avoid overload ambiguity at in-package call sites), while @targetName restores
+// the original JVM names, preserving linkage.
+
+@targetName("useInScope")
+private[ox] def useInScopeCompat[T](acquire: => T)(release: T => Unit)(using ox: OxUnsupervised): T =
+  useInScope(acquire)(release)
+@targetName("useCloseableInScope")
+private[ox] def useCloseableInScopeCompat[T <: AutoCloseable](c: => T)(using ox: OxUnsupervised): T =
+  useCloseableInScope(c)
+@targetName("releaseAfterScope")
+private[ox] def releaseAfterScopeCompat(release: => Unit)(using ox: OxUnsupervised): Unit =
+  releaseAfterScope(release)
+@targetName("releaseCloseableAfterScope")
+private[ox] def releaseCloseableAfterScopeCompat(toRelease: AutoCloseable)(using ox: OxUnsupervised): Unit =
+  releaseCloseableAfterScope(toRelease)
 
 /** Use the given resource, acquired using `acquire` and released using `release` in the given `f` code block. Releasing is
-  * [[uninterruptible]]. To use multiple resources, consider creating a [[supervised]] scope and [[useInScope]] method.
+  * [[uninterruptible]]. To use multiple resources, consider creating a [[resourceScope]] and using the [[useInScope]] method.
   */
 inline def use[R, T](inline acquire: R, inline release: R => Unit)(inline f: R => T): T =
   useInterruptible(acquire, r => uninterruptible(release(r)))(f)
 
 /** Use the given resource, acquired using `acquire` and released using `release` in the given `f` code block. Releasing might be
-  * interrupted. To use multiple resources, consider creating a [[supervised]] scope and [[useInScope]] method.
+  * interrupted. To use multiple resources, consider creating a [[resourceScope]] and using the [[useInScope]] method.
   *
   * Equivalent to a `try`-`finally` block.
   */
@@ -54,6 +109,6 @@ inline def useInterruptible[R, T](inline acquire: R, inline release: R => Unit)(
 end useInterruptible
 
 /** Use the given [[AutoCloseable]] resource, acquired using `acquire` in the given `f` code block. Releasing is [[uninterruptible]]. To use
-  * multiple resources, consider creating a [[supervised]] scope and [[useCloseableInScope]] method.
+  * multiple resources, consider creating a [[resourceScope]] and using the [[useCloseableInScope]] method.
   */
 inline def useCloseable[R <: AutoCloseable, T](inline acquire: R)(inline f: R => T): T = use(acquire, _.close())(f)

core/src/main/scala/ox/unsupervised.scala

@@ -25,30 +25,6 @@ private[ox] def unsupervised[T](locals: ForkLocalMap, f: OxUnsupervised ?=> T):
   scopedWithCapability(OxError(NoOpSupervisor, NoErrorMode, Option(currentScope.get()), locals))(f)
 
 private[ox] def scopedWithCapability[T](capability: Ox)(f: Ox ?=> T): T =
-  def throwWithSuppressed(es: List[Throwable]): Nothing =
-    val e = es.head
-    es.tail.foreach(e.addSuppressed)
-    throw e
-
-  def runFinalizers(result: Either[Throwable, T]): T =
-    val fs = capability.finalizers.get
-    if fs.isEmpty then result.fold(throw _, identity)
-    else
-      val es = uninterruptible {
-        fs.flatMap { f =>
-          try
-            f(); None
-          catch case e: Throwable => Some(e)
-        }
-      }
-
-      result match
-        case Left(e)                => throwWithSuppressed(e :: es)
-        case Right(t) if es.isEmpty => t
-        case _                      => throwWithSuppressed(es)
-    end if
-  end runFinalizers
-
   def runWithCurrentScopeSet =
     val result =
       try
@@ -59,8 +35,8 @@ private[ox] def scopedWithCapability[T](capability: Ox)(f: Ox ?=> T): T =
       catch case e: Throwable => Left(e)
 
     // running the finalizers only once we are sure that all child threads have been terminated, so that no new
-    // finalizers are added, and none are lost
-    runFinalizers(result)
+    // finalizers are added, and none are lost; registrations after the freeze (via leaked capabilities) throw
+    runFinalizers(capability, result)
   end runWithCurrentScopeSet
 
   val previousScope = currentScope.get()
@@ -69,3 +45,32 @@ private[ox] def scopedWithCapability[T](capability: Ox)(f: Ox ?=> T): T =
     runWithCurrentScopeSet
   finally currentScope.set(previousScope)
 end scopedWithCapability
+
+/** Runs the scope's finalizers (in reverse registration order, uninterruptibly), first freezing the finalizer list (by setting it to
+  * `null`), so that later registrations fail with an exception (see [[ResourceScope.addFinalizer]]) instead of being silently lost. Must be
+  * called exactly once per scope. Returns the scope's result: the body exception is re-thrown with finalizer exceptions suppressed;
+  * finalizer exceptions alone fail the scope.
+  */
+private[ox] def runFinalizers[T](scope: ResourceScope, result: Either[Throwable, T]): T =
+  def throwWithSuppressed(es: List[Throwable]): Nothing =
+    val e = es.head
+    es.tail.foreach(e.addSuppressed)
+    throw e
+
+  val fs = scope.finalizers.getAndSet(null)
+  val es =
+    if fs.isEmpty then Nil
+    else
+      uninterruptible {
+        fs.flatMap { f =>
+          try
+            f(); None
+          catch case e: Throwable => Some(e)
+        }
+      }
+
+  result match
+    case Left(e)                => throwWithSuppressed(e :: es)
+    case Right(t) if es.isEmpty => t
+    case _                      => throwWithSuppressed(es)
+end runFinalizers