Add chapter: subprocesses and external streams; cover Ox 1.0.6 additions

direct-style-scala/skills/direct-style-scala/100-resource-management.md

@@ -2,7 +2,8 @@
 
 ## Dependencies
 
-- `"com.softwaremill.ox" %% "core"` — `useInScope`, `useCloseableInScope`
+- `"com.softwaremill.ox" %% "core"` — `useInScope`, `useCloseableInScope`,
+  `resourceScope`
 
 ---
 
@@ -16,10 +17,12 @@ when the scope ends.
 ## Application resource setup
 
 The `Dependencies.create` method acquires all resources within the application's
-root scope:
+root scope. It only registers resources — no forks — so it declares
+`using ResourceScope` rather than `using Ox` (see below); the root `Ox` scope
+satisfies it:
 
 ```scala
-def create(using Ox): Dependencies =
+def create(using ResourceScope): Dependencies =
   val config = Config.read.tap(Config.log)
   val otel = initializeOtel()
   val sttpBackend = useInScope(
@@ -38,3 +41,33 @@ def create(using Ox): Dependencies =
 On scope termination (e.g., SIGTERM via `OxApp` — see [Background
 Processes](110-background-processes.md)), resources are released in reverse
 acquisition order: first the database pool, then the sttp backend.
+
+## Resources without concurrency: resourceScope
+
+When a method only needs scoped cleanup — no forks — don't open a `supervised`
+scope just to get `useInScope`: that claims a concurrency capability the code
+doesn't use. `resourceScope` (Ox ≥ 1.0.6) starts a dedicated, resource-only
+scope, Ox's analogue of `scala.util.Using.Manager`:
+
+```scala
+import ox.{resourceScope, useCloseableInScope}
+import java.io.{FileReader, FileWriter}
+
+def process(): Unit = resourceScope:
+  val in = useCloseableInScope(FileReader("in.txt"))
+  val out = useCloseableInScope(FileWriter("out.txt"))
+  // both closed when the scope completes, out first
+  out.write(in.read())
+```
+
+A resource scope can only be started where no concurrency scope is lexically
+visible — verified at compile time, because a fork started inside a visible
+resource scope could outlive it, using resources after they have been
+released. Within a concurrency scope, register
+resources directly instead; to use a resource scope e.g. in a fork's body,
+extract it to a method that doesn't take `using Ox` — good practice in itself.
+
+Methods that only register resources should declare exactly that capability:
+`using ResourceScope` instead of `using Ox`. Every concurrency scope is also a
+`ResourceScope`, so such a method can be called from both — while its
+signature no longer claims the ability to fork.

direct-style-scala/skills/direct-style-scala/150-shared-state-across-threads.md

@@ -3,7 +3,8 @@
 ## Dependencies
 
 - `"com.softwaremill.ox" %% "core"` — `Flow`, `par`, `supervised`, `fork`,
-  `forkDiscard`, `forkUserDiscard`, channels, actors
+  `forkDiscard`, `forkUserDiscard`, channels, actors, `computeIntensive`,
+  `cede`
 
 ---
 
@@ -26,6 +27,7 @@ when bridging a foreign API that Ox does not cover.
 | Mailbox or producer-consumer queue | `ox.channels.Channel[T]` |
 | Collection fan-out | `Flow.mapPar` or `Flow.mapParUnordered` |
 | Serialized access to mutable state | `Actor` |
+| Long CPU-bound computation | `computeIntensive` (see below) |
 
 ## Flows
 
@@ -287,6 +289,51 @@ stateRef.updateAndGet(state => process(state, item)).discard
 > **Warning:** Never use `AtomicReference` as a lifecycle, cancellation, or
 > shutdown flag. Model lifecycle with Ox scope structure or channel completion.
 
+## CPU-intensive work
+
+Ox forks run on virtual threads, which are never preempted: a thread yields
+only when it blocks. A long CPU-bound computation therefore monopolizes a
+carrier thread — and since there are only as many carriers as CPU cores (by
+default), a handful of such computations can starve every other virtual
+thread in the process, including latency-sensitive ones like HTTP handlers.
+Two remedies
+(Ox ≥ 1.0.6), depending on the computation:
+
+- for short bursts in code you control, call `cede()` about once every
+  millisecond of computation (a call costs ~1µs): it yields the virtual thread
+  back to the scheduler and checks the interrupt flag, making the loop both
+  fair and cancellable;
+- for long-running computations, or code that can't be instrumented with
+  yields (a third-party library), wrap the call in `computeIntensive`: the
+  computation runs on a pool of platform threads — which the OS preempts —
+  while the calling virtual thread blocks until the result is available.
+
+```scala
+import ox.*
+
+def expensive(): BigInt = (1 to 1_000_000).map(BigInt(_)).product
+
+supervised:
+  val f1 = fork(computeIntensive(expensive()))
+  val f2 = fork(computeIntensive(expensive()))
+  f1.join() + f2.join()
+```
+
+Because the caller blocks, `computeIntensive` stays structured — the
+computation never outlives the enclosing scope — and composes with `fork`,
+`par`, `race`, and `mapPar`. On cancellation the pool task is interrupted and
+the caller waits for it to complete, so computations should still call
+`checkInterrupt()` (or `cede()`) periodically where possible — a
+non-cooperating computation delays the scope's shutdown until it finishes, and
+a `timeout` around it overshoots accordingly.
+
+> **Warning:** Scope context does not propagate into the computation: `fork`
+> inside `computeIntensive` fails, `ForkLocal`s read their defaults, and
+> thread-local integrations (MDC, OpenTelemetry context — see [OpenTelemetry
+> Observability](510-opentelemetry-observability.md)) do not cross the
+> boundary. Compute the value on the pool; do the forking, logging context,
+> and tracing on the virtual-thread side.
+
 ## Scope propagation
 
 Prefer local, focused scopes. If a method only needs concurrency to compute its
@@ -312,8 +359,10 @@ def loadBoth(userId: UserId, accountId: AccountId): (User, Account) =
 ```
 
 > **Important:** `(using Ox)` in a method signature means "I will start
-> forks or register resources in your scope." If the method manages its own
-> concurrency lifecycle, use a local `supervised` block instead.
+> forks or register resources in your scope." If the method only registers
+> resources and starts no forks, declare the narrower `(using ResourceScope)`
+> instead (see [Resource Management](100-resource-management.md)); if it
+> manages its own concurrency lifecycle, use a local `supervised` block.
 
 ## Choosing the right pattern
 
@@ -326,3 +375,4 @@ Prefer the highest-level primitive that fits the shape of the problem:
 | **Channel** | Modeling an explicit protocol, mailbox, producer-consumer queue, or callback boundary. |
 | **Actor** | Multiple concurrent callers must access one mutable object serially. |
 | **AtomicReference** | A single shared value needs pure atomic updates; never use it for lifecycle flags. |
+| **computeIntensive** | Long or non-instrumentable CPU-bound work that must not starve virtual threads. |

direct-style-scala/skills/direct-style-scala/170-subprocesses-and-external-streams.md

@@ -0,0 +1,154 @@
+# Subprocesses and External Streams
+
+Driving a long-lived subprocess (or any blocking external connection — a raw
+socket, an SSE stream) whose output you consume line by line, under structured
+concurrency. Getting the ownership and teardown shape right *before* writing
+the code avoids a deadlock that only shows up at shutdown.
+
+## Dependencies
+
+- `"com.softwaremill.ox" %% "core"` — `supervised`, `fork`, `forkDiscard`,
+  `abandonOnInterruptReads`
+
+---
+
+## Own the work with a scope; return a result, not a live handle
+
+Decide the owning scope first. A reader that consumes a process's output is
+concurrent work, so it belongs to a `supervised` scope whose lifetime matches
+the work. The reader fork's **return value is the outcome** — read it with
+`join()`; don't publish it through a shared `AtomicReference` that another
+thread writes and the caller polls. There is one producer and one value.
+
+```scala
+def runAndCollect(command: Seq[String]): Summary =
+  supervised:
+    val process = ProcessBuilder(command*).start()
+    try
+      val reader = fork:
+        scala.io.Source
+          .fromInputStream(process.getInputStream)
+          .getLines()
+          .foldLeft(Summary.empty)(_.add(_)) // the fork returns the Summary
+      reader.join()                          // its return value IS the result
+    finally process.destroyForcibly().discard // see the next section
+```
+
+> **Warning:** Never return an object that owns running forks or threads, to
+> be driven by the caller afterwards — its lifetime then matches no scope, so
+> cancellation, error propagation, and cleanup become manual flags and
+> `try`/`finally` ladders, exactly the bookkeeping structured concurrency
+> removes. Keep the work inside the scope and return a plain value; if the
+> caller must interleave with the work (send input, react to events), pass a
+> *consumer* into the scope (`run(...)(use: Handle => T)`) rather than handing
+> a live handle out of it.
+
+## Teardown: a pipe read is not interruptible — destroy before the join
+
+A fork blocked reading a classic `java.io` stream — a subprocess's stdout
+pipe, a `FileInputStream`, stdin — does **not** observe `Thread.interrupt`,
+permanently and by design (see [which operations are
+interruptible](https://ox.softwaremill.com/latest/structured-concurrency/interruptions.html)
+in the Ox docs). Ox ends a scope by interrupting its forks and then joining
+them, so interruption alone will not stop such a fork, and the join blocks
+forever. For a subprocess, what unblocks the read is destroying the process:
+that closes the pipe's write-end, and the read returns EOF. Streams that don't
+support an asynchronous close — stdin, `FileInputStream` — can't be unblocked
+this way at all; wrap those with `abandonOnInterruptReads` (below).
+
+Blocking *socket* reads are different: on virtual threads — which Ox forks run
+on — interruption works since Java 21, destructively closing the socket. A
+socket-backed reader is thus ended by scope cancellation on its own; the
+deadlock here is specific to pipe and other classic stream reads.
+
+The destroy must be sequenced **before** the scope joins the fork — which is
+why it sits in the scope body's own `finally` above. Ox releases scope-managed
+resources (`useInScope`, `releaseAfterScope`) only *after* all forks complete,
+so a kill registered there runs when the join has already deadlocked. On the
+normal path the process has already exited and the destroy is a no-op; on
+cancellation or error it is what lets the scope complete.
+
+> **Required:** Close or destroy a blocking external resource in the scope
+> BODY's `finally`, before the scope joins the reader fork. Do NOT rely on
+> `releaseAfterScope` for it — that finalizer runs *after* the join and will
+> deadlock on a fork stuck in a non-interruptible read.
+
+## Kill the whole process tree, not just the direct child
+
+If the process you spawned is a launcher or wrapper that forks the real worker
+(`some-launcher run the-tool …`), destroying only your direct child orphans the
+worker. Destroy descendants first, then the root:
+
+```scala
+val handle = process.toHandle
+handle.descendants().forEach(_.destroyForcibly().discard)
+handle.destroyForcibly().discard
+```
+
+> **Warning:** A forked grandchild inherits the pipe file descriptors, and a
+> POSIX pipe reports EOF only once *every* write-end is closed. An orphan that
+> survives the kill keeps the reader blocked forever — terminate the
+> descendants, not just the process you directly spawned.
+
+## Reads you can't unblock by closing: `abandonOnInterruptReads`
+
+Destroy-and-EOF works because a subprocess supports asynchronous close. Some
+resources don't: stdin can't be meaningfully closed, and closing a
+`FileInputStream` does not unblock a pending read. And sometimes a single read
+should be cancellable (a `timeout` around one read) without tearing the
+resource down. For these, use `abandonOnInterruptReads` (Ox ≥ 1.0.6):
+it wraps an `InputStream` so the actual reads run on a *detached* virtual
+thread — unmanaged, never joined — while the calling fork awaits each chunk
+interruptibly. On interruption the wait is abandoned: the fork proceeds with an
+`InterruptedException`, and the in-flight chunk is not lost — the next read
+returns it. With `closeOnAbandon = true`, an interrupted read instead closes
+the underlying stream, and the wrapper becomes permanently closed.
+
+```scala
+import ox.*
+import scala.concurrent.duration.*
+
+// one process-wide wrapper: multiple wrappers over System.in would compete for input
+lazy val stdin = abandonOnInterruptReads(System.in)
+
+supervised:
+  val firstByte: Option[Int] = timeoutOption(1.second)(stdin.read())
+```
+
+The trade-off: an abandoned read leaves the detached thread blocked until the
+underlying read completes — for stdin, possibly for the application's lifetime.
+That is a cheap virtual thread in exchange for interruptibility. For a
+subprocess, keep the body-`finally` destroy as the primary teardown — it
+terminates the child *and* EOFs the pipe, leaving nothing behind; wrapping the
+process's stdout (`process.getInputStream`) additionally protects the reader
+from a mis-sequenced teardown, but does not replace the destroy.
+
+For one-off uninterruptible calls (a JDBC `execute`, a DNS lookup) there is
+`abandonOnInterrupt(op)(onAbandon)`, where the cleanup starts on abandonment —
+e.g. `abandonOnInterrupt(statement.execute())(connection.close())`. For
+resources supporting asynchronous close, the cleanup also unblocks the
+abandoned operation, so nothing is leaked. `abandonOnInterruptWrites` covers
+the write side — e.g. a write to the child's stdin blocked on a full pipe.
+
+## Don't let an unread pipe stall the process
+
+A subprocess blocks once an OS pipe buffer (~64 KB) fills. If you read stdout but
+ignore stderr, a chatty process wedges mid-run. Either let the child inherit the
+parent's stderr, or drain stderr in its own fork for the resource's lifetime —
+a `forkDiscard`, torn down by the same body-`finally` destroy:
+
+```scala
+supervised:
+  val process = ProcessBuilder(command*).start()
+  try
+    forkDiscard:
+      try scala.io.Source.fromInputStream(process.getErrorStream).getLines().foreach(log.debug)
+      catch case NonFatal(e) => log.debug("stderr drain ended", e)
+    val reader = fork(consume(process.getInputStream))
+    reader.join()
+  finally process.destroyForcibly().discard
+```
+
+The drain catches `NonFatal` — logged, so a real failure stays diagnosable —
+because a stray read error must not tear down the scope. It needs no stop
+signal: the body-`finally` destroy EOFs its read, and the scope then joins it.

direct-style-scala/skills/direct-style-scala/SKILL.md

@@ -85,6 +85,12 @@ def findUser(id: Id[User])(using DbTx): Either[Fail, User] =
 
 * NEVER materialize unbounded data into memory. Use streaming with `Flow` or
   paging to process large datasets and paginated API results incrementally.
+* virtual threads are never preempted — long CPU-bound computations (a few
+  suffice, e.g. a `mapPar` over such work) can starve every other virtual
+  thread in the process. Run long or non-instrumentable compute via
+  `computeIntensive` (platform-thread pool; the blocking caller keeps it
+  structured); in CPU-bound loops you control, call `cede()` about once per
+  millisecond.
 
 # Direct-style Scala
 
@@ -100,7 +106,23 @@ def findUser(id: Id[User])(using DbTx): Either[Fail, User] =
   job-level concurrency. Accept a parent `(using Ox)` only when a fork or
   resource must be tied to that parent scope's lifetime.
 * keep constructors plain; use factories that take `(using Ox)` and return
-  values that do not carry the capability.
+  values that do not carry the capability. If the factory only registers
+  resources and starts no forks, take the narrower `(using ResourceScope)`;
+  for scoped cleanup with no enclosing scope and no concurrency, use
+  `resourceScope` instead of `supervised`.
+* decide the owning scope BEFORE writing concurrent code. Model a stream reader
+  or worker as a fork in a `supervised` scope whose lifetime matches the work;
+  its result is the fork's **return value** (`fork{…}.join()`), not a value
+  published through a shared `AtomicReference`. NEVER return an object that owns
+  running forks/threads to be driven later — its lifetime escapes every scope,
+  making cancellation and cleanup manual again. Pass a consumer into the scope
+  instead of handing a live handle out.
+* a fork blocked reading a subprocess pipe, stdin, a file, or any other
+  classic `java.io` stream is NOT ended by scope cancellation — without the
+  right teardown shape, shutdown deadlocks. BEFORE writing code that drives a
+  subprocess or reads a blocking external stream (socket, SSE), read
+  [Subprocesses and External
+  Streams](170-subprocesses-and-external-streams.md).
 
 # Functional programming
 
@@ -247,7 +269,9 @@ https://raw.githubusercontent.com/virtuslab/scala-skill/refs/heads/master/direct
   sbt/Scalafix boundary enforcement.
 
 - [Resource Management](100-resource-management.md) — `useInScope`,
-  `useCloseableInScope`, reverse-order release, scope-based cleanup.
+  `useCloseableInScope`, reverse-order release, scope-based cleanup;
+  `resourceScope` for cleanup without concurrency, `using ResourceScope` as
+  the narrower capability.
 
 - [Background Processes](110-background-processes.md) — `OxApp` entry point,
   `forkDiscard`/`forkUserDiscard` for daemon vs. user threads,
@@ -264,7 +288,15 @@ https://raw.githubusercontent.com/virtuslab/scala-skill/refs/heads/master/direct
 - [Concurrency and Inter-Thread Communication](150-shared-state-across-threads.md)
   — Flows for declarative concurrent pipelines (`mapPar`, `merge`,
   `mapStateful`), Ox primitive selection, channels for worker mailboxes and
-  shutdown, actors for serialized mutable state.
+  shutdown, actors for serialized mutable state, `computeIntensive`/`cede` for
+  CPU-bound work on virtual threads.
+
+- [Subprocesses and External Streams](170-subprocesses-and-external-streams.md)
+  — driving a subprocess / socket / SSE reader as a fork whose return value is
+  the result; why a non-interruptible pipe read needs the resource destroyed
+  in the scope body's `finally` (before the join) rather than via
+  `releaseAfterScope`; process-tree teardown; `abandonOnInterruptReads` for
+  reads that can't be unblocked by closing; pipe back-pressure.
 
 ## Error Handling