Surface genuine pipeline copy-pump read failures

Author: ZelAnton

CHANGELOG.md

@@ -17,6 +17,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 - `ProcessKit.Testing.FaultInjectingRunner` adds scripted, first-N, and seeded deterministic failure scenarios with virtual-time latency.
 - A performance and scalability guide documents event-driven waits, buffer tuning, fleet sizing, and benchmark interpretation.
 
+### Changed
+- `Supervisor.LivenessMemory` now documents its intentional monotonic peak-memory contract: a transient peak is not forgiven by later lower usage, and `LivenessFailures` only delays the restart after the crossing.
+
 ### Fixed
 - Pipeline relays now surface genuine upstream read failures instead of treating truncated downstream input as successful, while keeping expected broken-pipe and teardown races quiet.
 - A supervision session stopped before any incarnation produced a result no longer starts one more child just to report one, and ends with the failure that kept the child from starting — or `ProcessError.Cancelled` when no incarnation was ever started.

README.md

@@ -523,6 +523,10 @@ of hammering restarts at backoff speed. Supervision runs through the `IProcessRu
 `.WithRunner(group)` to keep every incarnation in one shared kill-on-dispose group, or a
 `ScriptedRunner` to test supervision logic hermetically.
 
+The optional `LivenessMemory` probe intentionally samples attributable **peak** tree memory for each
+incarnation. A transient peak remains a violation after current usage falls, so choose a threshold above
+expected startup spikes when they should not cause a restart; unsupported backends return a typed error.
+
 *Deeper: [Supervision](docs/supervision.md).*
 
 ## Waiting for a child to be ready

docs/supervision.md

@@ -296,9 +296,12 @@ var byMemory = new Supervisor(new Command("worker"))
 `LivenessMemory(maxBytes)` uses the current liveness interval; the two-argument overload sets it in
 the same call. It samples whole-tree peak resident memory from the run's private Job Object or cgroup,
 so descendants count and a value that crosses the threshold remains over it for that incarnation.
-ProcessKit never substitutes leader-only or shared-group memory: when the active backend cannot
-provide an attributable tree metric, supervision stops the live child and returns a typed
-`ProcessError.Unsupported`.
+This is deliberately a peak contract, not a current-working-set contract: a transient spike is still a
+memory-liveness violation after current usage falls, and `LivenessFailures` only delays the restart
+after that crossing. Set `maxBytes` above expected startup/transient peaks when those peaks should not
+restart the child. ProcessKit never substitutes leader-only or shared-group memory: when the active
+backend cannot provide an attributable tree metric, supervision stops the live child and returns a
+typed `ProcessError.Unsupported` (including the POSIX process-group fallback).
 
 Every `LivenessHttp` form also accepts a caller-owned `HttpClient` immediately after the URI. Use it
 for authentication headers, custom certificate validation, proxies, or a custom transport such as HTTP
@@ -335,10 +338,13 @@ How it behaves:
 
 - **When it restarts.** After `LivenessFailures` **consecutive** failed attempts, the supervisor
   gracefully stops the child (a `LivenessGrace` soft-stop window, then a hard kill) and restarts it
-  through the **ordinary** restart path — the same `RestartPolicy`, backoff, jitter, `MaxRestarts`
-  budget, and storm guard apply. It is not a second, parallel restart mechanism. A single healthy
-  attempt resets the run, so a brief blip that recovers does not restart the child. The first attempt
-  runs one `LivenessInterval` after the child starts, a natural startup window.
+  through the **ordinary** restart path — the same `RestartPolicy`, backoff, jitter, `MaxRestarts`
+  budget, and storm guard apply. It is not a second, parallel restart mechanism. For HTTP and
+  predicate probes, a single healthy attempt resets the run, so a brief blip that recovers does not
+  restart the child. Memory uses the monotonic peak described above: healthy samples reset the run
+  only before the peak crosses `maxBytes`, and lower current usage afterward cannot make the sample
+  healthy again. The first attempt runs one `LivenessInterval` after the child starts, a natural
+  startup window.
 
 The soft phase is the supervised command's `Command.StopSignal` (default `Signal.Term`). The same
 setting is therefore honored by an explicit supervision-session `StopAsync`, a liveness restart, and

src/ProcessKit/Supervisor.fs

@@ -207,7 +207,8 @@ type internal LivenessProbe =
     /// Evaluate an arbitrary async predicate each attempt; the child is healthy when it returns `true`.
     | Custom of probe: (unit -> Task<bool>)
 
-    /// Sample attributable whole-tree peak memory; healthy while it is at or below `maxBytes`.
+    /// Sample attributable whole-tree peak memory since the incarnation started; healthy while the
+    /// peak is at or below `maxBytes`. This is intentionally not a current-working-set sample.
     | Memory of maxBytes: int64
 
 /// The immutable configuration behind a `Supervisor`. Internal — built through the `Supervisor`
@@ -552,6 +553,9 @@ type SupervisionSession internal (config: SupervisorConfig, cancellationToken: C
                     check, None
                 | LivenessProbe.Memory maxBytes ->
                     let check (_probeToken: CancellationToken) =
+                        // ProcessGroupStats.PeakMemoryBytes is monotonic for a private Job/cgroup
+                        // incarnation. A later lower working set therefore cannot forgive a peak that
+                        // already crossed the limit; this is the explicit memory-liveness contract.
                         match running.TreePeakMemoryBytes() with
                         | Ok bytes when bytes <= maxBytes -> Task.FromResult(Ok())
                         | Ok _ -> Task.FromResult(Error(ProcessError.NotReady(program, probeTimeout)))
@@ -1384,9 +1388,11 @@ type Supervisor internal (config: SupervisorConfig) =
         )
 
     /// Enable a whole-process-tree **memory liveness probe**. The supervisor samples attributable
-    /// peak resident memory every configured liveness interval and treats a value above `maxBytes`
-    /// as a failed attempt. After `LivenessFailures` consecutive failures it gracefully stops and
-    /// restarts the child through the ordinary liveness path. `maxBytes` must be positive.
+    /// peak resident memory since the incarnation started every configured liveness interval and
+    /// treats a value above `maxBytes` as a failed attempt. The peak is monotonic for that
+    /// incarnation: once it crosses the limit, later lower current usage does not produce a healthy
+    /// memory attempt. `LivenessFailures` therefore controls how many observations precede the
+    /// restart, but cannot forgive an already-crossed peak. `maxBytes` must be positive.
     ///
     /// Whole-tree accounting requires a private Job Object or cgroup. If the active backend cannot
     /// provide an attributable metric, supervision ends with a typed `ProcessError.Unsupported`
@@ -1410,10 +1416,12 @@ type Supervisor internal (config: SupervisorConfig) =
                 LivenessInterval = Liveness.clampInterval interval }
         )
 
-    /// How many **consecutive** failed liveness attempts trip a restart (default `3`). A single healthy
-    /// attempt resets the run, so a flaky endpoint that recovers does not restart the child. `count`
-    /// must be at least `1`. No effect unless a liveness probe (`LivenessHttp`/`LivenessCheck`/
-    /// `LivenessMemory`) is set.
+    /// How many **consecutive** failed liveness attempts trip a restart (default `3`). For HTTP and
+    /// predicate probes, a single healthy attempt resets the run, so a flaky endpoint that recovers
+    /// does not restart the child. For `LivenessMemory`, a healthy attempt resets the run only while
+    /// the incarnation's peak is still at or below the limit; after the monotonic peak crosses it,
+    /// later lower current usage remains failed. `count` must be at least `1`. No effect unless a
+    /// liveness probe (`LivenessHttp`/`LivenessCheck`/`LivenessMemory`) is set.
     member _.LivenessFailures(count: int) =
         ArgumentOutOfRangeException.ThrowIfLessThan(count, 1, nameof count)
         Supervisor({ config with LivenessFailures = count })

tests/ProcessKit.Tests/SupervisorTests.fs

@@ -1969,6 +1969,69 @@ type SupervisorTests() =
         }
         :> Task
 
+    [<Test>]
+    member _.``healthy attributable memory samples do not restart before a peak crossing``() : Task =
+        task {
+            let maxBytes = 1024L
+            let healthySamples = [| 256L; 768L; 1024L |]
+            let observedSamples = ResizeArray<int64>()
+
+            let sampledEnough =
+                TaskCompletionSource<unit>(TaskCreationOptions.RunContinuationsAsynchronously)
+
+            let mutable sampleIndex = 0
+            let mutable delayCalls = 0
+
+            let treeStats () =
+                let index = Interlocked.Increment(&sampleIndex) - 1
+                let bytes = healthySamples[min index (healthySamples.Length - 1)]
+                observedSamples.Add bytes
+
+                if index + 1 = healthySamples.Length then
+                    sampledEnough.TrySetResult() |> ignore
+
+                Some(ProcessGroupStats(1, None, Some bytes, None))
+
+            let deterministicDelay (_: TimeSpan) (cancellationToken: CancellationToken) : Task =
+                if Interlocked.Increment(&delayCalls) <= healthySamples.Length then
+                    Task.CompletedTask
+                else
+                    Task.Delay(Timeout.Infinite, cancellationToken)
+
+            let runner = MemoryLivenessRunner(Some treeStats)
+            let events = ResizeArray<SupervisorRestartEvent>()
+
+            let supervisor =
+                Supervisor(Command.create "memory-healthy")
+                    .WithRunner(runner)
+                    .Backoff(TimeSpan.Zero, 1.0)
+                    .Jitter(false)
+                    .MaxRestarts(1)
+                    .LivenessMemory(maxBytes, TimeSpan.FromMilliseconds 1.0)
+                    .LivenessFailures(1)
+                    .LivenessGrace(TimeSpan.Zero)
+                    .OnRestart(fun event -> events.Add event)
+                    .WithLivenessDelay
+                    deterministicDelay
+
+            let! session = supervisor.StartAsync()
+            do! sampledEnough.Task.WaitAsync(TimeSpan.FromSeconds 10.0)
+
+            Assert.That(observedSamples |> Seq.toArray, Is.EqualTo<int64> healthySamples)
+            Assert.That(session.Status.Restarts, Is.EqualTo 0, "healthy memory samples do not restart the child")
+            Assert.That(events.Count, Is.EqualTo 0, "healthy memory samples do not emit liveness restart events")
+            Assert.That(runner.Spawns, Is.EqualTo 1, "the healthy incarnation remains the only spawn")
+
+            let! outcome = session.StopAsync(TimeSpan.FromMilliseconds 100.0)
+
+            match outcome with
+            | Ok result ->
+                Assert.That(result.Stopped, Is.EqualTo StopReason.Stopped)
+                Assert.That(result.Restarts, Is.EqualTo 0)
+            | Error error -> Assert.Fail $"expected a clean stop after healthy memory samples, got {error}"
+        }
+        :> Task
+
     [<Test>]
     member _.``whole-tree memory above the threshold uses the ordinary liveness restart path``() : Task =
         task {
@@ -2005,6 +2068,48 @@ type SupervisorTests() =
         }
         :> Task
 
+    [<Test>]
+    member _.``memory liveness keeps a recovered current sample failed after a peak crossing``() : Task =
+        task {
+            let mutable samples = 0
+            let mutable peakBytes = 0L
+
+            let treeStats () =
+                let sample = Interlocked.Increment(&samples)
+                // Model a transient current-use spike followed by recovery. The backend exposes the
+                // accumulated peak, so the second sample remains over the limit even though current
+                // usage has fallen back below it.
+                let currentBytes = if sample = 1 then 4096L else 512L
+                peakBytes <- max peakBytes currentBytes
+                Some(ProcessGroupStats(1, None, Some peakBytes, None))
+
+            let runner = MemoryLivenessRunner(Some treeStats)
+            let events = ResizeArray<SupervisorRestartEvent>()
+
+            let supervisor =
+                Supervisor(Command.create "memory-spike")
+                    .WithRunner(runner)
+                    .Backoff(TimeSpan.Zero, 1.0)
+                    .Jitter(false)
+                    .MaxRestarts(1)
+                    .LivenessMemory(1024L, TimeSpan.FromMilliseconds 1.0)
+                    .LivenessFailures(2)
+                    .LivenessGrace(TimeSpan.Zero)
+                    .OnRestart(fun event -> events.Add event)
+
+            match! supervisor.RunAsync() with
+            | Ok outcome ->
+                Assert.That(outcome.Restarts, Is.EqualTo 1)
+                Assert.That(outcome.Stopped, Is.EqualTo StopReason.RestartsExhausted)
+                Assert.That(runner.Spawns, Is.EqualTo 2)
+                Assert.That(runner.GracefulStops, Is.EqualTo 2)
+                Assert.That(Volatile.Read(&samples), Is.GreaterThanOrEqualTo 4)
+                Assert.That(events.Count, Is.EqualTo 1)
+                Assert.That(events[0].Cause, Is.EqualTo RestartCause.Liveness)
+            | Error error -> Assert.Fail $"expected the peak violation to restart after recovery, got {error}"
+        }
+        :> Task
+
     [<Test>]
     member _.``memory liveness fails honestly when whole-tree accounting is unavailable``() : Task =
         task {