feat(materialize): worker supervisor with per-OS memory cap (TASK-2198 U3) (#1682)

* feat(materialize): parent-side worker supervisor with per-OS memory cap (TASK-2198 U3)

The Supervisor runs materializer jobs in ONE long-lived child process
(`pad __materialize-worker`), started lazily on the first job and fed
jobs serially. A job past PAD_MATERIALIZE_TIMEOUT (default 2s) kills the
child (goja's Interrupt, sent as timeout_ms a tenth earlier, is only the
soft layer); a dead child is respawned on the next job after an
exponential backoff (1s doubling, capped at 5 min, reset by a child that
lived a minute). Every stop is logged with its reason (deadline, memory,
exit, protocol, start, canceled, closed); every start logs the effective
timeout, memory limit and cap mechanism. The child's stderr goes to the
log, pid-tagged and bounded, never near the protocol. A job whose child
died is never retried here.

Memory cap (PAD_MATERIALIZE_MEM_LIMIT, default 2GiB, bytes or
KiB/MiB/GiB, clamped to 1536MiB-16GiB):
- Linux: RLIMIT_AS set on the child by prlimit before its first byte of
  input; the child only runs its own startup uncapped.
- Windows: Job Object with JOB_OBJECT_LIMIT_PROCESS_MEMORY; the child is
  created suspended, assigned, then resumed: no uncapped window.
- macOS: parent-side RSS watchdog every 100ms (proc_info syscall, ps
  fallback).
- Other: deadline kill only.

Not wired into the server yet (U4).

Claude-Session: https://claude.ai/code/session_01H63zaWhkb5pAafx2WGjrVC

* ci(materialize): run the supervisor on the real macOS and Windows caps (TASK-2198 U3)

native-smoke now builds the real materializer bundle (as the go job does)
and runs the supervisor, memory-cap and deadline tests on macos-latest
(proc_info RSS watchdog) and windows-latest (suspended start + Job
Object). PAD_MATERIALIZE_TEST_NO_SKIP turns the cap tests' skips into
failures there, so a cap that stops running cannot pass as skipped.

Tests: a real Windows processGone (OpenProcess + GetExitCodeProcess),
RSS samplers for darwin (the production proc_info path) and Windows
(K32GetProcessMemoryInfo), and an allocation helper that writes noise
into every byte so macOS memory compression cannot hide it from the
watchdog.

The worker-loop tests share one loaded Runner through serveWorker
(RunWorker's loop, split out); TestWorkerRoundTrip still goes through
RunWorker end to end. go test -race -count=3 on the package: 714s -> 413s.

docs: the materializer section says the worker and its knobs are inert
until U4 wires it in.

Claude-Session: https://claude.ai/code/session_01H63zaWhkb5pAafx2WGjrVC

* fix(materialize): a worker never outlives its supervising process (TASK-2198 U3, codex r1)

If the server died mid-job (SIGKILL, OOM kill, crash) its worker ran on
with no deadline kill and, on macOS, no memory watchdog. Two layers:

- Linux: Pdeathsig = SIGKILL. The fork runs on a fresh goroutine
  (startUnlocked), never locked to its thread, so it cannot land on a
  thread the runtime later retires (golang/go#27505); Go's forkExec
  re-checks getppid if the parent is already gone.
- Every Unix (the only layer on macOS): the worker command calls
  ExitWhenOrphaned before the bundle loads, polling its parent pid every
  200ms against PAD_MATERIALIZE_PARENT_PID and exiting (status 70) when
  it changes. It writes nothing.
- Windows was already covered by the Job Object's KILL_ON_JOB_CLOSE.

TestSupervisorOrphanedWorkerExits kills a helper process that supervises a
worker hung in a job and requires the worker gone within 1s: 0s on Linux,
105ms with the ppid watch alone; with both layers removed the worker
outlives its parent by 5s. It matches native-smoke's TestSupervisor.*
selection, so it runs on macOS and Windows too.

Claude-Session: https://claude.ai/code/session_01H63zaWhkb5pAafx2WGjrVC

* fix(materialize): relative memory cap, refuse when uncappable, GOMAXPROCS=2; nix-safe ps test (TASK-2198 U3)

The absolute Linux RLIMIT_AS failed on CI: a 64 MiB allocation died under a
1600 MiB cap, because the runtime there reserved more address space than on
the machine the floor was measured on. The cap is now RELATIVE on every OS:
once the worker has loaded and before it is sent any job, its usage is read
as a baseline and the cap is set to baseline + PAD_MATERIALIZE_MEM_LIMIT —
address space on Linux (/proc/<pid>/stat vsize, prlimit, read back),
committed memory on Windows (the Job Object's per-process limit, set after
the suspended start and load), resident memory on macOS (the watchdog's
threshold). Every start logs baseline, limit and effective cap.

Lead ruling, REFUSE: if the baseline probe fails or is implausible (0, or
above MaxPlausibleBaseline = 8 GiB, ~4.8x the largest measured 1704 MiB),
onBaselineProbeFailure kills the worker without sending it the job and
fails the job with ErrNoMemoryCap; one WARN until a probe succeeds again.
An OS with no cap mechanism never starts a worker and refuses every job
with ErrNoMemoryCap, one WARN naming runtime.GOOS.

The worker runs with GOMAXPROCS=2: measured loaded-worker baseline
1357 MiB vs 1597 MiB at this host's default (8) and 1704 MiB at 64.
MinMemLimit drops from 1536 MiB (an artefact of the absolute cap) to
256 MiB: the corpus grows the worker by at most +10 MiB of address space and
+18 MiB resident, and a 4,000-deep nested document by +129..194 MiB.
A cgo worker's "pthread_create failed" at the cap is now classified as a
memory death.

Nix: TestParsePSRSS parses fixed strings only; the live `ps` leg is
TestPSRSSLive and skips when there is no `ps` (the Nix sandbox has none;
ps is only macOS's fallback), unless PAD_MATERIALIZE_TEST_NO_SKIP is set.

Claude-Session: https://claude.ai/code/session_01H63zaWhkb5pAafx2WGjrVC

* fix(materialize): Windows kills on the job's memory notification; steady thread set before the baseline (TASK-2198 U3)

Windows (CI on c1866957): the Job Object refused the 1 GiB commit
(VirtualAlloc errno=1455), but the worker did not die and the job ran to its
deadline. The cap no longer depends on the runtime choosing to crash: the
job is associated with an I/O completion port, and
JOB_OBJECT_MSG_PROCESS_MEMORY_LIMIT for the worker's pid kills it at once
(ErrMemoryLimit). errno=1455 on stderr is a second signal. On every OS a
cap mechanism that sees the limit hit calls onLimit, and the kill is
reported as "reached its <mechanism> cap".

Linux (CI's cgo test binary at GOMAXPROCS=64): threads created after the
baseline ate the growth budget. The worker now brings its thread set to
steady state before readiness (warmThreads: an allocating goroutine per P,
then a GC). Measured on a cgo build: threads added after the baseline are 0
at GOMAXPROCS=2 with or without it; at 64, +3 threads and +272 MiB of
address space during the corpus without it, +0 and +0 MiB with it. 256 MiB
stays the floor (largest growth at GOMAXPROCS=2: +193 MiB, a 4,000-deep
nested document; the corpus adds +0 MiB).

The cap test's second leg no longer uses GOMAXPROCS=64 (which also changed
thread creation) to stand in for a bigger host: the helper inflates its
own baseline before readiness, with never-touched address space on Linux,
committed memory on Windows and resident memory on macOS.

Claude-Session: https://claude.ai/code/session_01H63zaWhkb5pAafx2WGjrVC

* feat(materialize): the worker stops after an idle timeout; the next job restarts it (TASK-2198 U3)

Dave's ruling on the worker process: GO, with an idle exit. A loaded worker
holds ~200 MB resident, which an install that rarely materializes should
not pay for all day.

PAD_MATERIALIZE_IDLE_TIMEOUT: default 5m, clamped to 30s-24h, 0 = never.
After each job the supervisor (re)arms a timer. When it fires, the stop
runs only while holding the job slot, so it never takes a worker with a job
in flight. It re-checks the idle time, forgets the worker WITHOUT recording
a death (backoff is untouched), and closes the worker's stdin so its read
loop ends with status 0, with a kill after a grace if it does not go. The
stop is logged at INFO, reason=idle. A job that arrives meanwhile waits for
the slot, finds no worker, and starts a new one with no backoff wait.

Legs: the idle exit after the timeout (INFO, status 0, no WARN); the next
job respawns and succeeds with no wait; an idle stop leaves the backoff
streak alone (crash, idle, crash waits 2s, not 4s); a job arriving during
the stop runs on a new worker; a timer that fires mid-job does not stop that
worker; NeverIdle arms no timer; env parsing and clamps. Mutations, each red
on its own: recording a death (3 legs), logging as a kill (1), not holding
the slot (1), never arming (3). Docs: the knob beside the other two, and
`idle` in the stop reasons.

Claude-Session: https://claude.ai/code/session_01H63zaWhkb5pAafx2WGjrVC

* fix(materialize): Close does not return while an idle-stopping worker still runs (TASK-2198 U3, codex r1)

An idle stop forgets the worker before it has exited, so a Close in that
window found no worker and returned with it still alive. The supervisor now
tracks the worker an idle stop is ending, and Close kills and reaps it. A
leg pins it, and goes red without the fix.

Claude-Session: https://claude.ai/code/session_01H63zaWhkb5pAafx2WGjrVC

* fix(materialize): an idle stop that Close ended says so (TASK-2198 U3, codex r2)

The idle stop stored its kill reason unconditionally, so it overwrote the
'closed' of a Close that had already killed the worker, and its stop line
said idle. It now claims the reason only if none is set, and the line
reports reason=closed. The Close-during-idle-stop leg asserts it.

Claude-Session: https://claude.ai/code/session_01H63zaWhkb5pAafx2WGjrVC

.github/workflows/ci.yml

@@ -201,7 +201,10 @@ jobs:
   native-smoke:
     name: Smoke (${{ matrix.os }})
     runs-on: ${{ matrix.os }}
-    timeout-minutes: 10
+    # 10 for the smoke itself, plus the materializer bundle build and the
+    # supervisor tests (TASK-2198 U3), which spawn real workers and compile
+    # the package's test binary.
+    timeout-minutes: 35
     defaults:
       run:
         shell: pwsh
@@ -240,14 +243,24 @@ jobs:
           go-version-file: go.mod
 
       - name: Create web build placeholder
-        # The materializer bundle (TASK-2198) is stubbed here too: this job
-        # builds and smoke-tests the binary, and nothing it runs loads the
-        # bundle. internal/materialize.New refuses a placeholder by name.
         run: |
           New-Item -ItemType Directory -Force web/build | Out-Null
           Set-Content web/build/index.html '<!doctype html>'
-          New-Item -ItemType Directory -Force web/build-materializer | Out-Null
-          Set-Content web/build-materializer/materializer.js '// placeholder'
+
+      - uses: actions/setup-node@820762786026740c76f36085b0efc47a31fe5020 # v7.0.0
+        with:
+          node-version: "24"
+          cache: "npm"
+          cache-dependency-path: web/package-lock.json
+
+      # The materializer bundle (TASK-2198) is built for real here, as in the
+      # go job, and not stubbed: the supervisor tests at the end of this job
+      # spawn real workers that load it, and internal/materialize.New refuses
+      # a placeholder by name. Only the bundle, not the whole web build (the
+      # go job's shape): web/build stays the placeholder above.
+      - name: Build the materializer bundle for embed
+        working-directory: web
+        run: npm ci && node scripts/build-materializer.mjs
 
       - name: Build binary
         run: go build -o "${{ matrix.binary }}" ./cmd/pad
@@ -349,6 +362,24 @@ jobs:
           }
           $global:LASTEXITCODE = 0
 
+      # The memory-cap tests below allocate 2 GiB against a 1600 MiB cap, and
+      # the adversarial one drives a real worker to a 2 GiB cap. Logged so a
+      # failure there can be read against what the runner actually had.
+      - name: Runner memory
+        run: |
+          if ($IsMacOS) { sysctl hw.memsize hw.ncpu } else { Get-CimInstance Win32_ComputerSystem | Select-Object TotalPhysicalMemory, NumberOfLogicalProcessors | Format-List }
+
+      # The materializer supervisor (TASK-2198 U3) on each OS's REAL memory
+      # cap, which only these runners have: macOS's parent-side RSS watchdog
+      # (proc_info) and Windows' Job Object with a suspended start. The Go job
+      # covers Linux's RLIMIT_AS. PAD_MATERIALIZE_TEST_NO_SKIP turns the cap
+      # tests' skips into failures, so a cap that stopped running cannot pass
+      # as "skipped".
+      - name: Materializer supervisor and memory cap (TASK-2198 U3)
+        env:
+          PAD_MATERIALIZE_TEST_NO_SKIP: "1"
+        run: go test ./internal/materialize/ -run '^(TestSupervisor.*|TestParseMemLimit|TestConfigFromEnv|TestParsePSRSS|TestPSRSSLive|TestParseStatVsize|TestStderrOOMMarkers|TestDeepNestFrameShallow)$' -count=1 -timeout 20m -v
+
   go-postgres:
     # Backstop only. A hung TEST is caught by -timeout=45m on the steps
     # below, which fails fast AND prints the goroutine dump that tells

cmd/pad/cmd_materialize.go

@@ -26,6 +26,9 @@ func materializeWorkerCmd() *cobra.Command {
 		SilenceUsage:  true,
 		SilenceErrors: false,
 		RunE: func(cmd *cobra.Command, args []string) error {
+			// Before the bundle loads: a worker whose server died must not
+			// run on without its deadline kill and memory watchdog.
+			materialize.ExitWhenOrphaned()
 			return materialize.RunWorker(os.Stdin, os.Stdout, pad.MaterializerJS)
 		},
 	}

docs/deployment.md

@@ -1222,6 +1222,72 @@ address inside `PAD_TRUSTED_PROXIES` (the proxy sends no `X-Forwarded-For` or
 - **The host check does not replace TLS.** It closes DNS rebinding; it does not
   protect tokens in transit. That is why http is token-only.
 
+### Op-log materializer worker
+
+**Not active yet.** The worker and the two variables below take effect once
+the materializer is wired into the server (TASK-2198 U4); until then they are
+inert, and setting them changes nothing.
+
+When a browser tab closes without saving, its last edits are only in the
+item's collaborative op-log. Pad can turn that op-log back into the item's
+markdown on the server by running the editor's own JavaScript. It does this in
+a **separate worker process** (the same `pad` binary, started as
+`pad __materialize-worker`), never inside the server: a crafted update can
+make that JavaScript use gigabytes of memory, and in a separate process that
+kills one worker instead of the server.
+
+- The worker starts on the first job, not at server start, and then stays up
+  while jobs keep coming (loading it takes a few seconds). Jobs run one at a
+  time.
+- A job that runs past the timeout is failed and the worker is killed. The
+  next job starts a new one.
+- If the server process dies, its worker exits too, so it never runs on
+  without its time and memory limits. On Linux it exits at once; on macOS
+  within about 200ms; on Windows at once.
+- A worker with no job for the idle timeout (5 minutes by default) is stopped
+  cleanly, because a loaded worker holds about 200 MB of resident memory for as
+  long as it runs. The next job starts a new one, with no wait. An idle stop is
+  not a death and does not count toward the restart wait below.
+- A worker that dies is restarted on the next job, waiting 1s after the first
+  death and doubling after each further death, up to 5 minutes. A worker that
+  stayed up for a minute resets the wait.
+- Every worker start logs the timeout, the worker's measured memory
+  baseline, the memory limit and the resulting cap, and every stop logs why
+  (`deadline`, `memory`, `exit`, `protocol`, `start`, `canceled`, `closed`,
+  or `idle`, which is logged at INFO rather than as a warning). The worker's
+  stderr goes to the server log, one line per record, tagged with the worker's
+  pid.
+
+| Variable | Default | Description |
+|----------|---------|-------------|
+| `PAD_MATERIALIZE_TIMEOUT` | `2s` | Per-job time limit, in Go duration syntax (`2s`, `1500ms`). Clamped to 250ms–60s. |
+| `PAD_MATERIALIZE_IDLE_TIMEOUT` | `5m` | How long the worker may sit with no job before it is stopped (it holds about 200 MB resident while it runs), in Go duration syntax. `0` means never stop it. Clamped to 30s–24h. |
+| `PAD_MATERIALIZE_MEM_LIMIT` | `2GiB` | How much memory one job may add to the loaded worker (see below): a whole number of bytes, or a whole number followed by `KiB`, `MiB` or `GiB` with no space (`2048MiB`, `2GiB`, `2147483648`). Decimal units such as `GB` are refused. Clamped to 256MiB–16GiB. |
+
+A value that cannot be parsed is ignored with a warning, and the default is
+used. A value outside the range is clamped with a warning.
+
+The memory limit is **growth beyond the loaded worker**. Once the worker has
+loaded, and before it is given any job, Pad measures its memory use (the
+baseline) and caps it at baseline + limit. What "memory" means depends on the
+OS, and the three are not the same measure:
+
+- **Linux:** virtual **address space** (`RLIMIT_AS`, set on the worker process
+  only). A loaded worker already holds about 1.4 GB of address space, more on
+  hosts with many CPUs; the limit is added on top of whatever it holds.
+- **macOS:** **resident memory** (RSS), watched by Pad every 100ms, because
+  macOS does not enforce `RLIMIT_AS`. A worker can go over the cap for up to
+  one check interval before it is stopped.
+- **Windows:** the worker's **committed memory**, as a Job Object limit. The
+  worker is started suspended and put in the job before it runs.
+
+Linux, macOS and Windows are the supported platforms. **If no memory cap can
+be established** (any other OS, or the baseline measurement failing or
+reading an implausible value), Pad does not materialize at all: no worker is
+given a job, one warning is logged, and items keep today's behaviour until a
+worker can be capped. That is safe, because materialization is a recovery
+enhancement; the server does not need it to run.
+
 ### Email (Optional)
 
 Email enables sending workspace invitation links. Without it, users can still join via CLI invite codes.

internal/materialize/adversarial_test.go

@@ -0,0 +1,123 @@
+package materialize
+
+import (
+	"context"
+	"errors"
+	"os"
+	"strconv"
+	"testing"
+	"time"
+)
+
+func putVarUint(b []byte, n uint64) []byte {
+	for n >= 0x80 {
+		b = append(b, byte(n)|0x80)
+		n >>= 7
+	}
+	return append(b, byte(n))
+}
+
+func putVarString(b []byte, s string) []byte {
+	b = putVarUint(b, uint64(len(s)))
+	return append(b, s...)
+}
+
+// deepNestFrame builds one y-protocols Update frame whose Yjs update (v1
+// encoding) nests depth XmlElements named node, each the only child of the
+// one before, under the 'default' fragment the editor binds. The update is
+// tiny — about 12 bytes per level — and what it costs to materialize is not:
+// every level is a recursion in the Yjs → ProseMirror conversion and in the
+// markdown serializer, and nested blockquotes repeat their "> " prefix on
+// every line, so the output is quadratic in depth.
+func deepNestFrame(depth int, node string) []byte {
+	const client = 7
+	var u []byte
+	u = putVarUint(u, 1) // one client's structs
+	u = putVarUint(u, uint64(depth+3))
+	u = putVarUint(u, client)
+	u = putVarUint(u, 0) // first clock
+	parent := func(i int) {
+		if i == 0 {
+			u = putVarUint(u, 1) // parent is a root type, named:
+			u = putVarString(u, "default")
+			return
+		}
+		u = putVarUint(u, 0) // parent is the item (client, i-1)
+		u = putVarUint(u, client)
+		u = putVarUint(u, uint64(i-1))
+	}
+	element := func(i int, name string) {
+		u = append(u, 7) // info: ContentType, no origins, no parentSub
+		parent(i)
+		u = putVarUint(u, 3) // YXmlElement
+		u = putVarString(u, name)
+	}
+	for i := 0; i < depth; i++ {
+		element(i, node)
+	}
+	// Innermost: a paragraph holding the text "x", so the tree is valid
+	// content all the way down and nothing is dropped as empty.
+	element(depth, "paragraph")
+	u = append(u, 7) // ContentType
+	parent(depth + 1)
+	u = putVarUint(u, 6) // YXmlText
+	u = append(u, 4)     // info: ContentString
+	parent(depth + 2)
+	u = putVarString(u, "x")
+	u = putVarUint(u, 0) // empty delete set
+	f := []byte{0, 2}    // MESSAGE_SYNC, messageYjsUpdate
+	f = putVarUint(f, uint64(len(u)))
+	return append(f, u...)
+}
+
+// The generator is well-formed: a shallow nest materializes to what the
+// editor makes of it.
+func TestDeepNestFrameShallow(t *testing.T) {
+	r := runner(t)
+	md, err := r.Materialize(context.Background(), Job{Rows: [][]byte{deepNestFrame(3, "blockquote")}, SchemaVersion: r.SchemaVersion()})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if md != "> > > x" {
+		t.Fatalf("3 nested blockquotes: %q, want %q", md, "> > > x")
+	}
+}
+
+// TestSupervisorAdversarialDeepNesting sends the spike's shape — a ~1 MB
+// deep-nesting update — to a REAL capped worker. In-process this is the input
+// that drove Go past 5 GB into its fatal out-of-memory. Here the worker must
+// fail the job (killed on its cap, or on the deadline), the parent must be
+// untouched, and the next job must succeed on a fresh worker.
+func TestSupervisorAdversarialDeepNesting(t *testing.T) {
+	if raceEnabled {
+		skipCapTest(t, "the race runtime cannot start under an address-space cap")
+	}
+	if testing.Short() {
+		t.Skip("spawns a real worker and drives it to its cap")
+	}
+	depth := 60_000 // a 1,063,536-byte frame: the spike's ~1 MB
+	if v := os.Getenv("PAD_TEST_NEST_DEPTH"); v != "" {
+		depth, _ = strconv.Atoi(v)
+	}
+	frame := deepNestFrame(depth, "blockquote")
+	h := newHarness(t, "worker", func(c *SupervisorConfig) {
+		c.MemLimit = 1 << 30
+		c.Timeout = 30 * time.Second
+		c.capOverride = nil
+	})
+	r := runner(t)
+	start := time.Now()
+	md, err := h.s.Materialize(context.Background(), Job{Rows: [][]byte{frame}, SchemaVersion: r.SchemaVersion()})
+	t.Logf("depth %d, frame %d bytes: %d bytes of markdown, err %v, after %s", depth, len(frame), len(md), err, time.Since(start).Round(time.Millisecond))
+	if !errors.Is(err, ErrMemoryLimit) && !errors.Is(err, ErrDeadline) && !errors.Is(err, ErrChildDied) {
+		t.Fatalf("got %d bytes, %v; want the worker to fail the job", len(md), err)
+	}
+	for _, l := range h.log.find("materialize worker stopped") {
+		t.Log(l)
+	}
+	ok := loadCorpus(t)[0]
+	got, err := h.s.Materialize(context.Background(), ok.job(t, r.SchemaVersion()))
+	if err != nil || got != ok.Expected {
+		t.Fatalf("job after the adversarial one: %q, %v", got, err)
+	}
+}

internal/materialize/inflate_darwin_test.go

@@ -0,0 +1,30 @@
+package materialize
+
+import "syscall"
+
+var inflated []byte
+
+// inflateBaseline raises the worker's macOS cap measure — resident memory —
+// by mib: an anonymous mapping, every page written with noise (macOS
+// compresses idle pages of zeros out of the resident set).
+func inflateBaseline(mib int) error {
+	b, err := syscall.Mmap(-1, 0, mib<<20, syscall.PROT_READ|syscall.PROT_WRITE, syscall.MAP_PRIVATE|syscall.MAP_ANON)
+	if err != nil {
+		return err
+	}
+	x := uint64(0x9E3779B97F4A7C15)
+	for i := 0; i+8 <= len(b); i += 8 {
+		x ^= x << 13
+		x ^= x >> 7
+		x ^= x << 17
+		for k := 0; k < 8; k++ {
+			b[i+k] = byte(x >> (8 * k))
+		}
+	}
+	inflated = b
+	return nil
+}
+
+// inflateMiB is how much the big-baseline leg inflates on this OS (resident,
+// on a 7.5 GB runner, beside the leg's own 1 GiB allocation).
+const inflateMiB = 512

internal/materialize/inflate_linux_test.go

@@ -0,0 +1,17 @@
+package materialize
+
+import "syscall"
+
+var inflated []byte
+
+// inflateBaseline raises the worker's Linux cap measure — address space —
+// by mib without touching it or creating threads: one PROT_NONE,
+// MAP_NORESERVE mapping, never touched.
+func inflateBaseline(mib int) error {
+	b, err := syscall.Mmap(-1, 0, mib<<20, syscall.PROT_NONE, syscall.MAP_PRIVATE|syscall.MAP_ANON|syscall.MAP_NORESERVE)
+	inflated = b
+	return err
+}
+
+// inflateMiB is how much the big-baseline leg inflates on this OS.
+const inflateMiB = 1536

internal/materialize/inflate_other_test.go

@@ -0,0 +1,7 @@
+//go:build !linux && !darwin && !windows
+
+package materialize
+
+func inflateBaseline(int) error { return nil }
+
+const inflateMiB = 0

internal/materialize/inflate_windows_test.go

@@ -0,0 +1,14 @@
+package materialize
+
+import "golang.org/x/sys/windows"
+
+// inflateBaseline raises the worker's Windows cap measure — committed memory
+// — by mib: one committed, never-touched region (commit charge counts it
+// whether or not it is touched).
+func inflateBaseline(mib int) error {
+	_, err := windows.VirtualAlloc(0, uintptr(mib)<<20, windows.MEM_RESERVE|windows.MEM_COMMIT, windows.PAGE_READWRITE)
+	return err
+}
+
+// inflateMiB is how much the big-baseline leg inflates on this OS.
+const inflateMiB = 1536

internal/materialize/measure_linux_test.go

@@ -0,0 +1,88 @@
+package materialize
+
+import (
+	"context"
+	"os"
+	"strconv"
+	"strings"
+	"testing"
+	"time"
+)
+
+// procThreads reads the Threads: count of /proc/<pid>/status.
+func procThreads(t *testing.T, pid int) int {
+	t.Helper()
+	b, err := os.ReadFile("/proc/" + strconv.Itoa(pid) + "/status")
+	if err != nil {
+		t.Fatal(err)
+	}
+	for _, l := range strings.Split(string(b), "\n") {
+		if strings.HasPrefix(l, "Threads:") {
+			n, _ := strconv.Atoi(strings.Fields(l)[1])
+			return n
+		}
+	}
+	return -1
+}
+
+// procStatusKB reads one "Name:   N kB" line of /proc/<pid>/status.
+func procStatusKB(t *testing.T, pid int, name string) uint64 {
+	t.Helper()
+	b, err := os.ReadFile("/proc/" + strconv.Itoa(pid) + "/status")
+	if err != nil {
+		t.Fatal(err)
+	}
+	for _, l := range strings.Split(string(b), "\n") {
+		if strings.HasPrefix(l, name+":") {
+			n, _ := strconv.ParseUint(strings.Fields(l)[1], 10, 64)
+			return n << 10
+		}
+	}
+	t.Fatalf("no %s", name)
+	return 0
+}
+
+// TestSupervisorBaselineMeasurements records, for a REAL loaded worker, its
+// address-space baseline and how much address space and resident memory the
+// corpus and a moderate adversarial document add — the evidence behind
+// MinMemLimit and MaxPlausibleBaseline. Run with PAD_MATERIALIZE_MEASURE=1.
+func TestSupervisorBaselineMeasurements(t *testing.T) {
+	if os.Getenv("PAD_MATERIALIZE_MEASURE") == "" {
+		t.Skip("measurement; set PAD_MATERIALIZE_MEASURE=1")
+	}
+	r := runner(t)
+	cases := loadCorpus(t)
+	for _, procs := range []string{"2", "8", "64"} {
+		var pid int
+		h := newHarness(t, "worker", func(c *SupervisorConfig) {
+			c.Timeout = 60 * time.Second
+			c.capOverride = fakeCap
+			c.extraEnv = []string{"GOMAXPROCS=" + procs}
+			c.onSpawn = func(p int) { pid = p }
+		})
+		if _, err := h.s.Materialize(context.Background(), Job{SchemaVersion: r.SchemaVersion()}); err != nil {
+			t.Fatal(err)
+		}
+		vsz0, rss0 := procStatusKB(t, pid, "VmSize"), procStatusKB(t, pid, "VmRSS")
+		th0 := procThreads(t, pid)
+		for _, tc := range cases {
+			if _, err := h.s.Materialize(context.Background(), tc.job(t, r.SchemaVersion())); err != nil {
+				t.Fatal(err)
+			}
+		}
+		vszC, rssC := procStatusKB(t, pid, "VmPeak"), procStatusKB(t, pid, "VmHWM")
+		line := "GOMAXPROCS=" + procs + ": baseline VmSize " + formatBytes(vsz0) + " (" + strconv.FormatUint(vsz0>>20, 10) + " MiB), VmRSS " + strconv.FormatUint(rss0>>20, 10) + " MiB; after " + strconv.Itoa(len(cases)) + " corpus cases: +" + strconv.FormatUint((vszC-vsz0)>>20, 10) + " MiB address space (peak), +" + strconv.FormatInt((int64(rssC)-int64(rss0))>>20, 10) + " MiB resident (peak)"
+		for _, depth := range []int{1000, 2000, 4000} {
+			_, err := h.s.Materialize(context.Background(), Job{Rows: [][]byte{deepNestFrame(depth, "blockquote")}, SchemaVersion: r.SchemaVersion()})
+			if err != nil {
+				line += "; depth " + strconv.Itoa(depth) + ": " + err.Error()
+				break
+			}
+			line += "; depth " + strconv.Itoa(depth) + ": +" + strconv.FormatUint((procStatusKB(t, pid, "VmPeak")-vsz0)>>20, 10) + " MiB AS, +" + strconv.FormatInt((int64(procStatusKB(t, pid, "VmHWM"))-int64(rss0))>>20, 10) + " MiB RSS"
+		}
+		th1 := procThreads(t, pid)
+		line += "; threads " + strconv.Itoa(th0) + " at baseline -> " + strconv.Itoa(th1) + " after all jobs (+" + strconv.Itoa(th1-th0) + ")"
+		t.Log(line)
+		_ = h.s.Close()
+	}
+}

internal/materialize/memcap_darwin.go

@@ -0,0 +1,106 @@
+package materialize
+
+import (
+	"errors"
+	"log/slog"
+	"strconv"
+	"sync"
+	"syscall"
+	"unsafe"
+
+	"golang.org/x/sys/unix"
+)
+
+// macOS: RLIMIT_AS is accepted but not enforced, so the cap is a parent-side
+// watchdog. Once the worker has loaded its bundle and before it is sent any
+// job, the parent reads its resident set size as the BASELINE; from then on,
+// every 100 ms, it kills the worker when its resident set exceeds baseline +
+// limit (ErrMemoryLimit). What it measures is RESIDENT
+// memory, where Linux caps address space and Windows committed memory.
+//
+// The window is the poll interval: a child can allocate for up to 100 ms
+// past the cap before it is seen. An allocation fast enough to exhaust the
+// machine inside 100 ms is not stopped; the kill still ends it.
+//
+// The read is proc_info(PROC_INFO_CALL_PIDINFO, pid, PROC_PIDTASKINFO) — the
+// syscall behind libproc's proc_pidinfo — with no fork. If it fails, the
+// watchdog falls back to `ps -o rss= -p <pid>` (a fork+exec per poll) and
+// says so once in the log.
+var platformMemCap = memCapImpl{
+	mechanism: "rss_watchdog",
+	rss:       darwinRSS,
+}
+
+const (
+	procInfoCallPidinfo = 2
+	procPidTaskInfo     = 4
+)
+
+// procTaskInfo is struct proc_taskinfo (sys/proc_info.h), 96 bytes.
+type procTaskInfo struct {
+	VirtualSize      uint64
+	ResidentSize     uint64
+	TotalUser        uint64
+	TotalSystem      uint64
+	ThreadsUser      uint64
+	ThreadsSystem    uint64
+	Policy           int32
+	Faults           int32
+	Pageins          int32
+	CowFaults        int32
+	MessagesSent     int32
+	MessagesReceived int32
+	SyscallsMach     int32
+	SyscallsUnix     int32
+	Csw              int32
+	Threadnum        int32
+	Numrunning       int32
+	Priority         int32
+}
+
+var (
+	psFallbackOnce sync.Once
+	useSyscall     = true
+	useSyscallMu   sync.Mutex
+)
+
+func darwinRSS(pid int) (uint64, error) {
+	useSyscallMu.Lock()
+	sys := useSyscall
+	useSyscallMu.Unlock()
+	if sys {
+		n, err := procPidRSS(pid)
+		if err == nil {
+			return n, nil
+		}
+		if errors.Is(err, syscall.ESRCH) {
+			return 0, err // the child is gone; not a reason to fall back
+		}
+		useSyscallMu.Lock()
+		useSyscall = false
+		useSyscallMu.Unlock()
+		psFallbackOnce.Do(func() {
+			logPSFallback(err)
+		})
+	}
+	return psRSS(pid)
+}
+
+func procPidRSS(pid int) (uint64, error) {
+	var ti procTaskInfo
+	size := unsafe.Sizeof(ti)
+	r1, _, errno := syscall.Syscall6(unix.SYS_PROC_INFO, procInfoCallPidinfo, uintptr(pid), procPidTaskInfo, 0,
+		uintptr(unsafe.Pointer(&ti)), size)
+	if errno != 0 {
+		return 0, errno
+	}
+	if r1 != size {
+		return 0, errors.New("proc_info: short PROC_PIDTASKINFO read of " + strconv.Itoa(int(r1)) + " bytes")
+	}
+	return ti.ResidentSize, nil
+}
+
+func logPSFallback(err error) {
+	slog.Warn("materialize: proc_info unavailable; the memory watchdog falls back to ps (a fork per 100 ms poll)",
+		"error", err)
+}

internal/materialize/memcap_linux.go

@@ -0,0 +1,63 @@
+package materialize
+
+import (
+	"fmt"
+	"os"
+	"os/exec"
+	"strconv"
+
+	"golang.org/x/sys/unix"
+)
+
+// Linux: RLIMIT_AS on the CHILD, RELATIVE to the loaded worker. Once the
+// worker has loaded its bundle (the readiness probe has answered) and before
+// it is sent any job, the parent reads the worker's address-space size
+// (/proc/<pid>/stat vsize) and sets RLIMIT_AS to that baseline plus the
+// limit with prlimit(2), then reads it back. The limit therefore bounds how
+// much ADDRESS SPACE a job may add to the loaded worker, whatever the host's
+// runtime reserved before it (an absolute cap measured on one machine failed
+// a 64 MiB allocation on CI runners, whose baseline was larger).
+//
+// Not setrlimit in the parent before exec: that would cap the server. Not the
+// child capping itself: the parent could not verify it.
+//
+// The window: the worker's own startup and bundle load run uncapped. That is
+// trusted code; the worker holds no job until the cap is set and confirmed.
+// If the baseline cannot be read, the job is refused (ErrNoMemoryCap).
+//
+// A breach fails the allocation and the Go runtime dies with "fatal error:
+// runtime: out of memory", which the stderr log recognises.
+var platformMemCap = memCapImpl{
+	mechanism: "rlimit_as",
+	attach: func(cmd *exec.Cmd, _ func()) (*capHandle, error) {
+		pid := cmd.Process.Pid
+		return &capHandle{
+			baseline: func() (uint64, error) { return procVsize(pid) },
+			set: func(capBytes uint64) error {
+				lim := unix.Rlimit{Cur: capBytes, Max: capBytes}
+				if err := unix.Prlimit(pid, unix.RLIMIT_AS, &lim, nil); err != nil {
+					return err
+				}
+				var got unix.Rlimit
+				if err := unix.Prlimit(pid, unix.RLIMIT_AS, nil, &got); err != nil {
+					return err
+				}
+				if got.Cur != capBytes {
+					return fmt.Errorf("RLIMIT_AS reads %d after setting %d", got.Cur, capBytes)
+				}
+				return nil
+			},
+		}, nil
+	},
+}
+
+// procVsize reads a process's virtual memory size in bytes: field 23 of
+// /proc/<pid>/stat, counted after the parenthesised command name (which may
+// itself hold spaces or parentheses).
+func procVsize(pid int) (uint64, error) {
+	b, err := os.ReadFile("/proc/" + strconv.Itoa(pid) + "/stat")
+	if err != nil {
+		return 0, err
+	}
+	return parseStatVsize(string(b))
+}

internal/materialize/memcap_other.go

@@ -0,0 +1,7 @@
+//go:build !linux && !windows && !darwin
+
+package materialize
+
+// Any other OS: no way to cap the worker, so the Supervisor never starts one
+// and refuses every job with ErrNoMemoryCap (one WARN per Supervisor).
+var platformMemCap = memCapImpl{mechanism: "none", unsupported: true}

internal/materialize/memcap_ps.go

@@ -0,0 +1,34 @@
+//go:build darwin || linux
+
+package materialize
+
+import (
+	"errors"
+	"os/exec"
+	"strconv"
+	"strings"
+)
+
+// psRSS reads a process's resident set size with `ps -o rss= -p <pid>`, the
+// macOS watchdog's fallback when proc_info is unavailable. Built on Linux too
+// so its parser and cost are tested where the tests run.
+func psRSS(pid int) (uint64, error) {
+	out, err := exec.Command("ps", "-o", "rss=", "-p", strconv.Itoa(pid)).Output()
+	if err != nil {
+		return 0, err
+	}
+	return parsePSRSS(out)
+}
+
+// parsePSRSS reads ps's rss column: KiB, one number, padded.
+func parsePSRSS(out []byte) (uint64, error) {
+	s := strings.TrimSpace(string(out))
+	if s == "" {
+		return 0, errors.New("ps: no such process")
+	}
+	kib, err := strconv.ParseUint(s, 10, 64)
+	if err != nil {
+		return 0, err
+	}
+	return kib << 10, nil
+}

internal/materialize/memcap_ps_test.go

@@ -0,0 +1,50 @@
+//go:build darwin || linux
+
+package materialize
+
+import (
+	"os"
+	"os/exec"
+	"testing"
+)
+
+// The parser alone: fixed strings, nothing executed, so it runs everywhere
+// including the Nix build sandbox.
+func TestParsePSRSS(t *testing.T) {
+	if n, err := parsePSRSS([]byte("  123456\n")); err != nil || n != 123456<<10 {
+		t.Fatalf("%d %v", n, err)
+	}
+	for _, bad := range []string{"", "\n", "abc", "12 34"} {
+		if _, err := parsePSRSS([]byte(bad)); err == nil {
+			t.Errorf("%q accepted", bad)
+		}
+	}
+}
+
+// TestPSRSSLive runs the real `ps` fallback against this process. It skips
+// when there is no `ps` to run (the Nix build sandbox has none), which is
+// acceptable: `ps` is only the macOS watchdog's FALLBACK, the production path
+// is proc_info. PAD_MATERIALIZE_TEST_NO_SKIP still turns the skip into a
+// failure where CI sets it.
+func TestPSRSSLive(t *testing.T) {
+	if _, err := exec.LookPath("ps"); err != nil {
+		skipCapTest(t, "no `ps` on PATH (the Nix build sandbox has none): "+err.Error())
+	}
+	n, err := psRSS(os.Getpid())
+	if err != nil || n == 0 {
+		t.Fatalf("ps on self: %d %v", n, err)
+	}
+}
+
+// BenchmarkPSRSS is the cost of one poll by the macOS watchdog's ps fallback.
+func BenchmarkPSRSS(b *testing.B) {
+	if _, err := exec.LookPath("ps"); err != nil {
+		b.Skip("no `ps` on PATH: " + err.Error())
+	}
+	pid := os.Getpid()
+	for b.Loop() {
+		if _, err := psRSS(pid); err != nil {
+			b.Fatal(err)
+		}
+	}
+}

internal/materialize/memcap_statparse.go

@@ -0,0 +1,22 @@
+package materialize
+
+import (
+	"fmt"
+	"strconv"
+	"strings"
+)
+
+// parseStatVsize extracts vsize (field 23) from a /proc/<pid>/stat line.
+// Platform-neutral so it is tested everywhere.
+func parseStatVsize(stat string) (uint64, error) {
+	i := strings.LastIndexByte(stat, ')')
+	if i < 0 {
+		return 0, fmt.Errorf("stat: no command name in %q", stat)
+	}
+	f := strings.Fields(stat[i+1:])
+	// f[0] is field 3 (state), so field 23 is f[20].
+	if len(f) < 21 {
+		return 0, fmt.Errorf("stat: %d fields after the command name", len(f))
+	}
+	return strconv.ParseUint(f[20], 10, 64)
+}

internal/materialize/memcap_windows.go

@@ -0,0 +1,205 @@
+package materialize
+
+import (
+	"fmt"
+	"math"
+	"os/exec"
+	"syscall"
+	"unsafe"
+
+	"golang.org/x/sys/windows"
+)
+
+// Windows: a Job Object. The worker is created SUSPENDED
+// (CREATE_SUSPENDED), assigned to a job with JOB_OBJECT_LIMIT_KILL_ON_JOB_CLOSE
+// (so it dies with the server), and only then resumed: it never runs outside
+// the job. Once it has loaded its bundle and before it is sent any job, the
+// parent reads its COMMITTED memory (the commit charge,
+// PROCESS_MEMORY_COUNTERS.PagefileUsage) and sets
+// JOB_OBJECT_LIMIT_PROCESS_MEMORY to that baseline plus the limit — the same
+// quantity the job limit measures. The bundle load itself runs without the
+// memory limit; it is trusted and holds no job.
+//
+// os/exec does not hand back the primary thread's handle, so the thread is
+// found by a toolhelp snapshot of the (brand new, single-threaded) process
+// and resumed by id. Any failure on the way kills the still-suspended child.
+//
+// A breach fails the allocation (VirtualAlloc, errno=1455). The Go runtime
+// does NOT reliably die of that — CI saw a worker whose 1 GiB allocation was
+// refused sit until the job deadline — so the supervisor does not wait for
+// it: the job is associated with an I/O completion port, and
+// JOB_OBJECT_MSG_PROCESS_MEMORY_LIMIT for the worker's pid kills the worker
+// at once and fails the job with ErrMemoryLimit. The runtime's own "out of
+// memory" / errno=1455 on stderr remains a second signal.
+var platformMemCap = memCapImpl{
+	mechanism: "job_object",
+	prepare: func(cmd *exec.Cmd) {
+		if cmd.SysProcAttr == nil {
+			cmd.SysProcAttr = &syscall.SysProcAttr{}
+		}
+		cmd.SysProcAttr.CreationFlags |= windows.CREATE_SUSPENDED
+	},
+	attach: attachJobObject,
+}
+
+type processMemoryCounters struct {
+	Cb                         uint32
+	PageFaultCount             uint32
+	PeakWorkingSetSize         uintptr
+	WorkingSetSize             uintptr
+	QuotaPeakPagedPoolUsage    uintptr
+	QuotaPagedPoolUsage        uintptr
+	QuotaPeakNonPagedPoolUsage uintptr
+	QuotaNonPagedPoolUsage     uintptr
+	PagefileUsage              uintptr
+	PeakPagefileUsage          uintptr
+}
+
+var procK32GetProcessMemoryInfo = windows.NewLazySystemDLL("kernel32.dll").NewProc("K32GetProcessMemoryInfo")
+
+// Job-object notification messages (winnt.h) and the association struct,
+// which x/sys/windows does not define.
+const (
+	jobObjectMsgProcessMemoryLimit = 9
+	jobObjectMsgJobMemoryLimit     = 10
+)
+
+type jobObjectAssociateCompletionPort struct {
+	CompletionKey  uintptr
+	CompletionPort windows.Handle
+}
+
+func attachJobObject(cmd *exec.Cmd, onLimit func()) (h *capHandle, err error) {
+	pid := uint32(cmd.Process.Pid)
+	job, err := windows.CreateJobObject(nil, nil)
+	if err != nil {
+		return nil, fmt.Errorf("CreateJobObject: %w", err)
+	}
+	var proc, port windows.Handle
+	defer func() {
+		if err != nil {
+			if proc != 0 {
+				windows.CloseHandle(proc)
+			}
+			if port != 0 {
+				windows.CloseHandle(port)
+			}
+			windows.CloseHandle(job)
+		}
+	}()
+	if err = setJobLimits(job, 0); err != nil {
+		return nil, err
+	}
+	port, err = windows.CreateIoCompletionPort(windows.InvalidHandle, 0, 0, 1)
+	if err != nil {
+		return nil, fmt.Errorf("CreateIoCompletionPort: %w", err)
+	}
+	assoc := jobObjectAssociateCompletionPort{CompletionKey: uintptr(job), CompletionPort: port}
+	if _, err = windows.SetInformationJobObject(job, windows.JobObjectAssociateCompletionPortInformation,
+		uintptr(unsafe.Pointer(&assoc)), uint32(unsafe.Sizeof(assoc))); err != nil {
+		return nil, fmt.Errorf("associate completion port: %w", err)
+	}
+	proc, err = windows.OpenProcess(windows.PROCESS_SET_QUOTA|windows.PROCESS_TERMINATE|
+		windows.PROCESS_QUERY_LIMITED_INFORMATION|windows.PROCESS_VM_READ, false, pid)
+	if err != nil {
+		return nil, fmt.Errorf("OpenProcess: %w", err)
+	}
+	if err = windows.AssignProcessToJobObject(job, proc); err != nil {
+		return nil, fmt.Errorf("AssignProcessToJobObject: %w", err)
+	}
+	if err = resumeProcess(pid); err != nil {
+		return nil, err
+	}
+	go watchJobPort(port, pid, onLimit)
+	return &capHandle{
+		baseline: func() (uint64, error) {
+			var c processMemoryCounters
+			c.Cb = uint32(unsafe.Sizeof(c))
+			r, _, e := procK32GetProcessMemoryInfo.Call(uintptr(proc), uintptr(unsafe.Pointer(&c)), uintptr(c.Cb))
+			if r == 0 {
+				return 0, fmt.Errorf("K32GetProcessMemoryInfo: %w", e)
+			}
+			return uint64(c.PagefileUsage), nil
+		},
+		set: func(capBytes uint64) error { return setJobLimits(job, capBytes) },
+		release: func() {
+			windows.CloseHandle(port) // ends watchJobPort
+			windows.CloseHandle(proc)
+			windows.CloseHandle(job)
+		},
+	}, nil
+}
+
+// setJobLimits sets KILL_ON_JOB_CLOSE, plus a per-process committed-memory
+// limit when limit > 0.
+func setJobLimits(job windows.Handle, limit uint64) error {
+	var info windows.JOBOBJECT_EXTENDED_LIMIT_INFORMATION
+	info.BasicLimitInformation.LimitFlags = windows.JOB_OBJECT_LIMIT_KILL_ON_JOB_CLOSE
+	if limit > 0 {
+		if limit > uint64(math.MaxUint) {
+			limit = uint64(math.MaxUint) // a 32-bit build: the most it can express
+		}
+		info.BasicLimitInformation.LimitFlags |= windows.JOB_OBJECT_LIMIT_PROCESS_MEMORY
+		info.ProcessMemoryLimit = uintptr(limit)
+	}
+	if _, err := windows.SetInformationJobObject(job, windows.JobObjectExtendedLimitInformation,
+		uintptr(unsafe.Pointer(&info)), uint32(unsafe.Sizeof(info))); err != nil {
+		return fmt.Errorf("SetInformationJobObject: %w", err)
+	}
+	return nil
+}
+
+// resumeProcess resumes every thread of pid (a CREATE_SUSPENDED process has
+// exactly one).
+func resumeProcess(pid uint32) error {
+	snap, err := windows.CreateToolhelp32Snapshot(windows.TH32CS_SNAPTHREAD, 0)
+	if err != nil {
+		return fmt.Errorf("CreateToolhelp32Snapshot: %w", err)
+	}
+	defer windows.CloseHandle(snap)
+	var te windows.ThreadEntry32
+	te.Size = uint32(unsafe.Sizeof(te))
+	resumed := 0
+	for err = windows.Thread32First(snap, &te); err == nil; err = windows.Thread32Next(snap, &te) {
+		if te.OwnerProcessID != pid {
+			continue
+		}
+		th, oerr := windows.OpenThread(windows.THREAD_SUSPEND_RESUME, false, te.ThreadID)
+		if oerr != nil {
+			return fmt.Errorf("OpenThread: %w", oerr)
+		}
+		_, rerr := windows.ResumeThread(th)
+		windows.CloseHandle(th)
+		if rerr != nil {
+			return fmt.Errorf("ResumeThread: %w", rerr)
+		}
+		resumed++
+	}
+	if resumed == 0 {
+		return fmt.Errorf("no thread of process %d to resume", pid)
+	}
+	return nil
+}
+
+// watchJobPort reads the job's notifications until the port is closed and
+// calls onLimit when the worker (pid) hits its memory limit. For job
+// messages the OVERLAPPED pointer carries the process id.
+func watchJobPort(port windows.Handle, pid uint32, onLimit func()) {
+	for {
+		var msg uint32
+		var key uintptr
+		var ov *windows.Overlapped
+		if err := windows.GetQueuedCompletionStatus(port, &msg, &key, &ov, windows.INFINITE); err != nil {
+			if ov == nil {
+				return // the port was closed (ERROR_ABANDONED_WAIT_0) or failed
+			}
+			continue
+		}
+		switch msg {
+		case jobObjectMsgProcessMemoryLimit, jobObjectMsgJobMemoryLimit:
+			if uint32(uintptr(unsafe.Pointer(ov))) == pid {
+				onLimit()
+			}
+		}
+	}
+}

internal/materialize/orphan.go

@@ -0,0 +1,31 @@
+package materialize
+
+import (
+	"os"
+	"os/exec"
+	"strconv"
+)
+
+// EnvParentPID carries the supervising process's pid to the worker, so the
+// worker's orphan watch (ExitWhenOrphaned) knows which parent it must keep.
+const EnvParentPID = "PAD_MATERIALIZE_PARENT_PID"
+
+// guardAgainstOrphaning prepares cmd so the worker cannot outlive the
+// supervising process. If the server dies mid-job (SIGKILL, the OOM killer,
+// a crash), no deadline kill and no RSS watchdog will ever reach the worker
+// again, so it would run on uncapped (macOS) or unbounded in time.
+//
+//   - Every Unix: the worker polls its parent pid (ExitWhenOrphaned) against
+//     the pid passed here, and exits when it changes. The only layer on
+//     macOS.
+//   - Linux, in addition: Pdeathsig (setParentDeathSignal), so the kernel
+//     sends SIGKILL at once.
+//   - Windows: the Job Object's KILL_ON_JOB_CLOSE (memcap_windows.go).
+func guardAgainstOrphaning(cmd *exec.Cmd) {
+	env := cmd.Env
+	if env == nil {
+		env = os.Environ()
+	}
+	cmd.Env = append(env, EnvParentPID+"="+strconv.Itoa(os.Getpid()))
+	setParentDeathSignal(cmd)
+}

internal/materialize/orphan_linux.go

@@ -0,0 +1,26 @@
+package materialize
+
+import (
+	"os/exec"
+	"syscall"
+)
+
+// setParentDeathSignal asks the kernel to SIGKILL the worker when its parent
+// dies.
+//
+// The Go caveat (golang/go#27505): PR_SET_PDEATHSIG fires when the THREAD
+// that forked the child exits, not the process, and Go may run the fork on
+// any thread. The runtime never retires an idle thread; the one case where it
+// ends a thread is a goroutine that exits while locked to it
+// (runtime.LockOSThread). spawn therefore runs cmd.Start on a FRESH goroutine
+// (startUnlocked), which is never locked whatever its caller is, and a thread
+// that is running a locked goroutine runs nothing else, so the fork cannot
+// land on a thread that will be retired. And if the parent is already gone by
+// the time the child sets the flag, Go's forkExec re-checks getppid and
+// signals the child itself.
+func setParentDeathSignal(cmd *exec.Cmd) {
+	if cmd.SysProcAttr == nil {
+		cmd.SysProcAttr = &syscall.SysProcAttr{}
+	}
+	cmd.SysProcAttr.Pdeathsig = syscall.SIGKILL
+}

internal/materialize/orphan_nolinux.go

@@ -0,0 +1,8 @@
+//go:build !linux
+
+package materialize
+
+import "os/exec"
+
+// setParentDeathSignal: Linux only; see orphan_linux.go.
+func setParentDeathSignal(*exec.Cmd) {}

internal/materialize/orphanwatch_other.go

@@ -0,0 +1,7 @@
+//go:build !unix
+
+package materialize
+
+// ExitWhenOrphaned: nothing to do on Windows, where the Job Object's
+// KILL_ON_JOB_CLOSE ends the worker with its parent (memcap_windows.go).
+func ExitWhenOrphaned() {}

internal/materialize/orphanwatch_unix.go

@@ -0,0 +1,40 @@
+//go:build unix
+
+package materialize
+
+import (
+	"os"
+	"strconv"
+	"time"
+)
+
+// OrphanPollInterval is how often the worker checks its parent.
+const OrphanPollInterval = 200 * time.Millisecond
+
+// ExitWhenOrphaned starts a goroutine that exits the process (status 70)
+// as soon as its parent pid is no longer the supervising process: the parent
+// died and the worker was reparented (to init, launchd or a subreaper). The
+// expected parent is EnvParentPID when set, else the parent at the time of
+// the call, so a worker whose parent died before this ran still exits.
+//
+// Called first thing by the worker command, before the bundle loads. It
+// writes nothing: stdout is the protocol channel, and nobody reads stderr
+// once the parent is gone.
+func ExitWhenOrphaned() {
+	want := os.Getppid()
+	if v, err := strconv.Atoi(os.Getenv(EnvParentPID)); err == nil && v > 0 {
+		want = v
+	}
+	check := func() {
+		if os.Getppid() != want {
+			os.Exit(70)
+		}
+	}
+	check()
+	go func() {
+		for {
+			time.Sleep(OrphanPollInterval)
+			check()
+		}
+	}()
+}

internal/materialize/procgone_other_test.go

@@ -0,0 +1,5 @@
+//go:build !unix && !windows
+
+package materialize
+
+func processGone(pid int) error { return nil }

internal/materialize/procgone_unix_test.go

@@ -0,0 +1,27 @@
+//go:build unix
+
+package materialize
+
+import (
+	"errors"
+	"fmt"
+	"os"
+	"strings"
+	"syscall"
+)
+
+// processGone reports nil once pid no longer exists. The supervisor reaps
+// what it kills; an ORPHAN is reaped by whoever adopted it, so a zombie
+// (Linux /proc state Z) also counts as gone: it runs nothing.
+func processGone(pid int) error {
+	err := syscall.Kill(pid, 0)
+	if errors.Is(err, syscall.ESRCH) {
+		return nil
+	}
+	if b, rerr := os.ReadFile(fmt.Sprintf("/proc/%d/stat", pid)); rerr == nil {
+		if i := strings.LastIndexByte(string(b), ')'); i >= 0 && strings.HasPrefix(strings.TrimSpace(string(b[i+1:])), "Z") {
+			return nil
+		}
+	}
+	return fmt.Errorf("pid %d still exists (kill 0: %v)", pid, err)
+}

internal/materialize/procgone_windows_test.go

@@ -0,0 +1,32 @@
+package materialize
+
+import (
+	"errors"
+	"fmt"
+
+	"golang.org/x/sys/windows"
+)
+
+// processGone reports nil once pid no longer names a live process. The
+// supervisor waits on (and so closes its handle to) every child it kills, so
+// a killed worker leaves no process object behind; one that is somehow still
+// open must at least have exited.
+func processGone(pid int) error {
+	h, err := windows.OpenProcess(windows.PROCESS_QUERY_LIMITED_INFORMATION, false, uint32(pid))
+	if err != nil {
+		if errors.Is(err, windows.ERROR_INVALID_PARAMETER) {
+			return nil // no such process
+		}
+		return fmt.Errorf("OpenProcess(%d): %w", pid, err)
+	}
+	defer windows.CloseHandle(h)
+	var code uint32
+	if err := windows.GetExitCodeProcess(h, &code); err != nil {
+		return fmt.Errorf("GetExitCodeProcess(%d): %w", pid, err)
+	}
+	const stillActive = 259
+	if code == stillActive {
+		return fmt.Errorf("pid %d is still running", pid)
+	}
+	return nil
+}

internal/materialize/race_off_test.go

@@ -0,0 +1,5 @@
+//go:build !race
+
+package materialize
+
+const raceEnabled = false

internal/materialize/race_on_test.go

@@ -0,0 +1,5 @@
+//go:build race
+
+package materialize
+
+const raceEnabled = true

internal/materialize/refuse_test.go

@@ -0,0 +1,160 @@
+package materialize
+
+import (
+	"context"
+	"errors"
+	"log/slog"
+	"os"
+	"os/exec"
+	"path/filepath"
+	"runtime"
+	"strings"
+	"sync/atomic"
+	"testing"
+)
+
+// probeCap is a cap whose baseline probe answers what the test says.
+func probeCap(probe func() (uint64, error)) *memCapImpl {
+	return &memCapImpl{
+		mechanism: "probe(test)",
+		attach: func(*exec.Cmd, func()) (*capHandle, error) {
+			return &capHandle{baseline: probe, set: func(uint64) error { return nil }}, nil
+		},
+	}
+}
+
+func journal(t *testing.T, path string) []string {
+	t.Helper()
+	b, err := os.ReadFile(path)
+	if errors.Is(err, os.ErrNotExist) {
+		return nil
+	}
+	if err != nil {
+		t.Fatal(err)
+	}
+	return strings.Fields(string(b))
+}
+
+// A worker whose memory baseline cannot be established is killed WITHOUT
+// being sent a job, and the job fails with ErrNoMemoryCap. One WARN names the
+// cause across any number of such refusals; each refused job's successor
+// starts a new worker and probes again, and a working probe ends the refusals.
+func TestSupervisorBaselineProbeFailureRefuses(t *testing.T) {
+	for name, bad := range map[string]func() (uint64, error){
+		"unreadable":  func() (uint64, error) { return 0, errors.New("probe exploded") },
+		"zero":        func() (uint64, error) { return 0, nil },
+		"implausible": func() (uint64, error) { return MaxPlausibleBaseline + 1, nil },
+	} {
+		t.Run(name, func(t *testing.T) {
+			var broken atomic.Bool
+			broken.Store(true)
+			jpath := filepath.Join(t.TempDir(), "journal")
+			h := newHarness(t, "script", func(c *SupervisorConfig) {
+				c.capOverride = probeCap(func() (uint64, error) {
+					if broken.Load() {
+						return bad()
+					}
+					return 100 << 20, nil
+				})
+				c.extraEnv = []string{journalEnv + "=" + jpath}
+			})
+			const n = 4
+			for i := range n {
+				_, err := h.s.Materialize(context.Background(), script("echo:refused"))
+				if got := journal(t, jpath); len(got) != 0 {
+					t.Fatalf("job %d reached an uncapped worker: %q", i, got)
+				}
+				if !errors.Is(err, ErrNoMemoryCap) {
+					t.Fatalf("job %d: got %v, want ErrNoMemoryCap", i, err)
+				}
+				if err := processGone(h.lastPid(t)); err != nil {
+					t.Fatalf("job %d: the uncapped worker was left running: %v", i, err)
+				}
+			}
+			if got := h.f.calls.Load(); got != n {
+				t.Fatalf("%d spawns for %d refused jobs; each must start and probe a new worker", got, n)
+			}
+			if got := journal(t, jpath); len(got) != 0 {
+				t.Fatalf("jobs reached an uncapped worker: %q", got)
+			}
+			warns := h.log.warns()
+			if len(warns) != 1 || !strings.Contains(warns[0], "could not establish") || !strings.Contains(warns[0], "cause=") {
+				t.Fatalf("want exactly one WARN naming the cause across %d refusals, got %d:\n%s", n, len(warns), strings.Join(warns, "\n"))
+			}
+			t.Log(warns[0])
+
+			// The probe works again: the next start re-probes and the job runs.
+			broken.Store(false)
+			if md, err := h.s.Materialize(context.Background(), script("echo:ok")); err != nil || md != "echo:ok" {
+				t.Fatalf("after the probe recovered: %q, %v", md, err)
+			}
+			if got := journal(t, jpath); len(got) != 1 || got[0] != "echo:ok" {
+				t.Fatalf("journal %q, want only the job after recovery", got)
+			}
+			// A failure after a success is news again: one more WARN.
+			broken.Store(true)
+			h.s.mu.Lock()
+			c := h.s.child
+			h.s.mu.Unlock()
+			if c != nil {
+				c.kill("test")
+				c.awaitExit()
+				h.s.stoppedQuiet(c)
+			}
+			if _, err := h.s.Materialize(context.Background(), script("echo:refused")); !errors.Is(err, ErrNoMemoryCap) {
+				t.Fatalf("after re-breaking: %v", err)
+			}
+			if got := len(h.log.warns()); got != 2 {
+				t.Fatalf("%d WARNs; a failure after a success must warn again", got)
+			}
+		})
+	}
+}
+
+// On an OS with no cap mechanism, no worker is ever started and every job is
+// refused, with one WARN naming the OS.
+func TestSupervisorUnsupportedOSRefuses(t *testing.T) {
+	h := newHarness(t, "script", func(c *SupervisorConfig) {
+		c.capOverride = &memCapImpl{mechanism: "none", unsupported: true}
+	})
+	for i := range 3 {
+		if _, err := h.s.Materialize(context.Background(), script("echo:x")); !errors.Is(err, ErrNoMemoryCap) {
+			t.Fatalf("job %d: %v", i, err)
+		}
+	}
+	if n := h.f.calls.Load(); n != 0 {
+		t.Fatalf("%d workers started on an OS with no cap", n)
+	}
+	warns := h.log.warns()
+	if len(warns) != 1 || !strings.Contains(warns[0], "goos="+runtime.GOOS) {
+		t.Fatalf("want one WARN naming the OS, got %q", warns)
+	}
+}
+
+func TestParseStatVsize(t *testing.T) {
+	line := "1234 (a (weird) name) S 1 1234 1234 0 -1 4194560 100 0 0 0 1 2 0 0 20 0 7 0 555 1423151104 18000 18446744073709551615 1 1 0 0 0 0 0 0 0 0 0 0 17 3 0 0 0 0 0"
+	if n, err := parseStatVsize(line); err != nil || n != 1423151104 {
+		t.Fatalf("%d %v", n, err)
+	}
+	for _, bad := range []string{"", "1234 no parens", "1 (x) S 1 2"} {
+		if _, err := parseStatVsize(bad); err == nil {
+			t.Errorf("%q accepted", bad)
+		}
+	}
+}
+
+func TestStderrOOMMarkers(t *testing.T) {
+	for line, want := range map[string]bool{
+		"fatal error: runtime: out of memory":                                  true,
+		"runtime: mmap: cannot allocate memory":                                true,
+		"runtime/cgo: pthread_create failed: Resource temporarily unavailable": true,
+		"helper: exiting on purpose":                                           false,
+		"panic: boom":                                                          false,
+	} {
+		l := &stderrLog{log: slog.New(&logSink{})}
+		_, _ = l.Write([]byte(line + "\n"))
+		if l.oom.Load() != want {
+			t.Errorf("%q: oom=%v, want %v", line, l.oom.Load(), want)
+		}
+	}
+}

internal/materialize/rss_darwin_test.go

@@ -0,0 +1,5 @@
+package materialize
+
+// testRSSSampler on macOS IS the production watchdog's sampler (proc_info
+// PROC_PIDTASKINFO, ps fallback).
+var testRSSSampler = darwinRSS

internal/materialize/rss_linux_test.go

@@ -0,0 +1,26 @@
+package materialize
+
+import (
+	"fmt"
+	"os"
+	"strconv"
+	"strings"
+)
+
+// testRSSSampler reads /proc/<pid>/statm's resident pages, so the macOS
+// watchdog mechanism can be exercised on Linux.
+var testRSSSampler = func(pid int) (uint64, error) {
+	b, err := os.ReadFile(fmt.Sprintf("/proc/%d/statm", pid))
+	if err != nil {
+		return 0, err
+	}
+	f := strings.Fields(string(b))
+	if len(f) < 2 {
+		return 0, fmt.Errorf("statm: %q", b)
+	}
+	pages, err := strconv.ParseUint(f[1], 10, 64)
+	if err != nil {
+		return 0, err
+	}
+	return pages * uint64(os.Getpagesize()), nil
+}

internal/materialize/rss_other_test.go

@@ -0,0 +1,5 @@
+//go:build !linux && !darwin && !windows
+
+package materialize
+
+var testRSSSampler func(pid int) (uint64, error)

internal/materialize/rss_windows_test.go

@@ -0,0 +1,25 @@
+package materialize
+
+import (
+	"fmt"
+	"unsafe"
+
+	"golang.org/x/sys/windows"
+)
+
+// testRSSSampler reads the working set, so the watchdog mechanism (macOS's
+// cap) is exercised on Windows too. Windows' own cap is the Job Object.
+var testRSSSampler = func(pid int) (uint64, error) {
+	h, err := windows.OpenProcess(windows.PROCESS_QUERY_LIMITED_INFORMATION|windows.PROCESS_VM_READ, false, uint32(pid))
+	if err != nil {
+		return 0, err
+	}
+	defer windows.CloseHandle(h)
+	var c processMemoryCounters
+	c.Cb = uint32(unsafe.Sizeof(c))
+	r, _, e := procK32GetProcessMemoryInfo.Call(uintptr(h), uintptr(unsafe.Pointer(&c)), uintptr(c.Cb))
+	if r == 0 {
+		return 0, fmt.Errorf("K32GetProcessMemoryInfo: %w", e)
+	}
+	return uint64(c.WorkingSetSize), nil
+}

internal/materialize/skip_test.go

@@ -0,0 +1,19 @@
+package materialize
+
+import (
+	"os"
+	"testing"
+)
+
+// noSkipEnv, set by CI's native-smoke job on macOS and Windows, turns the
+// memory-cap tests' skips into failures, so a platform cap that silently
+// stopped running cannot pass as "skipped".
+const noSkipEnv = "PAD_MATERIALIZE_TEST_NO_SKIP"
+
+func skipCapTest(t *testing.T, why string) {
+	t.Helper()
+	if os.Getenv(noSkipEnv) != "" {
+		t.Fatalf("%s is set, but this test would skip: %s", noSkipEnv, why)
+	}
+	t.Skip(why)
+}

internal/materialize/supervisor_config.go

@@ -0,0 +1,178 @@
+package materialize
+
+import (
+	"fmt"
+	"log/slog"
+	"math"
+	"regexp"
+	"strconv"
+	"strings"
+	"time"
+)
+
+// Environment knobs for the materializer worker (TASK-2198). Documented in
+// docs/deployment.md, "Op-log materializer worker".
+const (
+	// EnvTimeout is the per-job hard deadline, in Go duration syntax
+	// ("2s", "1500ms"). Past it the worker process is KILLED.
+	EnvTimeout = "PAD_MATERIALIZE_TIMEOUT"
+	// EnvMemLimit is how much memory one job may add to the loaded worker
+	// (the cap is baseline + limit; what "memory" is differs per OS, see
+	// memCapImpl): a whole number of bytes, or a whole number followed by a
+	// BINARY unit, KiB, MiB or GiB ("2048MiB", "2GiB"). Decimal units (MB,
+	// GB) are refused rather than guessed at.
+	EnvMemLimit = "PAD_MATERIALIZE_MEM_LIMIT"
+	// EnvIdleTimeout is how long the worker may sit with no job before it is
+	// stopped, in Go duration syntax ("5m"). "0" means never. A loaded worker
+	// holds ~200 MB resident, which an install that rarely materializes should
+	// not pay for all day; the next job starts a new one, as the first did.
+	EnvIdleTimeout = "PAD_MATERIALIZE_IDLE_TIMEOUT"
+)
+
+const (
+	DefaultTimeout  = 2 * time.Second
+	MinTimeout      = 250 * time.Millisecond
+	MaxTimeout      = 60 * time.Second
+	DefaultMemLimit = 2 << 30 // 2 GiB
+	// MinMemLimit: the limit is GROWTH beyond the loaded worker (see
+	// memCapImpl), and the measured growth of real work is small
+	// (TestSupervisorBaselineMeasurements, real worker, cgo build,
+	// GOMAXPROCS=2 as the supervisor sets it): the 50-case corpus adds
+	// +0 MiB of address space and +24 MiB resident, and a 4,000-deep nested
+	// document — pathological, not a real item — at most +193 MiB of address
+	// space. Threads created after the baseline: 0 in every run at
+	// GOMAXPROCS=2 (warmThreads makes sure of it at higher counts too). 256 MiB
+	// lets all of that through. The old 1536 MiB floor belonged to the
+	// absolute cap, which had to clear the runtime's own reservation.
+	MinMemLimit = 256 << 20
+	MaxMemLimit = 16 << 30
+
+	DefaultIdleTimeout = 5 * time.Minute
+	// MinIdleTimeout keeps a busy install from paying the ~2.6 s bundle load
+	// between jobs that arrive seconds apart.
+	MinIdleTimeout = 30 * time.Second
+	MaxIdleTimeout = 24 * time.Hour
+	// NeverIdle, as SupervisorConfig.IdleTimeout, keeps the worker until Close
+	// (EnvIdleTimeout "0"). The zero value is the default, as for the other
+	// fields.
+	NeverIdle time.Duration = -1
+)
+
+var memLimitRE = regexp.MustCompile(`^([0-9]+)(KiB|MiB|GiB)?$`)
+
+// ParseMemLimit reads the EnvMemLimit syntax: a whole number of bytes, or a
+// whole number with a KiB, MiB or GiB suffix (case-sensitive, no space).
+func ParseMemLimit(s string) (uint64, error) {
+	m := memLimitRE.FindStringSubmatch(strings.TrimSpace(s))
+	if m == nil {
+		return 0, fmt.Errorf("%q is not a byte count or a whole number of KiB, MiB or GiB", s)
+	}
+	n, err := strconv.ParseUint(m[1], 10, 64)
+	if err != nil {
+		return 0, fmt.Errorf("%q: %w", s, err)
+	}
+	shift := map[string]uint{"": 0, "KiB": 10, "MiB": 20, "GiB": 30}[m[2]]
+	if n > math.MaxUint64>>shift {
+		return 0, fmt.Errorf("%q overflows", s)
+	}
+	return n << shift, nil
+}
+
+// ConfigFromEnv reads EnvTimeout, EnvMemLimit and EnvIdleTimeout through getenv (os.Getenv in
+// production). An unset variable takes the default; an unparseable one takes
+// the default and is logged as a warning. Out-of-range values are clamped by
+// NewSupervisor, not here, so a config built by hand gets the same clamp.
+func ConfigFromEnv(getenv func(string) string, logger *slog.Logger) SupervisorConfig {
+	if logger == nil {
+		logger = slog.Default()
+	}
+	cfg := SupervisorConfig{Logger: logger}
+	if v := getenv(EnvTimeout); v != "" {
+		d, err := time.ParseDuration(strings.TrimSpace(v))
+		if err != nil {
+			logger.Warn("materialize: ignoring invalid "+EnvTimeout+"; using the default",
+				"value", v, "error", err, "default", DefaultTimeout.String())
+		} else {
+			cfg.Timeout = d
+		}
+	}
+	if v := getenv(EnvMemLimit); v != "" {
+		n, err := ParseMemLimit(v)
+		if err != nil {
+			logger.Warn("materialize: ignoring invalid "+EnvMemLimit+"; using the default",
+				"value", v, "error", err, "default", formatBytes(DefaultMemLimit))
+		} else {
+			cfg.MemLimit = n
+		}
+	}
+	if v := getenv(EnvIdleTimeout); v != "" {
+		d, err := time.ParseDuration(strings.TrimSpace(v))
+		switch {
+		case err != nil:
+			logger.Warn("materialize: ignoring invalid "+EnvIdleTimeout+"; using the default",
+				"value", v, "error", err, "default", DefaultIdleTimeout.String())
+		case d < 0:
+			logger.Warn("materialize: ignoring negative "+EnvIdleTimeout+"; using the default (0 means never)",
+				"value", v, "default", DefaultIdleTimeout.String())
+		case d == 0:
+			cfg.IdleTimeout = NeverIdle
+		default:
+			cfg.IdleTimeout = d
+		}
+	}
+	return cfg
+}
+
+// effectiveIdleTimeout applies the default and the clamp; a negative request
+// (NeverIdle) is returned as NeverIdle. The second result says whether a
+// positive request was changed.
+func effectiveIdleTimeout(d time.Duration) (time.Duration, bool) {
+	switch {
+	case d < 0:
+		return NeverIdle, false
+	case d == 0:
+		return DefaultIdleTimeout, false
+	case d < MinIdleTimeout:
+		return MinIdleTimeout, true
+	case d > MaxIdleTimeout:
+		return MaxIdleTimeout, true
+	}
+	return d, false
+}
+
+// effectiveTimeout applies the default and the clamp. The second result says
+// whether a non-zero request was changed.
+func effectiveTimeout(d time.Duration) (time.Duration, bool) {
+	switch {
+	case d == 0:
+		return DefaultTimeout, false
+	case d < MinTimeout:
+		return MinTimeout, true
+	case d > MaxTimeout:
+		return MaxTimeout, true
+	}
+	return d, false
+}
+
+func effectiveMemLimit(n uint64) (uint64, bool) {
+	switch {
+	case n == 0:
+		return DefaultMemLimit, false
+	case n < MinMemLimit:
+		return MinMemLimit, true
+	case n > MaxMemLimit:
+		return MaxMemLimit, true
+	}
+	return n, false
+}
+
+// formatBytes renders n in the EnvMemLimit syntax, exactly when it can.
+func formatBytes(n uint64) string {
+	switch {
+	case n != 0 && n%(1<<30) == 0:
+		return fmt.Sprintf("%dGiB", n>>30)
+	case n != 0 && n%(1<<20) == 0:
+		return fmt.Sprintf("%dMiB", n>>20)
+	}
+	return strconv.FormatUint(n, 10)
+}

internal/materialize/worker.go

@@ -9,6 +9,8 @@ import (
 	"errors"
 	"fmt"
 	"io"
+	"runtime"
+	"sync"
 	"time"
 )
 
@@ -57,6 +59,42 @@ func RunWorker(r io.Reader, w io.Writer, js []byte) error {
 	if err != nil {
 		return err
 	}
+	warmThreads()
+	return serveWorker(r, w, runner)
+}
+
+// warmThreads brings the worker's thread set to steady state BEFORE it reads
+// the readiness probe, i.e. before the supervisor measures the baseline the
+// memory cap is relative to (supervisor.go armCap). Every P runs an
+// allocating goroutine at once, so the runtime starts an M for each, and a
+// GC starts its workers. Threads created only later — during a job — would
+// otherwise come out of that job's budget: on a cgo build each reserves an
+// 8 MiB stack, and CI measured the growth this closes.
+func warmThreads() {
+	n := runtime.GOMAXPROCS(0)
+	var wg sync.WaitGroup
+	start := make(chan struct{})
+	for range n {
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			<-start
+			var keep [][]byte
+			for range 32 {
+				keep = append(keep, make([]byte, 64<<10))
+			}
+			runtime.KeepAlive(keep)
+		}()
+	}
+	close(start)
+	wg.Wait()
+	runtime.GC()
+}
+
+// serveWorker is RunWorker's loop over an already-loaded Runner. Split out so
+// the tests of the loop share one loaded bundle instead of paying the load
+// (~25 s under the race detector) per test.
+func serveWorker(r io.Reader, w io.Writer, runner *Runner) error {
 	br := bufio.NewReader(r)
 	bw := bufio.NewWriter(w)
 	for {

internal/materialize/worker_test.go

@@ -96,7 +96,7 @@ func TestWorkerRoundTrip(t *testing.T) {
 
 func TestWorkerEmptyInput(t *testing.T) {
 	var out bytes.Buffer
-	if err := RunWorker(bytes.NewReader(nil), &out, pad.MaterializerJS); err != nil {
+	if err := serveWorker(bytes.NewReader(nil), &out, runner(t)); err != nil {
 		t.Fatal(err)
 	}
 	if out.Len() != 0 {
@@ -111,7 +111,7 @@ func TestWorkerTruncatedFrame(t *testing.T) {
 		"body":   append(append([]byte{}, good...), good[:len(good)-3]...),
 	} {
 		var out bytes.Buffer
-		err := RunWorker(bytes.NewReader(in), &out, pad.MaterializerJS)
+		err := serveWorker(bytes.NewReader(in), &out, runner(t))
 		if !errors.Is(err, io.ErrUnexpectedEOF) {
 			t.Errorf("%s: got %v, want ErrUnexpectedEOF", name, err)
 		}
@@ -125,7 +125,7 @@ func TestWorkerOversizedFrame(t *testing.T) {
 	in := make([]byte, 4)
 	binary.BigEndian.PutUint32(in, MaxFrameBytes+1)
 	var out bytes.Buffer
-	if err := RunWorker(bytes.NewReader(in), &out, pad.MaterializerJS); !errors.Is(err, errFrameTooLarge) {
+	if err := serveWorker(bytes.NewReader(in), &out, runner(t)); !errors.Is(err, errFrameTooLarge) {
 		t.Fatalf("got %v", err)
 	}
 	resps := readResponses(t, out.Bytes())